amadan

carlos / platform Public

release: ChangeWatch polls the change index; ReadGate gates reads on generation and age (#336)

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
claude 4bc05b16ef3c42e9465cf4fc2bb00abe5305e054 parent 34cfaf9
2 files changed, +646 −0
  • internal/release/changewatch.go +210 −0
  • internal/release/changewatch_test.go +436 −0
diff --git a/internal/release/changewatch.go b/internal/release/changewatch.go
new file mode 100644
index 000000000..894fdd2eb
--- /dev/null
+++ b/internal/release/changewatch.go
@@ -0,0 +1,210 @@
+package release
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "sort"
+ "sync"
+ "time"
+)
+
+// ChangeIndexBackstop is the longest a gated reader goes without a successful
+// full read of a scope, however quiet the change index is (R3). It lives here
+// so provision, host, edge and console share one value without import cycles.
+// Callers jitter it by +-10%.
+const ChangeIndexBackstop = 30 * time.Minute
+
+// ChangeWatch follows the change index (ChangeIndexKey) with one conditional
+// GET per Poll: an unchanged index costs a 304 and no body. It is safe for
+// concurrent use.
+//
+// The watch is unusable until its first good poll, and again after any
+// failed one: an access error, a network error, an unknown schema, a
+// malformed body, or a store without ConditionalGetter. "Unknown" is never
+// read as "empty and unchanged"; callers walk as they did before the index
+// existed. The one exception is an absent index, which is the ordinary state
+// before the first bump and reads as usable and empty.
+type ChangeWatch struct {
+ store Store
+
+ pollMu sync.Mutex // serializes polls; held across the network call
+
+ mu sync.Mutex // guards the fields below; never held across I/O
+ usable bool
+ etag string // etag of the last good index body ("" when it was absent)
+ entries map[string]uint64 // the last good observation
+}
+
+// NewChangeWatch returns a watch over s. It has not polled yet, so it is
+// unusable.
+func NewChangeWatch(s Store) *ChangeWatch {
+ return &ChangeWatch{store: s, entries: map[string]uint64{}}
+}
+
+// Poll reads the index conditionally and returns, sorted, the scopes whose
+// generation differs from the last good observation: every scope on the
+// first good poll, then those that rose or appeared. On a 304 it returns
+// nil, nil. Any error marks the watch unusable until the next good poll and
+// is returned; that next poll reports every scope that differs from the last
+// good observation, including ones that rose while the watch was unusable.
+//
+// moved is advisory. Several consumers in one process share a watch (a box
+// agent's tick, the record view, the cert mirror), so whichever polls first
+// consumes a moved list and the others see nil. A consumer that must act on
+// movement compares Observed against the map it last acted on instead of
+// relying on Poll's return.
+func (w *ChangeWatch) Poll(ctx context.Context) (moved []string, err error) {
+ w.pollMu.Lock()
+ defer w.pollMu.Unlock()
+
+ g, ok := w.store.(ConditionalGetter)
+ if !ok {
+ return nil, w.fail(fmt.Errorf("release: change index unreadable: store has no conditional read: %w", errors.ErrUnsupported))
+ }
+ w.mu.Lock()
+ etag := w.etag
+ w.mu.Unlock()
+
+ body, newETag, changed, err := g.GetIfChanged(ctx, ChangeIndexKey, etag)
+ if errors.Is(err, os.ErrNotExist) {
+ return w.adopt(ChangeIndex{Schema: 1, Entries: map[string]uint64{}}, ""), nil
+ }
+ if err != nil {
+ return nil, w.fail(err)
+ }
+ if !changed {
+ w.mu.Lock()
+ w.usable = true // same bytes as the last good body, which are the last good observation
+ w.mu.Unlock()
+ return nil, nil
+ }
+ idx, _, err := readChangeIndex(ctx, func(context.Context, string) ([]byte, string, error) {
+ return body, newETag, nil
+ })
+ if err != nil {
+ return nil, w.fail(err)
+ }
+ return w.adopt(idx, newETag), nil
+}
+
+func (w *ChangeWatch) fail(err error) error {
+ w.mu.Lock()
+ w.usable = false
+ w.mu.Unlock()
+ return err
+}
+
+// adopt records idx as the last good observation and returns the scopes it
+// moved.
+func (w *ChangeWatch) adopt(idx ChangeIndex, etag string) []string {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ var moved []string
+ for scope, gen := range idx.Entries {
+ if old, ok := w.entries[scope]; !ok || old != gen {
+ moved = append(moved, scope)
+ }
+ }
+ sort.Strings(moved)
+ w.entries = idx.Entries
+ w.etag = etag
+ w.usable = true
+ return moved
+}
+
+// Usable reports whether the last poll succeeded.
+func (w *ChangeWatch) Usable() bool {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ return w.usable
+}
+
+// Gen is the last observed generation of scope. A scope the index does not
+// list has generation 0 with ok true; ok is false while the watch is
+// unusable.
+func (w *ChangeWatch) Gen(scope string) (gen uint64, ok bool) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if !w.usable {
+ return 0, false
+ }
+ return w.entries[scope], true
+}
+
+// Observed returns a copy of the last good entries, and false while the watch
+// is unusable. Poll's moved list is advisory when consumers share a watch, so
+// a consumer that must act on movement diffs Observed against its own
+// last-acted map.
+func (w *ChangeWatch) Observed() (map[string]uint64, bool) {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if !w.usable {
+ return nil, false
+ }
+ out := make(map[string]uint64, len(w.entries))
+ for k, v := range w.entries {
+ out[k] = v
+ }
+ return out, true
+}
+
+// ReadGate decides, per scope, whether a full read is due: when the watch
+// cannot vouch for the scope, the scope's generation has moved since the last
+// successful read, or the last successful read is maxAge old. It is safe for
+// concurrent use.
+type ReadGate struct {
+ w *ChangeWatch
+ maxAge time.Duration
+
+ mu sync.Mutex
+ marks map[string]gateMark
+}
+
+type gateMark struct {
+ gen uint64
+ at time.Time
+ ok bool
+}
+
+// NewReadGate gates reads on w. A nil w makes every read due. maxAge is the
+// longest a successful read stands in for the next one (use
+// ChangeIndexBackstop, jittered).
+func NewReadGate(w *ChangeWatch, maxAge time.Duration) *ReadGate {
+ return &ReadGate{w: w, maxAge: maxAge, marks: map[string]gateMark{}}
+}
+
+// Due reports whether scope must be read now and returns the generation to
+// pass to Mark once the read is done. Capture it BEFORE the read: a write that
+// lands during the read then leaves the scope due on the next check instead
+// of being hidden by it. The generation is 0 when the watch is nil or
+// unusable.
+func (g *ReadGate) Due(scope string, now time.Time) (gen uint64, due bool) {
+ if g == nil || g.w == nil {
+ return 0, true
+ }
+ gen, ok := g.w.Gen(scope)
+ if !ok {
+ return 0, true
+ }
+ g.mu.Lock()
+ m, marked := g.marks[scope]
+ g.mu.Unlock()
+ switch {
+ case !marked, !m.ok, m.gen != gen, now.Sub(m.at) >= g.maxAge:
+ return gen, true
+ }
+ return gen, false
+}
+
+// Mark records the outcome of a read of scope that began when Due returned
+// gen. A failed read (ok false) leaves the scope due.
+func (g *ReadGate) Mark(scope string, gen uint64, now time.Time, ok bool) {
+ if g == nil {
+ return
+ }
+ g.mu.Lock()
+ g.marks[scope] = gateMark{gen: gen, at: now, ok: ok}
+ g.mu.Unlock()
+}
diff --git a/internal/release/changewatch_test.go b/internal/release/changewatch_test.go
new file mode 100644
index 000000000..eb0d605a8
--- /dev/null
+++ b/internal/release/changewatch_test.go
@@ -0,0 +1,436 @@
+package release
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io/fs"
+ "reflect"
+ "sort"
+ "sync"
+ "testing"
+ "time"
+)
+
+// condCounter wraps a DirStore and records the etag every GetIfChanged call
+// carried, so a test can see that the second poll was conditional.
+type condCounter struct {
+ *DirStore
+ mu sync.Mutex
+ etags []string
+ getErr error // when set, GetIfChanged fails with it
+}
+
+func (c *condCounter) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
+ c.mu.Lock()
+ c.etags = append(c.etags, etag)
+ err := c.getErr
+ c.mu.Unlock()
+ if err != nil {
+ return nil, "", false, err
+ }
+ return c.DirStore.GetIfChanged(ctx, key, etag)
+}
+
+func (c *condCounter) setErr(err error) {
+ c.mu.Lock()
+ c.getErr = err
+ c.mu.Unlock()
+}
+
+func (c *condCounter) seen() []string {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ return append([]string(nil), c.etags...)
+}
+
+func writeScoped(t *testing.T, s Store, keys ...string) {
+ t.Helper()
+ for _, k := range keys {
+ if _, err := s.Put(context.Background(), k, []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+func TestChangeWatchFirstPollReportsAll(t *testing.T) {
+ t.Parallel()
+ d := indexedDirStore(t)
+ writeScoped(t, d, "apps/bab/shop/instances/h.json", "accounts/acme/members.json")
+ w := NewChangeWatch(d)
+ if w.Usable() {
+ t.Fatal("a watch that has not polled is usable")
+ }
+ if _, ok := w.Gen("apps/bab/shop"); ok {
+ t.Fatal("Gen ok before the first poll")
+ }
+ moved, err := w.Poll(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := []string{"accounts/acme", "apps/bab/shop"}
+ if !reflect.DeepEqual(moved, want) {
+ t.Fatalf("moved = %v, want %v", moved, want)
+ }
+ if !w.Usable() {
+ t.Fatal("not usable after a good poll")
+ }
+ if g, ok := w.Gen("apps/bab/shop"); !ok || g != 1 {
+ t.Fatalf("Gen = %d, %v; want 1, true", g, ok)
+ }
+ if g, ok := w.Gen("apps/bab/other"); !ok || g != 0 {
+ t.Fatalf("Gen of an unlisted scope = %d, %v; want 0, true", g, ok)
+ }
+}
+
+func TestChangeWatchUnchangedIs304(t *testing.T) {
+ t.Parallel()
+ c := &condCounter{DirStore: indexedDirStore(t)}
+ writeScoped(t, c.DirStore, "apps/bab/shop/instances/h.json")
+ w := NewChangeWatch(c)
+ if _, err := w.Poll(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ moved, err := w.Poll(context.Background())
+ if err != nil || moved != nil {
+ t.Fatalf("second poll = %v, %v; want nil, nil", moved, err)
+ }
+ seen := c.seen()
+ if len(seen) != 2 || seen[0] != "" || seen[1] == "" {
+ t.Fatalf("etags sent = %q; want an empty one, then the index's", seen)
+ }
+}
+
+func TestChangeWatchReportsOnlyMoved(t *testing.T) {
+ t.Parallel()
+ d := indexedDirStore(t)
+ writeScoped(t, d, "apps/bab/shop/instances/h.json", "accounts/acme/members.json")
+ w := NewChangeWatch(d)
+ if _, err := w.Poll(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ writeScoped(t, d, "apps/bab/shop/instances/i.json", "control/certs/example.com.json")
+ moved, err := w.Poll(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := []string{"apps/bab/shop", "control/certs"}
+ if !reflect.DeepEqual(moved, want) {
+ t.Fatalf("moved = %v, want %v", moved, want)
+ }
+ if g, _ := w.Gen("apps/bab/shop"); g != 2 {
+ t.Fatalf("Gen = %d, want 2", g)
+ }
+}
+
+func TestChangeWatchAbsentIndexIsUsableEmpty(t *testing.T) {
+ t.Parallel()
+ d := indexedDirStore(t)
+ w := NewChangeWatch(d)
+ moved, err := w.Poll(context.Background())
+ if err != nil || len(moved) != 0 {
+ t.Fatalf("poll = %v, %v; want nothing", moved, err)
+ }
+ if !w.Usable() {
+ t.Fatal("an absent index is the pre-first-bump state and must be usable")
+ }
+ if obs, ok := w.Observed(); !ok || len(obs) != 0 {
+ t.Fatalf("Observed = %v, %v; want empty, true", obs, ok)
+ }
+ // The first write then shows up as an appearance.
+ writeScoped(t, d, "apps/bab/shop/instances/h.json")
+ moved, err = w.Poll(context.Background())
+ if err != nil || !reflect.DeepEqual(moved, []string{"apps/bab/shop"}) {
+ t.Fatalf("poll after first write = %v, %v", moved, err)
+ }
+}
+
+func TestChangeWatchDeniedIsUnusable(t *testing.T) {
+ t.Parallel()
+ c := &condCounter{DirStore: indexedDirStore(t)}
+ writeScoped(t, c.DirStore, "apps/bab/shop/instances/h.json")
+ c.setErr(fmt.Errorf("get %s: %w", ChangeIndexKey, fs.ErrPermission))
+ w := NewChangeWatch(c)
+ moved, err := w.Poll(context.Background())
+ if !errors.Is(err, fs.ErrPermission) || moved != nil {
+ t.Fatalf("poll = %v, %v; want the access error", moved, err)
+ }
+ if w.Usable() {
+ t.Fatal("usable after a denied read")
+ }
+ if _, ok := w.Gen("apps/bab/shop"); ok {
+ t.Fatal("Gen ok while unusable")
+ }
+ if _, ok := w.Observed(); ok {
+ t.Fatal("Observed ok while unusable")
+ }
+}
+
+func TestChangeWatchUsableThenErrorThenRecovers(t *testing.T) {
+ t.Parallel()
+ c := &condCounter{DirStore: indexedDirStore(t)}
+ writeScoped(t, c.DirStore, "apps/bab/shop/instances/h.json", "accounts/acme/members.json")
+ w := NewChangeWatch(c)
+ if _, err := w.Poll(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ c.setErr(errors.New("network down"))
+ writeScoped(t, c.DirStore, "apps/bab/shop/instances/i.json")
+ if _, err := w.Poll(context.Background()); err == nil {
+ t.Fatal("poll succeeded through an error")
+ }
+ if w.Usable() {
+ t.Fatal("still usable after an error")
+ }
+ c.setErr(nil)
+ moved, err := w.Poll(context.Background())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(moved, []string{"apps/bab/shop"}) {
+ t.Fatalf("moved = %v; want the scope that rose while unusable", moved)
+ }
+ if !w.Usable() {
+ t.Fatal("not usable after recovering")
+ }
+ // An unchanged index after an error recovers without a spurious move.
+ c.setErr(errors.New("blip"))
+ _, _ = w.Poll(context.Background())
+ c.setErr(nil)
+ moved, err = w.Poll(context.Background())
+ if err != nil || moved != nil || !w.Usable() {
+ t.Fatalf("recovery on an unchanged index = %v, %v, usable %v", moved, err, w.Usable())
+ }
+}
+
+func TestChangeWatchUnknownSchemaIsUnusable(t *testing.T) {
+ t.Parallel()
+ d := NewDirStore(t.TempDir())
+ if _, err := d.Put(context.Background(), ChangeIndexKey, []byte(`{"schema":2,"entries":{"apps/a/b":1}}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ w := NewChangeWatch(d)
+ _, err := w.Poll(context.Background())
+ if !errors.Is(err, ErrChangeIndexSchema) {
+ t.Fatalf("err = %v, want ErrChangeIndexSchema", err)
+ }
+ if w.Usable() {
+ t.Fatal("usable on an unknown schema")
+ }
+ // A malformed body is unknown too.
+ _, etag, _ := d.Get(context.Background(), ChangeIndexKey)
+ if _, err := d.Put(context.Background(), ChangeIndexKey, []byte(`not json`), etag); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := w.Poll(context.Background()); err == nil || w.Usable() {
+ t.Fatalf("malformed index: err %v, usable %v", err, w.Usable())
+ }
+}
+
+func TestChangeWatchWithoutExtensionIsUnusable(t *testing.T) {
+ t.Parallel()
+ w := NewChangeWatch(bareStore{NewDirStore(t.TempDir())})
+ _, err := w.Poll(context.Background())
+ if !errors.Is(err, errors.ErrUnsupported) {
+ t.Fatalf("err = %v, want ErrUnsupported", err)
+ }
+ if w.Usable() {
+ t.Fatal("usable without ConditionalGetter")
+ }
+ // The prefix wrapper over such a store answers ErrUnsupported itself.
+ w = NewChangeWatch(NewPrefixStore(bareStore{NewDirStore(t.TempDir())}, "base/"))
+ if _, err := w.Poll(context.Background()); err == nil || w.Usable() {
+ t.Fatalf("prefixed bare store: err %v, usable %v", err, w.Usable())
+ }
+}
+
+func TestChangeWatchObservedIsACopy(t *testing.T) {
+ t.Parallel()
+ d := indexedDirStore(t)
+ writeScoped(t, d, "apps/bab/shop/instances/h.json")
+ w := NewChangeWatch(d)
+ if _, err := w.Poll(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ obs, _ := w.Observed()
+ obs["apps/bab/shop"] = 99
+ delete(obs, "nothing")
+ if g, _ := w.Gen("apps/bab/shop"); g != 1 {
+ t.Fatalf("mutating Observed changed the watch: Gen = %d", g)
+ }
+}
+
+func TestChangeWatchConcurrentPolls(t *testing.T) {
+ t.Parallel()
+ d := indexedDirStore(t)
+ writeScoped(t, d, "apps/bab/shop/instances/h.json")
+ w := NewChangeWatch(d)
+ var wg sync.WaitGroup
+ var mu sync.Mutex
+ var all []string
+ for range 8 {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ moved, err := w.Poll(context.Background())
+ if err != nil {
+ t.Error(err)
+ }
+ w.Gen("apps/bab/shop")
+ w.Observed()
+ mu.Lock()
+ all = append(all, moved...)
+ mu.Unlock()
+ }()
+ }
+ wg.Wait()
+ sort.Strings(all)
+ if !reflect.DeepEqual(all, []string{"apps/bab/shop"}) {
+ t.Fatalf("scope reported %v across racing polls; want once", all)
+ }
+}
+
+func TestReadGateDue(t *testing.T) {
+ t.Parallel()
+ const maxAge = 30 * time.Minute
+ t0 := time.Date(2026, 10, 22, 12, 0, 0, 0, time.UTC)
+ const scope = "apps/bab/shop"
+ ctx := context.Background()
+
+ usableWatch := func(t *testing.T) (*ChangeWatch, *DirStore) {
+ d := indexedDirStore(t)
+ writeScoped(t, d, "apps/bab/shop/instances/h.json")
+ w := NewChangeWatch(d)
+ if _, err := w.Poll(ctx); err != nil {
+ t.Fatal(err)
+ }
+ return w, d
+ }
+
+ tests := []struct {
+ name string
+ run func(t *testing.T) bool // returns Due's answer
+ want bool
+ }{
+ {"nil watch is due", func(t *testing.T) bool {
+ _, due := NewReadGate(nil, maxAge).Due(scope, t0)
+ return due
+ }, true},
+ {"unusable watch is due", func(t *testing.T) bool {
+ g := NewReadGate(NewChangeWatch(NewDirStore(t.TempDir())), maxAge)
+ _, due := g.Due(scope, t0)
+ return due
+ }, true},
+ {"never marked is due", func(t *testing.T) bool {
+ w, _ := usableWatch(t)
+ _, due := NewReadGate(w, maxAge).Due(scope, t0)
+ return due
+ }, true},
+ {"marked ok, same gen, within age is not due", func(t *testing.T) bool {
+ w, _ := usableWatch(t)
+ g := NewReadGate(w, maxAge)
+ gen, _ := g.Due(scope, t0)
+ g.Mark(scope, gen, t0, true)
+ _, due := g.Due(scope, t0.Add(maxAge-time.Second))
+ return due
+ }, false},
+ {"generation moved is due", func(t *testing.T) bool {
+ w, d := usableWatch(t)
+ g := NewReadGate(w, maxAge)
+ gen, _ := g.Due(scope, t0)
+ g.Mark(scope, gen, t0, true)
+ writeScoped(t, d, "apps/bab/shop/instances/i.json")
+ if _, err := w.Poll(ctx); err != nil {
+ t.Fatal(err)
+ }
+ _, due := g.Due(scope, t0.Add(time.Second))
+ return due
+ }, true},
+ {"other scope moving is not due", func(t *testing.T) bool {
+ w, d := usableWatch(t)
+ g := NewReadGate(w, maxAge)
+ gen, _ := g.Due(scope, t0)
+ g.Mark(scope, gen, t0, true)
+ writeScoped(t, d, "accounts/acme/members.json")
+ if _, err := w.Poll(ctx); err != nil {
+ t.Fatal(err)
+ }
+ _, due := g.Due(scope, t0.Add(time.Second))
+ return due
+ }, false},
+ {"age at maxAge is due", func(t *testing.T) bool {
+ w, _ := usableWatch(t)
+ g := NewReadGate(w, maxAge)
+ gen, _ := g.Due(scope, t0)
+ g.Mark(scope, gen, t0, true)
+ _, due := g.Due(scope, t0.Add(maxAge))
+ return due
+ }, true},
+ {"last mark failed is due", func(t *testing.T) bool {
+ w, _ := usableWatch(t)
+ g := NewReadGate(w, maxAge)
+ gen, _ := g.Due(scope, t0)
+ g.Mark(scope, gen, t0, false)
+ _, due := g.Due(scope, t0.Add(time.Second))
+ return due
+ }, true},
+ {"a failed mark after a good one is due", func(t *testing.T) bool {
+ w, _ := usableWatch(t)
+ g := NewReadGate(w, maxAge)
+ gen, _ := g.Due(scope, t0)
+ g.Mark(scope, gen, t0, true)
+ g.Mark(scope, gen, t0.Add(time.Minute), false)
+ _, due := g.Due(scope, t0.Add(2*time.Minute))
+ return due
+ }, true},
+ {"watch going unusable makes a marked scope due", func(t *testing.T) bool {
+ c := &condCounter{DirStore: indexedDirStore(t)}
+ writeScoped(t, c.DirStore, "apps/bab/shop/instances/h.json")
+ w := NewChangeWatch(c)
+ if _, err := w.Poll(ctx); err != nil {
+ t.Fatal(err)
+ }
+ g := NewReadGate(w, maxAge)
+ gen, _ := g.Due(scope, t0)
+ g.Mark(scope, gen, t0, true)
+ c.setErr(errors.New("down"))
+ _, _ = w.Poll(ctx)
+ _, due := g.Due(scope, t0.Add(time.Second))
+ return due
+ }, true},
+ }
+ for _, tc := range tests {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Parallel()
+ if got := tc.run(t); got != tc.want {
+ t.Fatalf("due = %v, want %v", got, tc.want)
+ }
+ })
+ }
+}
+
+func TestReadGateCapturedGenSurvivesRaceWithPoll(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ d := indexedDirStore(t)
+ writeScoped(t, d, "apps/bab/shop/instances/h.json")
+ w := NewChangeWatch(d)
+ if _, err := w.Poll(ctx); err != nil {
+ t.Fatal(err)
+ }
+ g := NewReadGate(w, time.Hour)
+ t0 := time.Now()
+ gen, due := g.Due("apps/bab/shop", t0)
+ if !due {
+ t.Fatal("never-marked scope not due")
+ }
+ // A write lands and is polled while the read is in flight.
+ writeScoped(t, d, "apps/bab/shop/instances/i.json")
+ if _, err := w.Poll(ctx); err != nil {
+ t.Fatal(err)
+ }
+ g.Mark("apps/bab/shop", gen, t0, true) // the generation captured BEFORE the read
+ if _, due := g.Due("apps/bab/shop", t0); !due {
+ t.Fatal("a mark of the pre-read generation hid a write that landed during the read")
+ }
+}