mesh-catalog/modules/mesh-delivery/cmd/mesh-delivery/holder.go

package main

import (
	"errors"
	"fmt"
	"sort"
	"strings"
	"sync"
	"time"
)

// The holder of the mesh-delivery seat (novox/hq ADR 0239, to-be 47): the one writer of deliveries and
// groups. Every input — a pull request's head, a verdict, a merge, a walk's record, a person's act — only
// changes a delivery's facts; settle then moves it through the table as far as its facts allow, so the
// order the inputs arrive in never decides where a delivery stands. A delivery is kept before anything
// about it is said, and what is owed outside — the event, the commit's note, the pull request's view and
// statuses — is retried until done.

// Holder is the seat's holder.
type Holder struct {
	Store      Store
	Controller Controller
	Forge      Forge
	Emit       func(event string, body any) error
	Now        func() time.Time
	Logf       func(format string, a ...any)

	mu         sync.Mutex
	deliveries map[string]*Delivery
	groups     map[string]*Group
	dirty      map[string]bool
	dirtyGroup map[string]bool
	unmatched  map[string]unmatched
	backlog    []Walk
	started    time.Time
	loaded     bool
	held       bool
	reconciled time.Time
	unreached  string
	// askTried is when a check this owner asks was last tried and could not be asked: tried again a minute on.
	askTried map[string]time.Time
	refusals []string
}

// unmatched is a walk no delivery is known for yet: its merge not heard, or merged before this owner held.
type unmatched struct {
	walk  Walk
	first time.Time
}

// The words this owner signs its acts with.
const (
	byOwner = "mesh-delivery"
	byH2    = "healer H2"
)

// How long a walk is given for its merge to be heard before it is made a delivery of its own.
const matchGrace = 2 * time.Minute

// How long a merged delivery may go without a walk before it is held, saying so.
const unwalked = 10 * time.Minute

// How long a final delivery is kept; its note on the commit keeps it after.
const keptFinal = 30 * 24 * time.Hour

func (h *Holder) init() {
	if h.Now == nil {
		h.Now = func() time.Time { return time.Now().UTC() }
	}
	if h.Logf == nil {
		h.Logf = func(string, ...any) {}
	}
	if h.Emit == nil {
		h.Emit = func(string, any) error { return nil }
	}
	h.deliveries = map[string]*Delivery{}
	h.groups = map[string]*Group{}
	h.dirty = map[string]bool{}
	h.dirtyGroup = map[string]bool{}
	h.unmatched = map[string]unmatched{}
	h.askTried = map[string]time.Time{}
	h.started = h.Now()
}

// Load reads back every delivery and group, before any input is taken: a restarted holder resumes each
// where it stood.
func (h *Holder) Load() error {
	ds, err := h.Store.Deliveries()
	if err != nil {
		return err
	}
	gs, err := h.Store.Groups()
	if err != nil {
		return err
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, d := range ds {
		h.deliveries[d.ID] = d
	}
	for _, g := range gs {
		h.groups[g.ID] = g
	}
	h.loaded = true
	h.Logf("[mesh-delivery] read back %d delivery(ies) and %d group(s)", len(ds), len(gs))
	return nil
}

// Refuse keeps a refusal from outside the lock.
func (h *Holder) Refuse(err error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.refuse(err)
}

// refuse keeps a refusal, said, for the status; under the lock.
func (h *Holder) refuse(err error) {
	h.Logf("[mesh-delivery] %v", err)
	h.refusals = append(h.refusals, h.Now().Format(time.RFC3339)+" "+err.Error())
	if len(h.refusals) > 50 {
		h.refusals = h.refusals[len(h.refusals)-50:]
	}
}

// moved records what a delivery's transitions owe outside, and keeps it.
func (h *Holder) moved(d *Delivery, ts []Transition) {
	now := h.Now()
	for _, t := range ts {
		h.Logf("[mesh-delivery] %s: %s → %s (%s): %s", d.ID, orNothing(t.From), t.To, t.Event, t.Why)
		d.Owed = append(d.Owed, Effect{Kind: EffectEmit, Line: string(mustJSON(transitionEvent(d, t))), Since: now})
		line := noteLine(d, t)
		d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.Commit, Line: line, Since: now})
		if d.MergedAs != "" && !sameCommit(d.MergedAs, d.Commit) && afterTheTrunk(t.To) {
			d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.MergedAs, Line: line, Since: now})
		}
		if t.To.Final() && d.MergedAs != "" && len(d.Steps) > 0 {
			d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.MergedAs, Line: executedLine(d), Since: now})
		}
	}
	if len(ts) > 0 && d.Number > 0 {
		d.owe(Effect{Kind: EffectView, Since: now})
		state, desc := statusOf(d)
		d.owe(Effect{Kind: EffectStatus, Commit: d.Commit, Context: "mesh/delivery", State: state, Line: desc, Since: now})
	}
	h.keep(d)
}

