PAGI-Server

 view release on metacpan or  search on metacpan

lib/PAGI/Server/Connection.pm  view on Meta::CPAN

        # Disable drain checking until next high watermark hit
        $self->{_drain_check_active} = 0;
    }
}

sub _setup_drain_detection {
    my ($self) = @_;

    # Avoid redundant setup
    return if $self->{_drain_check_active};
    $self->{_drain_check_active} = 1;

    weaken(my $weak_self = $self);

    # Primary mechanism: check when write queue empties
    # This guarantees we notice drain even for fast-draining connections
    # Store previous handler to chain if needed
    my $prev_on_empty = $self->{_prev_on_outgoing_empty};

    $self->{stream}->configure(
        on_outgoing_empty => sub {
            return unless $weak_self;
            $weak_self->_check_drain_waiters;
            # Call previous handler if any
            $prev_on_empty->(@_) if $prev_on_empty;
        },
    );
}

sub _wait_for_drain {
    my ($self) = @_;

    # Fast path: already below low watermark
    my $buffered = $self->_get_write_buffer_size;
    if ($buffered < $self->{write_low_watermark}) {
        return Future->done;
    }

    # Create Future to be resolved when drained
    my $f = $self->{server}->loop->new_future;
    push @{$self->{_drain_waiters}}, $f;

    # Ensure drain detection is active
    $self->_setup_drain_detection;

    return $f;
}

sub _cancel_drain_waiters {
    my ($self, $reason) = @_;
    $reason //= 'connection closed';

    my @waiters = splice @{$self->{_drain_waiters}};
    for my $f (@waiters) {
        # Resolve (not fail) - app should check connection state after await
        $f->done unless $f->is_ready;
    }
    # Drop (don't fire) the app's on_drain fires: the connection is going
    # away, not draining -- matches h2's teardown handling of
    # transport_drain_fires. The blocking waiters above still resume (so no
    # coroutine leak), but on_drain is a hysteresis signal for a buffer that
    # actually fell back below the low mark, which never happened here.
    $self->{_drain_fires} = [];
    $self->{_drain_check_active} = 0;
}

