PANA-8616: Snapshot state and incremental diffing

Author: jonathanmos

detekt_custom_safe_calls_third_party.yml

@@ -478,6 +478,7 @@ datadog:
       - "kotlin.collections.Collection.fold(kotlin.collections.MutableSet, kotlin.Function2)"
       - "kotlin.collections.Collection.forEach(kotlin.Function1)"
       - "kotlin.collections.Collection.groupBy(kotlin.Function1)"
+      - "kotlin.collections.Collection.map(kotlin.Function1)"
       - "kotlin.collections.Collection.mapNotNull(kotlin.Function1)"
       - "kotlin.collections.Collection.isNotEmpty()"
       - "kotlin.collections.Collection.sumOf(kotlin.Function1)"

features/dd-sdk-android-session-replay/src/main/kotlin/com/datadog/android/sessionreplay/internal/DefaultRecorderProvider.kt

@@ -41,6 +41,7 @@ import com.datadog.android.sessionreplay.internal.composition.CompositionCapture
 import com.datadog.android.sessionreplay.internal.composition.CompositionChangeListener
 import com.datadog.android.sessionreplay.internal.composition.CompositionChangeset
 import com.datadog.android.sessionreplay.internal.composition.CompositionViewOnDrawInterceptor
+import com.datadog.android.sessionreplay.internal.composition.DefaultOrientationProvider
 import com.datadog.android.sessionreplay.internal.composition.DefaultRumViewScopeProvider
 import com.datadog.android.sessionreplay.internal.composition.DefaultSnapshotCompletionProcessor
 import com.datadog.android.sessionreplay.internal.composition.HandlerCaptureMainThreadExecutor
@@ -235,6 +236,8 @@ internal class DefaultRecorderProvider(
             rumContextProvider = rumContextProvider,
             recordWriter = recordWriter,
             internalLogger = internalLogger,
+            timeProvider = sdkCore.timeProvider,
+            orientationProvider = DefaultOrientationProvider(),
             resourceDataQueueHandler = resourceDataQueueHandler
         ),
         internalLogger = internalLogger

features/dd-sdk-android-session-replay/src/main/kotlin/com/datadog/android/sessionreplay/internal/composition/AndroidCapturedSnapshotProducer.kt

@@ -13,17 +13,20 @@ import com.datadog.android.internal.sessionreplay.composition.CapturedChild
 import com.datadog.android.internal.sessionreplay.composition.CapturedLayer
 import com.datadog.android.internal.sessionreplay.composition.CapturedLayerKind
 import com.datadog.android.internal.sessionreplay.composition.CapturedWireframe
