164 lines
6.6 KiB
Go
164 lines
6.6 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
// mesh-mail send over a real bus (novox/hq ADR 0436 §7, issue 1195 AC2): a JetStream server of the test's
|
|
// own, nats-server v2.11.17, copied from modules/mesh-delivery/cmd/mesh-delivery/bus_test.go:52-58
|
|
// (newTestBus) — the SDK's own testbus is internal, so this module keeps its own copy. Compiled by the
|
|
// merge gate's `go vet -tags integration`; never run on this machine (the context file's hard rule).
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/process"
|
|
"github.com/nats-io/nats-server/v2/server"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// mailSendApprovedSubject is the real wire subject the controller composes for mesh-mail's own stream
|
|
// mail-sends' one event, mail.send.approved: "mesh.stream." + module + ".event." + event (mesh-controller
|
|
// internal/broker/modstreams.go ModuleStreamSubject, moduleStreamPrefix). The sender's own Consume binds
|
|
// by stream and consumer name alone (mesh-sdk go/process queue.go), so this subject matters only to how a
|
|
// test builds the stream and publishes to it — it must match what the controller actually composes, or an
|
|
// integration test proves nothing about the real path (novox/hq review round 1, finding 7).
|
|
const mailSendApprovedSubject = "mesh.stream." + Module + ".event." + MailSendApproved
|
|
|
|
// mailSendsStreamConfig is the work-queue stream mail-sends, composed exactly as the controller would
|
|
// (mesh-controller internal/catalogue/streams.go StreamDeclaration.Stream and
|
|
// internal/broker/modstreams.go ModuleStream.Stream): retention work-queue, discard new, module.json's
|
|
// max-age-days 1 and max-mib 64, and its "duplicates": 24 — in hours (novox/hq ADR 0436 clause 6) — which
|
|
// the controller turns into DuplicatesSeconds = Duplicates * 60 * 60, 24 hours here.
|
|
func mailSendsStreamConfig() jetstream.StreamConfig {
|
|
return jetstream.StreamConfig{
|
|
Name: MailSendsStream,
|
|
Subjects: []string{mailSendApprovedSubject},
|
|
Retention: jetstream.WorkQueuePolicy,
|
|
Discard: jetstream.DiscardNew,
|
|
MaxAge: 24 * time.Hour,
|
|
MaxBytes: 64 << 20,
|
|
Duplicates: 24 * time.Hour,
|
|
}
|
|
}
|
|
|
|
// mailSendsConsumerConfig is the one durable consumer the controller composes for a process's bus.queues
|
|
// (mesh-controller internal/broker/modstreams.go QueueConsumer): AckWaitSeconds 30, MaxDeliver from the
|
|
// stream's own max-deliveries (module.json: 5, MostAttempts here).
|
|
func mailSendsConsumerConfig(durable string) jetstream.ConsumerConfig {
|
|
return jetstream.ConsumerConfig{Durable: durable, AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: MostAttempts}
|
|
}
|
|
|
|
// newTestBus starts a JetStream-enabled nats-server for the test alone, on a random port with its own temp
|
|
// store (modules/mesh-delivery/cmd/mesh-delivery/bus_test.go:52-58).
|
|
func newTestBus(t *testing.T) string {
|
|
t.Helper()
|
|
opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
|
|
NoLog: true, NoSigs: true}
|
|
s, err := server.NewServer(opts)
|
|
if err != nil {
|
|
t.Fatalf("a bus for this test could not be made: %v", err)
|
|
}
|
|
go s.Start()
|
|
if !s.ReadyForConnections(30 * time.Second) {
|
|
s.Shutdown()
|
|
t.Fatal("a bus for this test did not come up within 30s")
|
|
}
|
|
t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
|
|
return s.ClientURL()
|
|
}
|
|
|
|
// 1195/AC2: given the mail-sends work-queue stream and its consumer <machine>_mesh-mail_send, when the
|
|
// process runs with a handler and a message is published, then the handler receives it; when the process
|
|
// gets SIGTERM (here, its context cancelled, the same signal main.go's send() reacts to) it stops without
|
|
// acknowledging a message in hand.
|
|
func TestSendConsumesMailSendsAndStopsCleanlyOnSignal(t *testing.T) {
|
|
url := newTestBus(t)
|
|
ctx := context.Background()
|
|
|
|
nc, err := nats.Connect(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nc.Close()
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := js.CreateStream(ctx, mailSendsStreamConfig()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
account := process.Account{URL: url, Node: "g14", Module: Module, User: "g14." + Module + "." + SendProcessID, Password: "pw"}
|
|
if _, err := js.CreateConsumer(ctx, MailSendsStream, mailSendsConsumerConfig(account.QueueConsumerName())); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
received := make(chan *process.Message, 1)
|
|
runCtx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- runSend(runCtx, account, func(m *process.Message) { received <- m }) }()
|
|
|
|
body := []byte(`{"send":"send-1","account":"mesh","ask":"send-1"}`)
|
|
// Published twice under the same message id (the send id, as the tools' own publish path always uses,
|
|
// novox/hq ADR 0436 clause 8): the stream's own 24-hour duplicate window keeps it one message, proving
|
|
// the dedup this test's stream config actually carries (novox/hq review round 1, finding 7).
|
|
if _, err := js.Publish(ctx, mailSendApprovedSubject, body, jetstream.WithMsgID("send-1")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := js.Publish(ctx, mailSendApprovedSubject, body, jetstream.WithMsgID("send-1")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
select {
|
|
case msg := <-received:
|
|
if msg.Subject != mailSendApprovedSubject {
|
|
t.Fatalf("received on %q, want %q", msg.Subject, mailSendApprovedSubject)
|
|
}
|
|
// Left exactly here, never acknowledged: the handler's own fate-deciding (Ack, NakWithDelay, Term) is
|
|
// issues 1196-1198's, not this skeleton's.
|
|
case <-time.After(10 * time.Second):
|
|
t.Fatal("the handler never received the published message")
|
|
}
|
|
select {
|
|
case <-received:
|
|
t.Fatal("the duplicate publish under the same message id was delivered a second time")
|
|
case <-time.After(500 * time.Millisecond):
|
|
// The duplicate was dropped by the stream's own window, as it should be.
|
|
}
|
|
|
|
cancel() // the signal send() itself reacts to (SIGTERM, or ^C)
|
|
select {
|
|
case err := <-done:
|
|
if err == nil {
|
|
t.Fatal("runSend returned nil, want ctx.Err() once its context was cancelled")
|
|
}
|
|
case <-time.After(10 * time.Second):
|
|
t.Fatal("runSend did not stop once its context was cancelled")
|
|
}
|
|
|
|
// The message was never acknowledged: it is still pending (or ack-pending) on the same durable consumer,
|
|
// read fresh over a new connection.
|
|
nc2, err := nats.Connect(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nc2.Close()
|
|
js2, err := jetstream.New(nc2)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
cons, err := js2.Consumer(ctx, MailSendsStream, account.QueueConsumerName())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
info, err := cons.Info(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if info.NumPending == 0 && info.NumAckPending == 0 {
|
|
t.Fatal("the message was acknowledged despite the process stopping mid-handle")
|
|
}
|
|
}
|