diff --git a/.changeset/replay-capture-off-main-thread.md b/.changeset/replay-capture-off-main-thread.md new file mode 100644 index 0000000..9e13061 --- /dev/null +++ b/.changeset/replay-capture-off-main-thread.md @@ -0,0 +1,5 @@ +--- +'com.posthog.unity': patch +--- + +Reduce session replay frame hitches by encoding screenshots to JPEG and base64 on a worker thread, and by serializing and gzipping replay batches on a worker thread instead of the main thread. diff --git a/com.posthog.unity/Runtime/SessionReplay/ReplayQueue.cs b/com.posthog.unity/Runtime/SessionReplay/ReplayQueue.cs index cb9868f..85af919 100644 --- a/com.posthog.unity/Runtime/SessionReplay/ReplayQueue.cs +++ b/com.posthog.unity/Runtime/SessionReplay/ReplayQueue.cs @@ -4,6 +4,7 @@ using System.IO; using System.IO.Compression; using System.Text; +using System.Threading.Tasks; using UnityEngine; using UnityEngine.Networking; @@ -20,7 +21,7 @@ class ReplayQueue readonly string _host; readonly Func _getDistinctId; readonly Func _getSessionId; - readonly Func, Action, IEnumerator> _sendBatch; + readonly Func, bool, Action, IEnumerator> _sendBatch; readonly Func _utcNow; readonly Func _isConnected; readonly object _lock = new(); @@ -62,7 +63,7 @@ internal ReplayQueue( string host, Func getDistinctId, Func getSessionId, - Func, Action, IEnumerator> sendBatch, + Func, bool, Action, IEnumerator> sendBatch, Func utcNow, Func isConnected ) @@ -156,7 +157,9 @@ public int Count /// /// Flushes all events in the queue. /// - public void Flush() + /// Serialize and compress the first batch on the main thread, so its request + /// starts before the coroutine first yields. Use it when the app is about to be suspended. + public void Flush(bool prepareInline = false) { if (!_isRunning) { @@ -166,7 +169,7 @@ public void Flush() if (_coroutineRunner != null) { - _coroutineRunner.StartCoroutine(FlushCoroutine()); + _coroutineRunner.StartCoroutine(FlushCoroutine(prepareInline)); } } @@ -226,10 +229,14 @@ IEnumerator FlushTimerCoroutine() } } - internal IEnumerator FlushCoroutine() + internal IEnumerator FlushCoroutine(bool prepareInline = false) { lock (_lock) { + // A flush already in flight makes an inline flush skip too. If it is still preparing on the + // worker when the app is suspended, its batch waits until the app resumes. That window is + // the prepare time (~100 ms) of each flush, so it is accepted rather than finished + // synchronously. A flush waiting on its response skips this one the same way main did. if (_isFlushing) { PostHogLogger.Debug("Already flushing replay queue, skipping"); @@ -276,6 +283,7 @@ internal IEnumerator FlushCoroutine() var statusCode = 0; yield return _sendBatch( batch, + prepareInline, (result, code) => { success = result; @@ -288,6 +296,9 @@ internal IEnumerator FlushCoroutine() RemoveBatch(batch); ResetRetryState(); PostHogLogger.Debug($"Successfully sent {batch.Count} replay events"); + // Only the first batch has to start before a suspend, and later batches would block + // the main thread again. + prepareInline = false; continue; } @@ -356,7 +367,11 @@ void ResetRetryState() _pausedUntil = null; } - IEnumerator SendBatch(List events, Action onComplete) + IEnumerator SendBatch( + List events, + bool prepareInline, + Action onComplete + ) { var url = $"{_host}/s/"; @@ -366,10 +381,42 @@ IEnumerator SendBatch(List events, Action onComplete) batchList.Add(evt.ToDictionary(_apiKey)); } - var json = JsonSerializer.Serialize(batchList); - var bodyBytes = Encoding.UTF8.GetBytes(json); + Task<(byte[] PayloadBytes, bool UseCompression, Exception CompressionError)> prepare; + if (prepareInline) + { + // The app is about to be suspended, and Unity's loop stops with it. Preparing here lets the + // request start before the first yield, so the OS sends it before suspending the app. + prepare = PrepareBodyInline(batchList); + } + else + { + // Serializing and compressing a batch of screenshots takes over 100 ms on mobile, so it runs + // on a worker thread while this coroutine waits. None of it touches Unity APIs. + prepare = Task.Run(() => PrepareBody(batchList)); + while (!prepare.IsCompleted) + { + yield return null; + } + } - var (payloadBytes, useCompression) = PreparePayload(bodyBytes, CompressGzip); + if (prepare.IsFaulted || prepare.IsCanceled) + { + PostHogLogger.Error( + "Failed to serialize replay batch", + prepare.Exception?.GetBaseException() + ); + // The same batch would fail again, so drop it like a rejected request instead of retrying it. + onComplete?.Invoke(false, 400); + yield break; + } + + var (payloadBytes, useCompression, compressionError) = prepare.Result; + if (compressionError != null) + { + PostHogLogger.Warning( + $"Failed to gzip replay batch, sending uncompressed: {compressionError.Message}" + ); + } PostHogLogger.Debug( $"Sending replay batch to {url} (size: {payloadBytes.Length} bytes)" ); @@ -405,25 +452,57 @@ IEnumerator SendBatch(List events, Action onComplete) } } - internal static (byte[] PayloadBytes, bool UseCompression) PreparePayload( - byte[] bodyBytes, - Func compressGzip - ) + /// + /// Runs on the calling thread, returning a failure as a faulted task so the + /// caller handles both paths the same way. + /// + internal static Task<( + byte[] PayloadBytes, + bool UseCompression, + Exception CompressionError + )> PrepareBodyInline(List> batchList) { try { - return (compressGzip(bodyBytes), true); + return Task.FromResult(PrepareBody(batchList)); } catch (Exception ex) { - PostHogLogger.Warning( - $"Failed to gzip replay batch, sending uncompressed: {ex.Message}" - ); - return (bodyBytes, false); + return Task.FromException<(byte[], bool, Exception)>(ex); + } + } + + /// + /// Runs on a worker thread, so it doesn't log; the caller reports a compression error. + /// + internal static ( + byte[] PayloadBytes, + bool UseCompression, + Exception CompressionError + ) PrepareBody(List> batchList) + { + var json = JsonSerializer.Serialize(batchList); + var bodyBytes = Encoding.UTF8.GetBytes(json); + return PreparePayload(bodyBytes, CompressGzip); + } + + internal static ( + byte[] PayloadBytes, + bool UseCompression, + Exception CompressionError + ) PreparePayload(byte[] bodyBytes, Func compressGzip) + { + try + { + return (compressGzip(bodyBytes), true, null); + } + catch (Exception ex) + { + return (bodyBytes, false, ex); } } - byte[] CompressGzip(byte[] data) + static byte[] CompressGzip(byte[] data) { using var output = new MemoryStream(); using (var gzip = new GZipStream(output, CompressionMode.Compress)) diff --git a/com.posthog.unity/Runtime/SessionReplay/ScreenshotCapture.cs b/com.posthog.unity/Runtime/SessionReplay/ScreenshotCapture.cs index 117dd2b..0d9b5f0 100644 --- a/com.posthog.unity/Runtime/SessionReplay/ScreenshotCapture.cs +++ b/com.posthog.unity/Runtime/SessionReplay/ScreenshotCapture.cs @@ -1,12 +1,15 @@ using System; +using System.Threading.Tasks; using UnityEngine; +using UnityEngine.Experimental.Rendering; using UnityEngine.Rendering; namespace PostHogUnity.SessionReplay { /// /// Captures and encodes screenshots for session replay with frame sampling. - /// Uses AsyncGPUReadback to avoid blocking the main thread during capture. + /// The GPU scales the screenshot, AsyncGPUReadback brings it back without blocking, and the JPEG and + /// base64 encoding run on a worker thread. hands the result back on the main thread. /// class ScreenshotCapture { @@ -15,8 +18,14 @@ class ScreenshotCapture readonly Vector2 _blitOffset; float _lastCaptureTime; + + // True from the capture request until Poll hands the encoded result back, so the next capture + // can't overwrite _pixels while the worker is still reading it. bool _isCapturing; + // Incremented by CancelPending so a readback requested before it is ignored. + int _generation; + // Reusable render textures to avoid allocations RenderTexture _fullRT; RenderTexture _scaledRT; @@ -25,6 +34,11 @@ class ScreenshotCapture int _lastScaledWidth; int _lastScaledHeight; + // Readback pixels, reused between captures. + byte[] _pixels; + Task _encodeTask; + Action _onComplete; + public ScreenshotCapture(PostHogSessionReplayConfig config) { _config = config; @@ -51,7 +65,8 @@ public bool CanCapture() /// Captures a screenshot asynchronously using AsyncGPUReadback. /// Requires AsyncGPUReadback support (checked at session replay start). /// - /// Callback with the screenshot result, or null if capture failed. + /// Called on the main thread with the screenshot result, or null if capture + /// failed. A successful result is delivered by . public void CaptureScreenshotAsync(Action onComplete) { if (!CanCapture()) @@ -75,6 +90,7 @@ public void CaptureScreenshotAsync(Action onComplete) long timestamp = GetTimestampMs(); int quality = _config.ScreenshotQuality; + int generation = _generation; EnsureRenderTextures(screenWidth, screenHeight, scaledWidth, scaledHeight); ScreenCapture.CaptureScreenshotIntoRenderTexture(_fullRT); @@ -91,6 +107,7 @@ public void CaptureScreenshotAsync(Action onComplete) { OnReadbackComplete( request, + generation, scaledWidth, scaledHeight, screenWidth, @@ -110,6 +127,34 @@ public void CaptureScreenshotAsync(Action onComplete) } } + /// + /// Hands a finished encode to its callback on the main thread. Call it every frame; it does nothing + /// while no encode is pending or the worker is still running. + /// + public void Poll() + { + if (_encodeTask == null || !_encodeTask.IsCompleted) + return; + + var task = _encodeTask; + var onComplete = _onComplete; + _encodeTask = null; + _onComplete = null; + _isCapturing = false; + + if (task.IsFaulted || task.IsCanceled) + { + PostHogLogger.Error( + "Failed to encode screenshot", + task.Exception?.GetBaseException() + ); + onComplete?.Invoke(null); + return; + } + + onComplete?.Invoke(task.Result); + } + void EnsureRenderTextures( int screenWidth, int screenHeight, @@ -167,6 +212,7 @@ int scaledHeight void OnReadbackComplete( AsyncGPUReadbackRequest request, + int generation, int scaledWidth, int scaledHeight, int screenWidth, @@ -176,10 +222,12 @@ void OnReadbackComplete( Action onComplete ) { - _isCapturing = false; + if (generation != _generation) + return; if (request.hasError) { + _isCapturing = false; PostHogLogger.Error("AsyncGPUReadback error during screenshot capture"); onComplete?.Invoke(null); return; @@ -187,37 +235,74 @@ Action onComplete try { + // The readback data is only valid during this callback, so copy it for the worker. var data = request.GetData(); - - var texture = new Texture2D(scaledWidth, scaledHeight, TextureFormat.RGBA32, false); - texture.LoadRawTextureData(data); - texture.Apply(); - - byte[] jpegBytes = texture.EncodeToJPG(quality); - UnityEngine.Object.Destroy(texture); - - string base64 = Convert.ToBase64String(jpegBytes); - string dataUrl = $"data:image/jpeg;base64,{base64}"; + if (_pixels == null || _pixels.Length != data.Length) + _pixels = new byte[data.Length]; + data.CopyTo(_pixels); var result = new ScreenshotResult { - Base64Data = dataUrl, Width = scaledWidth, Height = scaledHeight, OriginalWidth = screenWidth, OriginalHeight = screenHeight, Timestamp = timestamp, }; - - onComplete?.Invoke(result); + var pixels = _pixels; + _encodeTask = Task.Run(() => + { + result.Base64Data = Encode(pixels, result.Width, result.Height, quality); + return result; + }); + _onComplete = onComplete; } catch (Exception ex) { + _isCapturing = false; + _encodeTask = null; + _onComplete = null; PostHogLogger.Error("Failed to process screenshot readback", ex); onComplete?.Invoke(null); } } + /// + /// Runs on a worker thread, so it uses no Unity APIs except the thread-safe ImageConversion encoder. + /// + static string Encode(byte[] rgba, int width, int height, int quality) + { + // RGBA32 sRGB readback, matching the Texture2D(RGBA32, linear: false) previously encoded from. + byte[] jpegBytes = ImageConversion.EncodeArrayToJPG( + rgba, + GraphicsFormat.R8G8B8A8_SRGB, + (uint)width, + (uint)height, + 0, + quality + ); + return ToJpegDataUrl(jpegBytes); + } + + /// + /// Builds "data:image/jpeg;base64,..." in a single string instead of a base64 string plus a copy. + /// + internal static string ToJpegDataUrl(byte[] jpegBytes) + { + const string prefix = "data:image/jpeg;base64,"; + int base64Length = (jpegBytes.Length + 2) / 3 * 4; + return string.Create( + prefix.Length + base64Length, + jpegBytes, + (chars, bytes) => + { + prefix.AsSpan().CopyTo(chars); + if (!Convert.TryToBase64Chars(bytes, chars.Slice(prefix.Length), out _)) + throw new InvalidOperationException("Base64 buffer size mismatch"); + } + ); + } + /// /// Gets the current Unix timestamp in milliseconds. /// @@ -235,10 +320,25 @@ public void ResetThrottle() } /// - /// Releases render texture resources. + /// Drops a readback or encode still in flight, so its screenshot is never delivered. + /// + public void CancelPending() + { + _generation++; + _encodeTask = null; + _onComplete = null; + // The dropped worker may still be reading the old buffer. + _pixels = null; + _isCapturing = false; + } + + /// + /// Releases render texture resources. A readback or encode still in flight is dropped. /// public void Dispose() { + CancelPending(); + if (_fullRT != null) { _fullRT.Release(); diff --git a/com.posthog.unity/Runtime/SessionReplay/SessionReplayIntegration.cs b/com.posthog.unity/Runtime/SessionReplay/SessionReplayIntegration.cs index 96b5f7a..2f94d90 100644 --- a/com.posthog.unity/Runtime/SessionReplay/SessionReplayIntegration.cs +++ b/com.posthog.unity/Runtime/SessionReplay/SessionReplayIntegration.cs @@ -129,7 +129,8 @@ public void Pause() _isPaused = true; _consoleLogCapture?.Pause(); _networkTelemetry?.Pause(); - _replayQueue.Flush(); + // The app is going to the background, so the request has to start before Unity's loop stops. + _replayQueue.Flush(prepareInline: true); PostHogLogger.Debug("Session replay paused"); } @@ -248,6 +249,9 @@ IEnumerator CaptureLoopCoroutine() { yield return waitForEndOfFrame; + // Delivers a screenshot the worker thread finished encoding since the last frame. + _screenshotCapture.Poll(); + if (!IsActive) { yield return null; @@ -293,15 +297,24 @@ void CaptureAndEnqueueSnapshot() } var screenName = _currentScreenName; + var sessionId = _lastSessionId; _screenshotCapture.CaptureScreenshotAsync(result => { - OnScreenshotCaptured(result, touchEvents, networkSamples, logs, screenName); + OnScreenshotCaptured( + result, + sessionId, + touchEvents, + networkSamples, + logs, + screenName + ); }); } void OnScreenshotCaptured( ScreenshotResult result, + string sessionId, List touchEvents, List networkSamples, List logs, @@ -311,6 +324,14 @@ string screenName if (result == null) return; + // The session can change while the screenshot is read back and encoded (Reset, expiry), and the + // capture loop only notices on a later frame, or not at all while paused. + if (!ShouldEnqueueScreenshot(_isRunning, sessionId, _getSessionId())) + { + PostHogLogger.Debug("SessionReplay: Dropping screenshot from a previous session"); + return; + } + var events = new List(); events.Add( @@ -348,6 +369,19 @@ string screenName _replayQueue.Enqueue(events); } + /// + /// A screenshot is enqueued only under the session it was captured in, so a frame from before a + /// rotation never lands in the new session. + /// + internal static bool ShouldEnqueueScreenshot( + bool isRunning, + string capturedSessionId, + string currentSessionId + ) + { + return isRunning && capturedSessionId == currentSessionId; + } + void SendMetaEvent() { var timestamp = ScreenshotCapture.GetTimestampMs(); @@ -363,6 +397,8 @@ void OnSessionRotated() _lastSessionId = _getSessionId(); Clear(); + // A screenshot still being read back or encoded belongs to the previous session. + _screenshotCapture.CancelPending(); _screenshotCapture.ResetThrottle(); SendMetaEvent(); diff --git a/tests/PostHog.Unity.Tests/ReplayQueueTests.cs b/tests/PostHog.Unity.Tests/ReplayQueueTests.cs index 31fe81e..c3c8175 100644 --- a/tests/PostHog.Unity.Tests/ReplayQueueTests.cs +++ b/tests/PostHog.Unity.Tests/ReplayQueueTests.cs @@ -88,13 +88,119 @@ public void Clear_RemovesQueuedSnapshots() Assert.Equal(0, queue.Count); } + [Fact] + public void PrepareBodyGzipsSerializedBatch() + { + var batchList = new List> + { + new() { ["event"] = "$snapshot", ["text"] = "caf\u00e9 \"quoted\"" }, + new() { ["event"] = "$snapshot", ["n"] = 2 }, + }; + + var (payloadBytes, useCompression, compressionError) = ReplayQueue.PrepareBody( + batchList + ); + + Assert.Null(compressionError); + + Assert.True(useCompression); + using var input = new System.IO.MemoryStream(payloadBytes); + using var gzip = new System.IO.Compression.GZipStream( + input, + System.IO.Compression.CompressionMode.Decompress + ); + using var reader = new System.IO.StreamReader(gzip, System.Text.Encoding.UTF8); + Assert.Equal(JsonSerializer.Serialize(batchList), reader.ReadToEnd()); + } + + [Fact] + public async Task PrepareBodyInlineMatchesPrepareBody() + { + var batchList = new List> + { + new() { ["event"] = "$snapshot", ["n"] = 1 }, + }; + + var prepare = ReplayQueue.PrepareBodyInline(batchList); + + // Already completed: it ran on the calling thread. + Assert.True(prepare.IsCompletedSuccessfully); + var prepared = await prepare; + Assert.Equal(ReplayQueue.PrepareBody(batchList).PayloadBytes, prepared.PayloadBytes); + Assert.True(prepared.UseCompression); + } + + [Fact] + public void PrepareBodyInlineReturnsSerializationFailureAsFaultedTask() + { + var batchList = new List> + { + new() { ["event"] = "$snapshot", ["bad"] = new ThrowingToString() }, + }; + + var prepare = ReplayQueue.PrepareBodyInline(batchList); + + Assert.True(prepare.IsFaulted); + Assert.IsType(prepare.Exception.GetBaseException()); + } + + [Fact] + public void InlineFlushPreparesOnlyTheFirstBatchInline() + { + var harness = new ReplayHarness(); + // MaxBatchSize is 10, so this is two batches. + for (var i = 0; i < 12; i++) + harness.Enqueue($"event-{i}"); + harness.Results.Enqueue((true, 200)); + harness.Results.Enqueue((true, 200)); + + RunCoroutine(harness.Queue.FlushCoroutine(prepareInline: true)); + + Assert.Equal(new[] { 10, 2 }, harness.Attempts.Select(batch => batch.Count)); + Assert.Equal(new[] { true, false }, harness.PrepareInline); + Assert.Empty(GetQueuedEvents(harness.Queue)); + } + + [Fact] + public void DefaultFlushPreparesOnTheWorker() + { + var harness = new ReplayHarness(); + harness.Enqueue("event"); + harness.Results.Enqueue((true, 200)); + + RunCoroutine(harness.Queue.FlushCoroutine()); + + Assert.Equal(new[] { false }, harness.PrepareInline); + } + + [Fact] + public void InlineFlushRetriesWithTheWorkerAfterAFailure() + { + var harness = new ReplayHarness(); + harness.Enqueue("retained"); + harness.Results.Enqueue((false, 503)); + + RunCoroutine(harness.Queue.FlushCoroutine(prepareInline: true)); + harness.Now = harness.Now.AddSeconds(5); + harness.Results.Enqueue((true, 200)); + RunCoroutine(harness.Queue.FlushCoroutine()); + + Assert.Equal(new[] { true, false }, harness.PrepareInline); + Assert.Empty(GetQueuedEvents(harness.Queue)); + } + + sealed class ThrowingToString + { + public override string ToString() => throw new InvalidOperationException("bad value"); + } + [Fact] public void PreparePayloadCompressesSmallPayloads() { var bodyBytes = new byte[16]; var compressedBytes = new byte[] { 1, 2, 3 }; - var (payloadBytes, useCompression) = ReplayQueue.PreparePayload( + var (payloadBytes, useCompression, compressionError) = ReplayQueue.PreparePayload( bodyBytes, bytes => { @@ -105,6 +211,7 @@ public void PreparePayloadCompressesSmallPayloads() Assert.Same(compressedBytes, payloadBytes); Assert.True(useCompression); + Assert.Null(compressionError); } [Fact] @@ -112,13 +219,14 @@ public void PreparePayloadFallsBackToUncompressedWhenCompressionFails() { var bodyBytes = new byte[16]; - var (payloadBytes, useCompression) = ReplayQueue.PreparePayload( + var (payloadBytes, useCompression, compressionError) = ReplayQueue.PreparePayload( bodyBytes, _ => throw new InvalidOperationException("gzip failed") ); Assert.Same(bodyBytes, payloadBytes); Assert.False(useCompression); + Assert.Equal("gzip failed", compressionError.Message); } [Theory] @@ -367,15 +475,21 @@ public ReplayHarness(int maxQueueSize = 100) public DateTime Now { get; set; } = new(2026, 1, 1, 0, 0, 0, DateTimeKind.Utc); public bool Connected { get; set; } = true; public Action> OnSend { get; set; } + public List PrepareInline { get; } = new(); public void Enqueue(string screenName) { Queue.Enqueue(new List { RREvent.CreateMeta(100, 200, screenName, 123L) }); } - IEnumerator SendBatch(List batch, Action onComplete) + IEnumerator SendBatch( + List batch, + bool prepareInline, + Action onComplete + ) { Attempts.Add(new List(batch)); + PrepareInline.Add(prepareInline); OnSend?.Invoke(batch); var result = Results.Dequeue(); onComplete(result.Success, result.StatusCode); diff --git a/tests/PostHog.Unity.Tests/ScreenshotDataUrlTests.cs b/tests/PostHog.Unity.Tests/ScreenshotDataUrlTests.cs new file mode 100644 index 0000000..9ccbe3f --- /dev/null +++ b/tests/PostHog.Unity.Tests/ScreenshotDataUrlTests.cs @@ -0,0 +1,26 @@ +using PostHogUnity.SessionReplay; + +namespace PostHogUnity.Tests +{ + public class ScreenshotDataUrlTests + { + [Theory] + [InlineData(0)] + [InlineData(1)] + [InlineData(2)] + [InlineData(3)] + [InlineData(4)] + [InlineData(5)] + [InlineData(100_001)] + public void MatchesBase64StringWithDataUrlPrefix(int length) + { + var bytes = new byte[length]; + new Random(length).NextBytes(bytes); + + Assert.Equal( + "data:image/jpeg;base64," + Convert.ToBase64String(bytes), + ScreenshotCapture.ToJpegDataUrl(bytes) + ); + } + } +} diff --git a/tests/PostHog.Unity.Tests/ScreenshotSessionTests.cs b/tests/PostHog.Unity.Tests/ScreenshotSessionTests.cs new file mode 100644 index 0000000..51febb4 --- /dev/null +++ b/tests/PostHog.Unity.Tests/ScreenshotSessionTests.cs @@ -0,0 +1,37 @@ +using PostHogUnity.SessionReplay; + +namespace PostHogUnity.Tests +{ + public class ScreenshotSessionTests + { + [Theory] + // Same session as at capture time. + [InlineData(true, "session-a", "session-a", true)] + // Reset or expiry rotated the session while the screenshot was in flight. + [InlineData(true, "session-a", "session-b", false)] + // The session ended while the screenshot was in flight. + [InlineData(true, "session-a", null, false)] + // A session started after a capture taken without one. + [InlineData(true, null, "session-b", false)] + // No session at capture or delivery, same as other replay events then. + [InlineData(true, null, null, true)] + // Replay stopped while the screenshot was in flight. + [InlineData(false, "session-a", "session-a", false)] + public void EnqueuesOnlyUnderTheCaptureSession( + bool isRunning, + string capturedSessionId, + string currentSessionId, + bool expected + ) + { + Assert.Equal( + expected, + SessionReplayIntegration.ShouldEnqueueScreenshot( + isRunning, + capturedSessionId, + currentSessionId + ) + ); + } + } +}