Make the module mesh-delivery-assigner: its manifest's reads and states, its own credential and the states it announces, so the controller grants it what R5 reads (issue 1297, subtask B2a; ADR 0528, ADR 0429 §1)

2026-10-23 15:19:06 +02:00
parent 4ffe5f4d52
commit 034a6de9a7
6 changed files with 349 additions and 0 deletions
@@ -0,0 +1,127 @@
+// Command mesh-delivery-assigner is delivery's assigner (novox/hq ADR 0429 §1 and §6, ADR 0446, ADR 0528, issue 1297): a
+// process on its own bus credential that watches what the indexer, the builder, the controller, the node-engines and
+// the watchman write, and writes what each node should run (`want.<node>.<module>`, by the six rules), the
+// fingerprints it holds, every difference between a want and what the controller sends, and the composition check's
+// verdict of each group the checker judges. In release 5 the wants are only compared: the controller still sends what
+// each node runs (ADR 0429 §7). It holds no timer: every pass is started by a change it watched.
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"log"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/announce"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
+	"git.novox.be/novox/mesh-sdk/go/provider"
+)
+
+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-assigner:", err)
+		os.Exit(1)
+	}
+}
+
+// readBuckets are the buckets of other parts the assigner reads, each named in its manifest's reads (the manifest test
+// holds the two lists equal), so the controller's broker grants exactly these.
+var readBuckets = []string{
+	provider.AssignmentsBucket,
+	delivery.BuildsBucket,
+	delivery.ControllerGrantsBucket,
+	delivery.BuilderBuildsBucket,
+	delivery.CandidatesBucket,
+	delivery.ManifestsBucket,
+	delivery.ModulesBucket,
+	delivery.PresenceBucket,
+	delivery.HaveBucket,
+}
+
+// ownBuckets are the assigner's own states, as its manifest declares them.
+var ownBuckets = []string{delivery.AssignerCompositionsBucket, delivery.AssignerDifferencesBucket, delivery.AssignerWantsBucket}
+
+// keptStates is what the assigner keeps and who may read it (ADR 0495 §2-§3): its wants by the watchman and the view
+// (and the node agent from release 6), its differences by the view, its compositions by the checker.
+var keptStates = []announce.State{
+	{Name: "compositions", Readable: true},
+	{Name: "differences", Readable: true},
+	{Name: "wants", Readable: true},
+}
+
+// buckets is every bucket the assigner opened, by name.
+type buckets map[string]jetstream.KeyValue
+
+// open opens every bucket the assigner reads and writes; a bucket it cannot open stops the start, named, since a pass
+// without it would write wants from half the mesh.
+func open(ctx context.Context, js jetstream.JetStream) (buckets, error) {
+	out := buckets{}
+	for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
+		kv, err := js.KeyValue(ctx, name)
+		if err != nil {
+			return nil, fmt.Errorf("the bucket %s cannot be opened: %w", name, err)
+		}
+		out[name] = kv
+	}
+	return out, nil
+}
+
+// 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-assigner: ", 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 read the bus with")
+	}
+	account, err := process.ReadAccount(path)
+	if err != nil {
+		return err
+	}
+	opts := []nats.Option{
+		nats.Name(account.Node + "." + account.Module),
+		nats.UserInfo(account.User, account.Password),
+		nats.CustomInboxPrefix(account.InboxPrefix()),
+		nats.MaxReconnects(-1),
+		nats.RetryOnFailedConnect(true),
+	}
+	if strings.TrimSpace(account.Fingerprint) != "" {
+		opts = append(opts, nats.Secure(process.PinnedToFingerprint(account.Fingerprint)))
+	}
+	conn, err := nats.Connect(account.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
+	}
+	kvs, err := open(ctx, js)
+	if err != nil {
+		return err
+	}
+	// What the assigner keeps, said on the bus (ADR 0495 §2-§3): a reader is granted from it. Not fatal when it cannot
+	// be said now: process.Conn says it again at every reconnect.
+	said, err := process.Connect(account, nats.Name(account.Node+"."+account.Module+".keeps"), nats.RetryOnFailedConnect(true))
+	if err != nil {
+		return err
+	}
+	defer said.Close()
+	if err := said.Keeps(keptStates...); err != nil {
+		logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err)
+	}
+	logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
+	<-ctx.Done()
+	return nil
+}
@@ -0,0 +1,59 @@
+package main
+
+import (
+	"context"
+	"io"
+	"os"
+	"path/filepath"
+	"reflect"
+	"slices"
+	"strings"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+)
+
+// 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them
+// (`<module>.<state>` read as the bucket `<module>_<state>`), so it is granted what it reads and reads nothing else.
+func TestOpensWhatItsManifestNames(t *testing.T) {
+	raw, err := os.ReadFile(filepath.Join("..", "..", "module.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, err := catalogue.ParseManifest(raw)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var reads []string
+	for _, r := range m.Reads {
+		reads = append(reads, strings.Replace(r, ".", "_", 1))
+	}
+	var states []string
+	for _, s := range m.State {
+		states = append(states, m.Module+"_"+s.Name)
+	}
+	for _, pair := range []struct {
+		what       string
+		code, want []string
+	}{{"reads", readBuckets, reads}, {"states", ownBuckets, states}} {
+		got, want := slices.Sorted(slices.Values(pair.code)), slices.Sorted(slices.Values(pair.want))
+		if !reflect.DeepEqual(got, want) {
+			t.Errorf("the program's %s are %v, the manifest's %v", pair.what, got, want)
+		}
+	}
+	var kept []string
+	for _, s := range keptStates {
+		kept = append(kept, m.Module+"_"+s.Name)
+	}
+	if got, want := slices.Sorted(slices.Values(kept)), slices.Sorted(slices.Values(ownBuckets)); !reflect.DeepEqual(got, want) {
+		t.Errorf("the program announces %v, keeps %v", got, want)
+	}
+}
+
+// 1297/B2a: without a credential the program does not start, and says why.
+func TestNoCredential(t *testing.T) {
+	err := run(context.Background(), func(string) string { return "" }, io.Discard)
+	if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {
+		t.Errorf("run without a credential = %v", err)
+	}
+}
@@ -0,0 +1,9 @@
+module mesh-delivery-assigner
+
+go 1.26.0
+
+require (
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261023131240-d9384bcb239a
+	github.com/nats-io/nats-server/v2 v2.11.17
+	github.com/nats-io/nats.go v1.54.0
+)
@@ -0,0 +1,27 @@
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4 h1:yYDxFRJoH59kiCoEulD+FBwctyf9CrtKIxlEAsVL39Q=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261018184557-238440d6f8c4/go.mod h1:GFuZUElBZ9A++mxgIKo97aXXo+kV0uJ/UkbhQPPIbrY=
+github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op h1:1BOWQJweNyvZMlpAHXGLiZQn9S+QXGcz3xh94lC0w6E=
+github.com/antithesishq/antithesis-sdk-go v0.8.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
+github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+github.com/google/go-tpm v0.9.8/go.mod h1:h9jEsEECg7gtLis0upRBQU+GhYVH6jMjrFxI8u6bVUY=
+github.com/klauspost/compress v1.20.1 h1:T7kKElXUMXrUJ2E9QhQhxFtcK5rPyLdsGZvdbLMPdiQ=
+github.com/klauspost/compress v1.20.1/go.mod h1:LUdAzn7YLVvxLpc7y3V1m40wESHTgc1422pwwBSKYuI=
+github.com/minio/highwayhash v1.0.4 h1:asJizugGgchQod2ja9NJlGOWq4s7KsAWr5XUc9Clgl4=
+github.com/minio/highwayhash v1.0.4/go.mod h1:GGYsuwP/fPD6Y9hMiXuapVvlIUEhFhMTh0rxU3ik1LQ=
+github.com/nats-io/jwt/v2 v2.8.2 h1:XXRgB60MSTnqsRwejQurVDs/hcv2dkt+86GjI+I/bMc=
+github.com/nats-io/jwt/v2 v2.8.2/go.mod h1:Ag/56sq9OblL4JgdYufDd16Egb17Kr/8WwwuO/forVc=
+github.com/nats-io/nats-server/v2 v2.11.17 h1:GKEghcFK6A+aFx11Yf1LjgLC3txAwvyhnYzhBIQZA8I=
+github.com/nats-io/nats-server/v2 v2.11.17/go.mod h1:B1sFVz4StNosQ903ak4N1G01Fl/9f8e06mXpFIE2K24=
+github.com/nats-io/nats.go v1.54.0 h1:vsXoOxjHp/GmPUN+EcI7uOf/uB+iAP+kEsAFNQN0yzA=
+github.com/nats-io/nats.go v1.54.0/go.mod h1:y+DZoD1oBOYfZTU681eTUiUjI0vbqYGixNVFHcjHJ0k=
+github.com/nats-io/nkeys v0.4.16 h1:rd5oAuLOb8mnAycB0xleuEBNS1pVVnN0fv/FF34Eypg=
+github.com/nats-io/nkeys v0.4.16/go.mod h1:llLgWoI0o4z/Q57q2R1kHfmocyhGV6VG/U18Glg1Afs=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
+golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M=
+golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA=
+golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
+golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo=
+golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
+golang.org/x/time v0.16.0 h1:vMb6ptszcQMkcwiRTAuNNU50gom6++Q/6gY2hDM6VDE=
+golang.org/x/time v0.16.0/go.mod h1:rVKOqvZeKvrDKTQiAHJ7wmwP0RzleSphoEA9RcdLA0s=
@@ -0,0 +1,57 @@
+package manifest
+
+import (
+	"os"
+	"reflect"
+	"slices"
+	"testing"
+
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
+)
+
+func load(t *testing.T) catalogue.Manifest {
+	t.Helper()
+	raw, err := os.ReadFile("module.json")
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, err := catalogue.ParseManifest(raw)
+	if err != nil {
+		t.Fatalf("mesh-sdk's manifest reader refuses the manifest: %v", err)
+	}
+	return m
+}
+
+// 1297/B2a: the controller's reader (mesh-sdk catalogue, which the controller reads manifests with) takes the
+// manifest; it reads exactly the buckets the assigner opens (the program's list, held equal in its own test), keeps
+// its three states, and holds a credential of its own.
+func TestManifest(t *testing.T) {
+	m := load(t)
+	if m.Module != "mesh-delivery-assigner" {
+		t.Errorf("module = %q", m.Module)
+	}
+	reads := slices.Clone(m.Reads)
+	slices.Sort(reads)
+	want := []string{"mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.grants",
+		"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
+		"mesh-delivery-indexer.modules", "mesh-delivery-watchman.presence", "node-engine.have"}
+	if !reflect.DeepEqual(reads, want) {
+		t.Errorf("reads = %v, want exactly %v", reads, want)
+	}
+	states := map[string]int{}
+	for _, s := range m.State {
+		states[s.Name] = s.History
+	}
+	if want := map[string]int{"compositions": 5, "differences": 64, "wants": 20}; !reflect.DeepEqual(states, want) {
+		t.Errorf("state = %v, want %v", states, want)
+	}
+	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-assigner" {
+		t.Errorf("secrets-owner = %q", m.SecretsOwner)
+	}
+	if len(m.Claims) != 0 || len(m.Tools) != 0 || len(m.Invokes) != 0 {
+		t.Errorf("the assigner claims, serves or calls something: claims %v, tools %v, invokes %v", m.Claims, m.Tools, m.Invokes)
+	}
+}
@@ -0,0 +1,70 @@
+{
+  "module": "mesh-delivery-assigner",
+  "version": "1",
+  "slug": "assigner",
+  "reads": [
+    "mesh-controller.assignments",
+    "mesh-controller.builds",
+    "mesh-controller.grants",
+    "mesh-delivery-builder.builds",
+    "mesh-delivery-indexer.candidates",
+    "mesh-delivery-indexer.manifests",
+    "mesh-delivery-indexer.modules",
+    "mesh-delivery-watchman.presence",
+    "node-engine.have"
+  ],
+  "state": [
+    { "name": "compositions", "history": 5 },
+    { "name": "differences", "history": 64 },
+    { "name": "wants", "history": 20 }
+  ],
+  "own-secrets": {
+    "broker": "${dir:state}/broker"
+  },
+  "secrets-owner": "mesh-delivery-assigner",
+  "resources": [
+    {
+      "id": "account",
+      "type": "user",
+      "name": "mesh-delivery-assigner",
+      "shell": "/usr/bin/nologin",
+      "home": "/var/lib/mesh-delivery-assigner"
+    },
+    {
+      "id": "state",
+      "type": "directory",
+      "mode": "0700",
+      "owner": "mesh-delivery-assigner",
+      "place": ".",
+      "unknown": "program"
+    },
+    {
+      "id": "assigner",
+      "type": "process",
+      "name": "mesh-delivery-assigner",
+      "artifact": "program",
+      "run": [
+        "./mesh-delivery-assigner"
+      ],
+      "env": {
+        "MESH_BROKER_FILE": "${dir:state}/broker"
+      },
+      "user": "mesh-delivery-assigner",
+      "health": {
+        "kind": "unit"
+      }
+    }
+  ],
+  "build": {
+    "artifacts": [
+      {
+        "name": "program",
+        "kind": "bundle",
+        "language": "go",
+        "system": "arch",
+        "from": "cmd/mesh-delivery-assigner",
+        "binary": "mesh-delivery-assigner"
+      }
+    ]
+  }
+}