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{}