mesh-issues: resume an interrupted carry by revision and fault one issue instead of the whole start, so no kept value is lost and the tracker always starts (review of PR 228)

An open's lost answer is judged by the issue's first entry; an entry that does not
fold faults its issue on start; a value written beside the log is never folded over;
one issue is kept in at most 252 KiB so its whole operation fits a log entry.

modules/mesh-issues/cmd/mesh-issues/log_test.go

@@ -62,6 +62,9 @@ func (l *fakeLog) AppendAt(key string, value any, expect uint64) (uint64, error)
 	if err != nil {
 		return 0, err
 	}
+	if len(raw) > 256<<10 {
+		return 0, fmt.Errorf("the value for %s is %d bytes; a log entry holds at most 256 KiB", key, len(raw))
+	}
 	if last := l.lastOf(key); last != expect {
 		return 0, fmt.Errorf("the last entry of %s is %d, not %d: %w", key, last, expect, stdio.ErrChanged)
 	}
@@ -95,18 +98,6 @@ func (l *fakeLog) Read(from uint64, keys string, limit int) (*stdio.LogPage, err
 	return p, nil
 }
 
-func (l *fakeLog) Last(key string) (*stdio.LogEntry, error) {
-	l.mu.Lock()
-	defer l.mu.Unlock()
-	for i := len(l.entries) - 1; i >= 0; i-- {
-		if l.entries[i].Key == key {
-			e := l.entries[i]
-			return &e, nil
-		}
-	}
-	return nil, nil
-}
-
 // ops is every operation the log holds of one issue, oldest first.
 func (l *fakeLog) ops(t *testing.T, n uint64) []Operation {
 	t.Helper()
@@ -126,6 +117,34 @@ func (l *fakeLog) ops(t *testing.T, n uint64) []Operation {
 	return out
 }
 
+// foldCheck holds every issue the bucket folded to what its whole log adds up to, byte for byte: run after every
+// test of a harness on this log (the review of PR 228).
+func (l *fakeLog) foldCheck(t *testing.T, m *memory) {
+	l.mu.Lock()
+	entries := append([]stdio.LogEntry(nil), l.entries...)
+	l.mu.Unlock()
+	for k, v := range m.snapshot(IssuesState) {
+		var is Issue
+		if json.Unmarshal([]byte(v), &is) != nil || is.Folded == 0 {
+			continue
+		}
+		var f *Issue
+		var err error
+		for _, e := range entries {
+			if e.Key != k || e.Sequence > is.Folded {
+				continue
+			}
+			if f, err = foldEntry(f, e); err != nil {
+				t.Errorf("issue %s does not fold from its log: %v", k, err)
+				break
+			}
+		}
+		if raw, _ := json.Marshal(f); err == nil && string(raw) != v {
+			t.Errorf("issue %s in the bucket is not the fold of its log:\nbucket %s\nfold   %s", k, v, raw)
+		}
+	}
+}
+
 func (l *fakeLog) size() int {
 	l.mu.Lock()
 	defer l.mu.Unlock()
@@ -504,7 +523,7 @@ func TestTheCarryIsIdempotentAndCarriesEveryValueKept(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	if c.Issues != 2 || c.Values != 4 || !c.History {
+	if c.Issues != 2 || c.Values != 4 || len(c.Faults) != 0 {
 		t.Fatalf("carried %+v", c)
 	}
 	ops := h.log.ops(t, 300)
@@ -594,3 +613,207 @@ func TestAnOpenOverAnIssueKeptBeforeTheSwitchOverwritesNothing(t *testing.T) {
 		t.Fatalf("365 is %+v", is)
 	}
 }
+
+// titlesOf is each operation of an issue's log as its kind and, for one holding the issue whole, its title.
+func titlesOf(t *testing.T, h *harness, n uint64) string {
+	t.Helper()
+	var out []string
+	for _, op := range h.log.ops(t, n) {
+		if op.Issue != nil {
+			out = append(out, op.Op+":"+op.Issue.Title)
+		} else {
+			out = append(out, op.Op)
+		}
+	}
+	return strings.Join(out, " ")
+}
+
+// §6, the review of PR 228: a carry that fails partway, then a second carry, leaves every kept value carried
+// once, oldest first, and each carry says what it really carried.
+func TestAnInterruptedCarryResumesAfterTheLastValueCarried(t *testing.T) {
+	h := newHarness(t)
+	keptBefore(t, h, 300, "v1", "v2", "v3", "v4")
+	// The bus goes away after the first value is appended.
+	h.log.between = func() { h.log.between = func() { h.log.fail = errors.New("the bus went away") } }
+	c, err := h.Carry()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if c.Values != 1 || len(c.Faults) != 1 || !strings.Contains(c.Said, "could not be carried") ||
+		!strings.Contains(c.Faults[0], "the bus went away") {
+		t.Fatalf("the first carry said %+v", c)
+	}
+	if is, _ := h.Get(300); is.Folded != 0 || is.Title != "v4" {
+		t.Fatalf("after the failure the bucket holds %+v", is)
+	}
+	// A write meanwhile is refused, and changes nothing.
+	if _, err := h.Note(300, "while the bus is away"); err == nil || !strings.Contains(err.Error(), "not carried") {
+		t.Fatalf("a note on an issue half carried: %v", err)
+	}
+	h.log.fail = nil
+	c, err = h.Carry()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if c.Values != 3 || c.Issues != 1 || len(c.Faults) != 0 {
+		t.Fatalf("the second carry said %+v", c)
+	}
+	if got := titlesOf(t, h, 300); got != "carried:v1 carried:v2 carried:v3 carried:v4" {
+		t.Fatalf("the log of 300: %s", got)
+	}
+	if is, _ := h.Get(300); is.Title != "v4" || is.Folded != 4 {
+		t.Fatalf("300 is %+v", is)
+	}
+}
+
+// §6, the review of PR 228: one issue that cannot be carried is a fault, named, and the start carries the others,
+// folds, raises the counter and loads memory. Writes to that issue are refused with the reason.
+func TestOneIssueThatCannotBeCarriedDoesNotStopTheStart(t *testing.T) {
+	h := newHarness(t)
+	keptBefore(t, h, 300, "small")
+	// Kept before the switch at the bucket's limit: whole, in a carried operation, past what one log entry holds.
+	h.mem.put(IssuesState, "301", Issue{Number: 301, Title: "huge", Status: StatusOpen, Report: strings.Repeat("x", 256<<10-200),
+		OpenedBy: "x", Opening: "o301"})
+	keptBefore(t, h, 302, "also small")
+	s, err := h.Start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if s.Carried.Issues != 2 || len(s.Faults) != 1 || !strings.Contains(s.Faults[0], "issue 301") ||
+		!strings.Contains(s.Faults[0], "256 KiB") || !strings.Contains(s.Said, "issue 301") {
+		t.Fatalf("started %+v", s)
+	}
+	if !h.view().ready() {
+		t.Fatal("memory was not loaded")
+	}
+	l, err := h.List(ListAsk{})
+	if err != nil || l.Matched != 3 {
+		t.Fatalf("listed %+v, %v", l, err)
+	}
+	if _, err := h.Note(301, "a note"); err == nil || !strings.Contains(err.Error(), "256 KiB") {
+		t.Fatalf("a note on the issue not carried: %v", err)
+	}
+	if is, _ := h.Get(301); is.Title != "huge" || is.Folded != 0 || len(is.Notes) != 0 {
+		t.Fatal("the issue not carried was changed")
+	}
+	if is, _ := h.Get(302); is.Folded == 0 {
+		t.Fatal("302 was not carried")
+	}
+}
+
+// §6, the review of PR 228: two instances carrying the same issue at the same moment carry each value once.
+func TestTwoInstancesCarryingOneIssueAtOnceCarryEachValueOnce(t *testing.T) {
+	h := newHarness(t)
+	keptBefore(t, h, 300, "v1", "v2", "v3")
+	other := h.another(t)
+	var otherSaid *Carried
+	h.log.between = func() {
+		c, err := other.Carry()
+		if err != nil {
+			t.Errorf("the other carry: %v", err)
+		}
+		otherSaid = c
+	}
+	c, err := h.Carry()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got := titlesOf(t, h, 300); got != "carried:v1 carried:v2 carried:v3" {
+		t.Fatalf("the log of 300: %s", got)
+	}
+	if otherSaid.Values+c.Values != 3 || len(c.Faults)+len(otherSaid.Faults) != 0 {
+		t.Fatalf("this carry %+v, the other %+v", c, otherSaid)
+	}
+}
+
+// §6, the review of PR 228: a value written beside the log after the carry, by an instance from before the switch,
+// is never overwritten by the fold on start; it is carried as one more operation.
+func TestAValueWrittenBesideTheLogIsCarriedNotOverwritten(t *testing.T) {
+	h := newHarness(t)
+	n := h.open(t, "carried", "hq")
+	if _, err := h.Note(n, "after the switch"); err != nil {
+		t.Fatal(err)
+	}
+	// An instance from before the switch writes the bucket: no folded.
+	is, _ := h.Get(n)
+	is.Folded, is.Title = 0, "written beside the log"
+	h.mem.put(IssuesState, keyOf(n), is)
+	// Its carry fails at this start: the fold is not written over it.
+	h.log.failNext = 1
+	s, err := h.another(t).Start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if got, _ := h.Get(n); got.Title != "written beside the log" || got.Folded != 0 || len(s.Faults) != 2 {
+		t.Fatalf("issue %d is %+v after a start whose carry failed (faults %v)", n, got, s.Faults)
+	}
+	// The next start carries it.
+	if _, err := h.another(t).Start(); err != nil {
+		t.Fatal(err)
+	}
+	got, _ := h.Get(n)
+	if got.Title != "written beside the log" || got.Folded != 3 || len(got.Notes) != 1 {
+		t.Fatalf("issue %d is %+v", n, got)
+	}
+	if ops := titlesOf(t, h, n); ops != "opened:carried noted carried:written beside the log" {
+		t.Fatalf("the log: %s", ops)
+	}
+}
+
+// §3, the review of PR 228: an entry that cannot be folded faults its issue on start, and nothing folded past it
+// is written.
+func TestAnEntryThatCannotBeFoldedLeavesItsIssueAsItIs(t *testing.T) {
+	h := newHarness(t)
+	n := h.open(t, "fine")
+	m := h.open(t, "broken")
+	if _, err := h.Note(m, "kept"); err != nil {
+		t.Fatal(err)
+	}
+	h.log.mu.Lock()
+	for i, e := range h.log.entries {
+		if e.Key == keyOf(m) && e.Sequence == 2 {
+			h.log.entries[i].Value = json.RawMessage(`{"op":"noted","add":{"notes":"not a list"}}`)
+		}
+	}
+	h.log.mu.Unlock()
+	before := h.mem.snapshot(IssuesState)[keyOf(m)]
+	h.mem.empty(IssuesState)
+	h.mem.put(IssuesState, keyOf(m), json.RawMessage(before))
+	s, err := h.another(t).Start()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if s.Rebuilt != 1 || len(s.Faults) != 1 || !strings.Contains(s.Faults[0], fmt.Sprintf("issue %d cannot be folded", m)) {
+		t.Fatalf("started %+v", s)
+	}
+	if h.mem.snapshot(IssuesState)[keyOf(m)] != before {
+		t.Fatal("the broken issue's bucket value was written over")
+	}
+	if is, _ := h.Get(n); is.Title != "fine" {
+		t.Fatalf("%d is %+v", n, is)
+	}
+	// The broken entry is the test's doing: the fold check would find it, so take the issue out of the bucket.
+	h.mem.empty(IssuesState)
+}
+
+// §2, the review of PR 228: an open whose answer was lost, its issue already changed by another operation, is
+// still known as its own by its first entry.
+func TestALostOpenIsKnownByItsFirstEntryNotItsLast(t *testing.T) {
+	h := newHarness(t)
+	h.log.lost = errors.New("no answer in 30 s")
+	// After the first operation lands and its answer is lost, another operation follows before the retry.
+	h.log.between = func() {
+		h.log.between = func() {
+			h.log.mu.Lock()
+			h.log.entries = append(h.log.entries, stdio.LogEntry{Sequence: uint64(len(h.log.entries) + 1), Key: "365",
+				Value: json.RawMessage(`{"op":"triaged","by":"triage","set":{"severity":"\"minor\""}}`)})
+			h.log.mu.Unlock()
+		}
+	}
+	o, err := h.Open(OpenAsk{Title: "lost and followed", Report: "r"})
+	if err != nil || o.Number != 365 {
+		t.Fatalf("%+v %v", o, err)
+	}
+	// The entry added by hand is not one the tracker folded: take the issue out before the fold check.
+	h.mem.empty(IssuesState)
+}