Single executor for flags preload

main/src/androidTest/java/tests/integration/rollout/FreshInstallPrefetchPersistenceIntegrationTest.java

@@ -0,0 +1,131 @@
+package tests.integration.rollout;
+
+import static helper.IntegrationHelper.dummyApiKey;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+import android.content.Context;
+
+import androidx.test.platform.app.InstrumentationRegistry;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import helper.DatabaseHelper;
+import io.split.android.client.SplitFilter;
+import io.split.android.client.dtos.Split;
+import io.split.android.client.dtos.SplitChange;
+import io.split.android.client.dtos.Status;
+import io.split.android.client.service.splits.SplitChangeProcessor;
+import io.split.android.client.storage.cipher.SplitCipher;
+import io.split.android.client.storage.cipher.SplitCipherFactory;
+import io.split.android.client.storage.db.SplitRoomDatabase;
+import io.split.android.client.storage.splits.PersistentSplitsStorage;
+import io.split.android.client.storage.splits.SplitsStorage;
+import io.split.android.client.storage.splits.SplitsStorageImpl;
+import io.split.android.client.storage.splits.SqLitePersistentSplitsStorage;
+
+public class FreshInstallPrefetchPersistenceIntegrationTest {
+
+    private static final long CHANGE_NUMBER = 1778482333302L;
+    private static final int SPLIT_COUNT_OVER_ASYNC_THRESHOLD = 60;
+    private static final String FIRST_FLAG_NAME = "fresh_install_flag_0";
+
+    private SplitRoomDatabase mRoomDb;
+    private PersistentSplitsStorage mPersistentStorage;
+    private SplitChangeProcessor mSplitChangeProcessor;
+
+    @Before
+    public void setUp() {
+        Context context = InstrumentationRegistry.getInstrumentation().getContext();
+        mRoomDb = DatabaseHelper.getTestDatabase(context);
+        mRoomDb.clearAllTables();
+
+        SplitCipher cipher = SplitCipherFactory.create(dummyApiKey(), false);
+        mPersistentStorage = new SqLitePersistentSplitsStorage(mRoomDb, cipher);
+        mSplitChangeProcessor = new SplitChangeProcessor((Map<SplitFilter.Type, SplitFilter>) null, null);
+    }
+
+    @Test
+    public void processKillBeforeAsyncWriteCompletes_dbRemainsConsistent() throws InterruptedException {
+        // Block the executor so the first write doesn't complete
+        CountDownLatch blockLatch = new CountDownLatch(1);
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        executor.submit(() -> {
+            try {
+                blockLatch.await();
+            } catch (InterruptedException e) {
+                // shutdownNow will interrupt this
+            }
+        });
+
+        SplitsStorage storage = new SplitsStorageImpl(mPersistentStorage);
+
+        // First update queues behind the blocked task
+        storage.update(
+                mSplitChangeProcessor.process(SplitChange.create(-1, CHANGE_NUMBER, createSplits())),
+                executor);
+
+        // Simulate process kill — first write never completes
+        executor.shutdownNow();
+        executor.awaitTermination(1, TimeUnit.SECONDS);
+
+        // Second update (empty delta) — submit is rejected since executor is shut down
+        storage.update(
+                mSplitChangeProcessor.process(SplitChange.create(CHANGE_NUMBER, CHANGE_NUMBER, new ArrayList<>())),
+                executor);
+
+        // DB should be untouched — no partial CN write
+        SplitsStorage reloadedStorage = new SplitsStorageImpl(mPersistentStorage);
+        reloadedStorage.loadLocal();
+
+        assertEquals(-1, reloadedStorage.getTill());
+        assertEquals(0, mRoomDb.splitDao().getAll().size());
+    }
+
+    @Test
+    public void fullSnapshotAndEmptyDeltaPersistCorrectlyWhenExecutorIsRunning() throws InterruptedException {
+        ExecutorService executor = Executors.newSingleThreadExecutor();
+        SplitsStorage storage = new SplitsStorageImpl(mPersistentStorage);
+
+        storage.update(
+                mSplitChangeProcessor.process(SplitChange.create(-1, CHANGE_NUMBER, createSplits())),
+                executor);
+        storage.update(
+                mSplitChangeProcessor.process(SplitChange.create(CHANGE_NUMBER, CHANGE_NUMBER, new ArrayList<>())),
+                executor);
+
+        executor.shutdown();
+        executor.awaitTermination(5, TimeUnit.SECONDS);
+
+        SplitsStorage reloadedStorage = new SplitsStorageImpl(mPersistentStorage);
+        reloadedStorage.loadLocal();
+
+        assertEquals(CHANGE_NUMBER, reloadedStorage.getTill());
+        assertEquals(SPLIT_COUNT_OVER_ASYNC_THRESHOLD, mRoomDb.splitDao().getAll().size());
+        assertNotNull(reloadedStorage.get(FIRST_FLAG_NAME));
+    }
+
+    private static List<Split> createSplits() {
+        List<Split> splits = new ArrayList<>();
+        for (int i = 0; i < SPLIT_COUNT_OVER_ASYNC_THRESHOLD; i++) {
+            Split split = new Split();
+            split.name = "fresh_install_flag_" + i;
+            split.status = Status.ACTIVE;
+            split.changeNumber = CHANGE_NUMBER;
+            split.trafficTypeName = "user";
+            split.defaultTreatment = "on";
+            split.killed = false;
+            splits.add(split);
+        }
+        return splits;
+    }
+}

