From fe6552ba7c77061afb62afd2624ba79eb1d4ecc8 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 23:18:15 +0100 Subject: [PATCH] =?UTF-8?q?D4b:=20the=20asker=20calls=20the=20build=20seat?= =?UTF-8?q?'s=20drop=20on=20whichever=20holder=20answers,=20no=20node=20na?= =?UTF-8?q?med,=20and=20for=20an=20ask=20no=20node=20took=20(ADR=200664=20?= =?UTF-8?q?=C2=A74,=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 | 40 ++++++++++--------- .../buildrun/asks_test.go | 14 +++---- 2 files changed, 29 insertions(+), 25 deletions(-) diff --git a/modules/mesh-delivery-builder/buildrun/asks.go b/modules/mesh-delivery-builder/buildrun/asks.go index ca0b022c..f96af4d5 100644 --- a/modules/mesh-delivery-builder/buildrun/asks.go +++ b/modules/mesh-delivery-builder/buildrun/asks.go @@ -2,6 +2,7 @@ package buildrun import ( "context" + "encoding/json" "errors" "fmt" "strings" @@ -43,36 +44,41 @@ type Asker struct { 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). +// AskDropper ends an ask: the seat's verb drop, served by any holder of the build seat (no node is named). It writes the +// ask in the build module's shared state of pauses, so every node refuses it when the stream gives it, and the node that +// runs it ends the step (ADR 0664 §5). type AskDropper interface { - DropAsk(ctx context.Context, node string, a buildstep.StepAsk, why string) error + DropAsk(ctx context.Context, 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 } +// SeatCaller calls a seat's verb on whichever holder answers (mesh-sdk process.Conn.Call). +type SeatCaller interface { + Call(key string, body any) (json.RawMessage, error) +} -// 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 { +// SeatDrops is an AskDropper that calls the verb drop of the ask's build seat over the bus. +type SeatDrops struct{ Caller SeatCaller } + +// DropAsk calls .drop for a, naming its asker and why. +func (d SeatDrops) DropAsk(_ context.Context, 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) + if _, err := d.Caller.Call("seat:"+seat+".drop", map[string]any{"id": a.ID, "by": a.Asker, "why": why}); err != nil { + return fmt.Errorf("dropping the ask %s: %w", a.ID, 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 == "" { +// dropped ends a without its end. A failed drop is not the wait's error: the wait says why it ended. +func (k Asker) dropped(a buildstep.StepAsk, why string) { + if k.Dropper == nil { return } ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() - _ = k.Dropper.DropAsk(ctx, node, a, why) + _ = k.Dropper.DropAsk(ctx, a, why) } // Put puts a on its stream: the seat's accept subject of its language and kind. @@ -116,7 +122,6 @@ 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 == "" { @@ -129,12 +134,12 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e select { case <-ctx.Done(): if cause := context.Cause(ctx); cause != nil && !errors.Is(cause, context.Canceled) { - k.dropped(a, lastNode, cause.Error()) + k.dropped(a, 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") + k.dropped(a, "the bus did not give it again within its bound") } return buildstep.Step{}, standStill() case <-outerCap: @@ -144,7 +149,6 @@ 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 80e4c389..b80fc1ab 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_test.go +++ b/modules/mesh-delivery-builder/buildrun/asks_test.go @@ -71,8 +71,8 @@ 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. + // ADR 0664 §5: that ask is dropped; a wait ended by a named cause is dropped too, an ask no node took included (it waits in + // its stream); a shutdown's plain cancel is not. drops := &seatDrops{} h = &FakeAsks{} h.Write(buildstep.Step{ID: "mvl0frv18d1974", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now(), Started: time.Now()}) @@ -85,8 +85,8 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { 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) + if got := strings.Join(drops.got, "|"); !strings.Contains(got, "mvl0frv18d1974") || !strings.Contains(got, "mvl0frv18d1975 superseded by a newer fingerprint") || !strings.Contains(got, "mvl0frv18d1976 superseded by a newer fingerprint") || len(drops.got) != 3 { + t.Errorf("drops %q, want the ask never given again, the superseded one and the one no node took, and no other", 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. @@ -284,13 +284,13 @@ 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: " ". +// 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 { +func (d *seatDrops) DropAsk(_ context.Context, a buildstep.StepAsk, why string) error { if strings.Contains(why, "bound") { why = "" } - d.got = append(d.got, strings.TrimSpace(node+" "+a.ID+" "+why)) + d.got = append(d.got, strings.TrimSpace(a.ID+" "+why)) return nil }