Split the unit level from the integration level in cmd, builder, link, catalogue, inventory; cheap fixtures for the archive-entry bound, the registry waits and the catalogue copy (issue 895)

2026-10-17 21:32:23 +02:00
parent aac4bae5a4
commit 66dc501fc4
438 changed files with 22014 additions and 19825 deletions
@@ -0,0 +1,279 @@
+//go:build integration
+
+package link
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/novox/mesh-controller/internal/broker"
+)
+
+func TestADeadLetterIsDeliveredAgainToItsConsumerAlone(t *testing.T) {
+	js := aBus(t)
+	keeping(t, js)
+	failing := aModuleConsumer(t, js, "media", "sonarr", "gitea.pull.merged")
+	handling := aModuleConsumer(t, js, "media", "radarr", "gitea.pull.merged")
+
+	const subject = "mesh.mod.gitea.event.pull.merged"
+	msg := &nats.Msg{Subject: subject, Data: []byte(`{"n":1}`), Header: nats.Header{}}
+	msg.Header.Set("x-node", "forge")
+	msg.Header.Set(nats.MsgIdHdr, "event-1")
+	if _, err := js.Context().PublishMsg(msg); err != nil {
+		t.Fatal(err)
+	}
+	if m := next(t, handling); m == nil {
+		t.Fatal("the other module was not handed the event")
+	} else {
+		_ = m.Ack()
+	}
+	d := givenUp(t, js, failing)
+	if d.Consumer != "media_sonarr" || d.Stream != "EVENTS" || d.Subject != subject || d.Deliveries != 5 ||
+		d.GaveUp.IsZero() || d.Published.IsZero() || d.Lost != "" {
+		t.Fatalf("kept as %+v", d)
+	}
+	held, err := HeldDeadLetters(js.Context())
+	if err != nil || held["EVENTS.media_sonarr"] != 1 {
+		t.Fatalf("held %v (%v)", held, err)
+	}
+
+	whole, err := DeadLetterNamed(js.Context(), d.ID)
+	if err != nil || whole.Body != `{"n":1}` || len(whole.Headers["x-node"]) != 1 || whole.Headers["x-node"][0] != "forge" {
+		t.Fatalf("one asked whole is %+v (%v)", whole, err)
+	}
+
+	_, to, err := DeliverAgain(js.Context(), d.ID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if to != "mesh.again.media_sonarr.mod.gitea.event.pull.merged" {
+		t.Fatalf("delivered again on %s", to)
+	}
+	again := next(t, failing)
+	if again == nil {
+		t.Fatal("the consumer that gave it up was not handed it again")
+	}
+	if string(again.Data()) != `{"n":1}` || again.Headers().Get("x-node") != "forge" ||
+		again.Headers().Get(AgainHeader) == "" {
+		t.Fatalf("handed again as %s %v", again.Data(), again.Headers())
+	}
+	if original, ok := broker.OriginalOfAgain(again.Subject()); !ok || original != subject {
+		t.Fatalf("delivered again on %s, which does not say the event's own subject", again.Subject())
+	}
+	_ = again.Ack()
+	if m := next(t, handling); m != nil {
+		t.Fatalf("the module that handled it was handed it twice: %s", m.Subject())
+	}
+	if _, err := DeadLetterNamed(js.Context(), d.ID); !errors.Is(err, ErrNoDeadLetter) {
+		t.Fatalf("still kept after it was delivered again: %v", err)
+	}
+	held, _ = HeldDeadLetters(js.Context())
+	if held["EVENTS.media_sonarr"] != 0 {
+		t.Fatalf("still held: %v", held)
+	}
+	// Asked twice, delivered once: the second finds nothing kept.
+	if _, _, err := DeliverAgain(js.Context(), d.ID); !errors.Is(err, ErrNoDeadLetter) {
+		t.Fatalf("a second delivery answered %v", err)
+	}
+}
+
+func TestADeadLetterDroppedIsGone(t *testing.T) {
+	js := aBus(t)
+	keeping(t, js)
+	failing := aModuleConsumer(t, js, "media", "sonarr", "gitea.pull.merged")
+	if _, err := js.Context().Publish("mesh.mod.gitea.event.pull.merged", []byte(`{}`)); err != nil {
+		t.Fatal(err)
+	}
+	d := givenUp(t, js, failing)
+	if _, err := DropDeadLetter(js.Context(), d.ID); err != nil {
+		t.Fatal(err)
+	}
+	if left, total, err := DeadLetters(js.Context(), "", 0); err != nil || total != 0 || len(left) != 0 {
+		t.Fatalf("after the drop: %v %d %v", left, total, err)
+	}
+	if m := next(t, failing); m != nil {
+		t.Fatalf("a dropped event was handed over: %s", m.Subject())
+	}
+}
+
+func TestANoticeWhoseMessageIsGoneIsKeptAndSaysSo(t *testing.T) {
+	js := aBus(t)
+	d, err := KeepDeadLetter(js.Context(), []byte(`{"stream":"EVENTS","consumer":"media_sonarr","stream_seq":42,`+
+		`"deliveries":5,"timestamp":"2026-10-06T15:51:00Z"}`))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.Contains(d.Lost, "no longer held message 42") || d.Consumer != "media_sonarr" {
+		t.Fatalf("kept as %+v", d)
+	}
+	if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "Drop it") {
+		t.Fatalf("a record without its message was delivered again: %v", err)
+	}
+	// The same notice taken twice is kept once.
+	if _, err := KeepDeadLetter(js.Context(), []byte(`{"stream":"EVENTS","consumer":"media_sonarr","stream_seq":42,`+
+		`"deliveries":5}`)); err != nil {
+		t.Fatal(err)
+	}
+	if _, total, _ := DeadLetters(js.Context(), "media_sonarr", 0); total != 1 {
+		t.Fatalf("one notice taken twice is kept %d times", total)
+	}
+}
+
+func TestADeadLetterIsDeliveredAgainOnlyWhereItIsReceived(t *testing.T) {
+	js := aBus(t)
+	keeping(t, js)
+	failing := aModuleConsumer(t, js, "media", "sonarr", "gitea.pull.merged")
+	if _, err := js.Context().Publish("mesh.mod.gitea.event.pull.merged", []byte(`{}`)); err != nil {
+		t.Fatal(err)
+	}
+	d := givenUp(t, js, failing)
+
+	// A consumer as it was before this fix: its filters do not take what is delivered again.
+	c, _ := broker.ConsumerFor(broker.Principal{Kind: broker.KindModule, Node: "media", Module: "sonarr",
+		Consumes: []string{"gitea.pull.merged"}, PasswordHash: "x"})
+	c.Filters = c.Filters[:len(c.Filters)-1]
+	if err := js.EnsureConsumer(c); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "does not yet take") {
+		t.Fatalf("delivered to a consumer that does not filter its again subject: %v", err)
+	}
+	if _, err := DeadLetterNamed(js.Context(), d.ID); err != nil {
+		t.Fatalf("a refused delivery let the dead letter go: %v", err)
+	}
+
+	// The consumer gone: nothing would receive it.
+	if err := js.Context().DeleteConsumer(broker.EventsStream, c.Name); err != nil {
+		t.Fatal(err)
+	}
+	if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "no longer on the bus") {
+		t.Fatalf("delivered to a consumer that is gone: %v", err)
+	}
+	if _, err := DeadLetterNamed(js.Context(), d.ID); err != nil {
+		t.Fatalf("a refused delivery let the dead letter go: %v", err)
+	}
+
+	// Not an event: refused, whatever stream it is from.
+	for _, other := range []DeadLetter{
+		{ID: 2, Stream: "SEAT_TELEGRAM_SENDER", Consumer: "SEAT_TELEGRAM_SENDER_worker", Subject: "mesh.seat.telegram-sender.accept.send"},
+		{ID: 3, Stream: "KV_x", Consumer: "y", Subject: "$KV.x.k"},
+	} {
+		if _, err := AgainTo(js.Context(), other); err == nil {
+			t.Errorf("%s was given a subject to be delivered again on", other.Stream)
+		}
+	}
+}
+
+func TestTheListSaysTheTotalAndStopsAtItsLimit(t *testing.T) {
+	js := aBus(t)
+	for seq := 1; seq <= 5; seq++ {
+		if _, err := KeepDeadLetter(js.Context(), []byte(fmt.Sprintf(`{"stream":"EVENTS","consumer":"media_sonarr",`+
+			`"stream_seq":%d,"deliveries":5}`, seq))); err != nil {
+			t.Fatal(err)
+		}
+	}
+	list, total, err := DeadLetters(js.Context(), "media_sonarr", 2)
+	if err != nil || total != 5 || len(list) != 2 || list[0].Sequence != 5 {
+		t.Fatalf("%d of %d (%v): %+v", len(list), total, err, list)
+	}
+	if _, total, _ := DeadLetters(js.Context(), "another_one", 2); total != 0 {
+		t.Fatalf("another consumer's total is %d", total)
+	}
+}
+
+func TestTheControllerActsOnAnEventDeliveredAgain(t *testing.T) {
+	js := aBus(t)
+	told := &toldAbout{}
+	s := &Server{inbound: Nats(js), bus: OverNATS{Conn: js.Conn(), JS: js.Context()}, log: quiet()}
+	if err := s.Follows(told); err != nil {
+		t.Fatal(err)
+	}
+	if err := s.Answers(replaysWith{}); err != nil {
+		t.Fatal(err)
+	}
+	ctx, stop := context.WithCancel(context.Background())
+	defer stop()
+	go func() { _ = s.Serve(ctx) }()
+	eventually(t, "the controller's event consumer being made", func() bool {
+		_, err := js.Context().ConsumerInfo("EVENTS", broker.ControllerName)
+		return err == nil
+	})
+	moved, _ := json.Marshal(Upgraded{Module: "gitea", Commit: "abcdef0123"})
+	if _, err := js.Context().Publish(broker.AgainSubject(broker.ControllerName, broker.ControllerFollows[0]), moved); err != nil {
+		t.Fatal(err)
+	}
+	eventually(t, "the upgrade delivered again reaching the controller", func() bool { return told.count() == 1 })
+}
+
+func TestNoticesThatCannotBeTakenAreSaidAndServingGoesOn(t *testing.T) {
+	js := aBus(t)
+	if err := js.Context().DeleteStream(broker.DeadLetterNoticesStream); err != nil {
+		t.Fatal(err)
+	}
+	stop := keepingGivenUp(context.Background(), js, quiet())
+	defer stop()
+	eventually(t, "the failure said", func() bool {
+		for _, a := range Advisories.Since(time.Now().Add(-time.Minute)) {
+			if a.Token == AdvisoryNotKept && a.Stream == broker.DeadLetterNoticesStream {
+				return true
+			}
+		}
+		return false
+	})
+	held := &counted{}
+	_, stopServing := servingOn(t, js, held)
+	defer stopServing()
+	if _, err := js.Context().Publish("mesh.control.anchor.report", []byte(`{"node":"anchor"}`)); err != nil {
+		t.Fatal(err)
+	}
+	eventually(t, "a report heard while the notices cannot be taken", func() bool { return held.count() == 1 })
+}
+
+func TestTheNewestHeldDeadLetterIsWhenItWasKept(t *testing.T) {
+	js := aBus(t)
+	gaveUp := time.Date(2026, 10, 10, 21, 4, 0, 0, time.UTC)
+	keep := func(consumer string, seq int, at time.Time) {
+		t.Helper()
+		if _, err := KeepDeadLetter(js.Context(), []byte(fmt.Sprintf(`{"stream":"EVENTS","consumer":%q,`+
+			`"stream_seq":%d,"deliveries":5,"timestamp":%q}`, consumer, seq, at.Format(time.RFC3339Nano)))); err != nil {
+			t.Fatal(err)
+		}
+	}
+	look := func() map[string]time.Time {
+		t.Helper()
+		held, err := HeldDeadLetters(js.Context())
+		if err != nil {
+			t.Fatal(err)
+		}
+		newest, err := NewestDeadLetters(js.Context(), held)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return newest
+	}
+
+	// Sonarr gave up at 21:05 and that is kept first; radarr's letter is kept after it.
+	keep("media_sonarr", 2, gaveUp.Add(time.Minute))
+	keep("media_radarr", 3, gaveUp)
+	before := look()
+	if len(before) != 2 || before["EVENTS.media_sonarr"].IsZero() ||
+		!before["EVENTS.media_radarr"].After(before["EVENTS.media_sonarr"]) {
+		t.Fatalf("%v", before)
+	}
+	// Sonarr's give-up of 21:04, kept late, is newer than anything the condition has seen.
+	keep("media_sonarr", 1, gaveUp)
+	after := look()
+	if !after["EVENTS.media_sonarr"].After(before["EVENTS.media_sonarr"]) ||
+		!after["EVENTS.media_sonarr"].After(before["EVENTS.media_radarr"]) {
+		t.Fatalf("a letter kept late does not read as newer: before %v, after %v", before, after)
+	}
+	if !after["EVENTS.media_radarr"].Equal(before["EVENTS.media_radarr"]) {
+		t.Fatalf("radarr's newest moved: before %v, after %v", before, after)
+	}
+}