diff --git a/docs/architecture/server2.md b/docs/architecture/server2.md index 9b801eef6..2c5dbc9c0 100644 --- a/docs/architecture/server2.md +++ b/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. diff --git a/server2/config/tasks/KMA-RESPONSE-FIRST-WAITERS.json b/server2/config/tasks/KMA-RESPONSE-FIRST-WAITERS.json new file mode 100644 index 000000000..b7a5c1b62 --- /dev/null +++ b/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." + } + ] +} diff --git a/server2/src/storage/response_first.rs b/server2/src/storage/response_first.rs index 67816fe84..7c21fb062 100644 --- a/server2/src/storage/response_first.rs +++ b/server2/src/storage/response_first.rs @@ -78,9 +78,26 @@ pub struct Batch { pub declaration: GroupDeclaration, pub records: Vec, } +/// 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, 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 Publisher for CatalogStore { fn publish(&self, batch: Arc, deadline: Instant) -> BoxFuture<'_, Result<(), Error>> { @@ -595,11 +612,15 @@ impl Writer { F: Future, 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), diff --git a/server2/tests/response_first.rs b/server2/tests/response_first.rs index 643a8c462..6df9d7c1b 100644 --- a/server2/tests/response_first.rs +++ b/server2/tests/response_first.rs @@ -4,7 +4,9 @@ use server2::{ storage::{ CatalogId, CatalogLimits, Error, GroupDeclaration, GroupMember, Limits, Period, ProviderKey, RawRecord, - response_first::{Batch, Config, Publisher, ResponseScope, Writer}, + response_first::{ + Batch, Checkpoint, Config, Publisher, ReleaseReason, ResponseScope, Writer, + }, }, }; use std::{ @@ -830,3 +832,301 @@ fn inheriting_a_scope_keeps_nested_futures_small() { let size = std::mem::size_of_val(&nested); assert!(size < 4 * 1024, "{size} bytes on the caller's stack"); } + +/// A batch of one grid's current conditions (distinct natural keys for many acquisitions). +fn batch_grid(nx: u16, ny: u16) -> Batch { + let r = RawRecord::new( + "kma", + "current", + &ProviderKey::Grid { nx, ny }, + Period { + local_date: 20261025, + slot: "1200".into(), + }, + 1792897200000, + 200, + "application/json", + None, + Arc::from(&b"{\"t1h\":17}"[..]), + &Limits::default(), + ) + .unwrap(); + let d = GroupDeclaration::new( + vec![GroupMember { + envelope: r.envelope.clone(), + catalogs: vec![CatalogId::for_record(&r.envelope.identity, "20261025-1200").unwrap()], + }], + &CatalogLimits::default(), + ) + .unwrap(); + Batch { + declaration: d, + records: vec![r], + } +} +/// Stress check (not a regression test: its orders depend on the scheduler; the deterministic +/// ones are the `Checkpoint` tests below). 2,000 scoped acquisitions with instant providers, +/// created from the test thread; run on a saturated machine with +/// `cargo test --test response_first -- --ignored`. None may be released as unwaited +/// (NoWaiters) before the body ends; a dormancy release on a slow machine is not this defect. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "stress check: run under CPU saturation"] +async fn stress_scoped_instant_acquisitions_are_never_released_unwaited() { + let sink = Arc::new(Sink::default()); + let writer = Writer::new( + sink.clone(), + Config { + retained_bytes: 64 << 20, + group_bytes: 16384, + entries: 8192, + ..Config::default() + }, + ) + .unwrap(); + let scope = ResponseScope::default(); + scope + .run(async { + for i in 0..2000u16 { + writer + .acquire( + format!("current/{i}"), + 16384, + Instant::now() + Duration::from_secs(5), + async move { Ok(Some(batch_grid(1 + i % 140, 1 + i / 140))) }, + ) + .await + .unwrap(); + } + }) + .await; + let unwaited = released_early(&scope) + .into_iter() + .filter(|r| *r == ReleaseReason::NoWaiters) + .count(); + assert_eq!(unwaited, 0, "released as unwaited before the body ended"); + drop(scope.body(axum::body::Body::from("weather"))); + sink.allow.add_permits(4000); + assert!(writer.drain(Instant::now() + Duration::from_secs(10)).await); + assert_eq!(writer.stats().durable, 2000); +} + +/// A publisher that also holds the writer at one [`Checkpoint`]: the test learns when the writer +/// got there ([`Held::reached`]) and lets it go on ([`Held::go`]). Publications are counted and +/// wait for `allow`, as [`Sink`]'s. +struct Held { + at: Checkpoint, + reached: tokio::sync::watch::Sender, + open: tokio::sync::watch::Sender, + /// Every checkpoint passed so far. + passed: tokio::sync::watch::Sender>, + sink: Sink, +} +impl Held { + fn at(at: Checkpoint) -> Self { + Self { + at, + reached: tokio::sync::watch::channel(false).0, + open: tokio::sync::watch::channel(false).0, + passed: tokio::sync::watch::channel(Vec::new()).0, + sink: Sink::default(), + } + } + async fn reached(&self) { + let _ = self.reached.subscribe().wait_for(|r| *r).await; + } + /// Wait until the writer passed `point`. + async fn passed(&self, point: Checkpoint) { + let _ = self + .passed + .subscribe() + .wait_for(|p| p.contains(&point)) + .await; + } + fn go(&self) { + self.open.send_replace(true); + } +} +impl Publisher for Held { + fn publish(&self, batch: Arc, deadline: Instant) -> BoxFuture<'_, Result<(), Error>> { + self.sink.publish(batch, deadline) + } + fn checkpoint(&self, point: Checkpoint) -> BoxFuture<'_, ()> { + Box::pin(async move { + if point == self.at { + self.reached.send_replace(true); + let _ = self.open.subscribe().wait_for(|o| *o).await; + } + self.passed.send_modify(|p| p.push(point)); + }) + } +} +/// A writer whose checks before a body ends stay clear of its 10 s dormancy cap. +fn held_writer(held: Arc) -> Arc { + Writer::new( + held, + Config { + retained_bytes: 32768, + group_bytes: 16384, + dormancy: Duration::from_secs(10), + ..Config::default() + }, + ) + .unwrap() +} +/// The batches `scope` used that were released before its body ended, and why. +fn released_early(scope: &ResponseScope) -> Vec { + scope + .batches() + .into_iter() + .filter_map(|b| b.released.map(|r| r.1)) + .collect() +} + +/// RFW-zero-count-join: the response that created an acquisition is cancelled, so nobody waits +/// for it; another response joins it after its batch is valid and before it is Ready. That +/// response is a waiter: the batch is held until its body ends, not released as unwaited +/// (NoWaiters) and published while it is still being sent. +#[tokio::test] +async fn a_response_joining_before_ready_holds_a_cancelled_creators_batch() { + let held = Arc::new(Held::at(Checkpoint::Validated)); + let writer = held_writer(held.clone()); + let key = "grid/current/60/127/20261025-1200"; + let (provide, provided) = tokio::sync::oneshot::channel::<()>(); + let creator = tokio::spawn({ + let writer = writer.clone(); + async move { + let scope = ResponseScope::default(); + scope + .run(writer.acquire( + key.into(), + 16384, + Instant::now() + Duration::from_secs(5), + async move { + let _ = provided.await; + Ok(Some(batch())) + }, + )) + .await + } + }); + tokio::task::yield_now().await; + creator.abort(); + match creator.await { + Err(e) => assert!(e.is_cancelled()), + Ok(_) => panic!("the creator ran to its end"), + } + provide.send(()).unwrap(); + held.reached().await; + let joiner = ResponseScope::default(); + let joined = tokio::spawn({ + let (writer, joiner) = (writer.clone(), joiner.clone()); + async move { + joiner + .run(writer.acquire( + key.into(), + 16384, + Instant::now() + Duration::from_secs(5), + async { panic!("the joiner starts no provider") }, + )) + .await + } + }); + tokio::task::yield_now().await; + held.go(); + assert!(joined.await.unwrap().unwrap().is_some()); + assert_eq!( + released_early(&joiner), + Vec::new(), + "released before the body ended" + ); + assert_eq!( + held.sink.attempts.load(SeqCst), + 0, + "published before the body ended" + ); + drop(joiner.body(axum::body::Body::from("weather"))); + held.sink.allow.add_permits(1); + assert!(writer.drain(Instant::now() + Duration::from_secs(5)).await); + assert_eq!(writer.stats().durable, 1); +} + +/// One scoped acquisition of `writer` by a response of its own (`scope`), with an instant +/// provider. +fn respond( + writer: &Arc, + scope: &ResponseScope, +) -> tokio::task::JoinHandle>, Error>> +{ + let (writer, scope) = (writer.clone(), scope.clone()); + tokio::spawn(async move { + scope + .run(writer.acquire( + "grid/current/60/127/20261025-1200".into(), + 16384, + Instant::now() + Duration::from_secs(5), + async { Ok(Some(batch())) }, + )) + .await + }) +} +/// The body of `scope` ends and its batch becomes durable. +async fn body_ends(held: &Held, writer: &Arc, scope: &ResponseScope) { + drop(scope.body(axum::body::Body::from("weather"))); + held.sink.allow.add_permits(1); + assert!(writer.drain(Instant::now() + Duration::from_secs(5)).await); + assert_eq!(writer.stats().durable, 1); +} + +/// RFW-test-scheduling: the creating caller is paused right after it created the entry and +/// started its job, and the job makes the batch Ready meanwhile. The caller was a waiter from the +/// start, so its batch is held until its body ends. +#[tokio::test] +async fn a_creator_not_yet_waiting_when_its_batch_is_ready_holds_it() { + let held = Arc::new(Held::at(Checkpoint::Created)); + let writer = held_writer(held.clone()); + let scope = ResponseScope::default(); + let creator = respond(&writer, &scope); + held.reached().await; + held.passed(Checkpoint::Announced).await; + held.go(); + assert!(creator.await.unwrap().unwrap().is_some()); + assert_eq!( + released_early(&scope), + Vec::new(), + "released before the body ended" + ); + assert_eq!( + held.sink.attempts.load(SeqCst), + 0, + "published before the body ended" + ); + body_ends(&held, &writer, &scope).await; +} + +/// RFW-test-scheduling: the job is paused right after it made the batch Ready, and the waiting +/// response takes the batch and returns meanwhile. It was counted before Ready, so its batch is +/// held until its body ends. +#[tokio::test] +async fn a_waiter_that_takes_the_batch_at_once_holds_it() { + let held = Arc::new(Held::at(Checkpoint::Announced)); + let writer = held_writer(held.clone()); + let scope = ResponseScope::default(); + let waiter = respond(&writer, &scope); + held.reached().await; + assert!(waiter.await.unwrap().unwrap().is_some()); + held.go(); + for _ in 0..8 { + tokio::task::yield_now().await; + } + assert_eq!( + released_early(&scope), + Vec::new(), + "released before the body ended" + ); + assert_eq!( + held.sink.attempts.load(SeqCst), + 0, + "published before the body ended" + ); + body_ends(&held, &writer, &scope).await; +}