Merge remote-tracking branch 'origin/main' into feat/delivery-watch
mesh/repo-check head ac79e29319ee merged into trunk 5939bbdefe5a — pass: its merge-check.sh passed
mesh/merge-gate head ac79e29319ee merged into trunk 5939bbdefe5a — pass: group feat/delivery-watch composed: every machine composes with the change as it d…
mesh/delivery delivered
mesh/delivery-group group feat/delivery-watch delivered: every member is delivered
mesh/repo-check head ac79e29319ee merged into trunk 5939bbdefe5a — pass: its merge-check.sh passed
mesh/merge-gate head ac79e29319ee merged into trunk 5939bbdefe5a — pass: group feat/delivery-watch composed: every machine composes with the change as it d…
mesh/delivery delivered
mesh/delivery-group group feat/delivery-watch delivered: every member is delivered
# Conflicts: # cmd/mesh-controller/doctor.go
This commit is contained in:
@@ -176,6 +176,11 @@ func (f following) hearMerge(ctx context.Context, m link.SourceMoved, now time.T
|
||||
repository, m.Base, m.Commit, already)
|
||||
return nil
|
||||
}
|
||||
if already == 0 {
|
||||
if done, err := f.firstBuilds(ctx, m, entries, read, now); err != nil || done {
|
||||
return err
|
||||
}
|
||||
}
|
||||
fmt.Print(nothingReadsIt(repository, m))
|
||||
return nil
|
||||
}
|
||||
@@ -742,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.
|
||||
|
||||
@@ -125,6 +125,11 @@ var probeRegistry = []probe{
|
||||
{ID: probeStuckID, Asserts: "no key the watchman says stopped moving is left unsaid: each key of its stuck bucket " +
|
||||
"not moved is a condition of the delivery scope until it says moved", From: "ADR 0433 §2, issue 1156",
|
||||
Kind: kindStuck, Phase: 3, run: probeStuck},
|
||||
// A first merge into a repository the mesh builds nothing from, kept waiting for mesh-delivery's word on the
|
||||
// operator's Approve (novox/hq ADR 0432, issue 1146): said past a bound, never only in a log line (ADR 0426 §3, §8).
|
||||
{ID: probeFirstBuildsID, Asserts: "no first merge into a repository the mesh builds nothing from waits more than ten " +
|
||||
"minutes for mesh-delivery's announcement of the operator's Approve of its head", From: "ADR 0432, issue 1146",
|
||||
Kind: kindFirstBuildUnasked, Phase: 3, run: probeFirstBuilds},
|
||||
// The delivery budgets (novox/hq ADR 0282 decision 7): the newest delivery of each class within its budget,
|
||||
// read from mesh-delivery's `times`; a measurement said, never a delivery held.
|
||||
{ID: probeBudgetsID, Asserts: "the newest delivery of a leaf module ran on every machine within three minutes of " +
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// 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).
|
||||
//
|
||||
// **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)
|
||||
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, FirstBuild: waitingForWord})
|
||||
if err != nil {
|
||||
return 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)
|
||||
// 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.
|
||||
var errFirstNotApproved = errors.New("it is no merge made on the operator's Approve of its head, so a hand `build` " +
|
||||
"registers a module from a repository the mesh does not build yet")
|
||||
|
||||
// addedDirs are the module directories the merge adds to a repository no registered module is built from, none where
|
||||
// the merge is not into the trunk or some module is built from the repository already.
|
||||
func addedDirs(m link.SourceMoved, entries []inventory.Entry, read map[string][]inventory.ReadRepository) []string {
|
||||
if m.Base != "main" {
|
||||
return nil
|
||||
}
|
||||
for _, e := range entries {
|
||||
if !e.Provided && sameRepository(e.Source.Repository, m) {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
_, added, _ := touchedBy(nil, entries, m, read)
|
||||
return added
|
||||
}
|
||||
|
||||
// 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 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) ([]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})
|
||||
switch {
|
||||
case err == nil:
|
||||
return added, why, nil
|
||||
case errors.Is(err, errWordNotAnApprove):
|
||||
return nil, "", fmt.Errorf("%w (%v)", errFirstNotApproved, err)
|
||||
}
|
||||
// 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
|
||||
}
|
||||
|
||||
// probeFirstBuildsID is the probe that reads the first merges waiting for mesh-delivery's word.
|
||||
const probeFirstBuildsID = "D18"
|
||||
|
||||
// kindFirstBuildUnasked is what a first merge waiting too long for mesh-delivery's word raises.
|
||||
const kindFirstBuildUnasked = "first-build-unasked"
|
||||
|
||||
// firstBuildWaitBound is how long a first merge waits for mesh-delivery's announcement of the operator's Approve before
|
||||
// it is said: a bound in the sense of ADR 0426 §3, not a delay. Nothing waits on it: the word's arrival asks the builds
|
||||
// at once, and the merge is said, never dropped, once it is past this (issue 1146).
|
||||
const firstBuildWaitBound = 10 * time.Minute
|
||||
|
||||
// firstBuildObservations are the conditions of the first merges waiting past the bound, one for each; a merge answered is
|
||||
// no longer listed as waiting, so its condition clears with the next look.
|
||||
func firstBuildObservations(waiting []inventory.BatchedMerge, now time.Time) []conditions.Observation {
|
||||
var out []conditions.Observation
|
||||
for _, w := range waiting {
|
||||
k := announcedOf(w)
|
||||
if !k.FirstBuild || now.Sub(w.Heard) <= firstBuildWaitBound {
|
||||
continue
|
||||
}
|
||||
repository := k.Owner + "/" + k.Repo
|
||||
for_ := now.Sub(w.Heard).Round(time.Minute)
|
||||
out = append(out, conditions.Observation{Scope: conditions.ScopeMerge,
|
||||
ID: fmt.Sprintf("%s_%.8s", repository, k.Commit), Kind: kindFirstBuildUnasked, Severity: conditions.Warning,
|
||||
Summary: fmt.Sprintf("the merge %.8s of %s adds modules the mesh builds nothing from yet, and has waited %s for "+
|
||||
"mesh-delivery's announcement of the operator's Approve of its head: no build is asked", k.Commit, repository, for_),
|
||||
Said: fmt.Sprintf("%s at %s waits for mesh-delivery's word on pull request %d (head %.8s)", repository, k.Commit,
|
||||
k.Number, k.HeadSHA),
|
||||
Headline: fmt.Sprintf("The first build of %s is not asked: no Approve was announced", repository),
|
||||
Explanation: fmt.Sprintf("The merge %.8s into %s adds modules the mesh builds nothing from yet. It registers them "+
|
||||
"only when mesh-delivery announces that it merged the pull request on your Approve of that head, and no such "+
|
||||
"announcement has come in %s. The merge is kept, and the builds are asked the moment the announcement is kept.",
|
||||
k.Commit, repository, for_),
|
||||
Resolved: "mesh-delivery announces the merge, or a hand `build` registers the modules",
|
||||
})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// probeFirstBuilds is D18: no first merge waits past its bound for mesh-delivery's word unsaid.
|
||||
func probeFirstBuilds(ctx context.Context, d *doctor) ([]conditions.Observation, error) {
|
||||
waiting, err := d.open.inventory.WaitingMerges(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return firstBuildObservations(waiting, time.Now().UTC()), nil
|
||||
}
|
||||
@@ -0,0 +1,160 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"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, 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 _, 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)
|
||||
}
|
||||
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")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func inventoryHolds(t *testing.T, inv *inventory.Inventory, module string) (inventory.Entry, bool) {
|
||||
t.Helper()
|
||||
entries, err := inv.Catalogued(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, e := range entries {
|
||||
if e.Manifest.Module == module {
|
||||
return e, true
|
||||
}
|
||||
}
|
||||
return inventory.Entry{}, false
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/asks"
|
||||
"github.com/novox/mesh-controller/internal/conditions"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// 1146/AC6: the first merge into a repository no registered module is built from asks the first build of every module
|
||||
// the merge commit holds when it was made on the operator's Approve of that head, and asks nothing without it.
|
||||
func TestAFirstMergeIntoANewRepositoryAsksItsModulesBuildsOnlyOnTheApprove(t *testing.T) {
|
||||
const head = "a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1a1"
|
||||
now := time.Date(2026, 10, 19, 0, 6, 0, 0, time.UTC)
|
||||
mergedAt := now.Add(-time.Hour).Format(time.RFC3339Nano)
|
||||
heard := link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main", HeadSHA: head, Commit: "0177ea7a",
|
||||
Number: 1, MergedAt: mergedAt, ModuleDirsSaid: true, ModuleDirs: []string{"watchman", "view"},
|
||||
Paths: []string{"view/module.json", "watchman/module.json", "README.md"}}
|
||||
built := inventory.Entry{Source: inventory.Source{Repository: "novox/mesh-catalog", Seat: "git"}}
|
||||
for _, c := range []struct {
|
||||
what string
|
||||
change func(*anApprovedMerge, *link.SourceMoved, *[]inventory.Entry)
|
||||
unkept bool
|
||||
dirs []string
|
||||
refused 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 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"}})
|
||||
}},
|
||||
{what: "a merge into another branch", change: func(_ *anApprovedMerge, m *link.SourceMoved, _ *[]inventory.Entry) { m.Base = "feat" }},
|
||||
{what: "a merge that holds no module.json", change: func(_ *anApprovedMerge, m *link.SourceMoved, _ *[]inventory.Entry) {
|
||||
m.ModuleDirs, m.Paths = nil, []string{"README.md"}
|
||||
}},
|
||||
} {
|
||||
t.Run(c.what, func(t *testing.T) {
|
||||
a := aDeliveryMergeOf(t, head, map[string]string{"novox/mesh-delivery@a1a1a1a1a1a1": head})
|
||||
a.merge.Number = 1
|
||||
m, entries := heard, []inventory.Entry{built}
|
||||
if c.change != nil {
|
||||
c.change(&a, &m, &entries)
|
||||
}
|
||||
wasKept, wasAsk := keptDeliveryMerge, routerAskOf
|
||||
t.Cleanup(func() { keptDeliveryMerge, routerAskOf = wasKept, wasAsk })
|
||||
keptDeliveryMerge = func(context.Context, string, string, int, string) (deliveryMerge, bool, error) {
|
||||
return a.merge, !c.unkept, nil
|
||||
}
|
||||
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)
|
||||
switch {
|
||||
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.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)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// 1146/AC8: a first merge waiting past the bound for mesh-delivery's word is said, naming the repository, the commit and
|
||||
// the missing word; one inside the bound, one that is no first merge, and one answered (no longer waiting) are not.
|
||||
func TestAFirstMergeWaitingPastItsBoundIsSaidAndClearsWhenAnswered(t *testing.T) {
|
||||
now := time.Date(2026, 10, 19, 1, 0, 0, 0, time.UTC)
|
||||
waiting := func(commit string, first bool, heard time.Time) inventory.BatchedMerge {
|
||||
event, _ := json.Marshal(keptMerge{SourceMoved: link.SourceMoved{Owner: "novox", Repo: "mesh-delivery", Base: "main",
|
||||
Commit: commit, Number: 1, HeadSHA: "a1a1a1a1a1a1"}, FirstBuild: first})
|
||||
return inventory.BatchedMerge{Repository: "novox/mesh-delivery", Branch: "main", Commit: commit, Heard: heard, Event: event}
|
||||
}
|
||||
got := firstBuildObservations([]inventory.BatchedMerge{
|
||||
waiting("0177ea7a00", true, now.Add(-firstBuildWaitBound-time.Minute)),
|
||||
waiting("1111111100", true, now.Add(-time.Minute)),
|
||||
waiting("2222222200", false, now.Add(-time.Hour)),
|
||||
}, now)
|
||||
if len(got) != 1 {
|
||||
t.Fatalf("1146/AC8: %d conditions raised, want the one merge past the bound: %+v", len(got), got)
|
||||
}
|
||||
o := got[0]
|
||||
if o.Kind != kindFirstBuildUnasked || o.Scope != conditions.ScopeMerge || o.Severity != conditions.Warning ||
|
||||
!strings.Contains(o.Summary, "novox/mesh-delivery") || !strings.Contains(o.Summary, "0177ea7a") ||
|
||||
!strings.Contains(o.Summary, "announcement of the operator's Approve") {
|
||||
t.Fatalf("1146/AC8: the condition is %+v", o)
|
||||
}
|
||||
// Answered, the merge is no longer listed as waiting: nothing is observed, and the condition clears.
|
||||
if cleared := firstBuildObservations(nil, now); len(cleared) != 0 {
|
||||
t.Fatalf("1146/AC8: an answered merge still raises %+v", cleared)
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -236,16 +242,30 @@ func approvedMove(ctx context.Context, inv registrations, forms *sourceForms, bu
|
||||
if !known {
|
||||
return "", fmt.Errorf("the controller kept no merge of %s/%s at %.8s", owner, name, r.Commit)
|
||||
}
|
||||
k := announcedOf(merged)
|
||||
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")
|
||||
|
||||
// approvedAnnounced is the warrant of the merge the forge announced as k: mesh-delivery announced it, and the ask it made
|
||||
// it on holds the operator's Approve of that head. Why it holds, or why not. Shared by the approved move of a known
|
||||
// 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 {
|
||||
return "", err
|
||||
}
|
||||
if !kept {
|
||||
return "", fmt.Errorf("mesh-delivery announced no merge of %s/%s#%d at %.8s", owner, name, k.Number, k.HeadSHA)
|
||||
return "", fmt.Errorf("%w: %s/%s#%d at %.8s", errNotAnnounced, owner, name, k.Number, k.HeadSHA)
|
||||
}
|
||||
ask, w, err := routerAskOf(ctx, inv, d.AskID)
|
||||
if err != nil {
|
||||
@@ -253,7 +273,7 @@ func approvedMove(ctx context.Context, inv registrations, forms *sourceForms, bu
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -250,6 +250,68 @@ func appliedEach(module string, since func(node string) time.Time, running []str
|
||||
return len(waiting) == 0, waiting
|
||||
}
|
||||
|
||||
// gateNeed is a module the next tier waits for: the machines running it, and from when each must have reported.
|
||||
type gateNeed struct {
|
||||
module string
|
||||
running []string
|
||||
since func(node string) time.Time
|
||||
}
|
||||
|
||||
// silentNodes is the machines a walk sent and heard nothing from past ApplySilentAfter, on whichever tier it sent
|
||||
// them (novox/hq ADR 0360 §5, issue 1160): the one rule both the gates between tiers and the walk's end read, through
|
||||
// walkReports. A machine that reports on its send is not silent from then on.
|
||||
func silentNodes(p inventory.Plan, reports []inventory.Reported, now time.Time, prov inventory.ProvLookup) map[string]bool {
|
||||
_, _, silent := walkReports(p, reports, now, prov)
|
||||
out := map[string]bool{}
|
||||
for _, n := range silent {
|
||||
out[n] = true
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// withoutSilent is the machines of running that are not silent: a later tier is not sent to a machine the walk heard
|
||||
// nothing from, which keeps its want of the plan's builds for the send it gets when it is heard again (ADR 0360 §5).
|
||||
func withoutSilent(running []string, silent map[string]bool) []string {
|
||||
return slices.DeleteFunc(slices.Clone(running), func(n string) bool { return silent[n] })
|
||||
}
|
||||
|
||||
// settleTierGate is the wait between tiers on the reports of the machines running what the next tier needs: what is
|
||||
// still waited for ("module on a, b"), the silent machines left out of the wait, sorted, and whether every machine the
|
||||
// gate waits on is silent, so that nothing it needs was delivered anywhere (issue 1160).
|
||||
func settleTierGate(needs []gateNeed, reports []inventory.Reported, silent map[string]bool) (waiting, quiet []string, lost bool) {
|
||||
all, silentAll := 0, 0
|
||||
seenQuiet := map[string]bool{}
|
||||
seenNode := map[string]bool{}
|
||||
for _, need := range needs {
|
||||
for _, n := range need.running {
|
||||
if !seenNode[n] {
|
||||
seenNode[n] = true
|
||||
all++
|
||||
if silent[n] {
|
||||
silentAll++
|
||||
}
|
||||
}
|
||||
}
|
||||
_, on := appliedEach(need.module, need.since, need.running, reports)
|
||||
var still []string
|
||||
for _, n := range on {
|
||||
if silent[n] {
|
||||
if !seenQuiet[n] {
|
||||
seenQuiet[n] = true
|
||||
quiet = append(quiet, n)
|
||||
}
|
||||
continue
|
||||
}
|
||||
still = append(still, n)
|
||||
}
|
||||
if len(still) > 0 {
|
||||
waiting = append(waiting, fmt.Sprintf("%s on %s", need.module, strings.Join(still, ", ")))
|
||||
}
|
||||
}
|
||||
sort.Strings(quiet)
|
||||
return waiting, quiet, all > 0 && silentAll == all
|
||||
}
|
||||
|
||||
func applied(module string, builtAt time.Time, running []string, reports []inventory.Reported) (bool, []string) {
|
||||
at := map[string]*time.Time{}
|
||||
for _, r := range reports {
|
||||
@@ -869,6 +931,13 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
// being behind. Now every module of the tier whose first machine is the same is sent there in one
|
||||
// send, judged there by one gate — kept on the first of them, the others gated by it — and the rest
|
||||
// of the tier's machines are sent once, together, when the gates have passed.
|
||||
// **A machine silent past the bound is left as it was, on every tier** (novox/hq ADR 0360 §5, issue 1160): the
|
||||
// walk's end names it, and the gates between tiers and the sends of the tiers after do not wait for it or go to it.
|
||||
allReports, err := inv.LastReports(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
silent := silentNodes(*p, allReports, time.Now().UTC(), provLookup(ctx))
|
||||
var pending []string
|
||||
firstOn := map[string][]string{}
|
||||
runningOf := map[string][]string{}
|
||||
@@ -903,6 +972,9 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
if goes != nil {
|
||||
running = slices.DeleteFunc(running, func(n string) bool { return !goes[n] })
|
||||
}
|
||||
if state.FirstAt == nil {
|
||||
running = withoutSilent(running, silent)
|
||||
}
|
||||
runningOf[m] = running
|
||||
// **A build whose source is unchanged is never a move** (novox/hq issue 280): made from the source
|
||||
// of the build every machine running it was sent, it is registered with that build's artifacts —
|
||||
@@ -1096,13 +1168,14 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
return changed, nil
|
||||
}
|
||||
// And wait for what the next tier needs running.
|
||||
leftAsWas := ""
|
||||
needed := gates(*p, edges, rollsOut)
|
||||
if len(needed) > 0 {
|
||||
reports, err := inv.LastReports(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
var waiting []string
|
||||
var needs []gateNeed
|
||||
for _, m := range needed {
|
||||
running, err := inv.Running(ctx, m)
|
||||
if err != nil {
|
||||
@@ -1133,14 +1206,12 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
for _, r := range reports {
|
||||
byNode[r.Node] = r
|
||||
}
|
||||
if ok, on := appliedEach(m, func(node string) time.Time {
|
||||
needs = append(needs, gateNeed{m, running, func(node string) time.Time {
|
||||
if s := byNode[node].Sent; s != nil {
|
||||
return *s
|
||||
}
|
||||
return time.Time{}
|
||||
}, running, reports); !ok {
|
||||
waiting = append(waiting, fmt.Sprintf("%s on %s", m, strings.Join(on, ", ")))
|
||||
}
|
||||
}})
|
||||
continue
|
||||
}
|
||||
// **Each machine from its own send** (novox/hq issue 256). With one machine first (ADR 0218)
|
||||
@@ -1160,9 +1231,23 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
}
|
||||
return at
|
||||
}
|
||||
if ok, on := appliedEach(m, sinceFor, running, reports); !ok {
|
||||
waiting = append(waiting, fmt.Sprintf("%s on %s", m, strings.Join(on, ", ")))
|
||||
needs = append(needs, gateNeed{m, running, sinceFor})
|
||||
}
|
||||
waiting, quiet, lost := settleTierGate(needs, reports, silent)
|
||||
if lost {
|
||||
// Every machine the next tier waits on is silent: nothing it needs was delivered anywhere, so the plan does
|
||||
// not go on as if it had been (issue 1160).
|
||||
p.State = inventory.PlanFailed
|
||||
p.Note = fmt.Sprintf("tier %d built; every machine it waits on said nothing on its send within %s: %s",
|
||||
p.Tier, applySilentAfter, strings.Join(quiet, ", "))
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
return true, nil
|
||||
}
|
||||
if len(quiet) > 0 {
|
||||
fmt.Printf("%s: tier %d: %s said nothing on its send within %s and is left as it was; the plan goes on without it\n",
|
||||
p.ID, p.Tier, strings.Join(quiet, ", "), applySilentAfter)
|
||||
leftAsWas = fmt.Sprintf("%s said nothing on its send within %s and is left as it was", strings.Join(quiet, ", "),
|
||||
applySilentAfter)
|
||||
}
|
||||
if len(waiting) > 0 {
|
||||
note := "tier " + fmt.Sprint(p.Tier) + " built; waiting for " + strings.Join(waiting, "; ") + " to be applied"
|
||||
@@ -1204,6 +1289,9 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
carryUserList(ctx, open, p)
|
||||
}
|
||||
}
|
||||
if end == "" {
|
||||
end = leftAsWas
|
||||
}
|
||||
p.Tier++
|
||||
p.State = inventory.PlanBuilding
|
||||
p.Note = end
|
||||
|
||||
@@ -0,0 +1,212 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"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"
|
||||
)
|
||||
|
||||
// 1160/AC5, live, with the store and a real bus and no ticker: a plan of two tiers whose first tier has a machine that
|
||||
// never reports, and whose second tier waits on the first (the plan's own order), passes its first tier after the bound,
|
||||
// the silent key's expiry the only trigger, and sends its second tier to the machine that reported alone (issue 1160).
|
||||
func TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers(t *testing.T) {
|
||||
open := aMesh(t)
|
||||
ctx := t.Context()
|
||||
inv := open.inventory
|
||||
|
||||
// A quarter of an hour of silence is two seconds here.
|
||||
wasSilent := applySilentAfter
|
||||
applySilentAfter = 2 * time.Second
|
||||
t.Cleanup(func() { applySilentAfter = wasSilent })
|
||||
|
||||
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)
|
||||
}
|
||||
legacy, err := js.Conn().JetStream()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
checkEvents = link.OverNATS{Conn: js.Conn(), JS: legacy}
|
||||
t.Cleanup(func() { checkEvents = nil })
|
||||
sentHeard := make(chan *nats.Msg, 32)
|
||||
sub, err := js.Conn().ChanSubscribe(link.SeatEventSubject(link.MeshControllerSeat, link.KeySent), sentHeard)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
|
||||
wasSaved := inventory.PlanSaved
|
||||
inventory.PlanSaved = servingPlanSaved(ctx, checkEvents, open)
|
||||
t.Cleanup(func() { inventory.PlanSaved = wasSaved })
|
||||
|
||||
// The build machine and the machines, as far as the plan meets them: an ask is counted, a send is recorded and kept.
|
||||
var mu sync.Mutex
|
||||
var asks []string
|
||||
var sends [][]string
|
||||
wasAsk := askABuild
|
||||
askABuild = func(_ context.Context, source buildSource, _, _ string) (string, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
id := link.NewBuildID(time.Now().Add(time.Duration(len(asks)) * time.Millisecond))
|
||||
asks = append(asks, source.Repository+"#"+id)
|
||||
return id, nil
|
||||
}
|
||||
t.Cleanup(func() { askABuild = wasAsk })
|
||||
digests := 0
|
||||
wasSend := sendRollout
|
||||
sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
sends = append(sends, append([]string(nil), names...))
|
||||
current, err := open.inventory.CurrentBuilds(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
builds := map[string]string{}
|
||||
for _, m := range []string{"a", "b"} {
|
||||
builds[m] = current[m].Commit
|
||||
}
|
||||
for _, node := range names {
|
||||
digests++
|
||||
if err := open.inventory.RecordSent(ctx, nodeID(t, open, node), fmt.Sprintf("d-%s-%d", node, digests), builds); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
return names, nil
|
||||
}
|
||||
t.Cleanup(func() { sendRollout = wasSend })
|
||||
|
||||
register := func(m, commit string, asked time.Time) {
|
||||
t.Helper()
|
||||
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: m, Version: commit},
|
||||
inventory.Source{Repository: "novox/" + m, Seat: "git", Path: "modules/" + m, BuiltFrom: commit, Head: commit,
|
||||
Asked: asked}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
record := func(id, m, commit string, at time.Time) {
|
||||
t.Helper()
|
||||
b := inventory.Build{ID: id, Module: m, Commit: commit, Repository: "novox/" + m, Path: "modules/" + m, Asked: at, At: at}
|
||||
b.Manifest, _ = json.Marshal(catalogue.Manifest{Module: m, Version: commit})
|
||||
if err := inv.RecordBuild(ctx, b); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
old := time.Now().Add(-2 * time.Hour)
|
||||
for _, m := range []string{"a", "b"} {
|
||||
record("build-"+m+"-1", m, "c1", old)
|
||||
register(m, "c1", old)
|
||||
for _, n := range []string{"anchor", "laptop"} {
|
||||
if _, err := inv.Assign(ctx, n, m); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
for _, n := range []string{"anchor", "laptop"} {
|
||||
if err := inv.RecordSent(ctx, nodeID(t, open, n), "d-"+n+"-c1", map[string]string{"a": "c1", "b": "c1"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
record("build-a-2", "a", "c2", time.Now().Add(-time.Minute))
|
||||
register("a", "c2", time.Now().Add(-time.Minute))
|
||||
|
||||
// A machine's report on what it was sent, heard as the controller hears it.
|
||||
listener := nudgingListener{Enrolment: link.Enrolment{Inventory: inv}}
|
||||
seq := int64(0)
|
||||
reports := func(node string) {
|
||||
t.Helper()
|
||||
last, found, err := inv.SentTo(ctx, node)
|
||||
if err != nil || !found {
|
||||
t.Fatalf("%s was sent nothing: %v", node, err)
|
||||
}
|
||||
seq++
|
||||
body, _ := json.Marshal(map[string]any{"node": node, "applied": []string{"x.server"}, "declared": last.Digest,
|
||||
"report_sequence": seq})
|
||||
var report link.Report
|
||||
if err := json.Unmarshal(body, &report); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := listener.Heard(ctx, report); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
plan := func() inventory.Plan {
|
||||
t.Helper()
|
||||
p, err := inv.PlanByID(ctx, "plan-1160")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
built := time.Now().UTC()
|
||||
p := inventory.Plan{ID: "plan-1160", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2", Created: time.Now().UTC(),
|
||||
State: inventory.PlanBuilding, Tiers: [][]string{{"a"}, {"b"}},
|
||||
Delivery: &inventory.PlanDelivery{Order: []inventory.Edge{{From: "b", To: "a", Kind: edgeChangeOrder}}},
|
||||
Modules: map[string]*inventory.PlanModule{
|
||||
"a": {State: "built", BuiltAt: &built, Commit: "c2", Build: "build-a-2"},
|
||||
"b": {},
|
||||
}}
|
||||
first, stopFirst := context.WithCancel(ctx)
|
||||
defer stopFirst()
|
||||
startRolloutEvents(first, open, js)
|
||||
if err := inv.SavePlan(ctx, &p); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
eventually(t, "tier 0 sent to both machines", func() bool {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
return len(sends) == 1
|
||||
})
|
||||
reports("anchor")
|
||||
time.Sleep(time.Second)
|
||||
if got := plan().Tier; got != 0 {
|
||||
t.Fatalf("1160/AC5: the plan went on to tier %d before the bound, with laptop not heard from", got)
|
||||
}
|
||||
// The note of the move to tier 1 is read in the step that sees it: later steps of the tier say their own.
|
||||
var moved string
|
||||
eventually(t, "tier 1 passed once laptop's key expired", func() bool {
|
||||
if q := plan(); q.Tier == 1 {
|
||||
moved = q.Note
|
||||
return true
|
||||
}
|
||||
return false
|
||||
})
|
||||
if !strings.Contains(moved, "laptop") || strings.Contains(moved, "anchor") {
|
||||
t.Errorf("1160/AC5: the plan says %q; want laptop named silent and not anchor", moved)
|
||||
}
|
||||
// The ask for tier 1's build is the next step, and records its build id on the module.
|
||||
eventually(t, "tier 1 asked", func() bool { m := plan().Modules["b"]; return m != nil && m.Build != "" })
|
||||
b := plan().Modules["b"]
|
||||
askedAt, _ := link.BuildAskedAt(b.Build)
|
||||
record(b.Build, "b", "c2", askedAt)
|
||||
register("b", "c2", askedAt)
|
||||
planBuilt(ctx, open, "b", "c2", "", askedAt, b.Build)
|
||||
eventually(t, "tier 1 sent to anchor alone", func() bool {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
return len(sends) == 2 && strings.Join(sends[1], ",") == "anchor"
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"slices"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// A machine silent past the bound at a tier between the first and the last is named silent and left as it was, as the
|
||||
// last tier's is (novox/hq ADR 0282 §1, ADR 0360 §4-5, issue 1160): one rule, walkReports, read by the walk's end and by
|
||||
// the gates between tiers.
|
||||
|
||||
func silentWalk(now time.Time) (inventory.Plan, []inventory.Reported) {
|
||||
sent := now.Add(-time.Minute)
|
||||
long := now.Add(-inventory.ApplySilentAfter - time.Minute)
|
||||
at := now.Add(-10 * time.Second)
|
||||
p := inventory.Plan{Tier: 0, Tiers: [][]string{{"engine"}, {"app"}}, Modules: map[string]*inventory.PlanModule{
|
||||
"engine": {Rest: map[string]inventory.SentDeclaration{
|
||||
"anchor": {Digest: "d-anchor", At: sent},
|
||||
"g14": {Digest: "d-g14", At: long},
|
||||
"fresh": {Digest: "d-fresh", At: sent}}},
|
||||
"app": {}}}
|
||||
reports := []inventory.Reported{{Node: "anchor", Outcome: inventory.OutcomeApplied, At: &at, Declared: "d-anchor"}}
|
||||
return p, reports
|
||||
}
|
||||
|
||||
func needsOf(running ...string) []gateNeed {
|
||||
return []gateNeed{{module: "engine", running: running, since: func(string) time.Time { return time.Time{} }}}
|
||||
}
|
||||
|
||||
// 1160/AC1: at a middle tier the machine silent past the bound is named, not waited for, and the plan goes on with the
|
||||
// machines that reported; one still inside the bound is waited for.
|
||||
func TestAMiddleTierGateNamesTheSilentMachineAndGoesOnWithThoseThatReported(t *testing.T) {
|
||||
now := time.Now().UTC()
|
||||
p, reports := silentWalk(now)
|
||||
silent := silentNodes(p, reports, now, nil)
|
||||
waiting, quiet, lost := settleTierGate(needsOf("anchor", "g14"), reports, silent)
|
||||
if len(waiting) != 0 || !slices.Equal(quiet, []string{"g14"}) || lost {
|
||||
t.Fatalf("1160/AC1: waiting %v, silent %v, lost %v; want g14 named silent and nothing waited for", waiting, quiet, lost)
|
||||
}
|
||||
waiting, quiet, _ = settleTierGate(needsOf("anchor", "g14", "fresh"), reports, silent)
|
||||
if len(waiting) != 1 || waiting[0] != "engine on fresh" || !slices.Equal(quiet, []string{"g14"}) {
|
||||
t.Fatalf("1160/AC1: a machine inside the bound is still waited for: waiting %v, silent %v", waiting, quiet)
|
||||
}
|
||||
}
|
||||
|
||||
// 1160/AC2: the walk's end and the gate read the same silent machines.
|
||||
func TestTheGateAndTheWalksEndShareOneRuleOfSilence(t *testing.T) {
|
||||
now := time.Now().UTC()
|
||||
p, reports := silentWalk(now)
|
||||
_, _, ended := walkReports(p, reports, now, nil)
|
||||
got := silentNodes(p, reports, now, nil)
|
||||
if len(got) != len(ended) || !got["g14"] || !slices.Equal(ended, []string{"g14"}) {
|
||||
t.Fatalf("1160/AC2: the gate reads %v silent, the walk's end %v", got, ended)
|
||||
}
|
||||
}
|
||||
|
||||
// 1160/AC3: a later tier is not sent to a machine silent.
|
||||
func TestALaterTierIsNotSentToASilentMachine(t *testing.T) {
|
||||
got := withoutSilent([]string{"anchor", "g14"}, map[string]bool{"g14": true})
|
||||
if !slices.Equal(got, []string{"anchor"}) {
|
||||
t.Fatalf("1160/AC3: a later tier goes to %v; want anchor alone", got)
|
||||
}
|
||||
in := []string{"anchor", "g14"}
|
||||
withoutSilent(in, map[string]bool{"g14": true})
|
||||
if !slices.Equal(in, []string{"anchor", "g14"}) {
|
||||
t.Fatalf("1160/AC3: the machines running it were changed: %v", in)
|
||||
}
|
||||
}
|
||||
|
||||
// 1160/AC4: every machine the gate waits on silent: the plan is lost, not gone on as if delivered.
|
||||
func TestAGateWhoseEveryMachineIsSilentIsLost(t *testing.T) {
|
||||
now := time.Now().UTC()
|
||||
p, reports := silentWalk(now)
|
||||
reports = nil
|
||||
p.Modules["engine"].Rest = map[string]inventory.SentDeclaration{
|
||||
"g14": {Digest: "d", At: now.Add(-time.Hour)}, "laptop": {Digest: "d", At: now.Add(-time.Hour)}}
|
||||
silent := silentNodes(p, reports, now, nil)
|
||||
_, quiet, lost := settleTierGate(needsOf("g14", "laptop"), reports, silent)
|
||||
if !lost || !slices.Equal(quiet, []string{"g14", "laptop"}) {
|
||||
t.Fatalf("1160/AC4: lost %v, silent %v; want both named and the gate lost", lost, quiet)
|
||||
}
|
||||
}
|
||||
@@ -406,7 +406,12 @@ func askNewModules(ctx context.Context, inv *inventory.Inventory, m link.SourceM
|
||||
if len(added) == 0 || len(from) == 0 {
|
||||
return nil
|
||||
}
|
||||
source := buildSource{Repository: from[0].Source.Repository, Seat: from[0].Source.Seat}
|
||||
return askBuilds(ctx, inv, m, buildSource{Repository: from[0].Source.Repository, Seat: from[0].Source.Seat}, added)
|
||||
}
|
||||
|
||||
// askBuilds asks the first build of each directory of a merge's repository, as askNewModules does, from the source given.
|
||||
func askBuilds(ctx context.Context, inv *inventory.Inventory, m link.SourceMoved, source buildSource,
|
||||
added []string) []string {
|
||||
var asked []string
|
||||
for _, dir := range added {
|
||||
path := dir
|
||||
|
||||
@@ -462,7 +462,9 @@ func EnsureLeaseBucket(ctx context.Context, js jetstream.JetStream) error {
|
||||
}
|
||||
|
||||
// HaveBucketConfig is the options of the have bucket (novox/hq ADR 0429 §1): delivery.HaveHistory revisions of each
|
||||
// key, no age, a value under a kilobyte (a have is JSON of about 500 bytes), 64 MiB on disk like the assignments.
|
||||
// key, no age, 64 MiB on disk like the assignments. A module's have is JSON of about 500 bytes, but a node's own key
|
||||
// carries its whole facts (its profile among them), several kilobytes on a desktop: a value is as large as the chain's
|
||||
// buckets take, since a value is never cut (novox/hq ADR 0408 §6, issue 1162).
|
||||
func HaveBucketConfig() jetstream.KeyValueConfig {
|
||||
return jetstream.KeyValueConfig{
|
||||
Bucket: delivery.HaveBucket,
|
||||
@@ -470,7 +472,7 @@ func HaveBucketConfig() jetstream.KeyValueConfig {
|
||||
"own facts) (novox/hq ADR 0429, design 47): written by each machine's node-engine under its own node's keys " +
|
||||
"alone, never by the controller; read by the delivery view and the watchman",
|
||||
History: delivery.HaveHistory,
|
||||
MaxValueSize: 4 << 10,
|
||||
MaxValueSize: ChainValueBytes,
|
||||
MaxBytes: 64 << 20,
|
||||
Storage: jetstream.FileStorage,
|
||||
}
|
||||
|
||||
@@ -104,3 +104,12 @@ func TestAModuleReadingHaveAssignmentsAndBuildsReadsAndWritesNone(t *testing.T)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A node's own key carries its whole facts (its profile among them), several kilobytes on a desktop: the have
|
||||
// bucket takes any value the chain's buckets take, since a value is never cut (novox/hq ADR 0408 §6, issue 1162).
|
||||
// 1162/AC1
|
||||
func TestTheHaveBucketTakesAValueAsLargeAsTheChains(t *testing.T) {
|
||||
if got := HaveBucketConfig().MaxValueSize; got != ChainValueBytes {
|
||||
t.Fatalf("the have bucket takes values up to %d bytes, the chain's up to %d", got, ChainValueBytes)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -84,6 +84,16 @@ func (i *Inventory) AloneMerges(ctx context.Context) ([]BatchedMerge, error) {
|
||||
return i.chainAloneMerges(ctx, c)
|
||||
}
|
||||
|
||||
// WaitingMerges is every merge no plan answers that is not walked alone: kept, and waiting for something else to
|
||||
// happen to it (a first merge into a repository, waiting for mesh-delivery's word on it, issue 1146).
|
||||
func (i *Inventory) WaitingMerges(ctx context.Context) ([]BatchedMerge, error) {
|
||||
c, err := i.theChain(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return i.chainWaitingMerges(ctx, c)
|
||||
}
|
||||
|
||||
// AnswerMerge moves a merge to the plan that answers it now; an empty plan with alone leaves it waiting to
|
||||
// be walked on its own commit.
|
||||
func (i *Inventory) AnswerMerge(ctx context.Context, repository, commit, plan string, alone bool) error {
|
||||
|
||||
@@ -452,6 +452,22 @@ func (i *Inventory) chainAloneMerges(ctx context.Context, c *chain.Chain) ([]Bat
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// chainWaitingMerges is every merge no plan answers that waits and is not walked alone, the newest first.
|
||||
func (i *Inventory) chainWaitingMerges(ctx context.Context, c *chain.Chain) ([]BatchedMerge, error) {
|
||||
w, err := i.chainWinners(ctx, c)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []BatchedMerge
|
||||
for _, h := range w {
|
||||
if h.m.Waiting && !h.m.Alone {
|
||||
out = append(out, batchedOf(h))
|
||||
}
|
||||
}
|
||||
sortMerges(out, true)
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// pendingMerge is a merge added to a plan that is not saved yet: carried until that plan's save writes it.
|
||||
type pendingMerge struct {
|
||||
plan string
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A merge no plan answers waits, and is listed as waiting until it is answered where it is kept (novox/hq issue 1146):
|
||||
// what a first merge into a new repository does while it waits for mesh-delivery's word.
|
||||
func TestAMergeNoPlanAnswersIsListedWaitingUntilItIsAnswered(t *testing.T) {
|
||||
inv, c := anInventoryOnTheChain(t)
|
||||
ctx := context.Background()
|
||||
now := time.Now().UTC()
|
||||
if added, err := inv.chainAddMerge(ctx, c, BatchedMerge{Repository: "novox/mesh-delivery", Branch: "main", Commit: "c1",
|
||||
Merged: now, Heard: now}); err != nil || !added {
|
||||
t.Fatalf("added %v, %v", added, err)
|
||||
}
|
||||
waiting, err := inv.chainWaitingMerges(ctx, c)
|
||||
if err != nil || len(waiting) != 1 || waiting[0].Commit != "c1" {
|
||||
t.Fatalf("waiting %+v, %v", waiting, err)
|
||||
}
|
||||
move := []MergeMove{{Repository: "novox/mesh-delivery", Commit: "c1", Plan: WaitingRecord}}
|
||||
if err := inv.chainSavePlans(ctx, c, nil, move); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if waiting, err = inv.chainWaitingMerges(ctx, c); err != nil || len(waiting) != 0 {
|
||||
t.Fatalf("answered, it still waits: %+v, %v", waiting, err)
|
||||
}
|
||||
if _, known, err := inv.chainMergeOf(ctx, c, "novox/mesh-delivery", "c1"); err != nil || !known {
|
||||
t.Fatalf("an answered merge is no longer kept: %v %v", known, err)
|
||||
}
|
||||
}
|
||||
Vendored
+2
@@ -163,3 +163,5 @@
|
||||
./internal/inventory TestASendEmptiesTheColumnSoALaterCopyLeavesAClearedKeyBare
|
||||
./internal/inventory TestWhetherAMachineReadsModulesIsItsLatestWord
|
||||
./cmd/mesh-controller TestAComposedDeclarationCarriesModulesAndBuildsOnlyForANodeThatReadsThem
|
||||
./cmd/mesh-controller TestAPlanPassesATierWhoseMachineNeverReportsAfterTheBoundAndSendsTheNextToTheOthers
|
||||
./cmd/mesh-controller TestAMergeOnTheApproveIntoANewRepositoryAsksTheFirstBuildAndItsOutcomeRegistersTheModule
|
||||
|
||||
Reference in New Issue
Block a user