File Coverage

blib/lib/DBIx/QuickDB/Driver.pm

Criterion	Covered	Total	%
statement	33	214	15.4
branch	0	92	0.0
condition	0	41	0.0
subroutine	11	46	23.9
pod	26	31	83.8
total	70	424	16.5

package DBIx::QuickDB::Driver;
use strict;
use warnings;

our $VERSION = '0.000048';

use Carp qw/croak confess/;
use File::Path qw/remove_tree/;
use File::Temp qw/tempdir/;
use POSIX ":sys_wait_h";
use Scalar::Util qw/blessed/;
use Time::HiRes qw/sleep time/;

use DBIx::QuickDB::Util qw/clone_dir env_timeout/;

use DBIx::QuickDB::Watcher;

use DBIx::QuickDB::Util::HashBase qw{
    -root_pid
    -dir
    -_cleanup
    -autostop -autostart
    verbose
    -_log_id
    username
    password
    env_vars


};

sub viable { (0, "viable() is not implemented for the " . $_[0]->name . " driver") }

sub socket         { confess "socket() is not implemented for the " . $_[0]->name . " driver" }
sub load_sql       { confess "load_sql() is not implemented for the " . $_[0]->name . " driver" }
sub bootstrap      { confess "bootstrap() is not implemented for the " . $_[0]->name . " driver" }
sub connect_string { confess "connect_string() is not implemented for the " . $_[0]->name . " driver" }
sub start_command  { confess "start_command() is not implemented for the " . $_[0]->name . " driver" }
sub shell_command  { confess "shell_command() is not implemented for the " . $_[0]->name . " driver" }

sub error_log { undef }

sub read_error_log {
    my $self = shift;
    my $log = $self->error_log or return "";
    return "" unless -f $log;
    open(my $fh, '<', $log) or return "Could not open error log '$log': $!";
    return join "" => <$fh>;
}

sub list_env_vars { qw/DBI_USER DBI_PASS DBI_DSN/ }

sub version_string { 'unknown' }

sub stop_sig { 'TERM' }

sub write_config {}

sub do_in_env {
    my $self = shift;
    my ($code) = @_;

    my $old = $self->mask_env_vars;

    my $ok = eval { $code->(); 1 };
    my $err = $@;

    $self->unmask_env_vars($old);

    die $err unless $ok;

    return;
}

sub mask_env_vars {
    my $self = shift;

    my %old;

    for my $var ($self->list_env_vars) {
        next unless defined $ENV{$var};
        $old{$var} = delete $ENV{$var};
    }

    my $env_vars = $self->env_vars || {};
    for my $var (keys %$env_vars) {
        $old{$var} = delete $ENV{$var} unless defined $old{$var};
        $ENV{$var} = $env_vars->{$var};
    }

    return \%old;
}

sub unmask_env_vars {
    my $self = shift;
    my ($old) = @_;

    for my $var (keys %$old) {
        my $val = $old->{$var};

        if (defined $val) {
            $ENV{$var} = $val;
        }
        else {
            delete $ENV{$var};
        }
    }

    return;
}

sub name {
    my $in = shift;
    my $type = blessed($in) || $in;

    $type =~ s/^DBIx::QuickDB::Driver:://;

    return $type;
}

sub init {
    my $self = shift;

    confess "'dir' is a required attribute" unless $self->{+DIR};

    $self->{+ROOT_PID} = $$;
    $self->{+_CLEANUP} = delete $self->{cleanup};

    $self->{+USERNAME} = '' unless defined $self->{+USERNAME};
    $self->{+PASSWORD} = '' unless defined $self->{+PASSWORD};

    $self->{+ENV_VARS} ||= {};

    return;
}

sub clone_data {
    my $self = shift;

    return (
        USERNAME()  => $self->{+USERNAME},
        PASSWORD()  => $self->{+PASSWORD},
        VERBOSE()   => $self->{+VERBOSE},
        AUTOSTOP()  => $self->{+AUTOSTOP},
        AUTOSTART() => $self->{+AUTOSTART},

        cleanup => $self->{+_CLEANUP},

        ENV_VARS() => {%{$self->{+ENV_VARS}}},
    );
}

