Fix: stream could emit nothing when a write raced its initial query (+ exp 256: never emit stale)

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<double> xs) {
+  final s = [...xs]..sort();
+  return s[s.length ~/ 2];
+}
+
+double _p90(List<double> 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<void>();
+  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<void>.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<void> 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 = <double>[], corrects = <double>[];
+    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);
+}

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`.

experiments/255-stream-initial-emission-race.md

@@ -0,0 +1,149 @@
+# 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
+
+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
+if (entry.dirty) {
+  entry.lastRowCount = null;   // no baseline, so the re-query always emits
+  _requeryQueue.add(entry);
+  _flushQueue();
+} else {
+  entry.emit(initialRows);
+}
+```
+
+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.
+
+**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
+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:
+
+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.

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.