The end of the copy is marked in a state of its own, copies, read by assigner_read_copies; nothing is written to the retired state (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

@@ -309,9 +309,12 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
 			t.Errorf("%s was not copied into %s: %v", k, plc.AssignmentsBucket, err)
 		}
 	}
-	// The readers delivered before this one still read the retired state: nothing in it is deleted, and it gains the marker of the end of the copy.
-	if left, err := pb.buckets[retiredBucket].Keys(pb.ctx); err != nil || len(left) != len(values)+2 || !slices.Contains(left, copiedMarker) {
-		t.Errorf("the retired state holds %v (%v), want its %d keys as they were and the marker of the end of the copy", left, err, len(values)+1)
+	// The readers delivered before this one still read the retired state: nothing in it is deleted or written, and the end of the copy is marked in the state copies.
+	if _, err := pb.buckets[copiesBucket].Get(pb.ctx, copiedMarker); err != nil {
+		t.Errorf("the end of the copy is not marked in %s: %v", copiesBucket, err)
+	}
+	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/main.go

@@ -77,7 +77,7 @@ var readFamilies = []string{buckets.ComposedRead}
 var ownBuckets = []string{buckets.Asks, buckets.Differences,
 	buckets.Puts, buckets.Wants, plc.AssignmentsBucket, plc.ProposalsBucket,
 	plc.CandidatesBucket, plc.PulsesBucket, assign.FindingsBucket, assign.PoliciesBucket, plc.SeatHoldersBucket,
-	engineReadsBucket, retiredBucket}
+	engineReadsBucket, retiredBucket, copiesBucket}
 
 // engineReadsState is the assigner's state of what each node's engine states it reads (ADR 0618 §3), one key `<node>`, as
 // the verb manifest-reads last answered it or the fallback read it: what assigner_engine_reads lists and asks again.
@@ -89,7 +89,7 @@ const engineReadsBucket = assignment.Module + "_" + engineReadsState
 // seatBuckets are the seat's states, new with it: until the controller has made them from the manifest and the bus grants
 // them, the assigner starts without them and binds each once the bus says it was created (issue 1790's kind).
 var seatBuckets = []string{plc.AssignmentsBucket, plc.ProposalsBucket, plc.CandidatesBucket,
-	plc.PulsesBucket}
+	plc.PulsesBucket, copiesBucket}
 
 // keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view
 // (and the node agent from release 6), its differences by the view, its asks by the view
@@ -112,6 +112,7 @@ var keptStates = []announce.State{
 	{Name: views.StatusState, Readable: true}, // its first views of the indexer's states (views.go)
 	{Name: engineReadsState, Readable: true},
 	{Name: retiredState}, // read only by the copy into assignments (migrate.go)
+	{Name: copiesState, Readable: true},
 }
 
 // kvSet is every bucket the assigner opened, by name.
