Builder and checker read tools read only the keys they need, in parallel, at most 1000 a call (issue 1827)

builder_stuck, checker_kept, checker_verdicts, checker_status, checker_previews, checker_pull and checker_try listed a
bucket of thousands of keys and read each value (builder_stuck also its history) one round trip at a time before
filtering, so they timed out at 30 s. Now keys are narrowed by what the key carries first, history is read only for rows
kept, values are read sixteen at a time, at most 1000 a call with the answer saying it was cut, steps newest first, a new
try reads only the newest 64 tries, and each tool logs the values read and the time taken. Same kind as the view's fix
(PR 52); the helper is copied, not put in mesh-sdk, as a few lines do not earn a group delivery.

modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go

@@ -57,7 +57,13 @@ func read(s stdio.KeptState) build.StateRead {
 		if err != nil || e == nil {
 			return nil, time.Time{}, err
 		}
-		return e.Value, written(s, key, e.Revision), nil
+		return e.Value, time.Time{}, nil
+	}, Written: func(key string) time.Time {
+		e, err := s.Get(key)
+		if err != nil || e == nil {
+			return time.Time{}
+		}
+		return written(s, key, e.Revision)
 	}}
 }
 
@@ -81,7 +87,7 @@ func main() {
 		steps[bm] = read(stdio.State(bm + ".steps"))
 	}
 	reasks := stdio.State(build.ReAsksState)
-	r := build.StuckReader{Builds: read(stdio.State("builds")), ReAsks: read(reasks), Steps: steps}
+	r := build.StuckReader{Logf: func(f string, a ...any) { fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] "+f+"\n", a...) }, Builds: read(stdio.State("builds")), ReAsks: read(reasks), Steps: steps}
 	put := func(key string, value []byte) error {
 		_, err := reasks.Put(key, json.RawMessage(value))
 		return err

modules/mesh-delivery-builder/internal/build/run_test.go

@@ -8,6 +8,7 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/delivery/buildrun"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -807,6 +808,52 @@ func TestABuildAskWaitingInItsStreamIsNotBounded(t *testing.T) {
 			a[1].Asking != 1 || !strings.HasPrefix(a[1].Step, "running on g14") {
 			t.Fatalf("asks = %+v, want go at asking 2 with no step and image at asking 1 running on g14", a)
 		}
+
+		// 1827: on the live shape (3000 build keys over 100 modules, 1 ms a Get, three keys of each module building) a module reads only its
+		// own keys, the whole answer reads MaxReads values in parallel and says it was cut, and the history (Written) is asked
+		// only for the rows kept.
+		var gets, written, inFlight, maxFlight atomic.Int64
+		vals := map[string]string{}
+		for i := 0; i < 3000; i++ {
+			k, _ := delivery.BuilderBuildKey(fmt.Sprintf("mod%02d", i%100), fmt.Sprintf("src2:%064x", i))
+			vals[k] = string(finished)
+			if (i/100)%10 == 0 {
+				vals[k] = string(building)
+			}
+		}
+		slow := state(vals)
+		get := slow.Get
+		slow.Get = func(k string) (json.RawMessage, time.Time, error) {
+			gets.Add(1)
+			n := inFlight.Add(1)
+			for m := maxFlight.Load(); n > m && !maxFlight.CompareAndSwap(m, n); m = maxFlight.Load() {
+			}
+			time.Sleep(time.Millisecond)
+			inFlight.Add(-1)
+			raw, _, err := get(k)
+			return raw, time.Time{}, err
+		}
+		slow.Written = func(string) time.Time { written.Add(1); return at }
+		for _, row := range []struct {
+			module string
+			stuck  int
+			gets   int64
+			cut    bool
+		}{{"mod10", 3, 30, false}, {"", 100, MaxReads, true}} {
+			gets.Store(0)
+			written.Store(0)
+			out, err := StuckReader{Builds: slow, ReAsks: state(map[string]string{}), Steps: map[string]StateRead{}, Now: func() time.Time { return at }}.Stuck(row.module)
+			if err != nil {
+				t.Fatal(err)
+			}
+			_, cut := out["cut"]
+			if n := len(out["stuck"].([]Stuck)); n != row.stuck || cut != row.cut || written.Load() != int64(n) || gets.Load() != row.gets {
+				t.Errorf("%q: %d stuck, %d gets, %d written, cut %v; want %d stuck and gets, written for each, cut %v", row.module, n, gets.Load(), written.Load(), cut, row.stuck, row.cut)
+			}
+		}
+		if maxFlight.Load() < 2 {
+			t.Errorf("the values were read one at a time")
+		}
 	})
 }
 

modules/mesh-delivery-builder/internal/build/stuck.go

@@ -7,6 +7,8 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"time"
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -65,6 +67,9 @@ func NewReAsk(module, fp, why, by string, at time.Time) (string, []byte, error)
 type StateRead struct {
 	Keys func() ([]string, error)
 	Get  func(key string) (json.RawMessage, time.Time, error)
+	// Written is when the key's current value was written, from the state's history: one more round trip, so the tools ask it
+	// only for a row they keep, and only when Get did not say.
+	Written func(key string) time.Time
 }
 
 // StuckAsk is one ask a key waits on.
