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 29cdcdf6..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 @@ -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 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 compressionResult = compressData(data); RequestBody body = RequestBody.create(compressionResult.data, JSON_CONTENT_TYPE); - boolean mustShutDown = false; + boolean unexpected = false; if (compressionResult.wasCompressed) { headersBuilder.add("Content-Encoding", "gzip"); @@ -207,27 +208,24 @@ private Result sendEventData(boolean isDiagnostic, byte[] data, int eventCount, logger.debug("{} delivery took {} ms, response status {}", description, endTime - startTime, response.code()); if (response.isSuccessful()) { - return new Result(true, false, parseResponseDate(response)); + return Result.success(parseResponseDate(response)); } String errorDesc = httpErrorDescription(response.code()); - boolean recoverable = checkIfErrorIsRecoverableAndLog( - logger, - errorDesc, - errorContext, - response.code(), - nextActionMessage - ); - if (!recoverable) { - mustShutDown = true; + if (classifyHttpFailure(response.code()) == FailureClass.UNEXPECTED) { + logger.error("Error {} (the payload will not be retried): {}", errorContext, errorDesc); + unexpected = true; break; } + logger.warn("Error {} ({}): {}", errorContext, nextActionMessage, errorDesc); } catch (IOException e) { - checkIfErrorIsRecoverableAndLog(logger, e.toString(), errorContext, 0, nextActionMessage); + // Transport-level failures are always worth one retry here, whatever their cause; the + // request may simply have hit a transient network problem. + logger.warn("Error {} ({}): {}", errorContext, nextActionMessage, e.toString()); } } - return new Result(false, mustShutDown, null); + return unexpected ? Result.unexpectedFailure() : Result.failure(); } private final Date parseResponseDate(Response response) { 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..f4df1eb3 --- /dev/null +++ b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java @@ -0,0 +1,241 @@ +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: 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; + 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()) { + // Defensive; see recordSuccess. + 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..b2e5fcc6 --- /dev/null +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java @@ -0,0 +1,325 @@ +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() { + // 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(); + + 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()); + } +}