615 lines
24 KiB
Go
615 lines
24 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net/url"
|
|
"path"
|
|
"regexp"
|
|
"slices"
|
|
"sort"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/asks"
|
|
sdkdelivery "git.novox.be/novox/mesh-sdk/go/delivery"
|
|
|
|
"github.com/novox/mesh-controller/internal/conditions"
|
|
"github.com/novox/mesh-controller/internal/inventory"
|
|
"github.com/novox/mesh-controller/internal/link"
|
|
)
|
|
|
|
// **The controller catches up on merges it did not hear** (novox/hq ADR 0457, issue 1336). A merge the forge announced while the
|
|
// controller was being replaced was never heard, and nothing asked for it again: the trunk held a merge no walk answered. At its
|
|
// start, and each time it hears a merge, the controller reads the last catchUpWindow merge commits of a trunk from the forge, and
|
|
// hears each one it holds no record of as if the forge had announced it then, on the warrant the router holds for it.
|
|
// It lasts until release 6 of design 47 deletes delivery plans and batches (ADR 0429 §7, §8).
|
|
|
|
// catchUpWindow is how many of a trunk's newest merge commits are read; an older merge is a person's to decide.
|
|
const catchUpWindow = 50
|
|
|
|
// kindMergeNotHeard is the condition of a merge the controller did not hear and holds no warrant to plan.
|
|
const kindMergeNotHeard = "merge-not-heard"
|
|
|
|
// trunkMerge is a merge commit of a trunk as the forge lists it.
|
|
type trunkMerge struct {
|
|
Commit string
|
|
// Head is the merge commit's second parent: the pull request's head that was merged.
|
|
Head string
|
|
Subject string
|
|
At time.Time
|
|
}
|
|
|
|
// trunkCommit is a commit of a trunk as the forge lists it, newest first: a merge commit has a second parent.
|
|
type trunkCommit struct {
|
|
Commit, SecondParent, Subject string
|
|
At time.Time
|
|
}
|
|
|
|
// trunkCommitsPage is one page (fifty) of a trunk's commits, newest first; empty past the last. A variable so a test needs no
|
|
// forge.
|
|
var trunkCommitsPage = func(ctx context.Context, owner, repo, branch string, page int) ([]trunkCommit, error) {
|
|
var found struct {
|
|
Result []struct {
|
|
SHA string `json:"sha"`
|
|
Parents []struct {
|
|
SHA string `json:"sha"`
|
|
} `json:"parents"`
|
|
Commit struct {
|
|
Message string `json:"message"`
|
|
Committer struct {
|
|
Date string `json:"date"`
|
|
} `json:"committer"`
|
|
} `json:"commit"`
|
|
} `json:"result"`
|
|
}
|
|
apiPath := fmt.Sprintf("/repos/%s/%s/commits?sha=%s&limit=50&page=%d&stat=false&verification=false&files=false",
|
|
owner, repo, url.QueryEscape(branch), page)
|
|
missing, err := askForgeApi(ctx, apiPath, &found)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if missing {
|
|
return nil, fmt.Errorf("the forge holds no branch %s of %s/%s", branch, owner, repo)
|
|
}
|
|
var out []trunkCommit
|
|
for _, c := range found.Result {
|
|
at, _ := time.Parse(time.RFC3339, c.Commit.Committer.Date)
|
|
tc := trunkCommit{Commit: c.SHA, Subject: firstLine(c.Commit.Message), At: at.UTC()}
|
|
if len(c.Parents) >= 2 {
|
|
tc.SecondParent = c.Parents[1].SHA
|
|
}
|
|
out = append(out, tc)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// readTrunkMerges is the last catchUpWindow merge commits of a repository's trunk, oldest first: pages are read until that many
|
|
// merges are held or the trunk ends.
|
|
func readTrunkMerges(ctx context.Context, owner, repo, branch string) ([]trunkMerge, error) {
|
|
var merges []trunkMerge
|
|
for page := 1; page <= 8 && len(merges) < catchUpWindow; page++ {
|
|
commits, err := trunkCommitsPage(ctx, owner, repo, branch, page)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(commits) == 0 {
|
|
break
|
|
}
|
|
for _, c := range commits {
|
|
// A merge of the trunk is a pull request's merge into it. A commit with a second parent that is none is a merge made
|
|
// inside some branch ("Merge remote-tracking branch 'origin/main' into fix/..."), which the trunk's history lists too and
|
|
// which no Approve ever covered: read as a gap it would raise a condition for nothing a person decides.
|
|
if c.SecondParent != "" && mergesIntoTrunk(c.Subject, branch) {
|
|
merges = append(merges, trunkMerge{Commit: c.Commit, Head: c.SecondParent, Subject: c.Subject, At: c.At})
|
|
}
|
|
}
|
|
}
|
|
return lastMerges(merges), nil
|
|
}
|
|
|
|
// lastMerges is the newest catchUpWindow of merges, given newest first, oldest first.
|
|
func lastMerges(newestFirst []trunkMerge) []trunkMerge {
|
|
if len(newestFirst) > catchUpWindow {
|
|
newestFirst = newestFirst[:catchUpWindow]
|
|
}
|
|
out := slices.Clone(newestFirst)
|
|
slices.Reverse(out)
|
|
return out
|
|
}
|
|
|
|
var mergeSubject = regexp.MustCompile(`^Merge pull request '(.*)' \(#(\d+)\) from (\S+) into (\S+)$`)
|
|
|
|
// pullOfSubject is the pull request a merge commit's subject names: its title and number.
|
|
func pullOfSubject(subject string) (title string, number int, ok bool) {
|
|
m := mergeSubject.FindStringSubmatch(strings.TrimSpace(subject))
|
|
if m == nil {
|
|
return "", 0, false
|
|
}
|
|
n, err := strconv.Atoi(m[2])
|
|
if err != nil || n <= 0 {
|
|
return "", 0, false
|
|
}
|
|
return m[1], n, true
|
|
}
|
|
|
|
// mergesIntoTrunk says a commit's subject is a pull request's merge into this trunk.
|
|
func mergesIntoTrunk(subject, branch string) bool {
|
|
m := mergeSubject.FindStringSubmatch(strings.TrimSpace(subject))
|
|
return m != nil && m[4] == branch
|
|
}
|
|
|
|
// headBranchOfSubject is the branch a merge commit's subject says was merged.
|
|
func headBranchOfSubject(subject string) string {
|
|
if m := mergeSubject.FindStringSubmatch(strings.TrimSpace(subject)); m != nil {
|
|
return m[3]
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// readMergeEvent is the merge as the forge would have announced it: the commit, the pull request, the files it changed and the
|
|
// directories among them that hold a module.json. A variable so a test needs no forge.
|
|
var readMergeEvent = func(ctx context.Context, owner, repo, branch string, tm trunkMerge) (link.SourceMoved, error) {
|
|
m := link.SourceMoved{Owner: owner, Repo: repo, Base: branch, Head: headBranchOfSubject(tm.Subject), HeadSHA: tm.Head,
|
|
Commit: tm.Commit, MergedAt: tm.At.UTC().Format(time.RFC3339)}
|
|
if title, number, ok := pullOfSubject(tm.Subject); ok {
|
|
m.Number, m.Title = number, title
|
|
}
|
|
var found struct {
|
|
Result struct {
|
|
Files []struct {
|
|
Filename string `json:"filename"`
|
|
Status string `json:"status"`
|
|
} `json:"files"`
|
|
} `json:"result"`
|
|
}
|
|
missing, err := askForgeApi(ctx, "/repos/"+owner+"/"+repo+"/git/commits/"+url.PathEscape(tm.Commit), &found)
|
|
if err != nil {
|
|
return m, err
|
|
}
|
|
if missing || len(found.Result.Files) == 0 {
|
|
return m, fmt.Errorf("the forge named no files for the merge %.8s of %s/%s", tm.Commit, owner, repo)
|
|
}
|
|
for _, f := range found.Result.Files {
|
|
m.Paths = append(m.Paths, f.Filename)
|
|
if f.Status == "removed" || f.Status == "deleted" {
|
|
m.Removed = append(m.Removed, f.Filename)
|
|
}
|
|
}
|
|
// The forge lists a merge's files by the page; a list as long as a page may be a beginning.
|
|
m.PathsTruncated = len(m.Paths) >= 300
|
|
dirs := map[string]bool{}
|
|
for _, p := range m.Paths {
|
|
for d := path.Dir(p); ; d = path.Dir(d) {
|
|
dirs[d] = true
|
|
if d == "." || d == "/" {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
if len(dirs) > 64 {
|
|
return m, nil // too many to look at: the old rule stands, a directory is a module when its manifest changed
|
|
}
|
|
var holding []string
|
|
for d := range dirs {
|
|
file := "module.json"
|
|
if d != "." {
|
|
file = d + "/module.json"
|
|
}
|
|
_, exists, err := manifestCached(ctx, owner, repo, file, tm.Commit)
|
|
if err != nil {
|
|
return m, err
|
|
}
|
|
if exists {
|
|
holding = append(holding, d)
|
|
}
|
|
}
|
|
sort.Strings(holding)
|
|
m.ModuleDirs, m.ModuleDirsSaid = holding, true
|
|
return m, nil
|
|
}
|
|
|
|
// mergeAskPrefix is what the id of a merge ask made by the retired module mesh-delivery begins with: the pull request's number
|
|
// and the head's first twelve digits (`merge-<number>-<head>-<try>`). The teller's ids name neither (see findMergeWarrant).
|
|
func mergeAskPrefix(number int, head string) string {
|
|
if len(head) > 12 {
|
|
head = head[:12]
|
|
}
|
|
return fmt.Sprintf("merge-%d-%s-", number, head)
|
|
}
|
|
|
|
// askersOfMerges are the askers a merge's warrant is read under: the teller, and mesh-delivery for a merge before release 3e
|
|
// (ADR 0437 §6). Release 4's controller work stops accepting mesh-delivery (ADR 0450 §8); a merge asked under it earlier is
|
|
// still found here until then.
|
|
func askersOfMerges() []string { return []string{sdkdelivery.TellerModule, theDelivery} }
|
|
|
|
// caughtUpWarrant judges a chosen merge ask of the router as the warrant of a merge: the ask is the operator's, the warrant
|
|
// answers that very ask as it was published, and it is the operator's Approve proved by the channel it came through (ADR 0437
|
|
// §6, as movedOnApprove judges an announced one). The heads the Approve was bound to are checked by the caller, where it can
|
|
// name them.
|
|
func caughtUpWarrant(asker string, ask *asks.Ask, w *asks.Warrant) (string, error) {
|
|
if err := askerAccepted(asker); err != nil {
|
|
return "", err
|
|
}
|
|
switch {
|
|
case ask == nil || w == nil:
|
|
return "", errors.New("the router holds no closed record of the ask")
|
|
case ask.Who != asks.Operator:
|
|
return "", fmt.Errorf("the ask %s is for %q to answer, not the operator", ask.ID, ask.Who)
|
|
}
|
|
option, err := w.For(asker, *ask)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
if option.ID != optionApprove || w.Option != optionApprove {
|
|
return "", fmt.Errorf("the operator chose %q on the ask %s, not an Approve", w.Option, w.Ask)
|
|
}
|
|
if w.By == nil || w.By.Who != asks.Operator || w.By.Verified == "" {
|
|
return "", fmt.Errorf("the ask %s was not answered by the operator on a channel that proves who answered", w.Ask)
|
|
}
|
|
return w.Says() + " (ask " + w.Ask + ")", nil
|
|
}
|
|
|
|
// routerAskIDs are the ids of the asks an asker has in the router's record. A variable so a test needs no bus.
|
|
var routerAskIDs = func(ctx context.Context, inv catalogueReader, asker string) ([]string, error) {
|
|
serving := servingBus.Load()
|
|
if serving == nil {
|
|
return nil, errors.New("only the serving controller reads the router's record")
|
|
}
|
|
bucket, err := asksRecords(ctx, inv)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if bucket == "" {
|
|
return nil, errors.New("no module declares the operator channel's records")
|
|
}
|
|
api, err := jetstream.New(serving.Conn())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
kv, err := api.KeyValue(ctx, bucket)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return askIDsOf(ctx, kv, asker)
|
|
}
|
|
|
|
// askIDsOf are the ids of the asks an asker has in the router's bucket.
|
|
func askIDsOf(ctx context.Context, kv jetstream.KeyValue, asker string) ([]string, error) {
|
|
lister, err := kv.ListKeysFiltered(ctx, asker+".>")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var ids []string
|
|
for key := range lister.Keys() {
|
|
ids = append(ids, strings.TrimPrefix(key, asker+"."))
|
|
}
|
|
// The lister stops its own watch when it has sent the last key, so a stop of ours finds it stopped: nats answers that with
|
|
// "invalid subscription", which is no failure of the read (issue 1336). Any other error stays one.
|
|
if err := lister.Stop(); err != nil && !errors.Is(err, nats.ErrBadSubscription) {
|
|
return nil, err
|
|
}
|
|
return ids, nil
|
|
}
|
|
|
|
// closedRecord is a router record of an ask that ended: it never changes, so it is read once.
|
|
type closedRecord struct {
|
|
ask *asks.Ask
|
|
w *asks.Warrant
|
|
}
|
|
|
|
var closedRecords sync.Map // asker + "." + id -> closedRecord
|
|
|
|
// closedRecordOf is the router's record of an ask, kept once it is closed. A nil ask is an ask still open, ended without a
|
|
// choice, or not there.
|
|
func closedRecordOf(ctx context.Context, inv catalogueReader, asker, id string) (*asks.Ask, *asks.Warrant, error) {
|
|
if v, ok := closedRecords.Load(asker + "." + id); ok {
|
|
r := v.(closedRecord)
|
|
return r.ask, r.w, nil
|
|
}
|
|
ask, w, err := routerAskOf(ctx, inv, asker, id)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if ask != nil && w != nil {
|
|
closedRecords.Store(asker+"."+id, closedRecord{ask, w})
|
|
}
|
|
return ask, w, nil
|
|
}
|
|
|
|
// deliveryIDOf is the id the gate gives a delivery, and so the key of its head in the heads an Approve binds: the repository as
|
|
// owner/name, lower-cased, `@` and the first twelve characters of the head.
|
|
func deliveryIDOf(owner, repo, head string) string {
|
|
if len(head) > 12 {
|
|
head = head[:12]
|
|
}
|
|
return strings.ToLower(owner+"/"+repo) + "@" + head
|
|
}
|
|
|
|
// coversHead says an ask the operator approved is for this merge: its Approve is bound to the heads of the pull request alone
|
|
// (mergeAct over {delivery id: head}, which is how a pull request not in a group is asked), or, for a group's ask, whose other
|
|
// members' heads the router's record does not carry in full, the ask as published names this pull request at this head in its
|
|
// details ("<repo>#<number> ... at <head>:"), and the digest the warrant answers covers those words.
|
|
func coversHead(ask *asks.Ask, owner, repo string, number int, head string) (how string, ok bool) {
|
|
option, found := ask.Option(optionApprove)
|
|
if !found {
|
|
return "", false
|
|
}
|
|
heads := map[string]string{deliveryIDOf(owner, repo, head): head}
|
|
if option.Performs(mergeAct(ask.ID, optionApprove, heads)) == nil {
|
|
return "bound to the head of this pull request", true
|
|
}
|
|
short := head
|
|
if len(short) > 12 {
|
|
short = short[:12]
|
|
}
|
|
line := regexp.MustCompile(`(?m)^` + regexp.QuoteMeta(repo) + `#` + strconv.Itoa(number) + ` .* at ` + regexp.QuoteMeta(short) + `:`)
|
|
if line.MatchString(ask.Details) {
|
|
return "a group's ask naming this pull request at this head", true
|
|
}
|
|
return "", false
|
|
}
|
|
|
|
// findMergeWarrant finds the warrant of a merge, with no read the controller does not make today: the router's chosen merge
|
|
// ask for it. Under the teller (releases 3e on) the id is `merge-<group>-<heads hash>-<tail>`, which names neither the pull
|
|
// request nor the head, so the teller's closed asks are read and the one the operator approved whose heads cover the merge's
|
|
// second parent is the warrant (coversHead). Under mesh-delivery, which made the earlier merges, the id names the pull request
|
|
// and the head and is found by them. Why it holds; or found false with why not, for a person to read; or an error when the
|
|
// router could not be read, which is no answer. A variable so a test needs no bus.
|
|
var findMergeWarrant = func(ctx context.Context, inv catalogueReader, owner, repo string, number int, head string) (why string, found bool, reason string, err error) {
|
|
reason = "the router holds no chosen merge ask of " + strings.Join(askersOfMerges(), " or ") + " that covers the head " + head[:min(12, len(head))]
|
|
for _, asker := range askersOfMerges() {
|
|
ids, err := routerAskIDs(ctx, inv, asker)
|
|
if err != nil {
|
|
return "", false, "", err
|
|
}
|
|
for _, id := range ids {
|
|
named := number > 0 && strings.HasPrefix(id, mergeAskPrefix(number, head))
|
|
if asker == sdkdelivery.TellerModule && !strings.HasPrefix(id, sdkdelivery.AskKindMerge+"-") {
|
|
continue
|
|
}
|
|
if asker == theDelivery && !named {
|
|
continue
|
|
}
|
|
ask, w, err := closedRecordOf(ctx, inv, asker, id)
|
|
if err != nil {
|
|
return "", false, "", err
|
|
}
|
|
if ask == nil || w == nil {
|
|
continue
|
|
}
|
|
warrant, err := caughtUpWarrant(asker, ask, w)
|
|
if err != nil {
|
|
if named {
|
|
reason = fmt.Sprintf("%s's ask %s holds no operator's Approve: %v", asker, id, err)
|
|
}
|
|
continue
|
|
}
|
|
how, covers := coversHead(ask, owner, repo, number, head)
|
|
if !covers && !named {
|
|
continue
|
|
}
|
|
if !covers {
|
|
how = "the ask's id names this pull request and head"
|
|
}
|
|
return warrant + ", " + how, true, "", nil
|
|
}
|
|
}
|
|
return "", false, reason, nil
|
|
}
|
|
|
|
// caughtTrunk is one trunk the catalogue builds from.
|
|
type caughtTrunk struct{ Owner, Repo, Branch string }
|
|
|
|
func (t caughtTrunk) repository() string { return t.Owner + "/" + t.Repo }
|
|
|
|
// trunksOfCatalogue are the trunks of the repositories on the mesh's forge the catalogue builds a module from.
|
|
func trunksOfCatalogue(forms *sourceForms, entries []inventory.Entry) []caughtTrunk {
|
|
var out []caughtTrunk
|
|
for _, e := range entries {
|
|
if e.Provided || e.Source.Repository == "" {
|
|
continue
|
|
}
|
|
owner, name, onForge := forms.onTheForge(e.Source.Repository, e.Source.Seat)
|
|
if !onForge {
|
|
continue
|
|
}
|
|
branch := followedBranch(e.Source.Ref)
|
|
if branch == "" {
|
|
branch = "main"
|
|
}
|
|
t := caughtTrunk{owner, name, branch}
|
|
if !slices.Contains(out, t) {
|
|
out = append(out, t)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// catchUpBeside runs a catch-up beside the hearing of the merge that asked for it, so the bus's acknowledgement of that merge
|
|
// waits for no forge. A variable so a test runs it where it stands.
|
|
var catchUpBeside = func(run func()) { go run() }
|
|
|
|
// catchingUp makes one catch-up at a time.
|
|
var catchingUp sync.Mutex
|
|
|
|
// settledMerges are the merges this process found nothing to plan for: heard, and no record kept, because nothing reads them.
|
|
// They are not read from the forge again.
|
|
var settledMerges sync.Map
|
|
|
|
// catchUp hears the merges of the catalogue's trunks (only one, when only names it) that the controller holds no record of.
|
|
// It says whether a trunk was left for later: the forge, the router or the store did not answer.
|
|
func (f following) catchUp(ctx context.Context, only *caughtTrunk) (later bool) {
|
|
catchingUp.Lock()
|
|
defer catchingUp.Unlock()
|
|
inv := f.open.inventory
|
|
entries, err := inv.Catalogued(ctx)
|
|
if err != nil {
|
|
fmt.Printf("catch-up: the catalogue cannot be read, so no merge is caught up now: %v\n", err)
|
|
return true
|
|
}
|
|
trunks := trunksOfCatalogue(newSourceForms(ctx, inv), entries)
|
|
if only != nil {
|
|
trunks = slices.DeleteFunc(trunks, func(t caughtTrunk) bool {
|
|
return !strings.EqualFold(t.repository(), only.repository()) || t.Branch != only.Branch
|
|
})
|
|
}
|
|
for _, t := range trunks {
|
|
if f.catchUpTrunk(ctx, t) {
|
|
later = true
|
|
}
|
|
}
|
|
return later
|
|
}
|
|
|
|
func (f following) catchUpTrunk(ctx context.Context, t caughtTrunk) (later bool) {
|
|
inv := f.open.inventory
|
|
repository := t.repository()
|
|
merges, err := readTrunkMerges(ctx, t.Owner, t.Repo, t.Branch)
|
|
if err != nil {
|
|
// Not nothing to catch up: left for the next start or the next heard merge (ADR 0457 clause 3).
|
|
fmt.Printf("catch-up: the forge did not answer for the last merges of %s %s, so they are not caught up now; the next start "+
|
|
"or the next merge heard there does it: %v\n", repository, t.Branch, err)
|
|
return true
|
|
}
|
|
unwarranted := map[string]bool{}
|
|
for _, tm := range merges {
|
|
id := fmt.Sprintf("%s_%.8s", repository, tm.Commit)
|
|
if _, done := settledMerges.Load(repository + "@" + tm.Commit); done {
|
|
continue
|
|
}
|
|
if _, known, err := inv.MergeOf(ctx, repository, tm.Commit); err != nil {
|
|
fmt.Printf("catch-up: the record of %s cannot be read, so %s is not caught up now: %v\n", repository, tm.Commit[:8], err)
|
|
return true
|
|
} else if known {
|
|
continue
|
|
}
|
|
m, err := readMergeEvent(ctx, t.Owner, t.Repo, t.Branch, tm)
|
|
if err != nil {
|
|
fmt.Printf("catch-up: the merge %.8s of %s cannot be read from the forge, so it is not caught up now: %v\n", tm.Commit, repository, err)
|
|
later = true
|
|
continue
|
|
}
|
|
entries, err := inv.Catalogued(ctx)
|
|
if err != nil {
|
|
return true
|
|
}
|
|
read, err := readForPlanning(ctx, inv)
|
|
if err != nil {
|
|
return true
|
|
}
|
|
facts, err := factsOfMerge(ctx, inv, m, time.Now().UTC(), entries, read)
|
|
if err != nil {
|
|
fmt.Printf("catch-up: %s %.8s cannot be judged now: %v\n", repository, tm.Commit, err)
|
|
later = true
|
|
continue
|
|
}
|
|
// The very test hearMerge makes: a merge it would plan, or take as a first build, is a gap; any other it would say
|
|
// nothing reads, and is no gap for a person to decide.
|
|
if !facts.plans() && !facts.addsModuleToBuilds {
|
|
settledMerges.Store(repository+"@"+tm.Commit, true)
|
|
continue
|
|
}
|
|
why, found, reason, err := findMergeWarrant(ctx, inv, t.Owner, t.Repo, m.Number, tm.Head)
|
|
if err != nil {
|
|
fmt.Printf("catch-up: the router cannot be read, so %s %.8s is not caught up now: %v\n", repository, tm.Commit, err)
|
|
later = true
|
|
continue
|
|
}
|
|
if !found {
|
|
unwarranted[id] = true
|
|
f.raiseMergeNotHeard(ctx, repository, t.Branch, tm, m, reason)
|
|
continue
|
|
}
|
|
now := time.Now().UTC()
|
|
said := fmt.Sprintf("caught up: merged %s, not heard, planned %s", tm.At.Format(time.RFC3339), now.Format(time.RFC3339))
|
|
fmt.Printf("catch-up: %s merged into %s (%.8s) was not heard; %s; on the warrant %s\n", repository, t.Branch, tm.Commit, said, why)
|
|
m.Title = strings.TrimSpace(m.Title + " (" + said + ")")
|
|
if err := f.hearMerge(ctx, m, now); err != nil {
|
|
fmt.Printf("catch-up: %s %.8s could not be heard now: %v\n", repository, tm.Commit, err)
|
|
later = true
|
|
continue
|
|
}
|
|
if _, known, err := inv.MergeOf(ctx, repository, tm.Commit); err == nil && !known {
|
|
settledMerges.Store(repository+"@"+tm.Commit, true) // heard, and nothing reads it
|
|
}
|
|
}
|
|
// Cleared only when every gap was judged: one left for later may still be open.
|
|
if !later {
|
|
f.clearMergesHeard(ctx, repository, unwarranted)
|
|
}
|
|
return later
|
|
}
|
|
|
|
// raiseMergeNotHeard keeps open, and observes again, the condition of a merge with no warrant to plan it. One for each
|
|
// repository and commit, however many catch-ups find it.
|
|
func (f following) raiseMergeNotHeard(ctx context.Context, repository, branch string, tm trunkMerge, m link.SourceMoved, reason string) {
|
|
fmt.Printf("catch-up: %s merged into %s (%.8s) was not heard, and is not planned: %s\n", repository, branch, tm.Commit, reason)
|
|
if conditionsFrom == nil {
|
|
return
|
|
}
|
|
o := conditions.Observation{Scope: conditions.ScopeMerge, ID: fmt.Sprintf("%s_%.8s", repository, tm.Commit), Kind: kindMergeNotHeard,
|
|
Severity: conditions.Warning, Source: "catch-up", Resolver: conditions.ResolverOperator,
|
|
Summary: fmt.Sprintf("the merge %.8s of %s into %s was not heard by the controller, and no operator's Approve of it is on "+
|
|
"record to plan it on: a person decides", tm.Commit, repository, branch),
|
|
Said: fmt.Sprintf("%s %.8s (pull request %d, head %.8s, merged %s): %s", repository, tm.Commit, m.Number, tm.Head,
|
|
tm.At.Format(time.RFC3339), reason),
|
|
Headline: fmt.Sprintf("A merge into %s was never delivered", repository),
|
|
Explanation: fmt.Sprintf("A merge into %s was made while the controller did not hear it, and no Approve of yours for it is on "+
|
|
"record that the controller can plan it on. It stays open while the merge is among the latest %d merges of that repository. "+
|
|
"The details say what was looked for.", repository, catchUpWindow),
|
|
Needs: "decide whether that merge should be delivered, and say so if it should.",
|
|
Resolved: "the merge is planned, or is no longer among the latest merges of its repository"}
|
|
if _, err := conditionsFrom.Observe(ctx, o); err != nil {
|
|
fmt.Printf("catch-up: the condition of %s %.8s could not be kept: %v\n", repository, tm.Commit, err)
|
|
}
|
|
}
|
|
|
|
// clearMergesHeard clears the open conditions of a repository's merges not heard that this catch-up did not find again: the
|
|
// gap was planned, or left the window.
|
|
func (f following) clearMergesHeard(ctx context.Context, repository string, still map[string]bool) {
|
|
if conditionsFrom == nil {
|
|
return
|
|
}
|
|
open, err := conditionsFrom.Open(ctx)
|
|
if err != nil {
|
|
fmt.Printf("catch-up: the open conditions cannot be read, so none is cleared now: %v\n", err)
|
|
return
|
|
}
|
|
for _, c := range open {
|
|
if c.Kind != kindMergeNotHeard || c.Subject.Scope != conditions.ScopeMerge ||
|
|
!strings.HasPrefix(c.Subject.ID, repository+"_") || still[c.Subject.ID] {
|
|
continue
|
|
}
|
|
if _, err := conditionsFrom.Clear(ctx, c.Key, "the merge is planned, or left the last merges of its trunk"); err != nil {
|
|
fmt.Printf("catch-up: the condition %s could not be cleared: %v\n", c.Key, err)
|
|
}
|
|
}
|
|
}
|
|
|
|
// catchUpAtStart catches up every trunk once the controller serves, and again every minute while a trunk was left for later.
|
|
func catchUpAtStart(ctx context.Context, open *stores) {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(30 * time.Second):
|
|
}
|
|
for try := 0; try < 20; try++ {
|
|
if !(following{open: open}).catchUp(ctx, nil) {
|
|
return
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-time.After(time.Minute):
|
|
}
|
|
}
|
|
fmt.Println("catch-up: a trunk is still left for later after twenty tries; the next merge heard there does it")
|
|
}
|