Add Raven5 persistence & acceptance tests

.github/workflows/ci.yml

@@ -15,7 +15,7 @@ jobs:
     strategy:
       matrix:
         os: [ windows-2022 ]
-        test-category: [ Default, SqlServer, AzureServiceBus, RabbitMQ, AzureStorageQueues, MSMQ, SQS, Raven35Acceptance ]
+        test-category: [ Default, SqlServer, AzureServiceBus, RabbitMQ, AzureStorageQueues, MSMQ, SQS, Raven35Acceptance, Raven5Acceptance, Raven5Persistence ]
         include:
           - os: windows-2022
             os-name: Windows

src/ServiceControl.AcceptanceTesting/InfrastructureConfig/ConfigureEndpointLearningTransport.cs

@@ -1,5 +1,6 @@
 namespace ServiceControl.AcceptanceTesting.InfrastructureConfig
 {
+    using System;
     using System.IO;
     using System.Threading.Tasks;
     using NServiceBus;
@@ -11,14 +12,14 @@ public class ConfigureEndpointLearningTransport : ITransportIntegration
     {
         public ConfigureEndpointLearningTransport()
         {
-            var relativePath = Path.Combine(TestContext.CurrentContext.TestDirectory, @"..", "..", "..", ".transport");
-            ConnectionString = Path.GetFullPath(relativePath);
+            ConnectionString = Path.Combine(Path.GetTempPath(), "ServiceControlTests", "TestTransport", TestContext.CurrentContext.Test.ID);
         }
 
         public string ConnectionString { get; set; }
 
         public Task Configure(string endpointName, EndpointConfiguration configuration, RunSettings settings, PublisherMetadata publisherMetadata)
         {
+            Console.WriteLine($"Transport Directory: {ConnectionString}");
             Directory.CreateDirectory(ConnectionString);
 
             var transportConfig = configuration.UseTransport<LearningTransport>();

src/ServiceControl.AcceptanceTesting/PortPool.cs

@@ -0,0 +1,69 @@
+namespace ServiceControl.AcceptanceTesting
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Linq;
+    using NUnit.Framework;
+    using TestHelper;
+
+    public class PortPool
+    {
+        readonly int startingPort;
+        readonly HashSet<int> inUse;
+
+        public PortPool(int startingPort)
+        {
+            this.startingPort = startingPort;
+            inUse = new HashSet<int>();
+        }
+
+        public PortLease GetLease() => new PortLease(this);
+
+        internal int LeasePort()
+        {
+            lock (inUse)
+            {
+                var start = inUse.Any() ? inUse.Max() + 1 : startingPort;
+                var port = PortUtility.FindAvailablePort(start);
+                inUse.Add(port);
+                TestContext.Out.WriteLine($"Port leased: {port}");
+                return port;
+            }
+        }
+
+        internal void Return(int port)
+        {
+            lock (inUse)
+            {
+                inUse.Remove(port);
+            }
+        }
+    }
+
+    public class PortLease : IDisposable
+    {
+        readonly PortPool owner;
+        readonly List<int> leasedPorts;
+
+        internal PortLease(PortPool owner)
+        {
+            this.owner = owner;
+            leasedPorts = new List<int>();
+        }
+
+        public int GetPort()
+        {
+            var port = owner.LeasePort();
+            leasedPorts.Add(port);
+            return port;
+        }
+
+        public void Dispose()
+        {
+            foreach (var port in leasedPorts)
+            {
+                owner.Return(port);
+            }
+        }
+    }
+}

src/ServiceControl.AcceptanceTesting/ScenarioWithEndpointBehaviorExtensions.cs

@@ -53,7 +53,24 @@ public SequenceBuilder<TContext> Do(string step, Func<TContext, Task> handler)
         public IScenarioWithEndpointBehavior<TContext> Done(Func<TContext, bool> doneCriteria = null)
         {
             var behavior = new ServiceControlClient<TContext>(context => sequence.Continue(context));
-            return endpointBehavior.WithComponent(behavior).Done(ctx => sequence.IsFinished(ctx) && (doneCriteria == null || doneCriteria(ctx)));
+            return endpointBehavior.WithComponent(behavior).Done(async ctx =>
+            {
+                if (sequence.IsFinished(ctx))
+                {
+                    if (doneCriteria == null || doneCriteria(ctx))
+                    {
+                        return true;
+                    }
+                    else
+                    {
+                        // If sequence is done but test is not finished, small delay to avoid tight loop check
+                        await Task.Delay(250);
+                    }
+                }
+
+                // If sequence is not finished immediately return false, since each step will enforce delays 
+                return false;
+            });
         }
 
         IScenarioWithEndpointBehavior<TContext> endpointBehavior;

src/ServiceControl.AcceptanceTesting/Sequence.cs

@@ -56,6 +56,10 @@ public async Task<bool> Continue(TContext context)
                 context.Step = nextStep;
                 return finished;
             }
+            else
+            {
+                await Task.Delay(250);
+            }
 
             return false;
         }

