From f06ee99c31d992c450b6cac17597f72df6d23797 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Tue, 26 May 2026 16:54:50 +0900 Subject: [PATCH 01/14] Add server quiesce and stop-listening primitives for graceful shutdown Co-authored-by: Cursor --- libs/server/Servers/GarnetServerBase.cs | 17 ++++++++++++++ libs/server/Servers/GarnetServerOptions.cs | 5 +++++ libs/server/Servers/GarnetServerTcp.cs | 26 ++++++++++++++++++++-- libs/server/Servers/IGarnetServer.cs | 19 ++++++++++++++++ 4 files changed, 65 insertions(+), 2 deletions(-) diff --git a/libs/server/Servers/GarnetServerBase.cs b/libs/server/Servers/GarnetServerBase.cs index 2f01da25dd7..2cd27bd5b76 100644 --- a/libs/server/Servers/GarnetServerBase.cs +++ b/libs/server/Servers/GarnetServerBase.cs @@ -49,6 +49,17 @@ public abstract class GarnetServerBase : IGarnetServer, IConnectionSource /// public bool Disposed { get; set; } + /// + /// Backing field for . + /// + int isQuiescing = 0; + + /// + public bool IsQuiescing => isQuiescing != 0; + + /// + public void BeginQuiesce() => Interlocked.Exchange(ref isQuiescing, 1); + /// /// Logger /// @@ -196,6 +207,12 @@ public bool AddSession(WireFormat protocol, ref ISessionProvider provider, INetw /// public abstract void Start(); + /// + public virtual void StopListening() + { + // Base implementation does nothing; derived classes should override + } + /// public abstract void Close(); diff --git a/libs/server/Servers/GarnetServerOptions.cs b/libs/server/Servers/GarnetServerOptions.cs index 5d54e4abc8e..b1b71228a91 100644 --- a/libs/server/Servers/GarnetServerOptions.cs +++ b/libs/server/Servers/GarnetServerOptions.cs @@ -325,6 +325,11 @@ public class GarnetServerOptions : ServerOptions /// public bool QuietMode = false; + /// + /// Timeout (in seconds) for waiting on active connections to drain during graceful shutdown. + /// + public int ShutdownTimeoutSeconds = 5; + /// /// SAVE and BGSAVE: We will take a full (index + log) checkpoint when ReadOnlyAddress of log increases by this amount, from the last full checkpoint. /// diff --git a/libs/server/Servers/GarnetServerTcp.cs b/libs/server/Servers/GarnetServerTcp.cs index 307d6c80797..c4c8d5d88ec 100644 --- a/libs/server/Servers/GarnetServerTcp.cs +++ b/libs/server/Servers/GarnetServerTcp.cs @@ -147,10 +147,32 @@ public GarnetServerTcp( /// /// Stop listening for new connections. Frees the listening port /// without waiting for active connections to drain. + /// Safe to call multiple times (idempotent). /// public override void Close() { - listenSocket.Close(); + try + { + listenSocket.Close(); + } + catch (ObjectDisposedException) { } + } + + /// + public override void StopListening() + { + try + { + // Delegates to Close() which closes the listen socket. + // This will cause any pending AcceptAsync to complete with OperationAborted, + // handled as a Tier 1 (clean shutdown) error in HandleAcceptError. + Close(); + logger?.LogDebug("Stopped accepting new connections on {endpoint}", EndPoint); + } + catch (Exception ex) + { + logger?.LogDebug(ex, "Error closing listen socket on {endpoint}", EndPoint); + } } /// @@ -158,7 +180,7 @@ public override void Close() /// public override void Dispose() { - // Close listening socket to free the port and stop accepting new connections. + // Dispose listening socket (may already be closed by StopListening/Close during graceful shutdown). // This also prevents new connections from arriving while DisposeActiveHandlers drains existing ones. listenSocket.Dispose(); base.Dispose(); diff --git a/libs/server/Servers/IGarnetServer.cs b/libs/server/Servers/IGarnetServer.cs index 2b33e77648d..93a5a621751 100644 --- a/libs/server/Servers/IGarnetServer.cs +++ b/libs/server/Servers/IGarnetServer.cs @@ -47,10 +47,29 @@ public interface IGarnetServer : IDisposable /// public void Start(); + /// + /// Stop accepting new connections (for graceful shutdown). + /// Existing connections remain active until they complete or are disposed. + /// + public void StopListening(); + /// /// Stop listening for new connections. Frees the listening port /// without waiting for active connections to drain. /// public void Close(); + + /// + /// Signal all active sessions to reject incoming commands. + /// Sessions complete any in-flight command, respond with a LOADING error + /// on the next received message, then close the connection. + /// Must be called before for deterministic shutdown. + /// + public void BeginQuiesce(); + + /// + /// True after has been called. + /// + public bool IsQuiescing { get; } } } \ No newline at end of file From ec0efde401ced9e1cc4f54d228036304b722fbf7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Tue, 26 May 2026 16:55:08 +0900 Subject: [PATCH 02/14] Quiesce RESP sessions and pub/sub during graceful shutdown Co-authored-by: Cursor --- libs/server/PubSub/SubscribeBroker.cs | 10 ++++++++++ libs/server/Resp/RespServerSession.cs | 14 ++++++++++++++ 2 files changed, 24 insertions(+) diff --git a/libs/server/PubSub/SubscribeBroker.cs b/libs/server/PubSub/SubscribeBroker.cs index f4dcbbf4a27..01a656394d3 100644 --- a/libs/server/PubSub/SubscribeBroker.cs +++ b/libs/server/PubSub/SubscribeBroker.cs @@ -21,6 +21,7 @@ public sealed class SubscribeBroker : IDisposable, ILogEntryConsumer { int sid = 0; bool initialized = false; + volatile bool isQuiescing = false; ConcurrentDictionary> subscriptions; ReadOptimizedConcurrentSet patternSubscriptions; readonly TsavoriteLog aof; @@ -49,6 +50,13 @@ public SubscribeBroker(string logDir, long pageSize, LightEpoch epoch, bool star this.logger = logger; } + /// + /// Signal the broker to stop delivering messages to subscribers. + /// Any or calls made after this + /// are silently dropped, ensuring no new fan-out occurs during shutdown quiesce. + /// + public void BeginQuiesce() => isQuiescing = true; + /// /// Remove all subscriptions for a session, /// called during dispose of server session @@ -297,6 +305,7 @@ public unsafe List ListAllPatternSubscriptions(ServerSessionBa /// Number of subscribers notified public unsafe int PublishNow(PinnedSpanByte key, PinnedSpanByte value) { + if (isQuiescing) return 0; if (subscriptions == null && patternSubscriptions == null) return 0; return Broadcast(key, value); } @@ -308,6 +317,7 @@ public unsafe int PublishNow(PinnedSpanByte key, PinnedSpanByte value) /// value that has been updated public unsafe void Publish(PinnedSpanByte key, PinnedSpanByte value) { + if (isQuiescing) return; if (subscriptions == null && patternSubscriptions == null) return; diff --git a/libs/server/Resp/RespServerSession.cs b/libs/server/Resp/RespServerSession.cs index abf97d0c2f5..8955e355204 100644 --- a/libs/server/Resp/RespServerSession.cs +++ b/libs/server/Resp/RespServerSession.cs @@ -540,6 +540,20 @@ internal bool CanRunModule() public override int TryConsumeMessages(byte* reqBuffer, int bytesReceived) { + // Reject new commands when the server is quiescing for shutdown. + // Complete any in-flight operation is not possible here (this is the entry point before parsing), + // so we send a LOADING error and dispose the connection immediately. + if (server is GarnetServerBase { IsQuiescing: true }) + { + networkSender.EnterAndGetResponseObject(out dcurr, out dend); + while (!RespWriteUtils.TryWriteError("LOADING Garnet is shutting down"u8, ref dcurr, dend)) + SendAndReset(); + Send(networkSender.GetResponseObjectHead()); + networkSender.ExitAndReturnResponseObject(); + networkSender.DisposeNetworkSender(true); + return 0; + } + bytesRead = bytesReceived; if (!txnSkip) readHead = 0; From 7da3613863027813093ffe8ead38b53b92948587 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Tue, 26 May 2026 16:55:51 +0900 Subject: [PATCH 03/14] Add GarnetServer.ShutdownAsync with configurable shutdown timeout Co-authored-by: Cursor --- libs/host/Configuration/Options.cs | 6 + libs/host/GarnetServer.cs | 207 +++++++++++++++++++++++++++++ libs/host/defaults.conf | 3 + 3 files changed, 216 insertions(+) diff --git a/libs/host/Configuration/Options.cs b/libs/host/Configuration/Options.cs index 4694ca0f32a..996a7e025df 100644 --- a/libs/host/Configuration/Options.cs +++ b/libs/host/Configuration/Options.cs @@ -402,6 +402,11 @@ internal sealed class Options : ICloneable [Option("network-connection-limit", Required = false, HelpText = "Maximum number of simultaneously active network connections across all listeners, or -1 for unlimited. Equivalent to the Redis maxclients parameter, and settable at runtime with CONFIG SET maxclients.")] public int NetworkConnectionLimit { get; set; } + [IntRangeValidation(1, int.MaxValue)] + [Option("shutdown-timeout", Required = false, HelpText = "Timeout in seconds to wait for active connections to drain during graceful shutdown. " + + "The Windows SCM default pre-kill wait is 5 seconds, so values below 5 are not recommended when running as a Windows service.")] + public int ShutdownTimeoutSeconds { get; set; } + [OptionValidation] [Option("use-azure-storage", Required = false, HelpText = "Use Azure Page Blobs for storage instead of local storage.")] public bool? UseAzureStorage { get; set; } @@ -984,6 +989,7 @@ endpoint is IPEndPoint listenEp && clusterAnnounceEndpoint[0] is IPEndPoint anno ThreadPoolMinIOCompletionThreads = ThreadPoolMinIOCompletionThreads, ThreadPoolMaxIOCompletionThreads = ThreadPoolMaxIOCompletionThreads, NetworkConnectionLimit = NetworkConnectionLimit, + ShutdownTimeoutSeconds = ShutdownTimeoutSeconds, DeviceFactoryCreator = deviceType == DeviceType.AzureStorage ? azureFactoryCreator() : new LocalStorageNamedDeviceFactoryCreator( deviceType: deviceType, diff --git a/libs/host/GarnetServer.cs b/libs/host/GarnetServer.cs index 8826935c628..50884165a88 100644 --- a/libs/host/GarnetServer.cs +++ b/libs/host/GarnetServer.cs @@ -11,6 +11,7 @@ using System.Runtime.InteropServices; using System.Text; using System.Threading; +using System.Threading.Tasks; using Garnet.cluster; using Garnet.common; using Garnet.networking; @@ -542,6 +543,212 @@ public void Start() Console.WriteLine("* Ready to accept connections"); } + /// + /// Performs graceful shutdown of the server. + /// Stops accepting new connections, waits for active connections to complete, commits AOF, and takes checkpoint if needed. + /// + /// Timeout for waiting on active connections (default: 30 seconds) + /// If true, skip data persistence (AOF commit and checkpoint) during shutdown + /// Cancellation token + /// Task representing the async shutdown operation + public async Task ShutdownAsync(TimeSpan? timeout = null, bool noSave = false, CancellationToken token = default) + { + var shutdownTimeout = timeout ?? TimeSpan.FromSeconds(30); + + try + { + // Quiesce existing sessions first: they will reject the next incoming message + // and close themselves, so FinalizeDataAsync runs with no concurrent writers. + if (servers != null) + { + foreach (var server in servers) + server.BeginQuiesce(); + } + + // Quiesce pub/sub fan-out so no new messages are delivered after this point. + subscribeBroker?.BeginQuiesce(); + + // Stop accepting new connections. + StopListening(); + + // Wait for existing connections to complete (cancellable) + try + { + await WaitForActiveConnectionsAsync(shutdownTimeout, token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + logger?.LogWarning("Connection draining was cancelled. Proceeding with data finalization..."); + } + } + catch (Exception ex) + { + logger?.LogError(ex, "Error during graceful shutdown"); + } + finally + { + if (!noSave) + { + // Attempt AOF commit or checkpoint as best-effort, + // even if connection draining was cancelled or failed. + // Use a bounded timeout instead of the caller's token to ensure completion. + using var finalizeCts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); + try + { + await FinalizeDataAsync(finalizeCts.Token).ConfigureAwait(false); + } + catch (Exception ex) + { + logger?.LogError(ex, "Error during data finalization"); + } + } + else + { + logger?.LogInformation("Shutdown with noSave flag - skipping data persistence."); + } + } + } + + /// + /// Stop all servers from accepting new connections. + /// + private void StopListening() + { + if (servers == null) return; + + logger?.LogDebug("Stopping listeners to prevent new connections..."); + foreach (var server in servers) + { + try + { + server?.StopListening(); + } + catch (Exception ex) + { + logger?.LogWarning(ex, "Error stopping listener"); + } + } + } + + /// + /// Waits for active connections to complete within the specified timeout. + /// + private async Task WaitForActiveConnectionsAsync(TimeSpan timeout, CancellationToken token) + { + if (servers == null) return; + + // Linked Token : between external token and timeout + using var cts = CancellationTokenSource.CreateLinkedTokenSource(token); + cts.CancelAfter(timeout); + + var delays = new[] { 50, 300, 1000 }; + var delayIndex = 0; + + try + { + while (!cts.Token.IsCancellationRequested) + { + var activeConnections = GetActiveConnectionCount(); + if (activeConnections == 0) + { + logger?.LogInformation("All connections have been closed gracefully."); + return; + } + + logger?.LogInformation("Waiting for {ActiveConnections} active connections to complete...", activeConnections); + + var currentDelay = delays[delayIndex]; + if (delayIndex < delays.Length - 1) delayIndex++; + + await Task.Delay(currentDelay, cts.Token).ConfigureAwait(false); + } + } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + throw; + } + catch (OperationCanceledException) + { + // timeout reached error logging + logger?.LogWarning("Timeout reached after {TimeoutSeconds} seconds. Some connections may still be active.", + timeout.TotalSeconds); + } + catch (Exception ex) + { + logger?.LogWarning(ex, "Error checking active connections"); + await Task.Delay(500, token).ConfigureAwait(false); + } + } + + /// + /// Gets the current number of active connections directly from server instances. + /// + private long GetActiveConnectionCount() + { + long count = 0; + if (servers != null) + { + foreach (var garnetServer in servers) + { + if (garnetServer is GarnetServerBase garnetServerBase) + { + count += garnetServerBase.get_conn_active(); + } + } + } + return count; + } + + /// + /// Persists data during shutdown using AOF or checkpoint based on configuration. + /// + private async Task FinalizeDataAsync(CancellationToken token) + { + if (opts.EnableAOF) + { + logger?.LogDebug("Committing AOF before shutdown..."); + try + { + var commitSuccess = await Store.CommitAOFAsync(token).ConfigureAwait(false); + if (commitSuccess) + { + logger?.LogDebug("AOF committed successfully."); + } + else + { + logger?.LogInformation("AOF commit skipped (another commit in progress or replica mode)."); + } + } + catch (Exception ex) + { + logger?.LogError(ex, "Error committing AOF during shutdown"); + } + + return; + } + + if (!opts.EnableStorageTier) + return; + + logger?.LogDebug("Taking checkpoint for tiered storage..."); + try + { + var checkpointSuccess = await storeWrapper.TakeCheckpointAsync(background: false, token: token, logger: logger).ConfigureAwait(false); + if (checkpointSuccess) + { + logger?.LogDebug("Checkpoint completed successfully."); + } + else + { + logger?.LogInformation("Checkpoint skipped (another checkpoint in progress or replica mode)."); + } + } + catch (Exception ex) + { + logger?.LogError(ex, "Error taking checkpoint during shutdown"); + } + } + /// /// Dispose store (including log and checkpoint directory) /// diff --git a/libs/host/defaults.conf b/libs/host/defaults.conf index d73b74472b9..988f5eb09be 100644 --- a/libs/host/defaults.conf +++ b/libs/host/defaults.conf @@ -308,6 +308,9 @@ /* Maximum number of simultaneously active network connections. */ "NetworkConnectionLimit" : 10000, + /* Timeout in seconds to wait for active connections to drain during graceful shutdown. */ + "ShutdownTimeoutSeconds" : 5, + /* Use Azure Page Blobs for storage instead of local storage. */ "UseAzureStorage" : false, From 4adc1b5708f55a6063b52832ca77e67b6c9093d9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Tue, 26 May 2026 16:56:57 +0900 Subject: [PATCH 04/14] Wire graceful shutdown into hosts and add tests Co-authored-by: Cursor --- hosting/Windows/Garnet.worker/Program.cs | 41 +- hosting/Windows/Garnet.worker/Worker.cs | 36 +- libs/host/GarnetServer.cs | 7 +- libs/server/Servers/StoreApi.cs | 19 + main/GarnetServer/Program.cs | 51 +- .../Garnet.test/GarnetServerTcpTests.cs | 212 ++++++++- .../Garnet.test/RespAdminCommandsTests.cs | 105 +++++ .../ShutdownDataConsistencyTests.cs | 434 ++++++++++++++++++ 8 files changed, 894 insertions(+), 11 deletions(-) create mode 100644 test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs diff --git a/hosting/Windows/Garnet.worker/Program.cs b/hosting/Windows/Garnet.worker/Program.cs index 8418da86716..165199109b0 100644 --- a/hosting/Windows/Garnet.worker/Program.cs +++ b/hosting/Windows/Garnet.worker/Program.cs @@ -1,16 +1,35 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT license. +using System; using Garnet; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; class Program { + // Data finalization (AOF commit / checkpoint) uses up to 15 seconds internally (see GarnetServer.FinalizeDataAsync). + // Add this buffer on top of the connection-drain timeout so the host shutdown budget covers the full shutdown sequence. + private const int DataFinalizationBufferSeconds = 20; + static void Main(string[] args) { + // Pre-parse only the shutdown-timeout argument so we can configure both + // the host shutdown budget and the Worker's connection-drain timeout from a single value. + var shutdownTimeoutSeconds = ParseShutdownTimeoutSeconds(args, defaultSeconds: 5); + var shutdownTimeout = TimeSpan.FromSeconds(shutdownTimeoutSeconds); + var builder = Host.CreateApplicationBuilder(args); - builder.Services.AddHostedService(_ => new Worker(args)); + + // Tell the .NET host (and the Windows SCM via WindowsServiceLifetime) how long to wait + // before forcibly killing the process. We add DataFinalizationBufferSeconds so that AOF + // commit / checkpoint can complete after connection draining finishes. + builder.Services.Configure(opts => + { + opts.ShutdownTimeout = shutdownTimeout + TimeSpan.FromSeconds(DataFinalizationBufferSeconds); + }); + + builder.Services.AddHostedService(_ => new Worker(args, shutdownTimeout)); builder.Services.AddWindowsService(options => { @@ -20,4 +39,22 @@ static void Main(string[] args) var host = builder.Build(); host.Run(); } -} \ No newline at end of file + + /// + /// Scans for --shutdown-timeout <value> and returns + /// the parsed integer, or if the argument is absent or invalid. + /// This lightweight pre-parse avoids a full CommandLineParser pass before the host is built. + /// + private static int ParseShutdownTimeoutSeconds(string[] args, int defaultSeconds) + { + for (var i = 0; i < args.Length - 1; i++) + { + if (args[i] is "--shutdown-timeout" or "-shutdown-timeout" && + int.TryParse(args[i + 1], out var value) && value > 0) + { + return value; + } + } + return defaultSeconds; + } +} diff --git a/hosting/Windows/Garnet.worker/Worker.cs b/hosting/Windows/Garnet.worker/Worker.cs index d69adb7e3c0..6b3c6059fd9 100644 --- a/hosting/Windows/Garnet.worker/Worker.cs +++ b/hosting/Windows/Garnet.worker/Worker.cs @@ -12,12 +12,20 @@ public class Worker : BackgroundService { private bool _isDisposed = false; private readonly string[] args; + private readonly TimeSpan _shutdownTimeout; private GarnetServer server; - public Worker(string[] args) + /// Command line arguments forwarded to . + /// + /// How long to wait for active connections to drain during graceful shutdown. + /// Must be less than the host + /// so that data finalization (AOF commit / checkpoint) can also complete within the host budget. + /// + public Worker(string[] args, TimeSpan shutdownTimeout) { this.args = args; + _shutdownTimeout = shutdownTimeout; } protected override async Task ExecuteAsync(CancellationToken stoppingToken) @@ -43,8 +51,26 @@ protected override async Task ExecuteAsync(CancellationToken stoppingToken) /// Indicates that the shutdown process should no longer be graceful. public override async Task StopAsync(CancellationToken cancellationToken) { - Dispose(); - await base.StopAsync(cancellationToken).ConfigureAwait(false); + try + { + if (server != null) + { + // If cancellation is requested, we will skip the graceful shutdown and proceed to dispose immediately + bool isForceShutdown = cancellationToken.IsCancellationRequested; + // Perform graceful shutdown with AOF commit and checkpoint when not forced Shutdown From OS. + await server.ShutdownAsync(timeout: _shutdownTimeout, noSave: isForceShutdown, token: cancellationToken).ConfigureAwait(false); + } + } + catch (OperationCanceledException) + { + // Force shutdown requested - proceed to dispose + } + finally + { + // Ensure base class cleanup although cancellationToken is cancelled + await base.StopAsync(CancellationToken.None).ConfigureAwait(false); + Dispose(); + } } public override void Dispose() @@ -55,6 +81,8 @@ public override void Dispose() } server?.Dispose(); _isDisposed = true; + base.Dispose(); + GC.SuppressFinalize(this); } } -} \ No newline at end of file +} diff --git a/libs/host/GarnetServer.cs b/libs/host/GarnetServer.cs index 50884165a88..6bf1cfe4c94 100644 --- a/libs/host/GarnetServer.cs +++ b/libs/host/GarnetServer.cs @@ -72,6 +72,11 @@ static string GetVersion() /// public StoreApi Store; + /// + /// Configured shutdown drain timeout in seconds. + /// + public int ShutdownTimeoutSeconds => opts.ShutdownTimeoutSeconds > 0 ? opts.ShutdownTimeoutSeconds : 5; + /// /// Create Garnet Server instance using specified command line arguments; use Start to start the server. /// @@ -733,7 +738,7 @@ private async Task FinalizeDataAsync(CancellationToken token) logger?.LogDebug("Taking checkpoint for tiered storage..."); try { - var checkpointSuccess = await storeWrapper.TakeCheckpointAsync(background: false, token: token, logger: logger).ConfigureAwait(false); + var checkpointSuccess = await Store.TakeCheckpointAsync(background: false, token: token).ConfigureAwait(false); if (checkpointSuccess) { logger?.LogDebug("Checkpoint completed successfully."); diff --git a/libs/server/Servers/StoreApi.cs b/libs/server/Servers/StoreApi.cs index 9ad5d9db3f0..8773b889179 100644 --- a/libs/server/Servers/StoreApi.cs +++ b/libs/server/Servers/StoreApi.cs @@ -93,6 +93,25 @@ public bool FlushDB(int dbId = 0, bool unsafeTruncateLog = false) } } + /// + /// Take checkpoint for all active databases + /// + /// True if method can return before checkpoint is taken + /// Cancellation token + /// false if checkpoint was skipped due to node state or another checkpoint in progress + public async ValueTask TakeCheckpointAsync(bool background = false, CancellationToken token = default) + { + using (PreventRoleChange(out var acquired)) + { + if (!acquired || IsReplica) + { + return false; + } + + return await storeWrapper.TakeCheckpointAsync(background, token: token).ConfigureAwait(false); + } + } + /// /// Helper to disable role changes during a using block. /// diff --git a/main/GarnetServer/Program.cs b/main/GarnetServer/Program.cs index 7b2673ebc41..5b37eec3acd 100644 --- a/main/GarnetServer/Program.cs +++ b/main/GarnetServer/Program.cs @@ -10,7 +10,7 @@ namespace Garnet /// public class Program { - static void Main(string[] args) + static async Task Main(string[] args) { try { @@ -22,7 +22,52 @@ static void Main(string[] args) // Start the server server.Start(); - Thread.Sleep(Timeout.Infinite); + using var cts = new CancellationTokenSource(); + + ConsoleCancelEventHandler cancelKeyPressHandler = (sender, e) => + { + e.Cancel = true; // Prevent the process from terminating immediately + if (!cts.IsCancellationRequested) + cts.Cancel(); + }; + + EventHandler processExitHandler = (sender, e) => + { + try + { + if (!cts.IsCancellationRequested) + cts.Cancel(); + } + catch (ObjectDisposedException) + { + // The cancellation source may already be disposed during process teardown. + } + }; + + // Signal cancellation on Ctrl+C; avoid blocking the event handler with async work + Console.CancelKeyPress += cancelKeyPressHandler; + + // Signal cancellation on SIGTERM (e.g., container orchestrators, systemd) + AppDomain.CurrentDomain.ProcessExit += processExitHandler; + + // Wait until a shutdown signal is received + try + { + try + { + await Task.Delay(Timeout.Infinite, cts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Graceful shutdown: drain connections, commit AOF, take checkpoint + await server.ShutdownAsync(TimeSpan.FromSeconds(server.ShutdownTimeoutSeconds), cts.Token).ConfigureAwait(false); + } + } + finally + { + Console.CancelKeyPress -= cancelKeyPressHandler; + AppDomain.CurrentDomain.ProcessExit -= processExitHandler; + } } catch (Exception ex) { @@ -94,4 +139,4 @@ static void RegisterExtensions(GarnetServer server) server.Register.NewProcedure("SETMAINANDOBJECT", () => new SetStringAndList()); } } -} \ No newline at end of file +} diff --git a/test/standalone/Garnet.test/GarnetServerTcpTests.cs b/test/standalone/Garnet.test/GarnetServerTcpTests.cs index 2a23d68bc43..ebd4fe55598 100644 --- a/test/standalone/Garnet.test/GarnetServerTcpTests.cs +++ b/test/standalone/Garnet.test/GarnetServerTcpTests.cs @@ -2,22 +2,45 @@ // Licensed under the MIT license. using System; +using System.Linq; using System.Net; using System.Net.Sockets; +using System.Threading; +using System.Threading.Tasks; using Garnet.server; using NUnit.Framework; +using NUnit.Framework.Legacy; +using StackExchange.Redis; namespace Garnet.test { [TestFixture, NonParallelizable] public class GarnetServerTcpTests : TestBase { + private GarnetServer server; + [TearDown] public void TearDown() { + server?.Dispose(); + server = null; TestUtils.OnTearDown(); } + /// + /// Starts a full Garnet server on . + /// Called explicitly by tests that need a live server; the listener-reuse tests below + /// bind raw instances to the same port and require it to be free. + /// + private void StartServer(bool enableAOF = false) + { + TestUtils.DeleteDirectory(TestUtils.MethodTestDir, wait: true); + server = TestUtils.CreateGarnetServer(TestUtils.MethodTestDir, enableAOF: enableAOF); + server.Start(); + } + + #region Listener reuse semantics + static IPEndPoint GetEndPoint(bool ipv6) { if (OperatingSystem.IsWindows()) @@ -80,5 +103,192 @@ public void StartReusesEndpointAfterListenerCloses(bool ipv6) Assert.DoesNotThrow(() => second.Start()); } + + #endregion + + #region Graceful shutdown + + [Test] + public void StopListeningPreventsNewConnections() + { + StartServer(); + + // Arrange - Establish a working connection first + using var redis1 = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + var db1 = redis1.GetDatabase(0); + db1.StringSet("test", "value"); + ClassicAssert.AreEqual("value", (string)db1.StringGet("test")); + + // Act - Stop listening on all servers + foreach (var tcpServer in server.Provider.StoreWrapper.Servers.OfType()) + { + tcpServer.StopListening(); + } + + Thread.Sleep(100); // Brief delay to ensure socket is closed + + // Assert - New connections should fail + Assert.Throws(() => + { + using var redis2 = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + redis2.GetDatabase(0).Ping(); + }); + + // Existing connection should still work + ClassicAssert.AreEqual("value", (string)db1.StringGet("test")); + } + + [Test] + public void StopListeningIdempotent() + { + StartServer(); + + // Arrange + foreach (var tcpServer in server.Provider.StoreWrapper.Servers.OfType()) + { + tcpServer.StopListening(); + } + + // Act & Assert - Calling StopListening again should not throw + Assert.DoesNotThrow(() => + { + foreach (var tcpServer in server.Provider.StoreWrapper.Servers.OfType()) + { + tcpServer.StopListening(); + } + }); + } + + [Test] + public async Task StopListeningDuringActiveConnectionAttempts() + { + StartServer(); + + // Arrange - Start multiple connection attempts + var connectionTasks = new System.Collections.Generic.List(); + using var cts = new CancellationTokenSource(); + + for (int i = 0; i < 10; i++) + { + connectionTasks.Add(Task.Run(async () => + { + while (!cts.Token.IsCancellationRequested) + { + try + { + using var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + await redis.GetDatabase(0).PingAsync(); + await Task.Delay(10); + } + catch + { + // Connection failures are expected after StopListening + } + } + }, cts.Token)); + } + + await Task.Delay(50); // Let some connections establish + + // Act + foreach (var tcpServer in server.Provider.StoreWrapper.Servers.OfType()) + { + tcpServer.StopListening(); + } + + await Task.Delay(100); + cts.Cancel(); + + // Assert - All tasks should complete without unhandled exceptions + Assert.DoesNotThrowAsync(async () => await Task.WhenAll(connectionTasks)); + } + + [Test] + public async Task ShutdownAsyncCompletesGracefully() + { + StartServer(); + + // Arrange - Write data and then close the connection + using (var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig())) + { + var db = redis.GetDatabase(0); + db.StringSet("shutdown-test", "data"); + ClassicAssert.AreEqual("data", (string)db.StringGet("shutdown-test")); + } + + // Act - Graceful shutdown (no active connections) + await server.ShutdownAsync(timeout: TimeSpan.FromSeconds(5)).ConfigureAwait(false); + + // Assert - New connections should fail after shutdown + Assert.Throws(() => + { + using var redis2 = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + redis2.GetDatabase(0).Ping(); + }); + } + + [Test] + public async Task ShutdownAsyncRespectsTimeout() + { + StartServer(); + + // Arrange - Establish a connection that will stay open + using var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + var db = redis.GetDatabase(0); + db.Ping(); + + // Act - Shutdown with a very short timeout + var sw = System.Diagnostics.Stopwatch.StartNew(); + await server.ShutdownAsync(timeout: TimeSpan.FromMilliseconds(200)).ConfigureAwait(false); + sw.Stop(); + + // Assert - Should complete without hanging indefinitely + // Allow generous upper bound for CI environments + ClassicAssert.Less(sw.ElapsedMilliseconds, 10_000, + "ShutdownAsync should complete within a reasonable time even with active connections"); + } + + [Test] + public async Task ShutdownAsyncRespectsCancellation() + { + StartServer(); + + // Arrange + using var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + redis.GetDatabase(0).Ping(); + + using var cts = new CancellationTokenSource(); + + // Act - Cancel immediately + cts.Cancel(); + Assert.DoesNotThrowAsync(async () => + { + await server.ShutdownAsync(timeout: TimeSpan.FromSeconds(30), token: cts.Token).ConfigureAwait(false); + }); + } + + [Test] + public async Task ShutdownAsyncWithAofCommit() + { + // Arrange - Create server with AOF enabled + StartServer(enableAOF: true); + + using var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + var db = redis.GetDatabase(0); + + // Write some data + for (int i = 0; i < 100; i++) + { + db.StringSet($"aof-key-{i}", $"value-{i}"); + } + + // Act - Shutdown should commit AOF without errors + Assert.DoesNotThrowAsync(async () => + { + await server.ShutdownAsync(timeout: TimeSpan.FromSeconds(5)).ConfigureAwait(false); + }); + } + + #endregion } -} \ No newline at end of file +} diff --git a/test/standalone/Garnet.test/RespAdminCommandsTests.cs b/test/standalone/Garnet.test/RespAdminCommandsTests.cs index 46e452c91ac..e21efa24ced 100644 --- a/test/standalone/Garnet.test/RespAdminCommandsTests.cs +++ b/test/standalone/Garnet.test/RespAdminCommandsTests.cs @@ -779,5 +779,110 @@ public void ConfigGetWrongNumberOfArguments() ClassicAssert.AreEqual(expectedMessage, ex.Message); } #endregion + + #region GracefulShutdownTests + + [Test] + public async Task ShutdownAsyncStopsAcceptingNewConnections() + { + // Arrange - write data then close connection + using (var redis1 = ConnectionMultiplexer.Connect(TestUtils.GetConfig())) + { + var db1 = redis1.GetDatabase(0); + db1.StringSet("test", "value"); + } + + // Act - Initiate shutdown + await server.ShutdownAsync(TimeSpan.FromSeconds(5)); + + // Assert - New connections should fail + Assert.Throws(() => + { + using var redis2 = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + redis2.GetDatabase(0).Ping(); + }); + } + + [Test] + public async Task ShutdownAsyncWaitsForActiveConnections() + { + // Arrange - keep connection open to test timeout behavior + using var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + var db = redis.GetDatabase(0); + db.StringSet("key1", "value1"); + + // Act - Start shutdown with short timeout; active connection will force timeout + var sw = System.Diagnostics.Stopwatch.StartNew(); + await server.ShutdownAsync(TimeSpan.FromMilliseconds(500)); + sw.Stop(); + + // Assert - Should wait approximately the timeout duration before proceeding + ClassicAssert.GreaterOrEqual(sw.ElapsedMilliseconds, 400, + "Shutdown should wait for the timeout duration when connections are active"); + ClassicAssert.Less(sw.ElapsedMilliseconds, 5_000, + "Shutdown should not hang beyond a reasonable bound"); + } + + [Test] + public async Task ShutdownAsyncCommitsAOF() + { + // Arrange + server.Dispose(); + server = TestUtils.CreateGarnetServer(TestUtils.MethodTestDir, enableAOF: true); + server.Start(); + + using (var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig(allowAdmin: true))) + { + var db = redis.GetDatabase(0); + db.StringSet("aofKey", "aofValue"); + } + + // Act - Shutdown which should commit AOF + await server.ShutdownAsync(TimeSpan.FromSeconds(5)); + server.Dispose(false); + + // Assert - Recover and verify data persisted + server = TestUtils.CreateGarnetServer(TestUtils.MethodTestDir, enableAOF: true, tryRecover: true); + server.Start(); + + using (var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig())) + { + var db = redis.GetDatabase(0); + var recoveredValue = db.StringGet("aofKey"); + ClassicAssert.AreEqual("aofValue", recoveredValue.ToString()); + } + } + + [Test] + public async Task ShutdownAsyncTakesCheckpointWhenStorageTierEnabled() + { + // Arrange + server.Dispose(); + server = TestUtils.CreateGarnetServer(TestUtils.MethodTestDir); + server.Start(); + + using (var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig(allowAdmin: true))) + { + var db = redis.GetDatabase(0); + db.StringSet("checkpointKey", "checkpointValue"); + } + + // Act - Shutdown which should take checkpoint + await server.ShutdownAsync(TimeSpan.FromSeconds(5)); + server.Dispose(false); + + // Assert - Recover from checkpoint + server = TestUtils.CreateGarnetServer(TestUtils.MethodTestDir, tryRecover: true); + server.Start(); + + using (var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig())) + { + var db = redis.GetDatabase(0); + var recoveredValue = db.StringGet("checkpointKey"); + ClassicAssert.AreEqual("checkpointValue", recoveredValue.ToString()); + } + } + + #endregion } } \ No newline at end of file diff --git a/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs new file mode 100644 index 00000000000..9382cbf3b8c --- /dev/null +++ b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs @@ -0,0 +1,434 @@ +// Copyright (c) Microsoft Corporation. +// Licensed under the MIT license. + +using System; +using System.Threading; +using System.Threading.Tasks; +using NUnit.Framework; +using NUnit.Framework.Legacy; +using StackExchange.Redis; + +namespace Garnet.test +{ + /// + /// Tests to investigate data consistency under different shutdown finalization sequences: + /// 1. Checkpoint first, then AOF commit + /// 2. AOF commit first, then Checkpoint (current production order) + /// 3. AOF commit only + /// 4. Checkpoint only + /// + /// Each test writes data to main store (string keys) and object store (sorted sets), + /// performs the finalization sequence, disposes without cleanup, recovers, and verifies data. + /// + [TestFixture] + public class ShutdownDataConsistencyTests : TestBase + { + private GarnetServer server; + + private const int KeyCount = 50; + private const string MainStoreKeyPrefix = "shutdowntest:key:"; + private const string MainStoreValuePrefix = "shutdowntest:value:"; + private const string ObjectStoreKeyPrefix = "shutdowntest:zset:"; + + [SetUp] + public void Setup() + { + TestUtils.DeleteDirectory(TestUtils.MethodTestDir, wait: true); + } + + [TearDown] + public void TearDown() + { + server?.Dispose(); + TestUtils.DeleteDirectory(TestUtils.MethodTestDir); + } + + /// + /// Creates a server with both AOF and storage tier enabled (low memory forces spill to disk). + /// + private GarnetServer CreateServerWithAofAndStorage(bool tryRecover = false) + { + return TestUtils.CreateGarnetServer( + TestUtils.MethodTestDir, + enableAOF: true, + lowMemory: true, + tryRecover: tryRecover); + } + + /// + /// Populates main store with string key-value pairs and object store with sorted sets. + /// + private void PopulateData() + { + using var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig(allowAdmin: true)); + var db = redis.GetDatabase(0); + + // Main store: string keys + for (var i = 0; i < KeyCount; i++) + { + db.StringSet($"{MainStoreKeyPrefix}{i}", $"{MainStoreValuePrefix}{i}"); + } + + // Object store: sorted sets + for (var i = 0; i < KeyCount; i++) + { + var entries = new SortedSetEntry[] + { + new($"member_a_{i}", i * 10), + new($"member_b_{i}", i * 10 + 1), + new($"member_c_{i}", i * 10 + 2), + }; + db.SortedSetAdd($"{ObjectStoreKeyPrefix}{i}", entries); + } + } + + /// + /// Verifies all main store and object store data is recovered correctly. + /// Returns (mainStoreRecovered, objectStoreRecovered) counts. + /// + private (int mainStoreRecovered, int objectStoreRecovered) VerifyRecoveredData() + { + using var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + var db = redis.GetDatabase(0); + + var mainStoreRecovered = 0; + for (var i = 0; i < KeyCount; i++) + { + var value = db.StringGet($"{MainStoreKeyPrefix}{i}"); + if (value.HasValue) + { + ClassicAssert.AreEqual($"{MainStoreValuePrefix}{i}", value.ToString(), + $"Main store key {MainStoreKeyPrefix}{i} has wrong value after recovery"); + mainStoreRecovered++; + } + } + + var objectStoreRecovered = 0; + for (var i = 0; i < KeyCount; i++) + { + var members = db.SortedSetRangeByScoreWithScores($"{ObjectStoreKeyPrefix}{i}"); + if (members.Length > 0) + { + ClassicAssert.AreEqual(3, members.Length, + $"Object store key {ObjectStoreKeyPrefix}{i} should have 3 members"); + ClassicAssert.AreEqual($"member_a_{i}", members[0].Element.ToString()); + ClassicAssert.AreEqual(i * 10, members[0].Score); + ClassicAssert.AreEqual($"member_b_{i}", members[1].Element.ToString()); + ClassicAssert.AreEqual(i * 10 + 1, members[1].Score); + ClassicAssert.AreEqual($"member_c_{i}", members[2].Element.ToString()); + ClassicAssert.AreEqual(i * 10 + 2, members[2].Score); + objectStoreRecovered++; + } + } + + return (mainStoreRecovered, objectStoreRecovered); + } + + /// + /// Scenario 1: Checkpoint → AOF commit sequence. + /// Takes checkpoint first, then commits AOF. + /// + [Test] + public async Task CheckpointThenAofCommit_DataConsistencyTest() + { + server = CreateServerWithAofAndStorage(); + server.Start(); + + PopulateData(); + + // Create CancellationTokenSource with a long timeout to prevent hanging indefinitely in case of issues during shutdown + var cts = new CancellationTokenSource(TimeSpan.FromMinutes(5)); + TestContext.Progress.WriteLine($"Created CancellationTokenSource with 5 minute timeout for test shutdown"); + + // Sequence: Checkpoint first, then AOF commit + await server.Store.TakeCheckpointAsync(background: false); + await server.Store.CommitAOFAsync(cts.Token); + + server.Dispose(false); + + // Recover and verify + server = CreateServerWithAofAndStorage(tryRecover: true); + server.Start(); + + var (mainRecovered, objRecovered) = VerifyRecoveredData(); + + TestContext.Progress.WriteLine( + $"[Checkpoint→AOF] Main store: {mainRecovered}/{KeyCount}, Object store: {objRecovered}/{KeyCount}"); + + ClassicAssert.AreEqual(KeyCount, mainRecovered, + "Checkpoint→AOF: Not all main store keys recovered"); + ClassicAssert.AreEqual(KeyCount, objRecovered, + "Checkpoint→AOF: Not all object store keys recovered"); + } + + /// + /// Scenario 2: AOF commit → Checkpoint sequence (current production order). + /// Commits AOF first, then takes checkpoint. + /// + [Test] + public async Task AofCommitThenCheckpoint_DataConsistencyTest() + { + server = CreateServerWithAofAndStorage(); + server.Start(); + + PopulateData(); + + // Create CancellationTokenSource with a long timeout to prevent hanging indefinitely in case of issues during shutdown + var cts = new CancellationTokenSource(TimeSpan.FromMinutes(5)); + TestContext.Progress.WriteLine($"Created CancellationTokenSource with 5 minute timeout for test shutdown"); + + // Sequence: AOF commit first, then Checkpoint (matches current FinalizeDataAsync order) + await server.Store.CommitAOFAsync(cts.Token); + await server.Store.TakeCheckpointAsync(background: false); + + server.Dispose(false); + + // Recover and verify + server = CreateServerWithAofAndStorage(tryRecover: true); + server.Start(); + + var (mainRecovered, objRecovered) = VerifyRecoveredData(); + + TestContext.Progress.WriteLine( + $"[AOF→Checkpoint] Main store: {mainRecovered}/{KeyCount}, Object store: {objRecovered}/{KeyCount}"); + + ClassicAssert.AreEqual(KeyCount, mainRecovered, + "AOF→Checkpoint: Not all main store keys recovered"); + ClassicAssert.AreEqual(KeyCount, objRecovered, + "AOF→Checkpoint: Not all object store keys recovered"); + } + + /// + /// Scenario 3: AOF commit only (no checkpoint). + /// Only commits AOF before shutdown. + /// + [Test] + public async Task AofCommitOnly_DataConsistencyTest() + { + server = CreateServerWithAofAndStorage(); + server.Start(); + + PopulateData(); + + var cts = new CancellationTokenSource(TimeSpan.FromMinutes(5)); + + // Sequence: AOF commit only + await server.Store.CommitAOFAsync(cts.Token); + + server.Dispose(false); + + // Recover and verify + server = CreateServerWithAofAndStorage(tryRecover: true); + server.Start(); + + var (mainRecovered, objRecovered) = VerifyRecoveredData(); + + TestContext.Progress.WriteLine( + $"[AOF Only] Main store: {mainRecovered}/{KeyCount}, Object store: {objRecovered}/{KeyCount}"); + + ClassicAssert.AreEqual(KeyCount, mainRecovered, + "AOF Only: Not all main store keys recovered"); + ClassicAssert.AreEqual(KeyCount, objRecovered, + "AOF Only: Not all object store keys recovered"); + } + + /// + /// Scenario 4: Checkpoint only (no AOF commit). + /// Only takes checkpoint before shutdown. + /// + [Test] + public async Task CheckpointOnly_DataConsistencyTest() + { + server = CreateServerWithAofAndStorage(); + server.Start(); + + PopulateData(); + + // Sequence: Checkpoint only (no AOF commit) + await server.Store.TakeCheckpointAsync(background: false); + + server.Dispose(false); + + // Recover and verify + server = CreateServerWithAofAndStorage(tryRecover: true); + server.Start(); + + var (mainRecovered, objRecovered) = VerifyRecoveredData(); + + TestContext.Progress.WriteLine( + $"[Checkpoint Only] Main store: {mainRecovered}/{KeyCount}, Object store: {objRecovered}/{KeyCount}"); + + // Note: With checkpoint only (no AOF commit), data written after the last + // checkpoint but before the AOF commit may be lost. This test documents + // the actual behavior for investigation purposes. + ClassicAssert.AreEqual(KeyCount, mainRecovered, + "Checkpoint Only: Not all main store keys recovered"); + ClassicAssert.AreEqual(KeyCount, objRecovered, + "Checkpoint Only: Not all object store keys recovered"); + } + + /// + /// Scenario 5: No finalization at all (baseline - result is configuration-dependent). + /// Neither AOF commit nor checkpoint before shutdown. + /// Recovery results may vary if data was already persisted by background AOF or storage tier spill. + /// + [Test] + public void NoFinalization_DataConsistencyTest() + { + server = CreateServerWithAofAndStorage(); + server.Start(); + + PopulateData(); + + // No explicit finalization before shutdown + server.Dispose(false); + + // Recover and verify + server = CreateServerWithAofAndStorage(tryRecover: true); + server.Start(); + + var (mainRecovered, objRecovered) = VerifyRecoveredData(); + + TestContext.Progress.WriteLine( + $"[No Finalization] Main store: {mainRecovered}/{KeyCount}, Object store: {objRecovered}/{KeyCount}"); + + // Baseline observation only: recovery depends on prior persistence behavior. + TestContext.Progress.WriteLine( + $"[No Finalization] Data loss: main store={KeyCount - mainRecovered}, object store={KeyCount - objRecovered}"); + } + + /// + /// Scenario 6: Interleaved writes with checkpoint then additional writes with AOF commit. + /// Simulates the case where new writes happen between checkpoint and AOF commit. + /// + [Test] + public async Task CheckpointThenMoreWritesThenAofCommit_DataConsistencyTest() + { + server = CreateServerWithAofAndStorage(); + server.Start(); + + // Phase 1: Initial data + PopulateData(); + + var cts = new CancellationTokenSource(TimeSpan.FromMinutes(5)); + + // Take checkpoint + await server.Store.TakeCheckpointAsync(background: false); + + // Phase 2: Write additional data AFTER checkpoint + using (var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig())) + { + var db = redis.GetDatabase(0); + for (var i = KeyCount; i < KeyCount * 2; i++) + { + db.StringSet($"{MainStoreKeyPrefix}{i}", $"{MainStoreValuePrefix}{i}"); + } + } + + // Now commit AOF (should capture phase 2 writes) + await server.Store.CommitAOFAsync(cts.Token); + + server.Dispose(false); + + // Recover and verify + server = CreateServerWithAofAndStorage(tryRecover: true); + server.Start(); + + using (var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig())) + { + var db = redis.GetDatabase(0); + + var phase1Recovered = 0; + for (var i = 0; i < KeyCount; i++) + { + var value = db.StringGet($"{MainStoreKeyPrefix}{i}"); + if (value.HasValue && value.ToString() == $"{MainStoreValuePrefix}{i}") + phase1Recovered++; + } + + var phase2Recovered = 0; + for (var i = KeyCount; i < KeyCount * 2; i++) + { + var value = db.StringGet($"{MainStoreKeyPrefix}{i}"); + if (value.HasValue && value.ToString() == $"{MainStoreValuePrefix}{i}") + phase2Recovered++; + } + + TestContext.Progress.WriteLine( + $"[Checkpoint→Writes→AOF] Phase1: {phase1Recovered}/{KeyCount}, Phase2 (post-checkpoint): {phase2Recovered}/{KeyCount}"); + + ClassicAssert.AreEqual(KeyCount, phase1Recovered, + "Checkpoint→Writes→AOF: Not all phase 1 keys recovered"); + ClassicAssert.AreEqual(KeyCount, phase2Recovered, + "Checkpoint→Writes→AOF: Not all phase 2 (post-checkpoint) keys recovered"); + } + } + + /// + /// Scenario 7: AOF commit then additional writes then checkpoint. + /// Simulates the case where new writes happen between AOF commit and checkpoint. + /// + [Test] + public async Task AofCommitThenMoreWritesThenCheckpoint_DataConsistencyTest() + { + server = CreateServerWithAofAndStorage(); + server.Start(); + + // Phase 1: Initial data + PopulateData(); + + var cts = new CancellationTokenSource(TimeSpan.FromMinutes(5)); + + // Commit AOF + await server.Store.CommitAOFAsync(cts.Token); + + // Phase 2: Write additional data AFTER AOF commit + using (var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig())) + { + var db = redis.GetDatabase(0); + for (var i = KeyCount; i < KeyCount * 2; i++) + { + db.StringSet($"{MainStoreKeyPrefix}{i}", $"{MainStoreValuePrefix}{i}"); + } + } + + // Now take checkpoint (should capture phase 2 writes) + await server.Store.TakeCheckpointAsync(background: false); + + server.Dispose(false); + + // Recover and verify + server = CreateServerWithAofAndStorage(tryRecover: true); + server.Start(); + + using (var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig())) + { + var db = redis.GetDatabase(0); + + var phase1Recovered = 0; + for (var i = 0; i < KeyCount; i++) + { + var value = db.StringGet($"{MainStoreKeyPrefix}{i}"); + if (value.HasValue && value.ToString() == $"{MainStoreValuePrefix}{i}") + phase1Recovered++; + } + + var phase2Recovered = 0; + for (var i = KeyCount; i < KeyCount * 2; i++) + { + var value = db.StringGet($"{MainStoreKeyPrefix}{i}"); + if (value.HasValue && value.ToString() == $"{MainStoreValuePrefix}{i}") + phase2Recovered++; + } + + TestContext.Progress.WriteLine( + $"[AOF→Writes→Checkpoint] Phase1: {phase1Recovered}/{KeyCount}, Phase2 (post-AOF): {phase2Recovered}/{KeyCount}"); + + ClassicAssert.AreEqual(KeyCount, phase1Recovered, + "AOF→Writes→Checkpoint: Not all phase 1 keys recovered"); + ClassicAssert.AreEqual(KeyCount, phase2Recovered, + "AOF→Writes→Checkpoint: Not all phase 2 (post-AOF) keys recovered"); + } + } + } +} From b7209ece22cae503d6d445c4d9d95742f14ec1e9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Tue, 26 May 2026 17:06:49 +0900 Subject: [PATCH 05/14] formmating --- hosting/Windows/Garnet.worker/Program.cs | 2 +- hosting/Windows/Garnet.worker/Worker.cs | 2 +- main/GarnetServer/Program.cs | 4 ++-- test/standalone/Garnet.test/GarnetServerTcpTests.cs | 2 +- test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs | 2 +- 5 files changed, 6 insertions(+), 6 deletions(-) diff --git a/hosting/Windows/Garnet.worker/Program.cs b/hosting/Windows/Garnet.worker/Program.cs index 165199109b0..2ad736b8c53 100644 --- a/hosting/Windows/Garnet.worker/Program.cs +++ b/hosting/Windows/Garnet.worker/Program.cs @@ -57,4 +57,4 @@ private static int ParseShutdownTimeoutSeconds(string[] args, int defaultSeconds } return defaultSeconds; } -} +} \ No newline at end of file diff --git a/hosting/Windows/Garnet.worker/Worker.cs b/hosting/Windows/Garnet.worker/Worker.cs index 6b3c6059fd9..2de54e52200 100644 --- a/hosting/Windows/Garnet.worker/Worker.cs +++ b/hosting/Windows/Garnet.worker/Worker.cs @@ -85,4 +85,4 @@ public override void Dispose() GC.SuppressFinalize(this); } } -} +} \ No newline at end of file diff --git a/main/GarnetServer/Program.cs b/main/GarnetServer/Program.cs index 5b37eec3acd..e8abea37d2d 100644 --- a/main/GarnetServer/Program.cs +++ b/main/GarnetServer/Program.cs @@ -60,7 +60,7 @@ static async Task Main(string[] args) catch (OperationCanceledException) { // Graceful shutdown: drain connections, commit AOF, take checkpoint - await server.ShutdownAsync(TimeSpan.FromSeconds(server.ShutdownTimeoutSeconds), cts.Token).ConfigureAwait(false); + await server.ShutdownAsync(TimeSpan.FromSeconds(server.ShutdownTimeoutSeconds), token: cts.Token).ConfigureAwait(false); } } finally @@ -139,4 +139,4 @@ static void RegisterExtensions(GarnetServer server) server.Register.NewProcedure("SETMAINANDOBJECT", () => new SetStringAndList()); } } -} +} \ No newline at end of file diff --git a/test/standalone/Garnet.test/GarnetServerTcpTests.cs b/test/standalone/Garnet.test/GarnetServerTcpTests.cs index ebd4fe55598..2b4e21beba3 100644 --- a/test/standalone/Garnet.test/GarnetServerTcpTests.cs +++ b/test/standalone/Garnet.test/GarnetServerTcpTests.cs @@ -291,4 +291,4 @@ public async Task ShutdownAsyncWithAofCommit() #endregion } -} +} \ No newline at end of file diff --git a/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs index 9382cbf3b8c..26ec61fc3bf 100644 --- a/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs +++ b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs @@ -431,4 +431,4 @@ public async Task AofCommitThenMoreWritesThenCheckpoint_DataConsistencyTest() } } } -} +} \ No newline at end of file From a25810cec0cb01dde46f368ed0ab66a7378a3072 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Wed, 27 May 2026 09:25:34 +0900 Subject: [PATCH 06/14] Mark as Explicit test: ShutdownDataConsistencyTests To follow copilot's Suggestion, this test didn't require on CI just to keep for infomational test for develope https://github.com/microsoft/garnet/pull/1825#discussion_r3303117735 --- test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs | 1 + 1 file changed, 1 insertion(+) diff --git a/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs index 26ec61fc3bf..9cf039f5f7c 100644 --- a/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs +++ b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs @@ -21,6 +21,7 @@ namespace Garnet.test /// performs the finalization sequence, disposes without cleanup, recovers, and verifies data. /// [TestFixture] + [Explicit] public class ShutdownDataConsistencyTests : TestBase { private GarnetServer server; From 02dddf81d7d21b860abd2076ef7143e24a014464 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Wed, 27 May 2026 10:07:21 +0900 Subject: [PATCH 07/14] Fix by Copilot's suggestion Co-Authored-By: Copilot <198982749+Copilot@users.noreply.github.com> --- main/GarnetServer/Program.cs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/main/GarnetServer/Program.cs b/main/GarnetServer/Program.cs index e8abea37d2d..768f572e2d0 100644 --- a/main/GarnetServer/Program.cs +++ b/main/GarnetServer/Program.cs @@ -60,7 +60,10 @@ static async Task Main(string[] args) catch (OperationCanceledException) { // Graceful shutdown: drain connections, commit AOF, take checkpoint - await server.ShutdownAsync(TimeSpan.FromSeconds(server.ShutdownTimeoutSeconds), token: cts.Token).ConfigureAwait(false); + await server.ShutdownAsync( + TimeSpan.FromSeconds(server.ShutdownTimeoutSeconds), + token: CancellationToken.None + ).ConfigureAwait(false); } } finally From 797596140b9d9fedb57ecdb1ce304b7d8a54b852 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Wed, 27 May 2026 10:15:34 +0900 Subject: [PATCH 08/14] Fix finalizeData By copilot's suggestion I fixed to if AOF and checkpoint both enabled then we need to try if aof failed then checkpoint https://github.com/microsoft/garnet/pull/1825#discussion_r3303117692 --- libs/host/GarnetServer.cs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/libs/host/GarnetServer.cs b/libs/host/GarnetServer.cs index 6bf1cfe4c94..a6e8db0d0fa 100644 --- a/libs/host/GarnetServer.cs +++ b/libs/host/GarnetServer.cs @@ -706,6 +706,8 @@ private long GetActiveConnectionCount() /// /// Persists data during shutdown using AOF or checkpoint based on configuration. + /// Both Enabled AOF and checkpoint, but AOF is prioritized. + /// IF AOF commit failed, checkpoint will be performed. /// private async Task FinalizeDataAsync(CancellationToken token) { @@ -718,6 +720,7 @@ private async Task FinalizeDataAsync(CancellationToken token) if (commitSuccess) { logger?.LogDebug("AOF committed successfully."); + return; // skip checkpoint } else { @@ -728,8 +731,6 @@ private async Task FinalizeDataAsync(CancellationToken token) { logger?.LogError(ex, "Error committing AOF during shutdown"); } - - return; } if (!opts.EnableStorageTier) @@ -752,6 +753,7 @@ private async Task FinalizeDataAsync(CancellationToken token) { logger?.LogError(ex, "Error taking checkpoint during shutdown"); } + return; } /// From 45a8a5f21c2b57506b2c802947b039b1acd37c4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Wed, 27 May 2026 10:22:11 +0900 Subject: [PATCH 09/14] Fix resource leak on Test codes by copilot https: //github.com/microsoft/garnet/pull/1825#discussion_r3303117760 Co-Authored-By: Copilot <198982749+Copilot@users.noreply.github.com> --- test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs index 9cf039f5f7c..fbaa2fc8bbc 100644 --- a/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs +++ b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs @@ -138,7 +138,7 @@ public async Task CheckpointThenAofCommit_DataConsistencyTest() PopulateData(); // Create CancellationTokenSource with a long timeout to prevent hanging indefinitely in case of issues during shutdown - var cts = new CancellationTokenSource(TimeSpan.FromMinutes(5)); + using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(5)); TestContext.Progress.WriteLine($"Created CancellationTokenSource with 5 minute timeout for test shutdown"); // Sequence: Checkpoint first, then AOF commit From 2c49daf396235ecfbe0a6e3ce0fa526499c414c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Wed, 27 May 2026 10:40:49 +0900 Subject: [PATCH 10/14] Fix Test: StoplisteningDuringActiveConnectionAttempts To reflect Copilot's suggestion https://github.com/microsoft/garnet/pull/1825#discussion_r3303117784 and cancelation request will break test with new introduced catch blcok --- test/standalone/Garnet.test/GarnetServerTcpTests.cs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/test/standalone/Garnet.test/GarnetServerTcpTests.cs b/test/standalone/Garnet.test/GarnetServerTcpTests.cs index 2b4e21beba3..98566312201 100644 --- a/test/standalone/Garnet.test/GarnetServerTcpTests.cs +++ b/test/standalone/Garnet.test/GarnetServerTcpTests.cs @@ -180,12 +180,16 @@ public async Task StopListeningDuringActiveConnectionAttempts() await redis.GetDatabase(0).PingAsync(); await Task.Delay(10); } + catch (OperationCanceledException) when (cts.Token.IsCancellationRequested) + { + break; + } catch { // Connection failures are expected after StopListening } } - }, cts.Token)); + })); } await Task.Delay(50); // Let some connections establish From a13a12cd8d9d87952b4b4354b550782b7c09b037 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Wed, 27 May 2026 10:43:18 +0900 Subject: [PATCH 11/14] Update ShutdownAsync use configured timeout : to reflect copilot suggestion https: //github.com/microsoft/garnet/pull/1825#discussion_r3303117800 Co-Authored-By: Copilot <198982749+Copilot@users.noreply.github.com> --- libs/host/GarnetServer.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/libs/host/GarnetServer.cs b/libs/host/GarnetServer.cs index a6e8db0d0fa..c6184586eaa 100644 --- a/libs/host/GarnetServer.cs +++ b/libs/host/GarnetServer.cs @@ -552,13 +552,13 @@ public void Start() /// Performs graceful shutdown of the server. /// Stops accepting new connections, waits for active connections to complete, commits AOF, and takes checkpoint if needed. /// - /// Timeout for waiting on active connections (default: 30 seconds) + /// Timeout for waiting on active connections (default: configured value) /// If true, skip data persistence (AOF commit and checkpoint) during shutdown /// Cancellation token /// Task representing the async shutdown operation public async Task ShutdownAsync(TimeSpan? timeout = null, bool noSave = false, CancellationToken token = default) { - var shutdownTimeout = timeout ?? TimeSpan.FromSeconds(30); + var shutdownTimeout = timeout ?? TimeSpan.FromSeconds(ShutdownTimeoutSeconds); try { From 7ba6acf82f0cce0bc140227709f9af02e0a9192e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Wed, 27 May 2026 11:13:09 +0900 Subject: [PATCH 12/14] Centralize shutdown data-finalization timeout and add forced-shutdown path Add configurable data-finalization timeout and wire it through CLI, defaults.conf, and GarnetServerOptions so shutdown behavior stays consistent across hosts. - Add --data-finalization-timeout / DataFinalizationTimeoutSeconds (default 15s) - Replace hardcoded 15s finalize CTS in ShutdownAsync with linked caller token plus configured cap; skip persistence on noSave or forced cancellation - Define defaults on GarnetServerOptions (DefaultShutdownTimeoutSeconds, DefaultDataFinalizationTimeoutSeconds) as single source of truth - Windows service: HostOptions.ShutdownTimeout = drain + finalize + 5s margin (replaces fixed 20s buffer); pre-parse both timeout flags - Console host: second Ctrl+C / ProcessExit during shutdown cancels persistence (Redis-like fast path) via shutdownForceCts - Tests: forced cancellation, noSave, and AOF shutdown coverage; fix CS1998 by using void tests with Assert.DoesNotThrowAsync Co-authored-by: Cursor --- hosting/Windows/Garnet.worker/Program.cs | 29 ++++++------- hosting/Windows/Garnet.worker/Worker.cs | 4 +- libs/host/Configuration/Options.cs | 5 +++ libs/host/GarnetServer.cs | 42 ++++++++++++++----- libs/host/defaults.conf | 3 ++ libs/server/Servers/GarnetServerOptions.cs | 17 +++++++- main/GarnetServer/Program.cs | 34 ++++++++------- .../Garnet.test/GarnetServerTcpTests.cs | 24 +++++++++-- 8 files changed, 112 insertions(+), 46 deletions(-) diff --git a/hosting/Windows/Garnet.worker/Program.cs b/hosting/Windows/Garnet.worker/Program.cs index 2ad736b8c53..929047c0389 100644 --- a/hosting/Windows/Garnet.worker/Program.cs +++ b/hosting/Windows/Garnet.worker/Program.cs @@ -3,30 +3,32 @@ using System; using Garnet; +using Garnet.server; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; class Program { - // Data finalization (AOF commit / checkpoint) uses up to 15 seconds internally (see GarnetServer.FinalizeDataAsync). - // Add this buffer on top of the connection-drain timeout so the host shutdown budget covers the full shutdown sequence. - private const int DataFinalizationBufferSeconds = 20; + // Extra host budget beyond connection drain + data finalization (quiesce, stop listening, etc.). + private const int HostShutdownMarginSeconds = 5; static void Main(string[] args) { - // Pre-parse only the shutdown-timeout argument so we can configure both - // the host shutdown budget and the Worker's connection-drain timeout from a single value. - var shutdownTimeoutSeconds = ParseShutdownTimeoutSeconds(args, defaultSeconds: 5); + // Pre-parse shutdown-related arguments so we can configure both the host shutdown budget + // and the Worker's connection-drain timeout before the generic host is built. + var shutdownTimeoutSeconds = ParseIntOption(args, "--shutdown-timeout", "-shutdown-timeout", GarnetServerOptions.DefaultShutdownTimeoutSeconds); + var dataFinalizationTimeoutSeconds = ParseIntOption(args, "--data-finalization-timeout", "-data-finalization-timeout", GarnetServerOptions.DefaultDataFinalizationTimeoutSeconds); var shutdownTimeout = TimeSpan.FromSeconds(shutdownTimeoutSeconds); var builder = Host.CreateApplicationBuilder(args); // Tell the .NET host (and the Windows SCM via WindowsServiceLifetime) how long to wait - // before forcibly killing the process. We add DataFinalizationBufferSeconds so that AOF - // commit / checkpoint can complete after connection draining finishes. + // before forcibly killing the process: drain + data finalization + margin. builder.Services.Configure(opts => { - opts.ShutdownTimeout = shutdownTimeout + TimeSpan.FromSeconds(DataFinalizationBufferSeconds); + opts.ShutdownTimeout = shutdownTimeout + + TimeSpan.FromSeconds(dataFinalizationTimeoutSeconds) + + TimeSpan.FromSeconds(HostShutdownMarginSeconds); }); builder.Services.AddHostedService(_ => new Worker(args, shutdownTimeout)); @@ -41,15 +43,14 @@ static void Main(string[] args) } /// - /// Scans for --shutdown-timeout <value> and returns - /// the parsed integer, or if the argument is absent or invalid. - /// This lightweight pre-parse avoids a full CommandLineParser pass before the host is built. + /// Scans for (or ) and returns + /// the parsed positive integer, or if the argument is absent or invalid. /// - private static int ParseShutdownTimeoutSeconds(string[] args, int defaultSeconds) + private static int ParseIntOption(string[] args, string longName, string shortName, int defaultSeconds) { for (var i = 0; i < args.Length - 1; i++) { - if (args[i] is "--shutdown-timeout" or "-shutdown-timeout" && + if (args[i] is var name && (name == longName || name == shortName) && int.TryParse(args[i + 1], out var value) && value > 0) { return value; diff --git a/hosting/Windows/Garnet.worker/Worker.cs b/hosting/Windows/Garnet.worker/Worker.cs index 2de54e52200..d711974d13f 100644 --- a/hosting/Windows/Garnet.worker/Worker.cs +++ b/hosting/Windows/Garnet.worker/Worker.cs @@ -19,8 +19,8 @@ public class Worker : BackgroundService /// Command line arguments forwarded to . /// /// How long to wait for active connections to drain during graceful shutdown. - /// Must be less than the host - /// so that data finalization (AOF commit / checkpoint) can also complete within the host budget. + /// Must fit within the host , + /// which also budgets for data finalization and a small margin (see Program.cs). /// public Worker(string[] args, TimeSpan shutdownTimeout) { diff --git a/libs/host/Configuration/Options.cs b/libs/host/Configuration/Options.cs index 996a7e025df..4bb77988251 100644 --- a/libs/host/Configuration/Options.cs +++ b/libs/host/Configuration/Options.cs @@ -407,6 +407,10 @@ internal sealed class Options : ICloneable "The Windows SCM default pre-kill wait is 5 seconds, so values below 5 are not recommended when running as a Windows service.")] public int ShutdownTimeoutSeconds { get; set; } + [IntRangeValidation(1, int.MaxValue)] + [Option("data-finalization-timeout", Required = false, HelpText = "Timeout in seconds for AOF commit and checkpoint during graceful shutdown data finalization.")] + public int DataFinalizationTimeoutSeconds { get; set; } + [OptionValidation] [Option("use-azure-storage", Required = false, HelpText = "Use Azure Page Blobs for storage instead of local storage.")] public bool? UseAzureStorage { get; set; } @@ -990,6 +994,7 @@ endpoint is IPEndPoint listenEp && clusterAnnounceEndpoint[0] is IPEndPoint anno ThreadPoolMaxIOCompletionThreads = ThreadPoolMaxIOCompletionThreads, NetworkConnectionLimit = NetworkConnectionLimit, ShutdownTimeoutSeconds = ShutdownTimeoutSeconds, + DataFinalizationTimeoutSeconds = DataFinalizationTimeoutSeconds, DeviceFactoryCreator = deviceType == DeviceType.AzureStorage ? azureFactoryCreator() : new LocalStorageNamedDeviceFactoryCreator( deviceType: deviceType, diff --git a/libs/host/GarnetServer.cs b/libs/host/GarnetServer.cs index c6184586eaa..af25856fed4 100644 --- a/libs/host/GarnetServer.cs +++ b/libs/host/GarnetServer.cs @@ -75,7 +75,12 @@ static string GetVersion() /// /// Configured shutdown drain timeout in seconds. /// - public int ShutdownTimeoutSeconds => opts.ShutdownTimeoutSeconds > 0 ? opts.ShutdownTimeoutSeconds : 5; + public int ShutdownTimeoutSeconds => opts.ShutdownTimeoutSeconds > 0 ? opts.ShutdownTimeoutSeconds : GarnetServerOptions.DefaultShutdownTimeoutSeconds; + + /// + /// Configured data-finalization timeout in seconds (AOF commit / checkpoint during shutdown). + /// + public int DataFinalizationTimeoutSeconds => opts.DataFinalizationTimeoutSeconds > 0 ? opts.DataFinalizationTimeoutSeconds : GarnetServerOptions.DefaultDataFinalizationTimeoutSeconds; /// /// Create Garnet Server instance using specified command line arguments; use Start to start the server. @@ -554,11 +559,12 @@ public void Start() /// /// Timeout for waiting on active connections (default: configured value) /// If true, skip data persistence (AOF commit and checkpoint) during shutdown - /// Cancellation token + /// Cancellation token; when cancelled (e.g. second Ctrl+C), connection draining and data finalization are aborted /// Task representing the async shutdown operation public async Task ShutdownAsync(TimeSpan? timeout = null, bool noSave = false, CancellationToken token = default) { var shutdownTimeout = timeout ?? TimeSpan.FromSeconds(ShutdownTimeoutSeconds); + var skipPersistence = noSave; try { @@ -581,9 +587,14 @@ public async Task ShutdownAsync(TimeSpan? timeout = null, bool noSave = false, C { await WaitForActiveConnectionsAsync(shutdownTimeout, token).ConfigureAwait(false); } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + skipPersistence = true; + logger?.LogWarning("Connection draining was cancelled due to forced shutdown."); + } catch (OperationCanceledException) { - logger?.LogWarning("Connection draining was cancelled. Proceeding with data finalization..."); + logger?.LogWarning("Connection draining timed out. Proceeding with data finalization..."); } } catch (Exception ex) @@ -592,25 +603,34 @@ public async Task ShutdownAsync(TimeSpan? timeout = null, bool noSave = false, C } finally { - if (!noSave) + if (skipPersistence) + { + logger?.LogInformation("Skipping data persistence during shutdown."); + } + else { // Attempt AOF commit or checkpoint as best-effort, - // even if connection draining was cancelled or failed. - // Use a bounded timeout instead of the caller's token to ensure completion. - using var finalizeCts = new CancellationTokenSource(TimeSpan.FromSeconds(15)); + // even if connection draining timed out. Honor caller cancellation (forced shutdown) + // and cap total finalize time with DataFinalizationTimeoutSeconds. + using var finalizeCts = CancellationTokenSource.CreateLinkedTokenSource(token); + finalizeCts.CancelAfter(TimeSpan.FromSeconds(DataFinalizationTimeoutSeconds)); try { await FinalizeDataAsync(finalizeCts.Token).ConfigureAwait(false); } + catch (OperationCanceledException) when (token.IsCancellationRequested) + { + logger?.LogWarning("Data finalization was cancelled due to forced shutdown."); + } + catch (OperationCanceledException) + { + logger?.LogWarning("Data finalization timed out after {TimeoutSeconds} seconds.", DataFinalizationTimeoutSeconds); + } catch (Exception ex) { logger?.LogError(ex, "Error during data finalization"); } } - else - { - logger?.LogInformation("Shutdown with noSave flag - skipping data persistence."); - } } } diff --git a/libs/host/defaults.conf b/libs/host/defaults.conf index 988f5eb09be..66112378886 100644 --- a/libs/host/defaults.conf +++ b/libs/host/defaults.conf @@ -311,6 +311,9 @@ /* Timeout in seconds to wait for active connections to drain during graceful shutdown. */ "ShutdownTimeoutSeconds" : 5, + /* Timeout in seconds for AOF commit and checkpoint during graceful shutdown data finalization. */ + "DataFinalizationTimeoutSeconds" : 15, + /* Use Azure Page Blobs for storage instead of local storage. */ "UseAzureStorage" : false, diff --git a/libs/server/Servers/GarnetServerOptions.cs b/libs/server/Servers/GarnetServerOptions.cs index b1b71228a91..34fb86907cc 100644 --- a/libs/server/Servers/GarnetServerOptions.cs +++ b/libs/server/Servers/GarnetServerOptions.cs @@ -325,10 +325,25 @@ public class GarnetServerOptions : ServerOptions /// public bool QuietMode = false; + /// + /// Default connection-drain timeout in seconds when not configured. + /// + public const int DefaultShutdownTimeoutSeconds = 5; + + /// + /// Default data-finalization timeout in seconds when not configured. + /// + public const int DefaultDataFinalizationTimeoutSeconds = 15; + /// /// Timeout (in seconds) for waiting on active connections to drain during graceful shutdown. /// - public int ShutdownTimeoutSeconds = 5; + public int ShutdownTimeoutSeconds = DefaultShutdownTimeoutSeconds; + + /// + /// Timeout (in seconds) for AOF commit and checkpoint during graceful shutdown data finalization. + /// + public int DataFinalizationTimeoutSeconds = DefaultDataFinalizationTimeoutSeconds; /// /// SAVE and BGSAVE: We will take a full (index + log) checkpoint when ReadOnlyAddress of log increases by this amount, from the last full checkpoint. diff --git a/main/GarnetServer/Program.cs b/main/GarnetServer/Program.cs index 768f572e2d0..1fa4dfe1142 100644 --- a/main/GarnetServer/Program.cs +++ b/main/GarnetServer/Program.cs @@ -22,28 +22,35 @@ static async Task Main(string[] args) // Start the server server.Start(); - using var cts = new CancellationTokenSource(); + using var runCts = new CancellationTokenSource(); + using var shutdownForceCts = new CancellationTokenSource(); + var shutdownSignals = 0; - ConsoleCancelEventHandler cancelKeyPressHandler = (sender, e) => + void OnShutdownSignal() { - e.Cancel = true; // Prevent the process from terminating immediately - if (!cts.IsCancellationRequested) - cts.Cancel(); - }; + // First signal: graceful shutdown. Second signal (or ProcessExit while shutting down): fast path. + if (Interlocked.Increment(ref shutdownSignals) >= 2) + shutdownForceCts.Cancel(); - EventHandler processExitHandler = (sender, e) => - { try { - if (!cts.IsCancellationRequested) - cts.Cancel(); + if (!runCts.IsCancellationRequested) + runCts.Cancel(); } catch (ObjectDisposedException) { // The cancellation source may already be disposed during process teardown. } + } + + ConsoleCancelEventHandler cancelKeyPressHandler = (sender, e) => + { + e.Cancel = true; // Prevent the process from terminating immediately + OnShutdownSignal(); }; + EventHandler processExitHandler = (sender, e) => OnShutdownSignal(); + // Signal cancellation on Ctrl+C; avoid blocking the event handler with async work Console.CancelKeyPress += cancelKeyPressHandler; @@ -55,15 +62,14 @@ static async Task Main(string[] args) { try { - await Task.Delay(Timeout.Infinite, cts.Token).ConfigureAwait(false); + await Task.Delay(Timeout.Infinite, runCts.Token).ConfigureAwait(false); } catch (OperationCanceledException) { - // Graceful shutdown: drain connections, commit AOF, take checkpoint + // Graceful shutdown on first signal; second signal cancels persistence via shutdownForceCts. await server.ShutdownAsync( TimeSpan.FromSeconds(server.ShutdownTimeoutSeconds), - token: CancellationToken.None - ).ConfigureAwait(false); + token: shutdownForceCts.Token).ConfigureAwait(false); } } finally diff --git a/test/standalone/Garnet.test/GarnetServerTcpTests.cs b/test/standalone/Garnet.test/GarnetServerTcpTests.cs index 98566312201..11c97d16cfd 100644 --- a/test/standalone/Garnet.test/GarnetServerTcpTests.cs +++ b/test/standalone/Garnet.test/GarnetServerTcpTests.cs @@ -253,7 +253,7 @@ public async Task ShutdownAsyncRespectsTimeout() } [Test] - public async Task ShutdownAsyncRespectsCancellation() + public void ShutdownAsyncRespectsCancellation() { StartServer(); @@ -263,16 +263,32 @@ public async Task ShutdownAsyncRespectsCancellation() using var cts = new CancellationTokenSource(); - // Act - Cancel immediately + // Act - Cancel immediately (forced shutdown: skip data finalization) cts.Cancel(); + var sw = System.Diagnostics.Stopwatch.StartNew(); Assert.DoesNotThrowAsync(async () => { await server.ShutdownAsync(timeout: TimeSpan.FromSeconds(30), token: cts.Token).ConfigureAwait(false); }); + sw.Stop(); + + // Assert - Should not wait for the full data-finalization timeout + ClassicAssert.Less(sw.ElapsedMilliseconds, 5_000); + } + + [Test] + public void ShutdownAsyncNoSaveSkipsPersistence() + { + StartServer(); + + Assert.DoesNotThrowAsync(async () => + { + await server.ShutdownAsync(timeout: TimeSpan.FromSeconds(5), noSave: true).ConfigureAwait(false); + }); } [Test] - public async Task ShutdownAsyncWithAofCommit() + public void ShutdownAsyncWithAofCommit() { // Arrange - Create server with AOF enabled StartServer(enableAOF: true); @@ -286,7 +302,7 @@ public async Task ShutdownAsyncWithAofCommit() db.StringSet($"aof-key-{i}", $"value-{i}"); } - // Act - Shutdown should commit AOF without errors + // Act & Assert - Shutdown should commit AOF without errors Assert.DoesNotThrowAsync(async () => { await server.ShutdownAsync(timeout: TimeSpan.FromSeconds(5)).ConfigureAwait(false); From 3fcf5f7d31075ade3a967134620ddae6d63751e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Sat, 3 Oct 2026 21:51:17 +0900 Subject: [PATCH 13/14] Fix Conflict with #2145 Based on Context of #2145 I see Checkpoint can be failure, thus new Status Enum and api introduced. Then my PR intends to try best effort to save data thus, StoreApi side code change has try till cancelationToken has be requested --- libs/server/Servers/StoreApi.cs | 40 +++++++++++++++++++++++++++++++-- 1 file changed, 38 insertions(+), 2 deletions(-) diff --git a/libs/server/Servers/StoreApi.cs b/libs/server/Servers/StoreApi.cs index 8773b889179..47cfa7a553a 100644 --- a/libs/server/Servers/StoreApi.cs +++ b/libs/server/Servers/StoreApi.cs @@ -4,6 +4,7 @@ using System; using System.Threading; using System.Threading.Tasks; +using Microsoft.Extensions.Logging; namespace Garnet.server { @@ -98,7 +99,7 @@ public bool FlushDB(int dbId = 0, bool unsafeTruncateLog = false) /// /// True if method can return before checkpoint is taken /// Cancellation token - /// false if checkpoint was skipped due to node state or another checkpoint in progress + /// false if checkpoint was skipped due to node state, another checkpoint in progress for background, or failed public async ValueTask TakeCheckpointAsync(bool background = false, CancellationToken token = default) { using (PreventRoleChange(out var acquired)) @@ -108,7 +109,42 @@ public async ValueTask TakeCheckpointAsync(bool background = false, Cancel return false; } - return await storeWrapper.TakeCheckpointAsync(background, token: token).ConfigureAwait(false); + while (!token.IsCancellationRequested) + { + var status = await storeWrapper.TakeCheckpointAsync(background, token: token).ConfigureAwait(false); + + switch (status) + { + case CheckpointStatus.Success: + return true; + + case CheckpointStatus.AlreadyInProgress: + if (background) + { + return false; + } + + try + { + await Task.Delay(50, token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + return false; + } + + continue; + + case CheckpointStatus.Failed: + storeWrapper.logger?.LogError("TakeCheckpointAsync failed to complete checkpoint."); + return false; + + default: + return false; + } + } + + return false; } } From ad069d16765475ce23fe8cf17af604a76615926e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=EA=B9=80=EC=9C=A0=EC=84=9D=28Yu=20Seok=20Kim=29?= Date: Sat, 3 Oct 2026 21:52:23 +0900 Subject: [PATCH 14/14] Add Test that reflect StoreApi.TakeCheckpointAsync related change --- .../Garnet.test/RespAdminCommandsTests.cs | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/test/standalone/Garnet.test/RespAdminCommandsTests.cs b/test/standalone/Garnet.test/RespAdminCommandsTests.cs index e21efa24ced..9aa936598d8 100644 --- a/test/standalone/Garnet.test/RespAdminCommandsTests.cs +++ b/test/standalone/Garnet.test/RespAdminCommandsTests.cs @@ -883,6 +883,39 @@ public async Task ShutdownAsyncTakesCheckpointWhenStorageTierEnabled() } } + [Test] + public async Task StoreApiTakeCheckpointAsyncWaitsForInProgressCheckpoint() + { + server.Dispose(); + server = TestUtils.CreateGarnetServer(TestUtils.MethodTestDir); + server.Start(); + + // 1. Background checkpoint returns false immediately when checkpoint is already in progress + var paused = server.Provider.StoreWrapper.TryPauseCheckpoints(0); + ClassicAssert.IsTrue(paused); + + var bgResult = await server.Store.TakeCheckpointAsync(background: true); + ClassicAssert.IsFalse(bgResult, "Background checkpoint should not block and return false when already in progress"); + + // 2. Foreground checkpoint with short timeout returns false on cancellation + using (var shortCts = new CancellationTokenSource(TimeSpan.FromMilliseconds(100))) + { + var cancelResult = await server.Store.TakeCheckpointAsync(background: false, token: shortCts.Token); + ClassicAssert.IsFalse(cancelResult, "Foreground checkpoint should return false when cancelled while waiting"); + } + + // 3. Foreground checkpoint waits and succeeds once in-progress checkpoint completes + var checkpointTask = Task.Run(async () => await server.Store.TakeCheckpointAsync(background: false)); + + await Task.Delay(150); + ClassicAssert.IsFalse(checkpointTask.IsCompleted, "Task should still be waiting while checkpoint is paused"); + + server.Provider.StoreWrapper.ResumeCheckpoints(0); + + var result = await checkpointTask; + ClassicAssert.IsTrue(result, "Foreground checkpoint should complete successfully after in-progress checkpoint finishes"); + } + #endregion } } \ No newline at end of file