Add the mesh-delivery-view process, its manifest and the nox-delivery alias (issues 1134, 1137)

The process watches the three buckets on its own read-only credential; the manifest declares reads only, no state and no emits; the zsh item carries the alias as mesh-cli's does.
2026-10-18 19:31:50 +02:00
parent 1b6763de0d
commit eb8966c4b7
6 changed files with 617 additions and 1 deletions
@@ -0,0 +1,124 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-view/internal/view"
+)
+
+// TestProcessWholeSeesTheBuckets starts the program the way the node does (its entry, run, with its environment and a
+// broker file) against an embedded bus holding the three buckets, and asks its socket.
+func TestProcessWholeSeesTheBuckets(t *testing.T) { // 1134/AC8
+	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Start()
+	if !srv.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the bus did not come up")
+	}
+	t.Cleanup(srv.Shutdown)
+
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	kvs := map[string]jetstream.KeyValue{}
+	for _, b := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket} {
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b, History: 5})
+		if err != nil {
+			t.Fatal(err)
+		}
+		kvs[b] = kv
+	}
+	put := func(bucket, key string, value []byte) {
+		t.Helper()
+		if _, err := kvs[bucket].Put(ctx, key, value); err != nil {
+			t.Fatal(err)
+		}
+	}
+	haveValue := func(build string, phase delivery.Phase, health delivery.Health) []byte {
+		b, err := delivery.EncodeHave(delivery.Have{Build: build, Phase: phase, Health: health})
+		if err != nil {
+			t.Fatal(err)
+		}
+		return b
+	}
+	put(view.AssignmentsBucket, "a.x", []byte(`{"node":"a","module":"x","build":"b1"}`))
+	put(view.AssignmentsBucket, "a.y", []byte(`{"node":"a","module":"y","build":"b2"}`))
+	put(view.AssignmentsBucket, "b.x", []byte(`{"node":"b","module":"x","build":"b1","ended":"2026-10-18T11:30:00Z"}`))
+	put(view.BuildsBucket, "x.build.b1", []byte(`{"id":"b1","module":"x"}`))
+	put(view.BuildsBucket, "x.newest", []byte("b1"))
+	put(delivery.HaveBucket, "a.x", haveValue("b1", delivery.PhaseChecking, delivery.HealthWaits))
+	put(delivery.HaveBucket, "b.x", haveValue("b1", delivery.PhaseRunning, delivery.HealthOK))
+
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	broker := filepath.Join(dir, "broker")
+	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	socket := filepath.Join(dir, "run", "cli.sock")
+	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": socket, "MESH_DELIVERY_WEB": "127.0.0.1:0"}
+	runCtx, stop := context.WithCancel(ctx)
+	done := make(chan error, 1)
+	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
+	t.Cleanup(func() { stop(); <-done })
+
+	rowsMatch := func(check func(view.Snapshot) bool) view.Snapshot {
+		t.Helper()
+		var last view.Snapshot
+		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
+			if s, err := view.Ask(socket); err == nil {
+				last = s
+				if check(s) {
+					return s
+				}
+			}
+		}
+		t.Fatalf("the socket never showed the expected rows; last: %+v", last)
+		return last
+	}
+	byKey := func(s view.Snapshot, node, module string) view.Row {
+		for _, r := range s.Rows {
+			if r.Node == node && r.Module == module {
+				return r
+			}
+		}
+		return view.Row{}
+	}
+	s := rowsMatch(func(s view.Snapshot) bool {
+		return byKey(s, "a", "x").Health == "waits" && byKey(s, "a", "y").Waits == "no have yet" && byKey(s, "b", "x").Status == view.StatusLeftBehind
+	})
+	if s.Counted() != 2 {
+		t.Fatalf("counted = %d, want 2 (a.x, a.y): %+v", s.Counted(), s.Rows)
+	}
+
+	put(delivery.HaveBucket, "a.x", haveValue("b1", delivery.PhaseRunning, delivery.HealthOK))
+	rowsMatch(func(s view.Snapshot) bool { r := byKey(s, "a", "x"); return r.Phase == "running" && r.Health == "ok" })
+}
@@ -0,0 +1,194 @@
+// Command mesh-delivery-view is the operator's view of delivery (novox/hq ADR 0429 §3): it watches the have, assignments
+// and builds buckets on its own bus credential, derives one row per module and node in memory, and serves them on a
+// local socket (for mesh-delivery-cli) and on a local web page. It writes nothing and starts, stops or retries nothing.
+package main
+
+import (
+	"context"
+	"crypto/sha256"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+	"time"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"mesh-delivery-view/internal/view"
+)
+
+const (
+	defaultSocket = "/run/mesh-delivery-view/cli.sock"
+	defaultWeb    = "127.0.0.1:8471"
+)
+
+func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+	if err := run(ctx, os.Getenv, os.Stderr); err != nil {
+		fmt.Fprintln(os.Stderr, "mesh-delivery-view:", err)
+		os.Exit(1)
+	}
+}
+
+// credential is the broker file the controller issues to the module: JSON of the bus address, the certificate
+// fingerprint to pin, the principal and its password.
+type credential struct {
+	URL         string `json:"url"`
+	Fingerprint string `json:"fingerprint"`
+	Node        string `json:"node"`
+	Module      string `json:"module"`
+	User        string `json:"user"`
+	Password    string `json:"password"`
+}
+
+func readCredential(path string) (credential, error) {
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return credential{}, err
+	}
+	var c credential
+	if err := json.Unmarshal(raw, &c); err != nil {
+		return credential{}, fmt.Errorf("the broker credential is not one: %w", err)
+	}
+	if c.URL == "" || c.User == "" {
+		return credential{}, errors.New("the broker credential names no bus address or user")
+	}
+	return c, nil
+}
+
+// pinned accepts exactly the certificate with this SHA-256 and no other.
+func pinned(want string) *tls.Config {
+	return &tls.Config{
+		InsecureSkipVerify: true, //nolint:gosec // replaced by the pin below, which is stricter
+		MinVersion:         tls.VersionTLS12,
+		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
+			if len(rawCerts) == 0 {
+				return errors.New("the bus presented no certificate")
+			}
+			sum := sha256.Sum256(rawCerts[0])
+			if got := "sha256:" + hex.EncodeToString(sum[:]); got != want {
+				return fmt.Errorf("the bus presented a certificate this mesh does not know (%.23s…)", got)
+			}
+			return nil
+		},
+	}
+}
+
+// run is the program: it returns when ctx ends or a listener cannot be made.
+func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
+	logger := log.New(logOut, "mesh-delivery-view: ", log.LstdFlags)
+	path := strings.TrimSpace(getenv("MESH_BROKER_FILE"))
+	if path == "" {
+		return errors.New("MESH_BROKER_FILE is not set, so there is no credential to watch the buckets with")
+	}
+	cred, err := readCredential(path)
+	if err != nil {
+		return err
+	}
+	socket := strings.TrimSpace(getenv("MESH_DELIVERY_SOCKET"))
+	if socket == "" {
+		socket = defaultSocket
+	}
+	web := strings.TrimSpace(getenv("MESH_DELIVERY_WEB"))
+	if web == "" {
+		web = defaultWeb
+	}
+
+	model := view.New()
+	l, err := view.ListenSocket(socket)
+	if err != nil {
+		return fmt.Errorf("cannot listen on %s: %w", socket, err)
+	}
+	srv := &http.Server{Addr: web, Handler: view.Web(model), ReadHeaderTimeout: 10 * time.Second}
+	webUp := make(chan error, 1)
+	go func() { webUp <- srv.ListenAndServe() }()
+	go view.ServeSocket(ctx, l, model)
+
+	opts := []nats.Option{
+		nats.Name(cred.Node + "." + cred.Module),
+		nats.UserInfo(cred.User, cred.Password),
+		nats.CustomInboxPrefix("_INBOX." + cred.User),
+		nats.MaxReconnects(-1),
+		nats.RetryOnFailedConnect(true),
+	}
+	if strings.TrimSpace(cred.Fingerprint) != "" {
+		opts = append(opts, nats.Secure(pinned(cred.Fingerprint)))
+	}
+	conn, err := nats.Connect(cred.URL, opts...)
+	if err != nil {
+		return fmt.Errorf("connecting to the bus: %w", err)
+	}
+	defer conn.Close()
+	js, err := jetstream.New(conn)
+	if err != nil {
+		return err
+	}
+	for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket} {
+		go follow(ctx, js, bucket, model, logger)
+	}
+	select {
+	case <-ctx.Done():
+	case err := <-webUp:
+		if !errors.Is(err, http.ErrServerClosed) {
+			return fmt.Errorf("the web view stopped: %w", err)
+		}
+	}
+	shutdown, cancel := context.WithTimeout(context.Background(), 3*time.Second)
+	defer cancel()
+	srv.Shutdown(shutdown)
+	return nil
+}
+
+// follow watches a bucket with its last values first and then each change, for as long as ctx lives: a bucket that is
+// not there yet, or a watch that ends, is tried again.
+func follow(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model, logger *log.Logger) {
+	for ctx.Err() == nil {
+		if err := watchOnce(ctx, js, bucket, model); err != nil && ctx.Err() == nil {
+			logger.Printf("cannot watch %s, trying again: %v", bucket, err)
+		}
+		select {
+		case <-ctx.Done():
+		case <-time.After(5 * time.Second):
+		}
+	}
+}
+
+func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model *view.Model) error {
+	kv, err := js.KeyValue(ctx, bucket)
+	if err != nil {
+		return err
+	}
+	w, err := kv.WatchAll(ctx)
+	if err != nil {
+		return err
+	}
+	defer w.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case e, ok := <-w.Updates():
+			if !ok {
+				return errors.New("the watch ended")
+			}
+			if e == nil { // the last values have all been delivered
+				continue
+			}
+			model.Apply(view.Entry{Bucket: bucket, Key: e.Key(), Value: e.Value(), At: e.Created(),
+				Deleted: e.Operation() != jetstream.KeyValuePut})
+		}
+	}
+}
@@ -0,0 +1,34 @@
+package main
+
+import (
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestCredentialIsTheBrokerFile(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "broker")
+	os.WriteFile(path, []byte(`{"url":"nats://127.0.0.1:4222","fingerprint":"sha256:ab","node":"n","module":"mesh-delivery-view","user":"u","password":"p"}`), 0o600)
+	c, err := readCredential(path)
+	if err != nil || c.URL != "nats://127.0.0.1:4222" || c.User != "u" || c.Password != "p" || c.Fingerprint != "sha256:ab" {
+		t.Fatalf("credential = %+v, %v", c, err)
+	}
+	os.WriteFile(path, []byte(`{"url":"x"}`), 0o600)
+	if _, err := readCredential(path); err == nil {
+		t.Fatal("a credential naming no user must be refused")
+	}
+	os.WriteFile(path, []byte(`not json`), 0o600)
+	if _, err := readCredential(path); err == nil {
+		t.Fatal("a file that is not a credential must be refused")
+	}
+}
+
+func TestRunNeedsItsCredential(t *testing.T) {
+	err := run(context.Background(), func(string) string { return "" }, io.Discard)
+	if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {
+		t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err)
+	}
+}
@@ -1,2 +1 @@
-# The operator's command (module mesh-delivery-view, novox/hq ADR 0429 §3): nox-delivery is mesh-delivery-cli, in interactive zsh only.
 alias nox-delivery=mesh-delivery-cli
