fix(queue): circuit breaker on NATS/queue backend (sweep #2) (#42)

Queue/NATS was the only provisioner backend with no circuit breaker: a
wedged shared-NATS account/JWT call (or a kube-apiserver stall on the
dedicated path) could pile up with no fast-fail and no /readyz signal,
unlike postgres/redis/mongo which all trip after 5 consecutive failures.

Add a QueueAdmin breaker (BackendQueueAdmin = "queue_admin"), a
breakerForQueue(useDedicated) helper (shared → QueueAdmin, dedicated →
K8sAPI, mirroring breakerForMongo), and wrap all 4 queue dispatch sites
(Provision shared + dedicated, Deprovision shared + dedicated) in
callBackend/callBackendVoid. Register QueueAdmin in collectBreakerInspectors
so a tripped queue breaker surfaces as backend_queue_admin on /readyz.

Tests: TestServer_QueueCircuitTripsOnRepeatedFailures (5 failures trip the
breaker; 6th short-circuits to Unavailable WITHOUT invoking the backend,
asserted via a call counter). The two registry-drift guards
(TestBreakers_EveryStructFieldHasAnInspector, _AllBackendsSurfaced) caught
the new breaker and are updated — they enforce that a new breaker is wired
into BOTH the inspector slice and /readyz. Full `go test ./... -short` green.

Co-authored-by: Manas Srivastava <[email protected]>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: mastermanas805

coverage_registry_test.go

@@ -37,31 +37,32 @@ import (
 // collectBreakerInspectors fails this test.
 //
 // COVERAGE BLOCK (rule 17):
-//   Symptom:       a new circuit-protected backend is added to
-//                  *circuit.Breakers (new field, e.g. NATSAdmin) but
-//                  collectBreakerInspectors in main.go is not
-//                  extended. The new backend's circuit state never
-//                  surfaces on /readyz. A future tripped breaker
-//                  doesn't show up as a degraded check, the NR alert
-//                  filter (which keys on `circuit.opened` log lines)
-//                  fires but operators have no /readyz signal to
-//                  cross-reference.
-//   Enumeration:   reflect over the *circuit.Breakers struct fields
-//                  whose type is *circuit.Breaker. This IS the
-//                  registry.
-//   Sites found:   N fields on Breakers (currently 5: PostgresAdmin,
-//                  PostgresK8s, RedisAdmin, MongoAdmin, K8sAPI).
-//   Sites touched: each must be findable in the inspector slice
-//                  collectBreakerInspectors returns. The match is
-//                  by .Name() — every breaker carries the backend
-//                  name as its identity.
-//   Coverage test: missing-from-inspectors and missing-from-struct
-//                  both fail.
-//   Live verified: provisioner /readyz output today shows
-//                  backend_postgres_admin / backend_postgres_k8s /
-//                  backend_redis_admin / backend_mongo_admin /
-//                  backend_k8s_api — five degraded checks. This test
-//                  pins that count to whatever the struct says.
+//
+//	Symptom:       a new circuit-protected backend is added to
+//	               *circuit.Breakers (new field, e.g. NATSAdmin) but
+//	               collectBreakerInspectors in main.go is not
+//	               extended. The new backend's circuit state never
+//	               surfaces on /readyz. A future tripped breaker
+//	               doesn't show up as a degraded check, the NR alert
+//	               filter (which keys on `circuit.opened` log lines)
+//	               fires but operators have no /readyz signal to
+//	               cross-reference.
+//	Enumeration:   reflect over the *circuit.Breakers struct fields
+//	               whose type is *circuit.Breaker. This IS the
+//	               registry.
+//	Sites found:   N fields on Breakers (currently 5: PostgresAdmin,
+//	               PostgresK8s, RedisAdmin, MongoAdmin, K8sAPI).
+//	Sites touched: each must be findable in the inspector slice
+//	               collectBreakerInspectors returns. The match is
+//	               by .Name() — every breaker carries the backend
+//	               name as its identity.
+//	Coverage test: missing-from-inspectors and missing-from-struct
+//	               both fail.
+//	Live verified: provisioner /readyz output today shows
+//	               backend_postgres_admin / backend_postgres_k8s /
+//	               backend_redis_admin / backend_mongo_admin /
+//	               backend_k8s_api — five degraded checks. This test
+//	               pins that count to whatever the struct says.
 func TestBreakers_EveryStructFieldHasAnInspector(t *testing.T) {
 	bs := circuit.NewBreakers()
 	got := collectBreakerInspectors(bs)
@@ -90,6 +91,7 @@ func TestBreakers_EveryStructFieldHasAnInspector(t *testing.T) {
 		"PostgresAdmin": circuit.BackendPostgresAdmin,
 		"RedisAdmin":    circuit.BackendRedisAdmin,
 		"MongoAdmin":    circuit.BackendMongoAdmin,
+		"QueueAdmin":    circuit.BackendQueueAdmin,
 		"K8sAPI":        circuit.BackendK8sAPI,
 	}
 

internal/circuit/breakers.go

@@ -15,6 +15,7 @@ const (
 	BackendPostgresAdmin = "postgres_admin" // shared postgres-customers CREATE DATABASE / CREATE USER (local.go)
 	BackendRedisAdmin    = "redis_admin"    // shared redis-provision ACL SETUSER / namespace ops
 	BackendMongoAdmin    = "mongo_admin"    // shared mongo admin CREATE USER / role grants
+	BackendQueueAdmin    = "queue_admin"    // shared NATS account/JWT provisioning + deprovision
 	BackendK8sAPI        = "k8s_api"        // raw kube-apiserver client calls (kubectl-equivalent)
 )
 
@@ -40,6 +41,7 @@ type Breakers struct {
 	PostgresAdmin *Breaker
 	RedisAdmin    *Breaker
 	MongoAdmin    *Breaker
+	QueueAdmin    *Breaker
 	K8sAPI        *Breaker
 }
 
@@ -56,6 +58,7 @@ func NewBreakers() *Breakers {
 		PostgresAdmin: newDefault(BackendPostgresAdmin),
 		RedisAdmin:    newDefault(BackendRedisAdmin),
 		MongoAdmin:    newDefault(BackendMongoAdmin),
+		QueueAdmin:    newDefault(BackendQueueAdmin),
 		K8sAPI:        newDefault(BackendK8sAPI),
 	}
 }

internal/server/server.go

@@ -268,6 +268,18 @@ func (s *Server) breakerForMongo(useDedicated bool) *circuit.Breaker {
 	return s.breakers.MongoAdmin
 }
 
+// breakerForQueue — the shared NATS backend issues account/JWT provisioning on
+// the shared cluster (`queue_admin`); the dedicated k8s backend issues
+// kube-apiserver calls (`k8s_api`), the same split as Mongo. Queue was the only
+// backend with no breaker (sweep #2): a wedged NATS / kube-apiserver call could
+// pile up with no fast-fail and no /readyz signal.
+func (s *Server) breakerForQueue(useDedicated bool) *circuit.Breaker {
+	if useDedicated {
+		return s.breakers.K8sAPI
+	}
+	return s.breakers.QueueAdmin
+}
+
 // callBackend wraps a backend invocation behind the given breaker. Returns
 // circuit.ErrOpen verbatim when the breaker is open so the caller can map
 // it to a gRPC Unavailable via mapError. The breaker's caller-deadline
@@ -584,7 +596,9 @@ func (s *Server) provisionQueue(ctx context.Context, req *provisionerv1.Provisio
 	// Pro and team tiers with a dedicated k8s backend get their own NATS pod.
 	if isDedicatedTier(req.Tier) && s.dedicatedQueueBackend != nil {
 		slog.Info("server.provisionQueue: using dedicated backend", "token", req.Token, "tier", req.Tier)
-		creds, err := s.dedicatedQueueBackend.Provision(ctx, req.Token, req.Tier)
+		creds, err := callBackend(s.breakerForQueue(true), func() (*queue.Credentials, error) {
+			return s.dedicatedQueueBackend.Provision(ctx, req.Token, req.Tier)
+		})
 		if err != nil {
 			return nil, mapError("ProvisionResource.queue.dedicated", err)
 		}
@@ -611,7 +625,9 @@ func (s *Server) provisionQueue(ctx context.Context, req *provisionerv1.Provisio
 	}
 
 	slog.Info("server.provisionQueue: pool miss, provisioning live", "token", req.Token)
-	creds, err := s.queueBackend.Provision(ctx, req.Token, req.Tier)
+	creds, err := callBackend(s.breakerForQueue(false), func() (*queue.Credentials, error) {
+		return s.queueBackend.Provision(ctx, req.Token, req.Tier)
+	})
 	if err != nil {
 		return nil, mapError("ProvisionResource.queue", err)
 	}
@@ -718,12 +734,16 @@ func (s *Server) DeprovisionResource(ctx context.Context, req *provisionerv1.Dep
 			!strings.HasPrefix(req.ProviderResourceId, "instant-customer-") {
 			slog.Info("server.DeprovisionResource: queue using dedicated backend",
 				"token", req.Token, "provider_resource_id", req.ProviderResourceId)
-			if err := s.dedicatedQueueBackend.Deprovision(ctx, req.Token, req.ProviderResourceId); err != nil {
+			if err := callBackendVoid(s.breakerForQueue(true), func() error {
+				return s.dedicatedQueueBackend.Deprovision(ctx, req.Token, req.ProviderResourceId)
+			}); err != nil {
 				return nil, mapError("DeprovisionResource.queue.dedicated", err)
 			}
 			return &provisionerv1.DeprovisionResponse{Deprovisioned: true}, nil
 		}
-		if err := s.queueBackend.Deprovision(ctx, req.Token, req.ProviderResourceId); err != nil {
+		if err := callBackendVoid(s.breakerForQueue(false), func() error {
+			return s.queueBackend.Deprovision(ctx, req.Token, req.ProviderResourceId)
+		}); err != nil {
 			return nil, mapError("DeprovisionResource.queue", err)
 		}
 		return &provisionerv1.DeprovisionResponse{Deprovisioned: true}, nil

internal/server/server_coverage_test.go

@@ -21,6 +21,8 @@ import (
 	"google.golang.org/grpc/status"
 	"google.golang.org/grpc/test/bufconn"
 
+	commonv1 "instant.dev/proto/common/v1"
+	provisionerv1 "instant.dev/proto/provisioner/v1"
 	"instant.dev/provisioner/internal/backend/mongo"
 	"instant.dev/provisioner/internal/backend/postgres"
 	"instant.dev/provisioner/internal/backend/queue"
@@ -29,8 +31,6 @@ import (
 	"instant.dev/provisioner/internal/config"
 	"instant.dev/provisioner/internal/pool"
 	"instant.dev/provisioner/internal/server"
-	commonv1 "instant.dev/proto/common/v1"
-	provisionerv1 "instant.dev/proto/provisioner/v1"
 )
 
 // --- fake pool claimer for exercising the s.pool != nil branches ---
@@ -342,6 +342,58 @@ func TestProvisionResource_Queue_DedicatedFailure_ReturnsError(t *testing.T) {
 	assertCode(t, err, codes.InvalidArgument)
 }
 
+// countingFailQueueBackend records its call count so a breaker short-circuit
+// (the tripped breaker must NOT invoke the backend) can be asserted. The error
+// is non-retryable so mapError classifies it Internal — distinguishable from
+// the Unavailable an open breaker returns.
+type countingFailQueueBackend struct{ calls int }
+
+func (b *countingFailQueueBackend) Provision(context.Context, string, string) (*queue.Credentials, error) {
+	b.calls++
+	return nil, errors.New("permission denied: nsc account create")
+}
+
+func (b *countingFailQueueBackend) Deprovision(context.Context, string, string) error {
+	b.calls++
+	return errors.New("permission denied: nsc account delete")
+}
+
+// TestServer_QueueCircuitTripsOnRepeatedFailures — sweep #2: queue/NATS was the
+// only backend with no circuit breaker. After the default threshold of shared
+// queue failures the QueueAdmin breaker must open and the next Provision must
+// short-circuit to Unavailable WITHOUT invoking the backend.
+func TestServer_QueueCircuitTripsOnRepeatedFailures(t *testing.T) {
+	qb := &countingFailQueueBackend{}
+	srv := server.NewWithBackends(
+		&config.Config{},
+		&mockPostgresBackend{}, &mockRedisBackend{}, &mockMongoBackend{}, qb,
+		nil, nil, nil, nil, nil, nil,
+	)
+	srv.SetBreakers(freshBreakers())
+
+	for i := 0; i < 5; i++ {
+		_, err := srv.ProvisionResource(context.Background(), &provisionerv1.ProvisionRequest{
+			Token: "tok", Tier: "hobby", ResourceType: commonv1.ResourceType_RESOURCE_TYPE_QUEUE,
+		})
+		if err == nil {
+			t.Fatalf("attempt %d: expected queue backend error, got nil", i+1)
+		}
+	}
+	if qb.calls != 5 {
+		t.Fatalf("queue backend should have been called 5 times before tripping, got %d", qb.calls)
+	}
+
+	// 6th call: the QueueAdmin breaker is open → short-circuit to Unavailable
+	// without touching the backend.
+	_, err := srv.ProvisionResource(context.Background(), &provisionerv1.ProvisionRequest{
+		Token: "tok", Tier: "hobby", ResourceType: commonv1.ResourceType_RESOURCE_TYPE_QUEUE,
+	})
+	assertCode(t, err, codes.Unavailable)
+	if qb.calls != 5 {
+		t.Fatalf("open QueueAdmin breaker should have short-circuited; backend called %d times (expected 5)", qb.calls)
+	}
+}
+
 // Dedicated provision error → mapError'd.
 func TestProvisionResource_Redis_DedicatedFailure_ReturnsError(t *testing.T) {
 	srv := server.NewWithBackends(

main.go

@@ -5,14 +5,14 @@
 // Observability — relocated 2026-05-12 from the api repo's reference scaffold
 // (track B2 of the observability rollout). What this file wires up:
 //
-//   1. slog default handler decorated with instant.dev/common/logctx so every
-//      log line carries service / commit_id / trace_id / tid / team_id.
-//   2. New Relic Go agent (fail-open: an unset NEW_RELIC_LICENSE_KEY logs a
-//      warning and returns nil; a nil app is safe to pass to nrgrpc).
-//   3. nrgrpc.UnaryServerInterceptor chained with a trace-id stamper so
-//      W3C-propagated trace IDs reach downstream slog calls in handlers.
-//   4. HTTP sidecar on :8092 exposing /healthz with build metadata JSON.
-//      Same shape as api and worker /healthz so a single jq filter works.
+//  1. slog default handler decorated with instant.dev/common/logctx so every
+//     log line carries service / commit_id / trace_id / tid / team_id.
+//  2. New Relic Go agent (fail-open: an unset NEW_RELIC_LICENSE_KEY logs a
+//     warning and returns nil; a nil app is safe to pass to nrgrpc).
+//  3. nrgrpc.UnaryServerInterceptor chained with a trace-id stamper so
+//     W3C-propagated trace IDs reach downstream slog calls in handlers.
+//  4. HTTP sidecar on :8092 exposing /healthz with build metadata JSON.
+//     Same shape as api and worker /healthz so a single jq filter works.
 package main
 
 import (
@@ -77,6 +77,7 @@ func collectBreakerInspectors(bs *circuit.Breakers) []handlers.CircuitInspector
 		breakerAdapter{b: bs.PostgresK8s},
 		breakerAdapter{b: bs.RedisAdmin},
 		breakerAdapter{b: bs.MongoAdmin},
+		breakerAdapter{b: bs.QueueAdmin},
 		breakerAdapter{b: bs.K8sAPI},
 	}
 }

main_test.go

@@ -431,6 +431,7 @@ func TestCollectBreakerInspectors_AllBackendsSurfaced(t *testing.T) {
 		"postgres_k8s":   false,
 		"redis_admin":    false,
 		"mongo_admin":    false,
+		"queue_admin":    false,
 		"k8s_api":        false,
 	}
 	for _, ins := range got {