List aof persistent restore issues

[slqd3781]

Describe the bug

Hello, I have encountered an issue with AOF persistence. I use a list as a queue. After running the Garnet for a period of time, the list generates some data that has already been consumed. At this time, there are no values in the list. When I shut down all services that may insert data into the Garnet and restart the Garnet service, I will find that the previously consumed values reappear in the list. I also encounter this issue during local testing. Can you help me confirm if it is a persistence problem or other issues

Steps to reproduce the bug

This is my configuration file
{
"Port": 6379,
"EnableAOF":true,
"Recover": true,
"PubSubPageSize":"16m",
"AofSizeLimit": "1g"
}

Expected behavior

No response

Screenshots

No response

Release version

1.0.84

IDE

No response

OS version

No response

Additional context

No response

[yuseok-kim-edushare]

Hello, are you using garnet on windows? or not?

by the way, I met simmilar issue like you ( #1382 )
this caused by garnet's AOF processing is asynchronus Disk I/O
so, When we try to stop service, we need to save AOF data into Disk explicitaly.

If you using Garnet via Windows Service, my branch (with linked PR #1383 ) can help you

Else, you can use command Save to ensure data will preserved on disk

[slqd3781]

@yuseok-kim-edushare Hello, thank you for your answer. I have used the method you mentioned to explicitly save the AOF data to disk when the server is stopped, but it still doesn't work for the list data structure. The consumed values will still be restored after restarting the Garnet service

[TalZaccai]

Hi @slqd3781, could you please create a standalone repro with specific steps? That would be super helpful! Thanks!

[yuseok-kim-edushare]

AH. If you okay @slqd3781
could you check your configuration file key name?
our company server config is here
We use AOF instead of EnableAOF

{
  "EnableStorageTier": true,
  "AOF": true,
  "Recover": true,
  "CheckpointDir": "C:\\Garnet\\data",
  "LogDir": "C:\\Garnet\\data",
  "Compaction": {
    "Type": "Lookup",
    "FrequencySeconds": 60,
    "MaxSegments": 8
  }
}

---

I misunderstooded
EnableAOF is right

[slqd3781]

@TalZaccai Hello, I can simulate the production and consumption functions of a list using the following simple example, and the problem mentioned above will recur every time

using StackExchange.Redis;

var opts = new ConfigurationOptions
{
    EndPoints = new EndPointCollection
    {
        "127.0.0.1:48996"
    },
    AbortOnConnectFail = false
};

var redis = ConnectionMultiplexer.Connect(opts);
var db = redis.GetDatabase();

ThreadPool.QueueUserWorkItem(u =>
{
    var list = new List<string>();
    for (int i = 0; i < 100000; i++)
    {
        Console.WriteLine(i);
        db.ListLeftPush("list:test", i.ToString());
    }
});

ThreadPool.QueueUserWorkItem(u =>
{
    string? listvalue = string.Empty;
    while (true)
    {
        while ((listvalue = db.ListRightPop("list:test")) != null)
        {
            Console.WriteLine(listvalue + "---delete success");
        }
    }
});

[yuseok-kim-edushare]

Also I test with following codes
then terminal output shows consumption not reflected on AOF restore process

<Project Sdk="Microsoft.NET.Sdk">

  <PropertyGroup>
    <OutputType>Exe</OutputType>
    <TargetFramework>net9.0</TargetFramework>
    <ImplicitUsings>enable</ImplicitUsings>
    <Nullable>enable</Nullable>
    <AssemblyName>GarnetAofRestoreTest</AssemblyName>
    <RootNamespace>GarnetAofRestoreTest</RootNamespace>
  </PropertyGroup>

  <ItemGroup>
    <PackageReference Include="StackExchange.Redis" Version="2.8.16" />
  </ItemGroup>

</Project>

using StackExchange.Redis;
using System.Diagnostics;
using System.Text.Json;

namespace GarnetAofRestoreTest
{
    public class TestResult
    {
        public DateTime Timestamp { get; set; }
        public string TestPhase { get; set; } = "";
        public int QueueLength { get; set; }
        public int TotalProduced { get; set; }
        public int TotalConsumed { get; set; }
        public string Status { get; set; } = "";
        public List<string> QueueItems { get; set; } = new();
        public bool IsRestoreTest { get; set; }
        public string Notes { get; set; } = "";
    }

    class Program
    {
        private static readonly string TestListKey = "list:test";
        private static readonly string TestCounterKey = "test:counter";
        private static readonly string TestStatusKey = "test:status";
        private static readonly string TestConsumedKey = "test:consumed";
        private static readonly string TestResultsFile = "test_results.json";
        
        static async Task Main(string[] args)
        {
            Console.WriteLine("=== Garnet AOF Restore Test ===");
            Console.WriteLine("This test verifies if Garnet AOF persistence correctly restores data");
            Console.WriteLine("when the service is stopped and restarted.");
            Console.WriteLine();

            // Command line argument processing
            if (args.Length > 0)
            {
                switch (args[0].ToLower())
                {
                    case "--check":
                    case "-c":
                        await CheckRestoreStatus();
                        return;
                    case "--compare":
                    case "-cmp":
                        await CompareTestResults();
                        return;
                    case "--help":
                    case "-h":
                        ShowHelp();
                        return;
                }
            }

            var opts = new ConfigurationOptions
            {
                EndPoints = new EndPointCollection
                {
                    "127.0.0.1:6379"
                },
                AbortOnConnectFail = false,
                ConnectTimeout = 5000,
                SyncTimeout = 5000
            };

            try
            {
                var redis = ConnectionMultiplexer.Connect(opts);
                var db = redis.GetDatabase();

                // Connection test
                await db.PingAsync();
                Console.WriteLine("✓ Connected to Garnet server.");

                // Clean up previous test data
                await db.KeyDeleteAsync(TestListKey);
                await db.KeyDeleteAsync(TestCounterKey);
                await db.KeyDeleteAsync(TestStatusKey);
                await db.KeyDeleteAsync(TestConsumedKey);
                Console.WriteLine("✓ Previous test data cleaned up.");

                // Test start
                Console.WriteLine("\n=== Test Started ===");
                Console.WriteLine("1. Starting data producer and consumer.");
                Console.WriteLine("2. Stop Garnet service manually after some time.");
                Console.WriteLine("3. Restart Garnet service.");
                Console.WriteLine("4. Run this program again to check restore status.");
                Console.WriteLine();

                // Test status setup
                await db.StringSetAsync(TestStatusKey, "running");
                await db.StringSetAsync(TestCounterKey, "0");
                await db.StringSetAsync(TestConsumedKey, "0");

                // Data producer
                var producerTask = Task.Run(async () =>
                {
                    var counter = 0;
                    while (true)
                    {
                        try
                        {
                            var value = $"item_{counter}_{DateTime.Now:HH:mm:ss.fff}";
                            await db.ListLeftPushAsync(TestListKey, value);
                            
                            // Update counter
                            await db.StringSetAsync(TestCounterKey, counter.ToString());
                            
                            Console.WriteLine($"[Producer] {value} added (total {counter + 1})");
                            counter++;
                            
                            // Add data every 100ms
                            await Task.Delay(100);
                        }
                        catch (Exception ex)
                        {
                            Console.WriteLine($"[Producer Error] {ex.Message}");
                            await Task.Delay(1000);
                        }
                    }
                });

                // Data consumer
                var consumerTask = Task.Run(async () =>
                {
                    var consumedCount = 0;
                    while (true)
                    {
                        try
                        {
                            var value = await db.ListRightPopAsync(TestListKey);
                            if (value.HasValue)
                            {
                                consumedCount++;
                                await db.StringSetAsync(TestConsumedKey, consumedCount.ToString());
                                Console.WriteLine($"[Consumer] {value} consumed (total {consumedCount} consumed)");
                            }
                            else
                            {
                                // Wait briefly if queue is empty
                                await Task.Delay(50);
                            }
                        }
                        catch (Exception ex)
                        {
                            Console.WriteLine($"[Consumer Error] {ex.Message}");
                            await Task.Delay(1000);
                        }
                    }
                });

                // Test monitoring
                var monitorTask = Task.Run(async () =>
                {
                    while (true)
                    {
                        try
                        {
                            var listLength = await db.ListLengthAsync(TestListKey);
                            var counter = await db.StringGetAsync(TestCounterKey);
                            var consumed = await db.StringGetAsync(TestConsumedKey);
                            
                            Console.WriteLine($"[Monitor] Queue length: {listLength}, Total produced: {counter}, Total consumed: {consumed}");
                            
                            await Task.Delay(5000); // Display status every 5 seconds
                        }
                        catch (Exception ex)
                        {
                            Console.WriteLine($"[Monitor Error] {ex.Message}");
                            await Task.Delay(5000);
                        }
                    }
                });

                Console.WriteLine("Test is running...");
                Console.WriteLine("Press Ctrl+C to stop the test.");
                Console.WriteLine();

                // Ctrl+C handling
                Console.CancelKeyPress += async (sender, e) =>
                {
                    e.Cancel = true;
                    Console.WriteLine("\n=== Test Stopped ===");
                    await db.StringSetAsync(TestStatusKey, "stopped");
                    
                    var finalListLength = await db.ListLengthAsync(TestListKey);
                    var finalCounter = await db.StringGetAsync(TestCounterKey);
                    var finalConsumed = await db.StringGetAsync(TestConsumedKey);
                    
                    // Save test result
                    var testResult = new TestResult
                    {
                        Timestamp = DateTime.Now,
                        TestPhase = "BEFORE_RESTART",
                        QueueLength = (int)finalListLength,
                        TotalProduced = finalCounter.HasValue ? int.Parse(finalCounter) : 0,
                        TotalConsumed = finalConsumed.HasValue ? int.Parse(finalConsumed) : 0,
                        Status = "stopped",
                        IsRestoreTest = false,
                        Notes = "Garnet service state before restart"
                    };
                    
                    await SaveTestResult(testResult);
                    
                    Console.WriteLine($"Final state:");
                    Console.WriteLine($"- Queue length: {finalListLength}");
                    Console.WriteLine($"- Total produced: {finalCounter}");
                    Console.WriteLine($"- Total consumed: {finalConsumed}");
                    Console.WriteLine();
                    Console.WriteLine("Now stop and restart Garnet service, then run:");
                    Console.WriteLine("  dotnet run --check");
                    Console.WriteLine("  dotnet run --compare");
                    
                    Environment.Exit(0);
                };

                // Wait for all tasks to run
                await Task.WhenAll(producerTask, consumerTask, monitorTask);
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Connection error: {ex.Message}");
                Console.WriteLine("Please check if Garnet service is running.");
            }
        }

        // Check restore status method
        public static async Task CheckRestoreStatus()
        {
            var opts = new ConfigurationOptions
            {
                EndPoints = new EndPointCollection
                {
                    "127.0.0.1:6379"
                },
                AbortOnConnectFail = false,
                ConnectTimeout = 5000,
                SyncTimeout = 5000
            };

            try
            {
                var redis = ConnectionMultiplexer.Connect(opts);
                var db = redis.GetDatabase();

                await db.PingAsync();
                Console.WriteLine("✓ Connected to Garnet server.");

                var listLength = await db.ListLengthAsync(TestListKey);
                var counter = await db.StringGetAsync(TestCounterKey);
                var consumed = await db.StringGetAsync(TestConsumedKey);
                var status = await db.StringGetAsync(TestStatusKey);

                Console.WriteLine("\n=== Restore Status Check ===");
                Console.WriteLine($"Test status: {status}");
                Console.WriteLine($"Queue length: {listLength}");
                Console.WriteLine($"Total produced: {counter}");
                Console.WriteLine($"Total consumed: {consumed}");

                var queueItems = new List<string>();
                if (listLength > 0)
                {
                    Console.WriteLine("\nRemaining queue items:");
                    var items = await db.ListRangeAsync(TestListKey, 0, -1);
                    queueItems = items.Select(x => x.ToString()).ToList();
                    
                    foreach (var item in items.Take(10)) // Show first 10 items
                    {
                        Console.WriteLine($"  - {item}");
                    }
                    if (items.Length > 10)
                    {
                        Console.WriteLine($"  ... and {items.Length - 10} more");
                    }
                }

                // Save test result
                var testResult = new TestResult
                {
                    Timestamp = DateTime.Now,
                    TestPhase = "AFTER_RESTART",
                    QueueLength = (int)listLength,
                    TotalProduced = counter.HasValue ? int.Parse(counter) : 0,
                    TotalConsumed = consumed.HasValue ? int.Parse(consumed) : 0,
                    Status = status.HasValue ? status : "",
                    QueueItems = queueItems,
                    IsRestoreTest = true,
                    Notes = "Garnet service state after restart"
                };
                
                await SaveTestResult(testResult);

                // Check for issues
                if (listLength > 0 && status == "stopped")
                {
                    Console.WriteLine("\n⚠️  Warning: Queue has items remaining after test was stopped.");
                    Console.WriteLine("This indicates AOF restore is working correctly.");
                }
                else if (listLength == 0 && status == "stopped")
                {
                    Console.WriteLine("\n✓ Queue is empty. This is normal state.");
                }
                
                Console.WriteLine("\nTest results saved to test_results.json file.");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Restore status check error: {ex.Message}");
            }
        }

        // Test result comparison method
        public static async Task CompareTestResults()
        {
            try
            {
                if (!File.Exists(TestResultsFile))
                {
                    Console.WriteLine("Test results file not found. Please run the test first.");
                    return;
                }

                var json = await File.ReadAllTextAsync(TestResultsFile);
                var results = JsonSerializer.Deserialize<List<TestResult>>(json) ?? new List<TestResult>();

                var beforeRestart = results.FirstOrDefault(r => r.TestPhase == "BEFORE_RESTART");
                var afterRestart = results.FirstOrDefault(r => r.TestPhase == "AFTER_RESTART");

                Console.WriteLine("=== Test Result Comparison ===");
                Console.WriteLine();

                if (beforeRestart == null)
                {
                    Console.WriteLine("❌ No data found before service restart.");
                    return;
                }

                if (afterRestart == null)
                {
                    Console.WriteLine("❌ No data found after service restart.");
                    Console.WriteLine("Please run 'dotnet run --check' first.");
                    return;
                }

                Console.WriteLine("📊 Comparison Results:");
                Console.WriteLine($"Before service restart:");
                Console.WriteLine($"  - Queue length: {beforeRestart.QueueLength}");
                Console.WriteLine($"  - Total produced: {beforeRestart.TotalProduced}");
                Console.WriteLine($"  - Total consumed: {beforeRestart.TotalConsumed}");
                Console.WriteLine($"  - Time: {beforeRestart.Timestamp:yyyy-MM-dd HH:mm:ss}");
                Console.WriteLine();

                Console.WriteLine($"After service restart:");
                Console.WriteLine($"  - Queue length: {afterRestart.QueueLength}");
                Console.WriteLine($"  - Total produced: {afterRestart.TotalProduced}");
                Console.WriteLine($"  - Total consumed: {afterRestart.TotalConsumed}");
                Console.WriteLine($"  - Time: {afterRestart.Timestamp:yyyy-MM-dd HH:mm:ss}");
                Console.WriteLine();

                // Analysis
                Console.WriteLine("🔍 Analysis:");
                
                if (beforeRestart.QueueLength == afterRestart.QueueLength)
                {
                    Console.WriteLine("✅ Queue length matches. Normal restore.");
                }
                else
                {
                    Console.WriteLine($"❌ Queue length differs! ({beforeRestart.QueueLength} → {afterRestart.QueueLength})");
                    Console.WriteLine("   This indicates an AOF restore issue.");
                }

                if (beforeRestart.TotalProduced == afterRestart.TotalProduced)
                {
                    Console.WriteLine("✅ Total produced items match.");
                }
                else
                {
                    Console.WriteLine($"❌ Total produced items differ! ({beforeRestart.TotalProduced} → {afterRestart.TotalProduced})");
                }

                if (beforeRestart.TotalConsumed == afterRestart.TotalConsumed)
                {
                    Console.WriteLine("✅ Total consumed items match.");
                }
                else
                {
                    Console.WriteLine($"❌ Total consumed items differ! ({beforeRestart.TotalConsumed} → {afterRestart.TotalConsumed})");
                }

                // Problem detection
                var queueLengthDiff = afterRestart.QueueLength - beforeRestart.QueueLength;
                var consumedDiff = afterRestart.TotalConsumed - beforeRestart.TotalConsumed;

                if (queueLengthDiff > 0)
                {
                    Console.WriteLine($"\n🚨 Problem detected: {queueLengthDiff} additional items in queue!");
                    Console.WriteLine("This suggests previously consumed data has reappeared.");
                    
                    if (afterRestart.QueueItems.Any())
                    {
                        Console.WriteLine("\nRestored items:");
                        foreach (var item in afterRestart.QueueItems.Take(5))
                        {
                            Console.WriteLine($"  - {item}");
                        }
                        if (afterRestart.QueueItems.Count > 5)
                        {
                            Console.WriteLine($"  ... and {afterRestart.QueueItems.Count - 5} more");
                        }
                    }
                }
                else if (queueLengthDiff < 0)
                {
                    Console.WriteLine($"\n⚠️  Warning: {Math.Abs(queueLengthDiff)} items disappeared from queue.");
                }

                Console.WriteLine($"\n📝 Test results are saved in {TestResultsFile} file.");
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Comparison error: {ex.Message}");
            }
        }

        // Save test result
        private static async Task SaveTestResult(TestResult result)
        {
            try
            {
                List<TestResult> results = new();
                
                if (File.Exists(TestResultsFile))
                {
                    var json = await File.ReadAllTextAsync(TestResultsFile);
                    results = JsonSerializer.Deserialize<List<TestResult>>(json) ?? new List<TestResult>();
                }

                results.Add(result);
                
                var options = new JsonSerializerOptions
                {
                    WriteIndented = true,
                    Encoder = System.Text.Encodings.Web.JavaScriptEncoder.UnsafeRelaxedJsonEscaping
                };
                
                var jsonString = JsonSerializer.Serialize(results, options);
                await File.WriteAllTextAsync(TestResultsFile, jsonString);
            }
            catch (Exception ex)
            {
                Console.WriteLine($"Test result save error: {ex.Message}");
            }
        }

        // Show help
        private static void ShowHelp()
        {
            Console.WriteLine("=== Garnet AOF Restore Test Help ===");
            Console.WriteLine();
            Console.WriteLine("Usage:");
            Console.WriteLine("  dotnet run                    - Run main test");
            Console.WriteLine("  dotnet run --check (-c)       - Check restore status");
            Console.WriteLine("  dotnet run --compare (-cmp)   - Compare test results");
            Console.WriteLine("  dotnet run --help (-h)        - Show this help");
            Console.WriteLine();
            Console.WriteLine("Test Procedure:");
            Console.WriteLine("1. dotnet run                    - Start test");
            Console.WriteLine("2. Ctrl+C                        - Stop test");
            Console.WriteLine("3. Stop Garnet service");
            Console.WriteLine("4. Restart Garnet service");
            Console.WriteLine("5. dotnet run --check            - Check restore status");
            Console.WriteLine("6. dotnet run --compare         - Compare results");
            Console.WriteLine();
            Console.WriteLine("Output Files:");
            Console.WriteLine($"- {TestResultsFile}             - Test results JSON file");
        }
    }
}

[yuseok-kim-edushare]

@slqd3781

Sorry for confusing and my mistakes
I confused storageTier and AOF.

If we want save commit AOF
use command "COMMITAOF"
Else if we want save checkpoint StorageTier use command "Save"

furthermore I modified my PR #1383 to ensure garnet running as windows service that keep aof data to disk explicitly

[yuseok-kim-edushare]

@TalZaccai Sorry for confusing my initial answer in this issue
as above command, I misunderstand and confused between AOF and Garnet's StorageTier

[slqd3781]

@yuseok-kim-edushare I have passed the local test based on your submission. Thank you again for your answer. I wish you a happy life