Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cebc593627 | ||
|
|
df9e5728a9 | ||
|
|
2d18f438c5 | ||
|
|
3ce5a0239b | ||
|
|
70885ab3c4 | ||
|
|
65508d9df9 | ||
|
|
8123ca5654 | ||
|
|
d9b86376d3 | ||
|
|
ca3f43c400 | ||
|
|
b482122d66 | ||
|
|
2f5cc146fe | ||
|
|
c2ea9b4c51 | ||
|
|
b03d257f81 |
@@ -1,6 +0,0 @@
|
||||
# Release 57: config and provisioning
|
||||
|
||||
This branch carries the work of hq design 57 (ADR 0348, ADR 0349) in this repository. Each package
|
||||
merges here through its own reviewed pull request; the branch reaches main as one pull request, in one
|
||||
delivery group with the branch of the same name in mesh-controller, mesh-host, mesh-sdk, mesh-catalog
|
||||
and hq, after a rehearsal in the lab. This file is deleted by the last commit before that pull request.
|
||||
@@ -332,6 +332,12 @@ func answer(ctx context.Context, publisher builder.Publisher, on, workspace stri
|
||||
result.Check.Members = append(result.Check.Members, link.CheckedMember{Owner: m.Owner, Repo: m.Repo,
|
||||
Commit: m.Head, Trunk: m.Trunk})
|
||||
}
|
||||
// A branch tried before its pull request reads its outcome from this log, whole (novox/hq issue 686).
|
||||
if request.Check.Try {
|
||||
if line, err := json.Marshal(triedOf(v)); err == nil {
|
||||
say(link.StepTried, string(line))
|
||||
}
|
||||
}
|
||||
}
|
||||
} else if err == nil {
|
||||
// The package-registry credential is a build input, so it is resolved before the clone: a
|
||||
@@ -457,7 +463,8 @@ func checkSpecOf(request link.BuildRequest) builder.CheckSpec {
|
||||
spec := builder.CheckSpec{ID: request.ID, Repository: request.Repository, Ref: request.Ref,
|
||||
Owner: c.Owner, Repo: c.Repo, Number: c.Number, Paths: c.Paths, Beside: map[string]builder.Beside{},
|
||||
Modules: c.Modules, New: c.New, Manifests: c.Manifests, Judge: c.Judge, Base: c.Base,
|
||||
Toolchain: builder.ToolchainOf(request.Held), Toolchains: builder.ToolchainsOf(request.Held)}
|
||||
Toolchain: builder.ToolchainOf(request.Held), Toolchains: builder.ToolchainsOf(request.Held),
|
||||
Tests: c.Tests, StoreOnly: c.StoreOnly}
|
||||
for dir, b := range c.Beside {
|
||||
spec.Beside[dir] = builder.Beside{Repository: b.Repository, Ref: b.Ref}
|
||||
}
|
||||
@@ -472,6 +479,23 @@ func checkSpecOf(request link.BuildRequest) builder.CheckSpec {
|
||||
return spec
|
||||
}
|
||||
|
||||
// triedOf is a try's outcome as its log line says it (novox/hq issue 686): the repository's own check's verdict,
|
||||
// or the check's where that did not run — a head that does not merge — and the lines naming what failed.
|
||||
func triedOf(v builder.CheckVerdict) link.Tried {
|
||||
t := link.Tried{Verdict: v.Verdict, Summary: v.Summary, Took: v.Took.Round(time.Second).String(),
|
||||
Head: v.Head, Trunk: v.Trunk}
|
||||
if v.Repo != nil && (v.Gate == nil || v.Gate.Verdict == "pass") {
|
||||
t.Verdict, t.Summary = v.Repo.Verdict, v.Repo.Summary
|
||||
}
|
||||
if v.Repo != nil && v.Repo.Failed != "" {
|
||||
t.Failed = strings.Split(v.Repo.Failed, "\n")
|
||||
}
|
||||
if t.Verdict == "" {
|
||||
t.Verdict = "error"
|
||||
}
|
||||
return t
|
||||
}
|
||||
|
||||
// checkBuiltOf is a check's builds for its merge as the outcome carries them (novox/hq ADR 0350 §1).
|
||||
func checkBuiltOf(built []builder.CheckBuilt) []link.CheckBuilt {
|
||||
var out []link.CheckBuilt
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/builder"
|
||||
)
|
||||
|
||||
// A try's outcome, as the build seat says it in the build's log (novox/hq issue 686): the repository's own
|
||||
// check's verdict, how long it took, and each line naming what failed.
|
||||
func TestATrysOutcomeNamesWhatFailed(t *testing.T) {
|
||||
tried := triedOf(builder.CheckVerdict{Verdict: "pass", Summary: "the change touches no module", Took: 303 * time.Second,
|
||||
Head: "c0ffee", Trunk: "ca3f43c", Gate: &builder.Layer{Verdict: "pass"},
|
||||
Repo: &builder.Layer{Verdict: "fail", Summary: "its merge-check.sh failed: TestX",
|
||||
Failed: "--- FAIL: TestX (0.01s)\n x_test.go:12: got 1, want 2"}})
|
||||
if tried.Verdict != "fail" || tried.Took != "5m3s" || len(tried.Failed) != 2 || tried.Trunk != "ca3f43c" ||
|
||||
tried.Summary != "its merge-check.sh failed: TestX" {
|
||||
t.Fatalf("a failed try: %+v", tried)
|
||||
}
|
||||
// A head that does not merge fails both layers: the check's own word.
|
||||
conflict := triedOf(builder.CheckVerdict{Verdict: "fail", Summary: "does not merge", Gate: &builder.Layer{Verdict: "fail"},
|
||||
Repo: &builder.Layer{Verdict: "fail", Summary: "does not merge"}})
|
||||
if conflict.Verdict != "fail" || conflict.Summary != "does not merge" {
|
||||
t.Fatalf("a try that does not merge: %+v", conflict)
|
||||
}
|
||||
// No verdict is an error, never a pass.
|
||||
if none := triedOf(builder.CheckVerdict{}); none.Verdict != "error" {
|
||||
t.Fatalf("a try with no verdict: %+v", none)
|
||||
}
|
||||
}
|
||||
@@ -1443,6 +1443,12 @@ func probeGates(ctx context.Context, d *doctor) ([]conditions.Observation, error
|
||||
}
|
||||
// A release held after a failed one, while builds still wait for a gate (ADR 0236).
|
||||
out = append(out, backlogObservation()...)
|
||||
// And a rest that went silent (novox/hq ADR 0353 §4).
|
||||
silent, err := silentObservation(ctx, d.open.inventory)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out = append(out, silent...)
|
||||
return sortedFound(dedupeObservations(out)), nil
|
||||
}
|
||||
|
||||
@@ -1532,6 +1538,12 @@ func gateLine(g *inventory.PlanGate) string {
|
||||
case inventory.GateFailed:
|
||||
what = "FAILED"
|
||||
}
|
||||
switch {
|
||||
case g.Verdict == "" && g.Pending:
|
||||
what = "kept, not yet sent" // novox/hq ADR 0353 §8
|
||||
case g.Verdict == "" && g.Silent:
|
||||
what = "silent on its send, left" // ADR 0353 §4
|
||||
}
|
||||
if g.Verdict == "" && g.Waiting() {
|
||||
what = "waits for a busy container on " + g.Busy.Machine
|
||||
}
|
||||
|
||||
@@ -151,7 +151,7 @@ func heldMachines(ctx context.Context, open *stores, names []string) (map[string
|
||||
for _, mv := range f.moves(node, modules, same, known, false) {
|
||||
if planStillToSend(plans, mv.Module, node) == "" {
|
||||
why = append(why, fmt.Sprintf("%s would move from %s to %s, which has passed no gate yet — a release "+
|
||||
"plan sends it, one machine at a time, judged", mv.Module, buildName(mv.From), buildName(mv.To)))
|
||||
"plan sends it, one machine first and then the rest at once, judged", mv.Module, buildName(mv.From), buildName(mv.To)))
|
||||
}
|
||||
}
|
||||
if len(why) > 0 {
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// A tier whose module is no move — made from the source every machine running it runs, so nothing is
|
||||
// sent — is not waited on for a report after its marking (novox/hq issue 648). Its machines answered the
|
||||
// send that last carried that build before the walk got there, and nothing will make them report again:
|
||||
// the walk asks the next tier in the same step, rather than "waiting for app on anchor, laptop" for ever.
|
||||
func TestANoMoveTierWaitsForNoNewReport(t *testing.T) {
|
||||
judgeLeavesAsTheCore(t)
|
||||
g := aGateMesh(t)
|
||||
ctx := t.Context()
|
||||
inv := g.open.inventory
|
||||
|
||||
// app is the build machine, so a module of the next tier is built by it: the tier gates the next on app
|
||||
// running on its machines.
|
||||
for _, m := range []catalogue.Manifest{
|
||||
{Module: "app", Version: "c2", Claims: []catalogue.Claim{{Name: "node-build-agent", Scope: catalogue.ScopeNode}}},
|
||||
{Module: "web", Version: "w1"},
|
||||
} {
|
||||
if err := inv.RegisterModule(ctx, m, inventory.Source{Repository: "novox/mesh-catalog", Seat: "git",
|
||||
Path: "modules/" + m.Module, BuiltFrom: m.Version, Head: m.Version, Asked: time.Now().Add(-time.Minute)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
// What anchor and laptop were last sent (c1) was made from the source the plan's build of c2 is.
|
||||
for _, b := range []inventory.Build{
|
||||
{ID: "build-1s", Module: "app", Commit: "c1", SourceFingerprint: "src1:app",
|
||||
Asked: time.Now().Add(-30 * time.Second), At: time.Now().Add(-30 * time.Second)},
|
||||
{ID: "build-2s", Module: "app", Commit: "c2", SourceFingerprint: "src1:app", Asked: time.Now(), At: time.Now()},
|
||||
} {
|
||||
b.Manifest, _ = json.Marshal(catalogue.Manifest{Module: "app", Version: "1"})
|
||||
if err := inv.RecordBuild(ctx, b); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
// Each machine answered that send, applied, before the walk's tier was built.
|
||||
for _, n := range []string{"anchor", "laptop"} {
|
||||
if _, err := inv.RecordDoing(ctx, nodeID(t, g.open, n), inventory.Doing{Node: n,
|
||||
Outcome: inventory.OutcomeApplied, Declared: "d-" + n + "-c1", Applied: 1,
|
||||
At: time.Now().Add(-10 * time.Second)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
built := time.Now().UTC()
|
||||
p := g.plan(t)
|
||||
p.Tiers = [][]string{{"app"}, {"web"}}
|
||||
p.Modules["app"].Build, p.Modules["app"].BuiltAt = "build-2s", &built
|
||||
if err := inv.SavePlan(ctx, &p); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
advancePlans(ctx, g.open)
|
||||
|
||||
if len(g.sent) != 0 {
|
||||
t.Fatalf("a no-move build was sent: %v", g.sent)
|
||||
}
|
||||
p = g.plan(t)
|
||||
if s := p.Modules["app"]; s == nil || !strings.HasPrefix(s.Why, "no move") {
|
||||
t.Fatalf("the build was not marked no move: %+v", s)
|
||||
}
|
||||
if p.Tier < 1 || strings.Contains(p.Note, "waiting for app") {
|
||||
t.Fatalf("the walk waits on a no-move tier for a report nothing will cause: tier %d, %s %q", p.Tier, p.State, p.Note)
|
||||
}
|
||||
|
||||
// A machine that has not answered its last send is still waited on: no move is not "never sent".
|
||||
p.Tier, p.State, p.Note = 0, inventory.PlanRolling, ""
|
||||
if err := inv.RecordSent(ctx, nodeID(t, g.open, "laptop"), "d-laptop-later", map[string]string{"app": "c1"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.SavePlan(ctx, &p); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
advancePlans(ctx, g.open)
|
||||
p = g.plan(t)
|
||||
if p.Tier != 0 || !strings.Contains(p.Note, "waiting for app on laptop") {
|
||||
t.Fatalf("a machine that has not answered its last send was not waited on: tier %d, %q", p.Tier, p.Note)
|
||||
}
|
||||
}
|
||||
@@ -106,6 +106,8 @@ func run() error {
|
||||
return killCommand(ctx, args[1:])
|
||||
case "forget-checks":
|
||||
return forgetChecksCommand(ctx, args[1:])
|
||||
case "try":
|
||||
return tryCommand(ctx, args[1:])
|
||||
case "pause", "resume":
|
||||
return pauseCommand(ctx, args[0], args[1:])
|
||||
case "collection":
|
||||
@@ -381,6 +383,13 @@ func (b builds) Built(ctx context.Context, result link.BuildResult) error {
|
||||
// **A merge check builds nothing** (novox/hq to-be 45 §9): its verdict is said, and nothing of it is
|
||||
// recorded or registered.
|
||||
if result.Check != nil || result.Checked != nil {
|
||||
if isTry(result) {
|
||||
// A branch tried before its pull request (novox/hq issue 686): its asker reads the outcome from the
|
||||
// build's log. Never said as `checked`, so the forge's holder sets no status.
|
||||
fmt.Printf("%s: a try of %s at %.8s, read with `try --id`: said to nobody, nothing taken in\n", result.ID,
|
||||
result.Repository, result.Ref)
|
||||
return nil
|
||||
}
|
||||
checked(ctx, result)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -456,6 +456,9 @@ var plainWordings = map[string]func(conditions.Observation) words{
|
||||
"release-held": worded(func(o conditions.Observation) words {
|
||||
return releaseHeldWords(nil, o.Also)
|
||||
}),
|
||||
"release-silent": worded(func(o conditions.Observation) words {
|
||||
return releaseSilentWords(o.Also)
|
||||
}),
|
||||
"facts-stale": worded(func(o conditions.Observation) words {
|
||||
return words{Headline: "Merge checks use outdated facts",
|
||||
Explanation: "The facts a merge check judges a change against are out of date; they are gathered again " +
|
||||
@@ -841,6 +844,19 @@ func releaseHeldWords(modules, machines []string) words {
|
||||
Resolved: "Resolved: the held updates are released"}
|
||||
}
|
||||
|
||||
// releaseSilentWords are the plain words of machines that said nothing on an update they were sent (ADR 0353 §4).
|
||||
func releaseSilentWords(machines []string) words {
|
||||
where := "Some machines"
|
||||
if len(machines) > 0 {
|
||||
where = conditions.Capital(namesWords(machines, 4))
|
||||
}
|
||||
return words{Headline: "Machines did not answer an update",
|
||||
Needs: "check that they are awake and on the mesh; the next walk sends them what waits.",
|
||||
Explanation: where + " were sent module updates and said nothing about them within fifteen minutes, so " +
|
||||
"they were left as they were. What they run is not known to be the update.",
|
||||
Resolved: "Resolved: the next walk reached them"}
|
||||
}
|
||||
|
||||
// reloginNeeds is what an account waiting for its groups needs (ADR 0252).
|
||||
func reloginNeeds(node string) string {
|
||||
return fmt.Sprintf("log out of %s completely and log in again, or restart it.", node)
|
||||
|
||||
@@ -264,9 +264,6 @@ func serve(ctx context.Context) (err error) {
|
||||
// The facts snapshot a merge check is fed (novox/hq to-be 45 §9): kept current while this
|
||||
// controller holds the lease, read by the build seat from the artifact store.
|
||||
go exportingFacts(ctx, open, bus.Conn.ConnectedServerVersion)
|
||||
// The record mirrors (novox/hq ADR 0289, design 57 F3): the assignments and each registered module's
|
||||
// declared config on the bus, written from the store only when a record changed, by the lease holder.
|
||||
go mirroringRecords(ctx, inv, bus.Conn, bus)
|
||||
// Every call carries the lease's epoch, and its record is written only under the lease (novox/hq
|
||||
// to-be 45 §6).
|
||||
link.Calls.UnderLease(func() (uint64, error) { return theLease.epoch(ctx) })
|
||||
|
||||
@@ -1,105 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
"github.com/novox/mesh-controller/internal/recordmirror"
|
||||
)
|
||||
|
||||
// The controller's record mirrors on the bus (novox/hq ADR 0289, design 54, design 57 F3): the assignments,
|
||||
// and each registered module's declared config, which mesh-config resolves a module's config values from.
|
||||
//
|
||||
// **Built from the store, on a tick of their own, by the lease holder alone.** A record changes through a
|
||||
// verb that runs as a command of its own, a reconcile, a merge's registration or a pending assignment made,
|
||||
// so no one call site sees every change; what every change has in common is that it is committed to the
|
||||
// store. So the serving controller reads the store every mirrorEvery and writes only what differs from what
|
||||
// the bucket holds (ADR 0289 §1): a quiet mesh writes nothing, and a change rolled back is never read. A
|
||||
// mirror is therefore at most mirrorEvery behind its store.
|
||||
|
||||
// mirrorEvery is how often the mirrors are compared with the store: two small reads of the store.
|
||||
const mirrorEvery = 10 * time.Second
|
||||
|
||||
// mirroringRecords keeps the mirrors current for as long as this controller serves: ctx ends when it stops.
|
||||
func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.Conn, announce recordmirror.Announcer) {
|
||||
api, err := jetstream.New(conn)
|
||||
if err != nil {
|
||||
fmt.Printf("the record mirrors are not kept: %v\n", err)
|
||||
return
|
||||
}
|
||||
holds := func() bool {
|
||||
epoch, err := theLease.epoch(ctx)
|
||||
return err == nil && epoch > 0
|
||||
}
|
||||
epoch := func() uint64 {
|
||||
e, _ := theLease.epoch(ctx)
|
||||
return e
|
||||
}
|
||||
sources := map[string]func(context.Context) (map[string][]byte, error){
|
||||
recordmirror.Assignments.Name: inv.AssignmentsMirror,
|
||||
recordmirror.DeclaredConfig.Name: inv.DeclaredConfigMirror,
|
||||
}
|
||||
writers := map[string]*recordmirror.Writer{}
|
||||
failing := map[string]string{}
|
||||
tick := time.NewTicker(mirrorEvery)
|
||||
defer tick.Stop()
|
||||
for {
|
||||
if holds() {
|
||||
for _, m := range recordmirror.Mirrors() {
|
||||
why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
|
||||
if why != failing[m.Name] {
|
||||
if why != "" {
|
||||
fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
|
||||
} else {
|
||||
fmt.Printf("the %s mirror is kept again\n", m.Name)
|
||||
}
|
||||
failing[m.Name] = why
|
||||
}
|
||||
}
|
||||
}
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-tick.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// mirrorOnce brings one mirror to what the store holds, and answers why it could not; empty when it could.
|
||||
func mirrorOnce(ctx context.Context, api jetstream.JetStream, m recordmirror.Mirror, writers map[string]*recordmirror.Writer,
|
||||
source func(context.Context) (map[string][]byte, error), announce recordmirror.Announcer, epoch func() uint64,
|
||||
holds func() bool) string {
|
||||
reading, cancel := context.WithTimeout(ctx, 30*time.Second)
|
||||
defer cancel()
|
||||
w := writers[m.Name]
|
||||
if w == nil {
|
||||
kv, err := api.KeyValue(reading, m.Bucket)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("the bucket %s cannot be read: %v", m.Bucket, err)
|
||||
}
|
||||
w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
|
||||
writers[m.Name] = w
|
||||
}
|
||||
want, err := source(reading)
|
||||
if err != nil {
|
||||
return fmt.Sprintf("the store cannot be read: %v", err)
|
||||
}
|
||||
lost := w.LostEvents
|
||||
r, err := w.Sync(reading, want)
|
||||
if r.Written+r.Removed > 0 {
|
||||
fmt.Printf("the %s mirror moved: %d written, %d removed\n", m.Name, r.Written, r.Removed)
|
||||
}
|
||||
if w.LostEvents > lost {
|
||||
fmt.Printf("the %s mirror lost %d event(s): the bucket holds the change, its notice was refused twice\n",
|
||||
m.Name, w.LostEvents-lost)
|
||||
}
|
||||
if err != nil {
|
||||
return err.Error()
|
||||
}
|
||||
return ""
|
||||
}
|
||||
+537
-42
@@ -39,8 +39,8 @@ import (
|
||||
// from the same manifest is no move.
|
||||
//
|
||||
// **The release plan** is what walks the waiting moves through: whenever moves wait for a gate that no
|
||||
// started plan is walking, the mesh opens one — every such machine, one at a time, the control node last,
|
||||
// each sent and judged before the next. A release plan that fails stops the next one opening on its own:
|
||||
// started plan is walking, the mesh opens one — every such machine, the first sent and judged, then the rest
|
||||
// at once, each judged by its own gate (novox/hq ADR 0353, in place of one at a time). A release plan that fails stops the next one opening on its own:
|
||||
// a person releases it again (`upgrade release-backlog --why`), after the condition says why.
|
||||
|
||||
// sendScope is what a send may carry that no gate has seen.
|
||||
@@ -81,6 +81,15 @@ type moveFacts struct {
|
||||
srcs map[string]map[string]string
|
||||
passed map[string]map[string]bool
|
||||
plans []inventory.Plan
|
||||
// self is the release walking these moves, whose own gates are no other walk's (novox/hq ADR 0353).
|
||||
self string
|
||||
}
|
||||
|
||||
type walkingKey struct{}
|
||||
|
||||
// withWalking says a release plan is the one sending: its own gates never hold its own sends back.
|
||||
func withWalking(ctx context.Context, id string) context.Context {
|
||||
return context.WithValue(ctx, walkingKey{}, id)
|
||||
}
|
||||
|
||||
func readMoveFacts(ctx context.Context, inv *inventory.Inventory) (moveFacts, error) {
|
||||
@@ -99,6 +108,7 @@ func readMoveFacts(ctx context.Context, inv *inventory.Inventory) (moveFacts, er
|
||||
return f, err
|
||||
}
|
||||
f.plans, err = inv.OpenPlans(ctx)
|
||||
f.self, _ = ctx.Value(walkingKey{}).(string)
|
||||
return f, err
|
||||
}
|
||||
|
||||
@@ -193,17 +203,24 @@ func (f moveFacts) moves(node string, modules []string, sent map[string]string,
|
||||
// feet. A release keeps no module records: its open gate's carried moves are its walk.
|
||||
func (f moveFacts) walkedBy(module, node, to string) string {
|
||||
for _, p := range f.plans {
|
||||
if !p.Open() {
|
||||
if !p.Open() || (p.Release != nil && p.ID == f.self) {
|
||||
continue
|
||||
}
|
||||
if p.Release != nil {
|
||||
g := p.Release.Gate
|
||||
if g == nil || g.Verdict != "" {
|
||||
continue
|
||||
// The first machine's gate, and every gate of the rest still judging (novox/hq ADR 0353).
|
||||
gates := []*inventory.PlanGate{p.Release.Gate}
|
||||
for _, g := range p.Release.Gates {
|
||||
gates = append(gates, g)
|
||||
}
|
||||
for _, c := range g.Carried {
|
||||
if c.Module == module && !(slices.Contains(g.Machines, node) && sameCommit(c.To, to)) {
|
||||
return p.ID
|
||||
for _, g := range gates {
|
||||
// A send kept and not yet made walks nothing yet (ADR 0353 §8).
|
||||
if g == nil || g.Verdict != "" || g.Silent || g.Pending {
|
||||
continue
|
||||
}
|
||||
for _, c := range g.Carried {
|
||||
if c.Module == module && !(slices.Contains(g.Machines, node) && sameCommit(c.To, to)) {
|
||||
return p.ID
|
||||
}
|
||||
}
|
||||
}
|
||||
continue
|
||||
@@ -303,8 +320,8 @@ func ungatedIn(ctx context.Context, open *stores, names []string, addedHolder st
|
||||
}
|
||||
}
|
||||
if len(refused) > 0 {
|
||||
return nil, fmt.Errorf("%w: %s — a walk sends them, one machine at a time, each judged (novox/hq ADR "+
|
||||
"0236); `upgrade backlog` lists them", errUngated, strings.Join(refused, "; "))
|
||||
return nil, fmt.Errorf("%w: %s — a walk sends them, one machine first and then the rest at once, each judged (novox/hq ADR "+
|
||||
"0236, ADR 0353); `upgrade backlog` lists them", errUngated, strings.Join(refused, "; "))
|
||||
}
|
||||
return kept, nil
|
||||
}
|
||||
@@ -320,19 +337,35 @@ func ungatedIn(ctx context.Context, open *stores, names []string, addedHolder st
|
||||
// And the grants step first, when the send changes what the bus's user list must say (novox/hq issue 490):
|
||||
// answered for the gate to keep, nil when there was none.
|
||||
func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.CarriedMove) ([]inventory.CarriedMove, []string, *inventory.GrantsStep, error) {
|
||||
inv := open.inventory
|
||||
f, err := readMoveFacts(ctx, inv)
|
||||
moves, err := gatedMoves(ctx, open, node, owns)
|
||||
if err != nil || len(moves) == 0 {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
sayRecreations(ctx, open, moves)
|
||||
sent, grants, err := sendJudged(ctx, open, node)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
return moves, sent, grants, nil
|
||||
}
|
||||
|
||||
// gatedMoves is what a gated send to one machine would carry, with the build each moves to, made of nothing:
|
||||
// gatedSend's reading, which a walk also keeps before it sends (novox/hq ADR 0353 §8). Refused the same way
|
||||
// while a plan that has started walks one of those builds elsewhere.
|
||||
func gatedMoves(ctx context.Context, open *stores, node string, owns []inventory.CarriedMove) ([]inventory.CarriedMove, error) {
|
||||
inv := open.inventory
|
||||
f, err := readMoveFacts(ctx, inv)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
moves, err := machineMoves(ctx, open, f, node, true)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
// A move the send keeps is not made, and is no move for the gate to judge (novox/hq issue 201): a core module
|
||||
// kept over an older record judged as moving back to it would read the gate as superseded by its own send.
|
||||
if moves, err = withoutKept(ctx, open, node, moves); err != nil {
|
||||
return nil, nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
own := func(module string) bool {
|
||||
return slices.ContainsFunc(owns, func(o inventory.CarriedMove) bool { return o.Module == module })
|
||||
@@ -342,7 +375,7 @@ func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.
|
||||
continue
|
||||
}
|
||||
if id := f.walkedBy(mv.Module, node, mv.To); id != "" {
|
||||
return nil, nil, nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
|
||||
return nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
|
||||
mv.Module, short(mv.To), node, id)
|
||||
}
|
||||
}
|
||||
@@ -350,7 +383,7 @@ func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.
|
||||
// plan, or a release, on this machine is not crossed with a newer build; this send waits for its gate.
|
||||
for _, o := range owns {
|
||||
if id := f.walkedBy(o.Module, node, o.To); id != "" {
|
||||
return nil, nil, nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
|
||||
return nil, fmt.Errorf("%w: %s's build %s waits on %s, which %s is walking", errWalkedElsewhere,
|
||||
o.Module, short(o.To), node, id)
|
||||
}
|
||||
}
|
||||
@@ -363,30 +396,26 @@ func gatedSend(ctx context.Context, open *stores, node string, owns []inventory.
|
||||
moves[i].Build = o.Build
|
||||
}
|
||||
}
|
||||
if len(moves) == 0 && len(owns) == 0 {
|
||||
return nil, nil, nil, nil
|
||||
if len(moves) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
for i := range moves {
|
||||
if moves[i].Build == "" {
|
||||
if moves[i].Build, err = inv.BuildOf(ctx, moves[i].Module, moves[i].To); err != nil {
|
||||
return nil, nil, nil, err
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
}
|
||||
sort.Slice(moves, func(i, j int) bool { return moves[i].Module < moves[j].Module })
|
||||
sayRecreations(ctx, open, moves)
|
||||
sent, grants, err := sendJudged(ctx, open, node)
|
||||
if err != nil {
|
||||
return nil, nil, nil, err
|
||||
}
|
||||
return moves, sent, grants, nil
|
||||
return moves, nil
|
||||
}
|
||||
|
||||
// passCarried keeps a pass as the verdict of every build the gate judged beside its own module.
|
||||
func passCarried(ctx context.Context, open *stores, p *inventory.Plan, g *inventory.PlanGate, except string) {
|
||||
seen := map[string]bool{}
|
||||
for _, c := range g.Carried {
|
||||
if c.Module == except || c.Build == "" || seen[c.Build] {
|
||||
// A build put back with another machine's failure is not this gate's to pass (novox/hq ADR 0353 §3).
|
||||
if c.Module == except || c.Build == "" || seen[c.Build] || slices.Contains(g.Returned, c.Module) {
|
||||
continue
|
||||
}
|
||||
seen[c.Build] = true
|
||||
@@ -695,7 +724,7 @@ func releaseBacklog(ctx context.Context, open *stores, by string) (*inventory.Pl
|
||||
p := inventory.Plan{ID: fmt.Sprintf("release-%d", now.UnixNano()), Repository: releaseRepository,
|
||||
Created: now, State: inventory.PlanRolling, Tiers: [][]string{names}, Modules: map[string]*inventory.PlanModule{},
|
||||
Release: &inventory.PlanRelease{Order: order, By: by},
|
||||
Note: fmt.Sprintf("%d build(s) wait for a gate on %s; one machine at a time, each judged", len(names),
|
||||
Note: fmt.Sprintf("%d build(s) wait for a gate on %s; one machine first, then the rest at once, each judged", len(names),
|
||||
strings.Join(order, ", "))}
|
||||
if err := inv.SavePlan(ctx, &p); err != nil {
|
||||
return nil, err
|
||||
@@ -704,14 +733,23 @@ func releaseBacklog(ctx context.Context, open *stores, by string) (*inventory.Pl
|
||||
return &p, nil
|
||||
}
|
||||
|
||||
// advanceRelease takes one step of a release plan: the next machine sent everything waiting there under a
|
||||
// gate, or the machine being judged judged once more; the plan done when every machine is.
|
||||
// advanceRelease takes one step of a release plan: the first machine sent everything waiting there under a
|
||||
// gate and judged; once it passed, every other machine sent at once, each under its own gate (novox/hq ADR
|
||||
// 0353 §1), and judged; the plan done when every machine is.
|
||||
//
|
||||
// **A send is kept before it is made** (ADR 0353 §8): the step that chooses a machine keeps its gate pending with
|
||||
// what it will carry, and the next step makes the send. A controller replaced between the send and keeping it
|
||||
// finds the gate pending, reads the machine's last send, and neither sends again nor leaves the machine unjudged.
|
||||
func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool, error) {
|
||||
r := p.Release
|
||||
if r.Settings != nil {
|
||||
return advanceSettingsWalk(ctx, open, p)
|
||||
}
|
||||
ctx = withWalking(ctx, p.ID)
|
||||
now := time.Now().UTC()
|
||||
if r.Rest {
|
||||
return advanceRest(ctx, open, p, now)
|
||||
}
|
||||
if r.Gate == nil {
|
||||
heard, err := releaseHeard(ctx, open)
|
||||
if err != nil {
|
||||
@@ -724,7 +762,7 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
r.Next++
|
||||
continue
|
||||
}
|
||||
moves, sent, grants, err := gatedSend(ctx, open, node, nil)
|
||||
moves, err := gatedMoves(ctx, open, node, nil)
|
||||
if errors.Is(err, errWalkedElsewhere) {
|
||||
note := fmt.Sprintf("waiting before %s: %v", node, err)
|
||||
changed := p.Note != note
|
||||
@@ -732,23 +770,15 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
return changed, nil
|
||||
}
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("sending %s what waits there: %w", node, err)
|
||||
return false, fmt.Errorf("reading what waits on %s: %w", node, err)
|
||||
}
|
||||
if len(moves) == 0 {
|
||||
r.Done = append(r.Done, node)
|
||||
r.Next++
|
||||
continue
|
||||
}
|
||||
r.Gate = &inventory.PlanGate{Machines: sent, Since: &now, Carried: moves, Sent: sentNow(ctx, open.inventory, sent),
|
||||
Grants: grants}
|
||||
p.Note = fmt.Sprintf("sent %s %d build(s) that waited for a gate; judging them there", node, len(moves))
|
||||
if said := grantsSaid(grants); said != "" {
|
||||
p.Note += "; " + said
|
||||
}
|
||||
if said := recreationsSaid(moves); said != "" {
|
||||
p.Note += "; " + said
|
||||
}
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
r.Gate = &inventory.PlanGate{Machines: []string{node}, Carried: moves, Pending: true}
|
||||
p.Note = fmt.Sprintf("sending %s %d build(s) that wait for a gate, first", node, len(moves))
|
||||
return true, nil
|
||||
}
|
||||
p.State, p.Tier = inventory.PlanDone, len(p.Tiers)
|
||||
@@ -759,6 +789,19 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
return true, nil
|
||||
}
|
||||
g := r.Gate
|
||||
if g.Pending {
|
||||
node := firstOf(g.Machines)
|
||||
out, err := sendKept(ctx, open, p, node, g, now)
|
||||
if err != nil || out == keptWaiting {
|
||||
return out == keptWaiting, err
|
||||
}
|
||||
if out == keptNothing {
|
||||
r.Gate = nil
|
||||
r.Done = append(r.Done, node)
|
||||
r.Next++
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
verdict, err := judgeMoves(ctx, open, g, nil, now)
|
||||
if err != nil {
|
||||
return false, err
|
||||
@@ -777,6 +820,11 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
r.Done = append(r.Done, firstOf(g.Machines))
|
||||
r.Next++
|
||||
r.Gate = nil
|
||||
// **Then the rest at once** (novox/hq ADR 0353 §1): every other machine heard from, each judged by its own
|
||||
// gate, in place of one machine at a time (ADR 0236 §22).
|
||||
r.Rest = true
|
||||
p.Note = fmt.Sprintf("passed on %s; the rest are sent at once, each judged", firstOf(g.Machines))
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
return true, nil
|
||||
case inventory.GateSuperseded:
|
||||
// Another send moved a judged module on the judged machine (novox/hq issue 352): no verdict on what
|
||||
@@ -797,6 +845,453 @@ func advanceRelease(ctx context.Context, open *stores, p *inventory.Plan) (bool,
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// What sendKept did with a kept send.
|
||||
const (
|
||||
keptSent = iota // sent now
|
||||
keptAlready // the machine was already sent it: judged from that send, nothing sent
|
||||
keptNothing // nothing waits there any more: no gate
|
||||
keptWaiting // another walk judges a build it would carry: it waits
|
||||
)
|
||||
|
||||
// sendKept makes a send a gate kept before it was made (novox/hq ADR 0353 §8) — unless the machine's last send
|
||||
// already carries what the gate is for, made before a controller was replaced and before it was kept: then
|
||||
// nothing is sent, and the gate judges from that send (§6).
|
||||
func sendKept(ctx context.Context, open *stores, p *inventory.Plan, node string, g *inventory.PlanGate,
|
||||
now time.Time) (int, error) {
|
||||
inv := open.inventory
|
||||
moves, sent, grants, err := gatedSend(ctx, open, node, nil)
|
||||
if errors.Is(err, errWalkedElsewhere) {
|
||||
p.Note = fmt.Sprintf("waiting before %s: %v", node, err)
|
||||
return keptWaiting, nil
|
||||
}
|
||||
if err != nil {
|
||||
return keptSent, fmt.Errorf("sending %s what waits there: %w", node, err)
|
||||
}
|
||||
if len(moves) > 0 {
|
||||
g.Machines, g.Carried, g.Since, g.Grants, g.Pending = sent, moves, &now, grants, false
|
||||
g.Sent = sentNow(ctx, inv, sent)
|
||||
p.Note = fmt.Sprintf("sent %s %d build(s) that waited for a gate; judging them there", node, len(moves))
|
||||
if said := grantsSaid(grants); said != "" {
|
||||
p.Note += "; " + said
|
||||
}
|
||||
if said := recreationsSaid(moves); said != "" {
|
||||
p.Note += "; " + said
|
||||
}
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
return keptSent, nil
|
||||
}
|
||||
if !alreadySent(ctx, inv, node, g.Carried) {
|
||||
return keptNothing, nil
|
||||
}
|
||||
since := now
|
||||
if s, found, err := inv.SentTo(ctx, node); err == nil && found {
|
||||
if !s.At.IsZero() {
|
||||
since = s.At
|
||||
}
|
||||
g.Sent = map[string]inventory.SentDeclaration{node: s}
|
||||
}
|
||||
g.Since, g.Pending = &since, false
|
||||
p.Note = fmt.Sprintf("%s was already sent the %d build(s) kept for it, at %s: judged from that send, nothing sent again",
|
||||
node, len(g.Carried), since.Local().Format("15:04:05"))
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
return keptAlready, nil
|
||||
}
|
||||
|
||||
// alreadySent says a machine's last send carries every build of the moves: a send made and not kept.
|
||||
func alreadySent(ctx context.Context, inv *inventory.Inventory, node string, moves []inventory.CarriedMove) bool {
|
||||
if len(moves) == 0 {
|
||||
return false
|
||||
}
|
||||
sent, known, err := inv.SentBuilds(ctx, node)
|
||||
if err != nil || !known {
|
||||
return false
|
||||
}
|
||||
for _, mv := range moves {
|
||||
if !sameCommit(sent[mv.Module], mv.To) {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
// advanceRest takes one step of a release plan's rest (novox/hq ADR 0353): the sends kept for a step made, every
|
||||
// machine of the step judged by its own gate, one that fails put back by itself while the others go on to their
|
||||
// verdicts, and the next step — or the end — once every gate of the step has its verdict or is silent.
|
||||
func advanceRest(ctx context.Context, open *stores, p *inventory.Plan, now time.Time) (bool, error) {
|
||||
r := p.Release
|
||||
// The sends kept for this step, in the plan's order: the control node, last in it, last (§1).
|
||||
var pending []string
|
||||
for _, n := range r.Order {
|
||||
if g := r.Gates[n]; g != nil && g.Pending {
|
||||
pending = append(pending, n)
|
||||
}
|
||||
}
|
||||
if len(pending) > 0 {
|
||||
// **A kept send carries only what was kept** (ADR 0353 §2): a build that passed no gate and came to wait on
|
||||
// a machine since its step was chosen could reach a second machine in it; the step is chosen again.
|
||||
if grew, err := keptStepGrew(ctx, open, r, pending); err != nil {
|
||||
return false, err
|
||||
} else if grew != "" {
|
||||
for _, n := range pending {
|
||||
delete(r.Gates, n)
|
||||
}
|
||||
opened, waiting, err := openRestStep(ctx, open, p)
|
||||
if err != nil {
|
||||
return true, err
|
||||
}
|
||||
p.Note = "chosen again: " + grew
|
||||
if !opened && waiting != "" {
|
||||
p.Note += "; " + waiting
|
||||
}
|
||||
return true, nil
|
||||
}
|
||||
var sent, already, waiting []string
|
||||
for _, node := range pending {
|
||||
out, err := sendKept(ctx, open, p, node, r.Gates[node], now)
|
||||
if err != nil {
|
||||
return true, err
|
||||
}
|
||||
switch out {
|
||||
case keptSent:
|
||||
sent = append(sent, node)
|
||||
case keptAlready:
|
||||
already = append(already, node)
|
||||
case keptNothing:
|
||||
delete(r.Gates, node)
|
||||
r.Done = append(r.Done, node)
|
||||
case keptWaiting:
|
||||
waiting = append(waiting, node)
|
||||
}
|
||||
}
|
||||
var said []string
|
||||
if len(sent) > 0 {
|
||||
said = append(said, "sent "+strings.Join(sent, ", ")+" at once, each judged there")
|
||||
}
|
||||
if len(already) > 0 {
|
||||
said = append(said, strings.Join(already, ", ")+" already sent it: judged from that send")
|
||||
}
|
||||
if len(waiting) > 0 {
|
||||
said = append(said, "waiting before "+strings.Join(waiting, ", ")+": another walk judges a build it carries")
|
||||
}
|
||||
if len(said) > 0 {
|
||||
p.Note = strings.Join(said, "; ")
|
||||
}
|
||||
return len(sent)+len(already) > 0 || len(waiting) < len(pending), nil
|
||||
}
|
||||
reports, err := open.inventory.LastReports(ctx)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
byNode := map[string]inventory.Reported{}
|
||||
for _, rep := range reports {
|
||||
byNode[rep.Node] = rep
|
||||
}
|
||||
changed, noteWas := false, p.Note
|
||||
var notes, judging []string
|
||||
batched, back := batchingRollbacks(ctx)
|
||||
for _, node := range r.Order {
|
||||
g := r.Gates[node]
|
||||
if g == nil || g.Verdict != "" || g.Silent {
|
||||
continue
|
||||
}
|
||||
// What another machine's failure already put back is not this gate's to judge (ADR 0353 §3).
|
||||
if len(g.Carried) > 0 && len(g.Returned) >= len(g.Carried) {
|
||||
decide(g, inventory.GateFailed, "what it was sent was put back with another machine's failure", now)
|
||||
r.Failed = append(r.Failed, node)
|
||||
changed = true
|
||||
continue
|
||||
}
|
||||
// **A machine that says nothing on its send is left silent, not failed** (ADR 0353 §4): it is not judged
|
||||
// until it reports, and past ApplySilentAfter it is listed and left.
|
||||
if !reportedOnSend(byNode[node], node, g) {
|
||||
if now.Sub(*g.Since) > inventory.ApplySilentAfter {
|
||||
g.Silent = true
|
||||
r.Silent = append(r.Silent, node)
|
||||
changed = true
|
||||
notes = append(notes, fmt.Sprintf("%s said nothing on its send within %s: left silent", node,
|
||||
inventory.ApplySilentAfter))
|
||||
continue
|
||||
}
|
||||
judging = append(judging, node+": not yet reported on its send")
|
||||
continue
|
||||
}
|
||||
verdict, err := judgeMoves(ctx, open, g, nil, now)
|
||||
if err != nil {
|
||||
return changed, err
|
||||
}
|
||||
switch verdict {
|
||||
case "":
|
||||
if putBackBroken(ctx, open, p, g, "", nil) {
|
||||
changed = true
|
||||
notes = append(notes, p.Note)
|
||||
}
|
||||
judging = append(judging, node+": "+gateLine(g))
|
||||
case inventory.GatePassed:
|
||||
passCarried(ctx, open, p, g, "")
|
||||
r.Done = append(r.Done, node)
|
||||
changed = true
|
||||
case inventory.GateSuperseded:
|
||||
// Another send moved a judged module there (novox/hq issue 352): nothing judged, nothing put back on
|
||||
// it; what still waits is released again.
|
||||
r.Done = append(r.Done, node)
|
||||
changed = true
|
||||
notes = append(notes, fmt.Sprintf("superseded on %s: %s", node, g.Why))
|
||||
default:
|
||||
// **Put back by itself** (ADR 0353 §3): what was found wanting goes back at once, and every other
|
||||
// machine of the step goes on to its own verdict.
|
||||
stateWas := p.State
|
||||
p.Note = ""
|
||||
failCarried(batched, open, p, g, "")
|
||||
p.State = stateWas
|
||||
r.Failed = append(r.Failed, node)
|
||||
changed = true
|
||||
notes = append(notes, fmt.Sprintf("failed its gate on %s: %s — %s", node, g.Why, p.Note))
|
||||
returnedElsewhere(ctx, open, r, node, g)
|
||||
}
|
||||
}
|
||||
p.Note = ""
|
||||
sendRollbacks(ctx, open, p, back)
|
||||
if p.Note != "" {
|
||||
notes = append(notes, strings.TrimPrefix(p.Note, "; "))
|
||||
}
|
||||
if len(judging) > 0 {
|
||||
note := strings.Join(append(notes, "judging "+strings.Join(judging, "; ")), "; ")
|
||||
p.Note = note
|
||||
return changed || noteWas != note, nil
|
||||
}
|
||||
// Every gate of the step has its verdict or is silent: the next step, unless a machine failed (§3).
|
||||
if len(r.Failed) == 0 {
|
||||
opened, waiting, err := openRestStep(ctx, open, p)
|
||||
if err != nil {
|
||||
return changed, err
|
||||
}
|
||||
if opened {
|
||||
return true, nil
|
||||
}
|
||||
if waiting != "" {
|
||||
note := strings.Join(append(notes, waiting), "; ")
|
||||
p.Note = note
|
||||
return changed || noteWas != note, nil
|
||||
}
|
||||
}
|
||||
p.Tier = len(p.Tiers)
|
||||
end := fmt.Sprintf("released on %s", orNone(strings.Join(r.Done, ", ")))
|
||||
if len(r.Silent) > 0 {
|
||||
end += "; silent on their send, left: " + strings.Join(r.Silent, ", ")
|
||||
}
|
||||
if len(r.Skipped) > 0 {
|
||||
end += "; not heard from, left as they were: " + strings.Join(r.Skipped, ", ")
|
||||
}
|
||||
p.State = inventory.PlanDone
|
||||
if len(r.Failed) > 0 {
|
||||
p.State = inventory.PlanFailed
|
||||
end = "failed on " + strings.Join(r.Failed, ", ") + "; " + end
|
||||
} else if len(r.Silent) > 0 && len(r.Gates) == len(r.Silent) {
|
||||
// **A rest that said nothing is no success** (ADR 0353 §4): failure is loud; the condition says it.
|
||||
p.State = inventory.PlanFailed
|
||||
end = "no machine of the rest reported on its send; " + end
|
||||
}
|
||||
p.Note = strings.Join(append(notes, end), "; ")
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
return true, nil
|
||||
}
|
||||
|
||||
// keptStepGrew says why a step's kept sends would now carry a build no gate passed that they were not kept with;
|
||||
// empty when none would.
|
||||
func keptStepGrew(ctx context.Context, open *stores, r *inventory.PlanRelease, pending []string) (string, error) {
|
||||
f, err := readMoveFacts(ctx, open.inventory)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
shelf, err := open.inventory.Catalogue(ctx)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
for _, node := range pending {
|
||||
moves, err := gatedMoves(ctx, open, node, nil)
|
||||
if errors.Is(err, errWalkedElsewhere) {
|
||||
continue // sendKept says it waits
|
||||
}
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
kept := r.Gates[node].Carried
|
||||
for _, mv := range moves {
|
||||
// The node-engine and the resolver keep one machine at a time, passed or not (ADR 0353 §9).
|
||||
if (f.gated(mv.Module, mv.To) && !oneAtATime(mv.Module, shelf[mv.Module])) || slices.ContainsFunc(kept, func(c inventory.CarriedMove) bool {
|
||||
return c.Module == mv.Module && sameCommit(c.To, mv.To)
|
||||
}) {
|
||||
continue
|
||||
}
|
||||
return fmt.Sprintf("%s's build %s, which this step did not keep for it, came to wait on %s", mv.Module, short(mv.To), node), nil
|
||||
}
|
||||
}
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// reportedOnSend says a machine's last report is about the send its gate judges (novox/hq issue 352): the
|
||||
// declaration kept on the gate, or one sequenced after it; a gate that kept none reads the send made last.
|
||||
func reportedOnSend(r inventory.Reported, node string, g *inventory.PlanGate) bool {
|
||||
if r.At == nil {
|
||||
return false
|
||||
}
|
||||
if s, kept := g.Sent[node]; kept {
|
||||
return s.ReportsOn(r)
|
||||
}
|
||||
return r.Current
|
||||
}
|
||||
|
||||
// returnedElsewhere marks, on every other gate of the rest still judging, the builds a failed machine's
|
||||
// rollback put back wherever they were sent (ADR 0236 §5): no longer theirs to judge (ADR 0353 §3).
|
||||
func returnedElsewhere(ctx context.Context, open *stores, r *inventory.PlanRelease, failed string, g *inventory.PlanGate) {
|
||||
for _, c := range g.Carried {
|
||||
if c.Build == "" {
|
||||
continue
|
||||
}
|
||||
if marked, err := open.inventory.GateFailed(ctx, c.Build); err != nil || !marked {
|
||||
continue
|
||||
}
|
||||
for node, other := range r.Gates {
|
||||
if node == failed || other.Verdict != "" || slices.Contains(other.Returned, c.Module) {
|
||||
continue
|
||||
}
|
||||
for _, oc := range other.Carried {
|
||||
if oc.Module == c.Module && sameCommit(oc.To, c.To) {
|
||||
other.Returned = append(other.Returned, c.Module)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// openRestStep chooses the next step of a release plan's rest and keeps its sends (novox/hq ADR 0353 §1, §2):
|
||||
// every machine left that is heard from and has something waiting, except that a build no gate has passed goes
|
||||
// to one machine of the step only. Answers whether a step was kept, or why the walk waits.
|
||||
func openRestStep(ctx context.Context, open *stores, p *inventory.Plan) (bool, string, error) {
|
||||
r := p.Release
|
||||
heard, err := releaseHeard(ctx, open)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
f, err := readMoveFacts(ctx, open.inventory)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
waiting := map[string][]inventory.CarriedMove{}
|
||||
var nodes, walked []string
|
||||
for _, node := range r.Order[min(r.Next, len(r.Order)):] {
|
||||
if r.Gates[node] != nil || slices.Contains(r.Done, node) || slices.Contains(r.Skipped, node) ||
|
||||
slices.Contains(r.Failed, node) || slices.Contains(r.Silent, node) {
|
||||
continue
|
||||
}
|
||||
if !heard[node] {
|
||||
r.Skipped = append(r.Skipped, node)
|
||||
continue
|
||||
}
|
||||
moves, err := gatedMoves(ctx, open, node, nil)
|
||||
if errors.Is(err, errWalkedElsewhere) {
|
||||
walked = append(walked, fmt.Sprintf("%s (%v)", node, err))
|
||||
continue
|
||||
}
|
||||
if err != nil {
|
||||
return false, "", fmt.Errorf("reading what waits on %s: %w", node, err)
|
||||
}
|
||||
if len(moves) == 0 {
|
||||
r.Done = append(r.Done, node)
|
||||
continue
|
||||
}
|
||||
waiting[node] = moves
|
||||
nodes = append(nodes, node)
|
||||
}
|
||||
shelf, err := open.inventory.Catalogue(ctx)
|
||||
if err != nil {
|
||||
return false, "", err
|
||||
}
|
||||
step := restStep(nodes, waiting, f.gated, func(module string) bool { return oneAtATime(module, shelf[module]) })
|
||||
if len(step) == 0 {
|
||||
if len(walked) > 0 {
|
||||
return false, "waiting before " + strings.Join(walked, "; "), nil
|
||||
}
|
||||
return false, "", nil
|
||||
}
|
||||
if r.Gates == nil {
|
||||
r.Gates = map[string]*inventory.PlanGate{}
|
||||
}
|
||||
for _, node := range step {
|
||||
r.Gates[node] = &inventory.PlanGate{Machines: []string{node}, Carried: waiting[node], Pending: true}
|
||||
}
|
||||
p.Note = fmt.Sprintf("sending %s at once, each judged there", strings.Join(step, ", "))
|
||||
if left := len(nodes) - len(step); left > 0 {
|
||||
p.Note += fmt.Sprintf("; %d machine(s) wait for a build that passed no gate yet to pass on one", left)
|
||||
}
|
||||
return true, "", nil
|
||||
}
|
||||
|
||||
// oneAtATime is a module whose builds go to the rest one machine at a time (novox/hq ADR 0353 §9): the
|
||||
// node-engine, which applies every send, and a module holding the mesh's resolver, which names the bus.
|
||||
func oneAtATime(module string, m catalogue.Manifest) bool {
|
||||
if module == hostModule {
|
||||
return true
|
||||
}
|
||||
return slices.ContainsFunc(m.Claims, func(c catalogue.Claim) bool { return c.Name == resolverSeat })
|
||||
}
|
||||
|
||||
// restStep is the machines of one step of the rest (novox/hq ADR 0353 §2), in the order given: each machine, unless
|
||||
// a build it would carry has passed no gate and an earlier machine of the step already carries it — that build's
|
||||
// first machine. Pure.
|
||||
//
|
||||
// **The node-engine and the mesh's resolver still go one machine at a time** (ADR 0353 §9): a step carries a
|
||||
// build of either to one machine only, judged before the next, since a fault that cuts a machine off the bus
|
||||
// shows only as its silence and no rollback reaches it.
|
||||
func restStep(nodes []string, waiting map[string][]inventory.CarriedMove, passed func(module, commit string) bool,
|
||||
alone func(module string) bool) []string {
|
||||
first := map[string]string{}
|
||||
var out []string
|
||||
for _, n := range nodes {
|
||||
var unpassed []string
|
||||
for _, mv := range waiting[n] {
|
||||
if !passed(mv.Module, mv.To) || alone(mv.Module) {
|
||||
unpassed = append(unpassed, mv.Module+"@"+mv.To)
|
||||
}
|
||||
}
|
||||
if slices.ContainsFunc(unpassed, func(k string) bool { _, taken := first[k]; return taken }) {
|
||||
continue
|
||||
}
|
||||
for _, k := range unpassed {
|
||||
first[k] = n
|
||||
}
|
||||
out = append(out, n)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// silentObservation is what the gate's probe says of the newest release whose rest went silent (novox/hq ADR 0353
|
||||
// §4): machines that never reported on their send, left as they were — loud, never a success said by nothing.
|
||||
func silentObservation(ctx context.Context, inv *inventory.Inventory) ([]conditions.Observation, error) {
|
||||
recent, err := inv.RecentPlans(ctx, 20)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
for _, p := range recent {
|
||||
if p.Release == nil || p.Release.Settings != nil || p.Open() {
|
||||
continue
|
||||
}
|
||||
if len(p.Release.Silent) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
machines := append([]string(nil), p.Release.Silent...)
|
||||
sort.Strings(machines)
|
||||
o := conditions.Observation{Scope: conditions.ScopeMesh, ID: "release", Token: "silent", Kind: "release-silent",
|
||||
Severity: conditions.Warning, Resolver: conditions.ResolverOperator, Also: machines,
|
||||
Summary: fmt.Sprintf("%s: %s said nothing on the send of the builds that waited for a gate within %s, and were "+
|
||||
"left as they were", p.ID, strings.Join(machines, ", "), inventory.ApplySilentAfter)}
|
||||
w := releaseSilentWords(machines)
|
||||
o.Headline, o.Explanation, o.Resolved, o.Needs = w.Headline, w.Explanation, w.Resolved, w.Needs
|
||||
return []conditions.Observation{o}, nil
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// backlogObservation is what the gate's probe says of a release held after a failure.
|
||||
func backlogObservation() []conditions.Observation {
|
||||
if backlogNow.held == "" || len(backlogNow.waiting) == 0 {
|
||||
|
||||
@@ -251,6 +251,20 @@ func gates(p inventory.Plan, edges []inventory.Edge, rollsOut func(string) bool)
|
||||
return out
|
||||
}
|
||||
|
||||
// noMoveWhy is what a plan says of a module of its tier whose build was made from the source every machine
|
||||
// running it runs (novox/hq issue 280): nothing is sent for it.
|
||||
// noMovePrefix opens it, and is what noMove reads: one prefix for the writer and the reader.
|
||||
const (
|
||||
noMovePrefix = "no move"
|
||||
noMoveWhy = noMovePrefix + ": made from the source every machine running it runs"
|
||||
)
|
||||
|
||||
// noMove is whether the plan marked the module no move and sent it nothing. Read from what the plan says,
|
||||
// so a plan kept before the mark had a name of its own is read the same (novox/hq issue 648).
|
||||
func noMove(s *inventory.PlanModule) bool {
|
||||
return s != nil && s.FirstAt == nil && s.Gate == nil && strings.HasPrefix(s.Why, noMovePrefix)
|
||||
}
|
||||
|
||||
// applied says whether every machine running the module has reported since the module was built.
|
||||
// appliedEach is applied with a moment of its own for each machine: the reports that count are the ones
|
||||
// after that machine was sent the build (novox/hq issue 256).
|
||||
@@ -595,7 +609,7 @@ func advancePlans(ctx context.Context, open *stores) {
|
||||
return
|
||||
}
|
||||
defer release()
|
||||
// What waits for a gate, released one machine at a time (ADR 0236).
|
||||
// What waits for a gate, released one machine first, then the rest at once (ADR 0236, ADR 0353).
|
||||
if _, err := releaseBacklog(ctx, open, ""); err != nil {
|
||||
fmt.Printf("plans: what waits for a gate could not be looked at: %v\n", err)
|
||||
}
|
||||
@@ -805,7 +819,7 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
} else if same {
|
||||
now := time.Now().UTC()
|
||||
state.SentAt = &now
|
||||
state.Why = "no move: made from the source every machine running it runs"
|
||||
state.Why = noMoveWhy
|
||||
fmt.Printf("%s: %s built from the source every machine running it runs: no move, nothing sent\n", p.ID, m)
|
||||
continue
|
||||
}
|
||||
@@ -995,6 +1009,25 @@ func advanceOnce(ctx context.Context, open *stores, p *inventory.Plan,
|
||||
if state.BuiltAt != nil {
|
||||
since = *state.BuiltAt
|
||||
}
|
||||
// **A build that is no move is waited on by the send that last carried it** (novox/hq issue 648).
|
||||
// Nothing was sent for it, so no report follows its marking or its build: a machine that does not
|
||||
// happen to report again held the walk for ever. Each machine runs it once it answered its last
|
||||
// send — the one whose builds made it no move — whenever that was.
|
||||
if noMove(state) {
|
||||
byNode := map[string]inventory.Reported{}
|
||||
for _, r := range reports {
|
||||
byNode[r.Node] = r
|
||||
}
|
||||
if ok, on := appliedEach(m, 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)
|
||||
// a module is sent twice — the first machine, then the rest — and SentAt is the second.
|
||||
// Asked of every machine, the first machine's report, made between the two sends, read as
|
||||
@@ -1056,10 +1089,36 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string
|
||||
runningOf map[string][]string) error {
|
||||
inv := open.inventory
|
||||
before, beforeKnown, _ := inv.SentBuilds(ctx, node)
|
||||
// **A machine already sent what this send is for is not sent again** (novox/hq ADR 0353 §6): sent by this plan
|
||||
// before a controller was replaced and before it was kept, or by another walk. Judged from that send, and from
|
||||
// the readings of a gate already judging it there.
|
||||
// Only a send the machine reported it applied is one it has; one it never reported on is sent, as before
|
||||
// (issue 280: what it ran before is then that build, and a failure leaves it as it was).
|
||||
if beforeKnown && slices.IndexFunc(modules, func(m string) bool { return !sameCommit(before[m], p.Modules[m].Commit) }) < 0 &&
|
||||
appliedLastSend(ctx, inv, node) {
|
||||
if others, err := gatedMoves(ctx, open, node, nil); err == nil && len(others) == 0 {
|
||||
adoptSend(ctx, open, p, node, modules, before, runningOf)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
// **What the machine ran before is kept before the send is made** (novox/hq ADR 0353 §6): a controller replaced
|
||||
// between the send and keeping it reads it here, and a failure still puts the machine back.
|
||||
kept := false
|
||||
for _, m := range modules {
|
||||
s := p.Modules[m]
|
||||
if beforeKnown && (s.Previous == "" || !sameCommit(before[m], s.Commit)) && s.Previous != before[m] {
|
||||
s.Previous, kept = before[m], true
|
||||
}
|
||||
}
|
||||
if kept {
|
||||
if err := inv.SavePlan(ctx, p); err != nil {
|
||||
return fmt.Errorf("what %s ran before is not kept, so nothing is sent: %w", node, err)
|
||||
}
|
||||
}
|
||||
owns := make([]inventory.CarriedMove, 0, len(modules))
|
||||
for _, m := range modules {
|
||||
s := p.Modules[m]
|
||||
owns = append(owns, inventory.CarriedMove{Module: m, Node: node, From: before[m], To: s.Commit, Build: s.Build})
|
||||
owns = append(owns, inventory.CarriedMove{Module: m, Node: node, From: s.Previous, To: s.Commit, Build: s.Build})
|
||||
}
|
||||
// A gated send (ADR 0236): everything waiting on the machine goes with the tier, and the gate judges
|
||||
// all of it there.
|
||||
@@ -1074,10 +1133,7 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string
|
||||
sentWhat := sentNow(ctx, inv, sent)
|
||||
for _, m := range modules {
|
||||
s := p.Modules[m]
|
||||
// What the first machine ran before: what a failed gate puts back (ADR 0236).
|
||||
if beforeKnown {
|
||||
s.Previous = before[m]
|
||||
}
|
||||
// What the first machine ran before (ADR 0236), kept above before the send.
|
||||
s.First, s.FirstAt = sent, &now
|
||||
s.Gate = &inventory.PlanGate{Component: coreComponent(m), Machines: firstRunning(sent, runningOf[m]),
|
||||
From: s.Previous, To: s.Commit, Since: &now, Sent: sentWhat}
|
||||
@@ -1106,6 +1162,113 @@ func firstSend(ctx context.Context, open *stores, p *inventory.Plan, node string
|
||||
return nil
|
||||
}
|
||||
|
||||
// adoptSend records a first send the machine already had (novox/hq ADR 0353 §6, §7): nothing is sent; the gate counts
|
||||
// from the machine's last send, and takes over the readings of an open gate of another walk judging the same build
|
||||
// there — whose carried move also says what the machine ran before, for a rollback.
|
||||
func adoptSend(ctx context.Context, open *stores, p *inventory.Plan, node string, modules []string,
|
||||
before map[string]string, runningOf map[string][]string) {
|
||||
inv := open.inventory
|
||||
since := time.Now().UTC()
|
||||
last, found, err := inv.SentTo(ctx, node)
|
||||
if err == nil && found && !last.At.IsZero() {
|
||||
since = last.At
|
||||
}
|
||||
lead := modules[0]
|
||||
other, id, from := judgingThere(ctx, inv, p.ID, lead, node, p.Modules[lead].Commit, p.Modules[lead].Build)
|
||||
for _, m := range modules {
|
||||
s := p.Modules[m]
|
||||
// What it ran before: kept by this walk before its send was made (§6); else what the other walk's send
|
||||
// moved it from; else the build itself, so a failure leaves it as it was (issue 280).
|
||||
switch {
|
||||
case s.Previous != "":
|
||||
case m == lead && from != "":
|
||||
s.Previous = from
|
||||
default:
|
||||
s.Previous = before[m]
|
||||
}
|
||||
at := since
|
||||
s.First, s.FirstAt = []string{node}, &at
|
||||
s.Gate = &inventory.PlanGate{Component: coreComponent(m), Machines: firstRunning([]string{node}, runningOf[m]),
|
||||
From: s.Previous, To: s.Commit, Since: &at}
|
||||
if found {
|
||||
s.Gate.Sent = map[string]inventory.SentDeclaration{node: last}
|
||||
}
|
||||
s.GatedBy = ""
|
||||
if m == lead {
|
||||
s.Gate.TakeReadings(other, modules, id)
|
||||
} else {
|
||||
s.GatedBy = lead
|
||||
}
|
||||
}
|
||||
p.State = inventory.PlanRolling
|
||||
p.Note = fmt.Sprintf("tier %d built; %s already runs %s's build: nothing sent again, judged from that send",
|
||||
p.Tier, node, strings.Join(modules, ", "))
|
||||
if id != "" {
|
||||
p.Note += ", going on from the readings of " + id
|
||||
}
|
||||
fmt.Printf("%s: %s\n", p.ID, p.Note)
|
||||
}
|
||||
|
||||
// appliedLastSend says a machine reported it applied the declaration it was sent last.
|
||||
func appliedLastSend(ctx context.Context, inv *inventory.Inventory, node string) bool {
|
||||
reports, err := inv.LastReports(ctx)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
for _, r := range reports {
|
||||
if r.Node == node {
|
||||
return r.Current && r.At != nil && r.Outcome == inventory.OutcomeApplied
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// judgingThere is an open gate of another walk judging a module's build on a machine (novox/hq ADR 0353 §7), the
|
||||
// walk's id, and what the machine ran before; nil when none does.
|
||||
func judgingThere(ctx context.Context, inv *inventory.Inventory, self, module, node, commit, build string) (*inventory.PlanGate, string, string) {
|
||||
plans, err := inv.OpenPlans(ctx)
|
||||
if err != nil {
|
||||
return nil, "", ""
|
||||
}
|
||||
// The same build, by its id where both name one, else by its commit.
|
||||
same := func(b, c string) bool {
|
||||
if b != "" && build != "" {
|
||||
return b == build
|
||||
}
|
||||
return sameCommit(c, commit)
|
||||
}
|
||||
usable := func(g *inventory.PlanGate) bool {
|
||||
return g != nil && !g.Pending && g.Since != nil && (g.Verdict == "" || g.Verdict == inventory.GatePassed) &&
|
||||
slices.Contains(g.Machines, node)
|
||||
}
|
||||
for _, q := range plans {
|
||||
if q.ID == self {
|
||||
continue
|
||||
}
|
||||
if r := q.Release; r != nil {
|
||||
gates := []*inventory.PlanGate{r.Gate}
|
||||
for _, g := range r.Gates {
|
||||
gates = append(gates, g)
|
||||
}
|
||||
for _, g := range gates {
|
||||
if !usable(g) {
|
||||
continue
|
||||
}
|
||||
for _, c := range g.Carried {
|
||||
if c.Module == module && c.Node == node && same(c.Build, c.To) {
|
||||
return g, q.ID, c.From
|
||||
}
|
||||
}
|
||||
}
|
||||
continue
|
||||
}
|
||||
if s := q.Modules[module]; s != nil && usable(s.Gate) && same(s.Build, s.Commit) {
|
||||
return s.Gate, q.ID, s.Previous
|
||||
}
|
||||
}
|
||||
return nil, "", ""
|
||||
}
|
||||
|
||||
// passedWith keeps, on a module sent with others, the verdict of the gate that judged the send: the gate
|
||||
// on the first of them passed, and its pass was kept for every build it carried (passCarried).
|
||||
func passedWith(module string, s *inventory.PlanModule, lead string, g *inventory.PlanGate) {
|
||||
@@ -1594,9 +1757,19 @@ func plansCommand(ctx context.Context, args []string) error {
|
||||
// A release plan's walk (ADR 0236): machines done, the one judged, those to come.
|
||||
fmt.Printf(" machines in order: %s; done: %s; skipped: %s\n", strings.Join(r.Order, ", "),
|
||||
orNone(strings.Join(r.Done, ", ")), orNone(strings.Join(r.Skipped, ", ")))
|
||||
if r.Gate != nil {
|
||||
fmt.Printf(" %s\n", gateLine(r.Gate))
|
||||
for _, c := range r.Gate.Carried {
|
||||
if len(r.Failed)+len(r.Silent) > 0 {
|
||||
fmt.Printf(" failed: %s; silent: %s\n", orNone(strings.Join(r.Failed, ", ")), orNone(strings.Join(r.Silent, ", ")))
|
||||
}
|
||||
gates := []*inventory.PlanGate{r.Gate}
|
||||
for _, n := range r.Order {
|
||||
gates = append(gates, r.Gates[n])
|
||||
}
|
||||
for _, g := range gates {
|
||||
if g == nil {
|
||||
continue
|
||||
}
|
||||
fmt.Printf(" %s\n", gateLine(g))
|
||||
for _, c := range g.Carried {
|
||||
fmt.Printf(" %-22s %s → %s\n", c.Module, short(c.From), short(c.To))
|
||||
if c.Recreates != "" {
|
||||
fmt.Printf(" %-22s %s\n", "", c.Recreates)
|
||||
|
||||
@@ -0,0 +1,372 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// novox/hq ADR 0353: after its first machine passes, a build goes to every other machine at once, each judged by
|
||||
// its own gate and a failing one put back by itself; and a gate's readings outlive the controller that took them —
|
||||
// a machine already sent what a send is for is not sent again, and its judging goes on from what was read.
|
||||
//
|
||||
// Seen 2026-10-14: release-1791930194184991192 sent ace, g14, shanks and the control node one after another in
|
||||
// eleven minutes; the new controller its last send started sent the control node again the controller build it
|
||||
// had just applied, and judged it from its first reading.
|
||||
|
||||
// aBacklogOnThree is the backlog fixture with a third machine, desk, running app at c1 like the others.
|
||||
func aBacklogOnThree(t *testing.T) *backlogMesh {
|
||||
t.Helper()
|
||||
b := aBacklog(t)
|
||||
ctx := t.Context()
|
||||
inv := b.open.inventory
|
||||
if _, err := inv.AddNode(ctx, "desk"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := inv.Assign(ctx, "desk", "app"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RecordSent(ctx, nodeID(t, b.open, "desk"), "d-desk", map[string]string{"app": "c1"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
wasHeard := releaseHeard
|
||||
releaseHeard = func(context.Context, *stores) (map[string]bool, error) {
|
||||
return map[string]bool{"anchor": true, "laptop": true, "desk": true}, nil
|
||||
}
|
||||
t.Cleanup(func() { releaseHeard = wasHeard })
|
||||
return b
|
||||
}
|
||||
|
||||
// events records every send and every judging, in order.
|
||||
func events(t *testing.T) *[]string {
|
||||
t.Helper()
|
||||
var said []string
|
||||
wasSend, wasGather := sendRollout, gatherGateFacts
|
||||
sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
|
||||
said = append(said, "send "+strings.Join(names, ","))
|
||||
return wasSend(ctx, open, names)
|
||||
}
|
||||
gatherGateFacts = func(ctx context.Context, open *stores, component string) (gateFacts, error) {
|
||||
said = append(said, "judge")
|
||||
return wasGather(ctx, open, component)
|
||||
}
|
||||
t.Cleanup(func() { sendRollout, gatherGateFacts = wasSend, wasGather })
|
||||
return &said
|
||||
}
|
||||
|
||||
// Once the first machine passed, the rest are both sent before either is judged (ADR 0353 §1); before, the second
|
||||
// was judged to its verdict before the third was sent.
|
||||
func TestAReleaseSendsTheRestAtOnceAfterTheFirstPasses(t *testing.T) {
|
||||
judgeLeavesAsTheCore(t)
|
||||
b := aBacklogOnThree(t)
|
||||
ctx := t.Context()
|
||||
said := events(t)
|
||||
gateEvery = 0
|
||||
for i := 0; i < 12; i++ {
|
||||
advancePlans(ctx, b.open)
|
||||
}
|
||||
p := b.release(t)
|
||||
if p.State != inventory.PlanDone || len(p.Release.Done) != 3 {
|
||||
t.Fatalf("the release is %s: %s %+v", p.State, p.Note, p.Release)
|
||||
}
|
||||
desk, laptop := slices.Index(*said, "send desk"), slices.Index(*said, "send laptop")
|
||||
if desk < 0 || laptop < 0 || slices.Index(*said, "send anchor") > min(desk, laptop) {
|
||||
t.Fatalf("sends and judgings: %v", *said)
|
||||
}
|
||||
if slices.Contains((*said)[min(desk, laptop):max(desk, laptop)], "judge") {
|
||||
t.Fatalf("the rest were sent one at a time, each judged before the next: %v", *said)
|
||||
}
|
||||
if v, found, err := b.open.inventory.GateOf(ctx, "build-late-c2"); err != nil || !found || v.Verdict != inventory.GatePassed {
|
||||
t.Fatalf("late's pass on laptop was not kept: %+v %v %v", v, found, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A machine of the rest whose gate fails is put back at once, and the other machine of the step goes on to its own
|
||||
// verdict (ADR 0353 §3); before, the walk stopped at the failure and the machine after it was never sent.
|
||||
func TestARestMachineThatFailsIsPutBackAloneAndTheOthersAreJudgedOn(t *testing.T) {
|
||||
judgeLeavesAsTheCore(t)
|
||||
b := aBacklogOnThree(t)
|
||||
ctx := t.Context()
|
||||
inv := b.open.inventory
|
||||
gateEvery = 0
|
||||
b.broken["desk"] = true // desk fails what it is sent in the rest's step
|
||||
advancePlans(ctx, b.open)
|
||||
for i := 0; i < 12 && b.release(t).Open(); i++ {
|
||||
advancePlans(ctx, b.open)
|
||||
}
|
||||
p := b.release(t)
|
||||
if p.State != inventory.PlanFailed || !slices.Equal(p.Release.Failed, []string{"desk"}) {
|
||||
t.Fatalf("the release is %s: %s %+v", p.State, p.Note, p.Release)
|
||||
}
|
||||
if !slices.ContainsFunc(b.sent, func(s []string) bool { return slices.Equal(s, []string{"laptop"}) }) {
|
||||
t.Fatalf("laptop was never sent: %v", b.sent)
|
||||
}
|
||||
if !slices.Contains(p.Release.Done, "laptop") {
|
||||
t.Fatalf("laptop has no verdict of its own: %+v", p.Release)
|
||||
}
|
||||
// late waited on laptop alone and was healthy there: its pass is kept, whatever desk did.
|
||||
if v, found, err := inv.GateOf(ctx, "build-late-c2"); err != nil || !found || v.Verdict != inventory.GatePassed {
|
||||
t.Fatalf("late's pass on laptop was not kept: %+v %v %v", v, found, err)
|
||||
}
|
||||
// app failed on desk: marked, and put back wherever it was sent (ADR 0236).
|
||||
if failed, _ := inv.GateFailed(ctx, "build-app-c2"); !failed {
|
||||
t.Fatal("app's build is not marked failed")
|
||||
}
|
||||
if sent, _, _ := inv.SentBuilds(ctx, "desk"); sent["app"] != "c1" {
|
||||
t.Fatalf("desk was not put back: %v", sent)
|
||||
}
|
||||
}
|
||||
|
||||
// A walk whose first machine already runs its build under another walk's gate sends nothing and goes on from that
|
||||
// gate's readings (ADR 0353 §7); before, it sent the machine again and judged from its first reading.
|
||||
func TestAWalkTakesOverTheReadingsOfAGateJudgingTheSameBuildThere(t *testing.T) {
|
||||
judgeLeavesAsTheCore(t)
|
||||
b := aBacklog(t)
|
||||
ctx := t.Context()
|
||||
inv := b.open.inventory
|
||||
advancePlans(ctx, b.open) // the release sends anchor app c2, and reads it once
|
||||
release := b.release(t)
|
||||
judging := release.Release.Gate
|
||||
if judging == nil || judging.Pending || judging.Passes == 0 {
|
||||
t.Fatalf("the release is not judging anchor: %+v", judging)
|
||||
}
|
||||
built := time.Now().UTC()
|
||||
plan := inventory.Plan{ID: "plan-app", Repository: "novox/mesh-catalog", Branch: "main", Commit: "c2",
|
||||
Created: built, State: inventory.PlanBuilding, Tiers: [][]string{{"app"}},
|
||||
Modules: map[string]*inventory.PlanModule{"app": {State: "built", BuiltAt: &built, Commit: "c2",
|
||||
Build: "build-app-c2"}}}
|
||||
if err := inv.SavePlan(ctx, &plan); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
advancePlans(ctx, b.open)
|
||||
if len(b.sent) != 1 {
|
||||
t.Fatalf("anchor was sent again what it runs: %v", b.sent)
|
||||
}
|
||||
p, err := inv.PlanByID(ctx, "plan-app")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
g := p.Modules["app"].Gate
|
||||
if g == nil || g.Since == nil || !g.Since.Equal(*judging.Since) || g.Passes < judging.Passes || g.TookOver != release.ID {
|
||||
t.Fatalf("the walk's gate starts from zero: %+v (the release's: %+v)", g, judging)
|
||||
}
|
||||
if p.Modules["app"].Previous != "c1" {
|
||||
t.Fatalf("what anchor ran before is %q, not c1", p.Modules["app"].Previous)
|
||||
}
|
||||
}
|
||||
|
||||
// A walk's first send that was made and not kept — the controller replaced between the two — is not made again
|
||||
// by the controller that takes over: the gate counts from that send (ADR 0353 §5, §6).
|
||||
func TestAFirstSendMadeAndNotKeptIsNotSentAgain(t *testing.T) {
|
||||
judgeLeavesAsTheCore(t)
|
||||
g := aGateMesh(t)
|
||||
ctx := t.Context()
|
||||
// The old controller sends anchor app c2 and is replaced before it keeps the send.
|
||||
made := sendRollout
|
||||
sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
|
||||
if _, err := made(ctx, open, names); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return nil, context.Canceled
|
||||
}
|
||||
advancePlans(ctx, g.open)
|
||||
sendRollout = made
|
||||
if len(g.sent) != 1 || g.plan(t).Modules["app"].FirstAt != nil {
|
||||
t.Fatalf("the fixture's send was not made, or was kept: %v", g.sent)
|
||||
}
|
||||
last, _, err := g.open.inventory.SentTo(ctx, "anchor")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
advancePlans(ctx, g.open)
|
||||
if len(g.sent) != 1 {
|
||||
t.Fatalf("the send was made again: %v", g.sent)
|
||||
}
|
||||
s := g.plan(t).Modules["app"]
|
||||
if s.FirstAt == nil || s.Gate == nil || s.Gate.Since == nil || !s.Gate.Since.Equal(last.At) {
|
||||
t.Fatalf("the gate does not count from the send made: %+v %+v", s, s.Gate)
|
||||
}
|
||||
// What anchor ran before was kept before the send, so a failure still puts it back (§5).
|
||||
if s.Previous != "c1" || s.Gate.From != "c1" {
|
||||
t.Fatalf("what anchor ran before is %q (gate from %q), not c1", s.Previous, s.Gate.From)
|
||||
}
|
||||
}
|
||||
|
||||
// A release's send kept before it was made, and made, is judged by the controller that takes over without being
|
||||
// made again (ADR 0353 §8).
|
||||
func TestAReleaseSendKeptAndMadeIsNotMadeAgain(t *testing.T) {
|
||||
judgeLeavesAsTheCore(t)
|
||||
b := aBacklog(t)
|
||||
ctx := t.Context()
|
||||
inv := b.open.inventory
|
||||
p, err := releaseBacklog(ctx, b.open, "")
|
||||
if err != nil || p == nil {
|
||||
t.Fatalf("no release opened: %v", err)
|
||||
}
|
||||
if _, err := advanceRelease(ctx, b.open, p); err != nil || p.Release.Gate == nil || !p.Release.Gate.Pending {
|
||||
t.Fatalf("the send was not kept before it was made: %+v %v", p.Release, err)
|
||||
}
|
||||
if err := inv.SavePlan(ctx, p); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The send is made, and the controller is replaced before it keeps it.
|
||||
if _, err := sendRollout(ctx, b.open, []string{"anchor"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
advancePlans(ctx, b.open)
|
||||
if len(b.sent) != 1 {
|
||||
t.Fatalf("anchor was sent again: %v", b.sent)
|
||||
}
|
||||
g := b.release(t).Release.Gate
|
||||
if g == nil || g.Pending || g.Since == nil || len(g.Carried) == 0 {
|
||||
t.Fatalf("anchor is not judged on the send made: %+v", g)
|
||||
}
|
||||
}
|
||||
|
||||
// A build no gate has passed goes to one machine of a step, its first; a machine that would be its second waits
|
||||
// for the next step, and one whose every build passed goes now (ADR 0353 §2).
|
||||
func TestABuildThatPassedNoGateGoesToOneMachineOfAStep(t *testing.T) {
|
||||
waiting := map[string][]inventory.CarriedMove{
|
||||
"desk": {{Module: "app", To: "c2"}, {Module: "new", To: "n1"}},
|
||||
"laptop": {{Module: "app", To: "c2"}, {Module: "new", To: "n1"}},
|
||||
"shanks": {{Module: "app", To: "c2"}},
|
||||
"novox": {{Module: "app", To: "c2"}, {Module: "ctl", To: "k1"}},
|
||||
}
|
||||
passed := func(module, commit string) bool { return module == "app" && commit == "c2" }
|
||||
never := func(string) bool { return false }
|
||||
step := restStep([]string{"desk", "laptop", "shanks", "novox"}, waiting, passed, never)
|
||||
if !slices.Equal(step, []string{"desk", "shanks", "novox"}) {
|
||||
t.Fatalf("the step is %v: laptop would be new's second machine before it passed", step)
|
||||
}
|
||||
if step := restStep([]string{"laptop"}, waiting, func(string, string) bool { return true }, never); !slices.Equal(step, []string{"laptop"}) {
|
||||
t.Fatalf("a machine whose every build passed waits: %v", step)
|
||||
}
|
||||
}
|
||||
|
||||
// The node-engine and the mesh's resolver go to the rest one machine at a time, even passed (ADR 0353 §9): a step
|
||||
// carries either to one machine only, while what else passed goes everywhere at once.
|
||||
func TestTheNodeEngineAndTheResolverGoToTheRestOneAtATime(t *testing.T) {
|
||||
waiting := map[string][]inventory.CarriedMove{
|
||||
"desk": {{Module: hostModule, To: "h2"}},
|
||||
"laptop": {{Module: hostModule, To: "h2"}, {Module: "app", To: "c2"}},
|
||||
"shanks": {{Module: "app", To: "c2"}},
|
||||
"novox": {{Module: "dns", To: "d2"}},
|
||||
"ace": {{Module: "dns", To: "d2"}},
|
||||
}
|
||||
resolver := catalogue.Manifest{Module: "dns", Claims: []catalogue.Claim{{Name: resolverSeat}}}
|
||||
alone := func(m string) bool {
|
||||
if m == "dns" {
|
||||
return oneAtATime(m, resolver)
|
||||
}
|
||||
return oneAtATime(m, catalogue.Manifest{Module: m})
|
||||
}
|
||||
step := restStep([]string{"desk", "laptop", "shanks", "novox", "ace"}, waiting,
|
||||
func(string, string) bool { return true }, alone)
|
||||
if !slices.Equal(step, []string{"desk", "shanks", "novox"}) {
|
||||
t.Fatalf("the step is %v: the node-engine and the resolver each reach one machine of it", step)
|
||||
}
|
||||
if oneAtATime("app", catalogue.Manifest{Module: "app"}) {
|
||||
t.Fatal("a leaf module is held to one machine at a time")
|
||||
}
|
||||
}
|
||||
|
||||
// A first machine that fails still stops the walk (ADR 0236 §5, kept by ADR 0353 §3): the rest are never sent.
|
||||
func TestAFirstMachineThatFailsStillStopsTheWalk(t *testing.T) {
|
||||
b := aBacklogOnThree(t)
|
||||
ctx := t.Context()
|
||||
gateEvery = 0
|
||||
b.broken["anchor"] = true
|
||||
advancePlans(ctx, b.open)
|
||||
for i := 0; i < 6 && b.release(t).Open(); i++ {
|
||||
advancePlans(ctx, b.open)
|
||||
}
|
||||
if p := b.release(t); p.State != inventory.PlanFailed || p.Release.Rest {
|
||||
t.Fatalf("the release is %s: %s %+v", p.State, p.Note, p.Release)
|
||||
}
|
||||
for _, s := range b.sent {
|
||||
if !slices.Equal(s, []string{"anchor"}) {
|
||||
t.Fatalf("sent %v: the rest after a failed first machine", b.sent)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The control node is sent last in a step of the rest (ADR 0353 §1): the machine holding the controller is
|
||||
// ordered after every other.
|
||||
func TestTheControlNodeIsSentLastInAStep(t *testing.T) {
|
||||
judgeLeavesAsTheCore(t)
|
||||
b := aBacklogOnThree(t)
|
||||
ctx := t.Context()
|
||||
register(t, b.open, catalogue.Manifest{Module: catalogue.ControllerSeatName, Version: "1"})
|
||||
if _, err := b.open.inventory.Assign(ctx, "anchor", catalogue.ControllerSeatName); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
said := events(t)
|
||||
gateEvery = 0
|
||||
for i := 0; i < 12 && (len(b.sent) == 0 || b.release(t).Open()); i++ {
|
||||
advancePlans(ctx, b.open)
|
||||
}
|
||||
if p := b.release(t); !slices.Equal(p.Release.Order, []string{"desk", "laptop", "anchor"}) {
|
||||
t.Fatalf("the order is %v", p.Release.Order)
|
||||
}
|
||||
laptop, anchor := slices.Index(*said, "send laptop"), slices.Index(*said, "send anchor")
|
||||
if laptop < 0 || anchor < laptop || slices.Contains((*said)[laptop:anchor], "judge") {
|
||||
t.Fatalf("the control node is not sent last in the rest's step: %v", *said)
|
||||
}
|
||||
}
|
||||
|
||||
// A machine of the rest that says nothing on its send is left silent after fifteen minutes: not failed, not put
|
||||
// back; a rest that went silent whole ends failed, said as a condition (ADR 0353 §4).
|
||||
func TestARestMachineSilentPastItsBoundIsLeftAndSaid(t *testing.T) {
|
||||
judgeLeavesAsTheCore(t)
|
||||
b := aBacklog(t)
|
||||
ctx := t.Context()
|
||||
inv := b.open.inventory
|
||||
reporting := sendRollout
|
||||
sendRollout = func(ctx context.Context, open *stores, names []string) ([]string, error) {
|
||||
if slices.Equal(names, []string{"laptop"}) { // laptop sleeps: sent, and says nothing
|
||||
b.sent = append(b.sent, names)
|
||||
current, _ := open.inventory.CurrentBuilds(ctx)
|
||||
return names, open.inventory.RecordSent(ctx, nodeID(t, open, "laptop"), "d-laptop-asleep",
|
||||
map[string]string{"app": current["app"].Commit, "late": current["late"].Commit, "same": current["same"].Commit})
|
||||
}
|
||||
return reporting(ctx, open, names)
|
||||
}
|
||||
t.Cleanup(func() { sendRollout = reporting })
|
||||
gateEvery = 0
|
||||
for i := 0; i < 8; i++ {
|
||||
advancePlans(ctx, b.open)
|
||||
}
|
||||
p := b.release(t)
|
||||
g := p.Release.Gates["laptop"]
|
||||
if !p.Open() || g == nil || g.Verdict != "" || g.Silent {
|
||||
t.Fatalf("laptop is judged before fifteen minutes: %s %s %+v", p.State, p.Note, g)
|
||||
}
|
||||
since := time.Now().Add(-inventory.ApplySilentAfter - time.Minute)
|
||||
g.Since = &since
|
||||
if err := inv.SavePlan(ctx, &p); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
advancePlans(ctx, b.open)
|
||||
p = b.release(t)
|
||||
if p.State != inventory.PlanFailed || !slices.Equal(p.Release.Silent, []string{"laptop"}) || len(p.Release.Failed) > 0 {
|
||||
t.Fatalf("the release is %s: %s %+v", p.State, p.Note, p.Release)
|
||||
}
|
||||
if failed, _ := inv.GateFailed(ctx, "build-late-c2"); failed {
|
||||
t.Fatal("a silent machine's build was marked failed")
|
||||
}
|
||||
if sent, _, _ := inv.SentBuilds(ctx, "laptop"); sent["late"] != "c2" {
|
||||
t.Fatalf("laptop was put back: %v", sent)
|
||||
}
|
||||
obs, err := silentObservation(ctx, inv)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(obs) != 1 || obs[0].Kind != "release-silent" || !slices.Equal(obs[0].Also, []string{"laptop"}) {
|
||||
t.Fatalf("the silence is not said: %+v", obs)
|
||||
}
|
||||
}
|
||||
@@ -457,6 +457,28 @@ func (a *verbArguments) commandLine() ([]string, error) {
|
||||
return nil, err
|
||||
}
|
||||
return []string{"forget-checks", str("owner"), str("repo"), str("number")}, nil
|
||||
case "try":
|
||||
// A branch tried before its pull request (novox/hq issue 686): asked, or read by the id it answered.
|
||||
if id := str("id"); id != "" {
|
||||
return []string{"try", "--id", id}, nil
|
||||
}
|
||||
if a.given["repository"] == "" && a.given["ref"] == "" {
|
||||
// Neither shape named: the command says its usage, which names both, as `build` does.
|
||||
return []string{"try"}, nil
|
||||
}
|
||||
if err := need("repository", "ref"); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
argv := []string{"try", str("repository"), str("ref")}
|
||||
for _, flag := range []string{"base", "tests"} {
|
||||
if v := str(flag); v != "" {
|
||||
argv = append(argv, "--"+flag, v)
|
||||
}
|
||||
}
|
||||
if on("store-only") {
|
||||
argv = append(argv, "--store-only")
|
||||
}
|
||||
return argv, nil
|
||||
case "cancel", "kill":
|
||||
if err := need("id"); err != nil {
|
||||
return nil, err
|
||||
@@ -1018,6 +1040,8 @@ var jsonVerbs = map[string]bool{"status": true, "seats": true, "plan": true, "co
|
||||
"delivery": true,
|
||||
// The build queue, each ask with its position among the waiting (novox/hq issue 455).
|
||||
"queue": true,
|
||||
// A branch tried before its pull request (novox/hq issue 686).
|
||||
"try": true,
|
||||
// A plan followed (novox/hq issue 229): `plans wait` answers JSON; its other forms say text, and so answer no data.
|
||||
"plans": true}
|
||||
|
||||
@@ -1308,6 +1332,10 @@ func seatToolHandlers() (map[string]link.ToolHandler, []string, error) {
|
||||
link.AnswersWithin(ctx, bound+planWaitStart)
|
||||
}
|
||||
}
|
||||
if verb == "try" && len(argv) > 1 && argv[1] == "--id" {
|
||||
// Reading a try holds its answer for the verdict, up to its bound (novox/hq issue 686, 229).
|
||||
link.AnswersWithin(ctx, tryWait+planWaitStart)
|
||||
}
|
||||
if answer, read := readHere(ctx, argv); read {
|
||||
return answer, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,353 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"flag"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"os"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// A branch is tried before its pull request (novox/hq issue 686).
|
||||
//
|
||||
// A repository's store tests skip without a store, and no agent may start a database on its machine. So a
|
||||
// builder pushed, a reviewer read, and only the pull request's merge check ran those tests for the first time:
|
||||
// each failure there cost a round of fifteen to thirty minutes (mesh-controller #283, #310, #312).
|
||||
//
|
||||
// `try <owner>/<repository> <ref>` asks the build seat for the repository's own check of a pushed branch or
|
||||
// commit, as the merge check runs it (internal/builder/check.go): the ref merged into the trunk's head, the
|
||||
// trunk's merge-check.sh, each part in its toolchain's container, against a throwaway store and bus. It is a
|
||||
// check and nothing more:
|
||||
//
|
||||
// - **no pull request, so no status**: its outcome is never said as `checked`, which is what the forge's
|
||||
// holder sets statuses from (isTry, in Built);
|
||||
// - **nothing registered, nothing built**: it names no module of the graph, so the merge gate does not run and
|
||||
// no build is made for a merge (novox/hq ADR 0350's scratch builds are a pull request's check's alone);
|
||||
// - **a check in the build queue**: after deliveries (link.KindOf, novox/hq issue 558), and bounded by the
|
||||
// check's own bound (checkTimeout);
|
||||
// - **the code a pull request's check would run anyway**: only a repository of the mesh, whose every pull
|
||||
// request the build seat already checks, in containers that hold no runtime socket.
|
||||
//
|
||||
// A check outlasts a seat verb's one answer, so the ask answers its id at once and `try --id <id>` reads what
|
||||
// came of it from the build's log, where the build seat says it whole (link.StepTried): holding its answer up
|
||||
// to tryWait for a verdict, like `plans wait` (novox/hq issue 229). Nothing is kept beside the log.
|
||||
|
||||
// tryWait is how long reading a try holds its answer for the verdict. A variable so a test need not wait.
|
||||
var tryWait = 20 * time.Second
|
||||
|
||||
// A ref as the forge is asked for it — a branch or a commit, never an option — and a `go test -run` pattern.
|
||||
var (
|
||||
tryRef = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]*$`)
|
||||
tryTests = regexp.MustCompile(`^[A-Za-z0-9_^$|().*+?/\[\]-]+$`)
|
||||
)
|
||||
|
||||
// tryAnswer is what `try` answers, asking or reading.
|
||||
type tryAnswer struct {
|
||||
ID string `json:"id"`
|
||||
// State is asked (just now), waiting (nothing said of it yet: in the build queue), running, or the verdict:
|
||||
// pass, warning (the repository declares no merge-check.sh), fail, or error (it could not run).
|
||||
State string `json:"state"`
|
||||
Said string `json:"said"`
|
||||
Repository string `json:"repository,omitempty"`
|
||||
Ref string `json:"ref,omitempty"`
|
||||
Commit string `json:"commit,omitempty"`
|
||||
Base string `json:"base,omitempty"`
|
||||
Trunk string `json:"trunk,omitempty"`
|
||||
On string `json:"on,omitempty"`
|
||||
Took string `json:"took,omitempty"`
|
||||
Failed []string `json:"failed,omitempty"`
|
||||
// Last is the last thing the build seat said of one still running.
|
||||
Last string `json:"last,omitempty"`
|
||||
}
|
||||
|
||||
// isTry says a check's outcome is a try's: marked so, or — echoed by a build seat from before the mark — a
|
||||
// check of no pull request and no group, which is nobody's status either.
|
||||
func isTry(result link.BuildResult) bool {
|
||||
c := result.Checked
|
||||
return c != nil && (c.Try || (c.Number == 0 && c.Group == ""))
|
||||
}
|
||||
|
||||
// asATry makes a check's ask a try's: the repository's own script alone, for no pull request.
|
||||
func asATry(request link.BuildRequest, tests string, storeOnly bool) link.BuildRequest {
|
||||
c := *request.Check
|
||||
c.Number, c.Try, c.Tests, c.StoreOnly = 0, true, tests, storeOnly
|
||||
c.Paths, c.Modules, c.New, c.Dependents, c.Manifests, c.Builds = nil, nil, nil, nil, nil, nil
|
||||
c.Judge, c.Plan, c.Group, c.Members = "", nil, "", nil
|
||||
request.Check = &c
|
||||
return request
|
||||
}
|
||||
|
||||
// maxTries is how many tries may wait or run at once, mesh-wide: each holds a build machine for minutes.
|
||||
const maxTries = 4
|
||||
|
||||
// triesIn reads the build queue for a new try of one commit: the id of a try of that commit already waiting or
|
||||
// running — the answer to a second ask — and whether there is room for another.
|
||||
func triesIn(q link.Queue, owner, repo, commit string) (same string, room bool) {
|
||||
n := 0
|
||||
for _, a := range q.Asks {
|
||||
if !link.IsTry(a.Request) || (a.State != link.AskWaiting && a.State != link.AskInFlight) {
|
||||
continue
|
||||
}
|
||||
n++
|
||||
c := a.Request.Check
|
||||
if same == "" && strings.EqualFold(c.Owner, owner) && strings.EqualFold(c.Repo, repo) && a.Request.Ref == commit {
|
||||
same = a.ID
|
||||
}
|
||||
}
|
||||
return same, n < maxTries
|
||||
}
|
||||
|
||||
// forgeCommit asks the forge, through its module's tools on the bus, which commit a branch or commit of a
|
||||
// repository is, and the repository's default branch. A variable so a test needs no forge.
|
||||
var forgeCommit = func(ctx context.Context, owner, repo, ref string) (commit, trunk string, err error) {
|
||||
js, err := aBus()
|
||||
if err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
defer js.Close()
|
||||
bus := link.OverNATS{Conn: js.Conn()}
|
||||
ask := func(path string, into any) error {
|
||||
raw, _ := json.Marshal(map[string]any{"path": path})
|
||||
answer, err := link.Ask(ctx, bus, "gitea", "gitea_api", raw, 30*time.Second)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if answer.Error != "" {
|
||||
return errors.New(answer.Error)
|
||||
}
|
||||
return json.Unmarshal(answer.Result, into)
|
||||
}
|
||||
var commits struct {
|
||||
Result []struct {
|
||||
SHA string `json:"sha"`
|
||||
} `json:"result"`
|
||||
}
|
||||
if err := ask("/repos/"+owner+"/"+repo+"/commits?limit=1&stat=false&verification=false&files=false&sha="+
|
||||
url.QueryEscape(ref), &commits); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
if len(commits.Result) == 0 || commits.Result[0].SHA == "" {
|
||||
return "", "", errors.New("the forge named no commit for it")
|
||||
}
|
||||
var found struct {
|
||||
Result struct {
|
||||
DefaultBranch string `json:"default_branch"`
|
||||
} `json:"result"`
|
||||
}
|
||||
if err := ask("/repos/"+owner+"/"+repo, &found); err != nil {
|
||||
return "", "", err
|
||||
}
|
||||
return commits.Result[0].SHA, found.Result.DefaultBranch, nil
|
||||
}
|
||||
|
||||
// tryCommand is `try <owner>/<repository> <ref> [--base <branch>] [--tests <pattern>] [--store-only]`, and
|
||||
// `try --id <id>` to read what came of one.
|
||||
func tryCommand(ctx context.Context, args []string) error {
|
||||
set := flag.NewFlagSet("try", flag.ContinueOnError)
|
||||
base := set.String("base", "", "the branch it is merged into (the repository's default branch when none)")
|
||||
tests := set.String("tests", "", "only the tests this `go test -run` pattern names")
|
||||
storeOnly := set.Bool("store-only", false, "only the repository's store tests")
|
||||
id := set.String("id", "", "a try's id: what came of it")
|
||||
positionals, err := parseAround(set, args)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
usage := errors.New("try <owner>/<repository> <branch or commit> [--base <branch>] [--tests <pattern>] " +
|
||||
"[--store-only], or try --id <id>")
|
||||
if *id != "" {
|
||||
if len(positionals) != 0 {
|
||||
return usage
|
||||
}
|
||||
// Before it is a token of a bus subject: an id, never a pattern.
|
||||
if !link.SafeBuildID(*id) {
|
||||
return fmt.Errorf("%q is not a try's id", *id)
|
||||
}
|
||||
return tryRead(ctx, *id)
|
||||
}
|
||||
if len(positionals) != 2 {
|
||||
return usage
|
||||
}
|
||||
owner, repo, isPath := strings.Cut(positionals[0], "/")
|
||||
ref := positionals[1]
|
||||
switch {
|
||||
case !isPath || !tryRef.MatchString(owner) || !tryRef.MatchString(repo) || strings.Contains(repo, "/"):
|
||||
return fmt.Errorf("%q is not a repository on the forge, owner/name; nothing was asked", positionals[0])
|
||||
case !tryRef.MatchString(ref):
|
||||
return fmt.Errorf("%q is not a branch or a commit; nothing was asked", ref)
|
||||
case *base != "" && !tryRef.MatchString(*base):
|
||||
return fmt.Errorf("%q is not a branch; nothing was asked", *base)
|
||||
case *tests != "" && !tryTests.MatchString(*tests):
|
||||
return fmt.Errorf("%q is not a `go test -run` pattern of test names; nothing was asked", *tests)
|
||||
}
|
||||
// **An unknown ref is refused here, loudly** — not sent to a machine to fail at its clone minutes later.
|
||||
commit, trunk, err := forgeCommit(ctx, owner, repo, ref)
|
||||
if err != nil {
|
||||
return fmt.Errorf("the forge holds no branch or commit %s in %s/%s — is it pushed? — so nothing was asked: %w",
|
||||
ref, owner, repo, err)
|
||||
}
|
||||
if *base != "" {
|
||||
trunk = *base
|
||||
}
|
||||
if trunk == "" {
|
||||
return fmt.Errorf("the forge named no default branch for %s/%s, and no --base says what %s is merged into; "+
|
||||
"nothing was asked", owner, repo, ref)
|
||||
}
|
||||
|
||||
open, err := openStores(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer open.Close()
|
||||
entries, read, edges, err := theGraph(ctx, open.inventory)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
p := link.PullUpdated{Owner: owner, Repo: repo, Base: trunk, Commit: commit}
|
||||
// Whose repository it is, as a pull request's check reckons it: only the mesh's is tried, the code the
|
||||
// build seat runs for every pull request already.
|
||||
mesh := pullScope(p, entries, read, edges)
|
||||
if !mesh.Mesh {
|
||||
return fmt.Errorf("%s/%s is no repository of the mesh — no module is built from it and its owner is not "+
|
||||
"the core's — and only the mesh's are tried on the build seat; nothing was asked", owner, repo)
|
||||
}
|
||||
request, err := checkRequestFor(ctx, open, p, checkScope{Mesh: true, From: mesh.From}, entries)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
request = asATry(request, *tests, *storeOnly)
|
||||
seat := buildSeatHeld(ctx)
|
||||
// **Bounded in number** (novox/hq issue 686): one try of a commit at a time, and maxTries in all.
|
||||
js, err := dialTheBus()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
queue, err := link.ReadQueue(ctx, js, seat)
|
||||
js.Close()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
same, room := triesIn(queue, owner, repo, commit)
|
||||
if same != "" {
|
||||
return printTry(tryAnswer{ID: same, State: "asked", Repository: owner + "/" + repo, Ref: ref, Commit: commit,
|
||||
On: seat, Said: fmt.Sprintf("a try of %s/%s at %.12s already waits or runs, as %s: nothing new was asked. "+
|
||||
"Read it with try id=%s", owner, repo, commit, same, same)})
|
||||
}
|
||||
if !room {
|
||||
return fmt.Errorf("%d tries already wait or run on the build seat, the most at once (`queue` lists them); "+
|
||||
"nothing was asked — read one of yours, or ask again when one has ended", maxTries)
|
||||
}
|
||||
ask, err := askOverOn(seat)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer ask.Close()
|
||||
if err := ask.Ask(ctx, request); err != nil {
|
||||
return err
|
||||
}
|
||||
return printTry(tryAnswer{ID: request.ID, State: "asked", Repository: owner + "/" + repo, Ref: ref, Commit: commit,
|
||||
Base: trunk, On: seat, Said: fmt.Sprintf("asked %s to try %s/%s at %.12s (%s) merged into %s: its own "+
|
||||
"merge-check.sh, as a check in the build queue, bounded at %s; no status is set and nothing is registered. "+
|
||||
"Read it with try id=%s — a full run of this kind takes minutes", seat, owner, repo, commit, ref, trunk,
|
||||
checkTimeout, request.ID)})
|
||||
}
|
||||
|
||||
// tryRead answers what came of a try, holding the answer up to tryWait for its verdict.
|
||||
func tryRead(ctx context.Context, id string) error {
|
||||
until := time.Now().Add(tryWait)
|
||||
for {
|
||||
lines, err := buildLines(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
a := tryAnswerOf(id, lines)
|
||||
if (a.State != "waiting" && a.State != "running") || !time.Now().Add(2*time.Second).Before(until) {
|
||||
return printTry(a)
|
||||
}
|
||||
if err := sleepOrDone(ctx, 2*time.Second); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// tryAnswerOf reads a try's answer from what the build seat said of it: its outcome under StepTried; `failed`
|
||||
// for one that could not run — an error, never a pass; else how far it is.
|
||||
func tryAnswerOf(id string, lines []link.BuildLine) tryAnswer {
|
||||
a := tryAnswer{ID: id, State: "waiting", Said: "nothing is said of " + id + " yet: it waits in the build queue " +
|
||||
"(`queue` says where), or no try has that id; ask again"}
|
||||
for _, l := range lines {
|
||||
switch l.Step {
|
||||
case link.StepTried:
|
||||
var t link.Tried
|
||||
if err := json.Unmarshal([]byte(l.Message), &t); err != nil || t.Verdict == "" {
|
||||
return tryAnswer{ID: id, State: "error", Said: "the build seat's word on it cannot be read: " + l.Message}
|
||||
}
|
||||
return tryAnswer{ID: id, State: t.Verdict, Said: t.Summary, Took: t.Took, Commit: t.Head, Trunk: t.Trunk,
|
||||
Failed: t.Failed}
|
||||
case "failed":
|
||||
return tryAnswer{ID: id, State: "error", Said: "it could not run: " + l.Message}
|
||||
case "checked":
|
||||
// A check that is no try, or a build seat from before the verb: its verdict is not a try's.
|
||||
return tryAnswer{ID: id, State: "error", Said: "the build seat checked it and said no try's outcome — " +
|
||||
"it predates the verb (novox/hq issue 686), or " + id + " is no try: " + l.Message + "; `builds` with " +
|
||||
"the id says every line"}
|
||||
}
|
||||
a.State, a.Last = "running", l.Message
|
||||
a.Said = "still running; ask again. `builds` with the id says every line so far"
|
||||
}
|
||||
return a
|
||||
}
|
||||
|
||||
func printTry(a tryAnswer) error {
|
||||
fmt.Fprintln(os.Stderr, a.Said)
|
||||
for _, l := range a.Failed {
|
||||
fmt.Fprintln(os.Stderr, " "+l)
|
||||
}
|
||||
out, err := json.MarshalIndent(a, "", " ")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
fmt.Println(string(out))
|
||||
return nil
|
||||
}
|
||||
|
||||
// buildLines is everything the build seat said of one build, as the bus keeps it, in order.
|
||||
func buildLines(ctx context.Context, id string) ([]link.BuildLine, error) {
|
||||
js, err := aBus()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("cannot reach the bus to read a build's log: %w", err)
|
||||
}
|
||||
defer js.Close()
|
||||
subject := link.BuildLogOf("*", id)
|
||||
sub, err := js.Context().PullSubscribe(subject, "",
|
||||
nats.BindStream(broker.EventsStream), nats.DeliverAll(), nats.AckNone())
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("cannot read %s from the bus: %w", subject, err)
|
||||
}
|
||||
defer func() { _ = sub.Unsubscribe() }()
|
||||
var lines []link.BuildLine
|
||||
for {
|
||||
batch, err := sub.Fetch(200, nats.MaxWait(2*time.Second))
|
||||
if err != nil && !errors.Is(err, nats.ErrTimeout) && !errors.Is(err, context.DeadlineExceeded) {
|
||||
return nil, fmt.Errorf("reading a build's log: %w", err)
|
||||
}
|
||||
for _, msg := range batch {
|
||||
var line link.BuildLine
|
||||
if json.Unmarshal(msg.Data, &line) == nil {
|
||||
lines = append(lines, line)
|
||||
}
|
||||
}
|
||||
if len(batch) < 200 {
|
||||
return lines, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,194 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// A branch is tried before its pull request (novox/hq issue 686): the repository's own check asked for a pushed
|
||||
// branch or commit, with no pull request.
|
||||
|
||||
// What a try asks is a check — after deliveries in the build queue, bounded as one — of the repository's own
|
||||
// script alone: no module of the graph, so no gate; nothing built for a merge, so no reusable image.
|
||||
func TestATryAsksACheckOfTheRepositorysOwnScriptAlone(t *testing.T) {
|
||||
request := link.BuildRequest{ID: "build-1", Ref: "c0ffee", Check: &link.CheckRequest{Owner: "novox", Repo: "a",
|
||||
Number: 7, Base: "main", Modules: []string{"a"}, New: []string{"b"}, Manifests: []string{"module.json"},
|
||||
Dependents: []string{"c"}, Builds: []link.CheckBuild{{Module: "a"}}, Judge: link.JudgeSelf,
|
||||
Plan: &link.ChangePlan{}}}
|
||||
tried := asATry(request, "^TestStore", true)
|
||||
c := tried.Check
|
||||
if !c.Try || c.Number != 0 || c.Tests != "^TestStore" || !c.StoreOnly {
|
||||
t.Fatalf("not marked a try, or it kept a pull request's number: %+v", c)
|
||||
}
|
||||
if len(c.Modules)+len(c.New)+len(c.Manifests)+len(c.Dependents)+len(c.Builds) != 0 || c.Judge != "" || c.Plan != nil {
|
||||
t.Fatalf("a try runs the gate or builds for a merge: %+v", c)
|
||||
}
|
||||
if c.Base != "main" || c.Owner != "novox" || tried.Ref != "c0ffee" {
|
||||
t.Fatalf("a try lost what it merges into what: %+v", tried)
|
||||
}
|
||||
if link.KindOf(tried) != link.KindCheck {
|
||||
t.Fatal("a try does not count as a check in the build queue")
|
||||
}
|
||||
}
|
||||
|
||||
// A try's outcome is heard and left there: no `checked` is said, so the forge's holder sets no status, and
|
||||
// nothing is recorded or registered (the handler holds no store here, and would fail reaching for one).
|
||||
func TestATrysOutcomeSetsNoStatusAndRegistersNothing(t *testing.T) {
|
||||
said := &saidOnTheBus{}
|
||||
checkEvents = said
|
||||
defer func() { checkEvents = nil }()
|
||||
for name, result := range map[string]link.BuildResult{
|
||||
"a try": {ID: "build-1", Ref: "c0ffee", Checked: &link.CheckRequest{Owner: "novox", Repo: "a", Try: true},
|
||||
Check: &link.CheckOutcome{Verdict: "pass", RepoCheck: &link.CheckLayer{Verdict: "fail", Summary: "failed"}}},
|
||||
"a try that could not run": {ID: "build-2", Checked: &link.CheckRequest{Owner: "novox", Repo: "a", Try: true},
|
||||
Failed: "cannot clone"},
|
||||
// A build seat from before the mark echoes the ask without it: a check of no pull request and no
|
||||
// group is still nobody's status.
|
||||
"a try echoed by an older build seat": {ID: "build-3", Checked: &link.CheckRequest{Owner: "novox", Repo: "a"},
|
||||
Check: &link.CheckOutcome{Verdict: "pass"}},
|
||||
} {
|
||||
if err := (builds{}).Built(context.Background(), result); err != nil {
|
||||
t.Fatalf("%s: %v", name, err)
|
||||
}
|
||||
}
|
||||
if len(said.events) != 0 {
|
||||
t.Fatalf("a try's outcome was said for the forge: %v", said.events)
|
||||
}
|
||||
// A pull request's check is still said.
|
||||
_ = (builds{}).Built(context.Background(), link.BuildResult{ID: "build-4", Checked: &link.CheckRequest{
|
||||
Owner: "novox", Repo: "a", Number: 7}, Check: &link.CheckOutcome{Verdict: "pass"}})
|
||||
if len(said.events) != 1 {
|
||||
t.Fatalf("a pull request's check was said %d time(s)", len(said.events))
|
||||
}
|
||||
}
|
||||
|
||||
// saidOnTheBus keeps the seat events said, and nothing else.
|
||||
type saidOnTheBus struct {
|
||||
link.Bus
|
||||
events []string
|
||||
}
|
||||
|
||||
func (s *saidOnTheBus) PublishSeatEvent(_ context.Context, seat, event string, _ []byte) error {
|
||||
s.events = append(s.events, seat+" "+event)
|
||||
return nil
|
||||
}
|
||||
|
||||
// A ref the forge does not hold is refused before anything is asked, naming it.
|
||||
func TestATryOfAnUnknownRefIsRefusedLoudly(t *testing.T) {
|
||||
was := forgeCommit
|
||||
defer func() { forgeCommit = was }()
|
||||
forgeCommit = func(_ context.Context, owner, repo, ref string) (string, string, error) {
|
||||
return "", "", errors.New("404 object does not exist [id: " + ref + "]")
|
||||
}
|
||||
err := tryCommand(context.Background(), []string{"novox/mesh-controller", "fix/no-such-branch"})
|
||||
if err == nil {
|
||||
t.Fatal("a try of a ref the forge does not hold was asked")
|
||||
}
|
||||
for _, want := range []string{"fix/no-such-branch", "novox/mesh-controller", "nothing was asked", "404"} {
|
||||
if !strings.Contains(err.Error(), want) {
|
||||
t.Errorf("the refusal does not say %q: %v", want, err)
|
||||
}
|
||||
}
|
||||
for _, args := range [][]string{{"mesh-controller", "main"}, {"novox/mesh-controller"}, {"novox/a", "--upload-pack=x"},
|
||||
{"novox/a", "main", "--tests", "x; rm -rf /"}, {"novox/a", "main", "--id", "build-1"}} {
|
||||
if err := tryCommand(context.Background(), args); err == nil {
|
||||
t.Errorf("try %v was taken", args)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The answer of a try carries the lines that name what failed (novox/hq issue 460's shape) and how long it took.
|
||||
func TestATrysAnswerCarriesTheFailingLines(t *testing.T) {
|
||||
tried, _ := json.Marshal(link.Tried{Verdict: "fail", Summary: "its merge-check.sh failed: TestX", Took: "5m3s",
|
||||
Head: "c0ffee", Trunk: "ca3f43c", Failed: []string{"--- FAIL: TestX (0.01s)", " x_test.go:12: got 1, want 2",
|
||||
"FAIL\tgithub.com/novox/a/cmd\t3.1s"}})
|
||||
lines := []link.BuildLine{{Seq: 1, Step: "check", Message: "novox/a#0 at c0ffee"},
|
||||
{Seq: 2, Step: "output", Message: "ok github.com/novox/a/internal 1.2s"},
|
||||
{Seq: 3, Step: link.StepTried, Message: string(tried)}, {Seq: 4, Step: "checked", Message: "pass: x"}}
|
||||
a := tryAnswerOf("build-1", lines)
|
||||
if a.State != "fail" || a.Took != "5m3s" || len(a.Failed) != 3 || a.Failed[0] != "--- FAIL: TestX (0.01s)" ||
|
||||
a.Trunk != "ca3f43c" || a.ID != "build-1" {
|
||||
t.Fatalf("the answer lost the verdict, the time or the failing lines: %+v", a)
|
||||
}
|
||||
// Not answered yet: said as running, with how far it is — never as a pass.
|
||||
if a := tryAnswerOf("build-1", lines[:2]); a.State != "running" || a.Last == "" {
|
||||
t.Fatalf("a try still running: %+v", a)
|
||||
}
|
||||
if a := tryAnswerOf("build-1", nil); a.State != "waiting" {
|
||||
t.Fatalf("a try nothing was said of: %+v", a)
|
||||
}
|
||||
// One that could not run is an error with the build seat's word, never a pass.
|
||||
if a := tryAnswerOf("build-1", append(lines[:1:1], link.BuildLine{Seq: 2, Step: "failed",
|
||||
Message: "cannot clone"})); a.State != "error" || !strings.Contains(a.Said, "cannot clone") {
|
||||
t.Fatalf("a try that could not run: %+v", a)
|
||||
}
|
||||
}
|
||||
|
||||
// One try of a commit at a time, and four in all (novox/hq issue 686): a second ask is answered the first's id,
|
||||
// a fifth is refused; a try that ended, and a pull request's check, count for nothing.
|
||||
func TestTriesAreBoundedInNumber(t *testing.T) {
|
||||
ask := func(id, repo, ref, state string, try bool) link.QueuedAsk {
|
||||
return link.QueuedAsk{ID: id, State: state, Request: link.BuildRequest{ID: id, Ref: ref,
|
||||
Check: &link.CheckRequest{Owner: "novox", Repo: repo, Try: try}}}
|
||||
}
|
||||
q := link.Queue{Asks: []link.QueuedAsk{
|
||||
ask("t1", "a", "c1", link.AskInFlight, true), ask("t2", "a", "c2", link.AskWaiting, true),
|
||||
ask("t3", "b", "c1", link.AskDead, true), ask("p1", "a", "c9", link.AskWaiting, false)}}
|
||||
if same, room := triesIn(q, "novox", "a", "c2"); same != "t2" || !room {
|
||||
t.Fatalf("a second try of a commit: %q %v", same, room)
|
||||
}
|
||||
if same, room := triesIn(q, "novox", "b", "c1"); same != "" || !room {
|
||||
t.Fatalf("a dead try, or another repository's, was taken for this one: %q %v", same, room)
|
||||
}
|
||||
if same, _ := triesIn(q, "novox", "a", "c9"); same != "" {
|
||||
t.Fatalf("a pull request's check was taken for a try: %q", same)
|
||||
}
|
||||
q.Asks = append(q.Asks, ask("t4", "c", "c3", link.AskWaiting, true), ask("t5", "d", "c4", link.AskWaiting, true))
|
||||
if _, room := triesIn(q, "novox", "e", "c5"); room {
|
||||
t.Fatal("a fifth try was let in")
|
||||
}
|
||||
// An id is checked before it is a token of a bus subject.
|
||||
for _, id := range []string{"x.>", "*", "a b"} {
|
||||
if err := tryCommand(context.Background(), []string{"--id", id}); err == nil ||
|
||||
!strings.Contains(err.Error(), "not a try's id") {
|
||||
t.Errorf("try --id %q: %v", id, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// The verb runs its command, and is declared with the shell habit it replaces (novox/hq ADR 0245).
|
||||
func TestTheTryVerbRunsItsCommand(t *testing.T) {
|
||||
for _, c := range []struct {
|
||||
args map[string]any
|
||||
want []string
|
||||
}{
|
||||
{map[string]any{"repository": "novox/a", "ref": "fix/x"}, []string{"try", "novox/a", "fix/x"}},
|
||||
{map[string]any{"repository": "novox/a", "ref": "fix/x", "base": "main", "tests": "^TestX$", "store-only": "true"},
|
||||
[]string{"try", "novox/a", "fix/x", "--base", "main", "--tests", "^TestX$", "--store-only"}},
|
||||
{map[string]any{"id": "build-1"}, []string{"try", "--id", "build-1"}},
|
||||
} {
|
||||
got, err := argvFor("try", c.args)
|
||||
if err != nil || !reflect.DeepEqual(got, c.want) {
|
||||
t.Errorf("try %v: %v %v, want %v", c.args, got, err, c.want)
|
||||
}
|
||||
}
|
||||
if _, err := argvFor("try", map[string]any{"repository": "novox/a"}); err == nil {
|
||||
t.Error("a try of no ref was taken")
|
||||
}
|
||||
for _, v := range catalogue.ControllerVerbs {
|
||||
if v.Name == "try" {
|
||||
if len(v.Replaces) == 0 {
|
||||
t.Error("try names no shell habit it replaces")
|
||||
}
|
||||
return
|
||||
}
|
||||
}
|
||||
t.Error("try is not a verb of the controller seat")
|
||||
}
|
||||
@@ -79,7 +79,7 @@ func (f following) Upgraded(ctx context.Context, u link.Upgraded) error {
|
||||
// `rebuild`, a `replay` registered — was sent here one machine after another without any judging; it
|
||||
// now waits for a gate like any other build, and the release plan walks it through the machines, one
|
||||
// at a time, each judged.
|
||||
fmt.Printf("%s moved to %s outside a plan; a release plan sends it to %s, one machine at a time, each judged "+
|
||||
fmt.Printf("%s moved to %s outside a plan; a release plan sends it to %s, one machine first and then the rest at once, each judged "+
|
||||
"(`upgrade backlog` lists what waits)\n", u.Module, shortCommit(u.Commit), readableList(on))
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -177,6 +177,12 @@ func walkReached(p inventory.Plan, running func(string) []string) map[string][]s
|
||||
reached = appendNew(reached, r.Gate.Machines...)
|
||||
carried(r.Gate)
|
||||
}
|
||||
for _, g := range r.Gates {
|
||||
if g != nil && !g.Pending {
|
||||
reached = appendNew(reached, g.Machines...)
|
||||
carried(g)
|
||||
}
|
||||
}
|
||||
for _, tier := range p.Tiers {
|
||||
for _, m := range tier {
|
||||
for _, n := range running(m) {
|
||||
|
||||
@@ -40,39 +40,6 @@ var (
|
||||
AskedBucket = BucketName(ControllerSeat, "asked")
|
||||
)
|
||||
|
||||
// The controller's record mirrors (novox/hq ADR 0289, design 54, design 57 package F3): a bucket per kind of
|
||||
// record, one key per record, written from the store by the lease holder only when the record changed, with
|
||||
// one event on the controller's seat per write. Read by whatever names them in `reads` as
|
||||
// `mesh-controller.<mirror>` (ADR 0201 §1), written by nobody else. **No secret in either** (ADR 0201): an
|
||||
// assignment's value names its module and node; a module's declared config is what its manifest, which is
|
||||
// public, declares.
|
||||
//
|
||||
// AssignmentsBucket holds which module is assigned to which node, keyed `<node>.<module>`; and
|
||||
// DeclaredConfigBucket each registered module's config declarations as its manifest holds them, keyed by
|
||||
// the module: what mesh-config resolves a config value from (ADR 0348 clauses 2 and 4).
|
||||
var (
|
||||
AssignmentsBucket = BucketName(ControllerSeat, MirrorAssignments)
|
||||
DeclaredConfigBucket = BucketName(ControllerSeat, MirrorDeclaredConfig)
|
||||
)
|
||||
|
||||
// The mirrors' local names, as a reader names them in `reads` after `mesh-controller.`.
|
||||
const (
|
||||
MirrorAssignments = "assignments"
|
||||
MirrorDeclaredConfig = "declared-config"
|
||||
)
|
||||
|
||||
// ControllerMirrors are the local names of the controller's record mirrors, which a module may read.
|
||||
func ControllerMirrors() []string { return []string{MirrorAssignments, MirrorDeclaredConfig} }
|
||||
|
||||
// The mirrors' caps (design 54 §5): one value per key, no age, 64 KiB a value, 64 MiB a bucket.
|
||||
const (
|
||||
MirrorValueBytes = 64 << 10
|
||||
MirrorBytes = 64 << 20
|
||||
// MirrorMarkerTTL is the longest a purge's marker is kept (design 54 §2): the bucket allows a
|
||||
// per-message age for it, and the writer gives each marker a day.
|
||||
MirrorMarkerTTL = 24 * time.Hour
|
||||
)
|
||||
|
||||
// AskedKeptFor is how long an ask is kept after it was made: a month, as the router keeps its own.
|
||||
const AskedKeptFor = 30 * 24 * time.Hour
|
||||
|
||||
@@ -99,14 +66,12 @@ const (
|
||||
// IsControllerBucket says a bucket is the controller's own, not a module's state nothing declares.
|
||||
func IsControllerBucket(bucket string) bool {
|
||||
return bucket == CallsBucket || bucket == HandActsBucket || bucket == ConditionsBucket ||
|
||||
bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket ||
|
||||
bucket == AssignmentsBucket || bucket == DeclaredConfigBucket
|
||||
bucket == ConditionHistoryBucket || bucket == LeaseBucket || bucket == AskedBucket
|
||||
}
|
||||
|
||||
// ControllerBuckets are the controller's own buckets, in the order they are asserted.
|
||||
func ControllerBuckets() []string {
|
||||
return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket,
|
||||
AssignmentsBucket, DeclaredConfigBucket}
|
||||
return []string{LeaseBucket, CallsBucket, HandActsBucket, ConditionsBucket, ConditionHistoryBucket, AskedBucket}
|
||||
}
|
||||
|
||||
// ControllerBucketsAsserter is what raising the controller's buckets needs of a connection.
|
||||
@@ -203,27 +168,6 @@ func (j *JetStream) EnsureControllerBuckets() error {
|
||||
}); err != nil {
|
||||
return fmt.Errorf("asserting bucket %s: %w", AskedBucket, err)
|
||||
}
|
||||
// **The record mirrors: no age** (design 54 §5). A key is removed when its record is, by a purge whose
|
||||
// marker lives a day; a key that expired would be a record the bus forgot while the store still holds it.
|
||||
for _, m := range []struct{ bucket, why string }{
|
||||
{AssignmentsBucket, "which module is assigned to which node, keyed <node>.<module> (novox/hq ADR 0289, " +
|
||||
"design 54): written by the controller alone from its store when an assignment changes, read through `reads`"},
|
||||
{DeclaredConfigBucket, "each registered module's config declarations as its manifest holds them, keyed by " +
|
||||
"the module (novox/hq ADR 0348, design 57 F3): written by the controller alone from its store when a " +
|
||||
"registration changes them, read through `reads`"},
|
||||
} {
|
||||
if _, err := js.CreateOrUpdateKeyValue(ctx, jetstream.KeyValueConfig{
|
||||
Bucket: m.bucket,
|
||||
Description: m.why,
|
||||
History: 1,
|
||||
MaxValueSize: MirrorValueBytes,
|
||||
MaxBytes: MirrorBytes,
|
||||
Storage: jetstream.FileStorage,
|
||||
LimitMarkerTTL: MirrorMarkerTTL,
|
||||
}); err != nil {
|
||||
return fmt.Errorf("asserting bucket %s: %w", m.bucket, err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
|
||||
@@ -21,8 +21,7 @@ func TestTheControllerMayWriteEveryBucketItWrites(t *testing.T) {
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
want := []string{"$KV." + CallsBucket + ".>", "$KV." + HandActsBucket + ".>",
|
||||
"$KV." + AssignmentsBucket + ".>", "$KV." + DeclaredConfigBucket + ".>"}
|
||||
want := []string{"$KV." + CallsBucket + ".>", "$KV." + HandActsBucket + ".>"}
|
||||
for _, seat := range seatsTheControllerAsks {
|
||||
if hasCancelledSet(seat) {
|
||||
want = append(want, "$KV."+CancelledSetName(seat)+".>")
|
||||
@@ -38,34 +37,6 @@ func TestTheControllerMayWriteEveryBucketItWrites(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// **A module reads a record mirror through `reads`, and never writes it** (ADR 0201 §1, design 54 §5,
|
||||
// design 57 F3): granted the direct reads and watches of the bucket, and refused at composition a grant to
|
||||
// write it, whose writer is the controller.
|
||||
func TestAModuleReadsTheControllersMirrorsAndWritesNone(t *testing.T) {
|
||||
reader := Principal{Kind: KindModule, Node: "one", Module: "mesh-config", PasswordHash: "x",
|
||||
Reads: []string{"mesh-controller." + MirrorAssignments, "mesh-controller." + MirrorDeclaredConfig}}
|
||||
p, err := PermissionsFor(reader)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, bucket := range []string{AssignmentsBucket, DeclaredConfigBucket} {
|
||||
for _, subject := range []string{"$JS.API.STREAM.INFO.KV_" + bucket, "$JS.API.DIRECT.GET.KV_" + bucket + ".>",
|
||||
"$JS.API.CONSUMER.CREATE.KV_" + bucket + ".>"} {
|
||||
if !slices.Contains(p.Publish, subject) {
|
||||
t.Errorf("a module reading %s may not publish %s", bucket, subject)
|
||||
}
|
||||
}
|
||||
for _, subject := range p.Publish {
|
||||
if SubjectsOverlap(subject, "$KV."+bucket+".>") {
|
||||
t.Errorf("a module reading %s may write it through %s", bucket, subject)
|
||||
}
|
||||
}
|
||||
if err := CheckWriters(reader, []string{"$KV." + bucket + ".>"}); err == nil {
|
||||
t.Errorf("a module granted a write of %s would compose", bucket)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// **A machine's node tools may say they are there, as that machine and no other** (novox/hq to-be 45
|
||||
// S11), and the controller may hear the bus's advisories and ask who answers — read-only, named.
|
||||
func TestTheWatchedSignalsMayBeSaidAndHeard(t *testing.T) {
|
||||
@@ -125,56 +96,3 @@ func TestWhatTheControllerAskedIsBounded(t *testing.T) {
|
||||
t.Errorf("history %d, age %s, bytes %d", status.History(), status.TTL(), info.Config.MaxBytes)
|
||||
}
|
||||
}
|
||||
|
||||
// **A listed mirror is no undeclared bucket** (ADR 0356 §3, ADR 0201 §16): the controller's list of its mirrors
|
||||
// stands for their declaration, so the report of state nothing declares never names one.
|
||||
func TestAListedMirrorIsNotReportedAsUndeclared(t *testing.T) {
|
||||
names := []string{"gone_state"}
|
||||
for _, m := range ControllerMirrors() {
|
||||
names = append(names, BucketName(ControllerSeat, m))
|
||||
}
|
||||
undeclared, err := RaiseBuckets(fakeBuckets{names: names}, nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !slices.Equal(undeclared, []string{"gone_state"}) {
|
||||
t.Fatalf("the report of undeclared state names %v", undeclared)
|
||||
}
|
||||
}
|
||||
|
||||
// **The mirrors' list and the writers table agree** (ADR 0356 §3, to-be 45 §1): every listed mirror is a row of
|
||||
// the table whose one writer is the controller, on its bucket alone; and every row that writes a bucket of the
|
||||
// controller's named as a mirror is on the list.
|
||||
func TestTheMirrorsListAndTheWritersTableAgree(t *testing.T) {
|
||||
controller := Principal{Kind: KindController}
|
||||
module := Principal{Kind: KindModule, Module: ControllerSeat, Node: "one"}
|
||||
rowed := map[string]bool{}
|
||||
for _, row := range WritersTable {
|
||||
if !strings.HasSuffix(row.State, " mirror") {
|
||||
continue
|
||||
}
|
||||
for _, subject := range row.Subjects {
|
||||
bucket := strings.TrimSuffix(strings.TrimPrefix(subject, "$KV."), ".>")
|
||||
rowed[bucket] = true
|
||||
if row.Writes == nil || !row.Writes(controller, subject) || row.Writes(module, subject) {
|
||||
t.Errorf("%q is not written by the controller alone", row.State)
|
||||
}
|
||||
}
|
||||
}
|
||||
listed := map[string]bool{}
|
||||
for _, m := range ControllerMirrors() {
|
||||
bucket := BucketName(ControllerSeat, m)
|
||||
listed[bucket] = true
|
||||
if !rowed[bucket] {
|
||||
t.Errorf("the mirror %s is listed and no row of the writers table writes it", bucket)
|
||||
}
|
||||
if !IsControllerBucket(bucket) || !slices.Contains(ControllerBuckets(), bucket) {
|
||||
t.Errorf("the mirror %s is listed and not asserted as the controller's own", bucket)
|
||||
}
|
||||
}
|
||||
for bucket := range rowed {
|
||||
if !listed[bucket] {
|
||||
t.Errorf("the writers table writes the mirror %s, which the list does not name", bucket)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -172,9 +172,7 @@ func TestRaisingLogsReportsWhatNothingDeclares(t *testing.T) {
|
||||
// **No path of the mesh deletes or purges a log or a bucket, or recreates one** (novox/hq ADR 0297 §2,
|
||||
// ADR 0201 §15–16): no code of this repository outside its tests calls the client's stream or bucket
|
||||
// delete, or purges a stream, but for the controller lease's own stream, which purges its own history
|
||||
// below a sequence (internal/lease), and the controller's record mirrors, which purge the key of a record
|
||||
// the store no longer holds (internal/recordmirror; novox/hq ADR 0289, design 54 §2). A new call is a decision,
|
||||
// not a refactor.
|
||||
// below a sequence (internal/lease). A new call is a decision, not a refactor.
|
||||
func TestNoPathDeletesALogOrABucket(t *testing.T) {
|
||||
removers := map[string]bool{"DeleteStream": true, "DeleteKeyValue": true, "PurgeStream": true,
|
||||
"DeleteObjectStore": true}
|
||||
@@ -199,9 +197,7 @@ func TestNoPathDeletesALogOrABucket(t *testing.T) {
|
||||
return true
|
||||
}
|
||||
name := sel.Sel.Name
|
||||
decided := strings.Contains(filepath.ToSlash(path), "internal/lease/") ||
|
||||
strings.Contains(filepath.ToSlash(path), "internal/recordmirror/")
|
||||
if removers[name] || (name == "Purge" && !decided) {
|
||||
if removers[name] || (name == "Purge" && !strings.Contains(filepath.ToSlash(path), "internal/lease/")) {
|
||||
found = append(found, path+": "+name)
|
||||
}
|
||||
return true
|
||||
|
||||
@@ -34,8 +34,6 @@ func TestTheFactsTheGrantPermitsAreTheFactsTheMeshStates(t *testing.T) {
|
||||
states = append(states, link.KeyChecked)
|
||||
// And a walk kept in a new state, for the delivery it walks (novox/hq ADR 0239).
|
||||
states = append(states, link.KeyPlanMoved)
|
||||
// And a key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3; design 57 F3).
|
||||
states = append(states, link.KeyAssignmentChanged, link.KeyDeclaredConfigChanged)
|
||||
for _, event := range states {
|
||||
if !slices.Contains(broker.ControllerStates, event) {
|
||||
t.Errorf("the mesh states %q and its account may not publish it", event)
|
||||
@@ -44,11 +42,6 @@ func TestTheFactsTheGrantPermitsAreTheFactsTheMeshStates(t *testing.T) {
|
||||
if len(broker.ControllerStates) != len(states) {
|
||||
t.Errorf("the grant permits %v, which is more than the mesh states", broker.ControllerStates)
|
||||
}
|
||||
// The record mirrors a module may read are the ones the bus holds (design 54 §5, design 57 F3).
|
||||
if !slices.Equal(catalogue.ControllerMirrors, broker.ControllerMirrors()) {
|
||||
t.Errorf("the catalogue lets a module read the controller's mirrors %v and the bus holds %v",
|
||||
catalogue.ControllerMirrors, broker.ControllerMirrors())
|
||||
}
|
||||
// **And the seat says it.** A seat carries the protocol of its role (novox/hq ADR 0129), so the
|
||||
// facts the control plane states are the seat's `emits` — which is what lets anything else declare
|
||||
// that it consumes them, and what the subject-agreement check reads to know they have an owner.
|
||||
|
||||
@@ -313,10 +313,7 @@ var ControllerStates = []string{"applied", "refused", "built-before",
|
||||
// the pull request's status, and anybody else may read.
|
||||
"checked",
|
||||
// And a walk kept in a new state (novox/hq ADR 0239): what the delivery it walks reads its steps from.
|
||||
"plan-moved",
|
||||
// And a key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3): the assignments and
|
||||
// each registered module's declared config (design 57 F3).
|
||||
"assignment-changed", "declared-config-changed"}
|
||||
"plan-moved"}
|
||||
|
||||
// BusAdvisories are what the bus server says about the mesh's own account that the controller
|
||||
// reads (novox/hq to-be 45 §3, S9): a durable consumer that handed a message over as often as it
|
||||
|
||||
+1
-1
@@ -24,7 +24,7 @@ accounts {
|
||||
jetstream: enabled
|
||||
users = [
|
||||
{ user: "controller", password: "$2a$11$cccccccccccccccccccccc", permissions: {
|
||||
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_assignments.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_declared-config.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.assignment-changed", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.declared-config-changed", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*"] }
|
||||
publish: { allow: ["$JS.ACK.CONTROL.controller.>", "$JS.ACK.DEAD_LETTER_NOTICES.controller.>", "$JS.ACK.EVENTS.controller.>", "$JS.API.>", "$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>", "$KV.SEAT_NODE_BUILD_AGENT_cancelled.>", "$KV.mesh-controller_asked.>", "$KV.mesh-controller_calls.>", "$KV.mesh-controller_condition-history.>", "$KV.mesh-controller_conditions.>", "$KV.mesh-controller_hand-acts.>", "$KV.mesh-controller_lease.>", "$SRV.INFO", "_INBOX.enrol.>", "mesh.again.>", "mesh.assignment.>", "mesh.events.dead.>", "mesh.mod.*.tool.>", "mesh.node.>", "mesh.seat.mesh-build-machine.accept.>", "mesh.seat.mesh-build-machine.tool.>", "mesh.seat.mesh-controller.event.applied", "mesh.seat.mesh-controller.event.built-before", "mesh.seat.mesh-controller.event.checked", "mesh.seat.mesh-controller.event.condition-changed", "mesh.seat.mesh-controller.event.condition-cleared", "mesh.seat.mesh-controller.event.condition-raised", "mesh.seat.mesh-controller.event.doctor-heartbeat", "mesh.seat.mesh-controller.event.healer-acted", "mesh.seat.mesh-controller.event.plan-moved", "mesh.seat.mesh-controller.event.refused", "mesh.seat.mesh-controller.event.rolled-back", "mesh.seat.mesh-controller.event.secret-replaced", "mesh.seat.mesh-controller.tool.plans", "mesh.seat.mesh-delivery.tool.close", "mesh.seat.mesh-delivery.tool.groups", "mesh.seat.mesh-delivery.tool.release", "mesh.seat.mesh-delivery.tool.stalled", "mesh.seat.mesh-delivery.tool.stop", "mesh.seat.mesh-delivery.tool.times", "mesh.seat.node-backup.tool.backed-up.*", "mesh.seat.node-backup.tool.now.*", "mesh.seat.node-build-agent.accept.>", "mesh.seat.node-build-agent.tool.>", "mesh.seat.node-intrusion-prevention.tool.banned.*", "mesh.seat.node-launcher.tool.secret.*", "mesh.seat.node-service-manager.tool.restart.*"] }
|
||||
subscribe: { allow: ["$JS.API.>", "$JS.EVENT.ADVISORY.CONSUMER.DELETED.>", "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>", "$SRV.INFO", "$SRV.INFO.mesh-controller", "$SRV.INFO.mesh-controller.>", "$SRV.PING", "$SRV.PING.mesh-controller", "$SRV.PING.mesh-controller.>", "$SRV.STATS", "$SRV.STATS.mesh-controller", "$SRV.STATS.mesh-controller.>", "_DELIVER.controller", "_DELIVER.controller.>", "_INBOX.controller.>", "mesh.control.>", "mesh.mod.*.event.provisioner.failing", "mesh.mod.*.event.provisioner.recovered", "mesh.mod.*.event.provisioner.retirement", "mesh.mod.gitea.event.pull.merged", "mesh.mod.gitea.event.pull.updated", "mesh.mod.mesh-catalog.event.catching-up", "mesh.mod.mesh-catalog.event.upgraded", "mesh.seat.mesh-build-machine.event.built", "mesh.seat.mesh-controller.tool.>", "mesh.seat.node-build-agent.event.built", "mesh.seat.operator-channel.event.decided.mesh-controller"] }
|
||||
allow_responses: { max: 1, ttl: "1m" }
|
||||
} }
|
||||
|
||||
@@ -158,15 +158,6 @@ var WritersTable = []WriterRow{
|
||||
{State: "a message a consumer gave up on", Writer: "controller", KeptIn: "the bus, the stream " + DeadLettersStream,
|
||||
Others: "read, delivered again or dropped through the controller's dead-letters verb",
|
||||
Subjects: []string{deadLetterPrefix + ">", againPrefix + ">"}, Writes: isController},
|
||||
// The record mirrors (novox/hq ADR 0289 §9, design 54 §2): rows of this table, so a grant that lets
|
||||
// anything else write them is refused at composition. Built so far: the assignments and each registered
|
||||
// module's declared config (design 57 F3).
|
||||
{State: "the assignments mirror", Writer: "controller (lease holder), from its store, only on a change",
|
||||
KeptIn: "key-value " + AssignmentsBucket, Others: "read through `reads` as mesh-controller." + MirrorAssignments,
|
||||
Subjects: kvOf(AssignmentsBucket), Writes: isController},
|
||||
{State: "the declared-config mirror", Writer: "controller (lease holder), from its store, only on a change",
|
||||
KeptIn: "key-value " + DeclaredConfigBucket, Others: "read through `reads` as mesh-controller." + MirrorDeclaredConfig,
|
||||
Subjects: kvOf(DeclaredConfigBucket), Writes: isController},
|
||||
}
|
||||
|
||||
// CheckWriters refuses a grant that lets a principal publish on a subject the writers table gives
|
||||
|
||||
@@ -34,9 +34,6 @@ var designRows = []string{
|
||||
"the operator-channel's open messages",
|
||||
"the facts snapshot",
|
||||
"a message a consumer gave up on",
|
||||
// The record mirrors, rows of to-be 45 §1's table by design 54 §2 (ADR 0289 §9).
|
||||
"the assignments mirror",
|
||||
"the declared-config mirror",
|
||||
}
|
||||
|
||||
func TestTheWritersTableIsTheDesigns(t *testing.T) {
|
||||
|
||||
@@ -103,6 +103,23 @@ type CheckSpec struct {
|
||||
// for the walk to take. None for a group's composed check, whose members' own checks build.
|
||||
Builds []CheckBuildSpec
|
||||
BuildWith *CheckBuilding
|
||||
// Tests and StoreOnly narrow what the repository's own script runs, for a branch tried before its pull
|
||||
// request (novox/hq issue 686): handed to the script as EnvTests and EnvStoreOnly, which it may read.
|
||||
Tests string
|
||||
StoreOnly bool
|
||||
}
|
||||
|
||||
// selection is what a check hands the repository's script to narrow what it runs (novox/hq issue 686): nothing
|
||||
// for a pull request's check, which always runs whole.
|
||||
func selection(spec CheckSpec) []string {
|
||||
var env []string
|
||||
if spec.Tests != "" {
|
||||
env = append(env, EnvTests+"="+spec.Tests)
|
||||
}
|
||||
if spec.StoreOnly {
|
||||
env = append(env, EnvStoreOnly+"=1")
|
||||
}
|
||||
return env
|
||||
}
|
||||
|
||||
// GroupHead is one other head of a delivery group's composed check.
|
||||
@@ -193,6 +210,11 @@ const (
|
||||
// EnvGroup is a delivery group's other heads, as JSON, for the gate (novox/hq ADR 0239); empty for a
|
||||
// pull request checked alone.
|
||||
EnvGroup = "MESH_CHECK_GROUP"
|
||||
// EnvTests is a `go test -run` pattern and EnvStoreOnly is 1 when a branch tried before its pull request
|
||||
// asks for only those tests, or only the store tests (novox/hq issue 686). Never set for a pull request's
|
||||
// check. A script that reads neither runs whole.
|
||||
EnvTests = "MESH_CHECK_TESTS"
|
||||
EnvStoreOnly = "MESH_CHECK_STORE_ONLY"
|
||||
)
|
||||
|
||||
// CheckTimeout bounds one check; a check that runs past it is an error, not a pass.
|
||||
@@ -515,6 +537,7 @@ func Check(ctx context.Context, run Runner, spec CheckSpec, workspace, registry
|
||||
EnvModules + "=" + strings.Join(append(append([]string{}, spec.Modules...), spec.New...), ","),
|
||||
EnvGroup + "=" + groupFile,
|
||||
"GOCACHE=" + filepath.Join(workspace, "go-cache"), "GOMODCACHE=" + filepath.Join(workspace, "go-modules")}
|
||||
env = append(env, selection(spec)...)
|
||||
|
||||
v := CheckVerdict{}
|
||||
modules := append(append([]string{}, spec.Modules...), prefixed("new:", spec.New)...)
|
||||
|
||||
@@ -590,3 +590,15 @@ func TestARepositoryInTwoLanguagesIsCheckedInBoth(t *testing.T) {
|
||||
t.Errorf("a part in a toolchain the mesh does not hold answered %+v — never a pass", layer)
|
||||
}
|
||||
}
|
||||
|
||||
// A branch tried before its pull request may narrow what the repository's script runs (novox/hq issue 686); a
|
||||
// pull request's check is handed nothing, and runs whole.
|
||||
func TestOnlyATrysSelectionReachesTheScript(t *testing.T) {
|
||||
if env := selection(CheckSpec{Number: 7}); len(env) != 0 {
|
||||
t.Fatalf("a pull request's check was handed a selection: %v", env)
|
||||
}
|
||||
env := selection(CheckSpec{Tests: "^TestX$", StoreOnly: true})
|
||||
if len(env) != 2 || env[0] != "MESH_CHECK_TESTS=^TestX$" || env[1] != "MESH_CHECK_STORE_ONLY=1" {
|
||||
t.Fatalf("a try's selection: %v", env)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -107,9 +107,7 @@ var defaultSeats = append([]Seat{
|
||||
// A pull request's merge check, judged (novox/hq to-be 45 §9).
|
||||
"checked",
|
||||
// A walk kept in a new state, for the delivery it walks (novox/hq ADR 0239).
|
||||
"plan-moved",
|
||||
// A key of a record mirror written or removed (novox/hq ADR 0289, design 54 §3; design 57 F3).
|
||||
"assignment-changed", "declared-config-changed"},
|
||||
"plan-moved"},
|
||||
Serves: ControllerVerbs},
|
||||
// The store's first verbs (novox/hq ADR 0159): the smallest set that makes the store askable,
|
||||
// served by whichever module holds the seat with tools of these names.
|
||||
|
||||
@@ -124,14 +124,6 @@ func StateProblems(m Manifest) []string {
|
||||
return problems
|
||||
}
|
||||
|
||||
// ControllerMirrors are the controller's record mirrors a module may read, each `mesh-controller.<name>`
|
||||
// (novox/hq ADR 0289, design 54 §5, design 57 F3). **The controller keeps them, not its manifest's `state`**:
|
||||
// they are asserted with the controller's own buckets, with options of their own, and written by the
|
||||
// controller's account alone (the writers table); declared as a module's state, they would be asserted a
|
||||
// second way and granted to the module's own principal as well. Named here so a read of one is a read of
|
||||
// kept state. The bus's list is broker.ControllerMirrors; a test keeps the two agreeing.
|
||||
var ControllerMirrors = []string{"assignments", "declared-config"}
|
||||
|
||||
// StateReadsNothingDeclares is every read across a catalogue whose owner is present and declares no
|
||||
// such state. An absent owner says nothing — a module may be installed long before the one whose
|
||||
// state it reads, as a consumer may before its emitter (design 32 §1).
|
||||
@@ -144,11 +136,6 @@ func StateReadsNothingDeclares(manifests []Manifest) []string {
|
||||
}
|
||||
declared[m.Module] = own
|
||||
}
|
||||
if own, present := declared[ControllerSeatName]; present {
|
||||
for _, name := range ControllerMirrors {
|
||||
own[name] = true
|
||||
}
|
||||
}
|
||||
var problems []string
|
||||
for _, m := range manifests {
|
||||
for _, r := range m.Reads {
|
||||
|
||||
@@ -67,21 +67,6 @@ func TestAReadNamesStateItsOwnerKeeps(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// **The controller's record mirrors are kept state** (design 54 §5, design 57 F3): a module may read
|
||||
// `mesh-controller.assignments` and `mesh-controller.declared-config` although the controller's manifest
|
||||
// declares no `state`; a read of anything else of the controller's is still named.
|
||||
func TestAReadOfTheControllersMirrorsIsAReadOfKeptState(t *testing.T) {
|
||||
controller := Manifest{Module: ControllerSeatName}
|
||||
reader := Manifest{Module: "mesh-config", Reads: []string{"mesh-controller.assignments", "mesh-controller.declared-config"}}
|
||||
if p := StateReadsNothingDeclares([]Manifest{controller, reader}); len(p) != 0 {
|
||||
t.Fatalf("a read of the controller's mirrors was refused: %v", p)
|
||||
}
|
||||
other := Manifest{Module: "other", Reads: []string{"mesh-controller.calls"}}
|
||||
if p := StateReadsNothingDeclares([]Manifest{controller, other}); len(p) != 1 {
|
||||
t.Fatalf("a read of the controller's own calls was not named: %v", p)
|
||||
}
|
||||
}
|
||||
|
||||
// **Across the whole catalogue**: every state name is local, and every read whose owner is present
|
||||
// names state that owner keeps.
|
||||
func TestEveryManifestsStateIsLocalAndEveryReadIsKept(t *testing.T) {
|
||||
|
||||
@@ -362,6 +362,28 @@ var ControllerVerbs = []Verb{
|
||||
"register": "\"true\" to register what it builds",
|
||||
"older": "\"true\", with register: even though a newer build of the module is registered",
|
||||
}, []string{"id"}, "register", "older")},
|
||||
// novox/hq issue 686: a builder ran a branch's store tests for the first time in its pull request's merge check.
|
||||
{Name: "try", Description: "Try a pushed branch or commit before its pull request (novox/hq issue 686): the build " +
|
||||
"seat merges it into the repository's trunk and runs the repository's own merge-check.sh on the merge, in the " +
|
||||
"toolchain containers and against the throwaway store and bus a pull request's merge check gets. Answers at " +
|
||||
"once with an id; asked again with that id it waits up to 20 s and answers pass, fail or error, how long it " +
|
||||
"took and the lines naming what failed — or that it is still waiting or running, to ask again. Sets no " +
|
||||
"status on the forge, registers and builds nothing, runs no merge gate, counts as a check in the build queue " +
|
||||
"(after deliveries and after every pull request's check, however long it waits) and is bounded as one. One " +
|
||||
"try of a commit at a time — a second ask answers the first's id — and at most four waiting or running in " +
|
||||
"all: a fifth is refused. Only for a repository of the mesh; the ref must be on the forge.",
|
||||
Input: schema(map[string]string{
|
||||
"repository": "the repository on the forge, owner/name; with ref",
|
||||
"ref": "the branch or commit to try, as pushed to the forge",
|
||||
"base": "the branch it is merged into (the repository's default branch when none is named). The " +
|
||||
"merge-check.sh that runs is that branch's, never the ref's own: the same trust as a pull request's base",
|
||||
"tests": "only the tests this `go test -run` pattern names, where the repository's merge-check.sh reads " +
|
||||
"MESH_CHECK_TESTS; a script that does not runs whole",
|
||||
"store-only": "\"true\": only the repository's store tests, where its merge-check.sh reads " +
|
||||
"MESH_CHECK_STORE_ONLY",
|
||||
"id": "instead of a repository: the id a try answered, to read what came of it",
|
||||
}, nil, "store-only"),
|
||||
Replaces: []string{"sh merge-check.sh", "docker run postgres"}},
|
||||
// novox/hq ADR 0350 §5: asked by the delivery's owner when the forge says a pull request closed unmerged.
|
||||
{Name: "forget-checks", Description: "Let go of every build a pull request's merge checks made for its merge in " +
|
||||
"the artifact store's scratch namespace (novox/hq ADR 0350 §5): asked when it closed unmerged; a merged one's go " +
|
||||
|
||||
@@ -1,122 +0,0 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The values of the controller's record mirrors, built from the store (novox/hq ADR 0289, design 54 §1,
|
||||
// design 57 package F3).
|
||||
//
|
||||
// **Each value is built from a named list of fields, and nothing else** (design 54 §1): no epoch, no time
|
||||
// of writing and no time of the last look, so any lease holder builds the same bytes from the same record
|
||||
// and a pass over unchanged records writes nothing (ADR 0289 §1, §4). The encoding is fixed: the fields in
|
||||
// their struct's order, a declaration's maps sorted.
|
||||
|
||||
// AssignmentMirrored is one assignment as the assignments mirror holds it: the module, the node, and since
|
||||
// when it is assigned there. Design 54 §1 also lists whether it is pending, the builds sent and running and
|
||||
// the seats and provisions it holds there; those join with issue 419, which builds the rest of design 54.
|
||||
type AssignmentMirrored struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
Assigned time.Time `json:"assigned"`
|
||||
}
|
||||
|
||||
// DeclaredConfigMirrored is one registered module's config declarations as the declared-config mirror holds
|
||||
// them: the module, and each key its manifest's `settings` declares with that declaration's listed fields.
|
||||
// An empty object when the module declares none: registered, declaring nothing.
|
||||
type DeclaredConfigMirrored struct {
|
||||
Module string `json:"module"`
|
||||
Settings map[string]DeclaredMirrored `json:"settings"`
|
||||
}
|
||||
|
||||
// DeclaredMirrored is one key's declaration as the mirror holds it: **its listed fields and no other**
|
||||
// (ADR 0289 §12). Today a declaration is kind, default and why (catalogue.SettingDeclaration); a field the
|
||||
// manifest gains reaches the bus only once it is added here, deliberately, and a test fails until it is
|
||||
// listed or refused (ADR 0348 clause 4's fields arrive with package K1). A default is never a secret: the
|
||||
// catalogue refuses a default for a key whose name says secret.
|
||||
type DeclaredMirrored struct {
|
||||
Kind string `json:"kind"`
|
||||
Default json.RawMessage `json:"default"`
|
||||
Why string `json:"why"`
|
||||
}
|
||||
|
||||
// DeclaredMirroredFields are the fields of a declaration the mirror carries, by their manifest names.
|
||||
var DeclaredMirroredFields = []string{"kind", "default", "why"}
|
||||
|
||||
// AssignmentsMirror is every assignment's mirrored value, keyed `<node>.<module>`.
|
||||
func (i *Inventory) AssignmentsMirror(ctx context.Context) (map[string][]byte, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select n.name, a.module, a.assigned from assignment a join node n on n.id = a.node`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[string][]byte{}
|
||||
for rows.Next() {
|
||||
var a AssignmentMirrored
|
||||
if err := rows.Scan(&a.Node, &a.Module, &a.Assigned); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
a.Assigned = a.Assigned.UTC()
|
||||
value, err := json.Marshal(a)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[a.Node+"."+a.Module] = value
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// DeclaredConfigMirror is every registered module's mirrored config declarations, keyed by the module.
|
||||
func (i *Inventory) DeclaredConfigMirror(ctx context.Context) (map[string][]byte, error) {
|
||||
rows, err := i.store.Pool().Query(ctx,
|
||||
`select name, coalesce(manifest->'settings', '{}'::jsonb) from module`)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
out := map[string][]byte{}
|
||||
for rows.Next() {
|
||||
var (
|
||||
d DeclaredConfigMirrored
|
||||
raw []byte
|
||||
)
|
||||
if err := rows.Scan(&d.Module, &raw); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var declared map[string]DeclaredMirrored
|
||||
if err := json.Unmarshal(raw, &declared); err != nil {
|
||||
return nil, fmt.Errorf("the declarations of %s cannot be read: %w", d.Module, err)
|
||||
}
|
||||
d.Settings = map[string]DeclaredMirrored{}
|
||||
for key, decl := range declared {
|
||||
if len(decl.Default) > 0 {
|
||||
if decl.Default, err = canonical(decl.Default); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
d.Settings[key] = decl
|
||||
}
|
||||
value, err := json.Marshal(d)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
out[d.Module] = value
|
||||
}
|
||||
return out, rows.Err()
|
||||
}
|
||||
|
||||
// canonical is a JSON document encoded the one way: objects' keys sorted, numbers as written.
|
||||
func canonical(raw []byte) ([]byte, error) {
|
||||
dec := json.NewDecoder(bytes.NewReader(raw))
|
||||
dec.UseNumber()
|
||||
var v any
|
||||
if err := dec.Decode(&v); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return json.Marshal(v)
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
@@ -391,6 +392,65 @@ type PlanGate struct {
|
||||
// reports its module's container left running because it said busy, the judging waits and has not
|
||||
// started; once it reports the container replaced, Since is moved to then. Nil when nothing waited.
|
||||
Busy *GateBusy `json:"busy,omitempty"`
|
||||
// Pending says the send this gate judges is kept and not yet made (novox/hq ADR 0353 §8): a controller that
|
||||
// takes over reads the machine's last send before making it, and never sends it twice.
|
||||
Pending bool `json:"pending,omitempty"`
|
||||
// Silent says the machine reported nothing on its send within ApplySilentAfter (novox/hq ADR 0353 §4): left,
|
||||
// neither failed nor put back.
|
||||
Silent bool `json:"silent,omitempty"`
|
||||
// TookOver names the gate whose readings this one took over (novox/hq ADR 0353 §7): another walk's judging of
|
||||
// the same build on the same machine, which this gate goes on from instead of starting from zero.
|
||||
TookOver string `json:"took_over,omitempty"`
|
||||
}
|
||||
|
||||
// TakeReadings makes this gate go on from another's readings of the same build on the same machine (novox/hq ADR
|
||||
// 0353 §7): when the judging began, when each machine applied, the passes counted and when, and what each found.
|
||||
// A module's own count is taken only for the modules this gate judges.
|
||||
func (g *PlanGate) TakeReadings(from *PlanGate, modules []string, id string) {
|
||||
if from == nil {
|
||||
return
|
||||
}
|
||||
if from.Since != nil {
|
||||
since := *from.Since
|
||||
g.Since = &since
|
||||
}
|
||||
for n, s := range from.Sent {
|
||||
if slices.Contains(g.Machines, n) {
|
||||
if g.Sent == nil {
|
||||
g.Sent = map[string]SentDeclaration{}
|
||||
}
|
||||
g.Sent[n] = s
|
||||
}
|
||||
}
|
||||
for n, at := range from.Applied {
|
||||
if slices.Contains(g.Machines, n) {
|
||||
if g.Applied == nil {
|
||||
g.Applied = map[string]time.Time{}
|
||||
}
|
||||
g.Applied[n] = at
|
||||
}
|
||||
}
|
||||
g.Passes, g.Faulted, g.HealthySeen, g.Last = from.Passes, from.Faulted, from.HealthySeen, from.Last
|
||||
if from.LastPass != nil {
|
||||
at := *from.LastPass
|
||||
g.LastPass = &at
|
||||
}
|
||||
for _, m := range modules {
|
||||
if n, counted := from.Healthy[m]; counted {
|
||||
if g.Healthy == nil {
|
||||
g.Healthy = map[string]int{}
|
||||
}
|
||||
g.Healthy[m] = n
|
||||
}
|
||||
if at, counted := from.HealthyAt[m]; counted {
|
||||
if g.HealthyAt == nil {
|
||||
g.HealthyAt = map[string]time.Time{}
|
||||
}
|
||||
g.HealthyAt[m] = at
|
||||
}
|
||||
}
|
||||
g.Readings = append([]GateReading(nil), from.Readings...)
|
||||
g.TookOver = id
|
||||
}
|
||||
|
||||
// GateBusy is a first-node gate's wait for a busy container (novox/hq ADR 0335 §4).
|
||||
@@ -477,9 +537,17 @@ type CarriedMove struct {
|
||||
type PlanRelease struct {
|
||||
Order []string `json:"order"`
|
||||
Next int `json:"next"`
|
||||
// Gate is the machine being judged; nil between machines.
|
||||
// Gate is the first machine being judged; nil between machines.
|
||||
Gate *PlanGate `json:"gate,omitempty"`
|
||||
Done []string `json:"done,omitempty"`
|
||||
// Rest says the first machine passed (novox/hq ADR 0353 §1): every other machine is sent at once, each judged by
|
||||
// its own gate in Gates, by machine — a build no gate passed going to one machine of a step only (§2).
|
||||
Rest bool `json:"rest,omitempty"`
|
||||
Gates map[string]*PlanGate `json:"gates,omitempty"`
|
||||
// Failed are the machines of the rest whose gate failed, and Silent those that said nothing on their send
|
||||
// within ApplySilentAfter (ADR 0353 §3 and §4).
|
||||
Failed []string `json:"failed,omitempty"`
|
||||
Silent []string `json:"silent,omitempty"`
|
||||
Done []string `json:"done,omitempty"`
|
||||
// Skipped are the machines not heard from when their turn came, left as they were.
|
||||
Skipped []string `json:"skipped,omitempty"`
|
||||
// By is the person who released it, empty when the mesh did.
|
||||
|
||||
@@ -1,41 +0,0 @@
|
||||
package inventory
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
)
|
||||
|
||||
// jsonFields is the names a struct's fields are encoded under.
|
||||
func jsonFields(t reflect.Type) []string {
|
||||
var out []string
|
||||
for i := 0; i < t.NumField(); i++ {
|
||||
name, _, _ := strings.Cut(t.Field(i).Tag.Get("json"), ",")
|
||||
if name != "" && name != "-" {
|
||||
out = append(out, name)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// **A declaration reaches the bus with its listed fields and no other** (ADR 0289 §12): the mirror's value
|
||||
// carries exactly DeclaredMirroredFields, and every field a manifest's declaration has is on that list. A
|
||||
// field a declaration gains (package K1 widens it, ADR 0348 clause 4) fails here until it is listed — read,
|
||||
// judged not a secret and carried — so nothing reaches the declared-config mirror unreviewed.
|
||||
func TestTheDeclaredConfigMirrorCarriesOnlyItsListedFields(t *testing.T) {
|
||||
if got := jsonFields(reflect.TypeOf(DeclaredMirrored{})); !slices.Equal(got, DeclaredMirroredFields) {
|
||||
t.Errorf("the mirrored declaration carries %v, its list is %v", got, DeclaredMirroredFields)
|
||||
}
|
||||
for _, field := range jsonFields(reflect.TypeOf(catalogue.SettingDeclaration{})) {
|
||||
if !slices.Contains(DeclaredMirroredFields, field) {
|
||||
t.Errorf("a declaration has the field %q, which the declared-config mirror does not list: list it "+
|
||||
"in DeclaredMirrored (and say why it is no secret), or say here why it stays off the bus", field)
|
||||
}
|
||||
}
|
||||
if got := jsonFields(reflect.TypeOf(DeclaredConfigMirrored{})); !slices.Equal(got, []string{"module", "settings"}) {
|
||||
t.Errorf("the mirrored module carries %v", got)
|
||||
}
|
||||
}
|
||||
@@ -164,6 +164,30 @@ type CheckRequest struct {
|
||||
// each, from the merge it judged, into the scratch namespace once it has judged and not failed, for the walk to
|
||||
// take. A build seat from before it builds nothing, and the walk builds as before.
|
||||
Builds []CheckBuild `json:"builds,omitempty"`
|
||||
// Try marks a branch tried before its pull request (novox/hq issue 686): the repository's own check for a
|
||||
// pushed branch or commit, asked by a builder through the controller's `try` verb. Its outcome is read from
|
||||
// the build's log (StepTried) and said to nobody: no status on the forge, nothing recorded or registered,
|
||||
// nothing built for a merge. Tests narrows what the repository's script runs to the tests a pattern names,
|
||||
// and StoreOnly to its store tests; a script that reads neither runs whole.
|
||||
Try bool `json:"try,omitempty"`
|
||||
Tests string `json:"tests,omitempty"`
|
||||
StoreOnly bool `json:"store-only,omitempty"`
|
||||
}
|
||||
|
||||
// StepTried is the step a build seat says a try's outcome under, in the build's log: one line, a Tried as JSON
|
||||
// (novox/hq issue 686).
|
||||
const StepTried = "tried"
|
||||
|
||||
// Tried is what came of a branch tried before its pull request (novox/hq issue 686): the verdict of the
|
||||
// repository's own check for the ref merged into the trunk, how long it took, and the lines that name what
|
||||
// failed, picked from the whole of its output (issue 460).
|
||||
type Tried struct {
|
||||
Verdict string `json:"verdict"`
|
||||
Summary string `json:"summary"`
|
||||
Took string `json:"took"`
|
||||
Head string `json:"head,omitempty"`
|
||||
Trunk string `json:"trunk,omitempty"`
|
||||
Failed []string `json:"failed,omitempty"`
|
||||
}
|
||||
|
||||
// CheckBuild is one module a check builds for its merge: from the merge it made when Merged — a module built from
|
||||
|
||||
@@ -96,8 +96,27 @@ type BuildStart struct {
|
||||
// an earlier taker left unfinished — most often a build agent replaced mid-build — and started
|
||||
// again from the beginning (novox/hq issue 594). 0 from a machine that predates it.
|
||||
Delivery int `json:"delivery,omitempty"`
|
||||
// Try marks the start of a branch tried before its pull request (novox/hq issue 686), which names no
|
||||
// repository and no ref (startOf).
|
||||
Try bool `json:"try,omitempty"`
|
||||
}
|
||||
|
||||
// startOf is what a build machine says of the ask it took. **A try's start names no repository and no commit**
|
||||
// (novox/hq issue 686): the forge's holder writes mesh/merge-gate pending on an open pull request whose head is
|
||||
// the commit a start names, and a try of that head is no check of the pull request — nothing on the forge moves
|
||||
// for a try. Its id, machine and time are all the queue reads of a start.
|
||||
func startOf(r BuildRequest, on string, delivery int, at time.Time) BuildStart {
|
||||
s := BuildStart{ID: r.ID, Repository: r.Repository, Path: r.Path, Ref: r.Ref, On: on,
|
||||
At: at.UTC().Format(time.RFC3339Nano), Delivery: delivery}
|
||||
if IsTry(r) {
|
||||
s.Repository, s.Path, s.Ref, s.Try = "", "", "", true
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// IsTry says an ask is a branch tried before its pull request (novox/hq issue 686).
|
||||
func IsTry(r BuildRequest) bool { return r.Check != nil && r.Check.Try }
|
||||
|
||||
// Restarted is how a build handed out again says so (novox/hq issue 594), in its log and in its check's
|
||||
// report, so a second run never reads as the first. Neutral on why: a holder replaced mid-build, a paused
|
||||
// holder handing an ask back as it arrived, and a bus that held its acknowledgements past the wait all
|
||||
|
||||
@@ -401,10 +401,7 @@ func (b *natsBuild) Done() error { return b.msg.Ack() }
|
||||
// Began publishes that this machine has taken the build, into the stream like the outcome, so a
|
||||
// reader that asks afterwards sees when it started as well as how it ended.
|
||||
func (b *natsBuild) Began(ctx context.Context) error {
|
||||
body, err := json.Marshal(BuildStart{
|
||||
ID: b.request.ID, Repository: b.request.Repository, Path: b.request.Path,
|
||||
Ref: b.request.Ref, On: b.on, At: time.Now().UTC().Format(time.RFC3339Nano), Delivery: b.delivery,
|
||||
})
|
||||
body, err := json.Marshal(startOf(b.request, b.on, b.delivery, time.Now()))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -81,11 +81,6 @@ const (
|
||||
// steps from. Said after every save of the serving controller; a delivery's holder also asks for the
|
||||
// walks it follows, so a save made elsewhere is found by comparison, never lost.
|
||||
KeyPlanMoved = "plan-moved"
|
||||
// KeyAssignmentChanged and KeyDeclaredConfigChanged: a key of the controller's assignments or declared-config
|
||||
// mirror was written or removed (novox/hq ADR 0289, design 54 §3, design 57 F3): which key, how and at which
|
||||
// revision of the bucket — never the value, which the bucket holds.
|
||||
KeyAssignmentChanged = "assignment-changed"
|
||||
KeyDeclaredConfigChanged = "declared-config-changed"
|
||||
)
|
||||
|
||||
// Applied is what a machine now runs, as the mesh states it.
|
||||
|
||||
@@ -427,6 +427,9 @@ func ReadCheckBuilds(ctx context.Context, js *broker.JetStream, seat string, pul
|
||||
// safeKey is an id that can be a key, and a subject token, as it is: a build id, never a pattern.
|
||||
var safeKey = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
|
||||
|
||||
// SafeBuildID says an id can be a subject token as it is: a build's id, never a pattern.
|
||||
func SafeBuildID(id string) bool { return safeKey.MatchString(id) }
|
||||
|
||||
// Cancelled is what the controller writes for an ask it cancelled.
|
||||
type Cancelled struct {
|
||||
At string `json:"at"`
|
||||
|
||||
@@ -26,6 +26,8 @@ import (
|
||||
// - a check that has waited CheckWaitsAtMost goes before every delivery asked after it, so no check
|
||||
// waits for ever behind a stream of deliveries;
|
||||
// - within a kind, first come, first served, by when it was asked;
|
||||
// - a branch tried before its pull request goes after everything else, however long it has waited: it is
|
||||
// never promoted ahead of a delivery, and never ahead of a pull request's check (novox/hq issue 686);
|
||||
// - an ask whose kind cannot be told counts as a check: a delivery taken late breaks a budget, a check
|
||||
// taken late breaks nothing.
|
||||
//
|
||||
@@ -64,8 +66,12 @@ func (a QueuedAsk) askedOrStored() time.Time {
|
||||
// QueueOrder is the order the waiting asks are to be handed out in, now: the deliveries and every check
|
||||
// that has waited CheckWaitsAtMost, by when each was asked, then the other checks, by when each was asked.
|
||||
func QueueOrder(waiting []QueuedAsk, now time.Time) []QueuedAsk {
|
||||
var first, then []QueuedAsk
|
||||
var first, then, tries []QueuedAsk
|
||||
for _, a := range waiting {
|
||||
if IsTry(a.Request) {
|
||||
tries = append(tries, a)
|
||||
continue
|
||||
}
|
||||
if KindOf(a.Request) == KindCheck && now.Sub(a.askedOrStored()) < CheckWaitsAtMost {
|
||||
then = append(then, a)
|
||||
continue
|
||||
@@ -82,7 +88,8 @@ func QueueOrder(waiting []QueuedAsk, now time.Time) []QueuedAsk {
|
||||
}
|
||||
byAsking(first)
|
||||
byAsking(then)
|
||||
return append(first, then...)
|
||||
byAsking(tries)
|
||||
return append(append(first, then...), tries...)
|
||||
}
|
||||
|
||||
// toMove is the asks to move to the tail, in this order, for the queue to stand as order says: everything
|
||||
|
||||
@@ -0,0 +1,55 @@
|
||||
package link
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The start of a try names no repository and no commit (novox/hq issue 686): the forge's holder writes
|
||||
// mesh/merge-gate pending on the open pull request whose head a start names, and nothing on the forge moves
|
||||
// for a try. A pull request's check still names both.
|
||||
func TestATrysStartNamesNoCommitTheForgeCouldMatch(t *testing.T) {
|
||||
sha := strings.Repeat("c0ffee", 7)[:40]
|
||||
at := time.Date(2026, 10, 14, 0, 0, 0, 0, time.UTC)
|
||||
try := BuildRequest{ID: "build-1", Repository: "https://forge/novox/a.git", Ref: sha,
|
||||
Check: &CheckRequest{Owner: "novox", Repo: "a", Try: true}}
|
||||
body, _ := json.Marshal(startOf(try, "ace", 1, at))
|
||||
if strings.Contains(string(body), sha) || strings.Contains(string(body), "novox/a") {
|
||||
t.Fatalf("a try's start names what the forge matches a pull request by: %s", body)
|
||||
}
|
||||
if s := startOf(try, "ace", 2, at); s.ID != "build-1" || s.On != "ace" || !s.Try || s.Delivery != 2 || s.At == "" {
|
||||
t.Fatalf("a try's start lost what the queue reads: %+v", s)
|
||||
}
|
||||
pull := try
|
||||
pull.Check = &CheckRequest{Owner: "novox", Repo: "a", Number: 7}
|
||||
if s := startOf(pull, "ace", 1, at); s.Ref != sha || s.Repository == "" || s.Try {
|
||||
t.Fatalf("a pull request's check's start: %+v", s)
|
||||
}
|
||||
}
|
||||
|
||||
// A try stays behind every delivery and every pull request's check, however long it has waited: the ten
|
||||
// minutes that put a check ahead of later deliveries never promote a try (novox/hq issue 686).
|
||||
func TestATryIsNeverHandedOutBeforeADeliveryOrAPullRequestsCheck(t *testing.T) {
|
||||
now := time.Date(2026, 10, 12, 12, 0, 0, 0, time.UTC)
|
||||
try := func(seq uint64, id string, asked time.Time) QueuedAsk {
|
||||
a := anAsk(seq, id, asked, true)
|
||||
a.Request.Check.Try = true
|
||||
return a
|
||||
}
|
||||
waiting := []QueuedAsk{
|
||||
try(1, "try-old", now.Add(-40*time.Minute)),
|
||||
try(2, "try-young", now.Add(-time.Minute)),
|
||||
anAsk(3, "check-young", now.Add(-2*time.Minute), true),
|
||||
anAsk(4, "delivery", now.Add(-time.Minute), false),
|
||||
}
|
||||
want := []string{"delivery", "check-young", "try-old", "try-young"}
|
||||
if got := idsOf(QueueOrder(waiting, now)); !reflect.DeepEqual(got, want) {
|
||||
t.Fatalf("handed out as %v, want %v", got, want)
|
||||
}
|
||||
if !SafeBuildID("build-20261014T000000-1a2b") || SafeBuildID("x.>") || SafeBuildID("*") || SafeBuildID("") {
|
||||
t.Fatal("an id that is a pattern passes for a build's id")
|
||||
}
|
||||
}
|
||||
@@ -1,344 +0,0 @@
|
||||
// Package recordmirror writes the controller's record mirrors (novox/hq ADR 0289, design 54): a bucket per kind
|
||||
// of record, one key per record, written from the store only when the record changed, and one event on the
|
||||
// controller's seat per write.
|
||||
//
|
||||
// **The store stays the record.** A mirror is built from what the store committed, so a change rolled back
|
||||
// is never written. Every pass reads what the bucket holds and compares it byte for byte with what the store
|
||||
// says: a pass over unchanged records writes nothing and says nothing (ADR 0289 §1), and a value somebody
|
||||
// else wrote behind the writer's back (a former lease holder) differs from the store and is put right on
|
||||
// the next pass. Each write names the revision it read (compare-and-set), so a write in between is noticed,
|
||||
// never overwritten blind.
|
||||
//
|
||||
// **Only the lease holder writes** (ADR 0289 §9, design 54 §2): a pass begins by asking whether this
|
||||
// controller holds the lease, and asks again before every write; without it, it writes nothing.
|
||||
//
|
||||
// **The bucket is the authority, the event a notice** (design 54 §3): the event is sent after the write
|
||||
// succeeded, so there is no event for a value the bucket does not hold. A refused event is tried once more
|
||||
// and then counted as lost; a watcher that must not miss a change watches the bucket.
|
||||
//
|
||||
// What this package does not do yet is design 54 §4's audit (probe D17 and its healer H6), which compares a
|
||||
// mirror with its store behind the writer's back; novox/hq issue 419 carries it.
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"sort"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
)
|
||||
|
||||
// A Mirror is one kind of record the controller keeps on the bus: its local name, its bucket and the event
|
||||
// each write raises on the controller's seat.
|
||||
type Mirror struct {
|
||||
Name string
|
||||
Bucket string
|
||||
Event string
|
||||
}
|
||||
|
||||
// The mirrors this package writes (novox/hq design 57, package F3). The assignments mirror is design 54's,
|
||||
// keyed `<node>.<module>`; the declared config is each registered module's config declarations as its
|
||||
// manifest holds them, keyed by the module, which mesh-config resolves a module's config values from
|
||||
// (ADR 0348 clauses 2 and 4).
|
||||
var (
|
||||
Assignments = Mirror{Name: broker.MirrorAssignments, Bucket: broker.AssignmentsBucket,
|
||||
Event: link.KeyAssignmentChanged}
|
||||
DeclaredConfig = Mirror{Name: broker.MirrorDeclaredConfig, Bucket: broker.DeclaredConfigBucket,
|
||||
Event: link.KeyDeclaredConfigChanged}
|
||||
)
|
||||
|
||||
// Mirrors is every mirror this package writes, in the order they are written.
|
||||
func Mirrors() []Mirror { return []Mirror{Assignments, DeclaredConfig} }
|
||||
|
||||
// The changes an event says.
|
||||
const (
|
||||
Added = "added"
|
||||
Changed = "changed"
|
||||
Removed = "removed"
|
||||
)
|
||||
|
||||
// Event is what one write says on the controller's seat (design 54 §3): which mirror and key, what kind of
|
||||
// change, the names of the top-level fields that changed, the bucket's revision of the write, the lease's
|
||||
// epoch and when. Never the value: a watcher reads it from the bucket at that revision or later.
|
||||
type Event struct {
|
||||
Mirror string `json:"mirror"`
|
||||
Key string `json:"key"`
|
||||
Change string `json:"change"`
|
||||
Fields []string `json:"fields,omitempty"`
|
||||
Revision uint64 `json:"revision"`
|
||||
Epoch uint64 `json:"epoch,omitempty"`
|
||||
At time.Time `json:"at"`
|
||||
}
|
||||
|
||||
// Announcer says an event on a seat; link.OverNATS is one, and so is link.Bus.
|
||||
type Announcer interface {
|
||||
PublishSeatEvent(ctx context.Context, seat, event string, body []byte) error
|
||||
}
|
||||
|
||||
// MarkerTTL is how long the marker a purge leaves is kept (design 54 §2): a day, so the markers of removed
|
||||
// records do not pile up in a bucket that has no age.
|
||||
const MarkerTTL = 24 * time.Hour
|
||||
|
||||
// ErrNotHeld is a pass refused because this controller does not hold the lease: it wrote nothing.
|
||||
var ErrNotHeld = errors.New("this controller does not hold the lease, so it writes no mirror")
|
||||
|
||||
// Writer writes one mirror. Not safe for use by two goroutines at once: one controller's one loop owns it.
|
||||
type Writer struct {
|
||||
Mirror Mirror
|
||||
KV jetstream.KeyValue
|
||||
Announce Announcer
|
||||
// Epoch is the lease's epoch the events carry; nil carries none.
|
||||
Epoch func() uint64
|
||||
// Holds says this controller holds the lease. Asked before every pass and every write; a writer
|
||||
// without it writes nothing (design 54 §2). Required: nil holds nothing.
|
||||
Holds func() bool
|
||||
// Now is the clock the events are dated by; nil is time.Now.
|
||||
Now func() time.Time
|
||||
|
||||
known map[string]kept
|
||||
// LostEvents counts the events refused twice since the writer was made.
|
||||
LostEvents int
|
||||
}
|
||||
|
||||
type kept struct {
|
||||
value []byte
|
||||
revision uint64
|
||||
}
|
||||
|
||||
// Result is what one pass did: keys written, keys removed, and keys refused with why.
|
||||
type Result struct {
|
||||
Written int
|
||||
Removed int
|
||||
Refused int
|
||||
}
|
||||
|
||||
// Sync makes the bucket hold want, one key per record: writes each value that differs from what the key
|
||||
// holds, purges each key no record is left for, and raises one event per write. A key the bus cannot hold
|
||||
// is refused alone, and the others are written; the error then names it.
|
||||
func (w *Writer) Sync(ctx context.Context, want map[string][]byte) (Result, error) {
|
||||
var r Result
|
||||
if !w.holds() {
|
||||
return r, ErrNotHeld
|
||||
}
|
||||
// What the bucket holds now, every pass: a value written behind this writer's back differs from the
|
||||
// store's and is written again below.
|
||||
if err := w.fill(ctx); err != nil {
|
||||
return r, fmt.Errorf("reading the %s mirror: %w", w.Mirror.Name, err)
|
||||
}
|
||||
var problems []error
|
||||
keys := make([]string, 0, len(want))
|
||||
for k := range want {
|
||||
keys = append(keys, k)
|
||||
}
|
||||
sort.Strings(keys)
|
||||
for _, key := range keys {
|
||||
value := want[key]
|
||||
before, had := w.known[key]
|
||||
if had && bytes.Equal(before.value, value) {
|
||||
continue
|
||||
}
|
||||
if err := validKey(key); err != nil {
|
||||
r.Refused++
|
||||
problems = append(problems, err)
|
||||
continue
|
||||
}
|
||||
if !w.holds() {
|
||||
return r, errors.Join(append(problems, ErrNotHeld)...)
|
||||
}
|
||||
revision, wrote, err := w.put(ctx, key, value, before, had)
|
||||
if err != nil {
|
||||
problems = append(problems, fmt.Errorf("writing %s: %w", key, err))
|
||||
continue
|
||||
}
|
||||
w.known[key] = kept{value: append([]byte(nil), value...), revision: revision}
|
||||
if !wrote {
|
||||
continue
|
||||
}
|
||||
r.Written++
|
||||
change, fields := Added, []string(nil)
|
||||
if had {
|
||||
change, fields = Changed, changedFields(before.value, value)
|
||||
}
|
||||
w.say(ctx, key, change, fields, revision)
|
||||
}
|
||||
gone := make([]string, 0)
|
||||
for key := range w.known {
|
||||
if _, ok := want[key]; !ok {
|
||||
gone = append(gone, key)
|
||||
}
|
||||
}
|
||||
sort.Strings(gone)
|
||||
for _, key := range gone {
|
||||
k := w.known[key]
|
||||
if !w.holds() {
|
||||
return r, errors.Join(append(problems, ErrNotHeld)...)
|
||||
}
|
||||
err := w.KV.Purge(ctx, key, jetstream.LastRevision(k.revision), jetstream.PurgeTTL(MarkerTTL))
|
||||
if refusedByRevision(err) {
|
||||
// Read it again: purged already, or moved by another writer.
|
||||
if e, gerr := w.KV.Get(ctx, key); errors.Is(gerr, jetstream.ErrKeyNotFound) {
|
||||
delete(w.known, key)
|
||||
continue
|
||||
} else if gerr == nil && w.holds() {
|
||||
err = w.KV.Purge(ctx, key, jetstream.LastRevision(e.Revision()), jetstream.PurgeTTL(MarkerTTL))
|
||||
}
|
||||
}
|
||||
if err != nil {
|
||||
problems = append(problems, fmt.Errorf("removing %s: %w", key, err))
|
||||
continue
|
||||
}
|
||||
delete(w.known, key)
|
||||
r.Removed++
|
||||
revision := uint64(0)
|
||||
if status, err := w.KV.Status(ctx); err == nil {
|
||||
// The purge's own sequence is the stream's last one only when nothing wrote in between, which no
|
||||
// other writer may; said as the bucket's revision after it.
|
||||
if s, ok := status.(interface{ StreamInfo() *jetstream.StreamInfo }); ok {
|
||||
revision = s.StreamInfo().State.LastSeq
|
||||
}
|
||||
}
|
||||
w.say(ctx, key, Removed, nil, revision)
|
||||
}
|
||||
return r, errors.Join(problems...)
|
||||
}
|
||||
|
||||
// put writes one value by compare-and-set and answers its revision, and whether this call wrote it (false
|
||||
// when the bucket already held it).
|
||||
func (w *Writer) put(ctx context.Context, key string, value []byte, before kept, had bool) (uint64, bool, error) {
|
||||
var (
|
||||
revision uint64
|
||||
err error
|
||||
)
|
||||
if had {
|
||||
revision, err = w.KV.Update(ctx, key, value, before.revision)
|
||||
} else {
|
||||
revision, err = w.KV.Create(ctx, key, value)
|
||||
}
|
||||
if !refusedByRevision(err) {
|
||||
return revision, err == nil, err
|
||||
}
|
||||
// Refused: another writer in between, or this writer's own write whose acknowledgement was lost.
|
||||
e, gerr := w.KV.Get(ctx, key)
|
||||
switch {
|
||||
case errors.Is(gerr, jetstream.ErrKeyNotFound):
|
||||
if !w.holds() {
|
||||
return 0, false, errors.New("the key moved and this controller no longer holds the lease")
|
||||
}
|
||||
revision, err = w.KV.Create(ctx, key, value)
|
||||
return revision, err == nil, err
|
||||
case gerr != nil:
|
||||
return 0, false, gerr
|
||||
case bytes.Equal(e.Value(), value):
|
||||
return e.Revision(), false, nil
|
||||
case !w.holds():
|
||||
return 0, false, errors.New("the key moved and this controller no longer holds the lease")
|
||||
}
|
||||
revision, err = w.KV.Update(ctx, key, value, e.Revision())
|
||||
return revision, err == nil, err
|
||||
}
|
||||
|
||||
// say raises one write's event, tried once more when refused and then counted as lost.
|
||||
func (w *Writer) say(ctx context.Context, key, change string, fields []string, revision uint64) {
|
||||
if w.Announce == nil {
|
||||
return
|
||||
}
|
||||
e := Event{Mirror: w.Mirror.Name, Key: key, Change: change, Fields: fields, Revision: revision, At: w.now().UTC()}
|
||||
if w.Epoch != nil {
|
||||
e.Epoch = w.Epoch()
|
||||
}
|
||||
body, err := json.Marshal(e)
|
||||
if err != nil {
|
||||
w.LostEvents++
|
||||
return
|
||||
}
|
||||
for try := 0; try < 2; try++ {
|
||||
if err = w.Announce.PublishSeatEvent(ctx, broker.ControllerSeat, w.Mirror.Event, body); err == nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
w.LostEvents++
|
||||
}
|
||||
|
||||
// fill reads what the bucket holds now, so a controller started again writes only what differs.
|
||||
func (w *Writer) fill(ctx context.Context) error {
|
||||
w.known = map[string]kept{}
|
||||
watch, err := w.KV.WatchAll(ctx, jetstream.IgnoreDeletes())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = watch.Stop() }()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
case e := <-watch.Updates():
|
||||
if e == nil {
|
||||
return nil
|
||||
}
|
||||
w.known[e.Key()] = kept{value: append([]byte(nil), e.Value()...), revision: e.Revision()}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (w *Writer) holds() bool { return w.Holds != nil && w.Holds() }
|
||||
|
||||
func (w *Writer) now() time.Time {
|
||||
if w.Now == nil {
|
||||
return time.Now()
|
||||
}
|
||||
return w.Now()
|
||||
}
|
||||
|
||||
// refusedByRevision is the server refusing a compare-and-set: the key is not at the revision named.
|
||||
func refusedByRevision(err error) bool {
|
||||
if err == nil {
|
||||
return false
|
||||
}
|
||||
if errors.Is(err, jetstream.ErrKeyExists) {
|
||||
return true
|
||||
}
|
||||
var apiErr *jetstream.APIError
|
||||
return errors.As(err, &apiErr) && apiErr.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence
|
||||
}
|
||||
|
||||
// keyChars is what a bucket's key may be made of.
|
||||
var keyChars = regexp.MustCompile(`^[-/_=.a-zA-Z0-9]+$`)
|
||||
|
||||
// validKey refuses a key the bus cannot hold, naming it.
|
||||
func validKey(key string) error {
|
||||
if key == "" || !keyChars.MatchString(key) || key[0] == '.' || key[len(key)-1] == '.' {
|
||||
return fmt.Errorf("%q cannot be a key of a bucket: letters, digits, '-', '_', '=', '/' and '.', not at "+
|
||||
"either end", key)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// changedFields names the top-level fields whose values differ between two encoded objects, sorted; nil
|
||||
// when either is not an object.
|
||||
func changedFields(before, after []byte) []string {
|
||||
var a, b map[string]json.RawMessage
|
||||
if json.Unmarshal(before, &a) != nil || json.Unmarshal(after, &b) != nil || a == nil || b == nil {
|
||||
return nil
|
||||
}
|
||||
var out []string
|
||||
for k, v := range b {
|
||||
if old, ok := a[k]; !ok || !bytes.Equal(old, v) {
|
||||
out = append(out, k)
|
||||
}
|
||||
}
|
||||
for k := range a {
|
||||
if _, ok := b[k]; !ok {
|
||||
out = append(out, k)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
@@ -1,178 +0,0 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/catalogue"
|
||||
"github.com/novox/mesh-controller/internal/inventory"
|
||||
)
|
||||
|
||||
// **Design 57's F3, end to end** (novox/hq issue 652): from the controller's store to a real bus. An
|
||||
// assignment changes exactly one key of the assignments mirror and raises exactly one `assignment-changed`;
|
||||
// a registration changes exactly one key of the declared-config mirror and raises exactly one
|
||||
// `declared-config-changed`; and the same records written again change nothing and say nothing — a module
|
||||
// registered again unchanged included, although the store's registration time moved.
|
||||
func TestNatsAnAssignmentAndARegistrationEachChangeOneKeyAndRaiseOneEvent(t *testing.T) {
|
||||
inv := inventory.ForTest(t)
|
||||
ctx := t.Context()
|
||||
js, api := aMirrorBus(t)
|
||||
assignments := aWriter(t, js, api, Assignments)
|
||||
declared := aWriter(t, js, api, DeclaredConfig)
|
||||
heardAssignments := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event)
|
||||
heardDeclared := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+DeclaredConfig.Event)
|
||||
|
||||
pass := func() (Result, Result) {
|
||||
t.Helper()
|
||||
a, err := inv.AssignmentsMirror(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ra, err := assignments.Sync(ctx, a)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
d, err := inv.DeclaredConfigMirror(ctx)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rd, err := declared.Sync(ctx, d)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return ra, rd
|
||||
}
|
||||
|
||||
if _, err := inv.AddNode(ctx, "anchor"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ui := catalogue.Manifest{Module: "ui", Version: "1", Settings: map[string]catalogue.SettingDeclaration{
|
||||
"font-size": {Kind: catalogue.KindPreference, Default: 14, Why: "readable on the desktop"}}}
|
||||
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
pass()
|
||||
assignmentWrites, declaredWrites := writes(t, api, Assignments.Bucket), writes(t, api, DeclaredConfig.Bucket)
|
||||
assignmentsSaid, declaredSaid := len(settle(heardAssignments, 0)), len(settle(heardDeclared, 1))
|
||||
|
||||
// **A registration**: a module new to the catalogue.
|
||||
if err := inv.RegisterModule(ctx, catalogue.Manifest{Module: "notes", Version: "1"}, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ra, rd := pass()
|
||||
if ra != (Result{}) || rd != (Result{Written: 1}) {
|
||||
t.Errorf("a registration wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
if got := writes(t, api, DeclaredConfig.Bucket) - declaredWrites; got != 1 {
|
||||
t.Errorf("a registration changed %d keys of the declared config", got)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 0 {
|
||||
t.Errorf("a registration changed %d keys of the assignments", got)
|
||||
}
|
||||
said := settle(heardDeclared, declaredSaid+1)[declaredSaid:]
|
||||
if len(said) != 1 || said[0].Key != "notes" || said[0].Change != Added {
|
||||
t.Errorf("a registration raised %+v", said)
|
||||
}
|
||||
declaredWrites, declaredSaid = writes(t, api, DeclaredConfig.Bucket), declaredSaid+len(said)
|
||||
|
||||
// **An assignment.**
|
||||
if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ra, rd = pass()
|
||||
if ra != (Result{Written: 1}) || rd != (Result{}) {
|
||||
t.Errorf("an assignment wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - assignmentWrites; got != 1 {
|
||||
t.Errorf("an assignment changed %d keys of the assignments", got)
|
||||
}
|
||||
said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:]
|
||||
if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Added {
|
||||
t.Errorf("an assignment raised %+v", said)
|
||||
}
|
||||
assignmentWrites, assignmentsSaid = writes(t, api, Assignments.Bucket), assignmentsSaid+len(said)
|
||||
e, err := assignments.KV.Get(ctx, "anchor.ui")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var value inventory.AssignmentMirrored
|
||||
if err := json.Unmarshal(e.Value(), &value); err != nil || value.Node != "anchor" || value.Module != "ui" ||
|
||||
value.Assigned.IsZero() {
|
||||
t.Errorf("the assignment is mirrored as %s (%v)", e.Value(), err)
|
||||
}
|
||||
|
||||
// **The same records written again change nothing**: an assignment made again, and a module registered
|
||||
// again with the manifest it had.
|
||||
if _, err := inv.Assign(ctx, "anchor", "ui"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ra, rd := pass(); ra != (Result{}) || rd != (Result{}) {
|
||||
t.Errorf("unchanged records wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
if writes(t, api, Assignments.Bucket) != assignmentWrites || writes(t, api, DeclaredConfig.Bucket) != declaredWrites {
|
||||
t.Error("unchanged records changed a key")
|
||||
}
|
||||
if a, d := settle(heardAssignments, assignmentsSaid+1), settle(heardDeclared, declaredSaid+1); len(a) != assignmentsSaid ||
|
||||
len(d) != declaredSaid {
|
||||
t.Errorf("unchanged records raised %+v and %+v", a[assignmentsSaid:], d[declaredSaid:])
|
||||
}
|
||||
|
||||
// **A declaration changed by a registration**: one key, one event, naming the field that moved.
|
||||
ui.Version = "2"
|
||||
ui.Settings["font-size"] = catalogue.SettingDeclaration{Kind: catalogue.KindPreference, Default: 16,
|
||||
Why: "readable on the desktop"}
|
||||
if err := inv.RegisterModule(ctx, ui, inventory.Source{}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ra, rd := pass(); ra != (Result{}) || rd != (Result{Written: 1}) {
|
||||
t.Errorf("a changed declaration wrote assignments %+v and declared config %+v", ra, rd)
|
||||
}
|
||||
said = settle(heardDeclared, declaredSaid+1)[declaredSaid:]
|
||||
if len(said) != 1 || said[0].Key != "ui" || said[0].Change != Changed || strings.Join(said[0].Fields, ",") != "settings" {
|
||||
t.Errorf("a changed declaration raised %+v", said)
|
||||
}
|
||||
got, err := declared.KV.Get(ctx, "ui")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if want := `{"module":"ui","settings":{"font-size":{"kind":"preference","default":16,"why":"readable on the desktop"}}}`; string(got.Value()) != want {
|
||||
t.Errorf("the declaration is mirrored as %s, want %s", got.Value(), want)
|
||||
}
|
||||
|
||||
// **An unassignment** removes its key, said once.
|
||||
if err := inv.Unassign(ctx, "anchor", "ui"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if ra, _ := pass(); ra != (Result{Removed: 1}) {
|
||||
t.Errorf("an unassignment did %+v", ra)
|
||||
}
|
||||
said = settle(heardAssignments, assignmentsSaid+1)[assignmentsSaid:]
|
||||
if len(said) != 1 || said[0].Key != "anchor.ui" || said[0].Change != Removed {
|
||||
t.Errorf("an unassignment raised %+v", said)
|
||||
}
|
||||
|
||||
// **A module no longer registered** has its key removed, said once as `removed` (ADR 0356 §4).
|
||||
if err := inv.ForgetModule(ctx, "notes"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
declaredWrites = writes(t, api, DeclaredConfig.Bucket)
|
||||
declaredSaid = len(heardDeclared.all())
|
||||
if _, rd := pass(); rd != (Result{Removed: 1}) {
|
||||
t.Errorf("a module forgotten did %+v", rd)
|
||||
}
|
||||
if got := writes(t, api, DeclaredConfig.Bucket) - declaredWrites; got != 1 {
|
||||
t.Errorf("a module forgotten changed %d keys of the declared config", got)
|
||||
}
|
||||
if _, err := declared.KV.Get(ctx, "notes"); err == nil {
|
||||
t.Error("a module forgotten keeps its key")
|
||||
}
|
||||
said = settle(heardDeclared, declaredSaid+1)[declaredSaid:]
|
||||
if len(said) != 1 || said[0].Key != "notes" || said[0].Change != Removed {
|
||||
t.Errorf("a module forgotten raised %+v", said)
|
||||
}
|
||||
}
|
||||
@@ -1,356 +0,0 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"slices"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
"github.com/novox/mesh-controller/internal/link"
|
||||
"github.com/novox/mesh-controller/internal/testbus"
|
||||
)
|
||||
|
||||
// A record mirror against a real bus (novox/hq design 54 §2, §3; ADR 0289): a change of a record changes
|
||||
// exactly one key and raises exactly one event; a pass over unchanged records writes nothing and says
|
||||
// nothing; a record gone is a purge said as `removed`.
|
||||
|
||||
// aMirrorBus is a bus of this test alone with the mesh's streams and the controller's buckets, as the
|
||||
// controller asserts them at its start, and a counter of the writes made to one bucket.
|
||||
func aMirrorBus(t *testing.T) (*broker.JetStream, jetstream.JetStream) {
|
||||
t.Helper()
|
||||
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)
|
||||
}
|
||||
return js, api
|
||||
}
|
||||
|
||||
// heard counts what is said on one subject, each body kept.
|
||||
type heard struct {
|
||||
mu sync.Mutex
|
||||
bodies []Event
|
||||
}
|
||||
|
||||
func (h *heard) all() []Event {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return append([]Event(nil), h.bodies...)
|
||||
}
|
||||
|
||||
func listen(t *testing.T, conn *nats.Conn, subject string) *heard {
|
||||
t.Helper()
|
||||
h := &heard{}
|
||||
sub, err := conn.Subscribe(subject, func(m *nats.Msg) {
|
||||
var e Event
|
||||
if err := json.Unmarshal(m.Data, &e); err != nil {
|
||||
t.Errorf("an event that is not one: %s", m.Data)
|
||||
return
|
||||
}
|
||||
h.mu.Lock()
|
||||
h.bodies = append(h.bodies, e)
|
||||
h.mu.Unlock()
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
if err := conn.Flush(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return h
|
||||
}
|
||||
|
||||
// writes is how many messages the bucket's stream has taken: every put and every purge is one.
|
||||
func writes(t *testing.T, api jetstream.JetStream, bucket string) uint64 {
|
||||
t.Helper()
|
||||
s, err := api.Stream(t.Context(), "KV_"+bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
info, err := s.Info(t.Context())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return info.State.LastSeq
|
||||
}
|
||||
|
||||
// settle waits until n events are heard or a second passed, and answers what was heard.
|
||||
func settle(h *heard, n int) []Event {
|
||||
deadline := time.Now().Add(time.Second)
|
||||
for time.Now().Before(deadline) && len(h.all()) < n {
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
// And a moment more, so an event too many is heard too.
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
return h.all()
|
||||
}
|
||||
|
||||
func aWriter(t *testing.T, js *broker.JetStream, api jetstream.JetStream, m Mirror) *Writer {
|
||||
t.Helper()
|
||||
kv, err := api.KeyValue(t.Context(), m.Bucket)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return &Writer{Mirror: m, KV: kv, Announce: link.OverNATS{Conn: js.Conn(), JS: js.Context()},
|
||||
Epoch: func() uint64 { return 7 }, Holds: func() bool { return true }}
|
||||
}
|
||||
|
||||
func TestNatsAMirrorChangesOneKeyAndRaisesOneEventPerChange(t *testing.T) {
|
||||
for _, m := range []Mirror{Assignments, DeclaredConfig} {
|
||||
t.Run(m.Name, func(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+m.Event)
|
||||
w := aWriter(t, js, api, m)
|
||||
ctx := t.Context()
|
||||
|
||||
first := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}
|
||||
if _, err := w.Sync(ctx, first); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, m.Bucket)
|
||||
heardBefore := len(settle(events, 1))
|
||||
|
||||
// One record changes, one is added: two keys, two events, nothing else.
|
||||
second := map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)}
|
||||
r, err := w.Sync(ctx, second)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Written != 2 || r.Removed != 0 {
|
||||
t.Errorf("a pass over one changed and one new record wrote %+v", r)
|
||||
}
|
||||
if got := writes(t, api, m.Bucket) - before; got != 2 {
|
||||
t.Errorf("the bucket took %d writes for two changed records", got)
|
||||
}
|
||||
said := settle(events, heardBefore+2)[heardBefore:]
|
||||
if len(said) != 2 {
|
||||
t.Fatalf("two changes raised %d events: %+v", len(said), said)
|
||||
}
|
||||
changes := map[string]Event{}
|
||||
for _, e := range said {
|
||||
changes[e.Key] = e
|
||||
}
|
||||
if e := changes["anchor.gitea"]; e.Change != Changed || !slices.Equal(e.Fields, []string{"a"}) ||
|
||||
e.Mirror != m.Name || e.Epoch != 7 || e.Revision == 0 {
|
||||
t.Errorf("the change of anchor.gitea was said as %+v", e)
|
||||
}
|
||||
if e := changes["anchor.forgejo"]; e.Change != Added {
|
||||
t.Errorf("the new anchor.forgejo was said as %+v", e)
|
||||
}
|
||||
|
||||
// **An unchanged pass writes nothing**, and says nothing (ADR 0289 §1).
|
||||
heardBefore = len(events.all())
|
||||
before = writes(t, api, m.Bucket)
|
||||
r, err = w.Sync(ctx, second)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r != (Result{}) {
|
||||
t.Errorf("a pass over unchanged records did %+v", r)
|
||||
}
|
||||
if got := writes(t, api, m.Bucket) - before; got != 0 {
|
||||
t.Errorf("a pass over unchanged records wrote the bucket %d times", got)
|
||||
}
|
||||
if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 0 {
|
||||
t.Errorf("a pass over unchanged records raised %+v", said)
|
||||
}
|
||||
|
||||
// **A record gone is a purge**, one write and one `removed`.
|
||||
before = writes(t, api, m.Bucket)
|
||||
r, err = w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Removed != 1 || r.Written != 0 {
|
||||
t.Errorf("a record gone did %+v", r)
|
||||
}
|
||||
if got := writes(t, api, m.Bucket) - before; got != 1 {
|
||||
t.Errorf("a record gone wrote the bucket %d times", got)
|
||||
}
|
||||
if _, err := w.KV.Get(ctx, "anchor.forgejo"); !errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
t.Errorf("the key of a record gone is still read: %v", err)
|
||||
}
|
||||
said = settle(events, heardBefore+1)[heardBefore:]
|
||||
if len(said) != 1 || said[0].Change != Removed || said[0].Key != "anchor.forgejo" {
|
||||
t.Errorf("a record gone was said as %+v", said)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// **A controller started again writes nothing for what the bucket already holds** (design 54 §2: the writer's
|
||||
// memory is filled from the bucket), and a key nobody holds a record for is removed.
|
||||
func TestNatsAWriterStartedAgainWritesOnlyWhatDiffers(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
first := aWriter(t, js, api, Assignments)
|
||||
held := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`), "anchor.forgejo": []byte(`{"a":1}`)}
|
||||
if _, err := first.Sync(ctx, held); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, Assignments.Bucket)
|
||||
again := aWriter(t, js, api, Assignments)
|
||||
r, err := again.Sync(ctx, held)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r != (Result{}) || writes(t, api, Assignments.Bucket) != before {
|
||||
t.Errorf("a writer started again over unchanged records did %+v", r)
|
||||
}
|
||||
}
|
||||
|
||||
// **Another writer in between is no lost value** (design 54 §2, compare-and-set): a write refused because the
|
||||
// key moved reads it again, takes the revision when it already holds the value, and otherwise writes again.
|
||||
func TestNatsAWriteRefusedByCompareAndSetReadsAgain(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
if _, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// Behind its back, as a former lease holder would.
|
||||
if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r.Written != 1 {
|
||||
t.Errorf("a refused write was not made again: %+v", r)
|
||||
}
|
||||
e, err := w.KV.Get(ctx, "anchor.gitea")
|
||||
if err != nil || string(e.Value()) != `{"a":2}` {
|
||||
t.Fatalf("the bucket holds %v, %v", e, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A value written behind the writer's back is put right on the next pass** (ADR 0289 §10): a former lease
|
||||
// holder's write differs from the store, and the store's value is written again, said as one change.
|
||||
func TestNatsAForeignWriteIsCorrectedOnTheNextPass(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
events := listen(t, js.Conn(), "mesh.seat."+broker.ControllerSeat+".event."+Assignments.Event)
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
store := map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}
|
||||
if _, err := w.Sync(ctx, store); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
heardBefore := len(settle(events, 1))
|
||||
if _, err := w.KV.Put(ctx, "anchor.gitea", []byte(`{"a":9}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
r, err := w.Sync(ctx, store)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if r != (Result{Written: 1}) {
|
||||
t.Errorf("a pass after a foreign write did %+v", r)
|
||||
}
|
||||
e, err := w.KV.Get(ctx, "anchor.gitea")
|
||||
if err != nil || string(e.Value()) != `{"a":1}` {
|
||||
t.Fatalf("the bucket still holds %v, %v", e, err)
|
||||
}
|
||||
if said := settle(events, heardBefore+1)[heardBefore:]; len(said) != 1 || said[0].Change != Changed {
|
||||
t.Errorf("the correction was said as %+v", said)
|
||||
}
|
||||
// And put right, it stays: the next pass writes nothing.
|
||||
if r, err := w.Sync(ctx, store); err != nil || r != (Result{}) {
|
||||
t.Errorf("a pass after the correction did %+v, %v", r, err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A writer without the lease writes nothing** (ADR 0289 §9, design 54 §2): not a new key, not a change, not
|
||||
// a removal, and it says so.
|
||||
func TestNatsAWriterWithoutTheLeaseWritesNothing(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
held := aWriter(t, js, api, Assignments)
|
||||
if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, Assignments.Bucket)
|
||||
for _, holds := range []func() bool{nil, func() bool { return false }} {
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
w.Holds = holds
|
||||
r, err := w.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":2}`), "anchor.forgejo": []byte(`{"a":1}`)})
|
||||
if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
|
||||
t.Errorf("a writer without the lease did %+v, %v", r, err)
|
||||
}
|
||||
r, err = w.Sync(ctx, map[string][]byte{})
|
||||
if !errors.Is(err, ErrNotHeld) || r != (Result{}) {
|
||||
t.Errorf("a writer without the lease removed: %+v, %v", r, err)
|
||||
}
|
||||
}
|
||||
// And one that loses it mid-pass stops before its next write.
|
||||
asked := 0
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
w.Holds = func() bool { asked++; return asked <= 2 }
|
||||
r, err := w.Sync(ctx, map[string][]byte{"anchor.a": []byte(`{}`), "anchor.b": []byte(`{}`), "anchor.gitea": []byte(`{"a":1}`)})
|
||||
if !errors.Is(err, ErrNotHeld) || r.Written != 1 {
|
||||
t.Errorf("a writer that lost the lease mid-pass did %+v, %v", r, err)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - before; got != 1 {
|
||||
t.Errorf("writers without the lease wrote the bucket %d times", got)
|
||||
}
|
||||
}
|
||||
|
||||
// **A removal asks for the lease too** (ADR 0289 §9): a writer that held the lease when its pass began and lost
|
||||
// it before its first removal purges nothing, and says it does not hold the lease.
|
||||
func TestNatsARemovalWithoutTheLeaseRemovesNothing(t *testing.T) {
|
||||
js, api := aMirrorBus(t)
|
||||
ctx := t.Context()
|
||||
held := aWriter(t, js, api, Assignments)
|
||||
if _, err := held.Sync(ctx, map[string][]byte{"anchor.gitea": []byte(`{"a":1}`)}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
before := writes(t, api, Assignments.Bucket)
|
||||
asked := 0
|
||||
w := aWriter(t, js, api, Assignments)
|
||||
w.Holds = func() bool { asked++; return asked == 1 }
|
||||
r, err := w.Sync(ctx, map[string][]byte{})
|
||||
if !errors.Is(err, ErrNotHeld) || r != (Result{}) || asked != 2 {
|
||||
t.Errorf("a removal without the lease did %+v, %v (the lease asked %d times)", r, err, asked)
|
||||
}
|
||||
if got := writes(t, api, Assignments.Bucket) - before; got != 0 {
|
||||
t.Errorf("a removal without the lease wrote the bucket %d times", got)
|
||||
}
|
||||
if _, err := w.KV.Get(ctx, "anchor.gitea"); err != nil {
|
||||
t.Errorf("the key was removed without the lease: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// **A key the bus cannot hold is refused alone**, and the others are written.
|
||||
func TestAKeyTheBusCannotHoldIsRefusedAlone(t *testing.T) {
|
||||
if validKey("anchor.gitea") != nil || validKey("bad key") == nil || validKey(".lead") == nil ||
|
||||
validKey("") == nil || validKey("trail.") == nil {
|
||||
t.Error("the keys a bucket may hold are not told apart")
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheFieldsThatChangedAreNamed(t *testing.T) {
|
||||
got := changedFields([]byte(`{"a":1,"b":{"x":1},"c":"same"}`), []byte(`{"a":2,"b":{"x":1},"d":true,"c":"same"}`))
|
||||
if !slices.Equal(got, []string{"a", "d"}) {
|
||||
t.Errorf("the fields that changed are %v", got)
|
||||
}
|
||||
if got := changedFields(nil, []byte(`{"a":1}`)); got != nil {
|
||||
t.Errorf("an added key names fields %v", got)
|
||||
}
|
||||
}
|
||||
@@ -1,181 +0,0 @@
|
||||
package recordmirror
|
||||
|
||||
import (
|
||||
"go/ast"
|
||||
"go/parser"
|
||||
"go/token"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"slices"
|
||||
"strconv"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/novox/mesh-controller/internal/broker"
|
||||
)
|
||||
|
||||
// mayNameAMirror are the files of the controller that may name a record mirror's bucket outside their tests:
|
||||
// where the buckets are defined, asserted and granted (internal/broker), the list a reader names in `reads`
|
||||
// (internal/catalogue/state.go), and each mirror's own writer (this package and the loop that runs it). The
|
||||
// audit of a mirror against its store (ADR 0289 §4, issue 419) joins this list when it is built.
|
||||
var mayNameAMirror = []string{
|
||||
"cmd/mesh-controller/record_mirrors.go",
|
||||
"internal/broker/controller_buckets.go",
|
||||
"internal/broker/writers.go",
|
||||
"internal/catalogue/state.go",
|
||||
}
|
||||
|
||||
// The packages whose names stand for a mirror, by import path, and the names in each.
|
||||
const (
|
||||
brokerPath = "github.com/novox/mesh-controller/internal/broker"
|
||||
recordmirrorPath = "github.com/novox/mesh-controller/internal/recordmirror"
|
||||
)
|
||||
|
||||
var mirrorNames = map[string][]string{
|
||||
brokerPath: {"AssignmentsBucket", "DeclaredConfigBucket", "MirrorAssignments", "MirrorDeclaredConfig",
|
||||
"ControllerMirrors"},
|
||||
recordmirrorPath: {"Assignments", "DeclaredConfig", "Mirrors"},
|
||||
}
|
||||
|
||||
// **Nothing in the controller reads a mirror to decide anything** (ADR 0289 §19 as ADR 0356 §2 changes it): no
|
||||
// file of the controller's packages outside its tests names a mirror's bucket but the audit and a mirror's own
|
||||
// writer. A reader elsewhere fails here, naming the file and what it named.
|
||||
//
|
||||
// **What it catches**: a name that stands for a mirror, qualified by the package it comes from under any
|
||||
// local name the file imports it as (`broker.AssignmentsBucket`, `b.AssignmentsBucket`); the same names bare
|
||||
// in a file of package broker or package recordmirror; and a string literal holding a mirror's bucket name.
|
||||
// **What it does not catch**: a bucket name built at run time, such as `BucketName(ControllerSeat,
|
||||
// "assignments")` or literals concatenated into one, and a mirror reached through a value passed in from an
|
||||
// allowed file. Those are deliberate evasions rather than ordinary code; a reviewer judges them.
|
||||
func TestOnlyTheAuditAndAMirrorsWriterReadAMirror(t *testing.T) {
|
||||
var buckets []string
|
||||
for _, m := range broker.ControllerMirrors() {
|
||||
buckets = append(buckets, broker.BucketName(broker.ControllerSeat, m))
|
||||
}
|
||||
root := filepath.Join("..", "..")
|
||||
found := mirrorReaders(t, root, buckets)
|
||||
if len(found) > 0 {
|
||||
t.Fatalf("a package of the controller reads a record mirror, which only the audit and the mirror's "+
|
||||
"writer may (ADR 0289 §19, ADR 0356 §2):\n %s", strings.Join(found, "\n "))
|
||||
}
|
||||
}
|
||||
|
||||
// **The reader test sees what it says it sees**: in a planted tree, each planted reader is named, and an allowed
|
||||
// file and a selector on a value are not.
|
||||
func TestTheReaderTestFindsPlantedReaders(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
plant := func(path, src string) {
|
||||
full := filepath.Join(root, path)
|
||||
if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(full, []byte(src), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
plant("cmd/mesh-controller/qualified.go", "package main\n\nimport \""+brokerPath+"\"\n\nvar a = broker.AssignmentsBucket\n")
|
||||
plant("internal/x/aliased.go", "package x\n\nimport b \""+brokerPath+"\"\n\nvar a = b.DeclaredConfigBucket\n")
|
||||
plant("internal/broker/zz_reader.go", "package broker\n\nfunc zzR() string { return AssignmentsBucket }\n")
|
||||
plant("internal/inventory/literal.go", "package inventory\n\nvar a = \"mesh-controller_declared-config\"\n")
|
||||
plant("internal/y/writer.go", "package y\n\nimport \""+recordmirrorPath+"\"\n\nvar a = recordmirror.Assignments\n")
|
||||
// Allowed: a mirror's definition and its writer.
|
||||
plant("internal/broker/controller_buckets.go", "package broker\n\nvar AssignmentsBucket = \"x\"\n")
|
||||
plant("internal/broker/other.go", "package broker\n\nfunc other() string { return \"nothing\" }\n")
|
||||
// A selector on a value (v.AssignmentsBucket) is no use of the mirror, but declaring a field of a mirror's
|
||||
// name in package broker is named, on the safe side; and a bare use inside a call under a selector is found.
|
||||
plant("internal/broker/field.go", "package broker\n\ntype s struct{ AssignmentsBucket string }\n\nfunc f(v s) string { return v.AssignmentsBucket }\n")
|
||||
plant("internal/broker/nested.go", "package broker\n\nfunc g() int { return len(h(DeclaredConfigBucket).x) }\n")
|
||||
found := mirrorReaders(t, root, []string{"mesh-controller_assignments", "mesh-controller_declared-config"})
|
||||
want := []string{
|
||||
"cmd/mesh-controller/qualified.go: broker.AssignmentsBucket",
|
||||
"internal/broker/field.go: AssignmentsBucket",
|
||||
"internal/broker/nested.go: DeclaredConfigBucket",
|
||||
"internal/broker/zz_reader.go: AssignmentsBucket",
|
||||
"internal/inventory/literal.go: \"mesh-controller_declared-config\"",
|
||||
"internal/x/aliased.go: b.DeclaredConfigBucket",
|
||||
"internal/y/writer.go: recordmirror.Assignments",
|
||||
}
|
||||
slices.Sort(found)
|
||||
if !slices.Equal(found, want) {
|
||||
t.Errorf("the planted readers were found as\n %s\nwant\n %s", strings.Join(found, "\n "),
|
||||
strings.Join(want, "\n "))
|
||||
}
|
||||
}
|
||||
|
||||
// mirrorReaders is every place under root's cmd and internal, outside tests and the allowed files, that names a
|
||||
// mirror.
|
||||
func mirrorReaders(t *testing.T, root string, buckets []string) []string {
|
||||
t.Helper()
|
||||
var found []string
|
||||
for _, dir := range []string{"cmd", "internal"} {
|
||||
err := filepath.WalkDir(filepath.Join(root, dir), func(path string, e fs.DirEntry, err error) error {
|
||||
if err != nil || e.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
|
||||
return err
|
||||
}
|
||||
rel, err := filepath.Rel(root, path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
rel = filepath.ToSlash(rel)
|
||||
if strings.HasPrefix(rel, "internal/recordmirror/") || slices.Contains(mayNameAMirror, rel) {
|
||||
return nil
|
||||
}
|
||||
f, err := parser.ParseFile(token.NewFileSet(), path, nil, 0)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
// Each import's local name, mapped to the package it stands for.
|
||||
imported := map[string]string{}
|
||||
for _, imp := range f.Imports {
|
||||
p, _ := strconv.Unquote(imp.Path.Value)
|
||||
if _, ok := mirrorNames[p]; !ok {
|
||||
continue
|
||||
}
|
||||
local := filepath.Base(p)
|
||||
if imp.Name != nil {
|
||||
local = imp.Name.Name
|
||||
}
|
||||
imported[local] = p
|
||||
}
|
||||
// A file of the package that defines the names may use them bare.
|
||||
var bare []string
|
||||
switch f.Name.Name {
|
||||
case "broker":
|
||||
bare = mirrorNames[brokerPath]
|
||||
case "recordmirror":
|
||||
bare = mirrorNames[recordmirrorPath]
|
||||
}
|
||||
ast.Inspect(f, func(n ast.Node) bool {
|
||||
switch n := n.(type) {
|
||||
case *ast.SelectorExpr:
|
||||
pkg, ok := n.X.(*ast.Ident)
|
||||
if !ok {
|
||||
return true
|
||||
}
|
||||
if p, imp := imported[pkg.Name]; imp && slices.Contains(mirrorNames[p], n.Sel.Name) {
|
||||
found = append(found, rel+": "+pkg.Name+"."+n.Sel.Name)
|
||||
}
|
||||
// Neither a selector's own name nor its package is a bare use.
|
||||
return false
|
||||
case *ast.Ident:
|
||||
if slices.Contains(bare, n.Name) {
|
||||
found = append(found, rel+": "+n.Name)
|
||||
}
|
||||
case *ast.BasicLit:
|
||||
for _, b := range buckets {
|
||||
if n.Kind == token.STRING && strings.Contains(n.Value, b) {
|
||||
found = append(found, rel+": "+n.Value)
|
||||
}
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
return found
|
||||
}
|
||||
+19
-4
@@ -54,11 +54,26 @@ if [ -n "$unformatted" ]; then
|
||||
fi
|
||||
CGO_ENABLED=0 go vet ./...
|
||||
|
||||
if command -v gcc >/dev/null 2>&1; then
|
||||
CGO_ENABLED=1 go test -race -count=1 -timeout 30m ./...
|
||||
suite() {
|
||||
if command -v gcc >/dev/null 2>&1; then
|
||||
CGO_ENABLED=1 go test -race -count=1 -timeout 30m "$@"
|
||||
else
|
||||
echo "NOT RACE-CHECKED: the toolchain holds no C compiler; the suite runs without the race detector"
|
||||
CGO_ENABLED=0 go test -count=1 -timeout 30m "$@"
|
||||
fi
|
||||
}
|
||||
# **A branch tried before its pull request may ask for a part** (novox/hq issue 686): the controller's `try` verb
|
||||
# hands MESH_CHECK_TESTS, a `go test -run` pattern, or MESH_CHECK_STORE_ONLY=1. A pull request's merge check is
|
||||
# handed neither and always runs whole; a part that passes is said as a part, never as the check.
|
||||
if [ -n "${MESH_CHECK_TESTS:-}" ]; then
|
||||
echo "TRIED IN PART: only the tests matching $MESH_CHECK_TESTS, in every package"
|
||||
suite -run "$MESH_CHECK_TESTS" ./...
|
||||
exit 0
|
||||
fi
|
||||
if [ "${MESH_CHECK_STORE_ONLY:-}" = 1 ]; then
|
||||
echo "TRIED IN PART: only the store tests listed in testdata/store-tests"
|
||||
else
|
||||
echo "NOT RACE-CHECKED: the toolchain holds no C compiler; the suite runs without the race detector"
|
||||
CGO_ENABLED=0 go test -count=1 -timeout 30m ./...
|
||||
suite ./...
|
||||
fi
|
||||
|
||||
# **The store tests listed ran** (novox/hq issue 459): a test that needs the store skips without one, and a skip
|
||||
|
||||
@@ -77,6 +77,7 @@
|
||||
"replay",
|
||||
"kill",
|
||||
"forget-checks",
|
||||
"try",
|
||||
"pause",
|
||||
"resume",
|
||||
"hand-act",
|
||||
|
||||
Vendored
+1
-1
@@ -8,7 +8,7 @@ private addresses. That is acceptable while mesh-controller is private; if it ev
|
||||
a second place to clean besides the catalogue itself.
|
||||
|
||||
mesh-catalog e0bc919d4945bb5f13f971f74643a0b714cf45da modules/*/module.json, modules/nats/Dockerfile
|
||||
mesh-host c8920622516548978d92e734f3991a262f98a742 examples/foundation-first-node.lock
|
||||
mesh-host 7a41e120635b3f7b6cb94cca8b3fc4093e4e6110 examples/foundation-first-node.lock
|
||||
mesh-sdk 5933eb757d443293631be2bed27a96288c2348d8 conformance/events/module-event.json
|
||||
|
||||
To move them, from this repository's root, with the two repositories checked out beside it:
|
||||
|
||||
+29
-79
@@ -28,12 +28,11 @@
|
||||
// image, and this file is applied on a machine with no mesh to ask about anything (ADR 0006). The
|
||||
// registry prefix is the lab's; the digests are upstream's.
|
||||
//
|
||||
// **The store is raised where the `postgres` module keeps it** (novox/hq ADR 0338): the container
|
||||
// `postgres`, on the module's pgvector image, its data in the directory /var/lib/mesh-store owned by
|
||||
// the image's postgres user, its superuser password read from the module's own-secret file — so the
|
||||
// store step adopts it as it runs. Pinned upstream by the module's own digest, not through the lab's
|
||||
// registry prefix, because the two must be the same reference (mesh-host internal/bootstrap holds
|
||||
// them to each other against the module's manifest).
|
||||
// The store's data is a NAMED VOLUME, not a directory on the machine. A directory the host
|
||||
// creates is owned by root, and the database runs as somebody else inside the container — so it
|
||||
// could not write, and the container crash-looped. A named volume lets the image set up its own
|
||||
// ownership, and outlives the container, which is what you want for the thing holding the mesh's
|
||||
// state.
|
||||
|
||||
{
|
||||
"declaration": 1,
|
||||
@@ -84,28 +83,17 @@
|
||||
"boot": "enabled",
|
||||
"restart-on": ["base-filter"]
|
||||
},
|
||||
// The store's data is a directory owned by the image's postgres user (999), made before the
|
||||
// container starts: a directory the runtime made would be root's, and the database could not
|
||||
// write it. Measured on the module's image 2026-10-13: it initialises a new store on a directory
|
||||
// made so (novox/hq ADR 0338 §5).
|
||||
{
|
||||
"id": "store-data",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/mesh-store",
|
||||
"mode": "0700",
|
||||
"owner": "999:70"
|
||||
},
|
||||
{
|
||||
"id": "store",
|
||||
"type": "container",
|
||||
"name": "postgres",
|
||||
"image": "pgvector/pgvector@sha256:cf134a767f474095eeba57e0117be8e568e011a63f33fbf252f14c9b760f8e6f",
|
||||
"name": "mesh-store",
|
||||
"image": "192.0.2.250:5000/postgres@sha256:7abf537131b66ed5af448d90653abf1679b0c7e9a1f07efdd4c3108a401b259a",
|
||||
"env": {
|
||||
"POSTGRES_PASSWORD": "bootstrap",
|
||||
"PGDATA": "/var/lib/postgresql/data/pgdata"
|
||||
},
|
||||
"ports": ["5432:5432"],
|
||||
"volumes": ["/var/lib/mesh-store:/var/lib/postgresql/data"]
|
||||
"volumes": ["mesh-store-data:/var/lib/postgresql/data"]
|
||||
},
|
||||
// Over TCP, not the socket. While the store initialises it runs a temporary server on the
|
||||
// socket ONLY, then stops it and starts the real one — so a socket check passes, the action
|
||||
@@ -114,21 +102,21 @@
|
||||
{
|
||||
"id": "store-ready",
|
||||
"type": "action",
|
||||
"in": "postgres",
|
||||
"in": "mesh-store",
|
||||
"command": ["sh", "-c", "for i in $(seq 1 180); do pg_isready -h 127.0.0.1 -U postgres >/dev/null 2>&1 && exit 0; sleep 1; done; echo 'the store did not answer within 180s; its own last words follow'; pg_isready -h 127.0.0.1 -U postgres; tail -n 20 /var/lib/postgresql/data/log/*.log 2>/dev/null; exit 1"],
|
||||
"verify": ["pg_isready", "-h", "127.0.0.1", "-U", "postgres"]
|
||||
},
|
||||
{
|
||||
"id": "inventory-database",
|
||||
"type": "action",
|
||||
"in": "postgres",
|
||||
"in": "mesh-store",
|
||||
"command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE inventory'"],
|
||||
"verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw inventory"]
|
||||
},
|
||||
{
|
||||
"id": "identity-database",
|
||||
"type": "action",
|
||||
"in": "postgres",
|
||||
"in": "mesh-store",
|
||||
"command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE identity'"],
|
||||
"verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw identity"]
|
||||
},
|
||||
@@ -138,44 +126,20 @@
|
||||
{
|
||||
"id": "licences-database",
|
||||
"type": "action",
|
||||
"in": "postgres",
|
||||
"in": "mesh-store",
|
||||
"command": ["sh", "-c", "psql -U postgres -c 'CREATE DATABASE licences'"],
|
||||
"verify": ["sh", "-c", "psql -U postgres -lqt | cut -d'|' -f1 | grep -qw licences"]
|
||||
},
|
||||
{
|
||||
"id": "context-schemas",
|
||||
"type": "action",
|
||||
"command": ["docker", "run", "--rm", "--network", "container:postgres",
|
||||
"command": ["docker", "run", "--rm", "--network", "container:mesh-store",
|
||||
"-e", "MESH_STORE_INVENTORY=postgres://postgres:bootstrap@127.0.0.1:5432/inventory?sslmode=disable",
|
||||
"-e", "MESH_STORE_IDENTITY=postgres://postgres:bootstrap@127.0.0.1:5432/identity?sslmode=disable",
|
||||
"-e", "MESH_STORE_LICENCES=postgres://postgres:bootstrap@127.0.0.1:5432/licences?sslmode=disable",
|
||||
"192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
|
||||
"migrate"],
|
||||
"verify": ["sh", "-c", "docker exec postgres psql -U postgres -d inventory -tAc \"select to_regclass('public.node')\" | grep -qx node && docker exec postgres psql -U postgres -d identity -tAc \"select to_regclass('public.signing_key')\" | grep -qx signing_key && docker exec postgres psql -U postgres -d licences -tAc \"select to_regclass('public.licence')\" | grep -qx licence"]
|
||||
},
|
||||
// **The bus is raised where the `nats` module keeps it** (novox/hq ADR 0333 §1): the container
|
||||
// `mesh-broker-nats`, its configuration and user list in /var/lib/nats-module/conf, its streams in
|
||||
// /var/lib/mesh-broker-nats and its certificate in /var/lib/mesh-broker-tls — the names and paths the
|
||||
// catalogue's module declares. So adopting it as that module at the broker step keeps it: the
|
||||
// module's built image takes the place of the upstream one below, once, on the same streams, users
|
||||
// and certificate, and nothing else moves. mesh-host's internal/bootstrap holds the two to each other
|
||||
// against the module's own manifest (busLayout), at the broker step and in its tests.
|
||||
//
|
||||
// Directories on the machine, not volumes the runtime names: the module mounts these paths, and the
|
||||
// control plane, once it runs as the mesh's process, reads the certificate at its path on the machine.
|
||||
// The certificate's directory is readable by everyone because the certificate is public and the control
|
||||
// plane runs as nobody; its key is written 0600 by the program that makes it.
|
||||
{
|
||||
"id": "bus-tls-dir",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/mesh-broker-tls",
|
||||
"mode": "0755"
|
||||
},
|
||||
{
|
||||
"id": "bus-data-dir",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/mesh-broker-nats",
|
||||
"mode": "0700"
|
||||
"verify": ["sh", "-c", "docker exec mesh-store psql -U postgres -d inventory -tAc \"select to_regclass('public.node')\" | grep -qx node && docker exec mesh-store psql -U postgres -d identity -tAc \"select to_regclass('public.signing_key')\" | grep -qx signing_key && docker exec mesh-store psql -U postgres -d licences -tAc \"select to_regclass('public.licence')\" | grep -qx licence"]
|
||||
},
|
||||
{
|
||||
"id": "bus-certificate",
|
||||
@@ -187,63 +151,49 @@
|
||||
// it, and asking nothing of the image it writes into. Self-signed on purpose: a host pins
|
||||
// this server's exact certificate (novox/hq ADR 0004), and at this moment there is no mesh
|
||||
// to ask an authority of.
|
||||
// `--user 0:0` because the directory is root's and this image runs as nobody, which is right
|
||||
// for the long-running control plane and wrong for a one-shot writing into a fresh directory.
|
||||
"command": ["docker", "run", "--rm", "--user", "0:0", "-v", "/var/lib/mesh-broker-tls:/tls",
|
||||
// `--user 0:0` because the volume is root's and this image runs as nobody, which is right
|
||||
// for the long-running control plane and wrong for a one-shot writing into a fresh volume.
|
||||
"command": ["docker", "run", "--rm", "--user", "0:0", "-v", "mesh-broker-tls:/tls",
|
||||
"192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
|
||||
"broker", "certificate", "--into", "/tls"],
|
||||
"verify": ["docker", "run", "--rm", "--user", "0:0", "-v", "/var/lib/mesh-broker-tls:/tls",
|
||||
"verify": ["docker", "run", "--rm", "--user", "0:0", "-v", "mesh-broker-tls:/tls",
|
||||
"192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
|
||||
"broker", "certificate", "--check", "--into", "/tls"]
|
||||
},
|
||||
// **Seeds, written only where nothing is** (create-once). The bus's configuration and user list are
|
||||
// the bundle's until the bus is adopted, and the module's from then on (ADR 0333 §4): an installer
|
||||
// run again after the adoption applies this bundle again, and an ordinary file would put the
|
||||
// bundle's one user back over every user the mesh composed, and leave the node unable to reach its
|
||||
// bus. Until the adoption the installer places the composed list over the seed (enrol.go).
|
||||
{
|
||||
"id": "bus-conf-dir",
|
||||
"type": "directory",
|
||||
"path": "/var/lib/nats-module/conf",
|
||||
"path": "/var/lib/mesh-bus-conf",
|
||||
"mode": "0700"
|
||||
},
|
||||
{
|
||||
"id": "bus-conf",
|
||||
"type": "file",
|
||||
"path": "/var/lib/nats-module/conf/nats.conf",
|
||||
"path": "/var/lib/mesh-bus-conf/nats.conf",
|
||||
"mode": "0644",
|
||||
"create-once": true,
|
||||
"content": "port: 4222\nhttp: 0.0.0.0:8222\n\ntls {\n cert_file: \"/tls/tls.crt\"\n key_file: \"/tls/tls.key\"\n}\n\njetstream {\n store_dir: \"/data\"\n}\n\ninclude accounts.conf\n"
|
||||
"content": "port: 4222\nhttp: 127.0.0.1:8222\n\ntls {\n cert_file: \"/tls/tls.crt\"\n key_file: \"/tls/tls.key\"\n}\n\njetstream {\n store_dir: \"/data\"\n}\n\ninclude accounts.conf\n"
|
||||
},
|
||||
{
|
||||
"id": "bus-accounts",
|
||||
"type": "file",
|
||||
"path": "/var/lib/nats-module/conf/accounts.conf",
|
||||
"path": "/var/lib/mesh-bus-conf/accounts.conf",
|
||||
"mode": "0600",
|
||||
"create-once": true,
|
||||
"content": "// The first user list, carried by the installer because at genesis there is no mesh to\n// compose one. A bootstrap credential, rotated with the store's and replaced by the\n// controller's own composition from its first start onward.\naccounts {\n MESH {\n jetstream: enabled\n users = [\n { user: \"controller\", password: \"$2a$10$AHqJgOifIVbU41KmATiMhuXFs8xa7Wl2HuN4UVBCXdN2jIQzjqApy\", permissions: {\n publish: { allow: [\"$JS.ACK.CONTROL.controller.>\", \"$JS.ACK.EVENTS.controller.>\", \"$JS.ACK.DEAD_LETTER_NOTICES.controller.>\", \"$JS.API.>\", \"$KV.mesh-controller_asked.>\", \"$KV.mesh-controller_assignments.>\", \"$KV.mesh-controller_calls.>\", \"$KV.mesh-controller_hand-acts.>\", \"$KV.mesh-controller_conditions.>\", \"$KV.mesh-controller_condition-history.>\", \"$KV.mesh-controller_declared-config.>\", \"$KV.mesh-controller_lease.>\", \"$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>\", \"$KV.SEAT_NODE_BUILD_AGENT_cancelled.>\", \"_INBOX.enrol.>\", \"mesh.assignment.>\", \"mesh.mod.*.tool.>\", \"mesh.node.>\", \"mesh.seat.mesh-build-machine.accept.>\", \"mesh.seat.node-build-agent.accept.>\", \"mesh.seat.mesh-build-machine.tool.>\", \"mesh.seat.node-build-agent.tool.>\", \"mesh.seat.mesh-controller.event.applied\", \"mesh.seat.mesh-controller.event.built-before\", \"mesh.seat.mesh-controller.event.refused\", \"mesh.seat.mesh-controller.event.rolled-back\", \"mesh.seat.mesh-controller.event.checked\", \"mesh.seat.mesh-controller.event.plan-moved\", \"mesh.seat.mesh-controller.event.assignment-changed\", \"mesh.seat.mesh-controller.event.declared-config-changed\", \"mesh.seat.mesh-controller.event.condition-raised\", \"mesh.seat.mesh-controller.event.condition-changed\", \"mesh.seat.mesh-controller.event.condition-cleared\", \"mesh.seat.mesh-controller.event.doctor-heartbeat\", \"mesh.seat.mesh-controller.event.secret-replaced\", \"mesh.seat.mesh-controller.event.healer-acted\", \"$SRV.INFO\", \"mesh.seat.node-intrusion-prevention.tool.banned.*\", \"mesh.seat.node-backup.tool.backed-up.*\", \"mesh.seat.node-backup.tool.now.*\", \"mesh.seat.mesh-delivery.tool.stalled\", \"mesh.seat.mesh-delivery.tool.close\", \"mesh.seat.node-launcher.tool.secret.*\", \"mesh.seat.mesh-delivery.tool.release\", \"mesh.seat.mesh-delivery.tool.stop\", \"mesh.seat.mesh-delivery.tool.times\", \"mesh.seat.mesh-delivery.tool.groups\", \"mesh.seat.node-service-manager.tool.restart.*\", \"mesh.seat.mesh-controller.tool.plans\", \"mesh.events.dead.>\", \"mesh.again.>\"] }\n subscribe: { allow: [\"$JS.API.>\", \"$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>\", \"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>\", \"_DELIVER.controller\", \"_DELIVER.controller.>\", \"_INBOX.controller.>\", \"mesh.control.>\", \"mesh.mod.*.event.provisioner.failing\", \"mesh.mod.*.event.provisioner.recovered\", \"mesh.mod.*.event.provisioner.retirement\", \"mesh.mod.gitea.event.pull.merged\", \"mesh.mod.gitea.event.pull.updated\", \"mesh.mod.mesh-catalog.event.catching-up\", \"mesh.mod.mesh-catalog.event.upgraded\", \"mesh.seat.mesh-build-machine.event.built\", \"mesh.seat.node-build-agent.event.built\", \"mesh.seat.operator-channel.event.decided.mesh-controller\", \"mesh.seat.mesh-controller.tool.>\", \"$SRV.PING\", \"$SRV.INFO\", \"$SRV.PING.mesh-controller\", \"$SRV.PING.mesh-controller.>\", \"$SRV.INFO.mesh-controller\", \"$SRV.INFO.mesh-controller.>\", \"$SRV.STATS\", \"$SRV.STATS.mesh-controller\", \"$SRV.STATS.mesh-controller.>\"] }\n allow_responses: { max: 1, ttl: \"1m\" }\n } }\n ]\n }\n}\n"
|
||||
"content": "// The first user list, carried by the installer because at genesis there is no mesh to\n// compose one. A bootstrap credential, rotated with the store's and replaced by the\n// controller's own composition from its first start onward.\naccounts {\n MESH {\n jetstream: enabled\n users = [\n { user: \"controller\", password: \"$2a$10$AHqJgOifIVbU41KmATiMhuXFs8xa7Wl2HuN4UVBCXdN2jIQzjqApy\", permissions: {\n publish: { allow: [\"$JS.ACK.CONTROL.controller.>\", \"$JS.ACK.EVENTS.controller.>\", \"$JS.ACK.DEAD_LETTER_NOTICES.controller.>\", \"$JS.API.>\", \"$KV.mesh-controller_asked.>\", \"$KV.mesh-controller_calls.>\", \"$KV.mesh-controller_hand-acts.>\", \"$KV.mesh-controller_conditions.>\", \"$KV.mesh-controller_condition-history.>\", \"$KV.mesh-controller_lease.>\", \"$KV.SEAT_MESH_BUILD_MACHINE_cancelled.>\", \"$KV.SEAT_NODE_BUILD_AGENT_cancelled.>\", \"_INBOX.enrol.>\", \"mesh.assignment.>\", \"mesh.mod.*.tool.>\", \"mesh.node.>\", \"mesh.seat.mesh-build-machine.accept.>\", \"mesh.seat.node-build-agent.accept.>\", \"mesh.seat.mesh-build-machine.tool.>\", \"mesh.seat.node-build-agent.tool.>\", \"mesh.seat.mesh-controller.event.applied\", \"mesh.seat.mesh-controller.event.built-before\", \"mesh.seat.mesh-controller.event.refused\", \"mesh.seat.mesh-controller.event.rolled-back\", \"mesh.seat.mesh-controller.event.checked\", \"mesh.seat.mesh-controller.event.plan-moved\", \"mesh.seat.mesh-controller.event.condition-raised\", \"mesh.seat.mesh-controller.event.condition-changed\", \"mesh.seat.mesh-controller.event.condition-cleared\", \"mesh.seat.mesh-controller.event.doctor-heartbeat\", \"mesh.seat.mesh-controller.event.secret-replaced\", \"mesh.seat.mesh-controller.event.healer-acted\", \"$SRV.INFO\", \"mesh.seat.node-intrusion-prevention.tool.banned.*\", \"mesh.seat.node-backup.tool.backed-up.*\", \"mesh.seat.node-backup.tool.now.*\", \"mesh.seat.mesh-delivery.tool.stalled\", \"mesh.seat.mesh-delivery.tool.close\", \"mesh.seat.node-launcher.tool.secret.*\", \"mesh.seat.mesh-delivery.tool.release\", \"mesh.seat.mesh-delivery.tool.stop\", \"mesh.seat.mesh-delivery.tool.times\", \"mesh.seat.mesh-delivery.tool.groups\", \"mesh.seat.node-service-manager.tool.restart.*\", \"mesh.seat.mesh-controller.tool.plans\", \"mesh.events.dead.>\", \"mesh.again.>\"] }\n subscribe: { allow: [\"$JS.API.>\", \"$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>\", \"$JS.EVENT.ADVISORY.CONSUMER.DELETED.>\", \"_DELIVER.controller\", \"_DELIVER.controller.>\", \"_INBOX.controller.>\", \"mesh.control.>\", \"mesh.mod.*.event.provisioner.failing\", \"mesh.mod.*.event.provisioner.recovered\", \"mesh.mod.*.event.provisioner.retirement\", \"mesh.mod.gitea.event.pull.merged\", \"mesh.mod.gitea.event.pull.updated\", \"mesh.mod.mesh-catalog.event.catching-up\", \"mesh.mod.mesh-catalog.event.upgraded\", \"mesh.seat.mesh-build-machine.event.built\", \"mesh.seat.node-build-agent.event.built\", \"mesh.seat.operator-channel.event.decided.mesh-controller\", \"mesh.seat.mesh-controller.tool.>\", \"$SRV.PING\", \"$SRV.INFO\", \"$SRV.PING.mesh-controller\", \"$SRV.PING.mesh-controller.>\", \"$SRV.INFO.mesh-controller\", \"$SRV.INFO.mesh-controller.>\", \"$SRV.STATS\", \"$SRV.STATS.mesh-controller\", \"$SRV.STATS.mesh-controller.>\"] }\n allow_responses: { max: 1, ttl: \"1m\" }\n } }\n ]\n }\n}\n"
|
||||
},
|
||||
// The upstream server, until the broker step adopts it: the `nats` module's own image, built by the
|
||||
// mesh, takes its place there once, on these same directories (ADR 0333 §3), as the bus's planned
|
||||
// step (ADR 0236). The release is the one the module's image is built on (its NATS_BASE, 2.11.17),
|
||||
// pinned upstream by that same digest: the adoption changes no server release, so it can be undone
|
||||
// by putting the old image back and is said reversible. The published port is the
|
||||
// node's bus port, 5671 unless given, which every enrolment token names; the installer gives the
|
||||
// module the setting that maps its declared 4222 to it (ADR 0333 §2).
|
||||
{
|
||||
"id": "broker",
|
||||
"type": "container",
|
||||
"name": "mesh-broker-nats",
|
||||
"image": "nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00",
|
||||
"name": "mesh-broker",
|
||||
"image": "192.0.2.250:5000/nats@sha256:b83efabe3e7def1e0a4a31ec6e078999bb17c80363f881df35edc70fcb6bb927",
|
||||
"ports": ["5671:4222", "127.0.0.1:8222:8222"],
|
||||
"volumes": ["/var/lib/mesh-broker-nats:/data", "/var/lib/nats-module/conf:/etc/nats:ro", "/var/lib/mesh-broker-tls:/tls:ro"],
|
||||
"volumes": ["mesh-broker-data:/data", "mesh-broker-tls:/tls:ro", "/var/lib/mesh-bus-conf:/etc/nats:ro"],
|
||||
"args": ["-c", "/etc/nats/nats.conf", "-js"]
|
||||
},
|
||||
{
|
||||
"id": "broker-ready",
|
||||
"type": "action",
|
||||
"command": ["sh", "-c", "for i in $(seq 1 60); do docker run --rm --network host --entrypoint sh nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00 -c 'nc -z 127.0.0.1 5671' >/dev/null 2>&1 && exit 0; sleep 1; done; exit 1"],
|
||||
"verify": ["sh", "-c", "docker run --rm --network host --entrypoint sh nats@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00 -c 'nc -z 127.0.0.1 5671'"]
|
||||
"command": ["sh", "-c", "for i in $(seq 1 60); do docker run --rm --network host --entrypoint sh 192.0.2.250:5000/nats@sha256:b83efabe3e7def1e0a4a31ec6e078999bb17c80363f881df35edc70fcb6bb927 -c 'nc -z 127.0.0.1 5671' >/dev/null 2>&1 && exit 0; sleep 1; done; exit 1"],
|
||||
"verify": ["sh", "-c", "docker run --rm --network host --entrypoint sh 192.0.2.250:5000/nats@sha256:b83efabe3e7def1e0a4a31ec6e078999bb17c80363f881df35edc70fcb6bb927 -c 'nc -z 127.0.0.1 5671'"]
|
||||
},
|
||||
{
|
||||
"id": "control-plane",
|
||||
@@ -252,7 +202,7 @@
|
||||
"image": "192.0.2.250:5000/mesh-controller@sha256:c67db38439ff0aee242b467486765467bb95801f52175fc5727cc4e437338ace",
|
||||
"network": "host",
|
||||
"args": ["serve"],
|
||||
"volumes": ["/var/lib/mesh-broker-tls:/broker-tls:ro"],
|
||||
"volumes": ["mesh-broker-tls:/broker-tls:ro"],
|
||||
"env": {
|
||||
"MESH_STORE_INVENTORY": "postgres://postgres:bootstrap@127.0.0.1:5432/inventory?sslmode=disable",
|
||||
"MESH_STORE_IDENTITY": "postgres://postgres:bootstrap@127.0.0.1:5432/identity?sslmode=disable",
|
||||
|
||||
Vendored
+11
-1
@@ -49,7 +49,6 @@
|
||||
./internal/inventory TestACredentialIssuedOutsideIsNeverMade
|
||||
./internal/inventory TestOneAcceptForAProviderIsSealedToEveryConsumer
|
||||
./internal/inventory TestASharedOwnSecretIssuedOutsideIsNeverMade
|
||||
./internal/recordmirror TestNatsAnAssignmentAndARegistrationEachChangeOneKeyAndRaiseOneEvent
|
||||
./cmd/mesh-controller TestOneAcceptForAProviderIsPlacedForEveryConsumer
|
||||
./cmd/mesh-controller TestReplay164
|
||||
./cmd/mesh-controller TestAnApprovedLayerOpensTheWalkThatSendsIt
|
||||
@@ -62,3 +61,14 @@
|
||||
./cmd/mesh-controller TestAProviderElsewhereServesItsOwnComposedName
|
||||
./cmd/mesh-controller TestAReusedCheckBuildIsRegisteredByItsDigestAndGatedAsABuiltOne
|
||||
./internal/inventory TestAReportOfABusyContainerIsKeptAndReadBack
|
||||
./cmd/mesh-controller TestAReleaseSendsTheRestAtOnceAfterTheFirstPasses
|
||||
./cmd/mesh-controller TestARestMachineThatFailsIsPutBackAloneAndTheOthersAreJudgedOn
|
||||
./cmd/mesh-controller TestAWalkTakesOverTheReadingsOfAGateJudgingTheSameBuildThere
|
||||
./cmd/mesh-controller TestAFirstSendMadeAndNotKeptIsNotSentAgain
|
||||
./cmd/mesh-controller TestAReleaseSendKeptAndMadeIsNotMadeAgain
|
||||
./cmd/mesh-controller TestAFirstMachineThatFailsStillStopsTheWalk
|
||||
./cmd/mesh-controller TestTheControlNodeIsSentLastInAStep
|
||||
./cmd/mesh-controller TestARestMachineSilentPastItsBoundIsLeftAndSaid
|
||||
./cmd/mesh-controller TestANoMoveTierWaitsForNoNewReport
|
||||
./cmd/mesh-controller TestOneProviderDownAndThreeConsumersFailingAreOneCondition
|
||||
./cmd/mesh-controller TestAConsumerFailingBesideAHealthyProviderIsItsOwn
|
||||
|
||||
Reference in New Issue
Block a user