fix(queue): push tenant account claims to the NATS resolver

go.mod

@@ -16,6 +16,7 @@ require (
 	github.com/lib/pq v1.10.9
 	github.com/minio/madmin-go/v3 v3.0.110
 	github.com/minio/minio-go/v7 v7.0.90
+	github.com/nats-io/nats.go v1.53.1
 	github.com/nats-io/nkeys v0.4.15
 	github.com/newrelic/go-agent/v3 v3.43.3
 	github.com/oschwald/maxminddb-golang v1.13.0
@@ -103,6 +104,7 @@ require (
 	github.com/montanaflynn/stats v0.7.1 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nats-io/jwt/v2 v2.8.1 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/philhofer/fwd v1.1.3-0.20240916144458-20a13a1f6b7c // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect

go.sum

@@ -182,8 +182,12 @@ github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/nats-io/jwt/v2 v2.8.1 h1:V0xpGuD/N8Mi+fQNDynXohVvp7ZztevW5io8CUWlPmU=
 github.com/nats-io/jwt/v2 v2.8.1/go.mod h1:nWnOEEiVMiKHQpnAy4eXlizVEtSfzacZ1Q43LIRavZg=
+github.com/nats-io/nats.go v1.53.1 h1:Otsq3uLc/kLdjmkNHkXH0jBqwUquwdKFoe3fq6/3/Xo=
+github.com/nats-io/nats.go v1.53.1/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno=
 github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4=
 github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/newrelic/go-agent/v3 v3.43.3 h1:0A6DkUBYK2bidV6jJDJ1SD2XkRlg976nl+SiEqkGTUQ=
 github.com/newrelic/go-agent/v3 v3.43.3/go.mod h1:MFXnCId5xXMIJI6A/kbkg0DO48EVTsKcmNijMYphzTg=
 github.com/onsi/ginkgo/v2 v2.27.2 h1:LzwLj0b89qtIy6SSASkzlNvX6WktqurSHwkk2ipF/Ns=

internal/config/config.go

@@ -134,9 +134,23 @@ type Config struct {
 	NATSOperatorSeed     string // NATS_OPERATOR_SEED — operator NKey seed; empty = legacy_open fallback
 	NATSSystemAccountKey string // NATS_SYSTEM_ACCOUNT_PUBLIC_KEY — system account public key
 	NATSUseTLS           bool   // NATS_USE_TLS — true → tls:// URLs
-	R2Endpoint           string // R2_ENDPOINT — R2 endpoint hostname (default: r2.instant.dev)
-	R2BucketName         string // R2_BUCKET_NAME — shared R2 bucket name (default: instant-shared)
-	R2APIToken           string // R2_API_TOKEN — Cloudflare API token; if empty, R2 is not used
+
+	// System-account USER credentials (distinct from NATSSystemAccountKey,
+	// which is only the account's public key). Minting a tenant account JWT
+	// is useless unless the claim is pushed to the running nats-server on
+	// $SYS.REQ.CLAIMS.UPDATE, and only a connection authenticated INTO the
+	// system account may publish there. Both are secrets and are never
+	// logged. Required whenever NATSOperatorSeed is set; without them
+	// /queue/new would issue credentials the server rejects.
+	NATSSystemUserJWT  string // NATS_SYSTEM_USER_JWT — SYS-account user JWT (secret)
+	NATSSystemUserSeed string // NATS_SYSTEM_USER_SEED — SYS-account user NKey seed (secret)
+	// NATSSystemURL overrides the URL used for that system-account
+	// connection. Default: nats://<NATS_HOST>:4222 (in-cluster, plaintext).
+	NATSSystemURL string // NATS_SYSTEM_URL
+
+	R2Endpoint   string // R2_ENDPOINT — R2 endpoint hostname (default: r2.instant.dev)
+	R2BucketName string // R2_BUCKET_NAME — shared R2 bucket name (default: instant-shared)
+	R2APIToken   string // R2_API_TOKEN — Cloudflare API token; if empty, R2 is not used
 	// Object storage backend for /storage/new (provider-agnostic).
 	//
 	// ObjectStoreBackend selects the credential-issuance strategy:
@@ -443,6 +457,9 @@ func Load() *Config {
 	cfg.NATSPublicHost = getenv("NATS_PUBLIC_HOST", "nats.instanode.dev")
 	cfg.NATSOperatorSeed = os.Getenv("NATS_OPERATOR_SEED")
 	cfg.NATSSystemAccountKey = os.Getenv("NATS_SYSTEM_ACCOUNT_PUBLIC_KEY")
+	cfg.NATSSystemUserJWT = strings.TrimSpace(os.Getenv("NATS_SYSTEM_USER_JWT"))
+	cfg.NATSSystemUserSeed = strings.TrimSpace(os.Getenv("NATS_SYSTEM_USER_SEED"))
+	cfg.NATSSystemURL = strings.TrimSpace(os.Getenv("NATS_SYSTEM_URL"))
 	cfg.NATSUseTLS = os.Getenv("NATS_USE_TLS") == "true"
 	cfg.R2Endpoint = getenv("R2_ENDPOINT", "r2.instant.dev")
 	cfg.R2BucketName = getenv("R2_BUCKET_NAME", "instant-shared")

internal/config/config_test.go

@@ -54,6 +54,7 @@ func allKeys() []string {
 		"POSTGRES_CUSTOMERS_URL", "PROVISIONER_ADDR", "PROVISIONER_SECRET",
 		"NATS_HOST", "QUEUE_BACKEND", "NATS_PUBLIC_HOST",
 		"NATS_OPERATOR_SEED", "NATS_SYSTEM_ACCOUNT_PUBLIC_KEY", "NATS_USE_TLS",
+		"NATS_SYSTEM_USER_JWT", "NATS_SYSTEM_USER_SEED", "NATS_SYSTEM_URL",
 		"R2_ENDPOINT", "R2_BUCKET_NAME", "R2_API_TOKEN",
 		"OBJECT_STORE_MODE", "OBJECT_STORE_BACKEND", "OBJECT_STORE_ENDPOINT",
 		"OBJECT_STORE_PUBLIC_URL", "OBJECT_STORE_ACCESS_KEY",
@@ -307,6 +308,9 @@ func TestLoad_OverrideDefaults(t *testing.T) {
 		"NATS_OPERATOR_SEED":             "SO_seed",
 		"NATS_SYSTEM_ACCOUNT_PUBLIC_KEY": "ACSYS",
 		"NATS_USE_TLS":                   "true",
+		"NATS_SYSTEM_USER_JWT":           " eyJ0eXAiOiJKV1QifQ.sys.user\n",
+		"NATS_SYSTEM_USER_SEED":          "SUSYSSEED\n",
+		"NATS_SYSTEM_URL":                " nats://nats.x:4222 ",
 		"R2_ENDPOINT":                    "r2.x",
 		"R2_BUCKET_NAME":                 "x-bucket",
 		"R2_API_TOKEN":                   "r2tok",
@@ -333,6 +337,18 @@ func TestLoad_OverrideDefaults(t *testing.T) {
 	if !cfg.NATSUseTLS {
 		t.Error("NATSUseTLS must be true when env=true")
 	}
+	// SYS-account user credentials + system-URL override. All three are
+	// whitespace-trimmed: trailing newlines are what `kubectl create secret
+	// --from-file` leaves behind, and an untrimmed NKey seed fails to parse.
+	if cfg.NATSSystemUserJWT != "eyJ0eXAiOiJKV1QifQ.sys.user" {
+		t.Errorf("NATSSystemUserJWT trim: %q", cfg.NATSSystemUserJWT)
+	}
+	if cfg.NATSSystemUserSeed != "SUSYSSEED" {
+		t.Errorf("NATSSystemUserSeed trim: %q", cfg.NATSSystemUserSeed)
+	}
+	if cfg.NATSSystemURL != "nats://nats.x:4222" {
+		t.Errorf("NATSSystemURL trim: %q", cfg.NATSSystemURL)
+	}
 	// API_PUBLIC_URL — trailing slash must be trimmed.
 	if cfg.APIPublicURL != "https://api.x" {
 		t.Errorf("APIPublicURL trim: %q", cfg.APIPublicURL)

internal/handlers/export_queueresolver_test.go

@@ -0,0 +1,51 @@
+package handlers
+
+// export_queueresolver_test.go — test-only re-exports for the NATS
+// resolver-pusher wiring slice (queue_provider.go attachResolverPusher /
+// natsSystemURL / isolationUnavailable / unavailableCredProvider and
+// queue.go failQueueCredIssue).
+//
+// Kept in its own file (not the shared export_provarms_test.go) so concurrent
+// work on other handler slices never collides here. Go only compiles it in
+// test builds.
+
+import (
+	"github.com/gofiber/fiber/v2"
+
+	"instant.dev/common/queueprovider"
+	natsqp "instant.dev/common/queueprovider/nats"
+	"instant.dev/internal/config"
+	"instant.dev/internal/models"
+	"instant.dev/internal/natsresolver"
+)
+
+// SwapResolverPusherFactoryForTest replaces the system-account pusher
+// constructor and returns a restore func, so the attach / skip / fail arms of
+// attachResolverPusher can be driven without a NATS server.
+func SwapResolverPusherFactoryForTest(
+	fn func(natsresolver.Config) (natsqp.ResolverPusher, error),
+) (restore func()) {
+	prev := newResolverPusher
+	newResolverPusher = fn
+	return func() { newResolverPusher = prev }
+}
+
+// NATSSystemURLForTest re-exports natsSystemURL.
+func NATSSystemURLForTest(cfg *config.Config) string { return natsSystemURL(cfg) }
+
+// IsolationUnavailableForTest re-exports isolationUnavailable.
+func IsolationUnavailableForTest(err error) bool { return isolationUnavailable(err) }
+
+// NewUnavailableCredProviderForTest re-exports the stand-in provider installed
+// when isolation is configured but could not be initialised.
+func NewUnavailableCredProviderForTest(cause error) queueprovider.QueueCredentialProvider {
+	return unavailableCredProvider{cause: cause}
+}
+
+// FailQueueCredIssueForTest re-exports QueueHandler.failQueueCredIssue so the
+// mark-failed-error log branch can be driven with a closed DB.
+func (h *QueueHandler) FailQueueCredIssueForTest(
+	c *fiber.Ctx, resource *models.Resource, prid, token, logPrefix string, cause error,
+) error {
+	return h.failQueueCredIssue(c, resource, prid, token, logPrefix, cause)
+}

internal/handlers/queue.go

@@ -65,21 +65,40 @@ func NewQueueHandler(db *sql.DB, rdb *redis.Client, cfg *config.Config, provClie
 	// does not yet have a ProvisionQueue RPC. When it does, wire it here like
 	// CacheHandler.provisionCache does.
 	h.queueProvider = queueprovider.New(cfg.NATSHost)
-	// Build the credential issuer. Falls back to legacy_open when no operator
-	// seed is configured so api can deploy before the operator-key generation.
-	if cp, err := buildQueueProvider(cfg); err == nil {
+	// Build the credential issuer. Three outcomes: a working provider; a
+	// legacy_open fallback when NO operator seed is configured (so api can
+	// deploy before the operator-key generation); or — when the operator seed
+	// IS configured but the isolation path could not be initialised — a
+	// provider that refuses to issue, so /queue/new 503s instead of returning
+	// credentials nats-server would reject.
+	cp, err := buildQueueProvider(cfg)
+	switch {
+	case err == nil:
 		h.credProvider = cp
-	} else {
+	case cfg.NATSOperatorSeed != "":
+		// Isolation is CONFIGURED (operator seed present) but could not be
+		// initialised — typically the $SYS resolver publisher failed to
+		// connect. Falling back to legacy_open here would hand every caller a
+		// connection URL with no credentials against an auth_required server:
+		// the exact "issued but dead" failure this path exists to prevent.
+		// Fail the queue credential path loudly instead; /queue/new answers
+		// 503 until the operator fixes the NATS wiring.
+		slog.Error("queue.cred_provider_init_failed_isolation_unavailable",
+			"error", err,
+			"backend", cfg.QueueBackend,
+			"detail", "operator seed is set — refusing to downgrade to legacy_open; /queue/new will 503")
+		h.credProvider = unavailableCredProvider{cause: err}
+	default:
 		slog.Error("queue.cred_provider_init_failed_fallback_legacy_open",
 			"error", err,
 			"backend", cfg.QueueBackend)
 		// Defensive: never leave h.credProvider nil. The legacyopen provider
 		// is always registered so this fallback always succeeds.
 		fallback, _ := commonqp.Factory(commonqp.Config{
-			Backend:    "legacy_open",
+			Backend:    queueBackendLegacyOpen,
 			Host:       cfg.NATSHost,
 			PublicHost: cfg.NATSPublicHost,
-			Port:       4222,
+			Port:       natsClientPort,
 			UseTLS:     cfg.NATSUseTLS,
 		})
 		h.credProvider = fallback
@@ -150,6 +169,32 @@ func (h *QueueHandler) issueTenantCreds(ctx context.Context, token, subjectPrefi
 	return creds, nil
 }
 
+// failQueueCredIssue aborts a provision whose per-tenant credentials could not
+// be issued because the isolation path is broken (the account claim never
+// reached the nats-server resolver).
+//
+// CLAUDE.md rule 2: provisioning is synchronous and a backend failure is a
+// 503 — never a 201 carrying credentials for something the backend does not
+// know about. Returning the legacy_open response shape here would be exactly
+// that: a connection URL with no credentials against an auth_required server.
+// So the backend resource is torn down, the row is marked failed (failed rows
+// never count against quota) and the caller gets 503.
+func (h *QueueHandler) failQueueCredIssue(
+	c *fiber.Ctx, resource *models.Resource, prid, token, logPrefix string, cause error,
+) error {
+	ctx := c.UserContext()
+	metrics.ProvisionFailures.WithLabelValues("queue", "cred_issue_error").Inc()
+	middleware.RecordProvisionFail("queue", middleware.ProvisionFailBackendUnavailable)
+	slog.Error(logPrefix+".cred_issue_failed_isolation_unavailable",
+		"error", cause, "token", token, "resource_id", resource.ID)
+	deprovisionBestEffort(ctx, h.provClient, token, prid, "queue", logPrefix)
+	if delErr := models.MarkResourceFailed(ctx, h.db, resource.ID); delErr != nil {
+		slog.Error(logPrefix+".soft_delete_failed_cred_issue",
+			"error", delErr, "resource_id", resource.ID)
+	}
+	return respondProvisionFailed(c, cause, "Failed to issue isolated NATS credentials")
+}
+
 // NewQueue handles POST /queue/new.
 func (h *QueueHandler) NewQueue(c *fiber.Ctx) error {
 	if !h.cfg.IsServiceEnabled("queue") {
@@ -308,7 +353,10 @@ func (h *QueueHandler) NewQueue(c *fiber.Ctx) error {
 	// MR-P0-5: issue per-tenant credentials via the queueprovider abstraction.
 	// May return AuthMode=isolated (real per-tenant account JWT) or
 	// AuthMode=legacy_open (no auth — staged-cutover fallback).
-	tenantCreds, _ := h.issueTenantCreds(ctx, tokenStr, creds.SubjectPrefix)
+	tenantCreds, credErr := h.issueTenantCreds(ctx, tokenStr, creds.SubjectPrefix)
+	if isolationUnavailable(credErr) {
+		return h.failQueueCredIssue(c, resource, creds.ProviderResourceID, tokenStr, "queue.new", credErr)
+	}
 	authMode := commonqp.AuthModeLegacyOpen
 	if tenantCreds != nil && tenantCreds.AuthMode != "" {
 		authMode = tenantCreds.AuthMode
@@ -531,7 +579,10 @@ func (h *QueueHandler) newQueueAuthenticated(
 	}
 
 	// MR-P0-5: issue per-tenant credentials via the queueprovider abstraction.
-	tenantCreds, _ := h.issueTenantCreds(ctx, tokenStr, creds.SubjectPrefix)
+	tenantCreds, credErr := h.issueTenantCreds(ctx, tokenStr, creds.SubjectPrefix)
+	if isolationUnavailable(credErr) {
+		return h.failQueueCredIssue(c, resource, creds.ProviderResourceID, tokenStr, "queue.new.auth", credErr)
+	}
 	authMode := commonqp.AuthModeLegacyOpen
 	if tenantCreds != nil && tenantCreds.AuthMode != "" {
 		authMode = tenantCreds.AuthMode