mesh-controller/internal/link/deadletters.go

package link

import (
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"strconv"
	"strings"
	"time"

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

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

// What a consumer gave up on, kept until a person acts on it (novox/hq issue 330, design 25 §3).
//
// A durable consumer hands a message over as often as it may, gives up on it and says so on
// `$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES`. The server's notice is captured in its own stream
// (broker.DeadLetterNoticesStream), so one said while no controller listens waits for the next. The
// serving controller takes each notice, fetches the message it names by its sequence, while the source
// stream still holds it, and keeps a copy in DEAD_LETTERS with what the notice said: the consumer, the
// subject, how often it was handed over and when it was given up. Until then a message given up on was
// kept only by its source, which drops an event after a week, and nothing said which.
//
// The copy's headers are the message's own, but for the server's (`Nats-…`), which describe the first
// publish and would refuse or de-duplicate the copy; they are kept under DeadHeader names.

// The headers a kept message carries beside its own.
const (
	DeadHeaderStream     = "Mesh-Dead-Stream"
	DeadHeaderConsumer   = "Mesh-Dead-Consumer"
	DeadHeaderSequence   = "Mesh-Dead-Sequence"
	DeadHeaderSubject    = "Mesh-Dead-Subject"
	DeadHeaderDeliveries = "Mesh-Dead-Deliveries"
	DeadHeaderGaveUp     = "Mesh-Dead-Gave-Up"
	DeadHeaderPublished  = "Mesh-Dead-Published"
	DeadHeaderLost       = "Mesh-Dead-Lost"
	// DeadHeaderMsgID is the message's own de-duplication id, kept, since the copy carries its own.
	DeadHeaderMsgID = "Mesh-Dead-Msg-Id"
	// AgainHeader marks a message delivered again, with the kept copy's id it came from.
	AgainHeader = "Mesh-Delivered-Again"
)

// DeadLetter is one message a consumer gave up on, as DEAD_LETTERS keeps it.
type DeadLetter struct {
	// ID is its sequence in DEAD_LETTERS: what the verb names it by.
	ID       uint64 `json:"id"`
	Stream   string `json:"stream"`
	Consumer string `json:"consumer"`
	// Who is the consumer in the mesh's words: whose, and for what.
	Who        string    `json:"who"`
	Subject    string    `json:"subject"`
	Sequence   uint64    `json:"sequence"`
	Deliveries uint64    `json:"deliveries"`
	GaveUp     time.Time `json:"gave_up"`
	Published  time.Time `json:"published,omitzero"`
	// Lost says why the message itself is not kept: its source no longer held it when the notice was
	// taken. The record of it is kept all the same, so it is said and dropped, never silently missing.
	Lost string `json:"lost,omitempty"`
	Size int    `json:"size"`
	// Body and Headers are the message itself, given only for one dead letter asked by its id.
	Body    string            `json:"body,omitempty"`
	Headers map[string]string `json:"headers,omitempty"`
}

// maxDeliveries is the part of the server's notice a kept message is made from.
type maxDeliveries struct {
	Stream     string    `json:"stream"`
	Consumer   string    `json:"consumer"`
	StreamSeq  uint64    `json:"stream_seq"`
	Deliveries uint64    `json:"deliveries"`
	Timestamp  time.Time `json:"timestamp"`
}

// KeepDeadLetter keeps the message one maximum-deliveries notice is about. An error leaves nothing
// kept, and the notice is to be taken again: the source still holds the message, or a full
// DEAD_LETTERS has room again. A source that no longer holds it is no error: the record is kept with
// why, so it is said.
func KeepDeadLetter(js nats.JetStreamContext, notice []byte) (DeadLetter, error) {
	var a maxDeliveries
	if err := json.Unmarshal(notice, &a); err != nil || a.Stream == "" || a.Consumer == "" || a.StreamSeq == 0 {
		return DeadLetter{}, fmt.Errorf("not a maximum-deliveries notice the mesh can read: %.200s", notice)
	}
	gaveUp := a.Timestamp
	if gaveUp.IsZero() {
		gaveUp = time.Now()
	}
	kept := &nats.Msg{Subject: broker.DeadLetterSubject(a.Stream, a.Consumer), Header: nats.Header{}}
	raw, err := js.GetMsg(a.Stream, a.StreamSeq)
	switch {
	case err == nil:
		for k, v := range raw.Header {
			if strings.HasPrefix(k, "Nats-") {
				continue
			}
			kept.Header[k] = v
		}
		if id := raw.Header.Get(nats.MsgIdHdr); id != "" {
			kept.Header.Set(DeadHeaderMsgID, id)
		}
		kept.Header.Set(DeadHeaderSubject, raw.Subject)
		kept.Header.Set(DeadHeaderPublished, raw.Time.UTC().Format(time.RFC3339Nano))
		kept.Data = raw.Data
	case errors.Is(err, nats.ErrMsgNotFound) || errors.Is(err, nats.ErrStreamNotFound):
		kept.Header.Set(DeadHeaderLost, fmt.Sprintf("%s no longer held message %d when the notice was taken: %v",
			a.Stream, a.StreamSeq, err))
	default:
		return DeadLetter{}, fmt.Errorf("message %d of %s could not be read: %w", a.StreamSeq, a.Stream, err)
	}
	kept.Header.Set(DeadHeaderStream, a.Stream)
	kept.Header.Set(DeadHeaderConsumer, a.Consumer)
	kept.Header.Set(DeadHeaderSequence, strconv.FormatUint(a.StreamSeq, 10))
	kept.Header.Set(DeadHeaderDeliveries, strconv.FormatUint(a.Deliveries, 10))
	kept.Header.Set(DeadHeaderGaveUp, gaveUp.UTC().Format(time.RFC3339Nano))
	// One copy per message given up, however often its notice is taken: a controller that copied and
	// stopped before acknowledging, or two controllers each taking it.
	kept.Header.Set(nats.MsgIdHdr, fmt.Sprintf("%s.%s.%d", a.Stream, a.Consumer, a.StreamSeq))
	ack, err := js.PublishMsg(kept)
	if err != nil {
		return DeadLetter{}, fmt.Errorf("message %d of %s could not be kept in %s: %w", a.StreamSeq, a.Stream,
			broker.DeadLettersStream, err)
	}
	return deadLetterOf(ack.Sequence, kept.Subject, kept.Header, kept.Data, false), nil
}

// deadLetterOf reads a kept message.
func deadLetterOf(id uint64, subject string, h nats.Header, data []byte, whole bool) DeadLetter {
	d := DeadLetter{ID: id, Stream: h.Get(DeadHeaderStream), Consumer: h.Get(DeadHeaderConsumer),
		Subject: h.Get(DeadHeaderSubject), Lost: h.Get(DeadHeaderLost), Size: len(data)}
	if d.Stream == "" || d.Consumer == "" {
		d.Stream, d.Consumer, _ = broker.DeadLetterOf(subject)
	}
	d.Who = ConsumerInWords(d.Stream, d.Consumer)
	d.Sequence, _ = strconv.ParseUint(h.Get(DeadHeaderSequence), 10, 64)
	d.Deliveries, _ = strconv.ParseUint(h.Get(DeadHeaderDeliveries), 10, 64)
	d.GaveUp, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderGaveUp))
	d.Published, _ = time.Parse(time.RFC3339Nano, h.Get(DeadHeaderPublished))
	if whole {
		d.Body = string(data)
		d.Headers = map[string]string{}
		for k := range h {
			if !strings.HasPrefix(k, "Mesh-Dead-") && !strings.HasPrefix(k, "Nats-") {
				d.Headers[k] = h.Get(k)
			}
		}
	}
	return d
}

