Log outbound queue circuit breaker state changes (#11661)

### What changed
Sets gobreaker's OnStateChange hook on the outbound queue circuit
breaker pool, logging every transition.

### Why
Obtain more details for debugging curcuit breaker in production.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: stephanos

chasm/lib/callback/invocable_outbound.go

@@ -46,7 +46,7 @@ func (n invocableOutbound) Invoke(
 		traceLogger := log.With(h.logger,
 			tag.WorkflowNamespace(ns.Name().String()),
 			tag.Operation("CompleteNexusOperation"),
-			tag.String("destination", taskAttr.Destination),
+			tag.Destination(taskAttr.Destination),
 			tag.WorkflowID(n.workflowID),
 			tag.WorkflowRunID(n.runID),
 			tag.AttemptStart(time.Now().UTC()),

common/log/tag/tags.go

@@ -989,6 +989,11 @@ func Endpoint(endpoint string) ZapTag {
 	return NewStringTag("endpoint", endpoint)
 }
 
+// Destination returns a tag for an outbound task's destination.
+func Destination(destination string) ZapTag {
+	return NewStringTag("destination", destination)
+}
+
 func BuildId(buildId string) ZapTag {
 	return NewStringTag("build-id", buildId)
 }

components/callbacks/nexus_invocation.go

@@ -47,7 +47,7 @@ func (n nexusInvocation) Invoke(ctx context.Context, ns *namespace.Namespace, e
 		traceLogger := log.With(e.Logger,
 			tag.WorkflowNamespace(ns.Name().String()),
 			tag.Operation("CompleteNexusOperation"),
-			tag.String("destination", task.destination),
+			tag.Destination(task.destination),
 			tag.WorkflowID(n.workflowID),
 			tag.WorkflowRunID(n.runID),
 			tag.AttemptStart(time.Now().UTC()),

service/history/circuitbreakerpool/fx.go

@@ -3,7 +3,10 @@ package circuitbreakerpool
 import (
 	"fmt"
 
+	"github.com/sony/gobreaker"
 	"go.temporal.io/server/common/circuitbreaker"
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/log/tag"
 	"go.temporal.io/server/common/namespace"
 	"go.temporal.io/server/service/history/configs"
 	"go.temporal.io/server/service/history/tasks"
@@ -21,6 +24,7 @@ type OutboundQueueCircuitBreakerPool struct {
 func OutboundQueueCircuitBreakerPoolProvider(
 	namespaceRegistry namespace.Registry,
 	config *configs.Config,
+	logger log.SnTaggedLogger,
 ) *OutboundQueueCircuitBreakerPool {
 	return &OutboundQueueCircuitBreakerPool{
 		CircuitBreakerPool: NewCircuitBreakerPool(
@@ -37,6 +41,7 @@ func OutboundQueueCircuitBreakerPoolProvider(
 						key.NamespaceID,
 						key.Destination,
 					),
+					OnStateChange: onStateChange(key, nsName.String(), logger),
 				})
 				initial, cancel := config.OutboundQueueCircuitBreakerSettings(
 					nsName.String(),
@@ -50,3 +55,22 @@ func OutboundQueueCircuitBreakerPoolProvider(
 		),
 	}
 }
+
+// onStateChange logs breaker transitions.
+func onStateChange(
+	key tasks.TaskGroupNamespaceIDAndDestination,
+	nsName string,
+	logger log.SnTaggedLogger,
+) func(name string, from gobreaker.State, to gobreaker.State) {
+	return func(_ string, from gobreaker.State, to gobreaker.State) {
+		logger.Warn(
+			"outbound queue circuit breaker state change",
+			tag.WorkflowNamespace(nsName),
+			tag.WorkflowNamespaceID(key.NamespaceID),
+			tag.Destination(key.Destination),
+			tag.NewStringTag("task-group", key.TaskGroup),
+			tag.NewStringTag("from-state", from.String()),
+			tag.NewStringTag("to-state", to.String()),
+		)
+	}
+}