From b5ca0cf0f55634e191efdcc51c3c9cf4b864700f Mon Sep 17 00:00:00 2001 From: Nicolas Date: Thu, 7 Aug 2025 18:29:59 +0800 Subject: [PATCH 1/5] fix: update devops service to subscribe to correct reconciler heartbeat queue --- apps/devops/app/providers/message_queue.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/apps/devops/app/providers/message_queue.py b/apps/devops/app/providers/message_queue.py index 978e232..61e5c23 100644 --- a/apps/devops/app/providers/message_queue.py +++ b/apps/devops/app/providers/message_queue.py @@ -15,7 +15,7 @@ def register(app): try: # Initialize services during startup to avoid blocking app initialization print("🔧 Initializing services...") - app.deployment_heartbeat_subscriber = AsyncMQSubscriber("devops_reconcile_heartbeat") + app.deployment_heartbeat_subscriber = AsyncMQSubscriber("reconciler.output") app.deployment_status_service = DeploymentStatusUpdateService() print("✅ Services initialized") From e4fe9394b1a71ae0df946007c78cc29bdabf0e26 Mon Sep 17 00:00:00 2001 From: Nicolas Date: Fri, 8 Aug 2025 12:01:47 +0800 Subject: [PATCH 2/5] fix: use environment variable for RabbitMQ output queue name - Add RABBITMQ_OUTPUT_QUEUE_NAME environment variable support - Fix hardcoded queue name 'reconciler.output' to use configurable queue name - Default to 'freeleaps.devops.reconciler.output' if env var not set - Add debug logging to show which queue name is being used - This fixes the issue where 42 messages were stuck in the output queue due to queue name mismatch --- apps/devops/app/providers/message_queue.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/apps/devops/app/providers/message_queue.py b/apps/devops/app/providers/message_queue.py index 61e5c23..b2d87cb 100644 --- a/apps/devops/app/providers/message_queue.py +++ b/apps/devops/app/providers/message_queue.py @@ -1,4 +1,5 @@ import asyncio +import os from app.backend.infra.rabbitmq.async_subscriber import AsyncMQSubscriber from app.backend.services.deployment_status_update_service import DeploymentStatusUpdateService @@ -15,7 +16,9 @@ def register(app): try: # Initialize services during startup to avoid blocking app initialization print("🔧 Initializing services...") - app.deployment_heartbeat_subscriber = AsyncMQSubscriber("reconciler.output") + output_queue_name = os.getenv("RABBITMQ_OUTPUT_QUEUE_NAME", "freeleaps.devops.reconciler.output") + print(f"Using output queue: {output_queue_name}") + app.deployment_heartbeat_subscriber = AsyncMQSubscriber(output_queue_name) app.deployment_status_service = DeploymentStatusUpdateService() print("✅ Services initialized") From 4340949f5734f60eac280b0ef524e53039b5d7a3 Mon Sep 17 00:00:00 2001 From: Nicolas Date: Fri, 8 Aug 2025 12:33:14 +0800 Subject: [PATCH 3/5] fix: make startup and shutdown event handlers async in common provider - Fix TypeError: object NoneType can't be used in 'await' expression - FastAPI requires event handlers to be async functions - This was blocking the entire application startup sequence - Fixes the issue where message queue consumers were not starting properly --- apps/devops/app/providers/common.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/apps/devops/app/providers/common.py b/apps/devops/app/providers/common.py index 64a9a44..656bcc3 100644 --- a/apps/devops/app/providers/common.py +++ b/apps/devops/app/providers/common.py @@ -11,13 +11,13 @@ def register(app): # This hook ensures that a connection is opened to handle any queries # generated by the request. @app.on_event("startup") - def startup(): + async def startup(): pass # This hook ensures that the connection is closed when we've finished # processing the request. @app.on_event("shutdown") - def shutdown(): + async def shutdown(): pass From 04acd78d78359d791d6a7187a2eab4bf03cbea23 Mon Sep 17 00:00:00 2001 From: Nicolas Date: Fri, 8 Aug 2025 12:45:21 +0800 Subject: [PATCH 4/5] fix: connect to existing named queue instead of creating anonymous queue - Change AsyncMQClient to connect to existing persistent queue by name - Fix issue where DevOps Service created temporary anonymous queues instead of consuming from the correct named queue - This allows consuming the 42 backlogged messages in freeleaps.devops.reconciler.output queue - Change queue properties: exclusive=False, auto_delete=False, durable=True - Resolves the core issue where messages were split between persistent and temporary queues --- apps/devops/app/backend/infra/rabbitmq/async_client.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/devops/app/backend/infra/rabbitmq/async_client.py b/apps/devops/app/backend/infra/rabbitmq/async_client.py index a48fc58..39e9f28 100644 --- a/apps/devops/app/backend/infra/rabbitmq/async_client.py +++ b/apps/devops/app/backend/infra/rabbitmq/async_client.py @@ -37,8 +37,10 @@ class AsyncMQClient: name=self.exchange_name, type="direct", auto_delete=False ) + # Connect to existing named queue instead of creating anonymous queue + # channel_name already contains the full queue name from environment variable self.queue = await self.channel.declare_queue( - name=None, exclusive=True, auto_delete=True, durable=False + name=self.channel_name, exclusive=False, auto_delete=False, durable=True ) await self.queue.bind( exchange=self.exchange, routing_key=self.routing_key From 8caa8e6b31c46e205b0179024e080e2aac81c162 Mon Sep 17 00:00:00 2001 From: dongli Date: Sun, 10 Aug 2025 18:21:39 -0700 Subject: [PATCH 5/5] Add dummy change to trigger ci/cd --- apps/devops/app/main.py | 1 + 1 file changed, 1 insertion(+) diff --git a/apps/devops/app/main.py b/apps/devops/app/main.py index 559d7ed..1bfc3d4 100644 --- a/apps/devops/app/main.py +++ b/apps/devops/app/main.py @@ -13,4 +13,5 @@ async def root(): if __name__ == "__main__": import uvicorn + print("Starting FastAPI server...") uvicorn.run("main:app", host=site_settings.SERVER_HOST, port=site_settings.SERVER_PORT, reload=True) \ No newline at end of file