mesh-sdk names what the bus server holds in one list, for the assigner and the node agent (ADR 0594 §7)

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)
+		}
+	})
 }

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"
+      }
+    ]
+  }
+}

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 {

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")
 	}

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)

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.

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)