Handle Redis timeout in RedisSyncQueueService and improve SyncQueueProcessor logging

Author: davidwei

src/HappyNotes.Services/SyncQueue/Services/RedisSyncQueueService.cs

@@ -346,6 +346,11 @@ private async Task ProcessDelayedTasks(string service)
                 new RedisValue[] { now, batchLimit }
             );
         }
+        catch (RedisTimeoutException ex)
+        {
+            _logger.LogWarning(ex, "Redis timeout during delayed tasks processing for service {Service} - skipping this cycle", service);
+            return; // Skip this cycle, will retry on next dequeue attempt
+        }
         catch (StackExchange.Redis.RedisServerException ex) when (ex.Message.Contains("WRONGTYPE"))
         {
             _logger.LogWarning("Delayed queue {DelayedKey} has wrong type, recreating as sorted set", delayedKey);

src/HappyNotes.Services/SyncQueue/Services/SyncQueueProcessor.cs

@@ -77,71 +77,88 @@ private async Task ProcessServiceQueue(string serviceName, CancellationToken can
         _logger.LogInformation("Started processing queue for service: {ServiceName}", serviceName);
 
         using var semaphore = new SemaphoreSlim(_options.Processing.MaxConcurrentTasks);
+        var exitReason = "normal";
 
-        while (!cancellationToken.IsCancellationRequested)
+        try
         {
-            try
+            while (!cancellationToken.IsCancellationRequested)
             {
-                await semaphore.WaitAsync(cancellationToken);
-
-                var task = await _queueService.DequeueAsync<object>(serviceName, cancellationToken);
-
-                if (task == null)
+                try
                 {
-                    semaphore.Release();
-                    await Task.Delay(_options.Processing.PollingInterval, cancellationToken);
-                    continue;
-                }
+                    await semaphore.WaitAsync(cancellationToken);
 
-                // Process task in background with fresh scope per task
-                _ = Task.Run(async () =>
-                {
-                    using var taskScope = _serviceProvider.CreateScope();
-                    try
+                    var task = await _queueService.DequeueAsync<object>(serviceName, cancellationToken);
+
+                    if (task == null)
                     {
-                        // Resolve handler fresh for each task (with clean scoped dependencies)
-                        var handler = taskScope.ServiceProvider.GetServices<ISyncHandler>()
-                            .FirstOrDefault(h => h.ServiceName == serviceName);
+                        semaphore.Release();
+                        await Task.Delay(_options.Processing.PollingInterval, cancellationToken);
+                        continue;
+                    }
 
-                        if (handler == null)
+                    // Process task in background with fresh scope per task
+                    _ = Task.Run(async () =>
+                    {
+                        using var taskScope = _serviceProvider.CreateScope();
+                        try
                         {
-                            _logger.LogError("No handler found for service {ServiceName}", serviceName);
-                            return;
+                            // Resolve handler fresh for each task (with clean scoped dependencies)
+                            var handler = taskScope.ServiceProvider.GetServices<ISyncHandler>()
+                                .FirstOrDefault(h => h.ServiceName == serviceName);
+
+                            if (handler == null)
+                            {
+                                _logger.LogError("No handler found for service {ServiceName}", serviceName);
+                                return;
+                            }
+
+                            var syncTask = new SyncTask
+                            {
+                                Id = task.Id,
+                                Service = task.Service,
+                                Action = task.Action,
+                                EntityId = task.EntityId,
+                                UserId = task.UserId,
+                                Payload = task.Payload,
+                                AttemptCount = task.AttemptCount,
+                                CreatedAt = task.CreatedAt,
+                                ScheduledFor = task.ScheduledFor,
+                                Metadata = task.Metadata
+                            };
+                            await ProcessTask(handler, syncTask, cancellationToken);
                         }
-
-                        var syncTask = new SyncTask
+                        finally
                         {
-                            Id = task.Id,
-                            Service = task.Service,
-                            Action = task.Action,
-                            EntityId = task.EntityId,
-                            UserId = task.UserId,
-                            Payload = task.Payload,
-                            AttemptCount = task.AttemptCount,
-                            CreatedAt = task.CreatedAt,
-                            ScheduledFor = task.ScheduledFor,
-                            Metadata = task.Metadata
-                        };
-                        await ProcessTask(handler, syncTask, cancellationToken);
-                    }
-                    finally
-                    {
-                        semaphore.Release();
-                    }
-                }, cancellationToken);
+                            semaphore.Release();
+                        }
+                    }, cancellationToken);
+                }
+                catch (OperationCanceledException)
+                {
+                    exitReason = "cancelled";
+                    break;
+                }
+                catch (Exception ex)
+                {
+                    semaphore.Release(); // Release semaphore to prevent deadlock
+                    _logger.LogError(ex, "Error in queue processing loop for service {ServiceName}", serviceName);
+                    await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken);
+                }
             }
-            catch (OperationCanceledException)
+        }
+        finally
+        {
+            if (exitReason == "cancelled")
             {
-                break;
+                _logger.LogWarning("Processing queue for service {ServiceName} was cancelled at {StopTime}",
+                    serviceName, DateTime.UtcNow);
             }
-            catch (Exception ex)
+            else
             {
-                _logger.LogError(ex, "Error in queue processing loop for service {ServiceName}", serviceName);
-                await Task.Delay(TimeSpan.FromSeconds(10), cancellationToken);
+                _logger.LogInformation("Processing queue for service {ServiceName} stopped normally at {StopTime}",
+                    serviceName, DateTime.UtcNow);
             }
         }
-
-        _logger.LogInformation("Stopped processing queue for service: {ServiceName}", serviceName);
     }
 
     private async Task ProcessTask(ISyncHandler handler, SyncTask task, CancellationToken cancellationToken)