diff --git a/go/buildstep/pull.go b/go/buildstep/pull.go index 32f243e..9fb676b 100644 --- a/go/buildstep/pull.go +++ b/go/buildstep/pull.go @@ -281,6 +281,15 @@ func (p *Puller) take(ctx context.Context, msg *Pulled) { _ = msg.Nak() return } + if msg.Delivered > delivery.AskRuns { + // Run AskRuns times and never ended: each run took its node down or died with it. It is run no more, and said + // (issue 1907): the asker's key fails with why. + why := fmt.Errorf("the ask was taken %d times and no run of it ended: each took its build node down or died with it, so it is run no more (poison)", msg.Delivered-1) + p.logf("the ask %s is ended on %s: %v", ask.ID, p.Node, why) + finish(p.Put, key, step, nil, why) + _ = msg.Term() + return + } module := "" if ask.Build != nil { module = ask.Build.Module diff --git a/go/buildstep/pull_fake.go b/go/buildstep/pull_fake.go index 2e6624d..1e10e4a 100644 --- a/go/buildstep/pull_fake.go +++ b/go/buildstep/pull_fake.go @@ -17,6 +17,8 @@ type FakeAsks struct { // Acked, Termed and Naked are the asks' ids or data by fate, InProgress how many heartbeats were given. Acked, Termed, Naked []string InProgress int + // Delivered, when set, is the delivery count every ask handed over says (1 when 0): a test of a poisoned ask sets it. + Delivered uint64 } // NewFakeAsks is an empty double. @@ -102,6 +104,10 @@ func (f *FakeAsks) pulled(data []byte) *Pulled { return nil } } - return &Pulled{Data: data, Delivered: 1, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed), + delivered := f.Delivered + if delivered == 0 { + delivered = 1 + } + return &Pulled{Data: data, Delivered: delivered, Ack: fate(&f.Acked), Nak: fate(&f.Naked), Term: fate(&f.Termed), InProgress: func() error { f.mu.Lock(); f.InProgress++; f.mu.Unlock(); return nil }} } diff --git a/go/buildstep/pull_test.go b/go/buildstep/pull_test.go index 10fdedb..6d0b326 100644 --- a/go/buildstep/pull_test.go +++ b/go/buildstep/pull_test.go @@ -3,6 +3,7 @@ package buildstep import ( "context" "encoding/json" + "strings" "sync" "testing" "time" @@ -263,3 +264,27 @@ func TestPullHeartbeatAndStop(t *testing.T) { t.Fatalf("the step %s = %+v, want left running", key, s) } } + +// 1907: an ask taken past AskRuns (each run took its node down) is run no more: its step ends failed saying why, so the +// asker's key fails, and it is ended on the stream, not offered to another node. +func TestAPoisonedAskIsSaidAndRunNoMore(t *testing.T) { + src, st := NewFakeAsks(), newSteps() + src.Delivered = delivery.AskRuns + 1 + a := testAsk(delivery.AskDelivery) + _ = src.PublishAsk(a) + ran := false + p := &Puller{Source: src, Language: delivery.AskGo, Node: "shanks", Put: st.put, Capacity: func() int { return 1 }, + PullWait: 10 * time.Millisecond, Run: func(ctx context.Context, a delivery.StepAsk) (any, error) { ran = true; return nil, nil }} + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go p.Pull(ctx) + waitFor(t, "the ask termed", func() bool { _, tm, _ := src.Fates(); return len(tm) == 1 }) + key, _ := delivery.AskStepKey("shanks", a.ID) + s, ok := st.get(key) + if ran || !ok || s.State != delivery.StepFailed || !strings.Contains(s.Reason, "taken 3 times") || !strings.Contains(s.Reason, "poison") { + t.Fatalf("ran %v, step %+v", ran, s) + } + if acked, _, naked := src.Fates(); len(acked)+len(naked) != 0 { + t.Errorf("acked %v naked %v, want it termed alone", acked, naked) + } +} diff --git a/go/delivery/stepasks.go b/go/delivery/stepasks.go index a7ae9d2..4e1f001 100644 --- a/go/delivery/stepasks.go +++ b/go/delivery/stepasks.go @@ -41,6 +41,17 @@ const ( // AskLanguages are the build seats' languages. var AskLanguages = []string{AskGo, AskNode, AskImage} +// AskRuns is how many times an ask is run before it is taken for poison (issue 1907): an ask whose run never ends (it +// takes its node down, or its process dies with it) is delivered again by its stream to the next free node (ADR 0525 +// ยง6); the holder that takes it past AskRuns runs it no more, writes its step failed saying so, and ends it, so the +// asker's key fails with why and no further node is taken down. NATS-native: the count is the stream's own delivery +// count of the message, no timer. +const AskRuns = 3 + +// AskMaxDeliver is the build seats' workers' bound of deliveries (the broker's MaxDeliver): past AskRuns, so the holder +// that says the poison gets its delivery, with one more for a holder that dies saying it. +const AskMaxDeliver = AskRuns + 2 + // StepAskVersion is the version of an Ask value; a reader refuses another (ErrUnknownVersion). const StepAskVersion = 1