mesh-recovery A-143: the dump wait and the snapshot-restored wait are bound keys

Each wait's bound is a key of the module's state bounds with its own time to
live, dump-wait.<act> and restore-wait.<act>, 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
This commit is contained in:
2026-11-12 02:24:44 +01:00
parent 80ade3cacb
commit bf7d040489
9 changed files with 194 additions and 24 deletions
@@ -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"
@@ -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,
@@ -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
}
@@ -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, "|")
}
@@ -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.<act>` 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")
@@ -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.<act>` 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.<act>` 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.
@@ -174,12 +174,13 @@ func TestAProtectedSnapshotIsRestoredInPlaceByTheKeeper(t *testing.T) {
asked string
live string
unrestored int
bounds string // how each bound key restore-wait.<act> 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, "|"))
}
})
}
}
@@ -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): `<kind>.<act>`.
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
@@ -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