mesh-controller/internal/link/busfaults.go

package link

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"sort"
	"time"

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

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

// What the bus did to other users' connections, as the bus module says it (novox/hq ADR 0357, issue 270).
//
// **The server says another user's slow consumer only in its record of closed connections**, which the
// bus module reads on its own machine and says as events: `bus.slow-consumer` for a connection the server
// closed as a slow consumer, `bus.connection-closed` for one it closed for breaking the bus's rules. They
// are read back from EVENTS on every watchdog tick for the hour S9 says an advisory, rather than taken by
// the controller's durable consumer: that one takes a message at a time while a merge builds for
// minutes, and read back, a controller that restarts says the last hour again without keeping anything.
//
// **Said at least once, counted once.** The module says the last hour again whenever it starts; a closure
// is its user, its connection id and when it opened, and is counted once however often it is said.

// BusModule is the module whose events say what the bus did to its users' connections.
const BusModule = "nats"

// The bus module's events, by their local names (its manifest's `emits`).
const (
	BusSlowConsumerEvent     = "bus.slow-consumer"
	BusConnectionClosedEvent = "bus.connection-closed"
)

// AdvisoryConnectionClosed is a connection the bus closed for breaking its rules, as conditions name it.
const AdvisoryConnectionClosed = "connection-closed"

// BusClosure is one closed connection as the bus module says it: the user, never a credential.
type BusClosure struct {
	User    string    `json:"user"`
	Name    string    `json:"name,omitempty"`
	CID     uint64    `json:"cid"`
	Reason  string    `json:"reason"`
	Opened  time.Time `json:"opened"`
	Closed  time.Time `json:"closed"`
	Pending int64     `json:"pending_bytes,omitempty"`
}

func (c BusClosure) identity() string {
	return fmt.Sprintf("%s/%d/%s", c.User, c.CID, c.Opened.UTC().Format(time.RFC3339Nano))
}

// BusClosures are the advisories the bus module's closures make, closed at or after since: one per user
// and kind, each closure counted once, the newest said.
func BusClosures(closures []BusClosure, kinds []string, since time.Time) []Advisory {
	type entry struct {
		a      *Advisory
		seen   map[string]bool
		newest BusClosure
	}
	by := map[string]*entry{}
	var order []string
	for i, c := range closures {
		if c.User == "" || c.Closed.Before(since) {
			continue
		}
		kind := kinds[i]
		key := kind + "/" + c.User
		e := by[key]
		if e == nil {
			e = &entry{a: &Advisory{Kind: kind, ID: c.User, User: c.User}, seen: map[string]bool{}}
			by[key] = e
			order = append(order, key)
		}
		if e.seen[c.identity()] {
			continue
		}
		e.seen[c.identity()] = true
		e.a.Count++
		if e.a.First.IsZero() || c.Closed.Before(e.a.First) {
			e.a.First = c.Closed
		}
		if !c.Closed.Before(e.a.Last) {
			e.a.Last, e.newest = c.Closed, c
		}
	}
	sort.Strings(order)
	out := make([]Advisory, 0, len(order))
	for _, key := range order {
		e := by[key]
		c := e.newest
		name := ""
		if c.Name != "" && c.Name != c.User {
			name = fmt.Sprintf(" (connection %q)", c.Name)
		}
		pending := ""
		if c.Pending > 0 {
			pending = fmt.Sprintf(", %d bytes still to be written to it", c.Pending)
		}
		switch e.a.Kind {
		case AdvisorySlowConsumer:
			e.a.Said = fmt.Sprintf("the bus closed %s's connection %d%s as a slow consumer: %s%s", c.User, c.CID,
				name, c.Reason, pending)
		default:
			e.a.Said = fmt.Sprintf("the bus closed %s's connection %d%s: %s", c.User, c.CID, name, c.Reason)
		}
		e.a.Reason = c.Reason
		out = append(out, *e.a)
	}
	return out
}

// busEventKind is the advisory kind of one of the bus module's events, by its subject.
func busEventKind(subject string) (string, bool) {
	switch subject {
	case EventSubject(BusModule, BusSlowConsumerEvent):
		return AdvisorySlowConsumer, true
	case EventSubject(BusModule, BusConnectionClosedEvent):
		return AdvisoryConnectionClosed, true
	}
	return "", false
}

