diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index 6576c1284..a073a95d5 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/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) diff --git a/modules/nats/internal/keeper/jetstream.go b/modules/nats/internal/keeper/jetstream.go index 31e4ab8d8..fbb94b3f6 100644 --- a/modules/nats/internal/keeper/jetstream.go +++ b/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 +} diff --git a/modules/nats/internal/keeper/keeper.go b/modules/nats/internal/keeper/keeper.go index 073671dd4..7e9ba93d9 100644 --- a/modules/nats/internal/keeper/keeper.go +++ b/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 { diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index ef9af998f..99943317c 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/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 `/`. @@ -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}}) diff --git a/modules/nats/internal/keeper/requests.go b/modules/nats/internal/keeper/requests.go index d956a239e..692b98b20 100644 --- a/modules/nats/internal/keeper/requests.go +++ b/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 diff --git a/modules/nats/internal/keeper/snapshots.go b/modules/nats/internal/keeper/snapshots.go new file mode 100644 index 000000000..4a35cd4e9 --- /dev/null +++ b/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) + } +} diff --git a/modules/nats/module.json b/modules/nats/module.json index ad9d08019..b9bba9c08 100644 --- a/modules/nats/module.json +++ b/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": {