jschoubben
66d22dd53e
mesh/merge-gate head 66d22dd53eff merged into trunk 7ed119995f69 — pass: builds mesh-delivery → novox; no bus step; every machine composes with the change …
mesh/repo-check head 66d22dd53eff merged into trunk 7ed119995f69 — pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-…
mesh/delivery delivered
mesh/delivery-group group feat/350-one-build-per-change delivered: every member is delivered
A check's builds for its merge have no taker once the pull request closes unmerged, and the delivery should say whether a module was reused from the check or built, and why (novox/hq ADR 0350 §4, §5).
2423 lines
89 KiB
Go
2423 lines
89 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
// The holder of the mesh-delivery seat (novox/hq ADR 0239, to-be 47): the one writer of deliveries and
|
|
// groups. Every input — a pull request's head, a verdict, a merge, a walk's record, a person's act — only
|
|
// changes a delivery's facts; settle then moves it through the table as far as its facts allow, so the
|
|
// order the inputs arrive in never decides where a delivery stands. A delivery is kept before anything
|
|
// about it is said, and what is owed outside — the event, the commit's note, the pull request's view and
|
|
// statuses — is retried until done.
|
|
|
|
// Holder is the seat's holder.
|
|
type Holder struct {
|
|
Store Store
|
|
Controller Controller
|
|
Forge Forge
|
|
Emit func(event string, body any) error
|
|
Now func() time.Time
|
|
Logf func(format string, a ...any)
|
|
|
|
mu sync.Mutex
|
|
deliveries map[string]*Delivery
|
|
groups map[string]*Group
|
|
dirty map[string]bool
|
|
// 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
|
|
// 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 tick, for some deliveries.
|
|
walks map[string]seenWalk
|
|
orphans map[string]time.Time
|
|
turning map[string]*turnWant
|
|
// waitsWhy is, by walk, why the last tick 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
|
|
// 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
|
|
// tried counts the owed effects Flush tried since this owner started, by how each ended (novox/hq issue 407).
|
|
tried owedTries
|
|
}
|
|
|
|
// 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.deliveries = map[string]*Delivery{}
|
|
h.groups = map[string]*Group{}
|
|
h.dirty = 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.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)
|
|
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)
|
|
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"`
|
|
}
|
|
|
|
func (p PullEvent) repository() string { return strings.ToLower(p.Owner + "/" + p.Repo) }
|
|
|
|
// trunkLike are branch names that never make a group: a pull request from them is nobody's feature.
|
|
var trunkLike = map[string]bool{"main": true, "master": true, "trunk": true, "": true}
|
|
|
|
// PullUpdated takes a pull request's new head: a delivery proposed, an older head of the same pull request
|
|
// superseded, and the delivery put in the group its branch name makes.
|
|
func (h *Holder) PullUpdated(p PullEvent) {
|
|
if p.HeadSHA == "" || p.Number == 0 {
|
|
h.Refuse(errors.New("a pull request announced with no head or no number; nothing proposed"))
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
id := IDOf(p.repository(), p.HeadSHA)
|
|
now := h.Now()
|
|
if d := h.deliveries[id]; d != nil {
|
|
// 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)
|
|
}
|
|
if d.State == Proposed && d.outOfDate() != "" && d.AskedAgain == nil {
|
|
d.AskedAgain = &now
|
|
h.keep(d)
|
|
}
|
|
return
|
|
}
|
|
replaced := ""
|
|
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
|
|
}
|
|
}
|
|
}
|
|
d := &Delivery{ID: id, Repository: p.repository(), Commit: p.HeadSHA, Number: p.Number, Title: p.Title,
|
|
Base: p.Base, Branch: p.Head, HTMLURL: p.HTMLURL, CloneURL: p.CloneURL, Paths: p.Paths,
|
|
PathsTruncated: p.PathsTruncated, Removed: p.Removed, ModuleDirs: p.ModuleDirs, ModuleDirsSaid: p.ModuleDirsSaid,
|
|
After: AfterIn(p.Body), Created: now, BaseSHA: p.BaseSHA}
|
|
t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: fmt.Sprintf("%s#%d's head announced", d.Repository, d.Number)})
|
|
if err != nil {
|
|
h.refuse(err)
|
|
return
|
|
}
|
|
ts := []Transition{t}
|
|
// An announcement heard again from the bus's history may carry its head's verdict already: the forge holds
|
|
// the gate's status on that very commit. A decided one is taken; a pending or errored one is asked again.
|
|
if v := decidedOnHead(p.HeadChecks, now); v != nil {
|
|
d.Check = v
|
|
ts = append(ts, settle(d, Facts{Now: now})...)
|
|
}
|
|
h.moved(d, ts)
|
|
if !h.takePlaceOf(replaced, d) {
|
|
h.joinGroup(d)
|
|
}
|
|
}
|
|
|
|
// 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
|
|
}
|
|
for _, g := range h.groups {
|
|
_, ordered := g.OrderOf[old]
|
|
if !g.Closed || !ordered && !containsString(g.Order, old) {
|
|
continue
|
|
}
|
|
if ordered {
|
|
delete(g.OrderOf, old)
|
|
g.OrderOf[d.ID] = d.Commit
|
|
}
|
|
if !containsString(g.Members, d.ID) {
|
|
g.Members = append(g.Members, d.ID)
|
|
sort.Strings(g.Members)
|
|
}
|
|
for i, id := range g.Order {
|
|
if id == old {
|
|
g.Order[i] = d.ID
|
|
}
|
|
}
|
|
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()
|
|
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
|
|
}
|
|
}
|
|
}
|
|
now := h.Now()
|
|
var ts []Transition
|
|
if d == nil {
|
|
head := p.HeadSHA
|
|
if head == "" {
|
|
head = p.MergeCommit
|
|
}
|
|
d = &Delivery{ID: IDOf(p.repository(), head), Repository: p.repository(), Commit: head, Number: p.Number,
|
|
Title: p.Title, Base: p.Base, Branch: p.Head, HTMLURL: p.HTMLURL, CloneURL: p.CloneURL, Paths: p.Paths,
|
|
PathsTruncated: p.PathsTruncated, Removed: p.Removed, ModuleDirs: p.ModuleDirs, ModuleDirsSaid: p.ModuleDirsSaid,
|
|
After: AfterIn(p.Body), Created: now, 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: a recheck or the merge tool judges again.
|
|
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()
|
|
for _, d := range h.deliveries {
|
|
if d.Repository == p.repository() && d.Number == p.Number && !d.State.Final() && d.MergedAs == "" {
|
|
d.ClosedUnmerged = true
|
|
h.settleAndKeep(d)
|
|
if d.Group != "" {
|
|
h.leaveGroup(d)
|
|
}
|
|
}
|
|
}
|
|
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
|
|
// 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
|
|
}
|
|
}
|
|
}
|
|
g.Updated = now
|
|
h.keepGroup(g)
|
|
h.Logf("[mesh-delivery] group %s composed: %s — %s", g.ID, c.Verdict, c.Summary)
|
|
return
|
|
}
|
|
repository := strings.ToLower(c.Owner + "/" + c.Repo)
|
|
id := IDOf(repository, c.Commit)
|
|
d := h.deliveries[id]
|
|
if d == nil {
|
|
if c.Number == 0 {
|
|
return // a check of no pull request: nothing to deliver
|
|
}
|
|
d = &Delivery{ID: id, Repository: repository, Commit: c.Commit, Number: c.Number, Created: now}
|
|
t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: "its verdict arrived before its head was heard"})
|
|
if err != nil {
|
|
h.refuse(err)
|
|
return
|
|
}
|
|
h.moved(d, []Transition{t})
|
|
}
|
|
if d.State == 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
|
|
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).
|
|
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) {
|
|
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 {
|
|
return nil
|
|
}
|
|
h.walkAsked[id] = now
|
|
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 tick: groups, turns, adoption --------------------------------------------------------------------
|
|
|
|
// Tick takes every step the facts allow that needs another's word: a group's order and composed check, a
|
|
// delivery's turn, a walk with no delivery made one. The asking happens outside the lock.
|
|
func (h *Holder) Tick() {
|
|
for _, ask := range h.plan() {
|
|
ask()
|
|
}
|
|
for _, ask := range h.leaveStopped() {
|
|
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)
|
|
}
|
|
continue
|
|
}
|
|
if u.walk.Created.After(h.started) && now.Sub(u.first) < matchGrace {
|
|
continue
|
|
}
|
|
h.adopt(u.walk)
|
|
delete(h.unmatched, id)
|
|
}
|
|
|
|
// Plans not known for a delivery that has its diffset: asked once a tick.
|
|
for _, d := range h.deliveries {
|
|
if d.Plan == nil && !d.State.Final() && len(d.Paths) > 0 && (d.State == Ready || d.State == Held || d.MergedAs != "") {
|
|
d := d
|
|
id, repo, base, head, paths, dirs, removed := d.ID, d.Repository, d.Base, d.Commit, d.Paths, d.ModuleDirs, d.Removed
|
|
asks = append(asks, func() {
|
|
plan, err := h.Controller.Plan(repo, base, head, paths, dirs, removed)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] %s's plan cannot be asked yet: %v", id, err)
|
|
return
|
|
}
|
|
if d := h.deliveries[id]; d != nil && d.Plan == nil {
|
|
d.Plan = plan
|
|
h.settleAndKeep(d)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// Proposed, with no verdict and no check asked past the grace: its check is asked — a head adopted while
|
|
// it already waited, or one whose announcement the controller missed.
|
|
for _, d := range h.deliveries {
|
|
if a := h.askCheck(d, now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
|
|
// Merged 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.State.Final() || d.Walk == nil || d.Walk.TakenOverBy == "" {
|
|
continue
|
|
}
|
|
if s, known := h.walks[d.Walk.TakenOverBy]; known {
|
|
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 each tick, 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.
|
|
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 strandedUncovered(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, order, composed check, and their turns.
|
|
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")
|
|
}
|
|
// 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 {
|
|
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: stopped, %s", now.UTC().Format(time.RFC3339), d.ID, 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: stopped, %s", d.ID, 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 d.State == Stopped && d.Stranded != "" && d.CoveredBy == ""
|
|
}
|
|
|
|
// 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)
|
|
}
|
|
return d.State == Ready && d.MergedAs != "" && d.Walk == nil && !d.MovesNothing() && d.HeldWhy == "" &&
|
|
now.Sub(d.MergedAt) > unwalked && h.reconciled.After(d.MergedAt.Add(unwalked))
|
|
}
|
|
|
|
// 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 {
|
|
return nil
|
|
}
|
|
h.askTried[key] = now
|
|
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))
|
|
}
|
|
|
|
// askCheck is the ask of a proposed delivery's own check, when nothing else will bring its verdict: no verdict,
|
|
// no check this owner asked, and proposed past the grace the controller's own check of its announcement is
|
|
// given. Nil otherwise — once asked, a late verdict is the bound's and H2's to speak for.
|
|
func (h *Holder) askCheck(d *Delivery, now time.Time) func() {
|
|
if d.State != Proposed || d.Check != nil || d.CheckAsk != nil || d.MergedAs != "" || d.Number == 0 ||
|
|
now.Sub(d.Since) < checkGrace {
|
|
return nil
|
|
}
|
|
if tried, was := h.askTried[d.ID]; was && now.Sub(tried) < time.Minute {
|
|
return nil
|
|
}
|
|
h.askTried[d.ID] = now
|
|
id, m := d.ID, MemberOf(d)
|
|
why := fmt.Sprintf("proposed for %s with no verdict and no check asked: its announcement was heard from the "+
|
|
"bus's history, or the controller missed it", now.Sub(d.Since).Round(time.Minute))
|
|
return func() {
|
|
asked, err := h.Controller.Check("", []Member{m})
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] %s's check cannot be asked yet: %v", id, err)
|
|
return
|
|
}
|
|
delete(h.askTried, id)
|
|
d := h.deliveries[id]
|
|
if d == nil || d.State != Proposed || d.Check != nil {
|
|
return
|
|
}
|
|
d.CheckAsk = &CheckAsk{ID: asked, At: h.Now(), By: byOwner, Why: why}
|
|
h.Logf("[mesh-delivery] %s: its check asked of the controller as %s — %s", id, asked, why)
|
|
d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.Commit, Since: h.Now(),
|
|
Line: oneLine(fmt.Sprintf("%s delivery %s: its check asked by %s as %s: %s", h.Now().UTC().Format(time.RFC3339),
|
|
id, byOwner, asked, why))})
|
|
h.keep(d)
|
|
}
|
|
}
|
|
|
|
// adopt makes a delivery of a walk no pull request is known for: running already — at the switch, or on the
|
|
// controller's own path — it is delivering; waiting for a word, it waits for a person.
|
|
func (h *Holder) adopt(w Walk) {
|
|
now := h.Now()
|
|
d := &Delivery{ID: IDOf(w.Repository, w.Commit), Repository: strings.ToLower(w.Repository), Commit: w.Commit,
|
|
Base: w.Branch, MergedAs: w.Commit, Created: now}
|
|
if old := h.deliveries[d.ID]; old != nil {
|
|
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 order, worked out again whenever a member's head moved.
|
|
if !sameMembers(g.OrderOf, commits) && !g.Closed {
|
|
ms := make([]Member, 0, len(members))
|
|
for _, d := range members {
|
|
ms = append(ms, MemberOf(d))
|
|
}
|
|
id := g.ID
|
|
asks = append(asks, func() {
|
|
o, err := h.Controller.Order(ms)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
g := h.groups[id]
|
|
if g == nil {
|
|
return
|
|
}
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] group %s's order cannot be asked yet: %v", id, err)
|
|
return
|
|
}
|
|
g.Order, g.Pairs, g.Cycle, g.Contradictions, g.OrderOf = o.Order, o.Pairs, o.Cycle, o.Contradictions, commits
|
|
g.Check = nil // a check of another set of heads is no check of these
|
|
g.Updated = h.Now()
|
|
h.keepGroup(g)
|
|
})
|
|
return asks
|
|
}
|
|
// The composed check, once every member is ready and the order holds. 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))
|
|
if allReady && len(g.Cycle) == 0 && (g.Check == nil && !g.Closed || again) {
|
|
ms := h.inOrder(g, members)
|
|
id := g.ID
|
|
asks = append(asks, func() {
|
|
asked, err := h.Controller.Check(id, ms)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
g := h.groups[id]
|
|
if g == nil {
|
|
return
|
|
}
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] group %s's composed check cannot be asked yet: %v", id, err)
|
|
return
|
|
}
|
|
g.Check = &GroupCheck{Asked: asked, At: h.Now(), Commits: commits}
|
|
g.Updated = h.Now()
|
|
h.keepGroup(g)
|
|
})
|
|
}
|
|
// 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 != "" {
|
|
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.
|
|
if state, why := GroupState(g, members); state != g.Said {
|
|
g.Said = state
|
|
for _, d := range members {
|
|
g.Owed = append(g.Owed, Effect{Kind: EffectStatus, Commit: d.Commit, Context: "mesh/delivery-group",
|
|
State: groupStatusState(state), Line: clip(fmt.Sprintf("group %s %s: %s", g.ID, state, why), 140),
|
|
Since: now, Last: d.Repository + "#" + fmt.Sprint(d.Number) + " " + d.HTMLURL})
|
|
}
|
|
g.Owed = append(g.Owed, Effect{Kind: EffectEmit, Line: string(mustJSON(map[string]any{"group": g.ID,
|
|
"state": state, "why": why, "members": g.Members, "order": g.Order})), Since: now})
|
|
g.Updated = now
|
|
h.keepGroup(g)
|
|
}
|
|
// Its turns, in its order.
|
|
if g.Closed {
|
|
asks = append(asks, h.groupTurns(g, members)...)
|
|
}
|
|
return asks
|
|
}
|
|
|
|
// inOrder are a group's members as the controller ordered them.
|
|
func (h *Holder) inOrder(g *Group, members []*Delivery) []Member {
|
|
byID := map[string]*Delivery{}
|
|
for _, d := range members {
|
|
byID[d.ID] = d
|
|
}
|
|
var out []Member
|
|
for _, id := range g.Order {
|
|
if d := byID[id]; d != nil {
|
|
out = append(out, MemberOf(d))
|
|
delete(byID, id)
|
|
}
|
|
}
|
|
var rest []string
|
|
for id := range byID {
|
|
rest = append(rest, id)
|
|
}
|
|
sort.Strings(rest)
|
|
for _, id := range rest {
|
|
out = append(out, MemberOf(byID[id]))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// groupTurns walks a delivering group in its order: the next member is let go only when every member before
|
|
// it is delivered; a member that failed or was stopped stops every member after it, naming it. What was
|
|
// delivered before it stays, each judged at its own gate (ADR 0239 decision 4); the failed member's builds
|
|
// were put back at its gate by the controller. A superseded member counts as delivered once a later delivered
|
|
// walk carried it, holds the members after it while a later merge can still carry it, and stops them, saying
|
|
// why, only when none can (novox/hq issue 445).
|
|
func (h *Holder) groupTurns(g *Group, members []*Delivery) []func() {
|
|
waits := map[string]carryWait{}
|
|
asks := h.turnsOf(g, members, waits)
|
|
// What each member's turn waits on a carrier for is kept on it, and cleared once it no longer does, so `show`
|
|
// and `stalled` name the carrier and until when (ADR 0276 decision 9, novox/hq issue 445).
|
|
for _, d := range members {
|
|
w := waits[d.ID]
|
|
if d.State != Published || !d.LetGo.IsZero() {
|
|
w = carryWait{} // only a member still waiting for its turn waits for a carrier
|
|
}
|
|
var until *time.Time
|
|
if !w.until.IsZero() {
|
|
u := w.until
|
|
until = &u
|
|
}
|
|
if d.TurnAwaits != w.said || (d.TurnAwaitsUntil == nil) != (until == nil) ||
|
|
until != nil && !d.TurnAwaitsUntil.Equal(*until) {
|
|
d.TurnAwaits, d.TurnAwaitsUntil = w.said, until
|
|
h.keep(d)
|
|
}
|
|
}
|
|
return asks
|
|
}
|
|
|
|
// turnsOf is groupTurns' walk through the order; waits receives, by member, the carrier its turn waits for.
|
|
func (h *Holder) turnsOf(g *Group, members []*Delivery, waits map[string]carryWait) []func() {
|
|
ordered := h.inOrder(g, members)
|
|
byID := map[string]*Delivery{}
|
|
for _, d := range members {
|
|
byID[d.ID] = d
|
|
}
|
|
// Composed for these very heads: a verdict on another set of heads is no verdict on these.
|
|
composed := g.Check != nil && (g.Check.Verdict == "pass" || g.Check.Verdict == "warning") &&
|
|
sameMembers(g.Check.Commits, commitsOf(members)) && g.Check.OutOfDate == ""
|
|
// A member's turn registered, the members after it that the same walk carries take theirs too: one walk is let
|
|
// go once, for all of them (ADR 0276 decision 5). A member on another walk, or not at its turn, waits.
|
|
var asks []func()
|
|
sharing := ""
|
|
for i, m := range ordered {
|
|
d := byID[m.ID]
|
|
// A superseded member a later delivered walk carried counts as reached, as in GroupState: a group counts a
|
|
// covered member as delivered, so the member after it gets its turn (ADR 0270 decision 2, novox/hq
|
|
// issues 442 and 445).
|
|
if d.State.Reached() || d.State == Superseded && d.CoveredBy != "" {
|
|
continue
|
|
}
|
|
if sharing != "" && (d.Walk == nil || d.Walk.ID != sharing) {
|
|
return asks
|
|
}
|
|
switch {
|
|
case d.State == Superseded:
|
|
// Superseded is no failure: the merge that took its walk over may still deliver its change. While a later
|
|
// merge into its trunk that can carry it is open, within that merge's own bound, the members after it
|
|
// wait, as for a member still delivering (ADR 0239 decision 4: the next starts when the one before is
|
|
// delivered). Only when none can carry it are they stopped, saying why (novox/hq issue 445).
|
|
w := h.carryAwaited(d, h.Now())
|
|
// A delivered carrier is asked of the forge here too: coverOf asks only once every module has a
|
|
// delivered carrier, and while another module waits on an open one, this answer is what is awaited.
|
|
for _, in := range w.ask {
|
|
if a := h.askContains(d, in, h.Now()); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
if w.covered() {
|
|
continue // every module delivered by a later merge the forge says contains it: reached
|
|
}
|
|
if w.why != "" {
|
|
for _, later := range ordered[i+1:] {
|
|
if a := h.stopInGroup(byID[later.ID], fmt.Sprintf("its group %s stopped: %s superseded, and %s", g.ID,
|
|
d.ID, w.why)); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
return asks
|
|
}
|
|
w.said = fmt.Sprintf("its turn in group %s waits for %s, superseded, to be carried by %s, until %s", g.ID,
|
|
d.ID, w.said, w.until.UTC().Format(time.RFC3339))
|
|
for _, later := range ordered[i+1:] {
|
|
waits[later.ID] = w
|
|
}
|
|
// Under ADR 0276 a member is superseded when its waiting walk was folded into a successor that does not
|
|
// answer its merge, and the member after it was folded into the same successor: the carrier is then in
|
|
// the next member's own walk, which waits for that member's turn. Waiting there would wait for itself
|
|
// until the carrier's bound. That walk delivers both, in the group's order among its tiers (decision 5),
|
|
// so the members on it take their turn as members sharing one walk do; a member on another walk waits.
|
|
// It is shared only when that one walk carries every module of the member not yet carried, each by a
|
|
// merge the forge said contains the member's: a carrier of some modules on it, others elsewhere, would
|
|
// let the next member go on a walk that cannot deliver the member before it (review of #203).
|
|
shared, ask := h.sharedWalk(d, w, h.Now())
|
|
asks = append(asks, ask...)
|
|
if shared != "" && sharing == "" {
|
|
sharing = shared
|
|
continue
|
|
}
|
|
return asks
|
|
case d.State == Failed || d.State == Stopped:
|
|
for _, later := range ordered[i+1:] {
|
|
if a := h.stopInGroup(byID[later.ID], fmt.Sprintf("its group %s stopped: %s %s", g.ID, d.ID, d.State)); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
return asks
|
|
case d.State == Held:
|
|
// Held for a composed check that has since passed for the heads that merged: lifted, published again,
|
|
// and its turn taken in order (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, fmt.Sprintf("its turn in group %s (%d of %d)", g.ID, i+1, len(ordered))); a != nil {
|
|
asks, sharing = append(asks, a), d.Walk.ID
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
return asks
|
|
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 — a member rebased or pushed — 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))) {
|
|
return asks
|
|
}
|
|
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)
|
|
return asks
|
|
}
|
|
if a := h.turn(d, fmt.Sprintf("its turn in group %s (%d of %d)", g.ID, i+1, len(ordered))); a != nil {
|
|
asks, sharing = append(asks, a), d.Walk.ID
|
|
delete(waits, d.ID)
|
|
continue
|
|
}
|
|
return asks
|
|
default:
|
|
return asks // not on the trunk yet, held, or delivering: the members after it wait
|
|
}
|
|
}
|
|
return asks
|
|
}
|
|
|
|
// containsGrace is how long a later merge whose walk was delivered may still be awaited as a superseded member's
|
|
// carrier while the forge has not said whether its merge commit contains the member's: as long as a published
|
|
// delivery may wait for its turn (Bounds[Published]). A forge that cannot answer means "not yet covered" (ADR
|
|
// 0270), and without this bound the members after it would wait on that answer for ever (novox/hq issue 445).
|
|
const containsGrace = 30 * time.Minute
|
|
|
|
// readyCarrierBound is how long a merged carrier with no walk yet may be awaited from its merge. Its state, ready,
|
|
// has no bound of its own: it waits in a batch, which closes at the latest ten minutes after its first merge
|
|
// (ADR 0276 decision 1, `merge-window-at-most`), and then behind at most the one open walk, within that walk's
|
|
// bound (decisions 4 and 9). A merge no walk answers is held after `unwalked`, and a held one is no carrier.
|
|
const readyCarrierBound = 10*time.Minute + 2*time.Hour
|
|
|
|
// carryWait is what a superseded member's group waits for: the later merges that may still carry it, said, and
|
|
// the moment the wait ends at the latest; or why none can carry it. Neither means every module it moves was
|
|
// delivered by a later merge the forge says contains it: covered.
|
|
type carryWait struct {
|
|
by []*Delivery
|
|
// byModule are the carriers awaited for each module not yet carried.
|
|
byModule map[string][]*Delivery
|
|
said string
|
|
until time.Time
|
|
why string
|
|
// ask are the delivered carriers' merge commits the forge has not yet said contain the member's.
|
|
ask []string
|
|
}
|
|
|
|
func (w carryWait) covered() bool { return len(w.by) == 0 && w.why == "" }
|
|
|
|
// sharedWalk is the walk, not started and waiting for this owner's word, that carries every module of a superseded
|
|
// member not yet carried, each by a carrier the forge said contains the member's merge: the walk the members after
|
|
// it may take their turn on. Whether a carrier on it contains the merge, when not yet known, is asked of the forge,
|
|
// and the members wait for the answer within the carrier's bound (ADR 0270, novox/hq issue 445).
|
|
func (h *Holder) sharedWalk(d *Delivery, w carryWait, now time.Time) (string, []func()) {
|
|
var walks []string
|
|
for _, c := range w.by {
|
|
if !c.State.Final() && c.Walk != nil && c.Walk.Waits && !c.Walk.Started() && !containsString(walks, c.Walk.ID) {
|
|
walks = append(walks, c.Walk.ID)
|
|
}
|
|
}
|
|
sort.Strings(walks)
|
|
var asks []func()
|
|
for _, walk := range walks {
|
|
every, said := true, true
|
|
var unknown []string
|
|
for _, carriers := range w.byModule {
|
|
on, yes := false, false
|
|
for _, c := range carriers {
|
|
if c.State.Final() || c.Walk == nil || c.Walk.ID != walk || !c.Walk.Waits || c.Walk.Started() {
|
|
continue
|
|
}
|
|
on = true
|
|
contained, known := d.ContainedIn[c.MergedAs]
|
|
switch {
|
|
case known && contained:
|
|
yes = true
|
|
case !known && !containsString(unknown, c.MergedAs):
|
|
unknown = append(unknown, c.MergedAs)
|
|
}
|
|
}
|
|
every = every && on
|
|
said = said && yes
|
|
}
|
|
if !every {
|
|
continue // a module is carried elsewhere, or by nothing on this walk: it cannot deliver the member
|
|
}
|
|
if said {
|
|
return walk, asks
|
|
}
|
|
for _, in := range unknown {
|
|
if a := h.askContains(d, in, now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
}
|
|
return "", asks
|
|
}
|
|
|
|
// carryAwaited says which later merges into a superseded member's trunk may still carry every module it moves —
|
|
// the cover GroupState counts as reached (ADR 0270's covered test, novox/hq issue 442) — and until when, or,
|
|
// when none can, why. A later merge of the same repository into the same trunk on a passing check, whose plan
|
|
// moves the module and whose merge commit the forge has not said leaves this one out, can carry it while it is
|
|
// merged and not yet ended, not held for a person, and within its bound (ADR 0239 decision 9, ADR 0276 decision
|
|
// 9: every wait has a bound): its state's, or readyCarrierBound from its merge while no walk carries it yet; once
|
|
// delivered, while the forge has not yet said whether it contains this merge, within containsGrace. A carrier
|
|
// superseded in turn ends; the later merge that took it over is a carrier of its own (novox/hq issue 445).
|
|
func (h *Holder) carryAwaited(d *Delivery, now time.Time) carryWait {
|
|
if d.MergedAs == "" || d.MergedAt.IsZero() || d.Plan == nil || len(d.Plan.Moved) == 0 {
|
|
return carryWait{why: "nothing records which modules it moves, so no later merge can carry it"}
|
|
}
|
|
var w carryWait
|
|
seen := map[string]bool{}
|
|
modules:
|
|
for _, module := range d.Plan.Moved {
|
|
found := false
|
|
var latest time.Time // the last moment a carrier of this module may still be awaited
|
|
var not []string
|
|
for _, c := range h.deliveries {
|
|
if !mayCarry(c, d, module) {
|
|
continue
|
|
}
|
|
contained, known := d.ContainedIn[c.MergedAs]
|
|
var until time.Time
|
|
switch {
|
|
case known && !contained:
|
|
not = append(not, c.ID+" does not contain its merge")
|
|
continue
|
|
case c.State == Delivered && carries(c, d, module) && known:
|
|
continue modules // delivered and contained: this module is carried
|
|
case c.State == Delivered && carries(c, d, module):
|
|
until = c.Since.Add(containsGrace)
|
|
if now.After(until) {
|
|
not = append(not, fmt.Sprintf("the forge did not say within %s whether %s contains its merge", containsGrace, c.ID))
|
|
continue
|
|
}
|
|
if !containsString(w.ask, c.MergedAs) {
|
|
w.ask = append(w.ask, c.MergedAs)
|
|
}
|
|
case c.State.Final() || c.State == Held:
|
|
continue // ended without carrying it, or held for a person: no carrier to wait for
|
|
case c.State == Ready:
|
|
until = c.MergedAt.Add(readyCarrierBound)
|
|
if now.After(until) {
|
|
not = append(not, fmt.Sprintf("%s is merged and no walk carried it within %s", c.ID, readyCarrierBound))
|
|
continue
|
|
}
|
|
default:
|
|
b, bounded := boundOf(c)
|
|
if pastBound(c, now) {
|
|
not = append(not, fmt.Sprintf("%s is past its bound (%s %s)", c.ID, c.State, b.For))
|
|
continue
|
|
}
|
|
if until = now.Add(Bounds[Delivering].For); bounded && b.For > 0 {
|
|
until = boundSince(c).Add(b.For)
|
|
}
|
|
}
|
|
found = true
|
|
if w.byModule == nil {
|
|
w.byModule = map[string][]*Delivery{}
|
|
}
|
|
w.byModule[module] = append(w.byModule[module], c)
|
|
if until.After(latest) {
|
|
latest = until
|
|
}
|
|
if !seen[c.ID] {
|
|
seen[c.ID] = true
|
|
w.by = append(w.by, c)
|
|
}
|
|
}
|
|
if !found {
|
|
why := fmt.Sprintf("no later merge into %s can carry %s", d.Base, module)
|
|
if len(not) > 0 {
|
|
sort.Strings(not)
|
|
why += ": " + strings.Join(not, "; ")
|
|
}
|
|
return carryWait{why: why}
|
|
}
|
|
// The wait ends when the first module runs out of carriers, whatever the others still have.
|
|
if w.until.IsZero() || latest.Before(w.until) {
|
|
w.until = latest
|
|
}
|
|
}
|
|
sort.Slice(w.by, func(i, j int) bool { return w.by[i].ID < w.by[j].ID })
|
|
var said []string
|
|
for _, c := range w.by {
|
|
s := fmt.Sprintf("%s (%s", c.ID, c.State)
|
|
if c.Walk != nil {
|
|
s += ", walk " + c.Walk.ID
|
|
}
|
|
said = append(said, s+")")
|
|
}
|
|
w.said = strings.Join(said, ", ")
|
|
return w
|
|
}
|
|
|
|
// mayCarry is whether c is a later merge that could carry module for d, by what is known before its walk ends: of
|
|
// the same repository, merged later into the same trunk, on its own passing check, its plan moving the module.
|
|
func mayCarry(c, d *Delivery, module string) bool {
|
|
return c.ID != d.ID && 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.Check.Passes() &&
|
|
c.Plan != nil && slices.Contains(c.Plan.Moved, module)
|
|
}
|
|
|
|
// stopInGroup stops a member after one that failed, and leaves its merge out of its walk if it has one open.
|
|
//
|
|
// **Only its own change** (novox/hq issue 459, ADR 0299): one walk carries every merge of its batch, so ending the
|
|
// member's walk ended every other delivery on it, which nobody stopped. Its merge is left out instead: the
|
|
// controller ends the walk and walks the other merges again without it, and without each later merge of its
|
|
// repository, which contains it. Every member a pass stops on one walk is gathered first, and the walk is asked
|
|
// once, by the first member's ask, after the pass: a member is never walked again because its stop came second.
|
|
func (h *Holder) stopInGroup(d *Delivery, why string) func() {
|
|
if d == nil || d.State.Final() {
|
|
return nil
|
|
}
|
|
t, err := Apply(d, EvStop, true, Facts{Now: h.Now(), By: byOwner, Why: why})
|
|
if err != nil {
|
|
h.refuse(err)
|
|
return nil
|
|
}
|
|
h.moved(d, []Transition{t})
|
|
// Its open walk; or, its walk stopped for another merge, the batch or walk that walks its merge again (review
|
|
// of #221): a merge walked again and then stopped by its group is left out there too.
|
|
var walk string
|
|
switch {
|
|
case d.Walk == nil:
|
|
return nil
|
|
case d.Walk.State == walkBuilding || d.Walk.State == walkRolling:
|
|
walk = d.Walk.ID
|
|
case d.walkedAgain():
|
|
walk = d.Walk.WalkedAgainBy
|
|
default:
|
|
return nil
|
|
}
|
|
if !h.leave(walk, d) {
|
|
return nil // the first member left out of this walk asks for every one
|
|
}
|
|
return func() {
|
|
without, shared := h.leaving459(walk)
|
|
if !shared {
|
|
if err := h.Controller.Stop(walk, why, byOwner); err != nil {
|
|
h.Logf("[mesh-delivery] the walk %s could not be stopped: %v", walk, err)
|
|
}
|
|
return
|
|
}
|
|
h.Logf("[mesh-delivery] %s", h.stopLeaving(walk, why, byOwner, without))
|
|
}
|
|
}
|
|
|
|
// 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
|
|
}
|
|
// 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
|
|
}
|
|
|
|
// 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}
|
|
}
|
|
|
|
// 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 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
|
|
}
|
|
// 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 {
|
|
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.
|
|
//
|
|
// groupTurns agrees: the members after a carried member take their turn, and they wait while a later merge can still
|
|
// carry it (novox/hq issue 445).
|
|
func GroupState(g *Group, members []*Delivery) (string, string) {
|
|
count := map[State]int{}
|
|
reached, ended := 0, 0
|
|
var uncovered []string
|
|
for _, d := range members {
|
|
count[d.State]++
|
|
if d.State.Final() {
|
|
ended++
|
|
}
|
|
switch {
|
|
case d.State.Reached(), d.State == Superseded && d.CoveredBy != "":
|
|
reached++
|
|
case d.State == Superseded:
|
|
uncovered = append(uncovered, d.ID)
|
|
}
|
|
}
|
|
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 && 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:
|
|
return "delivering", fmt.Sprintf("%d of %d delivered", reached, n)
|
|
case len(g.Cycle) > 0:
|
|
return "rejected", cycleSaid(g)
|
|
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)
|
|
}
|
|
|
|
// 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: "
|
|
|
|
func groupStatusState(s string) string {
|
|
switch s {
|
|
case "ready", "delivered":
|
|
return "success"
|
|
case "rejected", "failed":
|
|
return "failure"
|
|
case "stopped":
|
|
return "error"
|
|
}
|
|
return "pending"
|
|
}
|
|
|
|
func clip(s string, n int) string {
|
|
if len(s) > n {
|
|
return s[:n-1] + "…"
|
|
}
|
|
return s
|
|
}
|
|
|
|
func containsString(xs []string, x string) bool {
|
|
for _, y := range xs {
|
|
if y == x {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// cycleSaid is why a group's order is refused: each contradiction no precedence resolves, with the rules on
|
|
// both sides and how to declare the order — or, for a cycle through more members, the pairs in it.
|
|
func cycleSaid(g *Group) string {
|
|
why := "its order contradicts itself: "
|
|
if len(g.Contradictions) > 0 {
|
|
var said []string
|
|
for _, c := range g.Contradictions {
|
|
said = append(said, c.Said)
|
|
}
|
|
return why + strings.Join(said, "; ")
|
|
}
|
|
var pairs []string
|
|
for _, p := range g.Pairs {
|
|
if slices.Contains(g.Cycle, p.Before) && slices.Contains(g.Cycle, p.After) {
|
|
pairs = append(pairs, p.orderSaid())
|
|
}
|
|
}
|
|
why += strings.Join(g.Cycle, ", ")
|
|
if len(pairs) > 0 {
|
|
why += " (" + strings.Join(pairs, "; ") + "); declare the order with a line `after: <repository>` in " +
|
|
"the description of the pull request that goes later"
|
|
}
|
|
return why
|
|
}
|
|
|
|
// rejudgedSince is whether a member was judged again after a moment: a verdict of its own heard since.
|
|
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
|
|
}
|