From 7eb96f2a1d6fad3e78ce4839f8cce36f93011a01 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 10 Oct 2026 19:46:25 +0200 Subject: [PATCH] mesh-delivery: write a delivery only when it changed (hq issue 326) 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. --- .../cmd/mesh-delivery/fakes_test.go | 3 + .../mesh-delivery/cmd/mesh-delivery/holder.go | 24 ++++- .../cmd/mesh-delivery/issue326_test.go | 89 +++++++++++++++++++ 3 files changed, 115 insertions(+), 1 deletion(-) create mode 100644 modules/mesh-delivery/cmd/mesh-delivery/issue326_test.go diff --git a/modules/mesh-delivery/cmd/mesh-delivery/fakes_test.go b/modules/mesh-delivery/cmd/mesh-delivery/fakes_test.go index c110d819..b140619a 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/fakes_test.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/fakes_test.go @@ -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") } diff --git a/modules/mesh-delivery/cmd/mesh-delivery/holder.go b/modules/mesh-delivery/cmd/mesh-delivery/holder.go index ce098104..efd79a18 100644 --- a/modules/mesh-delivery/cmd/mesh-delivery/holder.go +++ b/modules/mesh-delivery/cmd/mesh-delivery/holder.go @@ -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) } } } diff --git a/modules/mesh-delivery/cmd/mesh-delivery/issue326_test.go b/modules/mesh-delivery/cmd/mesh-delivery/issue326_test.go new file mode 100644 index 00000000..2588dd6d --- /dev/null +++ b/modules/mesh-delivery/cmd/mesh-delivery/issue326_test.go @@ -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") + } +}