22 Commits
Author SHA1 Message Date
jschoubben 36bf3b7143 Merge pull request 'The builder says it writes references only once its build modules proved it (issue 1886, ADR 0522)' (#23) from fix/1886-builder-says-references into main
Reviewed-on: #23
2026-10-23 00:26:18 +00:00
jschoubben 9fad90ee89 Say that the builder writes references only once each build module answered one (issue 1886, ADR 0522 review)
mesh/merge-gate head 9fad90ee890a merged into trunk b714eab995a5 — pass: builds mesh-delivery-builder → novox; no bus step; the change touches no module of…
mesh/repo-check head 9fad90ee890a merged into trunk b714eab995a5 — pass: its merge-check.sh passed
mesh/delivery delivered
mesh/delivery-group group fix/1886-builder-says-references delivered: every member is delivered
mesh/check failed: integration: github.com/novox/mesh-controller/internal/identity.TestACertificateChainsToTheMeshsOwnAuthority: authority_test.go:45:…
The references come from the build modules' own pushes, so a new builder beside
an old build module writes none; saying it at start would send the controller
to keys that fail not stored.
2026-10-23 00:52:50 +02:00
jschoubben 4b8b1c047c Say in the builder's own bucket of instances that it writes references (issue 1886)
The controller delivers from the builder's keys only while a running builder
says its done keys name references; a builder from before R4e says nothing,
so R4e's own delivery is asked of build-agent instead of waiting for ever.
The say is put once started and taken away when stopped.
2026-10-23 00:48:28 +02:00
mesh-tools b714eab995 Merge pull request 'R4e (mesh-delivery): the builder pushes and keys artifacts by reference, the checker's verdict and try, the view, and the checker's owed state read by the forge' (#22) from feat/r4e into main 2026-10-22 22:13:05 +00:00
jschoubben efca168c55 Stop the forge bridge performing mesh/check: the forge module writes it (R2-1)
mesh/repo-check head efca168c5522 merged into trunk 0e372c87f1df beside mesh-catalog#457@896bd883 +1 more — pass: its merge-check.sh passed
mesh/merge-gate head efca168c5522 merged into trunk 0e372c87f1df — pass: group feat/r4e composed: every machine composes with the change as it did without …
mesh/delivery-group group feat/r4e delivered: every member is delivered
mesh/check failed: unit: ..test: node:internal/modules/esm/resolve:989 | efca168c5522 on b714eab995a5 | unit 2092 run, 0 kept, 3 skipped | integration…
mesh/delivery covered after its stop: a later walk that contains it was delivered: plan-1792707174627079192 (novox/mesh-delivery at b714eab9 into main, d…
R4e round 2 finding R2-1, ADR 0521: nothing in the delivery plan orders the
forge module before the bridge, and the forge module's durable consumer keeps
its deliver policy when its filters change, so a check.status emitted before
it consumed the event was lost. The forge module now watches the checker's
owed state and writes mesh/check itself, once per revision.

The bridge emits no check.status, declares no emits, and neither reads the
checker's owed nor hears its stream; OwedAct still refuses mesh/check. The
integration test shows the bridge calls no tool and publishes nothing of its
own when the checker owes mesh/check.
2026-10-22 21:40:09 +02:00
jschoubben 19d06163de Announce the checker's states, owed readable, so the forge module may read it (R2-1)
R4e round 2 finding R2-1, ADR 0521: the forge module writes mesh/check from
the checker's owed state, read by a watch, because no event order between the
forge bridge and the forge module is kept by the delivery plan. A reader is
granted from the owner's announcement (ADR 0495 §2-§3), and the checker
announced nothing. It now says every state it keeps, with owed, previews and
verdicts readable: the forge module and the bridge read owed, the indexer
previews, the gate verdicts.
2026-10-22 21:38:53 +02:00
jschoubben 0997b071a3 Pin every R4e member module to mesh-sdk's final R4e head 104533e8, which holds review fixes S1 and S2, so the group pins one SDK 2026-10-22 21:29:20 +02:00
jschoubben bb4a4a82dc Emit mesh/check as the bridge's own check.status, never through a tool (K1)
Review finding K1 of R4e, issue 1301 S4 (ADR 0503 §3, ADR 0513 §4: the forge
requires mesh/check alone). gitea_check_status rested on "only the bridge
invokes it", but node-tools invokes every tool and a tool is told no caller,
so any session could set mesh/check and merge past the checker.

The bridge now emits check.status on mesh.mod.mesh-delivery-forge-bridge.event.
check.status, which the bus grants to its own account alone (emits), with the
owed status, its key and its revision; the event id check.<key>.<rev> is the
bus's dedup id. A status the bus did not take is not kept as done. OwedAct
refuses mesh/check, and gitea.gitea_check_status leaves the invokes. The
integration test shows the tool runner's account refused on that subject and
gitea's durable consumer handed statuses emitted before it fetched.
2026-10-22 21:23:27 +02:00
jschoubben 08bc672488 Pin every R4e member module to mesh-sdk's final R4e head 875c680d, which holds main's #68 and #69, so the group pins one SDK 2026-10-22 20:45:16 +02:00
jschoubben 90ac287e59 Drop the checker tools binary committed by mistake with the try verbs (7abe230): no module commits a build output 2026-10-22 20:29:53 +02:00
jschoubben 91e68ac13b Merge feat/r4e-builder into feat/r4e (R4e integration)
The checker's go.mod and go.sum conflicted on the mesh-sdk pin (checker lane 9cb4d59, builder lane e76a5a3): every
module touched is pinned to the sdk lane's newest head d1af87a until the final SDK head.
2026-10-22 20:29:53 +02:00
jschoubben 173becf35e Merge feat/r4e-checker into feat/r4e (R4e integration) 2026-10-22 20:29:10 +02:00
jschoubben 7d740be255 Fail a check whose forge read reached nobody at the forge, and judge it again when the nodes change (issue 1624, C7)
mesh-catalog #293 failed with 'go.mod cannot be read to check its mesh-sdk pin: nats: no responders', put at the
build node, and stayed failed: the forge's tool runner was not answering for a moment, which is not the change's
failure. The forge's reads now mark such a call unheard, as the build steps' calls are (issue 1820).
2026-10-22 18:07:22 +02:00
jschoubben 66ae4dc791 Format the judge (gofmt) 2026-10-22 18:06:11 +02:00
jschoubben 03eaf332ad Say the composition's time and the machines it composed in a check's line, once the assigner reports them (issue 1300, C10)
The line said only that the composition passed with an SDK commit, and its time sat among the parts that are dropped
first, so a forge status never said how long the composition took or which machines it covered.
2026-10-22 18:06:07 +02:00
jschoubben cbdcf6da0e Give a group's TypeScript members the mesh-sdk member's package when its merge changes the package's paths, and pin the SDK's R4e head (ADR 0500 §4-§5, issue 1309's checker half)
A TypeScript member of a group that changes mesh-sdk's package resolved the old version from the registry. Its Node
fetch now names the mesh-sdk member's head, which the Node build module packs and serves, publishing nothing. The paths
come from the indexer's modules value (lead decision L8), so a change of Go files or of the manifest alone resolves
the registry as before. A test holds that no step of a check carries publish.
2026-10-22 18:05:12 +02:00
jschoubben 7abe230701 Judge a pushed head as a try on the checker, through checker_try and checker_try_read (ADR 0501 §5)
mesh-controller's try asked build-agent, which leaves the checking at R4e. A try is now a key in the checker's state
tries: the checker judges the head merged onto its trunk as it judges a pull request's head, through the same
judge, places and holds, writes its end in the key and owes the forge nothing (no status, no preview, no verdict).
One try per head, at most four at once; hq and a repository whose trunk the gate does not name are refused with
"no level of <repo> is checked by the checker".
2026-10-22 18:03:28 +02:00
jschoubben 2855ae7ebf Set the owed mesh/check through gitea_check_status, the one tool that writes it (issue 1301's delivery part, option A)
The forge's gitea_commit_status refuses the context mesh/check from R4e, so that only the checker's account, through
this bridge, can make a pull request pass the one required status.
2026-10-22 17:59:30 +02:00
jschoubben d9ace64681 Mark mesh/check pending on every open pull request into a moved trunk, and carry the gate's heads hash in each verdict (issue 1302's checker part, lead decision L4; issue 1300)
The gate decides on mesh/check alone and matches an Approve's members by the heads hash over lowercase repositories
and heads. A success judged on an old trunk stood on the forge while the next check waited for a place or a delivery,
and a candidate whose repository the forge named in capitals carried a hash the gate never finds. A check that ends
after its trunk moved now owes pending, not its verdict.
2026-10-22 17:58:49 +02:00
jschoubben f7d8622deb Keep a core repository's passing package under a digest that covers its siblings' commits, so core checks keep verdicts (issue 1287, C3a/C3b, lead decision L5)
The checker took and kept nothing for a workspace given siblings, and every core repository has siblings, so the
kept store stayed empty. The digest step now folds the siblings' commits in and echoes them; a verdict is kept or
taken only when the echo equals the siblings the fetch was given, so a digest that misses one never poisons another
check (ADR 0444, issue 482).
2026-10-22 17:56:59 +02:00
jschoubben d83156fb21 Add checker_previews and checker_status read tools so the live acceptance of issues 1680 and 1685 can be read (C1, C2) 2026-10-22 17:25:11 +02:00
jschoubben 16ca51dc1f Bound builder_builds by key prefix, limit and read budget, list unreadable keys, refuse fp2 with commit, add candidate reads (issue 1827 V1-V3, 1818, 1681) 2026-10-22 17:23:46 +02:00
59 changed files with 2384 additions and 376 deletions
@@ -7,6 +7,7 @@ import (
"crypto/sha256" "crypto/sha256"
"encoding/hex" "encoding/hex"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"io" "io"
"os" "os"
@@ -155,6 +156,8 @@ type bed struct {
tools map[string]jetstream.KeyValue tools map[string]jetstream.KeyValue
presence jetstream.KeyValue presence jetstream.KeyValue
have jetstream.KeyValue have jetstream.KeyValue
instances jetstream.KeyValue
stop func() // stops the builder started last
fake *fakeBuildModules fake *fakeBuildModules
srvURL string srvURL string
dir string dir string
@@ -227,6 +230,7 @@ func startBed(t *testing.T) *bed {
b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory) b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
presence := mk(delivery.PresenceBucket, delivery.PresenceHistory) presence := mk(delivery.PresenceBucket, delivery.PresenceHistory)
b.presence, b.have = presence, mk(delivery.HaveBucket, delivery.HaveHistory) b.presence, b.have = presence, mk(delivery.HaveBucket, delivery.HaveHistory)
b.instances = mk(instancesBucket, 1)
for _, node := range []string{"n1", "n2"} { for _, node := range []string{"n1", "n2"} {
b.setOnline(t, node, true) b.setOnline(t, node, true)
for _, module := range []string{"mesh-build-go", "mesh-build-node"} { for _, module := range []string{"mesh-build-go", "mesh-build-node"} {
@@ -344,7 +348,9 @@ func (b *bed) start(t *testing.T, node string, env map[string]string) {
runCtx, stop := context.WithCancel(b.ctx) runCtx, stop := context.WithCancel(b.ctx)
done := make(chan error, 1) done := make(chan error, 1)
go func() { done <- run(runCtx, func(k string) string { return e[k] }, io.Discard) }() go func() { done <- run(runCtx, func(k string) string { return e[k] }, io.Discard) }()
t.Cleanup(func() { stop(); <-done }) var once sync.Once
b.stop = func() { once.Do(func() { stop(); <-done }) }
t.Cleanup(b.stop)
} }
// await reads the build key until pred holds of it, within 20 s. // await reads the build key until pred holds of it, within 20 s.
@@ -713,3 +719,26 @@ func TestAManifestFallingBackToAPredefinedKeyIsBuilt(t *testing.T) {
t.Errorf("build = %+v, want done: the manifest is read", v) t.Errorf("build = %+v, want done: the manifest is read", v)
} }
} }
// 1886/AC5: a builder on n1 whose Go, Node and image build modules each answered a reference says under n1, in its bucket
// of instances, that it writes references, and takes the say away when it stops: the controller delivers from its keys
// only while it holds.
func TestABuilderWhoseBuildModulesAnsweredReferencesSaysItAndStoppedSaysNothing(t *testing.T) {
b := startBed(t)
b.module(t, "netcheck", "catalogue-a67f0196-netcheck.module.json", "modules/netcheck") // a Go and a TypeScript bundle
b.moduleCommit(t, "imaged", "catalogue-a67f0196-imaged.module.json", "modules/imaged", nil, commit) // an image, for the trunk
b.start(t, "n1", nil)
var said string
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && said == ""; time.Sleep(50 * time.Millisecond) {
if e, err := b.instances.Get(b.ctx, "n1"); err == nil {
said = string(e.Value())
}
}
if !strings.Contains(said, `"writes":["references"]`) {
t.Fatalf("the builder said %q", said)
}
b.stop()
if _, err := b.instances.Get(b.ctx, "n1"); !errors.Is(err, jetstream.ErrKeyNotFound) {
t.Fatalf("the stopped builder still says: %v", err)
}
}
@@ -178,6 +178,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
Manifests: manifestReader{manifests}, Manifests: manifestReader{manifests},
Logf: logger.Printf, Logf: logger.Printf,
} }
// **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
// yet (the manifest's state not made) is said, and the controller keeps asking build-agent.
switch kv, err := js.KeyValue(ctx, instancesBucket); {
case errors.Is(err, jetstream.ErrBucketNotFound):
logger.Printf("the bucket %s does not exist yet: the controller is not told that this builder writes references until it starts again", instancesBucket)
case err != nil:
return fmt.Errorf("the bucket %s: %w", instancesBucket, err)
default:
says := newSayer(ctx, kv, cred.Node, []string{buildrun.GoBuildModule, buildrun.NodeBuildModule, build.ImageBuildModule}, logger.Printf)
b.Answered = says.Answered
defer func() {
stopping, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
defer cancel()
if err := says.unsay(stopping); err != nil {
logger.Printf("%v", err)
}
}()
}
// The toolchains first, so a build that starts from the first modules value finds the nodes. // The toolchains first, so a build that starts from the first modules value finds the nodes.
watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger) watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger)
if err := watch.start(); err != nil { if err := watch.start(); err != nil {
@@ -0,0 +1,79 @@
package main
import (
"context"
"fmt"
"sync"
"github.com/nats-io/nats.go/jetstream"
)
// What the builder says it writes (novox/hq issue 1886, ADR 0522): each running instance keeps a key, named by its node,
// in its own bucket of instances (`mesh-delivery-builder_instances`, the manifest's state `instances`), saying that every
// done key it writes names where its artifacts are (ADR 0498 §1). The references come from the build modules' own pushes,
// so an instance says it only once a step of every build module it calls answered one since it started: a builder beside
// a build module from before R4e says nothing. The controller delivers from the builder's keys only while an instance
// says so, and asks build-agent as before R4e otherwise. The bucket is read by the controller alone, so its name and shape
// are not in mesh-sdk.
const (
instancesBucket = "mesh-delivery-builder_instances"
writesReferences = `{"v":1,"writes":["references"]}`
)
// sayStore is the bucket of instances as the builder writes it.
type sayStore interface {
Put(ctx context.Context, key string, value []byte) (uint64, error)
Delete(ctx context.Context, key string, opts ...jetstream.KVDeleteOpt) error
}
// sayer puts this instance's say once every build module in need answered a reference, and takes it away at the stop.
type sayer struct {
ctx context.Context
store sayStore
node string
need []string
logf func(string, ...any)
mu sync.Mutex
answered map[string]bool
said bool
}
func newSayer(ctx context.Context, store sayStore, node string, need []string, logf func(string, ...any)) *sayer {
return &sayer{ctx: ctx, store: store, node: node, need: need, logf: logf, answered: map[string]bool{}}
}
// Answered is told the build module whose step answered a reference (build.Builder.Answered).
func (s *sayer) Answered(module string) {
s.mu.Lock()
defer s.mu.Unlock()
s.answered[module] = true
if s.said {
return
}
for _, m := range s.need {
if !s.answered[m] {
return
}
}
if _, err := s.store.Put(s.ctx, s.node, []byte(writesReferences)); err != nil {
s.logf("saying in %s that this builder writes references: %v; said again at the next reference", instancesBucket, err)
return
}
s.said = true
s.logf("every build module answered a reference: said in %s that this builder writes references", instancesBucket)
}
// unsay takes the say away, if it was made: a builder replaced by one that does not write references leaves none.
func (s *sayer) unsay(ctx context.Context) error {
s.mu.Lock()
defer s.mu.Unlock()
if !s.said {
return nil
}
if err := s.store.Delete(ctx, s.node); err != nil {
return fmt.Errorf("taking away the say in %s: %w", instancesBucket, err)
}
s.said = false
return nil
}
@@ -0,0 +1,46 @@
package main
import (
"context"
"encoding/json"
"testing"
"github.com/nats-io/nats.go/jetstream"
)
type saysOf map[string]string
func (s saysOf) Put(_ context.Context, key string, value []byte) (uint64, error) {
s[key] = string(value)
return 1, nil
}
func (s saysOf) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
delete(s, key)
return nil
}
// 1886/AC5: a builder says, under its node, that it writes references only once every build module it calls answered
// one; stopped, it says nothing any more.
func TestTheBuilderSaysItWritesReferencesOnceEveryBuildModuleAnsweredOne(t *testing.T) {
s := saysOf{}
say := newSayer(t.Context(), s, "shanks", []string{"mesh-build-go", "mesh-build-node", "mesh-build-image"}, t.Logf)
say.Answered("mesh-build-go")
say.Answered("mesh-build-node")
if v, ok := s["shanks"]; ok {
t.Fatalf("said before the image build module answered a reference: %q", v)
}
say.Answered("mesh-build-image")
var v struct {
Writes []string `json:"writes"`
}
if err := json.Unmarshal([]byte(s["shanks"]), &v); err != nil || len(v.Writes) != 1 || v.Writes[0] != "references" {
t.Fatalf("the builder's say: %q %v", s["shanks"], err)
}
if err := say.unsay(t.Context()); err != nil {
t.Fatal(err)
}
if _, ok := s["shanks"]; ok {
t.Fatalf("the stopped builder still says: %v", s)
}
}
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-builder
go 1.26.0 go 1.26.0
require ( require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
) )
+2 -8
View File
@@ -1,11 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc h1:ZJScJSnF6LgFl55/VJ9RT/lNroVWun469sXtx0nlVYw= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022154726-a90f523edc75 h1:e/aOJLHaNSBPFFHHjUF+0fAfjmtMUy/vDWqetauWp/U=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022154726-a90f523edc75/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7 h1:wK9O1CgNk7mo3IdJqW/CCx/hebg+1F0rkvW8CvJfnRM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b h1:OjlhfiGr+wKLWxafV+BRjZx3TbBtwRYcm9mgoam/jv8=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -383,3 +383,21 @@ func TestACandidatesFetchNamesTheNamedPathsTheIndexerGaveIt(t *testing.T) { // A
t.Errorf("fetch request = %+v, %v; want the candidate's base, the module's directory and its named paths", req, err) t.Errorf("fetch request = %+v, %v; want the candidate's base, the module's directory and its named paths", req, err)
} }
} }
// 1886/AC5: the builder tells who answered a reference: the build module whose step named one, and none for a step that
// answered a digest alone (a build module from before R4e).
func TestTheBuilderTellsWhichBuildModuleAnsweredAReference(t *testing.T) {
r := newRig(t, goBundle)
var answered []string
r.b.Answered = func(module string) { answered = append(answered, module) }
r.bm.noRef["ace/tool"] = true
r.consider(t, "modules/m")
if len(answered) != 0 {
t.Fatalf("a step that answered no reference was told as one: %v", answered)
}
delete(r.bm.noRef, "ace/tool")
r.consider(t, "modules/m")
if len(answered) != 1 || answered[0] != "mesh-build-go" {
t.Fatalf("the build module that answered a reference: %v, want mesh-build-go", answered)
}
}
@@ -45,6 +45,9 @@ type Builder struct {
// Limit is how many builds run at once; four when 0. // Limit is how many builds run at once; four when 0.
Limit int Limit int
Logf func(format string, args ...any) Logf func(format string, args ...any)
// 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)
once sync.Once once sync.Once
sem chan struct{} sem chan struct{}
@@ -749,6 +752,9 @@ func (b *Builder) runLane(ctx context.Context, l *lane, repository string, m del
if built.Reference != "" && (stored(t.Kind) || !candidate) { if built.Reference != "" && (stored(t.Kind) || !candidate) {
l.references[t.Artifact] = built.Reference l.references[t.Artifact] = built.Reference
} }
if built.Reference != "" && b.Answered != nil {
b.Answered(l.module)
}
} }
} }
@@ -83,7 +83,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
if m.Module != "mesh-delivery-builder" { if m.Module != "mesh-delivery-builder" {
t.Errorf("module = %q", m.Module) t.Errorf("module = %q", m.Module)
} }
if want := []state{{Name: "builds", History: 5}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory if want := []state{{Name: "builds", History: 5}, {Name: "instances", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
t.Errorf("state = %+v, want %+v", 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-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",
+2 -1
View File
@@ -7,7 +7,8 @@
"mesh-build-node.toolchain.reported" "mesh-build-node.toolchain.reported"
], ],
"state": [ "state": [
{ "name": "builds", "history": 5 } { "name": "builds", "history": 5 },
{ "name": "instances", "history": 1 }
], ],
"reads": [ "reads": [
"mesh-build-go.steps", "mesh-build-go.steps",
@@ -7,8 +7,10 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"os" "os"
"time"
stdio "git.novox.be/novox/mesh-sdk/go" stdio "git.novox.be/novox/mesh-sdk/go"
"git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-checker/internal/check" "mesh-delivery-checker/internal/check"
) )
@@ -23,6 +25,71 @@ func read(s stdio.KeptState) check.StateRead {
}} }}
} }
// stateStore is a state the tools write: the tries.
type stateStore struct {
check.StateRead
create func(key string, value []byte) error
}
func (s stateStore) Keys() ([]string, error) { return s.StateRead.Keys() }
func (s stateStore) Get(key string) (json.RawMessage, error) { return s.StateRead.Get(key) }
func (s stateStore) Create(key string, value []byte) error { return s.create(key, value) }
// Trying is what the try's tools work on: the tries, and the gate's trunk heads.
type Trying struct {
Tries check.TryStore
Trunks func() (map[string]string, error)
Now func() time.Time
}
// TryTools are the try's two tools (ADR 0501 §5): ask one, read one. mesh-controller's try calls them.
func TryTools(tr Trying) []stdio.Tool {
str := func(args map[string]any, k string) string { s, _ := args[k].(string); return s }
return []stdio.Tool{
{Name: check.TryTool,
Description: "Try a pushed head before its pull request is opened: the checker judges it as it judges a pull request's head, merged onto its trunk's head, " +
"its unit and integration levels through the build modules, and owes the forge nothing. Answers the try's id at once; " + check.TryReadTool + " reads " +
"where it stands. A try of the same head that waits or runs is the answer to a second ask; at most 4 wait or run at once, mesh-wide. A repository " +
"the checker checks no level of (hq, one whose trunk the gate does not name) is refused.",
Input: map[string]any{"type": "object", "required": []string{"repository", "head"}, "properties": map[string]any{
"repository": map[string]any{"type": "string", "description": "<owner>/<name>"},
"head": map[string]any{"type": "string", "description": "the full commit to try, 40 or 64 hex digits"},
"trunk": map[string]any{"type": "string", "description": "the trunk commit to merge it onto, or the trunk's branch for its head now; the trunk's head when absent"},
"tests": map[string]any{"type": "string", "description": "a go test -run pattern: only those tests run, and nothing is kept"},
"level": map[string]any{"type": "string", "enum": []string{"unit", "integration"}, "description": "only this level; both when absent"}}},
Run: func(args map[string]any) (any, error) {
if tr.Tries == nil || tr.Trunks == nil {
return nil, fmt.Errorf("the tries state cannot be written here")
}
trunks, err := tr.Trunks()
if err != nil {
return nil, fmt.Errorf("the trunk heads cannot be read: %w", err)
}
now := time.Now
if tr.Now != nil {
now = tr.Now
}
id, err := check.AskTry(tr.Tries, trunks, check.TryAsk{Repository: str(args, "repository"), Head: str(args, "head"),
Trunk: str(args, "trunk"), Tests: str(args, "tests"), Level: str(args, "level")}, now())
if err != nil {
return nil, err
}
return map[string]any{"id": id}, nil
}},
{Name: check.TryReadTool,
Description: "Where a try stands, as stored: queued, running, or its end (pass, fail, or error when it could not run), with what was said of it, " +
"the head and the trunk commit it was merged onto, the time it took and what failed first. Read-only.",
Input: map[string]any{"type": "object", "required": []string{"id"}, "properties": map[string]any{
"id": map[string]any{"type": "string", "description": "the try's id, try-<digits>"}}},
Run: func(args map[string]any) (any, error) {
if tr.Tries == nil {
return nil, fmt.Errorf("the tries state cannot be read here")
}
return check.ReadTry(tr.Tries, str(args, "id"))
}},
}
}
// Tools are the read tools over the checker's states. // Tools are the read tools over the checker's states.
func Tools(r check.Reader) []stdio.Tool { func Tools(r check.Reader) []stdio.Tool {
return []stdio.Tool{ return []stdio.Tool{
@@ -59,12 +126,69 @@ func Tools(r check.Reader) []stdio.Tool {
kept, err := r.Kept(level, digest) kept, err := r.Kept(level, digest)
return map[string]any{"kept": kept}, err return map[string]any{"kept": kept}, err
}}, }},
{Name: check.PreviewsReadTool,
Description: "The merge previews the checker holds, each as stored under previews.<repository>.<head>.<trunk>: the merge commit, per module directory " +
"its tree, manifest digest and lock hashes, or the files in conflict. With a repository, that repository's; with a head (7 or more characters) " +
"only that head's. Keys that are no preview key are listed as unreadable. Read-only.",
Input: map[string]any{"type": "object", "properties": map[string]any{
"repository": map[string]any{"type": "string", "description": "<owner>/<name>"},
"head": map[string]any{"type": "string", "description": "a pull request head commit, or a prefix of at least 7 characters"}}},
Run: func(args map[string]any) (any, error) {
repository, _ := args["repository"].(string)
head, _ := args["head"].(string)
if head != "" && len(head) < 7 {
return nil, fmt.Errorf("the head %q is too short: give at least 7 characters", head)
}
ps, bad, err := r.Previews(repository, head)
return map[string]any{"previews": ps, "unreadable": bad}, err
}},
{Name: check.StatusReadTool,
Description: "What holds the checker back and what runs: the capacity per node (the setting, the default one step per node), whether the controller's " +
"delivery plans in their build phase hold a key (no new check starts while one does), the steps in flight on the build modules by node, and the " +
"candidates without a verdict for their heads. The steps the checker holds waiting for a place are in its process, not in a bucket, and the " +
"answer says so. Read-only.",
Input: map[string]any{"type": "object", "properties": map[string]any{}},
Run: func(map[string]any) (any, error) { return r.Status() }},
} }
} }
func capacity() (string, error) {
path := os.Getenv("MESH_CHECKER_SETTINGS")
if path == "" {
return os.Getenv("MESH_CHECKER_CAPACITY"), nil
}
raw, err := os.ReadFile(path)
if err != nil {
return "", err
}
var s struct {
Capacity string `json:"capacity"`
}
return s.Capacity, json.Unmarshal(raw, &s)
}
func main() { func main() {
r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: read(stdio.State("mesh-delivery-indexer.candidates"))} r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: read(stdio.State("mesh-delivery-indexer.candidates")),
if err := stdio.Serve("", Tools(r)); err != nil { PreviewState: read(stdio.State("previews")), PlansState: read(stdio.State("mesh-controller.plans-building")),
StepStates: map[string]check.StateRead{"mesh-build-go": read(stdio.State("mesh-build-go.steps")), "mesh-build-node": read(stdio.State("mesh-build-node.steps"))},
Capacity: capacity}
tries := stdio.State(check.TriesState)
source := stdio.State("mesh-delivery-gate.source")
trying := Trying{
Tries: stateStore{StateRead: read(tries), create: func(key string, value []byte) error {
_, err := tries.Create(key, json.RawMessage(value))
return err
}},
Trunks: func() (map[string]string, error) {
e, err := source.Get(delivery.SourceMainKey)
if err != nil || e == nil {
return nil, err
}
s, err := delivery.DecodeSource(e.Value)
return s.Heads, err
},
}
if err := stdio.Serve("", append(Tools(r), TryTools(trying)...)); err != nil {
fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err) fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err)
os.Exit(1) os.Exit(1)
} }
@@ -1,12 +1,17 @@
package main package main
import ( import (
"encoding/json"
"errors"
"slices"
"strings"
"testing" "testing"
"time"
"mesh-delivery-checker/internal/check" "mesh-delivery-checker/internal/check"
) )
func TestTheToolsServedAreTheThreeReadTools(t *testing.T) { func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
var names []string var names []string
for _, tl := range Tools(check.Reader{}) { for _, tl := range Tools(check.Reader{}) {
names = append(names, tl.Name) names = append(names, tl.Name)
@@ -14,14 +19,63 @@ func TestTheToolsServedAreTheThreeReadTools(t *testing.T) {
t.Errorf("%s has no description or no run", tl.Name) t.Errorf("%s has no description or no run", tl.Name)
} }
} }
want := []string{check.VerdictsReadTool, check.PullReadTool, check.KeptReadTool} want := []string{check.VerdictsReadTool, check.PullReadTool, check.KeptReadTool, check.PreviewsReadTool, check.StatusReadTool}
if len(names) != 3 || names[0] != want[0] || names[1] != want[1] || names[2] != want[2] { if !slices.Equal(names, want) {
t.Errorf("tools = %v, want %v", names, want) t.Errorf("tools = %v, want %v", names, want)
} }
// With nothing to read from a tool says so and changes nothing. // With nothing to read from a tool says so and changes nothing.
for _, tl := range Tools(check.Reader{}) { for _, tl := range Tools(check.Reader{}) {
if tl.Name == check.StatusReadTool {
continue // it answers from what it can read, and says what it could not
}
if _, err := tl.Run(map[string]any{"repository": "novox/x", "number": float64(1)}); err == nil { if _, err := tl.Run(map[string]any{"repository": "novox/x", "number": float64(1)}); err == nil {
t.Errorf("%s answered with no state to read", tl.Name) t.Errorf("%s answered with no state to read", tl.Name)
} }
} }
} }
// memStore is a tries state in memory.
type memStore struct{ vals map[string]json.RawMessage }
func (m *memStore) Keys() ([]string, error) {
var out []string
for k := range m.vals {
out = append(out, k)
}
return out, nil
}
func (m *memStore) Get(key string) (json.RawMessage, error) { return m.vals[key], nil }
func (m *memStore) Create(key string, value []byte) error {
if _, ok := m.vals[key]; ok {
return errors.New("exists")
}
m.vals[key] = value
return nil
}
func TestTheTryToolsAskAndReadATry(t *testing.T) { // ADR 0501 §5, the agreed interface with mesh-controller's try
store := &memStore{vals: map[string]json.RawMessage{}}
tools := TryTools(Trying{Tries: store, Now: func() time.Time { return time.Unix(1, 5) },
Trunks: func() (map[string]string, error) {
return map[string]string{"novox/mesh-delivery": strings.Repeat("a", 40)}, nil
}})
if len(tools) != 2 || tools[0].Name != "checker_try" || tools[1].Name != "checker_try_read" {
t.Fatalf("tools = %+v", tools)
}
out, err := tools[0].Run(map[string]any{"repository": "novox/mesh-delivery", "head": strings.Repeat("1", 40), "trunk": "main"})
if err != nil {
t.Fatal(err)
}
id := out.(map[string]any)["id"].(string)
if id != "try-1000000005" {
t.Errorf("id = %q", id)
}
raw, err := tools[1].Run(map[string]any{"id": id})
if err != nil || !strings.Contains(string(raw.(json.RawMessage)), `"state":"queued"`) {
t.Errorf("read = %s, %v", raw, err)
}
if _, err := tools[0].Run(map[string]any{"repository": "novox/hq", "head": strings.Repeat("1", 40)}); err == nil ||
err.Error() != "no level of novox/hq is checked by the checker" {
t.Errorf("hq: %v", err)
}
}
@@ -57,6 +57,9 @@ func (s kvStore) Keys(ctx context.Context) ([]string, error) {
return keys, err return keys, err
} }
// Delete removes a key (an ended try past its keeping).
func (s kvStore) Delete(ctx context.Context, key string) error { return s.kv.Delete(ctx, key) }
func opOf(e jetstream.KeyValueEntry) bool { return e.Operation() == jetstream.KeyValuePut } func opOf(e jetstream.KeyValueEntry) bool { return e.Operation() == jetstream.KeyValuePut }
// settle reads the entries a watcher holds now, up to its mark of the last values, and returns the watcher for the // settle reads the entries a watcher holds now, up to its mark of the last values, and returns the watcher for the
@@ -3,10 +3,13 @@ package main
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"regexp" "regexp"
"strings" "strings"
"github.com/nats-io/nats.go"
"mesh-delivery-checker/internal/check" "mesh-delivery-checker/internal/check"
) )
@@ -27,6 +30,16 @@ type forgeReads struct {
call func(key string, body any) (json.RawMessage, error) call func(key string, body any) (json.RawMessage, error)
} }
// ask calls a forge tool; a call that reached nobody (no responders: the forge's tool runner re-serving, or the forge
// down) is marked Unheard, so the check fails at the forge and is judged again when the nodes change (issue 1624).
func (f forgeReads) ask(key string, body any) (json.RawMessage, error) {
raw, err := f.call(key, body)
if errors.Is(err, nats.ErrNoResponders) || errors.Is(err, nats.ErrTimeout) || errors.Is(err, context.DeadlineExceeded) {
return nil, &check.Unheard{Err: err}
}
return raw, err
}
func ownerRepo(repository string) (string, string) { func ownerRepo(repository string) (string, string) {
owner, repo, _ := strings.Cut(repository, "/") owner, repo, _ := strings.Cut(repository, "/")
return owner, repo return owner, repo
@@ -35,7 +48,7 @@ func ownerRepo(repository string) (string, string) {
// File is the text of a file of a repository at a ref. // File is the text of a file of a repository at a ref.
func (f forgeReads) File(_ context.Context, repository, path, ref string) (string, error) { func (f forgeReads) File(_ context.Context, repository, path, ref string) (string, error) {
owner, repo := ownerRepo(repository) owner, repo := ownerRepo(repository)
raw, err := f.call(getFileTool, map[string]string{"owner": owner, "repo": repo, "path": path, "ref": ref}) raw, err := f.ask(getFileTool, map[string]string{"owner": owner, "repo": repo, "path": path, "ref": ref})
if err != nil { if err != nil {
if notFoundStatus.MatchString(err.Error()) { if notFoundStatus.MatchString(err.Error()) {
// The forge answers 404 for a file the ref does not hold (issue 1845). The status is matched in its own place, // The forge answers 404 for a file the ref does not hold (issue 1845). The status is matched in its own place,
@@ -59,7 +72,7 @@ func (f forgeReads) File(_ context.Context, repository, path, ref string) (strin
// not. // not.
func (f forgeReads) OnTrunk(_ context.Context, repository, commit string) (bool, error) { func (f forgeReads) OnTrunk(_ context.Context, repository, commit string) (bool, error) {
owner, repo := ownerRepo(repository) owner, repo := ownerRepo(repository)
raw, err := f.call(compareTool, map[string]string{"owner": owner, "repo": repo, "base": "main", "head": commit}) raw, err := f.ask(compareTool, map[string]string{"owner": owner, "repo": repo, "base": "main", "head": commit})
if err != nil { if err != nil {
return false, err return false, err
} }
@@ -75,7 +88,7 @@ func (f forgeReads) OnTrunk(_ context.Context, repository, commit string) (bool,
// FullSHA is the whole sha of a commit named by its first digits: the newest commit of the history at that ref. // FullSHA is the whole sha of a commit named by its first digits: the newest commit of the history at that ref.
func (f forgeReads) FullSHA(_ context.Context, repository, commit string) (string, error) { func (f forgeReads) FullSHA(_ context.Context, repository, commit string) (string, error) {
owner, repo := ownerRepo(repository) owner, repo := ownerRepo(repository)
raw, err := f.call(commitsTool, map[string]any{"owner": owner, "repo": repo, "ref": commit, "limit": 1}) raw, err := f.ask(commitsTool, map[string]any{"owner": owner, "repo": repo, "ref": commit, "limit": 1})
if err != nil { if err != nil {
return "", err return "", err
} }
@@ -9,6 +9,8 @@ import (
"strings" "strings"
"testing" "testing"
"github.com/nats-io/nats.go"
"mesh-delivery-checker/internal/check" "mesh-delivery-checker/internal/check"
) )
@@ -86,3 +88,18 @@ func TestAFileTheRefDoesNotHoldIsNotFoundAndOtherFailuresAreNot(t *testing.T) {
t.Errorf("a 500 is %v, want a failure that is not ErrNotFound", err) t.Errorf("a 500 is %v, want a failure that is not ErrNotFound", err)
} }
} }
func TestAForgeReadThatReachesNobodyIsUnheard(t *testing.T) { // 1624 C7: "nats: no responders" on mesh-catalog #293
f := forgeReads{call: func(string, any) (json.RawMessage, error) { return nil, nats.ErrNoResponders }}
_, err := f.File(context.Background(), "novox/mesh-catalog", "checks/agenthome/go.mod", "950e37a46e8d")
var u *check.Unheard
if !errors.As(err, &u) {
t.Errorf("err = %v, want it marked as reaching nobody", err)
}
if _, err := f.OnTrunk(context.Background(), "novox/mesh-sdk", "abc"); !errors.As(err, &u) {
t.Errorf("compare: %v", err)
}
if _, err := f.FullSHA(context.Background(), "novox/mesh-sdk", "abc"); !errors.As(err, &u) {
t.Errorf("commits: %v", err)
}
}
@@ -19,7 +19,10 @@ import (
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream" "github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery"
"mesh-delivery-checker/internal/check"
) )
// busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the // busConf is the shape of the composed accounts file (mesh-controller internal/broker): one account with JetStream, the
@@ -267,7 +270,7 @@ func startBed(t *testing.T) *bed {
for name, history := range map[string]int{delivery.CheckerVerdictsBucket: delivery.CheckerVerdictsHistory, delivery.CheckerKeptBucket: delivery.CheckerKeptHistory, for name, history := range map[string]int{delivery.CheckerVerdictsBucket: delivery.CheckerVerdictsHistory, delivery.CheckerKeptBucket: delivery.CheckerKeptHistory,
delivery.CheckerOwedBucket: delivery.CheckerOwedHistory, delivery.PreviewsBucket: delivery.PreviewsHistory, delivery.CandidatesBucket: delivery.CandidatesHistory, delivery.CheckerOwedBucket: delivery.CheckerOwedHistory, delivery.PreviewsBucket: delivery.PreviewsHistory, delivery.CandidatesBucket: delivery.CandidatesHistory,
delivery.ModulesBucket: delivery.ModulesHistory, delivery.ManifestsBucket: delivery.ManifestsHistory, delivery.SourceBucket: delivery.SourceHistory, delivery.ModulesBucket: delivery.ModulesHistory, delivery.ManifestsBucket: delivery.ManifestsHistory, delivery.SourceBucket: delivery.SourceHistory,
delivery.PresenceBucket: delivery.PresenceHistory, delivery.HaveBucket: delivery.HaveHistory} { delivery.PresenceBucket: delivery.PresenceHistory, delivery.HaveBucket: delivery.HaveHistory, check.TriesBucket: 1} {
kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)}) kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: uint8(history)})
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -590,3 +593,91 @@ func TestAMemberNamingAMeshSDKCommitOffTheTrunkFailsNamingTheCommit(t *testing.T
t.Errorf("%d test steps ran after a stale pin", n) t.Errorf("%d test steps ran after a stale pin", n)
} }
} }
// triesStore is the tries bucket as the tools write it.
type triesStore struct {
kv jetstream.KeyValue
ctx context.Context
}
func (s triesStore) Keys() ([]string, error) {
keys, err := s.kv.Keys(s.ctx)
if err == jetstream.ErrNoKeysFound {
return nil, nil
}
return keys, err
}
func (s triesStore) Get(key string) (json.RawMessage, error) {
e, err := s.kv.Get(s.ctx, key)
if err == jetstream.ErrKeyNotFound {
return nil, nil
}
if err != nil {
return nil, err
}
return e.Value(), nil
}
func (s triesStore) Create(key string, value []byte) error {
_, err := s.kv.Create(s.ctx, key, value)
return err
}
// TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed: a try written as checker_try writes it is taken by the running checker,
// judged through the build modules as a pull request's head merged onto its trunk, and ended pass in its key, with no
// verdict and no status owed (ADR 0501 §5).
func TestATryIsJudgedAndItsEndWrittenWithNoStatusOwed(t *testing.T) {
b := startBed(t)
b.start(t, nil)
store := triesStore{b.buckets[check.TriesBucket], b.ctx}
id, err := check.AskTry(store, map[string]string{repo: trunk}, check.TryAsk{Repository: repo, Head: head1}, time.Now())
if err != nil {
t.Fatal(err)
}
var last check.Try
for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
raw, _ := store.Get(id)
if last, err = check.DecodeTry(raw); err == nil && last.Over() {
break
}
}
if last.State != check.TryPass || last.Trunk != trunk || last.Took == "" {
t.Fatalf("try = %+v, want passed on the trunk", last)
}
if b.fake.count("mesh-build-go.build_go_fetch") == 0 {
t.Errorf("the try fetched nothing")
}
for _, name := range []string{delivery.CheckerOwedBucket, delivery.CheckerVerdictsBucket, delivery.PreviewsBucket} {
if keys, _ := b.buckets[name].Keys(b.ctx); len(keys) != 0 {
t.Errorf("a try wrote %s: %v", name, keys)
}
}
}
// TestTheCheckerSaysOwedReadableOnTheBus: started, the checker says on its own states subject what it keeps, owed among
// the readable ones, which is what grants the forge module its read of owed (ADR 0495 §2-§3, ADR 0521; R4e finding R2-1).
func TestTheCheckerSaysOwedReadableOnTheBus(t *testing.T) { // R2-1
b := startBed(t)
if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
t.Fatal(err)
}
b.start(t, nil)
events, err := b.js.Stream(b.ctx, "EVENTS")
if err != nil {
t.Fatal(err)
}
var said announce.States
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
m, err := events.GetLastMsgForSubject(b.ctx, announce.Subject("mesh-delivery-checker"))
if err != nil {
continue
}
if err := json.Unmarshal(m.Data, &said); err != nil {
t.Fatal(err)
}
break
}
if said.Module != "mesh-delivery-checker" || said.Node != "n1" || !said.Keeps("owed") || !said.Readable("owed") ||
!said.Readable("previews") || !said.Readable("verdicts") || said.Readable("kept") || said.Readable("tries") || len(said.States) != 5 {
t.Fatalf("the checker said %+v, want its five states with owed, previews and verdicts readable", said)
}
}
@@ -3,6 +3,7 @@ package main
import ( import (
"context" "context"
"sort" "sort"
"strings"
"sync" "sync"
"git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/catalogue"
@@ -31,7 +32,7 @@ type judgeDeps struct {
func newJudge(d judgeDeps) check.Judge { func newJudge(d judgeDeps) check.Judge {
return &check.Examiner{Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places, return &check.Examiner{Steps: d.runner, Toolchains: d.tools, Presence: d.presence, Health: d.have, Places: d.places,
Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf} Kept: check.Kept{Store: d.kept}, Previews: d.previews, Modules: d.repos, Packages: d.repos, Forge: d.forge, Runs: d.repos, Builds: d.builds, Announce: d.announce, Trunks: d.trunks, Logf: d.logf}
} }
// manifestKV reads a manifest by its digest. // manifestKV reads a manifest by its digest.
@@ -47,10 +48,13 @@ type repoModules struct {
logf func(format string, args ...any) logf func(format string, args ...any)
mu sync.Mutex mu sync.Mutex
langs map[string]map[string]bool // by manifest digest: the languages of its artifacts langs map[string]map[string]bool // by manifest digest: the languages of its artifacts
repoOf map[string]string // by module: its repository tsPkgs map[string][]string // by manifest digest: the names of its TypeScript package artifacts
commit map[string]string // by module: the commit its record was computed at (the commit the mesh runs) path map[string]string // by module: its directory in its repository
typed map[string]string // by module: its manifest digest named map[string]map[string][]string // by module: each artifact's named paths, as the indexer keeps them
repoOf map[string]string // by module: its repository
commit map[string]string // by module: the commit its record was computed at (the commit the mesh runs)
typed map[string]string // by module: its manifest digest
} }
// Put takes a change of the indexer's modules bucket in: a module's repository and the languages its manifest names. // Put takes a change of the indexer's modules bucket in: a module's repository and the languages its manifest names.
@@ -58,11 +62,14 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted
r.mu.Lock() r.mu.Lock()
if r.repoOf == nil { if r.repoOf == nil {
r.repoOf, r.typed, r.langs, r.commit = map[string]string{}, map[string]string{}, map[string]map[string]bool{}, map[string]string{} r.repoOf, r.typed, r.langs, r.commit = map[string]string{}, map[string]string{}, map[string]map[string]bool{}, map[string]string{}
r.tsPkgs, r.path, r.named = map[string][]string{}, map[string]string{}, map[string]map[string][]string{}
} }
if deleted { if deleted {
delete(r.repoOf, key) delete(r.repoOf, key)
delete(r.typed, key) delete(r.typed, key)
delete(r.commit, key) delete(r.commit, key)
delete(r.path, key)
delete(r.named, key)
r.mu.Unlock() r.mu.Unlock()
return return
} }
@@ -75,12 +82,15 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted
r.repoOf[key] = m.Repository r.repoOf[key] = m.Repository
r.typed[key] = m.Manifest r.typed[key] = m.Manifest
r.commit[key] = m.Commit r.commit[key] = m.Commit
r.path[key] = m.Path
r.named[key] = m.ArtifactPaths
_, known := r.langs[m.Manifest] _, known := r.langs[m.Manifest]
r.mu.Unlock() r.mu.Unlock()
if m.Manifest == "" || known { if m.Manifest == "" || known {
return return
} }
set := map[string]bool{} set := map[string]bool{}
var pkgs []string
if mk, err := delivery.ManifestsKey(m.Manifest); err == nil { if mk, err := delivery.ManifestsKey(m.Manifest); err == nil {
if raw, _, found, err := r.manifests.Get(ctx, mk); err == nil && found { if raw, _, found, err := r.manifests.Get(ctx, mk); err == nil && found {
if kept, err := delivery.DecodeManifest(raw); err == nil { if kept, err := delivery.DecodeManifest(raw); err == nil {
@@ -89,6 +99,9 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted
if a.Language != "" { if a.Language != "" {
set[a.Language] = true set[a.Language] = true
} }
if a.Kind == "package" && a.Language == "typescript" {
pkgs = append(pkgs, a.Name)
}
} }
} else if err != nil && r.logf != nil { } else if err != nil && r.logf != nil {
r.logf("the manifest %s cannot be read, so its module counts as no TypeScript: %v", m.Manifest, err) r.logf("the manifest %s cannot be read, so its module counts as no TypeScript: %v", m.Manifest, err)
@@ -98,9 +111,37 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted
} }
r.mu.Lock() r.mu.Lock()
r.langs[m.Manifest] = set r.langs[m.Manifest] = set
r.tsPkgs[m.Manifest] = pkgs
r.mu.Unlock() r.mu.Unlock()
} }
// PackagePaths are the paths, relative to the repository, that the TypeScript package artifacts of a repository's modules
// are built from, sorted: the named paths the indexer keeps in each module's value (delivery.Module.ArtifactPaths; only the
// indexer reads them from a manifest, the lead's decision L8), under the module's directory. None for a module whose value
// holds no paths.
func (r *repoModules) PackagePaths(repository string) []string {
r.mu.Lock()
defer r.mu.Unlock()
var out []string
for module, repo := range r.repoOf {
if repo != repository {
continue
}
dir := strings.Trim(r.path[module], "/")
for _, name := range r.tsPkgs[r.typed[module]] {
for _, p := range r.named[module][name] {
p = strings.Trim(strings.TrimPrefix(p, "./"), "/")
if dir != "" && dir != delivery.RootPath {
p = dir + "/" + p
}
out = append(out, p)
}
}
}
sort.Strings(out)
return out
}
// BuildModules are the build modules that judge a repository, sorted. // BuildModules are the build modules that judge a repository, sorted.
func (r *repoModules) BuildModules(repository string) []string { func (r *repoModules) BuildModules(repository string) []string {
out := []string{buildrun.GoBuildModule} out := []string{buildrun.GoBuildModule}
@@ -73,3 +73,40 @@ func TestARepositoryWithATypeScriptModuleIsJudgedByTheNodeBuildModuleToo(t *test
t.Errorf("a repository with no module: %v, want the Go build module", got) t.Errorf("a repository with no module: %v, want the Go build module", got)
} }
} }
// sdkManifest is mesh-sdk's own manifest (its module.json at feat/r4e-sdk): one TypeScript package naming its paths.
const sdkManifest = `{"module":"mesh-sdk","version":"1","slug":"sdk","build":{"artifacts":[{"name":"lib","kind":"package","language":"typescript",` +
`"paths":["src","test","package.json","package-lock.json","tsconfig.json"]}]},"resources":[]}`
func TestTheTypeScriptPackagesPathsComeFromTheIndexersModulesNotTheManifest(t *testing.T) { // ADR 0500 §5 with lead decision L8
store := memManifests{}
data, err := delivery.EncodeManifest(delivery.Manifest{Module: "mesh-sdk", JSON: sdkManifest})
if err != nil {
t.Fatal(err)
}
store[delivery.ManifestDigest([]byte(sdkManifest))] = data
r := &repoModules{manifests: store}
put := func(path string, paths map[string][]string) {
v, err := delivery.EncodeModule(delivery.Module{Repository: "novox/mesh-sdk", Commit: strings.Repeat("a", 40), FP: "src1:" + strings.Repeat("a", 64),
Manifest: delivery.ManifestDigest([]byte(sdkManifest)), Path: path, ArtifactPaths: paths})
if err != nil {
t.Fatal(err)
}
r.Put(context.Background(), "mesh-sdk", v, false)
}
put(".", map[string][]string{"lib": {"src", "package.json"}}) // the indexer's word, not the manifest's five paths
if got := strings.Join(r.PackagePaths("novox/mesh-sdk"), ","); got != "package.json,src" {
t.Errorf("paths = %q, want the indexer's, at the repository's root", got)
}
put("ts", map[string][]string{"lib": {"src"}})
if got := strings.Join(r.PackagePaths("novox/mesh-sdk"), ","); got != "ts/src" {
t.Errorf("paths = %q, want them under the module's directory", got)
}
put(".", nil) // a value written before the indexer kept paths: nothing to serve by
if got := r.PackagePaths("novox/mesh-sdk"); len(got) != 0 {
t.Errorf("paths = %v, want none", got)
}
if got := r.PackagePaths("novox/mesh-delivery"); len(got) != 0 {
t.Errorf("another repository: %v", got)
}
}
@@ -0,0 +1,55 @@
package main
import (
"encoding/json"
"os"
"sort"
"testing"
)
// The checker says on the bus which states it keeps and which others may read (novox/hq ADR 0495 clauses 2 and 3): a
// reader is granted from that announcement. owed is read by the forge module, which writes mesh/check from it (ADR 0521,
// R4e round 2 finding R2-1), and by the forge bridge; previews by the indexer; verdicts by the gate. The list names every
// state of the manifest, so no state is refused at the call for want of being announced (clause 5).
func TestTheCheckerAnnouncesEveryStateItKeepsAndOwedReadable(t *testing.T) { // R2-1, ADR 0495 §2-§3, ADR 0521
raw, err := os.ReadFile("../../module.json")
if err != nil {
t.Fatal(err)
}
var m struct {
State []struct{ Name string } `json:"state"`
}
if err := json.Unmarshal(raw, &m); err != nil {
t.Fatal(err)
}
var manifest, announced, readable []string
for _, s := range m.State {
manifest = append(manifest, s.Name)
}
for _, s := range keptStates {
announced = append(announced, s.Name)
if s.Readable {
readable = append(readable, s.Name)
}
}
sort.Strings(manifest)
sort.Strings(announced)
if !equal(manifest, announced) {
t.Errorf("announced %v, the manifest keeps %v", announced, manifest)
}
if want := []string{"owed", "previews", "verdicts"}; !equal(readable, want) {
t.Errorf("readable %v, want %v", readable, want)
}
}
func equal(a, b []string) bool {
if len(a) != len(b) {
return false
}
for i := range a {
if a[i] != b[i] {
return false
}
}
return true
}
@@ -25,6 +25,7 @@ import (
"github.com/nats-io/nats.go" "github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream" "github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/announce"
"git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun" "git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
"git.novox.be/novox/mesh-sdk/go/process" "git.novox.be/novox/mesh-sdk/go/process"
@@ -113,6 +114,16 @@ func durationEnv(getenv func(string) string, name string, def time.Duration) (ti
} }
// run is the program: it returns when ctx ends or the bus cannot be used. // run is the program: it returns when ctx ends or the bus cannot be used.
// keptStates is every state the checker keeps and which others may read (ADR 0495 §2-§3): owed by the forge module,
// which writes mesh/check from it (ADR 0521), and by the forge bridge; previews by the indexer; verdicts by the gate.
var keptStates = []announce.State{
{Name: "kept"},
{Name: "owed", Readable: true},
{Name: "previews", Readable: true},
{Name: "tries"},
{Name: "verdicts", Readable: true},
}
func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error { func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
logger := log.New(logOut, "mesh-delivery-checker: ", log.LstdFlags) logger := log.New(logOut, "mesh-delivery-checker: ", log.LstdFlags)
path := strings.TrimSpace(getenv("MESH_BROKER_FILE")) path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
@@ -151,7 +162,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
} }
buckets := map[string]jetstream.KeyValue{} buckets := map[string]jetstream.KeyValue{}
for _, name := range []string{delivery.CheckerVerdictsBucket, delivery.CheckerKeptBucket, delivery.CheckerOwedBucket, delivery.PreviewsBucket, for _, name := range []string{delivery.CheckerVerdictsBucket, delivery.CheckerKeptBucket, delivery.CheckerOwedBucket, delivery.PreviewsBucket,
delivery.CandidatesBucket, delivery.ModulesBucket, delivery.ManifestsBucket} { delivery.CandidatesBucket, delivery.ModulesBucket, delivery.ManifestsBucket, check.TriesBucket} {
if buckets[name], err = bucket(name); err != nil { if buckets[name], err = bucket(name); err != nil {
return err return err
} }
@@ -167,6 +178,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
return err return err
} }
defer caller.Close() defer caller.Close()
// What the checker keeps, said on the bus (ADR 0495 §2-§3): a reader is granted from it. Not fatal when it cannot be said
// now: process.Conn says it again at every reconnect.
if err := caller.Keeps(keptStates...); err != nil {
logger.Printf("what the checker keeps could not be said on the bus, so readers of owed, previews and verdicts are not granted until it is: %v", err)
}
stepBound, err := durationEnv(getenv, "MESH_CHECKER_STEP_BOUND", defaultStepBound) stepBound, err := durationEnv(getenv, "MESH_CHECKER_STEP_BOUND", defaultStepBound)
if err != nil { if err != nil {
@@ -209,8 +225,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
default: default:
builds = buildReader{kvStore{kv}} builds = buildReader{kvStore{kv}}
} }
holds := newHoldWatch(ctx, js, ch.Reconsider, logger) // The tries (ADR 0501 §5) are judged by the same judge, held by the same deliveries.
tryer := &check.Tryer{Tries: kvStore{buckets[check.TriesBucket]}, Bound: checkBound, Logf: logger.Printf}
holds := newHoldWatch(ctx, js, func(ctx context.Context) { ch.Reconsider(ctx); tryer.Reconsider(ctx) }, logger)
ch.Hold = holds.Held ch.Hold = holds.Held
tryer.Hold = holds.Held
announce := func(ctx context.Context, event, key string) { announce := func(ctx context.Context, event, key string) {
body, _ := json.Marshal(map[string]string{"key": key}) body, _ := json.Marshal(map[string]string{"key": key})
if _, err := js.Publish(ctx, eventSubject(event), body); err != nil { if _, err := js.Publish(ctx, eventSubject(event), body); err != nil {
@@ -225,6 +244,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
announce: announce, announce: announce,
}) })
ch.Announce = announce ch.Announce = announce
tryer.Judge = ch.Judge
// The toolchains first, so a check that starts from the first candidate finds the nodes. // The toolchains first, so a check that starts from the first candidate finds the nodes.
watch := newToolchainWatch(ctx, js, tools, ch.NodesChanged, logger) watch := newToolchainWatch(ctx, js, tools, ch.NodesChanged, logger)
@@ -308,7 +328,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
go func() { go func() {
errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) { errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) {
if e.Key() == delivery.SourceMainKey && trunks.Put(e.Value(), !opOf(e)) { if e.Key() == delivery.SourceMainKey && trunks.Put(e.Value(), !opOf(e)) {
ch.Reconsider(ctx) // the trunk moved: a verdict for the old trunk is judged again ch.TrunkMoved(ctx) // the trunk moved: its statuses go pending, and a verdict for the old trunk is judged again
} }
}) })
}() }()
@@ -336,11 +356,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err := ch.Sweep(ctx); err != nil { if err := ch.Sweep(ctx); err != nil {
return err return err
} }
// The tries: a queued one starts, one left running is taken over, an ended one past its keeping is removed.
tw, err := settle(ctx, buckets[check.TriesBucket], func(jetstream.KeyValueEntry) {})
if err != nil {
return err
}
tryer.Reconsider(ctx)
go func() {
errs <- follow(ctx, tw, func(e jetstream.KeyValueEntry) {
if opOf(e) {
tryer.Put(ctx, e.Key(), e.Value())
}
})
}()
logger.Printf("started on %s", cred.Node) logger.Printf("started on %s", cred.Node)
go func() { errs <- follow(ctx, w, each) }() go func() { errs <- follow(ctx, w, each) }()
select { select {
case <-ctx.Done(): case <-ctx.Done():
ch.Wait() ch.Wait()
tryer.Wait()
return nil return nil
case err := <-errs: case err := <-errs:
return err return err
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-checker
go 1.26.0 go 1.26.0
require ( require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
) )
+2 -6
View File
@@ -1,9 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc h1:ZJScJSnF6LgFl55/VJ9RT/lNroVWun469sXtx0nlVYw= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7 h1:wK9O1CgNk7mo3IdJqW/CCx/hebg+1F0rkvW8CvJfnRM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b h1:OjlhfiGr+wKLWxafV+BRjZx3TbBtwRYcm9mgoam/jv8=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -4,6 +4,7 @@ import (
"fmt" "fmt"
"maps" "maps"
"sort" "sort"
"strings"
"sync" "sync"
"time" "time"
@@ -35,13 +36,17 @@ type Case struct {
Key string Key string
Group string Group string
HeadsHash string HeadsHash string
Pulls []Pull // Heads is the heads hash the gate matches an Approve's members by (GateHeads): a verdict carries it.
Heads string
Pulls []Pull
// Trunk is the trunk head of each repository of Pulls. // Trunk is the trunk head of each repository of Pulls.
Trunk map[string]string Trunk map[string]string
// Left says why pull requests of the candidate are not judged (hq's, a repository with no known trunk); empty when all are. // Left says why pull requests of the candidate are not judged (hq's, a repository with no known trunk); empty when all are.
Left []string Left []string
// Queued is when the checker took the case up, for the time of its queue. // Queued is when the checker took the case up, for the time of its queue.
Queued time.Time Queued time.Time
// Try narrows a try's check (its level, its tests); nil for a pull request's.
Try *TryOptions
// Progress says what the check is doing, so a check that ran past its bound names the step it was stuck at. // Progress says what the check is doing, so a check that ran past its bound names the step it was stuck at.
Progress *Progress Progress *Progress
} }
@@ -75,7 +80,7 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case {
if err != nil { if err != nil {
return nil return nil
} }
cs := &Case{Key: key, Group: c.Group, HeadsHash: c.HeadsHash, Trunk: map[string]string{}, Progress: &Progress{}} cs := &Case{Key: key, Group: c.Group, HeadsHash: c.HeadsHash, Heads: GateHeads(c), Trunk: map[string]string{}, Progress: &Progress{}}
for _, p := range c.Pulls { for _, p := range c.Pulls {
switch head, known := trunk[p.Repository]; { switch head, known := trunk[p.Repository]; {
case p.Repository == HQRepository: case p.Repository == HQRepository:
@@ -99,6 +104,22 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case {
return cs return cs
} }
// GateHeads is the hash the gate matches the members an Approve covers by (mesh-catalog mesh-delivery-gate onestatus.go
// headsOf): delivery.HeadsHash over every pull request of the candidate, hq's included, as
// `<owner/name lowercase>#<number>=<full head sha lowercase>`. A lone pull request is a group of one. Empty for a candidate
// with no pull request.
func GateHeads(c delivery.Candidate) string {
heads := make([]delivery.Head, 0, len(c.Pulls))
for _, p := range c.Pulls {
heads = append(heads, delivery.Head{Repository: strings.ToLower(p.Repository), Number: p.Number, Head: strings.ToLower(p.Head.Head)})
}
hash, err := delivery.HeadsHash(heads)
if err != nil {
return ""
}
return hash
}
// sameTrunk says whether two verdicts' trunk heads are the same. // sameTrunk says whether two verdicts' trunk heads are the same.
func sameTrunk(a, b map[string]string) bool { return maps.Equal(a, b) } func sameTrunk(a, b map[string]string) bool { return maps.Equal(a, b) }
@@ -18,6 +18,10 @@ const DefaultBound = 30 * time.Minute
// is judged again when the nodes change, by a watch and never a timer (ADR 0437 §8). // is judged again when the nodes change, by a watch and never a timer (ADR 0437 §8).
const NodesWords = "build node qualifies" const NodesWords = "build node qualifies"
// UnheardWords end the Words of a failure because a part of the mesh the check reads was not heard (the forge, issue 1624):
// like a failure of the nodes, it is not the change's, and is judged again when the nodes change.
const UnheardWords = "not heard; judged again when the nodes change"
// The events the checker says on its stream checks (module.json): a preview or a status was written, so a part that reads // The events the checker says on its stream checks (module.json): a preview or a status was written, so a part that reads
// the bucket, which is made when the checker is assigned and may not exist when that part starts, finds it (ADR 0437 §8: no // the bucket, which is made when the checker is assigned and may not exist when that part starts, finds it (ADR 0437 §8: no
// timer). The body is {"key": ...}. // timer). The body is {"key": ...}.
@@ -234,7 +238,7 @@ type claimed struct{ rev uint64 }
// - a pass or a failure for the trunk heads the case has stands; one for other trunk heads (the trunk moved) is taken // - a pass or a failure for the trunk heads the case has stands; one for other trunk heads (the trunk moved) is taken
// again by compare-and-set on the same key, and so is a failure of the nodes after the nodes changed. // again by compare-and-set on the same key, and so is a failure of the nodes after the nodes changed.
func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64) (*claimed, error) { func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64) (*claimed, error) {
running := delivery.Verdict{Group: cs.Group, Heads: cs.HeadsHash, Trunk: cs.Trunk, State: delivery.VerdictRunning} running := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictRunning}
data, err := delivery.EncodeVerdict(running) data, err := delivery.EncodeVerdict(running)
if err != nil { if err != nil {
return nil, err return nil, err
@@ -287,7 +291,7 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64
// IsNodesFailure says whether a verdict failed because no build node could take the check. // IsNodesFailure says whether a verdict failed because no build node could take the check.
func IsNodesFailure(v delivery.Verdict) bool { func IsNodesFailure(v delivery.Verdict) bool {
return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, NodesWords) return v.State == delivery.VerdictFail && (strings.Contains(v.Failed.Words, NodesWords) || strings.Contains(v.Failed.Words, UnheardWords))
} }
// try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends // try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends
@@ -313,14 +317,14 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
return return
} }
v := out.Verdict v := out.Verdict
v.Group, v.Heads, v.Trunk = cs.Group, cs.HeadsHash, cs.Trunk v.Group, v.Heads, v.Trunk = cs.Group, cs.Heads, cs.Trunk
if v.Took.Queue == 0 && !cs.Queued.IsZero() && began.After(cs.Queued) { if v.Took.Queue == 0 && !cs.Queued.IsZero() && began.After(cs.Queued) {
v.Took.Queue = began.Sub(cs.Queued).Milliseconds() v.Took.Queue = began.Sub(cs.Queued).Milliseconds()
} }
data, err := delivery.EncodeVerdict(v) data, err := delivery.EncodeVerdict(v)
if err != nil { if err != nil {
ch.logf("%s: the verdict cannot be written: %v", cs.Key, err) ch.logf("%s: the verdict cannot be written: %v", cs.Key, err)
v = delivery.Verdict{Group: cs.Group, Heads: cs.HeadsHash, Trunk: cs.Trunk, State: delivery.VerdictFail, v = delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictFail,
Failed: delivery.FailedFirst{Machine: "checker", Words: "the verdict could not be written: " + err.Error()}} Failed: delivery.FailedFirst{Machine: "checker", Words: "the verdict could not be written: " + err.Error()}}
if data, err = delivery.EncodeVerdict(v); err != nil { if data, err = delivery.EncodeVerdict(v); err != nil {
return return
@@ -335,6 +339,13 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
ch.nodesFail[cs.Key] = epoch ch.nodesFail[cs.Key] = epoch
ch.mu.Unlock() ch.mu.Unlock()
} }
if ch.movedSince(cs) {
// Judged on a trunk that moved while it ran: the verdict is kept as judged, and the status stays pending until the
// check of the new trunk, which Reconsider starts, ends.
ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, nil, nil)
ch.logf("%s: %s on a trunk that moved while it ran; the status stays pending", cs.Key, v.State)
return
}
ch.writeOwed(ctx, cs, v, out.Builds, out.Slowest) ch.writeOwed(ctx, cs, v, out.Builds, out.Slowest)
ch.logf("%s: %s", cs.Key, v.State) ch.logf("%s: %s", cs.Key, v.State)
} }
@@ -342,25 +353,7 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
// judge runs the case within the bound. A judge that fails to judge, and a check that ran past its bound, are verdicts of // judge runs the case within the bound. A judge that fails to judge, and a check that ran past its bound, are verdicts of
// failure naming the machine and the step. // failure naming the machine and the step.
func (ch *Checker) judge(ctx context.Context, cs *Case) Outcome { func (ch *Checker) judge(ctx context.Context, cs *Case) Outcome {
bound := ch.bound() out, _ := judgeWithin(ctx, ch.Judge, cs, ch.bound())
run, cancel := context.WithTimeoutCause(ctx, bound, errBound)
defer cancel()
out, err := ch.Judge.Judge(run, cs)
stage, node := cs.Progress.Now()
if node == "" {
node = "checker"
}
switch {
case ctx.Err() != nil:
return Outcome{}
case context.Cause(run) == errBound:
if stage == "" {
stage = "starting"
}
return Outcome{Verdict: failure(node, fmt.Sprintf("the check ran past its bound of %s, stuck at: %s", bound, stage))}
case err != nil:
return Outcome{Verdict: failure(node, err.Error())}
}
return out return out
} }
@@ -467,3 +460,44 @@ func (ch *Checker) Sweep(ctx context.Context) error {
} }
return nil return nil
} }
// TrunkMoved is Reconsider after a move of a trunk, which first owes mesh/check pending on every open pull request of a
// candidate whose verdict was judged on other trunk heads, every member's for a group (the lead's decision L4, the checker's
// part of issue 1302): the forge's success of the old trunk never stands while the check waits for a place or for a delivery
// that holds it, so the gate never reads it as the head's.
func (ch *Checker) TrunkMoved(ctx context.Context) {
ch.init()
ch.mu.Lock()
known := make(map[string]delivery.Candidate, len(ch.known))
for k, c := range ch.known {
known[k] = c
}
ch.mu.Unlock()
heads := ch.heads()
for _, c := range known {
cs := CaseOf(c, heads)
if cs == nil {
continue
}
value, _, found, err := ch.Verdicts.Get(ctx, cs.Key)
if err != nil || !found {
continue // never judged: no status to put out of date
}
if v, err := delivery.DecodeVerdict(value); err == nil && sameTrunk(v.Trunk, cs.Trunk) {
continue
}
ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, nil, nil)
}
ch.Reconsider(ctx)
}
// movedSince says whether a trunk the case was judged on is no longer the trunk's head.
func (ch *Checker) movedSince(cs *Case) bool {
now := ch.heads()
for repo, head := range cs.Trunk {
if now[repo] != head {
return true
}
}
return false
}
@@ -19,6 +19,7 @@ const (
trunkA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" trunkA = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
trunkA2 = "abababababababababababababababababababab" trunkA2 = "abababababababababababababababababababab"
trunkS = "5555555555555555555555555555555555555555" trunkS = "5555555555555555555555555555555555555555"
trunkS2 = "5656565656565656565656565656565656565656"
) )
// candidateOf is a candidate of a group with the given pull requests, as the indexer writes one. // candidateOf is a candidate of a group with the given pull requests, as the indexer writes one.
@@ -2,10 +2,14 @@ package check
import ( import (
"context" "context"
"crypto/sha256"
"encoding/hex"
"encoding/json" "encoding/json"
"fmt" "fmt"
"os" "os"
"path/filepath" "path/filepath"
"reflect"
"sort"
"strings" "strings"
"sync" "sync"
"testing" "testing"
@@ -565,7 +569,7 @@ func trunksOf(t *testing.T, heads map[string]string) *Trunks {
return tr return tr
} }
func TestNothingIsKeptForAWorkspaceGivenSiblingsBeside(t *testing.T) { // review D1, ADR 0444, issue 1287 func TestNothingIsKeptForAWorkspaceGivenSiblingsWhenTheDigestDoesNotSayItCoversThem(t *testing.T) { // review D1, ADR 0444, issue 1287
b := newBench(t) b := newBench(t)
b.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS, "novox/mesh-catalog": trunkA}) b.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS, "novox/mesh-catalog": trunkA})
b.judge(t, caseOf(t, pullOf(repoA, 7, head1))) b.judge(t, caseOf(t, pullOf(repoA, 7, head1)))
@@ -573,7 +577,7 @@ func TestNothingIsKeptForAWorkspaceGivenSiblingsBeside(t *testing.T) { // review
t.Fatalf("the fetch was not given siblings: %+v", b.steps.tool("fetch")) t.Fatalf("the fetch was not given siblings: %+v", b.steps.tool("fetch"))
} }
if keys, _ := b.kept.Keys(context.Background()); len(keys) != 0 { if keys, _ := b.kept.Keys(context.Background()); len(keys) != 0 {
t.Errorf("kept %v for a package whose tests may read a sibling the digest does not cover", keys) t.Errorf("kept %v for a package whose digest does not say it covers the siblings", keys)
} }
// And a kept verdict of the same digest from elsewhere is not taken for it: everything runs. // And a kept verdict of the same digest from elsewhere is not taken for it: everything runs.
if _, err := (Kept{Store: b.kept}).Keep(context.Background(), "unit", digA, keptValue("fixture/slow", 2)); err != nil { if _, err := (Kept{Store: b.kept}).Keep(context.Background(), "unit", digA, keptValue("fixture/slow", 2)); err != nil {
@@ -585,3 +589,137 @@ func TestNothingIsKeptForAWorkspaceGivenSiblingsBeside(t *testing.T) { // review
t.Errorf("%d test steps, want both levels run whole", n) t.Errorf("%d test steps, want both levels run whole", n)
} }
} }
// besideDigest stands for the Go build module's digest over a workspace with siblings (DigestVersion 3, lead decision L5):
// each package's digest folds the siblings' commits in, and the answer echoes the lines it folded.
func besideDigest(b *bench) func(c stepCall) (any, error) {
return func(c stepCall) (any, error) {
level := c.args["level"].(string)
var lines []string
for _, f := range b.steps.tool("fetch") {
refs, _ := f.args["beside"].([]BesideRef)
for _, r := range refs {
lines = append(lines, r.Repository+" "+r.Commit)
}
}
sort.Strings(lines)
sum := sha256.Sum256([]byte(level + "\n" + strings.Join(lines, "\n")))
return Digested{Level: level, Packages: map[string]string{"fixture/slow": hex.EncodeToString(sum[:])}, Beside: lines}, nil
}
}
func TestACoreRepositorysPassingPackageIsKeptUnderADigestCoveringItsSiblings(t *testing.T) { // 1287 AC1-AC3, AC7 with L5
b := newBench(t)
b.steps.handlers["build_go_digest"] = besideDigest(b)
b.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS, "novox/mesh-catalog": trunkA})
b.judge(t, caseOf(t, pullOf(repoA, 7, head1)))
keys, _ := b.kept.Keys(context.Background())
if len(keys) != 2 {
t.Fatalf("kept %v, want the passing package at both levels", keys)
}
// The same siblings again: everything is taken, nothing runs.
b.steps.calls = nil
v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
if n := len(b.steps.tool("test")); n != 0 || v.Levels[delivery.LevelUnit].Run != 0 || v.Levels[delivery.LevelUnit].Kept == 0 {
t.Fatalf("%d test steps, unit %+v, want none run and the kept verdict taken", n, v.Levels[delivery.LevelUnit])
}
// A sibling's commit moves: a new digest, so the package runs again and is kept under it.
b.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS2, "novox/mesh-catalog": trunkA})
b.steps.calls = nil
b.judge(t, caseOf(t, pullOf(repoA, 7, head1)))
if n := len(b.steps.tool("test")); n != 2 {
t.Errorf("%d test steps after a sibling moved, want both levels run", n)
}
if keys, _ := b.kept.Keys(context.Background()); len(keys) != 4 {
t.Errorf("kept %v, want a second key per level under the new digest", keys)
}
}
func TestADigestThatEchoesOtherSiblingsKeepsNothing(t *testing.T) { // 1287 with L5: an echo that differs is not trusted
b := newBench(t)
inner := besideDigest(b)
b.steps.handlers["build_go_digest"] = func(c stepCall) (any, error) {
v, err := inner(c)
d := v.(Digested)
d.Beside = d.Beside[:len(d.Beside)-1]
return d, err
}
b.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS, "novox/mesh-catalog": trunkA})
b.judge(t, caseOf(t, pullOf(repoA, 7, head1)))
if keys, _ := b.kept.Keys(context.Background()); len(keys) != 0 {
t.Errorf("kept %v under a digest that does not cover every sibling", keys)
}
}
type pkgPaths map[string][]string
func (p pkgPaths) PackagePaths(repository string) []string { return p[repository] }
// groupPackageBench is a group of a TypeScript member (mesh-delivery) and the mesh-sdk member, whose merge changes sdkChanged.
func groupPackageBench(t *testing.T, sdkChanged []string) *bench {
b := newBench(t)
b.ex.Modules = langs{repoA: {buildrun.GoBuildModule, buildrun.NodeBuildModule}, repoSDK: {buildrun.GoBuildModule, buildrun.NodeBuildModule}}
b.ex.Packages = pkgPaths{repoSDK: {"src", "test", "package.json", "package-lock.json", "tsconfig.json"}}
fetch := func(c stepCall) (any, error) {
f := b.fetch
if c.args["repository"] == repoSDK {
f.Changed = sdkChanged
}
return f, nil
}
b.steps.handlers["build_go_fetch"], b.steps.handlers["build_node_fetch"] = fetch, fetch
return b
}
func nodeFetchOf(b *bench, repo string) stepCall {
for _, c := range b.steps.tool("fetch") {
if c.module == buildrun.NodeBuildModule && c.args["repository"] == repo {
return c
}
}
return stepCall{}
}
func TestATypeScriptMemberTakesTheGroupsPackageWhenTheSDKMemberChangesItsPaths(t *testing.T) { // ADR 0500 §5, C11b
b := groupPackageBench(t, []string{"go/delivery/x.go", "src/state.ts"})
b.judge(t, caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2)))
want := map[string]any{"repository": repoSDK, "commit": head2, "ref": "refs/pull/12/head"}
if got := nodeFetchOf(b, repoA).args["package"]; !reflect.DeepEqual(got, want) {
t.Errorf("mesh-delivery's Node fetch was given the package %v, want %v", got, want)
}
if got := nodeFetchOf(b, repoSDK).args["package"]; got != nil {
t.Errorf("the mesh-sdk member's own fetch was given %v: it builds its package itself", got)
}
}
func TestATypeScriptMemberResolvesTheRegistryWhenTheSDKMemberLeavesItsPaths(t *testing.T) { // ADR 0500 §5: the manifest alone, or Go alone
for _, changed := range [][]string{{"go/delivery/x.go"}, {"module.json"}, {"srcx/a.ts"}} {
b := groupPackageBench(t, changed)
b.judge(t, caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2)))
if got := nodeFetchOf(b, repoA).args["package"]; got != nil {
t.Errorf("%v: the Node fetch was given %v, want the registry", changed, got)
}
}
// A lone TypeScript pull request: no group package either.
b := groupPackageBench(t, []string{"src/state.ts"})
b.judge(t, caseOf(t, pullOf(repoA, 7, head1)))
if got := nodeFetchOf(b, repoA).args["package"]; got != nil {
t.Errorf("a lone pull request's Node fetch was given %v", got)
}
}
func TestNoStepOfACheckCarriesPublish(t *testing.T) { // ADR 0500 §4, 1309 (the checker's half): only the builder's trunk build publishes
b := groupPackageBench(t, []string{"src/state.ts"})
b.judge(t, caseOf(t, pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2)))
if len(b.steps.calls) == 0 {
t.Fatal("no step ran")
}
for _, c := range b.steps.calls {
if _, ok := c.args["publish"]; ok {
t.Errorf("%s was given publish: %v", c.tool, c.args)
}
if strings.HasSuffix(c.tool, "_build") || strings.HasSuffix(c.tool, "_pack") {
t.Errorf("the checker asked %s: a check builds no artifact of its own", c.tool)
}
}
}
@@ -0,0 +1,106 @@
package check
import (
"context"
"strings"
"testing"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// What the gate reads of the checker (R4e, lead's interface with dev-r4e-gate): the verdict's heads are the gate's hash over
// `<owner/name lowercase>#<number>=<full head sha>`, a lone pull request is a group of one, and a move of a trunk marks
// mesh/check pending on every open pull request into it, every member's for a group, before any check starts.
func owedState(t testing.TB, r *rig, repo string, number int, head string) string {
t.Helper()
key, err := delivery.CheckerOwedKey(repo, number, head)
if err != nil {
t.Fatal(err)
}
raw, _, found, _ := r.owed.Get(context.Background(), key)
if !found {
return ""
}
o, err := delivery.DecodeCheckerOwed(raw)
if err != nil {
t.Fatal(err)
}
return o.State
}
func TestTheVerdictsHeadsAreTheGatesHashOverLowercaseHeads(t *testing.T) { // gate (b), (c)
ctx := context.Background()
r := newRig(t, passJudge())
upper := strings.ToUpper(head1[:20]) + head1[20:]
r.setTrunks(t, map[string]string{"Novox/Mesh-Delivery": trunkA, repoA: trunkA, repoSDK: trunkS})
cand := candidateOf(t, "feat/x", pullOf("Novox/Mesh-Delivery", 7, upper))
r.ch.Candidate(ctx, "feat%2Fx", cand)
r.ch.Wait()
want, _ := delivery.HeadsHash([]delivery.Head{{Repository: repoA, Number: 7, Head: head1}})
if v := r.verdict(t, keyOf(t, "feat/x", cand)); v.Heads != want {
t.Errorf("heads = %q, want the gate's hash of the lone pull request, a group of one, %q", v.Heads, want)
}
}
func TestAGroupsVerdictHeadsCoverEveryMember(t *testing.T) { // gate (b): hq's member too, as the gate's ask covers it
ctx := context.Background()
r := newRig(t, passJudge())
cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2), pullOf(HQRepository, 3, head2))
r.ch.Candidate(ctx, "feat%2Fx", cand)
r.ch.Wait()
want, _ := delivery.HeadsHash([]delivery.Head{{Repository: repoA, Number: 7, Head: head1}, {Repository: repoSDK, Number: 12, Head: head2},
{Repository: HQRepository, Number: 3, Head: head2}})
if v := r.verdict(t, keyOf(t, "feat/x", cand)); v.Heads != want {
t.Errorf("heads = %q, want %q", v.Heads, want)
}
}
func TestATrunkMoveMarksEveryOpenPullRequestIntoItPendingAtOnce(t *testing.T) { // gate (a), lead decision L4 (1302's checker part)
ctx := context.Background()
r := newRig(t, passJudge())
r.ch.Candidate(ctx, "feat%2Fg", candidateOf(t, "feat/g", pullOf(repoA, 7, head1), pullOf(repoSDK, 12, head2)))
r.ch.Candidate(ctx, "feat%2Fs", candidateOf(t, "feat/s", pullOf(repoSDK, 13, head1)))
r.ch.Wait()
for _, p := range []struct {
repo string
n int
head string
}{{repoA, 7, head1}, {repoSDK, 12, head2}, {repoSDK, 13, head1}} {
if s := owedState(t, r, p.repo, p.n, p.head); s != "success" {
t.Fatalf("%s#%d is %q before the move", p.repo, p.n, s)
}
}
// A delivery builds, so no check starts; mesh-delivery's trunk moves.
r.ch.Hold = func() bool { return true }
r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS})
r.ch.TrunkMoved(ctx)
r.ch.Wait()
if s := owedState(t, r, repoA, 7, head1); s != "pending" {
t.Errorf("mesh-delivery#7, into the moved trunk, is %q, want pending", s)
}
if s := owedState(t, r, repoSDK, 12, head2); s != "pending" {
t.Errorf("mesh-sdk#12, a member of the same group, is %q, want pending", s)
}
if s := owedState(t, r, repoSDK, 13, head1); s != "success" {
t.Errorf("mesh-sdk#13, alone into a trunk that did not move, is %q, want its success to stand", s)
}
}
func TestACheckThatEndsAfterItsTrunkMovedOwesPendingNotItsVerdict(t *testing.T) { // gate (a): a pass on an old trunk is never shown
ctx := context.Background()
var r *rig
j := &judgeFunc{f: func(ctx context.Context, c *Case, call int) (Outcome, error) {
if call == 1 {
r.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS}) // the trunk moves while it runs
r.ch.Hold = func() bool { return true } // and a delivery holds the next check
}
return Outcome{Verdict: passing()}, nil
}}
r = newRig(t, j)
r.ch.Candidate(ctx, "feat%2Fx", candidateOf(t, "feat/x", pullOf(repoA, 7, head1)))
r.ch.Wait()
if s := owedState(t, r, repoA, 7, head1); s != "pending" {
t.Errorf("the status is %q, want pending: the pass was judged on a trunk that moved", s)
}
}
@@ -5,6 +5,7 @@ import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"path" "path"
"slices"
"sort" "sort"
"strings" "strings"
"time" "time"
@@ -68,6 +69,11 @@ type Examiner struct {
Builds BuildStates Builds BuildStates
// Announce says a preview was written under the key (EventPreviewWritten); nil says nothing. // Announce says a preview was written under the key (EventPreviewWritten); nil says nothing.
Announce Announcer Announce Announcer
// Packages names, of a repository, the paths (repository-relative) its TypeScript package artifacts are built from, from
// the indexer's modules (delivery.Module.ArtifactPaths, lead decision L8); nil serves no group package.
Packages interface {
PackagePaths(repository string) []string
}
// Trunks are the trunk heads of every repository, for the siblings a repository's tests read. // Trunks are the trunk heads of every repository, for the siblings a repository's tests read.
Trunks interface{ Heads() map[string]string } Trunks interface{ Heads() map[string]string }
Now func() time.Time Now func() time.Time
@@ -87,6 +93,16 @@ func (j *Examiner) logf(format string, args ...any) {
} }
} }
// besideLines are the siblings as the digest step echoes them: `<owner/repo> <commit>`, sorted.
func besideLines(refs []BesideRef) []string {
out := make([]string, 0, len(refs))
for _, r := range refs {
out = append(out, r.Repository+" "+r.Commit)
}
sort.Strings(out)
return out
}
// target is a pull request on one build module: the node chosen, the workspace its fetch made and what it said. // target is a pull request on one build module: the node chosen, the workspace its fetch made and what it said.
type target struct { type target struct {
pull Pull pull Pull
@@ -98,8 +114,9 @@ type target struct {
fetchArgs map[string]any fetchArgs map[string]any
// priority are the Go packages (import paths) or the Node module directories whose test files the merge changes. // priority are the Go packages (import paths) or the Node module directories whose test files the merge changes.
changed []string changed []string
// beside says the workspace was given siblings: its tests may read what the digest does not cover, so nothing is kept for it. // beside are the lines `<owner/repo> <commit>` of the siblings the workspace was given, sorted: its tests may read them,
beside bool // so a verdict is kept or taken only under a digest that says it folded exactly these in (lead decision L5).
beside []string
plans map[string]*goPlan // by level plans map[string]*goPlan // by level
} }
@@ -142,6 +159,9 @@ func (j *Examiner) Judge(ctx context.Context, c *Case) (Outcome, error) {
return st.finish(ctx, *v), nil return st.finish(ctx, *v), nil
} }
for _, level := range []string{delivery.LevelUnit, delivery.LevelIntegration} { for _, level := range []string{delivery.LevelUnit, delivery.LevelIntegration} {
if c.Try != nil && c.Try.Level != "" && c.Try.Level != level {
continue // a try of one level runs that level alone
}
for _, priority := range []bool{true, false} { for _, priority := range []bool{true, false} {
for _, t := range st.targets { for _, t := range st.targets {
if v := st.runTarget(ctx, t, level, priority); v != nil { if v := st.runTarget(ctx, t, level, priority); v != nil {
@@ -254,7 +274,20 @@ func (st *state) choose(bm string) (node, toolchain string, err error) {
func (st *state) fetchAll(ctx context.Context) *delivery.Verdict { func (st *state) fetchAll(ctx context.Context) *delivery.Verdict {
began := st.j.now() began := st.j.now()
defer func() { st.clone = st.j.now().Sub(began) }() defer func() { st.clone = st.j.now().Sub(began) }()
for _, p := range st.c.Pulls { // The mesh-sdk member first: whether its merge changes the TypeScript package's paths decides whether the other members'
// Node fetches take the group's package (ADR 0500 §5).
pulls := slices.Clone(st.c.Pulls)
slices.SortStableFunc(pulls, func(a, b Pull) int {
if (a.Repository == SDKRepository) != (b.Repository == SDKRepository) {
if a.Repository == SDKRepository {
return -1
}
return 1
}
return 0
})
var groupPackage map[string]any
for _, p := range pulls {
var merged []Fetched var merged []Fetched
for _, bm := range st.j.Modules.BuildModules(p.Repository) { for _, bm := range st.j.Modules.BuildModules(p.Repository) {
node, tc, err := st.choose(bm) node, tc, err := st.choose(bm)
@@ -265,14 +298,17 @@ func (st *state) fetchAll(ctx context.Context) *delivery.Verdict {
st.nodes[node] = true st.nodes[node] = true
args := map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number), "base": p.Trunk} args := map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number), "base": p.Trunk}
t.fetchArgs = args t.fetchArgs = args
if bm == buildrun.NodeBuildModule && groupPackage != nil && p.Repository != SDKRepository {
args["package"] = groupPackage
}
if bm == buildrun.GoBuildModule { if bm == buildrun.GoBuildModule {
beside, err := st.j.besideFor(ctx, st.c, p) beside, err := st.j.besideFor(ctx, st.c, p)
if err != nil { if err != nil {
return fail(node, err.Error()) return fail(forgeMachine(err, node), err.Error()+unheardWords(err))
} }
if len(beside) > 0 { if len(beside) > 0 {
args["beside"] = beside args["beside"] = beside
t.beside = true t.beside = besideLines(beside)
} }
} }
raw, err := st.step(ctx, t, "fetch", fmt.Sprintf("fetch %s#%d on %s", p.Repository, p.Number, node), args) raw, err := st.step(ctx, t, "fetch", fmt.Sprintf("fetch %s#%d on %s", p.Repository, p.Number, node), args)
@@ -284,18 +320,45 @@ func (st *state) fetchAll(ctx context.Context) *delivery.Verdict {
return fail(node, fmt.Sprintf("the fetch of %s#%d answered no workspace: %s", p.Repository, p.Number, raw)) return fail(node, fmt.Sprintf("the fetch of %s#%d answered no workspace: %s", p.Repository, p.Number, raw))
} }
if len(t.fetched.Conflicts) > 0 { if len(t.fetched.Conflicts) > 0 {
st.j.writePreview(ctx, p, append(merged, t.fetched)) if st.c.Try == nil {
st.j.writePreview(ctx, p, append(merged, t.fetched))
}
return fail(node, fmt.Sprintf("%s#%d does not merge into the trunk: files in conflict: %s", p.Repository, p.Number, strings.Join(t.fetched.Conflicts, ", "))) return fail(node, fmt.Sprintf("%s#%d does not merge into the trunk: files in conflict: %s", p.Repository, p.Number, strings.Join(t.fetched.Conflicts, ", ")))
} }
t.changed = t.fetched.Changed t.changed = t.fetched.Changed
merged = append(merged, t.fetched) merged = append(merged, t.fetched)
st.targets = append(st.targets, t) st.targets = append(st.targets, t)
} }
st.j.writePreview(ctx, p, merged) if st.c.Try == nil { // a try owes the indexer no preview: no pull request holds its head
st.j.writePreview(ctx, p, merged)
}
if p.Repository == SDKRepository && len(st.c.Pulls) > 1 && st.j.changesPackage(p, merged) {
groupPackage = map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number)}
}
} }
return nil return nil
} }
// changesPackage says whether the merge of a pull request changes a file inside the named paths of one of its repository's
// TypeScript packages: a change of the manifest alone, or of files outside those paths, does not (ADR 0500 §5).
func (j *Examiner) changesPackage(p Pull, merged []Fetched) bool {
if j.Packages == nil {
return false
}
paths := j.Packages.PackagePaths(p.Repository)
for _, f := range merged {
for _, changed := range f.Changed {
for _, named := range paths {
named = strings.TrimSuffix(strings.TrimPrefix(named, "./"), "/")
if changed == named || strings.HasPrefix(changed, named+"/") {
return true
}
}
}
}
return false
}
// isTestFile says whether a repository-relative path is a test file of the languages the checker judges: Go's _test.go and // isTestFile says whether a repository-relative path is a test file of the languages the checker judges: Go's _test.go and
// the TypeScript module's .test. and .spec. files and test directories. // the TypeScript module's .test. and .spec. files and test directories.
func isTestFile(p string) bool { func isTestFile(p string) bool {
@@ -398,9 +461,8 @@ func (st *state) planGo(ctx context.Context, t *target, level string) *delivery.
} }
plan := &goPlan{digests: map[string]string{}} plan := &goPlan{digests: map[string]string{}}
t.plans[level] = plan t.plans[level] = plan
if t.beside { if st.c.Try != nil && st.c.Try.Run != "" {
// The digest covers the package's own files, the toolchain and the header, not the siblings placed beside the tree: a // A try of some tests runs them alone: a partial run proves no package, so nothing is taken or kept.
// package whose tests read one must never be kept, nor taken from a verdict kept for other inputs (ADR 0444).
plan.all = true plan.all = true
return nil return nil
} }
@@ -421,6 +483,13 @@ func (st *state) planGo(ctx context.Context, t *target, level string) *delivery.
plan.all = true plan.all = true
return nil return nil
} }
if !slices.Equal(d.Beside, t.beside) {
// The siblings the tests read are not all in the digest (a build module before DigestVersion 3, or another set): a
// package whose tests read one is never kept, nor taken from a verdict kept for other inputs (ADR 0444, L5).
st.j.logf("%s#%d: the %s level's digest covers the siblings %v, not %v, so it runs whole and keeps nothing", t.pull.Repository, t.pull.Number, level, d.Beside, t.beside)
plan.all = true
return nil
}
plan.digests = d.Packages plan.digests = d.Packages
a := st.levels[level] a := st.levels[level]
pkgs := make([]string, 0, len(d.Packages)) pkgs := make([]string, 0, len(d.Packages))
@@ -467,6 +536,9 @@ func (st *state) runGo(ctx context.Context, t *target, level string, priority bo
if run != nil { if run != nil {
args["packages"] = run args["packages"] = run
} }
if st.c.Try != nil && st.c.Try.Run != "" {
args["run"] = st.c.Try.Run // build_go_test's go test -run pattern
}
raw, err := st.step(ctx, t, "test", fmt.Sprintf("test the %s level of %s#%d on %s", level, t.pull.Repository, t.pull.Number, t.node), args) raw, err := st.step(ctx, t, "test", fmt.Sprintf("test the %s level of %s#%d on %s", level, t.pull.Repository, t.pull.Number, t.node), args)
if err != nil { if err != nil {
return fail(t.node, fmt.Sprintf("the %s level of %s#%d: %v", level, t.pull.Repository, t.pull.Number, err)) return fail(t.node, fmt.Sprintf("the %s level of %s#%d: %v", level, t.pull.Repository, t.pull.Number, err))
@@ -2,6 +2,7 @@ package check
import ( import (
"fmt" "fmt"
"slices"
"strings" "strings"
"unicode/utf8" "unicode/utf8"
@@ -50,7 +51,7 @@ func firstFailed(v delivery.Verdict) string {
// pieces are the parts of a line, from the most to the least needed. // pieces are the parts of a line, from the most to the least needed.
type pieces struct { type pieces struct {
head, commits, counts, shortCounts, times, shortTimes, over, builds, composition, group string head, commits, counts, shortCounts, times, shortTimes, over, builds, composition, shortComposition, group string
} }
func (p pieces) join(drop int) string { func (p pieces) join(drop int) string {
@@ -70,7 +71,10 @@ func (p pieces) join(drop int) string {
if drop < 2 && p.group != "" { if drop < 2 && p.group != "" {
parts = append(parts, p.group) parts = append(parts, p.group)
} }
if p.composition != "" { switch {
case drop >= 3 && p.shortComposition != "":
parts = append(parts, p.shortComposition)
case p.composition != "":
parts = append(parts, p.composition) parts = append(parts, p.composition)
} }
out := make([]string, 0, len(parts)) out := make([]string, 0, len(parts))
@@ -90,14 +94,37 @@ func levelWords(name string, r delivery.LevelResult) string {
return s return s
} }
func compositionWords(c *delivery.Composition) string { // compositionWords says the composition: not yet run, or once the assigner's check reports it (issue 1297) passed or
switch { // failed, with the time it took and the machines it composed (issue 1300).
case c == nil, c.State == delivery.CompositionNotRun: func compositionWords(c *delivery.Composition, took int64) string { return composition(c, took, false) }
// composition is compositionWords; short counts the machines and leaves the SDK commit to the report.
func composition(c *delivery.Composition, took int64, short bool) string {
if c == nil || c.State == delivery.CompositionNotRun {
return "composition not yet run" return "composition not yet run"
case c.State == delivery.CompositionPass:
return "composition passed with " + short12(c.SDK)
} }
return "composition failed: " + c.Words in := ""
if took > 0 {
in = " in " + seconds(took)
}
on := ""
if len(c.Machines) > 0 {
machines := slices.Sorted(slices.Values(c.Machines))
on = " on " + strings.Join(machines, ", ")
switch {
case len(machines) > 1 && short:
on = fmt.Sprintf(" on %d machines", len(machines))
case len(machines) > 1:
on = fmt.Sprintf(" on %d machines: %s", len(machines), strings.Join(machines, ", "))
}
}
switch {
case c.State == delivery.CompositionPass && short:
return "composition passed" + in + on // the SDK it composed with is in the report
case c.State == delivery.CompositionPass:
return "composition passed" + in + " with " + short12(c.SDK) + on
}
return "composition failed" + in + on + ": " + c.Words
} }
func buildWords(builds []BuildNote) string { func buildWords(builds []BuildNote) string {
@@ -149,7 +176,7 @@ func parts(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote, slowest []s
for _, x := range []struct { for _, x := range []struct {
name string name string
ms int64 ms int64
}{{"queue", t.Queue}, {"clone", t.Clone}, {"unit", t.Unit}, {"int setup", t.IntegrationSetup}, {"int run", t.IntegrationRun}, {"composition", t.Composition}} { }{{"queue", t.Queue}, {"clone", t.Clone}, {"unit", t.Unit}, {"int setup", t.IntegrationSetup}, {"int run", t.IntegrationRun}} {
if x.ms > 0 { if x.ms > 0 {
times = append(times, x.name+" "+seconds(x.ms)) times = append(times, x.name+" "+seconds(x.ms))
} }
@@ -181,7 +208,8 @@ func parts(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote, slowest []s
} }
pc.builds = buildWords(builds) pc.builds = buildWords(builds)
if v.State != delivery.VerdictRunning { if v.State != delivery.VerdictRunning {
pc.composition = compositionWords(v.Composition) pc.composition = compositionWords(v.Composition, v.Took.Composition)
pc.shortComposition = composition(v.Composition, v.Took.Composition, true)
} }
if cs != nil && len(cs.Pulls) > 1 { if cs != nil && len(cs.Pulls) > 1 {
pc.group = fmt.Sprintf("group %s of %d pull requests", cs.Group, len(cs.Pulls)) pc.group = fmt.Sprintf("group %s of %d pull requests", cs.Group, len(cs.Pulls))
@@ -133,3 +133,24 @@ func TestThePassedCompositionIsSaidOnlyWhenItWasRun(t *testing.T) { // ADR 0484
t.Errorf("a verdict with no composition: %q", r) t.Errorf("a verdict with no composition: %q", r)
} }
} }
func TestTheCompositionsTimeAndMachinesAreSaidOnceTheAssignerReportsThem(t *testing.T) { // 1300 (C10), the assigner's verdict (issue 1297)
cs, p := lineCase(t, pullOf(repoA, 7, head1))
v := passVerdict()
v.Took.Composition = 2400
v.Composition = &delivery.Composition{State: delivery.CompositionPass, SDK: "afbe371003a570f088be4a394d83aa7a9458f7ab", Machines: []string{"g14", "ace", "novox", "shanks"}}
r := Report(v, cs, p, nil, nil)
for _, want := range []string{"composition passed in 2.4s with afbe371003a5 on 4 machines: ace, g14, novox, shanks"} {
if !strings.Contains(r, want) {
t.Errorf("report %q lacks %q", r, want)
}
}
if l := Line(v, cs, p, nil, nil); !strings.Contains(l, "composition passed in 2.4s") || !strings.Contains(l, "on 4 machines") || utf8.RuneCountInString(l) > MaxLine {
t.Errorf("line = %q", l)
}
v.State, v.Failed = delivery.VerdictFail, delivery.FailedFirst{Machine: "novox", Words: "composition refused"}
v.Composition = &delivery.Composition{State: delivery.CompositionFail, Words: "novox: two seats claim dns", Machines: []string{"novox"}}
if r := Report(v, cs, p, nil, nil); !strings.Contains(r, "composition failed in 2.4s on novox: novox: two seats claim dns") {
t.Errorf("failed composition report = %q", r)
}
}
@@ -77,7 +77,7 @@ func (st *state) checkPins(ctx context.Context) verdictPtr {
text, err = st.j.Forge.File(ctx, t.pull.Repository, file, t.pull.Trunk) text, err = st.j.Forge.File(ctx, t.pull.Repository, file, t.pull.Trunk)
} }
if err != nil { if err != nil {
return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: %s cannot be read to check its mesh-sdk pin: %v", t.pull.Repository, t.pull.Number, file, err)) return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: %s cannot be read to check its mesh-sdk pin: %v%s", t.pull.Repository, t.pull.Number, file, err, unheardWords(err)))
} }
pin, ok := SDKPin(text) pin, ok := SDKPin(text)
if !ok { if !ok {
@@ -88,7 +88,7 @@ func (st *state) checkPins(ctx context.Context) verdictPtr {
} }
onTrunk, err := st.j.Forge.OnTrunk(ctx, SDKRepository, pin) onTrunk, err := st.j.Forge.OnTrunk(ctx, SDKRepository, pin)
if err != nil { if err != nil {
return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: whether mesh-sdk %s is on its trunk cannot be told: %v", t.pull.Repository, t.pull.Number, pin, err)) return failed(forgeMachine(err, t.node), fmt.Sprintf("%s#%d: whether mesh-sdk %s is on its trunk cannot be told: %v%s", t.pull.Repository, t.pull.Number, pin, err, unheardWords(err)))
} }
if onTrunk { if onTrunk {
continue continue
@@ -115,6 +115,14 @@ func pinHint(c *Case, fetchError string) string {
return " (a pin to the head of the group's mesh-sdk pull request names a commit only that head carries, which go mod download cannot resolve; " + Issue1148 + ")" return " (a pin to the head of the group's mesh-sdk pull request names a commit only that head carries, which go mod download cannot resolve; " + Issue1148 + ")"
} }
// unheardWords ends the words of a failed read of the forge that reached nobody: the forge was not heard (UnheardWords).
func unheardWords(err error) string {
if isUnheard(err) {
return " (the forge was " + UnheardWords + ")"
}
return ""
}
// forgeMachine is the machine a failed read of the forge is put at: the forge when nobody answered it (no build node is to // forgeMachine is the machine a failed read of the forge is put at: the forge when nobody answered it (no build node is to
// blame, issue 1820), the build node whose workspace was read otherwise. // blame, issue 1820), the build node whose workspace was read otherwise.
func forgeMachine(err error, node string) string { func forgeMachine(err error, node string) string {
@@ -253,3 +253,15 @@ func TestThePinOfTheControllerThatCannotBeReadFailsTheCheck(t *testing.T) { // r
t.Errorf("verdict = %+v", v) t.Errorf("verdict = %+v", v)
} }
} }
func TestAForgeThatIsNotHeardFailsAtTheForgeAndIsJudgedAgainWhenTheNodesChange(t *testing.T) { // 1624 C7
b := newBench(t)
b.ex.Forge = &fakeForge{err: &Unheard{Err: fmt.Errorf("nats: no responders available for request")}}
v := b.judge(t, caseOf(t, pullOf(repoA, 7, head1))).Verdict
if v.State != delivery.VerdictFail || v.Failed.Machine != "forge" {
t.Fatalf("verdict = %+v, want failed at the forge", v)
}
if !IsNodesFailure(v) {
t.Errorf("a forge that was not heard is not the change's failure: %+v is not judged again when the nodes change", v.Failed)
}
}
@@ -7,6 +7,7 @@ import (
"strings" "strings"
"git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
) )
// The read tools (novox/hq ADR 0484 clauses 8 and 13, issue 1686): what the checker stored, as stored, with the line of the // The read tools (novox/hq ADR 0484 clauses 8 and 13, issue 1686): what the checker stored, as stored, with the line of the
@@ -17,6 +18,9 @@ const (
VerdictsReadTool = "checker_verdicts" VerdictsReadTool = "checker_verdicts"
PullReadTool = "checker_pull" PullReadTool = "checker_pull"
KeptReadTool = "checker_kept" KeptReadTool = "checker_kept"
// PreviewsReadTool reads the previews state (issue 1680); StatusReadTool reads what the checker is held by and what it runs (issue 1685).
PreviewsReadTool = "checker_previews"
StatusReadTool = "checker_status"
) )
// StateRead reads one state: its keys and a key's stored value (nil for a key with none). // StateRead reads one state: its keys and a key's stored value (nil for a key with none).
@@ -26,7 +30,14 @@ type StateRead struct {
} }
// Reader answers the read tools from the checker's verdicts and kept states and the indexer's candidates. // Reader answers the read tools from the checker's verdicts and kept states and the indexer's candidates.
type Reader struct{ VerdictState, KeptState, CandidateState StateRead } type Reader struct {
VerdictState, KeptState, CandidateState StateRead
PreviewState StateRead // the checker's merge previews
PlansState StateRead // the controller's delivery plans in their build phase; zero where the bucket is absent
StepStates map[string]StateRead // the build modules' steps states, by module
Capacity func() (string, error) // the capacity setting as written, `node=steps,...`
}
// VerdictRead is a verdict as stored with the full report of the status of each pull request it judged (Report), when the // VerdictRead is a verdict as stored with the full report of the status of each pull request it judged (Report), when the
// candidate still holds those heads; without it the report is that of the verdict alone. // candidate still holds those heads; without it the report is that of the verdict alone.
@@ -57,7 +68,7 @@ func (r Reader) read(key string, raw json.RawMessage) (VerdictRead, error) {
return out, fmt.Errorf("%s holds what is not a verdict: %w", key, err) return out, fmt.Errorf("%s holds what is not a verdict: %w", key, err)
} }
c, ok := r.candidate(v.Group) c, ok := r.candidate(v.Group)
if !ok || c.HeadsHash != v.Heads { if !ok || (c.HeadsHash != v.Heads && GateHeads(c) != v.Heads) {
out.Note = "the candidate no longer holds these heads" out.Note = "the candidate no longer holds these heads"
return out, nil return out, nil
} }
@@ -191,3 +202,153 @@ func (r Reader) Kept(level, digest string) (map[string]json.RawMessage, error) {
} }
return out, nil return out, nil
} }
// PreviewRead is one preview as stored, with the parts of its key.
type PreviewRead struct {
Key string `json:"key"`
Repository string `json:"repository"`
Head string `json:"head"`
Trunk string `json:"trunk"`
Preview json.RawMessage `json:"preview"`
}
// Previews are the previews the checker holds, each as stored. With a repository only that repository's; with a head (a
// prefix of at least 7 characters is enough) only previews of that head. A key that is no preview key is listed as unreadable.
func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, error) {
if r.PreviewState.Keys == nil || r.PreviewState.Get == nil {
return nil, nil, fmt.Errorf("the previews state cannot be read here")
}
keys, err := r.PreviewState.Keys()
if err != nil {
return nil, nil, err
}
sort.Strings(keys)
out := []PreviewRead{}
var bad []string
for _, k := range keys {
repo, h, trunk, err := delivery.ParsePreviewKey(k)
if err != nil {
bad = append(bad, k)
continue
}
if (repository != "" && repo != repository) || (head != "" && !strings.HasPrefix(h, head)) {
continue
}
raw, err := r.PreviewState.Get(k)
if err != nil {
return nil, nil, err
}
if len(raw) > 0 {
out = append(out, PreviewRead{Key: k, Repository: repo, Head: h, Trunk: trunk, Preview: raw})
}
}
return out, bad, nil
}
// RunningStep is a step in flight on a build module.
type RunningStep struct {
Module string `json:"module"`
Node string `json:"node"`
ID string `json:"id"`
Tool string `json:"tool"`
Since string `json:"since,omitempty"`
}
// Status is what holds the checker back and what runs now, read from the buckets: the checker's held and running steps live in
// its process, so they are shown as what the buckets say (the steps in flight on the build modules, the candidates without a
// verdict for their heads, and the delivery plans in their build phase), not as the process's own counters.
type Status struct {
Capacity map[string]int `json:"capacity"`
CapacityNote string `json:"capacity_note,omitempty"`
DefaultPerNode int `json:"default_per_node"`
PlansBuilding struct {
Readable bool `json:"readable"`
Keys []string `json:"keys"`
Holds bool `json:"holds"`
} `json:"plans_building"`
Running []RunningStep `json:"running_steps"`
RunningByNode map[string]int `json:"running_by_node"`
Unjudged []string `json:"unjudged_candidates"`
Notes []string `json:"notes,omitempty"`
}
// Status reads the capacity, the plan bucket, the build modules' steps and the candidates against their verdicts.
func (r Reader) Status() (Status, error) {
st := Status{DefaultPerNode: buildrun.DefaultCapacity, Capacity: map[string]int{}, RunningByNode: map[string]int{}}
if r.Capacity == nil {
st.CapacityNote = "the capacity setting cannot be read from here"
} else if setting, err := r.Capacity(); err != nil {
st.CapacityNote = "the capacity setting cannot be read: " + err.Error()
} else if st.Capacity, err = buildrun.ParseCapacity(setting); err != nil {
st.Capacity, st.CapacityNote = map[string]int{}, "the capacity setting is not valid: "+err.Error()
}
if r.PlansState.Keys == nil {
st.Notes = append(st.Notes, PlansBuildingBucket+" is not readable here (absent), so it is read as empty: no delivery holds the checker")
} else if keys, err := r.PlansState.Keys(); err != nil {
st.Notes = append(st.Notes, PlansBuildingBucket+" cannot be listed, so it is read as empty: "+err.Error())
} else {
sort.Strings(keys)
st.PlansBuilding.Readable, st.PlansBuilding.Keys, st.PlansBuilding.Holds = true, keys, len(keys) > 0
}
mods := make([]string, 0, len(r.StepStates))
for m := range r.StepStates {
mods = append(mods, m)
}
sort.Strings(mods)
st.Running = []RunningStep{}
for _, m := range mods {
sr := r.StepStates[m]
if sr.Keys == nil || sr.Get == nil {
continue
}
keys, err := sr.Keys()
if err != nil {
st.Notes = append(st.Notes, "the steps of "+m+" cannot be listed: "+err.Error())
continue
}
sort.Strings(keys)
for _, k := range keys {
raw, err := sr.Get(k)
if err != nil || len(raw) == 0 {
continue
}
step, err := delivery.DecodeStep(raw)
if err != nil || step.State != delivery.StepRunning {
continue
}
rs := RunningStep{Module: m, Node: step.Node, ID: step.ID, Tool: step.Tool}
if !step.Started.IsZero() {
rs.Since = step.Started.UTC().Format("2006-01-02T15:04:05Z")
}
st.Running = append(st.Running, rs)
st.RunningByNode[step.Node]++
}
}
st.Unjudged = []string{}
if r.CandidateState.Keys != nil && r.CandidateState.Get != nil && r.VerdictState.Get != nil {
keys, err := r.CandidateState.Keys()
if err != nil {
return st, err
}
sort.Strings(keys)
for _, k := range keys {
raw, err := r.CandidateState.Get(k)
if err != nil || len(raw) == 0 {
continue
}
c, err := delivery.DecodeCandidate(raw)
if err != nil {
continue
}
vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
if err != nil {
continue
}
if v, err := r.VerdictState.Get(vkey); err == nil && len(v) == 0 {
st.Unjudged = append(st.Unjudged, c.Group)
}
}
}
st.Notes = append(st.Notes, "the steps the checker holds waiting for a place are in its process and in no bucket: a candidate without a verdict while a plan key exists, or while its node runs as many steps as its capacity, is held")
return st, nil
}
@@ -3,8 +3,10 @@ package check
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"os"
"strings" "strings"
"testing" "testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery"
) )
@@ -102,3 +104,64 @@ func TestTheKeptToolAnswersTheKeysAsStoredByLevelAndByDigest(t *testing.T) { //
t.Error("a digest nothing kept was answered") t.Error("a digest nothing kept was answered")
} }
} }
// 1680 live acceptance: checker_previews shows the preview key of an open pull request, as stored, narrowed by repository and head.
func TestThePreviewsToolShowsTheStoredPreviewOfAPullRequest(t *testing.T) {
pv := delivery.Preview{Repository: repoA, Head: head1, Trunk: head2, Merge: "m1", Modules: map[string]delivery.PreviewModule{}}
data, err := delivery.EncodePreview(pv)
if err != nil {
t.Fatal(err)
}
key, _ := delivery.PreviewKey(repoA, head1, head2)
rd := Reader{PreviewState: fakeState{key: data, "junk": []byte("{}")}.read()}
got, bad, err := rd.Previews(repoA, head1[:8])
if err != nil || len(got) != 1 || got[0].Key != key || got[0].Head != head1 || got[0].Trunk != head2 || string(got[0].Preview) != string(data) {
t.Fatalf("previews = %+v, %v", got, err)
}
if len(bad) != 1 || bad[0] != "junk" {
t.Errorf("unreadable = %v", bad)
}
if got, _, _ := rd.Previews("novox/other", ""); len(got) != 0 {
t.Errorf("another repository's = %+v", got)
}
if _, _, err := (Reader{}).Previews("", ""); err == nil {
t.Error("answered with no state to read")
}
}
// 1685 live acceptance: checker_status reads the capacity, whether a plan key holds the checker, the steps in flight by node and
// the candidates without a verdict.
func TestTheStatusToolReadsCapacityHoldsAndRunningSteps(t *testing.T) {
running, _ := delivery.EncodeStep(delivery.Step{ID: "abc1", Tool: "build_go_test", Node: "shanks", State: delivery.StepRunning, Started: time.Date(2026, 10, 22, 15, 0, 0, 0, time.UTC)})
passed, _ := delivery.EncodeStep(delivery.Step{ID: "abc2", Tool: "build_go_test", Node: "ace", State: delivery.StepPassed})
cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1))
ckey, _ := delivery.CandidatesKey("feat/x")
rd := Reader{
Capacity: func() (string, error) { return "shanks=2,ace=0", nil },
PlansState: fakeState{"p1": []byte("{}")}.read(),
StepStates: map[string]StateRead{"mesh-build-go": fakeState{"shanks.abc1": running, "ace.abc2": passed}.read()},
CandidateState: fakeState{ckey: cand}.read(),
VerdictState: fakeState{}.read(),
}
st, err := rd.Status()
if err != nil {
t.Fatal(err)
}
if st.Capacity["shanks"] != 2 || st.Capacity["ace"] != 0 || st.DefaultPerNode != 1 {
t.Errorf("capacity = %v default %d", st.Capacity, st.DefaultPerNode)
}
if !st.PlansBuilding.Holds || len(st.PlansBuilding.Keys) != 1 {
t.Errorf("plans = %+v", st.PlansBuilding)
}
if len(st.Running) != 1 || st.Running[0].Node != "shanks" || st.RunningByNode["shanks"] != 1 || st.RunningByNode["ace"] != 0 {
t.Errorf("running = %+v %v", st.Running, st.RunningByNode)
}
if len(st.Unjudged) != 1 || st.Unjudged[0] != "feat/x" {
t.Errorf("unjudged = %v", st.Unjudged)
}
// An absent plan bucket is read as empty and said; an unreadable capacity is said, not guessed.
st, err = Reader{Capacity: func() (string, error) { return "", os.ErrNotExist }}.Status()
if err != nil || st.PlansBuilding.Holds || st.CapacityNote == "" || len(st.Notes) == 0 {
t.Errorf("empty status = %+v, %v", st, err)
}
}
@@ -92,6 +92,9 @@ type Tested struct {
// Digested is the answer of the Go build module's digest step: one digest per package that has tests at the level, by import // Digested is the answer of the Go build module's digest step: one digest per package that has tests at the level, by import
// path. // path.
type Digested struct { type Digested struct {
// Beside are the lines `<owner/repo> <commit>` of the siblings the digest folded in (DigestVersion 3, lead decision L5),
// sorted; absent when the workspace has none or the build module folds none in.
Beside []string `json:"beside,omitempty"`
Level string `json:"level"` Level string `json:"level"`
Packages map[string]string `json:"packages"` Packages map[string]string `json:"packages"`
WithoutTests []string `json:"without_tests,omitempty"` WithoutTests []string `json:"without_tests,omitempty"`
@@ -0,0 +1,389 @@
package check
import (
"context"
"encoding/json"
"errors"
"fmt"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// A try is a pushed head judged by the checker as a pull request's head is, before its pull request is opened (ADR 0501
// §5; issue 686's `mesh-controller.try`, which calls these tools from R4e): the head merged onto its trunk, the unit and the
// integration level through the build modules. It owes the forge nothing (no status, no preview) and writes no verdict; its
// state is the value of its key in the checker's state tries.
// The try's tools and state.
const (
TryTool = "checker_try"
TryReadTool = "checker_try_read"
// TriesState is the checker's state of tries: the bucket mesh-delivery-checker_tries, keys try-<unix nanoseconds>.
TriesState = "tries"
TriesBucket = delivery.CheckerModule + "_" + TriesState
// MaxTries is how many tries may wait or run at once, mesh-wide: each holds a build node for minutes.
MaxTries = 4
// TryKeep is how long an ended try's key is kept before the checker removes it.
TryKeep = 7 * 24 * time.Hour
)
// The states of a try: asked and waiting, running, and its end.
const (
TryQueued = "queued"
TryRunning = "running"
TryPass = "pass"
TryFail = "fail"
TryError = "error"
)
var (
tryKeyOK = regexp.MustCompile(`^try-[0-9]{1,20}$`)
tryHeadOK = regexp.MustCompile(`^([0-9a-f]{40}|[0-9a-f]{64})$`)
tryTestsOK = regexp.MustCompile(`^[A-Za-z0-9_^$|().*+?/\[\]-]+$`)
tryRepoOK = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]*/[A-Za-z0-9][A-Za-z0-9._-]*$`)
tryTrunkOK = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._/-]*$`)
)
// Try is the value of a tries key.
type Try struct {
V int `json:"v"`
ID string `json:"id"`
Repository string `json:"repository"`
// Head is the commit tried; Trunk the trunk commit it is merged onto, and Branch the trunk's branch when it was asked by name.
Head string `json:"head"`
Trunk string `json:"trunk"`
Branch string `json:"branch,omitempty"`
// Tests is a go test -run pattern: only those tests run, and nothing is kept. Level is unit or integration: only that
// level runs. Empty is every test of both levels.
Tests string `json:"tests,omitempty"`
Level string `json:"level,omitempty"`
State string `json:"state"`
// 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"`
Failed []string `json:"failed,omitempty"`
Levels map[string]delivery.LevelResult `json:"levels,omitempty"`
Machines []string `json:"machines,omitempty"`
Asked time.Time `json:"asked"`
Started time.Time `json:"started,omitzero"`
Ended time.Time `json:"ended,omitzero"`
}
// Over says whether the try came to its end.
func (t Try) Over() bool { return t.State == TryPass || t.State == TryFail || t.State == TryError }
// DecodeTry reads a tries value.
func DecodeTry(raw []byte) (Try, error) {
var t Try
if err := json.Unmarshal(raw, &t); err != nil {
return Try{}, fmt.Errorf("not a try: %w", err)
}
if !tryKeyOK.MatchString(t.ID) || t.Repository == "" || t.Head == "" || t.Trunk == "" {
return Try{}, errors.New("not a try: it names no id, repository, head or trunk")
}
return t, nil
}
func encodeTry(t Try) ([]byte, error) {
t.V = delivery.Version
return json.Marshal(t)
}
// TryAsk is what checker_try is given.
type TryAsk struct {
Repository, Head, Trunk, Tests, Level string
}
// TryStore is the tries state as the tools see it.
type TryStore interface {
Keys() ([]string, error)
Get(key string) (json.RawMessage, error)
Create(key string, value []byte) error
}
// NotChecked is the error of a try of a repository the checker judges no level of: hq, judged by merge-check.sh (ADR 0484
// clause 8), and a repository whose trunk the gate's source does not name.
func NotChecked(repository string) error {
return fmt.Errorf("no level of %s is checked by the checker", repository)
}
// AskTry writes a new try, queued, and answers its id: a try of the same head that waits or runs is the answer to a second
// ask, and at most MaxTries wait or run at once. trunks are the gate's trunk heads of every repository (source.main): a
// repository with none, and hq, are no repository the checker checks. A trunk asked by name is its trunk head now.
func AskTry(store TryStore, trunks map[string]string, a TryAsk, now time.Time) (string, error) {
repo := strings.ToLower(strings.TrimSuffix(strings.TrimSpace(a.Repository), ".git"))
head := strings.ToLower(strings.TrimSpace(a.Head))
switch {
case !tryRepoOK.MatchString(repo) || strings.Contains(repo, ".."):
return "", fmt.Errorf("%q is not a repository, <owner>/<name>", a.Repository)
case !tryHeadOK.MatchString(head):
return "", fmt.Errorf("the head %q is not a full commit, 40 or 64 hex digits", a.Head)
case a.Level != "" && a.Level != delivery.LevelUnit && a.Level != delivery.LevelIntegration:
return "", fmt.Errorf("the level %q is neither unit nor integration", a.Level)
case a.Tests != "" && !tryTestsOK.MatchString(a.Tests):
return "", fmt.Errorf("the tests %q are not a go test -run pattern", a.Tests)
}
trunkHead := ""
for r, h := range trunks {
if strings.EqualFold(r, repo) {
trunkHead = h
}
}
if repo == HQRepository || trunkHead == "" {
return "", NotChecked(repo)
}
t := Try{Repository: repo, Head: head, Tests: a.Tests, Level: a.Level, State: TryQueued, Asked: now.UTC()}
switch trunk := strings.TrimSpace(a.Trunk); {
case trunk == "":
t.Trunk = trunkHead
case tryHeadOK.MatchString(strings.ToLower(trunk)):
t.Trunk = strings.ToLower(trunk)
case tryTrunkOK.MatchString(trunk) && !strings.Contains(trunk, ".."):
// The checker judges pull requests into the trunk alone (ADR 0484 clause 11): a branch named is the trunk's head now.
t.Trunk, t.Branch = trunkHead, trunk
default:
return "", fmt.Errorf("the trunk %q is neither a full commit nor a branch", a.Trunk)
}
keys, err := store.Keys()
if err != nil {
return "", fmt.Errorf("the tries cannot be listed: %w", err)
}
sort.Strings(keys)
var open []string
for _, k := range keys {
raw, err := store.Get(k)
if err != nil || len(raw) == 0 {
continue
}
cur, err := DecodeTry(raw)
if err != nil || cur.Over() {
continue
}
if cur.Repository == t.Repository && cur.Head == t.Head && cur.Trunk == t.Trunk && cur.Tests == t.Tests && cur.Level == t.Level {
return cur.ID, nil
}
open = append(open, fmt.Sprintf("%s (%s@%.12s, %s)", cur.ID, cur.Repository, cur.Head, cur.State))
}
if len(open) >= MaxTries {
return "", fmt.Errorf("%d tries wait or run, the most at once: %s; ask again when one ends", len(open), strings.Join(open, ", "))
}
t.ID = "try-" + strconv.FormatInt(now.UnixNano(), 10)
data, err := encodeTry(t)
if err != nil {
return "", err
}
if err := store.Create(t.ID, data); err != nil {
return "", fmt.Errorf("the try %s cannot be written: %w", t.ID, err)
}
return t.ID, nil
}
// ReadTry is a try's value, as stored.
func ReadTry(store TryStore, id string) (json.RawMessage, error) {
if !tryKeyOK.MatchString(id) {
return nil, fmt.Errorf("%q is not a try's id, try-<digits>", id)
}
raw, err := store.Get(id)
if err != nil {
return nil, err
}
if len(raw) == 0 {
return nil, fmt.Errorf("no try %s: it was never asked, or it ended more than %s ago and was removed", id, TryKeep)
}
return raw, nil
}
// TryOptions narrow what a try runs; a pull request's check has none.
type TryOptions struct {
Level string
Run string
}
// Tryer runs the tries: it takes a queued try, judges it through the same Judge as a pull request's check, and writes its
// end. It holds no timer: a try starts when its key is written, or when a delivery that held it ends (Reconsider).
type Tryer struct {
Tries KV
Judge Judge
// Hold says whether a delivery builds, so no try starts (ADR 0484 clause 12); nil never holds.
Hold func() bool
Bound time.Duration
Now func() time.Time
Logf func(format string, args ...any)
mu sync.Mutex
wg sync.WaitGroup
inflight map[string]bool
seen map[string]bool // looked at since this process started: a try left running is taken over the first time
}
func (tr *Tryer) now() time.Time {
if tr.Now != nil {
return tr.Now()
}
return time.Now()
}
func (tr *Tryer) logf(format string, args ...any) {
if tr.Logf != nil {
tr.Logf(format, args...)
}
}
// Wait returns when every try started so far has ended.
func (tr *Tryer) Wait() { tr.wg.Wait() }
// Put takes a value of the tries state in: a queued try is started, and a try left running is taken over the first time
// this process sees it (the checker that ran it stopped; there is one checker).
func (tr *Tryer) Put(ctx context.Context, key string, value []byte) {
t, err := DecodeTry(value)
if err != nil {
tr.logf("tries.%s holds what is not a try, left: %v", key, err)
return
}
if t.Over() {
return
}
held := tr.Hold != nil && tr.Hold()
tr.mu.Lock()
if tr.inflight == nil {
tr.inflight, tr.seen = map[string]bool{}, map[string]bool{}
}
if tr.inflight[key] || (t.State == TryRunning && tr.seen[key]) {
tr.mu.Unlock()
return
}
if held {
tr.mu.Unlock() // not seen yet: a try left running is still taken over when the hold ends
tr.logf("%s: not started, a delivery builds", key)
return
}
tr.seen[key] = true
tr.inflight[key] = true
tr.mu.Unlock()
tr.wg.Add(1)
go func() {
defer tr.wg.Done()
defer func() { tr.mu.Lock(); delete(tr.inflight, key); tr.mu.Unlock() }()
tr.run(ctx, key)
}()
}
// Reconsider looks at every try again (a delivery that held them ended), and removes the ended tries older than TryKeep.
func (tr *Tryer) Reconsider(ctx context.Context) {
keys, err := tr.Tries.Keys(ctx)
if err != nil {
tr.logf("the tries cannot be listed: %v", err)
return
}
for _, k := range keys {
raw, _, found, err := tr.Tries.Get(ctx, k)
if err != nil || !found {
continue
}
if t, err := DecodeTry(raw); err == nil && t.Over() && tr.now().Sub(t.Ended) > TryKeep {
if d, ok := tr.Tries.(interface {
Delete(ctx context.Context, key string) error
}); ok {
if err := d.Delete(ctx, k); err != nil {
tr.logf("%s ended %s ago and could not be removed: %v", k, tr.now().Sub(t.Ended).Round(time.Hour), err)
}
}
continue
}
tr.Put(ctx, k, raw)
}
}
// run claims the try by compare-and-set, judges it within its bound and writes its end.
func (tr *Tryer) run(ctx context.Context, key string) {
raw, rev, found, err := tr.Tries.Get(ctx, key)
if err != nil || !found {
return
}
t, err := DecodeTry(raw)
if err != nil || t.Over() {
return
}
t.State, t.Started = TryRunning, tr.now().UTC()
data, err := encodeTry(t)
if err != nil {
return
}
if rev, err = tr.Tries.Update(ctx, key, data, rev); err != nil {
if !errors.Is(err, ErrConflict) {
tr.logf("%s: %v", key, err)
}
return
}
cs := &Case{Key: key, Group: key,
Pulls: []Pull{{Repository: t.Repository, Head: t.Head, Trunk: t.Trunk}},
Trunk: map[string]string{t.Repository: t.Trunk},
Queued: t.Asked,
Progress: &Progress{},
Try: &TryOptions{Level: t.Level, Run: t.Tests}}
out, judged := judgeWithin(ctx, tr.Judge, cs, tr.Bound)
if ctx.Err() != nil {
return // left running: the checker takes it over when it starts
}
v := out.Verdict
t.Ended = tr.now().UTC()
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)
switch {
case !judged || IsNodesFailure(v):
t.State = TryError
case v.State == delivery.VerdictPass:
t.State = TryPass
default:
t.State = TryFail
if f := firstFailed(v); f != "" {
t.Failed = []string{f}
}
}
if t.Tests != "" {
t.Said += " | only the tests " + t.Tests + " (Go); nothing kept"
}
if data, err = encodeTry(t); err == nil {
_, err = tr.Tries.Update(ctx, key, data, rev)
}
if err != nil {
tr.logf("%s: its end could not be written: %v", key, err)
return
}
tr.logf("%s: %s", key, t.State)
}
// judgeWithin runs a case within the bound (DefaultBound when 0). A judge that fails to judge, and a check that ran past its
// bound, are verdicts of failure naming the machine and the step, and judged says false for them.
func judgeWithin(ctx context.Context, j Judge, cs *Case, bound time.Duration) (Outcome, bool) {
if bound <= 0 {
bound = DefaultBound
}
run, cancel := context.WithTimeoutCause(ctx, bound, errBound)
defer cancel()
out, err := j.Judge(run, cs)
stage, node := cs.Progress.Now()
if node == "" {
node = "checker"
}
switch {
case ctx.Err() != nil:
return Outcome{}, false
case context.Cause(run) == errBound:
if stage == "" {
stage = "starting"
}
return Outcome{Verdict: failure(node, fmt.Sprintf("the check ran past its bound of %s, stuck at: %s", bound, stage))}, false
case err != nil:
return Outcome{Verdict: failure(node, err.Error())}, false
}
return out, true
}
@@ -0,0 +1,196 @@
package check
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// memTries is the tries state as the tools see it, over a memKV.
type memTries struct{ kv *memKV }
func (m memTries) Keys() ([]string, error) { return m.kv.Keys(context.Background()) }
func (m memTries) Get(key string) (json.RawMessage, error) {
v, _, _, err := m.kv.Get(context.Background(), key)
return v, err
}
func (m memTries) Create(key string, value []byte) error {
_, err := m.kv.Create(context.Background(), key, value)
return err
}
var tryTrunks = map[string]string{repoA: trunkA, repoSDK: trunkS, HQRepository: trunkA}
func readTry(t testing.TB, kv *memKV, id string) Try {
t.Helper()
raw, _, found, _ := kv.Get(context.Background(), id)
if !found {
t.Fatalf("no try %s", id)
}
tr, err := DecodeTry(raw)
if err != nil {
t.Fatal(err)
}
return tr
}
func TestATryIsAskedQueuedOncePerHeadAndAtMostFourAtOnce(t *testing.T) { // ADR 0501 §5
kv := newMemKV()
store := memTries{kv}
at := time.Date(2026, 10, 22, 18, 0, 0, 0, time.UTC)
id, err := AskTry(store, tryTrunks, TryAsk{Repository: repoA, Head: head1, Trunk: "main"}, at)
if err != nil || !strings.HasPrefix(id, "try-") {
t.Fatalf("id %q, %v", id, err)
}
tr := readTry(t, kv, id)
if tr.State != TryQueued || tr.Trunk != trunkA || tr.Branch != "main" || tr.Head != head1 || tr.Repository != repoA {
t.Errorf("try = %+v, want queued on the trunk's head", tr)
}
again, err := AskTry(store, tryTrunks, TryAsk{Repository: repoA, Head: head1, Trunk: "main"}, at.Add(time.Second))
if err != nil || again != id {
t.Errorf("a second ask of the same head answered %q, %v, want %q", again, err, id)
}
heads := []string{head2, trunkA2, trunkS2}
for i, h := range heads {
if _, err := AskTry(store, tryTrunks, TryAsk{Repository: repoA, Head: h, Trunk: trunkA}, at.Add(time.Duration(i+2)*time.Second)); err != nil {
t.Fatalf("try %d: %v", i+2, err)
}
}
if _, err := AskTry(store, tryTrunks, TryAsk{Repository: repoSDK, Head: head1}, at.Add(time.Minute)); err == nil || !strings.Contains(err.Error(), "4 tries wait or run") {
t.Errorf("a fifth try answered %v, want refused naming the four", err)
}
}
func TestATryOfARepositoryWithNoCheckedLevelIsRefused(t *testing.T) { // the agreed error of the try
store := memTries{newMemKV()}
for _, repo := range []string{HQRepository, "novox/mesh-lab"} {
_, err := AskTry(store, tryTrunks, TryAsk{Repository: repo, Head: head1, Trunk: "main"}, time.Now())
if err == nil || err.Error() != "no level of "+repo+" is checked by the checker" {
t.Errorf("%s: %v", repo, err)
}
}
for _, bad := range []TryAsk{{Repository: repoA, Head: "abc"}, {Repository: repoA, Head: head1, Level: "e2e"},
{Repository: repoA, Head: head1, Tests: "a b"}, {Repository: "x", Head: head1}, {Repository: repoA, Head: head1, Trunk: "-x"}} {
if _, err := AskTry(store, tryTrunks, bad, time.Now()); err == nil {
t.Errorf("%+v was taken", bad)
}
}
}
func TestATryIsJudgedAsAPullRequestsHeadAndOwesTheForgeNothing(t *testing.T) { // ADR 0501 §5
ctx := context.Background()
kv := newMemKV()
id, err := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Trunk: trunkA}, time.Now())
if err != nil {
t.Fatal(err)
}
b := newBench(t)
tr := &Tryer{Tries: kv, Judge: b.ex}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
got := readTry(t, kv, id)
if got.State != TryPass || got.Took == "" || got.Head != head1 || got.Trunk != trunkA || got.Said == "" || got.Levels[delivery.LevelUnit].Run != 2 {
t.Fatalf("try = %+v", got)
}
fetch := b.steps.tool("fetch")
if len(fetch) != 1 || fetch[0].args["commit"] != head1 || fetch[0].args["base"] != trunkA || fetch[0].args["ref"] != "" {
t.Errorf("fetch = %+v, want the head merged onto the trunk, with no pull request's ref", fetch)
}
if keys, _ := b.prev.Keys(ctx); len(keys) != 0 {
t.Errorf("a try wrote the previews %v", keys)
}
// An ended try is not run again.
raw, _, _, _ = kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
if n := len(b.steps.tool("fetch")); n != 1 {
t.Errorf("an ended try was fetched again: %d fetches", n)
}
}
func TestATryOfOneLevelAndSomeTestsRunsThemAloneAndKeepsNothing(t *testing.T) {
ctx := context.Background()
kv := newMemKV()
id, err := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1, Level: delivery.LevelIntegration, Tests: "TestBus|TestStore"}, time.Now())
if err != nil {
t.Fatal(err)
}
b := newBench(t)
tr := &Tryer{Tries: kv, Judge: b.ex}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
tests := b.steps.tool("test")
if len(tests) != 1 || tests[0].args["level"] != delivery.LevelIntegration || tests[0].args["run"] != "TestBus|TestStore" {
t.Fatalf("tests = %+v, want the integration level alone with the run pattern", tests)
}
if n := len(b.steps.tool("digest")); n != 0 {
t.Errorf("%d digests for a try of some tests", n)
}
if keys, _ := b.kept.Keys(ctx); len(keys) != 0 {
t.Errorf("kept %v from a try of some tests", keys)
}
if got := readTry(t, kv, id); got.State != TryPass || !strings.Contains(got.Said, "only the tests TestBus|TestStore") {
t.Errorf("try = %+v", got)
}
}
func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T) {
ctx := context.Background()
kv := newMemKV()
b := newBench(t)
b.test = func(c stepCall) (any, error) { return captured[Tested](t, "go-test-failing"), nil }
id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
tr := &Tryer{Tries: kv, Judge: b.ex}
raw, _, _, _ := kv.Get(ctx, id)
tr.Put(ctx, id, raw)
tr.Wait()
if got := readTry(t, kv, id); got.State != TryFail || len(got.Failed) != 1 || !strings.Contains(got.Failed[0], "TestThreeIsFour") {
t.Errorf("try = %+v", got)
}
// No build node: the try could not run, which is an error, not a failure of the change.
kv2 := newMemKV()
b2 := newBench(t)
b2.ex.Toolchains = entries{}
id2, _ := AskTry(memTries{kv2}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
tr2 := &Tryer{Tries: kv2, Judge: b2.ex}
raw, _, _, _ = kv2.Get(ctx, id2)
tr2.Put(ctx, id2, raw)
tr2.Wait()
if got := readTry(t, kv2, id2); got.State != TryError {
t.Errorf("try with no node = %+v, want error", got)
}
}
func TestATryWaitsWhileADeliveryBuildsAndATryLeftRunningIsTakenOver(t *testing.T) {
ctx := context.Background()
kv := newMemKV()
id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
// Left running by a checker that stopped.
raw, rev, _, _ := kv.Get(ctx, id)
cur, _ := DecodeTry(raw)
cur.State = TryRunning
data, _ := encodeTry(cur)
if _, err := kv.Update(ctx, id, data, rev); err != nil {
t.Fatal(err)
}
b := newBench(t)
held := true
tr := &Tryer{Tries: kv, Judge: b.ex, Hold: func() bool { return held }}
tr.Reconsider(ctx)
tr.Wait()
if n := len(b.steps.tool("fetch")); n != 0 {
t.Fatalf("a try ran while a delivery builds")
}
held = false
tr.Reconsider(ctx)
tr.Wait()
if got := readTry(t, kv, id); got.State != TryPass {
t.Errorf("the try left running = %+v, want taken over and passed", got)
}
}
@@ -92,7 +92,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
t.Errorf("module = %q", m.Module) t.Errorf("module = %q", m.Module)
} }
// delivery.CheckerKeptHistory, CheckerOwedHistory, PreviewsHistory, CheckerVerdictsHistory; the names follow the SDK's buckets. // delivery.CheckerKeptHistory, CheckerOwedHistory, PreviewsHistory, CheckerVerdictsHistory; the names follow the SDK's buckets.
if want := []state{{Name: "kept", History: 1}, {Name: "owed", History: 5}, {Name: "previews", History: 5}, {Name: "verdicts", History: 5}}; !reflect.DeepEqual(m.State, want) { 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) 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",
@@ -118,7 +118,8 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
t.Errorf("streams = %+v, want %+v", m.Streams, want) t.Errorf("streams = %+v, want %+v", m.Streams, want)
} }
// The read tools (issue 1686) are served by the tools bundle, not by the checker's process. // The read tools (issue 1686) are served by the tools bundle, not by the checker's process.
if want := []string{"checker_kept", "checker_pull", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) { // The try's two tools (ADR 0501 §5) beside them: mesh-controller's try calls them.
if want := []string{"checker_kept", "checker_previews", "checker_pull", "checker_status", "checker_try", "checker_try_read", "checker_verdicts"}; !reflect.DeepEqual(sorted(m.Tools), want) {
t.Errorf("tools = %v, want %v", m.Tools, want) t.Errorf("tools = %v, want %v", m.Tools, want)
} }
for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
@@ -13,6 +13,7 @@
{ "name": "kept", "history": 1 }, { "name": "kept", "history": 1 },
{ "name": "owed", "history": 5 }, { "name": "owed", "history": 5 },
{ "name": "previews", "history": 5 }, { "name": "previews", "history": 5 },
{ "name": "tries", "history": 1 },
{ "name": "verdicts", "history": 5 } { "name": "verdicts", "history": 5 }
], ],
"reads": [ "reads": [
@@ -31,7 +32,11 @@
], ],
"tools": [ "tools": [
"checker_kept", "checker_kept",
"checker_previews",
"checker_pull", "checker_pull",
"checker_status",
"checker_try",
"checker_try_read",
"checker_verdicts" "checker_verdicts"
], ],
"invokes": [ "invokes": [
@@ -96,27 +96,6 @@ func (p *performer) owed(ctx context.Context, key string, rev uint64, value []by
p.perform(ctx, "owed."+key, key, rev, value) p.perform(ctx, "owed."+key, key, rev, value)
} }
// checkOwed performs the status the checker owes on a head (mesh-delivery-checker_owed; novox/hq ADR 0484 clause 2) at
// revision rev, once. Its keys are kept apart from the gate's, and a key that is not the checker's one act, the status
// mesh/check, is left.
func (p *performer) checkOwed(ctx context.Context, key string, rev uint64, value []byte) {
doneKey := "check." + key
if p.seen(ctx, doneKey, rev) {
return
}
if _, _, _, err := delivery.ParseCheckerOwedKey(key); err != nil {
p.logger.Printf("%s at revision %d is not a status of the checker, and is left: %v", key, rev, err)
p.keep(ctx, doneKey, rev)
return
}
if _, err := delivery.DecodeCheckerOwed(value); err != nil {
p.logger.Printf("%s at revision %d is not a status of the checker, and is left: %v", key, rev, err)
p.keep(ctx, doneKey, rev)
return
}
p.perform(ctx, doneKey, key, rev, value)
}
// perform does the act of an owed key once for the revision, under the bookkeeping key doneKey. // perform does the act of an owed key once for the revision, under the bookkeeping key doneKey.
func (p *performer) perform(ctx context.Context, doneKey, key string, rev uint64, value []byte) { func (p *performer) perform(ctx context.Context, doneKey, key string, rev uint64, value []byte) {
if p.seen(ctx, doneKey, rev) { if p.seen(ctx, doneKey, rev) {
@@ -208,57 +208,3 @@ func TestAMergeWithTwoHandedStopsAtTheFirstRefusal(t *testing.T) { // ADR 0441
t.Errorf("calls %d messages %d, want the first refusal to stop the pass", len(f.calls), len(pub.data)) t.Errorf("calls %d messages %d, want the first refusal to stop the pass", len(f.calls), len(pub.data))
} }
} }
// checkerOwed is the status the checker owes, as it writes it (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed).
func checkerOwed(t *testing.T, number int, state, text string) (string, []byte) {
t.Helper()
key, err := delivery.CheckerOwedKey("novox/mesh-delivery", number, headA)
if err != nil {
t.Fatal(err)
}
v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Text: text, Target: "https://view.example/checks/g.0123456789ab"})
if err != nil {
t.Fatal(err)
}
return key, v
}
func TestTheCheckersStatusIsSetOncePerRevisionAndAKeyOfTheGateIsNotItsKey(t *testing.T) { // 1683/AC5 (ADR 0484 clause 2)
f := &forgeFake{}
p, _, done, _ := newPerformer(f)
key, v := checkerOwed(t, 12, "success", "pass aaaaaaaaaaaa on bbbbbbbbbbbb")
p.checkOwed(context.Background(), key, 3, v)
p.checkOwed(context.Background(), key, 3, v)
if len(f.calls) != 1 || f.calls[0].key != "gitea.gitea_commit_status" ||
f.calls[0].body != `{"context":"mesh/check","description":"pass aaaaaaaaaaaa on bbbbbbbbbbbb","owner":"novox","repo":"mesh-delivery","sha":"`+headA+`","state":"success","target_url":"https://view.example/checks/g.0123456789ab"}` {
t.Fatalf("calls = %+v", f.calls)
}
p.checkOwed(context.Background(), key, 4, v)
if len(f.calls) != 2 {
t.Errorf("a new revision was not performed: %+v", f.calls)
}
if _, ok := done["check."+key]; !ok {
t.Errorf("revision kept under %v, want check.%s", done, key)
}
// The gate's own owed bookkeeping is another key: the same owed key seen in the gate's bucket is its own act.
if _, ok := done["owed."+key]; ok {
t.Errorf("the checker's status was kept under the gate's key: %v", done)
}
}
func TestAnythingButTheCheckersStatusInItsBucketIsLeft(t *testing.T) { // ADR 0484 clause 2: the checker owes one act
f := &forgeFake{}
p, _, done, logs := newPerformer(f)
// A comment, and a status of another context: the SDK refuses both to the checker's encoder, so they are written raw.
commentKey, commentValue := owedKV(t, "novox/mesh-delivery", 12, "comment.1", delivery.Owed{Act: delivery.OwedComment, Text: "hello"})
gateKey, gateValue := owedKV(t, "novox/mesh-delivery", 12, "status=mesh/merge-gate",
delivery.Owed{Act: delivery.OwedStatus, Context: "mesh/merge-gate", State: "success", Text: "x", Target: "https://x.example"})
p.checkOwed(context.Background(), commentKey, 1, commentValue)
p.checkOwed(context.Background(), gateKey, 1, gateValue)
if len(f.calls) != 0 {
t.Errorf("the forge was called: %+v", f.calls)
}
if !strings.Contains(logs.String(), commentKey) || !strings.Contains(logs.String(), gateKey) || len(done) != 2 {
t.Errorf("log %q, kept %v: both keys are named and kept so they are not tried again", logs.String(), done)
}
}
@@ -30,9 +30,8 @@ import (
// The gate's buckets mesh-delivery-gate_merges and mesh-delivery-gate_owed are read and watched (the manifest's reads), the // The gate's buckets mesh-delivery-gate_merges and mesh-delivery-gate_owed are read and watched (the manifest's reads), the
// bucket performed is the bridge's own (its state), and the five forge tools of the gate's acts are invoked (issue 1341, 1342). // bucket performed is the bridge's own (its state), and the five forge tools of the gate's acts are invoked (issue 1341, 1342).
// //
// The checker's owed bucket is read like the gate's (reads) and the checker's stream checks is heard like EVENTS is (consumes // The bridge neither reads the checker's owed nor hears its stream (ADR 0521, R4e finding R2-1): the forge module writes
// mesh-delivery-checker.owed.written; ADR 0484 clause 2): those entries are written by analogy with the ones the controller // mesh/check from it. The stream checks is made here so a test can say the checker wrote a status.
// composes for the gate's buckets and for the EVENTS consumer, not taken from a composed file.
// //
// The lists are what mesh-controller's broker.PermissionsFor answers (internal/broker/nats.go:577) for a module principal // The lists are what mesh-controller's broker.PermissionsFor answers (internal/broker/nats.go:577) for a module principal
// with the manifest of module.json: consumes the three gitea.pull events, invokes gitea.gitea_open_pulls, // with the manifest of module.json: consumes the three gitea.pull events, invokes gitea.gitea_open_pulls,
@@ -108,14 +107,6 @@ accounts {
"mesh.mod.gitea.tool.gitea_add_comment" "mesh.mod.gitea.tool.gitea_add_comment"
"mesh.mod.gitea.tool.gitea_add_comment.*" "mesh.mod.gitea.tool.gitea_add_comment.*"
"mesh.stream.mesh-delivery-forge-bridge.event.forge.>" "mesh.stream.mesh-delivery-forge-bridge.event.forge.>"
"$JS.ACK.CHECKS.n1_mesh-delivery-forge-bridge.>"
"$JS.API.CONSUMER.INFO.CHECKS.n1_mesh-delivery-forge-bridge"
"$JS.API.CONSUMER.MSG.NEXT.CHECKS.n1_mesh-delivery-forge-bridge"
"$JS.API.CONSUMER.CREATE.KV_mesh-delivery-checker_owed.>"
"$JS.API.CONSUMER.DELETE.KV_mesh-delivery-checker_owed.>"
"$JS.API.DIRECT.GET.KV_mesh-delivery-checker_owed.>"
"$JS.API.STREAM.INFO.KV_mesh-delivery-checker_owed"
"$JS.FC.KV_mesh-delivery-checker_owed.>"
] } ] }
subscribe { allow: [ subscribe { allow: [
"_INBOX.n1.mesh-delivery-forge-bridge.>" "_INBOX.n1.mesh-delivery-forge-bridge.>"
@@ -217,15 +208,10 @@ func startBed(t *testing.T, openReply string, prepare func(ctx context.Context,
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "FORGE", Subjects: []string{"mesh.stream.mesh-delivery-forge-bridge.event.forge.>"}}); err != nil {
t.Fatal(err) t.Fatal(err)
} }
// The checker's stream checks and the bridge's durable consumer on its owed.written event; the checker's owed bucket is made // The checker's stream checks, which the bridge no longer hears (ADR 0521): a test says on it that the checker wrote a status.
// only by the test that needs it, as when the checker is assigned after the bridge started.
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{"mesh.stream.mesh-delivery-checker.event.>"}}); err != nil { if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{"mesh.stream.mesh-delivery-checker.event.>"}}); err != nil {
t.Fatal(err) t.Fatal(err)
} }
if _, err := js.CreateOrUpdateConsumer(ctx, "CHECKS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-forge-bridge", AckPolicy: jetstream.AckExplicitPolicy,
FilterSubjects: []string{"mesh.stream.mesh-delivery-checker.event.owed.written"}}); err != nil {
t.Fatal(err)
}
// The bridge's state bucket, made by the controller from the manifest's state entry before the process starts. // The bridge's state bucket, made by the controller from the manifest's state entry before the process starts.
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: headsBucket, History: 1}); err != nil { if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: headsBucket, History: 1}); err != nil {
t.Fatal(err) t.Fatal(err)
@@ -728,16 +714,17 @@ func forgeMerged(t *testing.T, b *bed) []string {
func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] } func sha256Sum(b []byte) []byte { s := sha256.Sum256(b); return s[:] }
// TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey: the checker's owed bucket is made after the bridge started; // TestBridgePerformsNoMeshCheckAct: the checker owes mesh/check on a head and says so on its stream; the bridge calls no
// the checker writes the status mesh/check it owes on a head (the SDK's CheckerOwedKey and EncodeCheckerOwed, which the checker // tool with mesh/check and publishes nothing of its own on EVENTS for it. The forge module writes mesh/check from the
// itself writes with) and says so on its stream; the bridge sets the status on the forge within 10 s, with the context // checker's owed state itself (ADR 0521; R4e round 2 finding R2-1).
// mesh/check, the state and the line; the same revision is not set again, a new revision is (ADR 0484 clause 2). func TestBridgePerformsNoMeshCheckAct(t *testing.T) { // R2-1, ADR 0521
func TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { // 1683/AC5, ADR 0484 §2
b := startBed(t, `{"result":{"pulls":[]}}`, nil) b := startBed(t, `{"result":{"pulls":[]}}`, nil)
calls := make(chan string, 8) calls := make(chan string, 8)
sub, err := b.ctl.Subscribe("mesh.mod.gitea.tool.gitea_commit_status", func(m *nats.Msg) { // Every forge call is watched, and none answered here: the bed's stand-ins answer the ones the bridge asks.
calls <- string(m.Data) sub, err := b.ctl.Subscribe("mesh.mod.gitea.>", func(m *nats.Msg) {
_ = m.Respond([]byte(`{"result":{},"node":"n0"}`)) if strings.Contains(string(m.Data), "mesh/check") {
calls <- m.Subject + " " + string(m.Data)
}
}) })
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
@@ -754,38 +741,26 @@ func TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
put := func(state, text string) { v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.0123456789ab"})
v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: state, Text: text, Target: "https://view.example/checks/g.0123456789ab"}) if err != nil {
if err != nil { t.Fatal(err)
t.Fatal(err)
}
if _, err := kv.Put(b.ctx, key, v); err != nil {
t.Fatal(err)
}
if _, err := b.js.Publish(b.ctx, owedWrittenSubject, []byte(`{"key":"`+key+`"}`)); err != nil {
t.Fatal(err)
}
} }
next := func() string { if _, err := kv.Put(b.ctx, key, v); err != nil {
select { t.Fatal(err)
case got := <-calls:
return got
case <-time.After(10 * time.Second):
t.Fatal("the status was not set within 10 s")
}
return ""
} }
put("pending", "checking 0123456789ab on aaaaaaaaaaaa") if _, err := b.js.Publish(b.ctx, "mesh.stream.mesh-delivery-checker.event.owed.written", []byte(`{"key":"`+key+`"}`)); err != nil {
if got := next(); !strings.Contains(got, `"context":"mesh/check"`) || !strings.Contains(got, `"state":"pending"`) || !strings.Contains(got, `"sha":"`+head+`"`) || !strings.Contains(got, `"repo":"mesh-delivery"`) { t.Fatal(err)
t.Fatalf("first call = %s", got)
}
put("success", "pass 0123456789ab on aaaaaaaaaaaa | unit 2 run, 3 kept")
if got := next(); !strings.Contains(got, `"state":"success"`) || !strings.Contains(got, "unit 2 run, 3 kept") {
t.Fatalf("second call = %s", got)
} }
select { select {
case extra := <-calls: case c := <-calls:
t.Errorf("a third call was made: %s", extra) t.Errorf("a forge tool was called with mesh/check: %s", c)
case <-time.After(500 * time.Millisecond): case <-time.After(2 * time.Second):
}
events, err := b.js.Stream(b.ctx, "EVENTS")
if err != nil {
t.Fatal(err)
}
if _, err := events.GetLastMsgForSubject(b.ctx, "mesh.mod.mesh-delivery-forge-bridge.event.>"); err == nil {
t.Error("the bridge published an event of its own for the checker's mesh/check")
} }
} }
@@ -1,82 +0,0 @@
package main
import (
"context"
"errors"
"fmt"
"log"
"strings"
"sync"
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// checksStream is the stream the checker carries its events on, as the bus names it (declared by the checker as the stream
// checks, heard by this module through consumes; novox/hq ADR 0484 clause 2).
const checksStream = "CHECKS"
// owedWrittenSubject is the subject of the checker's owed.written event.
const owedWrittenSubject = "mesh.stream." + delivery.CheckerModule + ".event.owed.written"
// lateWatch watches a bucket that may be made after the bridge started: the checker's owed bucket. The bridge holds no timer
// (ADR 0437 §8), so it watches the bucket where it exists at start, and otherwise when the checker says it wrote a status
// (owed.written, handled by Written).
type lateWatch struct {
ctx context.Context
js interface {
KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error)
}
bucket string
react func(ctx context.Context, key string, rev uint64, value []byte)
logger *log.Logger
errs chan error
mu sync.Mutex
watched bool
}
// ensure starts the watch if the bucket exists and is not watched.
func (w *lateWatch) ensure() error {
w.mu.Lock()
defer w.mu.Unlock()
if w.watched {
return nil
}
kv, err := w.js.KeyValue(w.ctx, w.bucket)
switch {
case errors.Is(err, jetstream.ErrBucketNotFound):
w.logger.Printf("the bucket %s does not exist yet: it is watched when the checker says it wrote a status", w.bucket)
return nil
case err != nil:
return fmt.Errorf("the bucket %s: %w", w.bucket, err)
}
w.watched = true
go func() {
w.errs <- watch(w.ctx, kv, func(key string, rev uint64, value []byte) {
hctx, cancel := context.WithTimeout(w.ctx, syncDeadline)
defer cancel()
w.react(hctx, key, rev, value)
})
}()
return nil
}
// Written takes one owed.written event of the checker in: its bucket is watched if it is not. A subject that is not the
// checker's is terminated; a failure stays unacknowledged, and the server delivers it again.
func (w *lateWatch) Written(m interface {
Subject() string
Ack() error
Term() error
}) {
if !strings.HasPrefix(m.Subject(), owedWrittenSubject) {
w.logger.Printf("%s: not the checker's owed.written, dropped", m.Subject())
_ = m.Term()
return
}
if err := w.ensure(); err != nil {
w.logger.Printf("%s: left for redelivery: %v", m.Subject(), err)
return
}
_ = m.Ack()
}
@@ -225,24 +225,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}) })
}() }()
} }
// The checker's statuses (mesh/check, unrequired until release R4f): its bucket is made when the checker is assigned, which
// may be after the bridge started, and the bridge holds no timer; the checker says on its stream each time it wrote a
// status, and the bucket is then watched (late.go).
late := &lateWatch{ctx: ctx, js: js, bucket: delivery.CheckerOwedBucket, react: perf.checkOwed, logger: logger, errs: watching}
if err := late.ensure(); err != nil {
return err
}
switch ccons, err := js.Consumer(ctx, checksStream, cred.Node+"_"+cred.Module); {
case err != nil:
logger.Printf("the consumer %s_%s on %s does not exist yet, so the checker's statuses are not heard until the bridge starts again: %v", cred.Node, cred.Module, checksStream, err)
default:
cc, err := ccons.Consume(func(m jetstream.Msg) { late.Written(m) },
jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
if err != nil {
return err
}
defer cc.Stop()
}
if err := said(ctx); err != nil { if err := said(ctx); err != nil {
return err return err
} }
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge
go 1.26.0 go 1.26.0
require ( require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
) )
+2 -6
View File
@@ -1,9 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc h1:ZJScJSnF6LgFl55/VJ9RT/lNroVWun469sXtx0nlVYw= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7 h1:wK9O1CgNk7mo3IdJqW/CCx/hebg+1F0rkvW8CvJfnRM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b h1:OjlhfiGr+wKLWxafV+BRjZx3TbBtwRYcm9mgoam/jv8=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -77,6 +77,12 @@ func OwedAct(key string, value []byte) (Act, error) {
if o.Context != context { if o.Context != context {
return Act{}, fmt.Errorf("the owed status %s says the context %q, not the %q its key names", key, o.Context, context) return Act{}, fmt.Errorf("the owed status %s says the context %q, not the %q its key names", key, o.Context, context)
} }
if o.Context == delivery.StatusContextCheck {
// No tool writes mesh/check (R4e finding K1), and the bridge performs no act of it: the forge module writes it from
// the checker's state mesh-delivery-checker.owed, read by a watch (ADR 0521, finding R2-1).
return Act{}, fmt.Errorf("the owed status %s is %s, which the bridge does not perform: the forge module writes it from mesh-delivery-checker.owed",
key, delivery.StatusContextCheck)
}
a.Call = Call{Tool: commitStatusTool, Args: map[string]any{"owner": owner, "repo": repo, "sha": head, "context": o.Context, a.Call = Call{Tool: commitStatusTool, Args: map[string]any{"owner": owner, "repo": repo, "sha": head, "context": o.Context,
"state": o.State, "description": o.Text, "target_url": o.Target}} "state": o.State, "description": o.Text, "target_url": o.Target}}
case delivery.OwedView: case delivery.OwedView:
@@ -2,6 +2,7 @@ package bridge
import ( import (
"reflect" "reflect"
"strings"
"testing" "testing"
"git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/delivery"
@@ -43,6 +44,37 @@ func TestOwedStatusIsACommitStatusCall(t *testing.T) { // 1341/AC1
} }
} }
// checkerKV is the status the checker owes, as it writes it (mesh-sdk go/delivery CheckerOwedKey, EncodeCheckerOwed).
func checkerKV(t *testing.T, o delivery.Owed) (string, []byte) {
t.Helper()
key, err := delivery.CheckerOwedKey("novox/mesh-delivery", 12, headA)
if err != nil {
t.Fatal(err)
}
v, err := delivery.EncodeCheckerOwed(o)
if err != nil {
t.Fatal(err)
}
return key, v
}
// R4e findings K1 and R2-1 (issue 1301 S4; ADR 0503 §3, ADR 0521): no tool writes mesh/check, since a tool is told no
// caller and node-tools invokes every tool; the forge module writes it from the checker's owed state, and the bridge
// performs no mesh/check act.
func TestOwedMeshCheckIsNoToolCall(t *testing.T) { // K1
key, v := checkerKV(t, delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.1"})
a, err := OwedAct(key, v)
if err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") {
t.Fatalf("OwedAct(mesh/check) = %+v, %v; want refused, naming where the forge module reads it", a, err)
}
// Nor through the gate's own owed bucket: the context is refused whichever key carries it.
gk, gv := owedValue(t, "novox/mesh-delivery", 12, "status=mesh/check",
delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://x.example"})
if a, err := OwedAct(gk, gv); err == nil || !strings.Contains(err.Error(), "mesh-delivery-checker.owed") {
t.Fatalf("OwedAct(gate's mesh/check) = %+v, %v; want refused", a, err)
}
}
func TestOwedViewNoteAndCommentCalls(t *testing.T) { // 1341/AC2 func TestOwedViewNoteAndCommentCalls(t *testing.T) { // 1341/AC2
cases := []struct { cases := []struct {
act string act string
@@ -86,10 +86,11 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
if !reflect.DeepEqual(m.Streams, []stream{{Name: "forge", Events: []string{"forge.**"}}}) { if !reflect.DeepEqual(m.Streams, []stream{{Name: "forge", Events: []string{"forge.**"}}}) {
t.Errorf("streams = %+v", m.Streams) t.Errorf("streams = %+v", m.Streams)
} }
if want := []string{"gitea.pull.closed", "gitea.pull.merged", "gitea.pull.updated", "mesh-delivery-checker.owed.written"}; !reflect.DeepEqual(sorted(m.Consumes), want) { // The checker's owed is the forge module's to write (ADR 0521, R4e finding R2-1): the bridge neither hears nor reads it.
if want := []string{"gitea.pull.closed", "gitea.pull.merged", "gitea.pull.updated"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
t.Errorf("consumes = %v, want %v", m.Consumes, want) t.Errorf("consumes = %v, want %v", m.Consumes, want)
} }
if want := []string{"mesh-delivery-checker.owed", "mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) { if want := []string{"mesh-delivery-gate.merges", "mesh-delivery-gate.owed"}; !reflect.DeepEqual(sorted(m.Reads), want) {
t.Errorf("reads = %v, want %v", m.Reads, want) t.Errorf("reads = %v, want %v", m.Reads, want)
} }
if want := []string{"gitea.gitea_add_comment", "gitea.gitea_commit_status", "gitea.gitea_commit_trees", "gitea.gitea_delivery_view", if want := []string{"gitea.gitea_add_comment", "gitea.gitea_commit_status", "gitea.gitea_commit_trees", "gitea.gitea_delivery_view",
@@ -108,7 +109,7 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}}) { // 1279/AC6, 1341 if !reflect.DeepEqual(m.State, []state{{Name: "heads", History: 1}, {Name: "performed", History: 1}}) { // 1279/AC6, 1341
t.Errorf("state = %+v, want the buckets heads and performed", m.State) t.Errorf("state = %+v, want the buckets heads and performed", m.State)
} }
// No other state, no bus, no seat, no tool (ADR 0437 §4). // No other state, no bus, no seat, no tool, no event (ADR 0437 §4; ADR 0521: no check.status).
for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} { for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "tools", "seat", "seats"} {
if _, ok := top[k]; ok { if _, ok := top[k]; ok {
t.Errorf("the manifest declares %q: the bridge has none", k) t.Errorf("the manifest declares %q: the bridge has none", k)
@@ -5,8 +5,7 @@
"consumes": [ "consumes": [
"gitea.pull.updated", "gitea.pull.updated",
"gitea.pull.merged", "gitea.pull.merged",
"gitea.pull.closed", "gitea.pull.closed"
"mesh-delivery-checker.owed.written"
], ],
"streams": [ "streams": [
{ "name": "forge", "events": ["forge.**"] } { "name": "forge", "events": ["forge.**"] }
@@ -16,7 +15,6 @@
{ "name": "performed", "history": 1 } { "name": "performed", "history": 1 }
], ],
"reads": [ "reads": [
"mesh-delivery-checker.owed",
"mesh-delivery-gate.merges", "mesh-delivery-gate.merges",
"mesh-delivery-gate.owed" "mesh-delivery-gate.owed"
], ],
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-indexer
go 1.26.0 go 1.26.0
require ( require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
) )
+2 -6
View File
@@ -1,9 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc h1:ZJScJSnF6LgFl55/VJ9RT/lNroVWun469sXtx0nlVYw= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7 h1:wK9O1CgNk7mo3IdJqW/CCx/hebg+1F0rkvW8CvJfnRM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b h1:OjlhfiGr+wKLWxafV+BRjZx3TbBtwRYcm9mgoam/jv8=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -35,18 +35,39 @@ func Tools(r view.Reader) []stdio.Tool {
mods, err := r.Modules(only) mods, err := r.Modules(only)
return map[string]any{"modules": mods}, err return map[string]any{"modules": mods}, err
}}, }},
{Name: view.CandidateReadTool,
Description: "What the indexer stored for one candidate (the open pull requests of a branch) under candidates.<group>, as stored: " +
"the pull requests and heads, the base, and per pull request the touched modules with their new-form fingerprint (fp2) or why " +
"they have none (none2). Read-only.",
Input: map[string]any{"type": "object", "required": []string{"group"}, "properties": map[string]any{
"group": map[string]any{"type": "string", "description": "the candidate's group, the branch name its pull requests share"}}},
Run: func(args map[string]any) (any, error) {
group, _ := args["group"].(string)
return r.Candidate(group)
}},
{Name: view.CandidatesReadTool,
Description: "Every candidate the indexer holds, by group, each as stored under candidates.<group>; with module, only those with a " +
"pull request touching that module. Keys that cannot be read are listed as unreadable. Read-only.",
Input: map[string]any{"type": "object", "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "only candidates touching this module"}}},
Run: func(args map[string]any) (any, error) {
module, _ := args["module"].(string)
c, bad, err := r.Candidates(module)
return map[string]any{"candidates": c, "unreadable": bad}, err
}},
{Name: view.BuilderBuildsReadTool, {Name: view.BuilderBuildsReadTool,
Description: "The builder's build keys, each as stored in mesh-delivery-builder_builds: key, module, fingerprint (src2) and build " + Description: "The builder's build keys, each as stored in mesh-delivery-builder_builds: key, module, fingerprint (src2) and build " +
"(state, by, step with the nodes it was built on, manifest, artifacts with their digests, toolchains, error). Narrow by module " + "(state, by, step with the nodes it was built on, manifest, artifacts with their digests, toolchains, error). Narrow by module " +
"or state (queued, building, done, failed). Read-only.", "or state (queued, building, done, failed); at most limit builds (default 50), newest revision first for a module, and the answer says when it was cut and lists keys that are no build key (unreadable). Without a module, give one to get its newest. Read-only.",
Input: map[string]any{"type": "object", "properties": map[string]any{ Input: map[string]any{"type": "object", "properties": map[string]any{
"module": map[string]any{"type": "string", "description": "only this module's builds"}, "module": map[string]any{"type": "string", "description": "only this module's builds"},
"state": map[string]any{"type": "string", "description": "only builds in this state"}}}, "state": map[string]any{"type": "string", "description": "only builds in this state"},
"limit": map[string]any{"type": "integer", "description": "at most this many builds (default 50), newest revision first"}}},
Run: func(args map[string]any) (any, error) { Run: func(args map[string]any) (any, error) {
module, _ := args["module"].(string) module, _ := args["module"].(string)
state, _ := args["state"].(string) state, _ := args["state"].(string)
rows, err := r.BuilderBuilds(module, state) limit, _ := args["limit"].(float64)
return map[string]any{"builds": rows}, err return r.BuilderBuilds(module, state, int(limit))
}}, }},
{Name: view.BuilderBuildReadTool, {Name: view.BuilderBuildReadTool,
Description: "One build of the builder as stored. Give the module and its fingerprint (fp2, src2:...), or the module alone for the build " + Description: "One build of the builder as stored. Give the module and its fingerprint (fp2, src2:...), or the module alone for the build " +
@@ -65,7 +86,13 @@ func Tools(r view.Reader) []stdio.Tool {
} }
func read(state stdio.KeptState) view.StateRead { func read(state stdio.KeptState) view.StateRead {
return view.StateRead{Keys: state.Keys, Get: func(key string) (json.RawMessage, error) { return view.StateRead{Keys: state.Keys, GetEntry: func(key string) (json.RawMessage, uint64, error) {
e, err := state.Get(key)
if err != nil || e == nil {
return nil, 0, err
}
return e.Value, e.Revision, nil
}, Get: func(key string) (json.RawMessage, error) {
e, err := state.Get(key) e, err := state.Get(key)
if err != nil || e == nil { if err != nil || e == nil {
return nil, err return nil, err
@@ -75,7 +102,7 @@ func read(state stdio.KeptState) view.StateRead {
} }
func main() { func main() {
r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds"))} r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates"))}
if err := stdio.Serve("", Tools(r)); err != nil { if err := stdio.Serve("", Tools(r)); err != nil {
fmt.Fprintf(os.Stderr, "[mesh-delivery-view-tools] %v\n", err) fmt.Fprintf(os.Stderr, "[mesh-delivery-view-tools] %v\n", err)
os.Exit(1) os.Exit(1)
@@ -2,6 +2,7 @@ package main
import ( import (
"encoding/json" "encoding/json"
"fmt"
"strings" "strings"
"testing" "testing"
@@ -117,13 +118,16 @@ func TestBuilderBuildsListsKeysAsStored(t *testing.T) {
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
rows := out.(map[string]any)["builds"].([]view.BuilderRow) rows := out.(view.BuildsAnswer).Builds
if u := out.(view.BuildsAnswer).Unreadable; len(u) != 1 || u[0] != "junk" {
t.Errorf("unreadable = %v, want [junk] (issue 1818)", u)
}
if len(rows) != 2 || rows[0].Key != keyA || rows[0].Module != "netcheck" || rows[0].Fingerprint != fp2A || string(rows[0].Build) != doneBuild || string(rows[1].Build) != failBuild { if len(rows) != 2 || rows[0].Key != keyA || rows[0].Module != "netcheck" || rows[0].Fingerprint != fp2A || string(rows[0].Build) != doneBuild || string(rows[1].Build) != failBuild {
t.Errorf("rows = %+v", rows) t.Errorf("rows = %+v", rows)
} }
for args, want := range map[string]string{"state": keyB, "module": keyA} { for args, want := range map[string]string{"state": keyB, "module": keyA} {
out, _ := run(t, r, view.BuilderBuildsReadTool, map[string]any{args: map[string]string{keyB: "failed", keyA: "netcheck"}[want]}) out, _ := run(t, r, view.BuilderBuildsReadTool, map[string]any{args: map[string]string{keyB: "failed", keyA: "netcheck"}[want]})
got := out.(map[string]any)["builds"].([]view.BuilderRow) got := out.(view.BuildsAnswer).Builds
if len(got) != 1 || got[0].Key != want { if len(got) != 1 || got[0].Key != want {
t.Errorf("%s filter = %+v, want %s", args, got, want) t.Errorf("%s filter = %+v, want %s", args, got, want)
} }
@@ -156,3 +160,107 @@ func TestBuilderBuildJoinsTheIndexersFingerprint(t *testing.T) {
t.Errorf("an unbuilt fingerprint: %v", err) t.Errorf("an unbuilt fingerprint: %v", err)
} }
} }
// bigBuilds is a bucket of n keys over 100 modules that counts the values read.
type bigBuilds struct {
fake fake
reads *int
}
func newBig(n int) bigBuilds {
f := fake{}
for i := 0; i < n; i++ {
f[fmt.Sprintf("mod%02d.%064x", i%100, i)] = doneBuild
}
return bigBuilds{fake: f, reads: new(int)}
}
func (b bigBuilds) reader() view.Reader {
r := b.fake.reader()
get := r.State.Get
r.Builds = view.StateRead{Keys: r.State.Keys, Get: func(k string) (json.RawMessage, error) { *b.reads++; return get(k) },
GetEntry: func(k string) (json.RawMessage, uint64, error) {
*b.reads++
raw, err := get(k)
var rev uint64
fmt.Sscanf(k[strings.Index(k, ".")+1:], "%x", &rev) // the fake's revision is the key's number
return raw, rev + 1, err
}}
return r
}
// 1827/AC1: with 2,000 keys, a module's builds are found by key prefix, so only that module's values are read (20, not 2,000),
// newest revision first, within the limit, and the answer says it was cut.
func TestBuilderBuildsFiltersByPrefixBeforeAnyGet(t *testing.T) {
b := newBig(2000)
out, err := run(t, b.reader(), view.BuilderBuildsReadTool, map[string]any{"module": "mod07", "limit": float64(5)})
if err != nil {
t.Fatal(err)
}
a := out.(view.BuildsAnswer)
if *b.reads != 20 {
t.Errorf("read %d values for a module of 20 keys, want 20", *b.reads)
}
if len(a.Builds) != 5 || !a.Cut || !strings.Contains(a.Note, "5") {
t.Errorf("answer = %d rows, cut %v, note %q", len(a.Builds), a.Cut, a.Note)
}
for i := 1; i < len(a.Builds); i++ {
if a.Builds[i-1].Revision < a.Builds[i].Revision {
t.Errorf("not newest first: %d before %d", a.Builds[i-1].Revision, a.Builds[i].Revision)
}
}
if a.Builds[0].Revision != 1908 {
t.Errorf("newest revision = %d, want 1908", a.Builds[0].Revision)
}
}
// 1827/AC2: with no module the answer reads at most the limit's worth in key order (default 50), never the 2,000.
func TestBuilderBuildsWithoutAModuleIsBounded(t *testing.T) {
b := newBig(2000)
out, err := run(t, b.reader(), view.BuilderBuildsReadTool, map[string]any{})
if err != nil {
t.Fatal(err)
}
a := out.(view.BuildsAnswer)
if len(a.Builds) != view.DefaultBuildsLimit || !a.Cut || *b.reads > view.DefaultBuildsLimit {
t.Errorf("%d rows, cut %v, %d reads", len(a.Builds), a.Cut, *b.reads)
}
b = newBig(2000)
out, _ = run(t, b.reader(), view.BuilderBuildsReadTool, map[string]any{"state": "failed"})
if a := out.(view.BuildsAnswer); *b.reads != view.MaxBuildReads || !a.Cut || len(a.Builds) != 0 {
t.Errorf("a state matching nothing: %d reads (want the bound %d), cut %v", *b.reads, view.MaxBuildReads, a.Cut)
}
}
// 1818: fp2 together with commit is refused, naming both.
func TestBuilderBuildRefusesFP2WithCommit(t *testing.T) {
r := both(fake{keyA: doneBuild}, fake{"netcheck": `{"v":3,"fp2":"` + fp2A + `","commit":"c1","path":"."}`})
_, err := run(t, r, view.BuilderBuildReadTool, map[string]any{"module": "netcheck", "fp2": fp2A, "commit": "c1"})
if err == nil || !strings.Contains(err.Error(), "fp2") || !strings.Contains(err.Error(), "commit") || !strings.Contains(err.Error(), "not both") {
t.Errorf("err = %v", err)
}
}
// 1681: indexer_candidates and indexer_candidate answer the stored value; a pull request's fp2 per touched module shows.
func TestCandidateToolsAnswerTheStoredValue(t *testing.T) {
cand := `{"v":1,"group":"feat/x","base":3,"heads":"h","pulls":[{"modules":["zsh"],"fps":{"zsh":{"fp2":"` + fp2B + `","manifest":"m"}}}]}`
other := `{"v":1,"group":"fix","base":3,"heads":"h","pulls":[{"modules":["netcheck"]}]}`
r := fake{"feat/x": cand, "fix": other, "bad": "{"}.reader()
r.CandidatesState = r.State
out, err := run(t, r, view.CandidateReadTool, map[string]any{"group": "feat/x"})
if err != nil || string(out.(json.RawMessage)) != cand || !strings.Contains(string(out.(json.RawMessage)), fp2B) {
t.Fatalf("candidate = %v, %v", out, err)
}
if _, err := run(t, r, view.CandidateReadTool, map[string]any{"group": "gone"}); err == nil || !strings.Contains(err.Error(), "no candidate gone") {
t.Errorf("err = %v", err)
}
out, err = run(t, r, view.CandidatesReadTool, map[string]any{})
m := out.(map[string]any)
if err != nil || len(m["candidates"].(map[string]json.RawMessage)) != 2 || len(m["unreadable"].([]string)) != 1 {
t.Errorf("all = %v, %v", m, err)
}
out, _ = run(t, r, view.CandidatesReadTool, map[string]any{"module": "zsh"})
if c := out.(map[string]any)["candidates"].(map[string]json.RawMessage); len(c) != 1 || c["feat/x"] == nil {
t.Errorf("by module = %v", c)
}
}
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-view
go 1.26.0 go 1.26.0
require ( require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9
github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
) )
+2 -14
View File
@@ -1,17 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019203925-84a764372337 h1:cZyFnvDiBpnNx8ifSUG9WA61OHsn7PC2ml2s4eD7AkA= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019203925-84a764372337/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019223929-58373d65b47a h1:Vjk2Kq+g/Q86N8CCHq3OatJIJ0JSy6Kh980ekGxNLmE=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019223929-58373d65b47a/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69 h1:kvtV/SPsdRVUL/BZ3xxGS6U8YrXy/t7TRBjVrD6VLw8=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261019230201-5786ba190c69/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020060201-88581d0acaef h1:xy3wuGWJpqMAXYdXolocYG6w9xnWqBltVUffoaziIKo=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261020060201-88581d0acaef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc h1:ZJScJSnF6LgFl55/VJ9RT/lNroVWun469sXtx0nlVYw=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261021192148-78364e318bdc/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7 h1:wK9O1CgNk7mo3IdJqW/CCx/hebg+1F0rkvW8CvJfnRM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022160024-9cb4d59093a7/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b h1:OjlhfiGr+wKLWxafV+BRjZx3TbBtwRYcm9mgoam/jv8=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022161855-e76a5a31a07b/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+152 -14
View File
@@ -7,6 +7,7 @@ package view
import ( import (
"encoding/json" "encoding/json"
"fmt" "fmt"
"slices"
"sort" "sort"
"strings" "strings"
@@ -23,12 +24,16 @@ const (
type StateRead struct { type StateRead struct {
Keys func() ([]string, error) Keys func() ([]string, error)
Get func(key string) (json.RawMessage, error) Get func(key string) (json.RawMessage, error)
// GetEntry, when set, also gives the revision of the value; builder_builds orders by it.
GetEntry func(key string) (json.RawMessage, uint64, error)
} }
// Reader answers the read tools from a modules state. // Reader answers the read tools from a modules state.
type Reader struct { type Reader struct {
State StateRead // the indexer's modules State StateRead // the indexer's modules
Builds StateRead // the builder's builds Builds StateRead // the builder's builds
CandidatesState StateRead // the indexer's candidates
} }
// Module is the stored value of modules.<module>, exactly as the indexer wrote it. // Module is the stored value of modules.<module>, exactly as the indexer wrote it.
@@ -84,6 +89,67 @@ func (r Reader) Modules(none2 bool) (map[string]json.RawMessage, error) {
return out, nil return out, nil
} }
// The read tools of the indexer's candidates (issue 1681): what the indexer stored under candidates.<group>, as stored, so the
// new-form fingerprint of each touched module of an open pull request (pulls[].fps) can be read live.
const (
CandidateReadTool = "indexer_candidate"
CandidatesReadTool = "indexer_candidates"
)
// Candidate is the stored value of the candidates key of a group, exactly as the indexer wrote it.
func (r Reader) Candidate(group string) (json.RawMessage, error) {
key, err := delivery.CandidatesKey(strings.TrimSpace(group))
if err != nil {
return nil, err
}
if r.CandidatesState.Get == nil {
return nil, fmt.Errorf("the candidates state cannot be read here")
}
raw, err := r.CandidatesState.Get(key)
if err != nil {
return nil, err
}
if len(raw) == 0 {
return nil, fmt.Errorf("no candidate %s in the candidates state", group)
}
return raw, nil
}
// Candidates is every candidate, by group, each as stored; with a module only the candidates with a pull request touching it.
// A value that is not readable is listed under unreadable with its key, not skipped.
func (r Reader) Candidates(module string) (map[string]json.RawMessage, []string, error) {
if r.CandidatesState.Keys == nil || r.CandidatesState.Get == nil {
return nil, nil, fmt.Errorf("the candidates state cannot be read here")
}
keys, err := r.CandidatesState.Keys()
if err != nil {
return nil, nil, err
}
sort.Strings(keys)
out := map[string]json.RawMessage{}
var unreadable []string
for _, k := range keys {
raw, err := r.CandidatesState.Get(k)
if err != nil {
return nil, nil, err
}
if len(raw) == 0 {
continue
}
group, perr := delivery.ParseCandidatesKey(k)
var c delivery.Candidate
if perr != nil || json.Unmarshal(raw, &c) != nil {
unreadable = append(unreadable, k)
continue
}
if module != "" && !slices.ContainsFunc(c.Pulls, func(p delivery.CandidatePull) bool { return slices.Contains(p.Modules, module) }) {
continue
}
out[group] = raw
}
return out, unreadable, nil
}
// The read tools of the builder's builds (novox/hq issue 1723): what the builder stored under `<module>.<fingerprint hex>` // The read tools of the builder's builds (novox/hq issue 1723): what the builder stored under `<module>.<fingerprint hex>`
// in mesh-delivery-builder_builds, answered as stored, so its states, nodes (in the step) and digests can be checked live. // in mesh-delivery-builder_builds, answered as stored, so its states, nodes (in the step) and digests can be checked live.
const ( const (
@@ -91,48 +157,100 @@ const (
BuilderBuildsReadTool = "builder_builds" BuilderBuildsReadTool = "builder_builds"
) )
// BuilderRow is one key of the builder's builds: the key, the module and the fingerprint it parses to, and the stored value // BuilderRow is one key of the builder's builds: the key, the module and the fingerprint it parses to, the revision of its
// as the builder wrote it (state, by, step, manifest, artifacts with digests, toolchains, error). // stored value (when the read gives one) and the stored value as the builder wrote it.
type BuilderRow struct { type BuilderRow struct {
Key string `json:"key"` Key string `json:"key"`
Module string `json:"module"` Module string `json:"module"`
Fingerprint string `json:"fingerprint"` Fingerprint string `json:"fingerprint"`
Revision uint64 `json:"revision,omitempty"`
Build json.RawMessage `json:"build"` Build json.RawMessage `json:"build"`
} }
// The bounds of builder_builds (issue 1827): the rows it returns when no limit is given, and the values it reads in one call.
// Reading a value is a bus round trip, so the answer must not read the whole bucket (it timed out at 30 s on the live one).
const (
DefaultBuildsLimit = 50
MaxBuildReads = 200
)
// BuildsAnswer is the answer of builder_builds: the rows, newest revision first, whether the answer was cut and why, and
// the keys that are no build key (Unreadable), which are listed rather than silently skipped (issue 1818).
type BuildsAnswer struct {
Builds []BuilderRow `json:"builds"`
Cut bool `json:"cut,omitempty"`
Note string `json:"note,omitempty"`
Unreadable []string `json:"unreadable,omitempty"`
}
func (r Reader) get(key string) (json.RawMessage, uint64, error) {
if r.Builds.GetEntry != nil {
return r.Builds.GetEntry(key)
}
raw, err := r.Builds.Get(key)
return raw, 0, err
}
func (r Reader) builderRow(key string) (BuilderRow, bool, error) { func (r Reader) builderRow(key string) (BuilderRow, bool, error) {
module, fp, err := delivery.ParseBuilderBuildKey(key) module, fp, err := delivery.ParseBuilderBuildKey(key)
if err != nil { if err != nil {
return BuilderRow{}, false, nil // a key that is no build key is not a build return BuilderRow{}, false, nil // a key that is no build key is not a build
} }
raw, err := r.Builds.Get(key) raw, rev, err := r.get(key)
if err != nil { if err != nil {
return BuilderRow{}, false, err return BuilderRow{}, false, err
} }
if len(raw) == 0 { if len(raw) == 0 {
return BuilderRow{}, false, nil return BuilderRow{}, false, nil
} }
return BuilderRow{Key: key, Module: module, Fingerprint: fp, Build: raw}, true, nil return BuilderRow{Key: key, Module: module, Fingerprint: fp, Revision: rev, Build: raw}, true, nil
} }
// BuilderBuilds lists the builder's keys, each as stored, sorted by key; module and state, when given, keep only the builds // BuilderBuilds lists the builder's keys, each as stored, newest revision first. With a module it considers only the keys
// of that module or in that state. // that begin `<module>.`, before any value is read; with a state it keeps the builds in that state. It returns at most limit
func (r Reader) BuilderBuilds(module, state string) ([]BuilderRow, error) { // rows (DefaultBuildsLimit when limit is not above zero) and reads at most MaxBuildReads values, and says when it cut the
// answer. Keys that are no build key are listed as unreadable. Without a module the keys are walked in key order, so the
// answer is the first rows found, not the newest of the bucket: give a module to get its newest.
func (r Reader) BuilderBuilds(module, state string, limit int) (BuildsAnswer, error) {
if r.Builds.Keys == nil || r.Builds.Get == nil { if r.Builds.Keys == nil || r.Builds.Get == nil {
return nil, fmt.Errorf("the builder's builds cannot be read here") return BuildsAnswer{}, fmt.Errorf("the builder's builds cannot be read here")
}
if limit <= 0 {
limit = DefaultBuildsLimit
} }
keys, err := r.Builds.Keys() keys, err := r.Builds.Keys()
if err != nil { if err != nil {
return nil, err return BuildsAnswer{}, err
} }
sort.Strings(keys) sort.Strings(keys)
rows := []BuilderRow{} prefix := ""
if module = strings.TrimSpace(module); module != "" {
prefix = module + "."
}
out := BuildsAnswer{Builds: []BuilderRow{}}
var candidates []string
for _, k := range keys { for _, k := range keys {
if prefix != "" && !strings.HasPrefix(k, prefix) {
continue
}
if _, _, err := delivery.ParseBuilderBuildKey(k); err != nil {
out.Unreadable = append(out.Unreadable, k)
continue
}
candidates = append(candidates, k)
}
reads := 0
for _, k := range candidates {
if reads >= MaxBuildReads {
out.Cut, out.Note = true, fmt.Sprintf("read %d values and stopped; give a module (and a state) to narrow the answer", MaxBuildReads)
break
}
reads++
row, ok, err := r.builderRow(k) row, ok, err := r.builderRow(k)
if err != nil { if err != nil {
return nil, err return BuildsAnswer{}, err
} }
if !ok || (module != "" && row.Module != module) { if !ok {
continue continue
} }
if state != "" { if state != "" {
@@ -143,9 +261,26 @@ func (r Reader) BuilderBuilds(module, state string) ([]BuilderRow, error) {
continue continue
} }
} }
rows = append(rows, row) out.Builds = append(out.Builds, row)
// Without a module the walk is in key order and stops at the limit; with one, all its (few) keys are read to order them.
if prefix == "" && len(out.Builds) >= limit {
if reads < len(candidates) {
out.Cut, out.Note = true, fmt.Sprintf("stopped at the limit of %d builds in key order, not newest first; give a module for its newest", limit)
}
break
}
} }
return rows, nil sort.SliceStable(out.Builds, func(i, j int) bool {
if out.Builds[i].Revision != out.Builds[j].Revision {
return out.Builds[i].Revision > out.Builds[j].Revision
}
return out.Builds[i].Key < out.Builds[j].Key
})
if len(out.Builds) > limit {
out.Builds = out.Builds[:limit]
out.Cut, out.Note = true, fmt.Sprintf("cut to the newest %d builds of %s", limit, module)
}
return out, nil
} }
// BuilderBuild is the build of module at the new-form fingerprint fp2; with an empty fp2 it is the build at the fingerprint // BuilderBuild is the build of module at the new-form fingerprint fp2; with an empty fp2 it is the build at the fingerprint
@@ -156,6 +291,9 @@ func (r Reader) BuilderBuild(module, fp2, commit string) (BuilderRow, error) {
if r.Builds.Get == nil { if r.Builds.Get == nil {
return BuilderRow{}, fmt.Errorf("the builder's builds cannot be read here") return BuilderRow{}, fmt.Errorf("the builder's builds cannot be read here")
} }
if fp2 != "" && commit != "" {
return BuilderRow{}, fmt.Errorf("give fp2 or commit, not both: fp2 %s names a build directly, commit %s asks for the build at the indexer's fingerprint of that commit", fp2, commit)
}
if fp2 == "" { if fp2 == "" {
raw, err := r.Module(module) raw, err := r.Module(module)
if err != nil { if err != nil {
+1 -1
View File
@@ -95,7 +95,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
} }
// The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer // The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer
// itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads. // itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads.
if want := []string{"indexer_module", "indexer_modules", "builder_build", "builder_builds"}; !reflect.DeepEqual(m.Tools, want) { if want := []string{"indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds"}; !reflect.DeepEqual(m.Tools, want) {
t.Errorf("tools = %v, want %v", m.Tools, want) t.Errorf("tools = %v, want %v", m.Tools, want)
} }
for _, a := range m.Build.Artifacts { for _, a := range m.Build.Artifacts {
+2
View File
@@ -6,6 +6,8 @@
"tools": [ "tools": [
"indexer_module", "indexer_module",
"indexer_modules", "indexer_modules",
"indexer_candidate",
"indexer_candidates",
"builder_build", "builder_build",
"builder_builds" "builder_builds"
], ],