Merge pull request 'mesh-hand-acts answers bounded, newest first; mesh-healers seeds from a partial answer (issue 1966 S8, recovery)' (#642) from fix/1966-s8-hand-acts-window into main

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 true when every act of the window asked was read. Why and Cut set mean the read was cut before the
+	// fortnight's start (Why says at what, Cut how many older entries were not read), and Repeated counts only the part read.
+	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,17 @@ 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
+		// 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) {
@@ -318,20 +347,71 @@ 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
+		}
+		// 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-- {
+			read++
+			var a Act
+			if json.Unmarshal(page[i].Value, &a) != nil {
+				continue
+			}
+			if a.At.Before(since) {
+				return out, read, cutRead{}, nil
+			}
+			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
+			}
+			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
 }

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,9 @@ type memStore struct {
 	kv      map[string]map[string][]byte
 	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 {
@@ -51,15 +55,19 @@ 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
+	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})
 	}
 	return out, nil
 }
@@ -67,7 +75,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 +251,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 +287,73 @@ 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 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
+	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)

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() {

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 {

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"})

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
 }