view release on metacpan or  search on metacpan

t/12_bug_regressions.t  view on Meta::CPAN

    my $out;
    {
        local $/;
        $out = <$rd>;
    }
    close $rd;
    $out = '' unless defined $out;
    chomp $out;
    if (my $sig = $status & 127) { return ("signal:$sig", $out) }
    my $exit = $status >> 8;
    return ($exit == 0 ? 'ok' : "exit:$exit", $out);
}

# 1. Regression: skip off-by-one misdelivery.  In pipeline mode with a large
# single-row result A, calling skip_pending from A's first-row callback and
# then queueing D must deliver D's OWN result ('DDD'), never 'CCC'
# (pre-fix: D got 'CCC').
{
    my $pg;
    my $skipped = 0;
    my ($d_calls, $d_value) = (0, undef);
    $pg = EV::Pg->new(
        conninfo => $conninfo,
        on_connect => sub {
            $pg->enter_pipeline;
            $pg->query_params(
                "select repeat('x',200) from generate_series(1,50000)", [], sub {
                my ($r, $e) = @_;
                return if $skipped;
                $skipped = 1;
                $pg->skip_pending;
                $pg->query_params("select 'DDD'::text", [], sub {
                    my ($r2, $e2) = @_;
                    $d_calls++;
                    $d_value = $r2->[0][0] if ref $r2 && @$r2;
                });
                $pg->pipeline_sync(sub { EV::break });
            });
            $pg->set_single_row_mode;
            $pg->query_params("select 'BBB'::text", [], sub { });
            $pg->query_params("select 'CCC'::text", [], sub { });
            $pg->send_flush_request;
        },
        on_error => sub { diag "Error: $_[0]"; EV::break },
    );
    my $t = EV::timer(10, 0, sub { EV::break });
    EV::run;
    $pg->finish if $pg->is_connected;
    is($d_calls, 1, 'skip off-by-one: D callback fired exactly once');
    is($d_value, 'DDD', "skip off-by-one: D received its own 'DDD' (pre-fix: 'CCC')");
}

# 2. Regression: COPY-OUT skip livelock.  skip_pending from a COPY_OUT tag
# callback with ~20MB still in flight must not spin at 100% CPU, and the
# connection must recover.  Forked: a livelock regression hangs the process
# where no EV timer can fire, so only a parent-side wall-clock kills it.
{
    my ($st, $out) = run_isolated(sub {
        my $wr = shift;
        my $pg;
        my ($skipped, $alive) = (0, '');
        my $retry;
        $pg = EV::Pg->new(
            conninfo => $conninfo,
            on_connect => sub {
                $pg->enter_pipeline;
                $pg->query_params(
                    "copy (select repeat('x',100) from generate_series(1,200000)) to stdout",
                    [], sub {
                    my ($r, $e) = @_;
                    if (!$skipped && defined $r && !ref $r && $r eq 'COPY_OUT') {
                        $skipped = 1;
                        $pg->skip_pending;
                        # libpq is still in COPY state until the skip drain
                        # finishes in the io watcher; retry queueing 'alive'
                        # until the connection accepts new commands again.
                        $retry = EV::timer(0.05, 0.05, sub {
                            my $ok = eval {
                                $pg->query_params("select 'alive'::text", [], sub {
                                    my ($r2, $e2) = @_;
                                    $alive = $r2->[0][0] if ref $r2 && @$r2;
                                    EV::break;
                                });
                                1;
                            };
                            if ($ok) {
                                undef $retry;
                                $pg->send_flush_request;
                            }
                        });
                    }
                });
                $pg->query_params("select 'after'::text", [], sub { });
                $pg->pipeline_sync(sub { });
            },
            on_error => sub { warn "Error: $_[0]"; EV::break },
        );
        my $t = EV::timer(8, 0, sub { EV::break });
        EV::run;
        print $wr "$alive\n";
    }, 10);
    is($st, 'ok', 'COPY-OUT skip: no livelock (pre-fix: 100% CPU hang)');
    is($out, 'alive', "COPY-OUT skip: connection recovered, 'alive' returned");
}

# 3. Regression: use-after-free on a custom (non-default) loop.  Freeing the
# EV::Loop and then the EV::Pg must not segfault.  Forked: a regression kills
# the process with SIGSEGV, which the parent detects via the wait status.
{
    my ($st, $out) = run_isolated(sub {
        my $wr = shift;
        my $loop = EV::Loop->new;
        my $pg = EV::Pg->new(loop => $loop);
        $pg->on_error(sub { warn "Error: $_[0]"; $loop->break });
        $pg->on_connect(sub { $loop->break });
        $pg->connect($conninfo);
        my $guard = $loop->timer(5, 0, sub { $loop->break });
        $loop->run;
        undef $guard;
        $pg->on_connect(undef);   # drop closure capturing $loop
        $pg->on_error(undef);
        undef $loop;              # ev_loop_destroy; $pg's loop pointer dangles
        undef $pg;                # DESTROY must not touch the freed loop
        print $wr "survived\n";
    }, 10);
    is($st, 'ok', "custom loop UAF: clean exit after undef loop + undef pg (pre-fix: SIGSEGV)");
}

# 4. Regression: re-entrant skip double-count.  A's "skipped" callback
# re-enters skip_pending; queries queued afterwards must each get their OWN
# result.  Forked: pre-fix E was misdelivered and then the connection hung.
{
    my ($st, $out) = run_isolated(sub {
        my $wr = shift;
        my $pg;
        my $reentered = 0;
        my ($e_val, $f_val) = ('', '');
        $pg = EV::Pg->new(
            conninfo => $conninfo,
            on_connect => sub {
                $pg->enter_pipeline;
                $pg->query_params("select 'A'::text", [], sub {
                    my ($r, $e) = @_;
                    if (!$reentered) {
                        $reentered = 1;
                        $pg->skip_pending;    # inner re-entrant skip
                    }
                });
                $pg->query_params("select 'B'::text", [], sub { });
                $pg->query_params("select 'C'::text", [], sub { });
                $pg->skip_pending;            # outer skip over A,B,C
                $pg->query_params("select 'EEE'::text", [], sub {
                    my ($r, $e) = @_;
                    $e_val = $r->[0][0] if ref $r && @$r;
                });
                $pg->query_params("select 'FFF'::text", [], sub {
                    my ($r, $e) = @_;
                    $f_val = $r->[0][0] if ref $r && @$r;
                });
                $pg->pipeline_sync(sub { EV::break });
                $pg->send_flush_request;
            },
            on_error => sub { warn "Error: $_[0]"; EV::break },