fix(consensus): close metadata and consensus plane TODOs

Cargo.toml

@@ -278,6 +278,7 @@ reqwest-tracing = "0.7.1"
 ring = "0.17.14"
 ringbuffer = "0.16.0"
 rmcp = "2.2.0"
+rmp = "0.8.15"
 rmp-serde = "1.3.1"
 rolling-file = "0.2.0"
 rust-embed = "8.12.0"

core/consensus/src/lib.rs

@@ -34,6 +34,10 @@ pub trait Pipeline {
 
     fn pop(&mut self) -> Option<Self::Entry>;
 
+    /// Drop the newest entry when it is `op`, returning it; `None` and no
+    /// mutation otherwise. Unwinds a push that turned out not to be durable.
+    fn remove_tail(&mut self, op: u64) -> Option<Self::Entry>;
+
     fn clear(&mut self);
 
     fn entry_by_op(&self, op: u64) -> Option<&Self::Entry>;
@@ -52,6 +56,9 @@ pub trait Pipeline {
 
     fn len(&self) -> usize;
 
+    /// Requests parked waiting for a prepare slot (the second queue).
+    fn request_queue_len(&self) -> usize;
+
     /// In-flight prepare-queue capacity. `VsrConsensus` snapshots it at
     /// construction to size the loopback queue and to bound the uncommitted
     /// range a new primary may rebuild after a view change.

core/consensus/src/metadata_helpers.rs

@@ -102,11 +102,7 @@ where
 {
     // In-flight dedup: a live prepare from this client absorbs the retry.
     // Pump delivers the reply at commit.
-    if consensus
-        .pipeline()
-        .borrow()
-        .has_message_from_client(client_id)
-    {
+    if consensus.pipeline_has_message_from_client(client_id) {
         tracing::debug!(
             client_id,
             request,
@@ -262,11 +258,7 @@ where
     P: Pipeline<Entry = PipelineEntry>,
 {
     // In-flight dedup.
-    if consensus
-        .pipeline()
-        .borrow()
-        .has_message_from_client(client_id)
-    {
+    if consensus.pipeline_has_message_from_client(client_id) {
         tracing::debug!(client_id, "register_preflight: in-flight prepare, drop");
         return false;
     }

core/consensus/src/plane_helpers.rs

@@ -415,15 +415,15 @@ where
         return false;
     }
 
-    let pipeline = consensus.pipeline().borrow();
     let mut new_commit = consensus.commit_max();
-    while let Some(entry) = pipeline.entry_by_op(new_commit + 1) {
-        if !entry.ok_quorum_received {
-            break;
+    consensus.with_pipeline(|pipeline| {
+        while let Some(entry) = pipeline.entry_by_op(new_commit + 1) {
+            if !entry.ok_quorum_received {
+                break;
+            }
+            new_commit += 1;
         }
-        new_commit += 1;
-    }
-    drop(pipeline);
+    });
 
     if new_commit > consensus.commit_max() {
         consensus.advance_commit_max(new_commit);
@@ -447,17 +447,27 @@ where
 {
     let commit = consensus.commit_max();
     let mut drained = Vec::new();
-    let mut pipeline = consensus.pipeline().borrow_mut();
 
-    while let Some(head_op) = pipeline.head().map(|entry| entry.header.op) {
-        if head_op > commit {
-            break;
+    consensus.with_pipeline_mut(|pipeline| {
+        while let Some(head_op) = pipeline.head().map(|entry| entry.header.op) {
+            if head_op > commit {
+                break;
+            }
+
+            let entry = pipeline
+                .pop()
+                .expect("drain_committable_prefix: head exists");
+            drained.push(entry);
         }
+    });
 
-        let entry = pipeline
-            .pop()
-            .expect("drain_committable_prefix: head exists");
-        drained.push(entry);
+    // Popping through the pipeline directly bypasses
+    // `VsrConsensus::pop_committed_prepare`, so re-establish the prepare
+    // timeout's ticking-iff-non-empty invariant here: an emptied pipeline
+    // disarms it, and a remaining head becomes the entry the timer measures,
+    // timed from now rather than inheriting the drained entry's elapsed ticks.
+    if !drained.is_empty() {
+        consensus.sync_prepare_timeout();
     }
 
     drained
@@ -478,10 +488,8 @@ where
     P: Pipeline<Entry = PipelineEntry>,
 {
     let commit = consensus.commit_max();
-    let pipeline = consensus.pipeline().borrow();
-    pipeline
-        .head()
-        .map(|entry| entry.header)
+    consensus
+        .pipeline_head_header()
         .filter(|header| header.op <= commit)
 }
 
@@ -1971,11 +1979,7 @@ mod tests {
 
         assert_eq!(drained_ops, vec![5, 6]);
         assert_eq!(
-            consensus
-                .pipeline()
-                .borrow()
-                .head()
-                .map(|entry| entry.header.op),
+            consensus.pipeline_head_header().map(|header| header.op),
             Some(7)
         );
     }

core/metadata/Cargo.toml

@@ -48,6 +48,7 @@ journal = { workspace = true }
 left-right = { workspace = true }
 message_bus = { workspace = true }
 paste = { workspace = true }
+rmp = { workspace = true }
 rmp-serde = { workspace = true }
 serde = { workspace = true, features = ["derive"] }
 server_common = { workspace = true }