From b6884796e7fb87d7c4c66dc0d8077ada12f3dcdc Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 11 Nov 2026 12:08:05 +0100 Subject: [PATCH] NATS4: the bus module takes and restores a restore point of a module's kept data, beside the live data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Verbs snapshot-kept and restore-kept (with aside-only and discard-aside) on the seat mesh-broker, handed to the keeper; taken over from the join's 0480e16f and c4e88eaa, without the verbs aside and discard-aside and the tool nats_kept, which no design line gives. Builds: design 47 G §296, §297, §302, §337, §405 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- modules/nats/README.md | 5 +- modules/nats/cmd/nats-tools/keptools.go | 95 ++++---------------- modules/nats/cmd/nats-tools/keptools_test.go | 21 +++-- modules/nats/cmd/nats-tools/main.go | 6 +- modules/nats/internal/keeper/kept.go | 47 +++++++--- modules/nats/internal/keeper/kept_test.go | 13 +++ modules/nats/internal/keeper/requests.go | 6 +- modules/nats/module.json | 7 +- 8 files changed, 91 insertions(+), 109 deletions(-) diff --git a/modules/nats/README.md b/modules/nats/README.md index ec6b5700f..28e5b9a51 100644 --- a/modules/nats/README.md +++ b/modules/nats/README.md @@ -144,8 +144,9 @@ module's kept data on the bus*). Its seat verbs, served by the tools and handed `mesh-broker.snapshot-kept` (the data as the server's snapshot, the module's consumers' positions and the catch-up tool, in one directory under the keeper's state), `mesh-broker.restore-kept` (the live data is snapshotted into a set-aside directory under a name of its own and only then removed; the restore point is put in its place and the consumers on other -streams are set back), `mesh-broker.discard-aside` (only on a call that says why, recorded first) and `mesh-broker.aside`; -the tool `nats_kept` reads the restore points and the set-aside data from the keeper's state `kept`. The backup holder +streams are set back; with `aside-only` it sets the live data aside and restores nothing, with `discard-aside`, naming data +set aside, it removes that data on a call that says why, recorded first, and restores nothing). Both are called by the +holder of `node-backup` on the same node alone. The backup holder keeps the restore points; it calls these verbs through the seat. ## Files it does not know diff --git a/modules/nats/cmd/nats-tools/keptools.go b/modules/nats/cmd/nats-tools/keptools.go index e5efeeede..87545381a 100644 --- a/modules/nats/cmd/nats-tools/keptools.go +++ b/modules/nats/cmd/nats-tools/keptools.go @@ -1,9 +1,7 @@ package main import ( - "encoding/json" "fmt" - "sort" "strings" "time" @@ -16,49 +14,15 @@ import ( // the runtime's own bound on a call, so a long snapshot is answered as "handed, not ended" and read again with its request. const keptWait = 50 * time.Second -// KeptHeld reads the keeper's state kept: the restore points it took and the data it set aside. -type KeptHeld func() ([]keeper.Point, []keeper.Aside, error) - -// KeptInState is the state kept through the runtime that launched this bundle. -func KeptInState() ([]keeper.Point, []keeper.Aside, error) { - st := stdio.State(keeper.StateKept) - keys, err := st.Keys() - if err != nil { - return nil, nil, err - } - sort.Strings(keys) - var points []keeper.Point - var aside []keeper.Aside - for _, key := range keys { - e, err := st.Get(key) - if err != nil || e == nil { - continue - } - switch { - case strings.HasPrefix(key, "point."): - var p keeper.Point - if json.Unmarshal(e.Value, &p) == nil { - points = append(points, p) - } - case strings.HasPrefix(key, "aside."): - var a keeper.Aside - if json.Unmarshal(e.Value, &a) == nil { - aside = append(aside, a) - } - } - } - return points, aside, nil -} - func formOf(a map[string]any) int { f, _ := a["form"].(float64) return int(f) } -// keptTools are the seat mesh-broker's verbs for a module's kept data and the bus module's tool that reads what it holds: +// keptTools are the seat mesh-broker's verbs for a module's kept data (design 47 G §405): // each verb hands its request to the keeper, which acts on its own account, and waits for the end on the watch of its // requests (never a poll). Asked again with `request`, a verb reads the end of the request it made. -func keptTools(rs requestStore, held KeptHeld) []stdio.Tool { +func keptTools(rs requestStore) []stdio.Tool { entry := func(extra map[string]any) map[string]any { in := map[string]any{"module": str("the module whose kept data it is, e.g. mesh-delivery-indexer"), "local-name": str("the kept data's local name, e.g. modules"), "form": num("the kept data's form (a manifest field; absent reads 0)"), "fingerprint": str("the fingerprint the restore point was taken before"), @@ -112,16 +76,6 @@ func keptTools(rs requestStore, held KeptHeld) []stdio.Tool { } return nil } - readAside := func() (any, error) { - _, aside, err := held() - if err != nil { - return nil, err - } - if aside == nil { - aside = []keeper.Aside{} - } - return map[string]any{"aside": aside}, nil - } return []stdio.Tool{ {Name: "mesh-broker.snapshot-kept", Description: "Take the restore point of one module's kept data on the bus (milestone G, design 47): the data whole as the server's own snapshot, the position of each durable consumer the module reads with, and the catch-up tool the newer manifest declares, kept by the bus module's keeper. Answers the directory the restore point is in. Handed to the keeper, recorded as a hand act.", @@ -131,41 +85,28 @@ func keptTools(rs requestStore, held KeptHeld) []stdio.Tool { return point(a, r) })}, {Name: "mesh-broker.restore-kept", - Description: "Restore one module's kept data on the bus from a restore point beside the live data, never over it (design 47): the live data is snapshotted into a set-aside directory under a name of its own and then removed from the bus, the restore point's streams are put in its place and the module's other consumers are set back to their positions. Answers what was set aside and the catch-up tool the restore point carries. Handed to the keeper, recorded as a hand act.", - Input: entry(map[string]any{"from": str("the directory the restore point was restored to, an absolute path")}), + Description: "Restore one module's kept data on the bus from a restore point beside the live data, never over it (design 47 G §405): the live data is snapshotted into a set-aside directory under a name of its own and then removed from the bus, the restore point's streams are put in its place and the module's other consumers are set back to their positions. With `aside-only` it sets the live data aside and restores nothing; with `discard-aside`, naming data set aside, it removes that data and restores nothing. Answers what was set aside and the catch-up tool the restore point carries. Handed to the keeper, recorded as a hand act.", + Input: entry(map[string]any{"from": str("the directory the restore point was restored to, an absolute path"), + "aside-only": map[string]any{"type": "boolean", "description": "set the live data aside and restore nothing"}, + "discard-aside": str("the id of data set aside: remove it and restore nothing")}), Run: run(keeper.VerbRestoreKept, func(a map[string]any, r *keeper.Request) error { + if id := arg(a, "discard-aside"); id != "" { + r.Verb, r.From = keeper.VerbDiscardAside, id + return nil + } + if b, _ := a["aside-only"].(bool); b { + r.AsideOnly = true + r.Module, r.Local = arg(a, "module"), arg(a, "local-name") + if r.Module == "" || r.Local == "" { + return fmt.Errorf("aside-only is of one kept data: a module and its local name") + } + return nil + } r.From = arg(a, "from") if r.From == "" { return fmt.Errorf("a restore names the directory the restore point is in") } return point(a, r) })}, - {Name: "mesh-broker.discard-aside", - Description: "Remove the data a restore set aside, by its id (from `aside`), only on a person's call: it says who and why, and the keeper records the hand act before it removes anything.", - Input: map[string]any{"id": str("the id of the data set aside"), "why": str("why, in plain words: kept with the act"), "request": str("a request made before: its end is read")}, - Run: run(keeper.VerbDiscardAside, func(a map[string]any, r *keeper.Request) error { - r.From = arg(a, "id") - if r.From == "" { - return fmt.Errorf("discard-aside names the id of the data set aside") - } - return nil - })}, - {Name: "mesh-broker.aside", Description: "The data restores set aside, each with its module, kept data, size, when, who and why: what a person decides to discard.", - Run: func(map[string]any) (any, error) { return readAside() }}, - {Name: "nats_kept", - Description: "What the bus module holds of its modules' kept data besides the live data: every restore point it took (module, kept data, form, the fingerprint it was taken before, its consumers' positions, the catch-up tool, its size) and the data a restore set aside. Read from the keeper's state kept.", - Run: func(map[string]any) (any, error) { - points, aside, err := held() - if err != nil { - return nil, err - } - if points == nil { - points = []keeper.Point{} - } - if aside == nil { - aside = []keeper.Aside{} - } - return map[string]any{"points": points, "aside": aside}, nil - }}, } } diff --git a/modules/nats/cmd/nats-tools/keptools_test.go b/modules/nats/cmd/nats-tools/keptools_test.go index 4205dc0f1..a088cd704 100644 --- a/modules/nats/cmd/nats-tools/keptools_test.go +++ b/modules/nats/cmd/nats-tools/keptools_test.go @@ -33,9 +33,8 @@ func TestTheKeptDataVerbsAreHandedToTheKeeperAndReadBack(t *testing.T) { return &r }, } - aside := []keeper.Aside{{ID: "idx.modules.1", Module: "idx", Local: "modules", Bytes: 12}} tools := map[string]stdio.Tool{} - for _, tl := range keptTools(rs, func() ([]keeper.Point, []keeper.Aside, error) { return nil, aside, nil }) { + for _, tl := range keptTools(rs) { tools[tl.Name] = tl } for _, verb := range KeptVerbs { @@ -53,10 +52,11 @@ func TestTheKeptDataVerbsAreHandedToTheKeeperAndReadBack(t *testing.T) { t.Errorf("answered %+v: the request for form 1 before abc, for the caller the runtime proved, with the directory the keeper answered", r) } for name, args := range map[string]map[string]any{ - "mesh-broker.snapshot-kept": {"module": "idx", "local-name": "modules", "fingerprint": "abc"}, - "mesh-broker.restore-kept": {"module": "idx", "local-name": "modules", "fingerprint": "abc", "why": "revert"}, - "mesh-broker.discard-aside": {"why": "x"}, + "mesh-broker.snapshot-kept": {"module": "idx", "local-name": "modules", "fingerprint": "abc"}, + "mesh-broker.restore-kept": {"module": "idx", "local-name": "modules", "fingerprint": "abc", "why": "revert"}, + "mesh-broker.restore-kept#aside-only": {"aside-only": true, "why": "x"}, } { + name, _, _ = strings.Cut(name, "#") if _, err := tools[name].Run(args); err == nil { t.Errorf("%s with %v was handed to the keeper", name, args) } @@ -65,9 +65,14 @@ func TestTheKeptDataVerbsAreHandedToTheKeeperAndReadBack(t *testing.T) { "from": "/restored/f0-abc", "why": "revert"}); err != nil || out.(*keeper.Request).From != "/restored/f0-abc" { t.Errorf("restore answered %+v (%v)", out, err) } - read, err := tools["mesh-broker.aside"].Run(nil) - if err != nil || !strings.Contains(strings.ReplaceAll(strings.TrimSpace(toJSON(read)), " ", ""), `"id":"idx.modules.1"`) { - t.Errorf("aside answered %v (%v)", read, err) + // `aside-only` sets the live data aside and restores nothing; `discard-aside` removes the data set aside (§405). + if out, err := tools["mesh-broker.restore-kept"].Run(map[string]any{"module": "idx", "local-name": "modules", "aside-only": true, "why": "revert"}); err != nil || + !out.(*keeper.Request).AsideOnly || out.(*keeper.Request).Verb != keeper.VerbRestoreKept { + t.Errorf("aside-only answered %+v (%v)", out, err) + } + if out, err := tools["mesh-broker.restore-kept"].Run(map[string]any{"discard-aside": "idx.modules.1", "why": "operator's call"}); err != nil || + out.(*keeper.Request).Verb != keeper.VerbDiscardAside || out.(*keeper.Request).From != "idx.modules.1" { + t.Errorf("discard-aside answered %+v (%v)", out, err) } } diff --git a/modules/nats/cmd/nats-tools/main.go b/modules/nats/cmd/nats-tools/main.go index 1e277fdd5..d9f614485 100644 --- a/modules/nats/cmd/nats-tools/main.go +++ b/modules/nats/cmd/nats-tools/main.go @@ -166,10 +166,10 @@ func main() { // bus holds against what was asked, and what no manifest asks for. Its acting verbs (create, apply, recreate, drop) are // the keeper's own work, on its own account alone (keeper.go), so no other account, the tool runner's included, holds // them. -var ReadingVerbs = []string{"list", "show", "check", "orphans", "aside"} +var ReadingVerbs = []string{"list", "show", "check", "orphans"} // KeptVerbs are the seat mesh-broker's verbs for a module's kept data (keptools.go): each handed to the keeper. -var KeptVerbs = []string{"snapshot-kept", "restore-kept", "discard-aside"} +var KeptVerbs = []string{"snapshot-kept", "restore-kept"} // 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). @@ -195,5 +195,5 @@ func allTools() []stdio.Tool { out = append(out, readTools(jsz, LiveMsgGet(LiveBus()))...) requests := RequestsInState() out = append(out, requestTools(requests)...) - return append(out, keptTools(requests, KeptInState)...) + return append(out, keptTools(requests)...) } diff --git a/modules/nats/internal/keeper/kept.go b/modules/nats/internal/keeper/kept.go index e4caf5a5e..945f71ae4 100644 --- a/modules/nats/internal/keeper/kept.go +++ b/modules/nats/internal/keeper/kept.go @@ -132,7 +132,7 @@ func (k *Keeper) keptReady(r *Request) error { if k.Kept == nil || k.KeptDir == "" { return errors.New("no directory for restore points is set (MESH_NATS_SNAPSHOTS_DIR), so no kept data is taken or restored") } - if !safeName(r.Module) || !safeName(r.Local) || !safeName(r.Fingerprint) { + if !safeName(r.Module) || !safeName(r.Local) || (!r.AsideOnly && !safeName(r.Fingerprint)) { return fmt.Errorf("%q, %q and %q are not a module, a kept data and a fingerprint: each is one plain name", r.Module, r.Local, r.Fingerprint) } return nil @@ -244,17 +244,30 @@ func (k *Keeper) RestoreKept(ctx context.Context, r *Request) (string, error) { if err := k.keptReady(r); err != nil { return "", err } - if !filepath.IsAbs(r.From) { - return "", fmt.Errorf("the restore point's directory is given as an absolute path, not %q", r.From) - } - p, a, err := readPoint(filepath.Clean(r.From), r) - if err != nil { - return "", err - } - defer a.Close() - live, err := k.Kept.Existing(ctx, p.Streams) - if err != nil { - return "", err + var p *Point + var a *busarchive.Archive + var live []string + if r.AsideOnly { + // `aside-only` (design 47 G §405): the live data is set aside and nothing is restored. + streams, err := k.streamsOf(ctx, r.Module, r.Local) + if err != nil { + return "", err + } + live = streams + } else { + if !filepath.IsAbs(r.From) { + return "", fmt.Errorf("the restore point's directory is given as an absolute path, not %q", r.From) + } + var err error + p, a, err = readPoint(filepath.Clean(r.From), r) + if err != nil { + return "", err + } + defer a.Close() + live, err = k.Kept.Existing(ctx, p.Streams) + if err != nil { + return "", err + } } id := fmt.Sprintf("%s.%s.%s", r.Module, r.Local, k.now().Format("20060102t150405")) var aside *Aside @@ -288,6 +301,16 @@ func (k *Keeper) RestoreKept(ctx context.Context, r *Request) (string, error) { return "", fmt.Errorf("the live data was set aside in %s and could not be removed from the bus: %w", dir, err) } } + if r.AsideOnly { + r.Dir = "" + where := "nothing was live, so nothing was set aside" + if aside != nil { + r.Dir = aside.Dir + where = fmt.Sprintf("the live data is set aside as %s (%d bytes) in %s", aside.ID, aside.Bytes, aside.Dir) + } + k.record(r, fmt.Sprintf("set aside %s's %s and restored nothing; %s", r.Module, r.Local, where)) + return fmt.Sprintf("set aside %s's %s; %s", r.Module, r.Local, where), nil + } if err := k.Kept.RestoreArchive(ctx, a); err != nil { msg := fmt.Sprintf("the restore point could not be put on the bus: %v", err) if aside != nil { diff --git a/modules/nats/internal/keeper/kept_test.go b/modules/nats/internal/keeper/kept_test.go index 2c937daa8..95e0d2b12 100644 --- a/modules/nats/internal/keeper/kept_test.go +++ b/modules/nats/internal/keeper/kept_test.go @@ -175,6 +175,19 @@ func TestAKeptDataIsBackedUpAndRestoredBesideTheLiveData(t *testing.T) { t.Errorf("hand acts %v", *acts) } }}, + {"aside-only sets the live data aside whole and restores nothing", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) { + r := &Request{Verb: VerbRestoreKept, Module: "idx", Local: "modules", AsideOnly: true, Caller: "ace.restic", Why: "newer build stopped"} + if _, err := k.RestoreKept(context.Background(), r); err != nil { + t.Fatal(err) + } + if kb.streams["KV_idx_modules"] || len(kb.restored) != 0 { + t.Errorf("streams %v restored %v: the live data is removed and nothing restored", kb.streams, kb.restored) + } + aside, err := os.ReadDir(filepath.Join(k.KeptDir, "aside")) + if err != nil || len(aside) != 1 || r.Dir == "" { + t.Fatalf("aside: %v %v dir %q", aside, err, r.Dir) + } + }}, {"a kept data the bus does not hold is refused naming what was looked for, and nothing is written", func(t *testing.T, k *Keeper, kb *keptBus, store *memStore, acts *[]string) { r := &Request{Module: "idx", Local: "gone", Fingerprint: "abc123"} _, err := k.SnapshotKept(context.Background(), r) diff --git a/modules/nats/internal/keeper/requests.go b/modules/nats/internal/keeper/requests.go index f1bac747a..b7993c83d 100644 --- a/modules/nats/internal/keeper/requests.go +++ b/modules/nats/internal/keeper/requests.go @@ -51,8 +51,10 @@ type Request struct { Form int `json:"form,omitempty"` Fingerprint string `json:"fingerprint,omitempty"` CatchUp string `json:"catch-up,omitempty"` - From string `json:"from,omitempty"` - Dir string `json:"dir,omitempty"` + // AsideOnly is the argument `aside-only` of restore-kept: the live data is set aside and nothing is restored. + AsideOnly bool `json:"aside-only,omitempty"` + From string `json:"from,omitempty"` + Dir string `json:"dir,omitempty"` // Unrecorded is why the hand act of what it did could not be recorded, when it could not: said, never only logged. Unrecorded string `json:"unrecorded,omitempty"` } diff --git a/modules/nats/module.json b/modules/nats/module.json index 070b1a112..f239c2e49 100644 --- a/modules/nats/module.json +++ b/modules/nats/module.json @@ -33,10 +33,8 @@ "show", "check", "orphans", - "aside", "snapshot-kept", - "restore-kept", - "discard-aside" + "restore-kept" ] } ], @@ -141,8 +139,7 @@ "nats_drop", "nats_keeper_health", "nats_first_view_again", - "nats_requests", - "nats_kept" + "nats_requests" ], "guards": [ 8222