containers/talkbot: fix LLAMACPP_URL and retry failed queue jobs

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")

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