mesh-controller/internal/recordmirror/mirror.go

// Package recordmirror writes the controller's record mirrors (novox/hq ADR 0289, design 54): a bucket per kind
// of record, one key per record, written from the store only when the record changed, and one event on the
// controller's seat per write.
//
// **The store stays the record.** A mirror is built from what the store committed, so a change rolled back
// is never written. Every pass reads what the bucket holds and compares it byte for byte with what the store
// says: a pass over unchanged records writes nothing and says nothing (ADR 0289 §1), and a value somebody
// else wrote behind the writer's back (a former lease holder) differs from the store and is put right on
// the next pass. Each write names the revision it read (compare-and-set), so a write in between is noticed,
// never overwritten blind.
//
// **Only the lease holder writes** (ADR 0289 §9, design 54 §2): a pass begins by asking whether this
// controller holds the lease, and asks again before every write; without it, it writes nothing.
//
// **The bucket is the authority, the event a notice** (design 54 §3): the event is sent after the write
// succeeded, so there is no event for a value the bucket does not hold. A refused event is tried once more
// and then counted as lost; a watcher that must not miss a change watches the bucket.
//
// What this package does not do yet is design 54 §4's audit (probe D17 and its healer H6), which compares a
// mirror with its store behind the writer's back; novox/hq issue 419 carries it.
package recordmirror

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

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

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

// A Mirror is one kind of record the controller keeps on the bus: its local name, its bucket and the event
// each write raises on the controller's seat.
type Mirror struct {
	Name   string
	Bucket string
	Event  string
}

// The mirrors this package writes (novox/hq design 57, package F3). The assignments mirror is design 54's,
// keyed `<node>.<module>`; the declared config is each registered module's config declarations as its
// manifest holds them, keyed by the module, which mesh-config resolves a module's config values from
// (ADR 0348 clauses 2 and 4).
var (
	Assignments = Mirror{Name: broker.MirrorAssignments, Bucket: broker.AssignmentsBucket,
		Event: link.KeyAssignmentChanged}
	DeclaredConfig = Mirror{Name: broker.MirrorDeclaredConfig, Bucket: broker.DeclaredConfigBucket,
		Event: link.KeyDeclaredConfigChanged}
)

// Mirrors is every mirror this package writes, in the order they are written.
func Mirrors() []Mirror { return []Mirror{Assignments, DeclaredConfig} }

// The changes an event says.
const (
	Added   = "added"
	Changed = "changed"
	Removed = "removed"
)

// Event is what one write says on the controller's seat (design 54 §3): which mirror and key, what kind of
// change, the names of the top-level fields that changed, the bucket's revision of the write, the lease's
// epoch and when. Never the value: a watcher reads it from the bucket at that revision or later.
type Event struct {
	Mirror   string    `json:"mirror"`
	Key      string    `json:"key"`
	Change   string    `json:"change"`
	Fields   []string  `json:"fields,omitempty"`
	Revision uint64    `json:"revision"`
	Epoch    uint64    `json:"epoch,omitempty"`
	At       time.Time `json:"at"`
}

// Announcer says an event on a seat; link.OverNATS is one, and so is link.Bus.
type Announcer interface {
	PublishSeatEvent(ctx context.Context, seat, event string, body []byte) error
}

// MarkerTTL is how long the marker a purge leaves is kept (design 54 §2): a day, so the markers of removed
// records do not pile up in a bucket that has no age.
const MarkerTTL = 24 * time.Hour

// ErrNotHeld is a pass refused because this controller does not hold the lease: it wrote nothing.
var ErrNotHeld = errors.New("this controller does not hold the lease, so it writes no mirror")