src/ServiceControl.AcceptanceTesting/ServiceControl.AcceptanceTesting.csproj

@@ -7,6 +7,7 @@
   <ItemGroup>
     <ProjectReference Include="..\ServiceControl.Transports.Learning\ServiceControl.Transports.Learning.csproj" />
     <ProjectReference Include="..\ServiceControlInstaller.Engine\ServiceControlInstaller.Engine.csproj" />
+    <ProjectReference Include="..\TestHelper\TestHelper.csproj" />
   </ItemGroup>
 
   <ItemGroup>

src/ServiceControl.AcceptanceTests.RavenDB/AcceptanceTestStorageConfiguration.cs

@@ -2,22 +2,26 @@
 {
     using System;
     using System.IO;
-    using System.Linq;
-    using System.Net.NetworkInformation;
     using System.Threading.Tasks;
     using Persistence.RavenDb;
     using ServiceBus.Management.Infrastructure.Settings;
+    using ServiceControl.AcceptanceTesting;
 
     class AcceptanceTestStorageConfiguration
     {
+        static readonly PortPool portPool = new PortPool(33334);
+        readonly PortLease portLease = portPool.GetLease();
+
         public string PersistenceType { get; protected set; }
 
         public void CustomizeSettings(Settings settings)
         {
+            settings.Port = portLease.GetPort();
+
             settings.PersisterSpecificSettings = new RavenDBPersisterSettings
             {
                 RunInMemory = true,
-                DatabaseMaintenancePort = FindAvailablePort(settings.Port + 1),
+                DatabaseMaintenancePort = portLease.GetPort(),
                 DatabasePath = Path.Combine(Path.GetTempPath(), Path.GetRandomFileName()),
                 ErrorRetentionPeriod = TimeSpan.FromDays(10),
             };
@@ -30,24 +34,10 @@ public Task Configure()
             return Task.CompletedTask;
         }
 
-        static int FindAvailablePort(int startPort)
+        public Task Cleanup()
         {
-            var activeTcpListeners = IPGlobalProperties
-                .GetIPGlobalProperties()
-                .GetActiveTcpListeners();
-
-            for (var port = startPort; port < startPort + 1024; port++)
-            {
-                var portCopy = port;
-                if (activeTcpListeners.All(endPoint => endPoint.Port != portCopy))
-                {
-                    return port;
-                }
-            }
-
-            return startPort;
+            portLease.Dispose();
+            return Task.CompletedTask;
         }
-
-        public Task Cleanup() => Task.CompletedTask;
     }
 }

src/ServiceControl.AcceptanceTests.RavenDB5/.gitignore

@@ -0,0 +1 @@
+.transport

src/ServiceControl.AcceptanceTests.RavenDB5/AcceptanceTestStorageConfiguration.cs

@@ -0,0 +1,36 @@
+namespace ServiceControl.AcceptanceTests
+{
+    using System.Threading.Tasks;
+    using Persistence.RavenDb;
+    using ServiceBus.Management.Infrastructure.Settings;
+    using ServiceControl.AcceptanceTesting;
+
+    class AcceptanceTestStorageConfiguration
+    {
+        static readonly PortPool portPool = new PortPool(33334);
+
+        readonly DatabaseLease databaseLease = SharedDatabaseSetup.LeaseDatabase();
+        readonly PortLease portLease = portPool.GetLease();
+
+        public string PersistenceType { get; protected set; }
+
+        public void CustomizeSettings(Settings settings)
+        {
+            databaseLease.CustomizeSettings(settings);
+            settings.Port = portLease.GetPort();
+        }
+
+        public Task Configure()
+        {
+            PersistenceType = typeof(RavenDbPersistenceConfiguration).AssemblyQualifiedName;
+
+            return Task.CompletedTask;
+        }
+
+        public async ValueTask Cleanup()
+        {
+            portLease.Dispose();
+            await databaseLease.DisposeAsync();
+        }
+    }
+}

