diff --git a/modules/mesh-recovery/internal/recovery/control.go b/modules/mesh-recovery/internal/recovery/control.go new file mode 100644 index 000000000..868fd9f9e --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/control.go @@ -0,0 +1,164 @@ +package recovery + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + + seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" +) + +// Cancel is the verb `cancel` (§566, §682): it ends an accepted act that has not begun its work (one waiting on the +// dump, §563) with its ending event, outcome failed and error cancelled, the hand act's words as `why` (A-117); it +// refuses an act that has begun, naming its step, since a restore cut off after its removal is put back only by its own +// failure path (§516). It is written in the hand-act log first and answers in §691's one form. +func (e *Engine) Cancel(act, why, caller, asker string) Answer { + if err := e.Seats.HandAct(caller, asker, "cancel "+act, why); err != nil { + return Answer{Error: "the hand-act log was not written, so nothing was done: " + err.Error()} + } + e.mu.Lock() + defer e.mu.Unlock() + switch { + case e.running == nil || e.running.ID != act: + return Answer{Error: "the act " + act + " is not running"} + case e.running.Step != "waiting": + return Answer{Error: fmt.Sprintf("the act %s has begun (its step: %q) and is put back only by its own failure path", act, e.running.Step)} + } + e.why = why + e.cancel() + return Answer{Accepted: act} +} + +// Retry is the verb `retry` (§682): it asks again the last failed act of a module and kept data, with its fields, under +// an act of its own; `nothing to retry` when its last act did not fail. +func (e *Engine) Retry(module, kept, why, caller, asker string) Answer { + acts := e.readActs() + var last *Act + for i := range acts { + a := acts[i] + if a.Module == module && a.Kept == kept && a.Ended && a.Kind != KindExpire && (last == nil || !a.Since.Before(last.Since)) { + last = &a + } + } + if last == nil || last.Error == "" { + return Answer{Error: "nothing to retry"} + } + again := *last + again.ID, again.Ended, again.Error, again.Step, again.WaitsOn = "", false, "", "", "" + again.Why, again.Caller, again.Asker = why, caller, asker + return e.Accept(again) +} + +func (e *Engine) readActs() []Act { + var out []Act + files, _ := filepath.Glob(filepath.Join(e.Dir, "acts", "*.json")) + for _, f := range files { + var a Act + if readJSON(f, &a) == nil { + out = append(out, a) + } + } + return out +} + +// KeptPoints is the verb `kept-points` (§517): every restore point and aside with their sizes, narrowed to a module and a +// kept data, and the act that runs, with since when and what it waits on. +func (e *Engine) KeptPoints(module, kept string) (seat.KeptPointsAnswer, error) { + pts, err := e.Points() + if err != nil { + return seat.KeptPointsAnswer{}, err + } + out := seat.KeptPointsAnswer{Points: []seat.Point{}, Asides: []seat.Aside{}} + for _, p := range pts { + if (module == "" || p.Module == module) && (kept == "" || p.Kept == kept) { + out.Points = append(out.Points, p) + } + } + _, recs, _ := e.Asides() + for _, r := range recs { + if (module == "" || r.Module == module) && (kept == "" || r.Kept == kept) { + out.Asides = append(out.Asides, r.aside()) + } + } + e.mu.Lock() + if e.running != nil { + out.Running = &seat.Running{Act: e.running.ID, Step: e.running.Step, Since: e.running.Since, WaitsOn: e.running.WaitsOn} + } + e.mu.Unlock() + return out, nil +} + +// Unrestored are the asides whose record ends at `removed` that no running act holds (§554 health, §561): the live data +// of those names is gone and only the aside holds it. +func (e *Engine) Unrestored() []seat.Aside { + _, recs, _ := e.Asides() + e.mu.Lock() + defer e.mu.Unlock() + var out []seat.Aside + for _, r := range recs { + held := e.running != nil && e.running.Module == r.Module && e.running.Kept == r.Kept + if r.Step == stepRemoved && !held { + out = append(out, r.aside()) + } + } + return out +} + +// Start repairs what a restart cut off (§516, §564, §560): points under a temporary name are removed, an aside whose +// record ends at `set aside` (the live data never touched) is removed, an aside that ends at `removed` is put back under +// the live names, and every accepted act that never ended ends failed with `stopped`, so its caller asks again. It +// answers the asides it could not put back (the finding aside-unrestored, §561). +func (e *Engine) Start(ctx context.Context) []seat.Aside { + _ = filepath.WalkDir(filepath.Join(e.Dir, "points"), func(path string, d os.DirEntry, err error) error { + if err == nil && d.IsDir() && strings.HasPrefix(d.Name(), ".tmp-") { + _ = os.RemoveAll(path) + return filepath.SkipDir + } + return nil + }) + _, recs, _ := e.Asides() + for i := range recs { + r := &recs[i] + switch r.Step { + case stepSetAside: + _ = os.RemoveAll(e.asideDirOf(r)) + case stepRemoved: + if live, err := e.Bus.Objects(ctx, r.Streams); err == nil && !anyExists(live) { + if err := e.restoreFromAside(ctx, r, false); err != nil { + e.logf("the aside %s is not put back: %v", r.ID, err) + } + } + } + } + acts := e.readActs() + sort.Slice(acts, func(i, j int) bool { return acts[i].Since.Before(acts[j].Since) }) + for i := range acts { + a := acts[i] + if a.Ended { + continue + } + stopped := errors.New(ErrStopped) + switch a.Kind { + case KindTake: + _ = e.endedFailed(&a, stopped) + case KindRestore, KindRestoreAside: + _ = e.endedRestored(&a, stopped, nil, false) + } + a.Error = ErrStopped + e.markEnded(&a) + } + return e.Unrestored() +} + +func anyExists(os []Object) bool { + for _, o := range os { + if o.Exists { + return true + } + } + return false +} diff --git a/modules/mesh-recovery/internal/recovery/fakes_test.go b/modules/mesh-recovery/internal/recovery/fakes_test.go index 58ccdbe2a..a38f086f8 100644 --- a/modules/mesh-recovery/internal/recovery/fakes_test.go +++ b/modules/mesh-recovery/internal/recovery/fakes_test.go @@ -17,11 +17,11 @@ import ( // fakeBus holds streams as strings; a snapshot file is "snap::". type fakeBus struct { - mu sync.Mutex - streams map[string]string - consumers map[string][]seat.Consumer - setBack []seat.Consumer - failRestoreOnce bool + mu sync.Mutex + streams map[string]string + consumers map[string][]seat.Consumer + setBack []seat.Consumer + failRestores int } func (b *fakeBus) Objects(_ context.Context, names []string) ([]recovery.Object, error) { @@ -60,10 +60,9 @@ func (b *fakeBus) Restore(_ context.Context, file string) error { } b.mu.Lock() defer b.mu.Unlock() - if b.failRestoreOnce && len(raw) > 0 && b.streams != nil { - if string(raw[:5]) == "snap:" && !b.restoredOnce(raw) { - return errors.New("the server refused the archive") - } + if b.failRestores > 0 { + b.failRestores-- + return errors.New("the server refused the archive") } var name, content string rest := string(raw[5:]) @@ -80,13 +79,6 @@ func (b *fakeBus) Restore(_ context.Context, file string) error { return nil } -var once sync.Map - -func (b *fakeBus) restoredOnce(raw []byte) bool { - _, loaded := once.LoadOrStore(string(raw), true) - return loaded -} - func (b *fakeBus) SetConsumers(_ context.Context, cs []seat.Consumer) error { b.mu.Lock() defer b.mu.Unlock() @@ -177,6 +169,16 @@ type noDump struct{} func (noDump) Wait(context.Context, func(string)) error { return nil } +// holdDump waits until ctx ends, saying it waits. +type holdDump struct{ waiting chan string } + +func (h holdDump) Wait(ctx context.Context, waiting func(string)) error { + waiting("the dump of the bus") + h.waiting <- "waiting" + <-ctx.Done() + return ctx.Err() +} + type rig struct { e *recovery.Engine b *fakeBus diff --git a/modules/mesh-recovery/internal/recovery/points.go b/modules/mesh-recovery/internal/recovery/points.go index beb3688b1..9256efd47 100644 --- a/modules/mesh-recovery/internal/recovery/points.go +++ b/modules/mesh-recovery/internal/recovery/points.go @@ -26,11 +26,18 @@ func (e *Engine) run(ctx context.Context, a *Act) { err = e.take(ctx, a) case KindExpire: e.expire(a.ID) + case KindRestore: + err = e.restore(ctx, a) + case KindRestoreAside: + err = e.restoreAside(ctx, a) + case KindDiscardAside: + err = e.discardAside(a) default: err = e.endedFailed(a, fmt.Errorf("no act of kind %q", a.Kind)) } if err != nil { e.logf("act %s ended failed: %v", a.ID, err) + a.Error = err.Error() } e.markEnded(a) } diff --git a/modules/mesh-recovery/internal/recovery/recovery.go b/modules/mesh-recovery/internal/recovery/recovery.go index 153a18f6d..bdd0d05d8 100644 --- a/modules/mesh-recovery/internal/recovery/recovery.go +++ b/modules/mesh-recovery/internal/recovery/recovery.go @@ -116,6 +116,7 @@ type Engine struct { cancel context.CancelFunc wg sync.WaitGroup seq int + why string // the hand act words of a cancel (A-117) } // Act is one accepted act, written in the directory before its caller is answered (§554). @@ -137,6 +138,7 @@ type Act struct { 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. @@ -231,7 +233,7 @@ func (e *Engine) Accept(a Act) Answer { return Answer{Error: "the hand-act log was not written, so nothing was done: " + err.Error()} } e.seq++ - a.ID = fmt.Sprintf("act-%s-%d", e.now().Format("20060102T150405Z"), e.seq) + a.ID = fmt.Sprintf("act-%s-%06d", e.now().Format("20060102T150405Z"), e.seq) a.Since = e.now() if err := os.MkdirAll(filepath.Join(e.Dir, "acts"), 0o700); err != nil { return Answer{Error: err.Error()} @@ -268,3 +270,5 @@ func (e *Engine) markEnded(a *Act) { } func jsonMarshal(v any) ([]byte, error) { return json.Marshal(v) } + +func unmarshal(raw []byte, v any) error { return json.Unmarshal(raw, v) } diff --git a/modules/mesh-recovery/internal/recovery/restore.go b/modules/mesh-recovery/internal/recovery/restore.go new file mode 100644 index 000000000..cfec63109 --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/restore.go @@ -0,0 +1,335 @@ +package recovery + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "sort" + "strings" + + seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +const asideFile = "aside.json" + +// The steps an aside's record says, in their order (§516): a restore that fails after `removed` puts the aside back. +const ( + stepSetAside = "set aside" + stepRemoved = "removed" + stepRestored = "restored" +) + +// asideRecord is the record in an aside's directory (§516); nothing of an aside is on the bus. +type asideRecord struct { + ID string `json:"id"` + Module string `json:"module"` + Kept string `json:"kept"` + Streams []string `json:"streams"` + Consumers []seat.Consumer `json:"consumers,omitempty"` + Digests []seat.Digest `json:"digests"` + Size int64 `json:"size"` + Step string `json:"step"` +} + +func (r asideRecord) aside() seat.Aside { + return seat.Aside{ID: r.ID, Name: strings.Join(r.Streams, ","), Size: r.Size} +} + +func (e *Engine) asideRoot(module, kept string) (string, error) { + m, err := token(module) + if err != nil { + return "", err + } + k, err := token(kept) + if err != nil { + return "", err + } + return filepath.Join(e.Dir, "aside", m, k), nil +} + +// endedRestored says the end of a restore, or of a restore-aside, under the act's id (§560). +func (e *Engine) endedRestored(a *Act, outcomeErr error, aside *seat.Aside, missing bool) error { + ev := seat.RestoredEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form, Fingerprint: a.Fingerprint, + Snapshot: a.Snapshot, Outcome: seat.RestoredEventOutcomeDone, Aside: aside, Missing: missing} + if outcomeErr != nil { + ev.Outcome, ev.Error, ev.Why = seat.RestoredEventOutcomeFailed, outcomeErr.Error(), e.cancelledWhy(outcomeErr) + } + e.say(seat.EventRestored, ev, a.Module, a.Kept) + return outcomeErr +} + +func (e *Engine) cancelledWhy(err error) string { + e.mu.Lock() + defer e.mu.Unlock() + if err != nil && err.Error() == ErrCanceled { + return e.why + } + return "" +} + +// restore is `restore-kept` and the hand verb `restore` with a point (§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. +func (e *Engine) restore(ctx context.Context, a *Act) error { + e.say(seat.EventRestoreStarted, seat.RestoreStartedEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form, + Fingerprint: a.Fingerprint, Snapshot: a.Snapshot}, a.Module, a.Kept) + if e.Dump != nil { + if err := e.Dump.Wait(ctx, func(on string) { e.setStep(a, "waiting", on) }); err != nil { + if ctx.Err() != nil { + err = errors.New(ErrCanceled) + } + return e.endedRestored(a, err, nil, false) + } + } + 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 { + if !a.FromBackup { + return e.endedRestored(a, errors.New(ErrMissing), nil, true) + } + dir, err := e.Seats.Backup(ctx, Module) + if err != nil { + return e.endedRestored(a, errors.New(ErrMissing), nil, true) + } + defer os.RemoveAll(dir) + pointPath = filepath.Join(dir, filepath.FromSlash(a.Snapshot)) + if _, err := os.Stat(filepath.Join(pointPath, pointFile)); err != nil { + return e.endedRestored(a, errors.New(ErrMissing), nil, true) + } + } + var p seat.Point + if err := readJSON(filepath.Join(pointPath, pointFile), &p); err != nil { + return e.endedRestored(a, fmt.Errorf("%s: %w", ErrDamaged, err), nil, false) + } + if err := e.checkDigests(pointPath, a.Snapshot, p); err != nil { + return e.endedRestored(a, err, nil, false) + } + names := make([]string, 0, len(p.Digests)) + for _, d := range p.Digests { + names = append(names, d.Name) + } + rec, err := e.setAside(ctx, a, names, true) + if err != nil { + return e.endedRestored(a, err, nil, false) + } + // Past the removal, a failure puts the aside back (§516). + e.setStep(a, "restoring", "") + if err := e.putBack(ctx, pointPath, p.Digests, p.Consumers); err != nil { + 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) +} + +func asidePtr(r *asideRecord) *seat.Aside { + if r == nil { + return nil + } + x := r.aside() + return &x +} + +func (e *Engine) asideDirOf(r *asideRecord) string { + root, _ := e.asideRoot(r.Module, r.Kept) + return filepath.Join(root, r.ID) +} + +// checkDigests computes the digest of the bytes about to be restored and compares it with the key on the bus; the +// record in the directory is never the reference (§567). +func (e *Engine) checkDigests(pointPath, snapshot string, p seat.Point) error { + raw, ok, err := e.Digests.Get(keys.EscapeName(snapshot)) + if err != nil { + return fmt.Errorf("the digests of %s cannot be read: %w", snapshot, err) + } + if !ok { + return fmt.Errorf("%s: %s has no digests on the bus", ErrDamaged, snapshot) + } + var want []seat.Digest + if err := unmarshal(raw, &want); err != nil { + return fmt.Errorf("%s: %s: %w", ErrDamaged, snapshot, err) + } + if len(want) != len(p.Digests) { + return fmt.Errorf("%s: %s holds %d object(s), the bus says %d", ErrDamaged, snapshot, len(p.Digests), len(want)) + } + for _, w := range want { + got, _, err := digestOf(filepath.Join(pointPath, keys.EscapeName(w.Name)+".snap")) + if err != nil || got != w.Digest { + return fmt.Errorf("%s: %s, bus object %s", ErrDamaged, snapshot, w.Name) + } + } + return nil +} + +// setAside takes a snapshot of each live object into aside///, reads it back to check it, and only then +// (when remove) removes the live objects, each step written first in the record (§516). Past the cap it refuses before +// any object is removed (§564). +func (e *Engine) setAside(ctx context.Context, a *Act, names []string, remove bool) (*asideRecord, error) { + e.setStep(a, "setting aside", "") + objs, err := e.Bus.Objects(ctx, names) + if err != nil { + return nil, err + } + var need int64 + for _, o := range objs { + need += o.Bytes + } + if fill := e.filled(); fill+need > e.Settings.Cap { + return nil, fmt.Errorf("%s: the cap is %d bytes, the aside needs %d and the directory holds %d", ErrCap, e.Settings.Cap, need, fill) + } + root, err := e.asideRoot(a.Module, a.Kept) + if err != nil { + return nil, err + } + id := e.now().Format("20060102T150405Z") + for n := 2; ; n++ { + if _, err := os.Stat(filepath.Join(root, id)); err != nil { + break + } + id = fmt.Sprintf("%s-%d", e.now().Format("20060102T150405Z"), n) + } + dir := filepath.Join(root, id) + if err := os.MkdirAll(dir, 0o700); err != nil { + return nil, err + } + rec := &asideRecord{ID: id, Module: a.Module, Kept: a.Kept, Step: stepSetAside} + fail := func(err error) (*asideRecord, error) { _ = os.RemoveAll(dir); return nil, err } + for _, o := range objs { + if !o.Exists { + continue + } + sum, n, err := e.takeOne(ctx, o.Stream, filepath.Join(dir, keys.EscapeName(o.Stream)+".snap")) + if err != nil { + return fail(err) + } + if again, _, err := digestOf(filepath.Join(dir, keys.EscapeName(o.Stream)+".snap")); err != nil || again != sum { + return fail(fmt.Errorf("the aside of %s does not read back whole", o.Stream)) + } + rec.Streams = append(rec.Streams, o.Stream) + rec.Digests = append(rec.Digests, seat.Digest{Name: o.Stream, Digest: sum}) + rec.Consumers = append(rec.Consumers, o.Consumers...) + rec.Size += n + } + if err := writeJSON(filepath.Join(dir, asideFile), rec); err != nil { + return fail(err) + } + if !remove { + return rec, nil + } + rec.Step = stepRemoved + if err := writeJSON(filepath.Join(dir, asideFile), rec); err != nil { + return fail(err) + } + for _, s := range rec.Streams { + if err := e.Bus.Delete(ctx, s); err != nil { + if back := e.restoreFromAside(ctx, rec, false); back != nil { + return rec, fmt.Errorf("%w; and the aside is not put back: %v", err, back) + } + return rec, err + } + } + return rec, nil +} + +// putBack restores each stream of a point under its own name and sets each consumer to its recorded sequence. +func (e *Engine) putBack(ctx context.Context, dir string, ds []seat.Digest, cs []seat.Consumer) error { + for _, d := range ds { + if err := e.Bus.Restore(ctx, filepath.Join(dir, keys.EscapeName(d.Name)+".snap")); err != nil { + return fmt.Errorf("stream %s: %w", d.Name, err) + } + } + return e.Bus.SetConsumers(ctx, cs) +} + +// restoreFromAside puts an aside back under the live names: any partly restored stream is removed first. +func (e *Engine) restoreFromAside(ctx context.Context, rec *asideRecord, clear bool) error { + if clear { + for _, s := range rec.Streams { + if err := e.Bus.Delete(ctx, s); err != nil { + return err + } + } + } + if err := e.putBack(ctx, e.asideDirOf(rec), rec.Digests, rec.Consumers); err != nil { + return err + } + rec.Step = stepRestored + return writeJSON(filepath.Join(e.asideDirOf(rec), asideFile), rec) +} + +func (e *Engine) readAside(module, kept, id string) (*asideRecord, error) { + root, err := e.asideRoot(module, kept) + if err != nil { + return nil, err + } + if id == "" || id == "." || id == ".." || strings.ContainsAny(id, "/") { + return nil, fmt.Errorf("%q is not an aside", id) + } + var rec asideRecord + if err := readJSON(filepath.Join(root, id, asideFile), &rec); err != nil { + return nil, fmt.Errorf("no aside %s of %s's %s", id, module, kept) + } + return &rec, nil +} + +// restoreAside is `restore-aside` (§566): it puts an aside whose record ends at `removed` back under the live names it +// was taken from, refusing while a live object of those names exists, and ends with `restored`. +func (e *Engine) restoreAside(ctx context.Context, a *Act) error { + rec, err := e.readAside(a.Module, a.Kept, a.AsideID) + if err != nil { + return e.endedRestored(a, err, nil, false) + } + objs, err := e.Bus.Objects(ctx, rec.Streams) + if err != nil { + return e.endedRestored(a, err, nil, false) + } + for _, o := range objs { + if o.Exists { + return e.endedRestored(a, fmt.Errorf("the live stream %s exists; the aside %s is not put back over it", o.Stream, rec.ID), asidePtr(rec), false) + } + } + if err := e.restoreFromAside(ctx, rec, false); err != nil { + return e.endedRestored(a, err, asidePtr(rec), false) + } + return e.endedRestored(a, nil, asidePtr(rec), false) +} + +// discardAside is `discard-aside` (§554): it removes that directory and nothing else, an id holding `.`, `..` or `/` +// refused; it ends with `aside-discarded`. +func (e *Engine) discardAside(a *Act) error { + rec, err := e.readAside(a.Module, a.Kept, a.AsideID) + if err != nil { + return err + } + if err := os.RemoveAll(e.asideDirOf(rec)); err != nil { + return err + } + x := rec.aside() + e.say(seat.EventAsideDiscarded, seat.AsideDiscardedEvent{Act: a.ID, At: e.now(), ID: x.ID, Name: x.Name, Size: x.Size}, a.Module, a.Kept) + return nil +} + +// Asides lists every aside in the directory (§517). +func (e *Engine) Asides() ([]seat.Aside, []asideRecord, error) { + var out []seat.Aside + var recs []asideRecord + _ = filepath.WalkDir(filepath.Join(e.Dir, "aside"), func(path string, d os.DirEntry, err error) error { + if err == nil && !d.IsDir() && d.Name() == asideFile { + var r asideRecord + if readJSON(path, &r) == nil { + out = append(out, r.aside()) + recs = append(recs, r) + } + } + return nil + }) + sort.Slice(out, func(i, j int) bool { return out[i].ID < out[j].ID }) + return out, recs, nil +} diff --git a/modules/mesh-recovery/internal/recovery/restore_test.go b/modules/mesh-recovery/internal/recovery/restore_test.go new file mode 100644 index 000000000..1945a1823 --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/restore_test.go @@ -0,0 +1,234 @@ +package recovery_test + +import ( + "context" + "os" + "path/filepath" + "strings" + "testing" + + seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + + "mesh-recovery/internal/recovery" +) + +func restoreAct(snapshot string) recovery.Act { + return recovery.Act{Kind: recovery.KindRestore, Module: "app", Kept: "notes", Form: 2, Fingerprint: "fp1", Snapshot: snapshot, Why: "revert", Caller: "node.engine"} +} + +func takenRig(t *testing.T) *rig { + r := newRig(t) + r.do(takeAct()) + r.s.events = nil + return r +} + +func lastRestored(t *testing.T, r *rig) seat.RestoredEvent { + t.Helper() + ev := r.s.named(seat.EventRestored) + if len(ev) != 1 { + t.Fatalf("restored events %+v", r.s.events) + } + return ev[0].Body.(seat.RestoredEvent) +} + +// A restore says restore-started, sets the live data aside whole, removes it, restores the point under its names and sets +// each consumer to its recorded sequence, and says restored with the aside (design 47 G §516, §560). +func TestARestoreSetsTheLiveDataAsideAndRestoresThePoint(t *testing.T) { + r := takenRig(t) + r.b.streams["app_notes"] = "changed since" + ans := r.do(restoreAct("app/notes/2-fp1")) + ev := lastRestored(t, r) + if ev.Outcome != "done" || ev.Act != ans.Accepted || ev.Aside == nil || ev.Aside.Size == 0 { + t.Fatalf("restored %+v", ev) + } + if r.b.streams["app_notes"] != "hello" || r.b.streams["KV_app_seen"] != "world" { + t.Errorf("streams %v", r.b.streams) + } + if len(r.b.setBack) != 1 || r.b.setBack[0].Sequence != 7 { + t.Errorf("consumers %+v", r.b.setBack) + } + if len(r.s.named(seat.EventRestoreStarted)) != 1 { + t.Errorf("no restore-started") + } + kp, _ := r.e.KeptPoints("app", "notes") + if len(kp.Asides) != 1 || len(kp.Points) != 1 || kp.Running != nil { + t.Errorf("kept points %+v", kp) + } +} + +// A point missing from the directory says missing; with from-backup the backup holder brings it back and it is restored +// once its digest is checked; when no repository snapshot holds it, restore point missing (§516, §567). +func TestAMissingPointIsBroughtBackOrSaidMissing(t *testing.T) { + r := takenRig(t) + back := filepath.Join(t.TempDir(), "back") + if err := os.Rename(filepath.Join(r.e.Dir, "points"), back); err != nil { + t.Fatal(err) + } + r.do(restoreAct("app/notes/2-fp1")) + if ev := lastRestored(t, r); !ev.Missing || ev.Outcome != "failed" || ev.Error != recovery.ErrMissing { + t.Errorf("no backup asked: %+v", ev) + } + r.s.events = nil + a := restoreAct("app/notes/2-fp1") + a.FromBackup = true + r.do(a) + if ev := lastRestored(t, r); !ev.Missing || ev.Error != recovery.ErrMissing { + t.Errorf("backup holder holds none: %+v", ev) + } + r.s.events, r.s.backup = nil, back + r.do(a) + if ev := lastRestored(t, r); ev.Outcome != "done" { + t.Errorf("brought back: %+v", ev) + } +} + +// A point whose bytes differ from the digests on the bus is refused before any aside is taken, the record in the +// directory never being the reference (§567). +func TestAPointWhoseBytesWereAlteredIsRefusedBeforeAnyAside(t *testing.T) { + r := takenRig(t) + snap := filepath.Join(r.e.Dir, "points", "app", "notes", "2-fp1", "app_notes.snap") + if err := os.WriteFile(snap, []byte("snap:app_notes:evil"), 0o600); err != nil { + t.Fatal(err) + } + r.do(restoreAct("app/notes/2-fp1")) + if ev := lastRestored(t, r); ev.Outcome != "failed" || !strings.HasPrefix(ev.Error, recovery.ErrDamaged) { + t.Errorf("restored %+v", ev) + } + if kp, _ := r.e.KeptPoints("", ""); len(kp.Asides) != 0 || r.b.streams["app_notes"] != "hello" { + t.Errorf("an aside was taken or the live data changed: %+v %v", kp, r.b.streams) + } +} + +// A step after the removal that fails puts the aside back under the live names and the act ends failed, so the kept data +// is never left gone (§516). +func TestARestoreThatFailsAfterTheRemovalPutsTheAsideBack(t *testing.T) { + r := takenRig(t) + r.b.streams["app_notes"] = "live now" + r.b.failRestores = 1 + r.do(restoreAct("app/notes/2-fp1")) + if ev := lastRestored(t, r); ev.Outcome != "failed" || ev.Aside == nil { + t.Fatalf("restored %+v", ev) + } + if r.b.streams["app_notes"] != "live now" || r.b.streams["KV_app_seen"] != "world" { + t.Errorf("live data not as before: %v", r.b.streams) + } + if n := len(r.e.Unrestored()); n != 0 { + t.Errorf("%d asides unrestored", n) + } +} + +// A restore past the cap is refused with the live data untouched (§564). +func TestARestoreWhoseAsideWouldPassTheCapIsRefusedBeforeTheRemoval(t *testing.T) { + r := takenRig(t) + r.e.Settings.Cap = 1 + r.do(restoreAct("app/notes/2-fp1")) + if ev := lastRestored(t, r); ev.Outcome != "failed" || !strings.HasPrefix(ev.Error, recovery.ErrCap) { + t.Errorf("restored %+v", ev) + } + if r.b.streams["app_notes"] != "hello" { + t.Errorf("live data touched: %v", r.b.streams) + } +} + +// An act waiting on the dump is cancelled with its ending event failed/cancelled and the hand act's words as why; an act +// that has begun is refused, naming its step (§566, A-117). +func TestAWaitingActIsCancelledAndABegunOneRefused(t *testing.T) { + r := takenRig(t) + hd := holdDump{waiting: make(chan string, 1)} + r.e.Dump = hd + ans := r.e.Accept(restoreAct("app/notes/2-fp1")) + <-hd.waiting + kp, _ := r.e.KeptPoints("", "") + if kp.Running == nil || kp.Running.WaitsOn == "" || kp.Running.Act != ans.Accepted { + t.Errorf("running %+v", kp.Running) + } + if c := r.e.Cancel("act-nothing", "x", "tools", "op"); c.Error == "" { + t.Errorf("an unknown act cancelled: %+v", c) + } + if c := r.e.Cancel(ans.Accepted, "not now", "tools", "op"); c.Accepted != ans.Accepted { + t.Errorf("cancel %+v", c) + } + r.e.Wait() + ev := lastRestored(t, r) + if ev.Outcome != "failed" || ev.Error != recovery.ErrCanceled || ev.Why != "not now" || ev.Act != ans.Accepted { + t.Errorf("restored %+v", ev) + } +} + +// retry asks again the last failed act with its fields under an act of its own, and answers `nothing to retry` when its +// last act did not fail (§682). +func TestRetryAsksTheLastFailedActAgain(t *testing.T) { + r := takenRig(t) + r.e.Settings.Cap = 1 + r.do(restoreAct("app/notes/2-fp1")) + r.e.Settings.Cap = 1 << 20 + r.s.events = nil + ans := r.e.Retry("app", "notes", "again", "tools", "op") + r.e.Wait() + if ev := lastRestored(t, r); ans.Accepted == "" || ev.Outcome != "done" || ev.Act != ans.Accepted { + t.Errorf("answer %+v restored %+v", ans, ev) + } + if again := r.e.Retry("app", "notes", "again", "tools", "op"); again.Error != "nothing to retry" { + t.Errorf("answer %+v", again) + } +} + +// An aside whose record ends at removed is put back at the start, an act cut off by a restart ends failed with stopped, +// and a point left under a temporary name is removed (§516, §560, §564). +func TestAStartPutsBackAnAsideEndsACutOffActAndClearsTemporaryPoints(t *testing.T) { + r := takenRig(t) + r.do(restoreAct("app/notes/2-fp1")) // leaves an aside at "restored" + _, recs, _ := r.e.Asides() + if len(recs) != 1 { + t.Fatal("no aside") + } + // Cut a restore off after its removal: the record says removed, the live data is gone, an accepted act never ended. + dir := filepath.Join(r.e.Dir, "aside", "app", "notes", recs[0].ID) + raw, _ := os.ReadFile(filepath.Join(dir, "aside.json")) + _ = os.WriteFile(filepath.Join(dir, "aside.json"), []byte(strings.Replace(string(raw), `"step": "restored"`, `"step": "removed"`, 1)), 0o600) + delete(r.b.streams, "app_notes") + delete(r.b.streams, "KV_app_seen") + tmp := filepath.Join(r.e.Dir, "points", "app", "notes", ".tmp-act-x") + _ = os.MkdirAll(tmp, 0o700) + _ = os.WriteFile(filepath.Join(r.e.Dir, "acts", "act-cut.json"), []byte(`{"act":"act-cut","kind":"restore","module":"app","kept":"notes","why":"w","caller":"c","since":"2026-11-12T00:00:00Z"}`), 0o600) + r.s.events = nil + if left := r.e.Start(context.Background()); len(left) != 0 { + t.Errorf("unrestored %+v", left) + } + if r.b.streams["app_notes"] != "hello" { + t.Errorf("the aside was not put back: %v", r.b.streams) + } + if _, err := os.Stat(tmp); err == nil { + t.Errorf("the temporary point stays") + } + if ev := lastRestored(t, r); ev.Act != "act-cut" || ev.Error != recovery.ErrStopped { + t.Errorf("restored %+v", ev) + } +} + +// restore-aside puts an aside back under the live names, refused while live objects of those names exist; discard-aside +// removes the aside's directory and says aside-discarded (§554, §566). +func TestRestoreAsideRefusesLiveNamesAndDiscardAsideRemovesTheDirectory(t *testing.T) { + r := takenRig(t) + r.do(restoreAct("app/notes/2-fp1")) + _, recs, _ := r.e.Asides() + id := recs[0].ID + r.s.events = nil + r.do(recovery.Act{Kind: recovery.KindRestoreAside, Module: "app", Kept: "notes", AsideID: id, Why: "x", Caller: "tools"}) + if ev := lastRestored(t, r); ev.Outcome != "failed" || !strings.Contains(ev.Error, "live stream") { + t.Errorf("restored %+v", ev) + } + r.s.events = nil + r.do(recovery.Act{Kind: recovery.KindDiscardAside, Module: "app", Kept: "notes", AsideID: "../x", Why: "x", Caller: "tools"}) + if len(r.s.named(seat.EventAsideDiscarded)) != 0 { + t.Errorf("a path outside was discarded") + } + r.do(recovery.Act{Kind: recovery.KindDiscardAside, Module: "app", Kept: "notes", AsideID: id, Why: "x", Caller: "tools"}) + if ev := r.s.named(seat.EventAsideDiscarded); len(ev) != 1 || ev[0].Body.(seat.AsideDiscardedEvent).ID != id { + t.Errorf("events %+v", r.s.events) + } + if kp, _ := r.e.KeptPoints("", ""); len(kp.Asides) != 0 { + t.Errorf("asides %+v", kp.Asides) + } +}