// owe adds an effect that stands for the latest of its kind: a view or a commit's status is refreshed, not
// queued twice.
func (d *Delivery) owe(e Effect) {
	for i, o := range d.Owed {
		if o.Kind == e.Kind && o.Commit == e.Commit && o.Context == e.Context {
			d.Owed[i] = e
			return
		}
	}
	d.Owed = append(d.Owed, e)
}

func afterTheTrunk(s State) bool {
	return s == Published || s == Delivering || s == Held || s == Delivered || s == Failed || s == Stopped || s == Superseded
}

func orNothing(s State) string {
	if s == None {
		return "(new)"
	}
	return string(s)
}

// keep writes a delivery; one that could not be written is written again by Flush before anything about it
// is said.
func (h *Holder) keep(d *Delivery) {
	h.deliveries[d.ID] = d
	if err := h.Store.PutDelivery(d); err != nil {
		h.dirty[d.ID] = true
		h.Logf("[mesh-delivery] %s could not be kept yet (%v); kept again before it is said", d.ID, err)
		return
	}
	delete(h.dirty, d.ID)
}

func (h *Holder) keepGroup(g *Group) {
	h.groups[g.ID] = g
	if err := h.Store.PutGroup(g); err != nil {
		h.dirtyGroup[g.ID] = true
		h.Logf("[mesh-delivery] group %s could not be kept yet (%v)", g.ID, err)
		return
	}
	delete(h.dirtyGroup, g.ID)
}

// settleAndKeep moves a delivery as far as its facts allow, and keeps it.
func (h *Holder) settleAndKeep(d *Delivery) {
	ts := settle(d, Facts{Now: h.Now()})
	h.moved(d, ts)
	h.closeGroupOf(d)
}

// --- inputs ---------------------------------------------------------------------------------------------

// PullEvent is the forge's `pull.updated`, `pull.merged` or `pull.closed`.
type PullEvent struct {
	Owner          string            `json:"owner"`
	Repo           string            `json:"repo"`
	Number         int               `json:"number"`
	Title          string            `json:"title,omitempty"`
	Body           string            `json:"body,omitempty"`
	Base           string            `json:"base"`
	Head           string            `json:"head"`
	HeadSHA        string            `json:"head_sha"`
	MergeCommit    string            `json:"merge_commit_sha,omitempty"`
	MergedAt       string            `json:"merged_at,omitempty"`
	CloneURL       string            `json:"clone_url,omitempty"`
	HTMLURL        string            `json:"html_url,omitempty"`
	Paths          []string          `json:"paths,omitempty"`
	PathsTruncated bool              `json:"paths_truncated,omitempty"`
	Removed        []string          `json:"removed,omitempty"`
	ModuleDirs     []string          `json:"module_dirs,omitempty"`
	ModuleDirsSaid bool              `json:"module_dirs_said,omitempty"`
	HeadChecks     map[string]string `json:"head_checks,omitempty"`
}

func (p PullEvent) repository() string { return strings.ToLower(p.Owner + "/" + p.Repo) }

// trunkLike are branch names that never make a group: a pull request from them is nobody's feature.
var trunkLike = map[string]bool{"main": true, "master": true, "trunk": true, "": true}

// PullUpdated takes a pull request's new head: a delivery proposed, an older head of the same pull request
// superseded, and the delivery put in the group its branch name makes.
func (h *Holder) PullUpdated(p PullEvent) {
	if p.HeadSHA == "" || p.Number == 0 {
		h.Refuse(errors.New("a pull request announced with no head or no number; nothing proposed"))
		return
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	id := IDOf(p.repository(), p.HeadSHA)
	if d := h.deliveries[id]; d != nil {
		return // announced again: nothing new
	}
	now := h.Now()
	for _, old := range h.deliveries {
		if old.Repository == p.repository() && old.Number == p.Number && !sameCommit(old.Commit, p.HeadSHA) &&
			!old.State.Final() && old.MergedAs == "" {
			old.NewerHead = p.HeadSHA
			h.settleAndKeep(old)
			if old.Group != "" {
				h.leaveGroup(old)
			}
		}
	}
	d := &Delivery{ID: id, Repository: p.repository(), Commit: p.HeadSHA, Number: p.Number, Title: p.Title,
		Base: p.Base, Branch: p.Head, HTMLURL: p.HTMLURL, CloneURL: p.CloneURL, Paths: p.Paths,
		PathsTruncated: p.PathsTruncated, Removed: p.Removed, ModuleDirs: p.ModuleDirs, ModuleDirsSaid: p.ModuleDirsSaid,
		After: AfterIn(p.Body), Created: now}
	t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: fmt.Sprintf("%s#%d's head announced", d.Repository, d.Number)})
	if err != nil {
		h.refuse(err)
		return
	}
	ts := []Transition{t}
	// An announcement heard again from the bus's history may carry its head's verdict already: the forge holds
	// the gate's status on that very commit. A decided one is taken; a pending or errored one is asked again.
	if v := decidedOnHead(p.HeadChecks, now); v != nil {
		d.Check = v
		ts = append(ts, settle(d, Facts{Now: now})...)
	}
	h.moved(d, ts)
	h.joinGroup(d)
}

