C5: an integration test of the delayed redelivery and the drop of a running step against the real bus (ADR 0664 §1, §5, issue 1748)

Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ

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 {

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) }