mesh-controller/cmd/mesh-controller/deadletters_aging.go

package main

import (
	"context"
	"fmt"
	"sort"
	"strings"
	"time"

	"github.com/novox/mesh-controller/internal/conditions"
	"github.com/novox/mesh-controller/internal/link"
)

// A dead letter from a restricted stream ages out with its stream (novox/hq ADR 0376 §3, §6).
//
// DEAD_LETTERS lets such a message go when its stream's maximum age runs out, so a consumer's `max-deliveries`
// condition can clear with nobody having delivered it again or dropped it. That is said: the watchdogs remember,
// from one look to the next, when each restricted dead letter ages out, count those gone once their time passed,
// say each in the controller's log as it goes, and clear the condition with words saying how many aged out
// undelivered and from which stream.

// agingTally is what the watchdogs remember of restricted dead letters between looks.
type agingTally struct {
	// seen is each consumer's restricted dead letters at the last look, by id, with when each ages out.
	seen map[string]map[uint64]time.Time
	// agedOut is how many of each consumer's aged out undelivered while its condition was open.
	agedOut map[string]int
}

// look compares this look's restricted dead letters with the last: newly is how many of each consumer's aged out
// since, and cleared how many aged out in all for each consumer DEAD_LETTERS no longer holds anything of, when any
// did. held is every consumer's count this look; aging its restricted dead letters.
func (t *agingTally) look(now time.Time, held map[string]int, aging map[string][]link.Aging) (newly, cleared map[string]int) {
	if t.seen == nil {
		t.seen, t.agedOut = map[string]map[uint64]time.Time{}, map[string]int{}
	}
	newly, cleared = map[string]int{}, map[string]int{}
	for key, before := range t.seen {
		still := map[uint64]bool{}
		for _, a := range aging[key] {
			still[a.ID] = true
		}
		for id, agesOut := range before {
			if !still[id] && !agesOut.After(now) {
				newly[key]++
			}
		}
		if newly[key] == 0 {
			delete(newly, key)
		}
		t.agedOut[key] += newly[key]
	}
	for key, n := range t.agedOut {
		if held[key] > 0 {
			continue
		}
		if n > 0 {
			cleared[key] = n
		}
		delete(t.agedOut, key)
	}
	t.seen = map[string]map[uint64]time.Time{}
	for key, letters := range aging {
		t.seen[key] = map[uint64]time.Time{}
		for _, a := range letters {
			t.seen[key][a.ID] = a.AgesOut
		}
	}
	return newly, cleared
}

// agedOutResolved is the line a consumer's condition clears with when messages it gave up on aged out undelivered.
func agedOutResolved(key string, n int) string {
	stream, _, _ := strings.Cut(key, ".")
	messages := "a message"
	if n > 1 {
		messages = fmt.Sprintf("%d messages", n)
	}
	return fmt.Sprintf("Resolved: %s of the stream %s aged out undelivered", messages, stream)
}

// sayAgedOut says, in the log, what aged out since the last look, and clears the condition of every consumer whose
// dead letters are all gone with the words saying how many aged out. Before the rows are reconciled, so the
// condition is cleared with these words and not its own.
func (w *watchdogs) sayAgedOut(ctx context.Context, f *signalFacts) []string {
	if f.advisoriesErr != nil || f.deadLettersAging == nil {
		return nil
	}
	newly, cleared := w.aging.look(f.now, f.deadLetters, f.deadLettersAging)
	var problems []string
	for _, key := range countedKeys(newly) {
		stream, consumer, _ := strings.Cut(key, ".")
		fmt.Printf("%d message(s) %s gave up on aged out of %s undelivered: %s keeps them only for its maximum age "+
			"(novox/hq ADR 0376)\n", newly[key], link.ConsumerInWords(stream, consumer), "DEAD_LETTERS", stream)
	}
	for _, key := range countedKeys(cleared) {
		stream, consumer, _ := strings.Cut(key, ".")
		fmt.Printf("%s no longer has a message kept: %d aged out of %s undelivered, so its condition clears "+
			"(novox/hq ADR 0376)\n", link.ConsumerInWords(stream, consumer), cleared[key], stream)
		if _, err := w.keeper.ClearSaying(ctx, conditions.Key(conditions.ScopeBus, key, link.AdvisoryMaxDeliveries),
			fmt.Sprintf("its %d dead letter(s) of %s aged out undelivered", cleared[key], stream),
			agedOutResolved(key, cleared[key])); err != nil {
			problems = append(problems, err.Error())
		}
	}
	return problems
}

// countedKeys is a count's keys in order.
func countedKeys(m map[string]int) []string {
	out := make([]string, 0, len(m))
	for k := range m {
		out = append(out, k)
	}
	sort.Strings(out)
	return out
}