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 12494519..28aaa6ce 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,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) } } diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 3ff57552..23cdfc37 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/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 ``, 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, 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 cbf90e8a..f1d27e8b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/migrate.go +++ b/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 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 b228f0a5..45596114 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,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 +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go index 12e2e9c2..34179c8b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/views.go +++ b/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 { diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index d48e2112..a396d3a3 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/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) } } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index bc67d4a7..b1b56213 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/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 },