From c006f0b2bc9fdee0ab4a140777707691daf17182 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 5 Nov 2026 23:09:52 +0100 Subject: [PATCH] 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 --- .../integration_test.go | 6 +- .../cmd/mesh-delivery-assigner/migrate.go | 57 +++++++++++-------- .../cmd/mesh-delivery-assigner/pass.go | 2 - .../mesh-delivery-assigner/seat_tools_test.go | 30 ++++++++-- 4 files changed, 62 insertions(+), 33 deletions(-) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index 16c23f65..12494519 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -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) } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/migrate.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/migrate.go index b1366fcb..cbf90e8a 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/migrate.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/migrate.go @@ -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` or `.` 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` key there is a node leaving (ADR 0641 section 4); the -// state is dropped in the next release. Keys of the form `placement.` 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` or `.` 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` key there is a node leaving (ADR 0641 section 4); the state is dropped in the next release. Keys of the +// form `placement.` 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. 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) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index e0100ebf..eac3a2b1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -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 diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go index 71012108..b228f0a5 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/seat_tools_test.go @@ -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)