bus5 B1 (§681): the keeper records its snapshot (path, size, digest, protected) before it touches live data, holds it to snapshots-cap, removes it after snapshot-keep-days, tells snapshots-filling

Builds: design 47 G §681, ADR 0675

modules/nats/cmd/nats-tools/keep.go

@@ -217,6 +217,14 @@ func percent(getenv func(string) string, name string, fallback int64) int64 {
 	return fallback
 }
 
+// number reads a positive whole number from the environment, or the fallback.
+func number(getenv func(string) string, name string, fallback int64) int64 {
+	if n, err := strconv.ParseInt(strings.TrimSpace(getenv(name)), 10, 64); err == nil && n > 0 {
+		return n
+	}
+	return fallback
+}
+
 // runKeeper is the keeper's process: it returns when ctx ends, or when it has no credential.
 func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
 	logger := log.New(logOut, "nats-keeper: ", log.LstdFlags)
@@ -245,6 +253,9 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer
 		Warrant: WarrantFrom(conn.Ask),
 		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)),
 	}
 	logger.Printf("started on the account %s: asking %s.%s for the first view", account.User, keeper.FirstViewOwner, keeper.FirstViewVerb)
 	return Keep(ctx, k, hearer(js, account, logger.Printf), requestsOf(conn), logger.Printf)
@@ -291,6 +302,7 @@ func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func
 	tick := time.NewTicker(FillEvery)
 	defer tick.Stop()
 	k.FillPass()
+	k.SnapshotsAtStart()
 	var wait time.Duration
 	for {
 		next, err := k.AskFirstView(wait)

modules/nats/internal/keeper/jetstream.go

@@ -157,3 +157,24 @@ func retryable(err error) error {
 	}
 	return err
 }
+
+// RestrictedMetadata is the key a restricted stream's metadata says it under (ADR 0363 §4): the grants part writes it.
+const RestrictedMetadata = "mesh-restricted"
+
+// Restricted reads the streams' metadata: true when any says it is restricted.
+func (b JetStream) Restricted(ctx context.Context, streams []string) (bool, error) {
+	for _, name := range streams {
+		s, err := b.JS.Stream(ctx, name)
+		if err != nil {
+			return false, retryable(err)
+		}
+		info, err := s.Info(ctx)
+		if err != nil {
+			return false, retryable(err)
+		}
+		if _, ok := info.Config.Metadata[RestrictedMetadata]; ok {
+			return true, nil
+		}
+	}
+	return false, nil
+}

modules/nats/internal/keeper/keeper.go

@@ -180,6 +180,9 @@ type Bus interface {
 	Delete(ctx context.Context, o Object) error
 	// Snapshot takes a snapshot of the streams and says where it is kept.
 	Snapshot(ctx context.Context, streams []string) (string, error)
+	// Restricted says whether any of the streams is one ADR 0436 §3 lets no other account hear (its metadata says it), so
+	// a snapshot of them is never handed out (design 47 G §681).
+	Restricted(ctx context.Context, streams []string) (bool, error)
 }
 
 // Fill is how much one object holds: its bytes and messages.
