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 0362b1fc..16c23f65 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 @@ -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) } } 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 5f33b899..b1366fcb 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/migrate.go +++ b/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` or `.` 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.` 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` 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. 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. 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. 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 { 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 eac3a2b1..e0100ebf 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/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 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 1e6aeaac..71012108 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 @@ -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 +}