Merge pull request #5639 from Particular/john/raven_bugs

Fix RavenDB ingestion state handling

Author: johnsimons

src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenMonitoringIngestionUnitOfWork.cs

@@ -28,7 +28,23 @@ static PatchCommandData CreateKnownEndpointsPutCommand(KnownEndpoint endpoint)
 
             var docId = RavenMonitoringDataStore.MakeDocumentId(endpoint.EndpointDetails.GetDeterministicId());
 
-            var request = new PatchRequest
+            // Ingestion always observes endpoints with Monitored = false, so patching an
+            // already-known endpoint must not stomp a user-set Monitored = true flag back to false.
+            var existingDocPatch = new PatchRequest
+            {
+                Script = @$"
+                    var insert = {document};
+
+                    for(var key in insert) {{
+                        if(insert.hasOwnProperty(key) && key !== '{nameof(KnownEndpoint.Monitored)}') {{
+                            this[key] = insert[key];
+                        }}
+                    }}"
+            };
+
+            // A newly discovered endpoint has no existing Monitored value to preserve, so it
+            // still gets the full document, including Monitored = false.
+            var patchIfMissing = new PatchRequest
             {
                 Script = @$"
                     var insert = {document};
@@ -40,7 +56,7 @@ static PatchCommandData CreateKnownEndpointsPutCommand(KnownEndpoint endpoint)
                     }}"
             };
 
-            return new PatchCommandData(docId, null, request, request);
+            return new PatchCommandData(docId, null, existingDocPatch, patchIfMissing);
         }
 
         static RavenMonitoringIngestionUnitOfWork()

src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenRecoverabilityIngestionUnitOfWork.cs

