novox/mesh-controller
13
0
Fork 0
Code Issues Pull Requests 5 Releases Wiki Activity

A first merge into a new repository waits kept for mesh-delivery's word, and the word's arrival asks the builds, with no timer; AC7 on the bus in both orders (issue 1146)

Browse Source
This commit is contained in:
jschoubben
2026-10-19 00:16:36 +02:00
parent 19f15d8ea0
commit 4e12fc19b7
9 changed files with 301 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mesh-controller/batches.go
+3
View File
@@ -747,6 +747,9 @@ func carriedOf(merges []inventory.BatchedMerge) ([]inventory.PlanCommit, []inven
type keptMerge struct {
link.SourceMoved
Moves []string `json:"moves,omitempty"`
// FirstBuild says the merge is a first merge into a repository the mesh builds nothing from, kept to ask the first
// build of its modules once mesh-delivery's word on it is kept (novox/hq ADR 0432, issue 1146).
FirstBuild bool `json:"first_build,omitempty"`
}
// announcedOf is a kept merge as the forge announced it, and what it moved; empty where the record says none.
cmd/mesh-controller/first_builds.go
+91 -26
View File
@@ -12,46 +12,115 @@ import (
"github.com/novox/mesh-controller/internal/link"
)
// firstBuildWait is how long after the forge made a merge the controller keeps hearing it again for mesh-delivery's word
// on it, which may be on its way when the forge's announcement is heard (novox/hq ADR 0432).
const firstBuildWait = 2 * time.Minute
// firstBuilds asks the first build of every module a merge adds to a repository no registered module is built from, when
// the merge was announced as made on the operator's Approve of that head (novox/hq ADR 0432, issue 1146): the modules'
// directories are the ones the announcement says hold a module.json at the merge commit, as touchedBy reads them. The
// merge is kept, so the build's outcome is judged by the rule that registers it (mayRegisterFrom, approvedMove). Without
// the Approve nothing is asked, and it says why. Answers whether the merge was dealt with here.
// merge is kept, so the build's outcome is judged by the rule that registers it (mayRegisterFrom, approvedMove).
//
// **Complete in either order, with no timer** (ADR 0426): when mesh-delivery's word is kept first, it is found here; when
// the forge's merge is heard first, the merge is kept as a first merge waiting for the word, and keepingDeliveryMerges
// runs this again when the word is kept (resumeFirstBuilds). A word that says no (a Decline, another head) asks nothing and
// says why. Answers whether the merge was dealt with here.
func (f following) firstBuilds(ctx context.Context, m link.SourceMoved, entries []inventory.Entry,
read map[string][]inventory.ReadRepository, now time.Time) (bool, error) {
inv := f.open.inventory
repository := m.Owner + "/" + m.Repo
added, why, err := firstBuildsOf(ctx, inv, m, entries, read, now)
added, why, err := firstBuildsOf(ctx, inv, m, entries, read)
switch {
case errors.Is(err, errFirstNotApproved):
fmt.Printf("%s merged into %s (%.8s) adds %s, and no build is asked: %v\n", repository, m.Base, m.Commit,
strings.Join(addedDirs(m, entries, read), ", "), err)
return true, nil
case errors.Is(err, errNotAnnounced):
// Kept, waiting for mesh-delivery's word: a restart keeps it, and the word's arrival asks the builds.
if kerr := keepFirstMerge(ctx, inv, m, now, true); kerr != nil {
return true, notNow(kerr)
}
fmt.Printf("%s merged into %s (%.8s) adds %s; it is kept, and the first build is asked when mesh-delivery "+
"announces the merge on the operator's Approve\n", repository, m.Base, m.Commit,
strings.Join(addedDirs(m, entries, read), ", "))
return true, nil
case err != nil:
return true, notNow(err)
case len(added) == 0:
return false, nil
}
if err := askFirstBuilds(ctx, inv, m, now, added, why); err != nil {
return true, notNow(err)
}
return true, nil
}
// keepFirstMerge keeps a first merge as the controller heard it, waiting (no plan answers it).
func keepFirstMerge(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, now time.Time, waitingForWord bool) error {
merged, err := time.Parse(time.RFC3339Nano, m.MergedAt)
if err != nil {
merged = now
}
event, err := json.Marshal(keptMerge{SourceMoved: m})
event, err := json.Marshal(keptMerge{SourceMoved: m, FirstBuild: waitingForWord})
if err != nil {
return true, err
return err
}
if _, err := inv.AddMerge(ctx, inventory.BatchedMerge{Repository: repository, Branch: m.Base, Commit: m.Commit,
Merged: merged.UTC(), Heard: now, Event: event}); err != nil {
return true, notNow(err)
_, err = inv.AddMerge(ctx, inventory.BatchedMerge{Repository: m.Owner + "/" + m.Repo, Branch: m.Base, Commit: m.Commit,
Merged: merged.UTC(), Heard: now, Event: event})
return err
}
// askFirstBuilds keeps the merge, asks the builds, and marks the merge as no longer waiting for them.
func askFirstBuilds(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, now time.Time, added []string,
why string) error {
repository := m.Owner + "/" + m.Repo
if err := keepFirstMerge(ctx, inv, m, now, false); err != nil {
return err
}
fmt.Printf("%s merged into %s (%.8s) on the operator's Approve (%s): the mesh builds nothing from it yet, so the "+
"first build of %s is asked\n", repository, m.Base, m.Commit, why, strings.Join(added, ", "))
askBuilds(ctx, inv, m, buildSource{Repository: repository, Seat: gitSeat}, added)
return true, nil
// Answered: it no longer waits, so the word's arrival again asks nothing twice.
return inv.AnswerMerge(ctx, repository, m.Commit, inventory.WaitingRecord, false)
}
// resumeFirstBuilds runs firstBuilds for the first merges waiting for the word mesh-delivery just announced for a pull
// request's head: the other order of the same two events.
func (f following) resumeFirstBuilds(ctx context.Context, d deliveryMerge) error {
actingOnMerges.Lock()
defer actingOnMerges.Unlock()
inv := f.open.inventory
waiting, err := inv.WaitingMerges(ctx)
if err != nil {
return notNow(err)
}
for _, w := range waiting {
k := announcedOf(w)
if !k.FirstBuild || !strings.EqualFold(k.Owner+"/"+k.Repo, d.Owner+"/"+d.Repo) || k.Number != d.Number ||
k.HeadSHA != d.HeadSHA {
continue
}
entries, err := inv.Catalogued(ctx)
if err != nil {
return notNow(err)
}
read, err := readForPlanning(ctx, inv)
if err != nil {
return notNow(err)
}
added, why, err := firstBuildsOf(ctx, inv, k.SourceMoved, entries, read)
switch {
case errors.Is(err, errFirstNotApproved):
fmt.Printf("%s/%s merged into %s (%.8s) was waiting for mesh-delivery's word, and no build is asked: %v\n",
k.Owner, k.Repo, k.Base, k.Commit, err)
if aerr := inv.AnswerMerge(ctx, w.Repository, w.Commit, inventory.WaitingRecord, false); aerr != nil {
return notNow(aerr)
}
case err != nil:
return notNow(err)
case len(added) > 0:
if err := askFirstBuilds(ctx, inv, k.SourceMoved, time.Now().UTC(), added, why); err != nil {
return notNow(err)
}
}
}
return nil
}
// errFirstNotApproved says a merge into a repository the mesh builds nothing from is not one the operator approved.
@@ -74,25 +143,21 @@ func addedDirs(m link.SourceMoved, entries []inventory.Entry, read map[string][]
}
// firstBuildsOf is the directories whose first build a merge asks, and the warrant it was made on; none when the merge
// is no first merge into a repository the mesh builds nothing from. errFirstNotApproved when it is, and the operator's
// Approve of its head is not on record, a try-again error while mesh-delivery's word may still be on its way.
// is no first merge into a repository the mesh builds nothing from. errFirstNotApproved when it is and the word on record
// is no Approve of its head; errNotAnnounced when mesh-delivery's word is not kept yet.
func firstBuildsOf(ctx context.Context, inv catalogueReader, m link.SourceMoved, entries []inventory.Entry,
read map[string][]inventory.ReadRepository, now time.Time) ([]string, string, error) {
read map[string][]inventory.ReadRepository) ([]string, string, error) {
added := addedDirs(m, entries, read)
if len(added) == 0 {
return nil, "", nil
}
why, err := approvedAnnounced(ctx, inv, m.Owner, m.Repo, keptMerge{SourceMoved: m})
if err == nil {
switch {
case err == nil:
return added, why, nil
case errors.Is(err, errWordNotAnApprove):
return nil, "", fmt.Errorf("%w (%v)", errFirstNotApproved, err)
}
merged, perr := time.Parse(time.RFC3339Nano, m.MergedAt)
if errors.Is(err, errNotAnnounced) && perr == nil && now.Sub(merged) < firstBuildWait {
return nil, "", fmt.Errorf("%w: %w", link.ErrTryAgain, err)
}
var nope error = fmt.Errorf("%w (%v)", errFirstNotApproved, err)
if inventory.Unreachable(err) {
nope = err
}
return nil, "", nope
// Not announced yet, or the bus or the store could not be asked: the caller waits or tries again, and asks nothing.
return nil, "", err
}
cmd/mesh-controller/first_builds_integration_test.go
+128 -49
View File
@@ -4,65 +4,144 @@ package main
import (
"context"
"encoding/json"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/asks"
"github.com/nats-io/nats.go/jetstream"
"github.com/novox/mesh-controller/internal/broker"
"github.com/novox/mesh-controller/internal/catalogue"
"github.com/novox/mesh-controller/internal/inventory"
"github.com/novox/mesh-controller/internal/link"
"github.com/novox/mesh-controller/internal/testbus"
)
// 1146/AC7: mesh-delivery #1 merged on the operator's Approve into a repository the mesh builds nothing from. The forge's
// announcement, heard as the controller hears it (SourceMoved, hearMerge), asks the first build of the module the merge
// commit holds; the build's outcome registers it; and the module is in the catalogue. Without the Approve nothing is
// asked. Store-backed: skips without one.
// 1146/AC7, whole against a real bus and the store, in both orders of the two events (novox/hq ADR 0432, issue 1146):
// mesh-delivery #7 merged on the operator's Approve into a repository the mesh builds nothing from. The forge's merge,
// heard as the controller hears it (SourceMoved), and mesh-delivery's word, published on the subject the controller
// consumes with the router's record of the ask in its bucket, come in either order; each way the first build of the
// module the merge commit holds is asked exactly once, its outcome registers the module, and it is in the catalogue.
// Without the Approve in the record nothing is asked. Store-backed: skips without one.
func TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegistersTheModule(t *testing.T) {
const head, commit = "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1", "0177ea7a00000000"
for _, approved := range []bool{true, false} {
open := aMesh(t)
ctx := t.Context()
inv := open.inventory
asked := asksWithPaths(t)
a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@a1a1a1a1a1a1": head})
a.merge.Number = 1
wasKept, wasAsk := keptDeliveryMerge, routerAskOf
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
return a.merge, approved, nil
}
routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) {
return &a.ask, &a.warrant, nil
}
t.Cleanup(func() { keptDeliveryMerge, routerAskOf = wasKept, wasAsk })
m := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: commit, Number: 1,
MergedAt: time.Now().Add(-time.Hour).UTC().Format(time.RFC3339Nano), Paths: []string{"module.json", "main.go"},
ModuleDirs: []string{"."}, ModuleDirsSaid: true}
if err := (following{open: open}).SourceMoved(ctx, m); err != nil {
t.Fatalf("approved %v: %v", approved, err)
}
if !approved {
if len(*asked) != 0 {
t.Fatalf("1146/AC7: a merge with no Approve asked %v", *asked)
for _, c := range []struct {
what string
wordFirst bool
approved bool // the router's record holds the operator's Approve of the merged head
wantRegiter bool
}{
{what: "the forge's merge before mesh-delivery's word", approved: true, wantRegiter: true},
{what: "mesh-delivery's word before the forge's merge", wordFirst: true, approved: true, wantRegiter: true},
{what: "a word whose record holds no Approve", wordFirst: true},
} {
t.Run(c.what, func(t *testing.T) {
open := aMesh(t)
ctx := t.Context()
inv := open.inventory
asked := asksWithPaths(t)
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "operator-channel", Version: "1",
DefinesSeats: []catalogue.SeatDeclaration{{Name: broker.AsksSeat, Records: []string{"asks"}}}},
inventory.Source{Repository: "novox/mesh-catalog", Seat: "git", Path: "modules/operator-channel", BuiltFrom: "c0ffee"}); err != nil {
t.Fatal(err)
}
continue
}
want := [3]string{"novox/mesh-delivery", "", "main"}
if len(*asked) != 1 || (*asked)[0] != want {
t.Fatalf("1146/AC7: the merge asked %v, want %v", *asked, want)
}
build := onTrunk("b-", "novox/mesh-delivery", "git", "", map[string]any{"module": "mesh-delivery-view", "version": "1"})
build.Commit = commit
if _, _, err := takeIn(ctx, inv, build); err != nil {
t.Fatalf("1146/AC7: the outcome of the first build was refused: %v", err)
}
src, err := inv.SourceOf(ctx, "mesh-delivery-view")
if err != nil || src.Repository != "novox/mesh-delivery" {
t.Fatalf("1146/AC7: the module is built from %q (%v)", src.Repository, err)
}
if _, known := inventoryHolds(t, inv, "mesh-delivery-view"); !known {
t.Fatal("1146/AC7: the module is not in the catalogue")
}
js, err := broker.Dial(testbus.URL(t))
if err != nil {
t.Fatal(err)
}
t.Cleanup(js.Close)
if err := broker.AssertMeshStreams(js); err != nil {
t.Fatal(err)
}
if err := js.EnsureControllerBuckets(); err != nil {
t.Fatal(err)
}
api, err := jetstream.New(js.Conn())
if err != nil {
t.Fatal(err)
}
records, err := api.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: broker.BucketName("operator-channel", "asks")})
if err != nil {
t.Fatal(err)
}
a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@a1a1a1a1a1a1": head})
if c.approved {
record, _ := json.Marshal(map[string]any{"state": "closed", "ask": a.ask, "warrant": a.warrant})
if _, err := records.Put(ctx, theDelivery+"."+a.merge.AskID, record); err != nil {
t.Fatal(err)
}
}
following := following{open: open}
server := link.ConnectNats(js, nil, noListener{})
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream, kept: following.resumeFirstBuilds}); err != nil {
t.Fatal(err)
}
if err := broker.AssertServingConsumers(js); err != nil {
t.Fatal(err)
}
serving, stop := context.WithCancel(context.Background())
done := make(chan struct{})
go func() { defer close(done); _ = server.Serve(serving) }()
t.Cleanup(func() { stop(); <-done })
servingBus.Store(js)
t.Cleanup(func() { servingBus.Store(nil) })
word := func() {
t.Helper()
body, _ := json.Marshal(a.merge)
if _, err := js.Context().Publish(broker.DeliveryMergedSubject, body); err != nil {
t.Fatal(err)
}
becomes(t, 15*time.Second, "the word being kept", func() bool {
_, kept, err := keptDeliveryMerge(ctx, "novox", "mesh-delivery", 7, head)
return err == nil && kept
})
time.Sleep(500 * time.Millisecond) // the keeper's hook runs as the word is kept
}
forge := func() {
t.Helper()
m := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: commit, Number: 7,
MergedAt: time.Now().Add(-time.Second).UTC().Format(time.RFC3339Nano), Paths: []string{"module.json", "main.go"},
ModuleDirs: []string{"."}, ModuleDirsSaid: true}
if err := following.SourceMoved(ctx, m); err != nil {
t.Fatal(err)
}
}
if c.wordFirst {
word()
forge()
} else {
forge()
if len(*asked) != 0 {
t.Fatalf("1146/AC7: the merge asked %v before mesh-delivery's word was kept", *asked)
}
word()
}
if !c.wantRegiter {
if len(*asked) != 0 {
t.Fatalf("1146/AC7: a merge whose record holds no Approve asked %v", *asked)
}
return
}
want := [3]string{"novox/mesh-delivery", "", "main"}
becomes(t, 10*time.Second, "the first build asked", func() bool { return len(*asked) >= 1 })
if len(*asked) != 1 || (*asked)[0] != want {
t.Fatalf("1146/AC7: the merge asked %v, want %v once", *asked, want)
}
build := onTrunk("b-", "novox/mesh-delivery", "git", "", map[string]any{"module": "mesh-delivery-view", "version": "1"})
build.Commit = commit
if _, _, err := takeIn(ctx, inv, build); err != nil {
t.Fatalf("1146/AC7: the outcome of the first build was refused: %v", err)
}
src, err := inv.SourceOf(ctx, "mesh-delivery-view")
if err != nil || src.Repository != "novox/mesh-delivery" {
t.Fatalf("1146/AC7: the module is built from %q (%v)", src.Repository, err)
}
if _, known := inventoryHolds(t, inv, "mesh-delivery-view"); !known {
t.Fatal("1146/AC7: the module is not in the catalogue")
}
})
}
}
cmd/mesh-controller/first_builds_test.go
+6 -9
View File
@@ -28,16 +28,13 @@ func TestAFirstMergeIntoANewRepositoryAsksItsModulesBuildsOnlyOnTheApprove(t *te
unkept bool
dirs []string
refused bool
retry bool
waits bool
}{
{what: "the Approve of the merged head", dirs: []string{"view", "watchman"}},
{what: "a Decline", refused: true, change: func(a *anApprovedMerge, _ *link.SourceMoved, _ *[]inventory.Entry) {
a.warrant.Option, a.warrant.Label, a.merge.Option = "decline", "Decline", "decline"
}},
{what: "no word of mesh-delivery on a merge made long ago", unkept: true, refused: true},
{what: "no word yet on a merge just made", unkept: true, retry: true, change: func(_ *anApprovedMerge, m *link.SourceMoved, _ *[]inventory.Entry) {
m.MergedAt = now.Add(-time.Second).Format(time.RFC3339Nano)
}},
{what: "no word of mesh-delivery yet, however long ago the merge was made", unkept: true, waits: true},
{what: "a module already built from the repository", change: func(_ *anApprovedMerge, _ *link.SourceMoved, e *[]inventory.Entry) {
*e = append(*e, inventory.Entry{Source: inventory.Source{Repository: "novox/mesh-delivery", Seat: "git"}})
}},
@@ -61,13 +58,13 @@ func TestAFirstMergeIntoANewRepositoryAsksItsModulesBuildsOnlyOnTheApprove(t *te
routerAskOf = func(context.Context, catalogueReader, string) (*asks.Ask, *asks.Warrant, error) {
return &a.ask, &a.warrant, nil
}
dirs, _, err := firstBuildsOf(t.Context(), nil, m, entries, nil, now)
dirs, _, err := firstBuildsOf(t.Context(), nil, m, entries, nil)
switch {
case c.retry && !errors.Is(err, link.ErrTryAgain):
t.Fatalf("1146/AC6: %s: want a try-again, got %v", c.what, err)
case c.waits && !errors.Is(err, errNotAnnounced):
t.Fatalf("1146/AC6: %s: want a wait for the word, got %v", c.what, err)
case c.refused && !errors.Is(err, errFirstNotApproved):
t.Fatalf("1146/AC6: %s: want the Approve missing, got %v (dirs %v)", c.what, err, dirs)
case !c.refused && !c.retry && err != nil:
case !c.refused && !c.waits && err != nil:
t.Fatalf("1146/AC6: %s: %v", c.what, err)
case !slices.Equal(dirs, c.dirs):
t.Fatalf("1146/AC6: %s: first builds asked of %v, want %v", c.what, dirs, c.dirs)
cmd/mesh-controller/moved.go
+12 -2
View File
@@ -115,6 +115,9 @@ func movedOnApprove(d deliveryMerge, ask *asks.Ask, w *asks.Warrant) (string, er
// keepingDeliveryMerges keeps each merge mesh-delivery announces in the controller's bucket for them.
type keepingDeliveryMerges struct {
js func() *broker.JetStream
// kept, when set, is told each merge once it is kept: a first merge waiting for the word asks its builds then
// (novox/hq ADR 0432, issue 1146).
kept func(ctx context.Context, d deliveryMerge) error
}
func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, body []byte) error {
@@ -137,6 +140,9 @@ func (k keepingDeliveryMerges) KeepDeliveryMerge(ctx context.Context, body []byt
if _, err := kv.Put(ctx, deliveryMergeKey(d.Owner, d.Repo, d.Number, d.HeadSHA), body); err != nil {
return fmt.Errorf("%w: %w", link.ErrTryAgain, err)
}
if k.kept != nil {
return k.kept(ctx, d)
}
return nil
}
@@ -239,6 +245,10 @@ func approvedMove(ctx context.Context, inv registrations, forms *sourceForms, bu
return approvedAnnounced(ctx, inv, owner, name, announcedOf(merged))
}
// errWordNotAnApprove says the merge cannot be shown to be the operator's Approve of its head: its announcement names no
// pull request, or the word on record is not that Approve.
var errWordNotAnApprove = errors.New("the word on record is no Approve of the head merged")
// errNotAnnounced says mesh-delivery's word on a merge is not kept (yet): the merge may have been made by someone else,
// or its word may still be on its way.
var errNotAnnounced = errors.New("mesh-delivery announced no such merge")
@@ -248,7 +258,7 @@ var errNotAnnounced = errors.New("mesh-delivery announced no such merge")
// module and the first build of a new repository's modules (ADR 0420 §7, ADR 0432).
func approvedAnnounced(ctx context.Context, inv catalogueReader, owner, name string, k keptMerge) (string, error) {
if k.Number <= 0 || k.HeadSHA == "" {
return "", errors.New("the forge's announcement of the merge names no pull request or head")
return "", fmt.Errorf("%w: the forge's announcement of the merge names no pull request or head", errWordNotAnApprove)
}
d, kept, err := keptDeliveryMerge(ctx, owner, name, k.Number, k.HeadSHA)
if err != nil {
@@ -263,7 +273,7 @@ func approvedAnnounced(ctx context.Context, inv catalogueReader, owner, name str
}
why, err := movedOnApprove(d, ask, w)
if err != nil {
return "", err
return "", fmt.Errorf("%w: %w", errWordNotAnApprove, err)
}
return fmt.Sprintf("%s/%s#%d merged at %.8s on %s", owner, name, k.Number, k.HeadSHA, why), nil
}
cmd/mesh-controller/push.go
+2 -1
View File
@@ -209,7 +209,8 @@ func serve(ctx context.Context) (err error) {
}
// 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}); err != nil {
if err := server.KeepsDeliveryMerges(keepingDeliveryMerges{js: server.JetStream,
kept: following{open: open}.resumeFirstBuilds}); err != nil {
return err
}
// No timer reads the merges back (novox/hq ADR 0407 decision 3, issue 973): the forge's event and the durable