fix(queue): persist zero-loss message state in Redis with lease-based recovery

.github/workflows/architecture-check.yml

@@ -1,5 +1,12 @@
 name: Architecture Validation
 
+# Fork PRs run with a read-only GITHUB_TOKEN by default; the PR-comment
+# step needs write access to post the architecture report.
+permissions:
+  contents: read
+  issues: write
+  pull-requests: write
+
 on:
   push:
     branches:
@@ -130,6 +137,9 @@ jobs:
 
       - name: Comment on PR with results
         if: github.event_name == 'pull_request' && always()
+        # Fork PRs run with a read-only GITHUB_TOKEN that cannot write
+        # comments, so a skipped report comment must not fail the check.
+        continue-on-error: true
         uses: actions/github-script@v7
         with:
           script: |

README.md

@@ -275,6 +275,14 @@ sudo apt-get install redis-server
 redis-server
 ```
 
+> **Zero-loss messaging**: the queue's zero-loss message service persists
+> message state (payloads, attempts, processing leases, replication targets)
+> in Redis under the `zls:*` key namespace. All state survives process
+> restarts and is shared across horizontally scaled instances. A background
+> recovery sweep re-queues messages whose processing lease expired (worker
+> crash) exactly once, so Redis must be reachable for zero-loss guarantees
+> to hold.
+
 ### 4. Database Setup
 
 #### Development (SQLite)

src/queue/horizontal-scaling.controller.ts

@@ -310,16 +310,16 @@ export class HorizontalScalingController {
   @Get('zero-loss/stats')
   @ApiOperation({ summary: 'Get zero-loss message statistics' })
   @ApiResponse({ status: 200, description: 'Zero-loss stats' })
-  getZeroLossStats() {
+  async getZeroLossStats() {
     return this.zeroLoss.getStats();
   }
 
   @Get('zero-loss/messages/:messageId')
   @ApiOperation({ summary: 'Get message by ID' })
   @ApiResponse({ status: 200, description: 'Message details' })
   @ApiResponse({ status: 404, description: 'Message not found' })
-  getMessage(@Param('messageId') messageId: string) {
-    const message = this.zeroLoss.getMessage(messageId);
+  async getMessage(@Param('messageId') messageId: string) {
+    const message = await this.zeroLoss.getMessage(messageId);
     if (!message) {
       return { error: 'Message not found' };
     }
@@ -329,24 +329,24 @@ export class HorizontalScalingController {
   @Get('zero-loss/messages/queue/:queueName')
   @ApiOperation({ summary: 'Get messages for a queue' })
   @ApiResponse({ status: 200, description: 'Queue messages' })
-  getQueueMessages(@Param('queueName') queueName: string) {
+  async getQueueMessages(@Param('queueName') queueName: string) {
     return {
-      messages: this.zeroLoss.getQueueMessages(queueName),
+      messages: await this.zeroLoss.getQueueMessages(queueName),
     };
   }
 
   @Post('zero-loss/messages/:messageId/retry')
   @ApiOperation({ summary: 'Retry a failed message' })
   @ApiResponse({ status: 200, description: 'Retry initiated' })
-  retryMessage(@Param('messageId') messageId: string) {
-    const success = this.zeroLoss.retryMessage(messageId);
+  async retryMessage(@Param('messageId') messageId: string) {
+    const success = await this.zeroLoss.retryMessage(messageId);
     return { success };
   }
 
   @Get('zero-loss/messages/:messageId/verify')
   @ApiOperation({ summary: 'Verify message integrity' })
   @ApiResponse({ status: 200, description: 'Integrity check result' })
-  verifyMessageIntegrity(@Param('messageId') messageId: string) {
+  async verifyMessageIntegrity(@Param('messageId') messageId: string) {
     return this.zeroLoss.verifyMessageIntegrity(messageId);
   }
 

src/queue/horizontal-scaling.module.ts

@@ -3,6 +3,7 @@ import { Module } from '@nestjs/common';
 import { ConfigModule } from '@nestjs/config';
 import { ScheduleModule } from '@nestjs/schedule';
 import { EventEmitterModule } from '@nestjs/event-emitter';
+import { CustomCacheModule } from '../common/cache/cache.module';
 import { QueueWorkerManagerService } from './queue-worker-manager.service';
 import { QueueLoadBalancerService } from './queue-load-balancer.service';
 import { QueueFaultToleranceService } from './queue-fault-tolerance.service';
@@ -23,6 +24,9 @@ import { HorizontalScalingController } from './horizontal-scaling.controller';
     ConfigModule,
     ScheduleModule.forRoot(),
     EventEmitterModule.forRoot(),
+    // Provides RedisPoolService — the durable store backing the zero-loss
+    // message layer (same Redis infrastructure Bull already uses).
+    CustomCacheModule,
   ],
   controllers: [HorizontalScalingController],
   providers: [