From cc1c0c5d94119f8ab8e87e9b3e27e0fe36eae675 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 23:01:54 +0100 Subject: [PATCH] =?UTF-8?q?D4b:=20the=20asker=20drops=20an=20ask=20it=20st?= =?UTF-8?q?ops=20waiting=20for=20on=20the=20node=20that=20last=20ran=20it:?= =?UTF-8?q?=20given=20again=20past=20its=20bound,=20or=20a=20named=20cause?= =?UTF-8?q?=20(ADR=200664=20=C2=A75,=20issue=201748)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ --- .../mesh-delivery-builder/buildrun/asks.go | 44 +++++++++++++++++++ .../buildrun/asks_test.go | 28 ++++++++++++ .../cmd/mesh-delivery-builder/main.go | 2 +- .../mesh-delivery-builder/manifest_test.go | 2 +- modules/mesh-delivery-builder/module.json | 3 ++ .../cmd/mesh-delivery-checker/main.go | 4 +- .../mesh-delivery-checker/manifest_test.go | 4 +- modules/mesh-delivery-checker/module.json | 2 + 8 files changed, 83 insertions(+), 6 deletions(-) diff --git a/modules/mesh-delivery-builder/buildrun/asks.go b/modules/mesh-delivery-builder/buildrun/asks.go index bb60a103..ca0b022c 100644 --- a/modules/mesh-delivery-builder/buildrun/asks.go +++ b/modules/mesh-delivery-builder/buildrun/asks.go @@ -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 .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) } diff --git a/modules/mesh-delivery-builder/buildrun/asks_test.go b/modules/mesh-delivery-builder/buildrun/asks_test.go index 71694f56..80e4c389 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_test.go +++ b/modules/mesh-delivery-builder/buildrun/asks_test.go @@ -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: " ". +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 +} 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 ff15f021..224b428b 100644 --- a/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go +++ b/modules/mesh-delivery-builder/cmd/mesh-delivery-builder/main.go @@ -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}, diff --git a/modules/mesh-delivery-builder/manifest_test.go b/modules/mesh-delivery-builder/manifest_test.go index aa10d1bd..6e883afc 100644 --- a/modules/mesh-delivery-builder/manifest_test.go +++ b/modules/mesh-delivery-builder/manifest_test.go @@ -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) } diff --git a/modules/mesh-delivery-builder/module.json b/modules/mesh-delivery-builder/module.json index d9c7c566..2f4140d0 100644 --- a/modules/mesh-delivery-builder/module.json +++ b/modules/mesh-delivery-builder/module.json @@ -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", diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go index b178e634..9a3a26e6 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go @@ -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, diff --git a/modules/mesh-delivery-checker/manifest_test.go b/modules/mesh-delivery-checker/manifest_test.go index 22e1f177..ac47d009 100644 --- a/modules/mesh-delivery-checker/manifest_test.go +++ b/modules/mesh-delivery-checker/manifest_test.go @@ -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) diff --git a/modules/mesh-delivery-checker/module.json b/modules/mesh-delivery-checker/module.json index 7acb3125..49efcfe4 100644 --- a/modules/mesh-delivery-checker/module.json +++ b/modules/mesh-delivery-checker/module.json @@ -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" ],