Unify RabbitMQ handling with persistent, auto-reconnecting connections by Copilot · Pull Request #70 · maxi07/ScanSync

Copilot · 2026-07-02T15:01:50Z

Services each built their own RabbitMQ connections and rebuilt them constantly, so the broker dropped clients with missed heartbeats from client, timeout: 60s. The goal: establish connections once, keep them alive, and centralize the logic.

Root cause

sqlite_wrapper opened the SSE publisher with the default 60s heartbeat and never serviced it during long OCR/upload/naming work → broker closed it every ~60s.
forward_to_rabbitmq opened and closed a new connection per message.
Every service duplicated its own connect/reconnect loop.

Changes

New scansynclib/rabbitmq.py — single source of truth for messaging:
- RabbitMQClient: wraps one long-lived BlockingConnection, lazily connects, re-declares queues/exchanges after a reconnect, and transparently reconnects on publish/consume failures (default heartbeat 600s).
- Helpers: publish, forward_to_rabbitmq (shared publisher connection), publish_to_exchange, consume (unified reconnecting consumer loop), process_events (heartbeat servicing for polling publishers), and a backward-compatible connect_rabbitmq.
helpers.py re-exports these so existing imports keep working.
sqlite_wrapper.py — SSE notify_sse_clients now publishes through the self-healing publisher (fixes the 60s drops).
Services refactored — OCR / file_naming / upload / metadata consume via consume(...); metadata publishes to ocr_queue via publish(...); detection uses RabbitMQClient + process_events; the web SSE listener reconnects instead of killing its thread. file_naming's fragile manual re-ack was dropped (broker redelivers after reconnect).
tests/test_rabbitmq.py — covers connection reuse, reconnect on StreamLostError, publish-failure handling, exchange declaration, pickling in forward_to_rabbitmq, and consumer reconnection.

Before → after

# Before: new connection opened and torn down for every message
def forward_to_rabbitmq(queue_name, item):
    connection, channel = connect_rabbitmq([queue_name])
    channel.basic_publish(...)
    connection.close()

# After: shared, long-lived connection that reconnects on demand
forward_to_rabbitmq("upload_queue", item)   # -> _publisher.publish(...)

Publishers and consumers use separate connections (recommended by RabbitMQ), and all reconnect automatically, eliminating the connection churn.

…connecting connections

Copilot

Pull request overview

This PR centralizes RabbitMQ connectivity into scansynclib to eliminate per-message connection churn and missed-heartbeat disconnects across ScanSync’s microservices, improving reliability for long-running OCR/upload/naming workloads and SSE updates.

Changes:

Introduces a unified RabbitMQClient + helper API in scansynclib.rabbitmq with long-lived, auto-reconnecting publish/consume patterns (and a backward-compatible connect_rabbitmq).
Refactors services to consume/publish via the shared helpers instead of duplicating connect/reconnect logic.
Adds focused unit tests for connection reuse and reconnect behavior.

Reviewed changes

Copilot reviewed 11 out of 11 changed files in this pull request and generated 5 comments.

Show a summary per file

File	Description
web_service/src/main.py	Keeps SSE RabbitMQ listener thread alive via reconnect loop.
upload_service/main.py	Switches upload consumer to unified `consume(...)` helper.
ocr_service/main.py	Switches OCR consumer to unified `consume(...)` helper.
metadata_service/main.py	Publishes to OCR queue via unified `publish(...)` and consumes via `consume(...)`.
file_naming_service/main.py	Switches consumer to unified `consume(...)` and changes ack/reconnect behavior.
detection_service/main.py	Uses `RabbitMQClient` + `process_events()` to keep a single connection alive between scans.
scansynclib/scansynclib/rabbitmq.py	New unified, reconnecting RabbitMQ connection/publish/consume implementation.
scansynclib/scansynclib/helpers.py	Re-exports RabbitMQ helpers for backward-compatible imports.
scansynclib/scansynclib/sqlite_wrapper.py	Routes SSE publish through unified exchange publisher helper.
tests/test_rabbitmq.py	Adds tests covering connection reuse and reconnect logic.
scansynclib/scansynclib.egg-info/SOURCES.txt	Updates packaged file list (currently introduces duplicates).

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


+    client = RabbitMQClient(name="detection")
+    client.ensure_connection()
+    client.declare_queue(RABBITQUEUE)


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


+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


maxi07 · 2026-07-02T16:15:48Z

remember to update rabbitmq as well.

Copilot

Pull request overview

Copilot reviewed 11 out of 11 changed files in this pull request and generated 3 comments.

+            # 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


+_publisher = RabbitMQClient(name="publisher")
+_publisher_lock = threading.Lock()
+


+scansynclib/config.py
+scansynclib/helpers.py
+scansynclib/logging.py
+scansynclib/ollama_helper.py
+scansynclib/onedrive_api.py


Copilot

Pull request overview

Copilot reviewed 12 out of 12 changed files in this pull request and generated 1 comment.

Copilot · 2026-07-02T17:22:30Z

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


Fixed in commit a3aebb8. The finally block now tracks ack_ok and only proceeds with DB updates and upload_queue forwarding when the ack succeeded. The test was also updated to verify that these follow-up actions are skipped on ack failure.

…ck fails

maxi07 · 2026-07-04T12:09:22Z

