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 } // sinceTheRecord are the merges that can be gaps (novox/hq ADR 0463 §2): those made since the controller's record of merges begins on // the mesh (first, the earliest merge it holds of any repository and branch, less mergeGrace as owedLate judges). A merge before it is // history on every trunk, and a trunk with no merge kept of its own still has the merges made since as gaps. With no merge kept // anywhere there is no record to have a gap in. func sinceTheRecord(merges []trunkMerge, kept bool, first time.Time) []trunkMerge { if !kept { return nil } var out []trunkMerge for _, m := range merges { if !m.At.Before(first.Add(-mergeGrace)) { out = append(out, m) } } return out } // 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---`). 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 ("# ... at :"), 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---`, 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 } kept, first, err := inv.EarliestKept(ctx) if err != nil { fmt.Printf("catch-up: what the controller kept cannot be read, so no merge is caught up now: %v\n", err) return true } if kept { fmt.Printf("catch-up: the controller's record of merges begins %s; a merge made before it, less %s, is history on every trunk\n", first.UTC().Format(time.RFC3339), mergeGrace) } else { fmt.Println("catch-up: the controller holds no merge of any trunk yet, so none can have been missed") } 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, kept, first) { later = true } } return later } func (f following) catchUpTrunk(ctx context.Context, t caughtTrunk, kept bool, first time.Time) (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 } // **A merge made before the controller's record of merges begins is history, never a gap** (novox/hq ADR 0463 §2). read := len(merges) merges = sinceTheRecord(merges, kept, first) fmt.Printf("catch-up: %s %s: %d of the last %d pull request merges were made since the record began\n", repository, t.Branch, len(merges), read) 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") }