From f040a3ba6226f0a4107b61ed6669be483fc2196b Mon Sep 17 00:00:00 2001
From: Cam <cam@camm.id.au>
Date: Fri, 17 May 2024 17:59:02 +0200
Subject: [PATCH] fix same rowkey error in batch

---
 Sync.cs | 31 ++++++++++++++++++++++++++++++-
 1 file changed, 30 insertions(+), 1 deletion(-)

diff --git a/Sync.cs b/Sync.cs
index 9397ec4..cc95160 100644
--- a/Sync.cs
+++ b/Sync.cs
@@ -167,6 +167,11 @@ await BatchQueueProcess(tableActionQueue_Users,
     private static async Task BatchQueueProcess(ConcurrentQueue<TableTransactionAction> Queue, TableClient tableClient, CancellationToken ct)
     {
         List<TableTransactionAction> BatchTransactions = new List<TableTransactionAction>();
+        
+        // Used to re-queue transactions that cannot be put in this batch
+        // Such as transactions with a row key that is already present in the batch (cannot perform within the same batch)
+        
+        List<TableTransactionAction> RequeueTransactions = new List<TableTransactionAction>();
 
         // Take items out of the queue until it's empty or the max batch size hit
         while (!Queue.IsEmpty && BatchTransactions.Count < _maxTableBatchSize)
@@ -174,7 +179,24 @@ private static async Task BatchQueueProcess(ConcurrentQueue<TableTransactionActi
             TableTransactionAction dequeued;
 
             if (Queue.TryDequeue(out dequeued))
-                BatchTransactions.Add(dequeued);
+            {
+                // Validate row key is not already in batch transactions
+                // Batches cannot contain two transactions for the same partition key and row.
+                
+                if (BatchTransactions.Any(x =>
+                        x.Entity.PartitionKey == dequeued.Entity.PartitionKey &&
+                        x.Entity.RowKey == dequeued.Entity.RowKey))
+                {
+                    // Requeue the transaction for next batch as it is already existing in this batch
+                    RequeueTransactions.Add(dequeued);
+                }
+                else
+                {
+                    BatchTransactions.Add(dequeued);
+                }
+                
+            }
+                
         }
 
         if (BatchTransactions.Any())
@@ -200,6 +222,13 @@ private static async Task BatchQueueProcess(ConcurrentQueue<TableTransactionActi
                 }
             }
         }
+        
+        // Requeue transactions
+        if (RequeueTransactions.Any())
+        {
+            foreach(var transaction in RequeueTransactions)
+                Queue.Enqueue(transaction);
+        }
     }
     
     /// <summary>