From 0825476ec03a40abbbbe9ce87966f0a98a24b45f Mon Sep 17 00:00:00 2001 From: Marian Paul Date: Mon, 26 Oct 2026 21:11:37 +0100 Subject: [PATCH] fix: trim the persisted queue in one pass When uploads keep failing, every queue change re-encoded the whole queue and, past 512 KB, _trimQueueToTargetMap re-encoded it again after each removed event, which is quadratic and runs on the UI isolate. encodeQueue encodes events newest first, each at most once, and stops once the limit is passed, so a write costs at most ~512 KB of encoding regardless of queue length. Same 512 KB / 475 KB thresholds as before. Fixes #199 --- packages/core/lib/utils/store/io.dart | 88 +++++++-------------- packages/core/test/utils/store/io_test.dart | 58 ++++++++++++++ 2 files changed, 86 insertions(+), 60 deletions(-) create mode 100644 packages/core/test/utils/store/io_test.dart diff --git a/packages/core/lib/utils/store/io.dart b/packages/core/lib/utils/store/io.dart index 3f57ec7..9764b0d 100644 --- a/packages/core/lib/utils/store/io.dart +++ b/packages/core/lib/utils/store/io.dart @@ -4,6 +4,7 @@ import 'dart:convert'; import 'dart:io'; import 'dart:typed_data'; +import 'package:flutter/foundation.dart' show visibleForTesting; import 'package:segment_analytics/errors.dart'; import 'package:segment_analytics/utils/store/store.dart'; import 'package:path_provider/path_provider.dart'; @@ -61,31 +62,11 @@ class StoreImpl with Store { } Future _writeFile(String fileKey, Map data) async { - // Serialize once; may be replaced by trimmed version below - String serialized = json.encode(data); - List buffer = utf8.encode(serialized); - - // If this is the queue file, enforce size limits by trimming oldest events - if (fileKey == _kQueueFileKey) { - // Ensure we operate against the configured queue field name. - final currentEvents = []; - if (data[_kQueueField] is List) { - currentEvents.addAll(data[_kQueueField] as List); - } - - // If the serialized payload is already too large, attempt trim before writing - if (buffer.length > _kMaxQueueBytes) { - final trimmed = _trimQueueToTargetMap(data, _kTargetQueueBytes, _kQueueField); - serialized = json.encode(trimmed); - buffer = utf8.encode(serialized); - - // If still too big after trimming to target, write empty queue as last resort - if (buffer.length > _kMaxQueueBytes) { - serialized = json.encode({_kQueueField: []}); - buffer = utf8.encode(serialized); - } - } - } + final queue = data[_kQueueField]; + final String serialized = fileKey == _kQueueFileKey && queue is List + ? encodeQueue(queue, _kQueueField, _kMaxQueueBytes, _kTargetQueueBytes) + : json.encode(data); + final List buffer = utf8.encode(serialized); RandomAccessFile? file; try { @@ -210,40 +191,6 @@ class StoreImpl with Store { } } - /// Trim the queue in [data] (expected to be a map containing an array under [queueField]) - /// until the serialized size is <= targetBytes. Returns a new Map containing - /// the trimmed queue list. - Map _trimQueueToTargetMap( - Map data, int targetBytes, String queueField) { - final events = []; - if (data[queueField] is List) { - events.addAll(data[queueField] as List); - } - - // If no events or not a list, return minimal representation - if (events.isEmpty) { - return {queueField: []}; - } - - // Fast path: if current serialized size is already small enough, return input - var candidate = {queueField: events}; - var s = json.encode(candidate); - var b = utf8.encode(s); - if (b.length <= targetBytes) { - return candidate; - } - - // Iteratively remove oldest events until serialized size <= targetBytes or no events left - while (b.length > targetBytes && events.isNotEmpty) { - events.removeAt(0); - candidate = {queueField: events}; - s = json.encode(candidate); - b = utf8.encode(s); - } - - return {queueField: events}; - } - Future?> _readFile(String fileKey) async { RandomAccessFile? file = await _getFile(fileKey); if (file == null) { @@ -370,4 +317,25 @@ class StoreImpl with Store { return; } } -} \ No newline at end of file +} + +/// Encodes `{queueField: events}`. If that would exceed [maxBytes], the oldest +/// events are dropped so the result fits in [targetBytes]. +@visibleForTesting +String encodeQueue( + List events, String queueField, int maxBytes, int targetBytes) { + final parts = []; + var bytes = utf8.encode(json.encode({queueField: []})).length; + int? fitsTarget; + for (var i = events.length - 1; i >= 0; i--) { + final part = json.encode(events[i]); + bytes += utf8.encode(part).length + (parts.isEmpty ? 0 : 1); + if (bytes > targetBytes) fitsTarget ??= parts.length; + if (bytes > maxBytes) { + parts.length = fitsTarget!; + break; + } + parts.add(part); + } + return '{${json.encode(queueField)}:[${parts.reversed.join(',')}]}'; +} diff --git a/packages/core/test/utils/store/io_test.dart b/packages/core/test/utils/store/io_test.dart new file mode 100644 index 0000000..1d70688 --- /dev/null +++ b/packages/core/test/utils/store/io_test.dart @@ -0,0 +1,58 @@ +import 'dart:convert'; + +import 'package:flutter_test/flutter_test.dart'; +import 'package:segment_analytics/utils/store/io.dart'; + +class _Event { + _Event(this.id); + final int id; + int encodes = 0; + + Map toJson() { + encodes++; + return {'id': id, 'padding': 'x' * 1000}; + } +} + +List _ids(String encoded) => (json.decode(encoded)['queue'] as List) + .map((e) => e['id'] as int) + .toList(); + +void main() { + const maxBytes = 512 * 1024; + const targetBytes = 475 * 1024; + + test('encodeQueue keeps every event under the limit', () { + final events = List.generate(10, (i) => {'id': i}); + final encoded = encodeQueue(events, 'queue', maxBytes, targetBytes); + expect(encoded, json.encode({'queue': events})); + }); + + test('encodeQueue trims to target keeping the newest events', () { + final events = List.generate(2000, _Event.new); + final encoded = encodeQueue(events, 'queue', maxBytes, targetBytes); + final ids = _ids(encoded); + + expect(utf8.encode(encoded).length, lessThanOrEqualTo(targetBytes)); + expect(ids.last, 1999); + expect(ids, List.generate(ids.length, (i) => 2000 - ids.length + i)); + final oneMore = json.encode({'queue': events.sublist(1999 - ids.length)}); + expect(utf8.encode(oneMore).length, greaterThan(targetBytes)); + }); + + test('encodeQueue encodes each event at most once', () { + final events = List.generate(20000, _Event.new); + encodeQueue(events, 'queue', maxBytes, targetBytes); + + expect(events.every((e) => e.encodes <= 1), isTrue); + expect(events.where((e) => e.encodes == 1).length, lessThan(600)); + }); + + test('encodeQueue drops an event larger than the limit', () { + final events = [ + {'id': 0, 'padding': 'x' * (maxBytes + 1)} + ]; + expect(encodeQueue(events, 'queue', maxBytes, targetBytes), + '{"queue":[]}'); + }); +}