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() } } // askedToLook asks the serving controller's look at the plans, and says whether there was one to ask. func askedToLook() bool { lookerMu.RLock() l := theLooker lookerMu.RUnlock() if l == nil { return false } l.ask() return true } // 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); 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) if p.WalkSaid == inventory.WalkEnded { go forgetChecksOfWalk(context.WithoutCancel(ctx), open.inventory, 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..`): // 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) 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) } consuming, err := consumer.Consume(func(jetstream.Msg) { askToLook() }) 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) } }() } }