Fix up boadcasts, tests, and logging across the board (#1400)

This changes does the following:
- Moves the broadcast to be only before the master reconfiguration to both before *and* after - a fix following https://github.com/StackExchange/StackExchange.Redis/commit/88dcf0c989b25623fb8ffb1f4d48c24593246cfe to fix the gap.
- Tweaks tests by overall lessening runtime (and thus build server time). Overall, fixes a few static timeouts to be conditional (so they short circuit faster if met), and brings some tests that were some variant of the above into RELEASE since they're safe now.
- Changes `UntilCondition` to take a `TimeSpan`, just because clearer. Even though I IntelliSense completed `.FromMinutes()` earlier and watched it like an idiot for a while...I stand by this decision!
- Locks the `ITestOutputHelper` writer because...that was jacked up in races:

![off to the races](https://user-images.githubusercontent.com/454813/77232395-2f799980-6b77-11ea-8fb4-0398de25e313.png)

------

Note: a lot of the test changes are just optimizations to delays which allow longer but short-circuit sooner. The important changes are in broadcast and locking around the test runner. I can think of downsides to neither, but want some @mgravell eyes. This should resolve a lot of flaky-ness with local (and build agent) tests. Not all of it, but a lot of it!

src/StackExchange.Redis/ConnectionMultiplexer.cs

@@ -453,22 +453,31 @@ internal void MakeMaster(ServerEndPoint server, ReplicationChangeOptions options
             // ConfigurationOptions.ConfigCheckSeconds interval to identify the current (created by this method call) topology correctly.
             var blockingReconfig = Interlocked.CompareExchange(ref activeConfigCause, "Block: Pending Master Reconfig", null) == null;
 
-            // try and broadcast this everywhere, to catch the maximum audience
-            if ((options & ReplicationChangeOptions.Broadcast) != 0 && ConfigurationChangedChannel != null
-                && CommandMap.IsAvailable(RedisCommand.PUBLISH))
-            {
-                RedisValue channel = ConfigurationChangedChannel;
-                foreach (var node in nodes)
+            // Try and broadcast the fact a change happened to all members
+            // We want everyone possible to pick it up.
+            // We broadcast before *and after* the change to remote members, so that they don't go without detecting a change happened.
+            // This eliminates the race of pub/sub *then* re-slaving happening, since a method both preceeds and follows.
+            void Broadcast(ReadOnlySpan<ServerEndPoint> serverNodes)
+            {
+                if ((options & ReplicationChangeOptions.Broadcast) != 0 && ConfigurationChangedChannel != null
+                    && CommandMap.IsAvailable(RedisCommand.PUBLISH))
                 {
-                    if (!node.IsConnected) continue;
-                    log?.WriteLine($"Broadcasting via {Format.ToString(node.EndPoint)}...");
-                    msg = Message.Create(-1, flags, RedisCommand.PUBLISH, channel, newMaster);
+                    RedisValue channel = ConfigurationChangedChannel;
+                    foreach (var node in serverNodes)
+                    {
+                        if (!node.IsConnected) continue;
+                        log?.WriteLine($"Broadcasting via {Format.ToString(node.EndPoint)}...");
+                        msg = Message.Create(-1, flags, RedisCommand.PUBLISH, channel, newMaster);
 #pragma warning disable CS0618
-                    node.WriteDirectFireAndForgetSync(msg, ResultProcessor.Int64);
+                        node.WriteDirectFireAndForgetSync(msg, ResultProcessor.Int64);
 #pragma warning restore CS0618
+                    }
                 }
             }
 
+            // Send a message before it happens - because afterwards a new slave may be unresponsive
+            Broadcast(nodes);
+
             if ((options & ReplicationChangeOptions.EnslaveSubordinates) != 0)
             {
                 foreach (var node in nodes)
@@ -483,6 +492,11 @@ internal void MakeMaster(ServerEndPoint server, ReplicationChangeOptions options
                 }
             }
 
+            // ...and send one after it happens - because the first broadcast may have landed on a secondary client
+            // and it can reconfgure before any topology change actually happened. This is most likely to happen
+            // in low-latency environments.
+            Broadcast(nodes);
+
             // and reconfigure the muxer
             log?.WriteLine("Reconfiguring all endpoints...");
             // Yes, there is a tiny latency race possible between this code and the next call, but it's far more minute than before.

src/StackExchange.Redis/Interfaces/IServer.cs

@@ -771,8 +771,7 @@ public partial interface IServer : IRedis
         KeyValuePair<string, string>[] SentinelMaster(string serviceName, CommandFlags flags = CommandFlags.None);
 
         /// <summary>
-        /// Force a failover as if the master was not reachable, and without asking for agreement to other Sentinels 
-        /// (however a new version of the configuration will be published so that the other Sentinels will update their configurations).
+        /// Show the state and info of the specified master.
         /// </summary>
         /// <param name="serviceName">The sentinel service name.</param>
         /// <param name="flags">The command flags to use.</param>

tests/StackExchange.Redis.Tests/BasicOps.cs

@@ -292,12 +292,13 @@ public async Task TestSevered()
                 string key = Guid.NewGuid().ToString();
                 db.KeyDelete(key, CommandFlags.FireAndForget);
                 db.StringSet(key, key, flags: CommandFlags.FireAndForget);
-                GetServer(muxer).SimulateConnectionFailure();
+                var server = GetServer(muxer);
+                server.SimulateConnectionFailure();
                 var watch = Stopwatch.StartNew();
-                db.Ping();
+                await UntilCondition(TimeSpan.FromSeconds(10), () => server.IsConnected);
                 watch.Stop();
                 Log("Time to re-establish: {0}ms (any order)", watch.ElapsedMilliseconds);
-                await Task.Delay(2000).ForAwait();
+                await UntilCondition(TimeSpan.FromSeconds(10), () => key == db.StringGet(key));
                 Debug.WriteLine("Pinging...");
                 Assert.Equal(key, db.StringGet(key));
             }

tests/StackExchange.Redis.Tests/ConnectFailTimeout.cs

-using System.Threading;
+using System;
 using System.Threading.Tasks;
 using Xunit;
 using Xunit.Abstractions;
@@ -9,7 +9,6 @@ public class ConnectFailTimeout : TestBase
     {
         public ConnectFailTimeout(ITestOutputHelper output) : base (output) { }
 
-#if DEBUG
         [Fact]
         public async Task NoticesConnectFail()
         {
@@ -32,7 +31,7 @@ public async Task NoticesConnectFail()
                 Assert.Throws<RedisConnectionException>(() => server.Ping());
                 Log("pinged");
                 // Heartbeat should reconnect by now
-                await Task.Delay(5000).ConfigureAwait(false);
+                await UntilCondition(TimeSpan.FromSeconds(10), () => server.IsConnected);
 
                 Log("pinging - expect success");
                 var time = server.Ping();
@@ -40,6 +39,5 @@ public async Task NoticesConnectFail()
                 Log(time.ToString());
             }
         }
-#endif
     }
 }

tests/StackExchange.Redis.Tests/ConnectToUnexistingHost.cs

@@ -46,7 +46,7 @@ public void CanNotOpenNonsenseConnection_IP()
         {
             var ex = Assert.Throws<RedisConnectionException>(() =>
             {
-                using (var conn = ConnectionMultiplexer.Connect(TestConfig.Current.MasterServer + ":6500", Writer)) { }
+                using (var conn = ConnectionMultiplexer.Connect(TestConfig.Current.MasterServer + ":6500,connectTimeout=1000", Writer)) { }
             });
             Log(ex.ToString());
         }
@@ -56,7 +56,7 @@ public async Task CanNotOpenNonsenseConnection_DNS()
         {
             var ex = await Assert.ThrowsAsync<RedisConnectionException>(async () =>
             {
-                using (var conn = await ConnectionMultiplexer.ConnectAsync($"doesnot.exist.ds.{Guid.NewGuid():N}.com:6500", Writer).ForAwait()) { }
+                using (var conn = await ConnectionMultiplexer.ConnectAsync($"doesnot.exist.ds.{Guid.NewGuid():N}.com:6500,connectTimeout=1000", Writer).ForAwait()) { }
             }).ForAwait();
             Log(ex.ToString());
         }
@@ -64,7 +64,7 @@ public async Task CanNotOpenNonsenseConnection_DNS()
         [Fact]
         public void CreateDisconnectedNonsenseConnection_IP()
         {
-            using (var conn = ConnectionMultiplexer.Connect(TestConfig.Current.MasterServer + ":6500,abortConnect=false", Writer))
+            using (var conn = ConnectionMultiplexer.Connect(TestConfig.Current.MasterServer + ":6500,abortConnect=false,connectTimeout=1000", Writer))
             {
                 Assert.False(conn.GetServer(conn.GetEndPoints().Single()).IsConnected);
                 Assert.False(conn.GetDatabase().IsConnected(default(RedisKey)));
@@ -74,7 +74,7 @@ public void CreateDisconnectedNonsenseConnection_IP()
         [Fact]
         public void CreateDisconnectedNonsenseConnection_DNS()
         {
-            using (var conn = ConnectionMultiplexer.Connect($"doesnot.exist.ds.{Guid.NewGuid():N}.com:6500, abortConnect=false", Writer))
+            using (var conn = ConnectionMultiplexer.Connect($"doesnot.exist.ds.{Guid.NewGuid():N}.com:6500,abortConnect=false,connectTimeout=1000", Writer))
             {
                 Assert.False(conn.GetServer(conn.GetEndPoints().Single()).IsConnected);
                 Assert.False(conn.GetDatabase().IsConnected(default(RedisKey)));

tests/StackExchange.Redis.Tests/ConnectingFailDetection.cs

-using System.Threading;
+using System;
+using System.Threading;
 using System.Threading.Tasks;
 using Xunit;
 using Xunit.Abstractions;
@@ -113,7 +114,7 @@ public void ConnectsWhenBeginConnectCompletesSynchronously()
         }
 
         [Fact]
-        public void Issue922_ReconnectRaised()
+        public async Task Issue922_ReconnectRaised()
         {
             var config = ConfigurationOptions.Parse(TestConfig.Current.MasterServerAndPort);
             config.AbortOnConnectFail = true;
@@ -130,15 +131,14 @@ public void Issue922_ReconnectRaised()
                 muxer.ConnectionRestored += delegate { Interlocked.Increment(ref restoreCount); };
 
                 var db = muxer.GetDatabase();
-                db.Ping();
                 Assert.Equal(0, Volatile.Read(ref failCount));
                 Assert.Equal(0, Volatile.Read(ref restoreCount));
 
                 var server = muxer.GetServer(TestConfig.Current.MasterServerAndPort);
                 server.SimulateConnectionFailure();
-                Thread.Sleep(1000);
 
-                db.Ping(); // interactive+subscriber = 2
+                await UntilCondition(TimeSpan.FromSeconds(10), () => Volatile.Read(ref failCount) + Volatile.Read(ref restoreCount) == 4);
+                // interactive+subscriber = 2
                 Assert.Equal(2, Volatile.Read(ref failCount));
                 Assert.Equal(2, Volatile.Read(ref restoreCount));
             }

tests/StackExchange.Redis.Tests/ExceptionFactoryTests.cs

@@ -35,13 +35,12 @@ public void NullSnapshot()
             Assert.Null(ex.InnerException);
         }
 
-#if DEBUG
         [Fact]
         public void MultipleEndpointsThrowConnectionException()
         {
             try
             {
-                using (var muxer = Create(keepAlive: 1, connectTimeout: 10000, allowAdmin: true))
+                using (var muxer = Create(keepAlive: 1, connectTimeout: 10000, allowAdmin: true, shared: false))
                 {
                     var conn = muxer.GetDatabase();
                     muxer.AllowConnect = false;
@@ -55,7 +54,8 @@ public void MultipleEndpointsThrowConnectionException()
                     var outer = Assert.IsType<RedisConnectionException>(ex);
                     Assert.Equal(ConnectionFailureType.UnableToResolvePhysicalConnection, outer.FailureType);
                     var inner = Assert.IsType<RedisConnectionException>(outer.InnerException);
-                    Assert.Equal(ConnectionFailureType.SocketFailure, inner.FailureType);
+                    Assert.True(inner.FailureType == ConnectionFailureType.SocketFailure
+                             || inner.FailureType == ConnectionFailureType.InternalFailure);
                 }
             }
             finally
@@ -69,7 +69,7 @@ public void ServerTakesPrecendenceOverSnapshot()
         {
             try
             {
-                using (var muxer = Create(keepAlive: 1, connectTimeout: 10000, allowAdmin: true))
+                using (var muxer = Create(keepAlive: 1, connectTimeout: 10000, allowAdmin: true, shared: false))
                 {
                     var conn = muxer.GetDatabase();
                     muxer.AllowConnect = false;
@@ -87,7 +87,6 @@ public void ServerTakesPrecendenceOverSnapshot()
                 ClearAmbientFailures();
             }
         }
-#endif
 
         [Fact]
         public void NullInnerExceptionForMultipleEndpointsWithNoLastException()

tests/StackExchange.Redis.Tests/Failover.cs

-using System.IO;
+using System;
+using System.IO;
 using System.Threading;
 using System.Threading.Tasks;
 using Xunit;
@@ -103,6 +104,7 @@ public async Task ConfigVerifyReceiveConfigChangeBroadcast()
             }
         }
 
+
         [Fact]
         public async Task DeslaveGoesToPrimary()
         {
@@ -181,6 +183,8 @@ public async Task DeslaveGoesToPrimary()
                     Assert.Equal(secondary2.EndPoint, db2.IdentifyEndpoint(key, CommandFlags.DemandSlave));
                 }
 
+                await UntilCondition(TimeSpan.FromSeconds(20), () => !primary.IsSlave && secondary.IsSlave);
+
                 Assert.False(primary.IsSlave, $"{primary.EndPoint} should be a master.");
                 Assert.True(secondary.IsSlave, $"{secondary.EndPoint} should be a slave.");
 
@@ -205,6 +209,7 @@ public async Task SubscriptionsSurviveMasterSwitchAsync()
             using (var b = Create(allowAdmin: true, shared: false))
             {
                 RedisChannel channel = Me();
+                Log("Using Channel: " + channel);
                 var subA = a.GetSubscriber();
                 var subB = b.GetSubscriber();
 
@@ -247,7 +252,7 @@ public async Task SubscriptionsSurviveMasterSwitchAsync()
                 Log("  SubA ping: " + subA.Ping());
                 Log("  SubB ping: " + subB.Ping());
                 // If redis is under load due to this suite, it may take a moment to send across.
-                await UntilCondition(5000, () => Interlocked.Read(ref aCount) == 2 && Interlocked.Read(ref bCount) == 2).ForAwait();
+                await UntilCondition(TimeSpan.FromSeconds(5), () => Interlocked.Read(ref aCount) == 2 && Interlocked.Read(ref bCount) == 2).ForAwait();
 
                 Assert.Equal(2, Interlocked.Read(ref aCount));
                 Assert.Equal(2, Interlocked.Read(ref bCount));
@@ -264,7 +269,8 @@ public async Task SubscriptionsSurviveMasterSwitchAsync()
                         a.GetServer(TestConfig.Current.FailoverSlaveServerAndPort).MakeMaster(ReplicationChangeOptions.All, sw);
                         Log(sw.ToString());
                     }
-                    await UntilCondition(3000, () => b.GetServer(TestConfig.Current.FailoverMasterServerAndPort).IsSlave).ForAwait();
+                    Log("Waiting for connection B to detect...");
+                    await UntilCondition(TimeSpan.FromSeconds(10), () => b.GetServer(TestConfig.Current.FailoverMasterServerAndPort).IsSlave).ForAwait();
                     subA.Ping();
                     subB.Ping();
                     Log("Falover 2 Attempted. Pausing...");
@@ -281,10 +287,19 @@ public async Task SubscriptionsSurviveMasterSwitchAsync()
 
                     Assert.True(a.GetServer(TestConfig.Current.FailoverMasterServerAndPort).IsSlave, $"A Connection: {TestConfig.Current.FailoverMasterServerAndPort} should be a slave");
                     Assert.False(a.GetServer(TestConfig.Current.FailoverSlaveServerAndPort).IsSlave, $"A Connection: {TestConfig.Current.FailoverSlaveServerAndPort} should be a master");
+                    await UntilCondition(TimeSpan.FromSeconds(10), () => b.GetServer(TestConfig.Current.FailoverMasterServerAndPort).IsSlave).ForAwait();
                     var sanityCheck = b.GetServer(TestConfig.Current.FailoverMasterServerAndPort).IsSlave;
                     if (!sanityCheck)
                     {
-                        Skip.Inconclusive("Not enough latency.");
+                        Log("FAILURE: B has not detected the topology change.");
+                        foreach (var server in b.GetServerSnapshot().ToArray())
+                        {
+                            Log("  Server" + server.EndPoint);
+                            Log("    State: " + server.ConnectionState);
+                            Log("    IsMaster: " + !server.IsSlave);
+                            Log("    Type: " + server.ServerType);
+                        }
+                        //Skip.Inconclusive("Not enough latency.");
                     }
                     Assert.True(sanityCheck, $"B Connection: {TestConfig.Current.FailoverMasterServerAndPort} should be a slave");
                     Assert.False(b.GetServer(TestConfig.Current.FailoverSlaveServerAndPort).IsSlave, $"B Connection: {TestConfig.Current.FailoverSlaveServerAndPort} should be a master");
@@ -294,18 +309,20 @@ public async Task SubscriptionsSurviveMasterSwitchAsync()
                     Log("  B outstanding: " + b.GetCounters().TotalOutstanding);
                     subA.Ping();
                     subB.Ping();
-                    await Task.Delay(1000).ForAwait();
+                    await Task.Delay(5000).ForAwait();
                     epA = subA.SubscribedEndpoint(channel);
                     epB = subB.SubscribedEndpoint(channel);
                     Log("Subscription complete");
                     Log("  A: " + EndPointCollection.ToString(epA));
                     Log("  B: " + EndPointCollection.ToString(epB));
-                    Log("  A2 sent to: " + subA.Publish(channel, "A2"));
-                    Log("  B2 sent to: " + subB.Publish(channel, "B2"));
+                    var aSentTo = subA.Publish(channel, "A2");
+                    var bSentTo = subB.Publish(channel, "B2");
+                    Log("  A2 sent to: " + aSentTo);
+                    Log("  B2 sent to: " + bSentTo);
                     subA.Ping();
                     subB.Ping();
                     Log("Ping Complete. Checking...");
-                    await UntilCondition(10000, () => Interlocked.Read(ref aCount) == 2 && Interlocked.Read(ref bCount) == 2).ForAwait();
+                    await UntilCondition(TimeSpan.FromSeconds(10), () => Interlocked.Read(ref aCount) == 2 && Interlocked.Read(ref bCount) == 2).ForAwait();
 
                     Log("Counts so far:");
                     Log("  aCount: " + Interlocked.Read(ref aCount));
@@ -314,8 +331,15 @@ public async Task SubscriptionsSurviveMasterSwitchAsync()
 
                     Assert.Equal(2, Interlocked.Read(ref aCount));
                     Assert.Equal(2, Interlocked.Read(ref bCount));
-                    // Expect 6, because a sees a, but b sees a and b due to replication
-                    Assert.Equal(6, Interlocked.CompareExchange(ref masterChanged, 0, 0));
+                    // Expect 10, because a sees a, but b sees a and b due to replication
+                    Assert.Equal(10, Interlocked.CompareExchange(ref masterChanged, 0, 0));
+                }
+                catch
+                {
+                    LogNoTime("");
+                    Log("ERROR: Something went bad - see above! Roooooolling back. Back it up. Baaaaaack it on up.");
+                    LogNoTime("");
+                    throw;
                 }
                 finally
                 {

tests/StackExchange.Redis.Tests/Issues/Issue1101.cs

-using System.Collections.Generic;
+using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Threading;
 using System.Threading.Tasks;
@@ -40,9 +41,9 @@ public async Task ExecuteWithUnsubscribeViaChannel()
                     return Task.CompletedTask;
                 });
                 second.OnMessage(_ => Interlocked.Increment(ref i));
-                await Task.Delay(100);
+                await Task.Delay(200);
                 await pubsub.PublishAsync(name, "abc");
-                await Task.Delay(100);
+                await UntilCondition(TimeSpan.FromSeconds(10), () => values.Count == 1);
                 lock (values)
                 {
                     Assert.Equal("abc", Assert.Single(values));
@@ -53,9 +54,9 @@ public async Task ExecuteWithUnsubscribeViaChannel()
                 Assert.False(second.Completion.IsCompleted, "completed");
 
                 await first.UnsubscribeAsync();
-                await Task.Delay(100);
+                await Task.Delay(200);
                 await pubsub.PublishAsync(name, "def");
-                await Task.Delay(100);
+                await UntilCondition(TimeSpan.FromSeconds(10), () => values.Count == 1 && Volatile.Read(ref i) == 2);
                 lock (values)
                 {
                     Assert.Equal("abc", Assert.Single(values));
@@ -66,9 +67,9 @@ public async Task ExecuteWithUnsubscribeViaChannel()
                 AssertCounts(pubsub, name, true, 0, 1);
 
                 await second.UnsubscribeAsync();
-                await Task.Delay(100);
+                await Task.Delay(200);
                 await pubsub.PublishAsync(name, "ghi");
-                await Task.Delay(100);
+                await UntilCondition(TimeSpan.FromSeconds(10), () => values.Count == 1);
                 lock (values)
                 {
                     Assert.Equal("abc", Assert.Single(values));
@@ -110,7 +111,7 @@ public async Task ExecuteWithUnsubscribeViaSubscriber()
 
                 await Task.Delay(100);
                 await pubsub.PublishAsync(name, "abc");
-                await Task.Delay(100);
+                await UntilCondition(TimeSpan.FromSeconds(10), () => values.Count == 1);
                 lock (values)
                 {
                     Assert.Equal("abc", Assert.Single(values));
@@ -123,7 +124,7 @@ public async Task ExecuteWithUnsubscribeViaSubscriber()
                 await pubsub.UnsubscribeAsync(name);
                 await Task.Delay(100);
                 await pubsub.PublishAsync(name, "def");
-                await Task.Delay(100);
+                await UntilCondition(TimeSpan.FromSeconds(10), () => values.Count == 1);
                 lock (values)
                 {
                     Assert.Equal("abc", Assert.Single(values));
@@ -161,7 +162,7 @@ public async Task ExecuteWithUnsubscribeViaClearAll()
                 second.OnMessage(_ => Interlocked.Increment(ref i));
                 await Task.Delay(100);
                 await pubsub.PublishAsync(name, "abc");
-                await Task.Delay(100);
+                await UntilCondition(TimeSpan.FromSeconds(10), () => values.Count == 1);
                 lock (values)
                 {
                     Assert.Equal("abc", Assert.Single(values));
@@ -174,7 +175,7 @@ public async Task ExecuteWithUnsubscribeViaClearAll()
                 await pubsub.UnsubscribeAllAsync();
                 await Task.Delay(100);
                 await pubsub.PublishAsync(name, "def");
-                await Task.Delay(100);
+                await UntilCondition(TimeSpan.FromSeconds(10), () => values.Count == 1);
                 lock (values)
                 {
                     Assert.Equal("abc", Assert.Single(values));

tests/StackExchange.Redis.Tests/Issues/SO24807536.cs

@@ -21,7 +21,7 @@ public async Task Exec()
                 // setup some data
                 cache.KeyDelete(key, CommandFlags.FireAndForget);
                 cache.HashSet(key, "full", "some value", flags: CommandFlags.FireAndForget);
-                cache.KeyExpire(key, TimeSpan.FromSeconds(3), CommandFlags.FireAndForget);
+                cache.KeyExpire(key, TimeSpan.FromSeconds(1), CommandFlags.FireAndForget);
 
                 // test while exists
                 var keyExists = cache.KeyExists(key);
@@ -32,7 +32,7 @@ public async Task Exec()
                 Assert.Equal("some value", fullWait.Result);
 
                 // wait for expiry
-                await Task.Delay(4000).ForAwait();
+                await Task.Delay(2000).ForAwait();
 
                 // test once expired
                 keyExists = cache.KeyExists(key);

tests/StackExchange.Redis.Tests/Migrate.cs

@@ -30,11 +30,14 @@ public async Task Basic()
                 toDb.KeyDelete(key, CommandFlags.FireAndForget);
                 fromDb.StringSet(key, "foo", flags: CommandFlags.FireAndForget);
                 var dest = to.GetEndPoints(true).Single();
-                fromDb.KeyMigrate(key, dest);
-                await Task.Delay(1000); // this is *meant* to be synchronous at the redis level, but
+                fromDb.KeyMigrate(key, dest, migrateOptions: MigrateOptions.Replace);
+
+                // this is *meant* to be synchronous at the redis level, but
                 // we keep seeing it fail on the CI server where the key has *left* the origin, but
                 // has *not* yet arrived at the destination; adding a pause while we investigate with
                 // the redis folks
+                await UntilCondition(TimeSpan.FromSeconds(5), () => !fromDb.KeyExists(key) && toDb.KeyExists(key));
+
                 Assert.False(fromDb.KeyExists(key));
                 Assert.True(toDb.KeyExists(key));
                 string s = toDb.StringGet(key);

tests/StackExchange.Redis.Tests/PubSub.cs

@@ -27,21 +27,20 @@ public async Task ExplicitPublishMode()
                 pub.Subscribe(new RedisChannel("ab*d", RedisChannel.PatternMode.Auto), (x, y) => Interlocked.Increment(ref c));
                 pub.Subscribe("abc*", (x, y) => Interlocked.Increment(ref d));
 
-                await Task.Delay(4100).ForAwait();
+                await Task.Delay(1000).ForAwait();
                 pub.Publish("abcd", "efg");
-                await Task.Delay(500).ForAwait();
+                await UntilCondition(TimeSpan.FromSeconds(10),
+                    () => Thread.VolatileRead(ref b) == 1
+                       && Thread.VolatileRead(ref c) == 1
+                       && Thread.VolatileRead(ref d) == 1);
                 Assert.Equal(0, Thread.VolatileRead(ref a));
                 Assert.Equal(1, Thread.VolatileRead(ref b));
                 Assert.Equal(1, Thread.VolatileRead(ref c));
                 Assert.Equal(1, Thread.VolatileRead(ref d));
 
                 pub.Publish("*bcd", "efg");
-                await Task.Delay(500).ForAwait();
+                await UntilCondition(TimeSpan.FromSeconds(10), () => Thread.VolatileRead(ref a) == 1);
                 Assert.Equal(1, Thread.VolatileRead(ref a));
-                //Assert.Equal(1, Thread.VolatileRead(ref b));
-                //Assert.Equal(1, Thread.VolatileRead(ref c));
-                //Assert.Equal(1, Thread.VolatileRead(ref d));
-
             }
         }
 

tests/StackExchange.Redis.Tests/TestBase.cs

@@ -38,16 +38,22 @@ protected TestBase(ITestOutputHelper output, SharedConnectionFixture fixture = n
         protected void LogNoTime(string message) => LogNoTime(Writer, message);
         internal static void LogNoTime(TextWriter output, string message)
         {
-            output.WriteLine(message);
+            lock (output)
+            {
+                output.WriteLine(message);
+            }
             if (TestConfig.Current.LogToConsole)
             {
                 Console.WriteLine(message);
             }
         }
-        protected void Log(string message) => Log(Writer, message);
+        protected void Log(string message) => LogNoTime(Writer, message);
         public static void Log(TextWriter output, string message)
         {
-            output?.WriteLine(Time() + ": " + message);
+            lock (output)
+            {
+                output?.WriteLine(Time() + ": " + message);
+            }
             if (TestConfig.Current.LogToConsole)
             {
                 Console.WriteLine(message);
@@ -55,7 +61,10 @@ public static void Log(TextWriter output, string message)
         }
         protected void Log(string message, params object[] args)
         {
-            Output.WriteLine(Time() + ": " + message, args);
+            lock (Output)
+            {
+                Output.WriteLine(Time() + ": " + message, args);
+            }
             if (TestConfig.Current.LogToConsole)
             {
                 Console.WriteLine(message, args);
@@ -409,13 +418,15 @@ void callback()
             return watch.Elapsed;
         }
 
-        protected async Task UntilCondition(int maxMilliseconds, Func<bool> predicate, int perLoop = 100)
+        private static readonly TimeSpan DefaultWaitPerLoop = TimeSpan.FromMilliseconds(50);
+        protected async Task UntilCondition(TimeSpan maxWaitTime, Func<bool> predicate, TimeSpan? waitPerLoop = null)
         {
-            var spent = 0;
-            while (spent < maxMilliseconds && !predicate())
+            TimeSpan spent = TimeSpan.Zero;
+            while (spent < maxWaitTime && !predicate())
             {
-                await Task.Delay(perLoop).ForAwait();
-                spent += perLoop;
+                var wait = waitPerLoop ?? DefaultWaitPerLoop;
+                await Task.Delay(wait).ForAwait();
+                spent += wait;
             }
         }
     }