The assigner only copies the retired state, never deletes from it, and writes nothing over a filled assignments (ADR 0641 section 4, issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go

@@ -291,7 +291,7 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
 	case <-time.After(2 * time.Second):
 	}
 	// ADR 0641: the state the seat's assignments were kept in before holds n1's keys, and the form of one key per node; the
-	// assigner copies the keys into the state made after its start, drops the second form, and leaves the old state empty.
+	// assigner copies the keys into the state made after its start, not the second form, and leaves the old state as it is.
 	values, _ := mmac.AssignmentsValues(mma.Assignments{Node: "n1", Modules: []mma.PlacedModule{{Module: "web"}}})
 	for _, k := range slices.Sorted(maps.Keys(values)) { // a key per module, then the node key
 		pb.putJSON(t, retiredBucket, k, values[k])
@@ -309,8 +309,9 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
 			t.Errorf("%s was not copied into %s: %v", k, plc.AssignmentsBucket, err)
 		}
 	}
-	if left, err := pb.buckets[retiredBucket].Keys(pb.ctx); err == nil || !errors.Is(err, jetstream.ErrNoKeysFound) {
-		t.Errorf("the retired state still holds %v (%v)", left, err)
+	// The readers delivered before this one still read the retired state: nothing in it is deleted.
+	if left, err := pb.buckets[retiredBucket].Keys(pb.ctx); err != nil || len(left) != len(values)+1 {
+		t.Errorf("the retired state holds %v (%v), want its %d keys as they were", left, err, len(values)+1)
 	}
 }
 

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/migrate.go

@@ -33,12 +33,14 @@ type copiedSaid struct {
 	Skipped int `json:"skipped"`
 }
 
-// copyRetired is the holder's switch from the retired state to `assignments`, with no gap: for each key of the retired state
-// in the form `<node>.node` or `<node>.<module>` it copies the raw value into `assignments` when the key is missing there,
-// the module keys first and the node keys, which commit a node, last. A key copied or already there is then deleted from the
-// retired state, so a restart never brings back what `assignments` has dropped since. The keys of the form
-// `placement.<node>` are dropped, and a key that fits no form is left and counted. It answers whether the seat may go on: a
-// copy that failed is tried again at the next pass.
+// copyRetired is the holder's switch from the retired state to `assignments`, with no gap: while `assignments` holds no key,
+// it copies the raw value of each key of the retired state in the form `<node>.node` or `<node>.<module>` into it, the
+// module keys first and the node keys, which commit a node, last. It deletes nothing: the readers delivered before this
+// one still read the retired state, and a delete of a `<node>.node` key there is a node leaving (ADR 0641 section 4); the
+// state is dropped in the next release. Keys of the form `placement.<node>` are not copied, and a key that fits no form is
+// left and counted. A start that finds `assignments` filled writes nothing over it and says so, so a key the seat dropped
+// since is never brought back. It answers whether the seat may go on: a copy that failed is tried again at the next pass,
+// and goes on with the keys written until it ends.
 func (p *passer) copyRetired(ctx context.Context) bool {
 	if p.copied {
 		return true
@@ -52,6 +54,19 @@ func (p *passer) copyRetired(ctx context.Context) bool {
 		p.copied = true
 		return true
 	}
+	if !p.copying {
+		held, err := next.Keys(ctx)
+		if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
+			p.logf("assignments: the keys of %s cannot be listed, and are tried again at the next pass: %v", plc.AssignmentsBucket, err)
+			return false
+		}
+		if len(held) > 0 {
+			p.logf("assignments: %s holds %d keys already, so nothing is copied over it from %s", plc.AssignmentsBucket, len(held), retiredBucket)
+			p.copied = true
+			return true
+		}
+		p.copying = true
+	}
 	names, err := old.Keys(ctx)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
 		p.logf("assignments: the keys of %s cannot be listed, and are tried again at the next pass: %v", retiredBucket, err)
@@ -95,19 +110,15 @@ func (p *passer) copyRetired(ctx context.Context) bool {
 				p.w.putEntry(plc.AssignmentsBucket, key, entry.Value(), rev, p.clock())
 			}
 		}
-		if err := old.Delete(ctx, key); err != nil {
-			p.logf("assignments: %s is not deleted from %s after its copy, and the copy is tried again at the next pass: %v", key, retiredBucket, err)
-			return false
-		}
 	}
 	p.copied = true
 	if said == (copiedSaid{}) {
-		return true
+		return true // the retired state holds nothing
 	}
