From 75e3a26336d36d614f23a0b88f11a51d316ee04b Mon Sep 17 00:00:00 2001 From: jochen Date: Tue, 3 Nov 2026 13:37:42 +0100 Subject: [PATCH] =?UTF-8?q?The=20integration=20test=20of=20the=20bus=20mod?= =?UTF-8?q?ule's=20first=20want=20shows=20its=20node=20backing=20it=20up?= =?UTF-8?q?=20through=20the=20backup=20seat=20before=20that=20apply=20and?= =?UTF-8?q?=20before=20the=20one=20that=20replaces=20the=20server,=20as=20?= =?UTF-8?q?every=20module=20that=20keeps=20data=20(ADR=200601=20=C2=A72)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-host/nodeagent_integration_test.go | 38 ++++++++++++++++++++- 1 file changed, 37 insertions(+), 1 deletion(-) diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 4243082..ba61810 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -700,6 +700,8 @@ func (r *recordingRuntime) moved(container string) []string { // its changed file and records the server as the bus module's, and asks the container runtime to remove, make, stop, // rename or restart nothing, since the build does not change the server. A later want whose build changes the server // replaces it once, as any module's, once the want carries the yes (§2: the assigner gives it from the release's Approve). +// ADR 0601 §2: the bus module keeps data (its streams), so before each of those applies the node agent asks its node's +// backup seat to back it up, as for every module that keeps data, and applies only once the backup says it is done. func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *testing.T) { b := newItBed(t) runtime := &recordingRuntime{containers: map[string]string{}} @@ -749,7 +751,8 @@ func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *test kvOf(link.PackagesBucket) // The bus module at a build, done: its manifest written and the indexer's module naming it at fp. publish := func(fp, env string) { - man := `{"module":"nats","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}],"resources":[` + + man := `{"module":"nats","version":"1","provides":[{"name":"mesh-bus","scope":"mesh"}],` + + `"data":{"own":[{"id":"jetstream","path":"` + b.dir + `","class":"valuable","why":"the bus's streams"}]},"resources":[` + `{"id":"server-conf","type":"file","path":"` + conf + `","mode":"0644","content":"listen: new\n"},` + `{"id":"server","type":"container","name":"` + server + `","image":"` + image + `"` + env + `}]}` digest := delivery.ManifestDigest([]byte(man)) @@ -790,6 +793,33 @@ func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *test t.Fatal(err) } } + // The node's backup seat: its holder says the bus module has something to back up, starts a backup on now, and says + // when it ended on its event; each ask of now notes what the bus's file held then, so the order is seen. + var backups []string + var backupsMu sync.Mutex + if _, err := b.conn.Subscribe(link.SeatToolSubject("node-backup", "backed-up", itNode), func(m *nats.Msg) { + _ = m.Respond([]byte(`{"result":[{"module":"nats","runs":1}]}`)) + }); err != nil { + t.Fatal(err) + } + if _, err := b.conn.Subscribe(link.SeatToolSubject("node-backup", "now", itNode), func(m *nats.Msg) { + got, _ := os.ReadFile(conf) + backupsMu.Lock() + backups = append(backups, string(got)) + backupsMu.Unlock() + _ = m.Respond([]byte(`{"result":{"started":true}}`)) + go func() { + time.Sleep(20 * time.Millisecond) + _ = b.conn.Publish(backedUpEvent, backupEvent(itNode, "nats", true, time.Now(), "")) + }() + }); err != nil { + t.Fatal(err) + } + backedUp := func() []string { + backupsMu.Lock() + defer backupsMu.Unlock() + return append([]string(nil), backups...) + } first := fpIt("e") publish(first, "") busWant(delivery.Want{FP: first, Self: true}) @@ -809,6 +839,9 @@ func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *test if moved := runtime.moved(server); len(moved) != 0 || runtime.containers[server] != made { t.Fatalf("the first want moved the bus server its build does not change: %v", moved) } + if got := backedUp(); len(got) != 1 || got[0] != "listen: old\n" { + t.Fatalf("the bus module's first apply was not backed up first: the backups saw %q", got) + } state, err := store.Load(b.opts.state) if err != nil { t.Fatal(err) @@ -835,6 +868,9 @@ func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *test runs++ } } + if got := backedUp(); len(got) != 2 || got[1] != "listen: new\n" { + t.Errorf("the replacing apply of the bus module was not backed up first: the backups saw %q", got) + } if runs != 1 || runtime.containers[server] == made { t.Errorf("the build that changes the server made it %d times, its spec still the folded one %v: %v", runs, runtime.containers[server] == made, runtime.moved(server))