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..428e80de 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,79 @@ 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.isUnexpectedFailure()) { + noteUnexpectedFailure(token); + } else if (result.isSuccess()) { + if (cooldown.recordSuccess(token)) { + logger.info("Event delivery resumed after cooldown"); + } + } else { + cooldown.recordFailure(token); + } + } + + // The cooldown governs analytics delivery only (EVTPROC section 1.8 scopes it to flush + // payloads). A diagnostic delivery never touches cooldown state. + private void handleDiagnosticResponse(EventSender.Result result) { if (result.getTimeFromServer() != null) { lastKnownPastTime.set(result.getTimeFromServer().getTime()); } - if (result.isMustShutDown()) { - disabled.set(true); + } + + private void noteUnexpectedFailure(DeliveryCooldownTracker.WindowToken token) { + long windowMillis = cooldown.recordUnexpectedFailure(token); + if (windowMillis > 0) { + // The refused batch's index and identify events were lost with it, so forget every + // context key + if (contextDeduplicator != null) { + inbox.offer(new EventProcessorMessage(MessageType.FLUSH_USERS, null, false)); + } + logger.error("Event delivery failed with an unexpected error;" + + " event delivery will not be attempted for the next {} seconds", windowMillis / 1000); } } @@ -652,7 +732,7 @@ public void run() { writer.flush(); EventSender.Result result = eventsConfig.eventSender.sendDiagnosticEvent( buffer.toByteArray(), eventsConfig.eventsUri); - handleResponse(result); + handleDiagnosticResponse(result); if (diagnosticEvent.initEvent) { didSendInitEvent.set(true); } @@ -716,10 +796,10 @@ long getAndClearDroppedCount() { return res; } - FlushPayload getPayload() { + FlushPayload getPayload(DeliveryCooldownTracker.WindowToken token) { Event[] eventsOut = events.toArray(new Event[events.size()]); List summaries = summarizer.getSummariesAndReset(); - return new FlushPayload(eventsOut, summaries); + return new FlushPayload(eventsOut, summaries, token); } void clear() { @@ -731,15 +811,17 @@ void clear() { private static final class FlushPayload { final Event[] events; final List 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 { @@ -781,23 +863,45 @@ public void run() { continue; } try { - ByteArrayOutputStream buffer = new ByteArrayOutputStream(INITIAL_OUTPUT_BUFFER_SIZE); - Writer writer = new BufferedWriter(new OutputStreamWriter(buffer, Charset.forName("UTF-8")), INITIAL_OUTPUT_BUFFER_SIZE); - int outputEventCount = formatter.writeOutputEvents(payload.events, payload.summaries, writer); - writer.flush(); - EventSender.Result result = eventsConfig.eventSender.sendAnalyticsEvents( - buffer.toByteArray(), - outputEventCount, - eventsConfig.eventsUri - ); - responseListener.handleResponse(result); - } catch (Exception e) { - logger.error("Unexpected error in event processor: {}", LogValues.exceptionSummary(e)); - logger.debug(LogValues.exceptionTrace(e)); - } - synchronized (activeFlushWorkersCount) { - activeFlushWorkersCount.decrementAndGet(); - activeFlushWorkersCount.notifyAll(); + EventSender.Result result; + try { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(INITIAL_OUTPUT_BUFFER_SIZE); + Writer writer = new BufferedWriter(new OutputStreamWriter(buffer, Charset.forName("UTF-8")), INITIAL_OUTPUT_BUFFER_SIZE); + int outputEventCount = formatter.writeOutputEvents(payload.events, payload.summaries, writer); + writer.flush(); + result = eventsConfig.eventSender.sendAnalyticsEvents( + buffer.toByteArray(), + outputEventCount, + eventsConfig.eventsUri + ); + } catch (Exception e) { + logger.error("Unexpected error in event processor: {}", LogValues.exceptionSummary(e)); + logger.debug(LogValues.exceptionTrace(e)); + // The batch was not delivered, and the delivery schedule must know that: an exception + // here counts as an ordinary failure, which breaks the consecutive-success streak + // without opening a cooldown window. + result = EventSender.Result.failure(); + } + if (result == null) { + // A custom EventSender that returns no result reported no delivery, so treat it as an + // ordinary failure. + result = EventSender.Result.failure(); + } + try { + responseListener.handleResponse(result, payload.token); + } catch (Exception e) { + // Any remaining throw from response handling must not kill this worker thread. + logger.error("Unexpected error in event processor: {}", LogValues.exceptionSummary(e)); + logger.debug(LogValues.exceptionTrace(e)); + } + } finally { + // The busy-worker count must always come back down, even if an Error (not an Exception) + // escaped the work above, or close() would wait forever in + // waitUntilAllFlushWorkersInactive. The Error still ends this worker thread. + synchronized (activeFlushWorkersCount) { + activeFlushWorkersCount.decrementAndGet(); + activeFlushWorkersCount.notifyAll(); + } } } } 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..4a148b6d 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,29 @@ private Result sendEventData(boolean isDiagnostic, byte[] data, int eventCount, logger.debug("{} delivery took {} ms, response status {}", description, endTime - startTime, response.code()); if (response.isSuccessful()) { - return new Result(true, false, parseResponseDate(response)); + return Result.success(parseResponseDate(response)); } String errorDesc = httpErrorDescription(response.code()); - boolean recoverable = checkIfErrorIsRecoverableAndLog( - logger, - errorDesc, - errorContext, - response.code(), - nextActionMessage - ); - if (!recoverable) { - mustShutDown = true; + if (response.code() == 413) { + logger.warn("Error {} (the payload was too large; it was dropped and not retried): {}", + errorContext, errorDesc); break; } + if (classifyHttpFailure(response.code()) == FailureClass.UNEXPECTED) { + logger.error("Error {} (the payload will not be retried): {}", errorContext, errorDesc); + unexpected = true; + break; + } + logger.warn("Error {} ({}): {}", errorContext, nextActionMessage, errorDesc); } catch (IOException e) { - checkIfErrorIsRecoverableAndLog(logger, e.toString(), errorContext, 0, nextActionMessage); + // Transport-level failures are always worth one retry here, whatever their cause; the + // request may simply have hit a transient network problem. + logger.warn("Error {} ({}): {}", errorContext, nextActionMessage, LogValues.exceptionSummary(e)); } } - return new Result(false, mustShutDown, null); + return unexpected ? Result.unexpectedFailure() : Result.failure(); } private final Date parseResponseDate(Response response) { 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..cd171cc7 --- /dev/null +++ b/lib/shared/internal/src/main/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTracker.java @@ -0,0 +1,235 @@ +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()) { + return false; // defensive: a current token cannot originate inside a window + } + boolean resumed = false; + if (windowEndNanos != null) { + windowEndNanos = null; + resumed = true; + // The probe ends the window: the tokens previously given out go stale. + 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, + * even on a stale token, but never opens or lengthens a window. Unstamped outcomes update + * nothing. + * + * @param token the token issued when this delivery was dispatched + */ + synchronized void recordFailure(WindowToken token) { + if (token == null) { + return; // unstamped: not the outcome of a real delivery + } + // A real failure breaks the streak regardless of its era: the reset threshold counts + // consecutive successful deliveries, and a failure between two successes makes them + // non-consecutive no matter which round dispatched it. + consecutiveSuccesses = 0; + if (!isCurrent(token)) { + return; + } + if (inCooldown()) { + return; // defensive: a current token cannot originate inside a window + } + if (windowEndNanos != null) { + // A probe that fails recoverably still ends the window: clear it so a later success is not + // misreported as the resumption. + windowEndNanos = null; + epoch++; + } + } + + /** + * Records a delivery failure classified as unexpected. It opens a window that doubles from + * {@link #COOLDOWN_INITIAL_WINDOW_MILLIS} up to {@link #COOLDOWN_MAX_WINDOW_MILLIS} per failed + * probe. A stale token still breaks the consecutive-success streak but never opens or steps a + * window, so one incident's concurrent failures step the ladder only once. Unstamped outcomes + * update nothing. + * + * @param token the token issued when this delivery was dispatched + * @return the length in milliseconds of the newly opened window, or zero if no window was opened + */ + synchronized long recordUnexpectedFailure(WindowToken token) { + if (token == null) { + return 0; // unstamped: not the outcome of a real delivery + } + // See recordFailure: a real failure breaks the streak regardless of its era. + consecutiveSuccesses = 0; + if (!isCurrent(token)) { + return 0; + } + if (inCooldown()) { + return 0; // defensive: a current token cannot originate inside a window + } + 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..74b322b4 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,74 @@ 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 or + * network error. Unlike {@link #unexpectedFailure()}, it does not cause event delivery to + * enter a cooldown window. + * + * @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 +116,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..0a1c1229 --- /dev/null +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DefaultEventProcessorCooldownTest.java @@ -0,0 +1,557 @@ +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. The context is + // introduced in-window (the window-open wipe cleared it), then discarded; the discard must + // forget it again so a later payload re-introduces it. + ep.sendEvent(identifyEvent(user)); + ep.flushBlocking(); + es.expectNoRequests(100); + + // 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 windowOpenForgetsEvenDeliveredIntroductions() 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(); + + // Opening a window forgets every context key, including ones whose index events were + // delivered: the dispatcher cannot see which contexts the refused batch carried, and a + // redundant index event after the window is harmless. + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + Event.FeatureRequest fe = featureEvent(user, FLAG_KEY).build(); + ep.sendEvent(fe); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains(isIndexEvent(fe, userJson), 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 diagnosticFailureDoesNotOpenACooldownWindow() throws Exception { + // The cooldown governs analytics delivery only (EVTPROC section 1.8 scopes it to flush + // payloads). An unexpected failure on a diagnostic payload does not open a window, so + // analytics delivery is unaffected. + 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(); + + // No window opened: the next analytics flush is delivered rather than discarded. + es.result = EventSender.Result.success(null); + Event.Identify postEvent = identifyEvent(user); + ep.sendEvent(postEvent); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains(isIdentifyEvent(postEvent, userJson))); + } + } + + @Test + public void diagnosticOutcomesDoNotAffectAnOpenCooldownWindow() throws Exception { + // Diagnostic outcomes never move cooldown state: with an analytics window open, a diagnostic + // failure does not step the ladder, and a diagnostic success after the window elapses does + // not end it or log a resumption. Only analytics deliveries are the 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 + // The await returns when the payload is captured, before the diagnostic task has read the + // result field; wait for the task to finish before changing the result. + sharedExecutor.submit(() -> {}).get(); + enterCooldown(ep, es, EventSender.Result.unexpectedFailure()); + assertEquals(1, countLogLines(LDLogLevel.ERROR, "the next 300 seconds")); + + // A diagnostic failure during the window must not step the ladder. + es.result = EventSender.Result.unexpectedFailure(); + ep.postDiagnostic(); + es.awaitDiagnostic(); + sharedExecutor.submit(() -> {}).get(); + + // The window elapses; a diagnostic success must not end it or log a resumption. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + es.result = EventSender.Result.success(null); + ep.postDiagnostic(); + es.awaitDiagnostic(); + sharedExecutor.submit(() -> {}).get(); + assertEquals(0, countLogLines(LDLogLevel.INFO, "resumed after cooldown")); + + // The next analytics failure steps to the doubled window: the diagnostic failure did not + // advance the ladder (else this would be 1200), and the diagnostic success did not reset it + // (else this would be 300). + 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 + // Barrier: let the startup diagnostic task finish before the first analytics flush. + sharedExecutor.submit(() -> {}).get(); + + // 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 + public void nullResultBreaksTheSuccessStreak() throws Exception { + // A null result reports no delivery, like an exception from the sender, so it must restart + // the reset-threshold measurement and keep the ladder from resetting across it. (Go's result + // is a value type whose zero value is already an ordinary failure, so this matches Go.) + 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 returns null. + es.result = null; + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + es.awaitAnalytics(); + ep.waitUntilInactive(); + + // One more success: if the null had been ignored, this would be the second consecutive + // success and the ladder would reset to the initial delay. + es.result = EventSender.Result.success(null); + 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(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()); + + // A custom event for a context not in the cache (the window-open wipe emptied it) carries + // an index event; the in-window discard must forget that context again. + ep.sendEvent(customEvent(user, "eventkey").build()); + ep.flushBlocking(); + es.expectNoRequests(100); + + // After the window, the context re-introduces rather than referencing one the service never + // received. + clock.advanceMillis(COOLDOWN_INITIAL_WINDOW_MILLIS + 1000); + Event.Custom postEvent = customEvent(user, "eventkey").build(); + ep.sendEvent(postEvent); + ep.flushAsync(); + assertThat(es.getEventsFromLastRequest(), contains( + isIndexEvent(postEvent, userJson), + isCustomEvent(postEvent, userJson))); + } + } +} 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..b0dd0de6 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); @@ -279,7 +279,33 @@ public void uncheckedExceptionFromEventSenderDoesNotStopWorkerThread() throws Ex es.awaitRequest(); } } - + + @Test(timeout = 15000) + public void errorFromEventSenderDoesNotWedgeClose() throws Exception { + // An Error (not an Exception) from the sender escapes the flush worker's catch blocks and + // ends that worker thread, but the busy-worker count must still be decremented so that + // close() does not wait on it forever in waitUntilAllFlushWorkersInactive. + EventSender throwingSender = new EventSender() { + @Override + public Result sendAnalyticsEvents(byte[] data, int eventCount, URI eventsBaseUri) { + throw new AssertionError("boom"); + } + + @Override + public Result sendDiagnosticEvent(byte[] data, URI eventsBaseUri) { + return Result.success(null); + } + + @Override + public void close() throws IOException {} + }; + + DefaultEventProcessor ep = makeEventProcessor(baseConfig(throwingSender)); + ep.sendEvent(identifyEvent(user)); + ep.flushAsync(); + ep.close(); // would block forever if the dead worker's busy count were never decremented + } + @SuppressWarnings("unchecked") @Test public void eventsAreKeptInBufferIfAllFlushWorkersAreBusy() throws Exception { 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..de5c688e 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 @@ -246,6 +246,29 @@ public void http403ErrorIsUnrecoverable() throws Exception { testUnrecoverableHttpError(403); } + @Test + public void http413ErrorDropsPayloadWithoutDisablingDelivery() throws Exception { + // A payload-too-large response is locally unrecoverable: the payload is dropped and not + // retried, but it is not an unexpected failure, so delivery is not disabled and no cooldown + // window opens (EVTPROC 1.7.3.1). + Handler errorResponse = Handlers.status(413); + + try (HttpServer server = HttpServer.start(errorResponse)) { + try (EventSender es = makeEventSender()) { + EventSender.Result result = es.sendAnalyticsEvents(FAKE_DATA_BYTES, 1, server.getUri()); + + assertFalse(result.isSuccess()); + assertFalse(result.isUnexpectedFailure()); + assertFalse(result.isMustShutDown()); + } + + server.getRecorder().requireRequest(); + + // 413 is not retried. + server.getRecorder().requireNoRequests(Duration.ofMillis(100)); + } + } + // Cannot test our retry logic for 408, because OkHttp insists on doing its own retry on 408 so that // we never actually see that response status. // @Test @@ -396,6 +419,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 +431,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 +469,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..665e0178 --- /dev/null +++ b/lib/shared/internal/src/test/java/com/launchdarkly/sdk/internal/events/DeliveryCooldownTrackerTest.java @@ -0,0 +1,346 @@ +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 staleFailureBetweenSuccessesPreventsLadderReset() { + // The probe and a sibling are dispatched together in the boundary round; the probe succeeds, + // staling the sibling, and then the sibling's real failure lands. Success, failure, success: + // the successes were not consecutive, so the ladder must not reset. + failPastWindow(); // one failed probe on the ladder + WindowToken boundaryToken = tracker.beginDelivery(); + assertTrue(tracker.recordSuccess(boundaryToken)); + tracker.recordFailure(boundaryToken); + tracker.recordSuccess(tracker.beginDelivery()); + + assertEquals(2 * COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(tracker.beginDelivery())); + } + + @Test + public void staleUnexpectedFailureBetweenSuccessesPreventsLadderReset() { + failPastWindow(); + WindowToken boundaryToken = tracker.beginDelivery(); + assertTrue(tracker.recordSuccess(boundaryToken)); + + // The stale sibling's unexpected failure breaks the streak but neither opens nor steps. + assertEquals(0, tracker.recordUnexpectedFailure(boundaryToken)); + assertFalse(tracker.inCooldown()); + tracker.recordSuccess(tracker.beginDelivery()); + + assertEquals(2 * 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())); + + // Unstamped failures are not real delivery outcomes and must not break the streak either. + tracker.recordSuccess(tracker.beginDelivery()); // second consecutive success; resets the ladder + tracker.recordFailure(null); + assertEquals(0, tracker.recordUnexpectedFailure(null)); + // The ladder reset held (two consecutive real successes), undisturbed by the null failures. + assertEquals(COOLDOWN_INITIAL_WINDOW_MILLIS, tracker.recordUnexpectedFailure(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()); + } +}