diff --git a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go index 8e5ec44..4422450 100644 --- a/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go +++ b/modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go @@ -135,7 +135,8 @@ func TryTools(tr Trying) []stdio.Tool { }}, {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.", + "the head and the trunk commit it was merged onto, the time it took and what failed first; while it runs, each ask it put as soon as it is put, " + + "with the node that took it, its state and the phase its step says (fetch and merge, a level's digest, lint, tests) and since when. 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-"}}}, Run: func(args map[string]any) (any, error) { diff --git a/modules/mesh-delivery-checker/go.mod b/modules/mesh-delivery-checker/go.mod index 4e2c71e..cc0ac5a 100644 --- a/modules/mesh-delivery-checker/go.mod +++ b/modules/mesh-delivery-checker/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-checker go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026185254-02f50bd9d642 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-checker/go.sum b/modules/mesh-delivery-checker/go.sum index e793069..9e114f3 100644 --- a/modules/mesh-delivery-checker/go.sum +++ b/modules/mesh-delivery-checker/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026185254-02f50bd9d642 h1:mSBxF9KgIJyNL0gOKYq5naWewUj2yazU/tGOh12Tze0= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026185254-02f50bd9d642/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= diff --git a/modules/mesh-delivery-checker/internal/check/case.go b/modules/mesh-delivery-checker/internal/check/case.go index 169d55e..8670f7c 100644 --- a/modules/mesh-delivery-checker/internal/check/case.go +++ b/modules/mesh-delivery-checker/internal/check/case.go @@ -46,6 +46,11 @@ type Case struct { notes []string // asks are the ids of the asks the check put on the build seats' streams, in order: a try names them (Try.Asks). asks []string + // Watch, when set, hears each ask the check puts, as it is put (step nil), and each value its step key takes while + // the check waits for it: the node that took it, the phase it says, its end. A try writes them into its record, so a + // running try names its asks and where each stands (issue 1890). Called from the check's goroutines; it must not block + // for long. + Watch func(askID string, step *delivery.Step) // Progress says what the check is doing, so a check that ran past its bound names the step it was stuck at. Progress *Progress // Kind is what asked the check (KindNew, KindNodes, KindTrunk): a carry after a trunk or sibling move is asked on the diff --git a/modules/mesh-delivery-checker/internal/check/judge.go b/modules/mesh-delivery-checker/internal/check/judge.go index f493289..ec2f1c5 100644 --- a/modules/mesh-delivery-checker/internal/check/judge.go +++ b/modules/mesh-delivery-checker/internal/check/judge.go @@ -427,8 +427,14 @@ func (st *state) askNeeding(ctx context.Context, t *target, merged *[]Fetched, l if err := st.j.Asks.Put(ctx, a); err != nil { return fail(t.module, fmt.Sprintf("the check of %s#%d could not be asked: %v", p.Repository, p.Number, err)) } + waiting := ctx + if watch := st.c.Watch; watch != nil { + id := a.ID + watch(id, nil) // put: a try names it at once, not only when it ends (issue 1890) + waiting = buildrun.WithStepSeen(ctx, func(s delivery.Step) { watch(id, &s) }) + } var err error - step, err = st.j.Asks.Wait(ctx, a) + step, err = st.j.Asks.Wait(waiting, a) if err != nil { if ctx.Err() != nil { return fail(t.module, ctx.Err().Error()) diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go index af0b409..a44b3b4 100644 --- a/modules/mesh-delivery-checker/internal/check/try.go +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "regexp" + "slices" "sort" "strconv" "strings" @@ -79,12 +80,92 @@ type Try struct { Machines []string `json:"machines,omitempty"` // Asks are the ids of the asks the try put on the build seats' streams, each ending on the step key . of its // build module: the delivery view counts those steps as a try's, not a candidate's check (build_usage). - Asks []string `json:"asks,omitempty"` + Asks []string `json:"asks,omitempty"` + // Steps say where each ask stands while the try runs and after: asked until a node takes it, then the node, its + // state, the phase its step says and when (issue 1890: a 31-minute try could not be told queued from running). + Steps []TryStep `json:"steps,omitempty"` Asked time.Time `json:"asked"` Started time.Time `json:"started,omitzero"` Ended time.Time `json:"ended,omitzero"` } +// TryAsked is a TryStep's state before a build node took its ask: it waits in the seat's stream. +const TryAsked = "asked" + +// TryStep is one ask of a try as its step key last said it. +type TryStep struct { + Ask string `json:"ask"` + Node string `json:"node,omitempty"` + State string `json:"state"` + Phase string `json:"phase,omitempty"` + Started time.Time `json:"started,omitzero"` + PhaseAt time.Time `json:"phase-at,omitzero"` + Ended time.Time `json:"ended,omitzero"` +} + +// tryWriter writes a running try's record again as its asks are put and their steps move: one writer per try, holding +// the revision it wrote last, so the try's end is written at that revision. A write refused (the try ended by hand) +// stops the writes; the end's own write then says it. +type tryWriter struct { + mu sync.Mutex + tr *Tryer + ctx context.Context + key string + t Try + rev uint64 + stopped bool +} + +// heard takes an ask put (step nil) or a value of its step key in, and writes the record when it changed. +func (w *tryWriter) heard(askID string, step *delivery.Step) { + w.mu.Lock() + defer w.mu.Unlock() + if w.stopped { + return + } + changed := false + if !slices.Contains(w.t.Asks, askID) { + w.t.Asks, changed = append(w.t.Asks, askID), true + } + i := slices.IndexFunc(w.t.Steps, func(s TryStep) bool { return s.Ask == askID }) + if i < 0 { + w.t.Steps, changed = append(w.t.Steps, TryStep{Ask: askID, State: TryAsked}), true + i = len(w.t.Steps) - 1 + } + if step != nil { + now := TryStep{Ask: askID, Node: step.Node, State: step.State, Phase: step.Phase, Started: step.Started, + PhaseAt: step.PhaseAt, Ended: step.Ended} + changed = changed || now != w.t.Steps[i] + w.t.Steps[i] = now + } + if changed { + w.write() + } +} + +// write writes the record at the revision written last; a refusal stops the writes. +func (w *tryWriter) write() { + data, err := encodeTry(w.t) + if err != nil { + return + } + rev, err := w.tr.Tries.Update(w.ctx, w.key, data, w.rev) + if err != nil { + w.stopped = true + w.tr.logf("%s: its asks' progress is no longer written: %v", w.key, err) + return + } + w.rev = rev +} + +// end stops the writes and answers the record and its revision, for the try's end. +func (w *tryWriter) end() (Try, uint64) { + w.mu.Lock() + defer w.mu.Unlock() + w.stopped = true + return w.t, w.rev +} + // Over says whether the try came to its end. func (t Try) Over() bool { return t.State == TryPass || t.State == TryFail || t.State == TryError } @@ -395,7 +476,8 @@ func (tr *Tryer) run(ctx context.Context, key string) { } return } - cs := &Case{Key: key, Group: key, + progress := &tryWriter{tr: tr, ctx: ctx, key: key, t: t, rev: rev} + cs := &Case{Key: key, Group: key, Watch: progress.heard, Pulls: []Pull{{Repository: t.Repository, Head: t.Head, Trunk: t.Trunk}}, Trunk: map[string]string{t.Repository: t.Trunk}, Queued: t.Asked, @@ -409,6 +491,7 @@ func (tr *Tryer) run(ctx context.Context, key string) { if ctx.Err() != nil { return // left running: the checker takes it over when it starts; or ended by hand, which wrote its end } + t, rev = progress.end() v := out.Verdict t.Ended = tr.now().UTC() t.Took = t.Ended.Sub(t.Started).Round(time.Second).String() diff --git a/modules/mesh-delivery-checker/internal/check/try_test.go b/modules/mesh-delivery-checker/internal/check/try_test.go index 27eb986..cee1241 100644 --- a/modules/mesh-delivery-checker/internal/check/try_test.go +++ b/modules/mesh-delivery-checker/internal/check/try_test.go @@ -23,6 +23,19 @@ func (m memTries) Create(key string, value []byte) error { return err } +// writtenTries is a memKV that keeps every running try it was given to write, in order. +type writtenTries struct { + *memKV + running []Try +} + +func (w *writtenTries) Update(ctx context.Context, key string, value []byte, rev uint64) (uint64, error) { + if t, err := DecodeTry(value); err == nil && !t.Over() { + w.running = append(w.running, t) + } + return w.memKV.Update(ctx, key, value, rev) +} + var tryTrunks = map[string]string{repoA: trunkA, repoSDK: trunkS, hqRepo: trunkA} func readTry(t testing.TB, kv *memKV, id string) Try { @@ -89,7 +102,8 @@ func TestATryIsJudgedAsAPullRequestsHeadAndOwesTheForgeNothing(t *testing.T) { / t.Fatal(err) } b := newBench(t) - tr := &Tryer{Tries: kv, Judge: b.ex} + written := &writtenTries{memKV: kv} + tr := &Tryer{Tries: written, Judge: b.ex} raw, _, _, _ := kv.Get(ctx, id) tr.Put(ctx, id, raw) tr.Wait() @@ -97,6 +111,15 @@ func TestATryIsJudgedAsAPullRequestsHeadAndOwesTheForgeNothing(t *testing.T) { / 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) } + // Issue 1890: while it runs, the try names its ask as soon as it is put, and then where its step stands; the end + // keeps each ask's node and state. + if len(written.running) < 3 || len(written.running[1].Asks) != 1 || written.running[1].Steps[0].State != TryAsked || + written.running[2].Steps[0].Node != "shanks" { + t.Errorf("the running try's writes = %+v, want its ask named while it ran, then its node", written.running) + } + if len(got.Steps) != 1 || got.Steps[0].Node != "shanks" || got.Steps[0].State != delivery.StepPassed { + t.Errorf("the ended try's steps = %+v", got.Steps) + } // The try names the asks it put, so the delivery view counts their steps as a try's (build_usage). if len(got.Asks) != 1 { t.Errorf("asks = %v, want the one check ask the try put", got.Asks) diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index ae85fd6..c44cc9b 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-view go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026125731-2aae6f5580f2 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026185254-02f50bd9d642 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-view/go.sum b/modules/mesh-delivery-view/go.sum index f754077..9e114f3 100644 --- a/modules/mesh-delivery-view/go.sum +++ b/modules/mesh-delivery-view/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026125731-2aae6f5580f2 h1:s1WcG7Ll1OkEwwvMKEabdWvPvTiFCHghlXiY1LoIJDA= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026125731-2aae6f5580f2/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026185254-02f50bd9d642 h1:mSBxF9KgIJyNL0gOKYq5naWewUj2yazU/tGOh12Tze0= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026185254-02f50bd9d642/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= diff --git a/modules/mesh-delivery-view/internal/view/usage.go b/modules/mesh-delivery-view/internal/view/usage.go index 19d43ea..7849d80 100644 --- a/modules/mesh-delivery-view/internal/view/usage.go +++ b/modules/mesh-delivery-view/internal/view/usage.go @@ -67,7 +67,8 @@ type UsageAnswer struct { At time.Time `json:"at"` Nodes []NodeUsage `json:"nodes"` Backlog []BacklogRow `json:"backlog"` - // TriesRunning are the tries the checker runs now; their asks are counted under check until the try ends and names them. + // TriesRunning are the tries the checker runs now, each with its asks and where each stands (a checker that names its + // asks as they are put, issue 1890; an older one names them only at the end, and they are counted under check until then). TriesRunning []TryRunning `json:"tries_running"` Notes []string `json:"notes,omitempty"` } @@ -77,6 +78,9 @@ type TryRunning struct { ID string `json:"id"` Repository string `json:"repository"` Started time.Time `json:"started,omitzero"` + Asks []string `json:"asks,omitempty"` + // Steps are the try's own words for each ask: asked, or the node that took it, its state and the phase its step says. + Steps json.RawMessage `json:"steps,omitempty"` } // NodeUsage is one node: each build seat it holds or is set to build for, whether it is eligible to build, and its CPU. @@ -101,7 +105,11 @@ type SeatUsage struct { // Capacity is the asks it runs at once; CapacityFrom where that came from. Capacity int `json:"capacity"` CapacityFrom string `json:"capacity_from"` - Paused bool `json:"paused"` + // Configured is mesh-config's capacity when the module says it judges by a lower one (capacity. of its pauses + // state: the configured capacity bounded by its node's build CPU, issue 1806's follow-up); Capacity is then the + // module's and CapacityFrom its why. + Configured int `json:"configured,omitempty"` + Paused bool `json:"paused"` // Eligible says it takes asks: it holds the seat, has a capacity above 0 and is not paused; WhyNot says why not. // EligibleKnown is false when this tool cannot tell (it cannot read mesh-config's capacities): Eligible is then // absent, never false, and WhyNot starts "not known: ". @@ -384,8 +392,9 @@ func (u UsageReader) Usage() UsageAnswer { } } - // Pauses: paused. of each module. + // Pauses: paused. of each module; and capacity., the capacity the module says it judges by. paused := map[string]bool{} // . + judged := map[string]delivery.Capacity{} for _, lang := range delivery.AskLanguages { module, _ := delivery.BuildModuleOf(lang) pr, ok := u.Pauses[lang] @@ -411,6 +420,13 @@ func (u UsageReader) Usage() UsageAnswer { paused[node+"."+lang] = p.Paused } } + if key, err := delivery.CapacityKey(node); err == nil { + if raw, err := pr.Get(key); err == nil && len(raw) > 0 { + if c, err := delivery.DecodeCapacity(raw); err == nil { + judged[node+"."+lang] = c + } + } + } } } @@ -452,6 +468,11 @@ func (u UsageReader) Usage() UsageAnswer { s.Capacity = (&buildrun.Places{Capacity: caps}).CapacityOf(node) s.CapacityFrom = fmt.Sprintf("the modules' default %q: mesh-config holds no resolved value for %s on %s", buildstep.DefaultCapacities, s.Module, node) } + if c, ok := judged[node+"."+lang]; ok && s.Holds && unknown == "" && !isBad && c.Configured == s.Capacity && c.Effective != c.Configured { + // The module running on the node judges by less than mesh-config gives: its node's build CPU bounds it. + s.Configured, s.Capacity = s.Capacity, c.Effective + s.CapacityFrom = fmt.Sprintf("%s, as %s on %s judges it (said %s)", c.Why, s.Module, node, c.At.UTC().Format(time.RFC3339)) + } s.Paused = paused[node+"."+lang] s.EligibleKnown = true switch { @@ -517,6 +538,7 @@ func (u UsageReader) tries(now time.Time, ans *UsageAnswer, note func(string, .. want = append(want, k) } } + unnamed := 0 raws, err := u.Tries.values(want) if err != nil { note("the checker's tries cannot all be read: %v", err) @@ -526,11 +548,12 @@ func (u UsageReader) tries(now time.Time, ans *UsageAnswer, note func(string, .. continue } var t struct { - ID string `json:"id"` - Repository string `json:"repository"` - State string `json:"state"` - Started time.Time `json:"started"` - Asks []string `json:"asks"` + ID string `json:"id"` + Repository string `json:"repository"` + State string `json:"state"` + Started time.Time `json:"started"` + Asks []string `json:"asks"` + Steps json.RawMessage `json:"steps"` } if json.Unmarshal(raws[i], &t) != nil { continue @@ -539,11 +562,14 @@ func (u UsageReader) tries(now time.Time, ans *UsageAnswer, note func(string, .. asks[a] = true } if t.State == "running" { - ans.TriesRunning = append(ans.TriesRunning, TryRunning{ID: t.ID, Repository: t.Repository, Started: t.Started}) + ans.TriesRunning = append(ans.TriesRunning, TryRunning{ID: t.ID, Repository: t.Repository, Started: t.Started, Asks: t.Asks, Steps: t.Steps}) + if len(t.Asks) == 0 { + unnamed++ + } } } - if len(ans.TriesRunning) > 0 { - note("a running try's asks are counted under check until the try ends and names them") + if unnamed > 0 { + note("%d running tries name no ask yet (none put yet, or a checker that names them only at the end): their asks are counted under check until then", unnamed) } return asks } diff --git a/modules/mesh-delivery-view/internal/view/usage_test.go b/modules/mesh-delivery-view/internal/view/usage_test.go index 580018a..78cab03 100644 --- a/modules/mesh-delivery-view/internal/view/usage_test.go +++ b/modules/mesh-delivery-view/internal/view/usage_test.go @@ -191,4 +191,25 @@ func TestBuildUsageAnswersEveryNodesUseFromTheStates(t *testing.T) { if verdict["shanks"] != "not known" || verdict["ace"] != "not eligible" { t.Errorf("verdicts with mesh-config unreadable = %v, want shanks not known, ace not eligible (paused)", verdict) } + + // Issue 1806's follow-up: the module on shanks says it judges by 1 of mesh-config's 2 (its build CPU bounds it); the + // seat says the module's capacity and why. And issue 1890: a running try that names its asks is said with them. + u = usageFixture() + u.Pauses["go"] = states(map[string]string{"capacity.shanks": `{"v":1,"effective":1,"configured":2,"quota-from":"the cpu.max of /meshyield.slice","quota-cores":6,"cores-per-step":6,"by-cpu":1,"why":"1, bounded by the CPU","at":"2026-10-26T11:00:00Z"}`, + "capacity.ace": `{"v":1,"effective":2,"configured":3,"why":"stale: mesh-config says 2 now"}`}) + u.Tries = states(map[string]string{"try-" + strconv.FormatInt(at("11:58").UnixNano(), 10): `{"id":"try-2","repository":"novox/mesh-delivery","head":"h","trunk":"t","state":"running","asks":["` + tryAsk0 + `"],"steps":[{"ask":"` + tryAsk0 + `","node":"shanks","state":"running","phase":"unit: tests of 3 packages"}]}`}) + bound := u.Usage() + sc := seatOf(t, bound, "shanks", "node-build-go") + if sc.Capacity != 1 || sc.Configured != 2 || !strings.HasPrefix(sc.CapacityFrom, "1, bounded by the CPU, as mesh-build-go on shanks judges it") { + t.Errorf("a seat its CPU bounds: capacity %d of %d, from %q", sc.Capacity, sc.Configured, sc.CapacityFrom) + } + if sa := seatOf(t, bound, "ace", "node-build-go"); sa.Capacity != 2 || sa.Configured != 0 { + t.Errorf("a module's word older than mesh-config's value is not taken: %+v", sa) + } + if len(bound.TriesRunning) != 1 || len(bound.TriesRunning[0].Asks) != 1 || !strings.Contains(string(bound.TriesRunning[0].Steps), "unit: tests") || + strings.Contains(strings.Join(bound.Notes, " | "), "name no ask yet") { + t.Errorf("a running try naming its asks: %+v, notes %v", bound.TriesRunning, bound.Notes) + } } + +var tryAsk0 = stepID(at("11:58"), "a00010")