diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go index 581ef9c03..0a6b641e9 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go @@ -320,6 +320,11 @@ func (l *Log) HandActs(days int, cause, who string) Listing { } else { out.Kept[ActsLog] = kept out.Complete, out.Cut, out.Why = cut.why == "", cut.unread, cut.why + // Complete is said for the window asked: a read cut after it passed the window's start lists every act of it + // (Repeated, over the fortnight, is the part that is partial, and Why says so). + if n := len(all); !out.Complete && n > 0 && !all[n-1].At.After(since) { + out.Complete = true + } } for _, a := range all { if now.Sub(a.At) <= RepeatedWithin && handactsclient.Repair(a) { @@ -367,14 +372,26 @@ func (l *Log) read(since time.Time) ([]Act, int, cutRead, error) { if to-from+1 > pageSize { from = to - pageSize + 1 } - page, err := l.Store.Page(ActsLog, from, pageSize) - if err != nil { - return nil, 0, cutRead{}, err + // A page may end early (the runtime stops one at 4 MiB): read on from its last sequence to `to`, so no entry + // between is skipped. + var page []PageEntry + for cur := from; cur <= to; { + p, err := l.Store.Page(ActsLog, cur, pageSize) + if err != nil { + return nil, 0, cutRead{}, err + } + n := len(page) + for _, e := range p { + if e.Seq >= cur && e.Seq <= to { + page = append(page, e) + } + } + if len(page) == n { + break // nothing more reaches `to`: the rest of the range is gone from the log + } + cur = page[len(page)-1].Seq + 1 } for i := len(page) - 1; i >= 0; i-- { - if page[i].Seq > to { - continue - } read++ var a Act if json.Unmarshal(page[i].Value, &a) != nil { @@ -383,10 +400,12 @@ func (l *Log) read(since time.Time) ([]Act, int, cutRead, error) { if a.At.Before(since) { return out, read, cutRead{}, nil } - if size += len(page[i].Value); size > MaxAnswerBytes { + a.Log = handacts.Seat + // The act as answered counts, with its log: the answer stays under the bus's payload. + raw, _ := json.Marshal(a) + if size += len(raw); size > MaxAnswerBytes { return out, read, cutRead{"the answer reached " + strconv.Itoa(MaxAnswerBytes>>10) + " KiB", page[i].Seq - f.First + 1}, nil } - a.Log = handacts.Seat out = append(out, a) } if from <= f.First { diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go index f79237b63..d1cc2078c 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts_test.go @@ -22,6 +22,8 @@ type memStore struct { logs map[string][]entry keysErr map[string]error pages int + // pageCap ends a page early at this many entries, as the runtime does at its byte cap. + pageCap int } func newStore() *memStore { @@ -61,6 +63,9 @@ func (s *memStore) Page(log string, from uint64, limit int) ([]PageEntry, error) } s.pages++ var out []PageEntry + if s.pageCap > 0 { + limit = min(limit, s.pageCap) + } for i := max(int(from), 1) - 1; i < len(s.logs[log]) && len(out) < limit; i++ { out = append(out, PageEntry{Seq: uint64(i + 1), Value: s.logs[log][i].raw}) } @@ -308,6 +313,29 @@ func TestHandActsReadsItsOwnLogAlone(t *testing.T) { got.Acts[0].At.Before(got.Acts[1].At) { t.Errorf("a window over the cap: complete %v, cut %d, why %q, %d acts, %d bytes", got.Complete, got.Cut, got.Why, len(got.Acts), len(raw)) } + // A page the runtime ends early skips no entry: 10 acts, pages of 4 (issue 1966 review). + pc, pcStore, _, pnow := newLog() + for i := 0; i < 10; i++ { + at := pnow.Add(-time.Duration(10-i) * time.Minute) + _ = pcStore.Append(ActsLog, actKey(at), Act{At: at, Verb: VerbRecord, Cause: "c" + strconv.Itoa(i), By: "x", Why: "x"}) + } + pcStore.pageCap = 4 + if got := pc.HandActs(14, "", ""); len(got.Acts) != 10 || !got.Complete || got.Kept[ActsLog] != 10 { + t.Errorf("early-ended pages: %d acts, complete %v, kept %v", len(got.Acts), got.Complete, got.Kept) + } + // The answer is counted as it is sent, with its log: 20,000 short acts stay under 1 MiB, and a day wholly listed is complete. + sm, smStore, _, snow := newLog() + for i := 0; i < 20000; i++ { + at := snow.Add(-time.Duration(20000-i) * time.Second) + _ = smStore.Append(ActsLog, actKey(at), Act{At: at, Verb: VerbRecord, By: "x", Why: "x"}) + } + got = sm.HandActs(14, "", "") + if raw, _ := json.Marshal(got); got.Complete || len(raw) > 1<<20 { + t.Errorf("20,000 short acts: complete %v, %d bytes", got.Complete, len(raw)) + } + if got := sm.HandActs(0, "", ""); !got.Complete { + t.Errorf("a window wholly listed said partial: %q", got.Why) + } // A read that takes longer than its budget stops and says so. slow := &Log{Store: bigStore, Logf: func(string, ...any) {}} slowNow := *bnow