main/src/androidTest/java/tests/service/CompressionTest.java

@@ -8,18 +8,17 @@
 import org.junit.Before;
 import org.junit.Test;
 
-import java.nio.charset.Charset;
 import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
 
 import helper.CompressionHelper;
 import helper.FileHelper;
+import io.split.android.client.streaming.support.CompressionUtil;
+import io.split.android.client.streaming.support.Gzip;
+import io.split.android.client.streaming.support.Zlib;
 import io.split.android.client.utils.Base64Util;
-import io.split.android.client.utils.CompressionUtil;
-import io.split.android.client.utils.Gzip;
 import io.split.android.client.utils.StringHelper;
-import io.split.android.client.utils.Zlib;
 
 public class CompressionTest {
 

main/src/androidTest/java/tests/service/MySegmentV2PayloadDecoderTest.java

@@ -17,9 +17,8 @@
 import io.split.android.client.service.sseclient.notifications.KeyList;
 import io.split.android.client.service.sseclient.notifications.MySegmentsV2PayloadDecoder;
 import io.split.android.client.service.sseclient.notifications.NotificationParser;
-import io.split.android.client.utils.Gzip;
-import io.split.android.client.utils.MurmurHash3;
-import io.split.android.client.utils.Zlib;
+import io.split.android.client.streaming.support.Gzip;
+import io.split.android.client.streaming.support.Zlib;
 
 public class MySegmentV2PayloadDecoderTest {
 

main/src/androidTest/java/tests/storage/SplitsStorageTest.java

@@ -2,6 +2,7 @@
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -20,7 +21,10 @@
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Queue;
 import java.util.Set;
+import java.util.concurrent.AbstractExecutorService;
+import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
@@ -32,7 +36,9 @@
 import io.split.android.client.storage.db.GeneralInfoEntity;
 import io.split.android.client.storage.db.SplitEntity;
 import io.split.android.client.storage.db.SplitRoomDatabase;
+import io.split.android.client.storage.splits.PersistentSplitsStorage;
 import io.split.android.client.storage.splits.ProcessedSplitChange;
+import io.split.android.client.storage.splits.SplitsSnapshot;
 import io.split.android.client.storage.splits.SplitsStorage;
 import io.split.android.client.storage.splits.SplitsStorageImpl;
 import io.split.android.client.storage.splits.SqLitePersistentSplitsStorage;
@@ -534,6 +540,34 @@
         assertEquals("", flagsSpec);
     }
 
+    @Test
+    public void asyncPersistentUpdateReceivesMetadataSnapshot() {
+        CapturingPersistentSplitsStorage persistentStorage = new CapturingPersistentSplitsStorage();
+        ControlledExecutorService executor = new ControlledExecutorService();
+        SplitsStorage splitsStorage = new SplitsStorageImpl(persistentStorage);
+
+        splitsStorage.update(
+                new ProcessedSplitChange(
+                        Collections.singletonList(newSplit("split_1", Status.ACTIVE, "type_1", Collections.singleton("set_1"))),
+                        Collections.emptyList(),
+                        1L,
+                        0L),
+                executor);
+        splitsStorage.update(
+                new ProcessedSplitChange(
+                        Collections.singletonList(newSplit("split_2", Status.ACTIVE, "type_2", Collections.singleton("set_2"))),
+                        Collections.emptyList(),
+                        2L,
+                        0L),
+                executor);
+
+        executor.runNext();
+
+        assertEquals(Collections.singletonMap("type_1", 1), persistentStorage.lastTrafficTypes);
+        assertEquals(Collections.singleton("split_1"), persistentStorage.lastFlagSets.get("set_1"));
+        assertNull(persistentStorage.lastFlagSets.get("set_2"));
+    }
+
     private Split newSplit(String name, Status status, String trafficType) {
         return newSplit(name, status, trafficType, Collections.emptySet());
     }
@@ -564,4 +598,108 @@
 
         return entity;
     }
