From bf7d040489acff68f8198cc8ca34d92ff17d9eb3 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 02:24:44 +0100 Subject: [PATCH] mesh-recovery A-143: the dump wait and the snapshot-restored wait are bound keys MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Each wait's bound is a key of the module's state bounds with its own time to live, dump-wait. and restore-wait., its expiry heard on the key's watch, so no timer stays in the process. Builds: design 47 G A-143, A-77, A-138, §681, §561 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-recovery/cmd/mesh-recovery/expiry.go | 38 ++++++++++++++++ .../mesh-recovery/cmd/mesh-recovery/main.go | 1 + .../cmd/mesh-recovery/main_test.go | 31 +++++++++++++ .../internal/recovery/fakes_test.go | 38 +++++++++++++++- .../internal/recovery/findings.go | 25 +++++++---- .../mesh-recovery/internal/recovery/keeper.go | 43 ++++++++++++++++--- .../internal/recovery/keeper_test.go | 20 ++++++--- .../internal/recovery/recovery.go | 18 +++++++- .../internal/recovery/restore_test.go | 4 ++ 9 files changed, 194 insertions(+), 24 deletions(-) diff --git a/modules/mesh-recovery/cmd/mesh-recovery/expiry.go b/modules/mesh-recovery/cmd/mesh-recovery/expiry.go index 14b4ad65d..9dfb5b96a 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/expiry.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/expiry.go @@ -2,9 +2,12 @@ package main import ( "errors" + "sync/atomic" "time" stdio "git.novox.be/novox/mesh-sdk/go" + + "mesh-recovery/internal/recovery" ) // boundState is the module's own state bounds, whose keys live a time of their own (key-ttl): a bound key's expiry is @@ -12,8 +15,43 @@ import ( type boundState interface { CreateTTL(key string, value any, ttl time.Duration) (uint64, error) Watch(key string, handle func(stdio.StateChange) error) error + Delete(key string) error } +// errBoundEnded ends a bound key's watch once the key is gone. +var errBoundEnded = errors.New("the bound key is removed") + +// stateBounds times an act's waits by bound keys of the state bounds (A-77, A-143): the key is watched before it is +// written, so its expiry is never missed, and the watch ends with the key's removal. +type stateBounds struct{ s boundState } + +func (b stateBounds) Bound(key string, ttl time.Duration, expired func()) (func(), error) { + var ended atomic.Bool + if err := b.s.Watch(key, func(c stdio.StateChange) error { + if c.Op == "put" || c.Current { + return nil + } + if !ended.Swap(true) { + expired() + } + return errBoundEnded + }); err != nil { + return nil, err + } + if _, err := b.s.CreateTTL(key, map[string]any{"ttl": ttl.String()}, ttl); err != nil { + ended.Store(true) + _ = b.s.Delete(key) + return nil, err + } + return func() { + if !ended.Swap(true) { + _ = b.s.Delete(key) + } + }, nil +} + +var _ recovery.Bounds = stateBounds{} + // expiryKey is the bound key of §551's daily removal. const expiryKey = "expiry" diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go index 6d374b667..3c50f938c 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -159,6 +159,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro Digests: digests{conn.State("digests")}, Dump: dump, Keeper: liveKeeper{nc: nc, call: conn.Call, restored: restored}, + Bounds: stateBounds{conn.State("bounds")}, Settings: recovery.Settings{Cap: cap, KeepDays: keepDays, FillingMark: int(mark), DumpWait: dumpWait, RestoreWait: restoreWait}, Findings: writer, Node: account.Node, diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main_test.go b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go index 700a3514a..e0e4482d8 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main_test.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go @@ -110,6 +110,31 @@ func TestWhatTheProcessReads(t *testing.T) { t.Errorf("a standing bound key is refused: %v", err) } }) + t.Run("an act's wait is a bound key, its expiry heard on its watch (A-143)", func(t *testing.T) { + s := &fakeBounds{} + expired := 0 + end, err := stateBounds{s}.Bound("dump-wait.act-1", 3*time.Second, func() { expired++ }) + if err != nil || len(s.created) != 1 || s.created[0] != 3*time.Second { + t.Fatalf("%v, keys written %v", err, s.created) + } + if s.watch(stdio.StateChange{Key: "dump-wait.act-1", Op: "put"}) != nil || expired != 0 { + t.Error("the key's own write ended the wait") + } + if s.watch(stdio.StateChange{Key: "dump-wait.act-1", Op: "delete"}) == nil || expired != 1 { + t.Errorf("the key's expiry: %d calls, its watch not ended", expired) + } + end() + if expired != 1 || len(s.deleted) != 0 { + t.Errorf("an expired key's end: %d calls, deleted %v", expired, s.deleted) + } + s = &fakeBounds{} + end, _ = stateBounds{s}.Bound("restore-wait.act-2", time.Minute, func() { expired++ }) + end() + if s.watch(stdio.StateChange{Key: "restore-wait.act-2", Op: "delete"}) == nil || expired != 1 || + strings.Join(s.deleted, " ") != "restore-wait.act-2" { + t.Errorf("an ended key's removal: %d calls, deleted %v", expired, s.deleted) + } + }) t.Run("a dump of the bus module runs (§563)", func(t *testing.T) { d := newDumpView(nil, nil) start := time.Date(2026, 11, 12, 2, 0, 0, 0, time.UTC) @@ -134,6 +159,7 @@ type fakeBounds struct { created []time.Duration exists bool watch func(stdio.StateChange) error + deleted []string } func (f *fakeBounds) CreateTTL(_ string, _ any, ttl time.Duration) (uint64, error) { @@ -148,3 +174,8 @@ func (f *fakeBounds) Watch(_ string, handle func(stdio.StateChange) error) error f.watch = handle return nil } + +func (f *fakeBounds) Delete(key string) error { + f.deleted = append(f.deleted, key) + return nil +} diff --git a/modules/mesh-recovery/internal/recovery/fakes_test.go b/modules/mesh-recovery/internal/recovery/fakes_test.go index 38fca691b..254956aa7 100644 --- a/modules/mesh-recovery/internal/recovery/fakes_test.go +++ b/modules/mesh-recovery/internal/recovery/fakes_test.go @@ -6,6 +6,7 @@ import ( "io" "os" "path/filepath" + "strings" "sync" "testing" "time" @@ -192,14 +193,16 @@ type rig struct { s *fakeSeats d *memDigests t time.Time + // bounds is the state bounds the act's waits are timed by (A-143). + bounds *fakeBounds } func newRig(t *testing.T) *rig { t.Helper() - r := &rig{b: &fakeBus{streams: map[string]string{"app_notes": "hello", "KV_app_seen": "world"}, consumers: map[string][]seat.Consumer{}}, + r := &rig{bounds: &fakeBounds{}, b: &fakeBus{streams: map[string]string{"app_notes": "hello", "KV_app_seen": "world"}, consumers: map[string][]seat.Consumer{}}, s: &fakeSeats{streams: []string{"app_notes", "KV_app_seen"}}, d: &memDigests{m: map[string][]byte{}}, t: time.Date(2026, 11, 12, 1, 0, 0, 0, time.UTC)} r.b.consumers["app_notes"] = []seat.Consumer{{Name: "c1", Stream: "app_notes", Sequence: 7}} - r.e = &recovery.Engine{Dir: filepath.Join(t.TempDir(), "points-dir"), Bus: r.b, Seats: r.s, Digests: r.d, Dump: noDump{}, + r.e = &recovery.Engine{Dir: filepath.Join(t.TempDir(), "points-dir"), Bus: r.b, Seats: r.s, Digests: r.d, Dump: noDump{}, Bounds: r.bounds, Settings: recovery.Settings{Cap: 1 << 20, KeepDays: 90}, Now: func() time.Time { return r.t }} return r } @@ -266,3 +269,34 @@ func (h *heldFindings) has(kind string) bool { } return false } + +// fakeBounds is the state bounds: each key removed by the bus when its time to live passes, as the bus does, or by its +// end; written and removed say which keys, with each key's time to live (A-143). +type fakeBounds struct { + mu sync.Mutex + written []string + removed []string +} + +func (f *fakeBounds) Bound(key string, ttl time.Duration, expired func()) (func(), error) { + f.mu.Lock() + f.written = append(f.written, key+" "+ttl.String()) + f.mu.Unlock() + var once sync.Once + gone := func(by string, then func()) { + once.Do(func() { + f.mu.Lock() + f.removed = append(f.removed, key+" "+by) + f.mu.Unlock() + then() + }) + } + t := time.AfterFunc(ttl, func() { gone("expired", expired) }) + return func() { t.Stop(); gone("ended", func() {}) }, nil +} + +func (f *fakeBounds) keys() (written, removed string) { + f.mu.Lock() + defer f.mu.Unlock() + return strings.Join(f.written, "|"), strings.Join(f.removed, "|") +} diff --git a/modules/mesh-recovery/internal/recovery/findings.go b/modules/mesh-recovery/internal/recovery/findings.go index 09fa7310a..ca4be0c11 100644 --- a/modules/mesh-recovery/internal/recovery/findings.go +++ b/modules/mesh-recovery/internal/recovery/findings.go @@ -124,25 +124,32 @@ func (e *Engine) largest(n int) []string { return out } -// waitsOn is the waiting of one act on the dump: past the bound it writes recovery-waits, naming the item and since when; -// done clears it once the act starts its work or ends (§561, §563). +// waitsOn is the waiting of one act on the dump: past the bound, the key `dump-wait.` of the state bounds expiring +// (A-143), it writes recovery-waits, naming the item and since when; done removes the key and clears the finding once the +// act starts its work or ends (§561, §563). func (e *Engine) waitsOn(a *Act) (waiting func(on string), done func()) { - var timer *time.Timer + var end func() written := make(chan bool, 1) written <- false waiting = func(on string) { e.setStep(a, "waiting", on) since := e.now() - timer = time.AfterFunc(e.dumpWait(), func() { - <-written - e.write(e.finding(FindingRecoveryWaits, a.Module, a.ID, "A restore waits on the dump of the bus", - fmt.Sprintf("the act %s waits on %s since %s", a.ID, on, since.Format(time.RFC3339)), "the dump to end, or recovery_cancel")) + key := BoundDumpWait + "." + a.ID + var err error + end, err = e.bound(key, e.dumpWait(), func() { + if !<-written { + e.write(e.finding(FindingRecoveryWaits, a.Module, a.ID, "A restore waits on the dump of the bus", + fmt.Sprintf("the act %s waits on %s since %s", a.ID, on, since.Format(time.RFC3339)), "the dump to end, or recovery_cancel")) + } written <- true }) + if err != nil { + e.logf("the bound %s of the act's wait on the dump is not written, so recovery-waits is never written for it: %v", key, err) + } } done = func() { - if timer != nil { - timer.Stop() + if end != nil { + end() } if <-written { e.clear(a.Module, a.ID, FindingRecoveryWaits, "the act no longer waits on the dump") diff --git a/modules/mesh-recovery/internal/recovery/keeper.go b/modules/mesh-recovery/internal/recovery/keeper.go index 5c8167caa..38dde2e82 100644 --- a/modules/mesh-recovery/internal/recovery/keeper.go +++ b/modules/mesh-recovery/internal/recovery/keeper.go @@ -144,24 +144,39 @@ func (e *Engine) restoreInPlace(ctx context.Context, a *Act) error { return e.endedRestored(a, why, asidePtr(rec), false) } e.setStep(a, "restoring in place", "the keeper's snapshot-restored") + // The wait's bound is the key `restore-wait.` of the state bounds, written before the keeper is asked, so a bound + // that cannot be written asks nothing of the keeper (A-143). + wctx, passed, err := e.restoreBound(ctx, a.ID) + if err != nil { + return putBack(err) + } r, err := e.Keeper.Restore(ctx, a.Request, false, a.Why) if err == nil { err = keeperAnswer(r) } if err != nil { + passed() return putBack(err) } act := r.Accepted - ev, ok := e.restoredWithin(ctx, a.Request, act) + ev, ok := e.Keeper.Restored(wctx, a.Request, act) + passed() if !ok { e.setStep(a, "cancelling the keeper's restore", "the keeper's snapshot-restored after the cancel") why := fmt.Sprintf("the recovery module's wait of %s on the restore in place passed (act %s)", e.restoreWait(), a.ID) if c, err := e.Keeper.CancelRestore(context.Background(), act, why); err != nil || c.Error != "" { e.logf("the cancel of the keeper's restore %s is not accepted (%v%s); its outcome is waited on once more", act, err, c.Error) } - if ev, ok = e.restoredWithin(context.Background(), a.Request, act); !ok { - return e.endedRestored(a, errors.New(ErrUnanswered), asidePtr(rec), false) + wctx, passed, err := e.restoreBound(context.Background(), a.ID) + if err != nil { + e.logf("the bound of the wait after the cancel of %s is not written: %v", act, err) + return e.endedRestored(a, fmt.Errorf("%s: %v", ErrUnanswered, err), asidePtr(rec), false) } + ev, ok = e.Keeper.Restored(wctx, a.Request, act) + passed() + } + if !ok { + return e.endedRestored(a, errors.New(ErrUnanswered), asidePtr(rec), false) } if ev.Outcome != meshbroker.SnapshotRestoredEventOutcomeDone { return putBack(fmt.Errorf("the keeper's restore in place failed: %s", ev.Error)) @@ -171,10 +186,24 @@ func (e *Engine) restoreInPlace(ctx context.Context, a *Act) error { return e.endedRestored(a, nil, asidePtr(rec), false) } -func (e *Engine) restoredWithin(ctx context.Context, request, act string) (meshbroker.SnapshotRestoredEvent, bool) { - wctx, cancel := context.WithTimeout(ctx, e.restoreWait()) - defer cancel() - return e.Keeper.Restored(wctx, request, act) +// restoreBound is the bound of one wait on snapshot-restored (§681, A-138, A-143): the key `restore-wait.` with the time +// to live snapshot-restore-wait-minutes; its expiry ends the context it answers. end removes the key. +func (e *Engine) restoreBound(ctx context.Context, act string) (context.Context, func(), error) { + wctx, cancel := context.WithCancel(ctx) + end, err := e.bound(BoundRestoreWait+"."+act, e.restoreWait(), cancel) + if err != nil { + cancel() + return nil, nil, err + } + return wctx, func() { end(); cancel() }, nil +} + +// bound writes a bound key in the state bounds (A-77, A-143). +func (e *Engine) bound(key string, ttl time.Duration, expired func()) (func(), error) { + if e.Bounds == nil { + return nil, fmt.Errorf("the bound %s is not written: the engine has no state bounds", key) + } + return e.Bounds.Bound(key, ttl, expired) } // keeperAnswer reads snapshot-restore's one form (§691): accepted, or an error naming the keeper's refusal or its running act. diff --git a/modules/mesh-recovery/internal/recovery/keeper_test.go b/modules/mesh-recovery/internal/recovery/keeper_test.go index 5084e2b76..f816966ab 100644 --- a/modules/mesh-recovery/internal/recovery/keeper_test.go +++ b/modules/mesh-recovery/internal/recovery/keeper_test.go @@ -174,12 +174,13 @@ func TestAProtectedSnapshotIsRestoredInPlaceByTheKeeper(t *testing.T) { asked string live string unrestored int + bounds string // how each bound key restore-wait. was removed (A-143) }{ - {inPlace: "done", asked: "get req-1|verify req-1|restore req-1", live: "as dropped"}, - {inPlace: "failed", asked: "get req-1|verify req-1|restore req-1", live: "hello", err: "the keeper's restore in place failed"}, - {inPlace: "silent", asked: "get req-1|verify req-1|restore req-1|cancel keeper-act-1", live: "hello", err: "the keeper's restore in place failed: cancelled"}, - {inPlace: "deaf", asked: "get req-1|verify req-1|restore req-1|cancel keeper-act-1", err: recovery.ErrUnanswered, unrestored: 1}, - {inPlace: "done", busy: "keeper-act-0", asked: "get req-1|verify req-1|restore req-1", live: "hello", err: "busy: the keeper restores its act keeper-act-0"}, + {inPlace: "done", asked: "get req-1|verify req-1|restore req-1", live: "as dropped", bounds: "ended"}, + {inPlace: "failed", asked: "get req-1|verify req-1|restore req-1", live: "hello", err: "the keeper's restore in place failed", bounds: "ended"}, + {inPlace: "silent", asked: "get req-1|verify req-1|restore req-1|cancel keeper-act-1", live: "hello", err: "the keeper's restore in place failed: cancelled", bounds: "expired|ended"}, + {inPlace: "deaf", asked: "get req-1|verify req-1|restore req-1|cancel keeper-act-1", err: recovery.ErrUnanswered, unrestored: 1, bounds: "expired|expired"}, + {inPlace: "done", busy: "keeper-act-0", asked: "get req-1|verify req-1|restore req-1", live: "hello", err: "busy: the keeper restores its act keeper-act-0", bounds: "ended"}, } { t.Run(tc.inPlace+tc.busy, func(t *testing.T) { r, k := requestRig(t) @@ -201,6 +202,15 @@ func TestAProtectedSnapshotIsRestoredInPlaceByTheKeeper(t *testing.T) { if n := len(r.e.Unrestored()); n != tc.unrestored { t.Errorf("%d unrestored asides, want %d", n, tc.unrestored) } + w, rm := r.bounds.keys() + var wantW, wantR []string + for _, by := range strings.Split(tc.bounds, "|") { + wantW = append(wantW, "restore-wait."+ev.Act+" 20ms") + wantR = append(wantR, "restore-wait."+ev.Act+" "+by) + } + if w != strings.Join(wantW, "|") || rm != strings.Join(wantR, "|") { + t.Errorf("bound keys written %q, removed %q; want %q, %q", w, rm, strings.Join(wantW, "|"), strings.Join(wantR, "|")) + } }) } } diff --git a/modules/mesh-recovery/internal/recovery/recovery.go b/modules/mesh-recovery/internal/recovery/recovery.go index e5ee57af2..eeae6ea63 100644 --- a/modules/mesh-recovery/internal/recovery/recovery.go +++ b/modules/mesh-recovery/internal/recovery/recovery.go @@ -96,6 +96,20 @@ type Dump interface { Wait(ctx context.Context, waiting func(on string)) error } +// Bounds is the module's state bounds (design 47 G A-77, A-143): a wait's bound is a key with its own time to live, its +// expiry heard on the key's watch as the bus's own change, so no timer runs in the process. +type Bounds interface { + // Bound writes key with the time to live ttl and calls expired once when the bus removes it; end removes it first, and + // expired is then never called. + Bound(key string, ttl time.Duration, expired func()) (end func(), err error) +} + +// The bound keys of an act's waits (A-143): `.`. +const ( + BoundDumpWait = "dump-wait" + BoundRestoreWait = "restore-wait" +) + // Settings are the bounds a whole test sets in seconds (§557, §564, §633). type Settings struct { Cap int64 @@ -117,7 +131,9 @@ type Engine struct { Digests Digests Dump Dump // Keeper is the bus module's keeper, whose snapshots the hand verb restore takes by request (§681). - Keeper Keeper + Keeper Keeper + // Bounds times the dump wait and the snapshot-restored wait (A-143). + Bounds Bounds Settings Settings // Findings is the module's finding writer (§561); Node the machine it runs on, the id of points-filling. Findings Findings diff --git a/modules/mesh-recovery/internal/recovery/restore_test.go b/modules/mesh-recovery/internal/recovery/restore_test.go index cc168ee6c..b63c44fe7 100644 --- a/modules/mesh-recovery/internal/recovery/restore_test.go +++ b/modules/mesh-recovery/internal/recovery/restore_test.go @@ -164,6 +164,10 @@ func TestAWaitingActIsCancelledAndABegunOneRefused(t *testing.T) { if fs.has(recovery.FindingRecoveryWaits) { t.Errorf("recovery-waits stands after the act ended: %+v", fs.Held()) } + // The wait's bound is a key of the state bounds, its expiry what wrote recovery-waits (A-143). + if w, rm := r.bounds.keys(); w != "dump-wait."+ans.Accepted+" 1ms" || rm != "dump-wait."+ans.Accepted+" expired" { + t.Errorf("bound keys written %q, removed %q", w, rm) + } } // retry asks again the last failed act with its fields under an act of its own, and answers `nothing to retry` when its