From 6c2d6605a55469bfc8fcb5c03588b310deb2605e Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 02:36:19 +0100 Subject: [PATCH] mesh-recovery: its one whole test against the bus, and an act's end said once it is free MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The whole test runs the process on its own account against a bus of its own, every bound a setting of seconds, with a stand-in bus module for snapshot-get, snapshot-restore and snapshot-restored: a hand take's log, a restore waiting on the dump past its bound key, busy and the ask again, the expiry on its bound key, a keeper's snapshot in three chunks, a protected restore cancelled at its bound and its aside put back, one the keeper answers neither, and a stop putting the aside back. It found that an act said its ending event while it was still the running one, so a caller told busy that asked again on hearing it was told busy with the same act and heard nothing more: the act is now released and written ended before its ending event, and the removal at a snapshot-kept runs before its ended. Builds: design 47 G §557, §554, §681, §562, §551, A-138, A-140, A-141, A-143 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- .../mesh-recovery/whole_integration_test.go | 473 ++++++++++++++++++ .../mesh-recovery/internal/recovery/points.go | 33 +- .../internal/recovery/recovery.go | 4 +- .../internal/recovery/restore.go | 6 +- 4 files changed, 505 insertions(+), 11 deletions(-) create mode 100644 modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go diff --git a/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go b/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go new file mode 100644 index 000000000..5cbaf0e31 --- /dev/null +++ b/modules/mesh-recovery/cmd/mesh-recovery/whole_integration_test.go @@ -0,0 +1,473 @@ +//go:build integration + +package main + +// The recovery module's one whole test (design 47 G §557, §554, §681, §562, A-138, A-143): its run() as the node starts it, +// on its own account on a bus of this test alone, every bound a setting of a few seconds. A stand-in bus module answers +// the seat mesh-broker's snapshot-get (in chunks), snapshot-restore and snapshot-restore-cancel and says snapshot-restored; +// stand-ins answer the hand-act log, the assigner's assignments and the catalog's entry; the node-engine seat's running and +// the backup holder's item events are said on their streams. The live data is a real bucket, set aside and restored by +// the bus's own snapshots. + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/seats/meshbroker" + seat "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" + "git.novox.be/novox/mesh-sdk/go/buscalls" + "git.novox.be/novox/mesh-sdk/go/bussnap" + "git.novox.be/novox/mesh-sdk/go/process" + "git.novox.be/novox/mesh-sdk/go/testbus" + + "mesh-recovery/internal/recovery" +) + +// standKeeper is the bus module's keeper as the test plays it: a snapshot by request, protected or not, and what it says +// after a restore in place and after its cancel ("" says nothing). +type standKeeper struct { + mu sync.Mutex + file []byte // the snapshot an unprotected request sends + protected bool + onRestore string + onCancel string + asked []string +} + +func (k *standKeeper) set(protected bool, onRestore, onCancel string) { + k.mu.Lock() + k.protected, k.onRestore, k.onCancel, k.asked = protected, onRestore, onCancel, nil + k.mu.Unlock() +} + +func (k *standKeeper) said() string { + k.mu.Lock() + defer k.mu.Unlock() + return strings.Join(k.asked, "|") +} + +func TestRecoveryWhole(t *testing.T) { + url := testbus.URL(t) + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, _ := jetstream.New(nc) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute) + defer cancel() + acct := process.Account{URL: url, Node: "srv", Module: "mesh-recovery", User: "srv.mesh-recovery", Password: "unused"} + + // The module's own states (bounds takes keys with their own time to live), the app's live kept data, and the streams the + // grants part and the bus composer make: the seat events it hears with its durables, its own events and its findings. + for _, b := range []jetstream.KeyValueConfig{{Bucket: "mesh-recovery_bounds", LimitMarkerTTL: time.Second}, + {Bucket: "mesh-recovery_digests"}, {Bucket: "mesh-recovery_findings"}, {Bucket: "app_notes"}} { + if _, err := js.CreateKeyValue(ctx, b); err != nil { + t.Fatal(err) + } + } + for _, s := range []jetstream.StreamConfig{ + {Name: "RECOVERY_EVENTS", Subjects: []string{"mesh.seat.mesh-recovery.event.>"}}, + {Name: "RECOVERY_FINDINGS", Subjects: []string{"mesh.stream.mesh-recovery.event.>"}}, + {Name: "FINDINGS_ASKED", Subjects: []string{meshconditions.SubjectFindingsAsked}}, + } { + if _, err := js.CreateStream(ctx, s); err != nil { + t.Fatal(err) + } + } + for stream, subject := range map[string]string{ + nodeengine.StreamRunning: "mesh.seat.node-engine.event.running.>", + itemStartedSteam: "mesh.seat.node-backup.event.item-started.>", + itemEndedStream: "mesh.seat.node-backup.event.item-ended.>", + meshbroker.StreamSnapshotRestored: "mesh.seat.mesh-broker.event.snapshot-restored.>", + } { + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: stream, Subjects: []string{subject}, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, stream, acct.SeatEventConsumerConfig([]string{subject}, true)); err != nil { + t.Fatal(err) + } + } + notes, _ := js.KeyValue(ctx, "app_notes") + live := func() string { + e, err := notes.Get(ctx, "n") + if err != nil { + return "none: " + err.Error() + } + return string(e.Value()) + } + if _, err := notes.Put(ctx, "n", []byte("v1")); err != nil { + t.Fatal(err) + } + running, _ := json.Marshal(nodeengine.Running{V: 1, Node: "srv", Module: "app", Fingerprint: "src2:app1", At: time.Now()}) + if _, err := js.Publish(ctx, nodeengine.SubjectRunning("srv", "app"), running); err != nil { + t.Fatal(err) + } + + // The stand-ins of the seats the module calls. + answer := func(m *nats.Msg, result any) { + b, _ := json.Marshal(map[string]any{"result": result, "node": "srv"}) + _ = m.Respond(b) + } + var logMu sync.Mutex + var handActs []string + sub := func(subject string, h func(*nats.Msg)) { + if _, err := nc.Subscribe(subject, h); err != nil { + t.Fatal(err) + } + } + sub("mesh.seat.mesh-hand-acts.call.>", func(m *nats.Msg) { + logMu.Lock() + handActs = append(handActs, string(m.Data)) + logMu.Unlock() + answer(m, map[string]any{"recorded": true}) + }) + sub("mesh.seat.mesh-module-assigner.call.assignments.*", func(m *nats.Msg) { + answer(m, map[string]any{"seats": []any{map[string]any{"seat": meshbroker.Seat, "holders": []any{map[string]any{"module": "nats"}}}}}) + }) + sub("mesh.seat.mesh-catalog.call.entry.*", func(m *nats.Msg) { + answer(m, map[string]any{"entry": map[string]any{"module": "app", "fingerprint": "src2:app1", + "manifest": map[string]any{"module": "app", "version": "1", "state": []any{map[string]any{"name": "notes"}}}}}) + }) + keeper := &standKeeper{} + restored := func(request, act, outcome string) { + b, _ := json.Marshal(meshbroker.SnapshotRestoredEvent{Request: request, Act: act, Outcome: outcome, Error: "cancelled"}) + if _, err := js.Publish(ctx, meshbroker.SubjectSnapshotRestored(request, act), b); err != nil { + t.Error(err) + } + } + sub("mesh.seat.mesh-broker.call.snapshot-get.*", func(m *nats.Msg) { + var in meshbroker.SnapshotGetInput + _ = json.Unmarshal(m.Data, &in) + keeper.mu.Lock() + keeper.asked = append(keeper.asked, "get "+in.Request) + file, protected := keeper.file, keeper.protected + keeper.mu.Unlock() + if protected { + answer(m, meshbroker.SnapshotGetAnswer{Request: in.Request, Error: "snapshot protected"}) + return + } + sum := sha256.Sum256(file) + answer(m, meshbroker.SnapshotGetAnswer{Request: in.Request, Module: "app", Local: "notes", Stream: "KV_app_notes", + Size: int64(len(file)), Digest: "sha256:" + hex.EncodeToString(sum[:]), Chunks: 3}) + // Three chunks, each sent once the one before was acknowledged (§681). + go func() { + send := buscalls.SeatChunks{Seat: meshbroker.Seat, Verb: meshbroker.VerbSnapshotGet, Call: in.Call} + third := (len(file) + 2) / 3 + for n := int64(1); n <= 3; n++ { + acks, _ := nc.SubscribeSync(send.Ack(n)) + part := file[min(int(n-1)*third, len(file)):min(int(n)*third, len(file))] + body, _ := json.Marshal(buscalls.SeatChunk{N: n, Of: 3, Data: part}) + _ = nc.Publish(send.Chunk(n), body) + ack, err := acks.NextMsg(10 * time.Second) + _ = acks.Unsubscribe() + if err != nil { + t.Errorf("chunk %d not acknowledged: %v", n, err) + return + } + _ = ack.Respond([]byte("{}")) + } + }() + }) + sub("mesh.seat.mesh-broker.call.snapshot-restore.*", func(m *nats.Msg) { + var in meshbroker.SnapshotRestoreInput + _ = json.Unmarshal(m.Data, &in) + keeper.mu.Lock() + if in.Verify { + keeper.asked = append(keeper.asked, "verify "+in.Request) + keeper.mu.Unlock() + answer(m, meshbroker.SnapshotRestoreAnswer{Accepted: "keeper-verify"}) + return + } + keeper.asked = append(keeper.asked, "restore "+in.Request) + then := keeper.onRestore + keeper.mu.Unlock() + answer(m, meshbroker.SnapshotRestoreAnswer{Accepted: "keeper-" + in.Request}) + if then != "" { + restored(in.Request, "keeper-"+in.Request, then) + } + }) + sub("mesh.seat.mesh-broker.call.snapshot-restore-cancel.*", func(m *nats.Msg) { + var in meshbroker.SnapshotRestoreCancelInput + _ = json.Unmarshal(m.Data, &in) + keeper.mu.Lock() + keeper.asked = append(keeper.asked, "cancel "+in.Act) + then := keeper.onCancel + keeper.mu.Unlock() + answer(m, meshbroker.SnapshotRestoreCancelAnswer{Accepted: in.Act}) + if then != "" { + restored(strings.TrimPrefix(in.Act, "keeper-"), in.Act, then) + } + }) + + // Every event the module says, heard as it is said. + type heard struct { + subject string + body map[string]any + } + events := make(chan heard, 256) + sub("mesh.seat.mesh-recovery.event.>", func(m *nats.Msg) { + var b map[string]any + _ = json.Unmarshal(m.Data, &b) + events <- heard{m.Subject, b} + }) + var kept []heard + waitEvent := func(t *testing.T, what string, ok func(heard) bool) map[string]any { + t.Helper() + for i, h := range kept { + if ok(h) { + kept = append(kept[:i:i], kept[i+1:]...) + return h.body + } + } + deadline := time.After(30 * time.Second) + for { + select { + case h := <-events: + if ok(h) { + return h.body + } + kept = append(kept, h) + case <-deadline: + t.Fatalf("not said within 30s: %s", what) + } + } + } + endOf := func(t *testing.T, event, act string) map[string]any { + t.Helper() + return waitEvent(t, event+" of "+act, func(h heard) bool { + return strings.HasPrefix(h.subject, "mesh.seat.mesh-recovery.event."+event+".") && h.body["act"] == act + }) + } + ask := func(t *testing.T, verb, caller string, body any) seat.SnapshotKeptAnswer { + t.Helper() + data, _ := json.Marshal(body) + m, err := nc.Request("mesh.seat.mesh-recovery.call."+verb+"."+strings.ReplaceAll(caller, ".", "~"), data, 10*time.Second) + if err != nil { + t.Fatalf("%s: %v", verb, err) + } + var r struct { + Result seat.SnapshotKeptAnswer `json:"result"` + Error string `json:"error"` + } + _ = json.Unmarshal(m.Data, &r) + if r.Error != "" { + r.Result.Error = r.Error + } + return r.Result + } + // A caller told busy asks again when it hears the ending event of the act it was named (§554, §560): the expiry runs + // on its own bound beside the test's acts. + act := func(t *testing.T, verb, caller string, body any) string { + t.Helper() + for { + a := ask(t, verb, caller, body) + if a.Busy == "" { + if a.Accepted == "" { + t.Fatalf("%s answered %+v", verb, a) + } + return a.Accepted + } + waitEvent(t, "the ending of "+a.Busy, func(h heard) bool { + return h.body["act"] == a.Busy && !strings.Contains(h.subject, ".restore-started.") + }) + } + } + const engine, runner = "srv.node-engine", "srv.tool-runner" + + // The process, every bound a setting in seconds (§557, A-138): the dump wait and the restore wait 1.8 s, the expiry + // every 1.8 s, a point kept about 4 s. + dir := t.TempDir() + setting := func(name, v string) string { + p := filepath.Join(dir, name) + _ = os.WriteFile(p, []byte(v+"\n"), 0o600) + return p + } + acctFile, _ := json.Marshal(map[string]any{"url": url, "node": acct.Node, "module": acct.Module, "user": acct.User, "password": acct.Password}) + env := map[string]string{ + process.BrokerFileEnv: setting("broker", string(acctFile)), + envPoints: filepath.Join(dir, "points"), + envDumpWait: setting("dump-wait-minutes", "0.03"), + envRestore: setting("snapshot-restore-wait-minutes", "0.03"), + envExpire: setting("expire-every-hours", "0.0005"), + envKeepDays: setting("point-keep-days", "0.00005"), + } + runCtx, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- run(runCtx, func(k string) string { return env[k] }, testLog{t}) }() + defer func() { stop(); <-done }() + waitEvent(t, "the start's removal of points past their keep days", func(h heard) bool { + return h.subject == "mesh.seat.mesh-recovery.event."+seat.EventExpiryEnded + }) + + var first string + t.Run("a hand take's log names the account the bus proved and the asker (§554)", func(t *testing.T) { + ev := endOf(t, seat.EventEnded, act(t, seat.VerbTake, runner, seat.TakeInput{Module: "app", Kept: "notes", Fingerprint: "src2:app1", Why: "before a test", Asker: "op@srv"})) + if ev["outcome"] != "done" { + t.Fatalf("ended %v", ev) + } + first, _ = ev["snapshot"].(string) + logMu.Lock() + defer logMu.Unlock() + if len(handActs) == 0 || !strings.Contains(handActs[len(handActs)-1], runner) || !strings.Contains(handActs[len(handActs)-1], "op@srv") { + t.Errorf("the hand-act log holds %v", handActs) + } + }) + + t.Run("a restore waits on the dump past its bound key, a second ask is busy and asks again on its ended (§554, §563, A-143)", func(t *testing.T) { + if a := ask(t, seat.VerbRestoreKept, engine, map[string]any{"module": "app", "kept": "notes", "snapshot": first, "why": "w", "aside-only": true}); !strings.Contains(a.Error, "aside-only") { + t.Errorf("a node-engine's restore-kept with aside-only answered %+v", a) + } + item := func(event string) { + b, _ := json.Marshal(map[string]any{"at": time.Now()}) + if _, err := js.Publish(ctx, itemSubject(event, "srv", "nats", "data"), b); err != nil { + t.Fatal(err) + } + } + item(itemStarted) + _, _ = notes.Put(ctx, "n", []byte("changed")) + r := seat.SnapshotKeptAnswer{Accepted: act(t, seat.VerbRestoreKept, engine, seat.RestoreKeptInput{Module: "app", Kept: "notes", Form: 2, Fingerprint: "src2:app1", Snapshot: first, Why: "a form change"})} + if b := ask(t, seat.VerbSnapshotKept, engine, seat.SnapshotKeptInput{Module: "app", Kept: "notes", Form: 2, Fingerprint: "src2:app1", Why: "x"}); b.Busy != r.Accepted { + t.Errorf("a second ask answered %+v, want busy with %s", b, r.Accepted) + } + findings, _ := js.KeyValue(ctx, "mesh-recovery_findings") + w, _ := findings.Watch(ctx, "app."+r.Accepted+"."+recovery.FindingRecoveryWaits) + for e := range w.Updates() { + if e != nil && e.Operation() == jetstream.KeyValuePut { + break + } + } + _ = w.Stop() + item(itemEnded) + if ev := endOf(t, seat.EventRestored, r.Accepted); ev["outcome"] != "done" || live() != "v1" { + t.Fatalf("restored %v, live %q", ev, live()) + } + if e, err := findings.Get(ctx, "app."+r.Accepted+"."+recovery.FindingRecoveryWaits); err == nil { + t.Errorf("recovery-waits stands after the act ended: %s", e.Value()) + } + again := act(t, seat.VerbSnapshotKept, engine, seat.SnapshotKeptInput{Module: "app", Kept: "notes", Form: 2, Fingerprint: "src2:app1", Why: "asked again"}) + if ev := endOf(t, seat.EventEnded, again); ev["outcome"] != "done" { + t.Errorf("the ask again ended %v", ev) + } + }) + + t.Run("a point past its keep days is removed on the expiry's bound key, never the newest (§551, A-138)", func(t *testing.T) { + ev := waitEvent(t, "point-removed of "+first, func(h heard) bool { + return strings.HasPrefix(h.subject, "mesh.seat.mesh-recovery.event."+seat.EventPointRemoved+".") && h.body["snapshot"] == first + }) + if ev["module"] != "app" { + t.Errorf("point-removed %v", ev) + } + var kp seat.KeptPointsAnswer + m, _ := nc.Request("mesh.seat.mesh-recovery.call.kept-points."+strings.ReplaceAll(runner, ".", "~"), []byte(`{"module":"app","kept":"notes"}`), 10*time.Second) + var r struct { + Result seat.KeptPointsAnswer `json:"result"` + } + _ = json.Unmarshal(m.Data, &r) + kp = r.Result + if len(kp.Points) != 1 { + t.Errorf("the newest point is not kept alone: %+v", kp.Points) + } + }) + + t.Run("a keeper's snapshot of three chunks is received and restored whole (§681)", func(t *testing.T) { + _, _ = notes.Put(ctx, "n", []byte("from the keeper")) + var snap bytes.Buffer + if _, err := bussnap.Snapshot(ctx, nc, &snap, bussnap.SnapshotOptions{Streams: []string{"KV_app_notes"}, TempDir: t.TempDir()}); err != nil { + t.Fatal(err) + } + keeper.mu.Lock() + keeper.file = snap.Bytes() + keeper.mu.Unlock() + keeper.set(false, "", "") + _, _ = notes.Put(ctx, "n", []byte("since the drop")) + a := act(t, seat.VerbRestore, runner, seat.RestoreInput{Module: "app", Kept: "notes", Request: "req-1", Why: "test the drop's snapshot", Asker: "op@srv"}) + if ev := endOf(t, seat.EventRestored, a); ev["outcome"] != "done" || live() != "from the keeper" { + t.Errorf("restored %v, live %q", ev, live()) + } + if left, _ := os.ReadDir(filepath.Join(dir, "points", "incoming")); len(left) != 0 { + t.Errorf("incoming holds %v", left) + } + }) + + inPlace := func(t *testing.T, request string) string { + t.Helper() + _, _ = notes.Put(ctx, "n", []byte("live before "+request)) + return act(t, seat.VerbRestore, runner, seat.RestoreInput{Module: "app", Kept: "notes", Request: request, Why: "restore in place", Asker: "op@srv"}) + } + t.Run("a protected snapshot's bound asks the cancel first, then puts the aside back on its failed (§681, A-140, A-143)", func(t *testing.T) { + keeper.set(true, "", meshbroker.SnapshotRestoredEventOutcomeFailed) + id := inPlace(t, "req-2") + if ev := endOf(t, seat.EventRestored, id); ev["outcome"] != "failed" || live() != "live before req-2" { + t.Errorf("restored %v, live %q", ev, live()) + } + if got := keeper.said(); got != "get req-2|verify req-2|restore req-2|cancel keeper-req-2" { + t.Errorf("the keeper was asked %s", got) + } + }) + + t.Run("a keeper that answers neither leaves the aside unrestored, which restore-aside puts back (§681, §554)", func(t *testing.T) { + keeper.set(true, "", "") + id := inPlace(t, "req-3") + ev := endOf(t, seat.EventRestored, id) + if ev["outcome"] != "failed" || ev["error"] != "snapshot-restore unanswered" { + t.Fatalf("restored %v", ev) + } + var r struct { + Result seat.KeptPointsAnswer `json:"result"` + } + m, _ := nc.Request("mesh.seat.mesh-recovery.call.kept-points."+strings.ReplaceAll(runner, ".", "~"), []byte(`{}`), 10*time.Second) + _ = json.Unmarshal(m.Data, &r) + if len(r.Result.Unrestored) != 1 { + t.Fatalf("unrestored %+v", r.Result.Unrestored) + } + back := act(t, seat.VerbRestoreAside, runner, seat.RestoreAsideInput{Module: "app", Kept: "notes", ID: r.Result.Unrestored[0].ID, Why: "put it back", Asker: "op@srv"}) + if ev := endOf(t, seat.EventRestored, back); ev["outcome"] != "done" || live() != "live before req-3" { + t.Errorf("restore-aside %v, live %q", ev, live()) + } + }) + + t.Run("a stop during a restore past its removal puts the aside back and ends it stopped (§562, A-141)", func(t *testing.T) { + keeper.set(true, "", meshbroker.SnapshotRestoredEventOutcomeFailed) + id := inPlace(t, "req-4") + // The wait on snapshot-restored is a bound key: once it stands, the live data is removed and the keeper asked. + bounds, _ := js.KeyValue(ctx, "mesh-recovery_bounds") + w, _ := bounds.Watch(ctx, "restore-wait."+id) + for e := range w.Updates() { + if e != nil && e.Operation() == jetstream.KeyValuePut { + break + } + } + _ = w.Stop() + stop() + ev := endOf(t, seat.EventRestored, id) + if err := <-done; err != nil { + t.Errorf("the process ended %v", err) + } + done <- nil + if ev["outcome"] != "failed" || ev["error"] != "stopped" || live() != "live before req-4" { + t.Errorf("restored %v, live %q", ev, live()) + } + }) +} + +type testLog struct{ t *testing.T } + +func (l testLog) Write(p []byte) (int, error) { + l.t.Log(strings.TrimRight(string(p), "\n")) + return len(p), nil +} diff --git a/modules/mesh-recovery/internal/recovery/points.go b/modules/mesh-recovery/internal/recovery/points.go index 7f904c2f3..aa7207074 100644 --- a/modules/mesh-recovery/internal/recovery/points.go +++ b/modules/mesh-recovery/internal/recovery/points.go @@ -26,7 +26,7 @@ func (e *Engine) run(ctx context.Context, a *Act) { case KindTake: err = e.take(ctx, a) case KindExpire: - e.expire(a.ID) + e.expire(a.ID, true) case KindRestore: err = e.restore(ctx, a) case KindAside: @@ -45,6 +45,18 @@ func (e *Engine) run(ctx context.Context, a *Act) { e.markEnded(a) } +// sayEnd says the ending event of the act id (§554, §560): the act is first no longer the running one and written ended, so +// a caller told busy that asks again on hearing it is never told busy with the same act again. +func (e *Engine) sayEnd(id, event string, body any, parts ...string) { + e.mu.Lock() + if e.running != nil && e.running.ID == id { + e.markEnded(e.running) + e.running, e.cancel = nil, nil + } + e.mu.Unlock() + e.say(event, body, parts...) +} + // 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 { @@ -54,7 +66,7 @@ func (e *Engine) say(event string, body any, parts ...string) { func (e *Engine) endedFailed(a *Act, why error) error { why = e.stoppedOr(why) - e.say(seat.EventEnded, seat.EndedEvent{Act: a.ID, At: e.now(), Module: a.Module, Kept: a.Kept, Form: a.Form, + e.sayEnd(a.ID, 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 @@ -136,10 +148,11 @@ func (e *Engine) take(ctx context.Context, a *Act) error { _ = 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, + // The removal at each snapshot-kept (§551) runs within the take, before its ended, so one act runs at a time. + e.expire(a.ID, false) + e.sayEnd(a.ID, 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 } @@ -197,8 +210,9 @@ func (e *Engine) Points() ([]seat.Point, error) { } // 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) { +// said as point-removed and the run as expiry-ended, also when it removes nothing (§560). It runs inside an act: own when +// it is the act expire-points, whose ending event its expiry-ended is. +func (e *Engine) expire(act string, own bool) { pts, err := e.Points() if err != nil { e.logf("expiry: %v", err) @@ -228,5 +242,10 @@ func (e *Engine) expire(act string) { Fingerprint: p.Fingerprint, Snapshot: p.Snapshot}, p.Module, p.Kept) removed++ } - e.say(seat.EventExpiryEnded, seat.ExpiryEndedEvent{Act: act, At: e.now(), Removed: removed}) + ev := seat.ExpiryEndedEvent{Act: act, At: e.now(), Removed: removed} + if own { + e.sayEnd(act, seat.EventExpiryEnded, ev) + return + } + e.say(seat.EventExpiryEnded, ev) } diff --git a/modules/mesh-recovery/internal/recovery/recovery.go b/modules/mesh-recovery/internal/recovery/recovery.go index eeae6ea63..aa3568b3a 100644 --- a/modules/mesh-recovery/internal/recovery/recovery.go +++ b/modules/mesh-recovery/internal/recovery/recovery.go @@ -284,7 +284,9 @@ func (e *Engine) Accept(a Act) Answer { defer e.wg.Done() e.run(ctx, &a) e.mu.Lock() - e.running, e.cancel = nil, nil + if e.running == &a { // an act that said its ending event is no longer the running one (sayEnd) + e.running, e.cancel = nil, nil + } e.mu.Unlock() cancel() e.judge() diff --git a/modules/mesh-recovery/internal/recovery/restore.go b/modules/mesh-recovery/internal/recovery/restore.go index 6fd4fd60e..73f067e45 100644 --- a/modules/mesh-recovery/internal/recovery/restore.go +++ b/modules/mesh-recovery/internal/recovery/restore.go @@ -58,7 +58,7 @@ func (e *Engine) endedRestored(a *Act, outcomeErr error, aside *seat.Aside, miss 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) + e.sayEnd(a.ID, seat.EventRestored, ev, a.Module, a.Kept) return outcomeErr } @@ -310,7 +310,7 @@ func (e *Engine) aside(ctx context.Context, a *Act) error { return err } x := rec.aside() - e.say(seat.EventSetAside, seat.SetAsideEvent{Act: a.ID, At: e.now(), ID: x.ID, Names: x.Names, Size: x.Size}, a.Module, a.Kept) + e.sayEnd(a.ID, seat.EventSetAside, seat.SetAsideEvent{Act: a.ID, At: e.now(), ID: x.ID, Names: x.Names, Size: x.Size}, a.Module, a.Kept) return nil } @@ -347,7 +347,7 @@ func (e *Engine) discardAside(a *Act) error { return err } x := rec.aside() - e.say(seat.EventAsideDiscarded, seat.AsideDiscardedEvent{Act: a.ID, At: e.now(), ID: x.ID, Names: x.Names, Size: x.Size}, a.Module, a.Kept) + e.sayEnd(a.ID, seat.EventAsideDiscarded, seat.AsideDiscardedEvent{Act: a.ID, At: e.now(), ID: x.ID, Names: x.Names, Size: x.Size}, a.Module, a.Kept) return nil }