feat: replace permanent event-delivery disable with a bounded cooldown (SDK-3375)

lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessor.java

@@ -319,6 +319,7 @@ static final class EventDispatcher {
     private final AtomicInteger busyFlushWorkersCount;
     private final AtomicLong lastKnownPastTime = new AtomicLong(0);
     private final AtomicBoolean disabled = new AtomicBoolean(false);
+    private final DeliveryCooldownTracker cooldown;
     private final AtomicBoolean didSendInitEvent = new AtomicBoolean(false);
     final DiagnosticStore diagnosticStore; // visible for testing
     private final EventContextDeduplicator contextDeduplicator;
@@ -346,6 +347,7 @@ private EventDispatcher(
       this.diagnosticStore = eventsConfig.diagnosticStore;
       this.busyFlushWorkersCount = new AtomicInteger(0);
       this.logger = logger;
+      this.cooldown = new DeliveryCooldownTracker(eventsConfig.cooldownNanoTimeFn);
 
       ThreadFactory threadFactory = new ThreadFactory() {
         @Override
@@ -603,10 +605,21 @@ private boolean shouldDebugEvent(Event.FeatureRequest fe) {
     }
 
     private void triggerFlush(EventBuffer outbox, BlockingQueue<FlushPayload> payloadQueue) {
-      if (disabled.get() || outbox.isEmpty()) {
+      if (disabled.get()) {
+        return;
+      }
+      // The cooldown check and the token issue are one atomic step, so a window opening
+      // concurrently on a worker thread either diverts this flush to the discard path or makes
+      // the dispatched payload's token stale.
+      DeliveryCooldownTracker.WindowToken token = cooldown.beginDeliveryUnlessInCooldown();
+      if (token == null) {
+        discardPayloadInCooldown(outbox);
+        return;
+      }
+      if (outbox.isEmpty()) {
         return;
       }
-      FlushPayload payload = outbox.getPayload();
+      FlushPayload payload = outbox.getPayload(token);
       if (diagnosticStore != null) {
         int summaryCount = 0;
         for (EventSummary summary : payload.summaries) {
@@ -632,12 +645,80 @@ private void triggerFlush(EventBuffer outbox, BlockingQueue<FlushPayload> payloa
       }
     }
 
-    private void handleResponse(EventSender.Result result) {
+    // While a window is open, a flush of any kind is a no-op that behaves exactly as if the batch
+    // had been sent and the service had refused it: the buffered events are disposed of, and any
+    // context introductions they carried are forgotten so a later payload can introduce those
+    // contexts again. Only the network attempt and the serialization work are skipped.
+    private void discardPayloadInCooldown(EventBuffer outbox) {
+      if (outbox.isEmpty()) {
+        return;
+      }
+      // No batch reached the service, so the diagnostic batch-size stat must not report one.
+      if (diagnosticStore != null) {
+        diagnosticStore.recordEventsInBatch(0);
+      }
+      FlushPayload payload = outbox.getPayload(null);
+      outbox.clear();
+      int eventCount = payload.events.length;
+      for (EventSummary summary : payload.summaries) {
+        if (!summary.isEmpty()) {
+          eventCount++;
+        }
+      }
+      boolean hadContextIntroductions = false;
+      for (Event e : payload.events) {
+        if (e instanceof Event.Index || e instanceof Event.Identify) {
+          hadContextIntroductions = true;
+          break;
+        }
+      }
+      if (hadContextIntroductions && contextDeduplicator != null) {
+        contextDeduplicator.flush();
+      }
+      long remainingSeconds = Math.max(1, (cooldown.remainingMillis() + 999) / 1000);
+      if (cooldown.takeDiscardWarning()) {
+        logger.warn("Discarded {} events: event delivery is cooling down for another {} seconds",
+            eventCount, remainingSeconds);
+      } else {
+        logger.debug("Discarded {} events: event delivery is cooling down for another {} seconds",
+            eventCount, remainingSeconds);
+      }
+    }
+
+    private void handleResponse(EventSender.Result result, DeliveryCooldownTracker.WindowToken token) {
       if (result.getTimeFromServer() != null) {
         lastKnownPastTime.set(result.getTimeFromServer().getTime());
       }
-      if (result.isMustShutDown()) {
-        disabled.set(true);
+      if (result.isUnexpectedFailure()) {
+        noteUnexpectedFailure(token);
+      } else if (result.isSuccess()) {
+        if (cooldown.recordSuccess(token)) {
+          logger.info("Event delivery resumed after cooldown");
+        }
+      } else {
+        cooldown.recordFailure(token);
+      }
+    }
+
+    // Diagnostic deliveries are not gated by the cooldown, and their outcomes drive it only the
+    // way they drove the old permanent disable: an unexpected failure opens a window, while a
+    // diagnostic success or ordinary failure has no effect on delivery state. In particular, a
+    // diagnostic success is not the probe and cannot count toward the reset threshold; only
+    // analytics deliveries prove the service is accepting analytics payloads again.
+    private void handleDiagnosticResponse(EventSender.Result result, DeliveryCooldownTracker.WindowToken token) {
+      if (result.getTimeFromServer() != null) {
+        lastKnownPastTime.set(result.getTimeFromServer().getTime());
+      }
+      if (result.isUnexpectedFailure()) {
+        noteUnexpectedFailure(token);
+      }
+    }
+
+    private void noteUnexpectedFailure(DeliveryCooldownTracker.WindowToken token) {
+      long windowMillis = cooldown.recordUnexpectedFailure(token);
+      if (windowMillis > 0) {
+        logger.error("Event delivery failed with an unexpected error;"
+            + " event delivery will not be attempted for the next {} seconds", windowMillis / 1000);
       }
     }
 
@@ -650,9 +731,13 @@ public void run() {
             Writer writer = new BufferedWriter(new OutputStreamWriter(buffer, Charset.forName("UTF-8")), INITIAL_OUTPUT_BUFFER_SIZE);
             gson.toJson(diagnosticEvent.value, writer);
             writer.flush();
+            // The send itself is not gated by the cooldown, but a diagnostic dispatched inside a
+            // window gets a null token, so its outcome cannot move delivery state even if it
+            // lands after the window has elapsed.
+            DeliveryCooldownTracker.WindowToken token = cooldown.beginDeliveryUnlessInCooldown();
             EventSender.Result result = eventsConfig.eventSender.sendDiagnosticEvent(
                 buffer.toByteArray(), eventsConfig.eventsUri);
-            handleResponse(result);
+            handleDiagnosticResponse(result, token);
             if (diagnosticEvent.initEvent) {
               didSendInitEvent.set(true);
             }
@@ -716,10 +801,10 @@ long getAndClearDroppedCount() {
       return res;
     }
 
-    FlushPayload getPayload() {
+    FlushPayload getPayload(DeliveryCooldownTracker.WindowToken token) {
       Event[] eventsOut = events.toArray(new Event[events.size()]);
       List<EventSummarizer.EventSummary> summaries = summarizer.getSummariesAndReset();
-      return new FlushPayload(eventsOut, summaries);
+      return new FlushPayload(eventsOut, summaries, token);
     }
 
     void clear() {
@@ -731,15 +816,17 @@ void clear() {
   private static final class FlushPayload {
     final Event[] events;
     final List<EventSummary> summaries;
+    final DeliveryCooldownTracker.WindowToken token;
 
-    FlushPayload(Event[] events, List<EventSummary> summaries) {
+    FlushPayload(Event[] events, List<EventSummary> summaries, DeliveryCooldownTracker.WindowToken token) {
       this.events = events;
       this.summaries = summaries;
+      this.token = token;
     }
   }
 
   private static interface EventResponseListener {
-    void handleResponse(EventSender.Result result);
+    void handleResponse(EventSender.Result result, DeliveryCooldownTracker.WindowToken token);
   }
 
   private static final class SendEventsTask implements Runnable {
@@ -780,20 +867,33 @@ public void run() {
         } catch (InterruptedException e) {
           continue;
         }
+        EventSender.Result result;
         try {
           ByteArrayOutputStream buffer = new ByteArrayOutputStream(INITIAL_OUTPUT_BUFFER_SIZE);
           Writer writer = new BufferedWriter(new OutputStreamWriter(buffer, Charset.forName("UTF-8")), INITIAL_OUTPUT_BUFFER_SIZE);
           int outputEventCount = formatter.writeOutputEvents(payload.events, payload.summaries, writer);
           writer.flush();
-          EventSender.Result result = eventsConfig.eventSender.sendAnalyticsEvents(
+          result = eventsConfig.eventSender.sendAnalyticsEvents(
               buffer.toByteArray(),
               outputEventCount,
               eventsConfig.eventsUri
               );
-          responseListener.handleResponse(result);
         } catch (Exception e) {
           logger.error("Unexpected error in event processor: {}", LogValues.exceptionSummary(e));
           logger.debug(LogValues.exceptionTrace(e));
+          // The batch was not delivered, and the delivery schedule must know that: an exception
+          // here counts as an ordinary failure, which breaks the consecutive-success streak
+          // without opening a cooldown window.
+          result = EventSender.Result.failure();
+        }
+        try {
+          responseListener.handleResponse(result, payload.token);
+        } catch (Exception e) {
+          // An exception here (such as a null result from a custom EventSender) must not kill
+          // this worker thread: the busy-worker count would stay elevated forever and close()
+          // would never return.
+          logger.error("Unexpected error in event processor: {}", LogValues.exceptionSummary(e));
+          logger.debug(LogValues.exceptionTrace(e));
         }
         synchronized (activeFlushWorkersCount) {
           activeFlushWorkersCount.decrementAndGet();

lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventSender.java

@@ -2,6 +2,7 @@
 
 import com.launchdarkly.logging.LDLogger;
 import com.launchdarkly.logging.LogValues;
+import com.launchdarkly.sdk.internal.http.FailureClass;
 import com.launchdarkly.sdk.internal.http.HttpHelpers;
 import com.launchdarkly.sdk.internal.http.HttpProperties;
 
@@ -16,7 +17,7 @@
 import java.util.UUID;
 import java.util.zip.GZIPOutputStream;
 
-import static com.launchdarkly.sdk.internal.http.HttpErrors.checkIfErrorIsRecoverableAndLog;
+import static com.launchdarkly.sdk.internal.http.HttpErrors.classifyHttpFailure;
 import static com.launchdarkly.sdk.internal.http.HttpErrors.httpErrorDescription;
 
 import okhttp3.Headers;
@@ -152,7 +153,7 @@ private CompressionResult<byte[]> compressData(byte[] data) {
   private Result sendEventData(boolean isDiagnostic, byte[] data, int eventCount, URI eventsBaseUri) {
     if (data == null || data.length == 0) {
       // DefaultEventProcessor won't normally pass us an empty payload, but if it does, don't bother sending
-      return new Result(true, false, null);
+      return Result.success(null);
     }
 
     Headers.Builder headersBuilder = baseHeaders.newBuilder();
@@ -173,7 +174,7 @@ private Result sendEventData(boolean isDiagnostic, byte[] data, int eventCount,
     URI uri = HttpHelpers.concatenateUriPath(eventsBaseUri, path);
     CompressionResult<byte[]> compressionResult = compressData(data);
     RequestBody body = RequestBody.create(compressionResult.data, JSON_CONTENT_TYPE);
-    boolean mustShutDown = false;
+    boolean unexpected = false;
 
     if (compressionResult.wasCompressed) {
       headersBuilder.add("Content-Encoding", "gzip");
@@ -207,27 +208,24 @@ private Result sendEventData(boolean isDiagnostic, byte[] data, int eventCount,
         logger.debug("{} delivery took {} ms, response status {}", description, endTime - startTime, response.code());
 
         if (response.isSuccessful()) {
-          return new Result(true, false, parseResponseDate(response));
+          return Result.success(parseResponseDate(response));
         }
 
         String errorDesc = httpErrorDescription(response.code());
-        boolean recoverable = checkIfErrorIsRecoverableAndLog(
-            logger,
-            errorDesc,
-            errorContext,
-            response.code(),
-            nextActionMessage
-            );
-        if (!recoverable) {
-          mustShutDown = true;
+        if (classifyHttpFailure(response.code()) == FailureClass.UNEXPECTED) {
+          logger.error("Error {} (the payload will not be retried): {}", errorContext, errorDesc);
+          unexpected = true;
           break;
         }
+        logger.warn("Error {} ({}): {}", errorContext, nextActionMessage, errorDesc);
       } catch (IOException e) {
-        checkIfErrorIsRecoverableAndLog(logger, e.toString(), errorContext, 0, nextActionMessage);
+        // Transport-level failures are always worth one retry here, whatever their cause; the
+        // request may simply have hit a transient network problem.
+        logger.warn("Error {} ({}): {}", errorContext, nextActionMessage, e.toString());
       }
     }
 
-    return new Result(false, mustShutDown, null);
+    return unexpected ? Result.unexpectedFailure() : Result.failure();
   }
 
   private final Date parseResponseDate(Response response) {