From 828021bd8548d94d0ce947af6a440097fb723b1a Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Fri, 6 Nov 2026 21:52:16 -0500 Subject: [PATCH 1/3] refactor: migrate the event sender off the deprecated HTTP error classifier DefaultEventSender now classifies HTTP failures with HttpErrors.classifyHttpFailure instead of the deprecated isHttpErrorRecoverable/checkIfErrorIsRecoverableAndLog helpers. Behavior and log output are unchanged: the set of statuses treated as unexpected is identical, unexpected failures still skip the retry and still log at error level with the same wording, and transport-level exceptions keep their warn-and-retry-once handling. --- .../internal/events/DefaultEventSender.java | 18 ++++++++---------- 1 file changed, 8 insertions(+), 10 deletions(-) diff --git a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventSender.java b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventSender.java index 29cdcdf6..74e6c5f8 100644 --- a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventSender.java +++ b/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; @@ -211,19 +212,16 @@ private Result sendEventData(boolean isDiagnostic, byte[] data, int eventCount, } String errorDesc = httpErrorDescription(response.code()); - boolean recoverable = checkIfErrorIsRecoverableAndLog( - logger, - errorDesc, - errorContext, - response.code(), - nextActionMessage - ); - if (!recoverable) { + if (classifyHttpFailure(response.code()) == FailureClass.UNEXPECTED) { + logger.error("Error {} (giving up permanently): {}", errorContext, errorDesc); mustShutDown = 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()); } } From 0cd76e21c7a217110677115ba88033795d9c4484 Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Fri, 6 Nov 2026 21:52:17 -0500 Subject: [PATCH 2/3] feat: replace permanent event-delivery disable with a bounded cooldown (SDK-3375) On an unexpected delivery failure (a non-recoverable HTTP status such as 401 or 403), the event processor now enters a bounded, self-recovering cooldown window instead of permanently disabling delivery for the lifetime of the SDK instance, per the EVTPROC spec's Section 1.8. Windows double from 5 minutes to a 1 hour ceiling, one step per failed probe. In-window flushes of any kind are no-ops that dispose of the batch exactly as a sent-and-refused payload would be, without the network or serialization cost. Recording is never gated, retry state resets after two consecutive successful analytics deliveries, and no cooldown state persists across restarts. Diagnostic sends are not gated by the window; an unexpected diagnostic failure opens one, as it previously disabled delivery, but diagnostic successes do not drive the delivery schedule. EventSender.Result gains the unexpected classification with factory methods and the isUnexpectedFailure accessor; the legacy constructor and isMustShutDown are deprecated and still dual-written for consumers that have not migrated. --- .../events/DefaultEventProcessor.java | 124 +++- .../internal/events/DefaultEventSender.java | 12 +- .../events/DeliveryCooldownTracker.java | 243 ++++++++ .../sdk/internal/events/EventSender.java | 97 +++- .../internal/events/EventsConfiguration.java | 6 + .../sdk/internal/events/BaseEventTest.java | 2 +- .../DefaultEventProcessorCooldownTest.java | 539 ++++++++++++++++++ .../DefaultEventProcessorDiagnosticsTest.java | 19 +- .../DefaultEventProcessorOutputTest.java | 8 +- .../events/DefaultEventProcessorTest.java | 14 +- .../events/DefaultEventSenderTest.java | 31 + .../events/DeliveryCooldownTrackerTest.java | 324 +++++++++++ 12 files changed, 1362 insertions(+), 57 deletions(-) create mode 100644 lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java create mode 100644 lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorCooldownTest.java create mode 100644 lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java diff --git a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessor.java b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessor.java index f258243b..8b87cb31 100644 --- a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessor.java +++ b/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 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 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 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 summaries; + final DeliveryCooldownTracker.WindowToken token; - FlushPayload(Event[] events, List summaries) { + FlushPayload(Event[] events, List 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(); diff --git a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventSender.java b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventSender.java index 74e6c5f8..623abde6 100644 --- a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventSender.java +++ b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DefaultEventSender.java @@ -153,7 +153,7 @@ private CompressionResult 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(); @@ -174,7 +174,7 @@ private Result sendEventData(boolean isDiagnostic, byte[] data, int eventCount, URI uri = HttpHelpers.concatenateUriPath(eventsBaseUri, path); CompressionResult 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"); @@ -208,13 +208,13 @@ 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()); if (classifyHttpFailure(response.code()) == FailureClass.UNEXPECTED) { - logger.error("Error {} (giving up permanently): {}", errorContext, errorDesc); - mustShutDown = true; + logger.error("Error {} (the payload will not be retried): {}", errorContext, errorDesc); + unexpected = true; break; } logger.warn("Error {} ({}): {}", errorContext, nextActionMessage, errorDesc); @@ -225,7 +225,7 @@ private Result sendEventData(boolean isDiagnostic, byte[] data, int eventCount, } } - return new Result(false, mustShutDown, null); + return unexpected ? Result.unexpectedFailure() : Result.failure(); } private final Date parseResponseDate(Response response) { diff --git a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java new file mode 100644 index 00000000..e4a7e6c1 --- /dev/null +++ b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java @@ -0,0 +1,243 @@ +package com.launchdarkly.sdk.internal.events; + +/** + * Decides when analytics event delivery is in a cooldown window after an unexpected failure, so + * that a condition such as an invalid SDK key backs delivery off instead of permanently disabling + * it. + *

+ * Every delivery outcome is reported together with the {@link WindowToken} issued when that + * delivery was dispatched. A token stops being current when a window opens or ends, so an outcome + * from a delivery that was already in flight across that boundary updates nothing. This is what + * confines ladder movement and the success streak to actual probes even with concurrent + * deliveries straddling a window boundary. + *

+ * The windows are not floored at the flush interval: delivery is attempted only when a flush + * occurs (scheduled, manual, or at shutdown), and in-window flushes of any kind make no attempt, + * so a window shorter than the flush interval cannot make the processor try faster while failing + * than while healthy. + *

