D4b: the asker drops an ask it stops waiting for on the node that last ran it: given again past its bound, or a named cause (ADR 0664 §5, issue 1748)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
2026-11-09 23:01:54 +01:00
parent bfc73581ec
commit cc1c0c5d94
8 changed files with 83 additions and 6 deletions
@@ -37,6 +37,42 @@ type Asker struct {
 	// before the asker stops waiting; a step that keeps moving is waited for up to 3 Bounds from its start (issue 1930); zero waits as long as ctx. An ask waiting in its stream is not bounded: the stream's backlog is no fault
 	// (ADR 0525 §6, §9), and the bound starts again when another node takes the ask.
 	Bound time.Duration
+	// Dropper, when set, is told to end an ask the asker stops waiting for without its end (ADR 0664 §5): a step given
+	// again that the bus did not give within its bound, or a wait ended with a cause its caller names (a superseded
+	// fingerprint). Never a wait ended by a shutdown (a plain cancel): the ask is then taken over by the next run.
+	Dropper AskDropper
+}
+
+// AskDropper ends an ask on the build node that last ran it: the seat's verb drop (it writes the ask in the seat holder's
+// own state of pauses, so the holder refuses it when it is given again and ends it while it runs).
+type AskDropper interface {
+	DropAsk(ctx context.Context, node string, a buildstep.StepAsk, why string) error
+}
+
+// SeatDrops is an AskDropper that calls the verb drop of the ask's build seat on node over the bus.
+type SeatDrops struct{ Caller Caller }
+
+// DropAsk calls <seat>.drop on node for a, naming its asker and why.
+func (d SeatDrops) DropAsk(_ context.Context, node string, a buildstep.StepAsk, why string) error {
+	seat, err := buildstep.BuildSeat(a.Language)
+	if err != nil {
+		return err
+	}
+	if _, err := d.Caller.CallOn(node, "seat:"+seat+".drop", map[string]any{"id": a.ID, "by": a.Asker, "why": why}); err != nil {
+		return fmt.Errorf("dropping the ask %s on %s: %w", a.ID, node, err)
+	}
+	return nil
+}
+
+// dropped ends a without its end, on the node that last ran it; an ask no node ran has nothing to drop on a node (the
+// message left on its stream is the builder's own Drop). A failed drop is not the wait's error: the wait says why it ended.
+func (k Asker) dropped(a buildstep.StepAsk, node, why string) {
+	if k.Dropper == nil || node == "" {
+		return
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+	_ = k.Dropper.DropAsk(ctx, node, a, why)
 }

 // Put puts a on its stream: the seat's accept subject of its language and kind.
@@ -80,6 +116,7 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e
 		}
 	}()
 	running, since, phase, phaseAt := "", time.Time{}, "", time.Time{}
