Add LFBP Instrumentation, Server Shutdown Ordering, and Test TearDown Hardening (#1632)

* Fix dispose hang in network handler and buffer pool cleanup

Two bugs caused LimitedFixedBufferPool.Dispose() to hang indefinitely
during server teardown (blocking ClusterResetHardDuringDisklessReplicationAttach):

1. TcpNetworkHandlerBase.Dispose() never called DisposeImpl(), so when a
   handler thread was blocked synchronously (e.g. in TryBeginDisklessSync),
   the CTS was never cancelled and activeHandlerCount was never decremented.
   DisposeActiveHandlers() would spin forever waiting for it to reach 0.

2. GarnetTcpNetworkSender.DisposeNetworkSender() disposed the saeaStack
   but not the current responseObject, leaking a PoolEntry that was never
   returned to the pool. LimitedFixedBufferPool.Dispose() then spun
   forever waiting for totalReferences to reach 0.

Also adds PoolEntry source tracking infrastructure (PoolEntryBufferType
and PoolOwnerType enums) with DEBUG-only diagnostics that log unreturned
buffer details after a 5-second timeout during pool disposal.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* use stopwatch only in debug

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* add concurrent dictionary collection only in debug

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* reduce pool entry types to single byte

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Fix PoolEntry double-dispose in DisposeNetworkSender

Use Interlocked.Exchange to atomically take ownership of responseObject
and ReturnBuffer it back to the saeaStack before disposal. This ensures
the PoolEntry is disposed exactly once when saeaStack.Dispose() iterates
all items, and avoids a race with ReturnResponseObject() on the handler
thread that could cause Debug.Assert(!disposed) to fail.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* revert dispose calls and elevate logging for LFBP dispose timeout

* improve TearDown robustness

* propagate ownerType flag across clients

* fix ClusterReplicationCheckpointCleanupTest

* fix ClusterResetHardDuring tests by using ConfigureAwait for ExceptionInjection

* add logger for write/read asycn in device

* separate dispose from close and configure socket to allow rapid connect

* make ExceptionInjection ConfigureAwait(false)

* wip; restructuring cluster tests to reduce CI duration

* add logging for IDevice implementation

* log input to AOF for object and unified session functions

* fix ClusterSimpleFailoverAuth

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: 3 people

benchmark/BDN.benchmark/Embedded/GarnetServerEmbedded.cs

@@ -110,6 +110,11 @@ public override void Start()
         {
         }
 
+        /// <inheritdoc />
+        public override void Close()
+        {
+        }
+
         public bool TryCreateMessageConsumer(Span<byte> bytes, INetworkSender networkSender, out IMessageConsumer session)
         {
             session = null;

libs/client/ClientSession/GarnetClientSession.cs

@@ -127,7 +127,7 @@ public GarnetClientSession(
 
             this.usingManagedNetworkPool = networkPool != null;
             this.networkBufferSettings = networkBufferSettings;
-            this.networkPool = networkPool ?? networkBufferSettings.CreateBufferPool();
+            this.networkPool = networkPool ?? networkBufferSettings.CreateBufferPool(ownerType: PoolOwnerType.GarnetClientSession, logger: logger);
             this.bufferSizeDigits = NumUtils.CountDigits(this.networkBufferSettings.sendBufferSize);
 
             this.logger = logger;

libs/client/GarnetClient.cs

@@ -210,7 +210,7 @@ public GarnetClient(
         public void Connect(CancellationToken token = default)
         {
             socket = ConnectSendSocketAsync(timeoutMilliseconds).ConfigureAwait(false).GetAwaiter().GetResult();
-            networkWriter = new NetworkWriter(this, socket, bufferSize, sslOptions, out networkHandler, sendPageSize, networkSendThrottleMax, epoch, logger);
+            networkWriter = new NetworkWriter(this, socket, bufferSize, sslOptions, out networkHandler, sendPageSize, networkSendThrottleMax, epoch, PoolOwnerType.GarnetClient, logger);
             networkHandler.StartAsync(sslOptions, EndPoint.ToString(), token).ConfigureAwait(false).GetAwaiter().GetResult();
 
             if (timeoutMilliseconds > 0)
@@ -256,7 +256,7 @@ public void Connect(CancellationToken token = default)
         public async Task ConnectAsync(CancellationToken token = default)
         {
             socket = await ConnectSendSocketAsync(timeoutMilliseconds, token).ConfigureAwait(false);
-            networkWriter = new NetworkWriter(this, socket, bufferSize, sslOptions, out networkHandler, sendPageSize, networkSendThrottleMax, epoch, logger);
+            networkWriter = new NetworkWriter(this, socket, bufferSize, sslOptions, out networkHandler, sendPageSize, networkSendThrottleMax, epoch, PoolOwnerType.GarnetClient, logger);
             await networkHandler.StartAsync(sslOptions, EndPoint.ToString(), token).ConfigureAwait(false);
 
             if (timeoutMilliseconds > 0)

libs/client/NetworkWriter.cs

@@ -79,10 +79,10 @@ internal sealed class NetworkWriter : IDisposable
         /// <summary>
         /// Constructor
         /// </summary>
-        public NetworkWriter(GarnetClient serverHook, Socket socket, int messageBufferSize, SslClientAuthenticationOptions sslOptions, out GarnetClientTcpNetworkHandler networkHandler, int sendPageSize, int networkSendThrottleMax, LightEpoch epoch, ILogger logger = null)
+        public NetworkWriter(GarnetClient serverHook, Socket socket, int messageBufferSize, SslClientAuthenticationOptions sslOptions, out GarnetClientTcpNetworkHandler networkHandler, int sendPageSize, int networkSendThrottleMax, LightEpoch epoch, PoolOwnerType ownerType, ILogger logger = null)
         {
             this.networkBufferSettings = new NetworkBufferSettings(messageBufferSize, messageBufferSize);
-            this.networkPool = networkBufferSettings.CreateBufferPool(logger: logger);
+            this.networkPool = networkBufferSettings.CreateBufferPool(ownerType: ownerType, logger: logger);
 
             if (BufferSize > PageOffset.kPageMask) throw new Exception();
             this.networkHandler = networkHandler = new GarnetClientTcpNetworkHandler(serverHook, AsyncFlushPageCallback, socket, networkBufferSettings, networkPool, sslOptions != null, serverHook, networkSendThrottleMax: networkSendThrottleMax, logger: logger);

libs/cluster/Server/Migration/MigrateSessionSlots.cs

@@ -24,7 +24,7 @@ public async Task<bool> MigrateSlotsDriverInline()
 
 #if DEBUG
             // Only on Debug mode
-            ExceptionInjectionHelper.WaitOnSet(ExceptionInjectionType.Migration_Slot_End_Scan_Range_Acquisition).GetAwaiter().GetResult();
+            await ExceptionInjectionHelper.ResetAndWaitAsync(ExceptionInjectionType.Migration_Slot_End_Scan_Range_Acquisition).ConfigureAwait(false);
 #endif
 
             // Send store

libs/cluster/Server/Migration/MigrationManager.cs

@@ -43,7 +43,7 @@ public MigrationManager(ClusterProvider clusterProvider, ILogger logger = null)
             this.clusterProvider = clusterProvider;
             var sendBufferSize = 1 << clusterProvider.serverOptions.PageSizeBits();
             this.networkBufferSettings = new NetworkBufferSettings(sendBufferSize, initialReceiveBufferSize);
-            this.networkPool = networkBufferSettings.CreateBufferPool(logger: logger);
+            this.networkPool = networkBufferSettings.CreateBufferPool(ownerType: PoolOwnerType.Migration, logger: logger);
 
             logger?.LogInformation("NetworkBufferSettings.sendBufferSize:{sendBufferSize}", networkBufferSettings.sendBufferSize);
             logger?.LogInformation("NetworkBufferSettings.initialReceiveBufferSize:{initialReceiveBufferSize}", networkBufferSettings.initialReceiveBufferSize);

libs/cluster/Server/Replication/PrimaryOps/ReplicaSyncSession.cs

@@ -310,6 +310,7 @@ public async Task<bool> SendCheckpoint()
             const int maxOdcAttempts = 2;
             while (true)
             {
+                cts.Token.ThrowIfCancellationRequested();
                 logger?.LogInformation("AcquireCheckpointEntry iteration {iteration}", iteration);
                 iteration++;
 
@@ -338,7 +339,7 @@ public async Task<bool> SendCheckpoint()
 
 #if DEBUG
                 // Only on Debug mode
-                await ExceptionInjectionHelper.WaitOnSet(ExceptionInjectionType.Replication_Wait_After_Checkpoint_Acquisition);
+                await ExceptionInjectionHelper.ResetAndWaitAsync(ExceptionInjectionType.Replication_Wait_After_Checkpoint_Acquisition).ConfigureAwait(false);
 #endif
 
                 // Calculate the minimum start address covered by this checkpoint

libs/cluster/Server/Replication/ReplicaOps/ReplicaDisklessSync.cs

@@ -128,7 +128,7 @@ async Task<string> TryBeginReplicaSync(bool downgradeLock)
                     using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(ctsRepManager.Token, resetHandler.Token);
 
                     // Exception injection point for testing cluster reset during diskless replication
-                    await ExceptionInjectionHelper.WaitOnSet(ExceptionInjectionType.Replication_InProgress_During_Diskless_Replica_Attach_Sync).WaitAsync(storeWrapper.serverOptions.ReplicaAttachTimeout, linkedCts.Token).ConfigureAwait(false);
+                    await ExceptionInjectionHelper.ResetAndWaitAsync(ExceptionInjectionType.Replication_InProgress_During_Diskless_Replica_Attach_Sync).WaitAsync(storeWrapper.serverOptions.ReplicaAttachTimeout, linkedCts.Token).ConfigureAwait(false);
 
                     var resp = await gcs.ExecuteAttachSync(syncMetadata.ToByteArray()).WaitAsync(storeWrapper.serverOptions.ReplicaAttachTimeout, linkedCts.Token).ConfigureAwait(false);
                 }

libs/cluster/Server/Replication/ReplicaOps/ReplicaReceiveCheckpoint.cs

@@ -148,7 +148,7 @@ async Task<string> ReplicaSyncAttachTask(bool downgradeLock)
                     using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(ctsRepManager.Token, resetHandler.Token);
 
                     // Exception injection point for testing cluster reset during disk-based replication
-                    await ExceptionInjectionHelper.WaitOnSet(ExceptionInjectionType.Replication_InProgress_During_DiskBased_Replica_Attach_Sync).WaitAsync(storeWrapper.serverOptions.ReplicaAttachTimeout, linkedCts.Token).ConfigureAwait(false);
+                    await ExceptionInjectionHelper.ResetAndWaitAsync(ExceptionInjectionType.Replication_InProgress_During_DiskBased_Replica_Attach_Sync).WaitAsync(storeWrapper.serverOptions.ReplicaAttachTimeout, linkedCts.Token).ConfigureAwait(false);
                     var resp = await gcs.ExecuteReplicaSync(
                         nodeId,
                         PrimaryReplId,

libs/cluster/Server/Replication/ReplicationManager.cs

@@ -102,7 +102,7 @@ public ReplicationManager(ClusterProvider clusterProvider, ILogger logger = null
             this.pageSizeBits = storeWrapper.appendOnlyFile == null ? 0 : storeWrapper.appendOnlyFile.UnsafeGetLogPageSizeBits();
 
             networkBufferSettings.Log(logger, nameof(ReplicationManager));
-            this.networkPool = networkBufferSettings.CreateBufferPool(logger: logger);
+            this.networkPool = networkBufferSettings.CreateBufferPool(ownerType: PoolOwnerType.Replication, logger: logger);
             ValidateNetworkBufferSettings();
 
             aofProcessor = new AofProcessor(storeWrapper, recordToAof: false, clusterProvider: clusterProvider, logger: logger);