Conversation
…istributed cache & notification queue + separate scheduler process
format files fix fix fix fix fix fix fix tests fix fix fix fix fix fix fix fix fix try to fix tests fix test remove: redis files fix fix fix fix fix fix: path fix: database waits in tests
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the
WalkthroughAdds multi-worker support with NATS/JetStream KV, a role-aware FastAPI app factory (panel/node-worker/scheduler), NATS-based RPC/router/client, node worker service, pluggable notification queues, scheduler/jobstore support, lifecycle registries, many job gating changes, deployment scripts, and CI adjustments. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant API as Panel API
participant Router as NATS Router
participant KV as JetStream KV
participant Scheduler as Scheduler
participant NodeWorker as NodeWorker
API->>Router: publish Message(topic=CORE/SETTING, data)
Router->>NodeWorker: deliver message to handler
Router->>Scheduler: deliver message to handler
Scheduler->>KV: persist snapshot
NodeWorker->>KV: persist or request state (RPC)
NodeWorker->>NodeWorker: handle command / RPC, call local handlers
sequenceDiagram
autonumber
participant Caller as Caller
participant Op as NodeOperation
participant NatsClient as node_nats_client
participant Remote as Remote NodeWorker
participant Local as LocalImpl
Caller->>Op: update_node(node_id)
Op->>Op: check IS_NODE_WORKER
alt Running as node-worker
Op->>Local: _update_node_local(node_id)
Local-->>Op: result
else
Op->>NatsClient: request("update_node", payload, timeout)
NatsClient->>Remote: RPC via NATS
Remote-->>NatsClient: response
NatsClient-->>Op: response
end
Op-->>Caller: return result
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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 |
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
Note
Due to the large number of review comments, Critical, Major severity comments were prioritized as inline comments.
🤖 Fix all issues with AI agents
In `@app/core/hosts.py`:
- Around line 278-291: Replace insecure pickle serialization in
_persist_state/_snapshot_state with JSON-based serialization using the
SubscriptionInboundData Pydantic methods: when creating state bytes for
STATE_CACHE_KEY, convert each SubscriptionInboundData instance to JSON via
model_dump_json() (or use model_dump then json.dumps for nested dicts), store
the resulting UTF-8 bytes, and update loading logic elsewhere to reconstruct
objects with SubscriptionInboundData.model_validate_json() (or model_validate
after json.loads). Ensure all places that read from the KV use the new JSON
deserialization path and handle older pickle data gracefully (e.g., detect
pickle vs JSON and reject/ignore pickle) and keep references to _persist_state,
_snapshot_state, STATE_CACHE_KEY, and SubscriptionInboundData when making
changes.
In `@app/core/manager.py`:
- Line 1: The code currently imports and uses pickle for (de)serializing core
state (e.g., calls to pickle.loads/pickle.dumps when reading/writing the NATS KV
bucket) which is unsafe; replace pickle serialization with a safe format (JSON
or similar) by removing pickle usage and implementing to_json/from_json (or
serialize/deserialize) methods on the AbstractCore class hierarchy so each core
can produce a JSON-serializable dict and be reconstructed from it, then update
the KV read/write logic to call those methods (use json.dumps/json.loads)
instead of pickle.loads/pickle.dumps, and validate the deserialized shape before
constructing objects to avoid executing arbitrary code.
In `@app/jobs/reset_node_usage.py`:
- Around line 39-41: The call to the async coroutine connect_single_node is
incorrect: change the invocation inside the NodeStatus.connecting branch to
await the coroutine and pass the DB and node id parameters expected by the
signature; specifically replace the unawaited
node_operator.connect_single_node(node) with await
node_operator.connect_single_node(db, node.id) so the db from the enclosing
scope is used and the coroutine is awaited.
In `@app/lifecycle.py`:
- Around line 19-46: The lifespan function currently uses
func.__code__.co_varnames to detect an "app" parameter which can mis-detect
locals and fails for keyword-only params; update the logic in lifespan (handling
both startup_functions and shutdown_functions) to inspect each callable via
inspect.signature(func) and check for a parameter named "app" in the
signature.parameters mapping, then call the function with app passed as a
keyword argument (e.g., await func(app=app) or func(app=app)) when the parameter
exists, otherwise call with no args; preserve the existing coroutine vs sync
branching (asyncio.iscoroutinefunction) and ensure all calls use keyword passing
to avoid positional/kw-only issues.
In `@app/node/worker.py`:
- Around line 303-325: In _stream_logs, avoid leaking and calling result() on
cancelled tasks by creating explicit tasks for log_queue.get() and
stop_event.wait(), awaiting asyncio.wait(..., return_when=FIRST_COMPLETED), then
cancel any pending tasks and await them (suppressing asyncio.CancelledError via
contextlib.suppress) before accessing results; only call .result() on the task
that corresponds to log_queue.get() (handle the case where stop_event finished
first by breaking without reading result), and ensure you add "import
contextlib" at the top of the file so you can suppress cancelled-task exceptions
when cleaning up.
In `@node_worker.py`:
- Around line 20-29: The main() function currently only catches
KeyboardInterrupt and therefore won't exit cleanly on SIGTERM; modify main() to
register a SIGTERM handler that sets an asyncio.Event (e.g., shutdown_event) or
calls loop.stop so the while True sleep loop can break and allow async with
lifespan(app) to run shutdown hooks; use
asyncio.get_running_loop().add_signal_handler(signal.SIGTERM,
shutdown_event.set) (and also handle SIGINT similarly) and change the loop to
await shutdown_event.wait() (or check the event inside the sleep loop) so both
SIGTERM and KeyboardInterrupt trigger a graceful shutdown via lifespan(app) and
logger warnings remain unchanged.
🟡 Minor comments (16)
.github/workflows/test-database-migrations.yml-265-293 (1)
265-293:⚠️ Potential issue | 🟡 MinorAvoid logging DB passwords in CI output.
params_nativeincludes the password; redact it before printing to logs.Proposed fix
- print("Connecting with params:", params_native) + safe_params = dict(params_native) + if "password" in safe_params: + safe_params["password"] = "***" + print("Connecting with params:", safe_params) try: conn = pymysql.connect(**params_native) except Exception as exc: - print("FAILED to connect with params:", params_native) + print("FAILED to connect with params:", safe_params) raise.github/workflows/test-database-migrations.yml-24-28 (1)
24-28:⚠️ Potential issue | 🟡 MinorQuote
$GITHUB_ENVand group the writes.Shellcheck flags SC2086/SC2129 here; grouping the
echos and quoting the env file avoids word-splitting and repeated redirects. Apply the same pattern to the other “Set DATABASE_URL” steps.Proposed fix
- echo "TEST_FROM=github" >> $GITHUB_ENV - echo "TESTING=1" >> $GITHUB_ENV - echo "SQLALCHEMY_DATABASE_URL=sqlite+aiosqlite:///./test.db" >> $GITHUB_ENV + { + echo "TEST_FROM=github" + echo "TESTING=1" + echo "SQLALCHEMY_DATABASE_URL=sqlite+aiosqlite:///./test.db" + } >> "$GITHUB_ENV"healthcheck.sh-15-41 (1)
15-41:⚠️ Potential issue | 🟡 MinorAdd a curl timeout to avoid hanging healthchecks.
A stalled connection can block the script until Docker’s timeout; add a short, configurable timeout to keep probes responsive.Suggested fix
-DEBUG=${DEBUG:-false} +DEBUG=${DEBUG:-false} +TIMEOUT=${HEALTHCHECK_TIMEOUT:-5} @@ - local curl_flags="-sf" + local curl_flags="-sf --max-time $TIMEOUT" @@ - curl -sf --unix-socket "$socket_path" "http://localhost/health" 2>/dev/null + curl -sf --max-time "$TIMEOUT" --unix-socket "$socket_path" "http://localhost/health" 2>/dev/nullconfig.py-128-129 (1)
128-129:⚠️ Potential issue | 🟡 MinorTypo in comment.
"featueres" should be "features".
-## Experimental featueres +## Experimental featuresapp/notification/queue_manager.py-100-106 (1)
100-106:⚠️ Potential issue | 🟡 Minor
timeout=0will cause indefinite blocking.The condition
if timeout:evaluates toFalsewhentimeout=0, causing the method to fall through toawait self.q.get()which blocks indefinitely. Iftimeout=0is intended to mean "return immediately if empty", this is a bug.🐛 Proposed fix
async def dequeue(self, timeout: int | None = None): - if timeout: + if timeout is not None: try: return await asyncio.wait_for(self.q.get(), timeout=timeout) except asyncio.TimeoutError: return None return await self.q.get()app/telegram/__init__.py-122-127 (1)
122-127:⚠️ Potential issue | 🟡 MinorCancelled polling task should be awaited for clean shutdown.
After cancelling
_polling_task, the code doesn't await it. This can leave the task in a pending state and may cause "Task was destroyed but it is pending" warnings.🐛 Proposed fix
if self._polling_task is not None and not self._polling_task.done(): logger.info("stopping long polling") # Force stop the dispatcher first await self._dp.stop_polling() # Cancel the polling task self._polling_task.cancel() + try: + await self._polling_task + except asyncio.CancelledError: + passstart.sh-3-8 (1)
3-8:⚠️ Potential issue | 🟡 MinorBoolean environment variables from docker-compose won't be recognized in start.sh.
The docker-compose files setRUN_SCHEDULER: TrueandRUN_SCHEDULER: False, but the shell script checks[ "${RUN_SCHEDULER:-0}" = 1 ], which only matches the literal string"1". When Docker Compose passes these as environment variables, they become the strings"True"or"False", causing the conditions to fail and fall back to default behavior.♻️ Suggested normalization
ROLE="${NODE_ROLE:-panel}" -if [ "${ROLE}" = "node-worker" ] || [ "${IS_NODE_WORKER:-0}" = 1 ]; then +is_truthy() { + case "${1,,}" in + 1|true|yes|y|on) return 0 ;; + *) return 1 ;; + esac +} + +if [ "${ROLE}" = "node-worker" ] || is_truthy "${IS_NODE_WORKER:-0}"; then python node_worker.py -elif [ "${ROLE}" = "scheduler" ] || [ "${RUN_SCHEDULER:-0}" = 1 ]; then +elif [ "${ROLE}" = "scheduler" ] || is_truthy "${RUN_SCHEDULER:-0}"; then python scheduler_worker.py elseapp/operation/__init__.py-51-55 (1)
51-55:⚠️ Potential issue | 🟡 MinorGuard against non‑int RPC codes.
If
exc.codeisNoneor a non‑int,raise_errorwill hit aTypeErrorwhen comparingcode <= 0. Coerce defensively and fall back to 500.🛡️ Suggested fix
async def handle_rpc_error(self, exc: RuntimeError): """Convert NATS RPC errors to appropriate HTTP responses.""" - code = getattr(exc, "code", 500) - await self.raise_error(message=str(exc), code=code) + raw_code = getattr(exc, "code", None) + try: + code = int(raw_code) if raw_code is not None else 500 + except (TypeError, ValueError): + code = 500 + await self.raise_error(message=str(exc), code=code)scheduler_worker.py-15-24 (1)
15-24:⚠️ Potential issue | 🟡 MinorMake the NATS-disabled warning reachable (or remove it).
create_app(role="scheduler")runs at import time, so if NATS is disabled it raises beforemain()logs the warning or enters lifespan handling. Move app creation intomain()(or drop the warning) to keep behavior consistent.🔧 Proposed adjustment
- app = create_app(role="scheduler") - async def main(): if not is_nats_enabled(): logger.warning( "NATS is disabled; notification dispatching will only work when the scheduler shares a process with the API." ) + app = create_app(role="scheduler") async with lifespan(app): try: while True: await asyncio.sleep(3600) except KeyboardInterrupt: passtests/api/helpers.py-87-116 (1)
87-116:⚠️ Potential issue | 🟡 MinorFail fast when inbounds never appear during polling.
When
_WAIT_FOR_INBOUNDSis true,_poll()can return an empty list after retries, which may let tests proceed with invalid state and become flaky. Consider raising an assertion on timeout to surface readiness failures explicitly.✅ Suggested assertion on timeout
def _poll() -> list[str]: last_data: list[str] = [] for _ in range(_INBOUNDS_RETRIES): code, data = _fetch() if code == status.HTTP_200_OK: last_data = data if last_data: return last_data time.sleep(_INBOUNDS_DELAY_SEC) - return last_data + if _WAIT_FOR_INBOUNDS and not last_data: + raise AssertionError("Timed out waiting for /api/inbounds to be populated") + return last_datamain.py-27-33 (1)
27-33:⚠️ Potential issue | 🟡 MinorType mismatch:
require_nats_if_multiworkerexpects a boolean, but receives an integer.The function signature from
app/nats/__init__.pyshowsrequire_nats_if_multiworker(multi_worker: bool), but here you're passingworkers(an int). The function checksif multi_worker and not is_nats_enabled(), which will evaluate truthy for anyworkers > 0, but this is semantically incorrect for single-worker mode.🐛 Proposed fix
-require_nats_if_multiworker(workers) +require_nats_if_multiworker(workers > 1)app/operation/node.py-349-351 (1)
349-351:⚠️ Potential issue | 🟡 MinorDuplicate definition of
_connect_single_node_remote.The method
_connect_single_node_remoteis defined twice: at lines 349-350 and again at lines 613-614. Both have identical implementations. Remove the duplicate.Proposed fix - remove duplicate at lines 349-351
- async def _connect_single_node_remote(self, db: AsyncSession, node_id: int) -> None: - await node_nats_client.publish("connect_node", {"node_id": node_id}) - async def disconnect_single_node(self, node_id: int) -> None:Also applies to: 613-614
app/operation/node.py-556-557 (1)
556-557:⚠️ Potential issue | 🟡 MinorInconsistent user count calculation for max_message_size.
In
_connect_single_node_local,max_message_sizeis calculated usinglen(users)(line 557), but in_connect_nodes_bulk_local, it usesget_users_count_by_status(lines 481-483). Theuserslist fromcore_users()may not equal the count of active users from the database query, causing inconsistent sizing.Consider using consistent logic across both methods, preferably the explicit
get_users_count_by_statusapproach for accuracy.app/operation/node.py-818-822 (1)
818-822:⚠️ Potential issue | 🟡 MinorMissing error handling in
_update_core_remote.Unlike other remote methods (e.g.,
_update_node_api_remoteat line 802-806),_update_core_remotedoesn't wrap the RPC call in try/except withhandle_rpc_error. This could cause unhandledRuntimeErrorto propagate.Proposed fix
async def _update_core_remote(self, node_id: int, node_core_update: NodeCoreUpdate) -> dict: - return await node_nats_client.request( - "update_core", - {"node_id": node_id, "core_update": node_core_update.model_dump(mode="json")}, - ) + try: + return await node_nats_client.request( + "update_core", + {"node_id": node_id, "core_update": node_core_update.model_dump(mode="json")}, + ) + except RuntimeError as exc: + await self.handle_rpc_error(exc)app/operation/node.py-834-838 (1)
834-838:⚠️ Potential issue | 🟡 MinorMissing error handling in
_update_geofiles_remote.Same issue as
_update_core_remote- the method lacks try/except error handling for RPC failures.Proposed fix
async def _update_geofiles_remote(self, node_id: int, node_geofiles_update: NodeGeoFilesUpdate) -> dict: - return await node_nats_client.request( - "update_geofiles", - {"node_id": node_id, "geofiles_update": node_geofiles_update.model_dump(mode="json")}, - ) + try: + return await node_nats_client.request( + "update_geofiles", + {"node_id": node_id, "geofiles_update": node_geofiles_update.model_dump(mode="json")}, + ) + except RuntimeError as exc: + await self.handle_rpc_error(exc)app/node/worker.py-298-301 (1)
298-301:⚠️ Potential issue | 🟡 MinorStop subscription is never unsubscribed.
The subscription created for
stop_subjectat line 301 is never cleaned up. This could lead to subscription leaks if many log streams are started/stopped.Proposed fix - store and unsubscribe
+ stop_sub = await self._nc.subscribe(stop_subject, cb=_stop_cb) async def _stream_logs(): try: # ... existing code ... finally: + await stop_sub.unsubscribe() self._stop_events.pop(log_subject, None) self._log_tasks.pop(log_subject, None)
🧹 Nitpick comments (32)
dashboard/__init__.py (1)
57-63: Potential duplicate startup registration ifsetup_dashboardis called multiple times.The
@on_startupdecorator is applied insidesetup_dashboard, so each call tosetup_dashboard(app)will register a newrun_dashboardfunction tostartup_functions. Ifsetup_dashboardis invoked more than once (e.g., during tests or reloads), multiple handlers will accumulate.Consider moving the decorator outside or adding a guard:
♻️ Suggested refactor
-def setup_dashboard(app): - `@on_startup` - def run_dashboard(): - if DEBUG: - run_dev() - else: - run_build(app) +_dashboard_app = None + +@on_startup +def run_dashboard(): + if _dashboard_app is None: + return + if DEBUG: + run_dev() + else: + run_build(_dashboard_app) + +def setup_dashboard(app): + global _dashboard_app + _dashboard_app = appapp/notification/queue_manager.py (2)
82-88: Silent exception handling loses debugging context.Parse errors and other exceptions in
dequeueare silently swallowed. Consider logging these exceptions at debug/warning level to aid troubleshooting in production.♻️ Suggested improvement
except Exception: - await msg.nak() # Negative ack on parse error + await msg.nak() + logger.warning("Failed to parse notification message", exc_info=True) return None - except asyncio.TimeoutError: - return None - except Exception: + except asyncio.TimeoutError: + return None + except Exception: + logger.debug("Error fetching from notification queue", exc_info=True) return None
43-50: Use a specific exception type instead of catching bareException.Catching
Exceptionwhen adding a stream is fragile and may mask other errors (permissions issues, storage limits, configuration conflicts, etc.). In nats-py 2.12.0+, usenats.js.errors.APIErrorto catch JetStream-specific errors. For better clarity, consider checking if the stream exists first usingstream_info()before callingadd_stream(), which also handles edge cases like subject overlaps (error 10065) or misconfigured streams (error 10058).app/telegram/__init__.py (1)
86-87: SilentRuntimeErrorcatch obscures registration failures.The
except RuntimeError: passblock hides potential issues during handler/middleware registration. Consider logging at debug level to aid troubleshooting.app/settings/__init__.py (1)
74-77: Narrow the exception type toImportError.Using a bare
except Exception:is overly broad and may mask unrelated errors during the import or subsequentsync_from_settings()call. The intent appears to be handling cases where the telegram module isn't available.♻️ Suggested fix
try: from app.telegram import telegram_bot_manager + await telegram_bot_manager.sync_from_settings() - except Exception: + except ImportError: return - await telegram_bot_manager.sync_from_settings().env.example (1)
82-82: Add a trailing newline at end of file.Per POSIX convention and dotenv-linter, files should end with a newline character.
Proposed fix
# STOP_NODES_ON_SHUTDOWN=True +start.sh (1)
6-19: Considerexecfor Python entrypoints to preserve signal handling.
This lets the Python process receive SIGTERM/SIGINT directly; apply to all branches.♻️ Suggested update
- python node_worker.py + exec python node_worker.py elif [ "${ROLE}" = "scheduler" ] || [ "${RUN_SCHEDULER:-0}" = 1 ]; then - python scheduler_worker.py + exec python scheduler_worker.py else @@ - python main.py + exec python main.py fidocker-compose.multi.yml (1)
6-7: Consider restricting NATS port exposure.Port 4222 is exposed publicly. If NATS is only used for internal inter-service communication, consider binding to localhost only (
127.0.0.1:4222:4222) or removing the port mapping entirely since services communicate via Docker's internal network.🔒 Proposed fix for internal-only NATS access
ports: - - "4222:4222" + - "127.0.0.1:4222:4222"Or remove the
portssection entirely if external access is not needed.app/scheduler.py (1)
5-8: Remove unused logger.The
loggervariable is created but never used in this module.🧹 Proposed fix to remove unused import and variable
-from app.utils.logger import get_logger from config import SCHEDULER_JOBSTORE_URL, SQLALCHEMY_DATABASE_URL -logger = get_logger("scheduler") -app/node/user.py (1)
18-18: Parameteridshadows Python built-in.Consider renaming the parameter to
user_idto avoid shadowing the built-inidfunction.🧹 Proposed fix to avoid shadowing built-in
-def _serialize_user_for_node(id: int, username: str, user_settings: dict, inbounds: list[str] = None) -> ProtoUser: +def _serialize_user_for_node(user_id: int, username: str, user_settings: dict, inbounds: list[str] = None) -> ProtoUser: vmess_settings = user_settings.get("vmess", {}) vless_settings = user_settings.get("vless", {}) trojan_settings = user_settings.get("trojan", {}) shadowsocks_settings = user_settings.get("shadowsocks", {}) return create_user( - f"{id}.{username}", + f"{user_id}.{username}",app/nats/client.py (2)
8-13: Consider explicit connection configuration for resilience, but defaults already provide good recovery behavior.The
nats.connect()call relies on reasonable defaults (connect_timeout=2,max_reconnect_attempts=60,reconnect_time_wait=2), which handle network hiccups in multi-worker environments. However, if explicit configuration is desired, customize conservatively to avoid reducing resilience—the defaultmax_reconnect_attempts=60is preferable to a lower value like 10. Aconnect_timeoutincrease to 5 seconds may be beneficial for slower networks.
21-27: Improve exception handling specificity.The broad
except Exceptioncatches all errors, which may mask unexpected failures. Whencreate_key_value()is called on an existing bucket, nats-py raisesnats.js.errors.APIErrorwith error code10058. Catch this specific exception and check the error code instead of catching all exceptions.🔧 Proposed fix for specific exception handling
+from nats.js.errors import APIError + async def get_or_create_kv_bucket(js: JetStreamContext, bucket_name: str) -> KeyValue | None: """Get or create a JetStream KV bucket.""" try: return await js.create_key_value(bucket=bucket_name) - except Exception: - # Bucket already exists + except APIError as e: + if e.err_code == 10058: # stream name already in use + return await js.key_value(bucket=bucket_name) + raiseapp/nats/node_rpc.py (2)
43-60: Consider using a custom exception class instead of attaching attributes to RuntimeError.Attaching a
codeattribute to aRuntimeErrorinstance (line 57) works but is unconventional and can be error-prone for callers who may not expect this attribute. A custom exception class would be cleaner and more explicit.♻️ Proposed refactor using custom exception
+class NodeRPCError(Exception): + def __init__(self, message: str, code: int = 500): + super().__init__(message) + self.code = code + + class NodeNatsClient: # ... existing code ... async def request(self, action: str, payload: dict, timeout: float | None = None) -> dict: # ... existing code ... if not response.get("ok", False): error_msg = response.get("error", "Node RPC error") error_code = response.get("code", 500) - exc = RuntimeError(error_msg) - exc.code = error_code # Attach code to exception for caller to handle - raise exc + raise NodeRPCError(error_msg, error_code)
62-65: Minor: Consider draining the client before closing.The NATS client may have pending messages. Consider calling
drain()beforeclose()to ensure all pending messages are flushed.♻️ Proposed change
async def close(self): if self._nc and not self._nc.is_closed: + await self._nc.drain() await self._nc.close() self._nc = Nonemain.py (1)
27-33: Module-level side effects may cause issues during imports.The worker validation, NATS check, and app creation run at module level, which means they execute whenever
main.pyis imported (e.g., by test frameworks or tooling). Consider moving this logic inside theif __name__ == "__main__":block, or guard it appropriately.♻️ Proposed refactor
-workers = UVICORN_WORKERS or 1 -if workers < 1: - logger.warning(f"Invalid UVICORN_WORKERS value '{UVICORN_WORKERS}', defaulting to 1.") - workers = 1 -require_nats_if_multiworker(workers) - -app = create_app(role="panel") +app = None +workers = 1 +def _init_app(): + global app, workers + workers = UVICORN_WORKERS or 1 + if workers < 1: + logger.warning(f"Invalid UVICORN_WORKERS value '{UVICORN_WORKERS}', defaulting to 1.") + workers = 1 + require_nats_if_multiworker(workers > 1) + app = create_app(role="panel") + return app # ... rest of file ... if __name__ == "__main__": + _init_app() # Validate UVICORN_SSL_CA_TYPE valueNote: If uvicorn requires the
appto be available at module level for the"main:app"string reference, you may need to keepapp = create_app(role="panel")at module level but move only the validation/NATS check inside the guard.app/jobs/process_notification_queues.py (1)
19-24: Consider adding exception handling to prevent a single notification failure from breaking the drain loop.If
process_notification(item)raises an exception, the loop will terminate and remaining queued items won't be processed. Consider wrapping the call in a try-except to ensure all items are attempted.♻️ Proposed fix
queue = get_queue() while True: item = await queue.dequeue(timeout=0.1) if not item: break - await process_notification(item) + try: + await process_notification(item) + except Exception as exc: + logger.error(f"Failed to process notification item: {exc}")app/node/sync.py (1)
6-6: Importing a private function from another module breaks encapsulation.
_serialize_user_for_nodeis prefixed with an underscore, indicating it's intended to be private toapp.node.user. Consider either making it public (removing the underscore) or exposing a public wrapper in the source module.app/core/manager.py (2)
100-118: Fragile duck-typing with inner class_PayloadCore.The inner class
_PayloadCoremimics theCoreConfiginterface to pass data to_update_core_local. This is brittle—ifCoreConfigor_update_core_localchanges, this could silently break. Consider creating a proper dataclass or using the actualCoreConfigmodel.♻️ Proposed refactor using a dataclass
+from dataclasses import dataclass + +@dataclass +class CorePayload: + """Lightweight payload for core updates from NATS messages.""" + id: int + config: dict + exclude_inbound_tags: set + fallbacks_inbound_tags: set async def _apply_core_payload(self, payload: dict): try: core_id = payload["id"] config = payload["config"] except Exception: await self._reload_from_cache() return exclude_tags = set(payload.get("exclude_inbound_tags") or []) fallback_tags = set(payload.get("fallbacks_inbound_tags") or []) - class _PayloadCore: - def __init__(self, cid, cfg, exclude, fallbacks): - self.id = cid - self.config = cfg - self.exclude_inbound_tags = exclude - self.fallbacks_inbound_tags = fallbacks - - await self._update_core_local(_PayloadCore(core_id, config, exclude_tags, fallback_tags)) + await self._update_core_local(CorePayload(core_id, config, exclude_tags, fallback_tags))
271-275: Accessing private attribute_ncfrom outside the class.The shutdown function accesses
core_manager._ncdirectly, breaking encapsulation. Consider adding a publicclose()orshutdown()method toCoreManager.♻️ Proposed refactor
class CoreManager: # ... existing methods ... + + async def shutdown(self): + """Clean up resources on shutdown.""" + if self._nc and not self._nc.is_closed: + await self._nc.close() # ... at module level ... `@on_shutdown` async def shutdown_core_manager(): - # Close NATS connection - if core_manager._nc: - await core_manager._nc.close() + await core_manager.shutdown()app/notification/client.py (3)
152-163: Silent failure on send errors.When
_send_discord_webhook_directreturnsFalse(indicating a failure after retries), the function silently returns without logging the failure. Consider adding error logging for failed deliveries to aid debugging.Proposed improvement
if success: logger.debug("Discord notification delivered") + else: + logger.warning(f"Discord notification failed for webhook: {notification.webhook[:50]}...")
165-179: Silent failure on Telegram send errors.Similar to Discord, when
_send_telegram_message_directreturnsFalse, there's no logging of the failure. This makes it harder to diagnose delivery issues.Proposed improvement
if success: logger.debug("Telegram notification delivered") + else: + logger.warning(f"Telegram notification failed for chat_id: {notification.chat_id}")
182-197: Swallowed exceptions lose context for debugging.The
process_notificationfunction catches all exceptions and logs a generic message. For production debugging, consider logging the notification type and preserving the stack trace.Proposed improvement
except Exception as err: - logger.error(f"Failed to process notification: {err}") + logger.error(f"Failed to process notification (type={item.get('type')}): {err}", exc_info=True)app/node/__init__.py (1)
109-112:_update_usersiterates sequentially over nodes.The internal
_update_usersmethod awaits eachnode.update_users(users)call sequentially. For many nodes, this could be slow. Consider parallelizing withasyncio.gather.Proposed parallel implementation
async def _update_users(self, users: list): async with self._lock.reader_lock: - for node in self._nodes.values(): - await node.update_users(users) + await asyncio.gather( + *(node.update_users(users) for node in self._nodes.values()), + return_exceptions=True + )app/nats/router.py (3)
24-28: Lazy client creation lacks connection validation.
_get_clientcaches the client but doesn't verify the connection is still alive on subsequent calls. If the connection drops, stale client references may cause silent failures.Proposed improvement
async def _get_client(self) -> nats.NATS | None: """Get or create NATS client.""" - if not self._nc: + if not self._nc or self._nc.is_closed: self._nc = await create_nats_client() return self._nc
72-84: Race condition between_runningflag and task state.The
startmethod checksself._runningandself._listener_task.done()separately, but_runningis set toFalsein thefinallyblock of_listen. If_listenexits unexpectedly (e.g., exception),_runningbecomesFalsebut the task object still exists. The checkif self._listener_task and not self._listener_task.done()would prevent restart. Consider simplifying to rely only on task state.Proposed simplification
async def start(self): """Start the router listener.""" if not MULTI_WORKER: return - if self._running: - return - if self._listener_task and not self._listener_task.done(): return self._running = True self._listener_task = asyncio.create_task(self._listen())
101-114: Publish silently fails without propagating errors to caller.The
publishmethod catches exceptions and only logs a warning. Callers have no way to know if publishing failed. Consider at least returning a boolean or offering araise_on_erroroption for critical messages.app/node/worker.py (1)
139-148: Error code inference from string matching is fragile.The RPC error handling uses string matching (
"NotFound" in error_msg) to determine HTTP status codes. This is brittle and may misclassify errors if message text changes or contains false matches.Consider propagating structured error codes from the source exceptions (e.g.,
NodeAPIError.code) rather than parsing strings. If the underlying exceptions already have codes, preserve them through the call chain.app/core/hosts.py (3)
293-316:_load_state_from_cacheswallows all exceptions.The bare
except Exceptionat line 315 silently catches all errors including potentially important ones (e.g., data corruption). Consider logging the exception.Proposed improvement
- except Exception: + except Exception as exc: + self._logger.warning(f"Failed to load state from cache: {exc}") return False
444-455: Duplicate logic in_add_hosts_natsand_add_hosts_local.The two methods share nearly identical code for preparing hosts. Consider extracting the common preparation logic into a shared helper to reduce duplication.
497-503: Accessing private_ncattribute from outside the class.The
shutdown_hostsfunction directly accesseshost_manager._nc. Consider adding a publicclose()orshutdown()method to theHostManagerclass for cleaner encapsulation.Proposed improvement
Add to
HostManager:async def close(self): """Close NATS connection if open.""" if self._nc: await self._nc.close() self._nc = NoneThen in
shutdown_hosts:async def shutdown_hosts(): if IS_NODE_WORKER: return - # Close NATS connection - if host_manager._nc: - await host_manager._nc.close() + await host_manager.close()app/operation/node.py (2)
635-636:_get_logs_remoteraises 409 Conflict - consider 501 Not Implemented.Returning HTTP 409 (Conflict) for an unsupported operation is semantically incorrect. HTTP 501 (Not Implemented) or a custom error would be more appropriate.
Proposed fix
async def _get_logs_remote(self, node_id: Node) -> Callable[[], AsyncIterator[asyncio.Queue]]: - await self.raise_error(message="Node logs are only available via node-worker", code=409) + await self.raise_error(message="Node logs are only available via node-worker", code=501)
372-382: Type annotation suggestsdt = Noneis a default, but it'sOptional.The type hint
start: dt = Noneis technically incorrect. It should bestart: dt | None = Nonefor clarity and correctness with type checkers.Proposed fix
async def get_usage( self, db: AsyncSession, - start: dt = None, - end: dt = None, + start: dt | None = None, + end: dt | None = None, period: Period = Period.hour,
…g signal handlers and `exec` in the start script, while also improving asyncio compatibility.
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
There was a problem hiding this comment.
Actionable comments posted: 8
🤖 Fix all issues with AI agents
In `@app/core/hosts.py`:
- Around line 446-473: The NATS add path in _add_hosts_nats is publishing
SubscriptionInboundData objects directly which may not be JSON-safe and it also
never persists the KV state; modify the publish payloads to convert any
SubscriptionInboundData/Host model instances returned from _prepare_host_entry
into plain JSON-serializable primitives (e.g., via
model_dump()/model_validate()->dict or similar) before calling _publish, and
after updating local state (after _add_prepared_hosts_local and removing hosts
from self._hosts) call self._persist_state() to persist the KV so multi-worker
caches stay consistent; keep existing host_id-only removal publish behavior.
In `@app/core/xray.py`:
- Around line 415-437: to_json references self.fallbacks_inbound_tags which
isn't initialized and from_json passes invalid kwargs (_inbounds,
_inbounds_by_tag, _fallbacks_inbound) that __init__ doesn't accept; update
XRayConfig.__init__ to initialize fallbacks_inbound_tags (e.g., accept or set
self.fallbacks_inbound_tags = set(...) where appropriate) and modify from_json
to call cls with the real constructor parameter names (use inbounds=...,
inbounds_by_tag=..., fallbacks_inbound=... and pass
exclude_inbound_tags/fallbacks_inbound_tags as sets) so the reconstructed
instance matches the attributes used by to_json and the initializer.
In `@app/node/worker.py`:
- Around line 233-235: The RPC handler _get_nodes_system_stats has the wrong
signature for how handlers are invoked by _dispatch_rpc (which calls
handler(data)); update _get_nodes_system_stats to accept the dispatched payload
(e.g., add a second parameter like data=None or **kwargs) and keep the existing
behavior (ignore or validate data as needed), so the function becomes compatible
with _dispatch_rpc without changing its return transformation; reference
_get_nodes_system_stats and _dispatch_rpc when making the change.
- Around line 88-93: The shutdown loop currently cancels tasks in
self._log_tasks and immediately clears them, which can race with the
_stream_logs coroutine trying to publish to NATS; instead, after calling
task.cancel() collect the task objects from self._log_tasks.values() and await
their completion (e.g., await asyncio.gather(*tasks, return_exceptions=True))
before clearing self._log_tasks and self._stop_events and before closing the
NATS connection; also ensure _stream_logs handles asyncio.CancelledError/cleanup
in its finally block so awaited cancellations finish cleanly.
- Around line 299-302: The subscription for stop_subject created in _start_logs
via await self._nc.subscribe(stop_subject, cb=_stop_cb) is never stored or
unsubscribed causing a resource leak; modify _start_logs to capture the returned
subscription object (or subject sid) when calling subscribe, store it (e.g.,
local variable or attach to self like self._stop_sub), and ensure you call its
unsubscribe() or self._stop_sub.unsubscribe() in the cleanup/shutdown path that
stops log streaming (or when stop_event is set), and remove any references
afterwards so the subscription is cleaned up.
In `@docker-compose.multi.yml`:
- Around line 39-41: The node-worker service in docker-compose.multi.yml sets an
explicit container_name ("container_name: node-worker"), which prevents scaling;
remove the container_name entry from the node-worker service definition so
Docker Compose can create multiple instances (allowing docker compose up --scale
node-worker=N); ensure no other services rely on that fixed name (update any
references or networks if necessary) and keep the service name node-worker as
the identifier for scaling.
- Around line 10-14: The nats service healthcheck uses "nc" which isn't present
in the nats:2.10-alpine image; update the healthcheck block (healthcheck ->
test) to either use a shell TCP probe like ["/bin/sh","-c","exec
3<>/dev/tcp/localhost/4222"] or extend the image to install netcat (add
Dockerfile step RUN apk add --no-cache netcat-openbsd) so the existing test
["CMD","nc","-z","localhost","4222"] works; alternatively, point the healthcheck
to a bundled script (e.g., /code/healthcheck.sh) like
["CMD","/code/healthcheck.sh"] if that script is included in the image.
In `@start.sh`:
- Line 19: In start.sh change the final process invocation so the shell is
replaced by the Python process: replace the plain "python main.py" invocation
with an exec-style invocation so signals (SIGTERM) are delivered to the Python
process directly; update the panel startup line (the invocation of python
main.py) to use exec to match node-worker and scheduler behavior and enable
graceful shutdown.
🧹 Nitpick comments (5)
start.sh (1)
5-8: Dual environment variable patterns may cause confusion.The script checks both
NODE_ROLEand legacy-style flags (IS_NODE_WORKER,RUN_SCHEDULER). IfNODE_ROLEis set to something unexpected butIS_NODE_WORKER=1, the node-worker branch will still execute. This fallback pattern is fine for backward compatibility, but consider documenting the precedence or deprecating one approach to avoid ambiguity.scheduler_worker.py (1)
43-51: Consider simplifying to justasyncio.run().The fallback to
get_event_loop().run_until_complete()targets Python < 3.7, which reached end-of-life in June 2023. If the project's minimum Python version is 3.7+, this can be simplified.♻️ Simplified entry point for Python 3.7+
if __name__ == "__main__": try: - if hasattr(asyncio, "run"): - asyncio.run(main()) - else: - loop = asyncio.get_event_loop() - loop.run_until_complete(main()) + asyncio.run(main()) except KeyboardInterrupt: passnode_worker.py (1)
26-32: Guardadd_signal_handleron platforms where it’s unsupported.
loop.add_signal_handlercan raiseNotImplementedErroron some platforms (e.g., Windows). If Windows support is expected, wrap these calls to avoid crashing at startup.🔧 Proposed guard
loop = asyncio.get_running_loop() - loop.add_signal_handler(signal.SIGINT, handle_signal) - loop.add_signal_handler(signal.SIGTERM, handle_signal) + for sig in (signal.SIGINT, signal.SIGTERM): + try: + loop.add_signal_handler(sig, handle_signal) + except NotImplementedError: + passapp/node/worker.py (1)
42-43: UseCallablefromtypinginstead ofcallable.The type hint uses the built-in
callablefunction rather than theCallabletype from thetypingmodule. While this won't cause runtime errors, it's not the correct type annotation.Proposed fix
Add to imports:
from typing import CallableThen update the type hints:
- self._command_handlers: dict[str, callable] = {} - self._rpc_handlers: dict[str, callable] = {} + self._command_handlers: dict[str, Callable] = {} + self._rpc_handlers: dict[str, Callable] = {}docker-compose.multi.yml (1)
24-24: Use canonical boolean strings forRUN_SCHEDULER.Capitalized booleans are non-canonical in YAML 1.2 and may end up as strings; use
"true"/"false"to avoid parser ambiguity.✅ Suggested fix
- RUN_SCHEDULER: False + RUN_SCHEDULER: "false" ... - RUN_SCHEDULER: True + RUN_SCHEDULER: "true" ... - RUN_SCHEDULER: True + RUN_SCHEDULER: "true"Also applies to: 46-46, 60-60
| healthcheck: | ||
| test: ["CMD", "nc", "-z", "localhost", "4222"] | ||
| interval: 2s | ||
| timeout: 2s | ||
| retries: 10 |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
find . -name "docker-compose.multi.yml" -type fRepository: PasarGuard/panel
Length of output: 86
🏁 Script executed:
cat -n ./docker-compose.multi.ymlRepository: PasarGuard/panel
Length of output: 2025
🏁 Script executed:
# Search for healthcheck.sh in the codebase
find . -name "healthcheck.sh" -type fRepository: PasarGuard/panel
Length of output: 76
🏁 Script executed:
# Find Dockerfile
find . -name "Dockerfile*" -type fRepository: PasarGuard/panel
Length of output: 72
🏁 Script executed:
cat -n ./DockerfileRepository: PasarGuard/panel
Length of output: 1638
🏁 Script executed:
# Search for any documentation or references about nc in nats image
# First, let's check if there's any reference in the repo
rg "nc" --type md --type txt --type yamlRepository: PasarGuard/panel
Length of output: 1091
🏁 Script executed:
cat -n ./healthcheck.shRepository: PasarGuard/panel
Length of output: 2283
🌐 Web query:
nats:2.10-alpine netcat nc command availability
💡 Result:
In the official nats:2.10-alpine image, nc (netcat) is not guaranteed to be available by default—the Alpine variants intentionally keep extra utilities out unless explicitly installed. The published Dockerfile for the Alpine-based 2.10 images installs only minimal packages (for example ca-certificates and tzdata), and does not add netcat. [1], [2]
If you need nc, install it in a derived image, e.g.:
FROM nats:2.10-alpine
RUN apk add --no-cache netcat-openbsd(Alternative on Alpine: apk add --no-cache busybox-extras, depending on which nc you want.) [1]
Sources: [1] [2]
Remove nc from nats healthcheck or install it in the image.
The nats:2.10-alpine image does not include nc by default, causing the healthcheck on line 11 to fail. Either use an alternative command (e.g., /bin/sh -c 'exec 3<>/dev/tcp/localhost/4222') or extend the image to install netcat: RUN apk add --no-cache netcat-openbsd.
The /code/healthcheck.sh script in the panel service is correctly included in the image via the Dockerfile and will work as expected.
🤖 Prompt for AI Agents
In `@docker-compose.multi.yml` around lines 10 - 14, The nats service healthcheck
uses "nc" which isn't present in the nats:2.10-alpine image; update the
healthcheck block (healthcheck -> test) to either use a shell TCP probe like
["/bin/sh","-c","exec 3<>/dev/tcp/localhost/4222"] or extend the image to
install netcat (add Dockerfile step RUN apk add --no-cache netcat-openbsd) so
the existing test ["CMD","nc","-z","localhost","4222"] works; alternatively,
point the healthcheck to a bundled script (e.g., /code/healthcheck.sh) like
["CMD","/code/healthcheck.sh"] if that script is included in the image.
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@app/node/worker.py`:
- Around line 84-95: The stop() coroutine iterates directly over
self._stop_events and self._log_tasks while _stream_logs may pop from those
dicts concurrently, causing RuntimeError; fix by snapshotting the values before
iterating (use list(self._stop_events.values()) and
list(self._log_tasks.values())) and perform stop_event.set(), task.cancel(), and
the asyncio.gather on those snapshots, then clear/cleanup the original dicts
(e.g., self._log_tasks.clear()) as currently intended; reference: stop(),
_stop_events, _log_tasks, and _stream_logs.
🧹 Nitpick comments (4)
app/core/hosts.py (4)
319-322: Consider removing unreachable branch.Since
cached_statecomes fromjson.loads(),host_datawill always be a dict (or primitive). Theelsebranch at line 322 storinghost_datadirectly is unreachable for valid JSON payloads and could mask unexpected data.Suggested simplification
for host_id_str, host_data in cached_state.items(): try: host_id = int(host_id_str) - if isinstance(host_data, dict): - converted_state[host_id] = SubscriptionInboundData.model_validate(host_data) - else: - converted_state[host_id] = host_data + converted_state[host_id] = SubscriptionInboundData.model_validate(host_data) except (ValueError, TypeError):
340-356: Add defensive handling for malformed messages.If
host_entryis present but missing"id"or"data"keys, aKeyErrorwill propagate. While this is an internal message, defensive handling improves resilience.Suggested improvement
elif action == "add": host_entry = data.get("host") - if host_entry: - await self._add_prepared_hosts_local([(host_entry["id"], host_entry["data"])]) + if host_entry and "id" in host_entry and "data" in host_entry: + await self._add_prepared_hosts_local([(host_entry["id"], host_entry["data"])]) else: await self._reload_from_cache()
461-477: Consider batching host updates for improved performance.Publishing messages one-by-one in a loop (lines 462-467) could be slow with many hosts. Additionally, the redundant
_reset_cache()call at line 477 is unnecessary since_add_prepared_hosts_localalready resets the cache.Suggested optimization
# Persist state to NATS KV await self._persist_state() - await self._reset_cache()
519-525: Consider adding error handling for NATS connection close.If
_nc.close()raises an exception during shutdown, it could prevent clean termination. A try-except wrapper would improve robustness.Suggested improvement
`@on_shutdown` async def shutdown_hosts(): if IS_NODE_WORKER: return # Close NATS connection if host_manager._nc: - await host_manager._nc.close() + try: + await host_manager._nc.close() + except Exception as exc: + host_manager._logger.warning(f"Error closing NATS connection: {exc}")
| async def stop(self): | ||
| if not IS_NODE_WORKER: | ||
| return | ||
|
|
||
| for stop_event in self._stop_events.values(): | ||
| stop_event.set() | ||
| for task in self._log_tasks.values(): | ||
| task.cancel() | ||
| # Wait for all cancelled tasks to complete their cleanup | ||
| if self._log_tasks: | ||
| await asyncio.gather(*self._log_tasks.values(), return_exceptions=True) | ||
| self._log_tasks.clear() |
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
# First, let's find and examine the worker.py file
find . -name "worker.py" -path "*/app/node/*" | head -5Repository: PasarGuard/panel
Length of output: 80
🏁 Script executed:
# Read the worker.py file to understand the context
wc -l app/node/worker.pyRepository: PasarGuard/panel
Length of output: 82
🏁 Script executed:
# Read the relevant section and more context around the stop() method
cat -n app/node/worker.py | sed -n '1,120p'Repository: PasarGuard/panel
Length of output: 5751
🏁 Script executed:
# Search for _stream_logs method to verify if it mutates _stop_events or _log_tasks
rg "_stream_logs|_stop_events|_log_tasks" app/node/worker.py -A 5 -B 2Repository: PasarGuard/panel
Length of output: 2082
Snapshot dict values before iterating during shutdown to prevent RuntimeError: dictionary changed size during iteration.
The _stream_logs coroutine pops from _stop_events and _log_tasks during cleanup, and these pops can occur while stop() iterates over the same dictionaries. When stop_event.set() is called, it unblocks waiting tasks that then pop from the dicts concurrently with iteration, causing a runtime crash.
🛠️ Suggested fix
- for stop_event in self._stop_events.values():
+ for stop_event in list(self._stop_events.values()):
stop_event.set()
- for task in self._log_tasks.values():
+ for task in list(self._log_tasks.values()):
task.cancel()
# Wait for all cancelled tasks to complete their cleanup
if self._log_tasks:
- await asyncio.gather(*self._log_tasks.values(), return_exceptions=True)
+ await asyncio.gather(*list(self._log_tasks.values()), return_exceptions=True)🤖 Prompt for AI Agents
In `@app/node/worker.py` around lines 84 - 95, The stop() coroutine iterates
directly over self._stop_events and self._log_tasks while _stream_logs may pop
from those dicts concurrently, causing RuntimeError; fix by snapshotting the
values before iterating (use list(self._stop_events.values()) and
list(self._log_tasks.values())) and perform stop_event.set(), task.cancel(), and
the asyncio.gather on those snapshots, then clear/cleanup the original dicts
(e.g., self._log_tasks.clear()) as currently intended; reference: stop(),
_stop_events, _log_tasks, and _stream_logs.
2 participants
Summary by CodeRabbit
New Features
Chores