diff --git a/modules/mesh-build-go/buildstep/pull.go b/modules/mesh-build-go/buildstep/pull.go index 70a81162e..7f55b158c 100644 --- a/modules/mesh-build-go/buildstep/pull.go +++ b/modules/mesh-build-go/buildstep/pull.go @@ -67,7 +67,10 @@ type Puller struct { // Run runs one ask's job; a *Refusal ends its step refused. Its result is the step's result. Run func(ctx context.Context, ask sdkstep.StepAsk) (any, error) // PullWait and Heartbeat default to DefaultPullWait and DefaultHeartbeat. - PullWait time.Duration + PullWait time.Duration + // Delay is how long the bus waits before it gives an ask again after its run-th run ended on a transport fault; nil is + // sdkstep.AgainAfter (a minute, then four). A test's own is short. + Delay func(run int) time.Duration Heartbeat time.Duration Logf func(format string, args ...any) // Load reads one key of the state the node's pause is kept in (nil and no error when it is absent). When set, Pull @@ -715,6 +718,9 @@ func (p *Puller) run(ctx context.Context, msg *Pulled, ask sdkstep.StepAsk, key // 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 p.Delay != nil { + delay = p.Delay(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 { diff --git a/modules/mesh-build-go/buildstep/pull_integration_test.go b/modules/mesh-build-go/buildstep/pull_integration_test.go index 5fee7d300..38a5cc836 100644 --- a/modules/mesh-build-go/buildstep/pull_integration_test.go +++ b/modules/mesh-build-go/buildstep/pull_integration_test.go @@ -4,6 +4,7 @@ package buildstep import ( "context" + "errors" "strings" "sync" "testing" @@ -141,4 +142,57 @@ func TestPullRedeliversAKilledPullersAsk(t *testing.T) { } } +// ADR 0664 §1, §5 (issue 1748) on the real server: a run that ends on a transport fault is written again and the bus gives the +// ask again after the delay, not before, as the second run (its delivery count one higher), and that run's end is the ask's; +// a step that runs when its ask is dropped ends refused, naming who dropped it, and the ask is not given again. +func TestAnAskThatMetTheNetworkIsGivenAgainAndADroppedOneIsEnded(t *testing.T) { + js := askBus(t, 30*time.Second) + a, doomed := testAsk(sdkstep.AskDelivery), testAsk(sdkstep.AskDelivery) + publishAsk(t, js, a) + st, kv := newSteps(), &pauseKV{} + const delay = 600 * time.Millisecond + var mu sync.Mutex + var runs []time.Time + hold := make(chan struct{}) + p := &Puller{Source: &JetStreamAsks{JS: js, Language: sdkstep.AskGo}, Language: sdkstep.AskGo, Node: "ace", Put: st.put, PausePut: kv.put, Load: kv.load, + ForgetPause: kv.forget, PauseKeys: kv.keys, Capacity: func() int { return 1 }, PullWait: 100 * time.Millisecond, + Delay: func(int) time.Duration { return delay }, + Run: func(ctx context.Context, got sdkstep.StepAsk) (any, error) { + if got.ID == doomed.ID { + close(hold) + <-ctx.Done() + return nil, ctx.Err() + } + mu.Lock() + runs = append(runs, time.Now()) + n := len(runs) + mu.Unlock() + if n == 1 { + return nil, &sdkstep.Passing{Err: errors.New("lookup git.novox.be: i/o timeout")} + } + return nil, nil + }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go p.Pull(ctx) + key, _ := sdkstep.AskStepKey("ace", a.ID) + waitFor(t, "the ask's second run ended", func() bool { s, ok := st.get(key); return ok && s.State == sdkstep.StepPassed }) + mu.Lock() + gap := runs[1].Sub(runs[0]) + mu.Unlock() + if s, _ := st.get(key); len(runs) != 2 || gap < delay || s.Run != 2 { + t.Errorf("runs %d, %s apart (delay %s), the step's run %d: want two runs, the second after the delay, as run 2", len(runs), gap, delay, s.Run) + } + publishAsk(t, js, doomed) + <-hold + if err := p.Drop(doomed.ID, "mesh-delivery-builder", "superseded"); err != nil { + t.Fatal(err) + } + dkey, _ := sdkstep.AskStepKey("ace", doomed.ID) + waitFor(t, "the dropped step refused", func() bool { s, ok := st.get(dkey); return ok && s.State == sdkstep.StepRefused }) + if s, _ := st.get(dkey); !contains(s.Reason, "dropped by mesh-delivery-builder: superseded") { + t.Errorf("the dropped step says %q", s.Reason) + } +} + func contains(s, sub string) bool { return strings.Contains(s, sub) }