Contract-suite wake row (task suite-wake-rows): wake_receiver_shapes — the wake contract's pinnable core, tolerance-bounded to state outcomes (never delivery counts or latencies): a pre-attached listener receives a wake after a committed notify on its channel through all three recv forms (recv/try_recv/recv_timeout), every wake's channel field matching the listened channel (the one piece of semantic content a wake carries, ADR-008 §3); a three-notify burst floors at one wake — never an exact per-notify count (coalescing the documented engine asymmetry: SQLite's 1-slot feed vs pg per-notify, the row pins the floor not the shape); a listener attached after a commit never sees that commit's notify — recv_timeout idles a 400 ms absence window (a 300 ms pre-settle sleep closes SQLite's watcher baseline race: the last_version baseline is store-open-captured, so an unconsumed commit's version change would fire one late tick at the new subscriber indistinguishable from replay; pg's gap-commit no-replay hole and SQLite's burst coalescing both legal under the pin); and the channel-scoped leg — a foreign-channel notify never surfaces a wake naming it (SQLite's same-commit overtrigger may deliver but carries only the listened channel; the pg LISTEN fanout skips foreign channels; the reserved reconnect straggler tolerated), with a same-channel positive control proving the silence is scoping not a dead listener. The failure-surface close arms (SQLite watcher-death recv()->None, pg synthetic reconnect-wake) stay pinned engine-side and in receiver_close_and_save_arms's disposal leg — cross-referenced in the doc comment, not re-pinned. Stamped ADR-006 + ADR-008 §3. Wired into both engines' suite targets (SQLite tokio test, pg harness_row!), 24 rows per column up from 23. Dispositions in Notes: the backlog row stays in core-contract.md's inventory (flushes at review-wave-5's stable gate, like the other discharged rows), the absence windows are single recv_timeout calls (the documented Ok(None) idle arm as the bounded wait), and the scoping leg's observable is the channel field not absence (SQLite overtrigger makes an absence-only pin vacuous there). Verified: SQLite column green server-less, pg column green vs harness (postgres/poc@:15432), 3 solo re-runs of the row per engine (determinism), cargo test -p alkstore-sqlite -p alkstore-postgres green (SQLite 189+24, pg 121+24+9), workspace cargo test green (12 binaries), clippy -D warnings, fmt clean
This commit is contained in:
1 parent
98de4a49cd
commit
2a2ad7e11f
5 files changed
+302
-6
No files matched your search
@@ -1117,6 +1117,244 @@ pub async fn receiver_close_and_save_arms(factory: &dyn StoreFactory) {
|
||||
);
|
||||
}
|
||||
|
||||
/// **Wake semantics under the pinned `WakeReceiver` shapes** — the
|
||||
/// wake contract's pinnable core, tolerance-bounded (wakes are
|
||||
/// best-effort hints — the row asserts state outcomes, never delivery
|
||||
/// counts or latencies). A pre-attached listener receives a wake after
|
||||
/// a committed `notify` on its channel, with all three recv forms
|
||||
/// exercised (`recv`, `try_recv`, `recv_timeout`) and every wake
|
||||
/// observed carrying the listened channel — the one piece of semantic
|
||||
/// content a wake carries (ADR-008 §3). A burst of notifies yields
|
||||
/// *at least one* wake, never an exact per-notify count (coalescing is
|
||||
/// the documented engine asymmetry — possibly coalesced on SQLite's
|
||||
/// 1-slot feed, per-notify on pg; the row pins the floor, not the
|
||||
/// shape). A listener attached *after* a commit never sees that
|
||||
/// commit's notify (`recv_timeout` idles through a bounded absence
|
||||
/// window — no replay; the notify commits before the attach, and pg's
|
||||
/// no-replay hole — a commit during a forwarder gap is never
|
||||
/// re-delivered — and SQLite's burst coalescing are both legal under
|
||||
/// this pin). Wakes from a different channel never arrive on this
|
||||
/// listener: after a foreign-channel notify, the bounded window
|
||||
/// surfaces no wake carrying the foreign consumer channel (SQLite's
|
||||
/// documented overtrigger may deliver a wake on the same commit, but
|
||||
/// it carries the *listened* channel only — nothing about what was
|
||||
/// notified; the pg fanout skips non-LISTENed channels, and the
|
||||
/// reserved reconnect-wake carries its own reserved name — that
|
||||
/// reserved straggler is a tolerated engine arm, not a violation), and
|
||||
/// a same-channel positive control still delivers — the silence was
|
||||
/// channel scoping, never a dead listener.
|
||||
///
|
||||
/// Cross-reference: the failure-surface close arms — SQLite's
|
||||
/// watcher-death `recv() -> None` and the pg synthetic reconnect-wake
|
||||
/// — are the engine-differing close arms, already pinned engine-side
|
||||
/// and in `receiver_close_and_save_arms` (the stream receiver's
|
||||
/// disposal leg); this row does not re-pin them — the reserved
|
||||
/// reconnect-wake appears here only as a tolerated straggler.
|
||||
///
|
||||
/// Contract stamp: ADR-006 (the wake contract — opaque wake +
|
||||
/// re-read, best-effort hints not per-commit delivery, coalescing /
|
||||
/// overtrigger asymmetry, the channel-scoped fanout); ADR-008 §3 (the
|
||||
/// pinned `Wake { channel }` / recv forms — the wake's channel field
|
||||
/// matches the listened channel; `recv_timeout`'s expiry-without-wake
|
||||
/// is the documented `Ok(None)` idle arm).
|
||||
pub async fn wake_receiver_shapes(factory: &dyn StoreFactory) {
|
||||
let store = factory.open().await.expect("factory opens a store");
|
||||
|
||||
// Delivery + channel identity: a pre-attached listener, a
|
||||
// committed notify on its channel, every recv form once.
|
||||
let mut rx = store.listen("wake-a").await.expect("listen attaches");
|
||||
|
||||
store
|
||||
.notify("wake-a", json!({"n": 1}))
|
||||
.await
|
||||
.expect("notify commits");
|
||||
let wake = tokio::time::timeout(std::time::Duration::from_secs(10), rx.recv())
|
||||
.await
|
||||
.expect("the committed notify delivers a wake — the bounded wait is enough")
|
||||
.expect("the receiver is open through a healthy listen");
|
||||
assert_eq!(
|
||||
wake.channel, "wake-a",
|
||||
"the recv wake's channel field matches the listened channel"
|
||||
);
|
||||
|
||||
store
|
||||
.notify("wake-a", json!({"n": 2}))
|
||||
.await
|
||||
.expect("notify commits");
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
||||
let wake = loop {
|
||||
match rx.try_recv() {
|
||||
Ok(Some(w)) => break w,
|
||||
Ok(None) => {}
|
||||
Err(e) => panic!("the receiver is open through a healthy listen: {e}"),
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"the committed notify delivered no wake within the bounded wait"
|
||||
);
|
||||
tokio::time::sleep(std::time::Duration::from_millis(25)).await;
|
||||
};
|
||||
assert_eq!(
|
||||
wake.channel, "wake-a",
|
||||
"the try_recv wake's channel field matches the listened channel"
|
||||
);
|
||||
|
||||
store
|
||||
.notify("wake-a", json!({"n": 3}))
|
||||
.await
|
||||
.expect("notify commits");
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
||||
let wake = loop {
|
||||
match rx
|
||||
.recv_timeout(std::time::Duration::from_millis(50))
|
||||
.await
|
||||
.expect("recv_timeout's Err is closed/storage-failure only, idle is Ok(None)")
|
||||
{
|
||||
Some(w) => break w,
|
||||
None => {
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"the committed notify delivered no wake within the bounded wait"
|
||||
);
|
||||
}
|
||||
}
|
||||
};
|
||||
assert_eq!(
|
||||
wake.channel, "wake-a",
|
||||
"the recv_timeout wake's channel field matches the listened channel"
|
||||
);
|
||||
|
||||
// The burst floor: three committed notifies, at least one wake.
|
||||
// The count is never pinned — coalescing is the documented engine
|
||||
// asymmetry (SQLite's 1-slot feed; pg per-notify), the contract
|
||||
// leaves the shape engine-owned.
|
||||
for n in 0..3 {
|
||||
store
|
||||
.notify("wake-a", json!({"burst": n}))
|
||||
.await
|
||||
.expect("notify commits");
|
||||
}
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
||||
let mut heard = 0;
|
||||
while heard == 0 {
|
||||
match rx
|
||||
.recv_timeout(std::time::Duration::from_millis(50))
|
||||
.await
|
||||
.expect("the receiver is open through a healthy listen")
|
||||
{
|
||||
Some(w) => {
|
||||
assert_eq!(
|
||||
w.channel, "wake-a",
|
||||
"the burst wake's channel field matches the listened channel"
|
||||
);
|
||||
heard += 1;
|
||||
}
|
||||
None => {
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"a burst of notifies yields at least one wake — never zero \
|
||||
within the bounded wait"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// No replay: a listener attached after a commit never sees that
|
||||
// commit's notify. The pre-settle sleep closes SQLite's baseline
|
||||
// race — the watcher's tick past the settle has consumed the
|
||||
// commit's version change (a past-window spurious wake arriving
|
||||
// before the late attach is the documented overtrigger; the pin
|
||||
// starts at the attach).
|
||||
store
|
||||
.notify("wake-b", json!({"late": true}))
|
||||
.await
|
||||
.expect("notify commits");
|
||||
tokio::time::sleep(std::time::Duration::from_millis(300)).await;
|
||||
let mut late = store
|
||||
.listen("wake-b")
|
||||
.await
|
||||
.expect("the late listen attaches");
|
||||
let heard = late
|
||||
.recv_timeout(std::time::Duration::from_millis(400))
|
||||
.await
|
||||
.expect("the absence window is Ok(None) or a wake — never a hang");
|
||||
match heard {
|
||||
None => {}
|
||||
Some(w) if w.channel == alkstore::RESERVED_LISTENER_RECONNECTED => {}
|
||||
Some(w) => panic!(
|
||||
"a listener attached after a commit never sees that commit's \
|
||||
notify — got a wake carrying {}",
|
||||
w.channel
|
||||
),
|
||||
}
|
||||
|
||||
// Channel-scoped delivery: a foreign-channel notify never wakes
|
||||
// this listener. The bounded window drains no wake carrying the
|
||||
// foreign channel; what may arrive is the listened channel (SQLite's
|
||||
// legal overtrigger) or the reserved reconnect straggler (pg).
|
||||
let mut scoped = store.listen("scope-a").await.expect("listen attaches");
|
||||
store
|
||||
.notify("scope-b", json!({"foreign": true}))
|
||||
.await
|
||||
.expect("notify commits");
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_millis(400);
|
||||
loop {
|
||||
let wake = scoped
|
||||
.recv_timeout(std::time::Duration::from_millis(100))
|
||||
.await
|
||||
.expect("the scoped listener is open");
|
||||
match wake {
|
||||
None => break,
|
||||
Some(w)
|
||||
if w.channel != "scope-a"
|
||||
&& w.channel != alkstore::RESERVED_LISTENER_RECONNECTED =>
|
||||
{
|
||||
panic!(
|
||||
"a wake from a different channel never arrives on this \
|
||||
listener: {}",
|
||||
w.channel
|
||||
);
|
||||
}
|
||||
Some(_) => {}
|
||||
}
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"the scoped silence window overflowed with unnamed wakes"
|
||||
);
|
||||
}
|
||||
|
||||
// Positive control: a same-channel notify still delivers — the
|
||||
// idle above was channel scoping, not a dead listener.
|
||||
store
|
||||
.notify("scope-a", json!({"own": true}))
|
||||
.await
|
||||
.expect("notify commits");
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(10);
|
||||
let wake = loop {
|
||||
match scoped
|
||||
.recv_timeout(std::time::Duration::from_millis(50))
|
||||
.await
|
||||
.expect("the scoped listener is open")
|
||||
{
|
||||
Some(w) => break w,
|
||||
None => {
|
||||
assert!(
|
||||
std::time::Instant::now() < deadline,
|
||||
"the same-channel positive control delivered no wake — \
|
||||
the listener is dead, not idle, and the scoping pin \
|
||||
above is vacuous"
|
||||
);
|
||||
}
|
||||
}
|
||||
};
|
||||
assert_eq!(
|
||||
wake.channel, "scope-a",
|
||||
"the control wake's channel field matches the listened channel"
|
||||
);
|
||||
|
||||
factory.teardown().await.expect("factory teardown");
|
||||
}
|
||||
|
||||
/// **`PayloadTooLarge` produced pg-side** — the Postgres engine
|
||||
/// rejects an oversized `notify`/`notify_tx` payload client-side with
|
||||
/// the typed `PayloadTooLarge` carrying the limit, **before any round
|
||||
|
||||
Reference in new issue
Block a user