+	lastNode := ""
 	var agains []string // the node of each run that ended on a transport fault, in order (ADR 0664 §3)
 	standStill := func() error {
 		if phase == "" {
@@ -91,8 +128,14 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e
 	for {
 		select {
 		case <-ctx.Done():
+			if cause := context.Cause(ctx); cause != nil && !errors.Is(cause, context.Canceled) {
+				k.dropped(a, lastNode, cause.Error())
+			}
 			return buildstep.Step{}, ctx.Err()
 		case <-bound:
+			if len(agains) > 0 && phase == "given again after a transport fault" {
+				k.dropped(a, lastNode, "the bus did not give it again within its bound")
+			}
 			return buildstep.Step{}, standStill()
 		case <-outerCap:
 			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",
@@ -101,6 +144,7 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e
 			if !ok {
 				return buildstep.Step{}, errors.New("the watch of the ask's step ended")
 			}
+			lastNode = step.Node
 			if seen := stepSeenOf(ctx); seen != nil {
 				seen(step)
 			}
@@ -71,6 +71,23 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
 	if _, err := (Asker{Publisher: h, Waiter: h, Bound: 20 * time.Millisecond}).Wait(context.Background(), goAsk("mvl0frv18d1973")); err == nil || !strings.Contains(err.Error(), "given again after a transport fault") {
 		t.Errorf("an ask never given again: %v", err)
 	}
+	// ADR 0664 §5: that ask is dropped on the node that ran it; a wait ended by a named cause is dropped too; a shutdown's
+	// plain cancel is not, and nor is an ask no node ran.
+	drops := &seatDrops{}
+	h = &FakeAsks{}
+	h.Write(buildstep.Step{ID: "mvl0frv18d1974", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now(), Started: time.Now()})
+	(Asker{Publisher: h, Waiter: h, Bound: 20 * time.Millisecond, Dropper: drops}).Wait(context.Background(), goAsk("mvl0frv18d1974"))
+	named, stop := context.WithCancelCause(context.Background())
+	h.Write(buildstep.Step{ID: "mvl0frv18d1975", Node: "g14", State: buildstep.StepRunning, Started: time.Now()})
+	seenStep := WithStepSeen(named, func(buildstep.Step) { stop(errors.New("superseded by a newer fingerprint")) })
+	(Asker{Publisher: h, Waiter: h, Bound: time.Hour, Dropper: drops}).Wait(seenStep, goAsk("mvl0frv18d1975"))
+	shut, off := context.WithCancel(context.Background())
+	off()
+	(Asker{Publisher: h, Waiter: h, Bound: time.Hour, Dropper: drops}).Wait(shut, goAsk("mvl0frv18d1977"))
+	(Asker{Publisher: &FakeAsks{}, Waiter: &FakeAsks{}, Bound: time.Hour, Dropper: drops}).Wait(named, goAsk("mvl0frv18d1976"))
+	if got := strings.Join(drops.got, "|"); !strings.Contains(got, "ace mvl0frv18d1974") || !strings.Contains(got, "g14 mvl0frv18d1975 superseded by a newer fingerprint") || len(drops.got) != 2 {
+		t.Errorf("drops %q, want the ask never given again on ace and the superseded one on g14 only", got)
+	}
 	// review round 2 item 1: an ask that waits in its stream for a free slot is not bounded; once a node writes its step
 	// running, the step is bounded from its start, and the error names the node.
 	g := &FakeAsks{}
@@ -266,3 +283,14 @@ func (c *closingWatch) WatchAsk(ctx context.Context, module, askID string) (<-ch
 	}
 	return c.inner.WatchAsk(ctx, module, askID)
 }
+
+// seatDrops keeps the asks an Asker dropped: "<node> <id> <why>".
+type seatDrops struct{ got []string }
+
+func (d *seatDrops) DropAsk(_ context.Context, node string, a buildstep.StepAsk, why string) error {
+	if strings.Contains(why, "bound") {
+		why = ""
+	}
+	d.got = append(d.got, strings.TrimSpace(node+" "+a.ID+" "+why))
+	return nil
+}
@@ -175,7 +175,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	numberer.Publishing = pubs.arm
 	b := &build.Builder{
 		Claimer: build.Claimer{Store: kvStore{buildsKV}, By: cred.Node + "." + cred.Module},
-		Asks:    buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound},
+		Asks:    buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound, Dropper: buildrun.SeatDrops{Caller: calls}},
 		Held:    heldOnTheBus{js},
 		// An ask that can no longer be answered is asked again (issue 1945), read at the step bound: a context's bound.
 		Gone:       heldOnTheBus{js},
@@ -125,7 +125,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	// Design 51 *The floor*: the floor's three sources, the registry seats' versions, bridge_tags and indexer_requirements.
 	if want := []string{"mesh-delivery-assigner.assigner_read_assignments", "mesh-delivery-forge-bridge.bridge_tags", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
 		"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_requirements", "seat:container-image-registry.remove", "seat:container-image-registry.versions",
-		"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
+		"seat:go-package-registry.remove", "seat:go-package-registry.versions", "seat:mesh-artifact-sweep.held", "seat:mesh-hand-acts.hand-act", "seat:node-build-go.drop", "seat:node-build-image.drop", "seat:node-build-node.drop", "seat:node-engine.manifest-reads", "seat:npm-package-registry.remove",
 		"seat:npm-package-registry.versions", "seat:oci-artifact-registry.remove", "seat:oci-artifact-registry.versions"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
 		t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want)
 	}
@@ -67,6 +67,9 @@
    "seat:go-package-registry.versions",
    "seat:mesh-artifact-sweep.held",
    "seat:mesh-hand-acts.hand-act",
+    "seat:node-build-go.drop",
+    "seat:node-build-image.drop",
+    "seat:node-build-node.drop",
    "seat:node-engine.manifest-reads",
    "seat:npm-package-registry.remove",
    "seat:npm-package-registry.versions",
@@ -262,7 +262,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		Asker:    engineAsks{call: caller.CallOn},
 		Placed:   placedNodes{kv: func(context.Context) (jetstream.KeyValue, error) { return assignments, nil }}.Placed,
 		Bound:    answerBound,
-		Sandbox:  seatSandbox{asker: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound}},
+		Sandbox:  seatSandbox{asker: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}}},
 		Verdicts: verdicts.all,
 		// The node-engine's trunk head: a group with no node-engine member composes with the composer built there (ADR 0656 §5).
 		Composer: func() string { return trunks.Heads()[check.NodeEngineRepository] },
@@ -304,7 +304,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}
 	}
 	ch.Judge = newJudge(judgeDeps{
-		asks: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound}, tools: tools, trunks: trunks,
+		asks: buildrun.Asker{Publisher: asks, Waiter: waiter, Bound: stepBound, Dropper: buildrun.SeatDrops{Caller: caller}}, tools: tools, trunks: trunks,
 		kept: kvStore{buckets[check.CheckerKeptBucket]}, previews: kvStore{buckets[check.PreviewsBucket]}, candidates: tryCandidates{candidates},
 		repos: repos, published: published, builds: buildStates, ends: ends, buildWait: set.buildWait, forge: forgeReads{call: caller.Call}, logf: logger.Printf,
 		announce: announce, compositions: compositions, tags: tags, inputs: runInputs{awaitedKV{inputsKV}}, beginBound: beginBound,
@@ -131,8 +131,8 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
 		"mesh-delivery-assigner.assigner_read_assignments", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
 		"mesh-delivery-indexer.indexer_read_modules", "mesh-delivery-indexer.indexer_read_packages", "mesh-delivery-indexer.indexer_read_predefined",
 		"mesh-delivery-indexer.indexer_read_seat_definitions",
-		"seat:mesh-catalog.seat-definitions", "seat:mesh-config.renames", "seat:mesh-hand-acts.hand-act", "seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
-		t.Errorf("invokes = %v, want the forge's reads and compose-candidate", m.Invokes)
+		"seat:mesh-catalog.seat-definitions", "seat:mesh-config.renames", "seat:mesh-hand-acts.hand-act", "seat:node-build-go.drop", "seat:node-build-node.drop", "seat:node-engine.compose-candidate"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
+		t.Errorf("invokes = %v, want the forge's reads, the seats' drop and compose-candidate", m.Invokes)
 	}
 	if want := []string{"node-build-go", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
 		t.Errorf("uses = %v, want the Go and Node build seats", m.Uses)
@@ -93,6 +93,8 @@
    "seat:mesh-catalog.seat-definitions",
    "seat:mesh-config.renames",
    "seat:mesh-hand-acts.hand-act",
+    "seat:node-build-go.drop",
+    "seat:node-build-node.drop",
    "seat:node-engine.compose-candidate"
  ],