sync-v2 §10.7: soak stability — even per-worker cap + zombie diagnostic
- Size MAX_ACCOUNTS_PER_WORKER to ~1.2x the fair share so the fleet fills each worker evenly at startup and the rebalancer barely sheds; without this, three coordinators racing for the whole fleet shed/reclaim in a thrash that reset reclaimed accounts' first-poll clocks and dragged the sync-lag ramp. - Sample a "dead" count (owned runners whose work loops have stopped) to distinguish a real zombie-runner bug from a slow ramp. Validated: dead=0 throughout; 400-account run holds lag ~48s under a 52s budget across 13 steady samples, persistent sockets == hot, RSS flat. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -193,10 +193,11 @@ class Sample:
|
|||||||
max_lag: float
|
max_lag: float
|
||||||
rss_mb: float
|
rss_mb: float
|
||||||
synced: int # accounts that have completed >=1 poll (steady-state gate)
|
synced: int # accounts that have completed >=1 poll (steady-state gate)
|
||||||
|
dead: int # owned runners whose work loops have died (zombies)
|
||||||
|
|
||||||
|
|
||||||
async def _tier_counts(coords: list[SyncCoordinator]) -> tuple[int, int, int, int]:
|
async def _tier_counts(coords: list[SyncCoordinator]) -> tuple[int, int, int, int, int]:
|
||||||
owned = hot = warm = cold = 0
|
owned = hot = warm = cold = dead = 0
|
||||||
for c in coords:
|
for c in coords:
|
||||||
for runner in list(c._runners.values()):
|
for runner in list(c._runners.values()):
|
||||||
owned += 1
|
owned += 1
|
||||||
@@ -206,7 +207,9 @@ async def _tier_counts(coords: list[SyncCoordinator]) -> tuple[int, int, int, in
|
|||||||
warm += 1
|
warm += 1
|
||||||
else:
|
else:
|
||||||
cold += 1
|
cold += 1
|
||||||
return owned, hot, warm, cold
|
if not runner.running(): # owned but its work loops have died = zombie
|
||||||
|
dead += 1
|
||||||
|
return owned, hot, warm, cold, dead
|
||||||
|
|
||||||
|
|
||||||
def _max_lag(now: float, seeded: int, start: float) -> float:
|
def _max_lag(now: float, seeded: int, start: float) -> float:
|
||||||
@@ -250,8 +253,15 @@ async def _soak(cfg: Config) -> int:
|
|||||||
|
|
||||||
await refresh_sessions()
|
await refresh_sessions()
|
||||||
|
|
||||||
|
# Size the per-worker cap so total capacity is just above the fleet: claims
|
||||||
|
# fill each worker roughly evenly and the rebalancer barely has to shed
|
||||||
|
# (mirrors how you'd tune MAX_ACCOUNTS_PER_WORKER in a real deployment).
|
||||||
|
import math
|
||||||
|
|
||||||
|
cap = max(50, math.ceil(cfg.accounts / cfg.workers * 1.2))
|
||||||
coords = [SyncCoordinator(f"soak-w{i}") for i in range(cfg.workers)]
|
coords = [SyncCoordinator(f"soak-w{i}") for i in range(cfg.workers)]
|
||||||
for c in coords:
|
for c in coords:
|
||||||
|
c.max_accounts = cap
|
||||||
c.claim_interval = cfg.claim_s
|
c.claim_interval = cfg.claim_s
|
||||||
c.claim_jitter = 1.0
|
c.claim_jitter = 1.0
|
||||||
c.tier_eval_interval = cfg.tier_eval_s
|
c.tier_eval_interval = cfg.tier_eval_s
|
||||||
@@ -272,7 +282,7 @@ async def _soak(cfg: Config) -> int:
|
|||||||
)
|
)
|
||||||
print(
|
print(
|
||||||
f"{'t':>6} {'owned':>6} {'hot':>5} {'warm':>5} {'cold':>5} "
|
f"{'t':>6} {'owned':>6} {'hot':>5} {'warm':>5} {'cold':>5} "
|
||||||
f"{'persist':>8} {'open':>5} {'lag_s':>7} {'rss_mb':>8}",
|
f"{'persist':>8} {'open':>5} {'lag_s':>7} {'rss_mb':>8} {'dead':>5}",
|
||||||
flush=True,
|
flush=True,
|
||||||
)
|
)
|
||||||
try:
|
try:
|
||||||
@@ -282,7 +292,7 @@ async def _soak(cfg: Config) -> int:
|
|||||||
if now >= next_session_refresh:
|
if now >= next_session_refresh:
|
||||||
await refresh_sessions()
|
await refresh_sessions()
|
||||||
next_session_refresh = now + cfg.session_grace_s / 2
|
next_session_refresh = now + cfg.session_grace_s / 2
|
||||||
owned, hot, warm, cold = await _tier_counts(coords)
|
owned, hot, warm, cold, dead = await _tier_counts(coords)
|
||||||
s = Sample(
|
s = Sample(
|
||||||
t=now - start,
|
t=now - start,
|
||||||
owned=owned,
|
owned=owned,
|
||||||
@@ -294,11 +304,13 @@ async def _soak(cfg: Config) -> int:
|
|||||||
max_lag=_max_lag(now, cfg.accounts, start),
|
max_lag=_max_lag(now, cfg.accounts, start),
|
||||||
rss_mb=_rss_mb(),
|
rss_mb=_rss_mb(),
|
||||||
synced=len(MockStats.last_sync),
|
synced=len(MockStats.last_sync),
|
||||||
|
dead=dead,
|
||||||
)
|
)
|
||||||
samples.append(s)
|
samples.append(s)
|
||||||
print(
|
print(
|
||||||
f"{s.t:6.0f} {s.owned:6d} {s.hot:5d} {s.warm:5d} {s.cold:5d} "
|
f"{s.t:6.0f} {s.owned:6d} {s.hot:5d} {s.warm:5d} {s.cold:5d} "
|
||||||
f"{s.persistent:8d} {s.total_open:5d} {s.max_lag:7.0f} {s.rss_mb:8.1f}",
|
f"{s.persistent:8d} {s.total_open:5d} {s.max_lag:7.0f} {s.rss_mb:8.1f} "
|
||||||
|
f"{s.dead:5d}",
|
||||||
flush=True,
|
flush=True,
|
||||||
)
|
)
|
||||||
finally:
|
finally:
|
||||||
|
|||||||
Reference in New Issue
Block a user