feat(wal): add bounded quorum application queries

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"]

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,

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<Packet>,
     pub completions: Vec<Completion>,
+    pub reads: Vec<query::Completion>,
 }
 
 #[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<Arc<checkpoint::Image>>,
     incoming: Option<PreparedCheckpoint>,
 }
@@ -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<Page> {
         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<Option<Batch>> {

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<Box<publication::Table>>),
+    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<Value> {
+        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<u64>,
+}
+
+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<Vec<u8>> {
+        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<Vec<Completion>> {
+        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)
+    }
+}