Add failsafe to RedisLockExtension to resolve #181

src/CacheTower.Extensions.Redis/RedisLockExtension.cs

@@ -2,6 +2,7 @@
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using StackExchange.Redis;
 
@@ -87,7 +88,13 @@ public async ValueTask<CacheEntry<T>> WithRefreshAsync<T>(string cacheKey, Func<
 			}
 			else
 			{
-				var completionSource = LockedOnKeyRefresh.GetOrAdd(cacheKey, key => new TaskCompletionSource<bool>());
+				var completionSource = LockedOnKeyRefresh.GetOrAdd(cacheKey, key =>
+				{
+					var tcs = new TaskCompletionSource<bool>();
+					var cts = new CancellationTokenSource(Options.LockTimeout);
+					cts.Token.Register(() => tcs.TrySetCanceled(), useSynchronizationContext: false);
+					return tcs;
+				});
 
 				//Last minute check to confirm whether waiting is required (in case the notification is missed)
 				var currentEntry = await RegisteredStack!.GetAsync<T>(cacheKey);

tests/CacheTower.Tests/Extensions/Redis/RedisLockExtensionTests.cs

@@ -207,5 +207,47 @@ public async Task ObservedLockMultiple()
 
 			cacheStackMock.Verify(c => c.GetAsync<int>("TestKey"), Times.Exactly(4), "Two checks to the cache stack are expected");
 		}
+
+		[TestMethod]
+		public async Task FailsafeOnSubscriberFailure()
+		{
+			RedisHelper.ResetState();
+
+			var connection = RedisHelper.GetConnection();
+
+			var cacheStackMock = new Mock<ICacheStack>();
+			var extension = new RedisLockExtension(connection, new RedisLockOptions(lockTimeout: TimeSpan.FromSeconds(1)));
+			extension.Register(cacheStackMock.Object);
+
+			var cacheEntry = new CacheEntry<int>(13, TimeSpan.FromDays(1));
+
+			//Establish lock
+			await connection.GetDatabase().StringSetAsync("Lock:TestKey", RedisValue.EmptyString);
+
+			var refreshTask = extension.WithRefreshAsync("TestKey",
+				() =>
+				{
+					return new ValueTask<CacheEntry<int>>(cacheEntry);
+				},
+				new CacheSettings(TimeSpan.FromDays(1))
+			).AsTask();
+
+			//Delay to allow for Redis check and self-entry into lock
+			await Task.Delay(TimeSpan.FromSeconds(1));
+
+			Assert.IsTrue(extension.LockedOnKeyRefresh.ContainsKey("TestKey"), "Lock was not established");
+
+			//We don't publish to end lock
+
+			//However, we still expect to succeed
+			var succeedingTask = await Task.WhenAny(refreshTask, Task.Delay(TimeSpan.FromSeconds(10)));
+			if (!succeedingTask.Equals(refreshTask))
+			{
+				RedisHelper.DebugInfo(connection);
+				Assert.Fail("Refresh has timed out - something has gone very wrong");
+			}
+
+			cacheStackMock.Verify(c => c.GetAsync<int>("TestKey"), Times.Exactly(1), "One checks to the cache stack are expected as it will fail to resolve lock");
+		}
 	}
 }

tests/CacheTower.Tests/Utils/RedisHelper.cs

@@ -43,6 +43,7 @@ public static ConnectionMultiplexer GetConnection()
 		public static void ResetState()
 		{
 			GetConnection().GetServer(Endpoint).FlushDatabase();
+			GetConnection().GetSubscriber().UnsubscribeAll();
 
 			//.NET Framework doesn't support `Clear()` on Errors so we do it manually
 			while (!Errors.IsEmpty)