PostreSQL Transport

.github/workflows/ci.yml

@@ -56,5 +56,65 @@ jobs:
           sqlcmd -Q "CREATE SCHEMA receiver AUTHORIZATION db_owner" -d "nservicebus"
           sqlcmd -Q "CREATE SCHEMA sender AUTHORIZATION db_owner" -d "nservicebus"
           sqlcmd -Q "CREATE SCHEMA db@ AUTHORIZATION db_owner" -d "nservicebus"
+      - name: Azure login
+        uses: azure/login@v2.1.0
+        with:
+          creds: ${{ secrets.AZURE_ACI_CREDENTIALS }}    
+      - name: Setup PostgreSql 15.x
+        id: setup-postgresql
+        run: |
+          echo "Getting the Azure region in which this workflow is running..."
+          $hostInfo = curl -H Metadata:true --noproxy "*" "http://169.254.169.254/metadata/instance?api-version=2021-02-01" | ConvertFrom-Json
+          $region = $hostInfo.compute.location
+          echo "Actions agent running in Azure region: $region"
+
+          $name = "postgres-$(Get-Random)"
+          $packageTag = "Package=SqlPersistence"
+          $runnerOsTag = "RunnerOS=$($Env:RUNNER_OS)"
+          $dateTag = "Created=$(Get-Date -Format "yyyy-MM-dd")"
+
+          $password = [guid]::NewGuid().ToString("n")
+          echo "::add-mask::$password"
+
+          echo "Creating PostgreSQL container $name (This can take a while)"
+          $details = az container create --image postgres:15 --name $name --location $region --dns-name-label $name --resource-group GitHubActions-RG --cpu 2 --memory 8 --ports 5432 --ip-address public --environment-variables POSTGRES_PASSWORD="$password" --command-line "docker-entrypoint.sh postgres --max-prepared-transactions=10" | ConvertFrom-Json
+          echo "name=$name" | Out-File -FilePath $Env:GITHUB_OUTPUT -Encoding utf-8 -Append
+
+          echo "Tagging container"
+          $ignore = az tag create --resource-id $details.id --tags $packageTag $runnerOsTag $dateTag
+
+          $fqdn = $details.ipAddress.fqdn
+          echo "::add-mask::$fqdn"
+
+          $connectionString = "User ID=postgres;Password=$password;Host=$fqdn;Port=5432;Database=postgres;"
+          echo "PostgreSqlTransportConnectionString=$connectionString" | Out-File -FilePath $Env:GITHUB_ENV -Encoding utf-8 -Append
+
+          $env:PGPASSWORD = $password
+
+          # psql not in PATH on Windows
+          if ($Env:RUNNER_OS -eq 'Windows') {
+            $Env:PATH = $Env:PATH + ';' + $Env:PGBIN
+          }
+
+          for ($i = 0; $i -lt 24; $i++) { ## 2 minute timeout
+            echo "Checking for PostgreSQL connectivity $($i+1)/30..."
+            $ignore = psql --host $fqdn --username=postgres --list
+            if ($?) {
+              echo "Connection successful"
+              break;
+            }
+            sleep 5
+          }
+
+          # create schemas
+          $ignore = psql --host $fqdn --username=postgres --dbname=postgres --command 'CREATE SCHEMA IF NOT EXISTS sender'
+          $ignore = psql --host $fqdn --username=postgres --dbname=postgres --command 'CREATE SCHEMA IF NOT EXISTS receiver'
+          $ignore = psql --host $fqdn --username=postgres --dbname=postgres --command 'CREATE SCHEMA IF NOT EXISTS "db@"'
       - name: Run tests
         uses: Particular/run-tests-action@v1.7.0
+      - name: Teardown infrastructure
+        if: ${{ always() }}
+        run: |
+          echo "Removing container ${{ steps.setup-postgresql.outputs.name }}"
+          $ignore = az container delete --resource-group GitHubActions-RG --name ${{ steps.setup-postgresql.outputs.name }} --yes
+          echo "Removal complete"

src/NServiceBus.SqlServer.sln

