mesh-recovery: every bound a setting, the expiry on a bound key's expiry
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
This commit is contained in:
@@ -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()
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
${config:expire-every-hours}
|
||||
@@ -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) {
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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": [
|
||||
|
||||
Reference in New Issue
Block a user