diff --git a/modules/mesh-recovery/internal/recovery/fakes_test.go b/modules/mesh-recovery/internal/recovery/fakes_test.go new file mode 100644 index 000000000..58ccdbe2a --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/fakes_test.go @@ -0,0 +1,214 @@ +package recovery_test + +import ( + "context" + "errors" + "io" + "os" + "path/filepath" + "sync" + "testing" + "time" + + seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + + "mesh-recovery/internal/recovery" +) + +// 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 +} + +func (b *fakeBus) Objects(_ context.Context, names []string) ([]recovery.Object, error) { + b.mu.Lock() + defer b.mu.Unlock() + var out []recovery.Object + for _, n := range names { + c, ok := b.streams[n] + out = append(out, recovery.Object{Stream: n, Exists: ok, Bytes: int64(len(c)), Consumers: b.consumers[n]}) + } + return out, nil +} + +func (b *fakeBus) Take(_ context.Context, s string, w io.Writer) error { + b.mu.Lock() + defer b.mu.Unlock() + c, ok := b.streams[s] + if !ok { + return errors.New("no stream " + s) + } + _, err := io.WriteString(w, "snap:"+s+":"+c) + return err +} + +func (b *fakeBus) Delete(_ context.Context, s string) error { + b.mu.Lock() + defer b.mu.Unlock() + delete(b.streams, s) + return nil +} + +func (b *fakeBus) Restore(_ context.Context, file string) error { + raw, err := os.ReadFile(file) + if err != nil { + return err + } + 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") + } + } + var name, content string + rest := string(raw[5:]) + for i := range rest { + if rest[i] == ':' { + name, content = rest[:i], rest[i+1:] + break + } + } + if _, live := b.streams[name]; live { + return errors.New("stream name already in use") + } + b.streams[name] = content + 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() + b.setBack = append(b.setBack, cs...) + return nil +} + +type said struct { + Event string + Body any + Parts []string +} + +type fakeSeats struct { + mu sync.Mutex + events []said + handActs []string + handFail error + streams []string + backup string +} + +func (s *fakeSeats) Say(event string, body any, parts ...string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.events = append(s.events, said{event, body, parts}) + return nil +} + +func (s *fakeSeats) HandAct(caller, asker, what, why string) error { + s.mu.Lock() + defer s.mu.Unlock() + if s.handFail != nil { + return s.handFail + } + s.handActs = append(s.handActs, caller+"|"+asker+"|"+what+"|"+why) + return nil +} + +func (s *fakeSeats) Streams(context.Context, string, string, string) ([]string, error) { + return s.streams, nil +} + +func (s *fakeSeats) Backup(context.Context, string) (string, error) { + if s.backup == "" { + return "", errors.New("no repository snapshot holds it") + } + return s.backup, nil +} + +func (s *fakeSeats) named(event string) []said { + s.mu.Lock() + defer s.mu.Unlock() + var out []said + for _, e := range s.events { + if e.Event == event { + out = append(out, e) + } + } + return out +} + +type memDigests struct { + mu sync.Mutex + m map[string][]byte +} + +func (d *memDigests) Put(k string, v []byte) error { + d.mu.Lock() + defer d.mu.Unlock() + d.m[k] = v + return nil +} +func (d *memDigests) Get(k string) ([]byte, bool, error) { + d.mu.Lock() + defer d.mu.Unlock() + v, ok := d.m[k] + return v, ok, nil +} +func (d *memDigests) Delete(k string) error { + d.mu.Lock() + defer d.mu.Unlock() + delete(d.m, k) + return nil +} + +type noDump struct{} + +func (noDump) Wait(context.Context, func(string)) error { return nil } + +type rig struct { + e *recovery.Engine + b *fakeBus + s *fakeSeats + d *memDigests + t time.Time +} + +func newRig(t *testing.T) *rig { + t.Helper() + r := &rig{b: &fakeBus{streams: map[string]string{"app_notes": "hello", "KV_app_seen": "world"}, consumers: map[string][]seat.Consumer{}}, + s: &fakeSeats{streams: []string{"app_notes", "KV_app_seen"}}, d: &memDigests{m: map[string][]byte{}}, t: time.Date(2026, 11, 12, 1, 0, 0, 0, time.UTC)} + r.b.consumers["app_notes"] = []seat.Consumer{{Name: "c1", Stream: "app_notes", Sequence: 7}} + r.e = &recovery.Engine{Dir: filepath.Join(t.TempDir(), "points-dir"), Bus: r.b, Seats: r.s, Digests: r.d, Dump: noDump{}, + Settings: recovery.Settings{Cap: 1 << 20, KeepDays: 90}, Now: func() time.Time { return r.t }} + return r +} + +func (r *rig) do(a recovery.Act) recovery.Answer { + ans := r.e.Accept(a) + r.e.Wait() + return ans +} + +var errTest = errors.New("the log is unreachable") + +type gatedBus struct { + *fakeBus + gate chan struct{} +} + +func (g *gatedBus) Objects(ctx context.Context, n []string) ([]recovery.Object, error) { + <-g.gate + return g.fakeBus.Objects(ctx, n) +} diff --git a/modules/mesh-recovery/internal/recovery/points.go b/modules/mesh-recovery/internal/recovery/points.go new file mode 100644 index 000000000..beb3688b1 --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/points.go @@ -0,0 +1,221 @@ +package recovery + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sort" + "strings" + + seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +const pointFile = "point.json" + +// run is the work of one accepted act; it ends the act with its event under the act's id (§560). +func (e *Engine) run(ctx context.Context, a *Act) { + var err error + switch a.Kind { + case KindTake: + err = e.take(ctx, a) + case KindExpire: + e.expire(a.ID) + 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) + } + e.markEnded(a) +} + +// say publishes an event and logs the failure; an event lost is never silent. +func (e *Engine) say(event string, body any, parts ...string) { + if err := e.Seats.Say(event, body, parts...); err != nil { + e.logf("the event %s was not said: %v", event, err) + } +} + +func (e *Engine) endedFailed(a *Act, why error) error { + e.say(seat.EventEnded, seat.EndedEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form, + Fingerprint: a.Fingerprint, Snapshot: a.Snapshot, Outcome: seat.EndedEventOutcomeFailed, Error: why.Error(), Why: cancelWhy(a, why)}, + a.Module, a.Kept) + return why +} + +// cancelWhy is the hand act's words for a cancel (A-117). +func cancelWhy(a *Act, why error) string { + if why != nil && why.Error() == ErrCanceled { + return a.Why + } + return "" +} + +// filled is the fill of the points directory without the temporary names of acts still writing. +func (e *Engine) filled() int64 { + return dirSize(filepath.Join(e.Dir, "points")) + dirSize(filepath.Join(e.Dir, "aside")) +} + +// take is `snapshot-kept` and the hand verb `take` (§516, §564, §567): one stream at a time under a temporary name, +// renamed whole, its digests on the bus before `ended` is said. +func (e *Engine) take(ctx context.Context, a *Act) error { + dir, err := e.pointDir(a.Module, a.Kept, a.Form, a.Fingerprint) + if err != nil { + return e.endedFailed(a, err) + } + names, err := e.Seats.Streams(ctx, a.Module, a.Kept, a.Fingerprint) + if err != nil || len(names) == 0 { + if err == nil { + err = fmt.Errorf("%s's %s has no bus name at %s", a.Module, a.Kept, a.Fingerprint) + } + return e.endedFailed(a, err) + } + e.setStep(a, "measuring", "") + objs, err := e.Bus.Objects(ctx, names) + if err != nil { + return e.endedFailed(a, err) + } + var need int64 + for _, o := range objs { + need += o.Bytes + } + if fill := e.filled(); fill+need > e.Settings.Cap { + return e.endedFailed(a, fmt.Errorf("%s: the cap is %d bytes, the point needs %d and the directory holds %d", ErrCap, e.Settings.Cap, need, fill)) + } + tmp := filepath.Join(filepath.Dir(dir), ".tmp-"+a.ID) + _ = os.RemoveAll(tmp) + if err := os.MkdirAll(tmp, 0o700); err != nil { + return e.endedFailed(a, err) + } + defer os.RemoveAll(tmp) + p := seat.Point{At: e.now(), CatchUp: a.CatchUp, Fingerprint: a.Fingerprint, Form: a.Form, Kept: a.Kept, Module: a.Module, + Snapshot: strings.TrimPrefix(dir, filepath.Join(e.Dir, "points")+string(filepath.Separator))} + e.setStep(a, "taking", "") + for _, o := range objs { + if !o.Exists { + continue + } + file := filepath.Join(tmp, keys.EscapeName(o.Stream)+".snap") + sum, n, err := e.takeOne(ctx, o.Stream, file) + if err != nil { + return e.endedFailed(a, err) + } + p.Digests = append(p.Digests, seat.Digest{Name: o.Stream, Digest: sum}) + p.Size += n + p.Consumers = append(p.Consumers, o.Consumers...) + if e.filled()+p.Size > e.Settings.Cap { + return e.endedFailed(a, fmt.Errorf("%s: the stream grew while it was taken and passes the cap of %d bytes", ErrCap, e.Settings.Cap)) + } + } + if err := writeJSON(filepath.Join(tmp, pointFile), p); err != nil { + return e.endedFailed(a, err) + } + _ = os.RemoveAll(dir) + if err := os.Rename(tmp, dir); err != nil { + return e.endedFailed(a, err) + } + raw, _ := jsonMarshal(p.Digests) + if err := e.Digests.Put(keys.EscapeName(p.Snapshot), raw); err != nil { + _ = os.RemoveAll(dir) + return e.endedFailed(a, fmt.Errorf("the digests are not on the bus: %w", err)) + } + e.say(seat.EventEnded, seat.EndedEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form, + Fingerprint: a.Fingerprint, Snapshot: p.Snapshot, Outcome: seat.EndedEventOutcomeDone, Consumers: p.Consumers, + Digests: p.Digests, Size: p.Size, CatchUp: p.CatchUp}, a.Module, a.Kept) + e.expire(a.ID) + return nil +} + +func (e *Engine) takeOne(ctx context.Context, stream, file string) (string, int64, error) { + f, err := os.Create(file) + if err != nil { + return "", 0, err + } + if err := e.Bus.Take(ctx, stream, f); err != nil { + f.Close() + return "", 0, fmt.Errorf("stream %s: %w", stream, err) + } + if err := f.Close(); err != nil { + return "", 0, err + } + return digestOf(file) +} + +func digestOf(file string) (string, int64, error) { + f, err := os.Open(file) + if err != nil { + return "", 0, err + } + defer f.Close() + h := sha256.New() + n, err := io.Copy(h, f) + return "sha256:" + hex.EncodeToString(h.Sum(nil)), n, err +} + +// Points lists every restore point in the directory, newest first (§517): the one place they are kept. +func (e *Engine) Points() ([]seat.Point, error) { + var out []seat.Point + root := filepath.Join(e.Dir, "points") + err := filepath.WalkDir(root, func(path string, d os.DirEntry, err error) error { + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return err + } + if d.IsDir() && strings.HasPrefix(d.Name(), ".") { + return filepath.SkipDir + } + if !d.IsDir() && d.Name() == pointFile { + var p seat.Point + if err := readJSON(path, &p); err != nil { + return fmt.Errorf("%s: %w", path, err) + } + out = append(out, p) + } + return nil + }) + sort.Slice(out, func(i, j int) bool { return out[i].At.After(out[j].At) }) + return out, err +} + +// expire is §551's removal: a point 90 days after its `at` (the setting), never the newest of its kept data, each removal +// said as point-removed and the run as expiry-ended, also when it removes nothing (§560). It runs inside an act. +func (e *Engine) expire(act string) { + pts, err := e.Points() + if err != nil { + e.logf("expiry: %v", err) + } + newest := map[string]string{} + for _, p := range pts { // newest first + k := p.Module + "\x00" + p.Kept + if _, ok := newest[k]; !ok { + newest[k] = p.Snapshot + } + } + cutoff := e.now().AddDate(0, 0, -e.Settings.KeepDays) + removed := int64(0) + for _, p := range pts { + if !p.At.Before(cutoff) || newest[p.Module+"\x00"+p.Kept] == p.Snapshot { + continue + } + if err := os.RemoveAll(filepath.Join(e.Dir, "points", p.Snapshot)); err != nil { + e.logf("the point %s is not removed: %v", p.Snapshot, err) + continue + } + _ = e.Digests.Delete(keys.EscapeName(p.Snapshot)) + if err := e.Seats.HandAct(Module, "", "point-removed "+p.Snapshot, "older than the keep days"); err != nil { + e.logf("the removal of %s is not in the hand-act log: %v", p.Snapshot, err) + } + e.say(seat.EventPointRemoved, seat.PointRemovedEvent{Act: act, At: e.now(), Module: p.Module, Kept: p.Kept, Form: p.Form, + Fingerprint: p.Fingerprint, Snapshot: p.Snapshot}, p.Module, p.Kept) + removed++ + } + e.say(seat.EventExpiryEnded, seat.ExpiryEndedEvent{Act: act, At: e.now(), Removed: removed}) +} diff --git a/modules/mesh-recovery/internal/recovery/points_test.go b/modules/mesh-recovery/internal/recovery/points_test.go new file mode 100644 index 000000000..adef20c49 --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/points_test.go @@ -0,0 +1,121 @@ +package recovery_test + +import ( + "strings" + "testing" + "time" + + seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + + "mesh-recovery/internal/recovery" +) + +func takeAct() recovery.Act { + return recovery.Act{Kind: recovery.KindTake, Module: "app", Kept: "notes", Form: 2, Fingerprint: "fp1", Why: "form change", Caller: "node.engine"} +} + +// A point is taken stream by stream under a temporary name, renamed whole, its digests put on the bus before `ended` is +// said under the act's id with the point whole, and the act is written in the hand-act log with the caller the bus proved +// (design 47 G §516, §554, §560, §567). +func TestATakenPointIsListedWithItsDigestsAndEndedUnderItsAct(t *testing.T) { + r := newRig(t) + ans := r.do(takeAct()) + if ans.Accepted == "" { + t.Fatalf("answer %+v", ans) + } + ended := r.s.named(seat.EventEnded) + if len(ended) != 1 { + t.Fatalf("events %+v", r.s.events) + } + ev := ended[0].Body.(seat.EndedEvent) + if ev.Act != ans.Accepted || ev.Outcome != "done" || len(ev.Digests) != 2 || ev.Size == 0 || ev.Snapshot != "app/notes/2-fp1" { + t.Errorf("ended %+v", ev) + } + if got := strings.Join(ended[0].Parts, "."); got != "app.notes" { + t.Errorf("parts %q", got) + } + if len(ev.Consumers) != 1 || ev.Consumers[0].Sequence != 7 { + t.Errorf("consumers %+v", ev.Consumers) + } + pts, err := r.e.Points() + if err != nil || len(pts) != 1 || pts[0].Snapshot != ev.Snapshot { + t.Fatalf("points %+v %v", pts, err) + } + if _, ok, _ := r.d.Get("app/notes/2-fp1"); !ok { + t.Errorf("no digests on the bus: %v", r.d.m) + } + if len(r.s.handActs) != 1 || !strings.HasPrefix(r.s.handActs[0], "node.engine||take app notes|form change") { + t.Errorf("hand acts %v", r.s.handActs) + } +} + +// One act at a time and no queue: a second ask while one runs is answered busy with the running act (§554). +func TestASecondAskWhileOneRunsIsAnsweredBusy(t *testing.T) { + r := newRig(t) + gate := make(chan struct{}) + r.e.Dump = nil + r.e.Bus = &gatedBus{fakeBus: r.b, gate: gate} + first := r.e.Accept(takeAct()) + second := r.e.Accept(takeAct()) + close(gate) + r.e.Wait() + if first.Accepted == "" || second.Busy != first.Accepted || second.Accepted != "" || second.Error != "" { + t.Errorf("first %+v second %+v", first, second) + } +} + +// An act whose hand-act log write fails does nothing and answers an error naming the log (§554). +func TestAnActWhoseHandActLogWriteFailsDoesNothing(t *testing.T) { + r := newRig(t) + r.s.handFail = errTest + ans := r.do(takeAct()) + if !strings.Contains(ans.Error, "hand-act log") || ans.Accepted != "" || len(r.s.events) != 0 { + t.Errorf("answer %+v events %+v", ans, r.s.events) + } + if pts, _ := r.e.Points(); len(pts) != 0 { + t.Errorf("points %+v", pts) + } +} + +// An act that would pass the cap ends failed naming it, before anything is written (§564). +func TestAnActThatWouldPassTheCapEndsFailedBeforeItWritesAnything(t *testing.T) { + r := newRig(t) + r.e.Settings.Cap = 4 + r.do(takeAct()) + ev := r.s.named(seat.EventEnded)[0].Body.(seat.EndedEvent) + if ev.Outcome != "failed" || !strings.HasPrefix(ev.Error, recovery.ErrCap) { + t.Errorf("ended %+v", ev) + } + if pts, _ := r.e.Points(); len(pts) != 0 || len(r.d.m) != 0 { + t.Errorf("points %+v digests %v", pts, r.d.m) + } +} + +// A point older than the keep days is removed and said, the newest of its kept data stays however old, and the run says +// expiry-ended with the count, 0 included (§551, §560). +func TestExpiryRemovesOldPointsButNeverTheNewest(t *testing.T) { + r := newRig(t) + r.t = time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC) + r.do(takeAct()) + a := takeAct() + a.Fingerprint, a.Form = "fp2", 3 + r.t = time.Date(2026, 7, 1, 0, 0, 0, 0, time.UTC) + r.do(a) + r.s.events = nil + r.t = time.Date(2026, 11, 12, 0, 0, 0, 0, time.UTC) + r.do(recovery.Act{Kind: recovery.KindExpire, Why: "by hand", Caller: "tools"}) + pts, _ := r.e.Points() + if len(pts) != 1 || pts[0].Fingerprint != "fp2" { + t.Fatalf("points %+v", pts) + } + rem := r.s.named(seat.EventPointRemoved) + if len(rem) != 1 || rem[0].Body.(seat.PointRemovedEvent).Fingerprint != "fp1" { + t.Errorf("removed %+v", rem) + } + if ex := r.s.named(seat.EventExpiryEnded); len(ex) != 1 || ex[0].Body.(seat.ExpiryEndedEvent).Removed != 1 { + t.Errorf("expiry %+v", ex) + } + if _, ok, _ := r.d.Get("app/notes/2-fp1"); ok { + t.Errorf("its digests stay on the bus") + } +} diff --git a/modules/mesh-recovery/internal/recovery/recovery.go b/modules/mesh-recovery/internal/recovery/recovery.go new file mode 100644 index 000000000..153a18f6d --- /dev/null +++ b/modules/mesh-recovery/internal/recovery/recovery.go @@ -0,0 +1,270 @@ +// Package recovery is the recovery module's engine (design 47 G §554, §516): it takes, sets aside and restores a +// module's kept data on the live bus, one act at a time, each accepted act written in its directory before the caller is +// answered, the result said as an event under the act's id (§560). Every dependence on the bus, the seat list, the +// hand-act log and the backup holder is an interface here, so the one whole test runs the engine against the real bus +// and a unit test against a fake. +package recovery + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "os" + "path/filepath" + "sync" + "time" + + seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "git.novox.be/novox/mesh-sdk/go/keys" +) + +// Module and Seat are the module's name and the seat it claims. +const ( + Module = "mesh-recovery" + Seat = seat.Seat +) + +// The kinds of act, one per verb that starts one. +const ( + KindTake = "take" + KindRestore = "restore" + KindAside = "aside" + KindDiscardAside = "discard-aside" + KindRestoreAside = "restore-aside" + KindExpire = "expire-points" +) + +// Errors an act ends with: the words the node agent's table reads (§560). +const ( + ErrCap = "cap" + ErrStopped = "stopped" + ErrCanceled = "cancelled" + ErrMissing = "restore point missing" + ErrDamaged = "restore point damaged" +) + +// Object is one bus object of a kept data: a stream, by its name, with what the bus states of it. +type Object struct { + Stream string + Exists bool + Bytes int64 + Consumers []seat.Consumer +} + +// Bus is the bus as the module's account sees it (§554): a stream's snapshot and restore, its deletion, the +// durable consumers' positions. +type Bus interface { + // Objects states each stream of names: whether it exists, its bytes, its durable consumers' acknowledged sequences. + Objects(ctx context.Context, streams []string) ([]Object, error) + // Take writes one stream's snapshot file to w (ADR 0235 §10's pacing and §11's bounds are the implementation's). + Take(ctx context.Context, stream string, w io.Writer) error + // Delete removes a stream; one that is not there is no error. + Delete(ctx context.Context, stream string) error + // Restore puts the stream of a snapshot file back under its own name, where no stream of that name exists. + Restore(ctx context.Context, file string) error + // SetConsumers creates each durable consumer at its recorded sequence. + SetConsumers(ctx context.Context, cs []seat.Consumer) error +} + +// Seats is what the engine asks of the rest of the mesh. +type Seats interface { + // Say publishes one event of the seat under its subject parts (§560); an event of no parts has none. + Say(event string, body any, parts ...string) error + // HandAct writes the act in the hand-act log first (§554): the account the bus proved, who asked, what, why. + HandAct(caller, asker, what, why string) error + // Streams are the bus names of a module's kept data at a fingerprint, from its catalog graph entry (§554 Bus names). + Streams(ctx context.Context, module, kept, fingerprint string) ([]string, error) + // Backup brings the points directory of the module's item back beside its path through the backup holder and + // answers where (§516); an error when no repository snapshot holds it. + Backup(ctx context.Context, module string) (dir string, err error) +} + +// Digests keeps each point's digests on the bus, apart from its bytes (§567): the state digests. +type Digests interface { + Put(key string, v []byte) error + Get(key string) ([]byte, bool, error) + Delete(key string) error +} + +// Dump is what the act waits on while the bus's item is dumped (§563). +type Dump interface { + // Wait returns once no dump of the bus's item runs; it calls waiting(on) once if it has to wait. + Wait(ctx context.Context, waiting func(on string)) error +} + +// Settings are the bounds a whole test sets in seconds (§557, §564, §633). +type Settings struct { + Cap int64 + KeepDays int +} + +// Engine is the recovery module's one engine. +type Engine struct { + Dir string + Bus Bus + Seats Seats + Digests Digests + Dump Dump + Settings Settings + Now func() time.Time + Logf func(string, ...any) + + mu sync.Mutex + running *Act + cancel context.CancelFunc + wg sync.WaitGroup + seq int +} + +// 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"` +} + +// Answer is the one form every acting verb answers (§691): exactly one field set. +type Answer struct { + Accepted string `json:"accepted,omitempty"` + Busy string `json:"busy,omitempty"` + Error string `json:"error,omitempty"` +} + +func (e *Engine) now() time.Time { + if e.Now != nil { + return e.Now().UTC() + } + return time.Now().UTC() +} + +func (e *Engine) logf(f string, a ...any) { + if e.Logf != nil { + e.Logf(f, a...) + } +} + +// Tokens are escaped as a key's name is (§73); a token that is empty, `.` or `..` is refused. +func token(s string) (string, error) { + if s == "" || s == "." || s == ".." { + return "", fmt.Errorf("%q is not a name", s) + } + return keys.EscapeName(s), nil +} + +func (e *Engine) pointDir(module, kept string, form int64, fp string) (string, error) { + m, err := token(module) + if err != nil { + return "", err + } + k, err := token(kept) + if err != nil { + return "", err + } + f, err := token(fp) + if err != nil { + return "", err + } + return filepath.Join(e.Dir, "points", m, k, fmt.Sprintf("%d-%s", form, f)), nil +} + +func writeJSON(path string, v any) error { + raw, err := json.MarshalIndent(v, "", " ") + if err != nil { + return err + } + tmp := path + ".partial" + if err := os.WriteFile(tmp, raw, 0o600); err != nil { + return err + } + return os.Rename(tmp, path) +} + +func readJSON(path string, v any) error { + raw, err := os.ReadFile(path) + if err != nil { + return err + } + return json.Unmarshal(raw, v) +} + +func dirSize(dir string) int64 { + var n int64 + _ = filepath.WalkDir(dir, func(_ string, d os.DirEntry, err error) error { + if err == nil && !d.IsDir() { + if info, err := d.Info(); err == nil { + n += info.Size() + } + } + return nil + }) + return n +} + +var errBusy = errors.New("busy") + +// Accept is the one door of every acting verb (§554, §682): it writes the act in the hand-act log first (a failed +// write does nothing and answers an error naming the log), then answers busy with the running act's id, or writes the +// act in its directory and answers accepted; the work runs on, and its result is the event. +func (e *Engine) Accept(a Act) Answer { + e.mu.Lock() + defer e.mu.Unlock() + if e.running != nil { + return Answer{Busy: e.running.ID} + } + if err := e.Seats.HandAct(a.Caller, a.Asker, a.Kind+" "+a.Module+" "+a.Kept, a.Why); err != nil { + 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.Since = e.now() + if err := os.MkdirAll(filepath.Join(e.Dir, "acts"), 0o700); err != nil { + return Answer{Error: err.Error()} + } + if err := writeJSON(filepath.Join(e.Dir, "acts", a.ID+".json"), a); err != nil { + return Answer{Error: "the act cannot be written: " + err.Error()} + } + ctx, cancel := context.WithCancel(context.Background()) + e.running, e.cancel = &a, cancel + e.wg.Add(1) + go func() { + defer e.wg.Done() + e.run(ctx, &a) + e.mu.Lock() + e.running, e.cancel = nil, nil + e.mu.Unlock() + cancel() + }() + return Answer{Accepted: a.ID} +} + +// Wait returns once the running act, if any, has ended (a whole test's and a stop's join). +func (e *Engine) Wait() { e.wg.Wait() } + +func (e *Engine) setStep(a *Act, step, waitsOn string) { + e.mu.Lock() + a.Step, a.WaitsOn = step, waitsOn + e.mu.Unlock() +} + +func (e *Engine) markEnded(a *Act) { + a.Ended = true + _ = writeJSON(filepath.Join(e.Dir, "acts", a.ID+".json"), a) +} + +func jsonMarshal(v any) ([]byte, error) { return json.Marshal(v) }