feat(cluster): worker registry persistence, split-brain protection, circuit breaker - #1928
feat(cluster): worker registry persistence, split-brain protection, circuit breaker#1928hognek wants to merge 4 commits into
Conversation
Qodo reviews are paused for this user.Troubleshooting steps vary by plan Learn more → On a Teams plan? Using GitHub Enterprise Server, GitLab Self-Managed, or Bitbucket Data Center? |
|
Warning Review limit reached
Next review available in: 31 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughWalkthroughChangesCluster reliability
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant ClusterRoutes
participant ClusterManager
participant WorkerRegistryStore
participant TaskRouter
participant FailureTracker
participant Worker
Client->>ClusterRoutes: register or heartbeat with generation
ClusterRoutes->>ClusterManager: forward worker lifecycle request
ClusterManager->>WorkerRegistryStore: persist worker state
TaskRouter->>FailureTracker: check worker circuit
TaskRouter->>Worker: route request
Worker-->>TaskRouter: response or error
TaskRouter->>FailureTracker: record success or failure
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
|
![kilo-code-bot[bot]](https://avatars.githubusercontent.com/in/2193792?s=60&v=4){kind=small}
| # taOS #640: split-brain protection — reject heartbeat from a worker | ||
| # that is echoing a different generation (another active controller). | ||
| # Legacy workers that don't send generation get a pass (None). | ||
| if generation is not None and generation != self._generation: |
There was a problem hiding this comment.
WARNING: Split-brain protection is bypassed for any worker that omits generation (legacy / not-yet-upgraded agents).
if generation is not None and generation != self._generation: means a worker that simply does not send generation (or sends None) is always accepted, regardless of which controller it last talked to. In a true split-brain scenario where a stale controller is still alive and its workers never adopt the new generation, those workers would be happily accepted here — defeating the protection for the entire legacy fleet. Consider rejecting workers that should be on a generation once the fleet is expected to send one, or at least logging/metrics on how many None-generation heartbeats are accepted so the protection isn't silently inert.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| logger.info(f"Worker registered: {info.name} ({info.platform}, {len(info.capabilities)} capabilities)") | ||
|
|
||
| # taOS #640: persist worker to SQLite so the registry survives restarts. | ||
| if self._registry_store is not None: |
There was a problem hiding this comment.
WARNING: register_worker performs no generation check, so split-brain protection only covers heartbeat, not the re-enroll path.
The new generation guard lives only in heartbeat() (line 248). A worker that re-registers via register_worker (the normal re-enroll flow on worker boot) is accepted unconditionally and persisted, even if it was last talking to a now-stale controller. When a controller restarts, its generation increments, but a worker that re-registers (instead of heartbeating) is admitted regardless — the exact split-brain case this PR targets. Either pass/verify generation through register_worker too, or document why re-registration is exempt and why that can't race with a live stale controller.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| # taOS #640: persist updated worker state to SQLite. | ||
| if self._registry_store is not None: | ||
| try: | ||
| asyncio.get_running_loop().create_task( |
There was a problem hiding this comment.
WARNING: Fire-and-forget create_task(self._persist_worker(worker)) has no error handling and is never awaited.
If _persist_worker raises inside the spawned task (e.g. DB locked, disk full, serialization error), the exception becomes an unretrieved task exception logged as a generic asyncio warning, and the worker's persisted state silently goes stale with no logger.exception like the other call sites have. It is also not awaited before stop(), so a controller shutdown can drop the final heartbeat's persistence. Wrap the task body in try/except (mirroring the register_worker/unregister_worker handling) and/or track it in self._background_tasks so it can be awaited/cleaned up on stop.
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| first_failure_at: float | ||
|
|
||
|
|
||
| # Sensible defaults: 5 failures in 60 seconds trips the breaker. |
There was a problem hiding this comment.
SUGGESTION: Comment says "5 failures in 60 seconds" but DEFAULT_FAILURE_THRESHOLD = 3.
This is misleading to readers/maintainers; the actual default trips after 3 failures. Either update the comment to match the constant or change the constant if 5 was intended.
| # Sensible defaults: 5 failures in 60 seconds trips the breaker. | |
| # Sensible defaults: 3 failures in 60 seconds trips the breaker. |
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
| if self._db is None: | ||
| raise RuntimeError("WorkerRegistryStore not initialised") | ||
| await self._db.execute( | ||
| "UPDATE cluster_generation SET value = value + 1 WHERE id = 1" |
There was a problem hiding this comment.
SUGGESTION: increment_generation uses a separate UPDATE then SELECT, which is two round-trips and a non-atomic read of the value.
Use SQLite's RETURNING to get the new value in a single atomic statement, e.g. UPDATE cluster_generation SET value = value + 1 WHERE id = 1 RETURNING value. This removes the second query, avoids any window where another writer could shift the value between the UPDATE and the SELECT, and is supported by aiosqlite/SQLite 3.35+ (the same baseline the rest of the stores already rely on).
Reply with @kilocode-bot fix it to have Kilo Code address this issue.
Code Review SummaryStatus: 5 Issues Found | Recommendation: Address before merge Overview
Issue Details (click to expand)WARNING
SUGGESTION
Files Reviewed (6 files)
Fix these issues in Kilo Cloud Previous Review Summary (commit 9f1034e)Current summary above is authoritative. Previous snapshots are kept for context only. Previous review (commit 9f1034e)Status: 5 Issues Found | Recommendation: Address before merge Overview
Issue Details (click to expand)WARNING
SUGGESTION
Files Reviewed (6 files)
Reviewed by hy3:free · Input: 41.8K · Output: 3.4K · Cached: 255.2K |
9f1034e to
46aa5a6
Compare
…ircuit breaker Fixes jaylfc#640. Four interconnected changes: 1. **Worker registry persistence** — New WorkerRegistryStore (BaseStore + SQLite) persists worker registrations so the in-memory dict survives controller restarts. On startup, loaded workers are marked 'stale' until they re-heartbeat; local worker is skipped during load. 2. **Split-brain protection via generation counter** — A generation counter (stored in the same SQLite table) is incremented on each controller start. Workers echo it in heartbeats; stale generations are rejected. Legacy workers that don't send generation get a pass (None). 3. **Circuit breaker for failing workers** — New FailureTracker with sliding window (default: 3 failures in 60s). TaskRouter skips tripped workers instead of serially blocking 60s * N_workers. Success clears the circuit. 4. **In-flight routing recovery** — Addressed by Fix 1: persisted workers survive controller restart; stale-until-heartbeat prevents routing to workers that haven't reconnected. Files: - NEW: tinyagentos/cluster/worker_registry_store.py - NEW: tinyagentos/cluster/failure_tracker.py - MODIFIED: tinyagentos/cluster/manager.py (init, start, heartbeat, register_worker, unregister_worker, _persist_worker, _load_persisted_workers) - MODIFIED: tinyagentos/cluster/router.py (circuit breaker skip/record) - MODIFIED: tinyagentos/app.py (wire new stores) - NEW: tests/test_cluster_persistence.py (24 tests: store, tracker, manager persistence, router circuit breaker) Tests: 79/79 pass (55 existing + 24 new cluster tests)
- failure_tracker: fix comment (3 failures, not 5) - manager: add generation check to register_worker (mirrors heartbeat guard) - manager: add debug log for None-generation heartbeats from legacy workers - manager: wrap heartbeat-persist fire-and-forget in try/except with background task tracking - worker_registry_store: use RETURNING for atomic increment_generation - routes: wire generation through WorkerRegister and HeartbeatBody models Rebased onto origin/dev (8b97eb5a). All 96 cluster/persistence/worker-protocol tests pass.
46aa5a6 to
64e753e
Compare
|
All 5 Kilo findings addressed + rebased onto origin/dev (8b97eb5a): WARNING (3 fixed):
SUGGESTION (2 fixed): Routes wired: 96/96 cluster/persistence/worker-protocol tests pass. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
tinyagentos/routes/cluster.py (1)
381-385: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winReturn an error when generation validation rejects registration.
ClusterManager.register_worker()silently returns on mismatch, but this handler still records capabilities, surfaces backups, and responds"registered". Return a boolean or raise a typed error, and only run these side effects after acceptance.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tinyagentos/routes/cluster.py` around lines 381 - 385, Update the registration flow around ClusterManager.register_worker so generation mismatches are observable through a boolean result or typed exception. In the handler, only call _record_worker_capability, _surface_storage_backup, and return the "registered" response after registration is accepted; otherwise return an appropriate error response or propagate the typed error.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@tests/test_cluster_persistence.py`:
- Around line 579-591: Update the failure-tracking tests around
test_records_failure_on_error and the reset test so their assertions observe
mutations: use failure_threshold=1 when verifying a recorded failure, and in the
reset scenario use threshold=2, record an initial failure, verify the tracker is
tripped, perform a successful request, record another failure, then assert the
worker is untripped.
In `@tinyagentos/cluster/failure_tracker.py`:
- Around line 32-89: Replace FailureTracker’s first-failure counter model with
per-worker failure timestamp storage, using time.monotonic() rather than
time.time(). In record_failure, append the current timestamp and prune
timestamps older than window_seconds; in is_tripped, prune the worker’s
timestamps before comparing their count with failure_threshold, removing empty
records. Preserve record_success by clearing the worker’s stored timestamps.
In `@tinyagentos/cluster/manager.py`:
- Around line 408-421: Update the background persistence flow around
_safe_persist and its corresponding persistence path near the alternate location
to verify that the exact WorkerInfo instance is still registered immediately
before calling _persist_worker. Skip persistence when the worker has been
removed or replaced, while preserving existing exception logging and task
scheduling behavior.
- Around line 71-80: Update the controller lifecycle around start and its
request/heartbeat handling to periodically or before accepting work compare the
controller’s stored generation with registry_store.current_generation(); when
the durable generation is higher, fence and disable the older controller so it
no longer accepts or routes workers. Preserve normal operation for the current
generation, and add a regression test covering controller A at generation 2
being disabled after controller B advances the store to generation 3.
In `@tinyagentos/cluster/router.py`:
- Around line 45-54: Update the exception handling around raise_for_status in
the worker request flow to avoid calling ft.record_failure for request-specific
4xx client responses. Handle client errors separately, while continuing to
record transport/timeouts and retryable 5xx failures so healthy workers are not
tripped by malformed or unauthorized requests.
---
Outside diff comments:
In `@tinyagentos/routes/cluster.py`:
- Around line 381-385: Update the registration flow around
ClusterManager.register_worker so generation mismatches are observable through a
boolean result or typed exception. In the handler, only call
_record_worker_capability, _surface_storage_backup, and return the "registered"
response after registration is accepted; otherwise return an appropriate error
response or propagate the typed error.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: bd5f4c1e-e546-4ae2-99b5-5fd69cdd18e0
📒 Files selected for processing (7)
tests/test_cluster_persistence.pytinyagentos/app.pytinyagentos/cluster/failure_tracker.pytinyagentos/cluster/manager.pytinyagentos/cluster/router.pytinyagentos/cluster/worker_registry_store.pytinyagentos/routes/cluster.py
| async def test_records_failure_on_error(self): | ||
| ft = FailureTracker(failure_threshold=5) | ||
| mgr = ClusterManager(failure_tracker=ft) | ||
| await mgr.register_worker(_make_worker("w1", capabilities=["chat"], load=0.1, url="http://w1:8000")) | ||
|
|
||
| mock_client = AsyncMock(spec=httpx.AsyncClient) | ||
| mock_client.post.side_effect = Exception("connection refused") | ||
|
|
||
| router = TaskRouter(mgr, mock_client) | ||
| data, name = await router.route_request("chat", "POST", "/v1/chat/completions", {}) | ||
|
|
||
| assert data is None | ||
| assert ft.is_tripped("w1") is False # 1 failure, threshold=5 → not tripped |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Make these tests observe the tracker mutation.
The first test passes even if no failure is recorded because one failure is below five. The second remains untripped before and after success, so it does not prove reset behavior. Use threshold 1 for failure recording; for reset, record one failure at threshold 2, succeed, record another, then assert untripped.
Also applies to: 593-612
🧰 Tools
🪛 ast-grep (0.45.0)
[warning] 581-581: Do not make http calls without encryption
Context: "http://w1:8000"
Note: [CWE-319] Cleartext Transmission of Sensitive Information.
(requests-http)
🪛 Ruff (0.16.0)
[warning] 588-588: Unpacked variable name is never used
Prefix it with an underscore or any other dummy variable pattern
(RUF059)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tests/test_cluster_persistence.py` around lines 579 - 591, Update the
failure-tracking tests around test_records_failure_on_error and the reset test
so their assertions observe mutations: use failure_threshold=1 when verifying a
recorded failure, and in the reset scenario use threshold=2, record an initial
failure, verify the tracker is tripped, perform a successful request, record
another failure, then assert the worker is untripped.
| class FailureTracker: | ||
| """Per-worker failure counter with a sliding time window. | ||
|
|
||
| After ``failure_threshold`` failures within ``window_seconds``, | ||
| ``is_tripped(worker_name)`` returns ``True`` and the router skips | ||
| that worker. The window resets once enough time has passed since | ||
| the first failure in the current window. | ||
| """ | ||
|
|
||
| def __init__( | ||
| self, | ||
| failure_threshold: int = DEFAULT_FAILURE_THRESHOLD, | ||
| window_seconds: float = DEFAULT_WINDOW_SECONDS, | ||
| ): | ||
| self._failure_threshold = failure_threshold | ||
| self._window_seconds = window_seconds | ||
| self._records: dict[str, _FailureRecord] = defaultdict( | ||
| lambda: _FailureRecord(0, 0.0) | ||
| ) | ||
|
|
||
| def record_failure(self, worker_name: str) -> None: | ||
| """Register a failure for the given worker. | ||
|
|
||
| After the window has elapsed since the first failure, the counter | ||
| resets automatically on the next failure. | ||
| """ | ||
| now = time.time() | ||
| rec = self._records[worker_name] | ||
| if rec.first_failure_at > 0 and (now - rec.first_failure_at) > self._window_seconds: | ||
| # Window expired — reset | ||
| self._records[worker_name] = _FailureRecord(1, now) | ||
| elif rec.first_failure_at == 0: | ||
| self._records[worker_name] = _FailureRecord(1, now) | ||
| else: | ||
| self._records[worker_name] = _FailureRecord( | ||
| rec.failures + 1, rec.first_failure_at | ||
| ) | ||
|
|
||
| def record_success(self, worker_name: str) -> None: | ||
| """Clear the failure record for a worker after a successful request.""" | ||
| self._records.pop(worker_name, None) | ||
|
|
||
| def is_tripped(self, worker_name: str) -> bool: | ||
| """Return True if the worker's circuit breaker is currently open. | ||
|
|
||
| The breaker opens when ``failure_threshold`` failures have occurred | ||
| within ``window_seconds`` and not enough time has elapsed since the | ||
| first failure to reset the window. | ||
| """ | ||
| rec = self._records.get(worker_name) | ||
| if rec is None: | ||
| return False | ||
| now = time.time() | ||
| if rec.first_failure_at > 0 and (now - rec.first_failure_at) > self._window_seconds: | ||
| # Window expired — auto-reset | ||
| self._records.pop(worker_name, None) | ||
| return False | ||
| return rec.failures >= self._failure_threshold |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
Implement an actual sliding failure window.
Failures at t=0, t=59, and t=61 cause the current code to discard the still-recent t=59 failure. A breaker tripped at t=59 also remains open for only one second. Store timestamps per worker, prune entries older than the window, and use a monotonic clock.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/cluster/failure_tracker.py` around lines 32 - 89, Replace
FailureTracker’s first-failure counter model with per-worker failure timestamp
storage, using time.monotonic() rather than time.time(). In record_failure,
append the current timestamp and prune timestamps older than window_seconds; in
is_tripped, prune the worker’s timestamps before comparing their count with
failure_threshold, removing empty records. Preserve record_success by clearing
the worker’s stored timestamps.
| async def start(self): | ||
| # taOS #640: increment generation on each controller start (split-brain | ||
| # protection). Workers must echo the current generation in heartbeats; | ||
| # a stale generation means another controller instance is (or was) active. | ||
| if self._registry_store is not None: | ||
| self._generation = await self._registry_store.increment_generation() | ||
| logger.info("Cluster generation: %d", self._generation) | ||
| # Load persisted workers and mark them stale — they must | ||
| # re-register or heartbeat before being used for routing. | ||
| await self._load_persisted_workers() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Fence the superseded controller, not only mismatched workers.
If controller A starts at generation 2 and controller B advances the store to 3, A still accepts generation-2 workers indefinitely because it never rechecks current_generation(). Both controllers can remain active. Detect durable generation advancement and self-disable the older controller, with a concurrent-controller regression test.
Also applies to: 273-290
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/cluster/manager.py` around lines 71 - 80, Update the controller
lifecycle around start and its request/heartbeat handling to periodically or
before accepting work compare the controller’s stored generation with
registry_store.current_generation(); when the durable generation is higher,
fence and disable the older controller so it no longer accepts or routes
workers. Preserve normal operation for the current generation, and add a
regression test covering controller A at generation 2 being disabled after
controller B advances the store to generation 3.
| # taOS #640: persist updated worker state to SQLite. | ||
| if self._registry_store is not None: | ||
| async def _safe_persist() -> None: | ||
| try: | ||
| await self._persist_worker(worker) | ||
| except Exception: | ||
| logger.exception("Failed to persist worker '%s'", worker.name) | ||
|
|
||
| try: | ||
| task = asyncio.get_running_loop().create_task(_safe_persist()) | ||
| self._background_tasks.add(task) | ||
| task.add_done_callback(self._background_tasks.discard) | ||
| except RuntimeError: | ||
| pass # no running loop (e.g. sync tests) — skip gracefully |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Prevent queued heartbeats from resurrecting unregistered workers.
A pending heartbeat persistence task can run after remove_worker() and upsert the deleted row, causing a ghost worker after restart. Before persisting, verify this exact WorkerInfo is still registered.
Proposed guard
async def _safe_persist() -> None:
try:
+ if self._workers.get(worker.name) is not worker:
+ return
await self._persist_worker(worker)
except Exception:
logger.exception("Failed to persist worker '%s'", worker.name)Also applies to: 470-475
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/cluster/manager.py` around lines 408 - 421, Update the background
persistence flow around _safe_persist and its corresponding persistence path
near the alternate location to verify that the exact WorkerInfo instance is
still registered immediately before calling _persist_worker. Skip persistence
when the worker has been removed or replaced, while preserving existing
exception logging and task scheduling behavior.
| resp.raise_for_status() | ||
| # Success — clear any failure record for this worker. | ||
| if ft is not None: | ||
| ft.record_success(worker.name) | ||
| return resp.json(), worker.name | ||
| except Exception as e: | ||
| logger.warning(f"Worker '{worker.name}' failed for {capability}: {e}") | ||
| # taOS #640 Fix 3: record the failure for circuit breaker. | ||
| if ft is not None: | ||
| ft.record_failure(worker.name) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'Files matching router.py:\n'
fd -a 'router\.py$' . | sed 's#^\./##'
printf '\nRelevant file outline:\n'
ast-grep outline tinyagentos/cluster/router.py 2>/dev/null || true
printf '\nRelevant router.py lines 1-140:\n'
sed -n '1,140p' tinyagentos/cluster/router.py | nl -ba
printf '\nSearch for FailureTracker / record_failure / record_success:\n'
rg -n "class FailureTracker|record_failure|record_success|circuit|capacity|circuitbreaker|500|502|503|raise_for_status|requests\." tinyagentos tests -SRepository: jaylfc/taOS
Length of output: 638
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Verify behavioral semantics of requests.raise_for_status for client/server responses
# using installed requests library if available, and inspect any local FailureTracker implementation.
python3 - <<'PY'
import sys
try:
import requests
print("requests", getattr(requests, "__version__", "unknown"))
except Exception as e:
print("requests unavailable:", type(e).__name__, e)
try:
import tinyagentos.cluster.router as router
import inspect
print("router.file", inspect.getsourcefile(router))
src = inspect.getsource(routers.get(router, router))
print("has FailureTracker subclass:", "class Failure" in inspect.getsource(router) or False)
except Exception as e:
print("inspect tinyagentos.cluster.router failed:", type(e).__name__, e)
PY
printf '\nFailureTracker implementation candidates:\n'
rg -n "class FailureTracker|class .*Tracker|record_failure|record_success" tinyagentos/tests -S || trueRepository: jaylfc/taOS
Length of output: 455
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'tinyagentos/cluster/router.py lines 1-180:\n'
awk '{printf "%5d\t%s\n", NR, $0}' tinyagentos/cluster/router.py | sed -n '1,180p'
printf '\nSearch for FailureTracker / record_failure / record_success / circuit:\n'
rg -n "class FailureTracker|record_failure|record_success|circuit|circuitbreak|circuit breaker|500|502|503|raise_for_status|record_(success|failure)" tinyagentos tests tinyagentos/tests -S 2>/dev/null || true
printf '\nDependency files:\n'
git ls-files | rg '(^|/)(pyproject\.toml|requirements.*\.txt|setup\.py|setup\.cfg|poetry\.lock|uv\.lock)' | while read -r f; do
echo "--- $f"
rg -n "requests|httpx|uvicorn|fastapi" "$f" || true
doneRepository: jaylfc/taOS
Length of output: 50368
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'tinyagentos/cluster/failure_tracker.py:\n'
awk '{printf "%5d\t%s\n", NR, $0}' tinyagentos/cluster/failure_tracker.py | sed -n '1,130p'
printf '\nHTTP 4xx docs via pip-inspection/source availability:\n'
python3 - <<'PY'
try:
import httpx
import inspect
print("httpx", getattr(httpx, "__version__", "unknown"))
source = inspect.getsource(httpx.Response.raise_for_status)
print(source)
except Exception as e:
print("httpx unavailable or inspect failed:", type(e).__name__, e)
PYRepository: jaylfc/taOS
Length of output: 4443
🌐 Web query:
httpx Response.raise_for_status source code raises status_code 400 exception documentation
💡 Result:
In the httpx library, the response.raise_for_status method is used to check the HTTP status code of a response [1][2]. If the status code indicates an error—specifically 4xx (client error) or 5xx (server error)—the method raises an httpx.HTTPStatusError exception [2][3]. The exception httpx.HTTPStatusError is a subclass of httpx.HTTPError [2][3]. It contains information about the request that triggered the error and the corresponding response object [3]. When status codes are in the 400–599 range, raise_for_status triggers this exception [2][4]. If the response status code is successful (typically 2xx), the method returns the response object itself [4]. It is important to note that raise_for_status requires the response object to have a reference to the request that initiated it [3][5]. If a response is created manually without associating it with a request, calling raise_for_status will raise a RuntimeError [5].
Citations:
- 1: https://www.python-httpx.org/api/
- 2: https://www.python-httpx.org/exceptions/
- 3: https://github.com/encode/httpx/blob/def4778d/httpx/_exceptions.py
- 4: https://github.com/encode/httpx/blob/b5addb64/tests/client/test_client.py
- 5: https://github.com/encode/httpx/blob/def4778d/tests/models/test_responses.py
Do not trip healthy workers for request-specific 4xx responses.
raise_for_status() also raises for 4xx client errors, and this handler records every exception in FailureTracker. Repeated malformed or unauthorized requests can therefore trip every worker circuit; record transport/timeouts and retryable server 5xx failures, while treating client errors separately.
🧰 Tools
🪛 Ruff (0.16.0)
[warning] 50-50: Do not catch blind exception: Exception
(BLE001)
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@tinyagentos/cluster/router.py` around lines 45 - 54, Update the exception
handling around raise_for_status in the worker request flow to avoid calling
ft.record_failure for request-specific 4xx client responses. Handle client
errors separately, while continuing to record transport/timeouts and retryable
5xx failures so healthy workers are not tripped by malformed or unauthorized
requests.
…cking Replace the fixed-bucket anchor approach (tracking only first_failure_at + count) with a deque of individual failure timestamps. The old approach discarded recent failures when the window anchor expired — e.g., failures at t=0, t=59, t=61 would reset the count at t=61, discarding the still-recent t=59 failure. The new implementation prunes only timestamps older than now - window_seconds, correctly keeping all recent failures regardless of when the first failure occurred. - Replace _FailureRecord named tuple with deque[float] per worker - Add _prune() helper that removes stale timestamps - Add _time_func attribute for testability (controllable clock) - Add tests for sliding window edge cases (t=0,59,61 and prune behaviour) CodeRabbit: fixes MAJOR finding on PR jaylfc#1928
1 participant
Closes #640.
Four interconnected changes for multi-node reliability:
1. Worker registry persistence — New
WorkerRegistryStore(SQLite-backed viaBaseStore) persists worker registrations so the in-memoryClusterManager._workersdict survives controller restarts. On startup, loaded workers are marked"stale"until they re-heartbeat. The"local"worker is skipped during load (it re-registers on every boot).2. Split-brain protection via generation counter — A generation counter (stored in the same SQLite table) is incremented on each controller
start(). Workers echo it in heartbeats; stale generations are rejected. Legacy workers that don't send generation get a pass (None).3. Circuit breaker for failing workers — New
FailureTrackerwith sliding window (default: 3 failures in 60s).TaskRouterskips tripped workers instead of serially blocking60s * N_workers. Success clears the circuit; failures recorded per-worker.4. In-flight routing recovery — Addressed by Fix 1: persisted workers survive controller restart; stale-until-heartbeat prevents routing to workers that haven't reconnected.
Files changed
tinyagentos/cluster/worker_registry_store.py— SQLite store for worker records + generation countertinyagentos/cluster/failure_tracker.py— Sliding-window circuit breakertinyagentos/cluster/manager.py— init/start/heartbeat/register/unregister with persistence, generation counter, failure_tracker wiringtinyagentos/cluster/router.py— Circuit breaker skip + record on failure/successtinyagentos/app.py— Wire new stores into ClusterManager creationtests/test_cluster_persistence.py— 24 tests covering store CRUD, generation counter, stale-on-load, heartbeat revival, split-brain rejection, circuit breaker tripping/skipping/clearingTesting
Backward compatibility
ClusterManager.__init__accepts new optionalworker_registry_storeandfailure_trackerparams — defaults to Noneheartbeat()accepts new optionalgenerationparam — None = legacy worker, acceptedstore.init()failure_tracker is None(no circuit breaker)Summary by CodeRabbit
New Features
Tests