From 1ffbf9ea5b51b47001ed9d1ca98e57957bd962ff Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 19 Oct 2026 04:50:07 +0200 Subject: [PATCH] Make the Term test prove no redelivery instead of passing on a 30s AckWait it never waits out (issue 1190, review round 1 finding 2) --- go/process/call_integration_test.go | 2 +- go/process/queue_integration_test.go | 114 ++++++++++++++++++--------- 2 files changed, 79 insertions(+), 37 deletions(-) diff --git a/go/process/call_integration_test.go b/go/process/call_integration_test.go index 20f8292..d08be4d 100644 --- a/go/process/call_integration_test.go +++ b/go/process/call_integration_test.go @@ -148,7 +148,7 @@ func TestReleaseAcceptanceAProcessDoesExactlyItsBusDeclaration(t *testing.T) { }); err != nil { t.Fatalf("the admin could not make the events stream: %v", err) } - makeQueue(t, url, account) + makeQueue(t, url, account, 30*time.Second) publishApproved(t, url, `{"send":"s1"}`) adminNC, _ := adminJS(t, url) diff --git a/go/process/queue_integration_test.go b/go/process/queue_integration_test.go index 603f432..ce3736d 100644 --- a/go/process/queue_integration_test.go +++ b/go/process/queue_integration_test.go @@ -16,9 +16,10 @@ const mailSendsApprovedSubject = "mesh.stream.mesh-mail.event.mail.send.approved // makeQueue sets up MAIL_SENDS as a work-queue stream and binds the one durable consumer a process's // bus.queues names it (QueueConsumerName), both as the admin, mirroring what the controller composes -// (mesh-controller internal/broker/modstreams.go QueueConsumer): AckWaitSeconds 30, named -// __. -func makeQueue(t *testing.T, url string, account Account) { +// (mesh-controller internal/broker/modstreams.go QueueConsumer), named __. +// ackWait is the controller's 30s in most tests, shortened in the Term test so waiting past it (to prove +// no redelivery) does not make that test slow. +func makeQueue(t *testing.T, url string, account Account, ackWait time.Duration) { t.Helper() _, adminJetStream := adminJS(t, url) if _, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{ @@ -28,7 +29,7 @@ func makeQueue(t *testing.T, url string, account Account) { t.Fatalf("the admin could not make the work-queue stream: %v", err) } if _, err := adminJetStream.CreateConsumer(ctxFor(t), "MAIL_SENDS", jetstream.ConsumerConfig{ - Durable: account.QueueConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, + Durable: account.QueueConsumerName(), AckPolicy: jetstream.AckExplicitPolicy, AckWait: ackWait, MaxDeliver: 5, }); err != nil { t.Fatalf("the admin could not make the queue consumer: %v", err) @@ -46,11 +47,40 @@ func publishApproved(t *testing.T, url string, body string) { } } +// recv reads one value off ch, failing the test rather than hanging forever if nothing arrives within +// timeout. +func recv[T any](t *testing.T, ch <-chan T, timeout time.Duration, what string) T { + t.Helper() + select { + case v := <-ch: + return v + case <-time.After(timeout): + t.Fatalf("%s: nothing arrived within %s", what, timeout) + var zero T + return zero + } +} + +// streamMsgs is how many messages MAIL_SENDS still holds. +func streamMsgs(t *testing.T, url string) uint64 { + t.Helper() + _, adminJetStream := adminJS(t, url) + stream, err := adminJetStream.Stream(ctxFor(t), "MAIL_SENDS") + if err != nil { + t.Fatal(err) + } + info, err := stream.Info(ctxFor(t)) + if err != nil { + t.Fatal(err) + } + return info.State.Msgs +} + // 1190/AC1: given a work-queue stream with one durable pull consumer and a process connection, when a // message is consumed and acknowledged, then it is gone from the stream. func TestConsumeAckRemovesTheMessageFromTheStream(t *testing.T) { url, account := grantedSendProcess(t) - makeQueue(t, url, account) + makeQueue(t, url, account, 30*time.Second) publishApproved(t, url, `{"send":"s1"}`) conn, err := Connect(account) @@ -67,33 +97,18 @@ func TestConsumeAckRemovesTheMessageFromTheStream(t *testing.T) { cancel() }) }() - var msg *Message - select { - case msg = <-got: - case <-time.After(8 * time.Second): - t.Fatal("the message was never consumed") - } + msg := recv(t, got, 8*time.Second, "the message") <-ctx.Done() if err := msg.Ack(); err != nil { t.Fatalf("Ack: %v", err) } - _, adminJetStream := adminJS(t, url) deadline := time.Now().Add(5 * time.Second) for { - info, err := adminJetStream.Stream(ctxFor(t), "MAIL_SENDS") - if err != nil { - t.Fatal(err) - } - state, err := info.Info(ctxFor(t)) - if err != nil { - t.Fatal(err) - } - if state.State.Msgs == 0 { + if n := streamMsgs(t, url); n == 0 { return - } - if time.Now().After(deadline) { - t.Fatalf("the stream still holds %d messages after the ack", state.State.Msgs) + } else if time.Now().After(deadline) { + t.Fatalf("the stream still holds %d messages after the ack", n) } time.Sleep(100 * time.Millisecond) } @@ -103,7 +118,7 @@ func TestConsumeAckRemovesTheMessageFromTheStream(t *testing.T) { // then it comes back no sooner than the delay with its delivery count raised by one. func TestConsumeNakWithDelayRedeliversAfterTheDelayWithARaisedCount(t *testing.T) { url, account := grantedSendProcess(t) - makeQueue(t, url, account) + makeQueue(t, url, account, 30*time.Second) publishApproved(t, url, `{"send":"s2"}`) conn, err := Connect(account) @@ -115,16 +130,14 @@ func TestConsumeNakWithDelayRedeliversAfterTheDelayWithARaisedCount(t *testing.T const delay = 2 * time.Second ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) defer cancel() - deliveries := make(chan struct { + type delivery struct { at time.Time delivered uint64 - }, 2) + } + deliveries := make(chan delivery, 2) go func() { _ = conn.Consume(ctx, "MAIL_SENDS", func(m *Message) { - deliveries <- struct { - at time.Time - delivered uint64 - }{time.Now(), m.Delivered} + deliveries <- delivery{time.Now(), m.Delivered} if m.Delivered == 1 { _ = m.NakWithDelay(delay) } else { @@ -134,11 +147,11 @@ func TestConsumeNakWithDelayRedeliversAfterTheDelayWithARaisedCount(t *testing.T }) }() - first := <-deliveries + first := recv(t, deliveries, 8*time.Second, "the first delivery") if first.delivered != 1 { t.Fatalf("first delivery count: %d, want 1", first.delivered) } - second := <-deliveries + second := recv(t, deliveries, delay+8*time.Second, "the redelivery") if second.delivered != 2 { t.Fatalf("second delivery count: %d, want 2", second.delivered) } @@ -148,9 +161,13 @@ func TestConsumeNakWithDelayRedeliversAfterTheDelayWithARaisedCount(t *testing.T } // 1190/AC3: given a consumed message, when the handler terminates it, then it is never delivered again. +// A short AckWait (well under the 4s this then waits past) is what would otherwise make this pass +// whether or not Term did anything at all: without it, the message simply sits unredelivered until the +// controller's ordinary 30s AckWait, and the test would see the same "nothing arrived" result either way. func TestConsumeTermNeverDeliversTheMessageAgain(t *testing.T) { url, account := grantedSendProcess(t) - makeQueue(t, url, account) + const ackWait = time.Second + makeQueue(t, url, account, ackWait) publishApproved(t, url, `{"send":"s3"}`) conn, err := Connect(account) @@ -169,13 +186,38 @@ func TestConsumeTermNeverDeliversTheMessageAgain(t *testing.T) { }) }() - first := <-deliveries + first := recv(t, deliveries, 8*time.Second, "the message") if first.Delivered != 1 { t.Fatalf("first delivery count: %d, want 1", first.Delivered) } + + // Term removes it from the work-queue stream and leaves the consumer with nothing pending — not just + // "has not come back yet", which a message merely sitting past an un-acted AckWait would also show. + deadline := time.Now().Add(5 * time.Second) + for { + if n := streamMsgs(t, url); n == 0 { + break + } else if time.Now().After(deadline) { + t.Fatalf("the stream still holds %d messages after Term", n) + } + time.Sleep(100 * time.Millisecond) + } + _, adminJetStream := adminJS(t, url) + cons, err := adminJetStream.Consumer(ctxFor(t), "MAIL_SENDS", account.QueueConsumerName()) + if err != nil { + t.Fatal(err) + } + info, err := cons.Info(ctxFor(t)) + if err != nil { + t.Fatal(err) + } + if info.NumAckPending != 0 { + t.Fatalf("the consumer still has %d acks pending after Term", info.NumAckPending) + } + select { case again := <-deliveries: t.Fatalf("a terminated message was delivered again: %+v", again) - case <-time.After(4 * time.Second): + case <-time.After(4 * ackWait): } }