diff --git a/modules/mesh-issues/README.md b/modules/mesh-issues/README.md index 5c52946c..7b36ef7a 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 | @@ -167,7 +167,7 @@ It is a verb on the seat, callable by any agent, because it can do only this: `imported` event. What it writes comes from hq's `main` as the records module holds it, which is reviewed in pull requests. - **What it keeps is bounded** as every verb's text is: the title 200 bytes, the report 64 KiB, one issue - 256 KiB; a report that trips a credential's shape is refused. An entry of `located-in` or `fixed-by` hq + 252 KiB; a report that trips a credential's shape is refused. An entry of `located-in` or `fixed-by` hq wrote is prose at times, so it is taken within a reason's bound (2000 bytes), not a line's. - **An owner hq named that the registered list does not know is kept as written and marked**, in `located-in-unverified` on the issue and `unverified-owners` in the answer, never refused: the list is @@ -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,63 @@ 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**, 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. It still keeps 8 values per key, so the bus +cannot drop the older values before the carry has read them; it goes to one value per key (ADR 0300 §3) in a +follow-up, once the carry has run with a read of those values. + +**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, read through the +runtime's `mesh/state.history` (hq ADR 0307), is appended oldest first as a `carried` operation holding the +whole issue and the bucket's revision of that value, and the fold written. It resumes rather than restarts: a +kept value at or below the highest revision a carried operation holds, or one the tracker wrote itself (it has +`folded`), is known to the log, so an interrupted carry, or two instances carrying one issue at once, carry each +value once. Matched by revision, not content, because the same content may be kept twice. A value written beside +the log after the carry, by an instance from before the switch, is carried as one more operation and is never +written over by a fold. A deletion among the kept values is passed over and counted, since the tracker never +deletes an issue and every value after one holds the issue whole. + +An issue that cannot be carried — the log refusing it, or a value this module cannot read — is a fault, named +with its reason in the start line; the others are carried and the start finishes. That issue stays in the +bucket as it was, read and listed as before, and every write to it is refused with the same reason until it +can be carried, so no operation reaches the log before its carry. One issue is kept in at most 252 KiB, so that +the operation holding it whole fits one entry of the log (256 KiB). An entry of the log that cannot be folded +faults its issue on start: nothing folded past it is written, and its bucket value is left as it is. + +**In memory** the tracker holds every folded issue, kept current by the watch and its own writes, and a +full-text index of title, report and notes, ranked by BM25 (title words weigh three), rebuilt on every start. +`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 252 KiB; a note +that would pass it is refused. + +**The events announce; the log is the audit trail.** Every change is still an event: `mesh-issues.opened`, +`.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 +367,14 @@ 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, passing over a deletion among them. Over the bus, a +write to an issue the bucket kept before the switch carries its three kept values first, read through the +runtime's history. 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..62da6db5 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) @@ -239,6 +334,26 @@ func (i *instance) answer(m rpc) { case "mesh/state.put": rev, err := kv.Put(ctx, p.Key, p.Value) return map[string]any{"revision": rev}, err + case "mesh/state.history": + // As the runtime serves it (novox/hq ADR 0307): every value kept, oldest first, puts and deletes. + kept, err := kv.History(ctx, p.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return []any{}, nil + } + if err != nil { + return nil, err + } + out := []map[string]any{} + for _, e := range kept { + r := map[string]any{"key": e.Key(), "revision": e.Revision(), "op": "put", "at": e.Created()} + if e.Operation() == jetstream.KeyValuePut { + r["value"] = json.RawMessage(e.Value()) + } else { + r["op"] = "delete" + } + out = append(out, r) + } + return out, nil case "mesh/state.keys": keys, err := kv.Keys(ctx) if errors.Is(err, jetstream.ErrNoKeysFound) { @@ -471,6 +586,42 @@ func TestFortyReservationsAtOnceFromTwoInstancesGetFortyDistinctConsecutiveNumbe } } +// The carry over the bus (novox/hq ADR 0300 §6, ADR 0307): values the bucket kept before the switch are read +// through the runtime's history, oldest first, and carried before the write that touches the issue. +func TestAWriteOverTheBusCarriesEveryValueTheBucketKeptFirst(t *testing.T) { + b := newBus(t) + var events []event + var evMu sync.Mutex + kv := b.kvs[IssuesState] + for _, title := range []string{"first", "second", "third"} { + raw, _ := json.Marshal(Issue{Number: 300, Title: title, Status: StatusOpen, Report: "r", OpenedBy: "x", Opening: "o300"}) + if _, err := kv.Put(context.Background(), "300", raw); err != nil { + t.Fatal(err) + } + } + i := b.start(t, writeSettings(t, 365), &events, &evMu) + if _, err := i.call("note", map[string]any{"number": 300, "text": "after the switch"}); err != nil { + t.Fatal(err) + } + var titles []string + var ops []string + for seq := uint64(1); ; seq++ { + m, err := b.logs[ChangesLog].GetMsg(context.Background(), seq) + if err != nil { + break + } + var op Operation + _ = json.Unmarshal(m.Data, &op) + ops = append(ops, op.Op) + if op.Issue != nil { + titles = append(titles, op.Issue.Title) + } + } + if fmt.Sprint(ops) != "[carried carried carried noted]" || fmt.Sprint(titles) != "[first second third]" { + t.Fatalf("the log holds %v of %v", ops, titles) + } +} + // One instance, every verb over the bus: what an agent does with an issue from opening to resolving. func TestAnIssueIsOpenedNotedLinkedMovedAndListedOverTheBus(t *testing.T) { b := newBus(t) @@ -531,4 +682,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..0fc5af2f --- /dev/null +++ b/modules/mesh-issues/cmd/mesh-issues/changes.go @@ -0,0 +1,831 @@ +package main + +import ( + "bytes" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "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, and a read in pages. +type Log interface { + AppendAt(key string, value any, expect uint64) (uint64, error) + Read(from uint64, keys string, limit int) (*stdio.LogPage, error) +} + +// 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) +} + +// 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"` + // Revision is the bucket's revision of the value a carried operation holds: how a carry knows which kept + // values the log holds already. + Revision uint64 `json:"revision,omitempty"` +} + +// Act is who changes an issue and as what kind of operation. +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) { + first, lerr := t.firstOf(is.Number) + if lerr != nil { + return 0, fmt.Errorf("the log refused issue %d's first operation and could not be read back (%v): %w", + is.Number, lerr, errBegun) + } + if mine(first, is.Opening) { + return first.Sequence, nil + } + return 0, errBegun + } + } + // An append that timed out may still have landed: read back before saying it did not. + first, lerr := t.firstOf(is.Number) + switch { + case lerr != nil: + return 0, fmt.Errorf("%v, and the log could not be read back (%v): %w", err, lerr, errUnsure) + case mine(first, is.Opening): + return first.Sequence, nil + case first != nil: + return 0, errBegun + } + return 0, err +} + +// firstOf is an issue's first entry in the log, the one that begins it, or nil when it has none. The first, not +// the last: an open whose answer was lost may have been followed by another operation on it already, such as the +// triage's. +func (t *Tracker) firstOf(n uint64) (*stdio.LogEntry, error) { + p, err := t.Log.Read(1, keyOf(n), 1) + if err != nil { + return nil, err + } + if len(p.Entries) == 0 { + return nil, nil + } + return &p.Entries[0], nil +} + +// errUnsure is a first operation the log may have taken: its answer was lost, and the log could not be read back. +var errUnsure = errors.New("it may have been taken") + +// 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, fmt.Errorf("issue %d is kept from before the tracker's log and is not carried into it yet, "+ + "so nothing is changed: %v", n, err) + } + continue + } + 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 (252 KiB). Nothing "+ + "was changed: say it shorter, or in a document the issue links", n, len(raw), MostIssueBytes) + } + op, changed, err := operation(act, t.now(), before, is) + 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) +} + +// Carried is what the carry did (ADR 0300 §6), as it happened: the values it appended, not the ones it meant to. +type Carried struct { + // Issues is how many issues it appended values of, Values how many values; Already how many issues the log + // already began whole, which it left as they are. + Issues int `json:"issues"` + Values int `json:"values"` + Already int `json:"already"` + // Deletes is how many deletions among the kept values were passed over: an issue is never deleted by the + // tracker, and each value after one holds the issue whole, so the fold needs none. + Deletes int `json:"deletes,omitempty"` + // Faults is each issue that could not be carried, and why. It stays in the bucket as it was, readable and + // listed; every write to it is refused with the same reason until it can be carried, so nothing reaches the + // log before its carry. + Faults []string `json:"faults,omitempty"` + Said string `json:"said"` +} + +// carryOne carries one issue the bucket has not folded from the log: each value the bucket still keeps of it +// (novox/hq ADR 0307), oldest first, as `carried` operations holding the whole issue and the bucket's revision +// of that value, then the fold written to the bucket. Answers how many values it appended and how many +// deletions it passed over. +// +// It resumes rather than restarts. A value is known to the log when its revision is at or below the highest +// revision a carried operation of the issue holds, or when it is a value the tracker wrote itself (it carries +// `folded`): matched by revision, not by content, because the same content may be kept twice (a write that +// changed nothing) and a revision names one value exactly. So an interrupted carry, or another instance carrying +// the same issue at the same moment, carries each value once; a value written beside the log after the carry, +// by an instance from before the switch, is carried as one more operation and so is never overwritten. +func (t *Tracker) carryOne(n uint64) (int, int, error) { + appended := 0 + for i := 0; i < issueAttempts; i++ { + is, rev, err := read(t.Store, n) + if err != nil { + return appended, 0, err + } + if is.Folded != 0 { + return appended, 0, nil + } + entries, err := t.readKey(n, 1) + if err != nil { + return appended, 0, err + } + var folded *Issue + var expect, known uint64 + for _, e := range entries { + if folded, err = foldEntry(folded, e); err != nil { + return appended, 0, fmt.Errorf("issue %d cannot be folded from its log: %v", n, err) + } + expect = e.Sequence + var op Operation + if json.Unmarshal(e.Value, &op) == nil && op.Op == OpCarried { + known = max(known, op.Revision) + } + } + kept, err := t.Store.History(IssuesState, keyOf(n)) + if err != nil { + return appended, 0, fmt.Errorf("the values the bucket keeps of issue %d could not be read: %v", n, err) + } + type value struct { + issue *Issue + revision uint64 + } + var values []value + deletes := 0 + for _, e := range kept { + if e.Op == "delete" { + deletes++ + continue + } + var v Issue + if err := json.Unmarshal(e.Value, &v); err != nil { + return appended, deletes, fmt.Errorf("the value of issue %d at revision %d is in a form this module "+ + "cannot read (%v): not carried, so that nothing kept is left out", n, e.Revision, err) + } + switch { + case v.Folded != 0: + // The tracker's own fold: the log holds it already, and whatever came before it. + values = nil + known = max(known, e.Revision) + case e.Revision > known: + values = append(values, value{&v, e.Revision}) + } + } + if len(kept) == 0 || kept[len(kept)-1].Revision < rev { + // The value read is newer than the history read before it: carried too. + if rev > known { + values = append(values, value{is, rev}) + } + } + raced := false + for _, v := range values { + v.issue.Folded = 0 + seq, err := t.appendOp(n, &Operation{Op: OpCarried, By: Seat, At: t.now(), Issue: v.issue, Revision: v.revision}, expect) + if errors.Is(err, stdio.ErrChanged) { + raced = true + break + } + if err != nil { + return appended, deletes, fmt.Errorf("issue %d could not be carried into the log (%v); %d of its %d "+ + "values not carried yet were carried. Nothing is lost: the bucket still holds it, and the carry "+ + "resumes after the last value carried, at the next start or before its next change", n, err, appended, len(values)) + } + appended++ + expect = seq + if folded, err = fold(nil, &Operation{Op: OpCarried, Issue: v.issue}, seq); err != nil { + return appended, deletes, err + } + } + if raced { + // Another carrier was first: read again; what it appended is known by its revisions. + continue + } + if folded == nil { + return appended, deletes, fmt.Errorf("issue %d keeps no value to carry", n) + } + if _, err := t.Store.Update(IssuesState, keyOf(n), folded, rev); err != nil { + if errors.Is(err, stdio.ErrChanged) { + // Written meanwhile, by a carrier or an instance from before the switch: read again. + continue + } + return appended, deletes, fmt.Errorf("issue %d was carried into the log and its bucket could not be written (%v): "+ + "it is folded at the next change or start", n, err) + } + t.remember(folded) + return appended, deletes, nil + } + return appended, 0, fmt.Errorf("issue %d was written by another writer %d times while it was carried", n, issueAttempts) +} + +// same says whether two issues hold the same, whatever sequence each was folded at. +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 has not folded from the log (ADR 0300 §6). An issue that +// cannot be carried is a fault, named with its reason, and the others are carried. Run again, it carries +// nothing more, and resumes an issue whose carry was interrupted after the last value carried. +func (t *Tracker) Carry() (*Carried, error) { + keys, err := t.Store.Keys(IssuesState) + if err != nil { + return nil, fmt.Errorf("the issues kept cannot be listed: %v", err) + } + out := &Carried{} + 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 { + out.Faults = append(out.Faults, err.Error()) + continue + } + if is.Folded != 0 { + out.Already++ + continue + } + v, d, err := t.carryOne(n) + out.Values += v + out.Deletes += d + if v > 0 { + out.Issues++ + } else if err == nil { + out.Already++ + } + if err != nil { + t.Logf("[mesh-issues] %v", err) + out.Faults = append(out.Faults, err.Error()) + } + } + if out.Values == 0 { + out.Said = fmt.Sprintf("Carried nothing: the log begins every issue of the bucket it could (%d).", out.Already) + } else { + out.Said = fmt.Sprintf("Carried %d values of %d issues into the log, each value the bucket kept, oldest first; "+ + "%d issues were there already.", out.Values, out.Issues, out.Already) + } + if out.Deletes > 0 { + out.Said += fmt.Sprintf(" %d deletions among the kept values were passed over: each value after one holds "+ + "the issue whole.", out.Deletes) + } + if len(out.Faults) > 0 { + out.Said += fmt.Sprintf(" %d issues could not be carried, and stay as the bucket holds them, every write to "+ + "them refused until they can be: %s.", len(out.Faults), strings.Join(out.Faults, "; ")) + } + return out, nil +} + +// 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() + if err != nil { + return out, fmt.Errorf("the carry could not start (%v); writes carry each issue before they change it", err) + } + out.Carried = carried + for _, f := range carried.Faults { + out.Faults = append(out.Faults, "not carried: "+f) + } + folds := map[uint64]*Issue{} + // broken is each issue an entry of which could not be folded: none of its later entries is folded on top, + // and its bucket value is left as it is, so no fold that skipped an operation is ever written. + broken := map[uint64]bool{} + var highest uint64 + for from := uint64(1); from != 0; { + p, err := t.Log.Read(from, "", logPage) + 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) + if broken[n] { + continue + } + f, err := foldEntry(folds[n], e) + if err != nil { + out.Faults = append(out.Faults, fmt.Sprintf("issue %d cannot be folded from its log, and its bucket "+ + "value is left as it is: %v", n, err)) + broken[n] = true + delete(folds, n) + continue + } + folds[n] = f + } + from = p.Next + } + out.Issues = len(folds) + len(broken) + 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 == 0: + // Written beside the log, by an instance from before the switch, after the carry, and not carried again + // (its carry is in the faults): the fold would overwrite that write. Left for its carry, which appends it + // as one more operation before the issue's next change. + out.Faults = append(out.Faults, fmt.Sprintf("issue %d holds a value written beside the log and not carried "+ + "yet; it is not refolded over", n)) + continue + case is.Folded < f.Folded: + if _, err := t.Store.Update(IssuesState, keyOf(n), f, rev); err != nil && !errors.Is(err, stdio.ErrChanged) { + out.Faults = append(out.Faults, fmt.Sprintf("issue %d could not be refolded in the bucket: %v", n, err)) + 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..51e785d5 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. @@ -148,9 +151,10 @@ type Link struct { At time.Time `json:"at"` } -// MostIssueBytes is the most one issue may be as kept: a value the bus holds (to-be 53 §4). The largest -// issue folder in hq was 25 KiB on 2026-10-10. -const MostIssueBytes = 256 << 10 +// MostIssueBytes is the most one issue may be as kept: 252 KiB, so the operation holding it whole (an open, an +// import, a carry) fits in one entry of the log, which takes at most 256 KiB (novox/hq ADR 0297, ADR 0300). The +// largest issue folder in hq was 25 KiB on 2026-10-10. +const MostIssueBytes = 252 << 10 // Entry is one key's value and the revision it was read at. type Entry struct { @@ -166,6 +170,9 @@ type Store interface { Update(state, key string, value any, revision uint64) (uint64, error) Keys(state string) ([]string, error) Increment(state, key string, attempts int) (uint64, error) + // History is every value a key still keeps, oldest first, puts and deletes (novox/hq ADR 0307): what the + // carry into the log reads (ADR 0300 §6). + History(state, key string) ([]stdio.StateRevision, error) } // runtimeStore is Store through the runtime that launched this bundle. @@ -185,6 +192,11 @@ func (runtimeStore) Update(state, key string, value any, revision uint64) (uint6 return stdio.State(state).Update(key, value, revision) } func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } + +// History is every value the bucket still keeps of a key, oldest first (novox/hq ADR 0307). +func (runtimeStore) History(state, key string) ([]stdio.StateRevision, error) { + return stdio.State(state).History(key) +} func (runtimeStore) Increment(state, key string, attempts int) (uint64, error) { return stdio.State(state).Increment(key, attempts) } 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..c4876358 --- /dev/null +++ b/modules/mesh-issues/cmd/mesh-issues/log_test.go @@ -0,0 +1,819 @@ +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 len(raw) > 256<<10 { + return 0, fmt.Errorf("the value for %s is %d bytes; a log entry holds at most 256 KiB", key, len(raw)) + } + if last := l.lastOf(key); last != expect { + return 0, fmt.Errorf("the last entry of %s is %d, not %d: %w", key, last, expect, stdio.ErrChanged) + } + 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 +} + +// 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 +} + +// foldCheck holds every issue the bucket folded to what its whole log adds up to, byte for byte: run after every +// test of a harness on this log (the review of PR 228). +func (l *fakeLog) foldCheck(t *testing.T, m *memory) { + l.mu.Lock() + entries := append([]stdio.LogEntry(nil), l.entries...) + l.mu.Unlock() + for k, v := range m.snapshot(IssuesState) { + var is Issue + if json.Unmarshal([]byte(v), &is) != nil || is.Folded == 0 { + continue + } + var f *Issue + var err error + for _, e := range entries { + if e.Key != k || e.Sequence > is.Folded { + continue + } + if f, err = foldEntry(f, e); err != nil { + t.Errorf("issue %s does not fold from its log: %v", k, err) + break + } + } + if raw, _ := json.Marshal(f); err == nil && string(raw) != v { + t.Errorf("issue %s in the bucket is not the fold of its log:\nbucket %s\nfold %s", k, v, raw) + } + } +} + +func (l *fakeLog) size() int { + l.mu.Lock() + defer l.mu.Unlock() + 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 || len(c.Faults) != 0 { + 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 deletion among the values the bucket kept is passed over, said, and the values after it carried. +func TestTheCarryPassesOverADeletionAmongTheKeptValues(t *testing.T) { + h := newHarness(t) + keptBefore(t, h, 300, "before the deletion") + h.mem.deleteKey(IssuesState, keyOf(300)) + keptBefore(t, h, 300, "after the deletion", "now") + c, err := h.Carry() + if err != nil { + t.Fatal(err) + } + if c.Issues != 1 || c.Values != 3 || c.Deletes != 1 || !strings.Contains(c.Said, "1 deletions") { + t.Fatalf("carried %+v", c) + } + ops := h.log.ops(t, 300) + if len(ops) != 3 || ops[0].Issue.Title != "before the deletion" || ops[2].Issue.Title != "now" { + t.Fatalf("the log of 300: %+v", ops) + } + if is, _ := h.Get(300); is.Title != "now" || is.Folded != 3 { + t.Fatalf("300 is %+v", is) + } +} + +// §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) + } +} + +// titlesOf is each operation of an issue's log as its kind and, for one holding the issue whole, its title. +func titlesOf(t *testing.T, h *harness, n uint64) string { + t.Helper() + var out []string + for _, op := range h.log.ops(t, n) { + if op.Issue != nil { + out = append(out, op.Op+":"+op.Issue.Title) + } else { + out = append(out, op.Op) + } + } + return strings.Join(out, " ") +} + +// §6, the review of PR 228: a carry that fails partway, then a second carry, leaves every kept value carried +// once, oldest first, and each carry says what it really carried. +func TestAnInterruptedCarryResumesAfterTheLastValueCarried(t *testing.T) { + h := newHarness(t) + keptBefore(t, h, 300, "v1", "v2", "v3", "v4") + // The bus goes away after the first value is appended. + h.log.between = func() { h.log.between = func() { h.log.fail = errors.New("the bus went away") } } + c, err := h.Carry() + if err != nil { + t.Fatal(err) + } + if c.Values != 1 || len(c.Faults) != 1 || !strings.Contains(c.Said, "could not be carried") || + !strings.Contains(c.Faults[0], "the bus went away") { + t.Fatalf("the first carry said %+v", c) + } + if is, _ := h.Get(300); is.Folded != 0 || is.Title != "v4" { + t.Fatalf("after the failure the bucket holds %+v", is) + } + // A write meanwhile is refused, and changes nothing. + if _, err := h.Note(300, "while the bus is away"); err == nil || !strings.Contains(err.Error(), "not carried") { + t.Fatalf("a note on an issue half carried: %v", err) + } + h.log.fail = nil + c, err = h.Carry() + if err != nil { + t.Fatal(err) + } + if c.Values != 3 || c.Issues != 1 || len(c.Faults) != 0 { + t.Fatalf("the second carry said %+v", c) + } + if got := titlesOf(t, h, 300); got != "carried:v1 carried:v2 carried:v3 carried:v4" { + t.Fatalf("the log of 300: %s", got) + } + if is, _ := h.Get(300); is.Title != "v4" || is.Folded != 4 { + t.Fatalf("300 is %+v", is) + } +} + +// §6, the review of PR 228: one issue that cannot be carried is a fault, named, and the start carries the others, +// folds, raises the counter and loads memory. Writes to that issue are refused with the reason. +func TestOneIssueThatCannotBeCarriedDoesNotStopTheStart(t *testing.T) { + h := newHarness(t) + keptBefore(t, h, 300, "small") + // Kept before the switch at the bucket's limit: whole, in a carried operation, past what one log entry holds. + h.mem.put(IssuesState, "301", Issue{Number: 301, Title: "huge", Status: StatusOpen, Report: strings.Repeat("x", 256<<10-200), + OpenedBy: "x", Opening: "o301"}) + keptBefore(t, h, 302, "also small") + s, err := h.Start() + if err != nil { + t.Fatal(err) + } + if s.Carried.Issues != 2 || len(s.Faults) != 1 || !strings.Contains(s.Faults[0], "issue 301") || + !strings.Contains(s.Faults[0], "256 KiB") || !strings.Contains(s.Said, "issue 301") { + t.Fatalf("started %+v", s) + } + if !h.view().ready() { + t.Fatal("memory was not loaded") + } + l, err := h.List(ListAsk{}) + if err != nil || l.Matched != 3 { + t.Fatalf("listed %+v, %v", l, err) + } + if _, err := h.Note(301, "a note"); err == nil || !strings.Contains(err.Error(), "256 KiB") { + t.Fatalf("a note on the issue not carried: %v", err) + } + if is, _ := h.Get(301); is.Title != "huge" || is.Folded != 0 || len(is.Notes) != 0 { + t.Fatal("the issue not carried was changed") + } + if is, _ := h.Get(302); is.Folded == 0 { + t.Fatal("302 was not carried") + } +} + +// §6, the review of PR 228: two instances carrying the same issue at the same moment carry each value once. +func TestTwoInstancesCarryingOneIssueAtOnceCarryEachValueOnce(t *testing.T) { + h := newHarness(t) + keptBefore(t, h, 300, "v1", "v2", "v3") + other := h.another(t) + var otherSaid *Carried + h.log.between = func() { + c, err := other.Carry() + if err != nil { + t.Errorf("the other carry: %v", err) + } + otherSaid = c + } + c, err := h.Carry() + if err != nil { + t.Fatal(err) + } + if got := titlesOf(t, h, 300); got != "carried:v1 carried:v2 carried:v3" { + t.Fatalf("the log of 300: %s", got) + } + if otherSaid.Values+c.Values != 3 || len(c.Faults)+len(otherSaid.Faults) != 0 { + t.Fatalf("this carry %+v, the other %+v", c, otherSaid) + } +} + +// §6, the review of PR 228: a value written beside the log after the carry, by an instance from before the switch, +// is never overwritten by the fold on start; it is carried as one more operation. +func TestAValueWrittenBesideTheLogIsCarriedNotOverwritten(t *testing.T) { + h := newHarness(t) + n := h.open(t, "carried", "hq") + if _, err := h.Note(n, "after the switch"); err != nil { + t.Fatal(err) + } + // An instance from before the switch writes the bucket: no folded. + is, _ := h.Get(n) + is.Folded, is.Title = 0, "written beside the log" + h.mem.put(IssuesState, keyOf(n), is) + // Its carry fails at this start: the fold is not written over it. + h.log.failNext = 1 + s, err := h.another(t).Start() + if err != nil { + t.Fatal(err) + } + if got, _ := h.Get(n); got.Title != "written beside the log" || got.Folded != 0 || len(s.Faults) != 2 { + t.Fatalf("issue %d is %+v after a start whose carry failed (faults %v)", n, got, s.Faults) + } + // The next start carries it. + if _, err := h.another(t).Start(); err != nil { + t.Fatal(err) + } + got, _ := h.Get(n) + if got.Title != "written beside the log" || got.Folded != 3 || len(got.Notes) != 1 { + t.Fatalf("issue %d is %+v", n, got) + } + if ops := titlesOf(t, h, n); ops != "opened:carried noted carried:written beside the log" { + t.Fatalf("the log: %s", ops) + } +} + +// §3, the review of PR 228: an entry that cannot be folded faults its issue on start, and nothing folded past it +// is written. +func TestAnEntryThatCannotBeFoldedLeavesItsIssueAsItIs(t *testing.T) { + h := newHarness(t) + n := h.open(t, "fine") + m := h.open(t, "broken") + if _, err := h.Note(m, "kept"); err != nil { + t.Fatal(err) + } + h.log.mu.Lock() + for i, e := range h.log.entries { + if e.Key == keyOf(m) && e.Sequence == 2 { + h.log.entries[i].Value = json.RawMessage(`{"op":"noted","add":{"notes":"not a list"}}`) + } + } + h.log.mu.Unlock() + before := h.mem.snapshot(IssuesState)[keyOf(m)] + h.mem.empty(IssuesState) + h.mem.put(IssuesState, keyOf(m), json.RawMessage(before)) + s, err := h.another(t).Start() + if err != nil { + t.Fatal(err) + } + if s.Rebuilt != 1 || len(s.Faults) != 1 || !strings.Contains(s.Faults[0], fmt.Sprintf("issue %d cannot be folded", m)) { + t.Fatalf("started %+v", s) + } + if h.mem.snapshot(IssuesState)[keyOf(m)] != before { + t.Fatal("the broken issue's bucket value was written over") + } + if is, _ := h.Get(n); is.Title != "fine" { + t.Fatalf("%d is %+v", n, is) + } + // The broken entry is the test's doing: the fold check would find it, so take the issue out of the bucket. + h.mem.empty(IssuesState) +} + +// §2, the review of PR 228: an open whose answer was lost, its issue already changed by another operation, is +// still known as its own by its first entry. +func TestALostOpenIsKnownByItsFirstEntryNotItsLast(t *testing.T) { + h := newHarness(t) + h.log.lost = errors.New("no answer in 30 s") + // After the first operation lands and its answer is lost, another operation follows before the retry. + h.log.between = func() { + h.log.between = func() { + h.log.mu.Lock() + h.log.entries = append(h.log.entries, stdio.LogEntry{Sequence: uint64(len(h.log.entries) + 1), Key: "365", + Value: json.RawMessage(`{"op":"triaged","by":"triage","set":{"severity":"\"minor\""}}`)}) + h.log.mu.Unlock() + } + } + o, err := h.Open(OpenAsk{Title: "lost and followed", Report: "r"}) + if err != nil || o.Number != 365 { + t.Fatalf("%+v %v", o, err) + } + // The entry added by hand is not one the tracker folded: take the issue out before the fold check. + h.mem.empty(IssuesState) +} diff --git a/modules/mesh-issues/cmd/mesh-issues/main.go b/modules/mesh-issues/cmd/mesh-issues/main.go index 302a46aa..2f674e6b 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. @@ -181,6 +207,9 @@ func (emptyStore) Update(string, string, any, uint64) (uint64, error) { return 0, errors.New("a dry run at the terminal writes nothing") } func (emptyStore) Keys(string) ([]string, error) { return nil, nil } +func (emptyStore) History(string, string) ([]stdio.StateRevision, error) { + return []stdio.StateRevision{}, nil +} func (emptyStore) Increment(string, string, int) (uint64, error) { return 0, errors.New("a dry run at the terminal writes nothing") } @@ -376,7 +405,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 +416,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 +443,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..cfbd13ac 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,8 +88,13 @@ 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: issues the fold of the log (§3), + // still keeping its older values until the carry has read them (they go to one value in a follow-up), 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). + 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 || d[1].Name != CountersState || d[1].History > 1 || d[1].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..257a573b 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,50 @@ 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, as the SDK's State(...).History answers it: puts and +// deletes, empty for a key that keeps nothing. +func (m *memory) History(state, key string) ([]stdio.StateRevision, error) { + m.mu.Lock() + defer m.mu.Unlock() + out := []stdio.StateRevision{} + for _, e := range m.hist[state][key] { + op, value := "put", e.Value + if e.Value == nil { + op = "delete" + } + out = append(out, stdio.StateRevision{Key: key, Value: value, Revision: e.Revision, Op: op}) + } + return out, nil +} + +// deleteKey deletes a key, keeping the deletion among its values as the bucket does. +func (m *memory) deleteKey(state, key string) { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.data[state], key) + m.rev++ + m.hist[state][key] = append(m.hist[state][key], Entry{Revision: m.rev}) +} + +// 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 +199,7 @@ type said struct { type harness struct { *Tracker mem *memory + log *fakeLog mu sync.Mutex events []said settings Settings @@ -162,12 +208,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() @@ -185,6 +232,11 @@ func newHarness(t *testing.T) *harness { Logf: t.Logf, Sleep: func(time.Duration) {}, } + t.Cleanup(func() { + if h.Tracker.Log == Log(h.log) { + h.log.foldCheck(t, h.mem) + } + }) return h } diff --git a/modules/mesh-issues/cmd/mesh-issues/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..49d11876 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) Read(uint64, string, int) (*stdio.LogPage, 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) } } @@ -434,7 +447,7 @@ func TestANoteThatWouldMakeTheIssueTooLargeIsRefusedAndNothingChanges(t *testing for i := 0; i < 40 && err == nil; i++ { _, err = h.Note(n, note) } - if err == nil || !strings.Contains(err.Error(), "256 KiB") { + if err == nil || !strings.Contains(err.Error(), "252 KiB") { t.Fatalf("no refusal: %v", err) } e, _ := h.mem.Get(IssuesState, keyOf(n)) diff --git a/modules/mesh-issues/cmd/mesh-issues/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..62e67f33 --- /dev/null +++ b/modules/mesh-issues/cmd/mesh-issues/view.go @@ -0,0 +1,291 @@ +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 shallow copy: a caller may set an issue's fields without changing what is held, +// but its lists (notes, links, fixes and the rest) are shared with what is held, so a caller reads them and never +// changes them in place. Every caller in this module only reads; a write goes through change, which reads the +// bucket afresh. A deep copy per list would cost a JSON round trip of every issue on every list. +func (h *memoryView) all() map[uint64]*Issue { + h.mu.RLock() + defer h.mu.RUnlock() + 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..2aea2394 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.20261012082113-9747f3e1b1ac 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..b16d5979 100644 --- a/modules/mesh-issues/go.sum +++ b/modules/mesh-issues/go.sum @@ -1,5 +1,5 @@ -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.20261012082113-9747f3e1b1ac h1:jnZ/n+fFMo7VQPNNXK1gjStHIJPw3SPbzPxsf/PpZqw= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261012082113-9747f3e1b1ac/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= 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..8781fa9a 100644 --- a/modules/mesh-issues/module.json +++ b/modules/mesh-issues/module.json @@ -66,6 +66,9 @@ "records.records_read", "mesh-knowledge.knowledge_highest" ], + "logs": [ + "changes" + ], "state": [ { "name": "issues",