view release on metacpan or  search on metacpan

t/http2/15-sse-keepalive.t  view on Meta::CPAN

    my $ss_a = $conn->{h2_streams}{$a_id};
    ok($ss_a, 'server holds per-stream state for A immediately after open');

    open_sse_stream_tracked($client, $client_sock, '/b', \$b_id);
    my $ss_b = $conn->{h2_streams}{$b_id};
    ok($ss_b, 'server holds per-stream state for B immediately after open');

    # Both keepalives are armed immediately after each stream's sse.start.
    # A ticks every 0.2s; expect >= 2 pingA comments by ~0.4s. Ceiling 3s is
    # a >= 7x margin over that expectation.
    my $deadline = Time::HiRes::time() + 3;
    while (Time::HiRes::time() < $deadline && comment_count($a_data, 'pingA') < 2) {
        exchange_frames($client, $client_sock, 1);
    }
    ok(comment_count($a_data, 'pingA') >= 2,
        'stream A received >= 2 of its own keepalive comments')
        or diag("a_data: $a_data");

    # Isolation: the bug this task fixes is the h2 SSE keepalive writer being
    # connection-level, so the second stream's sse.start silently replaced
    # the first stream's writer and one stream ended up carrying (some of)
    # the other's comments. Neither stream may EVER see the other's text.
    is(comment_count($a_data, 'pingB'), 0, 'stream A never received a pingB comment');
    is(comment_count($b_data, 'pingA'), 0, 'stream B never received a pingA comment');

    my $timer_a = $ss_a->{sse_ka_timer};
    ok($timer_a && $timer_a->is_running, "A's per-stream keepalive timer is armed");

    # A closes itself at ~0.65s (app-driven, see make_two_stream_keepalive_app).
    # Ceiling 3s (fresh deadline from here) is a wide margin over whatever
    # remains of that 0.65s budget.
    $deadline = Time::HiRes::time() + 3;
    while (Time::HiRes::time() < $deadline && !$closed{$a_id}) {
        exchange_frames($client, $client_sock, 1);
    }
    ok($closed{$a_id}, 'stream A closed within the bounded window');

    # White-box: A's own per-stream keepalive timer is released; B's timer
    # (a DIFFERENT $ss, on the same connection) is untouched and still
    # running. This is design section 11.3's "closing one stream removes
    # only its timers".
    ok(!$ss_a->{sse_ka_timer}, "A's sse_ka_timer field is released after A's close");
    ok($timer_a && !$timer_a->is_running, "A's armed timer object itself is stopped");
    ok($ss_b->{sse_ka_timer} && $ss_b->{sse_ka_timer}->is_running,
        "B's sse_ka_timer is still running after A closed");
    ok(!$closed{$b_id},
        'stream B has not closed yet (well before its own 2.5s close budget)');

    # B keeps ticking at its own 0.5s cadence after A is gone. Window is
    # 0.7s (>= 1 more tick expected); B's own close (2.5s total) is not due
    # for a long while yet, so this cannot race B's teardown.
    my $before_b = comment_count($b_data, 'pingB');
    my $settle_deadline = Time::HiRes::time() + 0.7;
    while (Time::HiRes::time() < $settle_deadline) {
        exchange_frames($client, $client_sock, 1);
    }
    ok(comment_count($b_data, 'pingB') > $before_b,
        "B's keepalive kept ticking after A's stream closed")
        or diag("b_data: $b_data");

    # Drain: let B's own app coroutine reach its 2.5s close and return
    # naturally before tearing down, so no suspended async sub is abandoned
    # mid-await. Ceiling 6s is a wide margin over B's own budget.
    $deadline = Time::HiRes::time() + 6;
    while (Time::HiRes::time() < $deadline && !$closed{$b_id}) {
        exchange_frames($client, $client_sock, 1);
    }
    ok($closed{$b_id}, "B's own close completed before teardown");

    $stream_io->close_now;
    $loop->remove($server);
};

# ============================================================
# Per-stream SSE idle timeout independence (design section 11.3)
# ============================================================
# Before this task, sse_idle_timer lived on $self (the connection): only
# the FIRST stream's sse.start actually armed it (a `return if
# $self->{sse_idle_timer}` guard on the second), and on expiry it called
# the connection-wide _handle_disconnect_and_close, tearing down every
# multiplexed stream -- not just the idle one. This proves one idle stream
# closes on its own while an active sibling (which keeps resetting its own
# idle timer via sse.send) is unaffected.
my $active_done = 0;   # flips true once the '/active' app coroutine returns

