Delete the checker's hold, its own capacity and places, its node moves and its read of the controller's plans-building, so a check waits only in the build seats' streams (issue 1887 A5b, closes 1685; ADR 0525 §4, §7)
This commit is contained in:
@@ -143,32 +143,16 @@ func Tools(r check.Reader) []stdio.Tool {
|
||||
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), and the " +
|
||||
"candidates whose check waits by kind (waiting_by_kind: new, nodes, and trunk, which start after the other two). 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.",
|
||||
Description: "What runs: the steps in flight on the build modules by node, the candidates without a verdict for their heads, " +
|
||||
"and the candidates whose check is owed by kind (waiting_by_kind: new, nodes, and trunk, a re-judge pulled after the other two). 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. Nothing holds the checker and it has no capacity of its own: a check is an ask on the " +
|
||||
"build seats' streams, whose backlog the delivery view shows (novox/hq ADR 0525). 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 {
|
||||
@@ -180,9 +164,9 @@ func stepBound() time.Duration {
|
||||
|
||||
func main() {
|
||||
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()}
|
||||
PreviewState: read(stdio.State("previews")),
|
||||
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"))},
|
||||
HaveState: read(stdio.State("node-engine.have")), StepBound: stepBound()}
|
||||
tries := stdio.State(check.TriesState)
|
||||
source := stdio.State("mesh-delivery-gate.source")
|
||||
trunks := func() (map[string]string, error) {
|
||||
|
||||
@@ -105,43 +105,6 @@ func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.Key
|
||||
}
|
||||
}
|
||||
|
||||
// stepsWaiter waits for a build module's step by watching its key in the module's steps bucket (ADR 0469 clause 2): the
|
||||
// watch hands over the value the key holds now, so a step that ended before the watch began is read at once, and every
|
||||
// change after. No timer: the caller bounds the wait with its context.
|
||||
type stepsWaiter struct{ js bucketOpener }
|
||||
|
||||
func (w stepsWaiter) Wait(ctx context.Context, module, key string) (delivery.Step, error) {
|
||||
kv, err := w.js.KeyValue(ctx, delivery.StepsBucket(module))
|
||||
if err != nil {
|
||||
return delivery.Step{}, fmt.Errorf("the bucket %s: %w", delivery.StepsBucket(module), err)
|
||||
}
|
||||
watch, err := kv.Watch(ctx, key)
|
||||
if err != nil {
|
||||
return delivery.Step{}, err
|
||||
}
|
||||
defer watch.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return delivery.Step{}, ctx.Err()
|
||||
case e, ok := <-watch.Updates():
|
||||
if !ok {
|
||||
return delivery.Step{}, errors.New("the watch of the step ended")
|
||||
}
|
||||
if e == nil || !opOf(e) {
|
||||
continue
|
||||
}
|
||||
step, err := delivery.DecodeStep(e.Value())
|
||||
if err != nil {
|
||||
return delivery.Step{}, fmt.Errorf("the step %s is not a step: %w", key, err)
|
||||
}
|
||||
if step.State != delivery.StepRunning {
|
||||
return step, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// manifestReader reads manifests.<digest> of the indexer and parses the manifest it holds.
|
||||
type manifestReader struct{ kv jetstream.KeyValue }
|
||||
|
||||
|
||||
@@ -1,26 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
)
|
||||
|
||||
// heardCaller is the build runner's caller with the calls that reached nobody marked: no responders, or none answering within
|
||||
// the call's bound. The step never began there, so the check may give it to another node (issue 1820). A call that was
|
||||
// answered, with a failure or not, is left as it is.
|
||||
type heardCaller struct {
|
||||
call func(node, key string, body any) (json.RawMessage, error)
|
||||
}
|
||||
|
||||
func (h heardCaller) CallOn(node, key string, body any) (json.RawMessage, error) {
|
||||
raw, err := h.call(node, 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
|
||||
}
|
||||
@@ -1,30 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"testing"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
)
|
||||
|
||||
func TestACallThatReachedNobodyIsMarkedAndAnAnsweredFailureIsNot(t *testing.T) { // 1820/AC6
|
||||
for _, err := range []error{nats.ErrNoResponders, nats.ErrTimeout, context.DeadlineExceeded} {
|
||||
h := heardCaller{call: func(string, string, any) (json.RawMessage, error) { return nil, fmt.Errorf("call: %w", err) }}
|
||||
_, got := h.CallOn("g14", "m.t", nil)
|
||||
var u *check.Unheard
|
||||
if !errors.As(got, &u) {
|
||||
t.Errorf("%v: %v is not marked unheard", err, got)
|
||||
}
|
||||
}
|
||||
h := heardCaller{call: func(string, string, any) (json.RawMessage, error) { return nil, errors.New("the module refused") }}
|
||||
_, got := h.CallOn("g14", "m.t", nil)
|
||||
var u *check.Unheard
|
||||
if errors.As(got, &u) {
|
||||
t.Errorf("an answered failure %v is marked unheard", got)
|
||||
}
|
||||
}
|
||||
@@ -1,101 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log"
|
||||
"sync"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
"mesh-delivery-checker/internal/check"
|
||||
)
|
||||
|
||||
// holdWatch holds the checker back while a delivery builds (ADR 0484 clause 12): whether one does is the list of the keys of
|
||||
// the controller's bucket of the delivery plans in their build phase, read each time a check is about to start. Its keys age
|
||||
// out (a time to live, and no delete marker), which a watch never hears, so a watch cannot decide it (mesh-controller
|
||||
// internal/link/holds.go lists the keys too); the watch is only the wake-up, for a plan that ended by a delete. The bucket
|
||||
// is read as empty where it is absent (issue 1652 makes it) or cannot be read, saying so once. It may come into being after
|
||||
// the checker started and the checker holds no timer (ADR 0437 §8), so each time a check is about to start and the bucket is
|
||||
// not known yet, the checker looks for it.
|
||||
type holdWatch struct {
|
||||
ctx context.Context
|
||||
js bucketOpener
|
||||
keys func(ctx context.Context) ([]string, error)
|
||||
changed func(ctx context.Context) // a plan key changed: the checks that waited may start
|
||||
logger *log.Logger
|
||||
errs chan error
|
||||
mu sync.Mutex
|
||||
watched bool
|
||||
said bool
|
||||
}
|
||||
|
||||
func newHoldWatch(ctx context.Context, js bucketOpener, changed func(context.Context), logger *log.Logger) *holdWatch {
|
||||
return &holdWatch{ctx: ctx, js: js, changed: changed, logger: logger, errs: make(chan error, 1)}
|
||||
}
|
||||
|
||||
// Held says whether a delivery builds now.
|
||||
func (w *holdWatch) Held() bool {
|
||||
w.ensure()
|
||||
w.mu.Lock()
|
||||
keys := w.keys
|
||||
w.mu.Unlock()
|
||||
if keys == nil {
|
||||
return false
|
||||
}
|
||||
list, err := keys(w.ctx)
|
||||
if err != nil {
|
||||
w.say("the keys of %s cannot be listed, so it is read as empty: %v", check.PlansBuildingBucket, err)
|
||||
return false
|
||||
}
|
||||
return len(list) > 0
|
||||
}
|
||||
|
||||
func (w *holdWatch) say(format string, args ...any) {
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if !w.said {
|
||||
w.said = true
|
||||
w.logger.Printf(format, args...)
|
||||
}
|
||||
}
|
||||
|
||||
// ensure finds the bucket if it is not known and exists, and starts the wake-up watch.
|
||||
func (w *holdWatch) ensure() {
|
||||
w.mu.Lock()
|
||||
if w.watched || w.js == nil {
|
||||
w.mu.Unlock()
|
||||
return
|
||||
}
|
||||
w.mu.Unlock()
|
||||
kv, err := w.js.KeyValue(w.ctx, check.PlansBuildingBucket)
|
||||
switch {
|
||||
case errors.Is(err, jetstream.ErrBucketNotFound):
|
||||
w.say("the bucket %s does not exist yet: it is read as empty, so no check is held back", check.PlansBuildingBucket)
|
||||
return
|
||||
case err != nil:
|
||||
w.say("the bucket %s cannot be read, so it is read as empty: %v", check.PlansBuildingBucket, err)
|
||||
return
|
||||
}
|
||||
w.mu.Lock()
|
||||
defer w.mu.Unlock()
|
||||
if w.watched {
|
||||
return
|
||||
}
|
||||
w.watched, w.said = true, false
|
||||
w.keys = func(ctx context.Context) ([]string, error) {
|
||||
list, err := kv.Keys(ctx)
|
||||
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
||||
return nil, nil
|
||||
}
|
||||
return list, err
|
||||
}
|
||||
go func() {
|
||||
wake, err := kv.WatchAll(w.ctx)
|
||||
if err != nil {
|
||||
w.errs <- err
|
||||
return
|
||||
}
|
||||
w.errs <- follow(w.ctx, wake, func(jetstream.KeyValueEntry) { w.changed(w.ctx) })
|
||||
}()
|
||||
}
|
||||
@@ -1,35 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"log"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// The controller's plans-building keys age out of the bucket (a time to live, no delete marker), which a watch never hears:
|
||||
// whether a delivery builds is the list of the keys now, and the watch is only a wake-up (review D3).
|
||||
func TestADeliveryHoldsWhileItsKeyIsListedAndNotWhenItAgedOutWithNoEvent(t *testing.T) {
|
||||
listed := []string{"plan-1"}
|
||||
w := &holdWatch{watched: true, logger: log.New(io.Discard, "", 0), keys: func(context.Context) ([]string, error) { return listed, nil }}
|
||||
if !w.Held() {
|
||||
t.Fatal("a plan key is listed and nothing holds")
|
||||
}
|
||||
listed = nil // the key aged out: no delete reached the watch
|
||||
if w.Held() {
|
||||
t.Error("the key is gone from the bucket and the checker is still held")
|
||||
}
|
||||
}
|
||||
|
||||
func TestABucketThatCannotBeListedIsReadAsEmpty(t *testing.T) {
|
||||
w := &holdWatch{watched: true, logger: log.New(io.Discard, "", 0), keys: func(context.Context) ([]string, error) { return []string{"plan-1"}, errors.New("the bus is down") }}
|
||||
if w.Held() {
|
||||
t.Error("an unreadable bucket held the checker")
|
||||
}
|
||||
var absent holdWatch
|
||||
absent.logger = log.New(io.Discard, "", 0)
|
||||
if absent.Held() {
|
||||
t.Error("a watch with no bucket held the checker")
|
||||
}
|
||||
}
|
||||
@@ -561,34 +561,6 @@ func TestACheckLeftRunningIsTakenOverWhenTheCheckerStarts(t *testing.T) { // N6/
|
||||
}
|
||||
}
|
||||
|
||||
// TestNoCheckStartsWhileADeliveryBuildsAndOneStartsWhenItEnds: the controller's bucket of the plans in their build phase
|
||||
// holds a plan's key; the candidate waits, no step is called; the key deleted, the check starts and passes (ADR 0484 clause 12).
|
||||
func TestNoCheckStartsWhileADeliveryBuildsAndOneStartsWhenItEnds(t *testing.T) { // N8/AC1
|
||||
b := startBed(t)
|
||||
plans, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_plans-building", History: 1})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := plans.Put(b.ctx, "plan-1", []byte(`{"plan":"plan-1","since":"2026-10-21T19:00:00Z","seen":1}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
key := b.candidate(t, "feat/x", pull(repo, 7, head1))
|
||||
b.start(t, nil)
|
||||
time.Sleep(1500 * time.Millisecond) // a wait for nothing to happen: there is no event to wait on
|
||||
if _, err := b.buckets[delivery.CheckerVerdictsBucket].Get(b.ctx, key); err == nil {
|
||||
t.Fatal("a check started while a delivery builds")
|
||||
}
|
||||
if n := b.fake.count("mesh-build-go.build_go_fetch"); n != 0 {
|
||||
t.Fatalf("%d fetches while a delivery builds", n)
|
||||
}
|
||||
if err := plans.Delete(b.ctx, "plan-1"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if v := b.verdict(t, key, ended); v.State != delivery.VerdictPass {
|
||||
t.Errorf("verdict = %+v", v)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAMemberNamingAMeshSDKCommitOffTheTrunkFailsNamingTheCommit: the go.mod the forge answers names a commit that main does not
|
||||
// hold; the verdict fails naming the member and the commit and no test runs (ADR 0484 clause 9).
|
||||
func TestAMemberNamingAMeshSDKCommitOffTheTrunkFailsNamingTheCommit(t *testing.T) { // 1624/AC2
|
||||
|
||||
@@ -37,7 +37,8 @@ func TestTheCheckerAnnouncesEveryStateItKeepsAndOwedReadable(t *testing.T) { //
|
||||
if !equal(manifest, announced) {
|
||||
t.Errorf("announced %v, the manifest keeps %v", announced, manifest)
|
||||
}
|
||||
if want := []string{"owed", "previews", "verdicts"}; !equal(readable, want) {
|
||||
// kept is read by the Go build module for a check ask (issue 1887 design limit 1).
|
||||
if want := []string{"kept", "owed", "previews", "verdicts"}; !equal(readable, want) {
|
||||
t.Errorf("readable %v, want %v", readable, want)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -114,7 +114,9 @@ func durationEnv(getenv func(string) string, name string, def time.Duration) (ti
|
||||
// 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"},
|
||||
// The kept verdicts are read by the Go build module, which takes them for a check ask's unchanged packages (issue 1887's
|
||||
// design limit 1); the checker alone writes them (ADR 0444 clause 5).
|
||||
{Name: "kept", Readable: true},
|
||||
{Name: "owed", Readable: true},
|
||||
{Name: "previews", Readable: true},
|
||||
{Name: "tries"},
|
||||
@@ -189,10 +191,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
capacity, err := capacityFrom(strings.TrimSpace(getenv("MESH_CHECKER_SETTINGS")), getenv("MESH_CHECKER_CAPACITY"))
|
||||
if err != nil {
|
||||
return fmt.Errorf("the capacity setting: %w", err)
|
||||
}
|
||||
link := strings.TrimSpace(getenv("MESH_CHECKER_LINK"))
|
||||
if link == "" {
|
||||
link = defaultLink
|
||||
@@ -201,7 +199,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
presence := &buildrun.PresenceSet{}
|
||||
have := &buildrun.HaveSet{}
|
||||
trunks := &check.Trunks{}
|
||||
_ = capacity
|
||||
// A check is one ask per pull request and build module on the build seats' check stream, pulled by a build node with a
|
||||
// free slot (novox/hq ADR 0525 §1, §3): the checker chooses no node and starts no step.
|
||||
asks := buildrun.JetStreamAsks{JS: js}
|
||||
@@ -226,9 +223,6 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
}
|
||||
// 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 {
|
||||
@@ -379,7 +373,5 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
return err
|
||||
case err := <-watch.errs:
|
||||
return err
|
||||
case err := <-holds.errs:
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,29 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"os"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
// capacityFrom is the build capacity of each node (ADR 0478 clause 3): the module's setting `capacity`, `<node>=<steps at
|
||||
// once>` pairs separated by commas, laid into the settings file the controller composes (issue 1661); empty is one step at a
|
||||
// time on every node. With no settings file named, fallback is read instead, for a run outside the module's launch.
|
||||
func capacityFrom(path, fallback string) (map[string]int, error) {
|
||||
if path == "" {
|
||||
return buildrun.ParseCapacity(fallback)
|
||||
}
|
||||
raw, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("the settings file: %w", err)
|
||||
}
|
||||
var s struct {
|
||||
Capacity string `json:"capacity"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &s); err != nil {
|
||||
return nil, fmt.Errorf("the settings file %s is not JSON: %w", path, err)
|
||||
}
|
||||
return buildrun.ParseCapacity(s.Capacity)
|
||||
}
|
||||
@@ -1,68 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
)
|
||||
|
||||
func settingsFile(t *testing.T, body string) string {
|
||||
t.Helper()
|
||||
p := filepath.Join(t.TempDir(), "settings.json")
|
||||
if err := os.WriteFile(p, []byte(body), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return p
|
||||
}
|
||||
|
||||
// ADR 0484 clause 12: the build capacity is the module's setting `capacity`, laid into the settings file: empty is one step at a time everywhere.
|
||||
func TestCapacityIsReadFromTheSettingsFile(t *testing.T) {
|
||||
for body, want := range map[string]map[string]int{
|
||||
`{"capacity": "shanks=3,ace=2"}`: {"shanks": 3, "ace": 2},
|
||||
`{"capacity": ""}`: {},
|
||||
`{}`: {},
|
||||
} {
|
||||
got, err := capacityFrom(settingsFile(t, body), "")
|
||||
if err != nil || len(got) != len(want) || got["shanks"] != want["shanks"] || got["ace"] != want["ace"] {
|
||||
t.Errorf("%s: capacity %v, %v, want %v", body, got, err, want)
|
||||
}
|
||||
}
|
||||
p := &buildrun.Places{Capacity: mustCapacity(t, settingsFile(t, `{"capacity": ""}`))}
|
||||
if p.CapacityOf("shanks") != 1 || p.CapacityOf("ace") != 1 {
|
||||
t.Errorf("an empty setting gives %d and %d, want 1 everywhere", p.CapacityOf("shanks"), p.CapacityOf("ace"))
|
||||
}
|
||||
}
|
||||
|
||||
func mustCapacity(t *testing.T, path string) map[string]int {
|
||||
t.Helper()
|
||||
c, err := capacityFrom(path, "")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
func TestACapacitySettingThatIsNotOneIsRefusedNamingIt(t *testing.T) {
|
||||
if _, err := capacityFrom(settingsFile(t, `{"capacity": "shanks=-1"}`), ""); err == nil {
|
||||
t.Error("shanks=-1 accepted")
|
||||
}
|
||||
if c, err := capacityFrom(settingsFile(t, `{"capacity": "shanks=3,ace=2,g14=0,novox=0"}`), ""); err != nil || c["g14"] != 0 || len(c) != 4 {
|
||||
t.Errorf("the starting value: %v, %v", c, err)
|
||||
}
|
||||
if _, err := capacityFrom(settingsFile(t, `not json`), ""); err == nil {
|
||||
t.Error("a file that is not JSON accepted")
|
||||
}
|
||||
if _, err := capacityFrom(filepath.Join(t.TempDir(), "none"), ""); err == nil {
|
||||
t.Error("a settings file that is not there accepted")
|
||||
}
|
||||
}
|
||||
|
||||
// With no settings file named (a run outside the module's launch, as the tests do), the capacity is the environment's.
|
||||
func TestWithNoSettingsFileTheCapacityIsTheEnvironments(t *testing.T) {
|
||||
got, err := capacityFrom("", "ace=2")
|
||||
if err != nil || got["ace"] != 2 {
|
||||
t.Errorf("capacity %v, %v", got, err)
|
||||
}
|
||||
}
|
||||
@@ -1,3 +0,0 @@
|
||||
{
|
||||
"capacity": ""
|
||||
}
|
||||
@@ -4,7 +4,6 @@ import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"slices"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
@@ -69,10 +68,6 @@ type Checker struct {
|
||||
Trunks interface{ Heads() map[string]string }
|
||||
// Bound is how long one check may run; DefaultBound when 0.
|
||||
Bound time.Duration
|
||||
// Limit is how many checks run at once; two when 0. A freed place goes to the first waiting check by rank (ADR 0536 §4).
|
||||
Limit int
|
||||
// Hold says whether a delivery builds, so no new check starts (ADR 0484 clause 12); nil never holds.
|
||||
Hold func() bool
|
||||
// Link is the base of the address a status points at; the verdict key follows.
|
||||
Link string
|
||||
// Announce says a status was written (EventOwedWritten); nil says nothing.
|
||||
@@ -81,10 +76,6 @@ type Checker struct {
|
||||
Logf func(format string, args ...any)
|
||||
|
||||
once sync.Once
|
||||
limit int
|
||||
running int // checks holding a place
|
||||
waiting []*waiter // checks waiting for a place, in the order they asked
|
||||
seq uint64
|
||||
wg sync.WaitGroup
|
||||
mu sync.Mutex
|
||||
known map[string]delivery.Candidate // by candidates key
|
||||
@@ -97,11 +88,6 @@ type Checker struct {
|
||||
|
||||
func (ch *Checker) init() {
|
||||
ch.once.Do(func() {
|
||||
limit := ch.Limit
|
||||
if limit <= 0 {
|
||||
limit = 2
|
||||
}
|
||||
ch.limit = limit
|
||||
ch.known, ch.inflight, ch.again, ch.tried = map[string]delivery.Candidate{}, map[string]bool{}, map[string]bool{}, map[string]bool{}
|
||||
ch.nodesFail = map[string]uint64{}
|
||||
})
|
||||
@@ -190,10 +176,6 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat
|
||||
if cs == nil {
|
||||
return // nothing to judge: hq's pull requests, or no trunk known
|
||||
}
|
||||
if ch.Hold != nil && ch.Hold() {
|
||||
ch.logf("%s: not started, a delivery builds", cs.Key)
|
||||
return // started when the plan bucket empties (Reconsider)
|
||||
}
|
||||
cs.Queued = ch.now()
|
||||
ch.mu.Lock()
|
||||
if ch.inflight[cs.Key] {
|
||||
@@ -207,10 +189,11 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat
|
||||
go func() {
|
||||
defer ch.wg.Done()
|
||||
for {
|
||||
cs.Kind = ch.kindOf(ctx, cs) // a re-judge after a trunk move is asked on the recheck stream (ADR 0536 §4)
|
||||
ch.admit(cs, cs.Kind)
|
||||
// Every check is asked at once: the build seats' streams hold the backlog, a new head's check and a check after a
|
||||
// failure of the nodes on check, a re-judge after a trunk move on recheck, pulled after them (ADR 0525 §4, §9;
|
||||
// ADR 0536 §4). The checker holds no place and no capacity of its own.
|
||||
cs.Kind = ch.kindOf(ctx, cs)
|
||||
ch.try(ctx, key, cs)
|
||||
ch.leave()
|
||||
ch.mu.Lock()
|
||||
more := ch.again[cs.Key]
|
||||
delete(ch.again, cs.Key)
|
||||
@@ -223,62 +206,19 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat
|
||||
if next := CaseOf(ch.candidate(key, c), ch.heads()); next != nil {
|
||||
cs = next
|
||||
}
|
||||
cs.Queued = ch.now() // asked again: it waits for a place by its rank like any other
|
||||
cs.Queued = ch.now()
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// The kinds of a check waiting for a place, by the rank they start in (ADR 0536 §4): a head never judged and a head asked
|
||||
// again after a failure of the nodes first, then a head asked again only because its trunk moved.
|
||||
// The kinds of a check (ADR 0536 §4): a head never judged and a head asked again after a failure of the nodes are asked on
|
||||
// the build seats' check stream, a head asked again only because its trunk moved on the recheck stream, pulled after.
|
||||
const (
|
||||
KindNew = "new"
|
||||
KindNodes = "nodes"
|
||||
KindTrunk = "trunk"
|
||||
)
|
||||
|
||||
var kindRank = map[string]int{KindNew: 0, KindNodes: 0, KindTrunk: 1}
|
||||
|
||||
// waiter is a check waiting for a place.
|
||||
type waiter struct {
|
||||
group, key, kind string
|
||||
queued time.Time
|
||||
seq uint64
|
||||
ready chan struct{}
|
||||
}
|
||||
|
||||
// Waiting names a check waiting for a place: its group, its verdict key, its kind and when it was asked.
|
||||
type Waiting struct {
|
||||
Group string `json:"group"`
|
||||
Key string `json:"key"`
|
||||
Kind string `json:"kind"`
|
||||
Queued time.Time `json:"queued"`
|
||||
}
|
||||
|
||||
// Waiting are the checks waiting for a place, in the order they would start.
|
||||
func (ch *Checker) Waiting() []Waiting {
|
||||
ch.init()
|
||||
ch.mu.Lock()
|
||||
defer ch.mu.Unlock()
|
||||
ws := slices.Clone(ch.waiting)
|
||||
slices.SortStableFunc(ws, before)
|
||||
out := make([]Waiting, len(ws))
|
||||
for i, w := range ws {
|
||||
out[i] = Waiting{Group: w.group, Key: w.key, Kind: w.kind, Queued: w.queued}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// before orders waiting checks: the lower rank, then the one asked first, then the one that asked for a place first.
|
||||
func before(a, b *waiter) int {
|
||||
if ra, rb := kindRank[a.kind], kindRank[b.kind]; ra != rb {
|
||||
return ra - rb
|
||||
}
|
||||
if c := a.queued.Compare(b.queued); c != 0 {
|
||||
return c
|
||||
}
|
||||
return int(a.seq) - int(b.seq)
|
||||
}
|
||||
|
||||
// kindOf is the kind of a case's check, read from its verdict key: none (or one left running) is a head never judged; a
|
||||
// failure of the nodes is asked again after the nodes changed; a verdict on other trunk heads is a re-judge after a trunk
|
||||
// move. A verdict that stands is ranked with the re-judges: its claim finds nothing to do.
|
||||
@@ -297,47 +237,6 @@ func (ch *Checker) kindOf(ctx context.Context, cs *Case) string {
|
||||
return KindTrunk
|
||||
}
|
||||
|
||||
// admit returns when the check holds a place: at once when one is free and no check waits, else when a check's end hands its
|
||||
// place to this one as the first by rank. No timer: a check's end wakes the next (leave).
|
||||
func (ch *Checker) admit(cs *Case, kind string) {
|
||||
ch.mu.Lock()
|
||||
if ch.running < ch.limit && len(ch.waiting) == 0 {
|
||||
ch.running++
|
||||
ch.mu.Unlock()
|
||||
return
|
||||
}
|
||||
ch.seq++
|
||||
w := &waiter{group: cs.Group, key: cs.Key, kind: kind, queued: cs.Queued, seq: ch.seq, ready: make(chan struct{})}
|
||||
ch.waiting = append(ch.waiting, w)
|
||||
ch.handOn()
|
||||
ch.mu.Unlock()
|
||||
<-w.ready
|
||||
}
|
||||
|
||||
// leave frees the check's place and hands it to the first waiting check by rank.
|
||||
func (ch *Checker) leave() {
|
||||
ch.mu.Lock()
|
||||
ch.running--
|
||||
ch.handOn()
|
||||
ch.mu.Unlock()
|
||||
}
|
||||
|
||||
// handOn gives each free place to the first waiting check by rank; it is called with mu held.
|
||||
func (ch *Checker) handOn() {
|
||||
for ch.running < ch.limit && len(ch.waiting) > 0 {
|
||||
first := 0
|
||||
for i := range ch.waiting {
|
||||
if before(ch.waiting[i], ch.waiting[first]) < 0 {
|
||||
first = i
|
||||
}
|
||||
}
|
||||
w := ch.waiting[first]
|
||||
ch.waiting = slices.Delete(ch.waiting, first, first+1)
|
||||
ch.running++
|
||||
close(w.ready)
|
||||
}
|
||||
}
|
||||
|
||||
func (ch *Checker) candidate(key string, fallback delivery.Candidate) delivery.Candidate {
|
||||
ch.mu.Lock()
|
||||
defer ch.mu.Unlock()
|
||||
|
||||
@@ -298,24 +298,6 @@ func TestATrunkMoveRewritesTheSameVerdictKeyAndTheHeadsStatus(t *testing.T) { //
|
||||
}
|
||||
}
|
||||
|
||||
func TestNoCheckStartsWhileADeliveryBuildsAndOneStartsWhenItEnds(t *testing.T) { // N8/AC1
|
||||
ctx, j := context.Background(), passJudge()
|
||||
r := newRig(t, j)
|
||||
held := true
|
||||
r.ch.Hold = func() bool { return held }
|
||||
r.ch.Candidate(ctx, "feat/x", candidateOf(t, "feat/x", pullOf(repoA, 7, head1)))
|
||||
r.ch.Wait()
|
||||
if j.count() != 0 {
|
||||
t.Fatalf("judged %d times while held", j.count())
|
||||
}
|
||||
held = false
|
||||
r.ch.Reconsider(ctx)
|
||||
r.ch.Wait()
|
||||
if j.count() != 1 {
|
||||
t.Fatalf("judged %d times after the hold ended, want once", j.count())
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheStatusIsPendingWhileTheCheckRunsAndEndsSuccessOrFailure(t *testing.T) { // 1300/AC5 (unit part)
|
||||
ctx := context.Background()
|
||||
var seen []string
|
||||
|
||||
@@ -4,6 +4,7 @@ import (
|
||||
"context"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
@@ -71,11 +72,12 @@ func TestATrunkMoveMarksEveryOpenPullRequestIntoItPendingAtOnce(t *testing.T) {
|
||||
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 }
|
||||
// mesh-delivery's trunk moves; the re-judges it asks are held in their judge (in the stream, since R4g) meanwhile.
|
||||
release := make(chan struct{})
|
||||
r.ch.Judge = &judgeFunc{f: func(context.Context, *Case, int) (Outcome, error) { <-release; return Outcome{Verdict: passing()}, nil }}
|
||||
defer func() { close(release); r.ch.Wait() }()
|
||||
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)
|
||||
}
|
||||
@@ -90,17 +92,25 @@ func TestATrunkMoveMarksEveryOpenPullRequestIntoItPendingAtOnce(t *testing.T) {
|
||||
func TestACheckThatEndsAfterItsTrunkMovedOwesPendingNotItsVerdict(t *testing.T) { // gate (a): a pass on an old trunk is never shown
|
||||
ctx := context.Background()
|
||||
var r *rig
|
||||
release, judged := make(chan struct{}), make(chan struct{})
|
||||
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
|
||||
defer close(judged)
|
||||
} else {
|
||||
<-release // the check asked again for the new trunk waits in the stream meanwhile
|
||||
}
|
||||
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()
|
||||
<-judged
|
||||
for deadline := time.Now().Add(2 * time.Second); owedState(t, r, repoA, 7, head1) == "" && time.Now().Before(deadline); time.Sleep(time.Millisecond) {
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
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)
|
||||
}
|
||||
close(release)
|
||||
r.ch.Wait()
|
||||
}
|
||||
|
||||
@@ -1,7 +0,0 @@
|
||||
package check
|
||||
|
||||
// PlansBuildingBucket is the controller's bucket of the delivery plans in their build phase, one key for each plan, written
|
||||
// by the controller alone (issue 1652): a plan key there means a delivery builds on the build nodes, and no new check starts
|
||||
// meanwhile (ADR 0484 clause 12, ADR 0483 clause 6). The keys age out with no delete marker, so the keys are listed, not watched. The bucket may be absent: the controller's issue 1652 is delivered
|
||||
// before R4c-5b, and if it is not, the bucket is read as empty.
|
||||
const PlansBuildingBucket = "mesh-controller_plans-building"
|
||||
@@ -1,38 +0,0 @@
|
||||
package check
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
|
||||
func TestARunningCheckIsNotStoppedWhenADeliveryStartsToBuild(t *testing.T) { // ADR 0484 clause 12, ADR 0483 clause 4
|
||||
ctx := context.Background()
|
||||
held := false
|
||||
j := &judgeFunc{f: func(_ context.Context, _ *Case, call int) (Outcome, error) {
|
||||
if call == 1 {
|
||||
held = true // a delivery starts while this check runs
|
||||
}
|
||||
return Outcome{Verdict: passing()}, nil
|
||||
}}
|
||||
r := newRig(t, j)
|
||||
r.ch.Hold = func() bool { return held }
|
||||
r.ch.Candidate(ctx, "feat/x", candidateOf(t, "feat/x", pullOf(repoA, 7, head1)))
|
||||
r.ch.Wait()
|
||||
if v := r.verdict(t, keyOf(t, "feat/x", candidateOf(t, "feat/x", pullOf(repoA, 7, head1)))); v.State != delivery.VerdictPass {
|
||||
t.Errorf("verdict = %+v, want the running check to end as it was", v)
|
||||
}
|
||||
// A new candidate waits while the plan builds.
|
||||
r.ch.Candidate(ctx, "feat/y", candidateOf(t, "feat/y", pullOf(repoA, 8, head2)))
|
||||
r.ch.Wait()
|
||||
if j.count() != 1 {
|
||||
t.Errorf("judged %d times while a delivery builds, want only the one already running", j.count())
|
||||
}
|
||||
held = false
|
||||
r.ch.Reconsider(ctx)
|
||||
r.ch.Wait()
|
||||
if j.count() != 2 {
|
||||
t.Errorf("judged %d times after the delivery ended, want the waiting candidate too", j.count())
|
||||
}
|
||||
}
|
||||
@@ -3,10 +3,8 @@ package check
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"runtime"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
)
|
||||
@@ -27,146 +25,8 @@ type rankRig struct {
|
||||
errs []error // ctx errors the held check saw when released
|
||||
}
|
||||
|
||||
func newRankRig(t *testing.T) *rankRig {
|
||||
rr := &rankRig{held: make(chan struct{}), release: make(chan struct{})}
|
||||
j := &judgeFunc{f: func(ctx context.Context, c *Case, _ int) (Outcome, error) {
|
||||
rr.mu.Lock()
|
||||
rr.started = append(rr.started, c)
|
||||
hold := rr.hold
|
||||
rr.hold = false
|
||||
rr.mu.Unlock()
|
||||
if hold {
|
||||
close(rr.held)
|
||||
<-rr.release
|
||||
rr.mu.Lock()
|
||||
rr.errs = append(rr.errs, ctx.Err())
|
||||
rr.mu.Unlock()
|
||||
}
|
||||
return Outcome{Verdict: passing()}, nil
|
||||
}}
|
||||
rr.rig = newRig(t, j)
|
||||
rr.ch.Limit = 1
|
||||
now := time.Date(2026, 10, 23, 14, 0, 0, 0, time.UTC)
|
||||
var clock sync.Mutex
|
||||
rr.ch.Now = func() time.Time {
|
||||
clock.Lock()
|
||||
defer clock.Unlock()
|
||||
now = now.Add(time.Second)
|
||||
return now
|
||||
}
|
||||
return rr
|
||||
}
|
||||
|
||||
func group(n int) string { return fmt.Sprintf("feat/g%d", n) }
|
||||
|
||||
// judged puts n candidates in and judges them on the current trunk.
|
||||
func (rr *rankRig) judged(t *testing.T, n int) {
|
||||
for i := 1; i <= n; i++ {
|
||||
rr.ch.Candidate(context.Background(), group(i), candidateOf(t, group(i), pullOf(repoA, i, head1)))
|
||||
}
|
||||
rr.ch.Wait()
|
||||
rr.mu.Lock()
|
||||
rr.started = nil
|
||||
rr.mu.Unlock()
|
||||
}
|
||||
|
||||
// waitFor returns when n checks wait for a place.
|
||||
func (rr *rankRig) waitFor(t *testing.T, n int) {
|
||||
t.Helper()
|
||||
deadline := time.Now().Add(5 * time.Second)
|
||||
for {
|
||||
if w := len(rr.ch.Waiting()); w == n {
|
||||
return
|
||||
} else if time.Now().After(deadline) {
|
||||
t.Fatalf("%d checks wait, want %d", w, n)
|
||||
}
|
||||
runtime.Gosched()
|
||||
}
|
||||
}
|
||||
|
||||
// moveTrunkHoldingOne moves the trunk, so every judged candidate is asked again; the first re-judge to start is held.
|
||||
func (rr *rankRig) moveTrunkHoldingOne(t *testing.T, judged int) {
|
||||
rr.mu.Lock()
|
||||
rr.hold = true
|
||||
rr.mu.Unlock()
|
||||
rr.setTrunks(t, map[string]string{repoA: trunkA2, repoSDK: trunkS})
|
||||
rr.ch.TrunkMoved(context.Background())
|
||||
<-rr.held
|
||||
rr.waitFor(t, judged-1)
|
||||
}
|
||||
|
||||
func TestANewHeadStartsBeforeTheReJudgesOfATrunkMove(t *testing.T) { // 1897/AC1
|
||||
rr := newRankRig(t)
|
||||
rr.judged(t, 4)
|
||||
rr.moveTrunkHoldingOne(t, 4) // one runs, three re-judges wait
|
||||
rr.ch.Candidate(context.Background(), group(9), candidateOf(t, group(9), pullOf(repoA, 9, head2)))
|
||||
rr.waitFor(t, 4)
|
||||
close(rr.release)
|
||||
rr.ch.Wait()
|
||||
if len(rr.started) != 5 {
|
||||
t.Fatalf("%d checks started, want 5", len(rr.started))
|
||||
}
|
||||
if g := rr.started[1].Group; g != group(9) {
|
||||
t.Errorf("after the running check, %s started, want the new head's %s; order %v", g, group(9), groups(rr.started))
|
||||
}
|
||||
}
|
||||
|
||||
func TestTheReJudgeQueuedFirstStartsFirst(t *testing.T) { // 1897/AC2
|
||||
rr := newRankRig(t)
|
||||
rr.judged(t, 3)
|
||||
rr.moveTrunkHoldingOne(t, 3) // one runs, two re-judges wait
|
||||
close(rr.release)
|
||||
rr.ch.Wait()
|
||||
if len(rr.started) != 3 {
|
||||
t.Fatalf("%d checks started, want 3", len(rr.started))
|
||||
}
|
||||
if a, b := rr.started[1], rr.started[2]; !a.Queued.Before(b.Queued) {
|
||||
t.Errorf("%s (queued %s) started before %s (queued %s)", a.Group, a.Queued.Format(time.TimeOnly), b.Group, b.Queued.Format(time.TimeOnly))
|
||||
}
|
||||
}
|
||||
|
||||
func TestANewHeadDoesNotStopARunningReJudge(t *testing.T) { // 1897/AC3
|
||||
rr := newRankRig(t)
|
||||
rr.judged(t, 1)
|
||||
rr.moveTrunkHoldingOne(t, 1) // the re-judge runs
|
||||
rr.ch.Candidate(context.Background(), group(9), candidateOf(t, group(9), pullOf(repoA, 9, head2)))
|
||||
rr.waitFor(t, 1)
|
||||
close(rr.release)
|
||||
rr.ch.Wait()
|
||||
if len(rr.errs) != 1 || rr.errs[0] != nil {
|
||||
t.Fatalf("the running re-judge saw %v, want it never stopped", rr.errs)
|
||||
}
|
||||
if len(rr.started) != 2 || rr.started[0].Group != group(1) || rr.started[1].Group != group(9) {
|
||||
t.Errorf("started %v, want the re-judge to end before the new head starts", groups(rr.started))
|
||||
}
|
||||
key := keyOf(t, group(1), candidateOf(t, group(1), pullOf(repoA, 1, head1)))
|
||||
if v := rr.verdict(t, key); v.State != delivery.VerdictPass || v.Trunk[repoA] != trunkA2 {
|
||||
t.Errorf("the re-judge's verdict = %+v, want its pass on the moved trunk", v)
|
||||
}
|
||||
}
|
||||
|
||||
func TestWaitingChecksAreNamedByKind(t *testing.T) { // 1897/AC4, the checker's own view
|
||||
rr := newRankRig(t)
|
||||
rr.judged(t, 2)
|
||||
rr.moveTrunkHoldingOne(t, 2)
|
||||
rr.ch.Candidate(context.Background(), group(9), candidateOf(t, group(9), pullOf(repoA, 9, head2)))
|
||||
rr.waitFor(t, 2)
|
||||
kinds := map[string]string{}
|
||||
for _, w := range rr.ch.Waiting() {
|
||||
kinds[w.Group] = w.Kind
|
||||
}
|
||||
close(rr.release)
|
||||
rr.ch.Wait()
|
||||
running := rr.started[0].Group
|
||||
other := group(1)
|
||||
if running == other {
|
||||
other = group(2)
|
||||
}
|
||||
if kinds[group(9)] != KindNew || kinds[other] != KindTrunk || len(kinds) != 2 {
|
||||
t.Errorf("waiting = %v, want %s %s and %s %s", kinds, group(9), KindNew, other, KindTrunk)
|
||||
}
|
||||
}
|
||||
|
||||
func groups(cs []*Case) []string {
|
||||
out := make([]string, len(cs))
|
||||
for i, c := range cs {
|
||||
|
||||
@@ -8,7 +8,6 @@ import (
|
||||
"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
|
||||
@@ -34,12 +33,10 @@ type StateRead struct {
|
||||
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
|
||||
PreviewState StateRead // the checker's merge previews
|
||||
StepStates map[string]StateRead // the build modules' steps states, by module
|
||||
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
|
||||
// Trunks are the trunk heads of every repository (the gate's source); nil names no re-judge after a trunk move.
|
||||
Trunks func() (map[string]string, error)
|
||||
Now func() time.Time
|
||||
@@ -266,18 +263,10 @@ type LeftStep struct {
|
||||
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.
|
||||
// Status is what runs now, read from the buckets: the steps in flight on the build modules and the candidates without a
|
||||
// verdict for their heads. Nothing holds a check since R4g: a check is an ask on the build seats' streams, whose backlog
|
||||
// the delivery view shows (ADR 0525 §9).
|
||||
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"`
|
||||
@@ -288,24 +277,9 @@ type Status struct {
|
||||
Notes []string `json:"notes,omitempty"`
|
||||
}
|
||||
|
||||
// Status reads the capacity, the plan bucket, the build modules' steps and the candidates against their verdicts.
|
||||
// Status reads 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
|
||||
}
|
||||
st := Status{RunningByNode: map[string]int{}}
|
||||
mods := make([]string, 0, len(r.StepStates))
|
||||
for m := range r.StepStates {
|
||||
mods = append(mods, m)
|
||||
|
||||
@@ -3,7 +3,6 @@ package check
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -129,16 +128,14 @@ func TestThePreviewsToolShowsTheStoredPreviewOfAPullRequest(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
// 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) {
|
||||
// checker_status reads the steps in flight by node and the candidates without a verdict; since R4g nothing holds the checker
|
||||
// and it has no capacity of its own (ADR 0525 §4, §7; 1685 closed by the hold's deletion).
|
||||
func TestTheStatusToolReadsRunningSteps(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(),
|
||||
@@ -149,21 +146,15 @@ func TestTheStatusToolReadsCapacityHoldsAndRunningSteps(t *testing.T) {
|
||||
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 {
|
||||
// What cannot be read is said, not guessed.
|
||||
st, err = Reader{}.Status()
|
||||
if err != nil || len(st.Notes) == 0 {
|
||||
t.Errorf("empty status = %+v, %v", st, err)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -206,13 +206,11 @@ type TryOptions struct {
|
||||
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).
|
||||
// Tryer runs the tries: it takes a queued try, judges it through the same Judge as a pull request's check (a check ask, ADR
|
||||
// 0525 §1), and writes its end. It holds no timer: a try starts when its key is written.
|
||||
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)
|
||||
@@ -250,7 +248,6 @@ func (tr *Tryer) Put(ctx context.Context, key string, value []byte) {
|
||||
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{}
|
||||
@@ -259,11 +256,6 @@ func (tr *Tryer) Put(ctx context.Context, key string, value []byte) {
|
||||
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()
|
||||
|
||||
@@ -156,7 +156,7 @@ func TestAFailingTryNamesWhatFailedAndAStepThatCannotRunIsAnError(t *testing.T)
|
||||
// No build node is no error since R4g: the try's ask waits in the check stream until a node pulls it (ADR 0525 §4).
|
||||
}
|
||||
|
||||
func TestATryWaitsWhileADeliveryBuildsAndATryLeftRunningIsTakenOver(t *testing.T) {
|
||||
func TestATryLeftRunningIsTakenOver(t *testing.T) {
|
||||
ctx := context.Background()
|
||||
kv := newMemKV()
|
||||
id, _ := AskTry(memTries{kv}, tryTrunks, TryAsk{Repository: repoA, Head: head1}, time.Now())
|
||||
@@ -169,14 +169,7 @@ func TestATryWaitsWhileADeliveryBuildsAndATryLeftRunningIsTakenOver(t *testing.T
|
||||
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 := &Tryer{Tries: kv, Judge: b.ex}
|
||||
tr.Reconsider(ctx)
|
||||
tr.Wait()
|
||||
if got := readTry(t, kv, id); got.State != TryPass {
|
||||
|
||||
@@ -46,6 +46,7 @@ type manifest struct {
|
||||
Consumes []string `json:"consumes"`
|
||||
State []state `json:"state"`
|
||||
Reads []string `json:"reads"`
|
||||
Uses []string `json:"uses"`
|
||||
Invokes []string `json:"invokes"`
|
||||
Config map[string]struct {
|
||||
Type string `json:"type"`
|
||||
@@ -96,12 +97,17 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
||||
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",
|
||||
"mesh-controller.plans-building", "mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-builder.builds", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates",
|
||||
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}; !reflect.DeepEqual(sorted(m.Reads), want) {
|
||||
t.Errorf("reads = %v, want %v", m.Reads, want)
|
||||
}
|
||||
if want := []string{"gitea.gitea_commits", "gitea.gitea_compare", "gitea.gitea_get_file", "mesh-build-go.build_go_digest", "mesh-build-go.build_go_fetch", "mesh-build-go.build_go_test", "mesh-build-node.build_node_fetch", "mesh-build-node.build_node_test"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
|
||||
t.Errorf("invokes = %v, want only the build modules' fetch, digest and test (no run step before issue 1309, ADR 0484 clause 5) and the forge's two reads for a mesh-sdk pin (clause 9)", m.Invokes)
|
||||
// 1887/A5b: the checker reads nothing of the controller's plans-building (ADR 0483 superseded by ADR 0525 §7), calls no
|
||||
// build module's tool and holds no capacity: it puts check asks on the Go and Node build seats (ADR 0525 §1, §4).
|
||||
if want := []string{"gitea.gitea_commits", "gitea.gitea_compare", "gitea.gitea_get_file"}; !reflect.DeepEqual(sorted(m.Invokes), want) {
|
||||
t.Errorf("invokes = %v, want only the forge's reads for a mesh-sdk pin (ADR 0484 clause 9)", m.Invokes)
|
||||
}
|
||||
if want := []string{"node-build-go", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
|
||||
t.Errorf("uses = %v, want the Go and Node build seats", m.Uses)
|
||||
}
|
||||
// The checker is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it starts.
|
||||
if want := []string{"mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
|
||||
@@ -122,7 +128,7 @@ func TestManifestDeclarations(t *testing.T) { // 1678/AC1
|
||||
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"} {
|
||||
for _, k := range []string{"logs", "emits", "claims", "provides", "requires", "defines-seats", "seat", "seats", "bus", "config"} {
|
||||
if _, ok := top[k]; ok {
|
||||
t.Errorf("the manifest declares %q: the checker has none", k)
|
||||
}
|
||||
@@ -148,35 +154,19 @@ func TestManifestProcess(t *testing.T) { // 1678/AC1
|
||||
t.Errorf("process = %v", p)
|
||||
}
|
||||
env, _ := p["env"].(map[string]any)
|
||||
// The capacity of each node is the module's setting `capacity` (ADR 0484 clause 12, as the builder's, issue 1661), laid into the
|
||||
// settings file the program reads; empty is one step at a time everywhere. The environment holds no capacity.
|
||||
if len(env) != 2 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" || env["MESH_CHECKER_SETTINGS"] != "${dir:state}/settings.json" {
|
||||
t.Errorf("env = %v", p["env"])
|
||||
if len(env) != 1 || env["MESH_BROKER_FILE"] != "${dir:state}/broker" {
|
||||
t.Errorf("env = %v, want the broker file alone", p["env"])
|
||||
}
|
||||
if !reflect.DeepEqual(p["restart-on"], []any{"settings"}) {
|
||||
t.Errorf("restart-on = %v, want the settings file, so a changed capacity reaches the running program", p["restart-on"])
|
||||
}
|
||||
c, ok := m.Config["capacity"]
|
||||
if !ok || len(m.Config) != 1 || c.Type != "string" || c.Default != "" || c.Description == "" {
|
||||
t.Errorf("config = %+v, want one string setting, capacity, empty (one step at a time on every node) to start", m.Config)
|
||||
}
|
||||
var settings map[string]any
|
||||
for _, r := range m.Resources {
|
||||
if r["id"] == "settings" {
|
||||
settings = r
|
||||
t.Errorf("a settings resource %v: the checker has no capacity of its own (ADR 0525 §4)", r)
|
||||
}
|
||||
}
|
||||
if settings == nil || settings["type"] != "file" || settings["merge"] != "json" || !reflect.DeepEqual(settings["takes-config"], []any{"capacity"}) || settings["path"] != "${dir:state}/settings.json" {
|
||||
t.Errorf("settings resource = %v", settings)
|
||||
}
|
||||
if raw, err := os.ReadFile("files/settings.json"); err != nil || !strings.Contains(string(raw), `"capacity": ""`) {
|
||||
t.Errorf("files/settings.json = %q, %v, want the setting's starting value", raw, err)
|
||||
}
|
||||
if _, bus := p["bus"]; bus {
|
||||
t.Errorf("the process declares a bus: it runs on its own broker credential")
|
||||
}
|
||||
if len(m.Build.Artifacts) != 3 || m.Build.Artifacts[2].Name != "files" || m.Build.Artifacts[2].Kind != "archive" || m.Build.Artifacts[2].From != "files" {
|
||||
t.Fatalf("artifacts = %+v, want the program, the tools and the files", m.Build.Artifacts)
|
||||
if len(m.Build.Artifacts) != 2 {
|
||||
t.Fatalf("artifacts = %+v, want the program and the tools", m.Build.Artifacts)
|
||||
}
|
||||
if a := m.Build.Artifacts[1]; a.Name != "tools" || a.Kind != "bundle" || a.Language != "go" || a.Binary != "mesh-delivery-checker-tools" ||
|
||||
!reflect.DeepEqual(a.Loads, []string{"mesh-delivery-checker-tools"}) {
|
||||
|
||||
@@ -21,7 +21,6 @@
|
||||
"mesh-build-go.toolchains",
|
||||
"mesh-build-node.steps",
|
||||
"mesh-build-node.toolchains",
|
||||
"mesh-controller.plans-building",
|
||||
"mesh-delivery-builder.builds",
|
||||
"mesh-delivery-gate.source",
|
||||
"mesh-delivery-indexer.candidates",
|
||||
@@ -30,6 +29,10 @@
|
||||
"mesh-delivery-watchman.presence",
|
||||
"node-engine.have"
|
||||
],
|
||||
"uses": [
|
||||
"node-build-go",
|
||||
"node-build-node"
|
||||
],
|
||||
"tools": [
|
||||
"checker_kept",
|
||||
"checker_previews",
|
||||
@@ -42,21 +45,9 @@
|
||||
"invokes": [
|
||||
"gitea.gitea_commits",
|
||||
"gitea.gitea_compare",
|
||||
"gitea.gitea_get_file",
|
||||
"mesh-build-go.build_go_digest",
|
||||
"mesh-build-go.build_go_fetch",
|
||||
"mesh-build-go.build_go_test",
|
||||
"mesh-build-node.build_node_fetch",
|
||||
"mesh-build-node.build_node_test"
|
||||
"gitea.gitea_get_file"
|
||||
],
|
||||
"config": {
|
||||
"capacity": {
|
||||
"type": "string",
|
||||
"description": "How many build steps the checker runs at once on each node, as node=steps pairs separated by commas, such as shanks=2,ace=1 (ADR 0484 clause 12); a node set to 0 never checks, a node not named runs one at a time, and empty is one everywhere. The builder's capacity is the builder's own.",
|
||||
"pattern": "|[a-z0-9_-]+=[0-9]+(,[a-z0-9_-]+=[0-9]+)*",
|
||||
"default": ""
|
||||
}
|
||||
},
|
||||
|
||||
"own-secrets": {
|
||||
"broker": "${dir:state}/broker"
|
||||
},
|
||||
@@ -77,17 +68,6 @@
|
||||
"place": ".",
|
||||
"unknown": "program"
|
||||
},
|
||||
{
|
||||
"id": "settings",
|
||||
"trusted": false,
|
||||
"type": "file",
|
||||
"path": "${dir:state}/settings.json",
|
||||
"mode": "0600",
|
||||
"owner": "mesh-delivery-checker",
|
||||
"takes-config": ["capacity"],
|
||||
"merge": "json",
|
||||
"template": "files/settings.json"
|
||||
},
|
||||
{
|
||||
"id": "checker",
|
||||
"type": "process",
|
||||
@@ -97,12 +77,8 @@
|
||||
"./mesh-delivery-checker"
|
||||
],
|
||||
"env": {
|
||||
"MESH_BROKER_FILE": "${dir:state}/broker",
|
||||
"MESH_CHECKER_SETTINGS": "${dir:state}/settings.json"
|
||||
"MESH_BROKER_FILE": "${dir:state}/broker"
|
||||
},
|
||||
"restart-on": [
|
||||
"settings"
|
||||
],
|
||||
"user": "mesh-delivery-checker",
|
||||
"health": {
|
||||
"kind": "unit"
|
||||
@@ -129,11 +105,6 @@
|
||||
"loads": [
|
||||
"mesh-delivery-checker-tools"
|
||||
]
|
||||
},
|
||||
{
|
||||
"name": "files",
|
||||
"kind": "archive",
|
||||
"from": "files"
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user