From 790b5aac7ddc771bddb06acc271e8fd027ff915b Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:49:07 +0100 Subject: [PATCH] =?UTF-8?q?bus5=20B4=20(=C2=A7681,=20A-29,=20A-40,=20?= =?UTF-8?q?=C2=A7529):=20the=20bus=20module=20writes=20snapshots-filling?= =?UTF-8?q?=20through=20mesh-sdk's=20held=20writer=20over=20its=20state=20?= =?UTF-8?q?findings;=20nats-findings=20declared=20in=20the=20foundation=20?= =?UTF-8?q?form?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §681, §529, A-17, A-29, A-30, A-40 --- modules/nats/cmd/nats-tools/keep.go | 20 +++++++++-- modules/nats/cmd/nats-tools/manifest_test.go | 14 +++++--- modules/nats/internal/keeper/keeper.go | 4 ++- modules/nats/internal/keeper/keeper_test.go | 27 +++++++++++---- modules/nats/internal/keeper/snapshots.go | 35 ++++++++++++++++---- modules/nats/module.json | 20 +++++++++-- 6 files changed, 96 insertions(+), 24 deletions(-) diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index 1247d8e44..c36085ee1 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -19,6 +19,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" "git.novox.be/novox/mesh-contracts/health-and-repair/go/clients/meshhandactsclient" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" @@ -26,6 +27,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/seatdef" + "git.novox.be/novox/mesh-sdk/go/views" "nats-tools/internal/bussnap" "nats-tools/internal/keeper" @@ -251,6 +253,18 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer } defer nc.Close() recorder := meshhandactsclient.Recorder{Ask: conn.Call} + // The bus module's findings go through a held writer over its own state findings (design 47 G §681, A-29, A-40), made at once + // and started before the keeper's loop, so what the keeper writes first is held and written at the start. + writer, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: keeper.Module, Node: account.Node, State: conn.State("findings"), + Kinds: []string{keeper.FindingSnapshotsFilling}, JS: js, Conn: nc, Logf: logger.Printf}) + if err != nil { + return err + } + if err := writer.Start(ctx); err != nil { + return err + } + defer writer.Close() k := &keeper.Keeper{ Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR")), NC: nc, Dir: getenv("MESH_NATS_SNAPSHOTS_DIR"), Log: logger.Printf}, @@ -259,9 +273,9 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer _, err := recorder.Record(meshhandactsclient.Request{What: what, Why: why, Cause: cause}) return err }, - Warrant: WarrantFrom(conn.Ask), - Logf: logger.Printf, - Raise: percent(getenv, "MESH_NATS_FULL_RAISE", 75), Clear: percent(getenv, "MESH_NATS_FULL_CLEAR", 70), + Warrant: WarrantFrom(conn.Ask), Finding: writer, + Logf: logger.Printf, + Raise: percent(getenv, "MESH_NATS_FULL_RAISE", 75), Clear: percent(getenv, "MESH_NATS_FULL_CLEAR", 70), // The settings snapshots-cap (20 GiB) and snapshot-keep-days (90) of the manifest (design 47 G §681). SnapshotsCap: number(getenv, "MESH_NATS_SNAPSHOTS_CAP", keeper.DefaultSnapshotsCap), SnapshotKeepDays: int(number(getenv, "MESH_NATS_SNAPSHOT_KEEP_DAYS", keeper.DefaultSnapshotKeepDays)), diff --git a/modules/nats/cmd/nats-tools/manifest_test.go b/modules/nats/cmd/nats-tools/manifest_test.go index 2394a9eb8..c5a0d4a4c 100644 --- a/modules/nats/cmd/nats-tools/manifest_test.go +++ b/modules/nats/cmd/nats-tools/manifest_test.go @@ -192,8 +192,8 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { if len(nm.Claims) != 1 || nm.Claims[0].Name != "mesh-broker" || !slices.Equal(nm.Claims[0].Serves, ReadingVerbs) { t.Errorf("claims are %+v: mesh-broker, serving its reading verbs %v alone", nm.Claims, ReadingVerbs) } - if !slices.Equal(nm.Consumes, []string{"mesh-catalog.manifest-accepted"}) { - t.Errorf("consumes %v: the seat mesh-catalog's manifest-accepted alone (ADR 0595 §4)", nm.Consumes) + if !slices.Equal(nm.Consumes, []string{"mesh-catalog.manifest-accepted", "mesh-conditions.findings-asked"}) { + t.Errorf("consumes %v: the seat mesh-catalog's manifest-accepted (ADR 0595 §4) and the finding writer's findings-asked (design 47 G A-17)", nm.Consumes) } if AcceptedSubject != "mesh.seat.mesh-catalog.event.manifest-accepted" { t.Errorf("the keeper's durable consumer filters on %s, not on the seat mesh-catalog's event", AcceptedSubject) @@ -206,10 +206,14 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { if !slices.Contains(nm.Uses, "operator-channel") { t.Errorf("uses %v: the operator channel, for the warrant of recreate and drop", nm.Uses) } - if len(nm.Streams) != 1 || nm.Streams[0].Name != "broker-events" || !slices.Equal(nm.Streams[0].Events, Events) || + if len(nm.Streams) != 2 || nm.Streams[0].Name != "broker-events" || !slices.Equal(nm.Streams[0].Events, Events) || nm.Streams[0].MaxAgeDays != 0 || nm.Streams[0].Retention != "history" || nm.Streams[0].MaxMiB != 256 { t.Errorf("streams are %+v: broker-events carrying %v, keeping its history, 256 MiB", nm.Streams, Events) } + // A foundation part declares its finding stream in the old form, `-findings` (design 47 G §529, A-30). + if f := nm.Streams[1]; f.Name != "nats-findings" || !slices.Equal(f.Events, []string{"finding-written.*", "finding-cleared.*", "findings-stated.*"}) || f.MaxAgeDays != 7 || f.MaxMiB != 16 { + t.Errorf("the finding stream is %+v: nats-findings, its three events, 7 days and 16 MiB (design 47 G §529)", f) + } for _, e := range Events { if slices.Contains(m.Emits, e) { t.Errorf("emits lists %s, which its stream carries: an event is named in one place (ADR 0363)", e) @@ -224,7 +228,7 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { } // spent, the warrants the keeper spent, is written by the keeper alone and only read by the tools (review round 2, // security F2: the tool runner's account, which writes requests, may never delete a spend record). - if !slices.Equal(states, []string{"asked", "requests", "spent"}) || !slices.Equal(readOnly, []string{"spent"}) { + if !slices.Equal(states, []string{"asked", "requests", "spent", "findings"}) || !slices.Equal(readOnly, []string{"spent"}) { t.Errorf("state is %v, read-only %v: asked and requests, old sections (ADR 0595 §10), and spent, read-only", states, readOnly) } if _, ok := nm.OwnSecrets["broker-keeper"]; !ok { @@ -234,7 +238,7 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { bus, _ := keeper["bus"].(map[string]any) health, _ := keeper["health"].(map[string]any) if keeper["type"] != "process" || keeper["artifact"] != "tools" || fmt.Sprint(keeper["run"]) != "[./nats-tools keep]" || - fmt.Sprint(bus["state"]) != "[asked requests spent]" || + fmt.Sprint(bus["state"]) != "[asked requests spent findings]" || health["kind"] != "tool" || health["tool"] != "nats_keeper_health" { t.Errorf("the keeper is %v: a process of the tools bundle, run as keep, keeping asked, requests and spent, judged by nats_keeper_health", keeper) } diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index bb3747d27..4139151fd 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/modules/nats/internal/keeper/keeper.go @@ -236,7 +236,9 @@ type Keeper struct { Raise, Clear int64 // Files are the snapshot files (the disk when nil); SnapshotsCap the cap of their directory in bytes and // SnapshotKeepDays how long a snapshot is kept: the settings `snapshots-cap` (20 GiB) and `snapshot-keep-days` (90) (design 47 G §681). - Files Files + Files Files + // Finding is the writer of the finding snapshots-filling. + Finding Findings SnapshotsCap int64 SnapshotKeepDays int diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index a498b5f2d..ea76bf053 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -13,6 +13,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" "git.novox.be/novox/mesh-sdk/go/catalogue" @@ -277,10 +278,23 @@ func (s *said) reset() { s.events, s.bodies, s.conds = nil, nil, nil } +// findingLog is the finding writer's calls, as the held writer would be handed them. +type findingLog struct{ calls []string } + +func (f *findingLog) Write(_ context.Context, fi meshconditions.Finding) error { + f.calls = append(f.calls, "write "+fi.Scope+"."+fi.ID+"."+fi.Kind+" "+fi.Observer) + return nil +} + +func (f *findingLog) Clear(_ context.Context, scope, id, kind, _ string) error { + f.calls = append(f.calls, "clear "+scope+"."+id+"."+kind) + return nil +} + func newKeeper(view FirstView) (*Keeper, *fakeBus, *memStore, *said) { bus, store, out := newFakeBus(), newMemStore(), &said{} at := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) - k := &Keeper{Bus: bus, Store: store, Emit: out.emit, Say: out.say, Files: &memFiles{size: map[string]int64{}}, + k := &Keeper{Bus: bus, Store: store, Finding: &findingLog{}, Emit: out.emit, Say: out.say, Files: &memFiles{size: map[string]int64{}}, View: func() (FirstView, error) { return view, nil }, Now: func() time.Time { return at }} return k, bus, store, out @@ -722,8 +736,9 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { if got.Path != "/snapshots/KV_m_drafts.tar" || got.Size != 100 || !strings.HasPrefix(got.Digest, "sha256:") || !got.Protected || got.Error != "" { t.Fatalf("the drop's record: %+v", got) } - if !strings.Contains(fmt.Sprint(out.conds), "raised nats.snapshots-filling.nats_snapshots") { - t.Errorf("100 of 120 bytes is 83%%, so the finding is raised: %v", out.conds) + findings := k.Finding.(*findingLog) + if fmt.Sprint(findings.calls) != "[write nats.snapshots.snapshots-filling nats]" { + t.Errorf("100 of 120 bytes is 83%%, so the finding is written through the held writer: %v", findings.calls) } // Another snapshot of 100 bytes would pass the cap: refused, its file removed, nothing deleted from the bus. files.size["/snapshots/m_log.tar"] = 100 @@ -736,11 +751,11 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { _, _ = store.Put(StateRequests, "r0", old) base := k.now() k.Now = func() time.Time { return base.AddDate(0, 0, 91) } - out.reset() + findings.calls = nil k.ExpireSnapshots() if _, _, found, _ := store.Get(StateRequests, "r10"); found || fmt.Sprint(files.removed) != "[/snapshots/m_log.tar /snapshots/KV_m_drafts.tar]" || - len(journal) != 1 || !strings.Contains(journal[0], "past its 90 days") || fmt.Sprint(out.conds) != "[cleared nats.snapshots-filling.nats_snapshots]" { - t.Errorf("after 91 days: removed %v, journal %v, said %v", files.removed, journal, out.conds) + len(journal) != 1 || !strings.Contains(journal[0], "past its 90 days") || fmt.Sprint(findings.calls) != "[clear nats.snapshots.snapshots-filling]" { + t.Errorf("after 91 days: removed %v, journal %v, finding calls %v", files.removed, journal, findings.calls) } if _, _, found, _ := store.Get(StateRequests, "r0"); !found { t.Error("a snapshot of before G was removed by age") diff --git a/modules/nats/internal/keeper/snapshots.go b/modules/nats/internal/keeper/snapshots.go index 4a35cd4e9..6c1993284 100644 --- a/modules/nats/internal/keeper/snapshots.go +++ b/modules/nats/internal/keeper/snapshots.go @@ -11,7 +11,7 @@ import ( "time" "encoding/json" - occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" ) // The keeper's snapshots before a drop or recreate (design 47 G §681, ADR 0675): each is recorded in its request's record @@ -26,12 +26,19 @@ const ( // SnapshotsFillingMark is the fill of the cap, in percent, past which `snapshots-filling` is raised; it is cleared // back under it. SnapshotsFillingMark int64 = 80 - // conditionSnapshotsFilling is the finding's kind; its scope is the bus module and its id SnapshotsID. - conditionSnapshotsFilling = "snapshots-filling" + // FindingSnapshotsFilling is the finding's kind; its scope is the bus module and its id SnapshotsID. + FindingSnapshotsFilling = "snapshots-filling" // SnapshotsID is the id of the finding `snapshots-filling`. SnapshotsID = "snapshots" ) +// Findings is the bus module's finding writer, a held writer of mesh-sdk (views.HeldWriter): the keeper writes and clears +// `snapshots-filling` through it (design 47 G §681, A-29). +type Findings interface { + Write(ctx context.Context, finding meshconditions.Finding) error + Clear(ctx context.Context, scope, id, kind, why string) error +} + // Files is the keeper's snapshot files: read whole for their size and digest, and removed. type Files interface { // Digest reads the file's size and the `sha256:` digest of its bytes. @@ -214,12 +221,26 @@ func (k *Keeper) snapshotsFill(used int64) { k.filling, k.fillUnjudged = pct >= SnapshotsFillingMark, false now := k.filling k.mu.Unlock() + if k.Finding == nil { + return + } + ctx := context.Background() switch { case now: - k.raise(conditionSnapshotsFilling, Module, SnapshotsID, occ.SeverityWarning, "The bus module's snapshots are filling up", - "snapshots removed, or a larger snapshots-cap", - fmt.Sprintf("the snapshots hold %d of %d bytes (%d%%); at the cap a drop or recreate is refused; cleared under %d%%", used, capBytes, pct, SnapshotsFillingMark)) + f := meshconditions.Finding{Scope: Module, ID: SnapshotsID, Kind: FindingSnapshotsFilling, + Subject: meshconditions.Subject{Scope: "module", ID: Module}, + Severity: meshconditions.FindingSeverityWarning, Headline: "The bus module's snapshots are filling up", + Summary: fmt.Sprintf("the snapshots hold %d of %d bytes (%d%%)", used, capBytes, pct), + Needs: "snapshots removed, or a larger snapshots-cap", + Explanation: fmt.Sprintf("the snapshots hold %d of %d bytes (%d%%); at the cap a drop or recreate is refused; cleared under %d%%", + used, capBytes, pct, SnapshotsFillingMark), + Source: "snapshots", Observer: Module, At: k.now()} + if err := k.Finding.Write(ctx, f); err != nil { + k.logf("the finding %s cannot be written: %v", FindingSnapshotsFilling, err) + } case was || unjudged: - k.clear(conditionSnapshotsFilling, Module, SnapshotsID) + if err := k.Finding.Clear(ctx, Module, SnapshotsID, FindingSnapshotsFilling, "the snapshots are under 80 % of their cap"); err != nil { + k.logf("the finding %s cannot be cleared: %v", FindingSnapshotsFilling, err) + } } } diff --git a/modules/nats/module.json b/modules/nats/module.json index 8021aa955..cbaf95238 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -60,7 +60,8 @@ "mesh-delivery-indexer.indexer_bus_state" ], "consumes": [ - "mesh-catalog.manifest-accepted" + "mesh-catalog.manifest-accepted", + "mesh-conditions.findings-asked" ], "uses": [ "operator-channel" @@ -81,6 +82,16 @@ ], "retention": "history", "max-mib": 256 + }, + { + "name": "nats-findings", + "events": [ + "finding-written.*", + "finding-cleared.*", + "findings-stated.*" + ], + "max-age-days": 7, + "max-mib": 16 } ], "state": [ @@ -98,6 +109,10 @@ "history": 1, "ttl-seconds": 86400, "read-only": true + }, + { + "name": "findings", + "history": 1 } ], "listens": [ @@ -259,7 +274,8 @@ "state": [ "asked", "requests", - "spent" + "spent", + "findings" ], "asks": true, "keeps": true,