Envelopes that fail to send are now flushed when the transport recovers (#3438)

jamescrosswell · web-flow · commit d5d4b663b4c6 · 2024-07-11T14:32:35.000+12:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -2,6 +2,10 @@
 
 ## Unreleased
 
+### Fixes
+
+- Fixed envelopes getting stuck in processing when losing network connectivity ([#3438](https://github.com/getsentry/sentry-dotnet/pull/3438))
+
 ### Features
 
 - Client reports now include dropped spans ([#3463](https://github.com/getsentry/sentry-dotnet/pull/3463))
diff --git a/src/Sentry/Internal/Http/CachingTransport.cs b/src/Sentry/Internal/Http/CachingTransport.cs
@@ -17,6 +17,7 @@ namespace Sentry.Internal.Http;
 internal class CachingTransport : ITransport, IDisposable
 {
     private const string EnvelopeFileExt = "envelope";
+    private const string ProcessingFolder = "__processing";
 
     private readonly ITransport _innerTransport;
     private readonly SentryOptions _options;
@@ -77,7 +78,7 @@ private CachingTransport(ITransport innerTransport, SentryOptions options, bool
             options.TryGetProcessSpecificCacheDirectoryPath() ??
             throw new InvalidOperationException("Cache directory or DSN is not set.");
 
-        _processingDirectoryPath = Path.Combine(_isolatedCacheDirectoryPath, "__processing");
+        _processingDirectoryPath = Path.Combine(_isolatedCacheDirectoryPath, ProcessingFolder);
     }
 
     private void Initialize(bool startWorker)
@@ -271,6 +272,13 @@ private async Task ProcessCacheAsync(CancellationToken cancellation)
             // Signal that we can start waiting for _options.InitCacheFlushTimeout
             _preInitCacheResetEvent?.Set();
 
+            // Make sure no files got stuck in the processing directory
+            // See https://github.com/getsentry/sentry-dotnet/pull/3438#discussion_r1672524426
+            if (_options.NetworkStatusListener is { Online: false } listener)
+            {
+                MoveUnprocessedFilesBackToCache();
+            }
+
             // Process the cache
             _options.LogDebug("Flushing cached envelopes.");
             while (await TryPrepareNextCacheFileAsync(cancellation).ConfigureAwait(false) is { } file)
@@ -288,6 +296,13 @@ private async Task ProcessCacheAsync(CancellationToken cancellation)
         }
     }
 
+    private static bool IsNetworkError(Exception exception) =>
+        exception switch
+        {
+            HttpRequestException or WebException or IOException or SocketException => true,
+            _ => false
+        };
+
     private async Task InnerProcessCacheAsync(string file, CancellationToken cancellation)
     {
         if (_options.NetworkStatusListener is { Online: false } listener)
@@ -327,9 +342,12 @@ private async Task InnerProcessCacheAsync(string file, CancellationToken cancell
                         // Let the worker catch, log, wait a bit and retry.
                         throw;
                     }
-                    catch (Exception ex) when (ex is HttpRequestException or WebException or SocketException
-                                                   or IOException)
+                    catch (Exception ex) when (IsNetworkError(ex))
                     {
+                        if (_options.NetworkStatusListener is PollingNetworkStatusListener pollingListener)
+                        {
+                            pollingListener.Online = false;
+                        }
                         _options.LogError(ex, "Failed to send cached envelope: {0}, retrying after a delay.", file);
                         // Let the worker catch, log, wait a bit and retry.
                         throw;
diff --git a/src/Sentry/Internal/PollingNetworkStatusListener.cs b/src/Sentry/Internal/PollingNetworkStatusListener.cs
@@ -0,0 +1,70 @@
+using Sentry.Extensibility;
+
+namespace Sentry.Internal;
+
+internal class PollingNetworkStatusListener : INetworkStatusListener
+{
+    private readonly SentryOptions? _options;
+    private readonly IPing? _testPing;
+    internal int _delayInMilliseconds;
+    private readonly int _maxDelayInMilliseconds;
+    private readonly Func<int, int> _backoffFunction;
+
+    public PollingNetworkStatusListener(SentryOptions options, int initialDelayInMilliseconds = 500,
+        int maxDelayInMilliseconds = 32_000, Func<int, int>? backoffFunction = null)
+    {
+        _options = options;
+        _delayInMilliseconds = initialDelayInMilliseconds;
+        _maxDelayInMilliseconds = maxDelayInMilliseconds;
+        _backoffFunction = backoffFunction ?? (x => x * 2);
+    }
+
+    /// <summary>
+    /// Overload for testing
+    /// </summary>
+    internal PollingNetworkStatusListener(IPing testPing, int initialDelayInMilliseconds = 500,
+        int maxDelayInMilliseconds = 32_000, Func<int, int>? backoffFunction = null)
+    {
+        _testPing = testPing;
+        _delayInMilliseconds = initialDelayInMilliseconds;
+        _maxDelayInMilliseconds = maxDelayInMilliseconds;
+        _backoffFunction = backoffFunction ?? (x => x * 2);
+    }
+
+    private Lazy<IPing> LazyPing => new(() =>
+    {
+        if (_testPing != null)
+        {
+            return _testPing;
+        }
+        // If not running unit tests then _options will not be null and SDK init would fail without a Dsn
+        var uri = new Uri(_options!.Dsn!);
+        return new TcpPing(uri.DnsSafeHost, uri.Port);
+    });
+    private IPing Ping => LazyPing.Value;
+
+    private volatile bool _online = true;
+    public bool Online
+    {
+        get => _online;
+        set => _online = value;
+    }
+
+    public async Task WaitForNetworkOnlineAsync(CancellationToken cancellationToken = default)
+    {
+        while (!cancellationToken.IsCancellationRequested)
+        {
+            await Task.Delay(_delayInMilliseconds, cancellationToken).ConfigureAwait(false);
+            var checkResult = await Ping.IsAvailableAsync().ConfigureAwait(false);
+            if (checkResult)
+            {
+                Online = true;
+                return;
+            }
+            if (_delayInMilliseconds < _maxDelayInMilliseconds)
+            {
+                _delayInMilliseconds = _backoffFunction(_delayInMilliseconds);
+            }
+        }
+    }
+}
diff --git a/src/Sentry/Internal/TcpPing.cs b/src/Sentry/Internal/TcpPing.cs
@@ -0,0 +1,25 @@
+namespace Sentry.Internal;
+
+internal interface IPing
+{
+    Task<bool> IsAvailableAsync();
+}
+
+internal class TcpPing(string hostToCheck, int portToCheck = 443) : IPing
+{
+    private readonly Ping _ping = new();
+
+    public async Task<bool> IsAvailableAsync()
+    {
+        try
+        {
+            using var tcpClient = new TcpClient();
+            await tcpClient.ConnectAsync(hostToCheck, portToCheck).ConfigureAwait(false);
+            return true;
+        }
+        catch (Exception)
+        {
+            return false;
+        }
+    }
+}
diff --git a/src/Sentry/SentryOptions.cs b/src/Sentry/SentryOptions.cs
@@ -1297,6 +1297,10 @@ public SentryOptions()
                 UriComponents.SchemeAndServer,
                 UriFormat.Unescaped)
         );
+
+#if PLATFORM_NEUTRAL
+        NetworkStatusListener = new PollingNetworkStatusListener(this);
+#endif
     }
 
     /// <summary>
diff --git a/test/Sentry.Tests/Internals/Http/CachingTransportTests.cs b/test/Sentry.Tests/Internals/Http/CachingTransportTests.cs
@@ -545,14 +545,17 @@ await Assert.ThrowsAnyAsync<Exception>(async () =>
     [MemberData(nameof(NetworkTestData))]
     public async Task TestNetworkException(Exception exception)
     {
-        // Arrange
+        // Arrange - network unavailable
         using var cacheDirectory = new TempDirectory(_fileSystem);
+        var pingHost = Substitute.For<IPing>();
+        pingHost.IsAvailableAsync().Returns(Task.FromResult(true));
         var options = new SentryOptions
         {
             Dsn = ValidDsn,
             DiagnosticLogger = _logger,
             Debug = true,
             CacheDirectoryPath = cacheDirectory.Path,
+            NetworkStatusListener = new PollingNetworkStatusListener(pingHost),
             FileSystem = _fileSystem
         };
 
@@ -577,16 +580,25 @@ public async Task TestNetworkException(Exception exception)
         {
             receivedException = he;
         }
-        finally
-        {
-            // (transport stops failing)
-            innerTransport.ClearReceivedCalls();
-            await transport.FlushAsync();
-        }
 
         // Assert
-        Assert.Equal(exception, receivedException);
-        Assert.True(_fileSystem.EnumerateFiles(cacheDirectory.Path, "*", SearchOption.AllDirectories).Any());
+        receivedException.Should().Be(exception);
+        var files = _fileSystem.EnumerateFiles(cacheDirectory.Path, "*", SearchOption.AllDirectories).ToArray();
+        files.Should().NotBeEmpty();
+
+        // Arrange - network recovery
+        innerTransport.ClearReceivedCalls();
+        innerTransport
+            .SendEnvelopeAsync(Arg.Any<Envelope>(), Arg.Any<CancellationToken>())
+            .Returns(_ => Task.CompletedTask);
+
+        // Act
+        await transport.FlushAsync();
+
+        // Assert
+        receivedException.Should().Be(exception);
+        files = _fileSystem.EnumerateFiles(cacheDirectory.Path, "*", SearchOption.AllDirectories).ToArray();
+        files.Should().NotContain(file => file.Contains("__processing", StringComparison.OrdinalIgnoreCase));
     }
 
     [Fact]
diff --git a/test/Sentry.Tests/Internals/PollingNetworkStatusListenerTests.cs b/test/Sentry.Tests/Internals/PollingNetworkStatusListenerTests.cs
@@ -0,0 +1,53 @@
+namespace Sentry.Tests.Internals;
+
+public class PollingNetworkStatusListenerTest
+{
+    [Fact]
+    public async Task HostAvailable_CheckOnlyRunsOnce()
+    {
+        // Arrange
+        var initialDelay = 100;
+        var pingHost = Substitute.For<IPing>();
+        pingHost
+            .IsAvailableAsync()
+            .Returns(Task.FromResult(true));
+
+        var pollingListener = new PollingNetworkStatusListener(pingHost, initialDelay);
+        pollingListener.Online = false;
+
+        // Act
+        var waitForNetwork = pollingListener.WaitForNetworkOnlineAsync();
+        var timeout = Task.Delay(1000);
+        var completedTask = await Task.WhenAny(waitForNetwork, timeout);
+
+        // Assert
+        completedTask.Should().Be(waitForNetwork);
+        pollingListener.Online.Should().Be(true);
+        await pingHost.Received(1).IsAvailableAsync();
+    }
+
+    [Fact]
+    public async Task HostUnavailable_ShouldIncreaseDelay()
+    {
+        // Arrange
+        var initialDelay = 100; // set initial delay to ease the testing
+        var pingHost = Substitute.For<IPing>();
+        pingHost
+            .IsAvailableAsync()
+            .Returns(Task.FromResult(false));
+
+        var pollingListener = new PollingNetworkStatusListener(pingHost, initialDelay);
+        pollingListener.Online = false;
+
+        // Act
+        var waitForNetwork = pollingListener.WaitForNetworkOnlineAsync();
+        var timeout = Task.Delay(2000);
+        var completedTask = await Task.WhenAny(waitForNetwork, timeout);
+
+        // Assert
+        completedTask.Should().Be(timeout);
+        pollingListener.Online.Should().Be(false);
+        await pingHost.Received().IsAvailableAsync();
+        pollingListener._delayInMilliseconds.Should().BeGreaterThan(initialDelay);
+    }
+}