# HTTP/2 per-stream backpressure: the h2 analogue of _wait_for_drain. Resolves
# when this stream's send queue falls below the low watermark. Each multiplexed
# stream is bounded independently, so a quiet TCP buffer can't let one stream's
# queue grow without limit.
sub _h2_wait_for_stream_drain {
    my ($self, $stream_id) = @_;

    my $ss = $self->{h2_streams}{$stream_id} or return Future->done;

    # Fast path: already below low watermark
    if (($ss->{send_queue_bytes} // 0) < $self->{write_low_watermark}) {
        return Future->done;
    }

    # Create Future to be resolved when this stream's queue drains (in the
    # data_callback pull) or when the stream is torn down.
    my $f = $self->{server}->loop->new_future;
    push @{$ss->{stream_drain_waiters} //= []}, $f;

    return $f;
}

# Release any producer blocked on _h2_wait_for_stream_drain for a stream that
# is being torn down (close/RST/connection shutdown). Resolve, never fail - the
# producer rechecks connection/stream state after the await. Some teardown
# sites run inside nghttp2's feed() (e.g. the oversize-body 413 path); completing
# a waiter resumes the producer synchronously, so defer to the next loop tick to
# keep the resumed producer out of a re-entrant nghttp2 call.
sub _h2_resolve_stream_drain_waiters {
    my ($self, $ss) = @_;
    return unless $ss && $ss->{stream_drain_waiters};
    my @waiters = splice @{$ss->{stream_drain_waiters}};
    return unless @waiters;
    $self->{server}->loop->later(sub {
        $_->done for grep { !$_->is_ready } @waiters;
    });
}

# Release a send() parked in the http.response.trailers arm awaiting
# $deliver_trailer_eof (the data callback's own terminal invocation) for a
# stream that is being torn down before that ever happens. Resolve, never
# fail -- same h2_closed carve-out contract as every other post-close send
# (design §6.2 / §21 item 1): a trailers send racing a disconnect is a
# successful no-op, not an error the app must handle. Same re-entrancy
# discipline as _h2_resolve_stream_drain_waiters above (deferred one loop
# tick -- some teardown sites run inside nghttp2's feed()).
sub _h2_resolve_stream_trailer_wait {
    my ($self, $ss) = @_;
    return unless $ss;
    my $f = delete $ss->{trailer_wait};
    return unless $f;
    $self->{server}->loop->later(sub {
        $f->done unless $f->is_ready;
    });
}

lib/PAGI/Server/Connection.pm  view on Meta::CPAN

    session_complete
);

# Build the app-facing websocket.disconnect event for a server-detected close.
# The code and reason come from the close the server initiated; the defaults are
# the RFC 6455 "abnormal closure, no status received" pair (1006 / empty), used
# when the connection dropped with no close handshake (timeout, TCP FIN).
sub _ws_disconnect_event {
    my ($self) = @_;
    return {
        type   => 'websocket.disconnect',
        code   => $self->{ws_disconnect_code}   // 1006,
        reason => $self->{ws_disconnect_reason} // '',
    };
}

sub _handle_disconnect {
    my ($self, $reason) = @_;

    # Idempotency guard - prevent duplicate disconnect handling
    # Multiple paths can trigger disconnect (timeout, protocol error, session end)
    return if $self->{_disconnect_handled};
    $self->{_disconnect_handled} = 1;

    # Auto-detect server shutdown (PAGI spec compliance)
    # If no explicit reason and server is shutting down, use server_shutdown
    if (!$reason && $self->{server} && $self->{server}{shutting_down}) {
        $reason = 'server_shutdown';
    }

    # Default reason is client_closed (TCP FIN received)
    $reason //= 'client_closed';

    # A clean completion is not an abnormal disconnect: don't surface its reason.
    my $is_completion = $COMPLETION_REASON{$reason};

    # Mark HTTP connection state as disconnected (abnormal only).
    # Only for HTTP - WebSocket/SSE have their own patterns.
    if ($self->{current_connection_state} && !$self->{websocket_mode} && !$self->{sse_mode}) {
        $self->{current_connection_state}->_mark_disconnected($reason)
            unless $is_completion;
    }

    # HTTP/2: connection-level teardown (server shutdown, socket error, ...)
    # sweeps every open stream's own connection_state with this reason, so a
    # stream still mid-response when the whole connection dies still reports
    # why. _mark_disconnected is idempotent -- a stream _h2_on_close already
    # took to a terminal state (complete, or its own client_closed/server_error)
    # keeps that first reason; only still-open streams pick this one up.
    # WebSocket/SSE streams never attach a connection_state (N/A per spec),
    # so the guard on $stream->{connection_state} skips them naturally.
    if ($self->{is_h2} && $self->{h2_streams} && !$is_completion) {
        for my $stream (values %{$self->{h2_streams}}) {
            $stream->{connection_state}->_mark_disconnected($reason)
                if $stream->{connection_state};
        }
    }

    # Cancel any pending drain waiters (backpressure) AFTER connection state
    # is marked above: resolving a parked waiter can synchronously resume an
    # awaiting app coroutine (Future::AsyncAwait resumes inline off ->done),
    # and that resumed app's first act may be to read is_connected() /
    # disconnect_reason() -- those must already reflect this disconnect, not
    # a stale "still connected" snapshot from before it was detected.
    $self->_cancel_drain_waiters($reason);

    # Record the abnormal reason so the WebSocket disconnect event reports it
    # (instead of the old empty string). SSE tracks its own reason at the
    # detection sites via sse_disconnect_reason.
    if ($self->{websocket_mode} && !$is_completion) {
        $self->{ws_disconnect_reason} = $reason;
    }

    # Determine disconnect event type based on mode
    my $disconnect_event;
    if ($self->{websocket_mode}) {
        $disconnect_event = $self->_ws_disconnect_event;
    } elsif ($self->{sse_mode}) {
        # A completed decline is not an abnormal end -- the spec says a
        # decline delivers no events at all, so leave $disconnect_event
        # unset rather than synthesizing sse.disconnect for it.
        $disconnect_event = {
            type   => 'sse.disconnect',
            reason => $self->{sse_disconnect_reason} // 'client_closed',
        } unless $self->{sse_decline_completed};
    } else {
        $disconnect_event = { type => 'http.disconnect' };
    }

    # Queue disconnect event (do this even if already closed)
    push @{$self->{receive_queue}}, $disconnect_event if $disconnect_event;

    # Complete any pending receive
    if ($disconnect_event && $self->{receive_pending} && !$self->{receive_pending}->is_ready) {
        $self->{receive_pending}->done($disconnect_event);
        $self->{receive_pending} = undef;
    }
}

# Send a WebSocket close frame with status code and optional reason
# Per RFC 6455 Section 7.4, common codes:
#   1000 - Normal closure
#   1007 - Invalid frame payload data (e.g., invalid UTF-8)
#   1009 - Message too big
#   1011 - Unexpected condition
sub _send_close_frame {
    my ($self, $code, $reason) = @_;
    $reason //= '';

    return unless $self->{stream};
    return if $self->{close_sent};

    # Remember the wire code so the app-facing websocket.disconnect event reports
    # the same code the peer received, rather than the 1006 abnormal-close default.
    $self->{ws_disconnect_code} = $code;

    my $frame = Protocol::WebSocket::Frame->new(
        type   => 'close',
        buffer => pack('n', $code) . $reason,
    );

lib/PAGI/Server/Connection.pm  view on Meta::CPAN

        }
    }

    # Stop idle timer
    $self->_stop_idle_timer;

    # Stop stall timer
    $self->_stop_stall_timer;

    # Stop WS/SSE idle timers
    $self->_stop_ws_idle_timer;
    $self->_stop_sse_idle_timer;

    # Stop keepalive timers
    $self->_stop_ws_keepalive;
    $self->_stop_sse_keepalive;

    # Note: _close is resource cleanup ONLY. Callers should use
    # _handle_disconnect_and_close() which handles both protocol
    # notification and cleanup.

    # Determine disconnect event type based on mode
    my $disconnect_event;
    if ($self->{websocket_mode}) {
        $disconnect_event = $self->_ws_disconnect_event;
    } elsif ($self->{sse_mode}) {
        # See _handle_disconnect: a completed decline delivers no events.
        $disconnect_event = {
            type   => 'sse.disconnect',
            reason => $self->{sse_disconnect_reason} // 'client_closed',
        } unless $self->{sse_decline_completed};
    } else {
        $disconnect_event = { type => 'http.disconnect' };
    }

    # Cancel any tracked receive Futures that are still pending
    if ($disconnect_event) {
        for my $future (@{$self->{receive_futures}}) {
            if (!$future->is_ready) {
                # Complete with disconnect event instead of cancelling
                # This allows the async sub to complete cleanly
                $future->done($disconnect_event);
            }
        }
    }
    $self->{receive_futures} = [];

    if ($self->{stream}) {
        $self->{stream}->close_when_empty;
    }
}

