The watchman module: manifest, program on the bus, and its whole-process integration test (issue 1135)

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go

@@ -0,0 +1,220 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"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-watchman/internal/watch"
+)
+
+// busConf is the composed accounts file's shape (mesh-controller internal/broker): the account MESH imports the system
+// account's connect and disconnect events and its connections request, which the system account exports to MESH alone.
+const busConf = `
+listen: 127.0.0.1:-1
+jetstream { store_dir: %q }
+accounts {
+  MESH {
+    jetstream: enabled
+    imports = [
+      { stream: { account: "$SYS", subject: "$SYS.ACCOUNT.MESH.CONNECT" }, to: "mesh.bus.connect" }
+      { stream: { account: "$SYS", subject: "$SYS.ACCOUNT.MESH.DISCONNECT" }, to: "mesh.bus.disconnect" }
+      { service: { account: "$SYS", subject: "$SYS.REQ.ACCOUNT.MESH.CONNZ" }, to: "mesh.bus.connz" }
+    ]
+    users = [
+      { user: "ctl", password: "pw" }
+      { user: "n1.mesh-delivery-watchman", password: "pw" }
+      { user: "node.n0", password: "pw" }
+      { user: "node.n1", password: "pw" }
+      { user: "node.n2", password: "pw" }
+    ]
+  }
+  "$SYS" {
+    exports = [
+      { stream: "$SYS.ACCOUNT.MESH.CONNECT", accounts: ["MESH"] }
+      { stream: "$SYS.ACCOUNT.MESH.DISCONNECT", accounts: ["MESH"] }
+      { service: "$SYS.REQ.ACCOUNT.MESH.CONNZ", accounts: ["MESH"] }
+    ]
+  }
+}
+`
+
+// TestWatchmanWhole starts the program the way the node does (its entry, run, with its environment and a broker file)
+// against an embedded bus with the system exports and JetStream, and proves the four things the issue names.
+func TestWatchmanWhole(t *testing.T) { // 1135/AC8
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "w")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	conf := filepath.Join(dir, "bus.conf")
+	if err := os.WriteFile(conf, []byte(fmt.Sprintf(busConf, filepath.Join(dir, "js"))), 0o600); err != nil {
+		t.Fatal(err)
+	}
+	opts, err := server.ProcessConfigFile(conf)
+	if err != nil {
+		t.Fatal(err)
+	}
+	opts.NoLog, opts.NoSigs = true, true
+	srv, err := server.NewServer(opts)
+	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)
+
+	connect := func(user string) *nats.Conn {
+		t.Helper()
+		nc, err := nats.Connect(srv.ClientURL(), nats.UserInfo(user, "pw"))
+		if err != nil {
+			t.Fatalf("%s cannot connect: %v", user, err)
+		}
+		t.Cleanup(nc.Close)
+		return nc
+	}
+	ctl := connect("ctl")
+	js, err := jetstream.New(ctl)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
+	defer cancel()
+	kvs := map[string]jetstream.KeyValue{}
+	for _, c := range []jetstream.KeyValueConfig{
+		{Bucket: delivery.PresenceBucket, History: 20},
+		{Bucket: delivery.StuckBucket, History: 20},
+		{Bucket: delivery.PulseBucket, LimitMarkerTTL: time.Second},
+		{Bucket: watch.BuildsBucket, History: 5},
+		{Bucket: watch.ChecksBucket, History: 5},
+	} {
+		kv, err := js.CreateKeyValue(ctx, c)
+		if err != nil {
+			t.Fatal(err)
+		}
+		kvs[c.Bucket] = kv
+	}
+
+	// A node already connected when the watchman starts: its presence comes from the one connections request.
+	connect("node.n0")
+
+	broker := filepath.Join(dir, "broker")
+	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n1", Module: "mesh-delivery-watchman", User: "n1.mesh-delivery-watchman", Password: "pw"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_PULSE_BOUND": "3s"}
+	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 })
+
+	has := func(bucket, key string) (jetstream.KeyValueEntry, bool) {
+		e, err := kvs[bucket].Get(ctx, key)
+		return e, err == nil
+	}
+	online := func(key string) (bool, bool) {
+		e, ok := has(delivery.PresenceBucket, key)
+		if !ok {
+			return false, false
+		}
+		p, err := delivery.DecodePresence(e.Value())
+		return p.Online, err == nil
+	}
+	until := func(what string, ok func() bool) {
+		t.Helper()
+		for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+			if ok() {
+				return
+			}
+		}
+		t.Fatalf("never saw: %s", what)
+	}
+	pulseOf := func(bucket, key string) string {
+		k, err := delivery.PulseKey(bucket, key)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return k
+	}
+	stuckOf := func(bucket, key string) string {
+		k, err := delivery.StuckKey(bucket, key)
+		if err != nil {
+			t.Fatal(err)
+		}
+		return k
+	}
+
+	until("node n0, connected before the start, online", func() bool { on, ok := online("n0"); return ok && on })
+
+	// A connection opening and closing turns presence online and offline.
+	n1 := connect("node.n1")
+	until("node n1 online", func() bool { on, ok := online("n1"); return ok && on })
+	n1.Close()
+	until("node n1 offline", func() bool { on, ok := online("n1"); return ok && !on })
+
+	// A build asked and unanswered gets a pulse; let to expire, it writes a stuck key.
+	now := time.Now().UTC()
+	asked, _ := json.Marshal(map[string]any{"id": "b1", "module": "x", "state": "asked", "asked": now})
+	if _, err := kvs[watch.BuildsBucket].Put(ctx, "x.build.b1", asked); err != nil {
+		t.Fatal(err)
+	}
+	until("a pulse for the build asked", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(watch.BuildsBucket, "x.build.b1")); return ok })
+	var stuck delivery.Stuck
+	until("a stuck key for the build never answered", func() bool {
+		e, ok := has(delivery.StuckBucket, stuckOf(watch.BuildsBucket, "x.build.b1"))
+		if !ok {
+			return false
+		}
+		var err error
+		stuck, err = delivery.DecodeStuck(e.Value())
+		return err == nil
+	})
+	if stuck.Moved || stuck.Waited != "the build of x to be answered" || stuck.Bucket != watch.BuildsBucket || stuck.Key != "x.build.b1" {
+		t.Fatalf("stuck = %+v", stuck)
+	}
+	// The node n1, gone offline and never back, is stuck as well.
+	until("a stuck key for node n1 that did not come back", func() bool {
+		_, ok := has(delivery.StuckBucket, stuckOf(delivery.PresenceBucket, "n1"))
+		return ok
+	})
+
+	// A presence back online before the bound deletes its pulse and writes no stuck key.
+	n2 := connect("node.n2")
+	until("node n2 online", func() bool { on, ok := online("n2"); return ok && on })
+	n2.Close()
+	until("a pulse for node n2 gone offline", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(delivery.PresenceBucket, "n2")); return ok })
+	connect("node.n2")
+	until("node n2 online again", func() bool { on, ok := online("n2"); return ok && on })
+	until("the pulse of node n2 deleted", func() bool { _, ok := has(delivery.PulseBucket, pulseOf(delivery.PresenceBucket, "n2")); return !ok })
+	// A build answered at once is the same: no pulse, no stuck key, however long the bound passes.
+	answered, _ := json.Marshal(map[string]any{"id": "b2", "module": "x", "state": "asked", "asked": time.Now().UTC()})
+	if _, err := kvs[watch.BuildsBucket].Put(ctx, "x.build.b2", answered); err != nil {
+		t.Fatal(err)
+	}
+	done2, _ := json.Marshal(map[string]any{"id": "b2", "module": "x"})
+	if _, err := kvs[watch.BuildsBucket].Put(ctx, "x.build.b2", done2); err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(6 * time.Second) // twice the bound and the marker time
+	for _, k := range []string{stuckOf(delivery.PresenceBucket, "n2"), stuckOf(watch.BuildsBucket, "x.build.b2")} {
+		if _, ok := has(delivery.StuckBucket, k); ok {
+			t.Errorf("a stuck key %s was written for a move that happened before the bound", k)
+		}
+	}
+}

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go

