diff --git a/modules/nats/cmd/mesh-nats-snapshot/bussnap.go b/modules/nats/cmd/mesh-nats-snapshot/bussnap.go new file mode 100644 index 000000000..e4d8a8f59 --- /dev/null +++ b/modules/nats/cmd/mesh-nats-snapshot/bussnap.go @@ -0,0 +1,22 @@ +package main + +import "nats-tools/internal/bussnap" + +// The snapshot itself is bussnap's, which the keeper takes one entry's with too (design 25 § 12.3). +type ( + SnapshotOptions = bussnap.SnapshotOptions + apiError = bussnap.APIError + streamState = bussnap.StreamState +) + +var ( + Snapshot = bussnap.Snapshot + StreamNames = bussnap.StreamNames + request = bussnap.Request + truncate = bussnap.Truncate +) + +const ( + apiStreamInfo = bussnap.APIStreamInfo + apiStreamRestore = bussnap.APIStreamRestore +) diff --git a/modules/nats/cmd/mesh-nats-snapshot/main.go b/modules/nats/cmd/mesh-nats-snapshot/main.go index a3232c77e..983b33393 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/main.go +++ b/modules/nats/cmd/mesh-nats-snapshot/main.go @@ -31,6 +31,7 @@ import ( "fmt" "io" "nats-tools/internal/busarchive" + "nats-tools/internal/bussnap" "os" "time" ) @@ -200,15 +201,4 @@ func logf(format string, args ...any) { fmt.Fprintf(os.Stderr, format+"\n", args...) } -func human(n int64) string { - const unit = 1024 - if n < unit { - return fmt.Sprintf("%d B", n) - } - div, exp := int64(unit), 0 - for v := n / unit; v >= unit; v /= unit { - div *= unit - exp++ - } - return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp]) -} +func human(n int64) string { return bussnap.Human(n) } diff --git a/modules/nats/cmd/nats-tools/keep.go b/modules/nats/cmd/nats-tools/keep.go index 9006ac9b6..3ecaf53ce 100644 --- a/modules/nats/cmd/nats-tools/keep.go +++ b/modules/nats/cmd/nats-tools/keep.go @@ -18,11 +18,13 @@ import ( "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asks" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/handacts" "git.novox.be/novox/mesh-sdk/go/process" + "nats-tools/internal/bussnap" "nats-tools/internal/keeper" ) @@ -174,7 +176,7 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer defer nc.Close() recorder := handacts.Recorder{Ask: conn.Call} k := &keeper.Keeper{ - Bus: keeper.JetStream{JS: js}, + Bus: keeper.JetStream{JS: js, Snap: bussnap.Into(nc, getenv("MESH_NATS_SNAPSHOTS_DIR"))}, Store: processStore{conn}, Emit: streamEmitter(js, account), Say: sayer(js), View: firstView(conn), Act: func(what, why, cause string) error { _, err := recorder.Record(handacts.Request{What: what, Why: why, Cause: cause}) @@ -198,7 +200,7 @@ func runKeeper(ctx context.Context, getenv func(string) string, logOut io.Writer Raise: percent(getenv, "MESH_NATS_FULL_RAISE", 75), Clear: percent(getenv, "MESH_NATS_FULL_CLEAR", 70), } 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), nil, logger.Printf) + return Keep(ctx, k, hearer(js, account, logger.Printf), requestsOf(conn), logger.Printf) } // hearer binds the keeper's durable consumer on EVENTS (_nats_keeper, filtered on manifest-accepted) and hands @@ -276,6 +278,17 @@ func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func case <-time.After(backoff): } } + asked := make(chan struct{}, 1) + if requests != nil { + if err := requests(ctx, func() { + select { + case asked <- struct{}{}: + default: + } + }); err != nil { + logf("the persons' requests cannot be watched (%v): acted on at each fill pass", err) + } + } tick := time.NewTicker(time.Minute) defer tick.Stop() retry := 10 * time.Second @@ -290,6 +303,10 @@ func Keep(ctx context.Context, k *keeper.Keeper, hear func(context.Context, func return ctx.Err() case <-tick.C: k.FillPass() + // A recreate or a drop waits for the operator's warrant: read again once a minute, as an asker reads its asks. + k.ActOnRequests() + case <-asked: + k.ActOnRequests() case <-heard: case <-again: again = nil @@ -313,3 +330,13 @@ func keepMain() { os.Exit(1) } } + +// requestsOf watches the keeper's own state requests, waking act at each change. +func requestsOf(c *process.Conn) func(context.Context, func()) error { + return func(_ context.Context, act func()) error { + return c.State(keeper.StateRequests).Watch("", func(stdio.StateChange) error { + act() + return nil + }) + } +} diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 7945dfcc0..f5b560287 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -181,5 +181,6 @@ func allTools() []stdio.Tool { } out := append(tools(monitor, ReadUsers, UsersWritten), removalTools(LiveBus(), SaidByTheRuntime)...) out = append(out, expiryTools(LiveBus(), SaidByTheRuntime, RecordedBySeat)...) - return append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...) + out = append(out, brokerTools(HeldInState, jsz, StreamHistory(LiveBus()))...) + return append(out, requestTools(RequestsInState())...) } diff --git a/modules/nats/cmd/nats-tools/requests.go b/modules/nats/cmd/nats-tools/requests.go new file mode 100644 index 000000000..3384b064e --- /dev/null +++ b/modules/nats/cmd/nats-tools/requests.go @@ -0,0 +1,131 @@ +package main + +import ( + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "strings" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" + + "nats-tools/internal/keeper" +) + +// A person's acting tools (design 25 § 12.3; CAT-N8): nats_create and nats_apply (acts), nats_recreate and nats_drop +// (dangerous, granted only by name). None holds a grant of the seat's acting verbs: each hands its request to the +// keeper through the keeper's own state requests, stamped with the caller the tool runner proved, and reads its end +// back. recreate and drop first raise the operator's ask, binding exactly that act; the keeper acts only on its warrant, +// after a snapshot. + +// requestStore is the keeper's state requests, as this bundle reaches it. +type requestStore struct { + Put func(key string, r keeper.Request) error + Get func(key string) (*keeper.Request, error) + Ask func(a asks.Ask) error + Now func() time.Time + Wait func(d time.Duration) +} + +// RequestsInState is the store through the runtime that launched this bundle. +func RequestsInState() requestStore { + s := stdio.State(keeper.StateRequests) + return requestStore{ + Put: func(key string, r keeper.Request) error { _, err := s.Put(key, r); return err }, + Get: func(key string) (*keeper.Request, error) { + e, err := s.Get(key) + if err != nil || e == nil { + return nil, err + } + var r keeper.Request + return &r, json.Unmarshal(e.Value, &r) + }, + Ask: func(a asks.Ask) error { return stdio.SeatPublish(asks.AskSubject(keeper.Module), a, a.ID) }, + Now: func() time.Time { return time.Now().UTC() }, + Wait: time.Sleep, + } +} + +func requestID(now time.Time) string { + var b [4]byte + _, _ = rand.Read(b[:]) + return now.Format("20060102t150405") + "-" + hex.EncodeToString(b[:]) +} + +// askFor is the operator's ask for a recreate or a drop of one entry, its approve option binding exactly that act. +func askFor(r keeper.Request, now time.Time) (asks.Ask, error) { + binds, err := asks.ActDigest(r.Act()) + if err != nil { + return asks.Ask{}, err + } + what := map[string]string{keeper.VerbDrop: "remove", keeper.VerbRecreate: "rebuild"}[r.Verb] + name := keeper.BusName(r.Module, r.Local) + return asks.Ask{ID: "bus-" + r.ID, Who: asks.Operator, Expires: now.Add(24 * time.Hour), + OnExpiry: "nothing is removed and the entry stays as it is", + About: keeper.ConditionKey(map[string]string{keeper.VerbDrop: "orphaned", keeper.VerbRecreate: "drifted"}[r.Verb], r.Module, r.Local), + Headline: fmt.Sprintf("%s's kept data %s: %s it?", r.Module, r.Local, what), + Explanation: fmt.Sprintf("%s asks to %s %s on the bus, %s. A snapshot of it is taken first; its data is lost from the bus.", + r.Caller, what, name, r.Why), + Options: []asks.Option{ + {ID: "go", Label: strings.ToUpper(what[:1]) + what[1:], Does: fmt.Sprintf("%s %s after a snapshot", what, name), Level: asks.Destroy, Binds: binds}, + {ID: "keep", Label: "Keep it", Does: "nothing is removed", Level: asks.Acknowledge}, + }}, nil +} + +func requestTools(rs requestStore) []stdio.Tool { + entry := map[string]any{"module": str("the module, e.g. mesh-knowledge"), "local": str("its local name, e.g. records"), + "why": str("why, in plain words: kept with the act"), "request": str("a request this tool made before: its end is read, nothing new is asked")} + tool := func(verb, description string) stdio.Tool { + return stdio.Tool{Name: "nats_" + verb, Description: description, Input: entry, Run: func(a map[string]any) (any, error) { + if id := arg(a, "request"); id != "" { + r, err := rs.Get(id) + if err != nil || r == nil { + return nil, fmt.Errorf("no request %s is held (%v)", id, err) + } + return r, nil + } + r := keeper.Request{Verb: verb, Module: arg(a, "module"), Local: arg(a, "local"), Why: strings.TrimSpace(arg(a, "why")), + // The caller the tool runner proved, never a name the arguments give. + Caller: stdio.CallerOf(a), At: rs.Now()} + if r.Module == "" || r.Local == "" || r.Why == "" { + return nil, fmt.Errorf("nats_%s names one entry, a module and its local name, and says why", verb) + } + r.ID = requestID(r.At) + if verb == keeper.VerbRecreate || verb == keeper.VerbDrop { + ask, err := askFor(r, r.At) + if err != nil { + return nil, err + } + r.Ask = ask.ID + if err := rs.Put(r.ID, r); err != nil { + return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err) + } + if err := rs.Ask(ask); err != nil { + return nil, fmt.Errorf("the operator's ask cannot be raised, so nothing will be removed: %w", err) + } + return map[string]any{"request": r.ID, "ask": ask.ID, "state": "the operator is asked; the keeper acts once they approve, after a snapshot", + "read": fmt.Sprintf("nats_%s with request %s", verb, r.ID)}, nil + } + if err := rs.Put(r.ID, r); err != nil { + return nil, fmt.Errorf("the request cannot be handed to the keeper: %w", err) + } + // The keeper acts on its watch of requests at once; its end is read back for at most half a minute. + for wait := 100 * time.Millisecond; wait < 20*time.Second; wait *= 2 { + rs.Wait(wait) + if got, err := rs.Get(r.ID); err == nil && got != nil && !got.Done.IsZero() { + return got, nil + } + } + return map[string]any{"request": r.ID, "state": "handed to the keeper, not ended yet", + "read": fmt.Sprintf("nats_%s with request %s", verb, r.ID)}, nil + }} + } + return []stdio.Tool{ + tool(keeper.VerbCreate, "Create one entry of a module's kept data on the bus now, as its accepted manifest asks for it (design 25 § 12.3): handed to the bus module's keeper, which alone may create, and recorded as a hand act."), + tool(keeper.VerbApply, "Apply what a module's manifest asks of one entry where the bus holds less (a larger cap, a longer age): only a difference that grows; handed to the keeper and recorded as a hand act."), + tool(keeper.VerbRecreate, "Rebuild one entry whose settings drifted in a way that does not grow: the operator is asked first, and the keeper acts only on their warrant, after a snapshot of the entry. Its data is lost from the bus (the snapshot keeps it). Dangerous."), + tool(keeper.VerbDrop, "Remove one orphan, an entry no accepted manifest asks for: the operator is asked first, and the keeper acts only on their warrant, after a snapshot of the entry. Dangerous."), + } +} diff --git a/modules/nats/cmd/nats-tools/requests_test.go b/modules/nats/cmd/nats-tools/requests_test.go new file mode 100644 index 000000000..fdda6566f --- /dev/null +++ b/modules/nats/cmd/nats-tools/requests_test.go @@ -0,0 +1,70 @@ +package main + +import ( + "strings" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/asks" + + "nats-tools/internal/keeper" +) + +// A person's acting tools hand their request to the keeper (design 25 § 12.3; CAT-N8): stamped with the caller the +// tool runner proved, whatever name the arguments give; a create's end read back; a drop raises the operator's ask +// first, binding exactly that act, and is not acted on by the tool. +func TestAPersonsRequestIsHandedToTheKeeperAsTheProvedCaller(t *testing.T) { + held := map[string]keeper.Request{} + var raised []asks.Ask + rs := requestStore{ + Put: func(key string, r keeper.Request) error { + if r.Verb == keeper.VerbCreate { + // The keeper's end, as its watch writes it. + r.Done, r.Outcome = time.Unix(1, 0), "created" + } + held[key] = r + return nil + }, + Get: func(key string) (*keeper.Request, error) { + r, ok := held[key] + if !ok { + return nil, nil + } + return &r, nil + }, + Ask: func(a asks.Ask) error { raised = append(raised, a); return nil }, + Now: func() time.Time { return time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC) }, + Wait: func(time.Duration) {}, + } + tools := map[string]stdio.Tool{} + for _, tl := range requestTools(rs) { + tools[tl.Name] = tl + } + out, err := tools["nats_create"].Run(map[string]any{"module": "m", "local": "log", "why": "a test", "caller": "person.mallory"}) + if err != nil { + t.Fatal(err) + } + if r := out.(*keeper.Request); r.Caller != stdio.Unproven || r.Outcome != "created" { + t.Errorf("answered %+v: acted for the caller the runtime proved (here none, %q), never person.mallory", r, stdio.Unproven) + } + out, err = tools["nats_drop"].Run(map[string]any{"module": "m", "local": "old", "why": "no manifest asks for it"}) + if err != nil { + t.Fatal(err) + } + answer := out.(map[string]any) + r := held[answer["request"].(string)] + if len(raised) != 1 || r.Ask != raised[0].ID || !r.Done.IsZero() { + t.Fatalf("raised %+v, held %+v: the ask raised, the request waiting for it", raised, r) + } + binds, _ := asks.ActDigest(asks.Act{"verb": "drop", "module": "m", "local": "old"}) + if o, _ := raised[0].Option("go"); o.Level != asks.Destroy || o.Binds != binds || !strings.Contains(raised[0].Explanation, "snapshot") { + t.Errorf("the ask is %+v: its option destroys exactly this drop, after a snapshot", raised[0]) + } + if _, err := tools["nats_drop"].Run(map[string]any{"module": "m", "local": "old"}); err == nil || !strings.Contains(err.Error(), "says why") { + t.Errorf("a drop without why: %v", err) + } + if got, _ := tools["nats_drop"].Run(map[string]any{"request": r.ID}); got.(*keeper.Request).ID != r.ID { + t.Errorf("reading the request back answered %+v", got) + } +} diff --git a/modules/nats/cmd/mesh-nats-snapshot/snapshot.go b/modules/nats/internal/bussnap/snapshot.go similarity index 78% rename from modules/nats/cmd/mesh-nats-snapshot/snapshot.go rename to modules/nats/internal/bussnap/snapshot.go index 1a153c622..33c7ef03d 100644 --- a/modules/nats/cmd/mesh-nats-snapshot/snapshot.go +++ b/modules/nats/internal/bussnap/snapshot.go @@ -1,4 +1,7 @@ -package main +// Package bussnap takes a snapshot of the bus's streams with the server's own snapshot API (novox/hq ADR 0235): the +// nightly dump of every stream, and the bus module's keeper's snapshot of one entry before it is rebuilt or removed +// (design 25 § 12.3). +package bussnap import ( "archive/tar" @@ -10,22 +13,26 @@ import ( "fmt" "io" "os" + "path/filepath" + "slices" "sort" + "strings" "sync" "time" - "github.com/nats-io/nats.go" "nats-tools/internal/busarchive" + + "github.com/nats-io/nats.go" ) // The JetStream API's subjects this program speaks. Each is granted to the bus's own module and // nothing else is (mesh-controller internal/broker, BusSnapshotGrants); a test holds the two lists // equal against a real server. const ( - apiStreamNames = "$JS.API.STREAM.NAMES" - apiStreamInfo = "$JS.API.STREAM.INFO." - apiStreamSnapshot = "$JS.API.STREAM.SNAPSHOT." - apiStreamRestore = "$JS.API.STREAM.RESTORE." + APIStreamNames = "$JS.API.STREAM.NAMES" + APIStreamInfo = "$JS.API.STREAM.INFO." + APIStreamSnapshot = "$JS.API.STREAM.SNAPSHOT." + APIStreamRestore = "$JS.API.STREAM.RESTORE." ) // ChunkSize is what the server is asked to send at a time: its own default. The server sends up to @@ -42,20 +49,22 @@ type SnapshotOptions struct { // TempDir is where one stream's archive waits until it is written out; the system's when empty. TempDir string Log func(format string, args ...any) + // Streams, when set, are the only streams taken (the keeper's snapshot of one entry); every stream when empty. + Streams []string } -// apiError is the error every JetStream API answer may carry. -type apiError struct { +// APIError is the error every JetStream API answer may carry. +type APIError struct { Code int `json:"code"` ErrCode int `json:"err_code"` Description string `json:"description"` } -func (e *apiError) Error() string { return fmt.Sprintf("%s (%d)", e.Description, e.Code) } +func (e *APIError) Error() string { return fmt.Sprintf("%s (%d)", e.Description, e.Code) } -// streamState is what this program reads of a stream's state; the whole of it is kept as the server +// StreamState is what this program reads of a stream's state; the whole of it is kept as the server // gave it, in the stream's meta. -type streamState struct { +type StreamState struct { Messages uint64 `json:"messages"` Bytes uint64 `json:"bytes"` FirstSeq uint64 `json:"first_seq"` @@ -71,8 +80,8 @@ type streamConfig struct { Storage string `json:"storage"` } -// request asks the JetStream API and decodes its answer, refusing an answer that carries an error. -func request(ctx context.Context, nc *nats.Conn, subject string, body any, into any) error { +// Request asks the JetStream API and decodes its answer, refusing an answer that carries an error. +func Request(ctx context.Context, nc *nats.Conn, subject string, body any, into any) error { var payload []byte if body != nil { var err error @@ -90,10 +99,10 @@ func request(ctx context.Context, nc *nats.Conn, subject string, body any, into return fmt.Errorf("%s: %w", subject, err) } var wrapped struct { - Error *apiError `json:"error"` + Error *APIError `json:"error"` } if err := json.Unmarshal(msg.Data, &wrapped); err != nil { - return fmt.Errorf("%s answered something that is not JSON: %q", subject, truncate(msg.Data)) + return fmt.Errorf("%s answered something that is not JSON: %q", subject, Truncate(msg.Data)) } if wrapped.Error != nil { return fmt.Errorf("%s: %w", subject, wrapped.Error) @@ -104,7 +113,7 @@ func request(ctx context.Context, nc *nats.Conn, subject string, body any, into return nil } -func truncate(b []byte) string { +func Truncate(b []byte) string { if len(b) > 200 { return string(b[:200]) + "…" } @@ -121,7 +130,7 @@ func StreamNames(ctx context.Context, nc *nats.Conn) ([]string, error) { Limit int `json:"limit"` Streams []string `json:"streams"` } - if err := request(ctx, nc, apiStreamNames, map[string]int{"offset": len(names)}, &page); err != nil { + if err := Request(ctx, nc, APIStreamNames, map[string]int{"offset": len(names)}, &page); err != nil { return nil, err } names = append(names, page.Streams...) @@ -148,6 +157,18 @@ func Snapshot(ctx context.Context, nc *nats.Conn, out io.Writer, o SnapshotOptio if err != nil { return nil, err } + if len(o.Streams) > 0 { + var only []string + for _, n := range names { + if slices.Contains(o.Streams, n) { + only = append(only, n) + } + } + if len(only) != len(o.Streams) { + return nil, fmt.Errorf("the bus holds %v of the streams %v asked for: a snapshot missing one would pass for a whole one", only, o.Streams) + } + names = only + } if len(names) == 0 { // The mesh's own streams always exist once the controller has run; a bus with none is not // one to call backed up. @@ -178,7 +199,7 @@ func Snapshot(ctx context.Context, nc *nats.Conn, out io.Writer, o SnapshotOptio var info struct { Config streamConfig `json:"config"` } - if err := request(ctx, nc, apiStreamInfo+name, nil, &info); err != nil { + if err := Request(ctx, nc, APIStreamInfo+name, nil, &info); err != nil { return nil, err } if info.Config.Storage == "memory" { @@ -202,7 +223,7 @@ func Snapshot(ctx context.Context, nc *nats.Conn, out io.Writer, o SnapshotOptio m.StreamBytes += int64(entry.Bytes) m.ArchiveBytes += entry.ArchiveBytes o.Log("%s: %d message(s), %s, last sequence %d, %d consumer(s) → %s in %.2fs", - name, entry.Messages, human(int64(entry.Bytes)), entry.LastSeq, entry.Consumers, human(entry.ArchiveBytes), entry.Seconds) + name, entry.Messages, Human(int64(entry.Bytes)), entry.LastSeq, entry.Consumers, Human(entry.ArchiveBytes), entry.Seconds) } m.Finished = time.Now().UTC() m.Seconds = m.Finished.Sub(m.Started).Seconds() @@ -315,7 +336,7 @@ func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File) State json.RawMessage `json:"state"` } req := map[string]any{"deliver_subject": deliver, "chunk_size": ChunkSize} - if err := request(ctx, nc, apiStreamSnapshot+name, req, &resp); err != nil { + if err := Request(ctx, nc, APIStreamSnapshot+name, req, &resp); err != nil { return busarchive.StreamEntry{}, nil, err } select { @@ -339,7 +360,7 @@ func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File) } var cfg streamConfig - var state streamState + var state StreamState if err := json.Unmarshal(resp.Config, &cfg); err != nil { return busarchive.StreamEntry{}, nil, fmt.Errorf("its configuration: %w", err) } @@ -371,3 +392,42 @@ func snapshotStream(ctx context.Context, nc *nats.Conn, name string, f *os.File) ArchiveSHA256: hex.EncodeToString(hash.Sum(nil)), Seconds: time.Since(started).Seconds(), }, meta, nil } + +// Human is a size in bytes as a person reads it: 1.5 MiB. +func Human(n int64) string { + const unit = 1024 + if n < unit { + return fmt.Sprintf("%d B", n) + } + div, exp := int64(unit), 0 + for v := n / unit; v >= unit; v /= unit { + div *= unit + exp++ + } + return fmt.Sprintf("%.1f %ciB", float64(n)/float64(div), "KMGTPE"[exp]) +} + +// Into is the keeper's snapshot of some streams before it rebuilds or removes them (design 25 § 12.3): one tar under +// dir, named for the streams and the time, written whole before its name is given, so a partial one is never taken +// for it. It answers the file's path. +func Into(nc *nats.Conn, dir string) func(ctx context.Context, streams []string) (string, error) { + return func(ctx context.Context, streams []string) (string, error) { + 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") + f, err := os.CreateTemp(dir, ".entry-*.partial") + if err != nil { + return "", err + } + defer os.Remove(f.Name()) + if _, err := Snapshot(ctx, nc, f, SnapshotOptions{Streams: streams, TempDir: dir}); err != nil { + f.Close() + return "", err + } + if err := f.Close(); err != nil { + return "", err + } + return name, os.Rename(f.Name(), name) + } +} diff --git a/modules/nats/internal/keeper/keeper_test.go b/modules/nats/internal/keeper/keeper_test.go index 95934c3d9..88eafc6f3 100644 --- a/modules/nats/internal/keeper/keeper_test.go +++ b/modules/nats/internal/keeper/keeper_test.go @@ -444,6 +444,72 @@ func TestTheKeeperMakesWhatAVerdictAsksForAndSaysIt(t *testing.T) { t.Errorf("health %+v: unhealthy, naming the verb", h) } }}, + // CAT-N8: a person's requests: a drop removes an orphan alone, only on the operator's warrant binding exactly + // that drop, after a snapshot, said dropped and recorded; a recreate rebuilds a drifted entry the same way. + {"a drop waits for the warrant, refuses another act's, then acts 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{drafts, log}}) + k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{log}}) + var acts []string + k.Act = func(what, why, cause string) error { acts = append(acts, cause+": "+what); return nil } + ask := asks.Ask{ID: "bus-1", Who: asks.Operator, Expires: k.now().Add(time.Hour)} + var w *asks.Warrant + k.Warrant = func(string) (*asks.Ask, *asks.Warrant, error) { return &ask, w, nil } + put := func(r Request) Request { + raw, _ := json.Marshal(r) + _, _ = store.Put(StateRequests, r.ID, raw) + k.ActOnRequests() + raw, _, _, _ = store.Get(StateRequests, r.ID) + var got Request + _ = json.Unmarshal(raw, &got) + return got + } + if got := put(Request{ID: "r0", Verb: VerbDrop, Module: "m", Local: "log", Ask: "bus-1", Caller: "person.jo"}); !strings.Contains(got.Error, "drop removes an orphan alone") { + t.Errorf("dropping an asked entry: %+v", got) + } + if got := put(Request{ID: "r1", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo"}); !got.Done.IsZero() || len(bus.deleted) != 0 { + t.Fatalf("acted with no warrant: %+v, deleted %v", got, bus.deleted) + } + binds := func(act asks.Act) string { d, _ := asks.ActDigest(act); return d } + choose := func(act asks.Act) { + ask.Options = []asks.Option{{ID: "go", Label: "Remove", Level: asks.Destroy, Binds: binds(act)}} + w = &asks.Warrant{Ask: "bus-1", Asker: Module, Outcome: asks.OutcomeChosen, By: &asks.Person{Who: asks.Operator}, + Option: "go", Label: "Remove", Level: asks.Destroy, At: k.now(), AskDigest: ask.Digest()} + } + choose(asks.Act{"verb": "drop", "module": "m", "local": "log"}) + if got := put(Request{ID: "r1", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo"}); !strings.Contains(got.Error, "bound") || len(bus.deleted) != 0 { + t.Fatalf("acted on a warrant for another act: %+v, deleted %v", got, bus.deleted) + } + choose(asks.Act{"verb": "drop", "module": "m", "local": "drafts"}) + out.reset() + got := put(Request{ID: "r2", Verb: VerbDrop, Module: "m", Local: "drafts", Ask: "bus-1", Caller: "person.jo", Why: "unused"}) + if got.Error != "" || fmt.Sprint(bus.snaps) != "[[KV_m_drafts]]" || fmt.Sprint(bus.deleted) != "[KV_m_drafts]" || + fmt.Sprint(out.events) != "[dropped m drafts]" || len(acts) != 1 || !strings.Contains(acts[0], "nats_drop: dropped drafts of m") || + !strings.Contains(acts[0], "person.jo") { + t.Errorf("ended %+v, snapshots %v, deleted %v, said %v, recorded %v", got, bus.snaps, bus.deleted, out.events, acts) + } + if _, made := k.Held()["m"].Made["drafts"]; made { + t.Error("the dropped entry is still held") + } + }}, + {"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}}) + k.Heard(Accepted{Module: "m", Revision: 2, BusState: []catalogue.BusStateEntry{entry("log", catalogue.StorageStream, catalogue.TemplateLog, map[string]string{"max-mib": "16"})}}) + binds, _ := asks.ActDigest(asks.Act{"verb": "recreate", "module": "m", "local": "log"}) + ask := asks.Ask{ID: "bus-2", Who: asks.Operator, Expires: k.now().Add(time.Hour), Options: []asks.Option{{ID: "go", Label: "Rebuild", Level: asks.Destroy, Binds: binds}}} + w := &asks.Warrant{Ask: "bus-2", Asker: Module, Outcome: asks.OutcomeChosen, By: &asks.Person{Who: asks.Operator}, + Option: "go", Label: "Rebuild", Level: asks.Destroy, At: k.now(), AskDigest: ask.Digest()} + k.Warrant = func(string) (*asks.Ask, *asks.Warrant, error) { return &ask, w, nil } + out.reset() + raw, _ := json.Marshal(Request{ID: "r3", Verb: VerbRecreate, Module: "m", Local: "log", Ask: "bus-2"}) + _, _ = store.Put(StateRequests, "r3", raw) + k.ActOnRequests() + if bus.held["m_log"].MaxBytes != 16*MiB || fmt.Sprint(out.events) != "[recreated m log]" || + fmt.Sprint(out.conds) != "[cleared nats.drifted.m_log]" || k.Held()["m"].Made["log"].Drift != "" { + t.Errorf("held %+v, said %v %v: rebuilt as asked, the drift cleared", bus.held["m_log"], out.events, out.conds) + } + }}, } { t.Run(row.name, func(t *testing.T) { k, bus, _, out := newKeeper(FirstView{Revision: 1}) diff --git a/modules/nats/internal/keeper/requests.go b/modules/nats/internal/keeper/requests.go new file mode 100644 index 000000000..a686c6bbe --- /dev/null +++ b/modules/nats/internal/keeper/requests.go @@ -0,0 +1,205 @@ +package keeper + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "git.novox.be/novox/mesh-sdk/go/asks" +) + +// A person's request (design 25 § 12.3; CAT-N8): the bus module's tools hold no grant of the seat's acting verbs, so +// each hands its request to the keeper through the keeper's own state requests, stamped with the caller the tool runner +// proved (never a name the request's body gives), and reads the keeper's answer back from the same key. create and +// apply act at once; recreate and drop only once the operator's warrant on the ask the tool raised is held, after a +// snapshot of the entry's streams. Every act is recorded as a hand act. + +// The verbs a request names. +const ( + VerbCreate = "create" + VerbApply = "apply" + VerbRecreate = "recreate" + VerbDrop = "drop" +) + +// Request is one person's request, the value of its key in StateRequests. +type Request struct { + ID string `json:"id"` + Verb string `json:"verb"` + Module string `json:"module"` + Local string `json:"local"` + Why string `json:"why"` + // Caller is who asked, as the tool runner proved it (stdio.CallerOf), written by the tool. + Caller string `json:"caller"` + At time.Time `json:"at"` + // Ask is the id of the operator's ask the tool raised for a recreate or a drop; the keeper acts only on its warrant. + Ask string `json:"ask,omitempty"` + // Done is when the keeper ended it, Outcome what it did, Error why not. + Done time.Time `json:"done,omitempty"` + Outcome string `json:"outcome,omitempty"` + Error string `json:"error,omitempty"` + Snapshot string `json:"snapshot,omitempty"` +} + +// Act is what an option of the operator's ask binds for a recreate or a drop: the verb and the entry (asks.ActDigest). +func (r Request) Act() asks.Act { + return asks.Act{"verb": r.Verb, "module": r.Module, "local": r.Local} +} + +// 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") + +// ActOnRequests acts on every request not ended yet, oldest first, and writes each one's end back to its key. +func (k *Keeper) ActOnRequests() { + keys, err := k.Store.Keys(StateRequests) + if err != nil { + k.logf("the persons' requests cannot be read: %v", err) + return + } + for _, key := range keys { + raw, _, found, err := k.Store.Get(StateRequests, key) + if err != nil || !found { + continue + } + var r Request + if json.Unmarshal(raw, &r) != nil || !r.Done.IsZero() { + continue + } + outcome, err := k.act(&r) + if errors.Is(err, ErrNoWarrant) { + continue + } + r.Done, r.Outcome = k.now(), outcome + if err != nil { + r.Error = err.Error() + } + data, _ := json.Marshal(r) + if _, err := k.Store.Put(StateRequests, key, data); err != nil { + k.logf("the end of the request %s cannot be written: %v", key, err) + } + } +} + +func (k *Keeper) act(r *Request) (string, error) { + if !k.Viewed() { + return "", fmt.Errorf("%w: the keeper holds no first view yet, so it acts on nothing", ErrNoWarrant) + } + held, _, found := k.asked(r.Module) + if !found { + return "nothing done", fmt.Errorf("the bus module holds no entry of %s", r.Module) + } + m, made := held.Made[r.Local] + asked := false + for _, e := range held.Entries { + asked = asked || e.Name == r.Local + } + switch r.Verb { + case VerbCreate, VerbApply: + if !asked { + return "nothing done", fmt.Errorf("no accepted manifest of %s asks for %s: the bus module creates only what one asks", r.Module, r.Local) + } + for _, e := range held.Entries { + if e.Name == r.Local { + k.makeEntry(&held, e) + } + } + k.write(held) + m = held.Made[r.Local] + k.record(r, fmt.Sprintf("%s of %s's %s: created %v, drift %q, waiting %q", r.Verb, r.Module, r.Local, m.Created, m.Drift, m.Waiting)) + if m.Refused != "" || m.Waiting != "" { + return "not made", fmt.Errorf("%s%s", m.Refused, m.Waiting) + } + return fmt.Sprintf("%s: the bus holds %s as asked%s", r.Verb, BusName(r.Module, r.Local), map[bool]string{true: ", but for a drift a person decides: " + m.Drift}[m.Drift != ""]), nil + case VerbRecreate, VerbDrop: + default: + return "nothing done", fmt.Errorf("%q is no verb a person asks the bus module for: create, apply, recreate or drop", r.Verb) + } + if !made || !m.Created { + return "nothing done", fmt.Errorf("the bus module made no %s of %s", r.Local, r.Module) + } + if r.Verb == VerbDrop && !m.Orphaned { + return "nothing done", fmt.Errorf("%s of %s is asked for by its manifest: drop removes an orphan alone", r.Local, r.Module) + } + if r.Verb == VerbRecreate && m.Orphaned { + return "nothing done", fmt.Errorf("%s of %s is an orphan: drop it, or ask for it in a manifest again", r.Local, r.Module) + } + if err := k.warranted(r); err != nil { + return "nothing done", err + } + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Minute) + defer cancel() + var streams []string + for _, o := range m.Plan.Objects { + streams = append(streams, o.Stream()) + } + path, err := k.Bus.Snapshot(ctx, streams) + if err != nil { + return "nothing done", fmt.Errorf("the snapshot before %s failed, so nothing was removed: %w", r.Verb, err) + } + r.Snapshot = path + for _, o := range m.Plan.Objects { + if err := k.Bus.Delete(ctx, o); err != nil { + return "partly done", fmt.Errorf("%s could not be removed after its snapshot at %s: %w", o.Stream(), path, err) + } + } + if r.Verb == VerbDrop { + delete(held.Made, r.Local) + k.write(held) + k.emit(EventDropped, m.Plan, held.Revision, r.Why, nil, nil) + k.clear(conditionOrphaned, r.Module, r.Local) + k.record(r, fmt.Sprintf("dropped %s of %s on the operator's warrant (snapshot %s)", r.Local, r.Module, path)) + return fmt.Sprintf("dropped %s, its snapshot at %s", BusName(r.Module, r.Local), path), nil + } + for _, o := range m.Plan.Objects { + if err := k.Bus.Create(ctx, o); err != nil { + m.Created, m.Refused = false, fmt.Sprintf("%s: %v", o.Stream(), err) + held.Made[r.Local] = m + k.write(held) + return "partly done", fmt.Errorf("%s was removed and could not be made again (its snapshot at %s): %w", o.Stream(), path, err) + } + } + wasDrift := m.Drift != "" + m.Drift = "" + held.Made[r.Local] = m + k.write(held) + k.emit(EventRecreated, m.Plan, held.Revision, r.Why, nil, nil) + if wasDrift { + k.clear(conditionDrifted, r.Module, r.Local) + } + k.record(r, fmt.Sprintf("recreated %s of %s on the operator's warrant (snapshot %s)", r.Local, r.Module, path)) + return fmt.Sprintf("recreated %s as asked, its old data's snapshot at %s", BusName(r.Module, r.Local), path), nil +} + +// warranted holds a recreate or a drop to the operator's warrant on the ask its tool raised: chosen by the operator, +// on that ask, binding exactly this act. +func (k *Keeper) warranted(r *Request) error { + if r.Ask == "" || k.Warrant == nil { + return fmt.Errorf("a %s acts only on the operator's warrant, and this request names no ask", r.Verb) + } + ask, w, err := k.Warrant(r.Ask) + if err != nil { + return fmt.Errorf("%w: the router's record of the ask %s cannot be read now: %v", ErrNoWarrant, r.Ask, err) + } + if ask == nil || w == nil { + return ErrNoWarrant + } + o, err := w.For(Module, *ask) + if err != nil { + return fmt.Errorf("no warrant: %w", err) + } + if o.Level != asks.Destroy { + return fmt.Errorf("the operator chose %s, so nothing is removed", o.Label) + } + return o.Performs(r.Act()) +} + +func (k *Keeper) record(r *Request, what string) { + if k.Act == nil { + return + } + if err := k.Act(what+", asked by "+r.Caller, r.Why, "nats_"+r.Verb); err != nil { + k.logf("the hand act %q could not be recorded: %v", what, err) + } +} diff --git a/modules/nats/module.json b/modules/nats/module.json index a12bb1c63..b59e56ac2 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -115,7 +115,11 @@ "nats_bucket_remove", "nats_stream_remove", "nats_bucket_expiry", - "nats_history" + "nats_history", + "nats_create", + "nats_apply", + "nats_recreate", + "nats_drop" ], "guards": [ 8222 @@ -232,7 +236,8 @@ "env": { "MESH_BROKER_FILE": "${dir:keeper-state}/broker-keeper", "MESH_NATS_FULL_RAISE": "75", - "MESH_NATS_FULL_CLEAR": "70" + "MESH_NATS_FULL_CLEAR": "70", + "MESH_NATS_SNAPSHOTS_DIR": "${dir:keeper-state}" }, "bus": { "state": [