// Writer writes one mirror. Not safe for use by two goroutines at once: one controller's one loop owns it.
type Writer struct {
	Mirror   Mirror
	KV       jetstream.KeyValue
	Announce Announcer
	// Epoch is the lease's epoch the events carry; nil carries none.
	Epoch func() uint64
	// Holds says this controller holds the lease. Asked before every pass and every write; a writer
	// without it writes nothing (design 54 §2). Required: nil holds nothing.
	Holds func() bool
	// Now is the clock the events are dated by; nil is time.Now.
	Now func() time.Time

	known map[string]kept
	// LostEvents counts the events refused twice since the writer was made.
	LostEvents int
}

type kept struct {
	value    []byte
	revision uint64
}

// Result is what one pass did: keys written, keys removed, and keys refused with why.
type Result struct {
	Written int
	Removed int
	Refused int
}

// Sync makes the bucket hold want, one key per record: writes each value that differs from what the key
// holds, purges each key no record is left for, and raises one event per write. A key the bus cannot hold
// is refused alone, and the others are written; the error then names it.
func (w *Writer) Sync(ctx context.Context, want map[string][]byte) (Result, error) {
	var r Result
	if !w.holds() {
		return r, ErrNotHeld
	}
	// What the bucket holds now, every pass: a value written behind this writer's back differs from the
	// store's and is written again below.
	if err := w.fill(ctx); err != nil {
		return r, fmt.Errorf("reading the %s mirror: %w", w.Mirror.Name, err)
	}
	var problems []error
	keys := make([]string, 0, len(want))
	for k := range want {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, key := range keys {
		value := want[key]
		before, had := w.known[key]
		if had && bytes.Equal(before.value, value) {
			continue
		}
		if err := validKey(key); err != nil {
			r.Refused++
			problems = append(problems, err)
			continue
		}
		if !w.holds() {
			return r, errors.Join(append(problems, ErrNotHeld)...)
		}
		revision, wrote, err := w.put(ctx, key, value, before, had)
		if err != nil {
			problems = append(problems, fmt.Errorf("writing %s: %w", key, err))
			continue
		}
		w.known[key] = kept{value: append([]byte(nil), value...), revision: revision}
		if !wrote {
			continue
		}
		r.Written++
		change, fields := Added, []string(nil)
		if had {
			change, fields = Changed, changedFields(before.value, value)
		}
		w.say(ctx, key, change, fields, revision)
	}
	gone := make([]string, 0)
	for key := range w.known {
		if _, ok := want[key]; !ok {
			gone = append(gone, key)
		}
	}
	sort.Strings(gone)
	for _, key := range gone {
		k := w.known[key]
		if !w.holds() {
			return r, errors.Join(append(problems, ErrNotHeld)...)
		}
		err := w.KV.Purge(ctx, key, jetstream.LastRevision(k.revision), jetstream.PurgeTTL(MarkerTTL))
		if refusedByRevision(err) {
			// Read it again: purged already, or moved by another writer.
			if e, gerr := w.KV.Get(ctx, key); errors.Is(gerr, jetstream.ErrKeyNotFound) {
				delete(w.known, key)
				continue
			} else if gerr == nil && w.holds() {
				err = w.KV.Purge(ctx, key, jetstream.LastRevision(e.Revision()), jetstream.PurgeTTL(MarkerTTL))
			}
		}
		if err != nil {
			problems = append(problems, fmt.Errorf("removing %s: %w", key, err))
			continue
		}
		delete(w.known, key)
		r.Removed++
		revision := uint64(0)
		if status, err := w.KV.Status(ctx); err == nil {
			// The purge's own sequence is the stream's last one only when nothing wrote in between, which no
			// other writer may; said as the bucket's revision after it.
			if s, ok := status.(interface{ StreamInfo() *jetstream.StreamInfo }); ok {
				revision = s.StreamInfo().State.LastSeq
			}
		}
		w.say(ctx, key, Removed, nil, revision)
	}
	return r, errors.Join(problems...)
}

// put writes one value by compare-and-set and answers its revision, and whether this call wrote it (false
// when the bucket already held it).
func (w *Writer) put(ctx context.Context, key string, value []byte, before kept, had bool) (uint64, bool, error) {
	var (
		revision uint64
		err      error
	)
	if had {
		revision, err = w.KV.Update(ctx, key, value, before.revision)
	} else {
		revision, err = w.KV.Create(ctx, key, value)
	}
	if !refusedByRevision(err) {
		return revision, err == nil, err
	}
	// Refused: another writer in between, or this writer's own write whose acknowledgement was lost.
	e, gerr := w.KV.Get(ctx, key)
	switch {
	case errors.Is(gerr, jetstream.ErrKeyNotFound):
		if !w.holds() {
			return 0, false, errors.New("the key moved and this controller no longer holds the lease")
		}
		revision, err = w.KV.Create(ctx, key, value)
		return revision, err == nil, err
	case gerr != nil:
		return 0, false, gerr
	case bytes.Equal(e.Value(), value):
		return e.Revision(), false, nil
	case !w.holds():
		return 0, false, errors.New("the key moved and this controller no longer holds the lease")
	}
	revision, err = w.KV.Update(ctx, key, value, e.Revision())
	return revision, err == nil, err
}

// say raises one write's event, tried once more when refused and then counted as lost.
func (w *Writer) say(ctx context.Context, key, change string, fields []string, revision uint64) {
	if w.Announce == nil {
		return
	}
	e := Event{Mirror: w.Mirror.Name, Key: key, Change: change, Fields: fields, Revision: revision, At: w.now().UTC()}
	if w.Epoch != nil {
		e.Epoch = w.Epoch()
	}
	body, err := json.Marshal(e)
	if err != nil {
		w.LostEvents++
		return
	}
	for try := 0; try < 2; try++ {
		if err = w.Announce.PublishSeatEvent(ctx, broker.ControllerSeat, w.Mirror.Event, body); err == nil {
			return
		}
	}
	w.LostEvents++
}

// fill reads what the bucket holds now, so a controller started again writes only what differs.
func (w *Writer) fill(ctx context.Context) error {
	w.known = map[string]kept{}
	watch, err := w.KV.WatchAll(ctx, jetstream.IgnoreDeletes())
	if err != nil {
		return err
	}
	defer func() { _ = watch.Stop() }()
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case e := <-watch.Updates():
			if e == nil {
				return nil
			}
			w.known[e.Key()] = kept{value: append([]byte(nil), e.Value()...), revision: e.Revision()}
		}
	}
}

