Read the newest steps first: sort the node's keys by id (start order) before the cut, and say older steps were not read
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…
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.
This commit is contained in:
@@ -131,8 +131,9 @@ type StepLine struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only
|
// 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 (<node>.) are read, at most MaxStateReads of them; cut says more were stored than were read, so
|
// the node's own keys (<node>.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's
|
||||||
// the newest may be missing. A step's id says nothing of when it started, so the values are what orders them.
|
// 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) {
|
func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) {
|
||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
limit = 10
|
limit = 10
|
||||||
@@ -153,8 +154,9 @@ func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if len(own) > MaxStateReads {
|
sort.Sort(sort.Reverse(sort.StringSlice(own)))
|
||||||
own, cut = own[:MaxStateReads], true
|
if n := min(max(limit*4, 100), MaxStateReads); len(own) > n {
|
||||||
|
own, cut = own[:n], true
|
||||||
}
|
}
|
||||||
raws, err := readValues(own, m.ReadSteps.Get)
|
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)
|
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)
|
||||||
@@ -197,11 +199,12 @@ func (m *Module) ReadToolchains() (map[string]delivery.Toolchain, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if len(keys) > MaxStateReads {
|
cut := len(keys) > MaxStateReads
|
||||||
|
if cut {
|
||||||
keys = keys[:MaxStateReads] // a node has a key per language: far fewer in practice
|
keys = keys[:MaxStateReads] // a node has a key per language: far fewer in practice
|
||||||
}
|
}
|
||||||
raws, err := readValues(keys, m.ReadToolchainsState.Get)
|
raws, err := readValues(keys, m.ReadToolchainsState.Get)
|
||||||
log.Printf("toolchains read: %d keys read in %v", len(keys), time.Since(start).Round(time.Millisecond))
|
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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -242,7 +245,7 @@ func (m *Module) ReadTools() []stdio.Tool {
|
|||||||
ans := map[string]any{"node": m.Node, "steps": steps}
|
ans := map[string]any{"node": m.Node, "steps": steps}
|
||||||
if cut {
|
if cut {
|
||||||
ans["cut"] = true
|
ans["cut"] = true
|
||||||
ans["note"] = fmt.Sprintf("this node has more than %d steps stored; %d were read, so the newest may be missing", MaxStateReads, MaxStateReads)
|
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
|
return ans, err
|
||||||
}},
|
}},
|
||||||
|
|||||||
@@ -153,14 +153,14 @@ func TestReadToolsAreInTheManifestAndNeedNoToolchain(t *testing.T) {
|
|||||||
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
|
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
|
||||||
t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
|
t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
|
||||||
for _, row := range []struct {
|
for _, row := range []struct {
|
||||||
perNode int
|
perNode, limit int
|
||||||
reads int64
|
reads int64
|
||||||
steps int
|
steps int
|
||||||
cut bool
|
cut bool
|
||||||
}{
|
}{
|
||||||
{20, 20, 10, false},
|
{20, 0, 20, 10, false},
|
||||||
{600, 600, 10, false},
|
{600, 0, 100, 10, true},
|
||||||
{1500, MaxStateReads, 10, true},
|
{1500, 300, MaxStateReads, 300, true},
|
||||||
} {
|
} {
|
||||||
var reads, inFlight, maxFlight atomic.Int64
|
var reads, inFlight, maxFlight atomic.Int64
|
||||||
stored := map[string][]byte{}
|
stored := map[string][]byte{}
|
||||||
@@ -193,10 +193,13 @@ func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
lines, cut, err := m.RecentSteps(0)
|
lines, cut, err := m.RecentSteps(row.limit)
|
||||||
if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads {
|
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)
|
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 {
|
if maxFlight.Load() < 2 {
|
||||||
t.Errorf("%d a node: values read one at a time", row.perNode)
|
t.Errorf("%d a node: values read one at a time", row.perNode)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -130,8 +130,9 @@ type StepLine struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only
|
// 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 (<node>.) are read, at most MaxStateReads of them; cut says more were stored than were read, so
|
// the node's own keys (<node>.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's
|
||||||
// the newest may be missing. A step's id says nothing of when it started, so the values are what orders them.
|
// 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) {
|
func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) {
|
||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
limit = 10
|
limit = 10
|
||||||
@@ -152,8 +153,9 @@ func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if len(own) > MaxStateReads {
|
sort.Sort(sort.Reverse(sort.StringSlice(own)))
|
||||||
own, cut = own[:MaxStateReads], true
|
if n := min(max(limit*4, 100), MaxStateReads); len(own) > n {
|
||||||
|
own, cut = own[:n], true
|
||||||
}
|
}
|
||||||
raws, err := readValues(own, m.ReadSteps.Get)
|
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)
|
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)
|
||||||
@@ -209,7 +211,7 @@ func (m *Module) ReadTools() []stdio.Tool {
|
|||||||
ans := map[string]any{"node": m.Node, "steps": steps}
|
ans := map[string]any{"node": m.Node, "steps": steps}
|
||||||
if cut {
|
if cut {
|
||||||
ans["cut"] = true
|
ans["cut"] = true
|
||||||
ans["note"] = fmt.Sprintf("this node has more than %d steps stored; %d were read, so the newest may be missing", MaxStateReads, MaxStateReads)
|
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
|
return ans, err
|
||||||
}},
|
}},
|
||||||
|
|||||||
@@ -54,14 +54,14 @@ func TestTheReadToolsAnswerAStepsResult(t *testing.T) {
|
|||||||
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
|
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
|
||||||
t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
|
t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
|
||||||
for _, row := range []struct {
|
for _, row := range []struct {
|
||||||
perNode int
|
perNode, limit int
|
||||||
reads int64
|
reads int64
|
||||||
steps int
|
steps int
|
||||||
cut bool
|
cut bool
|
||||||
}{
|
}{
|
||||||
{20, 20, 10, false},
|
{20, 0, 20, 10, false},
|
||||||
{600, 600, 10, false},
|
{600, 0, 100, 10, true},
|
||||||
{1500, MaxStateReads, 10, true},
|
{1500, 300, MaxStateReads, 300, true},
|
||||||
} {
|
} {
|
||||||
var reads, inFlight, maxFlight atomic.Int64
|
var reads, inFlight, maxFlight atomic.Int64
|
||||||
stored := map[string][]byte{}
|
stored := map[string][]byte{}
|
||||||
@@ -94,10 +94,13 @@ func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
lines, cut, err := m.RecentSteps(0)
|
lines, cut, err := m.RecentSteps(row.limit)
|
||||||
if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads {
|
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)
|
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 {
|
if maxFlight.Load() < 2 {
|
||||||
t.Errorf("%d a node: values read one at a time", row.perNode)
|
t.Errorf("%d a node: values read one at a time", row.perNode)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -131,8 +131,9 @@ type StepLine struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// RecentSteps are this node's steps, the latest started first, at most limit (ten when limit is not above zero). Only
|
// 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 (<node>.) are read, at most MaxStateReads of them; cut says more were stored than were read, so
|
// the node's own keys (<node>.) are listed and the newest max(limit*4, 100) of them read (at most MaxStateReads): a step's
|
||||||
// the newest may be missing. A step's id says nothing of when it started, so the values are what orders them.
|
// 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) {
|
func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error) {
|
||||||
if limit <= 0 {
|
if limit <= 0 {
|
||||||
limit = 10
|
limit = 10
|
||||||
@@ -153,8 +154,9 @@ func (m *Module) RecentSteps(limit int) (lines []StepLine, cut bool, err error)
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if len(own) > MaxStateReads {
|
sort.Sort(sort.Reverse(sort.StringSlice(own)))
|
||||||
own, cut = own[:MaxStateReads], true
|
if n := min(max(limit*4, 100), MaxStateReads); len(own) > n {
|
||||||
|
own, cut = own[:n], true
|
||||||
}
|
}
|
||||||
raws, err := readValues(own, m.ReadSteps.Get)
|
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)
|
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)
|
||||||
@@ -197,11 +199,12 @@ func (m *Module) ReadToolchains() (map[string]delivery.Toolchain, error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
if len(keys) > MaxStateReads {
|
cut := len(keys) > MaxStateReads
|
||||||
|
if cut {
|
||||||
keys = keys[:MaxStateReads] // a node has a key per language: far fewer in practice
|
keys = keys[:MaxStateReads] // a node has a key per language: far fewer in practice
|
||||||
}
|
}
|
||||||
raws, err := readValues(keys, m.ReadToolchainsState.Get)
|
raws, err := readValues(keys, m.ReadToolchainsState.Get)
|
||||||
log.Printf("toolchains read: %d keys read in %v", len(keys), time.Since(start).Round(time.Millisecond))
|
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 {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
@@ -242,7 +245,7 @@ func (m *Module) ReadTools() []stdio.Tool {
|
|||||||
ans := map[string]any{"node": m.Node, "steps": steps}
|
ans := map[string]any{"node": m.Node, "steps": steps}
|
||||||
if cut {
|
if cut {
|
||||||
ans["cut"] = true
|
ans["cut"] = true
|
||||||
ans["note"] = fmt.Sprintf("this node has more than %d steps stored; %d were read, so the newest may be missing", MaxStateReads, MaxStateReads)
|
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
|
return ans, err
|
||||||
}},
|
}},
|
||||||
|
|||||||
@@ -153,14 +153,14 @@ func TestReadToolsAreInTheManifestAndNeedNoToolchain(t *testing.T) {
|
|||||||
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
|
func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
|
||||||
t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
|
t0 := time.Date(2026, 10, 25, 9, 0, 0, 0, time.UTC)
|
||||||
for _, row := range []struct {
|
for _, row := range []struct {
|
||||||
perNode int
|
perNode, limit int
|
||||||
reads int64
|
reads int64
|
||||||
steps int
|
steps int
|
||||||
cut bool
|
cut bool
|
||||||
}{
|
}{
|
||||||
{20, 20, 10, false},
|
{20, 0, 20, 10, false},
|
||||||
{600, 600, 10, false},
|
{600, 0, 100, 10, true},
|
||||||
{1500, MaxStateReads, 10, true},
|
{1500, 300, MaxStateReads, 300, true},
|
||||||
} {
|
} {
|
||||||
var reads, inFlight, maxFlight atomic.Int64
|
var reads, inFlight, maxFlight atomic.Int64
|
||||||
stored := map[string][]byte{}
|
stored := map[string][]byte{}
|
||||||
@@ -193,10 +193,13 @@ func TestRecentStepsReadsOnlyTheNodesKeysInParallelAndBounded(t *testing.T) {
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
start := time.Now()
|
start := time.Now()
|
||||||
lines, cut, err := m.RecentSteps(0)
|
lines, cut, err := m.RecentSteps(row.limit)
|
||||||
if err != nil || len(lines) != row.steps || cut != row.cut || reads.Load() != row.reads {
|
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)
|
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 {
|
if maxFlight.Load() < 2 {
|
||||||
t.Errorf("%d a node: values read one at a time", row.perNode)
|
t.Errorf("%d a node: values read one at a time", row.perNode)
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user