A-97: the builder takes the runner's build step ask on mesh-builds and answers its end

The seat mesh-builds' accepted kind step (cherry-picked contract 0dc8c723 of feat/47-G-recover-runner) names an entry by fingerprint; the builder reads it through the catalog half or the seat mesh-catalog's entry, builds it through its one claim, plan and lanes (a build per fingerprint, no pull request number) and says the end as step-ended.<node> under the ask's id, naming each entry as asked. A build already done is passed at once; the end of a build wakes its waiting step through the log entry. Builds: design 47 G §230, §194, §473, A-84, A-91, A-97, A-99 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
2026-11-12 00:07:53 +01:00
parent 009c4c6e73
commit 8160dc8d26
8 changed files with 399 additions and 1 deletions
@@ -138,7 +138,7 @@ func (h *catalogHolder) buildSeats(t *testing.T, nc *nats.Conn, holders map[stri
 	}
 	t.Cleanup(func() { _ = s.Unsubscribe() })
 	for seat, module := range holders {
-		h.put(t, meshcatalog.CatalogEntry{Module: module, Repository: "novox/mesh-catalog", Commit: "c0", Fp: "src2:" + strings.Repeat(fmt.Sprintf("%x", len(seat)%16), 64), ManifestDigest: "sha256:claims-" + module,
+		h.put(t, meshcatalog.CatalogEntry{Module: module, Repository: "novox/mesh-catalog", Commit: "c0", Fp: "src2:" + strings.Repeat(fmt.Sprintf("%x", 10+len(seat)%6), 64), ManifestDigest: "sha256:claims-" + module,
 			Manifest: map[string]any{"module": module, "claims": []any{map[string]any{"name": seat}}}})
 	}
 }
@@ -390,6 +390,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// one taker, so no pattern here covers them).
 	// The seat list over the connection and heard on the hub: a change of it, as of the catalog half's holders, names the build
 	// modules again (design 47 G A-82, §204).
+	// The runner's build step ask, the builder's one entry point for a candidate (design 47 G A-97).
+	b.StepEntry = newEntryWaits(branches, half.manifests).entry
+	if err := serveSteps(ctx, conn, js, cred, b, logger.Printf); err != nil {
+		return err
+	}
 	seatGate.Follow(seatlist.ProcessBus(calls.Call, hub), func() { seatsChanged(ctx) }, logger.Printf)
 	half.holders = func() { seatsChanged(ctx) }
 	for _, pattern := range seatEventPatterns {
@@ -0,0 +1,101 @@
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"mesh-delivery-builder/internal/build"
+)
+
+// **The runner's build step ask is the builder's one entry point for a candidate** (design 47 G A-97, A-84, A-99): the seat
+// mesh-builds' accepted kind `step`, heard on its subject, names the entry of one module by fingerprint; the builder reads it through
+// the seat mesh-catalog's `entry`, builds it (build.Builder.BuildStep) and says the end as the seat's event step-ended.<node>, the
+// ask's id as the event's id, which the runner holds the end to.
+
+// entryWaits answers an entry by fingerprint from the catalog half or the seat's `entry`, waiting for the answer the entries
+// keeper is told of (meshcatalogclient.Entries.Changed), never a timer.
+type entryWaits struct {
+	branches *meshcatalogclient.Branches
+	manifest *entryManifests
+	mu       sync.Mutex
+	waiting  map[string][]chan meshcatalogclient.EntryRow
+}
+
+func newEntryWaits(branches *meshcatalogclient.Branches, manifests *entryManifests) *entryWaits {
+	w := &entryWaits{branches: branches, manifest: manifests, waiting: map[string][]chan meshcatalogclient.EntryRow{}}
+	prior := branches.Entries.Changed
+	branches.Entries.Changed = func(row meshcatalogclient.EntryRow) {
+		w.mu.Lock()
+		chans := w.waiting[row.Fingerprint]
+		delete(w.waiting, row.Fingerprint)
+		w.mu.Unlock()
+		for _, ch := range chans {
+			ch <- row
+		}
+		if prior != nil {
+			prior(row)
+		}
+	}
+	return w
+}
+
+// entry is build.Builder.StepEntry.
+func (w *entryWaits) entry(ctx context.Context, fingerprint string) (meshcatalog.CatalogEntry, error) {
+	ch := make(chan meshcatalogclient.EntryRow, 1)
+	w.mu.Lock()
+	w.waiting[fingerprint] = append(w.waiting[fingerprint], ch)
+	w.mu.Unlock()
+	e, state := w.branches.Resolve(ctx, fingerprint)
+	if state == meshcatalogclient.EntryAsked {
+		select {
+		case row := <-ch:
+			if row.Entry == nil {
+				return meshcatalog.CatalogEntry{}, fmt.Errorf("the seat mesh-catalog holds no entry of it (%s)", row.State)
+			}
+			e, state = *row.Entry, row.State
+		case <-ctx.Done():
+			return meshcatalog.CatalogEntry{}, ctx.Err()
+		}
+	}
+	if state != meshcatalogclient.EntryAnswered {
+		return meshcatalog.CatalogEntry{}, fmt.Errorf("the seat mesh-catalog's entry of it is %s", state)
+	}
+	return e, w.manifest.keep(e) // the manifest the build is planned from
+}
+
+// serveSteps hears the step asks on the seat's accept subject and answers each one's end, in its own goroutine.
+func serveSteps(ctx context.Context, conn *nats.Conn, js jetstream.JetStream, cred process.Account, b *build.Builder, logf func(string, ...any)) error {
+	sub, err := conn.Subscribe(meshbuilds.AcceptSubjectStep, func(m *nats.Msg) {
+		id := m.Header.Get("x-event-id")
+		if id == "" {
+			id = m.Header.Get(nats.MsgIdHdr)
+		}
+		var run meshbuilds.StepRun
+		if err := json.Unmarshal(m.Data, &run); err != nil || id == "" || run.Step != meshbuilds.StepRunStepBuild {
+			logf("a step ask on %s that is no build step with an id is left: %v", m.Subject, err)
+			return
+		}
+		go func() {
+			end := b.BuildStep(ctx, run)
+			end.ID = id
+			if err := emitSeatEvent(context.WithoutCancel(ctx), js, cred, meshbuilds.SubjectStepEnded(cred.Node), id, end); err != nil {
+				logf("saying the end of the build step %s: %v", id, err)
+			}
+		}()
+	})
+	if err != nil {
+		return err
+	}
+	go func() { <-ctx.Done(); _ = sub.Drain() }()
+	return nil
+}
@@ -0,0 +1,52 @@
+//go:build integration
+
+package main
+
+import (
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
+)
+
+// The runner's build step ask on the seat mesh-builds' accept subject, naming an entry by its fingerprint, is built through the
+// module's one claim and answered as the seat's step-ended.<node> under the ask's id (design 47 G A-97, A-99).
+func TestTheRunnersBuildStepAskIsAnsweredAsStepEnded(t *testing.T) {
+	b := startBed(t)
+	b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck")
+	sub, err := nats.Connect(b.srvURL, nats.UserInfo("ctl", "pw"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer sub.Close()
+	ended := make(chan *nats.Msg, 4)
+	s, err := sub.ChanSubscribe(meshbuilds.SubjectStepEndedAll, ended)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer s.Unsubscribe()
+	b.start(t, "n1", nil)
+	b.await(t, "netcheck", isState(builds.BuildDone))
+	run := meshbuilds.StepRun{Pipeline: "main.1.netcheck", Step: meshbuilds.StepRunStepBuild, Branch: "main", Revision: 1,
+		Entries: []meshbuilds.BuildEntry{{Module: "netcheck", Fingerprint: fp}}}
+	data, _ := json.Marshal(run)
+	msg := nats.NewMsg(meshbuilds.AcceptSubjectStep)
+	msg.Data = data
+	msg.Header.Set("x-event-id", "ask-1")
+	if err := sub.PublishMsg(msg); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case m := <-ended:
+		var end meshbuilds.StepEnded
+		if err := json.Unmarshal(m.Data, &end); err != nil || end.ID != "ask-1" || end.Outcome != meshbuilds.StepEndedOutcomePassed || end.Pipeline != "main.1.netcheck" || m.Subject != meshbuilds.SubjectStepEnded("n1") {
+			t.Fatalf("step-ended on %s = %+v, %v; want the ask ask-1 passed on n1", m.Subject, end, err)
+		}
+	case <-time.After(20 * time.Second):
+		t.Fatal("no step-ended within 20 s")
+	}
+}
@@ -41,6 +41,8 @@ type ToolchainEntries interface {
 // change of the toolchains.
 type Builder struct {
 	Claimer Claimer
+	// StepEntry reads the entry of a fingerprint through the seat mesh-catalog's entry (design 47 G A-84, A-97).
+	StepEntry func(ctx context.Context, fingerprint string) (meshcatalog.CatalogEntry, error)
 	// Seats finds the build modules: the holders of the build seats on the seat list and the catalog half (design 47 G A-82).
 	Seats *buildrun.BuildSeats
 	// Asks puts each build module's part of a build as one ask on its build seat's stream and waits for its step (novox/hq
@@ -129,6 +131,7 @@ type Builder struct {
 	inflight  map[string]bool                               // by build key
 	again     map[string]meshcatalog.ModulesByModuleValue   // by build key: the newest value that arrived while the key was being built
 	plans     map[string]planned                            // by manifest digest
+	stepWaits map[string][]chan builds.BuildLogEntry        // by build key: the step asks waiting for its end (step.go)
 	cands     map[string]meshcatalog.CandidatesByGroupValue // by candidates key (candidates.go)
 	via       map[string]via                                // by build key: a candidate's fetch (candidates.go)
 	// outOfStep are, by build key, the out-of-step files a build's lanes answered, until its end entry takes them.
@@ -1202,6 +1205,7 @@ func (b *Builder) now() time.Time {

 // logBuild tells the log an entry, said in the journal when it cannot be one.
 func (b *Builder) logBuild(e builds.BuildLogEntry) {
+	b.stepEnded(e)
 	if b.Log == nil {
 		return
 	}
@@ -0,0 +1,156 @@
+package build
+
+import (
+	"context"
+	"fmt"
+	"strings"
+
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
+)
+
+// **A candidate's build is the runner's step ask** (design 47 G A-97, A-84, A-99): the runner's asking half asks the seat
+// mesh-builds a `build` step naming each entry by fingerprint, and the builder, its one entry point for a candidate, reads each
+// entry through the seat mesh-catalog's `entry` (the repository and the commit are the member's head), builds it as it builds any
+// fingerprint (the existing claim, plan and lanes) and answers the step's end. A build is per fingerprint, so no pull request
+// number is needed, and the builder follows no candidate.
+
+// BuildStep builds the entries of one step ask and answers its end: passed when every entry's build is done, failed when one
+// failed (said), could not run when an entry could not be read, nothing to run for an ask naming none. The ask's id and the node
+// are the caller's to set.
+func (b *Builder) BuildStep(ctx context.Context, run meshbuilds.StepRun) meshbuilds.StepEnded {
+	b.init()
+	started := b.now()
+	end := meshbuilds.StepEnded{V: 1, Pipeline: run.Pipeline, Step: meshbuilds.StepEndedStepBuild, Rerun: run.Rerun, By: run.By, Entries: run.Entries, Node: b.Node, Started: &started}
+	finish := func(outcome, said string) meshbuilds.StepEnded {
+		at := b.now()
+		end.Ended, end.Outcome, end.Said = &at, outcome, said
+		return end
+	}
+	if len(run.Entries) == 0 {
+		return finish(meshbuilds.StepEndedOutcomeNothingToRun, "the ask names no entry")
+	}
+	if b.StepEntry == nil {
+		return finish(meshbuilds.StepEndedOutcomeCouldNotRun, "the builder reads no entry of the seat mesh-catalog")
+	}
+	var failed []string
+	for _, e := range run.Entries {
+		outcome, said := b.buildEntry(ctx, e)
+		switch outcome {
+		case meshbuilds.StepEndedOutcomeCouldNotRun:
+			return finish(outcome, said)
+		case meshbuilds.StepEndedOutcomeFailed:
+			failed = append(failed, said)
+		}
+	}
+	if len(failed) > 0 {
+		return finish(meshbuilds.StepEndedOutcomeFailed, strings.Join(failed, "; "))
+	}
+	return finish(meshbuilds.StepEndedOutcomePassed, "")
+}
+
+// buildEntry builds one entry at its fingerprint and waits for its end.
+func (b *Builder) buildEntry(ctx context.Context, e meshbuilds.BuildEntry) (outcome, said string) {
+	entry, err := b.StepEntry(ctx, e.Fingerprint)
+	if err != nil {
+		return meshbuilds.StepEndedOutcomeCouldNotRun, fmt.Sprintf("the entry %s at %s cannot be read: %v", e.Module, e.Fingerprint, err)
+	}
+	if entry.Module != e.Module || (e.Commit != "" && entry.Commit != e.Commit) {
+		return meshbuilds.StepEndedOutcomeCouldNotRun, fmt.Sprintf("the entry of %s at %s is of %s at %s", e.Module, e.Fingerprint, entry.Module, entry.Commit)
+	}
+	key, err := builds.BuilderBuildKey(entry.Module, entry.Fp)
+	if err != nil {
+		return meshbuilds.StepEndedOutcomeCouldNotRun, err.Error()
+	}
+	ended := b.awaitEnd(key)
+	defer b.dropEnd(key, ended)
+	if done, failedWords, found := b.keyEnd(ctx, key); found {
+		if done {
+			return meshbuilds.StepEndedOutcomePassed, ""
+		}
+		return meshbuilds.StepEndedOutcomeFailed, failedWords
+	}
+	m := meshcatalog.ModulesByModuleValue{Repository: entry.Repository, Commit: entry.Commit, Fp2: entry.Fp, Manifest: entry.ManifestDigest, Path: entry.Path,
+		Toolchains: CatalogToolchains(b.agreedFor(ctx, entry.ManifestDigest)), Packages: entry.Packages}
+	b.mu.Lock()
+	if _, onTrunk := b.trunkFP[key]; !onTrunk {
+		b.via[key] = via{}
+	}
+	b.mu.Unlock()
+	b.consider(ctx, entry.Module, m)
+	select {
+	case log := <-ended:
+		if log.Outcome == builds.BuildDone {
+			return meshbuilds.StepEndedOutcomePassed, ""
+		}
+		return meshbuilds.StepEndedOutcomeFailed, fmt.Sprintf("%s at %s: %s %s", log.Module, log.FP, log.Outcome, log.Error)
+	case <-ctx.Done():
+		return meshbuilds.StepEndedOutcomeCouldNotRun, fmt.Sprintf("%s at %s: the builder stopped before its build ended", entry.Module, entry.Fp)
+	}
+}
+
+// keyEnd reads the build key: whether it is done, and its error when it failed for good; found only for a key that has ended.
+func (b *Builder) keyEnd(ctx context.Context, key string) (done bool, failed string, found bool) {
+	raw, _, ok, err := b.Claimer.Store.Get(ctx, key)
+	if err != nil || !ok {
+		return false, "", false
+	}
+	v, err := builds.DecodeBuilderBuild(raw)
+	if err != nil {
+		return false, "", false
+	}
+	switch {
+	case v.State == builds.BuildDone && !Unpublished(v, b.Seats):
+		return true, "", true
+	case v.State == builds.BuildFailed && builds.FailedUntilChanged(v.Error):
+		return false, key + ": " + v.Error, true
+	}
+	return false, "", false
+}
+
+// awaitEnd registers a wait for the end of the build of key, heard at its log entry (logBuild).
+func (b *Builder) awaitEnd(key string) chan builds.BuildLogEntry {
+	ch := make(chan builds.BuildLogEntry, 1)
+	b.mu.Lock()
+	if b.stepWaits == nil {
+		b.stepWaits = map[string][]chan builds.BuildLogEntry{}
+	}
+	b.stepWaits[key] = append(b.stepWaits[key], ch)
+	b.mu.Unlock()
+	return ch
+}
+
+func (b *Builder) dropEnd(key string, ch chan builds.BuildLogEntry) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	for i, c := range b.stepWaits[key] {
+		if c == ch {
+			b.stepWaits[key] = append(b.stepWaits[key][:i], b.stepWaits[key][i+1:]...)
+			break
+		}
+	}
+	if len(b.stepWaits[key]) == 0 {
+		delete(b.stepWaits, key)
+	}
+}
+
+// stepEnded tells the step asks waiting for a build that its end was logged.
+func (b *Builder) stepEnded(e builds.BuildLogEntry) {
+	if e.Event != builds.BuildEndedEvent {
+		return
+	}
+	key, err := builds.BuilderBuildKey(e.Module, e.FP)
+	if err != nil {
+		return
+	}
+	b.mu.Lock()
+	chans := append([]chan builds.BuildLogEntry(nil), b.stepWaits[key]...)
+	b.mu.Unlock()
+	for _, ch := range chans {
+		select {
+		case ch <- e:
+		default:
+		}
+	}
+}
@@ -0,0 +1,76 @@
+package build
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"testing"
+	"time"
+
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
+	"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+)
+
+// The runner's build step ask is the builder's one entry point for a candidate (design 47 G A-97): each entry is read by its
+// fingerprint, built through the one claim, and the step's end says passed only for a build that is done.
+func TestABuildStepAnswersItsEnd(t *testing.T) {
+	r := newRig(t, catalogue.Artifact{Name: "files", Kind: "archive"})
+	entryOf := func(fp, commit string) meshcatalog.CatalogEntry {
+		return meshcatalog.CatalogEntry{Module: "m", Repository: "novox/m", Commit: commit, Fp: fp, ManifestDigest: mdigest('a'), Path: "."}
+	}
+	r.b.StepEntry = func(_ context.Context, fp string) (meshcatalog.CatalogEntry, error) {
+		if fp == fp1 {
+			return entryOf(fp1, "c1"), nil
+		}
+		return meshcatalog.CatalogEntry{}, errors.New("not found")
+	}
+	ask := func(entries ...meshbuilds.BuildEntry) meshbuilds.StepEnded {
+		return r.b.BuildStep(t.Context(), meshbuilds.StepRun{Pipeline: "main.1.m", Step: meshbuilds.StepRunStepBuild, Entries: entries})
+	}
+	if end := ask(); end.Outcome != meshbuilds.StepEndedOutcomeNothingToRun {
+		t.Errorf("an ask naming no entry: %+v", end)
+	}
+	if end := ask(meshbuilds.BuildEntry{Module: "m", Fingerprint: "src2:" + strings.Repeat("2", 64)}); end.Outcome != meshbuilds.StepEndedOutcomeCouldNotRun || !strings.Contains(end.Said, "cannot be read") {
+		t.Errorf("an entry the seat does not hold: %+v", end)
+	}
+	if end := ask(meshbuilds.BuildEntry{Module: "m", Fingerprint: fp1, Commit: "other"}); end.Outcome != meshbuilds.StepEndedOutcomeCouldNotRun {
+		t.Errorf("an entry at another commit: %+v", end)
+	}
+	// A build already done is passed at once, naming each entry as it was asked.
+	done, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: mdigest('a')})
+	if _, err := r.st.Create(t.Context(), r.key, done); err != nil {
+		t.Fatal(err)
+	}
+	entries := []meshbuilds.BuildEntry{{Module: "m", Fingerprint: fp1, Commit: "c1"}}
+	if end := ask(entries...); end.Outcome != meshbuilds.StepEndedOutcomePassed || end.Step != meshbuilds.StepEndedStepBuild || len(end.Entries) != 1 || end.Entries[0].Commit != "c1" || end.Pipeline != "main.1.m" {
+		t.Errorf("a done build: %+v", end)
+	}
+}
+
+// A step ask waits for the end of its build without a timer: the log entry of that end wakes it (design 47 G A-97).
+func TestAStepAskIsWokenByItsBuildsEnd(t *testing.T) {
+	r := newRig(t)
+	key := r.key
+	ch := r.b.awaitEnd(key)
+	r.b.stepEnded(builds.BuildLogEntry{Event: builds.BuildStartedEvent, Module: "m", FP: fp1}) // a start wakes nothing
+	select {
+	case <-ch:
+		t.Fatal("a build's start woke the wait")
+	default:
+	}
+	r.b.stepEnded(builds.BuildLogEntry{Event: builds.BuildEndedEvent, Module: "m", FP: fp1, Outcome: builds.BuildDone})
+	select {
+	case e := <-ch:
+		if e.Outcome != builds.BuildDone {
+			t.Errorf("end = %+v", e)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("the end of the build did not wake the wait")
+	}
+	r.b.dropEnd(key, ch)
+	if len(r.b.stepWaits) != 0 {
+		t.Errorf("waits left: %v", r.b.stepWaits)
+	}
+}
@@ -260,6 +260,10 @@ func TestServesNoRequestSubjectAndHoldsNoTicker(t *testing.T) { // 1607/AC6 (ADR
 				return err
 			}
 			for _, f := range forbidden {
+				// The one subscription is the seat mesh-builds' accepted kind `step`, the runner's build ask (design 47 G A-97).
+				if f == "Subscribe(" && filepath.ToSlash(path) == "cmd/mesh-delivery-builder/step.go" {
+					continue
+				}
 				if strings.Contains(string(src), f) {
 					t.Errorf("%s holds %s: the builder serves no request subject and holds no timer", path, f)
 				}