novox/mesh-host
9
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Test 36: the recovery wiring is driven through the node agent by a revert want of db

Browse Source 
Builds: design 47 G §296, §297, §554, §560, A-113, A-320

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
jschoubben
2026-11-12 23:39:39 +01:00
parent 650910dc9e
commit 8656e57eb9
1 changed files with 132 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-host/recovery_integration_test.go
+132 -9
View File
@@ -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 {