Put again, under its own id, an ask a build key names that neither its stream holds nor any node ran, so a builder that stopped between naming and putting it leaves no key waiting for an ask nobody put (issue 1910)

2026-10-23 16:39:56 +02:00
parent c0f24f56b5
commit 84b2c92ccf
4 changed files with 131 additions and 0 deletions
@@ -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 `<node>.<ask id>` 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
+}
@@ -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,
@@ -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)
@@ -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)
+		}
+	}
+}