File Coverage

blib/lib/PAGI/Server.pm

Criterion	Covered	Total	%
statement	832	1120	74.2
branch	325	606	53.6
condition	202	364	55.4
subroutine	85	108	78.7
pod	10	12	83.3
total	1454	2210	65.7

package PAGI::Server;
use strict;
use warnings;

our $VERSION = '0.001012';

# Future::XS support - opt-in via PAGI_FUTURE_XS=1 environment variable
# Must be loaded before Future to take effect, so we check env var in BEGIN
# Note: We declare these without initialization so BEGIN block values persist
our ($FUTURE_XS_AVAILABLE, $FUTURE_XS_ENABLED);
BEGIN {
    $FUTURE_XS_AVAILABLE = eval { require Future::XS; 1 } ? 1 : 0;
    $FUTURE_XS_ENABLED = 0;  # Default to disabled

    if ($ENV{PAGI_FUTURE_XS}) {
        if ($FUTURE_XS_AVAILABLE) {
            # Future::XS is already loaded from the availability check
            $FUTURE_XS_ENABLED = 1;
        } else {
            die <<"END_FUTURE_XS_ERROR";
PAGI_FUTURE_XS=1 set but Future::XS is not installed.

To install Future::XS:
    cpanm Future::XS

Or unset the PAGI_FUTURE_XS environment variable.
END_FUTURE_XS_ERROR
        }
    } elsif ($FUTURE_XS_AVAILABLE) {
        # Available but not requested - unload it
        delete $INC{'Future/XS.pm'};
    }
}

use parent 'IO::Async::Notifier';
use IO::Async::Listener;
use IO::Async::Stream;
use IO::Async::Loop;
use IO::Async::Timer::Periodic;
use IO::Socket::INET;
use Future;
use Future::AsyncAwait;

use Scalar::Util qw(weaken refaddr);
use Socket qw(sockaddr_family unpack_sockaddr_in unpack_sockaddr_un AF_UNIX AF_INET);
use POSIX ();

use PAGI::Server::Connection;
use PAGI::Server::Protocol::HTTP1;


# Check TLS module availability (cached at load time for banner display)
our $TLS_AVAILABLE;
BEGIN {
    $TLS_AVAILABLE = eval {
        require IO::Async::SSL;
        require IO::Socket::SSL;
        1;
    } ? 1 : 0;
}

sub has_tls { return $TLS_AVAILABLE }

# Check HTTP/2 module availability (cached at load time)
our $HTTP2_AVAILABLE;
BEGIN {
    $HTTP2_AVAILABLE = eval {
        require PAGI::Server::Protocol::HTTP2;
        PAGI::Server::Protocol::HTTP2->available;
    } ? 1 : 0;
}

sub has_http2 { return $HTTP2_AVAILABLE }

# Windows doesn't support Unix signals - signal handling is conditional
use constant WIN32 => $^O eq 'MSWin32';

=encoding utf8

=head1 NAME

PAGI::Server - PAGI Reference Server Implementation

=head1 SYNOPSIS

    use IO::Async::Loop;
    use PAGI::Server;

    # If using Future::IO libraries (Async::Redis, SSE->every, etc.)
    # load the IO::Async implementation BEFORE loading them:
    use Future::IO::Impl::IOAsync;

    my $loop = IO::Async::Loop->new;

    my $server = PAGI::Server->new(
        app  => \&my_pagi_app,
        host => '127.0.0.1',
        port => 5000,
    );

    $loop->add($server);
    $server->listen->get;  # Start accepting connections

See L for details on Future::IO configuration.

=head1 DESCRIPTION

PAGI::Server is a reference implementation of a PAGI-compliant HTTP server.
It supports HTTP/1.1, WebSocket, and Server-Sent Events (SSE) as defined
in the PAGI specification. It prioritizes spec compliance and code clarity
over performance optimization. It serves as the canonical reference for how
PAGI servers should behave.

=head1 PROTOCOL SUPPORT

B

=over 4

=item * HTTP/1.1 (full support including chunked encoding, trailers, keepalive)

=item * HTTP/2 (B - via nghttp2, h2 over TLS and h2c cleartext)

=item * WebSocket (RFC 6455, including over HTTP/2 via RFC 8441)

=item * Server-Sent Events (SSE, including over HTTP/2)

=back

B

=over 4

=item * HTTP/3 (QUIC) - Under consideration

=back

For HTTP/2, see L.

=head1 UNIX DOMAIN SOCKET SUPPORT (EXPERIMENTAL)

B The API is subject to change in future
releases. Please report issues at L.

Unix domain sockets provide efficient local communication between a reverse
proxy (nginx, HAProxy, etc.) and PAGI::Server running on the same machine.
They bypass the TCP/IP stack entirely, reducing latency and overhead compared
to connecting over C<127.0.0.1>.

=head2 When to Use Unix Sockets

=over 4

=item * B — nginx or HAProxy on the same host handles
TLS termination, HTTP/2 negotiation, and static files, forwarding dynamic
requests to PAGI over a Unix socket.

=item * B — frameworks like TechEmpower FrameworkBenchmarks use
Unix sockets to eliminate network variable from application benchmarks.

=item * B — services on the same host communicate without
network overhead.

=back

B If clients connect over the network (remote browsers,
API consumers), use TCP. Unix sockets only accept connections from processes
on the same machine.

=head2 Basic Usage

B

    my $server = PAGI::Server->new(
        app    => $app,
        socket => '/tmp/pagi.sock',
    );

B

    pagi-server --socket /tmp/pagi.sock ./app.pl

B

    pagi-server --socket /tmp/pagi.sock --workers 4 ./app.pl

In multi-worker mode, the parent process creates the Unix socket and all
worker processes inherit the file descriptor via C. The kernel
distributes incoming connections across workers.

=head2 How It Works

=over 4

=item 1. On startup, any existing file at the socket path is removed (stale
socket cleanup).

=item 2. The server creates and binds a C Unix domain socket
at the specified path using C (multi-worker) or
C with C<< family => 'unix' >> (single-worker).

=item 3. If C is set, C is called immediately after
binding to set the file permissions.

=item 4. In multi-worker mode, the parent creates the socket before forking.
Workers inherit the listening fd and each runs its own C
wrapping the inherited handle.

=item 5. On graceful shutdown (SIGTERM/SIGINT), the socket file is unlinked
by both the single-worker C path and the multi-worker
C<_initiate_multiworker_shutdown()> path.

=back

=head2 nginx Configuration

B

    upstream pagi_backend {
        server unix:/var/run/myapp/pagi.sock;
        keepalive 32;
    }

    server {
        listen 80;
        server_name myapp.example.com;

        location / {
            proxy_pass http://pagi_backend;

            # Required for upstream keepalive:
            proxy_http_version 1.1;
            proxy_set_header Connection "";

            # Forward client info (since PAGI can't see it over Unix socket):
            proxy_set_header Host $host;
            proxy_set_header X-Real-IP $remote_addr;
            proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
            proxy_set_header X-Forwarded-Proto $scheme;
        }
    }

B The C directive is critical for performance. Without
it, nginx opens a new Unix socket connection for every request.
C and C are required
for keepalive to work.

B

    server {
        listen 443 ssl http2;
        ssl_certificate     /etc/ssl/myapp.crt;
        ssl_certificate_key /etc/ssl/myapp.key;

        location / {
            proxy_pass http://pagi_backend;
            # ... same proxy headers as above
        }
    }

nginx handles TLS and HTTP/2 with clients, then speaks plain HTTP/1.1 to
PAGI over the Unix socket. This is the recommended production pattern.

=head2 Socket Permissions and Security

By default, the socket file inherits permissions from the process umask.
Use C to set explicit permissions:

    # CLI
    pagi-server --socket /var/run/myapp/pagi.sock --socket-mode 0660 ./app.pl

    # Programmatic
    PAGI::Server->new(
        app         => $app,
        socket      => '/var/run/myapp/pagi.sock',
        socket_mode => 0660,
    );

B

=over 4

=item * B, not C. Directories like
C or C prevent symlink attacks and provide
an additional permission layer.

=item * B with a shared group.> Create a group (e.g., C)
that both the application user and the nginx user belong to:

    sudo groupadd myapp
    sudo usermod -aG myapp www-data    # nginx user
    sudo usermod -aG myapp myappuser   # app user
    sudo mkdir -p /var/run/myapp
    sudo chown myappuser:myapp /var/run/myapp
    sudo chmod 0750 /var/run/myapp

=item * B> for automatic directory management:

    # /etc/systemd/system/myapp.service
    [Service]
    User=myappuser
    Group=myapp
    RuntimeDirectory=myapp
    RuntimeDirectoryMode=0750
    ExecStart=/usr/local/bin/pagi-server \
        --socket /run/myapp/pagi.sock \
        --socket-mode 0660 \
        --workers 4 \
        /opt/myapp/app.pl

systemd creates C on service start and cleans it up on stop.

=back

=head2 TLS Over Unix Sockets

TLS can be used over Unix sockets, though this is unusual — normally the
reverse proxy handles TLS termination. When TLS is configured on a Unix
socket listener, the server logs an info-level note suggesting reverse proxy
TLS termination instead.

The combination is allowed because it has legitimate uses (encrypted
inter-container communication, compliance requirements). All major ASGI
servers (Uvicorn, Hypercorn, Granian) also allow it.

=head2 HTTP/2 Over Unix Sockets

h2c (HTTP/2 cleartext) works over Unix sockets. This is useful for gRPC
backends or reverse proxies that support HTTP/2 to upstreams (e.g., Envoy).
Note that nginx does B currently support HTTP/2 to upstream backends
(except for gRPC via C).

=head2 Scope Differences

For Unix socket connections, the PAGI scope differs from TCP connections:

=over 4

=item * B is absent> — Unix sockets have no peer IP address or
port. The C key is omitted entirely from the scope hashref (not set
to C). This is spec-compliant: the PAGI specification marks C
as optional.

=item * B is C<[$socket_path, undef]>>> — instead of C<[$host, $port]>.

=back

B Any middleware that accesses C<< $scope->{client} >>
must check C<< exists $scope->{client} >> first. For client IP identification
behind a reverse proxy, use C or C headers
instead of C<< $scope->{client} >>. The C
middleware (if available) handles this automatically.

B Unix socket connections log C as the client IP in the
access log instead of an IP address.

=head2 Stale Socket Cleanup

If a socket file already exists at the configured path (e.g., from a previous
crash), it is automatically removed before binding. This matches the behavior
of Starman, Gunicorn, Uvicorn, and other production servers. The socket file
is also removed during graceful shutdown (SIGTERM/SIGINT).

If the server is killed with SIGKILL (C), the socket file will
B be cleaned up. It will be removed on the next startup.

=head1 MULTI-LISTENER SUPPORT (EXPERIMENTAL)

B The API is subject to change in future
releases.

A single PAGI::Server instance can listen on multiple endpoints
simultaneously. This is useful for:

=over 4

=item * B — load
balancers probe a TCP port while nginx uses the Unix socket.

=item * B — serve different interfaces on different ports.

=item * B — listen on both old and new ports during
a transition.

=back

=head2 Programmatic API

    my $server = PAGI::Server->new(
        app    => $app,
        listen => [
            { host => '0.0.0.0', port => 8080 },
            { socket => '/tmp/pagi.sock', socket_mode => 0660 },
        ],
    );

Each spec in the C array is a hashref with either C<< { host, port } >>
for TCP or C<< { socket } >> (with optional C) for Unix sockets.

B Per-listener TLS configuration is not yet supported. TLS is configured
server-wide via the C constructor option and applies to all TCP listeners.
Unix socket listeners behind a reverse proxy do not need TLS — the proxy handles
TLS termination.

=head2 CLI

The C<--listen> flag is repeatable. The server auto-detects TCP vs Unix
socket: values containing C<:> are parsed as C, everything else
is treated as a Unix socket path.

    # TCP + Unix socket
    pagi-server --listen 0.0.0.0:8080 --listen /tmp/pagi.sock ./app.pl

    # Multiple TCP ports
    pagi-server --listen 0.0.0.0:8080 --listen 0.0.0.0:8443 ./app.pl

    # With workers
    pagi-server --listen 0.0.0.0:8080 --listen /tmp/pagi.sock -w 4 ./app.pl

    # IPv6
    pagi-server --listen [::1]:5000 ./app.pl

C<--listen> is B with C<--host>, C<--port>, and
C<--socket>. C<--socket-mode> applies to all Unix socket listeners when
using C<--listen>.

=head2 How It Works

=over 4

=item * In B, one C is created per
endpoint. All listeners share the same event loop and connection handler.

=item * In B, the parent process creates all listening
sockets (Unix and TCP) before forking. Workers inherit all file descriptors
and create their own C for each inherited socket.

=item * The C option applies only to TCP listeners. Unix socket
listeners always use the shared-socket model (parent creates, workers inherit).

=item * On shutdown, all listeners are stopped and all Unix socket files are
cleaned up.

=back

=head2 Accessors

    $server->port;          # Bound port of first TCP listener, or undef
    $server->socket_path;   # Path of first Unix socket listener, or undef
    $server->listeners;     # Arrayref of all listener specs

=head2 Backward Compatibility

The existing C/C constructor options continue to work exactly
as before. They are internally normalized to a single-element listener
array. The C option is similarly sugar for a single Unix socket
listener. Only C enables true multi-listener mode.

=head1 SYSTEMD SOCKET ACTIVATION (EXPERIMENTAL)

B The API is subject to change in future
releases.

PAGI::Server supports systemd socket activation, which allows systemd to
create and hold listening sockets on behalf of the server. This enables
zero-downtime restarts: when the server is restarted, the kernel continues
to queue incoming connections on the socket without refusing or dropping
them, even during the gap between the old process exiting and the new one
starting.

Benefits of systemd socket activation:

=over 4

=item * B — the kernel queues connections during restarts

=item * B — systemd creates the socket as root, then drops privileges before exec

=item * B — the server is started automatically when the first connection arrives

=back

=head2 Basic Setup

Create two systemd unit files: a C<.socket> unit that describes the socket,
and a C<.service> unit for the server itself.

B):>

    [Unit]
    Description=PAGI Application Socket

    [Socket]
    ListenStream=0.0.0.0:8080
    Accept=no

    [Install]
    WantedBy=sockets.target

B):>

    [Unit]
    Description=PAGI Application Socket

    [Socket]
    ListenStream=/run/pagi/app.sock
    SocketMode=0660
    SocketUser=www-data
    SocketGroup=www-data
    Accept=no

    [Install]
    WantedBy=sockets.target

B):>

    [Unit]
    Description=PAGI Application Server
    Requires=pagi.socket

    [Service]
    User=www-data
    ExecStart=/usr/local/bin/pagi-server -E production ./app.pl
    Restart=on-failure

    [Install]
    WantedBy=multi-user.target

C is B. PAGI::Server accepts connections itself via
C; systemd must not accept on its behalf.

=head2 How Auto-Detection Works

When PAGI::Server starts, it checks the C and C
environment variables set by systemd. If C matches the current
process PID, the server inspects each inherited file descriptor (starting at
fd 3) with C to determine its address.

Each inherited socket is then matched against the configured listeners. For
example, if you configure C<< port => 8080 >> and systemd has a socket bound
to C<0.0.0.0:8080>, PAGI::Server will use the inherited fd instead of
creating a new socket.

The same application code works identically with or without systemd:

    # Without systemd: PAGI::Server binds the socket itself
    # With systemd:    PAGI::Server inherits the socket from systemd
    my $server = PAGI::Server->new(
        app  => $app,
        host => '0.0.0.0',
        port => 8080,
    );

After reading the inherited fds, PAGI::Server removes C,
C, and C from the environment (per the
C specification), so child processes do not re-inherit them.

=head2 Unix Socket Cleanup

Normally, PAGI::Server unlinks its Unix socket file on shutdown. For
systemd-activated Unix sockets, the socket file is B unlinked because
systemd owns the socket and will recreate it for the next activation.

=head1 FD REUSE INTERNALS (EXPERIMENTAL)

B The API is subject to change in future
releases.

PAGI::Server uses a C environment variable to pass inherited
listening socket file descriptors to re-exec'd processes during hot restart
(see L). This mechanism also supports systemd
socket activation (see L).

=head2 PAGI_REUSE Format

The variable is a comma-separated list of C entries:

    # TCP listeners
    PAGI_REUSE=127.0.0.1:8080:3,0.0.0.0:8443:4

    # Unix socket listeners
    PAGI_REUSE=unix:/run/pagi/app.sock:5

    # Mixed
    PAGI_REUSE=0.0.0.0:8080:3,unix:/run/pagi/app.sock:4

