Keep a long ask alive with in-progress signals and leave a stopped holder's ask to the stream, so a node that dies hands its ask on (issue 1887 A2c, ADR 0525 §6)

go/buildstep/pull_integration_test.go

@@ -4,6 +4,7 @@ package buildstep
 
 import (
 	"context"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -96,3 +97,48 @@ func TestPullOrderAndCapacityOnTheBus(t *testing.T) {
 		return err == nil && si.State.Msgs == 0
 	})
 }
+
+// 1887/A2c, ADR 0525 §6: a killed puller's ask is redelivered by the stream's acknowledgement wait to the other node,
+// whose step key names the reference.
+func TestPullRedeliversAKilledPullersAsk(t *testing.T) {
+	js := askBus(t, 2*time.Second)
+	a := testAsk(delivery.AskDelivery)
+	publishAsk(t, js, a)
+	st := newSteps()
+	taken := make(chan struct{})
+	dying := &Puller{Source: &JetStreamAsks{JS: js, Language: delivery.AskGo}, Language: delivery.AskGo, Node: "ace", Put: st.put,
+		Capacity: func() int { return 1 }, PullWait: 200 * time.Millisecond, Heartbeat: time.Hour,
+		Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
+			close(taken)
+			<-ctx.Done()
+			return nil, ctx.Err()
+		}}
+	kill, killed := context.WithCancel(context.Background())
+	go dying.Pull(kill)
+	<-taken
+	killed()
+	other := &Puller{Source: &JetStreamAsks{JS: js, Language: delivery.AskGo}, Language: delivery.AskGo, Node: "shanks", Put: st.put,
+		Capacity: func() int { return 1 }, PullWait: 200 * time.Millisecond,
+		Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
+			return delivery.BuildAnswer{Artifacts: []delivery.AskBuilt{{Artifact: "bundle", Digest: "sha256:00", Reference: "artifact-store://x"}}}, nil
+		}}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go other.Pull(ctx)
+	key, _ := delivery.AskStepKey("shanks", a.ID)
+	deadline := time.Now().Add(10 * time.Second)
+	for {
+		if s, ok := st.get(key); ok && s.State == delivery.StepPassed {
+			if want := `artifact-store://x`; !contains(string(s.Result), want) {
+				t.Fatalf("the step's result %s names no %s", s.Result, want)
+			}
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("the killed puller's ask was not taken by shanks within 10 s")
+		}
+		time.Sleep(20 * time.Millisecond)
+	}
+}
+
+func contains(s, sub string) bool { return strings.Contains(s, sub) }

go/buildstep/pull_test.go

@@ -231,3 +231,35 @@ func TestPullPauseOutlivesARestart(t *testing.T) {
 	}
 	waitFor(t, "the ask taken after the resume", func() bool { a, _, _ := src.Fates(); return len(a) == 1 })
 }
+
+// 1887/A2c: a long step tells the bus it is still being worked; a holder that stops mid-step neither ends nor
+// acknowledges its ask, so the stream gives it to another node.
+func TestPullHeartbeatAndStop(t *testing.T) {
+	src, st := NewFakeAsks(), newSteps()
+	a := testAsk(delivery.AskDelivery)
+	_ = src.PublishAsk(a)
+	running := make(chan struct{})
+	p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 },
+		PullWait: 10 * time.Millisecond, Heartbeat: 5 * time.Millisecond,
+		Run: func(ctx context.Context, a delivery.StepAsk) (any, error) {
+			close(running)
+			<-ctx.Done()
+			return nil, ctx.Err()
+		}}
+	ctx, cancel := context.WithCancel(context.Background())
+	stopped := make(chan struct{})
+	go func() { p.Pull(ctx); close(stopped) }()
+	<-running
+	waitFor(t, "two heartbeats", func() bool { src.mu.Lock(); defer src.mu.Unlock(); return src.InProgress >= 2 })
+	cancel()
+	<-stopped
+	waitFor(t, "the step's goroutine gone", func() bool { return p.InHand() == 0 })
+	acked, termed, naked := src.Fates()
+	if len(acked)+len(termed)+len(naked) != 0 {
+		t.Fatalf("a stopped holder gave its ask a fate: acked %v termed %v naked %v", acked, termed, naked)
+	}
+	key, _ := delivery.AskStepKey("shanks", a.ID)
+	if s, ok := st.get(key); !ok || s.State != delivery.StepRunning {
+		t.Fatalf("the step %s = %+v, want left running", key, s)
+	}
+}