Merge pull request 'Catch up merges the controller did not hear, on the operator's Approve of each (ADR 0457, issue 1336)' (#407) from fix/1336-catch-up-merges-not-heard into main
This commit was merged in pull request #407.
This commit is contained in:
@@ -124,6 +124,52 @@ func durationSetting(v any) (time.Duration, bool) {
|
||||
return 0, false
|
||||
}
|
||||
|
||||
// mergeFacts is what hearMerge judges a merge by, kept apart so the catch-up (ADR 0457) judges a merge the way hearing it will.
|
||||
type mergeFacts struct {
|
||||
merged time.Time
|
||||
touches []inventory.Entry
|
||||
later []inventory.BatchedMerge
|
||||
owed bool
|
||||
from, packaging []inventory.Entry
|
||||
already int
|
||||
addsModuleToBuilds bool // a first merge into a repository the mesh builds nothing from: firstBuilds acts on it
|
||||
}
|
||||
|
||||
// factsOfMerge reads what hearing a merge turns on: the modules it touches whatever the history, the later merges of its
|
||||
// branch, whether it is owed late, and the modules it moves.
|
||||
func factsOfMerge(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, now time.Time,
|
||||
entries []inventory.Entry, read map[string][]inventory.ReadRepository) (mergeFacts, error) {
|
||||
repository := m.Owner + "/" + m.Repo
|
||||
f := mergeFacts{}
|
||||
var err error
|
||||
f.merged, err = time.Parse(time.RFC3339Nano, m.MergedAt)
|
||||
if err != nil {
|
||||
f.merged = now
|
||||
}
|
||||
// **What it touches, history or not** (review of this change): a merge heard after a later merge of its
|
||||
// branch was cut reads as history for every module that cut marked seen, and was dropped from the record
|
||||
// here, never named by the walk that carried it.
|
||||
raw := m
|
||||
raw.MergedAt = ""
|
||||
f.touches = wouldMove(raw, entries, read)
|
||||
if len(f.touches) > 0 {
|
||||
if f.later, err = inv.LaterMergesOf(ctx, repository, m.Base, f.merged.UTC()); err != nil {
|
||||
return f, err
|
||||
}
|
||||
}
|
||||
if f.owed, err = owedLate(ctx, inv, m, f.merged, f.touches); err != nil {
|
||||
return f, err
|
||||
}
|
||||
f.from, f.packaging, f.already = mergeCandidates(m, entries, read)
|
||||
f.addsModuleToBuilds = f.already == 0 && len(addedDirs(m, entries, read)) > 0
|
||||
return f, nil
|
||||
}
|
||||
|
||||
// plans says hearing the merge puts it in a batch or a walk.
|
||||
func (f mergeFacts) plans() bool {
|
||||
return len(f.from) > 0 || len(f.packaging) > 0 || len(f.later) > 0 || f.owed
|
||||
}
|
||||
|
||||
// hearMerge puts a merge the forge announced into the batch that answers it, at the moment it was heard,
|
||||
// and cuts what is due.
|
||||
func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.Time) error {
|
||||
@@ -148,27 +194,13 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
|
||||
if err != nil {
|
||||
return notNow(err)
|
||||
}
|
||||
merged, err := time.Parse(time.RFC3339Nano, m.MergedAt)
|
||||
if err != nil {
|
||||
merged = now
|
||||
}
|
||||
// **What it touches, history or not** (review of this change): a merge heard after a later merge of its
|
||||
// branch was cut reads as history for every module that cut marked seen, and was dropped from the record
|
||||
// here, never named by the walk that carried it.
|
||||
raw := m
|
||||
raw.MergedAt = ""
|
||||
touches := wouldMove(raw, entries, read)
|
||||
var later []inventory.BatchedMerge
|
||||
if len(touches) > 0 {
|
||||
if later, err = inv.LaterMergesOf(ctx, repository, m.Base, merged.UTC()); err != nil {
|
||||
return notNow(err)
|
||||
}
|
||||
}
|
||||
owed, err := owedLate(ctx, inv, m, merged, touches)
|
||||
facts, err := factsOfMerge(ctx, inv, m, now, entries, read)
|
||||
if err != nil {
|
||||
return notNow(err)
|
||||
}
|
||||
if from, packaging, already := mergeCandidates(m, entries, read); len(from) == 0 && len(packaging) == 0 && len(later) == 0 && !owed {
|
||||
merged, touches := facts.merged, facts.touches
|
||||
if !facts.plans() {
|
||||
already := facts.already
|
||||
// "Already built from it" and "nothing reads it" are different facts, and reading the first
|
||||
// as the second sends somebody looking for a broken trigger when the mesh is up to date.
|
||||
if already > 0 {
|
||||
|
||||
@@ -0,0 +1,597 @@
|
||||
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 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
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
// 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")
|
||||
}
|
||||
@@ -0,0 +1,291 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// The checks of ADR 0457 (issue 1336): what the controller does about merges it did not hear.
|
||||
|
||||
// aTrunk is a forge's trunk as the catch-up reads it: its merges oldest first, the module each moves, and which of them the
|
||||
// router holds an operator's Approve for.
|
||||
type aTrunk struct {
|
||||
merges []trunkMerge
|
||||
module map[string]string
|
||||
approved map[string]bool
|
||||
read []string // the trunks asked of the forge, as owner/repo
|
||||
events int // merges read in full from the forge
|
||||
silent bool // the forge does not answer
|
||||
}
|
||||
|
||||
func aTrunkOf(t *testing.T, repo string, modules ...string) *aTrunk {
|
||||
t.Helper()
|
||||
tr := &aTrunk{module: map[string]string{}, approved: map[string]bool{}}
|
||||
for i, m := range modules {
|
||||
c := fmt.Sprintf("%s-c%02d", repo, i+1)
|
||||
tr.merges = append(tr.merges, trunkMerge{Commit: c, Head: fmt.Sprintf("%shead%02d", repo, i+1),
|
||||
Subject: fmt.Sprintf("Merge pull request 'change %d' (#%d) from pr/%s into main", i+1, 100+i, c),
|
||||
At: t0.Add(time.Duration(i) * time.Minute)})
|
||||
tr.module[c] = m
|
||||
tr.approved[c] = true
|
||||
}
|
||||
was, wasEvent, wasWarrant := trunkCommitsPage, readMergeEvent, findMergeWarrant
|
||||
trunkCommitsPage = func(_ context.Context, owner, name, _ string, page int) ([]trunkCommit, error) {
|
||||
tr.read = append(tr.read, owner+"/"+name)
|
||||
if tr.silent {
|
||||
return nil, errors.New("the forge did not answer")
|
||||
}
|
||||
if page > 1 || owner+"/"+name != "novox/"+repo {
|
||||
return nil, nil
|
||||
}
|
||||
var newest []trunkCommit
|
||||
for i := len(tr.merges) - 1; i >= 0; i-- {
|
||||
m := tr.merges[i]
|
||||
newest = append(newest, trunkCommit{Commit: m.Commit, SecondParent: m.Head, Subject: m.Subject, At: m.At})
|
||||
}
|
||||
return newest, nil
|
||||
}
|
||||
readMergeEvent = func(_ context.Context, owner, name, branch string, tm trunkMerge) (link.SourceMoved, error) {
|
||||
tr.events++
|
||||
m := catalogueMerge(tm.Commit, tr.module[tm.Commit], tm.At)
|
||||
m.Owner, m.Repo, m.Base, m.HeadSHA = owner, name, branch, tm.Head
|
||||
_, m.Number, _ = pullOfSubject(tm.Subject)
|
||||
return m, nil
|
||||
}
|
||||
findMergeWarrant = func(_ context.Context, _ catalogueReader, _, _ string, number int, head string) (string, bool, string, error) {
|
||||
for _, m := range tr.merges {
|
||||
if strings.Contains(m.Subject, fmt.Sprintf("(#%d)", number)) && m.Head == head && tr.approved[m.Commit] {
|
||||
return "the operator chose Approve (ask merge-" + fmt.Sprint(number) + ")", true, "", nil
|
||||
}
|
||||
}
|
||||
return "", false, "the router holds no chosen merge ask for the pull request and head", nil
|
||||
}
|
||||
wasBeside := catchUpBeside
|
||||
catchUpBeside = func(run func()) { run() }
|
||||
t.Cleanup(func() {
|
||||
trunkCommitsPage, readMergeEvent, findMergeWarrant, catchUpBeside = was, wasEvent, wasWarrant, wasBeside
|
||||
settledMerges.Clear()
|
||||
})
|
||||
return tr
|
||||
}
|
||||
|
||||
func mergeKnown(t *testing.T, open *stores, commit string) bool {
|
||||
t.Helper()
|
||||
_, known, err := open.inventory.MergeOf(t.Context(), "novox/mesh-catalog", commit)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return known
|
||||
}
|
||||
|
||||
// Clause 1: the trunk holds merges 1, 2 and 3 and the controller holds records of 1 and 3 only; at its start it plans 2, said
|
||||
// as caught up; a second start plans nothing new.
|
||||
func TestAGapBeforeANewerHeardMergeIsPlannedAtTheControllersStart(t *testing.T) {
|
||||
open := windowed(t)
|
||||
everyHeadWaits(t)
|
||||
asksRecorded(t)
|
||||
tr := aTrunkOf(t, "mesh-catalog", "app", "notes", "app")
|
||||
hear(t, open, catalogueMerge("mesh-catalog-c01", "app", t0), t0.Add(time.Second))
|
||||
hear(t, open, catalogueMerge("mesh-catalog-c03", "app", t0.Add(2*time.Minute)), t0.Add(2*time.Minute+time.Second))
|
||||
if mergeKnown(t, open, "mesh-catalog-c02") {
|
||||
t.Fatal("the merge to be caught up is known before the catch-up")
|
||||
}
|
||||
if (following{open: open}).catchUp(t.Context(), nil) {
|
||||
t.Fatal("the catch-up says a trunk was left for later")
|
||||
}
|
||||
if !mergeKnown(t, open, "mesh-catalog-c02") {
|
||||
t.Fatal("the merge the controller did not hear was not planned")
|
||||
}
|
||||
kept, _, _ := open.inventory.MergeOf(t.Context(), "novox/mesh-catalog", "mesh-catalog-c02")
|
||||
if said := announcedOf(kept).Title; !strings.Contains(said, "caught up: merged ") || !strings.Contains(said, "not heard, planned ") {
|
||||
t.Fatalf("the merge does not say it was caught up: %q", said)
|
||||
}
|
||||
cutAt(t, open, t0.Add(10*time.Minute))
|
||||
ws, _ := walks(t, open)
|
||||
if len(ws) == 0 {
|
||||
t.Fatal("no walk answers the caught-up merge")
|
||||
}
|
||||
if _, in := ws[0].Modules["notes"]; !in {
|
||||
t.Fatalf("the walk does not build notes, which the caught-up merge moved: %v", ws[0].Modules)
|
||||
}
|
||||
before := tr.events
|
||||
if (following{open: open}).catchUp(t.Context(), nil) || tr.events != before {
|
||||
t.Fatalf("a second start read %d merge(s) in full again", tr.events-before)
|
||||
}
|
||||
}
|
||||
|
||||
// Clause 1: a heard merge of a repository whose trunk holds earlier merges the controller has no record of plans them all, the
|
||||
// earlier first, and the forge is read for that repository's trunk only.
|
||||
func TestAHeardMergeCatchesUpTheEarlierOnesOfItsOwnTrunkOnly(t *testing.T) {
|
||||
open := windowed(t)
|
||||
everyHeadWaits(t)
|
||||
asksRecorded(t)
|
||||
tr := aTrunkOf(t, "mesh-catalog", "app", "notes", "app")
|
||||
var order []string
|
||||
inner := readMergeEvent
|
||||
readMergeEvent = func(ctx context.Context, owner, name, branch string, tm trunkMerge) (link.SourceMoved, error) {
|
||||
order = append(order, tm.Commit)
|
||||
return inner(ctx, owner, name, branch, tm)
|
||||
}
|
||||
t.Cleanup(func() { readMergeEvent = inner })
|
||||
tr.read = nil
|
||||
heard := catalogueMerge("mesh-catalog-c03", "app", t0.Add(2*time.Minute))
|
||||
if err := (following{open: open}).SourceMoved(t.Context(), heard); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, c := range []string{"mesh-catalog-c01", "mesh-catalog-c02", "mesh-catalog-c03"} {
|
||||
if !mergeKnown(t, open, c) {
|
||||
t.Fatalf("%s is not planned: the heard merge and the earlier ones of its trunk are all wanted", c)
|
||||
}
|
||||
}
|
||||
if len(order) != 2 || order[0] != "mesh-catalog-c01" || order[1] != "mesh-catalog-c02" {
|
||||
t.Fatalf("the earlier merges were heard in the order %v, not the trunk's", order)
|
||||
}
|
||||
if len(tr.read) == 0 {
|
||||
t.Fatal("the forge was not read for the trunk of the merge heard")
|
||||
}
|
||||
for _, r := range tr.read {
|
||||
if r != "novox/mesh-catalog" {
|
||||
t.Fatalf("the forge was read for %s, not only for the trunk of the merge heard: %v", r, tr.read)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Clause 1: a trunk of fifty-one merges whose oldest is unheard plans nothing: the oldest is outside the window.
|
||||
func TestAMergeOlderThanTheLastFiftyIsNotRead(t *testing.T) {
|
||||
open := windowed(t)
|
||||
everyHeadWaits(t)
|
||||
asksRecorded(t)
|
||||
modules := make([]string, 51)
|
||||
for i := range modules {
|
||||
modules[i] = "app"
|
||||
}
|
||||
aTrunkOf(t, "mesh-catalog", modules...)
|
||||
// Fifty are known; the oldest is not.
|
||||
for i := 2; i <= 51; i++ {
|
||||
c := fmt.Sprintf("mesh-catalog-c%02d", i)
|
||||
hear(t, open, catalogueMerge(c, "app", t0.Add(time.Duration(i)*time.Minute)), t0.Add(time.Duration(i)*time.Minute+time.Second))
|
||||
}
|
||||
// Paged as the forge pages: fifty a page, and a trunk of another repository is empty (a short page, or none).
|
||||
pages := trunkCommitsPage
|
||||
trunkCommitsPage = func(ctx context.Context, owner, name, branch string, page int) ([]trunkCommit, error) {
|
||||
all, err := pages(ctx, owner, name, branch, 1)
|
||||
if err != nil || page < 1 || (page-1)*50 >= len(all) {
|
||||
return nil, err
|
||||
}
|
||||
return all[(page-1)*50 : min(page*50, len(all))], nil
|
||||
}
|
||||
t.Cleanup(func() { trunkCommitsPage = pages })
|
||||
(following{open: open}).catchUp(t.Context(), nil)
|
||||
if mergeKnown(t, open, "mesh-catalog-c01") {
|
||||
t.Fatal("the oldest of fifty-one merges was read and planned, outside the window")
|
||||
}
|
||||
}
|
||||
|
||||
// Clause 2: a gap with no chosen Approve in the router is not planned and raises one condition naming its repository and
|
||||
// commit; a second start leaves that one condition open, observed twice; planning the gap clears it.
|
||||
func TestAGapWithNoApproveRaisesOneConditionKeptOpenAndClearedWhenPlanned(t *testing.T) {
|
||||
open := windowed(t)
|
||||
everyHeadWaits(t)
|
||||
asksRecorded(t)
|
||||
keeper, _ := withConditionsInMemory(t)
|
||||
tr := aTrunkOf(t, "mesh-catalog", "notes")
|
||||
tr.approved["mesh-catalog-c01"] = false
|
||||
catch := func() { (following{open: open}).catchUp(t.Context(), nil) }
|
||||
catch()
|
||||
if mergeKnown(t, open, "mesh-catalog-c01") {
|
||||
t.Fatal("a merge with no Approve on record was planned")
|
||||
}
|
||||
raised := func() []conditions.Condition {
|
||||
t.Helper()
|
||||
all, err := keeper.Open(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var out []conditions.Condition
|
||||
for _, c := range all {
|
||||
if c.Kind == kindMergeNotHeard {
|
||||
out = append(out, c)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
one := raised()
|
||||
if len(one) != 1 || !strings.Contains(one[0].Subject.ID, "novox/mesh-catalog") || !strings.Contains(one[0].Subject.ID, "mesh-cata") {
|
||||
t.Fatalf("the gap raised %+v, not one condition naming its repository and commit", one)
|
||||
}
|
||||
catch()
|
||||
if again := raised(); len(again) != 1 || again[0].Observations < 2 {
|
||||
t.Fatalf("a second catch-up left %+v: one condition, observed twice, was wanted", again)
|
||||
}
|
||||
tr.approved["mesh-catalog-c01"] = true
|
||||
catch()
|
||||
if !mergeKnown(t, open, "mesh-catalog-c01") {
|
||||
t.Fatal("the gap was not planned once its Approve was on record")
|
||||
}
|
||||
if left := raised(); len(left) != 0 {
|
||||
t.Fatalf("planning the gap left its condition open: %+v", left)
|
||||
}
|
||||
}
|
||||
|
||||
// Clause 3: a forge that does not answer leaves the catch-up undone and says so, and the next start does it.
|
||||
func TestAForgeThatDoesNotAnswerLeavesTheCatchUpForTheNextStart(t *testing.T) {
|
||||
open := windowed(t)
|
||||
everyHeadWaits(t)
|
||||
asksRecorded(t)
|
||||
tr := aTrunkOf(t, "mesh-catalog", "notes")
|
||||
tr.silent = true
|
||||
if later := (following{open: open}).catchUp(t.Context(), nil); !later {
|
||||
t.Fatal("a forge that did not answer counted as nothing to catch up")
|
||||
}
|
||||
if mergeKnown(t, open, "mesh-catalog-c01") {
|
||||
t.Fatal("a merge was planned without the forge's word")
|
||||
}
|
||||
tr.silent = false
|
||||
if later := (following{open: open}).catchUp(t.Context(), nil); later || !mergeKnown(t, open, "mesh-catalog-c01") {
|
||||
t.Fatalf("the next start did not catch up (left for later: %v)", later)
|
||||
}
|
||||
}
|
||||
|
||||
// Finding 2 of the review: a gap that adds a module is a gap, judged as hearMerge judges it. With the operator's Approve on record it
|
||||
// is heard (hearMerge asks the new module in its walk); without, it raises the condition and is not lost in silence.
|
||||
func TestAGapThatAddsAModuleIsHeardOnItsWarrantAndRaisedWithout(t *testing.T) {
|
||||
open := windowed(t)
|
||||
everyHeadWaits(t)
|
||||
asksRecorded(t)
|
||||
keeper, _ := withConditionsInMemory(t)
|
||||
tr := aTrunkOf(t, "mesh-catalog", "newmodule")
|
||||
tr.approved["mesh-catalog-c01"] = false
|
||||
catch := func() { (following{open: open}).catchUp(t.Context(), nil) }
|
||||
catch()
|
||||
if mergeKnown(t, open, "mesh-catalog-c01") {
|
||||
t.Fatal("a merge adding a module was heard with no Approve on record")
|
||||
}
|
||||
all, err := keeper.Open(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var raised int
|
||||
for _, c := range all {
|
||||
if c.Kind == kindMergeNotHeard {
|
||||
raised++
|
||||
}
|
||||
}
|
||||
if raised != 1 {
|
||||
t.Fatalf("a merge adding a module with no Approve raised %d condition(s), not one: it was lost in silence", raised)
|
||||
}
|
||||
tr.approved["mesh-catalog-c01"] = true
|
||||
catch()
|
||||
if !mergeKnown(t, open, "mesh-catalog-c01") {
|
||||
t.Fatal("a merge adding a module was not heard once its Approve was on record")
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,294 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"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/link"
|
||||
)
|
||||
|
||||
// The pieces of the catch-up that need no store (novox/hq ADR 0457, issue 1336).
|
||||
|
||||
func TestAMergeCommitsSubjectNamesItsPullRequest(t *testing.T) {
|
||||
title, number, ok := pullOfSubject("Merge pull request 'Fix keycloak's sign-in (issue 1332)' (#378) from fix/1332-sign-in into main")
|
||||
if !ok || number != 378 || title != "Fix keycloak's sign-in (issue 1332)" {
|
||||
t.Fatalf("read %q %d %v", title, number, ok)
|
||||
}
|
||||
if b := headBranchOfSubject("Merge pull request 'x' (#378) from fix/1332-sign-in into main"); b != "fix/1332-sign-in" {
|
||||
t.Fatalf("the head branch read as %q", b)
|
||||
}
|
||||
if _, _, ok := pullOfSubject("Merge branch 'main' of somewhere"); ok {
|
||||
t.Fatal("a merge that names no pull request read as one")
|
||||
}
|
||||
}
|
||||
|
||||
func TestOnlyTheLastFiftyMergesOfATrunkAreRead(t *testing.T) {
|
||||
pages := map[int][]trunkCommit{}
|
||||
var all []trunkCommit // newest first: 51 merges, a plain commit between each pair
|
||||
for i := 51; i >= 1; i-- {
|
||||
all = append(all, trunkCommit{Commit: fmt.Sprintf("merge-%02d", i), SecondParent: fmt.Sprintf("head-%02d", i)},
|
||||
trunkCommit{Commit: fmt.Sprintf("plain-%02d", i)})
|
||||
}
|
||||
for i := 0; i < len(all); i += 50 {
|
||||
pages[i/50+1] = all[i:min(i+50, len(all))]
|
||||
}
|
||||
was := trunkCommitsPage
|
||||
trunkCommitsPage = func(_ context.Context, _, _, _ string, page int) ([]trunkCommit, error) { return pages[page], nil }
|
||||
t.Cleanup(func() { trunkCommitsPage = was })
|
||||
got, err := readTrunkMerges(t.Context(), "novox", "mesh-catalog", "main")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(got) != catchUpWindow || got[0].Commit != "merge-02" || got[len(got)-1].Commit != "merge-51" {
|
||||
t.Fatalf("read %d merges from %s to %s: the oldest of 51 is outside the window", len(got), got[0].Commit, got[len(got)-1].Commit)
|
||||
}
|
||||
if got[0].Head != "head-02" {
|
||||
t.Fatalf("a merge's head is its second parent, not %q", got[0].Head)
|
||||
}
|
||||
}
|
||||
|
||||
func TestAForgeThatDoesNotAnswerIsAnErrorNotAnEmptyTrunk(t *testing.T) {
|
||||
was := trunkCommitsPage
|
||||
trunkCommitsPage = func(context.Context, string, string, string, int) ([]trunkCommit, error) {
|
||||
return nil, errors.New("the forge did not answer")
|
||||
}
|
||||
t.Cleanup(func() { trunkCommitsPage = was })
|
||||
if got, err := readTrunkMerges(t.Context(), "novox", "mesh-catalog", "main"); err == nil {
|
||||
t.Fatalf("a forge that did not answer read as %d merges: nothing to catch up", len(got))
|
||||
}
|
||||
}
|
||||
|
||||
func TestACaughtUpMergeIsAcceptedOnTheSameWarrantAsAnAnnouncedOne(t *testing.T) {
|
||||
ok := approvedMerge(t)
|
||||
for _, asker := range []string{sdkdelivery.TellerModule, theDelivery} {
|
||||
w := ok.warrant
|
||||
w.Asker = asker
|
||||
if _, err := caughtUpWarrant(asker, &ok.ask, &w); err != nil {
|
||||
t.Errorf("the operator's Approve under %s was refused: %v", asker, err)
|
||||
}
|
||||
}
|
||||
if _, err := caughtUpWarrant("keycloak", &ok.ask, &ok.warrant); err == nil {
|
||||
t.Error("a warrant of an asker the controller takes none from was accepted")
|
||||
}
|
||||
declined := ok.warrant
|
||||
declined.Option, declined.Label = "decline", "Decline"
|
||||
if _, err := caughtUpWarrant(theDelivery, &ok.ask, &declined); err == nil {
|
||||
t.Error("a Decline was accepted as a warrant")
|
||||
}
|
||||
unproved := ok.warrant
|
||||
by := *unproved.By
|
||||
by.Verified = ""
|
||||
unproved.By = &by
|
||||
if _, err := caughtUpWarrant(theDelivery, &ok.ask, &unproved); err == nil {
|
||||
t.Error("an answer whose channel proves nobody was accepted")
|
||||
}
|
||||
if _, err := caughtUpWarrant(theDelivery, nil, nil); err == nil {
|
||||
t.Error("a router that holds nothing was accepted")
|
||||
}
|
||||
}
|
||||
|
||||
// aTellersAsk is the ask the delivery gate composes for a pull request, as the router holds it once the operator chose Approve
|
||||
// (the gate's composeMergeAsk: its id from sdkdelivery.AskingID of the group and the heads, its Approve bound to mergeAct over the
|
||||
// heads, the pull request and its head's first twelve digits in the details), with the warrant the router wrote.
|
||||
func aTellersAsk(t *testing.T, group string, members ...sdkdelivery.Head) (asks.Ask, asks.Warrant) {
|
||||
t.Helper()
|
||||
key, err := sdkdelivery.MergeAskKey(group, members)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
id := sdkdelivery.AskingID(key, 1)
|
||||
heads := map[string]string{}
|
||||
var details []string
|
||||
for _, m := range members {
|
||||
owner, repo, _ := strings.Cut(m.Repository, "/")
|
||||
heads[deliveryIDOf(owner, repo, m.Head)] = m.Head
|
||||
details = append(details, fmt.Sprintf("%s#%d %q at %s: issues: none named; review: approved; reaches: no delivery plan yet", repo, m.Number, "title", m.Head[:12]))
|
||||
}
|
||||
bind := func(option string) string {
|
||||
d, err := asks.ActDigest(mergeAct(id, option, heads))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return d
|
||||
}
|
||||
at := time.Date(2026, 10, 20, 12, 2, 1, 0, time.UTC)
|
||||
ask := asks.Ask{ID: id, Headline: "Merge into main", Explanation: "passed its checks.", Who: asks.Operator, Expires: at.Add(24 * time.Hour),
|
||||
OnExpiry: "nothing is merged; the ask is made again", Details: strings.Join(details, "\n"),
|
||||
Options: []asks.Option{
|
||||
{ID: optionApprove, Label: "Approve", Does: "merges", Level: asks.Approve, Binds: bind(optionApprove)},
|
||||
{ID: "decline", Label: "Decline", Does: "nothing is merged", Level: asks.Approve, Binds: bind("decline")}}}
|
||||
w := asks.Warrant{Ask: id, Asker: sdkdelivery.TellerModule, Outcome: asks.OutcomeChosen, Option: optionApprove, Label: "Approve",
|
||||
Level: asks.Approve, By: &asks.Person{Who: asks.Operator, Kind: "telegram", Identity: "1", Verified: "user id"},
|
||||
At: at.Add(time.Minute), AskDigest: ask.Digest()}
|
||||
return ask, w
|
||||
}
|
||||
|
||||
// withRouter hands findMergeWarrant these asks as the router's records, by asker.
|
||||
func withRouter(t *testing.T, records map[string]map[string]struct {
|
||||
ask asks.Ask
|
||||
w asks.Warrant
|
||||
}) {
|
||||
t.Helper()
|
||||
wasIDs, wasAsk := routerAskIDs, routerAskOf
|
||||
closedRecords.Range(func(k, _ any) bool { closedRecords.Delete(k); return true })
|
||||
routerAskIDs = func(_ context.Context, _ catalogueReader, asker string) ([]string, error) {
|
||||
var ids []string
|
||||
for id := range records[asker] {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
return ids, nil
|
||||
}
|
||||
routerAskOf = func(_ context.Context, _ catalogueReader, asker, id string) (*asks.Ask, *asks.Warrant, error) {
|
||||
r, ok := records[asker][id]
|
||||
if !ok {
|
||||
return nil, nil, nil
|
||||
}
|
||||
return &r.ask, &r.w, nil
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
routerAskIDs, routerAskOf = wasIDs, wasAsk
|
||||
closedRecords.Range(func(k, _ any) bool { closedRecords.Delete(k); return true })
|
||||
})
|
||||
}
|
||||
|
||||
type aRouterRecord = struct {
|
||||
ask asks.Ask
|
||||
w asks.Warrant
|
||||
}
|
||||
|
||||
// The teller's asks, as the router held them on 2026-10-20: the id is the router's own key, rebuilt here from the group and the
|
||||
// head the gate composed it for (mesh-controller#405 at 8a330afd, mesh-catalog#381 at 1c702422, mesh-host#124 at 52600ec7). It names
|
||||
// neither the pull request nor the head, so the warrant is found by the heads the Approve is bound to.
|
||||
func TestATellersWarrantIsFoundByTheHeadItsApproveIsBoundTo(t *testing.T) {
|
||||
real := []struct {
|
||||
group, key, owner, repo string
|
||||
number int
|
||||
head string
|
||||
}{
|
||||
{"feat/r4b-sdk-users", "merge-feat_r4b_sdk_users-ba59b03220f8-989afe0a3bcf", "novox", "mesh-controller", 405, "8a330afd44ba852873339cf1210e90989dca46e1"},
|
||||
{"fix/1351-remove-todays-delivery-module", "merge-fix_1351_remove_todays_delivery_module-64a353-d93b97b9697b", "novox", "mesh-catalog", 381, "1c7024224dd77c5b603299378883e0993b2e2fdd"},
|
||||
{"feat/r4b-sdk-users", "merge-feat_r4b_sdk_users-6a59ba48da14-424e053ebc9c", "novox", "mesh-host", 124, "52600ec7bd61ee271a5cc7f4136bcf0a2a8f00fc"},
|
||||
}
|
||||
teller := map[string]aRouterRecord{}
|
||||
for _, r := range real {
|
||||
ask, w := aTellersAsk(t, r.group, sdkdelivery.Head{Repository: r.owner + "/" + r.repo, Number: r.number, Head: r.head})
|
||||
if ask.ID != r.key {
|
||||
t.Fatalf("the id of %s#%d is %q, and the router held it as %q: the fixture is not the router's", r.repo, r.number, ask.ID, r.key)
|
||||
}
|
||||
teller[ask.ID] = aRouterRecord{ask, w}
|
||||
}
|
||||
withRouter(t, map[string]map[string]aRouterRecord{sdkdelivery.TellerModule: teller})
|
||||
for _, r := range real {
|
||||
why, found, reason, err := findMergeWarrant(t.Context(), nil, r.owner, r.repo, r.number, r.head)
|
||||
if err != nil || !found || !strings.Contains(why, r.key) || !strings.Contains(why, "bound to the head of this pull request") {
|
||||
t.Errorf("%s#%d: the warrant was not found: %q %v %q %v", r.repo, r.number, why, found, reason, err)
|
||||
}
|
||||
// Another head of the same pull request is another merge: the Approve was not given for it.
|
||||
if _, found, _, _ := findMergeWarrant(t.Context(), nil, r.owner, r.repo, r.number, "ffffffffffffffffffffffffffffffffffffffff"); found {
|
||||
t.Errorf("%s#%d: another head found the warrant of this one", r.repo, r.number)
|
||||
}
|
||||
}
|
||||
// One pull request's Approve is none for another repository's merge at the same head digits.
|
||||
if _, found, _, _ := findMergeWarrant(t.Context(), nil, "novox", "mesh-tools", 405, real[0].head); found {
|
||||
t.Error("a merge of another repository found mesh-controller#405's warrant")
|
||||
}
|
||||
}
|
||||
|
||||
// A group's ask binds several heads, and the router's record carries only twelve digits of the other members': the pull request is
|
||||
// found in the ask as it was published, which the warrant's digest covers.
|
||||
func TestAGroupsWarrantIsFoundByTheAsksWordsForThePullRequestAndItsHead(t *testing.T) {
|
||||
a := sdkdelivery.Head{Repository: "novox/mesh-host", Number: 124, Head: "52600ec7bd61ee271a5cc7f4136bcf0a2a8f00fc"}
|
||||
b := sdkdelivery.Head{Repository: "novox/mesh-controller", Number: 405, Head: "8a330afd44ba852873339cf1210e90989dca46e1"}
|
||||
ask, w := aTellersAsk(t, "feat/r4b-sdk-users", a, b)
|
||||
withRouter(t, map[string]map[string]aRouterRecord{sdkdelivery.TellerModule: {ask.ID: {ask, w}}})
|
||||
for _, h := range []sdkdelivery.Head{a, b} {
|
||||
owner, repo, _ := strings.Cut(h.Repository, "/")
|
||||
why, found, reason, err := findMergeWarrant(t.Context(), nil, owner, repo, h.Number, h.Head)
|
||||
if err != nil || !found || !strings.Contains(why, "a group's ask naming this pull request at this head") {
|
||||
t.Errorf("%s#%d: not found in the group's ask: %q %v %q %v", repo, h.Number, why, found, reason, err)
|
||||
}
|
||||
}
|
||||
if _, found, _, _ := findMergeWarrant(t.Context(), nil, "novox", "mesh-catalog", 381, a.Head); found {
|
||||
t.Error("a pull request the group's ask does not name found its warrant")
|
||||
}
|
||||
// A Decline is no warrant.
|
||||
declined := w
|
||||
declined.Option, declined.Label = "decline", "Decline"
|
||||
withRouter(t, map[string]map[string]aRouterRecord{sdkdelivery.TellerModule: {ask.ID: {ask, declined}}})
|
||||
if _, found, _, _ := findMergeWarrant(t.Context(), nil, "novox", "mesh-host", 124, a.Head); found {
|
||||
t.Error("a Decline found a warrant")
|
||||
}
|
||||
}
|
||||
|
||||
// Keycloak's merge was asked as mesh-delivery.merge-378-b9104454f9a5-1, before release 3e: the id names the pull request and
|
||||
// the first twelve digits of its head, and the warrant is found by them under mesh-delivery.
|
||||
func TestAWarrantAskedUnderMeshDeliveryIsFoundByThePullRequestAndItsHead(t *testing.T) {
|
||||
ok := approvedMerge(t)
|
||||
head := "b9104454f9a5e0c4d3b2a1f0e9d8c7b6a5948372"
|
||||
askID := "merge-378-b9104454f9a5-1"
|
||||
ask, w := ok.ask, ok.warrant
|
||||
ask.ID, w.Ask, w.Asker = askID, askID, theDelivery
|
||||
w.AskDigest = ask.Digest()
|
||||
withRouter(t, map[string]map[string]aRouterRecord{theDelivery: {
|
||||
"merge-377-aaaaaaaaaaaa-1": {},
|
||||
askID: {ask, w},
|
||||
}})
|
||||
why, found, reason, err := findMergeWarrant(t.Context(), nil, "novox", "mesh-catalog", 378, head)
|
||||
if err != nil || !found || !strings.Contains(why, askID) {
|
||||
t.Fatalf("the warrant was not found: %q %v %q %v", why, found, reason, err)
|
||||
}
|
||||
// Another head names another ask, and none is found.
|
||||
if _, found, _, _ := findMergeWarrant(t.Context(), nil, "novox", "mesh-catalog", 378, "ffffffffffff0000"); found {
|
||||
t.Fatal("another head found a warrant")
|
||||
}
|
||||
// A router that cannot be read is an error, never "no warrant".
|
||||
routerAskIDs = func(context.Context, catalogueReader, string) ([]string, error) { return nil, errors.New("no answer") }
|
||||
if _, found, _, err := findMergeWarrant(t.Context(), nil, "novox", "mesh-catalog", 378, head); found || err == nil {
|
||||
t.Fatalf("an unreadable router read as no warrant: %v %v", found, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A trunk shorter than a page, or with none, is read without a fault: the pages end when the forge's do.
|
||||
func TestAShortTrunkIsReadWholeAndAnEmptyOneIsNoMerges(t *testing.T) {
|
||||
was := trunkCommitsPage
|
||||
t.Cleanup(func() { trunkCommitsPage = was })
|
||||
trunkCommitsPage = func(_ context.Context, _, _, _ string, page int) ([]trunkCommit, error) {
|
||||
if page > 1 {
|
||||
return nil, nil
|
||||
}
|
||||
return []trunkCommit{{Commit: "m2", SecondParent: "h2"}, {Commit: "p"}, {Commit: "m1", SecondParent: "h1"}}, nil
|
||||
}
|
||||
got, err := readTrunkMerges(t.Context(), "novox", "x", "main")
|
||||
if err != nil || len(got) != 2 || got[0].Commit != "m1" || got[1].Commit != "m2" {
|
||||
t.Fatalf("a short trunk read as %+v %v", got, err)
|
||||
}
|
||||
trunkCommitsPage = func(context.Context, string, string, string, int) ([]trunkCommit, error) { return nil, nil }
|
||||
if got, err := readTrunkMerges(t.Context(), "novox", "x", "main"); err != nil || len(got) != 0 {
|
||||
t.Fatalf("an empty trunk read as %+v %v", got, err)
|
||||
}
|
||||
}
|
||||
|
||||
// ADR 0253: what the operator reads of a merge not heard is plain: no hash, no identifier, nothing to do "by hand".
|
||||
func TestTheConditionOfAMergeNotHeardIsInPlainWords(t *testing.T) {
|
||||
keeper, _ := withConditionsInMemory(t)
|
||||
tm := trunkMerge{Commit: "c0880c4d1f2e3a4b5c6d7e8f9a0b1c2d3e4f5a6b", Head: "b9104454f9a5e0c4d3b2a1f0e9d8c7b6a5948372"}
|
||||
(following{}).raiseMergeNotHeard(t.Context(), "novox/mesh-catalog", "main", tm, link.SourceMoved{Number: 378},
|
||||
"the router holds no chosen merge ask of mesh-delivery-teller or mesh-delivery that covers the head b9104454f9a5")
|
||||
open, err := keeper.Open(t.Context())
|
||||
if err != nil || len(open) != 1 {
|
||||
t.Fatalf("opened %d conditions, %v", len(open), err)
|
||||
}
|
||||
c := open[0]
|
||||
if why, ok := conditions.PlainWords(conditions.Words{Headline: c.Headline, Explanation: c.Explanation, Needs: c.Needs,
|
||||
Resolved: c.Resolved}); !ok {
|
||||
t.Fatalf("the condition's words are not plain: %s\n%q\n%q\n%q", why, c.Headline, c.Explanation, c.Needs)
|
||||
}
|
||||
}
|
||||
@@ -208,6 +208,8 @@ func serve(ctx context.Context) (err error) {
|
||||
if err := server.Follows(following{open: open}); err != nil {
|
||||
return err
|
||||
}
|
||||
// And the merges it did not hear while it was away, caught up once it serves (novox/hq ADR 0457, issue 1336).
|
||||
go catchUpAtStart(ctx, open)
|
||||
// And the merges mesh-delivery made, kept a week with the ask each was made on: what registers a module that
|
||||
// moved between repositories on the operator's Approve (novox/hq ADR 0420 §7, issue 1025).
|
||||
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, catalog: open.inventory.Catalogued,
|
||||
|
||||
@@ -316,7 +316,16 @@ func unsaved(err error) error {
|
||||
// SourceMoved is the forge announcing a merge. It is put into the open batch (novox/hq ADR 0276), which
|
||||
// becomes one walk when its merge window closes: hearMerge, and cutBatchesHeld in batches.go.
|
||||
func (f following) SourceMoved(ctx context.Context, m link.SourceMoved) error {
|
||||
return f.hearMerge(ctx, m, time.Now().UTC())
|
||||
if err := f.hearMerge(ctx, m, time.Now().UTC()); err != nil {
|
||||
return err
|
||||
}
|
||||
// And the merges of that trunk it did not hear, in the trunk's order (novox/hq ADR 0457): a catch-up that leaves
|
||||
// something for later is no failure of this merge's hearing, and is never a reason to hear it again.
|
||||
if !f.oneShot {
|
||||
trunk := &caughtTrunk{Owner: m.Owner, Repo: m.Repo, Branch: m.Base}
|
||||
catchUpBeside(func() { f.catchUp(context.WithoutCancel(ctx), trunk) })
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// movesOfMerge is what one merge moves, acted on: every module recorded as built from that repository and
|
||||
|
||||
Reference in New Issue
Block a user