feat: replace permanent event-delivery disable with a bounded cooldown

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,79 @@ 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.isUnexpectedFailure()) {
+        noteUnexpectedFailure(token);
+      } else if (result.isSuccess()) {
+        if (cooldown.recordSuccess(token)) {
+          logger.info("Event delivery resumed after cooldown");
+        }
+      } else {
+        cooldown.recordFailure(token);
+      }
+    }
+
+    // The cooldown governs analytics delivery only (EVTPROC section 1.8 scopes it to flush
+    // payloads). A diagnostic delivery never touches cooldown state.
+    private void handleDiagnosticResponse(EventSender.Result result) {
       if (result.getTimeFromServer() != null) {
         lastKnownPastTime.set(result.getTimeFromServer().getTime());
       }
-      if (result.isMustShutDown()) {
-        disabled.set(true);
+    }
+
+    private void noteUnexpectedFailure(DeliveryCooldownTracker.WindowToken token) {
+      long windowMillis = cooldown.recordUnexpectedFailure(token);
+      if (windowMillis > 0) {
+        // The refused batch's index and identify events were lost with it, so forget every
+        // context key
+        if (contextDeduplicator != null) {
+          inbox.offer(new EventProcessorMessage(MessageType.FLUSH_USERS, null, false));
+        }
+        logger.error("Event delivery failed with an unexpected error;"
+            + " event delivery will not be attempted for the next {} seconds", windowMillis / 1000);
       }
     }
 
@@ -652,7 +732,7 @@ public void run() {
             writer.flush();
             EventSender.Result result = eventsConfig.eventSender.sendDiagnosticEvent(
                 buffer.toByteArray(), eventsConfig.eventsUri);
-            handleResponse(result);
+            handleDiagnosticResponse(result);
             if (diagnosticEvent.initEvent) {
               didSendInitEvent.set(true);
             }
@@ -716,10 +796,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 +811,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 {
@@ -781,23 +863,45 @@ public void run() {
           continue;
         }
         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(
-              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));
-        }
-        synchronized (activeFlushWorkersCount) {
-          activeFlushWorkersCount.decrementAndGet();
-          activeFlushWorkersCount.notifyAll();
+          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();
+            result = eventsConfig.eventSender.sendAnalyticsEvents(
+                buffer.toByteArray(),
+                outputEventCount,
+                eventsConfig.eventsUri
+                );
+          } 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();
+          }
+          if (result == null) {
+            // A custom EventSender that returns no result reported no delivery, so treat it as an
+            // ordinary failure.
+            result = EventSender.Result.failure();
+          }
+          try {
+            responseListener.handleResponse(result, payload.token);
+          } catch (Exception e) {
+            // Any remaining throw from response handling must not kill this worker thread.
+            logger.error("Unexpected error in event processor: {}", LogValues.exceptionSummary(e));
+            logger.debug(LogValues.exceptionTrace(e));
+          }
+        } finally {
+          // The busy-worker count must always come back down, even if an Error (not an Exception)
+          // escaped the work above, or close() would wait forever in
+          // waitUntilAllFlushWorkersInactive. The Error still ends this worker thread.
+          synchronized (activeFlushWorkersCount) {
+            activeFlushWorkersCount.decrementAndGet();
+            activeFlushWorkersCount.notifyAll();
+          }
         }
       }
     }

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,29 @@ 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 (response.code() == 413) {
+          logger.warn("Error {} (the payload was too large; it was dropped and not retried): {}",
+              errorContext, errorDesc);
           break;
         }
+        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, LogValues.exceptionSummary(e));
       }
     }
 
-    return new Result(false, mustShutDown, null);
+    return unexpected ? Result.unexpectedFailure() : Result.failure();
   }
 
   private final Date parseResponseDate(Response response) {