W13.5: the node agent asks mesh-recovery for a restore point before a form change and restores it at a revert

Before a move across a kept data's form change the node agent asks snapshot-kept and waits for the ended of its own act; at a revert it stops the newer build (Env.Stop), asks kept-points, asks restore-kept with the snapshot of the point of the form and fingerprint it reverts from, waits for restored, applies the reverted build, and calls the point's catch-up as the module's tool through the tool runner. The wait is in the have (ask-at, act), bounded by a bounds key of the state bounds (recovery-wait-minutes, 90), and each ending is read as the table of 560 says. The manifest consumes mesh-recovery's ended and restored and invokes its three verbs.

Builds: design 47 G §296, §297, §554, §555, §560, §630, §631, §632, §633, A-77, A-92, A-98, A-108
Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

cmd/mesh-host/nodeagent.go

@@ -90,7 +90,7 @@ func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool
 		return agent.Have{}, false
 	}
 	return agent.Have{Module: module, FP: got.Fp, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail,
-		Sealed: got.SealedAt}, true
+		Sealed: got.SealedAt, Wait: got.Wait}, true
 }
 
 // ReadErr is Read telling a have absent from one the bus could not read (agent.HavesReadErr).
@@ -108,7 +108,7 @@ func (h sessionHaves) ReadErr(ctx context.Context, module string) (agent.Have, b
 		return agent.Have{}, false, err
 	}
 	return agent.Have{Module: module, FP: got.Fp, Composed: got.Composed, Phase: string(got.Phase), Detail: got.Detail,
-		Sealed: got.SealedAt}, true, nil
+		Sealed: got.SealedAt, Wait: got.Wait}, true, nil
 }
 
 func (h sessionHaves) Write(ctx context.Context, x agent.Have) {
@@ -125,7 +125,7 @@ func (h sessionHaves) Write(ctx context.Context, x agent.Have) {
 	}
 	h.session.NoteModule(ctx, link.ModuleFact{Module: x.Module, FP: x.FP, Composed: x.Composed, Phase: phase,
 		Failed: phase == string(nodeengineclient.PhaseFailed), Detail: shorten(x.Detail), Judgement: x.Judgement, Waits: x.Waits,
-		BuildAlone: x.BuildAlone, Kept: x.Kept, Sealed: x.Sealed, PullWait: x.PullWait})
+		BuildAlone: x.BuildAlone, Kept: x.Kept, Sealed: x.Sealed, PullWait: x.PullWait, Wait: x.Wait})
 }
 
 func (h sessionHaves) Removed(ctx context.Context, module string) {
@@ -688,6 +688,7 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, trus
 		go followRunning(run, session, composer, say)
 		apply.UseEngineModule(composer.EngineModule) // the guard of the engine's own trees fixes no name (§527)
 		env.Composed, env.SelfComposed, env.Policy, env.ComposedToday = composer, true, composer.Policy, composer.ComposedToday
+		env.Recovery = recoveryEnv(session, composer)
 		env.WhyUnmade, env.Waiting = composer.WhyUnmade, composer.Waiting
 		keptFetchFn := keptArtifactFetch(composer.KeptBuildOf, session.AskSeatOn, session.KeptObject)
 		keptFetch.Store(&keptFetchFn)

cmd/mesh-host/recovery.go

@@ -0,0 +1,83 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
+	stdio "git.novox.be/novox/mesh-sdk/go"
+
+	"github.com/novox/mesh-host/internal/agent"
+	"github.com/novox/mesh-host/internal/link"
+	"github.com/novox/mesh-host/internal/selfcompose"
+)
+
+// recoveryEnv is what the node agent asks of the recovery module and the manifests to move a module's kept data (design 47 G §296,
+// §297, §554, §560, §630, A-77, A-98, A-108): the seat's verbs asked in the node-engine's own name, the endings heard from the
+// seat's events, the bound of a wait kept as a key of the state bounds, the catch-up called as the module's tool.
+func recoveryEnv(session *link.Session, composer *selfcompose.Composer) *agent.Recovery {
+	ask := func(ctx context.Context, verb string, in, out any) error {
+		raw, err := session.AskMeshSeat(ctx, meshrecovery.Seat, verb, in)
+		if err != nil {
+			return err
+		}
+		return json.Unmarshal(raw, out)
+	}
+	return &agent.Recovery{
+		FormChanges: composer.FormChanges,
+		SnapshotKept: func(ctx context.Context, in meshrecovery.SnapshotKeptInput) (a meshrecovery.SnapshotKeptAnswer, err error) {
+			return a, ask(ctx, meshrecovery.VerbSnapshotKept, in, &a)
+		},
+		RestoreKept: func(ctx context.Context, in meshrecovery.RestoreKeptInput) (a meshrecovery.RestoreKeptAnswer, err error) {
+			return a, ask(ctx, meshrecovery.VerbRestoreKept, in, &a)
+		},
+		KeptPoints: func(ctx context.Context, in meshrecovery.KeptPointsInput) (a meshrecovery.KeptPointsAnswer, err error) {
+			return a, ask(ctx, meshrecovery.VerbKeptPoints, in, &a)
+		},
+		Hear: func(_ context.Context, kind, module, kept string) (<-chan agent.Ending, func(), error) {
+			heard, stop, err := session.HearRecoveryEnding(kind, module, kept)
+			if err != nil {
+				return nil, nil, err
+			}
+			out := make(chan agent.Ending, 64)
+			done := make(chan struct{})
+			go func() {
+				for {
+					select {
+					case ev := <-heard:
+						select {
+						case out <- agent.Ending(ev):
+						case <-done:
+							return
+						}
+					case <-done:
+						return
+					}
+				}
+			}()
+			return out, func() { stop(); close(done) }, nil
+		},
+		Bound: func(ctx context.Context, what, id, module string, since time.Time, d time.Duration) (<-chan struct{}, func(), error) {
+			return session.KeepBound(ctx, what, id, module, since, d)
+		},
+		CatchUp: func(ctx context.Context, module string, c meshrecovery.CatchUp) error {
+			return session.CallTool(ctx, module, c.Tool, c.Arguments)
+		},
+		WaitMinutes: recoveryWaitMinutes(),
+	}
+}
+
+// recoveryWaitMinutes is the manifest's config key recovery-wait-minutes, read through mesh-sdk's stdio.Config; its default is
+// the design's 90 (design 47 G §633).
+func recoveryWaitMinutes() int {
+	raw, err := stdio.Config("recovery-wait-minutes")
+	if err != nil {
+		return agent.RecoveryWaitMinutes
+	}
+	var n int
+	if json.Unmarshal(raw, &n) != nil || n <= 0 {
+		return agent.RecoveryWaitMinutes
+	}
+	return n
+}

