Merge pull request 'mesh-delivery: write a delivery only when it changed (hq issue 326)' (#187) from fix/326-write-a-delivery-only-when-it-changed into main
This commit was merged in pull request #187.
This commit is contained in:
@@ -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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user