From 5e9b244bebbb4a2dced69dcd077307f9c794064b Mon Sep 17 00:00:00 2001 From: Arif Burak Demiray Date: Thu, 22 Oct 2026 18:46:59 +0300 Subject: [PATCH] Gather log lines under their feature's consent, one queued batch at a time - A gathered line needs the consent of the feature its bracket names. A line of no feature needs the events and user profiles consents the app enabled, and removing a consent drops the held lines of that feature. - One sdk_logs request waits in the queue at a time, checked against the stored request, and a sent batch resumes only a delivery it held back. - A device ID change without merge queues the held lines and the tail of a stopped gather under the old ID and stops the gather. An answer requested before that change no longer arms it, while a change with merge keeps it. - A gather the server stops uploads its last lines once, under its ID. - The networking decides between a drain and a recheck under one lock, and a failing drain still rechecks. - SessionImpl logs event, location and view data after the feature check. - The test RSA key is 2048 bits (CodeQL). --- CHANGELOG.md | 4 +- .../sdk/java/internal/DefaultNetworking.java | 95 +- .../java/ly/count/sdk/java/internal/Log.java | 580 ++++++++++-- .../java/internal/ModuleConfiguration.java | 47 +- .../sdk/java/internal/ModuleDeviceIdCore.java | 16 + .../sdk/java/internal/ModuleRequests.java | 45 +- .../ly/count/sdk/java/internal/SDKCore.java | 52 +- .../count/sdk/java/internal/SDKStorage.java | 12 + .../count/sdk/java/internal/SessionImpl.java | 6 +- .../DefaultNetworkingRecheckTests.java | 79 ++ .../ly/count/sdk/java/internal/LogTests.java | 401 +++++++- .../internal/ModuleConnectionTestTests.java | 2 +- .../java/internal/SdkLogGatheringTests.java | 892 +++++++++++------- .../ly/count/sdk/java/internal/TestUtils.java | 20 + 14 files changed, 1756 insertions(+), 495 deletions(-) create mode 100644 sdk-java/src/test/java/ly/count/sdk/java/internal/DefaultNetworkingRecheckTests.java diff --git a/CHANGELOG.md b/CHANGELOG.md index b103a9b8b..02e9c9fdc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -33,10 +33,12 @@ * Added "disableSdkBehaviorSettingsUpdates()" init config method for disabling SDK behavior settings requests. * Improved request queue handling with a built-in backoff mechanism which is enabled by default. When the server takes 10 seconds or longer to answer while few requests are queued, the SDK pauses sending for 60 seconds. * Added "disableBackoffMechanism()" init config method to disable the backoff behavior. -* Added support for server requested SDK log gathering. When the server asks for it, the SDK uploads its own internal log lines, even if logging is disabled. +* Added support for server requested SDK log gathering. When the server asks for it, the SDK uploads its own internal log lines, even if logging is disabled. While consent is required, a line is uploaded only while the consent of the feature it belongs to is given. * Added support for the SDK connection test. When the server asks for it, the SDK checks that it can reach every endpoint it depends on and reports the results. * Mitigated an issue where saving the SDK storage while remote config values were changing could empty the storage file. +* Mitigated an issue where a storage file cut short by a stopped process made initialization fail. +* Mitigated an issue where a request recorded while the SDK was finishing a send could wait until the next request or session update before it was sent. * Updated JSON library version from "20250517" to "20260814". diff --git a/sdk-java/src/main/java/ly/count/sdk/java/internal/DefaultNetworking.java b/sdk-java/src/main/java/ly/count/sdk/java/internal/DefaultNetworking.java index 936f52622..10b00aab1 100644 --- a/sdk-java/src/main/java/ly/count/sdk/java/internal/DefaultNetworking.java +++ b/sdk-java/src/main/java/ly/count/sdk/java/internal/DefaultNetworking.java @@ -22,6 +22,11 @@ public class DefaultNetworking implements Networking { private Tasks tasks; private volatile boolean shutdown; private volatile boolean backedOff; + + private final Object recheckLock = new Object(); + //a check asked for while a task ran, which a drain that found the queue empty would otherwise lose: + //null for none, else whether every check asked for came from the log gathering's own transport work + private Boolean pendingRecheck = null; private boolean backendMode; private ScheduledExecutorService backoffScheduler; IStorageForRequestQueue storageForRequestQueue; @@ -56,10 +61,11 @@ public boolean isSending() { } /** - * Starts draining the request queue unless the SDK stopped, a request is being handled, there is - * no device ID yet or the backoff mechanism holds the queue back. A drain started from a thread - * marked as doing the log gathering's own transport work, which queued or sent a batch of - * gathered log lines, keeps that mark until it read the request to send. + * Starts draining the request queue unless the SDK stopped, there is no device ID yet or the + * backoff mechanism holds the queue back. While a request is being handled the check is kept + * for when the drain ends without handing a request over. A drain started from a thread marked + * as doing the log gathering's own transport work, which queued or sent a batch of gathered log + * lines, keeps that mark until it read the request to send. * * @param config configuration of the running SDK * @return whether a request is being handled @@ -67,12 +73,81 @@ public boolean isSending() { @Override public boolean check(InternalConfig config) { L.d("[Networking] [check] state: shutdown [" + shutdown + "], backed off [" + backedOff + "], tasks running [" + tasks.isRunning() + "], net running [" + tasks.isRunning() + "], device id [" + config.getDeviceId() + "]"); - if (!shutdown && !backedOff && !tasks.isRunning() && config.getDeviceId() != null) { - tasks.run(submit(config, L.isOwnTransportWork())); + if (!shutdown && !backedOff && config.getDeviceId() != null) { + boolean ownTransportWork = L.isOwnTransportWork(); + boolean drainNow; + //decided under the lock the end of a drain takes, so a check kept for that end is never kept too late + synchronized (recheckLock) { + drainNow = !tasks.isRunning(); + if (!drainNow) { + pendingRecheck = pendingRecheck == null ? ownTransportWork : pendingRecheck && ownTransportWork; + } + } + if (drainNow) { + drain(config, ownTransportWork); + } } return tasks.isRunning(); } + /** + * Starts a queue drain. When it ends without handing a request over for sending, failing + * included, it checks again if a check was asked for meanwhile, as a request stored just after the drain read the + * queue would otherwise wait for the next check. A drain that hands a request over is followed + * by the check of that send, so one request is handled at a time. + * + * @param config configuration of the running SDK + * @param followsOwnTransportWork whether the log gathering's own transport work started the drain + */ + private void drain(@Nonnull InternalConfig config, boolean followsOwnTransportWork) { + synchronized (recheckLock) { + pendingRecheck = null; + } + final Tasks.Task drain = submit(config, followsOwnTransportWork); + tasks.run(new Tasks.Task(drain.id) { + /** + * Runs the drain, a failure counting as a drain that handed nothing over. + */ + @Override + public Boolean call() { + try { + return drain.call(); + } catch (Exception e) { + L.e("[Networking] drain, failed to handle the next request, [" + e + "]"); + return false; + } + } + }, handedOver -> { + if (!Boolean.TRUE.equals(handedOver)) { + recheckIfRequested(config); + } + }); + } + + /** + * Runs a check kept while a task ran, marked as the log gathering's own transport work when + * every thread that asked for it was. + * + * @param config configuration of the running SDK + */ + private void recheckIfRequested(@Nonnull InternalConfig config) { + Boolean followsOwnTransportWork; + synchronized (recheckLock) { + followsOwnTransportWork = pendingRecheck; + pendingRecheck = null; + } + if (followsOwnTransportWork == null) { + return; + } + + L.setOwnTransportWork(followsOwnTransportWork); + try { + check(config); + } finally { + L.setOwnTransportWork(false); + } + } + /** * Sends the oldest queued request, unless networking is off or the backoff mechanism holds the * queue back. Queued requests older than the request drop age are dropped first, unsent. Once @@ -122,7 +197,7 @@ public Boolean call() throws Exception { * Hands a queued request to the transport once its owner says it is ready, or removes it when * its owner says it never will be. While the request carries a batch of gathered log lines, the * thread is marked as doing their transport work, also while the send completes, so nothing - * logged about the upload is gathered. + * logged about the upload is gathered, and once it is sent the logger may queue its next batch. * * @param config configuration of the running SDK * @param request the oldest queued request @@ -152,6 +227,9 @@ private boolean prepareAndSend(final InternalConfig config, final Request reques L.d("[Networking] Request " + request.storageId() + " sent?: " + result + ", response time [" + responseTimeMs.get() + "] ms"); if (result) { storageForRequestQueue.removeRequest(request); + if (uploadsGatheredLogs) { + L.onLogBatchSent(); + } if (!backOffIfTheServerIsSlow(config, request, responseTimeMs.get())) { check(config); } @@ -234,6 +312,9 @@ private boolean dropWithoutSending(@Nonnull InternalConfig config, @Nonnull Requ L.e("[Networking] dropWithoutSending, failed to remove request [" + request.storageId() + "], the queue drain stops here"); return false; } + if (request.params.has(Log.keySdkLogs)) { + L.onLogBatchSent(); + } Class requestOwner = request.owner(); SDKCore core = SDKCore.instance; diff --git a/sdk-java/src/main/java/ly/count/sdk/java/internal/Log.java b/sdk-java/src/main/java/ly/count/sdk/java/internal/Log.java index 089dde9c8..aee71824b 100644 --- a/sdk-java/src/main/java/ly/count/sdk/java/internal/Log.java +++ b/sdk-java/src/main/java/ly/count/sdk/java/internal/Log.java @@ -2,14 +2,20 @@ import java.util.ArrayDeque; import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.BooleanSupplier; +import java.util.function.IntSupplier; import javax.annotation.Nonnull; import javax.annotation.Nullable; import ly.count.sdk.java.Config; @@ -25,7 +31,8 @@ * requests of their own, shaped {@code {"i":gatherId,"d":dropped,"l":[{"t":ms,"l":"e","m":message}]}}. * Every level is captured, whatever is printed. Capture starts with the first line, into a * provisional buffer, since the init lines are the ones worth having: a live directive adopts the - * buffer filtered to its levels, or drops it. Nothing of it is ever stored. + * buffer filtered to its levels, or drops it. Nothing of it is ever stored. A line whose consent + * is not given when it is logged waits apart until it is, see {@link #logLineConsentFeatures(String)}. */ public class Log { @@ -46,6 +53,32 @@ public class Log { static final int gatheredCharCeiling = 128 * 1024; static final String DELIVERY_THREAD_NAME = "Countly-LogDelivery"; + //the longest bracketed name a line starts with, so finding it never scans a long line to its end + private static final int maxLogLineNameLength = 64; + //the consent a line of no feature needs, as it can quote any feature's data, such as a whole request + static final int UNATTRIBUTED_LINE_CONSENT = CoreFeature.Events.getIndex() | CoreFeature.UserProfiles.getIndex(); + + /** + * The consent a gathered line needs while consent is required, by the name its message starts + * with in brackets, as in {@code "[ModuleEvents] ..."}: the features any one of whose consent + * lets the line go. A line of any other name needs {@link #UNATTRIBUTED_LINE_CONSENT}. + */ + private static final Map logLineConsent; + + static { + Map consent = new HashMap<>(); + mapLogLineConsent(consent, CoreFeature.Events, "ModuleEvents", "Events", "EventImpl", "EventQueue", "TimedEvents"); + mapLogLineConsent(consent, CoreFeature.Sessions, "ModuleSessions", "SessionImpl"); + mapLogLineConsent(consent, CoreFeature.Views, "ModuleViews", "Views", "ViewImpl", "View"); + mapLogLineConsent(consent, CoreFeature.CrashReporting, "ModuleCrash", "Crashes", "CrashImpl"); + mapLogLineConsent(consent, CoreFeature.Location, "ModuleLocation", "Location"); + mapLogLineConsent(consent, CoreFeature.UserProfiles, "ModuleUserProfile", "UserProfile", "UserEditorImpl"); + mapLogLineConsent(consent, CoreFeature.RemoteConfig, "ModuleRemoteConfig", "RemoteConfig", "RemoteConfigValueStore"); + mapLogLineConsent(consent, CoreFeature.Feedback, "ModuleFeedback", "Feedback"); + mapLogLineConsent(consent, CoreFeature.Content, "ModuleContent", "Content", "ContentParser", "ContentRequestBuilder", "WidgetActionParser"); + logLineConsent = Collections.unmodifiableMap(consent); + } + private final LogCallback logListener; private final Config.LoggingLevel loggingLevel; //the lowest level printed: the developer level unless the SDK behavior settings override it @@ -60,6 +93,13 @@ public class Log { private final ArrayDeque gatheredLines = new ArrayDeque<>(); int droppedLogLineCount = 0; int gatheredChars = 0; + //lines whose consent was not given when they were logged, kept out of every batch until it is + private final ArrayDeque linesAwaitingConsent = new ArrayDeque<>(); + int awaitingConsentChars = 0; + //orders the lines of both lists, as the millisecond timestamps of a burst do not + private long nextLineSequence = 0; + //the last batch of a gather the server stopped, taken under the lock and queued outside it + private LogBatch pendingTailBatch = null; //the directive mirror, so capturing a line never reads the settings; the state alone is read without the lock volatile LogGatheringState logGatheringState = LogGatheringState.UNDECIDED; @@ -71,24 +111,37 @@ public class Log { private final ThreadLocal ownTransportWork = new ThreadLocal<>(); private final AtomicBoolean logDeliveryPending = new AtomicBoolean(false); private final AtomicBoolean partialLogDeliveryRequested = new AtomicBoolean(false); + //the storage IDs of the batches this logger queued, until the request queue no longer holds them + private final Set queuedLogBatchIds = ConcurrentHashMap.newKeySet(); + //a delivery left lines for the batch waiting in the request queue, so the send of that batch resumes it + private final AtomicBoolean logDeliveryAwaitsSentBatch = new AtomicBoolean(false); volatile ExecutorService logDeliveryExecutor = null; private volatile boolean sdkInitFinished = false; private volatile ConfigurationProvider configProvider = null; - private volatile BooleanSupplier consentCheck = null; + private volatile IntSupplier givenConsents = null; private volatile LogBatchQueue logBatchQueue = null; /** - * Queues one batch of gathered log lines as a request of its own. + * The request queue as the log gathering uses it: one request per batch of gathered log lines. */ interface LogBatchQueue { /** * Queues the batch. * * @param payload the {@code sdk_logs} value - * @return whether the batch was queued, {@code false} when the queue refused it + * @return the storage ID of the request carrying the batch, {@code null} when the queue refused it + */ + @Nullable + Long queueLogBatch(@Nonnull String payload); + + /** + * Whether a request still waits in the queue, until it is sent or dropped. + * + * @param requestId the storage ID {@link #queueLogBatch(String)} returned + * @return {@code true} while the request waits */ - boolean queueLogBatch(@Nonnull String payload); + boolean isQueued(@Nonnull Long requestId); } public Log(Config.LoggingLevel loggingLevel, LogCallback logListener) { @@ -194,26 +247,29 @@ private void informListener(String msg, Config.LoggingLevel level) { } /** - * Hands the logger what it needs to act on the log gathering directive, then applies the - * directive in effect. Setters rather than constructor parameters, as the logger exists before - * any of them. + * Hands the logger what it needs to act on the log gathering directive, lets the lines logged + * so far whose consent is given join the gather, then applies the directive in effect. Setters + * rather than constructor parameters, as the logger exists before any of them. * * @param configProvider the SDK behavior settings, which hold the directive - * @param consentCheck whether gathered lines may leave the device as far as consent goes + * @param givenConsents the {@link CoreFeature} indexes whose consent is given, every bit while + * no consent is required; read inside log calls, so it must never log * @param logBatchQueue queues one batch as a request of its own */ - void setLogGatheringProviders(@Nonnull ConfigurationProvider configProvider, @Nonnull BooleanSupplier consentCheck, @Nonnull LogBatchQueue logBatchQueue) { - v("[Log] setLogGatheringProviders, held lines:[" + heldLogLineCount() + "]"); - this.consentCheck = consentCheck; + void setLogGatheringProviders(@Nonnull ConfigurationProvider configProvider, @Nonnull IntSupplier givenConsents, @Nonnull LogBatchQueue logBatchQueue) { + v("[Log] setLogGatheringProviders, held lines:[" + heldLogLineCount() + "], awaiting consent:[" + awaitingConsentLineCount() + "]"); + this.givenConsents = givenConsents; this.logBatchQueue = logBatchQueue; this.configProvider = configProvider; + releaseConsentedLogLines(); applyLogGatheringDirective("init"); } /** * Applies the directive the settings hold to what is held. An armed directive adopts the held - * lines filtered to its levels, a decision against gathering drops them and stops the capture, - * and a new gather id drops the lines of the previous gather. Called after every decision, and + * lines filtered to its levels, and a new gather id drops the lines of the previous gather. A + * decision against gathering stops the capture: the lines a running gather held are queued once, + * with its id, and the lines held while undecided are dropped. Called after every decision, and * ignored once the gathering ended. * * @param source what decided, for the log line only @@ -241,7 +297,7 @@ void applyLogGatheringDirective(@Nonnull String source) { d("[Log] applyLogGatheringDirective, source:[" + source + "], state:[" + state + "], id:[" + gatherId + "], levels:[" + levels + "], batch size:[" + batchSize + "], held lines:[" + heldLogLineCount() + "]"); if (state != LogGatheringState.GATHERING) { - shutdownLogDelivery(); + shutdownLogDeliveryAfterTail(); return; } @@ -269,6 +325,101 @@ void flushGatheredLogs() { } } + /** + * Tells the logger that a request carrying a log batch left the request queue, sent or dropped + * unsent. One batch waits in the queue at a time, so while nothing is sent the lines stay in the + * capped buffer instead of filling the queue; a delivery that left lines behind for the waiting + * batch resumes now, a partial batch included. Called by the networking, on a thread marked as + * doing this feature's own transport work, and never logs. + */ + void onLogBatchSent() { + if (logDeliveryAwaitsSentBatch.compareAndSet(true, false) && isGatheringLogs()) { + scheduleLogDelivery(true); + } + } + + /** + * Stops the gather before the device ID changes without merge: queues every held line on the + * calling thread, past the one waiting batch, so each request carries the current device ID, + * then drops the lines still waiting for their consent and stops the capture, so nothing the + * next user does is gathered for the current gather. A later live directive can arm it again. + */ + void stopLogGatheringForDeviceIdChange() { + queueHeldLinesAndStop("stopLogGatheringForDeviceIdChange", true, false); + d("[Log] stopLogGatheringForDeviceIdChange, log gathering stopped before the device ID change"); + } + + /** + * Drops the held lines whose consent is no longer given, as after a consent was removed, the + * last lines of a stopped gather included, so nothing gathered for a feature goes out once its + * consent is withdrawn. They are counted as dropped. + */ + void dropLinesWithoutConsent() { + IntSupplier consents = givenConsents; + if (consents == null) { + return; + } + int given = consents.getAsInt(); + + synchronized (logBufferLock) { + Iterator iterator = gatheredLines.iterator(); + while (iterator.hasNext()) { + LogLine line = iterator.next(); + if (!isConsentGiven(line.consentFeatures, given)) { + iterator.remove(); + gatheredChars -= line.message.length(); + droppedLogLineCount++; + } + } + + LogBatch tail = pendingTailBatch; + if (tail != null) { + List kept = new ArrayList<>(tail.lines.size()); + for (LogLine line : tail.lines) { + if (isConsentGiven(line.consentFeatures, given)) { + kept.add(line); + } + } + pendingTailBatch = kept.isEmpty() ? null : new LogBatch(tail.gatherId, tail.dropped + tail.lines.size() - kept.size(), kept); + } + } + } + + /** + * Moves the lines whose consent is given by now out of the waiting list into the buffer, in + * capture order with the lines already there, and asks for a delivery when a full batch is + * held. Called when the providers are set and whenever consent is given. + */ + void releaseConsentedLogLines() { + IntSupplier consents = givenConsents; + if (consents == null) { + return; + } + int given = consents.getAsInt(); + + boolean fullBatch; + synchronized (logBufferLock) { + List released = new ArrayList<>(); + Iterator iterator = linesAwaitingConsent.iterator(); + while (iterator.hasNext()) { + LogLine line = iterator.next(); + if (isConsentGiven(line.consentFeatures, given)) { + iterator.remove(); + awaitingConsentChars -= line.message.length(); + released.add(line); + } + } + if (!released.isEmpty()) { + mergeIntoBufferLocked(released); + } + fullBatch = isGatheringLogs() && gatheredLines.size() >= logGatherBatchSize; + } + + if (fullBatch) { + scheduleLogDelivery(false); + } + } + /** * Marks the calling thread as doing this feature's own transport work, so nothing it logs * meanwhile is gathered, or every upload would report on itself. @@ -302,26 +453,58 @@ boolean isOwnTransportWork() { * @param queueHeldLines {@code true} to queue the held lines first, {@code false} to discard them */ void endLogGathering(boolean queueHeldLines) { + queueHeldLinesAndStop("endLogGathering", queueHeldLines, true); + configProvider = null; + givenConsents = null; + logBatchQueue = null; + v("[Log] endLogGathering, log gathering ended"); + } + + /** + * Queues what is held on the calling thread when asked, the held lines and then the last batch + * of a gather the server stopped, so each request carries the current device ID, then stops the + * capture and drops what is left, lines waiting for their consent included. Waits for a + * delivery in flight first, and releases the delivery thread in the same step as the stop, so a + * directive applied meanwhile either runs before it or starts a delivery thread of its own. + * + * @param caller the caller, for the log lines + * @param queueHeldLines {@code true} to queue what is held first, {@code false} to drop it + * @param endForGood {@code true} when the SDK stops, so nothing is gathered again, {@code false} + * when a later live directive may arm a gather again + */ + private void queueHeldLinesAndStop(@Nonnull String caller, boolean queueHeldLines, boolean endForGood) { + ExecutorService executor = null; Boolean previousMark = ownTransportWork.get(); setOwnTransportWork(true); try { synchronized (logDeliveryLock) { try { - if (queueHeldLines && isGatheringLogs()) { - int heldLines = heldLogLineCount(); - d("[Log] endLogGathering, queueing the [" + heldLines + "] held lines before the SDK stops"); - deliverHeldLogBatches(heldLines); + LogBatchQueue queue = logBatchQueue; + if (queueHeldLines && queue != null) { + if (isGatheringLogs()) { + int heldLines = heldLogLineCount(); + d("[Log] " + caller + ", queueing the [" + heldLines + "] held lines"); + deliverHeldLogBatches(heldLines, false); + } + LogBatch tail = takePendingTailBatch(); + if (tail != null) { + queueTail(queue, tail); + } } } catch (RuntimeException e) { - e("[Log] endLogGathering, failed to queue the held lines, [" + e + "]"); + e("[Log] " + caller + ", failed to queue the held lines, [" + e + "]"); } finally { synchronized (logBufferLock) { - logGatheringEnded = true; - logGatheringState = LogGatheringState.NOT_GATHERING; - logGatherId = null; - logGatherLevels = ModuleConfiguration.logGatheringAllLevels; - logGatherBatchSize = ModuleConfiguration.logGatheringDefaultBatchSize; - clearBufferLocked(); + if (!logGatheringEnded) { + logGatheringEnded = endForGood; + logGatheringState = LogGatheringState.NOT_GATHERING; + logGatherId = null; + logGatherLevels = ModuleConfiguration.logGatheringAllLevels; + logGatherBatchSize = ModuleConfiguration.logGatheringDefaultBatchSize; + pendingTailBatch = null; + clearBufferLocked(); + executor = takeLogDeliveryExecutor(); + } } } } @@ -329,11 +512,10 @@ void endLogGathering(boolean queueHeldLines) { setOwnTransportWork(Boolean.TRUE.equals(previousMark)); } - shutdownLogDelivery(); - configProvider = null; - consentCheck = null; - logBatchQueue = null; - v("[Log] endLogGathering, log gathering ended"); + if (executor != null) { + v("[Log] " + caller + ", stopping the log batch delivery executor"); + executor.shutdown(); + } } /** @@ -366,11 +548,22 @@ int heldLogLineCount() { } } + /** + * How many lines wait for their consent, outside the buffer. + * + * @return the number of waiting lines + */ + int awaitingConsentLineCount() { + synchronized (logBufferLock) { + return linesAwaitingConsent.size(); + } + } + /** * Holds a line for the server: every level while nothing is decided, the directive's levels * while gathering, nothing while not gathering. Lines that carry an uploaded batch and lines of a - * thread doing this feature's own transport work are skipped. A full batch is handed to the - * delivery thread. + * thread doing this feature's own transport work are skipped. A line whose consent is not given + * waits apart until it is. A full batch is handed to the delivery thread. * * @param message the line as logged * @param level the level character: e, w, i, d or v @@ -387,7 +580,9 @@ private void captureLogLine(@Nullable String message, char level) { return; } - boolean batchIsFull; + int consentFeatures = logLineConsentFeatures(msg); + boolean consentGiven = isConsentGiven(consentFeatures); + boolean batchIsFull = false; synchronized (logBufferLock) { LogGatheringState state = logGatheringState; if (state == LogGatheringState.NOT_GATHERING) { @@ -399,15 +594,88 @@ private void captureLogLine(@Nullable String message, char level) { return; } - LogLine line = new LogLine(System.currentTimeMillis(), level, trimLogMessage(msg)); - gatheredLines.addLast(line); - gatheredChars += line.message.length(); - trimBufferLocked(); - batchIsFull = gathering && gatheredLines.size() >= logGatherBatchSize; + LogLine line = new LogLine(nextLineSequence++, System.currentTimeMillis(), level, trimLogMessage(msg), consentFeatures); + if (consentGiven) { + gatheredLines.addLast(line); + gatheredChars += line.message.length(); + trimBufferLocked(); + batchIsFull = gathering && gatheredLines.size() >= logGatherBatchSize; + } else { + linesAwaitingConsent.addLast(line); + awaitingConsentChars += line.message.length(); + trimAwaitingConsentLocked(); + } } - if (batchIsFull) { + if (batchIsFull && !logDeliveryAwaitsSentBatch.get()) { scheduleLogDelivery(false); + } else if (!consentGiven && isConsentGiven(consentFeatures)) { + //given since it was read above, possibly right after the release that would have moved this line + releaseConsentedLogLines(); + } + } + + /** + * Whether a line needing the given consent may join the gather now. Says no until the + * providers are set, so the lines logged before that wait. + * + * @param consentFeatures what {@link #logLineConsentFeatures(String)} returned for the line + * @return {@code true} when the line may join the gather + */ + private boolean isConsentGiven(int consentFeatures) { + IntSupplier consents = givenConsents; + return consents != null && isConsentGiven(consentFeatures, consents.getAsInt()); + } + + /** + * Whether a line needing the given consent may join the gather. + * + * @param consentFeatures the features any one of whose consent lets the line go, 0 for + * {@link #UNATTRIBUTED_LINE_CONSENT} + * @param given the {@link CoreFeature} indexes whose consent is given + * @return {@code true} when the line may join the gather + */ + static boolean isConsentGiven(int consentFeatures, int given) { + if (consentFeatures == 0) { + return (given & UNATTRIBUTED_LINE_CONSENT) == UNATTRIBUTED_LINE_CONSENT; + } + return (given & consentFeatures) != 0; + } + + /** + * Finds the consent a log line needs from the name it starts with in brackets, up to the + * closing bracket or a comma. + * + * @param message the line as logged + * @return the {@link CoreFeature} indexes any one of whose consent lets the line go, 0 for a + * line that needs {@link #UNATTRIBUTED_LINE_CONSENT} + */ + static int logLineConsentFeatures(@Nonnull String message) { + if (message.isEmpty() || message.charAt(0) != '[') { + return 0; + } + int limit = Math.min(message.length(), maxLogLineNameLength + 2); + int end = 1; + while (end < limit && message.charAt(end) != ']' && message.charAt(end) != ',') { + end++; + } + if (end == 1 || end == limit) { + return 0; + } + Integer features = logLineConsent.get(message.substring(1, end)); + return features == null ? 0 : features; + } + + /** + * Maps the bracketed names a feature's log lines start with to the consent of that feature. + * + * @param consent the map to fill + * @param feature the feature whose consent the lines need + * @param names the names, without the brackets + */ + private static void mapLogLineConsent(@Nonnull Map consent, @Nonnull CoreFeature feature, @Nonnull String... names) { + for (String name : names) { + consent.put(name, feature.getIndex()); } } @@ -444,6 +712,10 @@ private boolean adoptLogGatheringDirective(@Nonnull LogGatheringState newState, switch (newState) { case NOT_GATHERING: + //the server accepts the last lines of a stopped gather under its id; lines held while undecided are dropped + if (wasGathering && previousGatherId != null && !gatheredLines.isEmpty()) { + pendingTailBatch = new LogBatch(previousGatherId, droppedLogLineCount, new ArrayList<>(gatheredLines)); + } clearBufferLocked(); break; case GATHERING: @@ -506,7 +778,7 @@ private void runLogDelivery() { if (logGatheringEnded) { return; } - deliverHeldLogBatches(includePartialBatch ? heldLogLineCount() : 0); + deliverHeldLogBatches(includePartialBatch ? heldLogLineCount() : 0, true); } } catch (RuntimeException e) { e("[Log] runLogDelivery, failed to upload the gathered log lines, [" + e + "]"); @@ -516,12 +788,14 @@ private void runLogDelivery() { /** * Queues the held lines batch by batch, oldest first, until nothing is left to queue or the * queue refuses a batch: every full batch, and within the given budget also a batch that is not - * full. Holds everything while tracking is off or no consent is given, as gathered lines quote - * event keys and whole requests. The caller holds the delivery lock and has marked its thread. + * full. Holds everything while tracking is off. The buffer holds only lines whose consent was + * given. The caller holds the delivery lock and has marked its thread. * * @param partialLineBudget how many lines may still go out in batches that are not full + * @param oneBatchAtATime whether to queue nothing while a batch this logger queued waits in the + * request queue, and at most one batch otherwise */ - private void deliverHeldLogBatches(int partialLineBudget) { + private void deliverHeldLogBatches(int partialLineBudget, boolean oneBatchAtATime) { LogBatchQueue queue = logBatchQueue; ConfigurationProvider provider = configProvider; if (queue == null || provider == null) { @@ -534,11 +808,16 @@ private void deliverHeldLogBatches(int partialLineBudget) { d("[Log] deliverHeldLogBatches, tracking is disabled, keeping the gathered lines buffered"); return; } - - BooleanSupplier consent = consentCheck; - if (consent != null && !consent.getAsBoolean()) { - d("[Log] deliverHeldLogBatches, no consent is given, keeping the gathered lines buffered"); - return; + if (oneBatchAtATime) { + if (forgetLogBatchesNoLongerQueued(queue) > 0) { + if (hasLinesToDeliver(budget > 0)) { + logDeliveryAwaitsSentBatch.set(true); + d("[Log] deliverHeldLogBatches, a log batch is still waiting in the request queue, delivering the rest once it is sent"); + } + return; + } + //no batch waits any more, as when the queue dropped it unsent, so a resume asked for earlier is moot + logDeliveryAwaitsSentBatch.set(false); } LogBatch batch = takeLogBatch(budget > 0); @@ -549,7 +828,7 @@ private void deliverHeldLogBatches(int partialLineBudget) { String payload = buildLogBatchPayload(batch.gatherId, batch.dropped, batch.lines); d("[Log] deliverHeldLogBatches, uploading a gathered log batch of [" + batch.lines.size() + "] lines, [" + payload.length() + "] characters"); - if (!queue.queueLogBatch(payload)) { + if (!queueLogBatch(queue, payload)) { d("[Log] deliverHeldLogBatches, the request queue refused the batch, keeping its lines buffered"); restoreLogBatch(batch); return; @@ -557,6 +836,107 @@ private void deliverHeldLogBatches(int partialLineBudget) { } } + /** + * Hands one batch to the request queue and keeps the storage ID of its request, so the next + * batch waits while the queue holds it. + * + * @param queue the request queue + * @param payload the batch + * @return whether the queue took the batch + */ + private boolean queueLogBatch(@Nonnull LogBatchQueue queue, @Nonnull String payload) { + Long requestId = queue.queueLogBatch(payload); + if (requestId == null) { + return false; + } + queuedLogBatchIds.add(requestId); + return true; + } + + /** + * Forgets the batches the request queue no longer holds, sent or dropped, which also covers a + * request dropped without a send, as when the queue is over its size. + * + * @param queue the request queue + * @return how many batches of this logger the queue still holds + */ + private int forgetLogBatchesNoLongerQueued(@Nonnull LogBatchQueue queue) { + queuedLogBatchIds.removeIf(requestId -> !queue.isQueued(requestId)); + return queuedLogBatchIds.size(); + } + + /** + * Whether a delivery would queue anything now. + * + * @param includePartialBatch whether a batch that is not full counts + * @return {@code true} when a batch could be taken + */ + private boolean hasLinesToDeliver(boolean includePartialBatch) { + synchronized (logBufferLock) { + return canTakeLogBatchLocked(includePartialBatch); + } + } + + /** + * Whether a batch can be taken out of the buffer. The caller holds the buffer lock. + * + * @param includePartialBatch whether a batch that is not full counts + * @return {@code true} when a batch can be taken + */ + private boolean canTakeLogBatchLocked(boolean includePartialBatch) { + if (logGatheringState != LogGatheringState.GATHERING || logGatherId == null || gatheredLines.isEmpty()) { + return false; + } + return includePartialBatch || gatheredLines.size() >= logGatherBatchSize; + } + + /** + * Queues the last batch of a gather the server stopped, once, from the delivery thread, as the + * decision that stopped the gather may hold a lock the request queue needs. + */ + private void queueTailBatch() { + synchronized (logDeliveryLock) { + LogBatch tail = takePendingTailBatch(); + LogBatchQueue queue = logBatchQueue; + if (tail == null || queue == null || !sdkInitFinished || logGatheringEnded) { + return; + } + try { + queueTail(queue, tail); + } catch (RuntimeException e) { + e("[Log] queueTailBatch, failed to upload the last batch, [" + e + "]"); + } + } + } + + /** + * Takes the last batch of a gather the server stopped, if one waits. + * + * @return the batch, {@code null} for none + */ + @Nullable + private LogBatch takePendingTailBatch() { + synchronized (logBufferLock) { + LogBatch tail = pendingTailBatch; + pendingTailBatch = null; + return tail; + } + } + + /** + * Queues the last batch of a stopped gather, with its id. Its lines are gone if the queue + * refuses it. The caller holds the delivery lock. + * + * @param queue the request queue + * @param tail the batch + */ + private void queueTail(@Nonnull LogBatchQueue queue, @Nonnull LogBatch tail) { + d("[Log] queueTail, uploading the last [" + tail.lines.size() + "] lines of the stopped gather"); + if (!queueLogBatch(queue, buildLogBatchPayload(tail.gatherId, tail.dropped, tail.lines))) { + d("[Log] queueTail, the request queue refused the last batch, its lines are dropped"); + } + } + /** * Takes up to one batch of the oldest lines out of the buffer, with the lines dropped since the * last batch, which are reported once, with the batch that follows the loss. @@ -567,10 +947,7 @@ private void deliverHeldLogBatches(int partialLineBudget) { @Nullable private LogBatch takeLogBatch(boolean includePartialBatch) { synchronized (logBufferLock) { - if (logGatheringState != LogGatheringState.GATHERING || logGatherId == null || gatheredLines.isEmpty()) { - return null; - } - if (!includePartialBatch && gatheredLines.size() < logGatherBatchSize) { + if (!canTakeLogBatchLocked(includePartialBatch)) { return null; } @@ -634,19 +1011,37 @@ private void ensureLogDeliveryExecutor() { } /** - * Releases the delivery thread. A delivery already queued still runs, and finds nothing to do - * once the gathering is off. + * Releases the delivery thread once it queued the last batch of a gather the server stopped, + * if there is one. The batch is queued on the delivery thread, behind any delivery already + * waiting there, which finds nothing to do once the gathering is off. */ - private void shutdownLogDelivery() { - ExecutorService executor; - synchronized (logDeliveryExecutorLock) { - executor = logDeliveryExecutor; - logDeliveryExecutor = null; + private void shutdownLogDeliveryAfterTail() { + ExecutorService executor = takeLogDeliveryExecutor(); + if (executor == null) { + takePendingTailBatch(); + return; } - if (executor != null) { - v("[Log] shutdownLogDelivery, stopping the log batch delivery executor"); - executor.shutdown(); + try { + executor.execute(this::queueTailBatch); + } catch (RejectedExecutionException e) { + takePendingTailBatch(); + } + v("[Log] shutdownLogDeliveryAfterTail, stopping the log batch delivery executor"); + executor.shutdown(); + } + + /** + * Takes the delivery thread's executor away, so no delivery is scheduled on it any more. + * + * @return the executor, {@code null} when there was none + */ + @Nullable + private ExecutorService takeLogDeliveryExecutor() { + synchronized (logDeliveryExecutorLock) { + ExecutorService executor = logDeliveryExecutor; + logDeliveryExecutor = null; + return executor; } } @@ -662,8 +1057,37 @@ private void trimBufferLocked() { } /** - * Drops the held lines of levels the directive does not ask for. They are not counted as - * dropped, as they were never wanted. + * Enforces both ceilings on the lines waiting for their consent by dropping the oldest. They + * are not counted into {@code d}, as a line waiting for its consent has not joined the gather. + */ + private void trimAwaitingConsentLocked() { + while (linesAwaitingConsent.size() > ModuleConfiguration.logGatheringMaxBufferedLines + || (awaitingConsentChars > gatheredCharCeiling && linesAwaitingConsent.size() > 1)) { + awaitingConsentChars -= linesAwaitingConsent.removeFirst().message.length(); + } + } + + /** + * Adds lines whose consent was given to the buffer, which stays in capture order. + * + * @param released the lines, in capture order + */ + private void mergeIntoBufferLocked(@Nonnull List released) { + List merged = new ArrayList<>(gatheredLines); + merged.addAll(released); + merged.sort(Comparator.comparingLong(line -> line.sequence)); + for (LogLine line : released) { + gatheredChars += line.message.length(); + } + + gatheredLines.clear(); + gatheredLines.addAll(merged); + trimBufferLocked(); + } + + /** + * Drops the lines of levels the directive does not ask for, held or waiting for their consent. + * They are not counted as dropped, as they were never wanted. */ private void filterBufferLocked() { Iterator iterator = gatheredLines.iterator(); @@ -674,14 +1098,25 @@ private void filterBufferLocked() { iterator.remove(); } } + + Iterator awaiting = linesAwaitingConsent.iterator(); + while (awaiting.hasNext()) { + LogLine line = awaiting.next(); + if (logGatherLevels.indexOf(line.level) < 0) { + awaitingConsentChars -= line.message.length(); + awaiting.remove(); + } + } } /** - * Empties the buffer and forgets the drop count. + * Empties the buffer and the lines waiting for their consent, and forgets the drop count. */ private void clearBufferLocked() { gatheredLines.clear(); gatheredChars = 0; + linesAwaitingConsent.clear(); + awaitingConsentChars = 0; droppedLogLineCount = 0; } @@ -735,21 +1170,28 @@ static String trimLogMessage(@Nonnull String message) { * One captured line. */ static final class LogLine { + final long sequence; final long timestamp; final char level; final String message; + final int consentFeatures; /** * Holds one line. * + * @param sequence its place in capture order * @param timestamp when it was logged, in milliseconds * @param level the level character * @param message the message, already cut + * @param consentFeatures the {@link CoreFeature} indexes any one of whose consent lets the + * line go, 0 for {@link #UNATTRIBUTED_LINE_CONSENT} */ - LogLine(long timestamp, char level, @Nonnull String message) { + LogLine(long sequence, long timestamp, char level, @Nonnull String message, int consentFeatures) { + this.sequence = sequence; this.timestamp = timestamp; this.level = level; this.message = message; + this.consentFeatures = consentFeatures; } } diff --git a/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleConfiguration.java b/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleConfiguration.java index 8c796a835..2e45936ca 100644 --- a/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleConfiguration.java +++ b/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleConfiguration.java @@ -112,6 +112,11 @@ public class ModuleConfiguration extends ModuleBase implements ConfigurationProv private boolean serverConfigRequestsDisabled = false; private volatile boolean running = false; + /** + * How many times the device ID changed without merge, written under this module's lock, so a + * response requested before such a change can tell its gather was armed for the previous user. + */ + private volatile int deviceIdChangesWithoutMerge = 0; private volatile ConnectionTestListener connectionTestListener = null; //resolved values; the constant seeds are here, the ones read from the developer configuration are in seedFromDeveloperConfig @@ -236,7 +241,7 @@ public void init(InternalConfig config) { loadConfigFromStorage(config.getSdkBehaviorSettings()); } - updateConfigVariables(config, null); + updateConfigVariables(config, null, true); running = true; config.configProvider = this; } @@ -486,19 +491,26 @@ synchronized void saveAndStoreDownloadedConfig(@Nonnull JSONObject config) { } /** - * Applies the merged settings to the resolved values, decides the log gathering directive of a - * live response and has the logger apply it, and restarts the refresh timer when its interval changed. + * Applies the stored settings to the resolved values, decides the log gathering directive of a + * live response requested since the last device ID change without merge and has the logger apply + * it, and restarts the refresh timer when its interval changed. A response requested before that + * change leaves the directive as it is, as its gather was armed for the previous user. * * @param config configuration of the running SDK * @param serverResponse the live response being applied, {@code null} at init, where only stored or provided settings exist + * @param forCurrentDeviceId whether the response was requested since the last device ID change without merge * @return whether a resolved value changed */ - synchronized boolean updateConfigVariables(@Nonnull InternalConfig config, @Nullable JSONObject serverResponse) { + synchronized boolean updateConfigVariables(@Nonnull InternalConfig config, @Nullable JSONObject serverResponse, boolean forCurrentDeviceId) { L.v("[ModuleConfiguration] updateConfigVariables, from server response:[" + (serverResponse != null) + "]"); - //read off the live response itself, so it applies even when its settings object was rejected - readLogGatheringDirective(serverResponse); - L.applyLogGatheringDirective(serverResponse == null ? "init" : "server response"); + if (forCurrentDeviceId) { + //read off the live response itself, so it applies even when its settings object was rejected + readLogGatheringDirective(serverResponse); + L.applyLogGatheringDirective(serverResponse == null ? "init" : "server response"); + } else { + L.d("[ModuleConfiguration] updateConfigVariables, the response was requested for a previous device ID, its log gathering directive is ignored"); + } if (latestRetrievedConfiguration == null) { return false; @@ -800,6 +812,19 @@ private void setLogGatheringOff() { currentVLogGathering = LogGatheringDirective.NOT_GATHERING; } + /** + * Stops log gathering before the device ID changes without merge: decides the directive off, + * then has the logger queue what it holds under the current device ID and stop, so nothing the + * next user does is gathered until a live response for the new device ID arms a gather. + */ + void stopLogGatheringForDeviceIdChange() { + synchronized (this) { + deviceIdChangesWithoutMerge++; + setLogGatheringOff(); + } + L.stopLogGatheringForDeviceIdChange(); + } + /** * Keeps only the level characters this SDK knows, lower cased, in the order the server sent them * and without duplicates. Falls back to every level when nothing usable is left. @@ -876,9 +901,10 @@ void fetchConfigFromServer(@Nonnull InternalConfig config) { String requestData = ModuleRequests.prepareRequiredParams(config).add("method", "sc").toString(); Transport transport = config.sdk.networking.getTransport(); final long fetchStartNs = System.nanoTime(); + final int requestedAfterDeviceIdChanges = deviceIdChangesWithoutMerge; config.immediateRequestGenerator.createImmediateRequestMaker().doWork(requestData, "/o/sdk?", transport, false, true, - response -> onServerConfigResponse(config, response, fetchStartNs), L); + response -> onServerConfigResponse(config, response, fetchStartNs, requestedAfterDeviceIdChanges), L); } catch (Exception e) { L.e("[ModuleConfiguration] fetchConfigFromServer, failed to request the configuration, [" + e + "]"); decideLogGatheringOffIfUndecided("the server config request failed"); @@ -892,8 +918,9 @@ void fetchConfigFromServer(@Nonnull InternalConfig config) { * @param config configuration of the running SDK * @param response the response, {@code null} when the request failed * @param fetchStartNs {@link System#nanoTime()} just before the request was made + * @param requestedAfterDeviceIdChanges how many device ID changes without merge happened before the request was made */ - private void onServerConfigResponse(@Nonnull InternalConfig config, @Nullable JSONObject response, long fetchStartNs) { + private void onServerConfigResponse(@Nonnull InternalConfig config, @Nullable JSONObject response, long fetchStartNs, int requestedAfterDeviceIdChanges) { if (!running) { L.d("[ModuleConfiguration] onServerConfigResponse, the SDK was stopped before the response arrived, ignoring it"); return; @@ -917,7 +944,7 @@ private void onServerConfigResponse(@Nonnull InternalConfig config, @Nullable JS return; } saveAndStoreDownloadedConfig(response); - changed = updateConfigVariables(config, response); + changed = updateConfigVariables(config, response, requestedAfterDeviceIdChanges == deviceIdChangesWithoutMerge); } if (changed && config.sdk != null) { diff --git a/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleDeviceIdCore.java b/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleDeviceIdCore.java index e77f93c54..fc9591aa4 100644 --- a/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleDeviceIdCore.java +++ b/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleDeviceIdCore.java @@ -149,6 +149,10 @@ protected void changeDeviceIdInternal(String id, DeviceIdType type, boolean with return; } + if (!withMerge) { + stopLogGatheringForDeviceIdChange(); + } + internalConfig.storageProvider.setDeviceIdType(type.name()); internalConfig.storageProvider.setDeviceID(id); internalConfig.setDeviceId(new Config.DID(type.index, id)); @@ -156,6 +160,18 @@ protected void changeDeviceIdInternal(String id, DeviceIdType type, boolean with SDKCore.instance.notifyModulesDeviceIdChanged(old.id, withMerge); } + /** + * Has the gathered SDK log lines of the current user queued under the current device ID and + * the gather stopped, as the device ID is about to change without merge. + */ + private void stopLogGatheringForDeviceIdChange() { + SDKCore core = SDKCore.instance; + ModuleConfiguration configuration = core == null ? null : core.module(ModuleConfiguration.class); + if (configuration != null) { + configuration.stopLogGatheringForDeviceIdChange(); + } + } + /** * Gets id of the strategy supplied. In case strategy is not available, returns a fallback strategy. * In case strategy is available but id cannot be acquired right now, returns null. diff --git a/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleRequests.java b/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleRequests.java index 752371a87..9d0479020 100644 --- a/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleRequests.java +++ b/sdk-java/src/main/java/ly/count/sdk/java/internal/ModuleRequests.java @@ -9,6 +9,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import javax.annotation.Nonnull; +import javax.annotation.Nullable; /** * Centralized place for all requests construction & handling. @@ -18,7 +19,6 @@ public class ModuleRequests extends ModuleBase { //how long queueing a log batch waits for the storage, which never answers once it is shut down static final long SDK_LOGS_STORE_TIMEOUT_MS = 10_000L; - private static final int MAX_CLOCK_WAIT_MS = 20; private static Params metrics; @@ -293,37 +293,36 @@ static boolean pushWithinQueueLimit(@Nonnull final InternalConfig config, @Nonnu /** * Queues one batch of gathered SDK log lines as a request of its own, the batch as its * {@code sdk_logs} parameter next to the required ones, through the checks of - * {@link #pushAsync(InternalConfig, Request)}, and waits until it is stored. Then waits for the - * clock to pass the ID of the request, as batches are queued back to back and a request ID is - * unique among the last few drawn only. When the wait times out or is interrupted, the store is - * cancelled, as a store left queued would still write the request after its lines were kept for - * another batch. + * {@link #pushAsync(InternalConfig, Request)}, and waits until it is stored. When the wait times + * out or is interrupted, the store is cancelled, as a store left queued would still write the + * request after its lines were kept for another batch. * * @param config configuration of the running SDK * @param payload the batch, as JSON text - * @return whether the request was stored, {@code false} when it was refused, as while tracking - * is off, could not be stored or its store was cancelled + * @return the storage ID of the stored request, {@code null} when it was refused, as while + * tracking is off, could not be stored or its store was cancelled */ - static boolean pushSdkLogs(@Nonnull InternalConfig config, @Nonnull String payload) { + @Nullable + static Long pushSdkLogs(@Nonnull InternalConfig config, @Nonnull String payload) { Request request = nonSessionRequest(config); request.params.add(Log.keySdkLogs, payload); Future stored = null; + boolean queued; try { stored = pushAsync(config, request); - return stored != null && Boolean.TRUE.equals(stored.get(SDK_LOGS_STORE_TIMEOUT_MS, TimeUnit.MILLISECONDS)); + queued = stored != null && Boolean.TRUE.equals(stored.get(SDK_LOGS_STORE_TIMEOUT_MS, TimeUnit.MILLISECONDS)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); config.getLogger().w("[ModuleRequests] pushSdkLogs, interrupted while storing request " + request.storageId()); - return !cancelStore(config, request, stored); + queued = !cancelStore(config, request, stored); } catch (TimeoutException e) { config.getLogger().w("[ModuleRequests] pushSdkLogs, storing request " + request.storageId() + " took over [" + SDK_LOGS_STORE_TIMEOUT_MS + "] ms"); - return !cancelStore(config, request, stored); + queued = !cancelStore(config, request, stored); } catch (ExecutionException | RuntimeException e) { config.getLogger().w("[ModuleRequests] pushSdkLogs, failed to store request " + request.storageId() + ", [" + e + "]"); - return false; - } finally { - waitForClockToPass(request.storageId()); + queued = false; } + return queued ? request.storageId() : null; } /** @@ -345,22 +344,6 @@ private static boolean cancelStore(@Nonnull InternalConfig config, @Nonnull Requ return false; } - /** - * Waits until the clock is past the given time, for a few milliseconds at most. - * - * @param timestampMs a time in milliseconds - */ - private static void waitForClockToPass(long timestampMs) { - for (int waited = 0; waited < MAX_CLOCK_WAIT_MS && System.currentTimeMillis() <= timestampMs; waited++) { - try { - Thread.sleep(1); - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; - } - } - } - /** * Drops the oldest stored requests while the queue holds more than the request queue size of the * SDK behavior settings. Without a size from the settings the queue has no limit. Must run on the diff --git a/sdk-java/src/main/java/ly/count/sdk/java/internal/SDKCore.java b/sdk-java/src/main/java/ly/count/sdk/java/internal/SDKCore.java index 7320d6ff3..296a15aa9 100644 --- a/sdk-java/src/main/java/ly/count/sdk/java/internal/SDKCore.java +++ b/sdk-java/src/main/java/ly/count/sdk/java/internal/SDKCore.java @@ -185,6 +185,7 @@ public void onConsent(final int consent) { } consents = consents | (consent & config.getFeatures1()); + L.releaseConsentedLogLines(); for (CoreFeature feature : moduleMappings.keySet()) { ModuleBase existing = module(moduleMappings.get(feature)); @@ -224,6 +225,7 @@ public void onConsentRemoval(final InternalConfig config, int noConsent) { } consents = consents & ~noConsent; + L.dropLinesWithoutConsent(); for (CoreFeature feature : moduleMappings.keySet()) { ModuleBase existing = module(moduleMappings.get(feature)); @@ -442,7 +444,7 @@ public void init(final InternalConfig givenConfig) { moduleConfiguration.init(config); applyLoggingSetting(config, moduleConfiguration); applyConsentRequirementAtInit(config, moduleConfiguration.getConsentRequired()); - L.setLogGatheringProviders(moduleConfiguration, this::isAnyConsentGiven, this::queueGatheredLogBatch); + L.setLogGatheringProviders(moduleConfiguration, this::givenLogLineConsents, gatheredLogBatchQueue); //create internal timer synchronized (lockTimer) { @@ -565,26 +567,48 @@ private void initFinished(final InternalConfig config) { } /** - * Whether gathered log lines may leave the device as far as consent goes: always while no - * consent is required, otherwise once any consent is given. + * The features whose consent lets a gathered log line go: every feature while no consent is + * required, otherwise the ones consent was given for. Read inside log calls, so it never logs. * - * @return {@code true} when the lines may be uploaded + * @return the {@link CoreFeature} indexes, every bit while no consent is required */ - private boolean isAnyConsentGiven() { + private int givenLogLineConsents() { InternalConfig current = config; - return current != null && (!current.requiresConsent() || consents != 0); + if (current == null) { + return 0; + } + if (!current.requiresConsent()) { + return ~0; + } + int given = consents; + //a feature the app never enabled holds no data and can never get consent, so it holds no line back + return given == 0 ? 0 : given | (~current.getFeatures1() & Log.UNATTRIBUTED_LINE_CONSENT); } /** - * Queues one batch of gathered log lines as a request of its own. - * - * @param payload the batch - * @return whether the request was stored + * The request queue as the log gathering uses it: one request per batch of gathered log lines, + * waiting until it is sent or dropped. */ - private boolean queueGatheredLogBatch(@Nonnull String payload) { - InternalConfig current = config; - return current != null && ModuleRequests.pushSdkLogs(current, payload); - } + private final Log.LogBatchQueue gatheredLogBatchQueue = new Log.LogBatchQueue() { + /** + * Stores the batch as a request of its own, unless the SDK stopped. + */ + @Nullable + @Override + public Long queueLogBatch(@Nonnull String payload) { + InternalConfig current = config; + return current == null ? null : ModuleRequests.pushSdkLogs(current, payload); + } + + /** + * Whether the request file is still stored, which it is until the request is sent or dropped. + */ + @Override + public boolean isQueued(@Nonnull Long requestId) { + InternalConfig current = config; + return current != null && sdkStorage.storableExists(current, Request.getStoragePrefix(), requestId); + } + }; /** * Makes the consent requirement the SDK behavior settings resolved the one every module reads diff --git a/sdk-java/src/main/java/ly/count/sdk/java/internal/SDKStorage.java b/sdk-java/src/main/java/ly/count/sdk/java/internal/SDKStorage.java index f4f2bf7ee..8bed32622 100644 --- a/sdk-java/src/main/java/ly/count/sdk/java/internal/SDKStorage.java +++ b/sdk-java/src/main/java/ly/count/sdk/java/internal/SDKStorage.java @@ -277,6 +277,18 @@ private String[] getFileList(InternalConfig config, Predicate fileValidati return fileNames.toArray(ret); } + /** + * Whether the file of a storable exists, as a stored request does until it is sent or dropped. + * + * @param config configuration of the running SDK + * @param prefix the storage prefix of the storable + * @param id the storage ID of the storable + * @return {@code true} while the file exists + */ + boolean storableExists(InternalConfig config, String prefix, Long id) { + return new File(createFileFullPath(config, getName(prefix, id.toString()))).exists(); + } + public List storableList(InternalConfig config, String prefix, int slice) { if (Utils.isEmptyOrNull(prefix)) { L.e("[SDKStorage] Cannot get list of ids without prefix"); diff --git a/sdk-java/src/main/java/ly/count/sdk/java/internal/SessionImpl.java b/sdk-java/src/main/java/ly/count/sdk/java/internal/SessionImpl.java index 547a5b61b..57877e17a 100644 --- a/sdk-java/src/main/java/ly/count/sdk/java/internal/SessionImpl.java +++ b/sdk-java/src/main/java/ly/count/sdk/java/internal/SessionImpl.java @@ -343,11 +343,11 @@ protected TimedEvents timedEvents() { */ @Override public void recordEvent(Event event) { - L.d("[SessionImpl] recordEvent: " + event.toString()); if (!SDKCore.enabled(CoreFeature.Events)) { L.i("[SessionImpl] recordEvent: Skipping event - feature is not enabled"); return; } + L.d("[SessionImpl] recordEvent"); if (began == null) { begin(); @@ -403,11 +403,11 @@ public Session addLocation(double latitude, double longitude) { return this; } - L.d("[SessionImpl] addLocation: latitude = " + latitude + " longitude = " + longitude); if (!SDKCore.enabled(CoreFeature.Location)) { L.i("[SessionImpl] addLocation: Skipping event - feature is not enabled"); return this; } + L.d("[SessionImpl] addLocation"); Countly.instance().location().setLocation(null, null, latitude + "," + longitude, null); return this; } @@ -421,11 +421,11 @@ public Session addLocation(double latitude, double longitude) { * @deprecated use {@link ModuleViews.Views#startView(String)} instead via {@link Countly#views()} */ public View view(String name, boolean start) { - L.d("[SessionImpl] view: name = " + name + " start = " + start); if (!SDKCore.enabled(CoreFeature.Views)) { L.i("[SessionImpl] view: Skipping view - feature is not enabled"); return null; } + L.d("[SessionImpl] view: start = " + start); if (currentView != null) { currentView.stop(false); } diff --git a/sdk-java/src/test/java/ly/count/sdk/java/internal/DefaultNetworkingRecheckTests.java b/sdk-java/src/test/java/ly/count/sdk/java/internal/DefaultNetworkingRecheckTests.java new file mode 100644 index 000000000..c6c071df2 --- /dev/null +++ b/sdk-java/src/test/java/ly/count/sdk/java/internal/DefaultNetworkingRecheckTests.java @@ -0,0 +1,79 @@ +package ly.count.sdk.java.internal; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import ly.count.sdk.java.Countly; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +import static org.mockito.Mockito.mock; + +/** + * The checks {@link DefaultNetworking} keeps while it drains the request queue. + */ +@RunWith(JUnit4.class) +public class DefaultNetworkingRecheckTests { + + @Before + public void beforeTest() { + TestUtils.createCleanTestState(); + } + + @After + public void afterTest() { + Countly.instance().halt(); + } + + /** + * A drain that fails still runs the check asked for while it ran, so a request queued meanwhile + * does not wait for the next check. + */ + @Test + public void aFailingDrain_stillRunsTheCheckAskedForMeanwhile() throws Exception { + Countly.instance().init(TestUtils.getBaseConfig()); + InternalConfig config = SDKCore.instance.config; + CountDownLatch firstDrainRuns = new CountDownLatch(1); + CountDownLatch checkAskedFor = new CountDownLatch(1); + CountDownLatch nextDrainRan = new CountDownLatch(1); + AtomicInteger drains = new AtomicInteger(); + DefaultNetworking networking = new DefaultNetworking() { + /** + * Fails the first drain once a check was asked for while it ran, and records the next. + */ + @Override + protected Tasks.Task submit(InternalConfig config, boolean followsOwnTransportWork) { + return new Tasks.Task(Tasks.ID_STRICT) { + /** + * Fails the first time, after the check was asked for, and records the next. + */ + @Override + public Boolean call() throws Exception { + if (drains.incrementAndGet() == 1) { + firstDrainRuns.countDown(); + Assert.assertTrue(checkAskedFor.await(10, TimeUnit.SECONDS)); + throw new IllegalStateException("the first drain fails"); + } + nextDrainRan.countDown(); + return false; + } + }; + } + }; + networking.init(config, mock(IStorageForRequestQueue.class)); + try { + networking.check(config); + Assert.assertTrue(firstDrainRuns.await(10, TimeUnit.SECONDS)); + networking.check(config); + checkAskedFor.countDown(); + + Assert.assertTrue("the check asked for during the failing drain never ran", nextDrainRan.await(10, TimeUnit.SECONDS)); + } finally { + networking.stop(config); + } + } +} diff --git a/sdk-java/src/test/java/ly/count/sdk/java/internal/LogTests.java b/sdk-java/src/test/java/ly/count/sdk/java/internal/LogTests.java index 9dea6d761..aa385c1ad 100644 --- a/sdk-java/src/test/java/ly/count/sdk/java/internal/LogTests.java +++ b/sdk-java/src/test/java/ly/count/sdk/java/internal/LogTests.java @@ -3,11 +3,15 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.Deque; import java.util.List; +import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import javax.annotation.Nonnull; import ly.count.sdk.java.internal.ConfigurationProvider.LogGatheringState; import org.json.JSONArray; import org.json.JSONObject; @@ -38,9 +42,41 @@ public class LogTests { private final List handedOver = new ArrayList<>(); private final List queuedBatches = new CopyOnWriteArrayList<>(); + //the storage IDs of the batches the played request queue still holds, oldest first + private final Deque waitingBatchIds = new ConcurrentLinkedDeque<>(); + private final AtomicLong nextRequestId = new AtomicLong(1); private final AtomicInteger queueAttempts = new AtomicInteger(); private volatile boolean queueAccepts = true; + /** + * Plays the request queue: records each batch, or refuses it while {@link #queueAccepts} is off, + * and holds it until {@link #sendOldestBatch(Log)}. + */ + private final Log.LogBatchQueue queue = new Log.LogBatchQueue() { + /** + * Records the batch and holds it, or refuses it. + */ + @Override + public Long queueLogBatch(@Nonnull String payload) { + queueAttempts.incrementAndGet(); + if (!queueAccepts) { + return null; + } + queuedBatches.add(new JSONObject(payload)); + long requestId = nextRequestId.getAndIncrement(); + waitingBatchIds.addLast(requestId); + return requestId; + } + + /** + * Whether the batch was not sent yet. + */ + @Override + public boolean isQueued(@Nonnull Long requestId) { + return waitingBatchIds.contains(requestId); + } + }; + @Before public void setupEveryTest() { String serverUrl = "http://www.serverurl.com"; @@ -66,8 +102,9 @@ public void logInit_null() { * capture whatever the print level. *

* Verifies, for a logger printing nothing, only errors and everything, that the listener hears - * every line with its level, that the print level stays as configured, that every line is held - * while nothing is decided, and that an armed directive uploads them with their level characters. + * every line with its level, that the print level stays as configured, that every line waits for + * its consent while no consent provider is set, and that an armed directive uploads them with + * their level characters. */ @Test public void levelMethods_informTheListenerAndFeedTheCapture_whateverThePrintLevel() throws Exception { @@ -83,7 +120,7 @@ public void levelMethods_informTheListenerAndFeedTheCapture_whateverThePrintLeve Assert.assertEquals(Arrays.asList("VERBOSE aa", "DEBUG bb", "INFO cc", "WARN dd", "ERROR ee"), heard); Assert.assertEquals(printLevel, log.getPrintLevel()); Assert.assertEquals(LogGatheringState.UNDECIDED, log.logGatheringState); - Assert.assertEquals(5, log.heldLogLineCount()); + Assert.assertEquals(5, log.awaitingConsentLineCount()); queuedBatches.clear(); handOver(log, settings(LogGatheringState.GATHERING, "gather_levels", ModuleConfiguration.logGatheringAllLevels, 100)); @@ -109,11 +146,13 @@ public void levelMethods_informTheListenerAndFeedTheCapture_whateverThePrintLeve @Test public void buffer_holdsAtMost500LinesAnd128KB_droppingTheOldestIntoD_andCutsLongMessages() throws Exception { Log lines = new Log(OFF, null); + handOver(lines, settings(LogGatheringState.UNDECIDED, null, ModuleConfiguration.logGatheringAllLevels, ModuleConfiguration.logGatheringDefaultBatchSize)); + int handOverLines = lines.heldLogLineCount(); for (int i = 0; i < 600; i++) { lines.e("line " + i); } Assert.assertEquals(ModuleConfiguration.logGatheringMaxBufferedLines, lines.heldLogLineCount()); - Assert.assertEquals(100, lines.droppedLogLineCount); + Assert.assertEquals(handOverLines + 100, lines.droppedLogLineCount); Log characters = new Log(OFF, null); handOver(characters, settings(LogGatheringState.GATHERING, "gather_characters", "e", 500)); @@ -154,12 +193,13 @@ public void buffer_holdsAtMost500LinesAnd128KB_droppingTheOldestIntoD_andCutsLon public void notGathering_holdsNothing_andStartsNoDeliveryThread() { Log log = new Log(OFF, null); log.e("held while undecided"); - Assert.assertEquals(1, log.heldLogLineCount()); + Assert.assertEquals(1, log.awaitingConsentLineCount()); handOver(log, settings(LogGatheringState.NOT_GATHERING, null, ModuleConfiguration.logGatheringAllLevels, ModuleConfiguration.logGatheringDefaultBatchSize)); Assert.assertEquals(LogGatheringState.NOT_GATHERING, log.logGatheringState); Assert.assertFalse(log.isCapturingLogs()); Assert.assertEquals(0, log.heldLogLineCount()); + Assert.assertEquals(0, log.awaitingConsentLineCount()); for (int i = 0; i < 1000; i++) { log.e("error " + i); @@ -180,25 +220,28 @@ public void notGathering_holdsNothing_andStartsNoDeliveryThread() { @Test public void selfCapture_batchCarryingLinesAndLinesOfAMarkedThread_areNotHeld() throws InterruptedException { Log log = new Log(OFF, null); + handOver(log, settings(LogGatheringState.UNDECIDED, null, ModuleConfiguration.logGatheringAllLevels, ModuleConfiguration.logGatheringDefaultBatchSize)); + int handOverLines = log.heldLogLineCount(); log.d("New request 1: [1] app_key=key&" + Log.transportMarker + "%7B%22i%22%3A%22gather%22%7D"); - Assert.assertEquals(0, log.heldLogLineCount()); + Assert.assertEquals(handOverLines, log.heldLogLineCount()); log.setOwnTransportWork(true); log.i("logged while doing transport work"); - Assert.assertEquals(0, log.heldLogLineCount()); + Assert.assertEquals(handOverLines, log.heldLogLineCount()); Thread other = new Thread(() -> log.i("logged by another thread meanwhile")); other.start(); other.join(); - Assert.assertEquals(1, log.heldLogLineCount()); + Assert.assertEquals(handOverLines + 1, log.heldLogLineCount()); log.setOwnTransportWork(false); log.i("logged once the mark is cleared"); - Assert.assertEquals(2, log.heldLogLineCount()); + Assert.assertEquals(handOverLines + 2, log.heldLogLineCount()); } /** - * A batch the queue refuses goes back to the front of the buffer with its drop count, so a later - * delivery uploads every line in order and still reports the loss. + * A batch the queue refuses goes back to the front of the buffer with its drop count, so the + * later deliveries, one batch at a time as each is sent, upload every line in order and still + * report the loss. */ @Test public void refusedBatch_goesBackToTheFrontOfTheBuffer_withItsDropCount() throws Exception { @@ -217,6 +260,7 @@ public void refusedBatch_goesBackToTheFrontOfTheBuffer_withItsDropCount() throws queueAccepts = true; log.flushGatheredLogs(); awaitDelivery(log); + sendQueuedBatches(log); Assert.assertEquals(0, log.heldLogLineCount()); Assert.assertEquals(5, queuedBatches.size()); List messages = new ArrayList<>(); @@ -249,7 +293,7 @@ public void endedGathering_queuesTheHeldLinesOnlyWhenAsked_andIgnoresEverythingA Assert.assertEquals(LogGatheringState.NOT_GATHERING, kept.logGatheringState); Assert.assertNull(kept.logDeliveryExecutor); - kept.setLogGatheringProviders(settings(LogGatheringState.GATHERING, "gather_again", "e", 10), () -> true, this::queue); + kept.setLogGatheringProviders(settings(LogGatheringState.GATHERING, "gather_again", "e", 10), () -> ~0, queue); kept.e("logged after the end"); kept.flushGatheredLogs(); Assert.assertEquals(LogGatheringState.NOT_GATHERING, kept.logGatheringState); @@ -267,13 +311,297 @@ public void endedGathering_queuesTheHeldLinesOnlyWhenAsked_andIgnoresEverythingA Assert.assertEquals(LogGatheringState.NOT_GATHERING, discarded.logGatheringState); } + /** + * A line is gathered only once the consent of the feature it belongs to is given, found from + * the name it starts with in brackets, and a line of no feature needs both the events and the + * user profiles consent. Until then it waits apart, and it joins the gather in capture order. + *

+ * Verifies that every line waits before the providers are set; that with only the sessions + * consent given only the session lines are held; that each consent given releases exactly its + * lines, the line of no feature once both of its consents are given; that the batch keeps the + * order the lines were logged in; and that the waiting lines are capped at 500, oldest first, + * without counting into {@code d}. + */ + @Test + public void perLineConsent_gathersALineOnlyWithTheConsentOfItsFeature_inCaptureOrder() throws Exception { + Log log = new Log(OFF, null); + log.e("[ModuleLocation] setLocation, gps:[12.345678,98.765432]"); + log.e("[SessionImpl] begin"); + Assert.assertEquals(0, log.heldLogLineCount()); + Assert.assertEquals(2, log.awaitingConsentLineCount()); + + AtomicInteger given = new AtomicInteger(CoreFeature.Sessions.getIndex()); + handedOver.add(log); + log.setLogGatheringProviders(settings(LogGatheringState.GATHERING, "gather_consent", "e", 100), given::get, queue); + Assert.assertEquals(1, log.heldLogLineCount()); + log.e("[ModuleEvents] recordEventInternal, key:[purchase]"); + log.e("[Test] a line of no feature"); + log.e("[Views] startView, name:[home]"); + Assert.assertEquals(1, log.heldLogLineCount()); + Assert.assertEquals(4, log.awaitingConsentLineCount()); + + given.set(given.get() | CoreFeature.Events.getIndex()); + log.releaseConsentedLogLines(); + Assert.assertEquals(2, log.heldLogLineCount()); + given.set(given.get() | CoreFeature.UserProfiles.getIndex() | CoreFeature.Location.getIndex()); + log.releaseConsentedLogLines(); + Assert.assertEquals(4, log.heldLogLineCount()); + Assert.assertEquals(1, log.awaitingConsentLineCount()); + + log.onSdkInitFinished(); + awaitDelivery(log); + Assert.assertEquals(1, queuedBatches.size()); + Assert.assertEquals(Arrays.asList("[ModuleLocation] setLocation, gps:[12.345678,98.765432]", "[SessionImpl] begin", + "[ModuleEvents] recordEventInternal, key:[purchase]", "[Test] a line of no feature"), messagesOf(queuedBatches.get(0))); + + for (int i = 0; i < 520; i++) { + log.e("[ModuleViews] waiting line " + i); + } + Assert.assertEquals(ModuleConfiguration.logGatheringMaxBufferedLines, log.awaitingConsentLineCount()); + Assert.assertEquals(0, log.droppedLogLineCount); + } + + /** + * One batch waits in the request queue at a time: until it is sent the lines stay in the + * buffer, and each send lets the next batch go, a partial one included. + *

+ * Verifies that a delivery queues a single batch however many are held; that neither a full + * batch nor a flush queues another while it waits; that each sent batch lets exactly one more + * go, until every line went out once, in order; and that a refused batch does not count as + * waiting. + */ + @Test + public void oneBatchAtATime_holdsTheLinesUntilTheWaitingBatchIsSent() throws Exception { + Log log = new Log(OFF, null); + handOver(log, settings(LogGatheringState.GATHERING, "gather_one", "e", 10)); + for (int i = 0; i < 35; i++) { + log.e("line " + i); + } + log.onSdkInitFinished(); + awaitDelivery(log); + Assert.assertEquals(1, queuedBatches.size()); + Assert.assertEquals(1, waitingBatchIds.size()); + Assert.assertEquals(25, log.heldLogLineCount()); + + log.e("line 35"); + log.flushGatheredLogs(); + awaitDelivery(log); + Assert.assertEquals(1, queuedBatches.size()); + Assert.assertEquals(26, log.heldLogLineCount()); + + for (int sent = 1; sent <= 3; sent++) { + sendOldestBatch(log); + awaitDelivery(log); + Assert.assertEquals(sent + 1, queuedBatches.size()); + Assert.assertEquals(1, waitingBatchIds.size()); + } + sendOldestBatch(log); + awaitDelivery(log); + Assert.assertEquals(4, queuedBatches.size()); + Assert.assertEquals(0, waitingBatchIds.size()); + List messages = new ArrayList<>(); + for (JSONObject batch : queuedBatches) { + messages.addAll(messagesOf(batch)); + } + List expected = new ArrayList<>(); + for (int i = 0; i < 36; i++) { + expected.add("line " + i); + } + Assert.assertEquals(expected, messages); + + queueAccepts = false; + log.e("refused line"); + log.flushGatheredLogs(); + awaitDelivery(log); + Assert.assertEquals(0, waitingBatchIds.size()); + queueAccepts = true; + log.flushGatheredLogs(); + awaitDelivery(log); + Assert.assertEquals(5, queuedBatches.size()); + Assert.assertEquals(Arrays.asList("refused line"), messagesOf(queuedBatches.get(4))); + } + + /** + * The batch waiting in the request queue holds the next one back only while the queue holds it, + * and its send resumes only a delivery it held back. + *

+ * Verifies that a batch the queue dropped without sending, as when the queue is over its size, + * no longer holds the lines back on the next flush; that a send that held nothing back queues + * nothing, so a line logged meanwhile waits for the next flush; and that a send after a delivery + * it held back queues the line that waited, a partial batch. + */ + @Test + public void theWaitingBatch_holdsTheNextBackOnlyWhileQueued_andItsSendResumesOnlyADeliveryItHeldBack() throws Exception { + Log log = new Log(OFF, null); + handOver(log, settings(LogGatheringState.GATHERING, "gather_waiting", "e", 10)); + log.e("line 0"); + log.onSdkInitFinished(); + awaitDelivery(log); + log.e("line 1"); + log.flushGatheredLogs(); + awaitDelivery(log); + Assert.assertEquals(1, queuedBatches.size()); + Assert.assertEquals(1, log.heldLogLineCount()); + + waitingBatchIds.clear(); + log.flushGatheredLogs(); + awaitDelivery(log); + Assert.assertEquals(2, queuedBatches.size()); + Assert.assertEquals(Collections.singletonList("line 1"), messagesOf(queuedBatches.get(1))); + + log.e("line 2"); + sendOldestBatch(log); + awaitDelivery(log); + Assert.assertEquals(2, queuedBatches.size()); + Assert.assertEquals(1, log.heldLogLineCount()); + + log.flushGatheredLogs(); + awaitDelivery(log); + log.e("line 3"); + log.flushGatheredLogs(); + awaitDelivery(log); + Assert.assertEquals(3, queuedBatches.size()); + sendOldestBatch(log); + awaitDelivery(log); + Assert.assertEquals(4, queuedBatches.size()); + Assert.assertEquals(Collections.singletonList("line 3"), messagesOf(queuedBatches.get(3))); + } + + /** + * A removed consent drops the held lines that needed it, so nothing gathered for a feature goes + * out once its consent is withdrawn, and the lines dropped are reported in {@code d}. + */ + @Test + public void removedConsent_dropsTheHeldLinesThatNeededIt_countingThemIntoD() throws Exception { + Log log = new Log(OFF, null); + AtomicInteger given = new AtomicInteger(CoreFeature.Events.getIndex() | CoreFeature.Location.getIndex() | CoreFeature.UserProfiles.getIndex()); + handedOver.add(log); + log.setLogGatheringProviders(settings(LogGatheringState.GATHERING, "gather_withdrawn", "e", 100), given::get, queue); + log.e("[ModuleEvents] an events line"); + log.e("[ModuleLocation] a location line"); + log.e("[Test] a line of no feature"); + Assert.assertEquals(3, log.heldLogLineCount()); + + given.set(CoreFeature.Events.getIndex()); + log.dropLinesWithoutConsent(); + + Assert.assertEquals(1, log.heldLogLineCount()); + log.onSdkInitFinished(); + awaitDelivery(log); + Assert.assertEquals(Collections.singletonList("[ModuleEvents] an events line"), messagesOf(queuedBatches.get(0))); + Assert.assertEquals(2, queuedBatches.get(0).getInt("d")); + } + + /** + * A gather the server stops queues the lines it still holds once, with its id, while a gather + * replaced by another drops them, and so does a decision against gathering made while nothing + * was decided. + *

+ * Verifies that stopping a gather queues one batch of exactly its held lines, with its id and + * past a batch already waiting; that the capture stops; that a gather replaced by a new id + * queues nothing; and that deciding off while undecided queues nothing. + */ + @Test + public void stoppedGather_queuesItsHeldLinesOnceWithItsId_whileAReplacedGatherDropsThem() throws Exception { + ModuleConfiguration stopped = settings(LogGatheringState.GATHERING, "gather_tail", "e", 10); + Log log = new Log(OFF, null); + handOver(log, stopped); + log.onSdkInitFinished(); + for (int i = 0; i < 13; i++) { + log.e("line " + i); + } + awaitDelivery(log); + Assert.assertEquals(1, queuedBatches.size()); + Assert.assertEquals(3, log.heldLogLineCount()); + + stopped.currentVLogGathering = ModuleConfiguration.LogGatheringDirective.NOT_GATHERING; + log.applyLogGatheringDirective("test"); + Assert.assertTrue("the held lines were not queued", TestUtils.waitFor(10_000, () -> queuedBatches.size() == 2)); + Assert.assertEquals("gather_tail", queuedBatches.get(1).getString("i")); + Assert.assertEquals(0, queuedBatches.get(1).getInt("d")); + Assert.assertEquals(Arrays.asList("line 10", "line 11", "line 12"), messagesOf(queuedBatches.get(1))); + Assert.assertFalse(log.isCapturingLogs()); + log.e("logged after the stop"); + Assert.assertEquals(0, log.heldLogLineCount()); + + ModuleConfiguration replaced = settings(LogGatheringState.GATHERING, "gather_one", "e", 100); + Log other = new Log(OFF, null); + handOver(other, replaced); + other.onSdkInitFinished(); + awaitDelivery(other); + other.e("line of the first gather"); + replaced.currentVLogGathering = new ModuleConfiguration.LogGatheringDirective(LogGatheringState.GATHERING, "gather_two", "e", 100); + other.applyLogGatheringDirective("test"); + awaitDelivery(other); + Assert.assertEquals(0, other.heldLogLineCount()); + + Log undecided = new Log(OFF, null); + undecided.e("line held while undecided"); + handOver(undecided, settings(LogGatheringState.NOT_GATHERING, null, ModuleConfiguration.logGatheringAllLevels, ModuleConfiguration.logGatheringDefaultBatchSize)); + undecided.onSdkInitFinished(); + Thread.sleep(100); + Assert.assertEquals(2, queuedBatches.size()); + Assert.assertEquals(2, queueAttempts.get()); + } + + /** + * Before the device ID changes without merge the gather stops: every held line is queued on + * the calling thread, past a batch already waiting, the lines waiting for their consent are + * dropped, and a later directive can still arm a new gather. + *

+ * Verifies that the held lines are queued before the call returns, in batches of at most + * {@code b} and in order; that a line waiting for its consent is not queued; that nothing is + * captured afterwards; and that a directive applied later arms a gather again. + */ + @Test + public void deviceIdChange_queuesTheHeldLinesThenStops_andALaterDirectiveCanArmAgain() throws Exception { + AtomicInteger given = new AtomicInteger(Log.UNATTRIBUTED_LINE_CONSENT); + ModuleConfiguration settings = settings(LogGatheringState.GATHERING, "gather_did", "e", 10); + Log log = new Log(OFF, null); + handedOver.add(log); + log.setLogGatheringProviders(settings, given::get, queue); + log.onSdkInitFinished(); + for (int i = 0; i < 25; i++) { + log.e("[Test] user A line " + i); + } + log.e("[ModuleLocation] user A location"); + awaitDelivery(log); + Assert.assertEquals(1, queuedBatches.size()); + Assert.assertEquals(1, log.awaitingConsentLineCount()); + + log.stopLogGatheringForDeviceIdChange(); + Assert.assertEquals(3, queuedBatches.size()); + List messages = new ArrayList<>(); + for (JSONObject batch : queuedBatches) { + Assert.assertEquals("gather_did", batch.getString("i")); + Assert.assertTrue(batch.getJSONArray("l").length() <= 10); + messages.addAll(messagesOf(batch)); + } + List expected = new ArrayList<>(); + for (int i = 0; i < 25; i++) { + expected.add("[Test] user A line " + i); + } + Assert.assertEquals(expected, messages); + Assert.assertEquals(LogGatheringState.NOT_GATHERING, log.logGatheringState); + Assert.assertEquals(0, log.awaitingConsentLineCount()); + log.e("[Test] user B line"); + Assert.assertEquals(0, log.heldLogLineCount()); + Assert.assertNull(log.logDeliveryExecutor); + + settings.currentVLogGathering = new ModuleConfiguration.LogGatheringDirective(LogGatheringState.GATHERING, "gather_b", "e", 10); + log.applyLogGatheringDirective("test"); + Assert.assertTrue(log.isGatheringLogs()); + log.e("[Test] user B armed line"); + Assert.assertEquals(1, log.heldLogLineCount()); + } + /** * A batch is valid JSON with its keys always in the same order, every message escaped. */ @Test public void batchPayload_isJsonWithItsKeysInOrder_andEscapesEveryMessage() { String awkward = "quote \" backslash \\ newline \n tab \t closing pair \uD83D\uDE00"; - List lines = Arrays.asList(new Log.LogLine(1755500000000L, 'e', awkward), new Log.LogLine(1755500000001L, 'd', "")); + List lines = Arrays.asList(new Log.LogLine(0, 1755500000000L, 'e', awkward, 0), new Log.LogLine(1, 1755500000001L, 'd', "", 0)); String payload = Log.buildLogBatchPayload("gather \"quoted\"", 3, lines); Assert.assertTrue(payload, payload.startsWith("{\"i\":\"gather \\\"quoted\\\"\",\"d\":3,\"l\":[{\"t\":1755500000000,\"l\":\"e\",\"m\":\"")); @@ -307,31 +635,16 @@ private static ModuleConfiguration settings(LogGatheringState state, String gath } /** - * Hands a logger the given settings, a consent check that always allows and {@link #queue(String)}, - * as an SDK does once its settings are resolved. Nothing is uploaded before the test tells the - * logger that init finished. + * Hands a logger the given settings, every consent given and {@link #queue}, as an SDK + * that requires no consent does once its settings are resolved. Nothing is uploaded before the + * test tells the logger that init finished. * * @param log the logger * @param settings the settings */ private void handOver(Log log, ModuleConfiguration settings) { handedOver.add(log); - log.setLogGatheringProviders(settings, () -> true, this::queue); - } - - /** - * Records a batch, or refuses it while {@link #queueAccepts} is off. - * - * @param payload the batch - * @return whether the batch was queued - */ - private boolean queue(String payload) { - queueAttempts.incrementAndGet(); - if (!queueAccepts) { - return false; - } - queuedBatches.add(new JSONObject(payload)); - return true; + log.setLogGatheringProviders(settings, () -> ~0, queue); } /** @@ -348,6 +661,30 @@ private static void awaitDelivery(Log log) throws Exception { } } + /** + * Plays the request queue sending every batch the logger queued, one at a time, each send + * letting the logger queue its next batch, until none waits. + * + * @param log the logger + * @throws Exception when a delivery does not finish in time + */ + private void sendQueuedBatches(Log log) throws Exception { + for (int i = 0; i < 1000 && !waitingBatchIds.isEmpty(); i++) { + sendOldestBatch(log); + awaitDelivery(log); + } + } + + /** + * Plays the request queue sending its oldest batch, then telling the logger. + * + * @param log the logger + */ + private void sendOldestBatch(Log log) { + waitingBatchIds.pollFirst(); + log.onLogBatchSent(); + } + /** * The messages of a batch, in order. * diff --git a/sdk-java/src/test/java/ly/count/sdk/java/internal/ModuleConnectionTestTests.java b/sdk-java/src/test/java/ly/count/sdk/java/internal/ModuleConnectionTestTests.java index 9427ee645..f3af81242 100644 --- a/sdk-java/src/test/java/ly/count/sdk/java/internal/ModuleConnectionTestTests.java +++ b/sdk-java/src/test/java/ly/count/sdk/java/internal/ModuleConnectionTestTests.java @@ -1378,7 +1378,7 @@ private static Transport initializedTransport(Config config) { */ private static String generatedPublicKeyPin() throws Exception { KeyPairGenerator generator = KeyPairGenerator.getInstance("RSA"); - generator.initialize(1024); + generator.initialize(2048); return Base64.getEncoder().encodeToString(generator.generateKeyPair().getPublic().getEncoded()); } diff --git a/sdk-java/src/test/java/ly/count/sdk/java/internal/SdkLogGatheringTests.java b/sdk-java/src/test/java/ly/count/sdk/java/internal/SdkLogGatheringTests.java index a737084d3..264667ad9 100644 --- a/sdk-java/src/test/java/ly/count/sdk/java/internal/SdkLogGatheringTests.java +++ b/sdk-java/src/test/java/ly/count/sdk/java/internal/SdkLogGatheringTests.java @@ -7,7 +7,6 @@ import java.io.InputStream; import java.io.OutputStream; import java.net.InetSocketAddress; -import java.net.URL; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.util.ArrayList; @@ -22,12 +21,9 @@ import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; -import java.util.function.BooleanSupplier; import java.util.regex.Matcher; import java.util.regex.Pattern; -import javax.annotation.Nullable; import ly.count.sdk.java.Config; import ly.count.sdk.java.Countly; import ly.count.sdk.java.internal.ConfigurationProvider.LogGatheringState; @@ -87,156 +83,172 @@ public void afterTest() { /** * An armed directive adopts what was captured before it arrived. *

- * Verifies that the lines of the directive's levels logged during init, before any response, - * are queued once init finishes, the very first line of the SDK first; that every batch echoes - * the gather id, carries the required request parameters and at most {@code b} lines, each - * with exactly {@code t}, {@code l} and {@code m}, none of another level and none quoting a - * batch; that a full batch goes out on its own while a partial one waits for a flush; and that a - * full batch is too long for a GET, so it is sent as a POST. An unknown top level key next to + * Verifies against a local server that the lines of the directive's levels logged during init, + * before any response, are uploaded once init finishes, the very first line of the SDK first; + * that every batch echoes the gather id, carries the required request parameters and at most + * {@code b} lines, each with exactly {@code t}, {@code l} and {@code m}, none of another level + * and none quoting a batch; that a partial batch waits for a flush while no batch is in flight; + * that a full batch goes out on its own, the lines after it following once it is sent; and that + * a full batch is too long for a GET, so it is sent as a POST. An unknown top level key next to * {@code lg} changes none of it. */ @Test public void armedDirective_uploadsTheInitLinesOfItsLevels_withTheGatherIdInBatchesOfAtMostB() throws Exception { long startMs = System.currentTimeMillis(); - server.respondWith(new ServerConfigBuilder().networking(false) - .logGatheringOn("gather_init", "wi", 10) - .topLevelKey("someFutureFeature", "a value this SDK has never heard of")); - init(TestUtils.getConfigSdkBehaviorSettings()); - assertGathering("gather_init", "wi", 10); - - awaitLogDelivery(); - List initBatches = batches(); - Assert.assertFalse("init end queued nothing", initBatches.isEmpty()); - Assert.assertEquals("[SDKCore] Initializing Countly", initBatches.get(0).getJSONArray("l").getJSONObject(0).getString("m")); - - String padding = repeat('p', 80); - List armedLines = new ArrayList<>(); - for (int i = 0; i < 25; i++) { - armedLines.add("[Test] armed line " + i + " " + padding); - logger().w(armedLines.get(i)); - } - awaitLogDelivery(); - List beforeFlush = linesStartingWith("[Test] armed line ", batches()); - Assert.assertTrue("[" + beforeFlush.size() + "] lines before a flush", beforeFlush.size() >= 10 && beforeFlush.size() < armedLines.size()); - Assert.assertEquals(armedLines.subList(0, beforeFlush.size()), beforeFlush); - - logger().flushGatheredLogs(); - awaitLogDelivery(); - Assert.assertEquals(armedLines, linesStartingWith("[Test] armed line ", batches())); - - List> requests = logRequests(); - for (Map request : requests) { - TestUtils.validateRequiredParams(request); - JSONObject batch = new JSONObject(request.get(Log.keySdkLogs)); - Assert.assertEquals(names("i", "d", "l"), batch.keySet()); - Assert.assertEquals("gather_init", batch.getString("i")); - Assert.assertEquals(0, batch.getInt("d")); - - JSONArray lines = batch.getJSONArray("l"); - Assert.assertTrue("a batch of [" + lines.length() + "] lines", lines.length() >= 1 && lines.length() <= 10); - for (int i = 0; i < lines.length(); i++) { - JSONObject line = lines.getJSONObject(i); - Assert.assertEquals(names("t", "l", "m"), line.keySet()); - long timestamp = line.getLong("t"); - Assert.assertTrue(timestamp >= startMs && timestamp <= System.currentTimeMillis()); - Assert.assertTrue(line.getString("l").equals("w") || line.getString("l").equals("i")); - Assert.assertFalse(line.getString("m").contains(Log.keySdkLogs)); + List requests = new CopyOnWriteArrayList<>(); + HttpServer httpServer = startRecordingServer(requests); + try { + server.respondWith(new ServerConfigBuilder() + .logGatheringOn("gather_init", "wi", 10) + .topLevelKey("someFutureFeature", "a value this SDK has never heard of")); + init(localConfig(httpServer)); + assertGathering("gather_init", "wi", 10); + + awaitSentBatches(); + List initBatches = receivedBatches(requests); + Assert.assertFalse("init end queued nothing", initBatches.isEmpty()); + Assert.assertEquals("[SDKCore] Initializing Countly", initBatches.get(0).getJSONArray("l").getJSONObject(0).getString("m")); + + List partialLines = numbered("[Test] partial line ", 0, 5); + for (String line : partialLines) { + logger().w(line); } - } + awaitLogDelivery(); + Thread.sleep(200); + Assert.assertTrue(linesStartingWith("[Test] partial line ", receivedBatches(requests)).isEmpty()); + logger().flushGatheredLogs(); + awaitSentBatches(); + Assert.assertEquals(partialLines, linesStartingWith("[Test] partial line ", receivedBatches(requests))); + + String padding = repeat('p', 80); + List armedLines = new ArrayList<>(); + for (int i = 0; i < 25; i++) { + armedLines.add("[Test] armed line " + i + " " + padding); + logger().w(armedLines.get(i)); + } + awaitSentBatches(); + Assert.assertEquals(armedLines, linesStartingWith("[Test] armed line ", receivedBatches(requests))); - int fullArmedBatches = 0; - for (String params : storedRequestParams()) { - Map decoded = decode(params); - if (decoded.containsKey(Log.keySdkLogs) && decoded.get(Log.keySdkLogs).contains("[Test] armed line 9 ")) { - Request request = new Request(0L); - request.params = new Params(params); - Assert.assertFalse("a full batch fits a GET", request.isGettable(new URL(TestUtils.SERVER_URL))); - fullArmedBatches++; + int fullArmedBatches = 0; + for (ReceivedRequest request : requests) { + if (!request.params.containsKey(Log.keySdkLogs)) { + continue; + } + TestUtils.validateRequiredParams(request.params); + JSONObject batch = new JSONObject(request.params.get(Log.keySdkLogs)); + Assert.assertEquals(names("i", "d", "l"), batch.keySet()); + Assert.assertEquals("gather_init", batch.getString("i")); + Assert.assertEquals(0, batch.getInt("d")); + + JSONArray lines = batch.getJSONArray("l"); + Assert.assertTrue("a batch of [" + lines.length() + "] lines", lines.length() >= 1 && lines.length() <= 10); + for (int i = 0; i < lines.length(); i++) { + JSONObject line = lines.getJSONObject(i); + Assert.assertEquals(names("t", "l", "m"), line.keySet()); + long timestamp = line.getLong("t"); + Assert.assertTrue(timestamp >= startMs && timestamp <= System.currentTimeMillis()); + Assert.assertTrue(line.getString("l").equals("w") || line.getString("l").equals("i")); + Assert.assertFalse(line.getString("m").contains(Log.keySdkLogs)); + } + if (request.params.get(Log.keySdkLogs).contains("[Test] armed line 9 ")) { + Assert.assertEquals("a full batch fits a GET", "POST", request.method); + fullArmedBatches++; + } } + Assert.assertEquals(1, fullArmedBatches); + } finally { + Countly.instance().halt(); + httpServer.stop(0); } - Assert.assertEquals(1, fullArmedBatches); } /** * While tracking is off the lines are held, the buffer keeps the newest 500 of them, and the * lines it drops are reported once, with the next batch. *

- * Verifies that nothing is queued while tracking is off, flush or not; that once tracking is back - * on the held lines go out oldest first in full batches of {@code b}, the first one reporting - * every dropped line in {@code d}, so the uploaded and the dropped lines add up to every line - * logged; and that a {@code b} over the buffer ceiling is clamped to it, so no batch ever holds - * more than 500 lines. + * Verifies against a local server that nothing is queued while tracking is off, flush or not; + * that once tracking is back on the held lines go out oldest first in full batches of {@code b}, + * one at a time as each is sent, the first one reporting every dropped line in {@code d}, so the + * uploaded and the dropped lines add up to every line logged; and that a {@code b} over the + * buffer ceiling is clamped to it, so no batch ever holds more than 500 lines. */ @Test public void trackingOffHold_keepsTheNewest500Lines_andDAccountsForEveryDroppedLine() throws Exception { - server.respondWith(heldGather("gather_hold", "e", 25)); - init(TestUtils.getConfigSdkBehaviorSettings()); - assertGathering("gather_hold", "e", 25); - - Log logger = logger(); - //what the SDK logged at the same level during init is held too, and is the oldest - int initLines = logger.heldLogLineCount(); - for (int i = 0; i < 520; i++) { - logger.e("[Test] held line " + i); - } - Assert.assertEquals(ModuleConfiguration.logGatheringMaxBufferedLines, logger.heldLogLineCount()); - Assert.assertEquals(initLines + 20, logger.droppedLogLineCount); - logger.flushGatheredLogs(); - awaitLogDelivery(); - Assert.assertEquals(ModuleConfiguration.logGatheringMaxBufferedLines, logger.heldLogLineCount()); - Assert.assertTrue(logRequests().isEmpty()); - - push(new ServerConfigBuilder().networking(false).tracking(true).logGatheringOn("gather_hold", "e", 25)); - assertGathering("gather_hold", "e", 25); - Assert.assertEquals(ModuleConfiguration.logGatheringMaxBufferedLines, logger.heldLogLineCount()); - logger.flushGatheredLogs(); - awaitLogDelivery(); - Assert.assertEquals(0, logger.heldLogLineCount()); - - List batches = batches(); - Assert.assertEquals(20, batches.size()); - List uploaded = new ArrayList<>(); - int dropped = 0; - for (int b = 0; b < batches.size(); b++) { - JSONObject batch = batches.get(b); - Assert.assertEquals("gather_hold", batch.getString("i")); - Assert.assertEquals(b == 0 ? initLines + 20 : 0, batch.getInt("d")); - Assert.assertEquals(25, batch.getJSONArray("l").length()); - dropped += batch.getInt("d"); - uploaded.addAll(messagesOf(batch)); - } - Assert.assertEquals(numbered("[Test] held line ", 20, 520), uploaded); - Assert.assertEquals(initLines + 520, uploaded.size() + dropped); - - push(heldGather("gather_hold", "e", 9999)); - assertGathering("gather_hold", "e", ModuleConfiguration.logGatheringMaxBufferedLines); - int heldBefore = logger.heldLogLineCount(); - for (int i = 0; i < 600; i++) { - logger.e("[Test] second line " + i); + List requests = new CopyOnWriteArrayList<>(); + HttpServer httpServer = startRecordingServer(requests); + try { + server.respondWith(heldGather("gather_hold", "e", 25)); + init(localConfig(httpServer)); + assertGathering("gather_hold", "e", 25); + + Log logger = logger(); + //what the SDK logged at the same level during init is held too, and is the oldest + int initLines = logger.heldLogLineCount(); + for (int i = 0; i < 520; i++) { + logger.e("[Test] held line " + i); + } + Assert.assertEquals(ModuleConfiguration.logGatheringMaxBufferedLines, logger.heldLogLineCount()); + Assert.assertEquals(initLines + 20, logger.droppedLogLineCount); + logger.flushGatheredLogs(); + awaitLogDelivery(); + Assert.assertEquals(ModuleConfiguration.logGatheringMaxBufferedLines, logger.heldLogLineCount()); + Assert.assertTrue(logRequests().isEmpty()); + + push(new ServerConfigBuilder().tracking(true).logGatheringOn("gather_hold", "e", 25)); + assertGathering("gather_hold", "e", 25); + logger.flushGatheredLogs(); + awaitSentBatches(); + Assert.assertEquals(0, logger.heldLogLineCount()); + + List batches = receivedBatches(requests); + Assert.assertEquals(20, batches.size()); + List uploaded = new ArrayList<>(); + int dropped = 0; + for (int b = 0; b < batches.size(); b++) { + JSONObject batch = batches.get(b); + Assert.assertEquals("gather_hold", batch.getString("i")); + Assert.assertEquals(b == 0 ? initLines + 20 : 0, batch.getInt("d")); + Assert.assertEquals(25, batch.getJSONArray("l").length()); + dropped += batch.getInt("d"); + uploaded.addAll(messagesOf(batch)); + } + Assert.assertEquals(numbered("[Test] held line ", 20, 520), uploaded); + Assert.assertEquals(initLines + 520, uploaded.size() + dropped); + + push(heldGather("gather_hold", "e", 9999)); + assertGathering("gather_hold", "e", ModuleConfiguration.logGatheringMaxBufferedLines); + int heldBefore = logger.heldLogLineCount(); + for (int i = 0; i < 600; i++) { + logger.e("[Test] second line " + i); + } + push(new ServerConfigBuilder().tracking(true).logGatheringOn("gather_hold", "e", 9999)); + logger.flushGatheredLogs(); + awaitSentBatches(); + + List later = receivedBatches(requests).subList(20, receivedBatches(requests).size()); + Assert.assertEquals(1, later.size()); + Assert.assertEquals(heldBefore + 100, later.get(0).getInt("d")); + Assert.assertEquals(numbered("[Test] second line ", 100, 600), messagesOf(later.get(0))); + } finally { + Countly.instance().halt(); + httpServer.stop(0); } - push(new ServerConfigBuilder().networking(false).tracking(true).logGatheringOn("gather_hold", "e", 9999)); - logger.flushGatheredLogs(); - awaitLogDelivery(); - - List later = batches().subList(20, batches().size()); - Assert.assertEquals(1, later.size()); - Assert.assertEquals(heldBefore + 100, later.get(0).getInt("d")); - Assert.assertEquals(numbered("[Test] second line ", 100, 600), messagesOf(later.get(0))); } /** - * Every way a gather ends drops what it holds, and a new gather never carries the lines of an + * Every way a gather ends leaves nothing held, and a new gather never carries the lines of an * earlier one. *

* Verifies, with tracking off so nothing leaves until the end, that the same directive repeated * and a change of levels or batch size within one gather keep the held lines; that * {@code "e":false}, a directive without a usable gather id and a live response without any - * directive each drop them, stop the capture and release the delivery thread; that a new gather - * id drops the lines of the previous gather; and that only the lines of the gather in effect, - * with its id, are ever uploaded. + * directive each stop the capture and release the delivery thread, their held lines offered to + * the request queue once, which refuses them while tracking is off; that a new gather id drops + * the lines of the previous gather; and that only the lines of the gather in effect, with its id, + * are ever uploaded. */ @Test - public void endedOrReplacedGather_dropsTheHeldLines_andANewGatherNeverCarriesTheOldOnes() throws Exception { + public void endedOrReplacedGather_leavesNothingHeld_andANewGatherNeverCarriesTheOldOnes() throws Exception { server.respondWith(heldGather("gather_a", "e", 10)); init(TestUtils.getConfigSdkBehaviorSettings()); Log logger = logger(); @@ -254,7 +266,7 @@ public void endedOrReplacedGather_dropsTheHeldLines_andANewGatherNeverCarriesThe assertGathering("gather_a", "e", 10); Assert.assertEquals(initLines + 2, logger.heldLogLineCount()); - push(new ServerConfigBuilder().networking(false).tracking(false).logGatheringOff()); + pushStop(new ServerConfigBuilder().networking(false).tracking(false).logGatheringOff()); assertNotGathering(); logger.e("[Test] logged while off"); Assert.assertEquals(0, logger.heldLogLineCount()); @@ -268,12 +280,12 @@ public void endedOrReplacedGather_dropsTheHeldLines_andANewGatherNeverCarriesThe Assert.assertEquals(0, logger.heldLogLineCount()); logger.e("[Test] c1"); - push(new ServerConfigBuilder().networking(false).tracking(false).logGatheringDirective(true, " ", "e", 10)); + pushStop(new ServerConfigBuilder().networking(false).tracking(false).logGatheringDirective(true, " ", "e", 10)); assertNotGathering(); push(heldGather("gather_d", "e", 10)); logger.e("[Test] d1"); - push(new ServerConfigBuilder().networking(false).tracking(false)); + pushStop(new ServerConfigBuilder().networking(false).tracking(false)); assertNotGathering(); push(heldGather("gather_e", "e", 10)); @@ -343,45 +355,109 @@ public void onlyALiveResponseDecides_storedDirectiveStaysUndecided_andAFailedFet } /** - * While consent is required and none is given, gathered lines are held, as they quote event - * keys and whole requests. + * While consent is required, a gathered line goes out only once the consent of the feature it + * belongs to is given, a line of no feature needs both the events and the user profiles + * consent, and the coordinates handed to a session are logged only with location consent. *

- * Verifies that neither a full batch nor a flush queues anything before any consent is given, - * and that once one is, a flush queues every held line. + * Verifies, with networking off, that the batch queued with only the sessions consent given + * holds no line of another feature and no coordinates; that the stop that follows events consent + * queues the events line but neither the location line nor the line of no feature; and that + * with location consent given the coordinates are gathered. */ @Test - public void requiredConsent_holdsTheLinesUntilAnyConsentIsGiven() throws Exception { - server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_consent", "e", 10)); - init(TestUtils.getConfigSdkBehaviorSettings().setRequiresConsent(true).enableFeatures(Config.Feature.Events)); - assertGathering("gather_consent", "e", 10); + public void requiredConsent_gathersEachLineOnlyWithTheConsentOfItsFeature() throws Exception { + Config config = TestUtils.getConfigSdkBehaviorSettings().setRequiresConsent(true) + .enableFeatures(Config.Feature.Events, Config.Feature.Sessions, Config.Feature.Location, Config.Feature.UserProfiles); + server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_consent", "ewidv", 500)); + init(config); + assertGathering("gather_consent", "ewidv", 500); Log logger = logger(); - int initLines = logger.heldLogLineCount(); - List consentLines = numbered("[Test] consent line ", 0, 12); - for (String line : consentLines) { - logger.e(line); - } - awaitLogDelivery(); + logger.e("[Test] a line of no feature"); + logger.e("[ModuleEvents] an events line"); + logger.e("[ModuleLocation] a location line"); + Countly.onConsent(Config.Feature.Sessions); + Countly.session().begin(); + ((SessionImpl) Countly.session()).addLocation(12.345678, 98.765432); logger.flushGatheredLogs(); awaitLogDelivery(); - Assert.assertTrue(logRequests().isEmpty()); - Assert.assertEquals(initLines + 12, logger.heldLogLineCount()); + + List queued = receivedLines(batches()); + Assert.assertFalse(queued.isEmpty()); + Assert.assertFalse(queued.contains("[Test] a line of no feature")); + Assert.assertFalse(queued.contains("[ModuleEvents] an events line")); + Assert.assertFalse(queued.contains("[ModuleLocation] a location line")); + for (String line : queued) { + Assert.assertFalse(line, line.contains("12.345678")); + } + Assert.assertTrue(logger.awaitingConsentLineCount() >= 3); Countly.onConsent(Config.Feature.Events); + Countly.instance().stop(); + List stopped = receivedLines(storedBatchesOnDisk()); + Assert.assertTrue(stopped.contains("[ModuleEvents] an events line")); + Assert.assertFalse(stopped.contains("[ModuleLocation] a location line")); + Assert.assertFalse(stopped.contains("[Test] a line of no feature")); + for (String line : stopped) { + Assert.assertFalse(line, line.contains("12.345678")); + } + + server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_location", "ewidv", 500)); + init(config); + Countly.onConsent(Config.Feature.Sessions, Config.Feature.Location); + Countly.session().begin(); + ((SessionImpl) Countly.session()).addLocation(12.345678, 98.765432); + Countly.instance().stop(); + boolean coordinatesGathered = false; + for (JSONObject batch : storedBatchesOnDisk()) { + if ("gather_location".equals(batch.getString("i"))) { + for (String line : messagesOf(batch)) { + coordinatesGathered |= line.contains("12.345678"); + } + } + } + Assert.assertTrue(coordinatesGathered); + } + + /** + * While consent is required, a feature the app did not enable holds no line back, and a removed + * consent drops the held lines of its feature. + *

+ * Verifies, with networking off and the user profiles feature not enabled, that a line of no + * feature is gathered once the events and sessions consents are given; and that after the + * events consent is removed the stop queues the held sessions line but not the held events line. + */ + @Test + public void requiredConsent_aFeatureNotEnabledHoldsNoLineBack_andARemovedConsentDropsItsLines() throws Exception { + Config config = TestUtils.getConfigSdkBehaviorSettings().setRequiresConsent(true) + .enableFeatures(Config.Feature.Events, Config.Feature.Sessions); + server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_enabled", "e", 500)); + init(config); + Log logger = logger(); + logger.e("[Test] a line of no feature"); + Countly.onConsent(Config.Feature.Events, Config.Feature.Sessions); logger.flushGatheredLogs(); awaitLogDelivery(); - Assert.assertEquals(0, logger.heldLogLineCount()); - Assert.assertEquals(consentLines, linesStartingWith("[Test] consent line ", batches())); + Assert.assertTrue(receivedLines(batches()).contains("[Test] a line of no feature")); + + logger.e("[ModuleEvents] a held events line"); + logger.e("[SessionImpl] a held sessions line"); + Countly.onConsentRemoval(Config.Feature.Events); + Countly.instance().stop(); + + List stopped = receivedLines(storedBatchesOnDisk()); + Assert.assertTrue(stopped.contains("[SessionImpl] a held sessions line")); + Assert.assertFalse(stopped.contains("[ModuleEvents] a held events line")); } /** * The global timer flushes a batch that is not full, a stop queues what is held before the * storage stops, and a halt discards it. Either way the logger ends for good. *

- * Verifies that a line below the batch size goes out on a tick of a fast global timer; that - * {@code stop()} queues the held line synchronously, so it is on disk once stop returns, with - * the gather id; that the stopped logger captures and queues nothing more and its delivery - * thread ends; and that {@code halt()} queues nothing at all. + * Verifies that a line below the batch size reaches a local server on a tick of a fast global + * timer; that, with networking off, {@code stop()} queues the held line synchronously, so it is + * on disk once stop returns, with the gather id; that the stopped logger captures and queues + * nothing more and its delivery thread ends; and that {@code halt()} queues nothing at all. */ @Test public void timerTick_flushesAPartialBatch_stopQueuesWhatIsHeld_andHaltDiscardsIt() throws Exception { @@ -391,69 +467,76 @@ public void timerTick_flushesAPartialBatch_stopQueuesWhatIsHeld_andHaltDiscardsI queuedLogRequests.add(message); } }; - server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_tick", "e", 100)); - init(TestUtils.getConfigSdkBehaviorSettings().setLogListener(listener)); - Log logger = logger(); - awaitLogDelivery(); - - //the timer ticks through the modules, so it is sped up only after init built them - CountlyTimer.TIMER_DELAY_MS = 50; - push(new ServerConfigBuilder().networking(false).sessionUpdateInterval(30).logGatheringOn("gather_tick", "e", 100)); - logger.e("[Test] tick line"); - Assert.assertTrue("no tick flushed the partial batch", waitFor(10_000, () -> queuedLinesContain(queuedLogRequests, "[Test] tick line"))); - - CountlyTimer.TIMER_DELAY_MS = 0; - push(new ServerConfigBuilder().networking(false).sessionUpdateInterval(60).logGatheringOn("gather_tick", "e", 100)); - //a tick of the replaced fast timer may still be running - Thread.sleep(250); - awaitLogDelivery(); - logger.e("[Test] tail line"); - Assert.assertEquals(1, logger.heldLogLineCount()); - Set deliveryThreads = liveDeliveryThreads(); - Assert.assertFalse(deliveryThreads.isEmpty()); - int queuedBeforeStop = queuedLogRequests.size(); - - Countly.instance().stop(); - Assert.assertEquals(queuedBeforeStop + 1, queuedLogRequests.size()); - JSONObject tail = null; - for (String params : storedRequestParams()) { - Map decoded = decode(params); - if (decoded.containsKey(Log.keySdkLogs) && decoded.get(Log.keySdkLogs).contains("[Test] tail line")) { - tail = new JSONObject(decoded.get(Log.keySdkLogs)); - } - } - Assert.assertNotNull("stop did not queue the held line", tail); - Assert.assertEquals("gather_tick", tail.getString("i")); - Assert.assertEquals(Arrays.asList("[Test] tail line"), messagesOf(tail)); - - Assert.assertEquals(LogGatheringState.NOT_GATHERING, logger.logGatheringState); - logger.e("[Test] logged after stop"); - logger.flushGatheredLogs(); - Assert.assertEquals(0, logger.heldLogLineCount()); - Assert.assertNull(logger.logDeliveryExecutor); - Assert.assertEquals(queuedBeforeStop + 1, queuedLogRequests.size()); - Assert.assertTrue("the delivery thread outlived the SDK", waitFor(5_000, () -> { - for (Thread thread : deliveryThreads) { - if (thread.isAlive()) { - return false; + List requests = new CopyOnWriteArrayList<>(); + HttpServer httpServer = startRecordingServer(requests); + try { + server.respondWith(new ServerConfigBuilder().logGatheringOn("gather_tick", "e", 100)); + init(localConfig(httpServer).setLogListener(listener)); + Log logger = logger(); + awaitSentBatches(); + + //the timer ticks through the modules, so it is sped up only after init built them + CountlyTimer.TIMER_DELAY_MS = 50; + push(new ServerConfigBuilder().sessionUpdateInterval(30).logGatheringOn("gather_tick", "e", 100)); + logger.e("[Test] tick line"); + Assert.assertTrue("no tick flushed the partial batch", TestUtils.waitFor(10_000, () -> !linesStartingWith("[Test] tick line", receivedBatches(requests)).isEmpty())); + + CountlyTimer.TIMER_DELAY_MS = 0; + push(new ServerConfigBuilder().networking(false).sessionUpdateInterval(60).logGatheringOn("gather_tick", "e", 100)); + //a tick of the replaced fast timer may still be running + Thread.sleep(250); + awaitLogDelivery(); + logger.e("[Test] tail line"); + Assert.assertEquals(1, logger.heldLogLineCount()); + Set deliveryThreads = liveDeliveryThreads(); + Assert.assertFalse(deliveryThreads.isEmpty()); + int queuedBeforeStop = queuedLogRequests.size(); + + Countly.instance().stop(); + Assert.assertEquals(queuedBeforeStop + 1, queuedLogRequests.size()); + JSONObject tail = null; + for (JSONObject batch : storedBatchesOnDisk()) { + if (messagesOf(batch).contains("[Test] tail line")) { + tail = batch; } } - return true; - })); + Assert.assertNotNull("stop did not queue the held line", tail); + Assert.assertEquals("gather_tick", tail.getString("i")); + Assert.assertEquals(Arrays.asList("[Test] tail line"), messagesOf(tail)); + + Assert.assertEquals(LogGatheringState.NOT_GATHERING, logger.logGatheringState); + logger.e("[Test] logged after stop"); + logger.flushGatheredLogs(); + Assert.assertEquals(0, logger.heldLogLineCount()); + Assert.assertNull(logger.logDeliveryExecutor); + Assert.assertEquals(queuedBeforeStop + 1, queuedLogRequests.size()); + Assert.assertTrue("the delivery thread outlived the SDK", TestUtils.waitFor(5_000, () -> { + for (Thread thread : deliveryThreads) { + if (thread.isAlive()) { + return false; + } + } + return true; + })); - server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_halt", "e", 100)); - init(TestUtils.getConfigSdkBehaviorSettings().setLogListener(listener)); - Log halted = logger(); - assertGathering("gather_halt", "e", 100); - awaitLogDelivery(); - halted.e("[Test] halted line"); - Assert.assertEquals(1, halted.heldLogLineCount()); - int queuedBeforeHalt = queuedLogRequests.size(); + server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_halt", "e", 100)); + init(TestUtils.getConfigSdkBehaviorSettings().setLogListener(listener)); + Log halted = logger(); + assertGathering("gather_halt", "e", 100); + awaitLogDelivery(); + int heldAtStart = halted.heldLogLineCount(); + halted.e("[Test] halted line"); + Assert.assertEquals(heldAtStart + 1, halted.heldLogLineCount()); + int queuedBeforeHalt = queuedLogRequests.size(); - Countly.instance().halt(); - Assert.assertEquals(queuedBeforeHalt, queuedLogRequests.size()); - Assert.assertEquals(LogGatheringState.NOT_GATHERING, halted.logGatheringState); - Assert.assertEquals(0, halted.heldLogLineCount()); + Countly.instance().halt(); + Assert.assertEquals(queuedBeforeHalt, queuedLogRequests.size()); + Assert.assertEquals(LogGatheringState.NOT_GATHERING, halted.logGatheringState); + Assert.assertEquals(0, halted.heldLogLineCount()); + } finally { + Countly.instance().halt(); + httpServer.stop(0); + } } /** @@ -466,49 +549,29 @@ public void timerTick_flushesAPartialBatch_stopQueuesWhatIsHeld_andHaltDiscardsI */ @Test public void uploadThreads_neverGatherTheirOwnLines_whileLinesAboutOrdinaryRequestsAreGathered() throws Exception { - List received = new CopyOnWriteArrayList<>(); - AtomicInteger eventRequests = new AtomicInteger(); - HttpServer httpServer = HttpServer.create(new InetSocketAddress("localhost", 0), 0); - httpServer.createContext("/", exchange -> { - String query = exchange.getRequestURI().getRawQuery(); - Map params = decode((query == null ? "" : query) + "&" + readAll(exchange.getRequestBody())); - if (params.containsKey(Log.keySdkLogs)) { - received.add(new JSONObject(params.get(Log.keySdkLogs))); - } - if (params.containsKey("events")) { - eventRequests.incrementAndGet(); - } - byte[] body = "{\"result\":\"Success\"}".getBytes(StandardCharsets.UTF_8); - exchange.sendResponseHeaders(200, body.length); - try (OutputStream os = exchange.getResponseBody()) { - os.write(body); - } - }); - httpServer.start(); + List requests = new CopyOnWriteArrayList<>(); + HttpServer httpServer = startRecordingServer(requests); List heard = new CopyOnWriteArrayList<>(); try { - Config config = new Config("http://localhost:" + httpServer.getAddress().getPort(), TestUtils.SERVER_APP_KEY, TestUtils.getTestSDirectory()) - .setApplicationVersion(TestUtils.APPLICATION_VERSION) - .setCustomDeviceId(TestUtils.DEVICE_ID) + Config config = localConfig(httpServer) .enableFeatures(Config.Feature.Events) .setEventQueueSizeToSend(1) - .setNetworkRequestCooldown(0) .setLogListener((message, level) -> heard.add(message)); server.respondWith(new ServerConfigBuilder().logGatheringOn("gather_net", "ewidv", 50)); init(config); - Assert.assertTrue("no upload arrived", waitFor(30_000, () -> !received.isEmpty())); - Assert.assertTrue("no upload completed", waitFor(30_000, () -> !completedRequests(heard, requestIds(heard, Log.transportMarker)).isEmpty())); + Assert.assertTrue("no upload arrived", TestUtils.waitFor(30_000, () -> !receivedBatches(requests).isEmpty())); + Assert.assertTrue("no upload completed", TestUtils.waitFor(30_000, () -> !completedRequests(heard, requestIds(heard, Log.transportMarker)).isEmpty())); Countly.instance().events().recordEvent("ordinary_event"); - Assert.assertTrue("the event request never arrived", waitFor(30_000, () -> eventRequests.get() > 0)); - Assert.assertTrue("the event request never completed", waitFor(30_000, () -> !completedRequests(heard, requestIds(heard, "events=")).isEmpty())); + Assert.assertTrue("the event request never arrived", TestUtils.waitFor(30_000, () -> requests.stream().anyMatch(request -> request.params.containsKey("events")))); + Assert.assertTrue("the event request never completed", TestUtils.waitFor(30_000, () -> !completedRequests(heard, requestIds(heard, "events=")).isEmpty())); Set eventIds = completedRequests(heard, requestIds(heard, "events=")); logger().flushGatheredLogs(); - Assert.assertTrue("the completion of the event request was not uploaded", waitFor(30_000, () -> { - for (String line : receivedLines(received)) { + Assert.assertTrue("the completion of the event request was not uploaded", TestUtils.waitFor(30_000, () -> { + for (String line : receivedLines(receivedBatches(requests))) { for (String eventId : eventIds) { if (line.startsWith("[Networking] Request " + eventId + " sent?: true")) { return true; @@ -520,10 +583,10 @@ public void uploadThreads_neverGatherTheirOwnLines_whileLinesAboutOrdinaryReques Set uploadIds = completedRequests(heard, requestIds(heard, Log.transportMarker)); Assert.assertFalse(uploadIds.isEmpty()); - for (JSONObject batch : received) { + for (JSONObject batch : receivedBatches(requests)) { Assert.assertEquals("gather_net", batch.getString("i")); } - for (String line : receivedLines(received)) { + for (String line : receivedLines(receivedBatches(requests))) { Assert.assertFalse(line, line.contains(Log.keySdkLogs)); Assert.assertFalse(line, line.startsWith("[Log] deliverHeldLogBatches")); for (String uploadId : uploadIds) { @@ -547,22 +610,19 @@ public void uploadThreads_neverGatherTheirOwnLines_whileLinesAboutOrdinaryReques */ @Test public void idleArmedSdk_uploadsNothingAboutItsOwnUploads_onTheTicksAfterAnUpload() throws Exception { - List received = new CopyOnWriteArrayList<>(); - HttpServer httpServer = startBatchServer(received); + List requests = new CopyOnWriteArrayList<>(); + HttpServer httpServer = startRecordingServer(requests); try { - Config config = new Config("http://localhost:" + httpServer.getAddress().getPort(), TestUtils.SERVER_APP_KEY, TestUtils.getTestSDirectory()) - .setApplicationVersion(TestUtils.APPLICATION_VERSION) - .setCustomDeviceId(TestUtils.DEVICE_ID) - .setNetworkRequestCooldown(0); server.respondWith(new ServerConfigBuilder().logGatheringOn("gather_idle", "ewidv", 100)); - init(config); + init(localConfig(httpServer)); - awaitIdleTicks(received); + awaitIdleTicks(requests); logger().e("[Test] idle probe"); - Assert.assertEquals(1, tickAndCountBatches(received)); + Assert.assertEquals(1, tickAndCountBatches(requests)); + List received = receivedBatches(requests); Assert.assertTrue(messagesOf(received.get(received.size() - 1)).contains("[Test] idle probe")); for (int tick = 0; tick < 3; tick++) { - Assert.assertEquals("tick " + tick, 0, tickAndCountBatches(received)); + Assert.assertEquals("tick " + tick, 0, tickAndCountBatches(requests)); } Assert.assertEquals(0, logger().heldLogLineCount()); } finally { @@ -572,27 +632,146 @@ public void idleArmedSdk_uploadsNothingAboutItsOwnUploads_onTheTicksAfterAnUploa } /** - * An armed SDK that is otherwise idle queues nothing about its own batches while networking is - * off. + * While nothing is sent, one log batch waits in the request queue and the rest stays in the + * capped buffer; once networking is back on, the batches go out one after another until nothing + * is held. + *

+ * Verifies, with networking off, that full batches and flushes queue nothing while a batch + * waits, the lines staying held; then, with networking on against a local server, that every + * held line arrives once, in order, with never more than one batch stored at a time. + */ + @Test + public void networkingOff_queuesOneBatchAtATime_andTheRestFollowsOnceItIsSent() throws Exception { + List requests = new CopyOnWriteArrayList<>(); + HttpServer httpServer = startRecordingServer(requests); + try { + server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_offline", "e", 10)); + init(localConfig(httpServer)); + awaitLogDelivery(); + Log logger = logger(); + for (int i = 0; i < 30; i++) { + logger.e("[Test] offline line " + i); + } + for (int flush = 0; flush < 3; flush++) { + logger.flushGatheredLogs(); + awaitLogDelivery(); + } + Assert.assertEquals(1, storedBatchCount()); + Assert.assertTrue(logger.heldLogLineCount() >= 20); + + push(new ServerConfigBuilder().logGatheringOn("gather_offline", "e", 10)); + int mostStored = 0; + long end = System.currentTimeMillis() + 30_000; + while (System.currentTimeMillis() < end) { + mostStored = Math.max(mostStored, storedBatchCount()); + if (logger.heldLogLineCount() == 0 && storedBatchCount() == 0) { + break; + } + Thread.sleep(5); + } + awaitSentBatches(); + Assert.assertEquals(1, mostStored); + Assert.assertEquals(numbered("[Test] offline line ", 0, 30), linesStartingWith("[Test] offline line ", receivedBatches(requests))); + } finally { + Countly.instance().halt(); + httpServer.stop(0); + } + } + + /** + * A device ID change without merge queues the gathered lines under the previous device ID and + * stops the gather, an answer requested for a previous device ID does not arm it again, and a + * change with merge keeps a gather running. + *

+ * Verifies, with networking off, that the line logged before the change is stored before the + * change returns, under the previous device ID and with the gather id; that nothing logged + * afterwards is held; that a settings answer requested before a later change and applied after + * it leaves the gather stopped while its other settings apply; that an answer for the current + * device ID arms a gather again; that a change with merge leaves that gather running; and that + * an answer requested before a change with merge arms its gather. + */ + @Test + public void deviceIdChangeWithoutMerge_queuesTheLinesUnderThePreviousId_andStopsTheGather() throws Exception { + server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_user_a", "e", 100)); + init(TestUtils.getConfigSdkBehaviorSettings()); + awaitLogDelivery(); + logger().e("[Test] line of user A"); + Countly.instance().deviceId().changeWithoutMerge("user_B"); + + Map userA = null; + for (Map request : logRequests()) { + if (request.get(Log.keySdkLogs).contains("[Test] line of user A")) { + userA = request; + } + } + Assert.assertNotNull("the line of user A was not queued before the change", userA); + Assert.assertEquals(TestUtils.DEVICE_ID, userA.get("device_id")); + Assert.assertEquals("gather_user_a", new JSONObject(userA.get(Log.keySdkLogs)).getString("i")); + assertNotGathering(); + logger().e("[Test] line of user B"); + Assert.assertEquals(0, logger().heldLogLineCount()); + + AtomicReference heldAnswer = new AtomicReference<>(); + InternalConfig config = SDKCore.instance.config; + config.immediateRequestGenerator = () -> (requestData, customEndpoint, transport, requestShouldBeDelayed, networkingIsEnabled, callback, log) -> heldAnswer.set(callback); + module().fetchConfigFromServer(config); + Assert.assertNotNull(heldAnswer.get()); + Countly.instance().deviceId().changeWithoutMerge("user_C"); + heldAnswer.get().callback(new JSONObject(new ServerConfigBuilder().networking(false).sessionUpdateInterval(45).logGatheringOn("gather_stale", "e", 100).build())); + assertNotGathering(); + Assert.assertEquals(45, module().getSessionUpdateInterval()); + + config.immediateRequestGenerator = server; + push(new ServerConfigBuilder().networking(false).logGatheringOn("gather_user_c", "e", 100)); + assertGathering("gather_user_c", "e", 100); + Countly.instance().deviceId().changeWithMerge("user_D"); + assertGathering("gather_user_c", "e", 100); + + config.immediateRequestGenerator = () -> (requestData, customEndpoint, transport, requestShouldBeDelayed, networkingIsEnabled, callback, log) -> heldAnswer.set(callback); + module().fetchConfigFromServer(config); + Countly.instance().deviceId().changeWithMerge("user_E"); + heldAnswer.get().callback(new JSONObject(new ServerConfigBuilder().networking(false).logGatheringOn("gather_after_merge", "e", 100).build())); + assertGathering("gather_after_merge", "e", 100); + } + + /** + * A gather the server stops queues the lines it held once, with its id, and a gather replaced + * by a new id drops them. *

- * Verifies on disk that once the lines logged before were queued a tick queues nothing, that a - * line logged then is queued in one batch on the next tick, and that the ticks after that queue - * nothing, as neither storing a batch nor the drain it asks for, which networking off skips, is - * gathered. + * Verifies, with networking off and tracking on, that after {@code "e":false} exactly the lines + * held until then are stored, in one batch with the stopped gather's id, past the batch already + * waiting, and nothing logged afterwards; and that a gather replaced by a new id holds none of + * the lines of the one before. */ @Test - public void idleArmedSdk_withNetworkingOff_queuesNothingAboutItsOwnBatches() throws Exception { - server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_idle_off", "ewidv", 100)); + public void stoppedGather_queuesItsHeldLinesOnce_whileAReplacedGatherDropsThem() throws Exception { + server.respondWith(new ServerConfigBuilder().networking(false).logGatheringOn("gather_stopped", "e", 100)); init(TestUtils.getConfigSdkBehaviorSettings()); + awaitLogDelivery(); + logger().e("[Test] stopped 1"); + logger().e("[Test] stopped 2"); - awaitIdleTicks(null); - logger().e("[Test] idle probe"); - Assert.assertEquals(1, tickAndCountBatches(null)); - List queued = batches(); - Assert.assertTrue(messagesOf(queued.get(queued.size() - 1)).contains("[Test] idle probe")); - for (int tick = 0; tick < 3; tick++) { - Assert.assertEquals("tick " + tick, 0, tickAndCountBatches(null)); + push(new ServerConfigBuilder().networking(false).logGatheringOff()); + assertNotGathering(); + Assert.assertTrue("the held lines were not queued", TestUtils.waitFor(10_000, () -> linesStartingWith("[Test] stopped ", batches()).size() == 2)); + JSONObject tail = null; + for (JSONObject batch : batches()) { + if (messagesOf(batch).contains("[Test] stopped 1")) { + tail = batch; + } } + Assert.assertNotNull(tail); + Assert.assertEquals("gather_stopped", tail.getString("i")); + Assert.assertEquals(Arrays.asList("[Test] stopped 1", "[Test] stopped 2"), linesStartingWith("[Test] stopped ", Arrays.asList(tail))); + logger().e("[Test] after the stop"); + Thread.sleep(200); + Assert.assertTrue(linesStartingWith("[Test] after the stop", batches()).isEmpty()); + + push(new ServerConfigBuilder().networking(false).logGatheringOn("gather_first", "e", 100)); + logger().e("[Test] first gather line"); + Assert.assertTrue(logger().heldLogLineCount() >= 1); + push(new ServerConfigBuilder().networking(false).logGatheringOn("gather_second", "e", 100)); + assertGathering("gather_second", "e", 100); Assert.assertEquals(0, logger().heldLogLineCount()); } @@ -617,19 +796,19 @@ public void queueingABatch_givenUpWhileTheStorageIsBusy_cancelsItsStore() throws storageBusy.countDown(); storageFree.await(30, TimeUnit.SECONDS); }); - AtomicReference queuedBatch = new AtomicReference<>(); + AtomicReference queuedBatch = new AtomicReference<>(-1L); Thread queueing = new Thread(() -> queuedBatch.set(ModuleRequests.pushSdkLogs(config, "{\"i\":\"gather_cancel\",\"d\":0,\"l\":[]}"))); try { Assert.assertTrue(storageBusy.await(30, TimeUnit.SECONDS)); queueing.start(); - Assert.assertTrue("the queueing never waited for its store", waitFor(30_000, () -> isWaitingForAFuture(queueing))); + Assert.assertTrue("the queueing never waited for its store", TestUtils.waitFor(30_000, () -> isWaitingForAFuture(queueing))); queueing.interrupt(); queueing.join(30_000); } finally { storageFree.countDown(); } - Assert.assertEquals(Boolean.FALSE, queuedBatch.get()); + Assert.assertNull(queuedBatch.get()); Storage.await(mock(Log.class)); Assert.assertEquals(storedBefore, logRequests().size()); } @@ -638,20 +817,36 @@ public void queueingABatch_givenUpWhileTheStorageIsBusy_cancelsItsStore() throws // region helpers /** - * Starts a local server that accepts every request and keeps the log batches it receives. + * A request a local server received. + */ + private static final class ReceivedRequest { + final String method; + final Map params; + + /** + * Holds one request. + * + * @param method the HTTP method it came with + * @param params its parameters, from the query and the body + */ + ReceivedRequest(String method, Map params) { + this.method = method; + this.params = params; + } + } + + /** + * Starts a local server that accepts every request and records it. * - * @param received where the batches go, in arrival order + * @param requests where the requests go, in arrival order * @return the started server * @throws IOException when the server cannot start */ - private static HttpServer startBatchServer(List received) throws IOException { + private static HttpServer startRecordingServer(List requests) throws IOException { HttpServer httpServer = HttpServer.create(new InetSocketAddress("localhost", 0), 0); httpServer.createContext("/", exchange -> { String query = exchange.getRequestURI().getRawQuery(); - Map params = decode((query == null ? "" : query) + "&" + readAll(exchange.getRequestBody())); - if (params.containsKey(Log.keySdkLogs)) { - received.add(new JSONObject(params.get(Log.keySdkLogs))); - } + requests.add(new ReceivedRequest(exchange.getRequestMethod(), decode((query == null ? "" : query) + "&" + readAll(exchange.getRequestBody())))); byte[] body = "{\"result\":\"Success\"}".getBytes(StandardCharsets.UTF_8); exchange.sendResponseHeaders(200, body.length); try (OutputStream os = exchange.getResponseBody()) { @@ -662,16 +857,81 @@ private static HttpServer startBatchServer(List received) throws IOE return httpServer; } + /** + * The log batches among the requests a local server received. + * + * @param requests the requests, in arrival order + * @return the batches, in arrival order + */ + private static List receivedBatches(List requests) { + List received = new ArrayList<>(); + for (ReceivedRequest request : requests) { + if (request.params.containsKey(Log.keySdkLogs)) { + received.add(new JSONObject(request.params.get(Log.keySdkLogs))); + } + } + return received; + } + + /** + * The test configuration, sending to a local server instead of the test server. + * + * @param httpServer the local server + * @return the configuration + */ + private static Config localConfig(HttpServer httpServer) { + return new Config("http://localhost:" + httpServer.getAddress().getPort(), TestUtils.SERVER_APP_KEY, TestUtils.getTestSDirectory()) + .setApplicationVersion(TestUtils.APPLICATION_VERSION) + .setCustomDeviceId(TestUtils.DEVICE_ID) + .setNetworkRequestCooldown(0); + } + + /** + * Waits until every batch the logger queued was sent and no delivery is left to run, as each + * sent batch lets the next one go. + * + * @throws Exception when the batches are not all sent in time + */ + private static void awaitSentBatches() throws Exception { + for (int round = 0; round < 200; round++) { + awaitLogDelivery(); + Assert.assertTrue("a queued batch never reached the server", TestUtils.waitFor(30_000, () -> storedBatchCount() == 0)); + awaitLogDelivery(); + if (storedBatchCount() == 0) { + return; + } + } + Assert.fail("the logger kept queueing batches"); + } + + /** + * The log batches stored on disk, read straight from the request files, which works once the + * SDK stopped too. + * + * @return the batches + * @throws IOException when a request file cannot be read + */ + private static List storedBatchesOnDisk() throws IOException { + List stored = new ArrayList<>(); + for (String params : storedRequestParams()) { + Map decoded = decode(params); + if (decoded.containsKey(Log.keySdkLogs)) { + stored.add(new JSONObject(decoded.get(Log.keySdkLogs))); + } + } + return stored; + } + /** * Ticks until a tick queues no batch, which an idle SDK reaches once the lines it logged before * went out, as nothing it logs about a batch is gathered. * - * @param received the batches a local server received, {@code null} while networking is off + * @param requests the requests a local server received * @throws Exception when a batch does not go out in time */ - private static void awaitIdleTicks(@Nullable List received) throws Exception { + private static void awaitIdleTicks(List requests) throws Exception { for (int tick = 0; tick < 10; tick++) { - if (tickAndCountBatches(received) == 0) { + if (tickAndCountBatches(requests) == 0) { return; } } @@ -679,37 +939,33 @@ private static void awaitIdleTicks(@Nullable List received) throws E } /** - * Flushes what is held, as a timer tick does, waits until the batches it queued are stored, and - * sent while networking is on, then lets the drain that follows them run. + * Flushes what is held, as a timer tick does, waits until the batches it queued are sent, then + * lets the drain that follows them run. * - * @param received the batches a local server received, {@code null} while networking is off + * @param requests the requests a local server received * @return how many batches the tick queued * @throws Exception when a batch does not go out in time */ - private static int tickAndCountBatches(@Nullable List received) throws Exception { - int before = settledBatchCount(received); + private static int tickAndCountBatches(List requests) throws Exception { + int before = settledBatchCount(requests); logger().flushGatheredLogs(); awaitLogDelivery(); - int after = settledBatchCount(received); + int after = settledBatchCount(requests); //nothing tells when the drain a stored batch asked for has finished, so it gets a moment Thread.sleep(200); return after - before; } /** - * How many batches were queued so far, counted while none is in flight: the stored ones while - * networking is off, otherwise the ones a local server received once every stored one was sent. + * How many batches a local server received, counted once every stored one was sent. * - * @param received the batches a local server received, {@code null} while networking is off + * @param requests the requests a local server received * @return the number of batches * @throws InterruptedException when interrupted while waiting */ - private static int settledBatchCount(@Nullable List received) throws InterruptedException { - if (received == null) { - return storedBatchCount(); - } - Assert.assertTrue("a queued batch never reached the server", waitFor(30_000, () -> storedBatchCount() == 0)); - return received.size(); + private static int settledBatchCount(List requests) throws InterruptedException { + Assert.assertTrue("a queued batch never reached the server", TestUtils.waitFor(30_000, () -> storedBatchCount() == 0)); + return receivedBatches(requests).size(); } /** @@ -770,6 +1026,22 @@ private void push(ServerConfigBuilder builder) { module().fetchConfigFromServer(SDKCore.instance.config); } + /** + * Answers the next settings request with a payload that stops the gather, makes the SDK fetch, + * and waits until the delivery thread offered the last batch of the stopped gather to the + * request queue and ended, so the settings that follow cannot change how the queue answers. + * + * @param builder the payload + * @throws InterruptedException when interrupted while waiting + */ + private void pushStop(ServerConfigBuilder builder) throws InterruptedException { + ExecutorService executor = logger().logDeliveryExecutor; + push(builder); + if (executor != null) { + Assert.assertTrue("the delivery thread did not end", executor.awaitTermination(30, TimeUnit.SECONDS)); + } + } + /** * An armed directive next to settings that switch tracking off, so the gathered lines are held, * and networking off, so nothing queued is sent. @@ -936,22 +1208,6 @@ private static List receivedLines(List received) { return messages; } - /** - * Whether any queued log request a listener heard of carries the given line. - * - * @param queuedLogRequests the lines the SDK logged when it queued each log request - * @param line the line - * @return {@code true} once a queued batch carries it - */ - private static boolean queuedLinesContain(List queuedLogRequests, String line) { - for (String queued : queuedLogRequests) { - if (Utils.urldecode(queued).contains(line)) { - return true; - } - } - return false; - } - /** * The storage IDs of the queued requests whose parameters contain the given text, read off * the lines the SDK logs as it queues a request. @@ -1093,24 +1349,6 @@ private static String repeat(char character, int count) { return text.toString(); } - /** - * Polls a condition until it holds or the time is up. - * - * @param timeoutMs how long to wait, in milliseconds - * @param condition the condition - * @return whether it held in time - * @throws InterruptedException when interrupted while waiting - */ - private static boolean waitFor(long timeoutMs, BooleanSupplier condition) throws InterruptedException { - long deadline = System.currentTimeMillis() + timeoutMs; - while (System.currentTimeMillis() < deadline) { - if (condition.getAsBoolean()) { - return true; - } - Thread.sleep(20); - } - return condition.getAsBoolean(); - } // endregion } diff --git a/sdk-java/src/test/java/ly/count/sdk/java/internal/TestUtils.java b/sdk-java/src/test/java/ly/count/sdk/java/internal/TestUtils.java index 356f935d1..fd0a0e593 100644 --- a/sdk-java/src/test/java/ly/count/sdk/java/internal/TestUtils.java +++ b/sdk-java/src/test/java/ly/count/sdk/java/internal/TestUtils.java @@ -13,6 +13,7 @@ import java.util.Scanner; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BooleanSupplier; import java.util.regex.Matcher; import java.util.regex.Pattern; import java.util.stream.Stream; @@ -740,4 +741,23 @@ static void letTheClockCatchUp() { } } } + + /** + * Polls a condition until it holds or the time is up. + * + * @param timeoutMs how long to wait at most, in milliseconds + * @param condition the condition + * @return whether it held in time + * @throws InterruptedException when interrupted while waiting + */ + static boolean waitFor(long timeoutMs, BooleanSupplier condition) throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMs; + while (System.currentTimeMillis() < deadline) { + if (condition.getAsBoolean()) { + return true; + } + Thread.sleep(10); + } + return condition.getAsBoolean(); + } }