From 3c3fdd34469167b7aee945992dae27ececd99ed6 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 9 Nov 2026 23:16:09 +0100 Subject: [PATCH] =?UTF-8?q?C5:=20an=20integration=20test=20of=20the=20dela?= =?UTF-8?q?yed=20redelivery=20and=20the=20drop=20of=20a=20running=20step?= =?UTF-8?q?=20against=20the=20real=20bus=20(ADR=200664=20=C2=A71,=20=C2=A7?= =?UTF-8?q?5,=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 --- modules/mesh-build-go/buildstep/pull.go | 8 ++- .../buildstep/pull_integration_test.go | 54 +++++++++++++++++++ 2 files changed, 61 insertions(+), 1 deletion(-) 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) }