AS5-5: the whole composition step against a stand-in node-engine and Go build seat

The assigner started whole takes the runner's composition step of a module on three machines, one away: it asks the two others'
engines for their inputs, puts one run ask per machine under its name, the stand-in build seat reads each request by mesh-wants'
input in chunks and says its step-ended with the composer's verdict, and the step's end fails naming the machine that does not
compose, warns of the one away, and keeps no input. The assigner says in its journal when it serves the step and when a machine
goes away or comes back.

Builds: design 47 G §354, §22, §63, §64, §389, A-240, A-252, A-254

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h

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)

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()
+}

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 {