subtest 'per-stream SSE idle timeout: an idle stream closes without killing an active sibling' => sub {
    $active_done = 0;
    my $app = async sub {
        my ($scope, $receive, $send) = @_;
        return unless $scope->{type} eq 'sse';

        await $receive->();
        await $send->({ type => 'sse.start', status => 200 });

        if ($scope->{path} eq '/idle') {
            # Sends nothing else -- only the per-stream idle timer (armed at
            # sse.start) can end this stream. Wait on receive() for the
            # eventual sse.disconnect it delivers, rather than a fixed
            # background delay that would outlive the test's own teardown
            # and get abandoned.
            await $receive->();
        }
        elsif ($scope->{path} eq '/active') {
            # First send immediately (no delay), so this stream already has
            # data well before the idle stream's independently-timed 0.3s
            # timeout can fire -- the two streams are dispatched moments
            # apart, so a delayed first send here would race that closure
            # instead of reliably preceding it. Then resets its own idle
            # timer on every further send (8 * 0.1s = 0.8s), comfortably
            # outliving the 0.3s idle timeout, but still short enough that
            # the coroutine returns (rather than being abandoned mid-await)
            # before this subtest's own teardown -- see $active_done below.
            await $send->({ type => 'sse.send', data => 'tick0' });
            for my $i (1 .. 8) {
                await $loop->delay_future(after => 0.1);
                await $send->({ type => 'sse.send', data => "tick$i" });
            }
            $active_done = 1;
        }
    };

    my $server = create_test_server(app => $app, sse_idle_timeout => 0.3);
    my ($conn, $stream_io, $client_sock) =
        create_h2c_connection(app => $app, server => $server);

    my ($idle_id, $active_id);
    my $active_data = '';
    my %closed;
    my $client = create_client(
        on_data_chunk_recv => sub {
            my ($sid, $data) = @_;
            $active_data .= $data if defined $active_id && $sid == $active_id;
            return 0;
        },
        on_stream_close => sub {
            my ($sid, $err) = @_;
            $closed{$sid} = 1;
            return 0;
        },
    );

    h2c_handshake($client, $client_sock);
    open_sse_stream_tracked($client, $client_sock, '/idle', \$idle_id);
    open_sse_stream_tracked($client, $client_sock, '/active', \$active_id);

    # Idle timeout is 0.3s and the idle stream sends nothing after
    # sse.start, so it should close at ~0.3s. Ceiling 5s is a >= 16x margin.
    my $deadline = Time::HiRes::time() + 5;
    while (Time::HiRes::time() < $deadline && !$closed{$idle_id}) {
        exchange_frames($client, $client_sock, 1);
    }
    ok($closed{$idle_id}, 'idle stream closed within the bounded window');

    # Independence: the active stream (resetting its OWN idle timer every
    # 0.1s) must not have been swept away by the idle stream's timeout --
    # that whole-connection blast radius is exactly the pre-fix behavior.
    ok(!$closed{$active_id}, 'active stream is still open when the idle stream closes');
    ok(length($active_data) > 0, 'active stream had already received data');

    # The active stream keeps receiving further data afterward too, proving
    # the connection (and the active stream's own scope) survived the idle
    # stream's teardown, not just the instant of closure. 0.3s is short
    # enough to land mid-loop (active's total budget is 0.8s from its own
    # dispatch, which starts only slightly after idle's).
    my $before = length($active_data);
    my $settle_deadline = Time::HiRes::time() + 0.3;
    while (Time::HiRes::time() < $settle_deadline) {
        exchange_frames($client, $client_sock, 1);
    }
    ok(length($active_data) > $before,
        'active stream continued receiving data after the sibling idle-timeout');

    # Drain: let the active app coroutine's bounded loop finish and return
    # naturally before tearing down, so no suspended async sub is abandoned
    # mid-await. Ceiling 5s is a wide margin over its ~0.8s own budget.
    my $drain_deadline = Time::HiRes::time() + 5;
    while (Time::HiRes::time() < $drain_deadline && !$active_done) {
        exchange_frames($client, $client_sock, 1);
    }
    ok($active_done, "active stream's app coroutine returned before teardown");

    $stream_io->close_now;
    $loop->remove($server);
};

done_testing;