perf: Schedule rate-limit notifications on shared executor (JAVA-653)

CHANGELOG.md

@@ -14,6 +14,7 @@
 ### Performance
 
 - Reduce the number of SDK threads: `LifecycleWatcher` now schedules the session-end task on the shared timer executor instead of creating a dedicated `java.util.Timer` thread ([#5819](https://github.com/getsentry/sentry-java/pull/5819))
+- Reduce the number of SDK threads: `RateLimiter` now schedules its rate-limit-lifted notifications on the shared timer executor instead of creating a dedicated `java.util.Timer` thread ([#5814](https://github.com/getsentry/sentry-java/pull/5814))
 - Speed up deserialization of arbitrary JSON objects by typing numbers without throwing exceptions ([#5783](https://github.com/getsentry/sentry-java/pull/5783))
 
 ### Dependencies

sentry/src/main/java/io/sentry/transport/RateLimiter.java

@@ -23,12 +23,13 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.Date;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.Timer;
-import java.util.TimerTask;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.Future;
+import java.util.concurrent.RejectedExecutionException;
 import org.jetbrains.annotations.NotNull;
 import org.jetbrains.annotations.Nullable;
 
@@ -42,8 +43,9 @@ public final class RateLimiter implements Closeable {
   private final @NotNull Map<DataCategory, @NotNull Date> sentryRetryAfterLimit =
       new ConcurrentHashMap<>();
   private final @NotNull List<IRateLimitObserver> rateLimitObservers = new CopyOnWriteArrayList<>();
-  private @Nullable Timer timer = null;
-  private final @NotNull AutoClosableReentrantLock timerLock = new AutoClosableReentrantLock();
+  private final @NotNull List<Future<?>> notifyObserversFutures = new ArrayList<>();
+  private final @NotNull AutoClosableReentrantLock notifyFuturesLock =
+      new AutoClosableReentrantLock();
 
   public RateLimiter(
       final @NotNull ICurrentDateProvider currentDateProvider,
@@ -278,11 +280,11 @@ public void updateRetryAfterLimits(
                   continue;
                 }
 
-                applyRetryAfterOnlyIfLonger(dataCategory, date);
+                applyRetryAfterOnlyIfLonger(dataCategory, date, retryAfterMillis);
               }
             } else {
               // if categories are empty, we should apply to "all" categories.
-              applyRetryAfterOnlyIfLonger(DataCategory.All, date);
+              applyRetryAfterOnlyIfLonger(DataCategory.All, date, retryAfterMillis);
             }
           }
         }
@@ -291,7 +293,7 @@ public void updateRetryAfterLimits(
       final long retryAfterMillis = parseRetryAfterOrDefault(retryAfterHeader);
       // we dont care if Date is UTC as we just add the relative seconds
       final Date date = new Date(currentDateProvider.getCurrentTimeMillis() + retryAfterMillis);
-      applyRetryAfterOnlyIfLonger(DataCategory.All, date);
+      applyRetryAfterOnlyIfLonger(DataCategory.All, date, retryAfterMillis);
     }
   }
 
@@ -300,10 +302,11 @@ public void updateRetryAfterLimits(
    *
    * @param dataCategory the DataCategory
    * @param date the Date to be applied
+   * @param delayMillis the millis until the rate limit is lifted
    */
   @SuppressWarnings({"JdkObsolete", "JavaUtilDate"})
   private void applyRetryAfterOnlyIfLonger(
-      final @NotNull DataCategory dataCategory, final @NotNull Date date) {
+      final @NotNull DataCategory dataCategory, final @NotNull Date date, final long delayMillis) {
     final Date oldDate = sentryRetryAfterLimit.get(dataCategory);
 
     // only overwrite its previous date if the limit is even longer
@@ -312,19 +315,25 @@ private void applyRetryAfterOnlyIfLonger(
 
       notifyRateLimitObservers();
 
-      try (final @NotNull ISentryLifecycleToken ignored = timerLock.acquire()) {
-        if (timer == null) {
-          timer = new Timer(true);
+      // notify observers again once the rate limit is lifted, using the shared timer executor
+      // instead of a dedicated Timer thread
+      try (final @NotNull ISentryLifecycleToken ignored = notifyFuturesLock.acquire()) {
+        final @NotNull Iterator<Future<?>> iterator = notifyObserversFutures.iterator();
+        while (iterator.hasNext()) {
+          if (iterator.next().isDone()) {
+            iterator.remove();
+          }
+        }
+        try {
+          notifyObserversFutures.add(
+              options
+                  .getTimerExecutorService()
+                  .schedule(this::notifyRateLimitObservers, delayMillis));
+        } catch (RejectedExecutionException e) {
+          options
+              .getLogger()
+              .log(SentryLevel.WARNING, "Failed to schedule rate limit lifted notification.", e);
         }
-
-        timer.schedule(
-            new TimerTask() {
-              @Override
-              public void run() {
-                notifyRateLimitObservers();
-              }
-            },
-            date);
       }
     }
   }
@@ -364,11 +373,11 @@ public void removeRateLimitObserver(@NotNull final IRateLimitObserver observer)
 
   @Override
   public void close() throws IOException {
-    try (final @NotNull ISentryLifecycleToken ignored = timerLock.acquire()) {
-      if (timer != null) {
-        timer.cancel();
-        timer = null;
+    try (final @NotNull ISentryLifecycleToken ignored = notifyFuturesLock.acquire()) {
+      for (Future<?> future : notifyObserversFutures) {
+        future.cancel(false);
       }
+      notifyObserversFutures.clear();
     }
     rateLimitObservers.clear();
   }

sentry/src/test/java/io/sentry/transport/RateLimiterTest.kt

@@ -18,6 +18,7 @@ import io.sentry.SentryEnvelope
 import io.sentry.SentryEnvelopeHeader
 import io.sentry.SentryEnvelopeItem
 import io.sentry.SentryEvent
+import io.sentry.SentryExecutorService
 import io.sentry.SentryLogEvent
 import io.sentry.SentryLogEvents
 import io.sentry.SentryLogLevel
@@ -37,12 +38,12 @@ import io.sentry.protocol.SentryId
 import io.sentry.protocol.SentryTransaction
 import io.sentry.protocol.User
 import io.sentry.test.getProperty
-import io.sentry.test.injectForField
 import io.sentry.util.HintUtils
 import java.io.File
-import java.util.Timer
 import java.util.UUID
+import java.util.concurrent.Future
 import java.util.concurrent.atomic.AtomicBoolean
+import kotlin.test.AfterTest
 import kotlin.test.Test
 import kotlin.test.assertEquals
 import kotlin.test.assertFalse
@@ -63,9 +64,14 @@ class RateLimiterTest {
     val currentDateProvider = mock<ICurrentDateProvider>()
     val clientReportRecorder = mock<IClientReportRecorder>()
     val serializer = mock<ISerializer>()
+    var executorService: SentryExecutorService? = null
 
     fun getSUT(): RateLimiter {
       val options = SentryOptions().apply { setLogger(NoOpLogger.getInstance()) }
+      // a real executor so scheduled rate-limit-lifted notifications actually run
+      val timerExecutorService = SentryExecutorService(options)
+      executorService = timerExecutorService
+      options.setTimerExecutorService(timerExecutorService)
 
       SentryOptionsManipulator.setClientReportRecorder(options, clientReportRecorder)
 
@@ -75,6 +81,12 @@ class RateLimiterTest {
 
   private val fixture = Fixture()
 
+  @AfterTest
+  fun `tear down`() {
+    // the executor's core thread never times out, so it would stay parked for the whole test JVM
+    fixture.executorService?.close(0)
+  }
+
   @Test
   fun `uses X-Sentry-Rate-Limit and allows sending if time has passed`() {
     val rateLimiter = fixture.getSUT()
@@ -654,7 +666,7 @@ class RateLimiterTest {
   }
 
   @Test
-  fun `apply rate limits schedules a timer to notify observers of lifted limits`() {
+  fun `apply rate limits schedules a task to notify observers of lifted limits`() {
     val rateLimiter = fixture.getSUT()
     whenever(fixture.currentDateProvider.currentTimeMillis).thenReturn(0, 1, 2001)
 
@@ -667,18 +679,19 @@ class RateLimiterTest {
   }
 
   @Test
-  fun `close cancels the timer`() {
+  fun `close cancels pending notify tasks`() {
     val rateLimiter = fixture.getSUT()
-    val timer = mock<Timer>()
-    rateLimiter.injectForField("timer", timer)
+    rateLimiter.updateRetryAfterLimits("60:replay:key", null, 1)
+
+    val futures = rateLimiter.getProperty<List<Future<*>>>("notifyObserversFutures")
+    assertEquals(1, futures.size)
+    val future = futures.first()
 
     // When the rate limiter is closed
     rateLimiter.close()
 
-    // Then the timer is cancelled
-    verify(timer).cancel()
-
-    // And is removed by the rateLimiter
-    assertNull(rateLimiter.getProperty("timer"))
+    // Then the pending notify task is cancelled and dropped
+    assertTrue(future.isCancelled)
+    assertTrue(rateLimiter.getProperty<List<Future<*>>>("notifyObserversFutures").isEmpty())
   }
 }