// decidedOnHead is the verdict the forge's statuses on a head hold, when the gate decided one: passed, warned
// or failed. Nil for none, pending, or a check that could not run — those are asked again.
func decidedOnHead(checks map[string]string, now time.Time) *Verdict {
	switch checks["mesh/merge-gate"] {
	case "success", "warning", "failure":
	default:
		return nil
	}
	return &Verdict{At: now, Gate: verdictOfState(checks["mesh/merge-gate"]),
		Summary: "the head's mesh/merge-gate, as the forge holds it", Repo: verdictOfState(checks["mesh/repo-check"])}
}

// joinGroup puts a new head in the group its branch name makes with the other repositories' open pull
// requests (ADR 0239 decision 4): one mechanism, the name the feature already has in every repository.
func (h *Holder) joinGroup(d *Delivery) {
	if trunkLike[strings.ToLower(d.Branch)] || strings.EqualFold(d.Branch, d.Base) {
		return
	}
	g := h.groups[d.Branch]
	if g != nil && g.Closed {
		h.Logf("[mesh-delivery] %s shares the branch %s with a group that is already delivering: a lone delivery",
			d.ID, d.Branch)
		return
	}
	var others []*Delivery
	for _, o := range h.deliveries {
		if o.ID != d.ID && o.Branch == d.Branch && o.Repository != d.Repository && o.Number > 0 &&
			!o.State.Final() && o.MergedAs == "" {
			others = append(others, o)
		}
	}
	if len(others) == 0 && g == nil {
		return
	}
	now := h.Now()
	if g == nil {
		g = &Group{ID: d.Branch, Created: now}
	}
	for _, o := range append(others, d) {
		if !containsString(g.Members, o.ID) {
			g.Members = append(g.Members, o.ID)
		}
		if o.Group != g.ID {
			o.Group = g.ID
			h.keep(o)
		}
	}
	sort.Strings(g.Members)
	g.Updated = now
	h.keepGroup(g)
	h.Logf("[mesh-delivery] group %s: %s", g.ID, strings.Join(g.Members, ", "))
}

// leaveGroup takes a superseded or stopped head out of its group's members.
func (h *Holder) leaveGroup(d *Delivery) {
	g := h.groups[d.Group]
	if g == nil {
		return
	}
	var kept []string
	for _, m := range g.Members {
		if m != d.ID {
			kept = append(kept, m)
		}
	}
	g.Members = kept
	g.Updated = h.Now()
	h.keepGroup(g)
}

// PullMerged takes a merge: the commit it landed as, and the delivery it was — made from the forge's word
// when this owner never heard its head (merged before it held the seat).
func (h *Holder) PullMerged(p PullEvent) {
	if p.MergeCommit == "" {
		h.Refuse(errors.New("a merge announced with no merge commit; nothing published"))
		return
	}
	h.mu.Lock()
	defer h.mu.Unlock()
	var d *Delivery
	if p.HeadSHA != "" {
		d = h.deliveries[IDOf(p.repository(), p.HeadSHA)]
	}
	if d == nil {
		var newest time.Time
		for _, c := range h.deliveries {
			if c.Repository == p.repository() && c.Number == p.Number && c.NewerHead == "" && c.Created.After(newest) {
				d, newest = c, c.Created
			}
		}
	}
	now := h.Now()
	if d == nil {
		head := p.HeadSHA
		if head == "" {
			head = p.MergeCommit
		}
		d = &Delivery{ID: IDOf(p.repository(), head), Repository: p.repository(), Commit: head, Number: p.Number,
			Title: p.Title, Base: p.Base, Branch: p.Head, HTMLURL: p.HTMLURL, CloneURL: p.CloneURL, Paths: p.Paths,
			PathsTruncated: p.PathsTruncated, Removed: p.Removed, ModuleDirs: p.ModuleDirs, ModuleDirsSaid: p.ModuleDirsSaid,
			After: AfterIn(p.Body), Created: now}
		// Its check, as the forge holds it on the head: the verdict the pull request merged with.
		if s := p.HeadChecks["mesh/merge-gate"]; s != "" {
			d.Check = &Verdict{At: now, Gate: verdictOfState(s), Summary: "the head's mesh/merge-gate, as the forge holds it",
				Repo: verdictOfState(p.HeadChecks["mesh/repo-check"])}
		}
		t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: fmt.Sprintf("%s#%d merged; its head was not "+
			"heard by this owner, so it is made from the forge's word", d.Repository, d.Number)})
		if err != nil {
			h.refuse(err)
			return
		}
		h.moved(d, []Transition{t})
	}
	d.MergedAs = p.MergeCommit
	if t, err := time.Parse(time.RFC3339, p.MergedAt); err == nil {
		d.MergedAt = t.UTC()
	} else {
		d.MergedAt = now
	}
	// A walk already heard of for this commit.
	for id, u := range h.unmatched {
		if strings.EqualFold(u.walk.Repository, d.Repository) && sameCommit(u.walk.Commit, d.MergedAs) {
			h.linkWalk(d, u.walk)
			delete(h.unmatched, id)
		}
	}
	ts := settle(d, Facts{Now: now})
	if d.State == Held && d.HeldWhy == "" {
		d.HeldWhy = "merged without a passing check: only a person decides that it goes on"
	}
	h.moved(d, ts)
	h.closeGroupOf(d)
}

