Coalesce concurrent download requests for the same object

When multiple git.exe processes request the same loose object simultaneously
via named pipes, GVFS previously started independent HTTP downloads for each
request. Under memory pressure, this caused 48K+ redundant requests, draining
the connection pool and effectively hanging the mount process.

Add a ConcurrentDictionary<string, Lazy<DownloadAndSaveObjectResult>> to
GVFSGitObjects that ensures only one download runs per objectId at a time.
Concurrent callers for the same SHA share the result of the single in-flight
download. The entry is removed after completion so subsequent requests get
fresh download attempts.

Work item: 60167591

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: Tyrie Vella

GVFS/GVFS.Common/Git/GVFSGitObjects.cs

@@ -14,12 +14,14 @@ public class GVFSGitObjects : GitObjects
         private static readonly TimeSpan NegativeCacheTTL = TimeSpan.FromSeconds(30);
 
         private ConcurrentDictionary<string, DateTime> objectNegativeCache;
+        private ConcurrentDictionary<string, Lazy<DownloadAndSaveObjectResult>> inflightDownloads;
 
         public GVFSGitObjects(GVFSContext context, GitObjectsHttpRequestor objectRequestor)
             : base(context.Tracer, context.Enlistment, objectRequestor, context.FileSystem)
         {
             this.Context = context;
             this.objectNegativeCache = new ConcurrentDictionary<string, DateTime>(StringComparer.OrdinalIgnoreCase);
+            this.inflightDownloads = new ConcurrentDictionary<string, Lazy<DownloadAndSaveObjectResult>>(StringComparer.OrdinalIgnoreCase);
         }
 
         public enum RequestSource
@@ -127,6 +129,40 @@ private DownloadAndSaveObjectResult TryDownloadAndSaveObject(
                 this.objectNegativeCache.TryRemove(objectId, out negativeCacheRequestTime);
             }
 
+            // Coalesce concurrent requests for the same objectId so that only one HTTP
+            // download runs per SHA at a time. All concurrent callers share the result.
+            // Note: the first caller's cancellationToken and retryOnFailure settings are
+            // captured by the Lazy factory. Subsequent coalesced callers inherit those
+            // settings. In practice this is fine because the primary concurrent path
+            // (NamedPipeMessage from git.exe) always uses CancellationToken.None.
+            Lazy<DownloadAndSaveObjectResult> newLazy = new Lazy<DownloadAndSaveObjectResult>(
+                () => this.DoDownloadAndSaveObject(objectId, cancellationToken, requestSource, retryOnFailure));
+            Lazy<DownloadAndSaveObjectResult> lazy = this.inflightDownloads.GetOrAdd(objectId, newLazy);
+
+            if (!ReferenceEquals(lazy, newLazy))
+            {
+                EventMetadata metadata = new EventMetadata();
+                metadata.Add("objectId", objectId);
+                metadata.Add("requestSource", requestSource.ToString());
+                this.Context.Tracer.RelatedEvent(EventLevel.Informational, "TryDownloadAndSaveObject_CoalescedRequest", metadata);
+            }
+
+            try
+            {
+                return lazy.Value;
+            }
+            finally
+            {
+                this.inflightDownloads.TryRemove(objectId, out _);
+            }
+        }
+
+        private DownloadAndSaveObjectResult DoDownloadAndSaveObject(
+            string objectId,
+            CancellationToken cancellationToken,
+            RequestSource requestSource,
+            bool retryOnFailure)
+        {
             // To reduce allocations, reuse the same buffer when writing objects in this batch
             byte[] bufToCopyWith = new byte[StreamUtil.DefaultCopyBufferSize];
 

GVFS/GVFS.UnitTests/Git/GVFSGitObjectsTests.cs

@@ -118,6 +118,157 @@ public void FailsNullBytePackDownloads()
                 gitObjects => gitObjects.TryDownloadCommit("object0"));
         }
 
