From f6d5d34368b2d49cfc764119bafa37c2d968a651 Mon Sep 17 00:00:00 2001 From: bponders12-cell Date: Tue, 1 Sep 2026 16:56:38 -0400 Subject: [PATCH] Harden persistent sync outbox capacity and key handling --- lib/services/sync/sync_outbox.dart | 153 +++++++++++------------------ 1 file changed, 60 insertions(+), 93 deletions(-) diff --git a/lib/services/sync/sync_outbox.dart b/lib/services/sync/sync_outbox.dart index de4d346..c5cdb75 100644 --- a/lib/services/sync/sync_outbox.dart +++ b/lib/services/sync/sync_outbox.dart @@ -3,14 +3,14 @@ import 'package:flutter/foundation.dart'; import 'package:shared_preferences/shared_preferences.dart'; import 'sync_models.dart'; -/// Persistent, crash-resilient local synchronization outbox. -/// Operates outside Drift Schema 5, ensuring zero database schema changes or migrations. +/// Persistent, crash-resilient synchronization outbox outside Drift Schema 5. class SyncOutbox { static const String _outboxStorageKey = 'motorsignal_sync_outbox_manifest_v1'; static const int maxOutboxCapacity = 500; final SharedPreferences _prefs; final Map _items = {}; + Future _saveChain = Future.value(); bool _isLoaded = false; SyncOutbox(this._prefs) { @@ -20,7 +20,6 @@ class SyncOutbox { bool get isLoaded => _isLoaded; int get count => _items.length; - /// Loads and deserializes outbox items from persistent storage with corruption recovery void load() { _items.clear(); final rawJson = _prefs.getString(_outboxStorageKey); @@ -28,96 +27,70 @@ class SyncOutbox { _isLoaded = true; return; } - try { final decoded = jsonDecode(rawJson); - if (decoded is List) { - for (final entry in decoded) { - try { - if (entry is Map) { - final item = SyncItem.fromJson(entry); - _items[item.key] = item; - } else if (entry is Map) { - final item = SyncItem.fromJson(Map.from(entry)); - _items[item.key] = item; - } - } catch (itemError) { - debugPrint('[SyncOutbox] Skipped malformed item in outbox: $itemError'); - } + if (decoded is! List) throw const FormatException('Outbox manifest is not a list'); + for (final entry in decoded) { + try { + final map = entry is Map + ? entry + : Map.from(entry as Map); + final item = SyncItem.fromJson(map); + _items[item.key] = item; + } catch (itemError) { + debugPrint('[SyncOutbox] Skipped malformed item: $itemError'); } } } catch (e) { - debugPrint('[SyncOutbox] Corruption detected in outbox manifest, resetting safely: $e'); + debugPrint('[SyncOutbox] Corrupt manifest recovered safely: $e'); _items.clear(); } - _isLoaded = true; } - /// Atomically writes current outbox manifest to persistent storage - Future _save() async { - try { - // Bound outbox capacity to avoid unbounded memory / storage growth - final list = _items.values.toList(); - if (list.length > maxOutboxCapacity) { - list.sort((a, b) => a.createdAt.compareTo(b.createdAt)); - // Keep most recent items - final trimmed = list.sublist(list.length - maxOutboxCapacity); - _items.clear(); - for (final item in trimmed) { - _items[item.key] = item; + Future _save() { + // Serialize writes so concurrent enqueue/update calls cannot overwrite + // one another with stale snapshots. + final operation = _saveChain.then((_) async { + if (_items.length > maxOutboxCapacity) { + final terminal = _items.values.where((i) => i.status.isTerminal).toList() + ..sort((a, b) => a.createdAt.compareTo(b.createdAt)); + while (_items.length > maxOutboxCapacity && terminal.isNotEmpty) { + final item = terminal.removeAt(0); + _items.remove(item.key); } } + if (_items.length > maxOutboxCapacity) { + throw StateError('Sync outbox capacity reached with no evictable terminal items.'); + } + return _prefs.setString( + _outboxStorageKey, + jsonEncode(_items.values.map((e) => e.toJson()).toList()), + ); + }); + _saveChain = operation.then((_) {}, onError: (_) {}); + return operation; + } - final encoded = jsonEncode(_items.values.map((e) => e.toJson()).toList()); - return await _prefs.setString(_outboxStorageKey, encoded); - } catch (e) { - debugPrint('[SyncOutbox] Error saving outbox manifest: $e'); - return false; + SyncItem? getItem(int localRideId, String providerId) => _items['$providerId:$localRideId']; + List getAllItems() => _items.values.toList(growable: false); + List getPendingItems() => _items.values.where((i) => i.status.isPending).toList(growable: false); + + List getEligibleItems(DateTime now) => _items.values + .where((item) => item.isEligibleForAttempt(now)) + .toList(growable: false); + + Future enqueue(int localRideId, String providerId, {DateTime? now}) async { + if (localRideId <= 0 || providerId.trim().isEmpty) { + throw ArgumentError('A positive localRideId and non-empty providerId are required.'); } - } - - /// Retrieves an item by its ride ID and provider ID - SyncItem? getItem(int localRideId, String providerId) { - final key = '${localRideId}_$providerId'; - return _items[key]; - } - - /// Returns all currently tracked items - List getAllItems() { - return _items.values.toList(growable: false); - } - - /// Returns all items requiring action or waiting for retry - List getPendingItems() { - return _items.values.where((item) => item.status.isPending).toList(growable: false); - } - - /// Returns all items currently eligible for an immediate upload attempt - List getEligibleItems(DateTime now) { - return _items.values - .where((item) => item.isEligibleForAttempt(now)) - .toList(growable: false); - } - - /// Enqueues a local ride for a target provider. - /// Deduplicates: if already queued or synced, returns existing item without duplicate insertion. - Future enqueue( - int localRideId, - String providerId, { - DateTime? now, - }) async { - final key = '${localRideId}_$providerId'; + final normalizedProvider = providerId.trim(); + final key = '$normalizedProvider:$localRideId'; final existing = _items[key]; - if (existing != null) { - if (existing.status == SyncStatus.synced) { - return existing; // Already completed + if (existing.status == SyncStatus.synced || existing.status == SyncStatus.queued || existing.status == SyncStatus.uploading) { + return existing; } - if (existing.status == SyncStatus.queued || existing.status == SyncStatus.uploading) { - return existing; // Already active in queue - } - // Re-arm failed or waiting item final rearmed = existing.copyWith( status: SyncStatus.queued, clearLastError: true, @@ -128,41 +101,35 @@ class SyncOutbox { return rearmed; } - final newItem = SyncItem( + if (_items.length >= maxOutboxCapacity && !_items.values.any((i) => i.status.isTerminal)) { + throw StateError('Sync outbox capacity reached; pending rides were not discarded.'); + } + + final item = SyncItem( localRideId: localRideId, - providerId: providerId, + providerId: normalizedProvider, status: SyncStatus.queued, createdAt: now ?? DateTime.now(), ); - - _items[key] = newItem; + _items[key] = item; await _save(); - return newItem; + return item; } - /// Updates the state of an existing outbox item Future update(SyncItem item) async { _items[item.key] = item; await _save(); } - /// Removes an item from the outbox Future remove(int localRideId, String providerId) async { - final key = '${localRideId}_$providerId'; - if (_items.remove(key) != null) { - await _save(); - } + if (_items.remove('$providerId:$localRideId') != null) await _save(); } - /// Clears the entire outbox Future clear() async { _items.clear(); + _saveChain = Future.value(); await _prefs.remove(_outboxStorageKey); } - /// Checks if a ride is marked synced for the specified provider - bool isRideSynced(int localRideId, String providerId) { - final item = getItem(localRideId, providerId); - return item?.status == SyncStatus.synced; - } + bool isRideSynced(int localRideId, String providerId) => getItem(localRideId, providerId)?.status == SyncStatus.synced; }