mesh-module-assigner client: the placement keys assembled into a node's placement, who stands where since when, who holds a seat, what a placement says to the composition; the node key's digest kept as the first writer made it (ADR 0634 §4; C-6; issue 1966)
This commit is contained in:
@@ -0,0 +1,257 @@
|
||||
package meshmoduleassignerclient
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
)
|
||||
|
||||
func TestPlacementClient(t *testing.T) {
|
||||
t.Run("key and file", func(t *testing.T) {
|
||||
// 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 KeyKind
|
||||
}{"g14.node": {"g14", "", NodeKind}, "g14.app": {"g14", "app", ModuleKind},
|
||||
"placement.g14": {"g14", "", LegacyKind}} {
|
||||
if n, m, k, err := ParseKey(key); err != nil || n != want.node || m != want.module || k != want.kind {
|
||||
t.Errorf("ParseKey(%q) = %q, %q, %d, %v", key, n, m, k, err)
|
||||
}
|
||||
}
|
||||
for _, bad := range []string{"want.g14.x", "placement.", "placement.a.b", "g14", "g14.a.b", ".node"} {
|
||||
if _, _, _, err := ParseKey(bad); err == nil {
|
||||
t.Errorf("ParseKey(%q) took it", bad)
|
||||
}
|
||||
}
|
||||
if k, err := NodeKey("g14"); err != nil || k != "g14.node" {
|
||||
t.Errorf("NodeKey = %q, %v", k, err)
|
||||
}
|
||||
if k, err := ModuleKey("g14", "app"); err != nil || k != "g14.app" {
|
||||
t.Errorf("ModuleKey = %q, %v", k, err)
|
||||
}
|
||||
if k, err := NodeKeys("g14"); err != nil || k != "g14.>" {
|
||||
t.Errorf("NodeKeys = %q, %v", k, err)
|
||||
}
|
||||
if k, err := LegacyKey("g14"); err != nil || k != "placement.g14" {
|
||||
t.Errorf("LegacyKey = %q, %v", k, err)
|
||||
}
|
||||
for name, err := range map[string]error{"a dotted node": func() error { _, err := NodeKey("a.b"); return err }(),
|
||||
"a node named placement": func() error { _, err := NodeKey("placement"); return err }(),
|
||||
"a module named node": func() error { _, err := ModuleKey("g14", "node"); return err }()} {
|
||||
if err == nil {
|
||||
t.Errorf("%s was taken", name)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
g14 := seat.Placement{Node: "g14", Revision: "abc",
|
||||
Decided: seat.NodeDecided{Endpoint: "g14.example:51820", Site: "home", Address: "10.77.0.2", Account: "op", Taken: []string{"zsh"}},
|
||||
Modules: []seat.PlacedModule{
|
||||
{Module: "zsh"},
|
||||
{Module: "app", Ports: map[string]int64{"8080": 18080}, Bindings: map[string]seat.Provider{"postgres": {Node: "novox", Module: "postgres"}}},
|
||||
},
|
||||
Pins: map[string]seat.Provider{"shell": {Node: "g14", Module: "zsh"}},
|
||||
Holds: []seat.Holding{{Claim: "node-shell", Scope: "node", Module: "zsh"}},
|
||||
}
|
||||
novox := seat.Placement{Node: "novox", Decided: seat.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []seat.PlacedModule{{Module: "postgres", Ports: map[string]int64{"5432": 5432}}},
|
||||
Holds: []seat.Holding{{Claim: "mesh-store", Scope: "mesh", Module: "postgres"}}}
|
||||
|
||||
t.Run("value", func(t *testing.T) {
|
||||
raw, err := EncodePlacement(g14)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
back, err := DecodePlacement(raw)
|
||||
if err != nil || back.V != Version || !SamePlacement(back, g14) {
|
||||
t.Fatalf("round trip: %+v, %v", back, err)
|
||||
}
|
||||
moved := g14
|
||||
moved.Revision = "def"
|
||||
if !SamePlacement(g14, moved) {
|
||||
t.Error("a new revision alone counts as a change")
|
||||
}
|
||||
if _, err := DecodePlacement([]byte(`{"v":9,"node":"g14","modules":[]}`)); !errors.Is(err, ErrUnknownVersion) {
|
||||
t.Errorf("version 9: %v", err)
|
||||
}
|
||||
for name, bad := range map[string]string{
|
||||
"module twice": `{"v":1,"node":"g14","modules":[{"module":"a"},{"module":"a"}]}`,
|
||||
"port not number": `{"v":1,"node":"g14","modules":[{"module":"a","ports":{"http":80}}]}`,
|
||||
"host port zero": `{"v":1,"node":"g14","modules":[{"module":"a","ports":{"80":0}}]}`,
|
||||
"binding no node": `{"v":1,"node":"g14","modules":[{"module":"a","bindings":{"pg":{"module":"pg"}}}]}`,
|
||||
"pin no module": `{"v":1,"node":"g14","modules":[],"pins":{"pg":{"node":"x"}}}`,
|
||||
"hold not placed": `{"v":1,"node":"g14","modules":[],"holds":[{"claim":"s","module":"a"}]}`,
|
||||
"node not a name": `{"v":1,"node":"g.14","modules":[]}`,
|
||||
"value not shaped": `{"v":1,"node":"g14","modules":{}}`,
|
||||
} {
|
||||
if _, err := DecodePlacement([]byte(bad)); err == nil {
|
||||
t.Errorf("%s: taken", name)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("as the composition reads it", func(t *testing.T) {
|
||||
if got := Assigned(g14); !reflect.DeepEqual(got, []string{"zsh", "app"}) {
|
||||
t.Errorf("Assigned = %v, want the file's order", got)
|
||||
}
|
||||
if got := ChosenPins(g14); !reflect.DeepEqual(got, map[string]catalogue.Chosen{"shell": {Node: "g14", Module: "zsh"}}) {
|
||||
t.Errorf("ChosenPins = %v", got)
|
||||
}
|
||||
if got := Bound(g14); !reflect.DeepEqual(got, map[string]map[string]catalogue.Chosen{"app": {"postgres": {Node: "novox", Module: "postgres"}}}) {
|
||||
t.Errorf("Bound = %v", got)
|
||||
}
|
||||
if got := HostPorts(g14); !reflect.DeepEqual(got, map[string]map[int]int{"app": {8080: 18080}}) {
|
||||
t.Errorf("HostPorts = %v", got)
|
||||
}
|
||||
if got := Held(g14); !reflect.DeepEqual(got, []catalogue.Held{{Claim: "node-shell", Scope: "node", Node: "g14", Module: "zsh"}}) {
|
||||
t.Errorf("Held = %v", got)
|
||||
}
|
||||
layered := seat.PlacedModule{Module: "app", HasMeshWide: true, Layer: map[string]any{"expose": map[string]any{"80": "mesh"}}}
|
||||
want := []catalogue.Layer{
|
||||
{From: catalogue.MeshWideLayer, Values: map[string]any{}},
|
||||
{From: "g14", Values: map[string]any{"mode": "x"}, Placement: map[string]any{"expose": map[string]any{"80": "mesh"}}},
|
||||
}
|
||||
if got := Layers(layered, "g14", map[string]any{"mode": "x"}); !reflect.DeepEqual(got, want) {
|
||||
t.Errorf("Layers = %#v, want %#v", got, want)
|
||||
}
|
||||
if got := Layers(seat.PlacedModule{Module: "a"}, "g14", map[string]any{"k": 1}); len(got) != 1 || got[0].From != "g14" {
|
||||
t.Errorf("values alone make the node's layer: %#v", got)
|
||||
}
|
||||
if got := Layers(seat.PlacedModule{Module: "a", HasLayer: true}, "g14", nil); len(got) != 1 || got[0].Values == nil {
|
||||
t.Errorf("an empty layer of the node's own is kept, with empty values: %#v", got)
|
||||
}
|
||||
if got := Layers(seat.PlacedModule{Module: "a"}, "g14", nil); got != nil {
|
||||
t.Errorf("no layer and no value is no layer: %#v", got)
|
||||
}
|
||||
empty := seat.Placement{Node: "x"}
|
||||
if ChosenPins(empty) != nil || Bound(empty) != nil || HostPorts(empty) != nil || len(Held(empty)) != 0 {
|
||||
t.Error("an empty placement reads as something")
|
||||
}
|
||||
})
|
||||
|
||||
// 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) != 4 || values["g14.node"] == nil || values["g14.zsh"] == nil || values["g14.app"] == nil ||
|
||||
values["placement.g14"] == 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)
|
||||
}
|
||||
if got := InsideAddresses(map[string]seat.Placement{"g14": g14, "novox": novox, "x": {Node: "x"}, "y": {Node: "y", Decided: seat.NodeDecided{Address: " 10.77.0.1"}}}); strings.Join(got, ",") != "10.77.0.1,10.77.0.2" {
|
||||
t.Errorf("InsideAddresses = %v", got)
|
||||
}
|
||||
legacy, _ := EncodePlacement(novox)
|
||||
stale, _ := EncodePlacement(seat.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 := NewKeys()
|
||||
rev := uint64(0)
|
||||
apply := func(writes []KeyWrite, 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 {
|
||||
// The legacy key holds the placement already: no mirror write.
|
||||
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 = []seat.PlacedModule{{Module: "app", Ports: map[string]int64{"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 placement.g14:true:1 g14.zsh:false:2" ||
|
||||
!writes[3].Mirror {
|
||||
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)
|
||||
}
|
||||
if legacy, _ := PlacementsOf(map[string][]byte{"placement.g14": writes[3].Value}); !SamePlacement(legacy["g14"], after) {
|
||||
t.Errorf("the mirror is not the placement written: %+v", legacy)
|
||||
}
|
||||
// A module key that stays is not written again, nor a mirror that holds it.
|
||||
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, _ := EncodeModule(seat.PlacementByNodeModuleValue{Node: "g14", Placed: seat.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 || !repair[1].Commit {
|
||||
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 := DecodeNode([]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")
|
||||
}
|
||||
})
|
||||
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
module git.novox.be/novox/mesh-contracts/placement/go/clients/meshmoduleassignerclient
|
||||
|
||||
go 1.26.0
|
||||
|
||||
require (
|
||||
git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner v0.0.0-20261105000000-000000000000
|
||||
git.novox.be/novox/mesh-sdk/go v0.0.0-20261105000000-000000000000
|
||||
)
|
||||
@@ -0,0 +1,53 @@
|
||||
package meshmoduleassignerclient
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
|
||||
)
|
||||
|
||||
// SeatHoldersKey is a seat's key in the state `seat-holders`: `<seat>` for a seat held once for the mesh (node empty),
|
||||
// `<seat>.<node>` for a node seat on one node.
|
||||
func SeatHoldersKey(name, node string) (string, error) {
|
||||
if err := checkName("seat", name); err != nil {
|
||||
return "", err
|
||||
}
|
||||
if node == "" {
|
||||
return seat.SeatHoldersBySeatKey(name), nil
|
||||
}
|
||||
if err := checkName("node", node); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return seat.SeatHoldersBySeatNodeKey(name, node), nil
|
||||
}
|
||||
|
||||
// HoldersOf reads who holds the seat named (on node, for a node seat) from the state's keys as a reader watched them: the
|
||||
// node's key first, then the seat's. None when nobody holds it.
|
||||
func HoldersOf(keys map[string][]byte, name, node string) ([]seat.SeatHoldersHoldersItem, error) {
|
||||
for _, n := range []string{node, ""} {
|
||||
key, err := SeatHoldersKey(name, n)
|
||||
if err != nil {
|
||||
if n == "" {
|
||||
return nil, err
|
||||
}
|
||||
continue
|
||||
}
|
||||
raw, ok := keys[key]
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
var sh seat.SeatHolders
|
||||
if err := json.Unmarshal(raw, &sh); err != nil {
|
||||
return nil, fmt.Errorf("meshmoduleassignerclient: the holders of %s are not readable: %w", key, err)
|
||||
}
|
||||
if sh.V != Version {
|
||||
return nil, fmt.Errorf("meshmoduleassignerclient: the holders of %s are of version %d, and this reads %d", key, sh.V, Version)
|
||||
}
|
||||
return sh.Holders, nil
|
||||
}
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
var _ = strings.Cut
|
||||
@@ -0,0 +1,65 @@
|
||||
package meshmoduleassignerclient
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
|
||||
)
|
||||
|
||||
// The digest a node key names is kept as the first writer made it, so a node key written before this client reads.
|
||||
func TestDigestIsTheFirstWritersAndHoldersAndStandingRead(t *testing.T) {
|
||||
m := seat.PlacedModule{Module: "app", Ports: map[string]int64{"8080": 18080}, Bindings: map[string]seat.Provider{"postgres": {Node: "novox", Module: "postgres"}},
|
||||
HasMeshWide: true, MeshWide: map[string]any{"a": 1.0}, Layer: map[string]any{"expose": "x"}, HasLayer: true,
|
||||
MeshWideValues: map[string]any{"k": "v"}, Values: map[string]any{"p": 2.0}}
|
||||
if got := Digest(m); got != "62b3c2796d264c42" {
|
||||
t.Errorf("Digest = %s, the first writer's is 62b3c2796d264c42", got)
|
||||
}
|
||||
|
||||
keys := map[string][]byte{
|
||||
"mesh-module-assigner": []byte(`{"v":1,"seat":"mesh-module-assigner","scope":"mesh","holders":[{"module":"mesh-delivery-assigner","node":"novox"}]}`),
|
||||
"node-files.ace": []byte(`{"v":1,"seat":"node-files","scope":"node","node":"ace","holders":[{"module":"node-files","node":"ace"}]}`),
|
||||
"mesh-vault": []byte(`{"v":2,"seat":"mesh-vault","scope":"mesh","holders":[]}`),
|
||||
}
|
||||
for _, c := range []struct {
|
||||
seat, node string
|
||||
want int
|
||||
bad bool
|
||||
}{{"mesh-module-assigner", "", 1, false}, {"node-files", "ace", 1, false}, {"node-files", "g14", 0, false}, {"mesh-vault", "", 0, true}} {
|
||||
got, err := HoldersOf(keys, c.seat, c.node)
|
||||
if (err != nil) != c.bad || len(got) != c.want {
|
||||
t.Errorf("HoldersOf(%s, %q) = %v, %v", c.seat, c.node, got, err)
|
||||
}
|
||||
}
|
||||
|
||||
// A module stands since the revision of its node's key that first named it; the first revision heard stands since
|
||||
// the zero time.
|
||||
p := NewPlaced()
|
||||
t0 := time.Date(2026, 11, 5, 12, 0, 0, 0, time.UTC)
|
||||
node := func(modules ...string) []byte {
|
||||
n := seat.PlacementByNodeNodeValue{Node: "ace"}
|
||||
for _, name := range modules {
|
||||
n.Modules = append(n.Modules, seat.PlacementByNodeNodeValueModulesItem{Module: name, Digest: "d"})
|
||||
}
|
||||
raw, err := EncodeNode(n)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return raw
|
||||
}
|
||||
if _, changed, err := p.Put("ace.node", node("a"), false, t0); err != nil || len(changed) != 1 {
|
||||
t.Fatalf("first: %v, %v", changed, err)
|
||||
}
|
||||
if _, changed, _ := p.Put("ace.node", node("a", "b"), false, t0.Add(time.Hour)); len(changed) != 1 || changed[0] != "b" {
|
||||
t.Fatalf("second: %v", changed)
|
||||
}
|
||||
if since, ok := p.Since("ace", "a"); !ok || !since.IsZero() {
|
||||
t.Errorf("a stands since the zero time: %v %v", since, ok)
|
||||
}
|
||||
if since, ok := p.Since("ace", "b"); !ok || !since.Equal(t0.Add(time.Hour)) {
|
||||
t.Errorf("b stands since the second revision: %v %v", since, ok)
|
||||
}
|
||||
if got := p.PlacedNodes("b"); len(got) != 1 || got[0] != "ace" {
|
||||
t.Errorf("PlacedNodes = %v", got)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,545 @@
|
||||
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 `<node>.>` alone and a change of one module writes that module's key:
|
||||
// `<node>.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 `<node>.<module>`, 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.<node>` 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>.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, `<node>.<module>`; 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, `<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.<node>`, 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>.node or <node>.<module>", 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.<node>`, 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.<node>` 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
|
||||
}
|
||||
@@ -0,0 +1,159 @@
|
||||
package meshmoduleassignerclient
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"maps"
|
||||
"slices"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
|
||||
)
|
||||
|
||||
// Placed is where the placement (the seat mesh-placement's state `placement`, held by the assigner; ADR 0567, ADR 0549 §3) places each module, and since when:
|
||||
// the one reader of that state for every part that asks "does module stand on node" (issue 774, design 47's third
|
||||
// milestone), in place of each part reading the controller's assignments (`mesh-controller_assignments`, which goes
|
||||
// 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, in either form of the state's keys (keys.go): a node's node key
|
||||
// names its modules (a module key says nothing of which stand), and its legacy key `placement.<node>` 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
|
||||
// (the state's history), and either way the module may have run there before it. A key's value that does not read places
|
||||
// nothing on its node, and Put says why.
|
||||
//
|
||||
// Safe for use from a watch and from a reader at once.
|
||||
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>.node` was heard: their legacy key `placement.<node>` is no longer
|
||||
// read (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{}, 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, _, kind, err := ParseKey(key)
|
||||
if err != nil {
|
||||
return "", nil, err
|
||||
}
|
||||
if kind == ModuleKind {
|
||||
return node, nil, nil // which modules stand is the node key's to say
|
||||
}
|
||||
p.mu.Lock()
|
||||
if kind == LegacyKind && p.newForm[node] {
|
||||
p.mu.Unlock()
|
||||
return node, nil, nil // the node is read from its node key once one was heard
|
||||
}
|
||||
if kind == NodeKind {
|
||||
p.newForm[node] = true
|
||||
}
|
||||
p.mu.Unlock()
|
||||
now := map[string]bool{}
|
||||
if !deleted {
|
||||
var modules []string
|
||||
var derr error
|
||||
var named string
|
||||
if kind == NodeKind {
|
||||
var n seat.PlacementByNodeNodeValue
|
||||
n, derr = DecodeNode(value)
|
||||
named = n.Node
|
||||
for _, r := range n.Modules {
|
||||
modules = append(modules, r.Module)
|
||||
}
|
||||
} else {
|
||||
var pl seat.Placement
|
||||
pl, derr = DecodePlacement(value)
|
||||
named, modules = pl.Node, Assigned(pl)
|
||||
}
|
||||
switch {
|
||||
case derr != nil:
|
||||
err = fmt.Errorf("the placement of %s does not read, so it places nothing there: %w", node, derr)
|
||||
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 modules {
|
||||
now[m] = true
|
||||
}
|
||||
}
|
||||
}
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
if !p.heard[node] {
|
||||
p.heard[node] = true
|
||||
at = time.Time{}
|
||||
}
|
||||
was := p.since[node]
|
||||
next := map[string]time.Time{}
|
||||
for m := range now {
|
||||
if s, ok := was[m]; ok {
|
||||
next[m] = s
|
||||
} else {
|
||||
next[m] = at
|
||||
changed = append(changed, m)
|
||||
}
|
||||
}
|
||||
for m := range was {
|
||||
if !now[m] {
|
||||
changed = append(changed, m)
|
||||
}
|
||||
}
|
||||
if len(next) == 0 {
|
||||
delete(p.since, node)
|
||||
} else {
|
||||
p.since[node] = next
|
||||
}
|
||||
slices.Sort(changed)
|
||||
return node, changed, err
|
||||
}
|
||||
|
||||
// Seeded says some node's placement is held: before the assigner's adopt writes the first placement nothing is placed,
|
||||
// and a reader keeps what it did before (ADR 0549 §5).
|
||||
func (p *Placed) Seeded() bool {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
return len(p.since) > 0
|
||||
}
|
||||
|
||||
// Since says whether module stands on node, and since when (the zero time: since before the revisions kept).
|
||||
func (p *Placed) Since(node, module string) (time.Time, bool) {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
s, ok := p.since[node][module]
|
||||
return s, ok
|
||||
}
|
||||
|
||||
// PlacedNodes is every node module is placed on, in name order.
|
||||
func (p *Placed) PlacedNodes(module string) []string {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
var out []string
|
||||
for node, mods := range p.since {
|
||||
if _, ok := mods[module]; ok {
|
||||
out = append(out, node)
|
||||
}
|
||||
}
|
||||
slices.Sort(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// ByNode is, by node, the modules placed on it in name order.
|
||||
func (p *Placed) ByNode() map[string][]string {
|
||||
p.mu.Lock()
|
||||
defer p.mu.Unlock()
|
||||
out := make(map[string][]string, len(p.since))
|
||||
for node, mods := range p.since {
|
||||
out[node] = slices.Sorted(maps.Keys(mods))
|
||||
}
|
||||
return out
|
||||
}
|
||||
@@ -0,0 +1,295 @@
|
||||
// Package meshmoduleassignerclient is the logic every module that reads where modules are placed shares over the seat
|
||||
// mesh-module-assigner (novox/hq ADR 0567, ADR 0634 §4, ADR 0625 §4): the keys of its state `placement` assembled into a
|
||||
// node's placement, who stands on which node since when, who holds a seat, and what a placement says to the composition.
|
||||
// The layouts are the seat's contract's (package meshmoduleassigner); the check of a proposal is the assigner's own.
|
||||
package meshmoduleassignerclient
|
||||
|
||||
import (
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
seat "git.novox.be/novox/mesh-contracts/placement/go/seats/meshmoduleassigner"
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/keys"
|
||||
)
|
||||
|
||||
// Version is the version of every value of the seat's states.
|
||||
const Version = keys.Version
|
||||
|
||||
// NodeWord is the last word of a node key, `<node>.node`; no module is named it.
|
||||
const NodeWord = "node"
|
||||
|
||||
const (
|
||||
legacyWord = "placement"
|
||||
digestLen = 16
|
||||
)
|
||||
|
||||
// ErrUnknownVersion is what a decode returns for a value whose version is missing or not Version.
|
||||
var ErrUnknownVersion = errors.New("meshmoduleassignerclient: unknown value version")
|
||||
|
||||
var nameOK = regexp.MustCompile(`^[-/_=a-zA-Z0-9]+$`)
|
||||
|
||||
func checkName(kind, name string) error {
|
||||
if name == "" {
|
||||
return fmt.Errorf("meshmoduleassignerclient: the %s name is empty", kind)
|
||||
}
|
||||
if !nameOK.MatchString(name) {
|
||||
return fmt.Errorf("meshmoduleassignerclient: the %s name %q holds a dot, a wildcard, whitespace or a character a key refuses", kind, name)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func checkVersion(data []byte) error {
|
||||
var probe struct {
|
||||
V *int `json:"v"`
|
||||
}
|
||||
if err := json.Unmarshal(data, &probe); err != nil {
|
||||
return fmt.Errorf("meshmoduleassignerclient: value is not JSON: %w", err)
|
||||
}
|
||||
if probe.V == nil {
|
||||
return fmt.Errorf("%w: no version in the value, this reader reads %d", ErrUnknownVersion, Version)
|
||||
}
|
||||
if *probe.V != Version {
|
||||
return fmt.Errorf("%w: the value is version %d, this reader reads %d", ErrUnknownVersion, *probe.V, Version)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// CheckPlacement refuses a placement that names a node or module that is not a name, a module twice, a port that is not
|
||||
// a number, a binding or pin naming no provider, and a holding by a module not placed on the node.
|
||||
func CheckPlacement(p seat.Placement) error {
|
||||
if err := checkName("node", p.Node); err != nil {
|
||||
return err
|
||||
}
|
||||
seen := map[string]bool{}
|
||||
for _, m := range p.Modules {
|
||||
if err := checkName("module", m.Module); err != nil {
|
||||
return err
|
||||
}
|
||||
if seen[m.Module] {
|
||||
return fmt.Errorf("meshmoduleassignerclient: the placement of %s names %s twice", p.Node, m.Module)
|
||||
}
|
||||
seen[m.Module] = true
|
||||
for port, at := range m.Ports {
|
||||
if n, err := strconv.Atoi(port); err != nil || n <= 0 || n > 65535 {
|
||||
return fmt.Errorf("meshmoduleassignerclient: %s's port %q on %s is not a port number", m.Module, port, p.Node)
|
||||
}
|
||||
if at <= 0 || at > 65535 {
|
||||
return fmt.Errorf("meshmoduleassignerclient: %s's port %s on %s is published on %d, which is not a port number", m.Module, port, p.Node, at)
|
||||
}
|
||||
}
|
||||
for provision, to := range m.Bindings {
|
||||
if provision == "" || to.Node == "" || to.Module == "" {
|
||||
return fmt.Errorf("meshmoduleassignerclient: %s on %s has a binding with no provision, node or module", m.Module, p.Node)
|
||||
}
|
||||
}
|
||||
}
|
||||
for provision, to := range p.Pins {
|
||||
if provision == "" || to.Node == "" || to.Module == "" {
|
||||
return fmt.Errorf("meshmoduleassignerclient: the placement of %s has a pin with no provision, node or module", p.Node)
|
||||
}
|
||||
}
|
||||
for _, h := range p.Holds {
|
||||
if h.Claim == "" || !seen[h.Module] {
|
||||
return fmt.Errorf("meshmoduleassignerclient: the placement of %s has %s hold %q, which is not a module placed there or names no seat", p.Node, h.Module, h.Claim)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// EncodePlacement is p as JSON at Version; it refuses what DecodePlacement refuses.
|
||||
func EncodePlacement(p seat.Placement) ([]byte, error) {
|
||||
if err := CheckPlacement(p); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
p.V = Version
|
||||
return json.Marshal(p)
|
||||
}
|
||||
|
||||
// DecodePlacement reads a placement file or key value (the form before the node and module keys).
|
||||
func DecodePlacement(data []byte) (seat.Placement, error) {
|
||||
if err := checkVersion(data); err != nil {
|
||||
return seat.Placement{}, err
|
||||
}
|
||||
var p seat.Placement
|
||||
if err := json.Unmarshal(data, &p); err != nil {
|
||||
return seat.Placement{}, fmt.Errorf("meshmoduleassignerclient: value does not fit its shape: %w", err)
|
||||
}
|
||||
if err := CheckPlacement(p); err != nil {
|
||||
return seat.Placement{}, err
|
||||
}
|
||||
return p, nil
|
||||
}
|
||||
|
||||
// SamePlacement says whether two placements are the same value, so the holder writes only when they differ.
|
||||
func SamePlacement(a, b seat.Placement) bool {
|
||||
a.Revision, b.Revision = "", ""
|
||||
ja, errA := EncodePlacement(a)
|
||||
jb, errB := EncodePlacement(b)
|
||||
return errA == nil && errB == nil && string(ja) == string(jb)
|
||||
}
|
||||
|
||||
// Assigned is the modules placed on the node, in the file's order.
|
||||
func Assigned(p seat.Placement) []string {
|
||||
out := make([]string, 0, len(p.Modules))
|
||||
for _, m := range p.Modules {
|
||||
out = append(out, m.Module)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Chosen is p as the composition names a chosen provider.
|
||||
func Chosen(p seat.Provider) catalogue.Chosen {
|
||||
return catalogue.Chosen{Node: p.Node, Module: p.Module}
|
||||
}
|
||||
|
||||
// ChosenPins is the node's pins as the composition reads them; nil when it has none.
|
||||
func ChosenPins(p seat.Placement) map[string]catalogue.Chosen {
|
||||
if len(p.Pins) == 0 {
|
||||
return nil
|
||||
}
|
||||
out := make(map[string]catalogue.Chosen, len(p.Pins))
|
||||
for provision, to := range p.Pins {
|
||||
out[provision] = Chosen(to)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Bound is, by consumer module and provision, where each binding sends it; nil when none.
|
||||
func Bound(p seat.Placement) map[string]map[string]catalogue.Chosen {
|
||||
var out map[string]map[string]catalogue.Chosen
|
||||
for _, m := range p.Modules {
|
||||
if len(m.Bindings) == 0 {
|
||||
continue
|
||||
}
|
||||
if out == nil {
|
||||
out = map[string]map[string]catalogue.Chosen{}
|
||||
}
|
||||
out[m.Module] = make(map[string]catalogue.Chosen, len(m.Bindings))
|
||||
for provision, to := range m.Bindings {
|
||||
out[m.Module][provision] = Chosen(to)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// HostPorts is, by module and the port it declares, the host port it is published on; nil when none. The check has
|
||||
// refused a port that is not a number, so none is dropped here.
|
||||
func HostPorts(p seat.Placement) map[string]map[int]int {
|
||||
var out map[string]map[int]int
|
||||
for _, m := range p.Modules {
|
||||
if len(m.Ports) == 0 {
|
||||
continue
|
||||
}
|
||||
if out == nil {
|
||||
out = map[string]map[int]int{}
|
||||
}
|
||||
out[m.Module] = make(map[int]int, len(m.Ports))
|
||||
for port, at := range m.Ports {
|
||||
n, _ := strconv.Atoi(port)
|
||||
out[m.Module][n] = int(at)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Layers is one module's settings layers on the node, as the composition reads them: the whole mesh's layer when it has
|
||||
// one, then the node's, which carries the module's config values as mesh-config resolved them for this node (values,
|
||||
// nil when none) beside its placement; a module with no layer and no value has none. A module in the settings form
|
||||
// carries its values in the file (MeshWideValues, Values), which then stand for mesh-config's.
|
||||
func Layers(m seat.PlacedModule, node string, values map[string]any) []catalogue.Layer {
|
||||
var out []catalogue.Layer
|
||||
if m.HasMeshWide || len(m.MeshWide) > 0 || len(m.MeshWideValues) > 0 {
|
||||
mesh := map[string]any{}
|
||||
for k, v := range m.MeshWideValues {
|
||||
mesh[k] = v
|
||||
}
|
||||
out = append(out, catalogue.Layer{From: catalogue.MeshWideLayer, Values: mesh, Placement: m.MeshWide})
|
||||
}
|
||||
if len(m.Values) > 0 {
|
||||
values = make(map[string]any, len(m.Values))
|
||||
for k, v := range m.Values {
|
||||
values[k] = v
|
||||
}
|
||||
}
|
||||
if m.HasLayer || len(m.Layer) > 0 || len(values) > 0 {
|
||||
if values == nil {
|
||||
values = map[string]any{}
|
||||
}
|
||||
out = append(out, catalogue.Layer{From: node, Values: values, Placement: m.Layer})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Held is the node's holdings as the composition reads them.
|
||||
func Held(p seat.Placement) []catalogue.Held {
|
||||
out := make([]catalogue.Held, 0, len(p.Holds))
|
||||
for _, h := range p.Holds {
|
||||
out = append(out, catalogue.Held{Claim: h.Claim, Scope: h.Scope, Node: p.Node, Module: h.Module, Site: h.Site, Kind: h.Kind})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// InsideAddresses is every node's address on the private network, sorted and without repeats.
|
||||
func InsideAddresses(placements map[string]seat.Placement) []string {
|
||||
seen := map[string]bool{}
|
||||
out := []string{}
|
||||
for _, p := range placements {
|
||||
if a := strings.TrimSpace(p.Decided.Address); a != "" && !seen[a] {
|
||||
seen[a] = true
|
||||
out = append(out, a)
|
||||
}
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
|
||||
// digestForm is a placed module as its digest has always been made: the fields in the order the first writer wrote them,
|
||||
// so a node key written before this client names a digest this client computes the same.
|
||||
type digestForm struct {
|
||||
Module string `json:"module"`
|
||||
Ports map[string]int64 `json:"ports,omitempty"`
|
||||
Bindings map[string]digestProvider `json:"bindings,omitempty"`
|
||||
MeshWide map[string]any `json:"mesh-wide,omitempty"`
|
||||
HasMesh bool `json:"has-mesh-wide,omitempty"`
|
||||
Layer map[string]any `json:"layer,omitempty"`
|
||||
HasLayer bool `json:"has-layer,omitempty"`
|
||||
MeshValues map[string]any `json:"mesh-wide-values,omitempty"`
|
||||
Values map[string]any `json:"values,omitempty"`
|
||||
}
|
||||
|
||||
type digestProvider struct {
|
||||
Node string `json:"node"`
|
||||
Module string `json:"module"`
|
||||
}
|
||||
|
||||
// Digest is the digest of a placed module's value, as a node key names it.
|
||||
func Digest(m seat.PlacedModule) string {
|
||||
var bindings map[string]digestProvider
|
||||
if len(m.Bindings) > 0 {
|
||||
bindings = make(map[string]digestProvider, len(m.Bindings))
|
||||
for k, v := range m.Bindings {
|
||||
bindings[k] = digestProvider{Node: v.Node, Module: v.Module}
|
||||
}
|
||||
}
|
||||
raw, _ := json.Marshal(digestForm{Module: m.Module, Ports: m.Ports, Bindings: bindings, MeshWide: m.MeshWide, HasMesh: m.HasMeshWide,
|
||||
Layer: m.Layer, HasLayer: m.HasLayer, MeshValues: m.MeshWideValues, Values: m.Values})
|
||||
sum := sha256.Sum256(raw)
|
||||
return hex.EncodeToString(sum[:])[:digestLen]
|
||||
}
|
||||
|
||||
func sortedKeys[V any](m map[string]V) []string {
|
||||
out := make([]string, 0, len(m))
|
||||
for k := range m {
|
||||
out = append(out, k)
|
||||
}
|
||||
sort.Strings(out)
|
||||
return out
|
||||
}
|
||||
Reference in New Issue
Block a user