PAGI-Server
view release on metacpan or search on metacpan
lib/PAGI/Server/Connection.pm view on Meta::CPAN
package PAGI::Server::Connection;
use strict;
use warnings;
our $VERSION = '0.002013';
use Future;
use Future::AsyncAwait;
use Scalar::Util qw(weaken refaddr);
use Protocol::WebSocket::Handshake::Server;
use Protocol::WebSocket::Frame;
use Digest::SHA qw(sha1_base64);
use Encode;
use URI::Escape qw(uri_unescape);
use IO::Async::Timer::Countdown;
use IO::Async::Timer::Periodic;
use Time::HiRes qw(gettimeofday tv_interval);
use PAGI::Server::AsyncFile;
use PAGI::Server::ConnectionState;
use PAGI::Server::TransportState;
use PAGI::Server::EventValidator;
use constant FILE_CHUNK_SIZE => 65536; # 64KB chunks for file streaming
# Per-second cache for CLF timestamp in access log (same pattern as HTTP1::format_date)
my $_cached_log_timestamp;
my $_cached_log_time = 0;
# =============================================================================
# Header Validation (CRLF Injection Prevention)
# =============================================================================
# RFC 7230 Section 3.2.6: Field values MUST NOT contain CR or LF
sub _validate_header_value { PAGI::Server::EventValidator::check_header_value($_[0]) }
sub _validate_header_name { PAGI::Server::EventValidator::check_header_name($_[0]) }
# =============================================================================
# HTTP/2 connection-specific header stripping (RFC 9113 section 8.2.2, design
# doc section 13.3)
# =============================================================================
# HTTP/2 forbids connection-specific header fields. An app-supplied
# connection, keep-alive, proxy-connection, transfer-encoding, or upgrade
# header -- or a te header carrying anything but the token 'trailers' --
# corrupts the response at the framing layer: the client receives only
# :status, with no body. HTTP/1.1 has no such prohibition, so this strip
# applies only to the HTTP/2 response paths that call it.
my %H2_CONNECTION_SPECIFIC_HEADER = map { $_ => 1 }
qw(connection keep-alive proxy-connection transfer-encoding upgrade te);
# Returns a new arrayref with connection-specific header pairs removed,
# warning once per stripped occurrence (not deduplicated by name -- two
# 'keep-alive' headers warn twice). Does not mutate $headers.
#
# $in_trailers (optional, default false) selects the trailer-block variant
# of this rule: RFC 9110 section 6.6.2 forbids every connection-specific
# field from a trailer section outright, so unlike a response's HEADERS
# block there is no 'te: trailers' carve-out inside a trailer block itself
# -- that carve-out is what lets a response ADVERTISE trailers are coming,
# not something a trailer block may then contain. A trailer-borne 'te'
# tuple is therefore stripped regardless of its value.
sub _h2_strip_connection_headers {
my $self = shift;
my ($headers, $in_trailers) = @_;
my $context = $in_trailers ? 'trailers' : 'response';
my @kept;
for my $h (@$headers) {
my ($name, $value) = @$h;
my $lc_name = lc $name;
if ($H2_CONNECTION_SPECIFIC_HEADER{$lc_name}) {
# RFC 9113 permits a response 'te' only with the exact token
# 'trailers'; a case-insensitive VALUE compare (not a name/list
# match) is what the token grammar calls for. OWS
# (leading/trailing whitespace) around the token is trimmed
# before the compare, per RFC 9110's field-value grammar -- a
# compound value like 'trailers, gzip' is not the bare token and
# is still stripped. This carve-out does not apply in a trailer
# block (see $in_trailers above).
if (!$in_trailers && $lc_name eq 'te') {
(my $v = lc $value) =~ s/^\s+|\s+\z//g;
if ($v eq 'trailers') {
# Submit the normalized token, never the original value:
# RFC 9113 8.2.1 forbids OWS in field values, and
lib/PAGI/Server/Connection.pm view on Meta::CPAN
my $body = $ss->{body};
$ss->{body} = '';
return {
type => 'http.request',
body => $body,
more => 0,
};
}
while (1) {
# Wait for body data (or, once the request has been fully
# delivered, for the stream to end)
if (!$ss->{body_pending}) {
$ss->{body_pending} = Future->new;
}
await $ss->{body_pending};
# Re-fetch stream state (may have changed)
$ss = $weak_self->{h2_streams}{$stream_id};
return { type => 'http.disconnect' } unless $ss;
# Check queue after waking -- a queued event wins over the
# body fallthrough (a close can set body_complete AND queue
# http.disconnect on the same wake)
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
# Terminal event already delivered: this wake brought
# nothing for the application -- park again rather than
# re-synthesize the final body event.
next if $ss->{final_request_delivered};
my $more = $ss->{body_complete} ? 0 : 1;
$ss->{final_request_delivered} = 1 unless $more;
my $body = $ss->{body};
$ss->{body} = '';
return {
type => 'http.request',
body => $body,
more => $more,
};
}
})->();
return $future;
};
}
sub _h2_create_send {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
my $status;
my @response_headers;
# Streaming state for deferred data provider pattern.
# The send queue lives on per-stream state ($ss->{send_queue} /
# $ss->{send_queue_bytes}) so the h2 transport handle can measure it;
# $eof_pending / $streaming_started stay closure-local.
my $eof_pending = 0;
my $streaming_started = 0;
# Mirrors the stream state's own starting point (see the seq_state =>
# 'initial' initializer in _h2_on_request); the two must stay in step.
my $seq = 'initial';
my $is_head = (($stream_state->{pseudo}{':method'} // '') eq 'HEAD');
# Set once, from http.response.start's own 'trailers' flag (section 6
# below), and never changed again. $data_callback's no_end computation
# MUST key off this rather than $ss->{seq_state}: the trailers arm
# advances that mirror to 'complete' as soon as the app's send() call
# is made, which can be BEFORE the data provider has actually drained
# (deferred submit, see below) -- keying off seq_state there raced the
# mirror update and let END_STREAM land back on the DATA frame.
my $trailers_declared = 0;
# Trailers-vs-data-provider handshake (design §8.3). Confirmed
# empirically: calling submit_trailer() BEFORE the data provider has
# actually handed nghttp2 its terminal (eof=1) chunk silently abandons
# any DATA nghttp2 has not yet pulled through $data_callback (observed
# under real per-stream flow control -- the still-queued tail of a
# file/fh body never reached the wire, yet the stream closed "cleanly"
# with the trailer -- a DEFERRED data-provider item is detached from
# nghttp2's own outbound queue, so an early trailing HEADERS orphans
# it). The Net::HTTP2::nghttp2 binding's own POD says submit_trailer()
# "can be called inside" the data-provider callback OR after it
# returns -- it does not say "at any later, unrelated time" -- so the
# invariant this handshake actually enforces is narrower and stricter:
# never BEFORE the provider has delivered its terminal EOF. The
# trailers arm below submits directly ONLY once $data_eof_delivered is
# already true (no further callback invocation will occur for this
# stream); otherwise it stages the headers here and PARKS the send()
# until the callback's own terminal invocation submits them.
#
# Contract note (flag for Task 6 / Compliance.pod): this means a
# trailers send() can now block for as long as the peer withholds
# flow-control window on a still-draining body -- new, unbounded-in-
# the-app's-view blocking that a pre-Task-4 (stub) reading of the spec
# would not have anticipated. This is arguably MORE correct, not a
# regression: trailers now participate in the same backpressure body
# sends already do, rather than racing ahead of undelivered DATA.
my $data_eof_delivered = 0;
my $pending_trailer_headers;
# Called by $data_callback at the exact point it hands nghttp2 the
# terminal (eof=1) chunk. If a trailers send() is waiting on this
# (staged $pending_trailer_headers), submits it HERE -- synchronously,
# from inside the data-provider callback, per nghttp2's own sanctioned
# pattern -- then wakes the waiting send() on the next loop tick (never
# resolve an app Future from inside a native nghttp2 callback -- same
# discipline as the drain waiters above).
my $deliver_trailer_eof = sub {
$data_eof_delivered = 1;
return unless defined $pending_trailer_headers;
my $headers = $pending_trailer_headers;
$pending_trailer_headers = undef;
my $ss2 = $weak_self && $weak_self->{h2_streams}{$stream_id};
# Dead-stream invariant, kept local here rather than inferred from
# the doomed-but-still-present carve-out pattern used across this
# file (h2_closed set, entry not yet reclaimed): don't reach
# nghttp2 a second time on a stream id it may already be tearing
# down. Two close paths, two mechanisms, same outcome -- there's
# nothing left to do here either way: _h2_on_close and the 413
# early-close branch set h2_closed on the still-present entry
# (caught by the check below), while a whole-connection _close
# deletes the h2_streams entry outright (so $ss2 comes back undef
# here) and releases trailer_wait itself via its own sweep. Either
# way, whichever close path is running has already released -- or
# is about to release -- trailer_wait.
return if $ss2 && $ss2->{h2_closed};
my $ok = eval {
$weak_self->{h2_session}->submit_trailer($stream_id, headers => $headers);
1;
};
my $err = $@;
my $f = $ss2 && delete $ss2->{trailer_wait};
return unless $f;
$weak_self->{server}->loop->later(sub {
return if $f->is_ready;
if ($ok) { $f->done(1) } else { $f->fail($err) }
});
};
# Data callback for nghttp2's streaming response.
# Returns ($data, $eof) when data is available, or undef to defer.
my $data_callback = sub {
my ($cb_stream_id, $max_len) = @_;
my $ss = $weak_self && $weak_self->{h2_streams}{$stream_id};
return undef unless $ss;
my $q = $ss->{send_queue} ||= [];
if (@$q) {
my $chunk = shift @$q;
# Respect max_len — XS truncates without preserving remainder
if (length($chunk) > $max_len) {
unshift @$q, substr($chunk, $max_len);
$chunk = substr($chunk, 0, $max_len);
}
$ss->{send_queue_bytes} -= length($chunk);
# Per-stream backpressure: once this stream's queue falls below the
# low watermark, release any producer blocked in
# _h2_wait_for_stream_drain. This callback runs inside nghttp2's
# extract(), so resolve on the next loop tick — completing the Future
# resumes the awaiting producer synchronously, and it must not call
# resume_stream/_h2_write_pending re-entrantly into nghttp2.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{stream_drain_waiters} && @{$ss->{stream_drain_waiters}}) {
my @waiters = splice @{$ss->{stream_drain_waiters}};
$weak_self->{server}->loop->later(sub {
$_->done for grep { !$_->is_ready } @waiters;
});
}
# Fire the app's on_drain hysteresis callbacks once this stream's
# queue falls below the low watermark. Like the waiters above, this
# runs inside nghttp2's extract(), and an on_drain callback may call
# $send to resume its source — which would re-enter nghttp2. Splice
# the fires out first (so they can't double-fire), then invoke them on
# the next loop tick.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{transport_drain_fires} && @{$ss->{transport_drain_fires}}) {
my @fires = splice @{$ss->{transport_drain_fires}};
$weak_self->{server}->loop->later(sub {
$_->() for @fires;
});
}
my $eof = (!@$q && $eof_pending) ? 1 : 0;
# Trailers declared: reserve END_STREAM for the trailing
# HEADERS block (design §8.3) instead of letting it land on
# this DATA frame. Has no effect unless $eof is also true
# (Net::HTTP2::nghttp2's own contract), so it is safe to key
# off $trailers_declared unconditionally here rather than the
# $ss->{seq_state} mirror, which the trailers arm may have
# already advanced past 'awaiting_trailers' by this point.
my $no_end = $trailers_declared ? 1 : 0;
$deliver_trailer_eof->() if $eof;
return ($chunk, $eof, $no_end);
}
# Queue empty but EOF pending — signal end of stream
if ($eof_pending) {
my $no_end = $trailers_declared ? 1 : 0;
$deliver_trailer_eof->();
return ('', 1, $no_end);
}
# Queue empty, more data expected — defer (NGHTTP2_ERR_DEFERRED in C layer)
return undef;
};
# Shared file/fh chunk pump: pushes produced chunks into this stream's
# send queue under the per-stream watermark, then marks EOF. The producer
# is an async sub that receives an async "emit" callback and must await it
# per chunk; emit dies with the sentinel below if the stream vanishes
# (client reset) so the pump stops reading without treating it as an error.
my $STREAM_GONE = "PAGI::h2 stream gone\n";
my $emit_chunk = async sub {
my ($chunk) = @_;
my $ss = $weak_self ? $weak_self->{h2_streams}{$stream_id} : undef;
die $STREAM_GONE unless $ss && !$ss->{h2_closed} && !$weak_self->{closed};
if (($ss->{send_queue_bytes} // 0) >= $weak_self->{write_high_watermark}) {
await $weak_self->_h2_wait_for_stream_drain($stream_id);
$ss = $weak_self ? $weak_self->{h2_streams}{$stream_id} : undef;
die $STREAM_GONE unless $ss && !$ss->{h2_closed} && !$weak_self->{closed};
}
if (length $chunk) {
push @{$ss->{send_queue}}, $chunk;
$ss->{send_queue_bytes} += length $chunk;
}
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
return;
};
# Shared tail for the file/fh arms below: mark EOF pending and resume the
# stream once the read loop finishes without error. A no-op if the stream
# vanished (client reset) while the last chunk was in flight.
my $finish_body_stream = sub {
$eof_pending = 1;
return unless $weak_self;
return unless $weak_self->{h2_streams}{$stream_id};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
};
# Shared preamble for the file/fh body arms below (extraction, not a
# redesign -- see the arms themselves for why each piece sits where it
# does). Split in two because the two pieces cross the arms' own
# eval boundary: the file arm must report an illegal-sequence error
# (from advance_http) ahead of a misleading "File not found" from its
# own -f/-r checks, but must NOT have submitted response headers to the
# client before those checks pass -- so the sequence advance runs
# BEFORE the arm's eval (a comment-preserving-only move: advance_http
# is a pure function, and a caught vs. uncaught throw here already
# behaved identically before this extraction, since $seq is simply
# never reassigned on a throw either way), while the streaming-start
# call must stay INSIDE the eval, at each arm's own correct position
# relative to its own pre-checks, so a failure there still rolls back
# via the shared tail below instead of leaking a submitted-but-broken
# response.
# Outside-eval half: snapshot+advance+mirror the sequence state ahead
# of either arm's own validity checks. Returns the pre-advance $seq so
# the tail helper below can roll back to it on failure.
my $advance_seq_for_body = sub {
my ($ss, $event) = @_;
my $seq_before = $seq;
$seq = PAGI::Server::EventValidator::advance_http($seq, $event);
$ss->{seq_state} = $seq if $ss;
return $seq_before;
};
# Inside-eval half: submit the streaming response exactly once, the
# first time either arm actually has a chunk ready to send. Each arm
# calls this at its own correct point (after its own pre-checks, so a
# failed check never causes headers to reach the client for a response
# that goes on to fail).
my $ensure_h2_streaming_started = sub {
my ($ss) = @_;
return if $streaming_started;
$streaming_started = 1;
$ss->{send_queue} //= []; $ss->{send_queue_bytes} //= 0;
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status, headers => \@response_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
};
# Shared failure-rollback-or-finish tail for the file/fh body arms:
# given the arm's own eval result, either rolls $seq back to its
# pre-event value and re-raises (recoverable per contract, unless the
# stream is simply gone -- a quiet no-op), or marks the body stream's
# EOF pending on success. Mirrors h1's D1 advance-then-rollback pattern.
my $finish_or_rollback_body_send = sub {
my ($ok, $err, $ss, $seq_before) = @_;
if (!$ok) {
return if $err eq $STREAM_GONE; # client reset: quiet no-op
$seq = $seq_before; # recoverable, per contract
$ss->{seq_state} = $seq if $ss;
die $err;
}
$finish_body_stream->();
};
return async sub {
my ($event) = @_;
return unless $weak_self;
my $type = $event->{type} // '';
# Once the machine has recorded the response complete, the machine
# decides what happens next -- 'complete' has no idempotent case, so
# any further send always raises -- not the stream-gone check below.
# h2_streams entries for a finished stream are reclaimed
# asynchronously by _h2_on_close, so by the time a post-complete send
# arrives $ss may already be gone.
my $already_closed = ($seq eq 'complete');
my $ss = $weak_self->{h2_streams}{$stream_id};
# A doomed-but-still-present entry (h2_closed set but not yet
# deleted -- see the 413-overrun branch in _h2_on_body) is treated
# the same as an absent one: both are post-close sends and must
# silently no-op, not reach nghttp2 a second time on this stream id.
return if (!$ss || $ss->{h2_closed}) && !$already_closed;
return if $weak_self->{closed} && !$already_closed;
# 1. Shape validation (mandatory)
PAGI::Server::EventValidator::validate_http_send(
$event, { extensions => $weak_self->{extensions} });
# 2. HEAD: the server suppresses the body (PAGI Www.pod "HEAD Requests").
# The app responds exactly as for GET; we discard payloads, never open
# file/fh, and accept-and-discard trailers. Sequence state still
# advances so the lifecycle (completion, post-complete raises) matches GET.
if ($is_head && ($type eq 'http.response.body' || $type eq 'http.response.trailers')) {
$seq = PAGI::Server::EventValidator::advance_http($seq, $event);
$ss->{seq_state} = $seq if $ss;
if ($seq eq 'complete' && !$ss->{h2_head_finished}) {
$ss->{h2_head_finished} = 1;
$weak_self->{h2_session}->submit_response($stream_id,
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# trailers at start. advance-then-rollback (h1 D1 pattern, mirrored
# from _create_send's chunked-framing check): the sequence machine
# and its $ss mirror advance FIRST, then the native submit is
# attempted (directly here, or via $deliver_trailer_eof above --
# see the closure-top comment for why). advance_http itself may
# croak here (trailers undeclared or body not yet complete) -- that
# propagates unrolled-back, same as validate_http_send's shape
# check above, since $seq was never reassigned in that case. If
# submit_trailer throws, both are rolled back to their pre-event
# value and the error propagates: the app's return then lands in
# the existing incomplete-response arm (RST NGHTTP2_INTERNAL_ERROR)
# -- the same machinery a dropped body chunk already relies on, not
# duplicated here.
if ($type eq 'http.response.trailers') {
my $seq_before = $seq;
$seq = PAGI::Server::EventValidator::advance_http($seq, $event);
$ss->{seq_state} = $seq if $ss;
# Empty/absent headers still calls submit_trailer with [] --
# trailers were declared and must still terminate the response.
my $trailer_headers = [
map { [_validate_header_name($_->[0]), _validate_header_value($_->[1])] }
@{ $event->{headers} // [] }
];
# RFC 9110 6.6.2 / design 13.3 -- an app-supplied connection,
# transfer-encoding, etc. in a trailer block corrupts it at the
# framing layer exactly like a response header: nghttp2 rejects
# the whole trailer HEADERS block, so the peer sees zero
# trailing HEADERS (on_stream_close never fires) even though
# submit_trailer below reports success. Strip before submission,
# same as every response-header path; $in_trailers=1 drops the
# response-only 'te: trailers' carve-out (RFC 9110 6.6.2 bans
# connection-specific fields from trailers outright).
$trailer_headers = $weak_self->_h2_strip_connection_headers($trailer_headers, 1);
my $ok;
if ($data_eof_delivered) {
# The data provider already handed nghttp2 its terminal
# chunk -- no further $data_callback invocation will occur
# for this stream, so it is safe to submit directly here.
$ok = eval {
$weak_self->{h2_session}->submit_trailer(
$stream_id, headers => $trailer_headers);
1;
};
} else {
# The data provider has not yet delivered EOF to nghttp2
# (still mid-transfer, possibly blocked on real per-stream
# flow control). Stage the headers for $deliver_trailer_eof
# to submit from inside the callback's own terminal
# invocation, and await that -- resolving this send() any
# earlier would let submit_trailer race ahead of not-yet-
# extracted DATA (see the closure-top comment). This PARKS
# the send() until the body actually finishes draining --
# e.g. until the peer grants enough flow-control window --
# which can be a real, unbounded wait from the app's point
# of view (closure-top comment has the full contract note).
$pending_trailer_headers = $trailer_headers;
my $f = $weak_self->{server}->loop->new_future;
$ss->{trailer_wait} = $f;
$weak_self->{h2_session}->resume_stream($stream_id) if $streaming_started;
$weak_self->_h2_write_pending;
# eval wraps the await (not just a plain assignment) so a
# failed $f (native rejection, via $deliver_trailer_eof's
# own $f->fail) is caught into $@ and folds into the same
# $ok/rollback handling the direct branch above uses.
$ok = eval { await $f; 1 };
# $f resolves DONE (not fail) on a whole-connection teardown
# too (_close's own sweep releases trailer_wait -- see
# _h2_resolve_stream_trailer_wait) -- the h2_closed carve-
# out's "successful no-op" contract, same discipline as the
# body arms' own post-drain-wait liveness re-check
# (:1616-1619 / :1635-1638 above). Without this, a
# connection that died while we awaited would leave
# $weak_self->{h2_session} undef and the shared tail below
# would call a method on it.
return unless $weak_self && !$weak_self->{closed} && $weak_self->{h2_session};
}
if (!$ok) {
my $err = $@;
$seq = $seq_before;
$ss->{seq_state} = $seq if $ss;
die $err;
}
# Stream-existence re-check, same discipline as every sibling
# await site's own $ss re-fetch (e.g. $emit_chunk above): the
# connection-liveness check at :1524 only covers the deferred
# branch and only the connection, not this stream specifically.
# Placed AFTER the rollback above (not folded into :1524) so a
# deferred native failure still reaches its rollback+die even if
# the stream looks gone by the time we get here -- only the
# success-path tail below, which is about to call resume_stream/
# _h2_write_pending on this stream id again, needs the guard.
# A gone stream here is the h2_closed carve-out's ordinary
# "successful no-op" case: the trailers already landed (ok=1)
# before the stream went away, so returning quietly is correct,
# not a swallowed error.
$ss = $weak_self ? $weak_self->{h2_streams}{$stream_id} : undef;
return if !$ss || $ss->{h2_closed};
# Flush any still-queued DATA before the trailing HEADERS --
# nghttp2 orders trailers after queued data on its own, but the
# stream must be resumed if the data callback had deferred.
$weak_self->{h2_session}->resume_stream($stream_id) if $streaming_started;
$weak_self->_h2_write_pending;
return;
}
# 4. file body: streamed through the send queue via $emit_chunk, one
# chunk at a time, under this stream's own backpressure. The type
# guard (not just "defined $event->{file}") keeps a nonconforming
# http.response.start carrying a stray 'file' key from being
# misrouted into this arm.
if ($type eq 'http.response.body' && defined $event->{file}) {
my $file = $event->{file};
my $offset = $event->{offset} // 0;
my $length = $event->{length};
# Snapshot+advance BEFORE the -f/-r checks (and the checks live
# inside the eval below) so a file event arriving in an illegal
# sequence state reports the SEQUENCE error, not a misleading
# "File not found" -- h1 parity. A failed file send must NOT
# mark the response complete, so on failure $seq rolls back to
# $seq_before exactly as the h1 twin in _create_send does.
my $seq_before = $advance_seq_for_body->($ss, $event);
my $ok = eval {
die "File not found: $file\n" unless -f $file;
die "Cannot read file: $file\n" unless -r $file;
$ensure_h2_streaming_started->($ss);
# Mirror h1's effective-length computation (_send_file_response)
# so the sync-vs-async choice below sees the same number h1
# would, including the Www.pod rule that an offset past EOF
# clamps to zero bytes rather than failing.
my $file_size = -s $file;
die "Cannot stat file $file: $!\n" unless defined $file_size;
my $effective_length = $length // ($file_size - $offset);
$effective_length = 0 if $effective_length < 0;
if ($weak_self->{sync_file_threshold} > 0
&& $effective_length <= $weak_self->{sync_file_threshold}) {
# Small-file fast path (parity with h1's sync_file_threshold):
# a file at or under the threshold is read synchronously,
# in-process, as ONE queue chunk -- avoiding a worker-pool
# round trip for a body small enough that it costs more
# than it saves. The no-slurp rule targets large files.
open my $fh, '<:raw', $file or die "Cannot open file $file: $!\n";
seek($fh, $offset, 0) if $offset;
my $bytes_read = read($fh, my $data, $effective_length);
die "Failed to read file $file: $!\n" unless defined $bytes_read;
close $fh;
await $emit_chunk->($data);
}
else {
my $loop = $weak_self->{server}->loop;
await PAGI::Server::AsyncFile->read_file_chunked(
$loop, $file, $emit_chunk,
offset => $offset,
(defined $length ? (length => $length) : ()),
chunk_size => FILE_CHUNK_SIZE,
);
}
1;
};
$finish_or_rollback_body_send->($ok, $@, $ss, $seq_before);
return;
}
# 5. fh body: streamed through the send queue via $emit_chunk, one
# chunk at a time, under this stream's own backpressure. The
# application owns $fh (opened before the send, closed or not by the
# app afterward) -- the server never closes it. Mirrors the file arm
# above; the read loop is adapted from h1's _send_fh_response. The
# type guard mirrors the file arm's M1 fix above.
if ($type eq 'http.response.body' && defined $event->{fh}) {
my $fh = $event->{fh};
my $seq_before = $advance_seq_for_body->($ss, $event);
my $ok = eval {
$ensure_h2_streaming_started->($ss);
if (my $off = $event->{offset}) {
seek($fh, $off, 0) or die "Cannot seek: $!\n";
}
my $remaining = $event->{length};
while (1) {
my $to_read = FILE_CHUNK_SIZE;
if (defined $remaining) {
$to_read = $remaining if $remaining < $to_read;
last if $to_read <= 0;
}
# The bare block scopes only the 'closed' warning
# suppression around read() -- a bare block is itself a
# one-iteration loop, so die/last/await must live outside
# it or 'last' would only exit the block, not this while.
my ($bytes_read, $chunk);
{ no warnings 'closed';
$bytes_read = read($fh, $chunk, $to_read);
}
die "Failed to read filehandle: $!\n" unless defined $bytes_read;
last if $bytes_read == 0;
await $emit_chunk->($chunk);
$remaining -= $bytes_read if defined $remaining;
}
1;
};
$finish_or_rollback_body_send->($ok, $@, $ss, $seq_before);
return;
}
# 6. Sequence enforcement (start / plain body / fullflush). advance_http
# is deliberately called from four places in this sub — the HEAD
# block above, the file arm, the fh arm, and here — each has
# different pre/post-state needs; do not consolidate them.
$seq = PAGI::Server::EventValidator::advance_http($seq, $event);
$ss->{seq_state} = $seq if $ss;
if ($type eq 'http.response.start') {
$ss->{response_started} = 1;
$ss->{connection_state}->_mark_response_started if $ss->{connection_state};
$trailers_declared = 1 if $event->{trailers};
$status = $event->{status} // 200;
@response_headers = map {
[_validate_header_name($_->[0]), _validate_header_value($_->[1])]
} @{$event->{headers} // []};
# RFC 9113 8.2.2 / design 13.3 — strips app-supplied connection,
# transfer-encoding, etc. before this list reaches nghttp2 (also
# covers the HEAD path below, which submits this same array).
@response_headers = @{ $weak_self->_h2_strip_connection_headers(\@response_headers) };
# Server-supplied Date header (HTTP/1.1 parity) — add if the app didn't.
unless (grep { lc($_->[0]) eq 'date' } @response_headers) {
push @response_headers, ['date', $weak_self->{protocol}->format_date];
}
}
elsif ($type eq 'http.response.body') {
my $body = $event->{body} // '';
my $more = $event->{more} // 0;
if ($more) {
if (!$streaming_started) {
# First streaming chunk — submit with data callback
$streaming_started = 1;
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
# Synchronous: we're in the app's send path (not nghttp2's
# extract), so on_high_water can fire here to tell the app to
# pause its source.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status,
headers => \@response_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
} else {
# Subsequent chunk — backpressure check then push and resume.
# Bound on THIS stream's send queue (per-stream), not the
# shared TCP buffer which is meaningless across multiplexed
# streams.
if (($ss->{send_queue_bytes} // 0) >= $weak_self->{write_high_watermark}) {
await $weak_self->_h2_wait_for_stream_drain($stream_id);
return unless $weak_self;
return if $weak_self->{closed};
return unless $weak_self->{h2_streams}{$stream_id};
}
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
# Synchronous — app send path, not nghttp2 extract.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
}
} else {
if ($streaming_started) {
# Final chunk on an already-streaming response. Bound on THIS
# stream's send queue (per-stream), not the shared TCP buffer.
if (($ss->{send_queue_bytes} // 0) >= $weak_self->{write_high_watermark}) {
await $weak_self->_h2_wait_for_stream_drain($stream_id);
return unless $weak_self;
return if $weak_self->{closed};
return unless $weak_self->{h2_streams}{$stream_id};
}
$eof_pending = 1;
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
# Synchronous — app send path, not nghttp2 extract.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
} elsif ($seq eq 'awaiting_trailers') {
# Trailers were declared and this is the (single-shot)
# terminal body event: a plain submit_response would set
# END_STREAM immediately, ending the stream before the
# trailers arrive -- design §8.3 forbids that. Route
# through the streaming path instead, mirroring the
# file/fh arms' $streaming_started idiom; the data
# callback above already reserves END_STREAM for the
# trailing HEADERS block once it observes
# 'awaiting_trailers'.
$streaming_started = 1;
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
$eof_pending = 1;
if (length $body) {
push @{$ss->{send_queue}}, $body;
$ss->{send_queue_bytes} += length $body;
}
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status,
headers => \@response_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
} else {
# Non-streaming: single response (unchanged one-shot path)
$weak_self->{h2_session}->submit_response($stream_id,
status => $status,
headers => \@response_headers,
body => $body,
);
$weak_self->_h2_write_pending;
}
}
}
elsif ($type eq 'http.fullflush') {
# Hand any pending frames to the session's write path (design §8.4).
$weak_self->{h2_session}->resume_stream($stream_id) if $streaming_started;
$weak_self->_h2_write_pending;
}
};
}
# =============================================================================
# HTTP/2 WebSocket over HTTP/2 (RFC 8441)
# =============================================================================
sub _h2_create_websocket_scope {
my ($self, $stream_id, $stream_state) = @_;
my $pseudo = $stream_state->{pseudo};
my $headers = $stream_state->{headers};
my $full_path = $pseudo->{':path'} // '/';
my ($path, $query_string) = split(/\?/, $full_path, 2);
$query_string //= '';
# Match HTTP/1.1 pipeline: URI::Escape + UTF-8 decode with fallback
my $unescaped = uri_unescape($path);
my $decoded_path = eval { decode('UTF-8', $unescaped, Encode::FB_CROAK) }
// $unescaped;
# Extract subprotocols from headers
my @subprotocols;
for my $header (@$headers) {
my ($name, $value) = @$header;
if ($name eq 'sec-websocket-protocol') {
push @subprotocols, map { s/^\s+|\s+$//gr } split /,/, $value;
}
}
return {
type => 'websocket',
pagi => {
version => '0.5',
spec_version => '0.5',
},
http_version => '2',
scheme => $self->_get_ws_scheme,
path => $decoded_path,
raw_path => $path,
query_string => $query_string,
root_path => '',
headers => $headers,
(defined $self->{client_host}
? (client => [$self->{client_host}, $self->{client_port}])
: ()
),
server => [$self->{server_host}, $self->{server_port}],
subprotocols => \@subprotocols,
state => keys %{$self->{state}} ? { %{$self->{state}} } : {},
extensions => do {
my %ext = (%{$self->_get_extensions_for_scope}, 'websocket.http.response' => {});
# fullflush has no validate_websocket_send arm; advertising it here
# would lie to the app (design 13.2).
delete $ext{fullflush};
\%ext;
},
# max_frame_size: omitted when unenforced (max_ws_frame_size 0/undef
# means unlimited, per Protocol::WebSocket::Frame's max_payload_size
# semantics -- a server that does not enforce a cap must not
# advertise one). max_receive_queue has no unlimited mode (a hard,
# always-enforced cap), so it is always present.
($self->{max_ws_frame_size}
? (max_frame_size => $self->{max_ws_frame_size})
: ()
),
max_receive_queue => $self->{max_receive_queue},
# Per-stream outbound flow-control handle. Like the h2 sse/streaming
# scopes, it measures THIS stream's send queue (h2 multiplexes many
# streams over one connection, so the shared TCP buffer is
# meaningless per stream). Gives WebSocket-over-h2 the same
# pagi.transport surface HTTP/1.1 WebSocket already provides.
'pagi.transport' => ($stream_state->{transport_state} = $self->_h2_transport_state($stream_state)),
};
}
sub _h2_create_websocket_receive {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
# Fallback disconnect for a receive() that resolves after this stream is
# gone or the connection is closed: 1006/'' by default (RFC 6455
# abnormal closure, no reason available), but prefers the per-stream
# server_close_reason token when the stream's own state is still
# reachable in h2_streams -- a server-initiated teardown (idle timeout,
# keepalive timeout, ...) records that token there before tearing the
# stream down, so a receive() racing that teardown still reports why.
my $fallback_disconnect = sub {
my $ss = $weak_self && $weak_self->{h2_streams}{$stream_id};
return {
type => 'websocket.disconnect',
code => 1006,
reason => ($ss && $ss->{server_close_reason}) // '',
};
};
return sub {
return Future->done($fallback_disconnect->())
unless $weak_self;
return Future->done($fallback_disconnect->())
if $weak_self->{closed};
my $ss = $weak_self->{h2_streams}{$stream_id};
return Future->done($fallback_disconnect->())
unless $ss;
my $future = (async sub {
return $fallback_disconnect->()
unless $weak_self;
my $ss = $weak_self->{h2_streams}{$stream_id};
return $fallback_disconnect->()
unless $ss;
# Check queue first
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
# First call returns websocket.connect
if (!$ss->{ws_connect_sent}) {
$ss->{ws_connect_sent} = 1;
return { type => 'websocket.connect' };
}
# Wait for events
while (1) {
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
return $fallback_disconnect->()
if $weak_self->{closed};
if (!$ss->{body_pending}) {
$ss->{body_pending} = Future->new;
}
await $ss->{body_pending};
$ss = $weak_self->{h2_streams}{$stream_id};
return $fallback_disconnect->()
unless $ss;
}
})->();
return $future;
};
}
sub _h2_create_websocket_send {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
my $seq = 'connecting';
# Data callback for nghttp2's streaming response (the same pull-based
# data-provider model _h2_create_send/_h2_create_sse_send already use).
# Pulls raw WS frame bytes from the per-stream queue -- app messages,
# protocol replies (pong, close-echo), and this stream's own keepalive
# ping are all pushed there in call order, so FIFO ordering on the wire
# is preserved exactly as it was under direct submit_data calls.
#
# $ss->{ws_eof_pending} lives on the stream state (not a closure-local,
# unlike the http streaming callback's $eof_pending) because it is set
# from other subs entirely -- _h2_ws_close and the close-frame arm of
# _h2_process_ws_frames -- not just from this closure. When set, it
# merges END_STREAM onto the LAST queued chunk (the close frame itself)
# rather than emitting a separate empty terminal frame, matching the
# previous submit_data($id, $close_frame_bytes, 1) behavior exactly.
my $data_callback = sub {
my ($cb_stream_id, $max_len) = @_;
my $ss = $weak_self && $weak_self->{h2_streams}{$stream_id};
return undef unless $ss;
my $q = $ss->{send_queue} ||= [];
if (@$q) {
my $chunk = shift @$q;
# Respect max_len — XS truncates without preserving remainder
if (length($chunk) > $max_len) {
unshift @$q, substr($chunk, $max_len);
$chunk = substr($chunk, 0, $max_len);
}
$ss->{send_queue_bytes} -= length($chunk);
# Per-stream backpressure: once this stream's queue falls below the
# low watermark, release any producer blocked in
# _h2_wait_for_stream_drain. This runs inside nghttp2's extract(), so
# resolve on the next loop tick — completing the Future resumes the
# awaiting producer synchronously, and it must not re-enter nghttp2.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{stream_drain_waiters} && @{$ss->{stream_drain_waiters}}) {
my @waiters = splice @{$ss->{stream_drain_waiters}};
$weak_self->{server}->loop->later(sub {
$_->done for grep { !$_->is_ready } @waiters;
});
}
# Fire the app's on_drain hysteresis callbacks once this stream's
# queue falls below the low watermark. Deferred for the same reason:
# an on_drain callback may call $send, which would re-enter nghttp2.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{transport_drain_fires} && @{$ss->{transport_drain_fires}}) {
my @fires = splice @{$ss->{transport_drain_fires}};
$weak_self->{server}->loop->later(sub {
$_->() for @fires;
});
}
my $eof = (!@$q && $ss->{ws_eof_pending}) ? 1 : 0;
return ($chunk, $eof);
}
# Queue empty but EOF pending (a close frame already delivered as
# the terminal chunk above) — signal end of stream.
return ('', 1) if $ss->{ws_eof_pending};
# Queue empty, more data expected — defer (NGHTTP2_ERR_DEFERRED in C layer)
return undef;
};
return async sub {
my ($event) = @_;
return unless $weak_self;
my $type = $event->{type} // '';
# Once the machine has recorded the connection closed or the denial
# response complete, the machine decides what happens next -- both
# 'closed' and 'denial_complete' have no idempotent case, so any
# further send always raises -- not the stream-gone check below.
# h2_streams entries for a finished stream are reclaimed
# asynchronously by _h2_on_close (websocket.close itself triggers
# this via END_STREAM, same mechanism as the h2 HTTP/SSE closures),
# so by the time a post-close send arrives $ss may already be gone.
my $already_closed = ($seq eq 'closed' || $seq eq 'denial_complete');
my $ss = $weak_self->{h2_streams}{$stream_id};
# A doomed-but-still-present entry (h2_closed set but not yet
# deleted -- see the 413-overrun branch in _h2_on_body) is treated
# the same as an absent one: both are post-close sends and must
# silently no-op, not reach nghttp2 a second time on this stream id.
return if (!$ss || $ss->{h2_closed}) && !$already_closed;
return if $weak_self->{closed} && !$already_closed;
# websocket.http.response is always available on this path (the
# scope advertises it unconditionally; see
# _h2_create_websocket_scope), unlike connection-level extensions
# such as fullflush.
PAGI::Server::EventValidator::validate_websocket_send(
$event, { extensions => { %{$weak_self->{extensions}}, 'websocket.http.response' => {} } });
$seq = PAGI::Server::EventValidator::advance_websocket($seq, $event);
if ($type eq 'websocket.accept') {
# A duplicate accept is already rejected by advance_websocket.
# HTTP/2 WebSocket: respond with 200 (not 101)
my @headers;
if (my $subprotocol = $event->{subprotocol}) {
$subprotocol = _validate_subprotocol($subprotocol);
push @headers, ['sec-websocket-protocol', $subprotocol];
}
if (my $extra = $event->{headers}) {
push @headers, map {
[_validate_header_name($_->[0]), _validate_header_value($_->[1])]
} @$extra;
}
# RFC 9113 8.2.2 / design 13.3 — strip app-supplied connection,
# transfer-encoding, etc. before submission.
@headers = @{ $weak_self->_h2_strip_connection_headers(\@headers) };
$ss->{ws_accepted} = 1;
$ss->{response_started} = 1;
$ss->{ws_frame} = Protocol::WebSocket::Frame->new(
max_payload_size => $weak_self->{max_ws_frame_size},
);
# Submit 200 response with a pull-based data provider (same
# model as h2 streaming/SSE): frames are pushed onto
# $ss->{send_queue} and pulled by $data_callback as nghttp2's
# per-stream flow-control window allows.
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
$weak_self->{h2_session}->submit_response_streaming($stream_id,
status => 200,
headers => \@headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
# Process any data that arrived before accept
if (length($ss->{body}) > 0) {
my $buffered = $ss->{body};
$ss->{body} = '';
$weak_self->_h2_process_ws_frames($stream_id, $ss, $buffered);
}
}
elsif ($type eq 'websocket.send') {
return unless $ss->{ws_accepted};
my $frame;
if (defined $event->{text}) {
$frame = Protocol::WebSocket::Frame->new(
buffer => $event->{text},
type => 'text',
);
}
elsif (defined $event->{bytes}) {
$frame = Protocol::WebSocket::Frame->new(
buffer => $event->{bytes},
type => 'binary',
);
}
else {
return;
}
my $bytes = $frame->to_bytes;
# Per-stream backpressure: bound on THIS stream's queue, not the
# shared TCP buffer (meaningless across multiplexed h2 streams).
if (($ss->{send_queue_bytes} // 0) >= $weak_self->{write_high_watermark}) {
await $weak_self->_h2_wait_for_stream_drain($stream_id);
return unless $weak_self;
return if $weak_self->{closed};
# Refetch (same idiom as emit_chunk, :1374-1375): a stream
# can close while this send was parked.
$ss = $weak_self->{h2_streams}{$stream_id};
return unless $ss;
return if $ss->{h2_closed};
# This send can wake AFTER this stream's close frame was
# already queued (window opened below the low watermark
# while ws_eof_pending was set) -- pushing app data now would
# land it BEHIND the close frame: Close would ship without
# END_STREAM and a Text/Binary frame would follow it,
# violating RFC 6455 5.5.1. Same post-close no-op contract
# as the top-of-closure $already_closed check.
return if $ss->{ws_eof_pending};
}
push @{$ss->{send_queue}}, $bytes;
$ss->{send_queue_bytes} = ($ss->{send_queue_bytes} // 0) + length $bytes;
# Synchronous — app send path, not nghttp2 extract — so on_high_water
# may fire here to tell the app to pause its source.
lib/PAGI/Server/Connection.pm view on Meta::CPAN
$weak_self->_h2_write_pending;
}
elsif ($type eq 'websocket.keepalive') {
return unless $ss->{ws_accepted};
my $interval = $event->{interval} // 0;
my $timeout = $event->{timeout};
if ($interval > 0) {
$weak_self->_h2_start_ws_keepalive($stream_id, $ss, $interval, $timeout);
}
else {
$weak_self->_h2_stop_ws_keepalive($ss);
}
}
return;
};
}
# =============================================================================
# HTTP/2 SSE (Server-Sent Events over HTTP/2)
# =============================================================================
sub _h2_create_sse_scope {
my ($self, $stream_id, $stream_state) = @_;
my $pseudo = $stream_state->{pseudo};
my $headers = $stream_state->{headers};
my $full_path = $pseudo->{':path'} // '/';
my ($path, $query_string) = split(/\?/, $full_path, 2);
$query_string //= '';
# Match HTTP/1.1 pipeline: URI::Escape + UTF-8 decode with fallback
my $unescaped = uri_unescape($path);
my $decoded_path = eval { decode('UTF-8', $unescaped, Encode::FB_CROAK) }
// $unescaped;
return {
type => 'sse',
pagi => {
version => '0.5',
spec_version => '0.5',
},
http_version => '2',
method => $pseudo->{':method'} // 'GET',
scheme => $pseudo->{':scheme'} // $self->_get_scheme,
path => $decoded_path,
raw_path => $path,
query_string => $query_string,
root_path => '',
headers => $headers,
(defined $self->{client_host}
? (client => [$self->{client_host}, $self->{client_port}])
: ()
),
server => [$self->{server_host}, $self->{server_port}],
state => keys %{$self->{state}} ? { %{$self->{state}} } : {},
extensions => $self->_get_extensions_for_scope,
# Per-stream outbound flow-control handle. Like the h2 streaming scope,
# it measures THIS stream's send queue (h2 multiplexes many streams over
# one connection, so the shared TCP buffer is meaningless per stream).
'pagi.transport' => ($stream_state->{transport_state} = $self->_h2_transport_state($stream_state)),
};
}
sub _h2_create_sse_receive {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
my $sse_disconnect = sub {
return {
type => 'sse.disconnect',
reason => 'client_closed',
};
};
return sub {
return Future->done($sse_disconnect->()) unless $weak_self;
return Future->done($sse_disconnect->()) if $weak_self->{closed};
my $ss = $weak_self->{h2_streams}{$stream_id};
return Future->done($sse_disconnect->()) unless $ss;
my $future = (async sub {
return $sse_disconnect->() unless $weak_self;
my $ss = $weak_self->{h2_streams}{$stream_id};
return $sse_disconnect->() unless $ss;
# Check queue first
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
# First call returns sse.request once the full body has arrived.
# A POST body still streaming across DATA frames must not be
# truncated behind a truthful more=>0 -- wait for body_complete
# (set on END_STREAM or stream close, see _h2_on_body/_h2_on_close)
# before delivering it as a single terminal event. This is the
# smaller change relative to reworking sse.request into a
# truthful multi-chunk stream: the wire shape here was already
# one-shot, so completing that contract fixes the dispatch-timing
# bug (design section 11.1) without touching the event shape.
if (!$ss->{sse_request_sent}) {
while (!$ss->{body_complete}) {
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
return $sse_disconnect->() if $weak_self->{closed};
if (!$ss->{body_pending}) {
$ss->{body_pending} = Future->new;
}
await $ss->{body_pending};
$ss = $weak_self->{h2_streams}{$stream_id};
return $sse_disconnect->() unless $ss;
}
# body_complete can flip true on the very wake that also
# queued a terminal event -- e.g. _h2_on_close sets
# body_complete AND pushes sse.disconnect before waking
# body_pending. That queued event must win over delivering
# a (possibly truncated) body: the loop's own queue check
# only runs at the top of an iteration, so a queue entry
# that arrives on the wake that also satisfies the while
# condition is never seen there.
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
$ss->{sse_request_sent} = 1;
return {
type => 'sse.request',
body => $ss->{body},
more => 0,
};
}
# Wait for disconnect
while (1) {
if (@{$ss->{receive_queue}}) {
return shift @{$ss->{receive_queue}};
}
return $sse_disconnect->()
if $weak_self->{closed};
if (!$ss->{body_pending}) {
$ss->{body_pending} = Future->new;
}
await $ss->{body_pending};
$ss = $weak_self->{h2_streams}{$stream_id};
return $sse_disconnect->() unless $ss;
}
})->();
return $future;
};
}
sub _h2_create_sse_send {
my ($self, $stream_id, $stream_state) = @_;
weaken(my $weak_self = $self);
my $seq = 'initial';
# Streaming state for the data-provider pull pattern. The send queue lives on
# per-stream state ($ss->{send_queue} / $ss->{send_queue_bytes}) so the
# pagi.transport handle can measure THIS stream's backlog. $streaming_started
# stays closure-local.
my $streaming_started = 0;
# Data callback for nghttp2's streaming response. Pulls from the per-stream
# queue; SSE responses stay open, so this never signals EOF (returns eof=0),
# or undef to defer when the queue is empty.
my $data_callback = sub {
my ($cb_stream_id, $max_len) = @_;
my $ss = $weak_self && $weak_self->{h2_streams}{$stream_id};
return undef unless $ss;
my $q = $ss->{send_queue} ||= [];
if (@$q) {
my $chunk = shift @$q;
# Respect max_len — XS truncates without preserving remainder
if (length($chunk) > $max_len) {
unshift @$q, substr($chunk, $max_len);
$chunk = substr($chunk, 0, $max_len);
}
$ss->{send_queue_bytes} -= length($chunk);
# Per-stream backpressure: once this stream's queue falls below the
# low watermark, release any producer blocked in
# _h2_wait_for_stream_drain. This runs inside nghttp2's extract(), so
# resolve on the next loop tick — completing the Future resumes the
# awaiting producer synchronously, and it must not re-enter nghttp2.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{stream_drain_waiters} && @{$ss->{stream_drain_waiters}}) {
my @waiters = splice @{$ss->{stream_drain_waiters}};
$weak_self->{server}->loop->later(sub {
$_->done for grep { !$_->is_ready } @waiters;
});
}
# Fire the app's on_drain hysteresis callbacks once this stream's
# queue falls below the low watermark. Deferred for the same reason:
# an on_drain callback may call $send, which would re-enter nghttp2.
if (($ss->{send_queue_bytes} // 0) < $weak_self->{write_low_watermark}
&& $ss->{transport_drain_fires} && @{$ss->{transport_drain_fires}}) {
my @fires = splice @{$ss->{transport_drain_fires}};
$weak_self->{server}->loop->later(sub {
$_->() for @fires;
});
}
return ($chunk, 0); # SSE streams never EOF via data_callback
}
# Queue empty. If the application closed this stream (sse.close), emit a
# final empty DATA frame with END_STREAM to terminate it; otherwise defer.
return ('', 1) if $ss->{sse_closing};
# Queue empty — defer (NGHTTP2_ERR_DEFERRED in the C layer)
return undef;
};
return async sub {
my ($event) = @_;
return unless $weak_self;
my $type = $event->{type} // '';
# Once the machine has already recorded this stream as closed --
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# anything else (decline_complete has no idempotent case; it croaks
# unconditionally). h2_streams entries for a closed stream are
# reclaimed asynchronously by _h2_on_close, so by the time a
# post-close send arrives $ss may already be gone.
my $already_closed = ($seq eq 'closed' || $seq eq 'decline_complete');
my $ss = $weak_self->{h2_streams}{$stream_id};
# A doomed-but-still-present entry (h2_closed set but not yet
# deleted -- see the 413-overrun branch in _h2_on_body) is treated
# the same as an absent one: both are post-close sends and must
# silently no-op, not reach nghttp2 a second time on this stream id.
return if (!$ss || $ss->{h2_closed}) && !$already_closed;
return if $weak_self->{closed} && !$already_closed;
# Reset THIS stream's SSE idle timer on send activity (skip once fully closed)
$weak_self->_h2_reset_sse_idle_timer($ss) unless $already_closed;
# Mandatory event validation and sequencing (PAGI spec compliance).
PAGI::Server::EventValidator::validate_sse_send(
$event, { extensions => $weak_self->{extensions} });
$seq = PAGI::Server::EventValidator::advance_sse($seq, $event);
if ($type eq 'sse.start') {
return if $ss->{response_started};
$ss->{response_started} = 1;
my $status = $event->{status} // 200;
my $headers = $event->{headers} // [];
# Ensure Content-Type is text/event-stream
my $has_content_type = 0;
for my $h (@$headers) {
if (lc($h->[0]) eq 'content-type') {
$has_content_type = 1;
last;
}
}
my @final_headers;
for my $h (@$headers) {
push @final_headers, [_validate_header_name($h->[0]), _validate_header_value($h->[1])];
}
# RFC 9113 8.2.2 / design 13.3 — strip app-supplied connection,
# transfer-encoding, etc. before submission.
@final_headers = @{ $weak_self->_h2_strip_connection_headers(\@final_headers) };
if (!$has_content_type) {
push @final_headers, ['content-type', 'text/event-stream'];
}
# Cache-Control and Date: server-supplied only when the app didn't
# supply them (design doc section 11.4).
unless (grep { lc($_->[0]) eq 'cache-control' } @final_headers) {
push @final_headers, ['cache-control', 'no-cache'];
}
# Server-supplied Date header (HTTP/1.1 parity) — the h1 SSE path adds
# this too; add it unless the app supplied one.
unless (grep { lc($_->[0]) eq 'date' } @final_headers) {
push @final_headers, ['date', $weak_self->{protocol}->format_date];
}
$streaming_started = 1;
$ss->{send_queue} //= [];
$ss->{send_queue_bytes} //= 0;
$weak_self->{h2_session}->submit_response_streaming(
$stream_id,
status => $status,
headers => \@final_headers,
data_callback => $data_callback,
);
$weak_self->_h2_write_pending;
# Protocol-specific keepalive writer (HTTP/2 DATA frames), scoped to
# THIS stream (design section 11.3): a second multiplexed SSE stream
# must not steal or replace this one's writer. Keepalive bytes are
# counted in the per-stream backlog so buffered_amount stays
# accurate, but they do not poke the watermark callbacks — a server
# heartbeat is not an application send.
$ss->{sse_ka_writer} = sub {
my ($text) = @_;
return unless $weak_self;
return if $weak_self->{closed};
my $ss = $weak_self->{h2_streams}{$stream_id} or return;
# PAGI Www.pod "Send SSE": encode to UTF-8 exactly once, at
# the wire boundary — all queue-length math below is on the
# resulting BYTE string.
my $bytes = eval { Encode::encode('UTF-8', $text, Encode::FB_CROAK) };
die "sse payload is not encodable as UTF-8: $@" unless defined $bytes;
push @{$ss->{send_queue} ||= []}, $bytes;
$ss->{send_queue_bytes} = ($ss->{send_queue_bytes} // 0) + length $bytes;
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
};
# Start THIS stream's SSE idle timer if configured
$weak_self->_h2_start_sse_idle_timer($stream_id, $ss);
}
elsif ($type eq 'sse.send') {
return unless $ss->{response_started};
# Per-stream backpressure: bound on THIS stream's queue, not the
# shared TCP buffer (meaningless across multiplexed h2 streams).
if (($ss->{send_queue_bytes} // 0) >= $weak_self->{write_high_watermark}) {
await $weak_self->_h2_wait_for_stream_drain($stream_id);
return unless $weak_self;
return if $weak_self->{closed};
return unless $weak_self->{h2_streams}{$stream_id};
}
my $sse_data = _format_sse_event($event);
# PAGI Www.pod "Send SSE": encode to UTF-8 exactly once, at the
# wire boundary — a failed encode fails this send's Future.
my $bytes = eval { Encode::encode('UTF-8', $sse_data, Encode::FB_CROAK) };
die "sse payload is not encodable as UTF-8: $@" unless defined $bytes;
push @{$ss->{send_queue} ||= []}, $bytes;
$ss->{send_queue_bytes} = ($ss->{send_queue_bytes} // 0) + length $bytes;
# Synchronous — app send path, not nghttp2 extract — so on_high_water
# may fire here to tell the app to pause its source.
$ss->{transport_state}->_check_watermarks if $ss->{transport_state};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
}
elsif ($type eq 'sse.comment') {
return unless $ss->{response_started};
lib/PAGI/Server/Connection.pm view on Meta::CPAN
# already been reclaimed (see the already-closed comment above)
# is the idempotent no-op advance_sse just allowed; nothing left
# to do.
return unless $ss;
# This stream is ending -- stop its keepalive and idle timers so
# neither fires (or leaks) after the stream is reclaimed.
$weak_self->_h2_stop_sse_keepalive($ss);
$weak_self->_h2_stop_sse_idle_timer($ss);
# End THIS HTTP/2 stream now: flush remaining queued events, then the
# data_callback emits a final END_STREAM frame. `reason` is
# server-side only and is never written to the wire.
$ss->{sse_close_sent} = 1;
$ss->{sse_closing} = 1;
$weak_self->{sse_disconnect_reason} = $event->{reason}
if defined $event->{reason};
$weak_self->{h2_session}->resume_stream($stream_id);
$weak_self->_h2_write_pending;
}
elsif ($type eq 'sse.http.response.start') {
# A decline-after-start attempt is already rejected by advance_sse.
return if $ss->{sse_decline_started}; # idempotent
$ss->{sse_decline_started} = 1;
$ss->{sse_decline_status} = $event->{status} // 200;
$ss->{sse_decline_headers} = [
map { [_validate_header_name($_->[0]), _validate_header_value($_->[1])] }
@{$event->{headers} // []}
];
# RFC 9113 8.2.2 / design 13.3 — strip app-supplied connection,
# transfer-encoding, etc. before submission.
$ss->{sse_decline_headers} = $weak_self->_h2_strip_connection_headers($ss->{sse_decline_headers});
$ss->{sse_decline_body} = '';
}
elsif ($type eq 'sse.http.response.body') {
return unless $ss->{sse_decline_started};
return if $ss->{response_started};
$ss->{sse_decline_body} .= $event->{body} // '';
return if $event->{more}; # more chunks coming — keep buffering
# Defensive: the send-sequence state machine never permits
# sse.keepalive before sse.start, so neither timer can actually be
# armed here -- stopped anyway for the same every-closure-path
# discipline the other SSE/WS stop sites follow.
$weak_self->_h2_stop_sse_keepalive($ss);
$weak_self->_h2_stop_sse_idle_timer($ss);
$ss->{response_started} = 1;
unless (grep { lc($_->[0]) eq 'date' } @{$ss->{sse_decline_headers}}) {
push @{$ss->{sse_decline_headers}}, ['date', $weak_self->{protocol}->format_date];
}
$weak_self->{h2_session}->submit_response($stream_id,
status => $ss->{sse_decline_status},
headers => $ss->{sse_decline_headers},
body => $ss->{sse_decline_body},
);
$weak_self->_h2_write_pending;
}
elsif ($type eq 'http.fullflush') {
# Hand any pending frames to the session's write path (design §8.4).
$weak_self->{h2_session}->resume_stream($stream_id) if $streaming_started;
$weak_self->_h2_write_pending;
}
return;
};
}
sub _h2_process_ws_frames {
my ($self, $stream_id, $stream, $data) = @_;
my $frame = $stream->{ws_frame};
return unless $frame;
$frame->append($data);
while (defined(my $bytes = $frame->next_bytes)) {
my $opcode = $frame->opcode;
# RFC 6455 Section 5.2: RSV1-3 MUST be 0 unless extension defines
# meaning. PAGI doesn't support compression extensions, so RSV must
# always be 0. Same enforcement as h1's _process_websocket_frames
# (Www.pod: transport-agnostic framing enforcement, and RFC 8441's
# "identical to HTTP/1.1" claim).
my $rsv = $frame->rsv;
if ($rsv && ref($rsv) eq 'ARRAY') {
if (grep { $_ } @$rsv) {
$self->_h2_ws_close($stream_id, 1002, 'RSV bits must be 0');
# Server-initiated protocol close (Www.pod: RFC code + 'protocol_error').
$self->_h2_ws_enqueue_disconnect($stream, 1002, 'protocol_error');
return;
}
}
# RFC 6455 Section 5.2: Opcodes 3-7 and 11-15 (0xB-0xF) are reserved.
# Must fail connection with 1002 Protocol Error.
if (($opcode >= 3 && $opcode <= 7) || ($opcode >= 11 && $opcode <= 15)) {
$self->_h2_ws_close($stream_id, 1002, 'Reserved opcode');
# Server-initiated protocol close (Www.pod: RFC code + 'protocol_error').
$self->_h2_ws_enqueue_disconnect($stream, 1002, 'protocol_error');
return;
}
# RFC 6455 Section 5.5: Control frames (close/ping/pong) MUST have
# payload length <= 125 bytes.
if (($opcode == 8 || $opcode == 9 || $opcode == 10) && length($bytes) > 125) {
$self->_h2_ws_close($stream_id, 1002, 'Control frame too large');
# Server-initiated protocol close (Www.pod: RFC code + 'protocol_error').
$self->_h2_ws_enqueue_disconnect($stream, 1002, 'protocol_error');
return;
}
if ($opcode == 1) {
# Text frame
my $text = eval { Encode::decode('UTF-8', $bytes, Encode::FB_CROAK) };
unless (defined $text) {
$self->_h2_ws_close($stream_id, 1007, 'Invalid UTF-8');
# Server-initiated protocol close (Www.pod: RFC code + 'protocol_error').
$self->_h2_ws_enqueue_disconnect($stream, 1007, 'protocol_error');
return;
}
lib/PAGI/Server/Connection.pm view on Meta::CPAN
no warnings 'closed';
$bytes_read = read($fh, $chunk, $to_read);
}
die "Failed to read filehandle: $!\n" unless defined $bytes_read;
last if $bytes_read == 0; # EOF
$self->{_response_size} += $bytes_read;
if ($chunked) {
my $len = sprintf("%x", length($chunk));
$stream->write("$len\r\n$chunk\r\n");
}
else {
$stream->write($chunk);
}
if (defined $remaining) {
$remaining -= $bytes_read;
}
}
# Send final chunk if chunked encoding
if ($chunked) {
$stream->write("0\r\n\r\n");
}
}
# Diagnostics go through the server so log_level governs them and a replaced
# sink sees them. A connection can outlive its server reference during
# shutdown, so falling back to STDERR is a real path, not a formality.
sub _log {
my ($self, $level, $msg) = @_;
my $server = $self->{server};
return $server->_log($level, $msg, __PACKAGE__) if $server;
warn "$msg\n";
return;
}
1;
__END__
=head1 SSE OVER HTTP/2
SSE events (C<sse.start>, C<sse.send>, C<sse.comment>, C<sse.keepalive>)
work transparently over both HTTP/1.1 and HTTP/2. Applications do not need
to change their SSE handling code based on protocol version.
=head2 How It Works
A request is detected as SSE when its combined C<Accept> header values
contain the exact media range C<text/event-stream>, case-insensitively,
with an effective quality value greater than zero (see L<PAGI::Spec::Www/
"SSE Connection Detection">); a C<q=0> refusal or a wildcard range such as
C<*/*> never signals SSE. Detection works identically regardless of HTTP
version. Over HTTP/1.1, SSE data is sent using chunked Transfer-Encoding.
Over HTTP/2, SSE data is sent as DATA frames via the
C<submit_response_streaming>/C<data_callback> mechanism. This difference is
transparent to the application.
The C<http_version> field in the scope hash will be C<'2'> for HTTP/2
connections, allowing applications to distinguish if needed.
=head2 SSE Idle Timeout over HTTP/2
The C<sse_idle_timeout> setting is enforced B<per stream> on HTTP/2: each
SSE stream owns its own idle timer, armed when that stream's
C<sse.start> is sent and reset by that stream's own send activity
(C<sse.send>, C<sse.comment>, C<sse.keepalive>, C<sse.close>). When a
stream's timer expires, only that stream ends -- the server marks the
stream closing, lets it flush any already-queued data, and then emits
the final HTTP/2 END_STREAM frame, the same path an application-initiated
C<sse.close> takes. Sibling SSE (and other) streams multiplexed on the
same HTTP/2 connection are unaffected, and the connection itself stays
open.
Over HTTP/1.1, each SSE stream already owns its own TCP connection, so
C<sse_idle_timeout> is enforced at the connection level there -- expiry
closes that connection, which only ever carries the one SSE stream.
=head2 Connection Reuse after an SSE Stream (HTTP/1.1)
C<sse.start> advertises C<Connection: keep-alive>, and the server honors it.
When an HTTP/1.1 SSE stream ends B<cleanly> -- the application returns, or it
sends C<sse.close> -- the server writes the chunked terminator, resets the
per-request state the stream accumulated, and hands the connection back to
ordinary keep-alive request handling, including serving any request already
pipelined in the read buffer. A pooled client (browser, C<Net::Async::HTTP>,
curl) can therefore reuse the same socket for its next request, which matters
for the short POST-SSE-exchange pattern used by fetch-event-source and
datastar.
Keep-alive yields to the usual overrides, each of which closes the connection
the same way it does outside SSE: a client C<Connection: close>, HTTP/1.0
semantics, server shutdown, an application exception, and any B<abnormal> end
(client disconnect, idle timeout, write error). An abnormal end is also the
only thing that delivers C<sse.disconnect> to the application; a clean end
never does.
Ending the stream is decoupled from the application returning. After
C<sse.close> the application keeps running against a live transport, and any
further send on that scope fails through the event sequence machine (C<after
sse.close>) rather than being silently swallowed by a closed transport.
=head1 SEE ALSO
L<PAGI::Server>, L<PAGI::Server::Protocol::HTTP1>
=head1 AUTHOR
John Napiorkowski E<lt>jjnapiork@cpan.orgE<gt>
=head1 LICENSE
This library is free software; you can redistribute it and/or modify
it under the same terms as Perl itself.
=cut
( run in 0.976 second using v1.01-cache-2.11-cpan-5c0b1e786e0 )