src/ServiceControl.AcceptanceTests.RavenDB5/App.config

@@ -0,0 +1,9 @@
+<?xml version="1.0" encoding="utf-8"?>
+<configuration>
+  <appSettings>
+    <add key="Raven/Esent/MaxVerPages" value="4096" />
+  </appSettings>
+  <runtime>
+    <gcServer enabled="true" />
+  </runtime>
+</configuration>

src/ServiceControl.AcceptanceTests.RavenDB5/DatabaseLease.cs

@@ -0,0 +1,28 @@
+using System;
+using System.Threading.Tasks;
+using Raven.Client.ServerWide.Operations;
+using ServiceBus.Management.Infrastructure.Settings;
+
+public class DatabaseLease : IAsyncDisposable
+{
+    public string DatabaseName { get; } = Guid.NewGuid().ToString("n");
+
+    public void CustomizeSettings(Settings settings)
+    {
+        settings.PersisterSpecificSettings = new RavenDBPersisterSettings
+        {
+            ErrorRetentionPeriod = TimeSpan.FromDays(10),
+            ConnectionString = SharedDatabaseSetup.SharedInstance.ServerUrl,
+            DatabaseName = DatabaseName
+        };
+    }
+
+    public async ValueTask DisposeAsync()
+    {
+        var documentStore = await SharedDatabaseSetup.SharedInstance.Connect();
+
+        // Comment this out temporarily to be able to inspect a database after the test has completed
+        var deleteDatabasesOperation = new DeleteDatabasesOperation(new DeleteDatabasesOperation.Parameters { DatabaseNames = new[] { DatabaseName }, HardDelete = true });
+        await documentStore.Maintenance.Server.SendAsync(deleteDatabasesOperation);
+    }
+}

src/ServiceControl.AcceptanceTests.RavenDB5/Recoverability/MessageFailures/FailedErrorsController.cs

@@ -0,0 +1,75 @@
+namespace ServiceControl.AcceptanceTests.RavenDB.Recoverability.MessageFailures
+{
+    using System;
+    using System.Net;
+    using System.Net.Http;
+    using System.Threading;
+    using System.Threading.Tasks;
+    using System.Web.Http;
+    using Infrastructure.WebApi;
+    using Operations;
+    using Raven.Client.Documents;
+    using Raven.Client.Documents.Operations;
+
+    public class FailedErrorsCountReponse
+    {
+        public int Count { get; set; }
+    }
+
+    class FailedErrorsController : ApiController
+    {
+        public FailedErrorsController(IDocumentStore store, ImportFailedErrors importFailedErrors)
+        {
+            this.store = store;
+            this.importFailedErrors = importFailedErrors;
+        }
+
+        [Route("failederrors/count")]
+        [HttpGet]
+        public async Task<HttpResponseMessage> GetFailedErrorsCount()
+        {
+            using (var session = store.OpenAsyncSession())
+            {
+                var query =
+                    session.Query<FailedErrorImport, FailedErrorImportIndex>().Statistics(out var stats);
+
+                var count = await query.CountAsync();
+
+                return Request.CreateResponse(HttpStatusCode.OK, new FailedErrorsCountReponse
+                {
+                    Count = count
+                })
+                    .WithEtag(stats.ResultEtag.ToString());
+            }
+        }
+
+        [Route("failederrors/import")]
+        [HttpPost]
+        public async Task<HttpResponseMessage> ImportFailedErrors(CancellationToken cancellationToken = default)
+        {
+            var tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
+            await importFailedErrors.Run(tokenSource.Token);
+            return Request.CreateResponse(HttpStatusCode.OK);
+        }
+
+        [Route("failederrors/forcecleanerrors")]
+        [HttpPost]
+        public Task<HttpResponseMessage> ForceErrorMessageCleanerRun()
+        {
+            // TODO: Is there a way to force the Raven5 expiration to happen? Or does it just happen? Won't be able to tell until we redesign that.
+
+            // May not even need WaitForIndexes given Raven5 implementation isn't index-based
+            WaitForIndexes(store);
+
+            return Task.FromResult(Request.CreateResponse(HttpStatusCode.OK));
+        }
+
+        static void WaitForIndexes(IDocumentStore store)
+        {
+            SpinWait.SpinUntil(() => store.Maintenance.Send(new GetStatisticsOperation()).StaleIndexes.Length == 0, TimeSpan.FromSeconds(10));
+        }
+
+        readonly IDocumentStore store;
+        readonly ImportFailedErrors importFailedErrors;
+    }
+}