+
+    private static class ControlledExecutorService extends AbstractExecutorService {
+        private final Queue<Runnable> tasks = new ConcurrentLinkedQueue<>();
+
+        void runNext() {
+            Runnable task = tasks.poll();
+            assertNotNull(task);
+            task.run();
+        }
+
+        @Override
+        public void shutdown() {
+        }
+
+        @Override
+        public List<Runnable> shutdownNow() {
+            return Collections.emptyList();
+        }
+
+        @Override
+        public boolean isShutdown() {
+            return false;
+        }
+
+        @Override
+        public boolean isTerminated() {
+            return false;
+        }
+
+        @Override
+        public boolean awaitTermination(long timeout, TimeUnit unit) {
+            return false;
+        }
+
+        @Override
+        public void execute(Runnable command) {
+            tasks.add(command);
+        }
+    }
+
+    private static class CapturingPersistentSplitsStorage implements PersistentSplitsStorage {
+        Map<String, Integer> lastTrafficTypes;
+        Map<String, Set<String>> lastFlagSets;
+
+        @Override
+        public boolean update(ProcessedSplitChange splitChange, Map<String, Integer> trafficTypes, Map<String, Set<String>> flagSets) {
+            lastTrafficTypes = new HashMap<>(trafficTypes);
+            lastFlagSets = new HashMap<>();
+            for (Map.Entry<String, Set<String>> entry : flagSets.entrySet()) {
+                lastFlagSets.put(entry.getKey(), new HashSet<>(entry.getValue()));
+            }
+            return true;
+        }
+
+        @Override
+        public SplitsSnapshot getSnapshot() {
+            return new SplitsSnapshot(Collections.emptyList(), -1L, 0L, "", "", Collections.emptyMap(), Collections.emptyMap());
+        }
+
+        @Override
+        public List<Split> getAll() {
+            return Collections.emptyList();
+        }
+
+        @Override
+        public void update(Split splitName) {
+            // no-op
+        }
+
+        @Override
+        public String getFilterQueryString() {
+            return "";
+        }
+
+        @Override
+        public void updateFilterQueryString(String queryString) {
+            // no-op
+        }
+
+        @Override
+        public String getFlagsSpec() {
+            return "";
+        }
+
+        @Override
+        public void updateFlagsSpec(String flagsSpec) {
+            // no-op
+        }
+
+        @Override
+        public void delete(List<String> splitNames) {
+            // no-op
+        }
+
+        @Override
+        public void clear() {
+            // no-op
+        }
+
+        @Override
+        public void close() {
+            // no-op
+        }
+    }
 }

main/src/main/java/io/split/android/client/service/splits/SplitsSyncHelper.java

@@ -331,7 +331,9 @@ private void updateStorage(boolean clearBeforeUpdate, SplitChange splitChange, R
             mLastProcessedSplitChange.set(processedSplitChange);
         }
         mSplitsStorage.update(processedSplitChange, mExecutor);
-        updateRbsStorage(ruleBasedSegmentChange);
+        if (ruleBasedSegmentChange != null) {
+            updateRbsStorage(ruleBasedSegmentChange);
+        }
     }
 
     private boolean hasFlagUpdates(@Nullable ProcessedSplitChange processedSplitChange) {

main/src/main/java/io/split/android/client/storage/splits/SplitsStorageImpl.java

@@ -24,12 +24,11 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import io.split.android.client.dtos.Split;
+import io.split.android.client.utils.logger.Logger;
 import io.split.android.client.utils.Json;
 
 public class SplitsStorageImpl implements SplitsStorage {
 
-    private static final int ASYNC_WRITE_THRESHOLD = 50;
-
     private final PersistentSplitsStorage mPersistentStorage;
     private final Map<String, Split> mInMemorySplits;
     private final Map<String, Set<String>> mFlagSets;
@@ -127,7 +126,7 @@
 
     @Override
     @WorkerThread
     public boolean update(ProcessedSplitChange splitChange, ExecutorService mExecutor) {
         if (splitChange == null) {
             return false;
         }
@@ -166,17 +165,30 @@
         mChangeNumber = splitChange.getChangeNumber();
         mUpdateTimestamp = splitChange.getUpdateTimestamp();
 
-        // If the amount of elements is greater than the threshold,
-        // we will use the executor to update the persistent storage asynchronously
-        if (((activeSplits != null && activeSplits.size() > ASYNC_WRITE_THRESHOLD) || (archivedSplits != null && archivedSplits.size() > ASYNC_WRITE_THRESHOLD)) && mExecutor != null) {
-            mExecutor.submit(() -> mPersistentStorage.update(splitChange, mTrafficTypes, mFlagSets));
+        if (mExecutor != null) {
+            try {
+                Map<String, Integer> trafficTypesSnapshot = new HashMap<>(mTrafficTypes);
+                Map<String, Set<String>> flagSetsSnapshot = copyFlagSets(mFlagSets);
+                mExecutor.submit(() -> mPersistentStorage.update(splitChange, trafficTypesSnapshot, flagSetsSnapshot));
+            } catch (Exception e) {
+                Logger.v("Failed to submit persistent write: " + e.getLocalizedMessage());
+            }
         } else {
             mPersistentStorage.update(splitChange, mTrafficTypes, mFlagSets);
         }
 
         return appliedUpdates;
     }
 
+    @NonNull
+    private static Map<String, Set<String>> copyFlagSets(Map<String, Set<String>> flagSets) {
+        Map<String, Set<String>> flagSetsSnapshot = new HashMap<>();
+        for (Map.Entry<String, Set<String>> entry : flagSets.entrySet()) {
+            flagSetsSnapshot.put(entry.getKey(), new HashSet<>(entry.getValue()));
+        }
+        return flagSetsSnapshot;
+    }
+
     @Override
     @WorkerThread
     public void updateWithoutChecks(Split split) {