mesh-issues: keep every issue in the log changes and the bucket as its fold, so the record outlives the bucket's cap and a week of events (hq ADR 0300, issues 501, 502)
mesh/merge-gate fail: builds mesh-issues → novox; no bus step; a manifest the change touches fails the module check: modules/mesh-issues/module.json: this …
mesh/repo-check fail: its merge-check.sh failed: files asking for a setting without saying whether it is trusted: 25
mesh/delivery superseded: a newer head of the same pull request

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