From 810254193c3f71b8190cafab2a8b3e86a1fcb986 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 02:12:32 +0100 Subject: [PATCH] mesh-recovery: the hand verb restore takes a keeper's snapshot by request MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An unprotected snapshot is received through snapshot-get into incoming/, refused before the first chunk past the cap, checked against its digest before any aside, and restored by the module; a protected one is verified, the running data set aside and restored in place by the keeper, waited on by snapshot-restored within snapshot-restore-wait-minutes, cancelled at the bound. The bus module sends its chunks with mesh-sdk's form. Builds: design 47 G §681, §682, A-137, A-138, A-139, A-140 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-recovery/cmd/mesh-recovery/keeper.go | 151 +++++++++++++ .../mesh-recovery/cmd/mesh-recovery/main.go | 13 +- .../cmd/mesh-recovery/main_test.go | 23 ++ .../files/snapshot-restore-wait-minutes | 1 + .../internal/recovery/control.go | 2 + .../mesh-recovery/internal/recovery/keeper.go | 191 ++++++++++++++++ .../internal/recovery/keeper_test.go | 206 ++++++++++++++++++ .../internal/recovery/manifest_test.go | 8 +- .../internal/recovery/recovery.go | 53 +++-- .../internal/recovery/restore.go | 5 +- .../mesh-recovery/internal/recovery/verbs.go | 6 +- modules/mesh-recovery/module.json | 21 +- modules/nats/cmd/nats-tools/snapshots.go | 33 ++- 13 files changed, 666 insertions(+), 47 deletions(-) create mode 100644 modules/mesh-recovery/cmd/mesh-recovery/keeper.go create mode 100644 modules/mesh-recovery/files/snapshot-restore-wait-minutes create mode 100644 modules/mesh-recovery/internal/recovery/keeper.go create mode 100644 modules/mesh-recovery/internal/recovery/keeper_test.go diff --git a/modules/mesh-recovery/cmd/mesh-recovery/keeper.go b/modules/mesh-recovery/cmd/mesh-recovery/keeper.go new file mode 100644 index 000000000..1d60219c4 --- /dev/null +++ b/modules/mesh-recovery/cmd/mesh-recovery/keeper.go @@ -0,0 +1,151 @@ +package main + +import ( + "context" + "crypto/rand" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "sync" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + "git.novox.be/novox/mesh-sdk/go/buscalls" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "mesh-recovery/internal/recovery" +) + +// liveKeeper is the bus module's keeper reached by the seat mesh-broker's verbs (design 47 G §681): snapshot-get's chunks +// received in mesh-sdk's form (A-139), snapshot-restore and its cancel called, and snapshot-restored heard. +type liveKeeper struct { + nc *nats.Conn + call Call + restored *restoredView +} + +func (k liveKeeper) verb(verb string, in, out any) error { + raw, err := k.call("seat:"+meshbroker.Seat+"."+verb, in) + if err != nil { + return fmt.Errorf("%s's %s: %w", meshbroker.Seat, verb, err) + } + if err := json.Unmarshal(raw, out); err != nil { + return fmt.Errorf("%s's %s answered what does not read: %w", meshbroker.Seat, verb, err) + } + return nil +} + +// Get subscribes to the chunks of a call of its own before it asks snapshot-get, so none is missed, then receives them in +// order, acknowledging each (§681, A-139). +func (k liveKeeper) Get(ctx context.Context, request string, fits func(meshbroker.SnapshotGetAnswer) error, w io.Writer) (meshbroker.SnapshotGetAnswer, error) { + var bits [16]byte + if _, err := rand.Read(bits[:]); err != nil { + return meshbroker.SnapshotGetAnswer{}, err + } + send := buscalls.SeatChunks{Seat: meshbroker.Seat, Verb: meshbroker.VerbSnapshotGet, Call: hex.EncodeToString(bits[:])} + chunks := make(chan *nats.Msg, 4) + sub, err := k.nc.ChanSubscribe(send.Chunks(), chunks) + if err != nil { + return meshbroker.SnapshotGetAnswer{}, fmt.Errorf("the chunks of snapshot-get cannot be heard: %w", err) + } + defer func() { _ = sub.Unsubscribe() }() + var ans meshbroker.SnapshotGetAnswer + if err := k.verb(meshbroker.VerbSnapshotGet, meshbroker.SnapshotGetInput{Call: send.Call, Request: request}, &ans); err != nil { + return ans, err + } + if ans.Error != "" { + return ans, nil + } + if err := fits(ans); err != nil { + return ans, err + } + if ans.Size > 0 && ans.Chunks == 0 { + return ans, fmt.Errorf("the keeper answered the request %s's %d bytes with no chunk: this account is granted no chunk subjects of snapshot-get", request, ans.Size) + } + next := func() (string, []byte, error) { + bound := time.NewTimer(buscalls.SeatAckWithin) + defer bound.Stop() + select { + case m := <-chunks: + return m.Subject, m.Data, nil + case <-bound.C: + return "", nil, buscalls.ErrSeatChunkMissed + case <-ctx.Done(): + return "", nil, ctx.Err() + } + } + ack := func(subject string) error { + _, err := k.nc.Request(subject, []byte("{}"), buscalls.SeatAckWithin) + return err + } + if _, err := send.Receive(ans.Chunks, next, ack, w); err != nil { + return ans, fmt.Errorf("the snapshot of the request %s: %w", request, err) + } + return ans, nil +} + +func (k liveKeeper) Restore(_ context.Context, request string, verify bool, why string) (meshbroker.SnapshotRestoreAnswer, error) { + var ans meshbroker.SnapshotRestoreAnswer + err := k.verb(meshbroker.VerbSnapshotRestore, meshbroker.SnapshotRestoreInput{Request: request, Verify: verify, Why: why}, &ans) + return ans, err +} + +func (k liveKeeper) CancelRestore(_ context.Context, act, why string) (meshbroker.SnapshotRestoreCancelAnswer, error) { + var ans meshbroker.SnapshotRestoreCancelAnswer + err := k.verb(meshbroker.VerbSnapshotRestoreCancel, meshbroker.SnapshotRestoreCancelInput{Act: act, Why: why}, &ans) + return ans, err +} + +func (k liveKeeper) Restored(ctx context.Context, request, act string) (meshbroker.SnapshotRestoredEvent, bool) { + return k.restored.wait(ctx, request, act) +} + +var _ recovery.Keeper = liveKeeper{} + +// restoredView is the keeper's snapshot-restored as heard, each request and act kept (§681): read as §556 has it, the last +// of each subject first, so an outcome said before a wait began wakes it too. +type restoredView struct { + mu sync.Mutex + by map[string]meshbroker.SnapshotRestoredEvent + changed chan struct{} +} + +func newRestoredView() *restoredView { + return &restoredView{by: map[string]meshbroker.SnapshotRestoredEvent{}, changed: make(chan struct{})} +} + +func (v *restoredView) hear(m jetstream.Msg) { + var ev meshbroker.SnapshotRestoredEvent + if err := json.Unmarshal(m.Data(), &ev); err == nil && ev.Request != "" && ev.Act != "" { + v.take(ev) + } + _ = m.Ack() +} + +func (v *restoredView) take(ev meshbroker.SnapshotRestoredEvent) { + v.mu.Lock() + v.by[ev.Request+"\x00"+ev.Act] = ev + close(v.changed) + v.changed = make(chan struct{}) + v.mu.Unlock() +} + +func (v *restoredView) wait(ctx context.Context, request, act string) (meshbroker.SnapshotRestoredEvent, bool) { + for { + v.mu.Lock() + ev, ok := v.by[request+"\x00"+act] + changed := v.changed + v.mu.Unlock() + if ok { + return ev, true + } + select { + case <-changed: + case <-ctx.Done(): + return meshbroker.SnapshotRestoredEvent{}, false + } + } +} diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main.go b/modules/mesh-recovery/cmd/mesh-recovery/main.go index 8f48f11a5..3e2f748de 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main.go @@ -35,6 +35,7 @@ const ( envKeepDays = "MESH_RECOVERY_KEEP_DAYS_FILE" envMark = "MESH_RECOVERY_FILLING_MARK_FILE" envDumpWait = "MESH_RECOVERY_DUMP_WAIT_FILE" + envRestore = "MESH_RECOVERY_RESTORE_WAIT_FILE" ) // The settings' defaults: the design's values (§551, §564). @@ -96,6 +97,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + restoreWait, err := readMinutes(getenv(envRestore), recovery.DefaultRestoreWait) + if err != nil { + return err + } account, err := process.ReadBrokerAccount(getenv) if err != nil { return err @@ -123,6 +128,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro return err } } + // The keeper's snapshot-restored, each request and act, for a restore in place (§681). + restored := newRestoredView() + if _, err := account.HearSeatEvents(ctx, js, meshbroker.StreamSnapshotRestored, restored.hear, logger.Printf); err != nil { + return err + } // The module's findings go through mesh-sdk's held writer over its own state findings (§561), started before the // engine's start, so what the start writes is held and written. writer, err := views.NewHeldWriter(views.WriterConfig[meshconditions.Finding, meshconditions.FindingCleared, meshconditions.FindingsStated, meshconditions.FindingsAskedEvent]{ @@ -143,7 +153,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro node: account.Node, points: dir, running: running}, Digests: digests{conn.State("digests")}, Dump: dump, - Settings: recovery.Settings{Cap: cap, KeepDays: int(keepDays), FillingMark: int(mark), DumpWait: dumpWait}, + Keeper: liveKeeper{nc: nc, call: conn.Call, restored: restored}, + Settings: recovery.Settings{Cap: cap, KeepDays: int(keepDays), FillingMark: int(mark), DumpWait: dumpWait, RestoreWait: restoreWait}, Findings: writer, Node: account.Node, Logf: logger.Printf, diff --git a/modules/mesh-recovery/cmd/mesh-recovery/main_test.go b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go index 7f1d18dce..a16eede55 100644 --- a/modules/mesh-recovery/cmd/mesh-recovery/main_test.go +++ b/modules/mesh-recovery/cmd/mesh-recovery/main_test.go @@ -1,10 +1,12 @@ package main import ( + "context" "strings" "testing" "time" + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/process" @@ -48,6 +50,27 @@ func TestWhatTheProcessReads(t *testing.T) { t.Errorf("two fingerprints answered %v", err) } }) + t.Run("the keeper's snapshot-restored of a request and act (§681)", func(t *testing.T) { + v := newRestoredView() + v.take(meshbroker.SnapshotRestoredEvent{Request: "req-1", Act: "act-0", Outcome: "failed"}) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, ok := v.wait(ctx, "req-1", "act-1"); ok { + t.Error("another act's outcome woke the wait") + } + woken := make(chan meshbroker.SnapshotRestoredEvent, 1) + go func() { + ev, _ := v.wait(context.Background(), "req-1", "act-1") + woken <- ev + }() + v.take(meshbroker.SnapshotRestoredEvent{Request: "req-1", Act: "act-1", Outcome: "done"}) + if ev := <-woken; ev.Outcome != "done" { + t.Errorf("woken by %+v", ev) + } + if ev, ok := v.wait(ctx, "req-1", "act-1"); !ok || ev.Outcome != "done" { + t.Errorf("an outcome said before the wait is not taken: %+v %v", ev, ok) + } + }) t.Run("a dump of the bus module runs (§563)", func(t *testing.T) { d := newDumpView(nil, nil) start := time.Date(2026, 11, 12, 2, 0, 0, 0, time.UTC) diff --git a/modules/mesh-recovery/files/snapshot-restore-wait-minutes b/modules/mesh-recovery/files/snapshot-restore-wait-minutes new file mode 100644 index 000000000..89cd4bdf5 --- /dev/null +++ b/modules/mesh-recovery/files/snapshot-restore-wait-minutes @@ -0,0 +1 @@ +${config:snapshot-restore-wait-minutes} diff --git a/modules/mesh-recovery/internal/recovery/control.go b/modules/mesh-recovery/internal/recovery/control.go index 941b22ccf..ef9ad13ea 100644 --- a/modules/mesh-recovery/internal/recovery/control.go +++ b/modules/mesh-recovery/internal/recovery/control.go @@ -134,6 +134,8 @@ func (e *Engine) Start(ctx context.Context) []UnrestoredAside { } return nil }) + // A keeper's snapshot a cut-off act left half or whole received is deleted (§681). + _ = os.RemoveAll(e.incoming()) _, recs, _ := e.Asides() for i := range recs { r := &recs[i] diff --git a/modules/mesh-recovery/internal/recovery/keeper.go b/modules/mesh-recovery/internal/recovery/keeper.go new file mode 100644 index 000000000..5c8167caa --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/keeper.go @@ -0,0 +1,191 @@ +package recovery + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" +) + +// Keeper is the bus module's keeper as the seat mesh-broker answers it (design 47 G §681): its snapshot taken before a +// `drop` or `recreate`, sent by snapshot-get, or restored in place by snapshot-restore when it holds a protected stream. +type Keeper interface { + // Get asks snapshot-get of the request; when its answer names a file and fits accepts it, it receives the chunks into + // w. A refusal of the keeper is the answer's Error, no error of Get. + Get(ctx context.Context, request string, fits func(meshbroker.SnapshotGetAnswer) error, w io.Writer) (meshbroker.SnapshotGetAnswer, error) + // Restore asks snapshot-restore; with verify the keeper checks the file and restores nothing. + Restore(ctx context.Context, request string, verify bool, why string) (meshbroker.SnapshotRestoreAnswer, error) + // CancelRestore asks snapshot-restore-cancel of a running restore. + CancelRestore(ctx context.Context, act, why string) (meshbroker.SnapshotRestoreCancelAnswer, error) + // Restored answers the keeper's snapshot-restored of that request and act once heard (read as §556 has it, so one said + // before the wait counts), or ok false when ctx ends first. + Restored(ctx context.Context, request, act string) (ev meshbroker.SnapshotRestoredEvent, ok bool) +} + +// The keeper's word for a snapshot it never hands out (§681). +const snapshotProtected = "snapshot protected" + +// ErrUnanswered is the end of a restore in place the keeper answered neither by its outcome nor by the cancel's (§681). +const ErrUnanswered = "snapshot-restore unanswered" + +// DefaultRestoreWait is how long a restore in place is waited on: past the keeper's own 60 minutes (§681, A-138). +const DefaultRestoreWait = 90 * time.Minute + +func (e *Engine) restoreWait() time.Duration { + if e.Settings.RestoreWait > 0 { + return e.Settings.RestoreWait + } + return DefaultRestoreWait +} + +func (e *Engine) incoming() string { return filepath.Join(e.Dir, "incoming") } + +// restoreRequest is the hand verb `restore` with `request`, a keeper's snapshot (§681 Its use, §682): an unprotected one is +// read through snapshot-get into incoming/, refused before the first chunk when it would pass the cap, checked +// against its digest before any aside, and restored by the module itself as a point is (§516), the file deleted when the +// act ends; a protected one is restored in place by the keeper (restoreInPlace). +func (e *Engine) restoreRequest(ctx context.Context, a *Act) error { + if e.Keeper == nil { + return e.endedRestored(a, errors.New("no keeper is reachable, so no snapshot of it is restored"), nil, false) + } + e.setStep(a, "receiving the keeper's snapshot", "") + name, err := token(a.Request) + if err != nil { + return e.endedRestored(a, err, nil, false) + } + if err := os.MkdirAll(e.incoming(), 0o700); err != nil { + return e.endedRestored(a, err, nil, false) + } + file := filepath.Join(e.incoming(), name) + defer os.Remove(file) + f, err := os.Create(file) + if err != nil { + return e.endedRestored(a, err, nil, false) + } + ans, err := e.Keeper.Get(ctx, a.Request, func(ans meshbroker.SnapshotGetAnswer) error { + if fill := e.filled(); fill+ans.Size > e.Settings.Cap { + return fmt.Errorf("%s: the cap is %d bytes, the snapshot of the request %s needs %d and the directory holds %d", + ErrCap, e.Settings.Cap, a.Request, ans.Size, fill) + } + return nil + }, f) + if cerr := f.Close(); err == nil { + err = cerr + } + if err != nil { + return e.endedRestored(a, err, nil, false) + } + if ans.Error == snapshotProtected { + _ = os.Remove(file) + return e.restoreInPlace(ctx, a) + } + if ans.Error != "" { + return e.endedRestored(a, errors.New(ans.Error), nil, false) + } + // The bytes are checked against the keeper's digest before any aside is taken (§567). + if got, n, err := digestOf(file); err != nil || got != ans.Digest || n != ans.Size { + return e.endedRestored(a, fmt.Errorf("snapshot damaged: the request %s's bytes received do not match its digest", a.Request), nil, false) + } + rec, err := e.setAside(ctx, a, []string{ans.Stream}, true) + if err != nil { + return e.endedRestored(a, err, nil, false) + } + e.setStep(a, "restoring", "") + if err := e.Bus.Restore(ctx, file); err != nil { + err = fmt.Errorf("stream %s: %w", ans.Stream, err) + if back := e.restoreFromAside(ctx, rec, true); back != nil { + err = fmt.Errorf("%w; and the aside is not put back: %v", err, back) + } + return e.endedRestored(a, err, asidePtr(rec), false) + } + rec.Step = stepRestored + _ = writeJSON(filepath.Join(e.asideDirOf(rec), asideFile), rec) + return e.endedRestored(a, nil, asidePtr(rec), false) +} + +// restoreInPlace is a protected snapshot's restore (§681, O-7): snapshot-restore with verify first, then the live data of +// the restore's module and kept data set aside, its names from the catalog entry of the fingerprint the node runs +// (A-137), then snapshot-restore, waited on by snapshot-restored of that request and the act it was answered for at most +// snapshot-restore-wait-minutes (A-138). Its `failed` puts the aside back. At the bound it asks snapshot-restore-cancel and +// waits one more bound for that act's outcome (A-140); when none comes it puts nothing back and ends failed with +// `snapshot-restore unanswered`, the aside left at `removed` for aside-unrestored (§561). A keeper answering busy ends the +// act failed before any aside, naming the keeper's running act (A-140). +func (e *Engine) restoreInPlace(ctx context.Context, a *Act) error { + e.setStep(a, "verifying the keeper's snapshot", "") + v, err := e.Keeper.Restore(ctx, a.Request, true, a.Why) + if err == nil { + err = keeperAnswer(v) + } + if err != nil { + return e.endedRestored(a, err, nil, false) + } + fp, err := e.Seats.RunningFingerprint(ctx, a.Module) + if err != nil { + return e.endedRestored(a, err, nil, false) + } + a.Fingerprint = fp + names, err := e.Seats.Streams(ctx, a.Module, a.Kept, fp) + if err != nil { + return e.endedRestored(a, err, nil, false) + } + rec, err := e.setAside(ctx, a, names, true) + if err != nil { + return e.endedRestored(a, err, nil, false) + } + putBack := func(why error) error { + if back := e.restoreFromAside(ctx, rec, true); back != nil { + why = fmt.Errorf("%w; and the aside is not put back: %v", why, back) + } + return e.endedRestored(a, why, asidePtr(rec), false) + } + e.setStep(a, "restoring in place", "the keeper's snapshot-restored") + r, err := e.Keeper.Restore(ctx, a.Request, false, a.Why) + if err == nil { + err = keeperAnswer(r) + } + if err != nil { + return putBack(err) + } + act := r.Accepted + ev, ok := e.restoredWithin(ctx, a.Request, act) + if !ok { + e.setStep(a, "cancelling the keeper's restore", "the keeper's snapshot-restored after the cancel") + why := fmt.Sprintf("the recovery module's wait of %s on the restore in place passed (act %s)", e.restoreWait(), a.ID) + if c, err := e.Keeper.CancelRestore(context.Background(), act, why); err != nil || c.Error != "" { + e.logf("the cancel of the keeper's restore %s is not accepted (%v%s); its outcome is waited on once more", act, err, c.Error) + } + if ev, ok = e.restoredWithin(context.Background(), a.Request, act); !ok { + return e.endedRestored(a, errors.New(ErrUnanswered), asidePtr(rec), false) + } + } + if ev.Outcome != meshbroker.SnapshotRestoredEventOutcomeDone { + return putBack(fmt.Errorf("the keeper's restore in place failed: %s", ev.Error)) + } + rec.Step = stepRestored + _ = writeJSON(filepath.Join(e.asideDirOf(rec), asideFile), rec) + return e.endedRestored(a, nil, asidePtr(rec), false) +} + +func (e *Engine) restoredWithin(ctx context.Context, request, act string) (meshbroker.SnapshotRestoredEvent, bool) { + wctx, cancel := context.WithTimeout(ctx, e.restoreWait()) + defer cancel() + return e.Keeper.Restored(wctx, request, act) +} + +// keeperAnswer reads snapshot-restore's one form (§691): accepted, or an error naming the keeper's refusal or its running act. +func keeperAnswer(r meshbroker.SnapshotRestoreAnswer) error { + switch { + case r.Error != "": + return errors.New(r.Error) + case r.Busy != "": + return fmt.Errorf("busy: the keeper restores its act %s", r.Busy) + case r.Accepted == "": + return errors.New("the keeper answered snapshot-restore with nothing") + } + return nil +} diff --git a/modules/mesh-recovery/internal/recovery/keeper_test.go b/modules/mesh-recovery/internal/recovery/keeper_test.go new file mode 100644 index 000000000..5084e2b76 --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/keeper_test.go @@ -0,0 +1,206 @@ +package recovery_test + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "io" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + + "mesh-recovery/internal/recovery" +) + +// fakeKeeper is the bus module's keeper holding one snapshot of app_notes (the fake bus's file form). in place: how its +// restore in place ends: "done", "failed", "silent" (it says nothing until cancelled, then failed) or "deaf" (never). +type fakeKeeper struct { + mu sync.Mutex + b *fakeBus + file string + protected bool + busy string + damaged bool + inPlace string + asked []string + wrote int + cancelled chan struct{} +} + +func newKeeper(b *fakeBus) *fakeKeeper { + return &fakeKeeper{b: b, file: "snap:app_notes:as dropped", inPlace: "done", cancelled: make(chan struct{})} +} + +func (k *fakeKeeper) note(s string) { + k.mu.Lock() + k.asked = append(k.asked, s) + k.mu.Unlock() +} + +func (k *fakeKeeper) Get(_ context.Context, request string, fits func(meshbroker.SnapshotGetAnswer) error, w io.Writer) (meshbroker.SnapshotGetAnswer, error) { + k.note("get " + request) + if k.protected { + return meshbroker.SnapshotGetAnswer{Error: "snapshot protected"}, nil + } + sum := sha256.Sum256([]byte(k.file)) + ans := meshbroker.SnapshotGetAnswer{Request: request, Module: "app", Local: "notes", Stream: "app_notes", Size: int64(len(k.file)), + Digest: "sha256:" + hex.EncodeToString(sum[:]), Chunks: 1} + if err := fits(ans); err != nil { + return ans, err + } + body := k.file + if k.damaged { + body = "snap:app_notes:altered on its way" + } + n, err := io.WriteString(w, body) + k.wrote += n + return ans, err +} + +func (k *fakeKeeper) Restore(_ context.Context, request string, verify bool, _ string) (meshbroker.SnapshotRestoreAnswer, error) { + if verify { + k.note("verify " + request) + return meshbroker.SnapshotRestoreAnswer{Accepted: request}, nil + } + k.note("restore " + request) + if k.busy != "" { + return meshbroker.SnapshotRestoreAnswer{Busy: k.busy}, nil + } + return meshbroker.SnapshotRestoreAnswer{Accepted: "keeper-act-1"}, nil +} + +func (k *fakeKeeper) CancelRestore(_ context.Context, act, _ string) (meshbroker.SnapshotRestoreCancelAnswer, error) { + k.note("cancel " + act) + close(k.cancelled) + return meshbroker.SnapshotRestoreCancelAnswer{Accepted: act}, nil +} + +func (k *fakeKeeper) Restored(ctx context.Context, request, act string) (meshbroker.SnapshotRestoredEvent, bool) { + ev := meshbroker.SnapshotRestoredEvent{Request: request, Act: act, Outcome: "failed", Error: "the server refused the archive"} + switch k.inPlace { + case "done": + k.b.mu.Lock() + if _, live := k.b.streams["app_notes"]; live { + k.b.mu.Unlock() + return meshbroker.SnapshotRestoredEvent{Request: request, Act: act, Outcome: "failed", Error: "names live"}, true + } + k.b.streams["app_notes"] = "as dropped" + k.b.mu.Unlock() + ev.Outcome, ev.Error = "done", "" + return ev, true + case "failed": + return ev, true + case "silent": + select { + case <-k.cancelled: + ev.Error = "cancelled" + return ev, true + case <-ctx.Done(): + return ev, false + } + } + <-ctx.Done() + return ev, false +} + +func requestRig(t *testing.T) (*rig, *fakeKeeper) { + r := newRig(t) + k := newKeeper(r.b) + r.e.Keeper = k + r.e.Settings.RestoreWait = 20 * time.Millisecond + return r, k +} + +func requestAct() recovery.Act { + return recovery.Act{Kind: recovery.KindRestore, Module: "app", Kept: "notes", Request: "req-1", Why: "test the drop's snapshot", Caller: "tool-runner", Asker: "op@node"} +} + +// A keeper's snapshot by request is read through snapshot-get into incoming/, refused before the first chunk past +// the cap, checked against its digest before any aside, restored by the module itself, and its file deleted at the end +// (design 47 G §681 Its use, §682). +func TestARestoreByRequestReadsTheKeepersSnapshot(t *testing.T) { + for _, tc := range []struct { + name string + cap int64 + damaged bool + err string + }{ + {name: "received and restored", cap: 1 << 20}, + {name: "altered on its way", cap: 1 << 20, damaged: true, err: "snapshot damaged"}, + {name: "past the cap", cap: 10, err: "cap:"}, + } { + t.Run(tc.name, func(t *testing.T) { + r, k := requestRig(t) + r.e.Settings.Cap, k.damaged = tc.cap, tc.damaged + ans := r.do(requestAct()) + ev := lastRestored(t, r) + if ev.Act != ans.Accepted || !strings.HasPrefix(ev.Error, tc.err) || (tc.err == "") != (ev.Outcome == "done") { + t.Fatalf("restored %+v, want error %q", ev, tc.err) + } + if left, _ := os.ReadDir(filepath.Join(r.e.Dir, "incoming")); len(left) != 0 { + t.Errorf("incoming holds %v after the act", left) + } + kp, _ := r.e.KeptPoints("app", "notes") + if tc.err != "" { + if r.b.streams["app_notes"] != "hello" || len(kp.Asides) != 0 { + t.Errorf("a refused restore touched the live data: %v, asides %+v", r.b.streams, kp.Asides) + } + if tc.name == "past the cap" && k.wrote != 0 { + t.Errorf("%d bytes received past the cap", k.wrote) + } + return + } + if r.b.streams["app_notes"] != "as dropped" || r.b.streams["KV_app_seen"] != "world" || len(kp.Asides) != 1 || + ev.Aside == nil || len(ev.Aside.Names) != 1 || ev.Aside.Names[0] != "app_notes" { + t.Errorf("streams %v, asides %+v, event aside %+v", r.b.streams, kp.Asides, ev.Aside) + } + }) + } +} + +// A protected snapshot is verified, the running data of the module and kept data set aside (A-137), and restored in place +// by the keeper; its failure puts the aside back; at the bound the module cancels first and puts the aside back on the +// cancel's outcome, and with no outcome at all ends `snapshot-restore unanswered`, the aside left unrestored; a busy keeper +// ends the act before any aside (§681, A-138, A-140). +func TestAProtectedSnapshotIsRestoredInPlaceByTheKeeper(t *testing.T) { + for _, tc := range []struct { + inPlace, busy string + err string + asked string + live string + unrestored int + }{ + {inPlace: "done", asked: "get req-1|verify req-1|restore req-1", live: "as dropped"}, + {inPlace: "failed", asked: "get req-1|verify req-1|restore req-1", live: "hello", err: "the keeper's restore in place failed"}, + {inPlace: "silent", asked: "get req-1|verify req-1|restore req-1|cancel keeper-act-1", live: "hello", err: "the keeper's restore in place failed: cancelled"}, + {inPlace: "deaf", asked: "get req-1|verify req-1|restore req-1|cancel keeper-act-1", err: recovery.ErrUnanswered, unrestored: 1}, + {inPlace: "done", busy: "keeper-act-0", asked: "get req-1|verify req-1|restore req-1", live: "hello", err: "busy: the keeper restores its act keeper-act-0"}, + } { + t.Run(tc.inPlace+tc.busy, func(t *testing.T) { + r, k := requestRig(t) + k.protected, k.inPlace, k.busy = true, tc.inPlace, tc.busy + r.do(requestAct()) + ev := lastRestored(t, r) + if !strings.HasPrefix(ev.Error, tc.err) || (tc.err == "") != (ev.Outcome == "done") || ev.Fingerprint != "fp-run" { + t.Fatalf("restored %+v, want error %q", ev, tc.err) + } + if got := strings.Join(k.asked, "|"); got != tc.asked { + t.Errorf("asked %s, want %s", got, tc.asked) + } + if r.b.streams["app_notes"] != tc.live { + t.Errorf("app_notes is %q, want %q", r.b.streams["app_notes"], tc.live) + } + if tc.live == "hello" && r.b.streams["KV_app_seen"] != "world" { + t.Errorf("the second name of the running data is not put back: %v", r.b.streams) + } + if n := len(r.e.Unrestored()); n != tc.unrestored { + t.Errorf("%d unrestored asides, want %d", n, tc.unrestored) + } + }) + } +} diff --git a/modules/mesh-recovery/internal/recovery/manifest_test.go b/modules/mesh-recovery/internal/recovery/manifest_test.go index b68c808b9..7e904fa1b 100644 --- a/modules/mesh-recovery/internal/recovery/manifest_test.go +++ b/modules/mesh-recovery/internal/recovery/manifest_test.go @@ -70,11 +70,17 @@ func TestTheManifestSaysWhatTheModuleIs(t *testing.T) { t.Errorf("no data item %q", id) } } - for k, want := range map[string]float64{"points-cap": 10 << 30, "points-filling-mark": 80, "point-keep-days": 90, "dump-wait-minutes": 60} { + for k, want := range map[string]float64{"points-cap": 10 << 30, "points-filling-mark": 80, "point-keep-days": 90, "dump-wait-minutes": 60, "snapshot-restore-wait-minutes": 90} { if m.Config[k].Default != want { t.Errorf("setting %s defaults to %v, want %v", k, m.Config[k].Default, want) } } + // The keeper's snapshot verbs the recovery module asks, the cancel at its bound among them (§681). + for _, inv := range []string{"seat:mesh-broker.snapshot-get", "seat:mesh-broker.snapshot-restore", "seat:mesh-broker.snapshot-restore-cancel"} { + if !slices.Contains(m.Invokes, inv) { + t.Errorf("does not invoke %s", inv) + } + } for _, r := range m.Resources { if r.Type != "process" { continue diff --git a/modules/mesh-recovery/internal/recovery/recovery.go b/modules/mesh-recovery/internal/recovery/recovery.go index f37c7b704..2e1f09076 100644 --- a/modules/mesh-recovery/internal/recovery/recovery.go +++ b/modules/mesh-recovery/internal/recovery/recovery.go @@ -104,15 +104,20 @@ type Settings struct { FillingMark int // DumpWait is how long an act waits on the dump before recovery-waits is written (dump-wait-minutes, 60 unless set). DumpWait time.Duration + // RestoreWait is how long a restore in place is waited on, and once more after its cancel (snapshot-restore-wait-minutes, + // 90 unless set; §681, A-138, A-140). + RestoreWait time.Duration } // Engine is the recovery module's one engine. type Engine struct { - Dir string - Bus Bus - Seats Seats - Digests Digests - Dump Dump + Dir string + Bus Bus + Seats Seats + Digests Digests + Dump Dump + // Keeper is the bus module's keeper, whose snapshots the hand verb restore takes by request (§681). + Keeper Keeper Settings Settings // Findings is the module's finding writer (§561); Node the machine it runs on, the id of points-filling. Findings Findings @@ -130,24 +135,26 @@ type Engine struct { // Act is one accepted act, written in the directory before its caller is answered (§554). type Act struct { - ID string `json:"act"` - Kind string `json:"kind"` - Module string `json:"module,omitempty"` - Kept string `json:"kept,omitempty"` - Form int64 `json:"form,omitempty"` - Fingerprint string `json:"fingerprint,omitempty"` - Snapshot string `json:"snapshot,omitempty"` - AsideID string `json:"aside-id,omitempty"` - FromBackup bool `json:"from-backup,omitempty"` - CatchUp *seat.CatchUp `json:"catch-up,omitempty"` - Why string `json:"why"` - Caller string `json:"caller"` - Asker string `json:"asker,omitempty"` - Since time.Time `json:"since"` - Step string `json:"step,omitempty"` - WaitsOn string `json:"waits-on,omitempty"` - Ended bool `json:"ended,omitempty"` - Error string `json:"error,omitempty"` + ID string `json:"act"` + Kind string `json:"kind"` + Module string `json:"module,omitempty"` + Kept string `json:"kept,omitempty"` + Form int64 `json:"form,omitempty"` + Fingerprint string `json:"fingerprint,omitempty"` + Snapshot string `json:"snapshot,omitempty"` + // Request is a keeper's snapshot by its request, the hand verb restore's other source (§681, §682). + Request string `json:"request,omitempty"` + AsideID string `json:"aside-id,omitempty"` + FromBackup bool `json:"from-backup,omitempty"` + CatchUp *seat.CatchUp `json:"catch-up,omitempty"` + Why string `json:"why"` + Caller string `json:"caller"` + Asker string `json:"asker,omitempty"` + Since time.Time `json:"since"` + Step string `json:"step,omitempty"` + WaitsOn string `json:"waits-on,omitempty"` + Ended bool `json:"ended,omitempty"` + Error string `json:"error,omitempty"` } // Answer is the one form every acting verb answers (§691): exactly one field set. diff --git a/modules/mesh-recovery/internal/recovery/restore.go b/modules/mesh-recovery/internal/recovery/restore.go index 4fdbdfd20..c6936f758 100644 --- a/modules/mesh-recovery/internal/recovery/restore.go +++ b/modules/mesh-recovery/internal/recovery/restore.go @@ -70,7 +70,7 @@ func (e *Engine) cancelledWhy(err error) string { return "" } -// restore is `restore-kept` and the hand verb `restore` with a point (§516, §554, §563, §567): it says restore-started, +// restore is `restore-kept` and the hand verb `restore` with a point (or, past the dump, a keeper's snapshot by request) (§516, §554, §563, §567): it says restore-started, // waits for the dump of the bus's item, finds the point (or has the backup holder bring it back), checks its digests // against the bus, sets the live data aside whole and checked, removes it, restores the point under its names and // sets each consumer to its recorded sequence; a step after the removal that fails puts the aside back. @@ -91,6 +91,9 @@ func (e *Engine) restore(ctx context.Context, a *Act) error { return e.endedRestored(a, err, nil, false) } } + if a.Request != "" { + return e.restoreRequest(ctx, a) + } e.setStep(a, "finding the point", "") pointPath := filepath.Join(e.Dir, "points", filepath.FromSlash(a.Snapshot)) if _, err := os.Stat(filepath.Join(pointPath, pointFile)); err != nil { diff --git a/modules/mesh-recovery/internal/recovery/verbs.go b/modules/mesh-recovery/internal/recovery/verbs.go index 105868eb7..34cbdf7ec 100644 --- a/modules/mesh-recovery/internal/recovery/verbs.go +++ b/modules/mesh-recovery/internal/recovery/verbs.go @@ -46,10 +46,10 @@ func (e *Engine) Handlers() map[string]process.SeatHandler { if (rs.Snapshot == "") == (rs.Request == "") { return Act{}, errors.New("restore takes exactly one of snapshot and request") } - if rs.Request != "" { - return Act{}, errors.New("the restore of a keeper's snapshot by request is not built yet (design 47 G §681)") + if rs.Request != "" && rs.FromBackup { + return Act{}, errors.New("from-backup is taken only with snapshot, a restore point (design 47 G §682)") } - return Act{Kind: KindRestore, Module: rs.Module, Kept: rs.Kept, Snapshot: rs.Snapshot, FromBackup: rs.FromBackup, Why: rs.Why, Asker: rs.Asker}, nil + return Act{Kind: KindRestore, Module: rs.Module, Kept: rs.Kept, Snapshot: rs.Snapshot, Request: rs.Request, FromBackup: rs.FromBackup, Why: rs.Why, Asker: rs.Asker}, nil }) h[seat.VerbAside] = actOf(e, func(ad seat.AsideInput) (Act, error) { return Act{Kind: KindAside, Module: ad.Module, Kept: ad.Kept, Why: ad.Why, Asker: ad.Asker}, nil diff --git a/modules/mesh-recovery/module.json b/modules/mesh-recovery/module.json index 4eef18760..3f1e6e024 100644 --- a/modules/mesh-recovery/module.json +++ b/modules/mesh-recovery/module.json @@ -45,6 +45,7 @@ "seat:mesh-hand-acts.hand-act", "seat:mesh-broker.snapshot-get", "seat:mesh-broker.snapshot-restore", + "seat:mesh-broker.snapshot-restore-cancel", "seat:node-backup.restore", "seat:mesh-recovery.kept-points", "seat:mesh-recovery.restore-aside", @@ -113,6 +114,14 @@ "maximum": 10080, "trusted": false, "default": 60 + }, + "snapshot-restore-wait-minutes": { + "type": "number", + "description": "minutes a restore in place of a protected keeper's snapshot is waited on, and once more after its cancel (design 47 G section 681, A-138, A-140).", + "minimum": 1, + "maximum": 10080, + "trusted": false, + "default": 90 } }, "data": { @@ -185,6 +194,14 @@ "mode": "0600", "template": "files/dump-wait-minutes" }, + { + "id": "restore-wait", + "type": "file", + "path": "${dir:state}/snapshot-restore-wait-minutes", + "owner": "mesh-recovery", + "mode": "0600", + "template": "files/snapshot-restore-wait-minutes" + }, { "id": "recovery", "type": "process", @@ -199,7 +216,8 @@ "MESH_RECOVERY_POINTS_CAP_FILE": "${dir:state}/points-cap", "MESH_RECOVERY_FILLING_MARK_FILE": "${dir:state}/points-filling-mark", "MESH_RECOVERY_KEEP_DAYS_FILE": "${dir:state}/point-keep-days", - "MESH_RECOVERY_DUMP_WAIT_FILE": "${dir:state}/dump-wait-minutes" + "MESH_RECOVERY_DUMP_WAIT_FILE": "${dir:state}/dump-wait-minutes", + "MESH_RECOVERY_RESTORE_WAIT_FILE": "${dir:state}/snapshot-restore-wait-minutes" }, "user": "mesh-recovery", "serves": [ @@ -222,6 +240,7 @@ "seat:mesh-hand-acts.hand-act", "seat:mesh-broker.snapshot-get", "seat:mesh-broker.snapshot-restore", + "seat:mesh-broker.snapshot-restore-cancel", "seat:node-backup.restore" ] }, diff --git a/modules/nats/cmd/nats-tools/snapshots.go b/modules/nats/cmd/nats-tools/snapshots.go index 01f2e03f8..a1d455588 100644 --- a/modules/nats/cmd/nats-tools/snapshots.go +++ b/modules/nats/cmd/nats-tools/snapshots.go @@ -19,6 +19,7 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/buscalls" "git.novox.be/novox/mesh-sdk/go/process" "nats-tools/internal/keeper" @@ -30,34 +31,32 @@ import ( // 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 - -// AckWithin is how long the next chunk waits for its acknowledgement. -const AckWithin = 30 * time.Second +// The form of the chunks is mesh-sdk's (buscalls.SeatChunks, A-139): the caller receives with the same one. +const ( + // ChunkSize is at most what one chunk carries: 512 KiB. + ChunkSize = buscalls.SeatChunkBytes + // AckWithin is how long the next chunk waits for its acknowledgement. + AckWithin = buscalls.SeatAckWithin +) // callToken is the shape of the subject token a caller chooses for a send. var callToken = regexp.MustCompile(`^[A-Za-z0-9_-]{16,64}$`) -// ChunkSubject and AckSubject are the subjects of the nth chunk of a send and of its acknowledgement. -func ChunkSubject(call string, n int64) string { - return "mesh.seat." + meshbroker.Seat + ".chunk." + meshbroker.VerbSnapshotGet + "." + call + "." + strconv.FormatInt(n, 10) +func sendOf(call string) buscalls.SeatChunks { + return buscalls.SeatChunks{Seat: meshbroker.Seat, Verb: meshbroker.VerbSnapshotGet, Call: call} } +// ChunkSubject and AckSubject are the subjects of the nth chunk of a send and of its acknowledgement. +func ChunkSubject(call string, n int64) string { return sendOf(call).Chunk(n) } + // AckSubject is the subject the caller acknowledges the nth chunk on. -func AckSubject(call string, n int64) string { - return "mesh.seat." + meshbroker.Seat + ".ack." + meshbroker.VerbSnapshotGet + "." + call + "." + strconv.FormatInt(n, 10) -} +func AckSubject(call string, n int64) string { return sendOf(call).Ack(n) } // AckPattern is every acknowledgement of snapshot-get. -const AckPattern = "mesh.seat.mesh-broker.ack.snapshot-get.*.*" +var AckPattern = buscalls.SeatAcks(meshbroker.Seat, meshbroker.VerbSnapshotGet) // Chunk is the body of one chunk: its number, how many there are, and its bytes. -type Chunk struct { - N int64 `json:"n"` - Of int64 `json:"of"` - Data []byte `json:"data"` -} +type Chunk = buscalls.SeatChunk // 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 {