fix(snapshot): avoid duplicate scheduled balance retry
This commit is contained in:
@@ -834,6 +834,46 @@ def test_maybe_tick_retries_due_worker_without_daily_refresh():
|
|||||||
assert health["core_holding_reconciliation"]["state"] == "ok"
|
assert health["core_holding_reconciliation"]["state"] == "ok"
|
||||||
|
|
||||||
|
|
||||||
|
def test_maybe_tick_daily_pass_does_not_duplicate_due_worker_retry():
|
||||||
|
from trex_core.application.broker_port import BalanceItem, BrokerBalanceError
|
||||||
|
|
||||||
|
class SequenceBroker(PaperBroker):
|
||||||
|
def __init__(self):
|
||||||
|
super().__init__()
|
||||||
|
self.steps = [BrokerBalanceError("http_5xx"), [BalanceItem("X", 0)]]
|
||||||
|
self.calls = 0
|
||||||
|
|
||||||
|
def get_balance(self):
|
||||||
|
self.calls += 1
|
||||||
|
step = self.steps.pop(0)
|
||||||
|
if isinstance(step, Exception):
|
||||||
|
raise step
|
||||||
|
return step
|
||||||
|
|
||||||
|
now = {"t": datetime(2026, 9, 12, 0, 0, tzinfo=UTC)}
|
||||||
|
broker = SequenceBroker()
|
||||||
|
sched = SnapshotScheduler(
|
||||||
|
read_port=_make_stub(),
|
||||||
|
store=SnapshotStore(),
|
||||||
|
broker=broker,
|
||||||
|
schedule_kst=datetime(2026, 9, 12, 0, 0).time(),
|
||||||
|
clock=lambda: now["t"],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert sched.run_once() is True
|
||||||
|
assert broker.calls == 1
|
||||||
|
|
||||||
|
# On the next day's scheduled tick, both the daily pass and the failed
|
||||||
|
# worker retry are due. The daily pass must satisfy both with one read.
|
||||||
|
now["t"] += timedelta(days=1)
|
||||||
|
sched._maybe_tick()
|
||||||
|
|
||||||
|
assert broker.calls == 2
|
||||||
|
assert broker.steps == []
|
||||||
|
state = sched.balance_worker_health(stale_after_sec=86_400)
|
||||||
|
assert state["position_reconciliation"]["state"] == "ok"
|
||||||
|
|
||||||
|
|
||||||
def test_run_forever_retries_due_worker_on_production_loop_path():
|
def test_run_forever_retries_due_worker_on_production_loop_path():
|
||||||
from trex_core.application.broker_port import BalanceItem, BrokerBalanceError
|
from trex_core.application.broker_port import BalanceItem, BrokerBalanceError
|
||||||
|
|
||||||
|
|||||||
@@ -117,15 +117,16 @@ class SnapshotScheduler:
|
|||||||
_log.exception("snapshot scheduler tick failed")
|
_log.exception("snapshot scheduler tick failed")
|
||||||
|
|
||||||
def _maybe_tick(self) -> None:
|
def _maybe_tick(self) -> None:
|
||||||
self._maybe_retry_balance_workers()
|
|
||||||
now_kst = self._clock().astimezone(_KST)
|
now_kst = self._clock().astimezone(_KST)
|
||||||
target = datetime.combine(now_kst.date(), self._schedule_kst, tzinfo=_KST)
|
target = datetime.combine(now_kst.date(), self._schedule_kst, tzinfo=_KST)
|
||||||
if now_kst < target:
|
|
||||||
return
|
|
||||||
last = self._last_fired_date
|
last = self._last_fired_date
|
||||||
if last is not None and last.date() == now_kst.date():
|
daily_due = now_kst >= target and (last is None or last.date() != now_kst.date())
|
||||||
return # already fired today
|
if daily_due:
|
||||||
|
# The daily pass already invokes every eligible balance worker.
|
||||||
|
# Do not run a due retry first and duplicate the same broker read.
|
||||||
self.run_once()
|
self.run_once()
|
||||||
|
return
|
||||||
|
self._maybe_retry_balance_workers()
|
||||||
|
|
||||||
# ---- entry-allowed gate (REQ-114) ------------------------------------
|
# ---- entry-allowed gate (REQ-114) ------------------------------------
|
||||||
def get_entry_allowed(self) -> bool:
|
def get_entry_allowed(self) -> bool:
|
||||||
|
|||||||
Reference in New Issue
Block a user