Handle CodeFlow conflicts gracefully

src/Microsoft.DotNet.Darc/DarcLib/AzureDevOpsClient.cs

@@ -1560,4 +1560,24 @@ private static string TruncateDescriptionIfNeeded(string str)
         }
         return str;
     }
+
+    public async Task CommentPullRequestAsync(string pullRequestUri, string comment)
+    {
+        (string accountName, string _, string repoName, int id) = ParsePullRequestUri(pullRequestUri);
+
+        using VssConnection connection = CreateVssConnection(accountName);
+        using GitHttpClient client = await connection.GetClientAsync<GitHttpClient>();
+
+        var prComment = new Comment()
+        {
+            CommentType = CommentType.Text,
+            Content = $"{comment}{CommentMarker}"
+        };
+
+        var newCommentThread = new GitPullRequestCommentThread()
+        {
+            Comments = [prComment]
+        };
+        await client.CreateThreadAsync(newCommentThread, repoName, id);
+    }
 }

src/Microsoft.DotNet.Darc/DarcLib/GitHubClient.cs

@@ -1309,4 +1309,10 @@ public async Task DeletePullRequestBranchAsync(string pullRequestUri)
         (string owner, string repo, int id) prInfo = ParsePullRequestUri(pullRequestUri);
         await DeleteBranchAsync(prInfo.owner, prInfo.repo, pr.HeadBranch);
     }
+
+    public async Task CommentPullRequestAsync(string pullRequestUri, string comment)
+    {
+        (string owner, string repo, int id) = ParsePullRequestUri(pullRequestUri);
+        await GetClient(owner, repo).Issue.Comment.Create(owner, repo, id, comment);
+    }
 }

src/Microsoft.DotNet.Darc/DarcLib/IRemote.cs

@@ -188,5 +188,13 @@ Task<List<GitFile>> CommitUpdatesAsync(
     /// <param name="gitDirParent">Location for the .git directory, or null for default</param>
     Task CloneAsync(string repoUri, string commit, string targetDirectory, bool checkoutSubmodules, string gitDirectory);
 
+    /// <summary>
+    ///    Comment on an existing pull request
+    /// </summary>
+    /// <param name="pullRequestUri">Uri of the pull request</param>
+    /// <param name="comment">Comment message</param>
+    /// <returns></returns>
+    Task CommentPullRequestAsync(string pullRequestUri, string comment);
+
     #endregion
 }

src/Microsoft.DotNet.Darc/DarcLib/IRemoteGitRepo.cs

@@ -108,7 +108,7 @@ Task<IEnumerable<int>> SearchPullRequestsAsync(
     ///     or remove each merge policy check that isn't in evaluations
     /// </summary>
     /// <param name="pullRequestUrl">Url of pull request</param>
-    /// <param name="evalutations">List of merge policies</param>
+    /// <param name="evaluations">List of merge policies</param>
     Task CreateOrUpdatePullRequestMergeStatusInfoAsync(string pullRequestUrl, IReadOnlyList<MergePolicyEvaluationResult> evaluations);
 
     /// <summary>
@@ -168,6 +168,13 @@ Task<IEnumerable<int>> SearchPullRequestsAsync(
     /// <param name="repoUri">Repository to find the branch in</param>
     /// <param name="branch">Branch to find</param>
     Task<bool> DoesBranchExistAsync(string repoUri, string branch);
+
+    /// <summary>
+    ///    Comment on an existing pull request
+    /// </summary>
+    /// <param name="pullRequestUri">Uri of the pull request</param>
+    /// <param name="comment">Comment message</param>
+    Task CommentPullRequestAsync(string pullRequestUri, string comment);
 }
 
 #nullable disable

src/Microsoft.DotNet.Darc/DarcLib/Remote.cs

@@ -405,4 +405,10 @@ public async Task<List<GitFile>> GetCommonScriptFilesAsync(string repoUri, strin
 
         return files;
     }
+
+    public async Task CommentPullRequestAsync(string pullRequestUri, string comment)
+    {
+        CheckForValidGitClient();
+        await _remoteGitClient.CommentPullRequestAsync(pullRequestUri, comment);
+    }
 }

src/Microsoft.DotNet.Darc/DarcLib/VirtualMonoRepo/Exceptions.cs

@@ -2,7 +2,10 @@
 // The .NET Foundation licenses this file to you under the MIT license.
 
 using System;
+using System.Collections.Generic;
 using System.IO;
+using System.Linq;
+using System.Text.RegularExpressions;
 using Microsoft.DotNet.DarcLib.Helpers;
 
 #nullable enable
@@ -24,8 +27,50 @@ private static string GetExceptionMessage(VmrIngestionPatch patch, ProcessExecut
             + result;
 }
 
