From 1191a10be17d5189e0dee59c8152550e309cefbc Mon Sep 17 00:00:00 2001 From: jochen Date: Sun, 18 Oct 2026 20:58:34 +0200 Subject: [PATCH] The watchman module: manifest, program on the bus, and its whole-process integration test (issue 1135) --- .../integration_test.go | 220 +++++++++++ .../cmd/mesh-delivery-watchman/main.go | 355 ++++++++++++++++++ .../cmd/mesh-delivery-watchman/main_test.go | 34 ++ .../mesh-delivery-watchman/manifest_test.go | 177 +++++++++ modules/mesh-delivery-watchman/module.json | 64 ++++ 5 files changed, 850 insertions(+) create mode 100644 modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go create mode 100644 modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go create mode 100644 modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go create mode 100644 modules/mesh-delivery-watchman/manifest_test.go create mode 100644 modules/mesh-delivery-watchman/module.json diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/integration_test.go new file mode 100644 index 0000000..f2758c6 --- /dev/null +++ b/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) + } + } +} diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go new file mode 100644 index 0000000..6eab4f9 --- /dev/null +++ b/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 + } +} diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main_test.go new file mode 100644 index 0000000..6a171c2 --- /dev/null +++ b/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) + } +} diff --git a/modules/mesh-delivery-watchman/manifest_test.go b/modules/mesh-delivery-watchman/manifest_test.go new file mode 100644 index 0000000..67e7342 --- /dev/null +++ b/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) + } +} diff --git a/modules/mesh-delivery-watchman/module.json b/modules/mesh-delivery-watchman/module.json new file mode 100644 index 0000000..f3d8ec2 --- /dev/null +++ b/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" + } + ] + } +}