diff --git a/modules/mesh-delivery-builder/buildrun/asks.go b/modules/mesh-delivery-builder/buildrun/asks.go index fdaf3057..1af73b84 100644 --- a/modules/mesh-delivery-builder/buildrun/asks.go +++ b/modules/mesh-delivery-builder/buildrun/asks.go @@ -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 `*.` 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...) } diff --git a/modules/mesh-delivery-builder/buildrun/asks_integration_test.go b/modules/mesh-delivery-builder/buildrun/asks_integration_test.go index 6c6961c0..31363fad 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_integration_test.go +++ b/modules/mesh-delivery-builder/buildrun/asks_integration_test.go @@ -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) }() diff --git a/modules/mesh-delivery-builder/buildrun/asks_test.go b/modules/mesh-delivery-builder/buildrun/asks_test.go index ec6b94f9..357ade87 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_test.go +++ b/modules/mesh-delivery-builder/buildrun/asks_test.go @@ -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 } diff --git a/modules/mesh-delivery-builder/buildrun/choose.go b/modules/mesh-delivery-builder/buildrun/choose.go index 06aea831..b8997280 100644 --- a/modules/mesh-delivery-builder/buildrun/choose.go +++ b/modules/mesh-delivery-builder/buildrun/choose.go @@ -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{} } diff --git a/modules/mesh-delivery-builder/buildrun/choose_test.go b/modules/mesh-delivery-builder/buildrun/choose_test.go index bb2e69ab..d39e70c6 100644 --- a/modules/mesh-delivery-builder/buildrun/choose_test.go +++ b/modules/mesh-delivery-builder/buildrun/choose_test.go @@ -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) diff --git a/modules/mesh-delivery-builder/buildrun/go.mod b/modules/mesh-delivery-builder/buildrun/go.mod index fceda6f3..ede9fa15 100644 --- a/modules/mesh-delivery-builder/buildrun/go.mod +++ b/modules/mesh-delivery-builder/buildrun/go.mod @@ -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 ) diff --git a/modules/mesh-delivery-builder/buildrun/have.go b/modules/mesh-delivery-builder/buildrun/have.go index f3c079cf..ae5be958 100644 --- a/modules/mesh-delivery-builder/buildrun/have.go +++ b/modules/mesh-delivery-builder/buildrun/have.go @@ -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 `.`): 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`) 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() diff --git a/modules/mesh-delivery-builder/buildrun/heardends.go b/modules/mesh-delivery-builder/buildrun/heardends.go index 25b8cbbb..3cae74f1 100644 --- a/modules/mesh-delivery-builder/buildrun/heardends.go +++ b/modules/mesh-delivery-builder/buildrun/heardends.go @@ -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 `*.`. 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 `*.` 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 } diff --git a/modules/mesh-delivery-builder/buildrun/lostask.go b/modules/mesh-delivery-builder/buildrun/lostask.go index 750acaa4..a3392412 100644 --- a/modules/mesh-delivery-builder/buildrun/lostask.go +++ b/modules/mesh-delivery-builder/buildrun/lostask.go @@ -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.., 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) } diff --git a/modules/mesh-delivery-builder/buildrun/names.go b/modules/mesh-delivery-builder/buildrun/names.go new file mode 100644 index 00000000..770324f8 --- /dev/null +++ b/modules/mesh-delivery-builder/buildrun/names.go @@ -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 `_`, 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" } diff --git a/modules/mesh-delivery-builder/buildrun/presence.go b/modules/mesh-delivery-builder/buildrun/presence.go index 63f82435..fafc431b 100644 --- a/modules/mesh-delivery-builder/buildrun/presence.go +++ b/modules/mesh-delivery-builder/buildrun/presence.go @@ -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 ``, 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 } diff --git a/modules/mesh-delivery-builder/buildrun/presence_test.go b/modules/mesh-delivery-builder/buildrun/presence_test.go index 5157609a..38e45577 100644 --- a/modules/mesh-delivery-builder/buildrun/presence_test.go +++ b/modules/mesh-delivery-builder/buildrun/presence_test.go @@ -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) } diff --git a/modules/mesh-delivery-builder/buildrun/standing.go b/modules/mesh-delivery-builder/buildrun/standing.go index fdaf1da6..19913896 100644 --- a/modules/mesh-delivery-builder/buildrun/standing.go +++ b/modules/mesh-delivery-builder/buildrun/standing.go @@ -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 // / -> 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) { diff --git a/modules/mesh-delivery-builder/buildrun/standing_test.go b/modules/mesh-delivery-builder/buildrun/standing_test.go index 6a55ef4b..47093903 100644 --- a/modules/mesh-delivery-builder/buildrun/standing_test.go +++ b/modules/mesh-delivery-builder/buildrun/standing_test.go @@ -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.` (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) diff --git a/modules/mesh-delivery-builder/buildrun/steps.go b/modules/mesh-delivery-builder/buildrun/steps.go index f08e0026..37e9bed8 100644 --- a/modules/mesh-delivery-builder/buildrun/steps.go +++ b/modules/mesh-delivery-builder/buildrun/steps.go @@ -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 diff --git a/modules/mesh-delivery-builder/buildrun/steps_test.go b/modules/mesh-delivery-builder/buildrun/steps_test.go index fe1d32e0..c94a6314 100644 --- a/modules/mesh-delivery-builder/buildrun/steps_test.go +++ b/modules/mesh-delivery-builder/buildrun/steps_test.go @@ -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 `/`; 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) diff --git a/modules/mesh-delivery-builder/buildrun/toolchains.go b/modules/mesh-delivery-builder/buildrun/toolchains.go index 61704252..acd8089b 100644 --- a/modules/mesh-delivery-builder/buildrun/toolchains.go +++ b/modules/mesh-delivery-builder/buildrun/toolchains.go @@ -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 .) 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 } diff --git a/modules/mesh-delivery-builder/buildrun/toolchains_test.go b/modules/mesh-delivery-builder/buildrun/toolchains_test.go index 4df0d38d..ebf796ac 100644 --- a/modules/mesh-delivery-builder/buildrun/toolchains_test.go +++ b/modules/mesh-delivery-builder/buildrun/toolchains_test.go @@ -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) }