diff --git a/cmd/mesh-controller/busobjects.go b/cmd/mesh-controller/busobjects.go index f7213d32..701692bc 100644 --- a/cmd/mesh-controller/busobjects.go +++ b/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 diff --git a/cmd/mesh-controller/composed_integration_test.go b/cmd/mesh-controller/composed_integration_test.go index ea84d260..2f59c7ab 100644 --- a/cmd/mesh-controller/composed_integration_test.go +++ b/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") + } +} diff --git a/cmd/mesh-controller/composed_rejudge.go b/cmd/mesh-controller/composed_rejudge.go index e550720f..0393123c 100644 --- a/cmd/mesh-controller/composed_rejudge.go +++ b/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) + } + } }) } } diff --git a/cmd/mesh-controller/composed_start.go b/cmd/mesh-controller/composed_start.go index 18838eb0..6b5b0f61 100644 --- a/cmd/mesh-controller/composed_start.go +++ b/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 +} diff --git a/cmd/mesh-controller/composed_start_test.go b/cmd/mesh-controller/composed_start_test.go index ef36ea16..35831c35 100644 --- a/cmd/mesh-controller/composed_start_test.go +++ b/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") + } +} diff --git a/internal/broker/composed.go b/internal/broker/composed.go index 16dc1a37..5506f9a8 100644 --- a/internal/broker/composed.go +++ b/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..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) +} diff --git a/internal/broker/composed_test.go b/internal/broker/composed_test.go index 3ddb231b..3997ac9d 100644 --- a/internal/broker/composed_test.go +++ b/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)) + } +} diff --git a/internal/broker/log_test.go b/internal/broker/log_test.go index ca618ca6..02b72e3a 100644 --- a/internal/broker/log_test.go +++ b/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) }