mesh-controller/cmd/mesh-controller/record_mirrors.go

package main

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

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

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

// The controller's record mirrors on the bus (novox/hq ADR 0289, design 54, design 57 F3): the assignments,
// and each registered module's declared config, which mesh-config resolves a module's config values from.
//
// **Built from the store, on a tick of their own, by the lease holder alone.** A record changes through a
// verb that runs as a command of its own, a reconcile, a merge's registration or a pending assignment made,
// so no one call site sees every change; what every change has in common is that it is committed to the
// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
// mirror is therefore at most mirrorEvery behind its store.

// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
const mirrorEvery = 10 * time.Second

// mirroringRecords keeps the mirrors current for as long as this controller serves: ctx ends when it stops.
func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) {
	api, err := jetstream.New(conn)
	if err != nil {
		fmt.Printf("the record mirrors are not kept: %v\n", err)
		return
	}
	holds := func() bool {
		epoch, err := theLease.epoch(ctx)
		return err == nil && epoch > 0
	}
	epoch := func() uint64 {
		e, _ := theLease.epoch(ctx)
		return e
	}
	sources := map[string]func(context.Context) (map[string][]byte, error){
		recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
		recordmirror.Predefined.Name:     inv.PredefinedMirror,
	}
	writers := map[string]*recordmirror.Writer{}
	failing := map[string]string{}
	tick := time.NewTicker(mirrorEvery)
	defer tick.Stop()
	events := &assignmentEvents{}
	for {
		if holds() {
			// The assignments are keys of the bus, written by the verbs that change them (novox/hq ADR 0408): this tick
			// keeps the provisioner's plan in their values and says what changed. They are no mirror of the store.
			keepingAssignments(ctx, inv, announce, epoch, events, failing)
			for _, m := range recordmirror.Mirrors() {
				why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
				if why != failing[m.Name] {
					if why != "" {
						fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
					} else {
						fmt.Printf("the %s mirror is kept again\n", m.Name)
					}
					failing[m.Name] = why
				}
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
		}
	}
}

// mirrorOnce brings one mirror to what the store holds, and answers why it could not; empty when it could.
func mirrorOnce(ctx context.Context, api jetstream.JetStream, m recordmirror.Mirror, writers map[string]*recordmirror.Writer,
	source func(context.Context) (map[string][]byte, error), announce recordmirror.Announcer, epoch func() uint64,
	holds func() bool) string {
	reading, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	w := writers[m.Name]
	if w == nil {
		kv, err := api.KeyValue(reading, m.Bucket)
		if err != nil {
			return fmt.Sprintf("the bucket %s cannot be read: %v", m.Bucket, err)
		}
		w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
		writers[m.Name] = w
	}
	want, err := source(reading)
	if err != nil {
		return fmt.Sprintf("the store cannot be read: %v", err)
	}
	lost := w.LostEvents
	r, err := w.Sync(reading, want)
	if r.Written+r.Removed > 0 {
		fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
	}
	if w.LostEvents > lost {
		fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
			m.Name, w.LostEvents-lost)
	}
	if err != nil {
		return err.Error()
	}
	return ""
}

// assignmentEvents is the assignments as the bucket held them at the last look, to say what changed since.
type assignmentEvents struct {
	known map[string]inventory.KeptValue
}

// keepingAssignments is the tick's work for the assignments once they are on the bus: the provisioner's plan written into
// the values it belongs to, and one `assignment-changed` event for each key that is new, changed or ended since the last
// look — whoever wrote it, a verb in this process or a command in another. False when they are not on the bus yet.
func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce recordmirror.Announcer, epoch func() uint64,
	events *assignmentEvents, failing map[string]string) {
	reading, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	const name = "assignments"
	if _, _, err := inv.KeepAssignmentPlans(reading); err != nil {
		if failing[name] != err.Error() {
			failing[name] = err.Error()
			fmt.Printf("the provisioner's plans are not kept in the assignments: %v\n", err)
		}
	} else if failing[name] != "" {
		fmt.Printf("the provisioner's plans are kept in the assignments again\n")
		failing[name] = ""
	}
	now, err := inv.KeptAssignments(reading)
	if err != nil {
		return
	}
	if events.known != nil {
		keys := make([]string, 0, len(now)+len(events.known))
		for key := range now {
			keys = append(keys, key)
		}
		for key := range events.known {
			if _, still := now[key]; !still {
				keys = append(keys, key)
			}
		}
		sort.Strings(keys)
		for _, key := range keys {
			was, had := events.known[key]
			is, has := now[key]
			e := recordmirror.Event{Mirror: recordmirror.Assignments.Name, Key: key, Revision: is.Revision, Epoch: epoch(),
				At: time.Now().UTC()}
			switch {
			case has && !had:
				e.Change = recordmirror.Added
			case has && had && was.Revision != is.Revision:
				e.Change, e.Fields = recordmirror.Changed, recordmirror.ChangedFields(was.Value, is.Value)
			case !has && had:
				e.Change, e.Revision = recordmirror.Removed, 0
			default:
				continue
			}
			body, err := json.Marshal(e)
			if err != nil {
				continue
			}
			for try := 0; try < 2; try++ {
				if err = announce.PublishSeatEvent(reading, broker.ControllerSeat, recordmirror.Assignments.Event, body); err == nil {
					break
				}
			}
		}
	}
	events.known = now
}