jochen
9b7f9a7fd2
ConsumersOf's carried-module path (a module the tool runner carries, no account of its own — mesh-mail's real shape) was not passing Queues through, so its work-queue consumer was missing from the composed list and PruneHeard would have removed what issueProcess made for it. Fixed, with a test through Users -> ConsumersOf. Extracted EnsureStream's config building into StreamConfigFor, a pure function, so the dead-letters test can assert the restricted metadata EnsureStream would actually send for a work-queue stream instead of setting the dead letter's header by hand, which passed whether or not modstreams.go's out.Restricted = true was there at all.
82 lines
3.6 KiB
Go
82 lines
3.6 KiB
Go
package broker
|
|
|
|
import "testing"
|
|
|
|
// mailSendsStream is mesh-mail's future `mail-sends` stream (novox/hq ADR 0436 §7, §8), as the bus holds it:
|
|
// work-queue retention, a 24-hour duplicate window, kept one day and at most 64 MiB.
|
|
var mailSendsStream = ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS",
|
|
Events: []string{"mail.send.approved"}, MaxAgeDays: 1, MaxMiB: 64, WorkQueue: true, Duplicates: 24}
|
|
|
|
// 1177/AC1: a work-queue stream is composed with work-queue retention, refusing a new message when full
|
|
// (DiscardNew), a 24-hour duplicate window, a one-day maximum age and a 64 MiB cap.
|
|
func TestAWorkQueueStreamIsComposedToRefuseWhenFull(t *testing.T) {
|
|
s := mailSendsStream.Stream()
|
|
if s.Retention != RetentionWorkQueue {
|
|
t.Fatalf("retention: %v, want %v", s.Retention, RetentionWorkQueue)
|
|
}
|
|
if !s.DiscardNew {
|
|
t.Fatal("a work-queue stream did not refuse a new message when full")
|
|
}
|
|
if s.DuplicatesSeconds != 24*60*60 {
|
|
t.Fatalf("duplicate window: %d seconds, want %d (24 hours)", s.DuplicatesSeconds, 24*60*60)
|
|
}
|
|
if s.MaxAge != 24*60*60 {
|
|
t.Fatalf("max age: %d seconds, want %d (1 day)", s.MaxAge, 24*60*60)
|
|
}
|
|
if s.MaxBytes != 64<<20 {
|
|
t.Fatalf("max bytes: %d, want %d (64 MiB)", s.MaxBytes, 64<<20)
|
|
}
|
|
}
|
|
|
|
// 1177/AC3: a stream with no work-queue clause composes exactly as main does: limits retention, dropping its
|
|
// oldest, never refusing a new message.
|
|
func TestANonWorkQueueStreamStillDropsItsOldest(t *testing.T) {
|
|
s := mailStream.Stream()
|
|
if s.Retention != RetentionLimits || s.DiscardNew || s.DuplicatesSeconds != 0 {
|
|
t.Fatalf("composed as %+v; want limits retention, never discarding new, no duplicate window", s)
|
|
}
|
|
}
|
|
|
|
// 1178/AC1: the process a work-queue stream's bus.queues names gets exactly one durable consumer,
|
|
// <node>_<module>_<process id>, with the stream's declared max-deliver — composed even though mesh-mail's
|
|
// sender otherwise consumes and watches nothing of its own, which alone would give ConsumerFor nothing
|
|
// needed. Explicit acknowledgement is EnsureConsumer's own rule for every consumer it makes (jetstream.go),
|
|
// so it is not re-asserted here.
|
|
func TestTheOneProcessAWorkQueuesBusNamesGetsItsOwnConsumer(t *testing.T) {
|
|
sender := Principal{Kind: KindModule, Node: "novox", Module: "mesh-mail", PasswordHash: "x",
|
|
Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}}}
|
|
got := ConsumersOf([]Principal{sender})
|
|
if len(got) != 1 {
|
|
t.Fatalf("mesh-mail's consumers: %+v", got)
|
|
}
|
|
c := got[0].Consumer
|
|
if c.Name != "novox_mesh-mail_send" || c.Stream != "MAIL_SENDS" || c.MaxDeliver != 5 {
|
|
t.Fatalf("the queue's consumer: %+v", c)
|
|
}
|
|
}
|
|
|
|
// review round 1, finding 2 (1178): mesh-mail's real shape has no account of its own (NoAccount) and is
|
|
// carried by the machine's tool runner — so its queue consumer has to come from the carried-module path in
|
|
// ConsumersOf, not only from a module with its own principal, or PruneHeard would remove what issueProcess
|
|
// made for it (novox/hq ADR 0436 §1, §2, §6).
|
|
func TestAWorkQueueConsumerSurvivesForAModuleTheRuntimeCarries(t *testing.T) {
|
|
users, err := Users(Records{Nodes: []string{"novox"}, Assigned: map[string][]Declared{"novox": {
|
|
{Module: RuntimeModule},
|
|
{Module: "mesh-mail", NoAccount: true,
|
|
Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}}},
|
|
}}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got := ConsumersOf(users)
|
|
var found bool
|
|
for _, c := range got {
|
|
if c.Module == "mesh-mail" && c.Consumer.Name == "novox_mesh-mail_send" && c.Consumer.Stream == "MAIL_SENDS" {
|
|
found = true
|
|
}
|
|
}
|
|
if !found {
|
|
t.Fatalf("mesh-mail's queue consumer is missing from the carried module's composition: %+v", got)
|
|
}
|
|
}
|