diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go new file mode 100644 index 0000000..6316cfe --- /dev/null +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/held.go @@ -0,0 +1,74 @@ +package main + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// heldOnTheBus says whether an ask a build key names is on its build seat's stream, or a build node wrote a step for it +// (issue 1910): read once, at the take-over of a key at start, never polled. +type heldOnTheBus struct{ js jetstream.JetStream } + +// Held is build.Builder's Held: a step key `.` in the build module's steps, or a message of the stream under +// the ask's subject whose Nats-Msg-Id is the ask's id. +func (h heldOnTheBus) Held(ctx context.Context, a delivery.StepAsk) (bool, error) { + module, err := delivery.BuildModuleOf(a.Language) + if err != nil { + return false, err + } + pattern, err := delivery.AskStepWatch(a.ID) + if err != nil { + return false, err + } + steps, err := h.js.KeyValue(ctx, delivery.StepsBucket(module)) + if err != nil { + return false, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err) + } + keys, err := steps.ListKeysFiltered(ctx, pattern) + if err != nil { + return false, err + } + for range keys.Keys() { + keys.Stop() + return true, nil // a node took it + } + subject, err := delivery.AskSubject(a.Language, a.Kind) + if err != nil { + return false, err + } + name, err := h.js.StreamNameBySubject(ctx, subject) + if errors.Is(err, jetstream.ErrStreamNotFound) { + return false, nil + } + if err != nil { + return false, err + } + stream, err := h.js.Stream(ctx, name) + if err != nil { + return false, err + } + // A seat's stream is a work queue, which takes no second consumer over its subjects: its messages are read by sequence + // instead, each a direct read that consumes nothing. + info, err := stream.Info(ctx) + if err != nil { + return false, err + } + for seq := info.State.FirstSeq; seq != 0 && seq <= info.State.LastSeq; seq++ { + m, err := stream.GetMsg(ctx, seq) + if errors.Is(err, jetstream.ErrMsgNotFound) { + continue // taken and acknowledged, or removed + } + if err != nil { + return false, err + } + if m.Subject == subject && m.Header.Get(jetstream.MsgIDHeader) == a.ID { + return true, nil + } + } + return false, nil +} diff --git a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go index 6d597a3..0cd20c8 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -155,6 +155,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro b := &build.Builder{ Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module}, Asks: buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound}, + Held: heldOnTheBus{js}, Toolchains: tools, Manifests: manifestReader{manifests}, Logf: logger.Printf, diff --git a/modules/mesh-delivery-builder/internal/build/run.go b/modules/mesh-delivery-builder/internal/build/run.go index 64e3bcd..9ca9dbf 100644 --- a/modules/mesh-delivery-builder/internal/build/run.go +++ b/modules/mesh-delivery-builder/internal/build/run.go @@ -35,6 +35,12 @@ type Builder struct { Put(ctx context.Context, a delivery.StepAsk) error Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, error) } + // Held says whether an ask a key names is on its stream or a build node wrote a step for it (issue 1910): a key taken + // over at start whose ask is neither was left by a builder that stopped between naming the ask and putting it, and + // the ask is put then, under the same id. Nil takes every named ask as held. + Held interface { + Held(ctx context.Context, a delivery.StepAsk) (bool, error) + } Toolchains ToolchainEntries Manifests Manifests // Limit is how many builds the builder has asked at once; their asks wait in the build seats' streams, which are the @@ -668,6 +674,21 @@ func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m del ask := delivery.StepAsk{ID: buildstep.NewID(), Kind: kind, Language: language, Asker: delivery.BuilderModule, At: time.Now(), Build: job} if id := asked.resume(language); id != "" { ask.ID = id // put before this builder started again: the stream holds it, or a node already ran it + if b.Held != nil { + held, err := b.Held.Held(ctx, ask) + if err != nil { + l.err = fmt.Errorf("whether the ask %s of %s the key names is on its stream cannot be read: %w", ask.ID, l.module, err) + return + } + if !held { + // The builder that named it stopped before it put it (issue 1910): put it now, under the same id. + b.logf("the ask %s of %s that the key names was never put: it is put now", ask.ID, l.module) + if err := b.Asks.Put(ctx, ask); err != nil { + l.err = err + return + } + } + } } else { if err := asked.note(ctx, language, ask.ID, "asked "+l.module); err != nil { l.err = fmt.Errorf("the ask of %s could not be written to the key: %w", l.module, err) diff --git a/modules/mesh-delivery-builder/internal/build/run_test.go b/modules/mesh-delivery-builder/internal/build/run_test.go index ba94b0e..4cc1a5c 100644 --- a/modules/mesh-delivery-builder/internal/build/run_test.go +++ b/modules/mesh-delivery-builder/internal/build/run_test.go @@ -613,3 +613,38 @@ func TestAFailedPublishOfACandidatesImageIsWrittenToTheKey(t *testing.T) { // re t.Errorf("build = %+v, want done with a reference", v) } } + +// heldAsks says an ask is held when its id is among held. +type heldAsks map[string]bool + +func (h heldAsks) Held(_ context.Context, a delivery.StepAsk) (bool, error) { return h[a.ID], nil } + +// 1910: a builder that stopped between naming the ask in its key and putting it left a key naming an ask neither the +// stream holds nor any node ran: started again, the builder puts that ask, under the same id, once, and the key is +// built from its step; an ask the stream holds is not put again. +func TestAnAskTheKeyNamesAndNobodyPutIsPutAtStart(t *testing.T) { + for _, held := range []bool{false, true} { + r := newRig(t, goBundle) + const id = "mvl0frv18d1958" + data, _ := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildBuilding, By: "novox.builder", Manifest: manifestD, + Step: "asked mesh-build-go", Asks: map[string]string{delivery.AskGo: id}}) + r.st.vals[r.key] = data + r.b.Held = heldAsks{id: held} + if held { + r.asks.Answer = nil + r.asks.End(delivery.Step{ID: id, Node: "shanks", State: delivery.StepPassed, + Result: []byte(`{"artifacts":[{"artifact":"tool","digest":"` + digest + `","reference":"artifact-store://m/tool/blobs/` + digest + `"}]}`)}) + } + v := r.consider(t, "modules/m") + asks := r.asks.Asks() + switch { + case held && len(asks) != 0: + t.Errorf("an ask the stream holds was put again: %+v", asks) + case !held && (len(asks) != 1 || asks[0].ID != id): + t.Errorf("the ask nobody put: asks %+v, want one under %s", asks, id) + } + if v.State != delivery.BuildDone { + t.Errorf("held %v: the key = %+v, want done", held, v) + } + } +}