diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go index 14acf61b..0aa88e22 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionrun.go @@ -465,6 +465,11 @@ func heardPresence(a *awayNodes, away bool, m jetstream.Msg, logf func(string, . logf("%s holds no machine, left: %v", m.Subject(), err) } else { a.set(e.Machine, away) + if away { + logf("%s is away (machine-away): a composition step does not compose it until its machine-back", e.Machine) + } else { + logf("%s is back (machine-back): a composition step composes it again", e.Machine) + } } if err := m.Ack(); err != nil { logf("acknowledging %s: %v", m.Subject(), err) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_integration_test.go new file mode 100644 index 00000000..a1f4cdf2 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compositionstep_integration_test.go @@ -0,0 +1,235 @@ +//go:build integration + +package main + +import ( + "bytes" + "encoding/json" + "strings" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient/enginereads" + "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" + "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" + meshwants "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/nodebuildgo" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/buildstep" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-sdk/composer" + "git.novox.be/novox/mesh-sdk/go/buscalls" + + "mesh-delivery-assigner/internal/buckets" +) + +// Design 47 G §354, §22, §63, §64, §389, A-240, A-252, A-254: the assigner started whole takes the runner's composition step of a +// module placed on three machines, one of them away: it asks the two others' node-engines for their inputs, puts one run ask per +// machine on the Go build seat's check queue under its name, its program the node-engine build each machine runs; the stand-in +// build seat reads each request by the seat mesh-wants' verb input in chunks and says its step-ended with the composer's verdict; +// the step's step-ended fails naming the machine that does not compose, warns of the machine away, and no input stays kept. +func TestTheCompositionStepComposesEachMachineThroughTheBuildSeat(t *testing.T) { + b := startBed(t, "") + nc, err := nats.Connect(b.srv.ClientURL()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + const engineCommit = "1111111111111111111111111111111111111111" + // What each node runs is the node-engine seat's own stream, kept last per subject (§199): the bed's EVENTS takes every seat's + // events, so here it takes those of the seats this case hears and says, the node-engine's reads changed among them, and not + // what each node runs. + if err := b.js.DeleteStream(b.ctx, "EVENTS"); err != nil { + t.Fatal(err) + } + var seatEvents []string + for _, seat := range []string{"mesh-catalog", "mesh-wants", "mesh-module-assigner", "node-build-go", "mesh-watchdogs", "mesh-delivery", "mesh-pipeline-runner", "mesh-conditions", "operator-channel"} { + seatEvents = append(seatEvents, "mesh.seat."+seat+".event.>") + } + if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: append([]string{"mesh.mod.*.event.>", enginereads.SubjectPrefix + "*"}, seatEvents...)}); err != nil { + t.Fatal(err) + } + // The assigner's durable on EVENTS hears, beside the bed's, the build seat's step-ended and the watchman's machine-away and + // machine-back, as the grants part derives them from its consumes; and what each node runs, kept last per subject. + if _, err := b.js.CreateOrUpdateConsumer(b.ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", + FilterSubjects: []string{enginereads.SubjectPrefix + "*", "mesh.mod." + buckets.IndexerModule + ".event.views.>", meshcatalog.SubjectCatalogRevisionAll, + nodebuildgo.SubjectStepEndedAll, meshwatchdogs.SubjectMachineAwayAll, meshwatchdogs.SubjectMachineBackAll}, AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: nodeengine.StreamRunning, Subjects: []string{nodeengine.SubjectRunningAll}, MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } + if _, err := b.js.CreateOrUpdateConsumer(b.ctx, nodeengine.StreamRunning, jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", AckPolicy: jetstream.AckExplicitPolicy}); err != nil { + t.Fatal(err) + } + for _, node := range []string{"n1", "n2"} { + raw, _ := json.Marshal(nodeengine.Running{V: 1, Node: node, Module: "engine", Fingerprint: "src2:e1", At: time.Now().UTC()}) + if _, err := b.js.Publish(b.ctx, nodeengine.SubjectRunning(node, "engine"), raw); err != nil { + t.Fatal(err) + } + } + away, _ := json.Marshal(meshwatchdogs.MachineAwayEvent{V: 1, Machine: "n3", At: time.Now().UTC(), LastHeard: time.Now().UTC()}) + if _, err := b.js.Publish(b.ctx, meshwatchdogs.SubjectMachineAway("n3"), away); err != nil { + t.Fatal(err) + } + // The trunk's catalog half holds the module and the node-engine, the module claiming the seat node-engine. + engine := meshcatalog.CatalogEntry{Module: "engine", Fp: "src2:e1", Repository: "novox/mesh-host", Commit: engineCommit, Path: ".", + Serves: []string{nodeengine.Seat + ".composition"}, Manifest: map[string]any{"module": "engine", "version": "0.1"}} + b.cat.mu.Lock() + b.cat.entries[trunkBranch] = map[string]meshcatalog.CatalogEntry{"engine": engine, + "m": {Module: "m", Fp: "src2:m1", Repository: "novox/m", Commit: engineCommit, Path: ".", Manifest: map[string]any{"module": "m", "version": "0.1"}}} + b.cat.rev[trunkBranch] = 3 + b.cat.mu.Unlock() + for _, node := range []string{"n1", "n2", "n3"} { + b.place(t, node, "m") + } + // The node-engines of n1 and n2 answer compose-candidate with their inputs; n3 is never asked. + asked := make(chan string, 8) + engines, err := nc.Subscribe("mesh.seat.node-engine.call.compose-candidate.*.*", func(m *nats.Msg) { + node := strings.Split(m.Subject, ".")[5] + asked <- node + raw, _ := json.Marshal(map[string]any{"result": map[string]any{"candidate": "main.3", "node": node, "verdict": "composed", "input": map[string]any{"node": node}, + "modules": []any{}, "own": map[string]any{}, "took": 1, "v": 1}, "node": node}) + _ = m.Respond(raw) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = engines.Unsubscribe() }) + // mesh-catalog's entry answers the node-engine build's entry by its fingerprint. + entries, err := nc.Subscribe("mesh.seat.mesh-catalog.call.entry.*", func(m *nats.Msg) { + raw, _ := json.Marshal(map[string]any{"result": meshcatalog.EntryAnswer{Entry: engine}, "node": "n0"}) + _ = m.Respond(raw) + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = entries.Unsubscribe() }) + subscribed(t, nc) + // The runner hears the step's end. + ended := make(chan meshwants.StepEnded, 1) + ends, err := nc.Subscribe("mesh.seat.mesh-wants.event.step-ended.n1", func(m *nats.Msg) { + var e meshwants.StepEnded + _ = json.Unmarshal(m.Data, &e) + ended <- e + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = ends.Unsubscribe() }) + subscribed(t, nc) + + from := b.journal.mark() + b.start(t) + b.journal.await(t, from, "the composition step is heard on", "the composition step served", 30*time.Second) + b.journal.await(t, from, "n3 is away (machine-away)", "the machine-away of n3 heard", 30*time.Second) + + step, _ := json.Marshal(meshwants.StepRun{Pipeline: "main.3.m", Step: meshwants.StepRunStepComposition, Module: "m", Branch: trunkBranch, + Revision: 3, Commit: engineCommit, Fingerprint: "src2:m1"}) + msg := nats.NewMsg(meshwants.AcceptSubjectStep) + msg.Header.Set("x-event-id", "step-ask-1") + msg.Data = step + if err := nc.PublishMsg(msg); err != nil { + t.Fatal(err) + } + + // The stand-in Go build seat takes the run asks put under the assigner's name, reads each request by the verb input and says + // its end with the composer's verdict: n1 composes, n2 does not. + queue, _ := buildstep.AskStream(buildstep.AskGo) + subject, _ := buildstep.AskSubjectBy(buildstep.AskGo, buildstep.AskCheck, runAsker) + worker, err := b.js.CreateOrUpdateConsumer(b.ctx, queue, jetstream.ConsumerConfig{Durable: "stand-in", FilterSubject: subject, AckPolicy: jetstream.AckExplicitPolicy}) + if err != nil { + t.Fatal(err) + } + programs := map[string]string{} + for range 2 { + got, err := worker.Next(jetstream.FetchMaxWait(30 * time.Second)) + if err != nil { + t.Fatalf("no run ask is put: %v", err) + } + _ = got.Ack() + a, err := buildstep.DecodeStepAsk(got.Data()) + if err != nil || a.Run == nil { + t.Fatalf("the run ask %s does not read: %v", got.Data(), err) + } + req, err := composer.Decode(readInput(t, nc, a.Run.Input)) + if err != nil || len(req.Machines) != 1 { + t.Fatalf("the request of %s: %+v %v", a.ID, req, err) + } + node := req.Machines[0] + programs[node] = a.Run.Program.Commit + m := composer.Machine{Composes: true} + if node == "n2" { + m = composer.Machine{Reason: "two modules own one path", Problems: []string{"/etc/x"}} + } + v, _ := json.Marshal(composer.Verdict{V: composer.Version, Cases: []composer.CaseVerdict{{Name: changeCase, Machines: map[string]composer.Machine{node: m}}}}) + end, _ := json.Marshal(buildstep.StepEnded{V: 1, ID: a.ID, Ask: a.ID, Asker: runAsker, Node: "b1", Outcome: buildstep.OutcomePassed, Output: string(v), At: time.Now().UTC()}) + if _, err := b.js.Publish(b.ctx, nodebuildgo.SubjectStepEnded("b1"), end); err != nil { + t.Fatal(err) + } + } + var e meshwants.StepEnded + select { + case e = <-ended: + case <-b.ctx.Done(): + t.Fatal("the composition step said no end") + } + if e.ID != "step-ask-1" || e.Outcome != meshwants.StepEndedOutcomeFailed || !strings.Contains(e.Said, "n2 does not compose") || + strings.Contains(e.Said, "n1") || len(e.Warnings) != 1 || !strings.Contains(e.Warnings[0], "n3 is away") { + t.Fatalf("the step ended %+v", e) + } + if programs["n1"] != engineCommit || programs["n2"] != engineCommit { + t.Errorf("the composers were built at %v, want the node-engine build each machine runs", programs) + } + close(asked) + for node := range asked { + if node == "n3" { + t.Error("the machine away was asked for its input") + } + } + keys, _ := b.buckets[buckets.Inputs].Keys(b.ctx) + if len(keys) != 0 { + t.Errorf("the inputs stay kept past their step-ended: %v", keys) + } +} + +// readInput reads a run's input as a build seat does: the seat mesh-wants' verb input answers its digest and chunk count, then its +// chunks on the call's subjects, each acknowledged. +func readInput(t *testing.T, nc *nats.Conn, in buildstep.RunInput) []byte { + t.Helper() + call := "standin" + strings.Repeat("0", 16) + chunks := buscalls.SeatChunks{Seat: meshwants.Seat, Verb: meshwants.VerbInput, Call: call} + sub, err := nc.SubscribeSync(chunks.Chunks()) + if err != nil { + t.Fatal(err) + } + defer sub.Unsubscribe() + body, _ := json.Marshal(meshwants.InputInput{Digest: in.Digest, Call: call}) + reply, err := nc.Request("mesh.seat.mesh-wants.tool.input", body, 10*time.Second) + if err != nil { + t.Fatal(err) + } + raw, said, ok := buscalls.Unwrap(reply.Data) + var a meshwants.InputAnswer + if !ok || json.Unmarshal(raw, &a) != nil || a.Chunks < 1 { + t.Fatalf("the verb input answered %s: %s", reply.Data, said) + } + var out bytes.Buffer + next := func() (string, []byte, error) { + m, err := sub.NextMsg(buscalls.SeatAckWithin) + if err != nil { + return "", nil, err + } + return m.Subject, m.Data, nil + } + ack := func(subject string) error { _, err := nc.Request(subject, nil, 10*time.Second); return err } + if _, err := chunks.Receive(a.Chunks, next, ack, &out); err != nil { + t.Fatal(err) + } + if err := in.Verify(out.Bytes()); err != nil { + t.Fatal(err) + } + return out.Bytes() +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 218d4761..560e1b0e 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -644,6 +644,8 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, steps := &compositionSteps{node: thisNode, say: p.publishWants, run: run.run, now: time.Now, logf: logger.Printf, asker: stepAsker} if err := serveCompositionSteps(ctx, nc, steps); err != nil { logger.Printf("the composition step is not heard, so no composition step runs until the assigner starts again: %v", err) + } else { + logger.Printf("the composition step is heard on %s", meshwants.AcceptSubjectStep) } } if bus != nil {