Files
mesh-catalog/modules/mesh-delivery-gate/cmd/mesh-delivery-gate/holder.go
T
jschoubben 30d04f6bef Pin the gate to mesh-sdk's release 6 verb tables and prove it holds the seat as they have it (issue 1946, R6-C3)
The integration tests hear the forge's own word that a pull request merged, as the real flow does, and fail on any call of the controller.
2026-10-24 16:35:15 +02:00

1088 lines
40 KiB
Go

package main
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"slices"
"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 {
// Settings reads the operator's values (settings.go); nil keeps every default.
Settings func() Settings
Store Store
Forge Forge
Emit func(event string, body any) error
Now func() time.Time
Logf func(format string, a ...any)
// Operator keeps the ask of the Approve of a merge into main in the gate's asks and reads the teller's answers
// (approval.go); Merges and Source are the gate's merges and source (merges.go).
Operator Operator
Merges MergeStore
Source SourceStore
// Owed is the gate's bucket owed: what a transition owes the forge, written as keys (owed.go).
Owed OwedStore
// Pulls keeps the pull requests' keys on the bus (pullkey.go); nil keeps none.
Pulls PullKeys
// cancels are asks to take back, gathered under the lock and run with the other asks of the look.
cancels []func()
// withdrawn are the ask keys withdrawn since the last look (approval.go withdrawAsk).
withdrawn map[string]bool
mu sync.Mutex
deliveries map[string]*Delivery
groups map[string]*Group
dirty map[string]bool
// written is each delivery as it was last written, encoded: a delivery that encodes the same is not
// written again (novox/hq issue 326).
written map[string][]byte
dirtyGroup map[string]bool
started time.Time
loaded bool
// 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
// lostRead is when the build queue was last read for a proposed delivery's asked check (issue 1274).
lostRead map[string]time.Time
refusals []string
// unchecked are the open pull requests' heads seen without a merge check (unchecked.go).
unchecked map[string]unchecked
// checker are the checker's verdicts by the hash of the heads they judged, as its verdicts bucket holds them
// (onestatus.go, issue 1302).
checker map[string]checkerVerdict
// tried counts the owed effects Flush tried since this owner started, by how each ended (novox/hq issue 407).
tried owedTries
// forgeWanted is whether the forge's open pull requests are to be read at the next React: a check may have
// answered (events.go). reactMu keeps one React at a time.
forgeWanted bool
reactMu sync.Mutex
// poke asks for a look (react.go); wanted is when the next is wanted, and its timer.
poke chan struct{}
wanted lookWanted
}
// owedTries is how many owed effects were tried since this owner started, and when the last was: three numbers
// and a time, however many were tried. Not kept: a restarted owner counts from its start, and says since when, so a
// window's count is the difference of two readings in the same run (novox/hq issue 407).
type owedTries struct {
done, failed, givenUp int
last time.Time
}
// The words this owner signs its acts with.
const (
byOwner = "mesh-delivery"
byH2 = "healer H2"
)
// 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.poke = make(chan struct{}, 1)
h.deliveries = map[string]*Delivery{}
h.groups = map[string]*Group{}
h.dirty = map[string]bool{}
h.withdrawn = map[string]bool{}
h.written = map[string][]byte{}
h.dirtyGroup = map[string]bool{}
h.askTried = map[string]time.Time{}
h.lostRead = map[string]time.Time{}
h.unchecked = map[string]unchecked{}
h.checker = map[string]checkerVerdict{}
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
if b, err := json.Marshal(d); err == nil {
h.written[d.ID] = b
}
// A delivery an earlier owner kept in a state that is gone (published, delivering, delivered, covered, stopped and the
// like) reads as the merge's state it came to, and is kept so (ADR 0539 §17).
if s := legacyState(d); s != d.State {
h.Logf("[mesh-delivery] %s was kept as %s: read as %s", d.ID, d.State, s)
d.State = s
h.dirty[d.ID] = true
}
}
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()
opened := h.openedOf(d)
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, opened))), 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 len(ts) > 0 && d.shownOnItsPullRequest() {
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)
}
// afterTheTrunk is whether a state is one a change is in once it reached the trunk: its note is kept on the merge
// commit too.
func afterTheTrunk(s State) bool { return s == Merged || s == Held }
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. A delivery that encodes exactly as it was last written is not written again: a revision of its key
// means the delivery changed (novox/hq issue 326). Any change to anything it holds — its state, steps, walk,
// owed effects and their tries — changes its encoding, so it is written.
func (h *Holder) keep(d *Delivery) {
h.deliveries[d.ID] = d
b, encoded := json.Marshal(d)
if encoded == nil && !h.dirty[d.ID] && h.written[d.ID] != nil && bytes.Equal(b, h.written[d.ID]) {
return
}
if err := h.Store.PutDelivery(d); err != nil {
h.dirty[d.ID] = true
delete(h.written, d.ID)
h.Logf("[mesh-delivery] %s could not be kept yet (%v); kept again before it is said", d.ID, err)
h.again("a delivery that could not be kept")
return
}
delete(h.dirty, d.ID)
if encoded == nil {
h.written[d.ID] = b
} else {
delete(h.written, 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)
h.again("a group that could not be kept")
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"`
// BaseSHA is the base branch's head when the head was announced (novox/hq ADR 0323).
BaseSHA string `json:"base_sha,omitempty"`
// Merged is the forge's own events' word that the pull request merged (gitea.forge.pull-request.*): its close is no
// close unmerged (issue 1107).
Merged bool `json:"merged,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) (isNew bool) {
if p.HeadSHA == "" || p.Number == 0 {
h.Refuse(errors.New("a pull request announced with no head or no number; nothing proposed"))
return false
}
h.mu.Lock()
defer h.mu.Unlock()
return h.pullUpdated(p)
}
// pullUpdated is PullUpdated under the lock.
func (h *Holder) pullUpdated(p PullEvent) (isNew bool) {
id := IDOf(p.repository(), p.HeadSHA)
isNew = h.deliveries[id] == nil
now := h.Now()
startedOver := ""
if d := h.deliveries[id]; d != nil {
if startedOver = startedOverFor(d, p.Number); startedOver == "" {
// Announced again: nothing new — but for a delivery whose verdict is out of date, the announcement is its
// check asked again against the trunk's head (the merge tool's ask, novox/hq ADR 0323 §5), and its bound
// runs from now (§7). The base head it was announced with is kept, for a check this owner asks after it.
if p.BaseSHA != "" && d.BaseSHA != p.BaseSHA {
d.BaseSHA = p.BaseSHA
h.keep(d)
}
// A delivery made from its verdict before its head was heard (issue 1726) holds none of what an announcement
// carries: filled from this one, so its base is known and the operator is asked.
h.fillFromAnnouncement(d, p)
// The review record, corrected in the description without a push, is read again (ADR 0421 §5).
if by, at := ReviewIn(p.Body); p.Body != "" && (by != d.ReviewedBy || at != d.ReviewedAt) {
d.ReviewedBy, d.ReviewedAt = by, at
h.keep(d)
}
// So is the line naming who pushed it (ADR 0505 §5).
if by := PushedIn(p.Body); p.Body != "" && by != d.PushedBy {
d.PushedBy = by
h.keep(d)
}
// An `after:` line written into the description later is told about too (novox/hq ADR 0419 §2).
if after := AfterIn(p.Body); p.Body != "" && !slices.Equal(after, d.After) {
d.After = after
h.keep(d)
}
if d.State == Proposed && d.outOfDate() != "" && d.AskedAgain == nil {
d.AskedAgain = &now
h.keep(d)
}
// Another pull request at this head, announced while this one is live (issue 1114): kept, so this one's close
// starts the record over for it.
if d.Number != p.Number && !d.State.Final() && d.MergedAs == "" {
d.Others = withOther(d.Others, p)
h.keep(d)
}
return
}
// A delivery stopped only because its pull request closed unmerged does not end its commit's delivery for
// another pull request with the same head (novox/hq issue 783): started over below, exactly as for a first
// announcement, under the same id — never a transition out of Stopped.
}
if startedOver != "" {
// The record the new pull request takes over leaves the group it was in (issue 1083): the new one joins its own.
if old := h.deliveries[id]; old != nil && old.Group != "" {
h.leaveGroup(old)
}
}
replaced := h.supersedeOlder(p)
d, t, err := h.freshAnnouncement(p, now, startedOver)
if err != nil {
h.refuse(err)
return false
}
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)
if !h.takePlaceOf(replaced, d) {
h.joinGroup(d)
}
return isNew
}
// supersedeOlder ends the older heads of p's pull request: each is told its newer head, and one in a group leaves it. It
// returns the id of the one that left a group, whose place the new head takes (takePlaceOf). Under the lock.
func (h *Holder) supersedeOlder(p PullEvent) (replaced string) {
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)
replaced = old.ID
}
}
}
return replaced
}
// fillFromAnnouncement fills a delivery made from its check's verdict before its pull request's head was heard (issue
// 1726: no base, branch, title, URLs or review record) from an announcement or an edit of that pull request, and then
// does what a first announcement does: the older heads of the pull request are superseded, and the delivery takes
// their place in a group or joins the group its branch makes. Nothing for a delivery that holds its base. Under the lock.
func (h *Holder) fillFromAnnouncement(d *Delivery, p PullEvent) {
if d.Base != "" || p.Base == "" || d.State.Final() || d.MergedAs != "" {
return
}
d.Title, d.Base, d.Branch, d.HTMLURL, d.CloneURL = p.Title, p.Base, p.Head, p.HTMLURL, p.CloneURL
d.Paths, d.PathsTruncated, d.Removed = p.Paths, p.PathsTruncated, p.Removed
d.ModuleDirs, d.ModuleDirsSaid = p.ModuleDirs, p.ModuleDirsSaid
d.Issues, d.After = IssuesIn(p.Title+"\n"+p.Body), AfterIn(p.Body)
d.ReviewedBy, d.ReviewedAt = ReviewIn(p.Body)
d.PushedBy = PushedIn(p.Body)
if p.BaseSHA != "" {
d.BaseSHA = p.BaseSHA
}
d.Number = p.Number
h.Logf("[mesh-delivery] %s: its pull request's announcement filled its base (%s), branch and title", d.ID, p.Base)
h.keep(d)
replaced := h.supersedeOlder(p)
if d.Group == "" && !h.takePlaceOf(replaced, d) {
h.joinGroup(d)
}
h.Poke()
}
// PullEdited takes a pull request's edit (novox/hq issue 1079): its description is read again for the review record, the
// `after:` lines and the issues it names, exactly as an announcement reads them. An Approve ask still unanswered whose words
// the new record changes is withdrawn and asked again with it read, without a word on the pull request: nothing was
// approved, so no Approve ended. An answered ask is left as it is. A record naming the head lifts the hold at once (ADR 0496 §4). An edit with no
// description, or of a pull request with no open delivery at that head, changes nothing.
func (h *Holder) PullEdited(p PullEvent) {
if p.HeadSHA == "" || p.Number == 0 {
return
}
h.mu.Lock()
defer h.mu.Unlock()
d := h.deliveries[IDOf(p.repository(), p.HeadSHA)]
if d != nil && startedOverFor(d, p.Number) != "" {
// The record at this head is stopped under another, closed pull request (issue 1114: this one was announced while
// that one was live and never heard again): the edit takes it over, as an announcement would. The forge's edit
// event carries no paths; the record's, for the same head, are the ones.
if p.Paths == nil {
p.Paths, p.PathsTruncated, p.Removed = d.Paths, d.PathsTruncated, d.Removed
p.ModuleDirs, p.ModuleDirsSaid = d.ModuleDirs, d.ModuleDirsSaid
}
if p.CloneURL == "" {
p.CloneURL = d.CloneURL
}
if p.BaseSHA == "" {
p.BaseSHA = d.BaseSHA
}
h.pullUpdated(p)
return
}
if d != nil && d.Number == p.Number {
h.fillFromAnnouncement(d, p) // a delivery made from its verdict, its head never heard (issue 1726)
}
if p.Body == "" {
return
}
if d == nil || d.Number != p.Number || d.State.Final() || d.MergedAs != "" {
return
}
before, _ := reviewWords(d)
d.ReviewedBy, d.ReviewedAt = ReviewIn(p.Body)
d.PushedBy = PushedIn(p.Body)
d.After = AfterIn(p.Body)
title := p.Title
if title == "" {
title = d.Title
}
d.Issues = IssuesIn(title + "\n" + p.Body)
h.keep(d)
if after, _ := reviewWords(d); after == before {
return
}
if r := h.covering(d); r != nil && r.AskID != "" && r.Approved == nil && r.Declined == nil {
h.cancels = append(h.cancels, h.withdrawAsk(r.AskID, d.ID+"'s review record changed: "+shortOf(d.ReviewedAt, 12)))
}
}
// withOther is others with p put in, replacing an earlier announcement of the same pull request.
func withOther(others []PullEvent, p PullEvent) []PullEvent {
out := withoutOther(others, p.Number)
return append(out, p)
}
// withoutOther is others without the pull request number.
func withoutOther(others []PullEvent, number int) []PullEvent {
var out []PullEvent
for _, o := range others {
if o.Number != number {
out = append(out, o)
}
}
return out
}
// startedOverFor is why a delivery already kept at a head's id is started over for a pull request's number rather
// than taken as announced again: it is Closed, never merged, and under another number. Stopped by an earlier, closed
// duplicate pull request with the same head (novox/hq issue 783), or by a person or a group before its pull request
// closed (issue 1083: the new pull request at that commit takes the record over, the key staying repository@commit).
// Empty for the ordinary 'announced again' path, or for the same pull request's own stop.
func startedOverFor(d *Delivery, number int) string {
if d == nil || d.State != Closed || d.NewerHead != "" || d.MergedAs != "" || d.Number == number {
return ""
}
if d.ClosedUnmerged {
return fmt.Sprintf("#%d with the same head closed unmerged", d.Number)
}
said := fmt.Sprintf("#%d with the same head was stopped", d.Number)
if n := len(d.Transitions); n > 0 && d.Transitions[n-1].To == Closed {
last := d.Transitions[n-1]
if last.By != "" {
said += " by " + last.By
}
if last.Why != "" {
said += ": " + last.Why
}
}
return said
}
// freshAnnouncement makes a delivery for p's head exactly as a first announcement would: its own id, proposed.
// startedOver, when not empty, is folded into the transition's Why, naming the pull request this replaces
// (novox/hq issue 783); the delivery it replaces is swapped out by the caller keeping the new one under the same
// id (keep), never by a transition out of its final state.
func (h *Holder) freshAnnouncement(p PullEvent, now time.Time, startedOver string) (*Delivery, Transition, error) {
d := &Delivery{ID: IDOf(p.repository(), p.HeadSHA), 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), Issues: IssuesIn(p.Title + "\n" + p.Body), Created: now, BaseSHA: p.BaseSHA}
d.ReviewedBy, d.ReviewedAt = ReviewIn(p.Body)
d.PushedBy = PushedIn(p.Body)
why := fmt.Sprintf("%s#%d's head announced", d.Repository, d.Number)
if startedOver != "" {
why = fmt.Sprintf("%s; %s", why, startedOver)
}
t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: why})
return d, t, err
}
// takePlaceOf puts a member's new head — a rebase, a push — in its group in the place of the head it superseded,
// when the group is closed: a member of it already reached the trunk, and a closed group takes no new member. The
// new head has its own check, and the group's composed check runs again for the new heads rather than holding the
// group for a person (novox/hq ADR 0323 §6, issue 406). An open group takes it as joinGroup does.
func (h *Holder) takePlaceOf(old string, d *Delivery) bool {
if old == "" {
return false
}
was := h.deliveries[old]
if was == nil || was.Group == "" {
return false
}
if g := h.groups[was.Group]; g != nil && g.Closed {
if !containsString(g.Members, d.ID) {
g.Members = append(g.Members, d.ID)
sort.Strings(g.Members)
}
d.Group = g.ID
g.Updated = h.Now()
h.keep(d)
h.keepGroup(g)
h.Logf("[mesh-delivery] group %s: %s takes the place of %s, which it superseded", g.ID, d.ID, old)
return true
}
return false
}
// decidedOnHead is the verdict the forge's statuses on a head hold, when the checker decided one: passed, warned or
// failed. Nil for none, pending, or a check that could not run — the checker checks those again.
func decidedOnHead(checks map[string]string, now time.Time) *Verdict {
switch checks[ctxCheck] {
case "success", "warning", "failure":
default:
return nil
}
return verdictOfCheck(checks[ctxCheck], "", now)
}
// 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()
h.forgetUnchecked(p.Owner, p.Repo, p.Number)
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 {
// One already on the trunk is not merged again: a forge whose numbers began again reuses them.
if c.Repository == p.repository() && c.Number == p.Number && c.NewerHead == "" && c.MergedAs == "" &&
c.Created.After(newest) {
d, newest = c, c.Created
}
}
}
if d != nil && d.MergedAs != "" && sameCommit(d.MergedAs, p.MergeCommit) {
return // this merge is already taken: heard again, past the bus's de-duplication window, it changes nothing
}
now := h.Now()
var ts []Transition
if over := startedOverFor(d, p.Number); over != "" {
// A delivery stopped only by an earlier, closed duplicate pull request with the same head is not lost when
// its merge is heard without the open announcement either: it is started over for the merging pull request
// the same way an announcement would, before the merge is taken as usual (novox/hq issue 783). Heard this
// way — its own merge, with no open announcement of this pull request — it is made exactly as a first-heard
// merge in every field but its id: FromForge, and its check the forge's word on the head, so a long-past
// merge replayed from the bus's history still falls under the history rule (issue 313) and is not held for
// a person (review round 1).
if d.Group != "" {
h.leaveGroup(d) // issue 1083: the record taken over leaves its group
}
fresh, t, err := h.freshAnnouncement(p, now, over)
if err != nil {
h.refuse(err)
return
}
fresh.FromForge = true
if s := p.HeadChecks[ctxCheck]; s != "" {
fresh.Check = verdictOfCheck(s, "", now)
}
d, ts = fresh, []Transition{t}
if d.Check != nil {
ts = append(ts, settle(d, Facts{Now: now})...)
}
} else 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), Issues: IssuesIn(p.Title + "\n" + p.Body), Created: now, FromForge: true}
// Its check, as the forge holds it on the head: the verdict the pull request merged with.
if s := p.HeadChecks[ctxCheck]; s != "" {
d.Check = verdictOfCheck(s, "", now)
}
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
}
ts = append(ts, t)
}
d.MergedAs = p.MergeCommit
if t, err := time.Parse(time.RFC3339, p.MergedAt); err == nil {
d.MergedAt = t.UTC()
} else {
d.MergedAt = now
}
ts = append(ts, settle(d, Facts{Now: now})...)
if d.State == Held && d.HeldWhy == "" {
d.HeldWhy = "merged without a passing check: only a person decides what follows"
}
h.moved(d, ts)
h.closeGroupOf(d)
h.trunkMoved(d, p.Base, p.MergeCommit, now)
}
// trunkMoved takes a merge into a trunk as that trunk's move (novox/hq ADR 0323 §7): every other open delivery into
// it keeps the trunk's new head, and a ready one whose verdict names the commit before goes back to proposed. A
// group's composed check goes out of date when the merge is not one of its members and a member still open merges
// into that trunk (§6); the group then reads checking. Nothing is asked here: at the same look askFresh asks the check of
// a group's member, or of a head with an Approve asked or given (issue 1047); a lone head's is judged again by a recheck
// or the merge tool (the rest of ADR 0395 §1 is issue 888).
func (h *Holder) trunkMoved(merged *Delivery, base, commit string, now time.Time) {
if base == "" || commit == "" {
return
}
for _, o := range h.deliveries {
if o == merged || o.Repository != merged.Repository || o.Base != base || o.MergedAs != "" || o.State.Final() ||
sameCommit(o.TrunkNow, commit) {
continue
}
// A check asked again against the trunk before this move judged that trunk: it is out of date too.
o.TrunkNow, o.AskedAgain = commit, nil
h.settleAndKeep(o)
}
for _, g := range h.groups {
if g.ID == merged.Group || g.Check == nil || g.Check.Verdict == "" || g.Check.OutOfDate != "" {
continue
}
for _, m := range h.membersOf(g) {
if m.Repository != merged.Repository || m.Base != base || m.MergedAs != "" || m.State.Final() {
continue
}
judged := g.Check.Trunks[m.ID]
if judged == "" {
judged = "?"
}
g.Check.OutOfDate = fmt.Sprintf("composed check out of date: %s was judged against trunk %s, the trunk is now %s "+
"(%s merged)", m.ID, shortOf(judged, 12), shortOf(commit, 12), merged.ID)
g.Check.OutOfDateAt = now
g.Updated = now
h.keepGroup(g)
h.Logf("[mesh-delivery] group %s: %s", g.ID, g.Check.OutOfDate)
break
}
}
}
// 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.
//
// What its checks built is no longer asked to be let go of here (issue 1318): the controller's forget-checks goes with
// the merge gate's checks (ADR 0503 §3), and the build seat's holders let go of it within the week.
func (h *Holder) PullClosed(p PullEvent) {
h.mu.Lock()
h.forgetUnchecked(p.Owner, p.Repo, p.Number)
var over *PullEvent
for _, d := range h.deliveries {
if d.Repository != p.repository() {
continue
}
if d.Number != p.Number {
// The other pull request at this head closed first: it is no longer one to take the record over.
if len(d.Others) > 0 {
if rest := withoutOther(d.Others, p.Number); len(rest) != len(d.Others) {
d.Others = rest
h.keep(d)
}
}
continue
}
if !d.State.Final() && d.MergedAs == "" {
d.ClosedUnmerged = true
h.settleAndKeep(d)
if d.Group != "" {
h.leaveGroup(d)
}
// Another open pull request carries this head (issue 1114): the record starts over for it at once, as a first
// announcement of it would, so it gets its check, its review record and its Approve ask.
if len(d.Others) > 0 {
next := d.Others[0]
over = &next
}
}
}
if over != nil {
h.pullUpdated(*over)
}
h.mu.Unlock()
}
// --- the look: checks, groups --------------------------------------------------------------------
// look takes every step the facts allow that needs another's word: a proposed delivery's check, a group's composed
// verdict, the Approve. The asking happens outside the lock.
func (h *Holder) look() {
for _, ask := range h.plan() {
ask()
}
h.hearAsks()
h.mu.Lock()
asks := h.readyAsks(h.Now())
h.mu.Unlock()
for _, ask := range asks {
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()
// Proposed, with no verdict: the checker's mesh/check that stands on its head at the forge is adopted (issue 1115);
// nothing is asked of the controller (issue 1318).
if a := h.askChecks(now); a != nil {
asks = append(asks, a)
}
// Groups: membership, composed check, and their word.
for _, g := range h.groups {
asks = append(asks, h.planGroup(g, h.membersOf(g), now)...)
}
// 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)
delete(h.written, id)
}
}
}
return asks
}
// askChecks reads the forge's open pull requests once for the proposed deliveries with no verdict (issue 1115), and takes the
// checker's mesh/check where one stands on a head (issue 1302); the base's required statuses are kept. Nothing is asked of
// the controller: the checker checks every head on every move (ADR 0447 §4, issue 1318). Under the lock; the read runs outside it.
func (h *Holder) askChecks(now time.Time) func() {
type want struct {
id, owner, repo, head string
number int
}
var wants []want
for _, d := range h.deliveries {
if d.State != Proposed || d.Check != nil || d.MergedAs != "" || d.Number == 0 {
continue
}
w := want{id: d.ID, head: d.Commit, number: d.Number}
w.owner, w.repo, _ = strings.Cut(d.Repository, "/")
wants = append(wants, w)
}
if len(wants) == 0 {
return nil
}
return func() {
open, err := h.Forge.OpenPulls()
if err != nil {
return
}
for _, w := range wants {
for _, p := range open {
if p.Number == w.number && strings.EqualFold(p.Owner, w.owner) && strings.EqualFold(p.Repo, w.repo) &&
sameCommit(p.HeadSHA, w.head) {
h.mu.Lock()
h.noteRequired(h.deliveries[w.id], p.Required)
if c := p.Checks[ctxCheck]; c.State != "" {
h.takeCheckStatus(w.owner+"/"+w.repo, w.head, c.State, c.Description)
}
h.mu.Unlock()
}
}
}
}
}
// 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
}
// A group is judged by mesh/check (issue 1302, ADR 0447 §2): its verdict is the checker's one verdict at its members'
// heads, read, not asked; nothing asks a composed check of the controller and no member is owed a mesh/merge-gate
// (issue 1318).
h.composeFromChecker(g, members, now)
// Its state, said when it changes.
state, why, held := h.groupState(g, members)
status := groupStatusState(state)
if held {
status = "pending" // no Approve exists while a member waits for its review (novox/hq ADR 0496 §2)
}
if state == "merging" && !(g.Check != nil && (g.Check.Verdict == "pass" || g.Check.Verdict == "warning") &&
sameMembers(g.Check.Commits, commitsOf(members)) && g.Check.OutOfDate == "") {
// Success there lets the merge tool merge its members: only for a composed check that passed for these heads.
status = "pending"
}
reviewSaid := ""
if held {
reviewSaid = why
}
if state != g.Said || status != g.SaidStatus || reviewSaid != g.SaidReview {
g.Said, g.SaidStatus, g.SaidReview = state, status, reviewSaid
for _, d := range members {
g.Owed = append(g.Owed, Effect{Kind: EffectStatus, Commit: d.Commit, Context: "mesh/delivery-group",
State: status, 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})), Since: now})
g.Updated = now
h.keepGroup(g)
}
return asks
}
// GroupState is a group's state, derived from its members and its composed check — never set apart (novox/hq
// ADR 0239 decision 4, cut to the merge's states by ADR 0539 §17): checking, rejected, ready, merging (some members
// merged), merged, held and closed. Nothing after the merge is derived here: whether a merged change reached the machines is
// the view's, from source.main, the wants and the haves.
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[Held] > 0 && g.Closed:
return "held", "a member is held: " + heldWords(members)
case count[Closed] > 0 && g.Closed:
return "closed", "a member was closed before the group merged"
case n > 0 && count[Merged] == n:
return "merged", "every member is merged"
case g.Closed && g.Check != nil && (g.Check.OutOfDate != "" || g.Check.Verdict == "" ||
!sameMembers(g.Check.Commits, commitsOf(members))):
// A member rebased or pushed once the group was merging, or its composed check out of date: it is composed
// again for these heads, and the group is checking until then, never held (novox/hq ADR 0323 §6, issue 406).
if g.Check.OutOfDate != "" {
return "checking", g.Check.OutOfDate
}
return "checking", "composed again for its members' new heads"
case g.Closed && g.Check != nil && g.Check.Verdict != "pass" && g.Check.Verdict != "warning":
// Composed again for new heads once it was merging, and those do not compose: its members wait (issue 902).
return "rejected", "composed together: " + g.Check.Summary
case g.Closed:
return "merging", fmt.Sprintf("%d of %d merged", count[Merged], n)
case count[Rejected] > 0:
return "rejected", "a member's own check failed"
case g.Check != nil && g.Check.OutOfDate != "":
// Out of date counts as outstanding (ADR 0239 §18, novox/hq ADR 0323 §7): checking, saying why.
return "checking", g.Check.OutOfDate
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)
}
// heldWords says the held members and why.
func heldWords(members []*Delivery) string {
var parts []string
for _, d := range members {
if d.State == Held {
parts = append(parts, d.ID+" ("+d.HeldWhy+")")
}
}
return strings.Join(parts, "; ")
}
// groupState is GroupState, but a group whose members all passed and compose waits for a review while any member the
// Approve would cover has none at its head: its why names the members (novox/hq ADR 0496 §2, issue 1803), and held
// says its mesh/delivery-group is pending, not a success. Under the lock.
func (h *Holder) groupState(g *Group, members []*Delivery) (state, why string, held bool) {
state, why = GroupState(g, members)
if state != "ready" || g.Closed {
return state, why, false
}
var waiting []string
for _, m := range members {
if m.MergedAs != "" || m.State.Final() || h.covering(m) != nil {
continue
}
if _, ok := reviewWords(m); !ok {
waiting = append(waiting, "at "+shortOf(m.Commit, 12)+" ("+pullSaid(m)+")")
}
}
if len(waiting) == 0 {
return state, why, false
}
return state, waitsForReview + strings.Join(waiting, " and ")[len("at "):], true
}
// groupStatusState is the forge status mesh/delivery-group takes for a group's state. A delivering group passed its composed check for these very heads (GroupState reads checking while it is out of
// date), so it is a success too: its members after the first merge only once mesh/delivery-group is one at their
// heads (novox/hq ADR 0382), and a pending one there would hold them for ever (novox/hq issue 902).
func groupStatusState(s string) string {
switch s {
case "ready", "merging", "merged":
return "success"
case "rejected":
return "failure"
case "held", "closed":
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
}