// closeGroupOf closes a delivery's group once it reached the trunk: a delivering group takes no new member.
func (h *Holder) closeGroupOf(d *Delivery) {
	if d.Group == "" || d.MergedAs == "" {
		return
	}
	if g := h.groups[d.Group]; g != nil && !g.Closed {
		g.Closed = true
		g.Updated = h.Now()
		h.keepGroup(g)
	}
}

// verdictOfState is a commit status's state as a verdict.
func verdictOfState(s string) string {
	switch s {
	case "success":
		return "pass"
	case "warning":
		return "warning"
	case "failure":
		return "fail"
	case "":
		return ""
	}
	return "error"
}

// PullClosed takes a pull request closed unmerged: its open deliveries are stopped.
func (h *Holder) PullClosed(p PullEvent) {
	h.mu.Lock()
	defer h.mu.Unlock()
	for _, d := range h.deliveries {
		if d.Repository == p.repository() && d.Number == p.Number && !d.State.Final() && d.MergedAs == "" {
			d.ClosedUnmerged = true
			h.settleAndKeep(d)
			if d.Group != "" {
				h.leaveGroup(d)
			}
		}
	}
}

// CheckedEvent is the controller's `checked`.
type CheckedEvent struct {
	Owner   string `json:"owner"`
	Repo    string `json:"repo"`
	Number  int    `json:"number,omitempty"`
	Commit  string `json:"commit"`
	Verdict string `json:"verdict"`
	Summary string `json:"summary"`
	ID      string `json:"id"`
	Gate    *struct {
		Verdict string `json:"verdict"`
		Summary string `json:"summary"`
	} `json:"gate,omitempty"`
	RepoCheck *struct {
		Verdict string `json:"verdict"`
		Summary string `json:"summary"`
	} `json:"repo-check,omitempty"`
	Plan    *DeliveryPlan `json:"plan,omitempty"`
	Group   string        `json:"group,omitempty"`
	Members []struct {
		Owner  string `json:"owner"`
		Repo   string `json:"repo"`
		Commit string `json:"commit"`
	} `json:"members,omitempty"`
}

// Checked takes a verdict: a head's own, or a group's composed check.
func (h *Holder) Checked(c CheckedEvent) {
	h.mu.Lock()
	defer h.mu.Unlock()
	now := h.Now()
	if c.Group != "" {
		g := h.groups[c.Group]
		if g == nil || g.Check == nil || g.Check.Asked != c.ID {
			h.Logf("[mesh-delivery] a composed check of %s (%s) that is not the one asked: history", c.Group, c.ID)
			return
		}
		g.Check.Verdict, g.Check.Summary = c.Verdict, c.Summary
		g.Updated = now
		h.keepGroup(g)
		h.Logf("[mesh-delivery] group %s composed: %s — %s", g.ID, c.Verdict, c.Summary)
		return
	}
	repository := strings.ToLower(c.Owner + "/" + c.Repo)
	id := IDOf(repository, c.Commit)
	d := h.deliveries[id]
	if d == nil {
		if c.Number == 0 {
			return // a check of no pull request: nothing to deliver
		}
		d = &Delivery{ID: id, Repository: repository, Commit: c.Commit, Number: c.Number, Created: now}
		t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: "its verdict arrived before its head was heard"})
		if err != nil {
			h.refuse(err)
			return
		}
		h.moved(d, []Transition{t})
	}
	if d.State != Proposed {
		h.Logf("[mesh-delivery] %s is %s: a verdict for it now is history (%s)", d.ID, d.State, c.Verdict)
		return
	}
	v := &Verdict{ID: c.ID, At: now, Gate: c.Verdict, Summary: c.Summary}
	if c.Gate != nil {
		v.Gate, v.Summary = c.Gate.Verdict, c.Gate.Summary
	}
	if c.RepoCheck != nil {
		v.Repo, v.RepoSaid = c.RepoCheck.Verdict, c.RepoCheck.Summary
	}
	if v.Gate == "" {
		v.Gate = "error"
	}
	d.Check = v
	if c.Plan != nil {
		d.Plan = c.Plan
	}
	h.settleAndKeep(d)
}

// WalkMoved takes a walk's record, from `plan-moved` or the controller's walks: the delivery it walks
// moves with it; a backlog walk's carried builds are credited to the deliveries that published them.
func (h *Holder) WalkMoved(w Walk) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.walkMoved(w)
}

func (h *Holder) walkMoved(w Walk) {
	if w.Release != nil {
		h.creditBacklog(w)
		return
	}
	d := h.deliveryWalking(w)
	if d == nil {
		if _, seen := h.unmatched[w.ID]; !seen {
			h.unmatched[w.ID] = unmatched{walk: w, first: h.Now()}
		} else {
			u := h.unmatched[w.ID]
			if w.Revision >= u.walk.Revision {
				u.walk = w
				h.unmatched[w.ID] = u
			}
		}
		return
	}
	if d.Walk != nil && d.Walk.ID == w.ID && w.Revision < d.Walk.Revision {
		return // an older record arriving late
	}
	h.linkWalk(d, w)
	h.settleAndKeep(d)
	// What its sends carried of other deliveries' builds.
	for _, m := range w.Modules {
		if m != nil && m.Gate != nil {
			h.creditCarried(m.Gate, w.ID, d.ID)
		}
	}
}

