From caa2a6a60b281a2dc40fac56fa49c3d796752ad7 Mon Sep 17 00:00:00 2001
From: Ziya Suzen <ziya@suzen.net>
Date: Tue, 16 Jun 2026 09:41:16 +0100
Subject: [PATCH] jetstream: dispose ordered push consumer sub on teardown

NatsJSOrderedPushConsumer.DisposeAsync left its current subscription alive, so the sub's ConnectionOpened handler stayed attached to the connection and leaked one subscription per object store Get/Watch call on long-lived connections. Dispose the current sub after the background loops stop.

Co-authored-by: haoguanjun <4115769+haoguanjun@users.noreply.github.com>
---
 .../Internal/NatsJSOrderedPushConsumer.cs     | 11 ++++++
 .../ObjectStoreTest.cs                        | 37 +++++++++++++++++++
 2 files changed, 48 insertions(+)

diff --git a/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs b/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs
index 2880a76d1..94726c950 100644
--- a/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs
+++ b/src/NATS.Client.JetStream/Internal/NatsJSOrderedPushConsumer.cs
@@ -178,6 +178,17 @@ public async ValueTask DisposeAsync()
 
         _msgChannel.Writer.TryComplete();
 
+        // Dispose the current subscription. Both background loops have completed
+        // by now, so no new subscription can be created. Without this the sub's
+        // ConnectionOpened handler stays attached to the connection, rooting the
+        // sub for the connection's lifetime (memory leak).
+        var sub = _sub;
+        if (sub != null)
+        {
+            await sub.UnsubscribeAsync();
+            await sub.DisposeAsync();
+        }
+
         using var cts = CancellationTokenSource.CreateLinkedTokenSource(_cancellationToken);
         cts.CancelAfter(_opts.CleanupTimeout);
         try
diff --git a/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs b/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs
index 023618421..2fd5c31cf 100644
--- a/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs
+++ b/tests/NATS.Client.ObjectStore.Tests/ObjectStoreTest.cs
@@ -1,4 +1,5 @@
 using System.Diagnostics;
+using System.Reflection;
 using System.Security.Cryptography;
 using System.Text;
 using NATS.Client.Core.Tests;
@@ -663,4 +664,40 @@ public async Task Metadata_field_types_match_spec()
 #pragma warning restore IDE0007
         Assert.Equal(1U, chunks);
     }
+
+    [Fact]
+    public async Task Get_does_not_leak_connection_opened_handlers()
+    {
+        var cts = new CancellationTokenSource(TimeSpan.FromSeconds(30));
+        var cancellationToken = cts.Token;
+
+        await using var server = await NatsServerProcess.StartAsync();
+        await using var nats = new NatsConnection(new NatsOpts { Url = server.Url });
+        var js = new NatsJSContext(nats);
+        var ob = new NatsObjContext(js);
+
+        var store = await ob.CreateObjectStoreAsync(new NatsObjConfig("b1"), cancellationToken);
+        await store.PutAsync("k1", new byte[] { 1, 2, 3 }, cancellationToken: cancellationToken);
+
+        // The ordered push consumer created per GetAsync registers a ConnectionOpened
+        // handler that must be removed when the consumer is disposed. If it is not, the
+        // handler (and the subscription it roots) leaks for the connection's lifetime.
+        var baseline = ConnectionOpenedHandlerCount(nats);
+
+        for (var i = 0; i < 10; i++)
+        {
+            var bytes = await store.GetBytesAsync("k1", cancellationToken);
+            Assert.Equal(new byte[] { 1, 2, 3 }, bytes);
+        }
+
+        Assert.Equal(baseline, ConnectionOpenedHandlerCount(nats));
+    }
+
+    private static int ConnectionOpenedHandlerCount(NatsConnection nats)
+    {
+        var field = typeof(NatsConnection).GetField("ConnectionOpened", BindingFlags.Instance | BindingFlags.NonPublic);
+        Assert.NotNull(field);
+        var handler = (Delegate?)field.GetValue(nats);
+        return handler?.GetInvocationList().Length ?? 0;
+    }
 }