novox/mesh-catalog
9
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

C1: a build step that ends on a transport fault below its last run is written again and handed back to the bus with a delay (ADR 0664 §1, issue 1748)

Browse Source 
Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
jschoubben
2026-11-09 23:04:44 +01:00
parent fd9147087e
commit 5a7561fe1d
6 changed files with 98 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-build-go/buildstep/pull.go
+19 -1
View File
@@ -32,6 +32,9 @@ type Pulled struct {
// Ack ends it: it is never delivered again. InProgress tells the bus it is still being worked, restarting its
// acknowledgement wait. Nak hands it back to be delivered again at once (to this node or another). Term drops it.
Ack, InProgress, Nak, Term func() error
// NakAfter hands it back to be delivered again after delay, not before (a step that ended on a transport fault, ADR 0664
// §1); nil when the source cannot, and a refusal (a tool runner that does not state the capability) is an error.
NakAfter func(delay time.Duration) error
}
// AskSource is where a puller takes asks from: the seat's workers on the bus (JetStreamAsks) or a test double
@@ -616,7 +619,7 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) {
p.decline(msg, ask, lacks)
return
}
step, key, err := begin(p.Put, p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)))
step, key, err := beginRun(p.Put, p.Node, ask.Kind+" ask", ask.ID, buildrun.AskWhat(askRead(ask)), int(msg.Delivered))
if err != nil {
p.logf("the ask %s cannot start on %s: %v", ask.ID, p.Node, err)
_ = msg.Nak()
@@ -688,6 +691,21 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key
// The process stops: the ask is neither ended nor acknowledged here, so the stream gives it to a node that runs.
return
}
if run := int(msg.Delivered); err != nil && sdkstep.IsPassing(err) && run < sdkstep.AskRuns && msg.NakAfter != nil {
// A transport fault (ADR 0664 §1): the step says it is given again and the bus gives it after a delay, to this node or
// another; it is not acknowledged, so its asker sees no end. Run AskRuns times it ends failed, said as a transport fault.
delay := sdkstep.AgainAfter(run)
if aerr := again(p.Put, key, step, err, delay); aerr == nil {
p.logf("the %s ask %s ended on %s on a transport fault in its run %d (%v), and is given again in %s", ask.Kind, ask.ID, p.Node, run, err, delay)
if nerr := msg.NakAfter(delay); nerr != nil {
p.logf("the ask %s could not be handed back with a delay (%v), so it is handed back at once", ask.ID, nerr)
_ = msg.Nak()
}
return
} else {
p.logf("the ask %s ended on a transport fault and its step could not be written again (%v): it ends failed", ask.ID, aerr)
}
}
if ferr := finish(p.Put, key, step, result, err, p.Logf); ferr != nil {
// No end was written: the ask is not acknowledged as done, so the stream gives it again and a run that can
// write its end (or the poison rule, issue 1907) ends it where its asker sees it (issue 1930).
modules/mesh-build-go/buildstep/pull_bus.go
+2 -1
View File
@@ -70,7 +70,8 @@ func (s *JetStreamAsks) Next(ctx context.Context, kind string, wait time.Duratio
delivered = meta.NumDelivered
}
return &Pulled{Data: append([]byte(nil), msg.Data()...), Delivered: delivered,
Ack: msg.Ack, InProgress: msg.InProgress, Nak: msg.Nak, Term: msg.Term}, nil
Ack: msg.Ack, InProgress: msg.InProgress, Nak: msg.Nak, Term: msg.Term,
NakAfter: func(delay time.Duration) error { return msg.NakWithDelay(delay) }}, nil
case <-ctx.Done():
return nil, ctx.Err()
}
modules/mesh-build-go/buildstep/pull_fake.go
+11 -2
View File
@@ -16,7 +16,10 @@ type FakeAsks struct {
added chan struct{}
// Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given.
Acked, Termed, Naked []string
InProgress int
// Delayed are the asks handed back to be given again after a delay (NakAfter), and Delays that delay of each.
Delayed []string
Delays []time.Duration
InProgress int
// Delivered, when set, is the delivery count every ask handed over says (1 when 0): a test of a poisoned ask sets it.
Delivered uint64
}
@@ -108,6 +111,12 @@ func (f *FakeAsks) pulled(data []byte) *Pulled {
if delivered == 0 {
delivered = 1
}
return &Pulled{Data: data, Delivered: delivered, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed),
nakAfter := func(delay time.Duration) error {
f.mu.Lock()
f.Delayed, f.Delays = append(f.Delayed, id), append(f.Delays, delay)
f.mu.Unlock()
return nil
}
return &Pulled{Data: data, Delivered: delivered, NakAfter: nakAfter, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed),
InProgress: func() error { f.mu.Lock(); f.InProgress++; f.mu.Unlock(); return nil }}
}
modules/mesh-build-go/buildstep/pull_seat.go
+8 -1
View File
@@ -16,6 +16,8 @@ type SeatAsks struct {
// Pull and Settle are stdio.SeatPull and stdio.SeatSettle when nil; a test gives its own.
Pull func(worker string, wait time.Duration) (*stdio.Pulled, error)
Settle func(id, fate string) error
// SettleAfter is stdio.SeatSettleAfter when nil (ADR 0664 §1).
SettleAfter func(id string, delay time.Duration) error
}
// Next is AskSource's.
@@ -38,7 +40,12 @@ func (s SeatAsks) Next(ctx context.Context, kind string, wait time.Duration) (*P
if err != nil || p == nil {
return nil, err
}
after := s.SettleAfter
if after == nil {
after = stdio.SeatSettleAfter
}
fate := func(f string) func() error { return func() error { return settle(p.ID, f) } }
return &Pulled{Data: p.Body, Delivered: p.Delivered, Ack: fate(stdio.SettleAck), InProgress: fate(stdio.SettleProgress),
Nak: fate(stdio.SettleNak), Term: fate(stdio.SettleTerm)}, nil
Nak: fate(stdio.SettleNak), Term: fate(stdio.SettleTerm),
NakAfter: func(delay time.Duration) error { return after(p.ID, delay) }}, nil
}
modules/mesh-build-go/buildstep/pull_test.go
+43
View File
@@ -518,6 +518,49 @@ func TestAPoisonedAskIsSaidAndRunNoMore(t *testing.T) {
if acked, _, naked := src.Fates(); len(acked)+len(naked) != 0 {
t.Errorf("acked %v naked %v, want it termed alone", acked, naked)
}
// ADR 0664 §1 (issue 1748): a run that ends on a transport fault below AskRuns is written again, with its run and when the
// bus gives it, and handed back with that delay, not acknowledged; the AskRuns-th ends failed, passing; a fault of the
// source ends failed at once, in its first run.
for _, c := range []struct {
name string
run uint64
err error
state string
delay time.Duration
passing bool
ackedOnly bool
}{
{"a transport fault in run 1", 1, &sdkstep.Passing{Err: errors.New("lookup proxy: i/o timeout")}, sdkstep.StepAgain, time.Minute, true, false},
{"a transport fault in run 2", 2, &sdkstep.Passing{Err: errors.New("lookup proxy: i/o timeout")}, sdkstep.StepAgain, 4 * time.Minute, true, false},
{"a transport fault in the last run", sdkstep.AskRuns, &sdkstep.Passing{Err: errors.New("lookup proxy: i/o timeout")}, sdkstep.StepFailed, 0, true, true},
{"a compile error in run 1", 1, errors.New("undefined: x"), sdkstep.StepFailed, 0, false, true},
} {
src, st := NewFakeAsks(), newSteps()
src.Delivered = c.run
a := testAsk(sdkstep.AskDelivery)
_ = src.PublishAsk(a)
p := &Puller{Source: src, Language: sdkstep.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a sdkstep.StepAsk) (any, error) { return nil, c.err }}
ctx, cancel := context.WithCancel(context.Background())
go p.Pull(ctx)
key, _ := sdkstep.AskStepKey("shanks", a.ID)
waitFor(t, c.name, func() bool {
ack, _, _ := src.Fates()
src.mu.Lock()
defer src.mu.Unlock()
return len(ack)+len(src.Delayed) == 1
})
cancel()
s, _ := st.get(key)
src.mu.Lock()
delays := append([]time.Duration(nil), src.Delays...)
src.mu.Unlock()
acked, _, _ := src.Fates()
if s.State != c.state || s.Passing != c.passing || s.Run != int(c.run) || (len(acked) == 1) != c.ackedOnly || (!c.ackedOnly && (len(delays) != 1 || delays[0] != c.delay)) ||
(c.state == sdkstep.StepAgain && s.AgainAt.IsZero()) {
t.Errorf("%s: step %+v, acked %v, delays %v; want %s, passing %v, delay %v", c.name, s, acked, delays, c.state, c.passing, c.delay)
}
}
}
// An ask whose job this build module does not know is answered on its step key as refused, naming why, never only dropped
modules/mesh-build-go/buildstep/step.go
+15 -1
View File
@@ -44,7 +44,12 @@ func Start(ctx context.Context, put Put, node, tool string, run func(ctx context
// begin writes the step running under id, owned by this process (Owner).
func begin(put Put, node, tool, id, what string) (sdkstep.Step, string, error) {
step := sdkstep.Step{ID: id, Tool: tool, Node: node, State: sdkstep.StepRunning, Started: now().UTC(), What: what, Owner: buildrun.Owner}
return beginRun(put, node, tool, id, what, 0)
}
// beginRun is begin for the run-th run of an ask (the bus's delivery count; 0 for a step that is no ask).
func beginRun(put Put, node, tool, id, what string, run int) (sdkstep.Step, string, error) {
step := sdkstep.Step{ID: id, Tool: tool, Node: node, State: sdkstep.StepRunning, Started: now().UTC(), What: what, Owner: buildrun.Owner, Run: run}
key, err := sdkstep.StepKey(node, step.ID)
if err != nil {
return sdkstep.Step{}, "", err
@@ -83,6 +88,7 @@ func finish(put Put, key string, step sdkstep.Step, result any, err error, logf
step.State, step.Reason = sdkstep.StepRefused, err.Error()
default:
step.State, step.Reason = sdkstep.StepFailed, err.Error()
step.Passing = sdkstep.IsPassing(err) // it ended on a transport fault: its asker may ask it again (ADR 0664 §3)
// A failed step may answer what it found on the way (a build's out-of-step files): kept on its end.
if result != nil {
if raw, merr := json.Marshal(result); merr == nil {
@@ -112,6 +118,14 @@ func finish(put Put, key string, step sdkstep.Step, result any, err error, logf
return nil
}
// again writes the step as given again after a transport fault: the state again with the run's cause, and when the bus gives
// the ask again (ADR 0664 §1). The key is not ended: its asker keeps waiting on it.
func again(put Put, key string, step sdkstep.Step, err error, delay time.Duration) error {
step.State, step.Reason, step.Passing, step.Result = sdkstep.StepAgain, err.Error(), true, nil
step.Ended, step.AgainAt = time.Time{}, now().UTC().Add(delay)
return writeEnd(put, key, step)
}
// writeEnd writes the step's end, trying EndAttempts times.
func writeEnd(put Put, key string, step sdkstep.Step) error {
for attempt := 1; ; attempt++ {