fred

To support pipelining, each Connection maintains an internal buffer: VecDeque<Command>. This buffer tracks in-flight requests that are waiting for a response from the server.

The Workflow:

1. Writing: When a command is written, it is encoded into a frame, pushed to the back of the buffer, and sent to the transport.
2. Reading: When a frame is received from the server, the client pops the command from the front of the buffer.

Because Redis/Valkey guarantees that responses are returned in the same order as requests, the VecDeque ensures that the next received frame is correctly associated with the oldest pending command.

impl Connection {
  pub async fn write(&mut self, command: Command) -> Result<(), Error> {
    let frame = encode_frame(&command)?;
    self.buffer.push_back(command);
    self.transport.send(frame).await
  }

  pub async fn read(&mut self) -> Result<Option<(Resp3Frame, Command)>, Error> {
    let frame = self.transport.next().await?;
    Ok(self.buffer.pop_front().map(|cmd| (frame, cmd)))
  }
}

Fred uses a decoupled architecture to separate request-response logic from connection management. This is achieved through a routing task (the connection manager) and a Client (the request interface).

The Routing Task

When you call client.connect(), it spawns a Tokio task that manages all connections to the servers. This task is responsible for:

• Managing private state (connections, retry buffers, replicas) via a Router struct.
• Listening for commands on a channel (command_rx).
• Handling connection-related events (errors, closures, reconnections) independently of individual requests.

The Client Interface

The Client is a thin, clonable wrapper around ClientInner. It provides the public API (e.g., get, set) by sending Command objects to the routing task via an unbounded channel (command_tx).

Request-Response Lifecycle

1. The caller invokes a method (e.g., client.get(key)).
2. A Command is created containing the command type, arguments, and a oneshot channel sender (tx).
3. The Command is sent to the routing task.
4. The routing task processes the command and sends the response back to the caller via the oneshot receiver (rx).

This model allows the client to handle connection failures or reconnections without requiring the user to manually manage connection state for every request.

// High-level pattern for request-response functions
impl Client {
    pub async fn get<K: Into<Key>>(&self, key: K) -> Result<Value, Error> {
        let (tx, rx) = oneshot_channel();
        let command = Command {
            kind: CommandKind::Get,
            args: vec![key.into().into()],
            tx
        };

        self.inner.command_tx.load().send(command);
        rx.await.and_then(|f| f.into())
    }
}

The write path in Fred is designed to allow automatic pipelining across different tasks. The routing task processes commands from its receiver and writes them to the appropriate server sockets without waiting for a response before moving to the next command.

Write Process

1. Receive: The router receives a RouterCommand from the command_rx channel.
2. Route: It identifies the target Server (applying cluster hashing if necessary).
3. Encode: The command is encoded as a Resp3Frame using the redis-protocol crate.
4. Buffer & Write: The frame is stored in a socket-associated buffer and written to the TcpStream.
5. Error Handling: If a write fails, the command is placed in the Router's retry_buffer.

RouterCommand Types

The routing task handles more than just standard Redis commands. It uses a RouterCommand enum to manage complex operations:

• Command(Command): A standard Valkey/Redis command.
• Reconnect(Server): Triggers a reconnection attempt for a specific server.
• Transaction(Vec<Command>): A group of commands to be executed as a transaction.
• Pipeline(Vec<Command>): A group of commands to be pipelined.

Reconnection Logic

When a Reconnect command is received, the router enters a loop using the provided ReconnectPolicy:

• It attempts to reconnect using reconnect_once.
• If it fails, it sleeps for a duration determined by reconnect_policy.next_delay().
• Once a connection is successfully re-established, the router calls retry_failed_commands() to drain the retry_buffer and retry commands that were in-flight when the connection was lost.

Note: While the router is in a sleep loop during a reconnection attempt, no other normal commands will be written to any other sockets.

// Simplified RouterCommand structure
enum RouterCommand {
  Command(Command),
  Reconnect(Server),
  Transaction(Vec<Command>),
  Pipeline(Vec<Command>),
  // ...
}

The Fred client manages multiple Redis connections within a single Tokio task. To handle concurrent reads from multiple sockets and writes from the user without losing data, the client uses a specialized architecture:

1. ReadAllFuture: Instead of using a standard Stream interface (which is not cancel-safe when combined with TcpStream in a select loop), Fred implements a custom ReadAllFuture. This future polls all active connections and returns a collection of responses (Responses) containing the server identifier and the received frame.
2. read_or_write Loop: The core execution loop uses tokio::select! with a biased configuration. It prioritizes reading inbound responses from servers over processing new outgoing commands.
3. Response Processing: When a frame is received, process_response matches the frame to the expected command in the connection's buffer and sends the result back to the caller via a channel.
4. Error Handling: If a connection closes due to an IO error, the client automatically disconnects the server, moves in-flight commands to a retry_buffer, and triggers a reconnection attempt via RouterCommand::Reconnect.