// deliveryWalking is the delivery a walk walks: the one that landed as its commit, or that it walked before.
func (h *Holder) deliveryWalking(w Walk) *Delivery {
	for _, d := range h.deliveries {
		if d.Walk != nil && d.Walk.ID == w.ID {
			return d
		}
	}
	for _, d := range h.deliveries {
		if strings.EqualFold(d.Repository, w.Repository) && sameCommit(d.MergedAs, w.Commit) {
			return d
		}
	}
	return nil
}

// linkWalk records a walk on its delivery: what it is at, and the machine steps its record says — each
// moved only as the step table allows.
func (h *Holder) linkWalk(d *Delivery, w Walk) {
	d.Walk = w.Seen()
	for _, s := range StepsOf(w) {
		h.step(d, s)
	}
}

// step moves one machine's step, or refuses a move the step table does not hold; it says whether it moved.
func (h *Holder) step(d *Delivery, s Step) bool {
	for i, old := range d.Steps {
		if old.Machine == s.Machine && old.Module == s.Module {
			if old.State == s.State {
				return false
			}
			if !StepMay(old.State, s.State) {
				h.refuse(fmt.Errorf("%s: %s on %s from %s to %s is refused — the step table holds no such move",
					d.ID, s.Module, s.Machine, old.State, s.State))
				return false
			}
			d.Steps[i] = s
			return true
		}
	}
	d.Steps = append(d.Steps, s)
	return true
}

// creditBacklog keeps a backlog walk, and credits what its gate carried.
func (h *Holder) creditBacklog(w Walk) {
	replaced := false
	for i, b := range h.backlog {
		if b.ID == w.ID {
			h.backlog[i], replaced = w, true
		}
	}
	if !replaced {
		h.backlog = append(h.backlog, w)
		if len(h.backlog) > 5 {
			h.backlog = h.backlog[len(h.backlog)-5:]
		}
	}
	if w.Release.Gate != nil {
		h.creditCarried(w.Release.Gate, w.ID, "")
	}
}

// creditCarried gives the delivery that published each carried build its machine step.
func (h *Holder) creditCarried(g *WalkGate, walk, except string) {
	state := StepJudging
	switch {
	case g.Rollback == "rolled-back":
		state = StepRolledBack
	case g.Verdict == "failed":
		state = StepFailed
	case g.Verdict == "passed":
		state = StepPassed
	}
	for _, c := range g.Carried {
		for _, d := range h.deliveries {
			if d.ID == except || d.MergedAs == "" || !sameCommit(d.MergedAs, c.To) {
				continue
			}
			if h.step(d, Step{Machine: c.Node, Module: c.Module, Build: c.Build, State: state, Since: h.Now(),
				Why: "carried by " + walk}) {
				h.keep(d)
			}
		}
	}
}

// --- the controller's walks, read back --------------------------------------------------------------------

// Reconcile asks the controller for the walks it keeps and takes each as if said: what was missed while
// this holder was down, or said by a command that says nothing, is found by comparison.
func (h *Holder) Reconcile() error {
	held, walks, err := h.Controller.Walks("")
	h.mu.Lock()
	defer h.mu.Unlock()
	if err != nil {
		h.unreached = err.Error()
		return err
	}
	h.unreached, h.held, h.reconciled = "", held, h.Now()
	for _, w := range walks {
		h.walkMoved(w)
	}
	return nil
}

// --- the tick: groups, turns, adoption --------------------------------------------------------------------

// Tick takes every step the facts allow that needs another's word: a group's order and composed check, a
// delivery's turn, a walk with no delivery made one. The asking happens outside the lock.
func (h *Holder) Tick() {
	for _, ask := range h.plan() {
		ask()
	}
}