@@ -93,11 +93,11 @@ ICommandData CreateFailedMessagesPatchCommand(string uniqueMessageId, FailedMess
             const string AttemptedAt = nameof(FailedMessage.ProcessingAttempt.AttemptedAt);
 
             //HINT: RavenDB 4.2 removed Lodash utility functions, but supports ECMAScript 5.1 and some 6.0 features like arrow functions and array primitive functions
-            return new PatchCommandData(documentId, null, new PatchRequest
+            var existingDocPatch = new PatchRequest
             {
                 Script = $@"this.{nameof(FailedMessage.Status)} = args.status;
                                 this.{nameof(FailedMessage.FailureGroups)} = args.failureGroups;
-                                
+
                                 var newAttempts = this.{nameof(FailedMessage.ProcessingAttempts)};
 
                                 //De-duplicate attempts by AttemptedAt value
@@ -109,7 +109,7 @@ ICommandData CreateFailedMessagesPatchCommand(string uniqueMessageId, FailedMess
 
                                 //Trim to the latest MaxProcessingAttempts
                                 newAttempts.sort((a, b) => a.{AttemptedAt} > b.{AttemptedAt} ? 1 : -1);
-                                
+
                                 if(newAttempts.length > {MaxProcessingAttempts})
                                 {{
                                     newAttempts = newAttempts.slice(newAttempts.length - {MaxProcessingAttempts}, newAttempts.length);
@@ -123,7 +123,13 @@ ICommandData CreateFailedMessagesPatchCommand(string uniqueMessageId, FailedMess
                         {"failureGroups", groups},
                         {"attempt", processingAttempt}
                     },
-            },
+            };
+
+            // A message re-failing must not keep an @expires stamp set by an earlier
+            // resolve/archive/retry, otherwise it can be silently deleted while still Unresolved.
+            expirationManager.CancelExpiration(existingDocPatch);
+
+            return new PatchCommandData(documentId, null, existingDocPatch,
                 patchIfMissing: new PatchRequest
                 {
                     Script = $@"this.{nameof(FailedMessage.Status)} = args.status;

src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs

@@ -141,6 +141,56 @@ public async Task SingleMessageMarkedAsResolvedShouldExpire()
             await WaitUntil(async () => (await GetAllMessages()).Results.Count == 0, "Archived message should be removed after archiving.");
         }
 
+        [Test]
+        public async Task MessageFailingAgainAfterRetryShouldNotKeepExpiration()
+        {
+            var (context, attempt) = CreateMessageContext();
+            var uniqueMessageId = context.Headers.UniqueId();
+
+            await DisableExpiration();
+
+            await using (var uow = await IngestionUnitOfWorkFactory.StartNew())
+            {
+                await uow.Recoverability.RecordFailedProcessingAttempt(context, attempt, []);
+
+                await uow.Complete(TestContext.CurrentContext.CancellationToken);
+            }
+
+            await CompleteDatabaseOperation();
+
+            // Successful retry stamps @expires on the FailedMessage document.
+            await using (var uow = await IngestionUnitOfWorkFactory.StartNew())
+            {
+                await uow.Recoverability.RecordSuccessfulRetry(uniqueMessageId);
+
+                await uow.Complete(TestContext.CurrentContext.CancellationToken);
+            }
+
+            await CompleteDatabaseOperation();
+
+            // The same logical message fails again before the retention period elapses.
+            var (context2, attempt2) = CreateMessageContext(uniqueMessageId);
+
+            await using (var uow = await IngestionUnitOfWorkFactory.StartNew())
+            {
+                await uow.Recoverability.RecordFailedProcessingAttempt(context2, attempt2, []);
+
+                await uow.Complete(TestContext.CurrentContext.CancellationToken);
+            }
+
+            await CompleteDatabaseOperation();
+
+            var documentId = FailedMessageIdGenerator.MakeDocumentId(uniqueMessageId);
+
+            using var session = DocumentStore.OpenAsyncSession();
+            var failedMessage = await session.LoadAsync<FailedMessage>(documentId);
+            var metadata = session.Advanced.GetMetadataFor(failedMessage);
+
+            Assert.That(failedMessage.Status, Is.EqualTo(FailedMessageStatus.Unresolved));
+            Assert.That(metadata.ContainsKey(Raven.Client.Constants.Documents.Metadata.Expires), Is.False,
+                "A message that fails again after being resolved should not retain its previous @expires stamp.");
+        }
+
         [Test]
         public async Task RetryConfirmationProcessingShouldTriggerExpiration()
         {
@@ -169,14 +219,21 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration()
             await WaitUntil(async () => (await GetAllMessages()).Results.Count == 0, "Retry confirmation should cause message removal.");
         }
 
-        static (MessageContext, FailedMessage.ProcessingAttempt) CreateMessageContext()
+        static (MessageContext, FailedMessage.ProcessingAttempt) CreateMessageContext(string forceUniqueMessageId = null)
         {
             var headers = new Dictionary<string, string>
             {
                 {Headers.ProcessingEndpoint, "SomeEndpoint"},
                 {Headers.MessageId, Guid.NewGuid().ToString() }
             };
 
+            // Forces context.Headers.UniqueId() to resolve to a specific, already-known
+            // UniqueMessageId, so a test can simulate the same logical message failing again.
+            if (forceUniqueMessageId != null)
+            {
+                headers["ServiceControl.Retry.UniqueMessageId"] = forceUniqueMessageId;
+            }
+
             var attempt = FailedMessageBuilder.Minimal().ProcessingAttempts.First();
 
             var message = new MessageContext(Guid.NewGuid().ToString(), headers, ReadOnlyMemory<byte>.Empty, new TransportTransaction(), "receiveAddress", new ContextBag());

src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs

@@ -227,5 +227,67 @@ public async Task Unit_of_work_detects_endpoint()
                 Assert.That(fromStorage.Monitored, Is.EqualTo(knownEndpoint.Monitored), "Monitored should match");
             }
         }
+
+        [Test]
+        public async Task Ingesting_a_known_endpoint_does_not_reset_monitored_flag()
+        {
+            var endpointDetails = new EndpointDetails { Host = "Host1", HostId = Guid.NewGuid(), Name = "Endpoint" };
+
+            await MonitoringDataStore.CreateIfNotExists(endpointDetails);
+            await MonitoringDataStore.UpdateEndpointMonitoring(endpointDetails, true);
+
+            await CompleteDatabaseOperation();
+
+            // Error ingestion always records endpoints it observes with Monitored = false
+            // (ErrorProcessor.RecordKnownEndpoints), regardless of the endpoint's current
+            // monitored state. Recording it again must not stomp a user-set Monitored = true.
+            var observedAgain = new KnownEndpoint
+            {
+                HostDisplayName = endpointDetails.Host,
+                EndpointDetails = endpointDetails,
+                Monitored = false
+            };
+
+            await using (var unitOfWork = await UnitOfWorkFactory.StartNew())
+            {
+                await unitOfWork.Monitoring.RecordKnownEndpoint(observedAgain);
+
+                await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken);
+            }
+
+            await CompleteDatabaseOperation();
+
+            var knownEndpoints = await MonitoringDataStore.GetAllKnownEndpoints();
+            var fromStorage = knownEndpoints.Single(e => e.EndpointDetails.HostId == endpointDetails.HostId);
+
+            Assert.That(fromStorage.Monitored, Is.True, "Ingestion must not reset an existing endpoint's Monitored flag back to false");
+        }
+
+        [Test]
+        public async Task Ingesting_a_previously_unknown_endpoint_creates_it_unmonitored()
+        {
+            var endpointDetails = new EndpointDetails { Host = "Host1", HostId = Guid.NewGuid(), Name = "Endpoint" };
+
+            var observed = new KnownEndpoint
+            {
+                HostDisplayName = endpointDetails.Host,
+                EndpointDetails = endpointDetails,
+                Monitored = false
+            };
+
+            await using (var unitOfWork = await UnitOfWorkFactory.StartNew())
+            {
+                await unitOfWork.Monitoring.RecordKnownEndpoint(observed);
+
+                await unitOfWork.Complete(TestContext.CurrentContext.CancellationToken);
+            }
+
+            await CompleteDatabaseOperation();
+
+            var knownEndpoints = await MonitoringDataStore.GetAllKnownEndpoints();
+            var fromStorage = knownEndpoints.Single(e => e.EndpointDetails.HostId == endpointDetails.HostId);
+
+            Assert.That(fromStorage.Monitored, Is.False, "A newly discovered endpoint should be created unmonitored");
+        }
     }
 }