The store's mirror of the assignments may not write while they are switching or moved, asked before every write (issue 1093)

Review round 2. The writer's leave to write is the lease's and AssignmentsMirrorMayWrite, a flag set before the settle of the assignments' switch, kept after the commit and cleared only if the switch fails, so a pass that read its source before the switch lays and purges nothing after it.

cmd/mesh-controller/record_mirrors.go

@@ -63,7 +63,7 @@ func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.
 				if m.Name == recordmirror.Assignments.Name && onTheBus {
 					continue
 				}
-				why := mirrorOnce(ctx, api, m, writers, sources[m.Name], announce, epoch, holds)
+				why := mirrorOnce(ctx, inv, api, m, writers, sources[m.Name], announce, epoch, holds)
 				if why != failing[m.Name] {
 					if why != "" {
 						fmt.Printf("the %s mirror cannot be kept: %s\n", m.Name, why)
@@ -83,7 +83,7 @@ func mirroringRecords(ctx context.Context, inv *inventory.Inventory, conn *nats.
 }
 
 // mirrorOnce brings one mirror to what the store holds, and answers why it could not; empty when it could.
-func mirrorOnce(ctx context.Context, api jetstream.JetStream, m recordmirror.Mirror, writers map[string]*recordmirror.Writer,
+func mirrorOnce(ctx context.Context, inv *inventory.Inventory, api jetstream.JetStream, m recordmirror.Mirror, writers map[string]*recordmirror.Writer,
 	source func(context.Context) (map[string][]byte, error), announce recordmirror.Announcer, epoch func() uint64,
 	holds func() bool) string {
 	reading, cancel := context.WithTimeout(ctx, 30*time.Second)
@@ -95,6 +95,9 @@ func mirrorOnce(ctx context.Context, api jetstream.JetStream, m recordmirror.Mir
 			return fmt.Sprintf("the bucket %s cannot be read: %v", m.Bucket, err)
 		}
 		w = &recordmirror.Writer{Mirror: m, KV: kv, Announce: announce, Epoch: epoch, Holds: holds}
+		if m.Name == recordmirror.Assignments.Name {
+			w.Holds = assignmentsHolds(inv, holds)
+		}
 		writers[m.Name] = w
 	}
 	want, err := source(reading)
@@ -193,3 +196,10 @@ func keepingAssignments(ctx context.Context, inv *inventory.Inventory, announce
 	events.known = now
 	return true
 }
+
+// assignmentsHolds is the assignments mirror's leave to write: the lease's, and the assignments neither being switched to
+// the bus nor on it. Asked by the writer before every write and purge, so a pass that read its source before the switch
+// writes nothing after it (ADR 0408 §4).
+func assignmentsHolds(inv *inventory.Inventory, holds func() bool) func() bool {
+	return func() bool { return holds() && inv.AssignmentsMirrorMayWrite() }
+}

cmd/mesh-controller/record_mirrors_switch_test.go

@@ -0,0 +1,29 @@
+package main
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/novox/mesh-controller/internal/inventory"
+	"github.com/novox/mesh-controller/internal/recordmirror"
+)
+
+// **A writer of the store's mirror of the assignments, made before the assignments are switched to the bus, writes and
+// purges nothing after** (ADR 0408 §4, issue 1093): its leave to write is asked before every write, and false once they are
+// switching or moved.
+func TestAMirrorWriterMadeBeforeTheSwitchWritesAndPurgesNothingAfter(t *testing.T) {
+	inv := inventory.NewForTest()
+	w := &recordmirror.Writer{Mirror: recordmirror.Assignments, Holds: assignmentsHolds(inv, func() bool { return true })}
+	if !w.Holds() {
+		t.Fatal("the writer may not write before the switch")
+	}
+	inv.BeginSwitchForTest(inventory.KindAssignments)
+	if w.Holds() {
+		t.Fatal("the writer may write while the assignments are switching")
+	}
+	// Refused before it reads the bucket (it has none here) or lays or ends a key.
+	r, err := w.Sync(t.Context(), map[string][]byte{"anchor.notes": []byte(`{}`)})
+	if !errors.Is(err, recordmirror.ErrNotHeld) || r.Written != 0 || r.Removed != 0 {
+		t.Fatalf("the writer wrote after the switch: %+v %v", r, err)
+	}
+}

internal/inventory/chain.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"sync"
+	"sync/atomic"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgconn"
@@ -41,6 +42,18 @@ type movedKinds struct {
 	// is not saved yet.
 	view    *rolloutView
 	pending map[string]pendingMerge
+	// switching is set when the assignments begin to be switched to the bus and kept once they are: the store's mirror of
+	// them writes nothing from then on (AssignmentsMirrorMayWrite). Cleared only if the switch fails.
+	switching atomic.Bool
+}
+
+// AssignmentsMirrorMayWrite says the store's mirror of the assignments may still write the bucket: the assignments are
+// neither being switched to the bus nor on it. Asked before every write of the mirror's writer, not once a pass.
+func (i *Inventory) AssignmentsMirrorMayWrite() bool {
+	i.moved.mu.Lock()
+	moved := i.moved.kind[KindAssignments]
+	i.moved.mu.Unlock()
+	return !moved && !i.moved.switching.Load()
 }
 
 // ChainOpener opens the chain on the bus when a process first needs it: a command that reads and never moved anything has
@@ -124,3 +137,13 @@ func (i *Inventory) ChainMoved(ctx context.Context) (map[string]int, error) {
 	}
 	return counts, err
 }
+
+// NewForTest is an inventory with no store, for a test of what needs none.
+func NewForTest() *Inventory { return &Inventory{moved: &movedKinds{}} }
+
+// BeginSwitchForTest sets the flag a kind's switch sets, as a switch under way does.
+func (i *Inventory) BeginSwitchForTest(kind string) {
+	if kind == KindAssignments {
+		i.moved.switching.Store(true)
+	}
+}

internal/inventory/chain_move.go

@@ -437,6 +437,7 @@ func (i *Inventory) moveKind(ctx context.Context, m mover, say func(string, ...a
 // switchKind records the kind as moved, in one transaction that holds the tables against writes and finds their count
 // as the move read it: false when they took a row since.
 func (i *Inventory) switchKind(ctx context.Context, m mover, r MoveReport) (bool, error) {
+	committed := false
 	tx, err := i.store.Pool().Begin(ctx)
 	if err != nil {
 		return false, err
@@ -452,6 +453,16 @@ func (i *Inventory) switchKind(ctx context.Context, m mover, r MoveReport) (bool
 	if n != r.TableRows {
 		return false, nil
 	}
+	// The store's mirror of the assignments writes nothing from here: it would lay values without the links over the keys
+	// the settle writes and end the ones it ended.
+	if m.kind == KindAssignments {
+		i.moved.switching.Store(true)
+		defer func() {
+			if !committed {
+				i.moved.switching.Store(false)
+			}
+		}()
+	}
 	// The rows read again under the lock: nothing writes them now, so what the bucket holds is made what they say.
 	rows, tableRows, err := m.rows(ctx)
 	if err != nil {
@@ -475,6 +486,7 @@ func (i *Inventory) switchKind(ctx context.Context, m mover, r MoveReport) (bool
 	if err := tx.Commit(ctx); err != nil {
 		return false, err
 	}
+	committed = true
 	i.moved.mu.Lock()
 	if i.moved.kind == nil {
 		i.moved.kind = map[string]bool{}