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 063987624..581ef9c03 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/acts.go @@ -62,21 +62,38 @@ type Store interface { Get(state, key string) ([]byte, bool, error) Create(state, key string, value any) error Keys(state string) ([]string, error) - // Append adds an entry to the end of a log; Last is a key's newest entry; Entries every entry in order; Fill how - // many entries and bytes the log holds against its cap. + // Append adds an entry to the end of a log; Last is a key's newest entry; Page at most limit entries from a + // sequence on, oldest first; Fill how many entries and bytes the log holds against its cap, and its first and + // last sequence. Append(log, key string, value any) error Last(log, key string) ([]byte, bool, error) - Entries(log string) ([][]byte, error) + Page(log string, from uint64, limit int) ([]PageEntry, error) Fill(log string) (Fill, error) } -// Fill is how full a log is: its entries, its bytes and its cap in bytes. +// PageEntry is one entry of a log read: its sequence and its value. +type PageEntry struct { + Seq uint64 + Value []byte +} + +// Fill is how full a log is: its entries, its bytes and its cap in bytes, and its first and last sequence. type Fill struct { Entries uint64 `json:"entries"` Bytes uint64 `json:"bytes"` MaxBytes uint64 `json:"max-bytes"` + First uint64 `json:"first"` + Last uint64 `json:"last"` } +// What bounds one answer, so it always comes, whatever the log holds (issue 1966): the bus carries at most 1 MiB, and +// the runtime's call is bound at 30 s. A read stops at either and says so. +const ( + MaxAnswerBytes = 900 << 10 + ReadBudget = 20 * time.Second + pageSize = 1000 +) + // ErrExists is a create of a key another writer made first. var ErrExists = errors.New("the key was made by another writer first") @@ -281,7 +298,13 @@ type Listing struct { Days int `json:"days"` // Unread says which log could not be read, so what is missing is said and never read as nothing. Unread map[string]string `json:"unread,omitempty"` - Kept map[string]int `json:"kept"` + // Kept is how many entries of the log were read, newest first. + Kept map[string]int `json:"kept"` + // Complete is false when the read stopped before the window's start (Why says at what, Cut how many older entries + // of the log were not read): what is listed is the newest part, and Repeated counts that part alone. + Complete bool `json:"complete"` + Cut uint64 `json:"cut,omitempty"` + Why string `json:"why,omitempty"` // Fill is how full this seat's log is against its cap: full, it refuses the next act. Fill *Fill `json:"fill,omitempty"` } @@ -291,11 +314,12 @@ func (l *Log) HandActs(days int, cause, who string) Listing { now := l.Now() since := now.Add(-time.Duration(days) * 24 * time.Hour) out := Listing{Acts: []Act{}, Repeated: map[string]int{}, Days: days, Kept: map[string]int{}} - all, kept, err := l.read(now.Add(-max(time.Duration(days)*24*time.Hour, RepeatedWithin))) + all, kept, cut, err := l.read(now.Add(-max(time.Duration(days)*24*time.Hour, RepeatedWithin))) if err != nil { out.Unread = map[string]string{ActsLog: err.Error()} } else { out.Kept[ActsLog] = kept + out.Complete, out.Cut, out.Why = cut.why == "", cut.unread, cut.why } for _, a := range all { if now.Sub(a.At) <= RepeatedWithin && handactsclient.Repair(a) { @@ -318,20 +342,57 @@ func (l *Log) HandActs(days int, cause, who string) Listing { return out } -// read is every act of this seat's log since a moment, entry by entry, and how many it keeps. -func (l *Log) read(since time.Time) ([]Act, int, error) { - var out []Act - entries, err := l.Store.Entries(ActsLog) +// cutRead says why a read stopped before the window's start, and how many older entries it left unread. +type cutRead struct { + why string + unread uint64 +} + +// read is the acts of this seat's log since a moment, from the newest entry backwards, page by page: it stops at the +// first entry older than the moment (the log is in time order), at MaxAnswerBytes of acts or at ReadBudget, and says +// which. It also answers how many entries it read. +func (l *Log) read(since time.Time) ([]Act, int, cutRead, error) { + f, err := l.Store.Fill(ActsLog) if err != nil { - return nil, 0, err + return nil, 0, cutRead{}, err } - for _, raw := range entries { - var a Act - if json.Unmarshal(raw, &a) != nil || a.At.Before(since) { - continue + var out []Act + var size, read int + deadline := l.Now().Add(ReadBudget) + for to := f.Last; to > 0 && to >= f.First; { + if l.Now().After(deadline) { + return out, read, cutRead{"the read took more than " + ReadBudget.String(), to - f.First + 1}, nil + } + from := f.First + 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 + } + 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 { + continue + } + if a.At.Before(since) { + return out, read, cutRead{}, nil + } + if size += len(page[i].Value); 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 { + break } - a.Log = handacts.Seat - out = append(out, a) + to = from - 1 } - return out, len(entries), nil + return out, read, cutRead{}, nil } 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 9f5d2bb8c..f79237b63 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 @@ -4,6 +4,7 @@ import ( "encoding/json" "errors" "sort" + "strconv" "strings" "sync" "testing" @@ -20,6 +21,7 @@ type memStore struct { kv map[string]map[string][]byte logs map[string][]entry keysErr map[string]error + pages int } func newStore() *memStore { @@ -51,15 +53,16 @@ func (s *memStore) Last(log, key string) ([]byte, bool, error) { return nil, false, nil } -func (s *memStore) Entries(log string) ([][]byte, error) { +func (s *memStore) Page(log string, from uint64, limit int) ([]PageEntry, error) { s.mu.Lock() defer s.mu.Unlock() if err := s.keysErr[log]; err != nil { return nil, err } - var out [][]byte - for _, e := range s.logs[log] { - out = append(out, e.raw) + s.pages++ + var out []PageEntry + 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}) } return out, nil } @@ -67,7 +70,8 @@ func (s *memStore) Entries(log string) ([][]byte, error) { func (s *memStore) Fill(log string) (Fill, error) { s.mu.Lock() defer s.mu.Unlock() - return Fill{Entries: uint64(len(s.logs[log])), MaxBytes: 1 << 30}, nil + n := uint64(len(s.logs[log])) + return Fill{Entries: n, MaxBytes: 1 << 30, First: min(n, 1), Last: n}, nil } // logKeys are a log's keys in order. @@ -242,12 +246,13 @@ func TestHandActsReadsItsOwnLogAlone(t *testing.T) { } _ = store.Put(log, actKey(at), a) } - put(ActsLog, VerbRecord, "relogin", "person.jochen", time.Hour) + // The log is in time order, oldest first, as the bus keeps it. + put(ActsLog, VerbRecord, "config", "ace.mesh-config", 30*24*time.Hour) put(ActsLog, VerbRecord, "relogin", "shanks.claude-code", 3*24*time.Hour) put(ActsLog, VerbDrill, CauseDrill, "person.jochen", 2*time.Hour) - put(ActsLog, VerbWarrant, CauseOperatorAnswer, "the operator", time.Hour) put(ActsLog, VerbWarrant, CauseOperatorAnswer, "the operator", 2*time.Hour) - put(ActsLog, VerbRecord, "config", "ace.mesh-config", 30*24*time.Hour) + put(ActsLog, VerbRecord, "relogin", "person.jochen", time.Hour) + put(ActsLog, VerbWarrant, CauseOperatorAnswer, "the operator", time.Hour) put(controllerActs, VerbRecord, "never-read", "ace.old", time.Hour) store.keysErr[controllerActs] = errors.New("the controller's bucket must not be read") cases := []struct { @@ -277,10 +282,50 @@ func TestHandActsReadsItsOwnLogAlone(t *testing.T) { got.Fill.Entries != 6 { t.Errorf("the log of an act or the kept counts are not said: %+v", got.Kept) } + if got := l.HandActs(14, "", ""); !got.Complete || got.Cut != 0 || got.Why != "" { + t.Errorf("a window within the log is read whole: %+v", got) + } + // An 84,000-entry log with 3 acts in the window reads one page and stops at the window's start (issue 1966). + big, bigStore, _, bnow := newLog() + for i := 0; i < 84000; i++ { + at := bnow.Add(-20*24*time.Hour + time.Duration(i)*time.Second) + if i >= 83997 { + at = bnow.Add(-time.Duration(84000-i) * time.Minute) + } + _ = bigStore.Append(ActsLog, actKey(at), Act{At: at, Verb: VerbRecord, Cause: "flood", By: "x", Why: "x"}) + } + bigStore.pages = 0 + if got := big.HandActs(1, "", ""); len(got.Acts) != 3 || !got.Complete || bigStore.pages != 1 { + t.Errorf("the window's 3 acts of 84,000: %d acts, complete %v, %d pages", len(got.Acts), got.Complete, bigStore.pages) + } + // A fortnight of it is cut at the answer's cap: partial, said, and under the bus's 1 MiB. + for i := 0; i < 40; i++ { + at := bnow.Add(-time.Duration(40-i) * time.Second) + _ = bigStore.Append(ActsLog, actKey(at), Act{At: at, Verb: VerbRecord, Cause: "flood", By: "x", Why: strings.Repeat("w", 40<<10)}) + } + got := big.HandActs(14, "", "") + if raw, _ := json.Marshal(got); got.Complete || got.Cut == 0 || got.Why == "" || len(raw) > 1<<20 || len(got.Acts) < 20 || + 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 read that takes longer than its budget stops and says so. + slow := &Log{Store: bigStore, Logf: func(string, ...any) {}} + slowNow := *bnow + slow.Now = func() time.Time { slowNow = slowNow.Add(ReadBudget + time.Second); return slowNow } + if got := slow.HandActs(14, "", ""); got.Complete || !strings.Contains(got.Why, "took more than") { + t.Errorf("a slow read: complete %v, why %q", got.Complete, got.Why) + } store.keysErr[ActsLog] = errors.New("the bus is away") if got := l.HandActs(14, "", ""); got.Unread[ActsLog] == "" || len(got.Acts) != 0 { t.Errorf("an unreadable log: %+v", got.Unread) } + for i := 0; i < 30; i++ { + _ = store.Put(WarrantsState, "asker.ask-"+strconv.Itoa(i), Warrant{Act: strings.Repeat("w", 40<<10)}) + } + if w, err := l.Warrants(); err != nil || w.(map[string]any)["complete"] != false || w.(map[string]any)["cut"].(int) == 0 { + t.Errorf("warrants over the cap answered %v, %v", w, err) + } + store.kv[WarrantsState] = nil w, err := l.Warrants() if err != nil || w.(map[string]any)["count"] != 0 { t.Errorf("warrants answered %v, %v", w, err) diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go index f8e6e5f96..cd3a14c3e 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/main.go @@ -49,21 +49,16 @@ func (runtimeStore) Last(log, key string) ([]byte, bool, error) { return e.Value, true, nil } -func (runtimeStore) Entries(log string) ([][]byte, error) { - var out [][]byte - for from := uint64(0); ; { - page, err := stdio.Log(log).Read(from, "", 1000) - if err != nil { - return nil, err - } - for _, e := range page.Entries { - out = append(out, e.Value) - } - if page.Next == 0 { - return out, nil - } - from = page.Next +func (runtimeStore) Page(log string, from uint64, limit int) ([]PageEntry, error) { + page, err := stdio.Log(log).Read(from, "", limit) + if err != nil { + return nil, err } + out := make([]PageEntry, 0, len(page.Entries)) + for _, e := range page.Entries { + out = append(out, PageEntry{Seq: e.Sequence, Value: e.Value}) + } + return out, nil } func (runtimeStore) Fill(log string) (Fill, error) { @@ -71,7 +66,7 @@ func (runtimeStore) Fill(log string) (Fill, error) { if err != nil { return Fill{}, err } - return Fill{Entries: i.Entries, Bytes: i.Bytes, MaxBytes: i.MaxBytes}, nil + return Fill{Entries: i.Entries, Bytes: i.Bytes, MaxBytes: i.MaxBytes, First: i.First, Last: i.Last}, nil } func main() { diff --git a/modules/mesh-hand-acts/cmd/mesh-hand-acts/tools.go b/modules/mesh-hand-acts/cmd/mesh-hand-acts/tools.go index 38bf64002..3b0e4b882 100644 --- a/modules/mesh-hand-acts/cmd/mesh-hand-acts/tools.go +++ b/modules/mesh-hand-acts/cmd/mesh-hand-acts/tools.go @@ -96,12 +96,17 @@ func (l *Log) Warrants() (any, error) { } sort.Strings(keys) rows := map[string]json.RawMessage{} - for _, k := range keys { + size := 0 + for i, k := range keys { if raw, found, err := l.Store.Get(WarrantsState, k); err == nil && found { + if size += len(raw) + len(k); size > MaxAnswerBytes { + // Bounded like hand-acts: the answer always comes, and says what it left out. + return map[string]any{"warrants": rows, "count": len(rows), "complete": false, "cut": len(keys) - i}, nil + } rows[k] = raw } } - return map[string]any{"warrants": rows, "count": len(rows)}, nil + return map[string]any{"warrants": rows, "count": len(rows), "complete": true}, nil } func arg(a map[string]any, k string) string { diff --git a/modules/mesh-healers/cmd/mesh-healers/healers_test.go b/modules/mesh-healers/cmd/mesh-healers/healers_test.go index 5dff69eca..1959fb3f0 100644 --- a/modules/mesh-healers/cmd/mesh-healers/healers_test.go +++ b/modules/mesh-healers/cmd/mesh-healers/healers_test.go @@ -360,6 +360,14 @@ func TestACauseRepairedByHandTwiceIsAHealerWanted(t *testing.T) { if n, err := w.h.Seeded([]byte(`{"acts":[` + string(seatAct) + `]}`)); err != nil || n != 1 { t.Fatalf("the seat's listing was not taken: %d, %v", n, err) } + // A partial answer (the seat bounds it) seeds all the same and says it is partial, so the start never retries forever (1966). + var said []string + w.h.Logf = func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) } + if n, err := w.h.Seeded([]byte(`{"acts":[` + string(seatAct) + `],"complete":false,"why":"the answer reached 900 KiB","cut":83000}`)); err != nil || + n != 1 || len(said) != 1 || !strings.Contains(said[0], "partial: the answer reached 900 KiB") { + t.Fatalf("a partial listing: %d, %v, said %v", n, err, said) + } + w.h.Logf = func(string, ...any) {} // A drill in each log counts none. act("a3", "hand-act drill", "drill", time.Hour) drill, _ := json.Marshal(HandAct{ID: "a4", At: w.now.Add(-2 * time.Hour), Verb: "hand-act drill", Why: "x", Cause: "drill"}) diff --git a/modules/mesh-healers/cmd/mesh-healers/wanted.go b/modules/mesh-healers/cmd/mesh-healers/wanted.go index f664d19ad..0e3274a43 100644 --- a/modules/mesh-healers/cmd/mesh-healers/wanted.go +++ b/modules/mesh-healers/cmd/mesh-healers/wanted.go @@ -95,7 +95,10 @@ func (h *Healers) ActRecorded(body []byte) { // answers how many it took. func (h *Healers) Seeded(answer []byte) (int, error) { var l struct { - Acts []json.RawMessage `json:"acts"` + Acts []json.RawMessage `json:"acts"` + Complete *bool `json:"complete"` + Why string `json:"why"` + Cut uint64 `json:"cut"` } if err := json.Unmarshal(answer, &l); err != nil { return 0, fmt.Errorf("the seat's hand-acts answer is not a list of acts: %w", err) @@ -103,6 +106,11 @@ func (h *Healers) Seeded(answer []byte) (int, error) { for _, raw := range l.Acts { h.ActRecorded(raw) } + // A partial answer (the seat bounds it, issue 1966) seeds all the same: the newest acts are what a fortnight's + // count needs most, and the older ones are heard no more, so asking again would only repeat it. + if l.Complete != nil && !*l.Complete { + h.Logf("[%s] took %d of the fortnight's hand acts, the newest (partial: %s; %d older entries not read)", Module, len(l.Acts), l.Why, l.Cut) + } return len(l.Acts), nil }