novox/mesh-sdk
13
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

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)

Browse Source
This commit is contained in:
jochen
2026-10-19 04:50:07 +02:00
parent e05e7130ae
commit 1ffbf9ea5b
2 changed files with 79 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/process/queue_integration_test.go
+78 -36
View File
@@ -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
// <node>_<module>_<process id>.
func makeQueue(t *testing.T, url string, account Account) {
// (mesh-controller internal/broker/modstreams.go QueueConsumer), named <node>_<module>_<process id>.
// 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):
}
}