Do not cancel operations after point of no return

src/Workspaces/Remote/Core/BrokeredServiceConnection.cs

@@ -150,6 +150,11 @@ internal static async ValueTask<TResult> InvokeStreamingServiceAsync<TResult>(
             Func<PipeReader, CancellationToken, ValueTask<TResult>> reader,
             CancellationToken cancellationToken)
         {
+            // We can cancel at entry, but once the pipe operations are scheduled we rely on both operations running to
+            // avoid deadlocks (the exception handler in 'writerTask' ensures progress is made in 'readerTask').
+            cancellationToken.ThrowIfCancellationRequested();
+            var mustNotCancelToken = CancellationToken.None;
+
             var pipe = new Pipe();
 
             // Create new tasks that both start executing, rather than invoking the delegates directly
@@ -170,7 +175,7 @@ internal static async ValueTask<TResult> InvokeStreamingServiceAsync<TResult>(
 
                     throw;
                 }
-            }, cancellationToken);
+            }, mustNotCancelToken);
 
             var readerTask = Task.Run(
                 async () =>
@@ -189,7 +194,7 @@ internal static async ValueTask<TResult> InvokeStreamingServiceAsync<TResult>(
                     {
                         await pipe.Reader.CompleteAsync(exception).ConfigureAwait(false);
                     }
-                }, cancellationToken);
+                }, mustNotCancelToken);
 
             await Task.WhenAll(writerTask, readerTask).ConfigureAwait(false);
 

src/Workspaces/Remote/Core/RemoteHostAssetSerialization.cs

@@ -88,6 +88,11 @@ static void WriteAsset(ObjectWriter writer, ISerializerService serializer, Check
 
         public static async ValueTask<ImmutableArray<(Checksum, object)>> ReadDataAsync(PipeReader pipeReader, int scopeId, ISet<Checksum> checksums, ISerializerService serializerService, CancellationToken cancellationToken)
         {
+            // We can cancel at entry, but once the pipe operations are scheduled we rely on both operations running to
+            // avoid deadlocks (the exception handler in 'copyTask' ensures progress is made in the blocking read).
+            cancellationToken.ThrowIfCancellationRequested();
+            var mustNotCancelToken = CancellationToken.None;
+
             // Workaround for ObjectReader not supporting async reading.
             // Unless we read from the RPC stream asynchronously and with cancallation support we might hang when the server cancels.
             // https://github.com/dotnet/roslyn/issues/47861
@@ -113,7 +118,7 @@ static void WriteAsset(ObjectWriter writer, ISerializerService serializer, Check
                     await localPipe.Writer.CompleteAsync(exception).ConfigureAwait(false);
                     await pipeReader.CompleteAsync(exception).ConfigureAwait(false);
                 }
-            }, cancellationToken);
+            }, mustNotCancelToken);
 
             // blocking read from the local pipe on the current thread:
             try

src/Workspaces/Remote/Core/SolutionAssetProvider.cs

@@ -52,6 +52,11 @@ public async ValueTask GetAssetsAsync(PipeWriter pipeWriter, int scopeId, Checks
                 assetMap = await assetStorage.GetAssetsAsync(scopeId, checksums, cancellationToken).ConfigureAwait(false);
             }
 
+            // We can cancel early, but once the pipe operations are scheduled we rely on both operations running to
+            // avoid deadlocks (the exception handler in 'task1' ensures progress is made in 'task2').
+            cancellationToken.ThrowIfCancellationRequested();
+            var mustNotCancelToken = CancellationToken.None;
+
             // Work around the lack of async stream writing in ObjectWriter, which is required when writing to the RPC pipe.
             // Run two tasks - the first synchronously writes to a local pipe and the second asynchronosly transfers the data to the RPC pipe.
             //
@@ -71,7 +76,7 @@ public async ValueTask GetAssetsAsync(PipeWriter pipeWriter, int scopeId, Checks
                 {
                     // no-op
                 }
-            }, cancellationToken);
+            }, mustNotCancelToken);
 
             // Complete RPC once we send the initial piece of data and start waiting for the writer to send more,
             // so the client can start reading from the stream. Once CopyPipeDataAsync completes the pipeWriter