mesh-recovery: its one whole test against the bus, and an act's end said once it is free

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

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
+}

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)
 }

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()

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
 }