@@ -221,12 +224,19 @@ type Keeper struct {
 	Logf    func(string, ...any)
 	// Raise and Clear are the fill marks in percent (ADR 0595 §6): settings of the bus module, 75 and 70 by default.
 	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
+	SnapshotsCap     int64
+	SnapshotKeepDays int
 
 	mu     sync.Mutex
 	status Status
 	viewed bool
 	// unwritten says the condition that the status cannot be written, with words unsaid, is raised.
 	unwritten bool
+	// filling is whether the snapshots directory was last judged past 80 % of its cap, fillJudged whether it is to be judged afresh (the keeper's start).
+	filling, fillUnjudged bool
 }
 
 func (k *Keeper) now() time.Time {

modules/nats/internal/keeper/keeper_test.go

@@ -2,6 +2,7 @@ package keeper
 
 import (
 	"context"
+	"encoding/hex"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -27,6 +28,8 @@ type fakeBus struct {
 	updated []string
 	deleted []string
 	snaps   [][]string
+	// restricted names the streams whose metadata says they are restricted.
+	restricted map[string]bool
 }
 
 func newFakeBus() *fakeBus {
@@ -77,6 +80,36 @@ func (b *fakeBus) Snapshot(_ context.Context, streams []string) (string, error)
 	return "/snapshots/" + strings.Join(streams, "+") + ".tar", nil
 }
 
+func (b *fakeBus) Restricted(_ context.Context, streams []string) (bool, error) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	for _, s := range streams {
+		if b.restricted[s] {
+			return true, nil
+		}
+	}
+	return false, nil
+}
+
+// memFiles are snapshot files in memory: each of size, 100 bytes unless a test says so, with a digest of its path.
+type memFiles struct {
+	size    map[string]int64
+	removed []string
+}
+
+func (f *memFiles) Digest(path string) (int64, string, error) {
+	n, ok := f.size[path]
+	if !ok {
+		n = 100
+	}
+	return n, "sha256:" + hex.EncodeToString([]byte(path)), nil
+}
+
+func (f *memFiles) Remove(path string) error {
+	f.removed = append(f.removed, path)
+	return nil
+}
+
 // memStore is a key-value state in memory.
 type memStore struct {
 	// refuse refuses a Put of `<state>/<key>`.
@@ -193,7 +226,7 @@ func (s *said) reset() {
 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,
+	k := &Keeper{Bus: bus, Store: store, 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
@@ -604,6 +637,61 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) {
 				t.Error("the dropped entry is still held")
 			}
 		}},
+		// Design 47 G §681: the snapshot is recorded in its request (path, size, digest, protected) before the live data is
+		// touched; one past the cap is refused before anything is removed; the keeper alone removes one past its days, said
+		// in its journal; the finding snapshots-filling is raised past 80 % of the cap and cleared under it; a snapshot of
+		// before G (no path, no digest) is never removed by age.
+		{"a snapshot is recorded, held to the cap, removed after its days, and told as a finding while it fills", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
+			store, files := k.Store.(*memStore), k.Files.(*memFiles)
+			var journal []string
+			k.Logf = func(f string, a ...any) { journal = append(journal, fmt.Sprintf(f, a...)) }
+			k.SnapshotsCap = 120
+			k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{drafts, log}})
+			k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{log}})
+			bus.restricted = map[string]bool{"KV_m_drafts": true}
+			binds, _ := occ.ActDigest(occ.Act{"verb": "drop", "module": "m", "local": "drafts"})
+			ask := oc.Ask{ID: "bus-9", Who: occ.Operator, Expires: occ.FormatTime(k.now().Add(time.Hour)), Options: []oc.Option{{ID: "go", Label: "Remove", Level: occ.Destroy, Binds: binds}}}
+			w := &oc.Warrant{Ask: "bus-9", Asker: Module, Outcome: occ.OutcomeChosen, By: &oc.Person{Who: occ.Operator},
+				Option: "go", Label: "Remove", Level: occ.Destroy, At: occ.FormatTime(k.now()), AskDigest: occ.Digest(ask)}
+			k.Warrant = func(string) (*oc.Ask, *oc.Warrant, error) { return &ask, w, nil }
+			get := func(id string) Request {
+				raw, _, _, _ := store.Get(StateRequests, id)
+				var got Request
+				_ = json.Unmarshal(raw, &got)
+				return got
+			}
+			raw, _ := json.Marshal(Request{ID: "r10", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-9", Caller: "person.jo", At: k.now()})
+			_, _ = store.Put(StateRequests, "r10", raw)
+			out.reset()
+			k.ActOnRequests()
+			got := get("r10")
+			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)
+			}
+			// 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
+			err := k.recordSnapshot(context.Background(), &Request{ID: "r11", At: k.now()}, "/snapshots/m_log.tar", []string{"KV_m_log"})
+			if err == nil || !strings.Contains(err.Error(), "would pass the cap 120") || fmt.Sprint(files.removed) != "[/snapshots/m_log.tar]" {
+				t.Errorf("a snapshot past the cap: %v, removed %v", err, files.removed)
+			}
+			// A snapshot of before G stays; one past 90 days goes with its record, said, and the finding clears.
+			old, _ := json.Marshal(Request{ID: "r0", Verb: VerbDrop, Snapshot: "/before/g.tar", At: k.now().AddDate(0, 0, -400), Done: k.now()})
+			_, _ = store.Put(StateRequests, "r0", old)
+			base := k.now()
+			k.Now = func() time.Time { return base.AddDate(0, 0, 91) }
+			out.reset()
+			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)
+			}
+			if _, _, found, _ := store.Get(StateRequests, "r0"); !found {
+				t.Error("a snapshot of before G was removed by age")
+			}
+		}},
 		{"a recreate rebuilds a drifted entry after a snapshot", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) {
 			store := k.Store.(*memStore)
 			k.Heard(Accepted{Module: "m", Revision: 1, BusState: []catalogue.BusStateEntry{log}})

modules/nats/internal/keeper/requests.go

@@ -44,6 +44,13 @@ type Request struct {
 	Outcome  string    `json:"outcome,omitempty"`
 	Error    string    `json:"error,omitempty"`
 	Snapshot string    `json:"snapshot,omitempty"`
+	// Path, Size, Digest and Protected are the snapshot the keeper took before a drop or recreate, written before it touches
+	// the live data (design 47 G §681): where it is, its bytes, the `sha256:` of them, and whether it holds a stream no other
+	// account may hear, which is never handed out.
+	Path      string `json:"path,omitempty"`
+	Size      int64  `json:"size,omitempty"`
+	Digest    string `json:"digest,omitempty"`
+	Protected bool   `json:"protected,omitempty"`
 	// Waiting is why it waits, while it waits: the operator's warrant, or the keeper's first view.
 	Waiting string `json:"waiting,omitempty"`
 	// Unrecorded is why the hand act of what it did could not be recorded, when it could not: said, never only logged.
@@ -85,6 +92,7 @@ var ErrNoWarrant = errors.New("the operator has not answered yet")
 
 // ActOnRequests acts on every request not ended yet, oldest first, and writes each one's end back to its key.
 func (k *Keeper) ActOnRequests() {
+	k.ExpireSnapshots()
 	keys, err := k.Store.Keys(StateRequests)
 	if err != nil {
 		k.logf("the persons' requests cannot be read: %v", err)
@@ -192,6 +200,10 @@ func (k *Keeper) act(r *Request) (string, error) {
 		return "nothing done", fmt.Errorf("the snapshot before %s failed, so nothing was removed: %w", r.Verb, err)
 	}
 	r.Snapshot = path
+	if err := k.recordSnapshot(ctx, r, path, streams); err != nil {
+		r.Snapshot = ""
+		return "nothing done", err
+	}
 	// The warrant is spent before anything is deleted: of two requests naming one ask, only the one whose claim stands acts.
 	if err := k.spend(r); err != nil {
 		return "nothing done", err

modules/nats/internal/keeper/snapshots.go

@@ -0,0 +1,225 @@
+package keeper
+
+import (
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"io"
+	"os"
+	"strings"
+	"time"
+
+	"encoding/json"
+	occ "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient"
+)
+
+// The keeper's snapshots before a drop or recreate (design 47 G §681, ADR 0675): each is recorded in its request's record
+// in the state requests (path, size, digest, protected) before the live data is touched, counts against the directory's
+// cap, is removed by the keeper after its days, and is told as a finding when the directory is filling.
+
+const (
+	// DefaultSnapshotsCap is the cap of the snapshots directory unless the setting `snapshots-cap` says otherwise: 20 GiB.
+	DefaultSnapshotsCap int64 = 20 << 30
+	// DefaultSnapshotKeepDays is how long a snapshot and its record are kept after their request (ADR 0667 §12).
+	DefaultSnapshotKeepDays = 90
+	// 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"
+	// SnapshotsID is the id of the finding `snapshots-filling`.
+	SnapshotsID = "snapshots"
+)
+
+// 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.
+	Digest(path string) (size int64, digest string, err error)
+	Remove(path string) error
+}
+
+// DiskFiles is Files on the keeper's own disk.
+type DiskFiles struct{}
+
+// Digest reads the file.
+func (DiskFiles) Digest(path string) (int64, string, error) {
+	f, err := os.Open(path)
+	if err != nil {
+		return 0, "", err
+	}
+	defer f.Close()
+	h := sha256.New()
+	n, err := io.Copy(h, f)
+	if err != nil {
+		return 0, "", err
+	}
+	return n, "sha256:" + hex.EncodeToString(h.Sum(nil)), nil
+}
+
+// Remove removes the file; one already gone is no error.
+func (DiskFiles) Remove(path string) error {
+	if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
+		return err
+	}
+	return nil
+}
+
+func (k *Keeper) files() Files {
+	if k.Files == nil {
+		return DiskFiles{}
+	}
+	return k.Files
+}
+
+func (k *Keeper) snapshotsCap() int64 {
+	if k.SnapshotsCap <= 0 {
+		return DefaultSnapshotsCap
+	}
+	return k.SnapshotsCap
+}
+
+func (k *Keeper) keepDays() int {
+	if k.SnapshotKeepDays <= 0 {
+		return DefaultSnapshotKeepDays
+	}
+	return k.SnapshotKeepDays
+}
+
+// Recorded says whether the request's snapshot was written whole by a keeper of milestone G: it holds a path and a
+// digest. A snapshot of before G holds neither, and the keeper never sends nor deletes it by age (§681).
+func (r Request) Recorded() bool { return r.Path != "" && r.Digest != "" }
+
+// Expired says whether the recorded snapshot is past its days at now.
+func (r Request) Expired(now time.Time, days int) bool {
+	return r.Recorded() && !r.At.IsZero() && now.After(r.At.AddDate(0, 0, days))
+}
+
+// snapshotRecords is every request in the state requests holding a recorded snapshot, by key.
+func (k *Keeper) snapshotRecords() (map[string]Request, error) {
+	keys, err := k.Store.Keys(StateRequests)
+	if err != nil {
+		return nil, err
+	}
+	out := map[string]Request{}
+	for _, key := range keys {
+		if strings.HasPrefix(key, ActedPrefix) {
+			continue
+		}
+		raw, _, found, err := k.Store.Get(StateRequests, key)
+		if err != nil || !found {
+			continue
+		}
+		var r Request
+		if json.Unmarshal(raw, &r) == nil && r.Recorded() {
+			out[key] = r
+		}
+	}
+	return out, nil
+}
+
+// SnapshotsUsed is the bytes the recorded snapshots hold, but for the request skip.
+func (k *Keeper) SnapshotsUsed(skip string) (int64, error) {
+	recs, err := k.snapshotRecords()
+	if err != nil {
+		return 0, err
+	}
+	var used int64
+	for key, r := range recs {
+		if key != skip {
+			used += r.Size
+		}
+	}
+	return used, nil
+}
+
+// recordSnapshot reads the snapshot's size and digest, refuses one that would pass the cap (removing the file it wrote,
+// so nothing is left over), and writes path, size, digest and protected into the request's record before anything is
+// removed from the live data.
+func (k *Keeper) recordSnapshot(ctx context.Context, r *Request, path string, streams []string) error {
+	size, digest, err := k.files().Digest(path)
+	if err != nil {
+		return fmt.Errorf("the snapshot at %s cannot be read whole, so nothing was removed: %w", path, err)
+	}
+	protected, err := k.Bus.Restricted(ctx, streams)
+	if err != nil {
+		_ = k.files().Remove(path)
+		return fmt.Errorf("whether the snapshot holds a stream no other account hears cannot be read, so nothing was removed: %w", err)
+	}
+	used, err := k.SnapshotsUsed(r.ID)
+	if err != nil {
+		_ = k.files().Remove(path)
+		return fmt.Errorf("the snapshots' fill cannot be read, so nothing was removed: %w", err)
+	}
+	if used+size > k.snapshotsCap() {
+		_ = k.files().Remove(path)
+		return fmt.Errorf("the snapshot (%d bytes) would pass the cap %d of the snapshots directory (%d held), so nothing was removed: raise snapshots-cap or remove a snapshot", size, k.snapshotsCap(), used)
+	}
+	r.Path, r.Size, r.Digest, r.Protected = path, size, digest, protected
+	data, _ := json.Marshal(r)
+	if _, err := k.Store.Put(StateRequests, r.ID, data); err != nil {
+		_ = k.files().Remove(path)
+		r.Path, r.Size, r.Digest, r.Protected = "", 0, "", false
+		return fmt.Errorf("the snapshot cannot be recorded, so nothing was removed: %w", err)
+	}
+	k.snapshotsFill(used + size)
+	return nil
+}
+
+// ExpireSnapshots removes every recorded snapshot past its days, and its record, each removal said in the journal:
+// at the keeper's start and at each new request, never on a timer. It judges the fill after.
+func (k *Keeper) ExpireSnapshots() {
+	recs, err := k.snapshotRecords()
+	if err != nil {
+		k.logf("the snapshots cannot be read: %v", err)
+		return
+	}
+	var used int64
+	for key, r := range recs {
+		if !r.Expired(k.now(), k.keepDays()) {
+			used += r.Size
+			continue
+		}
+		if err := k.files().Remove(r.Path); err != nil {
+			k.logf("the snapshot %s of %s, past its %d days, cannot be removed: %v", r.Path, key, k.keepDays(), err)
+			used += r.Size
+			continue
+		}
+		if err := k.Store.Delete(StateRequests, key); err != nil {
+			k.logf("the record of the removed snapshot %s cannot be removed: %v", key, err)
+			used += r.Size
+			continue
+		}
+		k.logf("removed the snapshot %s of the request %s, taken %s, past its %d days", r.Path, key, r.At.Format(time.RFC3339), k.keepDays())
+	}
+	k.snapshotsFill(used)
+}
+
+// SnapshotsAtStart is the keeper's first look at the snapshots at its start: what is past its days removed, and the fill
+// judged afresh, so a finding raised before a restart that no longer holds is cleared.
+func (k *Keeper) SnapshotsAtStart() {
+	k.mu.Lock()
+	k.fillUnjudged = true
+	k.mu.Unlock()
+	k.ExpireSnapshots()
+}
+
+// snapshotsFill raises the finding `snapshots-filling` past 80 % of the cap and clears it once back under, judged at
+// every snapshot taken and removed and at the keeper's start.
+func (k *Keeper) snapshotsFill(used int64) {
+	capBytes := k.snapshotsCap()
+	pct := used * 100 / capBytes
+	k.mu.Lock()
+	was, unjudged := k.filling, k.fillUnjudged
+	k.filling, k.fillUnjudged = pct >= SnapshotsFillingMark, false
+	now := k.filling
+	k.mu.Unlock()
+	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))
+	case was || unjudged:
+		k.clear(conditionSnapshotsFilling, Module, SnapshotsID)
+	}
+}

modules/nats/module.json

@@ -251,6 +251,8 @@
         "MESH_BROKER_FILE": "${dir:keeper-state}/broker-keeper",
         "MESH_NATS_FULL_RAISE": "75",
         "MESH_NATS_FULL_CLEAR": "70",
+        "MESH_NATS_SNAPSHOTS_CAP": "21474836480",
+        "MESH_NATS_SNAPSHOT_KEEP_DAYS": "90",
         "MESH_NATS_SNAPSHOTS_DIR": "${dir:keeper-state}"
       },
       "bus": {