From 034a6de9a706b7f9958597a6d0bb7b7208aaa17a Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 15:19:06 +0200 Subject: [PATCH] =?UTF-8?q?Make=20the=20module=20mesh-delivery-assigner:?= =?UTF-8?q?=20its=20manifest's=20reads=20and=20states,=20its=20own=20crede?= =?UTF-8?q?ntial=20and=20the=20states=20it=20announces,=20so=20the=20contr?= =?UTF-8?q?oller=20grants=20it=20what=20R5=20reads=20(issue=201297,=20subt?= =?UTF-8?q?ask=20B2a;=20ADR=200528,=20ADR=200429=20=C2=A71)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../cmd/mesh-delivery-assigner/main.go | 127 ++++++++++++++++++ .../cmd/mesh-delivery-assigner/main_test.go | 59 ++++++++ modules/mesh-delivery-assigner/go.mod | 9 ++ modules/mesh-delivery-assigner/go.sum | 27 ++++ .../mesh-delivery-assigner/manifest_test.go | 57 ++++++++ modules/mesh-delivery-assigner/module.json | 70 ++++++++++ 6 files changed, 349 insertions(+) create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go create mode 100644 modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go create mode 100644 modules/mesh-delivery-assigner/go.mod create mode 100644 modules/mesh-delivery-assigner/go.sum create mode 100644 modules/mesh-delivery-assigner/manifest_test.go create mode 100644 modules/mesh-delivery-assigner/module.json diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go new file mode 100644 index 0000000..4ac7045 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -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..`, 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 +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go new file mode 100644 index 0000000..f52e1fb --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -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 +// (`.` read as the bucket `_`), 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) + } +} diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod new file mode 100644 index 0000000..4998342 --- /dev/null +++ b/modules/mesh-delivery-assigner/go.mod @@ -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 +) diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum new file mode 100644 index 0000000..5c1a857 --- /dev/null +++ b/modules/mesh-delivery-assigner/go.sum @@ -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= diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go new file mode 100644 index 0000000..417ac68 --- /dev/null +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -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) + } +} diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json new file mode 100644 index 0000000..599c3f6 --- /dev/null +++ b/modules/mesh-delivery-assigner/module.json @@ -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" + } + ] + } +}