fix: pin retry_after_s to the next certain retry moment
retry_after_s never had a written definition, so each backend improvised and they drifted apart. It now answers exactly one question: how long until a retry is *certainly* worth attempting. OPEN has such a moment (the cooldown deadline); HALF_OPEN does not, because the probe can come back at any time -- so it reports 0.0, which already means "retry now" elsewhere in the library. Six exits are brought in line. The half-open rejection is the one issue 14 reported: it returned the probe lease remainder, a deadlock-guard value derived from 2x the slowest timeout, so a 60s cooldown told callers to wait 600s. Worse, retry.py fed that number into the source cooldown memo, whose set_until only moves forward -- a source stayed skipped in-process for the whole lease even after its probe succeeded and the gate closed. That now writes an already-expired deadline, so the memo goes back to recording only real OPEN cooldowns. The other five were pre-existing memory/redis divergences hidden by a contract-test blind spot (the suite pinned that a second caller gets rejected, never what number it got): redis reported the probe TTL on grant and the lease remainder on fenced-out writes, where memory has always reported 0. Contract cases now pin all four half-open exits on both backends, with 1:1 real-wait variants for redis since the fake-clock ones skip there.
This commit is contained in:
@@ -100,6 +100,22 @@ class InMemoryGate:
|
|||||||
streak = max(1, g.reopen_streak)
|
streak = max(1, g.reopen_streak)
|
||||||
return min(self._cfg.cooldown_s * (2 ** (streak - 1)), self._cfg.max_cooldown_s)
|
return min(self._cfg.cooldown_s * (2 ** (streak - 1)), self._cfg.max_cooldown_s)
|
||||||
|
|
||||||
|
def _remaining(self, g: _SourceGate) -> float:
|
||||||
|
"""距离**确定**可再试的时刻还有多久(issue #14 的契约定义)。
|
||||||
|
|
||||||
|
OPEN 的冷却截止是确定时刻;HALF_OPEN 下探针随时可能出结果,**不存在**
|
||||||
|
确定时刻,故 `0.0`——`0 = 可立即重试` 是库既有约定。此前这里返回探针
|
||||||
|
租约剩余,而租约长度是死锁保护参数(派生自 `2 × 最慢源 timeout`),与
|
||||||
|
"源多久能恢复"无因果关系;它还被喂进源冷却备忘,而备忘 `set_until`
|
||||||
|
取更晚者不可回退,于是门恢复 CLOSED 后本进程仍跳过该源整整一个租约。
|
||||||
|
|
||||||
|
三个出口(`try_enter` 拒绝、`_snapshot`、`retry_after_s`)共用本方法,
|
||||||
|
避免同一语义在三处各算一遍而漂移。
|
||||||
|
"""
|
||||||
|
if g.state is GateState.OPEN:
|
||||||
|
return max(0.0, g.open_until - self._now())
|
||||||
|
return 0.0
|
||||||
|
|
||||||
def _grant_probe(self, g: _SourceGate, source_name: str, owner: str) -> GateDecision:
|
def _grant_probe(self, g: _SourceGate, source_name: str, owner: str) -> GateDecision:
|
||||||
g.state = GateState.HALF_OPEN
|
g.state = GateState.HALF_OPEN
|
||||||
g.probe_owner = owner
|
g.probe_owner = owner
|
||||||
@@ -140,7 +156,7 @@ class InMemoryGate:
|
|||||||
epoch=g.epoch,
|
epoch=g.epoch,
|
||||||
is_probe=False,
|
is_probe=False,
|
||||||
probe_owner=None,
|
probe_owner=None,
|
||||||
retry_after_s=g.open_until - now,
|
retry_after_s=self._remaining(g),
|
||||||
)
|
)
|
||||||
# HALF_OPEN: 探针在途;租约过期则接管,否则拒绝(防惊群)
|
# HALF_OPEN: 探针在途;租约过期则接管,否则拒绝(防惊群)
|
||||||
if now >= g.probe_expires:
|
if now >= g.probe_expires:
|
||||||
@@ -152,7 +168,7 @@ class InMemoryGate:
|
|||||||
epoch=g.epoch,
|
epoch=g.epoch,
|
||||||
is_probe=False,
|
is_probe=False,
|
||||||
probe_owner=None,
|
probe_owner=None,
|
||||||
retry_after_s=g.probe_expires - now,
|
retry_after_s=self._remaining(g),
|
||||||
)
|
)
|
||||||
|
|
||||||
def _fenced(self, g: _SourceGate, entry: GateDecision) -> bool:
|
def _fenced(self, g: _SourceGate, entry: GateDecision) -> bool:
|
||||||
@@ -172,9 +188,7 @@ class InMemoryGate:
|
|||||||
state=g.state,
|
state=g.state,
|
||||||
epoch=g.epoch,
|
epoch=g.epoch,
|
||||||
failure_count=g.fails,
|
failure_count=g.fails,
|
||||||
retry_after_s=max(0.0, g.open_until - self._now())
|
retry_after_s=self._remaining(g),
|
||||||
if g.state is GateState.OPEN
|
|
||||||
else 0.0,
|
|
||||||
)
|
)
|
||||||
|
|
||||||
def _open(self, g: _SourceGate, reason: str, *, bump_streak: bool) -> None:
|
def _open(self, g: _SourceGate, reason: str, *, bump_streak: bool) -> None:
|
||||||
@@ -258,14 +272,4 @@ class InMemoryGate:
|
|||||||
"""集合中最早可尝试时间;健康/到期返回 0。"""
|
"""集合中最早可尝试时间;健康/到期返回 0。"""
|
||||||
if not sources:
|
if not sources:
|
||||||
raise ValueError("sources 不能为空")
|
raise ValueError("sources 不能为空")
|
||||||
now = self._now()
|
return min(self._remaining(self._gate(name)) for name in sources)
|
||||||
waits = []
|
|
||||||
for name in sources:
|
|
||||||
g = self._gate(name)
|
|
||||||
if g.state is GateState.OPEN:
|
|
||||||
waits.append(max(0.0, g.open_until - now))
|
|
||||||
elif g.state is GateState.HALF_OPEN:
|
|
||||||
waits.append(max(0.0, g.probe_expires - now))
|
|
||||||
else:
|
|
||||||
waits.append(0.0)
|
|
||||||
return min(waits)
|
|
||||||
|
|||||||
@@ -42,7 +42,8 @@ if state == 'open' and now < open_until then
|
|||||||
return {0, state, epoch, 0, '', open_until - now}
|
return {0, state, epoch, 0, '', open_until - now}
|
||||||
end
|
end
|
||||||
if state == 'half_open' and now < probe_until then
|
if state == 'half_open' and now < probe_until then
|
||||||
return {0, state, epoch, 0, '', probe_until - now}
|
-- 探针在途: 无确定的可再试时刻 → 0(issue #14,与 memory `_remaining` 同口径)
|
||||||
|
return {0, state, epoch, 0, '', 0}
|
||||||
end
|
end
|
||||||
|
|
||||||
local next_probe_until = now + tonumber(ARGV[2])
|
local next_probe_until = now + tonumber(ARGV[2])
|
||||||
@@ -50,7 +51,7 @@ redis.call('HSET', KEYS[1],
|
|||||||
'state', 'half_open',
|
'state', 'half_open',
|
||||||
'probe_owner', ARGV[1],
|
'probe_owner', ARGV[1],
|
||||||
'probe_until', next_probe_until)
|
'probe_until', next_probe_until)
|
||||||
return {1, 'half_open', epoch, 1, ARGV[1], tonumber(ARGV[2])}
|
return {1, 'half_open', epoch, 1, ARGV[1], 0}
|
||||||
"""
|
"""
|
||||||
|
|
||||||
# M2.5 窗口/退避公共片段(拼接进 success/failure 脚本;Lua 脚本间无法共享函数)
|
# M2.5 窗口/退避公共片段(拼接进 success/failure 脚本;Lua 脚本间无法共享函数)
|
||||||
@@ -124,8 +125,6 @@ end
|
|||||||
local deadline = 0
|
local deadline = 0
|
||||||
if state == 'open' then
|
if state == 'open' then
|
||||||
deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
||||||
elseif state == 'half_open' then
|
|
||||||
deadline = tonumber(redis.call('HGET', KEYS[1], 'probe_until') or '0')
|
|
||||||
end
|
end
|
||||||
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
||||||
"""
|
"""
|
||||||
@@ -156,8 +155,6 @@ if not matches then
|
|||||||
local deadline = 0
|
local deadline = 0
|
||||||
if state == 'open' then
|
if state == 'open' then
|
||||||
deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
||||||
elseif state == 'half_open' then
|
|
||||||
deadline = tonumber(redis.call('HGET', KEYS[1], 'probe_until') or '0')
|
|
||||||
end
|
end
|
||||||
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
||||||
end
|
end
|
||||||
@@ -255,8 +252,6 @@ end
|
|||||||
local deadline = 0
|
local deadline = 0
|
||||||
if state == 'open' then
|
if state == 'open' then
|
||||||
deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
deadline = tonumber(redis.call('HGET', KEYS[1], 'open_until') or '0')
|
||||||
elseif state == 'half_open' then
|
|
||||||
deadline = tonumber(redis.call('HGET', KEYS[1], 'probe_until') or '0')
|
|
||||||
end
|
end
|
||||||
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
return {0, state, epoch, failures, math.max(deadline - now, 0)}
|
||||||
"""
|
"""
|
||||||
@@ -272,9 +267,6 @@ for _, key in ipairs(KEYS) do
|
|||||||
if state == 'open' then
|
if state == 'open' then
|
||||||
local deadline = tonumber(redis.call('HGET', key, 'open_until') or '0')
|
local deadline = tonumber(redis.call('HGET', key, 'open_until') or '0')
|
||||||
remaining = math.max(deadline - now, 0)
|
remaining = math.max(deadline - now, 0)
|
||||||
elseif state == 'half_open' then
|
|
||||||
local deadline = tonumber(redis.call('HGET', key, 'probe_until') or '0')
|
|
||||||
remaining = math.max(deadline - now, 0)
|
|
||||||
end
|
end
|
||||||
if minimum == nil or remaining < minimum then minimum = remaining end
|
if minimum == nil or remaining < minimum then minimum = remaining end
|
||||||
end
|
end
|
||||||
|
|||||||
@@ -292,6 +292,51 @@ class TestRetryAfter:
|
|||||||
assert await gate.retry_after_s(("s1", "s2")) == 0.0
|
assert await gate.retry_after_s(("s1", "s2")) == 0.0
|
||||||
|
|
||||||
|
|
||||||
|
async def test_half_open_rejection_reports_no_certain_wait(self, gate_factory, clock):
|
||||||
|
"""探针在途时被拒 → 0.0(issue #14): 探针随时可能出结果,不存在确定时刻。
|
||||||
|
|
||||||
|
旧行为返回探针租约剩余,而租约长度是**死锁保护参数**(派生自
|
||||||
|
`2 × 最慢源 timeout`),与"这个源多久能恢复"没有因果关系。现场
|
||||||
|
`TIMEOUT_S=300` 时它是 600s,而冷却期只有 60s。
|
||||||
|
"""
|
||||||
|
gate = gate_factory(_CFG)
|
||||||
|
await _open_gate(gate)
|
||||||
|
clock.advance(_CFG.cooldown_s + 1)
|
||||||
|
probe = await gate.try_enter("s1", "w1")
|
||||||
|
assert probe.is_probe
|
||||||
|
blocked = await gate.try_enter("s1", "w2")
|
||||||
|
assert not blocked.allowed and blocked.state is GateState.HALF_OPEN
|
||||||
|
assert blocked.retry_after_s == 0.0
|
||||||
|
|
||||||
|
async def test_probe_grant_reports_no_certain_wait(self, gate_factory, clock):
|
||||||
|
"""准入被允许 → 恒 0.0(现在就能试);此前 redis 侧返回探针 TTL。"""
|
||||||
|
gate = gate_factory(_CFG)
|
||||||
|
await _open_gate(gate)
|
||||||
|
clock.advance(_CFG.cooldown_s + 1)
|
||||||
|
probe = await gate.try_enter("s1", "w1")
|
||||||
|
assert probe.allowed and probe.is_probe
|
||||||
|
assert probe.retry_after_s == 0.0
|
||||||
|
|
||||||
|
async def test_retry_after_zero_while_probe_in_flight(self, gate_factory, clock):
|
||||||
|
"""集合查询同口径: 探针在途的源不贡献等待时间。"""
|
||||||
|
gate = gate_factory(_CFG)
|
||||||
|
await _open_gate(gate)
|
||||||
|
clock.advance(_CFG.cooldown_s + 1)
|
||||||
|
assert (await gate.try_enter("s1", "w1")).is_probe
|
||||||
|
assert await gate.retry_after_s(("s1",)) == 0.0
|
||||||
|
|
||||||
|
async def test_fenced_write_in_half_open_reports_no_certain_wait(self, gate_factory, clock):
|
||||||
|
"""写回被 fencing 拒时的快照同口径;此前 redis 侧返回探针租约剩余。"""
|
||||||
|
gate = gate_factory(_CFG)
|
||||||
|
stale = await gate.try_enter("s1", "slow-worker") # epoch 0 的旧 entry
|
||||||
|
await _open_gate(gate) # 他人开路,epoch 推进
|
||||||
|
clock.advance(_CFG.cooldown_s + 1)
|
||||||
|
assert (await gate.try_enter("s1", "w1")).is_probe # 门此刻 HALF_OPEN
|
||||||
|
update = await gate.record_success(stale)
|
||||||
|
assert not update.applied and update.state is GateState.HALF_OPEN
|
||||||
|
assert update.retry_after_s == 0.0
|
||||||
|
|
||||||
|
|
||||||
class TestConsecutiveSuppression:
|
class TestConsecutiveSuppression:
|
||||||
"""迭代 6: 窗口证据充足且健康时,连败是噪声,不开路(设计 §3.39)。"""
|
"""迭代 6: 窗口证据充足且健康时,连败是噪声,不开路(设计 §3.39)。"""
|
||||||
|
|
||||||
|
|||||||
@@ -327,3 +327,49 @@ async def test_variant_probe_rate_limited_releases_not_hangs(redis_client):
|
|||||||
assert update.applied
|
assert update.applied
|
||||||
nxt = await gate.try_enter("s1", "w2")
|
nxt = await gate.try_enter("s1", "w2")
|
||||||
assert nxt.allowed and nxt.is_probe # 立即可再探,不等 probe_ttl
|
assert nxt.allowed and nxt.is_probe # 立即可再探,不等 probe_ttl
|
||||||
|
|
||||||
|
|
||||||
|
# —— issue #14: retry_after_s = 距离**确定**可再试的时刻,HALF_OPEN 无确定时刻 ——
|
||||||
|
|
||||||
|
|
||||||
|
@pytestmark_slow
|
||||||
|
async def test_variant_half_open_rejection_reports_no_certain_wait(redis_client):
|
||||||
|
gate = _gate(redis_client)
|
||||||
|
await _open_gate(gate)
|
||||||
|
await asyncio.sleep(_CFG.cooldown_s + 1)
|
||||||
|
probe = await gate.try_enter("s1", "w1")
|
||||||
|
assert probe.is_probe
|
||||||
|
blocked = await gate.try_enter("s1", "w2")
|
||||||
|
assert not blocked.allowed and blocked.state is GateState.HALF_OPEN
|
||||||
|
assert blocked.retry_after_s == 0.0
|
||||||
|
|
||||||
|
|
||||||
|
@pytestmark_slow
|
||||||
|
async def test_variant_probe_grant_reports_no_certain_wait(redis_client):
|
||||||
|
gate = _gate(redis_client)
|
||||||
|
await _open_gate(gate)
|
||||||
|
await asyncio.sleep(_CFG.cooldown_s + 1)
|
||||||
|
probe = await gate.try_enter("s1", "w1")
|
||||||
|
assert probe.allowed and probe.is_probe
|
||||||
|
assert probe.retry_after_s == 0.0
|
||||||
|
|
||||||
|
|
||||||
|
@pytestmark_slow
|
||||||
|
async def test_variant_retry_after_zero_while_probe_in_flight(redis_client):
|
||||||
|
gate = _gate(redis_client)
|
||||||
|
await _open_gate(gate)
|
||||||
|
await asyncio.sleep(_CFG.cooldown_s + 1)
|
||||||
|
assert (await gate.try_enter("s1", "w1")).is_probe
|
||||||
|
assert await gate.retry_after_s(("s1",)) == 0.0
|
||||||
|
|
||||||
|
|
||||||
|
@pytestmark_slow
|
||||||
|
async def test_variant_fenced_write_in_half_open_reports_no_certain_wait(redis_client):
|
||||||
|
gate = _gate(redis_client)
|
||||||
|
stale = await gate.try_enter("s1", "slow-worker") # epoch 0 的旧 entry
|
||||||
|
await _open_gate(gate) # 他人开路,epoch 推进
|
||||||
|
await asyncio.sleep(_CFG.cooldown_s + 1)
|
||||||
|
assert (await gate.try_enter("s1", "w1")).is_probe # 门此刻 HALF_OPEN
|
||||||
|
update = await gate.record_success(stale)
|
||||||
|
assert not update.applied and update.state is GateState.HALF_OPEN
|
||||||
|
assert update.retry_after_s == 0.0
|
||||||
|
|||||||
Reference in New Issue
Block a user