mesh-delivery: write a delivery only when it changed (hq issue 326)
mesh/merge-gate pass: builds mesh-delivery → 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 delivered

keep wrote the whole delivery on every call, and Reconcile passes every walk
the controller keeps through settleAndKeep every 30 seconds, so the
deliveries bucket was rewritten about once per walk per pass with nothing
changed: a key's revision no longer meant the delivery moved, and a watcher
of the bucket would receive a stream of non-changes.

keep now remembers each delivery as it last wrote it, encoded, and skips the
write when the delivery encodes exactly the same and is not waiting to be
written again. Anything a delivery holds that changes - its state, steps,
walk record, owed effects and their tries - changes its encoding, so every
change is still written; a failed write stays dirty and Flush writes it
again. Events, notes, views and forge statuses are owed and done as before.
A retry of an owed effect still writes its new try count: it is a change.

The test fails before (five passes over unmoved walks wrote five times) and
passes after; a walk that moves is still written and said.
This commit is contained in:
jochen
2026-10-10 19:46:25 +02:00
parent 63412ec36a
commit 7eb96f2a1d
3 changed files with 115 additions and 1 deletions
@@ -19,6 +19,8 @@ type memStore struct {
deliveries map[string][]byte
groups map[string][]byte
failing bool
// puts counts the deliveries written, failed or not.
puts int
}
func newMemStore() *memStore {
@@ -28,6 +30,7 @@ func newMemStore() *memStore {
func (s *memStore) PutDelivery(d *Delivery) error {
s.mu.Lock()
defer s.mu.Unlock()
s.puts++
if s.failing {
return errors.New("the bus is away")
}
@@ -1,6 +1,8 @@
package main
import (
"bytes"
"encoding/json"
"errors"
"fmt"
"slices"
@@ -30,6 +32,9 @@ type Holder struct {
deliveries map[string]*Delivery
groups map[string]*Group
dirty map[string]bool
// written is each delivery as it was last written, encoded: a delivery that encodes the same is not
// written again (novox/hq issue 326).
written map[string][]byte
dirtyGroup map[string]bool
unmatched map[string]unmatched
backlog []Walk
@@ -88,6 +93,7 @@ func (h *Holder) init() {
h.deliveries = map[string]*Delivery{}
h.groups = map[string]*Group{}
h.dirty = map[string]bool{}
h.written = map[string][]byte{}
h.dirtyGroup = map[string]bool{}
h.unmatched = map[string]unmatched{}
h.askTried = map[string]time.Time{}
@@ -114,6 +120,9 @@ func (h *Holder) Load() error {
defer h.mu.Unlock()
for _, d := range ds {
h.deliveries[d.ID] = d
if b, err := json.Marshal(d); err == nil {
h.written[d.ID] = b
}
}
for _, g := range gs {
h.groups[g.ID] = g
@@ -187,15 +196,27 @@ func orNothing(s State) string {
}
// keep writes a delivery; one that could not be written is written again by Flush before anything about it
// is said.
// is said. A delivery that encodes exactly as it was last written is not written again: a revision of its key
// means the delivery changed (novox/hq issue 326). Any change to anything it holds — its state, steps, walk,
// owed effects and their tries — changes its encoding, so it is written.
func (h *Holder) keep(d *Delivery) {
h.deliveries[d.ID] = d
b, encoded := json.Marshal(d)
if encoded == nil && !h.dirty[d.ID] && h.written[d.ID] != nil && bytes.Equal(b, h.written[d.ID]) {
return
}
if err := h.Store.PutDelivery(d); err != nil {
h.dirty[d.ID] = true
delete(h.written, d.ID)
h.Logf("[mesh-delivery] %s could not be kept yet (%v); kept again before it is said", d.ID, err)
return
}
delete(h.dirty, d.ID)
if encoded == nil {
h.written[d.ID] = b
} else {
delete(h.written, d.ID)
}
}
func (h *Holder) keepGroup(g *Group) {
@@ -971,6 +992,7 @@ func (h *Holder) plan() []func() {
if d.State.Final() && len(d.Owed) == 0 && now.Sub(d.Since) > keptFinal {
if err := h.Store.DeleteDelivery(id); err == nil {
delete(h.deliveries, id)
delete(h.written, id)
}
}
}
@@ -0,0 +1,89 @@
package main
import (
"testing"
"time"
)
// novox/hq issue 326: a delivery whose state did not change is not written again — a reconcile pass over
// walks that did not move writes nothing — and one that changed is written, and said.
func TestADeliveryIsWrittenOnlyWhenItChanged(t *testing.T) {
w := newWorld(t)
id := IDOf("novox/app", head)
w.h.PullUpdated(pr("novox/app", 7, head, "feat/x", "src/a.go"))
w.h.Checked(verdict("novox/app", 7, head, "pass"))
w.ctl.put(aWalk("plan-1", "novox/app", merge, true))
w.h.PullMerged(merged("novox/app", 7, head, merge))
w.settleAll()
if err := w.h.Reconcile(); err != nil {
t.Fatal(err)
}
w.settleAll()
// The word given moves the walk's record once more: read back, it is a change, and written.
if err := w.h.Reconcile(); err != nil {
t.Fatal(err)
}
w.settleAll()
if w.state(id) != Delivering {
t.Fatalf("let go, it is %s", w.state(id))
}
// Passes over the same walks, as main.go's tickers run them: nothing changed, nothing written.
before, said := w.store.puts, len(w.said)
for range 5 {
w.later(30 * time.Second)
if err := w.h.Reconcile(); err != nil {
t.Fatal(err)
}
w.settleAll()
}
if w.store.puts != before {
t.Fatalf("five passes over walks that did not move wrote %d delivery(ies)", w.store.puts-before)
}
if len(w.said) != said {
t.Fatalf("five passes over walks that did not move said %d event(s)", len(w.said)-said)
}
// The walk moves: the delivery changes, is written, and is said.
walk := w.ctl.walk("plan-1")
walk.State = walkDone
w.ctl.put(walk)
if err := w.h.Reconcile(); err != nil {
t.Fatal(err)
}
w.settleAll()
if w.state(id) != Delivered {
t.Fatalf("a walk done left it %s", w.state(id))
}
if w.store.puts == before {
t.Fatal("a delivery that changed was not written")
}
back, err := w.store.Deliveries()
if err != nil {
t.Fatal(err)
}
for _, d := range back {
if d.ID == id && d.State != Delivered {
t.Fatalf("the store holds it %s, the holder %s", d.State, Delivered)
}
}
if len(w.said) == said {
t.Fatal("a delivery that changed was not said")
}
}
// A delivery that could not be written is written again, even though it did not change since.
func TestADeliveryNotWrittenIsWrittenAgainUnchanged(t *testing.T) {
w := newWorld(t)
id := IDOf("novox/app", head)
w.store.failing = true
w.h.PullUpdated(pr("novox/app", 7, head, "feat/x", "src/a.go"))
if _, there := w.store.deliveries[id]; there {
t.Fatal("a failing store kept it")
}
w.store.failing = false
w.h.Flush()
if _, there := w.store.deliveries[id]; !there {
t.Fatal("a delivery that could not be written was not written again")
}
}