jschoubben
fb89990e2b
A proposed delivery whose asked check no queue entry holds stayed proposed for good (met in 1311, 1336, 1726, 1792, 1795, 1833). The sweep reads the controller's queue and asks again past max(10 minutes, the repository's last check time); recheck from proposed is taken on the same fact; never while the queue holds one.
2856 lines
108 KiB
Go
2856 lines
108 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"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
|
|
Controller Controller
|
|
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
|
|
// Seat says where today's module still stands; the gate holds the seat only when it stands nowhere (seat.go). Carry
|
|
// copies today's state over at the first holding, OnHold takes the reads of a holder just come.
|
|
Seat Occupancy
|
|
Carry func() (CarryReport, error)
|
|
OnHold func()
|
|
// 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
|
|
|
|
seatHeld atomic.Bool
|
|
seatMu sync.Mutex
|
|
seatWhy string
|
|
|
|
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
|
|
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
|
|
// lostRead is when the build queue was last read for a proposed delivery's asked check (issue 1274).
|
|
lostRead map[string]time.Time
|
|
// walkAsked is when a walk that carried a step still judging was last asked for by its id: once an hour.
|
|
walkAsked map[string]time.Time
|
|
refusals []string
|
|
// unchecked are the open pull requests' heads seen without a merge check (unchecked.go).
|
|
unchecked map[string]unchecked
|
|
// walks are the walks' and batches' last records (answers.go); orphans the waiting walks no live delivery
|
|
// holds, by when that was first seen; turning the walks whose turn came this look, for some deliveries.
|
|
walks map[string]seenWalk
|
|
orphans map[string]time.Time
|
|
turning map[string]*turnWant
|
|
// waitsWhy is, by walk, why the last look did not give a waiting walk its word: a delivery it carries not at
|
|
// its turn, or a stopped change it carries (novox/hq issue 529). Said by WalksWaiting.
|
|
waitsWhy map[string]string
|
|
// asking are the walks whose word is being asked right now, outside the lock: not asked twice meanwhile.
|
|
asking map[string]bool
|
|
// deliverErr is, by walk, the controller's last refusal of its word, until it is given (review of issue 753).
|
|
deliverErr map[string]string
|
|
// leaving are, by walk, the merges a stop gathered to leave out of it, asked once (novox/hq ADR 0299).
|
|
leaving map[string][]string
|
|
// leaveAsked is when a walk was last asked to leave out the stopped merges it carries (leaveStopped).
|
|
leaveAsked map[string]leaveAsk
|
|
// leaveRefused is, by walk, what the controller answered when it refused both to leave a stopped merge out and
|
|
// to stop the walk whole.
|
|
leaveRefused map[string]string
|
|
// coverAsked are the deliveries that failed or were stopped whose cover by a later finished walk waits for the
|
|
// forge's word on containment, worked out each pass (novox/hq issue 747): a walk carrying a merge stopped for one of
|
|
// them is held back meanwhile, never asked to leave it out before the answer.
|
|
coverAsked map[string]bool
|
|
// coverAskedSince is, by delivery, when the forge was first asked about its cover and has not answered: past
|
|
// coverAskBound the walk is no longer held back for it, and is asked to leave the merge out (review of issue 747).
|
|
coverAskedSince map[string]time.Time
|
|
// 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
|
|
// rereadWalks is whether the controller's walks are to be read once more: a merge passed the bound a walk opens
|
|
// within, and the walks were last read before it (unwalkedPast).
|
|
rereadWalks 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
|
|
}
|
|
|
|
// 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.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.unmatched = map[string]unmatched{}
|
|
h.waitsWhy = map[string]string{}
|
|
h.askTried = map[string]time.Time{}
|
|
h.lostRead = map[string]time.Time{}
|
|
h.walkAsked = map[string]time.Time{}
|
|
h.unchecked = map[string]unchecked{}
|
|
h.walks = map[string]seenWalk{}
|
|
h.orphans = map[string]time.Time{}
|
|
h.asking = map[string]bool{}
|
|
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
|
|
}
|
|
}
|
|
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 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.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)
|
|
}
|
|
|
|
func afterTheTrunk(s State) bool {
|
|
return s == Published || s == Delivering || s == Held || s == Delivered || s == Failed || s == Stopped ||
|
|
s == Superseded || s == Retired || s == Covered
|
|
}
|
|
|
|
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 Stopped, 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 != Stopped || 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 == Stopped {
|
|
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 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()
|
|
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["mesh/merge-gate"]; s != "" {
|
|
fresh.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"])}
|
|
}
|
|
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["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
|
|
}
|
|
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
|
|
}
|
|
// A walk already heard of that answers this merge: its own, or one that names it.
|
|
h.linkAnswering(d)
|
|
// A merge long past heard only now, from the forge's word — replayed from the bus's history when this owner
|
|
// first held the seat — is history: it reached the machines before this owner was there to deliver it, and
|
|
// holding it for a person asks a person about the past (novox/hq issue 313).
|
|
if d.FromForge && d.History == "" && isHistory(d) == nil {
|
|
d.History = fmt.Sprintf("merged %s, %s before this owner heard of it from the forge's word: history, "+
|
|
"never to be walked", d.MergedAt.UTC().Format(time.RFC3339), now.Sub(d.MergedAt).Round(time.Minute))
|
|
}
|
|
ts = append(ts, settle(d, Facts{Now: now})...)
|
|
if d.State == Held && d.HeldWhy == "" {
|
|
d.HeldWhy, d.Unwalked = "merged without a passing check: only a person decides that it goes on", false
|
|
}
|
|
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.
|
|
//
|
|
// **And what its checks built for its merge goes** (novox/hq ADR 0350 §5): nothing will take it now. Asked of the
|
|
// controller once the deliveries are settled, outside the lock; a refusal is said, and the build seat's holders let
|
|
// go of it within the week anyway.
|
|
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()
|
|
if h.Controller == nil || p.Owner == "" || p.Repo == "" || p.Number <= 0 {
|
|
return
|
|
}
|
|
if err := h.Controller.ForgetChecks(p.Owner, p.Repo, p.Number); err != nil {
|
|
h.mu.Lock()
|
|
h.refuse(fmt.Errorf("%s#%d closed, and its check builds stay until the week's sweep: %w", p.repository(),
|
|
p.Number, err))
|
|
h.mu.Unlock()
|
|
}
|
|
}
|
|
|
|
// 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"`
|
|
On string `json:"on,omitempty"`
|
|
Report string `json:"report,omitempty"`
|
|
Gate *struct {
|
|
Verdict string `json:"verdict"`
|
|
Summary string `json:"summary"`
|
|
Modules []string `json:"modules,omitempty"`
|
|
Dependents []string `json:"dependents,omitempty"`
|
|
} `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"`
|
|
Trunk string `json:"trunk,omitempty"`
|
|
} `json:"members,omitempty"`
|
|
// Trunk is the trunk commit the head was merged into and judged as, Beside the other repositories' commits the
|
|
// check read, and Took how long it ran (novox/hq ADR 0323 §3).
|
|
Trunk string `json:"trunk,omitempty"`
|
|
Beside map[string]string `json:"beside,omitempty"`
|
|
Took string `json:"took,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
|
|
for _, m := range h.membersOf(g) {
|
|
h.mergeAgainNow(m) // the group's composed check answered: its members' merges are asked at once
|
|
}
|
|
// Each member's head and the trunk commit it was merged into (novox/hq ADR 0323 §6).
|
|
for _, m := range c.Members {
|
|
for id := range g.Check.Commits {
|
|
if d := h.deliveries[id]; d != nil && strings.EqualFold(d.Repository, m.Owner+"/"+m.Repo) &&
|
|
sameCommit(d.Commit, m.Commit) && m.Trunk != "" {
|
|
if g.Check.Trunks == nil {
|
|
g.Check.Trunks = map[string]string{}
|
|
}
|
|
g.Check.Trunks[id] = m.Trunk
|
|
}
|
|
}
|
|
}
|
|
// A composed check that was on its way when a trunk moved lands judged against the old trunk: out of date as it lands
|
|
// (novox/hq ADR 0403 §6, issue 1047), and asked again once a member is judged again.
|
|
if g.Check.OutOfDate == "" {
|
|
for id, judged := range g.Check.Trunks {
|
|
if d := h.deliveries[id]; d != nil && d.MergedAs == "" && !d.State.Final() && d.TrunkNow != "" && !sameCommit(judged, d.TrunkNow) {
|
|
g.Check.OutOfDate = fmt.Sprintf("composed check out of date: %s was judged against trunk %s, the trunk is now %s",
|
|
id, shortOf(judged, 12), shortOf(d.TrunkNow, 12))
|
|
// From when the composed check was asked: a member judged since then, even before this verdict landed,
|
|
// counts as judged again.
|
|
g.Check.OutOfDateAt = g.Check.At
|
|
h.Logf("[mesh-delivery] group %s: %s", g.ID, g.Check.OutOfDate)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
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.Ready != nil && !d.Ready.OwnAskedAt.IsZero() {
|
|
// The own check asked for this head answered (issue 1792): one own ask is in flight at a time, so a status still
|
|
// pending or missing after its answer is asked again.
|
|
d.Ready.OwnAskedAt = time.Time{}
|
|
h.keep(d)
|
|
}
|
|
if d.State == Rejected && c.Trunk != "" && d.Check != nil && !sameCommit(d.Check.Trunk, c.Trunk) {
|
|
// Judged again against another trunk commit — the merge tool's ask of a verdict out of date, or a check asked
|
|
// elsewhere (novox/hq ADR 0323 §5, §7): taken as the recheck it is.
|
|
t, err := Apply(d, EvRecheck, true, Facts{Now: now, By: byOwner, Why: fmt.Sprintf("judged again against trunk %s "+
|
|
"(it was rejected against trunk %s)", shortOf(c.Trunk, 12), shortOf(d.Check.Trunk, 12))})
|
|
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, On: c.On, At: now, Gate: c.Verdict, Summary: c.Summary, Report: c.Report,
|
|
Head: shortOf(c.Commit, 12), Trunk: c.Trunk, Beside: c.Beside, Took: c.Took}
|
|
if c.Gate != nil {
|
|
v.Gate, v.Summary = c.Gate.Verdict, c.Gate.Summary
|
|
v.Modules, v.Dependents = c.Gate.Modules, c.Gate.Dependents
|
|
}
|
|
if c.RepoCheck != nil {
|
|
v.Repo, v.RepoSaid = c.RepoCheck.Verdict, c.RepoCheck.Summary
|
|
}
|
|
if v.Gate == "" {
|
|
v.Gate = "error"
|
|
}
|
|
d.Check, d.AskedAgain = v, nil
|
|
h.mergeAgainNow(d) // a verdict taken: the merge is asked at once, not a minute after a refusal
|
|
if c.Plan != nil {
|
|
d.Plan = c.Plan
|
|
}
|
|
if why := d.outOfDate(); why != "" {
|
|
// Kept and shown, and it does not count: the delivery stays proposed (ADR 0323 §7). If it passes, its check is
|
|
// asked against the trunk's head before the operator is (ADR 0421 §3, askFresh).
|
|
d.FreshWanted = d.Check.Passes()
|
|
h.Logf("[mesh-delivery] %s: its verdict %s is %s", d.ID, c.ID, why)
|
|
}
|
|
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) {
|
|
for _, d := range h.deliveries {
|
|
if d.Group != "" {
|
|
h.mergeAgainNow(d) // a walk moved: a group member's merge may be decided by it, asked at once
|
|
}
|
|
}
|
|
if w.Release != nil {
|
|
h.creditBacklog(w)
|
|
return
|
|
}
|
|
if old, seen := h.walks[w.ID]; !seen || w.Revision >= old.walk.Revision {
|
|
h.walks[w.ID] = seenWalk{walk: w, at: h.Now()}
|
|
}
|
|
if w.IsBatch() {
|
|
return // a batch is shown, never linked, adopted or let go (ADR 0276)
|
|
}
|
|
links := h.deliveriesWalking(w)
|
|
if len(links) == 0 {
|
|
if h.accounted(w) {
|
|
return // it answers only deliveries that ended or follow a newer walk
|
|
}
|
|
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
|
|
}
|
|
delete(h.unmatched, w.ID)
|
|
moved := map[string]bool{}
|
|
for _, a := range links {
|
|
d := a.d
|
|
if d.Walk != nil && d.Walk.ID == w.ID && w.Revision < d.Walk.Revision {
|
|
continue // an older record arriving late
|
|
}
|
|
h.linkWalk(d, w, a.carried)
|
|
h.settleAndKeep(d)
|
|
moved[d.ID] = true
|
|
}
|
|
// 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, moved)
|
|
}
|
|
}
|
|
// And how long each delivery it answers took, once its end is known (novox/hq ADR 0282): measured, never
|
|
// acted on.
|
|
h.timeWalk(w)
|
|
}
|
|
|
|
// linkWalk records a walk on its delivery: what it is at, the commit it carries for it, and the machine steps
|
|
// its record says — each moved only as the step table allows. A delivery that follows a new walk takes its
|
|
// steps as that walk's, a new journey; and one held because no walk answered it is lifted.
|
|
func (h *Holder) linkWalk(d *Delivery, w Walk, carried string) {
|
|
switching := d.Walk != nil && d.Walk.ID != w.ID
|
|
if switching {
|
|
h.Logf("[mesh-delivery] %s follows the walk %s now (it followed %s, %s)", d.ID, w.ID, d.Walk.ID, d.Walk.State)
|
|
d.LetGo, d.LetGoWhy = time.Time{}, "" // a new walk waits for a word of its own
|
|
}
|
|
d.Walk = w.Seen()
|
|
// Stopped without some of its merges (novox/hq ADR 0299): this delivery's merge was left out, maybe for
|
|
// containing a stopped one, or is walked again by the batch the record names.
|
|
if w.Delivery != nil && (w.Delivery.WalkedAgainBy != "" || len(w.Delivery.LeftOut) > 0) {
|
|
if o, out := w.leftOut(d.Repository, d.MergedAs); out {
|
|
d.Walk.Contains, d.Walk.Shares = o.Contains, o.Shares
|
|
} else {
|
|
d.Walk.WalkedAgainBy = w.Delivery.WalkedAgainBy
|
|
}
|
|
}
|
|
if carried != "" {
|
|
d.AnsweredBy = carried
|
|
}
|
|
d.Carrier = ""
|
|
if d.carriedLater() {
|
|
d.Carrier = fmt.Sprintf("merged as %s, carried by the walk %s", shortOf(d.AnsweredBy, 12), w.ID)
|
|
for _, o := range h.deliveries {
|
|
if o.ID != d.ID && strings.EqualFold(o.Repository, d.Repository) && sameCommit(o.MergedAs, d.AnsweredBy) {
|
|
d.Carrier = fmt.Sprintf("%s (merged as %s into %s, walk %s)", o.ID, shortOf(d.AnsweredBy, 8), orNone(d.Base), w.ID)
|
|
}
|
|
}
|
|
}
|
|
if d.State == Held && d.Unwalked && w.Open() {
|
|
d.HeldWhy = ""
|
|
d.Lifted = fmt.Sprintf("the walk %s, which answers its merge, opened", w.ID)
|
|
}
|
|
d.Unwalked = false // a walk answers it
|
|
for _, s := range StepsOf(w) {
|
|
if switching {
|
|
h.restep(d, s)
|
|
continue
|
|
}
|
|
h.step(d, s)
|
|
}
|
|
}
|
|
|
|
// restep puts a step of a walk a delivery newly follows, whatever the walk before said of that machine.
|
|
func (h *Holder) restep(d *Delivery, s Step) {
|
|
for i, old := range d.Steps {
|
|
if old.Machine == s.Machine && old.Module == s.Module {
|
|
d.Steps[i] = s
|
|
return
|
|
}
|
|
}
|
|
d.Steps = append(d.Steps, 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, nil)
|
|
}
|
|
if w.State == walkDone {
|
|
h.creditReleased(w)
|
|
}
|
|
}
|
|
|
|
// creditReleased passes the steps a finished backlog walk carried: its last record keeps no gate, only the
|
|
// machines it is done on, and a release ends done only when its gate passed on each of them (novox/hq issue
|
|
// 351). A step it failed on, and a step another walk carried, are left as they are.
|
|
func (h *Holder) creditReleased(w Walk) {
|
|
why := carriedBy + w.ID
|
|
for _, d := range h.deliveries {
|
|
moved := false
|
|
for _, s := range d.Steps {
|
|
if s.Why != why || !containsString(w.Release.Done, s.Machine) || s.State != StepSent && s.State != StepJudging {
|
|
continue
|
|
}
|
|
s.State, s.Since = StepPassed, h.Now()
|
|
if h.step(d, s) {
|
|
moved = true
|
|
}
|
|
}
|
|
if moved {
|
|
h.keep(d)
|
|
}
|
|
}
|
|
}
|
|
|
|
// carriedBy begins the why of a step another walk's send carried, followed by that walk's id.
|
|
const carriedBy = "carried by "
|
|
|
|
// askWalk is the ask of one walk's record by its id, at most once an hour; what it answers is taken as said.
|
|
// which says, for the log, why it is asked: what the walk is to the delivery that asks.
|
|
func (h *Holder) askWalk(id, which string, now time.Time) func() {
|
|
if asked, was := h.walkAsked[id]; was && now.Sub(asked) < time.Hour {
|
|
h.lookAt(asked.Add(time.Hour), "a walk's record asked of the controller by its id")
|
|
return nil
|
|
}
|
|
h.walkAsked[id] = now
|
|
h.lookAt(now.Add(time.Hour), "a walk's record asked of the controller by its id")
|
|
return func() {
|
|
_, walks, err := h.Controller.Walks(id)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] the walk %s, %s, cannot be read yet: %v", id, which, err)
|
|
h.walkAsked[id] = h.Now().Add(time.Minute - time.Hour) // asked again a minute on
|
|
return
|
|
}
|
|
for _, w := range walks {
|
|
if w.ID == id {
|
|
h.walkMoved(w)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// creditCarried gives the delivery that published each carried build its machine step.
|
|
func (h *Holder) creditCarried(g *WalkGate, walk string, except map[string]bool) {
|
|
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 except[d.ID] || 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: carriedBy + 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 look: groups, turns, adoption --------------------------------------------------------------------
|
|
|
|
// look takes every step the facts allow that needs another's word: a group's composed check, a
|
|
// delivery's turn, a walk with no delivery made one. The asking happens outside the lock.
|
|
func (h *Holder) look() {
|
|
for _, ask := range h.plan() {
|
|
ask()
|
|
}
|
|
for _, ask := range h.leaveStopped() {
|
|
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()
|
|
|
|
h.forgetWalks(now)
|
|
|
|
// 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 links := h.deliveriesWalking(u.walk); len(links) > 0 {
|
|
for _, a := range links {
|
|
h.linkWalk(a.d, u.walk, a.carried)
|
|
h.settleAndKeep(a.d)
|
|
}
|
|
delete(h.unmatched, id)
|
|
continue
|
|
}
|
|
if h.accounted(u.walk) {
|
|
delete(h.unmatched, id)
|
|
continue
|
|
}
|
|
if !u.walk.Open() {
|
|
if now.Sub(u.first) > 10*time.Minute {
|
|
delete(h.unmatched, id)
|
|
} else {
|
|
h.lookAt(u.first.Add(10*time.Minute), "a walk no merge was heard for, ended: forgotten after ten minutes")
|
|
}
|
|
continue
|
|
}
|
|
if u.walk.Created.After(h.started) && now.Sub(u.first) < matchGrace {
|
|
h.lookAt(u.first.Add(matchGrace), "a walk no merge was heard for: made a delivery after the grace")
|
|
continue
|
|
}
|
|
h.adopt(u.walk)
|
|
delete(h.unmatched, id)
|
|
}
|
|
|
|
// Plans not known for a delivery that has its diffset: asked at each look.
|
|
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)
|
|
h.again("a delivery's plan that could not be asked")
|
|
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: its check is asked at once (issue 1115), or the verdict that
|
|
// stands on its head at the forge is adopted.
|
|
if a := h.askChecks(now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
// Proposed, its check asked and held by no entry of the build queue past the bound: asked again (issue 1274).
|
|
if a := h.lostChecks(now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
|
|
// Merged with no walk yet: tied to the newest walk heard that answers it — by name, by its own commit, or by
|
|
// the forge's word that the walk's commit contains it, asked here (novox/hq issue 362).
|
|
for _, d := range h.deliveries {
|
|
if h.linkAnswering(d) {
|
|
h.settleAndKeep(d)
|
|
continue
|
|
}
|
|
asks = append(asks, h.containsAsks(d, now)...)
|
|
}
|
|
|
|
// Following a walk that was taken over: the walk that took it over, read as soon as it is heard of, asked for
|
|
// by its id when it is not.
|
|
for _, d := range h.deliveries {
|
|
if d.Walk == nil || d.Walk.TakenOverBy == "" {
|
|
continue
|
|
}
|
|
if d.State.Final() {
|
|
// Ended superseded by its walk before the walk that took it over was heard to answer (issue 1786): the
|
|
// state stays final, as a covered superseded member's does (issue 442), and the end of that walk is
|
|
// kept as DeliveredBy, which the group and the views read. Never logged, never changed otherwise.
|
|
if d.supersededByItsWalk() && d.DeliveredBy == "" {
|
|
if s, known := h.walks[d.Walk.TakenOverBy]; known {
|
|
if _, ok := walkAnswers(s.walk, d); ok && s.walk.State == walkDone {
|
|
d.DeliveredBy = fmt.Sprintf("the walk %s, which took its walk %s over, ended done", s.walk.ID, d.Walk.ID)
|
|
h.tellCovered(d, now)
|
|
}
|
|
} else if a := h.askWalk(d.Walk.TakenOverBy, "which took a delivery's walk over", now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
continue
|
|
}
|
|
if s, known := h.walks[d.Walk.TakenOverBy]; known {
|
|
if s.walk.IsBatch() {
|
|
// A batch is no walk yet (ADR 0276): the walk it is cut into answers the merge. Waited for, never
|
|
// read as a walk that does not answer it (issue 1786).
|
|
continue
|
|
}
|
|
if c, ok := walkAnswers(s.walk, d); ok {
|
|
h.linkWalk(d, s.walk, c)
|
|
} else {
|
|
// The walk that took its own over does not answer it: superseded, as before ADR 0276.
|
|
h.Logf("[mesh-delivery] %s: the walk %s, which took %s over, does not answer its merge", d.ID,
|
|
d.Walk.TakenOverBy, d.Walk.ID)
|
|
d.Walk.TakenOverBy = ""
|
|
}
|
|
h.settleAndKeep(d)
|
|
continue
|
|
}
|
|
if a := h.askWalk(d.Walk.TakenOverBy, "which took a delivery's walk over", now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
|
|
// Held, or merged with no walk opened for it: ended by what is already true of it, never asked of a person
|
|
// (novox/hq issue 319). A plan that moves nothing is delivered; one whose every module a later merge into the
|
|
// same trunk delivered is covered. Read again at each look, so a delivery read back after a restart, or held
|
|
// before this was known, ends by itself.
|
|
// A published delivery whose walk ended done before its word is read the same way (novox/hq issue 474): what
|
|
// that walk did not build, a later delivered merge may have; and when the cover test says none did, past the
|
|
// grace, it is stopped saying so — never left waiting for a start that cannot come.
|
|
// A delivery that failed or was stopped is read for a later finished walk of its repository that delivered its
|
|
// change after all: a walk that failed in an outage and was resumed, or the walk of a later merge that contains it
|
|
// (novox/hq issue 747). Read at each look, from what the walks and the kept deliveries say, so one stuck before this
|
|
// was built clears by itself after a restart.
|
|
h.coverAsked = map[string]bool{}
|
|
if h.coverAskedSince == nil {
|
|
h.coverAskedSince = map[string]time.Time{}
|
|
}
|
|
pending := map[string]bool{}
|
|
for _, d := range h.deliveries {
|
|
if d.State != Failed && d.State != Stopped || d.coveredStop() {
|
|
continue
|
|
}
|
|
by, ask := h.walkCover(d)
|
|
if len(ask) > 0 && by == "" {
|
|
pending[d.ID] = true
|
|
since, asked := h.coverAskedSince[d.ID]
|
|
if !asked {
|
|
since = now
|
|
h.coverAskedSince[d.ID] = now
|
|
}
|
|
// Held back only within the bound: a forge that keeps erroring never holds a walk for good.
|
|
h.coverAsked[d.ID] = now.Sub(since) < coverAskBound
|
|
if h.coverAsked[d.ID] {
|
|
h.lookAt(since.Add(coverAskBound), "a walk held back while the forge is asked what a later walk delivered")
|
|
}
|
|
}
|
|
if by != "" {
|
|
d.DeliveredBy = by
|
|
h.tellCovered(d, now)
|
|
continue
|
|
}
|
|
for _, in := range ask {
|
|
if a := h.askContains(d, in, now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
}
|
|
|
|
for id := range h.coverAskedSince {
|
|
if !pending[id] {
|
|
delete(h.coverAskedSince, id)
|
|
}
|
|
}
|
|
|
|
for _, d := range h.deliveries {
|
|
ended := d.endedUnstarted()
|
|
if d.coveredStop() && !d.CoverTold {
|
|
h.tellCovered(d, now) // covered before this holder told it, or before a restart (novox/hq issue 480)
|
|
continue
|
|
}
|
|
if strandedOpen(d) {
|
|
// Stopped and final, but still read for a later merge that delivered its modules: until one did,
|
|
// `stalled` keeps listing it (review of novox/hq issue 474).
|
|
var ask []string
|
|
if d.CoveredBy, ask = h.coverOf(d); d.CoveredBy != "" {
|
|
h.tellCovered(d, now)
|
|
}
|
|
for _, in := range ask {
|
|
if a := h.askContains(d, in, now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
continue
|
|
}
|
|
if d.State != Held && !h.unwalkedPast(d, now) && !ended {
|
|
continue
|
|
}
|
|
var ask []string
|
|
if d.CoveredBy == "" {
|
|
d.CoveredBy, ask = h.coverOf(d)
|
|
for _, in := range ask {
|
|
if a := h.askContains(d, in, now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
}
|
|
if ended {
|
|
if a := h.strand(d, ask, now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
if ts := settle(d, Facts{Now: now}); len(ts) > 0 {
|
|
h.moved(d, ts)
|
|
h.closeGroupOf(d)
|
|
}
|
|
}
|
|
|
|
// A step a walk carried and left judging: that walk's record is asked for by its id. A walk that ended
|
|
// while this holder was down may be older than the ended walks the controller lists back at its start,
|
|
// and its end is what passes the step (novox/hq issue 351).
|
|
for _, d := range h.deliveries {
|
|
for _, s := range d.Steps {
|
|
if s.State != StepSent && s.State != StepJudging || !strings.HasPrefix(s.Why, carriedBy) {
|
|
continue
|
|
}
|
|
if a := h.askWalk(strings.TrimPrefix(s.Why, carriedBy), "which carried a step still judging", 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 h.unwalkedPast(d, now) {
|
|
d.HeldWhy, d.Unwalked = h.unwalkedWhy(d), true
|
|
h.settleAndKeep(d)
|
|
}
|
|
}
|
|
|
|
// Groups: membership, composed check, and their word.
|
|
h.turning = map[string]*turnWant{}
|
|
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
|
|
}
|
|
h.turn(d, "its turn: a delivery in no group")
|
|
}
|
|
// A walk that answers the merge of a delivery whose stop lapsed: that delivery's turn (novox/hq issue 753).
|
|
h.lapsedTurns()
|
|
// The word, once per walk, when every live delivery it carries is at its turn.
|
|
asks = append(asks, h.words()...)
|
|
|
|
// Waiting walks no live delivery holds: listed by `stalled` (issue 362).
|
|
h.seeOrphans(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 && !strandedUncovered(d) {
|
|
if err := h.Store.DeleteDelivery(id); err == nil {
|
|
delete(h.deliveries, id)
|
|
delete(h.written, id)
|
|
}
|
|
}
|
|
}
|
|
return asks
|
|
}
|
|
|
|
// strandedAfter is how long after its walk ended unstarted a delivery that walk did not deliver waits for a later
|
|
// merge to cover it before it is stopped (novox/hq issue 474): within the bound of a published delivery, so it is
|
|
// stopped, saying why, before `stalled` would list it.
|
|
const strandedAfter = 20 * time.Minute
|
|
|
|
// strand says why a delivery whose walk ended done before its word is stopped, once that is so: the walk did not
|
|
// build every module it moves, the cover test asked the forge everything it had to and found no later delivered
|
|
// merge that carried them, and the walk ended strandedAfter ago. A walk record kept before it named its modules is
|
|
// asked for again (askWalk) rather than read as moving nothing.
|
|
func (h *Holder) strand(d *Delivery, pending []string, now time.Time) func() {
|
|
moved, known := d.walkMovedItsModules()
|
|
switch {
|
|
case !known && d.Plan != nil:
|
|
return h.askWalk(d.Walk.ID, "whose record, kept before it named the modules it built, is read again", now)
|
|
case !known, moved, d.CoveredBy != "", len(pending) > 0, d.Stranded != "", h.laterOnItsWay(d) != "":
|
|
return nil
|
|
}
|
|
ended := d.Walk.Updated
|
|
if ended.IsZero() {
|
|
ended = d.Since
|
|
}
|
|
if now.Sub(ended) < strandedAfter {
|
|
h.lookAt(ended.Add(strandedAfter), "a merge its walk did not build: stopped after the grace for a later merge to cover it")
|
|
return nil
|
|
}
|
|
note := ""
|
|
if d.Walk.Note != "" {
|
|
note = " (" + d.Walk.Note + ")"
|
|
}
|
|
d.Stranded = fmt.Sprintf("a merged change never reached the machines: its walk %s ended done before its word "+
|
|
"without building %s%s, and no later merge into %s that contains it delivered them (novox/hq issue 474)", d.Walk.ID,
|
|
strings.Join(d.notMovedBy(), ", "), note, orNone(d.Base))
|
|
return nil
|
|
}
|
|
|
|
// laterOnItsWay is a later merge of the same repository into the same trunk, not yet ended, whose plan moves a
|
|
// module this delivery's walk left unbuilt — its walk queued, waiting or running — or empty when there is none
|
|
// (review of novox/hq issue 474). A stop is final and nothing moves it to covered, so it waits for that merge:
|
|
// once delivered, the cover test reads it; if it ends otherwise, the stop follows.
|
|
func (h *Holder) laterOnItsWay(d *Delivery) string {
|
|
missing := d.notMovedBy()
|
|
for _, o := range h.deliveries {
|
|
if o.ID == d.ID || o.State.Final() || o.Plan == nil || o.MergedAs == "" || sameCommit(o.MergedAs, d.MergedAs) ||
|
|
!strings.EqualFold(o.Repository, d.Repository) || !strings.EqualFold(o.Base, d.Base) ||
|
|
!o.MergedAt.After(d.MergedAt) {
|
|
continue
|
|
}
|
|
for _, m := range missing {
|
|
if containsString(o.Plan.Moved, m) {
|
|
return o.ID
|
|
}
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// tellCovered tells the forge that a stop is covered since, once: its status says covered and by which merge, its view
|
|
// says so, and its note keeps the line on its head and its merge commit (novox/hq issue 480). No transition: the stop
|
|
// stays final, and the bus's transition event is not said again.
|
|
func (h *Holder) tellCovered(d *Delivery, now time.Time) {
|
|
d.CoverTold = true
|
|
line := oneLine(fmt.Sprintf("%s delivery %s: %s, %s", now.UTC().Format(time.RFC3339), d.ID, d.State, d.coveredStopSaid()))
|
|
d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.Commit, Line: line, Since: now})
|
|
if d.MergedAs != "" && !sameCommit(d.MergedAs, d.Commit) {
|
|
d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.MergedAs, Line: line, Since: now})
|
|
}
|
|
if 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.Logf("[mesh-delivery] %s: %s, %s", d.ID, d.State, d.coveredStopSaid())
|
|
h.keep(d)
|
|
}
|
|
|
|
// strandedUncovered is whether a delivery stopped because its ended walk delivered nothing is still uncovered: a
|
|
// merged change that never reached the machines, which `stalled` keeps listing until a later delivered merge
|
|
// carries its modules (review of novox/hq issue 474).
|
|
func strandedUncovered(d *Delivery) bool {
|
|
return strandedOpen(d) && !d.strandedEnded()
|
|
}
|
|
|
|
// strandedOpen is a stop for a walk that built none of its modules that no merge was found to cover: still read for
|
|
// a cover, and listed unless it is ended (novox/hq issue 1581).
|
|
func strandedOpen(d *Delivery) bool {
|
|
return d.State == Stopped && d.Stranded != "" && d.CoveredBy == ""
|
|
}
|
|
|
|
// walkCover says which later finished walk delivered the change of a delivery that failed or was stopped, or
|
|
// nothing, and which walk commits the forge must still be asked about (novox/hq issue 747). A walk covers it when it
|
|
// is of its repository and trunk, done, made after its merge, its commit containing its merge — the merge itself, or
|
|
// a later one the forge says contains it — and it built every module the delivery's plan moves, each it did not
|
|
// build being one the delivery's own walk said unchanged (issue 611). This is the containment "one walk answers every
|
|
// merge it contains" uses (issue 362, ADR 0276), read after the fact: a failed walk resumed to done answers its own
|
|
// merge, and a later merge's walk answers every merge it contains. A walk that only said a module unchanged for its
|
|
// own merges says nothing of this one's, so it covers no module that way. Walks are read from the controller's records
|
|
// this owner holds and from every delivery's kept walk, so a walk long ended still counts after a restart.
|
|
func (h *Holder) walkCover(d *Delivery) (string, []string) {
|
|
if d.State != Failed && d.State != Stopped || d.MergedAs == "" || d.Plan == nil || len(d.Plan.Moved) == 0 {
|
|
return "", nil
|
|
}
|
|
var ask []string
|
|
for _, c := range h.doneWalksOf(d.Repository) {
|
|
w := c.walk
|
|
if !sameBranch(w.Branch, d.Base) || !w.Created.IsZero() && w.Created.Before(d.MergedAt.Add(-time.Minute)) {
|
|
continue
|
|
}
|
|
var missing []string
|
|
for _, m := range d.Plan.Moved {
|
|
unchanged := false
|
|
if d.Walk != nil {
|
|
_, unchanged = d.Walk.Unchanged[m]
|
|
}
|
|
if !containsString(w.Moved, m) && !unchanged {
|
|
missing = append(missing, m)
|
|
}
|
|
}
|
|
if len(missing) > 0 {
|
|
continue
|
|
}
|
|
if !sameCommit(c.commit, d.MergedAs) {
|
|
yes, known := d.ContainedIn[c.commit]
|
|
if !known && !containsString(ask, c.commit) {
|
|
ask = append(ask, c.commit)
|
|
}
|
|
if !yes {
|
|
continue
|
|
}
|
|
}
|
|
return fmt.Sprintf("%s (%s at %s into %s, done, contains its merge %s)", w.ID, d.Repository, shortOf(c.commit, 8),
|
|
orNone(d.Base), shortOf(d.MergedAs, 8)), nil
|
|
}
|
|
return "", ask
|
|
}
|
|
|
|
// doneWalk is a finished walk and the commit of its repository it carried.
|
|
type doneWalk struct {
|
|
walk *WalkSeen
|
|
commit string
|
|
}
|
|
|
|
// doneWalksOf are the finished walks of a repository this owner knows, oldest first: the controller's records it
|
|
// holds, and each delivery's kept walk with the commit of its repository that walk carried for it.
|
|
func (h *Holder) doneWalksOf(repository string) []doneWalk {
|
|
byID := map[string]doneWalk{}
|
|
for id, s := range h.walks {
|
|
w := s.walk
|
|
if w.Release != nil || w.IsBatch() || w.State != walkDone || !strings.EqualFold(w.Repository, repository) {
|
|
continue
|
|
}
|
|
byID[id] = doneWalk{walk: w.Seen(), commit: w.Commit}
|
|
}
|
|
for _, o := range h.deliveries {
|
|
if o.Walk == nil || o.Walk.State != walkDone || !strings.EqualFold(o.Repository, repository) {
|
|
continue
|
|
}
|
|
if _, known := byID[o.Walk.ID]; known {
|
|
continue
|
|
}
|
|
commit := o.AnsweredBy
|
|
if commit == "" && o.MergedAs != "" && sameCommit(o.Walk.Commit, o.MergedAs) {
|
|
commit = o.MergedAs
|
|
}
|
|
if commit == "" {
|
|
continue
|
|
}
|
|
byID[o.Walk.ID] = doneWalk{walk: o.Walk, commit: commit}
|
|
}
|
|
out := make([]doneWalk, 0, len(byID))
|
|
for _, c := range byID {
|
|
out = append(out, c)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool {
|
|
if !out[i].walk.Created.Equal(out[j].walk.Created) {
|
|
return out[i].walk.Created.Before(out[j].walk.Created)
|
|
}
|
|
return out[i].walk.ID < out[j].walk.ID
|
|
})
|
|
return out
|
|
}
|
|
|
|
// stopLapsed is whether a member its group stopped for another is stopped for nothing any more (see stopLapsed).
|
|
func (h *Holder) stopLapsed(d *Delivery) bool {
|
|
return stopLapsed(d, func(id string) *Delivery { return h.deliveries[id] })
|
|
}
|
|
|
|
// unwalkedPast is whether a merged delivery has gone without a walk past the bound one opens within, read
|
|
// again after it: what holds it for a person, unless something already ends it.
|
|
func (h *Holder) unwalkedPast(d *Delivery, now time.Time) bool {
|
|
if _, batched := h.batchOf(d); batched {
|
|
return false // its batch is assembling or queued: a walk is not owed yet (ADR 0276)
|
|
}
|
|
if d.State != Ready || d.MergedAs == "" || d.Walk != nil || d.MovesNothing() || d.HeldWhy != "" {
|
|
return false
|
|
}
|
|
bound := d.MergedAt.Add(unwalked)
|
|
if now.Sub(d.MergedAt) <= unwalked {
|
|
h.lookAt(bound, "a merge no walk opened for within ten minutes")
|
|
return false
|
|
}
|
|
if !h.reconciled.After(bound) {
|
|
// The walks were not read since the bound: read once now (React), the only read besides the one at the start.
|
|
h.rereadWalks = true
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
// coverOf says which later merges cover a merged delivery, or nothing, and which later merge commits the forge
|
|
// must still be asked about (novox/hq ADR 0270). A module is covered by a later delivery that carries it (see
|
|
// carries) and whose merge commit the forge said contains this one's merge. Merge order is the cheap filter;
|
|
// ancestry decides. The forge is asked only once every module has a candidate, once per pair, and its answer
|
|
// is kept on the delivery: an answer not known yet is "not yet covered".
|
|
func (h *Holder) coverOf(d *Delivery) (string, []string) {
|
|
if d.MergedAs == "" || d.MergedAt.IsZero() || d.Base == "" || d.Plan == nil || len(d.Plan.Moved) == 0 {
|
|
return "", nil
|
|
}
|
|
candidates := map[string][]*Delivery{}
|
|
for _, module := range d.Plan.Moved {
|
|
for _, c := range h.deliveries {
|
|
if carries(c, d, module) {
|
|
candidates[module] = append(candidates[module], c)
|
|
}
|
|
}
|
|
if len(candidates[module]) == 0 {
|
|
return "", nil // nothing later carried it: nothing to ask
|
|
}
|
|
}
|
|
carried := map[string][]string{}
|
|
var by []*Delivery
|
|
var ask []string
|
|
for _, module := range d.Plan.Moved {
|
|
var first *Delivery
|
|
for _, c := range candidates[module] {
|
|
yes, known := d.ContainedIn[c.MergedAs]
|
|
if !known && !containsString(ask, c.MergedAs) {
|
|
ask = append(ask, c.MergedAs)
|
|
}
|
|
if yes && (first == nil || c.MergedAt.Before(first.MergedAt) || c.MergedAt.Equal(first.MergedAt) && c.ID < first.ID) {
|
|
first = c
|
|
}
|
|
}
|
|
if first == nil {
|
|
return "", ask
|
|
}
|
|
if carried[first.ID] == nil {
|
|
by = append(by, first)
|
|
}
|
|
carried[first.ID] = append(carried[first.ID], module)
|
|
}
|
|
sort.Slice(by, func(i, j int) bool { return by[i].MergedAt.Before(by[j].MergedAt) })
|
|
var parts []string
|
|
for _, c := range by {
|
|
mods := carried[c.ID]
|
|
sort.Strings(mods)
|
|
parts = append(parts, fmt.Sprintf("%s (merged as %s into %s, walk %s): %s", c.ID, shortOf(c.MergedAs, 8), d.Base,
|
|
c.Walk.ID, strings.Join(mods, ", ")))
|
|
}
|
|
return strings.Join(parts, "; "), nil
|
|
}
|
|
|
|
// carries is whether c delivered a module after d: of the same repository, merged later into the same trunk,
|
|
// delivered on its own passing check — a merge held without one is covered only by a later merge that passed
|
|
// its own — by its own walk, of its merge commit on that trunk, which sent the module to every machine running
|
|
// it. Its machine steps say so in one of two ways: the walk's own record, which sends to every machine at once
|
|
// or to the rest after its first machines passed; or, for a build the controller held back and a person
|
|
// released, one step per machine credited by the backlog walk that carried it — passed on every machine d's
|
|
// plan sends the module to (novox/hq issue 351). Whether c's merge commit contains d's is the forge's to say
|
|
// (coverOf).
|
|
func carries(c, d *Delivery, module string) bool {
|
|
if c.ID == d.ID || c.State != Delivered || !c.Check.Passes() || !strings.EqualFold(c.Repository, d.Repository) ||
|
|
!strings.EqualFold(c.Base, d.Base) || c.MergedAs == "" || sameCommit(c.MergedAs, d.MergedAs) ||
|
|
!c.MergedAt.After(d.MergedAt) || c.Walk == nil || c.Walk.State != walkDone || !sameCommit(c.Walk.Commit, c.MergedAs) ||
|
|
c.Walk.Branch != "" && !strings.EqualFold(c.Walk.Branch, d.Base) {
|
|
return false
|
|
}
|
|
everywhere := false
|
|
passed := map[string]bool{}
|
|
for _, s := range c.Steps {
|
|
if s.Module != module {
|
|
continue
|
|
}
|
|
switch {
|
|
case s.State == StepFailed || s.State == StepRolledBack:
|
|
return false
|
|
case s.State == StepSent && (s.Machine == stepEveryMachine || s.Machine == stepTheRest):
|
|
everywhere = true
|
|
case s.State == StepPassed:
|
|
passed[s.Machine] = true
|
|
}
|
|
}
|
|
if everywhere {
|
|
return true
|
|
}
|
|
machines := d.Plan.receiving(module)
|
|
if len(machines) == 0 {
|
|
return false
|
|
}
|
|
for _, m := range machines {
|
|
if !passed[m] {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// askContains is the ask of the forge whether a later merge commit contains a delivery's merge: at most once a
|
|
// minute while the forge cannot say, and never again once it said — its answer is kept on the delivery.
|
|
func (h *Holder) askContains(d *Delivery, in string, now time.Time) func() {
|
|
key := d.ID + " in " + in
|
|
if tried, was := h.askTried[key]; was && now.Sub(tried) < time.Minute {
|
|
h.lookAt(tried.Add(time.Minute), "the forge asked whether a later merge contains this one")
|
|
return nil
|
|
}
|
|
h.askTried[key] = now
|
|
h.lookAt(now.Add(time.Minute), "the forge asked whether a later merge contains this one")
|
|
id, owner, repo, merged := d.ID, d.Owner(), d.Repo(), d.MergedAs
|
|
return func() {
|
|
yes, err := h.Forge.Contains(owner, repo, merged, in)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] %s: whether %s contains its merge cannot be read yet (%v): not yet covered", id,
|
|
shortOf(in, 12), err)
|
|
return
|
|
}
|
|
delete(h.askTried, key)
|
|
d := h.deliveries[id]
|
|
if d == nil {
|
|
return
|
|
}
|
|
if d.ContainedIn == nil {
|
|
d.ContainedIn = map[string]bool{}
|
|
}
|
|
d.ContainedIn[in] = yes
|
|
h.keep(d)
|
|
}
|
|
}
|
|
|
|
// heldForGroup is whether a member is held only because its group's composed check had not passed.
|
|
func heldForGroup(d *Delivery, g *Group) bool {
|
|
return d.State == Held && strings.HasPrefix(d.HeldWhy, fmt.Sprintf("its group %s's composed check", g.ID))
|
|
}
|
|
|
|
// standingVerdict is the passing verdict an open pull request's head holds at the forge, when every merge check and
|
|
// every context its base requires stands as success or warning; nil otherwise, a missing context included (issue 1115).
|
|
func standingVerdict(p OpenPull, now time.Time) *Verdict {
|
|
if p.Checks == nil {
|
|
return nil
|
|
}
|
|
contexts := slices.Clone(mergeChecks)
|
|
for _, c := range p.Required {
|
|
if !slices.Contains(contexts, c) {
|
|
contexts = append(contexts, c)
|
|
}
|
|
}
|
|
for _, c := range contexts {
|
|
if st := p.Checks[c].State; st != "success" && st != "warning" {
|
|
return nil
|
|
}
|
|
}
|
|
return &Verdict{At: now, Gate: verdictOfState(p.Checks["mesh/merge-gate"].State),
|
|
Summary: "the head's mesh/merge-gate, as the forge holds it", Repo: verdictOfState(p.Checks["mesh/repo-check"].State)}
|
|
}
|
|
|
|
// askedCheck is whether a proposed delivery's own check stands asked: a kept ask with no id is none.
|
|
func askedCheck(d *Delivery) bool { return d.CheckAsk != nil && d.CheckAsk.ID != "" }
|
|
|
|
// askChecks are the asks of proposed deliveries with no verdict, at each look, at once (issue 1115): the forge's open
|
|
// pull requests are read once, a passing verdict standing on a head is adopted (the delivery goes on to ready with no
|
|
// new check), and a head with no check asked is asked its check. Under the lock; the asks run outside it.
|
|
func (h *Holder) askChecks(now time.Time) func() {
|
|
type want struct {
|
|
id, owner, repo, head string
|
|
number int
|
|
member Member
|
|
ask bool
|
|
why string
|
|
}
|
|
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, member: MemberOf(d)}
|
|
w.owner, w.repo, _ = strings.Cut(d.Repository, "/")
|
|
if !askedCheck(d) {
|
|
if tried, was := h.askTried[d.ID]; was && now.Sub(tried) < time.Minute {
|
|
h.lookAt(tried.Add(time.Minute), "a proposed delivery's check that could not be asked")
|
|
} else {
|
|
w.ask = true
|
|
h.askTried[d.ID] = now
|
|
h.lookAt(now.Add(time.Minute), "a proposed delivery's check that could not be asked")
|
|
w.why = fmt.Sprintf("proposed for %s with no verdict and no check asked", now.Sub(d.Since).Round(time.Second))
|
|
}
|
|
}
|
|
wants = append(wants, w)
|
|
}
|
|
if len(wants) == 0 {
|
|
return nil
|
|
}
|
|
return func() {
|
|
open, err := h.Forge.OpenPulls()
|
|
for _, w := range wants {
|
|
var standing *Verdict
|
|
if err == nil {
|
|
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) {
|
|
standing = standingVerdict(p, h.Now())
|
|
}
|
|
}
|
|
}
|
|
if standing != nil {
|
|
h.mu.Lock()
|
|
if d := h.deliveries[w.id]; d != nil && d.State == Proposed && d.Check == nil {
|
|
delete(h.askTried, w.id)
|
|
d.Check = standing
|
|
h.Logf("[mesh-delivery] %s: a passing verdict already stands on its head at the forge; adopted, no check asked", w.id)
|
|
h.settleAndKeep(d)
|
|
}
|
|
h.mu.Unlock()
|
|
continue
|
|
}
|
|
if w.ask {
|
|
h.askOwnCheck(w.id, w.member, w.why)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// askOwnCheck asks the controller a proposed delivery's own check and keeps the ask.
|
|
func (h *Holder) askOwnCheck(id string, m Member, why string) {
|
|
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
|
|
}
|
|
if asked == "" {
|
|
h.Logf("[mesh-delivery] ERROR: %s's check was asked and the controller's answer held no check id: it is not taken as asked, "+
|
|
"and asked again in a minute at most", id)
|
|
return
|
|
}
|
|
delete(h.askTried, id)
|
|
d := h.deliveries[id]
|
|
if d == nil || d.State != Proposed || d.Check != nil || askedCheck(d) {
|
|
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 {
|
|
if old.State.Final() {
|
|
// Never revived: an ended delivery is not the walk's; the walk, held by nothing, is listed by
|
|
// `stalled` (issue 362).
|
|
h.Logf("[mesh-delivery] the walk %s is at %s, whose delivery has ended (%s): not adopted by it", w.ID,
|
|
shortOf(w.Commit, 12), old.State)
|
|
return
|
|
}
|
|
h.linkWalk(old, w, w.Commit)
|
|
h.settleAndKeep(old)
|
|
return
|
|
}
|
|
h.linkWalk(d, w, w.Commit)
|
|
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 says a published delivery's waiting walk may start as far as that delivery goes; nil when it is not its
|
|
// turn. The word itself is asked once per walk by words, when every live delivery the walk carries is at its turn
|
|
// (ADR 0276 decision 5). What it answers is only whether this delivery's turn came.
|
|
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
|
|
}
|
|
if d.Walk.State != walkBuilding && d.Walk.State != walkRolling {
|
|
return nil // a walk superseded, done or failed is no walk to start
|
|
}
|
|
if h.turning == nil {
|
|
h.turning = map[string]*turnWant{}
|
|
}
|
|
want := h.turning[d.Walk.ID]
|
|
if want == nil {
|
|
want = &turnWant{}
|
|
h.turning[d.Walk.ID] = want
|
|
}
|
|
if !containsString(want.ids, d.ID) {
|
|
want.ids = append(want.ids, d.ID)
|
|
want.whys = append(want.whys, d.ID+": "+why)
|
|
}
|
|
return func() {}
|
|
}
|
|
|
|
// 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 composed check, once every member is ready. Asked again for new heads, even once the
|
|
// group is closed: a member rebased or pushed is a new head, and the group waits for its composed check rather
|
|
// than being held for a person (novox/hq ADR 0323 §6, issue 406). One out of date is asked again once a member
|
|
// was judged again after it went out of date: a recheck, or the merge tool's ask.
|
|
allReady := true
|
|
for _, d := range members {
|
|
// In a closed group a member on the trunk counts — merged and on its way, or already delivered (reached).
|
|
if d.State != Ready && !(g.Closed && d.MergedAs != "" && (!d.State.Final() || d.State.Reached())) {
|
|
allReady = false
|
|
}
|
|
}
|
|
again := g.Check != nil && (!sameMembers(g.Check.Commits, commits) || g.Check.OutOfDate != "" && rejudgedSince(members, g.Check.OutOfDateAt))
|
|
// A closed group's member whose merge the tool refused, after the composed check answered, is judged again at once
|
|
// (novox/hq ADR 0418 §4, issue 1082): the refusal is a sibling's merge having moved what the verdict stands on, and
|
|
// nothing else asks the composed check again. Not while a check is on its way (its verdict is empty then).
|
|
// **A refusal that names the member's OWN pending status** (mesh/repo-check, with the gate not pending: the composed
|
|
// check set it) is not that: the composed check sets no repository check, so asking it again changes nothing. The
|
|
// member's own check is asked, once for the head (issue 1049); its verdict sets the status.
|
|
if !again && g.Closed && g.Check != nil && g.Check.Verdict != "" {
|
|
for _, d := range members {
|
|
if !refusedAfter(d, g.Check.At) {
|
|
continue
|
|
}
|
|
if ownStatusPending(d.Ready.Answer) {
|
|
if a := h.askOwnAfterRefusal(d); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
continue
|
|
}
|
|
again = true
|
|
h.Logf("[mesh-delivery] group %s: the merge of %s was refused (%s), so its composed check is asked again", g.ID,
|
|
d.ID, d.Ready.Answer)
|
|
break
|
|
}
|
|
}
|
|
if allReady && (g.Check == nil && !g.Closed || again) {
|
|
ms := membersOfAsk(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)
|
|
h.again("a group's composed check that could not be asked")
|
|
return
|
|
}
|
|
g.Check = &GroupCheck{Asked: asked, At: h.Now(), Commits: commits}
|
|
g.Updated = h.Now()
|
|
h.keepGroup(g)
|
|
})
|
|
}
|
|
// A superseded member of a closed group: whether a later delivered walk on its trunk carried every module it
|
|
// moves, ADR 0270's covered test, kept on it as CoveredBy. Its own state stays superseded — a final state never
|
|
// moves — and its group counts it as reached (novox/hq issue 442).
|
|
if g.Closed {
|
|
for _, d := range members {
|
|
if d.State != Superseded || d.CoveredBy != "" || d.DeliveredBy != "" {
|
|
continue
|
|
}
|
|
var ask []string
|
|
if d.CoveredBy, ask = h.coverOf(d); d.CoveredBy != "" {
|
|
h.keep(d)
|
|
}
|
|
for _, in := range ask {
|
|
if a := h.askContains(d, in, now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// 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 == "delivering" && !(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)
|
|
}
|
|
// Each member's mesh/merge-gate is the composed check's (ADR 0403 §1, §6): pending, saying why, while no composed
|
|
// verdict stands at these heads. The verdict itself sets it, success or failure, in the forge module.
|
|
if why := composedPending(g, members); why != "" {
|
|
if g.SaidGate == nil {
|
|
g.SaidGate = map[string]string{}
|
|
}
|
|
for _, d := range members {
|
|
if d.MergedAs != "" || d.State.Final() || g.SaidGate[d.ID] == why {
|
|
continue
|
|
}
|
|
g.SaidGate[d.ID] = why
|
|
g.Owed = append(g.Owed, Effect{Kind: EffectStatus, Commit: d.Commit, Context: "mesh/merge-gate", State: "pending",
|
|
Line: clip("group "+g.ID+": "+why, 140), Since: now, Last: d.Repository + "#" + fmt.Sprint(d.Number) + " " + d.HTMLURL})
|
|
g.Updated = now
|
|
h.keepGroup(g)
|
|
}
|
|
} else if len(g.SaidGate) > 0 {
|
|
g.SaidGate = nil
|
|
h.keepGroup(g)
|
|
}
|
|
// Its word, once, when every member is merged (ADR 0418 §2).
|
|
if g.Closed {
|
|
asks = append(asks, h.groupWord(g, members)...)
|
|
}
|
|
return asks
|
|
}
|
|
|
|
// composedPending is why a group member's gate stands pending: the composed check has given no verdict at the members'
|
|
// heads as they are, or its verdict is out of date; empty once one stands, which the forge module then sets as every
|
|
// member's gate (novox/hq ADR 0403 §6).
|
|
func composedPending(g *Group, members []*Delivery) string {
|
|
c := g.Check
|
|
switch {
|
|
case c == nil || c.Verdict == "":
|
|
return "its group's composed check has not given a verdict at the members' heads yet"
|
|
case !sameMembers(c.Commits, commitsOf(members)):
|
|
return "composed again for the group's new heads"
|
|
case c.OutOfDate != "":
|
|
return c.OutOfDate
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// membersOfAsk are a group's members as the controller is asked about them: by id. No order is asked for or kept; the
|
|
// plan's tiers take it from the module graph (novox/hq ADR 0418 §2, ADR 0419).
|
|
func membersOfAsk(members []*Delivery) []Member {
|
|
sorted := append([]*Delivery(nil), members...)
|
|
sort.Slice(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID })
|
|
out := make([]Member, 0, len(sorted))
|
|
for _, d := range sorted {
|
|
out = append(out, MemberOf(d))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// groupWord gives a closed group's deliveries their turn for the plan's word, once every member is merged (novox/hq
|
|
// ADR 0418 §2, issue 978): no member has a turn of its own, and none waits for another's delivery. The word itself is
|
|
// asked once per plan by words, when every live delivery the plan carries is at its turn. A member that failed or was
|
|
// stopped stops no other: a group's plan that fails fails as every plan does (ADR 0418 §5), and a stop is a person's.
|
|
func (h *Holder) groupWord(g *Group, members []*Delivery) []func() {
|
|
for _, d := range members {
|
|
if d.MergedAs == "" && !d.State.Final() {
|
|
return nil // a member not yet on the trunk: the plan waits for it, nothing is let go
|
|
}
|
|
}
|
|
// 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)) && g.Check.OutOfDate == ""
|
|
sorted := append([]*Delivery(nil), members...)
|
|
sort.Slice(sorted, func(i, j int) bool { return sorted[i].ID < sorted[j].ID })
|
|
var asks []func()
|
|
for _, d := range sorted {
|
|
switch {
|
|
case memberReached(d), h.stopLapsed(d):
|
|
continue
|
|
case d.State == Held:
|
|
// Held for a composed check that has since passed for the heads that merged: lifted, published again
|
|
// (novox/hq issue 316). Any other hold is a person's.
|
|
if composed && heldForGroup(d, g) && walkCanGoOn(d) {
|
|
d.HeldWhy = ""
|
|
d.Lifted = fmt.Sprintf("its group %s's composed check said %s for the heads that merged: %s", g.ID,
|
|
g.Check.Verdict, g.Check.Summary)
|
|
h.settleAndKeep(d)
|
|
if d.State == Published {
|
|
if a := h.turn(d, "its group "+g.ID+": every member is merged"); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
}
|
|
case d.State == Published:
|
|
// Asked and not answered is not "did not pass": it waits, published, for the answer (issue 316). Nor is a
|
|
// composed check of other heads, or one out of date: it runs again for these heads, and the member waits
|
|
// for it (novox/hq ADR 0323 §6, issue 406).
|
|
if !composed && d.HeldWhy == "" && g.Check != nil && (g.Check.Verdict == "" || g.Check.OutOfDate != "" ||
|
|
!sameMembers(g.Check.Commits, commitsOf(members))) {
|
|
continue
|
|
}
|
|
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)
|
|
d.Unwalked = false
|
|
h.settleAndKeep(d)
|
|
continue
|
|
}
|
|
if a := h.turn(d, "its group "+g.ID+": every member is merged"); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
}
|
|
return asks
|
|
}
|
|
|
|
// stopLeaving asks the controller to stop a walk or batch without merges (ADR 0299) and says what came of it. A
|
|
// refusal — an older controller takes no `without`, or the leave-out could not be made — falls back to stopping the
|
|
// whole walk: the stopped change is held back, and every delivery on it ends stopped with the reason (review of
|
|
// #221). Outside the lock.
|
|
func (h *Holder) stopLeaving(walk, why, by string, without []string) string {
|
|
h.mu.Lock()
|
|
h.askedToLeave(walk)
|
|
h.mu.Unlock()
|
|
a, err := h.Controller.StopWithout(walk, why, by, without)
|
|
if err == nil {
|
|
return fmt.Sprintf("the walk %s is stopped without %s; %s", walk, strings.Join(without, ", "), againWords(a))
|
|
}
|
|
refused := err
|
|
if err := h.Controller.Stop(walk, fmt.Sprintf("%s, and it could not be left out alone (%v)", why, refused), by); err != nil {
|
|
h.mu.Lock()
|
|
if h.leaveRefused == nil {
|
|
h.leaveRefused = map[string]string{}
|
|
}
|
|
h.leaveRefused[walk] = fmt.Sprintf("leaving it out (%v), and stopping the walk whole (%v)", refused, err)
|
|
h.mu.Unlock()
|
|
return fmt.Sprintf("the walk %s could not be stopped without %s (%v), nor whole (%v): `plans stop %s --why …` "+
|
|
"ends it by hand", walk, strings.Join(without, ", "), refused, err, walk)
|
|
}
|
|
return fmt.Sprintf("the walk %s could not be stopped without %s (%v), so it was stopped whole: every delivery on it "+
|
|
"ends stopped", walk, strings.Join(without, ", "), refused)
|
|
}
|
|
|
|
// carriesStopped are the merges a walk or batch carries whose deliveries were stopped, as `<repository>@<commit>`,
|
|
// and why the first was: a record that names a stopped delivery's merge and has not left it out would deliver a
|
|
// stopped change (review of #221). Under the lock.
|
|
func (h *Holder) carriesStopped(walk string) ([]string, string) {
|
|
s, known := h.walks[walk]
|
|
if !known || !(s.walk.Open() || s.walk.IsBatch()) || s.walk.Delivery == nil || len(s.walk.Delivery.Merges) == 0 {
|
|
return nil, ""
|
|
}
|
|
var out []string
|
|
why := ""
|
|
ids := make([]string, 0, len(h.deliveries))
|
|
for id := range h.deliveries {
|
|
ids = append(ids, id)
|
|
}
|
|
slices.Sort(ids)
|
|
for _, id := range ids {
|
|
d := h.deliveries[id]
|
|
// Stopped by a person or by its group, the acts that hold a change back: never one ended by what its walk
|
|
// did (the observed rows, which name nobody), such as a merge stranded after its walk (issue 474).
|
|
if d.State != Stopped || d.MergedAs == "" || len(d.Transitions) == 0 {
|
|
continue
|
|
}
|
|
if last := d.Transitions[len(d.Transitions)-1]; last.Event != EvStop || last.By == "" {
|
|
continue
|
|
}
|
|
// A stop whose change a later walk delivered since, or one its group made for a member delivered since,
|
|
// holds nothing back: the group stops a walk only for a member whose change is still undelivered (novox/hq
|
|
// issue 747).
|
|
if d.coveredStop() || h.stopLapsed(d) {
|
|
continue
|
|
}
|
|
// By the merges it names alone: its own commit may be another's merge it no longer answers.
|
|
if !slices.ContainsFunc(s.walk.Delivery.Merges, func(m WalkMerge) bool {
|
|
return strings.EqualFold(m.Repository, d.Repository) && sameCommit(m.Commit, d.MergedAs)
|
|
}) {
|
|
continue
|
|
}
|
|
if _, out := s.walk.leftOut(d.Repository, d.MergedAs); out {
|
|
continue
|
|
}
|
|
out = append(out, d.Repository+"@"+d.MergedAs)
|
|
if why == "" {
|
|
why = fmt.Sprintf("%s was stopped: %s", d.ID, lastWhyOf(d))
|
|
}
|
|
}
|
|
return out, why
|
|
}
|
|
|
|
// coverAwaited is whether a stopped merge a walk carries, given as `<repository>@<commit>`, waits for the forge's
|
|
// word on whether a later finished walk delivered it or the member its group stopped it for (novox/hq issue 747).
|
|
// It says since when the forge was first asked, the earliest of them. Held only within coverAskBound of the first
|
|
// ask: past it the forge's silence or error lets the walk be asked to leave the merge out. Under the lock.
|
|
func (h *Holder) coverAwaited(merges []string) (time.Time, bool) {
|
|
var since time.Time
|
|
awaited := false
|
|
for _, d := range h.deliveries {
|
|
if d.State != Stopped || d.MergedAs == "" || !containsString(merges, d.Repository+"@"+d.MergedAs) {
|
|
continue
|
|
}
|
|
for _, id := range []string{d.ID, d.stoppedFor()} {
|
|
if id == "" || !h.coverAsked[id] {
|
|
continue
|
|
}
|
|
if at := h.coverAskedSince[id]; !awaited || at.Before(since) {
|
|
since = at
|
|
}
|
|
awaited = true
|
|
}
|
|
}
|
|
return since, awaited
|
|
}
|
|
|
|
// coverHeld says why a walk is held back while the forge is asked whether a later finished walk delivered a stopped
|
|
// merge it carries (novox/hq issue 747), or nothing when it is not. Said by `waits`, `stalled` and `stalled walks`, so
|
|
// a forge that cannot answer is loud. Under the lock.
|
|
func (h *Holder) coverHeld(walk string) string {
|
|
without, _ := h.carriesStopped(walk)
|
|
if len(without) == 0 {
|
|
return ""
|
|
}
|
|
since, awaited := h.coverAwaited(without)
|
|
if !awaited {
|
|
return ""
|
|
}
|
|
return fmt.Sprintf("it carries %s, whose delivery was stopped, and the forge is asked since %s whether a later "+
|
|
"finished walk delivered it; it is asked to leave the merge out if no answer comes by %s (novox/hq issue 747)",
|
|
strings.Join(without, ", "), since.Local().Format("15:04:05"), since.Add(coverAskBound).Local().Format("15:04:05"))
|
|
}
|
|
|
|
// coverAskBound is how long a walk carrying a stopped merge is held back while the forge is asked whether a later
|
|
// finished walk delivered it (novox/hq issue 747); past it, the walk is asked to leave the merge out.
|
|
const coverAskBound = 10 * time.Minute
|
|
|
|
// lastWhyOf is the reason of a delivery's last transition.
|
|
func lastWhyOf(d *Delivery) string {
|
|
if len(d.Transitions) == 0 {
|
|
return ""
|
|
}
|
|
return d.Transitions[len(d.Transitions)-1].Why
|
|
}
|
|
|
|
// leaveAsk is when a walk was last asked to leave merges out, and at which revision of its record.
|
|
type leaveAsk struct {
|
|
at time.Time
|
|
revision int64
|
|
}
|
|
|
|
// askedToLeave notes an ask to leave merges out of a walk, at its record's revision: under the lock.
|
|
func (h *Holder) askedToLeave(walk string) {
|
|
if h.leaveAsked == nil {
|
|
h.leaveAsked = map[string]leaveAsk{}
|
|
}
|
|
h.leaveAsked[walk] = leaveAsk{at: h.Now(), revision: h.walks[walk].walk.Revision}
|
|
h.lookAt(h.Now().Add(5*time.Minute), "a walk asked to leave a stopped merge out, asked again if it went nowhere")
|
|
}
|
|
|
|
// leaveStopped asks, for every open walk or batch that carries a stopped delivery's merge, that the merge be left
|
|
// out (ADR 0299, review of #221): a merge walked again and then stopped, or stopped while its batch assembled, never
|
|
// ships. Asked once per record of it while it is carried; such a walk is never given its word (words).
|
|
func (h *Holder) leaveStopped() []func() {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
now := h.Now()
|
|
var asks []func()
|
|
held := map[string]string{}
|
|
for id := range h.walks {
|
|
if without, _ := h.carriesStopped(id); len(without) > 0 {
|
|
said := fmt.Sprintf("its walk %s is not let go: it carries %s, whose delivery was stopped, and is asked to "+
|
|
"leave it out (novox/hq ADR 0299)", id, strings.Join(without, ", "))
|
|
if hold := h.coverHeld(id); hold != "" {
|
|
said = "its walk " + id + " is held back: " + hold
|
|
}
|
|
if r := h.leaveRefused[id]; r != "" {
|
|
said += "; the controller refused: " + r
|
|
}
|
|
held[id] = said
|
|
}
|
|
}
|
|
// What the waiting deliveries say: `waits`, and `stalled` past their bound.
|
|
for _, d := range h.deliveries {
|
|
said := ""
|
|
if d.Walk != nil && !d.State.Final() {
|
|
said = held[d.Walk.ID]
|
|
}
|
|
if d.HeldBack != said {
|
|
d.HeldBack = said
|
|
h.keep(d)
|
|
}
|
|
}
|
|
for id, seen := range h.walks {
|
|
without, why := h.carriesStopped(id)
|
|
if len(without) == 0 {
|
|
delete(h.leaveRefused, id)
|
|
continue
|
|
}
|
|
if _, awaited := h.coverAwaited(without); awaited {
|
|
continue // held back, not stopped, until the forge says whether a later walk delivered it (issue 747)
|
|
}
|
|
// Once per record of it: the record that answers an ask is newer. Asked again after five minutes, should
|
|
// the ask have gone nowhere.
|
|
if a, asked := h.leaveAsked[id]; asked && seen.walk.Revision <= a.revision && now.Sub(a.at) < 5*time.Minute {
|
|
h.lookAt(a.at.Add(5*time.Minute), "a walk asked to leave a stopped merge out, asked again if it went nowhere")
|
|
continue
|
|
}
|
|
walk := id
|
|
asks = append(asks, func() { h.Logf("[mesh-delivery] %s", h.stopLeaving(walk, why, byOwner, without)) })
|
|
}
|
|
return asks
|
|
}
|
|
|
|
// leave gathers a delivery's merge to leave out of a walk (ADR 0299), for a caller holding the lock; true for the
|
|
// first of that walk, whose ask asks for all.
|
|
func (h *Holder) leave(walk string, d *Delivery) bool {
|
|
if h.leaving == nil {
|
|
h.leaving = map[string][]string{}
|
|
}
|
|
first := len(h.leaving[walk]) == 0
|
|
if m := d.Repository + "@" + d.MergedAs; d.MergedAs != "" && !containsString(h.leaving[walk], m) {
|
|
h.leaving[walk] = append(h.leaving[walk], m)
|
|
}
|
|
return first
|
|
}
|
|
|
|
// leaving459 takes the merges gathered to leave out of a walk, and whether the walk carries anything else: another
|
|
// merge it names, or a delivery not ended that follows it. Taken under the lock, outside any.
|
|
func (h *Holder) leaving459(walk string) ([]string, bool) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
without := h.leaving[walk]
|
|
delete(h.leaving, walk)
|
|
return without, len(h.othersOn(walk, without)) > 0
|
|
}
|
|
|
|
// othersOn are what a walk carries besides the merges named: each delivery not ended that follows it, by id, and
|
|
// each merge it names that no such delivery holds, as `<repository>@<commit>`.
|
|
func (h *Holder) othersOn(walk string, without []string) []string {
|
|
out := map[string]bool{}
|
|
held := map[string]bool{}
|
|
for _, d := range h.deliveries {
|
|
if d.Walk == nil || d.Walk.ID != walk || d.MergedAs == "" {
|
|
continue
|
|
}
|
|
m := d.Repository + "@" + d.MergedAs
|
|
held[strings.ToLower(m)] = true
|
|
if !d.State.Final() && !containsString(without, m) {
|
|
out[d.ID] = true
|
|
}
|
|
}
|
|
if s, known := h.walks[walk]; known && s.walk.Delivery != nil {
|
|
for _, m := range s.walk.Delivery.Merges {
|
|
key := m.Repository + "@" + m.Commit
|
|
if !held[strings.ToLower(key)] && !slices.ContainsFunc(without, func(w string) bool { return strings.EqualFold(w, key) }) {
|
|
out[key] = true
|
|
}
|
|
}
|
|
}
|
|
keys := make([]string, 0, len(out))
|
|
for k := range out {
|
|
keys = append(keys, k)
|
|
}
|
|
slices.Sort(keys)
|
|
return keys
|
|
}
|
|
|
|
// againWords is what a stop that left merges out did, in a person's words: the walk that walks the others again,
|
|
// what it left out (the controller's word on what contains a stopped merge), what stays on the machines, and what
|
|
// will still deliver a stopped change (ADR 0299).
|
|
func againWords(a StopAnswer) string {
|
|
var left []string
|
|
for _, o := range a.LeftOut {
|
|
s := o.Repository + "@" + shortOf(o.Commit, 12)
|
|
switch {
|
|
case len(o.Shares) > 0:
|
|
s += " (moves " + strings.Join(o.Shares, ", ") + ", as " + shortOf(o.Contains, 12) + " does)"
|
|
case o.Contains != "":
|
|
s += " (contains " + shortOf(o.Contains, 12) + ")"
|
|
}
|
|
left = append(left, s)
|
|
}
|
|
var out string
|
|
switch {
|
|
case a.WalkedAgainBy != "":
|
|
out = fmt.Sprintf("%s walks its other merges again: %s; left out: %s", a.WalkedAgainBy,
|
|
strings.Join(a.WalkedAgain, ", "), orNone(strings.Join(left, ", ")))
|
|
case len(left) > 0:
|
|
out = "left out: " + strings.Join(left, ", ") + "; nothing else to walk"
|
|
default:
|
|
out = "nothing else to walk"
|
|
}
|
|
if len(a.Withheld) > 0 {
|
|
out += "; its walk withholds " + strings.Join(a.Withheld, ", ")
|
|
}
|
|
if len(a.Kept) > 0 {
|
|
var kept []string
|
|
for _, k := range a.Kept {
|
|
kept = append(kept, k.Module+" on "+strings.Join(k.Machines, ", "))
|
|
}
|
|
out += "; what the stopped walk already sent stays on the machines: " + strings.Join(kept, "; ")
|
|
}
|
|
if len(a.StillCarriedBy) > 0 {
|
|
out += "; still to deliver the stopped change when walked, a revert's to hold back: " +
|
|
strings.Join(a.StillCarriedBy, ", ")
|
|
}
|
|
return out
|
|
}
|
|
|
|
// GroupState is a group's state, derived from its members and its composed check — never set apart (novox/hq
|
|
// ADR 0239 decision 4, which lists checking, rejected, ready, delivering, delivered, failed and stopped).
|
|
//
|
|
// It extends that list for a superseded member, which ADR 0239 does not name (novox/hq issue 442: five closed
|
|
// groups read "delivering, 1 of 2 delivered" for days with nothing in progress). A superseded member is an end.
|
|
// It counts as reached when a later delivered walk on its trunk carried every module it moves (CoveredBy, ADR
|
|
// 0270's covered test); a closed group whose members all ended is then delivered, and otherwise stopped, naming
|
|
// the member. A group with nothing in progress never reads delivering.
|
|
//
|
|
// groupWord agrees: a group's plan is let go once, when every member is merged (novox/hq ADR 0418 §2).
|
|
func GroupState(g *Group, members []*Delivery) (string, string) {
|
|
count := map[State]int{}
|
|
reached, ended := 0, 0
|
|
var uncovered, lapsed []string
|
|
byID := map[string]*Delivery{}
|
|
for _, d := range members {
|
|
byID[d.ID] = d
|
|
}
|
|
for _, d := range members {
|
|
if d.State.Final() {
|
|
ended++
|
|
}
|
|
switch {
|
|
case memberReached(d):
|
|
// Reached, whatever its final state: a failure or stop a later walk delivered since is no failure of the
|
|
// group (novox/hq issue 747).
|
|
reached++
|
|
continue
|
|
case stopLapsed(d, func(id string) *Delivery { return byID[id] }):
|
|
lapsed = append(lapsed, d.ID)
|
|
continue
|
|
case d.State == Superseded:
|
|
uncovered = append(uncovered, d.ID)
|
|
}
|
|
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 && reached == n:
|
|
return "delivered", "every member is delivered"
|
|
case g.Closed && len(lapsed) > 0 && reached+len(lapsed) == n:
|
|
// Stopped for a member delivered since: its own change waits for a later walk of its repository, which then
|
|
// covers it (novox/hq issue 747).
|
|
return "delivering", fmt.Sprintf("%d of %d delivered; %s, stopped for a member delivered since, waits for a "+
|
|
"later walk of its repository", reached, n, strings.Join(lapsed, ", "))
|
|
case g.Closed && n > 0 && ended == n:
|
|
return "stopped", supersededWhy + strings.Join(uncovered, ", ")
|
|
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 delivering, 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 delivering, and those do not compose: its members wait (issue 902).
|
|
return "rejected", "composed together: " + g.Check.Summary
|
|
case g.Closed:
|
|
return "delivering", fmt.Sprintf("%d of %d delivered", reached, 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)
|
|
}
|
|
|
|
// 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
|
|
}
|
|
|
|
// supersededWhy begins why a group whose members all ended is stopped. It is kept short: the forge's status line,
|
|
// "group <id> stopped: <why>", is clipped at 140 characters, and the member's id must fit in it whole.
|
|
const supersededWhy = "superseded, not carried by a later delivery: "
|
|
|
|
// 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", "delivering", "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
|
|
}
|
|
|
|
// refusedAfter is whether a member's merge, not yet merged, was refused by the merge tool in a try made at or after at.
|
|
func refusedAfter(d *Delivery, at time.Time) bool {
|
|
r := d.Ready
|
|
return r != nil && r.Merged == nil && d.MergedAs == "" && r.RefusedSince != nil && !r.LastTry.Before(at)
|
|
}
|
|
|
|
// ownStatusPending is whether the merge tool refused a merge because a status of the member's own merge check is
|
|
// pending or missing (issue 1276) while its gate is not: the forge's "not a success: mesh/repo-check (pending)" (gitea merge.ts, mergeableUnder).
|
|
// A refusal that also names mesh/merge-gate is the group's composed check's to answer, as is a verdict out of date.
|
|
func ownStatusPending(answer string) bool {
|
|
_, listed, ok := strings.Cut(answer, "not a success:")
|
|
return ok && (strings.Contains(listed, "mesh/repo-check (pending)") || strings.Contains(listed, "mesh/repo-check (missing)")) &&
|
|
!strings.Contains(listed, "mesh/merge-gate")
|
|
}
|
|
|
|
// askOwnAfterRefusal asks a member's own check once for its head, after the merge tool refused it for a pending status of
|
|
// its own. Under the lock; the ask is made outside it. Nil when it was asked already for this head.
|
|
func (h *Holder) askOwnAfterRefusal(d *Delivery) func() {
|
|
r := d.Ready
|
|
if !r.OwnAskedAt.IsZero() {
|
|
return nil
|
|
}
|
|
r.OwnAskedAt = h.Now()
|
|
h.keep(d)
|
|
id, m := d.ID, MemberOf(d)
|
|
return func() {
|
|
asked, err := h.Controller.Check("", []Member{m})
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
d := h.deliveries[id]
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] %s's own check cannot be asked yet: %v", id, err)
|
|
if d != nil && d.Ready != nil {
|
|
d.Ready.OwnAskedAt = time.Time{}
|
|
h.keep(d)
|
|
}
|
|
h.again("a member's own check that could not be asked")
|
|
return
|
|
}
|
|
h.Logf("[mesh-delivery] %s: the merge tool refused it for a pending status of its own, so its own check is asked as %s",
|
|
id, asked)
|
|
}
|
|
}
|
|
|
|
func rejudgedSince(members []*Delivery, at time.Time) bool {
|
|
for _, d := range members {
|
|
if d.Check != nil && d.Check.At.After(at) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|