Idle subscription cleanup

.gitignore

@@ -24,4 +24,7 @@ dist_package/
 
 .claude
 data/
-CLAUDE.md
+CLAUDE.md
+
+# graft's local graph cache — regenerable, not committed (run `graft build`).
+/graft/

.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/

.mcp.json

@@ -0,0 +1,10 @@
+{
+  "mcpServers": {
+    "graft": {
+      "command": "graft",
+      "args": [
+        "mcp"
+      ]
+    }
+  }
+}

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
+{
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<TimeSpan> TopicStreamMaxAge()
+        {
+            var stream = (await _jsContextFactory.ListStreamsAsync()).Single(s => s.Config.Subjects!.Contains(_subscription.TopicPath));
+            return stream.Config.MaxAge;
+        }
+
+        private async Task<bool> StreamExists(string name)
+        {
+            return (await _jsContextFactory.ListStreamsAsync()).Any(s => s.Config.Name == name);
+        }
+    }
+}

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
+{
+    /// <summary>
+    ///     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.
+    /// </summary>
+    [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<double?> 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<long> 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<object> 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;
+        }
+    }
+}