func (w *Writer) holds() bool { return w.Holds != nil && w.Holds() }

func (w *Writer) now() time.Time {
	if w.Now == nil {
		return time.Now()
	}
	return w.Now()
}

// refusedByRevision is the server refusing a compare-and-set: the key is not at the revision named.
func refusedByRevision(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, jetstream.ErrKeyExists) {
		return true
	}
	var apiErr *jetstream.APIError
	return errors.As(err, &apiErr) && apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence
}

// keyChars is what a bucket's key may be made of.
var keyChars = regexp.MustCompile(`^[-/_=.a-zA-Z0-9]+$`)

// validKey refuses a key the bus cannot hold, naming it.
func validKey(key string) error {
	if key == "" || !keyChars.MatchString(key) || key[0] == '.' || key[len(key)-1] == '.' {
		return fmt.Errorf("%q cannot be a key of a bucket: letters, digits, '-', '_', '=', '/' and '.', not at "+
			"either end", key)
	}
	return nil
}

// changedFields names the top-level fields whose values differ between two encoded objects, sorted; nil
// when either is not an object.
func changedFields(before, after []byte) []string {
	var a, b map[string]json.RawMessage
	if json.Unmarshal(before, &a) != nil || json.Unmarshal(after, &b) != nil || a == nil || b == nil {
		return nil
	}
	var out []string
	for k, v := range b {
		if old, ok := a[k]; !ok || !bytes.Equal(old, v) {
			out = append(out, k)
		}
	}
	for k := range a {
		if _, ok := b[k]; !ok {
			out = append(out, k)
		}
	}
	sort.Strings(out)
	return out
}