@@ -0,0 +1,355 @@
+// Command mesh-delivery-watchman is delivery's watchman (novox/hq ADR 0429 §1, ADR 0431, ADR 0433, issue 1135): it
+// hears the bus's connect and disconnect events and writes presence, mirrors every move a machine owes (a build asked
+// and not answered, a check running, a node or part gone offline) into the pulse bucket with a time to live, and writes
+// a stuck key when the server expires one. It holds no timer: every reaction is an event, a watch or a request at
+// start or at a reconnect, and the server's expiry is the only clock.
+package main
+
+import (
+	"context"
+	"crypto/sha256"
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"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-watchman/internal/watch"
+)
+
+// The subjects the controller's broker imports the bus's own events and connections request under (mesh-controller
+// internal/broker BusConnectSubject, BusDisconnectSubject, BusConnzSubject).
+const (
+	connectSubject    = "mesh.bus.connect"
+	disconnectSubject = "mesh.bus.disconnect"
+	connzSubject      = "mesh.bus.connz"
+)
+
+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-watchman:", 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
+		},
+	}
+}
+
+// kvOut writes the watchman's three buckets.
+type kvOut struct {
+	ctx                   context.Context
+	presence, pulse, stuk jetstream.KeyValue
+}
+
+func (o *kvOut) PutPresence(key string, online bool) error {
+	value, err := delivery.EncodePresence(delivery.Presence{Online: online})
+	if err != nil {
+		return err
+	}
+	_, err = o.presence.Put(o.ctx, key, value)
+	return err
+}
+
+// PutPulse creates the key with its own time to live; a live key is deleted first, because a time to live is set only
+// when a key is created.
+func (o *kvOut) PutPulse(key string, value []byte, ttl time.Duration) error {
+	_, err := o.pulse.Create(o.ctx, key, value, jetstream.KeyTTL(ttl))
+	if errors.Is(err, jetstream.ErrKeyExists) {
+		if err := o.pulse.Delete(o.ctx, key); err != nil {
+			return err
+		}
+		_, err = o.pulse.Create(o.ctx, key, value, jetstream.KeyTTL(ttl))
+	}
+	return err
+}
+
+func (o *kvOut) DeletePulse(key string) error { return o.pulse.Delete(o.ctx, key) }
+
+func (o *kvOut) PutStuck(key string, value []byte) error {
+	_, err := o.stuk.Put(o.ctx, key, value)
+	return err
+}
+
+func opOf(e jetstream.KeyValueEntry) watch.Op {
+	switch e.Operation() {
+	case jetstream.KeyValueDelete:
+		return watch.Delete
+	case jetstream.KeyValuePurge:
+		return watch.Expired
+	}
+	return watch.Put
+}
+
+// settle reads the entries a watcher holds now, up to its mark of the last values, and returns the watcher for the
+// changes that follow.
+func settle(ctx context.Context, kv jetstream.KeyValue, each func(jetstream.KeyValueEntry)) (jetstream.KeyWatcher, error) {
+	w, err := kv.WatchAll(ctx)
+	if err != nil {
+		return nil, err
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			w.Stop()
+			return nil, ctx.Err()
+		case e, ok := <-w.Updates():
+			if !ok {
+				return nil, errors.New("the watch ended")
+			}
+			if e == nil {
+				return w, nil
+			}
+			each(e)
+		}
+	}
+}
+
+// follow hands each change of a watcher to each until ctx ends; a watch that ends is an error, and the process
+// manager starts the program again.
+func follow(ctx context.Context, w jetstream.KeyWatcher, each func(jetstream.KeyValueEntry)) error {
+	defer w.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case e, ok := <-w.Updates():
+			if !ok {
+				return errors.New("a watch ended")
+			}
+			if e != nil {
+				each(e)
+			}
+		}
+	}
+}
+
+// run is the program: it returns when ctx ends or the bus cannot be used.
+func run(ctx context.Context, getenv func(string) string, logOut io.Writer) error {
+	logger := log.New(logOut, "mesh-delivery-watchman: ", 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 bus with")
+	}
+	cred, err := readCredential(path)
+	if err != nil {
+		return err
+	}
+	again := make(chan struct{}, 1)
+	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),
+		nats.ReconnectHandler(func(*nats.Conn) {
+			select {
+			case again <- struct{}{}:
+			default:
+			}
+		}),
+	}
+	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
+	}
+	bucket := func(name string) (jetstream.KeyValue, error) {
+		kv, err := js.KeyValue(ctx, name)
+		if err != nil {
+			return nil, fmt.Errorf("the bucket %s: %w", name, err)
+		}
+		return kv, nil
+	}
+	var kvs struct{ presence, pulse, stuck, builds, checks jetstream.KeyValue }
+	for _, b := range []struct {
+		name string
+		into *jetstream.KeyValue
+	}{{delivery.PresenceBucket, &kvs.presence}, {delivery.PulseBucket, &kvs.pulse}, {delivery.StuckBucket, &kvs.stuck},
+		{watch.BuildsBucket, &kvs.builds}, {watch.ChecksBucket, &kvs.checks}} {
+		if *b.into, err = bucket(b.name); err != nil {
+			return err
+		}
+	}
+
+	w := watch.New(&kvOut{ctx: ctx, presence: kvs.presence, pulse: kvs.pulse, stuk: kvs.stuck}, time.Now)
+	w.Logf(logger.Printf)
+	// A test shortens how long a move may take; the mesh never sets it.
+	if v := strings.TrimSpace(getenv("MESH_DELIVERY_PULSE_BOUND")); v != "" {
+		d, err := time.ParseDuration(v)
+		if err != nil || d <= 0 {
+			return fmt.Errorf("MESH_DELIVERY_PULSE_BOUND %q is not a duration", v)
+		}
+		w.SetBound(d)
+	}
+
+	// What stands now: presence and stuck as written, pulses as the server holds them, then the builds and checks.
+	for _, s := range []struct {
+		kv   jetstream.KeyValue
+		each func(jetstream.KeyValueEntry)
+	}{
+		{kvs.presence, func(e jetstream.KeyValueEntry) {
+			if p, err := delivery.DecodePresence(e.Value()); err == nil && opOf(e) == watch.Put {
+				w.SeedPresence(e.Key(), p.Online, e.Created())
+			}
+		}},
+		{kvs.stuck, func(e jetstream.KeyValueEntry) {
+			if opOf(e) == watch.Put {
+				w.SeedStuck(e.Key(), e.Value())
+			}
+		}},
+	} {
+		sw, err := settle(ctx, s.kv, s.each)
+		if err != nil {
+			return err
+		}
+		sw.Stop()
+	}
+	pulses, err := settle(ctx, kvs.pulse, func(e jetstream.KeyValueEntry) {
+		if opOf(e) == watch.Put {
+			w.SeedPulse(e.Key())
+		}
+	})
+	if err != nil {
+		return err
+	}
+	builds, err := settle(ctx, kvs.builds, func(e jetstream.KeyValueEntry) {
+		w.Build(e.Key(), e.Value(), e.Created(), opOf(e))
+	})
+	if err != nil {
+		return err
+	}
+	checks, err := settle(ctx, kvs.checks, func(e jetstream.KeyValueEntry) {
+		w.Check(e.Key(), e.Value(), e.Created(), opOf(e))
+	})
+	if err != nil {
+		return err
+	}
+
+	// The events first, then the one request of the open connections, so no connection falls between them.
+	onEvent := func(open bool) nats.MsgHandler {
+		return func(m *nats.Msg) {
+			user, err := watch.ParseEvent(m.Data)
+			if err != nil {
+				logger.Print(err)
+				return
+			}
+			if open {
+				w.Connect(user)
+			} else {
+				w.Disconnect(user)
+			}
+		}
+	}
+	for subject, open := range map[string]bool{connectSubject: true, disconnectSubject: false} {
+		if _, err := conn.Subscribe(subject, onEvent(open)); err != nil {
+			return err
+		}
+	}
+	if err := conn.Flush(); err != nil {
+		return err
+	}
+	ask := func(body []byte) ([]byte, error) {
+		m, err := conn.RequestWithContext(ctx, connzSubject, body)
+		if err != nil {
+			return nil, err
+		}
+		return m.Data, nil
+	}
+	if err := w.Rebuild(ask); err != nil {
+		return err
+	}
+	w.Reconcile()
+
+	errs := make(chan error, 4)
+	go func() {
+		errs <- follow(ctx, pulses, func(e jetstream.KeyValueEntry) { w.Pulse(e.Key(), opOf(e), e.Created()) })
+	}()
+	go func() {
+		errs <- follow(ctx, builds, func(e jetstream.KeyValueEntry) { w.Build(e.Key(), e.Value(), e.Created(), opOf(e)) })
+	}()
+	go func() {
+		errs <- follow(ctx, checks, func(e jetstream.KeyValueEntry) { w.Check(e.Key(), e.Value(), e.Created(), opOf(e)) })
+	}()
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				errs <- nil
+				return
+			case <-again:
+				if err := w.Rebuild(ask); err != nil {
+					logger.Printf("the open connections could not be read after a reconnect: %v", err)
+				}
+			}
+		}
+	}()
+	select {
+	case <-ctx.Done():
+		return nil
+	case err := <-errs:
+		return err
+	}
+}

modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go

@@ -0,0 +1,34 @@
+package main
+
+import (
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+func TestCredentialIsTheBrokerFile(t *testing.T) { // 1135/AC6
+	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-watchman","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) { // 1135/AC6
+	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)
+	}
+}

modules/mesh-delivery-watchman/manifest_test.go

@@ -0,0 +1,177 @@
+package manifest
+
+import (
+	"bytes"
+	"encoding/json"
+	"io/fs"
+	"os"
+	"path/filepath"
+	"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 its
+// Manifest type names (DisallowUnknownFields, as its parseManifest does) and check the rules the issue names.
+// **The merge gate's module check is what proves the controller takes this manifest (1135/AC6); these tests hold the
+// rules so a unit run catches a refusal early.**
+
+type state struct {
+	Name    string `json:"name"`
+	History int    `json:"history,omitempty"`
+	KeyTTL  bool   `json:"key-ttl,omitempty"`
+}
+
+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 manifest struct {
+	Module       string            `json:"module"`
+	Moved        bool              `json:"moved"`
+	Version      string            `json:"version"`
+	Slug         string            `json:"slug"`
+	Reads        []string          `json:"reads"`
+	State        []state           `json:"state"`
+	OwnSecrets   map[string]string `json:"own-secrets"`
+	SecretsOwner string            `json:"secrets-owner"`
+	Resources    []map[string]any  `json:"resources"`
+	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 TestManifestStateAndReads(t *testing.T) { // 1135/AC6
+	m, top := load(t)
+	if m.Module != "mesh-delivery-watchman" || !m.Moved || m.Slug != "watchman" {
+		t.Errorf("module %q moved %v slug %q", m.Module, m.Moved, m.Slug)
+	}
+	want := []state{{Name: "presence", History: 20}, {Name: "pulse", KeyTTL: true}, {Name: "stuck", History: 20}}
+	if !reflect.DeepEqual(m.State, want) {
+		t.Errorf("state = %+v, want %+v", m.State, want)
+	}
+	reads := slices.Clone(m.Reads)
+	slices.Sort(reads)
+	if !reflect.DeepEqual(reads, []string{"mesh-controller.builds", "mesh-controller.checks-running"}) {
+		t.Errorf("reads = %v", reads)
+	}
+	for _, k := range []string{"emits", "context", "items", "consumes", "provides", "claims"} {
+		if _, ok := top[k]; ok {
+			t.Errorf("the manifest declares %q: the watchman has none", k)
+		}
+	}
+	if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
+		t.Errorf("own-secrets = %v, want exactly broker", m.OwnSecrets)
+	}
+	if m.SecretsOwner != "mesh-delivery-watchman" {
+		t.Errorf("secrets-owner = %q", m.SecretsOwner)
+	}
+}
+
+func TestManifestIsWholeInItsRepository(t *testing.T) { // 1135/AC6
+	m, _ := load(t)
+	if len(m.Build.Artifacts) != 1 {
+		t.Fatalf("artifacts = %+v, want the one bundle program", m.Build.Artifacts)
+	}
+	a := m.Build.Artifacts[0]
+	if a.Name != "program" || a.Kind != "bundle" || a.Language != "go" || a.System != "arch" || a.Context != nil {
+		t.Errorf("artifact = %+v", a)
+	}
+	if st, err := os.Stat(a.From); err != nil || !st.IsDir() {
+		t.Errorf("the artifact builds from %q, which is not a directory of this module", a.From)
+	}
+	if a.Binary != "mesh-delivery-watchman" || !strings.HasSuffix(a.From, a.Binary) {
+		t.Errorf("binary %q from %q", a.Binary, a.From)
+	}
+}
+
+func TestManifestResources(t *testing.T) { // 1135/AC6
+	m, _ := load(t)
+	byType := map[string][]map[string]any{}
+	for _, r := range m.Resources {
+		ty, _ := r["type"].(string)
+		byType[ty] = append(byType[ty], r)
+	}
+	for _, ty := range []string{"user", "directory", "process"} {
+		if len(byType[ty]) == 0 {
+			t.Fatalf("no %s resource", ty)
+		}
+	}
+	user := byType["user"][0]["name"]
+	if user != "mesh-delivery-watchman" {
+		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" {
+		t.Errorf("the process reads %v", env["MESH_BROKER_FILE"])
+	}
+	if h, _ := p["health"].(map[string]any); h["kind"] != "unit" {
+		t.Errorf("health = %v", p["health"])
+	}
+	d := byType["directory"][0]
+	if d["id"] != "state" || d["mode"] != "0700" || d["owner"] != user {
+		t.Errorf("state directory = %v", d)
+	}
+}
+
+func TestNoTickerTimerOrSleep(t *testing.T) { // 1135/AC7
+	forbidden := []string{"time.Tick", "time.NewTicker", "time.Sleep", "time.After", "time.NewTimer", "time.AfterFunc"}
+	var files int
+	for _, root := range []string{"cmd", "internal"} {
+		err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
+			if err != nil || d.IsDir() || !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
+				return err
+			}
+			files++
+			src, err := os.ReadFile(path)
+			if err != nil {
+				return err
+			}
+			for _, f := range forbidden {
+				if strings.Contains(string(src), f) {
+					t.Errorf("%s holds %s: the program holds no timer", path, f)
+				}
+			}
+			return nil
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+	}
+	if files < 3 {
+		t.Fatalf("read %d source files, want the program's", files)
+	}
+}