// BusFaults reads back what the bus module said of its users' connections since a moment, from the events
// stream, on a consumer of its own that acknowledges nothing (as PowerStates reads), and says it as
// advisories. Nothing to read is no advisory, and costs two direct reads: the newest of each event.
func (s *Server) BusFaults(ctx context.Context, since time.Time) ([]Advisory, error) {
	if s.js == nil {
		return nil, errors.New("this control plane is not on the bus, so it cannot read what the bus module said")
	}
	js := s.js.Context()
	recent := false
	for _, event := range []string{BusSlowConsumerEvent, BusConnectionClosedEvent} {
		last, err := js.GetLastMsg(broker.EventsStream, EventSubject(BusModule, event), nats.Context(ctx))
		if errors.Is(err, nats.ErrMsgNotFound) {
			continue
		}
		if err != nil {
			return nil, fmt.Errorf("reading what the bus module said of its users' connections: %w", err)
		}
		if !last.Time.Before(since) {
			recent = true
		}
	}
	if !recent {
		return nil, nil
	}
	sub, err := js.SubscribeSync(EventSubject(BusModule, "bus.>"), nats.OrderedConsumer(), nats.StartTime(since))
	if err != nil {
		return nil, fmt.Errorf("reading what the bus module said of its users' connections: %w", err)
	}
	defer func() { _ = sub.Unsubscribe() }()
	var closures []BusClosure
	var kinds []string
	read, cut := 0, false
	for {
		if read >= MaxBusEventsRead {
			cut = true
			break
		}
		read++
		wait, cancel := context.WithTimeout(ctx, readQuiet)
		msg, err := sub.NextMsgWithContext(wait)
		cancel()
		if err != nil {
			if ctx.Err() != nil {
				return nil, ctx.Err()
			}
			break
		}
		if kind, ok := busEventKind(msg.Subject); ok {
			var c BusClosure
			if json.Unmarshal(msg.Data, &c) == nil && c.User != "" && !c.Closed.IsZero() {
				closures, kinds = append(closures, c), append(kinds, kind)
			}
		}
		if meta, err := msg.Metadata(); err != nil || meta.NumPending == 0 {
			break
		}
	}
	out := BusClosures(closures, kinds, since)
	if cut {
		// Said, never silent: a busy hour is read up to the bound, and each advisory says it is a floor.
		for i := range out {
			out[i].Said += fmt.Sprintf(" (at least: the read stopped at %d of the bus module's events this hour)",
				MaxBusEventsRead)
		}
	}
	return out, nil
}

// MaxBusEventsRead bounds one tick's read of the bus module's events: a watchdog tick must not spend itself
// reading a storm of closures back. A read cut there says so in every advisory it makes.
var MaxBusEventsRead = 2000

// MergeAdvisories is two lists of advisories as one: the same kind about the same thing is one advisory,
// counted once (the larger count of the two), said by the newer. The controller's own slow consumer as its client library sees it and
// as the server closed it are the same user's (`controller`).
func MergeAdvisories(a, b []Advisory) []Advisory {
	out := append([]Advisory(nil), a...)
	at := map[string]int{}
	for i, x := range out {
		at[x.Kind+"/"+x.ID+"/"+x.Token] = i
	}
	for _, y := range b {
		key := y.Kind + "/" + y.ID + "/" + y.Token
		i, had := at[key]
		if !had {
			at[key] = len(out)
			out = append(out, y)
			continue
		}
		x := &out[i]
		// The same fault seen twice, not two faults: the controller's own slow consumer is told to its client
		// library and closed by the server, and is counted once — by whichever saw more of it.
		if y.Count > x.Count {
			x.Count = y.Count
		}
		if y.First.Before(x.First) {
			x.First = y.First
		}
		if y.Last.After(x.Last) {
			x.Last, x.Said = y.Last, y.Said
			if y.Reason != "" {
				x.Reason = y.Reason
			}
		}
		if x.User == "" {
			x.User = y.User
		}
	}
	return out
}