jochen
59f4d486b1
The controller acted only on what its events consumer handed it, so a merge the bus skipped left modules behind with nothing said. The stream is now read back every five minutes on a single-filter consumer, and any merge that would still move a module after ten minutes is said and acted on.
181 lines
6.9 KiB
Go
181 lines
6.9 KiB
Go
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)
|
|
}
|
|
}
|