bus5 B4 (§681, A-29, A-40, §529): the bus module writes snapshots-filling through mesh-sdk's held writer over its state findings; nats-findings declared in the foundation form

Builds: design 47 G §681, §529, A-17, A-29, A-30, A-40
This commit is contained in:
2026-11-11 22:49:07 +01:00
parent d627400cf1
commit 790b5aac7d
6 changed files with 96 additions and 24 deletions
+17 -3
View File
@@ -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)),
+9 -5
View File
@@ -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, `<module>-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)
}
+3 -1
View File
@@ -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
+21 -6
View File
@@ -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")
+28 -7
View File
@@ -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)
}
}
}
+18 -2
View File
@@ -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,