+ * The tracker is pure state: it starts no tasks and does no I/O or logging. Deliveries are + * dispatched from the main dispatcher thread while their outcomes arrive on delivery worker + * threads, so all methods are synchronized; callers must not do anything slow inside a pattern + * that would extend the lock's reach. + */ +final class DeliveryCooldownTracker { + static final long COOLDOWN_INITIAL_WINDOW_MILLIS = 5 * 60_000L; + static final long COOLDOWN_MAX_WINDOW_MILLIS = 60 * 60_000L; + static final int COOLDOWN_RESET_THRESHOLD = 2; + + private static final long NANOS_PER_MILLI = 1_000_000L; + + /** + * The tracker's time source. This stands in for {@code java.util.function.LongSupplier}, which + * cannot be used because this module must run on Android below API 24, where + * {@code java.util.function} does not exist. + */ + interface NanoTimeSource { + /** + * Returns a reading with {@link System#nanoTime()} semantics: a monotonic nanosecond value + * whose absolute magnitude is meaningless and may wrap. + * + * @return the reading + */ + long nanoTime(); + } + + /** + * Identifies the span between cooldown-window boundaries in which a delivery was dispatched; a + * token stops being current when a window opens or ends. The dispatcher obtains one from + * {@link #beginDelivery()} when it hands a payload off for delivery and presents it with the + * delivery's outcome; the tracker ignores outcomes whose token is no longer current. A null + * token is never current, so an unstamped payload cannot pass for a live delivery. + */ + static final class WindowToken { + private final long generation; + + private WindowToken(long generation) { + this.generation = generation; + } + } + + private final NanoTimeSource nanoTimeFn; + private int attempts = 0; + private int consecutiveSuccesses = 0; + private Long windowEndNanos = null; + private long epoch = 0; + private boolean warnedThisWindow = false; + + /** + * Creates an instance. + * + * @param nanoTimeFn the time source; null to use {@code System.nanoTime()}. Only tests pass a + * non-null value. + */ + DeliveryCooldownTracker(NanoTimeSource nanoTimeFn) { + this.nanoTimeFn = nanoTimeFn == null ? System::nanoTime : nanoTimeFn; + } + + /** + * Returns true if a cooldown window is in effect. + * + * @return true if delivery is cooling down + */ + synchronized boolean inCooldown() { + // The subtraction-based comparison stays correct when nanoTime wraps. + return windowEndNanos != null && nanoTimeFn.nanoTime() - windowEndNanos < 0; + } + + /** + * Returns the token that a delivery dispatched now must present with its outcome. + * + * @return the token + */ + synchronized WindowToken beginDelivery() { + return new WindowToken(epoch + 1); + } + + /** + * Returns the token for a delivery dispatched now, or null if a cooldown window is in effect. + * The check and the token issue are one atomic step, so a window cannot open in between: a + * delivery either sees the window, or carries a token that the window's opening made stale. + * + * @return the token, or null if delivery is cooling down + */ + synchronized WindowToken beginDeliveryUnlessInCooldown() { + return inCooldown() ? null : beginDelivery(); + } + + /** + * Returns true at most once per window, for the caller to make the first discard of each window + * more prominent than the rest. The check-and-set runs under the same lock that opens and arms + * windows. + * + * @return true if this is the first time this is called for the current window + */ + synchronized boolean takeDiscardWarning() { + if (warnedThisWindow) { + return false; + } + warnedThisWindow = true; + return true; + } + + /** + * Returns how much of the current window is left, or zero outside a window. + * + * @return the remaining window length in milliseconds + */ + synchronized long remainingMillis() { + if (windowEndNanos == null) { + return 0; + } + long remainingNanos = windowEndNanos - nanoTimeFn.nanoTime(); + return remainingNanos > 0 ? remainingNanos / NANOS_PER_MILLI : 0; + } + + /** + * Records a successful delivery. The retry state resets only after + * {@link #COOLDOWN_RESET_THRESHOLD} consecutive successes. Stale outcomes, and outcomes + * arriving while a window is open, update nothing. + * + * @param token the token issued when this delivery was dispatched + * @return true when this success is the probe that ended a window (the first success after the + * window elapsed), so the caller can log the resumption + */ + synchronized boolean recordSuccess(WindowToken token) { + if (!isCurrent(token)) { + return false; + } + if (inCooldown()) { + // Defensive: analytics tokens are issued atomically with the cooldown check and go stale + // when a window opens, and diagnostic successes do not reach the tracker, so no current + // caller gets here. The invariant stands on its own: an outcome arriving while a window is + // open must never move delivery state. + return false; + } + boolean resumed = false; + if (windowEndNanos != null) { + windowEndNanos = null; + resumed = true; + // The probe ends the window-boundary round: the tokens of its concurrently dispatched + // siblings go stale, so the boundary round cannot combine parallel deliveries into a + // reset. (A sibling unexpected failure goes stale with them; the next failed delivery of + // the new round opens the window instead, at most a cadence later.) Later rounds leave the + // epoch unchanged, so parallel deliveries of one such round each count toward the streak: + // the reset threshold counts successful deliveries, not rounds. + epoch++; + } + consecutiveSuccesses++; + if (consecutiveSuccesses >= COOLDOWN_RESET_THRESHOLD) { + attempts = 0; + } + return resumed; + } + + /** + * Records a delivery failure that was not unexpected. It breaks the consecutive-success streak + * but never opens or lengthens a window. Stale outcomes, and outcomes arriving while a window + * is open, update nothing. + * + * @param token the token issued when this delivery was dispatched + */ + synchronized void recordFailure(WindowToken token) { + if (!isCurrent(token)) { + return; + } + if (inCooldown()) { + // Defensive; see recordSuccess. + return; + } + if (windowEndNanos != null) { + // A probe that fails recoverably still ends the window: clear it so a much later success is + // not misreported as the resumption, and close the boundary round (see recordSuccess) so the + // probe's concurrently dispatched siblings cannot combine into a reset. The accepted cost is + // that a sibling unexpected failure in the same round goes stale with them, deferring the + // next window by at most one cadence: the next round's failure opens it, with the ladder + // position preserved. + windowEndNanos = null; + epoch++; + } + consecutiveSuccesses = 0; + } + + /** + * Records a delivery failure classified as unexpected, opening a cooldown window. Stale + * outcomes update nothing: in particular, once one failure of a round has opened a window, its + * concurrently dispatched siblings cannot lengthen it or step the ladder again for the same + * incident. The window doubles from {@link #COOLDOWN_INITIAL_WINDOW_MILLIS} up to + * {@link #COOLDOWN_MAX_WINDOW_MILLIS} per failed probe. + * + * @param token the token issued when this delivery was dispatched + * @return the length in milliseconds of the newly opened window, or zero if the outcome was + * stale and no window was opened + */ + synchronized long recordUnexpectedFailure(WindowToken token) { + if (!isCurrent(token)) { + return 0; + } + if (inCooldown()) { + // Load-bearing: diagnostic deliveries are not gated by the window, so an in-window + // diagnostic failure arrives with a current token, and it must not lengthen the window or + // step the ladder again for the same incident. + return 0; + } + consecutiveSuccesses = 0; + attempts++; + epoch++; + warnedThisWindow = false; + long windowMillis = COOLDOWN_MAX_WINDOW_MILLIS; + // Compare against the ceiling scaled down rather than the base scaled up, so the shift cannot + // overflow. (A shift of 63 or more would always be clamped anyway.) + int exponent = attempts - 1; + if (exponent >= 0 && exponent < 63 + && COOLDOWN_INITIAL_WINDOW_MILLIS <= (COOLDOWN_MAX_WINDOW_MILLIS >> exponent)) { + windowMillis = COOLDOWN_INITIAL_WINDOW_MILLIS << exponent; + } + windowEndNanos = nanoTimeFn.nanoTime() + windowMillis * NANOS_PER_MILLI; + return windowMillis; + } + + private boolean isCurrent(WindowToken token) { + return token != null && token.generation == epoch + 1; + } +} diff --git a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/EventSender.java b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/EventSender.java index 2f2a31e5..5d48fc73 100644 --- a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/EventSender.java +++ b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/EventSender.java @@ -10,7 +10,7 @@ * This is separate from the public EventSender interface in the SDK that applications can use to * provide a custom implementation. The latter is used as a wrapper for this one, so we do not have * to expose any types from the internal events code. The public interface is simpler because it - * only needs to return success/failure/shutdown status; the use of the Date header is an + * only needs to return a success or failure classification; the use of the Date header is an * implementation detail that is specific to the default HTTP implementation of event delivery. */ public interface EventSender extends Closeable { @@ -41,28 +41,73 @@ public interface EventSender extends Closeable { * Encapsulates the results of a call to an EventSender method. */ public static final class Result { - private boolean success; - private boolean mustShutDown; - private Date timeFromServer; - + private final boolean success; + private final boolean mustShutDown; + private final boolean unexpected; + private final Date timeFromServer; + + private Result(boolean success, boolean mustShutDown, boolean unexpected, Date timeFromServer) { + this.success = success; + this.mustShutDown = mustShutDown; + this.unexpected = unexpected; + this.timeFromServer = timeFromServer; + } + /** * Constructs an instance. - * + * * @param success true if the events were delivered - * @param mustShutDown true if an unrecoverable error (such as an HTTP 401 error, implying that the - * SDK key is invalid) means the SDK should permanently stop trying to send events + * @param mustShutDown true if the failure was an unexpected error (such as an HTTP 401 error, + * implying that the SDK key is invalid) * @param timeFromServer the parsed value of an HTTP Date header received from the remote server, * if any; this is used to compensate for differences between the application's time and server time + * + * @deprecated Use {@link #success(Date)}, {@link #failure()}, or {@link #unexpectedFailure()}. + * The {@code mustShutDown} flag predates the delivery cooldown: it is now read through + * {@link #isUnexpectedFailure()} and no longer shuts anything down permanently. */ + @Deprecated public Result(boolean success, boolean mustShutDown, Date timeFromServer) { - this.success = success; - this.mustShutDown = mustShutDown; - this.timeFromServer = timeFromServer; + this(success, mustShutDown, false, timeFromServer); + } + + /** + * Returns a result representing a successful delivery. + * + * @param timeFromServer the parsed value of an HTTP Date header received from the remote + * server, if any; may be null + * @return the result + */ + public static Result success(Date timeFromServer) { + return new Result(true, false, false, timeFromServer); + } + + /** + * Returns a result representing an ordinary delivery failure, such as a service outage, that + * does not affect the delivery schedule. + * + * @return the result + */ + public static Result failure() { + return new Result(false, false, false, null); + } + + /** + * Returns a result representing a delivery failure classified as unexpected (such as an HTTP + * 401 error, implying that the SDK key is invalid), which makes event delivery back off. + *

+ * For the benefit of consumers that still read {@link #isMustShutDown()}, this also sets that + * legacy flag. + * + * @return the result + */ + public static Result unexpectedFailure() { + return new Result(false, true, true, null); } /** * Returns true if the events were delivered. - * + * * @return true if the events were delivered */ public boolean isSuccess() { @@ -70,19 +115,37 @@ public boolean isSuccess() { } /** - * Returns true if an unrecoverable error (such as an HTTP 401 error, implying that the - * SDK key is invalid) means the SDK should permanently stop trying to send events - * - * @return true if event delivery should shut down + * Returns true if the failure was an unexpected error (such as an HTTP 401 error, implying + * that the SDK key is invalid). + * + * @return true if the failure was unexpected + * + * @deprecated Use {@link #isUnexpectedFailure()}. An unexpected failure no longer shuts event + * delivery down permanently; it makes delivery back off. */ + @Deprecated public boolean isMustShutDown() { return mustShutDown; } + /** + * Returns true if the failure was classified as unexpected (such as an HTTP 401 error, + * implying that the SDK key is invalid). Such a failure makes event delivery back off. + *

+ * This is true if the result was constructed with {@link #unexpectedFailure()}, or if it + * carries the legacy shutdown flag, so results from senders written before this + * classification existed are interpreted correctly. + * + * @return true if the failure was unexpected + */ + public boolean isUnexpectedFailure() { + return unexpected || mustShutDown; + } + /** * Returns the parsed value of an HTTP Date header received from the remote server, if any. This * is used to compensate for differences between the application's time and server time. - * + * * @return a date value or null */ public Date getTimeFromServer() { diff --git a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/EventsConfiguration.java b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/EventsConfiguration.java index 6dbd750c..e53bd89d 100644 --- a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/EventsConfiguration.java +++ b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/EventsConfiguration.java @@ -33,6 +33,12 @@ public final class EventsConfiguration { final boolean perContextSummarization; final boolean redactAnonymousAllEvents; + // Test seam only, which is why it is not part of any constructor: lets unit tests drive the + // delivery cooldown across window boundaries with a fake clock. Null means System.nanoTime() is + // used. Tests must set this before constructing the processor, on the same thread; the field is + // never written after construction, so it needs no synchronization. + DeliveryCooldownTracker.NanoTimeSource cooldownNanoTimeFn = null; + /** * Creates an instance. * diff --git a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/BaseEventTest.java b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/BaseEventTest.java index 776185cc..b336d81d 100644 --- a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/BaseEventTest.java +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/BaseEventTest.java @@ -116,7 +116,7 @@ static final class CapturedPayload { public final class MockEventSender implements EventSender { volatile boolean closed; - volatile Result result = new Result(true, false, null); + volatile Result result = Result.success(null); volatile RuntimeException fakeError = null; volatile IOException fakeErrorOnClose = null; volatile CountDownLatch receivedCounter = null; diff --git a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorCooldownTest.java b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorCooldownTest.java new file mode 100644 index 00000000..83e616a7 --- /dev/null +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorCooldownTest.java @@ -0,0 +1,539 @@ +package com.launchdarkly.sdk.internal.events; + +import com.launchdarkly.logging.LDLogLevel; +import com.launchdarkly.logging.LogCapture; +import com.launchdarkly.sdk.LDContext; +import com.launchdarkly.sdk.LDValue; + +import org.junit.Test; + +import java.util.HashSet; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static com.launchdarkly.sdk.internal.events.DeliveryCooldownTracker.COOLDOWN_INITIAL_WINDOW_MILLIS; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.contains; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +/** + * These DefaultEventProcessor tests cover the delivery cooldown: an unexpected delivery failure + * suppresses delivery for a bounded window instead of permanently disabling it. + */ +@SuppressWarnings("javadoc") +public class DefaultEventProcessorCooldownTest extends BaseEventTest { + private final DeliveryCooldownTrackerTest.FakeNanoClock clock = new DeliveryCooldownTrackerTest.FakeNanoClock(); + + // A deduplicator with the real component's shape: a context is new the first time it is seen, + // and flush() forgets everything. + static final class RememberingDeduplicator implements EventContextDeduplicator { + private final Set seen = new HashSet<>(); + private int flushCount = 0; + + @Override + public Long getFlushInterval() { + return null; + } + + @Override + public synchronized boolean processContext(LDContext context) { + return seen.add(context.getFullyQualifiedKey()); + } + + @Override + public synchronized void flush() { + seen.clear(); + flushCount++; + } + + synchronized int getFlushCount() { + return flushCount; + } + } + + private DefaultEventProcessor makeProcessorWithClock(EventsConfigurationBuilder ec) { + EventsConfiguration config = ec.build(); + config.cooldownNanoTimeFn = clock::now; + return new DefaultEventProcessor(config, sharedExecutor, Thread.MAX_PRIORITY, testLogger); + } + + // Drives the processor into a cooldown window: the sender fails one delivery with the given + // result, and the test waits until the result has been handled. A single sync is a true + // barrier, because the flush worker reports the result before it is counted inactive. + private void enterCooldown(DefaultEventProcessor ep, MockEventSender es, EventSender.Result result) + throws Exception { + es.result = result; + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + es.result = EventSender.Result.success(null); + } + + private int countLogLines(LDLogLevel level, String substring) { + int n = 0; + for (LogCapture.Message message : logCapture.getMessages()) { + if (message.getLevel() == level && message.getText().contains(substring)) { + n++; + } + } + return n; + } + + @Test + public void unexpectedErrorEntersCooldownAndDeliveryResumesAfterWindow() throws Exception { + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + + // An in-window flush is a no-op: it makes no delivery attempt, and it disposes of the + // drained events exactly as a delivered-and-refused payload would be disposed of. + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); + es.expectNoRequests(100); + + // Once the window has elapsed, delivery resumes: a fresh event is delivered by the probe, + // and the discarded in-window event is not resurrected. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + Event.Identify postWindowEvent = identifyEvent(user); + ep.sendEvent(postWindowEvent); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains(isIdentifyEvent(postWindowEvent, userJson))); + } + } + + @SuppressWarnings("deprecation") + @Test + public void legacyMustShutDownResultEntersCooldownInsteadOfDisabling() throws Exception { + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + // A custom EventSender written before the unexpected classification existed sets only the + // legacy shutdown flag. + enterCooldown(ep, es, new EventSender.Result(false, true, null)); + + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); + es.expectNoRequests(100); + + // Delivery is suppressed only for the window, not permanently. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + Event.Identify postWindowEvent = identifyEvent(user); + ep.sendEvent(postWindowEvent); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains(isIdentifyEvent(postWindowEvent, userJson))); + } + } + + @SuppressWarnings("deprecation") + @Test + public void resultClaimingBothSuccessAndShutdownIsTreatedAsUnexpected() throws Exception { + // Only a custom EventSender can produce a result that claims success and an unexpected + // failure at once; the failure takes precedence, so delivery health is never assumed from a + // contradictory result. + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + enterCooldown(ep, es, new EventSender.Result(true, true, null)); + + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); + es.expectNoRequests(100); + } + } + + @Test + public void recoverableFailureDoesNotEnterCooldown() throws Exception { + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + es.result = EventSender.Result.failure(); + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + + // Delivery continues immediately; no window was opened. + es.result = EventSender.Result.success(null); + Event.Identify secondEvent = identifyEvent(user); + ep.sendEvent(secondEvent); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains(isIdentifyEvent(secondEvent, userJson))); + } + } + + @Test + public void probeFailureOpensLongerWindow() throws Exception { + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + + // The probe after the first window fails again; the second window is doubled, so a flush + // that would be past a first window is still inside the second. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); // inside the 10-minute second window + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); + es.expectNoRequests(100); + + // Past the second window, delivery resumes. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + Event.Identify postWindowEvent = identifyEvent(user); + ep.sendEvent(postWindowEvent); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains(isIdentifyEvent(postWindowEvent, userJson))); + } + } + + @Test + public void shutdownDuringCooldownMakesNoDeliveryAttempt() throws Exception { + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + + // The shutdown flush is in-window, so the buffered event is discarded without a network + // attempt. (Closing twice is safe, so try-with-resources can close again after this.) + ep.sendEvent(identifyEvent(user)); + ep.close(); + assertEquals(0, es.receivedParams.size()); + } + } + + @Test + public void identifyOnlyInWindowDiscardForgetsTheContext() throws Exception { + MockEventSender es = new MockEventSender(); + RememberingDeduplicator deduplicator = new RememberingDeduplicator(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es).contextDeduplicator(deduplicator))) { + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + + // An identify event is a context introduction just as an index event is; when an in-window + // discard destroys it, the deduplicator must forget the context so a later payload can + // introduce it again. + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); + es.expectNoRequests(100); + assertEquals(1, deduplicator.getFlushCount()); + + // The post-window payload introduces the context afresh: a custom event for the context + // comes with an index event rather than referencing a context the service never received. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + Event.Custom customEvent = customEvent(user, "eventkey").build(); + ep.sendEvent(customEvent); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains( + isIndexEvent(customEvent, userJson), + isCustomEvent(customEvent, userJson))); + } + } + + @Test + public void summaryOnlyInWindowDiscardKeepsAlreadyIntroducedContexts() throws Exception { + MockEventSender es = new MockEventSender(); + RememberingDeduplicator deduplicator = new RememberingDeduplicator(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es).contextDeduplicator(deduplicator))) { + // Introduce the context successfully before any window exists. + ep.sendEvent(featureEvent(user, FLAG_KEY).build()); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + + // A summary-only batch discarded in-window destroys no context introduction, so the + // deduplicator must be left alone. + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + ep.sendEvent(featureEvent(user, FLAG_KEY).build()); + ep.flushBlocking(); + es.expectNoRequests(100); + assertEquals(0, deduplicator.getFlushCount()); + + // The post-window payload therefore carries no redundant index event. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + ep.sendEvent(featureEvent(user, FLAG_KEY).build()); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains(isSummaryEvent())); + } + } + + @Test + public void inWindowFlushWithNothingBufferedDoesNothing() throws Exception { + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + + // Nothing is buffered, so there is no batch to dispose of and nothing to log. + ep.flushBlocking(); + es.expectNoRequests(100); + assertEquals(0, countLogLines(LDLogLevel.WARN, "Discarded")); + assertEquals(0, countLogLines(LDLogLevel.DEBUG, "Discarded")); + } + } + + @Test + public void discardLogsWarnOncePerWindowThenDebug() throws Exception { + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "will not be attempted")); + + for (int i = 0; i < 3; i++) { + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); + } + assertEquals(1, countLogLines(LDLogLevel.WARN, "Discarded")); + assertEquals(2, countLogLines(LDLogLevel.DEBUG, "Discarded")); + + // A second window gets its own single Warn, and the resumption is logged at Info. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); + assertEquals(2, countLogLines(LDLogLevel.WARN, "Discarded")); + + clock.advanceMillis(2 * COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + assertEquals(1, countLogLines(LDLogLevel.INFO, "resumed after cooldown")); + } + } + + @Test + public void diagnosticFailureOpensWindowForAnalyticsDelivery() throws Exception { + // The diagnostic delivery path reports unexpected failures into the cooldown, so an + // unexpected failure on a diagnostic payload opens the window too. (Diagnostic sends + // themselves are not gated by the window; see DefaultEventProcessorDiagnosticsTest.) + LDValue fakePlatformData = LDValue.buildObject().put("cats", 2).build(); + DiagnosticStore diagnosticStore = new DiagnosticStore( + new DiagnosticStore.SdkDiagnosticParams(SDK_KEY, "fake-sdk", "1.2.3", "fake-platform", + fakePlatformData, null, null)); + MockEventSender es = new MockEventSender(); + es.result = EventSender.Result.unexpectedFailure(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es).diagnosticStore(diagnosticStore))) { + // The startup diagnostic event fails with the unexpected error. + es.awaitDiagnostic(); + // The diagnostic send runs on the shared executor, which is single-threaded in these + // tests, so a no-op task is a barrier for its response handling. + sharedExecutor.submit(() -> {}).get(); + + // Analytics delivery is now in-window. + es.result = EventSender.Result.success(null); + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); + es.expectNoRequests(100); + } + } + + @Test + public void diagnosticSuccessesDoNotResetTheLadderOrEndTheWindow() throws Exception { + // A diagnostic success is not the probe: only analytics deliveries prove the service is + // accepting analytics payloads again, so diagnostic successes neither end a window, nor log a + // resumption, nor count toward the reset threshold. + LDValue fakePlatformData = LDValue.buildObject().put("cats", 2).build(); + DiagnosticStore diagnosticStore = new DiagnosticStore( + new DiagnosticStore.SdkDiagnosticParams(SDK_KEY, "fake-sdk", "1.2.3", "fake-platform", + fakePlatformData, null, null)); + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es).diagnosticStore(diagnosticStore))) { + es.awaitDiagnostic(); // the startup diagnostic event succeeds + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "the next 300 seconds")); + + // The window elapses; two diagnostic successes land before any analytics delivery. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + for (int i = 0; i < 2; i++) { + ep.postDiagnostic(); + es.awaitDiagnostic(); + sharedExecutor.submit(() -> {}).get(); // barrier for the diagnostic response handling + } + assertEquals(0, countLogLines(LDLogLevel.INFO, "resumed after cooldown")); + + // The ladder did not reset: the next unexpected analytics failure opens the doubled window. + es.result = EventSender.Result.unexpectedFailure(); + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "the next 600 seconds")); + } + } + + @Test + public void senderExceptionBreaksTheSuccessStreak() throws Exception { + // A delivery that dies in an exception was not delivered, so it must restart the + // reset-threshold measurement like any other failure (without opening a window). + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "the next 300 seconds")); + + // The probe succeeds: one credited success. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + + // The next delivery throws inside the sender. + es.fakeError = new RuntimeException("sorry"); + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + es.fakeError = null; + + // One more success: if the exception had been dropped, this would be the second + // consecutive success and the ladder would reset. + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + + es.result = EventSender.Result.unexpectedFailure(); + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "the next 600 seconds")); + } + } + + @Test + public void inWindowDiscardReportsZeroEventsInLastBatch() throws Exception { + LDValue fakePlatformData = LDValue.buildObject().put("cats", 2).build(); + DiagnosticStore diagnosticStore = new DiagnosticStore( + new DiagnosticStore.SdkDiagnosticParams(SDK_KEY, "fake-sdk", "1.2.3", "fake-platform", + fakePlatformData, null, null)); + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es).diagnosticStore(diagnosticStore))) { + es.awaitDiagnostic(); // the startup diagnostic event + + // A delivered batch records its size. + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); // discarded in-window + + // No batch reached the service, so the batch-size stat must not report one. + ep.postDiagnostic(); + CapturedPayload diagnostic = es.awaitDiagnostic(); + assertEquals(0, LDValue.parse(diagnostic.data).get("eventsInLastBatch").intValue()); + } + } + + @Test(timeout = 15000) + public void nullResultFromSenderDoesNotStopDeliveryOrClose() throws Exception { + // A broken custom EventSender returning null must not kill the delivery worker: delivery + // continues and close() still completes. + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + es.result = null; + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); // would hang here if the worker died + + es.result = EventSender.Result.success(null); + Event.Identify secondEvent = identifyEvent(user); + ep.sendEvent(secondEvent); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains(isIdentifyEvent(secondEvent, userJson))); + } + } + + @Test(timeout = 15000) + public void diagnosticStraddlingTheWindowEndCannotStepTheLadder() throws Exception { + // A diagnostic dispatched inside the window gets no live token, so even if it is still in + // flight when the window elapses, its unexpected failure cannot step the ladder or displace + // the analytics probe. + LDValue fakePlatformData = LDValue.buildObject().put("cats", 2).build(); + DiagnosticStore diagnosticStore = new DiagnosticStore( + new DiagnosticStore.SdkDiagnosticParams(SDK_KEY, "fake-sdk", "1.2.3", "fake-platform", + fakePlatformData, null, null)); + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es).diagnosticStore(diagnosticStore))) { + es.awaitDiagnostic(); // the startup diagnostic event succeeds + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "the next 300 seconds")); + + Object gate = new Object(); + CountDownLatch inFlight = new CountDownLatch(1); + es.receivedCounter = inFlight; + es.waitSignal = gate; + es.result = EventSender.Result.unexpectedFailure(); + ep.postDiagnostic(); + assertTrue(inFlight.await(5, TimeUnit.SECONDS)); // the diagnostic send is parked in-flight + es.awaitDiagnostic(); // consume its capture, which was recorded before it parked + + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); // the window elapses meanwhile + synchronized (gate) { + gate.notifyAll(); + } + es.waitSignal = null; + sharedExecutor.submit(() -> {}).get(); // the diagnostic outcome has now been handled + + // The analytics probe's failure steps the ladder exactly once, to 10 minutes: had the + // straddling diagnostic stepped it, this would be 20. + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "the next 600 seconds")); + assertEquals(0, countLogLines(LDLogLevel.ERROR, "the next 1200 seconds")); + } + } + + @Test(timeout = 15000) + public void simultaneousFailuresAcrossAllWorkersStepTheLadderOnce() throws Exception { + // Five deliveries are held in flight across the full worker pool and all fail unexpectedly at + // once: exactly one window opens, and the incident steps the ladder exactly once. + MockEventSender es = new MockEventSender(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es))) { + Object gate = new Object(); + es.waitSignal = gate; + es.result = EventSender.Result.unexpectedFailure(); + for (int i = 0; i < EventsConfiguration.DEFAULT_EVENT_SENDING_THREAD_POOL_SIZE; i++) { + CountDownLatch inFlight = new CountDownLatch(1); + es.receivedCounter = inFlight; + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + assertTrue(inFlight.await(5, TimeUnit.SECONDS)); + } + synchronized (gate) { + gate.notifyAll(); + } + es.waitSignal = null; + ep.waitUntilInactive(); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "the next 300 seconds")); + assertEquals(0, countLogLines(LDLogLevel.ERROR, "the next 600 seconds")); + + // The probe's failure steps once, to 10 minutes. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + ep.waitUntilInactive(); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "the next 600 seconds")); + } + } + + @Test + public void indexEventInWindowDiscardForgetsTheContext() throws Exception { + // A custom event for a never-introduced context puts an index event in the batch; the + // in-window discard destroys that introduction, so the deduplicator must forget it. + MockEventSender es = new MockEventSender(); + RememberingDeduplicator deduplicator = new RememberingDeduplicator(); + try (DefaultEventProcessor ep = makeProcessorWithClock(baseConfig(es).contextDeduplicator(deduplicator))) { + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + + LDContext otherContext = LDContext.create("otherkey"); + ep.sendEvent(customEvent(otherContext, "eventkey").build()); + ep.flushBlocking(); + es.expectNoRequests(100); + assertEquals(1, deduplicator.getFlushCount()); + } + } +} diff --git a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorDiagnosticsTest.java b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorDiagnosticsTest.java index 3e95ac6e..04280c3b 100644 --- a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorDiagnosticsTest.java +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorDiagnosticsTest.java @@ -181,22 +181,21 @@ private EventsConfigurationBuilder makeEventsConfigurationWithBriefDiagnosticInt } @Test - public void diagnosticEventsStopAfter401Error() throws Exception { - // This is easier to test with a mock component than it would be in LDClientEndToEndTest, because - // we don't have to worry about the latency of a real HTTP request which could allow the periodic - // task to fire again before we received a response. In real life, that wouldn't matter because - // the minimum diagnostic interval is so long, but in a test we need to be able to use a short - // interval. + public void diagnosticEventsContinueAfter401Error() throws Exception { + // Diagnostic event delivery is not governed by the delivery cooldown: the unexpected error + // opens a cooldown window, which suppresses analytics payloads, but periodic diagnostic + // events keep being sent. MockEventSender es = new MockEventSender(); - es.result = new EventSender.Result(false, true, null); // mustShutdown=true; this is what would be returned for a 401 error + es.result = EventSender.Result.unexpectedFailure(); // what would be returned for a 401 error EventsConfigurationBuilder eventsConfig = makeEventsConfigurationWithBriefDiagnosticInterval(es); - + try (DefaultEventProcessor ep = makeEventProcessor(eventsConfig.diagnosticStore(diagnosticStore))) { - // Ignore the initial diagnostic event + // The initial diagnostic event fails with the 401... es.awaitDiagnostic(); - es.expectNoRequests(100); + // ...and the periodic diagnostic task keeps firing anyway. + es.awaitDiagnostic(); } } diff --git a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorOutputTest.java b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorOutputTest.java index 78c4a19e..4c592ff3 100644 --- a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorOutputTest.java +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorOutputTest.java @@ -248,7 +248,7 @@ public void debugModeExpiresBasedOnClientTimeIfClientTimeIsLaterThanServerTime() // Pick a server time that is somewhat behind the client time long serverTime = System.currentTimeMillis() - 20000; - es.result = new EventSender.Result(true, false, new Date(serverTime)); + es.result = EventSender.Result.success(new Date(serverTime)); long debugUntil = serverTime + 1000; Event.FeatureRequest fe = featureEvent(user, FLAG_KEY).debugEventsUntilDate(debugUntil).build(); @@ -261,7 +261,7 @@ public void debugModeExpiresBasedOnClientTimeIfClientTimeIsLaterThanServerTime() ep.waitUntilInactive(); // ensure the flush worker has finished processing the response (setting lastKnownPastTime) es.receivedParams.clear(); - es.result = new EventSender.Result(true, false, null); + es.result = EventSender.Result.success(null); // Now send an event with debug mode on, with a "debug until" time that is further in // the future than the server time, but in the past compared to the client. @@ -280,7 +280,7 @@ public void debugModeExpiresBasedOnServerTimeIfServerTimeIsLaterThanClientTime() // Pick a server time that is somewhat ahead of the client time long serverTime = System.currentTimeMillis() + 20000; - es.result = new EventSender.Result(true, false, new Date(serverTime)); + es.result = EventSender.Result.success(new Date(serverTime)); long debugUntil = serverTime - 1000; Event.FeatureRequest fe = featureEvent(user, FLAG_KEY).debugEventsUntilDate(debugUntil).build(); @@ -293,7 +293,7 @@ public void debugModeExpiresBasedOnServerTimeIfServerTimeIsLaterThanClientTime() ep.waitUntilInactive(); // ensure the flush worker has finished processing the response (setting lastKnownPastTime) es.receivedParams.clear(); - es.result = new EventSender.Result(true, false, null); + es.result = EventSender.Result.success(null); // Now send an event with debug mode on, with a "debug until" time that is further in // the future than the client time, but in the past compared to the server. diff --git a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorTest.java b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorTest.java index ab2fda81..4ea39277 100644 --- a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorTest.java +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorTest.java @@ -228,18 +228,18 @@ public void eventCapacityDoesNotPreventSummaryEventFromBeingSent() throws Except } @Test - public void noMoreEventsAreProcessedAfterUnrecoverableError() throws Exception { + public void eventsAreNotSentWhileDeliveryIsCoolingDownAfterUnexpectedError() throws Exception { + // Delivery suppression after an unexpected error is a bounded cooldown window, not a + // permanent stop; the window-crossing behavior is covered in DefaultEventProcessorCooldownTest. MockEventSender es = new MockEventSender(); - es.result = new EventSender.Result(false, true, null); // mustShutdown == true - + es.result = EventSender.Result.unexpectedFailure(); + try (DefaultEventProcessor ep = makeEventProcessor(baseConfig(es))) { ep.sendEvent(identifyEvent(user)); ep.flushAsync(); es.awaitRequest(); - - // allow a little time for the event processor to pass the "must shut down" signal back from the sender - Thread.sleep(50); - + ep.waitUntilInactive(); // the failure result has now been handled + ep.sendEvent(identifyEvent(user)); ep.flushAsync(); es.expectNoRequests(100); diff --git a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventSenderTest.java b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventSenderTest.java index ce9b2351..dc70e9d1 100644 --- a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventSenderTest.java +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventSenderTest.java @@ -396,6 +396,8 @@ private void testUnrecoverableHttpError(int status) throws Exception { EventSender.Result result = es.sendAnalyticsEvents(FAKE_DATA_BYTES, 1, server.getUri()); assertFalse(result.isSuccess()); + assertTrue(result.isUnexpectedFailure()); + // The legacy flag is still written for consumers that have not migrated. assertTrue(result.isMustShutDown()); } @@ -406,6 +408,34 @@ private void testUnrecoverableHttpError(int status) throws Exception { } } + @Test + public void transportErrorIsRecoverableAndIsRetriedOnce() throws Exception { + // A transport-level failure (no HTTP response) is never classified as unexpected, whatever + // its cause: the request is retried once and the failure leaves the delivery schedule alone. + URI unreachable; + try (java.net.ServerSocket socket = new java.net.ServerSocket(0)) { + unreachable = URI.create("http://127.0.0.1:" + socket.getLocalPort()); + } // closing the socket leaves the port with no listener + + long startTime = System.currentTimeMillis(); + try (EventSender es = makeEventSender()) { + EventSender.Result result = es.sendAnalyticsEvents(FAKE_DATA_BYTES, 1, unreachable); + + assertFalse(result.isSuccess()); + assertFalse(result.isUnexpectedFailure()); + assertFalse(result.isMustShutDown()); + } + // The one retry implies the sender waited out the retry delay between the two attempts. + assertTrue(System.currentTimeMillis() - startTime >= BRIEF_RETRY_DELAY_MILLIS); + int retryLines = 0; + for (com.launchdarkly.logging.LogCapture.Message message : logCapture.getMessages()) { + if (message.getText().contains("Will retry posting")) { + retryLines++; + } + } + assertEquals(1, retryLines); + } + private void testRecoverableHttpError(int status) throws Exception { Handler errorResponse = Handlers.status(status); Handler errorsThenSuccess = Handlers.sequential(errorResponse, errorResponse, eventsSuccessResponse()); @@ -416,6 +446,7 @@ private void testRecoverableHttpError(int status) throws Exception { EventSender.Result result = es.sendAnalyticsEvents(FAKE_DATA_BYTES, 1, server.getUri()); assertFalse(result.isSuccess()); + assertFalse(result.isUnexpectedFailure()); assertFalse(result.isMustShutDown()); } diff --git a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java new file mode 100644 index 00000000..8e2cd973 --- /dev/null +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java @@ -0,0 +1,324 @@ +package com.launchdarkly.sdk.internal.events; + +import com.launchdarkly.sdk.internal.BaseTest; +import com.launchdarkly.sdk.internal.events.DeliveryCooldownTracker.WindowToken; + +import org.junit.Test; + +import static com.launchdarkly.sdk.internal.events.DeliveryCooldownTracker.COOLDOWN_INITIAL_WINDOW_MILLIS; +import static com.launchdarkly.sdk.internal.events.DeliveryCooldownTracker.COOLDOWN_MAX_WINDOW_MILLIS; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +@SuppressWarnings("javadoc") +public class DeliveryCooldownTrackerTest extends BaseTest { + static final class FakeNanoClock { + private long nowNanos = 100_000_000_000L; + + synchronized long now() { + return nowNanos; + } + + synchronized void advanceMillis(long millis) { + nowNanos += millis * 1_000_000L; + } + } + + private final FakeNanoClock clock = new FakeNanoClock(); + private final DeliveryCooldownTracker tracker = new DeliveryCooldownTracker(clock::now); + + // Records an unexpected failure and then advances the clock past the window it opened, + // returning the window length. + private long failPastWindow() { + long windowMillis = tracker.recordUnexpectedFailure(tracker.beginDelivery()); + assertTrue(windowMillis > 0); + clock.advanceMillis(windowMillis + 1000); + return windowMillis; + } + + @Test + public void startsOutsideCooldown() { + assertFalse(tracker.inCooldown()); + assertEquals(0, tracker.remainingMillis()); + } + + @Test + public void windowDoublesPerFailedProbeUpToCeiling() { + long[] expected = { + 5 * 60_000L, 10 * 60_000L, 20 * 60_000L, 40 * 60_000L, + 60 * 60_000L, 60 * 60_000L, // clamped at the ceiling + }; + for (long want : expected) { + assertEquals(want, failPastWindow()); + } + } + + @Test + public void inWindowResultFromInFlightDeliveryDoesNotLengthenWindow() { + WindowToken preWindowToken = tracker.beginDelivery(); + assertEquals(COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(preWindowToken)); + + // A result arriving while the window is open is from a delivery that was already in flight; + // its token is stale, so it does not extend the window or advance the doubling sequence. + clock.advanceMillis(60_000); + assertEquals(0, tracker.recordUnexpectedFailure(preWindowToken)); + assertEquals(COOLDOWN_INITIAL_WINDOW_MILLIS - 60_000, tracker.remainingMillis()); + + // The next failed probe doubles once, to 10 minutes, not twice. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS - 60_000 + 1000); + assertEquals(2 * COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + } + + @Test + public void resetsAfterTwoConsecutiveSuccesses() { + failPastWindow(); + tracker.recordSuccess(tracker.beginDelivery()); + tracker.recordSuccess(tracker.beginDelivery()); + + // The retry state has reset, so a new unexpected failure starts over at the initial delay. + assertEquals(COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + } + + @Test + public void singleSuccessDoesNotReset() { + failPastWindow(); + tracker.recordSuccess(tracker.beginDelivery()); + + assertEquals(2 * COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + } + + @Test + public void ordinaryFailureBreaksSuccessStreakWithoutOpeningWindow() { + tracker.recordFailure(tracker.beginDelivery()); + assertFalse(tracker.inCooldown()); + + failPastWindow(); + tracker.recordSuccess(tracker.beginDelivery()); + tracker.recordFailure(tracker.beginDelivery()); // the two successes around it are not consecutive + tracker.recordSuccess(tracker.beginDelivery()); + + assertEquals(2 * COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + } + + @Test + public void remainingMillisCountsDown() { + assertEquals(COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + + clock.advanceMillis(2 * 60_000); + assertEquals(COOLDOWN_INITIAL_WINDOW_MILLIS - 2 * 60_000, tracker.remainingMillis()); + assertTrue(tracker.inCooldown()); + + clock.advanceMillis(3 * 60_000 + 1000); + assertEquals(0, tracker.remainingMillis()); + assertFalse(tracker.inCooldown()); + } + + @Test + public void reportsResumeOnlyOnFirstPostWindowSuccess() { + assertFalse(tracker.recordSuccess(tracker.beginDelivery())); // no window has ever opened + + assertTrue(tracker.recordUnexpectedFailure(tracker.beginDelivery()) > 0); + assertFalse(tracker.recordSuccess(tracker.beginDelivery())); // defensive: in-window success changes nothing + + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + assertTrue(tracker.recordSuccess(tracker.beginDelivery())); // the probe ends the window + assertFalse(tracker.recordSuccess(tracker.beginDelivery())); // only reported once + } + + @Test + public void noSpuriousResumeAfterRecoverableFailure() { + failPastWindow(); + tracker.recordFailure(tracker.beginDelivery()); // the probe fails recoverably; the window is over + + // Much later, an ordinary success must not be reported as a cooldown resumption. + clock.advanceMillis(1000 * 60_000); + assertFalse(tracker.recordSuccess(tracker.beginDelivery())); + } + + @Test + public void boundaryRoundSiblingsAreStaleAfterProbeSuccess() { + // Window opens; after it elapses, three deliveries are dispatched concurrently in the + // boundary round, all sharing the same token. + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + WindowToken boundaryToken = tracker.beginDelivery(); + + // The probe's success ends the round; the sibling success must not combine with it into a + // reset, and the sibling failure must not open a window (the next round's failure does). + assertTrue(tracker.recordSuccess(boundaryToken)); + assertFalse(tracker.recordSuccess(boundaryToken)); + assertEquals(0, tracker.recordUnexpectedFailure(boundaryToken)); + assertFalse(tracker.inCooldown()); + + // The retry state was not reset (one credited success, not two), so the next failed delivery + // opens a doubled window. + assertEquals(2 * COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + } + + @Test + public void boundaryRoundSiblingsAreStaleAfterRecoverableProbeFailure() { + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + WindowToken boundaryToken = tracker.beginDelivery(); + + // The probe fails recoverably, which also ends the boundary round: the two sibling successes + // are stale and must not combine into a reset. + tracker.recordFailure(boundaryToken); + assertFalse(tracker.recordSuccess(boundaryToken)); + assertFalse(tracker.recordSuccess(boundaryToken)); + + assertEquals(2 * COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + } + + @Test + public void staleOrdinaryFailureCannotCancelTheNextWindow() { + // A delivery is in flight when a concurrent delivery's failure opens the window. + WindowToken inFlight = tracker.beginDelivery(); + tracker.recordUnexpectedFailure(inFlight); + assertTrue(tracker.inCooldown()); + + // The window elapses and a probe is dispatched; then the straggler's recoverable failure + // lands. It is stale: were it allowed to end the already-elapsed window, it would close the + // boundary round and stale the probe, so the probe's failure would open no window at all. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + WindowToken probe = tracker.beginDelivery(); + tracker.recordFailure(inFlight); + + assertEquals(2 * COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(probe)); + assertTrue(tracker.inCooldown()); + } + + @Test + public void staleOrdinaryFailureDoesNotBreakSuccessStreak() { + failPastWindow(); + assertTrue(tracker.recordSuccess(tracker.beginDelivery())); // probe success ends the window + WindowToken staleToken = tracker.beginDelivery(); + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + clock.advanceMillis(2 * COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + assertTrue(tracker.recordSuccess(tracker.beginDelivery())); + tracker.recordFailure(staleToken); // stale; must not reset the streak + tracker.recordSuccess(tracker.beginDelivery()); // second consecutive success resets the ladder + + assertEquals(COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + } + + @Test + public void ordinaryFailureOutsideWindowDoesNotStaleConcurrentResults() { + // During healthy operation (no window has ever opened), an ordinary failure must not close a + // boundary round that does not exist: a concurrent unexpected failure in the same round must + // still open a window. + WindowToken roundToken = tracker.beginDelivery(); + tracker.recordFailure(roundToken); + assertTrue(tracker.recordUnexpectedFailure(roundToken) > 0); + assertTrue(tracker.inCooldown()); + } + + @Test + public void nullTokenIsNeverCurrent() { + // An unstamped payload must never pass for a live delivery, whether or not a window has + // ever existed. + assertFalse(tracker.recordSuccess(null)); + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + assertFalse(tracker.recordSuccess(null)); + + // The boundary round is still open for the real probe. + assertTrue(tracker.recordSuccess(tracker.beginDelivery())); + } + + @Test + public void defaultClockOpensWindow() { + // Production passes no clock; the default real-time clock path must work end to end. + DeliveryCooldownTracker realTimeTracker = new DeliveryCooldownTracker(null); + assertEquals(COOLDOWN_INITIAL_WINDOW_MILLIS, realTimeTracker.recordUnexpectedFailure(realTimeTracker.beginDelivery())); + assertTrue(realTimeTracker.inCooldown()); + assertTrue(realTimeTracker.remainingMillis() > COOLDOWN_INITIAL_WINDOW_MILLIS - 60_000); + } + + @Test + public void windowBoundaryInstantIsOutsideCooldown() { + long windowMillis = tracker.recordUnexpectedFailure(tracker.beginDelivery()); + clock.advanceMillis(windowMillis); + assertFalse(tracker.inCooldown()); + assertEquals(0, tracker.remainingMillis()); + } + + @Test + public void ladderIsClampedAtCeilingIndefinitely() { + // Runs well past the shift guard's exponent bound: without the bound, the shift distance + // wraps at 64 and the window would collapse back to the initial delay. + for (int i = 0; i < 70; i++) { + long expected = i < 4 ? COOLDOWN_INITIAL_WINDOW_MILLIS << i : COOLDOWN_MAX_WINDOW_MILLIS; + long windowMillis = tracker.recordUnexpectedFailure(tracker.beginDelivery()); + assertEquals("window for failed probe #" + (i + 1), expected, windowMillis); + clock.advanceMillis(windowMillis + 1000); + } + } + + @Test + public void inWindowOutcomesWithCurrentTokenUpdateNothing() { + // Diagnostic deliveries are not gated by the window, so an outcome can arrive in-window with + // a current token; no such outcome may move delivery state. + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + long remainingBefore = tracker.remainingMillis(); + WindowToken inWindowToken = tracker.beginDelivery(); + + assertFalse(tracker.recordSuccess(inWindowToken)); + tracker.recordFailure(inWindowToken); + assertEquals(0, tracker.recordUnexpectedFailure(inWindowToken)); + assertEquals(remainingBefore, tracker.remainingMillis()); + + // The state is untouched: the real probe still reports the resumption, and the ladder has + // stepped exactly once. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + assertTrue(tracker.recordSuccess(tracker.beginDelivery())); + tracker.recordSuccess(tracker.beginDelivery()); + assertEquals(COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + } + + @Test + public void discardWarningIsTakenOncePerWindow() { + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + assertTrue(tracker.takeDiscardWarning()); + assertFalse(tracker.takeDiscardWarning()); + + // A new window gets its own notice. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + assertTrue(tracker.takeDiscardWarning()); + assertFalse(tracker.takeDiscardWarning()); + } + + @Test + public void inWindowFailureDoesNotRearmTheDiscardWarning() { + // An in-window unexpected failure (an ungated diagnostic delivery) opens no window, so it + // must not hand the current window a second prominent notice. + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + assertTrue(tracker.takeDiscardWarning()); + assertEquals(0, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + assertFalse(tracker.takeDiscardWarning()); + } + + @Test + public void windowWithNoDiscardsStillArmsTheNextWindowsWarning() { + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + // Nothing is flushed during this window, so the notice is never taken. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + assertTrue(tracker.takeDiscardWarning()); + } + + @Test + public void gateIssuesTokenOnlyOutsideWindow() { + assertNotNull(tracker.beginDeliveryUnlessInCooldown()); + + tracker.recordUnexpectedFailure(tracker.beginDelivery()); + assertNull(tracker.beginDeliveryUnlessInCooldown()); + + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + assertNotNull(tracker.beginDeliveryUnlessInCooldown()); + } +} From f40324a2ca974c48eb0a5d6fce158d26da3b487e Mon Sep 17 00:00:00 2001 From: Todd Anderson Date: Fri, 6 Nov 2026 23:23:52 -0500 Subject: [PATCH 3/3] docs: correct the tracker's in-cooldown guard comments Diagnostics obtain their delivery token through the same atomic cooldown-check-plus-issue call that analytics flushes use, so a delivery dispatched inside a window carries no token at all and every pre-window token goes stale when the window opens. All three in-cooldown guards are therefore uniformly defensive; one of them previously claimed to be reachable from in-window diagnostic failures. --- .../sdk/internal/events/DeliveryCooldownTracker.java | 12 +++++------- .../internal/events/DeliveryCooldownTrackerTest.java | 5 +++-- 2 files changed, 8 insertions(+), 9 deletions(-) diff --git a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java index e4a7e6c1..f4df1eb3 100644 --- a/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java +++ b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java @@ -147,10 +147,10 @@ synchronized boolean recordSuccess(WindowToken token) { return false; } if (inCooldown()) { - // Defensive: analytics tokens are issued atomically with the cooldown check and go stale - // when a window opens, and diagnostic successes do not reach the tracker, so no current - // caller gets here. The invariant stands on its own: an outcome arriving while a window is - // open must never move delivery state. + // Defensive: tokens are issued atomically with the cooldown check (a diagnostic dispatched + // in-window gets no token at all), and a pre-window token goes stale when the window opens, + // so no current caller gets here. The invariant stands on its own: an outcome arriving + // while a window is open must never move delivery state. return false; } boolean resumed = false; @@ -216,9 +216,7 @@ synchronized long recordUnexpectedFailure(WindowToken token) { return 0; } if (inCooldown()) { - // Load-bearing: diagnostic deliveries are not gated by the window, so an in-window - // diagnostic failure arrives with a current token, and it must not lengthen the window or - // step the ladder again for the same incident. + // Defensive; see recordSuccess. return 0; } consecutiveSuccesses = 0; diff --git a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java index 8e2cd973..b2e5fcc6 100644 --- a/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java @@ -260,8 +260,9 @@ public void ladderIsClampedAtCeilingIndefinitely() { @Test public void inWindowOutcomesWithCurrentTokenUpdateNothing() { - // Diagnostic deliveries are not gated by the window, so an outcome can arrive in-window with - // a current token; no such outcome may move delivery state. + // No production caller currently presents a current token for an in-window outcome (tokens + // are issued atomically with the cooldown check), but the tracker's own contract is that no + // such outcome may move delivery state. tracker.recordUnexpectedFailure(tracker.beginDelivery()); long remainingBefore = tracker.remainingMillis(); WindowToken inWindowToken = tracker.beginDelivery();