feat(serve): add circuit-breaker middleware for deployment handles

[glatinone]

Introduce CircuitBreakerMiddleware under python/ray/serve/_private that guards deployment handle calls with configurable consecutive-error threshold and cooldown. Includes tests covering Closed→Open→Half-Open transitions and fail-fast behavior. Exported via _private.init for internal use.

[gemini-code-assist]

Code Review

This pull request introduces a new CircuitBreakerMiddleware and its corresponding state representation to guard deployment handle calls in Ray Serve, along with comprehensive unit tests. The reviewer feedback focuses on improving the implementation's robustness and code quality. Key recommendations include adding thread safety using a threading.Lock to prevent race conditions, defining a custom CircuitBreakerOpenError exception instead of raising a generic RuntimeError, and utilizing an Enum for circuit breaker states to avoid typos and improve type safety.

[gemini-code-assist]

Ray Serve deployment handles can be called concurrently from multiple threads or async tasks. To prevent race conditions during state transitions and counter updates, we should initialize a threading.Lock to synchronize access to the circuit breaker state.

Suggested change

	def __init__(
	self,
	*,
	error_threshold: int = 5,
	cooldown_seconds: float = 10.0,
	) -> None:
	self._threshold = max(1, error_threshold)
	self._cooldown = max(0.0, cooldown_seconds)
	self._state = CircuitBreakerState()
	def __init__(
	self,
	*,
	error_threshold: int = 5,
	cooldown_seconds: float = 10.0,
	) -> None:
	self._threshold = max(1, error_threshold)
	self._cooldown = max(0.0, cooldown_seconds)
	self._state = CircuitBreakerState()
	self._lock = threading.Lock()

[gemini-code-assist]

Acquire the lock to ensure thread-safe state checks and transitions, and raise the custom CircuitBreakerOpenError exception instead of RuntimeError.

Suggested change

	def before_call(self) -> None:
	if self._state.state == "open":
	if self._now() - self._state.opened_at >= self._cooldown:
	self._state.state = "half_open"
	self._state.failure_count = 0
	else:
	raise RuntimeError("CircuitBreaker: open")
	def before_call(self) -> None:
	with self._lock:
	if self._state.state == CircuitBreakerStateEnum.OPEN:
	if self._now() - self._state.opened_at >= self._cooldown:
	self._state.state = CircuitBreakerStateEnum.HALF_OPEN
	self._state.failure_count = 0
	else:
	raise CircuitBreakerOpenError("CircuitBreaker: open")

[gemini-code-assist]

Acquire the lock to ensure thread-safe state updates when recording a successful call.

Suggested change

	def record_success(self) -> None:
	if self._state.state in ("half_open", "open"):
	self._state = CircuitBreakerState()
	else:
	self._state.failure_count = 0
	def record_success(self) -> None:
	with self._lock:
	if self._state.state in (
	CircuitBreakerStateEnum.HALF_OPEN,
	CircuitBreakerStateEnum.OPEN,
	):
	self._state = CircuitBreakerState()
	else:
	self._state.failure_count = 0

[gemini-code-assist]

Acquire the lock to ensure thread-safe state updates and counter increments when recording a failed call.

Suggested change

	def record_failure(self) -> None:
	if self._state.state == "half_open":
	self._state.state = "open"
	self._state.opened_at = self._now()
	self._state.failure_count = 1
	return
	self._state.failure_count += 1
	if self._state.failure_count >= self._threshold:
	self._state.state = "open"
	self._state.opened_at = self._now()
	def record_failure(self) -> None:
	with self._lock:
	if self._state.state == CircuitBreakerStateEnum.HALF_OPEN:
	self._state.state = CircuitBreakerStateEnum.OPEN
	self._state.opened_at = self._now()
	self._state.failure_count = 1
	return
	self._state.failure_count += 1
	if self._state.failure_count >= self._threshold:
	self._state.state = CircuitBreakerStateEnum.OPEN
	self._state.opened_at = self._now()

[gemini-code-assist]

To improve code quality, maintainability, and robust error handling, we should:

1. Define a custom exception class (CircuitBreakerOpenError) instead of raising a generic RuntimeError.
2. Use an Enum (CircuitBreakerStateEnum) for the circuit breaker states instead of raw string literals to prevent typos and enable static type checking.
3. Import threading to support thread-safe operations.

from __future__ import annotations

from enum import Enum
import threading
import time
from dataclasses import dataclass
from typing import Optional


class CircuitBreakerOpenError(RuntimeError):
    """Raised when a call is made while the circuit breaker is open."""
    pass


class CircuitBreakerStateEnum(str, Enum):
    CLOSED = "closed"
    OPEN = "open"
    HALF_OPEN = "half_open"


@dataclass
class CircuitBreakerState:
    failure_count: int = 0
    state: CircuitBreakerStateEnum = CircuitBreakerStateEnum.CLOSED
    opened_at: float = 0.0

[gemini-code-assist]

To prevent race conditions when external callers read the state object while it is being modified by another thread, return a copy of the state under the lock.

Suggested change

	@property
	def state(self) -> CircuitBreakerState:
	return self._state
	@property
	def state(self) -> CircuitBreakerState:
	with self._lock:
	return CircuitBreakerState(
	failure_count=self._state.failure_count,
	state=self._state.state,
	opened_at=self._state.opened_at,
	)

[gemini-code-assist]

Import the new CircuitBreakerOpenError custom exception to use in tests.

Suggested change

	from ray.serve._private.circuit_breaker import CircuitBreakerMiddleware
	from ray.serve._private.circuit_breaker import CircuitBreakerMiddleware, CircuitBreakerOpenError

[gemini-code-assist]

Assert that CircuitBreakerOpenError is raised instead of a generic RuntimeError.

Suggested change

	with pytest.raises(RuntimeError):
	cb.before_call()
	with pytest.raises(CircuitBreakerOpenError):
	cb.before_call()

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 3 potential issues.

^{Reviewed by Cursor Bugbot for commit cabccb0. Configure here.}

[cursor]

Half-open allows multiple probes

High Severity

before_call only fail-fasts while state is open. Once it transitions to half_open, further calls pass through with no probe limit, even though the docstring says half-open should allow a single probe. After cooldown, traffic can stampede a still-failing deployment.

 

^{Reviewed by Cursor Bugbot for commit cabccb0. Configure here.}

[cursor]

Success resets open circuit

High Severity

record_success treats open like half_open and fully resets the breaker. In-flight calls that started while closed can still succeed after later failures have opened the circuit, immediately closing it again and re-admitting traffic to a dependency that just tripped the threshold.

 

^{Reviewed by Cursor Bugbot for commit cabccb0. Configure here.}

[cursor]

Tests omitted from Bazel CI

Medium Severity

test_circuit_breaker.py was added under python/ray/serve/tests/ but not listed in that package’s BUILD.bazel. Unlike tests/unit/ (globbed automatically), top-level serve tests are explicit, so these transition and fail-fast cases will not run in Ray’s Bazel CI.

 

^{Reviewed by Cursor Bugbot for commit cabccb0. Configure here.}