From cf48dbf2cd7426158ad9dbfea4260d2f28c73a50 Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Wed, 9 Sep 2026 11:59:28 -0700 Subject: [PATCH 001/186] feat(events): persist recorded events to an append-only log (tier 3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Tier 3 of the event durability spec: unrecoverable failure. EventStore stages serialized events in memory and commits them to an append-only log in the no-backup files directory, so events survive a process that never gets to flush. Commit points — track, identify, an explicit flush — write on the caller's thread before returning, and the evaluations counted since the last one are summarized and written with them. Delivery reads closed batches back from disk and deletes them only once LaunchDarkly has accepted them, under a payload ID that stays the same across retries so a redelivered batch is not counted twice. Each process of a multi-process application gets its own log. Spec: Event Durability, "Tier 3 — unrecoverable failure"; the mechanism is Page-Cache Event Persistence. Co-authored-by: Cursor --- .../example/FlushOnCrashHandler.java | 67 ++ .../sdk/android/AnalyticsEventSender.java | 130 +++ .../sdk/android/AndroidEventProcessor.java | 152 +++- .../sdk/android/AndroidPlatformState.java | 39 + .../sdk/android/ComponentsImpl.java | 20 + .../launchdarkly/sdk/android/EventStore.java | 858 ++++++++++++++++++ .../sdk/android/PlatformState.java | 24 + .../internal/events/AndroidEventBuffer.java | 86 ++ .../AndroidEventBufferSerializationTest.java | 128 +++ .../android/AndroidEventProcessorTest.java | 29 + .../sdk/android/EventProcessorTestBase.java | 18 +- .../android/EventStoreMultiProcessTest.java | 332 +++++++ .../sdk/android/MockPlatformState.java | 24 + 13 files changed, 1889 insertions(+), 18 deletions(-) create mode 100644 example/src/main/java/com/launchdarkly/example/FlushOnCrashHandler.java create mode 100644 launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AnalyticsEventSender.java create mode 100644 launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventStore.java create mode 100644 launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/AndroidEventBufferSerializationTest.java create mode 100644 launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventStoreMultiProcessTest.java diff --git a/example/src/main/java/com/launchdarkly/example/FlushOnCrashHandler.java b/example/src/main/java/com/launchdarkly/example/FlushOnCrashHandler.java new file mode 100644 index 000000000..dbe8a3e91 --- /dev/null +++ b/example/src/main/java/com/launchdarkly/example/FlushOnCrashHandler.java @@ -0,0 +1,67 @@ +package com.launchdarkly.example; + +import com.launchdarkly.sdk.android.LDClient; + +import java.util.concurrent.TimeUnit; + +import timber.log.Timber; + +/** + * Delivers buffered analytics events from the uncaught exception handler, which is the most an + * application can do about event loss while the SDK keeps its events only in memory. + *

+ * This is what makes the two instant buttons in {@link MainActivity} an experiment and its control. + * An uncaught exception runs this handler while the process is still alive and its other threads are + * still running, so the events recorded a moment earlier can still reach the network. + * {@code SIGKILL} runs nothing, and neither does an ANR, a native crash, or the system reclaiming a + * backgrounded process, so those lose the same events. The difference between the two buttons is the + * ground that on-disk persistence would cover and a crash handler cannot. + */ +final class FlushOnCrashHandler implements Thread.UncaughtExceptionHandler { + /** + * How long the crash is held open for the events. + *

+ * The SDK's HTTP timeouts are measured in seconds, and a request that hangs must not hold the + * process in a half-dead state for all of them: past this point the events are worth less than + * the delay, and the crash goes on to be reported. + */ + private static final long DELIVERY_BUDGET_MILLIS = 2_000; + + private final Thread.UncaughtExceptionHandler next; + + private FlushOnCrashHandler(Thread.UncaughtExceptionHandler next) { + this.next = next; + } + + /** + * Installs the handler in front of whatever was already there, which on a real application is + * the crash reporter, and on this one is the platform handler that prints the trace. + */ + static void install() { + Thread.UncaughtExceptionHandler previous = Thread.getDefaultUncaughtExceptionHandler(); + if (previous instanceof FlushOnCrashHandler) { + return; + } + Thread.setDefaultUncaughtExceptionHandler(new FlushOnCrashHandler(previous)); + } + + @Override + public void uncaughtException(Thread thread, Throwable throwable) { + try { + // Waiting here on the crashing thread is safe because the timeout is the SDK's to + // enforce: it stops waiting on the delivery rather than trusting it to finish. That also + // covers the case where this crash is the reason the delivery cannot complete, such as + // an exception thrown while the event buffer was locked. + boolean delivered = LDClient.get() + .flushAndWait(DELIVERY_BUDGET_MILLIS, TimeUnit.MILLISECONDS); + Timber.w("crash handler: events delivered = %b", delivered); + } catch (Throwable t) { + // Nothing that happens in here is worth losing the crash report over. + Timber.e(t, "Could not deliver events from the crash handler"); + } finally { + if (next != null) { + next.uncaughtException(thread, throwable); + } + } + } +} diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AnalyticsEventSender.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AnalyticsEventSender.java new file mode 100644 index 000000000..1dcd9bb31 --- /dev/null +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AnalyticsEventSender.java @@ -0,0 +1,130 @@ +package com.launchdarkly.sdk.android; + +import com.launchdarkly.logging.LDLogger; +import com.launchdarkly.logging.LogValues; +import com.launchdarkly.sdk.internal.events.EventSender; +import com.launchdarkly.sdk.internal.http.HttpHelpers; +import com.launchdarkly.sdk.internal.http.HttpProperties; + +import java.io.Closeable; +import java.io.IOException; +import java.net.URI; +import java.util.Date; + +import okhttp3.Headers; +import okhttp3.MediaType; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.RequestBody; +import okhttp3.Response; + +/** + * Posts a batch of analytics events under a payload ID the caller chooses. + *

+ * This exists for the sake of one header. java-sdk-internal's sender invents a payload ID per call, + * which is the right thing when the events being sent only ever existed in memory: the sole reason to + * send the same events twice is its own immediate retry, and it holds the ID for that. Once events are + * kept on disk, a delivery can also be repeated by a later run of the application, after a process died + * without learning whether the service had received the batch. Those two attempts are the same delivery + * and have to say so, or the events are counted twice - and counting them twice is worse than losing + * them, because an inflated exposure count silently biases an experiment rather than visibly shrinking + * it. + *

+ * The ID belongs to the batch, which is why it is the batch's identity in {@link EventStore} rather than + * something generated at send time. + */ +final class AnalyticsEventSender implements Closeable { + private static final MediaType JSON = MediaType.parse("application/json; charset=utf-8"); + private static final String EVENT_SCHEMA_HEADER = "X-LaunchDarkly-Event-Schema"; + private static final String EVENT_SCHEMA_VERSION = "4"; + private static final String PAYLOAD_ID_HEADER = "X-LaunchDarkly-Payload-ID"; + /** One immediate retry, as the other SDKs do, for a failure that may just be a bad moment. */ + private static final long RETRY_DELAY_MILLIS = 1000; + + private final OkHttpClient client; + private final boolean ownsClient; + private final Headers baseHeaders; + private final LDLogger logger; + + AnalyticsEventSender(HttpProperties httpProperties, LDLogger logger) { + this.logger = logger; + if (httpProperties.getSharedHttpClient() == null) { + this.client = httpProperties.toHttpClientBuilder().build(); + this.ownsClient = true; + } else { + this.client = httpProperties.getSharedHttpClient(); + this.ownsClient = false; + } + this.baseHeaders = httpProperties.toHeadersBuilder().build(); + } + + /** + * Posts a batch, retrying once if the failure looks temporary. + * + * @param body the JSON request body + * @param eventCount how many events the body holds, for logging + * @param payloadId identifies this delivery, the same on every attempt at it + * @param eventsBaseUri the events service + * @return whether the events arrived, whether the SDK must stop, and the service's clock + */ + EventSender.Result sendBatch(byte[] body, int eventCount, String payloadId, URI eventsBaseUri) { + URI uri = HttpHelpers.concatenateUriPath(eventsBaseUri, + StandardEndpoints.ANALYTICS_EVENTS_REQUEST_PATH); + + for (int attempt = 0; attempt < 2; attempt++) { + if (attempt > 0) { + try { + Thread.sleep(RETRY_DELAY_MILLIS); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + return new EventSender.Result(false, false, null); + } + logger.warn("Will retry posting {} event(s) after a failure", eventCount); + } + + Request request = new Request.Builder() + .url(uri.toString()) + .headers(baseHeaders) + .addHeader("Content-Type", "application/json") + .addHeader(EVENT_SCHEMA_HEADER, EVENT_SCHEMA_VERSION) + .addHeader(PAYLOAD_ID_HEADER, payloadId) + .post(RequestBody.create(body, JSON)) + .build(); + + try (Response response = client.newCall(request).execute()) { + if (response.isSuccessful()) { + logger.debug("Posted {} event(s)", eventCount); + return new EventSender.Result(true, false, parseDate(response)); + } + logger.warn("Error posting events: HTTP status {}", response.code()); + if (!LDUtil.isHttpErrorRecoverable(response.code())) { + return new EventSender.Result(false, true, null); + } + } catch (IOException e) { + logger.warn("Error posting events: {}", LogValues.exceptionSummary(e)); + } + } + return new EventSender.Result(false, false, null); + } + + /** + * The service's own clock, which the SDK trusts over the device's when deciding whether a debug + * window has closed. + */ + private Date parseDate(Response response) { + try { + return response.headers().getDate("Date"); + } catch (RuntimeException e) { + return null; + } + } + + @Override + public void close() { + if (!ownsClient) { + return; + } + client.dispatcher().executorService().shutdown(); + client.connectionPool().evictAll(); + } +} diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AndroidEventProcessor.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AndroidEventProcessor.java index 3af214035..72e21d2f9 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AndroidEventProcessor.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AndroidEventProcessor.java @@ -35,7 +35,9 @@ */ final class AndroidEventProcessor implements EventProcessor { private final AndroidEventBuffer buffer; + private final EventStore store; private final EventSender eventSender; + private final AnalyticsEventSender analyticsEventSender; private final URI eventsUri; private final DiagnosticStore diagnosticStore; private final long flushIntervalMillis; @@ -50,6 +52,17 @@ final class AndroidEventProcessor implements EventProcessor { private final AtomicBoolean disabled = new AtomicBoolean(false); private final AtomicBoolean diagnosticInitSent = new AtomicBoolean(false); private final AtomicLong lastKnownPastTime = new AtomicLong(0); + private final AtomicBoolean capacityExceeded = new AtomicBoolean(false); + private final AtomicLong droppedEvents = new AtomicLong(0); + /** + * Whether a previous run of the application left events behind. + *

+ * Those events are already as old as whatever happened to that run, so they are sent as soon as the + * SDK is allowed to send anything rather than at the next interval. The application in front of the + * user has moved on; an error report about a crash is not worth much thirty seconds late, and the + * process might not live that long either. + */ + private final AtomicBoolean hasEventsFromPreviousRun = new AtomicBoolean(false); private final Object stateLock = new Object(); private ScheduledFuture flushTask; @@ -57,7 +70,9 @@ final class AndroidEventProcessor implements EventProcessor { AndroidEventProcessor( AndroidEventBuffer buffer, + EventStore store, EventSender eventSender, + AnalyticsEventSender analyticsEventSender, URI eventsUri, DiagnosticStore diagnosticStore, long flushIntervalMillis, @@ -68,7 +83,9 @@ final class AndroidEventProcessor implements EventProcessor { LDLogger logger ) { this.buffer = buffer; + this.store = store; this.eventSender = eventSender; + this.analyticsEventSender = analyticsEventSender; this.eventsUri = eventsUri; this.diagnosticStore = diagnosticStore; this.flushIntervalMillis = flushIntervalMillis; @@ -78,6 +95,21 @@ final class AndroidEventProcessor implements EventProcessor { this.inBackground = new AtomicBoolean(initiallyInBackground); this.offline = new AtomicBoolean(initiallyOffline); + // Off the calling thread: this is on the path of LDClient.init, and it reads the filesystem. + submit(() -> { + store.recoverInterruptedLog(); + if (store.pendingBatches().isEmpty()) { + return; + } + // Either end of the race is handled: if the SDK is already allowed to send, these go now, + // and if it is not, going online later will pick them up. + if (offline.get() || isStopped()) { + hasEventsFromPreviousRun.set(true); + } else { + deliverPayload(); + } + }); + synchronized (stateLock) { updateScheduledTasks(initiallyInBackground, initiallyOffline); } @@ -103,11 +135,14 @@ public void recordEvaluationEvent( requireFullEvent, debugEventsUntilDate, false); buffer.summarize(event); if (requireFullEvent) { - buffer.addFullEvent(event); + record(event); } if (shouldDebugEvent(debugEventsUntilDate)) { - buffer.addFullEvent(event.toDebugEvent()); + record(event.toDebugEvent()); } + // Deliberately no commit. An evaluation is expected to cost what a map lookup costs, and it is + // usually the main thread doing it; the store writes these on its own thread once enough of them + // have piled up, and the next event recorded at a commit point makes them durable along with itself. } @Override @@ -115,7 +150,8 @@ public void recordIdentifyEvent(LDContext context) { if (isStopped() || context == null) { return; } - buffer.addFullEvent(new Event.Identify(System.currentTimeMillis(), context)); + record(new Event.Identify(System.currentTimeMillis(), context)); + commitDurably(); } @Override @@ -123,8 +159,53 @@ public void recordCustomEvent(LDContext context, String eventKey, LDValue data, if (isStopped() || context == null) { return; } - buffer.addFullEvent(new Event.Custom(System.currentTimeMillis(), eventKey, context, data, - metricValue)); + record(new Event.Custom(System.currentTimeMillis(), eventKey, context, data, metricValue)); + commitDurably(); + } + + /** + * Serializes an event and stages it, counting it as dropped if the store is full. + */ + private void record(Event event) { + byte[] serialized = buffer.serialize(event); + if (serialized == null) { + return; // sampled out, or unserializable, and already logged + } + if (!store.stage(serialized)) { + if (capacityExceeded.compareAndSet(false, true)) { + logger.warn("Exceeded event queue capacity. Increase capacity to avoid dropping events."); + } + droppedEvents.incrementAndGet(); + return; + } + capacityExceeded.set(false); + } + + /** + * Makes everything recorded so far survive the process, on the caller's thread. + *

+ * This is the point of the whole arrangement, so it is worth being precise about which callers get + * it: those recording an event the application asked for by name. An application that calls + * {@code track} to report an error is telling the SDK that this event matters more than the + * microseconds it costs to write it, and the crash it is reporting may be moments away. + *

+ * The evaluations counted since the last commit point are written too, because a summary is only worth + * having if it covers the evaluations that led up to whatever is about to happen. + */ + private void commitDurably() { + stageSummaries(); + store.commit(); + } + + /** + * Turns the evaluation counters into summary events in the store. + */ + private void stageSummaries() { + for (byte[] summary : buffer.serializeSummariesAndReset()) { + // Bypassing capacity: a summary is not a new event, it is the record of evaluations already + // counted, and dropping it would lose all of them at once. + store.stage(summary, true); + } } @Override @@ -158,6 +239,9 @@ public void flush() { if (isStopped()) { return; } + // Flush is a commit point: persist everything accepted before this call before returning, + // even when the queued delivery cannot run because the client is offline. + commitDurably(); submit(this::deliverPayload); } @@ -166,6 +250,7 @@ public void blockingFlush() { if (isStopped()) { return; } + commitDurably(); Future delivery = submit(this::deliverPayload); if (delivery == null) { return; @@ -184,6 +269,7 @@ public boolean blockingFlush(long timeout, TimeUnit unit) { if (isStopped()) { return false; } + commitDurably(); // Typed rather than inlined, so that it is unambiguously submitted as work with a result. Callable delivery = this::deliverPayloadReportingOutcome; Future pending; @@ -230,7 +316,11 @@ public void close() throws IOException { } } scheduler.shutdown(); + // Last: whatever could not be delivered is written down instead, so a caller who closed the + // client and then let the process end still has those events on the next run. + store.close(); eventSender.close(); + analyticsEventSender.close(); } /** @@ -252,24 +342,49 @@ private boolean deliverPayloadReportingOutcome() { if (disabled.get() || offline.get()) { return false; } - AndroidEventBuffer.Payload payload; - try { - payload = buffer.drain(); - } catch (IOException e) { - logUnexpectedError(e); - return false; + + stageSummaries(); + store.closeBatch(); + + // Every batch, not just the one just closed: the others are deliveries an earlier attempt did not + // finish, or that a previous run of the application never got to start. + boolean allDelivered = true; + for (EventStore.Batch batch : store.pendingBatches()) { + if (disabled.get() || offline.get()) { + return false; + } + allDelivered &= deliver(batch); } - if (payload == null) { + return allDelivered; + } + + /** + * @return true if the batch is no longer the SDK's problem, whether because it arrived or because it + * never can + */ + private boolean deliver(EventStore.Batch batch) { + byte[] body = store.body(batch); + if (body == null) { + // Unreadable, or already delivered by another process of this application. Either way there + // is nothing to send and nothing to keep. + store.remove(batch); return true; } if (diagnosticStore != null) { - diagnosticStore.recordEventsInBatch(payload.getEventCount()); + diagnosticStore.recordEventsInBatch(batch.eventCount); } try { - EventSender.Result result = eventSender.sendAnalyticsEvents(payload.getData(), - payload.getEventCount(), eventsUri); + EventSender.Result result = analyticsEventSender.sendBatch(body, batch.eventCount, + batch.payloadId, eventsUri); handleResponse(result); - return result != null && result.isSuccess(); + if (result != null && (result.isSuccess() || result.isMustShutDown())) { + // Forgotten once the service has either taken the events or told us to stop sending + // them. Anything else - a timeout, a 503, no network - leaves the batch where it is, to + // be tried again on the next flush or the next run of the application. + store.remove(batch); + return result.isSuccess(); + } + return false; } catch (Exception e) { logUnexpectedError(e); return false; @@ -296,7 +411,7 @@ private void sendDiagnosticStats() { if (disabled.get() || diagnosticStore == null) { return; } - sendDiagnosticEvent(diagnosticStore.createEventAndReset(buffer.getAndClearDroppedCount(), 0), + sendDiagnosticEvent(diagnosticStore.createEventAndReset(droppedEvents.getAndSet(0), 0), false); } @@ -333,6 +448,9 @@ private void updateScheduledTasks(boolean inBackground, boolean offline) { // events recorded before the app was backgrounded still get delivered. flushTask = enableOrDisableTask(!offline, flushTask, flushIntervalMillis, this::deliverPayload); + if (!offline && hasEventsFromPreviousRun.compareAndSet(true, false)) { + submit(this::deliverPayload); + } boolean diagnosticsEnabled = diagnosticStore != null && !offline && !inBackground; diagnosticTask = enableOrDisableTask(diagnosticsEnabled, diagnosticTask, diagnosticRecordingIntervalMillis, this::sendDiagnosticStats); diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AndroidPlatformState.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AndroidPlatformState.java index 357648917..3fb8c39e7 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AndroidPlatformState.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/AndroidPlatformState.java @@ -18,8 +18,13 @@ import android.os.Bundle; import com.launchdarkly.logging.LDLogger; +import com.launchdarkly.logging.LogValues; +import java.io.BufferedReader; import java.io.File; +import java.io.FileInputStream; +import java.io.InputStreamReader; +import java.nio.charset.Charset; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.atomic.AtomicBoolean; @@ -169,6 +174,40 @@ public File getCacheDir() { return application.getCacheDir(); } + @Override + public File getNoBackupFilesDir() { + return application.getNoBackupFilesDir(); + } + + @Override + public String getProcessName() { + // Returned as the platform gives it, punctuation and all. Whatever needs a name safe for some + // other purpose is responsible for making it so, because only that caller knows what "safe" + // means there; see EventStore. + if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.P) { + return Application.getProcessName(); + } + // Before API 28 there is no supported way to ask, so the process reads its own command line, + // which is where the platform gets the answer from as well. + try (BufferedReader reader = new BufferedReader( + new InputStreamReader(new FileInputStream("/proc/self/cmdline"), Charset.forName("UTF-8")))) { + String line = reader.readLine(); + if (line != null) { + // The command line is NUL-terminated, and Java keeps those bytes as characters. + int end = line.indexOf(0); + String name = end < 0 ? line : line.substring(0, end); + if (!name.trim().isEmpty()) { + return name.trim(); + } + } + } catch (Exception e) { + logger.debug("Could not determine the process name: {}", LogValues.exceptionSummary(e)); + } + // A wrong answer here would let two processes share one file, so the fallback is the pid, which + // is unique among live processes even though it does not survive a restart. + return "pid-" + android.os.Process.myPid(); + } + @Override public void close() { connectivityChangeListeners.clear(); diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/ComponentsImpl.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/ComponentsImpl.java index 62ea81560..0944ac50e 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/ComponentsImpl.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/ComponentsImpl.java @@ -113,7 +113,10 @@ public EventProcessor build(ClientContext clientContext) { privateAttributes, true, // perContextSummarization - enable for client SDK clientContext.getBaseLogger()), + makeEventStore(clientContext, clientContextImpl), eventSender, + new AnalyticsEventSender(LDUtil.makeHttpProperties(clientContext), + clientContext.getBaseLogger()), clientContext.getServiceEndpoints().getEventsBaseUri(), clientContextImpl.getDiagnosticStore(), flushIntervalMillis, @@ -125,6 +128,23 @@ public EventProcessor build(ClientContext clientContext) { ); } + /** + * Builds the store this environment's events are kept in until LaunchDarkly has them. + *

+ * The mobile key and the process are both part of where it writes, so an application configured + * for several environments, or running the SDK in several processes, gets one store per + * combination and they never touch each other's events. + */ + private EventStore makeEventStore(ClientContext clientContext, ClientContextImpl impl) { + PlatformState platformState = impl.getPlatformState(); + return EventStore.create( + platformState.getNoBackupFilesDir(), + clientContext.getMobileKey(), + platformState.getProcessName(), + capacity, + clientContext.getBaseLogger()); + } + @Override public LDValue describeConfiguration(ClientContext clientContext) { return LDValue.buildObject() diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventStore.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventStore.java new file mode 100644 index 000000000..b615902ed --- /dev/null +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/EventStore.java @@ -0,0 +1,858 @@ +package com.launchdarkly.sdk.android; + +import com.launchdarkly.logging.LDLogger; +import com.launchdarkly.logging.LogValues; + +import java.io.ByteArrayOutputStream; +import java.io.Closeable; +import java.io.File; +import java.io.FileInputStream; +import java.io.FileOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.nio.charset.Charset; +import java.security.MessageDigest; +import java.security.NoSuchAlgorithmException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; + +/** + * A crash-durable append log of serialized analytics events. + *

+ * Events are appended to an open log, which is closed into a batch when the event processor is ready to + * deliver one. A batch is deleted only once LaunchDarkly has accepted it, so a delivery interrupted by + * the process dying is retried on the next run rather than lost. + *

+ * Bytes reach the kernel through an ordinary write and are never {@code fsync}ed. What this defends + * against is the process dying - which is what an uncaught exception, an ANR kill, a native crash, and + * the system reclaiming a backgrounded application all are - and a process cannot take back bytes the + * kernel already holds. Only losing the kernel itself, to a panic or a power cut, can discard them, and + * covering that would mean an {@code fsync} per event, which costs milliseconds where this costs + * microseconds. + *

+ * Several clients. An application configured for more than one environment gets one + * {@code LDClient} and one store per mobile key, and the mobile key's hash is part of the directory, so + * two environments never read or write each other's events. + *

+ * Several processes. An application may declare components in more than one process, and each + * gets its own copy of the SDK that shares nothing but the filesystem. Only the open log is per-process, + * named after the process appending to it, which is what keeps two processes from interleaving frames in + * one file and keeps one process's startup recovery from stealing a log another is still writing to. + * Closed batches are deliberately left shared: a process that recorded events and then died may never + * run again, and a batch nobody but that process could see would never be delivered. If two processes + * do deliver the same batch, its payload ID is the same both times, and LaunchDarkly discards a repeated + * payload ID rather than counting the events twice. + *

+ * The format is byte-for-byte the one the iOS SDK writes, so the two can be reasoned about, and + * debugged, as one thing. + */ +final class EventStore implements Closeable { + /** Names a log closed off for delivery; the rest of the name is the batch's payload ID. */ + private static final String BATCH_PREFIX = "ready-"; + /** Names the log a single process appends to; the rest of the name is that process. */ + private static final String OPEN_PREFIX = "open-"; + private static final String DIRECTORY_NAME = "com.launchdarkly.events"; + + /** + * How many staged bytes may accumulate before one of them pays for a write. + *

+ * This is what keeps a syscall off most recordings while bounding what a crash can take with it, + * for an application that only evaluates flags and so never reaches a commit point of its own. + */ + private static final int STAGING_THRESHOLD = 16 * 1024; + + private final File directory; + private final File openLog; + private final int capacity; + private final LDLogger logger; + /** + * Where a commit runs when no caller is waiting on it. + *

+ * Evaluating a flag must not put a write on whichever thread evaluated it, and that thread is + * usually the main one. A caller at a commit point still commits on its own thread, because for + * them having returned is the guarantee. + */ + private final Executor commitExecutor; + + /** + * Guards the buffer and the counters. Held for a copy and never across a write, so an + * evaluation recording an event waits on another thread's array copy at worst, never on the disk. + */ + private final Object bufferLock = new Object(); + /** + * Serializes writes so that two threads committing at once cannot interleave a frame. Taken before + * {@link #bufferLock}, never after it. + */ + private final Object ioLock = new Object(); + + // Guarded by bufferLock. + private final ByteArrayOutputStream bufferData = new ByteArrayOutputStream(); + private int bufferedEventCount; + private int committedEvents; + private int closedEvents; + private boolean persistenceDisabled; + private boolean commitScheduled; + + // Guarded by ioLock. + private FileOutputStream output; + /** + * Batches held in memory because the filesystem would not take them. + *

+ * Persistence failing should cost durability and nothing else, so the store falls back to what the + * SDK did before it existed: hold the events, deliver them, lose them only if the process dies. + */ + private final Map inMemoryBatches = new LinkedHashMap<>(); + + EventStore(File directory, String processName, int capacity, LDLogger logger, Executor commitExecutor) { + this.directory = directory; + this.openLog = new File(directory, OPEN_PREFIX + logNameFor(processName)); + this.capacity = capacity >= 0 ? capacity : 1; + this.logger = logger; + this.commitExecutor = commitExecutor; + } + + /** + * Creates the store for one environment of one process. + * + * @param noBackupFilesDir where the platform lets the SDK keep files it must not lose + * @param mobileKey identifies the environment, so several clients stay out of each other's way + * @param processName identifies the process, so several processes stay out of each other's way + */ + static EventStore create( + File noBackupFilesDir, + String mobileKey, + String processName, + int capacity, + LDLogger logger + ) { + File directory = new File(new File(noBackupFilesDir, DIRECTORY_NAME), + environmentDirectoryName(mobileKey)); + return new EventStore(directory, processName, capacity, logger, defaultCommitExecutor()); + } + + /** + * Names the directory for one environment, hashed so the mobile key itself is never a filename. + *

+ * Deliberately not {@link LDUtil#urlSafeBase64Hash(String)}, which goes through the Android + * framework's Base64 and so is unavailable to plain JVM tests, and which pads with a character + * that has no business in a path. + */ + private static String environmentDirectoryName(String mobileKey) { + // Half a digest, which is far more than enough to keep two mobile keys apart. + return hexDigest(mobileKey, 16); + } + + /** + * Names the log file belonging to one process. + *

+ * The readable part is the process name with anything awkward in a filename replaced, which is what + * makes a directory of these diagnosable by a human. That reduction is not one-to-one, though: + * {@code com.app:remote} and {@code com.app.remote} are both legal process names and both come out as + * {@code com_app_remote}. Two processes sharing one log is the single thing this name exists to + * prevent, so the digest of the original name is appended to guarantee they never do. + */ + static String logNameFor(String processName) { + String name = processName == null ? "" : processName.trim(); + if (name.isEmpty()) { + name = "unknown"; + } + StringBuilder safe = new StringBuilder(name.length() + 9); + for (int i = 0; i < name.length(); i++) { + char c = name.charAt(i); + safe.append(Character.isLetterOrDigit(c) || c == '-' || c == '_' ? c : '_'); + } + return safe.append('-').append(hexDigest(name, 4)).toString(); + } + + /** + * @param bytes how many bytes of the digest to render, so a caller can trade length for headroom + * @return the first {@code bytes} bytes of the input's SHA-256, in lowercase hex + */ + private static String hexDigest(String input, int bytes) { + try { + byte[] hash = MessageDigest.getInstance("SHA-256") + .digest(input.getBytes(Charset.forName("UTF-8"))); + StringBuilder hex = new StringBuilder(bytes * 2); + for (int i = 0; i < bytes; i++) { + hex.append(Character.forDigit((hash[i] >> 4) & 0xf, 16)); + hex.append(Character.forDigit(hash[i] & 0xf, 16)); + } + return hex.toString(); + } catch (NoSuchAlgorithmException e) { + throw new RuntimeException(e); // shouldn't be possible; SHA-256 is built in + } + } + + private static Executor defaultCommitExecutor() { + return Executors.newSingleThreadExecutor(new ThreadFactory() { + @Override + public Thread newThread(Runnable r) { + Thread thread = new Thread(r, "LaunchDarkly-event-store"); + thread.setDaemon(true); + return thread; + } + }); + } + + // MARK: recording + + /** + * @return serialized events recorded but not yet accepted by LaunchDarkly, whether staged in + * memory, written to the open log, or sitting in a closed batch + */ + int getPendingEventCount() { + synchronized (bufferLock) { + return bufferedEventCount + committedEvents + closedEvents; + } + } + + /** + * Stages an already serialized event. + *

+ * Staging is a copy into a buffer; {@link #commit()} is what makes the event outlive the process. + * + * @return false if the event was dropped because capacity is reached + */ + boolean stage(byte[] serializedEvent) { + return stage(serializedEvent, false); + } + + /** + * @param bypassingCapacity true for an event that must be recorded even when the store is full + */ + boolean stage(byte[] serializedEvent, boolean bypassingCapacity) { + if (serializedEvent == null || serializedEvent.length == 0 + || serializedEvent.length > Format.MAX_FRAME_SIZE) { + return false; + } + + boolean needsCommit; + synchronized (bufferLock) { + if (!bypassingCapacity + && bufferedEventCount + committedEvents + closedEvents >= capacity) { + return false; + } + Format.writeFrame(bufferData, serializedEvent); + bufferedEventCount++; + needsCommit = !persistenceDisabled && bufferData.size() >= STAGING_THRESHOLD && !commitScheduled; + if (needsCommit) { + commitScheduled = true; + } + } + + // Handed to another thread rather than done here. Recording an event is something a flag + // evaluation does, and an evaluation is expected to be a memory operation: the caller is usually + // the main thread, where a write that happens to find a busy filesystem is a stall the + // application can do nothing about. + if (needsCommit) { + try { + commitExecutor.execute(this::commitScheduled); + } catch (RuntimeException e) { + // A rejected commit only means these bytes wait for the next one. + synchronized (bufferLock) { + commitScheduled = false; + } + } + } + return true; + } + + private void commitScheduled() { + // Cleared before the commit rather than after, so an event staged while this write is in flight + // asks for another one instead of finding a commit apparently already on its way. + synchronized (bufferLock) { + commitScheduled = false; + } + commit(); + } + + /** + * Hands every staged byte to the kernel, so the events recorded so far survive the process dying. + */ + void commit() { + synchronized (ioLock) { + commitHoldingIoLock(); + } + } + + // MARK: delivery + + /** + * Commits, then closes the open log into a batch to deliver. + * + * @return the batch, or null when there is nothing to send + */ + Batch closeBatch() { + synchronized (ioLock) { + commitHoldingIoLock(); + + int events; + int stillBuffered; + synchronized (bufferLock) { + events = committedEvents; + stillBuffered = bufferedEventCount; + } + + if (events > 0) { + return closeOpenLogHoldingIoLock(events); + } + if (stillBuffered > 0) { + // Only reachable once persistence has been given up on; a healthy commit leaves nothing + // staged behind. + return closeInMemoryBatchHoldingIoLock(); + } + return null; + } + } + + /** Requires {@code ioLock}. */ + private Batch closeOpenLogHoldingIoLock(int events) { + closeOutputHoldingIoLock(); + + String payloadId = UUID.randomUUID().toString(); + if (!openLog.renameTo(batchFile(payloadId))) { + logger.warn("Could not close the event log for delivery"); + if (!openLog.exists()) { + // The log itself is gone, so its events are too. They have to stop counting against + // capacity or the store would refuse events for the rest of the session. + synchronized (bufferLock) { + committedEvents = 0; + } + } + return null; + } + + synchronized (bufferLock) { + committedEvents = 0; + closedEvents += events; + } + return new Batch(payloadId, events); + } + + /** Requires {@code ioLock}. */ + private Batch closeInMemoryBatchHoldingIoLock() { + byte[] frames; + int events; + synchronized (bufferLock) { + frames = bufferData.toByteArray(); + events = bufferedEventCount; + bufferData.reset(); + bufferedEventCount = 0; + } + byte[] body = Format.assembleBody(Format.withFileHeader(frames)); + if (body == null) { + return null; + } + String payloadId = UUID.randomUUID().toString(); + inMemoryBatches.put(payloadId, new HeldBatch(body, events)); + synchronized (bufferLock) { + closedEvents += events; + } + return new Batch(payloadId, events); + } + + /** A batch the filesystem would not take, kept where the SDK used to keep all of them. */ + private static final class HeldBatch { + final byte[] body; + final int eventCount; + + HeldBatch(byte[] body, int eventCount) { + this.body = body; + this.eventCount = eventCount; + } + } + + /** + * @return batches awaiting delivery, oldest first, including any left by a previous run + */ + List pendingBatches() { + synchronized (ioLock) { + List batches = new ArrayList<>(); + File[] files = directory.listFiles(); + if (files != null) { + List ready = new ArrayList<>(); + for (File file : files) { + if (file.getName().startsWith(BATCH_PREFIX)) { + ready.add(file); + } + } + Collections.sort(ready, new Comparator() { + @Override + public int compare(File a, File b) { + return Long.compare(a.lastModified(), b.lastModified()); + } + }); + for (File file : ready) { + int events = Format.eventCount(readFile(file)); + if (events < 0) { + // Written by a version whose format this one does not read, or damaged beyond + // what the torn-tail recovery tolerates. Either way it can never be delivered. + deleteQuietly(file); + continue; + } + batches.add(new Batch(file.getName().substring(BATCH_PREFIX.length()), events)); + } + } + for (Map.Entry held : inMemoryBatches.entrySet()) { + batches.add(new Batch(held.getKey(), held.getValue().eventCount)); + } + + int total = 0; + for (Batch batch : batches) { + total += batch.eventCount; + } + synchronized (bufferLock) { + closedEvents = total; + } + return batches; + } + } + + /** + * @return the JSON request body for a batch, or null if it has since become unreadable + */ + byte[] body(Batch batch) { + synchronized (ioLock) { + HeldBatch held = inMemoryBatches.get(batch.payloadId); + if (held != null) { + return held.body; + } + return Format.assembleBody(readFile(batchFile(batch.payloadId))); + } + } + + /** + * Forgets a batch, which is only correct once LaunchDarkly has accepted it or permanently refused it. + */ + void remove(Batch batch) { + synchronized (ioLock) { + if (inMemoryBatches.remove(batch.payloadId) == null) { + deleteQuietly(batchFile(batch.payloadId)); + } + synchronized (bufferLock) { + closedEvents = Math.max(0, closedEvents - batch.eventCount); + } + } + } + + /** + * Closes any log this process left open on a previous run, so its events join the batches to deliver. + *

+ * The events in it were recorded by a process that is gone, so there is no one left to add to it. Only + * this process's own log is touched: another process's may still be open in a process that is alive. + */ + void recoverInterruptedLog() { + synchronized (ioLock) { + if (!openLog.exists()) { + return; + } + int events = Format.eventCount(readFile(openLog)); + if (events < 0) { + // Unreadable, and a log that cannot be read cannot be appended to either. + deleteQuietly(openLog); + return; + } + if (events == 0) { + deleteQuietly(openLog); + return; + } + String payloadId = UUID.randomUUID().toString(); + if (openLog.renameTo(batchFile(payloadId))) { + logger.info("Recovered {} event(s) that a previous run of this application did not deliver", + events); + } + } + } + + // MARK: writing + + /** Requires {@code ioLock}. */ + private void commitHoldingIoLock() { + byte[] bytes; + int events; + synchronized (bufferLock) { + if (persistenceDisabled || bufferedEventCount == 0) { + // Staged bytes are left where they are: with nowhere durable to put them, memory is + // better than dropping them. + return; + } + bytes = bufferData.toByteArray(); + events = bufferedEventCount; + bufferData.reset(); + bufferedEventCount = 0; + } + + if (append(bytes)) { + synchronized (bufferLock) { + committedEvents += events; + } + } else { + // Put back, to be delivered from memory rather than lost. + synchronized (bufferLock) { + byte[] laterFrames = bufferData.toByteArray(); + bufferData.reset(); + writeQuietly(bufferData, bytes); + writeQuietly(bufferData, laterFrames); + bufferedEventCount += events; + } + } + } + + /** + * Requires {@code ioLock}. + * + * @return false if the bytes could not be written, in which case persistence is given up on + */ + private boolean append(byte[] bytes) { + try { + FileOutputStream stream = outputForAppendingHoldingIoLock(); + if (stream == null) { + return false; + } + stream.write(bytes); + return true; + } catch (IOException | RuntimeException e) { + // Never let a full disk take the application down. The SDK gives up on persistence for the + // rest of the session instead, which is how other SDKs have crashed their hosts. + logger.warn("Giving up on persisting events: {}", LogValues.exceptionSummary(e)); + closeOutputHoldingIoLock(); + synchronized (bufferLock) { + persistenceDisabled = true; + } + return false; + } + } + + /** Requires {@code ioLock}. */ + private FileOutputStream outputForAppendingHoldingIoLock() throws IOException { + if (output != null) { + return output; + } + synchronized (bufferLock) { + if (persistenceDisabled) { + return null; + } + } + if (!directory.exists() && !directory.mkdirs() && !directory.isDirectory()) { + throw new IOException("could not create " + directory); + } + boolean isNew = !openLog.exists() || openLog.length() == 0; + // Append mode is what makes each write land at the end of the file as one step, so that a + // process cannot splice its bytes into the middle of what another wrote. + FileOutputStream stream = new FileOutputStream(openLog, true); + if (isNew) { + stream.write(Format.fileHeader()); + } + output = stream; + return output; + } + + /** Requires {@code ioLock}. */ + private void closeOutputHoldingIoLock() { + if (output == null) { + return; + } + try { + output.close(); + } catch (IOException e) { + logger.debug("Could not close the event log: {}", LogValues.exceptionSummary(e)); + } + output = null; + } + + @Override + public void close() { + synchronized (ioLock) { + commitHoldingIoLock(); + closeOutputHoldingIoLock(); + } + } + + private File batchFile(String payloadId) { + return new File(directory, BATCH_PREFIX + payloadId); + } + + File getDirectory() { + return directory; + } + + private void deleteQuietly(File file) { + if (file.exists() && !file.delete()) { + logger.debug("Could not delete {}", file); + } + } + + private static void writeQuietly(ByteArrayOutputStream target, byte[] bytes) { + try { + target.write(bytes); + } catch (IOException e) { + // ByteArrayOutputStream.write does not throw; this only satisfies the compiler. + } + } + + /** + * @return the file's bytes, or null if it could not be read + */ + private byte[] readFile(File file) { + ByteArrayOutputStream contents = new ByteArrayOutputStream(); + byte[] chunk = new byte[8192]; + try (InputStream in = new FileInputStream(file)) { + int read; + while ((read = in.read(chunk)) > 0) { + contents.write(chunk, 0, read); + } + } catch (IOException e) { + return null; + } + return contents.toByteArray(); + } + + /** + * A group of serialized events closed off for delivery. + *

+ * The identifier doubles as the payload ID sent to LaunchDarkly, so a batch retried after the process + * died mid-delivery is recognized upstream as the same delivery rather than counted twice. + */ + static final class Batch { + final String payloadId; + final int eventCount; + + Batch(String payloadId, int eventCount) { + this.payloadId = payloadId; + this.eventCount = eventCount; + } + + @Override + public boolean equals(Object other) { + if (!(other instanceof Batch)) { + return false; + } + Batch o = (Batch) other; + return payloadId.equals(o.payloadId) && eventCount == o.eventCount; + } + + @Override + public int hashCode() { + return payloadId.hashCode() * 31 + eventCount; + } + + @Override + public String toString() { + return "Batch(" + payloadId + ", " + eventCount + " events)"; + } + } + + /** + * The on-disk shape of an event log. + *

+ * A log opens with a magic and a format version, and holds a sequence of frames: + * 

+     *     +- 4 bytes -+- 2 bytes -+     +- 2 bytes -+-  4 bytes  -+- n bytes -+
+     *     |   LDEV    |  version  | ... |   type    |  length (n) |  payload  |
+     *     +-----------+-----------+     +-----------+-------------+-----------+
+     *
+ * The length prefix is what makes a log recoverable: a reader can tell a frame the writer finished + * from one it did not, without having to parse the event to find where it ends. The version is what + * makes an upgrade safe - a log this version does not understand is discarded rather than misread - + * and the frame type leaves room for a later version to write something new into a log this one + * still reads. + */ + static final class Format { + /** Bump this whenever the framing or the meaning of a frame changes. */ + static final int VERSION = 1; + static final byte[] MAGIC = "LDEV".getBytes(Charset.forName("US-ASCII")); + static final int FILE_HEADER_SIZE = 6; + static final int FRAME_HEADER_SIZE = 6; + /** The only frame type written today; a reader skips a frame whose type it does not know. */ + static final int EVENT_FRAME = 1; + /** A ceiling on a single frame, so a corrupt length cannot make recovery allocate wildly. */ + static final int MAX_FRAME_SIZE = 8 * 1024 * 1024; + + private Format() { + } + + static byte[] fileHeader() { + byte[] header = new byte[FILE_HEADER_SIZE]; + System.arraycopy(MAGIC, 0, header, 0, MAGIC.length); + header[4] = (byte) (VERSION >> 8); + header[5] = (byte) VERSION; + return header; + } + + static void writeFrame(ByteArrayOutputStream target, byte[] serializedEvent) { + target.write((byte) (EVENT_FRAME >> 8)); + target.write((byte) EVENT_FRAME); + int length = serializedEvent.length; + target.write((byte) (length >> 24)); + target.write((byte) (length >> 16)); + target.write((byte) (length >> 8)); + target.write((byte) length); + writeQuietly(target, serializedEvent); + } + + /** Frames on their own, as a log: only for the in-memory fallback, which has no file. */ + static byte[] withFileHeader(byte[] frames) { + byte[] log = new byte[FILE_HEADER_SIZE + frames.length]; + System.arraycopy(fileHeader(), 0, log, 0, FILE_HEADER_SIZE); + System.arraycopy(frames, 0, log, FILE_HEADER_SIZE, frames.length); + return log; + } + + /** + * @return how many event frames the log holds, or -1 if it is not a log this version reads + */ + static int eventCount(byte[] log) { + if (!isReadable(log)) { + return -1; + } + int events = 0; + int cursor = FILE_HEADER_SIZE; + while (cursor + FRAME_HEADER_SIZE <= log.length) { + int type = readUInt16(log, cursor); + int length = readInt32(log, cursor + 2); + int start = cursor + FRAME_HEADER_SIZE; + if (length <= 0 || length > MAX_FRAME_SIZE || start + length > log.length) { + // A process that died partway through a write leaves a torn frame at the end. + // Everything before it is intact, so recovery keeps that and drops only the tail. + break; + } + if (type == EVENT_FRAME) { + events++; + } + cursor = start + length; + } + return events; + } + + /** + * Assembles the JSON array LaunchDarkly expects out of the frames, without parsing the events: + * they were serialized on the way in and are shipped exactly as they were recorded. + * + * @return the request body, or null if the log is unreadable or holds no events + */ + static byte[] assembleBody(byte[] log) { + if (!isReadable(log)) { + return null; + } + ByteArrayOutputStream body = new ByteArrayOutputStream(log.length + 2); + body.write('['); + boolean isFirst = true; + int cursor = FILE_HEADER_SIZE; + while (cursor + FRAME_HEADER_SIZE <= log.length) { + int type = readUInt16(log, cursor); + int length = readInt32(log, cursor + 2); + int start = cursor + FRAME_HEADER_SIZE; + if (length <= 0 || length > MAX_FRAME_SIZE || start + length > log.length) { + break; + } + if (type == EVENT_FRAME) { + if (!isFirst) { + body.write(','); + } + body.write(log, start, length); + isFirst = false; + } + cursor = start + length; + } + if (isFirst) { + return null; + } + body.write(']'); + return body.toByteArray(); + } + + private static boolean isReadable(byte[] log) { + if (log == null || log.length < FILE_HEADER_SIZE) { + return false; + } + for (int i = 0; i < MAGIC.length; i++) { + if (log[i] != MAGIC[i]) { + return false; + } + } + return readUInt16(log, MAGIC.length) == VERSION; + } + + private static int readUInt16(byte[] bytes, int offset) { + return ((bytes[offset] & 0xff) << 8) | (bytes[offset + 1] & 0xff); + } + + private static int readInt32(byte[] bytes, int offset) { + return ((bytes[offset] & 0xff) << 24) | ((bytes[offset + 1] & 0xff) << 16) + | ((bytes[offset + 2] & 0xff) << 8) | (bytes[offset + 3] & 0xff); + } + } + + /** + * Every event the store is holding, serialized exactly as it will be sent. + *

+ * Reading the log means committing what is staged, so this is not free and is meant for tests and for + * diagnosing a store rather than for the recording path. + */ + List pendingEventPayloads() { + commit(); + synchronized (ioLock) { + List logs = new ArrayList<>(); + File[] files = directory.listFiles(); + if (files != null) { + List sorted = new ArrayList<>(); + for (File file : files) { + if (file.getName().startsWith(BATCH_PREFIX)) { + sorted.add(file); + } + } + Collections.sort(sorted, new Comparator() { + @Override + public int compare(File a, File b) { + return Long.compare(a.lastModified(), b.lastModified()); + } + }); + for (File file : sorted) { + logs.add(readFile(file)); + } + } + if (openLog.exists()) { + logs.add(readFile(openLog)); + } + synchronized (bufferLock) { + if (bufferedEventCount > 0) { + logs.add(Format.withFileHeader(bufferData.toByteArray())); + } + } + + List payloads = new ArrayList<>(); + for (byte[] log : logs) { + collectPayloads(log, payloads); + } + return payloads; + } + } + + private static void collectPayloads(byte[] log, List into) { + if (!Format.isReadable(log)) { + return; + } + int cursor = Format.FILE_HEADER_SIZE; + while (cursor + Format.FRAME_HEADER_SIZE <= log.length) { + int type = Format.readUInt16(log, cursor); + int length = Format.readInt32(log, cursor + 2); + int start = cursor + Format.FRAME_HEADER_SIZE; + if (length <= 0 || length > Format.MAX_FRAME_SIZE || start + length > log.length) { + break; + } + if (type == Format.EVENT_FRAME) { + into.add(Arrays.copyOfRange(log, start, start + length)); + } + cursor = start + length; + } + } +} diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/PlatformState.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/PlatformState.java index 6c66b28f2..d75e3903a 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/PlatformState.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/android/PlatformState.java @@ -55,4 +55,28 @@ interface ForegroundChangeListener { * @return a directory path */ File getCacheDir(); + + /** + * Returns the preferred filesystem location for files the SDK must not lose. + *

+ * This is not the cache directory: the system is free to delete cache files whenever it wants + * space, which is the opposite of what data written to survive a crash needs. It is also excluded + * from backups, so events recorded on one device are not restored onto another and reported as if + * they had happened there. + * + * @return a directory path + */ + File getNoBackupFilesDir(); + + /** + * Returns a name for the process the SDK is running in, stable across restarts of that process + * and different for each process of a multi-process application. + *

+ * An application can declare components in more than one process, and each process gets its own + * copy of the SDK with no memory shared between them. Anything the SDK keeps in files is shared + * whether it wants to be or not, so this is what lets each process keep its own. + * + * @return a process name safe to use as part of a filename + */ + String getProcessName(); } diff --git a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/internal/events/AndroidEventBuffer.java b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/internal/events/AndroidEventBuffer.java index 43d7fe6e4..879bc5923 100644 --- a/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/internal/events/AndroidEventBuffer.java +++ b/launchdarkly-android-client-sdk/src/main/java/com/launchdarkly/sdk/internal/events/AndroidEventBuffer.java @@ -1,6 +1,7 @@ package com.launchdarkly.sdk.internal.events; import com.launchdarkly.logging.LDLogger; +import com.launchdarkly.logging.LogValues; import com.launchdarkly.sdk.AttributeRef; import java.io.BufferedWriter; @@ -10,7 +11,9 @@ import java.io.Writer; import java.nio.charset.Charset; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.List; /** @@ -114,6 +117,89 @@ public synchronized void addFullEvent(Event event) { events.add(event); } + /** + * Serializes one event into the bytes it will be sent as, for a caller that keeps events + * somewhere other than this buffer's list. + *

+ * Serializing at record time rather than at flush is what lets an event be written somewhere that + * outlives the process: bytes can be appended to a log, an {@link Event} cannot. + * + * @param event the event + * @return the event's JSON object, or null if it was dropped by sampling or could not be + * serialized + */ + public synchronized byte[] serialize(Event event) { + if (!Sampler.shouldSample(event.getSamplingRatio())) { + return null; + } + return writeSingleObject(new Event[]{ event }, Collections.emptyList()); + } + + /** + * Turns the evaluations counted so far into their summary events, leaving the counters empty. + *

+ * Counters live only in memory, so whatever has not been summarized is what a crash takes with + * it. Summarizing at points where the events are about to be made durable is what bounds that + * loss, and it costs nothing in accuracy: LaunchDarkly sums the counters of every summary it + * receives, so several summaries covering the same evaluations count the same as one. + * + * @return one JSON object per summary, empty if there was nothing counted + */ + public synchronized List serializeSummariesAndReset() { + if (summarizer.isEmpty()) { + return Collections.emptyList(); + } + List summaries = summarizer.getSummariesAndReset(); + List serialized = new ArrayList<>(summaries.size()); + for (EventSummarizer.EventSummary summary : summaries) { + byte[] bytes = writeSingleObject(new Event[0], Collections.singletonList(summary)); + if (bytes != null) { + serialized.add(bytes); + } + } + if (serialized.isEmpty()) { + // The counters are put back rather than dropped: they are an aggregate of evaluations the + // SDK already promised to report, and nothing else is holding them now. + summarizer.restoreTo(summaries); + } + return serialized; + } + + /** + * Serializes exactly one output event and returns the JSON object on its own. + *

+ * {@link EventOutputFormatter} only offers to write a whole request body, which is a JSON array, + * and the single-event method beside it is private. So the array is written and its brackets are + * dropped, which is sound precisely because the count it returns says how many objects are in + * there: one object means the bytes between the brackets are that object. + * + * @return the object's bytes, or null if the formatter did not write exactly one event + */ + private byte[] writeSingleObject(Event[] events, List summaries) { + ByteArrayOutputStream buffer = new ByteArrayOutputStream(INITIAL_OUTPUT_BUFFER_SIZE); + Writer writer = new BufferedWriter( + new OutputStreamWriter(buffer, Charset.forName("UTF-8")), INITIAL_OUTPUT_BUFFER_SIZE); + int written; + try { + written = formatter.writeOutputEvents(events, summaries, writer); + writer.flush(); + } catch (Exception e) { + logger.warn("Failed to serialize an event: {}", LogValues.exceptionSummary(e)); + return null; + } + if (written != 1) { + return null; + } + byte[] all = buffer.toByteArray(); + if (all.length < 3 || all[0] != '[' || all[all.length - 1] != ']') { + // Not the shape this depends on. Refusing the event is the safe reading: a frame that is + // not one JSON object would corrupt every payload it was later spliced into. + logger.warn("Serialized event was not the expected shape and will not be sent"); + return null; + } + return Arrays.copyOfRange(all, 1, all.length - 1); + } + /** * @return true if there is nothing buffered and no summary counters */ diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/AndroidEventBufferSerializationTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/AndroidEventBufferSerializationTest.java new file mode 100644 index 000000000..b7b74ab94 --- /dev/null +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/AndroidEventBufferSerializationTest.java @@ -0,0 +1,128 @@ +package com.launchdarkly.sdk.android; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +import com.launchdarkly.logging.LDLogAdapter; +import com.launchdarkly.logging.LDLogger; +import com.launchdarkly.logging.Logs; +import com.launchdarkly.sdk.LDContext; +import com.launchdarkly.sdk.LDValue; +import com.launchdarkly.sdk.internal.events.AndroidEventBuffer; +import com.launchdarkly.sdk.internal.events.Event; + +import org.junit.Test; + +import java.nio.charset.Charset; +import java.util.Collections; +import java.util.List; + +/** + * The serialization the on-disk event log is built on: one event in, one JSON object out. + *

+ * These are deliberately narrow. Everything the log does depends on a frame holding exactly one + * event object, so the assumption is checked here rather than inferred from a delivered payload. + */ +public class AndroidEventBufferSerializationTest { + private static final LDContext CONTEXT = LDContext.create("user-key"); + private static final String FLAG_KEY = "flag-key"; + + private final LDLogAdapter logAdapter = Logs.none(); + + private AndroidEventBuffer makeBuffer() { + return new AndroidEventBuffer(100, false, Collections.emptyList(), true, + LDLogger.withAdapter(logAdapter, "")); + } + + private LDValue parse(byte[] serialized) { + return LDValue.parse(new String(serialized, Charset.forName("UTF-8"))); + } + + @Test + public void customEventSerializesToASingleObject() { + byte[] serialized = makeBuffer().serialize( + new Event.Custom(1000, "an-event", CONTEXT, LDValue.of("data"), 2.5)); + + assertNotNull(serialized); + // The bytes are one object, with no array around it, so that a payload can be assembled by + // joining frames with commas. + assertEquals('{', serialized[0]); + assertEquals('}', serialized[serialized.length - 1]); + + LDValue event = parse(serialized); + assertEquals(LDValue.of("custom"), event.get("kind")); + assertEquals(LDValue.of("an-event"), event.get("key")); + assertEquals(LDValue.of(2.5), event.get("metricValue")); + } + + @Test + public void identifyEventSerializesToASingleObject() { + byte[] serialized = makeBuffer().serialize(new Event.Identify(1000, CONTEXT)); + + assertNotNull(serialized); + LDValue event = parse(serialized); + assertEquals(LDValue.of("identify"), event.get("kind")); + assertEquals(LDValue.of(CONTEXT.getKey()), event.get("context").get("key")); + } + + @Test + public void featureEventSerializesToASingleObject() { + Event.FeatureRequest event = new Event.FeatureRequest(1000, FLAG_KEY, CONTEXT, 10, 1, + LDValue.of(true), LDValue.of(false), null, null, true, null, false); + + byte[] serialized = makeBuffer().serialize(event); + + assertNotNull(serialized); + LDValue parsed = parse(serialized); + assertEquals(LDValue.of("feature"), parsed.get("kind")); + assertEquals(LDValue.of(FLAG_KEY), parsed.get("key")); + assertEquals(LDValue.of(1), parsed.get("variation")); + } + + @Test + public void summariesSerializeOneObjectEachAndLeaveTheCountersEmpty() { + AndroidEventBuffer buffer = makeBuffer(); + buffer.summarize(new Event.FeatureRequest(1000, FLAG_KEY, CONTEXT, 10, 1, LDValue.of(true), + LDValue.of(false), null, null, false, null, false)); + buffer.summarize(new Event.FeatureRequest(1001, FLAG_KEY, CONTEXT, 10, 1, LDValue.of(true), + LDValue.of(false), null, null, false, null, false)); + + List summaries = buffer.serializeSummariesAndReset(); + + assertEquals(1, summaries.size()); + LDValue summary = parse(summaries.get(0)); + assertEquals(LDValue.of("summary"), summary.get("kind")); + int counted = 0; + for (LDValue counter : summary.get("features").get(FLAG_KEY).get("counters").values()) { + counted += counter.get("count").intValue(); + } + assertEquals(2, counted); + + // Reset, so the same evaluations are not summarized again by the next call. + assertTrue(buffer.serializeSummariesAndReset().isEmpty()); + } + + @Test + public void serializingSummariesWithNothingCountedReturnsNothing() { + assertTrue(makeBuffer().serializeSummariesAndReset().isEmpty()); + } + + @Test + public void framesJoinIntoTheSamePayloadTheBufferWouldHaveSent() throws Exception { + AndroidEventBuffer staged = makeBuffer(); + byte[] first = staged.serialize(new Event.Custom(1000, "first", CONTEXT, LDValue.ofNull(), null)); + byte[] second = staged.serialize(new Event.Custom(1001, "second", CONTEXT, LDValue.ofNull(), null)); + + // Assembled the way the log assembles a request body out of its frames. + String body = "[" + new String(first, Charset.forName("UTF-8")) + "," + + new String(second, Charset.forName("UTF-8")) + "]"; + + AndroidEventBuffer drained = makeBuffer(); + drained.addFullEvent(new Event.Custom(1000, "first", CONTEXT, LDValue.ofNull(), null)); + drained.addFullEvent(new Event.Custom(1001, "second", CONTEXT, LDValue.ofNull(), null)); + String expected = new String(drained.drain().getData(), Charset.forName("UTF-8")); + + assertEquals(LDValue.parse(expected), LDValue.parse(body)); + } +} diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/AndroidEventProcessorTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/AndroidEventProcessorTest.java index 825e166a8..e0d554475 100644 --- a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/AndroidEventProcessorTest.java +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/AndroidEventProcessorTest.java @@ -69,6 +69,35 @@ public void trackedEvaluationProducesAFeatureEventAndASummary() throws Exception } } + @Test + public void flushPersistsFeatureAndSummaryEventsSynchronouslyWhileOffline() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordEvaluationEvent(CONTEXT, FLAG_KEY, FLAG_VERSION, VARIATION, + FLAG_VALUE, EvaluationReason.off(), DEFAULT_VALUE, true, null); + eventProcessor.setOffline(true); + + eventProcessor.flush(); + + EventStore reader = EventStore.create(eventsDirectory.getRoot(), MOBILE_KEY, "test", + DEFAULT_CAPACITY, logging.logger); + List persisted = reader.pendingEventPayloads(); + int featureEvents = 0; + int summaryEvents = 0; + for (byte[] payload : persisted) { + String kind = LDValue.parse(new String(payload, "UTF-8")).get("kind").stringValue(); + featureEvents += "feature".equals(kind) ? 1 : 0; + summaryEvents += "summary".equals(kind) ? 1 : 0; + } + assertEquals(1, featureEvents); + assertEquals(1, summaryEvents); + } finally { + eventProcessor.close(); + } + } + } + @Test public void identifyAndCustomEventsAreDelivered() throws Exception { try (HttpServer server = startEventsServer()) { diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorTestBase.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorTestBase.java index 48597b0d6..18fd22fab 100644 --- a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorTestBase.java +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventProcessorTestBase.java @@ -16,6 +16,7 @@ import com.launchdarkly.testhelpers.httptest.RequestInfo; import org.junit.Rule; +import org.junit.rules.TemporaryFolder; import org.junit.rules.Timeout; import java.util.ArrayList; @@ -37,9 +38,24 @@ public abstract class EventProcessorTestBase { public Timeout globalTimeout = Timeout.seconds(60); @Rule public LogCaptureRule logging = new LogCaptureRule(); + /** + * A directory of its own per test, so that the events one test persists are never found by the next. + */ + @Rule + public TemporaryFolder eventsDirectory = new TemporaryFolder(); private final IEnvironmentReporter environmentReporter = new EnvironmentReporterBuilder().build(); + /** + * The processor persists events, so it needs somewhere to put them. Tests get a real directory + * rather than a stub, which is what makes them exercise the same path an application does. + */ + protected MockPlatformState platformState() { + MockPlatformState platformState = new MockPlatformState(); + platformState.setNoBackupFilesDir(eventsDirectory.getRoot()); + return platformState; + } + protected HttpServer startEventsServer() { return HttpServer.start(Handlers.status(202)); } @@ -72,7 +88,7 @@ protected EventProcessor makeEventProcessor(HttpServer server, EventProcessorBui .serviceEndpoints(Components.serviceEndpoints().events(server.getUri())) .build(); ClientContext clientContext = ClientContextImpl.fromConfig(config, MOBILE_KEY, "", - null, null, CONTEXT, logging.logger, null, environmentReporter, null); + null, null, CONTEXT, logging.logger, platformState(), environmentReporter, null); EventProcessor eventProcessor = config.events.build(clientContext); // The processor is built offline; LDClient turns it on once initialization decides the SDK // is not in offline mode. diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventStoreMultiProcessTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventStoreMultiProcessTest.java new file mode 100644 index 000000000..69719881c --- /dev/null +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventStoreMultiProcessTest.java @@ -0,0 +1,332 @@ +package com.launchdarkly.sdk.android; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import com.launchdarkly.logging.LDLogger; +import com.launchdarkly.logging.Logs; +import com.launchdarkly.sdk.LDValue; + +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.rules.Timeout; + +import java.io.File; +import java.nio.charset.Charset; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; + +/** + * The event store's behavior when an application runs the SDK in more than one process. + *

+ * An Android application may declare components in another process, and each process gets its own copy of + * the SDK sharing nothing but the filesystem. From the store's point of view a second process is simply a + * second instance over the same directory under a different process name, which is what these tests set + * up. That is not the same thing as a second OS process - it cannot reproduce a genuinely concurrent + * {@code O_APPEND} write from separate address spaces, which needs an instrumented test - but it does + * exercise every decision the design makes about sharing: which files a process may touch, which it must + * leave alone, and what happens when both reach for the same batch. + */ +public class EventStoreMultiProcessTest { + private static final String MAIN_PROCESS = "com.example.app"; + private static final String SERVICE_PROCESS = "com.example.app:service"; + private static final int CAPACITY = 100; + + @Rule + public Timeout globalTimeout = Timeout.seconds(30); + @Rule + public TemporaryFolder tempFolder = new TemporaryFolder(); + + private final LDLogger logger = LDLogger.withAdapter(Logs.none(), ""); + /** Commits run inline, so a test never waits on another thread to make an assertion true. */ + private final Executor inline = Runnable::run; + + private File sharedDirectory() { + return new File(tempFolder.getRoot(), "events"); + } + + private EventStore storeFor(String processName) { + return storeFor(processName, CAPACITY); + } + + private EventStore storeFor(String processName, int capacity) { + return new EventStore(sharedDirectory(), processName, capacity, logger, inline); + } + + private byte[] event(String key) { + return ("{\"kind\":\"custom\",\"key\":\"" + key + "\"}").getBytes(Charset.forName("UTF-8")); + } + + /** The event keys a store can currently see, which is its own log plus every shared batch. */ + private List visibleKeys(EventStore store) { + List keys = new ArrayList<>(); + for (byte[] payload : store.pendingEventPayloads()) { + keys.add(LDValue.parse(new String(payload, Charset.forName("UTF-8"))).get("key").stringValue()); + } + return keys; + } + + private List filesNamed(String prefix) { + List matching = new ArrayList<>(); + File[] files = sharedDirectory().listFiles(); + if (files != null) { + for (File file : files) { + if (file.getName().startsWith(prefix)) { + matching.add(file); + } + } + } + return matching; + } + + // MARK: each process writes only its own log + + @Test + public void eachProcessAppendsToItsOwnLog() { + EventStore main = storeFor(MAIN_PROCESS); + EventStore service = storeFor(SERVICE_PROCESS); + + main.stage(event("main-1")); + main.commit(); + service.stage(event("service-1")); + service.commit(); + main.stage(event("main-2")); + main.commit(); + + // Two logs, not one. This is the property that makes a spliced frame impossible: only one + // process ever has the file open for writing. + assertEquals(2, filesNamed("open-").size()); + assertEquals(Arrays.asList("main-1", "main-2"), visibleKeys(main)); + assertEquals(Arrays.asList("service-1"), visibleKeys(service)); + } + + @Test + public void processNamesThatLookAlikeInAFilenameStillGetSeparateLogs() { + // Both are legal Android process names, and a naive sanitization maps both to "com_app_remote". + EventStore colon = storeFor("com.app:remote"); + EventStore dot = storeFor("com.app.remote"); + + colon.stage(event("colon")); + colon.commit(); + dot.stage(event("dot")); + dot.commit(); + + assertNotEquals(EventStore.logNameFor("com.app:remote"), EventStore.logNameFor("com.app.remote")); + assertEquals(2, filesNamed("open-").size()); + assertEquals(Arrays.asList("colon"), visibleKeys(colon)); + assertEquals(Arrays.asList("dot"), visibleKeys(dot)); + } + + @Test + public void aLogNameIsStableAcrossRunsOfTheSameProcess() { + // Stability is what lets a process find the log its previous run left behind, so it must not + // include anything that changes from run to run. + assertEquals(EventStore.logNameFor(MAIN_PROCESS), EventStore.logNameFor(MAIN_PROCESS)); + } + + @Test + public void aProcessNameThePlatformCouldNotDetermineStillYieldsAName() { + assertTrue(EventStore.logNameFor("").length() > 0); + assertTrue(EventStore.logNameFor(null).length() > 0); + assertNotEquals(EventStore.logNameFor(null), EventStore.logNameFor(MAIN_PROCESS)); + } + + // MARK: recovery respects the boundary between processes + + @Test + public void recoveryLeavesAnotherProcessOpenLogAlone() { + EventStore service = storeFor(SERVICE_PROCESS); + service.stage(event("service-1")); + service.commit(); + + // The main process starts while the service process is still alive and still appending. Its + // recovery must not take the log out from under it. + EventStore main = storeFor(MAIN_PROCESS); + main.recoverInterruptedLog(); + + assertEquals("the other process's log was taken away", 1, + filesNamed("open-" + EventStore.logNameFor(SERVICE_PROCESS)).size()); + assertTrue("another process's events were claimed as recovered", main.pendingBatches().isEmpty()); + + // And the service process can still append to it and read back both events. + service.stage(event("service-2")); + service.commit(); + assertEquals(Arrays.asList("service-1", "service-2"), visibleKeys(service)); + } + + @Test + public void aProcessRecoversTheLogItsOwnPreviousRunLeftBehind() { + EventStore firstRun = storeFor(SERVICE_PROCESS); + firstRun.stage(event("before-the-crash")); + firstRun.commit(); + // No close() - the process died without getting the chance. + + EventStore secondRun = storeFor(SERVICE_PROCESS); + secondRun.recoverInterruptedLog(); + + List recovered = secondRun.pendingBatches(); + assertEquals(1, recovered.size()); + assertEquals(1, recovered.get(0).eventCount); + assertEquals(Arrays.asList("before-the-crash"), visibleKeys(secondRun)); + assertTrue("the recovered log should no longer be open", + filesNamed("open-" + EventStore.logNameFor(SERVICE_PROCESS)).isEmpty()); + } + + @Test + public void oneProcessRecoveringDoesNotDisturbWhatAnotherHasAlreadyClosed() { + EventStore service = storeFor(SERVICE_PROCESS); + service.stage(event("service-1")); + EventStore.Batch fromService = service.closeBatch(); + assertNotNull(fromService); + + // Recovery runs when a store is constructed, before that process has recorded anything of its own, + // which is the only time it is correct to go looking for a log to close. + EventStore main = storeFor(MAIN_PROCESS); + main.recoverInterruptedLog(); + main.stage(event("main-1")); + main.commit(); + + // The service process's batch is untouched and still deliverable, by either of them. + assertEquals(1, filesNamed("ready-").size()); + assertTrue(main.pendingBatches().contains(fromService)); + } + + // MARK: closed batches are shared on purpose + + @Test + public void aBatchClosedByOneProcessIsDeliverableByAnother() { + // The point of sharing: a service process that records events and is then killed may never run + // again, so a batch only it could see would never be delivered. + EventStore service = storeFor(SERVICE_PROCESS); + service.stage(event("recorded-by-the-service")); + EventStore.Batch batch = service.closeBatch(); + assertNotNull(batch); + + EventStore main = storeFor(MAIN_PROCESS); + List pending = main.pendingBatches(); + + assertEquals(1, pending.size()); + EventStore.Batch seenByMain = pending.get(0); + // The same payload ID, which is what keeps two processes delivering this batch from counting the + // events twice upstream. + assertEquals(batch.payloadId, seenByMain.payloadId); + assertEquals(batch.eventCount, seenByMain.eventCount); + + byte[] body = main.body(seenByMain); + assertNotNull("the other process could not read the batch", body); + LDValue parsed = LDValue.parse(new String(body, Charset.forName("UTF-8"))); + assertEquals(1, parsed.size()); + assertEquals(LDValue.of("recorded-by-the-service"), parsed.get(0).get("key")); + } + + @Test + public void aBatchTheOtherProcessAlreadyDeliveredReadsAsNothingToSend() { + EventStore service = storeFor(SERVICE_PROCESS); + service.stage(event("delivered-once")); + EventStore.Batch batch = service.closeBatch(); + assertNotNull(batch); + + EventStore main = storeFor(MAIN_PROCESS); + main.remove(batch); // main delivered it first + + // The other process asking for the body gets nothing, which its caller treats as already + // delivered rather than as a failure to retry. + assertNull(service.body(batch)); + } + + @Test + public void bothProcessesRemovingTheSameBatchIsHarmless() { + EventStore service = storeFor(SERVICE_PROCESS); + service.stage(event("delivered-twice")); + EventStore.Batch batch = service.closeBatch(); + assertNotNull(batch); + + EventStore main = storeFor(MAIN_PROCESS); + main.remove(batch); + service.remove(batch); // a remove of a file that is already gone + + assertTrue(filesNamed("ready-").isEmpty()); + assertEquals(0, main.pendingBatches().size()); + assertEquals(0, service.pendingBatches().size()); + } + + @Test + public void aProcessCountsAnotherProcessPendingBatchesAgainstItsOwnCapacity() { + // Deliberately conservative, and documented as such: the alternative is cross-process accounting, + // which would need cross-process locking to be correct. + EventStore service = storeFor(SERVICE_PROCESS); + for (int i = 0; i < 3; i++) { + service.stage(event("service-" + i)); + } + assertNotNull(service.closeBatch()); + + EventStore main = storeFor(MAIN_PROCESS); + main.pendingBatches(); // this is what reconciles the count with the directory + + assertEquals(3, main.getPendingEventCount()); + } + + // MARK: the two together + + @Test + public void concurrentRecordingFromTwoProcessesLosesAndCorruptsNothing() throws Exception { + int eventsPerProcess = 200; + // Capacity well clear of what the test records, so that a dropped event means a real loss rather + // than the store correctly refusing to exceed its limit. + EventStore main = storeFor(MAIN_PROCESS, 10 * eventsPerProcess); + EventStore service = storeFor(SERVICE_PROCESS, 10 * eventsPerProcess); + + CountDownLatch startTogether = new CountDownLatch(1); + CountDownLatch finished = new CountDownLatch(2); + Thread mainThread = recordingThread(main, "main", eventsPerProcess, startTogether, finished); + Thread serviceThread = recordingThread(service, "service", eventsPerProcess, startTogether, finished); + mainThread.start(); + serviceThread.start(); + startTogether.countDown(); + assertTrue(finished.await(20, TimeUnit.SECONDS)); + + // Every event either process recorded is readable, and every frame parses - a spliced frame would + // show up as a parse failure or a missing event, because a torn frame ends the reader. + Set everything = new HashSet<>(visibleKeys(main)); + everything.addAll(visibleKeys(service)); + assertEquals(2 * eventsPerProcess, everything.size()); + for (int i = 0; i < eventsPerProcess; i++) { + assertTrue("lost main-" + i, everything.contains("main-" + i)); + assertTrue("lost service-" + i, everything.contains("service-" + i)); + } + } + + private Thread recordingThread( + EventStore store, + String prefix, + int count, + CountDownLatch startTogether, + CountDownLatch finished + ) { + return new Thread(() -> { + try { + startTogether.await(); + for (int i = 0; i < count; i++) { + store.stage(event(prefix + "-" + i)); + // Committing on every event maximizes the chance of two writes overlapping, which is + // the interleaving this is looking for. + store.commit(); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + finished.countDown(); + } + }); + } +} diff --git a/shared-test-code/src/main/java/com/launchdarkly/sdk/android/MockPlatformState.java b/shared-test-code/src/main/java/com/launchdarkly/sdk/android/MockPlatformState.java index 1b08ef95d..8d9bccdf3 100644 --- a/shared-test-code/src/main/java/com/launchdarkly/sdk/android/MockPlatformState.java +++ b/shared-test-code/src/main/java/com/launchdarkly/sdk/android/MockPlatformState.java @@ -16,6 +16,8 @@ public class MockPlatformState implements PlatformState { private volatile boolean foreground = true; private volatile boolean networkAvailable = true; + private volatile File noBackupFilesDir = null; + private volatile String processName = "test"; @Override public boolean isNetworkAvailable() { @@ -105,6 +107,28 @@ public File getCacheDir() { return new File(System.getProperty("java.io.tmpdir")); } + @Override + public File getNoBackupFilesDir() { + return noBackupFilesDir != null ? noBackupFilesDir : getCacheDir(); + } + + public void setNoBackupFilesDir(File noBackupFilesDir) { + this.noBackupFilesDir = noBackupFilesDir; + } + + @Override + public String getProcessName() { + return processName; + } + + /** + * Lets a test act as a different process of a multi-process application while sharing the same + * directory, which is the situation the on-disk event log has to be safe in. + */ + public void setProcessName(String processName) { + this.processName = processName; + } + @Override public void close() throws IOException {} } From 75a39b1c6e048900e5f166d08f26c76433c4de33 Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Fri, 11 Sep 2026 15:16:04 -0700 Subject: [PATCH 002/186] test(events): benchmark what recording an event costs on Android The counterpart of the iOS EventPersistenceBenchmark, covering the same five context shapes and three privacy settings so the two platforms can be compared rather than assumed alike. Gated on LD_EVENT_BENCH=1, which also turns on stdout and disables up-to-date checks for the test task, so an ordinary run skips all four and is otherwise unaffected. The answer is that Android does not have the problem iOS had. Its unoptimized path beats iOS's fully optimized one on every row: evaluation, summary only 1.87 us (iOS) -> 68 ns evaluation, full event 17.35 us -> 1.51 us evaluation + track 38.40 us -> 4.92 us Cost still tracks attributes written rather than the redaction decision, as on iOS, but at 48-59 ns per attribute against Codable's 1.14 us. So there is no Codable-shaped overhead to remove, because there is no Codable. Co-authored-by: Cursor --- launchdarkly-android-client-sdk/build.gradle | 11 + .../android/EventPersistenceBenchmark.java | 454 ++++++++++++++++++ 2 files changed, 465 insertions(+) create mode 100644 launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventPersistenceBenchmark.java diff --git a/launchdarkly-android-client-sdk/build.gradle b/launchdarkly-android-client-sdk/build.gradle index c9197d62d..bb0038207 100644 --- a/launchdarkly-android-client-sdk/build.gradle +++ b/launchdarkly-android-client-sdk/build.gradle @@ -31,6 +31,17 @@ android { testOptions { execution = "ANDROIDX_TEST_ORCHESTRATOR" + + // EventPersistenceBenchmark reports its measurements on stdout, which Gradle otherwise + // swallows, and it is worth re-running even when nothing has changed. Both only apply when + // LD_EVENT_BENCH is set, which is also what un-skips the benchmark, so an ordinary test run + // is unaffected. + unitTests.all { test -> + if (System.getenv("LD_EVENT_BENCH") == "1") { + test.testLogging.showStandardStreams = true + test.outputs.upToDateWhen { false } + } + } } buildFeatures { diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventPersistenceBenchmark.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventPersistenceBenchmark.java new file mode 100644 index 000000000..9479ffba5 --- /dev/null +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventPersistenceBenchmark.java @@ -0,0 +1,454 @@ +package com.launchdarkly.sdk.android; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assume.assumeTrue; + +import com.launchdarkly.logging.LDLogger; +import com.launchdarkly.logging.Logs; +import com.launchdarkly.sdk.AttributeRef; +import com.launchdarkly.sdk.ContextKind; +import com.launchdarkly.sdk.LDContext; +import com.launchdarkly.sdk.LDValue; +import com.launchdarkly.sdk.internal.events.AndroidEventBuffer; +import com.launchdarkly.sdk.internal.events.Event; + +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.List; +import java.util.function.IntConsumer; + +/** + * The Android counterpart of the iOS {@code EventPersistenceBenchmark}: what it costs to record an + * analytics event, so the encoding research can be argued from measurements on this platform rather + * than from the ones taken on iOS. + *

+ * This exists because the two platforms are not obviously comparable. iOS encodes with {@code Codable}, + * Android encodes with {@code EventOutputFormatter} in java-sdk-internal, and a JIT with escape + * analysis behaves nothing like ahead-of-time compiled Swift. The iOS finding -- that the cost is in + * the serializer's per-value machinery rather than in walking and redacting the context -- is a claim + * about {@code Codable}, and nothing about it transfers here by assumption. + *

+ * Skipped unless {@code LD_EVENT_BENCH=1}, because nothing here asserts anything a regression would + * trip and the timing loops take far longer than a unit test should: + * 

+ *   LD_EVENT_BENCH=1 ./gradlew :launchdarkly-android-client-sdk:testDebugUnitTest \
+ *       --tests '*EventPersistenceBenchmark*'
+ *
+ *

+ * Read the numbers as ratios, not absolutes. Three reasons. This runs on a desktop JVM rather + * than on ART on a device, which is the same caveat the iOS figures carry about running on a Mac. It + * is a hand-rolled timing loop rather than JMH, so it defends against the two failure modes that + * matter -- an unwarmed JIT and dead-code elimination -- and against nothing else. And the JVM's + * escape analysis can delete allocations here that a real caller would pay for, which flatters any + * row whose result is discarded. + */ +public class EventPersistenceBenchmark { + private static final String FLAG_KEY = "benchmark-flag"; + private static final int FLAG_VERSION = 7; + private static final int VARIATION = 1; + + /** + * Fixed rather than {@code System.currentTimeMillis()}, so that repeated runs encode identical + * bytes and the clock call is not charged to the encoder. The event-construction row below covers + * what the real path pays around it. + */ + private static final long CREATION_DATE = 1_700_000_000_000L; + + /** How many times each timed loop is repeated; the fastest round is reported. */ + private static final int ROUNDS = 5; + + /** + * Consumes results so that the JIT cannot prove the work is unused and delete it. Never read for + * its value. + */ + private static long blackhole; + + /** + * Holds an allocation so that escape analysis cannot scalar-replace it. Assigning to a static field + * is what makes the object genuinely exist, which is the difference between measuring an allocation + * and measuring nothing at all. + */ + private static Object escaped; + + @Rule + public TemporaryFolder eventsDirectory = new TemporaryFolder(); + + private final LDLogger logger = LDLogger.withAdapter(Logs.none(), ""); + + @Before + public void requireBenchmarking() { + assumeTrue("Set LD_EVENT_BENCH=1 to measure event recording", + "1".equals(System.getenv("LD_EVENT_BENCH"))); + } + + /** + * M1 and M2 for Android: where the time goes when one feature event is serialized, across the + * context shapes and privacy settings that change how much work redaction has to do. + *

+ * The shape of the result is the interesting part. If the cost is dominated by walking and + * redacting the context, it should grow with the number of attributes and grow again when private + * attributes force the filtering path. If it is dominated by the serializer's fixed per-event + * machinery, it should be far flatter than the attribute counts suggest. + */ + @Test + public void encodingCostByContextShape() { + List building = new ArrayList<>(); + List serializing = new ArrayList<>(); + + for (ContextShape shape : ContextShape.values()) { + final LDContext context = makeContext(shape, "benchmark-key"); + + building.add(new Measurement(shape.label, measure(200_000, i -> { + escaped = featureEvent(context, true); + }))); + + for (PrivacyShape privacy : privacyShapes()) { + final AndroidEventBuffer buffer = makeBuffer(privacy); + + byte[] sample = buffer.serialize(featureEvent(context, true)); + assertNotNull("the corpus produced an event that will not serialize", sample); + + double nanos = measure(50_000, i -> { + byte[] out = buffer.serialize(featureEvent(context, true)); + blackhole += out == null ? 0 : out.length; + }); + serializing.add(new Measurement( + shape.label + ", " + privacy.name + ", " + sample.length + " bytes", nanos)); + } + } + + report("building one feature event, no serialization", building); + report("serializing one feature event", serializing); + } + + /** + * The figures a customer actually pays, mirroring what {@link AndroidEventProcessor} does per call. + *

+ * The three rows are the three paths through {@code recordEvaluationEvent} and {@code recordCustomEvent}: + * an evaluation of an untracked flag is a counter increment; an evaluation of a tracked flag also + * serializes and stages a full event; and a {@code track} is a commit point, which additionally turns + * the counters into summary events and writes everything to the log. + *

+ * The sequence is reproduced here rather than driven through the processor itself, which would need + * an HTTP server, a scheduler and a {@code ClientContext} whose costs would swamp what is being + * measured. It is kept deliberately close to the original; if the processor changes, this should too. + */ + @Test + public void recordingCostPerCall() throws IOException { + final AndroidEventBuffer buffer = makeBuffer(noRedaction()); + final EventStore store = EventStore.create(eventsDirectory.getRoot(), "benchmark-key", + "benchmark", Integer.MAX_VALUE, logger); + final LDContext context = makeContext(ContextShape.STUB, "benchmark-key"); + + try { + List results = new ArrayList<>(); + + results.add(new Measurement("evaluation, summary only", measure(200_000, i -> { + buffer.summarize(featureEvent(context, false)); + }))); + + results.add(new Measurement("evaluation, full event", measure(50_000, i -> { + Event.FeatureRequest event = featureEvent(context, true); + buffer.summarize(event); + byte[] out = buffer.serialize(event); + if (out != null) { + store.stage(out); + blackhole += out.length; + } + }))); + + results.add(new Measurement("evaluation + track (commit point)", measure(20_000, i -> { + buffer.summarize(featureEvent(context, false)); + byte[] out = buffer.serialize( + new Event.Custom(CREATION_DATE, "benchmark-event", context, LDValue.ofNull(), null)); + if (out != null) { + store.stage(out); + blackhole += out.length; + } + for (byte[] summary : buffer.serializeSummariesAndReset()) { + store.stage(summary, true); + blackhole += summary.length; + } + store.commit(); + }))); + + report("recording, per call", results); + } finally { + store.close(); + } + } + + /** + * Whether O2 -- caching the encoded context and reusing it while the context has not changed -- + * could work here, which comes down to what the cache key would cost to check. + *

+ * Android differs from iOS in a way that matters. {@code LDContext} is an immutable object passed by + * reference, and nothing on the event path copies it, so an application that holds one context hands + * the same instance to every event and {@code equals} can settle on reference identity. iOS had to + * reach that fast path by not rebuilding a value type. The row that decides the question is the other + * one: an application that rebuilds its context between evaluations produces an equal object sharing + * nothing, and that comparison is the price of a cache hit in the worst case. + */ + @Test + public void contextComparisonCost() { + List comparisons = new ArrayList<>(); + List hashing = new ArrayList<>(); + + for (ContextShape shape : ContextShape.values()) { + final LDContext context = makeContext(shape, "benchmark-key"); + final LDContext sameInstance = context; + final LDContext rebuilt = makeContext(shape, "benchmark-key"); + + // Both must be equal, or the figures below measure an early exit rather than a comparison. + assertEquals(shape.label, context, sameInstance); + assertEquals(shape.label, context, rebuilt); + + comparisons.add(new Measurement("same instance, " + shape.label, measure(500_000, i -> { + blackhole += context.equals(sameInstance) ? 1 : 0; + }))); + comparisons.add(new Measurement("independently built, " + shape.label, measure(200_000, i -> { + blackhole += context.equals(rebuilt) ? 1 : 0; + }))); + hashing.add(new Measurement(shape.label, measure(500_000, i -> { + blackhole += context.hashCode(); + }))); + } + + report("comparing two equal contexts", comparisons); + report("hashing a context", hashing); + } + + /** + * What the deferred half of the bill costs: summaries are not serialized per evaluation but at + * commit points and deliveries, and the size of that payment depends on how many distinct flags the + * counters have accumulated. + *

+ * Counting and serializing are reported separately, because only the second is deferred -- the + * counter increment is already paid on every evaluation. + */ + @Test + public void summarySerializationCost() { + final LDContext context = makeContext(ContextShape.STUB, "benchmark-key"); + List results = new ArrayList<>(); + + for (final int flagCount : new int[] {1, 10, 50}) { + final AndroidEventBuffer buffer = makeBuffer(noRedaction()); + + double countingOnly = measure(20_000, i -> { + for (int flag = 0; flag < flagCount; flag++) { + buffer.summarize(summaryEvent(context, flag)); + } + }); + // Drained, so the counters left behind above do not inflate the first serialized round. + buffer.serializeSummariesAndReset(); + + double countingAndSerializing = measure(20_000, i -> { + for (int flag = 0; flag < flagCount; flag++) { + buffer.summarize(summaryEvent(context, flag)); + } + for (byte[] summary : buffer.serializeSummariesAndReset()) { + blackhole += summary.length; + } + }); + + results.add(new Measurement(flagCount + " flags: counting only", countingOnly)); + results.add(new Measurement(flagCount + " flags: counting + serializing", + countingAndSerializing)); + results.add(new Measurement(flagCount + " flags: serializing alone", + countingAndSerializing - countingOnly)); + } + + report("turning counters into a summary event", results); + } + + // MARK: Corpus + + /** + * The same five shapes the iOS benchmark uses, so the two sets of numbers describe the same + * contexts. + */ + private enum ContextShape { + KEY_ONLY("key only"), + STUB("stub, 9 attributes"), + WIDE("20 flat attributes"), + NESTED("nested attributes"), + MULTI("multi-context"); + + final String label; + + ContextShape(String label) { + this.label = label; + } + } + + private static LDContext makeContext(ContextShape shape, String key) { + switch (shape) { + case KEY_ONLY: + return LDContext.create(key); + case STUB: + return stubContext(key); + case WIDE: { + com.launchdarkly.sdk.ContextBuilder builder = LDContext.builder(key).name("Wide"); + for (int i = 0; i < 20; i++) { + builder.set("attribute" + i, "value" + i); + } + return builder.build(); + } + case NESTED: + return LDContext.builder(key) + .name("Nested") + .set("address", LDValue.buildObject() + .put("street", "1 Main St") + .put("city", "Springfield") + .put("geo", LDValue.buildObject().put("lat", 1.5).put("lon", -2.5).build()) + .build()) + .set("tags", LDValue.buildArray() + .add("a").add("b").add("c").add("d").add("e").build()) + .build(); + case MULTI: + return LDContext.createMulti( + stubContext(key), + LDContext.builder(ContextKind.of("device"), "device-" + key) + .set("os", LDValue.buildObject() + .put("name", "Android").put("version", 14).build()) + .build()); + default: + throw new IllegalArgumentException(shape.name()); + } + } + + /** Nine attributes, matching the shape iOS calls its stub context. */ + private static LDContext stubContext(String key) { + return LDContext.builder(key) + .name("Stub Context") + .set("firstName", "Ada") + .set("lastName", "Lovelace") + .set("email", "ada@example.com") + .set("country", "US") + .set("age", 36) + .set("score", 12.5) + .set("verified", true) + .set("plan", "enterprise") + .build(); + } + + /** The privacy settings that change what redaction has to do. */ + private static final class PrivacyShape { + final String name; + final boolean allAttributesPrivate; + final Collection privateAttributes; + + PrivacyShape(String name, boolean allAttributesPrivate, Collection privateAttributes) { + this.name = name; + this.allAttributesPrivate = allAttributesPrivate; + this.privateAttributes = privateAttributes; + } + } + + private static PrivacyShape noRedaction() { + return new PrivacyShape("no redaction", false, Collections.emptyList()); + } + + private static List privacyShapes() { + return Arrays.asList( + noRedaction(), + new PrivacyShape("1 global private", false, + Collections.singletonList(AttributeRef.fromLiteral("email"))), + new PrivacyShape("all private", true, Collections.emptyList())); + } + + private AndroidEventBuffer makeBuffer(PrivacyShape privacy) { + return new AndroidEventBuffer(Integer.MAX_VALUE, privacy.allAttributesPrivate, + privacy.privateAttributes, true, logger); + } + + private static Event.FeatureRequest featureEvent(LDContext context, boolean requireFullEvent) { + return new Event.FeatureRequest(CREATION_DATE, FLAG_KEY, context, FLAG_VERSION, VARIATION, + LDValue.of(true), LDValue.of(false), null, null, requireFullEvent, null, false); + } + + private static Event.FeatureRequest summaryEvent(LDContext context, int flagIndex) { + return new Event.FeatureRequest(CREATION_DATE, FLAG_KEY + "-" + flagIndex, context, + FLAG_VERSION, VARIATION, LDValue.of(true), LDValue.of(false), null, null, false, null, + false); + } + + // MARK: Harness + + /** + * Times {@code body}, returning nanoseconds per iteration. + *

+ * Warmed first, because a JVM measured cold reports the interpreter rather than the compiled code. + * Then run several times with the fastest round reported: a garbage collection or a descheduled + * thread can only ever make a round slower, so the minimum is the closest estimate of the work + * itself. That is also why these numbers should not be read as what a busy device would see. + */ + private double measure(int iterations, IntConsumer body) { + int warmup = Math.max(10_000, iterations / 4); + for (int i = 0; i < warmup; i++) { + body.accept(i); + } + + double best = Double.MAX_VALUE; + for (int round = 0; round < ROUNDS; round++) { + long start = System.nanoTime(); + for (int i = 0; i < iterations; i++) { + body.accept(i); + } + long elapsed = System.nanoTime() - start; + best = Math.min(best, (double) elapsed / iterations); + } + return best; + } + + private static final class Measurement { + final String name; + final double nanosPerOp; + + Measurement(String name, double nanosPerOp) { + this.name = name; + this.nanosPerOp = nanosPerOp; + } + } + + private static void report(String title, List results) { + int width = 0; + for (Measurement result : results) { + width = Math.max(width, result.name.length()); + } + StringBuilder out = new StringBuilder("\n").append(title).append('\n'); + for (Measurement result : results) { + out.append(" ").append(pad(result.name, width)).append(" ") + .append(format(result.nanosPerOp)).append('\n'); + } + System.out.print(out); + System.out.flush(); + } + + private static String pad(String value, int width) { + StringBuilder padded = new StringBuilder(value); + while (padded.length() < width) { + padded.append(' '); + } + return padded.toString(); + } + + private static String format(double nanoseconds) { + if (nanoseconds >= 1_000_000) { + return String.format("%8.2f ms", nanoseconds / 1_000_000); + } + if (nanoseconds >= 1_000) { + return String.format("%8.2f \u00b5s", nanoseconds / 1_000); + } + return String.format("%8.1f ns", nanoseconds); + } +} From 4c5217644a002fdb312a96b015b1b8ef7a759fe1 Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Tue, 15 Sep 2026 08:40:21 -0700 Subject: [PATCH 003/186] bench --- .../android/EventPersistenceBenchmark.java | 407 ++++++++++++++++++ 1 file changed, 407 insertions(+) diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventPersistenceBenchmark.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventPersistenceBenchmark.java index 9479ffba5..c84bf1cf5 100644 --- a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventPersistenceBenchmark.java +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/EventPersistenceBenchmark.java @@ -18,12 +18,20 @@ import org.junit.Test; import org.junit.rules.TemporaryFolder; +import java.io.File; import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.UUID; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicLong; import java.util.function.IntConsumer; /** @@ -63,9 +71,35 @@ public class EventPersistenceBenchmark { */ private static final long CREATION_DATE = 1_700_000_000_000L; + /** + * One already serialized event, byte for byte what the iOS benchmark stages, so the store figures on the + * two platforms are measuring the same amount of work reaching the same kind of file. + */ + private static final String SERIALIZED_EVENT = + "{\"kind\":\"feature\",\"key\":\"benchmark-flag\",\"value\":true,\"default\":false," + + "\"variation\":1,\"version\":7,\"creationDate\":1740000000000}"; + /** How many times each timed loop is repeated; the fastest round is reported. */ private static final int ROUNDS = 5; + /** + * The bulk sizes the recording figures are swept over. + *

+ * Chosen around the store's 16 KiB staging threshold rather than for being round. At the event size used + * here a hundred events stay under it and cause no write at all, a thousand cross it a handful of times, + * and ten thousand cross it often enough that the deferred column is an average rather than a report on + * one commit. The same three are used by the iOS benchmark, so the two platforms' tables line up row for + * row. + */ + private static final int[] BULK_SIZES = {100, 1_000, 10_000}; + + /** + * How many calls it takes to get this loop compiled, independent of how many the measurement then makes. + *

+ * The smallest bulk size is below it on purpose, which is exactly why this is not a share of the round. + */ + private static final int JIT_WARMUP = 10_000; + /** * Consumes results so that the JIT cannot prove the work is unused and delete it. Never read for * its value. @@ -187,6 +221,145 @@ public void recordingCostPerCall() throws IOException { } } + /** + * What a bulk of recordings costs, split by which thread pays for it. + *

+ * Every other figure here prices one call, which hides the thing that decides whether the store is + * affordable: it does not write on the thread that staged an event. Bytes accumulate to 16 KiB and then + * the write is handed to {@code commitExecutor}, so a row that times {@code stage} alone reports an array + * copy and silently omits the write it caused. + *

+ * So this records a bulk the way an application would, and prices both sides of the hand-off: + *

+ *

+ * Deliberately no {@code closeBatch} and no {@code remove}. Those run on the delivery path once per flush + * interval, so folding them into a per-event figure answers a question nobody asked. + */ + @Test + public void callerThreadCostOfABulk() { + final byte[] serialized = SERIALIZED_EVENT.getBytes(StandardCharsets.UTF_8); + + final TimingExecutor commits = new TimingExecutor(); + final EventStore[] store = new EventStore[1]; + final List toClose = new ArrayList<>(); + + final Runnable newRound = () -> { + if (store[0] != null) { + toClose.add(store[0]); + } + store[0] = new EventStore(new File(eventsDirectory.getRoot(), UUID.randomUUID().toString()), + "benchmark", Integer.MAX_VALUE, logger, commits); + // Opens the output stream outside the timed region, so the round's first event does not pay for it. + store[0].stage(serialized); + store[0].commit(); + }; + + final AndroidEventBuffer buffer = makeBuffer(noRedaction()); + final LDContext context = makeContext(ContextShape.STUB, "benchmark-key"); + + try { + // Three sizes, because the 16 KiB threshold is what decides whether there is any deferred work at + // all. A hundred events of this size never reach it: nothing is written, and the whole cost is the + // caller's. A thousand cross it about eight times and ten thousand about eighty. + for (int bulk : BULK_SIZES) { + List results = new ArrayList<>(); + + results.add(new BulkMeasurement("staging a serialized event", + measureBulk(bulk, commits, newRound, i -> store[0].stage(serialized)))); + + // The same bulk through the whole recording path, which is what puts the write in proportion. + // Serializing is by far the larger cost on this platform, and a figure for the store alone + // invites being read as though it were not. + results.add(new BulkMeasurement("evaluation of a tracked flag, end to end", + measureBulk(bulk, commits, newRound, i -> { + Event.FeatureRequest event = featureEvent(context, true); + buffer.summarize(event); + byte[] out = buffer.serialize(event); + if (out != null) { + store[0].stage(out); + blackhole += out.length; + } + }))); + + reportBulk("recording " + count(bulk) + " events, nothing delivered", results); + } + } finally { + for (EventStore closing : toClose) { + closing.close(); + } + if (store[0] != null) { + store[0].close(); + } + commits.shutdown(); + } + } + + /** + * What a commit point costs when it goes badly, rather than on average. + *

+ * {@code track} and {@code identify} commit on the caller's thread by design -- that is the durable + * barrier -- so this is the one figure here an application can observe as a stall. The mean is the number + * that gets quoted; the tail is the number a user notices. + *

+ * The counterpart iOS test runs this against a SQLite store as well, where the tail is the point: a + * database in WAL mode folds its write-ahead log back into the file every thousand pages, on whichever + * thread happens to be committing, so one unlucky caller pays for all of them. An append has no such + * thing, and these rows are what that claim is measured against. + *

+ * Two rows per size, for the two things a caller's thread actually does. {@code track} commits where it + * stands, so its tail is the write. An evaluation only stages and lets the executor write, so its tail is + * whatever the executor made it wait for -- which is the cost the deferred column cannot show. + *

+ * p99.9 only separates from the worst at ten thousand samples. Below that the index falls on the + * last sample and the two columns are the same number by construction. They are left side by side so the + * table lines up across the sweep, and so the size at which they part is visible rather than asserted. + */ + @Test + public void commitPointOutliers() throws IOException { + final byte[] serialized = SERIALIZED_EVENT.getBytes(StandardCharsets.UTF_8); + final AndroidEventBuffer buffer = makeBuffer(noRedaction()); + final LDContext context = makeContext(ContextShape.STUB, "benchmark-key"); + + // A store of its own for each size and each row, so that a log inherited from an earlier measurement + // does not charge the next one for the file it had already grown. + for (int calls : BULK_SIZES) { + final EventStore committing = EventStore.create(eventsDirectory.newFolder(), "benchmark-key", + "benchmark", Integer.MAX_VALUE, logger); + final EventStore staging = EventStore.create(eventsDirectory.newFolder(), "benchmark-key", + "benchmark", Integer.MAX_VALUE, logger); + try { + List results = new ArrayList<>(); + + results.add(new DistributionMeasurement("track, commits on this thread", + measureDistribution(calls, i -> { + committing.stage(serialized); + committing.commit(); + }))); + + results.add(new DistributionMeasurement("evaluation, write handed off", + measureDistribution(calls, i -> { + Event.FeatureRequest event = featureEvent(context, true); + buffer.summarize(event); + byte[] out = buffer.serialize(event); + if (out != null) { + staging.stage(out); + blackhole += out.length; + } + }))); + + reportDistribution("one call, " + count(calls) + " of them, on the caller's thread", results); + } finally { + committing.close(); + staging.close(); + } + } + } + /** * Whether O2 -- caching the encoded context and reusing it while the context has not changed -- * could work here, which comes down to what the cache key would cost to check. @@ -410,6 +583,183 @@ private double measure(int iterations, IntConsumer body) { return best; } + /** + * An {@link Executor} that reports how long it spent running what it was handed. + *

+ * This is what makes the deferred column a measurement rather than an inference. The store hands its + * commits here, and each one is timed on the thread that runs it, so the figure is the write itself + * rather than the difference between two numbers taken elsewhere. + */ + private static final class TimingExecutor implements Executor { + private final ExecutorService delegate = Executors.newSingleThreadExecutor(runnable -> { + Thread thread = new Thread(runnable, "benchmark-commit"); + thread.setDaemon(true); + return thread; + }); + private final AtomicLong busyNanos = new AtomicLong(); + + @Override + public void execute(Runnable command) { + delegate.execute(() -> { + long start = System.nanoTime(); + try { + command.run(); + } finally { + busyNanos.addAndGet(System.nanoTime() - start); + } + }); + } + + /** Waits for everything queued so far, so a commit still in flight is counted rather than lost. */ + void drain() { + try { + delegate.submit(() -> { }).get(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } catch (ExecutionException e) { + throw new RuntimeException(e); + } + } + + long busyNanosAndReset() { + return busyNanos.getAndSet(0); + } + + void shutdown() { + delegate.shutdownNow(); + } + } + + /** What one bulk of recordings cost, per event, split by which thread paid for it. */ + private static final class BulkCost { + final double caller; + final double deferred; + + BulkCost(double caller, double deferred) { + this.caller = caller; + this.deferred = deferred; + } + + double total() { + return caller + deferred; + } + } + + /** + * Records {@code iterations} events per round, returning the cheapest round priced on both sides of the + * hand-off. + *

+ * {@code newRound} gets a fresh store, so that a log file growing across the rounds cannot make the later + * ones look worse than the earlier ones. + */ + private BulkCost measureBulk(int iterations, TimingExecutor commits, Runnable newRound, IntConsumer record) { + // Warmed against a store of its own, because a JVM measured cold reports the interpreter rather than + // the compiled code, and because what the warmup stages should not land in a measured round. + // + // A fixed count rather than a share of the round. What the JIT needs to compile this loop has nothing + // to do with how many events the round is going to record, and scaling the warmup down with the round + // is what made the hundred-event row report the interpreter while the ten-thousand-event row did not. + newRound.run(); + for (int i = 0; i < JIT_WARMUP; i++) { + record.accept(i); + } + commits.drain(); + commits.busyNanosAndReset(); + + BulkCost best = null; + for (int round = 0; round < ROUNDS; round++) { + newRound.run(); + commits.drain(); + // Whatever setting the round up cost belongs to no round. + commits.busyNanosAndReset(); + + long start = System.nanoTime(); + for (int i = 0; i < iterations; i++) { + record.accept(i); + } + long callerNanos = System.nanoTime() - start; + + // Drained before reading, so a commit still in flight when the loop ended is charged to the + // executor rather than escaping the figure entirely. + commits.drain(); + long deferredNanos = commits.busyNanosAndReset(); + + BulkCost cost = new BulkCost((double) callerNanos / iterations, + (double) deferredNanos / iterations); + if (best == null || cost.total() < best.total()) { + best = cost; + } + } + return best; + } + + /** How long one call took, across many of them, when the tail is the point. */ + private static final class Distribution { + final double median; + final double mean; + final double p999; + final double worst; + + Distribution(double median, double mean, double p999, double worst) { + this.median = median; + this.mean = mean; + this.p999 = p999; + this.worst = worst; + } + } + + /** + * Times every call rather than the enclosing loop, so the tail survives to be reported. + *

+ * One round, and the median rather than the fastest. Every other figure here reports the cheapest of + * several rounds, on the reasoning that noise can only make a round slower -- but here an outlier is the + * measurement, and that aggregation would throw away the answer. + */ + private Distribution measureDistribution(int iterations, IntConsumer body) { + int warmup = Math.max(10_000, iterations / 4); + for (int i = 0; i < warmup; i++) { + body.accept(i); + } + + long[] samples = new long[iterations]; + for (int i = 0; i < iterations; i++) { + long start = System.nanoTime(); + body.accept(i); + samples[i] = System.nanoTime() - start; + } + + long total = 0; + for (long sample : samples) { + total += sample; + } + Arrays.sort(samples); + return new Distribution(samples[iterations / 2], + (double) total / iterations, + samples[Math.min(iterations - 1, (int) ((long) iterations * 999 / 1000))], + samples[iterations - 1]); + } + + private static final class BulkMeasurement { + final String name; + final BulkCost cost; + + BulkMeasurement(String name, BulkCost cost) { + this.name = name; + this.cost = cost; + } + } + + private static final class DistributionMeasurement { + final String name; + final Distribution distribution; + + DistributionMeasurement(String name, Distribution distribution) { + this.name = name; + this.distribution = distribution; + } + } + private static final class Measurement { final String name; final double nanosPerOp; @@ -434,6 +784,63 @@ private static void report(String title, List results) { System.out.flush(); } + /** Three columns rather than one, because the split is the whole point of the table. */ + private static void reportBulk(String title, List results) { + int width = 0; + for (BulkMeasurement result : results) { + width = Math.max(width, result.name.length()); + } + StringBuilder out = new StringBuilder("\n").append(title).append(", per event\n") + .append(" ").append(pad("", width)) + .append(" ").append(column("caller thread")) + .append(" ").append(column("commit thread")) + .append(" ").append(column("total")).append('\n'); + for (BulkMeasurement result : results) { + out.append(" ").append(pad(result.name, width)) + .append(" ").append(format(result.cost.caller)) + .append(" ").append(format(result.cost.deferred)) + .append(" ").append(format(result.cost.total())).append('\n'); + } + System.out.print(out); + System.out.flush(); + } + + private static void reportDistribution(String title, List results) { + int width = 0; + for (DistributionMeasurement result : results) { + width = Math.max(width, result.name.length()); + } + StringBuilder out = new StringBuilder("\n").append(title).append('\n') + .append(" ").append(pad("", width)) + .append(" ").append(column("median")) + .append(" ").append(column("mean")) + .append(" ").append(column("p99.9")) + .append(" ").append(column("worst")).append('\n'); + for (DistributionMeasurement result : results) { + out.append(" ").append(pad(result.name, width)) + .append(" ").append(format(result.distribution.median)) + .append(" ").append(format(result.distribution.mean)) + .append(" ").append(format(result.distribution.p999)) + .append(" ").append(format(result.distribution.worst)).append('\n'); + } + System.out.print(out); + System.out.flush(); + } + + /** Renders a count with thousands separators, so a table title reads as a quantity, not a code literal. */ + private static String count(int value) { + return String.format("%,d", value); + } + + /** Right-aligns a column heading over the fixed width {@link #format} produces. */ + private static String column(String title) { + StringBuilder padded = new StringBuilder(); + for (int i = title.length(); i < 11; i++) { + padded.append(' '); + } + return padded.append(title).toString(); + } + private static String pad(String value, int width) { StringBuilder padded = new StringBuilder(value); while (padded.length() < width) { From a66701927af72755f9ee3334db362b2ceccccefe Mon Sep 17 00:00:00 2001 From: Andrey Belonogov Date: Thu, 17 Sep 2026 14:03:42 -0700 Subject: [PATCH 004/186] renaming --- .../launchdarkly/example/MainActivity.java | 58 ++ example/src/main/res/layout/activity_main.xml | 25 +- launchdarkly-android-client-sdk/build.gradle | 11 +- .../android/EventPersistenceBenchmark.java | 896 ++++++++++++++++++ .../sdk/android/DirectEventProcessor.java | 12 +- .../sdk/android/OutboundEventBuffer.java | 6 +- 6 files changed, 997 insertions(+), 11 deletions(-) create mode 100644 launchdarkly-android-client-sdk/src/androidTest/java/com/launchdarkly/sdk/android/EventPersistenceBenchmark.java diff --git a/example/src/main/java/com/launchdarkly/example/MainActivity.java b/example/src/main/java/com/launchdarkly/example/MainActivity.java index 990d5f4a0..2e400fb03 100644 --- a/example/src/main/java/com/launchdarkly/example/MainActivity.java +++ b/example/src/main/java/com/launchdarkly/example/MainActivity.java @@ -45,6 +45,18 @@ public class MainActivity extends AppCompatActivity { private static final String DEFAULT_USER_KEY = "user key"; + /** + * The default of {@code LDConfig.Builder#events} capacity. Past this the event store is full until a + * flush empties it, so every further full event is dropped. + */ + private static final int EVENT_CAPACITY = 100; + + /** + * Enough evaluations past capacity for the average to settle, and few enough that even a slow device + * finishes the tap well inside the ANR window. + */ + private static final int OVER_CAPACITY_EVALUATIONS = 900; + private LDClient ldClient; private LDStatusListener ldStatusListener; private LDAllFlagsListener allFlagsListener; @@ -105,6 +117,7 @@ public void onCreate(Bundle savedInstanceState) { setupKillUnsentButton(); setupKillNowButton(); setupCrashNowButton(); + setupOverRefreshButton(); setupOfflineSwitch(); // Rescues the events for "Eval+Crash now" and cannot run for "Eval+Kill now", which is what // makes the pair worth pressing. @@ -310,6 +323,51 @@ private void setupCrashNowButton() { }); } + /** + * Reproduces the over-refresh pathology: a UI that re-evaluates one flag far more often than anything + * about it changed -- a recomposition or layout loop -- against a flag whose events LaunchDarkly is + * tracking. Runs on the main thread, because that is where the loop it stands in for runs. + * + *

The comparison between the two phases is the point rather than either number on its own. The first + * {@link #EVENT_CAPACITY} evaluations have room in the event store; every one after that is dropped, + * because nothing empties it until a flush. So the two phases costing the same per evaluation means the + * SDK is serializing events it then discards, which is the whole of the waste. A cost of a couple of + * microseconds instead means the flag has event tracking off, and this measured the summary counters + * rather than the thing in question. + */ + private void setupOverRefreshButton() { + Button overRefreshButton = findViewById(R.id.over_refresh_button); + overRefreshButton.setOnClickListener(v -> { + if (ldClient == null) { + return; + } + final String flagKey = "over-refresh"; + Timber.w("over-refresh eval flag=%s", flagKey); + + double filling = averageMicroseconds(flagKey, EVENT_CAPACITY); + double overCapacity = averageMicroseconds(flagKey, OVER_CAPACITY_EVALUATIONS); + + String result = String.format(Locale.US, + "Over-refresh %s\n%d filling: %.0f \u00b5s/eval\n%d over capacity: %.0f \u00b5s/eval", + flagKey, + EVENT_CAPACITY, filling, + OVER_CAPACITY_EVALUATIONS, overCapacity); + Timber.w(result); + ((TextView) findViewById(R.id.result_textView)).setText(result); + }); + } + + /** + * @return the mean wall time, in microseconds, of {@code count} evaluations of the flag + */ + private double averageMicroseconds(String flagKey, int count) { + long start = System.nanoTime(); + for (int i = 0; i < count; i++) { + ldClient.boolVariation(flagKey, false); + } + return (System.nanoTime() - start) / 1_000.0 / count; + } + private void setupIdentifyButton() { Button identify = findViewById(R.id.identify_button); identify.setOnClickListener(v -> { diff --git a/example/src/main/res/layout/activity_main.xml b/example/src/main/res/layout/activity_main.xml index 412058b0f..c2008c1b0 100644 --- a/example/src/main/res/layout/activity_main.xml +++ b/example/src/main/res/layout/activity_main.xml @@ -122,7 +122,7 @@ android:id="@+id/kill_row" android:layout_width="match_parent" android:layout_height="wrap_content" - android:layout_above="@id/identify_button" + android:layout_above="@id/over_refresh_row" android:layout_alignParentStart="true" android:layout_alignParentLeft="true" android:layout_alignParentEnd="true" @@ -151,6 +151,29 @@ android:text="Eval+Crash now" /> + + + +