jschoubben
e901769cb2
mesh/delivery merged: source.main moved to novox/mesh-catalog@567736f7795a
mesh/check failed: integration: meshbuildnode/cmd/mesh-build-node.TestAFixtureBundleBuiltTwiceHasOneDigest: no toolchain of this lock is installed und…
The review found a cut read took an arbitrary 1000 keys, so the newest steps could be missing. A step id is base-36 milliseconds plus random digits, so its string order is its start order; the newest max(limit*4, 100) keys are read now, 1000 at most. The toolchains read logs when it drops keys past its cap.
110 lines
3.5 KiB
Go
110 lines
3.5 KiB
Go
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"strings"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
)
|
|
|
|
// The read tools answer what the steps state holds: a finished build's result is there, read by the step's id.
|
|
func TestTheReadToolsAnswerAStepsResult(t *testing.T) {
|
|
m, _, states := setup(t, same)
|
|
res, st := runBuild(t, m, states, nil)
|
|
if st.State != delivery.StepPassed {
|
|
t.Fatalf("step = %+v", st)
|
|
}
|
|
m.ReadSteps = StateRead{
|
|
Keys: func() ([]string, error) {
|
|
states.mu.Lock()
|
|
defer states.mu.Unlock()
|
|
var k []string
|
|
for key := range states.steps {
|
|
k = append(k, key)
|
|
}
|
|
return k, nil
|
|
},
|
|
Get: func(key string) (json.RawMessage, error) {
|
|
states.mu.Lock()
|
|
defer states.mu.Unlock()
|
|
v := states.steps[key]
|
|
return json.Marshal(v[len(v)-1])
|
|
},
|
|
}
|
|
got, err := m.ReadStep(st.ID)
|
|
if err != nil || got.State != delivery.StepPassed {
|
|
t.Fatalf("ReadStep = %+v, %v", got, err)
|
|
}
|
|
var back BuildResult
|
|
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)
|
|
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)
|
|
}
|
|
}
|