@@ -22,6 +22,16 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NServiceBus.Transport.SqlSe
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NServiceBus.Transport.SqlServer.UnitTests", "NServiceBus.Transport.SqlServer.UnitTests\NServiceBus.Transport.SqlServer.UnitTests.csproj", "{F08C8BAB-C329-4F96-A6AF-F25419C623FD}"
 EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NServiceBus.Transport.PostgreSql", "NServiceBus.Transport.PostgreSql\NServiceBus.Transport.PostgreSql.csproj", "{926EFB58-F51F-4BF8-9625-A11F53950FA4}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NServiceBus.Transport.PostgreSql.TransportTests", "NServiceBus.Transport.PostgreSql.TransportTests\NServiceBus.Transport.PostgreSql.TransportTests.csproj", "{76B19C3B-7AB6-4BB7-AF39-03F76701521F}"
+EndProject
+Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "NServiceBus.Transport.Sql.Shared", "NServiceBus.Transport.Sql.Shared\NServiceBus.Transport.Sql.Shared.csproj", "{8028DDF3-7565-49B8-B71C-F321A1755C51}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NServiceBus.Transport.PostgreSql.AcceptanceTests", "NServiceBus.Transport.PostgreSql.AcceptanceTests\NServiceBus.Transport.PostgreSql.AcceptanceTests.csproj", "{F0B7EDA3-AB83-4FE7-986F-EE2D17484B52}"
+EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "NServiceBus.Transport.PostgreSql.UnitTests", "NServiceBus.Transport.PostgreSql.UnitTests\NServiceBus.Transport.PostgreSql.UnitTests.csproj", "{738DD684-77B8-4A6E-AE73-ED98023E5957}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -48,6 +58,26 @@ Global
 		{F08C8BAB-C329-4F96-A6AF-F25419C623FD}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{F08C8BAB-C329-4F96-A6AF-F25419C623FD}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{F08C8BAB-C329-4F96-A6AF-F25419C623FD}.Release|Any CPU.Build.0 = Release|Any CPU