Each entry encodes the address the socket is bound to and the file descriptor
number to use.

=head2 Fd Matching

When starting, C<_collect_inherited_fds()> parses C and/or
C, building a table of C<< address => fd >> pairs. During
C, each configured listener looks up its own address in the table.
If a match is found, the existing fd is used instead of calling C and
C. This allows the kernel's accept queue to be preserved across
restarts.

=head2 File Descriptor Inheritance

To ensure that listening socket fds are inherited across C,
PAGI::Server sets C<$^F = 1023> before creating sockets. Perl uses C<$^F>
(the maximum system file descriptor, equivalent to C in
spirit) to decide which fds receive the C close-on-exec flag:
fds with numbers greater than C<$^F> get C set automatically.
By raising C<$^F> to 1023, listen socket fds remain open across C
without requiring explicit C calls.

=head1 HOT RESTART (EXPERIMENTAL)

B The API and signal behaviour are subject
to change in future releases.

Deploying new code normally requires a server restart. During that restart
there is a gap — however brief — where the listening socket is closed and
incoming connections are dropped. Hot restart eliminates this gap by having
the old master fork and exec a brand-new master process that B the
already-open listening sockets. Both the old master and the new master serve
requests during the transition; clients never see a refused connection.

=head2 How It Works

=over 4

=item 1.

An admin sends C to the running master: Cmaster_pidE>

=item 2.

The old master sets C (encoding each listening socket fd) and
C (its own PID) in the environment, then calls C
followed immediately by C of the original C command
(reconstructed from C).

=item 3.

The new master starts, finds the inherited file descriptors via C,
and reuses the existing listening sockets rather than calling C/C
again. The kernel's accept queue is preserved — no connections are dropped.

=item 4.

The new master spawns its worker pool and waits for each worker to complete
the lifespan startup handshake (heartbeat).

=item 5.

Once all workers are healthy, the new master sends C to the old
master (read from C).

=item 6.

The old master receives C, finishes in-flight requests within its
shutdown timeout, and exits cleanly.

=back

=head2 HUP vs USR2

    HUP   — Rolling worker restart. Workers are replaced one by one.
             Code loaded at master startup (middleware, startup modules)
             is NOT reloaded. Use for: config changes picked up per-worker.

    USR2  — Full master re-exec. Everything reloaded from disk including
             the perl binary, all modules, middleware stack. Use for:
             code deploys, Perl upgrades, PAGI::Server upgrades.

=head2 Deploy Workflow

    # Deploy new code
    rsync -a ./lib/ /opt/myapp/lib/

    # Hot restart (zero downtime)
    kill -USR2 $(cat /var/run/myapp/pagi.pid)

    # Verify (optional)
    curl http://localhost:8080/health

=head2 systemd Unit File

    [Service]
    Type=forking
    PIDFile=/var/run/myapp/pagi.pid
    ExecStart=/usr/local/bin/pagi-server \
        --host 0.0.0.0 --port 8080 \
        --workers 4 --pid /var/run/myapp/pagi.pid \
        --daemonize /opt/myapp/app.pl
    ExecReload=/bin/kill -USR2 $MAINPID
    KillMode=process

=head2 Failure Handling

The design ensures the old master never stops until the new master explicitly
sends C:

=over 4

=item * B — The old master logs the error and continues serving.
No new master is started.

=item * B — The forked child exits before loading any code.
The old master notices (via C) and continues serving.

=item * B — The old master never receives
C and continues serving indefinitely.

=item * B — The new master exits without
sending C. The old master is unaffected.

=back

=head2 PERL5LIB and Module Paths

When the new master is C'd it inherits the process environment, but
B any C<-I> flags that were on the original command line. If your
application uses C<-Ilib>, ensure C is set in the environment or
in the systemd unit file so the re-exec'd process can find your modules.
Alternatively, use the C<--lib> flag (C),
which is captured in C and replayed on re-exec.

=head1 WINDOWS SUPPORT

B

The server relies on Unix-specific features that are not available on Windows:

=over 4

=item * B - SIGTERM, SIGINT, SIGHUP for graceful shutdown and worker management

=item * B - Multi-worker mode requires real process forking, not thread emulation

=item * B - The event loop has Unix-specific optimizations

=back

For Windows development, consider using WSL (Windows Subsystem for Linux) to
run PAGI::Server in a Linux environment. The PAGI specification and middleware
components can still be developed and unit-tested on Windows, but the reference
server implementation requires a Unix-like operating system.


=head1 CONSTRUCTOR

=head2 new

    my $server = PAGI::Server->new(%options);

Creates a new PAGI::Server instance. Options:

=over 4

=item app => \&coderef (required)

The PAGI application coderef with signature: async sub ($scope, $receive, $send)

=item host => $host

Bind address (IP address or hostname). Default: C<'127.0.0.1'>

The default binds only to the loopback interface, accepting connections only
from localhost. This is B - development
servers won't accidentally be exposed to the network.

B

    '127.0.0.1'      - Localhost only (default, secure for development)
    '0.0.0.0'        - All IPv4 interfaces (required for remote access)
    '::'             - All IPv6 interfaces (may also accept IPv4)
    '192.168.1.100'  - Specific interface only

B where remote clients need
to connect, bind to all interfaces:

    my $server = PAGI::Server->new(
        app  => $app,
        host => '0.0.0.0',
        port => 8080,
    );

B When binding to C<0.0.0.0>, ensure appropriate firewall
rules are in place. For production, consider a reverse proxy (nginx, etc.)

=item port => $port

Bind port. Default: 5000

=item socket => $path

B Unix domain socket path for listening instead of TCP host:port.
B with C, C, and C.
See L for details.

    my $server = PAGI::Server->new(
        app    => $app,
        socket => '/tmp/pagi.sock',
    );

=item socket_mode => $mode

Set file permissions on the Unix domain socket after creation. The value
should be a numeric mode (e.g., C<0660>). If not specified, the socket
inherits the default permissions from the process umask. Silently ignored
if C is not set.

    my $server = PAGI::Server->new(
        app         => $app,
        socket      => '/tmp/pagi.sock',
        socket_mode => 0660,
    );

=item listen => \@specs

B Array of listener specifications for multi-endpoint listening.
Each spec is a hashref with either C<< { host, port } >> for TCP or
C<< { socket, socket_mode } >> for Unix domain sockets.
B with C, C, C, and C.
See L for details.

    my $server = PAGI::Server->new(
        app    => $app,
        listen => [
            { host => '0.0.0.0', port => 8080 },
            { socket => '/tmp/pagi.sock', socket_mode => 0660 },
        ],
    );

=item ssl => \%config

Optional TLS/HTTPS configuration. B - see
L below.

Configuration keys:

=over 4

=item cert_file => $path

Path to the SSL certificate file (PEM format).

=item key_file => $path

Path to the SSL private key file (PEM format).

=item ca_file => $path

Optional path to CA certificate for client verification.

=item verify_client => $bool

If true, require and verify client certificates.

=item min_version => $version

Minimum TLS version. Default: C<'TLSv1_2'>. Options: C<'TLSv1_2'>, C<'TLSv1_3'>.

=item cipher_list => $string

OpenSSL cipher list. Default uses modern secure ciphers.

=back

Example:

    my $server = PAGI::Server->new(
        app => $app,
        ssl => {
            cert_file => '/path/to/server.crt',
            key_file  => '/path/to/server.key',
        },
    );

=item disable_tls => $bool

Force-disable TLS even if ssl config is provided. Useful for testing
TLS configuration parsing without actually enabling TLS. Default: false.

=item extensions => \%extensions

Extensions to advertise (e.g., { fullflush => {} })

=item on_error => \&callback

Error callback receiving ($error)

=item access_log => $filehandle | undef

Access log filehandle. Default: STDERR

Set to C to disable access logging entirely. This eliminates
per-request I/O overhead, improving throughput by 5-15% depending on
workload. Useful for benchmarking or when access logs are handled
externally (e.g., by a reverse proxy).

    # Disable access logging
    my $server = PAGI::Server->new(
        app        => $app,
        access_log => undef,
    );

=item access_log_format => $format_or_preset

Access log format string or preset name. Default: C<'clf'>

Named presets:

    clf      - PAGI default: IP, timestamp, method/path, status, duration
    combined - Apache combined: adds Referer and User-Agent
    common   - Apache common: adds response size
    tiny     - Minimal: method, path, status, duration

Custom format strings use Apache-style atoms. See L.

    my $server = PAGI::Server->new(
        app               => $app,
        access_log_format => 'combined',
    );

=item log_level => $level

Controls the verbosity of server log messages. Default: 'info'

Valid levels (from least to most verbose):

=over 4

=item * B - Only errors (application errors, fatal conditions)

=item * B - Warnings and errors (connection issues, timeouts)

=item * B - Informational messages and above (startup, shutdown, worker spawning)

=item * B - Everything (verbose diagnostics, frame-level details)

=back

    my $server = PAGI::Server->new(
        app       => $app,
        log_level => 'debug',  # Very verbose
    );

B C<--log-level debug>

=item workers => $count

Number of worker processes for multi-worker mode. Default: 0 (single process mode).

When set to a value greater than 0, the server uses a pre-fork model:

=item listener_backlog => $number

Value for the listener queue size. Default: 2048

When in multi worker mode, the queue size for those workers inherits
from this value.

=item reuseport => $bool

Enable SO_REUSEPORT mode for multi-worker servers. Default: 0 (disabled).

When enabled, each worker process creates its own listening socket with
SO_REUSEPORT, allowing the kernel to load-balance incoming connections
across workers. This can reduce accept() contention and improve p99
latency under high concurrency.

B Parent creates one socket before forking,
all workers inherit and share that socket. Workers compete on a single
accept queue (potential thundering herd).

B Each worker creates its own socket with
SO_REUSEPORT. The kernel distributes connections across sockets, each
worker has its own accept queue (reduced contention).

B

=over 4

=item * B: Full kernel-level load balancing. Recommended for high
concurrency workloads.

=item * B: SO_REUSEPORT allows multiple binds but does NOT provide
kernel load balancing. May actually decrease performance compared to shared
socket mode. Use with caution - benchmark before deploying.

=back

=item max_receive_queue => $count

Maximum number of messages that can be queued in the WebSocket receive queue
before the connection is closed. This is a DoS protection mechanism.

B Message count (not bytes). Each WebSocket text or binary frame counts
as one message regardless of size.

B 1000 messages

B The server sends a WebSocket close frame with code 1008
(Policy Violation) and reason "Message queue overflow", then closes the
connection.

B

=over 4

=item * B Each queued message holds the full message payload.
With default of 1000 messages and average 1KB messages, worst case is ~1MB
per slow connection.

=item * B Total memory risk = workers × max_connections × max_receive_queue × avg_message_size.
For 4 workers, 100 connections each, 1000 queue, 1KB average = 400MB worst case.

=item * B If your app processes messages quickly, the queue
rarely grows. Default of 1000 is generous for most applications.

=item * B If your app does expensive processing per message,
consider lowering to 100-500 to limit memory exposure.

=item * B If you have trusted clients sending rapid bursts,
you may increase to 5000-10000, but monitor memory usage.

=back

B C<--max-receive-queue 500>

=item max_ws_frame_size => $bytes

Maximum size in bytes for a single WebSocket frame payload. When a client
sends a frame larger than this limit, the connection is closed with a
protocol error.

B Bytes

B 65536 (64KB) - matches Protocol::WebSocket default

B The server closes the connection. The error is logged as
"PAGI connection error: Payload is too big."

B

=over 4

=item * B For chat apps or control messages, default 64KB is plenty.

=item * B For binary data transfer via WebSocket, increase to 1MB-16MB
depending on expected file sizes.

=item * B Each connection can buffer up to max_ws_frame_size bytes
during frame parsing. High values increase memory per connection.

=item * B Lower values limit memory exhaustion from malicious clients
sending oversized frames.

=back

B C<--max-ws-frame-size 1048576>

=item max_connections => $count

Maximum number of concurrent connections before returning HTTP 503.
B (same as Mojolicious).

When at capacity, new connections receive a 503 Service Unavailable
response with a Retry-After header. This prevents resource exhaustion
under heavy load.

B

    my $server = PAGI::Server->new(
        app             => $app,
        max_connections => 5000,  # Higher limit for production
    );

B C<--max-connections 5000>

B Use C<< $server->connection_count >> and
C<< $server->effective_max_connections >> to monitor usage.

B

For high-traffic production deployments, you'll likely want to increase
this value. The optimal setting depends on your workload, available
memory, and system file descriptor limits.

I

    # Check current limits
    ulimit -n           # Soft limit (per-process)
    cat /proc/sys/fs/file-max  # System-wide limit

    # Increase for current session
    ulimit -n 65536

    # Permanent: add to /etc/security/limits.conf
    *  soft  nofile  65536
    *  hard  nofile  65536

    # Or for systemd services, in your unit file:
    [Service]
    LimitNOFILE=65536

I

    # Check current limits
    ulimit -n           # Soft limit
    sysctl kern.maxfilesperproc  # Per-process max

    # Increase for current session
    ulimit -n 65536

    # Permanent: add to /etc/launchd.conf or use launchctl
    sudo launchctl limit maxfiles 65536 200000

I Set C to roughly 80% of your file
descriptor limit to leave headroom for database connections, log files,
and other resources.

=item write_high_watermark => $bytes

B Maximum bytes to buffer in the socket write queue
before applying backpressure. When exceeded, C<< $send->() >> calls will
pause until the buffer drains below C.
Default: 65536 (64KB).

This prevents unbounded memory growth when the server writes data faster
than the client can receive it. The default matches Python's asyncio
transport defaults, providing a good balance between throughput and
memory efficiency.

B

=over 4

=item * B (e.g., 256KB-1MB) for high-throughput bulk transfers
where you want fewer context switches and higher throughput at the cost
of more per-connection memory.

=item * B (e.g., 16KB-32KB) if supporting many concurrent
connections where memory efficiency is critical.

=back

B

    # High-throughput file server - larger buffers
    my $server = PAGI::Server->new(
        app                  => $app,
        write_high_watermark => 262144,  # 256KB
        write_low_watermark  => 65536,   # 64KB
    );

=item write_low_watermark => $bytes

B Threshold below which sending resumes after
backpressure was applied. Must be less than or equal to
C. Default: 16384 (16KB, which is high/4).

A larger gap between high and low watermarks reduces oscillation
(frequent pause/resume cycles). A smaller gap provides more responsive
backpressure but may increase context switching.

B

    # Minimize oscillation with wider gap
    my $server = PAGI::Server->new(
        app                  => $app,
        write_high_watermark => 131072,  # 128KB
        write_low_watermark  => 16384,   # 16KB (8:1 ratio)
    );

=item max_body_size => $bytes

Maximum request body size in bytes. Default: 10,000,000 (10MB).
Set to 0 for unlimited (not recommended for public-facing servers).

Requests with Content-Length exceeding this limit receive HTTP 413
(Payload Too Large). Chunked requests are also checked as data arrives.

B

    my $server = PAGI::Server->new(
        app           => $app,
        max_body_size => 50_000_000,  # 50MB for file uploads
    );

    # Unlimited (use with caution)
    my $server = PAGI::Server->new(
        app           => $app,
        max_body_size => 0,
    );

B C<--max-body-size 50000000>

B Without a body size limit, attackers can exhaust server
memory with large requests. The 10MB default balances security with common
use cases (file uploads, JSON payloads). Increase for specific needs, or
use 0 only behind a reverse proxy that enforces its own limit.

=over 4

=item * A listening socket is created before forking

=item * Worker processes are spawned using C<< $loop->fork() >> which properly
handles IO::Async's C<$ONE_TRUE_LOOP> singleton

=item * Each worker gets a fresh event loop and runs lifespan startup independently

=item * Workers that exit are automatically respawned via C<< $loop->watch_process() >>

=item * SIGTERM/SIGINT triggers graceful shutdown of all workers

=back

=item sync_file_threshold => $bytes

Threshold in bytes for synchronous file reads. Files smaller than this value
are read synchronously in the event loop; larger files use async I/O via
a worker pool.

B 65536 (64KB)

Set to 0 for fully async file reads. This is recommended for:

=over 4

=item * Network filesystems (NFS, SMB, cloud storage)

=item * High-latency storage (spinning disks under load)

=item * Docker volumes with overlay filesystem

=back

The default (64KB) is optimized for local SSDs where small synchronous reads
are faster than the overhead of async I/O.

B C<--sync-file-threshold NUM>

