[DO NOT MERGE] Spike managing saga timeouts

src/ServiceControl.AcceptanceTests.RavenDB/SagaTimeouts/When_a_saga_timeout_is_requested.cs

@@ -0,0 +1,89 @@
+namespace ServiceControl.AcceptanceTests.RavenDB.SagaTimeouts
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Linq;
+    using System.Text;
+    using System.Threading.Tasks;
+    using AcceptanceTesting;
+    using AcceptanceTesting.EndpointTemplates;
+    using NServiceBus;
+    using NServiceBus.AcceptanceTesting;
+    using NServiceBus.Routing;
+    using NServiceBus.Transport;
+    using NUnit.Framework;
+    using Persistence.SagaTimeouts;
+    using ServiceBus.Management.Infrastructure.Settings;
+
+    class When_a_saga_timeout_is_requested : AcceptanceTest
+    {
+        [Test]
+        public async Task Should_list_it_as_pending()
+        {
+            SagaTimeoutView listed = null;
+            SagaTimeoutDetails details = null;
+
+            await Define<Context>()
+                .WithEndpoint<Sales>()
+                .Do("Wait for the timeout to be listed", async ctx =>
+                {
+                    var timeouts = await this.TryGetMany<SagaTimeoutView>("/api/timeouts?status=pending", timeout => timeout.TimeoutId == ctx.TimeoutId);
+                    listed = timeouts.HasResult ? timeouts.Items.Single() : null;
+                    return listed != null;
+                })
+                .Do("Read its details", async ctx =>
+                {
+                    var result = await this.TryGet<SagaTimeoutDetails>($"/api/timeouts/{ctx.TimeoutId}");
+                    details = result.HasResult ? result.Item : null;
+                    return details != null;
+                })
+                .Done(_ => true)
+                .Run();
+
+            Assert.Multiple(() =>
+            {
+                Assert.That(listed.Status, Is.EqualTo(SagaTimeoutStatus.Pending));
+                Assert.That(listed.Version, Is.EqualTo(1));
+                Assert.That(listed.ManagementAddress, Is.EqualTo("Sales.SagaTimeouts"));
+                Assert.That(details.Payload, Is.EqualTo("{\"OrderId\":\"order-1\"}"));
+                Assert.That(details.History.Single().Kind, Is.EqualTo(SagaTimeoutChangeKind.Requested));
+            });
+        }
+
+        class Context : ScenarioContext, ISequenceContext
+        {
+            public int Step { get; set; }
+            public string TimeoutId { get; } = Guid.NewGuid().ToString();
+        }
+
+        public class Sales : EndpointConfigurationBuilder
+        {
+            public Sales() => EndpointSetup<DefaultServerWithoutAudit>(c => c.EnableFeature<ReportTimeoutRequest>());
+
+            // Sent as NServiceBus sends it: the enclosed type is the full name without an assembly, and the body is System.Text.Json
+            class ReportTimeoutRequest : DispatchRawMessages<Context>
+            {
+                protected override TransportOperations CreateMessage(Context context)
+                {
+                    var payload = Convert.ToBase64String(Encoding.UTF8.GetBytes("{\"OrderId\":\"order-1\"}"));
+                    var body = $$"""
+                        {"TriggerId":"trigger-1","DueTime":"{{DateTimeOffset.UtcNow.AddHours(1):O}}","PayloadContentType":"application/json","Payload":"{{payload}}",
+                         "TimeoutId":"{{context.TimeoutId}}","TimeoutType":"Sales.PaymentTimeout","Version":1,"EndpointName":"Sales","SagaType":"Sales.OrderSaga",
+                         "SagaId":"{{Guid.NewGuid()}}","CorrelationPropertyValue":"order-1","ChangedAt":"{{DateTimeOffset.UtcNow:O}}",
+                         "Host":"host","HostId":"host-id","ManagementAddress":"Sales.SagaTimeouts"}
+                        """;
+
+                    var headers = new Dictionary<string, string>
+                    {
+                        [Headers.EnclosedMessageTypes] = "NServiceBus.SagaTimeoutRequested",
+                        [Headers.ContentType] = "application/json",
+                        [Headers.MessageIntent] = "Send"
+                    };
+
+                    var message = new OutgoingMessage(Guid.NewGuid().ToString(), headers, Encoding.UTF8.GetBytes(body));
+                    return new TransportOperations(new TransportOperation(message, new UnicastAddressTag(Settings.DEFAULT_INSTANCE_NAME)));
+                }
+            }
+        }
+    }
+}

src/ServiceControl.Infrastructure/Auth/MessageAction.cs

@@ -8,7 +8,10 @@ public enum MessageActionKind
     Archive,
     Unarchive,
     Edit,
-    Delete
+    Delete,
+    Reschedule,
+    Trigger,
+    Cancel
 }
 
 /// <summary>How the action selected the messages it acts on.</summary>

src/ServiceControl.Infrastructure/Auth/Permissions.cs

@@ -55,6 +55,11 @@ public static class Permissions
     /// <summary>Sagas area — viewing saga audit data.</summary>
     public const string ErrorSagasView = "error:sagas:view";
 
