Wait for the push that carries a missing event, not for a clock

Cargo.toml

@@ -3,7 +3,7 @@ members = ["codegen", "dsl", "examples", "performance_measurement", "performance
 
 [package]
 name = "worktable"
-version = "1.10.0-beta1"
+version = "1.10.0-beta2"
 edition = "2024"
 authors = ["Handy-caT"]
 license = "MIT"
@@ -147,12 +147,12 @@ walkdir = { version = "2", optional = true }
 # These pre-release workspace crates move as one train. The explicit caret
 # keeps the dependency policy consistent while the local path selects this
 # checkout during validation.
-worktable_codegen = { path = "codegen", version = "^1.10.0-beta1" }
+worktable_codegen = { path = "codegen", version = "^1.10.0-beta2" }
 # Re-exported below. Each generated table carries its declaration as a const
 # whose documentation says to read it with `worktable_dsl::Schema::parse`; that
 # instruction is only true if a plain `worktable` dependency can reach the
 # crate.
-worktable_dsl = { path = "dsl", version = "^1.10.0-beta1", optional = true }
+worktable_dsl = { path = "dsl", version = "^1.10.0-beta2", optional = true }
 
 [target.'cfg(unix)'.dependencies]
 libc = { version = "^0.2", default-features = false }

codegen/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "worktable_codegen"
-version = "1.10.0-beta1"
+version = "1.10.0-beta2"
 edition = "2024"
 license = "MIT"
 description = "Proc-macro companion crate for worktable: the worktable! macro and its derives."
@@ -26,7 +26,7 @@ proc-macro = true
 # a declaration. See its crate docs for why that needed a separate crate.
 # Name the reviewed prerelease while accepting compatible schema-model and
 # validator updates. Release checks verify the resolved generated API.
-worktable_dsl = { path = "../dsl", version = "^1.10.0-beta1" }
+worktable_dsl = { path = "../dsl", version = "^1.10.0-beta2" }
 # Test-only. As a normal dependency this proc-macro crate put `rkyv` with its
 # default features into the graph, which turned on `rkyv/std` for the target
 # build too and dragged `ptr_meta` with it. The generated code names `rkyv`

dsl/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "worktable_dsl"
-version = "1.10.0-beta1"
+version = "1.10.0-beta2"
 edition = "2024"
 license = "MIT"
 description = "The worktable! schema language: its model and parser, readable outside the proc macro"

src/persistence/mod.rs

@@ -118,10 +118,37 @@ mod space;
 mod task;
 
 // TODO: remove this
+/// Backstop for [`PersistenceConfig::event_gap_wait_cap`].
+///
+/// The worker waits on the queue's own wake-up, so this is only reached when no
+/// push arrives at all and no wake was delivered. It is short because nothing is
+/// bought by making it long: the wait exists to avoid spinning, not to give a
+/// producer time, and the producer's own push is what ends it.
+pub const DEFAULT_EVENT_GAP_WAIT_CAP: core::time::Duration = core::time::Duration::from_millis(100);
+
 pub trait PersistenceConfig {
     fn table_path(&self) -> &str;
 
     fn version(&self) -> u32;
+
+    /// Longest the persistence worker parks waiting for an index event that is
+    /// missing from the queue.
+    ///
+    /// The wait itself is event-driven: the worker registers on the queue's
+    /// wake-up and returns the moment an operation is pushed, so on a table
+    /// whose producers are live this cap is never reached and the value does
+    /// not matter. It bounds the one case the wake cannot cover, a wake lost
+    /// to a race, and it is the only place a number is still guessed.
+    ///
+    /// Override it per table when this table's producers are slower than the
+    /// default assumes -- a remote or batch producer that can genuinely be a
+    /// second between pushes -- or shorten it when a stall must surface fast.
+    /// Note what it does *not* bound: the worker gives up on a gap after a
+    /// fixed number of waits, so this cap multiplied by that count is the
+    /// worst-case time before a stalled gap fails the table.
+    fn event_gap_wait_cap(&self) -> core::time::Duration {
+        DEFAULT_EVENT_GAP_WAIT_CAP
+    }
 }
 
 /// Controls the consistency checks applied while loading persisted state.

src/persistence/operation/batch.rs

@@ -23,11 +23,17 @@ use crate::prelude::{Order, SelectQueryExecutor};
 /// fails the table.
 ///
 /// A gap is usually transient: the operation carrying the missing id has been
-/// pushed but not yet batched, or its producer has not reached its push. Each
-/// deferral sleeps 500ms in the worker loop, so this is about a minute of
-/// waiting. The previous value of eight was about four seconds, which a
-/// producer descheduled under load can lose, and the engine then blamed a
-/// permanent bug for what was a slow thread.
+/// pushed but not yet batched, or its producer has not reached its push.
+///
+/// This counts deferrals, not time, and the difference matters now that the
+/// worker waits on the push rather than on a clock. A deferral that ends
+/// because the awaited operation arrived costs whatever the producer took; one
+/// that ends on the backstop costs
+/// [`PersistenceConfig::event_gap_wait_cap`](crate::persistence::PersistenceConfig::event_gap_wait_cap),
+/// so that cap times this count is the worst case before a stalled gap fails
+/// the table. It was raised from eight for a reason that still holds: eight was
+/// about four seconds, which a producer descheduled under load can lose, and
+/// the engine then blamed a permanent bug for what was a slow thread.
 ///
 /// Widening the collection is a *separate* decision, taken far sooner and
 /// tracked by the analyzer's own no-progress counter. This one only decides

src/persistence/task.rs

@@ -72,6 +72,22 @@ const MAX_BATCH_OPERATIONS: usize = 512;
 /// contiguous prefix than a partial one, never something unsafe.
 const COLLECT_WHOLE_QUEUE_AFTER_ATTEMPTS: usize = 4;
 
+// History of the wait the drain loop no longer performs, kept because the number it
+// used is a tempting thing to reintroduce.
+//
+// The loop used to sleep a flat 500 ms whenever a collection found a hole in the event
+// stream. That was a clock standing in for an event, and it was measured downstream as
+// the dominant cost of shutting a daemon down: a settled store stepped from 0.42 s at
+// 700 indexed files to 1.45 s at 800 and then flattened, rising after that only at the
+// underlying linear rate of about 0.06 s per 100 files. A page drain cannot step by a
+// second and then flatten; the riser was two of these sleeps.
+//
+// It is now `Queue::await_more_operations`, which waits on the queue's own wake-up and
+// returns the instant the missing operation is pushed. The one number left is a
+// backstop against a lost wake, is reached only when nothing is pushed at all, and is a
+// per-table parameter rather than a constant: `PersistenceConfig::event_gap_wait_cap`,
+// defaulting to `crate::persistence::DEFAULT_EVENT_GAP_WAIT_CAP`.
+
 #[derive(Debug)]
 struct PersistenceLifecycle {
     state: ParkingMutex<PersistenceState>,
@@ -1771,6 +1787,55 @@ impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys> Queue<PrimaryKeyGenState, Pr
         self.notify.notify_waiters();
     }
 
+    /// Waits for an operation that could close an event gap, rather than for a
+    /// clock.
+    ///
+    /// Only the drain loop calls this, and only once collection has taken the
+    /// whole queue and *still* found a hole, which is the one situation where
+    /// the missing event genuinely has to arrive from somewhere else. That
+    /// somewhere else is a push, so a push is what this waits on: the wait ends
+    /// the instant the operation carrying the missing id is queued, and on a
+    /// table with live producers `cap` is never reached.
+    ///
+    /// Two things end it early, each for its own reason.
+    ///
+    /// A non-empty queue returns immediately. The push may have landed while
+    /// the collection that found the hole was still running, in which case the
+    /// event is already here and waiting for it is waiting for something that
+    /// has happened. The waiter is registered *before* the queue is read, so a
+    /// push landing between the two wakes this rather than being missed.
+    ///
+    /// Anything other than `Running` returns immediately too, and this one is
+    /// provable rather than a heuristic: a push takes the lifecycle lock and is
+    /// refused unless the state is `Running`, and `begin_close` takes the same
+    /// lock, so once `Closing` has been observed here no further push can ever
+    /// be accepted. The missing event can no longer arrive, and every moment
+    /// spent waiting for it is time bought for nothing -- which is exactly the
+    /// shape shutdown showed. The yield keeps the escalation off this worker's
+    /// only pool thread without pretending to wait.
+    ///
+    /// `cap` therefore covers one case: a wake lost to a race, with no later
+    /// push to deliver another. It comes from
+    /// [`PersistenceConfig::event_gap_wait_cap`](crate::persistence::PersistenceConfig::event_gap_wait_cap)
+    /// so a table whose producers are genuinely slower than the default assumes
+    /// can say so, instead of a constant deciding for every table at once.
+    async fn await_more_operations(&self, cap: Duration) {
+        let notified = self.notify.notified();
+        let mut notified = core::pin::pin!(notified);
+        // Registered before both checks below, for the reason spelled out in
+        // `pop_marking_in_progress`: `wake()` uses `notify_waiters`, which
+        // retains no permit, so a waiter created afterwards misses it forever.
+        notified.as_mut().enable();
+        if self.len() != 0 {
+            return;
+        }
+        if !matches!(self.lifecycle.state(), PersistenceState::Running) {
+            nagoya::yield_now().await;
+            return;
+        }
+        let _ = nagoya::timeout(cap, notified).await;
+    }
+
     fn immediate_pop(&self) -> Option<PersistenceMessage<PrimaryKeyGenState, PrimaryKey, SecondaryKeys>> {
         if let Some(v) = self.queue.lock().pop_front() {
             self.len.fetch_sub(1, Ordering::Release);
@@ -2032,6 +2097,11 @@ impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys, AvailableIndexes>
         AvailableIndexes: Copy + Clone + Debug + Hash + Eq + Send + Sync + 'static,
     {
         let table_path = engine.config().table_path().to_owned();
+        // Read once, here, rather than per wait: the worker owns the engine
+        // from this point on, and a parameter that could change under a
+        // running drain loop would be a worse thing to explain than a
+        // parameter that is fixed when the table starts.
+        let event_gap_wait_cap = engine.config().event_gap_wait_cap();
         let lifecycle = Arc::new(PersistenceLifecycle::new());
         let queue = Arc::new(Queue::new(lifecycle.clone(), &table_path));
 
@@ -2137,10 +2207,16 @@ impl<PrimaryKeyGenState, PrimaryKey, SecondaryKeys, AvailableIndexes>
                         // Only here is waiting the right thing: collection has
                         // already taken the whole queue and the stream still
                         // has a hole, so the event it needs is not yet queued.
-                        // Sleeping on the escalating retries instead charged
-                        // 500 ms for each step towards the fallback that fixes
-                        // them, which is how a 0.03s drain became 290s.
-                        nagoya::sleep(Duration::from_millis(500)).await;
+                        // Waiting on the escalating retries instead charged for
+                        // every step towards the fallback that fixes them,
+                        // which is how a 0.03s drain became 290s.
+                        //
+                        // This waits on the push that would carry the missing
+                        // event, not on a clock. See
+                        // `Queue::await_more_operations` for why a non-empty
+                        // queue and a non-`Running` lifecycle each end it at
+                        // once, and for what the cap is still there to cover.
+                        engine_queue.await_more_operations(event_gap_wait_cap).await;
                     }
                 } else if let Some(page_ids) = pending_reclaim.take() {
                     // `get_first_op_id_available() == None` is only sufficient