Fix learning competing consumers bug (#7324)

* AsyncFile.Move returns true only if the file was read (i.e. exists and not locked)
* Clean up the .pending directory if unable to move and unlock the file

Co-authored-by: Daniel Marbach <[email protected]>

Author: SzymonPobiega

src/NServiceBus.Core/Transports/Learning/AsyncFile.cs

@@ -121,15 +121,15 @@ public static async Task<bool> Move(string sourcePath, string targetPath, Cancel
 
             if (!IsFileLocked(targetPath))
             {
-                break;
+                return true;
             }
 
             await Task.Delay(100, cancellationToken).ConfigureAwait(false);
 
             count++;
         }
 
-        return true;
+        return false;
     }
 
     static bool IsFileLocked(string filePath)

src/NServiceBus.Core/Transports/Learning/DirectoryBasedTransaction.cs

@@ -19,12 +19,19 @@ public DirectoryBasedTransaction(string basePath, string pendingDirName, string
 
     public string FileToProcess { get; private set; }
 
-    public Task<bool> BeginTransaction(string incomingFilePath, CancellationToken cancellationToken = default)
+    public async Task<bool> BeginTransaction(string incomingFilePath, CancellationToken cancellationToken = default)
     {
         Directory.CreateDirectory(transactionDir);
         FileToProcess = Path.Combine(transactionDir, Path.GetFileName(incomingFilePath));
 
-        return AsyncFile.Move(incomingFilePath, FileToProcess, cancellationToken);
+        var succeeded = await AsyncFile.Move(incomingFilePath, FileToProcess, cancellationToken).ConfigureAwait(false);
+        if (succeeded)
+        {
+            return true;
+        }
+
+        Directory.Delete(transactionDir, true);
+        return false;
     }
 
     public async Task Commit(CancellationToken cancellationToken = default)