mesh-controller/internal/link/deadletters_restricted_test.go

package link

import (
	"fmt"
	"strings"
	"testing"
	"time"

	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"github.com/novox/mesh-controller/internal/broker"
)

// A restricted stream of a module, MAIL, kept a week, on the bus (novox/hq ADR 0363), and the subject an event of
// the account `mesh` is published on.
func aRestrictedStream(t *testing.T, js *broker.JetStream) string {
	t.Helper()
	mail := broker.ModuleStream{Module: "mesh-mail", Name: "MAIL", Events: []string{"mail.*.message.added"},
		MaxAgeDays: 7, MaxMiB: 16, Restricted: true}
	_ = js.Context().DeleteStream("MAIL")
	if err := js.EnsureStream(mail.Stream()); err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { _ = js.Context().DeleteStream("MAIL") })
	return broker.ModuleStreamSubject("mesh-mail", "mail.mesh.message.added")
}

// withClock sets the clock dead letters are kept and delivered again by, until the test ends.
func withClock(t *testing.T, now func() time.Time) {
	t.Helper()
	before := deadLetterNow
	deadLetterNow = now
	t.Cleanup(func() { deadLetterNow = before })
}

// publishMail puts one mail event on MAIL with a header of its own, and answers its sequence and when it was stored.
func publishMail(t *testing.T, js *broker.JetStream, subject string) (uint64, time.Time) {
	t.Helper()
	msg := &nats.Msg{Subject: subject, Data: []byte(`{"from":"someone@example.org"}`), Header: nats.Header{}}
	msg.Header.Set("x-account", "mesh")
	ack, err := js.Context().PublishMsg(msg)
	if err != nil {
		t.Fatal(err)
	}
	raw, err := js.Context().GetMsg("MAIL", ack.Sequence)
	if err != nil {
		t.Fatal(err)
	}
	return ack.Sequence, raw.Time
}

// notice is the server's maximum-deliveries notice for one message of MAIL.
func notice(seq uint64) []byte {
	return []byte(fmt.Sprintf(`{"stream":"MAIL","consumer":"ace_mailer","stream_seq":%d,"deliveries":5}`, seq))
}

// **A dead letter from a restricted stream is listed without its body, and ages out with its stream** (ADR 0376
// §3, §4): given up on by its consumer and kept, the list and the show by id give what it was and when it ages out,
// never its body or its own headers, and the kept copy carries a time to live of what is left of MAIL's week.
func TestARestrictedDeadLetterIsListedWithoutItsBody(t *testing.T) {
	js := aBus(t)
	keeping(t, js)
	subject := aRestrictedStream(t, js)
	heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{subject}}
	for _, c := range broker.ConsumersOf([]broker.Principal{{Kind: broker.KindModule, Node: "ace", Module: "mailer",
		Hears: []broker.Heard{heard}, PasswordHash: "x"}}) {
		if err := js.EnsureConsumer(c.Consumer); err != nil {
			t.Fatal(err)
		}
	}
	_, published := publishMail(t, js, subject)
	api, err := jetstream.New(js.Conn())
	if err != nil {
		t.Fatal(err)
	}
	mailer, err := api.Consumer(t.Context(), "MAIL", "ace_mailer")
	if err != nil {
		t.Fatal(err)
	}
	d := givenUp(t, js, mailer)
	weekLeft := published.Add(7 * 24 * time.Hour)
	if !d.Restricted || d.Body != "" || d.Headers != nil || d.Subject != subject || d.Sequence != 1 || d.Deliveries != 5 ||
		d.Published.IsZero() || d.GaveUp.IsZero() || d.Size == 0 || d.AgesOut.Sub(weekLeft).Abs() > time.Second {
		t.Fatalf("listed as %+v", d)
	}
	shown, err := DeadLetterNamed(js.Context(), d.ID)
	if err != nil {
		t.Fatal(err)
	}
	if shown.Body != "" || shown.Headers != nil || !shown.Restricted || shown.Withheld == "" || shown.AgesOut.IsZero() {
		t.Fatalf("shown by its id as %+v", shown)
	}
	raw, err := js.Context().GetMsg(broker.DeadLettersStream, d.ID)
	if err != nil {
		t.Fatal(err)
	}
	ttl, err := time.ParseDuration(raw.Header.Get(nats.MsgTTLHdr))
	if err != nil || ttl > 7*24*time.Hour || ttl < 7*24*time.Hour-time.Minute {
		t.Fatalf("kept with a time to live of %q (%v), want what is left of the week", raw.Header.Get(nats.MsgTTLHdr), err)
	}
	if string(raw.Data) != `{"from":"someone@example.org"}` {
		t.Fatal("the kept copy does not hold the message, so it could not be delivered again")
	}
}