@@ -0,0 +1,177 @@
+package manifest
+
+import (
+	"bytes"
+	"encoding/json"
+	"os"
+	"reflect"
+	"slices"
+	"strings"
+	"testing"
+)
+
+// The controller's manifest reader is mesh-controller internal/catalogue (ParseManifest), an internal package of another
+// repository, which this module cannot import. These tests therefore decode the manifest strictly into the fields the
+// controller's Manifest, Build and Artifact types name (internal/catalogue/manifest.go:475, :602, :659, :667, :737,
+// :809, :821, :892, :1010, :1065-:1090), as its parseManifest does (DisallowUnknownFields, manifest.go:1744), and check
+// the rules the issue names.
+
+type artifact struct {
+	Name     string `json:"name"`
+	Kind     string `json:"kind"`
+	Language string `json:"language"`
+	System   string `json:"system"`
+	From     string `json:"from"`
+	Binary   string `json:"binary"`
+	Context  *struct {
+		Repository string `json:"repository"`
+	} `json:"context"`
+}
+
+type item struct {
+	Collection string `json:"collection"`
+	Template   string `json:"template"`
+}
+
+type manifest struct {
+	Module       string            `json:"module"`
+	Version      string            `json:"version"`
+	Slug         string            `json:"slug"`
+	Reads        []string          `json:"reads"`
+	OwnSecrets   map[string]string `json:"own-secrets"`
+	SecretsOwner string            `json:"secrets-owner"`
+	Resources    []map[string]any  `json:"resources"`
+	Items        []item            `json:"items"`
+	Build        struct {
+		Artifacts []artifact `json:"artifacts"`
+	} `json:"build"`
+}
+
+func load(t *testing.T) (manifest, map[string]json.RawMessage) {
+	t.Helper()
+	raw, err := os.ReadFile("module.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var m manifest
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.DisallowUnknownFields()
+	if err := dec.Decode(&m); err != nil {
+		t.Fatalf("the controller's reader refuses a key this manifest has: %v", err)
+	}
+	var top map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &top); err != nil {
+		t.Fatal(err)
+	}
+	return m, top
+}
+
+func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
+	m, top := load(t)
+	if m.Module != "mesh-delivery-view" {
+		t.Errorf("module = %q", m.Module)
+	}
+	want := []string{"mesh-controller.assignments", "mesh-controller.builds", "node-engine.have"}
+	got := slices.Clone(m.Reads)
+	slices.Sort(got)
+	if !reflect.DeepEqual(got, want) {
+		t.Errorf("reads = %v, want exactly %v", got, want)
+	}
+	for _, k := range []string{"state", "emits", "consumes", "provides", "claims"} {
+		if _, ok := top[k]; ok {
+			t.Errorf("the manifest declares %q: the view writes nothing", k)
+		}
+	}
+	if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
+		t.Errorf("own-secrets = %v, want exactly broker (or no credential is issued)", m.OwnSecrets)
+	}
+	if m.SecretsOwner != "mesh-delivery-view" {
+		t.Errorf("secrets-owner = %q", m.SecretsOwner)
+	}
+}
+
+func TestManifestIsWholeInItsRepository(t *testing.T) { // 1134/AC7
+	m, _ := load(t)
+	if len(m.Build.Artifacts) != 2 {
+		t.Fatalf("artifacts = %+v", m.Build.Artifacts)
+	}
+	for _, a := range m.Build.Artifacts {
+		if a.Context != nil {
+			t.Errorf("artifact %s names a context of another repository", a.Name)
+		}
+		if a.Kind != "bundle" || a.Language != "go" || a.System != "arch" {
+			t.Errorf("artifact %s = %+v, want a go bundle for arch", a.Name, a)
+		}
+		if st, err := os.Stat(a.From); err != nil || !st.IsDir() {
+			t.Errorf("artifact %s builds from %q, which is not a directory of this module", a.Name, a.From)
+		}
+		if a.Binary == "" || !strings.HasSuffix(a.From, a.Binary) {
+			t.Errorf("artifact %s binary %q does not match %q", a.Name, a.Binary, a.From)
+		}
+	}
+	names := map[string]string{}
+	for _, a := range m.Build.Artifacts {
+		names[a.Name] = a.Binary
+	}
+	if names["program"] != "mesh-delivery-view" || names["cli"] != "mesh-delivery-cli" {
+		t.Errorf("artifacts = %v", names)
+	}
+}
+
+func TestManifestResources(t *testing.T) { // 1134/AC7
+	m, _ := load(t)
+	byType := map[string][]map[string]any{}
+	for _, r := range m.Resources {
+		byType[r["type"].(string)] = append(byType[r["type"].(string)], r)
+	}
+	for _, ty := range []string{"user", "directory", "process", "archive"} {
+		if len(byType[ty]) == 0 {
+			t.Fatalf("no %s resource", ty)
+		}
+	}
+	user := byType["user"][0]["name"]
+	if user != "mesh-delivery-view" {
+		t.Errorf("user = %v", user)
+	}
+	p := byType["process"][0]
+	if p["user"] != user || p["artifact"] != "program" {
+		t.Errorf("process = %v", p)
+	}
+	env, _ := p["env"].(map[string]any)
+	if env["MESH_BROKER_FILE"] != "${dir:state}/broker" || m.OwnSecrets["broker"] != "${dir:state}/broker" {
+		t.Errorf("the process reads %v, the own-secret is written at %v", env["MESH_BROKER_FILE"], m.OwnSecrets["broker"])
+	}
+	if h, _ := p["health"].(map[string]any); h["kind"] != "unit" {
+		t.Errorf("health = %v", p["health"])
+	}
+	var state, run bool
+	for _, d := range byType["directory"] {
+		if d["id"] == "state" && d["mode"] == "0700" && d["owner"] == user {
+			state = true
+		}
+		if d["path"] == "/run/mesh-delivery-view" && d["mode"] == "0755" && d["owner"] == user {
+			run = true
+		}
+	}
+	if !state || !run {
+		t.Errorf("state directory 0700 owned by the user: %v; run directory for the socket 0755: %v", state, run)
+	}
+	a := byType["archive"][0]
+	if a["artifact"] != "cli" || a["path"] != "/usr/local/bin" {
+		t.Errorf("archive = %v, want the cli unpacked into /usr/local/bin", a)
+	}
+}
+
+func TestAliasItem(t *testing.T) { // 1137/AC1
+	m, _ := load(t)
+	if !reflect.DeepEqual(m.Items, []item{{Collection: "shell:zsh:normal", Template: "files/zsh-normal"}}) {
+		t.Fatalf("items = %+v", m.Items)
+	}
+	raw, err := os.ReadFile("files/zsh-normal")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(raw) != "alias nox-delivery=mesh-delivery-cli\n" {
+		t.Fatalf("files/zsh-normal = %q, want the one line", raw)
+	}
+}
@@ -0,0 +1,88 @@
+{
+  "module": "mesh-delivery-view",
+  "version": "1",
+  "slug": "view",
+  "reads": [
+    "mesh-controller.assignments",
+    "mesh-controller.builds",
+    "node-engine.have"
+  ],
+  "own-secrets": {
+    "broker": "${dir:state}/broker"
+  },
+  "secrets-owner": "mesh-delivery-view",
+  "resources": [
+    {
+      "id": "account",
+      "type": "user",
+      "name": "mesh-delivery-view",
+      "shell": "/usr/bin/nologin",
+      "home": "/var/lib/mesh-delivery-view"
+    },
+    {
+      "id": "state",
+      "type": "directory",
+      "mode": "0700",
+      "owner": "mesh-delivery-view",
+      "place": ".",
+      "unknown": "program"
+    },
+    {
+      "id": "run",
+      "type": "directory",
+      "path": "/run/mesh-delivery-view",
+      "mode": "0755",
+      "owner": "mesh-delivery-view",
+      "unknown": "program"
+    },
+    {
+      "id": "view",
+      "type": "process",
+      "name": "mesh-delivery-view",
+      "artifact": "program",
+      "run": [
+        "./mesh-delivery-view"
+      ],
+      "env": {
+        "MESH_BROKER_FILE": "${dir:state}/broker"
+      },
+      "user": "mesh-delivery-view",
+      "health": {
+        "kind": "unit"
+      }
+    },
+    {
+      "id": "command",
+      "type": "archive",
+      "artifact": "cli",
+      "path": "/usr/local/bin",
+      "unknown": "program"
+    }
+  ],
+  "items": [
+    {
+      "collection": "shell:zsh:normal",
+      "template": "files/zsh-normal"
+    }
+  ],
+  "build": {
+    "artifacts": [
+      {
+        "name": "program",
+        "kind": "bundle",
+        "language": "go",
+        "system": "arch",
+        "from": "cmd/mesh-delivery-view",
+        "binary": "mesh-delivery-view"
+      },
+      {
+        "name": "cli",
+        "kind": "bundle",
+        "language": "go",
+        "system": "arch",
+        "from": "cmd/mesh-delivery-cli",
+        "binary": "mesh-delivery-cli"
+      }
+    ]
+  }
+}