// HeldDeadLetters is how many messages DEAD_LETTERS holds for each consumer, by `<stream>.<consumer>`.
func HeldDeadLetters(js nats.JetStreamContext) (map[string]int, error) {
	info, err := js.StreamInfo(broker.DeadLettersStream, &nats.StreamInfoRequest{SubjectsFilter: ">"})
	if err != nil {
		return nil, fmt.Errorf("%s cannot be read: %w", broker.DeadLettersStream, err)
	}
	held := map[string]int{}
	for subject, n := range info.State.Subjects {
		if stream, consumer, ok := broker.DeadLetterOf(subject); ok && n > 0 {
			held[stream+"."+consumer] += int(n)
		}
	}
	return held, nil
}

// DeadLetters lists what DEAD_LETTERS holds, newest first, at most most of them; for one consumer when
// consumer names one (its name, or `<stream>.<consumer>`).
func DeadLetters(js nats.JetStreamContext, consumer string, most int) ([]DeadLetter, int, error) {
	info, err := js.StreamInfo(broker.DeadLettersStream)
	if err != nil {
		return nil, 0, fmt.Errorf("%s cannot be read: %w", broker.DeadLettersStream, err)
	}
	var out []DeadLetter
	total := 0
	for seq := info.State.LastSeq; seq >= info.State.FirstSeq && seq > 0; seq-- {
		raw, err := js.GetMsg(broker.DeadLettersStream, seq)
		if errors.Is(err, nats.ErrMsgNotFound) {
			continue // delivered again or dropped
		}
		if err != nil {
			return out, total, fmt.Errorf("dead letter %d cannot be read: %w", seq, err)
		}
		d := deadLetterOf(seq, raw.Subject, raw.Header, raw.Data, false)
		if consumer != "" && consumer != d.Consumer && consumer != d.Stream+"."+d.Consumer {
			continue
		}
		total++
		if most <= 0 || len(out) < most {
			out = append(out, d)
		}
	}
	return out, total, nil
}