+		{926EFB58-F51F-4BF8-9625-A11F53950FA4}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{926EFB58-F51F-4BF8-9625-A11F53950FA4}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{926EFB58-F51F-4BF8-9625-A11F53950FA4}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{926EFB58-F51F-4BF8-9625-A11F53950FA4}.Release|Any CPU.Build.0 = Release|Any CPU
+		{76B19C3B-7AB6-4BB7-AF39-03F76701521F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{76B19C3B-7AB6-4BB7-AF39-03F76701521F}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{76B19C3B-7AB6-4BB7-AF39-03F76701521F}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{76B19C3B-7AB6-4BB7-AF39-03F76701521F}.Release|Any CPU.Build.0 = Release|Any CPU
+		{8028DDF3-7565-49B8-B71C-F321A1755C51}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{8028DDF3-7565-49B8-B71C-F321A1755C51}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{8028DDF3-7565-49B8-B71C-F321A1755C51}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{8028DDF3-7565-49B8-B71C-F321A1755C51}.Release|Any CPU.Build.0 = Release|Any CPU
+		{F0B7EDA3-AB83-4FE7-986F-EE2D17484B52}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{F0B7EDA3-AB83-4FE7-986F-EE2D17484B52}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{F0B7EDA3-AB83-4FE7-986F-EE2D17484B52}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{F0B7EDA3-AB83-4FE7-986F-EE2D17484B52}.Release|Any CPU.Build.0 = Release|Any CPU
+		{738DD684-77B8-4A6E-AE73-ED98023E5957}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{738DD684-77B8-4A6E-AE73-ED98023E5957}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{738DD684-77B8-4A6E-AE73-ED98023E5957}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{738DD684-77B8-4A6E-AE73-ED98023E5957}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

src/NServiceBus.Transport.PostgreSql.AcceptanceTests/.editorconfig

@@ -0,0 +1,10 @@
+[*.cs]
+
+# Justification: Test project
+dotnet_diagnostic.CA2007.severity = none
+
+# may be enabled in future
+dotnet_diagnostic.PS0018.severity = none  # A task-returning method should have a CancellationToken parameter unless it has a parameter implementing ICancellableContext
+
+# Justification: Tests don't support cancellation and don't need to forward IMessageHandlerContext.CancellationToken
+dotnet_diagnostic.NSB0002.severity = suggestion

src/NServiceBus.Transport.PostgreSql.AcceptanceTests/ConfigurationHelpers.cs

@@ -0,0 +1,22 @@
+using NServiceBus;
+using NServiceBus.Configuration.AdvancedExtensibility;
+using NServiceBus.Transport;
+using NServiceBus.Transport.PostgreSql;
+
+static class ConfigurationHelpers
+{
+    public static PostgreSqlTransport ConfigureSqlServerTransport(this EndpointConfiguration configuration)
+    {
+        return (PostgreSqlTransport)configuration.GetSettings().Get<TransportDefinition>();
+    }
+
+    public static string BuildAddressWithSchema(string endpointName, string schema)
+    {
+        return $"{schema}.{endpointName}";
+    }
+
+    public static string QuoteSchema(string schema)
+    {
+        return $"\"{schema}\"";
+    }
+}

src/NServiceBus.Transport.PostgreSql.AcceptanceTests/ConfigureEndpointPostgreSqlTransport.cs

@@ -0,0 +1,117 @@
+using System;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Npgsql;
+using NServiceBus;
+using NServiceBus.AcceptanceTesting.Support;
+using NServiceBus.Transport.PostgreSql;
+using NServiceBus.Transport.SqlServer;
+
+public class ConfigureEndpointPostgreSqlTransport : IConfigureEndpointTestExecution
+{
+    public ConfigureEndpointPostgreSqlTransport(PostgreSqlTransport transport)
+    {
+        this.transport = transport;
+    }
+
+    public ConfigureEndpointPostgreSqlTransport()
+    {
+        connectionString = Environment.GetEnvironmentVariable("PostgreSqlTransportConnectionString") ?? @"User ID=user;Password=admin;Host=localhost;Port=54320;Database=nservicebus;Pooling=true;Connection Lifetime=0;Include Error Detail=true";
+
+        transport = new PostgreSqlTransport(connectionString);
+        transport.Subscriptions.DisableCaching = true;
+
+        //On non windows operating systems we need to explicitly set the transaction mode to SendsAtomicWithReceive since distributed transactions is not available there
+        if (!OperatingSystem.IsWindows() && transport.TransportTransactionMode == TransportTransactionMode.TransactionScope)
+        {
+            transport.TransportTransactionMode = TransportTransactionMode.SendsAtomicWithReceive;
+        }
+    }
+
+    public Task Configure(string endpointName, EndpointConfiguration configuration, RunSettings runSettings, PublisherMetadata publisherMetadata)
+    {
+        doNotCleanNativeSubscriptions = runSettings.TryGet<bool>("DoNotCleanNativeSubscriptions", out _);
+
+        configuration.UseTransport(transport);
+        this.endpointName = endpointName;
+
+        return Task.FromResult(0);
+    }
+
+    public async Task Cleanup()
+    {
+        Func<Task<NpgsqlConnection>> factory = async () =>
+        {
+            if (transport.ConnectionString != null)
+            {
+                var connection = new NpgsqlConnection(transport.ConnectionString);
+                await connection.OpenAsync().ConfigureAwait(false);
+                return connection;
+            }
+
+            return await transport.ConnectionFactory(CancellationToken.None).ConfigureAwait(false);
+        };
+
+        using (var conn = await factory().ConfigureAwait(false))
+        {
+            var queueAddresses = transport.Testing.ReceiveAddresses;
+            var delayedQueueAddress = transport.Testing.DelayedDeliveryQueue;
+            var defaultSchema = transport.DefaultSchema;
+
+            var commandTextBuilder = new StringBuilder();
+
+            //No clean-up for send-only endpoints
+            if (queueAddresses != null)
+            {
+                foreach (var address in queueAddresses)
+                {
+                    commandTextBuilder.AppendLine($"DROP TABLE IF EXISTS {address};");
+                }
+            }
+
+            //Null-check because if an exception is thrown before startup these fields might be empty
+            if (delayedQueueAddress != null)
+            {
+                commandTextBuilder.AppendLine($"DROP TABLE IF EXISTS {delayedQueueAddress};");
+            }
+
+            var subscriptionTableName = transport.Testing.SubscriptionTable;
+
+            if (!doNotCleanNativeSubscriptions && !string.IsNullOrEmpty(subscriptionTableName))
+            {
+                commandTextBuilder.AppendLine($"DROP TABLE IF EXISTS {subscriptionTableName};");
+            }
+
+            var commandText = commandTextBuilder.ToString();
+            if (!string.IsNullOrEmpty(commandText))
+            {
+                await TryDeleteTables(conn, commandText);
+            }
+        }
+    }
+
+    async Task TryDeleteTables(NpgsqlConnection conn, string commandText)
+    {
+        try
+        {
+            using (var comm = conn.CreateCommand())
+            {
+                comm.CommandText = commandText;
+                await comm.ExecuteNonQueryAsync().ConfigureAwait(false);
+            }
+        }
+        catch (Exception e)
+        {
+            if (!e.Message.Contains("it does not exist or you do not have permission"))
+            {
+                throw new Exception($"Failed to execute query in {endpointName}: {commandText}", e);
+            }
+        }
+    }
+
+    bool doNotCleanNativeSubscriptions;
+    PostgreSqlTransport transport;
+    string connectionString;
+    string endpointName;
+}

src/NServiceBus.Transport.PostgreSql.AcceptanceTests/CustomizedServer.cs

@@ -0,0 +1,13 @@
+namespace NServiceBus.Transport.PostgreSql.AcceptanceTests
+{
+    using NServiceBus.AcceptanceTests.EndpointTemplates;
+    using PostgreSql;
+
+    public class CustomizedServer : DefaultServer
+    {
+        public CustomizedServer(PostgreSqlTransport transport)
+        {
+            TransportConfiguration = new ConfigureEndpointPostgreSqlTransport(transport);
+        }
+    }
+}

src/NServiceBus.Transport.PostgreSql.AcceptanceTests/MultiSchema/When_custom_schema_configured_for_error_queue.cs

@@ -0,0 +1,92 @@
+namespace NServiceBus.Transport.SqlServer.AcceptanceTests.MultiSchema
+{
+    using System;
+    using System.Threading.Tasks;
+    using AcceptanceTesting;
+    using NServiceBus.AcceptanceTests;
+    using NServiceBus.AcceptanceTests.EndpointTemplates;
+    using NUnit.Framework;
+
+    public class When_custom_schema_configured_for_error_queue : NServiceBusAcceptanceTest
+    {
+        [Test]
+        public async Task Error_should_be_sent_to_table_in_configured_schema()
+        {
+            var ctx = await Scenario.Define<Context>()
+                .WithEndpoint<Sender>(b =>
+                {
+                    b.DoNotFailOnErrorMessages();
+                    b.When((bus, c) => bus.SendLocal(new Message()));
+                })
+                .WithEndpoint<ErrorSpy>()
+                .Done(c => c.FailedMessageProcessed)
+                .Run();
+
+            Assert.True(ctx.FailedMessageProcessed, "Message should be moved to error queue in custom schema");
+        }
+
+        public class Context : ScenarioContext
+        {
+            public bool FailedMessageProcessed { get; set; }
+        }
+
+        public class Sender : EndpointConfigurationBuilder
+        {
+            public Sender()
+            {
+                EndpointSetup<DefaultServer>(c =>
+                {
+                    var errorSpyName = AcceptanceTesting.Customization.Conventions.EndpointNamingConvention(typeof(ErrorSpy));
+
+                    c.SendFailedMessagesTo(errorSpyName);
+
+                    c.Recoverability()
+                        .Immediate(i => i.NumberOfRetries(0))
+                        .Delayed(d => d.NumberOfRetries(0));
+
+                    var transport = c.ConfigureSqlServerTransport();
+                    transport.Schema.UseSchemaForQueue(errorSpyName, ErrorSpySchema);
+                });
+            }
+
+            class Handler : IHandleMessages<Message>
+            {
+                public Task Handle(Message message, IMessageHandlerContext context)
+                {
+                    throw new Exception("Simulated exception");
+                }
+            }
+        }
+
+        public class ErrorSpy : EndpointConfigurationBuilder
+        {
+            public ErrorSpy()
+            {
+                EndpointSetup<DefaultServer>(c =>
+                {
+                    c.ConfigureSqlServerTransport().DefaultSchema = ErrorSpySchema;
+                });
+            }
+
+            class Handler : IHandleMessages<Message>
+            {
+                readonly Context scenarioContext;
+                public Handler(Context scenarioContext)
+                {
+                    this.scenarioContext = scenarioContext;
+                }
+
+                public Task Handle(Message message, IMessageHandlerContext context)
+                {
+                    scenarioContext.FailedMessageProcessed = true;
+
+                    return Task.FromResult(0);
+                }
+            }
+        }
+
+        public class Message : ICommand { }
+
+        const string ErrorSpySchema = "receiver";
+    }
+}