The copy of the retired state ends with a marker: a crash mid-copy is completed at the next start, and nothing is copied after the marker (ADR 0641 section 4, issue 1966)

Claude-Session: https://claude.ai/code/session_01NcbgfFSnn24M3GcqttoU7i
2026-11-05 23:09:52 +01:00
parent c34d6a13cd
commit c006f0b2bc
4 changed files with 62 additions and 33 deletions
@@ -309,9 +309,9 @@ 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.
-	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)
+	// 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)
 	}
 }

@@ -2,9 +2,11 @@ package main

 import (
 	"context"
+	"encoding/json"
 	"errors"
 	"slices"
 	"strings"
+	"time"

 	"github.com/nats-io/nats.go/jetstream"

@@ -33,14 +35,23 @@ type copiedSaid struct {
 	Skipped int `json:"skipped"`
 }

-// 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.
+// copiedMarker is the key, in the retired state, 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"
+
+type copiedMark struct {
+	At     time.Time `json:"at"`
+	Copied int       `json:"copied"`
+}
+
+// copyRetired is the holder's switch from the retired state to `assignments`, with no gap: until the marker is there, it
+// creates in `assignments` each key of the retired state in the form `<node>.node` or `<node>.<module>` that is missing
+// there (never overwriting one), the module keys first and the node keys, which commit a node, last, and then writes the
+// marker. 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. It answers whether the seat may
+// go on: a copy that failed is tried again at the next pass.
 func (p *passer) copyRetired(ctx context.Context) bool {
 	if p.copied {
 		return true
@@ -54,18 +65,13 @@ 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
+	switch _, err := old.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)
+		return false
 	}
 	names, err := old.Keys(ctx)
 	if err != nil && !errors.Is(err, jetstream.ErrNoKeysFound) {
@@ -78,6 +84,9 @@ 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+"."):
@@ -111,15 +120,17 @@ 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)
+		return false
+	}
 	p.copied = true
 	if said == (copiedSaid{}) {
 		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> not copied, %d fitting no form left)",
 		said.Copied, retiredBucket, plc.AssignmentsBucket, said.Present, said.Dropped, said.Skipped)
-	if said.Copied == 0 {
-		return true
-	}
 	if p.seat != nil {
 		if err := p.seat.bus.Emit(CopiedEvent, said); err != nil {
 			p.logf("assignments: the copy is not said as an event: %v", err)
@@ -76,8 +76,6 @@ 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
@@ -308,6 +308,10 @@ 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()
@@ -323,9 +327,12 @@ func TestTheSeatsVerbs(t *testing.T) {
 		}{
 			{"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(),
+			{"a copy cut short by a crash is completed at the next start, and the marker is written", oldWeb(),
+				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()),
 				map[string][]byte{"anchor.node": shell["anchor.node"], "anchor.shell": shell["anchor.shell"]},
-				"", shell, 1, copiedSaid{}},
+				"", shell, 0, 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{}},
@@ -350,7 +357,15 @@ 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 !maps.EqualFunc(old.values, c.old, slices.Equal[[]byte]) || old.deletes != 0 {
+				_, 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)
+				}
+				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 {
 					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 {
@@ -363,9 +378,9 @@ func TestTheSeatsVerbs(t *testing.T) {
 				} 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 assignments filled: nothing is written over it, and no second event is said.
+				// A restart finds the marker: nothing is written, and no second event is said.
 				before := len(next.log)
-				p.copied, p.copying = false, false
+				p.copied = 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)
 				}
@@ -678,6 +693,11 @@ 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)