mesh-issues: carry each value the issues bucket kept, read through the runtime's history, so the log begins with all the bucket still knew (hq ADR 0300 §6, ADR 0307)
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate pass: builds mesh-issues → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
mesh/delivery superseded: a newer head of the same pull request
mesh/merge-gate pass: builds mesh-issues → novox; no bus step; every machine composes with the change as it did without (4 of 4 compose)
mesh/repo-check pass: its merge-check.sh passed, each part in its toolchain: merge-check.sh (go), merge-check-typescript.sh (typescript)
This commit is contained in:
@@ -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`.
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
@@ -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=
|
||||
|
||||
Reference in New Issue
Block a user