102 lines
3.2 KiB
Go
102 lines
3.2 KiB
Go
//go:build integration
|
|
|
|
package broker
|
|
|
|
import (
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/testbus"
|
|
)
|
|
|
|
// 1178/AC4, on a real bus: the controller ensuring mesh-mail's work-queue stream and its one consumer gives
|
|
// the bus's own guarantees — a duplicate publish within the window is taken once, a message naked past
|
|
// max-deliver is not handed over again, and a full stream refuses a new publish (ADR 0436 §6).
|
|
//
|
|
// go test -tags integration ./internal/broker/ -run TestAWorkQueueStreamOnARealBus
|
|
func TestAWorkQueueStreamOnARealBus(t *testing.T) {
|
|
js, err := Dial(testbus.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer js.Close()
|
|
|
|
// Small enough that a handful of messages fill it, so "the stream is full" is reachable within a test.
|
|
stream := ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"},
|
|
MaxAgeDays: 1, MaxMiB: 1, WorkQueue: true, Duplicates: 1}
|
|
if err := js.EnsureStream(stream.Stream()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(func() { _ = js.Context().DeleteStream(stream.Name) })
|
|
|
|
owner := QueueOwner{Process: "send", Stream: stream.Name, MaxDeliver: 5}
|
|
sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x"}
|
|
consumer := queueConsumer(sender, owner)
|
|
consumer.AckWaitSeconds = 1
|
|
if err := js.EnsureConsumer(consumer); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
subject := ModuleStreamSubject("mesh-mail", "mail.send.approved")
|
|
|
|
// A duplicate publish within the window is kept once (ADR 0436 §6's duplicate window).
|
|
publish := func(id string, size int) (*nats.PubAck, error) {
|
|
return js.Context().Publish(subject, make([]byte, size), nats.MsgId(id))
|
|
}
|
|
if _, err := publish("dup-1", 16); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := publish("dup-1", 16); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if info, err := js.Context().StreamInfo(stream.Name); err != nil || info.State.Msgs != 1 {
|
|
t.Fatalf("a message published twice with the same id: %v msgs, err %v; want 1", info, err)
|
|
}
|
|
|
|
// Naked five times, the consumer is not handed the message a sixth time.
|
|
api, err := jetstream.New(js.Conn())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
reading, err := api.Consumer(t.Context(), stream.Name, consumer.Name)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for i := 0; i < owner.MaxDeliver; i++ {
|
|
batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got := false
|
|
for msg := range batch.Messages() {
|
|
got = true
|
|
_ = msg.Nak()
|
|
}
|
|
if !got {
|
|
t.Fatalf("delivery %d of %d never arrived", i+1, owner.MaxDeliver)
|
|
}
|
|
}
|
|
batch, err := reading.Fetch(1, jetstream.FetchMaxWait(2*time.Second))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for range batch.Messages() {
|
|
t.Fatalf("the message was delivered a %dth time, past its max-deliver of %d", owner.MaxDeliver+1, owner.MaxDeliver)
|
|
}
|
|
|
|
// Full, the stream refuses a new publish rather than dropping what it holds (DiscardNew).
|
|
var fullErr error
|
|
for i := 0; i < 1<<16; i++ {
|
|
if _, err := publish("fill-"+time.Now().String(), 1<<16); err != nil {
|
|
fullErr = err
|
|
break
|
|
}
|
|
}
|
|
if fullErr == nil {
|
|
t.Fatal("a work-queue stream accepted every publish; want it to refuse one once full")
|
|
}
|
|
}
|