-public class ConflictInPrBranchException(VmrIngestionPatch patch, string targetBranch)
+/// <summary>
+///     Exception thrown when the service can't apply an update to the PR branch due to a conflict
+///     between the source repo and a change that was made in the PR after it was opened.
+/// </summary>
+public class ConflictInPrBranchException(ProcessExecutionResult gitMergeResult, string targetBranch, bool isForwardFlow)
     : Exception($"Failed to flow changes due to conflicts in the target branch ({targetBranch})")
 {
-    public VmrIngestionPatch Patch { get; } = patch;
+    public List<string> FilesInConflict { get; } = ParseResult(gitMergeResult, isForwardFlow);
+
+    private static readonly Regex AlreadyExistsRegex = new("patch failed: (.+): already exist in index");
+    private static readonly Regex PatchFailedRegex = new("error: patch failed: (.*):");
+    private static readonly Regex PatchDoesNotApplyRegex = new("error: (.+): patch does not apply");
+    private static readonly Regex FileDoesNotExistRegex = new("error: (.+): does not exist in index");
+
+    private static readonly Regex[] ConflictRegex =
+        [
+            AlreadyExistsRegex,
+            PatchFailedRegex,
+            PatchDoesNotApplyRegex,
+            FileDoesNotExistRegex
+        ];
+
+    private static List<string> ParseResult(ProcessExecutionResult gitMergeResult, bool isForwardFlow)
+    {
+        List<string> filesInConflict = new();
+        var errors = gitMergeResult.StandardError.Split(Environment.NewLine);
+        foreach (var error in errors)
+        {
+            foreach (var regex in ConflictRegex)
+            {
+                var match = regex.Match(error);
+                if (match.Success)
+                {
+                    filesInConflict.Add(match.Groups[1].Value);
+                    break;
+                }
+            }
+        }
+        if (isForwardFlow)
+        {
+            // Convert VMR paths to normal repo paths, for example src/repo/file.cs -> file.cs
+            return filesInConflict.Select(file => file.Split('/', 3)[2]).Distinct().ToList();
+        }
+        // If we're backflowing, the file paths are already normal
+        return filesInConflict;
+    }
 }

src/Microsoft.DotNet.Darc/DarcLib/VirtualMonoRepo/VmrBackflower.cs

@@ -251,7 +251,7 @@ protected override async Task<bool> SameDirectionFlowAsync(
             if (!rebaseConflicts)
             {
                 _logger.LogInformation("Failed to update a PR branch because of a conflict. Stopping the flow..");
-                throw new ConflictInPrBranchException(e.Patch, headBranch);
+                throw new ConflictInPrBranchException(e.Result, targetBranch, isForwardFlow: false);
             }
 
             // Otherwise, we have a conflicting change in the last backflow PR (before merging)

src/Microsoft.DotNet.Darc/DarcLib/VirtualMonoRepo/VmrForwardFlower.cs

@@ -217,7 +217,7 @@ protected override async Task<bool> SameDirectionFlowAsync(
             if (!rebaseConflicts)
             {
                 _logger.LogInformation("Failed to update a PR branch because of a conflict. Stopping the flow..");
-                throw new ConflictInPrBranchException(e.Patch, headBranch);
+                throw new ConflictInPrBranchException(e.Result, targetBranch, isForwardFlow: true);
             }
 
             // This happens when a conflicting change was made in the last backflow PR (before merging)

src/ProductConstructionService/ProductConstructionService.Common/RedisCache.cs

@@ -13,7 +13,7 @@ public interface IRedisCache
     Task<string?> GetAsync(string key);
     Task<string?> GetAsync();
     Task SetAsync(string value, TimeSpan? expiration = null);
-    Task TryDeleteAsync();
+    Task<bool> TryDeleteAsync();
     Task<string?> TryGetAsync();
     IAsyncEnumerable<string> GetKeysAsync(string pattern);
 }
@@ -44,9 +44,9 @@ public async Task SetAsync(string value, TimeSpan? expiration = null)
         return value.HasValue ? value.ToString() : null;
     }
 
-    public async Task TryDeleteAsync()
+    public async Task<bool> TryDeleteAsync()
     {
-        await Cache.KeyDeleteAsync(_stateKey);
+        return await Cache.KeyDeleteAsync(_stateKey);
     }
 
     public async Task<string?> GetAsync()

src/ProductConstructionService/ProductConstructionService.DependencyFlow/InProgressPullRequest.cs

@@ -50,4 +50,13 @@ public class InProgressPullRequest : DependencyFlowWorkItem, IPullRequest
 
     [DataMember]
     public DateTime? NextCheck { get; set; }
+
+    [DataMember]
+    public InProgressPullRequestState MergeState { get; set; }
+}
+
+public enum InProgressPullRequestState
+{
+    Mergeable,
+    Conflict
 }

src/ProductConstructionService/ProductConstructionService.DependencyFlow/PullRequestUpdater.cs

