31 Commits
Author SHA1 Message Date
Sebastian Monroy 207de4487c fix(worker_consumer): propagate submit failures to retry/DLQ and emit failed status
- raise a dedicated JobSubmissionError when ComfyUI prompt submission fails
- re-raise submission failures in process_message so consume loop does not delete SQS messages
- emit failed status updates on submit-time errors using status_policy.fail_status fallback
- remove failed-status dedupe guard so each retry attempt remains observable
- align parse-error logging with retry behavior ("Message will be retried")
2026-02-17 15:40:02 +00:00
Sebastian Monroy 62b71b2978 moved NILOR_WORKFLOW_OS_NORMALIZATION_ENABLED param to config 2026-01-15 22:52:37 +00:00
Sebastian Monroy 552252b84c worker consumer normalizes incoming workflows based on operating system 2026-01-15 21:17:35 +00:00
Sebastian Monroy 527e8dc4ab 663 implement comfyui memory guardian to prevent oom errors (#12)
* feat(config): add Memory Hygiene config scaffolding and defaults for nilor-nodes

- introduce MemoryHygieneConfig and wire into NilorNodesConfig
- parse NILOR_MEMORY_HYGIENE_* from JSON5 and apply env overrides
- add validation for thresholds, policy, cooldowns, retries, and durations
- extend config.json5 with sane Memory Guardian defaults
- update .env.example

* feat(client): add ComfyUI capability detection and supports_hygiene cache

- add one-time probe for /system_stats and /free, cached per session
- expose supports_hygiene() that logs a single warning when unsupported
- use short timeouts and no retries; mark false only on 404/405
- leave transient failures retryable by keeping capability as unknown

* feat(memory): add MemoryHygiene module with typed skeleton and API

- introduce RemediationAction and RemediationResult dataclass
- add MemoryHygiene class with DI for client/config/logger
- implement check_and_remediate skeleton with enablement/capability checks
- add safe stats helper; defer policy/remediation to later commits
- export public symbols via all

* feat(memory): add metrics collector with usage pct and vram_total

- extend SystemStats with vram_total and parse from /system_stats
- add DerivedStats and collect_metrics() computing vram/ram used pct (0–100)
- integrate metrics collection in check_and_remediate skeleton
- safe math with clamping and None handling for incomplete stats

* feat(memory): implement policy engine thresholds, cooldown, and action selection

- add cooldown handling and respect it in check_and_remediate
- detect pressure via percent or absolute MB thresholds for vram/ram
- normalize policy and choose staged initial action (auto => free)
- return actionable RemediationResult with reason; no remediation yet
- helper functions for conversions and comparisons

* feat(memory): add remediation loop with retries, time caps, and cooldown

- implement remediate_cycle with free/unload flags and staged auto escalation
- respect MAX_RETRIES, SLEEP_BETWEEN_ATTEMPTS_SECONDS, MAX_CYCLE_DURATION_SECONDS
- set cooldown after cycle; return after-stats, attempts, action, and outcome reason
- integrate cycle into check_and_remediate; keep helpers in module

* feat(worker): wire MemoryHygiene into idle and post-completion paths

- initialize MemoryHygiene with comfy client and config in consume_jobs
- run hygiene before polling when idle and after prompt finalize with 1s debounce
- guard remediation by setting is_busy to block new intake; reset after
- keep websocket listener unaffected; schedule post-completion hygiene as background task

* feat(worker): throttle hygiene checks and add cadence tracking

- throttle idle hygiene by cfg.hygiene.idle_poll_seconds using monotonic clock
- add last_hygiene_check_ts to avoid overly frequent checks
- keep is_busy gating and post-completion debounce execution

* chore(memory): add structured logs for decisions and remediation

- log disabled/unsupported/cooldown/no-pressure branches
- log start/end of remediation cycles with before/after VRAM/RAM stats
- log each /free invocation flags; guard logging to avoid exceptions

* fix(memory): harden hygiene with session disable and single-warning on unsupported

- add _capability_disabled to short-circuit future runs after unsupported endpoints
- emit a single warning then quietly skip further cycles for the session
- preserve existing retry/backoff/cooldown and safe exception handling

* .env.example update

* improved get_system_stats and added more logging

* improve logging formatting for comfyui_client and memory_hygiene

* feat(nilor-nodes): add startup hygiene summary log; convert hygiene lambda to class method

- add WorkerConsumer._run_memory_hygiene() async method; remove late-bound lambda injection
- delegate to guarded helper to respect busy gate and optional debounce
- emit startup hygiene summary with effective thresholds from _CFG.hygiene
  (enabled, idle_poll_s, vram/ram pct caps, min_free_mb, policy, retries, cooldown,
  sleep_between, max_cycle)
- keep existing call sites in consume_loop() and _finalize_prompt() using the new method
- no functional changes to remediation logic; new log improves observability at boot

* got rid of redundant .env loading in worker_consumer

* chore(config): simplify global config caching

- keep process-wide _CONFIG singleton for shared configuration instance
- no functional behavior change to config loading paths

* fixed valueerror logging in media_stream
2025-10-20 13:56:38 +01:00
Sebastian Monroy 5d05aeaf68 added a concise startup config log in worker_consumer.py 2025-10-16 20:56:06 +01:00
Sebastian Monroy 5668e09a79 removed legacy code that only existed for controlling rollout, idc bout dat 2025-10-16 20:51:03 +01:00
Sebastian Monroy 58ff7a0bce chore(flag): simplify websocket path to client iterator only
- use client.ws_connect exclusively; if disabled/not constructed, sleep and retry
- keep event parsing semantics and state transitions unchanged
- retain HTTP fallback gating; WS path now simplified for maintainability
2025-10-16 19:45:41 +01:00
Sebastian Monroy c33b06a092 refactor(worker/ws): consume websocket events via ComfyUILocalClient iterator
- replace manual websockets.connect loop with client.ws_connect iterator
- preserve parsing/side-effects for status/progress/executed/executing/success/error
- keep busy state, prompt_id mapping, and finalize logic unchanged
- rely on client’s bounded reconnect; log and retry on unexpected errors
2025-10-16 19:28:24 +01:00
Sebastian Monroy d5109ae0b3 refactor(worker): use ComfyUILocalClient for prompt submission
- import ComfyUILocalClient and ComfyUIClientError into worker
- construct a shared ComfyUILocalClient with the shared aiohttp session on startup
- replace direct POST to /prompt with client.submit_prompt(payload)
- preserve client_id injection, prompt_id mapping, and busy/current state updates
- catch ComfyUIClientError in submission error handling alongside aiohttp errors
2025-10-16 19:25:07 +01:00
Sebastian Monroy 4a56d4dff5 feat(nilor-nodes): integrate typed config loader into worker_consumer
- load Config once with env>JSON5 precedence and pass cfg into WorkerConsumer
- replace direct os.getenv reads with cfg.comfy and cfg.worker values
- compute Comfy HTTP/WS endpoints from cfg and reuse single aiohttp session
- keep behavior identical; no new features introduced
2025-10-16 17:36:42 +01:00
Sebastian Monroy bfba7b0265 fix(nilor-nodes): prevent ComfyUI websocket 1009 disconnects by allowing large frames
- Accept large preview frames from ComfyUI without dropping the socket.
- Binary preview frames are still ignored by the consumer; this change stops the
  connection from closing with code 1009 (message too big).
- Impact: keeps workers from getting stuck "busy" and allows long video job
  queues to finish reliably.

Notes:
- To reduce bandwidth instead of disabling client handling, start ComfyUI with
  --preview-method none or a smaller --preview-size.
2025-10-16 11:40:36 +01:00
Sebastian Monroy 848899298c update the worker consumer to poll jobs_to_process-comfyui and include job_type data 2025-10-14 14:57:19 +01:00
Sebastian Monroy 2333a0d45d refactor worker_consumer to never allow the consumption of other jobs while a job is already running. implemented robust prompt completion detection. reduced console noise. 2025-10-08 19:23:51 +01:00
Sebastian Monroy 6d17e664bb added handling for ClientError in worker_consumer 2025-10-08 15:32:46 +01:00
Sebastian Monroy 7beda857c8 added SQS_POLL_WAIT_TIME to .env 2025-10-08 15:31:42 +01:00
Sebastian Monroy 4fa9710aea forgot to include in last commit 2025-10-08 15:30:46 +01:00
Sebastian Monroy e8c0d451d9 added logger class 2025-10-08 15:30:09 +01:00
Sebastian Monroy 51777e3462 feat(workers): include required fields in status updates; honor optional status_policy
- copy content_id, venue, canvas, scene from job payload into status updates
- use running_status for first progress; fail_status on execution errors (fallbacks preserved)
- manage per-content context lifecycles
2025-09-26 16:06:33 +01:00
Sebastian Monroy 7806a92772 logging improvements for worker_consumer.py 2025-09-25 16:59:38 +01:00
Sebastian Monroy a42221565d refactor(nilor-nodes): adopt content_id in worker_consumer and MediaStreamOutput
- read `content_id` from job payloads in `worker_consumer.py` (replace `client_id`)
- rename internal mappings (e.g., `prompt_id_to_content_id_map`)
- MediaStreamOutput: require `content_id` input (replace `job_id`)
- completion messages: `{ "content_id": <uuid>, "status": "completed", "outputs": { ... } }`
- web widget: hide `content_id` instead of `job_id`
- BREAKING CHANGE: older payloads with `job_id`/`client_id` are no longer accepted
2025-09-25 16:10:06 +01:00
Sebastian Monroy 75a45c6c34 further improve logging everywhere, using thinspace instead of brackets to fix terminal rendering, including all raised errors 2025-09-25 11:50:16 +01:00
Sebastian Monroy 414679c676 simplified the logging when worker_consumer.py fails to initialize SQS, improved logging everywhere 2025-09-25 11:30:25 +01:00
Sebastian Monroy 5af42118fe add SQS_ENABLED flag to .env to toggle SQS functionality related to worker_consumer.py 2025-09-25 11:02:48 +01:00
Sebastian Monroy 35428a8287 black formatter pass on all python scripts 2025-09-10 14:47:43 +01:00
Sebastian Monroy 04ac2b655a set up websocket connection with ComfyUI so that it can report whether it has started a ComfyUI job, and then set job status to "running" via the queue 2025-09-08 14:33:55 +01:00
Sebastian Monroy cc9056e11c implement status update publishing to SQS
-   Adds configuration for the new `job_status_updates` queue in the worker consumer.
-   After successfully submitting a job to the local ComfyUI instance, the worker publishes a message with `{"status": "running"}`.
-   Includes a critical check for `client_id` in the workflow data to ensure a job ID is present before publishing.
-   Logs a warning and skips the update if `client_id` is missing, preventing silent failures.
-   Errors during the status update publication are logged but do not interrupt the primary job, maintaining system resilience.
-   Update .env.example
2025-09-08 12:16:57 +01:00
Sebastian Monroy 2387d0e0ad update worker_consumer.py to work with new SQS requirements and update .env.example 2025-08-08 16:54:09 +01:00
Sebastian Monroy 9e5ccb8bb7 added support for video to MediaStreamInput and MediaStreamOutput nodes 2025-08-04 14:25:03 +01:00
Sebastian Monroy 8f355e360c 3.3: successful end-to-end test of client initiating comfyui job, brain api creating the job, worker consuming the job, and comfui running the job and outputting to minio 2025-08-04 13:32:35 +01:00
Sebastian Monroy e8c5c449ac 3.2: full test of ComfyUI MediaStream nodes successful. got nodes to show up in ComfyUI properly. 2025-08-04 12:33:14 +01:00
Sebastian Monroy 2c7edfc535 added worker_consumer.py script which is responsible for connecting to ElasticMQ to poll for new comfyui jobs 2025-07-30 15:02:41 +01:00