=item max_requests => $count

Maximum number of requests a worker process will handle before restarting.
After serving this many requests, the worker gracefully shuts down and the
parent spawns a replacement.

B 0 (disabled - workers run indefinitely)

B

=over 4

=item * Long-running deployments where gradual memory growth is a concern

=item * Applications with known memory leaks that can't be easily fixed

=item * Defense against slow memory growth (~6.5 bytes/request observed in PAGI)

=back

B Only applies in multi-worker mode (C<< workers > 0 >>). In single-worker
mode, this setting is ignored.

B C<--max-requests 10000>

Example: With 4 workers and max_requests=10000, total capacity before any
restart is 40,000 requests. Workers restart individually without downtime.

=item timeout => $seconds

Connection idle timeout in seconds. Closes connections that are idle between
requests (applies to keep-alive connections waiting for the next request).

B 60

B Each connection with a non-zero timeout creates a timer
that is reset on every read event. For maximum throughput in high-performance
scenarios, set C 0> to disable the idle timer entirely. This
eliminates timer management overhead but means idle connections will never
be automatically closed.

B

    # Disable idle timeout for maximum performance
    my $server = PAGI::Server->new(
        app     => $app,
        timeout => 0,
    );

    # Short timeout to reclaim connections quickly
    my $server = PAGI::Server->new(
        app     => $app,
        timeout => 30,
    );

B C<--timeout 0> or C<--timeout 30>

B This differs from C (stall timeout during active
request processing). The C applies between requests; the
C applies during a request. For WebSocket and SSE, use
C and C respectively.

=item request_timeout => $seconds

Maximum time in seconds a request can stall without any I/O activity before
being terminated. This is a "stall timeout" - the timer resets whenever data
is read from the client or written to the client.

B 0 (disabled)

B Creating per-request timers adds overhead that
impacts throughput on high-performance workloads. For maximum performance,
this is disabled by default. Most production deployments run behind a reverse
proxy (nginx, haproxy) which provides its own timeout protection.

B

=over 4

=item * Running PAGI directly without a reverse proxy

=item * Small/internal apps where simplicity matters more than max throughput

=item * Untrusted clients that might send data slowly or hang

=item * Defense against application bugs that cause requests to hang indefinitely

=back

B

=over 4

=item * Timer starts when request processing begins

=item * Timer resets on any read activity (receiving request body)

=item * Timer resets on any write activity (sending response)

=item * If timer expires (no I/O for N seconds), connection is closed

=item * Not used for WebSocket/SSE (they have C/C)

=back

B

    # Enable 30 second stall timeout (recommended when not behind proxy)
    my $server = PAGI::Server->new(
        app             => $app,
        request_timeout => 30,
    );

B C<--request-timeout 30>

B This differs from C (idle connection timeout). The
C applies between requests on keep-alive connections. The
C applies during active request processing.

=item ws_idle_timeout => $seconds

Maximum time in seconds a WebSocket connection can be idle without any
activity (no messages sent or received) before being closed.

B 0 (disabled - WebSocket connections can be idle indefinitely)

When enabled, the timer resets on:

=over 4

=item * Sending any WebSocket frame (accept, send, ping, close)

=item * Receiving any WebSocket frame from client

=back

B

    # Close idle WebSocket connections after 5 minutes
    my $server = PAGI::Server->new(
        app             => $app,
        ws_idle_timeout => 300,
    );

B C<--ws-idle-timeout 300>

B For more sophisticated keep-alive behavior with ping/pong, use
C<< $ws->keepalive($interval, $timeout) >> for protocol-level ping/pong.

=item sse_idle_timeout => $seconds

Maximum time in seconds an SSE connection can be idle without any events
being sent before being closed.

B 0 (disabled - SSE connections can be idle indefinitely)

The timer resets each time an event is sent to the client (including
comments and the initial headers).

B

    # Close idle SSE connections after 2 minutes
    my $server = PAGI::Server->new(
        app              => $app,
        sse_idle_timeout => 120,
    );

B C<--sse-idle-timeout 120>

B For SSE connections that may be legitimately idle, use
C<< $sse->keepalive($interval) >> to send periodic comment keepalives.

B Over HTTP/2 this timeout applies at the connection level,
not per-stream. See L
for details and recommendations.

=item heartbeat_timeout => $seconds

Worker liveness timeout in seconds. Only active in multi-worker mode
(C<< workers >= 2 >>). Has no effect in single-worker mode — use
C for idle connection management there.

Each worker sends a heartbeat to the parent process via a Unix pipe at
an interval of C. The parent checks for missed
heartbeats every C. If a worker has not sent a
heartbeat within C seconds, the parent kills it with
SIGKILL and respawns a replacement.

B Event loop starvation — when the worker's event
loop is completely blocked and cannot process any events. This happens
with blocking syscalls (C, synchronous DNS, blocking database
drivers), deadlocks, runaway CPU-bound computation, or any code that
does not yield to the event loop.

B Slow async operations. A request handler
that does C<< await $db->query(...) >> for 5 minutes is fine — the
C returns control to the event loop, so heartbeats continue
normally. This value should be larger than the maximum time you expect
any single operation to block the event loop without yielding.

B 50 (seconds). Set to 0 to disable.

B

    # Tighter heartbeat for latency-sensitive service
    my $server = PAGI::Server->new(
        app               => $app,
        workers           => 4,
        heartbeat_timeout => 20,
    );

    # Disable heartbeat monitoring
    my $server = PAGI::Server->new(
        app               => $app,
        workers           => 4,
        heartbeat_timeout => 0,
    );

B C<--heartbeat-timeout 20>

=item loop_type => $backend

Specifies the IO::Async::Loop subclass to use when calling C.
This option is ignored when embedding the server in an existing loop.

B Auto-detect (IO::Async chooses the best available backend)

B

    'EPoll'   - Linux epoll (recommended for Linux)
    'EV'      - libev-based (cross-platform, requires EV module)
    'Poll'    - POSIX poll() (portable fallback)
    'Select'  - select() (most portable, least scalable)

B

    my $server = PAGI::Server->new(
        app       => $app,
        loop_type => 'EPoll',
    );
    $server->run;

B C<--loop EPoll> (via pagi-server)

B The specified backend module must be installed. For example,
C 'EPoll'> requires L.

=item h2_max_concurrent_streams => $count

B<(Experimental - HTTP/2 support may change in future releases.)>

Maximum number of concurrent HTTP/2 streams per connection. Each stream
represents an in-flight request/response exchange. Limits resource consumption
and provides protection against rapid-reset attacks.

B 100

This matches Apache httpd, H2O, and Hypercorn defaults. The RFC 7540 default
is unlimited, but 100 is the industry consensus for a safe maximum.

B Increase for API gateways handling many small concurrent requests.
Decrease for memory-constrained environments or when each request is expensive.

=item h2_initial_window_size => $bytes

B<(Experimental)>

Initial HTTP/2 flow control window size per stream, in bytes. Controls how
much data a client can send before the server must acknowledge receipt. Also
affects how much response data the server can buffer per stream before the
client acknowledges.

B 65535 (64KB minus 1, the RFC 7540 default)

B Increase to 131072-262144 for high-throughput file upload/download
workloads where the default window causes flow control stalls on high-latency
connections. The tradeoff is higher per-stream memory usage.

=item h2_max_frame_size => $bytes

B<(Experimental)>

Maximum size of a single HTTP/2 frame payload, in bytes. Must be between
16384 (16KB, the RFC minimum) and 16777215 (16MB, the RFC maximum).

B 16384 (16KB, the RFC 7540 default)

Most servers use the RFC default. Larger frames reduce framing overhead but
increase head-of-line blocking within a stream.

=item h2_enable_push => $bool

B<(Experimental)>

Enable HTTP/2 server push (SETTINGS_ENABLE_PUSH). When enabled, the server
can proactively push resources to the client before they are requested.

B 0 (disabled)

Server push is effectively deprecated. Chrome removed support in 2022,
and nginx deprecated it in version 1.25.1. Unless you have a specific use
case requiring server push, leave this disabled.

=item h2_enable_connect_protocol => $bool

B<(Experimental)>

Enable the Extended CONNECT protocol (RFC 8441, SETTINGS_ENABLE_CONNECT_PROTOCOL).
Required for WebSocket-over-HTTP/2 tunneling.

B 1 (enabled)

When enabled, clients can use the Extended CONNECT method with a C<:protocol>
pseudo-header to establish WebSocket connections over HTTP/2 streams. Disable
this only if you do not need WebSocket support over HTTP/2.

=item h2_max_header_list_size => $bytes

B<(Experimental)>

Maximum total size of the header block that the server will accept, in bytes.
This is the sum of all header name lengths, value lengths, and 32-byte per-entry
overhead as defined by RFC 7540 Section 6.5.2.

B 65536 (64KB)

Matches Hypercorn and Node.js defaults. Provides a guard against header-based
memory exhaustion attacks while being generous enough for normal use including
large cookies and authorization tokens.

=back

=head1 METHODS

=head2 listen

    my $future = $server->listen;

Starts listening for connections. Returns a Future that completes when
the server is ready to accept connections.

=head2 shutdown

    my $future = $server->shutdown;

Initiates graceful shutdown. Returns a Future that completes when
shutdown is complete.

=head2 port

    my $port = $server->port;

Returns the bound port number. Useful when port => 0 is used.

=head2 socket_path

    my $path = $server->socket_path;

Returns the Unix socket path of the first Unix socket listener,
or C if no Unix socket listeners are configured.

=head2 listeners

    my $listeners = $server->listeners;

Returns an arrayref of all normalized listener specifications.
Each entry is a hashref with C (C<'tcp'> or C<'unix'>)
and type-specific keys (C/C for TCP, C for Unix).

=head2 is_running

    my $bool = $server->is_running;

Returns true if the server is accepting connections.

=head2 connection_count

    my $count = $server->connection_count;

Returns the current number of active connections.

=head2 effective_max_connections

    my $max = $server->effective_max_connections;

Returns the effective maximum connections limit. If C
was set explicitly, returns that value. Otherwise returns the default
of 1000.

=head1 FILE RESPONSE STREAMING

PAGI::Server supports efficient file streaming via the C and C
keys in C events:

    # Stream entire file
    await $send->({
        type => 'http.response.body',
        file => '/path/to/file.mp4',
        more => 0,
    });

    # Stream partial file (for Range requests)
    await $send->({
        type => 'http.response.body',
        file => '/path/to/file.mp4',
        offset => 1000,
        length => 5000,
        more => 0,
    });

    # Stream from filehandle
    open my $fh, '<:raw', $file;
    await $send->({
        type => 'http.response.body',
        fh => $fh,
        length => $size,
        more => 0,
    });
    close $fh;

The server streams files in 64KB chunks to avoid memory bloat. Small files
(under 64KB) are read synchronously for speed; larger files use async I/O
via a worker pool to avoid blocking the event loop.

=head2 Production Recommendations for Static Files

B
serving to a reverse proxy:>

=over 4

=item 1. B

Place a reverse proxy in front of PAGI::Server and let it handle static
files directly. This provides:

=over 4

=item * Optimized file serving with kernel sendfile

=item * Efficient caching and compression

=item * Protection from slow client attacks

=item * HTTP/2 and HTTP/3 support

=back

=item 2. B>

For files that require authentication or authorization, use the XSendfile
middleware to delegate file serving to the reverse proxy:

    use PAGI::Middleware::Builder;

    my $app = builder {
        enable 'XSendfile',
            type    => 'X-Accel-Redirect',  # For Nginx
            mapping => { '/var/www/protected/' => '/internal/' };
        $my_app;
    };

See L for details.

=back

=head1 ENABLING TLS SUPPORT

PAGI::Server supports HTTPS/TLS connections, but requires additional modules
that are not installed by default. This keeps the base installation minimal
for users who don't need TLS.

=head2 When You Need TLS

You need TLS if you want to:

=over 4

=item * Serve HTTPS traffic directly from PAGI::Server

=item * Test TLS locally during development

=item * Use client certificate authentication

=back

You B need TLS if you:

=over 4

=item * Use a reverse proxy (nginx, Apache) that handles TLS termination

=item * Only serve HTTP traffic on localhost for development

=item * Deploy behind a load balancer that provides TLS

=back

B Use a reverse proxy (nginx, HAProxy, etc.) for
TLS termination. They offer better performance, easier certificate management,
and battle-tested security. PAGI::Server's TLS support is primarily for
development and testing.

=head2 Installing TLS Modules

To enable TLS support, install the required modules:

B

    cpanm IO::Async::SSL IO::Socket::SSL

B

    apt-get install libio-socket-ssl-perl

B

    yum install perl-IO-Socket-SSL

B

    perl -MIO::Async::SSL -MIO::Socket::SSL -e 'print "TLS modules installed\n"'

=head2 Basic TLS Configuration

Once the modules are installed, configure TLS with certificate and key files:

    my $server = PAGI::Server->new(
        app  => $app,
        host => '0.0.0.0',
        port => 5000,
        ssl  => {
            cert_file => '/path/to/server.crt',
            key_file  => '/path/to/server.key',
        },
    );

=head2 Generating Self-Signed Certificates (Development)

For local development and testing, you can generate a self-signed certificate:

B

    openssl req -x509 -newkey rsa:4096 -nodes \
        -keyout server.key -out server.crt -days 365 \
        -subj "/CN=localhost"

B

    # Create config file
    cat > ssl.conf <
    [req]
    distinguished_name = req_distinguished_name
    x509_extensions = v3_req
    prompt = no

    [req_distinguished_name]
    CN = localhost

    [v3_req]
    subjectAltName = @alt_names

    [alt_names]
    DNS.1 = localhost
    DNS.2 = *.localhost
    IP.1 = 127.0.0.1
    EOF

    # Generate certificate
    openssl req -x509 -newkey rsa:4096 -nodes \
        -keyout server.key -out server.crt -days 365 \
        -config ssl.conf -extensions v3_req

B

    # Start server
    pagi-server --app myapp.pl --ssl-cert server.crt --ssl-key server.key

    # Test with curl (ignore self-signed cert warning)
    curl -k https://localhost:5000/

B

