diff --git a/.gitignore b/.gitignore index c51fff37..8e138844 100644 --- a/.gitignore +++ b/.gitignore @@ -24,4 +24,7 @@ dist_package/ .claude data/ -CLAUDE.md \ No newline at end of file +CLAUDE.md + +# graft's local graph cache — regenerable, not committed (run `graft build`). +/graft/ diff --git a/.ignore b/.ignore new file mode 100644 index 00000000..615e581e --- /dev/null +++ b/.ignore @@ -0,0 +1,5 @@ +# graft's cards are gitignored but should stay greppable: ripgrep reads +# .ignore before .gitignore, so this re-admits the tree to search only. +!graft/ +graft/.cache/ +graft/.graph/ diff --git a/.mcp.json b/.mcp.json new file mode 100644 index 00000000..fcedf6ac --- /dev/null +++ b/.mcp.json @@ -0,0 +1,10 @@ +{ + "mcpServers": { + "graft": { + "command": "graft", + "args": [ + "mcp" + ] + } + } +} diff --git a/src/Nimbus.Tests.Integration/Tests/NatsTransportTests/WhenAnIdleSubscriptionExpires.cs b/src/Nimbus.Tests.Integration/Tests/NatsTransportTests/WhenAnIdleSubscriptionExpires.cs new file mode 100644 index 00000000..f649e470 --- /dev/null +++ b/src/Nimbus.Tests.Integration/Tests/NatsTransportTests/WhenAnIdleSubscriptionExpires.cs @@ -0,0 +1,174 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading.Tasks; +using Nimbus.Configuration.Settings; +using Nimbus.Infrastructure; +using Nimbus.Infrastructure.Logging; +using Nimbus.InfrastructureContracts; +using Nimbus.Infrastructure.MessageSendersAndReceivers; +using Nimbus.Serializers.Json; +using Nimbus.Transports.Nats; +using Nimbus.Transports.Nats.ConnectionManagement; +using Nimbus.Transports.Nats.MessageSendersAndReceivers; +using Nimbus.Transports.Nats.QueueManagement; +using NUnit.Framework; +using Shouldly; + +namespace Nimbus.Tests.Integration.Tests.NatsTransportTests +{ + /// + /// White-box tests against a live NATS server with JetStream. Set NIMBUS_TEST_NATS to a URL + /// (e.g. nats://localhost:4222) to run them. + /// + [TestFixture] + public class WhenAnIdleSubscriptionExpires + { + private NatsConnectionFactory _connectionFactory; + private NatsJetStreamContextFactory _jsContextFactory; + private AutoDeleteOnIdleSetting _autoDeleteOnIdle; + private NullLogger _logger; + private NatsSubscription _subscription; + + [SetUp] + public void SetUp() + { + var url = Environment.GetEnvironmentVariable("NIMBUS_TEST_NATS"); + if (string.IsNullOrWhiteSpace(url)) Assert.Ignore("Set NIMBUS_TEST_NATS to run NATS tests."); + + _logger = new NullLogger(); + var configuration = new NatsTransportConfiguration().WithUrl(url).WithJetStream(); + _connectionFactory = new NatsConnectionFactory(configuration, _logger); + _jsContextFactory = new NatsJetStreamContextFactory(_connectionFactory); + _autoDeleteOnIdle = new AutoDeleteOnIdleSetting {Value = TimeSpan.FromMinutes(7)}; + _subscription = new NatsSubscription($"t.idle.{Guid.NewGuid():N}", $"sub.{Guid.NewGuid():N}"); + } + + [TearDown] + public async Task TearDown() + { + if (_jsContextFactory != null) + { + foreach (var stream in await _jsContextFactory.ListStreamsAsync()) + { + if (stream.Config.Name != null && stream.Config.Subjects!.Any(s => s.StartsWith(_subscription.TopicPath, StringComparison.Ordinal))) + await _jsContextFactory.DeleteStreamAsync(stream.Config.Name); + } + } + + _connectionFactory?.Dispose(); + } + + [Test] + public async Task TopicConsumersAreCreatedWithAnInactiveThreshold() + { + var receiver = new NatsJetStreamTopicReceiver(_subscription, + _jsContextFactory, + new JsonSerializer(), + _autoDeleteOnIdle, + new ConcurrentHandlerLimitSetting(), + new GlobalHandlerThrottle(new GlobalConcurrentHandlerLimitSetting()), + _logger); + await receiver.Start(msg => Task.CompletedTask); + try + { + var consumers = new List<(string Stream, TimeSpan Threshold)>(); + foreach (var stream in await _jsContextFactory.ListStreamsAsync()) + { + if (!stream.Config.Subjects!.Any(s => s.StartsWith(_subscription.TopicPath, StringComparison.Ordinal))) continue; + var js = await _jsContextFactory.GetContextAsync(); + await foreach (var consumer in js.ListConsumersAsync(stream.Config.Name!)) + consumers.Add((stream.Config.Name!, consumer.Info.Config.InactiveThreshold)); + } + + consumers.Count.ShouldBe(2); // main + retry + consumers.ShouldAllBe(c => c.Threshold == _autoDeleteOnIdle.Value); + } + finally + { + receiver.Dispose(); + } + } + + [Test] + public async Task TheTopicStreamKeepsMessagesNoLongerThanTheIdleWindowWhetherTheSenderOrReceiverCreatedIt() + { + var sender = new NatsJetStreamTopicSender(_subscription.TopicPath, _jsContextFactory, new JsonSerializer(), _autoDeleteOnIdle); + await sender.Send(new NimbusMessage(_subscription.TopicPath)); + (await TopicStreamMaxAge()).ShouldBe(_autoDeleteOnIdle.Value); + + // A second process (fresh stream cache) with a receiver must agree, or the stream config would flip-flop. + var receiverFactory = new NatsJetStreamContextFactory(_connectionFactory); + var receiver = new NatsJetStreamTopicReceiver(_subscription, + receiverFactory, + new JsonSerializer(), + _autoDeleteOnIdle, + new ConcurrentHandlerLimitSetting(), + new GlobalHandlerThrottle(new GlobalConcurrentHandlerLimitSetting()), + _logger); + await receiver.Start(msg => Task.CompletedTask); + try + { + (await TopicStreamMaxAge()).ShouldBe(_autoDeleteOnIdle.Value); + } + finally + { + receiver.Dispose(); + } + } + + [Test] + public async Task TheReaperDeletesAConsumerlessRetryStreamOnlyAfterTwoSweeps() + { + var retrySubject = $"{_subscription.TopicPath}.sub.retry"; + var streamName = $"retry_{Guid.NewGuid():N}"; + await _jsContextFactory.EnsureStreamAsync(streamName, retrySubject); + var reaper = new NatsJetStreamRetryStreamReaper(_jsContextFactory, _autoDeleteOnIdle, _logger); + + await reaper.ReapOnce(); + (await StreamExists(streamName)).ShouldBeTrue(); + + await reaper.ReapOnce(); + (await StreamExists(streamName)).ShouldBeFalse(); + } + + [Test] + public async Task TheReaperKeepsRetryStreamsThatStillHaveAConsumer() + { + var retrySubject = $"{_subscription.TopicPath}.sub.retry"; + var streamName = $"retry_{Guid.NewGuid():N}"; + await _jsContextFactory.EnsureStreamAsync(streamName, retrySubject); + await _jsContextFactory.EnsureConsumerAsync(streamName, new NATS.Client.JetStream.Models.ConsumerConfig {Name = "c", DurableName = "c", FilterSubject = retrySubject}); + var reaper = new NatsJetStreamRetryStreamReaper(_jsContextFactory, _autoDeleteOnIdle, _logger); + + await reaper.ReapOnce(); + await reaper.ReapOnce(); + + (await StreamExists(streamName)).ShouldBeTrue(); + } + + [Test] + public async Task TheReaperIgnoresStreamsThatAreNotRetryStreams() + { + var streamName = $"plain_{Guid.NewGuid():N}"; + await _jsContextFactory.EnsureStreamAsync(streamName, _subscription.TopicPath); + var reaper = new NatsJetStreamRetryStreamReaper(_jsContextFactory, _autoDeleteOnIdle, _logger); + + await reaper.ReapOnce(); + await reaper.ReapOnce(); + + (await StreamExists(streamName)).ShouldBeTrue(); + } + + private async Task TopicStreamMaxAge() + { + var stream = (await _jsContextFactory.ListStreamsAsync()).Single(s => s.Config.Subjects!.Contains(_subscription.TopicPath)); + return stream.Config.MaxAge; + } + + private async Task StreamExists(string name) + { + return (await _jsContextFactory.ListStreamsAsync()).Any(s => s.Config.Name == name); + } + } +} diff --git a/src/Nimbus.Tests.Integration/Tests/PostgresTransportTests/WhenAnIdleSubscriptionExpires.cs b/src/Nimbus.Tests.Integration/Tests/PostgresTransportTests/WhenAnIdleSubscriptionExpires.cs new file mode 100644 index 00000000..f4dc021c --- /dev/null +++ b/src/Nimbus.Tests.Integration/Tests/PostgresTransportTests/WhenAnIdleSubscriptionExpires.cs @@ -0,0 +1,174 @@ +using System; +using System.Threading.Tasks; +using Nimbus.Configuration.Settings; +using Nimbus.Infrastructure; +using Nimbus.Infrastructure.Logging; +using Nimbus.Infrastructure.MessageSendersAndReceivers; +using Nimbus.InfrastructureContracts; +using Nimbus.Serializers.Json; +using Nimbus.Transports.Postgres; +using Nimbus.Transports.Postgres.MessageSendersAndReceivers; +using Nimbus.Transports.Postgres.QueueManagement; +using Nimbus.Transports.Postgres.Schema; +using Npgsql; +using NUnit.Framework; +using Shouldly; + +namespace Nimbus.Tests.Integration.Tests.PostgresTransportTests +{ + /// + /// White-box tests against a live PostgreSQL. Set NIMBUS_TEST_POSTGRES to a connection string to run them. + /// Staleness is simulated by back-dating last_seen_at rather than waiting out the 5 minute minimum TTL. + /// + [TestFixture] + public class WhenAnIdleSubscriptionExpires + { + private PostgresTransportConfiguration _configuration; + private PostgresSubscription _subscription; + private AutoDeleteOnIdleSetting _autoDeleteOnIdle; + private NullLogger _logger; + private JsonSerializer _serializer; + + [SetUp] + public async Task SetUp() + { + var connectionString = Environment.GetEnvironmentVariable("NIMBUS_TEST_POSTGRES"); + if (string.IsNullOrWhiteSpace(connectionString)) Assert.Ignore("Set NIMBUS_TEST_POSTGRES to run PostgreSQL tests."); + + _logger = new NullLogger(); + _configuration = new PostgresTransportConfiguration().WithConnectionString(connectionString).WithAutoCreateSchema(); + await new PostgresSchemaCreator(_configuration, _logger).EnsureSchemaExists(); + + _subscription = new PostgresSubscription($"t.Idle.{Guid.NewGuid():N}", $"sub.{Guid.NewGuid():N}"); + _autoDeleteOnIdle = new AutoDeleteOnIdleSetting {Value = TimeSpan.FromMinutes(5)}; + _serializer = new JsonSerializer(); + } + + [TearDown] + public async Task TearDown() + { + if (_subscription == null) return; + await Execute("DELETE FROM nimbus_subscriptions WHERE topic_name = @t; DELETE FROM nimbus_messages WHERE destination = @q", + ("@t", _subscription.TopicPath), ("@q", _subscription.SubscriberQueueName)); + } + + [Test] + public async Task WarmUpRegistersTheSubscriptionWithAFreshHeartbeat() + { + var receiver = CreateReceiver(); + await receiver.Start(msg => Task.CompletedTask); + try + { + (await SubscriptionAgeSeconds()).ShouldNotBeNull().ShouldBeLessThan(30); + } + finally + { + receiver.Dispose(); + } + } + + [Test] + public async Task ReRegisteringARegisteredSubscriptionRefreshesItsHeartbeat() + { + await RegisterSubscription(); + await BackdateSubscription(TimeSpan.FromHours(1)); + + await RegisterSubscription(); + + (await SubscriptionAgeSeconds()).ShouldNotBeNull().ShouldBeLessThan(30); + } + + [Test] + public async Task TheSenderSkipsSubscriptionsWithALapsedHeartbeat() + { + await RegisterSubscription(); + var sender = new PostgresTopicSender(_subscription.TopicPath, _configuration, _serializer, _autoDeleteOnIdle); + + await sender.Send(new NimbusMessage(_subscription.TopicPath)); + (await QueuedMessages()).ShouldBe(1); + + await BackdateSubscription(TimeSpan.FromHours(1)); + await sender.Send(new NimbusMessage(_subscription.TopicPath)); + (await QueuedMessages()).ShouldBe(1); + } + + [Test] + public async Task TheReaperRemovesStaleSubscriptionsAndTheirMessagesButKeepsLiveOnes() + { + var live = new PostgresSubscription(_subscription.TopicPath, $"live.{Guid.NewGuid():N}"); + await RegisterSubscription(); + await RegisterSubscription(live); + var sender = new PostgresTopicSender(_subscription.TopicPath, _configuration, _serializer, _autoDeleteOnIdle); + await sender.Send(new NimbusMessage(_subscription.TopicPath)); + await BackdateSubscription(TimeSpan.FromHours(1)); + + await new PostgresIdleSubscriptionReaper(_configuration, _autoDeleteOnIdle, _logger).ReapOnce(); + + (await SubscriptionAgeSeconds()).ShouldBeNull(); + (await QueuedMessages()).ShouldBe(0); + (await SubscriptionAgeSeconds(live)).ShouldNotBeNull(); + (await QueuedMessages(live)).ShouldBe(1); + + await Execute("DELETE FROM nimbus_subscriptions WHERE subscriber_queue = @q; DELETE FROM nimbus_messages WHERE destination = @q", + ("@q", live.SubscriberQueueName)); + } + + private PostgresTopicReceiver CreateReceiver() + { + return new PostgresTopicReceiver(_subscription, + _configuration, + _autoDeleteOnIdle, + _serializer, + new ConcurrentHandlerLimitSetting(), + new GlobalHandlerThrottle(new GlobalConcurrentHandlerLimitSetting()), + _logger); + } + + private Task RegisterSubscription(PostgresSubscription subscription = null) + { + subscription ??= _subscription; + return Execute(@"INSERT INTO nimbus_subscriptions (topic_name, subscriber_queue, last_seen_at) VALUES (@t, @q, now()) + ON CONFLICT (topic_name, subscriber_queue) DO UPDATE SET last_seen_at = now()", + ("@t", subscription.TopicPath), ("@q", subscription.SubscriberQueueName)); + } + + private Task BackdateSubscription(TimeSpan age) + { + return Execute("UPDATE nimbus_subscriptions SET last_seen_at = now() - make_interval(secs => @s) WHERE subscriber_queue = @q", + ("@s", age.TotalSeconds), ("@q", _subscription.SubscriberQueueName)); + } + + private async Task SubscriptionAgeSeconds(PostgresSubscription subscription = null) + { + subscription ??= _subscription; + var result = await Scalar("SELECT extract(epoch FROM now() - last_seen_at) FROM nimbus_subscriptions WHERE subscriber_queue = @q", + ("@q", subscription.SubscriberQueueName)); + return result == null ? (double?) null : Convert.ToDouble(result); + } + + private async Task QueuedMessages(PostgresSubscription subscription = null) + { + subscription ??= _subscription; + return Convert.ToInt64(await Scalar("SELECT count(*) FROM nimbus_messages WHERE destination = @q", ("@q", subscription.SubscriberQueueName))); + } + + private async Task Execute(string sql, params (string Name, object Value)[] parameters) + { + using var connection = new NpgsqlConnection(_configuration.ConnectionString); + await connection.OpenAsync(); + using var command = new NpgsqlCommand(sql, connection); + foreach (var (name, value) in parameters) command.Parameters.AddWithValue(name, value); + await command.ExecuteNonQueryAsync(); + } + + private async Task Scalar(string sql, params (string Name, object Value)[] parameters) + { + using var connection = new NpgsqlConnection(_configuration.ConnectionString); + await connection.OpenAsync(); + using var command = new NpgsqlCommand(sql, connection); + foreach (var (name, value) in parameters) command.Parameters.AddWithValue(name, value); + var result = await command.ExecuteScalarAsync(); + return result is DBNull ? null : result; + } + } +} diff --git a/src/Nimbus.Tests.Integration/Tests/RabbitMqTransportTests/WhenAnIdleSubscriptionExpires.cs b/src/Nimbus.Tests.Integration/Tests/RabbitMqTransportTests/WhenAnIdleSubscriptionExpires.cs new file mode 100644 index 00000000..7c31d239 --- /dev/null +++ b/src/Nimbus.Tests.Integration/Tests/RabbitMqTransportTests/WhenAnIdleSubscriptionExpires.cs @@ -0,0 +1,137 @@ +using System; +using System.Threading.Tasks; +using Nimbus.Configuration.Settings; +using Nimbus.Infrastructure; +using Nimbus.Infrastructure.Compression; +using Nimbus.Infrastructure.Logging; +using Nimbus.Infrastructure.MessageSendersAndReceivers; +using Nimbus.Serializers.Json; +using Nimbus.Transports.RabbitMQ; +using Nimbus.Transports.RabbitMQ.ConnectionManagement; +using Nimbus.Transports.RabbitMQ.MessageConversion; +using Nimbus.Transports.RabbitMQ.MessageSendersAndReceivers; +using NUnit.Framework; +using RabbitMQ.Client; +using RabbitMQ.Client.Exceptions; +using Shouldly; + +namespace Nimbus.Tests.Integration.Tests.RabbitMqTransportTests +{ + /// + /// White-box tests against a live RabbitMQ (with the delayed message exchange plugin) or LavinMQ. + /// Set NIMBUS_TEST_RABBITMQ to "host:port" (e.g. localhost:5674) to run them. The broker's default + /// user is "guest"/"guest"; set NIMBUS_TEST_RABBITMQ_CREDENTIALS to "user:password" to override. + /// + [TestFixture] + public class WhenAnIdleSubscriptionExpires + { + private RabbitMqTransportConfiguration _configuration; + private RabbitMqConnectionManager _connectionManager; + private RabbitMqSubscription _subscription; + private NullLogger _logger; + + [SetUp] + public void SetUp() + { + var endpoint = Environment.GetEnvironmentVariable("NIMBUS_TEST_RABBITMQ"); + if (string.IsNullOrWhiteSpace(endpoint)) Assert.Ignore("Set NIMBUS_TEST_RABBITMQ to run RabbitMQ tests."); + + var parts = endpoint.Split(':'); + var credentials = (Environment.GetEnvironmentVariable("NIMBUS_TEST_RABBITMQ_CREDENTIALS") ?? "guest:guest").Split(':'); + _logger = new NullLogger(); + _configuration = new RabbitMqTransportConfiguration().WithHost(parts[0]).WithPort(int.Parse(parts[1])).WithCredentials(credentials[0], credentials[1]); + _connectionManager = new RabbitMqConnectionManager(_configuration, _logger); + _subscription = new RabbitMqSubscription($"t.idle.{Guid.NewGuid():N}", $"sub.{Guid.NewGuid():N}"); + } + + [TearDown] + public async Task TearDown() + { + if (_connectionManager == null) return; + + var channel = await _connectionManager.CreateChannelAsync(); + try + { + await channel.QueueDeleteAsync(_subscription.SubscriptionName); + await channel.ExchangeDeleteAsync(_subscription.TopicPath); + } + catch { } + await _connectionManager.DisposeAsync(); + } + + [Test] + public async Task AnAbandonedSubscriptionQueueIsDeletedByTheBroker() + { + var receiver = CreateReceiver(TimeSpan.FromSeconds(2)); + await receiver.Start(msg => Task.CompletedTask); + (await QueueExists()).ShouldBeTrue(); + + receiver.Dispose(); + + // Don't poll: redeclaring a queue (even passively) counts as use and renews its expiry. + await Task.Delay(TimeSpan.FromSeconds(10)); + (await QueueExists()).ShouldBeFalse(); + } + + [Test] + public async Task ASubscriptionQueueWithALiveConsumerIsNotDeleted() + { + var receiver = CreateReceiver(TimeSpan.FromSeconds(2)); + await receiver.Start(msg => Task.CompletedTask); + try + { + await Task.Delay(TimeSpan.FromSeconds(8)); + (await QueueExists()).ShouldBeTrue(); + } + finally + { + receiver.Dispose(); + } + } + + [Test] + public async Task AQueueDeclaredBeforeIdleCleanupExistedIsStillUsable() + { + var channel = await _connectionManager.CreateChannelAsync(); + await channel.QueueDeclareAsync(_subscription.SubscriptionName, durable: true, exclusive: false, autoDelete: false); + + var receiver = CreateReceiver(TimeSpan.FromMinutes(5)); + await receiver.Start(msg => Task.CompletedTask); + try + { + (await QueueExists()).ShouldBeTrue(); + } + finally + { + receiver.Dispose(); + } + } + + private RabbitMqTopicReceiver CreateReceiver(TimeSpan idle) + { + return new RabbitMqTopicReceiver(_subscription, + null, + _connectionManager, + new RabbitMqMessageConverter(new JsonSerializer(), new NullCompressor(), _logger), + new AutoDeleteOnIdleSetting {Value = idle}, + new ConcurrentHandlerLimitSetting(), + new GlobalHandlerThrottle(new GlobalConcurrentHandlerLimitSetting()), + _logger); + } + + private async Task QueueExists() + { + var channel = await _connectionManager.CreateChannelAsync(); + try + { + await channel.QueueDeclarePassiveAsync(_subscription.SubscriptionName); + await channel.CloseAsync(); + return true; + } + catch (OperationInterruptedException) + { + return false; + } + } + } +} diff --git a/src/Nimbus.Tests.Integration/Tests/SqlServerTransportTests/WhenAnIdleSubscriptionExpires.cs b/src/Nimbus.Tests.Integration/Tests/SqlServerTransportTests/WhenAnIdleSubscriptionExpires.cs new file mode 100644 index 00000000..7cb2403e --- /dev/null +++ b/src/Nimbus.Tests.Integration/Tests/SqlServerTransportTests/WhenAnIdleSubscriptionExpires.cs @@ -0,0 +1,177 @@ +using System; +using System.Threading.Tasks; +using Nimbus.Configuration.Settings; +using Nimbus.Infrastructure; +using Nimbus.Infrastructure.Logging; +using Nimbus.Infrastructure.MessageSendersAndReceivers; +using Nimbus.InfrastructureContracts; +using Nimbus.Serializers.Json; +using Nimbus.Transports.SqlServer; +using Nimbus.Transports.SqlServer.MessageSendersAndReceivers; +using Nimbus.Transports.SqlServer.QueueManagement; +using Nimbus.Transports.SqlServer.Schema; +using Microsoft.Data.SqlClient; +using NUnit.Framework; +using Shouldly; + +namespace Nimbus.Tests.Integration.Tests.SqlServerTransportTests +{ + /// + /// White-box tests against a live SQL Server. Set NIMBUS_TEST_SQLSERVER to a connection string to run them. + /// Staleness is simulated by back-dating LastSeenAt rather than waiting out the 5 minute minimum TTL. + /// + [TestFixture] + public class WhenAnIdleSubscriptionExpires + { + private SqlServerTransportConfiguration _configuration; + private SqlServerSubscription _subscription; + private AutoDeleteOnIdleSetting _autoDeleteOnIdle; + private NullLogger _logger; + private JsonSerializer _serializer; + + [SetUp] + public async Task SetUp() + { + var connectionString = Environment.GetEnvironmentVariable("NIMBUS_TEST_SQLSERVER"); + if (string.IsNullOrWhiteSpace(connectionString)) Assert.Ignore("Set NIMBUS_TEST_SQLSERVER to run SQL Server tests."); + + _logger = new NullLogger(); + _configuration = new SqlServerTransportConfiguration().WithConnectionString(connectionString).WithAutoCreateSchema(); + await new SqlServerSchemaCreator(_configuration, _logger).EnsureSchemaExists(); + + _subscription = new SqlServerSubscription($"t.Idle.{Guid.NewGuid():N}", $"sub.{Guid.NewGuid():N}"); + _autoDeleteOnIdle = new AutoDeleteOnIdleSetting {Value = TimeSpan.FromMinutes(5)}; + _serializer = new JsonSerializer(); + } + + [TearDown] + public async Task TearDown() + { + if (_subscription == null) return; + await Execute("DELETE FROM NimbusSubscriptions WHERE TopicName = @t; DELETE FROM NimbusMessages WHERE Destination = @q", + ("@t", _subscription.TopicPath), ("@q", _subscription.SubscriberQueueName)); + } + + [Test] + public async Task WarmUpRegistersTheSubscriptionWithAFreshHeartbeat() + { + var receiver = CreateReceiver(); + await receiver.Start(msg => Task.CompletedTask); + try + { + (await SubscriptionAgeSeconds()).ShouldNotBeNull().ShouldBeLessThan(30); + } + finally + { + receiver.Dispose(); + } + } + + [Test] + public async Task ReRegisteringARegisteredSubscriptionRefreshesItsHeartbeat() + { + await RegisterSubscription(); + await BackdateSubscription(TimeSpan.FromHours(1)); + + await RegisterSubscription(); + + (await SubscriptionAgeSeconds()).ShouldNotBeNull().ShouldBeLessThan(30); + } + + [Test] + public async Task TheSenderSkipsSubscriptionsWithALapsedHeartbeat() + { + await RegisterSubscription(); + var sender = new SqlServerTopicSender(_subscription.TopicPath, _configuration, _serializer, _autoDeleteOnIdle); + + await sender.Send(new NimbusMessage(_subscription.TopicPath)); + (await QueuedMessages()).ShouldBe(1); + + await BackdateSubscription(TimeSpan.FromHours(1)); + await sender.Send(new NimbusMessage(_subscription.TopicPath)); + (await QueuedMessages()).ShouldBe(1); + } + + [Test] + public async Task TheReaperRemovesStaleSubscriptionsAndTheirMessagesButKeepsLiveOnes() + { + var live = new SqlServerSubscription(_subscription.TopicPath, $"live.{Guid.NewGuid():N}"); + await RegisterSubscription(); + await RegisterSubscription(live); + var sender = new SqlServerTopicSender(_subscription.TopicPath, _configuration, _serializer, _autoDeleteOnIdle); + await sender.Send(new NimbusMessage(_subscription.TopicPath)); + await BackdateSubscription(TimeSpan.FromHours(1)); + + await new SqlServerIdleSubscriptionReaper(_configuration, _autoDeleteOnIdle, _logger).ReapOnce(); + + (await SubscriptionAgeSeconds()).ShouldBeNull(); + (await QueuedMessages()).ShouldBe(0); + (await SubscriptionAgeSeconds(live)).ShouldNotBeNull(); + (await QueuedMessages(live)).ShouldBe(1); + + await Execute("DELETE FROM NimbusSubscriptions WHERE SubscriberQueue = @q; DELETE FROM NimbusMessages WHERE Destination = @q", + ("@q", live.SubscriberQueueName)); + } + + private SqlServerTopicReceiver CreateReceiver() + { + return new SqlServerTopicReceiver(_subscription, + _configuration, + _autoDeleteOnIdle, + _serializer, + new ConcurrentHandlerLimitSetting(), + new GlobalHandlerThrottle(new GlobalConcurrentHandlerLimitSetting()), + _logger); + } + + private Task RegisterSubscription(SqlServerSubscription subscription = null) + { + subscription ??= _subscription; + return Execute(@"MERGE NimbusSubscriptions WITH (HOLDLOCK) AS target + USING (SELECT @t AS TopicName, @q AS SubscriberQueue) AS source + ON target.TopicName = source.TopicName AND target.SubscriberQueue = source.SubscriberQueue + WHEN MATCHED THEN UPDATE SET LastSeenAt = SYSUTCDATETIME() + WHEN NOT MATCHED THEN INSERT (TopicName, SubscriberQueue, LastSeenAt) VALUES (source.TopicName, source.SubscriberQueue, SYSUTCDATETIME());", + ("@t", subscription.TopicPath), ("@q", subscription.SubscriberQueueName)); + } + + private Task BackdateSubscription(TimeSpan age) + { + return Execute("UPDATE NimbusSubscriptions SET LastSeenAt = DATEADD(SECOND, -@s, SYSUTCDATETIME()) WHERE SubscriberQueue = @q", + ("@s", (int) age.TotalSeconds), ("@q", _subscription.SubscriberQueueName)); + } + + private async Task SubscriptionAgeSeconds(SqlServerSubscription subscription = null) + { + subscription ??= _subscription; + var result = await Scalar("SELECT DATEDIFF(SECOND, LastSeenAt, SYSUTCDATETIME()) FROM NimbusSubscriptions WHERE SubscriberQueue = @q", + ("@q", subscription.SubscriberQueueName)); + return result == null ? (double?) null : Convert.ToDouble(result); + } + + private async Task QueuedMessages(SqlServerSubscription subscription = null) + { + subscription ??= _subscription; + return Convert.ToInt64(await Scalar("SELECT COUNT(*) FROM NimbusMessages WHERE Destination = @q", ("@q", subscription.SubscriberQueueName))); + } + + private async Task Execute(string sql, params (string Name, object Value)[] parameters) + { + using var connection = new SqlConnection(_configuration.ConnectionString); + await connection.OpenAsync(); + using var command = new SqlCommand(sql, connection); + foreach (var (name, value) in parameters) command.Parameters.AddWithValue(name, value); + await command.ExecuteNonQueryAsync(); + } + + private async Task Scalar(string sql, params (string Name, object Value)[] parameters) + { + using var connection = new SqlConnection(_configuration.ConnectionString); + await connection.OpenAsync(); + using var command = new SqlCommand(sql, connection); + foreach (var (name, value) in parameters) command.Parameters.AddWithValue(name, value); + var result = await command.ExecuteScalarAsync(); + return result is DBNull ? null : result; + } + } +} diff --git a/src/Nimbus.Transports.Nats/ConnectionManagement/NatsJetStreamContextFactory.cs b/src/Nimbus.Transports.Nats/ConnectionManagement/NatsJetStreamContextFactory.cs index 761b1577..0f1225cb 100644 --- a/src/Nimbus.Transports.Nats/ConnectionManagement/NatsJetStreamContextFactory.cs +++ b/src/Nimbus.Transports.Nats/ConnectionManagement/NatsJetStreamContextFactory.cs @@ -43,7 +43,7 @@ public async Task PublishAsync(string subject, byte[] data, NatsHeaders? headers await ctx.PublishAsync(subject, data, headers: headers, cancellationToken: ct); } - public async Task EnsureStreamAsync(string streamName, string subject, StreamConfigRetention retention = StreamConfigRetention.Limits, CancellationToken ct = default) + public async Task EnsureStreamAsync(string streamName, string subject, StreamConfigRetention retention = StreamConfigRetention.Limits, TimeSpan? maxAge = null, CancellationToken ct = default) { if (_ensuredStreams.ContainsKey(streamName)) return; var streamLock = _streamLocks.GetOrAdd(streamName, _ => new SemaphoreSlim(1, 1)); @@ -58,6 +58,7 @@ await ctx.CreateOrUpdateStreamAsync(new StreamConfig Subjects = [subject, subject + ".sched"], AllowMsgSchedules = true, Retention = retention, + MaxAge = maxAge ?? default, }, ct); _ensuredStreams.TryAdd(streamName, 0); } @@ -108,6 +109,22 @@ public async Task GetStreamAsync(string streamName, CancellationT return await ctx.GetStreamAsync(streamName, cancellationToken: ct); } + public async Task DeleteStreamAsync(string streamName, CancellationToken ct = default) + { + var ctx = await GetContextAsync(ct); + await ctx.DeleteStreamAsync(streamName, ct); + _ensuredStreams.TryRemove(streamName, out _); + } + + public async Task> ListStreamsAsync(CancellationToken ct = default) + { + var ctx = await GetContextAsync(ct); + var streams = new List(); + await foreach (var stream in ctx.ListStreamsAsync(cancellationToken: ct)) + streams.Add(stream.Info); + return streams; + } + public async Task DeleteAllStreamsAsync(CancellationToken ct = default) { var ctx = await GetContextAsync(ct); diff --git a/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamMessageReceiver.cs b/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamMessageReceiver.cs index aae55e24..61c03f22 100644 --- a/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamMessageReceiver.cs +++ b/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamMessageReceiver.cs @@ -26,6 +26,12 @@ internal abstract class NatsJetStreamMessageReceiver : ThrottlingMessageReceiver protected abstract string ConsumerName { get; } protected abstract StreamConfigRetention StreamRetention { get; } + // When set, the server deletes the consumer once nothing has pulled from it for this long. + protected virtual TimeSpan? ConsumerInactiveThreshold => null; + + // When set, the server discards messages older than this from the stream. + protected virtual TimeSpan? StreamMaxAge => null; + protected NatsJetStreamMessageReceiver( NatsJetStreamContextFactory jsContextFactory, ISerializer serializer, @@ -46,7 +52,7 @@ protected override async Task WarmUp() // SingleWriter = false: multiple consumer loops (main + retry) write concurrently. _channel = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleWriter = false }); - await _jsContextFactory.EnsureStreamAsync(StreamName, Subject, StreamRetention); + await _jsContextFactory.EnsureStreamAsync(StreamName, Subject, StreamRetention, StreamMaxAge); _mainConsumer = await _jsContextFactory.EnsureConsumerAsync(StreamName, new ConsumerConfig { Name = ConsumerName, @@ -54,6 +60,7 @@ protected override async Task WarmUp() FilterSubject = Subject, AckPolicy = ConsumerConfigAckPolicy.Explicit, DeliverPolicy = ConsumerConfigDeliverPolicy.All, + InactiveThreshold = ConsumerInactiveThreshold ?? default, }); await OnWarmingUp(); diff --git a/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamTopicReceiver.cs b/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamTopicReceiver.cs index 4323015c..cae5fe56 100644 --- a/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamTopicReceiver.cs +++ b/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamTopicReceiver.cs @@ -9,12 +9,20 @@ namespace Nimbus.Transports.Nats.MessageSendersAndReceivers internal class NatsJetStreamTopicReceiver : NatsJetStreamMessageReceiver { private readonly NatsSubscription _subscription; + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; protected override string StreamName { get; } protected override string Subject => _subscription.TopicPath; protected override string ConsumerName { get; } protected override StreamConfigRetention StreamRetention => StreamConfigRetention.Limits; + // Per-instance multicast subscriptions are orphaned by every restart; let the server drop consumers that stop pulling. + protected override TimeSpan? ConsumerInactiveThreshold => _autoDeleteOnIdle.Value; + + // Events are never scheduled ahead (retries use their own stream), so the topic stream only needs to + // keep messages as long as a subscriber could still come back for them. Must match NatsJetStreamTopicSender. + protected override TimeSpan? StreamMaxAge => _autoDeleteOnIdle.Value; + // Dedicated per-subscription subject for retries so that a failed handler // requeues only to its own subscription, not fan-out to all subscribers. private string RetrySubject => $"{_subscription.TopicPath}.{SanitiseName(_subscription.SubscriptionName)}.retry"; @@ -24,12 +32,14 @@ internal class NatsJetStreamTopicReceiver : NatsJetStreamMessageReceiver public NatsJetStreamTopicReceiver(NatsSubscription subscription, NatsJetStreamContextFactory jsContextFactory, ISerializer serializer, + AutoDeleteOnIdleSetting autoDeleteOnIdle, ConcurrentHandlerLimitSetting concurrentHandlerLimit, IGlobalHandlerThrottle globalHandlerThrottle, ILogger logger) : base(jsContextFactory, serializer, concurrentHandlerLimit, globalHandlerThrottle, logger) { _subscription = subscription; + _autoDeleteOnIdle = autoDeleteOnIdle; StreamName = SanitiseName(subscription.TopicPath); ConsumerName = SanitiseName(subscription.SubscriptionName); } @@ -44,6 +54,7 @@ protected override async Task OnWarmingUp() FilterSubject = RetrySubject, AckPolicy = ConsumerConfigAckPolicy.Explicit, DeliverPolicy = ConsumerConfigDeliverPolicy.All, + InactiveThreshold = _autoDeleteOnIdle.Value, }); RegisterAdditionalConsumer(retryConsumer); } diff --git a/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamTopicSender.cs b/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamTopicSender.cs index 77a985bb..0d94b9a1 100644 --- a/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamTopicSender.cs +++ b/src/Nimbus.Transports.Nats/MessageSendersAndReceivers/NatsJetStreamTopicSender.cs @@ -1,4 +1,5 @@ using System.Text; +using Nimbus.Configuration.Settings; using Nimbus.Infrastructure.MessageSendersAndReceivers; using Nimbus.InfrastructureContracts; using Nimbus.Transports.Nats.ConnectionManagement; @@ -11,20 +12,23 @@ internal class NatsJetStreamTopicSender : INimbusMessageSender private readonly string _streamName; private readonly NatsJetStreamContextFactory _jsContextFactory; private readonly ISerializer _serializer; + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; public NatsJetStreamTopicSender(string topicPath, NatsJetStreamContextFactory jsContextFactory, - ISerializer serializer) + ISerializer serializer, + AutoDeleteOnIdleSetting autoDeleteOnIdle) { _topicPath = topicPath; _streamName = SanitiseName(topicPath); _jsContextFactory = jsContextFactory; _serializer = serializer; + _autoDeleteOnIdle = autoDeleteOnIdle; } public async Task Send(NimbusMessage message) { - await _jsContextFactory.EnsureStreamAsync(_streamName, _topicPath); + await _jsContextFactory.EnsureStreamAsync(_streamName, _topicPath, maxAge: _autoDeleteOnIdle.Value); var bytes = Encoding.UTF8.GetBytes(_serializer.Serialize(message)); await _jsContextFactory.PublishAsync(_topicPath, bytes); } diff --git a/src/Nimbus.Transports.Nats/NatsTransport.cs b/src/Nimbus.Transports.Nats/NatsTransport.cs index 66f82203..77805fa3 100644 --- a/src/Nimbus.Transports.Nats/NatsTransport.cs +++ b/src/Nimbus.Transports.Nats/NatsTransport.cs @@ -4,14 +4,16 @@ using Nimbus.InfrastructureContracts.Filtering.Conditions; using Nimbus.Transports.Nats.ConnectionManagement; using Nimbus.Transports.Nats.MessageSendersAndReceivers; +using Nimbus.Transports.Nats.QueueManagement; namespace Nimbus.Transports.Nats { - internal class NatsTransport : INimbusTransport + internal class NatsTransport : INimbusTransport, IDisposable { private readonly PoorMansIoC _container; private readonly NatsConnectionFactory _connectionFactory; private readonly bool _isJetStream; + private NatsJetStreamRetryStreamReaper? _retryStreamReaper; public NatsTransport(PoorMansIoC container, NatsConnectionFactory connectionFactory, NatsTransportConfiguration config) { @@ -20,9 +22,20 @@ public NatsTransport(PoorMansIoC container, NatsConnectionFactory connectionFact _isJetStream = config.IsJetStream; } - public Task TestConnection() + public async Task TestConnection() { - return _connectionFactory.TestConnection(); + await _connectionFactory.TestConnection(); + + if (_isJetStream) + { + _retryStreamReaper = _container.Resolve(); + _retryStreamReaper.Start(); + } + } + + public void Dispose() + { + _retryStreamReaper?.Dispose(); } public INimbusMessageSender GetQueueSender(string queuePath) diff --git a/src/Nimbus.Transports.Nats/NatsTransportConfiguration.cs b/src/Nimbus.Transports.Nats/NatsTransportConfiguration.cs index c026999f..3e7206ab 100644 --- a/src/Nimbus.Transports.Nats/NatsTransportConfiguration.cs +++ b/src/Nimbus.Transports.Nats/NatsTransportConfiguration.cs @@ -66,6 +66,7 @@ protected override void RegisterComponents(PoorMansIoC container) if (IsJetStream) { container.RegisterType(ComponentLifetime.SingleInstance); + container.RegisterType(ComponentLifetime.SingleInstance); container.RegisterType(ComponentLifetime.InstancePerDependency); container.RegisterType(ComponentLifetime.InstancePerDependency); container.RegisterType(ComponentLifetime.InstancePerDependency); diff --git a/src/Nimbus.Transports.Nats/Nimbus.Transports.Nats.csproj b/src/Nimbus.Transports.Nats/Nimbus.Transports.Nats.csproj index d94c98c8..9df92cf7 100644 --- a/src/Nimbus.Transports.Nats/Nimbus.Transports.Nats.csproj +++ b/src/Nimbus.Transports.Nats/Nimbus.Transports.Nats.csproj @@ -1,6 +1,6 @@ - 1.0.0 + 1.1.0 net9.0 enable enable diff --git a/src/Nimbus.Transports.Nats/Properties/AssemblyVisibility.cs b/src/Nimbus.Transports.Nats/Properties/AssemblyVisibility.cs new file mode 100644 index 00000000..ffb1103e --- /dev/null +++ b/src/Nimbus.Transports.Nats/Properties/AssemblyVisibility.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Nimbus.Tests.Integration", AllInternalsVisible = true)] diff --git a/src/Nimbus.Transports.Nats/QueueManagement/NatsJetStreamRetryStreamReaper.cs b/src/Nimbus.Transports.Nats/QueueManagement/NatsJetStreamRetryStreamReaper.cs new file mode 100644 index 00000000..ca6cc702 --- /dev/null +++ b/src/Nimbus.Transports.Nats/QueueManagement/NatsJetStreamRetryStreamReaper.cs @@ -0,0 +1,77 @@ +using Nimbus.Configuration.Settings; +using Nimbus.Extensions; +using Nimbus.InfrastructureContracts; +using Nimbus.Transports.Nats.ConnectionManagement; +using Timer = System.Timers.Timer; + +namespace Nimbus.Transports.Nats.QueueManagement +{ + /// + /// Every topic subscription gets its own retry stream (see NatsJetStreamTopicReceiver). The + /// server drops idle consumers via InactiveThreshold, but streams have no idle expiry, so this + /// reaper deletes retry streams that have no consumers left. + /// + internal class NatsJetStreamRetryStreamReaper : IDisposable + { + private const string RetrySubjectSuffix = ".retry"; + + private readonly NatsJetStreamContextFactory _jsContextFactory; + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; + private readonly ILogger _logger; + private readonly HashSet _consumerlessLastSweep = new(); + private Timer? _timer; + + public NatsJetStreamRetryStreamReaper(NatsJetStreamContextFactory jsContextFactory, AutoDeleteOnIdleSetting autoDeleteOnIdle, ILogger logger) + { + _jsContextFactory = jsContextFactory; + _autoDeleteOnIdle = autoDeleteOnIdle; + _logger = logger; + } + + public void Start() + { + if (_timer != null) return; + + var interval = TimeSpan.FromTicks(_autoDeleteOnIdle.Value.Ticks / 4); + + _timer = new Timer(interval.TotalMilliseconds) { AutoReset = true }; + _timer.Elapsed += (_, _) => Task.Run(ReapOnce).ConfigureAwaitFalse(); + _timer.Start(); + } + + // internal so tests can trigger a sweep synchronously. + internal async Task ReapOnce() + { + try + { + var consumerless = (await _jsContextFactory.ListStreamsAsync()) + .Where(s => s.Config.Name != null) + .Where(s => s.Config.Subjects?.Any(subject => subject.EndsWith(RetrySubjectSuffix, StringComparison.Ordinal)) == true) + .Where(s => s.State.ConsumerCount == 0) + .Select(s => s.Config.Name!) + .ToHashSet(); + + // A stream is created a moment before its consumer, so only delete one that was + // already consumer-less on the previous sweep too. + foreach (var name in consumerless.Where(_consumerlessLastSweep.Contains)) + { + _logger.Info("Deleting orphaned retry stream {Stream}", name); + await _jsContextFactory.DeleteStreamAsync(name); + } + + _consumerlessLastSweep.Clear(); + foreach (var name in consumerless) _consumerlessLastSweep.Add(name); + } + catch (Exception exc) + { + _logger.Warn(exc, "Retry stream reaper sweep failed: {Message}", exc.Message); + } + } + + public void Dispose() + { + _timer?.Dispose(); + _timer = null; + } + } +} diff --git a/src/Nimbus.Transports.Postgres/MessageSendersAndReceivers/PostgresTopicReceiver.cs b/src/Nimbus.Transports.Postgres/MessageSendersAndReceivers/PostgresTopicReceiver.cs index dcfb79cd..29ec19da 100644 --- a/src/Nimbus.Transports.Postgres/MessageSendersAndReceivers/PostgresTopicReceiver.cs +++ b/src/Nimbus.Transports.Postgres/MessageSendersAndReceivers/PostgresTopicReceiver.cs @@ -1,3 +1,4 @@ +using System; using System.Threading; using System.Threading.Tasks; using Nimbus.Configuration.Settings; @@ -13,14 +14,20 @@ internal class PostgresTopicReceiver : PostgresQueueReceiver private readonly PostgresSubscription _subscription; private readonly PostgresTransportConfiguration _configuration; - // Idempotent subscription registration: safe to call on every bus start. + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; + private readonly ILogger _logger; + private DateTimeOffset _lastHeartbeat; + + // Idempotent subscription registration and heartbeat in one statement: safe to call on every bus start, + // and re-creates the row if the reaper removed it during a long outage. private const string RegisterSubscriptionSql = @" - INSERT INTO nimbus_subscriptions (topic_name, subscriber_queue) - VALUES (@topic_name, @subscriber_queue) - ON CONFLICT DO NOTHING"; + INSERT INTO nimbus_subscriptions (topic_name, subscriber_queue, last_seen_at) + VALUES (@topic_name, @subscriber_queue, now()) + ON CONFLICT (topic_name, subscriber_queue) DO UPDATE SET last_seen_at = now()"; public PostgresTopicReceiver(PostgresSubscription subscription, PostgresTransportConfiguration configuration, + AutoDeleteOnIdleSetting autoDeleteOnIdle, ISerializer serializer, ConcurrentHandlerLimitSetting concurrentHandlerLimit, IGlobalHandlerThrottle globalHandlerThrottle, @@ -29,9 +36,17 @@ public PostgresTopicReceiver(PostgresSubscription subscription, { _subscription = subscription; _configuration = configuration; + _autoDeleteOnIdle = autoDeleteOnIdle; + _logger = logger; } protected override async Task WarmUp() + { + await RegisterOrTouchSubscription(); + await base.WarmUp(); + } + + private async Task RegisterOrTouchSubscription() { using var connection = new NpgsqlConnection(_configuration.ConnectionString); await connection.OpenAsync(); @@ -41,8 +56,22 @@ protected override async Task WarmUp() command.Parameters.Add("@subscriber_queue", NpgsqlDbType.Text).Value = _subscription.SubscriberQueueName; await command.ExecuteNonQueryAsync(); + _lastHeartbeat = DateTimeOffset.UtcNow; + } - await base.WarmUp(); + // Refresh the heartbeat several times per idle window so one failed attempt doesn't get us reaped. + private async Task HeartbeatIfDue() + { + if (DateTimeOffset.UtcNow - _lastHeartbeat < TimeSpan.FromTicks(_autoDeleteOnIdle.Value.Ticks / 4)) return; + + try + { + await RegisterOrTouchSubscription(); + } + catch (Exception exc) + { + _logger.Warn(exc, "Failed to refresh heartbeat for subscription {SubscriberQueue}: {Message}", _subscription.SubscriberQueueName, exc.Message); + } } // Override Fetch to stamp the subscriber queue name on the message so that @@ -51,6 +80,8 @@ protected override async Task WarmUp() // fan-out to all subscribers again). protected override async Task Fetch(CancellationToken cancellationToken) { + await HeartbeatIfDue(); + var message = await base.Fetch(cancellationToken); if (message != null) message.Properties[MessagePropertyKeys.RedeliveryToSubscriptionName] = _subscription.SubscriberQueueName; diff --git a/src/Nimbus.Transports.Postgres/MessageSendersAndReceivers/PostgresTopicSender.cs b/src/Nimbus.Transports.Postgres/MessageSendersAndReceivers/PostgresTopicSender.cs index d0d063e6..a7e7434f 100644 --- a/src/Nimbus.Transports.Postgres/MessageSendersAndReceivers/PostgresTopicSender.cs +++ b/src/Nimbus.Transports.Postgres/MessageSendersAndReceivers/PostgresTopicSender.cs @@ -1,6 +1,7 @@ using System; using System.Text; using System.Threading.Tasks; +using Nimbus.Configuration.Settings; using Nimbus.Infrastructure.MessageSendersAndReceivers; using Nimbus.InfrastructureContracts; using Npgsql; @@ -13,17 +14,22 @@ internal class PostgresTopicSender : INimbusMessageSender private readonly string _topicPath; private readonly PostgresTransportConfiguration _configuration; private readonly ISerializer _serializer; + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; // Fan-out: one INSERT per subscriber, each with a fresh UUID, all in one atomic statement. // gen_random_uuid() is evaluated per row, so every subscriber gets a unique message_id. + // Subscriptions whose heartbeat has lapsed are skipped so we don't write messages nobody will read; + // PostgresIdleSubscriptionReaper removes them (and anything already queued for them) later. private const string FanOutSql = @" INSERT INTO nimbus_messages (message_id, destination, body, visible_after, expires_at) SELECT gen_random_uuid(), subscriber_queue, @body, @visible_after, @expires_at FROM nimbus_subscriptions - WHERE topic_name = @topic_name"; + WHERE topic_name = @topic_name + AND last_seen_at > now() - @idle_timeout"; - public PostgresTopicSender(string topicPath, PostgresTransportConfiguration configuration, ISerializer serializer) + public PostgresTopicSender(string topicPath, PostgresTransportConfiguration configuration, ISerializer serializer, AutoDeleteOnIdleSetting autoDeleteOnIdle) { + _autoDeleteOnIdle = autoDeleteOnIdle; _topicPath = topicPath; _configuration = configuration; _serializer = serializer; @@ -43,6 +49,7 @@ public async Task Send(NimbusMessage message) using var command = new NpgsqlCommand(FanOutSql, connection); command.Parameters.Add("@topic_name", NpgsqlDbType.Text).Value = _topicPath; + command.Parameters.Add("@idle_timeout", NpgsqlDbType.Interval).Value = _autoDeleteOnIdle.Value; command.Parameters.Add("@body", NpgsqlDbType.Bytea).Value = body; command.Parameters.Add("@visible_after", NpgsqlDbType.TimestampTz).Value = visibleAfter; command.Parameters.Add("@expires_at", NpgsqlDbType.TimestampTz).Value = (object) message.ExpiresAfter.UtcDateTime ?? DBNull.Value; diff --git a/src/Nimbus.Transports.Postgres/Nimbus.Transports.Postgres.csproj b/src/Nimbus.Transports.Postgres/Nimbus.Transports.Postgres.csproj index bb709cb7..324677be 100644 --- a/src/Nimbus.Transports.Postgres/Nimbus.Transports.Postgres.csproj +++ b/src/Nimbus.Transports.Postgres/Nimbus.Transports.Postgres.csproj @@ -10,7 +10,7 @@ - 1.0.0 + 1.1.0 true PostgreSQL transport for the Nimbus messaging framework using Npgsql. nimbus;messaging;transport;postgresql;postgres;npgsql diff --git a/src/Nimbus.Transports.Postgres/PostgresTransport.cs b/src/Nimbus.Transports.Postgres/PostgresTransport.cs index 3974cd06..ffa45d4c 100644 --- a/src/Nimbus.Transports.Postgres/PostgresTransport.cs +++ b/src/Nimbus.Transports.Postgres/PostgresTransport.cs @@ -1,22 +1,27 @@ +using System; using System.Threading.Tasks; using Nimbus.Configuration.PoorMansIocContainer; using Nimbus.Infrastructure; using Nimbus.Infrastructure.MessageSendersAndReceivers; using Nimbus.InfrastructureContracts.Filtering.Conditions; using Nimbus.Transports.Postgres.MessageSendersAndReceivers; +using Nimbus.Transports.Postgres.QueueManagement; using Nimbus.Transports.Postgres.Schema; using Npgsql; namespace Nimbus.Transports.Postgres { - internal class PostgresTransport : INimbusTransport + internal class PostgresTransport : INimbusTransport, IDisposable { private readonly PoorMansIoC _container; private readonly PostgresTransportConfiguration _configuration; private readonly PostgresSchemaCreator _schemaCreator; + private readonly PostgresIdleSubscriptionReaper _idleSubscriptionReaper; - public PostgresTransport(PoorMansIoC container, PostgresTransportConfiguration configuration, PostgresSchemaCreator schemaCreator) + public PostgresTransport(PoorMansIoC container, PostgresTransportConfiguration configuration, PostgresSchemaCreator schemaCreator, + PostgresIdleSubscriptionReaper idleSubscriptionReaper) { + _idleSubscriptionReaper = idleSubscriptionReaper; _container = container; _configuration = configuration; _schemaCreator = schemaCreator; @@ -29,6 +34,8 @@ public async Task TestConnection() if (_configuration.AutoCreateSchema) await _schemaCreator.EnsureSchemaExists(); + + _idleSubscriptionReaper.Start(); } public INimbusMessageSender GetQueueSender(string queuePath) @@ -51,5 +58,10 @@ public INimbusMessageReceiver GetTopicReceiver(string topicPath, string subscrip var subscription = new PostgresSubscription(topicPath, subscriptionName); return _container.ResolveWithOverrides(subscription); } + + public void Dispose() + { + _idleSubscriptionReaper.Dispose(); + } } } diff --git a/src/Nimbus.Transports.Postgres/PostgresTransportConfiguration.cs b/src/Nimbus.Transports.Postgres/PostgresTransportConfiguration.cs index 38c4b99a..0165645c 100644 --- a/src/Nimbus.Transports.Postgres/PostgresTransportConfiguration.cs +++ b/src/Nimbus.Transports.Postgres/PostgresTransportConfiguration.cs @@ -51,6 +51,7 @@ protected override void RegisterComponents(PoorMansIoC container) container.RegisterType(ComponentLifetime.SingleInstance, typeof(IDeadLetterOffice)); container.RegisterType(ComponentLifetime.SingleInstance, typeof(INamespaceCleanser)); container.RegisterType(ComponentLifetime.SingleInstance); + container.RegisterType(ComponentLifetime.SingleInstance); container.RegisterType(ComponentLifetime.SingleInstance, typeof(ILargeMessageBodyStore)); container.RegisterType(ComponentLifetime.SingleInstance, typeof(INimbusTransport)); diff --git a/src/Nimbus.Transports.Postgres/Properties/AssemblyVisibility.cs b/src/Nimbus.Transports.Postgres/Properties/AssemblyVisibility.cs new file mode 100644 index 00000000..ffb1103e --- /dev/null +++ b/src/Nimbus.Transports.Postgres/Properties/AssemblyVisibility.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Nimbus.Tests.Integration", AllInternalsVisible = true)] diff --git a/src/Nimbus.Transports.Postgres/QueueManagement/PostgresIdleSubscriptionReaper.cs b/src/Nimbus.Transports.Postgres/QueueManagement/PostgresIdleSubscriptionReaper.cs new file mode 100644 index 00000000..09c5438a --- /dev/null +++ b/src/Nimbus.Transports.Postgres/QueueManagement/PostgresIdleSubscriptionReaper.cs @@ -0,0 +1,77 @@ +using System; +using System.Threading.Tasks; +using Nimbus.Configuration.Settings; +using Nimbus.Extensions; +using Nimbus.InfrastructureContracts; +using Npgsql; +using NpgsqlTypes; +using Timer = System.Timers.Timer; + +namespace Nimbus.Transports.Postgres.QueueManagement +{ + /// + /// Removes subscriptions whose receivers have stopped sending heartbeats (see + /// PostgresTopicReceiver) along with any messages still queued for them. Every bus instance + /// runs a sweep; the statement is idempotent so concurrent sweeps are harmless. + /// + internal class PostgresIdleSubscriptionReaper : IDisposable + { + private readonly PostgresTransportConfiguration _configuration; + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; + private readonly ILogger _logger; + private Timer _timer; + + // Uses database time throughout so clock skew between application hosts doesn't matter. + private const string ReapSql = @" + WITH dead AS ( + DELETE FROM nimbus_subscriptions + WHERE last_seen_at < now() - @idle_timeout + RETURNING subscriber_queue + ) + DELETE FROM nimbus_messages + WHERE destination IN (SELECT subscriber_queue FROM dead)"; + + public PostgresIdleSubscriptionReaper(PostgresTransportConfiguration configuration, AutoDeleteOnIdleSetting autoDeleteOnIdle, ILogger logger) + { + _configuration = configuration; + _autoDeleteOnIdle = autoDeleteOnIdle; + _logger = logger; + } + + public void Start() + { + if (_timer != null) return; + + var interval = TimeSpan.FromTicks(_autoDeleteOnIdle.Value.Ticks / 4); + + _timer = new Timer(interval.TotalMilliseconds) {AutoReset = true}; + _timer.Elapsed += (s, e) => Task.Run(ReapOnce).ConfigureAwaitFalse(); + _timer.Start(); + } + + // internal so tests can trigger a sweep synchronously. + internal async Task ReapOnce() + { + try + { + using var connection = new NpgsqlConnection(_configuration.ConnectionString); + await connection.OpenAsync(); + + using var command = new NpgsqlCommand(ReapSql, connection); + command.Parameters.Add("@idle_timeout", NpgsqlDbType.Interval).Value = _autoDeleteOnIdle.Value; + + await command.ExecuteNonQueryAsync(); + } + catch (Exception exc) + { + _logger.Warn(exc, "Idle subscription reaper sweep failed: {Message}", exc.Message); + } + } + + public void Dispose() + { + _timer?.Dispose(); + _timer = null; + } + } +} diff --git a/src/Nimbus.Transports.Postgres/Schema/CreateSchema.sql b/src/Nimbus.Transports.Postgres/Schema/CreateSchema.sql index 92148c33..cd0a7735 100644 --- a/src/Nimbus.Transports.Postgres/Schema/CreateSchema.sql +++ b/src/Nimbus.Transports.Postgres/Schema/CreateSchema.sql @@ -16,9 +16,13 @@ CREATE INDEX IF NOT EXISTS ix_nimbus_messages_dequeue CREATE TABLE IF NOT EXISTS nimbus_subscriptions ( topic_name TEXT NOT NULL, subscriber_queue TEXT NOT NULL, + last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(), CONSTRAINT pk_nimbus_subscriptions PRIMARY KEY (topic_name, subscriber_queue) ); +-- Added for idle subscription cleanup; upgrades tables created by earlier versions. +ALTER TABLE nimbus_subscriptions ADD COLUMN IF NOT EXISTS last_seen_at TIMESTAMPTZ NOT NULL DEFAULT now(); + CREATE TABLE IF NOT EXISTS nimbus_dead_letters ( message_id UUID NOT NULL, original_destination TEXT NULL, diff --git a/src/Nimbus.Transports.RabbitMQ/MessageSendersAndReceivers/RabbitMqTopicReceiver.cs b/src/Nimbus.Transports.RabbitMQ/MessageSendersAndReceivers/RabbitMqTopicReceiver.cs index 74a29007..c4c5f9ee 100644 --- a/src/Nimbus.Transports.RabbitMQ/MessageSendersAndReceivers/RabbitMqTopicReceiver.cs +++ b/src/Nimbus.Transports.RabbitMQ/MessageSendersAndReceivers/RabbitMqTopicReceiver.cs @@ -1,3 +1,5 @@ +using System; +using System.Collections.Generic; using System.Threading.Tasks; using Nimbus.Configuration.Settings; using Nimbus.Infrastructure; @@ -6,6 +8,7 @@ using Nimbus.Transports.RabbitMQ.ConnectionManagement; using Nimbus.Transports.RabbitMQ.MessageConversion; using RabbitMQ.Client; +using RabbitMQ.Client.Exceptions; namespace Nimbus.Transports.RabbitMQ.MessageSendersAndReceivers { @@ -13,11 +16,13 @@ internal class RabbitMqTopicReceiver : RabbitMqReceiverBase { private readonly string _topicPath; private readonly string _subscriptionName; + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; public RabbitMqTopicReceiver(RabbitMqSubscription subscription, IFilterCondition filterCondition, RabbitMqConnectionManager connectionManager, RabbitMqMessageConverter messageConverter, + AutoDeleteOnIdleSetting autoDeleteOnIdle, ConcurrentHandlerLimitSetting concurrentHandlerLimit, IGlobalHandlerThrottle globalHandlerThrottle, ILogger logger) @@ -25,6 +30,7 @@ public RabbitMqTopicReceiver(RabbitMqSubscription subscription, { _topicPath = subscription.TopicPath; _subscriptionName = subscription.SubscriptionName; + _autoDeleteOnIdle = autoDeleteOnIdle; } protected override string ConsumeQueue => _subscriptionName; @@ -33,7 +39,7 @@ public RabbitMqTopicReceiver(RabbitMqSubscription subscription, protected override async Task DeclareTopologyAsync(IChannel channel) { await channel.ExchangeDeclareAsync(_topicPath, type: "fanout", durable: true, autoDelete: false); - await channel.QueueDeclareAsync(_subscriptionName, durable: true, exclusive: false, autoDelete: false); + channel = await DeclareSubscriptionQueueAsync(channel); // Immediate publishes: fanout exchange delivers to all subscriptions await channel.QueueBindAsync(_subscriptionName, _topicPath, routingKey: ""); // Targeted retries: delayed exchange routes to this specific subscription queue @@ -42,6 +48,31 @@ protected override async Task DeclareTopologyAsync(IChannel channel) await channel.QueueBindAsync(_subscriptionName, RabbitMqConnectionManager.DelayedExchangeName, routingKey: _topicPath); } + // x-expires makes the broker delete the queue (and its bindings) once it has had no consumers for the + // idle window, which cleans up the per-instance queues orphaned by every restart. + private async Task DeclareSubscriptionQueueAsync(IChannel channel) + { + var arguments = new Dictionary {{"x-expires", (int) _autoDeleteOnIdle.Value.TotalMilliseconds}}; + + try + { + await channel.QueueDeclareAsync(_subscriptionName, durable: true, exclusive: false, autoDelete: false, arguments: arguments); + return channel; + } + catch (OperationInterruptedException exc) when (exc.ShutdownReason?.ReplyCode == 406) + { + // The queue predates idle cleanup and was declared without x-expires. The broker closed the channel, + // so carry on over a new one and attach to the existing queue as-is. + _logger.Warn("Queue {Queue} already exists without x-expires, so it will not be removed when idle. Delete it, or apply an 'expires' policy, to enable idle cleanup.", _subscriptionName); + + try { channel.Dispose(); } catch { } + channel = await _connectionManager.CreateChannelAsync(); + _channel = channel; + await channel.QueueDeclarePassiveAsync(_subscriptionName); + return channel; + } + } + protected override void AfterDeserialize(NimbusMessage message) { // Tag so retries route directly to this subscription queue, bypassing the fanout diff --git a/src/Nimbus.Transports.RabbitMQ/Nimbus.Transports.RabbitMQ.csproj b/src/Nimbus.Transports.RabbitMQ/Nimbus.Transports.RabbitMQ.csproj index c9e511fb..c9b34b4e 100644 --- a/src/Nimbus.Transports.RabbitMQ/Nimbus.Transports.RabbitMQ.csproj +++ b/src/Nimbus.Transports.RabbitMQ/Nimbus.Transports.RabbitMQ.csproj @@ -2,7 +2,7 @@ net9.0 - 1.0.0 + 1.1.0 true RabbitMQ transport for the Nimbus messaging framework using the official RabbitMQ.Client v7 async API. nimbus;messaging;transport;rabbitmq;lavinmq;amqp diff --git a/src/Nimbus.Transports.RabbitMQ/Properties/AssemblyVisibility.cs b/src/Nimbus.Transports.RabbitMQ/Properties/AssemblyVisibility.cs new file mode 100644 index 00000000..ffb1103e --- /dev/null +++ b/src/Nimbus.Transports.RabbitMQ/Properties/AssemblyVisibility.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Nimbus.Tests.Integration", AllInternalsVisible = true)] diff --git a/src/Nimbus.Transports.SqlServer/MessageSendersAndReceivers/SqlServerTopicReceiver.cs b/src/Nimbus.Transports.SqlServer/MessageSendersAndReceivers/SqlServerTopicReceiver.cs index 628c94ac..abb4ed04 100644 --- a/src/Nimbus.Transports.SqlServer/MessageSendersAndReceivers/SqlServerTopicReceiver.cs +++ b/src/Nimbus.Transports.SqlServer/MessageSendersAndReceivers/SqlServerTopicReceiver.cs @@ -1,3 +1,4 @@ +using System; using System.Data; using System.Threading; using System.Threading.Tasks; @@ -13,17 +14,23 @@ internal class SqlServerTopicReceiver : SqlServerQueueReceiver private readonly SqlServerSubscription _subscription; private readonly SqlServerTransportConfiguration _configuration; - // Idempotent subscription registration: safe to call on every bus start. + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; + private readonly ILogger _logger; + private DateTimeOffset _lastHeartbeat; + + // Idempotent subscription registration and heartbeat in one statement: safe to call on every bus start, + // and re-creates the row if the reaper removed it during a long outage. private const string RegisterSubscriptionSql = @" - IF NOT EXISTS ( - SELECT 1 FROM NimbusSubscriptions - WHERE TopicName = @TopicName AND SubscriberQueue = @SubscriberQueue - ) - INSERT INTO NimbusSubscriptions (TopicName, SubscriberQueue) - VALUES (@TopicName, @SubscriberQueue)"; + MERGE NimbusSubscriptions WITH (HOLDLOCK) AS target + USING (SELECT @TopicName AS TopicName, @SubscriberQueue AS SubscriberQueue) AS source + ON target.TopicName = source.TopicName AND target.SubscriberQueue = source.SubscriberQueue + WHEN MATCHED THEN UPDATE SET LastSeenAt = SYSUTCDATETIME() + WHEN NOT MATCHED THEN INSERT (TopicName, SubscriberQueue, LastSeenAt) + VALUES (source.TopicName, source.SubscriberQueue, SYSUTCDATETIME());"; public SqlServerTopicReceiver(SqlServerSubscription subscription, SqlServerTransportConfiguration configuration, + AutoDeleteOnIdleSetting autoDeleteOnIdle, ISerializer serializer, ConcurrentHandlerLimitSetting concurrentHandlerLimit, IGlobalHandlerThrottle globalHandlerThrottle, @@ -32,9 +39,17 @@ public SqlServerTopicReceiver(SqlServerSubscription subscription, { _subscription = subscription; _configuration = configuration; + _autoDeleteOnIdle = autoDeleteOnIdle; + _logger = logger; } protected override async Task WarmUp() + { + await RegisterOrTouchSubscription(); + await base.WarmUp(); + } + + private async Task RegisterOrTouchSubscription() { using var connection = new SqlConnection(_configuration.ConnectionString); await connection.OpenAsync(); @@ -44,8 +59,22 @@ protected override async Task WarmUp() command.Parameters.Add("@SubscriberQueue", SqlDbType.NVarChar, 255).Value = _subscription.SubscriberQueueName; await command.ExecuteNonQueryAsync(); + _lastHeartbeat = DateTimeOffset.UtcNow; + } - await base.WarmUp(); + // Refresh the heartbeat several times per idle window so one failed attempt doesn't get us reaped. + private async Task HeartbeatIfDue() + { + if (DateTimeOffset.UtcNow - _lastHeartbeat < TimeSpan.FromTicks(_autoDeleteOnIdle.Value.Ticks / 4)) return; + + try + { + await RegisterOrTouchSubscription(); + } + catch (Exception exc) + { + _logger.Warn(exc, "Failed to refresh heartbeat for subscription {SubscriberQueue}: {Message}", _subscription.SubscriberQueueName, exc.Message); + } } // Override Fetch to stamp the subscriber queue name on the message so that @@ -54,6 +83,8 @@ protected override async Task WarmUp() // fan-out to all subscribers again). protected override async Task Fetch(CancellationToken cancellationToken) { + await HeartbeatIfDue(); + var message = await base.Fetch(cancellationToken); if (message != null) message.Properties[MessagePropertyKeys.RedeliveryToSubscriptionName] = _subscription.SubscriberQueueName; diff --git a/src/Nimbus.Transports.SqlServer/MessageSendersAndReceivers/SqlServerTopicSender.cs b/src/Nimbus.Transports.SqlServer/MessageSendersAndReceivers/SqlServerTopicSender.cs index 219db191..20ce87b8 100644 --- a/src/Nimbus.Transports.SqlServer/MessageSendersAndReceivers/SqlServerTopicSender.cs +++ b/src/Nimbus.Transports.SqlServer/MessageSendersAndReceivers/SqlServerTopicSender.cs @@ -3,6 +3,7 @@ using System.Text; using System.Threading.Tasks; using Microsoft.Data.SqlClient; +using Nimbus.Configuration.Settings; using Nimbus.Infrastructure.MessageSendersAndReceivers; using Nimbus.InfrastructureContracts; @@ -13,18 +14,23 @@ internal class SqlServerTopicSender : INimbusMessageSender private readonly string _topicPath; private readonly SqlServerTransportConfiguration _configuration; private readonly ISerializer _serializer; + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; // Fan-out: one INSERT per subscriber, each with a fresh GUID, all in one atomic statement. // NEWID() is evaluated per row, so every subscriber gets a unique MessageId. // Filters in NimbusSubscriptions are not evaluated yet (Phase 1 delivers to all subscribers). + // Subscriptions whose heartbeat has lapsed are skipped so we don't write messages nobody will read; + // SqlServerIdleSubscriptionReaper removes them (and anything already queued for them) later. private const string FanOutSql = @" INSERT INTO NimbusMessages (MessageId, Destination, Body, VisibleAfter, ExpiresAt) SELECT NEWID(), SubscriberQueue, @Body, @VisibleAfter, @ExpiresAt FROM NimbusSubscriptions - WHERE TopicName = @TopicName"; + WHERE TopicName = @TopicName + AND LastSeenAt > DATEADD(SECOND, -@IdleTimeoutSeconds, SYSUTCDATETIME())"; - public SqlServerTopicSender(string topicPath, SqlServerTransportConfiguration configuration, ISerializer serializer) + public SqlServerTopicSender(string topicPath, SqlServerTransportConfiguration configuration, ISerializer serializer, AutoDeleteOnIdleSetting autoDeleteOnIdle) { + _autoDeleteOnIdle = autoDeleteOnIdle; _topicPath = topicPath; _configuration = configuration; _serializer = serializer; @@ -44,6 +50,7 @@ public async Task Send(NimbusMessage message) using var command = new SqlCommand(FanOutSql, connection); command.Parameters.Add("@TopicName", SqlDbType.NVarChar, 255).Value = _topicPath; + command.Parameters.Add("@IdleTimeoutSeconds", SqlDbType.Int).Value = (int) _autoDeleteOnIdle.Value.TotalSeconds; command.Parameters.Add("@Body", SqlDbType.VarBinary, -1).Value = body; command.Parameters.Add("@VisibleAfter", SqlDbType.DateTime2).Value = visibleAfter; command.Parameters.Add("@ExpiresAt", SqlDbType.DateTime2).Value = (object) message.ExpiresAfter.UtcDateTime ?? DBNull.Value; diff --git a/src/Nimbus.Transports.SqlServer/Nimbus.Transports.SqlServer.csproj b/src/Nimbus.Transports.SqlServer/Nimbus.Transports.SqlServer.csproj index a7ff7d44..ef6d1c2b 100644 --- a/src/Nimbus.Transports.SqlServer/Nimbus.Transports.SqlServer.csproj +++ b/src/Nimbus.Transports.SqlServer/Nimbus.Transports.SqlServer.csproj @@ -10,7 +10,7 @@ - 1.0.0 + 1.1.0 true SQL Server transport for the Nimbus messaging framework using Microsoft.Data.SqlClient. nimbus;messaging;transport;sqlserver;sql-server diff --git a/src/Nimbus.Transports.SqlServer/Properties/AssemblyVisibility.cs b/src/Nimbus.Transports.SqlServer/Properties/AssemblyVisibility.cs new file mode 100644 index 00000000..ffb1103e --- /dev/null +++ b/src/Nimbus.Transports.SqlServer/Properties/AssemblyVisibility.cs @@ -0,0 +1,3 @@ +using System.Runtime.CompilerServices; + +[assembly: InternalsVisibleTo("Nimbus.Tests.Integration", AllInternalsVisible = true)] diff --git a/src/Nimbus.Transports.SqlServer/QueueManagement/SqlServerIdleSubscriptionReaper.cs b/src/Nimbus.Transports.SqlServer/QueueManagement/SqlServerIdleSubscriptionReaper.cs new file mode 100644 index 00000000..948ff4b3 --- /dev/null +++ b/src/Nimbus.Transports.SqlServer/QueueManagement/SqlServerIdleSubscriptionReaper.cs @@ -0,0 +1,83 @@ +using System; +using System.Data; +using System.Threading.Tasks; +using Microsoft.Data.SqlClient; +using Nimbus.Configuration.Settings; +using Nimbus.Extensions; +using Nimbus.InfrastructureContracts; +using Timer = System.Timers.Timer; + +namespace Nimbus.Transports.SqlServer.QueueManagement +{ + /// + /// Removes subscriptions whose receivers have stopped sending heartbeats (see + /// SqlServerTopicReceiver) along with any messages still queued for them. Every bus instance + /// runs a sweep; the batch is idempotent so concurrent sweeps are harmless. + /// + internal class SqlServerIdleSubscriptionReaper : IDisposable + { + private readonly SqlServerTransportConfiguration _configuration; + private readonly AutoDeleteOnIdleSetting _autoDeleteOnIdle; + private readonly ILogger _logger; + private Timer _timer; + + // Uses database time throughout so clock skew between application hosts doesn't matter. + private const string ReapSql = @" + SET XACT_ABORT ON; + BEGIN TRANSACTION; + + DECLARE @Dead TABLE (SubscriberQueue NVARCHAR(255) NOT NULL); + + DELETE FROM NimbusSubscriptions + OUTPUT DELETED.SubscriberQueue INTO @Dead + WHERE LastSeenAt < DATEADD(SECOND, -@IdleTimeoutSeconds, SYSUTCDATETIME()); + + DELETE FROM NimbusMessages + WHERE Destination IN (SELECT SubscriberQueue FROM @Dead); + + COMMIT TRANSACTION;"; + + public SqlServerIdleSubscriptionReaper(SqlServerTransportConfiguration configuration, AutoDeleteOnIdleSetting autoDeleteOnIdle, ILogger logger) + { + _configuration = configuration; + _autoDeleteOnIdle = autoDeleteOnIdle; + _logger = logger; + } + + public void Start() + { + if (_timer != null) return; + + var interval = TimeSpan.FromTicks(_autoDeleteOnIdle.Value.Ticks / 4); + + _timer = new Timer(interval.TotalMilliseconds) {AutoReset = true}; + _timer.Elapsed += (s, e) => Task.Run(ReapOnce).ConfigureAwaitFalse(); + _timer.Start(); + } + + // internal so tests can trigger a sweep synchronously. + internal async Task ReapOnce() + { + try + { + using var connection = new SqlConnection(_configuration.ConnectionString); + await connection.OpenAsync(); + + using var command = new SqlCommand(ReapSql, connection); + command.Parameters.Add("@IdleTimeoutSeconds", SqlDbType.Int).Value = (int) _autoDeleteOnIdle.Value.TotalSeconds; + + await command.ExecuteNonQueryAsync(); + } + catch (Exception exc) + { + _logger.Warn(exc, "Idle subscription reaper sweep failed: {Message}", exc.Message); + } + } + + public void Dispose() + { + _timer?.Dispose(); + _timer = null; + } + } +} diff --git a/src/Nimbus.Transports.SqlServer/Schema/CreateSchema.sql b/src/Nimbus.Transports.SqlServer/Schema/CreateSchema.sql index cdf7c65f..f1f724e4 100644 --- a/src/Nimbus.Transports.SqlServer/Schema/CreateSchema.sql +++ b/src/Nimbus.Transports.SqlServer/Schema/CreateSchema.sql @@ -24,10 +24,16 @@ BEGIN ( TopicName NVARCHAR(255) NOT NULL, SubscriberQueue NVARCHAR(255) NOT NULL, + LastSeenAt DATETIME2 NOT NULL CONSTRAINT DF_NimbusSubscriptions_LastSeenAt DEFAULT SYSUTCDATETIME(), CONSTRAINT PK_NimbusSubscriptions PRIMARY KEY (TopicName, SubscriberQueue) ); END +-- Added for idle subscription cleanup; upgrades tables created by earlier versions. +IF COL_LENGTH('dbo.NimbusSubscriptions', 'LastSeenAt') IS NULL + ALTER TABLE dbo.NimbusSubscriptions + ADD LastSeenAt DATETIME2 NOT NULL CONSTRAINT DF_NimbusSubscriptions_LastSeenAt DEFAULT SYSUTCDATETIME(); + IF NOT EXISTS (SELECT 1 FROM sys.tables WHERE name = 'NimbusDeadLetters' AND schema_id = SCHEMA_ID('dbo')) BEGIN CREATE TABLE dbo.NimbusDeadLetters diff --git a/src/Nimbus.Transports.SqlServer/SqlServerTransport.cs b/src/Nimbus.Transports.SqlServer/SqlServerTransport.cs index 34dec04c..efc148dc 100644 --- a/src/Nimbus.Transports.SqlServer/SqlServerTransport.cs +++ b/src/Nimbus.Transports.SqlServer/SqlServerTransport.cs @@ -1,3 +1,4 @@ +using System; using System.Threading.Tasks; using Microsoft.Data.SqlClient; using Nimbus.Configuration.PoorMansIocContainer; @@ -5,18 +6,22 @@ using Nimbus.Infrastructure.MessageSendersAndReceivers; using Nimbus.InfrastructureContracts.Filtering.Conditions; using Nimbus.Transports.SqlServer.MessageSendersAndReceivers; +using Nimbus.Transports.SqlServer.QueueManagement; using Nimbus.Transports.SqlServer.Schema; namespace Nimbus.Transports.SqlServer { - internal class SqlServerTransport : INimbusTransport + internal class SqlServerTransport : INimbusTransport, IDisposable { private readonly PoorMansIoC _container; private readonly SqlServerTransportConfiguration _configuration; private readonly SqlServerSchemaCreator _schemaCreator; + private readonly SqlServerIdleSubscriptionReaper _idleSubscriptionReaper; - public SqlServerTransport(PoorMansIoC container, SqlServerTransportConfiguration configuration, SqlServerSchemaCreator schemaCreator) + public SqlServerTransport(PoorMansIoC container, SqlServerTransportConfiguration configuration, SqlServerSchemaCreator schemaCreator, + SqlServerIdleSubscriptionReaper idleSubscriptionReaper) { + _idleSubscriptionReaper = idleSubscriptionReaper; _container = container; _configuration = configuration; _schemaCreator = schemaCreator; @@ -29,6 +34,8 @@ public async Task TestConnection() if (_configuration.AutoCreateSchema) await _schemaCreator.EnsureSchemaExists(); + + _idleSubscriptionReaper.Start(); } public INimbusMessageSender GetQueueSender(string queuePath) @@ -51,5 +58,10 @@ public INimbusMessageReceiver GetTopicReceiver(string topicPath, string subscrip var subscription = new SqlServerSubscription(topicPath, subscriptionName); return _container.ResolveWithOverrides(subscription); } + + public void Dispose() + { + _idleSubscriptionReaper.Dispose(); + } } } diff --git a/src/Nimbus.Transports.SqlServer/SqlServerTransportConfiguration.cs b/src/Nimbus.Transports.SqlServer/SqlServerTransportConfiguration.cs index ed40963b..b5a22e75 100644 --- a/src/Nimbus.Transports.SqlServer/SqlServerTransportConfiguration.cs +++ b/src/Nimbus.Transports.SqlServer/SqlServerTransportConfiguration.cs @@ -51,6 +51,7 @@ protected override void RegisterComponents(PoorMansIoC container) container.RegisterType(ComponentLifetime.SingleInstance, typeof(IDeadLetterOffice)); container.RegisterType(ComponentLifetime.SingleInstance, typeof(INamespaceCleanser)); container.RegisterType(ComponentLifetime.SingleInstance); + container.RegisterType(ComponentLifetime.SingleInstance); container.RegisterType(ComponentLifetime.SingleInstance, typeof(ILargeMessageBodyStore)); container.RegisterType(ComponentLifetime.SingleInstance, typeof(INimbusTransport));