From e8f3f2d4d7d98913b1fc02fb9aa614c1cb7786a1 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 15:01:48 +0000 Subject: [PATCH 1/9] Initial plan From 6dc680a23fdb75ed6a43c1e04a1394c73a2086ba Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 15:05:44 +0000 Subject: [PATCH 2/9] Plan: unify RabbitMQ connection handling --- scansynclib/scansynclib.egg-info/SOURCES.txt | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/scansynclib/scansynclib.egg-info/SOURCES.txt b/scansynclib/scansynclib.egg-info/SOURCES.txt index 87bba7e..77d4978 100644 --- a/scansynclib/scansynclib.egg-info/SOURCES.txt +++ b/scansynclib/scansynclib.egg-info/SOURCES.txt @@ -11,6 +11,18 @@ pyproject.toml ./scansynclib/settings.py ./scansynclib/settings_schema.py ./scansynclib/sqlite_wrapper.py +scansynclib/ProcessItem.py +scansynclib/__init__.py +scansynclib/config.py +scansynclib/helpers.py +scansynclib/logging.py +scansynclib/ollama_helper.py +scansynclib/onedrive_api.py +scansynclib/onedrive_smb_manager.py +scansynclib/openai_helper.py +scansynclib/settings.py +scansynclib/settings_schema.py +scansynclib/sqlite_wrapper.py scansynclib.egg-info/PKG-INFO scansynclib.egg-info/SOURCES.txt scansynclib.egg-info/dependency_links.txt From 8b26c143f8852016aedfe8c545f7a348e1581179 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 15:15:10 +0000 Subject: [PATCH 3/9] Unify RabbitMQ handling into scansynclib.rabbitmq with persistent, reconnecting connections --- detection_service/main.py | 24 +- file_naming_service/main.py | 25 +- metadata_service/main.py | 26 +- ocr_service/main.py | 17 +- scansynclib/scansynclib/helpers.py | 90 +----- scansynclib/scansynclib/rabbitmq.py | 343 ++++++++++++++++++++++ scansynclib/scansynclib/sqlite_wrapper.py | 51 ++-- tests/test_rabbitmq.py | 204 +++++++++++++ upload_service/main.py | 17 +- web_service/src/main.py | 40 ++- 10 files changed, 629 insertions(+), 208 deletions(-) create mode 100644 scansynclib/scansynclib/rabbitmq.py create mode 100644 tests/test_rabbitmq.py diff --git a/detection_service/main.py b/detection_service/main.py index abcc5c4..c90b618 100644 --- a/detection_service/main.py +++ b/detection_service/main.py @@ -4,7 +4,7 @@ from collections import defaultdict from scansynclib.logging import logger import pika -from scansynclib.helpers import reconnect_rabbitmq, setup_rabbitmq_connection +from scansynclib.rabbitmq import RabbitMQClient from scansynclib.config import config import json @@ -89,7 +89,9 @@ def publish_new_files(channel, queue_name, grouped_files): def main(): ensure_scan_directory_exists(SCAN_DIR) - connection, channel = setup_rabbitmq_connection(RABBITQUEUE) + client = RabbitMQClient(name="detection") + client.ensure_connection() + client.declare_queue(RABBITQUEUE) logger.info(f"Scanning {SCAN_DIR} for new files...") known_files = get_all_files(SCAN_DIR) @@ -98,11 +100,14 @@ def main(): while True: try: - # Check if the connection is still alive - if connection.is_closed or channel.is_closed: - connection, channel = reconnect_rabbitmq([RABBITQUEUE]) + # Keep the single connection alive and reconnect if it was dropped. + if not client.is_open(): + logger.warning("RabbitMQ connection lost. Reconnecting...") + client.ensure_connection() + client.declare_queue(RABBITQUEUE) else: - connection.process_data_events(1) + # Give pika a chance to send heartbeats between scans. + client.process_events(1) current_files = get_all_files(SCAN_DIR) new_files = current_files - known_files @@ -116,9 +121,10 @@ def main(): if pending_files and last_file_time and (time.time() - last_file_time >= DUPLICATE_DETECTION_WINDOW): logger.info(f"Processing {len(pending_files)} pending files after {DUPLICATE_DETECTION_WINDOW}s wait...") grouped_files = group_files_by_content(pending_files) - publish_new_files(channel, RABBITQUEUE, grouped_files) - pending_files = [] # Pending-Liste leeren - last_file_time = None + if client.is_open(): + publish_new_files(client.channel, RABBITQUEUE, grouped_files) + pending_files = [] # Pending-Liste leeren + last_file_time = None known_files = current_files diff --git a/file_naming_service/main.py b/file_naming_service/main.py index d8295cd..ed2cf58 100644 --- a/file_naming_service/main.py +++ b/file_naming_service/main.py @@ -4,8 +4,7 @@ from scansynclib.ProcessItem import ProcessItem, ProcessStatus, FileNamingStatus from scansynclib.logging import logger -from scansynclib.helpers import connect_rabbitmq, forward_to_rabbitmq -import time +from scansynclib.helpers import consume, forward_to_rabbitmq import pika.exceptions from scansynclib.openai_helper import generate_filename_openai from scansynclib.ollama_helper import generate_filename_ollama @@ -94,10 +93,10 @@ def callback(ch, method, properties, body): finally: try: ch.basic_ack(delivery_tag=method.delivery_tag) - except pika.exceptions.AMQPConnectionError: - logger.error("Connection lost while acknowledging message. Reconnecting...") - connection, channel = connect_rabbitmq([RABBITQUEUE], heartbeat=120) - channel.basic_ack(delivery_tag=method.delivery_tag) + except pika.exceptions.AMQPError: + # The connection was lost before we could acknowledge. The unified + # consumer will reconnect and the broker will redeliver the message. + logger.error("Connection lost while acknowledging message. It will be redelivered after reconnect.") if isinstance(item, ProcessItem): item_file_naming_db_id = getattr(item, "file_naming_db_id", None) if item_file_naming_db_id: @@ -110,19 +109,7 @@ def callback(ch, method, properties, body): def start_consuming_with_reconnect(): - global channel, connection - while True: - try: - connection, channel = connect_rabbitmq([RABBITQUEUE], heartbeat=120) - channel.basic_consume(queue=RABBITQUEUE, on_message_callback=callback) - logger.info("File naming service started, waiting for messages...") - channel.start_consuming() - except pika.exceptions.AMQPConnectionError as e: - logger.error(f"Connection lost: {e}. Reconnecting in 5 seconds...") - time.sleep(5) - except Exception as e: - logger.exception(f"Unexpected error: {e}. Restarting consumer...") - time.sleep(5) + consume(RABBITQUEUE, callback, heartbeat=120) if __name__ == "__main__": diff --git a/metadata_service/main.py b/metadata_service/main.py index b7290f1..f308556 100644 --- a/metadata_service/main.py +++ b/metadata_service/main.py @@ -5,17 +5,14 @@ from scansynclib.ProcessItem import ItemType, ProcessItem, ProcessStatus, OneDriveDestination from PIL import Image from pypdf import PdfReader -import pika -import pika.exceptions from scansynclib.sqlite_wrapper import execute_query, update_scanneddata_database -from scansynclib.helpers import connect_rabbitmq, move_to_failed +from scansynclib.helpers import consume, publish, move_to_failed from scansynclib.config import config import pymupdf import pickle RABBITQUEUE = "metadata_queue" TIMEOUT_PDF_VALIDATION = 300 -channel, connection = None, None def on_created(filepaths: list): @@ -165,12 +162,7 @@ def on_created(filepaths: list): logger.exception(f"Error reading PDF file: {item.local_file_path}") item.status = ProcessStatus.OCR_PENDING update_scanneddata_database(item, {"file_status": item.status.value}) - channel.basic_publish( - exchange="", - routing_key="ocr_queue", - body=pickle.dumps(item), - properties=pika.BasicProperties(delivery_mode=2) - ) + publish("ocr_queue", pickle.dumps(item)) logger.info(f"Added {item.local_file_path} to OCR queue") @@ -236,19 +228,7 @@ def callback(ch, method, properties, body): def start_consuming_with_reconnect(): - global channel, connection - while True: - try: - connection, channel = connect_rabbitmq([RABBITQUEUE, "ocr_queue"], heartbeat=600) - channel.basic_consume(queue=RABBITQUEUE, on_message_callback=callback) - logger.info("Metadata service started, waiting for messages...") - channel.start_consuming() - except pika.exceptions.AMQPConnectionError as e: - logger.error(f"Connection lost: {e}. Reconnecting in 5 seconds...") - time.sleep(5) - except Exception as e: - logger.exception(f"Unexpected error: {e}. Restarting consumer...") - time.sleep(5) + consume([RABBITQUEUE, "ocr_queue"], callback, heartbeat=600) # Start the consumer with reconnect logic diff --git a/ocr_service/main.py b/ocr_service/main.py index 5ca51eb..e533451 100644 --- a/ocr_service/main.py +++ b/ocr_service/main.py @@ -1,13 +1,11 @@ from scansynclib.logging import logger from scansynclib.ProcessItem import ProcessItem, ProcessStatus, OCRStatus from scansynclib.sqlite_wrapper import execute_query, update_scanneddata_database -from scansynclib.helpers import connect_rabbitmq, forward_to_rabbitmq, extract_text +from scansynclib.helpers import consume, forward_to_rabbitmq, extract_text import pickle import ocrmypdf import os from datetime import datetime -import time -import pika.exceptions from scansynclib.settings import settings logger.info("Starting OCR service...") @@ -124,18 +122,7 @@ def start_processing(item: ProcessItem): def start_consuming_with_reconnect(): - while True: - try: - connection, channel = connect_rabbitmq([RABBITQUEUE], heartbeat=600) - channel.basic_consume(queue=RABBITQUEUE, on_message_callback=callback) - logger.info("OCR service started, waiting for messages...") - channel.start_consuming() - except pika.exceptions.AMQPConnectionError as e: - logger.error(f"Connection lost: {e}. Reconnecting in 5 seconds...") - time.sleep(5) - except Exception as e: - logger.exception(f"Unexpected error: {e}. Restarting consumer...") - time.sleep(5) + consume(RABBITQUEUE, callback, heartbeat=600) # Start the consumer with reconnect logic diff --git a/scansynclib/scansynclib/helpers.py b/scansynclib/scansynclib/helpers.py index a12981b..2ad9496 100644 --- a/scansynclib/scansynclib/helpers.py +++ b/scansynclib/scansynclib/helpers.py @@ -1,14 +1,20 @@ from datetime import datetime, timedelta import os -import pickle import re -import pika -import socket -import time from scansynclib.ProcessItem import ProcessItem from scansynclib.config import config -import pika.exceptions from scansynclib.logging import logger +# RabbitMQ handling now lives in the unified scansynclib.rabbitmq module. The +# functions are re-exported here for backwards compatibility with existing +# imports (e.g. ``from scansynclib.helpers import forward_to_rabbitmq``). +from scansynclib.rabbitmq import ( # noqa: F401 + RabbitMQClient, + connect_rabbitmq, + consume, + forward_to_rabbitmq, + publish, + publish_to_exchange, +) from pypdf import PdfReader SMB_TAG_COLORS = [ @@ -32,80 +38,6 @@ ] -def connect_rabbitmq(queue_names: list = None, heartbeat: int = 30): - """ - Establishes a connection to a RabbitMQ server and declares multiple queues. - - This function attempts to connect to a RabbitMQ server up to 10 times, - with a 2-second delay between each attempt. If the connection is - successful, it declares durable queues with the specified names. - - Args: - queue_names (list): A list of RabbitMQ queue names to declare. - heartbeat (int): The heartbeat timeout in seconds for the RabbitMQ connection. - - Returns: - tuple: A tuple containing the RabbitMQ connection and channel objects - if the connection is successful. - None: If the connection could not be established after 10 attempts. - - Raises: - None: The function handles `socket.gaierror` and - `pika.exceptions.AMQPConnectionError` internally. - """ - for i in range(10): - try: - connection = pika.BlockingConnection(pika.ConnectionParameters(host="rabbitmq", heartbeat=heartbeat)) - channel = connection.channel() - if queue_names: - for queue_name in queue_names: - channel.queue_declare(queue=queue_name, durable=True) - channel.basic_qos(prefetch_count=1) - return connection, channel - except (socket.gaierror, pika.exceptions.AMQPConnectionError): - time.sleep(2) - logger.critical("Couldn't connect to RabbitMQ.") - return None - - -def setup_rabbitmq_connection(queue_name): - try: - connection, channel = connect_rabbitmq([queue_name]) - logger.debug(f"Connected to RabbitMQ on {channel.channel_number}") - logger.debug(f"Connected to queue {queue_name}") - return connection, channel - except Exception as e: - logger.critical(f"Failed to connect to RabbitMQ: {e}") - exit(1) - - -def reconnect_rabbitmq(queue_name): - while True: - try: - logger.warning("Attempting to reconnect to RabbitMQ...") - connection, channel = setup_rabbitmq_connection(queue_name) - logger.info("Reconnected to RabbitMQ successfully.") - return connection, channel - except Exception as e: - logger.critical(f"Failed to reconnect to RabbitMQ: {e}") - time.sleep(5) # Wait before retrying - - -def forward_to_rabbitmq(queue_name: str, item: ProcessItem): - try: - connection, channel = connect_rabbitmq([queue_name]) - channel.basic_publish( - exchange='', - routing_key=queue_name, - body=pickle.dumps(item), - properties=pika.BasicProperties(delivery_mode=2) # Make message persistent - ) - logger.info(f"Item {item.filename} forwarded to {queue_name}.") - connection.close() - except Exception as e: - logger.error(f"Failed to forward item {item.filename} to RabbitMQ queue {queue_name}: {e}") - - def parse_timestamp(timestamp: str) -> datetime: try: return datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S") diff --git a/scansynclib/scansynclib/rabbitmq.py b/scansynclib/scansynclib/rabbitmq.py new file mode 100644 index 0000000..0b66392 --- /dev/null +++ b/scansynclib/scansynclib/rabbitmq.py @@ -0,0 +1,343 @@ +"""Unified RabbitMQ connection handling for all ScanSync services. + +This module centralises every RabbitMQ interaction so that services no longer +each build and tear down their own connections. A single, long-lived +connection per process is established lazily and kept alive. Publishing and +consuming both transparently reconnect when the broker drops the connection, +which prevents the "missed heartbeats from client" churn where a new +connection was opened for every message. + +Two building blocks are provided: + +* :class:`RabbitMQClient` – a resilient wrapper around a single pika + ``BlockingConnection``. It reconnects automatically and re-declares queues + and exchanges after a reconnect. +* Module level helpers (:func:`publish`, :func:`forward_to_rabbitmq`, + :func:`consume`, ...) that operate on a shared, process-wide publisher + client so callers can keep the previous simple function based API. +""" + +import pickle +import socket +import threading +import time + +import pika +import pika.exceptions + +from scansynclib.logging import logger + +# Host of the RabbitMQ broker. All services run in the same docker network and +# reach the broker through the ``rabbitmq`` service name. +RABBITMQ_HOST = "rabbitmq" + +# A generous heartbeat keeps the connection alive across the long, blocking +# operations (OCR, uploads, AI file naming) that happen between two AMQP +# interactions. The previous default of 30/60 seconds caused the broker to +# close idle-looking connections while a service was busy. +DEFAULT_HEARTBEAT = 600 + +# Number of connection attempts (and delay between them) before giving up on +# the initial connect. +CONNECTION_ATTEMPTS = 10 +CONNECTION_RETRY_DELAY = 2 + +# Delay before a consumer loop retries after the connection was lost. +RECONNECT_DELAY = 5 + +# Exceptions that indicate the underlying connection/channel is gone and a +# reconnect should be attempted. +_CONNECTION_ERRORS = ( + pika.exceptions.AMQPConnectionError, + pika.exceptions.ConnectionClosed, + pika.exceptions.StreamLostError, + pika.exceptions.ChannelClosed, + pika.exceptions.ChannelWrongStateError, + socket.gaierror, +) + + +class RabbitMQClient: + """A resilient, reusable RabbitMQ connection wrapper. + + The client owns a single connection and channel. Both are created lazily on + first use and automatically recreated whenever the broker drops them, so a + connection is established once and then kept alive for the lifetime of the + process. + """ + + def __init__(self, heartbeat: int = DEFAULT_HEARTBEAT, host: str = RABBITMQ_HOST, name: str = "rabbitmq"): + self._heartbeat = heartbeat + self._host = host + self._name = name + self._connection = None + self._channel = None + self._declared_queues = set() + self._declared_exchanges = set() + # BlockingConnection is not thread safe; serialise access so the client + # can be shared between e.g. Flask request threads. + self._lock = threading.RLock() + + @property + def _parameters(self) -> pika.ConnectionParameters: + return pika.ConnectionParameters( + host=self._host, + heartbeat=self._heartbeat, + blocked_connection_timeout=300, + ) + + def _connect(self) -> bool: + """Establish the connection, retrying a bounded number of times.""" + self._close_quietly() + for attempt in range(1, CONNECTION_ATTEMPTS + 1): + try: + self._connection = pika.BlockingConnection(self._parameters) + self._channel = self._connection.channel() + self._channel.basic_qos(prefetch_count=1) + # Queues/exchanges must be re-declared on the fresh channel. + self._declared_queues.clear() + self._declared_exchanges.clear() + logger.info(f"Connected to RabbitMQ ({self._name}) on channel {self._channel.channel_number}.") + return True + except _CONNECTION_ERRORS as e: + logger.warning(f"RabbitMQ connection attempt {attempt}/{CONNECTION_ATTEMPTS} failed: {e}") + time.sleep(CONNECTION_RETRY_DELAY) + logger.critical("Couldn't connect to RabbitMQ.") + return False + + def _close_quietly(self): + for closable in (self._channel, self._connection): + try: + if closable is not None and closable.is_open: + closable.close() + except Exception: + pass + self._channel = None + self._connection = None + self._declared_queues.clear() + self._declared_exchanges.clear() + + def is_open(self) -> bool: + return ( + self._connection is not None + and self._connection.is_open + and self._channel is not None + and self._channel.is_open + ) + + @property + def channel(self): + return self._channel + + @property + def connection(self): + return self._connection + + def process_events(self, time_limit: float = 1): + """Service heartbeats and dispatch pending events on the connection. + + Useful for publisher-only services that otherwise sit in a polling loop + and would never give pika a chance to send heartbeats. + """ + with self._lock: + if not self.is_open(): + return + try: + self._connection.process_data_events(time_limit) + except _CONNECTION_ERRORS as e: + logger.warning(f"RabbitMQ connection lost while processing events: {e}") + self._close_quietly() + + def ensure_connection(self) -> bool: + """Make sure a usable connection and channel are available.""" + with self._lock: + if self.is_open(): + return True + return self._connect() + + def declare_queue(self, queue_name: str, durable: bool = True): + if not queue_name or queue_name in self._declared_queues: + return + self._channel.queue_declare(queue=queue_name, durable=durable) + self._declared_queues.add(queue_name) + + def declare_exchange(self, exchange: str, exchange_type: str = "fanout"): + if not exchange or exchange in self._declared_exchanges: + return + self._channel.exchange_declare(exchange=exchange, exchange_type=exchange_type) + self._declared_exchanges.add(exchange) + + def publish( + self, + body: bytes, + queue_name: str = "", + exchange: str = "", + routing_key: str = None, + persistent: bool = True, + declare_queue: bool = True, + exchange_type: str = None, + ) -> bool: + """Publish a message, transparently reconnecting on failure. + + Args: + body: The already serialised message body. + queue_name: Target queue for direct (default exchange) publishing. + exchange: Exchange to publish to (defaults to the direct exchange). + routing_key: Routing key; defaults to ``queue_name``. + persistent: Mark the message as persistent (delivery_mode=2). + declare_queue: Declare ``queue_name`` before publishing. + exchange_type: If given, declare ``exchange`` with this type. + + Returns: + ``True`` if the message was published, ``False`` otherwise. + """ + if routing_key is None: + routing_key = queue_name + properties = pika.BasicProperties(delivery_mode=2) if persistent else None + + with self._lock: + # Two attempts: the first may fail on a stale connection, the retry + # runs on a freshly established one. + for attempt in range(2): + try: + if not self.ensure_connection(): + return False + if exchange_type: + self.declare_exchange(exchange, exchange_type) + if declare_queue and queue_name: + self.declare_queue(queue_name) + self._channel.basic_publish( + exchange=exchange, + routing_key=routing_key, + body=body, + properties=properties, + ) + return True + except _CONNECTION_ERRORS as e: + logger.warning(f"RabbitMQ publish failed ({e}); reconnecting (attempt {attempt + 1}/2).") + self._close_quietly() + except Exception: + logger.exception("Unexpected error while publishing to RabbitMQ.") + return False + logger.error("Failed to publish message to RabbitMQ after reconnecting.") + return False + + def consume(self, queue_names, on_message_callback, prefetch_count: int = 1, auto_ack: bool = False): + """Consume messages forever, reconnecting when the connection drops. + + Args: + queue_names: A queue name or list of queue names to declare. The + first queue is the one that is actually consumed. + on_message_callback: The pika message callback. + prefetch_count: QoS prefetch count. + auto_ack: Whether to auto-acknowledge messages. + """ + if isinstance(queue_names, str): + queue_names = [queue_names] + consume_queue = queue_names[0] + + while True: + try: + if not self.ensure_connection(): + time.sleep(RECONNECT_DELAY) + continue + self._channel.basic_qos(prefetch_count=prefetch_count) + for queue_name in queue_names: + self.declare_queue(queue_name) + self._channel.basic_consume( + queue=consume_queue, + on_message_callback=on_message_callback, + auto_ack=auto_ack, + ) + logger.info(f"Consuming from queue '{consume_queue}', waiting for messages...") + self._channel.start_consuming() + except _CONNECTION_ERRORS as e: + logger.error(f"RabbitMQ connection lost: {e}. Reconnecting in {RECONNECT_DELAY} seconds...") + self._close_quietly() + time.sleep(RECONNECT_DELAY) + except Exception as e: + logger.exception(f"Unexpected error in consumer: {e}. Restarting consumer...") + self._close_quietly() + time.sleep(RECONNECT_DELAY) + + def close(self): + with self._lock: + self._close_quietly() + + +# --------------------------------------------------------------------------- +# Shared, process-wide publisher client and function based helpers. +# --------------------------------------------------------------------------- + +_publisher = RabbitMQClient(name="publisher") +_publisher_lock = threading.Lock() + + +def get_publisher() -> RabbitMQClient: + """Return the shared publisher client for the current process.""" + return _publisher + + +def publish(queue_name: str, body: bytes, persistent: bool = True) -> bool: + """Publish raw ``body`` to ``queue_name`` on the shared connection.""" + return _publisher.publish(body, queue_name=queue_name, persistent=persistent) + + +def forward_to_rabbitmq(queue_name: str, item) -> bool: + """Serialise ``item`` and forward it to ``queue_name``. + + Uses the shared, long-lived publisher connection instead of opening and + closing a new connection for every message. + """ + ok = _publisher.publish(pickle.dumps(item), queue_name=queue_name, persistent=True) + if ok: + logger.info(f"Item {getattr(item, 'filename', item)} forwarded to {queue_name}.") + else: + logger.error(f"Failed to forward item {getattr(item, 'filename', item)} to RabbitMQ queue {queue_name}.") + return ok + + +def publish_to_exchange(exchange: str, body: bytes, exchange_type: str = "fanout", persistent: bool = False) -> bool: + """Publish ``body`` to a (declared) exchange on the shared connection.""" + return _publisher.publish( + body, + exchange=exchange, + routing_key="", + persistent=persistent, + declare_queue=False, + exchange_type=exchange_type, + ) + + +def consume(queue_names, on_message_callback, prefetch_count: int = 1, auto_ack: bool = False, heartbeat: int = DEFAULT_HEARTBEAT): + """Start a resilient consumer loop on a dedicated connection. + + A dedicated client (separate from the shared publisher) is used so that + long running message callbacks do not interfere with publishing. + """ + client = RabbitMQClient(heartbeat=heartbeat, name="consumer") + client.consume(queue_names, on_message_callback, prefetch_count=prefetch_count, auto_ack=auto_ack) + + +def connect_rabbitmq(queue_names: list = None, heartbeat: int = DEFAULT_HEARTBEAT): + """Backwards compatible helper returning a ``(connection, channel)`` tuple. + + New code should prefer :class:`RabbitMQClient`, :func:`publish` or + :func:`consume`. This helper is retained for callers that need direct + access to a pika connection/channel (e.g. exclusive fanout queues). + """ + for _ in range(CONNECTION_ATTEMPTS): + try: + connection = pika.BlockingConnection( + pika.ConnectionParameters(host=RABBITMQ_HOST, heartbeat=heartbeat) + ) + channel = connection.channel() + if queue_names: + for queue_name in queue_names: + channel.queue_declare(queue=queue_name, durable=True) + channel.basic_qos(prefetch_count=1) + return connection, channel + except (socket.gaierror, pika.exceptions.AMQPConnectionError): + time.sleep(CONNECTION_RETRY_DELAY) + logger.critical("Couldn't connect to RabbitMQ.") + return None diff --git a/scansynclib/scansynclib/sqlite_wrapper.py b/scansynclib/scansynclib/sqlite_wrapper.py index c05dcf9..e46a43d 100644 --- a/scansynclib/scansynclib/sqlite_wrapper.py +++ b/scansynclib/scansynclib/sqlite_wrapper.py @@ -1,16 +1,14 @@ from contextlib import contextmanager import pickle import sqlite3 -import pika.exceptions from scansynclib.config import config from scansynclib.logging import logger from scansynclib.ProcessItem import ProcessItem, StatusProgressBar +from scansynclib.rabbitmq import publish_to_exchange import os -import pika -# Initialize RabbitMQ connection and channel globally -rabbit_connection = None -rabbit_channel = None +# Exchange used to broadcast live updates to the web service SSE clients. +SSE_EXCHANGE = "sse_updates_fanout" @contextmanager @@ -101,36 +99,21 @@ def update_scanneddata_database(item: ProcessItem, update_values: dict): logger.exception(f"Error updating database for id {id}.") -def initialize_rabbitmq(): - global rabbit_connection, rabbit_channel - try: - rabbit_connection = pika.BlockingConnection(pika.ConnectionParameters('rabbitmq')) - rabbit_channel = rabbit_connection.channel() - # Declare the fanout exchange (idempotent) - rabbit_channel.exchange_declare(exchange='sse_updates_fanout', exchange_type='fanout') - logger.info("RabbitMQ connection initialized successfully.") - except Exception: - logger.exception("Failed to initialize RabbitMQ connection.") +def notify_sse_clients(item: ProcessItem): + """Broadcast an item update to SSE clients via the shared RabbitMQ connection. - -def notify_sse_clients(item: ProcessItem, retry_count=0, max_retries=3): - try: - if rabbit_channel is None or rabbit_connection is None or rabbit_connection.is_closed: - initialize_rabbitmq() - rabbit_channel.basic_publish( - exchange='sse_updates_fanout', - routing_key='', # fanout ignores this - body=pickle.dumps(item), - ) - except pika.exceptions.StreamLostError: - if retry_count < max_retries: - logger.warning(f"RabbitMQ connection lost. Retrying... Attempt {retry_count + 1}/{max_retries}") - initialize_rabbitmq() - notify_sse_clients(item, retry_count=retry_count + 1, max_retries=max_retries) - else: - logger.error("Max retries reached. Failed to send update to SSE queue.") - except Exception: - logger.exception("Error sending update to SSE queue.") + Publishing goes through the unified, long-lived publisher connection which + keeps itself alive and reconnects automatically, so the previous + "missed heartbeats from client" connection churn no longer occurs. + """ + published = publish_to_exchange( + SSE_EXCHANGE, + pickle.dumps(item), + exchange_type="fanout", + persistent=False, + ) + if not published: + logger.error("Failed to send update to SSE queue.") def upgrade_sql_database(): diff --git a/tests/test_rabbitmq.py b/tests/test_rabbitmq.py new file mode 100644 index 0000000..bb95091 --- /dev/null +++ b/tests/test_rabbitmq.py @@ -0,0 +1,204 @@ +import pickle + +import pika.exceptions +import pytest + +from scansynclib import rabbitmq +from scansynclib.rabbitmq import RabbitMQClient + + +class _ForwardItem: + filename = "doc.pdf" + + +class FakeChannel: + """Minimal stand-in for a pika channel used in the tests.""" + + def __init__(self): + self.is_open = True + self.channel_number = 1 + self.queue_declare_calls = [] + self.exchange_declare_calls = [] + self.basic_publish_calls = [] + self.basic_qos_calls = [] + self.basic_consume_calls = [] + self.publish_side_effect = None + self.start_consuming_side_effect = None + + def basic_qos(self, prefetch_count=1): + self.basic_qos_calls.append(prefetch_count) + + def queue_declare(self, queue, durable=True): + self.queue_declare_calls.append(queue) + + def exchange_declare(self, exchange, exchange_type="fanout"): + self.exchange_declare_calls.append((exchange, exchange_type)) + + def basic_publish(self, exchange="", routing_key="", body=None, properties=None): + if self.publish_side_effect is not None: + effect = self.publish_side_effect + self.publish_side_effect = None + raise effect + self.basic_publish_calls.append((exchange, routing_key, body, properties)) + + def basic_consume(self, queue, on_message_callback, auto_ack=False): + self.basic_consume_calls.append((queue, on_message_callback, auto_ack)) + + def start_consuming(self): + if self.start_consuming_side_effect is not None: + effect = self.start_consuming_side_effect.pop(0) + raise effect + + +class FakeConnection: + def __init__(self, channel): + self.is_open = True + self._channel = channel + self.process_data_events_calls = [] + + def channel(self): + return self._channel + + def process_data_events(self, time_limit=1): + self.process_data_events_calls.append(time_limit) + + def close(self): + self.is_open = False + self._channel.is_open = False + + +@pytest.fixture +def fake_broker(mocker): + """Patch pika.BlockingConnection to hand out fresh fake connections.""" + channels = [] + connections = [] + + def factory(*args, **kwargs): + channel = FakeChannel() + connection = FakeConnection(channel) + channels.append(channel) + connections.append(connection) + return connection + + blocking = mocker.patch("scansynclib.rabbitmq.pika.BlockingConnection", side_effect=factory) + mocker.patch("scansynclib.rabbitmq.time.sleep") + return {"blocking": blocking, "channels": channels, "connections": connections} + + +def test_publish_establishes_connection_once_and_reuses(fake_broker): + client = RabbitMQClient(name="test") + + assert client.publish(b"one", queue_name="q") is True + assert client.publish(b"two", queue_name="q") is True + + # Connection is established a single time and then kept alive. + assert fake_broker["blocking"].call_count == 1 + channel = fake_broker["channels"][0] + assert len(channel.basic_publish_calls) == 2 + # The queue is declared only once thanks to the declared-queue cache. + assert channel.queue_declare_calls == ["q"] + + +def test_publish_marks_messages_persistent(fake_broker): + client = RabbitMQClient(name="test") + client.publish(b"payload", queue_name="q", persistent=True) + + channel = fake_broker["channels"][0] + _, routing_key, body, properties = channel.basic_publish_calls[0] + assert routing_key == "q" + assert body == b"payload" + assert properties.delivery_mode == 2 + + +def test_publish_reconnects_on_stream_lost(fake_broker): + client = RabbitMQClient(name="test") + assert client.publish(b"first", queue_name="q") is True + + # Make the next publish fail as if the broker dropped the connection. + fake_broker["channels"][0].publish_side_effect = pika.exceptions.StreamLostError("boom") + + assert client.publish(b"second", queue_name="q") is True + # A new connection was established to recover. + assert fake_broker["blocking"].call_count == 2 + # The message was delivered on the freshly created channel. + assert len(fake_broker["channels"][1].basic_publish_calls) == 1 + + +def test_publish_returns_false_when_broker_unavailable(mocker): + mocker.patch("scansynclib.rabbitmq.time.sleep") + mocker.patch( + "scansynclib.rabbitmq.pika.BlockingConnection", + side_effect=pika.exceptions.AMQPConnectionError("down"), + ) + client = RabbitMQClient(name="test") + assert client.publish(b"data", queue_name="q") is False + + +def test_publish_to_exchange_declares_exchange(fake_broker): + assert rabbitmq.publish_to_exchange("sse", b"body", exchange_type="fanout") is True + channel = fake_broker["channels"][0] + assert ("sse", "fanout") in channel.exchange_declare_calls + published_exchange = channel.basic_publish_calls[0][0] + assert published_exchange == "sse" + + +def test_forward_to_rabbitmq_pickles_item(mocker): + publish = mocker.patch.object(rabbitmq._publisher, "publish", return_value=True) + + item = _ForwardItem() + assert rabbitmq.forward_to_rabbitmq("upload_queue", item) is True + publish.assert_called_once() + _, kwargs = publish.call_args + body = publish.call_args.args[0] + assert pickle.loads(body).filename == "doc.pdf" + assert kwargs["queue_name"] == "upload_queue" + + +def test_consume_reconnects_on_connection_error(fake_broker): + client = RabbitMQClient(name="test") + + def on_message(ch, method, properties, body): + pass + + # First start_consuming raises a connection error (triggering reconnect), + # the second raises KeyboardInterrupt to break out of the infinite loop. + def prime_channel(*args, **kwargs): + channel = FakeChannel() + connection = FakeConnection(channel) + if len(fake_broker["channels"]) == 0: + channel.start_consuming_side_effect = [pika.exceptions.AMQPConnectionError("lost")] + else: + channel.start_consuming_side_effect = [KeyboardInterrupt()] + fake_broker["channels"].append(channel) + return connection + + fake_broker["blocking"].side_effect = prime_channel + + with pytest.raises(KeyboardInterrupt): + client.consume("q", on_message) + + # The consumer reconnected after the first failure. + assert fake_broker["blocking"].call_count == 2 + + +def test_connect_rabbitmq_returns_connection_and_channel(fake_broker): + result = rabbitmq.connect_rabbitmq(["a", "b"]) + assert result is not None + connection, channel = result + assert channel.queue_declare_calls == ["a", "b"] + + +def test_connect_rabbitmq_returns_none_on_failure(mocker): + mocker.patch("scansynclib.rabbitmq.time.sleep") + mocker.patch( + "scansynclib.rabbitmq.pika.BlockingConnection", + side_effect=pika.exceptions.AMQPConnectionError("down"), + ) + assert rabbitmq.connect_rabbitmq(["a"]) is None + + +def test_process_events_services_heartbeats(fake_broker): + client = RabbitMQClient(name="test") + client.ensure_connection() + client.process_events(1) + assert fake_broker["connections"][0].process_data_events_calls == [1] diff --git a/upload_service/main.py b/upload_service/main.py index ebdd255..9c701dd 100644 --- a/upload_service/main.py +++ b/upload_service/main.py @@ -2,13 +2,11 @@ import pickle from scansynclib.ProcessItem import ProcessItem, ProcessStatus from scansynclib.logging import logger -from scansynclib.helpers import connect_rabbitmq, move_to_failed +from scansynclib.helpers import consume, move_to_failed from scansynclib.sqlite_wrapper import update_scanneddata_database from scansynclib.onedrive_api import upload_small from scansynclib.config import config import os -import time -import pika.exceptions logger.info("Starting Upload service...") RABBITQUEUE = "upload_queue" @@ -99,18 +97,7 @@ def start_processing(item: ProcessItem): def start_consuming_with_reconnect(): - while True: - try: - connection, channel = connect_rabbitmq([RABBITQUEUE], heartbeat=600) - channel.basic_consume(queue=RABBITQUEUE, on_message_callback=callback) - logger.info("Upload service started, waiting for messages...") - channel.start_consuming() - except pika.exceptions.AMQPConnectionError as e: - logger.error(f"Connection lost: {e}. Reconnecting in 5 seconds...") - time.sleep(5) - except Exception as e: - logger.exception(f"Unexpected error: {e}. Restarting consumer...") - time.sleep(5) + consume(RABBITQUEUE, callback, heartbeat=600) # Start the consumer with reconnect logic diff --git a/web_service/src/main.py b/web_service/src/main.py index e471e39..3a3a8f8 100644 --- a/web_service/src/main.py +++ b/web_service/src/main.py @@ -6,6 +6,7 @@ import time from flask import Flask, Response, request, send_from_directory import sys +import pika.exceptions sys.path.append('/app/src') from scansynclib.ProcessItem import ProcessItem, StatusProgressBar from scansynclib.helpers import connect_rabbitmq, format_time_difference @@ -43,19 +44,7 @@ def start_rabbitmq_listener(): def rabbitmq_listener(): logger.info("Started RabbitMQ listener thread.") - result = connect_rabbitmq() - if result is None: - logger.warning("RabbitMQ is not available. SSE updates will be disabled.") - return - connection, channel = result - - # Use fanout as exchange type to broadcast messages to all connected clients exchange_name = "sse_updates_fanout" - channel.exchange_declare(exchange=exchange_name, exchange_type='fanout') - result = channel.queue_declare(queue='', exclusive=True) - queue_name = result.method.queue - - channel.queue_bind(exchange=exchange_name, queue=queue_name) def callback(ch, method, properties, body): if connected_clients > 0: @@ -103,8 +92,31 @@ def callback(ch, method, properties, body): else: logger.debug("No connected clients. Skipping SSE queue update.") - channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True) - channel.start_consuming() + # Keep the SSE listener alive: reconnect automatically if the broker drops + # the connection instead of silently ending the thread. + while True: + result = connect_rabbitmq() + if result is None: + logger.warning("RabbitMQ is not available. Retrying SSE listener in 5 seconds...") + time.sleep(5) + continue + connection, channel = result + + try: + # Use fanout as exchange type to broadcast messages to all connected clients + channel.exchange_declare(exchange=exchange_name, exchange_type='fanout') + queue_result = channel.queue_declare(queue='', exclusive=True) + queue_name = queue_result.method.queue + channel.queue_bind(exchange=exchange_name, queue=queue_name) + + channel.basic_consume(queue=queue_name, on_message_callback=callback, auto_ack=True) + channel.start_consuming() + except pika.exceptions.AMQPError as e: + logger.error(f"RabbitMQ SSE listener connection lost: {e}. Reconnecting in 5 seconds...") + time.sleep(5) + except Exception as e: + logger.exception(f"Unexpected error in SSE listener: {e}. Reconnecting in 5 seconds...") + time.sleep(5) def get_dashboard_info() -> dict: From 9b4015d269a5894b39fbd6a2dc5654337e1d7f32 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 16:20:18 +0000 Subject: [PATCH 4/9] Address PR review reliability fixes --- detection_service/main.py | 9 +++-- file_naming_service/main.py | 4 +++ scansynclib/scansynclib.egg-info/SOURCES.txt | 12 ------- scansynclib/scansynclib/rabbitmq.py | 36 +++++++++++++------- web_service/src/main.py | 6 ++++ 5 files changed, 40 insertions(+), 27 deletions(-) diff --git a/detection_service/main.py b/detection_service/main.py index c90b618..eb417e5 100644 --- a/detection_service/main.py +++ b/detection_service/main.py @@ -90,7 +90,9 @@ def publish_new_files(channel, queue_name, grouped_files): def main(): ensure_scan_directory_exists(SCAN_DIR) client = RabbitMQClient(name="detection") - client.ensure_connection() + while not client.ensure_connection(): + logger.warning("RabbitMQ is not available. Retrying detection startup in 5 seconds...") + time.sleep(5) client.declare_queue(RABBITQUEUE) logger.info(f"Scanning {SCAN_DIR} for new files...") @@ -103,7 +105,10 @@ def main(): # Keep the single connection alive and reconnect if it was dropped. if not client.is_open(): logger.warning("RabbitMQ connection lost. Reconnecting...") - client.ensure_connection() + if not client.ensure_connection(): + logger.warning("RabbitMQ reconnect failed. Retrying in 5 seconds...") + time.sleep(5) + continue client.declare_queue(RABBITQUEUE) else: # Give pika a chance to send heartbeats between scans. diff --git a/file_naming_service/main.py b/file_naming_service/main.py index ed2cf58..0f9e813 100644 --- a/file_naming_service/main.py +++ b/file_naming_service/main.py @@ -91,12 +91,16 @@ def callback(ch, method, properties, body): logger.exception(f"Failed processing {item.filename}.") execute_query("UPDATE file_naming_jobs SET file_naming_status = ?, error_description = ?, finished = DATETIME('now', 'localtime') WHERE id = ?", (FileNamingStatus.FAILED.name, str(e), item.file_naming_db_id)) finally: + acknowledged = False try: ch.basic_ack(delivery_tag=method.delivery_tag) + acknowledged = True except pika.exceptions.AMQPError: # The connection was lost before we could acknowledge. The unified # consumer will reconnect and the broker will redeliver the message. logger.error("Connection lost while acknowledging message. It will be redelivered after reconnect.") + if not acknowledged: + return if isinstance(item, ProcessItem): item_file_naming_db_id = getattr(item, "file_naming_db_id", None) if item_file_naming_db_id: diff --git a/scansynclib/scansynclib.egg-info/SOURCES.txt b/scansynclib/scansynclib.egg-info/SOURCES.txt index 77d4978..a387531 100644 --- a/scansynclib/scansynclib.egg-info/SOURCES.txt +++ b/scansynclib/scansynclib.egg-info/SOURCES.txt @@ -1,16 +1,4 @@ pyproject.toml -./scansynclib/ProcessItem.py -./scansynclib/__init__.py -./scansynclib/config.py -./scansynclib/helpers.py -./scansynclib/logging.py -./scansynclib/ollama_helper.py -./scansynclib/onedrive_api.py -./scansynclib/onedrive_smb_manager.py -./scansynclib/openai_helper.py -./scansynclib/settings.py -./scansynclib/settings_schema.py -./scansynclib/sqlite_wrapper.py scansynclib/ProcessItem.py scansynclib/__init__.py scansynclib/config.py diff --git a/scansynclib/scansynclib/rabbitmq.py b/scansynclib/scansynclib/rabbitmq.py index 0b66392..aecb4e3 100644 --- a/scansynclib/scansynclib/rabbitmq.py +++ b/scansynclib/scansynclib/rabbitmq.py @@ -155,17 +155,25 @@ def ensure_connection(self) -> bool: return True return self._connect() - def declare_queue(self, queue_name: str, durable: bool = True): - if not queue_name or queue_name in self._declared_queues: - return - self._channel.queue_declare(queue=queue_name, durable=durable) - self._declared_queues.add(queue_name) - - def declare_exchange(self, exchange: str, exchange_type: str = "fanout"): - if not exchange or exchange in self._declared_exchanges: - return - self._channel.exchange_declare(exchange=exchange, exchange_type=exchange_type) - self._declared_exchanges.add(exchange) + def declare_queue(self, queue_name: str, durable: bool = True) -> bool: + with self._lock: + if not queue_name or queue_name in self._declared_queues: + return True + if not self.ensure_connection(): + return False + self._channel.queue_declare(queue=queue_name, durable=durable) + self._declared_queues.add(queue_name) + return True + + def declare_exchange(self, exchange: str, exchange_type: str = "fanout") -> bool: + with self._lock: + if not exchange or exchange in self._declared_exchanges: + return True + if not self.ensure_connection(): + return False + self._channel.exchange_declare(exchange=exchange, exchange_type=exchange_type) + self._declared_exchanges.add(exchange) + return True def publish( self, @@ -203,9 +211,11 @@ def publish( if not self.ensure_connection(): return False if exchange_type: - self.declare_exchange(exchange, exchange_type) + if not self.declare_exchange(exchange, exchange_type): + return False if declare_queue and queue_name: - self.declare_queue(queue_name) + if not self.declare_queue(queue_name): + return False self._channel.basic_publish( exchange=exchange, routing_key=routing_key, diff --git a/web_service/src/main.py b/web_service/src/main.py index 3a3a8f8..974961b 100644 --- a/web_service/src/main.py +++ b/web_service/src/main.py @@ -117,6 +117,12 @@ def callback(ch, method, properties, body): except Exception as e: logger.exception(f"Unexpected error in SSE listener: {e}. Reconnecting in 5 seconds...") time.sleep(5) + finally: + try: + if connection.is_open: + connection.close() + except Exception: + logger.debug("Failed closing RabbitMQ SSE listener connection during reconnect.") def get_dashboard_info() -> dict: From 9ffdb3512d7a16d40e1acdd1b5f1d221269d8703 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:02:53 +0000 Subject: [PATCH 5/9] Fix review feedback for RabbitMQ follow-up --- file_naming_service/main.py | 4 -- scansynclib/scansynclib.egg-info/SOURCES.txt | 1 + scansynclib/scansynclib/rabbitmq.py | 1 - tests/test_file_naming_status.py | 73 +++++++++++++++++++- 4 files changed, 73 insertions(+), 6 deletions(-) diff --git a/file_naming_service/main.py b/file_naming_service/main.py index 0f9e813..ed2cf58 100644 --- a/file_naming_service/main.py +++ b/file_naming_service/main.py @@ -91,16 +91,12 @@ def callback(ch, method, properties, body): logger.exception(f"Failed processing {item.filename}.") execute_query("UPDATE file_naming_jobs SET file_naming_status = ?, error_description = ?, finished = DATETIME('now', 'localtime') WHERE id = ?", (FileNamingStatus.FAILED.name, str(e), item.file_naming_db_id)) finally: - acknowledged = False try: ch.basic_ack(delivery_tag=method.delivery_tag) - acknowledged = True except pika.exceptions.AMQPError: # The connection was lost before we could acknowledge. The unified # consumer will reconnect and the broker will redeliver the message. logger.error("Connection lost while acknowledging message. It will be redelivered after reconnect.") - if not acknowledged: - return if isinstance(item, ProcessItem): item_file_naming_db_id = getattr(item, "file_naming_db_id", None) if item_file_naming_db_id: diff --git a/scansynclib/scansynclib.egg-info/SOURCES.txt b/scansynclib/scansynclib.egg-info/SOURCES.txt index a387531..dea3bd1 100644 --- a/scansynclib/scansynclib.egg-info/SOURCES.txt +++ b/scansynclib/scansynclib.egg-info/SOURCES.txt @@ -8,6 +8,7 @@ scansynclib/ollama_helper.py scansynclib/onedrive_api.py scansynclib/onedrive_smb_manager.py scansynclib/openai_helper.py +scansynclib/rabbitmq.py scansynclib/settings.py scansynclib/settings_schema.py scansynclib/sqlite_wrapper.py diff --git a/scansynclib/scansynclib/rabbitmq.py b/scansynclib/scansynclib/rabbitmq.py index aecb4e3..79cc917 100644 --- a/scansynclib/scansynclib/rabbitmq.py +++ b/scansynclib/scansynclib/rabbitmq.py @@ -280,7 +280,6 @@ def close(self): # --------------------------------------------------------------------------- _publisher = RabbitMQClient(name="publisher") -_publisher_lock = threading.Lock() def get_publisher() -> RabbitMQClient: diff --git a/tests/test_file_naming_status.py b/tests/test_file_naming_status.py index f177116..19e43d2 100644 --- a/tests/test_file_naming_status.py +++ b/tests/test_file_naming_status.py @@ -1,8 +1,12 @@ import pickle import sys import types +from types import SimpleNamespace import pytest +from pika import exceptions as pika_exceptions + +from scansynclib.settings_schema import FileNamingMethod # Importing the service pulls in scansynclib.sqlite_wrapper, which initializes a @@ -16,23 +20,61 @@ _original_sqlite_wrapper = sys.modules.get("scansynclib.sqlite_wrapper") sys.modules["scansynclib.sqlite_wrapper"] = _sqlite_stub +_openai_stub = types.ModuleType("scansynclib.openai_helper") +_openai_stub.generate_filename_openai = lambda item: item.filename_without_extension +_original_openai_helper = sys.modules.get("scansynclib.openai_helper") +sys.modules["scansynclib.openai_helper"] = _openai_stub + +_ollama_stub = types.ModuleType("scansynclib.ollama_helper") +_ollama_stub.generate_filename_ollama = lambda item: item.filename_without_extension +_original_ollama_helper = sys.modules.get("scansynclib.ollama_helper") +sys.modules["scansynclib.ollama_helper"] = _ollama_stub + +_settings_stub = types.ModuleType("scansynclib.settings") +_settings_stub.settings = SimpleNamespace( + file_naming=SimpleNamespace( + method=FileNamingMethod.NONE, + openai_api_key="", + ollama_server_url="", + ollama_server_port=11434, + ollama_model="", + ) +) +_original_settings = sys.modules.get("scansynclib.settings") +sys.modules["scansynclib.settings"] = _settings_stub + def teardown_module(module): if _original_sqlite_wrapper is None: sys.modules.pop("scansynclib.sqlite_wrapper", None) else: sys.modules["scansynclib.sqlite_wrapper"] = _original_sqlite_wrapper + if _original_openai_helper is None: + sys.modules.pop("scansynclib.openai_helper", None) + else: + sys.modules["scansynclib.openai_helper"] = _original_openai_helper + if _original_ollama_helper is None: + sys.modules.pop("scansynclib.ollama_helper", None) + else: + sys.modules["scansynclib.ollama_helper"] = _original_ollama_helper + if _original_settings is None: + sys.modules.pop("scansynclib.settings", None) + else: + sys.modules["scansynclib.settings"] = _original_settings import file_naming_service.main as fn_main # noqa: E402 -from scansynclib.ProcessItem import ProcessItem, ItemType, FileNamingStatus # noqa: E402 +from scansynclib.ProcessItem import ProcessItem, ItemType, FileNamingStatus, ProcessStatus # noqa: E402 @pytest.fixture def item(tmp_path): file_path = tmp_path / "doc.pdf" file_path.write_bytes(b"%PDF-1.4 test") + ocr_file_path = tmp_path / "doc_OCR.pdf" + ocr_file_path.write_bytes(b"%PDF-1.4 ocr") process_item = ProcessItem(str(file_path), ItemType.PDF) + process_item.ocr_file = str(ocr_file_path) process_item.db_id = 7 process_item.file_naming_db_id = 11 process_item.file_naming_status = FileNamingStatus.PROCESSING @@ -82,3 +124,32 @@ def test_callback_with_non_processitem_does_not_crash(mocker): ch.basic_ack.assert_called_once_with(delivery_tag=123) execute_query.assert_not_called() forward.assert_not_called() + + +def test_callback_continues_pipeline_when_ack_fails(item, mocker): + mocker.patch.object(fn_main.settings.file_naming, "method", FileNamingMethod.OPENAI) + mocker.patch.object(fn_main.settings.file_naming, "openai_api_key", "test-key") + mocker.patch.object(fn_main, "generate_filename_openai", return_value="renamed") + execute_query = mocker.patch.object(fn_main, "execute_query", return_value=item.file_naming_db_id) + mocker.patch.object(fn_main, "get_latest_file_naming_status", return_value=FileNamingStatus.COMPLETED) + update = mocker.patch.object(fn_main, "update_scanneddata_database") + forward = mocker.patch.object(fn_main, "forward_to_rabbitmq") + + ch = mocker.Mock() + ch.basic_ack.side_effect = pika_exceptions.AMQPError("lost connection") + method = mocker.Mock() + method.delivery_tag = 456 + + fn_main.callback(ch, method, None, pickle.dumps(item)) + + ch.basic_ack.assert_called_once_with(delivery_tag=456) + execute_query.assert_called() + update.assert_called_once() + updated_item = update.call_args.args[0] + assert update.call_args.args[1] == {"file_status": ProcessStatus.SYNC_PENDING.value} + assert updated_item.filename == "renamed.pdf" + assert updated_item.status == ProcessStatus.SYNC_PENDING + forward.assert_called_once() + forwarded_item = forward.call_args.args[1] + assert forwarded_item.filename == "renamed.pdf" + assert forwarded_item.ocr_file.endswith("renamed_OCR.pdf") From b6bd72ce35f552e0fd9d57a5458c8bc723c41bc5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:20:11 +0000 Subject: [PATCH 6/9] fix: skip DB update and upload_queue forward when basic_ack fails --- file_naming_service/main.py | 21 ++++++++++++--------- 1 file changed, 12 insertions(+), 9 deletions(-) diff --git a/file_naming_service/main.py b/file_naming_service/main.py index ed2cf58..33c4ea7 100644 --- a/file_naming_service/main.py +++ b/file_naming_service/main.py @@ -91,21 +91,24 @@ def callback(ch, method, properties, body): logger.exception(f"Failed processing {item.filename}.") execute_query("UPDATE file_naming_jobs SET file_naming_status = ?, error_description = ?, finished = DATETIME('now', 'localtime') WHERE id = ?", (FileNamingStatus.FAILED.name, str(e), item.file_naming_db_id)) finally: + ack_ok = True try: ch.basic_ack(delivery_tag=method.delivery_tag) except pika.exceptions.AMQPError: + ack_ok = False # The connection was lost before we could acknowledge. The unified # consumer will reconnect and the broker will redeliver the message. logger.error("Connection lost while acknowledging message. It will be redelivered after reconnect.") - if isinstance(item, ProcessItem): - item_file_naming_db_id = getattr(item, "file_naming_db_id", None) - if item_file_naming_db_id: - item.file_naming_status = get_latest_file_naming_status(item) - item.status = ProcessStatus.SYNC_PENDING - update_scanneddata_database(item, {"file_status": item.status.value}) - forward_to_rabbitmq("upload_queue", item) - else: - logger.error("Item is None, cannot forward to upload queue.") + if ack_ok: + if isinstance(item, ProcessItem): + item_file_naming_db_id = getattr(item, "file_naming_db_id", None) + if item_file_naming_db_id: + item.file_naming_status = get_latest_file_naming_status(item) + item.status = ProcessStatus.SYNC_PENDING + update_scanneddata_database(item, {"file_status": item.status.value}) + forward_to_rabbitmq("upload_queue", item) + else: + logger.error("Item is None, cannot forward to upload queue.") def start_consuming_with_reconnect(): From 5adcbacf6f1bcd3904eb0c90e891d5f6797cf47a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Thu, 2 Jul 2026 17:22:23 +0000 Subject: [PATCH 7/9] test: update test to verify followup actions are skipped when basic_ack fails --- tests/test_file_naming_status.py | 16 ++++------------ 1 file changed, 4 insertions(+), 12 deletions(-) diff --git a/tests/test_file_naming_status.py b/tests/test_file_naming_status.py index 19e43d2..ee87e26 100644 --- a/tests/test_file_naming_status.py +++ b/tests/test_file_naming_status.py @@ -126,11 +126,11 @@ def test_callback_with_non_processitem_does_not_crash(mocker): forward.assert_not_called() -def test_callback_continues_pipeline_when_ack_fails(item, mocker): +def test_callback_skips_followup_when_ack_fails(item, mocker): mocker.patch.object(fn_main.settings.file_naming, "method", FileNamingMethod.OPENAI) mocker.patch.object(fn_main.settings.file_naming, "openai_api_key", "test-key") mocker.patch.object(fn_main, "generate_filename_openai", return_value="renamed") - execute_query = mocker.patch.object(fn_main, "execute_query", return_value=item.file_naming_db_id) + mocker.patch.object(fn_main, "execute_query", return_value=item.file_naming_db_id) mocker.patch.object(fn_main, "get_latest_file_naming_status", return_value=FileNamingStatus.COMPLETED) update = mocker.patch.object(fn_main, "update_scanneddata_database") forward = mocker.patch.object(fn_main, "forward_to_rabbitmq") @@ -143,13 +143,5 @@ def test_callback_continues_pipeline_when_ack_fails(item, mocker): fn_main.callback(ch, method, None, pickle.dumps(item)) ch.basic_ack.assert_called_once_with(delivery_tag=456) - execute_query.assert_called() - update.assert_called_once() - updated_item = update.call_args.args[0] - assert update.call_args.args[1] == {"file_status": ProcessStatus.SYNC_PENDING.value} - assert updated_item.filename == "renamed.pdf" - assert updated_item.status == ProcessStatus.SYNC_PENDING - forward.assert_called_once() - forwarded_item = forward.call_args.args[1] - assert forwarded_item.filename == "renamed.pdf" - assert forwarded_item.ocr_file.endswith("renamed_OCR.pdf") + update.assert_not_called() + forward.assert_not_called() From b3b4724f7538a95b8830c7a889b202fb719f9f8f Mon Sep 17 00:00:00 2001 From: Maximilian Krause Date: Sat, 4 Jul 2026 14:16:50 +0200 Subject: [PATCH 8/9] update rabbitmq --- docker-compose.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docker-compose.yml b/docker-compose.yml index 1daea7d..f5ea814 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -138,7 +138,7 @@ services: - app-network rabbitmq: - image: "rabbitmq:4.0.7-management" + image: "rabbitmq:4-management" restart: unless-stopped container_name: "rabbitmq" environment: From e2f01e1a4094c06bf349e225b0925302844cfbb5 Mon Sep 17 00:00:00 2001 From: Maximilian Krause Date: Sat, 4 Jul 2026 14:30:13 +0200 Subject: [PATCH 9/9] fix pytests --- tests/test_file_naming_status.py | 42 ++++++++++++++++++-------------- tests/test_settings.py | 22 +++++++++++++++++ 2 files changed, 46 insertions(+), 18 deletions(-) diff --git a/tests/test_file_naming_status.py b/tests/test_file_naming_status.py index ee87e26..f3b8220 100644 --- a/tests/test_file_naming_status.py +++ b/tests/test_file_naming_status.py @@ -44,26 +44,32 @@ sys.modules["scansynclib.settings"] = _settings_stub -def teardown_module(module): - if _original_sqlite_wrapper is None: - sys.modules.pop("scansynclib.sqlite_wrapper", None) - else: - sys.modules["scansynclib.sqlite_wrapper"] = _original_sqlite_wrapper - if _original_openai_helper is None: - sys.modules.pop("scansynclib.openai_helper", None) - else: - sys.modules["scansynclib.openai_helper"] = _original_openai_helper - if _original_ollama_helper is None: - sys.modules.pop("scansynclib.ollama_helper", None) - else: - sys.modules["scansynclib.ollama_helper"] = _original_ollama_helper - if _original_settings is None: - sys.modules.pop("scansynclib.settings", None) - else: - sys.modules["scansynclib.settings"] = _original_settings +import file_naming_service.main as fn_main # noqa: E402 -import file_naming_service.main as fn_main # noqa: E402 +def _restore_scansynclib_modules(): + """Undo the sys.modules stubbing done above. + + ``file_naming_service.main`` has already bound the stubbed objects it needs, + so the stubs can be removed from ``sys.modules`` immediately. Restoring them + here (rather than in ``teardown_module``) prevents the fileless stub modules + from leaking into the collection/import of other test modules, which would + otherwise fail to import the real ``scansynclib.settings`` members. + """ + for name, original in ( + ("scansynclib.sqlite_wrapper", _original_sqlite_wrapper), + ("scansynclib.openai_helper", _original_openai_helper), + ("scansynclib.ollama_helper", _original_ollama_helper), + ("scansynclib.settings", _original_settings), + ): + if original is None: + sys.modules.pop(name, None) + else: + sys.modules[name] = original + + +_restore_scansynclib_modules() + from scansynclib.ProcessItem import ProcessItem, ItemType, FileNamingStatus, ProcessStatus # noqa: E402 diff --git a/tests/test_settings.py b/tests/test_settings.py index ce636c6..bd4d933 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -1,5 +1,27 @@ +from unittest.mock import MagicMock from pydantic import ValidationError import pytest + +# settings.py connects to Redis at module level. Mock it so the module can be +# imported in a unit-test environment without a live Redis instance. +import redis as _real_redis +_orig_from_url = _real_redis.Redis.from_url + + +def _mock_from_url(*args, **kwargs): + mock_client = MagicMock() + mock_client.get.return_value = None # No existing settings in Redis + mock_client.set.return_value = True + mock_client.publish.return_value = 0 + mock_pubsub = MagicMock() + mock_pubsub.subscribe.return_value = None + mock_pubsub.listen.return_value = iter([]) # Empty iterator + mock_client.pubsub.return_value = mock_pubsub + return mock_client + + +_real_redis.Redis.from_url = _mock_from_url + from scansynclib.settings import SettingsProxy from scansynclib.settings_schema import FileNamingSettings, FileNamingMethod, SettingsSchema