// **Past its stream's age, a restricted dead letter is gone** (ADR 0376 §3), on a fake clock: kept with two seconds
// of MAIL's week left, DEAD_LETTERS no longer holds it once they pass; with none left, it is kept as a record
// without its body, which cannot be delivered again.
func TestARestrictedDeadLetterAgesOutWithItsStream(t *testing.T) {
	js := aBus(t)
	subject := aRestrictedStream(t, js)
	seq, published := publishMail(t, js, subject)
	withClock(t, func() time.Time { return published.Add(7*24*time.Hour - 2*time.Second) })
	d, err := KeepDeadLetter(js.Context(), notice(seq))
	if err != nil {
		t.Fatal(err)
	}
	if !d.Restricted || d.Lost != "" {
		t.Fatalf("kept as %+v", d)
	}
	eventually(t, "the dead letter aged out", func() bool {
		_, err := js.Context().GetMsg(broker.DeadLettersStream, d.ID)
		return err != nil
	})
	if held, _ := HeldDeadLetters(js.Context()); held["MAIL.ace_mailer"] != 0 {
		t.Fatalf("still held after its age: %v", held)
	}

	seq, published = publishMail(t, js, subject)
	withClock(t, func() time.Time { return published.Add(7*24*time.Hour + time.Minute) })
	past, err := KeepDeadLetter(js.Context(), notice(seq))
	if err != nil {
		t.Fatal(err)
	}
	raw, err := js.Context().GetMsg(broker.DeadLettersStream, past.ID)
	if err != nil {
		t.Fatal(err)
	}
	if len(raw.Data) != 0 || raw.Header.Get("x-account") != "" || !strings.Contains(past.Lost, "maximum age") {
		t.Fatalf("kept past its age as %+v with %q", past, raw.Data)
	}
	if _, _, err := DeliverAgain(js.Context(), past.ID); err == nil || !strings.Contains(err.Error(), "Drop it") {
		t.Fatalf("a record past its age was delivered again: %v", err)
	}
}

// **A restricted dead letter is delivered again only within its age, and the copy keeps that age** (ADR 0376 §5):
// with more than a minute left the copy goes on EVENTS under the consumer's again subject with a time to live of
// what is left; with less, it is refused as aged out.
func TestARestrictedDeadLetterGoesBackOnlyWithinItsAge(t *testing.T) {
	js := aBus(t)
	subject := aRestrictedStream(t, js)
	heard := broker.Heard{Stream: "MAIL", Emitter: "mesh-mail", Restricted: true, Filters: []string{subject}}
	for _, c := range broker.ConsumersOf([]broker.Principal{{Kind: broker.KindModule, Node: "ace", Module: "mailer",
		Hears: []broker.Heard{heard}, PasswordHash: "x"}}) {
		if err := js.EnsureConsumer(c.Consumer); err != nil {
			t.Fatal(err)
		}
	}
	seq, published := publishMail(t, js, subject)
	at := published.Add(7*24*time.Hour - 10*time.Minute)
	withClock(t, func() time.Time { return at })
	d, err := KeepDeadLetter(js.Context(), notice(seq))
	if err != nil {
		t.Fatal(err)
	}
	at = published.Add(7*24*time.Hour - 30*time.Second)
	if _, _, err := DeliverAgain(js.Context(), d.ID); err == nil || !strings.Contains(err.Error(), "aged out") {
		t.Fatalf("delivered again with under a minute left: %v", err)
	}
	at = published.Add(7*24*time.Hour - 5*time.Minute)
	_, to, err := DeliverAgain(js.Context(), d.ID)
	if err != nil {
		t.Fatal(err)
	}
	if to != broker.AgainSubject("ace_mailer", subject) {
		t.Fatalf("delivered again on %s", to)
	}
	raw, err := js.Context().GetLastMsg(broker.EventsStream, to)
	if err != nil {
		t.Fatal(err)
	}
	ttl, err := time.ParseDuration(raw.Header.Get(nats.MsgTTLHdr))
	if err != nil || ttl > 5*time.Minute || ttl < 4*time.Minute {
		t.Fatalf("the copy carries a time to live of %q (%v), want what is left", raw.Header.Get(nats.MsgTTLHdr), err)
	}
	if string(raw.Data) != `{"from":"someone@example.org"}` || raw.Header.Get("x-account") != "mesh" {
		t.Fatalf("the copy is %q %v", raw.Data, raw.Header)
	}
}