For production, use certificates from a trusted CA (Let's Encrypt, etc.):

    # Let's Encrypt with certbot
    certbot certonly --standalone -d yourdomain.com

    # Then configure PAGI::Server
    my $server = PAGI::Server->new(
        app => $app,
        ssl => {
            cert_file => '/etc/letsencrypt/live/yourdomain.com/fullchain.pem',
            key_file  => '/etc/letsencrypt/live/yourdomain.com/privkey.pem',
        },
    );

=head2 Advanced TLS Configuration

See the C option in L for details on:

=over 4

=item * Client certificate verification (C, C)

=item * TLS version requirements (C)

=item * Custom cipher suites (C)

=back

=head1 ENABLING HTTP/2 SUPPORT (EXPERIMENTAL)

B The API and behavior may change in future
releases. Please report issues and provide feedback.

PAGI::Server provides native HTTP/2 support via the nghttp2 C library
(L). When enabled, the server supports both TLS-based
HTTP/2 (h2 via ALPN negotiation) and cleartext HTTP/2 (h2c).

=head2 Requirements

L must be installed, which requires the nghttp2 C library:

    # Install the C library
    brew install nghttp2          # macOS
    apt-get install libnghttp2-dev  # Debian/Ubuntu

    # Install the Perl bindings
    cpanm Net::HTTP2::nghttp2

=head2 Enabling HTTP/2

B

    # HTTP/2 over TLS (recommended for production)
    pagi-server --http2 --ssl-cert cert.pem --ssl-key key.pem --app myapp.pl

    # HTTP/2 cleartext (h2c, for development/testing)
    pagi-server --http2 --app myapp.pl

B

    my $server = PAGI::Server->new(
        app   => $app,
        http2 => 1,
        ssl   => { cert_file => 'cert.pem', key_file => 'key.pem' },
    );

=head2 How It Works

With TLS, the server advertises C and C via ALPN during the
TLS handshake. Clients that support HTTP/2 will negotiate C automatically;
others fall back to HTTP/1.1 transparently.

Without TLS, the server detects HTTP/2 via the client connection preface
(h2c mode). HTTP/1.1 clients are handled normally.

=head2 HTTP/2 Features

=over 4

=item * Stream multiplexing (100 concurrent streams per connection by default)

=item * HPACK header compression

=item * Per-stream and connection-level flow control

=item * GOAWAY graceful session shutdown

=item * Stream state validation (RST_STREAM on protocol violations)

=item * WebSocket over HTTP/2 via Extended CONNECT (RFC 8441)

=back

=head2 Conformance

Tested against h2spec (the HTTP/2 conformance test suite): B<137/146 (93.8%)>.
All 9 remaining failures are shared with the bare nghttp2 C library and cannot
be fixed at the application level.

Load tested with h2load: 60,000 requests across 50 concurrent connections with
zero failures.

See L for full compliance details.

=head2 Configuration

HTTP/2 protocol settings are tuned via constructor options prefixed with
C. See L for details on:

=over 4

=item * C - Max streams per connection (default: 100)

=item * C - Flow control window (default: 65535)

=item * C - Max frame payload (default: 16384)

=item * C - Server push (default: disabled)

=item * C - WebSocket over HTTP/2 (default: enabled)

=item * C - Max header block size (default: 65536)

=back

=head1 SIGNAL HANDLING

PAGI::Server responds to Unix signals for process management. Signal behavior
differs between single-worker and multi-worker modes.

=head2 Supported Signals

=over 4

=item B - Graceful shutdown

Initiates graceful shutdown. The server stops accepting new connections,
waits for active requests to complete (up to C seconds),
then exits. In multi-worker mode, SIGTERM is forwarded to all workers.

    kill -TERM

=item B - Graceful shutdown (Ctrl-C)

Same behavior as SIGTERM. Triggered by Ctrl-C in the terminal. In multi-worker
mode, the parent process catches SIGINT and coordinates shutdown of all workers
to ensure proper lifespan.shutdown handling.

    kill -INT
    # or press Ctrl-C in terminal

=item B - Graceful worker restart (multi-worker only)

Performs a zero-downtime worker restart by spawning new workers before
terminating old ones. Useful for recycling workers to reclaim leaked memory
or reset per-worker state without dropping active connections.

B This does NOT reload application code. New workers fork from the
existing parent process and inherit the same loaded code. For code deploys,
perform a full server restart (SIGTERM + start).

    kill -HUP

In single-worker mode, SIGHUP is logged but ignored (no graceful restart
possible without multiple workers).

=item B - Increase worker count (multi-worker only)

Spawns an additional worker process. Use this to scale up capacity dynamically.

    kill -TTIN

=item B - Decrease worker count (multi-worker only)

Gracefully terminates one worker process. The minimum worker count is 1;
sending SIGTTOU when only one worker remains has no effect.

    kill -TTOU

=back

=head2 Signal Handling in Multi-Worker Mode

When running with C<< workers => N >> (where N > 1):

=over 4

=item * Parent process manages the worker pool

=item * Workers handle requests; parent handles signals

=item * SIGTERM/SIGINT to parent triggers coordinated shutdown of all workers

=item * Each worker runs lifespan.shutdown before exiting

=item * Workers that crash are automatically respawned

=item * Heartbeat monitoring detects workers with blocked event loops and replaces them automatically (see C)

=back

=head2 Examples

B

    # Deploy new code, then signal graceful restart
    kill -HUP $(cat /var/run/pagi.pid)

B

    # Add workers during peak hours
    kill -TTIN $(cat /var/run/pagi.pid)
    kill -TTIN $(cat /var/run/pagi.pid)

    # Remove workers during quiet periods
    kill -TTOU $(cat /var/run/pagi.pid)

B

    # Stop accepting new connections, drain existing ones
    kill -TERM $(cat /var/run/pagi.pid)

=cut

sub _init {
    my ($self, $params) = @_;

    $self->{app}              = delete $params->{app} or die "app is required";

    # Extract listener-related params
    my $listen      = delete $params->{listen};
    my $socket      = delete $params->{socket};
    my $socket_mode = delete $params->{socket_mode};
    my $host        = delete $params->{host};
    my $port        = delete $params->{port};
    $self->{ssl}    = delete $params->{ssl};
    $self->{disable_tls}      = delete $params->{disable_tls} // 0;  # Extract early for validation

    # Validate SSL certificate files at startup (fail fast)
    # Skip validation if TLS is explicitly disabled
    if (my $ssl = $self->{ssl}) {
        if ($self->{disable_tls}) {
            # Skip TLS setup and cert validation — ssl config is stored but not applied
            warn "PAGI::Server: TLS disabled via disable_tls option, ssl config ignored\n";
        } else {
            if (my $cert = $ssl->{cert_file}) {
                die "SSL certificate file not found: $cert\n" unless -e $cert;
                die "SSL certificate file not readable: $cert\n" unless -r $cert;
            }
            if (my $key = $ssl->{key_file}) {
                die "SSL key file not found: $key\n" unless -e $key;
                die "SSL key file not readable: $key\n" unless -r $key;
            }
            if (my $ca = $ssl->{ca_file}) {
                die "SSL CA file not found: $ca\n" unless -e $ca;
                die "SSL CA file not readable: $ca\n" unless -r $ca;
            }
        }
    }

    # Normalize all listener forms to $self->{listeners}
    if ($listen) {
        # Explicit listen array
        die "Cannot specify both 'listen' and 'host' options\n" if defined $host;
        die "Cannot specify both 'listen' and 'port' options\n" if defined $port;
        die "Cannot specify both 'listen' and 'socket' options\n" if defined $socket;
        die "Cannot specify both 'listen' and 'socket_mode' options\n" if defined $socket_mode;
        die "'listen' must be a non-empty arrayref\n"
            unless ref $listen eq 'ARRAY' && @$listen;

        $self->{listeners} = [];
        for my $spec (@$listen) {
            die "Each listen spec must be a hashref\n" unless ref $spec eq 'HASH';
            if ($spec->{socket}) {
                die "Cannot specify both 'socket' and 'host' in a listen spec\n" if $spec->{host};
                die "Cannot specify both 'socket' and 'port' in a listen spec\n" if $spec->{port};
                push @{$self->{listeners}}, {
                    type        => 'unix',
                    path        => $spec->{socket},
                    socket_mode => $spec->{socket_mode},
                };
            } else {
                die "TCP listen spec requires both 'host' and 'port'\n"
                    unless defined $spec->{host} && defined $spec->{port};
                push @{$self->{listeners}}, {
                    type => 'tcp',
                    host => $spec->{host},
                    port => $spec->{port},
                };
            }
        }
        $self->{host} = undef;
        $self->{port} = undef;
    } elsif (defined $socket) {
        # Socket sugar
        die "Cannot specify both 'socket' and 'host' options\n" if defined $host;
        die "Cannot specify both 'socket' and 'port' options\n" if defined $port;
        $self->{listeners} = [{
            type        => 'unix',
            path        => $socket,
            socket_mode => $socket_mode,
        }];
        $self->{host} = undef;
        $self->{port} = undef;
    } else {
        # Host/port sugar (backward compatible default)
        $host //= '127.0.0.1';
        $port //= 5000;
        $self->{listeners} = [{
            type => 'tcp',
            host => $host,
            port => $port,
        }];
        $self->{host} = $host;
        $self->{port} = $port;
    }

    # Apply server-wide SSL to all TCP listeners
    if ($self->{ssl}) {
        for my $listener (@{$self->{listeners}}) {
            if ($listener->{type} eq 'tcp') {
                $listener->{ssl} = $self->{ssl};
            }
        }
    }

    $self->{extensions}       = delete $params->{extensions} // {};
    $self->{on_error}         = delete $params->{on_error} // sub { warn @_ };
    $self->{access_log}       = exists $params->{access_log} ? delete $params->{access_log} : \*STDERR;
    $self->{access_log_format} = delete $params->{access_log_format} // 'clf';
    $self->{_access_log_formatter} = $self->_compile_access_log_format(
        $self->{access_log_format}
    );
    $self->{quiet}            = delete $params->{quiet} // 0;
    $self->{log_level}        = delete $params->{log_level} // 'info';
    # Validate log level
    my %valid_levels = (debug => 1, info => 2, warn => 3, error => 4);
    die "Invalid log_level '$self->{log_level}' - must be one of: debug, info, warn, error\n"
        unless $valid_levels{$self->{log_level}};
    $self->{_log_level_num}   = $valid_levels{$self->{log_level}};
    $self->{timeout}          = delete $params->{timeout} // 60;  # Connection idle timeout (seconds)
    $self->{max_header_size}  = delete $params->{max_header_size} // 8192;  # Max header size in bytes
    $self->{max_header_count} = delete $params->{max_header_count} // 100;  # Max number of headers
    $self->{max_body_size}    = delete $params->{max_body_size} // 10_000_000;  # Max body size in bytes (10MB default, 0 = unlimited)
    $self->{workers}          = delete $params->{workers} // 0;   # Number of worker processes (0 = single process)
    $self->{max_requests}     = delete $params->{max_requests} // 0;  # 0 = unlimited
    $self->{listener_backlog} = delete $params->{listener_backlog} // 2048;   # Listener queue size
    $self->{shutdown_timeout}  = delete $params->{shutdown_timeout} // 30;  # Graceful shutdown timeout (seconds)
    $self->{reuseport}         = delete $params->{reuseport} // 0;  # SO_REUSEPORT mode for multi-worker
    $self->{max_receive_queue} = delete $params->{max_receive_queue} // 1000;  # Max WebSocket receive queue size (messages)
    $self->{max_ws_frame_size} = delete $params->{max_ws_frame_size} // 65536;  # Max WebSocket frame size in bytes (64KB default)
    $self->{max_connections}     = delete $params->{max_connections} // 0;  # 0 = use default (1000)
    $self->{sync_file_threshold} = delete $params->{sync_file_threshold} // 65536;  # Threshold for sync file reads (0=always async)
    $self->{request_timeout}     = delete $params->{request_timeout} // 0;  # Request stall timeout in seconds (0 = disabled, default for performance)
    $self->{ws_idle_timeout}     = delete $params->{ws_idle_timeout} // 0;   # WebSocket idle timeout (0 = disabled)
    $self->{sse_idle_timeout}    = delete $params->{sse_idle_timeout} // 0;  # SSE idle timeout (0 = disabled)
    $self->{heartbeat_timeout}   = delete $params->{heartbeat_timeout} // 50;  # Worker heartbeat timeout (0 = disabled)
    $self->{write_high_watermark} = delete $params->{write_high_watermark} // 65536;   # 64KB - pause sending above this
    $self->{write_low_watermark}  = delete $params->{write_low_watermark}  // 16384;   # 16KB - resume sending below this
    $self->{loop_type}           = delete $params->{loop_type};  # Optional loop backend (EPoll, EV, Poll, etc.)
    if (my $lt = $self->{loop_type}) {
        die "Invalid loop_type '$lt': must contain only letters, digits, and ::\n"
            unless $lt =~ /\A[A-Za-z][A-Za-z0-9_]*(?:::[A-Za-z][A-Za-z0-9_]*)*\z/;
    }
    # Dev-mode event validation: explicit flag, or auto-enable in development mode
    $self->{validate_events}     = delete $params->{validate_events}
        // (($ENV{PAGI_ENV} // '') eq 'development' ? 1 : 0);

    # HTTP/2 support (opt-in, experimental)
    $self->{http2} = delete $params->{http2} // $ENV{_PAGI_SERVER_HTTP2} // 0;

    # HTTP/2 protocol settings (only used when http2 is enabled)
    my $h2_max_concurrent_streams  = delete $params->{h2_max_concurrent_streams}  // 100;
    my $h2_initial_window_size     = delete $params->{h2_initial_window_size}     // 65535;
    my $h2_max_frame_size          = delete $params->{h2_max_frame_size}          // 16384;
    my $h2_enable_push             = delete $params->{h2_enable_push}             // 0;
    my $h2_enable_connect_protocol = delete $params->{h2_enable_connect_protocol} // 1;
    my $h2_max_header_list_size    = delete $params->{h2_max_header_list_size}    // 65536;

    $self->{running}     = 0;
    $self->{bound_port}  = undef;
    $self->{listener}    = undef;
    $self->{connections} = {};  # Hash keyed by refaddr for O(1) add/remove
    $self->{protocol}    = PAGI::Server::Protocol::HTTP1->new(
        max_header_size  => $self->{max_header_size},
        max_header_count => $self->{max_header_count},
    );
    $self->{state}       = {};  # Shared state from lifespan
    $self->{worker_pids} = {};  # Track worker PIDs in multi-worker mode
    $self->{is_worker}   = 0;   # True if this is a worker process

    # Initialize HTTP/2 protocol handler if enabled and available
    if ($self->{http2}) {
        if ($HTTP2_AVAILABLE) {
            $self->{http2_protocol} = PAGI::Server::Protocol::HTTP2->new(
                max_concurrent_streams  => $h2_max_concurrent_streams,
                initial_window_size     => $h2_initial_window_size,
                max_frame_size          => $h2_max_frame_size,
                enable_push             => $h2_enable_push,
                enable_connect_protocol => $h2_enable_connect_protocol,
                max_header_list_size    => $h2_max_header_list_size,
            );
            $self->{http2_enabled} = 1;

            # h2c mode: HTTP/2 over cleartext (no TLS)
            if (!$self->{ssl}) {
                $self->{h2c_enabled} = 1;
            }
        } else {
            die <<"END_HTTP2_ERROR";
HTTP/2 support requested but Net::HTTP2::nghttp2 is not installed.

To install:
    cpanm Net::HTTP2::nghttp2

Or disable HTTP/2:
    http2 => 0
END_HTTP2_ERROR
        }
    }

    $self->SUPER::_init($params);
}

sub configure {
    my ($self, %params) = @_;

    if (exists $params{app}) {
        $self->{app} = delete $params{app};
    }
    if (exists $params{host}) {
        $self->{host} = delete $params{host};
    }
    if (exists $params{port}) {
        $self->{port} = delete $params{port};
    }
    if (exists $params{socket}) {
        delete $params{socket};
    }
    if (exists $params{socket_mode}) {
        delete $params{socket_mode};
    }
    if (exists $params{listen}) {
        delete $params{listen};
    }
    if (exists $params{ssl}) {
        $self->{ssl} = delete $params{ssl};
    }
    if (exists $params{extensions}) {
        $self->{extensions} = delete $params{extensions};
    }
    if (exists $params{on_error}) {
        $self->{on_error} = delete $params{on_error};
    }
    if (exists $params{access_log}) {
        $self->{access_log} = delete $params{access_log};
    }
    if (exists $params{access_log_format}) {
        $self->{access_log_format} = delete $params{access_log_format};
        $self->{_access_log_formatter} = $self->_compile_access_log_format(
            $self->{access_log_format}
        );
    }
    if (exists $params{quiet}) {
        $self->{quiet} = delete $params{quiet};
    }
    if (exists $params{log_level}) {
        my $level = delete $params{log_level};
        my %valid_levels = (debug => 1, info => 2, warn => 3, error => 4);
        die "Invalid log_level '$level' - must be one of: debug, info, warn, error\n"
            unless $valid_levels{$level};
        $self->{log_level} = $level;
        $self->{_log_level_num} = $valid_levels{$level};
    }
    if (exists $params{timeout}) {
        $self->{timeout} = delete $params{timeout};
    }
    if (exists $params{max_header_size}) {
        $self->{max_header_size} = delete $params{max_header_size};
    }
    if (exists $params{max_header_count}) {
        $self->{max_header_count} = delete $params{max_header_count};
    }
    if (exists $params{max_body_size}) {
        $self->{max_body_size} = delete $params{max_body_size};
    }
    if (exists $params{workers}) {
        $self->{workers} = delete $params{workers};
    }
    if (exists $params{max_requests}) {
        $self->{max_requests} = delete $params{max_requests};
    }
    if (exists $params{listener_backlog}) {
        $self->{listener_backlog} = delete $params{listener_backlog};
    }
    if (exists $params{shutdown_timeout}) {
        $self->{shutdown_timeout} = delete $params{shutdown_timeout};
    }
    if (exists $params{max_receive_queue}) {
        $self->{max_receive_queue} = delete $params{max_receive_queue};
    }
    if (exists $params{max_ws_frame_size}) {
        $self->{max_ws_frame_size} = delete $params{max_ws_frame_size};
    }
    if (exists $params{max_connections}) {
        $self->{max_connections} = delete $params{max_connections};
    }
    if (exists $params{request_timeout}) {
        $self->{request_timeout} = delete $params{request_timeout};
    }
    if (exists $params{ws_idle_timeout}) {
        $self->{ws_idle_timeout} = delete $params{ws_idle_timeout};
    }
    if (exists $params{sse_idle_timeout}) {
        $self->{sse_idle_timeout} = delete $params{sse_idle_timeout};
    }
    if (exists $params{http2}) {
        $self->{http2} = delete $params{http2};
    }

    $self->SUPER::configure(%params);
}

# Log levels: debug=1, info=2, warn=3, error=4
my %_LOG_LEVELS = (debug => 1, info => 2, warn => 3, error => 4);

sub _log {
    my ($self, $level, $msg) = @_;

    my $level_num = $_LOG_LEVELS{$level} // 2;
    return if $level_num < $self->{_log_level_num};
    return if $self->{quiet} && $level ne 'error';
    warn "$msg\n";
}

# Returns a human-readable TLS status string for the startup banner
sub _tls_status_string {
    my ($self) = @_;

    if ($self->{disable_tls}) {
        return $TLS_AVAILABLE ? 'disabled' : 'n/a (disabled)';
    }
    if ($self->{tls_enabled}) {
        return 'on';
    }
    return $TLS_AVAILABLE ? 'available' : 'not installed';
}

# Returns a human-readable HTTP/2 status string for the startup banner
sub _http2_status_string {
    my ($self) = @_;

    if ($self->{http2_enabled}) {
        return $self->{h2c_enabled} ? 'on (h2c)' : 'on';
    }
    return $HTTP2_AVAILABLE ? 'available' : 'not installed';
}

# Returns a human-readable Future::XS status string for the startup banner
sub _future_xs_status_string {
    return 'on' if $FUTURE_XS_ENABLED;
    return 'available' if $FUTURE_XS_AVAILABLE;
    return 'not installed';
}

# Check if TLS modules are available
sub _check_tls_available {
    my ($self) = @_;

    # Allow forcing TLS off for testing — return false to skip TLS setup
    if ($self->{disable_tls}) {
        return 0;
    }

    return 1 if $TLS_AVAILABLE;

    die <<"END_TLS_ERROR";
TLS support requested but required modules not installed.

To enable HTTPS/TLS support, install:

    cpanm IO::Async::SSL IO::Socket::SSL

Or on Debian/Ubuntu:

    apt-get install libio-socket-ssl-perl

Then restart your application.
END_TLS_ERROR
}

# Build SSL configuration parameters for use by both single-worker and multi-worker modes.
# Returns a hashref of SSL params (including SSL_reuse_ctx) or undef if no SSL configured.
sub _build_ssl_config {
    my ($self) = @_;
    my $ssl = $self->{ssl} or return;

    return unless $self->_check_tls_available;

    my %ssl_params;
    $ssl_params{SSL_server}      = 1;
    $ssl_params{SSL_cert_file}   = $ssl->{cert_file} if $ssl->{cert_file};
    $ssl_params{SSL_key_file}    = $ssl->{key_file}  if $ssl->{key_file};
    # Trailing colon means "this version or higher" — allows TLS 1.3 negotiation
    $ssl_params{SSL_version}     = ($ssl->{min_version} // 'TLSv1_2') . ':';
    $ssl_params{SSL_cipher_list} = $ssl->{cipher_list}
        // 'ECDHE+AESGCM:DHE+AESGCM:ECDHE+CHACHA20:!aNULL:!eNULL:!EXPORT:!DES:!RC4:!MD5:!PSK:!aECDH:!EDH-DSS-DES-CBC3-SHA:!KRB5-DES-CBC3-SHA';

    if ($ssl->{verify_client}) {
        # SSL_VERIFY_PEER (0x01) | SSL_VERIFY_FAIL_IF_NO_PEER_CERT (0x02)
        $ssl_params{SSL_verify_mode} = 0x03;
        $ssl_params{SSL_ca_file} = $ssl->{ca_file} if $ssl->{ca_file};
    } else {
        $ssl_params{SSL_verify_mode} = 0x00;  # SSL_VERIFY_NONE
    }

    # ALPN negotiation for HTTP/2 support
    if ($self->{http2} && $HTTP2_AVAILABLE) {
        $ssl_params{SSL_alpn_protocols} = ['h2', 'http/1.1'];
        $self->{http2_enabled} = 1;
        $self->{h2c_enabled} = 0;  # TLS mode, not cleartext
    }

    # Pre-create shared SSL context to avoid per-connection CA bundle parsing
    my $ssl_ctx = IO::Socket::SSL::SSL_Context->new(\%ssl_params);
    $self->{_ssl_ctx} = $ssl_ctx;
    $ssl_params{SSL_reuse_ctx} = $ssl_ctx;

    # Mark TLS enabled and auto-add tls extension
    $self->{tls_enabled} = 1;
    $self->{extensions}{tls} = {} unless exists $self->{extensions}{tls};

    return \%ssl_params;
}

=head2 run

    $server->run;

Standalone entry point that creates an event loop, starts the server,
and runs until shutdown. This is the simplest way to run a PAGI server:

    my $server = PAGI::Server->new(
        app  => $app,
        port => 8080,
    );
    $server->run;

For embedding in an existing IO::Async application, use the traditional
pattern instead:

    my $loop = IO::Async::Loop->new;
    $loop->add($server);
    $server->listen->get;
    $loop->run;

The C method handles:

=over 4

=item * Creating the event loop (respecting C if set)

=item * Adding the server to the loop

=item * Starting the listener

=item * Setting up signal handlers for graceful shutdown

=item * Running the event loop until shutdown

=back

=cut

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

    my $loop = $self->_create_loop;
    $loop->add($self);

    # Start listening with error handling
    eval { $self->listen->get };
    if ($@) {
        my $error = $@;
        my $port = $self->{port};
        if ($error =~ /Address already in use/i) {
            die "Error: Port $port is already in use\n";
        }
        elsif ($error =~ /Permission denied/i) {
            die "Error: Permission denied to bind to port $port\n";
        }
        die "Error starting server: $error\n";
    }

    # Run the event loop (signal handlers were set up by listen())
    $loop->run;
}

# Create an event loop, respecting loop_type config
sub _create_loop {
    my ($self) = @_;

    if (my $loop_type = $self->{loop_type}) {
        die "Invalid loop_type '$loop_type': must contain only letters, digits, and ::\n"
            unless $loop_type =~ /\A[A-Za-z][A-Za-z0-9_]*(?:::[A-Za-z][A-Za-z0-9_]*)*\z/;
        my $loop_class = "IO::Async::Loop::$loop_type";
        (my $loop_file = "$loop_class.pm") =~ s{::}{/}g;
        eval { require $loop_file }
            or die "Cannot load loop backend '$loop_type': $@\n" .
                   "Install it with: cpanm $loop_class\n";
        return $loop_class->new;
    }

    require IO::Async::Loop;
    return IO::Async::Loop->new;
}

async sub listen {
    my ($self) = @_;

    return if $self->{running};

    # Multi-worker mode uses a completely different code path
    if ($self->{workers} && $self->{workers} > 0) {
        return $self->_listen_multiworker;
    }

    return await $self->_listen_singleworker;
}

# Single-worker mode - uses IO::Async normally
async sub _listen_singleworker {
    my ($self) = @_;

    weaken(my $weak_self = $self);

    # Run lifespan startup before accepting connections
    my $startup_result = await $self->_run_lifespan_startup;

    if (!$startup_result->{success}) {
        my $message = $startup_result->{message} // 'Lifespan startup failed';
        $self->_log(error => "PAGI Server startup failed: $message");
        die "Lifespan startup failed: $message\n";
    }

    # Collect any inherited fds (from PAGI_REUSE or LISTEN_FDS)
    my $inherited = $self->_collect_inherited_fds;

    # Iterate over listeners array, creating one IO::Async::Listener per spec
    my @listen_entries;
    for my $spec (@{$self->{listeners}}) {

        # Check for inherited fd matching this spec
        my $match_key = $spec->{type} eq 'unix'
            ? "unix:$spec->{path}"
            : "$spec->{host}:$spec->{port}";

        if (my $inh = delete $inherited->{$match_key}) {
            # Reuse inherited fd — skip bind/listen entirely
            $spec->{_inherited} = 1;

            my $handle = $inh->{handle};
            if (!$handle) {
                my $class = $inh->{type} eq 'unix'
                    ? 'IO::Socket::UNIX' : 'IO::Socket::INET';
                require IO::Socket::UNIX if $inh->{type} eq 'unix';
                $handle = $class->new_from_fd($inh->{fd}, 'r')
                    or die "Cannot open inherited fd $inh->{fd}: $!\n";
            }

            if ($inh->{type} eq 'tcp' && $handle->can('sockport')) {
                $spec->{port} = $handle->sockport;
                $self->{bound_port} //= $spec->{port};
            }

            my $spec_ref = $spec;
            weaken(my $weak_inner = $self);
            my $listener = IO::Async::Listener->new(
                handle    => $handle,
                on_stream => sub {
                    my ($l, $stream) = @_;
                    return unless $weak_inner;
                    $weak_inner->_on_connection($stream, $spec_ref);
                },
            );
            $self->add_child($listener);

            $self->_log(info => "Reusing inherited fd $inh->{fd} for $match_key"
                . " (source: $inh->{source})");

            push @listen_entries, { listener => $listener, spec => $spec };
            next;  # Skip normal bind/listen
        }

        my $spec_copy = $spec;  # capture for closure
        my $listener = IO::Async::Listener->new(
            on_stream => sub {
                my ($listener, $stream) = @_;
                return unless $weak_self;
                $weak_self->_on_connection($stream, $spec_copy);
            },
        );

        $self->add_child($listener);

        # Build listener options
        my %listen_opts = (
            queuesize => $self->{listener_backlog},
        );

        if ($spec->{type} eq 'unix') {
            # Remove stale socket file if it exists
            unlink $spec->{path} if -e $spec->{path};

            $listen_opts{addr} = {
                family   => 'unix',
                socktype => 'stream',
                path     => $spec->{path},
            };

            if ($self->{tls_enabled}) {
                $self->_log(info => "Note: TLS is configured but does not apply to Unix socket $spec->{path}");
            }
        } else {
            # TCP listener
            $listen_opts{addr} = {
                family   => 'inet',
                socktype => 'stream',
                ip       => $spec->{host},
                port     => $spec->{port},
            };

            # Add SSL options if configured (TCP only)
            if (my $ssl_params = $self->_build_ssl_config) {
                $listen_opts{extensions} = ['SSL'];
                %listen_opts = (%listen_opts, %$ssl_params);

                $listen_opts{on_ssl_error} = sub {
                    return unless $weak_self;
                    $weak_self->_log(debug => "SSL handshake failed: $_[0]");
                };
            }
        }

        # Set restrictive umask for Unix socket bind to prevent brief
        # permission window (CVE-2023-45145 pattern in Redis)
        my $old_umask;
        if ($spec->{type} eq 'unix') {
            $old_umask = umask(0177);  # Owner-only until chmod
        }

        # Start listening ($^F raised so fd survives exec for hot restart)
        {
            local $^F = 1023;
            await $listener->listen(%listen_opts);
        }

        # Restore umask after bind
        umask($old_umask) if defined $old_umask;

        # Configure accept error handler after listen() to avoid SSL extension conflicts
        eval {
            $listener->configure(
                on_accept_error => sub {
                    my ($listener, $error) = @_;
                    return unless $weak_self;
                    $weak_self->_on_accept_error($error);
                },
            );
        };
        if ($@) {
            $self->_log(debug => "Could not configure on_accept_error (likely SSL listener): $@");
        }

        # Post-listen setup
        if ($spec->{type} eq 'unix') {
            # Apply socket permissions if configured
            if (defined $spec->{socket_mode}) {
                chmod $spec->{socket_mode}, $spec->{path};
            }
        } else {
            # Store the actual bound port from the listener's read handle
            my $socket = $listener->read_handle;
            if ($socket && $socket->can('sockport')) {
                my $bound = $socket->sockport;
                $spec->{port} = $bound;  # update spec with actual port
                $self->{bound_port} //= $bound;  # first TCP port wins
            }
        }

        # Register in PAGI_REUSE for hot restart fd inheritance
        my $rh = $listener->read_handle;
        if ($rh) {
            my $fd = fileno($rh);
            if (defined $fd) {
                my $reuse_key = $spec->{type} eq 'unix'
                    ? "unix:$spec->{path}:$fd"
                    : "$spec->{host}:$spec->{port}:$fd";
                $spec->{_reuse_key} = $reuse_key;
                $ENV{PAGI_REUSE} = length($ENV{PAGI_REUSE} // '')
                    ? "$ENV{PAGI_REUSE},$reuse_key"
                    : $reuse_key;
            }
        }

        push @listen_entries, { listener => $listener, spec => $spec };
    }

    # Warn about unmatched inherited fds
    for my $key (sort keys %$inherited) {
        my $inh = $inherited->{$key};
        $self->_log(warn => "Inherited fd $inh->{fd} ($key) does not match "
            . "any listener spec — closing");
        if ($inh->{handle}) {
            close($inh->{handle});
        } else {
            POSIX::close($inh->{fd});
        }
    }

    $self->{_listen_entries} = \@listen_entries;
    # Backward compat: keep $self->{listener} pointing to first entry
    $self->{listener} = $listen_entries[0]{listener} if @listen_entries;
    $self->{running} = 1;

    # Set up signal handlers for graceful shutdown (single-worker mode)
    # Note: Windows doesn't support Unix signals, so this is skipped there
    unless (WIN32) {
        my $shutdown_triggered = 0;
        my $shutdown_handler = sub {
            return if $shutdown_triggered;
            $shutdown_triggered = 1;
            $self->adopt_future(
                $self->shutdown->on_done(sub {
                    $self->loop->stop;
                })->on_fail(sub {
                    my ($error) = @_;
                    $self->_log(error => "Shutdown error: $error");
                    $self->loop->stop;  # Still stop even on error
                })
            );
        };
        $self->loop->watch_signal(TERM => $shutdown_handler);
        $self->loop->watch_signal(INT => $shutdown_handler);

        # HUP in single-worker mode just warns (graceful restart requires multi-worker)
        my $weak_self = $self;
        weaken($weak_self);
        $self->loop->watch_signal(HUP => sub {
            $weak_self->_log(warn => "Received HUP signal (graceful restart only works in multi-worker mode)")
                if $weak_self && !$weak_self->{quiet};
        });

        $self->loop->watch_signal(USR2 => sub {
            $weak_self->_log(warn => "Received USR2 signal (hot restart only works in multi-worker mode)")
                if $weak_self && !$weak_self->{quiet};
        });
    }

    my $loop_class = ref($self->loop);
    $loop_class =~ s/^IO::Async::Loop:://;  # Shorten for display
    my $max_conn = $self->effective_max_connections;
    my $tls_status = $self->_tls_status_string;
    my $http2_status = $self->_http2_status_string;
    my $future_xs_status = $self->_future_xs_status_string;

    # Warn if access_log is a terminal (slow for benchmarks)
    if ($self->{access_log} && -t $self->{access_log}) {
        $self->_log(warn =>
            "access_log is a terminal; this may impact performance. " .
            "Consider redirecting to a file or setting access_log => undef for benchmarks."
        );
    }

    # Log listening banner
    my $scheme = $self->{tls_enabled} ? 'https' : 'http';
    if (@listen_entries == 1) {
        my $spec = $listen_entries[0]{spec};
        if ($spec->{type} eq 'unix') {
            $self->_log(info => "PAGI Server listening on unix:$spec->{path} (loop: $loop_class, max_conn: $max_conn, http2: $http2_status, tls: $tls_status, future_xs: $future_xs_status)");
        } else {
            $self->_log(info => "PAGI Server listening on $scheme://$spec->{host}:$spec->{port}/ (loop: $loop_class, max_conn: $max_conn, http2: $http2_status, tls: $tls_status, future_xs: $future_xs_status)");
        }
    } else {
        my @addrs;
        for my $entry (@listen_entries) {
            my $s = $entry->{spec};
            if ($s->{type} eq 'unix') {
                push @addrs, "unix:$s->{path}";
            } else {
                push @addrs, "$scheme://$s->{host}:$s->{port}/";
            }
        }
        $self->_log(info => "PAGI Server listening on: " . join(', ', @addrs) . " (loop: $loop_class, max_conn: $max_conn, http2: $http2_status, tls: $tls_status, future_xs: $future_xs_status)");
    }

    # Warn in production if using default max_connections
    if (($ENV{PAGI_ENV} // '') eq 'production' && !$self->{max_connections}) {
        $self->_log(warn =>
            "Using default max_connections (1000). For production, consider tuning this value " .
            "based on your workload. See 'perldoc PAGI::Server' for guidance."
        );
    }

    return $self;
}

# Multi-worker mode - forks workers, each with their own event loop
sub _listen_multiworker {
    my ($self) = @_;

    my $workers = $self->{workers};
    my $reuseport = $self->{reuseport};

    # Create all listening sockets before forking workers
    my @listen_entries;

    # Collect any inherited fds (from PAGI_REUSE or LISTEN_FDS)
    my $inherited = $self->_collect_inherited_fds;

    for my $spec (@{$self->{listeners}}) {
        my $socket;

        # Check for inherited fd matching this spec
        my $match_key = $spec->{type} eq 'unix'
            ? "unix:$spec->{path}"
            : "$spec->{host}:$spec->{port}";

        if (my $inh = delete $inherited->{$match_key}) {
            $spec->{_inherited} = 1;

            if ($inh->{handle}) {
                $socket = $inh->{handle};
            } else {
                my $class = $inh->{type} eq 'unix'
                    ? 'IO::Socket::UNIX' : 'IO::Socket::INET';
                require IO::Socket::UNIX if $inh->{type} eq 'unix';
                $socket = $class->new_from_fd($inh->{fd}, 'r')
                    or die "Cannot open inherited fd $inh->{fd}: $!\n";
            }

            if ($inh->{type} eq 'tcp' && $socket->can('sockport')) {
                $spec->{bound_port} = $socket->sockport;
                $self->{bound_port} //= $spec->{bound_port};
            }

            $self->_log(info => "Reusing inherited fd $inh->{fd} for $match_key"
                . " (source: $inh->{source})");
        }
        elsif ($spec->{type} eq 'unix') {
            # Unix socket: parent creates, workers inherit
            unlink $spec->{path} if -e $spec->{path};

            # Set restrictive umask for bind (CVE-2023-45145 mitigation)
            my $old_umask = umask(0177);

            require IO::Socket::UNIX;
            {
                local $^F = 1023;
                $socket = IO::Socket::UNIX->new(
                    Local   => $spec->{path},
                    Type    => Socket::SOCK_STREAM(),
                    Listen  => $self->{listener_backlog},
                ) or die "Cannot create Unix socket $spec->{path}: $!";
            }

            umask($old_umask);

            if (defined $spec->{socket_mode}) {
                chmod($spec->{socket_mode}, $spec->{path})
                    or die "Cannot chmod $spec->{path}: $!\n";
            }

            # Register in PAGI_REUSE for hot restart fd inheritance
            if ($socket && !$spec->{_inherited}) {
                my $fd = fileno($socket);
                my $reuse_key = "unix:$spec->{path}:$fd";
                $spec->{_reuse_key} = $reuse_key;
                $ENV{PAGI_REUSE} = length($ENV{PAGI_REUSE} // '')
                    ? "$ENV{PAGI_REUSE},$reuse_key"
                    : $reuse_key;
            }
        } elsif ($reuseport) {
            # reuseport TCP: probe to get port, workers create their own
            # Note: reuseport sockets are not registered in PAGI_REUSE because
            # each worker creates its own socket. fd inheritance for reuseport
            # mode is not currently supported — use shared-socket mode for
            # hot restart / systemd socket activation.
            my $probe_socket = IO::Socket::INET->new(
                LocalAddr => $spec->{host},
                LocalPort => $spec->{port},
                Proto     => 'tcp',
                Listen    => 1,
                ReuseAddr => 1,
                ReusePort => 1,
            ) or die "Cannot bind to $spec->{host}:$spec->{port}: $!";
            $spec->{bound_port} = $probe_socket->sockport;
            $self->{bound_port} //= $spec->{bound_port};
            close($probe_socket);
        } else {
            # Shared-socket TCP: parent creates, workers inherit
            {
                local $^F = 1023;
                $socket = IO::Socket::INET->new(
                    LocalAddr => $spec->{host},
                    LocalPort => $spec->{port},
                    Proto     => 'tcp',
                    Listen    => $self->{listener_backlog},
                    ReuseAddr => 1,
                    Blocking  => 0,
                ) or die "Cannot create listening socket on $spec->{host}:$spec->{port}: $!";
            }
            $spec->{bound_port} = $socket->sockport;
            $self->{bound_port} //= $spec->{bound_port};

            # Register in PAGI_REUSE for hot restart fd inheritance
            if ($socket && !$spec->{_inherited}) {
                my $fd = fileno($socket);
                my $reuse_key = "$spec->{host}:" . $socket->sockport . ":$fd";
                $spec->{_reuse_key} = $reuse_key;
                $ENV{PAGI_REUSE} = length($ENV{PAGI_REUSE} // '')
                    ? "$ENV{PAGI_REUSE},$reuse_key"
                    : $reuse_key;
            }
        }

        push @listen_entries, { socket => $socket, spec => $spec };
    }

    # Warn about unmatched inherited fds
    for my $key (sort keys %$inherited) {
        my $inh = $inherited->{$key};
        $self->_log(warn => "Inherited fd $inh->{fd} ($key) does not match "
            . "any listener spec — closing");
        if ($inh->{handle}) {
            close($inh->{handle});
        } else {
            POSIX::close($inh->{fd});
        }
    }

    $self->{_listen_entries} = \@listen_entries;
    # Backward compat: keep listen_socket pointing to first entry's socket
    $self->{listen_socket} = $listen_entries[0]{socket} if @listen_entries && $listen_entries[0]{socket};

    $self->{running} = 1;

    # Validate TLS modules and set tls_enabled before forking workers
    if ($self->{ssl}) {
        $self->_check_tls_available;
        $self->{tls_enabled} = 1;
    }

    my $scheme = $self->{ssl} ? 'https' : 'http';
    my $loop_class = ref($self->loop);
    $loop_class =~ s/^IO::Async::Loop:://;  # Shorten for display
    my $mode = $reuseport ? 'reuseport' : 'shared-socket';
    my $max_conn = $self->effective_max_connections;
    my $tls_status = $self->_tls_status_string;
    my $http2_status = $self->_http2_status_string;
    my $future_xs_status = $self->_future_xs_status_string;

    # Warn if access_log is a terminal (slow for benchmarks)
    if ($self->{access_log} && -t $self->{access_log}) {
        $self->_log(warn =>
            "access_log is a terminal; this may impact performance. " .
            "Consider redirecting to a file or setting access_log => undef for benchmarks."
        );
    }

    # Log listening banner for all listeners
    my @addrs;
    for my $entry (@listen_entries) {
        my $s = $entry->{spec};
        if ($s->{type} eq 'unix') {
            push @addrs, "unix:$s->{path}";
        } else {
            my $port = $s->{bound_port} // $s->{port};
            push @addrs, "$scheme://$s->{host}:$port/";
        }
    }
    my $addr_str = join(', ', @addrs);
    $self->_log(info => "PAGI Server (multi-worker, $mode) listening on $addr_str with $workers workers (loop: $loop_class, max_conn: $max_conn/worker, http2: $http2_status, tls: $tls_status, future_xs: $future_xs_status)");

    # Warn in production if using default max_connections
    if (($ENV{PAGI_ENV} // '') eq 'production' && !$self->{max_connections}) {
        $self->_log(warn =>
            "Using default max_connections (1000). For production, consider tuning this value " .
            "based on your workload. See 'perldoc PAGI::Server' for guidance."
        );
    }

    my $loop = $self->loop;

    # Fork the workers FIRST, before setting up signal handlers.
    # This prevents children from inheriting the parent's sigpipe setup,
    # which can cause issues with Ctrl-C signal delivery on macOS.
    for my $i (1 .. $workers) {
        $self->_spawn_worker(\@listen_entries, $i);
    }

    # Set up signal handlers for parent process AFTER forking
    # Note: Windows doesn't support Unix signals, so this is skipped there
    unless (WIN32) {
        $loop->watch_signal(TERM => sub { $self->_initiate_multiworker_shutdown });
        $loop->watch_signal(INT  => sub { $self->_initiate_multiworker_shutdown });
        $loop->watch_signal(HUP => sub { $self->_graceful_restart });
        $loop->watch_signal(TTIN => sub { $self->_increase_workers });
        $loop->watch_signal(TTOU => sub { $self->_decrease_workers });
        $loop->watch_signal(USR2 => sub { $self->_hot_restart });
    }

    # Start heartbeat monitor if enabled
    if ($self->{heartbeat_timeout} && $self->{heartbeat_timeout} > 0) {
        my $hb_timeout = $self->{heartbeat_timeout};
        my $check_interval = $hb_timeout / 2;
        weaken(my $weak_self = $self);

        my $hb_check_timer = IO::Async::Timer::Periodic->new(
            interval => $check_interval,
            on_tick  => sub {
                return unless $weak_self;
                return if $weak_self->{shutting_down};

                my $now = time();
                for my $pid (keys %{$weak_self->{worker_pids}}) {
                    my $info = $weak_self->{worker_pids}{$pid};
                    next unless $info->{heartbeat_rd};

                    # Drain all available heartbeat bytes
                    while (sysread($info->{heartbeat_rd}, my $buf, 64)) {
                        $info->{last_heartbeat} = $now;
                    }

                    # Kill if heartbeat expired
                    if ($now - $info->{last_heartbeat} > $hb_timeout) {
                        $weak_self->_log(warn =>
                            "Worker $pid (worker $info->{worker_num}) heartbeat " .
                            "timeout after ${hb_timeout}s, sending SIGKILL");
                        kill 'KILL', $pid;
                    }
                }
            },
        );

        $self->add_child($hb_check_timer);
        $hb_check_timer->start;
        $self->{_heartbeat_check_timer} = $hb_check_timer;
    }

    # Hot restart handoff: if we were spawned by USR2, signal the old master
    if (my $old_master_pid = delete $ENV{PAGI_MASTER_PID}) {
        $old_master_pid = int($old_master_pid);

        # Wait for workers to be healthy before retiring old master
        # Delay = half the heartbeat timeout + 1 second buffer
        my $handoff_delay = ($self->{heartbeat_timeout} || 10) / 2 + 1;
        weaken(my $weak_self_handoff = $self);

        $self->loop->watch_time(
            after => $handoff_delay,
            code  => sub {
                return unless $weak_self_handoff;

                my $worker_count = scalar keys %{$weak_self_handoff->{worker_pids}};
                if ($worker_count == 0) {
                    $weak_self_handoff->_log(error =>
                        "Hot restart: no workers running, not retiring old master $old_master_pid");
                    return;
                }

                if (kill(0, $old_master_pid)) {
                    $weak_self_handoff->_log(info =>
                        "Hot restart: $worker_count workers healthy, "
                        . "sending SIGTERM to old master $old_master_pid");
                    kill('TERM', $old_master_pid);
                } else {
                    $weak_self_handoff->_log(warn =>
                        "Hot restart: old master $old_master_pid is no longer running");
                }
            },
        );
    }

    # Return immediately - caller (Runner) will call $loop->run()
    # This is consistent with single-worker mode behavior
    return $self;
}

# Collect inherited file descriptors from PAGI_REUSE and LISTEN_FDS.
# Returns a hashref keyed by "host:port" or "unix:path", values are
# { fd, type, host/port/path, handle?, source }
sub _collect_inherited_fds {
    my ($self) = @_;
    my %inherited;

    # Source 1: PAGI_REUSE (format: addr:port:fd,unix:path:fd,...)
    if (my $reuse = $ENV{PAGI_REUSE}) {
        for my $entry (split /,/, $reuse) {
            if ($entry =~ /^unix:(.+):(\d+)$/) {
                my ($path, $fd) = ($1, int($2));
                $inherited{"unix:$path"} = {
                    fd => $fd, type => 'unix', path => $path,
                    source => 'pagi_reuse',
                };
            } elsif ($entry =~ /^(\[.+?\]):(\d+):(\d+)$/) {
                my ($host, $port, $fd) = ($1, int($2), int($3));
                $inherited{"$host:$port"} = {
                    fd => $fd, type => 'tcp', host => $host, port => $port,
                    source => 'pagi_reuse',
                };
            } elsif ($entry =~ /^(.+):(\d+):(\d+)$/) {
                my ($host, $port, $fd) = ($1, int($2), int($3));
                $inherited{"$host:$port"} = {
                    fd => $fd, type => 'tcp', host => $host, port => $port,
                    source => 'pagi_reuse',
                };
            }
            # Malformed entries silently skipped
        }
    }

    # Source 2: LISTEN_FDS (systemd socket activation)
    my $listen_fds = $ENV{LISTEN_FDS};
    if (defined $listen_fds && $listen_fds =~ /^\d+$/ && $listen_fds > 0) {
        if (defined $ENV{LISTEN_PID} && $ENV{LISTEN_PID} == $$) {
            my $n = int($listen_fds);
            for my $i (0 .. $n - 1) {
                my $fd = 3 + $i;  # SD_LISTEN_FDS_START

                my $fh;
                unless (open($fh, '+<&=', $fd)) {
                    $self->_log(warn => "Cannot fdopen inherited fd $fd: $!");
                    next;
                }

                my $addr = getsockname($fh);
                unless ($addr) {
                    $self->_log(warn => "Cannot getsockname on inherited fd $fd: $!");
                    next;
                }

                my $family = sockaddr_family($addr);

                if ($family == AF_UNIX) {
                    my $path = unpack_sockaddr_un($addr);
                    my $key = "unix:$path";
                    $inherited{$key} //= {
                        fd => $fd, type => 'unix', path => $path,
                        handle => $fh, source => 'systemd',
                    };
                } elsif ($family == AF_INET) {
                    my ($port, $host_packed) = unpack_sockaddr_in($addr);
                    my $host = Socket::inet_ntoa($host_packed);
                    my $key = "$host:$port";
                    $inherited{$key} //= {
                        fd => $fd, type => 'tcp', host => $host, port => $port,
                        handle => $fh, source => 'systemd',
                    };
                } else {
                    $self->_log(warn =>
                        "Inherited fd $fd has unsupported address family $family");
                }
            }
        }

        # Always clean up systemd env vars (per sd_listen_fds spec)
        delete @ENV{qw(LISTEN_FDS LISTEN_PID LISTEN_FDNAMES)};
    }

    return \%inherited;
}

# Initiate graceful shutdown in multi-worker mode
sub _initiate_multiworker_shutdown {
    my ($self) = @_;

    return if $self->{shutting_down};
    $self->{shutting_down} = 1;
    $self->{running} = 0;

    # Stop heartbeat monitoring — shutdown escalation timer handles stuck workers
    if ($self->{_heartbeat_check_timer}) {
        $self->{_heartbeat_check_timer}->stop;
        $self->remove_child($self->{_heartbeat_check_timer});
        delete $self->{_heartbeat_check_timer};
    }

    # Close all listen sockets to stop accepting new connections
    # (skip during hot restart — new master is using these fds)
    if (!$self->{_hot_restart_in_progress}) {
        for my $entry (@{$self->{_listen_entries} // []}) {
            if ($entry->{socket}) {
                close($entry->{socket});
            }
        }
        if ($self->{listen_socket}) {
            delete $self->{listen_socket};
        }
    }

    # Clean up PAGI_REUSE entries (skip during hot restart)
    if (!$self->{_hot_restart_in_progress}) {
        for my $entry (@{$self->{_listen_entries} // []}) {
            my $key = $entry->{spec}{_reuse_key};
            if ($key && defined $ENV{PAGI_REUSE}) {
                $ENV{PAGI_REUSE} =~ s/(?:^|,)\Q$key\E//;
                $ENV{PAGI_REUSE} =~ s/^,// if defined $ENV{PAGI_REUSE};
            }
        }
    }

    # Clean up Unix socket files (skip inherited, skip during hot restart)
    if (!$self->{_hot_restart_in_progress}) {
        for my $entry (@{$self->{_listen_entries} // []}) {
            if ($entry->{spec}{type} eq 'unix'
                && !$entry->{spec}{_inherited}
                && -e $entry->{spec}{path}) {
                unlink $entry->{spec}{path};
            }
        }
    }

    # Signal all workers to shutdown
    for my $pid (keys %{$self->{worker_pids}}) {
        kill 'TERM', $pid;
    }

    # If no workers, stop the loop immediately
    if (!keys %{$self->{worker_pids}}) {
        $self->loop->stop;
        return;
    }

    # Escalate to SIGKILL after shutdown_timeout for workers that ignore SIGTERM
    my $timeout = $self->{shutdown_timeout} // 30;
    weaken(my $weak_self = $self);
    $self->{_shutdown_kill_timer} = $self->loop->watch_time(
        after => $timeout,
        code  => sub {
            return unless $weak_self;
            for my $pid (keys %{$weak_self->{worker_pids}}) {
                $weak_self->_log(warn =>
                    "Worker $pid did not exit after ${timeout}s, sending SIGKILL");
                kill 'KILL', $pid;
            }
        },
    );
}

# Graceful restart: replace all workers one by one
sub _graceful_restart {
    my ($self) = @_;

    return if $self->{shutting_down};

    $self->_log(info => "Received HUP, performing graceful restart");

    # Signal all current workers to shutdown
    # watch_process callbacks will respawn them
    my $timeout = $self->{shutdown_timeout} // 30;
    for my $pid (keys %{$self->{worker_pids}}) {
        kill 'TERM', $pid;

        # Escalate to SIGKILL if worker doesn't exit within shutdown_timeout
        weaken(my $weak_self = $self);
        $self->{_restart_kill_timers}{$pid} = $self->loop->watch_time(
            after => $timeout,
            code  => sub {
                return unless $weak_self;
                if (exists $weak_self->{worker_pids}{$pid}) {
                    $weak_self->_log(warn =>
                        "Worker $pid did not exit after ${timeout}s during restart, sending SIGKILL");
                    kill 'KILL', $pid;
                }
            },
        );
    }
}

# Hot restart: fork+exec a new master that inherits listen sockets via PAGI_REUSE
sub _hot_restart {
    my ($self) = @_;

    if ($self->{_hot_restart_in_progress}) {
        $self->_log(warn => "Hot restart already in progress, ignoring USR2");
        return;
    }

    if ($self->{shutting_down}) {
        $self->_log(warn => "Server is shutting down, ignoring USR2");
        return;
    }

    $self->{_hot_restart_in_progress} = 1;
    $self->_log(info => "Received USR2, starting hot restart");

    # Store our PID so the new master can signal us when ready
    $ENV{PAGI_MASTER_PID} = $$;

    # Fork and exec a new master process
    my $pid = fork();

    if (!defined $pid) {
        $self->_log(error => "Hot restart fork failed: $!");
        $self->{_hot_restart_in_progress} = 0;
        delete $ENV{PAGI_MASTER_PID};
        return;
    }

    if ($pid == 0) {
        # Child: exec new master
        my @args = defined $ENV{PAGI_ARGV}
            ? split(/\0/, $ENV{PAGI_ARGV})
            : ();
        exec($^X, $0, @args)
            or do {
                warn "Hot restart exec failed: $!\n";
                POSIX::_exit(1);
            };
    }

    # Parent: log and continue running
    $self->_log(info => "Hot restart: new master spawned as PID $pid");
}

# Increase worker pool by 1
sub _increase_workers {
    my ($self) = @_;

    return if $self->{shutting_down};

    my $current = scalar keys %{$self->{worker_pids}};
    my $new_worker_num = $current + 1;

    $self->_log(info => "Received TTIN, spawning worker $new_worker_num (total: $new_worker_num)");
    $self->_spawn_worker($self->{_listen_entries}, $new_worker_num);
}

# Decrease worker pool by 1
sub _decrease_workers {
    my ($self) = @_;

    return if $self->{shutting_down};

    my @pids = keys %{$self->{worker_pids}};
    return unless @pids > 1;  # Keep at least 1 worker

    my $victim_pid = $pids[-1];  # Kill most recent
    my $remaining = scalar(@pids) - 1;

    $self->_log(info => "Received TTOU, killing worker (remaining: $remaining)");

    # Mark as "don't respawn" by setting a flag before killing
    $self->{_dont_respawn}{$victim_pid} = 1;
    kill 'TERM', $victim_pid;
}

sub _spawn_worker {
    my ($self, $listen_entries, $worker_num) = @_;

    my $loop = $self->loop;
    weaken(my $weak_self = $self);

    # Create heartbeat pipe if enabled
    my ($hb_rd, $hb_wr);
    if ($self->{heartbeat_timeout} && $self->{heartbeat_timeout} > 0) {
        pipe($hb_rd, $hb_wr) or die "Cannot create heartbeat pipe: $!";
    }

    # Set IGNORE before fork - child inherits it. IO::Async only resets
    # CODE refs, so 'IGNORE' (a string) survives. Child must NOT call
    # watch_signal(INT) or it will overwrite the IGNORE.
    my $old_sigint = $SIG{INT};
    $SIG{INT} = 'IGNORE' unless WIN32;

    my $pid = $loop->fork(
        code => sub {
            close($hb_rd) if $hb_rd;
            $self->_run_as_worker($listen_entries, $worker_num, $hb_wr);
            return 0;
        },
    );

    # Restore parent's SIGINT handler
    $SIG{INT} = $old_sigint unless WIN32;

    die "Fork failed" unless defined $pid;

    # Parent — close write end, set read end non-blocking
    if ($hb_wr) {
        close($hb_wr);
        $hb_rd->blocking(0);
    }

    # Parent - track the worker
    $self->{worker_pids}{$pid} = {
        worker_num     => $worker_num,
        started        => time(),
        heartbeat_rd   => $hb_rd,
        last_heartbeat => time(),
    };

    # Use watch_process to handle worker exit (replaces manual SIGCHLD handling)
    $loop->watch_process($pid => sub {
        my ($exit_pid, $exitcode) = @_;
        return unless $weak_self;

        # Close heartbeat pipe read end
        if (my $info = $weak_self->{worker_pids}{$exit_pid}) {
            close($info->{heartbeat_rd}) if $info->{heartbeat_rd};
        }

        # Remove from tracking
        delete $weak_self->{worker_pids}{$exit_pid};

        # Cancel per-worker restart kill timer if one exists
        if (my $timer_id = delete $weak_self->{_restart_kill_timers}{$exit_pid}) {
            $loop->unwatch_time($timer_id);
        }

        # Check exit code: exit(2) = startup failure, don't respawn
        my $exit_code = $exitcode >> 8;
        if ($exit_code == 2) {
            $weak_self->_log(warn => "Worker $worker_num startup failed, not respawning");
            # Don't respawn - startup failure would just repeat
        }
        # Respawn if still running and not shutting down
        elsif ($weak_self->{running} && !$weak_self->{shutting_down}) {
            # Don't respawn if this was a TTOU reduction
            unless (delete $weak_self->{_dont_respawn}{$exit_pid}) {
                $weak_self->_spawn_worker($listen_entries, $worker_num);
            }
        }

        # Check if all workers have exited (for shutdown)
        if ($weak_self->{shutting_down} && !keys %{$weak_self->{worker_pids}}) {
            # Cancel the shutdown SIGKILL escalation timer
            if ($weak_self->{_shutdown_kill_timer}) {
                $loop->unwatch_time($weak_self->{_shutdown_kill_timer});
                delete $weak_self->{_shutdown_kill_timer};
            }
            $loop->stop;
        }
    });

    return $pid;
}

sub _run_as_worker {
    my ($self, $listen_entries, $worker_num, $heartbeat_wr) = @_;

    # Note: $ONE_TRUE_LOOP already cleared by $loop->fork(), so this creates a fresh loop
    # Note: $SIG{INT} = 'IGNORE' inherited from parent - do NOT call watch_signal(INT)
    #       or it will overwrite the IGNORE with a CODE ref!
    my $loop = IO::Async::Loop->new;

    # In reuseport mode, each worker creates its own TCP listening socket
    my $reuseport = $self->{reuseport};
    for my $entry (@$listen_entries) {
        my $spec = $entry->{spec};
        if (!$entry->{socket} && $reuseport && $spec->{type} eq 'tcp') {
            $entry->{socket} = IO::Socket::INET->new(
                LocalAddr => $spec->{host},
                LocalPort => $spec->{bound_port},
                Proto     => 'tcp',
                Listen    => $self->{listener_backlog},
                ReuseAddr => 1,
                ReusePort => 1,
                Blocking  => 0,
            ) or die "Worker $worker_num: Cannot create listening socket: $!";
        }
    }

    # Build listener specs for the worker server constructor
    my @listen_specs;
    for my $entry (@$listen_entries) {
        my $s = $entry->{spec};
        if ($s->{type} eq 'unix') {
            push @listen_specs, { socket => $s->{path}, (defined $s->{socket_mode} ? (socket_mode => $s->{socket_mode}) : ()) };
        } else {
            push @listen_specs, { host => $s->{host}, port => $s->{port}, ($s->{ssl} ? (ssl => $s->{ssl}) : ()) };
        }
    }

    # Create a fresh server instance for this worker (single-worker mode)
    my $worker_server = PAGI::Server->new(
        app             => $self->{app},
        listen          => \@listen_specs,
        ssl             => $self->{ssl},
        http2           => $self->{http2},
        extensions      => $self->{extensions},
        on_error        => $self->{on_error},
        access_log      => $self->{access_log},
        log_level       => $self->{log_level},
        quiet           => 1,  # Workers should be quiet
        timeout         => $self->{timeout},
        max_header_size  => $self->{max_header_size},
        max_header_count => $self->{max_header_count},
        max_body_size    => $self->{max_body_size},
        max_requests     => $self->{max_requests},
        shutdown_timeout    => $self->{shutdown_timeout},
        request_timeout     => $self->{request_timeout},
        ws_idle_timeout     => $self->{ws_idle_timeout},
        sse_idle_timeout    => $self->{sse_idle_timeout},
        sync_file_threshold => $self->{sync_file_threshold},
        max_receive_queue   => $self->{max_receive_queue},
        max_ws_frame_size   => $self->{max_ws_frame_size},
        write_high_watermark => $self->{write_high_watermark},
        write_low_watermark  => $self->{write_low_watermark},
        workers          => 0,  # Single-worker mode in worker process
    );
    $worker_server->{is_worker} = 1;
    $worker_server->{worker_num} = $worker_num;  # Store for lifespan scope
    $worker_server->{_request_count} = 0;  # Track requests handled

    # Set bound_port from first TCP listener's socket
    for my $entry (@$listen_entries) {
        if ($entry->{spec}{type} eq 'tcp' && $entry->{socket} && $entry->{socket}->can('sockport')) {
            $worker_server->{bound_port} = $entry->{socket}->sockport;
            last;
        }
    }

    $loop->add($worker_server);

    # Build SSL config for this worker (each worker gets its own SSL context post-fork)
    my $ssl_params = $worker_server->_build_ssl_config;

    # Set up graceful shutdown on SIGTERM using IO::Async's signal watching
    # (raw $SIG handlers don't work reliably when the loop is running)
    # Note: Windows doesn't support Unix signals, so this is skipped there
    # Note: We do NOT set up watch_signal(INT) here - workers inherit $SIG{INT}='IGNORE'
    #       from parent, so they ignore SIGINT (including Ctrl-C). Parent sends SIGTERM.
    unless (WIN32) {
        my $shutdown_triggered = 0;
        $loop->watch_signal(TERM => sub {
            return if $shutdown_triggered;
            $shutdown_triggered = 1;
            $worker_server->adopt_future(
                $worker_server->shutdown->on_done(sub {
                    $loop->stop;
                })->on_fail(sub {
                    my ($error) = @_;
                    $worker_server->_log(error => "Worker shutdown error: $error");
                    $loop->stop;  # Still stop even on error
                })
            );
        });
    }

    # Run lifespan startup using a proper async wrapper
    my $startup_done = 0;
    my $startup_error;

    my $startup_future = (async sub {
        eval {
            my $startup_result = await $worker_server->_run_lifespan_startup;
            if (!$startup_result->{success}) {
                $startup_error = $startup_result->{message} // 'Lifespan startup failed';
            }
        };
        if ($@) {
            $startup_error = $@;
        }
        $startup_done = 1;
        $loop->stop if $startup_error;  # Stop loop on error
    })->();

    # Use adopt_future instead of retain
    $worker_server->adopt_future($startup_future);

    # Run the loop briefly to let async startup complete
    $loop->loop_once while !$startup_done;

    if ($startup_error) {
        $self->_log(error => "Worker $worker_num ($$): startup failed: $startup_error");
        for my $entry (@$listen_entries) {
            close($entry->{socket}) if $entry->{socket};
        }
        exit(2);  # Exit code 2 = startup failure (don't respawn)
    }

    # Create IO::Async::Listener for each inherited socket
    weaken(my $weak_server = $worker_server);

    for my $entry (@$listen_entries) {
        next unless $entry->{socket};
        my $spec = $entry->{spec};

        # Build SSL config for TCP listeners if needed
        my $use_ssl = ($ssl_params && $spec->{type} eq 'tcp');

        my $listener = IO::Async::Listener->new(
            handle => $entry->{socket},
            on_stream => sub {
                my ($listener, $stream) = @_;
                return unless $weak_server;

                if ($use_ssl) {
                    $loop->SSL_upgrade(
                        handle        => $stream,
                        SSL_server    => 1,
                        SSL_reuse_ctx => $worker_server->{_ssl_ctx},
                    )->on_done(sub {
                        $weak_server->_on_connection($stream, $spec) if $weak_server;
                    })->on_fail(sub {
                        my ($failure) = @_;
                        $weak_server->_log(debug => "SSL handshake failed: $failure")
                            if $weak_server;
                    });
                } else {
                    $weak_server->_on_connection($stream, $spec);
                }
            },
        );

        $worker_server->add_child($listener);

        # Configure accept error handler - try but ignore if it fails
        eval {
            $listener->configure(
                on_accept_error => sub {
                    my ($listener, $error) = @_;
                    return unless $weak_server;
                    $weak_server->_on_accept_error($error);
                },
            );
        };
        # Silently ignore configuration errors in workers
    }

    # Set up heartbeat writer: periodically signal liveness to parent
    if ($heartbeat_wr) {
        my $interval = ($self->{heartbeat_timeout} || 50) / 5;
        my $hb_timer = IO::Async::Timer::Periodic->new(
            interval => $interval,
            on_tick  => sub {
                syswrite($heartbeat_wr, "\x00", 1);
            },
        );
        $worker_server->add_child($hb_timer);
        $hb_timer->start;
    }

    $worker_server->{running} = 1;

    # Run the event loop
    $loop->run;

    # Clean up FDs before exit
    close($heartbeat_wr) if $heartbeat_wr;
    for my $entry (@$listen_entries) {
        close($entry->{socket}) if $entry->{socket};
    }
    exit(0);
}

sub _on_connection {
    my ($self, $stream, $listener_spec) = @_;

    weaken(my $weak_self = $self);

    # Check if we're at capacity
    my $max = $self->effective_max_connections;
    if ($self->connection_count >= $max) {
        # Over capacity - send 503 and close
        $self->_send_503_and_close($stream);
        return;
    }

    # Detect ALPN-negotiated protocol from TLS handle
    my $alpn_protocol;
    if ($self->{tls_enabled} && $self->{http2_enabled}) {
        my $handle = $stream->write_handle // $stream->read_handle;
        if ($handle && $handle->can('alpn_selected')) {
            $alpn_protocol = eval { $handle->alpn_selected };
        }
    }

    my $conn = PAGI::Server::Connection->new(
        stream            => $stream,
        app               => $self->{app},
        protocol          => $self->{protocol},
        server            => $self,
        extensions        => $self->{extensions},
        state             => $self->{state},
        tls_enabled       => $self->{tls_enabled} // 0,
        timeout           => $self->{timeout},
        request_timeout   => $self->{request_timeout},
        ws_idle_timeout   => $self->{ws_idle_timeout},
        sse_idle_timeout  => $self->{sse_idle_timeout},
        max_body_size     => $self->{max_body_size},
        access_log        => $self->{access_log},
        _access_log_formatter => $self->{_access_log_formatter},
        max_receive_queue => $self->{max_receive_queue},
        max_ws_frame_size => $self->{max_ws_frame_size},
        sync_file_threshold => $self->{sync_file_threshold},
        validate_events   => $self->{validate_events},
        write_high_watermark => $self->{write_high_watermark},
        write_low_watermark  => $self->{write_low_watermark},
        transport_type    => ($listener_spec && $listener_spec->{type}) // 'tcp',
        transport_path    => ($listener_spec ? $listener_spec->{path} : undef),
        ($self->{http2_enabled} ? (
            h2_protocol   => $self->{http2_protocol},
            alpn_protocol => $alpn_protocol,
            h2c_enabled   => $self->{h2c_enabled} // 0,
        ) : ()),
    );

    # Track the connection (O(1) hash insert)
    $self->{connections}{refaddr($conn)} = $conn;

    # Configure stream with callbacks BEFORE adding to loop
    $conn->start;

    # Add stream to the loop so it can read/write
    $self->add_child($stream);
}

sub _send_503_and_close {
    my ($self, $stream) = @_;

    my $body = "503 Service Unavailable - Server at capacity\r\n";
    my $response = join("\r\n",
        "HTTP/1.1 503 Service Unavailable",
        "Content-Type: text/plain",
        "Content-Length: " . length($body),
        "Connection: close",
        "Retry-After: 5",
        "",
        $body
    );

    # Configure stream with minimal on_read handler (required by IO::Async)
    $stream->configure(
        on_read => sub { 0 },  # Ignore any incoming data
    );

    # Add stream to loop so it can write
    $self->add_child($stream);

    # Write response and close
    $stream->write($response);
    $stream->close_when_empty;

    $self->_log(warn => "Connection rejected: at capacity (" . $self->connection_count . "/" . $self->effective_max_connections . ")");
}

sub _on_accept_error {
    my ($self, $error) = @_;

    # EMFILE = "Too many open files" - we're out of file descriptors
    # ENFILE = System-wide FD limit reached
    if ($error =~ /Too many open files|EMFILE|ENFILE/i) {
        # Only log the first EMFILE in a burst (when we're not already paused)
        unless ($self->{_accept_paused}) {
            $self->_log(warn => "Accept error (FD exhaustion): $error - pausing accept for 100ms");
        }

        # Pause accepting for a short time to let connections drain
        $self->_pause_accepting(0.1);
    }
    else {
        # Log other accept errors but don't crash
        $self->_log(error => "Accept error: $error");
    }
}

sub _pause_accepting {
    my ($self, $duration) = @_;

    return if $self->{_accept_paused};
    $self->{_accept_paused} = 1;

    # Cancel any existing timer before creating new one
    if ($self->{_accept_pause_timer}) {
        $self->loop->unwatch_time($self->{_accept_pause_timer});
        delete $self->{_accept_pause_timer};
    }

    # Temporarily disable all listeners
    for my $entry (@{$self->{_listen_entries} // []}) {
        my $listener = $entry->{listener};
        if ($listener && $listener->read_handle) {
            $listener->want_readready(0);
        }
    }
    # Backward compat: also pause $self->{listener} if not in entries
    if ($self->{listener} && !$self->{_listen_entries}) {
        $self->{listener}->want_readready(0) if $self->{listener}->read_handle;
    }

    # Re-enable after duration
    weaken(my $weak_self = $self);
    my $timer_id = $self->loop->watch_time(after => $duration, code => sub {
        return unless $weak_self && $weak_self->{running};
        $weak_self->{_accept_paused} = 0;
        delete $weak_self->{_accept_pause_timer};

        # Resume all listeners
        for my $entry (@{$weak_self->{_listen_entries} // []}) {
            my $listener = $entry->{listener};
            if ($listener && $listener->read_handle) {
                $listener->want_readready(1);
            }
        }
        # Backward compat
        if ($weak_self->{listener} && !$weak_self->{_listen_entries}) {
            $weak_self->{listener}->want_readready(1) if $weak_self->{listener}->read_handle;
        }

        $weak_self->_log(debug => "Accept resumed after FD exhaustion pause");
    });

    $self->{_accept_pause_timer} = $timer_id;
}

# Called when a request completes (for max_requests tracking)
sub _on_request_complete {
    my ($self) = @_;

    return unless $self->{is_worker};
    return unless $self->{max_requests} && $self->{max_requests} > 0;

    $self->{_request_count}++;

    if ($self->{_request_count} >= $self->{max_requests}) {
        return if $self->{_max_requests_shutdown_triggered};  # Prevent duplicate shutdowns
        $self->{_max_requests_shutdown_triggered} = 1;
        $self->_log(info => "Worker $$: reached max_requests ($self->{max_requests}), shutting down");
        # Initiate graceful shutdown (finish current connections, then exit)
        $self->adopt_future(
            $self->shutdown->on_done(sub {
                $self->loop->stop;
            })->on_fail(sub {
                my ($error) = @_;
                $self->_log(error => "Worker $$: max_requests shutdown error: $error");
                $self->loop->stop;  # Still stop even on error
            })
        );
    }
}

# Lifespan Protocol Implementation

async sub _run_lifespan_startup {
    my ($self) = @_;

    # Create lifespan scope
    my $scope = {
        type => 'lifespan',
        pagi => {
            version      => '0.2',
            spec_version => '0.2',
            is_worker    => $self->{is_worker} // 0,
            worker_num   => $self->{worker_num},  # undef for single-worker, 1-N for multi-worker
        },
        state => $self->{state},  # App can populate this
    };

    # Create receive/send for lifespan protocol
    my @send_queue;
    my $receive_pending;
    my $startup_complete = Future->new;
    my $lifespan_supported = 1;  # Track if app supports lifespan

    # $receive for the app - returns events from the server
    my $receive = sub {
        if (@send_queue) {
            return Future->done(shift @send_queue);
        }
        $receive_pending = Future->new;
        return $receive_pending;
    };

    # $send for the app - handles app responses
    my $send = async sub  {
        my ($event) = @_;
        my $type = $event->{type} // '';

        if ($type eq 'lifespan.startup.complete') {
            $startup_complete->done({ success => 1 });
        }
        elsif ($type eq 'lifespan.startup.failed') {
            my $message = $event->{message} // '';
            $startup_complete->done({ success => 0, message => $message });
        }
        elsif ($type eq 'lifespan.shutdown.complete') {
            # Store for shutdown handling
            $self->{shutdown_complete} = 1;
            if ($self->{shutdown_pending}) {
                $self->{shutdown_pending}->done({ success => 1 });
            }
        }
        elsif ($type eq 'lifespan.shutdown.failed') {
            my $message = $event->{message} // '';
            $self->{shutdown_complete} = 1;
            if ($self->{shutdown_pending}) {
                $self->{shutdown_pending}->done({ success => 0, message => $message });
            }
        }

        return;
    };

    # Queue the startup event
    push @send_queue, { type => 'lifespan.startup' };
    if ($receive_pending && !$receive_pending->is_ready) {
        my $f = $receive_pending;
        $receive_pending = undef;
        $f->done(shift @send_queue);
    }

    # Store lifespan handlers for shutdown
    $self->{lifespan_receive} = $receive;
    $self->{lifespan_send} = $send;
    $self->{lifespan_send_queue} = \@send_queue;
    $self->{lifespan_receive_pending} = \$receive_pending;

    # Start the lifespan app handler
    # We run it in the background and wait for startup.complete
    my $app_future = (async sub {
        eval {
            await $self->{app}->($scope, $receive, $send);
        };
        # Per spec: if the app throws an exception for lifespan scope,
        # the server should continue without lifespan support.
        # This matches Uvicorn/Hypercorn "auto" mode behavior.
        # Apps that don't support lifespan should: die if $scope->{type} ne 'websocket';
        $lifespan_supported = 0;
        if (!$startup_complete->is_ready) {
            $self->_log(info => "Lifespan not supported, continuing without it");
            $startup_complete->done({ success => 1, lifespan_supported => 0 });
        }
    })->();

    # Keep the app future so we can trigger shutdown later
    $self->{lifespan_app_future} = $app_future;
    # Use adopt_future instead of retain for proper error handling
    $self->adopt_future($app_future);

    # Wait for startup complete (with timeout)
    my $result = await $startup_complete;

    # Track if lifespan is supported
    $self->{lifespan_supported} = $result->{lifespan_supported} // 1;

    return $result;
}

async sub _run_lifespan_shutdown {
    my ($self) = @_;

    # If lifespan is not supported or no lifespan was started, just return success
    return { success => 1 } unless $self->{lifespan_supported};
    return { success => 1 } unless $self->{lifespan_send_queue};

    $self->{shutdown_pending} = $self->loop->new_future;

    # Queue the shutdown event
    my $send_queue = $self->{lifespan_send_queue};
    my $receive_pending_ref = $self->{lifespan_receive_pending};

    push @$send_queue, { type => 'lifespan.shutdown' };

    # Trigger pending receive if waiting
    if ($$receive_pending_ref && !$$receive_pending_ref->is_ready) {
        my $f = $$receive_pending_ref;
        $$receive_pending_ref = undef;
        $f->done(shift @$send_queue);
    }

    # Wait for shutdown complete (with timeout to prevent hanging)
    my $timeout = $self->{shutdown_timeout} // 30;
    my $timeout_f = $self->loop->delay_future(after => $timeout);

    my $result = await Future->wait_any($self->{shutdown_pending}, $timeout_f);

    # If timeout won, return failure
    if ($timeout_f->is_ready && !$self->{shutdown_pending}->is_ready) {
        return { success => 0, message => "Lifespan shutdown timed out after ${timeout}s" };
    }

    return $result // { success => 1 };
}

async sub shutdown {
    my ($self) = @_;

    return unless $self->{running};
    $self->{running} = 0;
    $self->{shutting_down} = 1;

    # Cancel accept pause timer if active
    if ($self->{_accept_pause_timer}) {
        $self->loop->unwatch_time($self->{_accept_pause_timer});
        delete $self->{_accept_pause_timer};
        $self->{_accept_paused} = 0;
    }

    # Stop accepting new connections on all listeners
    for my $entry (@{$self->{_listen_entries} // []}) {
        eval { $self->remove_child($entry->{listener}) };
    }
    $self->{listener} = undef;

    # Clean up PAGI_REUSE entries for sockets we created (not inherited)
    for my $entry (@{$self->{_listen_entries} // []}) {
        my $key = $entry->{spec}{_reuse_key};
        if ($key && !$self->{_hot_restart_in_progress} && defined $ENV{PAGI_REUSE}) {
            $ENV{PAGI_REUSE} =~ s/(?:^|,)\Q$key\E//;
            $ENV{PAGI_REUSE} =~ s/^,// if defined $ENV{PAGI_REUSE};
        }
    }

    # Clean up Unix socket files (only those we created, not inherited)
    for my $entry (@{$self->{_listen_entries} // []}) {
        if ($entry->{spec}{type} eq 'unix'
            && !$entry->{spec}{_inherited}
            && -e $entry->{spec}{path}) {
            unlink $entry->{spec}{path};
        }
    }
    $self->{_listen_entries} = [];

    # Wait for active connections to drain (graceful shutdown)
    await $self->_drain_connections;

    # Run lifespan shutdown
    my $shutdown_result = await $self->_run_lifespan_shutdown;

    if (!$shutdown_result->{success}) {
        my $message = $shutdown_result->{message} // 'Lifespan shutdown failed';
        $self->_log(warn => "PAGI Server shutdown warning: $message");
    }

    return $self;
}

# Wait for active connections to complete, with timeout
# Uses event-driven approach: Connection._close() signals when last one closes
async sub _drain_connections {
    my ($self) = @_;

    my $timeout = $self->{shutdown_timeout} // 30;
    my $loop = $self->loop;

    # First, close all idle connections immediately (not processing a request)
    # Keep-alive connections waiting for next request should be closed
    my @idle = grep { !$_->{handling_request} } values %{$self->{connections}};
    for my $conn (@idle) {
        $conn->_handle_disconnect_and_close('server_shutdown');
    }

    # Also close long-lived connections (SSE, WebSocket) immediately
    # These never become "idle" so would wait for full timeout otherwise
    my @longlived = grep { $_->{sse_mode} || $_->{websocket_mode} } values %{$self->{connections}};
    for my $conn (@longlived) {
        $conn->_handle_disconnect_and_close('server_shutdown');
    }

    # If all connections are now closed, we're done
    return if keys %{$self->{connections}} == 0;

    # Create a Future that Connection._close() will resolve when last one closes
    $self->{drain_complete} = $loop->new_future;

    # Wait for either: all connections close OR timeout
    my $timeout_f = $loop->delay_future(after => $timeout);

    await Future->wait_any($self->{drain_complete}, $timeout_f);

    # Brief pause to let any final socket writes flush
    # (stream->write is async; data may still be in kernel buffer)
    await $loop->delay_future(after => 0.05) if keys %{$self->{connections}} == 0;

    # If timeout won (connections still remain), force close them
    if (keys %{$self->{connections}} > 0) {
        my $remaining = scalar keys %{$self->{connections}};
        $self->_log(warn => "Shutdown timeout: force-closing $remaining active connections");

        for my $conn (values %{$self->{connections}}) {
            $conn->_close if $conn && $conn->can('_close');
        }
    }

    delete $self->{drain_complete};
    return;
}

# --- Access log format compiler ---

my %ACCESS_LOG_PRESETS = (
    clf      => '%h - - [%t] "%m %U%q" %s %ds',
    combined => '%h - - [%t] "%r" %s %b "%{Referer}i" "%{User-Agent}i"',
    common   => '%h - - [%t] "%r" %s %b',
    tiny     => '%m %U%q %s %Dms',
);

sub _compile_access_log_format {
    my ($class_or_self, $format) = @_;

    # Resolve preset names
    if (exists $ACCESS_LOG_PRESETS{$format}) {
        $format = $ACCESS_LOG_PRESETS{$format};
    }

    # Parse format string into a list of fragments (closures or literal strings)
    my @fragments;
    my $pos = 0;
    my $len = length($format);

    while ($pos < $len) {
        my $ch = substr($format, $pos, 1);

        if ($ch eq '%') {
            $pos++;
            last if $pos >= $len;

            my $next = substr($format, $pos, 1);

            if ($next eq '%') {
                # Literal percent
                push @fragments, '%';
                $pos++;
            }
            elsif ($next eq '{') {
                # Header extraction: %{Name}i
                my $end = index($format, '}', $pos);
                die "Unterminated %{...} in access log format\n" if $end < 0;
                my $header_name = substr($format, $pos + 1, $end - $pos - 1);
                $pos = $end + 1;

                # Must be followed by 'i' (request header)
                die "Expected 'i' after %{$header_name} in access log format\n"
                    if $pos >= $len || substr($format, $pos, 1) ne 'i';
                $pos++;

                my $lc_name = lc($header_name);
                push @fragments, sub {
                    my ($info) = @_;
                    for my $h (@{$info->{request_headers}}) {
                        return $h->[1] if lc($h->[0]) eq $lc_name;
                    }
                    return '-';
                };
            }
            else {
                # Simple atom
                my $atom = $next;
                $pos++;

                my $frag = _access_log_atom($atom);
                push @fragments, $frag;
            }
        }
        else {
            # Literal text: collect until next %
            my $next_pct = index($format, '%', $pos);
            if ($next_pct < 0) {
                push @fragments, substr($format, $pos);
                $pos = $len;
            }
            else {
                push @fragments, substr($format, $pos, $next_pct - $pos);
                $pos = $next_pct;
            }
        }
    }

    # Build a single closure from fragments
    return sub {
        my ($info) = @_;
        return join('', map { ref $_ ? $_->($info) : $_ } @fragments);
    };
}

sub _access_log_atom {
    my ($atom) = @_;

    my %atoms = (
        h => sub { $_[0]->{client_ip} // '-' },
        l => sub { '-' },
        u => sub { '-' },
        t => sub { $_[0]->{timestamp} // '-' },
        r => sub {
            my $i = $_[0];
            my $uri = $i->{path} // '/';
            my $qs = $i->{query};
            $uri .= "?$qs" if defined $qs && length $qs;
            sprintf('%s %s HTTP/%s', $i->{method} // '-', $uri, $i->{http_version} // '1.1');
        },
        m => sub { $_[0]->{method} // '-' },
        U => sub { $_[0]->{path} // '/' },
        q => sub {
            my $qs = $_[0]->{query};
            (defined $qs && length $qs) ? "?$qs" : '';
        },
        H => sub { 'HTTP/' . ($_[0]->{http_version} // '1.1') },
        s => sub { $_[0]->{status} // '-' },
        b => sub {
            my $size = $_[0]->{size} // 0;
            $size ? $size : '-';
        },
        B => sub { $_[0]->{size} // 0 },
        d => sub { sprintf('%.3f', $_[0]->{duration} // 0) },
        D => sub { int(($_[0]->{duration} // 0) * 1_000_000) },
        T => sub { int($_[0]->{duration} // 0) },
    );

    if (my $frag = $atoms{$atom}) {
        return $frag;
    }

    die "Unknown access log format atom '%$atom'\n";
}

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

    return $self->{bound_port} // $self->{port};
}

sub socket_path {
    my ($self) = @_;
    for my $listener (@{$self->{listeners} // []}) {
        return $listener->{path} if $listener->{type} eq 'unix';
    }
    return undef;
}

sub listeners {
    my ($self) = @_;
    return $self->{listeners} // [];
}

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

    return $self->{running} ? 1 : 0;
}

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

    return scalar keys %{$self->{connections}};
}

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

    # If explicitly set, use that; otherwise default to 1000
    # (Same default as Mojolicious - simple, predictable, no platform-specific hacks)
    return $self->{max_connections} && $self->{max_connections} > 0
        ? $self->{max_connections}
        : 1000;
}

1;

__END__