Delete the declarations' stream once every node runs the new node-engine, and assert it no more (issue 1945, B13)

A node away during the switch finds its last declaration on the stream; once every node's have names the
node-engine registered, no node reads one, so the stream and the nodes' consumers on it go, said once, and a raise
leaves them out (ADR 0539 clause 11).

cmd/mesh-controller/busobjects.go

@@ -29,7 +29,13 @@ func assertBusObjects(ctx context.Context, inv *inventory.Inventory, r broker.Ra
 	for _, n := range nodes {
 		names = append(names, n.Name)
 	}
-	if err := broker.Raise(r, names); err != nil {
+	// Once every node runs the node-engine that reads no declaration, the declarations' stream and the nodes'
+	// consumers on it are not asserted again (novox/hq ADR 0539 clause 11; composed_start.go retires them).
+	raise := func() error { return broker.Raise(r, names) }
+	if declarationsRetiredNow(ctx, inv) {
+		raise = func() error { return broker.RaiseAfterDeclarations(r) }
+	}
+	if err := raise(); err != nil {
 		return nil, err
 	}
 	// And each node's bucket for the files of its declaration that travel by digest (novox/hq ADR 0486 §4), so a node finds its own

cmd/mesh-controller/composed_integration_test.go

@@ -154,7 +154,9 @@ type trunkKeys struct {
 	trunk sdkdelivery.Module
 }
 
-func (k trunkKeys) ModuleAt(context.Context, string, string) (sdkdelivery.Module, error) { return k.trunk, nil }
+func (k trunkKeys) ModuleAt(context.Context, string, string) (sdkdelivery.Module, error) {
+	return k.trunk, nil
+}
 
 // 1945/AC3 (and AC1, AC2's half for a new fingerprint): a done key of a new fingerprint of an assigned module is taken
 // in, its composition is written on the node it is assigned to, signed by the controller's key, with its digest beside
@@ -315,3 +317,59 @@ func TestAFreshStartComposesEveryNodeAndDeletesThePlansBucketsOnce(t *testing.T)
 	}
 	m.noDeclarationSent(t)
 }
+
+// 1945/AC14: the declarations' stream stays while one node's have names another node-engine than the one registered,
+// and goes, said once, when every node's names it (ADR 0539 clause 11).
+func TestTheDeclarationsStreamGoesOnceEveryNodeRunsTheNewEngine(t *testing.T) {
+	m := aComposingMesh(t)
+	ctx := t.Context()
+	legacy, err := m.conn.JetStream()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := legacy.AddStream(&nats.StreamConfig{Name: broker.DeclarationsStream, Subjects: []string{"mesh.node.*.declare"},
+		MaxMsgsPerSubject: 1}); err != nil {
+		t.Fatal(err)
+	}
+	// The new node-engine registered at its fingerprint, from its done key.
+	n6 := fpOf('6')
+	keys, trunk := aKeyFor("mesh-host", "6", n6, "novox/mesh-host", "", "f00d0123456789ab")
+	if _, err := liveDoneKeySteps(m.open, trunkKeys{keyBuckets: keys, trunk: trunk}).take(ctx, "mesh-host"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := m.js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: sdkdelivery.HaveBucket, History: sdkdelivery.HaveHistory}); err != nil {
+		t.Fatal(err)
+	}
+	haves, err := m.js.KeyValue(ctx, sdkdelivery.HaveBucket)
+	if err != nil {
+		t.Fatal(err)
+	}
+	have := func(node, fp string) {
+		key, _ := sdkdelivery.HaveKey(node, "mesh-host")
+		raw, _ := sdkdelivery.EncodeHave(sdkdelivery.Have{FP: fp, Phase: sdkdelivery.PhaseRunning, Health: sdkdelivery.HealthOK})
+		if _, err := haves.Put(ctx, key, raw); err != nil {
+			t.Fatal(err)
+		}
+	}
+	have("anchor", n6)
+	have("laptop", fpOf('5'))
+	if done, err := retireDeclarations(ctx, m.open); err != nil || done {
+		t.Fatalf("with laptop on the old node-engine: done %v (%v)", done, err)
+	}
+	if _, err := m.js.Stream(ctx, broker.DeclarationsStream); err != nil {
+		t.Fatalf("the stream went while laptop runs the old node-engine: %v", err)
+	}
+	have("laptop", n6)
+	if done, err := retireDeclarations(ctx, m.open); err != nil || !done {
+		t.Fatalf("with every node on the new node-engine: done %v (%v)", done, err)
+	}
+	if _, err := m.js.Stream(ctx, broker.DeclarationsStream); !errors.Is(err, jetstream.ErrStreamNotFound) {
+		t.Errorf("the stream is still there: %v", err)
+	}
+	if done, err := retireDeclarations(ctx, m.open); err != nil || !done {
+		t.Errorf("a second look: done %v (%v)", done, err)
+	}
+	if !declarationsRetiredNow(ctx, m.open.inventory) {
+		t.Errorf("a raise after the retirement would assert the stream again")
+	}
+}

