diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go index 91293e1fc..8f48f11a5 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -18,8 +18,10 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/seatlist" + "git.novox.be/novox/mesh-sdk/go/views" "github.com/nats-io/nats.go" @@ -31,6 +33,8 @@ const ( envPoints = "MESH_RECOVERY_POINTS" envCap = "MESH_RECOVERY_POINTS_CAP_FILE" envKeepDays = "MESH_RECOVERY_KEEP_DAYS_FILE" + envMark = "MESH_RECOVERY_FILLING_MARK_FILE" + envDumpWait = "MESH_RECOVERY_DUMP_WAIT_FILE" ) // The settings' defaults: the design's values (§551, §564). @@ -84,6 +88,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + mark, err := readNumber(getenv(envMark), recovery.DefaultFillingMark) + if err != nil { + return err + } + dumpWait, err := readMinutes(getenv(envDumpWait), recovery.DefaultDumpWait) + if err != nil { + return err + } account, err := process.ReadBrokerAccount(getenv) if err != nil { return err @@ -111,6 +123,18 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } } + // The module's findings go through mesh-sdk's held writer over its own state findings (§561), started before the + // engine's start, so what the start writes is held and written. + writer, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: account.Module, Node: account.Node, State: conn.State("findings"), + Kinds: recovery.FindingKinds, JS: js, Conn: nc, Logf: logger.Printf}) + if err != nil { + return err + } + if err := writer.Start(ctx); err != nil { + return err + } + defer writer.Close() e := &recovery.Engine{ Dir: dir, Bus: liveBus{nc: nc, js: js, scratch: dir, logf: logger.Printf}, @@ -119,7 +143,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro node: account.Node, points: dir, running: running}, Digests: digests{conn.State("digests")}, Dump: dump, - Settings: recovery.Settings{Cap: cap, KeepDays: int(keepDays)}, + Settings: recovery.Settings{Cap: cap, KeepDays: int(keepDays), FillingMark: int(mark), DumpWait: dumpWait}, + Findings: writer, + Node: account.Node, Logf: logger.Printf, } for _, x := range e.Start(ctx) { diff --git a/modules/mesh-recovery/cmd/mesh-recovery/seats.go b/modules/mesh-recovery/cmd/mesh-recovery/seats.go index 58937e038..26f0e06c1 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/seats.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/seats.go @@ -10,6 +10,7 @@ import ( "sort" "strings" "sync" + "time" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" @@ -205,6 +206,26 @@ 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. +func readMinutes(path string, fallback time.Duration) (time.Duration, error) { + if path == "" { + return fallback, nil + } + raw, err := os.ReadFile(path) + if errors.Is(err, os.ErrNotExist) { + return fallback, nil + } + if err != nil { + return 0, err + } + 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 time.Duration(n * float64(time.Minute)), nil +} + // readNumber reads a setting's file (design 47 G A-112): its whole content a positive number, the default when missing. func readNumber(path string, fallback int64) (int64, error) { if path == "" { diff --git a/modules/mesh-recovery/go.mod b/modules/mesh-recovery/go.mod index 8056787cb..87e7e287a 100644 --- a/modules/mesh-recovery/go.mod +++ b/modules/mesh-recovery/go.mod @@ -7,6 +7,7 @@ require ( git.novox.be/novox/mesh-contracts/go/seats/meshrecovery v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/go/seats/nodeengine v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient v0.0.0-00010101000000-000000000000 + git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog v0.0.0-00010101000000-000000000000 git.novox.be/novox/mesh-sdk/go v0.0.0-00010101000000-000000000000 github.com/nats-io/nats.go v1.54.0 diff --git a/modules/mesh-recovery/internal/recovery/control.go b/modules/mesh-recovery/internal/recovery/control.go index 66e725d4e..941b22ccf 100644 --- a/modules/mesh-recovery/internal/recovery/control.go +++ b/modules/mesh-recovery/internal/recovery/control.go @@ -165,6 +165,7 @@ func (e *Engine) Start(ctx context.Context) []UnrestoredAside { a.Error = ErrStopped e.markEnded(&a) } + e.judge() return e.Unrestored() } diff --git a/modules/mesh-recovery/internal/recovery/fakes_test.go b/modules/mesh-recovery/internal/recovery/fakes_test.go index 69c471955..38fca691b 100644 --- a/modules/mesh-recovery/internal/recovery/fakes_test.go +++ b/modules/mesh-recovery/internal/recovery/fakes_test.go @@ -11,6 +11,7 @@ import ( "time" seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" "mesh-recovery/internal/recovery" ) @@ -220,3 +221,48 @@ func (g *gatedBus) Objects(ctx context.Context, n []string) ([]recovery.Object, <-g.gate return g.fakeBus.Objects(ctx, n) } + +// heldFindings is the module's finding writer as it holds them; written says each write. +type heldFindings struct { + mu sync.Mutex + held map[string]meshconditions.Finding + written chan meshconditions.Finding +} + +func newHeldFindings() *heldFindings { + return &heldFindings{held: map[string]meshconditions.Finding{}, written: make(chan meshconditions.Finding, 64)} +} + +func (h *heldFindings) Write(_ context.Context, f meshconditions.Finding) error { + h.mu.Lock() + h.held[f.Scope+"."+f.ID+"."+f.Kind] = f + h.mu.Unlock() + h.written <- f + return nil +} + +func (h *heldFindings) Clear(_ context.Context, scope, id, kind, _ string) error { + h.mu.Lock() + defer h.mu.Unlock() + delete(h.held, scope+"."+id+"."+kind) + return nil +} + +func (h *heldFindings) Held() []meshconditions.Finding { + h.mu.Lock() + defer h.mu.Unlock() + var out []meshconditions.Finding + for _, f := range h.held { + out = append(out, f) + } + return out +} + +func (h *heldFindings) has(kind string) bool { + for _, f := range h.Held() { + if f.Kind == kind { + return true + } + } + return false +} diff --git a/modules/mesh-recovery/internal/recovery/findings.go b/modules/mesh-recovery/internal/recovery/findings.go new file mode 100644 index 000000000..09fa7310a --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/findings.go @@ -0,0 +1,153 @@ +package recovery + +import ( + "context" + "fmt" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" +) + +// The recovery module's findings (design 47 G §561), each written by mesh-sdk's finding writer as the key +// `..` and said on the module's finding stream. +const ( + FindingPointsFilling = "points-filling" + FindingRecoveryWaits = "recovery-waits" + FindingAsideUnrestored = "aside-unrestored" +) + +// FindingKinds are the kinds the module's writer owns. +var FindingKinds = []string{FindingPointsFilling, FindingRecoveryWaits, FindingAsideUnrestored} + +// Findings is the module's finding writer (mesh-sdk's held writer): what it writes and clears, and what it holds. +type Findings interface { + Write(ctx context.Context, f meshconditions.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error + Held() []meshconditions.Finding +} + +// The defaults of the settings §557 and §564 make bounds of. +const ( + DefaultFillingMark = 80 + DefaultDumpWait = 60 * time.Minute +) + +func (e *Engine) mark() int64 { + if e.Settings.FillingMark > 0 { + return int64(e.Settings.FillingMark) + } + return DefaultFillingMark +} + +func (e *Engine) dumpWait() time.Duration { + if e.Settings.DumpWait > 0 { + return e.Settings.DumpWait + } + return DefaultDumpWait +} + +func (e *Engine) finding(kind, scope, id, headline, summary, needs string) meshconditions.Finding { + return meshconditions.Finding{Scope: scope, ID: id, Kind: kind, Subject: meshconditions.Subject{Scope: "module", ID: Module}, + Severity: meshconditions.FindingSeverityWarning, Headline: headline, Summary: summary, Explanation: summary, Needs: needs, + Source: "points", Observer: Module, At: e.now()} +} + +func (e *Engine) write(f meshconditions.Finding) { + if e.Findings == nil { + return + } + if err := e.Findings.Write(context.Background(), f); err != nil { + e.logf("the finding %s %s.%s cannot be written: %v", f.Kind, f.Scope, f.ID, err) + } +} + +func (e *Engine) clear(scope, id, kind, why string) { + if e.Findings == nil { + return + } + if err := e.Findings.Clear(context.Background(), scope, id, kind, why); err != nil { + e.logf("the finding %s %s.%s cannot be cleared: %v", kind, scope, id, err) + } +} + +// judge measures the points directory against the fill mark of its cap (points-filling) and writes aside-unrestored for +// each aside whose record ends at removed that no running act holds, clearing those that no longer are (§561). It runs at +// the start, after each act and so after each expiry run. +func (e *Engine) judge() { + if e.Findings == nil { + return + } + fill, limit := e.filled(), e.Settings.Cap*e.mark()/100 + if fill > limit { + e.write(e.finding(FindingPointsFilling, Module, e.Node, "The recovery module's restore points are filling up", + fmt.Sprintf("the points directory holds %d of %d bytes, past its fill mark of %d %%; the largest: %s", fill, e.Settings.Cap, + e.mark(), strings.Join(e.largest(5), ", ")), "points removed or an aside discarded, or a larger points-cap")) + } else { + e.clear(Module, e.Node, FindingPointsFilling, fmt.Sprintf("the points directory holds %d bytes, under its fill mark", fill)) + } + open := map[string]bool{} + for _, u := range e.Unrestored() { + open[u.Module+"\x00"+u.Aside.ID] = true + e.write(e.finding(FindingAsideUnrestored, u.Module, u.Aside.ID, "An aside holds data the live bus lacks", + fmt.Sprintf("the aside %s of %s's %s (%s) ended at removed and was not restored", u.Aside.ID, u.Module, u.Kept, + strings.Join(u.Aside.Names, ", ")), "recovery_restore_aside, or recovery_discard_aside")) + } + for _, f := range e.Findings.Held() { + if f.Kind == FindingAsideUnrestored && !open[f.Scope+"\x00"+f.ID] { + e.clear(f.Scope, f.ID, f.Kind, "the aside was restored or discarded") + } + } +} + +// largest are the n largest points and asides, each with its size. +func (e *Engine) largest(n int) []string { + type sized struct { + what string + size int64 + } + var all []sized + pts, _ := e.Points() + for _, p := range pts { + all = append(all, sized{"point " + p.Snapshot, p.Size}) + } + _, recs, _ := e.Asides() + for _, r := range recs { + all = append(all, sized{"aside " + r.Module + "/" + r.Kept + "/" + r.ID, r.Size}) + } + sort.SliceStable(all, func(i, j int) bool { return all[i].size > all[j].size }) + var out []string + for i := 0; i < len(all) && i < n; i++ { + out = append(out, fmt.Sprintf("%s (%d bytes)", all[i].what, all[i].size)) + } + 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). +func (e *Engine) waitsOn(a *Act) (waiting func(on string), done func()) { + var timer *time.Timer + 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")) + written <- true + }) + } + done = func() { + if timer != nil { + timer.Stop() + } + if <-written { + e.clear(a.Module, a.ID, FindingRecoveryWaits, "the act no longer waits on the dump") + } + written <- false + } + return waiting, done +} diff --git a/modules/mesh-recovery/internal/recovery/recovery.go b/modules/mesh-recovery/internal/recovery/recovery.go index 91c1ab8f7..f37c7b704 100644 --- a/modules/mesh-recovery/internal/recovery/recovery.go +++ b/modules/mesh-recovery/internal/recovery/recovery.go @@ -100,6 +100,10 @@ type Dump interface { type Settings struct { Cap int64 KeepDays int + // 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). + DumpWait time.Duration } // Engine is the recovery module's one engine. @@ -110,6 +114,9 @@ type Engine struct { Digests Digests Dump Dump Settings Settings + // Findings is the module's finding writer (§561); Node the machine it runs on, the id of points-filling. + Findings Findings + Node string Now func() time.Time Logf func(string, ...any) @@ -253,6 +260,7 @@ func (e *Engine) Accept(a Act) Answer { e.running, e.cancel = nil, nil e.mu.Unlock() cancel() + e.judge() }() return Answer{Accepted: a.ID} } diff --git a/modules/mesh-recovery/internal/recovery/restore.go b/modules/mesh-recovery/internal/recovery/restore.go index fb6803b55..4fdbdfd20 100644 --- a/modules/mesh-recovery/internal/recovery/restore.go +++ b/modules/mesh-recovery/internal/recovery/restore.go @@ -81,7 +81,10 @@ func (e *Engine) restore(ctx context.Context, a *Act) error { return e.endedRestored(a, fmt.Errorf("restore-started was not acknowledged by the bus, so nothing was read or set aside: %w", err), nil, false) } if e.Dump != nil { - if err := e.Dump.Wait(ctx, func(on string) { e.setStep(a, "waiting", on) }); err != nil { + waiting, done := e.waitsOn(a) + err := e.Dump.Wait(ctx, waiting) + done() + if err != nil { if ctx.Err() != nil { err = errors.New(ErrCanceled) } diff --git a/modules/mesh-recovery/internal/recovery/restore_test.go b/modules/mesh-recovery/internal/recovery/restore_test.go index de2cbe0ff..88031c462 100644 --- a/modules/mesh-recovery/internal/recovery/restore_test.go +++ b/modules/mesh-recovery/internal/recovery/restore_test.go @@ -6,6 +6,7 @@ import ( "path/filepath" "strings" "testing" + "time" seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" @@ -137,8 +138,14 @@ func TestAWaitingActIsCancelledAndABegunOneRefused(t *testing.T) { r := takenRig(t) hd := holdDump{waiting: make(chan string, 1)} r.e.Dump = hd + // Past the bound (a setting, §557) the waiting act writes recovery-waits, cleared once it ends (§561). + fs := newHeldFindings() + r.e.Findings, r.e.Settings.DumpWait = fs, time.Millisecond ans := r.e.Accept(restoreAct("app/notes/2-fp1")) <-hd.waiting + if f := <-fs.written; f.Kind != recovery.FindingRecoveryWaits || f.Scope != "app" || f.ID != ans.Accepted { + t.Errorf("written %+v", f) + } kp, _ := r.e.KeptPoints("", "") if kp.Running == nil || kp.Running.WaitsOn == "" || kp.Running.Act != ans.Accepted { t.Errorf("running %+v", kp.Running) @@ -154,6 +161,9 @@ func TestAWaitingActIsCancelledAndABegunOneRefused(t *testing.T) { if ev.Outcome != "failed" || ev.Error != recovery.ErrCanceled || ev.Why != "not now" || ev.Act != ans.Accepted { t.Errorf("restored %+v", ev) } + if fs.has(recovery.FindingRecoveryWaits) { + t.Errorf("recovery-waits stands after the act ended: %+v", fs.Held()) + } } // retry asks again the last failed act with its fields under an act of its own, and answers `nothing to retry` when its @@ -237,6 +247,9 @@ func TestRestoreAsideRefusesLiveNamesAndDiscardAsideRemovesTheDirectory(t *testi // it back once no live name exists (§516, §554, §566, A-123, A-124). func TestAsideMovesTheLiveDataIntoAnAsideAndRestoreAsidePutsItBack(t *testing.T) { r := newRig(t) + // The findings of §561: aside-unrestored while the aside alone holds the data, points-filling past the fill mark. + fs := newHeldFindings() + r.e.Findings, r.e.Node, r.e.Settings.FillingMark, r.e.Settings.Cap = fs, "srv", 1, 4096 ans := r.do(recovery.Act{Kind: recovery.KindAside, Module: "app", Kept: "notes", Why: "look at it", Caller: "tools", Asker: "op"}) ev := r.s.named(seat.EventSetAside) if ans.Accepted == "" || len(ev) != 1 { @@ -253,11 +266,19 @@ func TestAsideMovesTheLiveDataIntoAnAsideAndRestoreAsidePutsItBack(t *testing.T) if kp, _ := r.e.KeptPoints("app", ""); len(kp.Unrestored) != 1 || kp.Unrestored[0].ID != sa.ID { t.Errorf("kept-points unrestored %+v", kp.Unrestored) } + if !fs.has(recovery.FindingAsideUnrestored) || !fs.has(recovery.FindingPointsFilling) { + t.Errorf("findings %+v", fs.Held()) + } r.s.events = nil r.do(recovery.Act{Kind: recovery.KindRestoreAside, Module: "app", Kept: "notes", AsideID: sa.ID, Why: "done looking", Caller: "tools"}) if lastRestored(t, r).Outcome != "done" || r.b.streams["app_notes"] != "hello" || len(r.e.Unrestored()) != 0 { t.Errorf("streams %v", r.b.streams) } + r.e.Settings.FillingMark = 100 + r.do(recovery.Act{Kind: recovery.KindDiscardAside, Module: "app", Kept: "notes", AsideID: sa.ID, Why: "not needed", Caller: "tools"}) + if fs.has(recovery.FindingAsideUnrestored) || fs.has(recovery.FindingPointsFilling) { + t.Errorf("findings stand after the aside was restored and discarded: %+v", fs.Held()) + } } // Said means acknowledged: a restore whose restore-started the bus did not take reads nothing and sets nothing aside, and