Remove the five-minute merge catch-up, its signal S5 and its watchdog: the event and its durable consumer keep every merge (issue 973)
This commit is contained in:
@@ -67,6 +67,10 @@ const mergeWindowFileVar = "MESH_MERGE_WINDOW_FILE"
|
|||||||
// the controller is said to have failed to cut it (ADR 0276 decision 9).
|
// the controller is said to have failed to cut it (ADR 0276 decision 9).
|
||||||
const batchLateAfter = time.Minute
|
const batchLateAfter = time.Minute
|
||||||
|
|
||||||
|
// mergeGrace is how long before a batch's first merge a merge may have been made and still be told apart from one
|
||||||
|
// made before the controller began keeping merges (owedLate).
|
||||||
|
const mergeGrace = 10 * time.Minute
|
||||||
|
|
||||||
// edgeGroupOrder is a delivery group's order made an edge of a walk's tiers (ADR 0276 decision 5): the
|
// edgeGroupOrder is a delivery group's order made an edge of a walk's tiers (ADR 0276 decision 5): the
|
||||||
// modules of the member after depend on the modules of the member before, and are gated on them.
|
// modules of the member after depend on the modules of the member before, and are gated on them.
|
||||||
const edgeGroupOrder = "group-order"
|
const edgeGroupOrder = "group-order"
|
||||||
@@ -124,8 +128,7 @@ func durationSetting(v any) (time.Duration, bool) {
|
|||||||
// hearMerge puts a merge the forge announced into the batch that answers it, at the moment it was heard,
|
// hearMerge puts a merge the forge announced into the batch that answers it, at the moment it was heard,
|
||||||
// and cuts what is due.
|
// and cuts what is due.
|
||||||
func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.Time) error {
|
func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.Time) error {
|
||||||
// One merge acted on at a time, whoever hands it over: the bus, or the catch-up that reads back
|
// One merge acted on at a time, whoever hands it over (novox/hq issue 266).
|
||||||
// what the bus did not hand over (novox/hq issue 266).
|
|
||||||
actingOnMerges.Lock()
|
actingOnMerges.Lock()
|
||||||
defer actingOnMerges.Unlock()
|
defer actingOnMerges.Unlock()
|
||||||
|
|
||||||
@@ -244,7 +247,7 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
|
|||||||
}
|
}
|
||||||
|
|
||||||
// owedLate says a merge that reads as history is still the record's to answer (review of this change): a cut
|
// owedLate says a merge that reads as history is still the record's to answer (review of this change): a cut
|
||||||
// marks every module it moved seen at that moment, so a merge of that branch the bus lost and the catch-up
|
// marks every module it moved seen at that moment, so a merge of that branch handed over late
|
||||||
// hands over later — made before the cut, heard after it — reads as history and was dropped. One that touches
|
// hands over later — made before the cut, heard after it — reads as history and was dropped. One that touches
|
||||||
// a module the mesh holds, of a branch whose merges the controller keeps, made since it began keeping them,
|
// a module the mesh holds, of a branch whose merges the controller keeps, made since it began keeping them,
|
||||||
// is kept; one made before it began is old news, as before.
|
// is kept; one made before it began is old news, as before.
|
||||||
|
|||||||
@@ -729,49 +729,6 @@ func TestARemovedFileIsTheLastMergesWord(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// The catch-up hands over what the bus lost after its branch's later merges were cut, and the record keeps it:
|
|
||||||
// an earlier merge is named by the walk that carried it; a merge made before a cut and heard after it, which
|
|
||||||
// no later merge carries, joins the next batch — neither reads as history and is dropped (review of this change).
|
|
||||||
func TestTheCatchUpKeepsWhatACutMadeHistory(t *testing.T) {
|
|
||||||
open := windowed(t)
|
|
||||||
asksWithPaths(t)
|
|
||||||
ctx := t.Context()
|
|
||||||
base := time.Now().UTC().Add(-time.Hour).Truncate(time.Second)
|
|
||||||
hear(t, open, repoMerge("one", "c2", base.Add(10*time.Second)), base.Add(11*time.Second))
|
|
||||||
cutAt(t, open, base.Add(2*time.Minute))
|
|
||||||
ws, _ := walks(t, open)
|
|
||||||
w := ws[0]
|
|
||||||
w.State = inventory.PlanDone
|
|
||||||
if err := open.inventory.SavePlan(ctx, &w); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
lost := []link.AnnouncedMerge{
|
|
||||||
{SourceMoved: repoMerge("one", "c1", base), At: base.Add(time.Second)},
|
|
||||||
{SourceMoved: repoMerge("one", "c3lost", base.Add(20*time.Second)), At: base.Add(21 * time.Second)},
|
|
||||||
}
|
|
||||||
catalogued := func(ctx context.Context) ([]inventory.Entry, map[string][]inventory.ReadRepository, error) {
|
|
||||||
entries, err := open.inventory.Catalogued(ctx)
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, err
|
|
||||||
}
|
|
||||||
read, err := readForPlanning(ctx, open.inventory)
|
|
||||||
return entries, read, err
|
|
||||||
}
|
|
||||||
if err := catchUpOnMerges(ctx, time.Now(), unheardMerges{announcedList(lost), open.inventory}, catalogued,
|
|
||||||
(following{open: open}).SourceMoved, t.Logf); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
got, _ := open.inventory.PlanByID(ctx, w.ID)
|
|
||||||
if len(got.Delivery.Merges) != 2 || !sameMerge(got.Delivery.Merges[0], inventory.PlanMerge{Repository: "novox/one",
|
|
||||||
Commit: "c1", Carried: "c2"}) {
|
|
||||||
t.Fatalf("the earlier merge the catch-up handed over is not named by the walk that carried it: %+v",
|
|
||||||
got.Delivery.Merges)
|
|
||||||
}
|
|
||||||
if _, kept, err := open.inventory.MergeOf(ctx, "novox/one", "c3lost"); err != nil || !kept {
|
|
||||||
t.Fatalf("the merge made before the cut and heard after it was dropped: %v %v", kept, err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// A merge on the controller's own path never shares a batch with one that waits for mesh-delivery's word
|
// A merge on the controller's own path never shares a batch with one that waits for mesh-delivery's word
|
||||||
// (decided during the build, 2026-10-10): the two are batches of their own, cut one after the other; a walk on
|
// (decided during the build, 2026-10-10): the two are batches of their own, cut one after the other; a walk on
|
||||||
// the own path folds no waiting catalogue walk but batches its merges again behind it; the catalogue's walk
|
// the own path folds no waiting catalogue walk but batches its merges again behind it; the catalogue's walk
|
||||||
|
|||||||
@@ -1,243 +0,0 @@
|
|||||||
package main
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"fmt"
|
|
||||||
"sync"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/novox/mesh-controller/internal/inventory"
|
|
||||||
"github.com/novox/mesh-controller/internal/link"
|
|
||||||
)
|
|
||||||
|
|
||||||
// **A merge the bus announced and the controller never acted on is caught up** (novox/hq issue 266).
|
|
||||||
//
|
|
||||||
// The forge's poll announces every merge on the events stream, and the controller acts on what its
|
|
||||||
// consumer there hands it. On 2026-10-06 one merge was on the stream and never handed over: the bus
|
|
||||||
// server moved the consumer past it — a fault of consumers with several filters in the server the
|
|
||||||
// mesh ran — and the controller, which only acts on what it is handed, said nothing. The modules
|
|
||||||
// built from that repository stayed behind and were built by hand.
|
|
||||||
//
|
|
||||||
// So the stream is read back on a timer, on a consumer of its own filtered on merges alone, and every
|
|
||||||
// announcement older than mergeGrace is judged as SourceMoved would judge it. **No record of what
|
|
||||||
// was handled is kept, because none is needed**: acting on a merge marks every module it moved as
|
|
||||||
// looked at since, so an announcement already acted on reads as history and moves nothing. One that
|
|
||||||
// would still move something was never acted on — it is said, and acted on now.
|
|
||||||
const (
|
|
||||||
// mergeGrace is how long an announcement is left to the controller's own consumer before it is
|
|
||||||
// judged missed. That consumer hands over one event at a time, and a merge waits behind a build
|
|
||||||
// outcome that is being acted on; acting on a merge itself asks builds and does not wait for them.
|
|
||||||
mergeGrace = 10 * time.Minute
|
|
||||||
// mergeLookBack is how far back a pass reads. A merge missed longer ago than this was missed by a
|
|
||||||
// controller that was not running this, and is the operator's to look at, not a surprise rebuild.
|
|
||||||
mergeLookBack = 24 * time.Hour
|
|
||||||
// mergeCatchUpEvery is how often the stream is read back.
|
|
||||||
mergeCatchUpEvery = 5 * time.Minute
|
|
||||||
)
|
|
||||||
|
|
||||||
// merges is what reads back the forge's announcements; the link server, or a test's list.
|
|
||||||
type merges interface {
|
|
||||||
AnnouncedMerges(ctx context.Context, since time.Time) ([]link.AnnouncedMerge, error)
|
|
||||||
}
|
|
||||||
|
|
||||||
// unheardMerges is the announcements of merges the controller has not heard (novox/hq ADR 0276): a merge kept
|
|
||||||
// in a batch is not acted on until its batch is cut, which can be past mergeGrace behind a long walk, and is
|
|
||||||
// not missed for that.
|
|
||||||
type unheardMerges struct {
|
|
||||||
merges
|
|
||||||
inv *inventory.Inventory
|
|
||||||
}
|
|
||||||
|
|
||||||
func (u unheardMerges) AnnouncedMerges(ctx context.Context, since time.Time) ([]link.AnnouncedMerge, error) {
|
|
||||||
all, err := u.merges.AnnouncedMerges(ctx, since)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
var out []link.AnnouncedMerge
|
|
||||||
for _, a := range all {
|
|
||||||
_, kept, err := u.inv.MergeOf(ctx, a.Owner+"/"+a.Repo, a.Commit)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
if !kept {
|
|
||||||
out = append(out, a)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// owedToTheRecord says a merge read as history is still owed to the record (novox/hq ADR 0276): the merges
|
|
||||||
// reader knows, when it keeps them (unheardMerges).
|
|
||||||
func owedToTheRecord(ctx context.Context, announced merges, m link.SourceMoved, entries []inventory.Entry,
|
|
||||||
read map[string][]inventory.ReadRepository) bool {
|
|
||||||
u, ok := announced.(unheardMerges)
|
|
||||||
if !ok {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
merged, err := time.Parse(time.RFC3339Nano, m.MergedAt)
|
|
||||||
if err != nil {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
raw := m
|
|
||||||
raw.MergedAt = ""
|
|
||||||
owed, err := owedLate(ctx, u.inv, m, merged, wouldMove(raw, entries, read))
|
|
||||||
return err == nil && owed
|
|
||||||
}
|
|
||||||
|
|
||||||
// catchingUpOnMerges reads back the forge's announcements on a timer, until the context ends.
|
|
||||||
func catchingUpOnMerges(ctx context.Context, open *stores, announced merges) {
|
|
||||||
f := following{open: open}
|
|
||||||
catalogued := func(ctx context.Context) ([]inventory.Entry, map[string][]inventory.ReadRepository, error) {
|
|
||||||
entries, err := open.inventory.Catalogued(ctx)
|
|
||||||
if err != nil {
|
|
||||||
return nil, nil, err
|
|
||||||
}
|
|
||||||
read, err := readForPlanning(ctx, open.inventory)
|
|
||||||
return entries, read, err
|
|
||||||
}
|
|
||||||
failing := ""
|
|
||||||
tick := time.NewTicker(mergeCatchUpEvery)
|
|
||||||
defer tick.Stop()
|
|
||||||
for {
|
|
||||||
select {
|
|
||||||
case <-ctx.Done():
|
|
||||||
return
|
|
||||||
case <-tick.C:
|
|
||||||
}
|
|
||||||
watchedMerges.begin()
|
|
||||||
err := catchUpOnMerges(ctx, time.Now(), unheardMerges{announced, open.inventory}, catalogued, f.SourceMoved, func(format string, args ...any) {
|
|
||||||
fmt.Printf(format+"\n", args...)
|
|
||||||
})
|
|
||||||
// What the pass found is what S5 says (novox/hq to-be 45 §3); a pass that could not read
|
|
||||||
// says that instead, and leaves what the last one found standing.
|
|
||||||
watchedMerges.end(time.Now(), err)
|
|
||||||
// A pass that cannot read says so once, not every five minutes, and says when it reads again.
|
|
||||||
why := ""
|
|
||||||
if err != nil {
|
|
||||||
why = err.Error()
|
|
||||||
}
|
|
||||||
if why != failing {
|
|
||||||
if why != "" {
|
|
||||||
fmt.Printf("merges the bus may not have handed over cannot be looked for: %s\n", why)
|
|
||||||
} else {
|
|
||||||
fmt.Println("merges the bus may not have handed over are looked for again")
|
|
||||||
}
|
|
||||||
failing = why
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// catchUpOnMerges is one pass: every announcement older than mergeGrace that acting on would still
|
|
||||||
// move something is said and acted on, oldest first.
|
|
||||||
//
|
|
||||||
// Judged twice: once against the catalogue as the pass found it, and again just before acting,
|
|
||||||
// because acting on an earlier missed merge of the same repository may have moved what a later one
|
|
||||||
// would have.
|
|
||||||
func catchUpOnMerges(ctx context.Context, now time.Time, announced merges,
|
|
||||||
catalogued func(context.Context) ([]inventory.Entry, map[string][]inventory.ReadRepository, error),
|
|
||||||
act func(context.Context, link.SourceMoved) error, say func(string, ...any)) error {
|
|
||||||
|
|
||||||
all, err := announced.AnnouncedMerges(ctx, now.Add(-mergeLookBack))
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
entries, read, err := catalogued(ctx)
|
|
||||||
if err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
for _, a := range all {
|
|
||||||
// A merge the forge said no time of is dated by its announcement, so a packaging module's look can
|
|
||||||
// make it history once acted on, and the catch-up does not act on it again every pass (ADR 0267).
|
|
||||||
if a.SourceMoved.MergedAt == "" && !a.At.IsZero() {
|
|
||||||
a.SourceMoved.MergedAt = a.At.UTC().Format(time.RFC3339)
|
|
||||||
}
|
|
||||||
if now.Sub(a.At) < mergeGrace {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if len(wouldMove(a.SourceMoved, entries, read)) == 0 && !owedToTheRecord(ctx, announced, a.SourceMoved, entries, read) {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if entries, read, err = catalogued(ctx); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
moves := wouldMove(a.SourceMoved, entries, read)
|
|
||||||
if len(moves) == 0 {
|
|
||||||
if !owedToTheRecord(ctx, announced, a.SourceMoved, entries, read) {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
raw := a.SourceMoved
|
|
||||||
raw.MergedAt = ""
|
|
||||||
moves = wouldMove(raw, entries, read)
|
|
||||||
}
|
|
||||||
var names []string
|
|
||||||
for _, e := range moves {
|
|
||||||
names = append(names, e.Manifest.Module)
|
|
||||||
}
|
|
||||||
watchedMerges.found(missedMerge{Owner: a.Owner, Repo: a.Repo, Base: a.Base, Commit: a.Commit,
|
|
||||||
At: a.At, Modules: names})
|
|
||||||
say("%s/%s merged into %s (%.8s), announced %s ago, and the controller never acted on it: the bus "+
|
|
||||||
"did not hand the announcement over (novox/hq issue 266). %s %s behind it; acting on it now",
|
|
||||||
a.Owner, a.Repo, a.Base, a.Commit, now.Sub(a.At).Round(time.Minute), readableList(names),
|
|
||||||
isAre(len(names)))
|
|
||||||
if err := act(ctx, a.SourceMoved); err != nil {
|
|
||||||
say("%s/%s moved to %.8s and the mesh could not act on it: %v; the next pass tries again",
|
|
||||||
a.Owner, a.Repo, a.Commit, err)
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if entries, read, err = catalogued(ctx); err != nil {
|
|
||||||
return err
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// missedMerge is one merge the bus announced and never handed over, as a pass found it.
|
|
||||||
type missedMerge struct {
|
|
||||||
Owner, Repo, Base, Commit string
|
|
||||||
// At is when the bus took the announcement.
|
|
||||||
At time.Time
|
|
||||||
Modules []string
|
|
||||||
}
|
|
||||||
|
|
||||||
// mergeWatch is what the passes found, for S5 (novox/hq to-be 45 §3): **a merge nothing read is
|
|
||||||
// said**, urgent, even though the pass acts on it at once — the bus skipping a message is a fault of
|
|
||||||
// the transport the mesh's every change rides on, and acting late is the repair, not the absence of
|
|
||||||
// the fault. The next pass, finding it acted on, clears it.
|
|
||||||
type mergeWatch struct {
|
|
||||||
mu sync.Mutex
|
|
||||||
passed time.Time
|
|
||||||
err error
|
|
||||||
finding []missedMerge
|
|
||||||
missed []missedMerge
|
|
||||||
}
|
|
||||||
|
|
||||||
var watchedMerges = &mergeWatch{}
|
|
||||||
|
|
||||||
func (w *mergeWatch) begin() {
|
|
||||||
w.mu.Lock()
|
|
||||||
defer w.mu.Unlock()
|
|
||||||
w.finding = nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (w *mergeWatch) found(m missedMerge) {
|
|
||||||
w.mu.Lock()
|
|
||||||
defer w.mu.Unlock()
|
|
||||||
w.finding = append(w.finding, m)
|
|
||||||
}
|
|
||||||
|
|
||||||
func (w *mergeWatch) end(at time.Time, err error) {
|
|
||||||
w.mu.Lock()
|
|
||||||
defer w.mu.Unlock()
|
|
||||||
w.passed, w.err = at, err
|
|
||||||
if err == nil {
|
|
||||||
w.missed = w.finding
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// last is when the last pass ended, what the last pass that read found, and what the last pass
|
|
||||||
// could not read.
|
|
||||||
func (w *mergeWatch) last() (time.Time, []missedMerge, error) {
|
|
||||||
w.mu.Lock()
|
|
||||||
defer w.mu.Unlock()
|
|
||||||
return w.passed, append([]missedMerge(nil), w.missed...), w.err
|
|
||||||
}
|
|
||||||
@@ -1,180 +0,0 @@
|
|||||||
package main
|
|
||||||
|
|
||||||
import (
|
|
||||||
"context"
|
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
"strings"
|
|
||||||
"testing"
|
|
||||||
"time"
|
|
||||||
|
|
||||||
"github.com/novox/mesh-controller/internal/inventory"
|
|
||||||
"github.com/novox/mesh-controller/internal/link"
|
|
||||||
)
|
|
||||||
|
|
||||||
// announcedList is the events stream's merges as a test gives them.
|
|
||||||
type announcedList []link.AnnouncedMerge
|
|
||||||
|
|
||||||
func (a announcedList) AnnouncedMerges(_ context.Context, since time.Time) ([]link.AnnouncedMerge, error) {
|
|
||||||
var out []link.AnnouncedMerge
|
|
||||||
for _, m := range a {
|
|
||||||
if !m.At.Before(since) {
|
|
||||||
out = append(out, m)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return out, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// aCatalogue is what the inventory holds, changed the way acting on a merge changes it: every module
|
|
||||||
// the merge moved is marked as looked at (inventory.SourceMoved writes source_seen = now()).
|
|
||||||
type aCatalogue struct {
|
|
||||||
entries []inventory.Entry
|
|
||||||
acted []string
|
|
||||||
fail error
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *aCatalogue) read(context.Context) ([]inventory.Entry, map[string][]inventory.ReadRepository, error) {
|
|
||||||
return append([]inventory.Entry(nil), c.entries...), nil, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (c *aCatalogue) act(now func() time.Time) func(context.Context, link.SourceMoved) error {
|
|
||||||
return func(_ context.Context, m link.SourceMoved) error {
|
|
||||||
if c.fail != nil {
|
|
||||||
return c.fail
|
|
||||||
}
|
|
||||||
c.acted = append(c.acted, m.Repo+"@"+m.Commit[:8])
|
|
||||||
for _, moved := range wouldMove(m, c.entries, nil) {
|
|
||||||
for i := range c.entries {
|
|
||||||
if c.entries[i].Manifest.Module == moved.Manifest.Module {
|
|
||||||
c.entries[i].Source.Head = m.Commit
|
|
||||||
c.entries[i].Source.Seen = now()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return nil
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func at(s string) time.Time {
|
|
||||||
t, err := time.Parse(time.RFC3339, s)
|
|
||||||
if err != nil {
|
|
||||||
panic(err)
|
|
||||||
}
|
|
||||||
return t
|
|
||||||
}
|
|
||||||
|
|
||||||
func announced(repo, commit, mergedAt, onTheBus string, paths ...string) link.AnnouncedMerge {
|
|
||||||
return link.AnnouncedMerge{
|
|
||||||
SourceMoved: link.SourceMoved{Owner: "novox", Repo: repo, Base: "main", Commit: commit,
|
|
||||||
MergedAt: mergedAt, Paths: paths,
|
|
||||||
CloneURL: "http://forge.internal:20000/novox/" + repo + ".git"},
|
|
||||||
At: at(onTheBus),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
func built(module, repo, path, commit, seen string) inventory.Entry {
|
|
||||||
e := fromRepo(module, "http://forge.internal:20000/novox/"+repo+".git", path)
|
|
||||||
e.Source.BuiltFrom, e.Source.Head, e.Source.Seen = commit, commit, at(seen)
|
|
||||||
return e
|
|
||||||
}
|
|
||||||
|
|
||||||
// **novox/hq issue 266, as it happened.** The forge announced a merge of the tools repository on the
|
|
||||||
// events stream; the bus never handed it to the controller, which acted on the merges around it and
|
|
||||||
// not on this one, and said nothing. Read back from the stream, it is the one merge that would still
|
|
||||||
// move something — so it is said and acted on, once, and only after the controller's own consumer
|
|
||||||
// has had its time with it.
|
|
||||||
func TestAMergeTheBusNeverHandedOverIsActedOnLate(t *testing.T) {
|
|
||||||
cat := &aCatalogue{entries: []inventory.Entry{
|
|
||||||
built("mesh-tools", "mesh-tools", "", "8b789578aaaaaaaa", "2026-10-04T15:24:32Z"),
|
|
||||||
built("node-tools", "mesh-tools", "node-tools", "8b789578aaaaaaaa", "2026-10-04T15:24:32Z"),
|
|
||||||
// Acted on when it was announced: looked at after it was merged.
|
|
||||||
built("gitea", "mesh-catalog", "modules/gitea", "5c2157b8bbbbbbbb", "2026-10-05T22:39:21Z"),
|
|
||||||
}}
|
|
||||||
stream := announcedList{
|
|
||||||
// Nothing the mesh holds is built from the records repository.
|
|
||||||
announced("hq", "88f7f79fcccccccc", "2026-10-05T22:43:00Z", "2026-10-05T22:43:04Z", "04-ISSUES/x.md"),
|
|
||||||
// Acted on: its module was looked at since.
|
|
||||||
announced("mesh-catalog", "78328d4adddddddd", "2026-10-05T22:39:00Z", "2026-10-05T22:39:21Z", "modules/gitea/x.ts"),
|
|
||||||
// Never handed over.
|
|
||||||
announced("mesh-tools", "9730bd89c3e48d0e", "2026-10-05T22:46:47Z", "2026-10-05T22:47:06Z",
|
|
||||||
"node-tools/internal/console/console.go"),
|
|
||||||
}
|
|
||||||
var said []string
|
|
||||||
say := func(format string, args ...any) { said = append(said, fmt.Sprintf(format, args...)) }
|
|
||||||
clock := at("2026-10-05T22:50:00Z")
|
|
||||||
now := func() time.Time { return clock }
|
|
||||||
pass := func() {
|
|
||||||
t.Helper()
|
|
||||||
if err := catchUpOnMerges(context.Background(), clock, stream, cat.read, cat.act(now), say); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
pass()
|
|
||||||
if len(cat.acted) != 0 {
|
|
||||||
t.Fatalf("a merge three minutes old was taken from the controller's own consumer: %v", cat.acted)
|
|
||||||
}
|
|
||||||
|
|
||||||
clock = at("2026-10-05T22:58:00Z")
|
|
||||||
pass()
|
|
||||||
if strings.Join(cat.acted, ",") != "mesh-tools@9730bd89" {
|
|
||||||
t.Fatalf("acted on %v, wanted the one merge never handed over", cat.acted)
|
|
||||||
}
|
|
||||||
if len(said) != 1 || !strings.Contains(said[0], "novox/mesh-tools merged into main (9730bd89)") ||
|
|
||||||
!strings.Contains(said[0], "mesh-tools and node-tools are behind it") {
|
|
||||||
t.Fatalf("the missed merge was not said as one: %q", said)
|
|
||||||
}
|
|
||||||
|
|
||||||
clock = at("2026-10-05T23:03:00Z")
|
|
||||||
pass()
|
|
||||||
if len(cat.acted) != 1 || len(said) != 1 {
|
|
||||||
t.Fatalf("a merge acted on was acted on again: %v %q", cat.acted, said)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// A merge that changed none of the held modules' files moves nothing, so it is never "missed"; one
|
|
||||||
// that could not be acted on is said and tried again on the next pass.
|
|
||||||
func TestAMissedMergeThatCouldNotBeActedOnIsTriedAgain(t *testing.T) {
|
|
||||||
cat := &aCatalogue{
|
|
||||||
entries: []inventory.Entry{built("gitea", "mesh-catalog", "modules/gitea", "5c2157b8bbbbbbbb", "2026-10-05T20:00:00Z")},
|
|
||||||
fail: errors.New("the store is restarting"),
|
|
||||||
}
|
|
||||||
stream := announcedList{
|
|
||||||
announced("mesh-catalog", "aaaaaaaa11111111", "2026-10-05T21:00:00Z", "2026-10-05T21:00:10Z",
|
|
||||||
"modules/plex/module.json", "modules/plex/x.ts"),
|
|
||||||
announced("mesh-catalog", "bbbbbbbb22222222", "2026-10-05T21:10:00Z", "2026-10-05T21:10:10Z", "modules/gitea/x.ts"),
|
|
||||||
}
|
|
||||||
var said []string
|
|
||||||
say := func(format string, args ...any) { said = append(said, fmt.Sprintf(format, args...)) }
|
|
||||||
clock := at("2026-10-05T22:00:00Z")
|
|
||||||
now := func() time.Time { return clock }
|
|
||||||
if err := catchUpOnMerges(context.Background(), clock, stream, cat.read, cat.act(now), say); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if len(said) != 2 || !strings.Contains(said[1], "could not act on it") {
|
|
||||||
t.Fatalf("a failed catch-up was not said: %q", said)
|
|
||||||
}
|
|
||||||
cat.fail = nil
|
|
||||||
clock = at("2026-10-05T22:05:00Z")
|
|
||||||
if err := catchUpOnMerges(context.Background(), clock, stream, cat.read, cat.act(now), say); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if strings.Join(cat.acted, ",") != "mesh-catalog@bbbbbbbb" {
|
|
||||||
t.Fatalf("acted on %v, wanted only the merge that changed a held module", cat.acted)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// A merge older than the look-back is left to the operator: a controller that did not run this
|
|
||||||
// missed it, and acting on it days later would be a surprise rebuild.
|
|
||||||
func TestAMergeOlderThanTheLookBackIsLeftAlone(t *testing.T) {
|
|
||||||
cat := &aCatalogue{entries: []inventory.Entry{built("gitea", "mesh-catalog", "modules/gitea", "5c2157b8bbbbbbbb", "2026-10-01T00:00:00Z")}}
|
|
||||||
stream := announcedList{announced("mesh-catalog", "cccccccc33333333", "2026-10-03T00:00:00Z", "2026-10-03T00:00:05Z", "modules/gitea/x.ts")}
|
|
||||||
clock := at("2026-10-05T22:00:00Z")
|
|
||||||
if err := catchUpOnMerges(context.Background(), clock, stream, cat.read, cat.act(func() time.Time { return clock }),
|
|
||||||
func(string, ...any) {}); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
if len(cat.acted) != 0 {
|
|
||||||
t.Fatalf("a merge of three days ago was acted on: %v", cat.acted)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -429,12 +429,6 @@ var plainWordings = map[string]func(conditions.Observation) words{
|
|||||||
"out) has not let it start yet.",
|
"out) has not let it start yet.",
|
||||||
Resolved: "Resolved: the delivery is no longer waiting"}
|
Resolved: "Resolved: the delivery is no longer waiting"}
|
||||||
}),
|
}),
|
||||||
"merge-not-acted": worded(func(o conditions.Observation) words {
|
|
||||||
return words{Headline: "A merge was never picked up",
|
|
||||||
Explanation: "A pull request was merged, and the controller never heard of it, so nothing was built or " +
|
|
||||||
"sent. The mesh catches up on missed merges by itself.",
|
|
||||||
Resolved: "Resolved: the merge was picked up"}
|
|
||||||
}),
|
|
||||||
"ask-lost": worded(func(o conditions.Observation) words {
|
"ask-lost": worded(func(o conditions.Observation) words {
|
||||||
return words{Headline: "A build has no result",
|
return words{Headline: "A build has no result",
|
||||||
Explanation: "A build was asked for and never answered. It is asked again with the next delivery.",
|
Explanation: "A build was asked for and never answered. It is asked again with the next delivery.",
|
||||||
|
|||||||
@@ -201,9 +201,8 @@ func serve(ctx context.Context) (err error) {
|
|||||||
if err := server.Follows(following{open: open}); err != nil {
|
if err := server.Follows(following{open: open}); err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
// And the merges the bus announced and never handed over, read back on a timer and acted on late
|
// No timer reads the merges back (novox/hq ADR 0407 decision 3, issue 973): the forge's event and the durable
|
||||||
// rather than never (novox/hq issue 266).
|
// consumer on the bus keep every merge, and one the controller cannot record yet is held and handed over again.
|
||||||
go catchingUpOnMerges(ctx, open, server)
|
|
||||||
// And a catalogue that has just started, asking for what it missed. The same type answers
|
// And a catalogue that has just started, asking for what it missed. The same type answers
|
||||||
// both: what a build meant and what the builds were are two questions about one record.
|
// both: what a build meant and what the builds were are two questions about one record.
|
||||||
if err := server.Answers(following{open: open}); err != nil {
|
if err := server.Answers(following{open: open}); err != nil {
|
||||||
|
|||||||
@@ -126,11 +126,6 @@ var signalsTable = []signalRow{
|
|||||||
Kind: "controller-deaf", Severity: conditions.Urgent, Phase: 1,
|
Kind: "controller-deaf", Severity: conditions.Urgent, Phase: 1,
|
||||||
needs: func(f *signalFacts) error { return f.loopErr }, watch: watchLoop,
|
needs: func(f *signalFacts) error { return f.loopErr }, watch: watchLoop,
|
||||||
newest: func(f *signalFacts) time.Time { return f.loop.took }},
|
newest: func(f *signalFacts) time.Time { return f.loop.took }},
|
||||||
{Row: "S5", Signal: "a merge announced becomes a plan, or nothing reads it", Emitter: "announcer → controller",
|
|
||||||
Trigger: "each merge", Bound: "10 min (the catch-up pass of issue 266)",
|
|
||||||
Kind: "merge-not-acted", Severity: conditions.Urgent, Phase: 1,
|
|
||||||
needs: func(f *signalFacts) error { return f.mergesErr }, watch: watchMerges,
|
|
||||||
newest: func(f *signalFacts) time.Time { return f.mergesPassed }},
|
|
||||||
{Row: "S6", Signal: "a build asked → its outcome", Emitter: "build seat", Trigger: "each ask",
|
{Row: "S6", Signal: "a build asked → its outcome", Emitter: "build seat", Trigger: "each ask",
|
||||||
Bound: "max(20 min, 3 × the p90 of measured builds), 1 h while nothing is measured, provisional — the " +
|
Bound: "max(20 min, 3 × the p90 of measured builds), 1 h while nothing is measured, provisional — the " +
|
||||||
"build seat declares no timeout; and an ask the queue gave up on (dead) at once",
|
"build seat declares no timeout; and an ask the queue gave up on (dead) at once",
|
||||||
@@ -459,18 +454,6 @@ func watchLoop(f *signalFacts) []conditions.Observation {
|
|||||||
Said: fmt.Sprintf("%d pending (%s), last taken %s", f.loop.pending, f.loop.where, since.UTC().Format(time.RFC3339))}}
|
Said: fmt.Sprintf("%d pending (%s), last taken %s", f.loop.pending, f.loop.where, since.UTC().Format(time.RFC3339))}}
|
||||||
}
|
}
|
||||||
|
|
||||||
func watchMerges(f *signalFacts) []conditions.Observation {
|
|
||||||
var out []conditions.Observation
|
|
||||||
for _, m := range f.merges {
|
|
||||||
out = append(out, conditions.Observation{Scope: conditions.ScopeMerge, ID: m.Repo + "." + short(m.Commit),
|
|
||||||
Kind: "merge-not-acted", Token: "not-acted", Severity: conditions.Urgent,
|
|
||||||
Summary: fmt.Sprintf("%s/%s merged into %s (%s) was never handed to the controller by the bus; "+
|
|
||||||
"acted on late by the catch-up", m.Owner, m.Repo, m.Base, short(m.Commit)),
|
|
||||||
Said: fmt.Sprintf("announced %s, %s behind it", m.At.UTC().Format(time.RFC3339), readableList(m.Modules))})
|
|
||||||
}
|
|
||||||
return out
|
|
||||||
}
|
|
||||||
|
|
||||||
func watchAsks(f *signalFacts) []conditions.Observation {
|
func watchAsks(f *signalFacts) []conditions.Observation {
|
||||||
var out []conditions.Observation
|
var out []conditions.Observation
|
||||||
for _, a := range f.asks {
|
for _, a := range f.asks {
|
||||||
|
|||||||
@@ -33,7 +33,6 @@ func calm(now time.Time) *signalFacts {
|
|||||||
plans: []planFacts{{id: "plan-1", repository: "novox/app", commit: "c0ffee00", tier: 0, tiers: 2,
|
plans: []planFacts{{id: "plan-1", repository: "novox/app", commit: "c0ffee00", tier: 0, tiers: 2,
|
||||||
entered: now.Add(-time.Minute), bound: 30 * time.Minute, waiting: "building"}},
|
entered: now.Add(-time.Minute), bound: 30 * time.Minute, waiting: "building"}},
|
||||||
loop: loopFacts{took: now.Add(-time.Second), pending: 1},
|
loop: loopFacts{took: now.Add(-time.Second), pending: 1},
|
||||||
mergesPassed: now.Add(-time.Minute),
|
|
||||||
selfCheck: selfCheckFacts{last: now.Add(-time.Minute), every: 5 * time.Minute},
|
selfCheck: selfCheckFacts{last: now.Add(-time.Minute), every: 5 * time.Minute},
|
||||||
lostConsumers: map[string]bool{}, epochs: map[int64]inventory.Epoch{},
|
lostConsumers: map[string]bool{}, epochs: map[int64]inventory.Epoch{},
|
||||||
lease: leaseFacts{held: true, epoch: 57, renewed: now.Add(-2 * time.Second)},
|
lease: leaseFacts{held: true, epoch: 57, renewed: now.Add(-2 * time.Second)},
|
||||||
@@ -72,12 +71,6 @@ var suppressions = map[string]suppression{
|
|||||||
inside: func(f *signalFacts) { f.loop.took = f.now.Add(-119 * time.Second) },
|
inside: func(f *signalFacts) { f.loop.took = f.now.Add(-119 * time.Second) },
|
||||||
past: func(f *signalFacts) { f.loop.took = f.now.Add(-121 * time.Second) },
|
past: func(f *signalFacts) { f.loop.took = f.now.Add(-121 * time.Second) },
|
||||||
},
|
},
|
||||||
"S5": {
|
|
||||||
inside: func(f *signalFacts) {},
|
|
||||||
past: func(f *signalFacts) {
|
|
||||||
f.merges = []missedMerge{{Owner: "novox", Repo: "app", Base: "main", Commit: "c0ffee0011", At: f.now.Add(-11 * time.Minute)}}
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"S6": {
|
"S6": {
|
||||||
inside: func(f *signalFacts) {
|
inside: func(f *signalFacts) {
|
||||||
f.asks = []askFacts{{id: "build-1", seat: "node-build-agent", what: "novox/app", state: link.AskInFlight,
|
f.asks = []askFacts{{id: "build-1", seat: "node-build-agent", what: "novox/app", state: link.AskInFlight,
|
||||||
|
|||||||
@@ -70,10 +70,6 @@ type signalFacts struct {
|
|||||||
loop loopFacts
|
loop loopFacts
|
||||||
loopErr error
|
loopErr error
|
||||||
|
|
||||||
merges []missedMerge
|
|
||||||
mergesPassed time.Time
|
|
||||||
mergesErr error
|
|
||||||
|
|
||||||
asks []askFacts
|
asks []askFacts
|
||||||
asksErr error
|
asksErr error
|
||||||
|
|
||||||
@@ -358,10 +354,6 @@ func (w *watchdogs) gather(ctx context.Context) *signalFacts {
|
|||||||
f.batches, f.plansErr = gatherBatches(ctx, inv, now)
|
f.batches, f.plansErr = gatherBatches(ctx, inv, now)
|
||||||
}
|
}
|
||||||
f.loop, f.loopErr = w.gatherLoop()
|
f.loop, f.loopErr = w.gatherLoop()
|
||||||
f.mergesPassed, f.merges, f.mergesErr = watchedMerges.last()
|
|
||||||
if f.mergesErr == nil && !f.mergesPassed.IsZero() && now.Sub(f.mergesPassed) > 3*mergeCatchUpEvery {
|
|
||||||
f.mergesErr = fmt.Errorf("the catch-up of merges has not passed since %s", f.mergesPassed.UTC().Format(time.RFC3339))
|
|
||||||
}
|
|
||||||
f.asks, f.asksErr = w.gatherAsks(ctx, inv)
|
f.asks, f.asksErr = w.gatherAsks(ctx, inv)
|
||||||
if open, err := w.keeper.Open(ctx); err != nil {
|
if open, err := w.keeper.Open(ctx); err != nil {
|
||||||
f.standingsErr = err
|
f.standingsErr = err
|
||||||
|
|||||||
Reference in New Issue
Block a user