@@ -282,6 +283,7 @@ var watched = []string{
 	plc.ProposalsBucket,
 	plc.CandidatesBucket,
 	plc.PulsesBucket,
+	copiesBucket,
 	buckets.Presence,
 	buckets.Have,
 	buckets.Asks,

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

@@ -35,7 +35,14 @@ type copiedSaid struct {
 	Skipped int `json:"skipped"`
 }
 
-// copiedMarker is the key, in the retired state, that says the copy ended: its time and the keys it copied. It is written
+// copiesState is the assigner's own state of its copies, read by its tool assigner_read_copies, and copiesBucket its bucket.
+// The retired state gets nothing written: readers of other modules still parse every key in it.
+const (
+	copiesState  = "copies"
+	copiesBucket = plc.Module + "_" + copiesState
+)
+
+// copiedMarker is the key, in the state copies, that says the copy ended: its time and the keys it copied. It is written
 // after the last key, and the copy is made at every start until it is there and never after, so a copy cut short by a
 // crash is completed at the next start and a key `assignments` dropped after the copy ended is never brought back.
 const copiedMarker = "copied-from-retired"
@@ -56,8 +63,8 @@ func (p *passer) copyRetired(ctx context.Context) bool {
 	if p.copied {
 		return true
 	}
-	old, next := p.kvs[retiredBucket], p.kvs[plc.AssignmentsBucket]
-	if next == nil {
+	old, next, copies := p.kvs[retiredBucket], p.kvs[plc.AssignmentsBucket], p.kvs[copiesBucket]
+	if next == nil || copies == nil {
 		return false
 	}
 	if old == nil {
@@ -65,12 +72,12 @@ func (p *passer) copyRetired(ctx context.Context) bool {
 		p.copied = true
 		return true
 	}
-	switch _, err := old.Get(ctx, copiedMarker); {
+	switch _, err := copies.Get(ctx, copiedMarker); {
 	case err == nil:
 		p.copied = true // the copy ended at an earlier start: nothing is copied again
 		return true
 	case !errors.Is(err, jetstream.ErrKeyNotFound):
-		p.logf("assignments: the marker %s of %s cannot be read, and the copy is tried again at the next pass: %v", copiedMarker, retiredBucket, err)
+		p.logf("assignments: the marker %s of %s cannot be read, and the copy is tried again at the next pass: %v", copiedMarker, copiesBucket, err)
 		return false
 	}
 	names, err := old.Keys(ctx)
@@ -84,9 +91,6 @@ func (p *passer) copyRetired(ctx context.Context) bool {
 	})
 	said := copiedSaid{}
 	for _, key := range names {
-		if key == copiedMarker {
-			continue
-		}
 		_, _, kind, perr := mmac.ParseKey(key)
 		switch {
 		case strings.HasPrefix(key, retiredState+"."):
@@ -121,8 +125,8 @@ func (p *passer) copyRetired(ctx context.Context) bool {
 		}
 	}
 	mark, _ := json.Marshal(copiedMark{At: p.clock(), Copied: said.Copied + said.Present})
-	if _, err := old.Put(ctx, copiedMarker, mark); err != nil {
-		p.logf("assignments: the end of the copy is not written as %s in %s, and the copy is tried again at the next pass: %v", copiedMarker, retiredBucket, err)
+	if _, err := copies.Put(ctx, copiedMarker, mark); err != nil {
+		p.logf("assignments: the end of the copy is not written as %s in %s, and the copy is tried again at the next pass: %v", copiedMarker, copiesBucket, err)
 		return false
 	}
 	p.copied = true

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

@@ -308,10 +308,6 @@ func TestTheSeatsVerbs(t *testing.T) {
 		oldWeb := func() map[string][]byte {
 			return map[string][]byte{"anchor.node": web["anchor.node"], "anchor.web": web["anchor.web"]}
 		}
-		markedOld := func(m map[string][]byte) map[string][]byte {
-			m[copiedMarker] = []byte(`{}`)
-			return m
-		}
 		withLegacy := oldWeb()
 		withLegacy["placement.anchor"] = []byte(`{"whole":1}`)
 		withOdd := oldWeb()
@@ -319,23 +315,24 @@ func TestTheSeatsVerbs(t *testing.T) {
 		for _, c := range []struct {
 			name       string
 			old        map[string][]byte // the retired state
+			marked     bool              // the copy ended at an earlier start
 			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 not copied", withLegacy, nil,
+			{"an empty new state takes every key, the node key last, and the old form is not copied", withLegacy, false, nil,
 				"create anchor.web create anchor.node", web, 1, copiedSaid{Copied: 2, Dropped: 1}},
-			{"a copy cut short by a crash is completed at the next start, and the marker is written", oldWeb(),
+			{"a copy cut short by a crash is completed at the next start, and the marker is written", oldWeb(), false,
 				map[string][]byte{"anchor.web": web["anchor.web"]},
 				"create anchor.node", web, 1, copiedSaid{Copied: 1, Present: 1}},
-			{"a start after the marker copies nothing, even with a key missing", markedOld(oldWeb()),
+			{"a start after the marker copies nothing, even with a key missing", oldWeb(), true,
 				map[string][]byte{"anchor.node": shell["anchor.node"], "anchor.shell": shell["anchor.shell"]},
 				"", shell, 0, copiedSaid{}},
-			{"a key that fits no form is left and counted", withOdd, nil,
+			{"a key that fits no form is left and counted", withOdd, false, 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{}},
+			{"an empty old state copies nothing and says nothing", nil, false, nil, "", nil, 0, copiedSaid{}},
 		} {
 			t.Run(c.name, func(t *testing.T) {
 				next := &revKV{values: map[string][]byte{}, revs: map[string]uint64{}}
@@ -344,9 +341,13 @@ func TestTheSeatsVerbs(t *testing.T) {
 				}
 				old := &retiredKV{values: map[string][]byte{}}
 				maps.Copy(old.values, c.old)
+				copies := &copiesKV{values: map[string][]byte{}}
+				if c.marked {
+					copies.values[copiedMarker] = []byte(`{}`)
+				}
 				var logs []string
 				bus := &emitted{}
-				p := &passer{w: newWorld(), kvs: kvSet{retiredBucket: old, plc.AssignmentsBucket: next}, now: time.Now, seat: &seatActs{bus: bus},
+				p := &passer{w: newWorld(), kvs: kvSet{retiredBucket: old, plc.AssignmentsBucket: next, copiesBucket: copies}, now: time.Now, seat: &seatActs{bus: bus},
 					logf: func(f string, a ...any) { logs = append(logs, fmt.Sprintf(f, a...)) }}
 				if !p.copyRetired(ctx) || !p.copied {
 					t.Fatalf("the copy did not end: %v", logs)
@@ -357,15 +358,10 @@ 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)
 				}
-				_, marked := c.old[copiedMarker]
-				if _, ok := old.values[copiedMarker]; !ok && len(c.old) > 0 || marked != (c.old[copiedMarker] != nil) {
-					t.Errorf("the marker of the end of the copy is missing: %v", old.values)
+				if _, ok := copies.values[copiedMarker]; !ok {
+					t.Errorf("the end of the copy is not marked in %s: %v", copiesBucket, copies.values)
 				}
-				rest := maps.Clone(old.values)
-				delete(rest, copiedMarker)
-				want := maps.Clone(c.old)
-				delete(want, copiedMarker)
-				if !maps.EqualFunc(rest, want, slices.Equal[[]byte]) || old.deletes != 0 {
+				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 {
@@ -693,11 +689,6 @@ func (k *retiredKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry,
 	return retiredEntry{key: key, value: v}, nil
 }
 
-func (k *retiredKV) Put(_ context.Context, key string, value []byte) (uint64, error) {
-	k.values[key] = value
-	return uint64(len(k.values)), nil
-}
-
 func (k *retiredKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error {
 	k.deletes++
 	delete(k.values, key)
@@ -732,3 +723,22 @@ func (k *revKV) Keys(context.Context, ...jetstream.WatchOpt) ([]string, error) {
 	}
 	return slices.Sorted(maps.Keys(k.values)), nil
 }
+
+// copiesKV is the state copies: the marker of the end of a copy.
+type copiesKV struct {
+	jetstream.KeyValue
+	values map[string][]byte
+}
+
+func (k *copiesKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) {
+	v, ok := k.values[key]
+	if !ok {
+		return nil, jetstream.ErrKeyNotFound
+	}
+	return retiredEntry{key: key, value: v}, nil
+}
+
+func (k *copiesKV) Put(_ context.Context, key string, value []byte) (uint64, error) {
+	k.values[key] = value
+	return uint64(len(k.values)), nil
+}

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

@@ -48,6 +48,7 @@ var readTools = []viewedTool{
 	{"assigner_read_findings", mcc.FindingsState, "the assigner's findings, which the condition keeper raises and clears."},
 	{"assigner_read_asks", "asks", "the asks the assigner put to a person, and their answers."},
 	{"assigner_read_differences", "differences", "every difference between a want and what a node runs."},
+	{"assigner_read_copies", copiesState, "the copies the assigner made of a retired state: when the copy into assignments ended and how many keys it copied."},
 }
 
 // viewedStates are the states the views family says, each by its bucket: the six its readers read.
@@ -58,6 +59,7 @@ var viewedStates = map[string]string{
 	mcc.FindingsState:    assign.FindingsBucket,
 	"asks":               buckets.Asks,
 	"differences":        buckets.Differences,
+	copiesState:          copiesBucket,
 }
 
 // viewTools are the assigner's first-view and read tools, reading only its own states, so they answer as soon as its
@@ -90,7 +92,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib
 	}
 	process.AwaitBucket(ctx, js, nc, marksBucket, "no change is said on the views family, so every reader waits on its first view", logf).Ready(func(mk jetstream.KeyValue) {
 		em := viewsEmitter(js, account, mk, findings, tooLarge, logf)
-		for _, state := range []string{"asks", "differences", mcc.FindingsState, mma.StateAssignments, mma.StateSeatHolders, "wants"} {
+		for _, state := range []string{"asks", "differences", mcc.FindingsState, mma.StateAssignments, mma.StateSeatHolders, "wants", copiesState} {
 			state := state
 			start := func(kv jetstream.KeyValue) {
 				if err := em.Start(ctx, state, kv, errs); err != nil {

modules/mesh-delivery-assigner/manifest_test.go

@@ -63,7 +63,7 @@ func TestManifest(t *testing.T) {
 		states[s.Name] = s.History
 	}
 	if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20,
-		"assignments": 64, "placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1,
+		"assignments": 64, "placement": 64, "copies": 1, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1,
 		"engine-reads": 1}; !reflect.DeepEqual(states, want) {
 		t.Errorf("state = %v, want %v", states, want)
 	}
@@ -94,7 +94,7 @@ func TestManifest(t *testing.T) {
 	// program's bundle, which runs in a runtime of its own on the assigner's account; ADR 0618 §3: and assigner_engine_reads,
 	// which lists and asks again what each node's engine reads.
 	if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "assignments", "seat-holders", "wants",
-		"assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences",
+		"assigner_read_assignments", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences", "assigner_read_copies",
 		"assigner_first_views", "assigner_first_view_again", "assigner_engine_reads"}) {
 		t.Errorf("the assigner serves tools %v, want its three put tools, three first views, six read tools and assigner_engine_reads", m.Tools)
 	}
@@ -130,7 +130,7 @@ func TestManifest(t *testing.T) {
 		"seat:node-engine.manifest-reads"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) { // ADR 0618 §3
 		t.Errorf("invokes = %v, want %v", m.Invokes, want)
 	}
-	if want := []string{"assignments-copied", "put", "views.asks", "views.differences", "views.findings", "views.assignments", "views.seat-holders", "views.wants"}; !slices.Equal(m.Emits, want) {
+	if want := []string{"assignments-copied", "put", "views.asks", "views.copies", "views.differences", "views.findings", "views.assignments", "views.seat-holders", "views.wants"}; !slices.Equal(m.Emits, want) {
 		t.Errorf("emits = %v, want the copy, put and its six views", m.Emits)
 	}
 }

modules/mesh-delivery-assigner/module.json

@@ -59,6 +59,7 @@
     "assignments-copied",
     "put",
     "views.asks",
+    "views.copies",
     "views.differences",
     "views.findings",
     "views.assignments",
@@ -78,6 +79,7 @@
     "assigner_read_findings",
     "assigner_read_asks",
     "assigner_read_differences",
+    "assigner_read_copies",
     "assigner_first_views",
     "assigner_first_view_again",
     "assigner_engine_reads"
@@ -85,6 +87,7 @@
   "state": [
     { "name": "assignments", "history": 64 },
     { "name": "placement", "history": 64 },
+    { "name": "copies", "history": 1 },
     { "name": "proposals", "history": 5 },
     { "name": "candidates", "history": 5 },
     { "name": "pulses", "key-ttl": true },