@@ -102,10 +107,56 @@ type StuckReader struct {
 	// Steps are the build modules' steps states, by build module.
 	Steps map[string]StateRead
 	Now   func() time.Time
+	// Logf, when set, gets the one line a call logs: the values it read and the time it took.
+	Logf func(format string, args ...any)
+}
+
+// ReadWorkers is how many values builder_stuck asks the bus for at once, and MaxReads how many it reads in a call (issue 1827:
+// every value is a round trip through the runtime, so one after the other a few thousand keys outlast the call's 30 s).
+// A copy of the delivery view's and the checker's: a few lines are not worth a group delivery through mesh-sdk.
+const (
+	ReadWorkers = 16
+	MaxReads    = 1000
+)
+
+// parallel runs read(i) for i in [0, n), at most ReadWorkers at a time, and answers the first error.
+func parallel(n int, read func(i int) error) error {
+	var (
+		next   atomic.Int64
+		failed atomic.Bool
+		once   sync.Once
+		first  error
+		wg     sync.WaitGroup
+	)
+	for w := 0; w < min(ReadWorkers, n); w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for !failed.Load() {
+				i := int(next.Add(1) - 1)
+				if i >= n {
+					return
+				}
+				if err := read(i); err != nil {
+					once.Do(func() { first = err })
+					failed.Store(true)
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	return first
 }
 
 // Stuck is every key building or queued with its asks, oldest first; with module, that module's only.
 func (r StuckReader) Stuck(module string) (map[string]any, error) {
+	started := time.Now()
+	reads := 0
+	defer func() {
+		if r.Logf != nil {
+			r.Logf("builder_stuck read %d values in %s", reads, time.Since(started).Round(time.Millisecond))
+		}
+	}()
 	keys, err := r.Builds.Keys()
 	if err != nil {
 		return nil, err
@@ -114,28 +165,64 @@ func (r StuckReader) Stuck(module string) (map[string]any, error) {
 	if r.Now != nil {
 		now = r.Now
 	}
-	stepKeys := map[string][]string{}
+	// Steps are found by the ask's id, which their key ends in: one index per build module, no value read until an ask names it.
+	stepKeys := map[string]map[string]string{}
 	for bm, s := range r.Steps {
 		ks, err := s.Keys()
 		if err != nil {
 			return nil, fmt.Errorf("the steps of %s: %w", bm, err)
 		}
-		stepKeys[bm] = ks
+		byID := make(map[string]string, len(ks))
+		for _, k := range ks {
+			if _, id, err := delivery.ParseStepKey(k); err == nil {
+				byID[id] = k
+			}
+		}
+		stepKeys[bm] = byID
 	}
-	out := []Stuck{}
+	// Narrowed by the key (the module) before any value is read; the state is in the value, so every matching key is read, in
+	// parallel, at most MaxReads of them.
+	type parsed struct{ key, mod, fp string }
+	var matching []parsed
 	for _, key := range keys {
 		mod, fp, err := delivery.ParseBuilderBuildKey(key)
 		if err != nil || (module != "" && mod != module) {
 			continue
 		}
-		raw, at, err := r.Builds.Get(key)
-		if err != nil || raw == nil {
+		matching = append(matching, parsed{key, mod, fp})
+	}
+	sort.Slice(matching, func(i, j int) bool { return matching[i].key < matching[j].key })
+	cut := len(matching) > MaxReads
+	total := len(matching)
+	if cut {
+		matching = matching[:MaxReads]
+	}
+	values := make([]json.RawMessage, len(matching))
+	ats := make([]time.Time, len(matching))
+	if err := parallel(len(matching), func(i int) error {
+		raw, at, err := r.Builds.Get(matching[i].key)
+		if err == nil {
+			values[i], ats[i] = raw, at
+		}
+		return nil
+	}); err != nil {
+		return nil, err
+	}
+	reads += len(matching)
+	out := []Stuck{}
+	for i, m := range matching {
+		key, mod, fp := m.key, m.mod, m.fp
+		raw, at := values[i], ats[i]
+		if raw == nil {
 			continue
 		}
 		v, err := delivery.DecodeBuilderBuild(raw)
 		if err != nil || (v.State != delivery.BuildBuilding && v.State != delivery.BuildQueued) {
 			continue
 		}
+		if at.IsZero() && r.Builds.Written != nil {
+			at = r.Builds.Written(key)
+		}
 		s := Stuck{Key: key, Module: mod, FP: fp, State: v.State, Step: v.Step, By: v.By, Since: at, Asks: []StuckAsk{}}
 		if !at.IsZero() {
 			s.Age = now().Sub(at).Round(time.Second).String()
@@ -147,10 +234,8 @@ func (r StuckReader) Stuck(module string) (map[string]any, error) {
 				a.Asked = t
 			}
 			if st, ok := r.Steps[bm]; ok {
-				for _, k := range stepKeys[bm] {
-					if !strings.HasSuffix(k, "."+id) {
-						continue
-					}
+				if k, ok := stepKeys[bm][id]; ok {
+					reads++
 					if raw, _, err := st.Get(k); err == nil && raw != nil {
 						if step, err := delivery.DecodeStep(raw); err == nil {
 							a.Step = fmt.Sprintf("%s on %s since %s", step.State, step.Node, step.Started.UTC().Format(time.RFC3339))
@@ -166,6 +251,7 @@ func (r StuckReader) Stuck(module string) (map[string]any, error) {
 		sort.Slice(s.Asks, func(i, j int) bool { return s.Asks[i].Language < s.Asks[j].Language })
 		s.Unasked = v.State == delivery.BuildBuilding && len(v.Asks) == 0
 		if r.ReAsks.Get != nil {
+			reads++
 			if raw, _, err := r.ReAsks.Get(key); err == nil && raw != nil {
 				if ra, err := DecodeReAsk(raw); err == nil {
 					s.ReAsk = &ra
@@ -180,8 +266,12 @@ func (r StuckReader) Stuck(module string) (map[string]any, error) {
 		}
 		return out[i].Key < out[j].Key
 	})
-	return map[string]any{"stuck": out, "count": len(out),
-		"delivery_counts": "not readable by the builder: the build seats' workers' information is granted to the delivery view alone (its backlog); an ask given up on is told by the builder at its check bound and asked again"}, nil
+	answer := map[string]any{"stuck": out, "count": len(out),
+		"delivery_counts": "not readable by the builder: the build seats' workers' information is granted to the delivery view alone (its backlog); an ask given up on is told by the builder at its check bound and asked again"}
+	if cut {
+		answer["cut"], answer["note"] = true, fmt.Sprintf("read %d of %d build keys in key order and stopped; give a module to read its keys alone", MaxReads, total)
+	}
+	return answer, nil
 }
 
 // StuckBuildModules are the build modules whose steps builder_stuck reads.

modules/mesh-delivery-checker/cmd/mesh-delivery-checker-tools/main.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"sync/atomic"
 	"time"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
@@ -15,8 +16,35 @@ import (
 	"mesh-delivery-checker/internal/check"
 )
 
+// reads counts the values this call read from the bus: the runtime launches the tools for one call, so the count is the call's.
+var reads atomic.Int64
+
+// cutAnswer says in the answer that it was cut, and how to narrow it.
+func cutAnswer(answer map[string]any, cut string) map[string]any {
+	if cut != "" {
+		answer["cut"], answer["note"] = true, cut
+	}
+	return answer
+}
+
+// logged runs every tool with one line to stderr per call: the tool, the values it read and the time it took.
+func logged(tools []stdio.Tool) []stdio.Tool {
+	for i := range tools {
+		name, run := tools[i].Name, tools[i].Run
+		tools[i].Run = func(args map[string]any) (any, error) {
+			reads.Store(0)
+			start := time.Now()
+			out, err := run(args)
+			fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %s read %d values in %s (error: %v)\n", name, reads.Load(), time.Since(start).Round(time.Millisecond), err)
+			return out, err
+		}
+	}
+	return tools
+}
+
 func read(s stdio.KeptState) check.StateRead {
 	return check.StateRead{Keys: s.Keys, Get: func(key string) (json.RawMessage, error) {
+		reads.Add(1)
 		e, err := s.Get(key)
 		if err != nil || e == nil {
 			return nil, err
@@ -100,8 +128,8 @@ func Tools(r check.Reader) []stdio.Tool {
 				"group": map[string]any{"type": "string", "description": "the delivery group's branch name"}}},
 			Run: func(args map[string]any) (any, error) {
 				group, _ := args["group"].(string)
-				vs, err := r.Verdicts(group)
-				return map[string]any{"verdicts": vs}, err
+				vs, cut, err := r.Verdicts(group)
+				return cutAnswer(map[string]any{"verdicts": vs}, cut), err
 			}},
 		{Name: check.PullReadTool,
 			Description: "The verdict that judges one pull request and the full report of its status mesh/check. Read-only.",
@@ -122,8 +150,8 @@ func Tools(r check.Reader) []stdio.Tool {
 			Run: func(args map[string]any) (any, error) {
 				level, _ := args["level"].(string)
 				digest, _ := args["digest"].(string)
-				kept, err := r.Kept(level, digest)
-				return map[string]any{"kept": kept}, err
+				kept, cut, err := r.Kept(level, digest)
+				return cutAnswer(map[string]any{"kept": kept}, cut), err
 			}},
 		{Name: check.PreviewsReadTool,
 			Description: "The merge previews the checker holds, each as stored under previews.<repository>.<head>.<trunk>: the merge commit, per module directory " +
@@ -138,8 +166,8 @@ func Tools(r check.Reader) []stdio.Tool {
 				if head != "" && len(head) < 7 {
 					return nil, fmt.Errorf("the head %q is too short: give at least 7 characters", head)
 				}
-				ps, bad, err := r.Previews(repository, head)
-				return map[string]any{"previews": ps, "unreadable": bad}, err
+				ps, bad, cut, err := r.Previews(repository, head)
+				return cutAnswer(map[string]any{"previews": ps, "unreadable": bad}, cut), err
 			}},
 		{Name: check.StatusReadTool,
 			Description: "What runs: the steps in flight on the build modules by node, the candidates without a verdict for their heads, " +
@@ -184,7 +212,7 @@ func main() {
 		}},
 		Trunks: trunks,
 	}
-	if err := stdio.Serve("", append(Tools(r), TryTools(trying)...)); err != nil {
+	if err := stdio.Serve("", logged(append(Tools(r), TryTools(trying)...))); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-checker-tools] %v\n", err)
 		os.Exit(1)
 	}

modules/mesh-delivery-checker/internal/check/bounded.go

@@ -0,0 +1,79 @@
+package check
+
+import (
+	"fmt"
+	"sort"
+	"sync"
+	"sync/atomic"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+)
+
+// ReadWorkers is how many values a read tool asks the bus for at once. Each value is a round trip from the tool through the
+// runtime to the bus, so reading a bucket of thousands of keys one after the other outlasts the 30 s a tool call is given
+// (issue 1827, as the delivery view's issue 1945). It is a copy of the view's: a few lines are not worth a group delivery
+// through mesh-sdk (rules-are-weighed).
+const ReadWorkers = 16
+
+// MaxReads bounds the values one call of a read tool reads; past it the answer says it was cut and how to narrow it.
+const MaxReads = 1000
+
+// parallel runs read(i) for i in [0, n), at most ReadWorkers at a time, and answers the first error; once one failed no
+// further read starts.
+func parallel(n int, read func(i int) error) error {
+	var (
+		next   atomic.Int64
+		failed atomic.Bool
+		once   sync.Once
+		first  error
+		wg     sync.WaitGroup
+	)
+	for w := 0; w < min(ReadWorkers, n); w++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for !failed.Load() {
+				i := int(next.Add(1) - 1)
+				if i >= n {
+					return
+				}
+				if err := read(i); err != nil {
+					once.Do(func() { first = err })
+					failed.Store(true)
+				}
+			}
+		}()
+	}
+	wg.Wait()
+	return first
+}
+
+// capKeys keeps the first MaxReads keys and answers a note when it cut the rest, "" otherwise.
+func capKeys(keys []string, narrow string) ([]string, string) {
+	if len(keys) <= MaxReads {
+		return keys, ""
+	}
+	return keys[:MaxReads], fmt.Sprintf("read %d of %d matching keys and stopped; %s", MaxReads, len(keys), narrow)
+}
+
+// newestFirst sorts step keys (<node>.<id>) by id, newest first: an id is the time in base 36, then six hex digits, so the
+// longer id is the later and equal lengths compare as text. A key that is no step key goes last.
+func newestFirst(keys []string) {
+	id := func(k string) (string, bool) {
+		_, id, err := delivery.ParseStepKey(k)
+		return id, err == nil
+	}
+	sort.Slice(keys, func(i, j int) bool {
+		a, aok := id(keys[i])
+		b, bok := id(keys[j])
+		switch {
+		case aok != bok:
+			return aok
+		case len(a) != len(b):
+			return len(a) > len(b)
+		case a != b:
+			return a > b
+		}
+		return keys[i] > keys[j]
+	})
+}

modules/mesh-delivery-checker/internal/check/read.go

@@ -89,37 +89,46 @@ func (r Reader) read(key string, raw json.RawMessage) (VerdictRead, error) {
 	return out, nil
 }
 
-// Verdicts are the verdicts the checker holds, by key; with a group only that group's.
-func (r Reader) Verdicts(group string) ([]VerdictRead, error) {
+// Verdicts are the verdicts the checker holds, by key; with a group only that group's. The keys are narrowed by the group
+// before any value is read, at most MaxReads are read (sixteen at a time), and cut says when it stopped there.
+func (r Reader) Verdicts(group string) (out []VerdictRead, cut string, err error) {
 	if r.VerdictState.Keys == nil || r.VerdictState.Get == nil {
-		return nil, fmt.Errorf("the verdicts state cannot be read here")
+		return nil, "", fmt.Errorf("the verdicts state cannot be read here")
 	}
 	keys, err := r.VerdictState.Keys()
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 	sort.Strings(keys)
-	var out []VerdictRead
+	var matching []string
 	for _, k := range keys {
 		if group != "" {
 			if g, _, err := delivery.ParseCheckerVerdictKey(k); err != nil || g != group {
 				continue
 			}
 		}
-		raw, err := r.VerdictState.Get(k)
-		if err != nil {
-			return nil, err
-		}
-		if len(raw) == 0 {
-			continue
-		}
-		vr, err := r.read(k, raw)
-		if err != nil {
-			return nil, err
-		}
-		out = append(out, vr)
+		matching = append(matching, k)
 	}
-	return out, nil
+	matching, cut = capKeys(matching, "give a group to narrow the answer")
+	rows := make([]VerdictRead, len(matching))
+	found := make([]bool, len(matching))
+	if err := parallel(len(matching), func(i int) error {
+		raw, err := r.VerdictState.Get(matching[i])
+		if err != nil || len(raw) == 0 {
+			return err
+		}
+		rows[i], err = r.read(matching[i], raw)
+		found[i] = err == nil
+		return err
+	}); err != nil {
+		return nil, "", err
+	}
+	for i, row := range rows {
+		if found[i] {
+			out = append(out, row)
+		}
+	}
+	return out, cut, nil
 }
 
 // Pull answers for one pull request: the verdict of the candidate that holds it, with its report; for one the indexer holds
@@ -133,74 +142,95 @@ func (r Reader) Pull(repository string, number int) (VerdictRead, error) {
 		return VerdictRead{}, err
 	}
 	sort.Strings(keys)
-	for _, k := range keys {
-		raw, err := r.CandidateState.Get(k)
-		if err != nil || len(raw) == 0 {
-			continue
-		}
-		c, err := delivery.DecodeCandidate(raw)
-		if err != nil {
-			continue
-		}
-		for _, p := range c.Pulls {
-			if p.Repository != repository || p.Number != number {
+	keys, cut := capKeys(keys, "")
+	// The candidates are read in batches, each in parallel and taken in key order, so the read stops at the batch that holds it.
+	for from := 0; from < len(keys); from += ReadWorkers * 4 {
+		batch := keys[from:min(from+ReadWorkers*4, len(keys))]
+		cands := make([]delivery.Candidate, len(batch))
+		ok := make([]bool, len(batch))
+		_ = parallel(len(batch), func(i int) error {
+			raw, err := r.CandidateState.Get(batch[i])
+			if err != nil || len(raw) == 0 {
+				return nil
+			}
+			c, err := delivery.DecodeCandidate(raw)
+			cands[i], ok[i] = c, err == nil
+			return nil
+		})
+		for i, c := range cands {
+			if !ok[i] {
 				continue
 			}
-			vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
-			if err != nil {
-				return VerdictRead{}, err
+			for _, p := range c.Pulls {
+				if p.Repository != repository || p.Number != number {
+					continue
+				}
+				vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
+				if err != nil {
+					return VerdictRead{}, err
+				}
+				vraw, err := r.VerdictState.Get(vkey)
+				if err != nil {
+					return VerdictRead{}, err
+				}
+				if len(vraw) == 0 {
+					return VerdictRead{Key: vkey, Note: "no verdict yet for the heads of its candidate"}, nil
+				}
+				return r.read(vkey, vraw)
 			}
-			vraw, err := r.VerdictState.Get(vkey)
-			if err != nil {
-				return VerdictRead{}, err
-			}
-			if len(vraw) == 0 {
-				return VerdictRead{Key: vkey, Note: "no verdict yet for the heads of its candidate"}, nil
-			}
-			return r.read(vkey, vraw)
 		}
 	}
+	if cut != "" {
+		return VerdictRead{}, fmt.Errorf("no candidate among the first %d holds %s#%d", MaxReads, repository, number)
+	}
 	return VerdictRead{}, fmt.Errorf("no candidate holds %s#%d", repository, number)
 }
 
 // Kept is the kept verdicts: one key's value as stored, or with an empty key the keys of a level (or every level) and their values.
-func (r Reader) Kept(level, digest string) (map[string]json.RawMessage, error) {
+func (r Reader) Kept(level, digest string) (map[string]json.RawMessage, string, error) {
 	if r.KeptState.Keys == nil || r.KeptState.Get == nil {
-		return nil, fmt.Errorf("the kept state cannot be read here")
+		return nil, "", fmt.Errorf("the kept state cannot be read here")
 	}
 	if digest != "" {
 		key, err := delivery.KeptKey(level, digest)
 		if err != nil {
-			return nil, err
+			return nil, "", err
 		}
 		raw, err := r.KeptState.Get(key)
 		if err != nil {
-			return nil, err
+			return nil, "", err
 		}
 		if len(raw) == 0 {
-			return nil, fmt.Errorf("no kept verdict %s", key)
+			return nil, "", fmt.Errorf("no kept verdict %s", key)
 		}
-		return map[string]json.RawMessage{key: raw}, nil
+		return map[string]json.RawMessage{key: raw}, "", nil
 	}
 	keys, err := r.KeptState.Keys()
 	if err != nil {
-		return nil, err
+		return nil, "", err
 	}
 	sort.Strings(keys)
-	out := map[string]json.RawMessage{}
+	var matching []string
 	for _, k := range keys {
-		if l, _, err := delivery.ParseKeptKey(k); err != nil || (level != "" && l != level) {
-			continue
-		}
-		raw, err := r.KeptState.Get(k)
-		if err != nil {
-			return nil, err
-		}
-		if len(raw) > 0 {
-			out[k] = raw
+		if l, _, err := delivery.ParseKeptKey(k); err == nil && (level == "" || l == level) {
+			matching = append(matching, k)
 		}
 	}
-	return out, nil
+	matching, cut := capKeys(matching, "give a level and a digest to read one")
+	values := make([]json.RawMessage, len(matching))
+	if err := parallel(len(matching), func(i int) (err error) {
+		values[i], err = r.KeptState.Get(matching[i])
+		return err
+	}); err != nil {
+		return nil, "", err
+	}
+	out := map[string]json.RawMessage{}
+	for i, k := range matching {
+		if len(values[i]) > 0 {
+			out[k] = values[i]
+		}
+	}
+	return out, cut, nil
 }
 
 // PreviewRead is one preview as stored, with the parts of its key.
@@ -214,17 +244,18 @@ type PreviewRead struct {
 
 // Previews are the previews the checker holds, each as stored. With a repository only that repository's; with a head (a
 // prefix of at least 7 characters is enough) only previews of that head. A key that is no preview key is listed as unreadable.
-func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, error) {
+func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, string, error) {
 	if r.PreviewState.Keys == nil || r.PreviewState.Get == nil {
-		return nil, nil, fmt.Errorf("the previews state cannot be read here")
+		return nil, nil, "", fmt.Errorf("the previews state cannot be read here")
 	}
 	keys, err := r.PreviewState.Keys()
 	if err != nil {
-		return nil, nil, err
+		return nil, nil, "", err
 	}
 	sort.Strings(keys)
-	out := []PreviewRead{}
-	var bad []string
+	type part struct{ repo, head, trunk string }
+	parts := map[string]part{}
+	var matching, bad []string
 	for _, k := range keys {
 		repo, h, trunk, err := delivery.ParsePreviewKey(k)
 		if err != nil {
@@ -234,15 +265,25 @@ func (r Reader) Previews(repository, head string) ([]PreviewRead, []string, erro
 		if (repository != "" && repo != repository) || (head != "" && !strings.HasPrefix(h, head)) {
 			continue
 		}
-		raw, err := r.PreviewState.Get(k)
-		if err != nil {
-			return nil, nil, err
-		}
-		if len(raw) > 0 {
-			out = append(out, PreviewRead{Key: k, Repository: repo, Head: h, Trunk: trunk, Preview: raw})
+		parts[k] = part{repo, h, trunk}
+		matching = append(matching, k)
+	}
+	matching, cut := capKeys(matching, "give a repository and a head to narrow the answer")
+	values := make([]json.RawMessage, len(matching))
+	if err := parallel(len(matching), func(i int) (err error) {
+		values[i], err = r.PreviewState.Get(matching[i])
+		return err
+	}); err != nil {
+		return nil, nil, "", err
+	}
+	out := []PreviewRead{}
+	for i, k := range matching {
+		if len(values[i]) > 0 {
+			p := parts[k]
+			out = append(out, PreviewRead{Key: k, Repository: p.repo, Head: p.head, Trunk: p.trunk, Preview: values[i]})
 		}
 	}
-	return out, bad, nil
+	return out, bad, cut, nil
 }
 
 // RunningStep is a step in flight on a build module.
@@ -300,16 +341,28 @@ func (r Reader) Status() (Status, error) {
 			st.Notes = append(st.Notes, "the steps of "+m+" cannot be listed: "+err.Error())
 			continue
 		}
-		sort.Strings(keys)
-		for _, k := range keys {
-			raw, err := sr.Get(k)
+		// A step that runs started lately: the newest ids are read first, at most MaxReads of them.
+		newestFirst(keys)
+		keys, cut := capKeys(keys, "")
+		if cut != "" {
+			st.Notes = append(st.Notes, "the steps of "+m+": read the newest "+fmt.Sprint(MaxReads)+" of its keys; a step older than those is not shown")
+		}
+		steps := make([]delivery.Step, len(keys))
+		isRunning := make([]bool, len(keys))
+		_ = parallel(len(keys), func(i int) error {
+			raw, err := sr.Get(keys[i])
 			if err != nil || len(raw) == 0 {
-				continue
+				return nil
 			}
 			step, err := delivery.DecodeStep(raw)
-			if err != nil || step.State != delivery.StepRunning {
+			steps[i], isRunning[i] = step, err == nil && step.State == delivery.StepRunning
+			return nil
+		})
+		for i := range keys {
+			if !isRunning[i] {
 				continue
 			}
+			step := steps[i]
 			rs := RunningStep{Module: m, Node: step.Node, ID: step.ID, Tool: step.Tool}
 			if !step.Started.IsZero() {
 				rs.Since = step.Started.UTC().Format("2006-01-02T15:04:05Z")
@@ -344,23 +397,41 @@ func (r Reader) Status() (Status, error) {
 			return st, err
 		}
 		sort.Strings(keys)
-		for _, k := range keys {
-			raw, err := r.CandidateState.Get(k)
+		total := len(keys)
+		keys, cut := capKeys(keys, "")
+		if cut != "" {
+			st.Notes = append(st.Notes, fmt.Sprintf("read the first %d candidates of %d", MaxReads, total))
+		}
+		type judged struct {
+			c   delivery.Candidate
+			raw json.RawMessage
+			ok  bool
+		}
+		rows := make([]judged, len(keys))
+		_ = parallel(len(keys), func(i int) error {
+			raw, err := r.CandidateState.Get(keys[i])
 			if err != nil || len(raw) == 0 {
-				continue
+				return nil
 			}
 			c, err := delivery.DecodeCandidate(raw)
 			if err != nil {
-				continue
+				return nil
 			}
 			vkey, err := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
 			if err != nil {
+				return nil
+			}
+			rows[i].c = c
+			if vraw, err := r.VerdictState.Get(vkey); err == nil {
+				rows[i].raw, rows[i].ok = vraw, true
+			}
+			return nil
+		})
+		for _, row := range rows {
+			if !row.ok {
 				continue
 			}
-			raw, err = r.VerdictState.Get(vkey)
-			if err != nil {
-				continue
-			}
+			c, raw := row.c, row.raw
 			if len(raw) == 0 {
 				st.Unjudged = append(st.Unjudged, c.Group)
 				st.Waiting[KindNew] = append(st.Waiting[KindNew], c.Group)

modules/mesh-delivery-checker/internal/check/read_test.go

@@ -3,7 +3,9 @@ package check
 import (
 	"context"
 	"encoding/json"
+	"fmt"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -51,7 +53,7 @@ func readerOf(t *testing.T) (Reader, string) {
 
 func TestTheVerdictToolsAnswerTheStoredVerdictWithTheFullReportOfEachPullRequest(t *testing.T) { // N9/AC1, 1686
 	rd, key := readerOf(t)
-	vs, err := rd.Verdicts("")
+	vs, _, err := rd.Verdicts("")
 	if err != nil || len(vs) != 1 || vs[0].Key != key {
 		t.Fatalf("verdicts = %+v, %v", vs, err)
 	}
@@ -61,7 +63,7 @@ func TestTheVerdictToolsAnswerTheStoredVerdictWithTheFullReportOfEachPullRequest
 	if got := vs[0].Reports["novox/mesh-delivery#7"]; !strings.HasPrefix(got, "pass 111111111111 on aaaaaaaaaaaa") || !strings.Contains(got, "composition not yet run") {
 		t.Errorf("reports = %v", vs[0].Reports)
 	}
-	if only, _ := rd.Verdicts("feat/other"); len(only) != 0 {
+	if only, _, _ := rd.Verdicts("feat/other"); len(only) != 0 {
 		t.Errorf("another group's verdicts = %+v", only)
 	}
 }
@@ -85,11 +87,11 @@ func TestAnHqPullRequestIsAnsweredAsAnyOther(t *testing.T) {
 
 func TestTheKeptToolAnswersTheKeysAsStoredByLevelAndByDigest(t *testing.T) { // N9/AC1
 	rd, _ := readerOf(t)
-	all, err := rd.Kept("", "")
+	all, _, err := rd.Kept("", "")
 	if err != nil || len(all) != 1 {
 		t.Fatalf("kept = %v, %v", all, err)
 	}
-	one, err := rd.Kept("unit", digA)
+	one, _, err := rd.Kept("unit", digA)
 	if err != nil || len(one) != 1 {
 		t.Fatalf("one kept = %v, %v", one, err)
 	}
@@ -98,10 +100,10 @@ func TestTheKeptToolAnswersTheKeysAsStoredByLevelAndByDigest(t *testing.T) { //
 			t.Errorf("kept value = %+v, %v", k, err)
 		}
 	}
-	if none, _ := rd.Kept("integration", ""); len(none) != 0 {
+	if none, _, _ := rd.Kept("integration", ""); len(none) != 0 {
 		t.Errorf("integration = %v", none)
 	}
-	if _, err := rd.Kept("unit", digB); err == nil {
+	if _, _, err := rd.Kept("unit", digB); err == nil {
 		t.Error("a digest nothing kept was answered")
 	}
 }
@@ -115,17 +117,17 @@ func TestThePreviewsToolShowsTheStoredPreviewOfAPullRequest(t *testing.T) {
 	}
 	key, _ := delivery.PreviewKey(repoA, head1, head2)
 	rd := Reader{PreviewState: fakeState{key: data, "junk": []byte("{}")}.read()}
-	got, bad, err := rd.Previews(repoA, head1[:8])
+	got, bad, _, err := rd.Previews(repoA, head1[:8])
 	if err != nil || len(got) != 1 || got[0].Key != key || got[0].Head != head1 || got[0].Trunk != head2 || string(got[0].Preview) != string(data) {
 		t.Fatalf("previews = %+v, %v", got, err)
 	}
 	if len(bad) != 1 || bad[0] != "junk" {
 		t.Errorf("unreadable = %v", bad)
 	}
-	if got, _, _ := rd.Previews("novox/other", ""); len(got) != 0 {
+	if got, _, _, _ := rd.Previews("novox/other", ""); len(got) != 0 {
 		t.Errorf("another repository's = %+v", got)
 	}
-	if _, _, err := (Reader{}).Previews("", ""); err == nil {
+	if _, _, _, err := (Reader{}).Previews("", ""); err == nil {
 		t.Error("answered with no state to read")
 	}
 }
@@ -233,3 +235,121 @@ func TestTheStatusCountsAnHQOnlyGroupAsUnjudged(t *testing.T) {
 		t.Errorf("unjudged = %v, want pb-x", st.Unjudged)
 	}
 }
+
+// slowBucket wraps a state with a Get that takes delay and counts the values read, the lists and the most Gets in flight.
+type slowBucket struct {
+	vals                fakeState
+	reads, lists        atomic.Int64
+	inFlight, maxFlight atomic.Int64
+}
+
+func (b *slowBucket) get(key string) (json.RawMessage, error) {
+	b.reads.Add(1)
+	now := b.inFlight.Add(1)
+	for m := b.maxFlight.Load(); now > m && !b.maxFlight.CompareAndSwap(m, now); m = b.maxFlight.Load() {
+	}
+	time.Sleep(time.Millisecond)
+	b.inFlight.Add(-1)
+	return b.vals.read().Get(key)
+}
+
+func (b *slowBucket) read() StateRead {
+	return StateRead{Keys: func() ([]string, error) { b.lists.Add(1); return b.vals.read().Keys() }, Get: b.get}
+}
+
+func (b *slowBucket) Keys() ([]string, error)                 { return b.read().Keys() }
+func (b *slowBucket) Get(key string) (json.RawMessage, error) { return b.get(key) }
+func (b *slowBucket) Create(string, []byte) error             { return nil }
+
+// 1827: a read tool lists the keys of a bucket of thousands and reads only the values its answer needs, sixteen at a time, at
+// most MaxReads, saying when it cut (on the live shape of 2026-10-26: kept 7990 keys, steps 10034, tries 1000s).
+func TestReadToolsReadOnlyWhatTheyNeedInParallelAndSayWhenCut(t *testing.T) { // 1827
+	kept := fakeState{}
+	for i := 0; i < 1800; i++ {
+		level := map[bool]string{true: "unit", false: "integration"}[i%9 != 0]
+		k, _ := delivery.KeptKey(level, fmt.Sprintf("%064x", i))
+		kept[k] = []byte(`{"v":1}`)
+	}
+	previews := fakeState{}
+	for i := 0; i < 1500; i++ {
+		repo := map[bool]string{true: repoA, false: repoSDK}[i < 300]
+		k, _ := delivery.PreviewKey(repo, fmt.Sprintf("%08x%032x", i, 0), trunkA)
+		previews[k] = []byte(`{"v":1}`)
+	}
+	for _, row := range []struct {
+		name         string
+		call         func(*slowBucket) (rows int, cut string, err error)
+		rows         int
+		reads, lists int64
+		cut          bool
+	}{
+		{"kept of one key", func(b *slowBucket) (int, string, error) {
+			m, c, err := Reader{KeptState: b.read()}.Kept("unit", fmt.Sprintf("%064x", 1))
+			return len(m), c, err
+		}, 1, 1, 0, false},
+		{"kept of a level", func(b *slowBucket) (int, string, error) {
+			m, c, err := Reader{KeptState: b.read()}.Kept("integration", "")
+			return len(m), c, err
+		}, 200, 200, 1, false},
+		{"kept of everything", func(b *slowBucket) (int, string, error) {
+			m, c, err := Reader{KeptState: b.read()}.Kept("", "")
+			return len(m), c, err
+		}, MaxReads, MaxReads, 1, true},
+		{"previews of a repository", func(b *slowBucket) (int, string, error) {
+			p, _, c, err := Reader{PreviewState: b.read()}.Previews(repoSDK, "")
+			return len(p), c, err
+		}, 1200, 1000, 1, true},
+		{"previews of a repository and a head", func(b *slowBucket) (int, string, error) {
+			p, _, c, err := Reader{PreviewState: b.read()}.Previews(repoA, fmt.Sprintf("%08x", 7))
+			return len(p), c, err
+		}, 1, 1, 1, false},
+	} {
+		b := &slowBucket{}
+		b.vals = kept
+		if strings.HasPrefix(row.name, "previews") {
+			b.vals = previews
+		}
+		start := time.Now()
+		rows, cut, err := row.call(b)
+		if err != nil {
+			t.Fatalf("%s: %v", row.name, err)
+		}
+		if rows != min(row.rows, MaxReads) || b.reads.Load() != row.reads || b.lists.Load() != row.lists || (cut != "") != row.cut {
+			t.Errorf("%s: %d rows, %d reads, %d lists, cut %q; want %d, %d, %d, cut %v", row.name, rows, b.reads.Load(), b.lists.Load(), cut, min(row.rows, MaxReads), row.reads, row.lists, row.cut)
+		}
+		if row.reads > 20 && b.maxFlight.Load() < 2 {
+			t.Errorf("%s: values were read one at a time", row.name)
+		}
+		t.Logf("%s: %d values in %v (one after the other: %v)", row.name, b.reads.Load(), time.Since(start), time.Duration(b.reads.Load())*time.Millisecond)
+	}
+
+	// The status reads the newest steps of a bucket of 1500 first, at most MaxReads, and finds the one that runs.
+	now := time.Date(2026, 10, 26, 7, 0, 0, 0, time.UTC)
+	steps := fakeState{}
+	for i := 0; i < 1500; i++ {
+		id := fmt.Sprintf("mvo5%04dabcdef", i)
+		step := delivery.Step{ID: id, Tool: "t", Node: "g14", State: delivery.StepPassed}
+		if i == 1499 {
+			step.State, step.Started = delivery.StepRunning, now.Add(-time.Minute)
+		}
+		raw, _ := delivery.EncodeStep(step)
+		steps["g14."+id] = raw
+	}
+	sb := &slowBucket{vals: steps}
+	st, err := Reader{StepStates: map[string]StateRead{"mesh-build-go": sb.read()}, HaveState: fakeState{"g14.mesh-build-go": haveOf(t, delivery.PhaseRunning)}.read(),
+		Now: func() time.Time { return now }}.Status()
+	if err != nil || sb.reads.Load() != MaxReads || len(st.Running) != 1 || sb.maxFlight.Load() < 2 {
+		t.Errorf("status: %d reads (want %d), running %+v, in flight %d, %v", sb.reads.Load(), MaxReads, st.Running, sb.maxFlight.Load(), err)
+	}
+
+	// A new try reads the newest TryScan of 1500 ended tries, not all, and still finds the same head asked twice.
+	tries := fakeState{}
+	for i := 0; i < 1500; i++ {
+		raw, _ := encodeTry(Try{ID: fmt.Sprintf("try-%d", 1700000000000000000+i), Repository: repoA, Head: head1, Trunk: trunkA, State: TryPass})
+		tries[fmt.Sprintf("try-%d", 1700000000000000000+i)] = raw
+	}
+	tb := &slowBucket{vals: tries}
+	if _, err := AskTry(tb, tryTrunks, TryAsk{Repository: repoA, Head: head2, Trunk: trunkA}, now); err != nil || tb.reads.Load() != TryScan {
+		t.Errorf("AskTry read %d of 1500 tries (want %d), %v", tb.reads.Load(), TryScan, err)
+	}
+}

modules/mesh-delivery-checker/internal/check/try.go

@@ -29,6 +29,8 @@ const (
 	TriesBucket = delivery.CheckerModule + "_" + TriesState
 	// MaxTries is how many tries may wait or run at once, mesh-wide: each holds a build node for minutes.
 	MaxTries = 4
+	// TryScan is how many of the newest tries a new ask reads to find the ones that wait or run (issue 1827).
+	TryScan = 64
 	// TryKeep is how long an ended try's key is kept before the checker removes it.
 	TryKeep = 7 * 24 * time.Hour
 )
@@ -155,15 +157,33 @@ func AskTry(store TryStore, trunks map[string]string, a TryAsk, now time.Time) (
 	if err != nil {
 		return "", fmt.Errorf("the tries cannot be listed: %w", err)
 	}
-	sort.Strings(keys)
-	var open []string
-	for _, k := range keys {
-		raw, err := store.Get(k)
+	// An ended try's key stays TryKeep, so the bucket holds thousands; one that waits or runs was asked lately. Only the
+	// newest TryScan keys are read, in parallel (a try id is the time in nanoseconds, so the newest sorts last).
+	sort.Slice(keys, func(i, j int) bool {
+		if len(keys[i]) != len(keys[j]) {
+			return len(keys[i]) > len(keys[j])
+		}
+		return keys[i] > keys[j]
+	})
+	if len(keys) > TryScan {
+		keys = keys[:TryScan]
+	}
+	tries := make([]Try, len(keys))
+	live := make([]bool, len(keys))
+	if err := parallel(len(keys), func(i int) error {
+		raw, err := store.Get(keys[i])
 		if err != nil || len(raw) == 0 {
-			continue
+			return nil
 		}
 		cur, err := DecodeTry(raw)
-		if err != nil || cur.Over() {
+		tries[i], live[i] = cur, err == nil && !cur.Over()
+		return nil
+	}); err != nil {
+		return "", err
+	}
+	var open []string
+	for i, cur := range tries {
+		if !live[i] {
 			continue
 		}
 		if cur.Repository == t.Repository && cur.Head == t.Head && cur.Trunk == t.Trunk && cur.Tests == t.Tests && cur.Level == t.Level {