From 8656e57eb96e859a75b287e345c2ae576f9c5173 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 23:39:39 +0100 Subject: [PATCH] Test 36: the recovery wiring is driven through the node agent by a revert want of db MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §296, §297, §554, §560, A-113, A-320 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h --- cmd/mesh-host/recovery_integration_test.go | 141 +++++++++++++++++++-- 1 file changed, 132 insertions(+), 9 deletions(-) diff --git a/cmd/mesh-host/recovery_integration_test.go b/cmd/mesh-host/recovery_integration_test.go index 2f665064..a45939ce 100644 --- a/cmd/mesh-host/recovery_integration_test.go +++ b/cmd/mesh-host/recovery_integration_test.go @@ -10,12 +10,24 @@ import ( "testing" "time" + "fmt" + "git.novox.be/novox/mesh-contracts/go/clients/meshmoduleassignerclient" + nec "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" + "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" "git.novox.be/novox/mesh-contracts/go/seats/meshrecovery" + nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" + "github.com/novox/mesh-host/internal/buckets" "github.com/novox/mesh-host/internal/link" "github.com/novox/mesh-host/internal/nodegrant" + "github.com/novox/mesh-host/internal/selfcompose" + "strings" + "sync" + "sync/atomic" ) // 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 @@ -50,15 +62,31 @@ func TestNatsTheRecoveryStepAsksHearsBoundsAndCatchesUpOverTheBus(t *testing.T) } } var restoreAsked meshrecovery.RestoreKeptInput + // The kept points the stand-in answers: the one of the first checks, then (below) what the agent's revert finds, and none. + var pointsMu sync.Mutex + 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}}}} + // The restore the agent's revert asks waits for the stand-in to say it restored: gated until released. + var gated atomic.Bool + release := make(chan struct{}) + var restores atomic.Int32 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}}}}} + pointsMu.Lock() + defer pointsMu.Unlock() + return meshrecovery.KeptPointsAnswer{Points: points} }, meshrecovery.VerbRestoreKept: func(m *nats.Msg) any { + pointsMu.Lock() _ = json.Unmarshal(m.Data, &restoreAsked) - say("r1", meshrecovery.RestoredEventOutcomeDone) - return meshrecovery.RestoreKeptAnswer{Accepted: "r1"} + pointsMu.Unlock() + act := fmt.Sprintf("r%d", restores.Add(1)) + if gated.Load() { + go func() { <-release; say(act, meshrecovery.RestoredEventOutcomeDone) }() + } else { + say(act, meshrecovery.RestoredEventOutcomeDone) + } + return meshrecovery.RestoreKeptAnswer{Accepted: act} }, } { answer := answer @@ -72,8 +100,14 @@ func TestNatsTheRecoveryStepAsksHearsBoundsAndCatchesUpOverTheBus(t *testing.T) t.Cleanup(func() { _ = sub.Unsubscribe() }) } var toolArgs map[string]any + var toolCalls atomic.Int32 toolSub, err := b.conn.Subscribe(link.ToolSubject("db", "rebuild", itNode), func(m *nats.Msg) { - _ = json.Unmarshal(m.Data, &toolArgs) + var args map[string]any + _ = json.Unmarshal(m.Data, &args) + pointsMu.Lock() + toolArgs = args + pointsMu.Unlock() + toolCalls.Add(1) _ = m.Respond([]byte(`{"result":{}}`)) }) if err != nil { @@ -113,14 +147,21 @@ func TestNatsTheRecoveryStepAsksHearsBoundsAndCatchesUpOverTheBus(t *testing.T) 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) + pointsMu.Lock() + firstAsked := restoreAsked + pointsMu.Unlock() + if err != nil || got.Accepted != "r1" || firstAsked.Snapshot != "snap-1" { + t.Fatalf("restore-kept: %+v, %v, asked %+v", got, err, firstAsked) } 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) + err = r.CatchUp(ctx, "db", *pts.Points[0].CatchUp) + pointsMu.Lock() + gotArgs := toolArgs + pointsMu.Unlock() + if err != nil || gotArgs["all"] != true { + t.Fatalf("catch-up: %v, arguments %v", err, gotArgs) } expired, end, err := r.Bound(ctx, "recovery", "db.restored.cache", "db", time.Now(), time.Second) if err != nil { @@ -133,6 +174,88 @@ func TestNatsTheRecoveryStepAsksHearsBoundsAndCatchesUpOverTheBus(t *testing.T) t.Fatal("the bound's end, the bus's marker, was not heard") } + // Design 47 G §296, §297, §554, §560, A-113: the node agent whole drives a revert want of db through the same recovery wiring + // (recoveryEnv over the session and the node's composer): db runs at a build whose cache is of form 2, and its want goes back to a + // build of form 1. With no point of form 1 taken before the newer build, its have says the restore point is missing, naming it, + // and the newer build runs again. With the point, the have is held (backing up) until the stand-in says restored, then the + // older build is applied and the point's catch-up tool is called. + dbManifest := func(form int, content string) string { + return `{"module":"db","version":"1","state":[{"name":"cache","form":` + fmt.Sprint(form) + `}],"resources":[{"id":"conf","type":"file","path":"` + + filepath.Join(b.dir, "db.conf") + `","mode":"0644","content":"` + content + `\n"}]}` + } + fpOld, fpNew := fpIt("a"), fpIt("b") + b.registerEntry("db", fpOld, dbManifest(1, "v1")) + b.registerEntry("db", fpNew, dbManifest(2, "v2")) + for _, name := range []string{buckets.Manifests, selfcompose.ResolvedBucket, buckets.ProposalCandidates, link.PackagesBucket} { + if _, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: name, History: 5}); err != nil { + t.Fatal(err) + } + } + assignment, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: buckets.Assignments, History: 5}) + if err != nil { + t.Fatal(err) + } + places, _ := meshmoduleassignerclient.AssignmentsValues(meshmoduleassigner.Assignments{Node: itNode, Modules: []meshmoduleassigner.PlacedModule{{Module: "db"}}}) + hkey, _ := nec.HaveNodeKey(itNode) + rec, _ := nec.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: itNode, Capabilities: []string{"service-manager", "package-manager"}}}) + if _, err := b.haves.Put(hkey, rec); err != nil { + t.Fatal(err) + } + for key, value := range places { + if _, err := assignment.Put(key, value); err != nil { + t.Fatal(err) + } + } + b.want("db", fpNew, "", "") + stopAgent := b.start() + defer stopAgent() + within(t, 20*time.Second, "db does not run its newer build", func() bool { + h := b.have("db") + return h.Fp == fpNew && h.Phase == nec.PhaseRunning && b.file("db") == "v2\n" + }) + // The engine says what runs, as it does after each apply: the composer keeps the entry of the running build while the want moves on. + b.running(itNode, "db", fpNew, meshcatalogclient.ManifestDigest([]byte(dbManifest(2, "v2")))) + // The stand-in holds no point of form 1: the revert is held, the have names the point missing, the newer build runs. + pointsMu.Lock() + points = nil + pointsMu.Unlock() + b.want("db", fpOld, "", "") + within(t, 20*time.Second, "the have does not say the restore point is missing", func() bool { + h := b.have("db") + return h.Phase == nec.PhaseHeld && strings.Contains(h.Detail, "restore point missing") && + strings.Contains(h.Detail, "db's cache has none of form 1 taken before "+fpNew) + }) + if b.have("db").Fp != fpNew || b.file("db") != "v2\n" || restores.Load() != 1 { + t.Fatalf("the missing point left db %+v with %q, %d restore(s) asked", b.have("db"), b.file("db"), restores.Load()) + } + // The stand-in now holds the point: the want answered "again" asks the restore, and the have waits for its ending. + pointsMu.Lock() + points = []meshrecovery.Point{{Module: "db", Kept: "cache", Form: 1, Fingerprint: fpNew, Snapshot: "snap-2", At: time.Now(), + CatchUp: &meshrecovery.CatchUp{Tool: "rebuild", Arguments: map[string]any{"all": "again"}}}} + pointsMu.Unlock() + gated.Store(true) + calls := toolCalls.Load() + b.want("db", fpOld, "", meshwants.WantsByWantNodeModuleValueBackupAgain) + within(t, 20*time.Second, "the restore of the revert is not asked", func() bool { return restores.Load() == 2 }) + within(t, 20*time.Second, "the have does not wait for the restore", func() bool { return b.have("db").Phase == nec.PhaseBackingUp }) + time.Sleep(300 * time.Millisecond) // the older build must not be applied while the stand-in has not said restored + if h := b.have("db"); h.Phase != nec.PhaseBackingUp || b.file("db") != "v2\n" || toolCalls.Load() != calls { + t.Fatalf("before the stand-in says restored the have is %+v, the file %q, %d catch-up call(s) more", h, b.file("db"), toolCalls.Load()-calls) + } + close(release) + within(t, 20*time.Second, "the older build is not applied after the restore", func() bool { + h := b.have("db") + return h.Fp == fpOld && h.Phase == nec.PhaseRunning && b.file("db") == "v1\n" + }) + within(t, 20*time.Second, "the catch-up tool is not called after the restore", func() bool { return toolCalls.Load() == calls+1 }) + pointsMu.Lock() + asked, gotArgs := restoreAsked, toolArgs + pointsMu.Unlock() + if asked.Snapshot != "snap-2" || asked.Fingerprint != fpNew || gotArgs["all"] != "again" { + t.Errorf("the revert asked %+v and called the catch-up with %v", asked, gotArgs) + } + stopAgent() + // Design 47 G A-293: over the same session, the builder's build ends the composer follows: the stream read whole first (the // last end of each module and fingerprint, in the stream's order), then the read's end, then each end said after it. if _, err := b.js.AddStream(&nats.StreamConfig{Name: link.BuildEndsStream, Subjects: []string{link.BuildEndsSubject}, MaxMsgsPerSubject: 1}); err != nil {