experiment: make all TCSs continue asynchronously.

Author: timothycoleman

src/Connectors/KurrentDB.Connectors.Tests/ClusterVNodeApp.cs

@@ -7,6 +7,7 @@
 using System.Net;
 using DotNext.Collections.Generic;
 using Kurrent.Surge;
+using KurrentDB.Common.Utils;
 using KurrentDB.Core;
 using KurrentDB.Core.Bus;
 using KurrentDB.Core.Certificates;
@@ -92,7 +93,7 @@ static ClusterVNodeOptions GetClusterVNodeOptions(Dictionary<string, string?> se
     class NodeReadinessProbe : IHandle<SystemMessage.SystemReady> {
 	    static readonly Serilog.ILogger Log = Serilog.Log.Logger.ForContext<NodeReadinessProbe>();
 
-	    TaskCompletionSource Ready { get; } = new();
+	    TaskCompletionSource Ready { get; } = TaskCompletionSourceFactory.CreateDefault();
 
 	    void IHandle<SystemMessage.SystemReady>.Handle(SystemMessage.SystemReady message) {
 		    if (!Ready.Task.IsCompleted)

src/Connectors/KurrentDB.Connectors.Tests/Infrastructure/Http/TestHttpMessageHandler.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
+using KurrentDB.Common.Utils;
 using static System.Threading.Tasks.TaskCreationOptions;
 
 namespace KurrentDB.Connectors.Tests.Infrastructure.Http;
@@ -10,7 +11,7 @@ protected override Task<HttpResponseMessage> SendAsync(HttpRequestMessage reques
         send(request, cancellationToken);
 
     public static TestHttpMessageHandler Create(Func<HttpRequestMessage, Task<HttpResponseMessage>> send) {
-        var tcs = new TaskCompletionSource<HttpResponseMessage>(RunContinuationsAsynchronously);
+        var tcs = TaskCompletionSourceFactory.CreateDefault<HttpResponseMessage>(RunContinuationsAsynchronously);
         return new TestHttpMessageHandler(async (req, ct) => {
             await using var registration = ct.Register(() => tcs.TrySetCanceled());
             var result = await Task.WhenAny(send(req), tcs.Task);

src/Connectors/KurrentDB.Connectors.Tests/MessageBus.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
+using KurrentDB.Common.Utils;
 using KurrentDB.Core.Bus;
 using KurrentDB.Core.Messaging;
 
@@ -19,7 +20,7 @@ public void Subscribe<T>(HandleMessage<T> handler) where T : Message =>
         });
 
     public Task SubscribeAndWait<T>(HandleMessageAsync<T> handler, CancellationToken timeoutToken = default) where T : Message {
-    	var completion = new TaskCompletionSource();
+    	var completion = TaskCompletionSourceFactory.CreateDefault();
 
     	timeoutToken.Register(() => completion.SetCanceled(timeoutToken));
 

src/Connectors/KurrentDB.Connectors.Tests/System/LeaderNodeBackgroundServiceTests.cs

@@ -6,6 +6,7 @@
 
 using System.Net;
 using Kurrent.Surge;
+using KurrentDB.Common.Utils;
 using KurrentDB.Connectors.Infrastructure.System.Node;
 using KurrentDB.Connectors.Infrastructure.System.Node.NodeSystemInfo;
 using KurrentDB.Core.Cluster;
@@ -70,7 +71,7 @@ public Task waits_for_leadership_again_when_leadership_revoked() => Fixture.Test
 
 			MessageBus.Publish(new SystemMessage.BecomeFollower(Guid.NewGuid(), FakeMemberInfo));
 
-			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = new();
+			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = TaskCompletionSourceFactory.CreateDefault<SystemMessage.ComponentTerminated>();
 			MessageBus.Subscribe<SystemMessage.ComponentTerminated>((message, _) => {
 				if (message.ComponentName == serviceName)
 					componentTerminated.SetResult(message);
@@ -123,7 +124,7 @@ public Task waits_for_leadership_again_when_leadership_revoked_multiple_times()
 			await sut.WaitUntilExecuted();
 			await Task.Delay(1000, cancellator.Token);
 
-			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = new();
+			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = TaskCompletionSourceFactory.CreateDefault<SystemMessage.ComponentTerminated>();
 			MessageBus.Subscribe<SystemMessage.ComponentTerminated>((message, _) => {
 				if (message.ComponentName == serviceName)
 					componentTerminated.SetResult(message);
@@ -178,7 +179,7 @@ public Task stops_gracefully_when_leadership_assigned_and_service_is_stopped() =
 			// Act
 			MessageBus.Publish(new SystemMessage.BecomeLeader(Guid.NewGuid()));
 
-			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = new();
+			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = TaskCompletionSourceFactory.CreateDefault<SystemMessage.ComponentTerminated>();
 			MessageBus.Subscribe<SystemMessage.ComponentTerminated>((message, _) => {
 				if (message.ComponentName == serviceName)
 					componentTerminated.SetResult(message);
@@ -234,7 +235,7 @@ public Task stops_gracefully_when_leadership_assigned_and_stopping_token_is_canc
 
 			await sut.WaitUntilExecuting();
 
-			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = new();
+			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = TaskCompletionSourceFactory.CreateDefault<SystemMessage.ComponentTerminated>();
 			MessageBus.Subscribe<SystemMessage.ComponentTerminated>((message, _) => {
 				if (message.ComponentName == serviceName)
 					componentTerminated.SetResult(message);
@@ -272,7 +273,7 @@ public Task stops_gracefully_when_leadership_revoked_and_waiting_for_leadership(
 
 			await sut.WaitUntilExecuted();
 
-			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = new();
+			TaskCompletionSource<SystemMessage.ComponentTerminated> componentTerminated = TaskCompletionSourceFactory.CreateDefault<SystemMessage.ComponentTerminated>();
 			MessageBus.Subscribe<SystemMessage.ComponentTerminated>((message, _) => {
 				if (message.ComponentName == serviceName)
 					componentTerminated.SetResult(message);
@@ -287,8 +288,8 @@ public Task stops_gracefully_when_leadership_revoked_and_waiting_for_leadership(
 
 class TestLeadershipAwareService(string serviceName, MessageBus bus, NodeSystemInfo nodeSystemInfo, ILoggerFactory loggerFactory)
 	: LeaderNodeBackgroundService(bus, bus, _ => new ValueTask<NodeSystemInfo>(nodeSystemInfo), loggerFactory, serviceName) {
-	volatile TaskCompletionSource<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)> _executingCompletionSource = new();
-	volatile TaskCompletionSource<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)> _executedCompletionSource  = new();
+	volatile TaskCompletionSource<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)> _executingCompletionSource = TaskCompletionSourceFactory.CreateDefault<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)>();
+	volatile TaskCompletionSource<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)> _executedCompletionSource  = TaskCompletionSourceFactory.CreateDefault<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)>();
 
 	public TimeSpan ExecuteDelay { get; set; } = TimeSpan.FromMinutes(10);
 
@@ -300,13 +301,13 @@ protected override async Task Execute(NodeSystemInfo nodeInfo, CancellationToken
 
 	public async Task<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)> WaitUntilExecuting() {
 		var result = await _executingCompletionSource.Task;
-		_executingCompletionSource = new();
+		_executingCompletionSource = TaskCompletionSourceFactory.CreateDefault<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)>();
 		return result;
 	}
 
 	public async Task<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)> WaitUntilExecuted() {
 		var result = await _executedCompletionSource.Task;
-		_executedCompletionSource = new();
+		_executedCompletionSource = TaskCompletionSourceFactory.CreateDefault<(NodeSystemInfo NodeInfo, CancellationToken StoppingToken)>();
 		return result;
 	}
 }

src/Connectors/KurrentDB.Connectors/Infrastructure/System/Node/NodeBackgroundService.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
+using KurrentDB.Common.Utils;
 using KurrentDB.Core.Bus;
 using KurrentDB.Core.Messages;
 using Microsoft.Extensions.Hosting;
@@ -85,7 +86,7 @@ public virtual async Task StopAsync(CancellationToken cancellationToken) {
         }
         finally {
             // Wait until the task completes or the stop token triggers
-            var completion = new TaskCompletionSource();
+            var completion = TaskCompletionSourceFactory.CreateDefault();
 
             await using var registration = cancellationToken
                 .Register(tcs => ((TaskCompletionSource)tcs!).SetCanceled(CancellationToken.None), completion);

src/Connectors/KurrentDB.Connectors/Infrastructure/System/PublisherManagementExtensions.cs

@@ -3,6 +3,7 @@
 
 // ReSharper disable CheckNamespace
 
+using KurrentDB.Common.Utils;
 using KurrentDB.Core.Bus;
 using KurrentDB.Core.Data;
 using KurrentDB.Core.Messages;
@@ -19,7 +20,7 @@ public static class PublisherManagementExtensions {
 	public static Task<(Position Position, StreamRevision Revision)> DeleteStream(this IPublisher publisher, string stream, long expectedRevision = -2, bool hardDelete = false, CancellationToken cancellationToken = default) {
 		cancellationToken.ThrowIfCancellationRequested();
 
-		var operation = new TaskCompletionSource<(Position Position, StreamRevision StreamRevision)>(TaskCreationOptions.RunContinuationsAsynchronously);
+		var operation = TaskCompletionSourceFactory.CreateDefault<(Position Position, StreamRevision StreamRevision)>(TaskCreationOptions.RunContinuationsAsynchronously);
 
 		var cid = Guid.NewGuid();
 

src/Connectors/KurrentDB.Connectors/Infrastructure/System/PublisherWriteExtensions.cs

@@ -3,6 +3,7 @@
 
 // ReSharper disable CheckNamespace
 
+using KurrentDB.Common.Utils;
 using KurrentDB.Core.Bus;
 using KurrentDB.Core.Data;
 using KurrentDB.Core.Messages;
@@ -87,7 +88,7 @@ public static async Task<WriteEventsResult> WriteEvents(
         this IPublisher publisher, string stream, Event[] events, long expectedRevision = ExpectedVersion.Any,
         CancellationToken cancellationToken = default
     ) {
-        var operation = new TaskCompletionSource<WriteEventsResult>(TaskCreationOptions.RunContinuationsAsynchronously);
+        var operation = TaskCompletionSourceFactory.CreateDefault<WriteEventsResult>(TaskCreationOptions.RunContinuationsAsynchronously);
 
         await publisher.WriteEvents(
             stream,

src/Connectors/KurrentDB.Connectors/Infrastructure/System/SystemReadinessProbe.cs

@@ -2,6 +2,7 @@
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
 using Kurrent.Surge;
+using KurrentDB.Common.Utils;
 using KurrentDB.Connectors.Infrastructure.System.Node.NodeSystemInfo;
 using KurrentDB.Core.Bus;
 using KurrentDB.Core.Messages;
@@ -15,7 +16,7 @@ public interface ISystemReadinessProbe {
 [UsedImplicitly]
 public class SystemReadinessProbe : IHandle<SystemMessage.BecomeLeader>, IHandle<SystemMessage.BecomeFollower>, IHandle<SystemMessage.BecomeReadOnlyReplica> {
     public SystemReadinessProbe(ISubscriber subscriber, GetNodeSystemInfo getNodeSystemInfo) {
-        CompletionSource = new();
+        CompletionSource = TaskCompletionSourceFactory.CreateDefault();
 
         Subscriber = subscriber.With(x => {
             x.Subscribe<SystemMessage.BecomeLeader>(this);

src/Connectors/KurrentDB.Connectors/Planes/Management/Data/ConnectorsStateProjection.cs

@@ -9,6 +9,7 @@
 using Kurrent.Surge.Connectors.Sinks;
 using KurrentDB.Core.Services.Transport.Enumerators;
 using static System.StringComparison;
+using KurrentDB.Common.Utils;
 
 namespace KurrentDB.Connectors.Planes.Management.Data;
 
@@ -101,7 +102,7 @@ public ConnectorsStateProjection(ISnapshotProjectionsStore store, string snapsho
         });
     }
 
-    TaskCompletionSource<(LogPosition Position, DateTimeOffset Timestamp)> HasCaughtUpTaskCompletionSource { get; } = new();
+    TaskCompletionSource<(LogPosition Position, DateTimeOffset Timestamp)> HasCaughtUpTaskCompletionSource { get; } = TaskCompletionSourceFactory.CreateDefault<(LogPosition Position, DateTimeOffset Timestamp)>();
 
     public Task<(LogPosition Position, DateTimeOffset Timestamp)> WaitUntilCaughtUp => HasCaughtUpTaskCompletionSource.Task;
 }

src/Connectors/KurrentDB.Connectors/Planes/Management/SystemStartupManager.cs

@@ -1,6 +1,7 @@
 // Copyright (c) Kurrent, Inc and/or licensed to Kurrent, Inc under one or more agreements.
 // Kurrent, Inc licenses this file to you under the Kurrent License v1 (see LICENSE.md).
 
+using KurrentDB.Common.Utils;
 using KurrentDB.Connectors.Infrastructure.System.Node;
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
@@ -9,7 +10,7 @@
 namespace KurrentDB.Connectors.Planes.Management;
 
 internal class SystemStartupManager(IServiceProvider serviceProvider) : BackgroundService, IStartupWorkCompletionMonitor {
-	private readonly TaskCompletionSource _completed = new();
+	private readonly TaskCompletionSource _completed = TaskCompletionSourceFactory.CreateDefault();
 
 	protected override async Task ExecuteAsync(CancellationToken stoppingToken) {
 		var workers = serviceProvider.GetServices<SystemStartupTaskWorker>().ToList();