cmd/mesh-host/recovery_integration_test.go

@@ -0,0 +1,125 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-contracts/go/seats/meshrecovery"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+	"github.com/novox/mesh-host/internal/link"
+	"github.com/novox/mesh-host/internal/nodegrant"
+)
+
+// Design 47 G §560, §630, A-77, A-92, A-108: the node agent's recovery step over a real bus. Its session asks the seat's verbs, hears
+// the last `restored` already said and the next, keeps a wait's bound as a key of the state bounds whose time ends with the bus's
+// marker, and calls the catch-up as the module's tool with its arguments. The recovery module is a stand-in that answers the verbs and
+// says the events; the grants part composes the node's subjects from the manifest, so the test adds them to the node's grant.
+func TestNatsTheRecoveryStepAsksHearsBoundsAndCatchesUpOverTheBus(t *testing.T) {
+	perms := nodegrant.For(itNode)
+	perms.Publish = append(perms.Publish, "$JS.API.>", "$KV.mesh-host_bounds.>", "mesh.seat.mesh-recovery.call.*.node~"+itNode,
+		"mesh.mod.db.tool.rebuild."+itNode)
+	perms.Subscribe = append(perms.Subscribe, "mesh.seat.mesh-recovery.event.>", "$KV.mesh-host_bounds.>", "_INBOX.>", "$JS.ACK.>", "$JS.FC.>")
+	b := newItBedAs(t, perms)
+	ctx := context.Background()
+
+	for _, s := range []struct{ stream, subject string }{{meshrecovery.StreamEnded, meshrecovery.SubjectEndedAll},
+		{meshrecovery.StreamRestored, meshrecovery.SubjectRestoredAll}} {
+		if _, err := b.js.AddStream(&nats.StreamConfig{Name: s.stream, Subjects: []string{s.subject}, MaxMsgsPerSubject: 1}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	js, err := jetstream.New(b.conn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: link.BoundsBucket, History: 1, LimitMarkerTTL: time.Second}); err != nil {
+		t.Fatal(err)
+	}
+	say := func(act, outcome string) {
+		raw, _ := json.Marshal(meshrecovery.RestoredEvent{Act: act, Module: "db", Kept: "cache", Outcome: outcome, At: time.Now().UTC()})
+		if _, err := b.js.Publish(meshrecovery.SubjectRestored("db", "cache"), raw); err != nil {
+			t.Fatal(err)
+		}
+	}
+	var restoreAsked meshrecovery.RestoreKeptInput
+	for verb, answer := range map[string]func(m *nats.Msg) any{
+		meshrecovery.VerbKeptPoints: func(*nats.Msg) any {
+			return meshrecovery.KeptPointsAnswer{Points: []meshrecovery.Point{{Module: "db", Kept: "cache", Form: 1, Fingerprint: "f2", Snapshot: "snap-1",
+				CatchUp: &meshrecovery.CatchUp{Tool: "rebuild", Arguments: map[string]any{"all": true}}}}}
+		},
+		meshrecovery.VerbRestoreKept: func(m *nats.Msg) any {
+			_ = json.Unmarshal(m.Data, &restoreAsked)
+			say("r1", meshrecovery.RestoredEventOutcomeDone)
+			return meshrecovery.RestoreKeptAnswer{Accepted: "r1"}
+		},
+	} {
+		answer := answer
+		sub, err := b.conn.Subscribe("mesh.seat.mesh-recovery.call."+verb+".*", func(m *nats.Msg) {
+			out, _ := json.Marshal(map[string]any{"result": answer(m)})
+			_ = m.Respond(out)
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = sub.Unsubscribe() })
+	}
+	var toolArgs map[string]any
+	toolSub, err := b.conn.Subscribe(link.ToolSubject("db", "rebuild", itNode), func(m *nats.Msg) {
+		_ = json.Unmarshal(m.Data, &toolArgs)
+		_ = m.Respond([]byte(`{"result":{}}`))
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = toolSub.Unsubscribe() })
+
+	say("old", meshrecovery.RestoredEventOutcomeFailed) // said before anyone listens: the last message of the subject
+	session := b.startSession()
+	r := recoveryEnv(session, nil)
+
+	events, stop, err := r.Hear(ctx, "restored", "db", "cache")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer stop()
+	if first := <-events; first.Act != "old" || first.Outcome != "failed" {
+		t.Fatalf("the first event heard is %+v, want the last one said before", first)
+	}
+	pts, err := r.KeptPoints(ctx, meshrecovery.KeptPointsInput{Module: "db", Kept: "cache"})
+	if err != nil || len(pts.Points) != 1 {
+		t.Fatalf("kept-points: %+v, %v", pts, err)
+	}
+	got, err := r.RestoreKept(ctx, meshrecovery.RestoreKeptInput{Module: "db", Kept: "cache", Form: 1, Fingerprint: "f2", Snapshot: pts.Points[0].Snapshot, Why: "w"})
+	if err != nil || got.Accepted != "r1" || restoreAsked.Snapshot != "snap-1" {
+		t.Fatalf("restore-kept: %+v, %v, asked %+v", got, err, restoreAsked)
+	}
+	if ev := <-events; ev.Act != "r1" || ev.Outcome != "done" {
+		t.Fatalf("the ending heard is %+v", ev)
+	}
+	if err := r.CatchUp(ctx, "db", *pts.Points[0].CatchUp); err != nil || toolArgs["all"] != true {
+		t.Fatalf("catch-up: %v, arguments %v", err, toolArgs)
+	}
+	expired, end, err := r.Bound(ctx, "recovery", "db.restored.cache", "db", time.Now(), time.Second)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer end()
+	select {
+	case <-expired:
+	case <-time.After(10 * time.Second):
+		t.Fatal("the bound's end, the bus's marker, was not heard")
+	}
+}
+
+// startSession is a session on the node's own connection, with no node agent run on it.
+func (b *itBed) startSession() *link.Session {
+	b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second,
+		Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }}
+	b.session.Attach(context.Background(), link.OverNATS{Conn: b.nodeConn, JS: b.nodeJS, Node: itNode})
+	return b.session
+}