server2: count response-first waiters before a batch is ready

docs/architecture/server2.md

@@ -342,7 +342,12 @@ instance headroom remains unqualified. Normal retries retain the same original
 bytes (9 s attempts, backoff capped at 30 s). Deterministic invalid input parks
 until shutdown; corrupt stored/control reads remain eligible for retry.
 Body end/error/drop releases publication; a 10 s dormancy cap, no-waiter detection
-and shutdown also release it. The hook proves server body termination, not remote
+and shutdown also release it. A caller is a waiter from the moment it creates or joins an
+acquisition, and the waiters are counted together with the Ready announcement, under the lock
+callers create and join entries under. A waiter that takes the batch and returns at once, one not
+yet polled again when a fast provider is done, or a response that joins after the creator was
+cancelled but before Ready therefore never makes its response's raw look unwaited: on a loaded
+runner such an order published a group before the body ended (CI run 42379684133, 2026-11-07). The hook proves server body termination, not remote
 receipt. Quiesce rejects queued misses and a 34 s shutdown drain counts abandoned
 served originals. No durable copy means exact crash recovery can be impossible.
 

server2/config/tasks/KMA-RESPONSE-FIRST-WAITERS.json

@@ -0,0 +1,17 @@
+{
+  "task": "KMA-RESPONSE-FIRST-WAITERS",
+  "issue": "https://github.com/WizardFactory/TodayWeather/issues/2614",
+  "scope": "Response-first writer: count a creating or joining caller as a waiter before its acquisition can finish, and settle who waits together with the Ready announcement under the lock entries are created and joined under, so a response's own raw is never released as NoWaiters and published before the response body ends. A publisher checkpoint is the tests' seam (no setting). No deployment or production change.",
+  "server2_paths": [
+    "server2/config/tasks/KMA-RESPONSE-FIRST-WAITERS.json",
+    "server2/src/storage/response_first.rs",
+    "server2/tests/response_first.rs"
+  ],
+  "outside": [
+    {
+      "path": "docs/architecture/server2.md",
+      "category": "documentation",
+      "reason": "Response-first writer paragraph: waiters are counted with the Ready announcement."
+    }
+  ]
+}

server2/src/storage/response_first.rs

@@ -78,9 +78,26 @@ pub struct Batch {
     pub declaration: GroupDeclaration,
     pub records: Vec<RawRecord>,
 }
+/// A step of an acquisition at which the writer awaits [`Publisher::checkpoint`].
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum Checkpoint {
+    /// The creating caller has created the entry and started its job, and waits for it next.
+    Created,
+    /// The job's batch is valid and charged; who waits for it is not settled and it is not
+    /// Ready yet.
+    Validated,
+    /// The batch is Ready and who waits for it is settled.
+    Announced,
+}
 /// Only this publisher can confer durable eligibility. RAM records carry no S06 token.
 pub trait Publisher: Send + Sync + 'static {
     fn publish(&self, batch: Arc<Batch>, deadline: Instant) -> BoxFuture<'_, Result<(), Error>>;
+    /// A test seam, not a setting: the writer awaits it at `point` of every acquisition, so a
+    /// test can order responses against the writer's steps. Production publishers keep this
+    /// default, which is ready at once.
+    fn checkpoint(&self, _point: Checkpoint) -> BoxFuture<'_, ()> {
+        Box::pin(std::future::ready(()))
+    }
 }
 impl<T: CatalogTransport + 'static> Publisher for CatalogStore<T> {
     fn publish(&self, batch: Arc<Batch>, deadline: Instant) -> BoxFuture<'_, Result<(), Error>> {
@@ -595,11 +612,15 @@ impl Writer {
         F: Future<Output = Result<Option<Batch>, Error>> + Send + 'static,
     {
         let scope = ResponseScope::current();
-        let entry = {
+        // The caller is a waiter from the moment it creates or joins the entry: its receiver
+        // exists before the provider can finish, so a provider that is done before the caller
+        // is polled again is not released as one nobody waits for (NoWaiters), which would
+        // publish a response's raw before its body ends.
+        let (mut receiver, created) = {
             let mut table = self.table.lock().unwrap_or_else(|e| e.into_inner());
             Self::prune(&mut table);
             if let Some(entry) = table.entries.get(&key) {
-                entry.clone()
+                (entry.state.subscribe(), false)
             } else {
                 if self.closing.load(SeqCst)
                     || reserve == 0
@@ -618,6 +639,7 @@ impl Writer {
                 if scope.is_none() {
                     entry.gate.release(ReleaseReason::Background);
                 }
+                let receiver = entry.state.subscribe();
                 table.entries.insert(key, entry.clone());
                 let this = self.clone();
                 let owned = entry.clone();
@@ -659,10 +681,12 @@ impl Writer {
                     this.run(owned, lease, reserve, provider_until, provider)
                         .await;
                 }));
-                entry
+                (receiver, true)
             }
         };
-        let mut receiver = entry.state.subscribe();
+        if created {
+            self.publisher.checkpoint(Checkpoint::Created).await;
+        }
         loop {
             let state = receiver.borrow_and_update().clone();
             match state {
@@ -772,11 +796,22 @@ impl Writer {
             _lease: lease,
         });
         self.counters.ready.fetch_add(1, SeqCst);
-        // The supervisor owns the charged batch before a waiter can observe Ready.
-        entry.state.send_replace(Stage::Ready(ready.clone()));
-        if entry.state.receiver_count() == 0 {
-            entry.gate.release(ReleaseReason::NoWaiters);
+        self.publisher.checkpoint(Checkpoint::Validated).await;
+        // Who waits is settled under the table lock, the one callers create and join entries
+        // under, together with the Ready announcement: a response that joins before Ready is
+        // counted, one that joins after it finds the batch Ready, and a waiter cannot take the
+        // batch and leave before it is counted. Released as unwaited (NoWaiters), a response's
+        // own raw would be published before its body ends.
+        {
+            let _joins = self.table.lock().unwrap_or_else(|e| e.into_inner());
+            let waiters = entry.state.receiver_count();
+            // The supervisor owns the charged batch before a waiter can observe Ready.
+            entry.state.send_replace(Stage::Ready(ready.clone()));
+            if waiters == 0 {
+                entry.gate.release(ReleaseReason::NoWaiters);
+            }
         }
+        self.publisher.checkpoint(Checkpoint::Announced).await;
         tokio::select! {
             _=entry.gate.wait()=>(),
             _=tokio::time::sleep(self.config.dormancy)=>entry.gate.release(ReleaseReason::DormancyCap),