diff --git a/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenMonitoringIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenMonitoringIngestionUnitOfWork.cs index 928445b078..996e9ce4f3 100644 --- a/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenMonitoringIngestionUnitOfWork.cs +++ b/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() diff --git a/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenRecoverabilityIngestionUnitOfWork.cs b/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenRecoverabilityIngestionUnitOfWork.cs index 2351e78e3e..906b2e1f36 100644 --- a/src/ServiceControl.Persistence.RavenDB/UnitOfWork/RavenRecoverabilityIngestionUnitOfWork.cs +++ b/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; diff --git a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs b/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs index 1b89d399b1..a0c3f8f080 100644 --- a/src/ServiceControl.Persistence.Tests.RavenDB/Expiration/MessageExpiryTests.cs +++ b/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(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,7 +219,7 @@ 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 { @@ -177,6 +227,13 @@ public async Task RetryConfirmationProcessingShouldTriggerExpiration() {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.Empty, new TransportTransaction(), "receiveAddress", new ContextBag()); diff --git a/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs b/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs index 51e796fd83..acc048ba26 100644 --- a/src/ServiceControl.Persistence.Tests/MonitoringDataStoreTests.cs +++ b/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"); + } } } \ No newline at end of file