From 47aaee122c607fbdffa6a9bd6b6076f0428d53d5 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 22:54:35 +0100 Subject: [PATCH] =?UTF-8?q?bus5=20B3=20(=C2=A7681,=20A-44,=20A-45,=20A-59)?= =?UTF-8?q?:=20the=20bus=20module=20serves=20snapshot-get,=20snapshot-rest?= =?UTF-8?q?ore,=20snapshot-restore-cancel=20and=20snapshot-remove=20throug?= =?UTF-8?q?h=20its=20tools=20bundle,=20sends=20the=20chunks,=20hands=20the?= =?UTF-8?q?=20work=20to=20the=20keeper=20through=20requests;=20nats=5Fsnap?= =?UTF-8?q?shots;=20the=20whole=20test=20covers=20send,=20restore=20in=20p?= =?UTF-8?q?lace=20and=20removal?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §681, ADR 0675, lead answers A-44, A-45, A-59 --- .../cmd/nats-tools/keeper_integration_test.go | 144 +++++++- modules/nats/cmd/nats-tools/main.go | 8 +- modules/nats/cmd/nats-tools/manifest_test.go | 4 +- modules/nats/cmd/nats-tools/snapshots.go | 328 ++++++++++++++++++ modules/nats/cmd/nats-tools/snapshots_test.go | 219 ++++++++++++ modules/nats/internal/bussnap/snapshot.go | 2 +- modules/nats/internal/keeper/keeper_test.go | 56 +++ modules/nats/internal/keeper/requests.go | 14 +- modules/nats/internal/keeper/restore.go | 1 + modules/nats/internal/keeper/snapshots.go | 52 ++- modules/nats/module.json | 13 +- 11 files changed, 824 insertions(+), 17 deletions(-) create mode 100644 modules/nats/cmd/nats-tools/snapshots.go create mode 100644 modules/nats/cmd/nats-tools/snapshots_test.go diff --git a/modules/nats/cmd/nats-tools/keeper_integration_test.go b/modules/nats/cmd/nats-tools/keeper_integration_test.go index 32456345f..90379740c 100644 --- a/modules/nats/cmd/nats-tools/keeper_integration_test.go +++ b/modules/nats/cmd/nats-tools/keeper_integration_test.go @@ -4,9 +4,12 @@ package main import ( "context" + "crypto/sha256" + "encoding/hex" "encoding/json" "errors" "fmt" + "io" "os" "slices" "strings" @@ -18,10 +21,14 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + "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" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/views" "nats-tools/internal/bussnap" "nats-tools/internal/keeper" @@ -153,7 +160,7 @@ func TestTheBusModuleRunsWhole(t *testing.T) { t.Fatal(err) } defer pconn.Close() - for _, st := range []string{keeper.StateAsked, keeper.StateRequests, keeper.StateSpent} { + for _, st := range []string{keeper.StateAsked, keeper.StateRequests, keeper.StateSpent, "findings"} { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "nats_" + st, History: 5}); err != nil { t.Fatal(err) } @@ -163,9 +170,20 @@ func TestTheBusModuleRunsWhole(t *testing.T) { var mu sync.Mutex view := keeper.FirstView{Revision: 1} var acts []string + // The bus module's findings go through a held writer over its state findings, started as the keeper's process starts it. + writer, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ + Asked: meshconditions.SubjectFindingsAsked, Module: keeper.Module, Node: account.Node, State: pconn.State("findings"), + Kinds: []string{keeper.FindingSnapshotsFilling}, JS: js, Conn: nc, Logf: t.Logf}) + if err != nil { + t.Fatal(err) + } + if err := writer.Start(ctx); err != nil { + t.Fatal(err) + } + defer writer.Close() newKeeper := func() *keeper.Keeper { - return &keeper.Keeper{Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, snapshots)}, Store: store, - Emit: streamEmitter(js, account), Say: sayer(js), + return &keeper.Keeper{Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, snapshots), NC: nc, Dir: snapshots, Log: t.Logf}, Store: store, + Finding: writer, Emit: streamEmitter(js, account), Say: sayer(js), View: func() (keeper.FirstView, error) { mu.Lock(); defer mu.Unlock(); return view, nil }, Act: func(what, why, cause string) error { mu.Lock() @@ -421,6 +439,126 @@ func TestTheBusModuleRunsWhole(t *testing.T) { t.Error("the spent warrant deleted KV_m_drafts a second time") } }) + // Design 47 G §681: the snapshot the keeper wrote is read by the bundle's snapshot-get, sent in chunks that hold its bytes, + // restored in place by the keeper (refused while its name is live, ended by snapshot-restored) and removed by snapshot-remove. + t.Run("a snapshot is sent by the bundle, restored in place by the keeper and removed", func(t *testing.T) { + get := func(id string) keeper.Request { + raw, _, _, _ := store.Get(keeper.StateRequests, id) + var r keeper.Request + _ = json.Unmarshal(raw, &r) + return r + } + rec := get("r2") + if rec.Path == "" || rec.Digest == "" || rec.Protected { + t.Fatalf("the drop's record holds no snapshot: %+v", rec) + } + rs := requestStore{ + Put: func(key string, r keeper.Request) error { + raw, _ := json.Marshal(r) + _, err := store.Put(keeper.StateRequests, key, raw) + return err + }, + Get: func(key string) (*keeper.Request, error) { + if _, _, found, _ := store.Get(keeper.StateRequests, key); !found { + return nil, nil + } + r := get(key) + return &r, nil + }, + Keys: func() ([]string, error) { return store.Keys(keeper.StateRequests) }, + Now: func() time.Time { return time.Now().UTC() }, + Ended: func(id string, within time.Duration) *keeper.Request { + for deadline := time.Now().Add(within); time.Now().Before(deadline); time.Sleep(20 * time.Millisecond) { + if r := get(id); !r.Done.IsZero() { + return &r + } + } + return nil + }, + } + var chunkMu sync.Mutex + var chunks []Chunk + var ack func(string, json.RawMessage) (any, error) + world := &snapshotWorld{ + Open: func(path string) (io.ReadCloser, error) { return os.Open(path) }, + Granted: func(string, string) bool { return true }, Logf: t.Logf, Within: 5 * time.Second, KeepDays: 90, + Answer: func(_ string, h func(string, json.RawMessage) (any, error)) error { ack = h; return nil }, + } + world.Publish = func(subject string, body any, id string) error { + c := body.(Chunk) + chunkMu.Lock() + chunks = append(chunks, c) + chunkMu.Unlock() + go func() { _, _ = ack(strings.Replace(subject, ".chunk.", ".ack.", 1), nil) }() + return nil + } + tl := map[string]stdio.Tool{} + for _, x := range snapshotTools(rs, world) { + tl[x.Name] = x + } + call := "0123456789abcdef0123456789abcdef" + out, err := tl["mesh-broker.snapshot-get"].Run(map[string]any{"request": "r2", "call": call}) + if err != nil { + t.Fatal(err) + } + if a := out.(meshbroker.SnapshotGetAnswer); a.Error != "" || a.Digest != rec.Digest || a.Chunks < 1 || a.Stream != "m_drafts" { + t.Fatalf("snapshot-get answered %+v for the keeper's record %+v", a, rec) + } + eventually(t, "the chunks sent", func() bool { chunkMu.Lock(); defer chunkMu.Unlock(); return len(chunks) > 0 }) + var whole []byte + eventually(t, "the last chunk", func() bool { + chunkMu.Lock() + defer chunkMu.Unlock() + return int64(len(chunks)) == chunks[0].Of + }) + chunkMu.Lock() + for _, c := range chunks { + whole = append(whole, c.Data...) + } + chunkMu.Unlock() + if sum := sha256.Sum256(whole); "sha256:"+hex.EncodeToString(sum[:]) != rec.Digest || int64(len(whole)) != rec.Size { + t.Errorf("the chunks hold %d bytes, not the snapshot the keeper wrote (%d)", len(whole), rec.Size) + } + restore := func(why string) meshbroker.SnapshotRestoreAnswer { + o, err := tl["mesh-broker.snapshot-restore"].Run(map[string]any{"request": "r2", "why": why}) + if err != nil { + t.Fatal(err) + } + return o.(meshbroker.SnapshotRestoreAnswer) + } + // KV_m_drafts is live (made again after the drop): nothing is restored over it. + if a := restore("live"); !strings.Contains(a.Error, "names live") { + t.Errorf("a restore over a live stream answered %+v", a) + } + if err := js.DeleteStream(ctx, "KV_m_drafts"); err != nil { + t.Fatal(err) + } + a := restore("it was removed by mistake") + if a.Accepted == "" { + t.Fatalf("restore answered %+v", a) + } + eventually(t, "snapshot-restored on the seat's subject", func() bool { + for _, l := range saidOn("EVENTS", "mesh.seat.mesh-broker.event.snapshot-restored.") { + if strings.HasPrefix(l, "r2."+a.Accepted+" ") { + return true + } + } + return false + }) + if _, ok := held("KV_m_drafts"); !ok || get(a.Accepted).Result != "done" { + t.Errorf("the stream is not restored in place: held %v, the act %+v", ok, get(a.Accepted)) + } + o, err := tl["mesh-broker.snapshot-remove"].Run(map[string]any{"request": "r2", "why": "no longer needed"}) + if err != nil || o.(meshbroker.SnapshotRemoveAnswer).Accepted == "" { + t.Fatalf("snapshot-remove answered %+v, %v", o, err) + } + if _, err := os.Stat(rec.Path); !os.IsNotExist(err) { + t.Errorf("the snapshot file %s is still there: %v", rec.Path, err) + } + if _, _, found, _ := store.Get(keeper.StateRequests, "r2"); found { + t.Error("the snapshot's record outlived its file") + } + }) t.Run("nats_first_view_again asks at once and records the hand act", func(t *testing.T) { raw, _ := json.Marshal(keeper.Request{ID: "r3", Verb: keeper.VerbFirstViewAgain, Why: "a verdict thought missed", Caller: "person.jo"}) _, _ = store.Put(keeper.StateRequests, "r3", raw) diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 034c5be3b..7b5a9b40d 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -168,6 +168,10 @@ func main() { // them. var ReadingVerbs = []string{"list", "show", "check", "orphans"} +// SnapshotVerbs are the verbs this bundle serves for the keeper's snapshots (design 47 G §681, A-45): the bus module answers them +// and hands the work that needs the bus and the files to the keeper through the state requests (snapshots.go). +var SnapshotVerbs = []string{"snapshot-get", "snapshot-restore", "snapshot-restore-cancel", "snapshot-remove"} + // Events are the seat mesh-broker's events (ADR 0595 §6, ADR 0620 §2): said on the seat's subjects on the shared events stream // and kept in full on the bus module's own stream broker-events (BROKER_EVENTS), which `history` reads (keep.go). var Events = []string{"created", "refused", "drifted", "drift-cleared", "full", "full-cleared", "orphaned", "recreated", "dropped"} @@ -190,5 +194,7 @@ func allTools() []stdio.Tool { out = append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...) out = append(out, keeperHealthTool(StatusInState, HeldInState, jsz)) out = append(out, readTools(jsz, LiveMsgGet(LiveBus()))...) - return append(out, requestTools(RequestsInState())...) + rs := RequestsInState() + out = append(out, snapshotTools(rs, liveSnapshots())...) + return append(out, requestTools(rs)...) } diff --git a/modules/nats/cmd/nats-tools/manifest_test.go b/modules/nats/cmd/nats-tools/manifest_test.go index c5a0d4a4c..5065ebf0e 100644 --- a/modules/nats/cmd/nats-tools/manifest_test.go +++ b/modules/nats/cmd/nats-tools/manifest_test.go @@ -189,8 +189,8 @@ func TestTheBussManifestMountsItsProgramsIntoTheUpstreamImage(t *testing.T) { if err := json.Unmarshal(raw, &nm); err != nil { t.Fatal(err) } - 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 len(nm.Claims) != 1 || nm.Claims[0].Name != "mesh-broker" || !slices.Equal(nm.Claims[0].Serves, append(slices.Clone(ReadingVerbs), SnapshotVerbs...)) { + t.Errorf("claims are %+v: mesh-broker, serving its reading verbs %v and its snapshot verbs %v alone", nm.Claims, ReadingVerbs, SnapshotVerbs) } 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) diff --git a/modules/nats/cmd/nats-tools/snapshots.go b/modules/nats/cmd/nats-tools/snapshots.go new file mode 100644 index 000000000..3d095b2a1 --- /dev/null +++ b/modules/nats/cmd/nats-tools/snapshots.go @@ -0,0 +1,328 @@ +package main + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "regexp" + "sort" + "strconv" + "strings" + "sync" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + stdio "git.novox.be/novox/mesh-sdk/go" + + "nats-tools/internal/keeper" +) + +// The seat mesh-broker's snapshot verbs and nats_snapshots (design 47 G §681, ADR 0675; A-44, A-45, A-59). The bus module's tools +// bundle answers them as it answers every seat verb, with the caller the tool runner proved. snapshot-get reads the keeper's +// snapshot file and sends it in chunks (SeatPublish), each acknowledged by the caller (SeatAnswer's subscription); the verbs +// that act on the bus or remove a file are handed to the keeper through the state requests, as the persons' tools are. + +// ChunkSize is at most what one chunk carries: 512 KiB. +const ChunkSize = 512 << 10 + +// AckWithin is how long the next chunk waits for its acknowledgement. +const AckWithin = 30 * time.Second + +// callToken is the shape of the subject token a caller chooses for a send. +var callToken = regexp.MustCompile(`^[A-Za-z0-9_-]{16,64}$`) + +// ChunkSubject and AckSubject are the subjects of the nth chunk of a send and of its acknowledgement. +func ChunkSubject(call string, n int64) string { + return "mesh.seat." + meshbroker.Seat + ".chunk." + meshbroker.VerbSnapshotGet + "." + call + "." + strconv.FormatInt(n, 10) +} + +// AckSubject is the subject the caller acknowledges the nth chunk on. +func AckSubject(call string, n int64) string { + return "mesh.seat." + meshbroker.Seat + ".ack." + meshbroker.VerbSnapshotGet + "." + call + "." + strconv.FormatInt(n, 10) +} + +// AckPattern is every acknowledgement of snapshot-get. +const AckPattern = "mesh.seat.mesh-broker.ack.snapshot-get.*.*" + +// Chunk is the body of one chunk: its number, how many there are, and its bytes. +type Chunk struct { + N int64 `json:"n"` + Of int64 `json:"of"` + Data []byte `json:"data"` +} + +// snapshotWorld is what the snapshot verbs reach beside the requests: the file, the runtime's seat traffic and the bus's +// users. +type snapshotWorld struct { + // Open opens the snapshot file the keeper wrote. + Open func(path string) (io.ReadCloser, error) + // Publish says one chunk on its subject (stdio.SeatPublish), Answer subscribes to the acknowledgements (stdio.SeatAnswer). + Publish func(subject string, body any, id string) error + Answer func(subject string, handle func(subject string, body json.RawMessage) (any, error)) error + // Granted says whether the caller may hear the chunks and acknowledge them (the bus's users, nats_user_can). + Granted func(caller, call string) bool + Logf func(string, ...any) + // Within is how long an acknowledgement is waited for. + Within time.Duration + // KeepDays is snapshot-keep-days. + KeepDays int + + once sync.Once + onceErr error + mu sync.Mutex + waiting map[string]chan struct{} +} + +// liveSnapshots is the world as this bundle reaches it. +func liveSnapshots() *snapshotWorld { + return &snapshotWorld{ + Open: func(path string) (io.ReadCloser, error) { return os.Open(path) }, + Publish: stdio.SeatPublish, Answer: stdio.SeatAnswer, + Granted: func(caller, call string) bool { + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Second) + defer cancel() + conf, err := ReadUsers(ctx) + if err != nil { + return false + } + a, err := UserCan(conf, caller, "subscribe", ChunkSubject(call, 1)) + if err != nil || a["allowed"] != true { + return false + } + a, err = UserCan(conf, caller, "publish", AckSubject(call, 1)) + return err == nil && a["allowed"] == true + }, + Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "nats-tools: "+f+"\n", a...) }, + Within: AckWithin, KeepDays: int(number(os.Getenv, "MESH_NATS_SNAPSHOT_KEEP_DAYS", keeper.DefaultSnapshotKeepDays)), + } +} + +// hearAcks subscribes once to the acknowledgements of every send. +func (w *snapshotWorld) hearAcks() error { + w.once.Do(func() { + w.waiting = map[string]chan struct{}{} + w.onceErr = w.Answer(AckPattern, func(subject string, _ json.RawMessage) (any, error) { + p := strings.Split(subject, ".") + if len(p) == 7 { + w.mu.Lock() + ch := w.waiting[p[5]+"."+p[6]] + w.mu.Unlock() + if ch != nil { + select { + case ch <- struct{}{}: + default: + } + } + } + return map[string]any{}, nil + }) + }) + return w.onceErr +} + +// digestOf reads the file whole: its size and the `sha256:` of its bytes. +func (w *snapshotWorld) digestOf(path string) (int64, string, error) { + f, err := w.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 +} + +// get answers snapshot-get: the fields, then the chunks sent after the answer when the caller is granted the chunk subjects. +func (w *snapshotWorld) get(rs requestStore, caller, call, request string) (meshbroker.SnapshotGetAnswer, error) { + refuse := func(why string) (meshbroker.SnapshotGetAnswer, error) { + return meshbroker.SnapshotGetAnswer{Error: why}, nil + } + if !callToken.MatchString(call) { + return refuse("call is one token of 16 to 64 letters, digits, - or _: 128 random bits the caller chooses") + } + rec, err := rs.Get(request) + if err != nil { + return meshbroker.SnapshotGetAnswer{}, fmt.Errorf("the snapshot of the request %s cannot be read: %w", request, err) + } + if rec == nil || (rec.Path == "" && rec.Snapshot == "") { + return refuse("no snapshot of the request " + request + " is held") + } + if !rec.Recorded() { + return refuse("snapshot taken before milestone G") + } + // A protected snapshot is never handed out: the keeper restores it in place (O-7). + if rec.Protected { + return refuse("snapshot protected") + } + size, digest, err := w.digestOf(rec.Path) + if errors.Is(err, os.ErrNotExist) { + return refuse("snapshot missing") + } + if err != nil { + return meshbroker.SnapshotGetAnswer{}, fmt.Errorf("the snapshot %s cannot be read: %w", rec.Path, err) + } + if digest != rec.Digest || size != rec.Size { + return refuse("snapshot damaged: request " + request) + } + at := rec.At + out := meshbroker.SnapshotGetAnswer{Request: rec.ID, Module: rec.Module, Local: rec.Local, Stream: keeper.BusName(rec.Module, rec.Local), + Size: size, Digest: digest, At: &at} + if size == 0 || !w.Granted(caller, call) { + return out, nil + } + if err := w.hearAcks(); err != nil { + return meshbroker.SnapshotGetAnswer{}, fmt.Errorf("the acknowledgements cannot be heard, so no chunk is sent: %w", err) + } + out.Chunks = (size + ChunkSize - 1) / ChunkSize + go w.send(rec.Path, call, out.Chunks) + return out, nil +} + +// send sends the file in chunks, each acknowledged before the next; a missed acknowledgement ends the send, said in the journal. +func (w *snapshotWorld) send(path, call string, total int64) { + f, err := w.Open(path) + if err != nil { + w.Logf("the snapshot %s cannot be opened to send (%v): the send %s ends", path, err, call) + return + } + defer f.Close() + buf := make([]byte, ChunkSize) + for n := int64(1); n <= total; n++ { + read, err := io.ReadFull(f, buf) + if err != nil && !errors.Is(err, io.ErrUnexpectedEOF) && !errors.Is(err, io.EOF) { + w.Logf("the snapshot %s cannot be read (%v): the send %s ends at chunk %d", path, err, call, n) + return + } + key := call + "." + strconv.FormatInt(n, 10) + ack := make(chan struct{}, 1) + w.mu.Lock() + w.waiting[key] = ack + w.mu.Unlock() + err = w.Publish(ChunkSubject(call, n), Chunk{N: n, Of: total, Data: buf[:read]}, key) + if err == nil { + select { + case <-ack: + case <-time.After(w.Within): + err = fmt.Errorf("no acknowledgement within %s", w.Within) + } + } + w.mu.Lock() + delete(w.waiting, key) + w.mu.Unlock() + if err != nil { + w.Logf("the send %s of the snapshot %s ends at chunk %d of %d: %v", call, path, n, total, err) + return + } + } +} + +// handed hands a request to the keeper and reads its answer: accepted or busy with the act's id, or why not. +func handed(rs requestStore, r keeper.Request) (accepted, busy, why string) { + r.ID, r.At = requestID(rs.Now()), rs.Now() + if err := rs.Put(r.ID, r); err != nil { + return "", "", fmt.Sprintf("the request cannot be handed to the keeper: %v", err) + } + end, ok := awaitEnd(rs, r, "nats_requests").(*keeper.Request) + if !ok { + return "", "", fmt.Sprintf("the keeper has not answered request %s yet: nats_requests reads it", r.ID) + } + switch { + case end.Error != "": + return "", "", end.Error + case end.Outcome == "busy": + return "", end.ActID, "" + case end.ActID != "": + return end.ActID, "", "" + } + return end.ID, "", "" +} + +func boolean(description string) map[string]any { + return map[string]any{"type": "boolean", "description": description} +} + +func flag(a map[string]any, k string) bool { b, _ := a[k].(bool); return b } + +// snapshotTools are the four verbs of the seat mesh-broker that concern the keeper's snapshots, and nats_snapshots. +func snapshotTools(rs requestStore, w *snapshotWorld) []stdio.Tool { + why := func(a map[string]any) string { return strings.TrimSpace(arg(a, "why")) } + need := func(a map[string]any, names ...string) error { + for _, n := range names { + if strings.TrimSpace(arg(a, n)) == "" { + return fmt.Errorf("%s is required", n) + } + } + return nil + } + return []stdio.Tool{ + {Name: "mesh-broker.snapshot-get", + Description: "Send the stream snapshot the bus module took before a drop or recreate: its fields, then the file in chunks of at most 512 KiB on the chunk subject of the call, each acknowledged by the caller on the ack subject within 30 seconds. A caller granted no chunk subjects is answered the fields alone, with chunks 0. Refuses snapshot damaged, snapshot missing, snapshot protected and snapshot taken before milestone G (design 47 G 681).", + Input: map[string]any{"request": str("the snapshot's request, as nats_snapshots lists it"), "call": str("128 random bits the caller chooses: the subject token of this send's chunks and acknowledgements")}, + Run: func(a map[string]any) (any, error) { + if err := need(a, "request", "call"); err != nil { + return nil, err + } + return w.get(rs, stdio.CallerOf(a), arg(a, "call"), arg(a, "request")) + }}, + {Name: "mesh-broker.snapshot-restore", + Description: "Restore, in place, the snapshot the bus module took before a drop or recreate of a stream no other account hears: the keeper checks the file against its digest (with verify true it says so and restores nothing), refuses `names live` while a stream of its names exists, and restores it under its own name within 60 minutes. Answers at once accepted with the act's id, busy or error; the outcome comes on snapshot-restored (design 47 G 681).", + Input: map[string]any{"request": str("the snapshot's request"), "verify": boolean("check the file against its digest and restore nothing"), "why": str("why, which the hand-act log keeps")}, + Run: func(a map[string]any) (any, error) { + if err := need(a, "request", "why"); err != nil { + return nil, err + } + acc, busy, bad := handed(rs, keeper.Request{Verb: keeper.VerbSnapshotRestore, Of: arg(a, "request"), Verify: flag(a, "verify"), Why: why(a), Caller: stdio.CallerOf(a)}) + return meshbroker.SnapshotRestoreAnswer{Accepted: acc, Busy: busy, Error: bad}, nil + }}, + {Name: "mesh-broker.snapshot-restore-cancel", + Description: "Cancel a running snapshot restore by its act: the keeper removes its own partial stream and says snapshot-restored failed. Answers accepted, or an error when that act has ended (design 47 G 681).", + Input: map[string]any{"act": str("the restore's act id"), "why": str("why, which the hand-act log keeps")}, + Run: func(a map[string]any) (any, error) { + if err := need(a, "act", "why"); err != nil { + return nil, err + } + acc, busy, bad := handed(rs, keeper.Request{Verb: keeper.VerbSnapshotRestoreCancel, Of: arg(a, "act"), Why: why(a), Caller: stdio.CallerOf(a)}) + return meshbroker.SnapshotRestoreCancelAnswer{Accepted: acc, Busy: busy, Error: bad}, nil + }}, + {Name: "mesh-broker.snapshot-remove", + Description: "Remove the keeper's snapshot of one request now, or every snapshot past snapshot-keep-days (expired true), written in the hand-act log first; a snapshot of before milestone G goes only by its request (design 47 G 681).", + Input: map[string]any{"request": str("the snapshot's request"), "expired": boolean("every snapshot past snapshot-keep-days"), "why": str("why, which the hand-act log keeps")}, + Run: func(a map[string]any) (any, error) { + if err := need(a, "why"); err != nil { + return nil, err + } + acc, busy, bad := handed(rs, keeper.Request{Verb: keeper.VerbSnapshotRemove, Of: arg(a, "request"), AllExpired: flag(a, "expired"), Why: why(a), Caller: stdio.CallerOf(a)}) + return meshbroker.SnapshotRemoveAnswer{Accepted: acc, Busy: busy, Error: bad}, nil + }}, + {Name: "nats_snapshots", + Description: "The keeper's snapshots of kept data taken before a drop or recreate, from its state requests: each one's request, module, local name, stream, size, digest, time, whether it is protected (restored only in place, never sent), whether it is expired (past snapshot-keep-days, removed at the keeper's next request) and whether it is recorded (a snapshot of before milestone G holds no digest and is sent or removed by hand only).", + Run: func(map[string]any) (any, error) { + keys, err := rs.Keys() + if err != nil { + return nil, err + } + sort.Strings(keys) + out := []map[string]any{} + for _, key := range keys { + if strings.HasPrefix(key, keeper.ActedPrefix) { + continue + } + r, err := rs.Get(key) + if err != nil || r == nil || (r.Path == "" && r.Snapshot == "") { + continue + } + out = append(out, map[string]any{"request": r.ID, "module": r.Module, "local": r.Local, "stream": keeper.BusName(r.Module, r.Local), + "size": r.Size, "digest": r.Digest, "at": r.At, "protected": r.Protected, "expired": r.Expired(rs.Now(), w.KeepDays), "recorded": r.Recorded()}) + } + return map[string]any{"snapshots": out}, nil + }}, + } +} diff --git a/modules/nats/cmd/nats-tools/snapshots_test.go b/modules/nats/cmd/nats-tools/snapshots_test.go new file mode 100644 index 000000000..54d847841 --- /dev/null +++ b/modules/nats/cmd/nats-tools/snapshots_test.go @@ -0,0 +1,219 @@ +package main + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "os" + "runtime" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + stdio "git.novox.be/novox/mesh-sdk/go" + + "nats-tools/internal/keeper" +) + +// The seat mesh-broker's snapshot verbs and nats_snapshots (design 47 G §681, A-44, A-45, A-59): snapshot-get answers the fields and +// sends the chunks of at most 512 KiB, each acknowledged before the next, to a caller granted the chunk subjects, and never a +// protected snapshot; the other three verbs are handed to the keeper with the proved caller and answered in the one acting +// form; nats_snapshots lists each snapshot with its expired and recorded. +func TestTheSnapshotVerbsAreServedByTheBundle(t *testing.T) { + at := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) + file := bytes.Repeat([]byte("snap"), (ChunkSize*5/2)/4) // two whole chunks and a half + sum := sha256.Sum256(file) + digest := "sha256:" + hex.EncodeToString(sum[:]) + files := map[string][]byte{"/k/ok.tar": file, "/k/damaged.tar": []byte("other bytes"), "/k/protected.tar": file} + held := map[string]keeper.Request{ + "r1": {ID: "r1", Verb: "drop", Module: "m", Local: "drafts", At: at, Path: "/k/ok.tar", Size: int64(len(file)), Digest: digest}, + "r2": {ID: "r2", Verb: "drop", Module: "m", Local: "secret", At: at, Path: "/k/protected.tar", Size: int64(len(file)), Digest: digest, Protected: true}, + "r3": {ID: "r3", Verb: "drop", Module: "m", Local: "bad", At: at, Path: "/k/damaged.tar", Size: int64(len(file)), Digest: digest}, + "r4": {ID: "r4", Verb: "drop", Module: "m", Local: "gone", At: at, Path: "/k/gone.tar", Size: 5, Digest: digest}, + "r5": {ID: "r5", Verb: "recreate", Module: "m", Local: "old", At: at.AddDate(0, 0, -400), Snapshot: "/before/g.tar"}, + "r6": {ID: "r6", Verb: "drop", Module: "m", Local: "aged", At: at.AddDate(0, 0, -91), Path: "/k/ok.tar", Size: 1, Digest: digest}, + } + var handedOver []keeper.Request + rs := requestStore{ + Put: func(key string, r keeper.Request) error { + handedOver = append(handedOver, r) + // The keeper's answer at once, as its watch writes it. + r.Done = at + switch r.Of { + case "busy": + r.Outcome, r.ActID = "busy", "a-running" + case "bad": + r.Outcome, r.Error = "error", "names live: the stream KV_m_drafts exists" + default: + r.Outcome, r.ActID = "accepted", key + } + held[key] = r + return nil + }, + Get: func(key string) (*keeper.Request, error) { + r, ok := held[key] + if !ok { + return nil, nil + } + return &r, nil + }, + Keys: func() ([]string, error) { + var out []string + for k := range held { + out = append(out, k) + } + return out, nil + }, + Now: func() time.Time { return at }, + } + chunks := make(chan string, 8) + var published []Chunk + var ackHandler func(subject string, body json.RawMessage) (any, error) + var journal []string + var mu sync.Mutex + locked := func(f func()) { mu.Lock(); defer mu.Unlock(); f() } + granted := true + w := &snapshotWorld{ + Open: func(path string) (io.ReadCloser, error) { + b, ok := files[path] + if !ok { + return nil, os.ErrNotExist + } + return io.NopCloser(bytes.NewReader(b)), nil + }, + Publish: func(subject string, body any, id string) error { + c := body.(Chunk) + locked(func() { published = append(published, c) }) + chunks <- subject + " " + id + return nil + }, + Answer: func(subject string, handle func(string, json.RawMessage) (any, error)) error { + if subject != "mesh.seat.mesh-broker.ack.snapshot-get.*.*" { + t.Errorf("the acknowledgements are heard on %s", subject) + } + ackHandler = handle + return nil + }, + Granted: func(string, string) bool { return granted }, + Logf: func(f string, a ...any) { locked(func() { journal = append(journal, fmt.Sprintf(f, a...)) }) }, + Within: time.Millisecond, + KeepDays: 90, + } + tools := map[string]stdio.Tool{} + for _, tl := range snapshotTools(rs, w) { + tools[tl.Name] = tl + } + get := func(request, call string) meshbroker.SnapshotGetAnswer { + out, err := tools["mesh-broker.snapshot-get"].Run(map[string]any{"request": request, "call": call}) + if err != nil { + t.Fatal(err) + } + return out.(meshbroker.SnapshotGetAnswer) + } + call := "0123456789abcdef0123456789abcdef" + for _, row := range []struct{ request, want string }{{"r2", "snapshot protected"}, {"r3", "snapshot damaged"}, {"r4", "snapshot missing"}, + {"r5", "snapshot taken before milestone G"}} { + if got := get(row.request, call); got.Error != row.want && !strings.HasPrefix(got.Error, row.want) || got.Chunks != 0 || len(published) != 0 { + t.Errorf("%s: answered %+v, want %q and no chunk", row.request, got, row.want) + } + } + // A caller granted no chunk subjects is answered the fields alone. + granted = false + if got := get("r1", call); got.Error != "" || got.Chunks != 0 || got.Digest != digest || got.Stream != "m_drafts" || got.Size != int64(len(file)) || len(published) != 0 { + t.Errorf("an ungranted caller: %+v, published %d", got, len(published)) + } + // A granted one hears each chunk after the answer, the next only once the last is acknowledged. + granted = true + got := get("r1", call) + if got.Chunks != 3 { + t.Fatalf("answered %+v: 2.5 chunks of 512 KiB are three", got) + } + for n := 1; n <= 3; n++ { + want := fmt.Sprintf("mesh.seat.mesh-broker.chunk.snapshot-get.%s.%d %s.%d", call, n, call, n) + if s := <-chunks; s != want { + t.Fatalf("chunk %d sent as %q, want %q", n, s, want) + } + if _, err := ackHandler(fmt.Sprintf("mesh.seat.mesh-broker.ack.snapshot-get.%s.%d", call, n), nil); err != nil { + t.Fatal(err) + } + } + var whole []byte + locked(func() {}) + for _, c := range published { + if len(c.Data) > ChunkSize { + t.Errorf("chunk %d holds %d bytes, past 512 KiB", c.N, len(c.Data)) + } + whole = append(whole, c.Data...) + } + if !bytes.Equal(whole, file) { + t.Errorf("the chunks hold %d bytes, not the file's %d", len(whole), len(file)) + } + // A missed acknowledgement ends the send, said in the journal; no second chunk goes. + locked(func() { published = nil }) + _ = get("r1", call) + <-chunks + var said []string + var sent int + for deadline := time.Now().Add(5 * time.Second); len(said) == 0 && time.Now().Before(deadline); { + locked(func() { said, sent = append([]string(nil), journal...), len(published) }) + runtime.Gosched() + } + if len(said) != 1 || !strings.Contains(said[0], "ends at chunk 1 of 3") || sent != 1 { + t.Errorf("a missed acknowledgement: journal %v, published %d", said, sent) + } + // The three other verbs are handed to the keeper with the proved caller and answered in the acting form. + restore := tools["mesh-broker.snapshot-restore"] + out, err := restore.Run(map[string]any{"request": "r2", "verify": true, "why": "w", "caller": "person.mallory"}) + if err != nil { + t.Fatal(err) + } + if a := out.(meshbroker.SnapshotRestoreAnswer); a.Accepted == "" || a.Busy != "" || a.Error != "" { + t.Errorf("restore answered %+v", a) + } + if h := handedOver[len(handedOver)-1]; h.Verb != keeper.VerbSnapshotRestore || h.Of != "r2" || !h.Verify || h.Caller != stdio.Unproven { + t.Errorf("handed %+v: the request, verify and the proved caller, never person.mallory", h) + } + out, _ = restore.Run(map[string]any{"request": "busy", "why": "w"}) + if a := out.(meshbroker.SnapshotRestoreAnswer); a.Busy != "a-running" || a.Accepted != "" { + t.Errorf("a second restore answered %+v", a) + } + out, _ = restore.Run(map[string]any{"request": "bad", "why": "w"}) + if a := out.(meshbroker.SnapshotRestoreAnswer); !strings.Contains(a.Error, "names live") || a.Accepted != "" { + t.Errorf("a refused restore answered %+v", a) + } + if _, err := restore.Run(map[string]any{"request": "r2"}); err == nil { + t.Error("a restore with no why was handed over") + } + out, _ = tools["mesh-broker.snapshot-restore-cancel"].Run(map[string]any{"act": "a1", "why": "w"}) + if h := handedOver[len(handedOver)-1]; h.Verb != keeper.VerbSnapshotRestoreCancel || h.Of != "a1" || out.(meshbroker.SnapshotRestoreCancelAnswer).Accepted == "" { + t.Errorf("cancel handed %+v, answered %+v", h, out) + } + out, _ = tools["mesh-broker.snapshot-remove"].Run(map[string]any{"expired": true, "why": "w"}) + if h := handedOver[len(handedOver)-1]; h.Verb != keeper.VerbSnapshotRemove || !h.AllExpired || out.(meshbroker.SnapshotRemoveAnswer).Accepted == "" { + t.Errorf("remove handed %+v, answered %+v", h, out) + } + // nats_snapshots lists every snapshot, the expired and the unrecorded said. + out, err = tools["nats_snapshots"].Run(nil) + if err != nil { + t.Fatal(err) + } + list := out.(map[string]any)["snapshots"].([]map[string]any) + byRequest := map[string]map[string]any{} + for _, s := range list { + byRequest[s["request"].(string)] = s + } + if r := byRequest["r2"]; r["protected"] != true || r["stream"] != "m_secret" { + t.Errorf("r2 listed as %v", r) + } + if r := byRequest["r6"]; r["expired"] != true || r["recorded"] != true { + t.Errorf("r6 listed as %v: 91 days old, recorded", r) + } + if r := byRequest["r5"]; r["recorded"] != false || r["expired"] != false { + t.Errorf("r5 (before G) listed as %v: unrecorded, never expired by age", r) + } +} diff --git a/modules/nats/internal/bussnap/snapshot.go b/modules/nats/internal/bussnap/snapshot.go index 33c7ef03d..dd4517d40 100644 --- a/modules/nats/internal/bussnap/snapshot.go +++ b/modules/nats/internal/bussnap/snapshot.go @@ -415,7 +415,7 @@ func Into(nc *nats.Conn, dir string) func(ctx context.Context, streams []string) if dir == "" { return "", errors.New("no snapshots directory is set (MESH_NATS_SNAPSHOTS_DIR), so no snapshot is taken and nothing is removed") } - name := filepath.Join(dir, "entry-"+strings.Join(streams, "+")+"-"+time.Now().UTC().Format("20060102T150405Z")+".tar") + name := filepath.Join(dir, "entry-"+strings.Join(streams, "+")+"-"+time.Now().UTC().Format("20060102T150405.000000Z")+".tar") f, err := os.CreateTemp(dir, ".entry-*.partial") if err != nil { return "", err diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index ea76bf053..b638e8557 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -845,6 +845,62 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { t.Errorf("a cut-off restore: said %q, record %+v, removed %v", got, get("a5"), bus.deletedStreams) } }}, + // Design 47 G §681: snapshot-remove removes the snapshot of one request, or every one past its days, written in the + // hand-act log first; a snapshot that cannot be recorded stays; one of before G goes only by its request. + {"snapshot-remove removes by request or all expired, after the hand act", func(t *testing.T, k *Keeper, bus *fakeBus, out *said) { + store, files := k.Store.(*memStore), k.Files.(*memFiles) + var acts []string + actErr := error(nil) + k.Act = func(what, why, cause string) error { acts = append(acts, cause+": "+what); return actErr } + put := func(r Request) { + raw, _ := json.Marshal(r) + _, _ = store.Put(StateRequests, r.ID, raw) + } + get := func(id string) Request { + raw, _, _, _ := store.Get(StateRequests, id) + var got Request + _ = json.Unmarshal(raw, &got) + return got + } + old := k.now().AddDate(0, 0, -100) + soon := k.now().AddDate(0, 0, -80) + put(Request{ID: "s1", Verb: VerbDrop, Path: "/snapshots/a.tar", Size: 10, Digest: "sha256:a", At: k.now(), Done: k.now()}) + put(Request{ID: "s2", Verb: VerbDrop, Path: "/snapshots/b.tar", Size: 10, Digest: "sha256:b", At: soon, Done: soon}) + put(Request{ID: "s3", Verb: VerbDrop, Snapshot: "/before/g.tar", At: old, Done: old}) + ask := func(r Request) Request { + r.Caller = "person.jo" + put(r) + k.ActOnRequests() + return get(r.ID) + } + for _, r := range []Request{{ID: "x1", Verb: VerbSnapshotRemove, Why: "w"}, {ID: "x2", Verb: VerbSnapshotRemove, Of: "s1", AllExpired: true, Why: "w"}} { + if got := ask(r); got.Outcome != "error" || !strings.Contains(got.Error, "not both and not neither") { + t.Errorf("%s: %+v", r.ID, got) + } + } + actErr = errors.New("the log is down") + if got := ask(Request{ID: "x3", Verb: VerbSnapshotRemove, Of: "s1", Why: "w"}); got.Outcome != "error" || len(files.removed) != 0 || get("s1").Path == "" { + t.Errorf("an unrecorded hand act: %+v, removed %v", got, files.removed) + } + actErr = nil + if got := ask(Request{ID: "x4", Verb: VerbSnapshotRemove, Of: "s1", Why: "w"}); got.Outcome != "accepted" || fmt.Sprint(files.removed) != "[/snapshots/a.tar]" || get("s1").ID != "" { + t.Errorf("by request: %+v, removed %v", got, files.removed) + } + // The keeper's pass at each request removes what is past its days first, so the verb is driven apart from it. + base := k.now() + k.Now = func() time.Time { return base.AddDate(0, 0, 20) } + all := &Request{ID: "x5", Verb: VerbSnapshotRemove, AllExpired: true, Why: "w", Caller: "person.jo"} + if outcome, err := k.removeSnapshots(all); outcome != "accepted" || err != nil || all.Result != "1 removed" || get("s2").ID != "" || get("s3").ID == "" { + t.Errorf("all expired: %q %v %+v: s2 goes, s3 (before G) stays", outcome, err, all) + } + k.Now = func() time.Time { return base } + if got := ask(Request{ID: "x6", Verb: VerbSnapshotRemove, Of: "s3", Why: "w"}); got.Outcome != "accepted" || get("s3").ID != "" || files.removed[len(files.removed)-1] != "/before/g.tar" { + t.Errorf("a snapshot of before G by its request: %+v, removed %v", got, files.removed) + } + if len(acts) != 4 || !strings.Contains(acts[len(acts)-1], "asked by person.jo") { + t.Errorf("the hand acts: %v", acts) + } + }}, {"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 ace02a3ac..78dd2281f 100644 --- a/modules/nats/internal/keeper/requests.go +++ b/modules/nats/internal/keeper/requests.go @@ -54,11 +54,13 @@ type Request struct { // Of is the keeper's snapshot a snapshot-restore or snapshot-remove names, by its request; Verify checks only; ActID is the // restore's act (the id of the running one for a busy answer); Ended is when the restore ended, and Result how: done or failed // (design 47 G §681). - Of string `json:"of,omitempty"` - Verify bool `json:"verify,omitempty"` - ActID string `json:"act-id,omitempty"` - Ended time.Time `json:"ended,omitempty"` - Result string `json:"result,omitempty"` + Of string `json:"of,omitempty"` + Verify bool `json:"verify,omitempty"` + // AllExpired is a snapshot-remove of every snapshot past its days, and Result then how many went. + AllExpired bool `json:"all-expired,omitempty"` + ActID string `json:"act-id,omitempty"` + Ended time.Time `json:"ended,omitempty"` + Result string `json:"result,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. @@ -157,6 +159,8 @@ func (k *Keeper) act(r *Request) (string, error) { return k.startRestore(r) case VerbSnapshotRestoreCancel: return k.cancelRestore(r) + case VerbSnapshotRemove: + return k.removeSnapshots(r) } if !k.Viewed() { return "", fmt.Errorf("%w: the keeper holds no first view yet, so it acts on nothing", ErrNoWarrant) diff --git a/modules/nats/internal/keeper/restore.go b/modules/nats/internal/keeper/restore.go index 4ada2e3f9..b402d4f4b 100644 --- a/modules/nats/internal/keeper/restore.go +++ b/modules/nats/internal/keeper/restore.go @@ -21,6 +21,7 @@ import ( const ( VerbSnapshotRestore = meshbroker.VerbSnapshotRestore VerbSnapshotRestoreCancel = meshbroker.VerbSnapshotRestoreCancel + VerbSnapshotRemove = meshbroker.VerbSnapshotRemove ) // RestoreBound is the longest a restore runs: past it the keeper ends it as failed. diff --git a/modules/nats/internal/keeper/snapshots.go b/modules/nats/internal/keeper/snapshots.go index 6c1993284..7f4132c1c 100644 --- a/modules/nats/internal/keeper/snapshots.go +++ b/modules/nats/internal/keeper/snapshots.go @@ -175,12 +175,16 @@ func (k *Keeper) recordSnapshot(ctx context.Context, r *Request, path string, st // 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() { +func (k *Keeper) ExpireSnapshots() { k.expireSnapshots() } + +// expireSnapshots is ExpireSnapshots, answering how many it removed. +func (k *Keeper) expireSnapshots() int { recs, err := k.snapshotRecords() if err != nil { k.logf("the snapshots cannot be read: %v", err) - return + return 0 } + removed := 0 var used int64 for key, r := range recs { if !r.Expired(k.now(), k.keepDays()) { @@ -198,8 +202,52 @@ func (k *Keeper) ExpireSnapshots() { 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()) + removed++ } k.snapshotsFill(used) + return removed +} + +// removeSnapshots is the seat mesh-broker's snapshot-remove: the snapshot of one request, or every one past its days +// (all-expired). The hand act is written first, and a snapshot that cannot be recorded stays; a snapshot of before G goes only +// by its request (design 47 G §681). The answer is `accepted` with the request's own id as the act. +func (k *Keeper) removeSnapshots(r *Request) (string, error) { + if (r.Of == "") == !r.AllExpired { + return "error", fmt.Errorf("snapshot-remove names a request, or all-expired, not both and not neither") + } + what := "removed every snapshot past its days" + if r.Of != "" { + what = "removed the snapshot of the request " + r.Of + } + if k.Act != nil { + if err := k.Act(what+", asked by "+r.Caller, r.Why, "mesh-broker_snapshot_remove"); err != nil { + return "error", fmt.Errorf("the hand act cannot be recorded, so nothing is removed: %w", err) + } + } + if r.AllExpired { + r.Result = fmt.Sprintf("%d removed", k.expireSnapshots()) + return "accepted", nil + } + raw, _, found, err := k.Store.Get(StateRequests, r.Of) + var rec Request + if err != nil || !found || json.Unmarshal(raw, &rec) != nil || (rec.Path == "" && rec.Snapshot == "") { + return "error", fmt.Errorf("no snapshot of the request %s is held", r.Of) + } + path := rec.Path + if path == "" { + path = rec.Snapshot + } + if err := k.files().Remove(path); err != nil { + return "error", fmt.Errorf("the snapshot %s cannot be removed: %w", path, err) + } + if err := k.Store.Delete(StateRequests, r.Of); err != nil { + return "error", fmt.Errorf("the record of the removed snapshot %s cannot be removed: %w", r.Of, err) + } + k.logf("removed the snapshot %s of the request %s by hand, asked by %s", path, r.Of, r.Caller) + if used, err := k.SnapshotsUsed(""); err == nil { + k.snapshotsFill(used) + } + return "accepted", nil } // SnapshotsAtStart is the keeper's first look at the snapshots at its start: what is past its days removed, and the fill diff --git a/modules/nats/module.json b/modules/nats/module.json index cbaf95238..456946488 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -32,7 +32,11 @@ "list", "show", "check", - "orphans" + "orphans", + "snapshot-get", + "snapshot-restore", + "snapshot-restore-cancel", + "snapshot-remove" ] } ], @@ -53,7 +57,8 @@ "consumer-removed", "bucket-keys-removed", "stream-removed", - "bucket-expiry-set" + "bucket-expiry-set", + "snapshot-restored" ], "invokes": [ "seat:mesh-hand-acts.hand-act", @@ -148,7 +153,8 @@ "nats_drop", "nats_keeper_health", "nats_first_view_again", - "nats_requests" + "nats_requests", + "nats_snapshots" ], "guards": [ 8222 @@ -370,6 +376,7 @@ "MESH_NATS_CONTAINER": "mesh-broker-nats", "MESH_NATS_USERS_FILE": "/etc/nats/accounts.conf", "MESH_NATS_SNAPSHOTS_DIR": "${dir:snapshots}", + "MESH_NATS_SNAPSHOT_KEEP_DAYS": "90", "NATS_TOOLS_CREDENTIAL": "${dir:tools-secret}/broker-nats-tools" } },