From b56c1d00a3ff64f6572d27dd8351bd55bfa048f0 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 03:38:25 +0200 Subject: [PATCH] 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 | 14 +++++++-- .../record_mirrors_switch_test.go | 29 +++++++++++++++++++ internal/inventory/chain.go | 23 +++++++++++++++ internal/inventory/chain_move.go | 12 ++++++++ 4 files changed, 76 insertions(+), 2 deletions(-) create mode 100644 cmd/mesh-controller/record_mirrors_switch_test.go diff --git a/cmd/mesh-controller/record_mirrors.go b/cmd/mesh-controller/record_mirrors.go index 6bd907f7..970956e8 100644 --- a/cmd/mesh-controller/record_mirrors.go +++ b/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() } +} diff --git a/cmd/mesh-controller/record_mirrors_switch_test.go b/cmd/mesh-controller/record_mirrors_switch_test.go new file mode 100644 index 00000000..88ed29a1 --- /dev/null +++ b/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) + } +} diff --git a/internal/inventory/chain.go b/internal/inventory/chain.go index 5b2b42c2..148d22da 100644 --- a/internal/inventory/chain.go +++ b/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) + } +} diff --git a/internal/inventory/chain_move.go b/internal/inventory/chain_move.go index d62363c2..281ce98f 100644 --- a/internal/inventory/chain_move.go +++ b/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{}