diff --git a/go/catalogue/bus_snapshot_test.go b/go/catalogue/bus_snapshot_test.go index 69bd737..789733f 100644 --- a/go/catalogue/bus_snapshot_test.go +++ b/go/catalogue/bus_snapshot_test.go @@ -43,4 +43,49 @@ func TestTheCataloguesBusIsBackedUpBySnapshot(t *testing.T) { if !found { t.Fatal("the bus declares no item for its streams") } + // 774/S3 (ADR 0594 §7): the one list of what is the bus server names the bus's container and every resource that + // restarts it, read from the catalogue's nats manifest at mesh-catalog 0b2d393b; a module not providing the bus + // has none. + t.Run("the bus server's list", func(t *testing.T) { + raw, err := os.ReadFile("testdata/catalogue-0b2d393b-nats.module.json") + if err != nil { + t.Fatal(err) + } + nats, err := decodeForTest(raw) + if err != nil { + t.Fatal(err) + } + server := BusServerOf(nats) + var containers int + for _, r := range nats.Resources { + if r["type"] != "container" { + continue + } + containers++ + restarts, _ := r["restart-on"].([]any) + if !server[r["id"].(string)] || len(restarts) == 0 { + t.Errorf("the bus's container %v (restarted on %v) is not all in the server's list %v", r["id"], restarts, server) + } + for _, id := range restarts { + if !server[id.(string)] { + t.Errorf("%v restarts the bus server and is not in its list %v", id, server) + } + } + } + if containers != 1 || len(server) != len(BusServer) { + t.Errorf("the bus has %d containers and its server is %v of %v", containers, server, BusServer) + } + xterm, err := os.ReadFile("testdata/catalogue-835bf22b-xterm.module.json") + if err != nil { + t.Fatal(err) + } + other, err := decodeForTest(xterm) + if err != nil { + t.Fatal(err) + } + other.Resources = append(other.Resources, map[string]any{"id": "server", "type": "container"}) + if got := BusServerOf(other); len(got) != 0 { + t.Errorf("a module not providing the bus has a bus server %v", got) + } + }) } diff --git a/go/catalogue/testdata/catalogue-0b2d393b-nats.module.json b/go/catalogue/testdata/catalogue-0b2d393b-nats.module.json new file mode 100644 index 0000000..01f3679 --- /dev/null +++ b/go/catalogue/testdata/catalogue-0b2d393b-nats.module.json @@ -0,0 +1,236 @@ +{ + "module": "nats", + "version": "1", + "upgrade": { + "policy": "record", + "why": "the bus: replaced only as a planned step a person starts (`bus upgrade`), its streams snapshotted first and checked after (hq ADR 0236); the controller holds this whatever is said here" + }, + "provides": [ + { + "name": "mesh-bus", + "scope": "mesh", + "credential": { + "form": "none" + } + } + ], + "claims": [ + { + "name": "mesh-broker", + "scope": "mesh" + } + ], + "bus-users": "/var/lib/nats-module/conf/accounts.conf", + "own-secrets": { + "broker": "${dir:mesh-state}/broker", + "broker-nats-tools": "${dir:tools-secret}/broker-nats-tools" + }, + "capabilities": [ + "container-runtime" + ], + "emits": [ + "consumer-removed", + "bucket-keys-removed", + "stream-removed", + "bucket-expiry-set" + ], + "invokes": [ + "seat:mesh-hand-acts.hand-act" + ], + "consumes": [], + "listens": [ + { + "name": "bus", + "port": 4222, + "protocol": "tcp", + "from": "mesh", + "why": "the mesh bus — every link the mesh has, over TLS, reached across the overlay" + } + ], + "tools": [ + "nats_server", + "nats_connections", + "nats_closed_connections", + "nats_subscriptions", + "nats_streams", + "nats_backlog", + "nats_buckets", + "nats_users", + "nats_user_can", + "nats_snapshot_verify", + "nats_consumer_remove", + "nats_bucket_remove", + "nats_stream_remove", + "nats_bucket_expiry" + ], + "guards": [ + 8222 + ], + "data": { + "own": [ + { + "id": "jetstream", + "path": "${dir:jetstream-data}", + "class": "valuable", + "active": "1d", + "backup": { + "dump": "docker exec -i mesh-broker-nats /opt/mesh/mesh-nats-snapshot snapshot < ${dir:mesh-state}/broker > ${dir:snapshots}/bus.tar.partial && mv ${dir:snapshots}/bus.tar.partial ${dir:snapshots}/bus.tar || { rm -f ${dir:snapshots}/bus.tar.partial; exit 1; }", + "into": "snapshots" + }, + "why": "the bus's streams and key-value buckets: the hand-act log, conditions, every module's state; written all the time, so copied by the server's own snapshot of each stream (novox/hq ADR 0235), never as live files" + }, + { + "id": "snapshots", + "path": "${dir:snapshots}", + "class": "rebuildable", + "why": "last night's snapshot of every stream with its manifest, made again every night" + } + ] + }, + "resources": [ + { + "id": "jetstream-data", + "type": "directory", + "path": "/var/lib/mesh-broker-nats", + "mode": "0700", + "unknown": "program" + }, + { + "id": "conf-dir", + "type": "directory", + "path": "/var/lib/nats-module/conf", + "mode": "0700", + "unknown": "program" + }, + { + "id": "tools-secret", + "type": "directory", + "path": "/var/lib/mesh/nats-tools", + "mode": "0711", + "unknown": "program" + }, + { + "id": "snapshots", + "type": "directory", + "mode": "0700", + "unknown": "program" + }, + { + "id": "mesh-state", + "type": "directory", + "mode": "0700", + "place": "mesh", + "unknown": "program" + }, + { + "id": "server-conf", + "type": "file", + "path": "/var/lib/nats-module/conf/nats.conf", + "template": "files/nats.conf", + "mode": "0644" + }, + { + "id": "state", + "type": "directory", + "mode": "0700", + "place": ".", + "unknown": "program" + }, + { + "id": "program-snapshot", + "type": "archive", + "path": "${dir:state}/snapshot", + "artifact": "snapshot", + "unknown": "program" + }, + { + "id": "program-entrypoint", + "type": "archive", + "path": "${dir:state}/entrypoint", + "artifact": "entrypoint", + "unknown": "program" + }, + { + "id": "server", + "type": "container", + "name": "mesh-broker-nats", + "ports": [ + "4222:4222", + "127.0.0.1:8222:8222" + ], + "volumes": [ + "/var/lib/mesh-broker-nats:/data", + "/var/lib/nats-module/conf:/etc/nats:ro", + "${access:tls}:/tls:ro", + "${dir:state}/snapshot:/opt/mesh:ro", + "${dir:state}/entrypoint:/opt/mesh-entrypoint:ro" + ], + "args": [ + "sh", + "/opt/mesh-entrypoint/entrypoint.sh" + ], + "artifact": "server", + "restart-on": [ + "program-snapshot", + "program-entrypoint" + ] + } + ], + "accesses": [ + { + "id": "tls", + "path": "/var/lib/mesh-broker-tls", + "mode": "read" + } + ], + "build": { + "artifacts": [ + { + "name": "server", + "kind": "upstream", + "from": "nats:2.11.17-alpine@sha256:e4bf19f15fd3218814a4e3c9e0064e1334bd8aa20d5984b9f1a0afd084f8cc00" + }, + { + "name": "snapshot", + "kind": "bundle", + "language": "go", + "system": "alpine", + "from": "cmd/mesh-nats-snapshot", + "binary": "mesh-nats-snapshot" + }, + { + "name": "entrypoint", + "kind": "archive", + "from": "program" + }, + { + "name": "tools", + "kind": "bundle", + "language": "go", + "system": "arch", + "from": "cmd/nats-tools", + "binary": "nats-tools", + "loads": [ + "nats-tools" + ], + "bus": { + "removes": [ + "controller-leftovers" + ] + }, + "env": { + "MESH_NATS_MONITOR": "http://127.0.0.1:8222", + "MESH_NATS_CONTAINER": "mesh-broker-nats", + "MESH_NATS_USERS_FILE": "/etc/nats/accounts.conf", + "MESH_NATS_SNAPSHOTS_DIR": "${dir:snapshots}", + "NATS_TOOLS_CREDENTIAL": "${dir:tools-secret}/broker-nats-tools" + } + }, + { + "name": "files", + "kind": "archive", + "from": "files" + } + ] + } +} diff --git a/go/catalogue/upgrade.go b/go/catalogue/upgrade.go index a89e994..c22c5e1 100644 --- a/go/catalogue/upgrade.go +++ b/go/catalogue/upgrade.go @@ -2,6 +2,7 @@ package catalogue import ( "fmt" + "slices" "strings" "time" ) @@ -69,6 +70,26 @@ func ProvidesBus(m Manifest) bool { return false } +// BusServer names the resources of a module that provides the mesh's bus which are its server (novox/hq ADR 0594 §7): +// the server's container and what restarts it. A node agent applying the bus module's want without the warrant's mark +// leaves these as they run (§2), and the assigner says which resources wait for the warrant from this list; neither +// keeps a copy of it. The ids are the catalogue's nats module's (modules/nats/module.json). +var BusServer = []string{"server", "program-snapshot", "program-entrypoint"} + +// BusServerOf is the ids of BusServer that m declares as resources, empty unless m provides the mesh's bus. +func BusServerOf(m Manifest) map[string]bool { + if !ProvidesBus(m) { + return nil + } + of := map[string]bool{} + for _, r := range m.Resources { + if id, _ := r["id"].(string); slices.Contains(BusServer, id) { + of[id] = true + } + } + return of +} + // irreplaceable names the first irreplaceable data the module keeps, its own or its consumers', or // nothing. func (m Manifest) irreplaceable() string { diff --git a/go/delivery/assigner.go b/go/delivery/assigner.go index 1c9cdf0..d477387 100644 --- a/go/delivery/assigner.go +++ b/go/delivery/assigner.go @@ -196,15 +196,37 @@ type Want struct { // Self says the node composes this module itself (ADR 0547 §2): the want names the build alone, and no composition. // A want that is not Self names the one composition it allows (ADR 0539 clause 1), as before. Self bool `json:"self,omitempty"` + // Warrant is the mark of a want that may replace the bus server (novox/hq ADR 0594 §6): WarrantPut on a want a put + // writes on the operator's warrant after the streams' snapshot, WarrantOperator on one the operator's credential + // writes by hand at the terminal. The node agent reads this mark and nothing else to decide whether the bus server + // may be replaced (§2); a want without it never replaces the server. Empty on every other want. + Warrant string `json:"warrant,omitempty"` } +// The warrants a want may carry (Want.Warrant, novox/hq ADR 0594 §6). +const ( + // WarrantPut marks a want the assigner's put wrote on the operator's warrant, after the streams' snapshot. + WarrantPut = "put" + // WarrantOperator marks a want the operator's credential wrote by hand at the terminal. + WarrantOperator = "operator" +) + func (w Want) check() error { if w.Removed { - if w.FP != "" || w.Composed != "" || w.Newest != "" || w.Rule != 0 || w.Backup != "" || w.Put || w.Hold || w.Self || w.Yes != "" { - return fmt.Errorf("delivery: a removed want names no fingerprint, composition, rule, backup answer or put") + if w.FP != "" || w.Composed != "" || w.Newest != "" || w.Rule != 0 || w.Backup != "" || w.Put || w.Hold || w.Self || w.Yes != "" || w.Warrant != "" { + return fmt.Errorf("delivery: a removed want names no fingerprint, composition, rule, backup answer, put or warrant") } return nil } + switch w.Warrant { + case "", WarrantOperator: + case WarrantPut: + if !w.Put { + return fmt.Errorf("delivery: only a put's want carries the put's warrant") + } + default: + return fmt.Errorf("delivery: the warrant %q is not %s or %s", w.Warrant, WarrantPut, WarrantOperator) + } if w.Hold && !w.Put { return fmt.Errorf("delivery: only a put holds a want") } diff --git a/go/delivery/assigner_test.go b/go/delivery/assigner_test.go index b26279f..bbaec4e 100644 --- a/go/delivery/assigner_test.go +++ b/go/delivery/assigner_test.go @@ -79,6 +79,9 @@ func TestAssignerValuesRoundTrip(t *testing.T) { {FP: fp2of('a'), Composed: hashOf('1'), Newest: fp2of('b'), Rule: RuleCanary, Words: "the canary ace does not run it healthy yet"}, {Removed: true, Words: "the module left g14"}, {FP: fp2of('a'), Composed: hashOf('1'), Put: true, Hold: true, Words: "put by the operator: the bus's user list"}, + // 774/S1: a want that may replace the bus server says its warrant (ADR 0594 §6). + {FP: fp2of('a'), Composed: hashOf('1'), Put: true, Warrant: WarrantPut}, + {FP: fp2of('a'), Self: true, Warrant: WarrantOperator}, } for _, w := range wants { data, err := EncodeWant(w) @@ -151,6 +154,10 @@ func TestAssignerValuesRefused(t *testing.T) { {FP: fp2of('a'), Composed: hashOf('1'), Hold: true}, {FP: fp2of('a'), Composed: hashOf('1'), Put: true, Newest: fp2of('b'), Rule: 3}, {Removed: true, Put: true}, + // 774/S1: a removed want carries no warrant, a warrant is put or operator, and put only on a put's want. + {Removed: true, Warrant: WarrantOperator}, + {FP: fp2of('a'), Self: true, Warrant: "yes"}, + {FP: fp2of('a'), Self: true, Warrant: WarrantPut}, } { if _, err := EncodeWant(w); err == nil { t.Errorf("EncodeWant(%+v) accepted", w) diff --git a/go/delivery/delivery.go b/go/delivery/delivery.go index 3b72bb7..839f3fd 100644 --- a/go/delivery/delivery.go +++ b/go/delivery/delivery.go @@ -360,6 +360,12 @@ type Have struct { // slot that holds it, so a provider and a consumer of one value on one node each keep their own. Written in Sealed's // place by a node-engine that knows it; Sealed is read only from a have written before. SealedAt Seals `json:"sealed-at,omitempty"` + // Server is, in the have of the bus module, what the bus server runs (novox/hq ADR 0594 §4): the fingerprint whose + // server the node agent last made, or, from the take-over of a server a node's own composition made, that old-form + // build's name (`build-...`). FP is the fingerprint applied: the two differ while a build changed the server and no + // warranted want replaced it yet. Empty on every other have, and on a bus module's have from a node agent that does + // not keep the server. + Server string `json:"server,omitempty"` } // UserListWritten is the bus's user list as the node-engine of the bus node wrote it to the bus module's file, applied @@ -421,8 +427,15 @@ type HaveNode struct { // controller ended may still say `controller` (composedByTheEndedController), which is read and never written. Composes string `json:"composes,omitempty"` ComposesWhy string `json:"composes-why,omitempty"` + // Keeps names what the node agent keeps as it runs on a want without the warrant's mark (novox/hq ADR 0594 §8): + // KeepsBusServer for a node agent that leaves the bus server alone. Written at each start; the assigner writes the + // bus module's want only to a node whose have names it. Empty from a node agent before it. + Keeps []string `json:"keeps,omitempty"` } +// KeepsBusServer is a node agent that keeps the bus server unless a want is warranted (HaveNode.Keeps, ADR 0594 §8). +const KeepsBusServer = "bus-server" + // ComposesOwn is a node applying its own compositions (HaveNode.Composes). const ComposesOwn = "own" @@ -494,6 +507,10 @@ func (h Have) check() error { return err } } + if h.Server != "" && checkFingerprint(h.Server, "the bus server's fingerprint") != nil && + (!strings.HasPrefix(h.Server, "build-") || checkName("bus server's build", h.Server) != nil) { + return fmt.Errorf("delivery: the bus server runs %q, neither a fingerprint nor a build's name", h.Server) + } switch h.Health { case HealthOK, HealthFailed, HealthWaits, HealthUnknown: default: @@ -529,6 +546,11 @@ func (n HaveNode) check() error { default: return fmt.Errorf("delivery: a node composes %q, not %s", n.Composes, ComposesOwn) } + for _, k := range n.Keeps { + if k == "" { + return fmt.Errorf("delivery: a node keeps an empty name") + } + } for group, a := range n.Candidates { if group == "" { return fmt.Errorf("delivery: a node's candidate answer names no group") @@ -615,7 +637,7 @@ func SameHaveNode(a, b HaveNode) bool { sameJSON(a.Strays, b.Strays) && sameJSON(a.Filters, b.Filters) && sameJSON(a.Profile, b.Profile) && (a.Record == nil) == (b.Record == nil) && (a.Record == nil || SameNodeRecord(*a.Record, *b.Record)) && jsonEqual(a.Sealed, b.Sealed) && maps.Equal(a.SealedAt, b.SealedAt) && jsonEqual(a.Candidates, b.Candidates) && a.Composes == b.Composes && - a.ComposesWhy == b.ComposesWhy + a.ComposesWhy == b.ComposesWhy && slices.Equal(a.Keeps, b.Keeps) } // SamePresence says whether two presences are the same value. diff --git a/go/delivery/delivery_test.go b/go/delivery/delivery_test.go index 328fad7..6dbf110 100644 --- a/go/delivery/delivery_test.go +++ b/go/delivery/delivery_test.go @@ -282,6 +282,51 @@ func TestRoundTrip(t *testing.T) { t.Error("a list was sealed to no key") } }) + // 774/S2: the bus module's have names the fingerprint applied and what the server runs, a fingerprint or, at the + // take-over, an old-form build's name (ADR 0594 §4); a have written before decodes as it was. + t.Run("have of the bus with its server", func(t *testing.T) { + fpA, fpB := fp2of('a'), fp2of('b') + for _, server := range []string{fpA, "build-1792674595610176744"} { + in := Have{FP: fpB, Server: server, Phase: PhaseRunning, Health: HealthOK} + b, err := EncodeHave(in) + if err != nil { + t.Fatal(err) + } + if !strings.Contains(string(b), `"server":"`+server+`"`) { + t.Errorf("the have says %s", b) + } + out, err := DecodeHave(b) + in.V = Version + if err != nil || !SameHave(out, in) || SameHave(out, Have{V: Version, FP: fpB, Phase: PhaseRunning, Health: HealthOK}) { + t.Errorf("got %+v, %v", out, err) + } + } + for _, server := range []string{"nats", "build-a.b", "src2:abc"} { + if _, err := EncodeHave(Have{FP: fpB, Server: server, Phase: PhaseRunning, Health: HealthOK}); err == nil { + t.Errorf("a server running %q was encoded", server) + } + } + old, err := DecodeHave([]byte(`{"v":1,"build":"build-17","phase":"running","health":"ok"}`)) + if err != nil || old.Server != "" || !SameHave(old, Have{V: Version, Build: "build-17", Phase: PhaseRunning, Health: HealthOK}) { + t.Errorf("an old have read %+v, %v", old, err) + } + keeper := HaveNode{Composes: ComposesOwn, Keeps: []string{KeepsBusServer}} + b, err := EncodeHaveNode(keeper) + if err != nil || !strings.Contains(string(b), `"keeps":["bus-server"]`) { + t.Fatalf("the keeper's have is %s, %v", b, err) + } + back, err := DecodeHaveNode(b) + keeper.V = Version + if err != nil || !SameHaveNode(back, keeper) || SameHaveNode(back, HaveNode{V: Version, Composes: ComposesOwn}) { + t.Errorf("the keeper came back %+v, %v", back, err) + } + if older, err := DecodeHaveNode([]byte(`{"v":1,"composes":"own"}`)); err != nil || older.Keeps != nil { + t.Errorf("an older node's have read %+v, %v", older, err) + } + if _, err := EncodeHaveNode(HaveNode{Keeps: []string{""}}); err == nil { + t.Error("a node keeping an empty name was encoded") + } + }) t.Run("have node", func(t *testing.T) { in := HaveNode{Windows: json.RawMessage(`{"a":1}`), Busy: json.RawMessage(`true`)} b, err := EncodeHaveNode(in)