Make checker_status count only steps that can still run, so a key a build module left running is not read as a full node (issue 1890)
mesh/delivery delivered
mesh/check pass e76202876032 on b714eab995a5 | unit 422 run, 0 kept | integration 467 run, 0 kept | over bound: integration (slowest integration mesh-…
mesh/delivery delivered
mesh/check pass e76202876032 on b714eab995a5 | unit 422 run, 0 kept | integration 467 run, 0 kept | over bound: integration (slowest integration mesh-…
The checker's own places are released when a step ends or its wait passes the step bound; checker_status read every 'running' key in the build modules' steps buckets as a step in flight, so keys left by stopped, restarted or unassigned build modules looked like full nodes. It now lists them under left_running with why (past the bound, module not on the node, module not running), and leaves hq-only groups out of the unjudged candidates.
This commit is contained in:
@@ -145,8 +145,9 @@ func Tools(r check.Reader) []stdio.Tool {
|
||||
{Name: check.StatusReadTool,
|
||||
Description: "What holds the checker back and what runs: the capacity per node (the setting, the default one step per node), whether the controller's " +
|
||||
"delivery plans in their build phase hold a key (no new check starts while one does), the steps in flight on the build modules by node, and the " +
|
||||
"candidates without a verdict for their heads. The steps the checker holds waiting for a place are in its process, not in a bucket, and the " +
|
||||
"answer says so. Read-only.",
|
||||
"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() }},
|
||||
}
|
||||
@@ -167,11 +168,20 @@ func capacity() (string, error) {
|
||||
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")),
|
||||
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}
|
||||
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{
|
||||
|
||||
@@ -40,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"
|
||||
)
|
||||
@@ -184,7 +181,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
|
||||
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
|
||||
}
|
||||
|
||||
@@ -723,3 +723,41 @@ func TestNoStepOfACheckCarriesPublish(t *testing.T) { // ADR 0500 §4, 1309 (the
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 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()
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery"
|
||||
"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
|
||||
@@ -37,6 +38,9 @@ type Reader struct {
|
||||
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
|
||||
@@ -254,6 +258,12 @@ type RunningStep struct {
|
||||
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.
|
||||
@@ -267,6 +277,7 @@ type Status struct {
|
||||
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"`
|
||||
@@ -295,7 +306,14 @@ func (r Reader) Status() (Status, error) {
|
||||
mods = append(mods, m)
|
||||
}
|
||||
sort.Strings(mods)
|
||||
st.Running = []RunningStep{}
|
||||
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 {
|
||||
@@ -320,10 +338,20 @@ func (r Reader) Status() (Status, error) {
|
||||
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()
|
||||
@@ -344,11 +372,56 @@ func (r Reader) Status() (Status, error) {
|
||||
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 steps the checker holds waiting for a place are in its process and in no bucket: a candidate without a verdict while a plan key exists, or while its node runs as many steps as its capacity, is held")
|
||||
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
|
||||
}
|
||||
|
||||
@@ -142,6 +142,8 @@ func TestTheStatusToolReadsCapacityHoldsAndRunningSteps(t *testing.T) {
|
||||
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 {
|
||||
@@ -165,3 +167,77 @@ func TestTheStatusToolReadsCapacityHoldsAndRunningSteps(t *testing.T) {
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user