Builder runner: the steps, asks and toolchain agreement are the Delivery SDK's, the toolchains the build seat's and the presence the watchdogs seat's types, the haves and placements their clients'; the buckets it reads are its own constants (D-11; ADR 0634 §1, §2; issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
2026-11-05 16:14:18 +01:00
parent 8448315b72
commit 6b610e1657
18 changed files with 217 additions and 185 deletions
+42 -42
View File
@@ -10,7 +10,7 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// **An asker puts an ask on its build seat's stream and waits for its step** (novox/hq ADR 0525 §1, §6; issue 1887 A4,
@@ -25,7 +25,7 @@ type AskPublisher interface {
// AskWaiter watches an ask's steps: every value a key `*.<ask id>` of the build module's steps state takes, running or
// ended, on the channel it answers, until ctx ends. A watch, not a poll.
type AskWaiter interface {
WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error)
WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error)
}
// Asker puts asks and waits for their steps.
@@ -39,12 +39,12 @@ type Asker struct {
}
// Put puts a on its stream: the seat's accept subject of its language and kind.
func (k Asker) Put(ctx context.Context, a delivery.StepAsk) error {
raw, err := delivery.EncodeStepAsk(a)
func (k Asker) Put(ctx context.Context, a buildstep.StepAsk) error {
raw, err := buildstep.EncodeStepAsk(a)
if err != nil {
return err
}
subject, err := delivery.AskSubject(a.Language, a.Kind)
subject, err := buildstep.AskSubject(a.Language, a.Kind)
if err != nil {
return err
}
@@ -57,16 +57,16 @@ func (k Asker) Put(ctx context.Context, a delivery.StepAsk) error {
// Wait waits for the step of an ask already put (a builder that started again takes over its own asks with it, never
// putting them twice) and answers it when it ended, whatever its state. The bound runs from the step's start on the node
// that took the ask, read from its running key, so an ask that waits in the stream for a free slot waits as long as ctx.
func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) {
module, err := delivery.BuildModuleOf(a.Language)
func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, error) {
module, err := buildstep.BuildModuleOf(a.Language)
if err != nil {
return delivery.Step{}, err
return buildstep.Step{}, err
}
watch, cancel := context.WithCancel(ctx)
defer cancel()
steps, err := k.Waiter.WatchAsk(watch, module, a.ID)
if err != nil {
return delivery.Step{}, err
return buildstep.Step{}, err
}
var bound, outerCap <-chan time.Time
var timer, capTimer *time.Timer
@@ -89,20 +89,20 @@ func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, err
for {
select {
case <-ctx.Done():
return delivery.Step{}, ctx.Err()
return buildstep.Step{}, ctx.Err()
case <-bound:
return delivery.Step{}, standStill()
return buildstep.Step{}, standStill()
case <-outerCap:
return delivery.Step{}, fmt.Errorf("the ask %s's step on %s did not end within %s of its start: it kept moving (last phase %q at %s) but never ended",
return buildstep.Step{}, fmt.Errorf("the ask %s's step on %s did not end within %s of its start: it kept moving (last phase %q at %s) but never ended",
a.ID, running, 3*k.Bound, phase, phaseAt.UTC().Format(time.RFC3339))
case step, ok := <-steps:
if !ok {
return delivery.Step{}, errors.New("the watch of the ask's step ended")
return buildstep.Step{}, errors.New("the watch of the ask's step ended")
}
if seen := stepSeenOf(ctx); seen != nil {
seen(step)
}
if step.State != delivery.StepRunning {
if step.State != buildstep.StepRunning {
return step, nil
}
if k.Bound <= 0 {
@@ -148,31 +148,31 @@ func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, err
type stepSeenKey struct{}
// WithStepSeen is ctx carrying seen, which Asker.Wait hands every value the ask's step key takes while it waits: the node
// that took it, each phase it says while running (delivery.Step.Phase), and its end. A try names its asks' progress so
// that took it, each phase it says while running (buildstep.Step.Phase), and its end. A try names its asks' progress so
// (issue 1890). seen runs in the waiting goroutine and must not block.
func WithStepSeen(ctx context.Context, seen func(delivery.Step)) context.Context {
func WithStepSeen(ctx context.Context, seen func(buildstep.Step)) context.Context {
return context.WithValue(ctx, stepSeenKey{}, seen)
}
func stepSeenOf(ctx context.Context) func(delivery.Step) {
seen, _ := ctx.Value(stepSeenKey{}).(func(delivery.Step))
func stepSeenOf(ctx context.Context) func(buildstep.Step) {
seen, _ := ctx.Value(stepSeenKey{}).(func(buildstep.Step))
return seen
}
// Ask puts a and waits for its step; a step that did not pass is an error naming its node, its state and its reason.
func (k Asker) Ask(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) {
func (k Asker) Ask(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, error) {
if err := k.Put(ctx, a); err != nil {
return delivery.Step{}, err
return buildstep.Step{}, err
}
return k.Passed(k.Wait(ctx, a))
}
// Passed is step when it passed, else an error naming where it ended and why.
func (k Asker) Passed(step delivery.Step, err error) (delivery.Step, error) {
func (k Asker) Passed(step buildstep.Step, err error) (buildstep.Step, error) {
if err != nil {
return step, err
}
if step.State != delivery.StepPassed {
if step.State != buildstep.StepPassed {
return step, fmt.Errorf("the ask %s on %s %s: %s", step.ID, step.Node, step.State, step.Reason)
}
return step, nil
@@ -194,20 +194,20 @@ func (s JetStreamAsks) PublishAsk(ctx context.Context, subject, id string, data
// WatchAsk is AskWaiter's: the watch hands over what the keys hold now, so a step that ended before it began is read at
// once, and every change after.
func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
pattern, err := delivery.AskStepWatch(askID)
func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
pattern, err := buildstep.AskStepWatch(askID)
if err != nil {
return nil, err
}
kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module))
kv, err := s.JS.KeyValue(ctx, StepsBucket(module))
if err != nil {
return nil, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
return nil, fmt.Errorf("the bucket %s: %w", StepsBucket(module), err)
}
w, err := kv.Watch(ctx, pattern)
if err != nil {
return nil, err
}
out := make(chan delivery.Step)
out := make(chan buildstep.Step)
go func() {
defer close(out)
defer w.Stop()
@@ -222,7 +222,7 @@ func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-ch
if e == nil || e.Operation() != jetstream.KeyValuePut {
continue
}
step, err := delivery.DecodeStep(e.Value())
step, err := buildstep.DecodeStep(e.Value())
if err != nil {
continue // not a step (a node's pause key never matches an ask id; anything else is not this ask's)
}
@@ -241,17 +241,17 @@ func (s JetStreamAsks) WatchAsk(ctx context.Context, module, askID string) (<-ch
// written at once with what Answer says a build node wrote (running or ended). Write writes a step later. Safe for
// concurrent use.
type FakeAsks struct {
Answer func(a delivery.StepAsk) delivery.Step
Answer func(a buildstep.StepAsk) buildstep.Step
mu sync.Mutex
asks []delivery.StepAsk
steps map[string][]delivery.Step
watchers map[string][]chan delivery.Step
asks []buildstep.StepAsk
steps map[string][]buildstep.Step
watchers map[string][]chan buildstep.Step
}
// PublishAsk is AskPublisher's.
func (f *FakeAsks) PublishAsk(_ context.Context, _, id string, data []byte) error {
a, err := delivery.DecodeStepAsk(data)
a, err := buildstep.DecodeStepAsk(data)
if err != nil {
return err
}
@@ -265,13 +265,13 @@ func (f *FakeAsks) PublishAsk(_ context.Context, _, id string, data []byte) erro
}
// Write writes an ask's step, as a build node does.
func (f *FakeAsks) Write(step delivery.Step) {
func (f *FakeAsks) Write(step buildstep.Step) {
f.mu.Lock()
if f.steps == nil {
f.steps, f.watchers = map[string][]delivery.Step{}, map[string][]chan delivery.Step{}
f.steps, f.watchers = map[string][]buildstep.Step{}, map[string][]chan buildstep.Step{}
}
f.steps[step.ID] = append(f.steps[step.ID], step)
ws := append([]chan delivery.Step(nil), f.watchers[step.ID]...)
ws := append([]chan buildstep.Step(nil), f.watchers[step.ID]...)
f.mu.Unlock()
for _, w := range ws {
w <- step
@@ -279,14 +279,14 @@ func (f *FakeAsks) Write(step delivery.Step) {
}
// End is Write: kept for the tests that end a step.
func (f *FakeAsks) End(step delivery.Step) { f.Write(step) }
func (f *FakeAsks) End(step buildstep.Step) { f.Write(step) }
// WatchAsk is AskWaiter's: what was written so far, then every write.
func (f *FakeAsks) WatchAsk(ctx context.Context, _ string, askID string) (<-chan delivery.Step, error) {
out := make(chan delivery.Step, 64)
func (f *FakeAsks) WatchAsk(ctx context.Context, _ string, askID string) (<-chan buildstep.Step, error) {
out := make(chan buildstep.Step, 64)
f.mu.Lock()
if f.steps == nil {
f.steps, f.watchers = map[string][]delivery.Step{}, map[string][]chan delivery.Step{}
f.steps, f.watchers = map[string][]buildstep.Step{}, map[string][]chan buildstep.Step{}
}
for _, s := range f.steps[askID] {
out <- s
@@ -309,8 +309,8 @@ func (f *FakeAsks) WatchAsk(ctx context.Context, _ string, askID string) (<-chan
}
// Asks are the asks put so far.
func (f *FakeAsks) Asks() []delivery.StepAsk {
func (f *FakeAsks) Asks() []buildstep.StepAsk {
f.mu.Lock()
defer f.mu.Unlock()
return append([]delivery.StepAsk(nil), f.asks...)
return append([]buildstep.StepAsk(nil), f.asks...)
}
@@ -11,7 +11,7 @@ import (
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/testbus"
)
@@ -29,7 +29,7 @@ func TestAnAskIsPutOnceAndItsStepIsRead(t *testing.T) {
Retention: jetstream.WorkQueuePolicy}); err != nil {
t.Fatal(err)
}
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket("mesh-build-go")})
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: StepsBucket("mesh-build-go")})
if err != nil {
t.Fatal(err)
}
@@ -47,10 +47,10 @@ func TestAnAskIsPutOnceAndItsStepIsRead(t *testing.T) {
}
go func() {
time.Sleep(100 * time.Millisecond)
key, _ := delivery.AskStepKey("ace", a.ID)
running, _ := delivery.EncodeStep(delivery.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: delivery.StepRunning, Started: time.Now()})
key, _ := buildstep.AskStepKey("ace", a.ID)
running, _ := buildstep.EncodeStep(buildstep.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: buildstep.StepRunning, Started: time.Now()})
_, _ = kv.Put(ctx, key, running)
passed, _ := delivery.EncodeStep(delivery.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: delivery.StepPassed, Started: time.Now(),
passed, _ := buildstep.EncodeStep(buildstep.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: buildstep.StepPassed, Started: time.Now(),
Result: json.RawMessage(`{"artifacts":[]}`)})
_, _ = kv.Put(ctx, key, passed)
}()
@@ -9,30 +9,30 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/health"
)
func goAsk(id string) delivery.StepAsk {
return delivery.StepAsk{ID: id, Kind: delivery.AskDelivery, Language: delivery.AskGo, Asker: "mesh-delivery-builder", At: time.Now(),
Build: &delivery.BuildJob{Module: "m", Path: ".", Fetch: map[string]any{"repository": "novox/x", "commit": strings.Repeat("a", 40)},
Artifacts: []delivery.AskArtifact{{Artifact: "program"}}}}
func goAsk(id string) buildstep.StepAsk {
return buildstep.StepAsk{ID: id, Kind: buildstep.AskDelivery, Language: buildstep.AskGo, Asker: "mesh-delivery-builder", At: time.Now(),
Build: &buildstep.BuildJob{Module: "m", Path: ".", Fetch: &buildstep.Fetch{Repository: "novox/x", Commit: strings.Repeat("a", 40)},
Artifacts: []buildstep.AskArtifact{{Artifact: "program"}}}}
}
// 1887/A4a: an asker puts the ask once and answers the step a build node wrote for it, naming the node; one that did not
// pass is an error naming the node and why; a wait of an ask already put puts nothing.
func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
f := &FakeAsks{Answer: func(a delivery.StepAsk) delivery.Step {
return delivery.Step{ID: a.ID, Node: "ace", State: delivery.StepPassed, Result: []byte(`{"artifacts":[]}`)}
f := &FakeAsks{Answer: func(a buildstep.StepAsk) buildstep.Step {
return buildstep.Step{ID: a.ID, Node: "ace", State: buildstep.StepPassed, Result: []byte(`{"artifacts":[]}`)}
}}
k := Asker{Publisher: f, Waiter: f}
var seen []delivery.Step // issue 1890: a waiter that asks to see each value of the step sees it
step, err := k.Ask(WithStepSeen(context.Background(), func(s delivery.Step) { seen = append(seen, s) }), goAsk("mvl0frv18d1957"))
var seen []buildstep.Step // issue 1890: a waiter that asks to see each value of the step sees it
step, err := k.Ask(WithStepSeen(context.Background(), func(s buildstep.Step) { seen = append(seen, s) }), goAsk("mvl0frv18d1957"))
if err != nil || step.Node != "ace" || len(f.Asks()) != 1 || len(seen) != 1 || seen[0].Node != "ace" {
t.Fatalf("step %+v, %v, asks %d, seen %+v", step, err, len(f.Asks()), seen)
}
f.Answer = func(a delivery.StepAsk) delivery.Step {
return delivery.Step{ID: a.ID, Node: "shanks", State: delivery.StepRefused, Reason: "go 1.27 installed"}
f.Answer = func(a buildstep.StepAsk) buildstep.Step {
return buildstep.Step{ID: a.ID, Node: "shanks", State: buildstep.StepRefused, Reason: "go 1.27 installed"}
}
if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") {
t.Fatalf("a refused step: %v", err)
@@ -46,7 +46,7 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
if _, err := k.Wait(waiting, goAsk("mvl0frv18d1959")); err != context.DeadlineExceeded {
t.Fatalf("an ask no node took: %v, want it still waiting when ctx ended", err)
}
g.Write(delivery.Step{ID: "mvl0frv18d1961", Node: "ace", State: delivery.StepRunning, Started: time.Now()})
g.Write(buildstep.Step{ID: "mvl0frv18d1961", Node: "ace", State: buildstep.StepRunning, Started: time.Now()})
if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1961")); err == nil || !strings.Contains(err.Error(), "on ace did not end within") {
t.Fatalf("a step that never ends: %v", err)
}
@@ -60,16 +60,16 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
// of 60 ms).
k = Asker{Publisher: g, Waiter: g, Bound: time.Hour}
now := time.Now()
g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepRunning, Started: now.Add(-90 * time.Minute), Phase: "pb", PhaseAt: now})
g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepPassed})
if step, err := k.Wait(context.Background(), goAsk("mvl0frv18d1980")); err != nil || step.State != delivery.StepPassed {
g.Write(buildstep.Step{ID: "mvl0frv18d1980", Node: "ace", State: buildstep.StepRunning, Started: now.Add(-90 * time.Minute), Phase: "pb", PhaseAt: now})
g.Write(buildstep.Step{ID: "mvl0frv18d1980", Node: "ace", State: buildstep.StepPassed})
if step, err := k.Wait(context.Background(), goAsk("mvl0frv18d1980")); err != nil || step.State != buildstep.StepPassed {
t.Fatalf("a step that kept moving past its bound: %+v, %v", step, err)
}
g.Write(delivery.Step{ID: "mvl0frv18d1981", Node: "ace", State: delivery.StepRunning, Started: now.Add(-time.Hour), Phase: "integration: tests", PhaseAt: now.Add(-time.Hour)})
g.Write(buildstep.Step{ID: "mvl0frv18d1981", Node: "ace", State: buildstep.StepRunning, Started: now.Add(-time.Hour), Phase: "integration: tests", PhaseAt: now.Add(-time.Hour)})
if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1981")); err == nil || !strings.Contains(err.Error(), `stood still in "integration: tests"`) {
t.Fatalf("a step that stood still in a phase: %v", err)
}
g.Write(delivery.Step{ID: "mvl0frv18d1982", Node: "ace", State: delivery.StepRunning, Started: now.Add(-3 * time.Hour), Phase: "qz", PhaseAt: now})
g.Write(buildstep.Step{ID: "mvl0frv18d1982", Node: "ace", State: buildstep.StepRunning, Started: now.Add(-3 * time.Hour), Phase: "qz", PhaseAt: now})
if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1982")); err == nil || !strings.Contains(err.Error(), "kept moving") {
t.Fatalf("a step that never stops moving: %v", err)
}
@@ -79,7 +79,7 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
dead := &DeadLetters{}
lw := LostAskWaiter{Inner: g, Dead: dead}
k = Asker{Publisher: g, Waiter: lw, Bound: time.Minute}
g.Write(delivery.Step{ID: "mvl0frv18d1970", Node: "ace", State: delivery.StepRunning, Started: time.Now()})
g.Write(buildstep.Step{ID: "mvl0frv18d1970", Node: "ace", State: buildstep.StepRunning, Started: time.Now()})
given, ok := GivenUpOf(map[string][]string{"Mesh-Dead-Msg-Id": {"mvl0frv18d1970"}, "Mesh-Dead-Deliveries": {"5"},
"Mesh-Dead-Stream": {"SEAT_NODE_BUILD_GO"}, "Mesh-Dead-Gave-Up": {"2026-10-26T12:26:40Z"}})
if !ok {
@@ -87,7 +87,7 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
}
go dead.Heard(given)
step, err = k.Wait(context.Background(), goAsk("mvl0frv18d1970"))
if err != nil || step.State != delivery.StepFailed || !IsLost(step) ||
if err != nil || step.State != buildstep.StepFailed || !IsLost(step) ||
!strings.Contains(step.Reason, LostWords+" after 5 deliveries at 2026-10-26T12:26:40Z") {
t.Fatalf("a given-up ask: %+v, %v", step, err)
}
@@ -98,7 +98,7 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
if err != nil || !IsNotTaken(step) || IsLost(step) {
t.Fatalf("an ask no node took: %+v, %v", step, err)
}
g.Write(delivery.Step{ID: "mvl0frv18d1972", Node: "g14", State: delivery.StepPassed})
g.Write(buildstep.Step{ID: "mvl0frv18d1972", Node: "g14", State: buildstep.StepPassed})
if step, err = k.Wait(NotTakenBound(context.Background(), time.Minute), goAsk("mvl0frv18d1972")); err != nil || step.Node != "g14" {
t.Fatalf("a step written: %+v, %v", step, err)
}
@@ -107,14 +107,14 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
fol := &refusingFollower{refuse: 1}
var mu sync.Mutex
var told []Hearing
hearing := NewDeadLetterHearing(context.Background(), fol, []string{delivery.AskGo}, dead, nil,
hearing := NewDeadLetterHearing(context.Background(), fol, []string{buildstep.AskGo}, dead, nil,
func(h Hearing) { mu.Lock(); told = append(told, h); mu.Unlock() })
hearing.Wait()
if why := hearing.Deaf(); !strings.Contains(why, "SEAT_NODE_BUILD_GO: permissions violation") {
t.Fatalf("a refused hearing says %q", why)
}
k = Asker{Publisher: g, Waiter: LostAskWaiter{Inner: g, Dead: dead, Hearing: hearing}, Bound: time.Minute}
g.Write(delivery.Step{ID: "mvl0frv18d1973", Node: "g14", State: delivery.StepPassed})
g.Write(buildstep.Step{ID: "mvl0frv18d1973", Node: "g14", State: buildstep.StepPassed})
if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1973")); err != nil {
t.Fatal(err)
}
@@ -134,8 +134,8 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
d := &FakeAsks{}
heard := &HeardEnds{Inner: deafWatch{}, Read: d}
k = Asker{Publisher: d, Waiter: heard, Bound: time.Minute}
d.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "shanks", State: delivery.StepPassed})
got := make(chan delivery.Step, 1)
d.Write(buildstep.Step{ID: "mvl0frv18d1980", Node: "shanks", State: buildstep.StepPassed})
got := make(chan buildstep.Step, 1)
go func() { s, _ := k.Wait(context.Background(), goAsk("mvl0frv18d1980")); got <- s }()
for heard.Waiting() == 0 {
time.Sleep(time.Millisecond)
@@ -147,11 +147,11 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
case <-time.After(20 * time.Millisecond):
}
heard.Nudge(GoBuildModule)
if s := <-got; s.State != delivery.StepPassed || s.Node != "shanks" {
if s := <-got; s.State != buildstep.StepPassed || s.Node != "shanks" {
t.Fatalf("a step its watch never handed over: %+v", s)
}
closing := &HeardEnds{Inner: &closingWatch{inner: d}, Read: d}
if s, err := (Asker{Publisher: d, Waiter: closing}).Wait(context.Background(), goAsk("mvl0frv18d1980")); err != nil || s.State != delivery.StepPassed {
if s, err := (Asker{Publisher: d, Waiter: closing}).Wait(context.Background(), goAsk("mvl0frv18d1980")); err != nil || s.State != buildstep.StepPassed {
t.Fatalf("a watch that closed once: %+v, %v", s, err)
}
for _, row := range []struct {
@@ -178,8 +178,8 @@ func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) {
k := Asker{Publisher: f, Waiter: f, Bound: time.Hour}
const id = "mvl0frv18d1962"
now := time.Now()
f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: now})
f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: now.Add(-time.Hour)})
f.Write(buildstep.Step{ID: id, Node: "ace", State: buildstep.StepRunning, Started: now})
f.Write(buildstep.Step{ID: id, Node: "ace", State: buildstep.StepRunning, Started: now.Add(-time.Hour)})
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
if _, err := k.Wait(ctx, goAsk(id)); err == nil || !strings.Contains(err.Error(), "on ace did not end within 1h0m0s of its start") {
@@ -213,8 +213,8 @@ func (f *refusingFollower) heard(g GivenUp) { f.mu.Lock(); h := f.hand; f.mu.Unl
// deafWatch is a watch of an ask's step that stays open and hears nothing.
type deafWatch struct{}
func (deafWatch) WatchAsk(context.Context, string, string) (<-chan delivery.Step, error) {
return make(chan delivery.Step), nil
func (deafWatch) WatchAsk(context.Context, string, string) (<-chan buildstep.Step, error) {
return make(chan buildstep.Step), nil
}
// closingWatch closes its first watch at once, and hands the later ones to inner.
@@ -223,10 +223,10 @@ type closingWatch struct {
closed bool
}
func (c *closingWatch) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
func (c *closingWatch) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
if !c.closed {
c.closed = true
ch := make(chan delivery.Step)
ch := make(chan buildstep.Step)
close(ch)
return ch, nil
}
@@ -5,7 +5,7 @@ import (
"sort"
"strings"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// ChooseNodes picks up to want build nodes for a language (ADR 0478 clause 3): the nodes whose toolchains key reports the
@@ -18,10 +18,10 @@ import (
// agree on, qualifying or not, as the indexer computes it. When nothing is reported, or the nodes report different
// versions (so the indexer agreed none), it fails naming the language and why; when no reporting node qualifies it fails
// naming the nodes. A node whose build module refuses on its toolchain refuses the step, and the build fails naming it.
func ChooseNodes(language string, entries []delivery.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, delivery.Toolchain2, error) {
tc, none := delivery.AgreedToolchain(entries)
func ChooseNodes(language string, entries []buildstep.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, buildstep.Toolchain, error) {
tc, none := buildstep.AgreedToolchain(entries)
if none != "" {
return nil, delivery.Toolchain2{}, fmt.Errorf("no %s build node qualifies: %s", language, none)
return nil, buildstep.Toolchain{}, fmt.Errorf("no %s build node qualifies: %s", language, none)
}
nodes, err := choose(language, entries, places, qualifies, want)
return nodes, tc, err
@@ -29,11 +29,11 @@ func ChooseNodes(language string, entries []delivery.NodeToolchain, places *Plac
// ChooseAnyNodes is ChooseNodes for artifacts that need no toolchain, such as an archive (ADR 0478 clause 1): the
// build module's nodes agree on nothing, so any node of the build module that qualifies is chosen, by capacity and load only.
func ChooseAnyNodes(language string, entries []delivery.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, error) {
func ChooseAnyNodes(language string, entries []buildstep.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, error) {
return choose(language, entries, places, qualifies, want)
}
func choose(language string, entries []delivery.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, error) {
func choose(language string, entries []buildstep.NodeToolchain, places *Places, qualifies func(node string) bool, want int) ([]string, error) {
if places == nil {
places = &Places{}
}
@@ -4,17 +4,17 @@ import (
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
const digest = "sha256:2222222222222222222222222222222222222222222222222222222222222222"
func hex64(c byte) string { return strings.Repeat(string(c), 64) }
func tcs(language, version string, nodes ...string) []delivery.NodeToolchain {
var out []delivery.NodeToolchain
func tcs(language, version string, nodes ...string) []buildstep.NodeToolchain {
var out []buildstep.NodeToolchain
for _, n := range nodes {
out = append(out, delivery.NodeToolchain{Node: n, Toolchain: delivery.Toolchain{Language: language, Version: version, Recipe: "1"}})
out = append(out, buildstep.NodeToolchain{Node: n, Toolchain: buildstep.Toolchain{Language: language, Version: version, Recipe: "1"}})
}
return out
}
@@ -42,8 +42,8 @@ func chooseWithOneNodeAnswersOneNode(t *testing.T) { // 1607/AC4
func chooseFailsNamingTheLanguageAndTheVersionsWhenNoNodeQualifies(t *testing.T) { // 1607/AC4
_, _, err := ChooseNodes("typescript", nil, nil, nil, 2)
if err == nil || !strings.Contains(err.Error(), "typescript") || !strings.Contains(err.Error(), delivery.ToolchainNoReport) {
t.Errorf("no report: error = %v, want it to name typescript and %q", err, delivery.ToolchainNoReport)
if err == nil || !strings.Contains(err.Error(), "typescript") || !strings.Contains(err.Error(), buildstep.ToolchainNoReport) {
t.Errorf("no report: error = %v, want it to name typescript and %q", err, buildstep.ToolchainNoReport)
}
entries := append(tcs("go", "go1.27.1", "novox"), tcs("go", "go1.26.0", "ace")...)
_, _, err = ChooseNodes("go", entries, nil, nil, 2)
@@ -3,6 +3,12 @@ module git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun
go 1.26.0
require (
git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261104174359-05b3e2e1b94f
github.com/nats-io/nats.go v1.54.0
)
@@ -3,7 +3,7 @@ package buildrun
import (
"sync"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient"
)
// HaveSet holds what each node's node-engine says of each module on it (node-engine_have, key `<node>.<module>`): the
@@ -19,17 +19,17 @@ type HaveSet struct {
// Put takes a change of the have bucket in and answers whether a module's healthy state changed. A node's own key
// (`<node>.node`) and a value that is not a have are left out.
func (s *HaveSet) Put(key string, value []byte, deleted bool) bool {
node, module, isNode, err := delivery.ParseHaveKey(key)
node, module, isNode, err := nodeengineclient.ParseHaveKey(key)
if err != nil || isNode {
return false
}
ok := false
if !deleted {
h, err := delivery.DecodeHave(value)
h, err := nodeengineclient.DecodeHave(value)
if err != nil {
return false
}
ok = h.Phase == delivery.PhaseRunning && h.Health == delivery.HealthOK
ok = h.Phase == nodeengineclient.PhaseRunning && h.Health == nodeengineclient.HealthOK
}
k := node + "." + module
s.mu.Lock()
@@ -7,7 +7,7 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// **A step's end is never only its own watch's to hear** (the lost check of 2026-10-28: mesh-catalog #553's verdict
@@ -22,7 +22,7 @@ import (
// StepReader reads, now, every step a build module's steps state holds for an ask: its keys `*.<ask id>`.
type StepReader interface {
AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error)
AskSteps(ctx context.Context, module, askID string) ([]buildstep.Step, error)
}
// reopenTries is how often a wait opens its ask's watch again after it closed, before the wait ends failed.
@@ -50,7 +50,7 @@ func (h *HeardEnds) logf(format string, args ...any) {
}
// WatchAsk is AskWaiter's: what Inner's watch hands, and, at each nudge, what the ask's keys hold now.
func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
in, err := h.Inner.WatchAsk(ctx, module, askID)
if err != nil {
return nil, err
@@ -62,7 +62,7 @@ func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan
}
h.waiting[askID] = w
h.mu.Unlock()
out := make(chan delivery.Step)
out := make(chan buildstep.Step)
go func() {
defer close(out)
defer func() {
@@ -72,7 +72,7 @@ func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan
}
h.mu.Unlock()
}()
send := func(s delivery.Step) bool {
send := func(s buildstep.Step) bool {
select {
case out <- s:
return true
@@ -112,7 +112,7 @@ func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan
continue
}
for _, s := range steps {
if s.State != delivery.StepRunning {
if s.State != buildstep.StepRunning {
h.logf("the ask %s's step on %s ended %s, read at a nudge: its watch had not handed it over", askID, s.Node, s.State)
}
if !send(s) {
@@ -147,26 +147,26 @@ func (h *HeardEnds) Waiting() int {
}
// AskSteps is StepReader's: the keys `*.<ask id>` of the build module's steps state, read now.
func (s JetStreamAsks) AskSteps(ctx context.Context, module, askID string) ([]delivery.Step, error) {
pattern, err := delivery.AskStepWatch(askID)
func (s JetStreamAsks) AskSteps(ctx context.Context, module, askID string) ([]buildstep.Step, error) {
pattern, err := buildstep.AskStepWatch(askID)
if err != nil {
return nil, err
}
kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module))
kv, err := s.JS.KeyValue(ctx, StepsBucket(module))
if err != nil {
return nil, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
return nil, fmt.Errorf("the bucket %s: %w", StepsBucket(module), err)
}
lister, err := kv.ListKeysFiltered(ctx, pattern)
if err != nil {
return nil, err
}
var out []delivery.Step
var out []buildstep.Step
for key := range lister.Keys() {
e, err := kv.Get(ctx, key)
if err != nil {
continue // deleted between the list and the read
}
if step, err := delivery.DecodeStep(e.Value()); err == nil {
if step, err := buildstep.DecodeStep(e.Value()); err == nil {
out = append(out, step)
}
}
@@ -182,7 +182,7 @@ func HearEnds(ctx context.Context, asks JetStreamAsks, inner AskWaiter, modules
}
h := &HeardEnds{Inner: inner, Read: asks, Logf: logf}
for _, module := range modules {
kv, err := asks.JS.KeyValue(ctx, delivery.StepsBucket(module))
kv, err := asks.JS.KeyValue(ctx, StepsBucket(module))
if err != nil {
logf("the steps of %s cannot be read (%v): an ask whose own watch goes deaf is read only at a reconnect", module, err)
continue
@@ -206,7 +206,7 @@ func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module st
continue
}
heard = true
if step, err := delivery.DecodeStep(e.Value()); err == nil && step.State != delivery.StepRunning {
if step, err := buildstep.DecodeStep(e.Value()); err == nil && step.State != buildstep.StepRunning {
h.Nudge(module)
}
}
@@ -226,11 +226,11 @@ func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module st
}
// AskSteps is StepReader's: the last step written for the ask, if any.
func (f *FakeAsks) AskSteps(_ context.Context, _ string, askID string) ([]delivery.Step, error) {
func (f *FakeAsks) AskSteps(_ context.Context, _ string, askID string) ([]buildstep.Step, error) {
f.mu.Lock()
defer f.mu.Unlock()
if s := f.steps[askID]; len(s) > 0 {
return []delivery.Step{s[len(s)-1]}, nil
return []buildstep.Step{s[len(s)-1]}, nil
}
return nil, nil
}
@@ -12,13 +12,13 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
"git.novox.be/novox/mesh-sdk/go/health"
"git.novox.be/novox/mesh-sdk/go/process"
)
// **An ask the build seat gave up on ends its step** (novox/hq issue 734). A build seat's worker offers an ask
// delivery.AskMaxDeliver times; one handed out that often without an answer (a bus reload that drops every hand-out, a node
// buildstep.AskMaxDeliver times; one handed out that often without an answer (a bus reload that drops every hand-out, a node
// that dies taking it each time) is given up by the server, and the controller keeps a copy of it in DEAD_LETTERS under
// mesh.events.dead.<seat stream>.<worker>, with the ask's message id (its ask id), how often it was handed out and when it
// was given up. No build node writes a step for it, so an asker that waits for the step waits for ever: a try stays
@@ -193,7 +193,7 @@ func (w LostAskWaiter) now() time.Time {
}
// WatchAsk is AskWaiter's.
func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan delivery.Step, error) {
func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-chan buildstep.Step, error) {
if w.Hearing != nil && w.Hearing.Deaf() != "" {
w.Hearing.Again()
}
@@ -215,12 +215,12 @@ func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-ch
nt, stopBound = context.WithTimeout(context.Background(), bound)
notTaken = nt.Done()
}
out := make(chan delivery.Step)
out := make(chan buildstep.Step)
go func() {
defer close(out)
defer cancel()
defer func() { stopBound() }()
send := func(s delivery.Step) bool {
send := func(s buildstep.Step) bool {
select {
case out <- s:
return true
@@ -228,9 +228,9 @@ func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-ch
return false
}
}
ended := func(reason string) delivery.Step {
ended := func(reason string) buildstep.Step {
at := w.now().UTC()
return delivery.Step{ID: askID, Node: LostNode, State: delivery.StepFailed, Reason: reason, Ended: at}
return buildstep.Step{ID: askID, Node: LostNode, State: buildstep.StepFailed, Reason: reason, Ended: at}
}
for {
select {
@@ -261,10 +261,12 @@ func (w LostAskWaiter) WatchAsk(ctx context.Context, module, askID string) (<-ch
}
// IsLost says whether a step was ended by LostAskWaiter because the build seat gave its ask up (LostWords).
func IsLost(s delivery.Step) bool { return s.Node == LostNode && strings.Contains(s.Reason, LostWords) }
func IsLost(s buildstep.Step) bool {
return s.Node == LostNode && strings.Contains(s.Reason, LostWords)
}
// IsNotTaken says whether a step was ended by LostAskWaiter because no build node took its ask within the bound.
func IsNotTaken(s delivery.Step) bool {
func IsNotTaken(s buildstep.Step) bool {
return s.Node == LostNode && strings.Contains(s.Reason, NotTakenWords)
}
@@ -0,0 +1,15 @@
package buildrun
// The holders' states the runner reads, named from the builder's own manifest (module.json `reads`; a unit row of the
// builder checks them against it) and from the build seats' states: a bucket is `<holder module>_<state>`, and no contract
// names a holder (novox/hq ADR 0625 §4, ADR 0634 §1).
const (
// PlacementBucket is the assigner's state placement, which the toolchain gate reads for the placing of a build module.
PlacementBucket = "mesh-delivery-assigner_placement"
)
// StepsBucket is the bucket of a build module's state steps.
func StepsBucket(module string) string { return module + "_steps" }
// ToolchainsBucket is the bucket of a build module's state toolchains.
func ToolchainsBucket(module string) string { return module + "_toolchains" }
@@ -3,7 +3,7 @@ package buildrun
import (
"sync"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
)
// PresenceSet holds which nodes are online, as the watchman writes it (mesh-delivery-watchman_presence, key `<node>`, the
@@ -17,13 +17,16 @@ type PresenceSet struct {
// Put takes a change of the presence bucket in and answers whether a node's online state changed. A part's key and a
// value that is not a presence are left out.
func (s *PresenceSet) Put(key string, value []byte, deleted bool) bool {
node, part, _, err := delivery.ParsePresenceKey(key)
if err != nil || part != "" {
if _, _, isPart := meshwatchdogs.ParsePresenceByPartPartInstanceKey(key); isPart {
return false
}
node, ok := meshwatchdogs.ParsePresenceByNodeKey(key)
if !ok {
return false
}
on := false
if !deleted {
p, err := delivery.DecodePresence(value)
p, err := meshwatchdogs.DecodePresenceByNodeValue(value)
if err != nil {
return false
}
@@ -1,14 +1,15 @@
package buildrun
import (
"encoding/json"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshwatchdogs"
)
func presenceValue(t *testing.T, online bool) []byte {
t.Helper()
b, err := delivery.EncodePresence(delivery.Presence{Online: online})
b, err := json.Marshal(meshwatchdogs.PresenceByNodeValue{Online: online})
if err != nil {
t.Fatal(err)
}
@@ -12,7 +12,8 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
"git.novox.be/novox/mesh-sdk/go/process"
)
@@ -23,7 +24,7 @@ import (
//
// ToolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a
// ToolchainSet): an entry is passed on only while the placement (the seat mesh-placement's state `placement`, held by
// the assigner, delivery.PlacementBucket, read by delivery.Placed; ADR 0567, ADR 0549 §3) places its build module on
// the assigner, PlacementBucket, read by meshmoduleassignerclient.Placed; ADR 0567, ADR 0549 §3) places its build module on
// its node, and only once it was written since that placement; it is passed on as deleted otherwise, said once, and
// passed on when its node writes it again. A report written before the placement is one an earlier run of the module
// left (issue 1940, 2026-10-25 20:56: mesh-build-node put on ace again brought ace.typescript v26.12.0 of 2026-10-23
@@ -39,7 +40,7 @@ type ToolchainGate struct {
modules map[string]bool // the build modules whose toolchains pass the gate
forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds
logf func(format string, a ...any)
placed *delivery.Placed
placed *meshmoduleassignerclient.Placed
known bool // the placement is settled
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
left map[string]string // <module>/<key> -> why it was said left out, so each reason is said once
@@ -51,7 +52,7 @@ type ToolchainGate struct {
// NewToolchainGate is a gate over the toolchains of modules, passing what stands on to forward and saying with logf what
// it leaves out.
func NewToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *ToolchainGate {
g := &ToolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, placed: delivery.NewPlaced(),
g := &ToolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, placed: meshmoduleassignerclient.NewPlaced(),
raw: map[string]map[string][]byte{}, left: map[string]string{}}
for _, m := range modules {
g.modules[m] = true
@@ -82,7 +83,7 @@ func (g *ToolchainGate) Placement(key string, value []byte, deleted bool, at tim
was := g.placed.Seeded()
node, changed, err := g.placed.Put(key, value, deleted, at)
if err != nil {
g.logf("%s: %v (issue 1940)", delivery.PlacementBucket, err)
g.logf("%s: %v (issue 1940)", PlacementBucket, err)
}
if !g.known {
return false
@@ -135,7 +136,7 @@ func (g *ToolchainGate) again(node string, modules []string) bool {
continue
}
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node {
if n, _, ok := nodebuildgo.ParseToolchainsByNodeLanguageKey(k); ok && n == node {
changed = g.pass(module, k) || changed
}
}
@@ -150,9 +151,9 @@ func (g *ToolchainGate) pass(module, key string) bool {
if !held {
return g.forward(module, key, nil, true)
}
node, _, err := delivery.ParseToolchainKey(key)
node, _, ok := nodebuildgo.ParseToolchainsByNodeLanguageKey(key)
why := ""
if err == nil && g.known && g.placed.Seeded() {
if ok && g.known && g.placed.Seeded() {
why = g.leftOut(module, node, value)
}
if why == "" {
@@ -161,7 +162,7 @@ func (g *ToolchainGate) pass(module, key string) bool {
}
if g.left[module+"/"+key] != why {
g.left[module+"/"+key] = why
g.logf("%s.%s is left out of the agreed toolchain: %s (issue 1940)", delivery.ToolchainsBucket(module), key, why)
g.logf("%s.%s is left out of the agreed toolchain: %s (issue 1940)", ToolchainsBucket(module), key, why)
}
return g.forward(module, key, nil, true)
}
@@ -174,8 +175,8 @@ func (g *ToolchainGate) leftOut(module, node string, value []byte) string {
if !placed {
return fmt.Sprintf("the catalogue does not place %s on %s, so it is no builder", module, node)
}
tc, err := delivery.DecodeToolchain(value)
if err == nil && tc.At.Before(since) {
tc, err := nodebuildgo.DecodeToolchainsByNodeLanguageValue(value)
if err == nil && tc.At != nil && tc.At.Before(since) {
return fmt.Sprintf("it was written at %s, before %s was placed on %s at %s: it counts once %s writes it again",
tc.At.UTC().Format(time.RFC3339), module, node, since.UTC().Format(time.RFC3339), node)
}
@@ -188,14 +189,14 @@ func (g *ToolchainGate) leftOut(module, node string, value []byte) string {
// stops for it stops every delivery (issue 1945). An error of the watch after that ends on the channel.
func FollowPlacement(ctx context.Context, js process.BucketOpener, gate *ToolchainGate, changed func(context.Context), logf func(string, ...any)) (<-chan error, error) {
errs := make(chan error, 1)
kv, err := js.KeyValue(ctx, delivery.PlacementBucket)
kv, err := js.KeyValue(ctx, PlacementBucket)
if err != nil {
logf("the state %s cannot be read, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", delivery.PlacementBucket, err)
logf("the state %s cannot be read, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", PlacementBucket, err)
return errs, nil
}
w, err := kv.WatchAll(ctx, jetstream.IncludeHistory())
if err != nil {
logf("the state %s cannot be watched, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", delivery.PlacementBucket, err)
logf("the state %s cannot be watched, so every node's toolchain report counts, a build node's the catalogue does not place too (issue 1940): %v", PlacementBucket, err)
return errs, nil
}
put := func(e jetstream.KeyValueEntry) bool {
@@ -208,7 +209,7 @@ func FollowPlacement(ctx context.Context, js process.BucketOpener, gate *Toolcha
return errs, ctx.Err()
case e, ok := <-w.Updates():
if !ok {
return errs, errors.New("the watch of " + delivery.PlacementBucket + " ended before it settled")
return errs, errors.New("the watch of " + PlacementBucket + " ended before it settled")
}
if e == nil {
settled = true
@@ -232,7 +233,7 @@ func FollowPlacement(ctx context.Context, js process.BucketOpener, gate *Toolcha
return
case e, ok := <-w.Updates():
if !ok {
errs <- errors.New("the watch of " + delivery.PlacementBucket + " ended")
errs <- errors.New("the watch of " + PlacementBucket + " ended")
return
}
if e != nil && put(e) {
@@ -3,6 +3,7 @@ package buildrun
import (
"bytes"
"context"
"encoding/json"
"errors"
"fmt"
"log"
@@ -12,16 +13,16 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient"
"git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
func gateReport(t *testing.T, node, version string, at time.Time) (string, []byte) {
t.Helper()
key, err := delivery.ToolchainKey(node, "typescript")
if err != nil {
t.Fatal(err)
}
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: at})
key := nodebuildgo.ToolchainsByNodeLanguageKey(node, "typescript")
value, err := json.Marshal(nodebuildgo.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: &at})
if err != nil {
t.Fatal(err)
}
@@ -31,27 +32,27 @@ func gateReport(t *testing.T, node, version string, at time.Time) (string, []byt
// placing is a value of the state `placement` placing modules on node.
func placing(t *testing.T, node string, modules ...string) (string, []byte) {
t.Helper()
p := delivery.Placement{Node: node}
p := meshmoduleassigner.Placement{Node: node}
for _, m := range modules {
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
p.Modules = append(p.Modules, meshmoduleassigner.PlacedModule{Module: m})
}
values, err := delivery.PlacementValues(p)
values, err := meshmoduleassignerclient.PlacementValues(p)
if err != nil {
t.Fatal(err)
}
key, _ := delivery.PlacementNodeKey(node)
key, _ := meshmoduleassignerclient.NodeKey(node)
return key, values[key]
}
// placingLegacy is placing in the form before, `placement.<node>` (research 072 row 37).
func placingLegacy(t *testing.T, node string, modules ...string) (string, []byte) {
t.Helper()
p := delivery.Placement{Node: node}
p := meshmoduleassigner.Placement{Node: node}
for _, m := range modules {
p.Modules = append(p.Modules, delivery.PlacedModule{Module: m})
p.Modules = append(p.Modules, meshmoduleassigner.PlacedModule{Module: m})
}
key, _ := delivery.LegacyPlacementKey(node)
value, err := delivery.EncodePlacement(p)
key, _ := meshmoduleassignerclient.LegacyKey(node)
value, err := meshmoduleassignerclient.EncodePlacement(p)
if err != nil {
t.Fatal(err)
}
@@ -59,7 +60,7 @@ func placingLegacy(t *testing.T, node string, modules ...string) (string, []byte
}
func agreedOf(set *ToolchainSet) (string, string) {
tc, none := delivery.AgreedToolchain(set.Entries("typescript"))
tc, none := buildstep.AgreedToolchain(set.Entries("typescript"))
return tc.Version, none
}
@@ -91,7 +92,7 @@ func TestToolchainGate(t *testing.T) {
gate, set, said := start(t, true)
key, value := gateReport(t, "ace", "v26.12.0", old)
gate.Toolchain(NodeBuildModule, key, value, false)
if a := delivery.ToolchainAgreement(set.Entries("typescript")); len(a.SittingOut) != 1 || a.SittingOut[0].Node != "ace" { // ADR 0584: the majority is agreed, ace's report counts as the one that sits out
if a := buildstep.ToolchainAgreement(set.Entries("typescript")); len(a.SittingOut) != 1 || a.SittingOut[0].Node != "ace" { // ADR 0584: the majority is agreed, ace's report counts as the one that sits out
t.Fatalf("before the placement is settled every report counts; got %+v", a)
}
k, v := placing(t, "ace", GoBuildModule) // another build module on ace: no matter
@@ -111,7 +112,7 @@ func TestToolchainGate(t *testing.T) {
// Placed on ace with a report written since: it counts, and the builders really disagree.
k, v = placing(t, "ace", GoBuildModule, NodeBuildModule)
gate.Placement(k, v, false, old.Add(-time.Minute))
if a := delivery.ToolchainAgreement(set.Entries("typescript")); a.Toolchain.Version != "v26.12.1" || len(a.SittingOut) != 1 || a.SittingOut[0].Version != "v26.12.0" {
if a := buildstep.ToolchainAgreement(set.Entries("typescript")); a.Toolchain.Version != "v26.12.1" || len(a.SittingOut) != 1 || a.SittingOut[0].Version != "v26.12.0" {
t.Fatalf("builders that run different versions: the majority is agreed and ace sits out; got %+v", a)
}
// Taken off ace again (its key deleted): its report goes again.
@@ -188,7 +189,7 @@ func TestToolchainGate(t *testing.T) {
})
t.Run("the history: a module in the first revision heard of a key stands since before it", func(t *testing.T) {
p := delivery.NewPlaced()
p := meshmoduleassignerclient.NewPlaced()
// ace in the form before, then converted: its module stands since before, the legacy key read no more.
k, v := placingLegacy(t, "ace", NodeBuildModule)
p.Put(k, v, false, old)
@@ -6,7 +6,7 @@ import (
"fmt"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// Caller calls a tool of a module on one named machine (mesh-sdk process.Conn.CallOn; ADR 0469 clause 3).
@@ -18,7 +18,7 @@ type Caller interface {
// (ADR 0469 clause 2) once its state is not running. It is a watch of that key, not a poll; it returns when the step
// ends or ctx does.
type Steps interface {
Wait(ctx context.Context, module, key string) (delivery.Step, error)
Wait(ctx context.Context, module, key string) (buildstep.Step, error)
}
// Runner runs a build module's long steps: the call answers at once with the step's key, and the result is read from that
@@ -48,7 +48,7 @@ func (r Runner) Run(ctx context.Context, module, node, tool string, args map[str
if answer.Key == "" {
return nil, fmt.Errorf("%s on %s answered no step key: %s", tool, node, raw)
}
if n, _, err := delivery.ParseStepKey(answer.Key); err != nil || n != node {
if n, _, err := buildstep.ParseStepKey(answer.Key); err != nil || n != node {
return nil, fmt.Errorf("%s on %s answered the step key %q, which is not a key of that node", tool, node, answer.Key)
}
wait, cancel := context.WithTimeoutCause(ctx, r.Bound, fmt.Errorf("the step %s did not end within %s", answer.Key, r.Bound))
@@ -60,7 +60,7 @@ func (r Runner) Run(ctx context.Context, module, node, tool string, args map[str
}
return nil, fmt.Errorf("%s on %s: %w", tool, node, err)
}
if step.State != delivery.StepPassed {
if step.State != buildstep.StepPassed {
return nil, fmt.Errorf("%s on %s: the step %s %s: %s", tool, node, answer.Key, step.State, step.Reason)
}
return step.Result, nil
@@ -9,7 +9,7 @@ import (
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
type call struct {
@@ -35,12 +35,12 @@ func (f *fakeCaller) CallOn(node, key string, body any) (json.RawMessage, error)
// fakeSteps holds the steps values by `<module>/<key>`; Wait on a key it does not hold blocks until the context ends.
type fakeSteps struct {
mu sync.Mutex
steps map[string]delivery.Step
steps map[string]buildstep.Step
// during, when set, is called as a wait starts and the function it answers as the wait ends.
during func(key string) func()
}
func (f *fakeSteps) Wait(ctx context.Context, module, key string) (delivery.Step, error) {
func (f *fakeSteps) Wait(ctx context.Context, module, key string) (buildstep.Step, error) {
f.mu.Lock()
s, ok := f.steps[module+"/"+key]
during := f.during
@@ -52,11 +52,11 @@ func (f *fakeSteps) Wait(ctx context.Context, module, key string) (delivery.Step
return s, nil
}
<-ctx.Done()
return delivery.Step{}, context.Cause(ctx)
return buildstep.Step{}, context.Cause(ctx)
}
func stepAt(node, id, state, reason string, result string) delivery.Step {
s := delivery.Step{ID: id, Tool: "build_go_fetch", Node: node, State: state, Reason: reason}
func stepAt(node, id, state, reason string, result string) buildstep.Step {
s := buildstep.Step{ID: id, Tool: "build_go_fetch", Node: node, State: state, Reason: reason}
if result != "" {
s.Result = json.RawMessage(result)
}
@@ -65,7 +65,7 @@ func stepAt(node, id, state, reason string, result string) delivery.Step {
func runCallsTheToolOnTheNodeAndReadsTheStepsKey(t *testing.T) { // 1607/AC5
caller := &fakeCaller{answer: `{"step":"abc1","key":"ace.abc1","state":"running"}`}
steps := &fakeSteps{steps: map[string]delivery.Step{"mesh-build-go/ace.abc1": stepAt("ace", "abc1", delivery.StepPassed, "", `{"workspace":"w1","commit":"c"}`)}}
steps := &fakeSteps{steps: map[string]buildstep.Step{"mesh-build-go/ace.abc1": stepAt("ace", "abc1", buildstep.StepPassed, "", `{"workspace":"w1","commit":"c"}`)}}
res, err := Runner{Caller: caller, Steps: steps, Bound: time.Second}.Run(context.Background(), "mesh-build-go", "ace", "build_go_fetch", map[string]any{"repository": "novox/r", "commit": "c"})
if err != nil {
t.Fatal(err)
@@ -80,9 +80,9 @@ func runCallsTheToolOnTheNodeAndReadsTheStepsKey(t *testing.T) { // 1607/AC5
}
func runFailsOnAFailedOrRefusedStepNamingItsReason(t *testing.T) { // 1607/AC5
for _, state := range []string{delivery.StepFailed, delivery.StepRefused} {
for _, state := range []string{buildstep.StepFailed, buildstep.StepRefused} {
caller := &fakeCaller{answer: `{"step":"abc1","key":"ace.abc1","state":"running"}`}
steps := &fakeSteps{steps: map[string]delivery.Step{"mesh-build-go/ace.abc1": stepAt("ace", "abc1", state, "go is go1.26.0, the recipe builds with go1.27.1", "")}}
steps := &fakeSteps{steps: map[string]buildstep.Step{"mesh-build-go/ace.abc1": stepAt("ace", "abc1", state, "go is go1.26.0, the recipe builds with go1.27.1", "")}}
_, err := Runner{Caller: caller, Steps: steps, Bound: time.Second}.Run(context.Background(), "mesh-build-go", "ace", "build_go_build", nil)
if err == nil || !strings.Contains(err.Error(), "go1.27.1") || !strings.Contains(err.Error(), state) || !strings.Contains(err.Error(), "ace") {
t.Errorf("%s: error = %v, want the node, the state and the reason", state, err)
@@ -3,7 +3,8 @@ package buildrun
import (
"sync"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
"git.novox.be/novox/mesh-delivery/sdk/buildstep"
)
// LanguageOf is the language each build module's toolchains state holds (the indexer's ToolchainModules, reversed).
@@ -13,7 +14,7 @@ var LanguageOf = map[string]string{GoBuildModule: "go", NodeBuildModule: "typesc
// put them in. Safe for use from the watches and from a build at once.
type ToolchainSet struct {
mu sync.Mutex
by map[string]map[string]delivery.Toolchain // language -> node -> entry
by map[string]map[string]nodebuildgo.ToolchainsByNodeLanguageValue // language -> node -> entry
}
// Put takes a change of a build module's toolchains bucket in: a put of a node's entry (key <node>.<language>) or its
@@ -24,8 +25,8 @@ func (s *ToolchainSet) Put(module, key string, value []byte, deleted bool) bool
if !ok {
return false
}
node, keyLanguage, err := delivery.ParseToolchainKey(key)
if err != nil || keyLanguage != language {
node, keyLanguage, ok := nodebuildgo.ParseToolchainsByNodeLanguageKey(key)
if !ok || keyLanguage != language {
return false
}
s.mu.Lock()
@@ -37,15 +38,15 @@ func (s *ToolchainSet) Put(module, key string, value []byte, deleted bool) bool
delete(s.by[language], node)
return true
}
tc, err := delivery.DecodeToolchain(value)
tc, err := nodebuildgo.DecodeToolchainsByNodeLanguageValue(value)
if err != nil || tc.Language != language {
return false
}
if s.by == nil {
s.by = map[string]map[string]delivery.Toolchain{}
s.by = map[string]map[string]nodebuildgo.ToolchainsByNodeLanguageValue{}
}
if s.by[language] == nil {
s.by[language] = map[string]delivery.Toolchain{}
s.by[language] = map[string]nodebuildgo.ToolchainsByNodeLanguageValue{}
}
cur, had := s.by[language][node]
s.by[language][node] = tc
@@ -53,12 +54,12 @@ func (s *ToolchainSet) Put(module, key string, value []byte, deleted bool) bool
}
// Entries are the nodes' entries of a language, in no particular order.
func (s *ToolchainSet) Entries(language string) []delivery.NodeToolchain {
func (s *ToolchainSet) Entries(language string) []buildstep.NodeToolchain {
s.mu.Lock()
defer s.mu.Unlock()
var out []delivery.NodeToolchain
var out []buildstep.NodeToolchain
for node, tc := range s.by[language] {
out = append(out, delivery.NodeToolchain{Node: node, Toolchain: tc})
out = append(out, buildstep.NodeToolchain{Node: node, Toolchain: buildstep.Toolchain{Language: tc.Language, Version: tc.Version, Recipe: tc.Recipe}})
}
return out
}
@@ -1,15 +1,17 @@
package buildrun
import (
"encoding/json"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-delivery/contracts/go/seats/nodebuildgo"
)
func tcValue(t *testing.T, language, version string) []byte {
t.Helper()
b, err := delivery.EncodeToolchain(delivery.Toolchain{Language: language, Version: version, Recipe: "1", At: time.Now()})
now := time.Now()
b, err := json.Marshal(nodebuildgo.Toolchain{Language: language, Version: version, Recipe: "1", At: &now})
if err != nil {
t.Fatal(err)
}