diff --git a/modules/mesh-build-go/cmd/mesh-build-go/read.go b/modules/mesh-build-go/cmd/mesh-build-go/read.go index fcb6ef4a..2ab40bbb 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/read.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/read.go @@ -7,9 +7,12 @@ package main import ( "encoding/json" "fmt" + "log" "sort" "strconv" "strings" + "sync" + "time" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -39,6 +42,53 @@ func readOf(s stdio.KeptState) StateRead { }} } +// MaxStateReads is the most values one read tool reads from a state in a call, and stateReadParallel how many of them +// are in flight at once. Every value is a round trip from the tool through the runtime to the bus, so a call that read +// every key of a large state one after the other outlasted the call's 30 s (issue 1827: 1164 keys, 20 MB). +const ( + MaxStateReads = 1000 + stateReadParallel = 16 +) + +// readValues reads the values of keys, stateReadParallel at a time, in the order of keys (nil for a key with none). It +// stops starting reads after the first error and returns it. +func readValues(keys []string, get func(string) (json.RawMessage, error)) ([]json.RawMessage, error) { + out := make([]json.RawMessage, len(keys)) + var ( + wg sync.WaitGroup + mu sync.Mutex + fail error + next = make(chan int) + ) + for w := 0; w < stateReadParallel && w < len(keys); w++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := range next { + raw, err := get(keys[i]) + mu.Lock() + if err != nil && fail == nil { + fail = err + } + mu.Unlock() + out[i] = raw + } + }() + } + for i := range keys { + mu.Lock() + stop := fail != nil + mu.Unlock() + if stop { + break + } + next <- i + } + close(next) + wg.Wait() + return out, fail +} + // stepKey is the key of a step of this node: the id alone, or the whole key . as a step tool answered it. func (m *Module) stepKey(id string) (string, error) { id = strings.TrimSpace(id) @@ -80,27 +130,41 @@ type StepLine struct { Reason string `json:"reason,omitempty"` } -// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). -func (m *Module) RecentSteps(limit int) ([]StepLine, error) { +// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only +// the node's own keys (.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's +// id is base-36 milliseconds and random digits, so its string order is its start order. cut says older steps were not +// read. The values still order the lines, since an ask step's id carries the ask time. +func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) { if limit <= 0 { limit = 10 } if m.ReadSteps.Keys == nil || m.ReadSteps.Get == nil { - return nil, fmt.Errorf("the steps state cannot be read here") + return nil, false, fmt.Errorf("the steps state cannot be read here") } + start := time.Now() keys, err := m.ReadSteps.Keys() if err != nil { - return nil, err + return nil, false, err + } + own := keys[:0:0] + for _, k := range keys { + if strings.HasPrefix(k, m.Node+".") { + if node, _, err := delivery.ParseStepKey(k); err == nil && node == m.Node { + own = append(own, k) + } + } + } + sort.Sort(sort.Reverse(sort.StringSlice(own))) + if n := min(max(limit*4, 100), MaxStateReads); len(own) > n { + own, cut = own[:n], true + } + raws, err := readValues(own, m.ReadSteps.Get) + log.Printf("steps read: %d of %d keys listed read in %v (cut %v)", len(own), len(keys), time.Since(start).Round(time.Millisecond), cut) + if err != nil { + return nil, cut, err } var steps []delivery.Step - for _, k := range keys { - if node, _, err := delivery.ParseStepKey(k); err != nil || node != m.Node { - continue - } - raw, err := m.ReadSteps.Get(k) - if err != nil { - return nil, err - } + for _, raw := range raws { if len(raw) == 0 { continue } @@ -122,7 +186,7 @@ func (m *Module) RecentSteps(limit int) ([]StepLine, error) { } out = append(out, l) } - return out, nil + return out, cut, nil } // ReadToolchains is every key of the toolchains state (.) with its stored value. @@ -130,24 +194,30 @@ func (m *Module) ReadToolchains() (map[string]delivery.Toolchain, error) { if m.ReadToolchainsState.Keys == nil || m.ReadToolchainsState.Get == nil { return nil, fmt.Errorf("the toolchains state cannot be read here") } + start := time.Now() keys, err := m.ReadToolchainsState.Keys() if err != nil { return nil, err } + cut := len(keys) > MaxStateReads + if cut { + keys = keys[:MaxStateReads] // a node has a key per language: far fewer in practice + } + raws, err := readValues(keys, m.ReadToolchainsState.Get) + log.Printf("toolchains read: %d keys read in %v (cut %v: keys past %d were not read)", len(keys), time.Since(start).Round(time.Millisecond), cut, MaxStateReads) + if err != nil { + return nil, err + } out := map[string]delivery.Toolchain{} - for _, k := range keys { - raw, err := m.ReadToolchainsState.Get(k) - if err != nil { - return nil, err - } + for i, raw := range raws { if len(raw) == 0 { continue } t, err := delivery.DecodeToolchain(raw) if err != nil { - return nil, fmt.Errorf("%s: %w", k, err) + return nil, fmt.Errorf("%s: %w", keys[i], err) } - out[k] = t + out[keys[i]] = t } return out, nil } @@ -171,8 +241,13 @@ func (m *Module) ReadTools() []stdio.Tool { Input: map[string]any{"type": "object", "properties": map[string]any{ "limit": map[string]any{"type": "integer", "description": "how many, at most; 10 when left out"}}}, Run: func(args map[string]any) (any, error) { - steps, err := m.RecentSteps(intArg(args["limit"])) - return map[string]any{"node": m.Node, "steps": steps}, err + steps, cut, err := m.RecentSteps(intArg(args["limit"])) + ans := map[string]any{"node": m.Node, "steps": steps} + if cut { + ans["cut"] = true + ans["note"] = fmt.Sprintf("this node has more steps stored than one call reads; older steps were not read (ask for a larger limit, at most %d reads)", MaxStateReads) + } + return ans, err }}, {Name: ToolchainsReadTool, Description: "What the module's toolchains state holds: each key . with the version installed, the " + diff --git a/modules/mesh-build-go/cmd/mesh-build-go/read_test.go b/modules/mesh-build-go/cmd/mesh-build-go/read_test.go index 38accef1..04049540 100644 --- a/modules/mesh-build-go/cmd/mesh-build-go/read_test.go +++ b/modules/mesh-build-go/cmd/mesh-build-go/read_test.go @@ -3,8 +3,10 @@ package main import ( "context" "encoding/json" + "fmt" "slices" "strings" + "sync/atomic" "testing" "time" @@ -145,3 +147,62 @@ func TestReadToolsAreInTheManifestAndNeedNoToolchain(t *testing.T) { t.Errorf("the read refused: %v", err) } } + +// 1827: on the live shape (4 nodes, 600 steps each; a steps state of 1164 keys / 20 MB timed out at 30 s) the steps tool +// lists, then reads only this node's keys, in parallel, at most MaxStateReads of them, and says when it cut. +func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) { + t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC) + for _, row := range []struct { + perNode, limit int + reads int64 + steps int + cut bool + }{ + {20, 0, 20, 10, false}, + {600, 0, 100, 10, true}, + {1500, 300, MaxStateReads, 300, true}, + } { + var reads, inFlight, maxFlight atomic.Int64 + stored := map[string][]byte{} + for _, n := range []string{"ace", "g14", "novox", "shanks"} { + for i := 0; i < row.perNode; i++ { + b, _ := delivery.EncodeStep(delivery.Step{ID: fmt.Sprintf("s%04d", i), Tool: "t", Node: n, State: delivery.StepPassed, Started: t0.Add(time.Duration(i) * time.Second)}) + stored[n+".s"+fmt.Sprintf("%04d", i)] = b + } + } + m := &Module{Node: "shanks"} + m.ReadSteps = StateRead{ + Keys: func() ([]string, error) { + var k []string + for key := range stored { + k = append(k, key) + } + return k, nil + }, + Get: func(key string) (json.RawMessage, error) { + reads.Add(1) + now := inFlight.Add(1) + for o := maxFlight.Load(); now > o && !maxFlight.CompareAndSwap(o, now); o = maxFlight.Load() { + } + time.Sleep(time.Millisecond) + inFlight.Add(-1) + if !strings.HasPrefix(key, "shanks.") { + t.Errorf("read %s, a key of another node", key) + } + return stored[key], nil + }, + } + start := time.Now() + lines, cut, err := m.RecentSteps(row.limit) + if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads { + t.Errorf("%d a node: %d steps, cut %v, %d reads, err %v; want %d, %v, %d", row.perNode, len(lines), cut, reads.Load(), err, row.steps, row.cut, row.reads) + } + if row.cut && len(lines) > 0 && lines[0].ID != fmt.Sprintf("s%04d", row.perNode-1) { + t.Errorf("%d a node: newest answered %s", row.perNode, lines[0].ID) + } + if maxFlight.Load() < 2 { + t.Errorf("%d a node: values read one at a time", row.perNode) + } + t.Logf("%d a node: %d reads in %v (one after the other: %v)", row.perNode, reads.Load(), time.Since(start), time.Duration(reads.Load())*time.Millisecond) + } +} diff --git a/modules/mesh-build-image/cmd/mesh-build-image/read.go b/modules/mesh-build-image/cmd/mesh-build-image/read.go index cbae0998..b97ddccc 100644 --- a/modules/mesh-build-image/cmd/mesh-build-image/read.go +++ b/modules/mesh-build-image/cmd/mesh-build-image/read.go @@ -7,9 +7,12 @@ package main import ( "encoding/json" "fmt" + "log" "sort" "strconv" "strings" + "sync" + "time" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -38,6 +41,53 @@ func readOf(s stdio.KeptState) StateRead { }} } +// MaxStateReads is the most values one read tool reads from a state in a call, and stateReadParallel how many of them +// are in flight at once. Every value is a round trip from the tool through the runtime to the bus, so a call that read +// every key of a large state one after the other outlasted the call's 30 s (issue 1827: 1164 keys, 20 MB). +const ( + MaxStateReads = 1000 + stateReadParallel = 16 +) + +// readValues reads the values of keys, stateReadParallel at a time, in the order of keys (nil for a key with none). It +// stops starting reads after the first error and returns it. +func readValues(keys []string, get func(string) (json.RawMessage, error)) ([]json.RawMessage, error) { + out := make([]json.RawMessage, len(keys)) + var ( + wg sync.WaitGroup + mu sync.Mutex + fail error + next = make(chan int) + ) + for w := 0; w < stateReadParallel && w < len(keys); w++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := range next { + raw, err := get(keys[i]) + mu.Lock() + if err != nil && fail == nil { + fail = err + } + mu.Unlock() + out[i] = raw + } + }() + } + for i := range keys { + mu.Lock() + stop := fail != nil + mu.Unlock() + if stop { + break + } + next <- i + } + close(next) + wg.Wait() + return out, fail +} + // stepKey is the key of a step of this node: the id alone, or the whole key . as a step tool answered it. func (m *Module) stepKey(id string) (string, error) { id = strings.TrimSpace(id) @@ -79,27 +129,41 @@ type StepLine struct { Reason string `json:"reason,omitempty"` } -// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). -func (m *Module) RecentSteps(limit int) ([]StepLine, error) { +// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only +// the node's own keys (.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's +// id is base-36 milliseconds and random digits, so its string order is its start order. cut says older steps were not +// read. The values still order the lines, since an ask step's id carries the ask time. +func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) { if limit <= 0 { limit = 10 } if m.ReadSteps.Keys == nil || m.ReadSteps.Get == nil { - return nil, fmt.Errorf("the steps state cannot be read here") + return nil, false, fmt.Errorf("the steps state cannot be read here") } + start := time.Now() keys, err := m.ReadSteps.Keys() if err != nil { - return nil, err + return nil, false, err + } + own := keys[:0:0] + for _, k := range keys { + if strings.HasPrefix(k, m.Node+".") { + if node, _, err := delivery.ParseStepKey(k); err == nil && node == m.Node { + own = append(own, k) + } + } + } + sort.Sort(sort.Reverse(sort.StringSlice(own))) + if n := min(max(limit*4, 100), MaxStateReads); len(own) > n { + own, cut = own[:n], true + } + raws, err := readValues(own, m.ReadSteps.Get) + log.Printf("steps read: %d of %d keys listed read in %v (cut %v)", len(own), len(keys), time.Since(start).Round(time.Millisecond), cut) + if err != nil { + return nil, cut, err } var steps []delivery.Step - for _, k := range keys { - if node, _, err := delivery.ParseStepKey(k); err != nil || node != m.Node { - continue - } - raw, err := m.ReadSteps.Get(k) - if err != nil { - return nil, err - } + for _, raw := range raws { if len(raw) == 0 { continue } @@ -121,7 +185,7 @@ func (m *Module) RecentSteps(limit int) ([]StepLine, error) { } out = append(out, l) } - return out, nil + return out, cut, nil } // ReadTools are the module's read tools. All are read-only. @@ -143,8 +207,13 @@ func (m *Module) ReadTools() []stdio.Tool { Input: map[string]any{"type": "object", "properties": map[string]any{ "limit": map[string]any{"type": "integer", "description": "how many, at most; 10 when left out"}}}, Run: func(args map[string]any) (any, error) { - steps, err := m.RecentSteps(intArg(args["limit"])) - return map[string]any{"node": m.Node, "steps": steps}, err + steps, cut, err := m.RecentSteps(intArg(args["limit"])) + ans := map[string]any{"node": m.Node, "steps": steps} + if cut { + ans["cut"] = true + ans["note"] = fmt.Sprintf("this node has more steps stored than one call reads; older steps were not read (ask for a larger limit, at most %d reads)", MaxStateReads) + } + return ans, err }}, } } diff --git a/modules/mesh-build-image/cmd/mesh-build-image/read_test.go b/modules/mesh-build-image/cmd/mesh-build-image/read_test.go index 8ef4ce26..61c6fc95 100644 --- a/modules/mesh-build-image/cmd/mesh-build-image/read_test.go +++ b/modules/mesh-build-image/cmd/mesh-build-image/read_test.go @@ -2,7 +2,11 @@ package main import ( "encoding/json" + "fmt" + "strings" + "sync/atomic" "testing" + "time" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -39,8 +43,67 @@ func TestTheReadToolsAnswerAStepsResult(t *testing.T) { if err := json.Unmarshal(got.Result, &back); err != nil || back.Digest != res.Digest { t.Errorf("result = %s, %v", got.Result, err) } - lines, err := m.RecentSteps(0) + lines, _, err := m.RecentSteps(0) if err != nil || len(lines) != 1 || lines[0].Tool != BuildTool { t.Errorf("steps = %+v, %v", lines, err) } } + +// 1827: on the live shape (4 nodes, 600 steps each; a steps state of 1164 keys / 20 MB timed out at 30 s) the steps tool +// lists, then reads only this node's keys, in parallel, at most MaxStateReads of them, and says when it cut. +func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) { + t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC) + for _, row := range []struct { + perNode, limit int + reads int64 + steps int + cut bool + }{ + {20, 0, 20, 10, false}, + {600, 0, 100, 10, true}, + {1500, 300, MaxStateReads, 300, true}, + } { + var reads, inFlight, maxFlight atomic.Int64 + stored := map[string][]byte{} + for _, n := range []string{"ace", "g14", "novox", "shanks"} { + for i := 0; i < row.perNode; i++ { + b, _ := delivery.EncodeStep(delivery.Step{ID: fmt.Sprintf("s%04d", i), Tool: "t", Node: n, State: delivery.StepPassed, Started: t0.Add(time.Duration(i) * time.Second)}) + stored[n+".s"+fmt.Sprintf("%04d", i)] = b + } + } + m := &Module{Node: "shanks"} + m.ReadSteps = StateRead{ + Keys: func() ([]string, error) { + var k []string + for key := range stored { + k = append(k, key) + } + return k, nil + }, + Get: func(key string) (json.RawMessage, error) { + reads.Add(1) + now := inFlight.Add(1) + for o := maxFlight.Load(); now > o && !maxFlight.CompareAndSwap(o, now); o = maxFlight.Load() { + } + time.Sleep(time.Millisecond) + inFlight.Add(-1) + if !strings.HasPrefix(key, "shanks.") { + t.Errorf("read %s, a key of another node", key) + } + return stored[key], nil + }, + } + start := time.Now() + lines, cut, err := m.RecentSteps(row.limit) + if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads { + t.Errorf("%d a node: %d steps, cut %v, %d reads, err %v; want %d, %v, %d", row.perNode, len(lines), cut, reads.Load(), err, row.steps, row.cut, row.reads) + } + if row.cut && len(lines) > 0 && lines[0].ID != fmt.Sprintf("s%04d", row.perNode-1) { + t.Errorf("%d a node: newest answered %s", row.perNode, lines[0].ID) + } + if maxFlight.Load() < 2 { + t.Errorf("%d a node: values read one at a time", row.perNode) + } + t.Logf("%d a node: %d reads in %v (one after the other: %v)", row.perNode, reads.Load(), time.Since(start), time.Duration(reads.Load())*time.Millisecond) + } +} diff --git a/modules/mesh-build-node/cmd/mesh-build-node/read.go b/modules/mesh-build-node/cmd/mesh-build-node/read.go index 983587fd..4b290eaf 100644 --- a/modules/mesh-build-node/cmd/mesh-build-node/read.go +++ b/modules/mesh-build-node/cmd/mesh-build-node/read.go @@ -7,9 +7,12 @@ package main import ( "encoding/json" "fmt" + "log" "sort" "strconv" "strings" + "sync" + "time" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/delivery" @@ -39,6 +42,53 @@ func readOf(s stdio.KeptState) StateRead { }} } +// MaxStateReads is the most values one read tool reads from a state in a call, and stateReadParallel how many of them +// are in flight at once. Every value is a round trip from the tool through the runtime to the bus, so a call that read +// every key of a large state one after the other outlasted the call's 30 s (issue 1827: 1164 keys, 20 MB). +const ( + MaxStateReads = 1000 + stateReadParallel = 16 +) + +// readValues reads the values of keys, stateReadParallel at a time, in the order of keys (nil for a key with none). It +// stops starting reads after the first error and returns it. +func readValues(keys []string, get func(string) (json.RawMessage, error)) ([]json.RawMessage, error) { + out := make([]json.RawMessage, len(keys)) + var ( + wg sync.WaitGroup + mu sync.Mutex + fail error + next = make(chan int) + ) + for w := 0; w < stateReadParallel && w < len(keys); w++ { + wg.Add(1) + go func() { + defer wg.Done() + for i := range next { + raw, err := get(keys[i]) + mu.Lock() + if err != nil && fail == nil { + fail = err + } + mu.Unlock() + out[i] = raw + } + }() + } + for i := range keys { + mu.Lock() + stop := fail != nil + mu.Unlock() + if stop { + break + } + next <- i + } + close(next) + wg.Wait() + return out, fail +} + // stepKey is the key of a step of this node: the id alone, or the whole key . as a step tool answered it. func (m *Module) stepKey(id string) (string, error) { id = strings.TrimSpace(id) @@ -80,27 +130,41 @@ type StepLine struct { Reason string `json:"reason,omitempty"` } -// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). -func (m *Module) RecentSteps(limit int) ([]StepLine, error) { +// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only +// the node's own keys (.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's +// id is base-36 milliseconds and random digits, so its string order is its start order. cut says older steps were not +// read. The values still order the lines, since an ask step's id carries the ask time. +func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) { if limit <= 0 { limit = 10 } if m.ReadSteps.Keys == nil || m.ReadSteps.Get == nil { - return nil, fmt.Errorf("the steps state cannot be read here") + return nil, false, fmt.Errorf("the steps state cannot be read here") } + start := time.Now() keys, err := m.ReadSteps.Keys() if err != nil { - return nil, err + return nil, false, err + } + own := keys[:0:0] + for _, k := range keys { + if strings.HasPrefix(k, m.Node+".") { + if node, _, err := delivery.ParseStepKey(k); err == nil && node == m.Node { + own = append(own, k) + } + } + } + sort.Sort(sort.Reverse(sort.StringSlice(own))) + if n := min(max(limit*4, 100), MaxStateReads); len(own) > n { + own, cut = own[:n], true + } + raws, err := readValues(own, m.ReadSteps.Get) + log.Printf("steps read: %d of %d keys listed read in %v (cut %v)", len(own), len(keys), time.Since(start).Round(time.Millisecond), cut) + if err != nil { + return nil, cut, err } var steps []delivery.Step - for _, k := range keys { - if node, _, err := delivery.ParseStepKey(k); err != nil || node != m.Node { - continue - } - raw, err := m.ReadSteps.Get(k) - if err != nil { - return nil, err - } + for _, raw := range raws { if len(raw) == 0 { continue } @@ -122,7 +186,7 @@ func (m *Module) RecentSteps(limit int) ([]StepLine, error) { } out = append(out, l) } - return out, nil + return out, cut, nil } // ReadToolchains is every key of the toolchains state (.) with its stored value. @@ -130,24 +194,30 @@ func (m *Module) ReadToolchains() (map[string]delivery.Toolchain, error) { if m.ReadToolchainsState.Keys == nil || m.ReadToolchainsState.Get == nil { return nil, fmt.Errorf("the toolchains state cannot be read here") } + start := time.Now() keys, err := m.ReadToolchainsState.Keys() if err != nil { return nil, err } + cut := len(keys) > MaxStateReads + if cut { + keys = keys[:MaxStateReads] // a node has a key per language: far fewer in practice + } + raws, err := readValues(keys, m.ReadToolchainsState.Get) + log.Printf("toolchains read: %d keys read in %v (cut %v: keys past %d were not read)", len(keys), time.Since(start).Round(time.Millisecond), cut, MaxStateReads) + if err != nil { + return nil, err + } out := map[string]delivery.Toolchain{} - for _, k := range keys { - raw, err := m.ReadToolchainsState.Get(k) - if err != nil { - return nil, err - } + for i, raw := range raws { if len(raw) == 0 { continue } t, err := delivery.DecodeToolchain(raw) if err != nil { - return nil, fmt.Errorf("%s: %w", k, err) + return nil, fmt.Errorf("%s: %w", keys[i], err) } - out[k] = t + out[keys[i]] = t } return out, nil } @@ -171,8 +241,13 @@ func (m *Module) ReadTools() []stdio.Tool { Input: map[string]any{"type": "object", "properties": map[string]any{ "limit": map[string]any{"type": "integer", "description": "how many, at most; 10 when left out"}}}, Run: func(args map[string]any) (any, error) { - steps, err := m.RecentSteps(intArg(args["limit"])) - return map[string]any{"node": m.Node, "steps": steps}, err + steps, cut, err := m.RecentSteps(intArg(args["limit"])) + ans := map[string]any{"node": m.Node, "steps": steps} + if cut { + ans["cut"] = true + ans["note"] = fmt.Sprintf("this node has more steps stored than one call reads; older steps were not read (ask for a larger limit, at most %d reads)", MaxStateReads) + } + return ans, err }}, {Name: ToolchainsReadTool, Description: "What the module's toolchains state holds: each key . with the version installed, the " + diff --git a/modules/mesh-build-node/cmd/mesh-build-node/read_test.go b/modules/mesh-build-node/cmd/mesh-build-node/read_test.go index 11cfa6fc..cc0d85ef 100644 --- a/modules/mesh-build-node/cmd/mesh-build-node/read_test.go +++ b/modules/mesh-build-node/cmd/mesh-build-node/read_test.go @@ -3,8 +3,10 @@ package main import ( "context" "encoding/json" + "fmt" "slices" "strings" + "sync/atomic" "testing" "time" @@ -145,3 +147,62 @@ func TestReadToolsAreInTheManifestAndNeedNoToolchain(t *testing.T) { t.Errorf("the read refused: %v", err) } } + +// 1827: on the live shape (4 nodes, 600 steps each; a steps state of 1164 keys / 20 MB timed out at 30 s) the steps tool +// lists, then reads only this node's keys, in parallel, at most MaxStateReads of them, and says when it cut. +func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) { + t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC) + for _, row := range []struct { + perNode, limit int + reads int64 + steps int + cut bool + }{ + {20, 0, 20, 10, false}, + {600, 0, 100, 10, true}, + {1500, 300, MaxStateReads, 300, true}, + } { + var reads, inFlight, maxFlight atomic.Int64 + stored := map[string][]byte{} + for _, n := range []string{"ace", "g14", "novox", "shanks"} { + for i := 0; i < row.perNode; i++ { + b, _ := delivery.EncodeStep(delivery.Step{ID: fmt.Sprintf("s%04d", i), Tool: "t", Node: n, State: delivery.StepPassed, Started: t0.Add(time.Duration(i) * time.Second)}) + stored[n+".s"+fmt.Sprintf("%04d", i)] = b + } + } + m := &Module{Node: "shanks"} + m.ReadSteps = StateRead{ + Keys: func() ([]string, error) { + var k []string + for key := range stored { + k = append(k, key) + } + return k, nil + }, + Get: func(key string) (json.RawMessage, error) { + reads.Add(1) + now := inFlight.Add(1) + for o := maxFlight.Load(); now > o && !maxFlight.CompareAndSwap(o, now); o = maxFlight.Load() { + } + time.Sleep(time.Millisecond) + inFlight.Add(-1) + if !strings.HasPrefix(key, "shanks.") { + t.Errorf("read %s, a key of another node", key) + } + return stored[key], nil + }, + } + start := time.Now() + lines, cut, err := m.RecentSteps(row.limit) + if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads { + t.Errorf("%d a node: %d steps, cut %v, %d reads, err %v; want %d, %v, %d", row.perNode, len(lines), cut, reads.Load(), err, row.steps, row.cut, row.reads) + } + if row.cut && len(lines) > 0 && lines[0].ID != fmt.Sprintf("s%04d", row.perNode-1) { + t.Errorf("%d a node: newest answered %s", row.perNode, lines[0].ID) + } + if maxFlight.Load() < 2 { + t.Errorf("%d a node: values read one at a time", row.perNode) + } + t.Logf("%d a node: %d reads in %v (one after the other: %v)", row.perNode, reads.Load(), time.Since(start), time.Duration(reads.Load())*time.Millisecond) + } +}