// ErrNoDeadLetter is a dead letter's id DEAD_LETTERS does not hold.
var ErrNoDeadLetter = errors.New("no such dead letter")

// DeadLetterNamed is one kept message whole: what it said, and the headers it said it with.
func DeadLetterNamed(js nats.JetStreamContext, id uint64) (DeadLetter, error) {
	raw, err := js.GetMsg(broker.DeadLettersStream, id)
	if errors.Is(err, nats.ErrMsgNotFound) {
		return DeadLetter{}, fmt.Errorf("%w: %s holds no message %d — delivered again or dropped already, "+
			"or never kept; dead-letters lists what it holds", ErrNoDeadLetter, broker.DeadLettersStream, id)
	}
	if err != nil {
		return DeadLetter{}, fmt.Errorf("dead letter %d cannot be read: %w", id, err)
	}
	return deadLetterOf(id, raw.Subject, raw.Header, raw.Data, true), nil
}

// AgainTo is where a kept message is delivered again so that only the consumer that gave it up gets
// it: an event under that consumer's own again subject on EVENTS; an ask back onto its seat's queue,
// whose one worker is that consumer. Any other stream's message is refused, with why.
func AgainTo(d DeadLetter) (string, error) {
	switch {
	case d.Lost != "":
		return "", fmt.Errorf("dead letter %d holds no message to deliver: %s. Drop it", d.ID, d.Lost)
	case d.Subject == "":
		return "", fmt.Errorf("dead letter %d does not say the subject it was published on, so it cannot be "+
			"delivered again. Drop it", d.ID)
	case d.Stream == broker.EventsStream:
		return broker.AgainSubject(d.Consumer, d.Subject), nil
	case strings.HasPrefix(d.Stream, "SEAT_") && strings.HasSuffix(d.Consumer, "_worker"):
		return d.Subject, nil
	}
	return "", fmt.Errorf("dead letter %d is from %s, which nothing delivers again to one consumer alone: "+
		"publishing it again would reach every consumer of its subject. Drop it, and have its sender say it again",
		d.ID, d.Stream)
}

