Improved error handling around message queue initialization

Author: sweetlandj

Source/Platibus.MongoDB/MongoDBEventType.cs

@@ -16,5 +16,11 @@ public class MongoDBEventType
         /// Thrown when an index creation operation fails
         /// </summary>
         public static readonly DiagnosticEventType IndexCreationFailed = new DiagnosticEventType("IndexCreationFailed", DiagnosticEventLevel.Warn);
+
+        /// <summary>
+        /// Emitted when message headers or content cannot be read from the database
+        /// </summary>
+        public static readonly DiagnosticEventType MessageDocumentFormatError = new DiagnosticEventType("MessageDocumentFormatError", DiagnosticEventLevel.Error);
+
     }
 }

Source/Platibus.MongoDB/MongoDBMessageQueue.cs

@@ -200,6 +200,7 @@ private static string MapToCollectionName(QueueName queueName)
             return _queuedMessages.UpdateOneAsync(filter, update, cancellationToken: cancellationToken);
         }
 
+        /// <inheritdoc />
         /// <summary>
         /// Returns messages in the queue that are pending
         /// </summary>
@@ -216,10 +217,23 @@ private static string MapToCollectionName(QueueName queueName)
             var queuedMessages = new List<QueuedMessage>();
             foreach (var queuedMessage in existingMessages)
             {
-                var messageHeaders = new MessageHeaders(queuedMessage.Headers);
-                var principal = await _securityTokenService.NullSafeValidate(messageHeaders.SecurityToken);
-                var message = new Message(messageHeaders, queuedMessage.Content).WithoutSecurityToken();
-                queuedMessages.Add(new QueuedMessage(message, principal, queuedMessage.Attempts));
+                try
+                {
+                    var messageHeaders = new MessageHeaders(queuedMessage.Headers);
+                    var principal = await _securityTokenService.NullSafeValidate(messageHeaders.SecurityToken);
+                    var message = new Message(messageHeaders, queuedMessage.Content).WithoutSecurityToken();
+                    queuedMessages.Add(new QueuedMessage(message, principal, queuedMessage.Attempts));
+                }
+                catch (Exception ex)
+                {
+                    DiagnosticService.Emit(new MongoDBEventBuilder(this, MongoDBEventType.MessageDocumentFormatError)
+                    {
+                        Detail = "Error reading previously queued message document ID " + queuedMessage.Id + "; skipping",
+                        CollectionName = _queuedMessages.CollectionNamespace.CollectionName,
+                        DatabaseName = _queuedMessages.Database.DatabaseNamespace.DatabaseName,
+                        Exception = ex
+                    }.Build());
+                }
             }
             return queuedMessages;
         }

Source/Platibus/Diagnostics/DiagnosticEventType.cs

@@ -36,6 +36,12 @@ public class DiagnosticEventType
         /// </summary>
         public static readonly DiagnosticEventType ComponentInitialization = new DiagnosticEventType("ComponentInitialization", DiagnosticEventLevel.Info);
 
+        /// <summary>
+        /// Emitted when an error occurs during component or service initialization
+        /// </summary>
+        public static readonly DiagnosticEventType ComponentInitializationError = new DiagnosticEventType("ComponentInitializationError", DiagnosticEventLevel.Error);
+
+
         /// <summary>
         /// Emitted when a default configuration is used
         /// </summary>

Source/Platibus/Queueing/AbstractMessageQueue.cs

@@ -30,6 +30,7 @@
 
 namespace Platibus.Queueing
 {
+    /// <inheritdoc />
     /// <summary>
     /// An abstract base class for implementing message queues
     /// </summary>
@@ -123,7 +124,19 @@ protected AbstractMessageQueue(QueueName queueName, IQueueListener listener, Que
         {
             if (Interlocked.Exchange(ref _initialized, 1) == 0)
             {
-                await EnqueueExistingMessages(cancellationToken);
+                try
+                {
+                    await EnqueueExistingMessages(cancellationToken);
+                }
+                catch (Exception ex)
+                {
+                    DiagnosticService.Emit(
+                        new DiagnosticEventBuilder(this, DiagnosticEventType.ComponentInitializationError)
+                        {
+                            Detail = "Error enqueueing previously queued message(s)",
+                            Exception = ex
+                        }.Build());
+                }
             }
         }
 

Source/Platibus/SQL/SQLEventType.cs

@@ -31,5 +31,11 @@ public static class SQLEventType
         /// Emitted when there is an error opening or closing a connection
         /// </summary>
         public static readonly DiagnosticEventType CommandError = new DiagnosticEventType("CommandError", DiagnosticEventLevel.Error);
+
+        /// <summary>
+        /// Emitted when message headers or content cannot be read from the database
+        /// </summary>
+        public static readonly DiagnosticEventType MessageRecordFormatError = new DiagnosticEventType("MessageFormatError", DiagnosticEventLevel.Error);
+
     }
 }

Source/Platibus/SQL/SQLMessageQueue.cs

@@ -148,16 +148,27 @@ await DiagnosticService.EmitAsync(
                         {
                             while (await reader.ReadAsync(cancellationToken))
                             {
-                                var record = commandBuilder.BuildQueuedMessageRecord(reader);
-                                var messageContent = record.Content;
-                                var headers = DeserializeHeaders(record.Headers);
+                                try
+                                {
+                                    var record = commandBuilder.BuildQueuedMessageRecord(reader);
+                                    var messageContent = record.Content;
+                                    var headers = DeserializeHeaders(record.Headers);
 #pragma warning disable 612
-                                var principal = await ResolvePrincipal(headers, record.SenderPrincipal);
+                                    var principal = await ResolvePrincipal(headers, record.SenderPrincipal);
 #pragma warning restore 612
-                                var message = new Message(headers, messageContent).WithoutSecurityToken();
-                                var attempts = record.Attempts;
-                                var queuedMessage = new QueuedMessage(message, principal, attempts);
-                                queuedMessages.Add(queuedMessage);
+                                    var message = new Message(headers, messageContent).WithoutSecurityToken();
+                                    var attempts = record.Attempts;
+                                    var queuedMessage = new QueuedMessage(message, principal, attempts);
+                                    queuedMessages.Add(queuedMessage);
+                                }
+                                catch (Exception ex)
+                                {
+                                    DiagnosticService.Emit(new SQLEventBuilder(this, SQLEventType.MessageRecordFormatError)
+                                    {
+                                        Detail = "Error reading previously queued message record; skipping",
+                                        Exception = ex
+                                    }.Build());
+                                }
                             }
                         }
                     }