Compare commits
21
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6fe7b7dd6c | ||
|
|
171bddacab | ||
|
|
65496fc4f7 | ||
|
|
4c1d36c9cd | ||
|
|
991589fdd2 | ||
|
|
fa46e7df7d | ||
|
|
18b522184f | ||
|
|
1d4dcdf41a | ||
|
|
f5fd74380c | ||
|
|
e78bbfbdee | ||
|
|
ce7dcccb1d | ||
|
|
42d880f075 | ||
|
|
3553cbaf08 | ||
|
|
d8b19cb332 | ||
|
|
2259d98c5b | ||
|
|
4702c6bd77 | ||
|
|
23f4304d70 | ||
|
|
28ee22ef27 | ||
|
|
f0c5be8c30 | ||
|
|
80a0183567 | ||
|
|
840e56eddb |
+236
@@ -0,0 +1,236 @@
|
||||
{
|
||||
"module": "xterm",
|
||||
"version": "1",
|
||||
"moved": true,
|
||||
"capabilities": [
|
||||
"package-manager"
|
||||
],
|
||||
"requires": [
|
||||
"x11-display"
|
||||
],
|
||||
"claims": [
|
||||
{
|
||||
"name": "node-terminal-emulator",
|
||||
"scope": "node",
|
||||
"serves": [
|
||||
"open"
|
||||
]
|
||||
}
|
||||
],
|
||||
"tools": [
|
||||
"xterm_font",
|
||||
"xterm_colours"
|
||||
],
|
||||
"config": {
|
||||
"face": {
|
||||
"type": "string",
|
||||
"description": "The font family xterm draws with, as fc-list names it: JetBrainsMono Nerd Font, because research 026/04 chose it for its glyphs the prompt uses. A family the machine does not have falls back to the nearest installed face (xterm_font says which).",
|
||||
"pattern": "[A-Za-z0-9 ._-]{1,64}",
|
||||
"default": "JetBrainsMono Nerd Font"
|
||||
},
|
||||
"size": {
|
||||
"type": "number",
|
||||
"description": "The size of xterm's text, in points: 12 reads well on a screen of about 100 DPI, and a denser screen needs a larger size.",
|
||||
"minimum": 6,
|
||||
"maximum": 72,
|
||||
"predefined": "font-size",
|
||||
"default": 12
|
||||
},
|
||||
"save-lines": {
|
||||
"type": "number",
|
||||
"description": "How many lines of scrollback xterm keeps per terminal: 10000 holds a long build log, and each line costs memory in every open terminal.",
|
||||
"minimum": 0,
|
||||
"maximum": 1000000,
|
||||
"default": 10000
|
||||
},
|
||||
"term-name": {
|
||||
"type": "string",
|
||||
"description": "The TERM value xterm sets for the programs it runs: xterm-256color, because the palette has 256 colours and every program knows the name.",
|
||||
"pattern": "[A-Za-z0-9._-]{1,64}",
|
||||
"default": "xterm-256color"
|
||||
},
|
||||
"render-font": {
|
||||
"type": "boolean",
|
||||
"description": "Whether xterm draws with fontconfig faces (true) or the old X core fonts (false): true, because the face is a fontconfig family and core fonts cannot show it.",
|
||||
"default": true
|
||||
},
|
||||
"bold-mode": {
|
||||
"type": "boolean",
|
||||
"description": "Whether xterm draws bold text by overstriking the normal face (true): false, because the face has a real bold style and overstriking blurs it.",
|
||||
"default": false
|
||||
},
|
||||
"always-bold-mode": {
|
||||
"type": "boolean",
|
||||
"description": "Whether xterm overstrikes bold even where the face has a bold style (true): false, for the same reason as bold-mode.",
|
||||
"default": false
|
||||
},
|
||||
"allow-bold-fonts": {
|
||||
"type": "boolean",
|
||||
"description": "Whether xterm may use the face's bold style for bold text (true): true, so bold text is drawn with the real bold weight.",
|
||||
"default": true
|
||||
},
|
||||
"background": {
|
||||
"type": "string",
|
||||
"description": "The terminal's background colour, as #rrggbb: black, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#000000"
|
||||
},
|
||||
"foreground": {
|
||||
"type": "string",
|
||||
"description": "The terminal's text colour, as #rrggbb: a light grey, easier on the eyes than white on black.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#a8a8a8"
|
||||
},
|
||||
"color0": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 0, the normal black, as #rrggbb: #000000, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#000000"
|
||||
},
|
||||
"color1": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 1, the normal red, as #rrggbb: #9e1828, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#9e1828"
|
||||
},
|
||||
"color2": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 2, the normal green, as #rrggbb: #008800, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#008800"
|
||||
},
|
||||
"color3": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 3, the normal yellow, as #rrggbb: #d2bb4b, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#d2bb4b"
|
||||
},
|
||||
"color4": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 4, the normal blue, as #rrggbb: #414171, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#414171"
|
||||
},
|
||||
"color5": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 5, the normal magenta, as #rrggbb: #963c59, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#963c59"
|
||||
},
|
||||
"color6": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 6, the normal cyan, as #rrggbb: #418179, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#418179"
|
||||
},
|
||||
"color7": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 7, the normal white, as #rrggbb: #bebebe, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#bebebe"
|
||||
},
|
||||
"color8": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 8, the bright black, as #rrggbb: #666666, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#666666"
|
||||
},
|
||||
"color9": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 9, the bright red, as #rrggbb: #bc5766, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#bc5766"
|
||||
},
|
||||
"color10": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 10, the bright green, as #rrggbb: #61a171, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#61a171"
|
||||
},
|
||||
"color11": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 11, the bright yellow, as #rrggbb: #e7db52, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#e7db52"
|
||||
},
|
||||
"color12": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 12, the bright blue, as #rrggbb: #5085af, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#5085af"
|
||||
},
|
||||
"color13": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 13, the bright magenta, as #rrggbb: #a97a99, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#a97a99"
|
||||
},
|
||||
"color14": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 14, the bright cyan, as #rrggbb: #6ba4a4, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#6ba4a4"
|
||||
},
|
||||
"color15": {
|
||||
"type": "string",
|
||||
"description": "Palette entry 15, the bright white, as #rrggbb: #ffffff, as both workstations ran it.",
|
||||
"pattern": "#[0-9a-fA-F]{6}",
|
||||
"default": "#ffffff"
|
||||
},
|
||||
"copy-binding": {
|
||||
"type": "string",
|
||||
"description": "The key combination that copies the selection to the clipboard, in X translation syntax: Ctrl Shift ~Alt <Key> C, so a plain Ctrl+C still interrupts the program. It holds no colon, backslash or line break, which would end the resource.",
|
||||
"pattern": "[^:\\\\\\n]+",
|
||||
"default": "Ctrl Shift ~Alt <Key> C"
|
||||
},
|
||||
"paste-binding": {
|
||||
"type": "string",
|
||||
"description": "The key combination that pastes the clipboard, in X translation syntax: Ctrl Shift ~Alt <Key> V, the pair of copy-binding. It holds no colon, backslash or line break, which would end the resource.",
|
||||
"pattern": "[^:\\\\\\n]+",
|
||||
"default": "Ctrl Shift ~Alt <Key> V"
|
||||
}
|
||||
},
|
||||
"items": [
|
||||
{
|
||||
"collection": "environment",
|
||||
"name": "terminal",
|
||||
"attributes": {
|
||||
"kind": "variable",
|
||||
"name": "TERMINAL"
|
||||
},
|
||||
"template": "files/env-terminal"
|
||||
},
|
||||
{
|
||||
"collection": "shell:xresources:normal",
|
||||
"name": "xresources-normal",
|
||||
"template": "files/xresources-normal"
|
||||
}
|
||||
],
|
||||
"resources": [
|
||||
{
|
||||
"id": "package",
|
||||
"type": "package",
|
||||
"package": "xterm"
|
||||
}
|
||||
],
|
||||
"build": {
|
||||
"artifacts": [
|
||||
{
|
||||
"name": "tools",
|
||||
"kind": "bundle",
|
||||
"language": "go",
|
||||
"system": "arch",
|
||||
"from": "cmd/xterm-tools",
|
||||
"binary": "xterm-tools",
|
||||
"loads": [
|
||||
"xterm-tools"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "files",
|
||||
"kind": "archive",
|
||||
"from": "files"
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -148,7 +148,9 @@ func (w *world) manifest(ctx context.Context, get getter, digest string) parsed
|
||||
m, err := delivery.DecodeManifest(value)
|
||||
if err != nil {
|
||||
p = parsed{err: err.Error()}
|
||||
} else if man, err := catalogue.ParseManifest([]byte(m.JSON)); err != nil {
|
||||
} else if man, err := catalogue.ParseManifestForInputs([]byte(m.JSON)); err != nil {
|
||||
// Read as the indexer reads it (issue 1751): a value falling back to a predefined key is judged by the merge gate
|
||||
// in the catalogue, against the predefined.json at its root that this process never holds (ADR 0348 §7).
|
||||
p = parsed{module: m.Module, err: err.Error()}
|
||||
} else {
|
||||
p = parsed{module: m.Module, manifest: man}
|
||||
|
||||
@@ -2,6 +2,7 @@ package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"os"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
@@ -88,3 +89,25 @@ func TestAChangeWaitsForItsManifests(t *testing.T) {
|
||||
t.Errorf("an unreadable manifest: waits %q, change %+v", waits, ch)
|
||||
}
|
||||
}
|
||||
|
||||
// A manifest of the change whose config value falls back to a predefined key (novox/hq ADR 0348 §7) is read, not
|
||||
// unreadable for the catalogue's predefined.json the assigner never holds: the merge gate judges the fallback in the
|
||||
// catalogue, and the indexer reads it the same way (catalogue.ParseManifestForInputs, issue 1751). A group whose
|
||||
// mesh-catalog member touched pacman and xterm failed its composition on this (group check 18df81380db3). The manifest is
|
||||
// xterm's module.json from mesh-catalog at 835bf22b.
|
||||
func TestAManifestFallingBackToAPredefinedKeyIsReadable(t *testing.T) {
|
||||
raw, err := os.ReadFile("testdata/catalogue-835bf22b-xterm.module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !strings.Contains(string(raw), `"predefined"`) {
|
||||
t.Fatal("the fixture names no predefined key")
|
||||
}
|
||||
f := fakeManifests{}
|
||||
xterm := f.add(t, "xterm", string(raw))
|
||||
ch, waits := newWorld().change(context.Background(), f.get, delivery.Candidate{Group: "g", HeadsHash: "0123456789ab",
|
||||
Pulls: []delivery.CandidatePull{pull("novox/mesh-catalog", 1, map[string]delivery.CandidateModule{"xterm": {FP2: fp('b'), Manifest: xterm}})}}, nil)
|
||||
if waits != "" || ch.Unreadable["xterm"] != "" || ch.Manifests["xterm"].Module != "xterm" {
|
||||
t.Errorf("xterm: waits %q, unreadable %q, want read", waits, ch.Unreadable["xterm"])
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-assigner
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024021031-66bf26c0d721
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024125710-bfc3883af980
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,9 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024015135-4784762a70ce h1:9pNnygyhuYIpVs0HPzgheGLNjPXigGGQpnFTp8lw+X4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024015135-4784762a70ce/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024021031-66bf26c0d721 h1:N//dDGmry1F3HKeBTWd+z/SfK+wb8bUUMJF0OlgR0/g=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024021031-66bf26c0d721/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024125710-bfc3883af980 h1:ZEEeYibEPD1rngOAqbODzRY5rUStGC46OpyQwtNCYbs=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024125710-bfc3883af980/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -0,0 +1,185 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// A done key whose blob the artifact store no longer holds is built again, by the builder whole against the bus
|
||||
// (novox/hq issue 1937): on 2026-10-24 the store's 03:30 collection deleted the bare blobs of three done keys the
|
||||
// controller had never taken in, and the builder kept them done.
|
||||
|
||||
// heldVerb answers the controller seat's verb held on the bus, as the serving controller does: {result: {output, ok,
|
||||
// answer}}; a reference is gone while gone says so.
|
||||
type heldVerb struct {
|
||||
mu sync.Mutex
|
||||
gone func(reference string) bool
|
||||
asked []string
|
||||
}
|
||||
|
||||
func (h *heldVerb) serve(t *testing.T, b *bed) {
|
||||
t.Helper()
|
||||
sub, err := b.js.Conn().Subscribe("mesh.seat.mesh-controller.call.held.>", func(m *nats.Msg) {
|
||||
var args struct {
|
||||
References []string `json:"references"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Data, &args)
|
||||
a := delivery.HeldAnswer{Store: "store.test:5100", Held: []string{}, Gone: []string{}}
|
||||
h.mu.Lock()
|
||||
for _, r := range args.References {
|
||||
h.asked = append(h.asked, r)
|
||||
if h.gone(r) {
|
||||
a.Gone = append(a.Gone, r)
|
||||
} else {
|
||||
a.Held = append(a.Held, r)
|
||||
}
|
||||
}
|
||||
h.mu.Unlock()
|
||||
reply, _ := json.Marshal(map[string]any{"result": map[string]any{"output": "held", "ok": true, "answer": a}, "node": "ctl"})
|
||||
_ = m.Respond(reply)
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(func() { _ = sub.Unsubscribe() })
|
||||
}
|
||||
|
||||
func (h *heldVerb) asks() int {
|
||||
h.mu.Lock()
|
||||
defer h.mu.Unlock()
|
||||
return len(h.asked)
|
||||
}
|
||||
|
||||
// eventsConsumer makes EVENTS and the builder's durable consumer of artifacts-gone on it, as the controller makes them
|
||||
// from the manifest's consumes.
|
||||
func eventsConsumer(t *testing.T, b *bed, node string) {
|
||||
t.Helper()
|
||||
if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: eventsStream, Subjects: []string{delivery.ArtifactsGoneSubject}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := b.js.CreateOrUpdateConsumer(b.ctx, eventsStream, jetstream.ConsumerConfig{Durable: node + "_mesh-delivery-builder",
|
||||
AckPolicy: jetstream.AckExplicitPolicy, FilterSubjects: []string{delivery.ArtifactsGoneSubject}}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// doneKey writes builds.<module>.<fp> done, naming one archive at reference, and answers its revision.
|
||||
func doneKey(t *testing.T, b *bed, module, manifest, reference, digest string) uint64 {
|
||||
t.Helper()
|
||||
key, _ := delivery.BuilderBuildKey(module, fp)
|
||||
data, err := delivery.EncodeBuilderBuild(delivery.BuilderBuild{State: delivery.BuildDone, By: "n2.mesh-delivery-builder",
|
||||
Manifest: manifest, Step: "built mesh-build-go on n2",
|
||||
Artifacts: []delivery.BuiltArtifact{{Name: "files", Kind: "archive", Digest: digest, Reference: reference}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
rev, err := b.builds.Put(b.ctx, key, data)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return rev
|
||||
}
|
||||
|
||||
func revisionOf(t *testing.T, b *bed, module string) uint64 {
|
||||
t.Helper()
|
||||
key, _ := delivery.BuilderBuildKey(module, fp)
|
||||
e, err := b.builds.Get(b.ctx, key)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return e.Revision()
|
||||
}
|
||||
|
||||
// 1937/AC3 (integration): the builder, started whole, finds the done key of the trunk's fingerprint naming a blob held
|
||||
// answers gone, builds it again as a new revision whose step says so, and leaves the rebuilt key done.
|
||||
func TestADoneKeyWhoseBlobIsGoneIsBuiltAgainAtStart(t *testing.T) {
|
||||
b := startBed(t)
|
||||
eventsConsumer(t, b, "n1")
|
||||
manifest := b.archive(t, "ca-trust", "modules/ca-trust", commit)
|
||||
old := digestOf("collected last night")
|
||||
oldRef := "artifact-store://ca-trust/files/blobs/" + old
|
||||
rev := doneKey(t, b, "ca-trust", manifest, oldRef, old)
|
||||
h := &heldVerb{gone: func(r string) bool { return r == oldRef }}
|
||||
h.serve(t, b)
|
||||
|
||||
b.start(t, "n1", nil)
|
||||
v := b.await(t, "ca-trust", func(v delivery.BuilderBuild) bool {
|
||||
return v.State == delivery.BuildDone && len(v.Artifacts) == 1 && v.Artifacts[0].Digest == digestOf("files")
|
||||
})
|
||||
if !strings.Contains(v.Step, "built again: the artifact store no longer held "+oldRef) {
|
||||
t.Errorf("the rebuilt key's step %q does not say why it was built again", v.Step)
|
||||
}
|
||||
if got := revisionOf(t, b, "ca-trust"); got <= rev {
|
||||
t.Errorf("revision %d, was %d: want a new revision", got, rev)
|
||||
}
|
||||
if n := b.fake.count("mesh-build-go.build_go_build"); n != 1 {
|
||||
t.Errorf("the archive was built %d times, want once", n)
|
||||
}
|
||||
if h.asks() == 0 {
|
||||
t.Error("the builder never asked held")
|
||||
}
|
||||
}
|
||||
|
||||
// 1937/AC4 (integration): a done key whose blob held answers held at start stays done; when the controller says
|
||||
// artifacts-gone for it, the builder asks again and builds it again.
|
||||
func TestArtifactsGoneLeadsTheBuilderToBuildTheKeyAgain(t *testing.T) {
|
||||
b := startBed(t)
|
||||
eventsConsumer(t, b, "n1")
|
||||
manifest := b.archive(t, "ca-trust", "modules/ca-trust", commit)
|
||||
d := digestOf("files")
|
||||
ref := "artifact-store://ca-trust/files/blobs/" + d
|
||||
rev := doneKey(t, b, "ca-trust", manifest, ref, d)
|
||||
var mu sync.Mutex
|
||||
collected := false
|
||||
h := &heldVerb{gone: func(r string) bool {
|
||||
mu.Lock()
|
||||
defer mu.Unlock()
|
||||
if r == ref && collected {
|
||||
collected = false // gone once: the rebuild pushes it again
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}}
|
||||
h.serve(t, b)
|
||||
|
||||
b.start(t, "n1", nil)
|
||||
for deadline := time.Now().Add(20 * time.Second); h.asks() == 0 && time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
}
|
||||
if h.asks() == 0 {
|
||||
t.Fatal("the builder never asked held at its start")
|
||||
}
|
||||
time.Sleep(300 * time.Millisecond)
|
||||
if got := revisionOf(t, b, "ca-trust"); got != rev || b.fake.count("mesh-build-go.build_go_build") != 0 {
|
||||
t.Fatalf("a key whose blob is held was built again (revision %d, was %d)", got, rev)
|
||||
}
|
||||
|
||||
mu.Lock()
|
||||
collected = true
|
||||
mu.Unlock()
|
||||
key, _ := delivery.BuilderBuildKey("ca-trust", fp)
|
||||
body, err := delivery.EncodeArtifactsGone(delivery.ArtifactsGone{Module: "ca-trust", Key: key, Fingerprint: fp,
|
||||
Gone: []string{ref}, At: time.Now().UTC()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := b.js.Publish(b.ctx, delivery.ArtifactsGoneSubject, body); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
v := b.await(t, "ca-trust", func(v delivery.BuilderBuild) bool {
|
||||
return v.State == delivery.BuildDone && strings.Contains(v.Step, "built again")
|
||||
})
|
||||
if got := revisionOf(t, b, "ca-trust"); got <= rev || b.fake.count("mesh-build-go.build_go_build") != 1 {
|
||||
t.Errorf("build = %+v at revision %d (was %d), %d builds: want built again once on the event", v, got, rev,
|
||||
b.fake.count("mesh-build-go.build_go_build"))
|
||||
}
|
||||
}
|
||||
@@ -26,6 +26,7 @@ import (
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
|
||||
"mesh-delivery-builder/internal/build"
|
||||
)
|
||||
@@ -153,6 +154,17 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
// Each build module's part of a build is an ask on its build seat's stream, pulled by a build node with a free slot
|
||||
// (novox/hq ADR 0525): the builder chooses no node, so it reads neither presence nor have and holds no capacity.
|
||||
asks := buildrun.JetStreamAsks{JS: js}
|
||||
// The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs
|
||||
// (the seat verb held, novox/hq issue 1937) before it leaves the key done.
|
||||
account, err := process.ReadAccount(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
calls, err := process.Connect(account, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer calls.Close()
|
||||
b := &build.Builder{
|
||||
Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
|
||||
Asks: buildrun.Asker{Publisher: asks, Waiter: asks, Bound: bound},
|
||||
@@ -160,6 +172,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
Toolchains: tools,
|
||||
Manifests: manifestReader{manifests},
|
||||
Logf: logger.Printf,
|
||||
Stored: storedVia(calls.Call),
|
||||
}
|
||||
// **Said once every build module answered a reference, taken away when stopped** (issue 1886, ADR 0522): the
|
||||
// controller delivers from this builder's keys only while an instance says it writes references. A bucket not there
|
||||
@@ -183,6 +196,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
// The toolchains first, so a build that starts from the first modules value finds the nodes.
|
||||
watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger)
|
||||
// Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's
|
||||
// assignments are settled first, so the toolchains settle through the gate already knowing them.
|
||||
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
|
||||
watch.put = gate.Toolchain
|
||||
assignErrs, err := watchAssignments(ctx, js, gate, b.NodesChanged, logger)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := watch.start(); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -203,6 +224,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
defer cc.Stop()
|
||||
}
|
||||
// The controller says artifacts-gone when its take-in found a key's blob gone (issue 1937): the builder's durable
|
||||
// consumer on EVENTS, made by the controller from the manifest's consumes. Where it is not there yet the builder says
|
||||
// so; it still asks held at its start and at every module value.
|
||||
switch cons, err := js.Consumer(ctx, eventsStream, cred.Node+"_"+cred.Module); {
|
||||
case err != nil:
|
||||
logger.Printf("the consumer %s_%s on %s does not exist yet, so %s is not heard until the builder starts again: %v",
|
||||
cred.Node, cred.Module, eventsStream, delivery.ArtifactsGoneEvent, err)
|
||||
default:
|
||||
cc, err := cons.Consume(func(m jetstream.Msg) { heardGone(ctx, b, m, logger.Printf) },
|
||||
jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer cc.Stop()
|
||||
}
|
||||
each := func(e jetstream.KeyValueEntry) {
|
||||
if opOf(e) {
|
||||
b.Module(ctx, e.Key(), e.Value())
|
||||
@@ -240,6 +276,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return nil
|
||||
case err := <-errs:
|
||||
return err
|
||||
case err := <-assignErrs:
|
||||
return err
|
||||
case err := <-watch.errs:
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
package main
|
||||
|
||||
// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves
|
||||
// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very
|
||||
// toolchain agreement this file repairs.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
)
|
||||
|
||||
// A build module writes <node>.<language> of its toolchains state on every node it runs on, and nothing deletes the key
|
||||
// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of
|
||||
// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace
|
||||
// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree").
|
||||
//
|
||||
// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a
|
||||
// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments,
|
||||
// key <node>.<module>) hold an open assignment of its build module on its node; it is passed on as deleted otherwise,
|
||||
// and passed on again when the module is assigned there again. Until the assignments are settled (or where they cannot
|
||||
// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that
|
||||
// refusal is the guard of novox/hq ADR 0465, and it stays.
|
||||
type toolchainGate struct {
|
||||
mu sync.Mutex
|
||||
modules map[string]bool // the build modules whose toolchains pass the gate
|
||||
forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds
|
||||
logf func(format string, a ...any)
|
||||
known bool // the assignments are settled
|
||||
standing map[string]bool // <node>.<module> with an open assignment
|
||||
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
|
||||
left map[string]bool // <module>/<key> said as left out, so it is said once
|
||||
}
|
||||
|
||||
func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate {
|
||||
g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]bool{},
|
||||
raw: map[string]map[string][]byte{}, left: map[string]bool{}}
|
||||
for _, m := range modules {
|
||||
g.modules[m] = true
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed.
|
||||
func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.raw[module] == nil {
|
||||
g.raw[module] = map[string][]byte{}
|
||||
}
|
||||
if deleted {
|
||||
delete(g.raw[module], key)
|
||||
} else {
|
||||
g.raw[module][key] = value
|
||||
}
|
||||
return g.pass(module, key)
|
||||
}
|
||||
|
||||
// Assignment takes one change of the controller's assignments in and answers whether what it passed on changed.
|
||||
func (g *toolchainGate) Assignment(key string, value []byte, deleted bool) bool {
|
||||
node, module, ok := strings.Cut(key, ".")
|
||||
if !ok || !g.modules[module] {
|
||||
return false
|
||||
}
|
||||
open := !deleted && openAssignment(value)
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.standing[key] == open {
|
||||
return false
|
||||
}
|
||||
if open {
|
||||
g.standing[key] = true
|
||||
} else {
|
||||
delete(g.standing, key)
|
||||
}
|
||||
if !g.known {
|
||||
return false
|
||||
}
|
||||
changed := false
|
||||
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
||||
if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node {
|
||||
changed = g.pass(module, k) || changed
|
||||
}
|
||||
}
|
||||
return changed
|
||||
}
|
||||
|
||||
// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it
|
||||
// passed on changed.
|
||||
func (g *toolchainGate) Settled() bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
g.known = true
|
||||
changed := false
|
||||
for _, module := range slices.Sorted(maps.Keys(g.raw)) {
|
||||
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
||||
changed = g.pass(module, k) || changed
|
||||
}
|
||||
}
|
||||
return changed
|
||||
}
|
||||
|
||||
// pass hands one key on as it stands: its value when its node is assigned (or the assignments are not known), deleted
|
||||
// otherwise. The caller holds mu.
|
||||
func (g *toolchainGate) pass(module, key string) bool {
|
||||
value, held := g.raw[module][key]
|
||||
if !held {
|
||||
return g.forward(module, key, nil, true)
|
||||
}
|
||||
node, _, err := delivery.ParseToolchainKey(key)
|
||||
if err != nil || !g.known || g.standing[node+"."+module] {
|
||||
delete(g.left, module+"/"+key)
|
||||
return g.forward(module, key, value, false)
|
||||
}
|
||||
if !g.left[module+"/"+key] {
|
||||
g.left[module+"/"+key] = true
|
||||
g.logf("%s.%s is left out of the agreed toolchain: %s has no assignment on %s, so it is no builder (issue 1940)",
|
||||
delivery.ToolchainsBucket(module), key, module, node)
|
||||
}
|
||||
return g.forward(module, key, nil, true)
|
||||
}
|
||||
|
||||
// openAssignment is whether a value of the controller's assignments holds an assignment that stands: one that has not
|
||||
// ended and has no state (a pending one is not on the machine yet; mesh-controller internal/chain Assignment).
|
||||
func openAssignment(value []byte) bool {
|
||||
var a struct {
|
||||
Ended time.Time `json:"ended"`
|
||||
State string `json:"state"`
|
||||
}
|
||||
if json.Unmarshal(value, &a) != nil {
|
||||
return false
|
||||
}
|
||||
return a.Ended.IsZero() && a.State == ""
|
||||
}
|
||||
|
||||
// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is
|
||||
// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read
|
||||
// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel.
|
||||
func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) {
|
||||
errs := make(chan error, 1)
|
||||
kv, err := js.KeyValue(ctx, provider.AssignmentsBucket)
|
||||
if err != nil {
|
||||
// Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery).
|
||||
logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
|
||||
return errs, nil
|
||||
}
|
||||
each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e)) }
|
||||
w, err := settle(ctx, kv, each)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
|
||||
}
|
||||
if gate.Settled() {
|
||||
changed(ctx)
|
||||
}
|
||||
go func() {
|
||||
errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) {
|
||||
if gate.Assignment(e.Key(), e.Value(), !opOf(e)) {
|
||||
changed(ctx)
|
||||
}
|
||||
})
|
||||
}()
|
||||
return errs, nil
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
package main
|
||||
|
||||
// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940).
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
func gateReport(t *testing.T, node, version string) (string, []byte) {
|
||||
t.Helper()
|
||||
key, err := delivery.ToolchainKey(node, "typescript")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return key, value
|
||||
}
|
||||
|
||||
func agreedOf(set *buildrun.ToolchainSet) (string, string) {
|
||||
tc, none := delivery.AgreedToolchain(set.Entries("typescript"))
|
||||
return tc.Version, none
|
||||
}
|
||||
|
||||
// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the
|
||||
// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them
|
||||
// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again.
|
||||
func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
var said []string
|
||||
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) })
|
||||
for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} {
|
||||
key, value := gateReport(t, node, version)
|
||||
gate.Toolchain(buildrun.NodeBuildModule, key, value, false)
|
||||
}
|
||||
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") {
|
||||
t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none)
|
||||
}
|
||||
for _, node := range []string{"g14", "novox", "shanks"} {
|
||||
gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false)
|
||||
}
|
||||
gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false)
|
||||
gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false) // another module's: no matter
|
||||
if !gate.Settled() {
|
||||
t.Fatal("settling the assignments left ace's report out, so what was passed on changed")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none)
|
||||
}
|
||||
if len(said) != 1 {
|
||||
t.Fatalf("ace's report is said left out once; said %v", said)
|
||||
}
|
||||
// ace reports again while unassigned (an old value seen again): still left out.
|
||||
key, value := gateReport(t, "ace", "v26.12.0")
|
||||
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("an unassigned node's report changed what was passed on")
|
||||
}
|
||||
// Assigned to ace again: its report counts, and the builders really disagree: the guard stays.
|
||||
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false) {
|
||||
t.Fatal("assigning ace again passed nothing on")
|
||||
}
|
||||
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") {
|
||||
t.Fatalf("builders that run different versions still disagree; got %q", none)
|
||||
}
|
||||
// The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again.
|
||||
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false) {
|
||||
t.Fatal("a pending assignment is no assignment on the machine")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("got %q, %q", v, none)
|
||||
}
|
||||
gate.Assignment("ace.mesh-build-node", nil, true)
|
||||
if v, _ := agreedOf(set); v != "v26.12.1" {
|
||||
t.Fatalf("got %q", v)
|
||||
}
|
||||
// A report deleted from the bucket goes whether its node is assigned or not.
|
||||
key, _ = gateReport(t, "g14", "v26.12.1")
|
||||
if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) {
|
||||
t.Fatal("deleting g14's report passed nothing on")
|
||||
}
|
||||
if n := len(set.Entries("typescript")); n != 2 {
|
||||
t.Fatalf("novox and shanks are left; got %d entries", n)
|
||||
}
|
||||
}
|
||||
|
||||
// A newly assigned build node that has not reported yet changes nothing; its report counts as it comes.
|
||||
func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(string, ...any) {})
|
||||
gate.Settled()
|
||||
if gate.Assignment("g14.mesh-build-node", []byte(`{}`), false) {
|
||||
t.Fatal("an assignment with no report passed something on")
|
||||
}
|
||||
key, value := gateReport(t, "g14", "v26.12.1")
|
||||
if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("an assigned node's report was not passed on")
|
||||
}
|
||||
key, value = gateReport(t, "ace", "v26.12.0")
|
||||
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("a report of a node with no assignment key was passed on after the assignments settled")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("got %q, %q", v, none)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"mesh-delivery-builder/internal/build"
|
||||
)
|
||||
|
||||
// eventsStream is the stream every module's and role's events are on, where the controller makes the builder's durable
|
||||
// consumer of artifacts-gone (mesh-controller internal/broker EventsStream).
|
||||
const eventsStream = "EVENTS"
|
||||
|
||||
// toolCaller calls a tool or a seat verb as the builder's own account (mesh-sdk process.Conn.Call).
|
||||
type toolCaller func(key string, body any) (json.RawMessage, error)
|
||||
|
||||
// storedVia is build.Builder's Stored: it asks the controller's verb held (mesh-sdk delivery.HeldVerb; novox/hq issue
|
||||
// 1937) and answers the references the store no longer holds. One it could not ask about, with none gone, is an error:
|
||||
// the key is then left done, and the controller's take-in still refuses a gone blob.
|
||||
func storedVia(call toolCaller) func(ctx context.Context, references []string) ([]string, error) {
|
||||
return func(_ context.Context, references []string) ([]string, error) {
|
||||
raw, err := call(delivery.HeldVerb, map[string]any{"references": references})
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("asking %s: %w", delivery.HeldVerb, err)
|
||||
}
|
||||
a, err := delivery.DecodeHeldAnswer(raw, references)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if len(a.Gone) == 0 && len(a.Unknown) > 0 {
|
||||
var why []string
|
||||
for r, w := range a.Unknown {
|
||||
why = append(why, r+": "+w)
|
||||
}
|
||||
sort.Strings(why)
|
||||
return nil, fmt.Errorf("the artifact store could not be asked about %s", strings.Join(why, "; "))
|
||||
}
|
||||
return a.Gone, nil
|
||||
}
|
||||
}
|
||||
|
||||
// heardGone takes one artifacts-gone event in: acknowledged, then handed to the builder, which asks the store and builds
|
||||
// the key again. A body that is not one is said and acknowledged, so it is not handed over again.
|
||||
func heardGone(ctx context.Context, b *build.Builder, m jetstream.Msg, logf func(string, ...any)) {
|
||||
e, err := delivery.DecodeArtifactsGone(m.Data())
|
||||
if ackErr := m.Ack(); ackErr != nil {
|
||||
logf("acknowledging %s: %v", m.Subject(), ackErr)
|
||||
}
|
||||
if err != nil {
|
||||
logf("%s holds what is not %s, left: %v", m.Subject(), delivery.ArtifactsGoneEvent, err)
|
||||
return
|
||||
}
|
||||
b.ArtifactsGone(ctx, e)
|
||||
}
|
||||
@@ -23,9 +23,11 @@ var toolchainStreams = map[string]string{buildrun.GoBuildModule: "TOOLCHAINS_GO"
|
||||
// §8): it watches the buckets that exist at start, and a build module says each time it wrote its toolchains key
|
||||
// (toolchain.reported, handled by Reported); a bucket not watched yet is then watched.
|
||||
type toolchainWatch struct {
|
||||
ctx context.Context
|
||||
js bucketOpener
|
||||
set *buildrun.ToolchainSet
|
||||
ctx context.Context
|
||||
js bucketOpener
|
||||
set *buildrun.ToolchainSet
|
||||
// put takes a change of a toolchains bucket in: set.Put, or the gate of the assigned build nodes before it (issue 1940).
|
||||
put func(module, key string, value []byte, deleted bool) bool
|
||||
changed func(ctx context.Context) // a language's entries changed: the failed builds are tried again
|
||||
logger *log.Logger
|
||||
errs chan error
|
||||
@@ -34,7 +36,7 @@ type toolchainWatch struct {
|
||||
}
|
||||
|
||||
func newToolchainWatch(ctx context.Context, js bucketOpener, set *buildrun.ToolchainSet, changed func(context.Context), logger *log.Logger) *toolchainWatch {
|
||||
return &toolchainWatch{ctx: ctx, js: js, set: set, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}}
|
||||
return &toolchainWatch{ctx: ctx, js: js, set: set, put: set.Put, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}}
|
||||
}
|
||||
|
||||
// start watches the buckets that exist now; a missing one is said and waits for its build module's event.
|
||||
@@ -62,14 +64,14 @@ func (w *toolchainWatch) ensure(module string) error {
|
||||
case err != nil:
|
||||
return fmt.Errorf("the bucket %s: %w", name, err)
|
||||
}
|
||||
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.set.Put(module, e.Key(), e.Value(), !opOf(e)) })
|
||||
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) })
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
w.watched[module] = true
|
||||
go func() {
|
||||
w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) {
|
||||
if w.set.Put(module, e.Key(), e.Value(), !opOf(e)) {
|
||||
if w.put(module, e.Key(), e.Value(), !opOf(e)) {
|
||||
w.changed(w.ctx)
|
||||
}
|
||||
})
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-builder
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468 h1:UcMee+IV2ii5E4dRZdQUaSaeh9IedTFYumuQbBiB0k0=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023150321-a2c813a6b468/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1 h1:fzOkrz9CSp3mEjeV7NfZx8yoivI1sD1AOtIPtyyF8RI=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024020722-34f210e5eba1/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
@@ -37,6 +38,11 @@ type Claim struct {
|
||||
Key string
|
||||
Rev uint64
|
||||
Build delivery.BuilderBuild
|
||||
// Gone are the references of a done key the artifact store no longer holds, for which it was claimed again
|
||||
// (ClaimOptions.Stored; issue 1937): said in the log and the key's step.
|
||||
Gone []string
|
||||
// Unasked is why the store could not be asked about a done key taken as it stands to publish: said in the log.
|
||||
Unasked error
|
||||
}
|
||||
|
||||
// ClaimOptions say what the builder knows about the moment of a claim.
|
||||
@@ -55,6 +61,30 @@ type ClaimOptions struct {
|
||||
// a package's reference. A key queued waiting for a version (BuilderBuild.Waits) is taken only once it is (ADR 0500
|
||||
// clause 6). Nil means none is.
|
||||
Published func(ref string) bool
|
||||
// Stored answers which of the references given the artifact store no longer holds (the controller's verb held,
|
||||
// mesh-sdk delivery.HeldVerb; novox/hq issue 1937). A done key of the trunk's fingerprint is asked before it is left
|
||||
// done: the store's nightly collector deletes a blob nothing holds, and a key the controller never took in holds
|
||||
// nothing, so a key whose blob is gone is claimed building as a new revision and built again. Nil asks nothing.
|
||||
Stored func(ctx context.Context, references []string) (gone []string, err error)
|
||||
}
|
||||
|
||||
// StoredReferences are the references of a done build's bundles and archives that name a blob in the artifact store
|
||||
// (`artifact-store://<module>/<artifact>/blobs/<digest>`): what the store's collector can delete. An image or an upstream
|
||||
// copy is a manifest, which the collector keeps, and a package lives in the package registry.
|
||||
func StoredReferences(b delivery.BuilderBuild) []string {
|
||||
var out []string
|
||||
for _, a := range b.Artifacts {
|
||||
if strings.HasPrefix(a.Reference, "artifact-store://") && strings.Contains(a.Reference, "/blobs/") &&
|
||||
!slices.Contains(out, a.Reference) {
|
||||
out = append(out, a.Reference)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// GoneStep is the step a key claimed again for gone blobs says (issue 1937).
|
||||
func GoneStep(gone []string) string {
|
||||
return "claimed again: the artifact store no longer holds " + strings.Join(gone, ", ")
|
||||
}
|
||||
|
||||
// Unpublished is whether a done build holds an image built without a push or a package built without a publish: it has
|
||||
@@ -118,6 +148,8 @@ func Unreferenced(b delivery.BuilderBuild) bool {
|
||||
// names it (ClaimOptions.Trunk): that is taken as it stands, with the done value, and its builder writes the new
|
||||
// revision when the publishing build ends (ADR 0498 clause 6). From R4e one node builds each fingerprint (ADR 0478
|
||||
// clause 4), so a key built on one node is never built again for a second;
|
||||
// - a done key of the trunk's fingerprint whose bundle or archive blob the artifact store no longer holds
|
||||
// (ClaimOptions.Stored, issue 1937) is claimed building as a new revision and built whole, its step saying which;
|
||||
// - a failed key is taken again as a new revision of the same key (ADR 0429 clause 18), except one failed for a kind no
|
||||
// builder builds, for an upstream pinned by no digest or a package version taken (delivery.FailedUntilChanged: ADR 0478
|
||||
// clause 2, ADR 0498 clauses 4 and 5) or for digests that differ (buildrun.DigestsDifferPrefix, clause 4),
|
||||
@@ -149,6 +181,28 @@ func (c Claimer) Claim(ctx context.Context, module, fingerprint, manifest string
|
||||
if Unreferenced(cur) {
|
||||
return c.write(ctx, key, building, func(data []byte) (uint64, error) { return c.Store.Update(ctx, key, data, rev) })
|
||||
}
|
||||
if opt.Trunk && opt.Stored != nil {
|
||||
if refs := StoredReferences(cur); len(refs) > 0 {
|
||||
gone, err := opt.Stored(ctx, refs)
|
||||
if err != nil {
|
||||
err = fmt.Errorf("%s: whether the artifact store still holds %s cannot be asked, so the key is "+
|
||||
"left done (the controller's take-in refuses it if one is gone): %w", key, strings.Join(refs, ", "), err)
|
||||
if len(publishable(cur, opt.ImageBuilder)) > 0 {
|
||||
return &Claim{Key: key, Rev: rev, Build: cur, Unasked: err}, nil
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
if len(gone) > 0 {
|
||||
again := building
|
||||
again.Step = GoneStep(gone)
|
||||
cl, err := c.write(ctx, key, again, func(data []byte) (uint64, error) { return c.Store.Update(ctx, key, data, rev) })
|
||||
if cl != nil {
|
||||
cl.Gone = gone
|
||||
}
|
||||
return cl, err
|
||||
}
|
||||
}
|
||||
}
|
||||
if opt.Trunk && len(publishable(cur, opt.ImageBuilder)) > 0 {
|
||||
return &Claim{Key: key, Rev: rev, Build: cur}, nil
|
||||
}
|
||||
|
||||
@@ -0,0 +1,204 @@
|
||||
package build
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// A done key whose blob the artifact store no longer holds is built again (novox/hq issue 1937): on 2026-10-24 the
|
||||
// store's 03:30 collection deleted the bare blobs of node-tools', mesh-host's and the forge bridge's done keys, which the
|
||||
// controller had never taken in, and the builder kept them done.
|
||||
|
||||
// fakeHeld answers the controller's verb held: the references in gone are gone.
|
||||
type fakeHeld struct {
|
||||
mu sync.Mutex
|
||||
gone map[string]bool
|
||||
err error
|
||||
asked [][]string
|
||||
}
|
||||
|
||||
func (f *fakeHeld) stored(_ context.Context, refs []string) ([]string, error) {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
f.asked = append(f.asked, slices.Clone(refs))
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
var out []string
|
||||
for _, r := range refs {
|
||||
if f.gone[r] {
|
||||
out = append(out, r)
|
||||
}
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (f *fakeHeld) asks() int {
|
||||
f.mu.Lock()
|
||||
defer f.mu.Unlock()
|
||||
return len(f.asked)
|
||||
}
|
||||
|
||||
// logged keeps what the builder says.
|
||||
type logged struct {
|
||||
mu sync.Mutex
|
||||
lines []string
|
||||
}
|
||||
|
||||
func (l *logged) logf(format string, args ...any) {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
l.lines = append(l.lines, fmt.Sprintf(format, args...))
|
||||
}
|
||||
|
||||
func (l *logged) has(words ...string) bool {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
for _, line := range l.lines {
|
||||
all := true
|
||||
for _, w := range words {
|
||||
all = all && strings.Contains(line, w)
|
||||
}
|
||||
if all {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
const goneRef = "artifact-store://m/tool/blobs/" + digest
|
||||
|
||||
func doneKeyNaming(ref string) delivery.BuilderBuild {
|
||||
return delivery.BuilderBuild{State: delivery.BuildDone, By: "shanks.builder", Manifest: manifestD, Step: "built mesh-build-go on shanks",
|
||||
Artifacts: []delivery.BuiltArtifact{{Name: "tool", Kind: "bundle", Digest: digest, Reference: ref}}}
|
||||
}
|
||||
|
||||
// 1937/AC3 (unit, the claim): a done key of the trunk's fingerprint whose blob is gone is claimed building as a new
|
||||
// revision, its step naming the gone reference; a candidate's done key is not asked about.
|
||||
func TestADoneTrunkKeyWhoseBlobIsGoneIsClaimedAgain(t *testing.T) {
|
||||
st := newMem()
|
||||
c := Claimer{Store: st, By: "novox.builder"}
|
||||
key, _ := delivery.BuilderBuildKey("m", fp1)
|
||||
rev := putBuild(t, st, key, doneKeyNaming(goneRef))
|
||||
held := &fakeHeld{gone: map[string]bool{goneRef: true}}
|
||||
|
||||
cl, err := c.Claim(t.Context(), "m", fp1, manifestD, ClaimOptions{Stored: held.stored})
|
||||
if err != nil || cl != nil || held.asks() != 0 {
|
||||
t.Fatalf("a candidate's done key: claim %+v, %v, %d asks; want nothing and no ask", cl, err, held.asks())
|
||||
}
|
||||
cl, err = c.Claim(t.Context(), "m", fp1, manifestD, ClaimOptions{Trunk: true, Stored: held.stored})
|
||||
if err != nil || cl == nil {
|
||||
t.Fatalf("claim = %+v, %v; want the key claimed again", cl, err)
|
||||
}
|
||||
if cl.Rev <= rev || cl.Build.State != delivery.BuildBuilding || !slices.Equal(cl.Gone, []string{goneRef}) {
|
||||
t.Errorf("claim = %+v (revision was %d); want building as a new revision naming the gone reference", cl, rev)
|
||||
}
|
||||
v := valueOf(t, st, key)
|
||||
if v.State != delivery.BuildBuilding || !strings.Contains(v.Step, "no longer holds") || !strings.Contains(v.Step, goneRef) {
|
||||
t.Errorf("the key holds %+v; want building, its step naming the gone reference", v)
|
||||
}
|
||||
}
|
||||
|
||||
// 1937/AC5 (unit, the claim): a done trunk key whose blobs are all held stays done; a store that cannot be asked leaves
|
||||
// it done and says why; images and packages are not asked about.
|
||||
func TestADoneTrunkKeyStaysDoneWhileTheStoreHoldsItsBlobs(t *testing.T) {
|
||||
st := newMem()
|
||||
c := Claimer{Store: st, By: "novox.builder"}
|
||||
key, _ := delivery.BuilderBuildKey("m", fp1)
|
||||
b := doneKeyNaming(goneRef)
|
||||
b.Artifacts = append(b.Artifacts, delivery.BuiltArtifact{Name: "img", Kind: "image", Digest: digest, Reference: "registry.example/m/img@" + digest})
|
||||
rev := putBuild(t, st, key, b)
|
||||
held := &fakeHeld{gone: map[string]bool{}}
|
||||
if cl, err := c.Claim(t.Context(), "m", fp1, manifestD, ClaimOptions{Trunk: true, Stored: held.stored}); err != nil || cl != nil {
|
||||
t.Fatalf("claim = %+v, %v; want nothing: the store holds the blob", cl, err)
|
||||
}
|
||||
if len(held.asked) != 1 || !slices.Equal(held.asked[0], []string{goneRef}) {
|
||||
t.Errorf("asked %v; want the bundle's reference alone", held.asked)
|
||||
}
|
||||
held.err = errors.New("the controller did not answer")
|
||||
cl, err := c.Claim(t.Context(), "m", fp1, manifestD, ClaimOptions{Trunk: true, Stored: held.stored})
|
||||
if cl != nil || err == nil || !strings.Contains(err.Error(), "cannot be asked") || !strings.Contains(err.Error(), "left done") {
|
||||
t.Errorf("claim = %+v, %v; want the key left done and said", cl, err)
|
||||
}
|
||||
if st.revs[key] != rev {
|
||||
t.Errorf("the key moved to revision %d (was %d)", st.revs[key], rev)
|
||||
}
|
||||
}
|
||||
|
||||
// 1937/AC3 (unit, the builder): the builder's look at a module whose done key's blob is gone builds it again, says so
|
||||
// in a log line and in the done key's step, and asks the store once per process while it answers held.
|
||||
func TestTheBuilderBuildsAgainADoneKeyWhoseBlobIsGone(t *testing.T) {
|
||||
r := newRig(t, goBundle)
|
||||
log := &logged{}
|
||||
r.b.Logf = log.logf
|
||||
held := &fakeHeld{gone: map[string]bool{goneRef: true}}
|
||||
r.b.Stored = held.stored
|
||||
rev := putBuild(t, r.st, r.key, doneKeyNaming(goneRef))
|
||||
v := r.consider(t, "modules/m")
|
||||
if v.State != delivery.BuildDone || r.st.revs[r.key] <= rev || len(r.bm.callsOf("build_go_build")) == 0 {
|
||||
t.Fatalf("build = %+v at revision %d (was %d); want built again", v, r.st.revs[r.key], rev)
|
||||
}
|
||||
if !strings.Contains(v.Step, "built again: the artifact store no longer held "+goneRef) {
|
||||
t.Errorf("the done key's step %q does not say it was built again for the gone blob", v.Step)
|
||||
}
|
||||
if !log.has(r.key, "built again", goneRef) {
|
||||
t.Errorf("no log line says the key was built again: %v", log.lines)
|
||||
}
|
||||
// The new revision's blob is held: asked once, then not again in this process.
|
||||
held.mu.Lock()
|
||||
held.gone = map[string]bool{}
|
||||
held.mu.Unlock()
|
||||
rev = r.st.revs[r.key]
|
||||
asks := held.asks()
|
||||
r.consider(t, "modules/m")
|
||||
r.consider(t, "modules/m")
|
||||
if r.st.revs[r.key] != rev {
|
||||
t.Errorf("a key whose blob is held was built again")
|
||||
}
|
||||
if held.asks() != asks+1 {
|
||||
t.Errorf("asked the store %d times for two looks at a held key; want once", held.asks()-asks)
|
||||
}
|
||||
}
|
||||
|
||||
// 1937/AC4 (unit, the builder): artifacts-gone for the module's fingerprint makes the builder ask again, even for a key
|
||||
// it found held before, and build it again; one for another fingerprint builds nothing.
|
||||
func TestArtifactsGoneMakesTheBuilderAskAndBuildAgain(t *testing.T) {
|
||||
r := newRig(t, goBundle)
|
||||
log := &logged{}
|
||||
r.b.Logf = log.logf
|
||||
held := &fakeHeld{gone: map[string]bool{}}
|
||||
r.b.Stored = held.stored
|
||||
putBuild(t, r.st, r.key, doneKeyNaming(goneRef))
|
||||
if v := r.consider(t, "modules/m"); v.State != delivery.BuildDone || len(r.bm.callsOf("build_go_build")) != 0 {
|
||||
t.Fatalf("a key whose blob is held was built: %+v", v)
|
||||
}
|
||||
rev := r.st.revs[r.key]
|
||||
|
||||
held.mu.Lock()
|
||||
held.gone[goneRef] = true // collected after the builder's look; the controller's take-in found it
|
||||
held.mu.Unlock()
|
||||
r.b.ArtifactsGone(context.Background(), delivery.ArtifactsGone{Event: delivery.ArtifactsGoneEvent, Module: "m",
|
||||
Fingerprint: "src2:" + strings.Repeat("9", 64), Gone: []string{goneRef}})
|
||||
r.b.Wait()
|
||||
if r.st.revs[r.key] != rev || !log.has("a fingerprint the module does not have now") {
|
||||
t.Fatalf("an event for another fingerprint moved the key (revision %d, was %d): %v", r.st.revs[r.key], rev, log.lines)
|
||||
}
|
||||
|
||||
r.b.ArtifactsGone(context.Background(), delivery.ArtifactsGone{Event: delivery.ArtifactsGoneEvent, Module: "m",
|
||||
Key: r.key, Fingerprint: fp1, Gone: []string{goneRef}})
|
||||
r.b.Wait()
|
||||
v := valueOf(t, r.st, r.key)
|
||||
if v.State != delivery.BuildDone || r.st.revs[r.key] <= rev || len(r.bm.callsOf("build_go_build")) == 0 {
|
||||
t.Fatalf("build = %+v at revision %d (was %d); want built again on the event", v, r.st.revs[r.key], rev)
|
||||
}
|
||||
if !log.has("heard artifacts-gone") || !log.has(r.key, "built again", goneRef) {
|
||||
t.Errorf("the builder did not say what it heard and did: %v", log.lines)
|
||||
}
|
||||
}
|
||||
@@ -50,6 +50,9 @@ type Builder struct {
|
||||
// Answered is told the build module whose step answered a reference (issue 1886): what the builder says it writes
|
||||
// rests on it. Nil tells nobody.
|
||||
Answered func(buildModule string)
|
||||
// Stored answers which references the artifact store no longer holds (the controller's verb held; issue 1937): a done
|
||||
// key of the trunk's fingerprint is asked before it is left done (ClaimOptions.Stored). Nil asks nothing.
|
||||
Stored func(ctx context.Context, references []string) (gone []string, err error)
|
||||
|
||||
once sync.Once
|
||||
sem chan struct{}
|
||||
@@ -66,6 +69,9 @@ type Builder struct {
|
||||
// published are the package versions `<name>@<version>` a done key of this bucket names as a package's reference: a key
|
||||
// waiting for one is built once it is here (ADR 0500 clause 6).
|
||||
published map[string]bool
|
||||
// held are the done keys' references the store answered held, by `<key> <references>`, asked once per process: a
|
||||
// blob collected later is found by the take-in, whose artifacts-gone forgets the key here (issue 1937).
|
||||
held map[string]bool
|
||||
}
|
||||
|
||||
type planned struct {
|
||||
@@ -82,7 +88,7 @@ func (b *Builder) init() {
|
||||
b.sem = make(chan struct{}, limit)
|
||||
b.cands, b.via, b.trunkFP = map[string]delivery.Candidate{}, map[string]via{}, map[string]bool{}
|
||||
b.known, b.inflight, b.again, b.plans, b.tried = map[string]delivery.Module{}, map[string]bool{}, map[string]delivery.Module{}, map[string]planned{}, map[string]bool{}
|
||||
b.published = map[string]bool{}
|
||||
b.published, b.held = map[string]bool{}, map[string]bool{}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -202,6 +208,55 @@ func (b *Builder) Forget(key string) {
|
||||
b.mu.Unlock()
|
||||
}
|
||||
|
||||
// storedOnce is Stored for one key, asked once per process while the store answers every reference held.
|
||||
func (b *Builder) storedOnce(key string) func(ctx context.Context, references []string) ([]string, error) {
|
||||
if b.Stored == nil {
|
||||
return nil
|
||||
}
|
||||
return func(ctx context.Context, references []string) ([]string, error) {
|
||||
id := key + " " + strings.Join(references, " ")
|
||||
b.mu.Lock()
|
||||
known := b.held[id]
|
||||
b.mu.Unlock()
|
||||
if known {
|
||||
return nil, nil
|
||||
}
|
||||
gone, err := b.Stored(ctx, references)
|
||||
if err == nil && len(gone) == 0 {
|
||||
b.mu.Lock()
|
||||
b.held[id] = true
|
||||
b.mu.Unlock()
|
||||
}
|
||||
return gone, err
|
||||
}
|
||||
}
|
||||
|
||||
// ArtifactsGone takes the controller's event artifacts-gone in (issue 1937): a take-in found that the store does not
|
||||
// hold what the module's key names. The module is looked at again, which asks the store (ClaimOptions.Stored) and
|
||||
// builds the key again; an event for a fingerprint the module no longer has is said and left. The event only wakes the
|
||||
// builder: what is built rests on the store's answer, never on the event alone.
|
||||
func (b *Builder) ArtifactsGone(ctx context.Context, e delivery.ArtifactsGone) {
|
||||
b.init()
|
||||
b.mu.Lock()
|
||||
m, ok := b.known[e.Module]
|
||||
if key, err := delivery.BuilderBuildKey(e.Module, e.Fingerprint); err == nil {
|
||||
for id := range b.held {
|
||||
if strings.HasPrefix(id, key+" ") {
|
||||
delete(b.held, id)
|
||||
}
|
||||
}
|
||||
}
|
||||
b.mu.Unlock()
|
||||
if !ok || m.FP2 != e.Fingerprint {
|
||||
b.logf("%s: heard %s for %s, a fingerprint the module does not have now: nothing to build", e.Module,
|
||||
delivery.ArtifactsGoneEvent, e.Fingerprint)
|
||||
return
|
||||
}
|
||||
b.logf("%s: heard %s (%s): asking the artifact store and building the key again", e.Module,
|
||||
delivery.ArtifactsGoneEvent, strings.Join(e.Gone, ", "))
|
||||
b.consider(ctx, e.Module, m)
|
||||
}
|
||||
|
||||
// Reconsider looks at every module again: the toolchains changed, so a build that failed for them may build now.
|
||||
func (b *Builder) Reconsider(ctx context.Context) {
|
||||
b.init()
|
||||
@@ -312,7 +367,8 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) {
|
||||
first := !b.tried[key]
|
||||
b.tried[key] = true
|
||||
b.mu.Unlock()
|
||||
cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished})
|
||||
cl, err := b.Claimer.Claim(ctx, module, m.FP2, m.Manifest, ClaimOptions{TakeOver: first, ImageBuilder: imageSeat, Trunk: trunk, Published: b.isPublished,
|
||||
Stored: b.storedOnce(key)})
|
||||
if err != nil {
|
||||
b.logf("%s: %v", module, err)
|
||||
return
|
||||
@@ -320,6 +376,13 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) {
|
||||
if cl == nil {
|
||||
return
|
||||
}
|
||||
if cl.Unasked != nil {
|
||||
b.logf("%v", cl.Unasked)
|
||||
}
|
||||
if len(cl.Gone) > 0 {
|
||||
// Said aloud (issue 1937): the key was done, and the store's collector deleted what it names.
|
||||
b.logf("%s: built again: the artifact store no longer holds %s", cl.Key, strings.Join(cl.Gone, ", "))
|
||||
}
|
||||
// A done key taken again builds only what is missing: the image or the package built unpublished as a candidate's; the
|
||||
// other artifacts stay as the done key has them (ADR 0487 clauses 5 and 8, ADR 0498 clause 6).
|
||||
var prior *delivery.BuilderBuild
|
||||
@@ -350,6 +413,9 @@ func (b *Builder) try(ctx context.Context, module string, m delivery.Module) {
|
||||
return
|
||||
}
|
||||
result.By, result.Manifest = b.Claimer.By, m.Manifest
|
||||
if len(cl.Gone) > 0 && result.State == delivery.BuildDone {
|
||||
result.Step += " (built again: the artifact store no longer held " + strings.Join(cl.Gone, ", ") + ")"
|
||||
}
|
||||
if publishing && result.State == delivery.BuildFailed {
|
||||
// The trunk's image or package was built only as a candidate's and its push or publish failed: the key says so, so the builds view, the
|
||||
// checker and the failed-key retake see it, and a look does not build the image again for nothing.
|
||||
|
||||
@@ -89,18 +89,21 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
}
|
||||
// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
|
||||
// holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5).
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains",
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments",
|
||||
"mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
|
||||
t.Errorf("uses = %v, want the three build seats", m.Uses)
|
||||
}
|
||||
if len(m.Invokes) != 0 || len(m.Config) != 0 {
|
||||
t.Errorf("invokes = %v, config = %v, want none: no tool call from the builder and no capacity", m.Invokes, m.Config)
|
||||
// 1937/AC3: the builder asks the controller whether the artifact store still holds a done key's blobs (held), and no
|
||||
// other tool; it holds no capacity.
|
||||
if want := []string{"seat:mesh-controller.held"}; !reflect.DeepEqual(m.Invokes, want) || len(m.Config) != 0 {
|
||||
t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want)
|
||||
}
|
||||
// The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it starts.
|
||||
if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
// The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it
|
||||
// starts; and by the controller's artifacts-gone, a take-in refused for a gone blob (1937/AC4).
|
||||
if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", "mesh-controller.artifacts-gone"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
t.Errorf("consumes = %v, want %v", m.Consumes, want)
|
||||
}
|
||||
if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
|
||||
@@ -109,7 +112,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
|
||||
if m.SecretsOwner != "mesh-delivery-builder" {
|
||||
t.Errorf("secrets-owner = %q", m.SecretsOwner)
|
||||
}
|
||||
for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "tools", "seat", "seats", "bus", "invokes", "config"} {
|
||||
for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "defines-seats", "tools", "seat", "seats", "bus", "config"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the builder has none", k)
|
||||
}
|
||||
|
||||
@@ -4,7 +4,11 @@
|
||||
"slug": "builder",
|
||||
"consumes": [
|
||||
"mesh-build-go.toolchain.reported",
|
||||
"mesh-build-node.toolchain.reported"
|
||||
"mesh-build-node.toolchain.reported",
|
||||
"mesh-controller.artifacts-gone"
|
||||
],
|
||||
"invokes": [
|
||||
"seat:mesh-controller.held"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "builds", "history": 5 },
|
||||
@@ -16,6 +20,7 @@
|
||||
"mesh-build-image.steps",
|
||||
"mesh-build-node.steps",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-controller.assignments",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests",
|
||||
"mesh-delivery-indexer.modules"
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -20,6 +20,8 @@ const (
|
||||
getFileTool = "gitea.gitea_get_file"
|
||||
compareTool = "gitea.gitea_compare"
|
||||
commitsTool = "gitea.gitea_commits"
|
||||
// gitea_list_branches answers {branches: [{name, commit}]}, every branch with the commit it points at.
|
||||
branchesTool = "gitea.gitea_list_branches"
|
||||
)
|
||||
|
||||
// notFoundStatus is the 404 of the forge's error, `Gitea API <path>: <status> <body>`.
|
||||
@@ -102,3 +104,26 @@ func (f forgeReads) FullSHA(_ context.Context, repository, commit string) (strin
|
||||
}
|
||||
return a.Commits[0].SHA, nil
|
||||
}
|
||||
|
||||
// BranchHeads is the commit each branch of the repository points at.
|
||||
func (f forgeReads) BranchHeads(_ context.Context, repository string) (map[string]string, error) {
|
||||
owner, repo := ownerRepo(repository)
|
||||
raw, err := f.ask(branchesTool, map[string]string{"owner": owner, "repo": repo})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var a struct {
|
||||
Branches *[]struct {
|
||||
Name string `json:"name"`
|
||||
Commit string `json:"commit"`
|
||||
} `json:"branches"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &a); err != nil || a.Branches == nil {
|
||||
return nil, fmt.Errorf("the answer of %s is not {branches: [{name, commit}]}: %v", branchesTool, err)
|
||||
}
|
||||
heads := map[string]string{}
|
||||
for _, b := range *a.Branches {
|
||||
heads[b.Name] = b.Commit
|
||||
}
|
||||
return heads, nil
|
||||
}
|
||||
|
||||
@@ -257,6 +257,14 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
|
||||
// The toolchains first, so a check that starts from the first candidate finds the nodes.
|
||||
watch := newToolchainWatch(ctx, js, tools, ch.NodesChanged, logger)
|
||||
// Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's
|
||||
// assignments are settled first, so the toolchains settle through the gate already knowing them.
|
||||
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
|
||||
watch.put = gate.Toolchain
|
||||
assignErrs, err := watchAssignments(ctx, js, gate, ch.NodesChanged, logger)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := watch.start(); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -369,6 +377,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return nil
|
||||
case err := <-errs:
|
||||
return err
|
||||
case err := <-assignErrs:
|
||||
return err
|
||||
case err := <-watch.errs:
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
package main
|
||||
|
||||
// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves
|
||||
// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very
|
||||
// toolchain agreement this file repairs.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
)
|
||||
|
||||
// A build module writes <node>.<language> of its toolchains state on every node it runs on, and nothing deletes the key
|
||||
// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of
|
||||
// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace
|
||||
// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree").
|
||||
//
|
||||
// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a
|
||||
// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments,
|
||||
// key <node>.<module>) hold an open assignment of its build module on its node; it is passed on as deleted otherwise,
|
||||
// and passed on again when the module is assigned there again. Until the assignments are settled (or where they cannot
|
||||
// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that
|
||||
// refusal is the guard of novox/hq ADR 0465, and it stays.
|
||||
type toolchainGate struct {
|
||||
mu sync.Mutex
|
||||
modules map[string]bool // the build modules whose toolchains pass the gate
|
||||
forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds
|
||||
logf func(format string, a ...any)
|
||||
known bool // the assignments are settled
|
||||
standing map[string]bool // <node>.<module> with an open assignment
|
||||
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
|
||||
left map[string]bool // <module>/<key> said as left out, so it is said once
|
||||
}
|
||||
|
||||
func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate {
|
||||
g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]bool{},
|
||||
raw: map[string]map[string][]byte{}, left: map[string]bool{}}
|
||||
for _, m := range modules {
|
||||
g.modules[m] = true
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed.
|
||||
func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.raw[module] == nil {
|
||||
g.raw[module] = map[string][]byte{}
|
||||
}
|
||||
if deleted {
|
||||
delete(g.raw[module], key)
|
||||
} else {
|
||||
g.raw[module][key] = value
|
||||
}
|
||||
return g.pass(module, key)
|
||||
}
|
||||
|
||||
// Assignment takes one change of the controller's assignments in and answers whether what it passed on changed.
|
||||
func (g *toolchainGate) Assignment(key string, value []byte, deleted bool) bool {
|
||||
node, module, ok := strings.Cut(key, ".")
|
||||
if !ok || !g.modules[module] {
|
||||
return false
|
||||
}
|
||||
open := !deleted && openAssignment(value)
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.standing[key] == open {
|
||||
return false
|
||||
}
|
||||
if open {
|
||||
g.standing[key] = true
|
||||
} else {
|
||||
delete(g.standing, key)
|
||||
}
|
||||
if !g.known {
|
||||
return false
|
||||
}
|
||||
changed := false
|
||||
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
||||
if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node {
|
||||
changed = g.pass(module, k) || changed
|
||||
}
|
||||
}
|
||||
return changed
|
||||
}
|
||||
|
||||
// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it
|
||||
// passed on changed.
|
||||
func (g *toolchainGate) Settled() bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
g.known = true
|
||||
changed := false
|
||||
for _, module := range slices.Sorted(maps.Keys(g.raw)) {
|
||||
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
||||
changed = g.pass(module, k) || changed
|
||||
}
|
||||
}
|
||||
return changed
|
||||
}
|
||||
|
||||
// pass hands one key on as it stands: its value when its node is assigned (or the assignments are not known), deleted
|
||||
// otherwise. The caller holds mu.
|
||||
func (g *toolchainGate) pass(module, key string) bool {
|
||||
value, held := g.raw[module][key]
|
||||
if !held {
|
||||
return g.forward(module, key, nil, true)
|
||||
}
|
||||
node, _, err := delivery.ParseToolchainKey(key)
|
||||
if err != nil || !g.known || g.standing[node+"."+module] {
|
||||
delete(g.left, module+"/"+key)
|
||||
return g.forward(module, key, value, false)
|
||||
}
|
||||
if !g.left[module+"/"+key] {
|
||||
g.left[module+"/"+key] = true
|
||||
g.logf("%s.%s is left out of the agreed toolchain: %s has no assignment on %s, so it is no builder (issue 1940)",
|
||||
delivery.ToolchainsBucket(module), key, module, node)
|
||||
}
|
||||
return g.forward(module, key, nil, true)
|
||||
}
|
||||
|
||||
// openAssignment is whether a value of the controller's assignments holds an assignment that stands: one that has not
|
||||
// ended and has no state (a pending one is not on the machine yet; mesh-controller internal/chain Assignment).
|
||||
func openAssignment(value []byte) bool {
|
||||
var a struct {
|
||||
Ended time.Time `json:"ended"`
|
||||
State string `json:"state"`
|
||||
}
|
||||
if json.Unmarshal(value, &a) != nil {
|
||||
return false
|
||||
}
|
||||
return a.Ended.IsZero() && a.State == ""
|
||||
}
|
||||
|
||||
// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is
|
||||
// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read
|
||||
// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel.
|
||||
func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) {
|
||||
errs := make(chan error, 1)
|
||||
kv, err := js.KeyValue(ctx, provider.AssignmentsBucket)
|
||||
if err != nil {
|
||||
// Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery).
|
||||
logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
|
||||
return errs, nil
|
||||
}
|
||||
each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e)) }
|
||||
w, err := settle(ctx, kv, each)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
|
||||
}
|
||||
if gate.Settled() {
|
||||
changed(ctx)
|
||||
}
|
||||
go func() {
|
||||
errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) {
|
||||
if gate.Assignment(e.Key(), e.Value(), !opOf(e)) {
|
||||
changed(ctx)
|
||||
}
|
||||
})
|
||||
}()
|
||||
return errs, nil
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
package main
|
||||
|
||||
// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940).
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
func gateReport(t *testing.T, node, version string) (string, []byte) {
|
||||
t.Helper()
|
||||
key, err := delivery.ToolchainKey(node, "typescript")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return key, value
|
||||
}
|
||||
|
||||
func agreedOf(set *buildrun.ToolchainSet) (string, string) {
|
||||
tc, none := delivery.AgreedToolchain(set.Entries("typescript"))
|
||||
return tc.Version, none
|
||||
}
|
||||
|
||||
// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the
|
||||
// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them
|
||||
// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again.
|
||||
func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
var said []string
|
||||
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) })
|
||||
for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} {
|
||||
key, value := gateReport(t, node, version)
|
||||
gate.Toolchain(buildrun.NodeBuildModule, key, value, false)
|
||||
}
|
||||
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") {
|
||||
t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none)
|
||||
}
|
||||
for _, node := range []string{"g14", "novox", "shanks"} {
|
||||
gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false)
|
||||
}
|
||||
gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false)
|
||||
gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false) // another module's: no matter
|
||||
if !gate.Settled() {
|
||||
t.Fatal("settling the assignments left ace's report out, so what was passed on changed")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none)
|
||||
}
|
||||
if len(said) != 1 {
|
||||
t.Fatalf("ace's report is said left out once; said %v", said)
|
||||
}
|
||||
// ace reports again while unassigned (an old value seen again): still left out.
|
||||
key, value := gateReport(t, "ace", "v26.12.0")
|
||||
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("an unassigned node's report changed what was passed on")
|
||||
}
|
||||
// Assigned to ace again: its report counts, and the builders really disagree: the guard stays.
|
||||
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false) {
|
||||
t.Fatal("assigning ace again passed nothing on")
|
||||
}
|
||||
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") {
|
||||
t.Fatalf("builders that run different versions still disagree; got %q", none)
|
||||
}
|
||||
// The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again.
|
||||
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false) {
|
||||
t.Fatal("a pending assignment is no assignment on the machine")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("got %q, %q", v, none)
|
||||
}
|
||||
gate.Assignment("ace.mesh-build-node", nil, true)
|
||||
if v, _ := agreedOf(set); v != "v26.12.1" {
|
||||
t.Fatalf("got %q", v)
|
||||
}
|
||||
// A report deleted from the bucket goes whether its node is assigned or not.
|
||||
key, _ = gateReport(t, "g14", "v26.12.1")
|
||||
if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) {
|
||||
t.Fatal("deleting g14's report passed nothing on")
|
||||
}
|
||||
if n := len(set.Entries("typescript")); n != 2 {
|
||||
t.Fatalf("novox and shanks are left; got %d entries", n)
|
||||
}
|
||||
}
|
||||
|
||||
// A newly assigned build node that has not reported yet changes nothing; its report counts as it comes.
|
||||
func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(string, ...any) {})
|
||||
gate.Settled()
|
||||
if gate.Assignment("g14.mesh-build-node", []byte(`{}`), false) {
|
||||
t.Fatal("an assignment with no report passed something on")
|
||||
}
|
||||
key, value := gateReport(t, "g14", "v26.12.1")
|
||||
if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("an assigned node's report was not passed on")
|
||||
}
|
||||
key, value = gateReport(t, "ace", "v26.12.0")
|
||||
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("a report of a node with no assignment key was passed on after the assignments settled")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("got %q, %q", v, none)
|
||||
}
|
||||
}
|
||||
@@ -24,9 +24,11 @@ var toolchainStreams = map[string]string{buildrun.GoBuildModule: "TOOLCHAINS_GO"
|
||||
// §8): it watches the buckets that exist at start, and a build module says each time it wrote its toolchains key
|
||||
// (toolchain.reported, handled by Reported); a bucket not watched yet is then watched.
|
||||
type toolchainWatch struct {
|
||||
ctx context.Context
|
||||
js bucketOpener
|
||||
set *buildrun.ToolchainSet
|
||||
ctx context.Context
|
||||
js bucketOpener
|
||||
set *buildrun.ToolchainSet
|
||||
// put takes a change of a toolchains bucket in: set.Put, or the gate of the assigned build nodes before it (issue 1940).
|
||||
put func(module, key string, value []byte, deleted bool) bool
|
||||
changed func(ctx context.Context) // a language's entries changed: the failed builds are tried again
|
||||
logger *log.Logger
|
||||
errs chan error
|
||||
@@ -35,7 +37,7 @@ type toolchainWatch struct {
|
||||
}
|
||||
|
||||
func newToolchainWatch(ctx context.Context, js bucketOpener, set *buildrun.ToolchainSet, changed func(context.Context), logger *log.Logger) *toolchainWatch {
|
||||
return &toolchainWatch{ctx: ctx, js: js, set: set, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}}
|
||||
return &toolchainWatch{ctx: ctx, js: js, set: set, put: set.Put, changed: changed, logger: logger, errs: make(chan error, len(buildrun.LanguageOf)), watched: map[string]bool{}}
|
||||
}
|
||||
|
||||
// start watches the buckets that exist now; a missing one is said and waits for its build module's event.
|
||||
@@ -63,14 +65,14 @@ func (w *toolchainWatch) ensure(module string) error {
|
||||
case err != nil:
|
||||
return fmt.Errorf("the bucket %s: %w", name, err)
|
||||
}
|
||||
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.set.Put(module, e.Key(), e.Value(), !opOf(e)) })
|
||||
watcher, err := settle(w.ctx, kv, func(e jetstream.KeyValueEntry) { w.put(module, e.Key(), e.Value(), !opOf(e)) })
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
w.watched[module] = true
|
||||
go func() {
|
||||
w.errs <- follow(w.ctx, watcher, func(e jetstream.KeyValueEntry) {
|
||||
if w.set.Put(module, e.Key(), e.Value(), !opOf(e)) {
|
||||
if w.put(module, e.Key(), e.Value(), !opOf(e)) {
|
||||
w.changed(w.ctx)
|
||||
}
|
||||
})
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-checker
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024214154-bc16740feb0d
|
||||
github.com/nats-io/nats-server/v2 v2.11.17
|
||||
github.com/nats-io/nats.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93 h1:FAiFY2KBKEoZSH49OZTt7/zRBIQxyl5g/KhYrpggbN4=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023213200-403a5d72ab93/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024214154-bc16740feb0d h1:oJREw0ggKqJWJ5DEISSL0i2ZC5XqdXAidxo9NJ74TCc=
|
||||
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024214154-bc16740feb0d/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -83,6 +83,10 @@ func (j *Examiner) besideFor(ctx context.Context, c *Case, p Pull) ([]BesideRef,
|
||||
out = append(out, BesideRef{Repository: repo, Commit: member.Head, Ref: refOf(member.Number)})
|
||||
} else if head := heads[repo]; head != "" {
|
||||
out = append(out, BesideRef{Repository: repo, Commit: head})
|
||||
} else if repo == controllerRepo && p.Repository == nodeEngineRepo {
|
||||
// The node-engine's step-nine test judges the controller's own manifest and fails without it (issue 1295): a
|
||||
// checker that cannot place it says so, where placing none left that test to skip.
|
||||
return nil, fmt.Errorf("the trunk head of %s is not known, so the controller the tests of %s#%d read cannot be placed beside them", repo, p.Repository, p.Number)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,6 +3,7 @@ package check
|
||||
import (
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
@@ -40,6 +41,9 @@ type Case struct {
|
||||
Queued time.Time
|
||||
// Try narrows a try's check (its level, its tests); nil for a pull request's.
|
||||
Try *TryOptions
|
||||
|
||||
notesMu sync.Mutex
|
||||
notes []string
|
||||
// Progress says what the check is doing, so a check that ran past its bound names the step it was stuck at.
|
||||
Progress *Progress
|
||||
// Kind is what asked the check (KindNew, KindNodes, KindTrunk): a re-judge after a trunk move is asked on the build
|
||||
@@ -148,3 +152,19 @@ func (t *Trunks) Heads() map[string]string {
|
||||
defer t.mu.Unlock()
|
||||
return maps.Clone(t.heads)
|
||||
}
|
||||
|
||||
// note puts a word of the check's way on the case, for a try's line (issue 1921).
|
||||
func (c *Case) note(s string) {
|
||||
c.notesMu.Lock()
|
||||
defer c.notesMu.Unlock()
|
||||
if !slices.Contains(c.notes, s) {
|
||||
c.notes = append(c.notes, s)
|
||||
}
|
||||
}
|
||||
|
||||
// Notes are the words noted on the case, in the order they were noted.
|
||||
func (c *Case) Notes() []string {
|
||||
c.notesMu.Lock()
|
||||
defer c.notesMu.Unlock()
|
||||
return slices.Clone(c.notes)
|
||||
}
|
||||
|
||||
@@ -0,0 +1,90 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
// failingResult is a level's test result in which the named tests failed, each in its own package.
|
||||
func failingResult(t *testing.T, tests int, failed ...Failure) json.RawMessage {
|
||||
t.Helper()
|
||||
raw, err := json.Marshal(Tested{Level: delivery.LevelIntegration, Tests: tests, Elapsed: 2, Failed: failed, Passed: len(failed) == 0})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
func judgeLevel(t *testing.T, tests ...json.RawMessage) (*delivery.Verdict, *state) {
|
||||
t.Helper()
|
||||
st := &state{j: &Examiner{}, c: &Case{}, levels: map[string]*acc{delivery.LevelUnit: {}, delivery.LevelIntegration: {}}, nodes: map[string]bool{}}
|
||||
v, _ := st.level(t.Context(), &target{pull: Pull{Repository: "novox/x"}}, delivery.LevelAnswer{Level: delivery.LevelIntegration, Tests: tests})
|
||||
return v, st
|
||||
}
|
||||
|
||||
// Issue 1921: two failing tests in two packages are two entries of the verdict's Failures and two lines of a try's
|
||||
// failed list, each with its own first line (its cause, issue 1901), whether the build module answered them in one test
|
||||
// result (one go test run over both packages) or in two (the changed packages first, then the rest); every result is
|
||||
// counted, the ones after a failure too; and Failed keeps the first one's words whole.
|
||||
func TestTwoFailingTestsInTwoPackagesAreTwoEntries(t *testing.T) {
|
||||
a := Failure{Package: "x/a", Test: "TestA", Cause: "a_test.go:9: want 1, got 2", Output: " a_test.go:3: logged\n a_test.go:9: want 1, got 2\n--- FAIL: TestA (0.00s)"}
|
||||
b := Failure{Package: "x/b", Test: "TestB", Cause: "b_test.go:4: boom", Output: " b_test.go:4: boom\n--- FAIL: TestB (0.00s)"}
|
||||
for name, tests := range map[string][]json.RawMessage{
|
||||
"one result": {failingResult(t, 10, a, b)},
|
||||
"two results": {failingResult(t, 4, a), failingResult(t, 3), failingResult(t, 3, b)},
|
||||
} {
|
||||
v, st := judgeLevel(t, tests...)
|
||||
if v == nil {
|
||||
t.Fatalf("%s: the level passed", name)
|
||||
}
|
||||
want := []string{"integration: x/a.TestA: a_test.go:9: want 1, got 2", "integration: x/b.TestB: b_test.go:4: boom"}
|
||||
if got := failedLines(*v); strings.Join(got, "\n") != strings.Join(want, "\n") {
|
||||
t.Errorf("%s: failed lines\n%s\nwant\n%s", name, strings.Join(got, "\n"), strings.Join(want, "\n"))
|
||||
}
|
||||
if len(v.Failures) != 2 || v.Failed.Test != "integration: x/a.TestA" || !strings.Contains(v.Failed.Words, "a_test.go:3: logged") {
|
||||
t.Errorf("%s: verdict %+v", name, v)
|
||||
}
|
||||
if run := st.levels[delivery.LevelIntegration].run; run != 10 {
|
||||
t.Errorf("%s: %d tests counted, want all 10", name, run)
|
||||
}
|
||||
if _, err := delivery.EncodeVerdict(delivery.Verdict{Group: "g", Heads: "0123456789ab", State: v.State, Failed: v.Failed, Failures: v.Failures}); err != nil {
|
||||
t.Errorf("%s: the verdict is not written: %v", name, err)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Issue 1921: past delivery.MaxFailures the list names the first fifty and how many more, the status line says how many
|
||||
// failed, and each entry's line is cut to MaxFailureLine.
|
||||
func TestFailuresAreCappedWithHowManyMore(t *testing.T) {
|
||||
var fs []Failure
|
||||
for i := range 60 {
|
||||
fs = append(fs, Failure{Package: fmt.Sprintf("x/p%d", i), Test: "TestT", Cause: strings.Repeat("y", 500)})
|
||||
}
|
||||
v, _ := judgeLevel(t, failingResult(t, 60, fs...))
|
||||
if len(v.Failures) != delivery.MaxFailures+1 || v.Failures[delivery.MaxFailures].Test != "and 10 more" {
|
||||
t.Fatalf("%d entries, last %+v", len(v.Failures), v.Failures[len(v.Failures)-1])
|
||||
}
|
||||
if len(v.Failures[0].Words) > MaxFailureLine {
|
||||
t.Errorf("an entry's line is %d bytes", len(v.Failures[0].Words))
|
||||
}
|
||||
if n := failedCount(*v); n != 60 {
|
||||
t.Errorf("failedCount = %d, want 60", n)
|
||||
}
|
||||
cs := &Case{Pulls: []Pull{{Repository: "novox/x", Head: "0123456789abcdef", Trunk: "fedcba9876543210"}}}
|
||||
if line := Line(*v, cs, cs.Pulls[0], nil, nil, nil); !strings.HasPrefix(line, "failed (60 tests): integration: x/p0.TestT") {
|
||||
t.Errorf("line %q", line)
|
||||
}
|
||||
}
|
||||
|
||||
// A verdict that names no list (a conflict, a refused step, a verdict written before issue 1921) is one failed line, as
|
||||
// before.
|
||||
func TestAVerdictWithoutFailuresIsOneLine(t *testing.T) {
|
||||
v := delivery.Verdict{State: delivery.VerdictFail, Failed: delivery.FailedFirst{Machine: "g14", Words: "files in conflict: go.mod\nmore"}}
|
||||
if got := failedLines(v); len(got) != 1 || got[0] != "g14: files in conflict: go.mod" {
|
||||
t.Fatalf("%q", got)
|
||||
}
|
||||
}
|
||||
@@ -341,7 +341,9 @@ func TestANodePackageIsNeverKeptAndRunsEveryCheck(t *testing.T) { // 1288/AC4, 1
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheChangesOwnFailingTestRunsFirstAndNothingElseRuns(t *testing.T) { // 1289/AC1, AC2
|
||||
// 1289/AC1, AC2 as issue 1921 changed it: the change's own failing test runs first, the level's other packages still run
|
||||
// after it (so one check names every failure), and no later level runs.
|
||||
func TestTheChangesOwnFailingTestRunsFirstThenTheRestAndNoLaterLevel(t *testing.T) {
|
||||
b := newBench(t)
|
||||
b.fetch.Changed = []string{"modules/m/internal/b/b_test.go"}
|
||||
b.digest[delivery.LevelUnit] = Digested{Level: "unit", Packages: map[string]string{"m/internal/a": digA, "m/internal/b": digB}}
|
||||
@@ -357,8 +359,8 @@ func TestTheChangesOwnFailingTestRunsFirstAndNothingElseRuns(t *testing.T) { //
|
||||
t.Fatalf("verdict = %+v", out.Verdict)
|
||||
}
|
||||
tests := b.steps.tool("test")
|
||||
if len(tests) != 1 || strings.Join(packagesOf(tests[0]), ",") != "m/internal/b" {
|
||||
t.Fatalf("test calls = %+v, want only m/internal/b, once", tests)
|
||||
if len(tests) != 2 || strings.Join(packagesOf(tests[0]), ",") != "m/internal/b" || strings.Join(packagesOf(tests[1]), ",") != "m/internal/a" {
|
||||
t.Fatalf("test calls = %+v, want m/internal/b, then m/internal/a, each once", tests)
|
||||
}
|
||||
if n := len(b.steps.tool("digest")); n != 1 {
|
||||
t.Errorf("%d digest calls, want the unit level's alone (the check stopped)", n)
|
||||
|
||||
@@ -438,8 +438,9 @@ func (st *state) ask(ctx context.Context, t *target, merged *[]Fetched) *deliver
|
||||
}
|
||||
|
||||
// level judges one level of a check ask's answer as the tools' answers were judged: the digest keys the kept verdicts
|
||||
// of the packages that passed, the kept ones taken are counted, and the first test result that did not pass is the
|
||||
// verdict. It answers the level's running time (its tests' and setup's).
|
||||
// of the packages that passed, the kept ones taken are counted, and every test result is read and counted, also after
|
||||
// one that did not pass, so the verdict names every failed test of the level (issue 1921). It answers the level's
|
||||
// running time (its tests' and setup's).
|
||||
func (st *state) level(ctx context.Context, t *target, la delivery.LevelAnswer) (*delivery.Verdict, float64) {
|
||||
level := la.Level
|
||||
a := st.levels[level]
|
||||
@@ -461,6 +462,7 @@ func (st *state) level(ctx context.Context, t *target, la delivery.LevelAnswer)
|
||||
a.kept += k.Tests
|
||||
}
|
||||
elapsed := 0.0
|
||||
var failedResults []Tested
|
||||
for _, raw := range la.Tests {
|
||||
var res Tested
|
||||
if err := json.Unmarshal(raw, &res); err != nil {
|
||||
@@ -489,9 +491,12 @@ func (st *state) level(ctx context.Context, t *target, la delivery.LevelAnswer)
|
||||
}
|
||||
}
|
||||
if !res.Passed {
|
||||
return levelFailure(t, level, res), elapsed
|
||||
failedResults = append(failedResults, res)
|
||||
}
|
||||
}
|
||||
if len(failedResults) > 0 {
|
||||
return levelFailure(t, level, failedResults...), elapsed
|
||||
}
|
||||
return nil, elapsed
|
||||
}
|
||||
|
||||
@@ -558,23 +563,60 @@ func firstLine(s string) string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// failure turns a level that did not pass into a verdict naming what failed first.
|
||||
func levelFailure(t *target, level string, res Tested) *delivery.Verdict {
|
||||
// MaxFailureLine is the most bytes of a failure's line in a verdict's Failures: the first line of its words, which is
|
||||
// its cause where the build module told one apart (issue 1901). Failed keeps the first failure's words whole.
|
||||
const MaxFailureLine = 300
|
||||
|
||||
// levelFailure turns a level that did not pass into a verdict: Failed names what failed first with its words whole,
|
||||
// and Failures every failed test of the level's results with its first line, at most delivery.MaxFailures and an entry
|
||||
// saying how many more (issue 1921).
|
||||
func levelFailure(t *target, level string, results ...Tested) *delivery.Verdict {
|
||||
v := delivery.Verdict{State: delivery.VerdictFail}
|
||||
switch {
|
||||
case len(res.Failed) > 0:
|
||||
f := res.Failed[0]
|
||||
name := failedName(f)
|
||||
if t.module == buildrun.NodeBuildModule && f.Package == "" {
|
||||
name = res.Module
|
||||
var all []delivery.FailedFirst
|
||||
for _, res := range results {
|
||||
if len(res.Failed) == 0 {
|
||||
why := res.Why
|
||||
if why == "" {
|
||||
why = "the " + level + " level did not pass"
|
||||
}
|
||||
all = append(all, delivery.FailedFirst{Test: level + ": " + t.pull.Repository, Words: why})
|
||||
continue
|
||||
}
|
||||
v.Failed = delivery.FailedFirst{Test: level + ": " + name, Words: failureWords(f)}
|
||||
default:
|
||||
why := res.Why
|
||||
if why == "" {
|
||||
why = "the " + level + " level did not pass"
|
||||
for _, f := range res.Failed {
|
||||
name := failedName(f)
|
||||
if t.module == buildrun.NodeBuildModule && f.Package == "" {
|
||||
name = res.Module
|
||||
}
|
||||
all = append(all, delivery.FailedFirst{Test: level + ": " + name, Words: failureWords(f)})
|
||||
}
|
||||
v.Failed = delivery.FailedFirst{Test: level + ": " + t.pull.Repository, Words: why}
|
||||
}
|
||||
if len(all) == 0 { // no result: a level said failed with nothing to name
|
||||
all = append(all, delivery.FailedFirst{Test: level + ": " + t.pull.Repository, Words: "the " + level + " level did not pass"})
|
||||
}
|
||||
v.Failed = all[0]
|
||||
v.Failures = failureLines(all)
|
||||
return &v
|
||||
}
|
||||
|
||||
// failureLines is every failure with the first line of its words, cut to MaxFailureLine, at most delivery.MaxFailures
|
||||
// and then one entry saying how many more failed.
|
||||
func failureLines(all []delivery.FailedFirst) []delivery.FailedFirst {
|
||||
n := min(len(all), delivery.MaxFailures)
|
||||
out := make([]delivery.FailedFirst, 0, n+1)
|
||||
for _, f := range all[:n] {
|
||||
out = append(out, delivery.FailedFirst{Test: f.Test, Machine: f.Machine, Words: clipLine(firstLine(f.Words), MaxFailureLine)})
|
||||
}
|
||||
if more := len(all) - n; more > 0 {
|
||||
out = append(out, delivery.FailedFirst{Test: fmt.Sprintf("and %d more", more)})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// clipLine cuts a line to at most n bytes on a whole character, marking the cut.
|
||||
func clipLine(s string, n int) string {
|
||||
if len(s) <= n {
|
||||
return s
|
||||
}
|
||||
s = strings.ToValidUTF8(s[:n-3], "")
|
||||
return s + "..."
|
||||
}
|
||||
|
||||
@@ -49,6 +49,34 @@ func firstFailed(v delivery.Verdict) string {
|
||||
return firstWordsLine(f.Words)
|
||||
}
|
||||
|
||||
// failedLines is everything a verdict names as failed, one line each: every failed test with its first line (issue
|
||||
// 1921), or what failed first when the verdict names no list (a conflict, a refused step, an older verdict).
|
||||
func failedLines(v delivery.Verdict) []string {
|
||||
if len(v.Failures) == 0 {
|
||||
if f := firstFailed(v); f != "" {
|
||||
return []string{f}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
lines := make([]string, 0, len(v.Failures))
|
||||
for _, f := range v.Failures {
|
||||
lines = append(lines, firstFailed(delivery.Verdict{Failed: f}))
|
||||
}
|
||||
return lines
|
||||
}
|
||||
|
||||
// failedCount is how many tests a verdict names as failed, the ones after delivery.MaxFailures counted too.
|
||||
func failedCount(v delivery.Verdict) int {
|
||||
n := len(v.Failures)
|
||||
if n > delivery.MaxFailures {
|
||||
var more int
|
||||
if _, err := fmt.Sscanf(v.Failures[n-1].Test, "and %d more", &more); err == nil {
|
||||
return delivery.MaxFailures + more
|
||||
}
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
||||
// firstWordsLine is the first line of a failure's words: the words are several lines (issue 1901) and the status line one.
|
||||
func firstWordsLine(w string) string { return firstLine(w) }
|
||||
|
||||
@@ -155,6 +183,9 @@ func parts(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote, slowest []s
|
||||
pc.head = "pass " + commits
|
||||
default:
|
||||
pc.head = "failed: " + firstFailed(v)
|
||||
if n := failedCount(v); n > 1 {
|
||||
pc.head = fmt.Sprintf("failed (%d tests): %s", n, firstFailed(v))
|
||||
}
|
||||
pc.commits = commits
|
||||
}
|
||||
var counts []string
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
"fmt"
|
||||
"path"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
)
|
||||
|
||||
@@ -36,6 +37,8 @@ type Forge interface {
|
||||
OnTrunk(ctx context.Context, repository, commit string) (bool, error)
|
||||
// FullSHA is the 40-digit sha of a commit named by its first 12 digits.
|
||||
FullSHA(ctx context.Context, repository, commit string) (string, error)
|
||||
// BranchHeads is the commit every branch of a repository points at, by branch name (issue 1921).
|
||||
BranchHeads(ctx context.Context, repository string) (map[string]string, error)
|
||||
}
|
||||
|
||||
var pseudoVersion = regexp.MustCompile(`-([0-9a-f]{12})$`)
|
||||
@@ -93,6 +96,18 @@ func (st *state) checkPinsOf(ctx context.Context, t *target) verdictPtr {
|
||||
if onTrunk {
|
||||
continue
|
||||
}
|
||||
if st.c.Try != nil {
|
||||
// A try only (issue 1921): a repository tried on a branch that mesh-sdk shares by name (a milestone's branch across
|
||||
// repositories) is built with that branch's mesh-sdk head, which no group pull request holds before the pull requests exist.
|
||||
branch, err := st.sameBranchPin(ctx, t, pin)
|
||||
if err != nil {
|
||||
return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: whether mesh-sdk %s is the head of a branch cannot be told: %v%s", t.pull.Repository, t.pull.Number, pin, err, unheardWords(err)))
|
||||
}
|
||||
if branch != "" {
|
||||
st.c.note(fmt.Sprintf("pin accepted: head of mesh-sdk %s", branch))
|
||||
continue
|
||||
}
|
||||
}
|
||||
nor := "the group has no mesh-sdk pull request"
|
||||
if sdk != nil {
|
||||
nor = "the head of the group's mesh-sdk pull request is " + short12(sdk.Head)
|
||||
@@ -103,6 +118,35 @@ func (st *state) checkPinsOf(ctx context.Context, t *target) verdictPtr {
|
||||
return nil
|
||||
}
|
||||
|
||||
// sameBranchPin names the branch of the tried repository that points at the tried head and that mesh-sdk also has, at the
|
||||
// commit the pin names; empty when there is none.
|
||||
func (st *state) sameBranchPin(ctx context.Context, t *target, pin string) (string, error) {
|
||||
mine, err := st.j.Forge.BranchHeads(ctx, t.pull.Repository)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
var same []string
|
||||
for name, head := range mine {
|
||||
if head == t.pull.Head && name != "main" {
|
||||
same = append(same, name)
|
||||
}
|
||||
}
|
||||
if len(same) == 0 {
|
||||
return "", nil
|
||||
}
|
||||
sdk, err := st.j.Forge.BranchHeads(ctx, SDKRepository)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
sort.Strings(same)
|
||||
for _, name := range same {
|
||||
if h, ok := sdk[name]; ok && strings.HasPrefix(h, pin) {
|
||||
return name, nil
|
||||
}
|
||||
}
|
||||
return "", nil
|
||||
}
|
||||
|
||||
// failureVerdict is a verdict of failure the check ends with.
|
||||
|
||||
// pinHint is added to the failure of a Go member's fetch that names mesh-sdk while the group holds a mesh-sdk pull request:
|
||||
|
||||
@@ -18,6 +18,15 @@ type fakeForge struct {
|
||||
onTrunk map[string]bool // by commit prefix
|
||||
err error
|
||||
errAt map[string]error // by ref: a read at that ref fails with it
|
||||
// branches are the branches of a repository with the commit each points at.
|
||||
branches map[string]map[string]string
|
||||
}
|
||||
|
||||
func (f *fakeForge) BranchHeads(_ context.Context, repository string) (map[string]string, error) {
|
||||
if f.err != nil {
|
||||
return nil, f.err
|
||||
}
|
||||
return f.branches[repository], nil
|
||||
}
|
||||
|
||||
func (f *fakeForge) File(_ context.Context, repository, path, ref string) (string, error) {
|
||||
@@ -346,3 +355,56 @@ func TestAMeshLabHeadGetsACheck(t *testing.T) {
|
||||
t.Fatalf("mesh-lab was not fetched: %+v", fetched)
|
||||
}
|
||||
}
|
||||
|
||||
// 1295: the node-engine's step-nine test fails without the controller beside it, so a checker that knows no trunk head of the
|
||||
// controller fails the check at the placing, loudly, and does not place none.
|
||||
func TestTheNodeEnginesCheckFailsWhereTheControllersTrunkIsNotKnown(t *testing.T) {
|
||||
const host = "novox/mesh-host"
|
||||
b := newBench(t)
|
||||
b.ex.Trunks = trunksOf(t, map[string]string{host: trunkA, catalogueRepository: trunkS, SDKRepository: trunkS})
|
||||
v := b.judge(t, hostCase(t, pullOf(host, 5, head1))).Verdict
|
||||
if v.State != delivery.VerdictFail || !strings.Contains(v.Failed.Words, "mesh-controller") || !strings.Contains(v.Failed.Words, "cannot be placed") {
|
||||
t.Fatalf("verdict = %+v", v)
|
||||
}
|
||||
}
|
||||
|
||||
// sameBranchBench is a try of mesh-delivery's feat/r6 at head1 whose go.mod pins the SDK at pin, with the SDK's feat/r6 at head2.
|
||||
func sameBranchBench(t *testing.T, pin string) (*bench, *Case) {
|
||||
b := groupBench(t, pin)
|
||||
b.ex.Forge.(*fakeForge).branches = map[string]map[string]string{
|
||||
repoA: {"main": trunkA, "feat/r6": head1},
|
||||
repoSDK: {"main": "cb54756c3865" + strings.Repeat("0", 28), "feat/r6": head2},
|
||||
}
|
||||
c := caseOf(t, pullOf(repoA, 7, head1))
|
||||
c.Try = &TryOptions{}
|
||||
return b, c
|
||||
}
|
||||
|
||||
func TestATryPinnedToTheHeadOfTheSDKBranchOfTheTriedBranchPasses(t *testing.T) { // 1921/AC1
|
||||
b, c := sameBranchBench(t, head2[:12])
|
||||
if v := b.judge(t, c).Verdict; v.State != delivery.VerdictPass {
|
||||
t.Fatalf("verdict = %+v", v)
|
||||
}
|
||||
if n := c.Notes(); len(n) != 1 || n[0] != "pin accepted: head of mesh-sdk feat/r6" {
|
||||
t.Errorf("notes = %q", n)
|
||||
}
|
||||
}
|
||||
|
||||
func TestATryPinnedToAnotherCommitIsStillRefused(t *testing.T) { // 1921/AC2
|
||||
b, c := sameBranchBench(t, "9aaa5c80ac5c")
|
||||
v := b.judge(t, c).Verdict
|
||||
if v.State != delivery.VerdictFail || !strings.Contains(v.Failed.Words, "9aaa5c80ac5c") || !strings.Contains(v.Failed.Words, "neither on mesh-sdk's trunk nor the group's head") {
|
||||
t.Fatalf("verdict = %+v", v)
|
||||
}
|
||||
if len(c.Notes()) != 0 {
|
||||
t.Errorf("notes = %q", c.Notes())
|
||||
}
|
||||
}
|
||||
|
||||
func TestAMergeCheckWithTheSameBranchPinIsStillRefusedWithoutAGroupPullRequest(t *testing.T) { // 1921/AC3
|
||||
b, _ := sameBranchBench(t, head2[:12])
|
||||
v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict // no Try: a pull request's check
|
||||
if v.State != delivery.VerdictFail || !strings.Contains(v.Failed.Words, "the group has no mesh-sdk pull request") {
|
||||
t.Fatalf("verdict = %+v", v)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -67,7 +67,9 @@ type Try struct {
|
||||
// Said is the try's report in words: what failed first, the tests run, kept and skipped per level, the time of each part.
|
||||
Said string `json:"said,omitempty"`
|
||||
// Took is the time from its start to its end, as Go writes a duration.
|
||||
Took string `json:"took,omitempty"`
|
||||
Took string `json:"took,omitempty"`
|
||||
// Failed is every test that failed, each with its first line, at most delivery.MaxFailures and then how many more
|
||||
// (issue 1921); or what else failed first (a conflict, a refused step).
|
||||
Failed []string `json:"failed,omitempty"`
|
||||
Levels map[string]delivery.LevelResult `json:"levels,omitempty"`
|
||||
Machines []string `json:"machines,omitempty"`
|
||||
@@ -329,6 +331,9 @@ func (tr *Tryer) run(ctx context.Context, key string) {
|
||||
t.Took = t.Ended.Sub(t.Started).Round(time.Second).String()
|
||||
t.Levels, t.Machines = v.Levels, v.Machines
|
||||
t.Said = Report(v, cs, cs.Pulls[0], out.Builds, out.Slowest, out.Judged)
|
||||
for _, n := range cs.Notes() {
|
||||
t.Said += " | " + n
|
||||
}
|
||||
switch {
|
||||
case !judged || IsNodesFailure(v):
|
||||
t.State = TryError
|
||||
@@ -336,9 +341,7 @@ func (tr *Tryer) run(ctx context.Context, key string) {
|
||||
t.State = TryPass
|
||||
default:
|
||||
t.State = TryFail
|
||||
if f := firstFailed(v); f != "" {
|
||||
t.Failed = []string{f}
|
||||
}
|
||||
t.Failed = failedLines(v)
|
||||
}
|
||||
if t.Tests != "" {
|
||||
t.Said += " | only the tests " + t.Tests + " (Go); nothing kept"
|
||||
|
||||
@@ -96,14 +96,14 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
||||
if want := []state{{Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "tries", History: 1}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) {
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains",
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-node.steps", "mesh-build-node.toolchains", "mesh-controller.assignments",
|
||||
"mesh-delivery-assigner.compositions", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
// 1887/A5b: the checker reads nothing of the controller's plans-building (ADR 0483 superseded by ADR 0525 §7), calls no
|
||||
// build module's tool and holds no capacity: it puts check asks on the Go and Node build seats (ADR 0525 §1, §4).
|
||||
if want := []string{"gitea.gitea_commits", "gitea.gitea_compare", "gitea.gitea_get_file"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
|
||||
if want := []string{"gitea.gitea_commits", "gitea.gitea_compare", "gitea.gitea_get_file", "gitea.gitea_list_branches"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
|
||||
t.Errorf("invokes = %v, want only the forge's reads for a mesh-sdk pin (ADR 0484 clause 9)", m.Invokes)
|
||||
}
|
||||
if want := []string{"node-build-go", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
|
||||
|
||||
@@ -21,6 +21,7 @@
|
||||
"mesh-build-go.toolchains",
|
||||
"mesh-build-node.steps",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-controller.assignments",
|
||||
"mesh-delivery-assigner.compositions",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-gate.source",
|
||||
@@ -46,7 +47,8 @@
|
||||
"invokes": [
|
||||
"gitea.gitea_commits",
|
||||
"gitea.gitea_compare",
|
||||
"gitea.gitea_get_file"
|
||||
"gitea.gitea_get_file",
|
||||
"gitea.gitea_list_branches"
|
||||
],
|
||||
|
||||
"own-secrets": {
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -549,3 +549,88 @@ func TestIndexerJudgesAnUnreadableVerdictAgainAtStart(t *testing.T) { // 1923/AC
|
||||
t.Errorf("avahi = fp2 %q none2 %q within 5 s of the start, want a fingerprint and no reason", got.FP2, got.None2)
|
||||
}
|
||||
}
|
||||
|
||||
// TestIndexerCountsOnlyTheAssignedBuildNodesToolchains (issue 1940): the toolchains bucket holds the report of n2, where
|
||||
// mesh-build-go was unassigned (its assignment ended) at another version than n1's, as ace's v26.12.0 stayed beside the
|
||||
// v26.12.1 of the nodes mesh-build-node runs on. The indexer gives the merged module a fingerprint from n1's toolchain
|
||||
// alone; when the controller assigns mesh-build-go to n2 again, its report counts and the builders disagree (the guard
|
||||
// stays); when the assignment ends again, the fingerprint comes back. No restart, no timer: the assignments' watch.
|
||||
func TestIndexerCountsOnlyTheAssignedBuildNodesToolchains(t *testing.T) {
|
||||
manifest, err := os.ReadFile("../../internal/index/testdata/catalogue-a67f0196-avahi.module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
const goSum = "3b5028f0c1e9bde04dd766abd63eb5931bc516bae695cf46f00abcd8510b0381"
|
||||
var assignments jetstream.KeyValue
|
||||
js, ctx, _, modules := startIndexer(t, func(context.Context, jetstream.KeyValue) {}, func(ctx context.Context, js jetstream.JetStream) {
|
||||
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: delivery.ToolchainsBucket("mesh-build-go"), History: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for node, version := range map[string]string{"n1": "go1.27.3", "n2": "go1.27.2"} {
|
||||
key, _ := delivery.ToolchainKey(node, "go")
|
||||
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "go", Version: version, Recipe: "1", At: time.Now()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := kv.Put(ctx, key, value); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
// The controller's assignments mirror, in its layout (mesh-controller internal/chain Assignment).
|
||||
if assignments, err = js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_assignments", History: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for key, value := range map[string]string{
|
||||
"n1.mesh-build-go": `{"node":"n1","module":"mesh-build-go","assigned":"2026-10-20T08:00:00Z"}`,
|
||||
"n2.mesh-build-go": `{"node":"n2","module":"mesh-build-go","assigned":"2026-10-20T08:00:00Z","ended":"2026-10-23T08:00:00Z"}`,
|
||||
} {
|
||||
if _, err := assignments.Put(ctx, key, []byte(value)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
})
|
||||
data, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: "novox/mesh-catalog", Branch: "feat/x", Number: 302,
|
||||
Head: "0123456789abcdef0123456789abcdef01234567", Merge: "89abcdef0123456789abcdef0123456789abcdef", Root: "r0",
|
||||
Trees: map[string]string{"modules/avahi": "newtree"},
|
||||
Manifests: map[string]string{"modules/avahi": string(manifest)},
|
||||
Locks: map[string]map[string]string{"modules/avahi": {"go.sum": goSum}}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.Publish(ctx, "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-catalog.merged", data); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
read := func(ok func(delivery.Module) bool) delivery.Module {
|
||||
var got delivery.Module
|
||||
for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
|
||||
if e, err := modules.Get(ctx, "avahi"); err == nil {
|
||||
if m, err := delivery.DecodeModule(e.Value()); err == nil {
|
||||
if got = m; ok(m) {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return got
|
||||
}
|
||||
agreed := func(m delivery.Module) bool { return m.FP2 != "" && m.Toolchains["go"].Version == "go1.27.3" }
|
||||
if m := read(agreed); !agreed(m) {
|
||||
t.Fatalf("within 5 s avahi has fp2 %q, none2 %q, toolchains %v: want a fingerprint of n1's go1.27.3 alone", m.FP2, m.None2, m.Toolchains)
|
||||
}
|
||||
if _, err := assignments.Put(ctx, "n2.mesh-build-go", []byte(`{"node":"n2","module":"mesh-build-go","assigned":"2026-10-24T08:00:00Z"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
disagree := func(m delivery.Module) bool {
|
||||
return m.FP2 == "" && strings.HasPrefix(m.None2, "builders disagree: go1.27.2 recipe 1 on n2")
|
||||
}
|
||||
if m := read(disagree); !disagree(m) {
|
||||
t.Fatalf("within 5 s of n2's assignment avahi has fp2 %q, none2 %q: want the builders to disagree", m.FP2, m.None2)
|
||||
}
|
||||
if _, err := assignments.Put(ctx, "n2.mesh-build-go", []byte(`{"node":"n2","module":"mesh-build-go","ended":"2026-10-24T09:00:00Z"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if m := read(agreed); !agreed(m) {
|
||||
t.Fatalf("within 5 s of n2's unassignment avahi has fp2 %q, none2 %q", m.FP2, m.None2)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -313,7 +313,26 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
go func() { sourceErrs <- follow(ctx, sw, each) }()
|
||||
}
|
||||
tools := newToolchainWatch(ctx, js, toolchainSink{put: ix.PutToolchain, apply: ix.ApplyToolchain, recompute: ix.Recompute}, logger)
|
||||
// Only the toolchain reports of nodes the build module is assigned to count (issue 1940): the controller's
|
||||
// assignments are settled first, so the toolchains settle through the gate already knowing them.
|
||||
gate := newToolchainGate(sortedModules(), ix.PutToolchain, logger.Printf)
|
||||
assignErrs, err := watchAssignments(ctx, js, gate, func(ctx context.Context) {
|
||||
if err := ix.Recompute(ctx); err != nil {
|
||||
logger.Printf("writing the modules again after an assignment of a build module changed: %v", err)
|
||||
}
|
||||
}, logger)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
tools := newToolchainWatch(ctx, js, toolchainSink{
|
||||
put: gate.Toolchain,
|
||||
apply: func(ctx context.Context, module, key string, value []byte, deleted bool) error {
|
||||
if !gate.Toolchain(module, key, value, deleted) {
|
||||
return nil
|
||||
}
|
||||
return ix.Recompute(ctx)
|
||||
},
|
||||
recompute: ix.Recompute}, logger)
|
||||
if err := tools.start(); err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -377,6 +396,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return err
|
||||
case err := <-sourceErrs:
|
||||
return err
|
||||
case err := <-assignErrs:
|
||||
return err
|
||||
case err := <-tools.errs:
|
||||
return err
|
||||
case err := <-previews.errs:
|
||||
|
||||
@@ -0,0 +1,174 @@
|
||||
package main
|
||||
|
||||
// This file is the same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker (issue 1940): it moves
|
||||
// to mesh-sdk go/delivery/buildrun once mesh-sdk can be delivered again, since a change of mesh-sdk waits on the very
|
||||
// toolchain agreement this file repairs.
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"maps"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/provider"
|
||||
)
|
||||
|
||||
// A build module writes <node>.<language> of its toolchains state on every node it runs on, and nothing deletes the key
|
||||
// when the module is unassigned from a node: the controller writes no module's state, and the module is gone. A report of
|
||||
// a node the build module no longer runs on is therefore not a builder (issue 1940: mesh-build-node unassigned from ace
|
||||
// on 2026-10-23 left ace.typescript at v26.12.0, and every Node fingerprint said "builders disagree").
|
||||
//
|
||||
// toolchainGate stands between the toolchains buckets and what takes their entries in (the indexer's PutToolchain, a
|
||||
// buildrun.ToolchainSet): an entry is passed on only while the controller's assignments (mesh-controller_assignments,
|
||||
// key <node>.<module>) hold an open assignment of its build module on its node; it is passed on as deleted otherwise,
|
||||
// and passed on again when the module is assigned there again. Until the assignments are settled (or where they cannot
|
||||
// be read) every entry is passed on, as before. Builders that run and report different versions still disagree: that
|
||||
// refusal is the guard of novox/hq ADR 0465, and it stays.
|
||||
type toolchainGate struct {
|
||||
mu sync.Mutex
|
||||
modules map[string]bool // the build modules whose toolchains pass the gate
|
||||
forward func(module, key string, value []byte, deleted bool) bool // answers whether it changed what it holds
|
||||
logf func(format string, a ...any)
|
||||
known bool // the assignments are settled
|
||||
standing map[string]bool // <node>.<module> with an open assignment
|
||||
raw map[string]map[string][]byte // module -> toolchains key -> value, as the bucket holds it
|
||||
left map[string]bool // <module>/<key> said as left out, so it is said once
|
||||
}
|
||||
|
||||
func newToolchainGate(modules []string, forward func(module, key string, value []byte, deleted bool) bool, logf func(string, ...any)) *toolchainGate {
|
||||
g := &toolchainGate{modules: map[string]bool{}, forward: forward, logf: logf, standing: map[string]bool{},
|
||||
raw: map[string]map[string][]byte{}, left: map[string]bool{}}
|
||||
for _, m := range modules {
|
||||
g.modules[m] = true
|
||||
}
|
||||
return g
|
||||
}
|
||||
|
||||
// Toolchain takes one change of a build module's toolchains bucket in and answers whether what it passed on changed.
|
||||
func (g *toolchainGate) Toolchain(module, key string, value []byte, deleted bool) bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.raw[module] == nil {
|
||||
g.raw[module] = map[string][]byte{}
|
||||
}
|
||||
if deleted {
|
||||
delete(g.raw[module], key)
|
||||
} else {
|
||||
g.raw[module][key] = value
|
||||
}
|
||||
return g.pass(module, key)
|
||||
}
|
||||
|
||||
// Assignment takes one change of the controller's assignments in and answers whether what it passed on changed.
|
||||
func (g *toolchainGate) Assignment(key string, value []byte, deleted bool) bool {
|
||||
node, module, ok := strings.Cut(key, ".")
|
||||
if !ok || !g.modules[module] {
|
||||
return false
|
||||
}
|
||||
open := !deleted && openAssignment(value)
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
if g.standing[key] == open {
|
||||
return false
|
||||
}
|
||||
if open {
|
||||
g.standing[key] = true
|
||||
} else {
|
||||
delete(g.standing, key)
|
||||
}
|
||||
if !g.known {
|
||||
return false
|
||||
}
|
||||
changed := false
|
||||
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
||||
if n, _, err := delivery.ParseToolchainKey(k); err == nil && n == node {
|
||||
changed = g.pass(module, k) || changed
|
||||
}
|
||||
}
|
||||
return changed
|
||||
}
|
||||
|
||||
// Settled says the assignments are read: from now on only the reports of assigned nodes pass. It answers whether what it
|
||||
// passed on changed.
|
||||
func (g *toolchainGate) Settled() bool {
|
||||
g.mu.Lock()
|
||||
defer g.mu.Unlock()
|
||||
g.known = true
|
||||
changed := false
|
||||
for _, module := range slices.Sorted(maps.Keys(g.raw)) {
|
||||
for _, k := range slices.Sorted(maps.Keys(g.raw[module])) {
|
||||
changed = g.pass(module, k) || changed
|
||||
}
|
||||
}
|
||||
return changed
|
||||
}
|
||||
|
||||
// pass hands one key on as it stands: its value when its node is assigned (or the assignments are not known), deleted
|
||||
// otherwise. The caller holds mu.
|
||||
func (g *toolchainGate) pass(module, key string) bool {
|
||||
value, held := g.raw[module][key]
|
||||
if !held {
|
||||
return g.forward(module, key, nil, true)
|
||||
}
|
||||
node, _, err := delivery.ParseToolchainKey(key)
|
||||
if err != nil || !g.known || g.standing[node+"."+module] {
|
||||
delete(g.left, module+"/"+key)
|
||||
return g.forward(module, key, value, false)
|
||||
}
|
||||
if !g.left[module+"/"+key] {
|
||||
g.left[module+"/"+key] = true
|
||||
g.logf("%s.%s is left out of the agreed toolchain: %s has no assignment on %s, so it is no builder (issue 1940)",
|
||||
delivery.ToolchainsBucket(module), key, module, node)
|
||||
}
|
||||
return g.forward(module, key, nil, true)
|
||||
}
|
||||
|
||||
// openAssignment is whether a value of the controller's assignments holds an assignment that stands: one that has not
|
||||
// ended and has no state (a pending one is not on the machine yet; mesh-controller internal/chain Assignment).
|
||||
func openAssignment(value []byte) bool {
|
||||
var a struct {
|
||||
Ended time.Time `json:"ended"`
|
||||
State string `json:"state"`
|
||||
}
|
||||
if json.Unmarshal(value, &a) != nil {
|
||||
return false
|
||||
}
|
||||
return a.Ended.IsZero() && a.State == ""
|
||||
}
|
||||
|
||||
// watchAssignments settles the controller's assignments into the gate, says them settled, and follows them; changed is
|
||||
// told whenever what the gate passes on changed. Where the bucket cannot be read (not there, or this module may not read
|
||||
// it) it says so and the gate passes every report on, as before; an error of the watch after that ends on the channel.
|
||||
func watchAssignments(ctx context.Context, js bucketOpener, gate *toolchainGate, changed func(context.Context), logger *log.Logger) (<-chan error, error) {
|
||||
errs := make(chan error, 1)
|
||||
kv, err := js.KeyValue(ctx, provider.AssignmentsBucket)
|
||||
if err != nil {
|
||||
// Not fatal: an indexer, builder or checker that stops for it stops every delivery (a grant comes with a delivery).
|
||||
logger.Printf("the bucket %s cannot be read, so every node's toolchain report counts, an unassigned build node's too (issue 1940): %v", provider.AssignmentsBucket, err)
|
||||
return errs, nil
|
||||
}
|
||||
each := func(e jetstream.KeyValueEntry) { gate.Assignment(e.Key(), e.Value(), !opOf(e)) }
|
||||
w, err := settle(ctx, kv, each)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the bucket %s: %w", provider.AssignmentsBucket, err)
|
||||
}
|
||||
if gate.Settled() {
|
||||
changed(ctx)
|
||||
}
|
||||
go func() {
|
||||
errs <- follow(ctx, w, func(e jetstream.KeyValueEntry) {
|
||||
if gate.Assignment(e.Key(), e.Value(), !opOf(e)) {
|
||||
changed(ctx)
|
||||
}
|
||||
})
|
||||
}()
|
||||
return errs, nil
|
||||
}
|
||||
@@ -0,0 +1,112 @@
|
||||
package main
|
||||
|
||||
// The same in mesh-delivery-indexer, mesh-delivery-builder and mesh-delivery-checker, as standing.go is (issue 1940).
|
||||
|
||||
import (
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
func gateReport(t *testing.T, node, version string) (string, []byte) {
|
||||
t.Helper()
|
||||
key, err := delivery.ToolchainKey(node, "typescript")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
value, err := delivery.EncodeToolchain(delivery.Toolchain{Language: "typescript", Version: version, Recipe: "f59571cf4be75adf", At: time.Unix(0, 0).UTC()})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return key, value
|
||||
}
|
||||
|
||||
func agreedOf(set *buildrun.ToolchainSet) (string, string) {
|
||||
tc, none := delivery.AgreedToolchain(set.Entries("typescript"))
|
||||
return tc.Version, none
|
||||
}
|
||||
|
||||
// The live case of issue 1940: mesh-build-node was unassigned from ace, whose report stayed at v26.12.0 while the
|
||||
// three nodes it runs on report v26.12.1. Without the assignments every report counts (and they disagree); with them
|
||||
// ace's is left out and the three agree; assigned to ace again, its report counts again and the guard refuses again.
|
||||
func TestAnUnassignedBuildNodesReportIsNoBuilder(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
var said []string
|
||||
gate := newToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, set.Put, func(f string, a ...any) { said = append(said, f) })
|
||||
for node, version := range map[string]string{"ace": "v26.12.0", "g14": "v26.12.1", "novox": "v26.12.1", "shanks": "v26.12.1"} {
|
||||
key, value := gateReport(t, node, version)
|
||||
gate.Toolchain(buildrun.NodeBuildModule, key, value, false)
|
||||
}
|
||||
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree") {
|
||||
t.Fatalf("before the assignments are read every report counts, so they disagree; got %q", none)
|
||||
}
|
||||
for _, node := range []string{"g14", "novox", "shanks"} {
|
||||
gate.Assignment(node+"."+buildrun.NodeBuildModule, []byte(`{"node":"`+node+`","module":"mesh-build-node"}`), false)
|
||||
}
|
||||
gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","ended":"2026-10-23T08:00:00Z"}`), false)
|
||||
gate.Assignment("ace.mesh-build-go", []byte(`{"node":"ace","module":"mesh-build-go"}`), false) // another module's: no matter
|
||||
if !gate.Settled() {
|
||||
t.Fatal("settling the assignments left ace's report out, so what was passed on changed")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("the assigned nodes agree on v26.12.1; got %q, %q", v, none)
|
||||
}
|
||||
if len(said) != 1 {
|
||||
t.Fatalf("ace's report is said left out once; said %v", said)
|
||||
}
|
||||
// ace reports again while unassigned (an old value seen again): still left out.
|
||||
key, value := gateReport(t, "ace", "v26.12.0")
|
||||
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("an unassigned node's report changed what was passed on")
|
||||
}
|
||||
// Assigned to ace again: its report counts, and the builders really disagree: the guard stays.
|
||||
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node"}`), false) {
|
||||
t.Fatal("assigning ace again passed nothing on")
|
||||
}
|
||||
if _, none := agreedOf(set); !strings.HasPrefix(none, "builders disagree: v26.12.0 recipe f59571cf4be75adf on ace") {
|
||||
t.Fatalf("builders that run different versions still disagree; got %q", none)
|
||||
}
|
||||
// The assignment's key deleted, or a pending one written: no assignment stands, and ace's report goes again.
|
||||
if !gate.Assignment("ace.mesh-build-node", []byte(`{"node":"ace","module":"mesh-build-node","state":"pending"}`), false) {
|
||||
t.Fatal("a pending assignment is no assignment on the machine")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("got %q, %q", v, none)
|
||||
}
|
||||
gate.Assignment("ace.mesh-build-node", nil, true)
|
||||
if v, _ := agreedOf(set); v != "v26.12.1" {
|
||||
t.Fatalf("got %q", v)
|
||||
}
|
||||
// A report deleted from the bucket goes whether its node is assigned or not.
|
||||
key, _ = gateReport(t, "g14", "v26.12.1")
|
||||
if !gate.Toolchain(buildrun.NodeBuildModule, key, nil, true) {
|
||||
t.Fatal("deleting g14's report passed nothing on")
|
||||
}
|
||||
if n := len(set.Entries("typescript")); n != 2 {
|
||||
t.Fatalf("novox and shanks are left; got %d entries", n)
|
||||
}
|
||||
}
|
||||
|
||||
// A newly assigned build node that has not reported yet changes nothing; its report counts as it comes.
|
||||
func TestANewlyAssignedNodeCountsFromItsReport(t *testing.T) {
|
||||
set := &buildrun.ToolchainSet{}
|
||||
gate := newToolchainGate([]string{buildrun.NodeBuildModule}, set.Put, func(string, ...any) {})
|
||||
gate.Settled()
|
||||
if gate.Assignment("g14.mesh-build-node", []byte(`{}`), false) {
|
||||
t.Fatal("an assignment with no report passed something on")
|
||||
}
|
||||
key, value := gateReport(t, "g14", "v26.12.1")
|
||||
if !gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("an assigned node's report was not passed on")
|
||||
}
|
||||
key, value = gateReport(t, "ace", "v26.12.0")
|
||||
if gate.Toolchain(buildrun.NodeBuildModule, key, value, false) {
|
||||
t.Fatal("a report of a node with no assignment key was passed on after the assignments settled")
|
||||
}
|
||||
if v, none := agreedOf(set); v != "v26.12.1" || none != "" {
|
||||
t.Fatalf("got %q, %q", v, none)
|
||||
}
|
||||
}
|
||||
@@ -81,7 +81,7 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
|
||||
if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}}; !reflect.DeepEqual(m.State, want) {
|
||||
t.Errorf("state = %+v, want %+v", m.State, want)
|
||||
}
|
||||
if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
// "Reads FORGE" is a consumes of the indexer's stream: the broker grants only buckets by reads (mesh-control
|
||||
|
||||
@@ -16,6 +16,7 @@
|
||||
"reads": [
|
||||
"mesh-build-go.toolchains",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-controller.assignments",
|
||||
"mesh-controller.builds",
|
||||
"mesh-delivery-checker.previews",
|
||||
"mesh-delivery-gate.source"
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
@@ -0,0 +1 @@
|
||||
These tests read files outside this directory (the whole catalogue, the repository, or its root files): mesh-build-go never keeps this package's verdict (issue 1935).
|
||||
Reference in New Issue
Block a user