perf(observability): count what the pipeline ordering guard costs (#513)

CHANGELOG.md

@@ -6,6 +6,42 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+- **`INFO stats` reports what the pipeline ordering guarantee costs** (`total_pipeline_remote_defer`,
+  and the `moon_pipeline_remote_defer_total` Prometheus counter) — groundwork for #513.
+
+  #512 made a pipelined command that cannot route by its own single key wait for the batch's
+  pending cross-shard commands, which is what stopped the silent write loss of #507. It is also
+  expensive, and nothing said so: the only symptom was throughput that looked bad for no
+  visible reason.
+
+  Two things bound the cost, and the counter is what made both checkable:
+
+  - A deferral needs an **undispatched cross-shard command already in the batch** — the guard
+    is `!remote_groups.is_empty() && must_wait_for_pending_remote(..)`. A shard-spanning `MGET`
+    on its own never defers: 64 spread `MGET`s with no preceding writes measure **0**. A
+    preceding foreign *read* counts too, since the E2 read fast path is disabled and foreign
+    reads are slotted alongside writes.
+  - **At most one deferral per batch pass.** The cut re-parses the tail with `remote_groups`
+    cleared, so the command at the head of the next pass runs inline whatever its shape.
+
+  Together those explain why 64 interleavings produce fewer than 64 deferrals, and fewer still
+  at `--shards 2` (48) than at `--shards 4` (55) — with two shards, more of the preceding `SET`s
+  land locally and never reach `remote_groups` at all. The counts are shape- and
+  placement-specific, not constants: the same shape re-measured on a different key set gave 59.
+
+  Measured on moon-dev (aarch64, 6 vCPU), one connection, 9 reps alternating leg order, median:
+
+  | pipeline shape | shards=1 | shards=2 | shards=4 |
+  |---|---|---|---|
+  | `MGET` after every 2 `SET`s | 1,296,360 ops/s (0 deferrals) | 49,203 (48) | 38,856 (55) |
+  | 128 `SET`s then one `MGET` | 1,156,693 (0) | 659,436 (1) | 539,996 (1) |
+  | `SET`,`SET`,`GET` — routes by its own key | 1,700,287 (0) | 1,122,573 (0) | 993,784 (0) |
+
+  The deferral counts are the server's own, not inferred: reading the code suggested 64 for the
+  first shape and the counter says 48, which is exactly why it exists. The `--shards 1` column
+  is the control — `remote_groups` is always empty there, so the guard structurally cannot fire.
+
 ### Changed
 - **Sharded pub/sub channels are no longer workspace-scoped** (#703, fallout of #668). The
   hand-rolled workspace key walker had `PUBLISH`/`SUBSCRIBE`/`PSUBSCRIBE` in its no-key

src/admin/metrics_setup/memory.rs

@@ -9,7 +9,8 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use metrics::gauge;
 
 use crate::admin::metrics_setup::{
-    DISPATCH_CROSS_READ_SPSC_TOTAL, METRICS_INITIALIZED, TOTAL_CONNECTIONS, total_commands_sum,
+    DISPATCH_CROSS_READ_SPSC_TOTAL, METRICS_INITIALIZED, PIPELINE_REMOTE_DEFER_TOTAL,
+    TOTAL_CONNECTIONS, total_commands_sum,
 };
 
 // ── Memory metrics ──────────────────────────────────────────────────────
@@ -226,6 +227,15 @@ pub fn total_dispatch_cross_spsc() -> u64 {
     DISPATCH_CROSS_READ_SPSC_TOTAL.load(Ordering::Relaxed)
 }
 
+/// Total pipeline batches cut short by the moon#507 ordering guard (moon#513).
+///
+/// One per extra dispatch/await boundary. Always accurate — does not require
+/// Prometheus to be initialised.
+#[inline]
+pub fn total_pipeline_remote_defer() -> u64 {
+    PIPELINE_REMOTE_DEFER_TOTAL.load(Ordering::Relaxed)
+}
+
 /// Read process CPU usage via `getrusage(RUSAGE_SELF)`.
 ///
 /// Returns `(used_cpu_sys, used_cpu_user)` in seconds (f64).

src/admin/metrics_setup/mod.rs

@@ -318,6 +318,12 @@ static WAL_AGGRESSIVE_RECYCLE_BYTES_TOTAL: AtomicU64 = AtomicU64::new(0);
 // SPSC (when fast-path is off) and writes. INFO exposes the unified total
 // as `total_dispatch_cross_spsc`.
 static DISPATCH_CROSS_READ_SPSC_TOTAL: AtomicU64 = AtomicU64::new(0);
+/// moon#513: pipeline batches cut short because a command could not execute
+/// against shards whose earlier writes in the same batch were still pending.
+/// Each increment is one extra dispatch/await boundary — measured at ~57us on
+/// moon-dev — so this is the counter that says whether a slow pipeline is
+/// paying the moon#512 ordering guarantee or something else entirely.
+static PIPELINE_REMOTE_DEFER_TOTAL: AtomicU64 = AtomicU64::new(0);
 
 // ── INFO-readable counter accessors ─────────────────────────────────────
 

src/admin/metrics_setup/recorders.rs

@@ -11,8 +11,8 @@ use metrics::{counter, gauge, histogram};
 
 use crate::admin::metrics_setup::{
     CONNECTED_CLIENTS, DISPATCH_CROSS_READ_SPSC_TOTAL, EVICTED_KEYS, EXPIRING_SPILL_SKIPPED,
-    KEYSPACE_HITS, KEYSPACE_MISSES, METRICS_INITIALIZED, SPILLED_KEYS, TOTAL_CONNECTIONS,
-    WAL_AGGRESSIVE_RECYCLE_BYTES_TOTAL, WAL_AGGRESSIVE_RECYCLE_SEGMENTS_TOTAL,
+    KEYSPACE_HITS, KEYSPACE_MISSES, METRICS_INITIALIZED, PIPELINE_REMOTE_DEFER_TOTAL, SPILLED_KEYS,
+    TOTAL_CONNECTIONS, WAL_AGGRESSIVE_RECYCLE_BYTES_TOTAL, WAL_AGGRESSIVE_RECYCLE_SEGMENTS_TOTAL,
 };
 
 // ── Connection metrics ──────────────────────────────────────────────────
@@ -324,6 +324,26 @@ pub fn record_xshard_reply_timeout(kind: &'static str) {
     counter!("moon_xshard_reply_timeout_total", "kind" => kind).increment(1);
 }
 
+/// A pipeline batch was cut short by the moon#507 ordering guard.
+///
+/// Recorded at the two sites that set `deferred_tail_from` for
+/// `must_wait_for_pending_remote`: the command and the unconsumed tail move to
+/// the next loop iteration so the pending `remote_groups` resolve first.
+///
+/// Exists because the cost is invisible otherwise. A client interleaving reads
+/// between write groups at `--shards >= 2` pays one boundary per interleaving,
+/// and nothing in `INFO` said so — the throughput just looked bad. It is also
+/// what keeps a benchmark for moon#513 honest: a shape that claims to trigger
+/// the guard has to be able to PROVE it did, and how often.
+#[inline]
+pub fn record_pipeline_remote_defer() {
+    PIPELINE_REMOTE_DEFER_TOTAL.fetch_add(1, Ordering::Relaxed);
+    if !METRICS_INITIALIZED.load(Ordering::Relaxed) {
+        return;
+    }
+    counter!("moon_pipeline_remote_defer_total").increment(1);
+}
+
 /// Batched variant of `record_dispatch_cross_spsc`.
 #[inline]
 pub fn record_dispatch_cross_spsc_batch(count: u64) {

src/command/connection.rs

@@ -657,13 +657,15 @@ fn info_raw(db: &Database, facts: &InstanceFacts) -> String {
         "total_commands_processed:{}\r\n\
          total_connections_received:{}\r\n\
          total_dispatch_cross_spsc:{}\r\n\
+         total_pipeline_remote_defer:{}\r\n\
          spsc_notify_wakes:{}\r\n\
          spsc_drain_renotify:{}\r\n\
          spsc_notify_skipped:{}\r\n\
          ft_search_cooperative_yields_total:{}\r\n",
         crate::admin::metrics_setup::total_commands_processed(),
         crate::admin::metrics_setup::total_connections_received(),
         crate::admin::metrics_setup::total_dispatch_cross_spsc(),
+        crate::admin::metrics_setup::total_pipeline_remote_defer(),
         crate::admin::metrics_setup::spsc_notify_wakes(),
         crate::admin::metrics_setup::spsc_drain_renotify(),
         crate::admin::metrics_setup::spsc_notify_skipped(),

src/server/conn/handler_monoio/mod.rs

@@ -1678,6 +1678,7 @@ pub(crate) async fn handle_connection_sharded_monoio<
                 && crate::server::conn::shared::must_wait_for_pending_remote(cmd, cmd_args)
             {
                 frames[frame_idx - 1] = frame;
+                crate::admin::metrics_setup::record_pipeline_remote_defer();
                 deferred_tail_from = Some(frame_idx - 1);
                 break;
             }

src/server/conn/handler_sharded/mod.rs

@@ -818,6 +818,7 @@ pub(crate) async fn handle_connection_sharded_inner<
                         && crate::server::conn::shared::must_wait_for_pending_remote(cmd, cmd_args)
                     {
                         batch[frame_idx - 1] = frame;
+                        crate::admin::metrics_setup::record_pipeline_remote_defer();
                         deferred_tail_from = Some(frame_idx - 1);
                         break;
                     }

tests/pipeline_cross_shard_ordering.rs

@@ -166,6 +166,101 @@ fn each_trial(port: u16, tag: &str, mut body: impl FnMut(&mut Conn, &str) -> Opt
 // ---------------------------------------------------------------------------
 
 /// moon#507 as reported: MGET returns nulls for keys its own batch just wrote.
+/// The ordering guarantee has a PRICE, and `INFO` now names it — moon#513.
+///
+/// Every other test in this file asserts the guarantee holds. This one asserts
+/// the cost of holding it is observable, because it was not: a client
+/// interleaving reads between write groups at `--shards >= 2` pays one extra
+/// dispatch/await boundary per interleaving, and nothing reported it. The
+/// throughput simply looked bad. Measured on moon-dev, `--shards 2`, one
+/// connection: an `MGET` after every two `SET`s runs at 49,203 ops/s against
+/// 1,122,573 for the same shape with single-key `GET`s.
+///
+/// It is also the acceptance criterion for moon#513, stated as a number rather
+/// than a description: letting multi-key commands join the slotted batch means
+/// `interleaved` stops deferring. When that lands, the assertion below flips
+/// from "> 0" to "== 0" and the test stays meaningful either way.
+///
+/// The `--shards 1` leg is what keeps it honest. There `remote_groups` is
+/// always empty, so the guard structurally cannot fire — a counter that moved
+/// there would be counting something else, and every claim built on it would
+/// be wrong.
+#[test]
+fn pco12_the_ordering_guard_reports_what_it_costs() {
+    fn defers(port: u16) -> u64 {
+        let mut c = Conn::open(port);
+        let info = c.send(&["INFO", "stats"]);
+        info.split("\r\n")
+            .find_map(|l| l.strip_prefix("total_pipeline_remote_defer:"))
+            .and_then(|v| v.trim().parse().ok())
+            .unwrap_or_else(|| {
+                panic!("INFO stats has no total_pipeline_remote_defer field: {info:?}")
+            })
+    }
+
+    /// `MGET` after every two `SET`s — the shape that triggers the guard.
+    fn interleaved(c: &mut Conn, tag: &str, n: usize) {
+        for i in (0..n).step_by(2) {
+            let (a, b) = (format!("{tag}:{i}"), format!("{tag}:{}", i + 1));
+            c.pipeline(&[&["SET", &a, "v"], &["SET", &b, "v"], &["MGET", &a, &b]]);
+        }
+    }
+
+    /// The same shape with single-key reads, which route by their own key and
+    /// therefore never need to wait.
+    fn single_key(c: &mut Conn, tag: &str, n: usize) {
+        for i in (0..n).step_by(2) {
+            let (a, b) = (format!("{tag}:{i}"), format!("{tag}:{}", i + 1));
+            c.pipeline(&[&["SET", &a, "v"], &["SET", &b, "v"], &["GET", &a]]);
+        }
+    }
+
+    // --- shards = 1: the guard cannot fire, whatever the shape ---
+    {
+        let m = spawn_moon("1");
+        let base = defers(m.port);
+        let mut c = Conn::open(m.port);
+        interleaved(&mut c, "pco12s1", 64);
+        drop(c);
+        assert_eq!(
+            defers(m.port) - base,
+            0,
+            "at --shards 1 `remote_groups` is always empty, so the moon#507 \
+             guard cannot fire — a non-zero count here means the counter is \
+             measuring something other than the deferral"
+        );
+    }
+
+    // --- shards = 4: the interleaved shape defers, the single-key one does not ---
+    let m = spawn_moon(SHARDS);
+
+    let base = defers(m.port);
+    let mut c = Conn::open(m.port);
+    single_key(&mut c, "pco12ctl", 64);
+    drop(c);
+    let control_defers = defers(m.port) - base;
+    assert_eq!(
+        control_defers, 0,
+        "a pipeline of single-key commands routes every command by its own key \
+         and must never defer — {control_defers} deferrals means the guard is \
+         firing on the fast path, which would be a throughput regression for \
+         every pipelined client"
+    );
+
+    let base = defers(m.port);
+    let mut c = Conn::open(m.port);
+    interleaved(&mut c, "pco12int", 64);
+    drop(c);
+    let interleaved_defers = defers(m.port) - base;
+    assert!(
+        interleaved_defers > 0,
+        "an MGET interleaved between writes at --shards {SHARDS} must trigger \
+         the moon#507 guard, and INFO must say so. Zero here means either the \
+         counter is not wired to the deferral site, or moon#513 has landed — \
+         in which case flip this assertion to `== 0` rather than deleting it"
+    );
+}
+
 #[test]
 fn pco1_mget_sees_writes_from_its_own_batch() {
     let m = spawn_moon(SHARDS);