// plan works out, under the lock, what to ask; each ask applies its own answer.
func (h *Holder) plan() []func() {
	h.mu.Lock()
	defer h.mu.Unlock()
	now := h.Now()
	var asks []func()

	// Walks no merge was heard for: made deliveries of their own once their merge had time to be heard.
	for id, u := range h.unmatched {
		if d := h.deliveryWalking(u.walk); d != nil {
			h.linkWalk(d, u.walk)
			h.settleAndKeep(d)
			delete(h.unmatched, id)
			continue
		}
		if !u.walk.Open() {
			if now.Sub(u.first) > 10*time.Minute {
				delete(h.unmatched, id)
			}
			continue
		}
		if u.walk.Created.After(h.started) && now.Sub(u.first) < matchGrace {
			continue
		}
		h.adopt(u.walk)
		delete(h.unmatched, id)
	}

	// Plans not known for a delivery that has its diffset: asked once a tick.
	for _, d := range h.deliveries {
		if d.Plan == nil && !d.State.Final() && len(d.Paths) > 0 && (d.State == Ready || d.State == Held || d.MergedAs != "") {
			d := d
			id, repo, base, head, paths, dirs, removed := d.ID, d.Repository, d.Base, d.Commit, d.Paths, d.ModuleDirs, d.Removed
			asks = append(asks, func() {
				plan, err := h.Controller.Plan(repo, base, head, paths, dirs, removed)
				h.mu.Lock()
				defer h.mu.Unlock()
				if err != nil {
					h.Logf("[mesh-delivery] %s's plan cannot be asked yet: %v", id, err)
					return
				}
				if d := h.deliveries[id]; d != nil && d.Plan == nil {
					d.Plan = plan
					h.settleAndKeep(d)
				}
			})
		}
	}

	// Proposed, with no verdict and no check asked past the grace: its check is asked — a head adopted while
	// it already waited, or one whose announcement the controller missed.
	for _, d := range h.deliveries {
		if a := h.askCheck(d, now); a != nil {
			asks = append(asks, a)
		}
	}

	// Merged, and no walk opened for it after the walks were read again: never silent — held, saying so.
	for _, d := range h.deliveries {
		if d.State == Ready && d.MergedAs != "" && d.Walk == nil && !d.MovesNothing() && d.HeldWhy == "" &&
			now.Sub(d.MergedAt) > unwalked && h.reconciled.After(d.MergedAt.Add(unwalked)) {
			d.HeldWhy = "merged, and the controller opened no walk for it within " + unwalked.String() + " — nothing it " +
				"holds follows that branch, or the merge was not heard: a person decides"
			h.settleAndKeep(d)
		}
	}

	// Groups: membership, order, composed check, and their turns.
	grouped := map[string]bool{}
	for _, g := range h.groups {
		members := h.membersOf(g)
		for _, m := range members {
			grouped[m.ID] = true
		}
		asks = append(asks, h.planGroup(g, members, now)...)
	}

	// A lone delivery's turn: published, its walk waiting, nothing before it.
	for _, d := range h.deliveries {
		if grouped[d.ID] || d.Group != "" && h.groups[d.Group] != nil {
			continue
		}
		if a := h.turn(d, "its turn: a delivery in no group"); a != nil {
			asks = append(asks, a)
		}
	}

	// Final deliveries past their keeping, once nothing is owed for them.
	for id, d := range h.deliveries {
		if d.State.Final() && len(d.Owed) == 0 && now.Sub(d.Since) > keptFinal {
			if err := h.Store.DeleteDelivery(id); err == nil {
				delete(h.deliveries, id)
			}
		}
	}
	return asks
}

// askCheck is the ask of a proposed delivery's own check, when nothing else will bring its verdict: no verdict,
// no check this owner asked, and proposed past the grace the controller's own check of its announcement is
// given. Nil otherwise — once asked, a late verdict is the bound's and H2's to speak for.
func (h *Holder) askCheck(d *Delivery, now time.Time) func() {
	if d.State != Proposed || d.Check != nil || d.CheckAsk != nil || d.MergedAs != "" || d.Number == 0 ||
		now.Sub(d.Since) < checkGrace {
		return nil
	}
	if tried, was := h.askTried[d.ID]; was && now.Sub(tried) < time.Minute {
		return nil
	}
	h.askTried[d.ID] = now
	id, m := d.ID, MemberOf(d)
	why := fmt.Sprintf("proposed for %s with no verdict and no check asked: its announcement was heard from the "+
		"bus's history, or the controller missed it", now.Sub(d.Since).Round(time.Minute))
	return func() {
		asked, err := h.Controller.Check("", []Member{m})
		h.mu.Lock()
		defer h.mu.Unlock()
		if err != nil {
			h.Logf("[mesh-delivery] %s's check cannot be asked yet: %v", id, err)
			return
		}
		delete(h.askTried, id)
		d := h.deliveries[id]
		if d == nil || d.State != Proposed || d.Check != nil {
			return
		}
		d.CheckAsk = &CheckAsk{ID: asked, At: h.Now(), By: byOwner, Why: why}
		h.Logf("[mesh-delivery] %s: its check asked of the controller as %s — %s", id, asked, why)
		d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.Commit, Since: h.Now(),
			Line: oneLine(fmt.Sprintf("%s delivery %s: its check asked by %s as %s: %s", h.Now().UTC().Format(time.RFC3339),
				id, byOwner, asked, why))})
		h.keep(d)
	}
}

