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) } }