cmd/mesh-controller/composed_rejudge.go

@@ -156,6 +156,12 @@ func watchHavesForJudgement(ctx context.Context, open *stores) {
 					if err := rejudgeOn(ctx, open, []string{node}, module); err != nil {
 						fmt.Printf("compositions: %s on %s is not judged again: %v\n", module, node, err)
 					}
+					// A node's node-engine moved: the declarations' stream goes once every node runs the new one.
+					if module == nodeEngineModule {
+						if _, err := retireDeclarations(ctx, open); err != nil {
+							fmt.Printf("compositions: whether the declarations' stream may go cannot be read: %v\n", err)
+						}
+					}
 				})
 			}
 		}

cmd/mesh-controller/composed_start.go

@@ -62,7 +62,11 @@ type BucketDeleted struct {
 
 // sayBucketDeleted says a deleted bucket in the log and as the controller's event.
 func sayBucketDeleted(ctx context.Context, bucket string) {
-	why := "a bucket of the plans' state, which release 6 deleted (novox/hq ADR 0539 clause 13)"
+	sayDeleted(ctx, bucket, "a bucket of the plans' state, which release 6 deleted (novox/hq ADR 0539 clause 13)")
+}
+
+// sayDeleted says a deleted bucket or stream, with why, in the log and as the controller's event bucket-deleted.
+func sayDeleted(ctx context.Context, bucket, why string) {
 	fmt.Printf("start: deleted %s: %s\n", bucket, why)
 	bus := checkEvents
 	if bus == nil {
@@ -108,6 +112,84 @@ func composeAtStart(ctx context.Context, open *stores) {
 	if err := composeReplacedController(ctx, open); err != nil {
 		fmt.Printf("start: the controller this one replaced is not composed: %v\n", err)
 	}
+	if _, err := retireDeclarations(ctx, open); err != nil {
+		fmt.Printf("start: whether the declarations' stream may go cannot be read: %v\n", err)
+	}
+}
+
+// nodeEngineModule is the node-engine's module.
+const nodeEngineModule = "mesh-host"
+
+// declarationsDone says every node runs the node-engine at fingerprint current, as its have names: no node still reads
+// a declaration, and none away will need its last one (ADR 0539 clause 11). No node, or no fingerprint of the new
+// form, is not done.
+func declarationsDone(nodes []string, runs map[string]string, current string) bool {
+	if len(nodes) == 0 || !strings.HasPrefix(current, sdkdelivery.SourcePrefix2) {
+		return false
+	}
+	for _, n := range nodes {
+		if runs[n] != current {
+			return false
+		}
+	}
+	return true
+}
+
+// declarationsRetiredOn reads whether the declarations are done on the mesh now: every node's have of the node-engine
+// against the fingerprint it is registered at.
+func declarationsRetiredOn(ctx context.Context, inv *inventory.Inventory, js jetstream.JetStream) (bool, error) {
+	fps, err := currentFingerprints(ctx, inv)
+	if err != nil {
+		return false, err
+	}
+	nodes, err := inv.Nodes(ctx)
+	if err != nil {
+		return false, err
+	}
+	names := make([]string, 0, len(nodes))
+	runs := map[string]string{}
+	for _, n := range nodes {
+		names = append(names, n.Name)
+		h, found, err := haveOn(ctx, js, n.Name, nodeEngineModule)
+		if err != nil {
+			return false, err
+		}
+		if found {
+			runs[n.Name] = h.FP
+		}
+	}
+	return declarationsDone(names, runs, fps[nodeEngineModule]), nil
+}
+
+// retireDeclarations deletes the declarations' stream, with every node's consumer on it, once every node runs the new
+// node-engine, and says it once; a stream already gone is said by nobody. It answers whether the declarations are done.
+func retireDeclarations(ctx context.Context, open *stores) (bool, error) {
+	if _, err := theLease.epoch(ctx); err != nil {
+		return false, err
+	}
+	bus, err := aBus()
+	if err != nil {
+		return false, err
+	}
+	defer bus.Close()
+	js, err := jetstream.New(bus.Conn())
+	if err != nil {
+		return false, err
+	}
+	done, err := declarationsRetiredOn(ctx, open.inventory, js)
+	if err != nil || !done {
+		return done, err
+	}
+	err = js.DeleteStream(ctx, broker.DeclarationsStream)
+	switch {
+	case errors.Is(err, jetstream.ErrStreamNotFound):
+		return true, nil
+	case err != nil:
+		return true, fmt.Errorf("deleting the stream %s: %w", broker.DeclarationsStream, err)
+	}
+	sayDeleted(ctx, broker.DeclarationsStream, "the stream of each node's last declaration: every node runs the new "+
+		"node-engine, which reads none (novox/hq ADR 0539 clause 11)")
+	return true, nil
 }
 
 // replacedController is the controller a control-node ran before this one: its fingerprint and manifest.
@@ -241,3 +323,19 @@ func replacedComposition(in compose.Input, r replacedController) (sdkdelivery.Co
 	}
 	return c, nil
 }
+
+// declarationsRetiredNow says the declarations are done on the mesh now, for a raise of the bus's objects; a bus or a
+// store it cannot read is not done, so the stream is asserted as before.
+func declarationsRetiredNow(ctx context.Context, inv *inventory.Inventory) bool {
+	bus, err := aBus()
+	if err != nil {
+		return false
+	}
+	defer bus.Close()
+	js, err := jetstream.New(bus.Conn())
+	if err != nil {
+		return false
+	}
+	done, err := declarationsRetiredOn(ctx, inv, js)
+	return err == nil && done
+}

cmd/mesh-controller/composed_start_test.go

@@ -91,3 +91,22 @@ func TestTheReplacedControllerIsItsNewestEarlierBuild(t *testing.T) {
 		t.Errorf("a controller with no earlier build replaced one")
 	}
 }
+
+// 1945/AC14 (unit half): the declarations' stream stays while one node's have names another node-engine than the one
+// registered, or none, and goes once every node's names it.
+func TestTheDeclarationsGoOnlyOnceEveryNodeRunsTheNewEngine(t *testing.T) {
+	n6 := sdkdelivery.SourcePrefix2 + strings.Repeat("6", 64)
+	nodes := []string{"ace", "g14", "novox"}
+	if declarationsDone(nodes, map[string]string{"ace": n6, "novox": n6}, n6) {
+		t.Errorf("done while g14 says nothing (it sleeps)")
+	}
+	if declarationsDone(nodes, map[string]string{"ace": n6, "g14": "src2:old", "novox": n6}, n6) {
+		t.Errorf("done while g14 runs the old node-engine")
+	}
+	if !declarationsDone(nodes, map[string]string{"ace": n6, "g14": n6, "novox": n6}, n6) {
+		t.Errorf("not done with every node on the new node-engine")
+	}
+	if declarationsDone(nodes, map[string]string{"ace": "", "g14": "", "novox": ""}, "") || declarationsDone(nil, nil, n6) {
+		t.Errorf("done with no fingerprint registered, or no node")
+	}
+}

internal/broker/composed.go

@@ -121,3 +121,23 @@ func NodeAgentAsks(node string) ([]string, error) {
 // NodeAgentHears are the events a node's node-engine subscribes while it applies a want: the backup holder's word that a
 // backup is taken (`backed-up`), from whichever module holds node-backup on its machine (restic today).
 var NodeAgentHears = []string{"mesh.mod.*.event.backed-up"}
+
+// DeclarationsStream is the stream of each node's last declaration, `mesh.node.<node>.declare` (issue 107). Release 6
+// sends none (ADR 0539 clauses 5 and 11): the stream is kept, with each node's consumer, only so a node away during the
+// switch finds its last declaration and with it the new node-engine; the controller deletes it once every node's have
+// names the new node-engine, and from then on neither asserts it nor any node's consumer on it.
+const DeclarationsStream = "NODES"
+
+// RaiseAfterDeclarations is Raise once the declarations' stream is retired: the mesh's streams and consumers but the
+// declarations' stream, and no node's consumer of a declaration.
+func RaiseAfterDeclarations(r Raiser) error {
+	for _, s := range MeshStreams() {
+		if s.Name == DeclarationsStream {
+			continue
+		}
+		if err := r.EnsureStream(s); err != nil {
+			return fmt.Errorf("asserting stream %s: %w", s.Name, err)
+		}
+	}
+	return AssertMeshConsumers(r)
+}

internal/broker/composed_test.go

@@ -72,3 +72,37 @@ func TestANodeAgentAsksItsOwnBackupHolder(t *testing.T) {
 		t.Errorf("a node-engine of ace reaches another machine: %v", p.Publish)
 	}
 }
+
+// recordingRaiser records what a raise asserts.
+type recordingRaiser struct{ streams, consumers []string }
+
+func (r *recordingRaiser) EnsureStream(s Stream) error {
+	r.streams = append(r.streams, s.Name)
+	return nil
+}
+func (r *recordingRaiser) EnsureConsumer(c Consumer) error {
+	r.consumers = append(r.consumers, c.Stream+"/"+c.Name)
+	return nil
+}
+
+// 1945/AC14 (unit half): once the declarations' stream is retired, a raise asserts every stream of the mesh but it,
+// and no node's consumer of a declaration; before, a raise asserts both (ADR 0539 clause 11).
+func TestARaiseAfterTheDeclarationsAssertsNoneOfThem(t *testing.T) {
+	before := &recordingRaiser{}
+	if err := Raise(before, []string{"ace"}); err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(strings.Join(before.streams, " "), DeclarationsStream) || !strings.Contains(strings.Join(before.consumers, " "), "NODES/ace") {
+		t.Fatalf("before the retirement a raise asserts %v and %v", before.streams, before.consumers)
+	}
+	after := &recordingRaiser{}
+	if err := RaiseAfterDeclarations(after); err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(strings.Join(after.streams, " "), DeclarationsStream) || strings.Contains(strings.Join(after.consumers, " "), "NODES/") {
+		t.Errorf("after the retirement a raise asserts %v and %v", after.streams, after.consumers)
+	}
+	if len(after.streams) != len(before.streams)-1 {
+		t.Errorf("after the retirement %d streams are asserted, before %d", len(after.streams), len(before.streams))
+	}
+}

internal/broker/log_test.go

@@ -201,7 +201,8 @@ func TestNoPathDeletesALogOrABucket(t *testing.T) {
 					strings.Contains(filepath.ToSlash(path), "internal/recordmirror/")
 				// And the buckets of the plans' state release 6 deleted, at the controller's first start, each said (novox/hq
 				// ADR 0539 clause 13): the one decided deletion of a bucket.
-				release6 := strings.HasSuffix(filepath.ToSlash(path), "cmd/mesh-controller/composed_start.go") && name == "DeleteKeyValue"
+				release6 := strings.HasSuffix(filepath.ToSlash(path), "cmd/mesh-controller/composed_start.go") &&
+					(name == "DeleteKeyValue" || name == "DeleteStream") // and the declarations' stream once no node reads one (clause 11)
 				if (removers[name] && !release6) || (name == "Purge" && !decided) {
 					found = append(found, path+": "+name)
 				}