view release on metacpan or  search on metacpan

t/23-connection-cleanup.t  view on Meta::CPAN

    my $port = $server->port;

    my @sockets;
    # Send 10 requests (5 will throw exceptions)
    for my $i (1..10) {
        my $sock = IO::Socket::INET->new(
            PeerAddr => '127.0.0.1',
            PeerPort => $port,
            Proto    => 'tcp',
            Timeout  => 2,
        ) or die "Cannot connect: $!";

        # Use keep-alive to verify exception connections are closed, not due to Connection: close
        print $sock "GET /$i HTTP/1.1\r\nHost: localhost\r\nConnection: keep-alive\r\n\r\n";

        my $response = '';
        $sock->blocking(0);
        my $deadline = time + 2;
        while (time < $deadline) {
            $loop->loop_once(0.1);
            my $data;
            my $bytes = sysread($sock, $data, 4096);
            if (defined $bytes && $bytes > 0) {
                $response .= $data;
            }
            elsif (defined $bytes && $bytes == 0) {
                last;  # Server closed connection (exception case)
            }
            # For keep-alive success case, response ends with body
            last if $response =~ /OK$/;
        }
        push @sockets, $sock;  # Keep socket open to simulate keep-alive
    }

    # Let server process
    $loop->loop_once(0.2);

    is($request_count, 10, "All 10 requests processed");
    is($exception_count, 5, "5 exceptions thrown");

    # Exception connections (5) should be closed immediately
    # Keep-alive successful connections (5) should still be tracked (waiting for more requests)
    my $conn_count = keys %{$server->{connections}};
    is($conn_count, 5, "Exception connections closed, keep-alive connections still tracked");

    # Clean up: close client sockets
    close($_) for @sockets;

    $server->shutdown->get;
    eval { $loop->remove($server) };
};

# =============================================================================
# Test: h1 teardown with the outbound buffer above the high mark must NOT
# fire on_drain (the connection is going away, not draining -- h2 already
# gets this right via its separate transport_drain_fires list; h1's arm_drain
# used to piggyback directly on the same _drain_waiters Futures that
# blocking producers await, so tearing down resolved both indiscriminately
# and fired on_drain with the buffer nowhere near the low mark). A producer
# genuinely parked on a blocking backpressure await must still resume (no
# coroutine leak) -- only the app-facing on_drain callback must be dropped.
#
# Exercises the real Connection-level machinery (_h1_transport_state's
# arm_drain, _wait_for_drain, _cancel_drain_waiters) directly against a
# fake stream that reports a controlled, non-draining buffer size -- a
# unit-level handler invocation, not a real socket, so the buffer-stays-full
# precondition is exact rather than timing-dependent.
# =============================================================================

package Local::FakeWriter {
    sub new { my ($c, $d) = @_; return bless { data => $d }, $c }
    sub data { return $_[0]->{data} }
}

package Local::FakeStream {
    sub new { return bless { writequeue => [] }, shift }
    sub configure { my $self = shift; my %args = @_; %$self = (%$self, %args); return }
}

package Local::FakeServer {
    sub new { my ($c, $loop) = @_; return bless { loop => $loop }, $c }
    sub loop { return $_[0]->{loop} }
}

subtest 'h1 teardown with buffer above the high mark drops on_drain but resumes a parked wait' => sub {
    my $fake_loop   = IO::Async::Loop->new;
    my $fake_server = Local::FakeServer->new($fake_loop);
    my $fake_stream = Local::FakeStream->new;

    my $conn = PAGI::Server::Connection->new(
        stream => $fake_stream, server => $fake_server, app => sub { },
        write_high_watermark => 10, write_low_watermark => 2,
    );

    # Buffer is (and stays, for this test) well above the high mark.
    $fake_stream->{writequeue} = [ Local::FakeWriter->new('x' x 100) ];

    my $transport = $conn->_h1_transport_state;
    my $drain_fired = 0;
    $transport->on_drain(sub { $drain_fired++ });

    # Crossing the high mark arms drain detection -- h1's arm_drain fires
    # into $conn->{_drain_fires} (post-fix) rather than reusing the blocking
    # producer queue.
    $transport->_check_watermarks;

    # A genuinely parked producer, awaiting the buffer to drain before its
    # next write -- must still resume when the connection tears down.
    my $parked = $conn->_wait_for_drain;
    ok(!$parked->is_ready, 'parked wait is pending before teardown (buffer still above low mark)');

    # Teardown: the buffer is still full (nothing has actually drained).
    $conn->_cancel_drain_waiters('connection closing');

    is($drain_fired, 0, 'on_drain did NOT fire on teardown (connection going away, not draining)');
    ok($parked->is_ready, 'the parked blocking wait resumed anyway (no coroutine leak)');
};

subtest 'h1 on_drain still fires normally when the buffer genuinely drains (not torn down)' => sub {
    my $fake_loop   = IO::Async::Loop->new;
    my $fake_server = Local::FakeServer->new($fake_loop);
    my $fake_stream = Local::FakeStream->new;

    my $conn = PAGI::Server::Connection->new(
        stream => $fake_stream, server => $fake_server, app => sub { },
        write_high_watermark => 10, write_low_watermark => 2,
    );

    $fake_stream->{writequeue} = [ Local::FakeWriter->new('x' x 100) ];

    my $transport = $conn->_h1_transport_state;
    my $drain_fired = 0;
    $transport->on_drain(sub { $drain_fired++ });
    $transport->_check_watermarks;

    my $parked = $conn->_wait_for_drain;

    # The buffer actually falls back below the low mark -- a real drain, the
    # same event on_outgoing_empty reports on a live stream.
    $fake_stream->{writequeue} = [];
    $conn->_check_drain_waiters;

    is($drain_fired, 1, 'on_drain fired on a genuine drain');
    ok($parked->is_ready, 'the parked blocking wait also resolved');
};

# =============================================================================
# Test: socket-error handlers (spec tokens read_error/write_error).
#
# IO::Async::Stream's own contract (perldoc IO::Async::Stream, "on_read_error"
# / "on_write_error"): "If an error occurs when the corresponding error
# callback is not supplied, ... the close method is called instead" --
# confirmed in source (_do_read / _do_write):
#   $self->maybe_invoke_event(on_read_error => $errno) or $self->close_now;
# maybe_invoke_event always returns a (truthy) arrayref once ANY handler is
# registered, regardless of what that handler itself returns, so simply
# registering on_read_error/on_write_error is sufficient by itself to
# suppress IO::Async's own close_now -- there is no risk of double-teardown
# from IO::Async's side; our handler becomes solely responsible for tearing
# the connection down, via the same _handle_disconnect_and_close every other
# reason already uses (whose _disconnect_handled guard is independently
# idempotent against any other path -- e.g. on_closed -- that might also
# fire afterward).
#
# A live EPIPE/ECONNRESET provocation (RST-closing the client socket via
# SO_LINGER=>0, then forcing a server write) was attempted first, per the
# audit brief, using several variants (immediate release, delayed release,
# large single write, many small writes). On this platform/sandbox the
# server's on_read handler's EOF/close detection reliably wins the race
# against any write-side failure -- every variant tried settled on
# client_closed (or a same-tick 200, meaning the write itself succeeded
# despite the peer RST) rather than a genuine write-side error, so this pins
# the fix with a unit-level handler invocation instead, per the brief's own
# fallback allowance: a real Connection, with start() actually called (so
# on_read_error/on_write_error are registered by the real production code,
# not hand-rolled by the test), against a real IO::Async::Stream backed by a