mesh-controller/cmd/mesh-controller/plan_looks.go

package main

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"slices"
	"sort"
	"strings"
	"sync"
	"time"

	"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/link"
)

// A rollout advances on events, not on a timer (novox/hq ADR 0406 §5, issue 974). One look at the open plans runs on
// each thing that can let a plan move: a plan saved (a build's `built`, a tier's send kept, a command's move), a
// machine's report on a send, a silent key's expiry, and the controller taking the lease. The looks are asked for
// with askToLook and run one at a time by a worker, so an ask made while a look runs is not lost and many asks are
// one look. Only two things are still started by time, both named and both one-shot: a failed advance step is
// looked at again (retryLook), and the end of a walk whose end is not known is recorded when it can be
// (armWalkEnd).

// looker runs the look at the plans that events ask for.
type looker struct {
	signal chan struct{} // one place: asks made together are one look
	look   func(ctx context.Context) (failed bool)
	// gap is the least time between two looks, so a plan that kept changing could not make the worker spin.
	gap time.Duration
	// first runs before the first look: what a controller that took the lease sets up.
	first func(ctx context.Context)

	mu      sync.Mutex
	retry   *time.Timer
	backoff time.Duration
	base    time.Duration
	most    time.Duration
}

const (
	retryLookFirst = 5 * time.Second
	retryLookMost  = time.Minute
	lookGap        = 200 * time.Millisecond
)

func newLooker(look func(ctx context.Context) bool) *looker {
	return &looker{signal: make(chan struct{}, 1), look: look, gap: lookGap, base: retryLookFirst, most: retryLookMost}
}

// ask has a look run, soon; never blocks.
func (l *looker) ask() {
	select {
	case l.signal <- struct{}{}:
	default:
	}
}

// run takes the asks one at a time until the context ends.
func (l *looker) run(ctx context.Context) {
	if l.first != nil {
		l.first(ctx)
	}
	for {
		select {
		case <-ctx.Done():
			l.mu.Lock()
			if l.retry != nil {
				l.retry.Stop()
			}
			l.mu.Unlock()
			return
		case <-l.signal:
		}
		l.after(l.look(ctx))
		if l.gap > 0 {
			select {
			case <-ctx.Done():
			case <-time.After(l.gap):
			}
		}
	}
}

// after takes the outcome of a look: a failed one is looked at again later, a good one forgets the failures.
func (l *looker) after(failed bool) {
	l.mu.Lock()
	defer l.mu.Unlock()
	if !failed {
		l.backoff = 0
		if l.retry != nil {
			l.retry.Stop()
			l.retry = nil
		}
		return
	}
	if l.retry != nil {
		return // one is armed already
	}
	switch {
	case l.backoff == 0:
		l.backoff = l.base
	case l.backoff*2 > l.most:
		l.backoff = l.most
	default:
		l.backoff *= 2
	}
	// retries: an advance step that failed (store, bus or send). One at a time, five seconds doubling to a minute, forgotten by
	// the first look that fails nothing; the step is otherwise started by an event, and a failure is none.
	l.retry = time.AfterFunc(l.backoff, func() {
		l.mu.Lock()
		l.retry = nil
		l.mu.Unlock()
		l.ask()
	})
}

// theLooker is the serving controller's; nil in a command, whose looks are its own.
var (
	lookerMu  sync.RWMutex
	theLooker *looker
)

// askToLook has the serving controller look at the open plans soon: a plan was saved, a machine reported on a send,
// a silent key expired. Nothing in a process that is not serving.
func askToLook() {
	lookerMu.RLock()
	l := theLooker
	lookerMu.RUnlock()
	if l != nil {
		l.ask()
	}
}

// startPlanLooks starts the look at the open plans, and asks for the first: the controller took the lease, and a
// plan left by a replaced controller resumes from there (ADR 0406 §6). It replaces the rollout's 30 second ticker.
// It answers a channel that is closed once the look has returned after the context ended, for a test that restores what
// the look reads (issue 1338).
func startPlanLooks(ctx context.Context, open *stores) <-chan struct{} {
	l := newLooker(func(ctx context.Context) bool { return lookAtPlans(ctx, open) })
	l.first = func(ctx context.Context) { rearmWalkEnds(ctx, open) }
	lookerMu.Lock()
	theLooker = l
	lookerMu.Unlock()
	go func() {
		<-ctx.Done()
		lookerMu.Lock()
		if theLooker == l {
			theLooker = nil
		}
		lookerMu.Unlock()
	}()
	done := make(chan struct{})
	go func() {
		defer close(done)
		l.run(ctx)
	}()
	l.ask()
	return done
}