sub resync {
    my $self = shift;

    my $from = $self->{+CLONED_FROM} or croak "No original source to sync from";

    my $started = $self->started;
    $self->stop if $started;

    clone_dir($from, $self->{+DIR}, verbose => (($self->{+VERBOSE} // 0) > 2) ? 1 : 0, checksum => 1);

    $self->write_config();

    $self->start if $started;
}

sub clone {
    my $self = shift;
    my %params = @_;

    confess "Cannot clone a started database, please stop it first."
        if $self->started;

    my $orig_dir = $self->{+DIR};
    my $new_dir  = delete $params{dir} // tempdir('DB-QUICK-CLONE-XXXXXX', CLEANUP => 0, $ENV{QDB_TMPDIR} ? (DIR => $ENV{QDB_TMPDIR}) : (TMPDIR => 1));

    clone_dir($orig_dir, $new_dir, verbose => (($self->{+VERBOSE} // 0) > 2) ? 1 : 0);

    my $class = ref($self);
    my %ok = (
        cleanup => 1,
        map {$_ => 1} DBIx::QuickDB::Util::HashBase::attr_list($class),
    );
    my @bad = grep { !$ok{$_} } keys %params;

    confess "Invalid options to clone(): " . join(', ' => @bad)
        if @bad;

    my $clone = $class->new(
        $self->clone_data,

        %params,

        DIR() => $new_dir,

        WATCHER()  => undef,

        CLONED_FROM() => $orig_dir,
    );

    $clone->write_config();
    $clone->start if $clone->{+AUTOSTART};

    return $clone;
}

sub gen_log {
    my $self = shift;
    return if $self->no_log(@_);
    return $self->{+DIR} . "/cmd-log-$$-" . $self->{+_LOG_ID}++;
}

sub no_log {
    my $self = shift;
    my ($params) = @_;
    return $self->{+VERBOSE} || $params->{no_log} || $ENV{DB_VERBOSE};
}

sub run_command {
    my $self = shift;
    my ($cmd, $params) = @_;

    my $no_log = $self->no_log($params);
    my $log_file = $params->{log_file} || ($no_log ? undef : $self->gen_log);

    my $pid = fork();
    croak "Could not fork" unless defined $pid;

    if ($pid) {
        local $?;
        return ($pid, $log_file) if $params->{no_wait};
        my $ret = waitpid($pid, 0);
        my $exit = $?;
        die "waitpid returned $ret" unless $ret == $pid;

        return unless $exit;

        my $log = "";
        unless ($no_log) {
            open(my $fh, '<', $log_file) or warn "Failed to open log: $!";
            $log = eval { join "" => <$fh> };
        }
        my $error_log = $self->read_error_log;
        $log .= "\n=== error log ===\n$error_log" if length $error_log;
        croak "Failed to run command '" . join(' ' => @$cmd) . "' ($exit)\n$log";
    }

    $self->mask_env_vars;

    unless ($no_log) {
        open(my $log, '>', $log_file) or die "Could not open log file ($log_file): $!";
        close(STDOUT);
        open(STDOUT, '>&', $log);
        close(STDERR);
        open(STDERR, '>&', $log);
    }

    if (my $file = $params->{stdin}) {
        close(STDIN);
        open(STDIN, '<', $file) or die "Could not open new STDIN ($file): $!";
    }

    exec(@$cmd);
}

sub should_cleanup { shift->{+_CLEANUP} }

# Flush server state durably to disk so a hard kill mid-shutdown cannot leave
# the data dir needing crash recovery. Drivers that benefit (e.g. PostgreSQL)
# override this; the default is a no-op.
sub checkpoint { }

sub cleanup {
    my $self = shift;

    # Ignore errors here.
    my $err = [];
    remove_tree($self->{+DIR}, {safe => 1, error => \$err}) if -d $self->{+DIR};
    return;
}

sub connect {
    my $self = shift;
    my ($db_name, %params) = @_;

    %params = (AutoCommit => 1, RaiseError => 1) unless @_ > 1;

    my $dbh;
    $self->do_in_env(
        sub {
            my $cstring = $self->connect_string($db_name);
            require DBI;
            $dbh = DBI->connect($cstring, $self->username, $self->password, \%params);
        }
    );

    return $dbh;
}

sub started {
    my $self = shift;

    my $socket = $self->socket;
    return 1 if $self->{+WATCHER} || -S $socket;
    return 0;
}

sub start {
    my $self = shift;
    my @args = @_;

    my $dir = $self->{+DIR};
    my $socket = $self->socket;

    return if $self->{+WATCHER} || -S $socket;

    my $watcher = $self->{+WATCHER} = DBIx::QuickDB::Watcher->new(db => $self, args => \@args);

    # Defaults to 10s; tunable via QDB_START_TIMEOUT for slow hosts that need
    # longer to bring a server up (e.g. a clone doing crash recovery).
    my $timeout = env_timeout(QDB_START_TIMEOUT => 10);

    my $start = time;
    until (-S $socket) {
        my $waited = time - $start;

        if ($waited > $timeout) {
            my $error_log = $self->read_error_log;
            $watcher->eliminate();
            confess "Timed out waiting for server to start\n$error_log";
            last;
        }

        sleep 0.01;
    }

    return;
}

sub stop {
    my $self = shift;
    my %params = @_;

    my $watcher = delete $self->{+WATCHER} or return;

    # Flush a durable checkpoint while the server is still running. If shutdown
    # is slow enough to be SIGKILLed, this ensures the on-disk state is already
    # consistent so a later clone does not crash-recover and jump SERIAL
    # sequences forward (PostgreSQL SEQ_LOG_VALS=32), corrupting the clone.
    # No-op for drivers that do not need it.
    $self->checkpoint;

    DBI->visit_handles(
        sub {
            my ($driver_handle) = @_;

            $driver_handle->disconnect
               if $driver_handle->{Type} && $driver_handle->{Type} eq 'db'
               && $driver_handle->{Name} && index($driver_handle->{Name}, $self->{+DIR}) >= 0;

            return 1;
        }
    );

    $watcher->stop();

    unless ($params{no_wait}) {
        # wait() blocks until the watcher process exits, and the watcher reaps
        # the server before it exits -- so once wait() returns the server is
        # gone. Trust that instead of polling a stored server pid: after the
        # watcher exits that pid may have been recycled by the OS to an
        # unrelated process, and polling it could hang/confess on the wrong
        # process (the same pid-reuse hazard the watcher teardown guards against).
        $watcher->wait();

        # Remove a stale unix socket left behind by a hard kill so it does not
        # confuse callers or a later run that reuses the same directory.
        my $socket = $self->socket;
        unlink($socket) if $socket && -S $socket;
    }

    return;
}

sub shell {
    my $self = shift;
    my ($db_name) = @_;
    $db_name = 'quickdb' unless defined $db_name;

    system($self->shell_command($db_name));
}

sub DESTROY {
    my $self = shift;
    return unless $self->{+ROOT_PID} && $self->{+ROOT_PID} == $$;

    if (my $watcher = delete $self->{+WATCHER}) {
        # eliminate() signals the watcher to stop the server and delete the data
        # dir; destroying the watcher then blocks (via Watcher::wait) until the
        # watcher process has exited, and the watcher reaps the server before it
        # exits. So once $watcher is gone the server is gone too. We deliberately
        # do NOT fall back to signalling a stored server pid here: after the
        # watcher exits that pid may have been recycled to an unrelated process
        # (the pid-reuse hazard), so a stray TERM/KILL could hit the wrong one.
        $watcher->eliminate();
        undef $watcher;

        $self->cleanup() if $self->should_cleanup;
    }
    elsif ($self->should_cleanup) {
        $self->cleanup();
    }

    return;
}

1;

__END__