package meshmoduleassignerclient import ( "encoding/json" "errors" "fmt" "sort" "strings" "sync" seat "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner" ) // **One key per module and node** (research 072 row 37; issue 774). The seat'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, its pins and holdings, and the modules placed on it, // each with the digest of its module key's value), and `.`, a module key. The node key is the commit of a // write; 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. The form before, one value `placement.` per node, is still read and mirrored until // a later change deletes it. // KeyKind is the kind of a key of the state `placement`. type KeyKind int // The kinds of key: a node key, a module key, and the form before (one value per node). const ( NodeKind KeyKind = iota + 1 ModuleKind LegacyKind ) // NodeKey is a node's node key, `.node`. A node named `placement` is refused: its keys would read as the // legacy form's. func NodeKey(node string) (string, error) { if err := checkPlacementNode(node); err != nil { return "", err } return node + "." + NodeWord, nil } // ModuleKey is a module's key on a node, `.`; a module named `node` is refused. func ModuleKey(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 } // NodeKeys is the pattern of every key of one node, `.>`: what a node watches of its own placement. func NodeKeys(node string) (string, error) { if err := checkPlacementNode(node); err != nil { return "", err } return node + ".>", nil } // LegacyKey is a node's key in the form before, `placement.`, read while this change is delivered. func LegacyKey(node string) (string, error) { if err := checkName("node", node); err != nil { return "", err } return legacyWord + "." + node, nil } // ParseKey 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 ParseKey(key string) (node, module string, kind KeyKind, 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("meshmoduleassignerclient: %q is not a placement key, .node or .", key) } switch { case first == legacyWord: return rest, "", LegacyKind, nil case rest == NodeWord: return first, "", NodeKind, nil default: return first, rest, ModuleKind, nil } } func checkPlacementNode(node string) error { if err := checkName("node", node); err != nil { return err } if node == legacyWord { return fmt.Errorf("meshmoduleassignerclient: no node is named %q: its placement keys would read as the form before", legacyWord) } return nil } func checkPlacementModule(module string) error { if err := checkName("module", module); err != nil { return err } if module == NodeWord { return fmt.Errorf("meshmoduleassignerclient: no module is named %q: it is the word of a node's own key", NodeWord) } return nil } // EncodeNode is n as JSON at Version. func EncodeNode(n seat.PlacementByNodeNodeValue) ([]byte, error) { if err := checkPlacementNode(n.Node); err != nil { return nil, err } n.V = Version return json.Marshal(n) } // DecodeNode 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 DecodeNode(data []byte) (seat.PlacementByNodeNodeValue, error) { if err := checkVersion(data); err != nil { return seat.PlacementByNodeNodeValue{}, err } n, err := seat.DecodePlacementByNodeNodeValue(data) if err != nil { return n, fmt.Errorf("meshmoduleassignerclient: value does not fit its shape: %w", err) } if err := checkPlacementNode(n.Node); err != nil { return seat.PlacementByNodeNodeValue{}, err } seen := map[string]bool{} for _, r := range n.Modules { if err := checkPlacementModule(r.Module); err != nil { return seat.PlacementByNodeNodeValue{}, err } if seen[r.Module] || r.Digest == "" { return seat.PlacementByNodeNodeValue{}, fmt.Errorf("meshmoduleassignerclient: the node key of %s names %s twice or without a digest", n.Node, r.Module) } seen[r.Module] = true } return n, nil } // EncodeModule is m as JSON at Version. func EncodeModule(m seat.PlacementByNodeModuleValue) ([]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("meshmoduleassignerclient: 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) } // DecodeModule reads a module key's value. func DecodeModule(data []byte) (seat.PlacementByNodeModuleValue, error) { if err := checkVersion(data); err != nil { return seat.PlacementByNodeModuleValue{}, err } m, err := seat.DecodePlacementByNodeModuleValue(data) if err != nil { return m, fmt.Errorf("meshmoduleassignerclient: value does not fit its shape: %w", err) } if err := checkPlacementNode(m.Node); err != nil { return seat.PlacementByNodeModuleValue{}, err } if err := checkPlacementModule(m.Placed.Module); err != nil { return seat.PlacementByNodeModuleValue{}, err } return m, nil } // Keys 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 NodeWord), 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 Keys struct { mu sync.Mutex nodes map[string]*keysOf } // keysOf is what is held of one node's keys. type keysOf struct { node *seat.PlacementByNodeNodeValue nodeRev uint64 modules map[string]seat.PlacementByNodeModuleValue moduleRevs map[string]uint64 legacy *seat.Placement legacyRev uint64 last *seat.Placement // the placement last assembled whole, nil when none } // NewKeys holds no key yet. func NewKeys() *Keys { return &Keys{nodes: map[string]*keysOf{}} } // 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]seat.Placement, error) { k := NewKeys() 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 *Keys) 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 *Keys) put(key string, value []byte, revision uint64, deleted bool) (node string, readErr, err error) { node, module, kind, err := ParseKey(key) if err != nil { return "", err, nil } k.mu.Lock() defer k.mu.Unlock() of := k.nodes[node] if of == nil { of = &keysOf{modules: map[string]seat.PlacementByNodeModuleValue{}, moduleRevs: map[string]uint64{}} k.nodes[node] = of } switch kind { case NodeKind: of.node, of.nodeRev = nil, 0 if !deleted { n, derr := DecodeNode(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 ModuleKind: delete(of.modules, module) delete(of.moduleRevs, module) if !deleted { m, derr := DecodeModule(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 LegacyKind: 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 *keysOf) assemble(node string) (*seat.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 := seat.Placement{V: Version, Node: node, Decided: n.Decided, Pins: n.Pins, Holds: n.Holds, Modules: make([]seat.PlacedModule, 0, len(n.Modules))} var astray []string for _, r := range n.Modules { m, ok := of.modules[r.Module] switch { case ok && Digest(m.Placed) == r.Digest: p.Modules = append(p.Modules, m.Placed) case ok && m.Previous != nil && Digest(*m.Previous) == 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 *Keys) Of(node string) (seat.Placement, uint64, bool) { k.mu.Lock() defer k.mu.Unlock() of := k.nodes[node] if of == nil || of.last == nil { return seat.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 *Keys) All() map[string]seat.Placement { out, _ := k.AllRevisions() return out } // AllRevisions is All with the revision of each node's key. func (k *Keys) AllRevisions() (map[string]seat.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]seat.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 } // KeyWrite 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 KeyWrite 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 // Mirror says this write is the legacy key's, `placement.`, kept in step after the commit for the readers not // yet delivered: a reader of both forms never reads it once the node key is there, so a mirror write that fails // harms no such reader, and is made again at the node's next write. Mirror 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 legacy key `placement.` with the whole placement, a mirror for the readers not yet delivered, unless it // holds it already; 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 *Keys) Writes(node string, after *seat.Placement) ([]KeyWrite, error) { nodeKey, err := NodeKey(node) if err != nil { return nil, err } k.mu.Lock() defer k.mu.Unlock() of := k.nodes[node] if of == nil { of = &keysOf{modules: map[string]seat.PlacementByNodeModuleValue{}, moduleRevs: map[string]uint64{}} } var out []KeyWrite deleteModules := func(keep map[string]bool) { for _, module := range sortedKeys(of.moduleRevs) { if keep[module] { continue } key, _ := ModuleKey(node, module) out = append(out, KeyWrite{Node: node, Key: key, Base: of.moduleRevs[module]}) } } if after == nil { if of.legacy != nil { key, _ := LegacyKey(node) out = append(out, KeyWrite{Node: node, Key: key, Base: of.legacyRev}) } if of.node != nil { out = append(out, KeyWrite{Node: node, Key: nodeKey, Base: of.nodeRev, Commit: true}) } deleteModules(nil) return out, nil } if after.Node != node { return nil, fmt.Errorf("meshmoduleassignerclient: the placement written under %s names the node %q", node, after.Node) } if err := CheckPlacement(*after); err != nil { return nil, err } committed := map[string]seat.PlacedModule{} if of.last != nil { for _, m := range of.last.Modules { committed[m.Module] = m } } n := seat.PlacementByNodeNodeValue{Node: node, Decided: after.Decided, Pins: after.Pins, Holds: after.Holds, Modules: make([]seat.PlacementByNodeNodeValueModulesItem, 0, len(after.Modules))} keep := map[string]bool{} for _, m := range after.Modules { if err := checkPlacementModule(m.Module); err != nil { return nil, err } digest := Digest(m) n.Modules = append(n.Modules, seat.PlacementByNodeNodeValueModulesItem{Module: m.Module, Digest: digest}) keep[m.Module] = true was, wasPlaced := committed[m.Module] if held, ok := of.modules[m.Module]; ok && Digest(held.Placed) == digest && (held.Previous == nil || wasPlaced && Digest(was) == digest) { continue // the key holds it, and no previous value a reader could take in its place } v := seat.PlacementByNodeModuleValue{Node: node, Placed: m} if wasPlaced && Digest(was) != digest { v.Previous = &was } raw, err := EncodeModule(v) if err != nil { return nil, err } key, _ := ModuleKey(node, m.Module) out = append(out, KeyWrite{Node: node, Key: key, Value: raw, Base: of.moduleRevs[m.Module]}) } raw, err := EncodeNode(n) if err != nil { return nil, err } out = append(out, KeyWrite{Node: node, Key: nodeKey, Value: raw, Base: of.nodeRev, Commit: true}) // The mirror of the form before, while readers of it alone may run (the delivery of this change); a later change // stops writing it and deletes it. if of.legacy == nil || !SamePlacement(*of.legacy, *after) { legacy, err := EncodePlacement(*after) if err != nil { return nil, err } key, _ := LegacyKey(node) out = append(out, KeyWrite{Node: node, Key: key, Value: legacy, Base: of.legacyRev, Mirror: 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 kept as the mirror 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 *Keys) Migrations() (map[string][]KeyWrite, 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][]KeyWrite{} 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 *Keys) 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, its node key and the // legacy mirror, as a writer that holds no key of the node (a test's bus, a fresh mesh) writes them. func PlacementValues(p seat.Placement) (map[string][]byte, error) { writes, err := NewKeys().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 } // clonePlacement is p with its own maps and slices, so a reader's copy is not a held one. func clonePlacement(p seat.Placement) seat.Placement { raw, _ := json.Marshal(p) var out seat.Placement _ = json.Unmarshal(raw, &out) if out.Modules == nil { out.Modules = []seat.PlacedModule{} } return out }