+    /// <summary>Saga timeouts area — viewing saga timeouts and requesting changes to them.</summary>
+    public const string ErrorTimeoutsView = "error:timeouts:view";
+    /// <inheritdoc cref="ErrorTimeoutsView"/>
+    public const string ErrorTimeoutsManage = "error:timeouts:manage";
+
     /// <summary>Event log area — viewing the event log.</summary>
     public const string ErrorEventLogView = "error:eventlog:view";
 

src/ServiceControl.Infrastructure/Auth/RolePermissions.cs

@@ -19,6 +19,7 @@ public static class RolePermissions
         Permissions.ErrorHeartbeatsView,
         Permissions.ErrorCustomChecksView,
         Permissions.ErrorSagasView,
+        Permissions.ErrorTimeoutsView,
         Permissions.ErrorEventLogView,
         Permissions.ErrorQueuesView,
         Permissions.ErrorConnectionsView,
@@ -50,6 +51,7 @@ public static class RolePermissions
         Permissions.ErrorEndpointsManage,
         Permissions.ErrorEndpointsDelete,
         Permissions.ErrorCustomChecksDelete,
+        Permissions.ErrorTimeoutsManage,
         Permissions.ErrorQueuesDelete,
         Permissions.ErrorConnectionsManage,
         Permissions.MonitoringEndpointDelete,

src/ServiceControl.Persistence.EFCore/Abstractions/BasePersistence.cs

@@ -15,6 +15,7 @@ namespace ServiceControl.Persistence.EFCore.Abstractions;
 using ServiceControl.Persistence.EFCore.Infrastructure.Metrics;
 using ServiceControl.Persistence.MessageRedirects;
 using ServiceControl.Persistence.Recoverability;
+using ServiceControl.Persistence.SagaTimeouts;
 using ServiceControl.Persistence.UnitOfWork;
 using ServiceControl.Recoverability;
 using ServiceControl.Recoverability.Archiving.Metrics;
