From 48fa29aca2396cd81557d8ff3fb78af10497aa19 Mon Sep 17 00:00:00 2001 From: bponders12-cell Date: Tue, 1 Sep 2026 16:57:09 -0400 Subject: [PATCH] Recover interrupted sync uploads after process restart --- lib/services/sync/sync_outbox.dart | 110 +++++++---------------------- 1 file changed, 26 insertions(+), 84 deletions(-) diff --git a/lib/services/sync/sync_outbox.dart b/lib/services/sync/sync_outbox.dart index c5cdb75..e02d077 100644 --- a/lib/services/sync/sync_outbox.dart +++ b/lib/services/sync/sync_outbox.dart @@ -3,133 +3,75 @@ import 'package:flutter/foundation.dart'; import 'package:shared_preferences/shared_preferences.dart'; import 'sync_models.dart'; -/// 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) { - load(); - } - + SyncOutbox(this._prefs) { load(); } bool get isLoaded => _isLoaded; int get count => _items.length; void load() { _items.clear(); final rawJson = _prefs.getString(_outboxStorageKey); - if (rawJson == null || rawJson.trim().isEmpty) { - _isLoaded = true; - return; - } + if (rawJson == null || rawJson.trim().isEmpty) { _isLoaded = true; return; } try { final decoded = jsonDecode(rawJson); 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 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'); - } + // An app/process death cannot leave an upload genuinely in progress. + // Requeue it so the provider can safely retry its deterministic operation. + _items[item.key] = item.status == SyncStatus.uploading + ? item.copyWith(status: SyncStatus.queued, clearNextRetry: true) + : item; + } catch (itemError) { debugPrint('[SyncOutbox] Skipped malformed item: $itemError'); } } - } catch (e) { - debugPrint('[SyncOutbox] Corrupt manifest recovered safely: $e'); - _items.clear(); - } + } catch (e) { debugPrint('[SyncOutbox] Corrupt manifest recovered safely: $e'); _items.clear(); } _isLoaded = true; } 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); - } + final terminal = _items.values.where((i) => i.status.isTerminal).toList()..sort((a, b) => a.createdAt.compareTo(b.createdAt)); + while (_items.length > maxOutboxCapacity && terminal.isNotEmpty) _items.remove(terminal.removeAt(0).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()), - ); + 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; } - SyncItem? getItem(int localRideId, String providerId) => _items['$providerId:$localRideId']; + SyncItem? getItem(int localRideId, String providerId) => _items['${providerId.trim()}:$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); + List getEligibleItems(DateTime now) => _items.values.where((i) => i.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.'); - } + if (localRideId <= 0 || providerId.trim().isEmpty) throw ArgumentError('A positive localRideId and non-empty providerId are required.'); final normalizedProvider = providerId.trim(); final key = '$normalizedProvider:$localRideId'; final existing = _items[key]; if (existing != null) { - if (existing.status == SyncStatus.synced || existing.status == SyncStatus.queued || existing.status == SyncStatus.uploading) { - return existing; - } - final rearmed = existing.copyWith( - status: SyncStatus.queued, - clearLastError: true, - clearNextRetry: true, - ); - _items[key] = rearmed; - await _save(); - return rearmed; + if (existing.status == SyncStatus.synced || existing.status == SyncStatus.queued || existing.status == SyncStatus.uploading) return existing; + final rearmed = existing.copyWith(status: SyncStatus.queued, clearLastError: true, clearNextRetry: true); + _items[key] = rearmed; await _save(); return rearmed; } - - 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: normalizedProvider, - status: SyncStatus.queued, - createdAt: now ?? DateTime.now(), - ); - _items[key] = item; - await _save(); - return item; - } - - Future update(SyncItem item) async { - _items[item.key] = item; - await _save(); - } - - Future remove(int localRideId, String providerId) async { - if (_items.remove('$providerId:$localRideId') != null) await _save(); - } - - Future clear() async { - _items.clear(); - _saveChain = Future.value(); - await _prefs.remove(_outboxStorageKey); + 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: normalizedProvider, status: SyncStatus.queued, createdAt: now ?? DateTime.now()); + _items[key] = item; await _save(); return item; } + Future update(SyncItem item) async { _items[item.key] = item; await _save(); } + Future remove(int localRideId, String providerId) async { if (_items.remove('${providerId.trim()}:$localRideId') != null) await _save(); } + Future clear() async { _items.clear(); _saveChain = Future.value(); await _prefs.remove(_outboxStorageKey); } bool isRideSynced(int localRideId, String providerId) => getItem(localRideId, providerId)?.status == SyncStatus.synced; }