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 {