diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index c36085ee1..483de4537 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -37,7 +37,7 @@ import ( // started by the node-engine as the manifest's process keeper. It reads the fill every minute from its start, asks the // indexer's first view once, then hears manifest-accepted and the operator's decided on its durable consumer, makes what // each verdict asks for, and acts on the persons' requests its tools put in its own state requests (requests.go) as they -// are written or answered. It serves nothing (ADR 0436 §3). +// are written or answered. It answers the seat mesh-broker's snapshot verbs itself (snapshots.go; design 47 G A-107). // AcceptedOf reads a manifest-accepted body: the seat mesh-catalog's typed event, with the indexer's revision of the // verdict when the event carries it (0 from an indexer that does not say it). @@ -280,6 +280,12 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer SnapshotsCap: number(getenv, "MESH_NATS_SNAPSHOTS_CAP", keeper.DefaultSnapshotsCap), SnapshotKeepDays: int(number(getenv, "MESH_NATS_SNAPSHOT_KEEP_DAYS", keeper.DefaultSnapshotKeepDays)), } + // The process answers the seat mesh-broker's snapshot verbs itself (design 47 G §681, A-107), with the caller the bus proved. + stopAnswers, err := conn.AnswerSeat(meshbroker.Seat, snapshotAnswers(k, heldRequest(k.Store), keeperSnapshots(js, nc, account, logger.Printf))) + if err != nil { + return fmt.Errorf("the seat %s's snapshot verbs cannot be answered: %w", meshbroker.Seat, err) + } + defer stopAnswers() 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) } diff --git a/modules/nats/cmd/nats-tools/keeper_integration_test.go b/modules/nats/cmd/nats-tools/keeper_integration_test.go index 90379740c..f0031effe 100644 --- a/modules/nats/cmd/nats-tools/keeper_integration_test.go +++ b/modules/nats/cmd/nats-tools/keeper_integration_test.go @@ -9,7 +9,6 @@ import ( "encoding/json" "errors" "fmt" - "io" "os" "slices" "strings" @@ -25,7 +24,6 @@ import ( "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" @@ -439,9 +437,10 @@ 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) { + // Design 47 G §681, A-107: the snapshot the keeper wrote is sent by the keeper's own process (snapshot-get answered through + // AnswerSeat, the chunks published by the process and acknowledged to its own subscription), restored in place by the keeper + // (refused while its name is live, ended by snapshot-restored) and removed by snapshot-remove, no call handed through the requests. + t.Run("a snapshot is sent, restored in place and removed by the keeper's process, which answers the seat itself", func(t *testing.T) { get := func(id string) keeper.Request { raw, _, _, _ := store.Get(keeper.StateRequests, id) var r keeper.Request @@ -452,64 +451,62 @@ func TestTheBusModuleRunsWhole(t *testing.T) { 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 - }, + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHUNKS", Subjects: []string{"mesh.seat.mesh-broker.chunk.>"}}); err != nil { + t.Fatal(err) } - 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}) + world := keeperSnapshots(js, nc, account, t.Logf) + world.Granted = func(string, string) bool { return true } + stopAnswering, err := pconn.AnswerSeat(meshbroker.Seat, snapshotAnswers(k, heldRequest(store), world)) 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) + defer stopAnswering() + // A caller of the seat: its call carries its bus name as the last token, and it acknowledges each chunk it hears. + verb := func(name string, args map[string]any, into any) { + t.Helper() + raw, _ := json.Marshal(args) + msg, err := nc.Request("mesh.seat.mesh-broker.call."+name+".person~jo", raw, 10*time.Second) + if err != nil { + t.Fatal(err) + } + var reply struct { + Result json.RawMessage `json:"result"` + Error string `json:"error"` + } + if err := json.Unmarshal(msg.Data, &reply); err != nil || reply.Error != "" { + t.Fatalf("%s answered %s (%v)", name, msg.Data, err) + } + if err := json.Unmarshal(reply.Result, into); err != nil { + t.Fatal(err) + } + } + call := "0123456789abcdef0123456789abcdef" + var chunkMu sync.Mutex + var chunks []Chunk + hear, err := nc.Subscribe("mesh.seat.mesh-broker.chunk.snapshot-get."+call+".*", func(m *nats.Msg) { + var c Chunk + _ = json.Unmarshal(m.Data, &c) + chunkMu.Lock() + chunks = append(chunks, c) + chunkMu.Unlock() + go func() { + _, _ = nc.Request(strings.Replace(m.Subject, ".chunk.", ".ack.", 1), []byte("{}"), 5*time.Second) + }() + }) + if err != nil { + t.Fatal(err) + } + defer hear.Unsubscribe() + var got meshbroker.SnapshotGetAnswer + verb(meshbroker.VerbSnapshotGet, map[string]any{"request": "r2", "call": call}, &got) + if got.Error != "" || got.Digest != rec.Digest || got.Chunks < 1 || got.Stream != "m_drafts" { + t.Fatalf("snapshot-get answered %+v for the keeper's record %+v", got, 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 + return len(chunks) > 0 && int64(len(chunks)) == chunks[0].Of }) chunkMu.Lock() for _, c := range chunks { @@ -520,11 +517,9 @@ func TestTheBusModuleRunsWhole(t *testing.T) { 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) + var o meshbroker.SnapshotRestoreAnswer + verb(meshbroker.VerbSnapshotRestore, map[string]any{"request": "r2", "why": why}, &o) + return o } // 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") { @@ -548,9 +543,10 @@ func TestTheBusModuleRunsWhole(t *testing.T) { 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) + var o meshbroker.SnapshotRemoveAnswer + verb(meshbroker.VerbSnapshotRemove, map[string]any{"request": "r2", "why": "no longer needed"}, &o) + if o.Accepted == "" { + t.Fatalf("snapshot-remove answered %+v", o) } if _, err := os.Stat(rec.Path); !os.IsNotExist(err) { t.Errorf("the snapshot file %s is still there: %v", rec.Path, err) diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 7b5a9b40d..535eeb75b 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -12,6 +12,7 @@ import ( stdio "git.novox.be/novox/mesh-sdk/go" "nats-tools/internal/busarchive" + "nats-tools/internal/keeper" ) func str(description string) map[string]any { @@ -168,8 +169,8 @@ 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). +// SnapshotVerbs are the verbs of the seat mesh-broker the keeper's process answers (design 47 G §681, A-107), over the bus through +// process.Conn.AnswerSeat (snapshots.go); the tools bundle serves none of them. 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 @@ -195,6 +196,6 @@ func allTools() []stdio.Tool { out = append(out, keeperHealthTool(StatusInState, HeldInState, jsz)) out = append(out, readTools(jsz, LiveMsgGet(LiveBus()))...) rs := RequestsInState() - out = append(out, snapshotTools(rs, liveSnapshots())...) + out = append(out, snapshotListTool(rs, int(number(os.Getenv, "MESH_NATS_SNAPSHOT_KEEP_DAYS", keeper.DefaultSnapshotKeepDays)))...) return append(out, requestTools(rs)...) } diff --git a/modules/nats/cmd/nats-tools/snapshots.go b/modules/nats/cmd/nats-tools/snapshots.go index 3d095b2a1..f1266342c 100644 --- a/modules/nats/cmd/nats-tools/snapshots.go +++ b/modules/nats/cmd/nats-tools/snapshots.go @@ -16,16 +16,21 @@ import ( "sync" "time" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/process" "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. +// The seat mesh-broker's snapshot verbs and nats_snapshots (design 47 G §681, ADR 0675; A-44, A-107). The keeper's process answers the +// four verbs itself over the bus (process.Conn.AnswerSeat), with the caller the bus proved: snapshot-get answers the fields and +// sends the file in chunks, published by the process and acknowledged by the caller on a subscription of the process's own; +// snapshot-restore, snapshot-restore-cancel and snapshot-remove are acted on by the keeper at once, with no hand-over through the +// state requests. The tools bundle keeps nats_snapshots, which lists them. // ChunkSize is at most what one chunk carries: 512 KiB. const ChunkSize = 512 << 10 @@ -56,21 +61,18 @@ type Chunk struct { 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. +// snapshotWorld is what the keeper's snapshot verbs reach: the file, the process's own publish and subscription, 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 says one chunk on its subject, the process's own publish; Hear subscribes to the acknowledgements, once. Publish func(subject string, body any, id string) error - Answer func(subject string, handle func(subject string, body json.RawMessage) (any, error)) error + Hear func(ack func(subject string)) 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 @@ -78,27 +80,54 @@ type snapshotWorld struct { waiting map[string]chan struct{} } -// liveSnapshots is the world as this bundle reaches it. -func liveSnapshots() *snapshotWorld { +// grantedBy is whether the bus's users let caller hear a send's chunks and acknowledge them. +func grantedBy(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 +} + +// keeperSnapshots is the world as the keeper's process reaches it: its own JetStream publish of a chunk, as the seat's events are +// said, and its own core subscription for the acknowledgements, answered at once since the caller asks them as requests. +func keeperSnapshots(js jetstream.JetStream, nc *nats.Conn, a process.Account, logf func(string, ...any)) *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) + Open: func(path string) (io.ReadCloser, error) { return os.Open(path) }, + Publish: func(subject string, body any, id string) error { + data, err := json.Marshal(body) if err != nil { - return false + return err } - 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 + msg := nats.NewMsg(subject) + msg.Header.Set("x-event-id", a.Module+"."+id) + msg.Header.Set("x-source", a.Module) + msg.Header.Set("x-node", a.Node) + msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano)) + msg.Header.Set("content-type", "application/json") + msg.Data = data + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(a.Module+"."+id)) + return err }, - 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)), + Hear: func(ack func(subject string)) error { + _, err := nc.Subscribe(AckPattern, func(m *nats.Msg) { + ack(m.Subject) + if m.Reply != "" { + _ = m.Respond([]byte("{}")) + } + }) + return err + }, + Granted: grantedBy, Logf: logf, Within: AckWithin, } } @@ -106,20 +135,20 @@ func liveSnapshots() *snapshotWorld { 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) { + w.onceErr = w.Hear(func(subject string) { 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: - } + if len(p) != 7 { + return + } + 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 @@ -141,14 +170,14 @@ func (w *snapshotWorld) digestOf(path string) (int64, string, error) { } // 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) { +func (w *snapshotWorld) get(held func(id string) (*keeper.Request, error), 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) + rec, err := held(request) if err != nil { return meshbroker.SnapshotGetAnswer{}, fmt.Errorf("the snapshot of the request %s cannot be read: %w", request, err) } @@ -224,84 +253,110 @@ func (w *snapshotWorld) send(path, call string, total int64) { } } -// 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, "", "" +// snapshotArgs are the arguments of the four verbs. +type snapshotArgs struct { + Request string `json:"request"` + Call string `json:"call"` + Act string `json:"act"` + Why string `json:"why"` + Verify bool `json:"verify"` + Expired bool `json:"expired"` } -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) - } +// argsOf reads a call's arguments and refuses one that lacks a name in need. +func argsOf(body json.RawMessage, need ...string) (snapshotArgs, error) { + var a snapshotArgs + if len(body) > 0 { + if err := json.Unmarshal(body, &a); err != nil { + return a, fmt.Errorf("the arguments are not JSON: %w", err) } - return nil } + a.Why = strings.TrimSpace(a.Why) + have := map[string]string{"request": a.Request, "call": a.Call, "act": a.Act, "why": a.Why} + for _, n := range need { + if have[n] == "" { + return a, fmt.Errorf("%s is required", n) + } + } + return a, nil +} + +// snapshotAnswers are the four verbs of the seat mesh-broker that concern the keeper's snapshots, as the keeper's process answers +// them (process.Conn.AnswerSeat): each with the caller the bus proved, "unproven" for a call that named none. +func snapshotAnswers(k *keeper.Keeper, held func(id string) (*keeper.Request, error), w *snapshotWorld) map[string]process.SeatHandler { + acting := func(verb, caller, of string, a snapshotArgs) keeper.Request { + if caller == "" { + caller = stdio.Unproven + } + return k.AnswerSnapshot(keeper.Request{ID: requestID(time.Now().UTC()), Verb: verb, Of: of, Verify: a.Verify, AllExpired: a.Expired, Why: a.Why, Caller: caller}) + } + return map[string]process.SeatHandler{ + meshbroker.VerbSnapshotGet: func(caller string, body json.RawMessage) (any, error) { + a, err := argsOf(body, "request", "call") + if err != nil { + return nil, err + } + if caller == "" { + caller = stdio.Unproven + } + return w.get(held, caller, a.Call, a.Request) + }, + meshbroker.VerbSnapshotRestore: func(caller string, body json.RawMessage) (any, error) { + a, err := argsOf(body, "request", "why") + if err != nil { + return nil, err + } + r := acting(keeper.VerbSnapshotRestore, caller, a.Request, a) + switch { + case r.Error != "": + return meshbroker.SnapshotRestoreAnswer{Error: r.Error}, nil + case r.Outcome == "busy": + return meshbroker.SnapshotRestoreAnswer{Busy: r.ActID}, nil + case r.ActID != "": + return meshbroker.SnapshotRestoreAnswer{Accepted: r.ActID}, nil + } + return meshbroker.SnapshotRestoreAnswer{Accepted: r.ID}, nil + }, + meshbroker.VerbSnapshotRestoreCancel: func(caller string, body json.RawMessage) (any, error) { + a, err := argsOf(body, "act", "why") + if err != nil { + return nil, err + } + r := acting(keeper.VerbSnapshotRestoreCancel, caller, a.Act, a) + if r.Error != "" { + return meshbroker.SnapshotRestoreCancelAnswer{Error: r.Error}, nil + } + return meshbroker.SnapshotRestoreCancelAnswer{Accepted: r.ActID}, nil + }, + meshbroker.VerbSnapshotRemove: func(caller string, body json.RawMessage) (any, error) { + a, err := argsOf(body, "why") + if err != nil { + return nil, err + } + r := acting(keeper.VerbSnapshotRemove, caller, a.Request, a) + if r.Error != "" { + return meshbroker.SnapshotRemoveAnswer{Error: r.Error}, nil + } + return meshbroker.SnapshotRemoveAnswer{Accepted: r.ID}, nil + }, + } +} + +// heldRequest reads one record of the keeper's own state requests. +func heldRequest(store keeper.Store) func(id string) (*keeper.Request, error) { + return func(id string) (*keeper.Request, error) { + raw, _, found, err := store.Get(keeper.StateRequests, id) + if err != nil || !found { + return nil, err + } + var r keeper.Request + return &r, json.Unmarshal(raw, &r) + } +} + +// snapshotListTool is nats_snapshots, which the tools bundle keeps beside the verbs the keeper's process answers. +func snapshotListTool(rs requestStore, keepDays int) []stdio.Tool { 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) { @@ -320,7 +375,7 @@ func snapshotTools(rs requestStore, w *snapshotWorld) []stdio.Tool { 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()}) + "size": r.Size, "digest": r.Digest, "at": r.At, "protected": r.Protected, "expired": r.Expired(rs.Now(), 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 index 54d847841..96a7ee8df 100644 --- a/modules/nats/cmd/nats-tools/snapshots_test.go +++ b/modules/nats/cmd/nats-tools/snapshots_test.go @@ -20,11 +20,48 @@ import ( "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) { +// mapStore is the keeper's state in memory. +type mapStore struct { + mu sync.Mutex + m map[string][]byte +} + +func (s *mapStore) Get(state, key string) ([]byte, uint64, bool, error) { + s.mu.Lock() + defer s.mu.Unlock() + v, ok := s.m[state+"/"+key] + return v, 1, ok, nil +} +func (s *mapStore) Put(state, key string, v []byte) (uint64, error) { + s.mu.Lock() + defer s.mu.Unlock() + s.m[state+"/"+key] = v + return 1, nil +} +func (s *mapStore) Delete(state, key string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.m, state+"/"+key) + return nil +} +func (s *mapStore) Keys(state string) ([]string, error) { + s.mu.Lock() + defer s.mu.Unlock() + var out []string + for k := range s.m { + if strings.HasPrefix(k, state+"/") { + out = append(out, strings.TrimPrefix(k, state+"/")) + } + } + return out, nil +} +func (s *mapStore) Create(state, key string, v []byte) (uint64, error) { return s.Put(state, key, v) } + +// The seat mesh-broker's snapshot verbs and nats_snapshots (design 47 G §681, A-44, A-107): the keeper's process answers snapshot-get +// with the fields and 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 it acts on itself, with the proved caller, and answers in the one acting form; +// nats_snapshots lists each snapshot with its expired and recorded. +func TestTheKeepersProcessAnswersTheSnapshotVerbs(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) @@ -38,23 +75,7 @@ func TestTheSnapshotVerbsAreServedByTheBundle(t *testing.T) { "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 { @@ -73,7 +94,7 @@ func TestTheSnapshotVerbsAreServedByTheBundle(t *testing.T) { } chunks := make(chan string, 8) var published []Chunk - var ackHandler func(subject string, body json.RawMessage) (any, error) + var ackHandler func(subject string) var journal []string var mu sync.Mutex locked := func(f func()) { mu.Lock(); defer mu.Unlock(); f() } @@ -92,24 +113,40 @@ func TestTheSnapshotVerbsAreServedByTheBundle(t *testing.T) { 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 + Hear: func(ack func(subject string)) error { + ackHandler = ack 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, + 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, } - tools := map[string]stdio.Tool{} - for _, tl := range snapshotTools(rs, w) { - tools[tl.Name] = tl + store := &mapStore{m: map[string][]byte{}} + snapFile := t.TempDir() + "/rm.tar" + if err := os.WriteFile(snapFile, []byte("x"), 0o600); err != nil { + t.Fatal(err) + } + put := func(r keeper.Request) { + raw, _ := json.Marshal(r) + _, _ = store.Put(keeper.StateRequests, r.ID, raw) + } + for _, r := range held { + put(r) + } + put(keeper.Request{ID: "r7", Verb: "drop", Module: "m", Local: "rm", At: at, Path: snapFile, Size: 1, Digest: "sha256:x"}) + put(keeper.Request{ID: "r5", Verb: "recreate", Module: "m", Local: "old", At: at, Snapshot: "/before/g.tar"}) + var acts []string + k := &keeper.Keeper{Store: store, Act: func(what, why, cause string) error { acts = append(acts, what+" | "+cause); return nil }} + answers := snapshotAnswers(k, heldRequest(store), w) + if len(answers) != 4 { + t.Fatalf("the process answers %d verbs, not the four snapshot verbs: %v", len(answers), answers) + } + answer := func(verb, caller string, args map[string]any) (any, error) { + raw, _ := json.Marshal(args) + return answers[verb](caller, raw) } get := func(request, call string) meshbroker.SnapshotGetAnswer { - out, err := tools["mesh-broker.snapshot-get"].Run(map[string]any{"request": request, "call": call}) + out, err := answers[meshbroker.VerbSnapshotGet]("person.jo", json.RawMessage(fmt.Sprintf(`{"request":%q,"call":%q}`, request, call))) if err != nil { t.Fatal(err) } @@ -138,9 +175,7 @@ func TestTheSnapshotVerbsAreServedByTheBundle(t *testing.T) { 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) - } + ackHandler(fmt.Sprintf("mesh.seat.mesh-broker.ack.snapshot-get.%s.%d", call, n)) } var whole []byte locked(func() {}) @@ -166,39 +201,36 @@ func TestTheSnapshotVerbsAreServedByTheBundle(t *testing.T) { 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"}) + // The three other verbs are acted on by the keeper itself, with the proved caller, and answered in the acting form. + out, err := answer(meshbroker.VerbSnapshotRestore, "person.jo", map[string]any{"request": "r5", "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 a := out.(meshbroker.SnapshotRestoreAnswer); a.Accepted != "" || a.Error != "snapshot taken before milestone G" { + t.Errorf("a restore of a snapshot of before G 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) + if _, err := answer(meshbroker.VerbSnapshotRestore, "person.jo", map[string]any{"request": "r2"}); err == nil { + t.Error("a restore with no why was answered") } - 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, _ = answer(meshbroker.VerbSnapshotRestoreCancel, "", map[string]any{"act": "a1", "why": "w"}) + if a := out.(meshbroker.SnapshotRestoreCancelAnswer); a.Accepted != "" || !strings.Contains(a.Error, "a1 has ended") { + t.Errorf("cancel of an act that is none 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) + out, err = answer(meshbroker.VerbSnapshotRemove, "", map[string]any{"request": "r7", "why": "w"}) + if a, _ := out.(meshbroker.SnapshotRemoveAnswer); err != nil || a.Accepted == "" || a.Error != "" { + t.Fatalf("remove answered %+v, %v", out, err) } - if _, err := restore.Run(map[string]any{"request": "r2"}); err == nil { - t.Error("a restore with no why was handed over") + if _, err := os.Stat(snapFile); !os.IsNotExist(err) { + t.Errorf("the snapshot file is still there: %v", err) } - 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) + if _, _, found, _ := store.Get(keeper.StateRequests, "r7"); found { + t.Error("the removed snapshot's record is still held") } - 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) + if len(acts) != 1 || !strings.Contains(acts[0], "asked by "+stdio.Unproven) { + t.Errorf("the hand acts %v: the caller of a call that named none is %s", acts, stdio.Unproven) } // nats_snapshots lists every snapshot, the expired and the unrecorded said. - out, err = tools["nats_snapshots"].Run(nil) + out, err = snapshotListTool(rs, 90)[0].Run(nil) if err != nil { t.Fatal(err) } diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index 9d936d2ad..360c6f5d8 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -783,9 +783,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { } ask := func(r Request) Request { r.Caller = "person.jo" - put(r) - k.ActOnRequests() - return get(r.ID) + return k.AnswerSnapshot(r) } put(Request{ID: "r20", Verb: VerbDrop, Path: path, Size: 100, Digest: digest, Protected: true, At: k.now(), Done: k.now()}) put(Request{ID: "r21", Verb: VerbDrop, Path: path, Size: 100, Digest: "sha256:bad", At: k.now(), Done: k.now()}) @@ -837,6 +835,13 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { if got := ask(Request{ID: "c2", Verb: VerbSnapshotRestoreCancel, Of: "a3", Why: "w"}); got.Outcome != "error" { t.Errorf("cancel of an ended act: %+v", got) } + // A restore written into the state requests is no request of a person's: the keeper answers the verb on the seat alone (A-107). + before := len(bus.restored) + put(Request{ID: "z1", Verb: VerbSnapshotRestore, Of: "r20", Why: "w", Caller: "person.jo"}) + k.ActOnRequests() + if got := get("z1"); len(bus.restored) != before || got.Outcome == "accepted" || got.Error == "" { + t.Errorf("a restore handed through the requests: %+v, restored %v", got, bus.restored) + } // A restore a restart cut off is ended at the keeper's start. put(Request{ID: "a5", Verb: VerbSnapshotRestore, Of: "r20", Outcome: "accepted", Done: k.now()}) k.EndCutOffRestores() @@ -869,9 +874,7 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { 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) + return k.AnswerSnapshot(r) } 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") { diff --git a/modules/nats/internal/keeper/requests.go b/modules/nats/internal/keeper/requests.go index 78dd2281f..ef07a9742 100644 --- a/modules/nats/internal/keeper/requests.go +++ b/modules/nats/internal/keeper/requests.go @@ -97,6 +97,33 @@ func (r Request) Act() occ.Act { return occ.Act{"verb": r.Verb, "module": r.Module, "local": r.Local} } +// AnswerSnapshot answers one call of the seat mesh-broker's snapshot-restore, snapshot-restore-cancel or snapshot-remove at once, as +// the keeper's process answers it (design 47 G §681, A-107): r names the verb, its arguments and the proved caller, and comes back +// ended with its Outcome (accepted, busy, verified or error), ActID and Error. Nothing is handed through the state requests but the +// record of a restore that runs, which its end and a restart's cut-off read. +func (k *Keeper) AnswerSnapshot(r Request) Request { + if r.At.IsZero() { + r.At = k.now() + } + var outcome string + var err error + switch r.Verb { + case VerbSnapshotRestore: + outcome, err = k.startRestore(&r) + case VerbSnapshotRestoreCancel: + outcome, err = k.cancelRestore(&r) + case VerbSnapshotRemove: + outcome, err = k.removeSnapshots(&r) + default: + outcome, err = "error", fmt.Errorf("%q is no snapshot verb", r.Verb) + } + r.Done, r.Outcome = k.now(), outcome + if err != nil { + r.Error = err.Error() + } + return r +} + // ErrNoWarrant is a request that waits for the operator's word: kept, acted on once the warrant is held. var ErrNoWarrant = errors.New("the operator has not answered yet") @@ -154,14 +181,6 @@ func (k *Keeper) act(r *Request) (string, error) { } return fmt.Sprintf("the first view answered at revision %d", s.Revision), nil } - switch r.Verb { - case VerbSnapshotRestore: - 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 b402d4f4b..55a3005d1 100644 --- a/modules/nats/internal/keeper/restore.go +++ b/modules/nats/internal/keeper/restore.go @@ -111,6 +111,18 @@ func (k *Keeper) startRestore(r *Request) (string, error) { k.restoring = run k.mu.Unlock() r.ActID = run.act + // The act's record is written before the restore runs, as the end of the restore and a restart's cut-off read it; a record that + // cannot be written ends the start, so no restore runs that nothing could find again. + held := *r + held.Done, held.Outcome = k.now(), "accepted" + data, _ := json.Marshal(held) + if _, err := k.Store.Put(StateRequests, r.ID, data); err != nil { + k.mu.Lock() + k.restoring = nil + k.mu.Unlock() + cancel() + return "error", fmt.Errorf("the restore's record cannot be written, so nothing is restored: %w", err) + } k.record(r, fmt.Sprintf("restores the snapshot of the request %s (%s) in place", rec.ID, rec.Path)) go k.restore(ctx, run, rec, r.Caller, r.Why) return "accepted", nil