fix(replication): stream SELECT for multi-db writes + serve SELECT on replicas

CHANGELOG.md

@@ -6,6 +6,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed — multi-db replication: master streams `SELECT`, replicas serve it (HIGH-2)
+
+- **Writes outside db 0 landed in db 0 on the replica**: the replica-side
+  drain already understood in-stream `SELECT` context, but the master never
+  emitted it — every replicated command applied to the replica's db 0
+  regardless of the db it executed in on the master. `record_local_write_db`
+  now prepends a `SELECT <db>` record whenever the writing connection's db
+  differs from the stream's per-shard db context (`ReplicationState::
+  stream_db`); the context is reset in the same synchronous stretch as every
+  FULLRESYNC snapshot capture (Redis's `slaveseldb = -1` idiom), so a fresh
+  replica always sees an explicit context before its first non-0-db write.
+- **`+CONTINUE` keeps the db context across reconnects**: resumed backlog
+  bytes only carry `SELECT` at db CHANGES, so the replica now preserves its
+  drain-side db in `ReplicaTaskConfig::stream_db` across link drops (reset to
+  0 on FULLRESYNC). In-memory only — a replica process restart starts at
+  offset 0 and always full-resyncs.
+- **`SELECT` was rejected on read-only replicas** (task #23): flagged W in
+  the metadata table, so the READONLY guard blocked it — a client could
+  never read a replica's non-zero dbs. All three dispatch paths now serve
+  SELECT on replicas (connection-state only, Redis parity).
+
 ### Fixed — replication round-2 hardening: TEMPORAL.INVALIDATE, replay liveness, blob endpoint checks
 
 - **TEMPORAL.INVALIDATE never replicated** (round-2 finding B): the handler

src/replication/master.rs

@@ -670,7 +670,21 @@ pub async fn handle_psync_inline_single_shard(
             let snapshot_offset = {
                 let off = repl_state
                     .read()
-                    .map(|g| g.total_offset())
+                    .map(|g| {
+                        // HIGH-2 (task #22): reset the stream's db context in
+                        // the SAME synchronous stretch as the snapshot capture
+                        // — every byte at offset ≥ snapshot_offset then starts
+                        // from "db unknown", so the first post-snapshot write
+                        // re-emits `SELECT <db>` and this replica's drain
+                        // (which starts at db 0 after loading the RDB) can
+                        // never bind a write to the wrong db. Redis's
+                        // `slaveseldb = -1` idiom. Redundant re-SELECTs for
+                        // already-attached replicas are idempotent.
+                        if let Some(slot) = g.stream_db.first() {
+                            slot.store(-1, std::sync::atomic::Ordering::Relaxed);
+                        }
+                        g.total_offset()
+                    })
                     .map_err(|_| anyhow::anyhow!("lock poisoned"))?;
                 // Shard 0 is this thread's shard — use the thread-local slice.
                 crate::shard::slice::with_shard(|s| {

src/replication/replica.rs

@@ -26,6 +26,16 @@ pub struct ReplicaTaskConfig {
     pub num_shards: usize,
     pub persistence_dir: Option<String>,
     pub listening_port: u16,
+    /// Logical-db context of the replication stream, preserved ACROSS
+    /// reconnects (HIGH-2, task #22): a `+CONTINUE` partial resync replays
+    /// backlog bytes that only contain `SELECT` at db CHANGES — if the stream
+    /// was in db N when the link dropped, the resumed bytes carry no fresh
+    /// SELECT and a 0-reset drain would misapply them to db 0. Seeded into
+    /// `stream_commands`'s drain state on `+CONTINUE`; reset to 0 on
+    /// `+FULLRESYNC` (the master resets its own stream-db at snapshot capture,
+    /// so post-snapshot bytes always re-establish context). In-memory only: a
+    /// replica process restart starts at offset 0 → always FULLRESYNC.
+    pub stream_db: std::sync::atomic::AtomicUsize,
 }
 
 /// Entry point for the outbound replica task.
@@ -235,6 +245,10 @@ async fn run_handshake_and_stream(
                 *state = ReplicaHandshakeState::Streaming;
             }
         }
+        // FULLRESYNC resets the stream's db context: the master reset its own
+        // stream-db in the snapshot-capture stretch, so post-snapshot bytes
+        // always re-establish it with an explicit SELECT (task #22).
+        cfg.stream_db.store(0, Ordering::Relaxed);
         stream_commands(&mut stream, cfg).await?;
     } else if response.starts_with(b"+CONTINUE") {
         // Partial resync: stream from current offset
@@ -258,7 +272,10 @@ async fn run_handshake_and_stream(
 #[cfg(feature = "runtime-tokio")]
 async fn stream_commands(stream: &mut TcpStream, cfg: &ReplicaTaskConfig) -> anyhow::Result<()> {
     let mut buf = BytesMut::with_capacity(65536);
-    let mut selected_db = 0usize;
+    // Seeded from the task-level slot (NOT 0): a +CONTINUE resume must keep
+    // the db context the stream was in when the link dropped — see
+    // `ReplicaTaskConfig::stream_db`.
+    let mut selected_db = cfg.stream_db.load(Ordering::Relaxed);
 
     loop {
         let n = stream.read_buf(&mut buf).await?;
@@ -284,6 +301,9 @@ async fn stream_commands(stream: &mut TcpStream, cfg: &ReplicaTaskConfig) -> any
                     .fetch_add(outcome.consumed as u64, Ordering::Relaxed);
             }
         }
+        // Persist the drain's db context so a reconnect (+CONTINUE) resumes
+        // in the same logical db (HIGH-2, task #22).
+        cfg.stream_db.store(selected_db, Ordering::Relaxed);
         if outcome.fatal {
             return Err(anyhow::anyhow!(
                 "replication stream parse error — dropping connection to force resync"
@@ -496,6 +516,10 @@ async fn run_handshake_and_stream(
                 *state = ReplicaHandshakeState::Streaming;
             }
         }
+        // FULLRESYNC resets the stream's db context: the master reset its own
+        // stream-db in the snapshot-capture stretch, so post-snapshot bytes
+        // always re-establish it with an explicit SELECT (task #22).
+        cfg.stream_db.store(0, Ordering::Relaxed);
         stream_commands(&mut stream, cfg).await?;
     } else if response.starts_with(b"+CONTINUE") {
         if let Ok(mut rs) = cfg.repl_state.write() {
@@ -520,7 +544,10 @@ async fn stream_commands(
     use monoio::io::AsyncReadRent;
 
     let mut buf = BytesMut::with_capacity(65536);
-    let mut selected_db = 0usize;
+    // Seeded from the task-level slot (NOT 0): a +CONTINUE resume must keep
+    // the db context the stream was in when the link dropped — see
+    // `ReplicaTaskConfig::stream_db`.
+    let mut selected_db = cfg.stream_db.load(Ordering::Relaxed);
 
     loop {
         let tmp = vec![0u8; 65536];
@@ -549,6 +576,9 @@ async fn stream_commands(
                     .fetch_add(outcome.consumed as u64, Ordering::Relaxed);
             }
         }
+        // Persist the drain's db context so a reconnect (+CONTINUE) resumes
+        // in the same logical db (HIGH-2, task #22).
+        cfg.stream_db.store(selected_db, Ordering::Relaxed);
         if outcome.fatal {
             return Err(anyhow::anyhow!(
                 "replication stream parse error — dropping connection to force resync"

src/replication/state.rs

@@ -45,6 +45,15 @@ pub struct ReplicationState {
     /// startup via [`set_backlog_capacity`](Self::set_backlog_capacity); never
     /// resizes already-allocated backlogs.
     pub backlog_capacity: usize,
+    /// Per-shard logical-db context of the replication byte stream (HIGH-2,
+    /// task #22): the db of the LAST data command recorded into the shard's
+    /// backlog, or `-1` = unknown. `record_local_write_db` prepends a
+    /// `SELECT <db>` record whenever the writing connection's db differs, so
+    /// a replica's drain binds each command to the master's db. Reset to `-1`
+    /// in the SAME synchronous stretch as every FULLRESYNC snapshot capture —
+    /// the first post-snapshot write then re-establishes the context for the
+    /// freshly-attached replica (Redis's `slaveseldb = -1` idiom).
+    pub stream_db: Vec<std::sync::atomic::AtomicI64>,
 }
 
 pub enum ReplicationRole {
@@ -81,6 +90,9 @@ impl ReplicationState {
                 .collect(),
             is_replica_mirror: Arc::new(AtomicBool::new(false)),
             backlog_capacity: DEFAULT_REPL_BACKLOG_SIZE,
+            stream_db: (0..num_shards)
+                .map(|_| std::sync::atomic::AtomicI64::new(-1))
+                .collect(),
         }
     }
 

src/server/conn/handler_monoio/dispatch.rs

@@ -471,6 +471,7 @@ pub(super) fn try_handle_replicaof(
                         num_shards: ctx.num_shards,
                         persistence_dir: None,
                         listening_port: 0,
+                        stream_db: std::sync::atomic::AtomicUsize::new(0),
                     };
                     monoio::spawn(crate::replication::replica::run_replica_task(cfg));
                 }
@@ -680,6 +681,13 @@ pub(super) fn try_enforce_readonly(
         return false;
     }
     if metadata::is_write(cmd) {
+        // SELECT is flagged W in the metadata table (it routes through the
+        // write dispatch paths) but only mutates CONNECTION state — Redis
+        // serves it on replicas, and a client cannot read a replica's
+        // non-zero dbs without it (task #23).
+        if cmd.eq_ignore_ascii_case(b"SELECT") {
+            return false;
+        }
         // GRAPH.QUERY is blanket-W in the metadata table because Cypher CAN
         // write; a read-only MATCH/RETURN must still be served by a replica.
         // Reuse the token-scan classifier the write dispatch path branches

src/server/conn/handler_monoio/ft.rs

@@ -89,6 +89,52 @@ pub(super) fn record_local_write(ctx: &ConnectionContext, bytes: Bytes) {
     crate::shard::self_msg::push(crate::shard::dispatch::ShardMessage::ReplicaLiveFanout { bytes });
 }
 
+/// Db-aware variant of [`record_local_write`] (HIGH-2, task #22): prepends a
+/// `SELECT <db>` record whenever the writing connection's logical db differs
+/// from the stream's current db context (`ReplicationState::stream_db`), so a
+/// replica's drain binds this command to the SAME db the master executed it
+/// in. Both records go through `record_local_write` — backlog append + offset
+/// advance stay synchronous with the mutation, and the SELECT is delivered in
+/// order ahead of the payload on the same self-queue.
+///
+/// GRAPH.\* / TEMPORAL.\* records are db-agnostic on the replica (routed
+/// before db resolution) but still use this variant: the uniform invariant is
+/// "stream db context == the writing connection's selected db", so a
+/// db-agnostic record can never silently strand a stale context for the next
+/// KV write.
+pub(super) fn record_local_write_db(ctx: &ConnectionContext, db: usize, bytes: Bytes) {
+    let needs_select = ctx.repl_state.as_ref().is_some_and(|rs| {
+        rs.read().is_ok_and(|g| {
+            g.stream_db.get(ctx.shard_id).is_some_and(|slot| {
+                if slot.load(std::sync::atomic::Ordering::Relaxed) != db as i64 {
+                    slot.store(db as i64, std::sync::atomic::Ordering::Relaxed);
+                    true
+                } else {
+                    false
+                }
+            })
+        })
+    });
+    if needs_select {
+        record_local_write(ctx, Bytes::from(serialize_select(db)));
+    }
+    record_local_write(ctx, bytes);
+}
+
+/// RESP-serialize `SELECT <db>` for the replication stream.
+fn serialize_select(db: usize) -> Vec<u8> {
+    let mut n = itoa::Buffer::new();
+    let db_str = n.format(db);
+    let mut ln = itoa::Buffer::new();
+    let mut buf = Vec::with_capacity(32);
+    buf.extend_from_slice(b"*2\r\n$6\r\nSELECT\r\n$");
+    buf.extend_from_slice(ln.format(db_str.len()).as_bytes());
+    buf.extend_from_slice(b"\r\n");
+    buf.extend_from_slice(db_str.as_bytes());
+    buf.extend_from_slice(b"\r\n");
+    buf
+}
+
 /// Fail-loud marker for planes NOT yet wired into replication (round-2
 /// finding A): WS.* and MQ.* writes persist durably on the master but never
 /// reach a replica — deterministic record forms + replica apply arms are the
@@ -872,7 +918,7 @@ pub(super) async fn try_handle_ft_command(
             && replication_fanout_active(ctx)
         {
             let serialized = crate::persistence::aof::serialize_command(frame);
-            record_local_write(ctx, serialized);
+            record_local_write_db(ctx, conn.selected_db, serialized);
         }
         let mut response = response;
         if let Some(ws_id) = conn.workspace_id.as_ref() {

src/server/conn/handler_monoio/mod.rs

@@ -1337,7 +1337,11 @@ pub(crate) async fn handle_connection_sharded_monoio<
                         if repl_active || ctx.aof_pool.is_some() {
                             let serialized = aof::serialize_command(&frame);
                             let lsn = if repl_active {
-                                ft::record_local_write(ctx, serialized.clone());
+                                ft::record_local_write_db(
+                                    ctx,
+                                    conn.selected_db,
+                                    serialized.clone(),
+                                );
                                 0
                             } else {
                                 aof::AofWriterPool::issue_append_lsn(
@@ -1416,7 +1420,11 @@ pub(crate) async fn handle_connection_sharded_monoio<
                             if repl_active || ctx.aof_pool.is_some() {
                                 let serialized = aof::serialize_command(&frame);
                                 let lsn = if repl_active {
-                                    ft::record_local_write(ctx, serialized.clone());
+                                    ft::record_local_write_db(
+                                        ctx,
+                                        conn.selected_db,
+                                        serialized.clone(),
+                                    );
                                     0
                                 } else {
                                     aof::AofWriterPool::issue_append_lsn(
@@ -1719,7 +1727,11 @@ pub(crate) async fn handle_connection_sharded_monoio<
                         if repl_active || ctx.aof_pool.is_some() {
                             let serialized = aof::serialize_command(&frame);
                             let lsn = if repl_active {
-                                ft::record_local_write(ctx, serialized.clone());
+                                ft::record_local_write_db(
+                                    ctx,
+                                    conn.selected_db,
+                                    serialized.clone(),
+                                );
                                 0
                             } else {
                                 aof::AofWriterPool::issue_append_lsn(

src/server/conn/handler_monoio/txn.rs

@@ -355,6 +355,11 @@ pub(super) async fn try_handle_temporal_invalidate(
                                 entity_id,
                                 wall_ms,
                             );
+                            // Db-agnostic record (the replica routes it before
+                            // db resolution) — the primitive leaves the
+                            // stream's SELECT context untouched, which stays
+                            // truthful: this record neither needs nor changes
+                            // the db context.
                             super::ft::record_local_write(ctx, Bytes::from(record));
                         }
                         for record in wal_records {

src/server/conn/handler_monoio/write.rs

@@ -766,7 +766,7 @@ pub(super) async fn try_handle_multi_exec(
                 // Each drain iteration is just a try_send per replica, so the
                 // burst is cheap; revisit only if EXEC bodies grow unbounded.
                 for bytes in &aof_entries {
-                    super::ft::record_local_write(ctx, bytes.clone());
+                    super::ft::record_local_write_db(ctx, conn.selected_db, bytes.clone());
                 }
             }
             // DURABILITY: append every successful write in the body to THIS
@@ -984,7 +984,7 @@ pub(super) async fn try_handle_graph_command(
         // keeps mutation + replication record atomic w.r.t. the inline PSYNC
         // task's snapshot capture on this thread.
         for record in &wal_records {
-            super::ft::record_local_write(ctx, record.clone());
+            super::ft::record_local_write_db(ctx, conn.selected_db, record.clone());
         }
     }
     for record in wal_records {

src/server/conn/handler_sharded/dispatch.rs

@@ -288,6 +288,7 @@ pub(super) fn try_handle_replicaof(
                         num_shards: ctx.num_shards,
                         persistence_dir: None,
                         listening_port: 0,
+                        stream_db: std::sync::atomic::AtomicUsize::new(0),
                     };
                     tokio::task::spawn_local(crate::replication::replica::run_replica_task(cfg));
                 }
@@ -573,6 +574,11 @@ pub(super) fn try_enforce_readonly(
         return false;
     }
     if metadata::is_write(cmd) {
+        // SELECT is flagged W but only mutates CONNECTION state — Redis
+        // serves it on replicas (task #23, see handler_monoio::dispatch).
+        if cmd.eq_ignore_ascii_case(b"SELECT") {
+            return false;
+        }
         // GRAPH.QUERY is blanket-W (Cypher CAN write); serve read-only
         // MATCH/RETURN on replicas. The classifier never false-negatives
         // for a write query — see handler_monoio::dispatch.

src/server/conn/handler_single.rs

@@ -907,7 +907,11 @@ pub async fn handle_connection(
                                         && !crate::command::graph::is_cypher_write_query(cmd_args);
                                     #[cfg(not(feature = "graph"))]
                                     let graph_ro = false;
-                                    if metadata::is_write(cmd) && !graph_ro {
+                                    // SELECT is flagged W but only mutates
+                                    // CONNECTION state — serve it on replicas
+                                    // (task #23).
+                                    let conn_only = cmd.eq_ignore_ascii_case(b"SELECT");
+                                    if metadata::is_write(cmd) && !graph_ro && !conn_only {
                                         responses.push(Frame::Error(Bytes::from_static(
                                             b"READONLY You can't write against a read only replica.",
                                         )));