@copilot resolve the merge conflicts in this pull request

Resolve conflicts: - detection_service/main.py: keep RabbitMQClient setup, add cleanup_dangling_documents() from main - upload_service/main.py: keep consume() from helpers, add execute_query import from main

Copilot · 2026-07-04T12:12:17Z

@copilot resolve the merge conflicts in this pull request

Resolved in merge commit 2bb5f61. The two conflicts were:

detection_service/main.py: kept our RabbitMQClient + retry loop, and added cleanup_dangling_documents() from main (Add startup cleanup for dangling documents #71).
upload_service/main.py: kept consume() from our unified helpers, and added the execute_query import + finalize_sync_job function from main (Add OCR and sync logging overviews to settings #67).

…om/maxi07/ScanSync into copilot/handle-rabbitmq-connection

Initial plan

e8f3f2d

Copilot AI assigned Copilot and maxi07 Jul 2, 2026

Copilot started work on behalf of maxi07 July 2, 2026 15:02 View session

Copilot AI linked an issue Jul 2, 2026 that may be closed by this pull request

RabbitMQ is not used properly #14

Closed

2 tasks

Copilot AI added 2 commits July 2, 2026 15:05

Plan: unify RabbitMQ connection handling

6dc680a

Unify RabbitMQ handling into scansynclib.rabbitmq with persistent, re…

8b26c14

…connecting connections

Copilot AI changed the title ~~[WIP] Fix RabbitMQ connection handling by preserving the connection~~ Unify RabbitMQ handling with persistent, auto-reconnecting connections Jul 2, 2026

Copilot AI requested a review from maxi07 July 2, 2026 15:17

Copilot finished work on behalf of maxi07 July 2, 2026 15:17

maxi07 requested a review from Copilot July 2, 2026 16:11

Copilot started reviewing on behalf of maxi07 July 2, 2026 16:11 View session

Copilot AI reviewed Jul 2, 2026

View reviewed changes

Copilot started work on behalf of maxi07 July 2, 2026 16:16 View session

Address PR review reliability fixes

9b4015d

Copilot finished work on behalf of maxi07 July 2, 2026 16:21

maxi07 requested a review from Copilot July 2, 2026 16:23

Copilot started reviewing on behalf of maxi07 July 2, 2026 16:23 View session

Copilot AI reviewed Jul 2, 2026

View reviewed changes

Copilot started work on behalf of maxi07 July 2, 2026 16:57 View session

Fix review feedback for RabbitMQ follow-up

9ffdb35

Copilot finished work on behalf of maxi07 July 2, 2026 17:04

maxi07 requested a review from Copilot July 2, 2026 17:12

Copilot started reviewing on behalf of maxi07 July 2, 2026 17:12 View session

Copilot AI reviewed Jul 2, 2026

View reviewed changes

Copilot started work on behalf of maxi07 July 2, 2026 17:19 View session

Copilot AI added 2 commits July 2, 2026 17:20

fix: skip DB update and upload_queue forward when basic_ack fails

b6bd72c

test: update test to verify followup actions are skipped when basic_a…

5adcbac

…ck fails

Copilot finished work on behalf of maxi07 July 2, 2026 17:22

maxi07 marked this pull request as ready for review July 4, 2026 12:09

Copilot started work on behalf of maxi07 July 4, 2026 12:09 View session

Merge origin/main into copilot/handle-rabbitmq-connection

2bb5f61

Resolve conflicts: - detection_service/main.py: keep RabbitMQClient setup, add cleanup_dangling_documents() from main - upload_service/main.py: keep consume() from helpers, add execute_query import from main

Copilot finished work on behalf of maxi07 July 4, 2026 12:12

maxi07 added 3 commits July 4, 2026 14:16

update rabbitmq

b3b4724

Merge branch 'copilot/handle-rabbitmq-connection' of https://github.c…

eaeba5e

…om/maxi07/ScanSync into copilot/handle-rabbitmq-connection

fix pytests

e2f01e1

maxi07 merged commit c9c60ed into main Jul 4, 2026
3 checks passed

maxi07 deleted the copilot/handle-rabbitmq-connection branch July 4, 2026 12:32

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Unify RabbitMQ handling with persistent, auto-reconnecting connections#70

Unify RabbitMQ handling with persistent, auto-reconnecting connections#70
maxi07 merged 11 commits into
mainfrom
copilot/handle-rabbitmq-connection

Copilot AI commented Jul 2, 2026 •

edited

Loading

Uh oh!

Copilot AI left a comment

Uh oh!

maxi07 commented Jul 2, 2026

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI left a comment

Uh oh!

Copilot AI Jul 2, 2026

Uh oh!

maxi07 commented Jul 4, 2026

Uh oh!

Copilot AI commented Jul 4, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

		_publisher = RabbitMQClient(name="publisher")
		_publisher_lock = threading.Lock()

Conversation

Copilot AI commented Jul 2, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Root cause

Changes

Before → after

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

maxi07 commented Jul 2, 2026

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Uh oh!

Copilot AI Jul 2, 2026

Choose a reason for hiding this comment

Uh oh!

maxi07 commented Jul 4, 2026

Uh oh!

Copilot AI commented Jul 4, 2026

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Copilot AI commented Jul 2, 2026 •

edited

Loading