// DeliverAgain hands a kept message to the consumer that gave it up, and nobody else, then removes it
// from DEAD_LETTERS. The message carries its own headers and AgainHeader; its de-duplication id is the
// kept copy's, so asking twice delivers it once.
func DeliverAgain(js nats.JetStreamContext, id uint64) (DeadLetter, string, error) {
	d, err := DeadLetterNamed(js, id)
	if err != nil {
		return d, "", err
	}
	to, err := AgainTo(d)
	if err != nil {
		return d, "", err
	}
	again := &nats.Msg{Subject: to, Header: nats.Header{}, Data: []byte(d.Body)}
	for k, v := range d.Headers {
		again.Header.Set(k, v)
	}
	again.Header.Set(AgainHeader, strconv.FormatUint(id, 10))
	again.Header.Set(nats.MsgIdHdr, "again."+broker.DeadLettersStream+"."+strconv.FormatUint(id, 10))
	if _, err := js.PublishMsg(again); err != nil {
		return d, to, fmt.Errorf("dead letter %d could not be delivered again on %s: %w; it is still kept", id, to, err)
	}
	if err := js.DeleteMsg(broker.DeadLettersStream, id); err != nil && !errors.Is(err, nats.ErrMsgNotFound) {
		return d, to, fmt.Errorf("dead letter %d was delivered again on %s and could not be removed from %s: %w",
			id, to, broker.DeadLettersStream, err)
	}
	return d, to, nil
}

// DropDeadLetter removes a kept message for good.
func DropDeadLetter(js nats.JetStreamContext, id uint64) (DeadLetter, error) {
	d, err := DeadLetterNamed(js, id)
	if err != nil {
		return d, err
	}
	if err := js.DeleteMsg(broker.DeadLettersStream, id); err != nil {
		return d, fmt.Errorf("dead letter %d could not be dropped: %w", id, err)
	}
	return d, nil
}

// noticeRetry is how long a notice whose message could not be kept waits before it is taken again.
var noticeRetry = time.Minute

// keepGivenUp takes the server's maximum-deliveries notices off their stream and keeps the message
// each is about, for as long as the subscription stands. A notice that could not be kept is offered
// again after noticeRetry, and said: in the log, and as the consumer's max-deliveries condition.
func keepGivenUp(js nats.JetStreamContext, logger *log.Logger) (*nats.Subscription, error) {
	return js.Subscribe("", func(m *nats.Msg) {
		d, err := KeepDeadLetter(js, m.Data)
		if err != nil {
			logger.Printf("a message a consumer gave up on could NOT be kept: %v", err)
			var a maxDeliveries
			if json.Unmarshal(m.Data, &a) == nil && a.Stream != "" && a.Consumer != "" {
				Advisories.Heard(Advisory{Kind: AdvisoryMaxDeliveries, ID: a.Stream + "." + a.Consumer,
					Stream: a.Stream, Consumer: a.Consumer, Token: AdvisoryNotKept,
					Said: fmt.Sprintf("%s gave up on message %d, and it could not be kept: %v",
						ConsumerInWords(a.Stream, a.Consumer), a.StreamSeq, err)}, time.Now())
			}
			_ = m.NakWithDelay(noticeRetry)
			return
		}
		if d.Lost != "" {
			logger.Printf("%s gave up on message %d of %s, which its stream no longer held; kept as dead letter %d "+
				"without it", d.Who, d.Sequence, d.Stream, d.ID)
		} else {
			logger.Printf("%s gave up on message %d of %s (%s) after %d deliveries; kept as dead letter %d",
				d.Who, d.Sequence, d.Stream, d.Subject, d.Deliveries, d.ID)
		}
		_ = m.Ack()
	}, nats.Bind(broker.DeadLetterNoticesStream, broker.ControllerName), nats.ManualAck())
}