BL17b: a key failed on a transport fault is built again on the events that say its cause passed, counted in event-builds

Builds: design 47 G §568

Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
2026-11-11 23:19:08 +01:00
parent 381b55ef53
commit 2339eb60b8
7 changed files with 429 additions and 4 deletions
@@ -245,6 +245,11 @@ type BuilderBuild struct {
// oci-artifact-registry (design 47 G §671): every artifact of the build is kept there, each with its digest in
// Artifacts. A build not kept names none; the builder carries it onto the build's end (BuildLogEntry.KeptOn).
KeptOn string `json:"kept-on,omitempty"`
// EventBuilds counts, in the key's newest revision, the revisions an event started since the last one a person started
// (design 47 G §568, ADR 0673): the builder builds a key that failed on a transport fault again on an event for the third
// time at most, then asks the operator. A person's build (builder_build_again, the operator's answer) starts the count
// again. A reader that does not know the field reads none.
EventBuilds int `json:"event-builds,omitempty"`
}
func (b BuilderBuild) check() error {
@@ -365,3 +370,18 @@ func DecodeBuilderBuild(data []byte) (BuilderBuild, error) {
// SameBuilderBuild says whether two builds are the same value, so the builder writes only when they differ.
func SameBuilderBuild(a, b BuilderBuild) bool { return reflect.DeepEqual(a, b) }
// TransportFaultOf reads a key's error `transport fault: <class>, <seat> <node>, <cause>` (design 47 G §568, A-76): ok is false
// for an error that does not open so. The node of a fault no node ran (the put of an ask) is `no node`.
func TransportFaultOf(err string) (class, seat, node, cause string, ok bool) {
rest, found := strings.CutPrefix(err, TransportFaultPrefix+" ")
if !found {
return "", "", "", "", false
}
parts := strings.SplitN(rest, ", ", 3)
if len(parts) != 3 {
return "", "", "", "", false
}
seat, node, _ = strings.Cut(parts[1], " ")
return parts[0], seat, node, parts[2], seat != "" && node != ""
}
@@ -233,6 +233,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
Unreleased: reserved.unreleasedWords, Max: republishMax, Logf: logger.Printf}
acts.Republish = republish.Republish
again := &republishing{rp: republish}
retake := &retaking{rt: &build.Retaker{Builder: b, Acted: awaitedReserved{awaitedVersions{b: actedKV, bucket: versions.ActedBucket}}, Entry: entryVia(calls.Call), Logf: logger.Printf}}
queueKV.Ready(again.bound(ctx, js, logger.Printf))
actedKV.Ready(again.bound(ctx, js, logger.Printf))
// A candidate's preview is merged onto the trunk head the indexer previewed it onto: source.main at the candidate's base
@@ -379,6 +380,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if again.heard(ctx, m, logger.Printf) { // the publish again's own subjects, left for the bus to give again when its state cannot take them
return
}
if retake.heard(ctx, m, logger.Printf) { // a key failed on a transport fault, built again when its cause passed (§568)
return
}
if reads.heard(m, logger.Printf) || heardTools.Reported(ctx, m) || heardTools.Running(ctx, m) || floor.heard(ctx, m) || reserved.heard(ctx, m) || seats.heard(ctx, m) { // reads: the node-engine seat's manifest-reads-changed (ADR 0618 §4)
return
}
@@ -0,0 +1,31 @@
package main
import (
"context"
"github.com/nats-io/nats.go/jetstream"
"mesh-delivery-builder/internal/build"
)
// The builder's retake of a key failed on a transport fault, on the node-engine seat's events (design 47 G §568, ADR 0673): the
// wiring of build.Retaker to the events consumer.
type retaking struct{ rt *build.Retaker }
// heard hands a message of the events consumer to the retake: one of its subjects whose state cannot take it is left for the bus to
// give again (Nak); every message goes on to the other parts.
func (g *retaking) heard(ctx context.Context, m jetstream.Msg, logf func(string, ...any)) (stop bool) {
meta, err := m.Metadata()
if err != nil {
return false
}
handled, err := g.rt.Heard(ctx, m.Subject(), m.Data(), meta.Sequence.Stream)
if handled && err != nil {
logf("%v; heard again", err)
if nakErr := m.Nak(); nakErr != nil {
logf("leaving %s for the bus to give again: %v", m.Subject(), nakErr)
}
return true
}
return false
}
@@ -54,6 +54,9 @@ type ClaimOptions struct {
// ByHand is a person's hand act (BuildAgain): a key failed on a transport fault is taken again only by it, or by a builder
// that has just started (TakeOver), never by the builder's own retake (ADR 0664 §3).
ByHand bool
// ByEvent is the builder's retake on an event that says a transport fault's cause passed (design 47 G §568): it takes a key
// failed on a transport fault, as a new revision that counts one more of EventBuilds.
ByEvent bool
// ImageBuilder says whether a node runs a healthy image build module now: a key failed for NoImageBuilderPrefix is taken
// again only then. Nil means none does.
ImageBuilder func() bool
@@ -213,11 +216,18 @@ func (c Claimer) Claim(ctx context.Context, module, fingerprint, manifest string
return nil, nil
case cur.State == builds.BuildFailed && (builds.FailedUntilChanged(cur.Error) || strings.HasPrefix(cur.Error, buildrun.DigestsDifferPrefix)):
return nil, nil
case cur.State == builds.BuildFailed && builds.FailedOnTransportFault(cur.Error) && !opt.ByHand && !opt.TakeOver:
case cur.State == builds.BuildFailed && builds.FailedOnTransportFault(cur.Error) && !opt.ByHand && !opt.TakeOver && !opt.ByEvent:
return nil, nil
case cur.State == builds.BuildFailed && strings.HasPrefix(cur.Error, NoImageBuilderPrefix) && (opt.ImageBuilder == nil || !opt.ImageBuilder()):
return nil, nil
case cur.State == builds.BuildFailed:
building.EventBuilds = cur.EventBuilds // a person's build starts the count again, an event's adds one (design 47 G §568)
switch {
case opt.ByHand:
building.EventBuilds = 0
case opt.ByEvent:
building.EventBuilds++
}
return c.write(ctx, key, building, func(data []byte) (uint64, error) { return c.Store.Update(ctx, key, data, rev) })
case cur.State == builds.BuildQueued && cur.Waits != "":
if opt.Published == nil || !opt.Published(cur.Waits) {
@@ -0,0 +1,226 @@
package build
import (
"context"
"encoding/json"
"errors"
"fmt"
"slices"
"strings"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
"mesh-delivery-builder/internal/versions"
)
// **A key that failed on a transport fault is built again on an event that says its cause passed** (design 47 G §568, ADR 0673
// §1 to §4, which changes ADR 0664 §3): the node-engine seat's `running` of a module whose catalog graph entry claims the build
// seat the failed step ran on, on any node, and its `manifest-reads-changed` of the node the step ran on with `why` `returned`.
// Each message is acted on once, by its stream sequence (state `acted`). A key is built again once per event and only while its
// newest revision is that failure, as a new revision counting one more of EventBuilds; after MostEventBuilds the next event
// starts nothing and asks the operator once.
// MostEventBuilds is how many revisions events start, since the last a person started, before the operator is asked.
const MostEventBuilds = 3
// Retake builds the failed key again on an event: it starts the look of the key's module at its fingerprint, which claims the
// key as an event's (ClaimOptions.ByEvent). started is false when the key is not failed on a transport fault, when the module
// has another fingerprint now, or when a look holds the key; ask is true when MostEventBuilds are spent, with the key as it
// stands and its revision.
func (b *Builder) Retake(ctx context.Context, key string) (started, ask bool, cur builds.BuilderBuild, rev uint64, err error) {
b.init()
module, fp, err := builds.ParseBuilderBuildKey(key)
if err != nil {
return false, false, cur, 0, err
}
value, rev, found, err := b.Claimer.Store.Get(ctx, key)
if err != nil || !found {
return false, false, cur, 0, err
}
if cur, err = builds.DecodeBuilderBuild(value); err != nil {
return false, false, cur, rev, fmt.Errorf("%s holds what is not a build: %w", key, err)
}
if cur.State != builds.BuildFailed || !builds.FailedOnTransportFault(cur.Error) {
return false, false, cur, rev, nil
}
if cur.EventBuilds >= MostEventBuilds {
return false, true, cur, rev, nil
}
b.mu.Lock()
m, known := b.known[module]
if !known || m.Fp2 != fp || m.Manifest == "" || b.inflight[key] || b.hands[key] != nil {
b.mu.Unlock()
return false, false, cur, rev, nil
}
b.inflight[key], b.events[key] = true, true
b.mu.Unlock()
b.logf("%s: failed (%s); built again on an event, build %d of %d since a person's", key, cur.Error, cur.EventBuilds+1, MostEventBuilds)
b.startOnce(ctx, key, module, m)
return true, false, cur, rev, nil
}
// Retaker hears the node-engine seat's events and builds again what they say can pass (Builder.Retake).
type Retaker struct {
Builder *Builder
// Acted is the state acted: the last stream sequence acted on for each subject, and the revision the operator was asked at.
Acted versions.ReserveStore
// Entry reads the catalog graph entry of a fingerprint (the seat mesh-catalog's verb entry).
Entry func(ctx context.Context, fingerprint string) (meshcatalog.CatalogEntry, error)
// Ask asks the operator about a key whose event builds are spent (the seat operator-channel's ask).
Ask func(ctx context.Context, key string, cur builds.BuilderBuild) error
Logf func(string, ...any)
}
func (r *Retaker) logf(f string, a ...any) {
if r.Logf != nil {
r.Logf(f, a...)
}
}
// retakeKey is the acted key of a subject's sequence, apart from the publish again's.
func retakeKey(subject string) string { return versions.ActedKey("retake:" + subject) }
// askedKey is the acted key of the revision of a build key the operator was asked at.
func askedKey(key string) string { return versions.ActedKey("retake-asked:" + key) }
// Heard takes one message of the node-engine seat's running or manifest-reads-changed in, at stream sequence seq, and acts on
// it once. A subject that is neither is not its; an error leaves it not acted on, so it is read again.
func (r *Retaker) Heard(ctx context.Context, subject string, data []byte, seq uint64) (bool, error) {
running := strings.HasPrefix(subject, strings.TrimSuffix(nodeengine.SubjectRunningAll, "*.*"))
changed := strings.HasPrefix(subject, "mesh.seat.node-engine.event."+nodeengine.EventManifestReadsChanged+".")
if !running && !changed {
return false, nil
}
key := retakeKey(subject)
raw, rev, found, err := r.Acted.Get(ctx, key)
if err != nil {
return true, fmt.Errorf("the acted sequence of %s cannot be read: %w", subject, err)
}
if found {
var a versions.Acted
if json.Unmarshal(raw, &a) == nil && a.Sequence >= seq {
return true, nil
}
}
var want func(seat, node string) bool
if running {
want, err = r.runningWant(ctx, data)
} else {
want, err = r.returnedWant(data)
}
if err == nil && want != nil {
err = r.retakeAll(ctx, want)
}
if err != nil {
return true, err
}
val, _ := json.Marshal(versions.Acted{Sequence: seq})
if found {
_, err = r.Acted.Update(ctx, key, val, rev)
} else {
_, err = r.Acted.Create(ctx, key, val)
}
if err != nil && !errors.Is(err, versions.ErrConflict) {
return true, fmt.Errorf("the acted sequence of %s cannot be kept: %w", subject, err)
}
return true, nil
}
// runningWant is what a running says passed: the build seats the running module's catalog graph entry claims, on any node.
func (r *Retaker) runningWant(ctx context.Context, data []byte) (func(seat, node string) bool, error) {
var ev nodeengine.Running
if err := json.Unmarshal(data, &ev); err != nil || ev.Module == "" || ev.Node == "" {
r.logf("a running that does not read is not acted on: %v", err)
return nil, nil
}
if ev.Stopped || ev.Fingerprint == "" {
return nil, nil
}
e, err := r.Entry(ctx, ev.Fingerprint)
if err != nil {
return nil, fmt.Errorf("the catalog entry %s of %s cannot be read: %w", ev.Fingerprint, ev.Module, err)
}
var seats []string
for _, s := range e.Serves {
if seat, _, _ := strings.Cut(s, "."); strings.HasPrefix(seat, "node-build-") && !slices.Contains(seats, seat) {
seats = append(seats, seat)
}
}
if len(seats) == 0 {
return nil, nil
}
return func(seat, _ string) bool { return slices.Contains(seats, seat) }, nil
}
// returnedWant is what a manifest-reads-changed says passed: only one with why returned, for the node it names.
func (r *Retaker) returnedWant(data []byte) (func(seat, node string) bool, error) {
var ev nodeengine.ManifestReadsChanged
if err := json.Unmarshal(data, &ev); err != nil || ev.Node == "" {
r.logf("a manifest-reads-changed that does not read is not acted on: %v", err)
return nil, nil
}
if ev.Why != nodeengine.ManifestReadsChangedWhyReturned {
return nil, nil
}
return func(_, node string) bool { return node == ev.Node }, nil
}
// retakeAll builds again every key failed on a transport fault whose seat and node want says passed.
func (r *Retaker) retakeAll(ctx context.Context, want func(seat, node string) bool) error {
keys, err := r.Builder.Claimer.Store.Keys(ctx)
if err != nil {
return fmt.Errorf("the build keys cannot be listed: %w", err)
}
for _, key := range keys {
raw, _, found, err := r.Builder.Claimer.Store.Get(ctx, key)
if err != nil || !found {
continue
}
v, err := builds.DecodeBuilderBuild(raw)
if err != nil || v.State != builds.BuildFailed {
continue
}
_, seat, node, _, ok := builds.TransportFaultOf(v.Error)
if !ok || !want(seat, node) {
continue
}
started, ask, cur, rev, err := r.Builder.Retake(ctx, key)
if err != nil {
return err
}
if started || !ask || r.Ask == nil {
continue
}
if err := r.askOnce(ctx, key, cur, rev); err != nil {
return err
}
}
return nil
}
// askOnce asks the operator about key once per revision of the key.
func (r *Retaker) askOnce(ctx context.Context, key string, cur builds.BuilderBuild, rev uint64) error {
ak := askedKey(key)
raw, arev, found, err := r.Acted.Get(ctx, ak)
if err != nil {
return err
}
val := []byte(fmt.Sprint(rev))
if found && string(raw) == string(val) {
return nil
}
if err := r.Ask(ctx, key, cur); err != nil {
return fmt.Errorf("the operator cannot be asked about %s: %w", key, err)
}
if found {
_, err = r.Acted.Update(ctx, ak, val, arev)
} else {
_, err = r.Acted.Create(ctx, ak, val)
}
if err != nil && !errors.Is(err, versions.ErrConflict) {
return err
}
return nil
}
@@ -0,0 +1,130 @@
package build
import (
"context"
"encoding/json"
"fmt"
"sync"
"testing"
"git.novox.be/novox/mesh-contracts/go/seats/nodeengine"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
)
type actedStore struct {
mu sync.Mutex
vals map[string][]byte
revs map[string]uint64
}
func (s *actedStore) Get(_ context.Context, k string) ([]byte, uint64, bool, error) {
s.mu.Lock()
defer s.mu.Unlock()
v, ok := s.vals[k]
return v, s.revs[k], ok, nil
}
func (s *actedStore) Create(_ context.Context, k string, v []byte) (uint64, error) {
return s.Update(context.Background(), k, v, 0)
}
func (s *actedStore) Update(_ context.Context, k string, v []byte, _ uint64) (uint64, error) {
s.mu.Lock()
defer s.mu.Unlock()
if s.vals == nil {
s.vals, s.revs = map[string][]byte{}, map[string]uint64{}
}
s.vals[k], s.revs[k] = v, s.revs[k]+1
return s.revs[k], nil
}
func (s *actedStore) KeysOf(context.Context, string) ([]string, error) { return nil, nil }
func (s *actedStore) Delete(_ context.Context, k string) error { delete(s.vals, k); return nil }
// Design 47 G §568, ADR 0673: a key failed on a transport fault is built again on the node-engine seat's events that say its
// cause passed, once per event, counted in event-builds; after three the next event asks the operator once.
func TestRetakeOnAnEvent(t *testing.T) {
r := newRig(t, goBundle)
// The first look creates the key and its note meets the fault; a retake's claim is an update that succeeds and the note after
// it is the second.
fault := fmt.Errorf("the key: %w", context.DeadlineExceeded)
failOnce := func() { r.st.before = func() { r.st.before = func() { r.st.faults = []error{fault} } } }
r.st.faults = []error{fault}
if v := r.consider(t, "modules/m"); !builds.FailedOnTransportFault(v.Error) {
t.Fatalf("the key = %+v, want failed on a transport fault", v)
}
var asked []string
rt := &Retaker{Builder: r.b, Acted: &actedStore{},
Entry: func(context.Context, string) (meshcatalog.CatalogEntry, error) {
return meshcatalog.CatalogEntry{Serves: []string{"node-build-go.steps"}}, nil
},
Ask: func(_ context.Context, key string, _ builds.BuilderBuild) error {
asked = append(asked, key)
return nil
}}
running, _ := json.Marshal(nodeengine.Running{Module: "mesh-build-go", Node: "g14", Fingerprint: "f1"})
started, _ := json.Marshal(nodeengine.ManifestReadsChanged{Node: "ace", Why: "started"})
heard := func(seq uint64, subject string, data []byte) {
t.Helper()
if ok, err := rt.Heard(context.Background(), subject, data, seq); !ok || err != nil {
t.Fatalf("heard %s: %v, %v", subject, ok, err)
}
r.b.Wait()
}
runningSubject := "mesh.seat.node-engine.event.running.g14.mesh-build-go"
readsSubject := "mesh.seat.node-engine.event.manifest-reads-changed.ace"
// A `started` builds nothing, and the same message again builds nothing.
heard(1, readsSubject, started)
if v := valueOf(t, r.st, r.key); v.EventBuilds != 0 || v.State != builds.BuildFailed {
t.Fatalf("a started: the key = %+v", v)
}
for i, ev := range []struct {
seq uint64
subject string
data []byte
}{{2, runningSubject, running}, {3, runningSubject + ".2", running}, {4, runningSubject + ".3", running}} {
failOnce()
heard(ev.seq, ev.subject, ev.data)
if v := valueOf(t, r.st, r.key); v.EventBuilds != i+1 || !builds.FailedOnTransportFault(v.Error) {
t.Fatalf("event %d: the key = %+v, want event-builds %d failed on a transport fault", i+1, v, i+1)
}
}
heard(4, runningSubject+".3", running) // a sequence heard is not acted on again
if v := valueOf(t, r.st, r.key); v.EventBuilds != 3 || len(asked) != 0 {
t.Fatalf("a sequence heard again: the key = %+v, asked %v", v, asked)
}
heard(5, runningSubject, running)
heard(6, runningSubject, running)
if v := valueOf(t, r.st, r.key); v.EventBuilds != 3 || len(asked) != 1 || asked[0] != r.key {
t.Fatalf("the fourth event: the key = %+v, asked %v, want nothing built and the operator asked once", v, asked)
}
// A person's build starts the count again.
if res := r.b.BuildAgain(context.Background(), "m", fp1, "Build once more", "operator", 2e9); res.Refused != "" {
t.Fatalf("built again by hand: %+v", res)
}
r.b.Wait()
if v := valueOf(t, r.st, r.key); v.EventBuilds != 0 {
t.Fatalf("after a person's build: the key = %+v, want event-builds 0", v)
}
}
// A manifest-reads-changed with why returned builds again the keys whose step ran on its node, and no other's.
func TestRetakeOnANodeReturned(t *testing.T) {
r := newRig(t, goBundle)
r.consider(t, "modules/m")
seed(t, r.st, builds.BuilderBuild{State: builds.BuildFailed, By: "novox.builder", Manifest: mdigest('a'),
Error: "transport fault: connection, node-build-go ace, dial tcp: connection refused"})
rt := &Retaker{Builder: r.b, Acted: &actedStore{}}
for seq, c := range []struct {
node, why string
built bool
}{{"g14", "returned", false}, {"ace", "started", false}, {"ace", "returned", true}} {
data, _ := json.Marshal(nodeengine.ManifestReadsChanged{Node: c.node, Why: c.why})
if _, err := rt.Heard(context.Background(), "mesh.seat.node-engine.event.manifest-reads-changed."+c.node, data, uint64(seq+1)); err != nil {
t.Fatal(err)
}
r.b.Wait()
if v := valueOf(t, r.st, r.key); (v.State == builds.BuildDone) != c.built {
t.Fatalf("%s %s: the key = %+v, want built %v", c.node, c.why, v, c.built)
}
}
}
@@ -132,6 +132,8 @@ type Builder struct {
waiting map[string]map[string]chan string
// hands are the requests to build a key again by hand that wait for the claim and its asks (BuildAgain), by build key.
hands map[string]*handWatch
// events are the keys an event asks the builder to build again (Retake), until their look claims them.
events map[string]bool
// released are the keys builder_release is building now, whatever toolchain their fingerprint names (§276, §672).
released map[string]bool
// builds are the ends that withdraw the asks of a key's build in flight, by build key (ADR 0664 §4): a newer fingerprint of
@@ -162,7 +164,7 @@ func (b *Builder) init() {
b.cands, b.via, b.trunkFP = map[string]meshcatalog.CandidatesByGroupValue{}, map[string]via{}, map[string]bool{}
b.known, b.inflight, b.again, b.plans, b.tried = map[string]meshcatalog.ModulesByModuleValue{}, map[string]bool{}, map[string]meshcatalog.ModulesByModuleValue{}, map[string]planned{}, map[string]bool{}
b.published, b.held, b.waiting = map[string]bool{}, map[string]bool{}, map[string]map[string]chan string{}
b.hands = map[string]*handWatch{}
b.hands, b.events = map[string]*handWatch{}, map[string]bool{}
b.builds = map[string]context.CancelCauseFunc{}
})
}
@@ -493,6 +495,8 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
first := !b.tried[key]
b.tried[key] = true
byHand := b.hands[key] != nil
byEvent := b.events[key]
delete(b.events, key)
b.mu.Unlock()
var numbering versions.Numbering
if trunk && planErr == nil && !nothing {
@@ -507,7 +511,7 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
}
// A key built again by hand is taken over as it stands too: one left building by a look that ended without its write
// (issue 1966) is the hand tool's to finish.
cl, err := b.Claimer.Claim(ctx, module, m.Fp2, m.Manifest, ClaimOptions{TakeOver: first || byHand, ByHand: byHand, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished,
cl, err := b.Claimer.Claim(ctx, module, m.Fp2, m.Manifest, ClaimOptions{TakeOver: first || byHand, ByHand: byHand, ByEvent: byEvent, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished,
Stored: b.storedOnce(key)})
if err != nil {
b.logf("%s: %v", module, err)
@@ -576,7 +580,7 @@ func (b *Builder) try(ctx context.Context, module string, m meshcatalog.ModulesB
b.logf("%s: left building, the toolchains moved under it", cl.Key)
return
}
result.By, result.Manifest = b.Claimer.By, m.Manifest
result.By, result.Manifest, result.EventBuilds = b.Claimer.By, m.Manifest, cl.Build.EventBuilds
if len(cl.Gone) > 0 && result.State == builds.BuildDone {
result.Step += " (built again: the artifact store no longer held " + strings.Join(cl.Gone, ", ") + ")"
}