@@ -1,6 +1,7 @@
 // Licensed to the .NET Foundation under one or more agreements.
 // The .NET Foundation licenses this file to you under the MIT license.
 
+using System.Text;
 using Maestro.Data.Models;
 using Maestro.MergePolicies;
 using Maestro.MergePolicyEvaluation;
@@ -26,7 +27,7 @@ namespace ProductConstructionService.DependencyFlow;
 internal abstract class PullRequestUpdater : IPullRequestUpdater
 {
 #if DEBUG
-    private static readonly TimeSpan DefaultReminderDelay = TimeSpan.FromMinutes(3);
+    private static readonly TimeSpan DefaultReminderDelay = TimeSpan.FromMinutes(1);
 #else
     private static readonly TimeSpan DefaultReminderDelay = TimeSpan.FromMinutes(5);
 #endif
@@ -138,7 +139,6 @@ public async Task<bool> UpdateAssetsAsync(
     public async Task<bool> ProcessPendingUpdatesAsync(SubscriptionUpdateWorkItem update)
     {
         _logger.LogInformation("Processing pending updates for subscription {subscriptionId}", update.SubscriptionId);
-
         // Check if we track an on-going PR already
         InProgressPullRequest? pr = await _pullRequestState.TryGetStateAsync();
         bool isCodeFlow = update.SubscriptionType == SubscriptionType.DependenciesAndSources;
@@ -170,7 +170,7 @@ public async Task<bool> ProcessPendingUpdatesAsync(SubscriptionUpdateWorkItem up
             }
         }
 
-        // Code flow updates are handled separetely
+        // Code flow updates are handled separately
         if (isCodeFlow)
         {
             return await ProcessCodeFlowUpdateAsync(update, pr);
@@ -206,8 +206,8 @@ public async Task<bool> CheckPullRequestAsync(PullRequestCheck pullRequestCheck)
         if (inProgressPr == null)
         {
             _logger.LogInformation("No in-progress pull request found for a PR check");
-            await ClearAllStateAsync(isCodeFlow: true);
-            await ClearAllStateAsync(isCodeFlow: false);
+            await ClearAllStateAsync(isCodeFlow: true, clearPendingUpdates: true);
+            await ClearAllStateAsync(isCodeFlow: false, clearPendingUpdates: true);
             return false;
         }
 
@@ -232,7 +232,7 @@ protected virtual Task TagSourceRepositoryGitHubContactsIfPossibleAsync(InProgre
         _logger.LogInformation("Querying status for pull request {prUrl}", pr.Url);
 
         (var targetRepository, _) = await GetTargetAsync();
-        IRemote remote = await _remoteFactory.CreateRemoteAsync(targetRepository);
+        var remote = await _remoteFactory.CreateRemoteAsync(targetRepository);
 
         PrStatus status;
         try
@@ -267,7 +267,8 @@ await AddDependencyFlowEventsAsync(
                             mergePolicyResult,
                             pr.Url);
 
-                        await ClearAllStateAsync(isCodeFlow);
+                        // If the PR we just merged was in conflict with an update we previously tried to apply, we shouldn't delete the reminder for the update
+                        await ClearAllStateAsync(isCodeFlow, clearPendingUpdates: pr.MergeState == InProgressPullRequestState.Mergeable);
                         return PullRequestStatus.Completed;
 
                     case MergePolicyCheckResult.FailedPolicies:
@@ -277,6 +278,17 @@ await AddDependencyFlowEventsAsync(
                     case MergePolicyCheckResult.NoPolicies:
                     case MergePolicyCheckResult.FailedToMerge:
                         _logger.LogInformation("Pull request {url} still active (updatable) - keeping tracking it", pr.Url);
+                        // Check if we think the PR has a conflict
+                        if (pr.MergeState == InProgressPullRequestState.Conflict)
+                        {
+                            // If we think so, check if the PR head branch still has the same commit as the one we remembered.
+                            // If it doesn't, we should try to update the PR again, the conflicts might be resolved
+                            var latestCommit = await remote.GetLatestCommitAsync(targetRepository, pr.HeadBranch);
+                            if (latestCommit == pr.SourceSha)
+                            {
+                                return PullRequestStatus.InProgressCannotUpdate;
+                            }
+                        }
                         await SetPullRequestCheckReminder(pr, isCodeFlow);
                         return PullRequestStatus.InProgressCanUpdate;
 
@@ -311,7 +323,7 @@ await AddDependencyFlowEventsAsync(
 
                 _logger.LogInformation("PR {url} has been manually {action}. Stopping tracking it", pr.Url, status.ToString().ToLowerInvariant());
 
-                await ClearAllStateAsync(isCodeFlow);
+                await ClearAllStateAsync(isCodeFlow, clearPendingUpdates: pr.MergeState == InProgressPullRequestState.Mergeable);
 
                 // Also try to clean up the PR branch.
                 try
@@ -800,11 +812,16 @@ private async Task SetPullRequestCheckReminder(InProgressPullRequest prState, bo
         await _pullRequestState.SetAsync(prState);
     }
 
-    private async Task ClearAllStateAsync(bool isCodeFlow)
+    private async Task ClearAllStateAsync(bool isCodeFlow, bool clearPendingUpdates)
     {
         await _pullRequestState.TryDeleteAsync();
         await _pullRequestCheckReminders.UnsetReminderAsync(isCodeFlow);
-        await _pullRequestUpdateReminders.UnsetReminderAsync(isCodeFlow);
+        // If the pull request we deleted from the cache had a conflict, we shouldn't unset the update reminder
+        // as there was an update that was previously blocked
+        if (!clearPendingUpdates)
+        {
+            await _pullRequestUpdateReminders.UnsetReminderAsync(isCodeFlow);
+        }
     }
 
     /// <summary>
@@ -905,6 +922,13 @@ await CreateCodeFlowPullRequestAsync(
         {
             await UpdateAssetsAndSources(update, pr);
         }
+        catch (ConflictInPrBranchException conflictException)
+        {
+            return await HandlePrUpdateConflictAsync(
+                conflictException,
+                update,
+                pr);
+        }
         catch (Exception e)
         {
             // TODO https://github.com/dotnet/arcade-services/issues/4198: Notify us about these kind of failures
@@ -962,6 +986,10 @@ private async Task<bool> UpdateAssetsAndSources(SubscriptionUpdateWorkItem updat
                     cancellationToken: default);
             }
         }
+        catch (Exception e) when (e is ConflictInPrBranchException)
+        {
+            throw;
+        }
         catch (Exception e)
         {
             _logger.LogError(e, "Failed to flow changes for build {buildId} in subscription {subscriptionId}",
@@ -1016,6 +1044,7 @@ private async Task<bool> UpdateAssetsAndSources(SubscriptionUpdateWorkItem updat
         });
 
         pullRequest.LastUpdate = DateTime.UtcNow;
+        pullRequest.MergeState = InProgressPullRequestState.Mergeable;
         await SetPullRequestCheckReminder(pullRequest, true);
         await _pullRequestUpdateReminders.UnsetReminderAsync(true);
 
@@ -1171,5 +1200,54 @@ await AddDependencyFlowEventsAsync(
         }
     }
 
+    private async Task<bool> HandlePrUpdateConflictAsync(
+        ConflictInPrBranchException conflictException,
+        SubscriptionUpdateWorkItem update,
+        InProgressPullRequest pr)
+    {
+        // The PR we're trying to update has a conflict with the source repo. We will mark it as blocked, not allowing any updates from this
+        // subscription till it's merged, or the conflict resolved. We'll set a reminder to check on it.
+        StringBuilder sb = new();
+        sb.AppendLine($"There was a conflict in the PR branch when flowing source from {update.GetRepoAtCommitUri()}");
+        sb.AppendLine("Files conflicting with the head branch:");
+        foreach (var file in conflictException.FilesInConflict)
+        {
+            sb.AppendLine($" - [{file}]({update.GetFileAtCommitUri(file)})");
+        }
+        sb.AppendLine();
+        sb.AppendLine("Updates from this subscription will be blocked until the conflict is resolved, or the PR is merged");
+
+        (var targetRepository, _) = await GetTargetAsync();
+        var remote = await _remoteFactory.CreateRemoteAsync(targetRepository);
+
+        try
+        {
+            await remote.CommentPullRequestAsync(pr.Url, sb.ToString());
+        }
+        catch (Exception e)
+        {
+            _logger.LogWarning("Posting comment to {prUrl} failed with exception {message}", pr.Url, e.Message);
+        }
+        // If the headBranch gets updated, we will retry to update it with previously conflicting changes. If these changes still cause a conflict, we should update the
+        // InProgressPullRequest with the latest commit from the remote branch
+        var remoteCommit = pr.SourceSha;
+        try
+        {
+            remoteCommit = await remote.GetLatestCommitAsync(targetRepository, pr.HeadBranch);
+        }
+        catch (Exception e)
+        {
+            _logger.LogWarning("Couldn't get latest commit of {repo}/{commit}. Failed with exception {message}", targetRepository, pr.HeadBranch, e.Message);
+        }
+
+        pr.MergeState = InProgressPullRequestState.Conflict;
+        pr.SourceSha = remoteCommit;
+        await _pullRequestState.SetAsync(pr);
+        await _pullRequestUpdateReminders.SetReminderAsync(update, DefaultReminderDelay, isCodeFlow: true);
+        await _pullRequestCheckReminders.UnsetReminderAsync(isCodeFlow: true);
+
+        return true;
+    }
+
     #endregion
 }