[SDK-499] Synchronize EmbeddedSessionManager to fix embedded session races

CHANGELOG.md

@@ -9,6 +9,7 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 ### Fixed
 - Fixed the keychain treating a transient crypto timeout as a permanent decryption failure. A slow AndroidKeyStore operation that exceeded the 500 ms timeout would wipe the stored email, userId, and auth token and disable encryption, forcing the user to re-authenticate (and request a new auth token) on the next launch. Crypto timeouts are now handled as transient without wiping credentials or disabling encryption for the device: a read that times out returns no value for that call (the stored ciphertext is left intact for the next attempt), and a write that times out stores that one value unencrypted (as the non-encrypted fallback already did) rather than clearing everything. The timed-out crypto operation is also cancelled so it no longer blocks subsequent reads/writes.
 - `setExpiringAuthTokenRefreshPeriod` now validates its input instead of silently producing a broken refresh schedule. Previously a negative value was converted to a negative millisecond period and then *subtracted* when computing the refresh time, scheduling the refresh after the token had already expired; a very large value overflowed to a negative period with the same effect; and `null` threw a `NullPointerException` on unboxing. Invalid values (`null`, `NaN`, negatives) are now logged and ignored, leaving the period at whatever it was before the call — the 60 second default unless an earlier call set something else. Values above ~10 years are clamped to that ceiling rather than ignored. Zero remains valid and means the token is refreshed only once it has expired.
