diff --git a/cmd/mesh-controller/catch_up.go b/cmd/mesh-controller/catch_up.go new file mode 100644 index 00000000..01e827dc --- /dev/null +++ b/cmd/mesh-controller/catch_up.go @@ -0,0 +1,537 @@ +package main + +import ( + "context" + "errors" + "fmt" + "net/url" + "path" + "regexp" + "slices" + "sort" + "strconv" + "strings" + "sync" + "time" + + "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 { + if c.SecondParent != "" { + 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 +} + +// 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 of the delivery begins with when it names a pull request and its head: the +// number and the head's first twelve digits (mesh-delivery's `merge---`). +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). What it cannot re-prove is the heads the Approve was bound to: the +// router's record does not carry them, and the ask's id names the pull request and the head merged. +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 + } + 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+".")) + } + if err := lister.Stop(); err != nil { + return nil, err + } + return ids, nil +} + +// findMergeWarrant finds the warrant of a merge: the router's chosen merge ask whose id names the pull request and its head, +// under an asker the controller takes a warrant from. 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, number int, head string) (why string, found bool, reason string, err error) { + if number <= 0 { + return "", false, "the merge commit names no pull request, so no ask of the router can name it", nil + } + prefix := mergeAskPrefix(number, head) + reason = fmt.Sprintf("the router holds no merge ask %s* of %s", prefix, strings.Join(askersOfMerges(), " or ")) + for _, asker := range askersOfMerges() { + ids, err := routerAskIDs(ctx, inv, asker) + if err != nil { + return "", false, "", err + } + for _, id := range ids { + if !strings.HasPrefix(id, prefix) { + continue + } + ask, w, err := routerAskOf(ctx, inv, asker, id) + if err != nil { + return "", false, "", err + } + if why, err := caughtUpWarrant(asker, ask, w); err == nil { + return why, true, "", nil + } else { + reason = fmt.Sprintf("%s's ask %s holds no operator's Approve: %v", asker, id, err) + } + } + } + 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 + } + if !wouldBePlanned(ctx, inv, m, entries, read) { + settledMerges.Store(repository+"@"+tm.Commit, true) + continue + } + why, found, reason, err := findMergeWarrant(ctx, inv, 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 +} + +// wouldBePlanned says hearing the merge would plan something: it touches a module the mesh holds and is not history for every +// one of them (or is owed, as hearMerge judges). A merge that moves nothing is no gap for a person to decide. +func wouldBePlanned(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, entries []inventory.Entry, + read map[string][]inventory.ReadRepository) bool { + raw := m + raw.MergedAt = "" + touches := wouldMove(raw, entries, read) + if len(touches) == 0 { + return false + } + from, packaging, _ := mergeCandidates(m, entries, read) + if len(from) > 0 || len(packaging) > 0 { + return true + } + merged, err := time.Parse(time.RFC3339, m.MergedAt) + if err != nil { + return true + } + owed, err := owedLate(ctx, inv, m, merged, touches) + return err != nil || owed +} + +// 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("The merge %.8s into %s was made while the controller did not hear it, and the router holds no "+ + "Approve of yours for it that the controller can plan it on: %s. It stays open while the merge is among the last %d "+ + "of the trunk.", tm.Commit, repository, reason, catchUpWindow), + Needs: "decide whether it is delivered: build its modules by hand, or leave it", + Resolved: "the merge is planned, or is no longer among the last merges of its trunk"} + 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") +}