+        [TestCase]
+        public void CoalescesMultipleConcurrentRequestsForSameObject()
+        {
+            ManualResetEventSlim downloadStarted = new ManualResetEventSlim(false);
+            ManualResetEventSlim downloadGate = new ManualResetEventSlim(false);
+            int downloadCount = 0;
+
+            CoalescingTestHttpGitObjects httpObjects = new CoalescingTestHttpGitObjects(
+                this.validTestObjectFileContents,
+                onDownloadStarting: () =>
+                {
+                    Interlocked.Increment(ref downloadCount);
+                    downloadStarted.Set();
+                    downloadGate.Wait();
+                });
+
+            MockFileSystemWithCallbacks fileSystem = new MockFileSystemWithCallbacks();
+            fileSystem.OnFileExists = (path) => false;
+            fileSystem.OnMoveFile = (source, target) => { };
+            fileSystem.OnOpenFileStream = (path, mode, access) =>
+            {
+                if (access == FileAccess.Read)
+                {
+                    return new MemoryStream(this.validTestObjectFileContents);
+                }
+
+                return new MemoryStream();
+            };
+
+            GVFSGitObjects dut = this.CreateTestableGVFSGitObjects(httpObjects, fileSystem);
+
+            const int threadCount = 10;
+            GitObjects.DownloadAndSaveObjectResult[] results = new GitObjects.DownloadAndSaveObjectResult[threadCount];
+            Thread[] threads = new Thread[threadCount];
+            CountdownEvent allReady = new CountdownEvent(threadCount);
+            ManualResetEventSlim go = new ManualResetEventSlim(false);
+
+            for (int i = 0; i < threadCount; i++)
+            {
+                int idx = i;
+                threads[i] = new Thread(() =>
+                {
+                    allReady.Signal();
+                    go.Wait();
+                    results[idx] = dut.TryDownloadAndSaveObject(
+                        ValidTestObjectFileSha1,
+                        GVFSGitObjects.RequestSource.NamedPipeMessage);
+                });
+                threads[i].Start();
+            }
+
+            // Release all threads simultaneously
+            allReady.Wait();
+            go.Set();
+
+            // Wait for the first download to start (proves one thread entered the factory)
+            downloadStarted.Wait(TimeSpan.FromSeconds(5)).ShouldBeTrue("Download should have started");
+
+            // Give other threads time to pile up on the Lazy<T>
+            Thread.Sleep(200);
+
+            // Release the download
+            downloadGate.Set();
+
+            // Wait for all threads
+            foreach (Thread t in threads)
+            {
+                t.Join(TimeSpan.FromSeconds(10)).ShouldBeTrue("Thread should complete");
+            }
+
+            // Only one download should have occurred
+            downloadCount.ShouldEqual(1);
+
+            // All threads should have gotten Success
+            foreach (GitObjects.DownloadAndSaveObjectResult result in results)
+            {
+                result.ShouldEqual(GitObjects.DownloadAndSaveObjectResult.Success);
+            }
+        }
+
+        [TestCase]
+        public void DifferentObjectsAreNotCoalesced()
+        {
+            string secondSha = "b376885ac8452b6cbf9ced81b1080bfd570d9b91";
+            int downloadCount = 0;
+
+            CoalescingTestHttpGitObjects httpObjects = new CoalescingTestHttpGitObjects(
+                this.validTestObjectFileContents,
+                onDownloadStarting: () => Interlocked.Increment(ref downloadCount));
+
+            MockFileSystemWithCallbacks fileSystem = new MockFileSystemWithCallbacks();
+            fileSystem.OnFileExists = (path) => false;
+            fileSystem.OnMoveFile = (source, target) => { };
+            fileSystem.OnOpenFileStream = (path, mode, access) =>
+            {
+                if (access == FileAccess.Read)
+                {
+                    return new MemoryStream(this.validTestObjectFileContents);
+                }
+
+                return new MemoryStream();
+            };
+
+            GVFSGitObjects dut = this.CreateTestableGVFSGitObjects(httpObjects, fileSystem);
+
+            dut.TryDownloadAndSaveObject(ValidTestObjectFileSha1, GVFSGitObjects.RequestSource.NamedPipeMessage)
+                .ShouldEqual(GitObjects.DownloadAndSaveObjectResult.Success);
+
+            dut.TryDownloadAndSaveObject(secondSha, GVFSGitObjects.RequestSource.NamedPipeMessage)
+                .ShouldEqual(GitObjects.DownloadAndSaveObjectResult.Success);
+
+            downloadCount.ShouldEqual(2);
+        }
+
+        [TestCase]
+        public void FailedDownloadAllowsSubsequentRetry()
+        {
+            int downloadCount = 0;
+
+            CoalescingTestHttpGitObjects httpObjects = new CoalescingTestHttpGitObjects(
+                this.validTestObjectFileContents,
+                onDownloadStarting: () => Interlocked.Increment(ref downloadCount),
+                failUntilAttempt: 2);
+
+            MockFileSystemWithCallbacks fileSystem = new MockFileSystemWithCallbacks();
+            fileSystem.OnFileExists = (path) => false;
+            fileSystem.OnMoveFile = (source, target) => { };
+            fileSystem.OnOpenFileStream = (path, mode, access) =>
+            {
+                if (access == FileAccess.Read)
+                {
+                    return new MemoryStream(this.validTestObjectFileContents);
+                }
+
+                return new MemoryStream();
+            };
+
+            GVFSGitObjects dut = this.CreateTestableGVFSGitObjects(httpObjects, fileSystem);
+
+            // First attempt fails
+            dut.TryDownloadAndSaveObject(ValidTestObjectFileSha1, GVFSGitObjects.RequestSource.NamedPipeMessage)
+                .ShouldEqual(GitObjects.DownloadAndSaveObjectResult.Error);
+
+            // Second attempt should start a new download (not reuse cached failure)
+            dut.TryDownloadAndSaveObject(ValidTestObjectFileSha1, GVFSGitObjects.RequestSource.NamedPipeMessage)
+                .ShouldEqual(GitObjects.DownloadAndSaveObjectResult.Success);
+
+            // Two separate downloads should have occurred
+            downloadCount.ShouldEqual(2);
+        }
+
         private void AssertRetryableExceptionOnDownload(
             MemoryStream inputStream,
             string mediaType,
@@ -140,7 +291,7 @@ private void AssertRetryableExceptionOnDownload(
             }
         }
 
