diff --git a/crates/lance-context-wal/Cargo.toml b/crates/lance-context-wal/Cargo.toml index 3867197..62d614f 100644 --- a/crates/lance-context-wal/Cargo.toml +++ b/crates/lance-context-wal/Cargo.toml @@ -8,7 +8,7 @@ description = "Independent Rust WAL experiments owned by lance-context" [features] client-journal = ["dep:anyhow", "dep:raft-engine", "dep:bytes", "dep:serde", "dep:serde_json", "dep:tokio", "tokio/sync"] -replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog"] +replication = ["raft-log", "record", "contract", "dep:bincode", "dep:slog", "uuid/v4"] raft-log = ["placement", "dep:raft", "dep:raft-engine", "dep:raft-proto", "dep:protobuf"] vectored-io = ["dep:bytes", "dep:tokio"] runner = ["dep:anyhow", "dep:clap", "dep:libc", "dep:serde", "dep:serde_json", "dep:sha2"] diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index b9b057a..fb65272 100644 --- a/crates/lance-context-wal/README.md +++ b/crates/lance-context-wal/README.md @@ -172,6 +172,10 @@ The replication layer also provides [durable publisher bindings and epochs](../. through its catalog group's normal consensus path. This is an incremental publication prerequisite, not a production Lance writer or a manifest lease. +[Quorum application queries](../../docs/design/wal-quorum-queries.md) capture bounded +catalog, producer-tail and accepted-page views after a safe read-index barrier. +They share pending capacity with proposals and do not fetch bulk payloads. + The optional `raft-log` feature exposes `raft_log::PhysicalLog` and its cloneable `GroupLog` handles. One raft-engine directory hosts all groups assigned to a node by the existing immutable `placement::Layout`. Opening binds the full layout, diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 04b8eca..da9fc73 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -15,6 +15,7 @@ pub mod checkpoint; mod images; pub mod publication; +pub mod query; mod state; #[cfg(test)] mod tests; @@ -124,6 +125,7 @@ pub struct Completion { pub struct Events { pub messages: Vec, pub completions: Vec, + pub reads: Vec, } #[derive(Clone, Debug, Serialize, Deserialize)] @@ -135,6 +137,7 @@ pub struct Status { pub accepted_records: usize, pub producers: usize, pub pending: usize, + pub pending_reads: usize, } /// A bounded local view captured together with its next cursor. Preserve the @@ -152,6 +155,7 @@ struct Group { state: State, // Index AND term prevents suffix replacement from completing another input. pending: BTreeMap<(u64, u64), (u64, BatchMeta)>, + reads: query::Tracker, image: Option>, incoming: Option, } @@ -232,6 +236,7 @@ impl Node { heartbeat_tick: 2, check_quorum: true, pre_vote: true, + read_only_option: raft::ReadOnlyOption::Safe, max_size_per_msg: 1024 * 1024, max_inflight_msgs: 16, max_uncommitted_size: limits.max_uncommitted_bytes, @@ -245,6 +250,7 @@ impl Node { raw, state, pending: BTreeMap::new(), + reads: query::Tracker::default(), image, incoming: None, }, @@ -300,9 +306,45 @@ impl Node { accepted_records: g.state.accepted.len(), producers: g.state.producers.len(), pending: g.pending.len(), + pending_reads: g.reads.len(), }) } + fn request_pending(&self, request: u64) -> bool { + self.groups + .values() + .any(|g| g.pending.values().any(|p| p.0 == request) || g.reads.contains(request)) + } + + /// Queue a bounded safe read-index request at this group's leader. Only a + /// successful poll can return the captured application view. This neither + /// forwards follower reads nor treats the leader's local state as a lease. + /// Until an entry in the leader's current term commits, admission fails + /// without reserving capacity or the token; retry after driving replication. + pub fn query(&mut self, group: u64, request: u64, query: query::Query) -> Result<()> { + self.healthy()?; + ensure!( + !self.request_pending(request), + "request token already pending" + ); + query.validate(&self.layout, group)?; + let g = self.groups.get_mut(&group).context("unhosted group")?; + ensure!(g.raw.raft.state == StateRole::Leader, "not leader"); + // raft-rs 0.7 silently drops ReadIndex before this condition holds. + // Reject before tracking so a new leader cannot strand accepted reads. + ensure!( + g.raw.raft.commit_to_current_term(), + "leader has not committed a current-term entry; retry after replication" + ); + ensure!( + g.pending.len() + g.reads.len() < self.limits.max_pending as usize, + "pending request capacity" + ); + let context = g.reads.enqueue(request, g.raw.raft.term, query)?; + g.raw.read_index(context); + Ok(()) + } + /// Queue a binding/epoch CAS in the one catalog group. Only a committed /// Publisher completion from poll grants authority; no local cache shortcut. pub fn acquire_publisher(&mut self, request: u64, acquire: publication::Acquire) -> Result<()> { @@ -328,10 +370,7 @@ impl Node { pub fn propose(&mut self, group: u64, request: u64, batch: Batch) -> Result<()> { self.healthy()?; ensure!( - !self - .groups - .values() - .any(|g| g.pending.values().any(|p| p.0 == request)), + !self.request_pending(request), "request token already pending" ); state::validate_meta( @@ -345,7 +384,7 @@ impl Node { let g = self.groups.get_mut(&group).context("unhosted group")?; ensure!(g.raw.raft.state == StateRole::Leader, "not leader"); ensure!( - g.pending.len() < self.limits.max_pending as usize, + g.pending.len() + g.reads.len() < self.limits.max_pending as usize, "pending proposal capacity" ); let index = g.raw.raft.raft_log.last_index() + 1; @@ -452,6 +491,9 @@ impl Node { } outbound.extend(ready.take_messages()); outbound.extend(ready.take_persisted_messages()); + for read in ready.take_read_states() { + group.reads.confirm(&read.request_ctx, read.index)?; + } apply( group, *id, @@ -494,6 +536,18 @@ impl Node { outcome: Outcome::Unknown, }); } + events.reads.extend(group.reads.finish( + query::Cut { + cluster: self.layout.cluster, + layout: self.layout_hash, + group: *id, + term: group.raw.raft.term, + quorum_index: 0, + applied: group.state.applied, + }, + &group.state, + group.raw.raft.state == StateRole::Leader, + )?); events .messages .extend(outbound.into_iter().map(|message| Packet { @@ -512,26 +566,8 @@ impl Node { /// read or a producer ACK and cannot be used to shortcut retry completion. pub fn accepted(&self, group: u64, after: u64, limit: usize) -> Result { self.healthy()?; - ensure!((1..=128).contains(&limit), "invalid accepted page limit"); let state = &self.groups.get(&group).context("unhosted group")?.state; - ensure!( - after <= state.applied, - "cursor exceeds local applied position" - ); - let mut remaining = state - .accepted - .range((std::ops::Bound::Excluded(after), std::ops::Bound::Unbounded)); - let entries: Vec<_> = remaining - .by_ref() - .take(limit) - .map(|(index, meta)| (*index, meta.clone())) - .collect(); - let through = if remaining.next().is_some() { - entries.last().expect("positive page limit").0 - } else { - state.applied - }; - Ok(Page { entries, through }) + state.page(after, limit) } pub fn read_accepted(&self, group: u64, index: u64) -> Result> { diff --git a/crates/lance-context-wal/src/replication/query.rs b/crates/lance-context-wal/src/replication/query.rs new file mode 100644 index 0000000..e1a5cb1 --- /dev/null +++ b/crates/lance-context-wal/src/replication/query.rs @@ -0,0 +1,205 @@ +//! Bounded read-index queries with an application view captured at completion. +//! A result describes committed state during the request; it is not a lease, +//! authorization token or permission to write an external Lance manifest. +use super::*; + +#[cfg(test)] +mod tests; + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Query { + Publisher { + table: Uuid, + }, + Accepted { + after: u64, + limit: usize, + }, + Producer { + table: Uuid, + bucket: u32, + producer: Uuid, + }, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Value { + Publisher(Option>), + Accepted(Page), + Producer(Option<(BatchMeta, u64)>), + CursorAhead { applied: u64 }, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Cut { + pub cluster: Uuid, + pub layout: [u8; 32], + pub group: u64, + pub term: u64, + pub quorum_index: u64, + /// The query and its cursor are captured together after this position applies. + pub applied: u64, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub enum Outcome { + Observed { + cut: Cut, + value: Value, + }, + /// No read value was released before leadership changed. Retry at a leader. + Unknown, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Completion { + pub group: u64, + pub request: u64, + pub outcome: Outcome, +} + +impl Query { + pub(super) fn validate(&self, layout: &Layout, group: u64) -> Result<()> { + match self { + Self::Publisher { table } => ensure!( + !table.is_nil() && group == publication::control_group(layout), + "invalid publisher query scope" + ), + Self::Accepted { after, limit } => ensure!( + *after < u64::MAX && (1..=128).contains(limit), + "invalid accepted query bounds" + ), + Self::Producer { + table, + bucket, + producer, + } => ensure!( + !table.is_nil() && !producer.is_nil() && layout.group_for_bucket(*bucket)? == group, + "invalid producer query scope" + ), + } + Ok(()) + } + + fn evaluate(&self, state: &State) -> Result { + Ok(match self { + Self::Publisher { table } => { + Value::Publisher(state.catalog.tables.get(table).cloned().map(Box::new)) + } + Self::Accepted { after, .. } if *after > state.applied => Value::CursorAhead { + applied: state.applied, + }, + Self::Accepted { after, limit } => Value::Accepted(state.page(*after, *limit)?), + Self::Producer { + table, + bucket, + producer, + } => Value::Producer(state.producers.get(&(*table, *bucket, *producer)).cloned()), + }) + } +} + +struct Pending { + request: u64, + term: u64, + query: Query, + confirmed: Option, +} + +pub(super) struct Tracker { + incarnation: Uuid, + next: u64, + pending: BTreeMap<[u8; 32], Pending>, +} + +impl Default for Tracker { + fn default() -> Self { + Self { + incarnation: Uuid::new_v4(), + next: 0, + pending: BTreeMap::new(), + } + } +} + +impl Tracker { + pub fn len(&self) -> usize { + self.pending.len() + } + pub fn contains(&self, request: u64) -> bool { + self.pending.values().any(|p| p.request == request) + } + pub fn enqueue(&mut self, request: u64, term: u64, query: Query) -> Result> { + self.next = self.next.checked_add(1).context("read context exhausted")?; + let mut context = [0; 32]; + context[..8].copy_from_slice(b"LCREAD01"); + context[8..24].copy_from_slice(self.incarnation.as_bytes()); + context[24..].copy_from_slice(&self.next.to_le_bytes()); + ensure!( + self.pending + .insert( + context, + Pending { + request, + term, + query, + confirmed: None + } + ) + .is_none(), + "reused read context" + ); + Ok(context.to_vec()) + } + pub fn confirm(&mut self, context: &[u8], index: u64) -> Result<()> { + let Ok(key) = <[u8; 32]>::try_from(context) else { + return Ok(()); + }; + let Some(pending) = self.pending.get_mut(&key) else { + return Ok(()); + }; + ensure!( + index > 0 && index < u64::MAX, + "invalid confirmed read index" + ); + ensure!( + pending.confirmed.is_none_or(|old| old == index), + "read index changed" + ); + pending.confirmed = Some(index); + Ok(()) + } + pub fn finish(&mut self, cut: Cut, state: &State, leader: bool) -> Result> { + let done: Vec<_> = self + .pending + .iter() + .filter_map(|(key, p)| { + (!leader + || p.term != cut.term + || p.confirmed.is_some_and(|index| index <= state.applied)) + .then_some(*key) + }) + .collect(); + let mut results = Vec::with_capacity(done.len()); + for key in done { + let p = self.pending.remove(&key).expect("selected pending read"); + let outcome = if !leader || p.term != cut.term { + Outcome::Unknown + } else { + Outcome::Observed { + cut: Cut { + quorum_index: p.confirmed.expect("confirmed completed read"), + ..cut.clone() + }, + value: p.query.evaluate(state)?, + } + }; + results.push(Completion { + group: cut.group, + request: p.request, + outcome, + }); + } + Ok(results) + } +} diff --git a/crates/lance-context-wal/src/replication/query/tests.rs b/crates/lance-context-wal/src/replication/query/tests.rs new file mode 100644 index 0000000..0b9078e --- /dev/null +++ b/crates/lance-context-wal/src/replication/query/tests.rs @@ -0,0 +1,262 @@ +use super::*; + +fn nodes(limits: Limits) -> Result<(tempfile::TempDir, Vec)> { + let root = tempfile::tempdir()?; + let nodes = (1..=3) + .map(|id| { + Node::open( + &root.path().join(id.to_string()), + super::super::tests::layout(), + id, + limits.clone(), + ) + }) + .collect::>>()?; + Ok((root, nodes)) +} +fn pump(nodes: &mut [Node]) -> Result { + let mut result = Events::default(); + for _ in 0..100 { + let mut messages = Vec::new(); + for node in &mut *nodes { + let events = node.poll()?; + result.completions.extend(events.completions); + result.reads.extend(events.reads); + messages.extend(events.messages); + } + if messages.is_empty() { + return Ok(result); + } + for packet in messages { + nodes[packet.message.to as usize - 1].step(packet)?; + } + } + anyhow::bail!("query message pump did not settle") +} + +#[test] +fn new_leader_rejects_reads_until_its_first_current_term_entry_commits() -> Result<()> { + let (_root, mut nodes) = nodes(Limits { + max_pending: 1, + ..Default::default() + })?; + nodes[0].campaign(10)?; + let mut held_appends = Vec::new(); + for round in 0..100 { + let mut messages = Vec::new(); + for node in &mut nodes { + let events = node.poll()?; + assert!(events.reads.is_empty() && events.completions.is_empty()); + messages.extend(events.messages); + } + if messages.is_empty() { + break; + } + assert!(round < 99, "election did not settle"); + for packet in messages { + if packet.message.get_msg_type() == MessageType::MsgAppend { + held_appends.push(packet); + } else { + nodes[packet.message.to as usize - 1].step(packet)?; + } + } + } + let group = &nodes[0].groups[&10]; + assert_eq!(group.raw.raft.state, StateRole::Leader); + assert!(!held_appends.is_empty()); + assert!(!group.raw.raft.commit_to_current_term()); + assert!(group.raw.store().last_index()? > group.state.applied); + + // raft-rs silently drops ReadIndex here. Do not consume the one available + // request slot or token for a request it cannot yet process. + let query = Query::Accepted { after: 0, limit: 1 }; + assert!(nodes[0].query(10, 7, query.clone()).is_err()); + assert_eq!(nodes[0].status(10)?.pending_reads, 0); + assert!(nodes[0].poll()?.reads.is_empty()); + for packet in held_appends { + nodes[packet.message.to as usize - 1].step(packet)?; + } + assert!(pump(&mut nodes)?.reads.is_empty()); + assert!(nodes[0].groups[&10].raw.raft.commit_to_current_term()); + + // Restoring replication makes the same token admissible without a timeout, + // capacity reset or leader change. The retry completes through a quorum. + nodes[0].query(10, 7, query)?; + let reads = pump(&mut nodes)?.reads; + assert_eq!(reads.len(), 1); + assert_eq!(reads[0].request, 7); + assert!(matches!(reads[0].outcome, Outcome::Observed { .. })); + assert_eq!(nodes[0].status(10)?.pending_reads, 0); + Ok(()) +} + +#[test] +fn reads_share_capacity_and_tokens_with_appends_and_never_use_a_local_lease() -> Result<()> { + let (_root, mut nodes) = nodes(Limits { + max_pending: 1, + ..Default::default() + })?; + nodes[0].campaign(10)?; + nodes[0].campaign(20)?; + pump(&mut nodes)?; + let log_positions = nodes + .iter() + .map(|node| node.groups[&10].raw.store().last_index()) + .collect::, _>>()?; + let query = Query::Accepted { after: 0, limit: 1 }; + nodes[0].query(10, 1, query.clone())?; + assert_eq!(nodes[0].status(10)?.pending_reads, 1); + assert!(nodes[0] + .propose(10, 2, super::super::tests::batch(1)) + .is_err()); + assert!(nodes[0].query(20, 1, query.clone()).is_err()); + assert!(nodes[0].query(10, 2, query.clone()).is_err()); + assert!(nodes[1].query(10, 1, query.clone()).is_err()); + let events = nodes[0].poll()?; + assert!( + events.reads.is_empty(), + "local leader released a read without quorum" + ); + for packet in events.messages { + nodes[packet.message.to as usize - 1].step(packet)?; + } + let events = pump(&mut nodes)?; + assert_eq!(events.reads.len(), 1); + assert!( + matches!(&events.reads[0].outcome, Outcome::Observed { cut, value: Value::Accepted(page) } if cut.quorum_index > 0 && cut.applied >= cut.quorum_index && page.through == cut.applied) + ); + for (node, expected) in nodes.iter().zip(log_positions) { + assert_eq!(node.groups[&10].raw.store().last_index()?, expected); + assert_eq!(node.groups[&10].raw.raft.raft_log.last_index(), expected); + } + nodes[0].propose(10, 2, super::super::tests::batch(1))?; + assert!(nodes[0].query(10, 3, query.clone()).is_err()); + pump(&mut nodes)?; + nodes[0].query( + 10, + 3, + Query::Accepted { + after: u64::MAX - 1, + limit: 1, + }, + )?; + assert!(matches!( + &pump(&mut nodes)?.reads[0].outcome, + Outcome::Observed { + value: Value::CursorAhead { .. }, + .. + } + )); + assert_eq!(nodes[0].status(10)?.pending_reads, 0); + assert!(nodes[0] + .query( + 20, + 4, + Query::Publisher { + table: Uuid::from_u128(50) + } + ) + .is_err()); + assert!(nodes[0] + .query( + 10, + 4, + Query::Producer { + table: Uuid::from_u128(50), + bucket: 1, + producer: Uuid::from_u128(51) + } + ) + .is_err()); + assert!(nodes[0] + .query( + 10, + 4, + Query::Accepted { + after: 0, + limit: 129 + } + ) + .is_err()); + Ok(()) +} + +#[test] +fn read_contexts_do_not_cross_request_reuse_reopen_or_unapplied_cuts() -> Result<()> { + let mut tracker = Tracker::default(); + let query = Query::Accepted { after: 0, limit: 1 }; + let context = tracker.enqueue(10, 2, query.clone())?; + let mut state = State { + applied: 4, + ..Default::default() + }; + let mut cut = Cut { + cluster: Uuid::from_u128(1), + layout: [2; 32], + group: 10, + term: 2, + quorum_index: 0, + applied: 4, + }; + tracker.confirm(&context, 5)?; + assert!(tracker.finish(cut.clone(), &state, true)?.is_empty()); + state.applied = 5; + cut.applied = 5; + assert_eq!(tracker.finish(cut.clone(), &state, true)?.len(), 1); + let next = tracker.enqueue(10, 2, query.clone())?; + assert_ne!(next, context); + tracker.confirm(&context, 5)?; + assert!(tracker.finish(cut.clone(), &state, true)?.is_empty()); + cut.term = 3; + assert_eq!( + tracker.finish(cut.clone(), &state, true)?[0].outcome, + Outcome::Unknown + ); + let mut reopened = Tracker::default(); + assert_ne!(reopened.enqueue(10, 3, query)?, context); + reopened.confirm(&context, 5)?; + assert!(reopened.finish(cut.clone(), &state, true)?.is_empty()); + assert_eq!( + reopened.finish(cut, &state, false)?[0].outcome, + Outcome::Unknown + ); + Ok(()) +} + +#[test] +fn shared_physical_sync_failure_fences_already_confirmed_reads() -> Result<()> { + let (_root, mut nodes) = nodes(Limits::default())?; + nodes[0].campaign(10)?; + nodes[0].campaign(20)?; + pump(&mut nodes)?; + nodes[0].query(10, 1, Query::Accepted { after: 0, limit: 1 })?; + let events = nodes[0].poll()?; + assert!(events.reads.is_empty()); + for packet in events.messages { + nodes[packet.message.to as usize - 1].step(packet)?; + } + let mut responses = Vec::new(); + for node in &mut nodes[1..] { + responses.extend(node.poll()?.messages); + } + for packet in responses { + nodes[packet.message.to as usize - 1].step(packet)?; + } + assert!( + !nodes[0].groups[&10].raw.raft.read_states.is_empty(), + "quorum read was not confirmed before sibling failure" + ); + let mut batch = super::super::tests::batch(1); + batch.meta.bucket = 1; + nodes[0].propose(20, 2, batch)?; + nodes[0].groups[&20].raw.store().fail_next_sync_for_test(); + assert!( + nodes[0].poll().is_err(), + "sibling persistence error released read events" + ); + assert!(nodes[0].status(10).is_err()); + assert!(nodes[0] + .query(10, 3, Query::Accepted { after: 0, limit: 1 }) + .is_err()); + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication/state.rs b/crates/lance-context-wal/src/replication/state.rs index abf95f9..08140dd 100644 --- a/crates/lance-context-wal/src/replication/state.rs +++ b/crates/lance-context-wal/src/replication/state.rs @@ -85,6 +85,28 @@ pub(super) struct State { pub producers: BTreeMap<(Uuid, u32, Uuid), (BatchMeta, u64)>, } impl State { + pub fn page(&self, after: u64, limit: usize) -> Result { + ensure!((1..=128).contains(&limit), "invalid accepted page limit"); + ensure!( + after <= self.applied, + "cursor exceeds local applied position" + ); + let mut remaining = self + .accepted + .range((std::ops::Bound::Excluded(after), std::ops::Bound::Unbounded)); + let entries: Vec<_> = remaining + .by_ref() + .take(limit) + .map(|(index, meta)| (*index, meta.clone())) + .collect(); + let through = if remaining.next().is_some() { + entries.last().expect("positive page limit").0 + } else { + self.applied + }; + Ok(Page { entries, through }) + } + /// The image retains every accepted record. Producer tails must exactly /// match those records, including sequence continuity within each lane. /// Retirement will need a new format with independently retained tails. diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index 6c91232..c6a28df 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -11,6 +11,7 @@ use lance_context_wal::{ publication::{ Acquire, Destination, Rejection as PublisherRejection, Reply as PublisherReply, Table, }, + query::{Completion as ReadCompletion, Outcome as ReadOutcome, Query, Value}, Applied, Completion, Events, Limits, Node, Outcome, Packet, Page, PreparedCheckpoint, SnapshotReceiver, Status, }, @@ -59,6 +60,7 @@ impl WirePacket { } #[derive(Serialize, Deserialize)] enum Request { + Query(u64, u64, Query), Acquire(u64, Acquire), Publisher(Uuid), Campaign(u64), @@ -80,7 +82,7 @@ enum Request { #[derive(Serialize, Deserialize)] enum Response { Publisher(Option), - Events(Vec, Vec), + Events(Vec, Vec, Vec), Status(Status), Record(Option), Page(Page), @@ -123,6 +125,7 @@ fn part(payload: bool) -> Part { fn run(worker: &mut Worker, request: Request) -> Result { let node = &mut worker.node; match request { + Request::Query(group, token, query) => node.query(group, token, query)?, Request::Acquire(token, acquire) => node.acquire_publisher(token, acquire)?, Request::Publisher(table) => { return Ok(Response::Publisher(node.publisher(table)?.cloned())) @@ -198,6 +201,7 @@ fn run(worker: &mut Worker, request: Request) -> Result { let Events { messages, completions, + reads, } = node.poll()?; for packet in &messages { if packet.message.get_msg_type() == raft_proto::eraftpb::MessageType::MsgSnapshot { @@ -214,6 +218,7 @@ fn run(worker: &mut Worker, request: Request) -> Result { .map(WirePacket::encode) .collect::>()?, completions, + reads, )) } @@ -324,6 +329,7 @@ struct Cluster { incarnations: BTreeMap, queue: VecDeque, completions: Vec, + reads: Vec, transfers: u64, interrupt_next_snapshot: bool, } @@ -348,6 +354,7 @@ impl Cluster { incarnations: BTreeMap::new(), queue: VecDeque::new(), completions: Vec::new(), + reads: Vec::new(), transfers: 0, interrupt_next_snapshot: false, }; @@ -368,7 +375,7 @@ impl Cluster { Ok(()) } fn command(&mut self, id: u64, request: Request) -> Result<()> { - let Response::Events(packets, completions) = self + let Response::Events(packets, completions, reads) = self .peers .get_mut(&id) .context("missing peer")? @@ -380,6 +387,7 @@ impl Cluster { self.queue.push_back(packet.decode()?); } self.completions.extend(completions); + self.reads.extend(reads); Ok(()) } fn pump(&mut self, allowed: &[u64]) -> Result<()> { @@ -487,6 +495,12 @@ impl Cluster { .position(|c| c.request == token) .map(|position| self.completions.remove(position).outcome) } + fn read_outcome(&mut self, token: u64) -> Option { + self.reads + .iter() + .position(|c| c.request == token) + .map(|position| self.reads.remove(position).outcome) + } fn elect(&mut self, id: u64, group: u64, allowed: &[u64]) -> Result { self.command(id, Request::Campaign(group))?; self.pump(allowed)?; @@ -857,6 +871,28 @@ fn publisher_epochs_require_quorum_and_survive_snapshot_transfer_and_restart() - assert_eq!(table.destination, acquire.destination); assert_eq!(table.ingest, acquire.ingest); + c.command( + 2, + Request::Query( + 10, + 8000, + Query::Publisher { + table: acquire.table, + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + cut, + value: Value::Publisher(Some(captured)), + }) = c.read_outcome(8000) + else { + anyhow::bail!("publisher quorum view missing"); + }; + assert_eq!(*captured, table); + assert!(cut.quorum_index >= table.authority.grant_index); + assert!(cut.applied >= cut.quorum_index); + c.command(2, Request::Acquire(703, acquire.clone()))?; c.pump(&[2])?; assert!( @@ -908,5 +944,134 @@ fn publisher_epochs_require_quorum_and_survive_snapshot_transfer_and_restart() - Some(latest), "restart changed the committed grant index" ); + assert_eq!( + captured.authority.epoch, 2, + "later takeover mutated a captured read" + ); + Ok(()) +} + +#[test] +fn quorum_queries_capture_committed_tails_without_appends_and_reject_isolated_reads() -> Result<()> +{ + let mut c = Cluster::new()?; + c.command(1, Request::Campaign(10))?; + c.pump(&[1, 2, 3])?; + let first = input(0, 1, 70); + c.command(1, Request::Propose(10, 900, first.clone()))?; + c.pump(&[1, 2, 3])?; + let Some(Outcome::Committed(Applied::Accepted { index: first_index })) = c.outcome(900) else { + anyhow::bail!("first append did not commit"); + }; + let query = Query::Producer { + table: first.meta.stream, + bucket: 0, + producer: first.meta.producer, + }; + let Response::Status(before) = c.peers.get_mut(&1).unwrap().call(Request::Status(10))? else { + anyhow::bail!("missing initial status"); + }; + c.command(1, Request::Query(10, 901, query.clone()))?; + assert!( + c.read_outcome(901).is_none(), + "local tail answered without quorum" + ); + c.pump(&[1, 2])?; + let Some(ReadOutcome::Observed { + cut, + value: Value::Producer(Some((meta, index))), + }) = c.read_outcome(901) + else { + anyhow::bail!("missing quorum producer tail"); + }; + assert_eq!(meta, first.meta); + assert_eq!(index, first_index); + assert!(cut.quorum_index >= index && cut.applied >= cut.quorum_index); + c.command( + 1, + Request::Query( + 10, + 902, + Query::Accepted { + after: 0, + limit: 128, + }, + ), + )?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + cut, + value: Value::Accepted(page), + }) = c.read_outcome(902) + else { + anyhow::bail!("missing quorum accepted page"); + }; + assert_eq!(page.entries, vec![(first_index, first.meta.clone())]); + assert_eq!(page.through, cut.applied); + let Response::Status(after) = c.peers.get_mut(&1).unwrap().call(Request::Status(10))? else { + anyhow::bail!("missing post-read status"); + }; + assert_eq!( + after.committed, before.committed, + "read-index queries appended log entries" + ); + + c.command(1, Request::Query(10, 903, query.clone()))?; + c.pump(&[1])?; + assert!(c.read_outcome(903).is_none()); + let leader = c.elect(2, 10, &[2, 3])?; + let second = input(0, 2, 71); + c.command(leader, Request::Propose(10, 904, second.clone()))?; + c.pump(&[2, 3])?; + let Some(Outcome::Committed(Applied::Accepted { + index: second_index, + })) = c.outcome(904) + else { + anyhow::bail!("new leader did not append"); + }; + for _ in 0..25 { + c.command(1, Request::Tick)?; + c.pump(&[1])?; + } + assert_eq!(c.read_outcome(903), Some(ReadOutcome::Unknown)); + for _ in 0..4 { + c.command(leader, Request::Tick)?; + c.pump(&[1, 2, 3])?; + } + c.command(leader, Request::Query(10, 905, query.clone()))?; + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + value: Value::Producer(Some((meta, index))), + .. + }) = c.read_outcome(905) + else { + anyhow::bail!("missing new leader's tail"); + }; + assert_eq!(meta, second.meta); + assert_eq!(index, second_index); + c.queue.clear(); + c.peers.clear(); + for id in 1..=3 { + c.start(id)?; + } + let leader = c.elect(1, 10, &[1, 2, 3])?; + c.command(leader, Request::Query(10, 901, query))?; + assert!( + c.read_outcome(901).is_none(), + "restart reused an old read result" + ); + c.pump(&[1, 2, 3])?; + let Some(ReadOutcome::Observed { + value: Value::Producer(Some((meta, index))), + .. + }) = c.read_outcome(901) + else { + anyhow::bail!("missing restarted tail"); + }; + assert_eq!(meta, second.meta); + assert_eq!(index, second_index); + for peer in c.peers.values_mut() { + verify_records(peer, 10, &[first.clone(), second.clone()])?; + } Ok(()) } diff --git a/docs/design/wal-quorum-queries.md b/docs/design/wal-quorum-queries.md new file mode 100644 index 0000000..eba7781 --- /dev/null +++ b/docs/design/wal-quorum-queries.md @@ -0,0 +1,46 @@ +# Quorum application queries + +Refs #378. + +Cross-group publication needs a committed view of catalog plans and source state. +A local `publisher`, `accepted` or `status` observation cannot establish that the +responding node still participates in a majority. The optional replication Node +now queues bounded `Query` requests using raft-rs safe read-index processing. + +Only the group's current leader accepts a query. It must obtain a read-index +confirmation and apply through that index before capturing the query result. +Admission rejects a newly elected leader until an entry from its current term +commits: raft-rs otherwise silently drops the request. This rejection reserves +neither capacity nor the token; callers can retry after driving replication. +`Events::reads` includes the request token, group, cluster/layout scope, current +term, confirmed index and captured applied position. Accepted-page contents and +their cursor come from that same state. Producer queries return only the latest +metadata/index; they do not fetch, deserialize or clone payload bodies. + +Reads and proposals share each group's `max_pending` capacity and the owner's +request-token namespace. Accepted pages are limited to 128 records; queries contain +fixed identifiers and bounds. Publisher answers use the existing bounded table +representation. An ahead-of-state cursor returns `CursorAhead` after the quorum +read, without fencing the node. Source placement and catalog-group routing are +validated before admission. + +Each group's reader uses a new incarnation UUID on reopen and a never-reused +monotonic context counter. Caller token reuse cannot match an earlier context. +Confirmed reads wait for apply; term or leadership loss yields `Unknown` for +unresolved queries. The owner must keep ticking and polling to observe these +transitions. No follower forwarding, cached lease or cancellation shortcut is +provided. Storage failure fences the physical node, and a failing poll releases +none of its messages, append completions or query results. + +A returned value describes state observed during its request. It is not a lease, +authentication proof, producer ACK or external commit fence. Publication commands +must still validate immutable plans and dependencies at deterministic apply; +source retirement needs a durable publication outcome. A later takeover may occur +after a successful read. Exact append retries still use the existing proposal +path; using these reads to optimize retries is subsequent integration work. + +Tests cover new-leader readiness, shared admission, request-token reuse, stale +contexts, ahead-of-state cursors, apply barriers, physical sync failure and isolated leaders. Real TCP +child processes verify majority reads, independent appends during isolation, +unchanged log positions for reads, retained publisher snapshots and full process +restart with exact producer tails and payloads.