File Coverage

blib/lib/Async/Redis.pm

Criterion	Covered	Total	%
statement	129	768	16.8
branch	22	280	7.8
condition	27	141	19.1
subroutine	32	109	29.3
pod	18	49	36.7
total	228	1347	16.9

package Async::Redis;

use strict;
use warnings;
use 5.018;

our $VERSION = '0.001006';

use Future;
use Future::AsyncAwait;
use Future::IO 0.19;
use Socket qw(pack_sockaddr_in pack_sockaddr_un inet_aton AF_INET AF_UNIX SOCK_STREAM);
use IO::Handle ();
use IO::Socket::INET;
use Time::HiRes ();

# Error classes
use Async::Redis::Error::Connection;
use Async::Redis::Error::Timeout;
use Async::Redis::Error::Disconnected;
use Async::Redis::Error::Redis;
use Async::Redis::Error::Protocol;

# Import auto-generated command methods
use Async::Redis::Commands;
our @ISA = qw(Async::Redis::Commands);

# Key extraction for prefixing
use Async::Redis::KeyExtractor;

# Transaction support
use Async::Redis::Transaction;

# Script support
use Async::Redis::Script;

# Iterator support
use Async::Redis::Iterator;

# Pipeline support
use Async::Redis::Pipeline;
use Async::Redis::AutoPipeline;

# PubSub support
use Async::Redis::Subscription;

# Telemetry support
use Async::Redis::Telemetry;

# Try XS version first, fall back to pure Perl
BEGIN {
    eval { require Protocol::Redis::XS; 1 }
        or require Protocol::Redis;
}

sub _parser_class {
    return $INC{'Protocol/Redis/XS.pm'} ? 'Protocol::Redis::XS' : 'Protocol::Redis';
}

sub _calculate_backoff {
    my ($self, $attempt) = @_;

    # Exponential: delay * 2^(attempt-1)
    my $delay = $self->{reconnect_delay} * (2 ** ($attempt - 1));

    # Cap at max
    $delay = $self->{reconnect_delay_max} if $delay > $self->{reconnect_delay_max};

    # Apply jitter: delay * (1 +/- jitter)
    if ($self->{reconnect_jitter} > 0) {
        my $jitter_range = $delay * $self->{reconnect_jitter};
        my $jitter = (rand(2) - 1) * $jitter_range;
        $delay += $jitter;
    }

    return $delay;
}

