diff --git a/adapters/breaker.py b/adapters/breaker.py index 1555142..f3506c2 100644 --- a/adapters/breaker.py +++ b/adapters/breaker.py @@ -20,21 +20,32 @@ class CircuitBreaker: self._cooldown_s = cooldown_s self._fails: dict[str, int] = {} self._open_until: dict[str, float] = {} + self._half_open_inflight: dict[str, bool] = {} def is_open(self, source_name: str, now: float) -> bool: """判断指定源是否处于开路状态。 - 冷却截止时刻之前为开路;到期返回 False(放行一个试探,即半开)。 + 冷却截止时刻之前为开路;到期进入半开,**只放行一个探针**(其余仍被挡), + 避免冷却到期瞬间惊群重连再次压垮上游。"检查+标记探针"在 asyncio 单线程内 + 同步执行,天然原子无竞态。 Args: source_name: 被熔断的源标识。 now: 当前时刻(秒级时间戳),由调用方注入。 Returns: - True 表示开路(拒绝请求),False 表示关闭或半开(放行)。 + True 表示开路(拒绝请求),False 表示关闭或半开放行探针。 """ until = self._open_until.get(source_name) - return until is not None and now < until + if until is None: + return False + if now < until: + return True # 冷却中,全挡 + # 冷却到期:half-open,只放行一个探针 + if self._half_open_inflight.get(source_name): + return True # 已有探针在途,继续挡 + self._half_open_inflight[source_name] = True + return False def record_failure(self, source_name: str, now: float) -> None: """记录一次失败;累计达阈值则开路至 now + cooldown。 @@ -47,6 +58,8 @@ class CircuitBreaker: self._fails[source_name] = count if count >= self._fail_threshold: self._open_until[source_name] = now + self._cooldown_s + # 探针失败清在途标记,使下一轮 cooldown 到期后可再放行探针 + self._half_open_inflight.pop(source_name, None) def force_open(self, source_name: str, now: float) -> None: """强制开路(用于 401/403 等不可恢复错误),一次即熔断。 @@ -59,6 +72,7 @@ class CircuitBreaker: """ self._fails[source_name] = self._fail_threshold self._open_until[source_name] = now + self._cooldown_s + self._half_open_inflight.pop(source_name, None) def record_success(self, source_name: str) -> None: """记录一次成功;清零失败计数与开路状态(关闭熔断器)。 @@ -68,3 +82,4 @@ class CircuitBreaker: """ self._fails.pop(source_name, None) self._open_until.pop(source_name, None) + self._half_open_inflight.pop(source_name, None) diff --git a/tests/unit/test_breaker.py b/tests/unit/test_breaker.py index 348c2da..345c635 100644 --- a/tests/unit/test_breaker.py +++ b/tests/unit/test_breaker.py @@ -58,6 +58,18 @@ class TestCircuitBreaker: breaker.force_open("llm", now=5.0) assert breaker.is_open("llm", now=5.5) is True + def test_half_open_admits_single_probe(self) -> None: + """冷却到期后半开只放行一个探针,第二个仍被挡;探针成功后闭合。""" + b = CircuitBreaker(fail_threshold=2, cooldown_s=10.0) + b.record_failure("p", now=0.0) + b.record_failure("p", now=0.0) # 开路至 t=10 + assert b.is_open("p", now=5.0) is True # 冷却中 + # 冷却到期:只放行第一个探针 + assert b.is_open("p", now=11.0) is False # 探针 1 放行 + assert b.is_open("p", now=11.0) is True # 探针 2 被挡(探针在途) + b.record_success("p") # 探针成功 → 闭合 + assert b.is_open("p", now=12.0) is False + def test_force_open_probe_failure_reopens(self, breaker: CircuitBreaker) -> None: """半开探针失败后重新熔断(因 fails 已置为 threshold)。""" breaker.force_open("llm", now=0.0)