// applySilentAfter is how long a machine may say nothing on a send before the plan names it silent: the mesh's
// ApplySilentAfter, a variable so a test of the wait needs no quarter of an hour.
var applySilentAfter = inventory.ApplySilentAfter

// startRolloutEvents starts what a rollout advances on, for as long as the context lives (novox/hq ADR 0406 §5, issue 974):
// the wait for each machine's report as a key on the bus that expires (where the bus will not take it a timer of its own
// for that wait looks instead), a provisioner client's words, and the look at the plans, asked for first so a plan left
// by a replaced controller resumes. The serving controller's start and the tests of a rollout call this one. It answers
// a function that waits until the look at the plans has returned, which only a test that ends the context needs (issue
// 1338): the serving controller ignores it.
func startRolloutEvents(ctx context.Context, open *stores, js *broker.JetStream) (wait func()) {
	if err := startRolloutSilent(ctx, js); err != nil {
		fmt.Printf("plans: the wait for the machines' reports could not be put on the bus (%v); each is looked at by a timer of its own\n", err)
	}
	if err := startProvAppliedLooks(ctx, js); err != nil {
		fmt.Printf("plans: the provisioner clients' words could not be watched (%v); a walk waits for them until the next look\n", err)
	}
	if err := startPlanMovedLooks(ctx, js, open); err != nil {
		fmt.Printf("plans: the plans moved by other processes could not be followed (%v)\n", err)
	}
	done := startPlanLooks(ctx, open)
	return func() { <-done }
}

// servingPlanSaved is what the serving controller does for each plan it keeps: said as plan-moved, looked at (a build, a
// send kept or a command's move is followed by the look that asks the next tier, in place of the timer that did), and,
// for a walk that ended, its merges' check builds forgotten (novox/hq ADR 0350 §5).
func servingPlanSaved(ctx context.Context, bus link.Bus, open *stores) func(*inventory.Inventory, inventory.Plan) {
	return func(_ *inventory.Inventory, p inventory.Plan) {
		sayPlanMoved(ctx, bus, open.inventory, p)
		planKept(p)
	}
}

// planLookPasses is the most times one look goes over the plans again because the first pass changed which are open.
const planLookPasses = 6

// lookAtPlans moves every open plan as far as the facts allow, as the ticker did each pass: what waits for a gate is
// released, the plans advanced, the wait for each machine's report kept as a key on the bus, and the phases of the
// walks that ended recorded. It says whether a step failed. Waits for the plans rather than giving up: an ask is
// answered once.
func lookAtPlans(ctx context.Context, open *stores) (failed bool) {
	release, err := open.inventory.HoldPlans(ctx, true)
	if err != nil {
		fmt.Printf("plans: cannot hold them: %v\n", err)
		return true
	}
	func() {
		defer release()
		// What waits for a gate, released one machine first, then the rest at once (ADR 0236, ADR 0353).
		if _, err := releaseBacklog(ctx, open, ""); err != nil {
			fmt.Printf("plans: what waits for a gate could not be looked at: %v\n", err)
			failed = true
		}
		// Again while the plans that are open changed: a plan that ended lets a deferred one before it start, and a
		// batch cut makes one.
		for pass := 0; pass < planLookPasses; pass++ {
			before := openPlansNow(ctx, open.inventory)
			if advanceHeld(ctx, open) {
				failed = true
			}
			if openPlansNow(ctx, open.inventory) == before {
				break
			}
		}
		if err := syncSilentKeys(ctx, open); err != nil {
			fmt.Printf("plans: the wait for the machines' reports could not be kept on the bus: %v\n", err)
			failed = true // so the look is retried
		}
	}()
	keepWalkPhases(ctx, open)
	return failed
}