+import com.datadog.android.internal.sessionreplay.composition.RumViewIdentityScope
 import com.datadog.android.internal.time.TimeProvider
 import com.datadog.android.sessionreplay.utils.DefaultViewIdentifierResolver
 import com.datadog.android.sessionreplay.utils.ViewIdentifierResolver
 
 /**
  * The real [CapturedSnapshotProducer] for the plain Android View hierarchy - the workstream-3
  * implementation of the extension point [SnapshotCaptureOrchestrator] drives every generation.
- * Builds a fresh [CapturedIdentityFactory] per call (this workstream only produces full snapshots;
- * an identity factory persisting across generations for incremental diffing is a later workstream's
- * concern), walks every currently active window via [AndroidWindowTraversal], and assembles them
- * under one synthetic screen root in [ActiveWindowSource.currentWindows] order (already z-ordered).
+ * Retains one [CapturedIdentityFactory] per active [com.datadog.android.internal.sessionreplay.composition.RumViewIdentityScope],
+ * reusing it across generations so the same View keeps the same identity - required for
+ * [CapturedSnapshotDiffer] to mean anything - and minting a fresh one, implicitly starting a new
+ * identity scope, only when the RUM view changes. Walks every currently active window via
+ * [AndroidWindowTraversal], and assembles them under one synthetic screen root in
+ * [ActiveWindowSource.currentWindows] order (already z-ordered).
  */
 internal class AndroidCapturedSnapshotProducer(
     private val windowSource: ActiveWindowSource,
@@ -33,11 +36,13 @@ internal class AndroidCapturedSnapshotProducer(
     private val viewIdentifierResolver: ViewIdentifierResolver = DefaultViewIdentifierResolver
 ) : CapturedSnapshotProducer {
 
+    private var retainedIdentityFactory: DefaultCapturedIdentityFactory? = null
+
     @MainThread
     @Suppress("ReturnCount")
     override fun capture(context: CaptureGenerationContext, changeset: CaptureChangeset): CaptureOutput? {
         val rumViewScope = scopeProvider.currentScope() ?: return null
-        val identityFactory = DefaultCapturedIdentityFactory(rumViewScope.scope)
+        val identityFactory = identityFactoryFor(rumViewScope.scope)
         val walk = walkWindows(windowSource.currentWindows(), identityFactory, context)
 
         return walk?.let {
@@ -61,6 +66,12 @@ internal class AndroidCapturedSnapshotProducer(
         }
     }
 
+    /** Reuses the retained factory while the RUM view scope is unchanged; mints a fresh one otherwise. */
+    private fun identityFactoryFor(scope: RumViewIdentityScope): DefaultCapturedIdentityFactory {
+        retainedIdentityFactory?.takeIf { it.scope == scope }?.let { return it }
+        return DefaultCapturedIdentityFactory(scope).also { retainedIdentityFactory = it }
+    }
+
     private fun walkWindows(
         windows: List<View>,
         identityFactory: CapturedIdentityFactory,

features/dd-sdk-android-session-replay/src/main/kotlin/com/datadog/android/sessionreplay/internal/composition/CapturedSnapshotDiffer.kt

@@ -0,0 +1,92 @@
+/*
+ * Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+ * This product includes software developed at Datadog (https://www.datadoghq.com/).
+ * Copyright 2016-Present Datadog, Inc.
+ */
+
+package com.datadog.android.sessionreplay.internal.composition
+
+import com.datadog.android.internal.sessionreplay.composition.CapturedLayer
+
+/**
+ * Diffs two accepted [CapturedFullSnapshot]s of the same [com.datadog.android.internal.sessionreplay.composition.RumViewIdentityScope]
+ * into a [CapturedMutationSet], or returns null when the change can't be expressed as a mutation
+ * under the current wire contract - the caller must fall back to a full snapshot in that case.
+ *
+ * The wire mutation model (block 1) can only express layer-level structural changes (a layer's own
+ * bounds/children/modifiers/composite operation) - it has no operation for a wireframe's own
+ * content changing (text, its independently-absolute bounds, an image resource, a style), nor for
+ * delivering a brand-new wireframe's definition out of band. [diff] therefore requires every
+ * wireframe referenced by [current] to already have existed, unchanged, in [previous] before it
+ * will attempt a layer diff at all; a wireframe simply disappearing needs no such check, since it
+ * becomes unreachable exactly when its owning layer is removed or stops referencing it, which the
+ * layer diff below already produces correctly by construction.
+ */
+internal object CapturedSnapshotDiffer {
+
+    @Suppress("ReturnCount")
+    fun diff(previous: CapturedFullSnapshot, current: CapturedFullSnapshot): CapturedMutationSet? {
+        if (previous.scope != current.scope) return null
+        val currentRoot = current.root ?: return null
+        if (!isWireframeContentStable(previous, current)) return null
+
+        val previousLayers = previous.layers.associateBy { it.identity.wireId }
+        val currentLayers = current.layers.associateBy { it.identity.wireId }
+
+        val adds = currentLayers.filterKeys { it !in previousLayers }.values.toList()
+        val removes = previousLayers.filterKeys { it !in currentLayers }.values.map { it.identity }
+        val updates = currentLayers.mapNotNull { (wireId, layer) ->
+            previousLayers[wireId]?.let { diffLayer(it, layer) }
+        }
+
+        return CapturedMutationSet(
+            timestamp = current.timestamp,
+            scope = current.scope,
+            root = if (previous.root != currentRoot) CapturedChange.Set(currentRoot) else CapturedChange.Unchanged,
+            adds = adds.toChange(),
+            removes = removes.toChange(),
+            updates = updates.toChange()
+        )
+    }
+
+    /**
+     * True only if every wireframe [current] references already existed, with identical content,
+     * in [previous]. A brand-new wireframe id has no delivery mechanism on the wire today, and a
+     * persisting wireframe with different content has no update mechanism either - both force a
+     * full snapshot. Wireframes present only in [previous] are deliberately not examined here.
+     */
+    private fun isWireframeContentStable(previous: CapturedFullSnapshot, current: CapturedFullSnapshot): Boolean {
+        val previousWireframesById = previous.wireframes.associateBy { it.identity.wireId }
+        return current.wireframes.all { wireframe ->
+            previousWireframesById[wireframe.identity.wireId] == wireframe
+        }
+    }
+
+    /** Null if nothing about [current] differs from [previous]; otherwise a sparse per-field update. */
+    private fun diffLayer(previous: CapturedLayer, current: CapturedLayer): CapturedLayerUpdate? {
+        val update = CapturedLayerUpdate(
+            identity = current.identity,
+            x = changeIfDiffers(previous.bounds.x, current.bounds.x),
+            y = changeIfDiffers(previous.bounds.y, current.bounds.y),
+            width = changeIfDiffers(previous.bounds.width, current.bounds.width),
+            height = changeIfDiffers(previous.bounds.height, current.bounds.height),
+            children = changeIfDiffers(previous.children, current.children),
+            modifiers = changeIfDiffers(previous.modifiers, current.modifiers),
+            compositeOperation = changeIfDiffers(previous.compositeOperation, current.compositeOperation)
+        )
+        val isUnchanged = update.x == CapturedChange.Unchanged &&
+            update.y == CapturedChange.Unchanged &&
+            update.width == CapturedChange.Unchanged &&
+            update.height == CapturedChange.Unchanged &&
+            update.children == CapturedChange.Unchanged &&
+            update.modifiers == CapturedChange.Unchanged &&
+            update.compositeOperation == CapturedChange.Unchanged
+        return update.takeUnless { isUnchanged }
+    }
+
+    private fun <T> changeIfDiffers(previous: T, current: T): CapturedChange<T> =
+        if (previous == current) CapturedChange.Unchanged else CapturedChange.Set(current)
+
+    private fun <T> List<T>.toChange(): CapturedChange<List<T>> =
+        takeIf { it.isNotEmpty() }?.let { CapturedChange.Set(it) } ?: CapturedChange.Unchanged
+}

features/dd-sdk-android-session-replay/src/main/kotlin/com/datadog/android/sessionreplay/internal/composition/OrientationProvider.kt

@@ -0,0 +1,32 @@
+/*
+ * Unless explicitly stated otherwise all files in this repository are licensed under the Apache License Version 2.0.
+ * This product includes software developed at Datadog (https://www.datadoghq.com/).
+ * Copyright 2016-Present Datadog, Inc.
+ */
+
+package com.datadog.android.sessionreplay.internal.composition
+
+import android.content.res.Configuration
+import android.content.res.Resources
+
+/** Isolates the current device orientation so it can be substituted in tests. */
+internal fun interface OrientationProvider {
+    fun currentOrientation(): Int
+}
+
+/**
+ * Reads [Resources.getSystem] rather than requiring an Activity/Application context - it reflects
+ * true device rotation without needing any context threaded into the composition pipeline. Falls
+ * back to a constant sentinel on failure rather than throwing: a session-replay signal must never
+ * be able to crash the host app, and a constant value simply makes orientation-change gating inert
+ * (the other full-snapshot triggers - new view, periodic checkpoint - are unaffected) rather than
+ * unsafe in either direction.
+ */
+internal class DefaultOrientationProvider : OrientationProvider {
+    @Suppress("TooGenericExceptionCaught", "UnsafeThirdPartyFunctionCall")
+    override fun currentOrientation(): Int = try {
+        Resources.getSystem().configuration.orientation
+    } catch (@Suppress("SwallowedException") e: Exception) {
+        Configuration.ORIENTATION_UNDEFINED
+    }
+}

features/dd-sdk-android-session-replay/src/main/kotlin/com/datadog/android/sessionreplay/internal/composition/SnapshotCompletionQueue.kt

@@ -8,6 +8,7 @@ package com.datadog.android.sessionreplay.internal.composition
 
 import com.datadog.android.api.InternalLogger
 import com.datadog.android.core.internal.utils.executeSafe
+import com.datadog.android.internal.time.TimeProvider
 import com.datadog.android.sessionreplay.internal.async.DataQueueHandler
 import com.datadog.android.sessionreplay.internal.processor.EnrichedRecord
 import com.datadog.android.sessionreplay.internal.storage.RecordWriter
@@ -16,12 +17,20 @@ import com.datadog.android.sessionreplay.internal.utils.SessionReplayRumContext
 import com.datadog.android.sessionreplay.model.MobileSegment
 import java.util.concurrent.ConcurrentLinkedQueue
 import java.util.concurrent.ExecutorService
+import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
 
 internal fun interface SnapshotCompletionProcessor {
     fun process(capture: CompletedSnapshotCapture)
 }
 
+/** The last accepted snapshot a generation can be diffed against, and when it was last a full one. */
+private data class RetainedSnapshotState(
+    val snapshot: CapturedFullSnapshot,
+    val orientation: Int,
+    val lastFullSnapshotAtNs: Long
+)
+
 /**
  * Bundles composition wire-mapping with the view-lifecycle records the player needs around it -
  * mirrors legacy `RecordedDataProcessor`'s new-view handling: every genuinely new RUM view opens
@@ -40,18 +49,29 @@ internal fun interface SnapshotCompletionProcessor {
  * calls [DataQueueHandler.tryToConsumeItems] - without a caller, those items sat in memory
  * forever, so every composition-tree pixel capture's `resourceId` referenced a resource that had
  * never actually been persisted anywhere.
+ *
+ * Also diffs each completed generation against the last *accepted* one (retained only inside the
+ * [CaptureGenerationContext.tryAccept]-gated success branch below, so an expired/rejected
+ * generation never corrupts it) and emits an incremental mutation via [CapturedSnapshotDiffer]
+ * where possible, falling back to a full snapshot on a new RUM view, a periodic checkpoint, an
+ * orientation change, or a mutation that unexpectedly fails validation (self-healing rather than
+ * dropping the generation) - mirroring legacy `RecordedDataProcessor`'s
+ * `isNewView`/`isTimeForFullSnapshot`/`screenOrientationChanged` gating for this pipeline's state.
  */
 internal class DefaultSnapshotCompletionProcessor(
     private val rumContextProvider: RumContextProvider,
     private val recordWriter: RecordWriter,
     private val internalLogger: InternalLogger,
+    private val timeProvider: TimeProvider,
+    private val orientationProvider: OrientationProvider = DefaultOrientationProvider(),
     private val wireMapper: CapturedTreeWireMapper = DefaultCapturedTreeWireMapper(),
     private val resourceDataQueueHandler: DataQueueHandler? = null
 ) : SnapshotCompletionProcessor {
 
     // Only ever read/written from SnapshotCompletionQueue's single draining thread - same
     // single-threaded-processor assumption legacy RecordedDataProcessor's own prevRumContext relies on.
     private var lastViewContext: SessionReplayRumContext? = null
+    private var retained: RetainedSnapshotState? = null
 
     override fun process(capture: CompletedSnapshotCapture) {
         // Whatever resources this generation's pixel captures resolved were already queued
@@ -66,7 +86,8 @@ internal class DefaultSnapshotCompletionProcessor(
             return
         }
 
-        when (val mapping = wireMapper.mapFullSnapshot(capture.snapshot)) {
+        val currentOrientation = orientationProvider.currentOrientation()
+        when (val mapping = resolveMapping(capture.snapshot, currentOrientation)) {
             is CaptureWireMappingResult.Success -> {
                 if (capture.generation.tryAccept()) {
                     writeViewEndRecordIfViewChanged(rumContext, capture.snapshot.timestamp)
@@ -84,6 +105,11 @@ internal class DefaultSnapshotCompletionProcessor(
                             records = records
                         )
                     )
+                    retained = RetainedSnapshotState(
+                        snapshot = capture.snapshot,
+                        orientation = currentOrientation,
+                        lastFullSnapshotAtNs = lastFullSnapshotAtNs(mapping.value)
+                    )
                 }
             }
 
@@ -125,6 +151,48 @@ internal class DefaultSnapshotCompletionProcessor(
             )
         )
     }
+
+    /** Only a full snapshot resets the periodic-checkpoint clock; a mutation cycle leaves it running. */
+    private fun lastFullSnapshotAtNs(record: MobileSegment.MobileRecord): Long =
+        if (record is MobileSegment.MobileRecord.MobileFullSnapshotRecord) {
+            timeProvider.getDeviceElapsedTimeNanos()
+        } else {
+            retained?.lastFullSnapshotAtNs ?: timeProvider.getDeviceElapsedTimeNanos()
+        }
+
+    @Suppress("ReturnCount")
+    private fun resolveMapping(
+        snapshot: CapturedFullSnapshot,
+        currentOrientation: Int
+    ): CaptureWireMappingResult<MobileSegment.MobileRecord> {
+        val retainedState = retained ?: return wireMapper.mapFullSnapshot(snapshot)
+        val fullSnapshotRequired = retainedState.snapshot.scope != snapshot.scope ||
+            currentOrientation != retainedState.orientation ||
+            isTimeForFullSnapshot(retainedState)
+        if (fullSnapshotRequired) return wireMapper.mapFullSnapshot(snapshot)
+
+        val mutation = CapturedSnapshotDiffer.diff(retainedState.snapshot, snapshot)
+            ?: return wireMapper.mapFullSnapshot(snapshot)
+
+        return when (val mutationMapping = wireMapper.mapMutation(mutation, retainedState.snapshot)) {
+            is CaptureWireMappingResult.Success -> mutationMapping
+            is CaptureWireMappingResult.Invalid -> {
+                internalLogger.log(
+                    InternalLogger.Level.WARN,
+                    InternalLogger.Target.TELEMETRY,
+                    { "Computed mutation failed validation, retrying as a full snapshot: ${mutationMapping.failures}" }
+                )
+                wireMapper.mapFullSnapshot(snapshot)
+            }
+        }
+    }
+
+    private fun isTimeForFullSnapshot(retainedState: RetainedSnapshotState): Boolean =
+        timeProvider.getDeviceElapsedTimeNanos() - retainedState.lastFullSnapshotAtNs >= FULL_SNAPSHOT_INTERVAL_NS
+
+    private companion object {
+        val FULL_SNAPSHOT_INTERVAL_NS = TimeUnit.MILLISECONDS.toNanos(3000)
+    }
 }
 
 /**