# Combined disconnect and close - use this from callbacks where $weak_self may
# become undefined after _handle_disconnect completes its Future callbacks.
# This method holds a strong reference to $self throughout the operation.
sub _handle_disconnect_and_close {
    my ($self, $reason) = @_;

    # Mark the transport closed before notifying: _handle_disconnect below
    # completes any pending receive(), which can synchronously resume the
    # app coroutine (it may run straight through a subsequent send()), so
    # the send-side closed-check needs "closed" to already be true at that
    # point (spec order: closed-check precedes validation). Resource
    # cleanup itself still happens in _close, gated by its own idempotency
    # flag so it isn't skipped by this early flip.
    $self->{closed} = 1;

    $self->_handle_disconnect($reason);
    $self->_close;
}

#
# TLS Support Methods
#

sub _extract_tls_info {
    my ($self) = @_;

    my $stream = $self->{stream};
    my $handle = $stream->read_handle;

    # Check if handle is an IO::Socket::SSL
    return unless $handle && $handle->isa('IO::Socket::SSL');

    my $tls_info = {
        server_cert       => undef,
        client_cert_chain => [],
        client_cert_name  => undef,
        client_cert_error => undef,
        tls_version       => undef,
        cipher_suite      => undef,
    };

    # Get TLS version - IO::Socket::SSL returns something like 'TLSv1_3'
    if (my $version_str = $handle->get_sslversion) {
        # Map version string to numeric value per TLS spec
        my %version_map = (
            'SSLv3'   => 0x0300,
            'TLSv1'   => 0x0301,
            'TLSv1_1' => 0x0302,
            'TLSv1_2' => 0x0303,
            'TLSv1_3' => 0x0304,
        );
        $tls_info->{tls_version} = $version_map{$version_str};
    }

    # Cipher suite (numeric IANA id). Net::SSLeay/IO::Socket::SSL expose only the
    # cipher *name*, not the 16-bit id the spec asks for. For TLS 1.3 the OpenSSL
    # name IS the IANA name and the registry is frozen at five suites, so we map
    # those exactly. For TLS 1.2 the names are OpenSSL-specific (a large, shifting
    # set), so we leave cipher_suite undef -- the spec permits undef when the
    # server cannot determine the value.
    if (my $cipher_name = $handle->get_cipher) {
        my %tls13_cipher_suites = (
            'TLS_AES_128_GCM_SHA256'       => 0x1301,
            'TLS_AES_256_GCM_SHA384'       => 0x1302,
            'TLS_CHACHA20_POLY1305_SHA256' => 0x1303,
            'TLS_AES_128_CCM_SHA256'       => 0x1304,
            'TLS_AES_128_CCM_8_SHA256'     => 0x1305,
        );
        $tls_info->{cipher_suite} = $tls13_cipher_suites{$cipher_name}