diff --git a/go/buildstep/pull_integration_test.go b/go/buildstep/pull_integration_test.go index 7240e9b..c210e4b 100644 --- a/go/buildstep/pull_integration_test.go +++ b/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) } diff --git a/go/buildstep/pull_test.go b/go/buildstep/pull_test.go index a2c0fb4..e0d7af6 100644 --- a/go/buildstep/pull_test.go +++ b/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) + } +}