diff --git a/modules/mesh-build-go/buildstep/pull.go b/modules/mesh-build-go/buildstep/pull.go index b17eb790f..10c097416 100644 --- a/modules/mesh-build-go/buildstep/pull.go +++ b/modules/mesh-build-go/buildstep/pull.go @@ -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). diff --git a/modules/mesh-build-go/buildstep/pull_bus.go b/modules/mesh-build-go/buildstep/pull_bus.go index 9f2a8f12a..8633ed77b 100644 --- a/modules/mesh-build-go/buildstep/pull_bus.go +++ b/modules/mesh-build-go/buildstep/pull_bus.go @@ -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() } diff --git a/modules/mesh-build-go/buildstep/pull_fake.go b/modules/mesh-build-go/buildstep/pull_fake.go index ddefb2338..84ed06d18 100644 --- a/modules/mesh-build-go/buildstep/pull_fake.go +++ b/modules/mesh-build-go/buildstep/pull_fake.go @@ -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 }} } diff --git a/modules/mesh-build-go/buildstep/pull_seat.go b/modules/mesh-build-go/buildstep/pull_seat.go index 9441618b7..105f410c3 100644 --- a/modules/mesh-build-go/buildstep/pull_seat.go +++ b/modules/mesh-build-go/buildstep/pull_seat.go @@ -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 } diff --git a/modules/mesh-build-go/buildstep/pull_test.go b/modules/mesh-build-go/buildstep/pull_test.go index b538dccba..d30cd77c4 100644 --- a/modules/mesh-build-go/buildstep/pull_test.go +++ b/modules/mesh-build-go/buildstep/pull_test.go @@ -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 diff --git a/modules/mesh-build-go/buildstep/step.go b/modules/mesh-build-go/buildstep/step.go index c145958ff..5981000b3 100644 --- a/modules/mesh-build-go/buildstep/step.go +++ b/modules/mesh-build-go/buildstep/step.go @@ -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++ {