fix(cache): a refused telemetry report no longer defeats the catch that made it

src/UiPath.Caching/Broadcast/Redis/RedisStreamSubjectWriter.cs

@@ -294,10 +294,11 @@ private ValueTask ProcessEvent(StreamEntry @event, List<RedisValue> 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<R
 
     private void HandleInvalidEvent(T ev, StreamEntry @event, List<RedisValue> 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()),

src/UiPath.Caching/FactoryTimeout.cs

@@ -37,7 +37,7 @@ public static async Task<T> RunAsync<T>(
         }
         catch (OperationCanceledException) when (linkedCts.IsCancellationRequested && !token.IsCancellationRequested)
         {
-            telemetry.TrackEvent(EventName,
+            telemetry.TryTrackEvent(EventName,
             [
                 new(TagCacheName, cacheName),
                 new(TagCacheKey, cacheKey.Name),

src/UiPath.Caching/Locking/RedisDistributedLock.cs

@@ -72,8 +72,8 @@ public async ValueTask<IAsyncDisposable> 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<IAsyncDisposable> 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())]);
             }
         }
     }

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),

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

src/UiPath.Caching/Redis/RedisConnector.cs

@@ -645,7 +645,7 @@ private void RecordConfirmedMembers(List<EndPoint> 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<Task<IConnectionMultiplexer>> judged)
         }
 
         _staleEndpointTimer?.Dispose();
-        _telemetryProvider.TrackEvent("Redis.StaleEndpointScanDisabled", [new("Reason", "NoClusterConfiguration")]);
+        _telemetryProvider.TryTrackEvent("Redis.StaleEndpointScanDisabled", [new("Reason", "NoClusterConfiguration")]);
     }
 
     private void TryDisposeMultiplexer(IConnectionMultiplexer multiplexer)

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;
             }
         }

src/UiPath.Caching/TelemetrySafeguards.cs

@@ -44,11 +44,11 @@ public static bool TryTrackEvent(this ICachingTelemetryProvider telemetryProvide
     }
 
     /// <summary>Reports a caught failure; a sink that refuses it leaves nowhere else to put it.</summary>
-    public static void TryTrackException(this ICachingTelemetryProvider telemetryProvider, Exception ex)
+    public static void TryTrackException(this ICachingTelemetryProvider telemetryProvider, Exception ex, ReadOnlySpan<KeyValuePair<string, string>> properties = default)
     {
         try
         {
-            telemetryProvider.TrackException(ex);
+            telemetryProvider.TrackException(ex, properties);
         }
         catch (Exception)
         {

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<ICacheEvent>(new BoundedChannelOptions(10));
+        var acked = new TaskCompletionSource<RedisValue[]>(TaskCreationOptions.RunContinuationsAsynchronously);
+        _database.StreamAcknowledgeAsync(Arg.Any<RedisKey>(), Arg.Any<RedisValue>(), Arg.Any<RedisValue[]>())
+            .Returns(call => { acked.TrySetResult(call.Arg<RedisValue[]>()); return Task.FromResult(1L); });
+        var id = _fixture.Create<string>();
+        var entries = new[] { new StreamEntry(id, [new NameValueEntry(_fieldName, _fixture.Create<string>())]) };
+        _formatter.Decode(Arg.Any<ReadOnlyMemory<byte>>()).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<ICacheEvent>(new BoundedChannelOptions(10));
+        var acked = new TaskCompletionSource<RedisValue[]>(TaskCreationOptions.RunContinuationsAsynchronously);
+        _database.StreamAcknowledgeAsync(Arg.Any<RedisKey>(), Arg.Any<RedisValue>(), Arg.Any<RedisValue[]>())
+            .Returns(call => { acked.TrySetResult(call.Arg<RedisValue[]>()); return Task.FromResult(1L); });
+        var id = _fixture.Create<string>();
+        var entries = new[] { new StreamEntry(id, [new NameValueEntry(_fieldName, _fixture.Create<string>())]) };
+        _formatter.Decode(Arg.Any<ReadOnlyMemory<byte>>()).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<ICacheEvent>(new BoundedChannelOptions(10));
+        var acked = new TaskCompletionSource<RedisValue[]>(TaskCreationOptions.RunContinuationsAsynchronously);
+        _database.StreamAcknowledgeAsync(Arg.Any<RedisKey>(), Arg.Any<RedisValue>(), Arg.Any<RedisValue[]>())
+            .Returns(call => { acked.TrySetResult(call.Arg<RedisValue[]>()); return Task.FromResult(1L); });
+        var id = _fixture.Create<string>();
+        var entries = new[] { new StreamEntry(id, [new NameValueEntry(_fieldName, _fixture.Create<string>())]) };
+        _formatter.Decode(Arg.Any<ReadOnlyMemory<byte>>()).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<ICacheEvent> CreateSut(ChannelWriter<ICacheEvent> writer, ILogger logger)
+    private RedisStreamSubjectWriter<ICacheEvent> CreateSut(ChannelWriter<ICacheEvent> writer, ILogger logger, ICachingTelemetryProvider? telemetry = null)
     {
         var connectionState = _fixture.Create<IConnectionState>();
         connectionState.IsConnected.Returns(true);
@@ -577,7 +648,7 @@ private RedisStreamSubjectWriter<ICacheEvent> CreateSut(ChannelWriter<ICacheEven
             writer,
             _formatter,
             logger,
-            _fixture.Create<ICachingTelemetryProvider>(),
+            telemetry ?? _fixture.Create<ICachingTelemetryProvider>(),
             _fixture.Create<IRedisProfiler>(),
             new TimedFetchWaiter(_pollInterval),
             _cancellationTokenSource.Token);

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<Task> act = () => RunTimingOutFactoryAsync(telemetry);
+
+        await act.Should().ThrowAsync<TimeoutException>("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<Task> act = () => RunTimingOutFactoryAsync(telemetry);
+
+        await act.Should().ThrowAsync<TimeoutException>();
+        telemetry.Events.Should().Contain(TimedOutEvent);
+        telemetry.Exceptions.Should().BeEmpty();
+    }
+
+    private Task<string> RunTimingOutFactoryAsync(RefusingTelemetryProvider telemetry) =>
+        FactoryTimeout.RunAsync<string>(
+            async ct =>
+            {
+                await Task.Delay(Timeout.InfiniteTimeSpan, ct);
+                return "unreachable";
+            },
+            TimeSpan.FromMilliseconds(20),
+            new CacheKey("k"),
+            "cache",
+            telemetry,
+            testContextAccessor.Current.CancellationToken);
+}