sub new {
    my ($class, %args) = @_;

    # Parse URI if provided
    if ($args{uri}) {
        require Async::Redis::URI;
        my $uri = Async::Redis::URI->parse($args{uri});
        if ($uri) {
            my %uri_args = $uri->to_hash;
            # URI values are defaults, explicit args override
            %args = (%uri_args, %args);
            delete $args{uri};  # don't store the string
        }
    }

    my $self = bless {
        path     => $args{path},
        host     => $args{path} ? undef : ($args{host} // 'localhost'),
        port     => $args{path} ? undef : ($args{port} // 6379),
        socket   => undef,
        parser   => undef,
        connected => 0,

        # Timeout settings
        connect_timeout         => $args{connect_timeout} // 10,
        request_timeout         => $args{request_timeout} // 5,
        blocking_timeout_buffer => $args{blocking_timeout_buffer} // 2,

        # Inflight tracking with deadlines
        # Entry: { future => $f, cmd => $cmd, args => \@args, deadline => $t, sent_at => $t }
        inflight => [],

        # Response reader synchronization
        _reading_responses => 0,

        # Reconnection settings
        reconnect           => $args{reconnect} // 0,
        reconnect_delay     => $args{reconnect_delay} // 0.1,
        reconnect_delay_max => $args{reconnect_delay_max} // 60,
        reconnect_jitter    => $args{reconnect_jitter} // 0.25,
        _reconnect_attempt  => 0,

        # Callbacks
        on_connect    => $args{on_connect},
        on_disconnect => $args{on_disconnect},
        on_error      => $args{on_error},

        # Authentication
        password    => $args{password},
        username    => $args{username},
        database    => $args{database} // 0,
        client_name => $args{client_name},

        # TLS
        tls => $args{tls},

        # Key prefixing
        prefix => $args{prefix},

        # Pipeline settings
        pipeline_depth => $args{pipeline_depth} // 10000,
        auto_pipeline  => $args{auto_pipeline} // 0,

        # Transaction state
        in_multi => 0,
        watching => 0,

        # PubSub state
        in_pubsub     => 0,
        _subscription => undef,
        _pump_running => 0,

        # Fork safety
        _pid => $$,

        # Script registry
        _scripts => {},

        # Current read future for clean disconnect cancellation
        _current_read_future => undef,

        # Telemetry options
        debug              => $args{debug},
        otel_tracer        => $args{otel_tracer},
        otel_meter         => $args{otel_meter},
        otel_include_args  => $args{otel_include_args} // 1,
        otel_redact        => $args{otel_redact} // 1,
    }, $class;

    # Initialize telemetry if any observability enabled
    if ($self->{debug} || $self->{otel_tracer} || $self->{otel_meter}) {
        $self->{_telemetry} = Async::Redis::Telemetry->new(
            tracer       => $self->{otel_tracer},
            meter        => $self->{otel_meter},
            debug        => $self->{debug},
            include_args => $self->{otel_include_args},
            redact       => $self->{otel_redact},
            host         => $self->{host},
            port         => $self->{port},
            database     => $self->{database} // 0,
        );
    }

    return $self;
}

# Connect to Redis server
async sub connect {
    my ($self) = @_;

    return $self if $self->{connected};

    # Create socket — AF_UNIX for path, AF_INET for host:port
    my ($socket, $sockaddr);

    if ($self->{path}) {
        socket($socket, AF_UNIX, SOCK_STREAM, 0)
            or die Async::Redis::Error::Connection->new(
                message => "Cannot create unix socket: $!",
                host    => $self->{path},
                port    => 0,
            );
        IO::Handle::blocking($socket, 0);
        $sockaddr = pack_sockaddr_un($self->{path});
    } else {
        $socket = IO::Socket::INET->new(
            Proto    => 'tcp',
            Blocking => 0,
        ) or die Async::Redis::Error::Connection->new(
            message => "Cannot create socket: $!",
            host    => $self->{host},
            port    => $self->{port},
        );

        # Build sockaddr
        my $addr = inet_aton($self->{host})
            or die Async::Redis::Error::Connection->new(
                message => "Cannot resolve host: $self->{host}",
                host    => $self->{host},
                port    => $self->{port},
            );
        $sockaddr = pack_sockaddr_in($self->{port}, $addr);
    }

    # Connect with timeout using Future->wait_any
    my $connect_f = Future::IO->connect($socket, $sockaddr);
    my $sleep_f = Future::IO->sleep($self->{connect_timeout});

    my $timeout_f = $sleep_f->then(sub {
        return Future->fail('connect_timeout');
    });

    my $wait_f = Future->wait_any($connect_f, $timeout_f);

    # Use followed_by to handle both success and failure without await propagating failure
    my $result_f = $wait_f->followed_by(sub {
        my ($f) = @_;
        return Future->done($f);  # wrap the future itself
    });

    my $completed_f = await $result_f;

    # Now check the result
    if ($completed_f->is_failed) {
        my ($error) = $completed_f->failure;
        # Don't call close() - let $socket go out of scope when we die.
        # Perl's DESTROY will close it after the exception unwinds.

        if ($error eq 'connect_timeout') {
            die Async::Redis::Error::Timeout->new(
                message => "Connect timed out after $self->{connect_timeout}s",
                timeout => $self->{connect_timeout},
            );
        }
        die Async::Redis::Error::Connection->new(
            message => "$error",
            host    => $self->{path} // $self->{host},
            port    => $self->{port} // 0,
        );
    }

    # TLS upgrade if enabled
    if ($self->{tls}) {
        eval {
            $socket = await $self->_tls_upgrade($socket);
        };
        if ($@) {
            # Don't call close() - let $socket go out of scope when we die.
            # Perl's DESTROY will close it after the exception unwinds.
            die $@;
        }
    }

    $self->{socket} = $socket;
    $self->{parser} = _parser_class()->new(api => 1);
    $self->{connected} = 1;
    $self->{inflight} = [];
    $self->{_reading_responses} = 0;
    $self->{_pid} = $$;  # Track PID for fork safety
    $self->{_current_read_future} = undef;

    # Run Redis protocol handshake (AUTH, SELECT, CLIENT SETNAME)
    await $self->_redis_handshake;

    # Initialize auto-pipeline if enabled
    if ($self->{auto_pipeline}) {
        $self->{_auto_pipeline} = Async::Redis::AutoPipeline->new(
            redis     => $self,
            max_depth => $self->{pipeline_depth},
        );
    }

    # Fire on_connect callback and reset reconnect counter
    if ($self->{on_connect}) {
        $self->{on_connect}->($self);
    }
    $self->{_reconnect_attempt} = 0;

    # Telemetry: record connection
    if ($self->{_telemetry}) {
        $self->{_telemetry}->record_connection(1);
        $self->{_telemetry}->log_event('connected',
            $self->{path} // "$self->{host}:$self->{port}");
    }

    return $self;
}

# Redis protocol handshake after TCP connect
async sub _redis_handshake {
    my ($self) = @_;

    # Use connect_timeout for the entire handshake (AUTH, SELECT, CLIENT SETNAME)
    # This ensures the handshake can't block forever if Redis hangs
    my $deadline = Time::HiRes::time() + $self->{connect_timeout};

    # AUTH (password or username+password for ACL)
    if ($self->{password}) {
        my @auth_args = ('AUTH');
        push @auth_args, $self->{username} if $self->{username};
        push @auth_args, $self->{password};

        my $cmd = $self->_build_command(@auth_args);
        await $self->_send($cmd);

        my $response = await $self->_read_response_with_deadline($deadline, ['AUTH']);
        my $result = $self->_decode_response($response);

        # AUTH returns OK on success, throws on failure
        unless ($result && $result eq 'OK') {
            die Async::Redis::Error::Redis->new(
                message => "Authentication failed: $result",
                type    => 'NOAUTH',
            );
        }
    }

    # SELECT database
    if ($self->{database} && $self->{database} != 0) {
        my $cmd = $self->_build_command('SELECT', $self->{database});
        await $self->_send($cmd);

        my $response = await $self->_read_response_with_deadline($deadline, ['SELECT', $self->{database}]);
        my $result = $self->_decode_response($response);

        unless ($result && $result eq 'OK') {
            die Async::Redis::Error::Redis->new(
                message => "SELECT failed: $result",
                type    => 'ERR',
            );
        }
    }

    # CLIENT SETNAME
    if ($self->{client_name}) {
        my $cmd = $self->_build_command('CLIENT', 'SETNAME', $self->{client_name});
        await $self->_send($cmd);

        my $response = await $self->_read_response_with_deadline($deadline, ['CLIENT', 'SETNAME']);
        # Ignore result - SETNAME failing shouldn't prevent connection
    }
}

# Check if connected to Redis
sub is_connected {
    my ($self) = @_;
    return $self->{connected} ? 1 : 0;
}

# Disconnect from Redis
sub disconnect {
    my ($self, $reason) = @_;
    $reason //= 'client_disconnect';

    my $was_connected = $self->{connected};

    # Cancel any active read future BEFORE closing socket
    # This ensures Future::IO unregisters its watcher while fileno is still valid
    if ($self->{_current_read_future} && !$self->{_current_read_future}->is_ready) {
        $self->{_current_read_future}->cancel;
        $self->{_current_read_future} = undef;
    }

    # Cancel any pending inflight operations before closing socket
    if (my $inflight = $self->{inflight}) {
        for my $entry (@$inflight) {
            if ($entry->{future} && !$entry->{future}->is_ready) {
                $entry->{future}->cancel;
            }
        }
        $self->{inflight} = [];
    }

    if ($self->{socket}) {
        $self->_close_socket;
    }
    $self->{connected} = 0;
    $self->{parser} = undef;
    $self->{_reading_responses} = 0;

    if ($was_connected && $self->{on_disconnect}) {
        $self->{on_disconnect}->($self, $reason);
    }

    # Telemetry: record disconnection
    if ($was_connected && $self->{_telemetry}) {
        $self->{_telemetry}->record_connection(-1);
        $self->{_telemetry}->log_event('disconnected', $reason);
    }

    return $self;
}

# Destructor - clean up socket when object is garbage collected
sub DESTROY {
    my ($self) = @_;
    # Only clean up if we have a socket and it's still open
    if ($self->{socket} && fileno($self->{socket})) {
        $self->_close_socket;
    }
}

# Properly close socket, canceling any pending futures first
sub _close_socket {
    my ($self) = @_;

    # Take ownership - removes from $self immediately
    my $socket = delete $self->{socket} or return;
    my $fileno = fileno($socket);

    # Cancel any pending inflight futures - this propagates to
    # Future::IO internals and cleans up any watchers on this socket.
    # Important: must happen while fileno is still valid!
    if (my $inflight = delete $self->{inflight}) {
        for my $entry (@$inflight) {
            if ($entry->{future} && !$entry->{future}->is_ready) {
                $entry->{future}->cancel;
            }
        }
    }
    $self->{inflight} = [];

    # Initiate clean TCP shutdown (FIN) while fileno still valid
    shutdown($socket, 2) if defined $fileno;

    # DON'T call close()!
    # $socket falls out of scope here, Perl's DESTROY calls close().
    # By this point, Future::IO has already unregistered its watchers
    # via the cancel() calls above.
}

# Check if fork occurred and invalidate connection
sub _check_fork {
    my ($self) = @_;

    if ($self->{_pid} && $self->{_pid} != $$) {
        # Fork detected - invalidate connection (parent owns the socket)
        # Don't cancel futures - they belong to the parent's event loop
        # Just clear references so we don't try to use them
        $self->{connected} = 0;
        $self->{socket} = undef;
        $self->{parser} = undef;
        $self->{inflight} = [];
        $self->{_reading_responses} = 0;
        $self->{_current_read_future} = undef;  # Clear stale reference

        my $old_pid = $self->{_pid};
        $self->{_pid} = $$;

        if ($self->{_telemetry}) {
            $self->{_telemetry}->log_event('fork_detected', "old PID: $old_pid, new PID: $$");
        }

        return 1;  # Fork occurred
    }

    return 0;
}

# Build Redis command in RESP format
sub _build_command {
    my ($self, @args) = @_;

    my $cmd = "*" . scalar(@args) . "\r\n";
    for my $arg (@args) {
        $arg //= '';
        my $bytes = "$arg";  # stringify
        utf8::encode($bytes) if utf8::is_utf8($bytes);
        $cmd .= "\$" . length($bytes) . "\r\n" . $bytes . "\r\n";
    }
    return $cmd;
}

# Send raw data
async sub _send {
    my ($self, $data) = @_;
    await Future::IO->write_exactly($self->{socket}, $data);
    return length($data);
}

# Add command to inflight queue - returns queue depth
sub _add_inflight {
    my ($self, $future, $cmd, $args, $deadline) = @_;
    push @{$self->{inflight}}, {
        future   => $future,
        cmd      => $cmd,
        args     => $args,
        deadline => $deadline,
        sent_at  => Time::HiRes::time(),
    };
    return scalar @{$self->{inflight}};
}

# Shift first entry from inflight queue
sub _shift_inflight {
    my ($self) = @_;
    return shift @{$self->{inflight}};
}

# Fail all pending inflight futures with given error
sub _fail_all_inflight {
    my ($self, $error) = @_;
    while (my $entry = $self->_shift_inflight) {
        if ($entry->{future} && !$entry->{future}->is_ready) {
            $entry->{future}->fail($error);
        }
    }
}

# Ensure response reader is running - the core response queue mechanism
# Only one reader should be active at a time, processing responses in FIFO order
async sub _ensure_response_reader {
    my ($self) = @_;

    # Already reading - don't start another reader
    return if $self->{_reading_responses};

    $self->{_reading_responses} = 1;

    while (@{$self->{inflight}} && $self->{connected}) {
        my $entry = $self->{inflight}[0];

        # Read response with deadline from the entry
        my $response;
        my $read_ok = eval {
            $response = await $self->_read_response_with_deadline(
                $entry->{deadline},
                $entry->{args}
            );
            1;
        };

        if (!$read_ok) {
            my $read_error = $@;
            # Connection/timeout error - fail all inflight and abort
            $self->_fail_all_inflight($read_error);
            $self->{_reading_responses} = 0;
            return;
        }

        # Remove this entry from the queue now that we have its response
        $self->_shift_inflight;

        # Decode response (sync operation, eval works fine here)
        my $result;
        my $decode_ok = eval {
            $result = $self->_decode_response($response);
            1;
        };

        # Complete the future
        if (!$decode_ok) {
            my $decode_error = $@;
            # Redis error (like WRONGTYPE) - fail just this future
            $entry->{future}->fail($decode_error) unless $entry->{future}->is_ready;
        } else {
            # Success - complete the future with result
            $entry->{future}->done($result) unless $entry->{future}->is_ready;
        }
    }

    $self->{_reading_responses} = 0;
}

# Read and parse one response
async sub _read_response {
    my ($self) = @_;

    # First check if parser already has a complete message
    # (from previous read that contained multiple responses)
    if (my $msg = $self->{parser}->get_message) {
        return $msg;
    }

    # Read until we get a complete message
    while (1) {
        my $buf = await Future::IO->read($self->{socket}, 65536);

        # EOF
        if (!defined $buf || length($buf) == 0) {
            die "Connection closed by server";
        }

        $self->{parser}->parse($buf);

        if (my $msg = $self->{parser}->get_message) {
            return $msg;
        }
    }
}

# Calculate deadline based on command type
sub _calculate_deadline {
    my ($self, $cmd, @args) = @_;

    $cmd = uc($cmd // '');

    # Blocking commands get extended deadline
    if ($cmd =~ /^(BLPOP|BRPOP|BLMOVE|BRPOPLPUSH|BLMPOP|BZPOPMIN|BZPOPMAX|BZMPOP)$/) {
        # Last arg is the timeout for these commands
        my $server_timeout = $args[-1] // 0;
        return Time::HiRes::time() + $server_timeout + $self->{blocking_timeout_buffer};
    }

    if ($cmd =~ /^(XREAD|XREADGROUP)$/) {
        # XREAD/XREADGROUP have BLOCK option
        for my $i (0 .. $#args - 1) {
            if (uc($args[$i]) eq 'BLOCK') {
                my $block_ms = $args[$i + 1] // 0;
                return Time::HiRes::time() + ($block_ms / 1000) + $self->{blocking_timeout_buffer};
            }
        }
    }

    # Normal commands use request_timeout
    return Time::HiRes::time() + $self->{request_timeout};
}

# Non-blocking TLS upgrade
async sub _tls_upgrade {
    my ($self, $socket) = @_;

    require IO::Socket::SSL;

    # Build SSL options
    my %ssl_opts = (
        SSL_startHandshake => 0,  # Don't block during start_SSL!
    );

    if (ref $self->{tls} eq 'HASH') {
        $ssl_opts{SSL_ca_file}    = $self->{tls}{ca_file} if $self->{tls}{ca_file};
        $ssl_opts{SSL_cert_file}  = $self->{tls}{cert_file} if $self->{tls}{cert_file};
        $ssl_opts{SSL_key_file}   = $self->{tls}{key_file} if $self->{tls}{key_file};

        if (exists $self->{tls}{verify}) {
            $ssl_opts{SSL_verify_mode} = $self->{tls}{verify}
                ? IO::Socket::SSL::SSL_VERIFY_PEER()
                : IO::Socket::SSL::SSL_VERIFY_NONE();
        } else {
            $ssl_opts{SSL_verify_mode} = IO::Socket::SSL::SSL_VERIFY_PEER();
        }
    } else {
        $ssl_opts{SSL_verify_mode} = IO::Socket::SSL::SSL_VERIFY_PEER();
    }

    # Start SSL (does not block because SSL_startHandshake => 0)
    IO::Socket::SSL->start_SSL($socket, %ssl_opts)
        or die Async::Redis::Error::Connection->new(
            message => "SSL setup failed: " . IO::Socket::SSL::errstr(),
            host    => $self->{host},
            port    => $self->{port},
        );

    # Drive handshake with non-blocking loop
    my $deadline = Time::HiRes::time() + $self->{connect_timeout};

    while (1) {
        # Check timeout
        if (Time::HiRes::time() >= $deadline) {
            die Async::Redis::Error::Timeout->new(
                message => "TLS handshake timed out",
                timeout => $self->{connect_timeout},
            );
        }

        # Attempt handshake step
        my $rv = $socket->connect_SSL;

        if ($rv) {
            # Handshake complete!
            return $socket;
        }

        # Check what the handshake needs
        my $remaining = $deadline - Time::HiRes::time();
        $remaining = 0.1 if $remaining <= 0;

        if ($IO::Socket::SSL::SSL_ERROR == IO::Socket::SSL::SSL_ERROR_WANT_READ()) {
            # Wait for socket to become readable with timeout
            my $read_f = Future::IO->waitfor_readable($socket);
            my $timeout_f = Future::IO->sleep($remaining)->then(sub {
                return Future->fail('tls_timeout');
            });

            my $wait_f = Future->wait_any($read_f, $timeout_f);
            await $wait_f;

            if ($wait_f->is_failed) {
                die Async::Redis::Error::Timeout->new(
                    message => "TLS handshake timed out",
                    timeout => $self->{connect_timeout},
                );
            }
        }
        elsif ($IO::Socket::SSL::SSL_ERROR == IO::Socket::SSL::SSL_ERROR_WANT_WRITE()) {
            # Wait for socket to become writable with timeout
            my $write_f = Future::IO->waitfor_writable($socket);
            my $timeout_f = Future::IO->sleep($remaining)->then(sub {
                return Future->fail('tls_timeout');
            });

            my $wait_f = Future->wait_any($write_f, $timeout_f);
            await $wait_f;

            if ($wait_f->is_failed) {
                die Async::Redis::Error::Timeout->new(
                    message => "TLS handshake timed out",
                    timeout => $self->{connect_timeout},
                );
            }
        }
        else {
            # Actual error
            die Async::Redis::Error::Connection->new(
                message => "TLS handshake failed: " . IO::Socket::SSL::errstr(),
                host    => $self->{host},
                port    => $self->{port},
            );
        }
    }
}

# Reconnect with exponential backoff
async sub _reconnect {
    my ($self) = @_;

    while (!$self->{connected}) {
        $self->{_reconnect_attempt}++;
        my $delay = $self->_calculate_backoff($self->{_reconnect_attempt});

        eval {
            await $self->connect;
        };

        if ($@) {
            my $error = $@;

            # Fire on_error callback
            if ($self->{on_error}) {
                $self->{on_error}->($self, $error);
            }

            # Wait before next attempt
            await Future::IO->sleep($delay);
        }
    }
}

# Reconnect and replay pubsub subscriptions
async sub _reconnect_pubsub {
    my ($self) = @_;

    my $sub = $self->{_subscription}
        or die Async::Redis::Error::Disconnected->new(
            message => "No subscription to replay",
        );

    my @replay = $sub->get_replay_commands;

    # Ensure connection state is fully cleaned up before reconnecting.
    # _reset_connection may have already been called by _read_response,
    # but if the socket was closed externally, we need to clean up
    # stale IO watchers and state here. It is safe to call twice —
    # the on_disconnect callback is guarded by $was_connected.
    $self->_reset_connection('pubsub_reconnect');

    await $self->_reconnect;

    # Replay all subscription commands
    for my $cmd (@replay) {
        my ($command, @args) = @$cmd;

        await $self->_send_command($command, @args);

        # Read and discard subscription confirmations
        for my $arg (@args) {
            await $self->_read_pubsub_frame();
        }
    }

    # Re-enter pubsub mode
    $self->{in_pubsub} = 1;
}

# Execute a Redis command
async sub command {
    my ($self, $cmd, @args) = @_;

    # Check for fork - invalidate connection if PID changed
    $self->_check_fork;

    # Block regular commands on pubsub connection
    if ($self->{in_pubsub}) {
        my $ucmd = uc($cmd // '');
        unless ($ucmd =~ /^(SUBSCRIBE|UNSUBSCRIBE|PSUBSCRIBE|PUNSUBSCRIBE|SSUBSCRIBE|SUNSUBSCRIBE|PING|QUIT)$/) {
            die Async::Redis::Error::Protocol->new(
                message => "Cannot execute '$cmd' on connection in PubSub mode",
            );
        }
    }

    # Apply key prefixing if configured
    if (defined $self->{prefix} && $self->{prefix} ne '') {
        @args = Async::Redis::KeyExtractor::apply_prefix(
            $self->{prefix}, $cmd, @args
        );
    }

    # Route through auto-pipeline if enabled
    if ($self->{_auto_pipeline}) {
        return await $self->{_auto_pipeline}->command($cmd, @args);
    }

    # If disconnected and reconnect enabled, try to reconnect
    if (!$self->{connected} && $self->{reconnect}) {
        await $self->_reconnect;
    }

    die Async::Redis::Error::Disconnected->new(
        message => "Not connected",
    ) unless $self->{connected};

    # Telemetry: start span and log send
    my $span_context;
    my $start_time = Time::HiRes::time();
    if ($self->{_telemetry}) {
        $span_context = $self->{_telemetry}->start_command_span($cmd, @args);
        $self->{_telemetry}->log_send($cmd, @args);
    }

    my $raw_cmd = $self->_build_command($cmd, @args);

    # Calculate deadline based on command type
    my $deadline = $self->_calculate_deadline($cmd, @args);

    # Create response future and register in inflight queue BEFORE sending
    # This ensures responses are matched in order
    my $response_future = Future->new;
    $self->_add_inflight($response_future, $cmd, \@args, $deadline);

    my $result;
    my $error;

    my $send_ok = eval {
        # Send command
        await $self->_send($raw_cmd);
        1;
    };

    if (!$send_ok) {
        $error = $@;
        # Send failed - remove from inflight and fail
        $self->_shift_inflight;  # Remove the entry we just added
        $response_future->fail($error) unless $response_future->is_ready;
    } else {
        # Trigger the response reader (fire and forget - it runs in background)
        $self->_ensure_response_reader->retain;

        # Wait for our response future to be completed by the reader
        my $await_ok = eval {
            $result = await $response_future;
            1;
        };

        if (!$await_ok) {
            $error = $@;
        }
    }

    # Telemetry: log result and end span
    if ($self->{_telemetry}) {
        my $elapsed_ms = (Time::HiRes::time() - $start_time) * 1000;
        if ($error) {
            $self->{_telemetry}->log_error($error);
        }
        else {
            $self->{_telemetry}->log_recv($result, $elapsed_ms);
        }
        $self->{_telemetry}->end_command_span($span_context, $error);
    }

    die $error if $error;
    return $result;
}

# Read response with deadline enforcement
async sub _read_response_with_deadline {
    my ($self, $deadline, $cmd_ref) = @_;

    # First check if parser already has a complete message
    if (my $msg = $self->{parser}->get_message) {
        return $msg;
    }

    # Read until we get a complete message
    while (1) {
        my $remaining = $deadline - Time::HiRes::time();

        if ($remaining <= 0) {
            $self->_reset_connection;
            die Async::Redis::Error::Timeout->new(
                message        => "Request timed out after $self->{request_timeout}s",
                command        => $cmd_ref,
                timeout        => $self->{request_timeout},
                maybe_executed => 1,  # already sent the command
            );
        }

        # Use wait_any for timeout
        my $read_f = Future::IO->read($self->{socket}, 65536);

        # Store reference so disconnect() can cancel it
        $self->{_current_read_future} = $read_f;

        my $timeout_f = Future::IO->sleep($remaining)->then(sub {
            return Future->fail('read_timeout');
        });

        my $wait_f = Future->wait_any($read_f, $timeout_f);
        await $wait_f;

        # Clear stored reference after await completes
        $self->{_current_read_future} = undef;

        # Check if read was cancelled (by disconnect)
        if ($read_f->is_cancelled) {
            die Async::Redis::Error::Disconnected->new(
                message => "Disconnected during read",
            );
        }

        if ($wait_f->is_failed) {
            my ($error) = $wait_f->failure;
            if ($error eq 'read_timeout') {
                $self->_reset_connection;
                die Async::Redis::Error::Timeout->new(
                    message        => "Request timed out after $self->{request_timeout}s",
                    command        => $cmd_ref,
                    timeout        => $self->{request_timeout},
                    maybe_executed => 1,
                );
            }
            $self->_reset_connection;
            die Async::Redis::Error::Connection->new(
                message => "$error",
            );
        }

        # Get the read result
        my $buf = $wait_f->get;

        # EOF
        if (!defined $buf || length($buf) == 0) {
            $self->_reset_connection;
            die Async::Redis::Error::Connection->new(
                message => "Connection closed by server",
            );
        }

        $self->{parser}->parse($buf);

        if (my $msg = $self->{parser}->get_message) {
            return $msg;
        }
    }
}

# Reset connection after timeout (stream is desynced)
sub _reset_connection {
    my ($self, $reason) = @_;
    $reason //= 'timeout';

    my $was_connected = $self->{connected};

    # Cancel any active read future BEFORE closing socket
    # This ensures Future::IO unregisters its watcher while fileno is still valid
    if ($self->{_current_read_future} && !$self->{_current_read_future}->is_ready) {
        $self->{_current_read_future}->cancel;
        $self->{_current_read_future} = undef;
    }

    # Cancel any pending inflight operations before closing socket
    if (my $inflight = $self->{inflight}) {
        for my $entry (@$inflight) {
            if ($entry->{future} && !$entry->{future}->is_ready) {
                $entry->{future}->cancel;
            }
        }
        $self->{inflight} = [];
    }

    if ($self->{socket}) {
        $self->_close_socket;
    }

    $self->{connected} = 0;
    $self->{parser} = undef;
    $self->{_reading_responses} = 0;
    $self->{in_pubsub} = 0;

    if ($was_connected && $self->{on_disconnect}) {
        $self->{on_disconnect}->($self, $reason);
    }
}

# Decode Protocol::Redis response to Perl value
sub _decode_response {
    my ($self, $msg) = @_;

    return undef unless $msg;

    my $type = $msg->{type};
    my $data = $msg->{data};

    # Simple string (+)
    if ($type eq '+') {
        return $data;
    }
    # Error (-)
    elsif ($type eq '-') {
        die Async::Redis::Error::Redis->from_message($data);
    }
    # Integer (:)
    elsif ($type eq ':') {
        return 0 + $data;
    }
    # Bulk string ($)
    elsif ($type eq '$') {
        return $data;  # undef for null bulk
    }
    # Array (*)
    elsif ($type eq '*') {
        return undef unless defined $data;  # null array
        return [ map { $self->_decode_response($_) } @$data ];
    }

    return $data;
}

# ============================================================================
# Convenience Commands
# ============================================================================

async sub ping {
    my ($self) = @_;
    return await $self->command('PING');
}

async sub set {
    my ($self, $key, $value, %opts) = @_;
    my @cmd = ('SET', $key, $value);
    push @cmd, 'EX', $opts{ex} if exists $opts{ex};
    push @cmd, 'PX', $opts{px} if exists $opts{px};
    push @cmd, 'NX' if $opts{nx};
    push @cmd, 'XX' if $opts{xx};
    return await $self->command(@cmd);
}

async sub get {
    my ($self, $key) = @_;
    return await $self->command('GET', $key);
}

async sub del {
    my ($self, @keys) = @_;
    return await $self->command('DEL', @keys);
}

async sub incr {
    my ($self, $key) = @_;
    return await $self->command('INCR', $key);
}

async sub lpush {
    my ($self, $key, @values) = @_;
    return await $self->command('LPUSH', $key, @values);
}

async sub rpush {
    my ($self, $key, @values) = @_;
    return await $self->command('RPUSH', $key, @values);
}

async sub lpop {
    my ($self, $key) = @_;
    return await $self->command('LPOP', $key);
}

async sub lrange {
    my ($self, $key, $start, $stop) = @_;
    return await $self->command('LRANGE', $key, $start, $stop);
}

async sub keys {
    my ($self, $pattern) = @_;
    return await $self->command('KEYS', $pattern // '*');
}

async sub flushdb {
    my ($self) = @_;
    return await $self->command('FLUSHDB');
}

# ============================================================================
# Lua Scripting
# ============================================================================

async sub script_load {
    my ($self, $script) = @_;
    return await $self->command('SCRIPT', 'LOAD', $script);
}

async sub script_exists {
    my ($self, @shas) = @_;
    return await $self->command('SCRIPT', 'EXISTS', @shas);
}

async sub script_flush {
    my ($self, $mode) = @_;
    my @args = ('SCRIPT', 'FLUSH');
    push @args, $mode if $mode;  # ASYNC or SYNC
    return await $self->command(@args);
}

async sub script_kill {
    my ($self) = @_;
    return await $self->command('SCRIPT', 'KILL');
}

async sub evalsha_or_eval {
    my ($self, $sha, $script, $numkeys, @keys_and_args) = @_;

    # Try EVALSHA first
    my $result;
    eval {
        $result = await $self->evalsha($sha, $numkeys, @keys_and_args);
    };

    if ($@) {
        my $error = $@;

        # Check if it's a NOSCRIPT error
        if ("$error" =~ /NOSCRIPT/i) {
            # Fall back to EVAL (which also loads the script)
            $result = await $self->eval($script, $numkeys, @keys_and_args);
        }
        else {
            # Re-throw other errors
            die $error;
        }
    }

    return $result;
}

sub script {
    my ($self, $code) = @_;
    return Async::Redis::Script->new(
        redis  => $self,
        script => $code,
    );
}

# Define a named script command
# Usage: $redis->define_command(name => { keys => N, lua => '...' })
sub define_command {
    my ($self, $name, $def) = @_;

    die "Command name required" unless defined $name && length $name;
    die "Command definition required" unless ref $def eq 'HASH';
    die "Lua script required (lua => '...')" unless defined $def->{lua};

    # Validate name (alphanumeric and underscore only)
    die "Invalid command name '$name' - use only alphanumeric and underscore"
        unless $name =~ /^[a-zA-Z_][a-zA-Z0-9_]*$/;

    my $script = Async::Redis::Script->new(
        redis       => $self,
        script      => $def->{lua},
        name        => $name,
        num_keys    => $def->{keys} // 'dynamic',
        description => $def->{description},
    );

    $self->{_scripts}{$name} = $script;

    # Optional: install as method on this instance
    if ($def->{install}) {
        $self->_install_script_method($name);
    }

    return $script;
}

# Run a registered script by name
# Usage: $redis->run_script('name', @keys_then_args)
# If num_keys is 'dynamic', first arg is the key count
async sub run_script {
    my ($self, $name, @args) = @_;

    my $script = $self->{_scripts}{$name}
        or die "Unknown script: '$name' - use define_command() first";

    my $num_keys = $script->num_keys;

    # Handle dynamic key count
    if ($num_keys eq 'dynamic') {
        $num_keys = shift @args;
        die "Key count required as first argument for dynamic script '$name'"
            unless defined $num_keys;
    }

    # Split args into keys and argv
    my @keys = splice(@args, 0, $num_keys);
    return await $script->run(\@keys, \@args);
}

# Get a registered script by name
sub get_script {
    my ($self, $name) = @_;
    return $self->{_scripts}{$name};
}

# List all registered script names
sub list_scripts {
    my ($self) = @_;
    return CORE::keys %{$self->{_scripts}};
}

# Preload all registered scripts to Redis
# Useful before pipeline execution
async sub preload_scripts {
    my ($self) = @_;

    my @names = $self->list_scripts;
    return 0 unless @names;

    for my $name (@names) {
        my $script = $self->{_scripts}{$name};
        await $self->script_load($script->script);
    }

    return scalar @names;
}

# Install a script as a method (internal)
sub _install_script_method {
    my ($self, $name) = @_;

    # Create closure that captures $name
    my $method = sub {
        my ($self, @args) = @_;
        return $self->run_script($name, @args);
    };

    # Install on the class (affects all instances)
    no strict 'refs';
    no warnings 'redefine';
    *{"Async::Redis::$name"} = $method;
}

# ============================================================================
# SCAN Iterators
# ============================================================================

sub scan_iter {
    my ($self, %opts) = @_;
    return Async::Redis::Iterator->new(
        redis   => $self,
        command => 'SCAN',
        match   => $opts{match},
        count   => $opts{count},
        type    => $opts{type},
    );
}

sub hscan_iter {
    my ($self, $key, %opts) = @_;
    return Async::Redis::Iterator->new(
        redis   => $self,
        command => 'HSCAN',
        key     => $key,
        match   => $opts{match},
        count   => $opts{count},
    );
}

sub sscan_iter {
    my ($self, $key, %opts) = @_;
    return Async::Redis::Iterator->new(
        redis   => $self,
        command => 'SSCAN',
        key     => $key,
        match   => $opts{match},
        count   => $opts{count},
    );
}

sub zscan_iter {
    my ($self, $key, %opts) = @_;
    return Async::Redis::Iterator->new(
        redis   => $self,
        command => 'ZSCAN',
        key     => $key,
        match   => $opts{match},
        count   => $opts{count},
    );
}

# ============================================================================
# Transactions
# ============================================================================

async sub multi {
    my ($self, $callback) = @_;

    # Prevent nested multi() calls
    die "Cannot nest multi() calls - already in a transaction"
        if $self->{in_multi};

    # Mark that we're collecting transaction commands
    $self->{in_multi} = 1;

    my @commands;
    eval {
        # Create transaction collector
        my $tx = Async::Redis::Transaction->new(redis => $self);

        # Run callback to collect commands
        await $callback->($tx);

        @commands = $tx->commands;
    };
    my $collect_error = $@;

    if ($collect_error) {
        $self->{in_multi} = 0;
        die $collect_error;
    }

    # If no commands queued, return empty result
    unless (@commands) {
        $self->{in_multi} = 0;
        return [];
    }

    # Execute transaction (in_multi already set)
    return await $self->_execute_transaction(\@commands);
}

async sub _execute_transaction {
    my ($self, $commands) = @_;

    # in_multi should already be set by caller

    my $results;
    eval {
        # Send MULTI
        await $self->command('MULTI');

        # Queue all commands (they return +QUEUED)
        for my $cmd (@$commands) {
            await $self->command(@$cmd);
        }

        # Execute and get results
        $results = await $self->command('EXEC');
    };
    my $error = $@;

    # Always clear transaction state
    $self->{in_multi} = 0;

    if ($error) {
        # Try to clean up
        eval { await $self->command('DISCARD') };
        die $error;
    }

    return $results;
}

# Accessor for pool cleanliness tracking
sub in_multi { shift->{in_multi} }
sub watching { shift->{watching} }
sub in_pubsub { shift->{in_pubsub} }
sub inflight_count { scalar @{shift->{inflight} // []} }

# Is connection dirty (unsafe to reuse)?
sub is_dirty {
    my ($self) = @_;

    return 1 if $self->{in_multi};
    return 1 if $self->{watching};
    return 1 if $self->{in_pubsub};
    return 1 if @{$self->{inflight} // []} > 0;

    return 0;
}

async sub watch {
    my ($self, @keys) = @_;
    $self->{watching} = 1;
    return await $self->command('WATCH', @keys);
}

async sub unwatch {
    my ($self) = @_;
    my $result = await $self->command('UNWATCH');
    $self->{watching} = 0;
    return $result;
}

async sub multi_start {
    my ($self) = @_;
    $self->{in_multi} = 1;
    return await $self->command('MULTI');
}

async sub exec {
    my ($self) = @_;
    my $result = await $self->command('EXEC');
    $self->{in_multi} = 0;
    $self->{watching} = 0;  # EXEC clears watches
    return $result;
}

async sub discard {
    my ($self) = @_;
    my $result = await $self->command('DISCARD');
    $self->{in_multi} = 0;
    # Note: DISCARD does NOT clear watches
    return $result;
}

async sub watch_multi {
    my ($self, $keys, $callback) = @_;

    # WATCH the keys
    await $self->watch(@$keys);

    # Get current values of watched keys
    my %watched;
    for my $key (@$keys) {
        $watched{$key} = await $self->get($key);
    }

    # Create transaction collector
    my $tx = Async::Redis::Transaction->new(redis => $self);

    # Run callback with watched values
    await $callback->($tx, \%watched);

    my @commands = $tx->commands;

    # If no commands queued, just unwatch and return empty
    unless (@commands) {
        await $self->unwatch;
        return [];
    }

    # Execute transaction
    $self->{in_multi} = 1;

    my $results;
    eval {
        await $self->command('MULTI');

        for my $cmd (@commands) {
            await $self->command(@$cmd);
        }

        $results = await $self->command('EXEC');
    };
    my $error = $@;

    $self->{in_multi} = 0;
    $self->{watching} = 0;

    if ($error) {
        eval { await $self->command('DISCARD') };
        die $error;
    }

    # EXEC returns undef/nil if WATCH failed
    return $results;
}

# ============================================================================
# PUB/SUB
# ============================================================================

# Wait for inflight commands to complete before mode change
async sub _wait_for_inflight_drain {
    my ($self, $timeout) = @_;
    $timeout //= 30;

    return unless @{$self->{inflight}};

    my $deadline = Time::HiRes::time() + $timeout;

    while (@{$self->{inflight}} && Time::HiRes::time() < $deadline) {
        await Future::IO->sleep(0.001);
    }

    if (@{$self->{inflight}}) {
        $self->_fail_all_inflight("Timeout waiting for inflight commands");
    }
}

async sub publish {
    my ($self, $channel, $message) = @_;
    return await $self->command('PUBLISH', $channel, $message);
}

async sub spublish {
    my ($self, $channel, $message) = @_;
    return await $self->command('SPUBLISH', $channel, $message);
}

# Subscribe to channels - returns a Subscription object
async sub subscribe {
    my ($self, @channels) = @_;

    die Async::Redis::Error::Disconnected->new(
        message => "Not connected",
    ) unless $self->{connected};

    # Wait for pending commands before entering PubSub mode
    await $self->_wait_for_inflight_drain;

    # Create or reuse subscription
    my $sub = $self->{_subscription} //= Async::Redis::Subscription->new(redis => $self);

    # Send SUBSCRIBE command
    await $self->_send_command('SUBSCRIBE', @channels);

    # Read subscription confirmations
    for my $ch (@channels) {
        my $msg = await $self->_read_pubsub_frame();
        # Response: ['subscribe', $channel, $count]
        $sub->_add_channel($ch);
    }

    $self->{in_pubsub} = 1;

    return $sub;
}

# Pattern subscribe
async sub psubscribe {
    my ($self, @patterns) = @_;

    die Async::Redis::Error::Disconnected->new(
        message => "Not connected",
    ) unless $self->{connected};

    # Wait for pending commands before entering PubSub mode
    await $self->_wait_for_inflight_drain;

    my $sub = $self->{_subscription} //= Async::Redis::Subscription->new(redis => $self);

    await $self->_send_command('PSUBSCRIBE', @patterns);

    for my $p (@patterns) {
        my $msg = await $self->_read_pubsub_frame();
        $sub->_add_pattern($p);
    }

    $self->{in_pubsub} = 1;

    return $sub;
}

# Sharded subscribe (Redis 7+)
async sub ssubscribe {
    my ($self, @channels) = @_;

    die Async::Redis::Error::Disconnected->new(
        message => "Not connected",
    ) unless $self->{connected};

    # Wait for pending commands before entering PubSub mode
    await $self->_wait_for_inflight_drain;

    my $sub = $self->{_subscription} //= Async::Redis::Subscription->new(redis => $self);

    await $self->_send_command('SSUBSCRIBE', @channels);

    for my $ch (@channels) {
        my $msg = await $self->_read_pubsub_frame();
        $sub->_add_sharded_channel($ch);
    }

    $self->{in_pubsub} = 1;

    return $sub;
}

# Read pubsub frame (subscription confirmation or message)
async sub _read_pubsub_frame {
    my ($self) = @_;

    die Async::Redis::Error::Disconnected->new(
        message => "Not connected",
    ) unless $self->{connected};

    my $msg = await $self->_read_response();
    return $self->_decode_response($msg);
}

# Send command without reading response (for pubsub)
async sub _send_command {
    my ($self, @args) = @_;

    my $cmd = $self->_build_command(@args);
    await $self->_send($cmd);
}

# Read next pubsub message (blocking) - for compatibility
async sub _read_pubsub_message {
    my ($self) = @_;

    my $msg = await $self->_read_response();

    # Message format: ['message', $channel, $payload]
    # or: ['pmessage', $pattern, $channel, $payload]
    return $msg;
}

# ============================================================================
# Pipelining
# ============================================================================

sub pipeline {
    my ($self, %opts) = @_;
    return Async::Redis::Pipeline->new(
        redis     => $self,
        max_depth => $opts{max_depth} // $self->{pipeline_depth},
    );
}

# Execute multiple commands, return all responses
async sub _execute_pipeline {
    my ($self, $commands) = @_;

    die "Not connected" unless $self->{connected};

    return [] unless @$commands;

    # Wait for any inflight regular commands to complete before pipeline
    # This prevents interleaving pipeline responses with regular command responses
    await $self->_wait_for_inflight_drain;

    # Take over reading - prevent response reader from running
    $self->{_reading_responses} = 1;

    my $start_time = Time::HiRes::time();
    my @responses;
    my $count = scalar @$commands;

    my $ok = eval {
        # Send all commands
        my $data = '';
        for my $cmd (@$commands) {
            $data .= $self->_build_command(@$cmd);
        }
        await $self->_send($data);

        # Read all responses, capturing per-slot Redis errors
        for my $i (1 .. $count) {
            my $msg = await $self->_read_response();

            # Capture Redis errors inline rather than dying
            my $result;
            eval {
                $result = $self->_decode_response($msg);
            };
            if ($@) {
                # Capture the error object inline in the results
                $result = $@;
            }
            push @responses, $result;
        }
        1;
    };

    my $error = $@;

    # Release reading lock
    $self->{_reading_responses} = 0;

    if (!$ok) {
        die $error;
    }

    # Telemetry: record pipeline metrics
    if ($self->{_telemetry}) {
        my $elapsed_ms = (Time::HiRes::time() - $start_time) * 1000;
        $self->{_telemetry}->record_pipeline($count, $elapsed_ms);
    }

    return \@responses;
}

1;

__END__