novox/mesh-delivery
10
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

D4b: the asker calls the build seat's drop on whichever holder answers, no node named, and for an ask no node took (ADR 0664 §4, §5, issue 1748)

Browse Source 
Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-09 23:18:15 +01:00
parent c072a271e5
commit fe6552ba7c
2 changed files with 29 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-builder/buildrun/asks.go
+22 -18
View File
@@ -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 <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 {
// 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 <seat>.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)
}
modules/mesh-delivery-builder/buildrun/asks_test.go
+7 -7
View File
@@ -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: "<node> <id> <why>".
// seatDrops keeps the asks an Asker dropped: "<id> <why>".
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
}