Compare commits
23
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
e762028760 | ||
|
|
36bf3b7143 | ||
|
|
9fad90ee89 | ||
|
|
4b8b1c047c | ||
|
|
b714eab995 | ||
|
|
efca168c55 | ||
|
|
19d06163de | ||
|
|
0997b071a3 | ||
|
|
bb4a4a82dc | ||
|
|
08bc672488 | ||
|
|
90ac287e59 | ||
|
|
91e68ac13b | ||
|
|
173becf35e | ||
|
|
7d740be255 | ||
|
|
66ae4dc791 | ||
|
|
03eaf332ad | ||
|
|
cbdcf6da0e | ||
|
|
7abe230701 | ||
|
|
2855ae7ebf | ||
|
|
d9ace64681 | ||
|
|
f7d8622deb | ||
|
|
d83156fb21 | ||
|
|
16ca51dc1f |
@@ -7,6 +7,7 @@ import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
@@ -155,6 +156,8 @@ type bed struct {
|
||||
tools map[string]jetstream.KeyValue
|
||||
presence jetstream.KeyValue
|
||||
have jetstream.KeyValue
|
||||
instances jetstream.KeyValue
|
||||
stop func() // stops the builder started last
|
||||
fake *fakeBuildModules
|
||||
srvURL string
|
||||
dir string
|
||||
@@ -227,6 +230,7 @@ func startBed(t *testing.T) *bed {
|
||||
b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
|
||||
presence := mk(delivery.PresenceBucket, delivery.PresenceHistory)
|
||||
b.presence, b.have = presence, mk(delivery.HaveBucket, delivery.HaveHistory)
|
||||
b.instances = mk(instancesBucket, 1)
|
||||
for _, node := range []string{"n1", "n2"} {
|
||||
b.setOnline(t, node, true)
|
||||
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)
|
||||
done := make(chan error, 1)
|
||||
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.
|
||||
@@ -713,3 +719,26 @@ func TestAManifestFallingBackToAPredefinedKeyIsBuilt(t *testing.T) {
|
||||
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},
|
||||
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.
|
||||
watch := newToolchainWatch(ctx, js, tools, b.NodesChanged, logger)
|
||||
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)
|
||||
}
|
||||
}
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-builder
|
||||
go 1.26.0
|
||||
|
||||
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.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -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.20261021192148-78364e318bdc/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=
|
||||
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.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
// 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 int
|
||||
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
|
||||
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) {
|
||||
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" {
|
||||
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)
|
||||
}
|
||||
if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains",
|
||||
|
||||
@@ -7,7 +7,8 @@
|
||||
"mesh-build-node.toolchain.reported"
|
||||
],
|
||||
"state": [
|
||||
{ "name": "builds", "history": 5 }
|
||||
{ "name": "builds", "history": 5 },
|
||||
{ "name": "instances", "history": 1 }
|
||||
],
|
||||
"reads": [
|
||||
"mesh-build-go.steps",
|
||||
|
||||
@@ -7,8 +7,10 @@ import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"time"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
|
||||
"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.
|
||||
func Tools(r check.Reader) []stdio.Tool {
|
||||
return []stdio.Tool{
|
||||
@@ -59,12 +126,79 @@ func Tools(r check.Reader) []stdio.Tool {
|
||||
kept, err := r.Kept(level, digest)
|
||||
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 (a group of hq's pull requests alone is left out: merge-check.sh judges it). A key a build " +
|
||||
"module left running (started past the step bound, or its build module not running on its node) is listed apart under left_running " +
|
||||
"with why, and not counted. The places the checker holds 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)
|
||||
}
|
||||
|
||||
// stepBound is the checker's step bound as the checker reads it (MESH_CHECKER_STEP_BOUND), its default when unset or not a
|
||||
// duration.
|
||||
func stepBound() time.Duration {
|
||||
if d, err := time.ParseDuration(os.Getenv("MESH_CHECKER_STEP_BOUND")); err == nil && d > 0 {
|
||||
return d
|
||||
}
|
||||
return check.DefaultStepBound
|
||||
}
|
||||
|
||||
func main() {
|
||||
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 {
|
||||
r := check.Reader{VerdictState: read(stdio.State("verdicts")), KeptState: read(stdio.State("kept")), CandidateState: read(stdio.State("mesh-delivery-indexer.candidates")),
|
||||
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, HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound()}
|
||||
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)
|
||||
os.Exit(1)
|
||||
}
|
||||
|
||||
@@ -1,12 +1,17 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"slices"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
)
|
||||
|
||||
func TestTheToolsServedAreTheThreeReadTools(t *testing.T) {
|
||||
func TestTheToolsServedAreTheFiveReadTools(t *testing.T) {
|
||||
var names []string
|
||||
for _, tl := range Tools(check.Reader{}) {
|
||||
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)
|
||||
}
|
||||
}
|
||||
want := []string{check.VerdictsReadTool, check.PullReadTool, check.KeptReadTool}
|
||||
if len(names) != 3 || names[0] != want[0] || names[1] != want[1] || names[2] != want[2] {
|
||||
want := []string{check.VerdictsReadTool, check.PullReadTool, check.KeptReadTool, check.PreviewsReadTool, check.StatusReadTool}
|
||||
if !slices.Equal(names, want) {
|
||||
t.Errorf("tools = %v, want %v", names, want)
|
||||
}
|
||||
// With nothing to read from a tool says so and changes nothing.
|
||||
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 {
|
||||
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
|
||||
}
|
||||
|
||||
// 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 }
|
||||
|
||||
// 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 (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strings"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
)
|
||||
|
||||
@@ -27,6 +30,16 @@ type forgeReads struct {
|
||||
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) {
|
||||
owner, repo, _ := strings.Cut(repository, "/")
|
||||
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.
|
||||
func (f forgeReads) File(_ context.Context, repository, path, ref string) (string, error) {
|
||||
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 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,
|
||||
@@ -59,7 +72,7 @@ func (f forgeReads) File(_ context.Context, repository, path, ref string) (strin
|
||||
// not.
|
||||
func (f forgeReads) OnTrunk(_ context.Context, repository, commit string) (bool, error) {
|
||||
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 {
|
||||
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.
|
||||
func (f forgeReads) FullSHA(_ context.Context, repository, commit string) (string, error) {
|
||||
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 {
|
||||
return "", err
|
||||
}
|
||||
|
||||
@@ -9,6 +9,8 @@ import (
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"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)
|
||||
}
|
||||
}
|
||||
|
||||
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/jetstream"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/announce"
|
||||
"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
|
||||
@@ -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,
|
||||
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.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)})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -590,3 +593,91 @@ func TestAMemberNamingAMeshSDKCommitOffTheTrunkFailsNamingTheCommit(t *testing.T
|
||||
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 (
|
||||
"context"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
@@ -31,7 +32,7 @@ type judgeDeps struct {
|
||||
|
||||
func newJudge(d judgeDeps) check.Judge {
|
||||
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.
|
||||
@@ -47,10 +48,13 @@ type repoModules struct {
|
||||
logf func(format string, args ...any)
|
||||
|
||||
mu sync.Mutex
|
||||
langs map[string]map[string]bool // by manifest digest: the languages of its artifacts
|
||||
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
|
||||
langs map[string]map[string]bool // by manifest digest: the languages of its artifacts
|
||||
tsPkgs map[string][]string // by manifest digest: the names of its TypeScript package artifacts
|
||||
path map[string]string // by module: its directory in its repository
|
||||
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.
|
||||
@@ -58,11 +62,14 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted
|
||||
r.mu.Lock()
|
||||
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.tsPkgs, r.path, r.named = map[string][]string{}, map[string]string{}, map[string]map[string][]string{}
|
||||
}
|
||||
if deleted {
|
||||
delete(r.repoOf, key)
|
||||
delete(r.typed, key)
|
||||
delete(r.commit, key)
|
||||
delete(r.path, key)
|
||||
delete(r.named, key)
|
||||
r.mu.Unlock()
|
||||
return
|
||||
}
|
||||
@@ -75,12 +82,15 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted
|
||||
r.repoOf[key] = m.Repository
|
||||
r.typed[key] = m.Manifest
|
||||
r.commit[key] = m.Commit
|
||||
r.path[key] = m.Path
|
||||
r.named[key] = m.ArtifactPaths
|
||||
_, known := r.langs[m.Manifest]
|
||||
r.mu.Unlock()
|
||||
if m.Manifest == "" || known {
|
||||
return
|
||||
}
|
||||
set := map[string]bool{}
|
||||
var pkgs []string
|
||||
if mk, err := delivery.ManifestsKey(m.Manifest); err == nil {
|
||||
if raw, _, found, err := r.manifests.Get(ctx, mk); err == nil && found {
|
||||
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 != "" {
|
||||
set[a.Language] = true
|
||||
}
|
||||
if a.Kind == "package" && a.Language == "typescript" {
|
||||
pkgs = append(pkgs, a.Name)
|
||||
}
|
||||
}
|
||||
} 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)
|
||||
@@ -98,9 +111,37 @@ func (r *repoModules) Put(ctx context.Context, key string, value []byte, deleted
|
||||
}
|
||||
r.mu.Lock()
|
||||
r.langs[m.Manifest] = set
|
||||
r.tsPkgs[m.Manifest] = pkgs
|
||||
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.
|
||||
func (r *repoModules) BuildModules(repository string) []string {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// 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/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/buildrun"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
@@ -39,9 +40,6 @@ func eventSubject(event string) string {
|
||||
}
|
||||
|
||||
const (
|
||||
// defaultStepBound is how long the checker waits for one step of a build module (a fetch downloads the Go modules, a
|
||||
// test level runs a repository's tests).
|
||||
defaultStepBound = 30 * time.Minute
|
||||
// defaultLink is the base of the address a status points at; the verdict key follows.
|
||||
defaultLink = "mesh-delivery-view/checks"
|
||||
)
|
||||
@@ -113,6 +111,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.
|
||||
// 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 {
|
||||
logger := log.New(logOut, "mesh-delivery-checker: ", log.LstdFlags)
|
||||
path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
|
||||
@@ -151,7 +159,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
buckets := map[string]jetstream.KeyValue{}
|
||||
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 {
|
||||
return err
|
||||
}
|
||||
@@ -167,8 +175,13 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return err
|
||||
}
|
||||
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", check.DefaultStepBound)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -209,8 +222,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
default:
|
||||
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
|
||||
tryer.Hold = holds.Held
|
||||
announce := func(ctx context.Context, event, key string) {
|
||||
body, _ := json.Marshal(map[string]string{"key": key})
|
||||
if _, err := js.Publish(ctx, eventSubject(event), body); err != nil {
|
||||
@@ -225,6 +241,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
announce: announce,
|
||||
})
|
||||
ch.Announce = announce
|
||||
tryer.Judge = ch.Judge
|
||||
|
||||
// The toolchains first, so a check that starts from the first candidate finds the nodes.
|
||||
watch := newToolchainWatch(ctx, js, tools, ch.NodesChanged, logger)
|
||||
@@ -308,7 +325,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
go func() {
|
||||
errs <- follow(ctx, sw, func(e jetstream.KeyValueEntry) {
|
||||
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 +353,25 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err := ch.Sweep(ctx); err != nil {
|
||||
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)
|
||||
go func() { errs <- follow(ctx, w, each) }()
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
ch.Wait()
|
||||
tryer.Wait()
|
||||
return nil
|
||||
case err := <-errs:
|
||||
return err
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-checker
|
||||
go 1.26.0
|
||||
|
||||
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.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -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.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=
|
||||
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.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"fmt"
|
||||
"maps"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
@@ -35,13 +36,17 @@ type Case struct {
|
||||
Key string
|
||||
Group 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 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 []string
|
||||
// Queued is when the checker took the case up, for the time of its queue.
|
||||
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 *Progress
|
||||
}
|
||||
@@ -75,7 +80,7 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case {
|
||||
if err != 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 {
|
||||
switch head, known := trunk[p.Repository]; {
|
||||
case p.Repository == HQRepository:
|
||||
@@ -99,6 +104,22 @@ func CaseOf(c delivery.Candidate, trunk map[string]string) *Case {
|
||||
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.
|
||||
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).
|
||||
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 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": ...}.
|
||||
@@ -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
|
||||
// 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) {
|
||||
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)
|
||||
if err != nil {
|
||||
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.
|
||||
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
|
||||
@@ -313,14 +317,14 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
|
||||
return
|
||||
}
|
||||
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) {
|
||||
v.Took.Queue = began.Sub(cs.Queued).Milliseconds()
|
||||
}
|
||||
data, err := delivery.EncodeVerdict(v)
|
||||
if err != nil {
|
||||
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()}}
|
||||
if data, err = delivery.EncodeVerdict(v); err != nil {
|
||||
return
|
||||
@@ -335,6 +339,13 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
|
||||
ch.nodesFail[cs.Key] = epoch
|
||||
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.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
|
||||
// failure naming the machine and the step.
|
||||
func (ch *Checker) judge(ctx context.Context, cs *Case) Outcome {
|
||||
bound := 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())}
|
||||
}
|
||||
out, _ := judgeWithin(ctx, ch.Judge, cs, ch.bound())
|
||||
return out
|
||||
}
|
||||
|
||||
@@ -467,3 +460,44 @@ func (ch *Checker) Sweep(ctx context.Context) error {
|
||||
}
|
||||
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"
|
||||
trunkA2 = "abababababababababababababababababababab"
|
||||
trunkS = "5555555555555555555555555555555555555555"
|
||||
trunkS2 = "5656565656565656565656565656565656565656"
|
||||
)
|
||||
|
||||
// candidateOf is a candidate of a group with the given pull requests, as the indexer writes one.
|
||||
|
||||
@@ -2,10 +2,14 @@ package check
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"reflect"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
@@ -565,7 +569,7 @@ func trunksOf(t *testing.T, heads map[string]string) *Trunks {
|
||||
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.ex.Trunks = trunksOf(t, map[string]string{repoSDK: trunkS, "novox/mesh-catalog": trunkA})
|
||||
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"))
|
||||
}
|
||||
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.
|
||||
if _, err := (Kept{Store: b.kept}).Keep(context.Background(), "unit", digA, keptValue("fixture/slow", 2)); err != nil {
|
||||
@@ -585,3 +589,175 @@ func TestNothingIsKeptForAWorkspaceGivenSiblingsBeside(t *testing.T) { // review
|
||||
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)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// keyCaller is a build module that starts every long step and answers its key on the node it was asked on.
|
||||
type keyCaller struct{}
|
||||
|
||||
func (keyCaller) CallOn(node, key string, _ any) (json.RawMessage, error) {
|
||||
return json.Marshal(map[string]string{"step": "lost1", "key": node + ".lost1", "state": delivery.StepRunning})
|
||||
}
|
||||
|
||||
// lostSteps is a steps state whose key never ends: the build module stopped, restarted or was unassigned with the step
|
||||
// running, and nothing writes its end.
|
||||
type lostSteps struct{}
|
||||
|
||||
func (lostSteps) Wait(ctx context.Context, _, _ string) (delivery.Step, error) {
|
||||
<-ctx.Done()
|
||||
return delivery.Step{}, ctx.Err()
|
||||
}
|
||||
|
||||
func TestAStepThatNeverEndsEndsTheCheckLoudlyAndFreesItsNodesPlace(t *testing.T) { // 1890/AC1
|
||||
b := newBench(t)
|
||||
places := &buildrun.Places{Capacity: map[string]int{"shanks": 1}}
|
||||
b.ex.Places = places
|
||||
b.ex.Steps = buildrun.Runner{Caller: keyCaller{}, Steps: lostSteps{}, Bound: 20 * time.Millisecond}
|
||||
out := b.judge(t, caseOf(t, pullOf(repoA, 7, head1)))
|
||||
if f := out.Verdict.Failed; out.Verdict.State != delivery.VerdictFail || f.Machine != "shanks" || !strings.Contains(f.Words, "shanks.lost1 did not end within") {
|
||||
t.Fatalf("verdict = %+v", out.Verdict)
|
||||
}
|
||||
if r, res := places.Running(), places.Reserved(); len(r) != 0 || len(res) != 0 {
|
||||
t.Fatalf("the lost step still holds shanks: running %v, reserved %v", r, res)
|
||||
}
|
||||
// The node's one place is free: the next check takes it at once.
|
||||
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
|
||||
defer cancel()
|
||||
release, err := places.Acquire(ctx, "shanks")
|
||||
if err != nil {
|
||||
t.Fatalf("shanks's place was not freed: %v", err)
|
||||
}
|
||||
release()
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
"fmt"
|
||||
"path"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
@@ -68,6 +69,11 @@ type Examiner struct {
|
||||
Builds BuildStates
|
||||
// Announce says a preview was written under the key (EventPreviewWritten); nil says nothing.
|
||||
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 interface{ Heads() map[string]string }
|
||||
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.
|
||||
type target struct {
|
||||
pull Pull
|
||||
@@ -98,8 +114,9 @@ type target struct {
|
||||
fetchArgs map[string]any
|
||||
// priority are the Go packages (import paths) or the Node module directories whose test files the merge changes.
|
||||
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 bool
|
||||
// beside are the lines `<owner/repo> <commit>` of the siblings the workspace was given, sorted: its tests may read them,
|
||||
// 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
|
||||
}
|
||||
|
||||
@@ -142,6 +159,9 @@ func (j *Examiner) Judge(ctx context.Context, c *Case) (Outcome, error) {
|
||||
return st.finish(ctx, *v), nil
|
||||
}
|
||||
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 _, t := range st.targets {
|
||||
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 {
|
||||
began := st.j.now()
|
||||
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
|
||||
for _, bm := range st.j.Modules.BuildModules(p.Repository) {
|
||||
node, tc, err := st.choose(bm)
|
||||
@@ -265,14 +298,17 @@ func (st *state) fetchAll(ctx context.Context) *delivery.Verdict {
|
||||
st.nodes[node] = true
|
||||
args := map[string]any{"repository": p.Repository, "commit": p.Head, "ref": refOf(p.Number), "base": p.Trunk}
|
||||
t.fetchArgs = args
|
||||
if bm == buildrun.NodeBuildModule && groupPackage != nil && p.Repository != SDKRepository {
|
||||
args["package"] = groupPackage
|
||||
}
|
||||
if bm == buildrun.GoBuildModule {
|
||||
beside, err := st.j.besideFor(ctx, st.c, p)
|
||||
if err != nil {
|
||||
return fail(node, err.Error())
|
||||
return fail(forgeMachine(err, node), err.Error()+unheardWords(err))
|
||||
}
|
||||
if len(beside) > 0 {
|
||||
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)
|
||||
@@ -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))
|
||||
}
|
||||
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, ", ")))
|
||||
}
|
||||
t.changed = t.fetched.Changed
|
||||
merged = append(merged, t.fetched)
|
||||
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
|
||||
}
|
||||
|
||||
// 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
|
||||
// the TypeScript module's .test. and .spec. files and test directories.
|
||||
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{}}
|
||||
t.plans[level] = plan
|
||||
if t.beside {
|
||||
// The digest covers the package's own files, the toolchain and the header, not the siblings placed beside the tree: a
|
||||
// package whose tests read one must never be kept, nor taken from a verdict kept for other inputs (ADR 0444).
|
||||
if st.c.Try != nil && st.c.Try.Run != "" {
|
||||
// A try of some tests runs them alone: a partial run proves no package, so nothing is taken or kept.
|
||||
plan.all = true
|
||||
return nil
|
||||
}
|
||||
@@ -421,6 +483,13 @@ func (st *state) planGo(ctx context.Context, t *target, level string) *delivery.
|
||||
plan.all = true
|
||||
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
|
||||
a := st.levels[level]
|
||||
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 {
|
||||
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)
|
||||
if err != nil {
|
||||
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 (
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"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.
|
||||
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 {
|
||||
@@ -70,7 +71,10 @@ func (p pieces) join(drop int) string {
|
||||
if drop < 2 && 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)
|
||||
}
|
||||
out := make([]string, 0, len(parts))
|
||||
@@ -90,14 +94,37 @@ func levelWords(name string, r delivery.LevelResult) string {
|
||||
return s
|
||||
}
|
||||
|
||||
func compositionWords(c *delivery.Composition) string {
|
||||
switch {
|
||||
case c == nil, c.State == delivery.CompositionNotRun:
|
||||
// compositionWords says the composition: not yet run, or once the assigner's check reports it (issue 1297) passed or
|
||||
// failed, with the time it took and the machines it composed (issue 1300).
|
||||
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"
|
||||
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 {
|
||||
@@ -149,7 +176,7 @@ func parts(v delivery.Verdict, cs *Case, p Pull, builds []BuildNote, slowest []s
|
||||
for _, x := range []struct {
|
||||
name string
|
||||
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 {
|
||||
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)
|
||||
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 {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
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)
|
||||
if !ok {
|
||||
@@ -88,7 +88,7 @@ func (st *state) checkPins(ctx context.Context) verdictPtr {
|
||||
}
|
||||
onTrunk, err := st.j.Forge.OnTrunk(ctx, SDKRepository, pin)
|
||||
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 {
|
||||
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 + ")"
|
||||
}
|
||||
|
||||
// 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
|
||||
// blame, issue 1820), the build node whose workspace was read otherwise.
|
||||
func forgeMachine(err error, node string) string {
|
||||
|
||||
@@ -253,3 +253,15 @@ func TestThePinOfTheControllerThatCannotBeReadFailsTheCheck(t *testing.T) { // r
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -5,8 +5,10 @@ import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"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
|
||||
@@ -17,6 +19,9 @@ const (
|
||||
VerdictsReadTool = "checker_verdicts"
|
||||
PullReadTool = "checker_pull"
|
||||
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).
|
||||
@@ -26,7 +31,17 @@ type StateRead struct {
|
||||
}
|
||||
|
||||
// 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,...`
|
||||
HaveState StateRead // the node-engine's have: whether a build module is on a node and running
|
||||
StepBound time.Duration // how long the checker waits for one step; DefaultStepBound when 0
|
||||
Now func() time.Time
|
||||
}
|
||||
|
||||
// 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.
|
||||
@@ -57,7 +72,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)
|
||||
}
|
||||
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"
|
||||
return out, nil
|
||||
}
|
||||
@@ -191,3 +206,222 @@ func (r Reader) Kept(level, digest string) (map[string]json.RawMessage, error) {
|
||||
}
|
||||
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"`
|
||||
}
|
||||
|
||||
// LeftStep is a key a build module left running that no step behind it runs any more, and why.
|
||||
type LeftStep struct {
|
||||
RunningStep
|
||||
Why string `json:"why"`
|
||||
}
|
||||
|
||||
// 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"`
|
||||
LeftRunning []LeftStep `json:"left_running"`
|
||||
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, st.LeftRunning = []RunningStep{}, []LeftStep{}
|
||||
now, bound := time.Now(), r.StepBound
|
||||
if r.Now != nil {
|
||||
now = r.Now()
|
||||
}
|
||||
if bound <= 0 {
|
||||
bound = DefaultStepBound
|
||||
}
|
||||
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")
|
||||
}
|
||||
if why := r.leftWhy(m, step, now, bound); why != "" {
|
||||
st.LeftRunning = append(st.LeftRunning, LeftStep{RunningStep: rs, Why: why})
|
||||
continue
|
||||
}
|
||||
st.Running = append(st.Running, rs)
|
||||
st.RunningByNode[step.Node]++
|
||||
}
|
||||
}
|
||||
if r.HaveState.Get == nil {
|
||||
st.Notes = append(st.Notes, "the node-engine's have is not readable here, so a step is not checked against its build module being on its node")
|
||||
}
|
||||
if len(st.LeftRunning) > 0 {
|
||||
st.Notes = append(st.Notes, "left_running are keys a build module left running (it stopped, restarted or was unassigned before it wrote the step's end): the checker's wait for each ended at the step bound, so none holds a place; the build module never ends them")
|
||||
}
|
||||
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 !checksAny(c) {
|
||||
continue // hq's pull requests alone: merge-check.sh judges them, the checker never does
|
||||
}
|
||||
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 places the checker holds are in its process and in no bucket: running_steps are the build modules' keys that can still be running (started within the step bound, their build module running on the node); 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
|
||||
}
|
||||
|
||||
// DefaultStepBound is how long the checker waits for one step of a build module (a fetch downloads the Go modules, a test
|
||||
// level runs a repository's tests); past it the check fails naming the step, and the step's place on its node is free.
|
||||
const DefaultStepBound = 30 * time.Minute
|
||||
|
||||
// leftWhy says why a key in the running state runs no step any more, or "" when it can still be running: its build module
|
||||
// is not on its node or is not running there (the node-engine's have), or it started longer ago than the step bound, when
|
||||
// the checker stopped waiting for it.
|
||||
func (r Reader) leftWhy(module string, step delivery.Step, now time.Time, bound time.Duration) string {
|
||||
if r.HaveState.Get != nil {
|
||||
key, err := delivery.HaveKey(step.Node, module)
|
||||
if err == nil {
|
||||
raw, err := r.HaveState.Get(key)
|
||||
switch {
|
||||
case err != nil:
|
||||
case len(raw) == 0:
|
||||
return fmt.Sprintf("%s is not on %s", module, step.Node)
|
||||
default:
|
||||
if h, err := delivery.DecodeHave(raw); err == nil && h.Phase != delivery.PhaseRunning {
|
||||
return fmt.Sprintf("%s on %s is %s", module, step.Node, h.Phase)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if step.Started.IsZero() {
|
||||
return fmt.Sprintf("it says no start, so no wait of the checker's (bound %s) can be behind it", bound)
|
||||
}
|
||||
if end := step.Started.Add(bound); now.After(end) {
|
||||
return fmt.Sprintf("started %s, past the step bound of %s: the checker stopped waiting for it at %s", step.Started.UTC().Format(time.RFC3339), bound, end.UTC().Format(time.RFC3339))
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// checksAny says whether a candidate has a pull request the checker judges: one outside hq.
|
||||
func checksAny(c delivery.Candidate) bool {
|
||||
for _, p := range c.Pulls {
|
||||
if p.Repository != HQRepository {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
@@ -3,8 +3,10 @@ package check
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
@@ -102,3 +104,140 @@ func TestTheKeptToolAnswersTheKeysAsStoredByLevelAndByDigest(t *testing.T) { //
|
||||
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(),
|
||||
HaveState: fakeState{"shanks.mesh-build-go": haveOf(t, delivery.PhaseRunning)}.read(),
|
||||
Now: func() time.Time { return time.Date(2026, 10, 22, 15, 10, 0, 0, time.UTC) },
|
||||
}
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
func haveOf(t *testing.T, phase delivery.Phase) []byte {
|
||||
t.Helper()
|
||||
raw, err := delivery.EncodeHave(delivery.Have{Phase: phase, Health: delivery.HealthOK})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
func stepOf(t *testing.T, id, node string, started time.Time) []byte {
|
||||
t.Helper()
|
||||
raw, err := delivery.EncodeStep(delivery.Step{ID: id, Tool: "build_go_fetch", Node: node, State: delivery.StepRunning, Started: started})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
// 1890: a key a build module left running (it stopped, restarted or was unassigned before it wrote the step's end) holds
|
||||
// no place in the checker, whose wait for it ended at the step bound; the status says it is left running and why, and
|
||||
// counts only the steps that can still run.
|
||||
func TestTheStatusCountsOnlyStepsThatCanStillRunAndSaysWhyTheOthersAreLeft(t *testing.T) { // 1890/AC2, AC3
|
||||
now := time.Date(2026, 10, 23, 7, 0, 0, 0, time.UTC)
|
||||
rd := Reader{
|
||||
StepStates: map[string]StateRead{
|
||||
"mesh-build-go": fakeState{
|
||||
"g14.fresh1": stepOf(t, "fresh1", "g14", now.Add(-5*time.Minute)),
|
||||
"g14.old1": stepOf(t, "old1", "g14", now.Add(-33*time.Hour)),
|
||||
"ace.gone1": stepOf(t, "gone1", "ace", now.Add(-2*time.Minute)),
|
||||
"shanks.down1": stepOf(t, "down1", "shanks", now.Add(-time.Minute)),
|
||||
}.read(),
|
||||
},
|
||||
HaveState: fakeState{
|
||||
"g14.mesh-build-go": haveOf(t, delivery.PhaseRunning),
|
||||
"shanks.mesh-build-go": haveOf(t, delivery.PhaseFailed),
|
||||
}.read(),
|
||||
StepBound: 30 * time.Minute,
|
||||
Now: func() time.Time { return now },
|
||||
}
|
||||
st, err := rd.Status()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(st.Running) != 1 || st.Running[0].ID != "fresh1" || len(st.RunningByNode) != 1 || st.RunningByNode["g14"] != 1 {
|
||||
t.Fatalf("running = %+v by node %v", st.Running, st.RunningByNode)
|
||||
}
|
||||
why := map[string]string{}
|
||||
for _, l := range st.LeftRunning {
|
||||
why[l.ID] = l.Why
|
||||
}
|
||||
if len(why) != 3 {
|
||||
t.Fatalf("left running = %+v", st.LeftRunning)
|
||||
}
|
||||
for id, want := range map[string]string{"old1": "past the step bound of 30m0s", "gone1": "mesh-build-go is not on ace", "down1": "mesh-build-go on shanks is failed"} {
|
||||
if !strings.Contains(why[id], want) {
|
||||
t.Errorf("why %s = %q, want it to say %q", id, why[id], want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 1890: a group whose pull requests are all hq's is never judged by the checker (merge-check.sh judges hq), so it is not
|
||||
// a candidate held back.
|
||||
func TestTheStatusDoesNotCountAnHQOnlyGroupAsUnjudged(t *testing.T) { // 1890/AC4
|
||||
cand := candidateOf(t, "pb-x", pullOf(HQRepository, 3, head2))
|
||||
ckey, _ := delivery.CandidatesKey("pb-x")
|
||||
st, err := Reader{CandidateState: fakeState{ckey: cand}.read(), VerdictState: fakeState{}.read()}.Status()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(st.Unjudged) != 0 {
|
||||
t.Errorf("unjudged = %v", st.Unjudged)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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
|
||||
// path.
|
||||
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"`
|
||||
Packages map[string]string `json:"packages"`
|
||||
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)
|
||||
}
|
||||
// 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)
|
||||
}
|
||||
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)
|
||||
}
|
||||
// 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)
|
||||
}
|
||||
for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
{ "name": "kept", "history": 1 },
|
||||
{ "name": "owed", "history": 5 },
|
||||
{ "name": "previews", "history": 5 },
|
||||
{ "name": "tries", "history": 1 },
|
||||
{ "name": "verdicts", "history": 5 }
|
||||
],
|
||||
"reads": [
|
||||
@@ -31,7 +32,11 @@
|
||||
],
|
||||
"tools": [
|
||||
"checker_kept",
|
||||
"checker_previews",
|
||||
"checker_pull",
|
||||
"checker_status",
|
||||
"checker_try",
|
||||
"checker_try_read",
|
||||
"checker_verdicts"
|
||||
],
|
||||
"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)
|
||||
}
|
||||
|
||||
// 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.
|
||||
func (p *performer) perform(ctx context.Context, doneKey, key string, rev uint64, value []byte) {
|
||||
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))
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
|
||||
+29
-54
@@ -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
|
||||
// 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
|
||||
// mesh-delivery-checker.owed.written; ADR 0484 clause 2): those entries are written by analogy with the ones the controller
|
||||
// composes for the gate's buckets and for the EVENTS consumer, not taken from a composed file.
|
||||
// The bridge neither reads the checker's owed nor hears its stream (ADR 0521, R4e finding R2-1): the forge module writes
|
||||
// mesh/check from it. The stream checks is made here so a test can say the checker wrote a status.
|
||||
//
|
||||
// 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,
|
||||
@@ -108,14 +107,6 @@ accounts {
|
||||
"mesh.mod.gitea.tool.gitea_add_comment"
|
||||
"mesh.mod.gitea.tool.gitea_add_comment.*"
|
||||
"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: [
|
||||
"_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 {
|
||||
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
|
||||
// only by the test that needs it, as when the checker is assigned after the bridge started.
|
||||
// The checker's stream checks, which the bridge no longer hears (ADR 0521): a test says on it that the checker wrote a status.
|
||||
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CHECKS", Subjects: []string{"mesh.stream.mesh-delivery-checker.event.>"}}); err != nil {
|
||||
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.
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: headsBucket, History: 1}); err != nil {
|
||||
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[:] }
|
||||
|
||||
// TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey: the checker's owed bucket is made after the bridge started;
|
||||
// the checker writes the status mesh/check it owes on a head (the SDK's CheckerOwedKey and EncodeCheckerOwed, which the checker
|
||||
// itself writes with) and says so on its stream; the bridge sets the status on the forge within 10 s, with the context
|
||||
// mesh/check, the state and the line; the same revision is not set again, a new revision is (ADR 0484 clause 2).
|
||||
func TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T) { // 1683/AC5, ADR 0484 §2
|
||||
// TestBridgePerformsNoMeshCheckAct: the checker owes mesh/check on a head and says so on its stream; the bridge calls no
|
||||
// tool with mesh/check and publishes nothing of its own on EVENTS for it. The forge module writes mesh/check from the
|
||||
// checker's owed state itself (ADR 0521; R4e round 2 finding R2-1).
|
||||
func TestBridgePerformsNoMeshCheckAct(t *testing.T) { // R2-1, ADR 0521
|
||||
b := startBed(t, `{"result":{"pulls":[]}}`, nil)
|
||||
calls := make(chan string, 8)
|
||||
sub, err := b.ctl.Subscribe("mesh.mod.gitea.tool.gitea_commit_status", func(m *nats.Msg) {
|
||||
calls <- string(m.Data)
|
||||
_ = m.Respond([]byte(`{"result":{},"node":"n0"}`))
|
||||
// Every forge call is watched, and none answered here: the bed's stand-ins answer the ones the bridge asks.
|
||||
sub, err := b.ctl.Subscribe("mesh.mod.gitea.>", func(m *nats.Msg) {
|
||||
if strings.Contains(string(m.Data), "mesh/check") {
|
||||
calls <- m.Subject + " " + string(m.Data)
|
||||
}
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -754,38 +741,26 @@ func TestBridgePerformsTheCheckersStatusOncePerRevisionOfItsOwedKey(t *testing.T
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
put := func(state, text string) {
|
||||
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)
|
||||
}
|
||||
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)
|
||||
}
|
||||
v, err := delivery.EncodeCheckerOwed(delivery.Owed{Act: delivery.OwedStatus, Context: delivery.StatusContextCheck, State: "success", Text: "pass", Target: "https://view.example/checks/g.0123456789ab"})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
next := func() string {
|
||||
select {
|
||||
case got := <-calls:
|
||||
return got
|
||||
case <-time.After(10 * time.Second):
|
||||
t.Fatal("the status was not set within 10 s")
|
||||
}
|
||||
return ""
|
||||
if _, err := kv.Put(b.ctx, key, v); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
put("pending", "checking 0123456789ab on aaaaaaaaaaaa")
|
||||
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.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)
|
||||
if _, err := b.js.Publish(b.ctx, "mesh.stream.mesh-delivery-checker.event.owed.written", []byte(`{"key":"`+key+`"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
select {
|
||||
case extra := <-calls:
|
||||
t.Errorf("a third call was made: %s", extra)
|
||||
case <-time.After(500 * time.Millisecond):
|
||||
case c := <-calls:
|
||||
t.Errorf("a forge tool was called with mesh/check: %s", c)
|
||||
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 {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge
|
||||
go 1.26.0
|
||||
|
||||
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.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -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.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=
|
||||
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.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -77,6 +77,12 @@ func OwedAct(key string, value []byte) (Act, error) {
|
||||
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)
|
||||
}
|
||||
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,
|
||||
"state": o.State, "description": o.Text, "target_url": o.Target}}
|
||||
case delivery.OwedView:
|
||||
|
||||
@@ -2,6 +2,7 @@ package bridge
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"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
|
||||
cases := []struct {
|
||||
act string
|
||||
|
||||
@@ -86,10 +86,11 @@ func TestManifestDeclarations(t *testing.T) { // 1235/AC1
|
||||
if !reflect.DeepEqual(m.Streams, []stream{{Name: "forge", Events: []string{"forge.**"}}}) {
|
||||
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)
|
||||
}
|
||||
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)
|
||||
}
|
||||
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
|
||||
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"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the bridge has none", k)
|
||||
|
||||
@@ -5,8 +5,7 @@
|
||||
"consumes": [
|
||||
"gitea.pull.updated",
|
||||
"gitea.pull.merged",
|
||||
"gitea.pull.closed",
|
||||
"mesh-delivery-checker.owed.written"
|
||||
"gitea.pull.closed"
|
||||
],
|
||||
"streams": [
|
||||
{ "name": "forge", "events": ["forge.**"] }
|
||||
@@ -16,7 +15,6 @@
|
||||
{ "name": "performed", "history": 1 }
|
||||
],
|
||||
"reads": [
|
||||
"mesh-delivery-checker.owed",
|
||||
"mesh-delivery-gate.merges",
|
||||
"mesh-delivery-gate.owed"
|
||||
],
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-indexer
|
||||
go 1.26.0
|
||||
|
||||
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.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -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.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=
|
||||
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.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -35,18 +35,39 @@ func Tools(r view.Reader) []stdio.Tool {
|
||||
mods, err := r.Modules(only)
|
||||
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,
|
||||
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 " +
|
||||
"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{
|
||||
"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) {
|
||||
module, _ := args["module"].(string)
|
||||
state, _ := args["state"].(string)
|
||||
rows, err := r.BuilderBuilds(module, state)
|
||||
return map[string]any{"builds": rows}, err
|
||||
limit, _ := args["limit"].(float64)
|
||||
return r.BuilderBuilds(module, state, int(limit))
|
||||
}},
|
||||
{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 " +
|
||||
@@ -65,7 +86,13 @@ func Tools(r view.Reader) []stdio.Tool {
|
||||
}
|
||||
|
||||
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)
|
||||
if err != nil || e == nil {
|
||||
return nil, err
|
||||
@@ -75,7 +102,7 @@ func read(state stdio.KeptState) view.StateRead {
|
||||
}
|
||||
|
||||
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 {
|
||||
fmt.Fprintf(os.Stderr, "[mesh-delivery-view-tools] %v\n", err)
|
||||
os.Exit(1)
|
||||
|
||||
@@ -2,6 +2,7 @@ package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -117,13 +118,16 @@ func TestBuilderBuildsListsKeysAsStored(t *testing.T) {
|
||||
if err != nil {
|
||||
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 {
|
||||
t.Errorf("rows = %+v", rows)
|
||||
}
|
||||
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]})
|
||||
got := out.(map[string]any)["builds"].([]view.BuilderRow)
|
||||
got := out.(view.BuildsAnswer).Builds
|
||||
if len(got) != 1 || got[0].Key != 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)
|
||||
}
|
||||
}
|
||||
|
||||
// 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)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,7 +3,7 @@ module mesh-delivery-view
|
||||
go 1.26.0
|
||||
|
||||
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.go v1.54.0
|
||||
)
|
||||
|
||||
@@ -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.20261019203925-84a764372337/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=
|
||||
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.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
|
||||
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
|
||||
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
|
||||
|
||||
@@ -7,6 +7,7 @@ package view
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"slices"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
@@ -23,12 +24,16 @@ const (
|
||||
type StateRead struct {
|
||||
Keys func() ([]string, 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.
|
||||
type Reader struct {
|
||||
State StateRead // the indexer's modules
|
||||
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.
|
||||
@@ -84,6 +89,67 @@ func (r Reader) Modules(none2 bool) (map[string]json.RawMessage, error) {
|
||||
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>`
|
||||
// in mesh-delivery-builder_builds, answered as stored, so its states, nodes (in the step) and digests can be checked live.
|
||||
const (
|
||||
@@ -91,48 +157,100 @@ const (
|
||||
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
|
||||
// as the builder wrote it (state, by, step, manifest, artifacts with digests, toolchains, error).
|
||||
// BuilderRow is one key of the builder's builds: the key, the module and the fingerprint it parses to, the revision of its
|
||||
// stored value (when the read gives one) and the stored value as the builder wrote it.
|
||||
type BuilderRow struct {
|
||||
Key string `json:"key"`
|
||||
Module string `json:"module"`
|
||||
Fingerprint string `json:"fingerprint"`
|
||||
Revision uint64 `json:"revision,omitempty"`
|
||||
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) {
|
||||
module, fp, err := delivery.ParseBuilderBuildKey(key)
|
||||
if err != nil {
|
||||
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 {
|
||||
return BuilderRow{}, false, err
|
||||
}
|
||||
if len(raw) == 0 {
|
||||
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
|
||||
// of that module or in that state.
|
||||
func (r Reader) BuilderBuilds(module, state string) ([]BuilderRow, error) {
|
||||
// BuilderBuilds lists the builder's keys, each as stored, newest revision first. With a module it considers only the keys
|
||||
// that begin `<module>.`, before any value is read; with a state it keeps the builds in that state. It returns at most limit
|
||||
// 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 {
|
||||
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()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return BuildsAnswer{}, err
|
||||
}
|
||||
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 {
|
||||
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)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return BuildsAnswer{}, err
|
||||
}
|
||||
if !ok || (module != "" && row.Module != module) {
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
if state != "" {
|
||||
@@ -143,9 +261,26 @@ func (r Reader) BuilderBuilds(module, state string) ([]BuilderRow, error) {
|
||||
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
|
||||
@@ -156,6 +291,9 @@ func (r Reader) BuilderBuild(module, fp2, commit string) (BuilderRow, error) {
|
||||
if r.Builds.Get == nil {
|
||||
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 == "" {
|
||||
raw, err := r.Module(module)
|
||||
if err != nil {
|
||||
|
||||
@@ -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
|
||||
// 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)
|
||||
}
|
||||
for _, a := range m.Build.Artifacts {
|
||||
|
||||
@@ -6,6 +6,8 @@
|
||||
"tools": [
|
||||
"indexer_module",
|
||||
"indexer_modules",
|
||||
"indexer_candidate",
|
||||
"indexer_candidates",
|
||||
"builder_build",
|
||||
"builder_builds"
|
||||
],
|
||||
|
||||
Reference in New Issue
Block a user