mesh-recovery: points-filling, recovery-waits and aside-unrestored through mesh-sdk's finding writer

The module measures its points directory against the fill mark of its cap at its start and
after each act, writes recovery-waits for an act that waits on the dump past its bound and
clears it once the act starts its work or ends, and writes aside-unrestored for each aside
only it holds the data of, cleared once the aside is restored or discarded. The bounds are
the settings points-filling-mark and dump-wait-minutes, read at the start.

Builds: design 47 G §561, §557, §564, §633, A-112, A-130

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

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) {

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 == "" {

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

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()
 }
 

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
+}

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
+// `<scope>.<id>.<kind>` 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
+}

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}
 }

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)
 			}

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