+- Fixed a `NullPointerException` in `EmbeddedSessionManager.updateDisplayCountAndDuration()` that could crash apps calling embedded session methods off the main thread. `EmbeddedSessionManager` is now internally synchronized, which also fixes concurrent modification of its impression map and duplicate session tracking when `endSession()` raced with itself. Thanks to [@Shamyyoun](https://github.com/Shamyyoun) for the report and initial fix.
 
 ### Changed
 - Clarified that `setExpiringAuthTokenRefreshPeriod` takes **seconds**, with a default of 60. The unit and default are unchanged and match every other Iterable SDK.

iterableapi/src/main/java/com/iterable/iterableapi/EmbeddedSessionManager.kt

@@ -6,48 +6,58 @@ public class EmbeddedSessionManager {
 
     private val TAG = "EmbeddedSessionManager"
 
+    // Callers reach this class from arbitrary threads (see issue #1052), so every access to
+    // impressions, session, and the impression fields happens under this lock.
+    private val lock = Any()
+
     private var impressions: MutableMap<String, EmbeddedImpressionData> = mutableMapOf()
 
     var session: IterableEmbeddedSession = IterableEmbeddedSession(
         null,
         null,
         null
     )
+        get() = synchronized(lock) { field }
+        set(value) = synchronized(lock) { field = value }
 
     fun isTracking(): Boolean {
-        return session.start != null
+        return synchronized(lock) { session.start != null }
     }
 
     fun startSession() {
-        if (isTracking()) {
-            IterableLogger.e(TAG, "Embedded session started twice")
-            return
-        }
+        synchronized(lock) {
+            if (isTracking()) {
+                IterableLogger.e(TAG, "Embedded session started twice")
+                return
+            }
 
-        session = IterableEmbeddedSession(
-            Date(),
-            null,
-            null
-        )
+            session = IterableEmbeddedSession(
+                Date(),
+                null,
+                null
+            )
+        }
     }
 
     fun endSession() {
-        if (!isTracking()) {
-            IterableLogger.e(TAG, "Embedded session ended without start")
-            return
-        }
+        val sessionToTrack = synchronized(lock) {
+            if (!isTracking()) {
+                IterableLogger.e(TAG, "Embedded session ended without start")
+                return
+            }
 
-        if(impressions.isNotEmpty()) {
-            endAllImpressions()
+            if (impressions.isEmpty()) {
+                return
+            }
 
-            val sessionToTrack = IterableEmbeddedSession(
+            endAllImpressionsLocked()
+
+            val tracked = IterableEmbeddedSession(
                 session.start,
                 Date(),
-                getImpressionList()
+                getImpressionListLocked()
             )
 
-            IterableApi.getInstance().trackEmbeddedSession(sessionToTrack)
-
             //reset session for next session start
             session = IterableEmbeddedSession(
                 null,
@@ -56,43 +66,54 @@ public class EmbeddedSessionManager {
             )
 
             impressions = mutableMapOf()
+
+            tracked
         }
+
+        // Tracking calls into IterableApi, so it runs after the lock is released.
+        IterableApi.getInstance().trackEmbeddedSession(sessionToTrack)
     }
 
     fun startImpression(messageId: String, placementId: Long) {
-        var impressionData: EmbeddedImpressionData? = impressions[messageId]
+        synchronized(lock) {
+            var impressionData: EmbeddedImpressionData? = impressions[messageId]
 
-        if (impressionData == null) {
-            impressionData = EmbeddedImpressionData(messageId, placementId)
-            impressions[messageId] = impressionData
-        }
+            if (impressionData == null) {
+                impressionData = EmbeddedImpressionData(messageId, placementId)
+                impressions[messageId] = impressionData
+            }
 
-        impressionData.start = Date()
+            impressionData.start = Date()
+        }
     }
 
     fun pauseImpression(messageId: String) {
-        val impressionData: EmbeddedImpressionData? = impressions[messageId]
+        synchronized(lock) {
+            val impressionData: EmbeddedImpressionData? = impressions[messageId]
 
-        if (impressionData == null) {
-            IterableLogger.e(TAG, "onMessageImpressionEnded: impressionData not found")
-            return
-        }
+            if (impressionData == null) {
+                IterableLogger.e(TAG, "onMessageImpressionEnded: impressionData not found")
+                return
+            }
 
-        if (impressionData.start == null) {
-            IterableLogger.e(TAG, "onMessageImpressionEnded: impressionStarted is null")
-            return
-        }
+            if (impressionData.start == null) {
+                IterableLogger.e(TAG, "onMessageImpressionEnded: impressionStarted is null")
+                return
+            }
 
-        updateDisplayCountAndDuration(impressionData)
+            updateDisplayCountAndDurationLocked(impressionData)
+        }
     }
 
-    private fun endAllImpressions() {
+    // The Locked suffix marks helpers that read or write impressions without taking the lock
+    // themselves: every caller must already hold it.
+    private fun endAllImpressionsLocked() {
         for (impressionData in impressions.values) {
-            updateDisplayCountAndDuration(impressionData)
+            updateDisplayCountAndDurationLocked(impressionData)
         }
     }
 
-    private fun getImpressionList(): List<IterableEmbeddedImpression>? {
+    private fun getImpressionListLocked(): List<IterableEmbeddedImpression>? {
         val impressionList: MutableList<IterableEmbeddedImpression> = ArrayList()
         for (impressionData in impressions.values) {
             impressionList.add(
@@ -107,14 +128,15 @@ public class EmbeddedSessionManager {
         return impressionList
     }
 
-    private fun updateDisplayCountAndDuration(impressionData: EmbeddedImpressionData): EmbeddedImpressionData {
-        if (impressionData.start != null) {
+    private fun updateDisplayCountAndDurationLocked(impressionData: EmbeddedImpressionData): EmbeddedImpressionData {
+        val start = impressionData.start
+        if (start != null) {
             impressionData.displayCount = impressionData.displayCount.plus(1)
             impressionData.duration =
-                impressionData.duration.plus((Date().time - impressionData.start!!.time) / 1000.0)
+                impressionData.duration.plus((Date().time - start.time) / 1000.0)
                     .toFloat()
             impressionData.start = null
         }
         return impressionData
     }
-}
+}

iterableapi/src/test/java/com/iterable/iterableapi/EmbeddedSessionManagerThreadSafetyTest.java

@@ -0,0 +1,179 @@
+package com.iterable.iterableapi;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class EmbeddedSessionManagerThreadSafetyTest extends BaseTest {
+
+    private EmbeddedSessionManager sessionManager;
+
+    @Before
+    public void setUp() {
+        IterableApi.sharedInstance = new IterableApi();
+        sessionManager = new EmbeddedSessionManager();
+    }
+
+    // Pins existing behavior, not intended behavior: with no impressions, endSession() returns
+    // early and leaves the session open. Flip this assertion when SDK-701 is fixed.
+    @Test
+    public void endSessionWithoutImpressionsLeavesSessionRunning() {
+        sessionManager.startSession();
+        sessionManager.endSession();
+
+        assertTrue(sessionManager.isTracking());
+    }
+
+    @Test
+    public void endSessionWithImpressionsResetsSession() {
+        sessionManager.startSession();
+        sessionManager.startImpression("message-1", 1L);
+        sessionManager.pauseImpression("message-1");
+        sessionManager.endSession();
+
+        assertFalse(sessionManager.isTracking());
+    }
+
+    @Test
+    public void concurrentEndSessionTracksSessionOnlyOnce() throws Exception {
+        BlockingRecordingIterableApi recordingApi = new BlockingRecordingIterableApi();
+        IterableApi.sharedInstance = recordingApi;
+
+        sessionManager.startSession();
+        sessionManager.startImpression("message-1", 1L);
+        sessionManager.pauseImpression("message-1");
+
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        try {
+            // Keep the first tracking call open while a second thread ends the same session.
+            Future<?> firstEnd = executor.submit(sessionManager::endSession);
+            assertTrue(
+                    "first endSession did not reach tracking",
+                    recordingApi.awaitFirstTrack(5, TimeUnit.SECONDS)
+            );
+
+            // The session must already be cleared, even though its first tracking call is blocked.
+            // Ending it again must therefore return without tracking the same session twice.
+            Future<?> secondEnd = executor.submit(sessionManager::endSession);
+            secondEnd.get(5, TimeUnit.SECONDS);
+
+            recordingApi.allowFirstTrackToFinish();
+            firstEnd.get(5, TimeUnit.SECONDS);
+        } finally {
+            recordingApi.allowFirstTrackToFinish();
+            executor.shutdownNow();
+        }
+
+        List<IterableEmbeddedSession> trackedSessions = recordingApi.getTrackedSessions();
+        assertEquals("the active session should be tracked exactly once", 1, trackedSessions.size());
+
+        List<IterableEmbeddedImpression> impressions = trackedSessions.get(0).getImpressions();
+        assertNotNull(impressions);
+        assertEquals(1, impressions.size());
+        assertEquals("message-1", impressions.get(0).getMessageId());
+        assertEquals(1, impressions.get(0).getDisplayCount());
+    }
+
+    @Test
+    public void concurrentSessionAndImpressionUpdatesDoNotThrow() throws Exception {
+        final int threadCount = 8;
+        final int iterations = 2000;
+        final CountDownLatch startGate = new CountDownLatch(1);
+        final CountDownLatch finishGate = new CountDownLatch(threadCount);
+        final List<Throwable> failures = Collections.synchronizedList(new ArrayList<Throwable>());
+
+        sessionManager.startSession();
+
+        for (int threadIndex = 0; threadIndex < threadCount; threadIndex++) {
+            final int role = threadIndex % 4;
+            new Thread(new Runnable() {
+                @Override
+                public void run() {
+                    try {
+                        startGate.await();
+                        for (int i = 0; i < iterations; i++) {
+                            String messageId = "message-" + (i % 4);
+                            switch (role) {
+                                case 0:
+                                    sessionManager.startImpression(messageId, i % 3);
+                                    break;
+                                case 1:
+                                    sessionManager.pauseImpression(messageId);
+                                    break;
+                                case 2:
+                                    sessionManager.endSession();
+                                    break;
+                                default:
+                                    sessionManager.startSession();
+                                    break;
+                            }
+                        }
+                    } catch (Throwable throwable) {
+                        failures.add(throwable);
+                    } finally {
+                        finishGate.countDown();
+                    }
+                }
+            }, "embedded-session-" + threadIndex).start();
+        }
+
+        startGate.countDown();
+
+        assertTrue("threads did not finish in time", finishGate.await(30, TimeUnit.SECONDS));
+        assertEquals("concurrent access failed: " + failures, 0, failures.size());
+    }
+
+    private static class BlockingRecordingIterableApi extends IterableApi {
+        private final AtomicInteger trackCallCount = new AtomicInteger();
+        private final List<IterableEmbeddedSession> trackedSessions =
+                Collections.synchronizedList(new ArrayList<IterableEmbeddedSession>());
+        private final CountDownLatch firstTrackStarted = new CountDownLatch(1);
+        private final CountDownLatch allowFirstTrackToFinish = new CountDownLatch(1);
+
+        @Override
+        public void trackEmbeddedSession(IterableEmbeddedSession session) {
+            int callNumber = trackCallCount.incrementAndGet();
+            trackedSessions.add(session);
+
+            if (callNumber == 1) {
+                firstTrackStarted.countDown();
+                try {
+                    if (!allowFirstTrackToFinish.await(5, TimeUnit.SECONDS)) {
+                        throw new AssertionError("first tracking call was not released");
+                    }
+                } catch (InterruptedException exception) {
+                    Thread.currentThread().interrupt();
+                    throw new AssertionError("interrupted while waiting to finish tracking", exception);
+                }
+            }
+        }
+
+        boolean awaitFirstTrack(long timeout, TimeUnit unit) throws InterruptedException {
+            return firstTrackStarted.await(timeout, unit);
+        }
+
+        void allowFirstTrackToFinish() {
+            allowFirstTrackToFinish.countDown();
+        }
+
+        List<IterableEmbeddedSession> getTrackedSessions() {
+            synchronized (trackedSessions) {
+                return new ArrayList<>(trackedSessions);
+            }
+        }
+    }
+}