From eb8966c4b7184e390bb30dd68abd27546c98a337 Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 19:31:50 +0200 Subject: [PATCH] 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. --- .../mesh-delivery-view/integration_test.go | 124 +++++++++++ .../cmd/mesh-delivery-view/main.go | 194 ++++++++++++++++++ .../cmd/mesh-delivery-view/main_test.go | 34 +++ modules/mesh-delivery-view/files/zsh-normal | 1 - modules/mesh-delivery-view/manifest_test.go | 177 ++++++++++++++++ modules/mesh-delivery-view/module.json | 88 ++++++++ 6 files changed, 617 insertions(+), 1 deletion(-) create mode 100644 modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go create mode 100644 modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go create mode 100644 modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go create mode 100644 modules/mesh-delivery-view/manifest_test.go create mode 100644 modules/mesh-delivery-view/module.json diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go new file mode 100644 index 0000000..cb32676 --- /dev/null +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/integration_test.go @@ -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" }) +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go new file mode 100644 index 0000000..040a7ca --- /dev/null +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main.go @@ -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}) + } + } +} diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go new file mode 100644 index 0000000..f536aab --- /dev/null +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view/main_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-view/files/zsh-normal b/modules/mesh-delivery-view/files/zsh-normal index da079e8..40bf85e 100644 --- a/modules/mesh-delivery-view/files/zsh-normal +++ b/modules/mesh-delivery-view/files/zsh-normal @@ -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 diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go new file mode 100644 index 0000000..13f83cc --- /dev/null +++ b/modules/mesh-delivery-view/manifest_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json new file mode 100644 index 0000000..c74f080 --- /dev/null +++ b/modules/mesh-delivery-view/module.json @@ -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" + } + ] + } +}