@@ -73,6 +74,7 @@ protected static void RegisterDataStores(IServiceCollection services, EFPersiste
         services.AddSingleton<IRetryHistoryDataStore, RetryHistoryDataStore>();
         services.AddSingleton<IEndpointSettingsStore, EndpointSettingsStore>();
         services.AddSingleton<ITrialLicenseDataProvider, TrialLicenseDataProvider>();
+        services.AddSingleton<ISagaTimeoutDataStore, SagaTimeoutDataStore>();
 
         services.AddSingleton<ILicensingDataStore, LicensingDataStore>();
 

src/ServiceControl.Persistence.EFCore/Implementation/SagaTimeoutDataStore.cs

@@ -0,0 +1,26 @@
+namespace ServiceControl.Persistence.EFCore.Implementation;
+
+using ServiceControl.Persistence.Infrastructure;
+using ServiceControl.Persistence.SagaTimeouts;
+
+// Saga timeout management is implemented for RavenDB only so far
+public class SagaTimeoutDataStore : ISagaTimeoutDataStore
+{
+    public Task<SagaTimeoutApplyOutcome> Apply(SagaTimeoutChange change, CancellationToken cancellationToken = default) => throw NotSupported();
+
+    public Task<SagaTimeoutApplyOutcome> ApplySagaCompleted(SagaCompletedChange change, CancellationToken cancellationToken = default) => throw NotSupported();
+
+    public Task<QueryResult<IList<SagaTimeoutView>>> Query(SagaTimeoutQuery query, PagingInfo paging, SortInfo sort, CancellationToken cancellationToken = default) => throw NotSupported();
+
+    public Task<SagaTimeoutCounts> GetCounts(CancellationToken cancellationToken = default) => throw NotSupported();
+
+    public Task<SagaTimeoutDetails?> Get(string timeoutId, CancellationToken cancellationToken = default) => throw NotSupported();
+
+    public Task<IList<SagaTimeoutView>> GetBySaga(Guid sagaId, CancellationToken cancellationToken = default) => throw NotSupported();
+
+    public Task<SagaTimeoutOperatorRequestResult> RecordOperatorRequest(SagaTimeoutOperatorRequest request, CancellationToken cancellationToken = default) => throw NotSupported();
+
+    public Task Forget(string timeoutId, CancellationToken cancellationToken = default) => throw NotSupported();
+
+    static NotSupportedException NotSupported() => new("Saga timeout management is not supported by this persister yet.");
+}

src/ServiceControl.Persistence.RavenDB/ExpirationManager.cs

@@ -34,6 +34,19 @@ public void EnableExpiration(IAsyncDocumentSession session, FailedMessage failed
             session.Advanced.GetMetadataFor(failedMessage)[Constants.Documents.Metadata.Expires] = expiresAt;
         }
 
+        // For error-instance data other than failed messages, such as finished saga timeouts
+        public void EnableErrorRetentionExpiration(IAsyncDocumentSession session, object document)
+        {
+            var expiresAt = timeProvider.GetUtcNow().UtcDateTime + errorRetentionPeriod;
+
+            session.Advanced.GetMetadataFor(document)[Constants.Documents.Metadata.Expires] = expiresAt;
+        }
+
+        public void CancelExpiration(IAsyncDocumentSession session, object document)
+        {
+            session.Advanced.GetMetadataFor(document).Remove(Constants.Documents.Metadata.Expires);
+        }
+
         public void EnableExpiration(IAsyncDocumentSession session, EventLogItem eventLogItem)
         {
             var expiresAt = timeProvider.GetUtcNow().UtcDateTime + eventsRetentionPeriod;

src/ServiceControl.Persistence.RavenDB/RavenPersistence.cs

@@ -13,8 +13,10 @@ namespace ServiceControl.Persistence.RavenDB;
 using Persistence.MessageRedirects;
 using Persistence.Recoverability;
 using Recoverability;
+using SagaTimeouts;
 using ServiceControl.CustomChecks;
 using ServiceControl.Infrastructure.RavenDB.Subscriptions;
+using ServiceControl.Persistence.SagaTimeouts;
 using ServiceControl.Recoverability;
 using Throughput;
 using UnitOfWork;
@@ -75,6 +77,7 @@ public void AddPersistence(IServiceCollection services)
         services.AddSingleton<IRetryHistoryDataStore, RetryHistoryDataStore>();
         services.AddSingleton<IEndpointSettingsStore, EndpointSettingsStore>();
         services.AddSingleton<ITrialLicenseDataProvider, TrialLicenseDataProvider>();
+        services.AddSingleton<ISagaTimeoutDataStore, SagaTimeoutDataStore>();
     }
 
     public void AddInstaller(IServiceCollection services)

src/ServiceControl.Persistence.RavenDB/SagaTimeouts/SagaTimeout.cs

@@ -0,0 +1,87 @@
+namespace ServiceControl.Persistence.RavenDB.SagaTimeouts
+{
+    using System;
+    using System.Collections.Generic;
+    using ServiceControl.Persistence.SagaTimeouts;
+
+    // The type names give the collection names SagaTimeouts and CompletedSagas
+    class SagaTimeout
+    {
+        public string Id { get; set; }
+        public string TimeoutId { get; set; }
+        public string EndpointName { get; set; }
+        public string SagaType { get; set; }
+        public Guid SagaId { get; set; }
+        public string CorrelationValue { get; set; }
+        public string TimeoutType { get; set; }
+        public int Version { get; set; }
+        public int Rank { get; set; }
+        public string TriggerId { get; set; }
+        public DateTime? DueTime { get; set; }
+        public SagaTimeoutStatus Status { get; set; }
+        public DateTime? RequestedAt { get; set; }
+        public SagaTimeoutChangeKind LastChangeKind { get; set; }
+        public SagaTimeoutChangeSource LastChangeSource { get; set; }
+        public DateTime? LastChangedAt { get; set; }
+        public DateTime? StatusChangedAt { get; set; }
+
+        // ChangedAt of the newest report, which decides whether its host and management address replace the stored ones
+        public DateTime? LastReportedAt { get; set; }
+        public string Host { get; set; }
+        public string HostId { get; set; }
+        public string ManagementAddress { get; set; }
+
+        public PendingOperation PendingOperation { get; set; }
+        public bool HistoryTruncated { get; set; }
+        public List<HistoryEntry> Changes { get; set; } = [];
+
+        // The payload of each version is an attachment named by PayloadAttachmentName
+        public List<PayloadVersion> Payloads { get; set; } = [];
+
+        public static string MakeDocumentId(string timeoutId) => $"SagaTimeouts/{timeoutId}";
+
+        public static string PayloadAttachmentName(int version) => $"payload-v{version}";
+    }
+
+    class PendingOperation
+    {
+        public Guid OperationId { get; set; }
+        public SagaTimeoutOperation Change { get; set; }
+        public DateTime RequestedAt { get; set; }
+        public DateTime? RequestedDueTime { get; set; }
+    }
+
+    class HistoryEntry
+    {
+        public string DeduplicationKey { get; set; }
+        public SagaTimeoutChangeKind Kind { get; set; }
+        public int Version { get; set; }
+        public string TriggerId { get; set; }
+        public DateTime? DueTime { get; set; }
+        public DateTime ChangedAt { get; set; }
+        public DateTime ReceivedAt { get; set; }
+        public SagaTimeoutChangeSource Source { get; set; }
+        public Guid? OperationId { get; set; }
+        public string RequestedChange { get; set; }
+        public DateTime? RequestedDueTime { get; set; }
+        public string Reason { get; set; }
+    }
+
+    class PayloadVersion
+    {
+        public int Version { get; set; }
+        public string ContentType { get; set; }
+    }
+
+    class CompletedSaga
+    {
+        public string Id { get; set; }
+        public Guid SagaId { get; set; }
+        public string EndpointName { get; set; }
+        public string SagaType { get; set; }
+        public DateTime ChangedAt { get; set; }
+        public DateTime ReceivedAt { get; set; }
+
+        public static string MakeDocumentId(Guid sagaId) => $"CompletedSagas/{sagaId}";
+    }
+}