jochen
32d55beb47
mesh/merge-gate pass: builds gitea, mesh-delivery → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery superseded: a newer delivery to the same trunk took over its walk
The confirmation review asked for the delivery domain's own vocabulary instead of a provider standing. A gitea health check would need a `tools` list (turning gitea's tools into seat verbs), a check of another kind beside it, and would hold the forge's own deliveries at the first-node gate; so the delivery's owner says it, with the word it already has. - gitea_open_pulls (gitea): every open pull request from the forge itself, with what its base requires and the statuses its head has. - mesh-delivery asks it every five minutes and keeps its own first-seen time per head (the forge's update time moves on every comment). A head on a base requiring mesh/merge-gate OR mesh/repo-check (hq requires only the second) with neither for ten minutes is a `stalled` line in state `unannounced`, which the controller raises as delivery.<id>.stalled (D14); H2 has nothing to close. Once announced it is a delivery of its own. - The gitea-side watcher and its provider standing are gone.
1332 lines
43 KiB
Go
1332 lines
43 KiB
Go
package main
|
|
|
|
import (
|
|
"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
|
|
dirtyGroup map[string]bool
|
|
unmatched map[string]unmatched
|
|
backlog []Walk
|
|
started time.Time
|
|
loaded bool
|
|
held bool
|
|
reconciled time.Time
|
|
unreached string
|
|
// askTried is when a check this owner asks was last tried and could not be asked: tried again a minute on.
|
|
askTried map[string]time.Time
|
|
refusals []string
|
|
// unchecked are the open pull requests' heads seen without a merge check (unchecked.go).
|
|
unchecked map[string]unchecked
|
|
}
|
|
|
|
// unmatched is a walk no delivery is known for yet: its merge not heard, or merged before this owner held.
|
|
type unmatched struct {
|
|
walk Walk
|
|
first time.Time
|
|
}
|
|
|
|
// The words this owner signs its acts with.
|
|
const (
|
|
byOwner = "mesh-delivery"
|
|
byH2 = "healer H2"
|
|
)
|
|
|
|
// How long a walk is given for its merge to be heard before it is made a delivery of its own.
|
|
const matchGrace = 2 * time.Minute
|
|
|
|
// How long a merged delivery may go without a walk before it is held, saying so.
|
|
const unwalked = 10 * time.Minute
|
|
|
|
// How long a final delivery is kept; its note on the commit keeps it after.
|
|
const keptFinal = 30 * 24 * time.Hour
|
|
|
|
func (h *Holder) init() {
|
|
if h.Now == nil {
|
|
h.Now = func() time.Time { return time.Now().UTC() }
|
|
}
|
|
if h.Logf == nil {
|
|
h.Logf = func(string, ...any) {}
|
|
}
|
|
if h.Emit == nil {
|
|
h.Emit = func(string, any) error { return nil }
|
|
}
|
|
h.deliveries = map[string]*Delivery{}
|
|
h.groups = map[string]*Group{}
|
|
h.dirty = map[string]bool{}
|
|
h.dirtyGroup = map[string]bool{}
|
|
h.unmatched = map[string]unmatched{}
|
|
h.askTried = map[string]time.Time{}
|
|
h.unchecked = map[string]unchecked{}
|
|
h.started = h.Now()
|
|
}
|
|
|
|
// Load reads back every delivery and group, before any input is taken: a restarted holder resumes each
|
|
// where it stood.
|
|
func (h *Holder) Load() error {
|
|
ds, err := h.Store.Deliveries()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
gs, err := h.Store.Groups()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
for _, d := range ds {
|
|
h.deliveries[d.ID] = d
|
|
}
|
|
for _, g := range gs {
|
|
h.groups[g.ID] = g
|
|
}
|
|
h.loaded = true
|
|
h.Logf("[mesh-delivery] read back %d delivery(ies) and %d group(s)", len(ds), len(gs))
|
|
return nil
|
|
}
|
|
|
|
// Refuse keeps a refusal from outside the lock.
|
|
func (h *Holder) Refuse(err error) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
h.refuse(err)
|
|
}
|
|
|
|
// refuse keeps a refusal, said, for the status; under the lock.
|
|
func (h *Holder) refuse(err error) {
|
|
h.Logf("[mesh-delivery] %v", err)
|
|
h.refusals = append(h.refusals, h.Now().Format(time.RFC3339)+" "+err.Error())
|
|
if len(h.refusals) > 50 {
|
|
h.refusals = h.refusals[len(h.refusals)-50:]
|
|
}
|
|
}
|
|
|
|
// moved records what a delivery's transitions owe outside, and keeps it.
|
|
func (h *Holder) moved(d *Delivery, ts []Transition) {
|
|
now := h.Now()
|
|
for _, t := range ts {
|
|
h.Logf("[mesh-delivery] %s: %s → %s (%s): %s", d.ID, orNothing(t.From), t.To, t.Event, t.Why)
|
|
d.Owed = append(d.Owed, Effect{Kind: EffectEmit, Line: string(mustJSON(transitionEvent(d, t))), Since: now})
|
|
line := noteLine(d, t)
|
|
d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.Commit, Line: line, Since: now})
|
|
if d.MergedAs != "" && !sameCommit(d.MergedAs, d.Commit) && afterTheTrunk(t.To) {
|
|
d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.MergedAs, Line: line, Since: now})
|
|
}
|
|
if t.To.Final() && d.MergedAs != "" && len(d.Steps) > 0 {
|
|
d.Owed = append(d.Owed, Effect{Kind: EffectNote, Commit: d.MergedAs, Line: executedLine(d), Since: now})
|
|
}
|
|
}
|
|
if len(ts) > 0 && d.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.
|
|
func (h *Holder) keep(d *Delivery) {
|
|
h.deliveries[d.ID] = d
|
|
if err := h.Store.PutDelivery(d); err != nil {
|
|
h.dirty[d.ID] = true
|
|
h.Logf("[mesh-delivery] %s could not be kept yet (%v); kept again before it is said", d.ID, err)
|
|
return
|
|
}
|
|
delete(h.dirty, d.ID)
|
|
}
|
|
|
|
func (h *Holder) keepGroup(g *Group) {
|
|
h.groups[g.ID] = g
|
|
if err := h.Store.PutGroup(g); err != nil {
|
|
h.dirtyGroup[g.ID] = true
|
|
h.Logf("[mesh-delivery] group %s could not be kept yet (%v)", g.ID, err)
|
|
return
|
|
}
|
|
delete(h.dirtyGroup, g.ID)
|
|
}
|
|
|
|
// settleAndKeep moves a delivery as far as its facts allow, and keeps it.
|
|
func (h *Holder) settleAndKeep(d *Delivery) {
|
|
ts := settle(d, Facts{Now: h.Now()})
|
|
h.moved(d, ts)
|
|
h.closeGroupOf(d)
|
|
}
|
|
|
|
// --- inputs ---------------------------------------------------------------------------------------------
|
|
|
|
// PullEvent is the forge's `pull.updated`, `pull.merged` or `pull.closed`.
|
|
type PullEvent struct {
|
|
Owner string `json:"owner"`
|
|
Repo string `json:"repo"`
|
|
Number int `json:"number"`
|
|
Title string `json:"title,omitempty"`
|
|
Body string `json:"body,omitempty"`
|
|
Base string `json:"base"`
|
|
Head string `json:"head"`
|
|
HeadSHA string `json:"head_sha"`
|
|
MergeCommit string `json:"merge_commit_sha,omitempty"`
|
|
MergedAt string `json:"merged_at,omitempty"`
|
|
CloneURL string `json:"clone_url,omitempty"`
|
|
HTMLURL string `json:"html_url,omitempty"`
|
|
Paths []string `json:"paths,omitempty"`
|
|
PathsTruncated bool `json:"paths_truncated,omitempty"`
|
|
Removed []string `json:"removed,omitempty"`
|
|
ModuleDirs []string `json:"module_dirs,omitempty"`
|
|
ModuleDirsSaid bool `json:"module_dirs_said,omitempty"`
|
|
HeadChecks map[string]string `json:"head_checks,omitempty"`
|
|
}
|
|
|
|
func (p PullEvent) repository() string { return strings.ToLower(p.Owner + "/" + p.Repo) }
|
|
|
|
// trunkLike are branch names that never make a group: a pull request from them is nobody's feature.
|
|
var trunkLike = map[string]bool{"main": true, "master": true, "trunk": true, "": true}
|
|
|
|
// PullUpdated takes a pull request's new head: a delivery proposed, an older head of the same pull request
|
|
// superseded, and the delivery put in the group its branch name makes.
|
|
func (h *Holder) PullUpdated(p PullEvent) {
|
|
if p.HeadSHA == "" || p.Number == 0 {
|
|
h.Refuse(errors.New("a pull request announced with no head or no number; nothing proposed"))
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
id := IDOf(p.repository(), p.HeadSHA)
|
|
if d := h.deliveries[id]; d != nil {
|
|
return // announced again: nothing new
|
|
}
|
|
now := h.Now()
|
|
for _, old := range h.deliveries {
|
|
if old.Repository == p.repository() && old.Number == p.Number && !sameCommit(old.Commit, p.HeadSHA) &&
|
|
!old.State.Final() && old.MergedAs == "" {
|
|
old.NewerHead = p.HeadSHA
|
|
h.settleAndKeep(old)
|
|
if old.Group != "" {
|
|
h.leaveGroup(old)
|
|
}
|
|
}
|
|
}
|
|
d := &Delivery{ID: id, Repository: p.repository(), Commit: p.HeadSHA, Number: p.Number, Title: p.Title,
|
|
Base: p.Base, Branch: p.Head, HTMLURL: p.HTMLURL, CloneURL: p.CloneURL, Paths: p.Paths,
|
|
PathsTruncated: p.PathsTruncated, Removed: p.Removed, ModuleDirs: p.ModuleDirs, ModuleDirsSaid: p.ModuleDirsSaid,
|
|
After: AfterIn(p.Body), Created: now}
|
|
t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: fmt.Sprintf("%s#%d's head announced", d.Repository, d.Number)})
|
|
if err != nil {
|
|
h.refuse(err)
|
|
return
|
|
}
|
|
ts := []Transition{t}
|
|
// An announcement heard again from the bus's history may carry its head's verdict already: the forge holds
|
|
// the gate's status on that very commit. A decided one is taken; a pending or errored one is asked again.
|
|
if v := decidedOnHead(p.HeadChecks, now); v != nil {
|
|
d.Check = v
|
|
ts = append(ts, settle(d, Facts{Now: now})...)
|
|
}
|
|
h.moved(d, ts)
|
|
h.joinGroup(d)
|
|
}
|
|
|
|
// decidedOnHead is the verdict the forge's statuses on a head hold, when the gate decided one: passed, warned
|
|
// or failed. Nil for none, pending, or a check that could not run — those are asked again.
|
|
func decidedOnHead(checks map[string]string, now time.Time) *Verdict {
|
|
switch checks["mesh/merge-gate"] {
|
|
case "success", "warning", "failure":
|
|
default:
|
|
return nil
|
|
}
|
|
return &Verdict{At: now, Gate: verdictOfState(checks["mesh/merge-gate"]),
|
|
Summary: "the head's mesh/merge-gate, as the forge holds it", Repo: verdictOfState(checks["mesh/repo-check"])}
|
|
}
|
|
|
|
// joinGroup puts a new head in the group its branch name makes with the other repositories' open pull
|
|
// requests (ADR 0239 decision 4): one mechanism, the name the feature already has in every repository.
|
|
func (h *Holder) joinGroup(d *Delivery) {
|
|
if trunkLike[strings.ToLower(d.Branch)] || strings.EqualFold(d.Branch, d.Base) {
|
|
return
|
|
}
|
|
g := h.groups[d.Branch]
|
|
if g != nil && g.Closed {
|
|
h.Logf("[mesh-delivery] %s shares the branch %s with a group that is already delivering: a lone delivery",
|
|
d.ID, d.Branch)
|
|
return
|
|
}
|
|
var others []*Delivery
|
|
for _, o := range h.deliveries {
|
|
if o.ID != d.ID && o.Branch == d.Branch && o.Repository != d.Repository && o.Number > 0 &&
|
|
!o.State.Final() && o.MergedAs == "" {
|
|
others = append(others, o)
|
|
}
|
|
}
|
|
if len(others) == 0 && g == nil {
|
|
return
|
|
}
|
|
now := h.Now()
|
|
if g == nil {
|
|
g = &Group{ID: d.Branch, Created: now}
|
|
}
|
|
for _, o := range append(others, d) {
|
|
if !containsString(g.Members, o.ID) {
|
|
g.Members = append(g.Members, o.ID)
|
|
}
|
|
if o.Group != g.ID {
|
|
o.Group = g.ID
|
|
h.keep(o)
|
|
}
|
|
}
|
|
sort.Strings(g.Members)
|
|
g.Updated = now
|
|
h.keepGroup(g)
|
|
h.Logf("[mesh-delivery] group %s: %s", g.ID, strings.Join(g.Members, ", "))
|
|
}
|
|
|
|
// leaveGroup takes a superseded or stopped head out of its group's members.
|
|
func (h *Holder) leaveGroup(d *Delivery) {
|
|
g := h.groups[d.Group]
|
|
if g == nil {
|
|
return
|
|
}
|
|
var kept []string
|
|
for _, m := range g.Members {
|
|
if m != d.ID {
|
|
kept = append(kept, m)
|
|
}
|
|
}
|
|
g.Members = kept
|
|
g.Updated = h.Now()
|
|
h.keepGroup(g)
|
|
}
|
|
|
|
// PullMerged takes a merge: the commit it landed as, and the delivery it was — made from the forge's word
|
|
// when this owner never heard its head (merged before it held the seat).
|
|
func (h *Holder) PullMerged(p PullEvent) {
|
|
if p.MergeCommit == "" {
|
|
h.Refuse(errors.New("a merge announced with no merge commit; nothing published"))
|
|
return
|
|
}
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
var d *Delivery
|
|
if p.HeadSHA != "" {
|
|
d = h.deliveries[IDOf(p.repository(), p.HeadSHA)]
|
|
}
|
|
if d == nil {
|
|
var newest time.Time
|
|
for _, c := range h.deliveries {
|
|
// 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 for this commit.
|
|
for id, u := range h.unmatched {
|
|
if strings.EqualFold(u.walk.Repository, d.Repository) && sameCommit(u.walk.Commit, d.MergedAs) {
|
|
h.linkWalk(d, u.walk)
|
|
delete(h.unmatched, id)
|
|
}
|
|
}
|
|
// 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 = "merged without a passing check: only a person decides that it goes on"
|
|
}
|
|
h.moved(d, ts)
|
|
h.closeGroupOf(d)
|
|
}
|
|
|
|
// closeGroupOf closes a delivery's group once it reached the trunk: a delivering group takes no new member.
|
|
func (h *Holder) closeGroupOf(d *Delivery) {
|
|
if d.Group == "" || d.MergedAs == "" {
|
|
return
|
|
}
|
|
if g := h.groups[d.Group]; g != nil && !g.Closed {
|
|
g.Closed = true
|
|
g.Updated = h.Now()
|
|
h.keepGroup(g)
|
|
}
|
|
}
|
|
|
|
// verdictOfState is a commit status's state as a verdict.
|
|
func verdictOfState(s string) string {
|
|
switch s {
|
|
case "success":
|
|
return "pass"
|
|
case "warning":
|
|
return "warning"
|
|
case "failure":
|
|
return "fail"
|
|
case "":
|
|
return ""
|
|
}
|
|
return "error"
|
|
}
|
|
|
|
// PullClosed takes a pull request closed unmerged: its open deliveries are stopped.
|
|
func (h *Holder) PullClosed(p PullEvent) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
for _, d := range h.deliveries {
|
|
if d.Repository == p.repository() && d.Number == p.Number && !d.State.Final() && d.MergedAs == "" {
|
|
d.ClosedUnmerged = true
|
|
h.settleAndKeep(d)
|
|
if d.Group != "" {
|
|
h.leaveGroup(d)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// CheckedEvent is the controller's `checked`.
|
|
type CheckedEvent struct {
|
|
Owner string `json:"owner"`
|
|
Repo string `json:"repo"`
|
|
Number int `json:"number,omitempty"`
|
|
Commit string `json:"commit"`
|
|
Verdict string `json:"verdict"`
|
|
Summary string `json:"summary"`
|
|
ID string `json:"id"`
|
|
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"`
|
|
} `json:"members,omitempty"`
|
|
}
|
|
|
|
// Checked takes a verdict: a head's own, or a group's composed check.
|
|
func (h *Holder) Checked(c CheckedEvent) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
now := h.Now()
|
|
if c.Group != "" {
|
|
g := h.groups[c.Group]
|
|
if g == nil || g.Check == nil || g.Check.Asked != c.ID {
|
|
h.Logf("[mesh-delivery] a composed check of %s (%s) that is not the one asked: history", c.Group, c.ID)
|
|
return
|
|
}
|
|
g.Check.Verdict, g.Check.Summary = c.Verdict, c.Summary
|
|
g.Updated = now
|
|
h.keepGroup(g)
|
|
h.Logf("[mesh-delivery] group %s composed: %s — %s", g.ID, c.Verdict, c.Summary)
|
|
return
|
|
}
|
|
repository := strings.ToLower(c.Owner + "/" + c.Repo)
|
|
id := IDOf(repository, c.Commit)
|
|
d := h.deliveries[id]
|
|
if d == nil {
|
|
if c.Number == 0 {
|
|
return // a check of no pull request: nothing to deliver
|
|
}
|
|
d = &Delivery{ID: id, Repository: repository, Commit: c.Commit, Number: c.Number, Created: now}
|
|
t, err := Apply(d, EvAnnounced, false, Facts{Now: now, Why: "its verdict arrived before its head was heard"})
|
|
if err != nil {
|
|
h.refuse(err)
|
|
return
|
|
}
|
|
h.moved(d, []Transition{t})
|
|
}
|
|
if d.State != Proposed {
|
|
h.Logf("[mesh-delivery] %s is %s: a verdict for it now is history (%s)", d.ID, d.State, c.Verdict)
|
|
return
|
|
}
|
|
v := &Verdict{ID: c.ID, On: c.On, At: now, Gate: c.Verdict, Summary: c.Summary, Report: c.Report}
|
|
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 = v
|
|
if c.Plan != nil {
|
|
d.Plan = c.Plan
|
|
}
|
|
h.settleAndKeep(d)
|
|
}
|
|
|
|
// WalkMoved takes a walk's record, from `plan-moved` or the controller's walks: the delivery it walks
|
|
// moves with it; a backlog walk's carried builds are credited to the deliveries that published them.
|
|
func (h *Holder) WalkMoved(w Walk) {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
h.walkMoved(w)
|
|
}
|
|
|
|
func (h *Holder) walkMoved(w Walk) {
|
|
if w.Release != nil {
|
|
h.creditBacklog(w)
|
|
return
|
|
}
|
|
d := h.deliveryWalking(w)
|
|
if d == nil {
|
|
if _, seen := h.unmatched[w.ID]; !seen {
|
|
h.unmatched[w.ID] = unmatched{walk: w, first: h.Now()}
|
|
} else {
|
|
u := h.unmatched[w.ID]
|
|
if w.Revision >= u.walk.Revision {
|
|
u.walk = w
|
|
h.unmatched[w.ID] = u
|
|
}
|
|
}
|
|
return
|
|
}
|
|
if d.Walk != nil && d.Walk.ID == w.ID && w.Revision < d.Walk.Revision {
|
|
return // an older record arriving late
|
|
}
|
|
h.linkWalk(d, w)
|
|
h.settleAndKeep(d)
|
|
// What its sends carried of other deliveries' builds.
|
|
for _, m := range w.Modules {
|
|
if m != nil && m.Gate != nil {
|
|
h.creditCarried(m.Gate, w.ID, d.ID)
|
|
}
|
|
}
|
|
}
|
|
|
|
// deliveryWalking is the delivery a walk walks: the one that landed as its commit, or that it walked before.
|
|
func (h *Holder) deliveryWalking(w Walk) *Delivery {
|
|
for _, d := range h.deliveries {
|
|
if d.Walk != nil && d.Walk.ID == w.ID {
|
|
return d
|
|
}
|
|
}
|
|
for _, d := range h.deliveries {
|
|
if strings.EqualFold(d.Repository, w.Repository) && sameCommit(d.MergedAs, w.Commit) {
|
|
return d
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// linkWalk records a walk on its delivery: what it is at, and the machine steps its record says — each
|
|
// moved only as the step table allows.
|
|
func (h *Holder) linkWalk(d *Delivery, w Walk) {
|
|
d.Walk = w.Seen()
|
|
for _, s := range StepsOf(w) {
|
|
h.step(d, s)
|
|
}
|
|
}
|
|
|
|
// step moves one machine's step, or refuses a move the step table does not hold; it says whether it moved.
|
|
func (h *Holder) step(d *Delivery, s Step) bool {
|
|
for i, old := range d.Steps {
|
|
if old.Machine == s.Machine && old.Module == s.Module {
|
|
if old.State == s.State {
|
|
return false
|
|
}
|
|
if !StepMay(old.State, s.State) {
|
|
h.refuse(fmt.Errorf("%s: %s on %s from %s to %s is refused — the step table holds no such move",
|
|
d.ID, s.Module, s.Machine, old.State, s.State))
|
|
return false
|
|
}
|
|
d.Steps[i] = s
|
|
return true
|
|
}
|
|
}
|
|
d.Steps = append(d.Steps, s)
|
|
return true
|
|
}
|
|
|
|
// creditBacklog keeps a backlog walk, and credits what its gate carried.
|
|
func (h *Holder) creditBacklog(w Walk) {
|
|
replaced := false
|
|
for i, b := range h.backlog {
|
|
if b.ID == w.ID {
|
|
h.backlog[i], replaced = w, true
|
|
}
|
|
}
|
|
if !replaced {
|
|
h.backlog = append(h.backlog, w)
|
|
if len(h.backlog) > 5 {
|
|
h.backlog = h.backlog[len(h.backlog)-5:]
|
|
}
|
|
}
|
|
if w.Release.Gate != nil {
|
|
h.creditCarried(w.Release.Gate, w.ID, "")
|
|
}
|
|
}
|
|
|
|
// creditCarried gives the delivery that published each carried build its machine step.
|
|
func (h *Holder) creditCarried(g *WalkGate, walk, except string) {
|
|
state := StepJudging
|
|
switch {
|
|
case g.Rollback == "rolled-back":
|
|
state = StepRolledBack
|
|
case g.Verdict == "failed":
|
|
state = StepFailed
|
|
case g.Verdict == "passed":
|
|
state = StepPassed
|
|
}
|
|
for _, c := range g.Carried {
|
|
for _, d := range h.deliveries {
|
|
if d.ID == except || d.MergedAs == "" || !sameCommit(d.MergedAs, c.To) {
|
|
continue
|
|
}
|
|
if h.step(d, Step{Machine: c.Node, Module: c.Module, Build: c.Build, State: state, Since: h.Now(),
|
|
Why: "carried by " + walk}) {
|
|
h.keep(d)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// --- the controller's walks, read back --------------------------------------------------------------------
|
|
|
|
// Reconcile asks the controller for the walks it keeps and takes each as if said: what was missed while
|
|
// this holder was down, or said by a command that says nothing, is found by comparison.
|
|
func (h *Holder) Reconcile() error {
|
|
held, walks, err := h.Controller.Walks("")
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.unreached = err.Error()
|
|
return err
|
|
}
|
|
h.unreached, h.held, h.reconciled = "", held, h.Now()
|
|
for _, w := range walks {
|
|
h.walkMoved(w)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// --- the tick: groups, turns, adoption --------------------------------------------------------------------
|
|
|
|
// Tick takes every step the facts allow that needs another's word: a group's order and composed check, a
|
|
// delivery's turn, a walk with no delivery made one. The asking happens outside the lock.
|
|
func (h *Holder) Tick() {
|
|
for _, ask := range h.plan() {
|
|
ask()
|
|
}
|
|
}
|
|
|
|
// plan works out, under the lock, what to ask; each ask applies its own answer.
|
|
func (h *Holder) plan() []func() {
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
now := h.Now()
|
|
var asks []func()
|
|
|
|
// Walks no merge was heard for: made deliveries of their own once their merge had time to be heard.
|
|
for id, u := range h.unmatched {
|
|
if d := h.deliveryWalking(u.walk); d != nil {
|
|
h.linkWalk(d, u.walk)
|
|
h.settleAndKeep(d)
|
|
delete(h.unmatched, id)
|
|
continue
|
|
}
|
|
if !u.walk.Open() {
|
|
if now.Sub(u.first) > 10*time.Minute {
|
|
delete(h.unmatched, id)
|
|
}
|
|
continue
|
|
}
|
|
if u.walk.Created.After(h.started) && now.Sub(u.first) < matchGrace {
|
|
continue
|
|
}
|
|
h.adopt(u.walk)
|
|
delete(h.unmatched, id)
|
|
}
|
|
|
|
// Plans not known for a delivery that has its diffset: asked once a tick.
|
|
for _, d := range h.deliveries {
|
|
if d.Plan == nil && !d.State.Final() && len(d.Paths) > 0 && (d.State == Ready || d.State == Held || d.MergedAs != "") {
|
|
d := d
|
|
id, repo, base, head, paths, dirs, removed := d.ID, d.Repository, d.Base, d.Commit, d.Paths, d.ModuleDirs, d.Removed
|
|
asks = append(asks, func() {
|
|
plan, err := h.Controller.Plan(repo, base, head, paths, dirs, removed)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] %s's plan cannot be asked yet: %v", id, err)
|
|
return
|
|
}
|
|
if d := h.deliveries[id]; d != nil && d.Plan == nil {
|
|
d.Plan = plan
|
|
h.settleAndKeep(d)
|
|
}
|
|
})
|
|
}
|
|
}
|
|
|
|
// Proposed, with no verdict and no check asked past the grace: its check is asked — a head adopted while
|
|
// it already waited, or one whose announcement the controller missed.
|
|
for _, d := range h.deliveries {
|
|
if a := h.askCheck(d, now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
|
|
// 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.
|
|
for _, d := range h.deliveries {
|
|
if d.State != Held && !h.unwalkedPast(d, now) {
|
|
continue
|
|
}
|
|
if d.CoveredBy == "" {
|
|
var ask []string
|
|
d.CoveredBy, ask = h.coverOf(d)
|
|
for _, in := range ask {
|
|
if a := h.askContains(d, in, now); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
}
|
|
if ts := settle(d, Facts{Now: now}); len(ts) > 0 {
|
|
h.moved(d, ts)
|
|
h.closeGroupOf(d)
|
|
}
|
|
}
|
|
|
|
// 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 = "merged, and the controller opened no walk for it within " + unwalked.String() + " — nothing it " +
|
|
"holds follows that branch, or the merge was not heard: a person decides"
|
|
h.settleAndKeep(d)
|
|
}
|
|
}
|
|
|
|
// Groups: membership, order, composed check, and their turns.
|
|
grouped := map[string]bool{}
|
|
for _, g := range h.groups {
|
|
members := h.membersOf(g)
|
|
for _, m := range members {
|
|
grouped[m.ID] = true
|
|
}
|
|
asks = append(asks, h.planGroup(g, members, now)...)
|
|
}
|
|
|
|
// A lone delivery's turn: published, its walk waiting, nothing before it.
|
|
for _, d := range h.deliveries {
|
|
if grouped[d.ID] || d.Group != "" && h.groups[d.Group] != nil {
|
|
continue
|
|
}
|
|
if a := h.turn(d, "its turn: a delivery in no group"); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
|
|
// Final deliveries past their keeping, once nothing is owed for them.
|
|
for id, d := range h.deliveries {
|
|
if d.State.Final() && len(d.Owed) == 0 && now.Sub(d.Since) > keptFinal {
|
|
if err := h.Store.DeleteDelivery(id); err == nil {
|
|
delete(h.deliveries, id)
|
|
}
|
|
}
|
|
}
|
|
return asks
|
|
}
|
|
|
|
// 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 {
|
|
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. 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
|
|
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
|
|
}
|
|
}
|
|
return everywhere
|
|
}
|
|
|
|
// 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 {
|
|
h.linkWalk(old, w)
|
|
h.settleAndKeep(old)
|
|
return
|
|
}
|
|
h.linkWalk(d, w)
|
|
var t Transition
|
|
var err error
|
|
if d.Walk.Waits {
|
|
d.HeldWhy = "a commit on the trunk with no pull request this owner heard of — merged before it held the " +
|
|
"seat, or pushed to the trunk — waits for a person's release"
|
|
t, err = Apply(d, EvAppeared, false, Facts{Now: now, Why: d.HeldWhy})
|
|
} else {
|
|
t, err = Apply(d, EvAdopted, false, Facts{Now: now, Why: "a walk already running when this owner first heard " +
|
|
"of it (" + w.ID + "), adopted as its delivery"})
|
|
}
|
|
if err != nil {
|
|
h.refuse(err)
|
|
return
|
|
}
|
|
h.moved(d, []Transition{t})
|
|
h.settleAndKeep(d)
|
|
}
|
|
|
|
// turn is the ask that lets a published delivery's waiting walk start; nil when it is not its turn.
|
|
func (h *Holder) turn(d *Delivery, why string) func() {
|
|
if d.Walk == nil || !d.Walk.Waits || !d.LetGo.IsZero() {
|
|
return nil
|
|
}
|
|
if d.State != Published && d.State != Delivering {
|
|
return nil
|
|
}
|
|
id, walk := d.ID, d.Walk.ID
|
|
return func() {
|
|
err := h.Controller.Deliver(walk, why)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
d := h.deliveries[id]
|
|
if d == nil {
|
|
return
|
|
}
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] %s's walk %s could not be let go yet: %v", id, walk, err)
|
|
return
|
|
}
|
|
d.LetGo, d.LetGoWhy = h.Now(), why
|
|
if d.Walk != nil && d.Walk.ID == walk {
|
|
d.Walk.Waits, d.Walk.GoBy = false, byOwner
|
|
}
|
|
h.settleAndKeep(d)
|
|
}
|
|
}
|
|
|
|
// membersOf are a group's current members, by id order.
|
|
func (h *Holder) membersOf(g *Group) []*Delivery {
|
|
var out []*Delivery
|
|
for _, id := range g.Members {
|
|
if d := h.deliveries[id]; d != nil {
|
|
out = append(out, d)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// planGroup is what a group needs asked, and its turns.
|
|
func (h *Holder) planGroup(g *Group, members []*Delivery, now time.Time) []func() {
|
|
var asks []func()
|
|
if len(members) == 0 {
|
|
return nil
|
|
}
|
|
if len(members) == 1 && !g.Closed {
|
|
// A group of one is no group: its member is a lone delivery again.
|
|
members[0].Group = ""
|
|
h.keep(members[0])
|
|
g.Members = nil
|
|
h.keepGroup(g)
|
|
return nil
|
|
}
|
|
commits := commitsOf(members)
|
|
// The order, worked out again whenever a member's head moved.
|
|
if !sameMembers(g.OrderOf, commits) && !g.Closed {
|
|
ms := make([]Member, 0, len(members))
|
|
for _, d := range members {
|
|
ms = append(ms, MemberOf(d))
|
|
}
|
|
id := g.ID
|
|
asks = append(asks, func() {
|
|
o, err := h.Controller.Order(ms)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
g := h.groups[id]
|
|
if g == nil {
|
|
return
|
|
}
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] group %s's order cannot be asked yet: %v", id, err)
|
|
return
|
|
}
|
|
g.Order, g.Pairs, g.Cycle, g.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.
|
|
allReady := true
|
|
for _, d := range members {
|
|
if d.State != Ready {
|
|
allReady = false
|
|
}
|
|
}
|
|
if allReady && len(g.Cycle) == 0 && !g.Closed && (g.Check == nil || !sameMembers(g.Check.Commits, commits)) {
|
|
ms := h.inOrder(g, members)
|
|
id := g.ID
|
|
asks = append(asks, func() {
|
|
asked, err := h.Controller.Check(id, ms)
|
|
h.mu.Lock()
|
|
defer h.mu.Unlock()
|
|
g := h.groups[id]
|
|
if g == nil {
|
|
return
|
|
}
|
|
if err != nil {
|
|
h.Logf("[mesh-delivery] group %s's composed check cannot be asked yet: %v", id, err)
|
|
return
|
|
}
|
|
g.Check = &GroupCheck{Asked: asked, At: h.Now(), Commits: commits}
|
|
g.Updated = h.Now()
|
|
h.keepGroup(g)
|
|
})
|
|
}
|
|
// Its state, said when it changes.
|
|
if state, why := GroupState(g, members); state != g.Said {
|
|
g.Said = state
|
|
for _, d := range members {
|
|
g.Owed = append(g.Owed, Effect{Kind: EffectStatus, Commit: d.Commit, Context: "mesh/delivery-group",
|
|
State: groupStatusState(state), Line: clip(fmt.Sprintf("group %s %s: %s", g.ID, state, why), 140),
|
|
Since: now, Last: d.Repository + "#" + fmt.Sprint(d.Number) + " " + d.HTMLURL})
|
|
}
|
|
g.Owed = append(g.Owed, Effect{Kind: EffectEmit, Line: string(mustJSON(map[string]any{"group": g.ID,
|
|
"state": state, "why": why, "members": g.Members, "order": g.Order})), Since: now})
|
|
g.Updated = now
|
|
h.keepGroup(g)
|
|
}
|
|
// Its turns, in its order.
|
|
if g.Closed {
|
|
asks = append(asks, h.groupTurns(g, members)...)
|
|
}
|
|
return asks
|
|
}
|
|
|
|
// inOrder are a group's members as the controller ordered them.
|
|
func (h *Holder) inOrder(g *Group, members []*Delivery) []Member {
|
|
byID := map[string]*Delivery{}
|
|
for _, d := range members {
|
|
byID[d.ID] = d
|
|
}
|
|
var out []Member
|
|
for _, id := range g.Order {
|
|
if d := byID[id]; d != nil {
|
|
out = append(out, MemberOf(d))
|
|
delete(byID, id)
|
|
}
|
|
}
|
|
var rest []string
|
|
for id := range byID {
|
|
rest = append(rest, id)
|
|
}
|
|
sort.Strings(rest)
|
|
for _, id := range rest {
|
|
out = append(out, MemberOf(byID[id]))
|
|
}
|
|
return out
|
|
}
|
|
|
|
// groupTurns walks a delivering group in its order: the next member is let go only when every member before
|
|
// it is delivered; a member that failed or was stopped stops every member after it, naming it. What was
|
|
// delivered before it stays, each judged at its own gate (ADR 0239 decision 4); the failed member's builds
|
|
// were put back at its gate by the controller.
|
|
func (h *Holder) groupTurns(g *Group, members []*Delivery) []func() {
|
|
ordered := h.inOrder(g, members)
|
|
byID := map[string]*Delivery{}
|
|
for _, d := range members {
|
|
byID[d.ID] = d
|
|
}
|
|
// Composed for these very heads: a verdict on another set of heads is no verdict on these.
|
|
composed := g.Check != nil && (g.Check.Verdict == "pass" || g.Check.Verdict == "warning") &&
|
|
sameMembers(g.Check.Commits, commitsOf(members))
|
|
for i, m := range ordered {
|
|
d := byID[m.ID]
|
|
switch {
|
|
case d.State.Reached():
|
|
continue
|
|
case d.State == Failed || d.State == Stopped || d.State == Superseded:
|
|
var asks []func()
|
|
for _, later := range ordered[i+1:] {
|
|
if a := h.stopInGroup(byID[later.ID], fmt.Sprintf("its group %s stopped: %s %s", g.ID, d.ID, d.State)); a != nil {
|
|
asks = append(asks, a)
|
|
}
|
|
}
|
|
return asks
|
|
case d.State == 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 {
|
|
return []func(){a}
|
|
}
|
|
}
|
|
}
|
|
return nil
|
|
case d.State == Published:
|
|
// Asked and not answered is not "did not pass": it waits, published, for the answer (issue 316).
|
|
if !composed && d.HeldWhy == "" && g.Check != nil && g.Check.Verdict == "" &&
|
|
sameMembers(g.Check.Commits, commitsOf(members)) {
|
|
return nil
|
|
}
|
|
if !composed && d.HeldWhy == "" {
|
|
d.HeldWhy = fmt.Sprintf("its group %s's composed check did not pass for the heads that merged; a person "+
|
|
"decides that it goes on", g.ID)
|
|
h.settleAndKeep(d)
|
|
return nil
|
|
}
|
|
if a := h.turn(d, fmt.Sprintf("its turn in group %s (%d of %d)", g.ID, i+1, len(ordered))); a != nil {
|
|
return []func(){a}
|
|
}
|
|
return nil
|
|
default:
|
|
return nil // not on the trunk yet, held, or delivering: the members after it wait
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// stopInGroup stops a member after one that failed, and its walk if it has one open.
|
|
func (h *Holder) stopInGroup(d *Delivery, why string) func() {
|
|
if d == nil || d.State.Final() {
|
|
return nil
|
|
}
|
|
t, err := Apply(d, EvStop, true, Facts{Now: h.Now(), By: byOwner, Why: why})
|
|
if err != nil {
|
|
h.refuse(err)
|
|
return nil
|
|
}
|
|
h.moved(d, []Transition{t})
|
|
if d.Walk == nil || d.Walk.State == walkDone || d.Walk.State == walkFailed || d.Walk.State == walkSuperseded {
|
|
return nil
|
|
}
|
|
walk := d.Walk.ID
|
|
return func() {
|
|
if err := h.Controller.Stop(walk, why, byOwner); err != nil {
|
|
h.Logf("[mesh-delivery] the walk %s could not be stopped: %v", walk, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// GroupState is a group's state, derived from its members and its composed check — never set apart.
|
|
func GroupState(g *Group, members []*Delivery) (string, string) {
|
|
count := map[State]int{}
|
|
for _, d := range members {
|
|
count[d.State]++
|
|
}
|
|
n := len(members)
|
|
switch {
|
|
case count[Failed] > 0:
|
|
return "failed", "a member failed"
|
|
case count[Stopped] > 0 && g.Closed:
|
|
return "stopped", "a member was stopped"
|
|
case n > 0 && count[Delivered]+count[Retired]+count[Covered] == n:
|
|
return "delivered", "every member is delivered"
|
|
case g.Closed:
|
|
return "delivering", fmt.Sprintf("%d of %d delivered", count[Delivered]+count[Covered], 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.Verdict != "" && g.Check.Verdict != "pass" && g.Check.Verdict != "warning":
|
|
return "rejected", "composed together: " + g.Check.Summary
|
|
case count[Ready] == n && g.Check != nil && sameMembers(g.Check.Commits, commitsOf(members)) &&
|
|
(g.Check.Verdict == "pass" || g.Check.Verdict == "warning"):
|
|
return "ready", "every member ready, and composed together they pass"
|
|
}
|
|
return "checking", fmt.Sprintf("%d of %d member(s) ready", count[Ready], n)
|
|
}
|
|
|
|
func groupStatusState(s string) string {
|
|
switch s {
|
|
case "ready", "delivered":
|
|
return "success"
|
|
case "rejected", "failed":
|
|
return "failure"
|
|
case "stopped":
|
|
return "error"
|
|
}
|
|
return "pending"
|
|
}
|
|
|
|
func clip(s string, n int) string {
|
|
if len(s) > n {
|
|
return s[:n-1] + "…"
|
|
}
|
|
return s
|
|
}
|
|
|
|
func containsString(xs []string, x string) bool {
|
|
for _, y := range xs {
|
|
if y == x {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// 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
|
|
}
|