// advancePlans is one pass over the open plans taking the plans without waiting for them, for a caller that wants a
// single step rather than the look (the tests of the steps).
func advancePlans(ctx context.Context, open *stores) {
	release, err := open.inventory.HoldPlans(ctx, false)
	if err != nil {
		if !errors.Is(err, inventory.ErrPlansBusy) {
			fmt.Printf("plans: cannot hold them: %v\n", err)
		}
		return
	}
	defer release()
	if _, err := releaseBacklog(ctx, open, ""); err != nil {
		fmt.Printf("plans: what waits for a gate could not be looked at: %v\n", err)
	}
	advanceHeld(ctx, open)
}

// openPlansNow is which plans are open and in what state, for telling a pass that changed them from one that did not.
func openPlansNow(ctx context.Context, inv *inventory.Inventory) string {
	plans, err := inv.OpenPlans(ctx)
	if err != nil {
		return "unreadable " + time.Now().String()
	}
	ids := make([]string, 0, len(plans))
	for _, p := range plans {
		ids = append(ids, p.ID+":"+string(p.State))
	}
	sort.Strings(ids)
	return strings.Join(ids, ",")
}

// --- the wait for a machine's report, as keys on the bus ---

var (
	silentMu    sync.RWMutex
	silentKeyed silentClock // nil in a process that is not the serving controller
)

// setSilentClock makes clock the rollout's; nil takes it away.
func setSilentClock(c silentClock) {
	silentMu.Lock()
	defer silentMu.Unlock()
	silentKeyed = c
}

func silentClockOf() silentClock {
	silentMu.RLock()
	defer silentMu.RUnlock()
	return silentKeyed
}

// silentWaitsOf is every machine an open plan waits on for a report, with when it was sent: the machines a walk sent
// that have not reported on the send and are not silent yet (walkReports), the machine a gate of a first machine
// judges, and the machines a release's gates wait on. Pure.
func silentWaitsOf(p inventory.Plan, reports []inventory.Reported, now time.Time, prov inventory.ProvLookup) map[silentWait]time.Time {
	waits := map[silentWait]time.Time{}
	put := func(w silentWait, at time.Time) {
		if at.IsZero() {
			return
		}
		if was, ok := waits[w]; !ok || at.Before(was) {
			waits[w] = at
		}
	}
	if !p.Open() {
		return waits
	}
	if r := p.Release; r != nil {
		gates := []*inventory.PlanGate{r.Gate}
		for _, g := range r.Gates {
			gates = append(gates, g)
		}
		for _, g := range gates {
			if g == nil || g.Verdict != "" || g.Silent || g.Pending || g.Since == nil {
				continue
			}
			for _, node := range g.Machines {
				put(silentWait{Plan: p.ID, Tier: 0, Node: node}, *g.Since)
			}
		}
		return waits
	}
	if p.Waiting() {
		return waits
	}
	tierOf := func(module string) int {
		for i, tier := range p.Tiers {
			if slices.Contains(tier, module) {
				return i
			}
		}
		return 0
	}
	waiting, _, _ := walkReports(p, reports, now, prov)
	for module, s := range p.Modules {
		if s == nil {
			continue
		}
		for _, node := range waiting {
			sent, ok := s.Rest[node]
			if !ok {
				continue
			}
			at := sent.At
			if at.IsZero() && s.SentAt != nil {
				at = *s.SentAt
			}
			put(silentWait{Plan: p.ID, Tier: tierOf(module), Node: node}, at)
		}
		if g := s.Gate; g != nil && g.Verdict == "" && !g.Silent && g.Since != nil && s.FirstAt != nil {
			for _, node := range g.Machines {
				put(silentWait{Plan: p.ID, Tier: tierOf(module), Node: node}, *g.Since)
			}
		}
	}
	return waits
}

// silentTTL is how long a wait's key lives: what is left of ApplySilentAfter from the send, and a second over, so the
// look it wakes finds the machine silent and not a fraction early.
func silentTTL(sent, now time.Time) time.Duration {
	left := sent.Add(applySilentAfter).Sub(now) + time.Second
	if left < time.Second {
		return time.Second
	}
	return left
}

// silentDiff is what the keys on the bus need to be as the waits are: the waits to put a key for (none yet, or the
// key is of an earlier send) and the keys to delete (the wait is over). Pure.
func silentDiff(want, held map[silentWait]time.Time) (arm, drop []silentWait) {
	for w, sent := range want {
		if was, ok := held[w]; !ok || !was.Equal(sent) {
			arm = append(arm, w)
		}
	}
	for w := range held {
		if _, ok := want[w]; !ok {
			drop = append(drop, w)
		}
	}
	return arm, drop
}

