From 80ade3cacb2d27a2a384440d9ca01804fdab9f49 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 02:18:22 +0100 Subject: [PATCH] mesh-recovery: every bound a setting, the expiry on a bound key's expiry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit point-keep-days, dump-wait-minutes and snapshot-restore-wait-minutes may be fractions, and expire-every-hours (24 unless set) times the removal of old points by a key of the state bounds whose own time to live the bus ends, heard as its change: no timer in the process. Builds: design 47 G §551, §557, A-77, A-130, A-138 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-recovery/cmd/mesh-recovery/expiry.go | 44 ++++++++++++++ .../mesh-recovery/cmd/mesh-recovery/main.go | 37 ++++++------ .../cmd/mesh-recovery/main_test.go | 59 +++++++++++++++++++ .../mesh-recovery/cmd/mesh-recovery/seats.go | 14 +++-- .../mesh-recovery/files/expire-every-hours | 1 + .../internal/recovery/manifest_test.go | 8 ++- .../mesh-recovery/internal/recovery/points.go | 3 +- .../internal/recovery/recovery.go | 2 +- modules/mesh-recovery/module.json | 33 +++++++++-- 9 files changed, 173 insertions(+), 28 deletions(-) create mode 100644 modules/mesh-recovery/cmd/mesh-recovery/expiry.go create mode 100644 modules/mesh-recovery/files/expire-every-hours diff --git a/modules/mesh-recovery/cmd/mesh-recovery/expiry.go b/modules/mesh-recovery/cmd/mesh-recovery/expiry.go new file mode 100644 index 000000000..14b4ad65d --- /dev/null +++ b/modules/mesh-recovery/cmd/mesh-recovery/expiry.go @@ -0,0 +1,44 @@ +package main + +import ( + "errors" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// boundState is the module's own state bounds, whose keys live a time of their own (key-ttl): a bound key's expiry is +// heard as the bus's own change, so nothing in the process waits on a timer (design 47 G A-77, A-138). +type boundState interface { + CreateTTL(key string, value any, ttl time.Duration) (uint64, error) + Watch(key string, handle func(stdio.StateChange) error) error +} + +// expiryKey is the bound key of §551's daily removal. +const expiryKey = "expiry" + +// expireOnBound runs due each time the bound key `expiry`, written with the time to live `every` (expire-every-hours), is +// removed by the bus, and writes it again (§551, A-138). A key a run before this start wrote is kept, so a restart does not +// put the next removal off; its removal at the start is the start's own run. +func expireOnBound(s boundState, every time.Duration, due func(), logf func(string, ...any)) error { + arm := func() error { + _, err := s.CreateTTL(expiryKey, map[string]any{"every": every.String()}, every) + if errors.Is(err, stdio.ErrChanged) { + return nil + } + return err + } + if err := s.Watch(expiryKey, func(c stdio.StateChange) error { + if c.Op == "put" || c.Current { + return nil + } + due() + if err := arm(); err != nil { + logf("the bound of the next removal of points past their keep days is not written: %v", err) + } + return nil + }); err != nil { + return err + } + return arm() +} diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go index 78987366b..6d374b667 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -36,6 +36,7 @@ const ( envMark = "MESH_RECOVERY_FILLING_MARK_FILE" envDumpWait = "MESH_RECOVERY_DUMP_WAIT_FILE" envRestore = "MESH_RECOVERY_RESTORE_WAIT_FILE" + envExpire = "MESH_RECOVERY_EXPIRE_EVERY_FILE" ) // The settings' defaults: the design's values (§551, §564). @@ -44,8 +45,8 @@ const ( defaultKeepDays = 90 ) -// ExpireEvery is how often §551's removal runs beside its start and each snapshot-kept: once a day. -var ExpireEvery = 24 * time.Hour +// defaultExpireEvery is how often §551's removal runs beside its start and each snapshot-kept: once a day (expire-every-hours). +const defaultExpireEvery = 24 * time.Hour func main() { ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) @@ -85,7 +86,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } - keepDays, err := readNumber(getenv(envKeepDays), defaultKeepDays) + keepDays, err := readFraction(getenv(envKeepDays), defaultKeepDays) if err != nil { return err } @@ -101,6 +102,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + expireEvery, err := readFraction(getenv(envExpire), defaultExpireEvery.Hours()) + if err != nil { + return err + } account, err := process.ReadBrokerAccount(getenv) if err != nil { return err @@ -154,7 +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}, - Settings: recovery.Settings{Cap: cap, KeepDays: int(keepDays), FillingMark: int(mark), DumpWait: dumpWait, RestoreWait: restoreWait}, + Settings: recovery.Settings{Cap: cap, KeepDays: keepDays, FillingMark: int(mark), DumpWait: dumpWait, RestoreWait: restoreWait}, Findings: writer, Node: account.Node, Logf: logger.Printf, @@ -174,18 +179,16 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } } expire("the removal of points past their keep days at the start (design 47 G section 551)") - tick := time.NewTicker(ExpireEvery) - defer tick.Stop() - for { - select { - case <-ctx.Done(): - // A move or a restart stops the process: it takes no new call, the running act ends through its own failure - // path, and every accepted act ends stopped, so its caller asks again (§562, A-141). - e.Stop("its stop signal") - stopAnswers() - return nil - case <-tick.C: - expire("the daily removal of points past their keep days (design 47 G section 551)") - } + every := time.Duration(expireEvery * float64(time.Hour)) + if err := expireOnBound(conn.State("bounds"), every, func() { + expire("the removal of points past their keep days on its bound of " + every.String() + " (design 47 G section 551)") + }, logger.Printf); err != nil { + return fmt.Errorf("the bound of the removal of points past their keep days: %w", err) } + <-ctx.Done() + // A move or a restart stops the process: it takes no new call, the running act ends through its own failure path, and + // every accepted act ends stopped, so its caller asks again (§562, A-141). + e.Stop("its stop signal") + stopAnswers() + return nil } diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main_test.go b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go index a16eede55..700a3514a 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main_test.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go @@ -2,12 +2,15 @@ package main import ( "context" + "os" + "path/filepath" "strings" "testing" "time" "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" ) @@ -71,6 +74,42 @@ func TestWhatTheProcessReads(t *testing.T) { t.Errorf("an outcome said before the wait is not taken: %+v %v", ev, ok) } }) + t.Run("every bound is a setting, a fraction allowed (§557, A-138)", func(t *testing.T) { + dir := t.TempDir() + write := func(name, v string) string { + p := filepath.Join(dir, name) + _ = os.WriteFile(p, []byte(v+"\n"), 0o600) + return p + } + if d, err := readMinutes(write("wait", "0.05"), time.Hour); err != nil || d != 3*time.Second { + t.Errorf("0.05 minutes read %s, %v", d, err) + } + if n, err := readFraction(write("days", "0.00002"), 90); err != nil || n != 0.00002 { + t.Errorf("a fraction of a day read %v, %v", n, err) + } + if n, err := readFraction(filepath.Join(dir, "none"), 24); err != nil || n != 24 { + t.Errorf("a missing setting read %v, %v", n, err) + } + if _, err := readFraction(write("zero", "0"), 24); err == nil { + t.Error("a bound of 0 was taken") + } + }) + t.Run("the expiry runs on its bound key's expiry (§551, A-77, A-138)", func(t *testing.T) { + s := &fakeBounds{} + runs := 0 + if err := expireOnBound(s, 2*time.Second, func() { runs++ }, t.Logf); err != nil { + t.Fatal(err) + } + _ = s.watch(stdio.StateChange{Key: expiryKey, Op: "put"}) + _ = s.watch(stdio.StateChange{Key: expiryKey, Op: "delete"}) + if runs != 1 || len(s.created) != 2 || s.created[1] != 2*time.Second { + t.Errorf("%d runs, keys written %v", runs, s.created) + } + s.exists = true // a key a run before this start wrote stands + if err := expireOnBound(s, time.Hour, func() {}, t.Logf); err != nil { + t.Errorf("a standing bound key is refused: %v", err) + } + }) 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) @@ -89,3 +128,23 @@ func TestWhatTheProcessReads(t *testing.T) { } }) } + +// fakeBounds is the state bounds: the keys written with their time to live, and the watch's handler. +type fakeBounds struct { + created []time.Duration + exists bool + watch func(stdio.StateChange) error +} + +func (f *fakeBounds) CreateTTL(_ string, _ any, ttl time.Duration) (uint64, error) { + if f.exists { + return 0, stdio.ErrChanged + } + f.created = append(f.created, ttl) + return 1, nil +} + +func (f *fakeBounds) Watch(_ string, handle func(stdio.StateChange) error) error { + f.watch = handle + return nil +} diff --git a/modules/mesh-recovery/cmd/mesh-recovery/seats.go b/modules/mesh-recovery/cmd/mesh-recovery/seats.go index 26f0e06c1..688c596fb 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/seats.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/seats.go @@ -206,9 +206,15 @@ func (v *runningView) Fingerprint(module string) (string, error) { return "", fmt.Errorf("%s runs at more than one fingerprint (%s)", module, strings.Join(said, "; ")) } -// readMinutes reads a bound in minutes from its setting's file, a fraction allowed so a whole test sets it in seconds -// (§557, A-130); the default when missing. +// readMinutes reads a bound in minutes from its setting's file (§557, A-130, A-138); the default when missing. func readMinutes(path string, fallback time.Duration) (time.Duration, error) { + n, err := readFraction(path, fallback.Minutes()) + return time.Duration(n * float64(time.Minute)), err +} + +// readFraction reads a bound's setting (minutes, hours, days), a fraction allowed so a whole test sets it in seconds +// (§557, A-138); the default when missing. +func readFraction(path string, fallback float64) (float64, error) { if path == "" { return fallback, nil } @@ -221,9 +227,9 @@ func readMinutes(path string, fallback time.Duration) (time.Duration, error) { } var n float64 if err := json.Unmarshal([]byte(strings.TrimSpace(string(raw))), &n); err != nil || n <= 0 { - return 0, fmt.Errorf("the setting in %s is %q, not a positive number of minutes", path, strings.TrimSpace(string(raw))) + return 0, fmt.Errorf("the setting in %s is %q, not a positive number", path, strings.TrimSpace(string(raw))) } - return time.Duration(n * float64(time.Minute)), nil + return n, nil } // readNumber reads a setting's file (design 47 G A-112): its whole content a positive number, the default when missing. diff --git a/modules/mesh-recovery/files/expire-every-hours b/modules/mesh-recovery/files/expire-every-hours new file mode 100644 index 000000000..68cafde19 --- /dev/null +++ b/modules/mesh-recovery/files/expire-every-hours @@ -0,0 +1 @@ +${config:expire-every-hours} diff --git a/modules/mesh-recovery/internal/recovery/manifest_test.go b/modules/mesh-recovery/internal/recovery/manifest_test.go index 7e904fa1b..185840f3d 100644 --- a/modules/mesh-recovery/internal/recovery/manifest_test.go +++ b/modules/mesh-recovery/internal/recovery/manifest_test.go @@ -70,7 +70,7 @@ func TestTheManifestSaysWhatTheModuleIs(t *testing.T) { t.Errorf("no data item %q", id) } } - for k, want := range map[string]float64{"points-cap": 10 << 30, "points-filling-mark": 80, "point-keep-days": 90, "dump-wait-minutes": 60, "snapshot-restore-wait-minutes": 90} { + for k, want := range map[string]float64{"points-cap": 10 << 30, "points-filling-mark": 80, "point-keep-days": 90, "dump-wait-minutes": 60, "snapshot-restore-wait-minutes": 90, "expire-every-hours": 24} { if m.Config[k].Default != want { t.Errorf("setting %s defaults to %v, want %v", k, m.Config[k].Default, want) } @@ -107,6 +107,7 @@ func TestTheManifestSaysWhatTheModuleIs(t *testing.T) { if m.RuntimeHealth.Kind != "tool" || m.RuntimeHealth.Tool != "recovery_kept_points" || !slices.Contains(m.Tools, m.RuntimeHealth.Tool) { t.Errorf("the tool check is %+v, want recovery_kept_points, a tool of %v", m.RuntimeHealth, m.Tools) } + bounds := false for _, raw := range m.Requires { var q struct { Name string `json:"name"` @@ -115,6 +116,11 @@ func TestTheManifestSaysWhatTheModuleIs(t *testing.T) { if json.Unmarshal(raw, &q) == nil && q.Name == "digests" && len(q.Options) != 0 { t.Errorf("digests gives the options %v beside the template state's fixed cap", q.Options) } + // The expiry's bound is a bound key of its own state, its keys living a time of their own (A-77, A-138). + bounds = bounds || q.Name == "bounds" && q.Options["key-ttl"] == true + } + if !bounds { + t.Error("no state bounds with key-ttl for the expiry's bound key") } for _, c := range []string{"node-backup.item-started", "node-backup.item-ended", "mesh-broker.snapshot-restored"} { if !slices.Contains(m.Consumes, c) { diff --git a/modules/mesh-recovery/internal/recovery/points.go b/modules/mesh-recovery/internal/recovery/points.go index b3d9f1bed..7f904c2f3 100644 --- a/modules/mesh-recovery/internal/recovery/points.go +++ b/modules/mesh-recovery/internal/recovery/points.go @@ -11,6 +11,7 @@ import ( "path/filepath" "sort" "strings" + "time" seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" "git.novox.be/novox/mesh-sdk/go/keys" @@ -209,7 +210,7 @@ func (e *Engine) expire(act string) { newest[k] = p.Snapshot } } - cutoff := e.now().AddDate(0, 0, -e.Settings.KeepDays) + cutoff := e.now().Add(-time.Duration(e.Settings.KeepDays * float64(24*time.Hour))) removed := int64(0) for _, p := range pts { if !p.At.Before(cutoff) || newest[p.Module+"\x00"+p.Kept] == p.Snapshot { diff --git a/modules/mesh-recovery/internal/recovery/recovery.go b/modules/mesh-recovery/internal/recovery/recovery.go index efd0faddf..e5ee57af2 100644 --- a/modules/mesh-recovery/internal/recovery/recovery.go +++ b/modules/mesh-recovery/internal/recovery/recovery.go @@ -99,7 +99,7 @@ type Dump interface { // Settings are the bounds a whole test sets in seconds (§557, §564, §633). type Settings struct { Cap int64 - KeepDays int + KeepDays float64 // FillingMark is the percent of the cap past which points-filling is written (points-filling-mark, 80 unless set). FillingMark int // DumpWait is how long an act waits on the dump before recovery-waits is written (dump-wait-minutes, 60 unless set). diff --git a/modules/mesh-recovery/module.json b/modules/mesh-recovery/module.json index 3f1e6e024..bbfe35474 100644 --- a/modules/mesh-recovery/module.json +++ b/modules/mesh-recovery/module.json @@ -68,6 +68,14 @@ "storage": "key-value", "template": "state" }, + { + "name": "bounds", + "storage": "key-value", + "template": "state", + "options": { + "key-ttl": true + } + }, { "name": "finding-events", "storage": "stream", @@ -102,7 +110,7 @@ "point-keep-days": { "type": "number", "description": "days after which a restore point is removed, the newest of its kept data never (design 47 G sections 551, 557).", - "minimum": 1, + "minimum": 0, "maximum": 3650, "trusted": false, "default": 90 @@ -110,7 +118,7 @@ "dump-wait-minutes": { "type": "number", "description": "minutes an act waits on the dump of the bus's item before recovery-waits is written (design 47 G sections 561, 557).", - "minimum": 1, + "minimum": 0, "maximum": 10080, "trusted": false, "default": 60 @@ -118,10 +126,18 @@ "snapshot-restore-wait-minutes": { "type": "number", "description": "minutes a restore in place of a protected keeper's snapshot is waited on, and once more after its cancel (design 47 G section 681, A-138, A-140).", - "minimum": 1, + "minimum": 0, "maximum": 10080, "trusted": false, "default": 90 + }, + "expire-every-hours": { + "type": "number", + "description": "hours between two removals of the points past their keep days, beside the one at the start and at each snapshot-kept; timed by a bound key of the state bounds (design 47 G sections 551, 557, A-138).", + "minimum": 0, + "maximum": 8760, + "trusted": false, + "default": 24 } }, "data": { @@ -202,6 +218,14 @@ "mode": "0600", "template": "files/snapshot-restore-wait-minutes" }, + { + "id": "expire-every", + "type": "file", + "path": "${dir:state}/expire-every-hours", + "owner": "mesh-recovery", + "mode": "0600", + "template": "files/expire-every-hours" + }, { "id": "recovery", "type": "process", @@ -217,7 +241,8 @@ "MESH_RECOVERY_FILLING_MARK_FILE": "${dir:state}/points-filling-mark", "MESH_RECOVERY_KEEP_DAYS_FILE": "${dir:state}/point-keep-days", "MESH_RECOVERY_DUMP_WAIT_FILE": "${dir:state}/dump-wait-minutes", - "MESH_RECOVERY_RESTORE_WAIT_FILE": "${dir:state}/snapshot-restore-wait-minutes" + "MESH_RECOVERY_RESTORE_WAIT_FILE": "${dir:state}/snapshot-restore-wait-minutes", + "MESH_RECOVERY_EXPIRE_EVERY_FILE": "${dir:state}/expire-every-hours" }, "user": "mesh-recovery", "serves": [