Assert every declared state's bucket on each push, before the memberships that name it (novox/hq ADR 0201)

The raise at start was the only place buckets were asserted, so a module
registered and assigned since had none until the control plane restarted —
found on the first module to declare state.

internal/broker/jetstream.go

@@ -86,6 +86,13 @@ func pinnedTo(path string) (*tls.Config, error) {
 	return PinnedToFingerprint(want), nil
 }
 
+// OnConn is the JetStream handle over a connection the caller already holds — the control plane's
+// link — for asserting what the bus holds without dialling a second time.
+func OnConn(conn *nats.Conn) *JetStream {
+	js, _ := conn.JetStream()
+	return &JetStream{conn: conn, js: js}
+}
+
 // DialPinned is Dial with the server's certificate pinned by a fingerprint the caller already holds
 // — a module or a build machine that was handed one beside its credential, and has no file.
 func DialPinned(url, fingerprint string, opts ...nats.Option) (*JetStream, error) {

internal/broker/state_test.go

@@ -164,3 +164,17 @@ func TestABucketIsAssertedInPlace(t *testing.T) {
 		}
 	}
 }
+
+// Against a real server: the handle over a connection the control plane already holds asserts a
+// bucket as Dial's does — what a push uses, so a module registered since the last start has its
+// bucket before its membership names it.
+func TestABucketIsAssertedOverAHeldConnection(t *testing.T) {
+	js := aLiveBus(t)
+	held := OnConn(js.Conn())
+	if _, err := RaiseBuckets(held, []Bucket{{Module: "statetest", Name: "held"}}); err != nil {
+		t.Fatalf("asserting over a held connection failed: %v", err)
+	}
+	if _, err := js.Context().KeyValue("statetest_held"); err != nil {
+		t.Fatalf("the bucket is not there: %v", err)
+	}
+}