Cluster-wide transactions on client failover with concurrent processing

[krivulcik]

We got hit by the issue with failing cluster transactions on client failover (reported and fixed here: #15700).

This still occurs when multiple threads are processing the documents. When the used database node is brought down, some threads fail with the message like Failed to execute cluster transaction due to the following issues: Guard compare exchange value 'rvn-atomic/doc-c7752570-44a2-44cb-a9af-a2e7f205c527' index does not match the transaction index's 0 change vector on doc-c7752570-44a2-44cb-a9af-a2e7f205c527 Concurrency check failed for putting the key 'rvn-atomic/doc-c7752570-44a2-44cb-a9af-a2e7f205c527'. Requested index: 0, actual index: 5959.

I can't provide a neat unit test, but the testing program below reproduces the issue pretty reliably.

Steps to reproduce:

1. Configure the DocumentStore to a multi-node cluster with an existing database.
2. Run the program.
3. When the program runs, bring down the node used for processing (I identify which node to shut down by looking at the traffic watch).
4. This usually (but not always) causes one or more threads to fail with the exception mentioned above. The exception will be logged.

Also, sometimes, a task fails with an exception "A task was canceled". This looks like the failover didn't work properly -- is this expected?

Reproduction code:

using NLog;
using Raven.Client.Documents;
using Raven.Client.Documents.Session;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace RavenDB5Tests.Tryouts
{
    internal class ClusterWideTransactionsConcurrent
    {
        private static Logger _logger = LogManager.GetCurrentClassLogger();

        public static void ClusterWideTransactionsConcurrentTest()
        {
            using (var store = new DocumentStore
            {
                Urls = new[] { "http://localhost:8080", "http://localhost:8081", "http://localhost:8082" },
                Database = "ClusterWideTransactionsConcurrentTest",
            }.Initialize())
            {
                var tasks = new Dictionary<string, Task>();

                for (int i = 0; i < 5; i++)
                {
                    var taskIdent = "task-" + i;
                    tasks[taskIdent] = Task.Run(() => ProcessDocument(store, taskIdent))
                        .ContinueWith(task =>
                        {
                            _logger.Warn($"{taskIdent} finished.");
                            if (task.Exception != null)
                            {
                                _logger.Error(task.Exception);
                            }
                        });
                }

                var tasksValues = tasks.Values.ToArray();

                _logger.Info("Waiting for tasks to finish.");
                Task.WaitAll(tasksValues);
                _logger.Info("Tasks finished.");
            }
        }

        private static async Task ProcessDocument(IDocumentStore store, string taskId)
        {
            var ident = Guid.NewGuid().ToString();

            _logger.Info($"[{taskId}] Creating doc {ident}.");
            using (var session = store.OpenAsyncSession(new SessionOptions { TransactionMode = TransactionMode.ClusterWide }))
            {
                var doc = new Doc
                {
                    Id = "doc-" + ident,
                };
                await session.StoreAsync(doc);
                await session.SaveChangesAsync();
            }

            var rnd = new Random();

            for (int i = 0; i < 1000; i++)
            {
                _logger.Info($"[{taskId}] {i} Processing doc {ident}.");
                using (var session = store.OpenAsyncSession(new SessionOptions { TransactionMode = TransactionMode.ClusterWide }))
                {
                    var doc = await session.LoadAsync<Doc>("doc-" + ident);

                    doc.Progress = i;

                    await Task.Delay(1000 + rnd.Next(0, 250));

                    await session.StoreAsync(doc);
                    await session.SaveChangesAsync();
                }
            }
        }

        public class Doc
        {
            public string Id { get; set; }
            public int Progress { get; set; }
        }
    }
}

[ayende]

Assume that the failure happens when the transaction has been accepted to the cluster, but not confirmed by the client.
We have a failure to talk to the node, which means that we'll retry. However, we use the same command id, which should ensure that you'll get the same result in the end.

The error you list here can only happen on the initial creation, which is very strange and unlikely.
The task cancelled should be caught internally, not external one.

Here is a stand alone test, which I was able to reproduce the error you are seeing.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;
using Raven.Client.Documents;
using Raven.Client.Documents.Session;
using Raven.Client.ServerWide;
using Raven.Client.ServerWide.Operations;
using Tests.Infrastructure;
using Xunit;
using Xunit.Abstractions;

namespace SlowTests.Issues;

public class GH_16671 : ReplicationTestBase
{
    public GH_16671(ITestOutputHelper output) : base(output)
    {
    }

    [Fact]
    public async Task CanFailoverClsuterTx()
    {
        var (nodes, leader) = await CreateRaftCluster(3);
        string databaseName = GetDatabaseName();
        using var store = new DocumentStore { Urls = nodes.Select(x => x.WebUrl).ToArray(), Database = databaseName }.Initialize();
        // var orderedNodes = new[] { leader.ServerStore.NodeTag }.Concat(nodes.Where(x => x != leader).Select(x => x.ServerStore.NodeTag)).ToArray();
        // await store.Maintenance.Server.SendAsync(new CreateDatabaseOperation(new DatabaseRecord(databaseName)
        // {
        //     Topology = new DatabaseTopology
        //     {
        //         Members = orderedNodes.ToList()
        //     }
        // },3));
        await store.Maintenance.Server.SendAsync(new CreateDatabaseOperation(new DatabaseRecord(databaseName),3));

        var tasks = new Dictionary<string, Task>();
        var cq = new ConcurrentQueue<Exception>();

        for (int i = 0; i < 5; i++)
        {
            var taskIdent = "task-" + i;
            tasks[taskIdent] = Task.Run(() => ProcessDocument(store, taskIdent))
                .ContinueWith(task =>
                {
                    if (task.Exception != null)
                    {
                        cq.Enqueue(task.Exception);
                    }
                });
        }

        await Task.Delay(1000);

        //leader.Dispose();
        var recordWithEtag = await store.Maintenance.Server.SendAsync(new GetDatabaseRecordOperation(databaseName));
        nodes.Single(x => x.ServerStore.NodeTag == recordWithEtag.Topology.Members[0]).Dispose();

        var tasksValues = tasks.Values.ToArray();
        await Task.WhenAll(tasksValues);
        

        Assert.Empty(cq);
    }
    
    private static async Task ProcessDocument(IDocumentStore store, string taskId)
    {
        using (var session = store.OpenAsyncSession(new SessionOptions { TransactionMode = TransactionMode.ClusterWide }))
        {
            var doc = new Doc
            {
                Id = "doc-" + taskId,
            };
            await session.StoreAsync(doc);
            await session.SaveChangesAsync();
        }

        var rnd = new Random();

        for (int i = 0; i < 100; i++)
        {
            using (var session = store.OpenAsyncSession(new SessionOptions { TransactionMode = TransactionMode.ClusterWide }))
            {
                var doc = await session.LoadAsync<Doc>("doc-" + taskId);

                doc.Progress = i;

                await Task.Delay(10 + rnd.Next(0, 250));

                await session.StoreAsync(doc);
                await session.SaveChangesAsync();
            }
        }
    }

    private class Doc
    {
        public string Id { get; set; }
        public int Progress { get; set; }
    }
}

[krivulcik]

Thanks for looking into this.

Is it possible to run this test using RavenDB.TestDriver? Or is it necessary to run as part of the RavenDB project/solution?

Since you managed to reproduce, I assume that you are going to try to handle this situation, is that right?

Also, did you manage to reproduce the task canceled error?

[ayende]

I didn't try using TestDriver, basically, we have facilities in place to test clusters, etc that make doing this a lot easier.
The test itself doesn't fail consistently, so we'll need to investigate, I reproduced the concurrency issue.

Will probably take a few days to resolve properly. Once we have that, I'll look into the cancelled exception

[krivulcik]

Great, thanks for the information.

[ayende]

FYI, this is handled here: #16701

[ayende]

#16909

[krivulcik]

Great news, thanks!

[krivulcik]

Also #17181 and possibly #17082. See #17490 (comment).