modules/mesh-delivery-watchman/module.json

@@ -0,0 +1,64 @@
+{
+  "module": "mesh-delivery-watchman",
+  "version": "1",
+  "moved": true,
+  "slug": "watchman",
+  "reads": [
+    "mesh-controller.builds",
+    "mesh-controller.checks-running"
+  ],
+  "state": [
+    { "name": "presence", "history": 20 },
+    { "name": "pulse", "key-ttl": true },
+    { "name": "stuck", "history": 20 }
+  ],
+  "own-secrets": {
+    "broker": "${dir:state}/broker"
+  },
+  "secrets-owner": "mesh-delivery-watchman",
+  "resources": [
+    {
+      "id": "account",
+      "type": "user",
+      "name": "mesh-delivery-watchman",
+      "shell": "/usr/bin/nologin",
+      "home": "/var/lib/mesh-delivery-watchman"
+    },
+    {
+      "id": "state",
+      "type": "directory",
+      "mode": "0700",
+      "owner": "mesh-delivery-watchman",
+      "place": ".",
+      "unknown": "program"
+    },
+    {
+      "id": "watchman",
+      "type": "process",
+      "name": "mesh-delivery-watchman",
+      "artifact": "program",
+      "run": [
+        "./mesh-delivery-watchman"
+      ],
+      "env": {
+        "MESH_BROKER_FILE": "${dir:state}/broker"
+      },
+      "user": "mesh-delivery-watchman",
+      "health": {
+        "kind": "unit"
+      }
+    }
+  ],
+  "build": {
+    "artifacts": [
+      {
+        "name": "program",
+        "kind": "bundle",
+        "language": "go",
+        "system": "arch",
+        "from": "cmd/mesh-delivery-watchman",
+        "binary": "mesh-delivery-watchman"
+      }
+    ]
+  }
+}