123 lines
4.2 KiB
Go
123 lines
4.2 KiB
Go
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.
|