// adopt makes a delivery of a walk no pull request is known for: running already — at the switch, or on the
// controller's own path — it is delivering; waiting for a word, it waits for a person.
func (h *Holder) adopt(w Walk) {
	now := h.Now()
	d := &Delivery{ID: IDOf(w.Repository, w.Commit), Repository: strings.ToLower(w.Repository), Commit: w.Commit,
		Base: w.Branch, MergedAs: w.Commit, Created: now}
	if old := h.deliveries[d.ID]; old != nil {
		h.linkWalk(old, w)
		h.settleAndKeep(old)
		return
	}
	h.linkWalk(d, w)
	var t Transition
	var err error
	if d.Walk.Waits {
		d.HeldWhy = "a commit on the trunk with no pull request this owner heard of — merged before it held the " +
			"seat, or pushed to the trunk — waits for a person's release"
		t, err = Apply(d, EvAppeared, false, Facts{Now: now, Why: d.HeldWhy})
	} else {
		t, err = Apply(d, EvAdopted, false, Facts{Now: now, Why: "a walk already running when this owner first heard " +
			"of it (" + w.ID + "), adopted as its delivery"})
	}
	if err != nil {
		h.refuse(err)
		return
	}
	h.moved(d, []Transition{t})
	h.settleAndKeep(d)
}

// turn is the ask that lets a published delivery's waiting walk start; nil when it is not its turn.
func (h *Holder) turn(d *Delivery, why string) func() {
	if d.Walk == nil || !d.Walk.Waits || !d.LetGo.IsZero() {
		return nil
	}
	if d.State != Published && d.State != Delivering {
		return nil
	}
	id, walk := d.ID, d.Walk.ID
	return func() {
		err := h.Controller.Deliver(walk, why)
		h.mu.Lock()
		defer h.mu.Unlock()
		d := h.deliveries[id]
		if d == nil {
			return
		}
		if err != nil {
			h.Logf("[mesh-delivery] %s's walk %s could not be let go yet: %v", id, walk, err)
			return
		}
		d.LetGo, d.LetGoWhy = h.Now(), why
		if d.Walk != nil && d.Walk.ID == walk {
			d.Walk.Waits, d.Walk.GoBy = false, byOwner
		}
		h.settleAndKeep(d)
	}
}

// membersOf are a group's current members, by id order.
func (h *Holder) membersOf(g *Group) []*Delivery {
	var out []*Delivery
	for _, id := range g.Members {
		if d := h.deliveries[id]; d != nil {
			out = append(out, d)
		}
	}
	return out
}

// planGroup is what a group needs asked, and its turns.
func (h *Holder) planGroup(g *Group, members []*Delivery, now time.Time) []func() {
	var asks []func()
	if len(members) == 0 {
		return nil
	}
	if len(members) == 1 && !g.Closed {
		// A group of one is no group: its member is a lone delivery again.
		members[0].Group = ""
		h.keep(members[0])
		g.Members = nil
		h.keepGroup(g)
		return nil
	}
	commits := commitsOf(members)
	// The order, worked out again whenever a member's head moved.
	if !sameMembers(g.OrderOf, commits) && !g.Closed {
		ms := make([]Member, 0, len(members))
		for _, d := range members {
			ms = append(ms, MemberOf(d))
		}
		id := g.ID
		asks = append(asks, func() {
			o, err := h.Controller.Order(ms)
			h.mu.Lock()
			defer h.mu.Unlock()
			g := h.groups[id]
			if g == nil {
				return
			}
			if err != nil {
				h.Logf("[mesh-delivery] group %s's order cannot be asked yet: %v", id, err)
				return
			}
			g.Order, g.Pairs, g.Cycle, g.OrderOf = o.Order, o.Pairs, o.Cycle, commits
			g.Check = nil // a check of another set of heads is no check of these
			g.Updated = h.Now()
			h.keepGroup(g)
		})
		return asks
	}
	// The composed check, once every member is ready and the order holds.
	allReady := true
	for _, d := range members {
		if d.State != Ready {
			allReady = false
		}
	}
	if allReady && len(g.Cycle) == 0 && !g.Closed && (g.Check == nil || !sameMembers(g.Check.Commits, commits)) {
		ms := h.inOrder(g, members)
		id := g.ID
		asks = append(asks, func() {
			asked, err := h.Controller.Check(id, ms)
			h.mu.Lock()
			defer h.mu.Unlock()
			g := h.groups[id]
			if g == nil {
				return
			}
			if err != nil {
				h.Logf("[mesh-delivery] group %s's composed check cannot be asked yet: %v", id, err)
				return
			}
			g.Check = &GroupCheck{Asked: asked, At: h.Now(), Commits: commits}
			g.Updated = h.Now()
			h.keepGroup(g)
		})
	}
	// Its state, said when it changes.
	if state, why := GroupState(g, members); state != g.Said {
		g.Said = state
		for _, d := range members {
			g.Owed = append(g.Owed, Effect{Kind: EffectStatus, Commit: d.Commit, Context: "mesh/delivery-group",
				State: groupStatusState(state), Line: clip(fmt.Sprintf("group %s %s: %s", g.ID, state, why), 140),
				Since: now, Last: d.Repository + "#" + fmt.Sprint(d.Number) + " " + d.HTMLURL})
		}
		g.Owed = append(g.Owed, Effect{Kind: EffectEmit, Line: string(mustJSON(map[string]any{"group": g.ID,
			"state": state, "why": why, "members": g.Members, "order": g.Order})), Since: now})
		g.Updated = now
		h.keepGroup(g)
	}
	// Its turns, in its order.
	if g.Closed {
		asks = append(asks, h.groupTurns(g, members)...)
	}
	return asks
}

