diff --git a/go/delivery/builder.go b/go/delivery/builder.go index 47bcab7..981bca7 100644 --- a/go/delivery/builder.go +++ b/go/delivery/builder.go @@ -184,6 +184,10 @@ type BuilderBuild struct { // does not yet hold: the key is not claimed until that package's trunk key is done and published (novox/hq ADR 0500 // clause 6). A reader that does not know the field reads the key as queued. Waits string `json:"waits,omitempty"` + // Asks are, on a building key, the ids of the asks the builder put for it (novox/hq ADR 0525 §1; issue 1887 A4b): a + // builder that takes the key over at its start waits for those asks' steps and never puts them again. A reader that + // does not know the field reads the key as before. + Asks []string `json:"asks,omitempty"` } func (b BuilderBuild) check() error { @@ -216,6 +220,14 @@ func (b BuilderBuild) check() error { return fmt.Errorf("delivery: the reference %q of the %s %q %s", a.Reference, a.Kind, a.Name, p) } } + for _, id := range b.Asks { + if b.State != BuildBuilding { + return fmt.Errorf("delivery: a %s build names the ask %s; only a building key does", b.State, id) + } + if err := checkAskID(id); err != nil { + return err + } + } if b.Waits != "" { if b.State != BuildQueued { return fmt.Errorf("delivery: a %s build waits for %q; only a queued key waits for a package", b.State, b.Waits) diff --git a/go/delivery/buildrun/asks.go b/go/delivery/buildrun/asks.go new file mode 100644 index 0000000..de6dac2 --- /dev/null +++ b/go/delivery/buildrun/asks.go @@ -0,0 +1,220 @@ +package buildrun + +import ( + "context" + "errors" + "fmt" + "sync" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// **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, +// A5). It chooses no node: whichever build node pulls the ask writes `.` in its build module's steps state, +// and the asker reads the first such key that is not running. The builder and the checker share it. + +// AskPublisher puts an encoded ask on subject, de-duplicated by its id. +type AskPublisher interface { + PublishAsk(ctx context.Context, subject, id string, data []byte) error +} + +// AskWaiter waits for an ask's step: the first key `*.` of the build module's steps state whose state is not +// running. A watch, not a poll; it returns when the step ends or ctx does. +type AskWaiter interface { + WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error) +} + +// Asker puts asks and waits for their steps. +type Asker struct { + Publisher AskPublisher + Waiter AskWaiter + // Bound is how long the asker waits for an ask's step before it stops waiting; zero waits as long as ctx. + Bound time.Duration +} + +// 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) + if err != nil { + return err + } + subject, err := delivery.AskSubject(a.Language, a.Kind) + if err != nil { + return err + } + if err := k.Publisher.PublishAsk(ctx, subject, a.ID, raw); err != nil { + return fmt.Errorf("putting the %s ask %s: %w", a.Kind, a.ID, err) + } + return nil +} + +// 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. +func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) { + module, err := delivery.BuildModuleOf(a.Language) + if err != nil { + return delivery.Step{}, err + } + wait, cancel := ctx, context.CancelFunc(func() {}) + if k.Bound > 0 { + wait, cancel = context.WithTimeoutCause(ctx, k.Bound, fmt.Errorf("the ask %s's step did not end within %s", a.ID, k.Bound)) + } + defer cancel() + step, err := k.Waiter.WaitAsk(wait, module, a.ID) + if err != nil { + if cause := context.Cause(wait); cause != nil && ctx.Err() == nil { + err = cause + } + return delivery.Step{}, err + } + return step, nil +} + +// 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) { + if err := k.Put(ctx, a); err != nil { + return delivery.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) { + if err != nil { + return step, err + } + if step.State != delivery.StepPassed { + return step, fmt.Errorf("the ask %s on %s %s: %s", step.ID, step.Node, step.State, step.Reason) + } + return step, nil +} + +// JetStreamAsks is the bus side of an Asker: a JetStream publish de-duplicated by the ask's id (Nats-Msg-Id), and a watch +// of `*.` in the build module's steps bucket. +type JetStreamAsks struct { + JS jetstream.JetStream +} + +// PublishAsk is AskPublisher's. +func (s JetStreamAsks) PublishAsk(ctx context.Context, subject, id string, data []byte) error { + msg := &nats.Msg{Subject: subject, Data: data, Header: nats.Header{}} + msg.Header.Set(jetstream.MsgIDHeader, id) + _, err := s.JS.PublishMsg(ctx, msg) + return err +} + +// WaitAsk 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) WaitAsk(ctx context.Context, module, askID string) (delivery.Step, error) { + pattern, err := delivery.AskStepWatch(askID) + if err != nil { + return delivery.Step{}, err + } + kv, err := s.JS.KeyValue(ctx, delivery.StepsBucket(module)) + if err != nil { + return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) + } + w, err := kv.Watch(ctx, pattern) + if err != nil { + return delivery.Step{}, err + } + defer w.Stop() + for { + select { + case <-ctx.Done(): + return delivery.Step{}, ctx.Err() + case e, ok := <-w.Updates(): + if !ok { + return delivery.Step{}, errors.New("the watch of the ask's step ended") + } + if e == nil || e.Operation() != jetstream.KeyValuePut { + continue + } + step, err := delivery.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) + } + if step.State != delivery.StepRunning { + return step, nil + } + } + } +} + +// FakeAsks is a test double of the bus for an asker's tests: each ask put is kept and, when Answer is set, its step is +// ended at once with what Answer says a build node wrote. Safe for concurrent use. +type FakeAsks struct { + Answer func(a delivery.StepAsk) delivery.Step + + mu sync.Mutex + asks []delivery.StepAsk + steps map[string]delivery.Step + waits map[string]chan struct{} +} + +// PublishAsk is AskPublisher's. +func (f *FakeAsks) PublishAsk(_ context.Context, _, id string, data []byte) error { + a, err := delivery.DecodeStepAsk(data) + if err != nil { + return err + } + f.mu.Lock() + f.asks = append(f.asks, a) + f.mu.Unlock() + if f.Answer != nil { + f.End(f.Answer(a)) + } + return nil +} + +// End writes an ask's step end, as a build node does. +func (f *FakeAsks) End(step delivery.Step) { + f.mu.Lock() + if f.steps == nil { + f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{} + } + f.steps[step.ID] = step + ch := f.waits[step.ID] + delete(f.waits, step.ID) + f.mu.Unlock() + if ch != nil { + close(ch) + } +} + +// WaitAsk is AskWaiter's. +func (f *FakeAsks) WaitAsk(ctx context.Context, _ string, askID string) (delivery.Step, error) { + f.mu.Lock() + if f.steps == nil { + f.steps, f.waits = map[string]delivery.Step{}, map[string]chan struct{}{} + } + if s, ok := f.steps[askID]; ok { + f.mu.Unlock() + return s, nil + } + ch := f.waits[askID] + if ch == nil { + ch = make(chan struct{}) + f.waits[askID] = ch + } + f.mu.Unlock() + select { + case <-ctx.Done(): + return delivery.Step{}, ctx.Err() + case <-ch: + } + f.mu.Lock() + defer f.mu.Unlock() + return f.steps[askID], nil +} + +// Asks are the asks put so far. +func (f *FakeAsks) Asks() []delivery.StepAsk { + f.mu.Lock() + defer f.mu.Unlock() + return append([]delivery.StepAsk(nil), f.asks...) +} diff --git a/go/delivery/buildrun/asks_integration_test.go b/go/delivery/buildrun/asks_integration_test.go new file mode 100644 index 0000000..6c6961c --- /dev/null +++ b/go/delivery/buildrun/asks_integration_test.go @@ -0,0 +1,61 @@ +//go:build integration + +package buildrun + +import ( + "context" + "encoding/json" + "testing" + "time" + + "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-sdk/go/testbus" +) + +// 1887/A4a, 1887/AC3 on the real bus: an ask put twice under one id is one message, and the asker reads the step a node +// wrote under `.`. +func TestAnAskIsPutOnceAndItsStepIsRead(t *testing.T) { + nc, err := nats.Connect(testbus.URL(t)) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, _ := jetstream.New(nc) + ctx := context.Background() + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "SEAT_NODE_BUILD_GO", Subjects: []string{"mesh.seat.node-build-go.accept.>"}, + Retention: jetstream.WorkQueuePolicy}); err != nil { + t.Fatal(err) + } + kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.StepsBucket("mesh-build-go")}) + if err != nil { + t.Fatal(err) + } + bus := JetStreamAsks{JS: js} + k := Asker{Publisher: bus, Waiter: bus, Bound: 5 * time.Second} + a := goAsk("mvl0frv18d1960") + for i := 0; i < 2; i++ { + if err := k.Put(ctx, a); err != nil { + t.Fatal(err) + } + } + s, _ := js.Stream(ctx, "SEAT_NODE_BUILD_GO") + if info, _ := s.Info(ctx); info.State.Msgs != 1 { + t.Fatalf("%d messages for one ask put twice", info.State.Msgs) + } + 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()}) + _, _ = kv.Put(ctx, key, running) + passed, _ := delivery.EncodeStep(delivery.Step{ID: a.ID, Tool: "delivery ask", Node: "ace", State: delivery.StepPassed, Started: time.Now(), + Result: json.RawMessage(`{"artifacts":[]}`)}) + _, _ = kv.Put(ctx, key, passed) + }() + step, err := k.Passed(k.Wait(ctx, a)) + if err != nil || step.Node != "ace" { + t.Fatalf("step %+v, %v", step, err) + } +} diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go new file mode 100644 index 0000000..b34466f --- /dev/null +++ b/go/delivery/buildrun/asks_test.go @@ -0,0 +1,43 @@ +package buildrun + +import ( + "context" + "strings" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +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"}}}} +} + +// 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":[]}`)} + }} + k := Asker{Publisher: f, Waiter: f} + step, err := k.Ask(context.Background(), goAsk("mvl0frv18d1957")) + if err != nil || step.Node != "ace" || len(f.Asks()) != 1 { + t.Fatalf("step %+v, %v, asks %d", step, err, len(f.Asks())) + } + f.Answer = func(a delivery.StepAsk) delivery.Step { + return delivery.Step{ID: a.ID, Node: "shanks", State: delivery.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) + } + g := &FakeAsks{} + k = Asker{Publisher: g, Waiter: g, Bound: 20 * time.Millisecond} + if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1959")); err == nil || !strings.Contains(err.Error(), "did not end within") { + t.Fatalf("an ask whose step never ends: %v", err) + } + if len(g.Asks()) != 0 { + t.Fatal("a wait put an ask") + } +} diff --git a/go/delivery/stepasks_test.go b/go/delivery/stepasks_test.go index 578b8a8..a8d2d11 100644 --- a/go/delivery/stepasks_test.go +++ b/go/delivery/stepasks_test.go @@ -124,3 +124,24 @@ func TestAskSubjects(t *testing.T) { t.Error("BuildSeat(agent)") } } + +// 1887/A4b: a building key names the asks put for it; a done key or a name that is no ask id is refused. +func TestABuildingKeyNamesItsAsks(t *testing.T) { + b := BuilderBuild{State: BuildBuilding, By: "novox.builder", Asks: []string{"mh3k2x9qa1b2c3"}} + raw, err := EncodeBuilderBuild(b) + if err != nil { + t.Fatal(err) + } + got, err := DecodeBuilderBuild(raw) + if err != nil || len(got.Asks) != 1 || got.Asks[0] != "mh3k2x9qa1b2c3" { + t.Fatalf("decoded %+v, %v", got, err) + } + for _, bad := range []BuilderBuild{ + {State: BuildBuilding, Asks: []string{"postgres"}}, + {State: BuildFailed, Error: "x", Asks: []string{"mh3k2x9qa1b2c3"}}, + } { + if _, err := EncodeBuilderBuild(bad); err == nil { + t.Errorf("encoded %+v", bad) + } + } +}