// syncSilentKeys keeps the keys on the bus as the open plans' waits are: a wait without a key gets one that lives out
// its time, a send made again gets its key again, and a key whose wait is over — the machine reported on the send, the
// gate has its verdict, the plan ended — is deleted. A report deletes the key by way of the look it asks for (the
// look alone knows the report was on that send). Only the lease holder looks, so only it writes.
func syncSilentKeys(ctx context.Context, open *stores) error {
	c := silentClockOf()
	inv := open.inventory
	plans, err := inv.OpenPlans(ctx)
	if err != nil {
		return err
	}
	reports, err := inv.LastReports(ctx)
	if err != nil {
		return err
	}
	now := time.Now().UTC()
	want := map[silentWait]time.Time{}
	for _, p := range plans {
		for w, at := range silentWaitsOf(p, reports, now, provLookup(ctx)) {
			want[w] = at
		}
	}
	return keepSilentKeys(ctx, c, want, now)
}

// keepSilentKeys puts the keys on the bus as the waits are. Where the bus will not take a key — no bucket, the keys
// cannot be read, one cannot be put — the refusal is raised as a condition naming the plan and the machine, never
// stood in for by a timer, and the error is given back so the look is retried and writes the key again (novox/hq ADR
// 0433 §4). The set of refusals is what this look found: a key written at the next look is no longer in it.
func keepSilentKeys(ctx context.Context, c silentClock, want map[silentWait]time.Time, now time.Time) error {
	refused := func(err error) map[silentWait]error {
		out := map[silentWait]error{}
		for w := range want {
			out[w] = err
		}
		return out
	}
	if c == nil {
		noteRefusedSendKeys(refused(errors.New("the bus's clock for the waits is not there")))
		return nil
	}
	c = boundedClock{c}
	held, err := c.Held(ctx)
	if err != nil {
		noteRefusedSendKeys(refused(err))
		return err
	}
	arm, drop := silentDiff(want, held)
	var first error
	notTaken := map[silentWait]error{}
	for _, w := range arm {
		if err := c.Arm(ctx, w, silentTTL(want[w], now), want[w]); err != nil {
			notTaken[w] = err
			if first == nil {
				first = err
			}
		}
	}
	noteRefusedSendKeys(notTaken)
	for _, w := range drop {
		if err := c.Reported(ctx, w); err != nil && first == nil {
			first = err
		}
	}
	return first
}

// silentCallBound is how long one call to the silent clock may take while the look holds the plans (novox/hq issue 1121).
// The vendored key listing reads its watcher without the context, so a consumer lost when the broker reloaded its user
// list left the look waiting for good, with the plans and every node held.
var silentCallBound = 20 * time.Second

// boundedClock is a clock whose every call is answered within silentCallBound, or fails: a call that never returns is
// left behind on its own goroutine.
type boundedClock struct{ silentClock }

func boundedCall[T any](ctx context.Context, what string, call func(context.Context) (T, error)) (T, error) {
	ctx, cancel := context.WithTimeout(ctx, silentCallBound)
	defer cancel()
	type answer struct {
		v   T
		err error
	}
	done := make(chan answer, 1)
	go func() { v, err := call(ctx); done <- answer{v, err} }()
	select {
	case a := <-done:
		return a.v, a.err
	case <-ctx.Done():
		var zero T
		return zero, fmt.Errorf("%s: no answer from the bus within %s: %w", what, silentCallBound, ctx.Err())
	}
}

func (b boundedClock) Arm(ctx context.Context, w silentWait, ttl time.Duration, sent time.Time) error {
	_, err := boundedCall(ctx, "arming a wait", func(ctx context.Context) (struct{}, error) {
		return struct{}{}, b.silentClock.Arm(ctx, w, ttl, sent)
	})
	return err
}

func (b boundedClock) Held(ctx context.Context) (map[silentWait]time.Time, error) {
	return boundedCall(ctx, "listing the waits", b.silentClock.Held)
}

