feat: add breakerpool all-down fallback and expose CA status in inspect

When all circuit breakers are open, bypass them and try entries directly
instead of returning AllUnavailableError immediately. This fixes the case
where a transient outage (e.g., no network) trips all breakers, locking
out all CAs for the full 10-minute cooldown even after recovery.

Also expose per-CA-endpoint circuit breaker state (healthy/broken) in
`epithet agent inspect` output, both human-readable and JSON.

Author: brianm

cmd/epithet/inspect.go

@@ -82,6 +82,21 @@ func (i *AgentInspectCLI) Run(parent *AgentCLI, logger *slog.Logger) error {
 	fmt.Printf("Agent Dir: %s\n", resp.AgentSocketDir)
 	fmt.Printf("Discovery Patterns: %v\n\n", resp.DiscoveryPatterns)
 
+	fmt.Printf("CA Endpoints (%d)\n", len(resp.CaEndpoints))
+	fmt.Printf("-----------------\n")
+	if len(resp.CaEndpoints) == 0 {
+		fmt.Printf("  (none)\n")
+	} else {
+		for _, ep := range resp.CaEndpoints {
+			label := "healthy"
+			if ep.State == "open" || ep.State == "half-open" {
+				label = "broken"
+			}
+			fmt.Printf("  %s (priority %d) — %s\n", ep.Url, ep.Priority, label)
+		}
+	}
+	fmt.Println()
+
 	fmt.Printf("Agents (%d)\n", len(resp.Agents))
 	fmt.Printf("-----------\n")
 	if len(resp.Agents) == 0 {
@@ -124,11 +139,18 @@ func (i *AgentInspectCLI) Run(parent *AgentCLI, logger *slog.Logger) error {
 
 // inspectResponseJSON is a simplified structure for JSON output.
 type inspectResponseJSON struct {
-	SocketPath        string          `json:"socketPath"`
-	AgentSocketDir    string          `json:"agentSocketDir"`
-	DiscoveryPatterns []string        `json:"discoveryPatterns,omitempty"`
-	Agents            []agentInfoJSON `json:"agents"`
-	Certificates      []certInfoJSON  `json:"certificates"`
+	SocketPath        string               `json:"socketPath"`
+	AgentSocketDir    string               `json:"agentSocketDir"`
+	DiscoveryPatterns []string             `json:"discoveryPatterns,omitempty"`
+	Agents            []agentInfoJSON      `json:"agents"`
+	Certificates      []certInfoJSON       `json:"certificates"`
+	CAEndpoints       []caEndpointInfoJSON `json:"caEndpoints"`
+}
+
+type caEndpointInfoJSON struct {
+	URL      string `json:"url"`
+	Priority int32  `json:"priority"`
+	State    string `json:"state"`
 }
 
 type agentInfoJSON struct {
@@ -168,12 +190,22 @@ func inspectResponseToJSON(resp *pb.InspectResponse) inspectResponseJSON {
 		}
 	}
 
+	caEndpoints := make([]caEndpointInfoJSON, len(resp.CaEndpoints))
+	for i, ep := range resp.CaEndpoints {
+		caEndpoints[i] = caEndpointInfoJSON{
+			URL:      ep.Url,
+			Priority: ep.Priority,
+			State:    ep.State,
+		}
+	}
+
 	return inspectResponseJSON{
 		SocketPath:        resp.SocketPath,
 		AgentSocketDir:    resp.AgentSocketDir,
 		DiscoveryPatterns: resp.DiscoveryPatterns,
 		Agents:            agents,
 		Certificates:      certs,
+		CAEndpoints:       caEndpoints,
 	}
 }
 

pkg/breakerpool/pool.go

@@ -175,6 +175,19 @@ func (p *Pool[T, S]) Execute(fn func(S) (T, error)) (T, error) {
 		// Get next available entry index
 		idx := p.next()
 		if idx < 0 {
+			// All breakers open — try all entries directly, bypassing circuit breakers.
+			// When every endpoint is marked down, breaker protection has no value
+			// (nothing to fail over to). This allows recovery from transient outages
+			// without waiting for the full cooldown.
+			for _, tier := range p.tiers {
+				for _, tidx := range tier {
+					result, err := fn(p.entries[tidx].State)
+					if err == nil {
+						return result, nil
+					}
+					lastErr = err
+				}
+			}
 			return zero, &AllUnavailableError{LastError: lastErr}
 		}
 
@@ -264,6 +277,38 @@ func (p *Pool[T, S]) AllUnavailable() bool {
 	return true
 }
 
+// EndpointStatus represents the current state of a pool entry.
+type EndpointStatus[S any] struct {
+	State        S
+	Priority     int
+	BreakerState string // "closed", "open", "half-open"
+}
+
+// Status returns the current state of each entry in priority order.
+func (p *Pool[T, S]) Status() []EndpointStatus[S] {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	statuses := make([]EndpointStatus[S], len(p.entries))
+	for i, entry := range p.entries {
+		var state string
+		switch p.breakers[i].State() {
+		case gobreaker.StateClosed:
+			state = "closed"
+		case gobreaker.StateOpen:
+			state = "open"
+		case gobreaker.StateHalfOpen:
+			state = "half-open"
+		}
+		statuses[i] = EndpointStatus[S]{
+			State:        entry.State,
+			Priority:     entry.Priority,
+			BreakerState: state,
+		}
+	}
+	return statuses
+}
+
 // Len returns the number of entries in the pool.
 func (p *Pool[T, S]) Len() int {
 	return len(p.entries)

pkg/breakerpool/pool_test.go

@@ -305,6 +305,130 @@ func TestPool_PerEntrySettings(t *testing.T) {
 	require.Len(t, calls, 2)
 }
 
+func TestPool_Execute_AllOpenFallbackRecovers(t *testing.T) {
+	entries := []Entry[testState]{
+		{State: testState{url: "https://ca1.example.com"}, Priority: 100},
+		{State: testState{url: "https://ca2.example.com"}, Priority: 50},
+	}
+	pool := New[string](entries, testDefaults(time.Minute))
+
+	// Trip both breakers.
+	_, _ = pool.Execute(func(s testState) (string, error) {
+		return "", &infraError{msg: "down"}
+	})
+	require.True(t, pool.AllUnavailable())
+
+	// Network recovers — fallback should try directly and succeed.
+	result, err := pool.Execute(func(s testState) (string, error) {
+		return "recovered", nil
+	})
+	require.NoError(t, err)
+	require.Equal(t, "recovered", result)
+}
+
+func TestPool_Execute_AllOpenFallbackStillFails(t *testing.T) {
+	entries := []Entry[testState]{
+		{State: testState{url: "https://ca1.example.com"}, Priority: 100},
+		{State: testState{url: "https://ca2.example.com"}, Priority: 50},
+	}
+	pool := New[string](entries, testDefaults(time.Minute))
+
+	// Trip both breakers.
+	_, _ = pool.Execute(func(s testState) (string, error) {
+		return "", &infraError{msg: "down"}
+	})
+
+	// Still failing — should return AllUnavailableError.
+	_, err := pool.Execute(func(s testState) (string, error) {
+		return "", &infraError{msg: "still down"}
+	})
+	require.Error(t, err)
+	var allUnavail *AllUnavailableError
+	require.ErrorAs(t, err, &allUnavail)
+	require.Contains(t, allUnavail.Error(), "still down")
+}
+
+func TestPool_Execute_AllOpenFallbackPriorityOrder(t *testing.T) {
+	entries := []Entry[testState]{
+		{State: testState{url: "https://ca-low.example.com"}, Priority: 50},
+		{State: testState{url: "https://ca-high.example.com"}, Priority: 100},
+	}
+	pool := New[string](entries, testDefaults(time.Minute))
+
+	// Trip both breakers.
+	_, _ = pool.Execute(func(s testState) (string, error) {
+		return "", &infraError{msg: "down"}
+	})
+
+	// Track call order in fallback.
+	var calls []string
+	_, _ = pool.Execute(func(s testState) (string, error) {
+		calls = append(calls, s.url)
+		return "", &infraError{msg: "down"}
+	})
+
+	// Should try high priority first, then low.
+	require.Equal(t, []string{"https://ca-high.example.com", "https://ca-low.example.com"}, calls)
+}
+
+func TestPool_Execute_AllOpenFallbackFirstSuccessWins(t *testing.T) {
+	entries := []Entry[testState]{
+		{State: testState{url: "https://ca1.example.com"}, Priority: 100},
+		{State: testState{url: "https://ca2.example.com"}, Priority: 100},
+		{State: testState{url: "https://ca3.example.com"}, Priority: 50},
+	}
+	pool := New[string](entries, testDefaults(time.Minute))
+
+	// Trip all breakers.
+	for i := 0; i < 3; i++ {
+		_, _ = pool.Execute(func(s testState) (string, error) {
+			return "", &infraError{msg: "down"}
+		})
+	}
+	require.True(t, pool.AllUnavailable())
+
+	// Second entry succeeds — third should not be tried.
+	var calls []string
+	result, err := pool.Execute(func(s testState) (string, error) {
+		calls = append(calls, s.url)
+		if s.url == "https://ca2.example.com" {
+			return "from-ca2", nil
+		}
+		return "", &infraError{msg: "down"}
+	})
+	require.NoError(t, err)
+	require.Equal(t, "from-ca2", result)
+	require.Equal(t, []string{"https://ca1.example.com", "https://ca2.example.com"}, calls)
+}
+
+func TestPool_Status(t *testing.T) {
+	entries := []Entry[testState]{
+		{State: testState{url: "https://ca1.example.com"}, Priority: 100},
+		{State: testState{url: "https://ca2.example.com"}, Priority: 50},
+	}
+	pool := New[string](entries, testDefaults(time.Minute))
+
+	// Both should start closed.
+	statuses := pool.Status()
+	require.Len(t, statuses, 2)
+	require.Equal(t, "closed", statuses[0].BreakerState)
+	require.Equal(t, "closed", statuses[1].BreakerState)
+	require.Equal(t, 100, statuses[0].Priority)
+	require.Equal(t, 50, statuses[1].Priority)
+
+	// Trip the high-priority breaker.
+	_, _ = pool.Execute(func(s testState) (string, error) {
+		if s.url == "https://ca1.example.com" {
+			return "", &infraError{msg: "down"}
+		}
+		return "ok", nil
+	})
+
+	statuses = pool.Status()
+	require.Equal(t, "open", statuses[0].BreakerState)
+	require.Equal(t, "closed", statuses[1].BreakerState)
+}
+
 func TestPool_Execute_StringState(t *testing.T) {
 	// Test with simple string state (common case)
 	entries := []Entry[string]{

pkg/broker/broker.go

@@ -190,13 +190,21 @@ type CertInfo struct {
 	ExpiresAt   time.Time              `json:"expiresAt"`
 }
 
+// CAEndpointInfo contains the current state of a CA endpoint.
+type CAEndpointInfo struct {
+	URL      string `json:"url"`
+	Priority int    `json:"priority"`
+	State    string `json:"state"` // "closed", "open", "half-open"
+}
+
 // InspectResponse contains the current broker state
 type InspectResponse struct {
-	SocketPath        string      `json:"socketPath"`
-	AgentSocketDir    string      `json:"agentSocketDir"`
-	DiscoveryPatterns []string    `json:"discoveryPatterns,omitempty"` // Fetched live from CA (HTTP cached)
-	Agents            []AgentInfo `json:"agents"`
-	Certificates      []CertInfo  `json:"certificates"`
+	SocketPath        string           `json:"socketPath"`
+	AgentSocketDir    string           `json:"agentSocketDir"`
+	DiscoveryPatterns []string         `json:"discoveryPatterns,omitempty"` // Fetched live from CA (HTTP cached)
+	Agents            []AgentInfo      `json:"agents"`
+	Certificates      []CertInfo       `json:"certificates"`
+	CAEndpoints       []CAEndpointInfo `json:"caEndpoints"`
 }
 
 // Match is invoked via rpc from `epithet match` invocations.
@@ -701,5 +709,14 @@ func (b *Broker) Inspect(_ InspectRequest, output *InspectResponse) error {
 	// Get certificate info
 	output.Certificates = b.certStore.List()
 
+	// Get CA endpoint status
+	for _, ep := range b.caClient.EndpointStatus() {
+		output.CAEndpoints = append(output.CAEndpoints, CAEndpointInfo{
+			URL:      ep.URL,
+			Priority: ep.Priority,
+			State:    ep.BreakerState,
+		})
+	}
+
 	return nil
 }

pkg/broker/grpc_server.go

@@ -104,12 +104,22 @@ func inspectResponseToProto(resp *InspectResponse) *pb.InspectResponse {
 		}
 	}
 
+	caEndpoints := make([]*pb.CAEndpointInfo, len(resp.CAEndpoints))
+	for i, ep := range resp.CAEndpoints {
+		caEndpoints[i] = &pb.CAEndpointInfo{
+			Url:      ep.URL,
+			Priority: int32(ep.Priority),
+			State:    ep.State,
+		}
+	}
+
 	return &pb.InspectResponse{
 		SocketPath:        resp.SocketPath,
 		AgentSocketDir:    resp.AgentSocketDir,
 		DiscoveryPatterns: resp.DiscoveryPatterns,
 		Agents:            agents,
 		Certificates:      certs,
+		CaEndpoints:       caEndpoints,
 	}
 }