From 983d434623b9b4d136ba7c33213078f38ae514c6 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 12 Oct 2026 02:41:54 +0200 Subject: [PATCH] mesh-issues: keep every issue in the log changes and the bucket as its fold, so the record outlives the bucket's cap and a week of events (hq ADR 0300, issues 501, 502) Every write appends one operation at the issue's last sequence before the bucket is written; a refused append writes nothing. The start carries what the bucket holds, refolds what is behind and indexes every issue in memory, so list answers from memory with a ranked search and a before cursor. --- modules/mesh-issues/README.md | 80 +- .../mesh-issues/cmd/mesh-issues/bus_test.go | 114 ++- .../mesh-issues/cmd/mesh-issues/changes.go | 780 ++++++++++++++++++ modules/mesh-issues/cmd/mesh-issues/claims.go | 4 +- .../mesh-issues/cmd/mesh-issues/delivery.go | 8 +- modules/mesh-issues/cmd/mesh-issues/groom.go | 26 +- modules/mesh-issues/cmd/mesh-issues/import.go | 37 +- .../cmd/mesh-issues/import_test.go | 3 +- modules/mesh-issues/cmd/mesh-issues/issue.go | 3 + .../mesh-issues/cmd/mesh-issues/log_test.go | 574 +++++++++++++ modules/mesh-issues/cmd/mesh-issues/main.go | 39 +- .../cmd/mesh-issues/manifest_test.go | 11 +- .../cmd/mesh-issues/memory_test.go | 36 +- modules/mesh-issues/cmd/mesh-issues/ready.go | 12 +- .../mesh-issues/cmd/mesh-issues/tracker.go | 189 +++-- .../cmd/mesh-issues/tracker_test.go | 33 +- .../cmd/mesh-issues/triage_test.go | 4 +- modules/mesh-issues/cmd/mesh-issues/view.go | 288 +++++++ modules/mesh-issues/go.mod | 2 +- modules/mesh-issues/go.sum | 2 + modules/mesh-issues/module.json | 5 +- 21 files changed, 2074 insertions(+), 176 deletions(-) create mode 100644 modules/mesh-issues/cmd/mesh-issues/changes.go create mode 100644 modules/mesh-issues/cmd/mesh-issues/log_test.go create mode 100644 modules/mesh-issues/cmd/mesh-issues/view.go diff --git a/modules/mesh-issues/README.md b/modules/mesh-issues/README.md index 5c52946c..3edb33f0 100644 --- a/modules/mesh-issues/README.md +++ b/modules/mesh-issues/README.md @@ -19,7 +19,7 @@ what it refuses. |---|---|---| | `open` | `title`, `report`, optionally `located-in`, `record` | checks the issue, then takes the next number and keeps it under that number; answers the number | | `get` | `number` | the issue whole: fields, notes, links | -| `list` | optionally `status`, `owner`, `phrase`, `limit`, `severity`, `needs`, `ready`, `flag` (`delivered-not-resolved` or `abandoned`) | short rows, newest first: number, status, owner, title, and each flag that holds | +| `list` | optionally `status`, `owner`, `phrase`, `limit`, `severity`, `needs`, `ready`, `flag` (`delivered-not-resolved` or `abandoned`), `search` (words, ranked), `before` (a number, the cursor) | short rows, newest first, or best first for a `search`: number, status, owner, title, and each flag that holds; `next`, the `before` that reads on, when more match | | `move` | `number`, `status`, and what that status needs (`by` for `diagnosing` and `fixing`, optionally `take-over`) | changes the status under the development cycle's rules | | `note` | `number`, `text` | appends a note; a note is never edited | | `link` | `number`, `kind`, `target`, optionally `remove` | adds or removes a typed link | @@ -200,14 +200,15 @@ in one instance or two — never take one number. A write another opener won is bounded number of times with a short random wait; past the bound the open is refused and takes no number. - **A refused open takes no number**: an issue is checked before its number is taken. -- **A number is never handed out twice, and may be skipped.** A number taken and then not kept — the bus - failing between the counter and the issue — stays unused, and the answer says which. A write whose answer - was lost is recognised as its own when tried again, so it leaves no gap; when it cannot be confirmed either - way, the answer says the issue may have been written and to `get` that number before opening it again, - never that nothing was opened. -- **An issue is never overwritten.** It is kept by a write that is refused when the key exists; if the - counter ever hands out a number already kept, the open is refused loudly, the number stays unused, and - nothing is overwritten. +- **A number is never handed out twice, and may be skipped.** A number taken and then not taken by the log — + the bus failing between the counter and the issue's first operation, or a full log — is spent, and the open + answers the failure (hq ADR 0300 §2). An append whose answer was lost is recognised as its own by the open's + id when tried again or read back, so it leaves no gap; when it cannot be confirmed either way, the answer says + the issue may have been written and to `get` that number before opening it again, never that nothing was + opened. +- **An issue is never overwritten.** Its first operation is appended at `expect` 0, refused when the log + already holds one for that number; if the counter ever hands out a number already kept, the open is refused + loudly, the number stays unused, and nothing is overwritten. - **The counter starts once**, the first time a number is taken: at the setting `first-number`, or after the highest issue already kept, whichever is higher, made only if it does not exist. The setting is never read again, so a number never moves. @@ -242,16 +243,52 @@ sessions writing a record at once never take one number. `decision-floor` or `research-floor` (the controller's settings verb) to the highest number on any open branch of hq; once the counter has started, the floor is not read again. -## State and events +## The log, the bucket and the events (hq ADR 0300) -The states, declared in the manifest and made by the controller; the module never makes a bucket -(ADR 0201): `issues`, one key per issue by number, keeping 8 values per key; `counters`, one value per key -(`issue`, `decision`, `research`); `grooming`, the backlog's pass; `reservations`, one key per reserved number. -Neither expires. One issue is kept in at most 256 KiB; a note that would pass it is refused. +**The record is the log `changes`** (declared in the manifest's `logs`, made by the controller, hq ADR 0297), +one key per issue by number. Every write — open, move, note, link, set, the triage, a delivery's word on a +fix, a warning said or cleared, the import — is one operation appended under the issue's number with what +changed, who acted and when, at the sequence of the last operation the issue was folded at +(`AppendAt`). Another operation in between refuses it: nothing is written, the log is folded again and the +change applied to that, a bounded number of times. A write the log refused writes nothing; only an open's +number, once taken, stays spent. An operation that begins an issue (`opened`, `imported`, `carried`) holds it +whole; every other one names the fields given a new value (`set`), the lists added to at their end (`add`: +a note is kept as itself, not as every note again) and the fields taken away (`unset`). The fold applies them +as written, so a new kind of operation needs nothing new in the fold. -Every change is an event: `mesh-issues.opened`, `.moved`, `.noted`, `.linked`, `.imported`, `.reserved`, each with the -number, what changed, who changed it and when. The events are the audit trail. A change whose event the bus did not take -is kept, and its answer says the event was not said (`announced: false`). +**The bucket `issues` is the fold of the log**, one value per key, each issue carrying `folded`, the sequence +of its last operation. It is written after the append, at the revision read; one whose write fails after the +log took the change is behind, and caught up at its next change (the stale append is refused and the log +folded) or at the next start. Agents and views watch it as before. + +**On start**, the tracker watches its own bucket, carries what the bucket holds and the log does not (below), +reads the whole log page by page, folds every issue whose last operation is newer than its `folded` — an +emptied bucket is rebuilt whole — and raises the issue counter to the highest number in the log where it is +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 runtime offers no read of a +key's older values today, so the carry takes each issue's current value and says so in its start line; with a +store that reads them, it takes every one. It runs on start rather than as a verb because it must precede every +write, and a verb could be called late or never. + +**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. +`list` answers from memory, with `search` ranked best first, and `before` paging the newest-first list past +the 500 rows one answer holds. The index is the module's own: a few thousand issues need a hundred lines, not a +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 +that would pass it is refused. + +**The events announce; the log is the audit trail.** Every change is still an event: `mesh-issues.opened`, +`.moved`, `.noted`, `.linked`, `.imported`, `.reserved`, each with the number, what changed, who changed it and +when. A change whose event the bus did not take is kept, and its answer says the event was not said +(`announced: false`). ## Who acted @@ -319,3 +356,12 @@ date, a replay, an empty list, a wontfix): the frontmatter read as written, each number with the counter untouched, a second run changing nothing, an issue the tracker holds differently left alone, a folder-backed issue filled once, a folder past the counter or past the first number refused, a dry run writing nothing, and a report that cannot be read refused alone with why. + +The log (hq ADR 0300) is tested in `log_test.go` over a log kept in the test with the runtime's semantics +(`expect`, `ErrChanged`): a write whose append is refused changing nothing, an operation in between not lost, an +open refused at validation taking no number, an append failing after the number spending it and answering the +failure, an emptied bucket restored exactly by the fold on start, a bucket behind its log catching up, the +counter raised to the log, each write saying its event beside its one operation, `before` paging 530 issues +without a gap or a repeat, a ranked search with the index rebuilt on start and kept current by the watch, and +the carry carrying each value kept, oldest first, once. Over the bus, the stream the controller makes holds one +operation per change of `TestAnIssueIsOpenedNotedLinkedMovedAndListedOverTheBus`. diff --git a/modules/mesh-issues/cmd/mesh-issues/bus_test.go b/modules/mesh-issues/cmd/mesh-issues/bus_test.go index f4b523d7..b85247e5 100644 --- a/modules/mesh-issues/cmd/mesh-issues/bus_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/bus_test.go @@ -12,6 +12,7 @@ import ( "path/filepath" "sort" "strconv" + "strings" "sync" "sync/atomic" "testing" @@ -48,6 +49,8 @@ type bus struct { nc *nats.Conn js jetstream.JetStream kvs map[string]jetstream.KeyValue + // logs is each declared log's stream, made as the controller makes it (novox/hq ADR 0297). + logs map[string]jetstream.Stream } func newBus(t *testing.T) *bus { @@ -73,7 +76,7 @@ func newBus(t *testing.T) *bus { if err != nil { t.Fatal(err) } - b := &bus{nc: nc, js: js, kvs: map[string]jetstream.KeyValue{}} + b := &bus{nc: nc, js: js, kvs: map[string]jetstream.KeyValue{}, logs: map[string]jetstream.Stream{}} // The controller's job: a bucket per declared state, with the options the manifest gives it. m, _ := readManifest(t) for _, d := range m.stateDeclarations(t) { @@ -88,9 +91,98 @@ func newBus(t *testing.T) *bus { } b.kvs[d.Name] = kv } + for _, name := range m.Logs { + st, err := js.CreateStream(context.Background(), jetstream.StreamConfig{ + Name: "LOG_" + m.Module + "_" + name, Subjects: []string{"mesh.log." + m.Module + "." + name + ".>"}, + Storage: jetstream.FileStorage, MaxMsgsPerSubject: -1, Discard: jetstream.DiscardNew, AllowDirect: true, + DenyDelete: true, DenyPurge: true}) + if err != nil { + t.Fatal(err) + } + b.logs[name] = st + } return b } +// logVerb is the runtime's mesh/log verbs over a stream, as the tool runner serves them (ADR 0297). +func (b *bus) logVerb(ctx context.Context, verb string, params json.RawMessage) (any, error) { + var p struct { + Log string `json:"log"` + Key string `json:"key"` + Value json.RawMessage `json:"value"` + Expect *uint64 `json:"expect"` + From uint64 `json:"from"` + Keys string `json:"keys"` + Limit int `json:"limit"` + } + if err := json.Unmarshal(params, &p); err != nil { + return nil, err + } + st := b.logs[p.Log] + if st == nil { + return nil, fmt.Errorf("the log %q is not declared", p.Log) + } + subject := func(key string) string { return "mesh.log.mesh-issues." + p.Log + "." + key } + entry := func(m *jetstream.RawStreamMsg) map[string]any { + return map[string]any{"sequence": m.Sequence, "key": m.Subject[len(subject("")):], "value": json.RawMessage(m.Data), + "at": m.Time} + } + switch verb { + case "append": + msg := nats.NewMsg(subject(p.Key)) + msg.Data = p.Value + if p.Expect != nil { + msg.Header.Set(jetstream.ExpectedLastSubjSeqHeader, strconv.FormatUint(*p.Expect, 10)) + } + ack, err := b.js.PublishMsg(ctx, msg) + if err != nil { + return nil, err + } + return map[string]any{"sequence": ack.Sequence}, nil + case "last": + m, err := st.GetLastMsgForSubject(ctx, subject(p.Key)) + if errors.Is(err, jetstream.ErrMsgNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + return entry(m), nil + case "read": + filter := subject(">") + if p.Keys != "" { + filter = subject(p.Keys) + } + limit := p.Limit + if limit <= 0 { + limit = 100 + } + info, err := st.Info(ctx) + if err != nil { + return nil, err + } + entries := []map[string]any{} + var next uint64 + for seq := max(p.From, 1); seq <= info.State.LastSeq; { + m, err := st.GetMsg(ctx, seq, jetstream.WithGetMsgSubject(filter)) + if errors.Is(err, jetstream.ErrMsgNotFound) { + break + } + if err != nil { + return nil, err + } + if len(entries) == limit { + next = m.Sequence + break + } + entries = append(entries, entry(m)) + seq = m.Sequence + 1 + } + return map[string]any{"entries": entries, "next": next, "last": info.State.LastSeq}, nil + } + return nil, fmt.Errorf("the runtime answers no mesh/log.%s", verb) +} + // instance is one running bundle, and the runtime's half of its conversation. type instance struct { t *testing.T @@ -213,6 +305,9 @@ func (i *instance) answer(m rpc) { } return map[string]any{"decision": 291, "research": 45, "records": 336}, nil } + if verb, ok := strings.CutPrefix(m.Method, "mesh/log."); ok { + return i.b.logVerb(ctx, verb, m.Params) + } kv := i.b.kvs[p.State] if kv == nil { return nil, fmt.Errorf("the state %q is not declared", p.State) @@ -531,4 +626,21 @@ func TestAnIssueIsOpenedNotedLinkedMovedAndListedOverTheBus(t *testing.T) { if fmt.Sprint(kinds) != "[opened noted linked moved moved moved]" { t.Fatalf("events %v", kinds) } + // The record is the log (novox/hq ADR 0300 §1): one operation per change, in the stream the controller makes, + // and the issue in the bucket folded at the last of them. + var ops []string + for seq := uint64(1); ; seq++ { + m, err := b.logs[ChangesLog].GetMsg(context.Background(), seq) + if err != nil { + break + } + var op Operation + if err := json.Unmarshal(m.Data, &op); err != nil || m.Subject != "mesh.log.mesh-issues.changes.365" { + t.Fatalf("entry %d on %s: %s", seq, m.Subject, m.Data) + } + ops = append(ops, op.Op) + } + if fmt.Sprint(ops) != fmt.Sprint(kinds) || is.Folded != uint64(len(ops)) { + t.Fatalf("the log holds %v, the issue is folded at %d", ops, is.Folded) + } } diff --git a/modules/mesh-issues/cmd/mesh-issues/changes.go b/modules/mesh-issues/cmd/mesh-issues/changes.go new file mode 100644 index 00000000..64a1a5ab --- /dev/null +++ b/modules/mesh-issues/cmd/mesh-issues/changes.go @@ -0,0 +1,780 @@ +package main + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "sort" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// The tracker's record is its log `changes`, not its bucket (novox/hq ADR 0300 §1). Every write is one +// operation appended under the issue's number, refused when another operation on that issue came in between; +// only then is the folded issue written to the bucket `issues`, which is what the operations add up to (§3). +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. +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. +type runtimeLog struct{} + +func (runtimeLog) AppendAt(key string, value any, expect uint64) (uint64, error) { + return stdio.Log(ChangesLog).AppendAt(key, value, expect) +} +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") + +// logPage is how many entries one read of the log asks for: the most the runtime gives. +const logPage = 1000 + +// The kinds of operation. An operation that begins an issue holds it whole: opened, imported, carried. +const ( + OpOpened = "opened" + OpImported = "imported" + // OpCarried is a value the bucket kept when the tracker switched to its log (ADR 0300 §6), whole. + OpCarried = "carried" +) + +// Operation is one change to one issue, as the log keeps it: what changed, who acted and when (ADR 0300 §1). +// One that begins an issue, or carries a value the bucket kept, holds the issue whole; every other one names +// the fields it gave a new value, the lists it added to at their end, and the fields it took away. The fold +// applies them as they are written, so it needs no knowledge of the kind: a new kind is folded as it comes. +type Operation struct { + Op string `json:"op"` + By string `json:"by"` + At time.Time `json:"at"` + Issue *Issue `json:"issue,omitempty"` + Set map[string]json.RawMessage `json:"set,omitempty"` + Add map[string]json.RawMessage `json:"add,omitempty"` + Unset []string `json:"unset,omitempty"` +} + +// Act is who changes an issue and as what kind of operation. +type Act struct { + Op string + By string +} + +// fields is an issue as its fields in JSON, without the sequence it was folded at: that is the log's, not +// the issue's. +func fields(is *Issue) (map[string]json.RawMessage, error) { + raw, err := json.Marshal(is) + if err != nil { + return nil, err + } + var m map[string]json.RawMessage + if err := json.Unmarshal(raw, &m); err != nil { + return nil, err + } + delete(m, "folded") + return m, nil +} + +// operation is what changed from before to after as one operation of act, and false when nothing did. +func operation(act Act, at time.Time, before, after *Issue) (*Operation, bool, error) { + b, err := fields(before) + if err != nil { + return nil, false, err + } + a, err := fields(after) + if err != nil { + return nil, false, err + } + op := &Operation{Op: act.Op, By: act.By, At: at} + for k, v := range a { + old, had := b[k] + if had && bytes.Equal(old, v) { + continue + } + if !had { + // A list left out while empty: its first element is added like any other. + old = json.RawMessage("[]") + } + if tail, ok := appended(old, v); ok { + if op.Add == nil { + op.Add = map[string]json.RawMessage{} + } + op.Add[k] = tail + continue + } + if op.Set == nil { + op.Set = map[string]json.RawMessage{} + } + op.Set[k] = v + } + for k := range b { + if _, still := a[k]; !still { + op.Unset = append(op.Unset, k) + } + } + sort.Strings(op.Unset) + return op, len(op.Set)+len(op.Add)+len(op.Unset) > 0, nil +} + +// appended is what a list gained at its end, when after is before with elements added and nothing else +// changed: a note or a link is kept in the log as itself, not as every note again. +func appended(before, after json.RawMessage) (json.RawMessage, bool) { + var b, a []json.RawMessage + if len(after) == 0 || after[0] != '[' || json.Unmarshal(before, &b) != nil || json.Unmarshal(after, &a) != nil || len(a) <= len(b) { + return nil, false + } + for i := range b { + if !bytes.Equal(compact(b[i]), compact(a[i])) { + return nil, false + } + } + tail, err := json.Marshal(a[len(b):]) + if err != nil { + return nil, false + } + return tail, true +} + +func compact(raw json.RawMessage) []byte { + var buf bytes.Buffer + if json.Compact(&buf, raw) != nil { + return raw + } + return buf.Bytes() +} + +// ErrNoBeginning is an operation on an issue the log holds no beginning for. +var ErrNoBeginning = errors.New("the log holds no operation that begins it") + +// fold applies one operation to an issue (nil before its first) and answers the issue it adds up to, at seq. +func fold(base *Issue, op *Operation, seq uint64) (*Issue, error) { + var m map[string]json.RawMessage + switch { + case op.Issue != nil: + f, err := fields(op.Issue) + if err != nil { + return nil, err + } + m = f + case base == nil: + return nil, fmt.Errorf("a %s operation at %d: %w", op.Op, seq, ErrNoBeginning) + default: + f, err := fields(base) + if err != nil { + return nil, err + } + m = f + } + for k, v := range op.Set { + m[k] = v + } + for k, tail := range op.Add { + var have, more []json.RawMessage + if old, ok := m[k]; ok { + if err := json.Unmarshal(old, &have); err != nil { + return nil, fmt.Errorf("a %s operation at %d adds to %s, which is not a list", op.Op, seq, k) + } + } + if err := json.Unmarshal(tail, &more); err != nil { + return nil, fmt.Errorf("a %s operation at %d adds to %s what is not a list", op.Op, seq, k) + } + joined, err := json.Marshal(append(have, more...)) + if err != nil { + return nil, err + } + m[k] = joined + } + for _, k := range op.Unset { + delete(m, k) + } + raw, err := json.Marshal(m) + if err != nil { + return nil, err + } + var is Issue + if err := json.Unmarshal(raw, &is); err != nil { + return nil, fmt.Errorf("a %s operation at %d folds to no issue: %v", op.Op, seq, err) + } + is.Folded = seq + return &is, nil +} + +// foldEntry is fold of one entry of the log. +func foldEntry(base *Issue, e stdio.LogEntry) (*Issue, error) { + var op Operation + if err := json.Unmarshal(e.Value, &op); err != nil { + return nil, fmt.Errorf("the entry at %d is no operation this module reads: %v", e.Sequence, err) + } + return fold(base, &op, e.Sequence) +} + +// readKey is every entry of one issue's key from the sequence from on, oldest first, read page by page. +func (t *Tracker) readKey(n uint64, from uint64) ([]stdio.LogEntry, error) { + if t.Log == nil { + return nil, errNoLog + } + var out []stdio.LogEntry + for from != 0 || out == nil { + p, err := t.Log.Read(max(from, 1), keyOf(n), logPage) + if err != nil { + return nil, fmt.Errorf("the log of issue %d could not be read: %v", n, err) + } + if out == nil { + out = []stdio.LogEntry{} + } + out = append(out, p.Entries...) + from = p.Next + } + return out, nil +} + +// append adds one operation to an issue at the sequence its fold was read at. A refusal is stdio.ErrChanged: +// another operation on the issue came in between, and nothing was written. +func (t *Tracker) appendOp(n uint64, op *Operation, expect uint64) (uint64, error) { + if t.Log == nil { + return 0, errNoLog + } + raw, err := json.Marshal(op) + if err != nil { + return 0, err + } + return t.Log.AppendAt(keyOf(n), json.RawMessage(raw), expect) +} + +// errBegun is a first operation refused because the log already begins that issue, with another one. +var errBegun = errors.New("the log already holds another issue under that number") + +// begin appends the first operation of a new issue — expect 0 — and answers its sequence. A first operation +// whose answer was lost is known as its own by the issue's opening id. Any other refusal is errBegun; a +// failure is answered as it is (ADR 0300 §2: the number it was given stays spent). +func (t *Tracker) begin(is *Issue, act Act) (uint64, error) { + op := &Operation{Op: act.Op, By: act.By, At: t.now(), Issue: is} + var err error + for i := 0; i < createAttempts; i++ { + if i > 0 { + t.Sleep(jitter(time.Duration(i) * 50 * time.Millisecond)) + } + var seq uint64 + seq, err = t.appendOp(is.Number, op, 0) + if err == nil { + return seq, nil + } + if errors.Is(err, stdio.ErrChanged) { + last, lerr := t.Log.Last(keyOf(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 + } + 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)) + 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: + return 0, errBegun + } + return 0, err +} + +// 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") + +// mine says whether a log entry is the first operation of the issue opened by opening. +func mine(e *stdio.LogEntry, opening string) bool { + if e == nil || opening == "" { + return false + } + var op Operation + return json.Unmarshal(e.Value, &op) == nil && op.Issue != nil && op.Issue.Opening == opening +} + +// change reads an issue, applies f, appends what changed to the log at the sequence the issue was folded at, +// and then writes the fold to the bucket (ADR 0300 §1). An append another operation won is read and applied +// again, so no note, link or move is lost and none is applied to a stale issue; a refused append writes +// nothing. A change that changes nothing appends nothing. +func (t *Tracker) change(n uint64, act Act, f func(*Issue) error) (*Issue, error) { + for i := 0; i < issueAttempts; i++ { + is, rev, err := t.current(n) + if err != nil { + return nil, err + } + 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 + } + continue + } + before, err := clone(is) + if err != nil { + return nil, err + } + if err := f(is); err != nil { + return nil, err + } + raw, err := json.Marshal(is) + if err != nil { + 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 "+ + "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) + if err != nil { + return nil, err + } + if !changed { + return is, nil + } + seq, err := t.appendOp(n, op, is.Folded) + if errors.Is(err, stdio.ErrChanged) { + // Another operation came in between, or the bucket is behind the log: fold what the log holds and + // apply this change again to that. + if _, err := t.catchUp(n); err != nil { + return nil, err + } + t.Sleep(jitter(time.Duration(i+1) * 5 * time.Millisecond)) + continue + } + if err != nil { + return nil, fmt.Errorf("issue %d: the log did not take the change (%v), so the bucket was not written. If "+ + "the log took it before the failure it is kept and the bucket folds it at the next change or start: "+ + "get %d before changing it again", n, err, n) + } + is.Folded = seq + t.store(n, is, rev) + return is, nil + } + return nil, fmt.Errorf("issue %d was changed by another caller %d times while this change was made; nothing was "+ + "changed. Try again", n, issueAttempts) +} + +// current is an issue as the bucket holds it, and the revision it was read at; an issue the log holds and the +// bucket does not is folded into the bucket first. +func (t *Tracker) current(n uint64) (*Issue, uint64, error) { + is, rev, err := read(t.Store, n) + if errors.Is(err, ErrNoIssue) && t.Log != nil { + if _, ferr := t.catchUp(n); ferr != nil { + return nil, 0, ferr + } + is, rev, err = read(t.Store, n) + } + return is, rev, err +} + +// store writes the fold of an issue at the revision read, or creates it at 0. The log already holds the +// change, so a write another writer won is folded again from the log, and one that fails is said and left for +// the next change or start, which fold the log again; it is never answered as a change not made. +func (t *Tracker) store(n uint64, is *Issue, rev uint64) { + var err error + if rev == 0 { + _, err = t.Store.Create(IssuesState, keyOf(n), is) + } else { + _, err = t.Store.Update(IssuesState, keyOf(n), is, rev) + } + if err == nil { + t.remember(is) + return + } + if errors.Is(err, stdio.ErrChanged) { + if folded, cerr := t.catchUp(n); cerr == nil { + if folded != nil { + t.remember(folded) + } + return + } else { + err = cerr + } + } + t.remember(is) + t.Logf("[mesh-issues] issue %d is kept in the log at %d and its bucket could not be written (%v): the bucket is "+ + "behind until the next change of it or the next start folds it", n, is.Folded, err) +} + +// catchUp folds into the bucket every operation of an issue the bucket has not folded, and answers the issue +// as folded (nil when neither holds it). The bucket is written only at the revision read; a write another +// writer won is read again. +func (t *Tracker) catchUp(n uint64) (*Issue, error) { + for i := 0; i < issueAttempts; i++ { + is, rev, err := read(t.Store, n) + if err != nil && !errors.Is(err, ErrNoIssue) { + return nil, err + } + var from uint64 = 1 + if is != nil { + if is.Folded == 0 { + // Not carried yet: what the log holds is no fold of it. + return is, nil + } + from = is.Folded + 1 + } + entries, err := t.readKey(n, from) + if err != nil { + return nil, err + } + if len(entries) == 0 { + return is, nil + } + folded := is + for _, e := range entries { + if folded, err = foldEntry(folded, e); err != nil { + return nil, fmt.Errorf("issue %d cannot be folded from its log: %v", n, err) + } + } + if is == nil { + _, err = t.Store.Create(IssuesState, keyOf(n), folded) + } else { + _, err = t.Store.Update(IssuesState, keyOf(n), folded, rev) + } + if err == nil { + t.remember(folded) + return folded, nil + } + if !errors.Is(err, stdio.ErrChanged) { + return nil, fmt.Errorf("issue %d is behind its log in the bucket and could not be written (%v)", n, err) + } + } + return nil, fmt.Errorf("issue %d was written by another writer %d times while it was folded from its log", n, issueAttempts) +} + +func clone(is *Issue) (*Issue, error) { + raw, err := json.Marshal(is) + if err != nil { + return nil, err + } + var c Issue + return &c, json.Unmarshal(raw, &c) +} + +// History is a Store that reads every value a key still keeps, oldest first. The carry reads it where the +// store offers it; the runtime offers no such read today (novox/hq ADR 0300 §6), and then the carry takes +// each issue's current value, and says so. +type History interface { + History(state, key string) ([]Entry, error) +} + +// Carried is what the carry did (ADR 0300 §6). +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 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"` + 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) { + for i := 0; i < issueAttempts; i++ { + is, rev, err := read(t.Store, n) + if err != nil { + return 0, err + } + if is.Folded != 0 { + return 0, nil + } + entries, err := t.readKey(n, 1) + if err != nil { + return 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) + } + } + expect = folded.Folded + if same(folded, is) { + values = nil + } else { + values = []*Issue{is} + } + } 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) + 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) + } + appendedNow++ + expect = seq + if folded, err = fold(nil, &Operation{Op: OpCarried, Issue: v}, seq); err != nil { + return appendedNow, err + } + } + if raced { + // Another carrier was first: read again; what it appended is recognised as the log's. + continue + } + 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 + } + continue + } + return appendedNow, 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 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. +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 { + var v Issue + if json.Unmarshal(e.Value, &v) != nil { + t.Logf("[mesh-issues] an older value of issue %d is in a form this module cannot read: not carried", n) + continue + } + out = append(out, &v) + } + if len(out) == 0 || !same(out[len(out)-1], current) { + out = append(out, current) + } + return out, nil +} + +// same says whether two issues hold the same, whatever sequence each was folded at. +func same(a, b *Issue) bool { + fa, err1 := fields(a) + fb, err2 := fields(b) + if err1 != nil || err2 != nil || len(fa) != len(fb) { + return false + } + for k, v := range fa { + if !bytes.Equal(compact(v), compact(fb[k])) { + return false + } + } + 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. +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} + for _, k := range keys { + n, ok := numberOfKey(k) + if !ok { + continue + } + is, _, err := read(t.Store, n) + if errors.Is(err, ErrNoIssue) { + continue + } + if err != nil { + return out, err + } + if is.Folded != 0 { + out.Already++ + continue + } + v, err := t.carryOne(n) + out.Values += v + if err != nil { + return out, err + } + if v > 0 { + out.Issues++ + } else { + out.Already++ + } + } + 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: + 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) + 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) + } + return out, nil +} + +// Started is what the tracker did on start (ADR 0300 §3, §6). +type Started struct { + // Entries is how many operations the log holds; Issues how many issues they add up to. + Entries int `json:"entries"` + Issues int `json:"issues"` + // Refolded is how many issues the bucket held behind the log, Rebuilt how many it did not hold at all. + Refolded int `json:"refolded"` + Rebuilt int `json:"rebuilt"` + Carried *Carried `json:"carried"` + // Counter is the issue counter after start, raised to the highest number in the log where it was below. + Counter uint64 `json:"counter"` + Faults []string `json:"faults,omitempty"` + Said string `json:"said"` +} + +// Start carries what the bucket holds and the log does not, reads the whole log, folds every issue whose +// last operation is newer than the bucket's fold of it — an emptied bucket is rebuilt whole — keeps the +// counter at or above the highest number in the log, and holds every issue in memory with its index +// (ADR 0300 §3, §5, §6). +func (t *Tracker) Start() (*Started, error) { + out := &Started{} + if t.Log == nil { + 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) + } + folds := map[uint64]*Issue{} + var highest uint64 + for from := uint64(1); from != 0; { + p, err := t.Log.Read(from, "", logPage) + if err != nil { + return out, fmt.Errorf("the log could not be read from %d: %v", from, err) + } + for _, e := range p.Entries { + out.Entries++ + n, ok := numberOfKey(e.Key) + if !ok { + out.Faults = append(out.Faults, fmt.Sprintf("the entry at %d has the key %q, which is no issue's number", e.Sequence, e.Key)) + continue + } + highest = max(highest, n) + f, err := foldEntry(folds[n], e) + if err != nil { + out.Faults = append(out.Faults, fmt.Sprintf("issue %d: %v", n, err)) + continue + } + folds[n] = f + } + from = p.Next + } + out.Issues = len(folds) + numbers := make([]uint64, 0, len(folds)) + for n := range folds { + numbers = append(numbers, n) + } + sort.Slice(numbers, func(i, j int) bool { return numbers[i] < numbers[j] }) + for _, n := range numbers { + f := folds[n] + is, rev, err := read(t.Store, n) + switch { + case errors.Is(err, ErrNoIssue): + if _, err := t.Store.Create(IssuesState, keyOf(n), f); err != nil && !errors.Is(err, stdio.ErrChanged) { + out.Faults = append(out.Faults, fmt.Sprintf("issue %d could not be rebuilt in the bucket: %v", n, err)) + continue + } else if err == nil { + out.Rebuilt++ + } + case err != nil: + out.Faults = append(out.Faults, err.Error()) + 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)) + continue + } else if err == nil { + out.Refolded++ + } + } + } + if out.Counter, err = t.raiseCounter(highest); err != nil { + out.Faults = append(out.Faults, err.Error()) + } + all, err := t.readAll() + if err != nil { + return out, err + } + t.view().load(all) + out.Said = fmt.Sprintf("The log holds %d operations of %d issues; the bucket was behind on %d and missed %d, now "+ + "folded. %s The issue counter is at %d. %d issues are held in memory and indexed.", out.Entries, out.Issues, + out.Refolded, out.Rebuilt, carried.Said, out.Counter, len(all)) + if len(out.Faults) > 0 { + out.Said += fmt.Sprintf(" %d faults: %v", len(out.Faults), out.Faults) + } + return out, nil +} + +// raiseCounter keeps the issue counter at or above the highest number in the log, so no number the log holds +// is handed out again; a counter not started yet is left to the first open, which starts it above every issue +// the bucket holds. +func (t *Tracker) raiseCounter(highest uint64) (uint64, error) { + for i := 0; i < issueAttempts; i++ { + e, err := t.Store.Get(CountersState, IssueCounter) + if err != nil { + return 0, fmt.Errorf("the issue counter cannot be read: %v", err) + } + if e == nil { + return 0, nil + } + var n uint64 + if err := json.Unmarshal(e.Value, &n); err != nil { + return 0, fmt.Errorf("the issue counter holds %s, no number", e.Value) + } + if n >= highest { + return n, nil + } + _, err = t.Store.Update(CountersState, IssueCounter, highest, e.Revision) + if err == nil { + t.Logf("[mesh-issues] the issue counter was at %d, below issue %d in the log: raised to %d", n, highest, highest) + return highest, nil + } + if !errors.Is(err, stdio.ErrChanged) { + return n, fmt.Errorf("the issue counter is at %d, below issue %d in the log, and could not be raised: %v", n, highest, err) + } + } + return 0, errors.New("the issue counter was advanced by others while it was raised to the log's highest number") +} diff --git a/modules/mesh-issues/cmd/mesh-issues/claims.go b/modules/mesh-issues/cmd/mesh-issues/claims.go index 03aae0f5..fdbcd00c 100644 --- a/modules/mesh-issues/cmd/mesh-issues/claims.go +++ b/modules/mesh-issues/cmd/mesh-issues/claims.go @@ -91,7 +91,7 @@ func (t *Tracker) sayNotResolved(is *Issue, now time.Time) (bool, error) { prs, ago, at.Format(time.RFC3339), worker), "", false); err != nil { return false, err } - _, err := t.change(is.Number, func(i *Issue) error { i.NotResolvedOpen = true; return nil }) + _, err := t.change(is.Number, Act{Op: "warned", By: Seat}, func(i *Issue) error { i.NotResolvedOpen = true; return nil }) return err == nil, err } @@ -105,7 +105,7 @@ func (t *Tracker) clearNotResolved(is *Issue) error { "fix not delivered yet", is.Number, is.Status), true); err != nil { return err } - _, err := t.change(is.Number, func(i *Issue) error { i.NotResolvedOpen = false; return nil }) + _, err := t.change(is.Number, Act{Op: "warned", By: Seat}, func(i *Issue) error { i.NotResolvedOpen = false; return nil }) return err } diff --git a/modules/mesh-issues/cmd/mesh-issues/delivery.go b/modules/mesh-issues/cmd/mesh-issues/delivery.go index c1a4240c..45e6bfa4 100644 --- a/modules/mesh-issues/cmd/mesh-issues/delivery.go +++ b/modules/mesh-issues/cmd/mesh-issues/delivery.go @@ -167,7 +167,7 @@ func (t *Tracker) recordFix(pr, state, why string, merged bool, at time.Time, on // the same pull request at most, and an older one is history. continue } - if _, err := t.change(n, func(i *Issue) error { + if _, err := t.change(n, Act{Op: "delivery", By: Seat}, func(i *Issue) error { f := fixOf(i, pr) if f == nil { i.Fixes = append(i.Fixes, Fix{PullRequest: pr}) @@ -242,7 +242,7 @@ func (t *Tracker) afterFix(n uint64) error { } } claimed := false - if _, err := t.change(n, func(i *Issue) error { + if _, err := t.change(n, Act{Op: EventReadyToVerify, By: Seat}, func(i *Issue) error { claimed = false if strings.Join(i.VerifyFor, ",") == strings.Join(prs, ",") { return nil @@ -290,7 +290,7 @@ func (t *Tracker) clearStuck(is *Issue) error { if err := t.Notify(stuckKey(is.Number), "", "", "", "", "the fix of issue "+keyOf(is.Number)+" is delivered or no longer waited for", true); err != nil { return err } - _, err := t.change(is.Number, func(i *Issue) error { i.StuckOpen = false; return nil }) + _, err := t.change(is.Number, Act{Op: "warned", By: Seat}, func(i *Issue) error { i.StuckOpen = false; return nil }) return err } @@ -514,7 +514,7 @@ func (t *Tracker) Sweep() (int, error) { } } pr := f.PullRequest - if _, err := t.change(n, func(i *Issue) error { + if _, err := t.change(n, Act{Op: "warned", By: Seat}, func(i *Issue) error { if g := fixOf(i, pr); g != nil { g.StuckSaid = true } diff --git a/modules/mesh-issues/cmd/mesh-issues/groom.go b/modules/mesh-issues/cmd/mesh-issues/groom.go index bf2b2877..38e98321 100644 --- a/modules/mesh-issues/cmd/mesh-issues/groom.go +++ b/modules/mesh-issues/cmd/mesh-issues/groom.go @@ -188,30 +188,6 @@ func hasLink(is *Issue, kind string, target uint64) bool { return false } -// allIssues is every issue kept, by number. -func (t *Tracker) allIssues() (map[uint64]*Issue, error) { - keys, err := t.Store.Keys(IssuesState) - if err != nil { - return nil, fmt.Errorf("the issues cannot be listed: %v", err) - } - out := map[uint64]*Issue{} - for _, k := range keys { - n, ok := numberOfKey(k) - if !ok { - continue - } - is, _, err := read(t.Store, n) - if errors.Is(err, ErrNoIssue) { - continue - } - if err != nil { - return nil, err - } - out[n] = is - } - return out, nil -} - // unresolvedIssues is every issue at open, diagnosing or located, oldest first. func (t *Tracker) unresolvedIssues() ([]*Issue, error) { all, err := t.allIssues() @@ -538,7 +514,7 @@ func (t *Tracker) apply(newer uint64, v *Verdict) ([]string, error) { } } kept := *v - if _, err := t.change(newer, func(i *Issue) error { + if _, err := t.change(newer, Act{Op: EventTriaged, By: GroomBy}, func(i *Issue) error { if i.Triage == nil { i.Triage = &kept } diff --git a/modules/mesh-issues/cmd/mesh-issues/import.go b/modules/mesh-issues/cmd/mesh-issues/import.go index ef60fb16..f6e4e202 100644 --- a/modules/mesh-issues/cmd/mesh-issues/import.go +++ b/modules/mesh-issues/cmd/mesh-issues/import.go @@ -600,29 +600,42 @@ func differences(kept, folder *Issue) []string { return d } -// importCreate writes a new issue under its number, never over one. +// importCreate writes a new issue under its number, never over one: its first operation appended to the log at +// expect 0 (novox/hq ADR 0300 §1), then its fold to the bucket. func (t *Tracker) importCreate(is *Issue, out *Imported) error { - _, err := t.Store.Create(IssuesState, keyOf(is.Number), is) - if errors.Is(err, stdio.ErrChanged) { + if there, _, err := read(t.Store, is.Number); err == nil && there.Opening != is.Opening { return fmt.Errorf("an issue %d appeared while the import ran; nothing was overwritten. Run the import again", is.Number) } - if err != nil { - // A write that timed out may still have landed: read back before saying it did not. - if there, _, rerr := read(t.Store, is.Number); rerr == nil && there.Opening == is.Opening { - t.announceImport(is, out) - return nil - } - return fmt.Errorf("issue %d could not be written: %v", is.Number, err) + seq, err := t.begin(is, Act{Op: OpImported, By: importer(is)}) + if errors.Is(err, errBegun) { + return fmt.Errorf("an issue %d appeared while the import ran; nothing was overwritten. Run the import again", is.Number) } + if errors.Is(err, errUnsure) { + return fmt.Errorf("issue %d may have been written: the log did not confirm it (%v); run the import again, "+ + "which finds it the same or creates it", is.Number, err) + } + if err != nil { + return fmt.Errorf("issue %d could not be written: the log did not take it (%v)", is.Number, err) + } + is.Folded = seq + t.store(is.Number, is, 0) t.announceImport(is, out) return nil } +// importer is who ran the import that took an issue from hq. +func importer(is *Issue) string { + if is.Imported != nil && is.Imported.By != "" { + return is.Imported.By + } + return ImportedBy +} + // importFill takes the status and the fields of an issue the tracker holds without a status: opened with a // folder before slice 3, whose status lived in that folder. Its title, report, opening, notes and links // stay; links the report names are added. func (t *Tracker) importFill(from *Issue, out *Imported) error { - is, err := t.change(from.Number, func(is *Issue) error { + is, err := t.change(from.Number, Act{Op: OpImported, By: importer(from)}, func(is *Issue) error { if is.Status != "" { return fmt.Errorf("issue %d took a status while the import ran; nothing was changed", is.Number) } @@ -663,7 +676,7 @@ func (t *Tracker) announceImport(is *Issue, out *Imported) { // fillOpen gives an issue the tracker holds without a status — opened with a folder before slice 3, whose // folder hq does not hold at the import's commit — the status open, so no issue is left without one. func (t *Tracker) fillOpen(n uint64, commit string, at time.Time, by string) error { - is, err := t.change(n, func(is *Issue) error { + is, err := t.change(n, Act{Op: OpImported, By: by}, func(is *Issue) error { if is.Status != "" { return fmt.Errorf("issue %d took a status while the import ran; nothing was changed", is.Number) } diff --git a/modules/mesh-issues/cmd/mesh-issues/import_test.go b/modules/mesh-issues/cmd/mesh-issues/import_test.go index 46b73449..0e01a0d4 100644 --- a/modules/mesh-issues/cmd/mesh-issues/import_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/import_test.go @@ -1,7 +1,6 @@ package main import ( - "errors" "os" "path/filepath" "reflect" @@ -304,7 +303,7 @@ func TestTheImportRefusesWhatItMustNotWrite(t *testing.T) { // An issue that appears under the number while the import runs is never overwritten. h = newHarness(t) - h.mem.fail[IssuesState] = errors.New("the bus is away") + h.log.failNext = createAttempts out, err = h.Import(ImportAsk{}, newSample(t)) if err != nil { t.Fatal(err) diff --git a/modules/mesh-issues/cmd/mesh-issues/issue.go b/modules/mesh-issues/cmd/mesh-issues/issue.go index 4a9e8b01..8bf96b8a 100644 --- a/modules/mesh-issues/cmd/mesh-issues/issue.go +++ b/modules/mesh-issues/cmd/mesh-issues/issue.go @@ -118,6 +118,9 @@ type Issue struct { // slice 3: when, by whom, at which commit of hq, from which document. An issue the import wrote has // OpenedBy ImportedBy, and its Opened is the date its report gave. Imported *Import `json:"imported,omitempty"` + // Folded is the sequence in the log `changes` of the last operation this issue folds (novox/hq ADR 0300 + // §3); 0 for an issue kept before the switch and not carried into the log yet. + Folded uint64 `json:"folded,omitempty"` } // ImportedBy is who opened an issue the import wrote: not a caller of `open`, and not a name anybody typed. diff --git a/modules/mesh-issues/cmd/mesh-issues/log_test.go b/modules/mesh-issues/cmd/mesh-issues/log_test.go new file mode 100644 index 00000000..969391fb --- /dev/null +++ b/modules/mesh-issues/cmd/mesh-issues/log_test.go @@ -0,0 +1,574 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + "testing" + "time" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// fakeLog is the module's log kept in this process with the runtime's semantics (novox/hq ADR 0297, mesh-sdk +// go/log.go): an append at expect is refused with stdio.ErrChanged when the key's last sequence is another, +// 0 standing for a key with no entry; a read pages from a sequence, narrowed to one key or none. +type fakeLog struct { + mu sync.Mutex + entries []stdio.LogEntry + // fail, when set, is what every append answers, writing nothing: a full log, or a bus that is down. + fail error + // failNext is how many of the next appends fail, writing nothing, as a bus that does not answer. + failNext int + // lost, when set, is what the next append answers after it was written: an answer lost on the way. + lost error + // between, when set, runs once before the next append is judged: another writer coming in between. + between func() + clock time.Time +} + +func newFakeLog() *fakeLog { + return &fakeLog{clock: time.Date(2026, 10, 12, 9, 0, 0, 0, time.UTC)} +} + +func (l *fakeLog) lastOf(key string) uint64 { + for i := len(l.entries) - 1; i >= 0; i-- { + if l.entries[i].Key == key { + return l.entries[i].Sequence + } + } + return 0 +} + +func (l *fakeLog) AppendAt(key string, value any, expect uint64) (uint64, error) { + l.mu.Lock() + if b := l.between; b != nil { + l.between = nil + l.mu.Unlock() + b() + l.mu.Lock() + } + defer l.mu.Unlock() + if l.fail != nil { + return 0, l.fail + } + if l.failNext > 0 { + l.failNext-- + return 0, errors.New("the bus did not answer") + } + raw, err := json.Marshal(value) + if err != nil { + return 0, err + } + 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) + } + seq := uint64(len(l.entries) + 1) + l.entries = append(l.entries, stdio.LogEntry{Sequence: seq, Key: key, Value: raw, At: l.clock}) + if lost := l.lost; lost != nil { + l.lost = nil + return 0, lost + } + return seq, nil +} + +func (l *fakeLog) Read(from uint64, keys string, limit int) (*stdio.LogPage, error) { + l.mu.Lock() + defer l.mu.Unlock() + if limit <= 0 { + limit = 100 + } + limit = min(limit, 1000) + p := &stdio.LogPage{Entries: []stdio.LogEntry{}, Last: uint64(len(l.entries))} + for _, e := range l.entries { + if e.Sequence < from || (keys != "" && e.Key != keys) { + continue + } + if len(p.Entries) == limit { + p.Next = e.Sequence + break + } + p.Entries = append(p.Entries, e) + } + 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() + l.mu.Lock() + defer l.mu.Unlock() + var out []Operation + for _, e := range l.entries { + if e.Key != keyOf(n) { + continue + } + var op Operation + if err := json.Unmarshal(e.Value, &op); err != nil { + t.Fatal(err) + } + out = append(out, op) + } + return out +} + +func (l *fakeLog) size() int { + l.mu.Lock() + defer l.mu.Unlock() + return len(l.entries) +} + +// another is a second tracker on the same bucket and log: another instance, or the same after a restart. +func (h *harness) another(t *testing.T) *harness { + t.Helper() + o := newHarness(t) + o.mem, o.log = h.mem, h.log + o.Tracker.Store, o.Tracker.Log = h.mem, h.log + o.settings = h.settings + return o +} + +// §1: a write whose append is refused changes nothing: not the bucket, not the log, and says no event. +func TestAWriteWhoseAppendIsRefusedChangesNothing(t *testing.T) { + h := newHarness(t) + n := h.open(t, "the bucket fills", "mesh-catalog modules/mesh-issues") + bucket, entries, events := h.mem.snapshot(IssuesState), h.log.size(), len(h.said()) + h.log.fail = fmt.Errorf("the log is full: %w", errors.New("maximum bytes exceeded")) + if _, err := h.Note(n, "a note the log refuses"); err == nil || !strings.Contains(err.Error(), "log did not take") { + t.Fatalf("a note whose append is refused answers %v", err) + } + if _, err := h.Move(MoveAsk{Number: n, Status: StatusDiagnosing, By: "fix@g14"}); err == nil { + t.Fatal("a move whose append is refused was answered as made") + } + if got := h.mem.snapshot(IssuesState); fmt.Sprint(got) != fmt.Sprint(bucket) { + t.Fatalf("the bucket changed:\n%v\n%v", bucket, got) + } + if h.log.size() != entries || len(h.said()) != events { + t.Fatalf("the log holds %d entries (was %d), %d events said (was %d)", h.log.size(), entries, len(h.said()), events) + } +} + +// §1: an operation that came in between refuses the stale append; the change is read again and applied to +// what the log holds, so neither is lost. +func TestAnOperationInBetweenIsNotLost(t *testing.T) { + h := newHarness(t) + n := h.open(t, "two notes at once", "hq") + other := h.another(t) + h.log.between = func() { + if _, err := other.Note(n, "the other's note"); err != nil { + t.Errorf("the other note: %v", err) + } + } + if _, err := h.Note(n, "this note"); err != nil { + t.Fatal(err) + } + is, err := h.Get(n) + if err != nil { + t.Fatal(err) + } + if len(is.Notes) != 2 || is.Notes[0].Text != "the other's note" || is.Notes[1].Text != "this note" { + t.Fatalf("notes %+v", is.Notes) + } + ops := h.log.ops(t, n) + if len(ops) != 3 || ops[0].Op != OpOpened || ops[1].Op != EventNoted || ops[2].Op != EventNoted { + t.Fatalf("the log of %d: %+v", n, ops) + } + if is.Folded != uint64(h.log.size()) { + t.Fatalf("folded at %d, the log's last is %d", is.Folded, h.log.size()) + } +} + +// §2: an open refused at validation takes no number. +func TestAnOpenRefusedAtValidationTakesNoNumber(t *testing.T) { + h := newHarness(t) + if _, err := h.Open(OpenAsk{Title: "", Report: "no title"}); err == nil { + t.Fatal("an open without a title was taken") + } + if _, err := h.Open(OpenAsk{Title: "t", Report: "r", LocatedIn: []string{"nowhere"}}); err == nil { + t.Fatal("an open naming an unregistered owner was taken") + } + if c := h.mem.counter(t); c != 0 || h.log.size() != 0 { + t.Fatalf("counter %d, log %d entries", c, h.log.size()) + } + if n := h.open(t, "the first"); n != 365 { + t.Fatalf("the first open got %d", n) + } +} + +// §2: an append that fails after the number was taken answers the failure and spends that number. +func TestAnAppendFailingAfterTheNumberSpendsIt(t *testing.T) { + h := newHarness(t) + h.log.fail = errors.New("the log is full") + _, err := h.Open(OpenAsk{Title: "refused by a full log", Report: "r"}) + if err == nil || !strings.Contains(err.Error(), "365 is spent") || !strings.Contains(err.Error(), "the log is full") { + t.Fatalf("the open answered %v", err) + } + if c := h.mem.counter(t); c != 365 { + t.Fatalf("the counter is at %d: the number was not taken", c) + } + if len(h.mem.snapshot(IssuesState)) != 0 || h.log.size() != 0 || len(h.said()) != 0 { + t.Fatal("an open the log refused left an issue, an entry or an event") + } + h.log.fail = nil + if n := h.open(t, "the next"); n != 366 { + t.Fatalf("the next open got %d, not 366", n) + } +} + +// §2: an append whose answer was lost is known as the open's own by its opening id: one issue, one number. +func TestAnAppendWhoseAnswerWasLostIsTheOpensOwn(t *testing.T) { + h := newHarness(t) + h.log.lost = errors.New("timeout") + o, err := h.Open(OpenAsk{Title: "lost answer", Report: "r"}) + if err != nil { + t.Fatal(err) + } + if o.Number != 365 || h.log.size() != 1 { + t.Fatalf("number %d, %d entries", o.Number, h.log.size()) + } + if is, err := h.Get(365); err != nil || is.Folded != 1 { + t.Fatalf("issue %+v, %v", is, err) + } +} + +// writeEverything makes issues with every kind of write: open, move, note, link, set, delivery, triage. +func writeEverything(t *testing.T, h *harness) []uint64 { + t.Helper() + a := h.open(t, "the issues bucket fills toward its cap", "mesh-catalog modules/mesh-issues") + b := h.open(t, "status changes are kept for seven days", "mesh-controller") + c := h.open(t, "an unrelated one") + if _, err := h.Note(a, "measured 10.9 MB of 64 MiB"); err != nil { + t.Fatal(err) + } + if _, err := h.Link(LinkAsk{Number: b, Kind: LinkRelatesTo, Target: keyOf(a)}); err != nil { + t.Fatal(err) + } + if _, err := h.Set(SetAsk{Number: a, Severity: "major", Needs: "decision"}); err != nil { + t.Fatal(err) + } + if _, err := h.Move(MoveAsk{Number: a, Status: StatusFixing, By: "fix@g14", FixedBy: []string{"novox/mesh-catalog#1"}}); err != nil { + t.Fatal(err) + } + if _, err := h.Move(MoveAsk{Number: c, Status: StatusWontfix, Reason: "not ours"}); err != nil { + t.Fatal(err) + } + if _, err := h.Link(LinkAsk{Number: b, Kind: LinkRelatesTo, Target: keyOf(a), Remove: true}); err != nil { + t.Fatal(err) + } + return []uint64{a, b, c} +} + +// §3: the bucket emptied, the fold on start restores every issue exactly. +func TestTheFoldOnStartRestoresAnEmptiedBucketExactly(t *testing.T) { + h := newHarness(t) + writeEverything(t, h) + before := h.mem.snapshot(IssuesState) + if len(before) != 3 { + t.Fatalf("%d issues", len(before)) + } + h.mem.empty(IssuesState) + r := h.another(t) + s, err := r.Start() + if err != nil { + t.Fatal(err) + } + if s.Rebuilt != 3 || s.Issues != 3 || len(s.Faults) != 0 { + t.Fatalf("started %+v", s) + } + after := h.mem.snapshot(IssuesState) + for k, v := range before { + if after[k] != v { + t.Errorf("issue %s was\n%s\nand is folded as\n%s", k, v, after[k]) + } + } + if len(after) != len(before) { + t.Fatalf("%d issues after, %d before", len(after), len(before)) + } +} + +// §3: a bucket behind its log (the write after the append failed) catches up on start, and at the next change. +func TestABucketBehindItsLogCatchesUp(t *testing.T) { + h := newHarness(t) + n := h.open(t, "crash between the two", "hq") + h.mem.fail[IssuesState] = errors.New("the bus went away") + if _, err := h.Note(n, "kept in the log only"); err != nil { + t.Fatalf("a note the log took is a note made: %v", err) + } + is, _ := h.Get(n) + if len(is.Notes) != 0 { + t.Fatal("the bucket write was meant to fail") + } + r := h.another(t) + s, err := r.Start() + if err != nil { + t.Fatal(err) + } + if s.Refolded != 1 { + t.Fatalf("started %+v", s) + } + if is, _ = h.Get(n); len(is.Notes) != 1 || is.Folded != 2 { + t.Fatalf("after start: %+v", is) + } + + // And at the next change without a start: the stale append is refused, the log folded, the change applied. + h.mem.fail[IssuesState] = errors.New("the bus went away") + if _, err := h.Note(n, "second"); err != nil { + t.Fatal(err) + } + if _, err := h.Note(n, "third"); err != nil { + t.Fatal(err) + } + if is, _ = h.Get(n); len(is.Notes) != 3 || is.Folded != 4 { + t.Fatalf("after the next change: notes %d, folded %d", len(is.Notes), is.Folded) + } +} + +// §3: the counter is kept at or above the highest number in the log. +func TestStartRaisesTheCounterToTheLog(t *testing.T) { + h := newHarness(t) + h.open(t, "one") + h.open(t, "two") + h.mem.put(CountersState, IssueCounter, uint64(365)) + s, err := h.another(t).Start() + if err != nil { + t.Fatal(err) + } + if s.Counter != 366 || h.mem.counter(t) != 366 { + t.Fatalf("counter %d, kept %d", s.Counter, h.mem.counter(t)) + } + if n := h.open(t, "three"); n != 367 { + t.Fatalf("the next open got %d", n) + } +} + +// §4: each write still says its event, alongside its one operation in the log. +func TestEachWriteSaysItsEventAlongsideItsOperation(t *testing.T) { + h := newHarness(t) + type step struct { + event string + do func() error + } + var n uint64 + steps := []step{ + {EventOpened, func() error { + o, err := h.Open(OpenAsk{Title: "t", Report: "r", LocatedIn: []string{"hq"}}) + n = o.Number + return err + }}, + {EventNoted, func() error { _, err := h.Note(n, "a note"); return err }}, + {EventLinked, func() error { _, err := h.Link(LinkAsk{Number: n, Kind: LinkDecision, Target: "0300"}); return err }}, + {EventSet, func() error { _, err := h.Set(SetAsk{Number: n, Severity: "minor"}); return err }}, + {EventMoved, func() error { _, err := h.Move(MoveAsk{Number: n, Status: StatusLocated}); return err }}, + } + for i, s := range steps { + events, entries := len(h.said()), h.log.size() + if err := s.do(); err != nil { + t.Fatalf("%s: %v", s.event, err) + } + said := h.said() + if len(said) != events+1 || said[len(said)-1].Event != s.event { + t.Fatalf("step %d said %v", i, said[events:]) + } + ops := h.log.ops(t, n) + if h.log.size() != entries+1 || ops[len(ops)-1].Op != s.event || ops[len(ops)-1].By != Unproven { + t.Fatalf("step %d: %d entries (was %d), last %+v", i, h.log.size(), entries, ops[len(ops)-1]) + } + } + // A note is kept as itself, not as every note again. + if op := h.log.ops(t, n)[1]; op.Set != nil || len(op.Add) != 1 || op.Add["notes"] == nil { + t.Fatalf("the note's operation is %+v", op) + } +} + +// §5: before pages through more than 500 issues, without a gap or a repeat. +func TestBeforePagesThroughMoreThanFiveHundredIssues(t *testing.T) { + h := newHarness(t) + const issues = 530 + for i := 0; i < issues; i++ { + h.open(t, fmt.Sprintf("issue %d", i)) + } + if _, err := h.Start(); err != nil { + t.Fatal(err) + } + seen := map[uint64]bool{} + var last uint64 = 0 + var before uint64 + pages := 0 + for { + l, err := h.List(ListAsk{Limit: 100, Before: before}) + if err != nil { + t.Fatal(err) + } + pages++ + for _, r := range l.Rows { + if seen[r.Number] { + t.Fatalf("%d repeated", r.Number) + } + if last != 0 && r.Number != last-1 { + t.Fatalf("a gap from %d to %d", last, r.Number) + } + seen[r.Number], last = true, r.Number + } + if l.Next == 0 { + break + } + before = l.Next + } + if len(seen) != issues || pages != 6 || last != 365 { + t.Fatalf("%d issues seen in %d pages, ending at %d", len(seen), pages, last) + } + // Over the tool too, as JSON gives the cursor. + out, err := h.call(t, "list", map[string]any{"before": 400, "limit": 2}) + if err != nil { + t.Fatal(err) + } + if l := out.(*Listed); len(l.Rows) != 2 || l.Rows[0].Number != 399 || l.Next != 398 { + t.Fatalf("listed %+v", l) + } +} + +// §5: a search is ranked by relevance; the index is rebuilt on start and kept current by the bucket's watch. +func TestASearchIsRankedAndTheIndexRebuiltOnStart(t *testing.T) { + h := newHarness(t) + full := h.open(t, "the issues bucket fills toward its cap") + h.open(t, "a bucket of the controller") + week := h.open(t, "status changes are kept for seven days") + h.open(t, "nothing alike") + if _, err := h.Note(week, "the events stream drops them after a week; the bucket is not involved"); err != nil { + t.Fatal(err) + } + r := h.another(t) + if _, err := r.Start(); err != nil { + t.Fatal(err) + } + if !r.view().ready() { + t.Fatal("the index was not built on start") + } + l, err := r.List(ListAsk{Search: "issues bucket cap"}) + if err != nil { + t.Fatal(err) + } + if len(l.Rows) != 3 || l.Rows[0].Number != full { + t.Fatalf("ranked %+v", l.Rows) + } + // A note's words are found: notes are indexed. + if l, _ = r.List(ListAsk{Search: "week"}); len(l.Rows) != 1 || l.Rows[0].Number != week { + t.Fatalf("a note's word: %+v", l.Rows) + } + // Another instance's write reaches this one's index through the watch of the bucket. + if _, err := h.Note(full, "seen again by the operator on shanks"); err != nil { + t.Fatal(err) + } + e, _ := h.mem.Get(IssuesState, keyOf(full)) + if err := r.Watched(stdio.StateChange{State: IssuesState, Key: keyOf(full), Op: "put", Value: e.Value}); err != nil { + t.Fatal(err) + } + if l, _ = r.List(ListAsk{Search: "operator"}); len(l.Rows) != 1 || l.Rows[0].Number != full { + t.Fatalf("after the watch: %+v", l.Rows) + } + if _, err := r.List(ListAsk{Search: "bucket", Before: 400}); err == nil { + t.Fatal("a search with before was taken") + } +} + +// keptBefore puts an issue in the bucket as the tracker kept it before the switch: values, no log. +func keptBefore(t *testing.T, h *harness, n uint64, titles ...string) { + t.Helper() + for _, title := range titles { + h.mem.put(IssuesState, keyOf(n), Issue{Number: n, Title: title, Status: StatusOpen, Report: "r", + Opened: time.Date(2026, 10, 10, 0, 0, 0, 0, time.UTC), OpenedBy: "x", Opening: fmt.Sprint("o", n)}) + } +} + +// §6: the carry appends each value the bucket keeps, oldest first, as carried operations holding the whole +// issue; run again it carries nothing; a write after it is one entry in the log. +func TestTheCarryIsIdempotentAndCarriesEveryValueKept(t *testing.T) { + h := newHarness(t) + keptBefore(t, h, 300, "first title", "second title", "third title") + keptBefore(t, h, 301, "only") + c, err := h.Carry() + if err != nil { + t.Fatal(err) + } + if c.Issues != 2 || c.Values != 4 || !c.History { + t.Fatalf("carried %+v", c) + } + ops := h.log.ops(t, 300) + if len(ops) != 3 { + t.Fatalf("%d operations of 300", len(ops)) + } + for i, title := range []string{"first title", "second title", "third title"} { + if ops[i].Op != OpCarried || ops[i].Issue == nil || ops[i].Issue.Title != title { + t.Fatalf("operation %d: %+v", i, ops[i]) + } + } + is, _ := h.Get(300) + if is.Title != "third title" || is.Folded != 3 { + t.Fatalf("300 is %+v", is) + } + again, err := h.Carry() + if err != nil { + t.Fatal(err) + } + if again.Issues != 0 || again.Values != 0 || again.Already != 2 || h.log.size() != 4 { + t.Fatalf("carried again %+v, %d entries", again, h.log.size()) + } + if _, err := h.Move(MoveAsk{Number: 300, Status: StatusDiagnosing, By: "fix@g14"}); err != nil { + t.Fatal(err) + } + if h.log.size() != 5 || h.log.ops(t, 300)[3].Op != EventMoved { + t.Fatalf("a move after the carry: %d entries", h.log.size()) + } + // The start carries too, and finds nothing more. + s, err := h.another(t).Start() + if err != nil || s.Carried.Issues != 0 || h.log.size() != 5 { + t.Fatalf("start %+v, %v", s, err) + } +} + +// §6: a write to an issue not carried yet carries it first, so its first operation in the log is a carried one. +func TestAWriteBeforeTheCarryCarriesItsIssueFirst(t *testing.T) { + h := newHarness(t) + keptBefore(t, h, 300, "before", "now") + if _, err := h.Note(300, "after the switch"); err != nil { + t.Fatal(err) + } + ops := h.log.ops(t, 300) + if len(ops) != 3 || ops[0].Op != OpCarried || ops[1].Op != OpCarried || ops[2].Op != EventNoted { + t.Fatalf("the log of 300: %+v", ops) + } + h.mem.empty(IssuesState) + if _, err := h.another(t).Start(); err != nil { + t.Fatal(err) + } + if is, _ := h.Get(300); is.Title != "now" || len(is.Notes) != 1 { + t.Fatalf("300 folds to %+v", is) + } +} + +// The counter behind an issue the bucket kept before the switch: the open refuses, overwrites nothing. +func TestAnOpenOverAnIssueKeptBeforeTheSwitchOverwritesNothing(t *testing.T) { + h := newHarness(t) + keptBefore(t, h, 365, "kept") + h.mem.put(CountersState, IssueCounter, uint64(364)) + if _, err := h.Open(OpenAsk{Title: "new", Report: "r"}); err == nil || !strings.Contains(err.Error(), "fault") { + t.Fatalf("the open answered %v", err) + } + if is, _ := h.Get(365); is.Title != "kept" { + t.Fatalf("365 is %+v", is) + } +} diff --git a/modules/mesh-issues/cmd/mesh-issues/main.go b/modules/mesh-issues/cmd/mesh-issues/main.go index 302a46aa..ffd92a0f 100644 --- a/modules/mesh-issues/cmd/mesh-issues/main.go +++ b/modules/mesh-issues/cmd/mesh-issues/main.go @@ -108,6 +108,8 @@ func main() { if _, err := t.Settings(); err != nil { logf("[mesh-issues] the settings cannot be read (%v): every open and move will say so", err) } + t.Log = runtimeLog{} + go start(t) go hearOpened(t) go sweep(t) if err := stdio.Serve("", Tools(t)); err != nil { @@ -116,6 +118,30 @@ func main() { } } +// start watches the module's own bucket into memory, then carries, folds and indexes (novox/hq ADR 0300 §3, §5, +// §6), each asked again until it is done: the runtime answers only once Serve runs. Until it is done, a list +// reads the bucket, and a write carries the issue it changes first, so no write reaches the log before its +// issue's carry. +func start(t *Tracker) { + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + time.Sleep(wait) + if err := stdio.State(IssuesState).Watch("", t.Watched); err != nil { + logf("[mesh-issues] the bucket is not watched yet (%v); asking again in %s", err, wait) + continue + } + break + } + for wait := 2 * time.Second; ; wait = min(wait*2, time.Minute) { + s, err := t.Start() + if err == nil { + logf("[mesh-issues] started: %s", s.Said) + return + } + logf("[mesh-issues] the start did not finish (%v); trying again in %s", err, wait) + time.Sleep(wait) + } +} + // hearOpened subscribes the triage to the tracker's own `opened` and `moved` events and to mesh-delivery's // `transition` (ADR 0278 §1, ADR 0279 §3), each asked again until the runtime takes it: the runtime answers only // once Serve runs. @@ -376,7 +402,8 @@ func Tools(t *Tracker) []stdio.Tool { "taken over. flag delivered-not-resolved keeps the issues in fixing whose every fix (fixed-by and the " + "pull requests linked) is delivered; flag abandoned the issues whose worker did nothing on them for a " + "day with no pull request open, which a move takes over without a reason. Each row says both. Answers " + - "what is open in one call.", + "what is open in one call. search ranks the issues whose title, report or notes hold its words, best first; " + + "before pages the newest-first list: give the answer's next to read on below it.", Input: schema(nil, map[string]any{ "status": map[string]any{"type": "string", "enum": Statuses, "description": "only issues at this status"}, "owner": text("only issues located in this owner, such as mesh-controller"), @@ -386,9 +413,11 @@ func Tools(t *Tracker) []stdio.Tool { "needs": map[string]any{"type": "string", "enum": NeedsKinds, "description": "only issues that need this"}, "ready": map[string]any{"type": "boolean", "description": "only issues ready (true) or waiting (false)"}, "flag": map[string]any{"type": "string", "enum": Flags, "description": "only issues delivered and not resolved, or the abandoned ones"}, + "search": text("rank the issues whose title, report or notes hold these words, best first"), + "before": number("only issues numbered below this: the next of the page before, to read on"), }), Run: func(m map[string]any) (any, error) { - a, err := readArgs("list", m, "status", "owner", "phrase", "limit", "severity", "needs", "ready", "flag") + a, err := readArgs("list", m, "status", "owner", "phrase", "limit", "severity", "needs", "ready", "flag", "search", "before") if err != nil { return nil, err } @@ -411,6 +440,12 @@ func Tools(t *Tracker) []stdio.Tool { if ask.Flag, err = a.str("flag"); err != nil { return nil, err } + if ask.Search, err = a.str("search"); err != nil { + return nil, err + } + if ask.Before, err = a.number("before", false); err != nil { + return nil, err + } if v, given := m["ready"]; given && v != nil { r, err := a.flag("ready") if err != nil { diff --git a/modules/mesh-issues/cmd/mesh-issues/manifest_test.go b/modules/mesh-issues/cmd/mesh-issues/manifest_test.go index 2bf6dbac..655656eb 100644 --- a/modules/mesh-issues/cmd/mesh-issues/manifest_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/manifest_test.go @@ -22,6 +22,7 @@ type manifest struct { Consumes []string `json:"consumes"` Invokes []string `json:"invokes"` State []json.RawMessage `json:"state"` + Logs []string `json:"logs"` Tools []string `json:"tools"` Resources []struct { ID string `json:"id"` @@ -87,10 +88,14 @@ func TestItDeclaresAndHoldsTheIssueTrackerSeatOnceForTheMesh(t *testing.T) { EventTriaged, EventReady, EventReadyToVerify, EventReserved}) { t.Fatalf("emits %v", m.Emits) } - // Four states, none expiring: issues keep a short history per key, the counters one value each, the - // backlog's pass of the duplicate check one value (ADR 0278 §6), each reserved number one value (issue 424). + // The record is the log `changes` (ADR 0300 §1). Four states, none expiring, each one value per key: issues + // the fold of the log (§3), the counters, the backlog's pass of the duplicate check (ADR 0278 §6), each + // reserved number (issue 424). + if !reflect.DeepEqual(m.Logs, []string{ChangesLog}) { + t.Fatalf("logs %v", m.Logs) + } d := m.stateDeclarations(t) - if len(d) != 4 || d[0].Name != IssuesState || d[0].History < 2 || d[0].TTLSeconds != 0 || + if len(d) != 4 || d[0].Name != IssuesState || d[0].History > 1 || d[0].TTLSeconds != 0 || d[1].Name != CountersState || d[1].History > 1 || d[1].TTLSeconds != 0 || d[2].Name != GroomingState || d[2].History > 1 || d[2].TTLSeconds != 0 || d[3].Name != ReservationsState || d[3].History > 1 || d[3].TTLSeconds != 0 { diff --git a/modules/mesh-issues/cmd/mesh-issues/memory_test.go b/modules/mesh-issues/cmd/mesh-issues/memory_test.go index f0d663ad..14962dfd 100644 --- a/modules/mesh-issues/cmd/mesh-issues/memory_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/memory_test.go @@ -22,10 +22,14 @@ type memory struct { data map[string]map[string]Entry // fail, when set, is the error the next write of that state answers. fail map[string]error + // hist is every value each key was given, the last eight kept, as the bucket kept them before the tracker + // switched to its log (novox/hq ADR 0300 §6). + hist map[string]map[string][]Entry } func newMemory() *memory { - return &memory{data: map[string]map[string]Entry{IssuesState: {}, CountersState: {}, GroomingState: {}, ReservationsState: {}}, fail: map[string]error{}} + return &memory{data: map[string]map[string]Entry{IssuesState: {}, CountersState: {}, GroomingState: {}, ReservationsState: {}}, fail: map[string]error{}, + hist: map[string]map[string][]Entry{}} } func (m *memory) bucket(state string) (map[string]Entry, error) { @@ -71,9 +75,35 @@ func (m *memory) write(state, key string, value any, check func(Entry, bool) boo } m.rev++ b[key] = Entry{Value: raw, Revision: m.rev} + if m.hist[state] == nil { + m.hist[state] = map[string][]Entry{} + } + kept := append(m.hist[state][key], b[key]) + if len(kept) > 8 { + kept = kept[len(kept)-8:] + } + m.hist[state][key] = kept return m.rev, nil } +// History is every value the key still keeps, oldest first. +func (m *memory) History(state, key string) ([]Entry, error) { + m.mu.Lock() + defer m.mu.Unlock() + if _, ok := m.data[state][key]; !ok { + return nil, nil + } + return append([]Entry(nil), m.hist[state][key]...), nil +} + +// empty takes every key out of a state, as an emptied bucket. +func (m *memory) empty(state string) { + m.mu.Lock() + defer m.mu.Unlock() + m.data[state] = map[string]Entry{} + delete(m.hist, state) +} + func (m *memory) Create(state, key string, value any) (uint64, error) { return m.write(state, key, value, func(_ Entry, there bool) bool { return !there }) } @@ -154,6 +184,7 @@ type said struct { type harness struct { *Tracker mem *memory + log *fakeLog mu sync.Mutex events []said settings Settings @@ -162,12 +193,13 @@ type harness struct { func newHarness(t *testing.T) *harness { t.Helper() - h := &harness{mem: newMemory(), settings: Settings{FirstNumber: 365, Owners: []string{ + h := &harness{mem: newMemory(), log: newFakeLog(), settings: Settings{FirstNumber: 365, Owners: []string{ "hq", "mesh-catalog", "mesh-controller", "mesh-host", "mesh-sdk", "mesh-tools", "mesh-catalog modules/nats", "mesh-catalog modules/records", "mesh-catalog modules/mesh-issues"}}} clock := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC) h.Tracker = &Tracker{ Store: h.mem, + Log: h.log, Emit: func(event string, body any) error { h.mu.Lock() defer h.mu.Unlock() diff --git a/modules/mesh-issues/cmd/mesh-issues/ready.go b/modules/mesh-issues/cmd/mesh-issues/ready.go index f2d989b2..07e66006 100644 --- a/modules/mesh-issues/cmd/mesh-issues/ready.go +++ b/modules/mesh-issues/cmd/mesh-issues/ready.go @@ -45,6 +45,14 @@ type SetDone struct { // Set changes an issue's severity, needs or suggested owner, as a person decides. func (t *Tracker) Set(a SetAsk) (*SetDone, error) { return t.set(t.by(), a, false) } +// setOp is the kind of operation a set is in the log: a person's set, or the triage filling what is empty. +func setOp(onlyEmpty bool) string { + if onlyEmpty { + return EventTriaged + } + return EventSet +} + // set is Set as the one who acts. onlyEmpty fills only a value that is empty, as the triage does: a value a // person gave meanwhile stands. func (t *Tracker) set(by string, a SetAsk, onlyEmpty bool) (*SetDone, error) { @@ -72,7 +80,7 @@ func (t *Tracker) set(by string, a SetAsk, onlyEmpty bool) (*SetDone, error) { return nil, errors.New("set changes severity, needs or suggested-owner: give at least one") } var changed []string - is, err := t.change(a.Number, func(is *Issue) error { + is, err := t.change(a.Number, Act{Op: setOp(onlyEmpty), By: by}, func(is *Issue) error { changed = nil for _, f := range []struct { name string @@ -344,7 +352,7 @@ func (t *Tracker) letGo(dep uint64) error { deps := dependsOn(is) sort.Slice(deps, func(i, j int) bool { return deps[i] < deps[j] }) claimed := false - if _, err := t.change(w, func(i *Issue) error { + if _, err := t.change(w, Act{Op: EventReady, By: Seat}, func(i *Issue) error { claimed = false if equalNumbers(i.ReadyFor, deps) { return nil diff --git a/modules/mesh-issues/cmd/mesh-issues/tracker.go b/modules/mesh-issues/cmd/mesh-issues/tracker.go index 0afc420d..e735ab26 100644 --- a/modules/mesh-issues/cmd/mesh-issues/tracker.go +++ b/modules/mesh-issues/cmd/mesh-issues/tracker.go @@ -9,6 +9,7 @@ import ( "math/big" "sort" "strings" + "sync" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -19,7 +20,7 @@ import ( // change as an event. type Tracker struct { Store Store - // Emit says an event as this module (ADR 0041); the events are the audit trail. + // Emit says an event as this module (ADR 0041); the events announce, the log is the audit trail (ADR 0300 §4). Emit func(event string, body any) error Settings func() (Settings, error) Now func() time.Time @@ -39,7 +40,11 @@ type Tracker struct { Notify Notifier // CallerOf is who called, for the arguments one call was handed (stdio.CallerOf); nil proves nobody. CallerOf func(map[string]any) string + // Log is the tracker's record, the log `changes` (novox/hq ADR 0300); the bucket `issues` is its fold. + Log Log groom grooming + held *memoryView + heldOnce sync.Once } // The events this module says, one per kind of change. @@ -122,9 +127,10 @@ type Opened struct { Warning string `json:"warning,omitempty"` } -// Open checks an issue, then takes the next number and writes the issue under it (to-be 53 §4). A refused -// issue takes no number. A number taken and then not written — the bus failing between the two — stays -// unused, and the answer says which: numbers may skip, never repeat. +// Open checks an issue, then takes the next number, appends the issue's first operation to the log at expect +// 0, and writes its fold under its number (to-be 53 §4, novox/hq ADR 0300 §2). An issue refused at validation +// takes no number. A number taken and then not taken by the log — the bus failing between the two, or a full +// log — is spent, and the open answers the failure: numbers may skip, never repeat. func (t *Tracker) Open(a OpenAsk) (*Opened, error) { return t.open(t.by(), a) } // open is Open as the one who acts. @@ -174,9 +180,33 @@ func (t *Tracker) open(by string, a OpenAsk) (*Opened, error) { if record != "" { is.Record = RecordOf(n, record) } - if err := t.create(&is); err != nil { - return nil, err + // The first operation, at expect 0 (ADR 0300 §2): one that fails after the number was taken spends the + // number, and the open answers the failure. + var seq uint64 + if there, _, rerr := read(t.Store, n); rerr == nil && there.Opening != is.Opening { + // Kept from before the switch and not carried yet: the log does not know it, the bucket does. + err = errBegun + } else { + seq, err = t.begin(&is, Act{Op: OpOpened, By: by}) } + if errors.Is(err, errBegun) { + t.Logf("[mesh-issues] the counter handed out %d and an issue %d is already kept: not overwritten", n, n) + return nil, fmt.Errorf("a fault, not your mistake: the counter handed out %d, and an issue %d is already kept. "+ + "Nothing was overwritten and nothing was opened; %d stays unused. Open it again for the next number, "+ + "and tell the operator the counter is behind the issues kept", n, n, n) + } + if errors.Is(err, errUnsure) { + t.Logf("[mesh-issues] number %d was taken and the log's answer to the issue was lost: %v", n, err) + return nil, fmt.Errorf("number %d was taken and the log did not confirm the issue (%v): it may have been written. "+ + "Call get %d before opening it again; if there is no issue %d, it is spent", n, err, n, n) + } + if err != nil { + t.Logf("[mesh-issues] number %d was taken and the log did not take the issue: %v", n, err) + return nil, fmt.Errorf("number %d was taken and the log did not take the issue (%v): nothing was opened and %d "+ + "is spent. Open it again for the next number", n, err, n) + } + is.Folded = seq + t.store(n, &is, 0) out := &Opened{Number: n, Title: title, Status: is.Status, LocatedIn: located, Record: is.Record, By: is.OpenedBy} out.Said = fmt.Sprintf("Opened issue %d: %s.", n, title) if is.Record != "" { @@ -188,45 +218,6 @@ func (t *Tracker) open(by string, a OpenAsk) (*Opened, error) { return out, nil } -// create writes a new issue under its number, never over one: a key that is there already is a fault said -// loudly, and the number stays unused. -func (t *Tracker) create(is *Issue) error { - key := keyOf(is.Number) - var err error - for i := 0; i < createAttempts; i++ { - if i > 0 { - t.Sleep(jitter(time.Duration(i) * 50 * time.Millisecond)) - } - _, err = t.Store.Create(IssuesState, key, is) - if err == nil { - return nil - } - if errors.Is(err, stdio.ErrChanged) { - // Ours, written by an attempt whose answer was lost? - there, _, rerr := read(t.Store, is.Number) - if rerr != nil { - t.Logf("[mesh-issues] issue %d is kept and could not be read back: %v", is.Number, rerr) - return fmt.Errorf("issue %d may have been written by this open: the key is taken and could not be read "+ - "back (%v). Call get %d before opening it again, so it is not opened twice", is.Number, rerr, is.Number) - } - if there.Opening == is.Opening { - return nil - } - t.Logf("[mesh-issues] the counter handed out %d and an issue %d is already kept: not overwritten", is.Number, is.Number) - return fmt.Errorf("a fault, not your mistake: the counter handed out %d, and an issue %d is already kept. "+ - "Nothing was overwritten and nothing was opened; %d stays unused. Open it again for the next number, "+ - "and tell the operator the counter is behind the issues kept", is.Number, is.Number, is.Number) - } - } - // A write that timed out may still have landed: say so, never "nothing was opened". - t.Logf("[mesh-issues] number %d was taken and the issue's write was not confirmed: %v", is.Number, err) - if there, _, rerr := read(t.Store, is.Number); rerr == nil && there.Opening == is.Opening { - return nil - } - return fmt.Errorf("number %d was taken and the issue's write was not confirmed (%v): it may have been written. "+ - "Call get %d before opening it again; if there is no issue %d, it stays unused", is.Number, err, is.Number, is.Number) -} - // next takes the next issue number: the counter advanced only at the revision read, so two openers — in // one instance or two — never take one number (ADR 0275 §3). func (t *Tracker) next() (uint64, error) { @@ -285,38 +276,6 @@ func (t *Tracker) seed() error { return nil } -// change reads an issue, applies f, and writes it only at the revision read; a write another writer won -// is read and applied again, so no note, link or move is lost and none is applied to a stale issue. -func (t *Tracker) change(n uint64, f func(*Issue) error) (*Issue, error) { - for i := 0; i < issueAttempts; i++ { - is, rev, err := read(t.Store, n) - if err != nil { - return nil, err - } - if err := f(is); err != nil { - return nil, err - } - raw, err := json.Marshal(is) - if err != nil { - 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 "+ - "was changed: say it shorter, or in a document the issue links", n, len(raw), MostIssueBytes) - } - _, err = t.Store.Update(IssuesState, keyOf(n), json.RawMessage(raw), rev) - if err == nil { - return is, nil - } - if !errors.Is(err, stdio.ErrChanged) { - return nil, fmt.Errorf("issue %d could not be written (%v); nothing was changed", n, err) - } - t.Sleep(jitter(time.Duration(i+1) * 5 * time.Millisecond)) - } - return nil, fmt.Errorf("issue %d was changed by another caller %d times while this change was made; nothing was "+ - "changed. Try again", n, issueAttempts) -} - // Get is one issue whole. func (t *Tracker) Get(n uint64) (*Issue, error) { is, _, err := read(t.Store, n) @@ -451,7 +410,7 @@ func (t *Tracker) move(by string, a MoveAsk) (*Moved, error) { var from string var changed []string var work workOutcome - is, err := t.change(a.Number, func(is *Issue) error { + is, err := t.change(a.Number, Act{Op: EventMoved, By: by}, func(is *Issue) error { changed = nil from = is.Status at := t.now() @@ -610,7 +569,7 @@ func (t *Tracker) note(by string, n uint64, text string) (*Noted, error) { return nil, err } note := Note{By: by, At: t.now(), Text: text} - is, err := t.change(n, func(is *Issue) error { + is, err := t.change(n, Act{Op: EventNoted, By: by}, func(is *Issue) error { is.Notes = append(is.Notes, note) touch(is, by, note.At) return nil @@ -680,7 +639,7 @@ func (t *Tracker) link(by string, a LinkAsk) (*Linked, error) { } } at := t.now() - is, err := t.change(a.Number, func(is *Issue) error { + is, err := t.change(a.Number, Act{Op: EventLinked, By: by}, func(is *Issue) error { for i, l := range is.Links { if l.Kind == kind && l.Target == target { if !a.Remove { @@ -727,6 +686,10 @@ type ListAsk struct { Ready *bool // Flag keeps only the issues delivered and not resolved, or the abandoned ones (novox/hq issue 483). Flag string + // Search ranks the issues whose title, report or notes hold its words, best first (novox/hq ADR 0300 §5). + Search string + // Before keeps only the issues numbered below it: the cursor that pages the newest-first list. + Before uint64 } // Row is one issue in a list: short enough to read in a chat. @@ -755,12 +718,15 @@ type Row struct { // Listed is the answer to a list. type Listed struct { - Rows []Row `json:"rows"` - Matched int `json:"matched"` - Said string `json:"said"` + Rows []Row `json:"rows"` + Matched int `json:"matched"` + // Next is the before that continues the list where this page ends; 0 when nothing more matches. + Next uint64 `json:"next,omitempty"` + Said string `json:"said"` } -// List is the issues that match, newest first. +// List is the issues that match, newest first, or ranked by a search: from memory once it is loaded on start +// (novox/hq ADR 0300 §5). func (t *Tracker) List(a ListAsk) (*Listed, error) { status := strings.TrimSpace(a.Status) if status != "" && !known(status) { @@ -795,6 +761,19 @@ func (t *Tracker) List(a ListAsk) (*Listed, error) { return nil, err } } + search := "" + if strings.TrimSpace(a.Search) != "" { + if search, err = singleLine("search", a.Search, MostPhrase); err != nil { + return nil, err + } + if len(tokens(search)) == 0 { + return nil, fmt.Errorf("search %q holds no word of two letters or digits", search) + } + if a.Before != 0 { + return nil, errors.New("before pages the newest-first list; a search is ranked best first, so give limit, " + + "or more words, instead") + } + } limit := a.Limit if limit <= 0 { limit = 50 @@ -806,10 +785,30 @@ func (t *Tracker) List(a ListAsk) (*Listed, error) { } now, staleAfter := t.now(), t.workerStaleAfter() var numbers []uint64 - for n := range all { - numbers = append(numbers, n) + if search != "" { + var hits []hit + if v := t.view(); v.ready() { + hits = v.search(search) + } else { + x := newIndex() + for _, is := range all { + x.add(is) + } + hits = x.search(search) + } + for _, h := range hits { + if all[h.Number] != nil { + numbers = append(numbers, h.Number) + } + } + } else { + for n := range all { + if a.Before == 0 || n < a.Before { + numbers = append(numbers, n) + } + } + sort.Slice(numbers, func(i, j int) bool { return numbers[i] > numbers[j] }) } - sort.Slice(numbers, func(i, j int) bool { return numbers[i] > numbers[j] }) out := &Listed{Rows: []Row{}} lower := strings.ToLower(phrase) for _, n := range numbers { @@ -849,13 +848,23 @@ func (t *Tracker) List(a ListAsk) (*Listed, error) { out.Rows = append(out.Rows, row) } } + below := "" + if a.Before != 0 { + below = fmt.Sprintf(" below %d", a.Before) + } switch { case out.Matched == 0: - out.Said = "No issue matches." + out.Said = "No issue matches" + below + "." + case search != "" && out.Matched > len(out.Rows): + out.Said = fmt.Sprintf("%d issues match the search; the best %d are shown, best first.", out.Matched, len(out.Rows)) + case search != "": + out.Said = fmt.Sprintf("%d issues match the search, best first.", out.Matched) case out.Matched > len(out.Rows): - out.Said = fmt.Sprintf("%d issues match; the newest %d are shown.", out.Matched, len(out.Rows)) + out.Next = out.Rows[len(out.Rows)-1].Number + out.Said = fmt.Sprintf("%d issues match%s; the newest %d are shown. The next page is before %d.", out.Matched, + below, len(out.Rows), out.Next) default: - out.Said = fmt.Sprintf("%d issues match, newest first.", out.Matched) + out.Said = fmt.Sprintf("%d issues match%s, newest first.", out.Matched, below) } return out, nil } diff --git a/modules/mesh-issues/cmd/mesh-issues/tracker_test.go b/modules/mesh-issues/cmd/mesh-issues/tracker_test.go index 1b0c8d15..1d9be5a2 100644 --- a/modules/mesh-issues/cmd/mesh-issues/tracker_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/tracker_test.go @@ -106,19 +106,19 @@ func TestAnIssuesKeyIsNeverOverwrittenAndTheFaultIsSaid(t *testing.T) { func TestANumberTakenAndNotWrittenIsSaidAndNeverHandedOutAgain(t *testing.T) { h := newHarness(t) h.open(t, "first") - h.mem.fail[IssuesState] = errors.New("the bus did not answer") + h.log.failNext = 1 // One failure is tried again and succeeds: no gap. if got := h.open(t, "second"); got != 366 { t.Fatalf("got %d", got) } - // Failing every time leaves a gap, said. - failing := &failingStore{memory: h.mem, failCreates: createAttempts} - h.Store = failing + // Failing every time spends the number, said (novox/hq ADR 0300 §2). + h.log.failNext = createAttempts _, err := h.Open(OpenAsk{Title: "third", Report: "r"}) - if err == nil || !strings.Contains(err.Error(), "may have been written") || !strings.Contains(err.Error(), "get 367") { - t.Fatalf("a write that never landed: %v", err) + if err == nil || !strings.Contains(err.Error(), "367 is spent") || !strings.Contains(err.Error(), "did not answer") { + t.Fatalf("an append that never landed: %v", err) } - h.Store = h.mem + // A bucket that takes no write after the log took the issue: the issue is opened, the bucket folds it later. + h.mem.fail[IssuesState] = errors.New("the bus did not answer") if got := h.open(t, "fourth"); got != 368 { t.Fatalf("got %d, want 368: a number is never handed out twice", got) } @@ -189,17 +189,30 @@ func (u unreadable) Get(state, key string) (*Entry, error) { return u.memory.Get(state, key) } -// An open whose write may have landed never says nothing was opened (the review of 2026-10-10). +// unreadableLog loses every append's answer and cannot be read back. +type unreadableLog struct{ *fakeLog } + +func (u unreadableLog) AppendAt(key string, value any, expect uint64) (uint64, error) { + _, _ = u.fakeLog.AppendAt(key, value, expect) + return 0, errors.New("no answer in 30 s") +} +func (u unreadableLog) Last(string) (*stdio.LogEntry, error) { + return nil, errors.New("no answer in 30 s") +} + +// An open whose write may have landed never says nothing was opened (the review of 2026-10-10), in the bucket +// or in the log. func TestAnOpenWhoseWriteMayHaveLandedNeverSaysNothingWasOpened(t *testing.T) { h := newHarness(t) h.Store = lostEvery{h.mem} if o, err := h.Open(OpenAsk{Title: "t", Report: "r"}); err != nil || o.Number != 365 { t.Fatalf("a write that landed, every answer lost: %v %v", o, err) } - h.Store = unreadable{h.mem} + h.Store = h.mem + h.Log = unreadableLog{h.log} _, err := h.Open(OpenAsk{Title: "t", Report: "r"}) if err == nil || strings.Contains(err.Error(), "nothing was opened") || !strings.Contains(err.Error(), "get 366") { - t.Fatalf("a taken key that cannot be read back: %v", err) + t.Fatalf("an append whose answer was lost and cannot be read back: %v", err) } } diff --git a/modules/mesh-issues/cmd/mesh-issues/triage_test.go b/modules/mesh-issues/cmd/mesh-issues/triage_test.go index 98482d7f..9aca9120 100644 --- a/modules/mesh-issues/cmd/mesh-issues/triage_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/triage_test.go @@ -305,7 +305,7 @@ func TestTheSameEventTwiceJudgesOnceAndFinishesWhatIsMissing(t *testing.T) { o2 := h2.open(t, "the counter skips a number") n2 := h2.open(t, "a number is skipped") kept := &Verdict{Verdict: VerdictDuplicate, Issue: o2, Certain: true, Reason: "the same", Severity: "minor", Needs: "none", By: "the agent"} - if _, err := h2.change(n2, func(i *Issue) error { i.Triage = kept; return nil }); err != nil { + if _, err := h2.change(n2, Act{Op: EventTriaged, By: GroomBy}, func(i *Issue) error { i.Triage = kept; return nil }); err != nil { t.Fatal(err) } s2 := withJudge(h2, &stubJudge{err: errors.New("must not be asked")}) @@ -690,7 +690,7 @@ func TestATriagedIssueAPersonResolvedIsNotTouchedAgain(t *testing.T) { older := h.open(t, "the counter skips a number") newer := h.open(t, "a number is skipped") kept := &Verdict{Verdict: VerdictDuplicate, Issue: older, Certain: true, Reason: "same"} - if _, err := h.change(newer, func(i *Issue) error { i.Triage = kept; return nil }); err != nil { + if _, err := h.change(newer, Act{Op: EventTriaged, By: GroomBy}, func(i *Issue) error { i.Triage = kept; return nil }); err != nil { t.Fatal(err) } resolve(t, h, newer) diff --git a/modules/mesh-issues/cmd/mesh-issues/view.go b/modules/mesh-issues/cmd/mesh-issues/view.go new file mode 100644 index 00000000..3f574b61 --- /dev/null +++ b/modules/mesh-issues/cmd/mesh-issues/view.go @@ -0,0 +1,288 @@ +package main + +import ( + "encoding/json" + "errors" + "fmt" + "math" + "sort" + "strings" + "sync" + "unicode" + + stdio "git.novox.be/novox/mesh-sdk/go" +) + +// memoryView is every folded issue in memory, kept current by watching the module's own bucket, with a full-text +// index of title, report and notes (novox/hq ADR 0300 §5). Until it is loaded on start, the tracker reads +// the bucket as before. +type memoryView struct { + mu sync.RWMutex + loaded bool + issues map[uint64]*Issue + index *index +} + +func newMemoryView() *memoryView { return &memoryView{issues: map[uint64]*Issue{}, index: newIndex()} } + +// view is the tracker's issues in memory. +func (t *Tracker) view() *memoryView { + t.heldOnce.Do(func() { t.held = newMemoryView() }) + return t.held +} + +// load holds these issues and rebuilds the index from them. An issue the watch handed over meanwhile is kept +// where it was folded later than the one read. +func (h *memoryView) load(all map[uint64]*Issue) { + h.mu.Lock() + defer h.mu.Unlock() + merged := map[uint64]*Issue{} + for n, is := range all { + merged[n] = is + } + for n, is := range h.issues { + if old := merged[n]; old == nil || old.Folded < is.Folded { + merged[n] = is + } + } + h.issues = merged + h.index = newIndex() + for _, is := range merged { + h.index.add(is) + } + h.loaded = true +} + +func (h *memoryView) ready() bool { + h.mu.RLock() + defer h.mu.RUnlock() + return h.loaded +} + +// put holds an issue unless what is held was folded later. Before the load it is held for the load to merge. +func (h *memoryView) put(is *Issue) { + h.mu.Lock() + defer h.mu.Unlock() + old := h.issues[is.Number] + if old != nil && old.Folded > is.Folded { + return + } + h.issues[is.Number] = is + if h.loaded { + if old != nil { + h.index.remove(old.Number) + } + h.index.add(is) + } +} + +func (h *memoryView) drop(n uint64) { + h.mu.Lock() + defer h.mu.Unlock() + delete(h.issues, n) + if h.loaded { + h.index.remove(n) + } +} + +// all is every issue held, each a copy of its fields: a caller that changes one changes nothing held. +func (h *memoryView) all() map[uint64]*Issue { + h.mu.RLock() + defer h.mu.RUnlock() + out := make(map[uint64]*Issue, len(h.issues)) + for n, is := range h.issues { + c := *is + out[n] = &c + } + return out +} + +func (h *memoryView) search(q string) []hit { + h.mu.RLock() + defer h.mu.RUnlock() + return h.index.search(q) +} + +// remember holds an issue the tracker just wrote, so a list right after a write shows it before the watch +// hands it back. +func (t *Tracker) remember(is *Issue) { + if is == nil { + return + } + c, err := clone(is) + if err != nil { + return + } + t.view().put(c) +} + +// Watched takes one change of the module's own bucket into memory: a put held unless memory holds a later +// fold, a delete dropped (the next start folds it back from the log). +func (t *Tracker) Watched(c stdio.StateChange) error { + n, ok := numberOfKey(c.Key) + if !ok { + return nil + } + if c.Op != "put" { + t.view().drop(n) + return nil + } + var is Issue + if err := json.Unmarshal(c.Value, &is); err != nil { + return fmt.Errorf("issue %d is kept in a form this module cannot read: %v", n, err) + } + t.view().put(&is) + return nil +} + +// readAll is every issue the bucket holds, by number, each read by its key. +func (t *Tracker) readAll() (map[uint64]*Issue, error) { + keys, err := t.Store.Keys(IssuesState) + if err != nil { + return nil, fmt.Errorf("the issues cannot be listed: %v", err) + } + out := map[uint64]*Issue{} + for _, k := range keys { + n, ok := numberOfKey(k) + if !ok { + continue + } + is, _, err := read(t.Store, n) + if errors.Is(err, ErrNoIssue) { + continue + } + if err != nil { + return nil, err + } + out[n] = is + } + return out, nil +} + +// allIssues is every issue kept, by number: from memory once it is loaded, from the bucket before. +func (t *Tracker) allIssues() (map[uint64]*Issue, error) { + if v := t.view(); v.ready() { + return v.all(), nil + } + return t.readAll() +} + +// index is a full-text index of the issues' title, report and notes, ranked by BM25. Written here rather than +// taken from a library (bleve): the tracker holds a few thousand issues, the index is rebuilt on every start in +// well under a second, and its whole need — words in, issues ranked out — is a hundred lines, against a +// dependency of dozens of modules for a bundle that has two. +type index struct { + // postings is, for each word, how often it stands in each issue, the title's words counted titleWeight + // times. + postings map[string]map[uint64]int + // lengths is each issue's words, weighted as postings counts them; total their sum. + lengths map[uint64]int + words map[uint64][]string + total int +} + +const titleWeight = 3 + +func newIndex() *index { + return &index{postings: map[string]map[uint64]int{}, lengths: map[uint64]int{}, words: map[uint64][]string{}} +} + +// tokens is a text's words: lower case, letters and digits, two characters at least. +func tokens(s string) []string { + var out []string + for _, w := range strings.FieldsFunc(strings.ToLower(s), func(r rune) bool { return !unicode.IsLetter(r) && !unicode.IsDigit(r) }) { + if len([]rune(w)) >= 2 { + out = append(out, w) + } + } + return out +} + +func (x *index) add(is *Issue) { + counts := map[string]int{} + for _, w := range tokens(is.Title) { + counts[w] += titleWeight + } + body := []string{is.Report} + for _, n := range is.Notes { + body = append(body, n.Text) + } + for _, w := range tokens(strings.Join(body, "\n")) { + counts[w]++ + } + length := 0 + words := make([]string, 0, len(counts)) + for w, c := range counts { + p := x.postings[w] + if p == nil { + p = map[uint64]int{} + x.postings[w] = p + } + p[is.Number] = c + length += c + words = append(words, w) + } + x.lengths[is.Number] = length + x.words[is.Number] = words + x.total += length +} + +func (x *index) remove(n uint64) { + for _, w := range x.words[n] { + if p := x.postings[w]; p != nil { + delete(p, n) + if len(p) == 0 { + delete(x.postings, w) + } + } + } + x.total -= x.lengths[n] + delete(x.lengths, n) + delete(x.words, n) +} + +// hit is one issue a search found, and how well it matched. +type hit struct { + Number uint64 + Score float64 +} + +// search ranks every issue holding at least one of the query's words by BM25: a word few issues hold weighs +// more, a word standing often weighs more with diminishing return, a long issue weighs each word less. Equal +// scores rank the newer first. +func (x *index) search(q string) []hit { + docs := len(x.lengths) + if docs == 0 { + return nil + } + const k1, b = 1.2, 0.75 + avg := float64(x.total) / float64(docs) + scores := map[uint64]float64{} + seen := map[string]bool{} + for _, w := range tokens(q) { + if seen[w] { + continue + } + seen[w] = true + p := x.postings[w] + if len(p) == 0 { + continue + } + idf := math.Log(1 + (float64(docs)-float64(len(p))+0.5)/(float64(len(p))+0.5)) + for n, tf := range p { + f := float64(tf) + scores[n] += idf * f * (k1 + 1) / (f + k1*(1-b+b*float64(x.lengths[n])/avg)) + } + } + out := make([]hit, 0, len(scores)) + for n, s := range scores { + out = append(out, hit{n, s}) + } + sort.Slice(out, func(i, j int) bool { + if out[i].Score != out[j].Score { + return out[i].Score > out[j].Score + } + return out[i].Number > out[j].Number + }) + return out +} diff --git a/modules/mesh-issues/go.mod b/modules/mesh-issues/go.mod index 9b0e32bb..d10e51b0 100644 --- a/modules/mesh-issues/go.mod +++ b/modules/mesh-issues/go.mod @@ -3,7 +3,7 @@ module mesh-issues go 1.25.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011091942-4cf45b5ca0cf + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011180852-5794ccd569be github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.51.0 ) diff --git a/modules/mesh-issues/go.sum b/modules/mesh-issues/go.sum index 70e80d38..c7036ba2 100644 --- a/modules/mesh-issues/go.sum +++ b/modules/mesh-issues/go.sum @@ -1,5 +1,7 @@ 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= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-issues/module.json b/modules/mesh-issues/module.json index 91e01be3..a4e96e38 100644 --- a/modules/mesh-issues/module.json +++ b/modules/mesh-issues/module.json @@ -66,10 +66,13 @@ "records.records_read", "mesh-knowledge.knowledge_highest" ], + "logs": [ + "changes" + ], "state": [ { "name": "issues", - "history": 8 + "history": 1 }, "counters", "grooming",