amadan

carlos / platform Public

release, provision, host: any failed index poll leaves the watch unknown; record view polls are bounded; no canary gate on a tick that could not read the index (#336)

Final review finding 3. A poll that gives up waiting for another caller's
hung index GET now marks the watch unknown like any other failure (ETag and
entries kept, so recovery is a 304). RecordView's Refresh and Rebuild bound
their polls with ChangeIndexPollTimeout, and a refresh whose poll failed
walks without polling again. The fast tick installs the canary pointer gate
only when its own index observation succeeded.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude cd418ebf7462cf35282029ef007085dca92c53ac parent 4512018
7 files changed, +283 −22
  • internal/host/agent.go +6 −1
  • internal/host/changeindex_test.go +89 −0
  • internal/host/readgate_test.go +34 −0
  • internal/provision/recordview.go +30 −9
  • internal/provision/recordview_test.go +71 −0
  • internal/release/changewatch.go +8 −6
  • internal/release/changewatch_test.go +45 −6
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 9eabb586d..ba17bd6dc 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -2729,7 +2729,12 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
member[h] = true
}
opts := d.passOpts
- if d.canaryGate != nil {
+ // The canary gate only on a tick whose own index observation succeeded:
+ // a tick that could not read the index reads every pointer, as before
+ // the index. (A failed poll also leaves the watch unknown, so the gate
+ // would find every group due anyway; this keeps it off whatever another
+ // caller's poll did in between.)
+ if d.canaryGate != nil && indexOK {
opts = append(slices.Clip(opts), release.WithPointerGate(d.canaryPointerGate(now)))
}
if err := release.PassFilter(ctx, d.reg, d.store, d.pub, d.optDir, d.restart,
diff --git a/internal/host/changeindex_test.go b/internal/host/changeindex_test.go
index b27dd528a..a4321f052 100644
--- a/internal/host/changeindex_test.go
+++ b/internal/host/changeindex_test.go
@@ -30,6 +30,24 @@ type ciCountingStore struct {
mu sync.Mutex
calls []string
failList string // List and ListETags of exactly this prefix fail
+ // stall, when set, holds every GetIfChanged of the change index until
+ // the caller's ctx ends or stall is closed: a hung index GET while the
+ // rest of the bucket answers.
+ stall chan struct{}
+}
+
+// stallIndex hangs the index GET until the returned func is called.
+func (s *ciCountingStore) stallIndex() (release func()) {
+ ch := make(chan struct{})
+ s.mu.Lock()
+ s.stall = ch
+ s.mu.Unlock()
+ return func() {
+ s.mu.Lock()
+ s.stall = nil
+ s.mu.Unlock()
+ close(ch)
+ }
}
func (s *ciCountingStore) setFailList(p string) { s.mu.Lock(); s.failList = p; s.mu.Unlock() }
@@ -82,6 +100,16 @@ func (s *ciCountingStore) Get(ctx context.Context, key string) ([]byte, string,
func (s *ciCountingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
s.note("GetIfChanged", key)
+ s.mu.Lock()
+ stall := s.stall
+ s.mu.Unlock()
+ if stall != nil && key == release.ChangeIndexKey {
+ select {
+ case <-ctx.Done():
+ return nil, "", false, ctx.Err()
+ case <-stall:
+ }
+ }
return s.DirStore.GetIfChanged(ctx, key, etag)
}
@@ -330,6 +358,67 @@ func TestViewStoreFallsBackWhenTheViewCannotVouch(t *testing.T) {
}
}
+// A hung index GET while the rest of the bucket answers (amadan #336 final
+// review, finding 3): the reconcile pass gives up on the index after
+// ChangeIndexPollTimeout, walks, and mints what the walk found; the watch
+// is unknown meanwhile, so a gated read marked before the stall is due
+// again; and the next good poll brings the watch back.
+func TestStalledIndexGetKeepsReconcileWalking(t *testing.T) {
+ w := newCIWorld(t)
+ v := armView(t, w.live)
+ if err := v.Rebuild(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ reg := restartTestRegistry(t)
+ cfg := ciReconcileConfig(t)
+ if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil {
+ t.Fatal(err)
+ }
+ gate := newChangeGate()
+ clock := newGateClock()
+ gen, _ := gate.Due(w.scope("shop"), clock.now())
+ gate.Mark(w.scope("shop"), gen, clock.now(), true)
+ if _, due := gate.Due(w.scope("shop"), clock.now()); due {
+ t.Fatal("fixture: a marked scope is due before the stall")
+ }
+
+ w.record(t, "shop", "two.shop.example.com")
+ unstall := w.live.stallIndex()
+ defer func() {
+ if unstall != nil {
+ unstall()
+ }
+ }()
+ done := make(chan error, 1)
+ go func() { done <- reconcileSerialized(context.Background(), w.live, reg, cfg) }()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(4 * release.ChangeIndexPollTimeout):
+ t.Fatal("a reconcile pass hung behind a stalled index GET")
+ }
+ if rt, err := reg.Get("two.shop.example.com"); err != nil || rt == nil {
+ t.Fatalf("the pass over a stalled index did not walk to the new record: %v", err)
+ }
+ if armedChangeWatch.Load().Usable() {
+ t.Fatal("the watch vouches after its poll gave up")
+ }
+ if _, due := gate.Due(w.scope("shop"), clock.now()); !due {
+ t.Fatal("a gated read trusted the frozen observation")
+ }
+
+ unstall()
+ unstall = nil
+ if err := reconcileSerialized(context.Background(), w.live, reg, cfg); err != nil {
+ t.Fatal(err)
+ }
+ if !armedChangeWatch.Load().Usable() {
+ t.Fatal("the watch did not recover on the next good poll")
+ }
+}
+
// ---- moves -----------------------------------------------------------------
// The moves pass takes its own snapshot inside w.pass, never the one the
diff --git a/internal/host/readgate_test.go b/internal/host/readgate_test.go
index 13c7b2b7b..8d3405ea8 100644
--- a/internal/host/readgate_test.go
+++ b/internal/host/readgate_test.go
@@ -262,6 +262,40 @@ func TestCanaryPointerReadWhileUnsettled(t *testing.T) {
}
}
+// A tick whose own index poll failed reads the canary pointer (amadan #336
+// final review, finding 3). Here another caller's index GET hangs holding
+// the watch's poll slot, and the tick gives up waiting for it: before the
+// fix that give-up left the frozen observation vouching, and the gate
+// skipped the read. The next good poll restores the gate.
+func TestCanaryPointerReadWhenTheTicksIndexPollFails(t *testing.T) {
+ t.Parallel()
+ k := newCanaryKit(t, true)
+ if n := k.tick(t, 0); n != 1 {
+ t.Fatalf("first tick: %d pointer reads, want 1", n)
+ }
+ if n := k.tick(t, 2*time.Second); n != 1 {
+ t.Fatalf("a quiet tick read the pointer (%d reads)", n)
+ }
+ unstall := k.cs.stallIndex()
+ polls := k.cs.count("GetIfChanged", release.ChangeIndexKey)
+ hung := make(chan error, 1)
+ go func() { _, err := k.d.watch.Poll(context.Background()); hung <- err }()
+ for k.cs.count("GetIfChanged", release.ChangeIndexKey) == polls {
+ time.Sleep(time.Millisecond)
+ }
+ if n := k.tick(t, 2*time.Second); n != 2 {
+ unstall()
+ t.Fatalf("a tick whose index poll gave up skipped the pointer read (%d reads, want 2): it trusted a frozen observation", n)
+ }
+ unstall()
+ if err := <-hung; err != nil {
+ t.Fatalf("the hung poll, released: %v", err)
+ }
+ if n := k.tick(t, 2*time.Second); n != 2 {
+ t.Fatalf("after the watch recovered the quiet tick read the pointer (%d reads)", n)
+ }
+}
+
// A stable route keeps today's path: the gate never skips its pointer.
func TestCanaryGateLeavesStableRoutes(t *testing.T) {
t.Parallel()
diff --git a/internal/provision/recordview.go b/internal/provision/recordview.go
index 52e5c6f0e..7324b541a 100644
--- a/internal/provision/recordview.go
+++ b/internal/provision/recordview.go
@@ -12,6 +12,7 @@ import (
"strings"
"sync"
"sync/atomic"
+ "time"
"github.com/carlosframework/platform/internal/release"
)
@@ -33,6 +34,9 @@ import (
type RecordView struct {
live release.Store
watch *release.ChangeWatch
+ // pollTimeout bounds each index poll (release.ChangeIndexPollTimeout;
+ // tests shorten it).
+ pollTimeout time.Duration
// mu serializes Refresh and Rebuild for their whole duration, the backstop
// walk included, so a publication always reflects reads that began after
@@ -55,7 +59,7 @@ var recordViewRebuildHook func()
// NewRecordView returns a view over live that follows watch. It has published
// nothing: Snapshot is nil until the first Refresh or Rebuild succeeds.
func NewRecordView(live release.Store, watch *release.ChangeWatch) *RecordView {
- return &RecordView{live: live, watch: watch, applied: map[string]uint64{}, pending: map[string]bool{}}
+ return &RecordView{live: live, watch: watch, pollTimeout: release.ChangeIndexPollTimeout, applied: map[string]uint64{}, pending: map[string]bool{}}
}
// Snapshot is the latest publication, or nil before the first. It never
@@ -115,15 +119,15 @@ func (v *RecordView) Refresh(ctx context.Context) error {
defer v.mu.Unlock()
prev := v.snap.Load()
if prev == nil || v.rebuildDue {
- return v.rebuildLocked(ctx)
+ return v.rebuildLocked(ctx, true)
}
- if _, err := v.watch.Poll(ctx); err != nil {
+ if err := v.poll(ctx); err != nil {
log.Printf("provision: record view: change index unusable (%v): walking", err)
- return v.rebuildLocked(ctx)
+ return v.rebuildLocked(ctx, false)
}
obs, ok := v.watch.Observed()
if !ok {
- return v.rebuildLocked(ctx)
+ return v.rebuildLocked(ctx, false)
}
var due []string
for scope, gen := range obs {
@@ -186,12 +190,29 @@ func (v *RecordView) Refresh(ctx context.Context) error {
func (v *RecordView) Rebuild(ctx context.Context) error {
v.mu.Lock()
defer v.mu.Unlock()
- return v.rebuildLocked(ctx)
+ return v.rebuildLocked(ctx, true)
}
-func (v *RecordView) rebuildLocked(ctx context.Context) error {
- if _, err := v.watch.Poll(ctx); err != nil {
- log.Printf("provision: record view: change index unusable (%v): the walk records no generations", err)
+// poll is one index poll, bounded by pollTimeout inside ctx: it runs under
+// mu, and the callers' contexts are long-lived, so a stalled index GET must
+// not hold the view (and with it every reconcile pass) or the watch's shared
+// poll slot. A poll that runs out marks the watch unknown, and the caller
+// walks.
+func (v *RecordView) poll(ctx context.Context) error {
+ pctx, cancel := context.WithTimeout(ctx, v.pollTimeout)
+ defer cancel()
+ _, err := v.watch.Poll(pctx)
+ return err
+}
+
+// rebuildLocked is Rebuild under mu. poll is false when the caller's own
+// poll just failed: the watch is unknown, so a second poll could only spend
+// another bound before the same walk.
+func (v *RecordView) rebuildLocked(ctx context.Context, poll bool) error {
+ if poll {
+ if err := v.poll(ctx); err != nil {
+ log.Printf("provision: record view: change index unusable (%v): the walk records no generations", err)
+ }
}
obs, usable := v.watch.Observed()
prev := v.snap.Load()
diff --git a/internal/provision/recordview_test.go b/internal/provision/recordview_test.go
index 2160ba1e8..3d9685bdf 100644
--- a/internal/provision/recordview_test.go
+++ b/internal/provision/recordview_test.go
@@ -32,8 +32,11 @@ type viewCountingStore struct {
ghost string // ListETags lists this key, which is not there: deleted between list and read
cancelOn string // ListETags of exactly this prefix calls cancel first
cancel context.CancelFunc
+ stall string // GetIfChanged of exactly this key blocks until the caller's ctx ends
}
+func (s *viewCountingStore) setStall(k string) { s.mu.Lock(); s.stall = k; s.mu.Unlock() }
+
func (s *viewCountingStore) note(method, key string) {
s.mu.Lock()
s.calls = append(s.calls, method+" "+key)
@@ -82,6 +85,13 @@ func (s *viewCountingStore) Get(ctx context.Context, key string) ([]byte, string
func (s *viewCountingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
s.note("GetIfChanged", key)
+ s.mu.Lock()
+ stall := s.stall != "" && key == s.stall
+ s.mu.Unlock()
+ if stall {
+ <-ctx.Done()
+ return nil, "", false, ctx.Err()
+ }
if s.getFails(key) {
return nil, "", false, errors.New("get refused")
}
@@ -625,6 +635,67 @@ func TestViewRetriesFailedScopeOn304(t *testing.T) {
}
}
+// A stalled index GET never holds a refresh, or the backstop rebuild, for
+// longer than the poll bound, whatever the caller's context: each gives up
+// on the index, walks the bucket (which answers), and publishes. The watch
+// is unknown meanwhile, and the next good poll brings the view back to the
+// index.
+func TestViewStalledIndexIsBounded(t *testing.T) {
+ t.Parallel()
+ f := newViewFixture(t, true, true)
+ v := f.view()
+ v.pollTimeout = 50 * time.Millisecond
+ mustRebuild(t, v)
+ f.record(t, f.live.DirStore, f.acct[1], "docs", "two")
+ f.live.setStall(release.ChangeIndexKey)
+ f.live.reset()
+ for _, step := range []struct {
+ name string
+ run func(context.Context) error
+ }{{"Refresh", v.Refresh}, {"Rebuild", v.Rebuild}} {
+ done := make(chan error, 1)
+ go func() { done <- step.run(context.Background()) }()
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatalf("%s over a stalled index: %v", step.name, err)
+ }
+ case <-time.After(10 * time.Second):
+ t.Fatalf("%s over a stalled index did not return: its poll is unbounded", step.name)
+ }
+ sn := v.Snapshot()
+ if !sn.Complete || !slices.Contains(viewHosts(t, sn.Store(f.live)), "two.docs.example.com") {
+ t.Fatalf("%s over a stalled index: complete %v, records %v; want the walk's", step.name, sn.Complete, viewHosts(t, sn.Store(f.live)))
+ }
+ if v.watch.Usable() {
+ t.Fatalf("%s: the watch vouches after its poll gave up", step.name)
+ }
+ }
+ if n := slices.Index(f.live.got(), "GetIfChanged "+release.ChangeIndexKey); n < 0 {
+ t.Fatalf("no index poll at all: %v", f.live.got())
+ }
+ polls := 0
+ for _, c := range f.live.got() {
+ if c == "GetIfChanged "+release.ChangeIndexKey {
+ polls++
+ }
+ }
+ if polls != 2 {
+ t.Fatalf("the index was polled %d time(s) for one Refresh and one Rebuild, want 2: a refresh that gave up polls again before its walk", polls)
+ }
+ f.live.setStall("")
+ f.live.reset()
+ mustRefresh(t, v)
+ if !v.watch.Usable() {
+ t.Fatal("the watch did not recover on the next good poll")
+ }
+ f.live.reset()
+ mustRefresh(t, v)
+ if got, want := f.live.got(), []string{"GetIfChanged " + release.ChangeIndexKey}; !reflect.DeepEqual(got, want) {
+ t.Fatalf("a quiet refresh after recovery read %v, want %v", got, want)
+ }
+}
+
func TestViewFirstRefreshWithoutIndexRebuilds(t *testing.T) {
t.Parallel()
f := newViewFixture(t, true, false)
diff --git a/internal/release/changewatch.go b/internal/release/changewatch.go
index 018cb7854..df0257711 100644
--- a/internal/release/changewatch.go
+++ b/internal/release/changewatch.go
@@ -28,7 +28,8 @@ const ChangeIndexPollTimeout = 5 * time.Second
//
// 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
+// malformed body, a store without ConditionalGetter, or a caller's context
+// ending, while waiting for another caller's poll included. "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.
@@ -66,17 +67,18 @@ func NewChangeWatch(s Store) *ChangeWatch {
// relying on Poll's return.
//
// A Poll that waits behind another caller's in-flight poll stops waiting when
-// ctx ends and returns ctx's error with the watch's state unchanged: it read
-// nothing, so it has nothing to say about the index. Callers treat that error
-// as they treat any other, as "unknown".
+// ctx ends and returns ctx's error. That give-up is a failed poll like any
+// other: the watch is unusable until the next good poll, so no gated reader
+// trusts an observation a stalled index GET has frozen. The ETag and entries
+// are kept, so the next good poll of an unchanged index is a 304 again.
func (w *ChangeWatch) Poll(ctx context.Context) (moved []string, err error) {
if err := ctx.Err(); err != nil {
- return nil, err
+ return nil, w.fail(err)
}
select {
case w.pollSem <- struct{}{}:
case <-ctx.Done():
- return nil, ctx.Err()
+ return nil, w.fail(ctx.Err())
}
defer func() { <-w.pollSem }()
diff --git a/internal/release/changewatch_test.go b/internal/release/changewatch_test.go
index 4baaf9a97..59ee9950b 100644
--- a/internal/release/changewatch_test.go
+++ b/internal/release/changewatch_test.go
@@ -462,8 +462,11 @@ func (b *blockingCond) GetIfChanged(ctx context.Context, key, etag string) ([]by
return b.DirStore.GetIfChanged(ctx, key, etag)
}
-// A Poll waiting behind a stuck one gives up when its context ends, and its
-// give-up leaves the watch's last good observation untouched.
+// A Poll waiting behind a stuck one gives up when its context ends. The
+// give-up is a failed poll like any other: the watch is unknown until the
+// next good one (a frozen observation must never vouch for a gated read),
+// but it keeps its ETag and entries, so that next poll is a 304 that
+// reports nothing moved.
func TestChangeWatchPollLockWaitHonoursContext(t *testing.T) {
t.Parallel()
d := indexedDirStore(t)
@@ -499,14 +502,50 @@ func TestChangeWatchPollLockWaitHonoursContext(t *testing.T) {
close(release)
t.Fatal("a Poll waiting behind a stuck one ignored its context")
}
- if !w.Usable() {
- t.Fatal("a give-up on the lock must leave the watch usable as it was")
+ if w.Usable() {
+ t.Fatal("a give-up on the lock left the watch usable: its observation is frozen")
}
- if g, ok := w.Gen("apps/bab/shop"); !ok || g != gen {
- t.Fatalf("Gen after a give-up = %d, %v; want %d, true", g, ok, gen)
+ if _, ok := w.Gen("apps/bab/shop"); ok {
+ t.Fatal("Gen vouched for a scope after a give-up")
+ }
+ if _, ok := w.Observed(); ok {
+ t.Fatal("Observed vouched after a give-up")
}
close(release)
if err := <-stuck; err != nil {
t.Fatalf("the stuck Poll, released: %v", err)
}
+ if g, ok := w.Gen("apps/bab/shop"); !ok || g != gen {
+ t.Fatalf("Gen after the stuck poll answered = %d, %v; want %d, true", g, ok, gen)
+ }
+}
+
+// A Poll whose context has already ended reads nothing, and is a failed
+// poll: the watch is unknown until the next good one, which is a 304
+// because the watch kept its ETag.
+func TestChangeWatchCancelledPollIsUnknownThenA304(t *testing.T) {
+ t.Parallel()
+ d := indexedDirStore(t)
+ writeScoped(t, d, "apps/bab/shop/instances/h.json")
+ c := &condCounter{DirStore: d}
+ w := NewChangeWatch(c)
+ if _, err := w.Poll(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ if _, err := w.Poll(ctx); !errors.Is(err, context.Canceled) {
+ t.Fatalf("a cancelled Poll returned %v", err)
+ }
+ if w.Usable() {
+ t.Fatal("a cancelled Poll left the watch usable")
+ }
+ moved, err := w.Poll(context.Background())
+ if err != nil || moved != nil || !w.Usable() {
+ t.Fatalf("recovery = %v, %v, usable %v; want nothing moved and usable", moved, err, w.Usable())
+ }
+ seen := c.seen()
+ if last := seen[len(seen)-1]; last == "" {
+ t.Fatalf("the recovery poll was unconditional (etags %q): the give-up dropped the ETag", seen)
+ }
}