-        private GVFSGitObjects CreateTestableGVFSGitObjects(MockHttpGitObjects httpObjects, MockFileSystemWithCallbacks fileSystem)
+        private GVFSGitObjects CreateTestableGVFSGitObjects(GitObjectsHttpRequestor httpObjects, MockFileSystemWithCallbacks fileSystem)
         {
             MockTracer tracer = new MockTracer();
             GVFSEnlistment enlistment = new GVFSEnlistment(TestEnlistmentRoot, "https://fakeRepoUrl", "fakeGitBinPath", authentication: null);
@@ -224,5 +375,71 @@ public UnsafeGVFSGitObjects(GVFSContext context, GitObjectsHttpRequestor objectR
                 this.checkData = false;
             }
         }
+
+        private class CoalescingTestHttpGitObjects : GitObjectsHttpRequestor
+        {
+            private readonly byte[] objectContents;
+            private readonly Action onDownloadStarting;
+            private readonly int failUntilAttempt;
+            private int attemptCount;
+
+            public CoalescingTestHttpGitObjects(byte[] objectContents, Action onDownloadStarting, int failUntilAttempt = 0)
+                : this(new MockGVFSEnlistment(), objectContents, onDownloadStarting, failUntilAttempt)
+            {
+            }
+
+            private CoalescingTestHttpGitObjects(MockGVFSEnlistment enlistment, byte[] objectContents, Action onDownloadStarting, int failUntilAttempt)
+                : base(new MockTracer(), enlistment, new MockCacheServerInfo(), new RetryConfig(maxRetries: 1))
+            {
+                this.objectContents = objectContents;
+                this.onDownloadStarting = onDownloadStarting;
+                this.failUntilAttempt = failUntilAttempt;
+            }
+
+            public override RetryWrapper<GitObjectTaskResult>.InvocationResult TryDownloadLooseObject(
+                string objectId,
+                bool retryOnFailure,
+                CancellationToken cancellationToken,
+                string requestSource,
+                Func<int, GitEndPointResponseData, RetryWrapper<GitObjectTaskResult>.CallbackResult> onSuccess)
+            {
+                this.onDownloadStarting?.Invoke();
+
+                int attempt = Interlocked.Increment(ref this.attemptCount);
+                if (attempt < this.failUntilAttempt)
+                {
+                    GitObjectTaskResult failResult = new GitObjectTaskResult(false);
+                    return new RetryWrapper<GitObjectTaskResult>.InvocationResult(0, false, failResult);
+                }
+
+                using (MemoryStream stream = new MemoryStream(this.objectContents))
+                using (GitEndPointResponseData response = new GitEndPointResponseData(
+                    HttpStatusCode.OK,
+                    GVFSConstants.MediaTypes.LooseObjectMediaType,
+                    stream,
+                    message: null,
+                    onResponseDisposed: null))
+                {
+                    onSuccess(0, response);
+                }
+
+                GitObjectTaskResult result = new GitObjectTaskResult(true);
+                return new RetryWrapper<GitObjectTaskResult>.InvocationResult(0, true, result);
+            }
+
+            public override RetryWrapper<GitObjectTaskResult>.InvocationResult TryDownloadObjects(
+                IEnumerable<string> objectIds,
+                Func<int, GitEndPointResponseData, RetryWrapper<GitObjectTaskResult>.CallbackResult> onSuccess,
+                Action<RetryWrapper<GitObjectTaskResult>.ErrorEventArgs> onFailure,
+                bool preferBatchedLooseObjects)
+            {
+                throw new NotImplementedException();
+            }
+
+            public override List<GitObjectSize> QueryForFileSizes(IEnumerable<string> objectIds, CancellationToken cancellationToken)
+            {
+                throw new NotImplementedException();
+            }
+        }
     }
 }