// What of a consumer's dead letters ages out, and when, is read without their bodies, and only for a restricted
// stream's (ADR 0376 §6).
func TestTheAgingOfRestrictedDeadLettersIsRead(t *testing.T) {
	js := aBus(t)
	subject := aRestrictedStream(t, js)
	seq, published := publishMail(t, js, subject)
	withClock(t, func() time.Time { return published.Add(time.Hour) })
	d, err := KeepDeadLetter(js.Context(), notice(seq))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := KeepDeadLetter(js.Context(), []byte(`{"stream":"EVENTS","consumer":"media_sonarr","stream_seq":42,`+
		`"deliveries":5}`)); err != nil {
		t.Fatal(err)
	}
	held, err := HeldDeadLetters(js.Context())
	if err != nil {
		t.Fatal(err)
	}
	aging, err := AgingDeadLetters(js.Context(), held)
	if err != nil {
		t.Fatal(err)
	}
	if len(aging) != 1 || len(aging["MAIL.ace_mailer"]) != 1 || aging["MAIL.ace_mailer"][0].ID != d.ID ||
		!aging["MAIL.ace_mailer"][0].AgesOut.Equal(d.AgesOut) {
		t.Fatalf("aging %+v, kept %+v", aging, d)
	}
}

// **A look never waits out NextMsg's bound for a dead letter that is not there** (novox/hq issue 728 review round
// 1): held can count more than DEAD_LETTERS still holds for a consumer, one aged out, delivered again or dropped
// since it was counted. AgingDeadLetters reads the ordered consumer's pending count once and looks for at most
// that many, so it returns in well under the 2-second NextMsg bound rather than waiting it out for messages that
// are gone.
func TestTheAgingOfRestrictedDeadLettersDoesNotWaitForWhatHeldOvercounts(t *testing.T) {
	js := aBus(t)
	subject := aRestrictedStream(t, js)
	seq, published := publishMail(t, js, subject)
	withClock(t, func() time.Time { return published.Add(time.Hour) })
	d, err := KeepDeadLetter(js.Context(), notice(seq))
	if err != nil {
		t.Fatal(err)
	}
	// held says 3 for this consumer; DEAD_LETTERS holds only the one just kept.
	held := map[string]int{"MAIL.ace_mailer": 3}

	started := time.Now()
	aging, err := AgingDeadLetters(js.Context(), held)
	elapsed := time.Since(started)
	if err != nil {
		t.Fatal(err)
	}
	if elapsed >= time.Second {
		t.Fatalf("took %s, well over the 2-second bound it should never wait out", elapsed)
	}
	if len(aging["MAIL.ace_mailer"]) != 1 || aging["MAIL.ace_mailer"][0].ID != d.ID {
		t.Fatalf("aging %+v, kept %+v", aging, d)
	}
}