diff --git a/go/compose/frombus/frombus_test.go b/go/compose/frombus/frombus_test.go index f91e4dc..aa521a4 100644 --- a/go/compose/frombus/frombus_test.go +++ b/go/compose/frombus/frombus_test.go @@ -819,10 +819,11 @@ func statesOf(t *testing.T, ctx context.Context, b *bus) mapStates { out[state][key] = raw } placements, _ := b.Placements(ctx) - for node, p := range placements { - k, _ := delivery.PlacementKey(node) - raw, err := delivery.EncodePlacement(p) - put(PlacementRead, k, raw, err) + for _, p := range placements { + values, err := delivery.PlacementValues(p) + for k, raw := range values { + put(PlacementRead, k, raw, err) + } } records, _ := b.Records(ctx) for node, r := range records { diff --git a/go/compose/frombus/states.go b/go/compose/frombus/states.go index 9962cb9..3ea4618 100644 --- a/go/compose/frombus/states.go +++ b/go/compose/frombus/states.go @@ -68,15 +68,11 @@ func (r statesReader) each(read string, visit func(key string, value []byte)) er } func (r statesReader) Placements(context.Context) (map[string]delivery.Placement, error) { - out := map[string]delivery.Placement{} + keys := delivery.NewPlacementKeys() err := r.each(PlacementRead, func(k string, v []byte) { - if node, err := delivery.ParsePlacementKey(k); err == nil { - if p, err := delivery.DecodePlacement(v); err == nil && p.Node == node { - out[node] = p - } - } + _, _ = keys.Put(k, v, 0, false) // a key that does not read places nothing, as before }) - return out, err + return keys.All(), err } func (r statesReader) Records(context.Context) (map[string]delivery.NodeRecord, error) { diff --git a/go/delivery/buildrun/standing_test.go b/go/delivery/buildrun/standing_test.go index 79845c0..e60b5a6 100644 --- a/go/delivery/buildrun/standing_test.go +++ b/go/delivery/buildrun/standing_test.go @@ -35,10 +35,22 @@ func placing(t *testing.T, node string, modules ...string) (string, []byte) { for _, m := range modules { p.Modules = append(p.Modules, delivery.PlacedModule{Module: m}) } - key, err := delivery.PlacementKey(node) + values, err := delivery.PlacementValues(p) if err != nil { t.Fatal(err) } + key, _ := delivery.PlacementNodeKey(node) + return key, values[key] +} + +// placingLegacy is placing in the form before, `placement.` (research 072 row 37). +func placingLegacy(t *testing.T, node string, modules ...string) (string, []byte) { + t.Helper() + p := delivery.Placement{Node: node} + for _, m := range modules { + p.Modules = append(p.Modules, delivery.PlacedModule{Module: m}) + } + key, _ := delivery.LegacyPlacementKey(node) value, err := delivery.EncodePlacement(p) if err != nil { t.Fatal(err) @@ -177,8 +189,21 @@ func TestToolchainGate(t *testing.T) { t.Run("the history: a module in the first revision heard of a key stands since before it", func(t *testing.T) { p := delivery.NewPlaced() - k, v := placing(t, "ace", NodeBuildModule) + // ace in the form before, then converted: its module stands since before, the legacy key read no more. + k, v := placingLegacy(t, "ace", NodeBuildModule) p.Put(k, v, false, old) + k, v = placing(t, "ace", NodeBuildModule) + if _, changed, err := p.Put(k, v, false, placed); len(changed) != 0 || err != nil { + t.Errorf("the conversion of ace's key changed %v, %v", changed, err) + } + if _, changed, _ := p.Put("ace."+NodeBuildModule, []byte("{}"), false, placed); len(changed) != 0 { + t.Errorf("a module key says which modules stand: %v", changed) + } + if k, v := placingLegacy(t, "ace"); true { + if _, changed, _ := p.Put(k, v, false, placed); len(changed) != 0 { + t.Errorf("the legacy key is read after the node key: %v", changed) + } + } k2, v2 := placing(t, "g14") p.Put(k2, v2, false, old) k2, v2 = placing(t, "g14", NodeBuildModule) diff --git a/go/delivery/control_records_test.go b/go/delivery/control_records_test.go index 51aed3c..0356807 100644 --- a/go/delivery/control_records_test.go +++ b/go/delivery/control_records_test.go @@ -86,7 +86,7 @@ func TestTheControlRecordsReadTheirWritersValues(t *testing.T) { if _, err := ModuleNames(map[string][]byte{"a.b": nil}); err == nil { t.Error("a key that is no module's was read as one") } - if n := MachineNames(map[string][]byte{"placement.novox": nil, "placement.ace": nil, "proposal.p1": nil}); strings.Join(n, ",") != "ace,novox" { + if n := MachineNames(map[string][]byte{"placement.novox": nil, "placement.ace": nil, "ace.node": nil, "ace.app": nil, "g14.app": nil, "proposal.p1": nil}); strings.Join(n, ",") != "ace,novox" { t.Errorf("the machines are %v", n) } if h, _ := HoldersOf(keys, "node-files", "novox"); h != nil { diff --git a/go/delivery/history_records.go b/go/delivery/history_records.go index 714ff85..2649a74 100644 --- a/go/delivery/history_records.go +++ b/go/delivery/history_records.go @@ -264,11 +264,14 @@ func ModuleNames(keys map[string][]byte) ([]string, error) { } // MachineNames is every machine the placement seat places modules on, sorted, from its state `placement` as a reader -// holds its keys: each `placement.`. Keys of the state's other kinds (a proposal's) are passed over. +// holds its keys: each with a node key `.node` or, in the form before, `placement.` (placement_keys.go). +// Module keys and keys of the state's other kinds (a proposal's) are passed over. func MachineNames(keys map[string][]byte) []string { + seen := map[string]bool{} out := []string{} for k := range keys { - if node, err := ParsePlacementKey(k); err == nil { + if node, _, kind, err := ParsePlacementKey(k); err == nil && kind != PlacementModuleKind && !seen[node] { + seen[node] = true out = append(out, node) } } diff --git a/go/delivery/placed.go b/go/delivery/placed.go index 60f2b14..3a4a749 100644 --- a/go/delivery/placed.go +++ b/go/delivery/placed.go @@ -14,7 +14,9 @@ import ( // with the controller's lease, ADR 0549 §6). // // Its input is the state's revisions in the order the bus wrote them, as a watch with the history (jetstream's -// IncludeHistory) hands them: Put takes each. A module stands since the revision of its node's key that first named it +// IncludeHistory) hands them: Put takes each, in either form of the state's keys (placement_keys.go): a node's node key +// names its modules (a module key says nothing of which stand), and its legacy key `placement.` does until a node +// key of it is heard. A module stands since the revision of its node's key that first named it // in an unbroken run. A module in the first revision heard of a key stands since the zero time: that revision is the // key's first (the seed of ADR 0549 §4 placed what already ran there) or the oldest the bus still keeps // (PlacementHistory), and either way the module may have run there before it. A key's value that does not read places @@ -25,31 +27,60 @@ type Placed struct { mu sync.Mutex since map[string]map[string]time.Time // node -> module -> since when it stands heard map[string]bool // the nodes whose key was heard at all + // newForm are the nodes whose node key `.node` was heard: their legacy key `placement.` is no longer + // read (placement_keys.go). + newForm map[string]bool } // NewPlaced is a Placed that holds no placement yet. func NewPlaced() *Placed { - return &Placed{since: map[string]map[string]time.Time{}, heard: map[string]bool{}} + return &Placed{since: map[string]map[string]time.Time{}, heard: map[string]bool{}, newForm: map[string]bool{}} } // Put takes one revision of a key of the state `placement` (its value, whether it is a delete, and when the bus wrote // it). It answers the node the key is of and the modules whose standing on it changed (placed or removed), in name // order, and an error for a key or a value that does not read (the node then places nothing). func (p *Placed) Put(key string, value []byte, deleted bool, at time.Time) (node string, changed []string, err error) { - node, err = ParsePlacementKey(key) + node, _, kind, err := ParsePlacementKey(key) if err != nil { return "", nil, err } + if kind == PlacementModuleKind { + return node, nil, nil // which modules stand is the node key's to say + } + p.mu.Lock() + if kind == PlacementLegacyKind && p.newForm[node] { + p.mu.Unlock() + return node, nil, nil // the node is read from its node key once one was heard + } + if kind == PlacementNodeKind { + p.newForm[node] = true + } + p.mu.Unlock() now := map[string]bool{} if !deleted { - pl, derr := DecodePlacement(value) + var modules []string + var derr error + var named string + if kind == PlacementNodeKind { + var n PlacementNode + n, derr = DecodePlacementNode(value) + named = n.Node + for _, r := range n.Modules { + modules = append(modules, r.Module) + } + } else { + var pl Placement + pl, derr = DecodePlacement(value) + named, modules = pl.Node, pl.Assigned() + } switch { case derr != nil: err = fmt.Errorf("the placement of %s does not read, so it places nothing there: %w", node, derr) - case pl.Node != node: - err = fmt.Errorf("the placement under %s names the node %q, so it places nothing there", key, pl.Node) + case named != node: + err = fmt.Errorf("the placement under %s names the node %q, so it places nothing there", key, named) default: - for _, m := range pl.Assigned() { + for _, m := range modules { now[m] = true } } diff --git a/go/delivery/placement.go b/go/delivery/placement.go index d71a3f3..ec8c438 100644 --- a/go/delivery/placement.go +++ b/go/delivery/placement.go @@ -11,8 +11,9 @@ import ( ) // Placement of modules on nodes is the state of the seat mesh-placement (novox/hq ADR 0567), held once for the mesh by -// the assigner: one key `placement.` per node in its state `placement`, written only on the operator's Approve of a -// proposal the nodes it touches answered (§2-§5), by compare-and-set at the revision the proposal was laid over. No +// the assigner: in its state `placement`, one node key `.node` and one key `.` per module placed +// (placement_keys.go; research 072 row 37), written only on the operator's Approve of a proposal the nodes it touches +// answered (§2-§5), by compare-and-set at the revision the proposal was laid over. No // repository holds it. The node agents, the grants part, mesh-config and the status view read it (§9); nothing in it is // secret. Its states are named after the holding module, as every module state is, and these constants are the seat's // contract: every reader takes the names from here (§1). @@ -42,23 +43,6 @@ const ( // PlacementEvents are the seat's four events, in the order a proposal says them. var PlacementEvents = []string{PlacementProposedEvent, PlacementJudgedEvent, PlacementChangedEvent, PlacementWithdrawnEvent} -// PlacementKey is the key, in PlacementBucket, of one node's placement: `placement.`. -func PlacementKey(node string) (string, error) { - if err := checkName("node", node); err != nil { - return "", err - } - return placementWord + "." + node, nil -} - -// ParsePlacementKey is the inverse of PlacementKey; it refuses any other key, naming it. -func ParsePlacementKey(key string) (node string, err error) { - node, ok := strings.CutPrefix(key, placementWord+".") - if !ok || checkName("node", node) != nil { - return "", fmt.Errorf("delivery: %q is not a placement key, placement.", key) - } - return node, nil -} - // Provider is the module on a node that answers a requirement or holds a seat: a pin's or a binding's provider. type Provider struct { Node string `json:"node"` diff --git a/go/delivery/placement_keys.go b/go/delivery/placement_keys.go new file mode 100644 index 0000000..9f9a962 --- /dev/null +++ b/go/delivery/placement_keys.go @@ -0,0 +1,563 @@ +package delivery + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "errors" + "fmt" + "sort" + "strings" + "sync" +) + +// **One key per module and node** (research 072 row 37, the operator's decision of 2026-10-31; issue 774). The seat +// mesh-placement's state `placement` keeps a node's placement under the node's own keys, so a node watches `.>` +// alone and a change of one module writes that module's key: +// +// - `.node`, the node key: what a person decided of the node (NodeDecided), its pins and its holdings, and the +// modules placed on it in the order the composition reads them, each with the digest of its module key's value. +// The word `node` is the have state's own word for a node's key (HaveNodeWord), so no module is named it. +// - `.`, a module key: the module's ports, bindings and layers (PlacedModule). +// +// The node key is the commit of a write (ADR 0567 §6: a proposal's keys share one revision in history). A writer writes +// the module keys first, each carrying the value it replaces as Previous, then the node key, by compare-and-set at the +// revision the proposal was laid over, and deletes the module keys no longer named last. A reader takes, for each module +// the node key names, the value of its module key whose digest the node key names: the current one once the node key +// is written, the previous one while the write is under way. So a reader never composes a part of a write, and every +// write of a node is one revision of its node key, whose history is the node's placement history as `placement.`'s +// was (Placed, the warrant's record). A module key the node key does not name is not placed. +// +// The form before, one value `placement.` per node (LegacyPlacementKey), is still read during the delivery of +// this change, so readers and the holder may move in any order: a node with a node key is read from its new keys alone, +// one without from its legacy key. The holder converts each legacy value once at its start (PlacementKeys.Migrations) +// and deletes a legacy key with its node key, so a placement handed back never returns from it. +const ( + // PlacementNodeWord is the last word of a node key, `.node`. + PlacementNodeWord = HaveNodeWord + placementDigestLen = 16 +) + +// PlacementKeyKind is the kind of a key of the state `placement`. +type PlacementKeyKind int + +// The kinds of key: a node key, a module key, and the form before (one value per node). +const ( + PlacementNodeKind PlacementKeyKind = iota + 1 + PlacementModuleKind + PlacementLegacyKind +) + +// PlacementNodeKey is a node's node key, `.node`. A node named `placement` is refused: its keys would read as the +// legacy form's. +func PlacementNodeKey(node string) (string, error) { + if err := checkPlacementNode(node); err != nil { + return "", err + } + return node + "." + PlacementNodeWord, nil +} + +// PlacementModuleKey is a module's key on a node, `.`; a module named `node` is refused. +func PlacementModuleKey(node, module string) (string, error) { + if err := checkPlacementNode(node); err != nil { + return "", err + } + if err := checkPlacementModule(module); err != nil { + return "", err + } + return node + "." + module, nil +} + +// PlacementNodeKeys is the pattern of every key of one node, `.>`: what a node watches of its own placement. +func PlacementNodeKeys(node string) (string, error) { + if err := checkPlacementNode(node); err != nil { + return "", err + } + return node + ".>", nil +} + +// LegacyPlacementKey is a node's key in the form before, `placement.`, read while this change is delivered. +func LegacyPlacementKey(node string) (string, error) { + if err := checkName("node", node); err != nil { + return "", err + } + return placementWord + "." + node, nil +} + +// ParsePlacementKey reads a key of the state `placement` in either form: its node, its module (empty for a node key and +// a legacy key) and its kind. Any other key is refused, naming it. +func ParsePlacementKey(key string) (node, module string, kind PlacementKeyKind, err error) { + first, rest, ok := strings.Cut(key, ".") + if !ok || strings.Contains(rest, ".") || checkName("node", first) != nil || checkName("module", rest) != nil { + return "", "", 0, fmt.Errorf("delivery: %q is not a placement key, .node or .", key) + } + switch { + case first == placementWord: + return rest, "", PlacementLegacyKind, nil + case rest == PlacementNodeWord: + return first, "", PlacementNodeKind, nil + default: + return first, rest, PlacementModuleKind, nil + } +} + +func checkPlacementNode(node string) error { + if err := checkName("node", node); err != nil { + return err + } + if node == placementWord { + return fmt.Errorf("delivery: no node is named %q: its placement keys would read as the form before", placementWord) + } + return nil +} + +func checkPlacementModule(module string) error { + if err := checkName("module", module); err != nil { + return err + } + if module == PlacementNodeWord { + return fmt.Errorf("delivery: no module is named %q: it is the word of a node's own key", PlacementNodeWord) + } + return nil +} + +// PlacementRef is one module the node key names, with the digest of the module key's value it stands for. +type PlacementRef struct { + Module string `json:"module"` + Digest string `json:"digest"` +} + +// PlacementNode is the value of a node key: the node's decisions, pins and holdings, and the modules placed on it. +type PlacementNode struct { + V int `json:"v"` + Node string `json:"node"` + Decided NodeDecided `json:"decided"` + Modules []PlacementRef `json:"modules"` + Pins map[string]Provider `json:"pins,omitempty"` + Holds []Holding `json:"holds,omitempty"` +} + +// PlacementModule is the value of a module key: the module as placed, and, while a write that replaces it is under way, +// the value it replaces (Previous), which a reader takes until the node key names the new one. +type PlacementModule struct { + V int `json:"v"` + Node string `json:"node"` + Placed PlacedModule `json:"placed"` + Previous *PlacedModule `json:"previous,omitempty"` +} + +// Digest is the digest of a placed module's value, as a node key names it. +func (m PlacedModule) Digest() string { + raw, _ := json.Marshal(m) + sum := sha256.Sum256(raw) + return hex.EncodeToString(sum[:])[:placementDigestLen] +} + +// EncodePlacementNode is n as JSON at Version. +func EncodePlacementNode(n PlacementNode) ([]byte, error) { + if err := checkPlacementNode(n.Node); err != nil { + return nil, err + } + n.V = Version + return json.Marshal(n) +} + +// DecodePlacementNode reads a node key's value; it refuses an unknown version, a node or module that is not a name, and +// a module named twice or without a digest. +func DecodePlacementNode(data []byte) (PlacementNode, error) { + var n PlacementNode + if err := decode(data, &n); err != nil { + return PlacementNode{}, err + } + if err := checkPlacementNode(n.Node); err != nil { + return PlacementNode{}, err + } + seen := map[string]bool{} + for _, r := range n.Modules { + if err := checkPlacementModule(r.Module); err != nil { + return PlacementNode{}, err + } + if seen[r.Module] || r.Digest == "" { + return PlacementNode{}, fmt.Errorf("delivery: the node key of %s names %s twice or without a digest", n.Node, r.Module) + } + seen[r.Module] = true + } + return n, nil +} + +// EncodePlacementModule is m as JSON at Version. +func EncodePlacementModule(m PlacementModule) ([]byte, error) { + if err := checkPlacementNode(m.Node); err != nil { + return nil, err + } + if err := checkPlacementModule(m.Placed.Module); err != nil { + return nil, err + } + if m.Previous != nil && m.Previous.Module != m.Placed.Module { + return nil, fmt.Errorf("delivery: the key of %s on %s carries %s as its previous value", m.Placed.Module, m.Node, m.Previous.Module) + } + m.V = Version + return json.Marshal(m) +} + +// DecodePlacementModule reads a module key's value. +func DecodePlacementModule(data []byte) (PlacementModule, error) { + var m PlacementModule + if err := decode(data, &m); err != nil { + return PlacementModule{}, err + } + if err := checkPlacementNode(m.Node); err != nil { + return PlacementModule{}, err + } + if err := checkPlacementModule(m.Placed.Module); err != nil { + return PlacementModule{}, err + } + return m, nil +} + +// PlacementKeys is the state `placement` as a reader holds it, key by key in either form, and every node's placement +// assembled from it: the one reader of the state's keys for every part (issue 774), so no part parses them itself. A +// write of the holder's under way is never assembled half: a module key's value is taken only as far as the node key +// names its digest (see the package's comment above PlacementNodeWord), and a node whose keys do not agree keeps the +// placement last assembled for it, Put saying why. +// +// Safe for use from a watch and from a reader at once. +type PlacementKeys struct { + mu sync.Mutex + nodes map[string]*placementKeysOf +} + +// placementKeysOf is what is held of one node's keys. +type placementKeysOf struct { + node *PlacementNode + nodeRev uint64 + modules map[string]PlacementModule + moduleRevs map[string]uint64 + legacy *Placement + legacyRev uint64 + last *Placement // the placement last assembled whole, nil when none +} + +// NewPlacementKeys holds no key yet. +func NewPlacementKeys() *PlacementKeys { return &PlacementKeys{nodes: map[string]*placementKeysOf{}} } + +// PlacementsOf is every node's placement assembled from the state's keys as a reader holds them (key to value; a +// deleted key absent), and an error naming every key that did not read and every node whose keys do not agree (that +// node is then assembled from its module keys' values as they are). +func PlacementsOf(keys map[string][]byte) (map[string]Placement, error) { + k := NewPlacementKeys() + var errs []error + names := make([]string, 0, len(keys)) + for key := range keys { + names = append(names, key) + } + sort.Strings(names) + for _, key := range names { + // A node key read before its module keys does not agree yet; whether the keys agree is asked once all are in. + if _, readErr, _ := k.put(key, keys[key], 0, false); readErr != nil { + errs = append(errs, readErr) + } + } + k.mu.Lock() + for _, node := range sortedKeys(k.nodes) { + if _, _, err := k.nodes[node].assemble(node); err != nil { + errs = append(errs, err) + } + } + k.mu.Unlock() + return k.All(), errors.Join(errs...) +} + +// Put takes one key's value in (revision: the bus's revision of it, 0 when the reader has none; deleted: a delete or a +// purge). It answers the node the key is of and an error for a key or a value that does not read, or for a node whose +// keys do not agree; the node then keeps what it had. +func (k *PlacementKeys) Put(key string, value []byte, revision uint64, deleted bool) (string, error) { + node, readErr, err := k.put(key, value, revision, deleted) + if readErr != nil { + return node, readErr + } + return node, err +} + +// put is Put telling an error of the key or value it took (readErr) from the node's keys not agreeing after it (err). +func (k *PlacementKeys) put(key string, value []byte, revision uint64, deleted bool) (node string, readErr, err error) { + node, module, kind, err := ParsePlacementKey(key) + if err != nil { + return "", err, nil + } + k.mu.Lock() + defer k.mu.Unlock() + of := k.nodes[node] + if of == nil { + of = &placementKeysOf{modules: map[string]PlacementModule{}, moduleRevs: map[string]uint64{}} + k.nodes[node] = of + } + switch kind { + case PlacementNodeKind: + of.node, of.nodeRev = nil, 0 + if !deleted { + n, derr := DecodePlacementNode(value) + switch { + case derr != nil: + readErr = fmt.Errorf("the node key %s does not read, so it places nothing there: %w", key, derr) + case n.Node != node: + readErr = fmt.Errorf("the node key %s names the node %q, so it places nothing there", key, n.Node) + default: + of.node, of.nodeRev = &n, revision + } + } + case PlacementModuleKind: + delete(of.modules, module) + delete(of.moduleRevs, module) + if !deleted { + m, derr := DecodePlacementModule(value) + switch { + case derr != nil: + readErr = fmt.Errorf("the module key %s does not read: %w", key, derr) + case m.Node != node || m.Placed.Module != module: + readErr = fmt.Errorf("the module key %s holds %s on %s", key, m.Placed.Module, m.Node) + default: + of.modules[module] = m + } + // Its revision is kept even when its value does not read, so the holder can write over it. + of.moduleRevs[module] = revision + } + case PlacementLegacyKind: + of.legacy, of.legacyRev = nil, 0 + if !deleted { + p, derr := DecodePlacement(value) + switch { + case derr != nil: + readErr = fmt.Errorf("the placement of %s does not read, so it places nothing there: %w", node, derr) + case p.Node != node: + readErr = fmt.Errorf("the placement under %s names the node %q, so it places nothing there", key, p.Node) + default: + of.legacy, of.legacyRev = &p, revision + } + } + } + p, whole, err := of.assemble(node) + switch { + case p == nil: + of.last = nil + case whole || of.last == nil: + of.last = p + } + return node, readErr, err +} + +// assemble is the node's placement from its keys: nil when none places it, whole false (and an error) when the node key +// names a module digest no module key's value has. +func (of *placementKeysOf) assemble(node string) (*Placement, bool, error) { + if of.node == nil { + if of.legacy == nil { + return nil, true, nil + } + p := clonePlacement(*of.legacy) + return &p, true, nil + } + n := of.node + p := Placement{V: Version, Node: node, Decided: n.Decided, Pins: n.Pins, Holds: n.Holds, Modules: make([]PlacedModule, 0, len(n.Modules))} + var astray []string + for _, r := range n.Modules { + m, ok := of.modules[r.Module] + switch { + case ok && m.Placed.Digest() == r.Digest: + p.Modules = append(p.Modules, m.Placed) + case ok && m.Previous != nil && m.Previous.Digest() == r.Digest: + p.Modules = append(p.Modules, *m.Previous) + case ok: + astray = append(astray, r.Module) + p.Modules = append(p.Modules, m.Placed) + default: + astray = append(astray, r.Module) + } + } + if len(astray) > 0 { + return &p, false, fmt.Errorf("the node key of %s names %s at a value no module key holds: %s keeps the placement it had", + node, strings.Join(astray, ", "), node) + } + return &p, true, nil +} + +// Of is node's placement and the revision of the key it was read from (its node key, or its legacy key), and whether +// any key places it. +func (k *PlacementKeys) Of(node string) (Placement, uint64, bool) { + k.mu.Lock() + defer k.mu.Unlock() + of := k.nodes[node] + if of == nil || of.last == nil { + return Placement{}, 0, false + } + rev := of.nodeRev + if of.node == nil { + rev = of.legacyRev + } + return clonePlacement(*of.last), rev, true +} + +// All is every node's placement, by node, and the revision of each one's key (Of). +func (k *PlacementKeys) All() map[string]Placement { + out, _ := k.AllRevisions() + return out +} + +// AllRevisions is All with the revision of each node's key. +func (k *PlacementKeys) AllRevisions() (map[string]Placement, map[string]uint64) { + k.mu.Lock() + nodes := make([]string, 0, len(k.nodes)) + for n := range k.nodes { + nodes = append(nodes, n) + } + k.mu.Unlock() + out, revs := map[string]Placement{}, map[string]uint64{} + for _, n := range nodes { + if p, rev, ok := k.Of(n); ok { + out[n], revs[n] = p, rev + } + } + return out, revs +} + +// PlacementKeyWrite is one write of a key of the state `placement`, by compare-and-set: Value nil deletes the key at +// Base, Base 0 creates it, any other Base updates it at that revision. +type PlacementKeyWrite struct { + Node string + Key string + Value []byte + Base uint64 + // Commit says this write is the node key's: once it is written, the node's placement is the new one. + Commit bool +} + +// Writes are the writes that take node from the placement its keys hold to after (nil: no placement), in the order +// they are made: the module keys that change, each carrying the value it replaces; then the node key, the commit; then +// the deletes of the module keys after no longer names. For no placement the legacy key goes first, then the node key, +// then the module keys, so a crash between them never leaves the legacy value to be read again. A module key whose value +// stays is not written. +func (k *PlacementKeys) Writes(node string, after *Placement) ([]PlacementKeyWrite, error) { + nodeKey, err := PlacementNodeKey(node) + if err != nil { + return nil, err + } + k.mu.Lock() + defer k.mu.Unlock() + of := k.nodes[node] + if of == nil { + of = &placementKeysOf{modules: map[string]PlacementModule{}, moduleRevs: map[string]uint64{}} + } + var out []PlacementKeyWrite + deleteModules := func(keep map[string]bool) { + for _, module := range sortedKeys(of.moduleRevs) { + if keep[module] { + continue + } + key, _ := PlacementModuleKey(node, module) + out = append(out, PlacementKeyWrite{Node: node, Key: key, Base: of.moduleRevs[module]}) + } + } + if after == nil { + if of.legacy != nil { + key, _ := LegacyPlacementKey(node) + out = append(out, PlacementKeyWrite{Node: node, Key: key, Base: of.legacyRev}) + } + if of.node != nil { + out = append(out, PlacementKeyWrite{Node: node, Key: nodeKey, Base: of.nodeRev, Commit: true}) + } + deleteModules(nil) + return out, nil + } + if after.Node != node { + return nil, fmt.Errorf("delivery: the placement written under %s names the node %q", node, after.Node) + } + if err := after.check(); err != nil { + return nil, err + } + committed := map[string]PlacedModule{} + if of.last != nil { + for _, m := range of.last.Modules { + committed[m.Module] = m + } + } + n := PlacementNode{Node: node, Decided: after.Decided, Pins: after.Pins, Holds: after.Holds, Modules: make([]PlacementRef, 0, len(after.Modules))} + keep := map[string]bool{} + for _, m := range after.Modules { + if err := checkPlacementModule(m.Module); err != nil { + return nil, err + } + digest := m.Digest() + n.Modules = append(n.Modules, PlacementRef{Module: m.Module, Digest: digest}) + keep[m.Module] = true + was, wasPlaced := committed[m.Module] + if held, ok := of.modules[m.Module]; ok && held.Placed.Digest() == digest && + (held.Previous == nil || wasPlaced && was.Digest() == digest) { + continue // the key holds it, and no previous value a reader could take in its place + } + v := PlacementModule{Node: node, Placed: m} + if wasPlaced && was.Digest() != digest { + v.Previous = &was + } + raw, err := EncodePlacementModule(v) + if err != nil { + return nil, err + } + key, _ := PlacementModuleKey(node, m.Module) + out = append(out, PlacementKeyWrite{Node: node, Key: key, Value: raw, Base: of.moduleRevs[m.Module]}) + } + raw, err := EncodePlacementNode(n) + if err != nil { + return nil, err + } + out = append(out, PlacementKeyWrite{Node: node, Key: nodeKey, Value: raw, Base: of.nodeRev, Commit: true}) + deleteModules(keep) + return out, nil +} + +// Migrations are, by node, the writes that convert every node held only in the form before (its legacy key, no node +// key) into its node and module keys, its legacy key left as it is for the readers not yet delivered (Writes); empty +// once every node is converted, so the holder's conversion at each start is idempotent. +func (k *PlacementKeys) Migrations() (map[string][]PlacementKeyWrite, error) { + k.mu.Lock() + var legacy []string + for node, of := range k.nodes { + if of.node == nil && of.legacy != nil && checkPlacementNode(node) == nil { + legacy = append(legacy, node) + } + } + k.mu.Unlock() + sort.Strings(legacy) + out := map[string][]PlacementKeyWrite{} + var errs []error + for _, node := range legacy { + p, _, _ := k.Of(node) + w, err := k.Writes(node, &p) + if err != nil { + errs = append(errs, fmt.Errorf("%s: %w", node, err)) + continue + } + out[node] = w + } + return out, errors.Join(errs...) +} + +// Legacy says node is held only in the form before. +func (k *PlacementKeys) Legacy(node string) bool { + k.mu.Lock() + defer k.mu.Unlock() + of := k.nodes[node] + return of != nil && of.node == nil && of.legacy != nil +} + +// PlacementValues is every key and value of p written whole onto no keys before: its module keys and its node key, as +// a writer that holds no key of the node (a test's bus, a fresh mesh) writes them. +func PlacementValues(p Placement) (map[string][]byte, error) { + writes, err := NewPlacementKeys().Writes(p.Node, &p) + if err != nil { + return nil, err + } + out := make(map[string][]byte, len(writes)) + for _, w := range writes { + out[w.Key] = w.Value + } + return out, nil +} diff --git a/go/delivery/placement_test.go b/go/delivery/placement_test.go index 6ea2dde..1d78e55 100644 --- a/go/delivery/placement_test.go +++ b/go/delivery/placement_test.go @@ -18,20 +18,39 @@ import ( // a change says, and the catalogue's check of every file. func TestPlacement(t *testing.T) { t.Run("key and file", func(t *testing.T) { - key, err := PlacementKey("g14") - if err != nil || key != "placement.g14" { - t.Fatalf("PlacementKey = %q, %v", key, err) + // Research 072 row 37: one node key and one key per module, a legacy key read beside them, any other refused. + for key, want := range map[string]struct { + node, module string + kind PlacementKeyKind + }{"g14.node": {"g14", "", PlacementNodeKind}, "g14.app": {"g14", "app", PlacementModuleKind}, + "placement.g14": {"g14", "", PlacementLegacyKind}} { + if n, m, k, err := ParsePlacementKey(key); err != nil || n != want.node || m != want.module || k != want.kind { + t.Errorf("ParsePlacementKey(%q) = %q, %q, %d, %v", key, n, m, k, err) + } } - if n, err := ParsePlacementKey(key); err != nil || n != "g14" { - t.Fatalf("ParsePlacementKey = %q, %v", n, err) - } - for _, bad := range []string{"want.g14.x", "placement.", "placement.a.b"} { - if _, err := ParsePlacementKey(bad); err == nil { + for _, bad := range []string{"want.g14.x", "placement.", "placement.a.b", "g14", "g14.a.b", ".node"} { + if _, _, _, err := ParsePlacementKey(bad); err == nil { t.Errorf("ParsePlacementKey(%q) took it", bad) } } - if _, err := PlacementKey("a.b"); err == nil { - t.Error("PlacementKey took a dotted node") + if k, err := PlacementNodeKey("g14"); err != nil || k != "g14.node" { + t.Errorf("PlacementNodeKey = %q, %v", k, err) + } + if k, err := PlacementModuleKey("g14", "app"); err != nil || k != "g14.app" { + t.Errorf("PlacementModuleKey = %q, %v", k, err) + } + if k, err := PlacementNodeKeys("g14"); err != nil || k != "g14.>" { + t.Errorf("PlacementNodeKeys = %q, %v", k, err) + } + if k, err := LegacyPlacementKey("g14"); err != nil || k != "placement.g14" { + t.Errorf("LegacyPlacementKey = %q, %v", k, err) + } + for name, err := range map[string]error{"a dotted node": func() error { _, err := PlacementNodeKey("a.b"); return err }(), + "a node named placement": func() error { _, err := PlacementNodeKey("placement"); return err }(), + "a module named node": func() error { _, err := PlacementModuleKey("g14", "node"); return err }()} { + if err == nil { + t.Errorf("%s was taken", name) + } } // ADR 0567 §1: the seat's three states are named after the holding module, and every reader takes the names here. for got, want := range map[string]string{PlacementBucket: "mesh-delivery-assigner_placement", @@ -149,6 +168,118 @@ func TestPlacement(t *testing.T) { } }) + // Research 072 row 37: a node's placement as its node key and module keys, written so a reader never takes a part of + // a write, read beside the form before, and converted from it once. + t.Run("one key per module and node", func(t *testing.T) { + values, err := PlacementValues(g14) + if err != nil || len(values) != 3 || values["g14.node"] == nil || values["g14.zsh"] == nil || values["g14.app"] == nil { + t.Fatalf("PlacementValues = %v, %v", values, err) + } + got, err := PlacementsOf(values) + if err != nil || !SamePlacement(got["g14"], g14) || len(got) != 1 { + t.Fatalf("PlacementsOf = %+v, %v", got, err) + } + legacy, _ := EncodePlacement(novox) + stale, _ := EncodePlacement(Placement{Node: "g14"}) + both := map[string][]byte{"placement.novox": legacy, "placement.g14": stale} + for k, v := range values { + both[k] = v + } + if got, err := PlacementsOf(both); err != nil || !SamePlacement(got["g14"], g14) || !SamePlacement(got["novox"], novox) { + t.Errorf("a node key wins over a legacy key, and a legacy key alone places its node: %+v, %v", got, err) + } + + keys := NewPlacementKeys() + rev := uint64(0) + apply := func(writes []PlacementKeyWrite, upTo int) { + t.Helper() + for i, w := range writes { + if i == upTo { + return + } + rev++ + if _, err := keys.Put(w.Key, w.Value, rev, w.Value == nil); err != nil { + t.Fatalf("write %d (%s): %v", i, w.Key, err) + } + } + } + // The form before converts once: its node and module keys, the legacy key left for the readers not yet delivered. + rev++ + legacyG14, _ := EncodePlacement(g14) + if _, err := keys.Put("placement.g14", legacyG14, rev, false); err != nil || !keys.Legacy("g14") { + t.Fatalf("the legacy key: %v", err) + } + migrations, err := keys.Migrations() + if err != nil || len(migrations["g14"]) != 3 || !migrations["g14"][2].Commit { + t.Fatalf("Migrations = %+v, %v", migrations, err) + } + apply(migrations["g14"], -1) + if m, _ := keys.Migrations(); len(m) != 0 || keys.Legacy("g14") { + t.Fatalf("a converted node converts again: %+v", m) + } + if p, r, ok := keys.Of("g14"); !ok || !SamePlacement(p, g14) || r != rev { + t.Fatalf("Of after the conversion = %+v, %d, %v", p, r, ok) + } + + // A write under way is never read half: app's new port is not read before the node key names it. + after := clonePlacement(g14) + after.Modules = []PlacedModule{{Module: "app", Ports: map[string]int{"8080": 18081}}, {Module: "new"}} + after.Holds, after.Pins = nil, nil + writes, err := keys.Writes("g14", &after) + if err != nil { + t.Fatal(err) + } + var kinds []string + for _, w := range writes { + kinds = append(kinds, fmt.Sprintf("%s:%v:%d", w.Key, w.Value != nil, w.Base)) + } + if strings.Join(kinds, " ") != "g14.app:true:3 g14.new:true:0 g14.node:true:4 g14.zsh:false:2" { + t.Fatalf("Writes = %v", kinds) + } + apply(writes, 2) + if p, _, _ := keys.Of("g14"); !SamePlacement(p, g14) { + t.Errorf("the module keys written before the node key are read: %+v", p) + } + apply(writes[2:], -1) + if p, _, _ := keys.Of("g14"); !SamePlacement(p, after) { + t.Errorf("the write done is not read whole: %+v", p) + } + // A module key that stays is not written again. + if again, _ := keys.Writes("g14", &after); len(again) != 1 || !again[0].Commit { + t.Errorf("an unchanged placement writes %+v", again) + } + // Keys that do not agree keep what the node had, and say so. + stray, _ := EncodePlacementModule(PlacementModule{Node: "g14", Placed: PlacedModule{Module: "app"}}) + if _, err := keys.Put("g14.app", stray, 99, false); err == nil { + t.Error("a module key the node key does not name the value of was taken silently") + } + if p, _, _ := keys.Of("g14"); !SamePlacement(p, after) { + t.Errorf("a stray module key moved the placement: %+v", p) + } + // The holder's next write puts the module key right again. + repair, _ := keys.Writes("g14", &after) + if len(repair) != 2 || repair[0].Key != "g14.app" || repair[0].Base != 99 { + t.Fatalf("the repair writes %+v", repair) + } + apply(repair, -1) + // No placement: the legacy key first, then the node key, then the module keys; none is read after. + gone, _ := keys.Writes("g14", nil) + kinds = nil + for _, w := range gone { + kinds = append(kinds, w.Key) + } + if strings.Join(kinds, " ") != "placement.g14 g14.node g14.app g14.new" { + t.Errorf("Writes(nil) = %v", kinds) + } + apply(gone, 2) + if _, _, ok := keys.Of("g14"); ok { + t.Error("a node key deleted still places its node") + } + if _, err := DecodePlacementNode([]byte(`{"v":1,"node":"g14","modules":[{"module":"a","digest":"x"},{"module":"a","digest":"x"}]}`)); err == nil { + t.Error("a node key naming a module twice reads") + } + }) + t.Run("the catalogue's check", func(t *testing.T) { held := func(m string) bool { return m != "ghost" } enrolled := func(n string) bool { return n != "stranger" }