-	p.logf("assignments: copied %d keys of %s into %s (%d already there kept as they are, %d of the form placement.<node> dropped, %d fitting no form left)",
+	p.logf("assignments: copied %d keys of %s into %s (%d already there kept as they are, %d of the form placement.<node> not copied, %d fitting no form left)",
 		said.Copied, retiredBucket, plc.AssignmentsBucket, said.Present, said.Dropped, said.Skipped)
-	if said.Copied+said.Present+said.Dropped == 0 {
-		return true // only keys left behind: the line above says them at each start, and the event was said with the copy
+	if said.Copied == 0 {
+		return true
 	}
 	if p.seat != nil {
 		if err := p.seat.bus.Emit(CopiedEvent, said); err != nil {

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go

@@ -76,6 +76,8 @@ type passer struct {
 	saidProposed map[string]bool
 	// copied says the retired state was copied into `assignments` by this process, or found empty (copyRetired).
 	copied bool
+	// copying says the copy found `assignments` empty and began: its retries go on with the keys written (copyRetired).
+	copying bool
 	// saidWaits is the last said of the seat's wait for its states (sayWaits).
 	saidWaits string
 	intended  map[string]map[string]map[string]bool

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go

@@ -301,29 +301,34 @@ func TestTheSeatsVerbs(t *testing.T) {
 			t.Errorf("the delete = %v, %v", kv.log, err)
 		}
 	})
-	t.Run("ADR 0641: the retired state is copied into assignments once, key by key, with no gap", func(t *testing.T) {
+	t.Run("ADR 0641: the retired state is copied into assignments, key by key, with no gap and nothing deleted", func(t *testing.T) {
 		ctx := context.Background()
 		web, _ := mmac.AssignmentsValues(mma.Assignments{Node: "anchor", Modules: []mma.PlacedModule{{Module: "web"}}})
 		shell, _ := mmac.AssignmentsValues(mma.Assignments{Node: "anchor", Modules: []mma.PlacedModule{{Module: "shell"}}})
+		oldWeb := func() map[string][]byte {
+			return map[string][]byte{"anchor.node": web["anchor.node"], "anchor.web": web["anchor.web"]}
+		}
+		withLegacy := oldWeb()
+		withLegacy["placement.anchor"] = []byte(`{"whole":1}`)
+		withOdd := oldWeb()
+		withOdd["a.b.c"] = []byte("x")
 		for _, c := range []struct {
-			name         string
-			old, present map[string][]byte // the retired state, and what assignments holds already
-			wantLog      string            // the writes into assignments, in order
-			wantValues   map[string][]byte // assignments afterwards
-			wantLeft     []string          // what the retired state still holds
-			said         copiedSaid        // the event; none when zero
+			name       string
+			old        map[string][]byte // the retired state
+			present    map[string][]byte // what assignments holds at the start
+			wantLog    string            // the writes into assignments, in order
+			wantValues map[string][]byte // assignments afterwards
+			journal    int               // lines said
+			said       copiedSaid        // the event; none when zero
 		}{
-			{"an empty new state takes every key, the node key last, and the old form is dropped",
-				map[string][]byte{"anchor.node": web["anchor.node"], "anchor.web": web["anchor.web"], "placement.anchor": []byte(`{"whole":1}`)},
-				nil, "create anchor.web create anchor.node", web, nil, copiedSaid{Copied: 2, Dropped: 1}},
-			{"a key already in the new state is kept as it is",
-				map[string][]byte{"anchor.node": web["anchor.node"], "anchor.web": web["anchor.web"]},
+			{"an empty new state takes every key, the node key last, and the old form is not copied", withLegacy, nil,
+				"create anchor.web create anchor.node", web, 1, copiedSaid{Copied: 2, Dropped: 1}},
+			{"a start with assignments filled writes nothing over it and says so", oldWeb(),
 				map[string][]byte{"anchor.node": shell["anchor.node"], "anchor.shell": shell["anchor.shell"]},
-				"create anchor.web", map[string][]byte{"anchor.node": shell["anchor.node"], "anchor.shell": shell["anchor.shell"], "anchor.web": web["anchor.web"]}, nil, copiedSaid{Copied: 1, Present: 1}},
-			{"a key that fits no form is left and counted",
-				map[string][]byte{"anchor.node": web["anchor.node"], "anchor.web": web["anchor.web"], "a.b.c": []byte("x")},
-				nil, "create anchor.web create anchor.node", web, []string{"a.b.c"}, copiedSaid{Copied: 2, Skipped: 1}},
-			{"an empty old state copies nothing and says nothing", nil, nil, "", nil, nil, copiedSaid{}},
+				"", shell, 1, copiedSaid{}},
+			{"a key that fits no form is left and counted", withOdd, nil,
+				"create anchor.web create anchor.node", web, 1, copiedSaid{Copied: 2, Skipped: 1}},
+			{"an empty old state copies nothing and says nothing", nil, nil, "", nil, 0, copiedSaid{}},
 		} {
 			t.Run(c.name, func(t *testing.T) {
 				next := &revKV{values: map[string][]byte{}, revs: map[string]uint64{}}
@@ -345,23 +350,25 @@ func TestTheSeatsVerbs(t *testing.T) {
 				if c.wantValues != nil && !maps.EqualFunc(next.values, c.wantValues, slices.Equal[[]byte]) {
 					t.Errorf("assignments = %v, want %v", next.values, c.wantValues)
 				}
-				if left := slices.Sorted(maps.Keys(old.values)); !slices.Equal(left, c.wantLeft) && (len(left) != 0 || len(c.wantLeft) != 0) {
-					t.Errorf("the retired state still holds %v, want %v", left, c.wantLeft)
+				if !maps.EqualFunc(old.values, c.old, slices.Equal[[]byte]) || old.deletes != 0 {
+					t.Errorf("the retired state changed: %v, %d deletes; the readers delivered before still read it", old.values, old.deletes)
+				}
+				if len(logs) != c.journal {
+					t.Errorf("said %d lines %v, want %d", len(logs), logs, c.journal)
 				}
 				if c.said == (copiedSaid{}) {
-					if len(bus.events) != 0 || len(logs) != 0 {
-						t.Errorf("a copy of nothing said %v, %v", bus.events, logs)
+					if len(bus.events) != 0 {
+						t.Errorf("no copy was made, and said %v", bus.events)
 					}
-				} else if len(bus.events) != 1 || bus.events[0] != c.said || len(logs) != 1 {
-					t.Errorf("said %v (journal %v), want the event %+v once", bus.events, logs, c.said)
+				} else if len(bus.events) != 1 || bus.events[0] != c.said {
+					t.Errorf("said %v, want the event %+v once", bus.events, c.said)
 				}
-				// A restart finds the retired state emptied: nothing comes back that assignments has dropped since.
+				// A restart finds assignments filled: nothing is written over it, and no second event is said.
 				before := len(next.log)
-				p.copied = false
+				p.copied, p.copying = false, false
 				if !p.copyRetired(ctx) || len(next.log) != before || len(bus.events) > 1 {
 					t.Errorf("a second copy wrote %v, said %v", next.log[before:], bus.events)
 				}
-				// The copied keys are in the world at once, so the seat's next step reads them.
 				if got := len(p.w.read(plc.AssignmentsBucket)); got != c.said.Copied {
 					t.Errorf("%d copied keys are in the world, want %d", got, c.said.Copied)
 				}
@@ -652,7 +659,8 @@ func setAssignments(kv *memKV, a mma.Assignments) {
 // retiredKV is the retired state: its keys, its values, and deletes of them.
 type retiredKV struct {
 	jetstream.KeyValue
-	values map[string][]byte
+	values  map[string][]byte
+	deletes int
 }
 
 func (k *retiredKV) Keys(context.Context, ...jetstream.WatchOpt) ([]string, error) {
@@ -671,6 +679,7 @@ func (k *retiredKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry,
 }
 
 func (k *retiredKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
+	k.deletes++
 	delete(k.values, key)
 	return nil
 }
@@ -696,3 +705,10 @@ func (e *emitted) Emit(eventType string, body any) error {
 	}
 	return nil
 }
+
+func (k *revKV) Keys(context.Context, ...jetstream.WatchOpt) ([]string, error) {
+	if len(k.values) == 0 {
+		return nil, jetstream.ErrNoKeysFound
+	}
+	return slices.Sorted(maps.Keys(k.values)), nil
+}