diff --git a/CHANGELOG.md b/CHANGELOG.md index e0235570f..2ef608730 100644 --- a/CHANGELOG.md +++ b/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 ` 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 diff --git a/src/replication/master.rs b/src/replication/master.rs index 41025bb16..fd84716fe 100644 --- a/src/replication/master.rs +++ b/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 ` 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| { diff --git a/src/replication/replica.rs b/src/replication/replica.rs index 157bfa3ce..329b2e013 100644 --- a/src/replication/replica.rs +++ b/src/replication/replica.rs @@ -26,6 +26,16 @@ pub struct ReplicaTaskConfig { pub num_shards: usize, pub persistence_dir: Option, 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" diff --git a/src/replication/state.rs b/src/replication/state.rs index 6135e3ef4..3ec316362 100644 --- a/src/replication/state.rs +++ b/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 ` 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, } 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(), } } diff --git a/src/server/conn/handler_monoio/dispatch.rs b/src/server/conn/handler_monoio/dispatch.rs index 3a0071928..4729f29c0 100644 --- a/src/server/conn/handler_monoio/dispatch.rs +++ b/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 diff --git a/src/server/conn/handler_monoio/ft.rs b/src/server/conn/handler_monoio/ft.rs index 6adeba187..524a88519 100644 --- a/src/server/conn/handler_monoio/ft.rs +++ b/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 ` 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 ` for the replication stream. +fn serialize_select(db: usize) -> Vec { + 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() { diff --git a/src/server/conn/handler_monoio/mod.rs b/src/server/conn/handler_monoio/mod.rs index c1036f04c..110f9d35e 100644 --- a/src/server/conn/handler_monoio/mod.rs +++ b/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( diff --git a/src/server/conn/handler_monoio/txn.rs b/src/server/conn/handler_monoio/txn.rs index fbfb71cd2..0d7fc4ff1 100644 --- a/src/server/conn/handler_monoio/txn.rs +++ b/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 { diff --git a/src/server/conn/handler_monoio/write.rs b/src/server/conn/handler_monoio/write.rs index d33e66c71..d7ef0e4e9 100644 --- a/src/server/conn/handler_monoio/write.rs +++ b/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 { diff --git a/src/server/conn/handler_sharded/dispatch.rs b/src/server/conn/handler_sharded/dispatch.rs index 41c5fbe31..7cf6545c9 100644 --- a/src/server/conn/handler_sharded/dispatch.rs +++ b/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. diff --git a/src/server/conn/handler_single.rs b/src/server/conn/handler_single.rs index 649b062bb..fd8c6703f 100644 --- a/src/server/conn/handler_single.rs +++ b/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.", ))); diff --git a/tests/replication_streaming.rs b/tests/replication_streaming.rs index d6292ffe6..a161a7aad 100644 --- a/tests/replication_streaming.rs +++ b/tests/replication_streaming.rs @@ -688,3 +688,78 @@ fn replica_applies_multi_exec_bodies() { "MULTI/EXEC INCRs did not replicate exactly once (None=lost, 4=double-applied)" ); } + +/// HIGH-2 (task #22 + #23): multi-db writes must land in the SAME logical db +/// on the replica. The master prepends `SELECT ` to the stream whenever +/// the writing connection's db differs from the stream's current db context; +/// the replica-side drain already binds subsequent commands to it. Also +/// exercises #23: reading db 2 on the replica requires SELECT on a replica +/// connection (previously rejected with READONLY — SELECT is flagged W). +#[test] +#[ignore] +fn replica_applies_multi_db_stream() { + let master_dir = tempfile::tempdir().unwrap(); + let replica_dir = tempfile::tempdir().unwrap(); + + let master_addr = "127.0.0.1:16734"; + let replica_addr = "127.0.0.1:16735"; + + let _master = Killer(start_moon(16734, master_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(master_addr, "PING") + .starts_with("+PONG")), + "master never became ready" + ); + let _replica = Killer(start_moon(16735, replica_dir.path().to_str().unwrap())); + assert!( + wait_until(Duration::from_secs(5), || send_cmd(replica_addr, "PING") + .starts_with("+PONG")), + "replica never became ready" + ); + send_cmd(replica_addr, "REPLICAOF 127.0.0.1 16734"); + + // Prove the stream is up in db 0 first. + send_cmd(master_addr, "SET d0key d0val"); + assert!( + wait_until(Duration::from_secs(10), || { + get(replica_addr, "d0key").as_deref() == Some("d0val") + }), + "db-0 live stream not flowing" + ); + + // #23: SELECT must be served by the read-only replica. + let sel = send_cmd(replica_addr, "SELECT 2"); + assert!( + sel.starts_with("+OK"), + "SELECT on a read-only replica must succeed, got: {sel}" + ); + + // Write in db 2, then hop back to db 0 — the stream must carry both + // context switches. + let r = send_seq(master_addr, &["SELECT 2", "SET d2key d2val"]); + assert!(r.starts_with("+OK"), "master db-2 SET failed: {r}"); + send_cmd(master_addr, "SET d0post after"); + + assert!( + wait_until(Duration::from_secs(10), || { + get(replica_addr, "d0post").as_deref() == Some("after") + }), + "post-hop db-0 sentinel never replicated" + ); + assert_eq!( + get_in_db(replica_addr, 2, "d2key").as_deref(), + Some("d2val"), + "db-2 write did not land in db 2 on the replica" + ); + assert_eq!( + get_in_db(replica_addr, 0, "d2key"), + None, + "db-2 write leaked into db 0 on the replica (SELECT not streamed)" + ); + // The db-0 sentinel must not have leaked into db 2 either. + assert_eq!( + get_in_db(replica_addr, 2, "d0post"), + None, + "db-0 write leaked into db 2 on the replica" + ); +}