// inOrder are a group's members as the controller ordered them.
func (h *Holder) inOrder(g *Group, members []*Delivery) []Member {
	byID := map[string]*Delivery{}
	for _, d := range members {
		byID[d.ID] = d
	}
	var out []Member
	for _, id := range g.Order {
		if d := byID[id]; d != nil {
			out = append(out, MemberOf(d))
			delete(byID, id)
		}
	}
	var rest []string
	for id := range byID {
		rest = append(rest, id)
	}
	sort.Strings(rest)
	for _, id := range rest {
		out = append(out, MemberOf(byID[id]))
	}
	return out
}

// groupTurns walks a delivering group in its order: the next member is let go only when every member before
// it is delivered; a member that failed or was stopped stops every member after it, naming it. What was
// delivered before it stays, each judged at its own gate (ADR 0239 decision 4); the failed member's builds
// were put back at its gate by the controller.
func (h *Holder) groupTurns(g *Group, members []*Delivery) []func() {
	ordered := h.inOrder(g, members)
	byID := map[string]*Delivery{}
	for _, d := range members {
		byID[d.ID] = d
	}
	// Composed for these very heads: a verdict on another set of heads is no verdict on these.
	composed := g.Check != nil && (g.Check.Verdict == "pass" || g.Check.Verdict == "warning") &&
		sameMembers(g.Check.Commits, commitsOf(members))
	for i, m := range ordered {
		d := byID[m.ID]
		switch {
		case d.State == Delivered:
			continue
		case d.State == Failed || d.State == Stopped || d.State == Superseded:
			var asks []func()
			for _, later := range ordered[i+1:] {
				if a := h.stopInGroup(byID[later.ID], fmt.Sprintf("its group %s stopped: %s %s", g.ID, d.ID, d.State)); a != nil {
					asks = append(asks, a)
				}
			}
			return asks
		case d.State == Published:
			if !composed && d.HeldWhy == "" {
				d.HeldWhy = fmt.Sprintf("its group %s's composed check did not pass for the heads that merged; a person "+
					"decides that it goes on", g.ID)
				h.settleAndKeep(d)
				return nil
			}
			if a := h.turn(d, fmt.Sprintf("its turn in group %s (%d of %d)", g.ID, i+1, len(ordered))); a != nil {
				return []func(){a}
			}
			return nil
		default:
			return nil // not on the trunk yet, held, or delivering: the members after it wait
		}
	}
	return nil
}

// stopInGroup stops a member after one that failed, and its walk if it has one open.
func (h *Holder) stopInGroup(d *Delivery, why string) func() {
	if d == nil || d.State.Final() {
		return nil
	}
	t, err := Apply(d, EvStop, true, Facts{Now: h.Now(), By: byOwner, Why: why})
	if err != nil {
		h.refuse(err)
		return nil
	}
	h.moved(d, []Transition{t})
	if d.Walk == nil || d.Walk.State == walkDone || d.Walk.State == walkFailed || d.Walk.State == walkSuperseded {
		return nil
	}
	walk := d.Walk.ID
	return func() {
		if err := h.Controller.Stop(walk, why, byOwner); err != nil {
			h.Logf("[mesh-delivery] the walk %s could not be stopped: %v", walk, err)
		}
	}
}

// GroupState is a group's state, derived from its members and its composed check — never set apart.
func GroupState(g *Group, members []*Delivery) (string, string) {
	count := map[State]int{}
	for _, d := range members {
		count[d.State]++
	}
	n := len(members)
	switch {
	case count[Failed] > 0:
		return "failed", "a member failed"
	case count[Stopped] > 0 && g.Closed:
		return "stopped", "a member was stopped"
	case n > 0 && count[Delivered] == n:
		return "delivered", "every member is delivered"
	case g.Closed:
		return "delivering", fmt.Sprintf("%d of %d delivered", count[Delivered], n)
	case len(g.Cycle) > 0:
		return "rejected", "its order contradicts itself: " + strings.Join(g.Cycle, ", ")
	case count[Rejected] > 0:
		return "rejected", "a member's own check failed"
	case g.Check != nil && g.Check.Verdict != "" && g.Check.Verdict != "pass" && g.Check.Verdict != "warning":
		return "rejected", "composed together: " + g.Check.Summary
	case count[Ready] == n && g.Check != nil && sameMembers(g.Check.Commits, commitsOf(members)) &&
		(g.Check.Verdict == "pass" || g.Check.Verdict == "warning"):
		return "ready", "every member ready, and composed together they pass"
	}
	return "checking", fmt.Sprintf("%d of %d member(s) ready", count[Ready], n)
}

func groupStatusState(s string) string {
	switch s {
	case "ready", "delivered":
		return "success"
	case "rejected", "failed":
		return "failure"
	case "stopped":
		return "error"
	}
	return "pending"
}

func clip(s string, n int) string {
	if len(s) > n {
		return s[:n-1] + "…"
	}
	return s
}

func containsString(xs []string, x string) bool {
	for _, y := range xs {
		if y == x {
			return true
		}
	}
	return false
}