diff --git a/modules/mesh-issues/README.md b/modules/mesh-issues/README.md index 191feb88..7b36ef7a 100644 --- a/modules/mesh-issues/README.md +++ b/modules/mesh-issues/README.md @@ -167,7 +167,7 @@ It is a verb on the seat, callable by any agent, because it can do only this: `imported` event. What it writes comes from hq's `main` as the records module holds it, which is reviewed in pull requests. - **What it keeps is bounded** as every verb's text is: the title 200 bytes, the report 64 KiB, one issue - 256 KiB; a report that trips a credential's shape is refused. An entry of `located-in` or `fixed-by` hq + 252 KiB; a report that trips a credential's shape is refused. An entry of `located-in` or `fixed-by` hq wrote is prose at times, so it is taken within a reason's bound (2000 bytes), not a line's. - **An owner hq named that the registered list does not know is kept as written and marked**, in `located-in-unverified` on the issue and `unverified-owners` in the answer, never refused: the list is @@ -269,15 +269,22 @@ emptied bucket is rebuilt whole — and raises the issue counter to the highest below. Until then a `list` reads the bucket and a write carries the issue it changes first. **The switch** (ADR 0300 §6) is the carry, run on every start and before any write of an issue not carried: -for every issue the bucket holds with no `folded`, each value the bucket still keeps of it, oldest first, is -appended as a `carried` operation holding the whole issue, and the fold written. It is idempotent: an issue the -log already begins is not carried again (where the bucket holds something other than its fold — written by an -instance from before the switch — that value is carried as one more operation). The values kept are read through -the runtime's `mesh/state.history` (hq ADR 0307, the SDK's `State("issues").History`), oldest first; a deletion -among them is passed over and counted in the start line, since the tracker never deletes an issue and every -value after one holds the issue whole. A store that reads no history gives each issue's current value alone, -and the start line says so. It runs on start rather than as a verb because it must precede every -write, and a verb could be called late or never. +for every issue the bucket holds with no `folded`, each value the bucket still keeps of it, read through the +runtime's `mesh/state.history` (hq ADR 0307), is appended oldest first as a `carried` operation holding the +whole issue and the bucket's revision of that value, and the fold written. It resumes rather than restarts: a +kept value at or below the highest revision a carried operation holds, or one the tracker wrote itself (it has +`folded`), is known to the log, so an interrupted carry, or two instances carrying one issue at once, carry each +value once. Matched by revision, not content, because the same content may be kept twice. A value written beside +the log after the carry, by an instance from before the switch, is carried as one more operation and is never +written over by a fold. A deletion among the kept values is passed over and counted, since the tracker never +deletes an issue and every value after one holds the issue whole. + +An issue that cannot be carried — the log refusing it, or a value this module cannot read — is a fault, named +with its reason in the start line; the others are carried and the start finishes. That issue stays in the +bucket as it was, read and listed as before, and every write to it is refused with the same reason until it +can be carried, so no operation reaches the log before its carry. One issue is kept in at most 252 KiB, so that +the operation holding it whole fits one entry of the log (256 KiB). An entry of the log that cannot be folded +faults its issue on start: nothing folded past it is written, and its bucket value is left as it is. **In memory** the tracker holds every folded issue, kept current by the watch and its own writes, and a full-text index of title, report and notes, ranked by BM25 (title words weigh three), rebuilt on every start. @@ -286,7 +293,7 @@ the 500 rows one answer holds. The index is the module's own: a few thousand iss library of dozens of modules. The other states: `counters`, one value per key (`issue`, `decision`, `research`); `grooming`, the backlog's -pass; `reservations`, one key per reserved number. None expires. One issue is kept in at most 256 KiB; a note +pass; `reservations`, one key per reserved number. None expires. One issue is kept in at most 252 KiB; a note that would pass it is refused. **The events announce; the log is the audit trail.** Every change is still an event: `mesh-issues.opened`, diff --git a/modules/mesh-issues/cmd/mesh-issues/changes.go b/modules/mesh-issues/cmd/mesh-issues/changes.go index 20b8a482..0fc5af2f 100644 --- a/modules/mesh-issues/cmd/mesh-issues/changes.go +++ b/modules/mesh-issues/cmd/mesh-issues/changes.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "sort" + "strings" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -17,11 +18,10 @@ import ( const ChangesLog = "changes" // Log is the module's log as the runtime offers it (novox/hq ADR 0297): an append at the key's last sequence, -// refused with stdio.ErrChanged when another entry came in between, a read in pages, and a key's last entry. +// refused with stdio.ErrChanged when another entry came in between, and a read in pages. type Log interface { AppendAt(key string, value any, expect uint64) (uint64, error) Read(from uint64, keys string, limit int) (*stdio.LogPage, error) - Last(key string) (*stdio.LogEntry, error) } // runtimeLog is Log through the runtime that launched this bundle. @@ -33,7 +33,6 @@ func (runtimeLog) AppendAt(key string, value any, expect uint64) (uint64, error) func (runtimeLog) Read(from uint64, keys string, limit int) (*stdio.LogPage, error) { return stdio.Log(ChangesLog).Read(from, keys, limit) } -func (runtimeLog) Last(key string) (*stdio.LogEntry, error) { return stdio.Log(ChangesLog).Last(key) } // errNoLog is a tracker started without its log: the terminal's dry run, which writes nothing. var errNoLog = errors.New("this tracker runs without its log, so it writes nothing") @@ -61,6 +60,9 @@ type Operation struct { Set map[string]json.RawMessage `json:"set,omitempty"` Add map[string]json.RawMessage `json:"add,omitempty"` Unset []string `json:"unset,omitempty"` + // Revision is the bucket's revision of the value a carried operation holds: how a carry knows which kept + // values the log holds already. + Revision uint64 `json:"revision,omitempty"` } // Act is who changes an issue and as what kind of operation. @@ -269,30 +271,44 @@ func (t *Tracker) begin(is *Issue, act Act) (uint64, error) { return seq, nil } if errors.Is(err, stdio.ErrChanged) { - last, lerr := t.Log.Last(keyOf(is.Number)) + first, lerr := t.firstOf(is.Number) if lerr != nil { return 0, fmt.Errorf("the log refused issue %d's first operation and could not be read back (%v): %w", is.Number, lerr, errBegun) } - if mine(last, is.Opening) { - return last.Sequence, nil + if mine(first, is.Opening) { + return first.Sequence, nil } return 0, errBegun } } // An append that timed out may still have landed: read back before saying it did not. - last, lerr := t.Log.Last(keyOf(is.Number)) + first, lerr := t.firstOf(is.Number) switch { case lerr != nil: return 0, fmt.Errorf("%v, and the log could not be read back (%v): %w", err, lerr, errUnsure) - case mine(last, is.Opening): - return last.Sequence, nil - case last != nil: + case mine(first, is.Opening): + return first.Sequence, nil + case first != nil: return 0, errBegun } return 0, err } +// firstOf is an issue's first entry in the log, the one that begins it, or nil when it has none. The first, not +// the last: an open whose answer was lost may have been followed by another operation on it already, such as the +// triage's. +func (t *Tracker) firstOf(n uint64) (*stdio.LogEntry, error) { + p, err := t.Log.Read(1, keyOf(n), 1) + if err != nil { + return nil, err + } + if len(p.Entries) == 0 { + return nil, nil + } + return &p.Entries[0], nil +} + // errUnsure is a first operation the log may have taken: its answer was lost, and the log could not be read back. var errUnsure = errors.New("it may have been taken") @@ -317,8 +333,9 @@ func (t *Tracker) change(n uint64, act Act, f func(*Issue) error) (*Issue, error } if is.Folded == 0 { // Kept before the switch, and not carried yet: carried first, so the log begins it (ADR 0300 §6). - if _, err := t.carryOne(n); err != nil { - return nil, err + if _, _, err := t.carryOne(n); err != nil { + return nil, fmt.Errorf("issue %d is kept from before the tracker's log and is not carried into it yet, "+ + "so nothing is changed: %v", n, err) } continue } @@ -334,7 +351,7 @@ func (t *Tracker) change(n uint64, act Act, f func(*Issue) error) (*Issue, error return nil, err } if len(raw) > MostIssueBytes { - return nil, fmt.Errorf("issue %d would be %d bytes; one issue is kept in at most %d (256 KiB). Nothing "+ + return nil, fmt.Errorf("issue %d would be %d bytes; one issue is kept in at most %d (252 KiB). Nothing "+ "was changed: say it shorter, or in a document the issue links", n, len(raw), MostIssueBytes) } op, changed, err := operation(act, t.now(), before, is) @@ -464,133 +481,133 @@ func clone(is *Issue) (*Issue, error) { return &c, json.Unmarshal(raw, &c) } -// History is a Store that reads every value a key still keeps, oldest first, as the runtime serves it -// (novox/hq ADR 0307, mesh-sdk State(...).History): each a put or a delete. The carry reads it where the store -// offers it, and takes each issue's current value alone where it does not, and says so. -type History interface { - History(state, key string) ([]stdio.StateRevision, error) -} - -// Carried is what the carry did (ADR 0300 §6). +// Carried is what the carry did (ADR 0300 §6), as it happened: the values it appended, not the ones it meant to. type Carried struct { - // Issues is how many issues it carried, Values how many values of them; Already how many issues the log - // already began, which it left as they are. + // Issues is how many issues it appended values of, Values how many values; Already how many issues the log + // already began whole, which it left as they are. Issues int `json:"issues"` Values int `json:"values"` Already int `json:"already"` - // History says whether each value the bucket kept was carried, or the current value alone. - History bool `json:"history"` // Deletes is how many deletions among the kept values were passed over: an issue is never deleted by the // tracker, and each value after one holds the issue whole, so the fold needs none. - Deletes int `json:"deletes,omitempty"` - Said string `json:"said"` + Deletes int `json:"deletes,omitempty"` + // Faults is each issue that could not be carried, and why. It stays in the bucket as it was, readable and + // listed; every write to it is refused with the same reason until it can be carried, so nothing reaches the + // log before its carry. + Faults []string `json:"faults,omitempty"` + Said string `json:"said"` } -// carryOne carries one issue the bucket holds and the bucket has not folded from the log: every value the -// bucket still keeps of it, oldest first, as `carried` operations holding the whole issue, then the fold -// written to the bucket. An issue the log already begins is not carried again: when the bucket holds what the -// log adds up to, the fold is written; when it holds something else, written beside the log, that value is -// carried as one more operation. Answers how many values it appended. -func (t *Tracker) carryOne(n uint64) (int, error) { +// carryOne carries one issue the bucket has not folded from the log: each value the bucket still keeps of it +// (novox/hq ADR 0307), oldest first, as `carried` operations holding the whole issue and the bucket's revision +// of that value, then the fold written to the bucket. Answers how many values it appended and how many +// deletions it passed over. +// +// It resumes rather than restarts. A value is known to the log when its revision is at or below the highest +// revision a carried operation of the issue holds, or when it is a value the tracker wrote itself (it carries +// `folded`): matched by revision, not by content, because the same content may be kept twice (a write that +// changed nothing) and a revision names one value exactly. So an interrupted carry, or another instance carrying +// the same issue at the same moment, carries each value once; a value written beside the log after the carry, +// by an instance from before the switch, is carried as one more operation and so is never overwritten. +func (t *Tracker) carryOne(n uint64) (int, int, error) { + appended := 0 for i := 0; i < issueAttempts; i++ { is, rev, err := read(t.Store, n) if err != nil { - return 0, err + return appended, 0, err } if is.Folded != 0 { - return 0, nil + return appended, 0, nil } entries, err := t.readKey(n, 1) if err != nil { - return 0, err + return appended, 0, err } - var values []*Issue - var expect uint64 var folded *Issue - if len(entries) > 0 { - for _, e := range entries { - if folded, err = foldEntry(folded, e); err != nil { - return 0, fmt.Errorf("issue %d cannot be folded from its log: %v", n, err) - } + var expect, known uint64 + for _, e := range entries { + if folded, err = foldEntry(folded, e); err != nil { + return appended, 0, fmt.Errorf("issue %d cannot be folded from its log: %v", n, err) } - expect = folded.Folded - if same(folded, is) { + expect = e.Sequence + var op Operation + if json.Unmarshal(e.Value, &op) == nil && op.Op == OpCarried { + known = max(known, op.Revision) + } + } + kept, err := t.Store.History(IssuesState, keyOf(n)) + if err != nil { + return appended, 0, fmt.Errorf("the values the bucket keeps of issue %d could not be read: %v", n, err) + } + type value struct { + issue *Issue + revision uint64 + } + var values []value + deletes := 0 + for _, e := range kept { + if e.Op == "delete" { + deletes++ + continue + } + var v Issue + if err := json.Unmarshal(e.Value, &v); err != nil { + return appended, deletes, fmt.Errorf("the value of issue %d at revision %d is in a form this module "+ + "cannot read (%v): not carried, so that nothing kept is left out", n, e.Revision, err) + } + switch { + case v.Folded != 0: + // The tracker's own fold: the log holds it already, and whatever came before it. values = nil - } else { - values = []*Issue{is} + known = max(known, e.Revision) + case e.Revision > known: + values = append(values, value{&v, e.Revision}) + } + } + if len(kept) == 0 || kept[len(kept)-1].Revision < rev { + // The value read is newer than the history read before it: carried too. + if rev > known { + values = append(values, value{is, rev}) } - } else if values, err = t.kept(n, is); err != nil { - return 0, err } - appendedNow := 0 raced := false for _, v := range values { - v.Folded = 0 - seq, err := t.appendOp(n, &Operation{Op: OpCarried, By: Seat, At: t.now(), Issue: v}, expect) + v.issue.Folded = 0 + seq, err := t.appendOp(n, &Operation{Op: OpCarried, By: Seat, At: t.now(), Issue: v.issue, Revision: v.revision}, expect) if errors.Is(err, stdio.ErrChanged) { raced = true break } if err != nil { - return appendedNow, fmt.Errorf("issue %d could not be carried into the log (%v); nothing is lost: the "+ - "bucket still holds it, and it is carried again before its next change", n, err) + return appended, deletes, fmt.Errorf("issue %d could not be carried into the log (%v); %d of its %d "+ + "values not carried yet were carried. Nothing is lost: the bucket still holds it, and the carry "+ + "resumes after the last value carried, at the next start or before its next change", n, err, appended, len(values)) } - appendedNow++ + appended++ expect = seq - if folded, err = fold(nil, &Operation{Op: OpCarried, Issue: v}, seq); err != nil { - return appendedNow, err + if folded, err = fold(nil, &Operation{Op: OpCarried, Issue: v.issue}, seq); err != nil { + return appended, deletes, err } } if raced { - // Another carrier was first: read again; what it appended is recognised as the log's. + // Another carrier was first: read again; what it appended is known by its revisions. continue } + if folded == nil { + return appended, deletes, fmt.Errorf("issue %d keeps no value to carry", n) + } if _, err := t.Store.Update(IssuesState, keyOf(n), folded, rev); err != nil { if errors.Is(err, stdio.ErrChanged) { - // Written meanwhile (by a writer from before the switch, or a carrier): read again. - if appendedNow > 0 { - return appendedNow, nil - } + // Written meanwhile, by a carrier or an instance from before the switch: read again. continue } - return appendedNow, fmt.Errorf("issue %d was carried into the log and its bucket could not be written (%v): "+ + return appended, deletes, fmt.Errorf("issue %d was carried into the log and its bucket could not be written (%v): "+ "it is folded at the next change or start", n, err) } t.remember(folded) - return appendedNow, nil + return appended, deletes, nil } - return 0, fmt.Errorf("issue %d was written by another writer %d times while it was carried", n, issueAttempts) -} - -// kept is every value the bucket keeps of an issue, oldest first, ending with the one read; the current one -// alone where the store reads no history. A deletion among them is passed over and counted: the tracker never -// deletes an issue, the fold has no deletion, and the next value holds the issue whole again. -func (t *Tracker) kept(n uint64, current *Issue) ([]*Issue, error) { - h, ok := t.Store.(History) - if !ok { - return []*Issue{current}, nil - } - old, err := h.History(IssuesState, keyOf(n)) - if err != nil { - return nil, fmt.Errorf("the values the bucket keeps of issue %d could not be read: %v", n, err) - } - var out []*Issue - for _, e := range old { - if e.Op == "delete" { - t.carryDeletes.Add(1) - continue - } - var v Issue - if json.Unmarshal(e.Value, &v) != nil { - t.Logf("[mesh-issues] the value of issue %d at revision %d is in a form this module cannot read: not carried", n, e.Revision) - continue - } - out = append(out, &v) - } - if len(out) == 0 || !same(out[len(out)-1], current) { - out = append(out, current) - } - return out, nil + return appended, 0, fmt.Errorf("issue %d was written by another writer %d times while it was carried", n, issueAttempts) } // same says whether two issues hold the same, whatever sequence each was folded at. @@ -608,16 +625,15 @@ func same(a, b *Issue) bool { return true } -// Carry carries every issue the bucket holds and the log does not begin (ADR 0300 §6). Run again, it carries -// nothing more: an issue is carried once, before any other operation of it goes to the log. +// Carry carries every issue the bucket holds and has not folded from the log (ADR 0300 §6). An issue that +// cannot be carried is a fault, named with its reason, and the others are carried. Run again, it carries +// nothing more, and resumes an issue whose carry was interrupted after the last value carried. func (t *Tracker) Carry() (*Carried, error) { keys, err := t.Store.Keys(IssuesState) if err != nil { return nil, fmt.Errorf("the issues kept cannot be listed: %v", err) } - _, history := t.Store.(History) - out := &Carried{History: history} - deletes := t.carryDeletes.Load() + out := &Carried{} for _, k := range keys { n, ok := numberOfKey(k) if !ok { @@ -628,37 +644,39 @@ func (t *Tracker) Carry() (*Carried, error) { continue } if err != nil { - return out, err + out.Faults = append(out.Faults, err.Error()) + continue } if is.Folded != 0 { out.Already++ continue } - v, err := t.carryOne(n) + v, d, err := t.carryOne(n) out.Values += v - if err != nil { - return out, err - } + out.Deletes += d if v > 0 { out.Issues++ - } else { + } else if err == nil { out.Already++ } + if err != nil { + t.Logf("[mesh-issues] %v", err) + out.Faults = append(out.Faults, err.Error()) + } } - out.Deletes = int(t.carryDeletes.Load() - deletes) - switch { - case out.Issues == 0: - out.Said = fmt.Sprintf("Nothing to carry: the log begins every issue of the bucket (%d).", out.Already) - case history: + if out.Values == 0 { + out.Said = fmt.Sprintf("Carried nothing: the log begins every issue of the bucket it could (%d).", out.Already) + } else { out.Said = fmt.Sprintf("Carried %d values of %d issues into the log, each value the bucket kept, oldest first; "+ "%d issues were there already.", out.Values, out.Issues, out.Already) - if out.Deletes > 0 { - out.Said += fmt.Sprintf(" %d deletions among the kept values were passed over: each value after one holds "+ - "the issue whole.", out.Deletes) - } - default: - out.Said = fmt.Sprintf("Carried %d issues into the log, %d values: the current value of each, as the runtime "+ - "reads no older value of a key; %d issues were there already.", out.Issues, out.Values, out.Already) + } + if out.Deletes > 0 { + out.Said += fmt.Sprintf(" %d deletions among the kept values were passed over: each value after one holds "+ + "the issue whole.", out.Deletes) + } + if len(out.Faults) > 0 { + out.Said += fmt.Sprintf(" %d issues could not be carried, and stay as the bucket holds them, every write to "+ + "them refused until they can be: %s.", len(out.Faults), strings.Join(out.Faults, "; ")) } return out, nil } @@ -688,11 +706,17 @@ func (t *Tracker) Start() (*Started, error) { return out, errNoLog } carried, err := t.Carry() - out.Carried = carried if err != nil { - return out, fmt.Errorf("the carry did not finish (%v); writes carry each issue before they change it", err) + return out, fmt.Errorf("the carry could not start (%v); writes carry each issue before they change it", err) + } + out.Carried = carried + for _, f := range carried.Faults { + out.Faults = append(out.Faults, "not carried: "+f) } folds := map[uint64]*Issue{} + // broken is each issue an entry of which could not be folded: none of its later entries is folded on top, + // and its bucket value is left as it is, so no fold that skipped an operation is ever written. + broken := map[uint64]bool{} var highest uint64 for from := uint64(1); from != 0; { p, err := t.Log.Read(from, "", logPage) @@ -707,16 +731,22 @@ func (t *Tracker) Start() (*Started, error) { continue } highest = max(highest, n) + if broken[n] { + continue + } f, err := foldEntry(folds[n], e) if err != nil { - out.Faults = append(out.Faults, fmt.Sprintf("issue %d: %v", n, err)) + out.Faults = append(out.Faults, fmt.Sprintf("issue %d cannot be folded from its log, and its bucket "+ + "value is left as it is: %v", n, err)) + broken[n] = true + delete(folds, n) continue } folds[n] = f } from = p.Next } - out.Issues = len(folds) + out.Issues = len(folds) + len(broken) numbers := make([]uint64, 0, len(folds)) for n := range folds { numbers = append(numbers, n) @@ -736,6 +766,13 @@ func (t *Tracker) Start() (*Started, error) { case err != nil: out.Faults = append(out.Faults, err.Error()) continue + case is.Folded == 0: + // Written beside the log, by an instance from before the switch, after the carry, and not carried again + // (its carry is in the faults): the fold would overwrite that write. Left for its carry, which appends it + // as one more operation before the issue's next change. + out.Faults = append(out.Faults, fmt.Sprintf("issue %d holds a value written beside the log and not carried "+ + "yet; it is not refolded over", n)) + continue case is.Folded < f.Folded: if _, err := t.Store.Update(IssuesState, keyOf(n), f, rev); err != nil && !errors.Is(err, stdio.ErrChanged) { out.Faults = append(out.Faults, fmt.Sprintf("issue %d could not be refolded in the bucket: %v", n, err)) diff --git a/modules/mesh-issues/cmd/mesh-issues/issue.go b/modules/mesh-issues/cmd/mesh-issues/issue.go index 64ac267b..51e785d5 100644 --- a/modules/mesh-issues/cmd/mesh-issues/issue.go +++ b/modules/mesh-issues/cmd/mesh-issues/issue.go @@ -151,9 +151,10 @@ type Link struct { At time.Time `json:"at"` } -// MostIssueBytes is the most one issue may be as kept: a value the bus holds (to-be 53 §4). The largest -// issue folder in hq was 25 KiB on 2026-10-10. -const MostIssueBytes = 256 << 10 +// MostIssueBytes is the most one issue may be as kept: 252 KiB, so the operation holding it whole (an open, an +// import, a carry) fits in one entry of the log, which takes at most 256 KiB (novox/hq ADR 0297, ADR 0300). The +// largest issue folder in hq was 25 KiB on 2026-10-10. +const MostIssueBytes = 252 << 10 // Entry is one key's value and the revision it was read at. type Entry struct { @@ -169,6 +170,9 @@ type Store interface { Update(state, key string, value any, revision uint64) (uint64, error) Keys(state string) ([]string, error) Increment(state, key string, attempts int) (uint64, error) + // History is every value a key still keeps, oldest first, puts and deletes (novox/hq ADR 0307): what the + // carry into the log reads (ADR 0300 §6). + History(state, key string) ([]stdio.StateRevision, error) } // runtimeStore is Store through the runtime that launched this bundle. diff --git a/modules/mesh-issues/cmd/mesh-issues/log_test.go b/modules/mesh-issues/cmd/mesh-issues/log_test.go index a626504e..c4876358 100644 --- a/modules/mesh-issues/cmd/mesh-issues/log_test.go +++ b/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) +} diff --git a/modules/mesh-issues/cmd/mesh-issues/main.go b/modules/mesh-issues/cmd/mesh-issues/main.go index ffd92a0f..2f674e6b 100644 --- a/modules/mesh-issues/cmd/mesh-issues/main.go +++ b/modules/mesh-issues/cmd/mesh-issues/main.go @@ -207,6 +207,9 @@ func (emptyStore) Update(string, string, any, uint64) (uint64, error) { return 0, errors.New("a dry run at the terminal writes nothing") } func (emptyStore) Keys(string) ([]string, error) { return nil, nil } +func (emptyStore) History(string, string) ([]stdio.StateRevision, error) { + return []stdio.StateRevision{}, nil +} func (emptyStore) Increment(string, string, int) (uint64, error) { return 0, errors.New("a dry run at the terminal writes nothing") } diff --git a/modules/mesh-issues/cmd/mesh-issues/memory_test.go b/modules/mesh-issues/cmd/mesh-issues/memory_test.go index 75d00a5b..257a573b 100644 --- a/modules/mesh-issues/cmd/mesh-issues/memory_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/memory_test.go @@ -232,6 +232,11 @@ func newHarness(t *testing.T) *harness { Logf: t.Logf, Sleep: func(time.Duration) {}, } + t.Cleanup(func() { + if h.Tracker.Log == Log(h.log) { + h.log.foldCheck(t, h.mem) + } + }) return h } diff --git a/modules/mesh-issues/cmd/mesh-issues/tracker.go b/modules/mesh-issues/cmd/mesh-issues/tracker.go index 32c9e819..e735ab26 100644 --- a/modules/mesh-issues/cmd/mesh-issues/tracker.go +++ b/modules/mesh-issues/cmd/mesh-issues/tracker.go @@ -10,7 +10,6 @@ import ( "sort" "strings" "sync" - "sync/atomic" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -46,8 +45,6 @@ type Tracker struct { groom grooming held *memoryView heldOnce sync.Once - // carryDeletes counts the deletions the carry passed over among the values the bucket kept. - carryDeletes atomic.Int64 } // The events this module says, one per kind of change. diff --git a/modules/mesh-issues/cmd/mesh-issues/tracker_test.go b/modules/mesh-issues/cmd/mesh-issues/tracker_test.go index 1d9be5a2..49d11876 100644 --- a/modules/mesh-issues/cmd/mesh-issues/tracker_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/tracker_test.go @@ -196,7 +196,7 @@ func (u unreadableLog) AppendAt(key string, value any, expect uint64) (uint64, e _, _ = u.fakeLog.AppendAt(key, value, expect) return 0, errors.New("no answer in 30 s") } -func (u unreadableLog) Last(string) (*stdio.LogEntry, error) { +func (u unreadableLog) Read(uint64, string, int) (*stdio.LogPage, error) { return nil, errors.New("no answer in 30 s") } @@ -447,7 +447,7 @@ func TestANoteThatWouldMakeTheIssueTooLargeIsRefusedAndNothingChanges(t *testing for i := 0; i < 40 && err == nil; i++ { _, err = h.Note(n, note) } - if err == nil || !strings.Contains(err.Error(), "256 KiB") { + if err == nil || !strings.Contains(err.Error(), "252 KiB") { t.Fatalf("no refusal: %v", err) } e, _ := h.mem.Get(IssuesState, keyOf(n)) diff --git a/modules/mesh-issues/cmd/mesh-issues/view.go b/modules/mesh-issues/cmd/mesh-issues/view.go index 3f574b61..62e67f33 100644 --- a/modules/mesh-issues/cmd/mesh-issues/view.go +++ b/modules/mesh-issues/cmd/mesh-issues/view.go @@ -85,7 +85,10 @@ func (h *memoryView) drop(n uint64) { } } -// all is every issue held, each a copy of its fields: a caller that changes one changes nothing held. +// all is every issue held, each a shallow copy: a caller may set an issue's fields without changing what is held, +// but its lists (notes, links, fixes and the rest) are shared with what is held, so a caller reads them and never +// changes them in place. Every caller in this module only reads; a write goes through change, which reads the +// bucket afresh. A deep copy per list would cost a JSON round trip of every issue on every list. func (h *memoryView) all() map[uint64]*Issue { h.mu.RLock() defer h.mu.RUnlock() diff --git a/modules/mesh-issues/go.sum b/modules/mesh-issues/go.sum index a888f498..b16d5979 100644 --- a/modules/mesh-issues/go.sum +++ b/modules/mesh-issues/go.sum @@ -1,9 +1,3 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011091942-4cf45b5ca0cf h1:ED/+RszKwm86U7H/NZRbS4b5LzQ8n7rubqKmeFqFtvM= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011091942-4cf45b5ca0cf/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011180852-5794ccd569be h1:G2ZwFVQsk5Df55gV6OwhJCykrZNQ1/UmO/OxbOJof28= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011180852-5794ccd569be/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261012081756-d41d255f28ad h1:rg8MlukAHgAlcxT1jJFwHaMRRd0sBU87gaOJ4Ykz9wY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261012081756-d41d255f28ad/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261012082113-9747f3e1b1ac h1:jnZ/n+fFMo7VQPNNXK1gjStHIJPw3SPbzPxsf/PpZqw= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261012082113-9747f3e1b1ac/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=