Merge pull request 'Issues 203 and 206: an assignment issues its credential; the controller owns a worker's shape; the build seat's holder follows the controller' (#233) from fix/issues-203-206 into main

internal/broker/jetstream.go

@@ -214,6 +214,51 @@ func (j *JetStream) EnsureConsumer(c Consumer) error {
 
 	switch have, err := j.js.ConsumerInfo(c.Stream, c.Name); {
 	case err == nil:
+		// **The controller owns the worker's shape, type included** (novox/hq issue 206). A holder
+		// built for a pull worker cannot bind a push one — `cannot pull subscribe to push based
+		// consumer` — and on 2026-10-03 the build machine rolled before the controller that would
+		// have redefined its worker, restarted on that for an hour, and nothing could build the
+		// controller that would have ended it. The server cannot change a consumer's type in place,
+		// so one of the wrong type is re-made: on a work queue nothing is lost, because what was
+		// acknowledged is gone from the stream and what was not is delivered again from the start.
+		// On any other stream a re-made consumer would replay what this one acknowledged (issue
+		// 156), so there it is said and left, and the person re-makes it knowing the cost.
+		if havePush, wantPush := have.Config.DeliverSubject != "", want.DeliverSubject != ""; havePush != wantPush {
+			shape := func(push bool) string {
+				if push {
+					return "push"
+				}
+				return "pull"
+			}
+			info, err := j.js.StreamInfo(c.Stream)
+			if err != nil {
+				return fmt.Errorf("asking about stream %s to re-make consumer %s: %w", c.Stream, c.Name, err)
+			}
+			if info.Config.Retention != nats.WorkQueuePolicy {
+				// **A stream that keeps its history is re-made from now on, never from the start.**
+				// Left for a hand, the hand re-makes it with the server's default — everything the
+				// stream holds — which on 2026-10-03 replayed every build ask since 1 October and
+				// re-registered nine modules from the past (novox/hq issue 207). What this consumer
+				// had not yet acknowledged is lost with it, and said: on a history stream that is
+				// the smaller cost, and the asks in flight are visible to whoever asked.
+				j.note("consumer %s on %s changes from %s to %s delivery on a stream that keeps its history: "+
+					"re-made to deliver from now on, so nothing this one acknowledged comes back (novox/hq issue "+
+					"207); %d ask(s) it had not acknowledged are not carried over and must be asked again",
+					c.Name, c.Stream, shape(havePush), shape(wantPush), have.NumPending+uint64(have.NumAckPending))
+				want.DeliverPolicy = nats.DeliverNewPolicy
+			} else {
+				j.note("consumer %s on %s changes from %s to %s delivery: re-made where it left off, nothing "+
+					"acknowledged comes back and nothing pending is lost (novox/hq issue 206); a holder bound to "+
+					"the old shape binds again", c.Name, c.Stream, shape(havePush), shape(wantPush))
+			}
+			if err := j.js.DeleteConsumer(c.Stream, c.Name); err != nil {
+				return fmt.Errorf("re-making consumer %s on %s as %s: %w", c.Name, c.Stream, shape(wantPush), err)
+			}
+			if _, err := j.js.AddConsumer(c.Stream, want); err != nil {
+				return fmt.Errorf("re-making consumer %s on %s as %s: %w", c.Name, c.Stream, shape(wantPush), err)
+			}
+			return nil
+		}
 		// Where an existing consumer starts is its history, not something an assertion may move:
 		// the server refuses a changed deliver policy outright. Carried across, so asserting twice
 		// is the no-op a restart depends on.

internal/broker/worker_type_change_test.go

@@ -0,0 +1,167 @@
+package broker
+
+import (
+	"os"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats.go"
+)
+
+// A seat's worker that changed from push to pull delivery strands a holder built for the new shape
+// (novox/hq issue 206): the server refuses a pull subscription on a push consumer, and the controller
+// that would redefine it was the build that nobody could take. The controller owns the worker's
+// shape, type included: on a work queue it re-makes one of the wrong type, losing nothing, and a
+// pull subscription then binds and takes what was pending.
+//
+//	docker run -d --rm --name t -p 14231:4222 nats:2.10-alpine -js
+//	MESH_TEST_NATS=nats://127.0.0.1:14231 go test ./internal/broker/ -run TestAWorker
+func TestAWorkerOfTheWrongTypeIsRemadeOnAWorkQueueAndAPullThenBinds(t *testing.T) {
+	url := os.Getenv("MESH_TEST_NATS")
+	if url == "" {
+		t.Skip("MESH_TEST_NATS unset")
+	}
+	js, err := Dial(url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer js.Close()
+
+	const stream, worker, filter = "SEAT_T_SHELF", "SEAT_T_SHELF_worker", "mesh.seat.t-shelf.accept.>"
+	_ = js.js.DeleteStream(stream)
+	if _, err := js.js.AddStream(&nats.StreamConfig{
+		Name: stream, Subjects: []string{filter}, Retention: nats.WorkQueuePolicy, Storage: nats.MemoryStorage,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = js.js.DeleteStream(stream) }()
+
+	// The worker as the previous controller defined it: push, in a queue group.
+	if _, err := js.js.AddConsumer(stream, &nats.ConsumerConfig{
+		Durable: worker, AckPolicy: nats.AckExplicitPolicy, AckWait: 60 * time.Second, MaxDeliver: 5,
+		FilterSubject: filter, DeliverSubject: "_DELIVER." + worker, DeliverGroup: "holders",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	for _, body := range []string{"one", "two", "three"} {
+		if _, err := js.js.Publish("mesh.seat.t-shelf.accept.build", []byte(body)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	// The old holder took and acknowledged the first ask, then went away.
+	old, err := js.js.QueueSubscribeSync(filter, "holders", nats.Bind(stream, worker))
+	if err != nil {
+		t.Fatal(err)
+	}
+	m, err := old.NextMsg(twoSeconds)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(m.Data) != "one" {
+		t.Fatalf("the first ask is %q", m.Data)
+	}
+	if err := m.AckSync(); err != nil {
+		t.Fatal(err)
+	}
+	if err := old.Unsubscribe(); err != nil {
+		t.Fatal(err)
+	}
+
+	// The new controller asserts the worker as the mesh derives it now: pull.
+	if err := js.EnsureConsumer(Consumer{
+		Name: worker, Stream: stream, Filters: []string{filter}, AckWaitSeconds: 60, MaxDeliver: 5,
+		Why: "the test's worker",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	have, err := js.js.ConsumerInfo(stream, worker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if have.Config.DeliverSubject != "" || have.Config.DeliverGroup != "" {
+		t.Fatalf("the worker is still push: %+v", have.Config)
+	}
+
+	// A holder built for the new shape binds, and takes exactly what the old one left.
+	sub, err := js.js.PullSubscribe(filter, worker, nats.Bind(stream, worker), nats.ManualAck())
+	if err != nil {
+		t.Fatalf("a pull subscription does not bind the re-made worker: %v", err)
+	}
+	got, err := sub.Fetch(3, nats.MaxWait(twoSeconds))
+	if err != nil && len(got) == 0 {
+		t.Fatalf("nothing pending was delivered: %v", err)
+	}
+	var bodies []string
+	for _, g := range got {
+		bodies = append(bodies, string(g.Data))
+		_ = g.Ack()
+	}
+	if len(bodies) != 2 || bodies[0] != "two" || bodies[1] != "three" {
+		t.Fatalf("the pending asks after the acknowledged one, in order: %v", bodies)
+	}
+
+	// Asserted again, the pull worker is the no-op a restart depends on.
+	if err := js.EnsureConsumer(Consumer{
+		Name: worker, Stream: stream, Filters: []string{filter}, AckWaitSeconds: 60, MaxDeliver: 5,
+	}); err != nil {
+		t.Fatal(err)
+	}
+}
+
+// On a stream that keeps its history, a worker of the wrong type is re-made to deliver from now on:
+// re-making it from the start would replay what it acknowledged (novox/hq issue 156), and leaving it
+// for a hand re-made it exactly that way on 2026-10-03 (issue 207).
+func TestAWorkerOfTheWrongTypeOnAHistoryStreamIsRemadeFromNowOn(t *testing.T) {
+	url := os.Getenv("MESH_TEST_NATS")
+	if url == "" {
+		t.Skip("MESH_TEST_NATS unset")
+	}
+	js, err := Dial(url)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer js.Close()
+	const stream, worker, filter = "EVENTS_T", "EVENTS_T_reader", "mesh.t.event.>"
+	_ = js.js.DeleteStream(stream)
+	if _, err := js.js.AddStream(&nats.StreamConfig{Name: stream, Subjects: []string{filter}, Storage: nats.MemoryStorage}); err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = js.js.DeleteStream(stream) }()
+	if _, err := js.js.AddConsumer(stream, &nats.ConsumerConfig{
+		Durable: worker, AckPolicy: nats.AckExplicitPolicy, AckWait: 60 * time.Second,
+		FilterSubject: filter, DeliverSubject: "_DELIVER." + worker,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	// History the old consumer would have acknowledged long ago, and must not come back.
+	for i := 0; i < 3; i++ {
+		if _, err := js.js.Publish("mesh.t.event.old", []byte("old")); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if err := js.EnsureConsumer(Consumer{Name: worker, Stream: stream, Filters: []string{filter}, AckWaitSeconds: 60}); err != nil {
+		t.Fatal(err)
+	}
+	have, err := js.js.ConsumerInfo(stream, worker)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if have.Config.DeliverSubject != "" {
+		t.Fatal("a history stream's consumer of the wrong type was left as it was")
+	}
+	if have.Config.DeliverPolicy != nats.DeliverNewPolicy || have.NumPending != 0 {
+		t.Fatalf("re-made consumer delivers %v with %d pending; it must deliver from now on with nothing of the past", have.Config.DeliverPolicy, have.NumPending)
+	}
+	// And what arrives from now on is delivered.
+	if _, err := js.js.Publish("mesh.t.event.new", []byte("new")); err != nil {
+		t.Fatal(err)
+	}
+	sub, err := js.js.PullSubscribe(filter, worker, nats.Bind(stream, worker))
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, err := sub.Fetch(1, nats.MaxWait(3*time.Second))
+	if err != nil || len(got) != 1 || string(got[0].Data) != "new" {
+		t.Fatalf("the re-made consumer delivered %v, %v; want the one new message", got, err)
+	}
+}

internal/catalogue/declaration.go

@@ -872,6 +872,15 @@ func (r Resolution) compose(with Rendering, owner map[string]string,
 			if renamed := reflectsRenamed(m.Module, resource["reload-on"]); renamed != nil {
 				copied["reload-on"] = renamed
 			}
+			// **What reads one of this module's own secrets is restarted when it changes** (novox/hq
+			// issue 203, issue 206). A credential is re-issued by the mesh, and a container that
+			// mounted the old file keeps the old one open: the build machine ran for an hour on a
+			// credential the mesh had replaced, because its manifest restarted it on its
+			// environment file and nobody had thought to name the credential too. Composed here so
+			// no manifest has to say it, for a container or a daemon that names the secret's path.
+			if reads := secretsReadBy(copied, m); len(reads) > 0 {
+				copied["restart-on"] = withRestartOn(copied["restart-on"], reads)
+			}
 			// **A version prepares its state before it runs** (novox/hq ADR 0135). Derived from the
 			// module's own resource rather than declared beside it: what prepares the state is the
 			// module's own code, so what it is given has to be what that code is given — and a
@@ -2079,3 +2088,87 @@ func portOfEndpoint(values map[string]any, ports map[string]int) {
 		values["port"] = port
 	}
 }
+
+// secretsReadBy is the file resources of this module's own secrets that a container or a daemon reads
+// — named in its volumes, its environment or its env-files by the secret's placed path — as
+// restart-on ids. Nothing for other shapes, and nothing for a scheduled or run-once process, which
+// the host refuses a restart-on for (it runs again anyway, and reads the file afresh).
+func secretsReadBy(resource map[string]any, m Manifest) []string {
+	kind := fmt.Sprint(resource["type"])
+	if kind != "container" && kind != "process" {
+		return nil
+	}
+	if resource["schedule"] != nil || resource["run-once"] == true {
+		return nil
+	}
+	var mentioned []string
+	for _, key := range []string{"volumes", "env", "env-file"} {
+		mentioned = append(mentioned, stringsIn(resource[key])...)
+	}
+	var out []string
+	for _, name := range sortedKeys(m.OwnSecrets) {
+		path := m.OwnSecrets[name].Path
+		if path == "" {
+			continue
+		}
+		for _, s := range mentioned {
+			// A volume is `source:destination[:mode]`; an env value or an env-file is the path itself.
+			if s == path || strings.HasPrefix(s, path+":") {
+				out = append(out, m.Module+"."+NeedID(name))
+				break
+			}
+		}
+	}
+	return out
+}
+
+// stringsIn is every string in a list or a map's values; nothing for anything else.
+func stringsIn(v any) []string {
+	switch x := v.(type) {
+	case []any:
+		var out []string
+		for _, item := range x {
+			if s, ok := item.(string); ok {
+				out = append(out, s)
+			}
+		}
+		return out
+	case []string:
+		return x
+	case map[string]any:
+		var out []string
+		for _, k := range sortedKeys(x) {
+			if s, ok := x[k].(string); ok {
+				out = append(out, s)
+			}
+		}
+		return out
+	case map[string]string:
+		var out []string
+		for _, k := range sortedKeys(x) {
+			out = append(out, x[k])
+		}
+		return out
+	}
+	return nil
+}
+
+// withRestartOn is a resource's restart-on list with these ids added once each.
+func withRestartOn(have any, add []string) []any {
+	var out []any
+	seen := map[string]bool{}
+	for _, id := range reflectsRenamed("", have) {
+		s := fmt.Sprint(id)
+		if !seen[s] {
+			seen[s] = true
+			out = append(out, s)
+		}
+	}
+	for _, id := range add {
+		if !seen[id] {
+			seen[id] = true
+			out = append(out, id)
+		}
+	}
+	return out
+}

internal/catalogue/secret_restart_test.go

@@ -0,0 +1,52 @@
+package catalogue
+
+import (
+	"reflect"
+	"strings"
+	"testing"
+)
+
+// What reads one of a module's own secrets is restarted when the secret changes (novox/hq issue 203,
+// issue 206): the build machine kept an hour-old credential open because its manifest restarted it
+// on its environment file alone. Composed, so a manifest need not say it; a scheduled process is
+// left alone, because the host refuses a restart-on for one and it reads the file afresh each run.
+func TestAContainerReadingAnOwnSecretIsRestartedWhenItChanges(t *testing.T) {
+	m := Manifest{Module: "agent", Version: "1",
+		OwnSecrets: OwnSecrets{"broker": {Path: "/var/lib/mesh/agent/broker"}},
+		Resources: []map[string]any{
+			{"id": "mesh-state", "type": "directory", "path": "/var/lib/mesh/agent", "mode": "0700"},
+			{"id": "settings", "type": "file", "path": "/var/lib/mesh/agent/agent.env", "mode": "0600", "content": "A=1\n"},
+			{"id": "server", "type": "container", "name": "agent", "network": "host",
+				"image":      "registry.example/agent@sha256:" + strings.Repeat("a", 64),
+				"volumes":    []any{"/var/lib/mesh/agent:/run/mesh:ro", "/var/lib/mesh/agent/broker:/run/mesh/broker:ro"},
+				"env-file":   []any{"/var/lib/mesh/agent/agent.env"},
+				"restart-on": []any{"settings"}},
+			{"id": "nightly", "type": "container", "name": "agent-nightly", "schedule": "0 3 * * *",
+				"image":   "registry.example/agent@sha256:" + strings.Repeat("a", 64),
+				"volumes": []any{"/var/lib/mesh/agent/broker:/run/mesh/broker:ro"}},
+			{"id": "other", "type": "container", "name": "agent-other",
+				"image": "registry.example/agent@sha256:" + strings.Repeat("a", 64)},
+		}}
+	got, err := Resolve(shelf(m), []string{m.Module},
+		Node{Name: "anchor", At: "10.0.0.1", Capabilities: map[string]bool{"container-runtime": true}}, World{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	out, err := got.Declaration(Rendering{Needed: map[string]map[string]string{"agent": {"broker": "SEALED"}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	by := map[string]map[string]any{}
+	for _, r := range out {
+		by[r["id"].(string)] = r
+	}
+	if want := []any{"agent.settings", "agent.needs-broker"}; !reflect.DeepEqual(by["agent.server"]["restart-on"], want) {
+		t.Fatalf("the server reads the credential and is not restarted on it: %v", by["agent.server"]["restart-on"])
+	}
+	if _, has := by["agent.nightly"]["restart-on"]; has {
+		t.Fatalf("a scheduled container was given a restart-on, which the host refuses: %v", by["agent.nightly"]["restart-on"])
+	}
+	if _, has := by["agent.other"]["restart-on"]; has {
+		t.Fatalf("a container that reads no secret was given one to restart on: %v", by["agent.other"]["restart-on"])
+	}
+}

internal/inventory/dependencies.go

@@ -18,8 +18,17 @@ const (
 	EdgeBuiltBy = "built-by"
 	// EdgeDeclared: the manifest's own `build.on`.
 	EdgeDeclared = "declared"
+	// EdgeWorkerOf: the module holds the build seat, whose worker the control plane defines
+	// (novox/hq issue 206). The one place a *running* order enters the graph: a build machine rolled
+	// before the controller that redefines its worker cannot bind it, and nothing can then build the
+	// controller that would end that — so the holder of the build seat follows the controller, and
+	// the controller is built by whichever build machine is running, as it always was.
+	EdgeWorkerOf = "worker-of"
 )
 
+// TheControlPlane is the module that defines every seat's worker on the bus.
+const TheControlPlane = "mesh-controller"
+
 // Edge is one dependency: From depends on To, in the way Kind says.
 type Edge struct {
 	From string `json:"from"`
@@ -106,6 +115,13 @@ func dependenciesOf(entries []Entry, against map[string][]string, read map[strin
 			}
 		}
 	}
+	if known[TheControlPlane] {
+		for _, b := range builders {
+			if b != TheControlPlane {
+				add(b, TheControlPlane, EdgeWorkerOf)
+			}
+		}
+	}
 	sort.Slice(out, func(a, b int) bool {
 		if out[a].From != out[b].From {
 			return out[a].From < out[b].From