From aa62930163e7069f49f7906a65269bb5602ebcf6 Mon Sep 17 00:00:00 2001 From: Dan Reynolds Date: Tue, 28 Jul 2026 13:00:57 -0400 Subject: [PATCH 1/5] Fix: a stream could emit nothing when a write raced its initial query MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A write that landed while a stream's initial query was in flight marked the entry dirty, and the initial result was then discarded in favour of a catch-up re-query. But `lastResultHash` had already been set from those same rows, so `selectIfChanged` compared against them, reported "unchanged", and emitted nothing — leaving the stream permanently silent, never emitting even its first value. Opening a stream and writing immediately is ordinary usage, and the documented contract is that a stream emits current results immediately. Always emit the initial result, then re-query if the entry was dirtied; the re-query still suppresses itself when nothing actually changed, so no duplicate emission. This was reproducing at ~33% as the long-standing write_coalescing "flake". It is not a flake: the regression test added here (seeded so the write reliably races the initial query) fails 5/5 without the fix and passes 6/6 with it, and that test file's comment claiming the shape "can't flake under load" is corrected to describe what it actually guards. Co-Authored-By: Claude Opus 4.8 --- lib/src/stream_engine.dart | 16 ++++++++--- test/stream_test.dart | 49 +++++++++++++++++++++++++++++++++ test/write_coalescing_test.dart | 8 ++++-- 3 files changed, 66 insertions(+), 7 deletions(-) diff --git a/lib/src/stream_engine.dart b/lib/src/stream_engine.dart index cf6c5e8c..5b71ec4c 100644 --- a/lib/src/stream_engine.dart +++ b/lib/src/stream_engine.dart @@ -307,13 +307,21 @@ final class StreamEngine { }; } - // If an invalidation occurred while performing the entry's initial query then the entry - // needs to be re-queried since its dependencies were not known at the time and this result could be stale. + // The initial result is a valid observation and the stream's contract + // is to emit current results immediately, so it is always emitted — + // even when a write dirtied this entry mid-flight. Skipping it used to + // strand the stream permanently: `lastResultHash` was already set from + // these rows, so the catch-up re-query below compared against them, + // reported "unchanged", and emitted nothing at all. + entry.emit(initialRows); + + // An invalidation that landed while the initial query was in flight + // arrived before this entry's dependencies were known, so re-query to + // pick up anything the initial result missed. `selectIfChanged` + // suppresses the emission if nothing actually changed. if (entry.dirty) { _requeryQueue.add(entry); _flushQueue(); - } else { - entry.emit(initialRows); } } catch (e, stackTrace) { // Propagate error to all subscribers so they don't hang. diff --git a/test/stream_test.dart b/test/stream_test.dart index e53082a7..3ee4419c 100644 --- a/test/stream_test.dart +++ b/test/stream_test.dart @@ -102,6 +102,55 @@ void main() { ); }); + test( + 'stream always emits its initial result even when a write lands mid-query', + () async { + // Regression: a write that dirtied an entry while its initial query was + // in flight used to suppress the initial emission in favour of a + // catch-up re-query — but `lastResultHash` was already set from those + // rows, so `selectIfChanged` reported "unchanged" and the stream + // emitted NOTHING, permanently. Writing immediately after opening a + // stream is ordinary usage, so the stream must still emit. + // Seed enough rows that the stream's initial query is still running + // when the writes below dispatch — that is the race window. + await db.executeBatch('INSERT INTO items(name, value) VALUES (?, ?)', [ + for (var i = 0; i < 40000; i++) ['seed_$i', i], + ]); + + final seen = []; + final sub = db + .stream('SELECT id, name, value FROM items') + .map((rows) => rows.length) + .listen(seen.add); + addTearDown(sub.cancel); + + // No await between opening the stream and writing: the writes race the + // initial query on purpose. + await Future.wait([ + for (var i = 0; i < 8; i++) + db.execute('INSERT INTO items(name, value) VALUES (?, ?)', [ + 'r_$i', + i, + ]), + ]); + + await Future.doWhile(() async { + if (seen.isNotEmpty && seen.last == 40008) return false; + await Future.delayed(const Duration(milliseconds: 10)); + return true; + }).timeout( + const Duration(seconds: 5), + onTimeout: () => fail( + 'stream never reached the post-write count; ' + 'emitted: $seen', + ), + ); + + expect(seen, isNotEmpty, reason: 'the initial result must be emitted'); + expect(seen.last, 40008); + }, + ); + tearDown(() async { await db.close(); if (await tempDir.exists()) { diff --git a/test/write_coalescing_test.dart b/test/write_coalescing_test.dart index d2b495a8..d12dd899 100644 --- a/test/write_coalescing_test.dart +++ b/test/write_coalescing_test.dart @@ -102,9 +102,11 @@ void main() { .stream('SELECT COUNT(*) AS c FROM items') .map((rows) => rows.first['c'] as int); - // emitsThrough waits for the post-burst count, tolerating the initial 0 - // and any intermediate emissions — no fixed-delay sleeps, so it can't - // flake under load. + // emitsThrough tolerates the initial count and any intermediate + // emissions. This shape is also a live guard on the initial-emission + // race: these writes dispatch while the stream's initial query is in + // flight, and a regression there strands the stream with no emissions + // at all (see stream_test's 'emits its initial result' case). final reachedEight = expectLater(counts, emitsThrough(8)); await Future.wait([ From 9acd90ab340c3d642270d32414b83b2d84001be2 Mon Sep 17 00:00:00 2001 From: Dan Reynolds Date: Tue, 28 Jul 2026 13:10:13 -0400 Subject: [PATCH 2/5] Exp 255: record the stream initial-emission race as a correctness experiment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Writeup, README row, and signal entry with two claims: 255.1 (the emission guarantee and why suppressing it stranded streams) and 255.2 (a 1-in-3 reproduction is a bug, not a flake — and a comment asserting otherwise redirects investigation). Co-Authored-By: Claude Opus 4.8 --- .../255-stream-initial-emission-race.md | 125 ++++++++++++++++++ experiments/index/255.json | 7 + experiments/signals/entries/255.json | 24 ++++ 3 files changed, 156 insertions(+) create mode 100644 experiments/255-stream-initial-emission-race.md create mode 100644 experiments/index/255.json create mode 100644 experiments/signals/entries/255.json diff --git a/experiments/255-stream-initial-emission-race.md b/experiments/255-stream-initial-emission-race.md new file mode 100644 index 00000000..ccf4d6e1 --- /dev/null +++ b/experiments/255-stream-initial-emission-race.md @@ -0,0 +1,125 @@ +# Experiment 255: A stream could emit nothing when a write raced its initial query + +**Date:** 2026-07-28 +**Status:** Accepted +**Category:** Correctness +**Direction:** `stream-rerun-dispatch` +**Benchmark Run:** none — correctness fix, guarded by a deterministic + regression test rather than a benchmark. + +## Problem + +`db.stream(sql)` promises to emit current results immediately and then re-emit +when writes change them. It could do neither: a stream whose initial query was +still in flight when a write landed went **permanently silent**, never emitting +even its first value. + +This surfaced as a long-standing intermittent failure in +`test/write_coalescing_test.dart` — "coalesced writes invalidate watching +streams" — which had been dismissed as a flake (including three times in one +session while shepherding unrelated PRs). It reproduced at roughly **1 in 3** +runs on a loaded machine, and the test file's own comment asserted the shape +"can't flake under load," which is precisely what discouraged looking closer. + +The failure output said what was actually happening, once read: + +``` +Which: emitted x Stream closed. + which never did emit an event that <8> +``` + +`emitted x` — nothing at all. Not a stale value: no value. + +## Mechanism + +`StreamEngine._createStream` registers an entry, adds it to +`_unknownDepsEntries` (so invalidations that arrive before its dependencies are +known still mark it dirty), and runs the initial query. On completion it stored +the result's hash and row count as the comparison baseline, then chose one of +two paths: + +```dart +entry.lastResultHash = initialHash; // baseline set from these rows +... +if (entry.dirty) { + _requeryQueue.add(entry); // catch-up re-query + _flushQueue(); +} else { + entry.emit(initialRows); // emit only when NOT dirtied +} +``` + +When a write dirtied the entry mid-flight, the initial rows were discarded +unemitted in favour of the catch-up re-query — but the baseline had already +been taken *from those same rows*. The re-query calls `selectIfChanged`, which +compares against that baseline, finds it identical, and returns +`rows == null` meaning "unchanged, nothing to emit" +([exp 075](075-native-hash-selectifchanged.md) established that short-circuit). +So the initial emission was suppressed as stale and the replacement emission +was suppressed as unchanged. The stream emitted nothing, forever. + +The race window is the initial query's duration, so it widens with result size: +trivial on a 4-row table, reliable on a 40k-row one. Opening a stream and +immediately writing is ordinary usage — seeding a list on first paint, for +example — which is why this was reachable at all. + +## Approach + +Always emit the initial result; it is a valid observation of the database at +the moment it was read, and the stream's contract is to deliver it. Then +re-query if the entry was dirtied, so anything the initial query missed still +arrives: + +```dart +entry.emit(initialRows); + +if (entry.dirty) { + _requeryQueue.add(entry); + _flushQueue(); +} +``` + +No duplicate-emission risk: the catch-up re-query still runs through +`selectIfChanged`, which suppresses itself when nothing actually changed. + +Two hypotheses were tried and discarded before this one — worth recording, +because both looked plausible and neither moved the failure rate: + +1. **Test-side registration race** (await the first emission before writing) — + made the *existing* test pass, but only by avoiding the window rather than + fixing anything; a fresh test written that way passed with and without the + library fix, guarding nothing. +2. **Stranded re-query queue** (`_flushQueue` never re-running after + `inFlight` cleared) — 3/6 failures, unchanged from baseline. + +Reading the actual failure text instead of theorising would have reached the +answer immediately; that is the transferable lesson. + +## Results + +A regression test in `test/stream_test.dart` seeds 40k rows so the write +reliably races the initial query, then asserts the stream emits at all: + +| | without fix | with fix | +|---|---|---| +| new regression test | **0 / 5 pass** | **6 / 6 pass** | +| `write_coalescing_test` (the old "flake") | 3 / 6 pass | 8 / 8 pass | + +A standalone reproduction on unmodified `main` versus this branch: + +``` +main: ✗ stream emitted NOTHING in 4s (permanently silent) +this branch: ✓ emissions: [40008] → final 40008 +``` + +Full stream/write/transaction suites: 140 tests pass. + +## Decision + +**Accepted.** A user-visible correctness fix: streams no longer drop their +initial emission, and the failure mode was silence rather than staleness, which +is the hardest kind to notice in an application. + +The `write_coalescing_test` comment claiming the shape "can't flake under load" +is corrected to describe what it actually guards — a false confidence marker is +worse than no comment, since it actively redirects investigation. diff --git a/experiments/index/255.json b/experiments/index/255.json new file mode 100644 index 00000000..e564c153 --- /dev/null +++ b/experiments/index/255.json @@ -0,0 +1,7 @@ +{ + "file": "255-stream-initial-emission-race.md", + "title": "Stream could emit nothing when a write raced its initial query", + "impact": "`db.stream()` promises to emit current results immediately, then re-emit on change. It could do neither: a stream whose initial query was still in flight when a write landed went **permanently silent**, never emitting even its first value. `StreamEngine._createStream` stored the initial result's hash as the comparison baseline and then, if the entry had been dirtied mid-flight, discarded those rows unemitted in favour of a catch-up re-query — but that re-query's `selectIfChanged` compared against the baseline taken from the very rows it replaced, found them identical, and returned 'unchanged, nothing to emit'. Initial emission suppressed as stale; replacement suppressed as unchanged; stream silent forever. The race window is the initial query's duration, so it is trivial on small tables and reliable on large ones — and opening a stream then immediately writing (seeding a list on first paint) is ordinary usage. Fix: always emit the initial result (a valid observation, and the documented contract), then re-query if dirtied; `selectIfChanged` still suppresses the follow-up when nothing changed, so no duplicate. This had been surfacing for a long time as an intermittent `write_coalescing_test` failure dismissed as a flake — at ~1-in-3 on a loaded machine — helped along by that test's own comment asserting the shape 'can't flake under load'. New regression test (40k seeded rows so the write reliably races the query) fails 0/5 without the fix and passes 6/6 with it; the old test goes 3/6 -> 8/8; 140 tests pass across the stream, write-coalescing, trigger-cascade, database and transaction suites.", + "status": "accepted", + "link": "" +} diff --git a/experiments/signals/entries/255.json b/experiments/signals/entries/255.json new file mode 100644 index 00000000..0db309d3 --- /dev/null +++ b/experiments/signals/entries/255.json @@ -0,0 +1,24 @@ +{ + "directions": ["stream-rerun-dispatch"], + "outcomeClass": "accepted", + "changedBeliefs": [ + "A stream's initial emission was not guaranteed: if a write dirtied the entry while its initial query was in flight, `_createStream` discarded the initial rows unemitted and relied on a catch-up re-query, which then compared against the hash baseline taken from those same discarded rows and reported 'unchanged'. Both emissions suppressed each other and the stream stayed permanently silent. The initial result is always emitted now, with the catch-up re-query following when the entry was dirtied.", + "The failure mode of this class of bug is silence, not staleness — the hardest kind for an application to notice, and the reason it survived so long behind an intermittent test failure." + ], + "claims": [ + { + "id": "255.1", + "text": "A stream always emits its initial result, even when a write dirties it mid-query; the catch-up re-query follows and self-suppresses via selectIfChanged when nothing changed. Suppressing the initial emission in favour of the re-query stranded streams permanently silent, because the re-query's baseline came from the very rows it replaced.", + "conditions": "M1 Pro · deterministic regression test (40k-row seed) · 2026-07" + }, + { + "id": "255.2", + "text": "An intermittent test failure reproducing at ~1-in-3 is a bug, not a flake. This one was dismissed three times in one session, partly because the test's own comment asserted the shape could not flake — a false confidence marker actively redirects investigation, and reading the actual failure text ('emitted x Stream closed') would have identified the mechanism immediately.", + "conditions": "methodology · 2026-07" + } + ], + "nextSignals": [ + "Stream emission is now guaranteed at registration; the remaining stream-rerun-dispatch work is throughput (exp 239's parked-read batching, preserved) rather than correctness.", + "Audit sibling paths for the same shape — any place that takes a comparison baseline from a result it then declines to emit will suppress both the original and its replacement. `selectIfChanged`'s short-circuit is only safe when the baseline came from rows the subscriber actually received." + ] +} From 591636714f867ae85aae38792ff7424e06eab904 Mon Sep 17 00:00:00 2001 From: Dan Reynolds Date: Tue, 28 Jul 2026 13:54:54 -0400 Subject: [PATCH 3/5] Exp 256: suppress the stale initial emission rather than emit-then-correct MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Exp 255 guaranteed a stream always emits; this settles which rows the first emission carries when a write raced the initial query. Measured both designs across race windows widened by seed size: time-to-correct (p50) A 8.1ms -> B 9.5ms at 20k; 18.5 -> 18.9 at 60k stale frames/stream A 1.00 -> B 0.00 at every raced size no-race control identical (13.3ms, 2 emissions) A's faster first emission measures the wrong clock: it delivers rows already known superseded, so the subscriber must disregard them anyway. B costs about a millisecond of time-to-truth and removes the list-jump artifact entirely. Shipping B: poison the comparison baseline when the entry is dirtied so the catch-up re-query is guaranteed to emit. This is not the exp 255 bug — that was suppression WITHOUT poisoning, which let both emissions cancel. Exp 255's writeup and claim are restated for the shipped semantics, and 255.1 now refines 256.1. Co-Authored-By: Claude Opus 4.8 --- .../stream_initial_emission_ab.dart | 124 ++++++++++++++++++ .../2026-07-28-exp256-initial-emission.md | 22 ++++ .../255-stream-initial-emission-race.md | 26 ++-- .../256-stale-initial-emission-tradeoff.md | 94 +++++++++++++ experiments/index/256.json | 7 + experiments/signals/entries/255.json | 16 ++- experiments/signals/entries/256.json | 19 +++ lib/src/stream_engine.dart | 34 +++-- 8 files changed, 318 insertions(+), 24 deletions(-) create mode 100644 benchmark/experiments/stream_initial_emission_ab.dart create mode 100644 benchmark/results/2026-07-28-exp256-initial-emission.md create mode 100644 experiments/256-stale-initial-emission-tradeoff.md create mode 100644 experiments/index/256.json create mode 100644 experiments/signals/entries/256.json diff --git a/benchmark/experiments/stream_initial_emission_ab.dart b/benchmark/experiments/stream_initial_emission_ab.dart new file mode 100644 index 00000000..8e667f2e --- /dev/null +++ b/benchmark/experiments/stream_initial_emission_ab.dart @@ -0,0 +1,124 @@ +// EXP-256: when a write races a stream's initial query, should the stream emit +// the known-stale initial rows immediately and correct itself, or wait and emit +// only the corrected result? +// +// Lane B shipped, so this harness measures it directly. To reproduce lane A, +// replace the poisoned-baseline branch in `StreamEngine._createStream` with a +// plain `entry.emit(initialRows)` before the re-query enqueue. +// +// The decision metric is NOT time-to-first-emission — lane A wins that by +// construction, by emitting data it already knows is superseded. What matters +// is time-to-CORRECT-value (when the subscriber can trust what it sees) and +// how many stale frames a UI renders before then. +import 'dart:async'; +import 'dart:io'; + +import 'package:resqlite/resqlite.dart'; + +const _samples = 12; +const _writes = 8; + +double _median(List xs) { + final s = [...xs]..sort(); + return s[s.length ~/ 2]; +} + +double _p90(List xs) { + final s = [...xs]..sort(); + return s[(0.9 * (s.length - 1)).round()]; +} + +class _Obs { + double? firstMs; + double? correctMs; + int emissions = 0; + int staleEmissions = 0; +} + +/// One observation: open a stream over [seedRows] rows, fire [_writes] inserts +/// without awaiting the first emission (so they race the initial query), and +/// record when the subscriber first sees anything and first sees the truth. +Future<_Obs> _observe(Database db, int seedRows, {required bool race}) async { + final expected = seedRows + _writes; + final obs = _Obs(); + final done = Completer(); + final sw = Stopwatch()..start(); + + final sub = db.stream('SELECT id, name FROM items').listen((rows) { + final t = sw.elapsedMicroseconds / 1000.0; + obs.emissions++; + obs.firstMs ??= t; + if (rows.length != expected) obs.staleEmissions++; + if (rows.length == expected && obs.correctMs == null) { + obs.correctMs = t; + if (!done.isCompleted) done.complete(); + } + }); + + if (!race) { + // Control: let the stream register and emit before writing at all. + while (obs.emissions == 0) { + await Future.delayed(const Duration(milliseconds: 1)); + } + } + await Future.wait([ + for (var i = 0; i < _writes; i++) + db.execute('INSERT INTO items(name) VALUES (?)', ['w_$i']), + ]); + + await done.future.timeout( + const Duration(seconds: 10), + onTimeout: () => throw StateError('never reached the correct value'), + ); + await sub.cancel(); + await db.execute('DELETE FROM items WHERE name LIKE ?', ['w_%']); + return obs; +} + +Future main() async { + const lane = 'B suppress-stale (shipped)'; + final tmp = await Directory.systemTemp.createTemp('resqlite-exp256-'); + final db = await Database.open('${tmp.path}/s.db'); + await db.execute('CREATE TABLE items(id INTEGER PRIMARY KEY, name TEXT)'); + + stdout.writeln( + '| lane | shape | first emit ms | correct ms (p50/p90) | emissions | stale frames |', + ); + stdout.writeln('|---|---|---:|---:|---:|---:|'); + + for (final (label, seed, race) in [ + ('racing · 1k rows', 1000, true), + ('racing · 20k rows', 20000, true), + ('racing · 60k rows', 60000, true), + ('no race · 20k rows (control)', 20000, false), + ]) { + await db.execute('DELETE FROM items'); + await db.executeBatch('INSERT INTO items(name) VALUES (?)', [ + for (var i = 0; i < seed; i++) ['seed_$i'], + ]); + + // Warm the pool and statement caches. + for (var w = 0; w < 3; w++) { + await _observe(db, seed, race: race); + } + + final firsts = [], corrects = []; + var emissions = 0, stale = 0; + for (var i = 0; i < _samples; i++) { + final o = await _observe(db, seed, race: race); + firsts.add(o.firstMs!); + corrects.add(o.correctMs!); + emissions += o.emissions; + stale += o.staleEmissions; + } + stdout.writeln( + '| $lane | $label | ${_median(firsts).toStringAsFixed(1)} ' + '| ${_median(corrects).toStringAsFixed(1)} / ${_p90(corrects).toStringAsFixed(1)} ' + '| ${(emissions / _samples).toStringAsFixed(2)} ' + '| ${(stale / _samples).toStringAsFixed(2)} |', + ); + } + + await db.close(); + await tmp.delete(recursive: true); +} diff --git a/benchmark/results/2026-07-28-exp256-initial-emission.md b/benchmark/results/2026-07-28-exp256-initial-emission.md new file mode 100644 index 00000000..16019336 --- /dev/null +++ b/benchmark/results/2026-07-28-exp256-initial-emission.md @@ -0,0 +1,22 @@ +# Exp 256 — stale-then-correct vs suppress-stale initial emission + +Harness: `benchmark/experiments/stream_initial_emission_ab.dart`. Apple M1 Pro. +Stream opened over a seeded table; 8 inserts fired without awaiting the first +emission so they race the initial query. 12 samples per shape after 3 warmup +observations. Lane A reproduced by replacing the poisoned-baseline branch in +`StreamEngine._createStream` with a plain `entry.emit(initialRows)`. + +| lane | shape | first emit ms | correct ms (p50/p90) | emissions | stale frames | +|---|---|---:|---:|---:|---:| +| A emit-then-correct | racing · 1k | 0.3 | 0.9 / 2.0 | 2.08 | 1.08 | +| A emit-then-correct | racing · 20k | 3.1 | 8.1 / 10.0 | 2.00 | 1.00 | +| A emit-then-correct | racing · 60k | 8.1 | 18.5 / 22.9 | 2.00 | 1.00 | +| A emit-then-correct | no race · 20k (control) | 3.4 | 13.3 / 15.1 | 2.00 | 1.00 | +| B suppress-stale | racing · 1k | 0.9 | 1.1 / 2.5 | 1.42 | 0.42 | +| B suppress-stale | racing · 20k | 9.5 | 9.5 / 15.8 | 1.00 | 0.00 | +| B suppress-stale | racing · 60k | 18.9 | 18.9 / 21.4 | 1.00 | 0.00 | +| B suppress-stale | no race · 20k (control) | 2.9 | 13.3 / 19.4 | 2.00 | 1.00 | + +B shipped: ~1 ms of time-to-correct-value buys the removal of every stale +frame, and the unraced control is identical. See +`experiments/256-stale-initial-emission-tradeoff.md`. diff --git a/experiments/255-stream-initial-emission-race.md b/experiments/255-stream-initial-emission-race.md index ccf4d6e1..8f240310 100644 --- a/experiments/255-stream-initial-emission-race.md +++ b/experiments/255-stream-initial-emission-race.md @@ -65,22 +65,32 @@ example — which is why this was reachable at all. ## Approach -Always emit the initial result; it is a valid observation of the database at -the moment it was read, and the stream's contract is to deliver it. Then -re-query if the entry was dirtied, so anything the initial query missed still -arrives: +The stream must always resolve to an emission — but *which* rows it emits when +a write raced is a second question, and the two answers were measured against +each other in [exp 256](256-stale-initial-emission-tradeoff.md) before choosing. +The shipped behaviour poisons the comparison baseline so the catch-up re-query +is guaranteed to emit, and lets that emission be the stream's first: ```dart -entry.emit(initialRows); - if (entry.dirty) { + entry.lastResultHash = _poisonedHash; // guarantees the re-query emits + entry.lastRowCount = -1; _requeryQueue.add(entry); _flushQueue(); +} else { + entry.emit(initialRows); } ``` -No duplicate-emission risk: the catch-up re-query still runs through -`selectIfChanged`, which suppresses itself when nothing actually changed. +The poisoning is what makes suppression safe, and is exactly what the bug was +missing: suppressing the initial emission while leaving the baseline set to +those same rows is what let both emissions cancel each other. `_requery` +propagates errors to subscribers, so the stream always resolves to a value or +an error — never silence. + +The alternative — emit the known-stale rows immediately, then correct — is +simpler and also fixes the silence, but exp 256 measured it rendering exactly +one stale frame on every raced stream while saving ~1 ms of time-to-correct. Two hypotheses were tried and discarded before this one — worth recording, because both looked plausible and neither moved the failure rate: diff --git a/experiments/256-stale-initial-emission-tradeoff.md b/experiments/256-stale-initial-emission-tradeoff.md new file mode 100644 index 00000000..bcf5ef83 --- /dev/null +++ b/experiments/256-stale-initial-emission-tradeoff.md @@ -0,0 +1,94 @@ +# Experiment 256: Emit stale-then-correct, or wait and emit only the truth? + +**Date:** 2026-07-28 +**Status:** Accepted +**Direction:** `stream-rerun-dispatch` +**Benchmark Run:** none — focused + [`benchmark/experiments/stream_initial_emission_ab.dart`](../benchmark/experiments/stream_initial_emission_ab.dart); + raw table in + [`benchmark/results/2026-07-28-exp256-initial-emission.md`](../benchmark/results/2026-07-28-exp256-initial-emission.md). + +## Problem + +[Exp 255](255-stream-initial-emission-race.md) fixed a stream that could go +permanently silent when a write raced its initial query. But it left a design +question the bug had obscured: once you guarantee an emission, *which rows* +should the first one carry? + +- **A — emit-then-correct.** Emit the initial rows immediately even though a + write has already superseded them, then correct via the catch-up re-query. +- **B — suppress-stale.** Skip that emission, poison the comparison baseline so + the re-query is guaranteed to emit, and let the corrected result be the + stream's first. + +Both fix the silence. A is unconditional by construction, which is why it +shipped first. B is what the stream's consumers arguably want — but it keeps +the "suppress the initial emission" shape that caused the bug, so it needed +evidence rather than preference. + +## Hypothesis + +A's advantage — a faster first emission — is measured against the wrong clock. +What a subscriber cares about is **time-to-correct-value**: when it can trust +what it is looking at. Since the catch-up re-query runs in both designs, B +should cost little on that clock while removing the stale frame entirely. + +## Approach + +The harness opens a stream over a seeded table and fires eight inserts +*without* awaiting the first emission, so the writes race the initial query. +Per observation it records time-to-first-emission, time-to-correct-value, +emission count, and how many emissions carried a row count that was already +wrong. Seed sizes of 1k / 20k / 60k widen the race window (the window is the +initial query's duration), plus a no-race control that must be identical +between lanes. 12 samples per shape after warmup. + +## Results + +Medians, Apple M1 Pro: + +| shape | metric | A emit-then-correct | B suppress-stale | +|---|---|---:|---:| +| racing · 20k | time to correct (p50 / p90) | 8.1 / 10.0 ms | **9.5 / 15.8 ms** | +| racing · 20k | stale frames per stream | 1.00 | **0.00** | +| racing · 60k | time to correct (p50 / p90) | 18.5 / 22.9 ms | **18.9 / 21.4 ms** | +| racing · 60k | stale frames per stream | 1.00 | **0.00** | +| racing · 1k | stale frames per stream | 1.08 | **0.42** | +| no race · 20k (control) | time to correct | 13.3 ms | 13.3 ms | +| no race · 20k (control) | emissions | 2.00 | 2.00 | + +**B costs ~1.4 ms at 20k rows and ~0.4 ms at 60k on time-to-correct-value** — +far less than a full extra query, because the re-query is already in flight in +both lanes; B simply declines to paint while it waits. + +**B removes the stale frame completely** (1.00 → 0.00 at every raced size). A +renders exactly one wrong frame on every raced stream — the list-jump artifact +a reactive stream exists to prevent. + +**A's fast first emission measures the wrong thing.** Its 3.1 ms first emission +at 20k delivers rows already known to be superseded; the subscriber must +disregard them for another 5 ms. + +**The control is identical** — unraced streams behave the same in both lanes +(13.3 ms, 2 emissions), confirming the change is scoped to the raced path. + +At 1k rows the race window is small enough that it often does not land at all, +which is why B still shows 0.42 stale frames there: those are the samples where +the writes completed before the initial query started, so both lanes correctly +emit the pre-write state first. + +## Decision + +**Accepted — ship B.** It is better on the metric that matters (never showing +known-stale data), costs about a millisecond on time-to-truth, and is identical +when nothing races. + +Crucially it is *not* a return to the exp 255 bug: the poisoned baseline +guarantees the re-query emits, and `_requery` propagates errors to subscribers, +so the stream always resolves to a value or an error. The bug was suppression +*without* poisoning. + +Would revisit if a workload cared more about first-paint latency than +correctness of first paint — a skeleton-UI pattern that would rather show stale +rows than nothing. That is a public-API question (an opt-in emission policy), +not a default worth changing. diff --git a/experiments/index/256.json b/experiments/index/256.json new file mode 100644 index 00000000..a55e022f --- /dev/null +++ b/experiments/index/256.json @@ -0,0 +1,7 @@ +{ + "file": "256-stale-initial-emission-tradeoff.md", + "title": "Stale-then-correct vs suppress-stale initial stream emission", + "impact": "Exp 255 guaranteed a stream always emits; this settles WHICH rows the first emission carries when a write raced the initial query. Lane A emits the known-superseded initial rows immediately and corrects via the catch-up re-query; lane B suppresses that emission, poisons the comparison baseline so the re-query is guaranteed to emit, and lets the corrected result be the stream's first. A's advantage — a faster first emission — is measured against the wrong clock: what a subscriber cares about is time-to-CORRECT-value, and the catch-up re-query runs in both lanes. Measured across race windows widened by seed size: B costs **~1.4 ms at 20k rows and ~0.4 ms at 60k** on time-to-correct (8.1 -> 9.5 and 18.5 -> 18.9 p50) and removes the stale frame **entirely** (1.00 -> 0.00 per raced stream at every size); A renders exactly one wrong frame on every raced stream, the list-jump artifact a reactive stream exists to prevent. The no-race control is identical in both lanes (13.3 ms, 2 emissions), confirming the change is scoped to the raced path. **Accepted: ship B.** It is not a return to the exp 255 bug — that was suppression WITHOUT poisoning, which let the initial emission and its replacement cancel each other; the sentinel baseline guarantees an emission and `_requery` propagates errors, so a stream always resolves to a value or an error. Would revisit only for a skeleton-UI workload that prefers stale rows to nothing, which is an opt-in public-API question rather than a default.", + "status": "accepted", + "link": "" +} diff --git a/experiments/signals/entries/255.json b/experiments/signals/entries/255.json index 0db309d3..6f1a43be 100644 --- a/experiments/signals/entries/255.json +++ b/experiments/signals/entries/255.json @@ -1,15 +1,23 @@ { - "directions": ["stream-rerun-dispatch"], + "directions": [ + "stream-rerun-dispatch" + ], "outcomeClass": "accepted", "changedBeliefs": [ - "A stream's initial emission was not guaranteed: if a write dirtied the entry while its initial query was in flight, `_createStream` discarded the initial rows unemitted and relied on a catch-up re-query, which then compared against the hash baseline taken from those same discarded rows and reported 'unchanged'. Both emissions suppressed each other and the stream stayed permanently silent. The initial result is always emitted now, with the catch-up re-query following when the entry was dirtied.", + "A stream's initial emission was not guaranteed: if a write dirtied the entry while its initial query was in flight, `_createStream` discarded the initial rows unemitted and relied on a catch-up re-query, which then compared against the hash baseline taken from those same discarded rows and reported 'unchanged'. Both emissions suppressed each other and the stream stayed permanently silent. The fix keeps the suppression (exp 256 measured it as the better semantics) but poisons the baseline so the re-query always emits.", "The failure mode of this class of bug is silence, not staleness — the hardest kind for an application to notice, and the reason it survived so long behind an intermittent test failure." ], "claims": [ { "id": "255.1", - "text": "A stream always emits its initial result, even when a write dirties it mid-query; the catch-up re-query follows and self-suppresses via selectIfChanged when nothing changed. Suppressing the initial emission in favour of the re-query stranded streams permanently silent, because the re-query's baseline came from the very rows it replaced.", - "conditions": "M1 Pro · deterministic regression test (40k-row seed) · 2026-07" + "text": "A stream always resolves to an emission or an error, never silence. When a write dirties an entry mid-initial-query the comparison baseline is poisoned so the catch-up re-query is guaranteed to emit, and that emission is the stream's first. Suppressing the initial emission while leaving the baseline set to the rows it replaced is what stranded streams permanently silent.", + "conditions": "M1 Pro · deterministic regression test (40k-row seed) · 2026-07", + "edges": [ + { + "type": "refines", + "target": "256.1" + } + ] }, { "id": "255.2", diff --git a/experiments/signals/entries/256.json b/experiments/signals/entries/256.json new file mode 100644 index 00000000..ac710ab7 --- /dev/null +++ b/experiments/signals/entries/256.json @@ -0,0 +1,19 @@ +{ + "directions": ["stream-rerun-dispatch"], + "outcomeClass": "accepted", + "changedBeliefs": [ + "When a write races a stream's initial query, the stream suppresses the known-stale initial rows and emits only the corrected result. Measured cost is ~1.4 ms at 20k rows and ~0.4 ms at 60k on time-to-correct-value, because the catch-up re-query is already in flight in either design; the benefit is removing the one stale frame that emit-then-correct rendered on every raced stream.", + "Time-to-first-emission is the wrong metric for a reactive stream when the first emission is known to be superseded. The metric that matters is time-to-correct-value — when the subscriber can trust what it sees — and on that clock the two designs are within about a millisecond." + ], + "claims": [ + { + "id": "256.1", + "text": "A stream never emits rows it already knows a write superseded: when the initial query is dirtied mid-flight the comparison baseline is poisoned and only the corrected result is emitted. This costs ~0.4-1.4 ms of time-to-correct-value and removes the stale frame that emit-then-correct rendered on every raced stream; unraced streams are unaffected.", + "conditions": "M1 Pro · 1k/20k/60k seed race windows · 12 samples/shape · 2026-07" + } + ], + "nextSignals": [ + "Stream emission semantics are settled: always resolves (exp 255), never stale (exp 256). Remaining stream-rerun-dispatch work is throughput — exp 239's parked-read batching, preserved but unshipped — not correctness.", + "A skeleton-UI workload that prefers stale rows to no rows would want the opposite policy. That is an opt-in public-API question (an emission-policy parameter on `stream()`), not a default to flip; it needs a real consumer asking before it is worth the surface area." + ] +} diff --git a/lib/src/stream_engine.dart b/lib/src/stream_engine.dart index 5b71ec4c..27110cac 100644 --- a/lib/src/stream_engine.dart +++ b/lib/src/stream_engine.dart @@ -47,6 +47,10 @@ import 'extensions/set.dart'; /// deduplication, initial query with dependency tracking, write /// invalidation, re-query with result-change detection, and /// per-subscriber buffered delivery. +/// Sentinel comparison baseline that no real result hash equals, so a +/// catch-up re-query is guaranteed to report a change and emit. +const int _poisonedHash = 0; + final class StreamEngine { StreamEngine(this._pool); @@ -307,21 +311,27 @@ final class StreamEngine { }; } - // The initial result is a valid observation and the stream's contract - // is to emit current results immediately, so it is always emitted — - // even when a write dirtied this entry mid-flight. Skipping it used to - // strand the stream permanently: `lastResultHash` was already set from - // these rows, so the catch-up re-query below compared against them, - // reported "unchanged", and emitted nothing at all. - entry.emit(initialRows); - - // An invalidation that landed while the initial query was in flight - // arrived before this entry's dependencies were known, so re-query to - // pick up anything the initial result missed. `selectIfChanged` - // suppresses the emission if nothing actually changed. + // A write landed while this initial query was in flight, so these rows + // are already known to be superseded. Don't paint them: poison the + // comparison baseline so the catch-up re-query is guaranteed to report + // a change, and let that emission be the stream's first. + // + // Poisoning is what makes this safe. Suppressing the initial emission + // while leaving the baseline set to these rows is the exp 255 bug: the + // re-query then compares against the very rows it replaced, reports + // "unchanged", and the stream goes permanently silent. The sentinel + // guarantees an emission, and `_requery` propagates errors to + // subscribers, so the stream always resolves to a value or an error. + // + // Measured (exp 256): this costs ~1 ms of time-to-correct-value and + // removes the one stale frame lane A rendered every time. if (entry.dirty) { + entry.lastResultHash = _poisonedHash; + entry.lastRowCount = -1; _requeryQueue.add(entry); _flushQueue(); + } else { + entry.emit(initialRows); } } catch (e, stackTrace) { // Propagate error to all subscribers so they don't hang. From 9ba9461a29ea500054ecbd75444b9f5fa5b15fd9 Mon Sep 17 00:00:00 2001 From: Dan Reynolds Date: Tue, 28 Jul 2026 14:17:35 -0400 Subject: [PATCH 4/5] =?UTF-8?q?Review:=20poison=20the=20row=20count,=20not?= =?UTF-8?q?=20the=20hash=20=E2=80=94=200=20is=20a=20real=20hash?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review pass on the folded PR found the sentinel was wrong. `selectIfChanged` reports "unchanged" only when hash AND row count both match, so poisoning either suffices — but only the row count has an impossible value. An empty result hashes to 0 (`_finishInitialHash` short-circuits on rowCount == 0), so a hash sentinel of 0 is a legitimate baseline, not an unreachable one: a stream over a query matching no rows would compare 0 against 0, report "unchanged", and go silent — exactly the exp 255 bug, reachable only for empty results. The shipped code already set both, so behaviour was correct, but the comment credited the hash for a guarantee the row count was providing. Anyone simplifying by dropping the row count would have reintroduced the bug for empty results with no test to catch it. Now poisons the row count alone, named `_neverMatchingRowCount` and documented with why a hash sentinel cannot work. Adds an empty-result regression test: 0/3 against the hash-only sentinel, 4/4 against this one. Recorded as claim 255.3 with the general lesson — a forced-mismatch sentinel needs a value the compared domain cannot produce. Co-Authored-By: Claude Opus 4.8 --- .../255-stream-initial-emission-race.md | 13 +++++- experiments/signals/entries/255.json | 8 +++- lib/src/stream_engine.dart | 31 ++++++++------ test/stream_test.dart | 41 +++++++++++++++++++ 4 files changed, 78 insertions(+), 15 deletions(-) diff --git a/experiments/255-stream-initial-emission-race.md b/experiments/255-stream-initial-emission-race.md index 8f240310..bb290a7b 100644 --- a/experiments/255-stream-initial-emission-race.md +++ b/experiments/255-stream-initial-emission-race.md @@ -73,8 +73,7 @@ is guaranteed to emit, and lets that emission be the stream's first: ```dart if (entry.dirty) { - entry.lastResultHash = _poisonedHash; // guarantees the re-query emits - entry.lastRowCount = -1; + entry.lastRowCount = _neverMatchingRowCount; // guarantees the re-query emits _requeryQueue.add(entry); _flushQueue(); } else { @@ -88,6 +87,16 @@ those same rows is what let both emissions cancel each other. `_requery` propagates errors to subscribers, so the stream always resolves to a value or an error — never silence. +**It has to be the row count, not the hash.** `selectIfChanged` reports +"unchanged" only when the hash *and* the row count both match, so poisoning +either one suffices — but only the row count has an impossible value. An empty +result hashes to **0** (`_finishInitialHash` short-circuits on `rowCount == 0`), +so a hash sentinel of 0 is a legitimate baseline rather than an unreachable one, +and a stream over a query matching no rows would go silent exactly as before. +A review pass caught this in the first version of the fix; the empty-result +regression test in `stream_test.dart` fails 0/3 against the hash-only sentinel +and passes 4/4 against the row-count one. + The alternative — emit the known-stale rows immediately, then correct — is simpler and also fixes the silence, but exp 256 measured it rendering exactly one stale frame on every raced stream while saving ~1 ms of time-to-correct. diff --git a/experiments/signals/entries/255.json b/experiments/signals/entries/255.json index 6f1a43be..08681a21 100644 --- a/experiments/signals/entries/255.json +++ b/experiments/signals/entries/255.json @@ -23,10 +23,16 @@ "id": "255.2", "text": "An intermittent test failure reproducing at ~1-in-3 is a bug, not a flake. This one was dismissed three times in one session, partly because the test's own comment asserted the shape could not flake — a false confidence marker actively redirects investigation, and reading the actual failure text ('emitted x Stream closed') would have identified the mechanism immediately.", "conditions": "methodology · 2026-07" + }, + { + "id": "255.3", + "text": "Poisoning a stream's comparison baseline must use the row count, not the hash: selectIfChanged requires both to match, and an empty result legitimately hashes to 0, so a hash sentinel of 0 is reachable rather than impossible and would leave empty-result streams silent. A real row count is always >= 0, so -1 can never match.", + "conditions": "review finding · guarded by the empty-result regression test · 2026-07" } ], "nextSignals": [ "Stream emission is now guaranteed at registration; the remaining stream-rerun-dispatch work is throughput (exp 239's parked-read batching, preserved) rather than correctness.", - "Audit sibling paths for the same shape — any place that takes a comparison baseline from a result it then declines to emit will suppress both the original and its replacement. `selectIfChanged`'s short-circuit is only safe when the baseline came from rows the subscriber actually received." + "Audit sibling paths for the same shape — any place that takes a comparison baseline from a result it then declines to emit will suppress both the original and its replacement. `selectIfChanged`'s short-circuit is only safe when the baseline came from rows the subscriber actually received.", + "Any future 'force this comparison to report a change' needs a value the compared domain cannot produce. Row counts have one (-1); hashes do not, because the empty-result short-circuit makes 0 meaningful. Check the domain before picking a sentinel." ] } diff --git a/lib/src/stream_engine.dart b/lib/src/stream_engine.dart index 27110cac..b0f7052c 100644 --- a/lib/src/stream_engine.dart +++ b/lib/src/stream_engine.dart @@ -47,9 +47,16 @@ import 'extensions/set.dart'; /// deduplication, initial query with dependency tracking, write /// invalidation, re-query with result-change detection, and /// per-subscriber buffered delivery. -/// Sentinel comparison baseline that no real result hash equals, so a -/// catch-up re-query is guaranteed to report a change and emit. -const int _poisonedHash = 0; +/// Row count that no real result can have, used to poison a stream's +/// comparison baseline so its next re-query is guaranteed to report a change. +/// +/// The row count — not the hash — is what makes this safe. `selectIfChanged` +/// treats a result as unchanged only when the hash *and* the row count both +/// match, and a real count is always >= 0, so this value can never match. A +/// hash sentinel would not work: an empty result hashes to 0 (see +/// `_finishInitialHash`), so 0 is a legitimate value rather than an impossible +/// one, and a stream over an empty query would go silent again. +const int _neverMatchingRowCount = -1; final class StreamEngine { StreamEngine(this._pool); @@ -316,18 +323,18 @@ final class StreamEngine { // comparison baseline so the catch-up re-query is guaranteed to report // a change, and let that emission be the stream's first. // - // Poisoning is what makes this safe. Suppressing the initial emission - // while leaving the baseline set to these rows is the exp 255 bug: the - // re-query then compares against the very rows it replaced, reports - // "unchanged", and the stream goes permanently silent. The sentinel - // guarantees an emission, and `_requery` propagates errors to - // subscribers, so the stream always resolves to a value or an error. + // Poisoning is what makes suppression safe. Suppressing the initial + // emission while leaving the baseline set to these rows is the exp 255 + // bug: the re-query then compares against the very rows it replaced, + // reports "unchanged", and the stream goes permanently silent. With the + // baseline poisoned the re-query always emits, and `_requery` + // propagates errors to subscribers, so a stream always resolves to a + // value or an error. // // Measured (exp 256): this costs ~1 ms of time-to-correct-value and - // removes the one stale frame lane A rendered every time. + // removes the one stale frame emit-then-correct rendered every time. if (entry.dirty) { - entry.lastResultHash = _poisonedHash; - entry.lastRowCount = -1; + entry.lastRowCount = _neverMatchingRowCount; _requeryQueue.add(entry); _flushQueue(); } else { diff --git a/test/stream_test.dart b/test/stream_test.dart index 3ee4419c..c687b5cd 100644 --- a/test/stream_test.dart +++ b/test/stream_test.dart @@ -151,6 +151,47 @@ void main() { }, ); + test( + 'stream over an empty result still emits when a write races it', + () async { + // An empty result hashes to 0, so 0 is a legitimate baseline rather + // than an impossible one. Poisoning the baseline by hash alone would + // leave this stream comparing 0 against 0, reporting "unchanged", and + // going silent — the exp 255 failure, reachable only when the query + // matches nothing. The row-count sentinel is what rules it out. + await db.executeBatch('INSERT INTO items(name, value) VALUES (?, ?)', [ + for (var i = 0; i < 40000; i++) ['seed_$i', i], + ]); + + final seen = []; + final sub = db + .stream("SELECT id FROM items WHERE name = 'no_such_row'") + .map((rows) => rows.length) + .listen(seen.add); + addTearDown(sub.cancel); + + // Races the initial query, and leaves the result set still empty. + await Future.wait([ + for (var i = 0; i < 8; i++) + db.execute('INSERT INTO items(name, value) VALUES (?, ?)', [ + 'other_$i', + i, + ]), + ]); + + await Future.doWhile(() async { + if (seen.isNotEmpty) return false; + await Future.delayed(const Duration(milliseconds: 10)); + return true; + }).timeout( + const Duration(seconds: 5), + onTimeout: () => fail('empty-result stream never emitted'), + ); + + expect(seen.first, 0, reason: 'an empty result is still a result'); + }, + ); + tearDown(() async { await db.close(); if (await tempDir.exists()) { From c207f3878318618e6d4a91fba3fc38a4a1da09b2 Mon Sep 17 00:00:00 2001 From: Dan Reynolds Date: Tue, 28 Jul 2026 14:27:05 -0400 Subject: [PATCH 5/5] Use a null baseline instead of a magic row count MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `entry.lastRowCount = null` says what it means — no baseline to compare against — where `_neverMatchingRowCount = -1` needed a comment to explain that -1 was chosen because real counts are non-negative. The row count is nullable end to end now (StreamEntry, SelectIfChangedRequest, the comparison), and null cannot collide with any real observation, so the guarantee holds by construction rather than by argument. This also retires a magic -1 that predated this work: the request type already documented "-1 if unknown", so the codebase ends with one fewer sentinel than it started with rather than one more. Behaviour unchanged: 141 tests pass, both regression tests 10/10, and the empty-result guard still fails 0/3 against a hash-only sentinel. Co-Authored-By: Claude Opus 4.8 --- .../255-stream-initial-emission-race.md | 25 ++++++++------ experiments/signals/entries/255.json | 4 +-- lib/src/reader/read_worker.dart | 9 ++--- lib/src/reader/reader_pool.dart | 2 +- lib/src/stream_engine.dart | 33 ++++++++----------- 5 files changed, 36 insertions(+), 37 deletions(-) diff --git a/experiments/255-stream-initial-emission-race.md b/experiments/255-stream-initial-emission-race.md index bb290a7b..a658186c 100644 --- a/experiments/255-stream-initial-emission-race.md +++ b/experiments/255-stream-initial-emission-race.md @@ -73,7 +73,7 @@ is guaranteed to emit, and lets that emission be the stream's first: ```dart if (entry.dirty) { - entry.lastRowCount = _neverMatchingRowCount; // guarantees the re-query emits + entry.lastRowCount = null; // no baseline, so the re-query always emits _requeryQueue.add(entry); _flushQueue(); } else { @@ -87,15 +87,20 @@ those same rows is what let both emissions cancel each other. `_requery` propagates errors to subscribers, so the stream always resolves to a value or an error — never silence. -**It has to be the row count, not the hash.** `selectIfChanged` reports -"unchanged" only when the hash *and* the row count both match, so poisoning -either one suffices — but only the row count has an impossible value. An empty -result hashes to **0** (`_finishInitialHash` short-circuits on `rowCount == 0`), -so a hash sentinel of 0 is a legitimate baseline rather than an unreachable one, -and a stream over a query matching no rows would go silent exactly as before. -A review pass caught this in the first version of the fix; the empty-result -regression test in `stream_test.dart` fails 0/3 against the hash-only sentinel -and passes 4/4 against the row-count one. +**It has to be the row count, and it is expressed as `null`.** +`selectIfChanged` reports "unchanged" only when the hash *and* the row count +both match, so clearing either suffices — but the hash has no unreachable +value. An empty result hashes to **0** (`_finishInitialHash` short-circuits on +`rowCount == 0`), so a hash sentinel of 0 is a legitimate baseline rather than +an impossible one, and a stream over a query matching no rows would go silent +exactly as before. A review pass caught that in the first version of this fix. + +The row count is now nullable end to end — `StreamEntry`, the +`SelectIfChangedRequest`, and the comparison — with `null` meaning "no baseline +to compare against". That replaced a magic `-1` that predated this work +(the request type documented "`-1` if unknown"), so the codebase carries one +fewer sentinel than before rather than one more. The empty-result regression +test fails 0/3 against a hash-only sentinel and passes 10/10 against this. The alternative — emit the known-stale rows immediately, then correct — is simpler and also fixes the silence, but exp 256 measured it rendering exactly diff --git a/experiments/signals/entries/255.json b/experiments/signals/entries/255.json index 08681a21..833836aa 100644 --- a/experiments/signals/entries/255.json +++ b/experiments/signals/entries/255.json @@ -26,13 +26,13 @@ }, { "id": "255.3", - "text": "Poisoning a stream's comparison baseline must use the row count, not the hash: selectIfChanged requires both to match, and an empty result legitimately hashes to 0, so a hash sentinel of 0 is reachable rather than impossible and would leave empty-result streams silent. A real row count is always >= 0, so -1 can never match.", + "text": "Clearing a stream's comparison baseline must clear the row count, not the hash: selectIfChanged requires both to match, and an empty result legitimately hashes to 0, so a hash sentinel of 0 is reachable rather than impossible and would leave empty-result streams silent. The row count is nullable end to end, with null meaning 'no baseline' — which also retired a pre-existing magic -1 in the request type.", "conditions": "review finding · guarded by the empty-result regression test · 2026-07" } ], "nextSignals": [ "Stream emission is now guaranteed at registration; the remaining stream-rerun-dispatch work is throughput (exp 239's parked-read batching, preserved) rather than correctness.", "Audit sibling paths for the same shape — any place that takes a comparison baseline from a result it then declines to emit will suppress both the original and its replacement. `selectIfChanged`'s short-circuit is only safe when the baseline came from rows the subscriber actually received.", - "Any future 'force this comparison to report a change' needs a value the compared domain cannot produce. Row counts have one (-1); hashes do not, because the empty-result short-circuit makes 0 meaningful. Check the domain before picking a sentinel." + "Prefer absence over a magic value when forcing a comparison to report a change: null cannot collide with any real observation, whereas a sentinel must be a value the domain cannot produce — and hashes have no such value here, since the empty-result short-circuit makes 0 meaningful." ] } diff --git a/lib/src/reader/read_worker.dart b/lib/src/reader/read_worker.dart index 9c58069b..93b70d51 100644 --- a/lib/src/reader/read_worker.dart +++ b/lib/src/reader/read_worker.dart @@ -65,9 +65,10 @@ final class SelectIfChangedRequest extends ReadRequest { }); final int lastResultHash; - /// Previously-emitted row count, or `-1` if unknown. Compared with the - /// fresh count alongside the canonical result hash. - final int lastRowCount; + /// Row count previously emitted to subscribers, or null when there is no + /// baseline. Compared with the fresh count alongside the result hash; a null + /// baseline can never match, so the result is always treated as changed. + final int? lastRowCount; } /// How large a result's *structure* (rows × columns) can grow before handing @@ -404,7 +405,7 @@ RawQueryResult executeQuery( String sql, List parameters, int lastResultHash, - int lastRowCount, + int? lastRowCount, ) => _withAcquiredStmt(handleAddr, readerId, sql, parameters, (_, stmt) { final (newHash, newRowCount) = callQueryHash(stmt); if (newHash == lastResultHash && newRowCount == lastRowCount) { diff --git a/lib/src/reader/reader_pool.dart b/lib/src/reader/reader_pool.dart index 92027813..d094420f 100644 --- a/lib/src/reader/reader_pool.dart +++ b/lib/src/reader/reader_pool.dart @@ -133,7 +133,7 @@ final class ReaderPool { String sql, List parameters, int lastResultHash, - int lastRowCount, [ + int? lastRowCount, [ int? traceCorrelationId, ]) async { final result = await _dispatch( diff --git a/lib/src/stream_engine.dart b/lib/src/stream_engine.dart index b0f7052c..96d777b5 100644 --- a/lib/src/stream_engine.dart +++ b/lib/src/stream_engine.dart @@ -47,17 +47,6 @@ import 'extensions/set.dart'; /// deduplication, initial query with dependency tracking, write /// invalidation, re-query with result-change detection, and /// per-subscriber buffered delivery. -/// Row count that no real result can have, used to poison a stream's -/// comparison baseline so its next re-query is guaranteed to report a change. -/// -/// The row count — not the hash — is what makes this safe. `selectIfChanged` -/// treats a result as unchanged only when the hash *and* the row count both -/// match, and a real count is always >= 0, so this value can never match. A -/// hash sentinel would not work: an empty result hashes to 0 (see -/// `_finishInitialHash`), so 0 is a legitimate value rather than an impossible -/// one, and a stream over an empty query would go silent again. -const int _neverMatchingRowCount = -1; - final class StreamEngine { StreamEngine(this._pool); @@ -323,18 +312,18 @@ final class StreamEngine { // comparison baseline so the catch-up re-query is guaranteed to report // a change, and let that emission be the stream's first. // - // Poisoning is what makes suppression safe. Suppressing the initial - // emission while leaving the baseline set to these rows is the exp 255 - // bug: the re-query then compares against the very rows it replaced, - // reports "unchanged", and the stream goes permanently silent. With the - // baseline poisoned the re-query always emits, and `_requery` - // propagates errors to subscribers, so a stream always resolves to a - // value or an error. + // Clearing the baseline is what makes suppression safe. Suppressing + // the initial emission while leaving the baseline set to these rows is + // the exp 255 bug: the re-query then compares against the very rows it + // replaced, reports "unchanged", and the stream goes permanently + // silent. With no baseline there is nothing to match, so the re-query + // always emits — and `_requery` propagates errors to subscribers, so a + // stream always resolves to a value or an error. // // Measured (exp 256): this costs ~1 ms of time-to-correct-value and // removes the one stale frame emit-then-correct rendered every time. if (entry.dirty) { - entry.lastRowCount = _neverMatchingRowCount; + entry.lastRowCount = null; _requeryQueue.add(entry); _flushQueue(); } else { @@ -493,7 +482,11 @@ final class StreamEntry { /// ([EXP-077](../../experiments/077-cheap-check-first-sweep.md)). -1 means /// "no baseline yet" — the initial query hasn't returned. Compared with the /// fresh count as an additional equality guard alongside the result hash. - int lastRowCount = -1; + /// Row count of the last result emitted to subscribers, or null when there + /// is no baseline to compare against — a fresh entry, or one whose initial + /// result was withheld because a write superseded it. A null baseline + /// guarantees the next re-query reports a change and emits. + int? lastRowCount; /// Whether the stream is dirty and needs to be requeried. bool dirty = false;