From 4384529612746622965a58ee25d9b401d24d0d40 Mon Sep 17 00:00:00 2001 From: Cosmin Staicu Date: Wed, 23 Sep 2026 15:52:09 +0300 Subject: [PATCH] fix(cache): a refused telemetry report no longer defeats the catch that made it Each of these catches exists so the surrounding work carries on, and each reported the caught failure through the telemetry sink directly, so a sink that throws took the recovery with it. They now go through TryTrackException and TryTrackEvent, which gains an optional properties parameter for the lock sites. - RedisDistributedLock: acquire, release, acquired and timeout. A refusal handed the caller an exception instead of the no-op lease, or left a lock Redis had granted held until it expired. - FactoryTimeout: a refusal replaced the TimeoutException callers expect. - MemoryCacheSetter: a refusal in the finally replaced the exception on its way out. - RedisConnectionWarmup: the one thing that could make best effort not be. - RehydrationCoordinator: a refused trigger skipped the rehydrate, and a refused timed-out or failed outcome released the per-key locks instead of holding them for the cooldown. - RedisStreamSubjectWriter: the entry is acknowledged before anything is reported, so a refusal cannot leave it pending. - RedisConnector: a refused stale-endpoint member report counted as a failed scan and backed off the next one. No raw TrackEvent or TrackException call remains in src outside TelemetrySafeguards. Each new test fails against the raw call it guards. Signed-off-by: Cosmin Staicu --- .../Redis/RedisStreamSubjectWriter.cs | 10 +- src/UiPath.Caching/FactoryTimeout.cs | 2 +- .../Locking/RedisDistributedLock.cs | 14 +-- src/UiPath.Caching/MemoryCacheSetter.cs | 2 +- .../Redis/RedisConnectionWarmup.cs | 2 +- src/UiPath.Caching/Redis/RedisConnector.cs | 4 +- src/UiPath.Caching/RehydrationCoordinator.cs | 10 +- src/UiPath.Caching/TelemetrySafeguards.cs | 4 +- .../RedisStreamSubjectWriterTests.cs | 75 ++++++++++- .../FactoryTimeoutTests.cs | 44 +++++++ .../Locking/RedisDistributedLockTests.cs | 91 ++++++++++++++ .../MemoryCacheSetterTests.cs | 38 ++++++ .../Redis/RedisConnectorStaleEndpointTests.cs | 26 +++- .../RehydrationCoordinatorTests.cs | 118 +++++++++++++++++- .../Telemetry/RefusingTelemetryProvider.cs | 65 ++++++++++ 15 files changed, 477 insertions(+), 28 deletions(-) create mode 100644 tests/UiPath.Caching.Tests/FactoryTimeoutTests.cs create mode 100644 tests/UiPath.Caching.Tests/Telemetry/RefusingTelemetryProvider.cs diff --git a/src/UiPath.Caching/Broadcast/Redis/RedisStreamSubjectWriter.cs b/src/UiPath.Caching/Broadcast/Redis/RedisStreamSubjectWriter.cs index e9bb63ef..9e5803dd 100644 --- a/src/UiPath.Caching/Broadcast/Redis/RedisStreamSubjectWriter.cs +++ b/src/UiPath.Caching/Broadcast/Redis/RedisStreamSubjectWriter.cs @@ -294,10 +294,11 @@ private ValueTask ProcessEvent(StreamEntry @event, List ids) if (ev.SameSource(_context.SourceUri)) { + // Acknowledged first, so a throw below cannot leave the entry pending. + ids.Add(@event.Id); LogEventFromCurrentSource(ev.Id, _context.Topic, @event.Id); _cachingTelemetryProvider.TrackTopicReadMetric(_context.Topic.ToString(), @event.Id); TraceReceipt(ev); - ids.Add(@event.Id); return default; } @@ -342,13 +343,14 @@ private async ValueTask DispatchValidEventAsync(T ev, StreamEntry @event, List ids) { - _cachingTelemetryProvider.TrackEvent(EventInvalid, + // Acknowledged first, so a throw below cannot leave the entry pending. + ids.Add(@event.Id); + _cachingTelemetryProvider.TryTrackEvent(EventInvalid, [ new(PropTopicKey, _context.Topic.ToString()), new(PropTransportId, @event.Id.ToString()), ]); LogEventInvalid(ev.Id, _context.Topic, @event.Id); - ids.Add(@event.Id); } private void TraceReceipt(T ev) @@ -357,7 +359,7 @@ private void TraceReceipt(T ev) if (_context.EmitStreamReceivedEvent) { - _cachingTelemetryProvider.TrackEvent(EventReceived, + _cachingTelemetryProvider.TryTrackEvent(EventReceived, [ new(PropEventId, ev.Id!), new(PropTopicKey, _context.Topic.ToString()), diff --git a/src/UiPath.Caching/FactoryTimeout.cs b/src/UiPath.Caching/FactoryTimeout.cs index 319362c9..f9662a2e 100644 --- a/src/UiPath.Caching/FactoryTimeout.cs +++ b/src/UiPath.Caching/FactoryTimeout.cs @@ -37,7 +37,7 @@ public static async Task RunAsync( } catch (OperationCanceledException) when (linkedCts.IsCancellationRequested && !token.IsCancellationRequested) { - telemetry.TrackEvent(EventName, + telemetry.TryTrackEvent(EventName, [ new(TagCacheName, cacheName), new(TagCacheKey, cacheKey.Name), diff --git a/src/UiPath.Caching/Locking/RedisDistributedLock.cs b/src/UiPath.Caching/Locking/RedisDistributedLock.cs index 8abd977c..069bbc9c 100644 --- a/src/UiPath.Caching/Locking/RedisDistributedLock.cs +++ b/src/UiPath.Caching/Locking/RedisDistributedLock.cs @@ -72,8 +72,8 @@ public async ValueTask AcquireAsync(string key, TimeSpan expir catch (Exception ex) { var contendedStr = contended.ToString(); - _telemetry.TrackException(ex, [new(PropOperation, OperationAcquire), new(PropKey, key), new(PropContended, contendedStr)]); - _telemetry.TrackEvent(EventUnavailable, [new(PropKey, key), new(PropContended, contendedStr)]); + _telemetry.TryTrackException(ex, [new(PropOperation, OperationAcquire), new(PropKey, key), new(PropContended, contendedStr)]); + _telemetry.TryTrackEvent(EventUnavailable, [new(PropKey, key), new(PropContended, contendedStr)]); return NoOpAsyncDisposable.Instance; } @@ -108,8 +108,8 @@ public async ValueTask AcquireAsync(string key, TimeSpan expir } catch (Exception ex) { - _telemetry.TrackException(ex, [new(PropOperation, OperationAcquire), new(PropKey, key), new(PropContended, bool.FalseString)]); - _telemetry.TrackEvent(EventUnavailable, [new(PropKey, key), new(PropContended, bool.FalseString)]); + _telemetry.TryTrackException(ex, [new(PropOperation, OperationAcquire), new(PropKey, key), new(PropContended, bool.FalseString)]); + _telemetry.TryTrackEvent(EventUnavailable, [new(PropKey, key), new(PropContended, bool.FalseString)]); return null; } @@ -149,13 +149,13 @@ private string BuildLockToken() => private Releaser BuildAcquiredLease(RedisKey redisKey, RedisValue lockToken, string key, bool contended) { - _telemetry.TrackEvent(EventAcquired, [new(PropKey, key), new(PropContended, contended.ToString())]); + _telemetry.TryTrackEvent(EventAcquired, [new(PropKey, key), new(PropContended, contended.ToString())]); return new Releaser(_redis, _telemetry, redisKey, lockToken); } private NoOpAsyncDisposable TrackTimeoutNoOp(string key) { - _telemetry.TrackEvent(EventTimeout, [new(PropKey, key), new(PropContended, bool.TrueString)]); + _telemetry.TryTrackEvent(EventTimeout, [new(PropKey, key), new(PropContended, bool.TrueString)]); return NoOpAsyncDisposable.Instance; } @@ -176,7 +176,7 @@ public async ValueTask DisposeAsync() } catch (Exception ex) { - telemetry.TrackException(ex, [new(PropOperation, OperationRelease), new(PropKey, redisKey.ToString())]); + telemetry.TryTrackException(ex, [new(PropOperation, OperationRelease), new(PropKey, redisKey.ToString())]); } } } diff --git a/src/UiPath.Caching/MemoryCacheSetter.cs b/src/UiPath.Caching/MemoryCacheSetter.cs index 9bc64715..a520b1a6 100644 --- a/src/UiPath.Caching/MemoryCacheSetter.cs +++ b/src/UiPath.Caching/MemoryCacheSetter.cs @@ -101,7 +101,7 @@ private void RefreshMetadata(RefreshMetadataState metadataState) { if (!set) { - telemetryProvider.TrackEvent(EventRefreshMetadataFailed, + telemetryProvider.TryTrackEvent(EventRefreshMetadataFailed, [ new(PropCacheKey, metadataState.CacheKey), new(PropTopicKey, metadataState.TopicKey), diff --git a/src/UiPath.Caching/Redis/RedisConnectionWarmup.cs b/src/UiPath.Caching/Redis/RedisConnectionWarmup.cs index 71df14ca..f05d0c2a 100644 --- a/src/UiPath.Caching/Redis/RedisConnectionWarmup.cs +++ b/src/UiPath.Caching/Redis/RedisConnectionWarmup.cs @@ -41,7 +41,7 @@ private async Task WarmUpAsync(CancellationToken cancellationToken) } catch (Exception ex) when (ex is not OperationCanceledException) { - telemetryProvider.TrackException(ex); + telemetryProvider.TryTrackException(ex); } } } diff --git a/src/UiPath.Caching/Redis/RedisConnector.cs b/src/UiPath.Caching/Redis/RedisConnector.cs index bc9bef9e..d2abbfe2 100644 --- a/src/UiPath.Caching/Redis/RedisConnector.cs +++ b/src/UiPath.Caching/Redis/RedisConnector.cs @@ -645,7 +645,7 @@ private void RecordConfirmedMembers(List confirmed, long now) if (firstTime.Count > 0) { - _telemetryProvider.TrackEvent( + _telemetryProvider.TryTrackEvent( "Redis.StaleEndpointStillAMember", [ new("EndPoints", string.Join(";", firstTime.Select(FormatEndPoint))), @@ -670,7 +670,7 @@ private void DisableStaleEndpointScan(Lazy> judged) } _staleEndpointTimer?.Dispose(); - _telemetryProvider.TrackEvent("Redis.StaleEndpointScanDisabled", [new("Reason", "NoClusterConfiguration")]); + _telemetryProvider.TryTrackEvent("Redis.StaleEndpointScanDisabled", [new("Reason", "NoClusterConfiguration")]); } private void TryDisposeMultiplexer(IConnectionMultiplexer multiplexer) diff --git a/src/UiPath.Caching/RehydrationCoordinator.cs b/src/UiPath.Caching/RehydrationCoordinator.cs index e5aaaffc..296f6e1a 100644 --- a/src/UiPath.Caching/RehydrationCoordinator.cs +++ b/src/UiPath.Caching/RehydrationCoordinator.cs @@ -163,32 +163,32 @@ private async Task SpawnAsync( (keys, handles) = await AcquirePerKeyLocksAsync(reservedKeys, lockExpiry, factoryTimeout, entryType).ConfigureAwait(false); if (keys.Length == 0) { - telemetry.TrackEvent(EventDeduped, Tags(KeyValuePair.Create(TagReason, ReasonNotAcquired))); + telemetry.TryTrackEvent(EventDeduped, Tags(KeyValuePair.Create(TagReason, ReasonNotAcquired))); return; } groupKey = CompositeCacheKey.For(keys); - telemetry.TrackEvent(EventTriggered, Tags()); + telemetry.TryTrackEvent(EventTriggered, Tags()); using var cts = new CancellationTokenSource(factoryTimeout); try { await rehydrateAsync(keys, cts.Token).ConfigureAwait(false); ClearFailureCounts(keys); - telemetry.TrackEvent(EventSucceeded, Tags()); + telemetry.TryTrackEvent(EventSucceeded, Tags()); await ReleaseLocksAsync(handles, groupKey, entryType).ConfigureAwait(false); handles = null; } catch (OperationCanceledException) when (cts.IsCancellationRequested) { IncrementFailureCounts(keys); - telemetry.TrackEvent(EventTimedOut, Tags()); + telemetry.TryTrackEvent(EventTimedOut, Tags()); handles = null; } catch (Exception ex) { IncrementFailureCounts(keys); - telemetry.TrackEvent(EventFailed, Tags(KeyValuePair.Create(TagExceptionType, ex.GetType().Name))); + telemetry.TryTrackEvent(EventFailed, Tags(KeyValuePair.Create(TagExceptionType, ex.GetType().Name))); handles = null; } } diff --git a/src/UiPath.Caching/TelemetrySafeguards.cs b/src/UiPath.Caching/TelemetrySafeguards.cs index eb0cee46..9f922086 100644 --- a/src/UiPath.Caching/TelemetrySafeguards.cs +++ b/src/UiPath.Caching/TelemetrySafeguards.cs @@ -44,11 +44,11 @@ public static bool TryTrackEvent(this ICachingTelemetryProvider telemetryProvide } /// Reports a caught failure; a sink that refuses it leaves nowhere else to put it. - public static void TryTrackException(this ICachingTelemetryProvider telemetryProvider, Exception ex) + public static void TryTrackException(this ICachingTelemetryProvider telemetryProvider, Exception ex, ReadOnlySpan> properties = default) { try { - telemetryProvider.TrackException(ex); + telemetryProvider.TrackException(ex, properties); } catch (Exception) { diff --git a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs index 1a1c9bb3..d291480c 100644 --- a/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs +++ b/tests/UiPath.Caching.Tests/Broadcast/RedisStreamSubjectWriterTests.cs @@ -5,6 +5,7 @@ using NSubstitute.ReceivedExtensions; using StackExchange.Redis; using UiPath.Caching.Telemetry; +using UiPath.Caching.Tests.Telemetry; namespace UiPath.Caching.Tests.Broadcast; @@ -407,6 +408,76 @@ public async Task SameSource_event_is_acknowledged_without_writing_to_channel() channel.Reader.TryRead(out _).Should().BeFalse("events from the current source must not enter the dispatcher channel"); } + [Fact] + public async Task Invalid_event_is_acknowledged_when_the_telemetry_sink_refuses_the_record() + { + var channel = Channel.CreateBounded(new BoundedChannelOptions(10)); + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => { acked.TrySetResult(call.Arg()); return Task.FromResult(1L); }); + var id = _fixture.Create(); + var entries = new[] { new StreamEntry(id, [new NameValueEntry(_fieldName, _fixture.Create())]) }; + _formatter.Decode(Arg.Any>()).Returns(new TestCacheEvent { Valid = false }); + SetupSingleBatch(entries); + var telemetry = new RefusingTelemetryProvider("Caching.RedisStreamSubjectWriter.DispatchEventsAsync.InvalidEvent"); + + using var sut = CreateSut(channel.Writer, _logger, telemetry); + + var ids = await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + ids.Select(v => v.ToString()).Should().Contain(id, "the poison entry must still be acknowledged"); + telemetry.Exceptions.Should().Contain(RefusingTelemetryProvider.Failure, "the refusal is reported rather than swallowed"); + } + + [Fact] + public async Task SameSource_event_is_acknowledged_when_the_telemetry_sink_refuses_the_receipt() + { + var channel = Channel.CreateBounded(new BoundedChannelOptions(10)); + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => { acked.TrySetResult(call.Arg()); return Task.FromResult(1L); }); + var id = _fixture.Create(); + var entries = new[] { new StreamEntry(id, [new NameValueEntry(_fieldName, _fixture.Create())]) }; + _formatter.Decode(Arg.Any>()).Returns(new TestCacheEvent { Valid = true, Source = _sourceUri }); + SetupSingleBatch(entries); + // Metrics too: TrackTopicReadMetric also runs on this path. + var telemetry = new RefusingTelemetryProvider("Caching.RedisStreamSubjectWriter.DispatchEventsAsync.EventReceived", refuseMetrics: true); + + using var sut = CreateSut(channel.Writer, _logger, telemetry); + + var ids = await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + ids.Select(v => v.ToString()).Should().Contain(id, "the entry must still be acknowledged"); + } + + [Fact] + public async Task Valid_event_is_dispatched_and_acknowledged_when_the_telemetry_sink_refuses_the_receipt() + { + var channel = Channel.CreateBounded(new BoundedChannelOptions(10)); + var acked = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _database.StreamAcknowledgeAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(call => { acked.TrySetResult(call.Arg()); return Task.FromResult(1L); }); + var id = _fixture.Create(); + var entries = new[] { new StreamEntry(id, [new NameValueEntry(_fieldName, _fixture.Create())]) }; + _formatter.Decode(Arg.Any>()).Returns(new TestCacheEvent { Valid = true, Source = new Uri("urn:other-source") }); + SetupSingleBatch(entries); + var telemetry = new RefusingTelemetryProvider("Caching.RedisStreamSubjectWriter.DispatchEventsAsync.EventReceived"); + + using var sut = CreateSut(channel.Writer, _logger, telemetry); + + var ids = await acked.Task.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + _cancellationTokenSource.Cancel(); + await sut.FetchTask.WaitAsync(WaitTimeout, TestContext.Current.CancellationToken); + + channel.Reader.TryRead(out _).Should().BeTrue("the event must still reach the dispatcher"); + ids.Select(v => v.ToString()).Should().Contain(id); + telemetry.Exceptions.Should().Contain(RefusingTelemetryProvider.Failure); + } + [Fact] public async Task Valid_event_is_written_to_channel_and_acknowledged() { @@ -564,7 +635,7 @@ private void SetupSingleBatch(StreamEntry[] entries) .ReturnsForAnyArgs(_ => ++emitted == 1 ? entries : []); } - private RedisStreamSubjectWriter CreateSut(ChannelWriter writer, ILogger logger) + private RedisStreamSubjectWriter CreateSut(ChannelWriter writer, ILogger logger, ICachingTelemetryProvider? telemetry = null) { var connectionState = _fixture.Create(); connectionState.IsConnected.Returns(true); @@ -577,7 +648,7 @@ private RedisStreamSubjectWriter CreateSut(ChannelWriter(), + telemetry ?? _fixture.Create(), _fixture.Create(), new TimedFetchWaiter(_pollInterval), _cancellationTokenSource.Token); diff --git a/tests/UiPath.Caching.Tests/FactoryTimeoutTests.cs b/tests/UiPath.Caching.Tests/FactoryTimeoutTests.cs new file mode 100644 index 00000000..29e43b23 --- /dev/null +++ b/tests/UiPath.Caching.Tests/FactoryTimeoutTests.cs @@ -0,0 +1,44 @@ +using UiPath.Caching.Tests.Telemetry; + +namespace UiPath.Caching.Tests; + +public class FactoryTimeoutTests(ITestContextAccessor testContextAccessor) +{ + private const string TimedOutEvent = "cache.factory.timed_out"; + + [Fact] + public async Task RunAsync_still_throws_TimeoutException_when_the_telemetry_sink_refuses_the_record() + { + var telemetry = new RefusingTelemetryProvider(TimedOutEvent); + + Func act = () => RunTimingOutFactoryAsync(telemetry); + + await act.Should().ThrowAsync("a refused record must not replace the exception the caller is documented to get"); + telemetry.Exceptions.Should().Contain(RefusingTelemetryProvider.Failure, "the refusal is reported rather than swallowed"); + } + + [Fact] + public async Task RunAsync_records_the_timeout_when_the_telemetry_sink_accepts_it() + { + var telemetry = new RefusingTelemetryProvider("some.other.event"); + + Func act = () => RunTimingOutFactoryAsync(telemetry); + + await act.Should().ThrowAsync(); + telemetry.Events.Should().Contain(TimedOutEvent); + telemetry.Exceptions.Should().BeEmpty(); + } + + private Task RunTimingOutFactoryAsync(RefusingTelemetryProvider telemetry) => + FactoryTimeout.RunAsync( + async ct => + { + await Task.Delay(Timeout.InfiniteTimeSpan, ct); + return "unreachable"; + }, + TimeSpan.FromMilliseconds(20), + new CacheKey("k"), + "cache", + telemetry, + testContextAccessor.Current.CancellationToken); +} diff --git a/tests/UiPath.Caching.Tests/Locking/RedisDistributedLockTests.cs b/tests/UiPath.Caching.Tests/Locking/RedisDistributedLockTests.cs index 8e6eef5e..88dc70dd 100644 --- a/tests/UiPath.Caching.Tests/Locking/RedisDistributedLockTests.cs +++ b/tests/UiPath.Caching.Tests/Locking/RedisDistributedLockTests.cs @@ -374,6 +374,97 @@ public void Ctor_throws_when_DistributedLockMaxPollInterval_is_less_than_Distrib .Which.ParamName.Should().Be("cacheOptions.DistributedLockMaxPollInterval"); } + [Fact] + public async Task Acquire_returns_the_lease_when_the_telemetry_sink_refuses_the_acquired_event() + { + var redis = _fixture.Freeze(); + redis.Database + .LockTakeAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + + var telemetry = new RefusingTelemetryProvider("cache.distributedlock.acquired"); + var sut = NewLock(redis, telemetry: telemetry); + var token = testContextAccessor.Current.CancellationToken; + + var lease = await sut.AcquireAsync("k", TimeSpan.FromSeconds(5), TimeSpan.Zero, token); + + lease.Should().NotBeSameAs(NoOpAsyncDisposable.Instance, "Redis granted the lock, so the caller must get the releaser"); + telemetry.Exceptions.Should().Contain(RefusingTelemetryProvider.Failure); + + await lease!.DisposeAsync(); + await redis.Database.Received().LockReleaseAsync(Arg.Any(), Arg.Any(), Arg.Any()); + } + + [Fact] + public async Task Acquire_returns_the_no_op_lease_when_the_telemetry_sink_refuses_the_timeout_event() + { + var redis = _fixture.Freeze(); + redis.Database + .LockTakeAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(false); + + var telemetry = new RefusingTelemetryProvider("cache.distributedlock.timeout"); + var sut = NewLock(redis, new CacheOptions { DistributedLockPollInterval = TimeSpan.FromMilliseconds(1) }, telemetry); + var token = testContextAccessor.Current.CancellationToken; + + var lease = await sut.AcquireAsync("k", TimeSpan.FromSeconds(5), TimeSpan.FromMilliseconds(5), token); + + lease.Should().BeSameAs(NoOpAsyncDisposable.Instance, "waiting out the deadline degrades to the no-op lease"); + telemetry.Exceptions.Should().Contain(RefusingTelemetryProvider.Failure); + } + + [Fact] + public async Task Acquire_returns_the_no_op_lease_when_the_telemetry_sink_refuses_the_failure_reports() + { + var redis = _fixture.Freeze(); + redis.Database + .LockTakeAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns>(_ => throw new RedisException("down")); + + var telemetry = new RefusingTelemetryProvider("cache.distributedlock.unavailable", refuseExceptions: true); + var sut = NewLock(redis, telemetry: telemetry); + + var lease = await sut.AcquireAsync("k", TimeSpan.FromSeconds(5), TimeSpan.Zero, testContextAccessor.Current.CancellationToken); + + lease.Should().BeSameAs(NoOpAsyncDisposable.Instance); + } + + [Fact] + public async Task TryAcquire_returns_null_when_the_telemetry_sink_refuses_the_failure_reports() + { + var redis = _fixture.Freeze(); + redis.Database + .LockTakeAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns>(_ => throw new RedisException("down")); + + var telemetry = new RefusingTelemetryProvider("cache.distributedlock.unavailable", refuseExceptions: true); + var sut = NewLock(redis, telemetry: telemetry); + + var lease = await sut.TryAcquireAsync("k", TimeSpan.FromSeconds(5), testContextAccessor.Current.CancellationToken); + + lease.Should().BeNull(); + } + + [Fact] + public async Task Release_does_not_throw_when_the_telemetry_sink_refuses_the_failure_report() + { + var redis = _fixture.Freeze(); + redis.Database + .LockTakeAsync(Arg.Any(), Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(true); + redis.Database + .LockReleaseAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns>(_ => throw new RedisException("down")); + + var telemetry = new RefusingTelemetryProvider("none", refuseExceptions: true); + var sut = NewLock(redis, telemetry: telemetry); + var lease = await sut.AcquireAsync("k", TimeSpan.FromSeconds(5), TimeSpan.Zero, testContextAccessor.Current.CancellationToken); + + var release = async () => await lease.DisposeAsync(); + + await release.Should().NotThrowAsync(); + } + private RedisDistributedLock NewLock(IRedisConnector? redis = null, CacheOptions? options = null, ICachingTelemetryProvider? telemetry = null) { redis ??= _fixture.Freeze(); diff --git a/tests/UiPath.Caching.Tests/MemoryCacheSetterTests.cs b/tests/UiPath.Caching.Tests/MemoryCacheSetterTests.cs index 6fad8319..94817b34 100644 --- a/tests/UiPath.Caching.Tests/MemoryCacheSetterTests.cs +++ b/tests/UiPath.Caching.Tests/MemoryCacheSetterTests.cs @@ -147,6 +147,44 @@ public void RefreshMetadata_swallows_NewEntry_exception_and_emits_failure_event( e.Properties["CacheKey"] == _cacheKey); } + [Fact] + public void RefreshMetadata_callback_does_not_throw_when_the_telemetry_sink_refuses_the_failure_event() + { + var telemetry = new RefusingTelemetryProvider($"Caching.{nameof(MemoryCacheSetter)}.{nameof(MemoryCacheSetter.RefreshMetadata)}.Failed"); + _fixture.Inject(telemetry); + _memoryCache = new MemoryCache(Options.Create(new MemoryCacheOptions + { + Clock = _clock, + })); + _fixture.Inject(_memoryCache); + + var token = new TestChangeToken + { + ActiveChangeCallbacks = true, + HasChanged = false, + Expiration = _clock.UtcNow.AddDays(1), + }; + _changeTokenFactory.Create(Arg.Any(), Arg.Any>(), Arg.Any(), Arg.Any()) + .Returns(token); + + var cacheEntity = _fixture.Create(); + cacheEntity.NewEntry(Arg.Any(), Arg.Any?>()) + .Returns(_ => throw new InvalidOperationException("simulated NewEntry failure")); + + var x = new InternalHashCacheEntryOptions() + { + CacheKey = _cacheKey, + TopicKey = _topicKey, + Expiration = _clock.UtcNow.AddDays(1), + }; + + Sut.Set(x, cacheEntity, _fixture.Create(), TimeSpan.FromMinutes(1)); + Action act = () => token.InvokeCallbacks(); + + act.Should().NotThrow(); + telemetry.Exceptions.Should().Contain(RefusingTelemetryProvider.Failure); + } + [Fact] public void Setter_max_duration() { diff --git a/tests/UiPath.Caching.Tests/Redis/RedisConnectorStaleEndpointTests.cs b/tests/UiPath.Caching.Tests/Redis/RedisConnectorStaleEndpointTests.cs index c5b73777..8637b321 100644 --- a/tests/UiPath.Caching.Tests/Redis/RedisConnectorStaleEndpointTests.cs +++ b/tests/UiPath.Caching.Tests/Redis/RedisConnectorStaleEndpointTests.cs @@ -127,6 +127,21 @@ public async Task Scan_IgnoresConfiguredEndpoints() h.Connector.Dispose(); } + [Fact] + public async Task A_refused_member_report_does_not_back_off_the_next_scan() + { + // Thrown into the scan's catch, a refusal would count as a failed scan and skip the next one. + var h = new Harness(retiredIsMember: true); + h.Telemetry.Refuse = "Redis.StaleEndpointStillAMember"; + await h.ScanTwiceAcrossThresholdAsync(); + + h.Clock.Advance(TimeSpan.FromMinutes(5)); + await h.Connector.ScanStaleEndpointsAsync(); + + await h.Multiplexer.Received(2).ConfigureAsync(Arg.Any()); + h.Connector.Dispose(); + } + [Fact] public async Task Scan_DisablesItself_OnlyAfterANodeKeepsReportingNoConfiguration() { @@ -512,8 +527,17 @@ private sealed class RecordingTelemetry : ICachingTelemetryProvider { public List Events { get; } = []; public List Exceptions { get; } = []; + public string? Refuse { get; set; } public void TrackException(Exception ex, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) => Exceptions.Add(ex); - public void TrackEvent(string eventName, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) => Events.Add(eventName); + public void TrackEvent(string eventName, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) + { + if (eventName == Refuse) + { + throw new InvalidOperationException("telemetry sink refused"); + } + + Events.Add(eventName); + } } private sealed class FakeTopology : IClusterTopologyReader diff --git a/tests/UiPath.Caching.Tests/RehydrationCoordinatorTests.cs b/tests/UiPath.Caching.Tests/RehydrationCoordinatorTests.cs index 67c28e83..81c7dd88 100644 --- a/tests/UiPath.Caching.Tests/RehydrationCoordinatorTests.cs +++ b/tests/UiPath.Caching.Tests/RehydrationCoordinatorTests.cs @@ -1,6 +1,8 @@ using System.Collections.Concurrent; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using UiPath.Caching.Locking; +using UiPath.Caching.Telemetry; using UiPath.Caching.Tests.Telemetry; namespace UiPath.Caching.Tests; @@ -338,9 +340,101 @@ public async Task Batch_cooldown_uses_the_max_failure_count_across_the_set() Assert.Fail("\"failing\" never left the in-flight set, so the max-failure-count path was never exercised."); } + [Fact] + public async Task SpawnAsync_still_rehydrates_when_the_telemetry_sink_refuses_the_triggered_event() + { + var distributedLock = Substitute.For(); + distributedLock.TryAcquireAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(Substitute.For()); + + var telemetry = new RefusingTelemetryProvider("cache.rehydrate.triggered"); + var sut = NewCoordinator(distributedLock, telemetry); + var rehydrated = new TaskCompletionSource(); + + var triggered = sut.TryTrigger( + (CacheKey)"k", + DateTimeOffset.UtcNow.Add(TimeSpan.FromSeconds(1)), + RehydratePolicy(), + Duration, + "cache", + _ => + { + rehydrated.TrySetResult(); + return ValueTask.CompletedTask; + }); + + triggered.Should().BeTrue(); + await rehydrated.Task.WaitAsync(TimeSpan.FromSeconds(30), TestContext.Current.CancellationToken); + telemetry.Exceptions.Should().Contain(RefusingTelemetryProvider.Failure, "the refusal is reported rather than swallowed"); + } + + [Theory] + [InlineData("cache.rehydrate.timed_out")] + [InlineData("cache.rehydrate.failed")] + public async Task SpawnAsync_keeps_the_locks_for_the_cooldown_when_the_telemetry_sink_refuses_the_outcome_event(string refusedEvent) + { + // The draining probe gets its own handle, or it would dispose the one under test. + var handle = Substitute.For(); + var distributedLock = Substitute.For(); + distributedLock.TryAcquireAsync(Arg.Any(), Arg.Any(), Arg.Any()) + .Returns(handle, Substitute.For()); + + var telemetry = new RefusingTelemetryProvider(refusedEvent); + var logger = new CapturingLogger(); + var sut = NewCoordinator(distributedLock, telemetry, logger); + var policy = RehydratePolicy(timeoutFraction: 0.001, baseCooldown: TimeSpan.FromMilliseconds(50)); + + var triggered = sut.TryTrigger( + (CacheKey)"k", + DateTimeOffset.UtcNow.Add(TimeSpan.FromSeconds(1)), + policy, + Duration, + "cache", + refusedEvent == "cache.rehydrate.failed" + ? _ => ValueTask.FromException(new InvalidOperationException("factory failed")) + : async ct => await Task.Delay(TimeSpan.FromSeconds(30), ct)); + + triggered.Should().BeTrue(); + await WaitForRefusalAsync(telemetry); + + // Same key, since _inFlight is keyed by name. An escaped refusal is logged before the finally clears the + // reservation, so once it clears the outer catch has already run. + await WaitForReservationReleaseAsync(sut, policy); + + logger.Errors.Should().BeEmpty("a refused report must not reach the outer catch, which is what lets the finally release the locks"); + await handle.DidNotReceive().DisposeAsync(); + } + + private static async Task WaitForReservationReleaseAsync(RehydrationCoordinator sut, CachePolicy policy) + { + for (var i = 0; i < 3000; i++) + { + if (!sut.TryTrigger((CacheKey)"k", DateTimeOffset.UtcNow.Add(TimeSpan.FromSeconds(1)), policy, Duration, "cache", _ => ValueTask.CompletedTask)) + { + await Task.Delay(10, TestContext.Current.CancellationToken); + continue; + } + + return; + } + + throw new InvalidOperationException("the first spawn never released its in-flight reservation"); + } + + private static async Task WaitForRefusalAsync(RefusingTelemetryProvider telemetry) + { + for (var i = 0; i < 3000 && telemetry.Exceptions.Count == 0; i++) + { + await Task.Delay(10, TestContext.Current.CancellationToken); + } + + telemetry.Exceptions.Should().Contain(RefusingTelemetryProvider.Failure); + } + private static RehydrationCoordinator NewCoordinator( IDistributedLock? distributedLock = null, - RecordingTelemetryProvider? telemetry = null) + ICachingTelemetryProvider? telemetry = null, + ILogger? logger = null) { var clock = TimeProvider.System; var lockKeyStrategy = new DefaultDistributedLockKeyStrategy(separator: ':'); @@ -350,7 +444,7 @@ private static RehydrationCoordinator NewCoordinator( distributedLock ?? NullDistributedLock.Instance, lockKeyStrategy, telemetry ?? new RecordingTelemetryProvider(), - NullLogger.Instance); + logger ?? NullLogger.Instance); } private static CachePolicy RehydratePolicy( @@ -397,4 +491,24 @@ private static async Task WaitForEvent(RecordingTelemetryProvider telemetry, str } throw new TimeoutException($"Event '{eventName}' was not emitted within {timeout}."); } + + private sealed class CapturingLogger : ILogger + { + private readonly ConcurrentQueue _errors = new(); + + public IReadOnlyCollection Errors => _errors; + + public IDisposable? BeginScope(TState state) + where TState : notnull => null; + + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + if (logLevel >= LogLevel.Error) + { + _errors.Enqueue(formatter(state, exception)); + } + } + } } diff --git a/tests/UiPath.Caching.Tests/Telemetry/RefusingTelemetryProvider.cs b/tests/UiPath.Caching.Tests/Telemetry/RefusingTelemetryProvider.cs new file mode 100644 index 00000000..75f90eae --- /dev/null +++ b/tests/UiPath.Caching.Tests/Telemetry/RefusingTelemetryProvider.cs @@ -0,0 +1,65 @@ +using UiPath.Caching.Telemetry; + +namespace UiPath.Caching.Tests.Telemetry; + +/// Refuses one event name, and optionally every metric or exception. +internal sealed class RefusingTelemetryProvider(string failingEvent, bool refuseMetrics = false, bool refuseExceptions = false) : ICachingTelemetryProvider +{ + public static readonly InvalidOperationException Failure = new("telemetry sink refused"); + + private readonly object _gate = new(); + private readonly List _events = []; + private readonly List _exceptions = []; + + public IReadOnlyList Events => Snapshot(_events); + + public IReadOnlyList Exceptions => Snapshot(_exceptions); + + public void TrackDependency(string type, string target, string name, string data, DateTimeOffset startTime, TimeSpan duration, string resultCode, bool success, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) + { + } + + public void TrackEvent(string eventName, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) + { + if (eventName == failingEvent) + { + throw Failure; + } + + Record(_events, eventName); + } + + public void TrackException(Exception ex, ReadOnlySpan> properties = default, ReadOnlySpan> metrics = default) + { + if (refuseExceptions) + { + throw Failure; + } + + Record(_exceptions, ex); + } + + public void TrackMetric(string name, double value, ReadOnlySpan> properties = default) + { + if (refuseMetrics) + { + throw Failure; + } + } + + private void Record(List target, T record) + { + lock (_gate) + { + target.Add(record); + } + } + + private T[] Snapshot(List source) + { + lock (_gate) + { + return source.ToArray(); + } + } +}