diff --git a/hosting/Windows/Garnet.worker/Program.cs b/hosting/Windows/Garnet.worker/Program.cs index 8418da86716..929047c0389 100644 --- a/hosting/Windows/Garnet.worker/Program.cs +++ b/hosting/Windows/Garnet.worker/Program.cs @@ -1,16 +1,37 @@ // Copyright (c) Microsoft Corporation. // Licensed under the MIT license. +using System; using Garnet; +using Garnet.server; using Microsoft.Extensions.DependencyInjection; using Microsoft.Extensions.Hosting; class Program { + // Extra host budget beyond connection drain + data finalization (quiesce, stop listening, etc.). + private const int HostShutdownMarginSeconds = 5; + static void Main(string[] args) { + // 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); - 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: drain + data finalization + margin. + builder.Services.Configure(opts => + { + opts.ShutdownTimeout = shutdownTimeout + + TimeSpan.FromSeconds(dataFinalizationTimeoutSeconds) + + TimeSpan.FromSeconds(HostShutdownMarginSeconds); + }); + + builder.Services.AddHostedService(_ => new Worker(args, shutdownTimeout)); builder.Services.AddWindowsService(options => { @@ -20,4 +41,21 @@ static void Main(string[] args) var host = builder.Build(); host.Run(); } + + /// + /// Scans for (or ) and returns + /// the parsed positive integer, or if the argument is absent or invalid. + /// + 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 var name && (name == longName || name == shortName) && + int.TryParse(args[i + 1], out var value) && value > 0) + { + return value; + } + } + 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 d69adb7e3c0..d711974d13f 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 fit within the host , + /// which also budgets for data finalization and a small margin (see Program.cs). + /// + 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/Configuration/Options.cs b/libs/host/Configuration/Options.cs index ec20aac2ff3..0347e66ea0b 100644 --- a/libs/host/Configuration/Options.cs +++ b/libs/host/Configuration/Options.cs @@ -423,6 +423,15 @@ 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; } + + [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; } @@ -1015,6 +1024,8 @@ public GarnetServerOptions GetServerOptions(ILogger logger = null) ThreadPoolMinIOCompletionThreads = ThreadPoolMinIOCompletionThreads, 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 0e32400f6e3..0f72e8fc37f 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; @@ -71,6 +72,16 @@ static string GetVersion() /// public StoreApi Store; + /// + /// Configured shutdown drain timeout in seconds. + /// + 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. /// @@ -551,6 +562,229 @@ 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: configured value) + /// If true, skip data persistence (AOF commit and checkpoint) during shutdown + /// 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 + { + // 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) when (token.IsCancellationRequested) + { + skipPersistence = true; + logger?.LogWarning("Connection draining was cancelled due to forced shutdown."); + } + catch (OperationCanceledException) + { + logger?.LogWarning("Connection draining timed out. Proceeding with data finalization..."); + } + } + catch (Exception ex) + { + logger?.LogError(ex, "Error during graceful shutdown"); + } + finally + { + if (skipPersistence) + { + logger?.LogInformation("Skipping data persistence during shutdown."); + } + else + { + // Attempt AOF commit or checkpoint as best-effort, + // 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"); + } + } + } + } + + /// + /// 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. + /// Both Enabled AOF and checkpoint, but AOF is prioritized. + /// IF AOF commit failed, checkpoint will be performed. + /// + 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."); + return; // skip checkpoint + } + else + { + logger?.LogInformation("AOF commit skipped (another commit in progress or replica mode)."); + } + } + catch (Exception ex) + { + logger?.LogError(ex, "Error committing AOF during shutdown"); + } + } + + if (!opts.EnableStorageTier) + return; + + logger?.LogDebug("Taking checkpoint for tiered storage..."); + try + { + var checkpointSuccess = await Store.TakeCheckpointAsync(background: false, token: token).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"); + } + return; + } + /// /// Dispose store (including log and checkpoint directory) /// diff --git a/libs/host/defaults.conf b/libs/host/defaults.conf index 3f287174ff3..cca16ae95d6 100644 --- a/libs/host/defaults.conf +++ b/libs/host/defaults.conf @@ -323,6 +323,12 @@ /* Maximum number of simultaneously active network connections. */ "NetworkConnectionLimit" : 10000, + /* 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/PubSub/SubscribeBroker.cs b/libs/server/PubSub/SubscribeBroker.cs index 05204f2c9e6..a1e66644dd2 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 02e6ea82cca..e4b626d1964 100644 --- a/libs/server/Resp/RespServerSession.cs +++ b/libs/server/Resp/RespServerSession.cs @@ -544,6 +544,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; 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 80ce876e458..5ae6a24cbd6 100644 --- a/libs/server/Servers/GarnetServerOptions.cs +++ b/libs/server/Servers/GarnetServerOptions.cs @@ -354,6 +354,26 @@ 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 = 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/libs/server/Servers/GarnetServerTcp.cs b/libs/server/Servers/GarnetServerTcp.cs index 191f81cc182..1cbcd338643 100644 --- a/libs/server/Servers/GarnetServerTcp.cs +++ b/libs/server/Servers/GarnetServerTcp.cs @@ -167,10 +167,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); + } } /// @@ -178,7 +200,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 diff --git a/libs/server/Servers/StoreApi.cs b/libs/server/Servers/StoreApi.cs index 9ad5d9db3f0..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 { @@ -93,6 +94,60 @@ 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, another checkpoint in progress for background, or failed + public async ValueTask TakeCheckpointAsync(bool background = false, CancellationToken token = default) + { + using (PreventRoleChange(out var acquired)) + { + if (!acquired || IsReplica) + { + return 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; + } + } + /// /// Helper to disable role changes during a using block. /// diff --git a/main/GarnetServer/Program.cs b/main/GarnetServer/Program.cs index 7b2673ebc41..1fa4dfe1142 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,61 @@ static void Main(string[] args) // Start the server server.Start(); - Thread.Sleep(Timeout.Infinite); + using var runCts = new CancellationTokenSource(); + using var shutdownForceCts = new CancellationTokenSource(); + var shutdownSignals = 0; + + void OnShutdownSignal() + { + // First signal: graceful shutdown. Second signal (or ProcessExit while shutting down): fast path. + if (Interlocked.Increment(ref shutdownSignals) >= 2) + shutdownForceCts.Cancel(); + + try + { + 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; + + // 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, runCts.Token).ConfigureAwait(false); + } + catch (OperationCanceledException) + { + // Graceful shutdown on first signal; second signal cancels persistence via shutdownForceCts. + await server.ShutdownAsync( + TimeSpan.FromSeconds(server.ShutdownTimeoutSeconds), + token: shutdownForceCts.Token).ConfigureAwait(false); + } + } + finally + { + Console.CancelKeyPress -= cancelKeyPressHandler; + AppDomain.CurrentDomain.ProcessExit -= processExitHandler; + } } catch (Exception ex) { diff --git a/test/standalone/Garnet.test/GarnetServerTcpTests.cs b/test/standalone/Garnet.test/GarnetServerTcpTests.cs index 290bd27d16a..80485f95ef5 100644 --- a/test/standalone/Garnet.test/GarnetServerTcpTests.cs +++ b/test/standalone/Garnet.test/GarnetServerTcpTests.cs @@ -1,23 +1,47 @@ // Copyright (c) Microsoft Corporation. // 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.common; 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 (ipv6 && !Socket.OSSupportsIPv6) @@ -84,5 +108,212 @@ 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 (OperationCanceledException) when (cts.Token.IsCancellationRequested) + { + break; + } + catch + { + // Connection failures are expected after StopListening + } + } + })); + } + + 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 void ShutdownAsyncRespectsCancellation() + { + StartServer(); + + // Arrange + using var redis = ConnectionMultiplexer.Connect(TestUtils.GetConfig()); + redis.GetDatabase(0).Ping(); + + using var cts = new CancellationTokenSource(); + + // 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 void 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 & Assert - 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..9aa936598d8 100644 --- a/test/standalone/Garnet.test/RespAdminCommandsTests.cs +++ b/test/standalone/Garnet.test/RespAdminCommandsTests.cs @@ -779,5 +779,143 @@ 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()); + } + } + + [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 diff --git a/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs new file mode 100644 index 00000000000..fbaa2fc8bbc --- /dev/null +++ b/test/standalone/Garnet.test/ShutdownDataConsistencyTests.cs @@ -0,0 +1,435 @@ +// 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] + [Explicit] + 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 + 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 + 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"); + } + } + } +} \ No newline at end of file