package link import ( "io" "log" "strings" "testing" "time" "github.com/nats-io/nats.go/jetstream" "github.com/novox/mesh-controller/internal/broker" ) // A module's consumer as the controller makes it, on a bus whose streams the controller asserted. func aModuleConsumer(t *testing.T, js *broker.JetStream, node, module string, consumes ...string) jetstream.Consumer { t.Helper() c, ok := broker.ConsumerFor(broker.Principal{Kind: broker.KindModule, Node: node, Module: module, Consumes: consumes, PasswordHash: "x"}) if !ok { t.Fatal("a module that consumes got no consumer") } if err := js.EnsureConsumer(c); err != nil { t.Fatal(err) } api, err := jetstream.New(js.Conn()) if err != nil { t.Fatal(err) } reading, err := api.Consumer(t.Context(), broker.EventsStream, c.Name) if err != nil { t.Fatal(err) } return reading } // next is the one message a consumer hands over within a second, or nil. func next(t *testing.T, c jetstream.Consumer) jetstream.Msg { t.Helper() batch, err := c.Fetch(1, jetstream.FetchMaxWait(time.Second)) if err != nil { t.Fatal(err) } for msg := range batch.Messages() { return msg } return nil } // keeping takes the notices as the serving controller does, until the test ends. func keeping(t *testing.T, js *broker.JetStream) { t.Helper() if err := broker.AssertMeshConsumers(js); err != nil { t.Fatal(err) } sub, err := keepGivenUp(js.Context(), log.New(io.Discard, "", 0)) if err != nil { t.Fatal(err) } t.Cleanup(func() { _ = sub.Unsubscribe() }) } // givenUp hands one event to a consumer as often as it may, failing each time, and waits for it kept. func givenUp(t *testing.T, js *broker.JetStream, c jetstream.Consumer) DeadLetter { t.Helper() for { msg := next(t, c) if msg == nil { break } _ = msg.Nak() } var kept []DeadLetter eventually(t, "the event kept", func() bool { kept, _, _ = DeadLetters(js.Context(), c.CachedInfo().Name, 0) return len(kept) == 1 }) return kept[0] } // **Delivered again to the consumer that gave it up, and nobody else**: another module consuming the // same event handled it the first time and must not handle it twice. // A notice whose message its stream no longer holds is kept as a record that says so — said, and // dropped by a person, never silently missing — and cannot be delivered again. // What only the giving-up consumer filters is its own: a subject delivered again reads back as the // event's, and a module's runtime reads the same key from it (the tokens around `.event.`). func TestASubjectDeliveredAgainSaysTheEventsOwn(t *testing.T) { for _, original := range []string{"mesh.mod.gitea.event.pull.merged", "mesh.seat.node-build-agent.event.built"} { again := broker.AgainSubject("ace_sonarr", original) if back, ok := broker.OriginalOfAgain(again); !ok || back != original { t.Errorf("%s reads back as %s", again, back) } key := func(subject string) string { before, event, _ := strings.Cut(subject, ".event.") parts := strings.Split(before, ".") return parts[len(parts)-1] + "." + event } if key(again) != key(original) { t.Errorf("%s reads as key %s, the event as %s", again, key(again), key(original)) } } if _, ok := broker.OriginalOfAgain("mesh.mod.gitea.event.pull.merged"); ok { t.Error("an event's own subject reads as delivered again") } } // Only an event is delivered again, and only to a consumer that is on the bus and filters its again // subject: a dead letter is never let go as delivered while nobody receives it. // A total says how many are held, however few the list carries. // An event the controller itself gave up on, delivered again by a person, is acted on as the event it // was: it arrives under the controller's again subject, which its consumer filters. // The notices cannot be taken (their stream is gone): said as a condition, and the controller serves on. // When DEAD_LETTERS kept each consumer's newest message, for the condition that counts what was given up // on rather than how often it was looked at (novox/hq issue 440). It rises with every message kept, even // one the consumer gave up on before the last: a notice that could not be kept is offered again a minute // later, so a later give-up can be kept first, and the one kept after it must still count.