diff --git a/containers/talkbot/server.py b/containers/talkbot/server.py index c68edbe..b50477d 100644 --- a/containers/talkbot/server.py +++ b/containers/talkbot/server.py @@ -4,7 +4,7 @@ import os import re import sqlite3 import threading -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import Optional import httpx @@ -52,6 +52,14 @@ QUEUE_DB_PATH = os.environ.get( "/app/data/queue.db", ) +# A job is retried up to this many times (across restarts, since the +# attempt count lives in the queue row) before being abandoned. Between +# attempts we wait RETRY_BACKOFF_SECONDS so a job that fails because a +# downstream service is briefly unavailable doesn't spin the worker in +# a tight loop. +MAX_ATTEMPTS = 3 +RETRY_BACKOFF_SECONDS = 20 + MANAGER_USER = NC_USER app = FastAPI(title="Nextcloud Talk voice translator") @@ -79,10 +87,25 @@ def init_queue_db() -> None: CREATE TABLE IF NOT EXISTS queue ( id INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT NOT NULL, - created_at TEXT NOT NULL + created_at TEXT NOT NULL, + attempts INTEGER NOT NULL DEFAULT 0, + next_attempt_at TEXT ) """ ) + + existing_columns = { + row[1] for row in conn.execute("PRAGMA table_info(queue)") + } + + if "attempts" not in existing_columns: + conn.execute( + "ALTER TABLE queue ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0" + ) + + if "next_attempt_at" not in existing_columns: + conn.execute("ALTER TABLE queue ADD COLUMN next_attempt_at TEXT") + conn.commit() @@ -100,17 +123,37 @@ def enqueue_event(job: dict) -> None: new_job_signal.set() -def dequeue_next_event() -> Optional[tuple[int, dict]]: +def dequeue_next_event() -> Optional[tuple[int, dict, int]]: + now = datetime.now(timezone.utc).isoformat() + with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn: row = conn.execute( - "SELECT id, payload FROM queue ORDER BY id LIMIT 1" + """ + SELECT id, payload, attempts FROM queue + WHERE next_attempt_at IS NULL OR next_attempt_at <= ? + ORDER BY id LIMIT 1 + """, + (now,), ).fetchone() if not row: return None - row_id, payload = row - return row_id, json.loads(payload) + row_id, payload, attempts = row + return row_id, json.loads(payload), attempts + + +def reschedule_event(row_id: int, attempts: int) -> None: + next_attempt_at = ( + datetime.now(timezone.utc) + timedelta(seconds=RETRY_BACKOFF_SECONDS) + ).isoformat() + + with sqlite3.connect(QUEUE_DB_PATH, timeout=30) as conn: + conn.execute( + "UPDATE queue SET attempts = ?, next_attempt_at = ? WHERE id = ?", + (attempts, next_attempt_at, row_id), + ) + conn.commit() def remove_queued_event(row_id: int) -> None: @@ -1145,6 +1188,25 @@ def process_event(job: dict) -> None: ) +def notify_processing_failure(job: dict) -> None: + token = job.get("token") + + if not token: + return + + try: + post_manager_message( + token, + "Non รจ stato possibile elaborare il messaggio.", + ) + except Exception: + log.exception( + "Could not post failure notice for message %s in conversation %s", + job.get("message_id"), + token, + ) + + def event_worker() -> None: while True: job_row = dequeue_next_event() @@ -1154,18 +1216,32 @@ def event_worker() -> None: new_job_signal.clear() continue - row_id, job = job_row + row_id, job, attempts = job_row + attempts += 1 try: process_event(job) except Exception as exc: log.exception( - "Unhandled error processing message %s: %s", + "Error processing message %s (attempt %s/%s): %s", job.get("message_id"), + attempts, + MAX_ATTEMPTS, exc, ) - finally: - remove_queued_event(row_id) + + if attempts < MAX_ATTEMPTS: + reschedule_event(row_id, attempts) + continue + + log.error( + "Giving up on message %s after %s attempts", + job.get("message_id"), + attempts, + ) + notify_processing_failure(job) + + remove_queued_event(row_id) @app.on_event("startup") diff --git a/containers/talkbot/talkbot.container b/containers/talkbot/talkbot.container index 503997d..3e07d45 100644 --- a/containers/talkbot/talkbot.container +++ b/containers/talkbot/talkbot.container @@ -33,7 +33,7 @@ Environment=NC_PASSWORD=changeme # --- Backend services (adjust host:port to match your setup/network) --- Environment=WHISPER_URL=http://whisper:8080 -Environment=LLAMACPP_URL=http://llm:7000 +Environment=LLAMACPP_URL=http://llamacpp:8090 #Environment=LLAMACPP_MODEL= Environment=LLAMACPP_ENABLE_THINKING=false Environment=QWEN_TTS_URL=http://qwen-tts:8000