novox/mesh-controller
6
0
Fork 0
Code Issues Pull Requests 4 Releases Wiki Activity

Merge pull request 'A gate judges its own send and the build it sent, and never puts the controller back behind its store (hq issue 352)' (#185) from fix/a-gate-outlives-the-controller-and-judges-the-build-it-sent into main

Browse Source
This commit was merged in pull request #185.
This commit is contained in:
mesh-admin
2026-10-09 16:10:13 +00:00
parent 65ff6159ad e6e1e3bc89
commit 1c7c385839
16 changed files with 869 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/catalogue/build.go
+26 -5
View File
@@ -173,11 +173,7 @@ func (m Manifest) Resolve(built []Built) (Manifest, error) {
// had — so re-composing a declaration moves nothing, where a commit would move the
// path of an identical binary and recreate everything that reads it.
for key, value := range filled {
text, isText := value.(string)
if !isText || !strings.Contains(text, versionRef) {
continue
}
filled[key] = strings.ReplaceAll(text, versionRef, versionOf(artifact.Digest))
filled[key] = withVersion(value, versionOf(artifact.Digest))
}
default:
return Manifest{}, fmt.Errorf("%s: %q is a %q, and an artifact is %q, %q, %q or %q",
@@ -189,6 +185,31 @@ func (m Manifest) Resolve(built []Built) (Manifest, error) {
return out, nil
}
// withVersion is a resource's value with `${version}` filled: in a string, and in each string of a map —
// a process's env, where the controller is told which build it is (novox/hq issue 352). Anything else is
// left as it is.
func withVersion(value any, version string) any {
switch v := value.(type) {
case string:
if strings.Contains(v, versionRef) {
return strings.ReplaceAll(v, versionRef, version)
}
case map[string]any:
out := make(map[string]any, len(v))
for k, x := range v {
out[k] = withVersion(x, version)
}
return out
case map[string]string:
out := make(map[string]string, len(v))
for k, x := range v {
out[k] = strings.ReplaceAll(x, versionRef, version)
}
return out
}
return value
}
// checkBuild is the manifest's own account of what it builds.
func (b *Build) problems(module string) []string {
if b == nil {
internal/catalogue/version_in_a_path_test.go
+25
View File
@@ -95,3 +95,28 @@ func TestAResourceWithoutAVersionReferenceIsUntouched(t *testing.T) {
t.Fatalf("a path naming no version became %q", path)
}
}
// A process's env can name the build's own version too (novox/hq issue 352): the controller is told which
// build it is, and records what that build reads of the store's schema under it.
func TestAProcessEnvCanNameTheBuildsOwnVersion(t *testing.T) {
m := Manifest{
Module: "mesh-controller",
Build: &Build{Artifacts: []Artifact{{Name: "controller", Kind: ArtifactBundle, Language: "go", System: "arch"}}},
Resources: []map[string]any{{
"id": "controller", "type": "process", "artifact": "controller", "run": []any{"./mesh-controller", "serve"},
"env": map[string]any{"MESH_CONTROLLER_VERSION": "${version}", "OTHER": "kept"},
}},
}
got, err := m.Resolve([]Built{{Name: "controller", Kind: ArtifactBundle,
Reference: "artifact-store://mesh-controller/controller", Digest: aDigest}})
if err != nil {
t.Fatal(err)
}
env, _ := got.Resources[0]["env"].(map[string]any)
if env["MESH_CONTROLLER_VERSION"] != "ad62528c47c7" || env["OTHER"] != "kept" {
t.Fatalf("the env resolved to %v", env)
}
if run, _ := got.Resources[0]["run"].([]any); len(run) != 2 {
t.Fatalf("the run was changed: %v", run)
}
}
internal/inventory/gate.go
+3
View File
@@ -23,6 +23,9 @@ import (
const (
GatePassed = "passed"
GateFailed = "failed"
// GateSuperseded is a judging ended by a later send to the judged machine that moved the module to
// another build (novox/hq issue 352): no verdict on the build, and nothing put back.
GateSuperseded = "superseded"
RollingBack = "rolling-back"
RolledBack = "rolled-back"
internal/inventory/migrations/0087-a-controller-records-what-its-stores-schema-reaches.sql
+10
View File
@@ -0,0 +1,10 @@
-- A controller records, when it serves, how far the store's schema reaches in the build it is (novox/hq
-- issue 352): the highest migration it carries, by its build's version. A gate that fails the controller's
-- build puts the build before it back, and on 2026-10-09 that build was older than the migrations the
-- failed one had applied: it started, said it was behind its own row, and judged the next gate half-blind.
-- A put-back now reads this and keeps the current build when the one before it reaches less than the store.
create table controller_schema (
build text primary key,
reach integer not null,
recorded timestamptz not null default now()
);
internal/inventory/nodes.go
+46 -2
View File
@@ -1054,13 +1054,57 @@ type Reported struct {
// acted on the current words, not merely spoken after they were written. False also covers
// a machine that has not said which, which is every host from before reports carried it.
Current bool
// Declared is the digest of the declaration the last report was about, and ReportedSequence that
// declaration's sequence as the report claimed it (zero from an engine that claims none): what a
// gate holds against the send it made, rather than against the send made last (novox/hq issue 352).
Declared string
ReportedSequence int64
}
// SentDeclaration is what one send carried to a machine, as a gate keeps it: the declaration's digest and
// its sequence (novox/hq issue 352). A report about this declaration, or about one sequenced after it, is a
// report on what the gate sent — whatever the machine was sent since.
type SentDeclaration struct {
Digest string `json:"digest"`
Sequence int64 `json:"sequence,omitempty"`
}
// ReportsOn says a report is about this send: the declaration itself; one the same machine was sequenced
// after it; or the declaration the machine was sent last (Current), which is this send or a later one —
// sends to a machine are made one after another. A send kept without a sequence is matched by its digest
// and by the last send alone.
func (s SentDeclaration) ReportsOn(r Reported) bool {
if r.Current || (s.Digest != "" && r.Declared == s.Digest) {
return true
}
return s.Sequence > 0 && r.ReportedSequence >= s.Sequence
}
// SentTo is the declaration a machine was last sent, by name: its digest and sequence, and false when it
// was never sent one.
func (i *Inventory) SentTo(ctx context.Context, name string) (SentDeclaration, bool, error) {
var digest *string
var seq *int64
err := i.store.Pool().QueryRow(ctx, `select sent, sequence from node where name = $1`, name).Scan(&digest, &seq)
if errors.Is(err, pgx.ErrNoRows) {
return SentDeclaration{}, false, fmt.Errorf("%w: %s", ErrNoSuchNode, name)
}
if err != nil || digest == nil || *digest == "" {
return SentDeclaration{}, false, err
}
s := SentDeclaration{Digest: *digest}
if seq != nil {
s.Sequence = *seq
}
return s, true, nil
}
// LastReports is every machine's last report beside when it was last sent a declaration.
func (i *Inventory) LastReports(ctx context.Context) ([]Reported, error) {
rows, err := i.store.Pool().Query(ctx,
`select n.name, coalesce(r.outcome, ''), r.at, n.sent_at,
r.declared is not null and r.declared <> '' and r.declared = n.sent
r.declared is not null and r.declared <> '' and r.declared = n.sent,
coalesce(r.declared, ''), coalesce(r.reported_sequence, 0)
from node n left join node_report r on r.node = n.id
order by n.name`)
if err != nil {
@@ -1070,7 +1114,7 @@ func (i *Inventory) LastReports(ctx context.Context) ([]Reported, error) {
var out []Reported
for rows.Next() {
var r Reported
if err := rows.Scan(&r.Node, &r.Outcome, &r.At, &r.Sent, &r.Current); err != nil {
if err := rows.Scan(&r.Node, &r.Outcome, &r.At, &r.Sent, &r.Current, &r.Declared, &r.ReportedSequence); err != nil {
return nil, err
}
out = append(out, r)
internal/inventory/plans.go
+4
View File
@@ -126,6 +126,10 @@ type PlanGate struct {
To string `json:"to,omitempty"`
// Since is when the judging began: the first machine reported the new build applied.
Since *time.Time `json:"since,omitempty"`
// Sent is, per machine, the declaration the gate's send carried there (novox/hq issue 352): what a
// machine's report is held against. Absent on a gate kept before it was, which reads the report
// against the send made last, as before.
Sent map[string]SentDeclaration `json:"sent,omitempty"`
// Passes counts the consecutive judgings that found it healthy, LastPass the newest; a judging that
// does not resets them.
Passes int `json:"passes,omitempty"`
internal/inventory/plans_test.go
+80
View File
@@ -115,3 +115,83 @@ func TestTheNewestMergeOfABranchIsTheOneMergedLast(t *testing.T) {
t.Fatalf("one merge time, two plans: %s", p.ID)
}
}
// novox/hq issue 352: what a machine was last sent is read back by name with its sequence, a report keeps
// the declaration it was about and that declaration's sequence, and a gate's sends are kept with the plan.
func TestASendAndAReportAreKnownByTheirDeclaration(t *testing.T) {
inv := ForTest(t)
ctx := t.Context()
record, err := inv.AddNode(ctx, "anchor")
if err != nil {
t.Fatal(err)
}
if _, found, err := inv.SentTo(ctx, "anchor"); err != nil || found {
t.Fatalf("a machine never sent anything: %v %v", found, err)
}
if _, _, err := inv.SentTo(ctx, "nobody"); err == nil {
t.Fatal("a machine that does not exist was answered")
}
seq, err := inv.NextSequence(ctx, record.ID)
if err != nil {
t.Fatal(err)
}
if err := inv.RecordSent(ctx, record.ID, "d-1", map[string]string{"app": "c1"}); err != nil {
t.Fatal(err)
}
sent, found, err := inv.SentTo(ctx, "anchor")
if err != nil || !found || sent.Digest != "d-1" || sent.Sequence != seq {
t.Fatalf("sent %+v %v %v", sent, found, err)
}
if _, err := inv.RecordOrderedDoing(ctx, record.ID, Doing{Node: "anchor", Outcome: OutcomeApplied, Declared: "d-1", Applied: 1},
ReportOrder{Sequence: seq, ReportSequence: 1}, func(ReportOrder) bool { return false }); err != nil {
t.Fatal(err)
}
reports, err := inv.LastReports(ctx)
if err != nil || len(reports) != 1 || reports[0].Declared != "d-1" || reports[0].ReportedSequence != seq || !reports[0].Current {
t.Fatalf("reports %+v %v", reports, err)
}
// Sent again, unreported: the report is no longer on the last send, and is still on the first.
if err := inv.RecordSent(ctx, record.ID, "d-2", map[string]string{"app": "c1"}); err != nil {
t.Fatal(err)
}
reports, _ = inv.LastReports(ctx)
if reports[0].Current || !sent.ReportsOn(reports[0]) {
t.Fatalf("after a newer send: %+v", reports[0])
}
at := time.Now().UTC()
p := Plan{ID: "plan-352", Repository: "novox/x", Commit: "c", Created: at, State: PlanRolling, Tiers: [][]string{{"app"}},
Modules: map[string]*PlanModule{"app": {Gate: &PlanGate{Machines: []string{"anchor"}, Since: &at,
Sent: map[string]SentDeclaration{"anchor": sent}}}}}
if err := inv.SavePlan(ctx, &p); err != nil {
t.Fatal(err)
}
kept, err := inv.PlanByID(ctx, "plan-352")
if err != nil || kept.Modules["app"].Gate.Sent["anchor"] != sent {
t.Fatalf("the gate's send was not kept with the plan: %+v %v", kept.Modules["app"].Gate, err)
}
}
// A controller build's reach of the store's schema is kept by its version, and the store's own is read.
func TestASchemaReachIsKeptByBuild(t *testing.T) {
inv := ForTest(t)
ctx := t.Context()
applied, err := inv.SchemaApplied(ctx)
if err != nil || applied < 87 {
t.Fatalf("applied %d %v", applied, err)
}
if _, known, err := inv.SchemaReachOf(ctx, "ad62528c47c7"); err != nil || known {
t.Fatalf("an unrecorded build: %v %v", known, err)
}
if err := inv.RecordSchemaReach(ctx, "", 87); err == nil {
t.Fatal("a reach without a build was recorded")
}
if err := inv.RecordSchemaReach(ctx, "ad62528c47c7", 86); err != nil {
t.Fatal(err)
}
if err := inv.RecordSchemaReach(ctx, "ad62528c47c7", 87); err != nil {
t.Fatal(err)
}
if reach, known, err := inv.SchemaReachOf(ctx, "ad62528c47c7"); err != nil || !known || reach != 87 {
t.Fatalf("reach %d %v %v", reach, known, err)
}
}
internal/inventory/schema.go
+47
View File
@@ -0,0 +1,47 @@
package inventory
import (
"context"
"errors"
"github.com/jackc/pgx/v5"
)
// What a controller build knows of the store's schema (novox/hq issue 352): the highest migration it
// carries, recorded by its version when it serves, and the highest migration the store has applied. A
// put-back of the controller to a build that reaches less than the store is refused (gate.go), because
// such a controller starts behind its own records and judges with what it can read.
// RecordSchemaReach keeps the highest migration the build serving now carries.
func (i *Inventory) RecordSchemaReach(ctx context.Context, build string, reach int) error {
if build == "" {
return errors.New("a schema reach is recorded by a build's version, and this controller has none")
}
_, err := i.store.Pool().Exec(ctx,
`insert into controller_schema (build, reach) values ($1, $2)
on conflict (build) do update set reach = excluded.reach, recorded = now()`, build, reach)
return err
}
// SchemaReachOf is the highest migration a build carries, as it recorded when it served; false for a
// build that never did.
func (i *Inventory) SchemaReachOf(ctx context.Context, build string) (int, bool, error) {
var reach int
err := i.store.Pool().QueryRow(ctx, `select reach from controller_schema where build = $1`, build).Scan(&reach)
if errors.Is(err, pgx.ErrNoRows) {
return 0, false, nil
}
return reach, err == nil, err
}
// SchemaApplied is the highest migration the store has applied.
func (i *Inventory) SchemaApplied(ctx context.Context) (int, error) {
var n *int
if err := i.store.Pool().QueryRow(ctx, `select max(number) from migration`).Scan(&n); err != nil {
return 0, err
}
if n == nil {
return 0, nil
}
return *n, nil
}