From dcc7510cd839dd74eeb0f5a31ff1f3570d9769ab Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 12 Oct 2026 10:20:51 +0200 Subject: [PATCH] =?UTF-8?q?mesh-issues:=20carry=20each=20value=20the=20iss?= =?UTF-8?q?ues=20bucket=20kept,=20read=20through=20the=20runtime's=20histo?= =?UTF-8?q?ry,=20so=20the=20log=20begins=20with=20all=20the=20bucket=20sti?= =?UTF-8?q?ll=20knew=20(hq=20ADR=200300=20=C2=A76,=20ADR=200307)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- modules/mesh-issues/README.md | 12 ++-- .../mesh-issues/cmd/mesh-issues/bus_test.go | 56 +++++++++++++++++++ .../mesh-issues/cmd/mesh-issues/changes.go | 28 +++++++--- modules/mesh-issues/cmd/mesh-issues/issue.go | 5 ++ .../mesh-issues/cmd/mesh-issues/log_test.go | 22 ++++++++ .../cmd/mesh-issues/memory_test.go | 25 +++++++-- .../mesh-issues/cmd/mesh-issues/tracker.go | 3 + modules/mesh-issues/go.mod | 2 +- modules/mesh-issues/go.sum | 2 + 9 files changed, 138 insertions(+), 17 deletions(-) diff --git a/modules/mesh-issues/README.md b/modules/mesh-issues/README.md index 55504f01..191feb88 100644 --- a/modules/mesh-issues/README.md +++ b/modules/mesh-issues/README.md @@ -272,9 +272,11 @@ below. Until then a `list` reads the bucket and a write carries the issue it cha 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 +instance from before the switch — that value is carried as one more operation). The values kept are read through +the runtime's `mesh/state.history` (hq ADR 0307, the SDK's `State("issues").History`), oldest first; a deletion +among them is passed over and counted in the start line, since the tracker never deletes an issue and every +value after one holds the issue whole. A store that reads no history gives each issue's current value alone, +and the start line says so. It runs on start rather than as a verb because it must precede every write, and a verb could be called late or never. **In memory** the tracker holds every folded issue, kept current by the watch and its own writes, and a @@ -365,5 +367,7 @@ open refused at validation taking no number, an append failing after the number 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 +the carry carrying each value kept, oldest first, once, passing over a deletion among them. Over the bus, a +write to an issue the bucket kept before the switch carries its three kept values first, read through the +runtime's history. Over the bus, the stream the controller makes holds one operation per change of `TestAnIssueIsOpenedNotedLinkedMovedAndListedOverTheBus`. diff --git a/modules/mesh-issues/cmd/mesh-issues/bus_test.go b/modules/mesh-issues/cmd/mesh-issues/bus_test.go index b85247e5..62da6db5 100644 --- a/modules/mesh-issues/cmd/mesh-issues/bus_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/bus_test.go @@ -334,6 +334,26 @@ func (i *instance) answer(m rpc) { case "mesh/state.put": rev, err := kv.Put(ctx, p.Key, p.Value) return map[string]any{"revision": rev}, err + case "mesh/state.history": + // As the runtime serves it (novox/hq ADR 0307): every value kept, oldest first, puts and deletes. + kept, err := kv.History(ctx, p.Key) + if errors.Is(err, jetstream.ErrKeyNotFound) { + return []any{}, nil + } + if err != nil { + return nil, err + } + out := []map[string]any{} + for _, e := range kept { + r := map[string]any{"key": e.Key(), "revision": e.Revision(), "op": "put", "at": e.Created()} + if e.Operation() == jetstream.KeyValuePut { + r["value"] = json.RawMessage(e.Value()) + } else { + r["op"] = "delete" + } + out = append(out, r) + } + return out, nil case "mesh/state.keys": keys, err := kv.Keys(ctx) if errors.Is(err, jetstream.ErrNoKeysFound) { @@ -566,6 +586,42 @@ func TestFortyReservationsAtOnceFromTwoInstancesGetFortyDistinctConsecutiveNumbe } } +// The carry over the bus (novox/hq ADR 0300 §6, ADR 0307): values the bucket kept before the switch are read +// through the runtime's history, oldest first, and carried before the write that touches the issue. +func TestAWriteOverTheBusCarriesEveryValueTheBucketKeptFirst(t *testing.T) { + b := newBus(t) + var events []event + var evMu sync.Mutex + kv := b.kvs[IssuesState] + for _, title := range []string{"first", "second", "third"} { + raw, _ := json.Marshal(Issue{Number: 300, Title: title, Status: StatusOpen, Report: "r", OpenedBy: "x", Opening: "o300"}) + if _, err := kv.Put(context.Background(), "300", raw); err != nil { + t.Fatal(err) + } + } + i := b.start(t, writeSettings(t, 365), &events, &evMu) + if _, err := i.call("note", map[string]any{"number": 300, "text": "after the switch"}); err != nil { + t.Fatal(err) + } + var titles []string + var ops []string + for seq := uint64(1); ; seq++ { + m, err := b.logs[ChangesLog].GetMsg(context.Background(), seq) + if err != nil { + break + } + var op Operation + _ = json.Unmarshal(m.Data, &op) + ops = append(ops, op.Op) + if op.Issue != nil { + titles = append(titles, op.Issue.Title) + } + } + if fmt.Sprint(ops) != "[carried carried carried noted]" || fmt.Sprint(titles) != "[first second third]" { + t.Fatalf("the log holds %v of %v", ops, titles) + } +} + // One instance, every verb over the bus: what an agent does with an issue from opening to resolving. func TestAnIssueIsOpenedNotedLinkedMovedAndListedOverTheBus(t *testing.T) { b := newBus(t) diff --git a/modules/mesh-issues/cmd/mesh-issues/changes.go b/modules/mesh-issues/cmd/mesh-issues/changes.go index 64a1a5ab..20b8a482 100644 --- a/modules/mesh-issues/cmd/mesh-issues/changes.go +++ b/modules/mesh-issues/cmd/mesh-issues/changes.go @@ -464,11 +464,11 @@ func clone(is *Issue) (*Issue, error) { return &c, json.Unmarshal(raw, &c) } -// History is a Store that reads every value a key still keeps, oldest first. 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. +// History is a Store that reads every value a key still keeps, oldest first, as the runtime serves it +// (novox/hq ADR 0307, mesh-sdk State(...).History): each a put or a delete. The carry reads it where the store +// offers it, and takes each issue's current value alone where it does not, and says so. type History interface { - History(state, key string) ([]Entry, error) + History(state, key string) ([]stdio.StateRevision, error) } // Carried is what the carry did (ADR 0300 §6). @@ -479,7 +479,10 @@ type Carried struct { 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"` + History bool `json:"history"` + // Deletes is how many deletions among the kept values were passed over: an issue is never deleted by the + // tracker, and each value after one holds the issue whole, so the fold needs none. + Deletes int `json:"deletes,omitempty"` Said string `json:"said"` } @@ -560,7 +563,8 @@ func (t *Tracker) carryOne(n uint64) (int, error) { } // 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. +// alone where the store reads no history. A deletion among them is passed over and counted: the tracker never +// deletes an issue, the fold has no deletion, and the next value holds the issue whole again. func (t *Tracker) kept(n uint64, current *Issue) ([]*Issue, error) { h, ok := t.Store.(History) if !ok { @@ -572,9 +576,13 @@ func (t *Tracker) kept(n uint64, current *Issue) ([]*Issue, error) { } var out []*Issue for _, e := range old { + if e.Op == "delete" { + t.carryDeletes.Add(1) + continue + } var v Issue if json.Unmarshal(e.Value, &v) != nil { - t.Logf("[mesh-issues] an older value of issue %d is in a form this module cannot read: not carried", n) + t.Logf("[mesh-issues] the value of issue %d at revision %d is in a form this module cannot read: not carried", n, e.Revision) continue } out = append(out, &v) @@ -609,6 +617,7 @@ func (t *Tracker) Carry() (*Carried, error) { } _, history := t.Store.(History) out := &Carried{History: history} + deletes := t.carryDeletes.Load() for _, k := range keys { n, ok := numberOfKey(k) if !ok { @@ -636,12 +645,17 @@ func (t *Tracker) Carry() (*Carried, error) { out.Already++ } } + out.Deletes = int(t.carryDeletes.Load() - deletes) switch { case out.Issues == 0: out.Said = fmt.Sprintf("Nothing to carry: the log begins every issue of the bucket (%d).", out.Already) case history: out.Said = fmt.Sprintf("Carried %d values of %d issues into the log, each value the bucket kept, oldest first; "+ "%d issues were there already.", out.Values, out.Issues, out.Already) + if out.Deletes > 0 { + out.Said += fmt.Sprintf(" %d deletions among the kept values were passed over: each value after one holds "+ + "the issue whole.", out.Deletes) + } default: out.Said = fmt.Sprintf("Carried %d issues into the log, %d values: the current value of each, as the runtime "+ "reads no older value of a key; %d issues were there already.", out.Issues, out.Values, out.Already) diff --git a/modules/mesh-issues/cmd/mesh-issues/issue.go b/modules/mesh-issues/cmd/mesh-issues/issue.go index 8bf96b8a..64ac267b 100644 --- a/modules/mesh-issues/cmd/mesh-issues/issue.go +++ b/modules/mesh-issues/cmd/mesh-issues/issue.go @@ -188,6 +188,11 @@ func (runtimeStore) Update(state, key string, value any, revision uint64) (uint6 return stdio.State(state).Update(key, value, revision) } func (runtimeStore) Keys(state string) ([]string, error) { return stdio.State(state).Keys() } + +// History is every value the bucket still keeps of a key, oldest first (novox/hq ADR 0307). +func (runtimeStore) History(state, key string) ([]stdio.StateRevision, error) { + return stdio.State(state).History(key) +} func (runtimeStore) Increment(state, key string, attempts int) (uint64, error) { return stdio.State(state).Increment(key, attempts) } diff --git a/modules/mesh-issues/cmd/mesh-issues/log_test.go b/modules/mesh-issues/cmd/mesh-issues/log_test.go index 969391fb..a626504e 100644 --- a/modules/mesh-issues/cmd/mesh-issues/log_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/log_test.go @@ -540,6 +540,28 @@ func TestTheCarryIsIdempotentAndCarriesEveryValueKept(t *testing.T) { } } +// §6: a deletion among the values the bucket kept is passed over, said, and the values after it carried. +func TestTheCarryPassesOverADeletionAmongTheKeptValues(t *testing.T) { + h := newHarness(t) + keptBefore(t, h, 300, "before the deletion") + h.mem.deleteKey(IssuesState, keyOf(300)) + keptBefore(t, h, 300, "after the deletion", "now") + c, err := h.Carry() + if err != nil { + t.Fatal(err) + } + if c.Issues != 1 || c.Values != 3 || c.Deletes != 1 || !strings.Contains(c.Said, "1 deletions") { + t.Fatalf("carried %+v", c) + } + ops := h.log.ops(t, 300) + if len(ops) != 3 || ops[0].Issue.Title != "before the deletion" || ops[2].Issue.Title != "now" { + t.Fatalf("the log of 300: %+v", ops) + } + if is, _ := h.Get(300); is.Title != "now" || is.Folded != 3 { + t.Fatalf("300 is %+v", is) + } +} + // §6: a write to an issue not carried yet carries it first, so its first operation in the log is a carried one. func TestAWriteBeforeTheCarryCarriesItsIssueFirst(t *testing.T) { h := newHarness(t) diff --git a/modules/mesh-issues/cmd/mesh-issues/memory_test.go b/modules/mesh-issues/cmd/mesh-issues/memory_test.go index 14962dfd..75d00a5b 100644 --- a/modules/mesh-issues/cmd/mesh-issues/memory_test.go +++ b/modules/mesh-issues/cmd/mesh-issues/memory_test.go @@ -86,14 +86,29 @@ func (m *memory) write(state, key string, value any, check func(Entry, bool) boo return m.rev, nil } -// History is every value the key still keeps, oldest first. -func (m *memory) History(state, key string) ([]Entry, error) { +// History is every value the key still keeps, oldest first, as the SDK's State(...).History answers it: puts and +// deletes, empty for a key that keeps nothing. +func (m *memory) History(state, key string) ([]stdio.StateRevision, error) { m.mu.Lock() defer m.mu.Unlock() - if _, ok := m.data[state][key]; !ok { - return nil, nil + out := []stdio.StateRevision{} + for _, e := range m.hist[state][key] { + op, value := "put", e.Value + if e.Value == nil { + op = "delete" + } + out = append(out, stdio.StateRevision{Key: key, Value: value, Revision: e.Revision, Op: op}) } - return append([]Entry(nil), m.hist[state][key]...), nil + return out, nil +} + +// deleteKey deletes a key, keeping the deletion among its values as the bucket does. +func (m *memory) deleteKey(state, key string) { + m.mu.Lock() + defer m.mu.Unlock() + delete(m.data[state], key) + m.rev++ + m.hist[state][key] = append(m.hist[state][key], Entry{Revision: m.rev}) } // empty takes every key out of a state, as an emptied bucket. diff --git a/modules/mesh-issues/cmd/mesh-issues/tracker.go b/modules/mesh-issues/cmd/mesh-issues/tracker.go index e735ab26..32c9e819 100644 --- a/modules/mesh-issues/cmd/mesh-issues/tracker.go +++ b/modules/mesh-issues/cmd/mesh-issues/tracker.go @@ -10,6 +10,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" stdio "git.novox.be/novox/mesh-sdk/go" @@ -45,6 +46,8 @@ type Tracker struct { groom grooming held *memoryView heldOnce sync.Once + // carryDeletes counts the deletions the carry passed over among the values the bucket kept. + carryDeletes atomic.Int64 } // The events this module says, one per kind of change. diff --git a/modules/mesh-issues/go.mod b/modules/mesh-issues/go.mod index d10e51b0..3566dac1 100644 --- a/modules/mesh-issues/go.mod +++ b/modules/mesh-issues/go.mod @@ -3,7 +3,7 @@ module mesh-issues go 1.25.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011180852-5794ccd569be + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261012081756-d41d255f28ad github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.51.0 ) diff --git a/modules/mesh-issues/go.sum b/modules/mesh-issues/go.sum index c7036ba2..7186dbfd 100644 --- a/modules/mesh-issues/go.sum +++ b/modules/mesh-issues/go.sum @@ -2,6 +2,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011091942-4cf45b5ca0cf h1:ED/+RszK git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011091942-4cf45b5ca0cf/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011180852-5794ccd569be h1:G2ZwFVQsk5Df55gV6OwhJCykrZNQ1/UmO/OxbOJof28= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261011180852-5794ccd569be/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261012081756-d41d255f28ad h1:rg8MlukAHgAlcxT1jJFwHaMRRd0sBU87gaOJ4Ykz9wY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261012081756-d41d255f28ad/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY= 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=