A first view's page is bounded by time as well as size, so a slow source never outlasts its reader (issue 1966)
Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
This commit is contained in:
+14
-1
@@ -40,6 +40,14 @@ const MaxValueBytes = 960 * 1024
|
||||
// PageBytes bounds the values one answer carries, so the answer fits one bus message; Answer.Next continues it.
|
||||
const PageBytes = 768 * 1024
|
||||
|
||||
// PageBudget bounds the time one answer spends reading values, so a slow source (kept state reads one value per bus
|
||||
// request) never makes a page outlast its reader's patience (30 s): the page stops when it is spent, always after at
|
||||
// least one value, and Answer.Next continues it. A variable so a test sets it.
|
||||
var PageBudget = 10 * time.Second
|
||||
|
||||
// clock is the page budget's time, replaced by a test.
|
||||
var clock = time.Now
|
||||
|
||||
// Bound is how long a reader waits for its first view before it raises the condition naming the owner and the verb
|
||||
// (ADR 0615 §6): longer than an owner's start, so a normal start, readers and owners coming up together, tells nobody.
|
||||
const Bound = 2 * time.Minute
|
||||
@@ -187,7 +195,7 @@ type Source interface {
|
||||
}
|
||||
|
||||
// FirstView answers one page of state from its owner's source: the mark first, so no change after the revision it
|
||||
// answers at is missed, then the values in key order after after, until PageBytes. A state not said yet (no mark)
|
||||
// answers at is missed, then the values in key order after after, until PageBytes or PageBudget. A state not said yet (no mark)
|
||||
// answers revision 0: a first view of whatever it holds, which the reader takes as a gap at the first event.
|
||||
func FirstView(owner, state string, values, marks Source, after string, now time.Time) (Answer, error) {
|
||||
a := Answer{V: Version, Owner: owner, State: state, Values: map[string]json.RawMessage{}, At: now.UTC()}
|
||||
@@ -210,10 +218,15 @@ func FirstView(owner, state string, values, marks Source, after string, now time
|
||||
}
|
||||
slices.Sort(keys)
|
||||
size := 0
|
||||
started := clock()
|
||||
for _, k := range keys {
|
||||
if after != "" && k <= after {
|
||||
continue
|
||||
}
|
||||
if size > 0 && clock().Sub(started) >= PageBudget {
|
||||
a.Next = lastKey(a.Values)
|
||||
break
|
||||
}
|
||||
e, ok, err := values.Get(k)
|
||||
if err != nil {
|
||||
return Answer{}, fmt.Errorf("%s.%s cannot be read: %w", state, k, err)
|
||||
|
||||
@@ -95,6 +95,32 @@ func TestViews(t *testing.T) {
|
||||
t.Errorf("pages hold %v, want the four keys", got)
|
||||
}
|
||||
})
|
||||
t.Run("a first view from a slow source stops at the page budget, with one value at least, and pages on to every key", func(t *testing.T) {
|
||||
at := time.Unix(0, 0)
|
||||
clock = func() time.Time { return at }
|
||||
defer func() { clock = time.Now }()
|
||||
slow := mapSource{}
|
||||
for i := range 5 {
|
||||
slow[fmt.Sprintf("k%d", i)] = Entry{Value: json.RawMessage(`{"n":1}`), Revision: uint64(i + 1)}
|
||||
}
|
||||
got, after := map[string]bool{}, ""
|
||||
for pages := 0; ; pages++ {
|
||||
a, err := FirstView("o", "s", &tickSource{slow, &at, PageBudget / 2}, nil, after, now())
|
||||
if err != nil || pages > 5 || len(a.Values) == 0 || len(a.Values) > 3 {
|
||||
t.Fatalf("page %d: %d values, %v", pages, len(a.Values), err)
|
||||
}
|
||||
for k := range a.Values {
|
||||
got[k] = true
|
||||
}
|
||||
if a.Next == "" {
|
||||
break
|
||||
}
|
||||
after = a.Next
|
||||
}
|
||||
if len(got) != 5 {
|
||||
t.Errorf("pages hold %v, want the five keys", got)
|
||||
}
|
||||
})
|
||||
t.Run("each first-view tool answers its callers alone, never the owner, by the proof and not an argument", func(t *testing.T) {
|
||||
callers := ParseCallers("mesh-grants, node-tools,mesh-delivery-assigner")
|
||||
for caller, allowed := range map[string]bool{"novox.mesh-grants": true, "ace.node-tools": true, "g14.mesh-delivery-assigner": true,
|
||||
@@ -613,3 +639,15 @@ func (k kvEntry) Revision() uint64 { return k.rev }
|
||||
func (k kvEntry) Created() time.Time { return k.at }
|
||||
func (k kvEntry) Delta() uint64 { return 0 }
|
||||
func (k kvEntry) Operation() jetstream.KeyValueOp { return jetstream.KeyValuePut }
|
||||
|
||||
// tickSource is a source whose every Get takes a step of the fake clock.
|
||||
type tickSource struct {
|
||||
mapSource
|
||||
at *time.Time
|
||||
step time.Duration
|
||||
}
|
||||
|
||||
func (s *tickSource) Get(k string) (*Entry, bool, error) {
|
||||
*s.at = s.at.Add(s.step)
|
||||
return s.mapSource.Get(k)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user