func (b boundedClock) Reported(ctx context.Context, w silentWait) error {
	_, err := boundedCall(ctx, "dropping a wait", func(ctx context.Context) (struct{}, error) {
		return struct{}{}, b.silentClock.Reported(ctx, w)
	})
	return err
}

// startRolloutSilent puts the wait for the machines' reports on the bus for as long as the context lives: the
// controller is the lease holder here, the only one that writes it. The watcher asks for a look when a key expires.
func startRolloutSilent(ctx context.Context, js *broker.JetStream) error {
	api, err := jetstream.New(js.Conn())
	if err != nil {
		return err
	}
	clock, err := newBusSilent(ctx, api, func(silentWait) { askToLook() })
	if err != nil {
		return err
	}
	setSilentClock(clock)
	go func() {
		<-ctx.Done()
		silentMu.Lock()
		if silentKeyed == silentClock(clock) {
			silentKeyed = nil
		}
		silentMu.Unlock()
	}()
	return nil
}

// startProvAppliedLooks has the plans looked at when a machine's provisioner client says a word (`prov.applied.<node>.<module>`):
// a send that handed modules to the provisioner is applied only once the client said so too, and without this the walk
// would wait for its silent key (ADR 0362, ADR 0406 §5, issue 974). New words only: the first look reads the rest.
func startProvAppliedLooks(ctx context.Context, js *broker.JetStream) error {
	api, err := jetstream.New(js.Conn())
	if err != nil {
		return err
	}
	stream, err := api.Stream(ctx, broker.ProvAppliedStream)
	if err != nil {
		return fmt.Errorf("the provisioner clients' words: %w", err)
	}
	consumer, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{DeliverPolicy: jetstream.DeliverNewPolicy})
	if err != nil {
		return fmt.Errorf("watching the provisioner clients' words: %w", err)
	}
	consuming, err := consumer.Consume(func(jetstream.Msg) { askToLook() })
	if err != nil {
		return fmt.Errorf("watching the provisioner clients' words: %w", err)
	}
	go func() {
		<-ctx.Done()
		consuming.Stop()
	}()
	return nil
}

// startPlanMovedLooks has the plans looked at when any process says one moved: the serving controller follows its own
// `plan-moved`, so a plan a person moved with a verb at the control node's terminal — a process this one did not spawn,
// whose saves it does not hear — is looked at all the same (novox/hq ADR 0406 §2, §5, issue 974). New messages only; the
// first look reads the rest. A look its own save asks is the same look: asks are one.
func startPlanMovedLooks(ctx context.Context, js *broker.JetStream, open *stores) error {
	api, err := jetstream.New(js.Conn())
	if err != nil {
		return err
	}
	stream, err := api.Stream(ctx, broker.EventsStream)
	if err != nil {
		return fmt.Errorf("the events stream: %w", err)
	}
	consumer, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{
		FilterSubjects: []string{link.SeatEventSubject(link.MeshControllerSeat, link.KeyPlanMoved)},
		DeliverPolicy:  jetstream.DeliverNewPolicy})
	if err != nil {
		return fmt.Errorf("following the plans moved: %w", err)
	}
	var bus link.Bus
	if core, err := js.Conn().JetStream(); err == nil {
		bus = link.OverNATS{Conn: js.Conn(), JS: core}
	}
	consuming, err := consumer.Consume(func(m jetstream.Msg) {
		askToLook()
		if bus != nil && open != nil {
			// A walk's start or end, said by this controller or a verb: said again with the machines whose
			// declaration it changes (novox/hq ADR 0505 §2, walk_declared.go).
			go sayDeclared(context.WithoutCancel(ctx), bus, open, m.Data())
		}
	})
	if err != nil {
		return fmt.Errorf("following the plans moved: %w", err)
	}
	go func() {
		<-ctx.Done()
		consuming.Stop()
	}()
	return nil
}

// --- the end of a walk whose end is not known ---

var walkEnds = struct {
	mu     sync.Mutex
	timers map[string]*time.Timer
}{timers: map[string]*time.Timer{}}

