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(); + } + } +}