// armWalkEnd has the walk recorded once its end can be known: a walk that ended with a machine silent is recorded
// without its total ApplySilentAfter and a minute after its last save (recordWalkPhases), which no event says. Armed at
// each save of the done walk, so it counts from the last.
func armWalkEnd(id string, savedAt time.Time) {
	d := time.Until(savedAt.Add(inventory.ApplySilentAfter + time.Minute + time.Second))
	if d <= 0 {
		return
	}
	walkEnds.mu.Lock()
	defer walkEnds.mu.Unlock()
	if t := walkEnds.timers[id]; t != nil {
		t.Stop()
	}
	// retries: recording the end of a walk whose end is not known. One per walk, once, from its last save; the first
	// look after the lease was taken arms it again for the walks still unrecorded.
	walkEnds.timers[id] = time.AfterFunc(d, func() {
		walkEnds.mu.Lock()
		delete(walkEnds.timers, id)
		walkEnds.mu.Unlock()
		askToLook()
	})
}

// rearmWalkEnds is armWalkEnd for the walks that ended lately, for a controller that has just taken the lease.
func rearmWalkEnds(ctx context.Context, open *stores) {
	recent, err := open.inventory.RecentPlans(ctx, 30)
	if err != nil {
		fmt.Printf("plans: the walks ended lately could not be read to record their phases: %v\n", err)
		return
	}
	for _, p := range recent {
		if p.State == inventory.PlanDone && p.Release == nil && time.Since(p.Updated) <= 2*time.Hour {
			armWalkEnd(p.ID, p.Updated)
		}
	}
}

// planKept is what the serving controller does when a plan was saved: a look is asked for, so a send, a built tier or
// a command's move is followed by the plan's next step; and a walk that ended is armed to be recorded.
func planKept(p inventory.Plan) {
	if p.State == inventory.PlanDone && p.Release == nil {
		armWalkEnd(p.ID, time.Now())
	}
	askToLook()
}

// --- `sent`: a tier sent to a machine, said ---

// sendMade is one machine a plan's tier was sent to.
type sendMade struct {
	Tier    int
	Node    string
	Modules []string
	At      time.Time
}

// sendsMade collects the sends one step of a plan made, to be said once the plan is saved.
type sendsMade struct {
	mu   sync.Mutex
	list []sendMade
}

type sendsMadeKey struct{}

func withSendsMade(ctx context.Context, s *sendsMade) context.Context {
	return context.WithValue(ctx, sendsMadeKey{}, s)
}

// noteSends records that modules of the plan's tier were sent to the nodes.
func noteSends(ctx context.Context, tier int, nodes, modules []string, at time.Time) {
	s, _ := ctx.Value(sendsMadeKey{}).(*sendsMade)
	if s == nil {
		return
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	for _, n := range nodes {
		s.list = append(s.list, sendMade{Tier: tier, Node: n, Modules: slices.Clone(modules), At: at})
	}
}

// planSentBody is the body of the seat event `sent` (novox/hq ADR 0406 §5, issue 974): the plan, the tier, the machine
// and the modules it was sent, and when.
type planSentBody struct {
	Event   string    `json:"event"`
	Plan    string    `json:"plan"`
	Tier    int       `json:"tier"`
	Node    string    `json:"node"`
	Modules []string  `json:"modules"`
	At      time.Time `json:"at"`
}

// sentBodies are the events a step's sends are said as.
func sentBodies(plan string, sends *sendsMade) []planSentBody {
	sends.mu.Lock()
	defer sends.mu.Unlock()
	out := make([]planSentBody, 0, len(sends.list))
	for _, s := range sends.list {
		out = append(out, planSentBody{Event: link.KeySent, Plan: plan, Tier: s.Tier, Node: s.Node,
			Modules: s.Modules, At: s.At})
	}
	return out
}

// saySends says each send a step made as `sent`, once the plan that records it is saved. What an outsider hears; the
// controller's own next step does not wait for it.
func saySends(ctx context.Context, p inventory.Plan, sends *sendsMade) {
	bus := checkEvents
	if bus == nil {
		return
	}
	for _, body := range sentBodies(p.ID, sends) {
		raw, err := json.Marshal(body)
		if err != nil {
			continue
		}
		go func() {
			stating, stop := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
			defer stop()
			if err := bus.PublishSeatEvent(stating, link.MeshControllerSeat, link.KeySent, raw); err != nil {
				fmt.Printf("%s: tier %d was sent to %s, and could not be said as sent: %v\n", body.Plan, body.Tier, body.Node, err)
			}
		}()
	}
}