Harden persistent sync outbox capacity and key handling
This commit is contained in:
parent
ec12bf4dfc
commit
f6d5d34368
|
|
@ -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<String, SyncItem> _items = {};
|
||||
Future<void> _saveChain = Future<void>.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<String, dynamic>) {
|
||||
final item = SyncItem.fromJson(entry);
|
||||
_items[item.key] = item;
|
||||
} else if (entry is Map) {
|
||||
final item = SyncItem.fromJson(Map<String, dynamic>.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<String, dynamic>
|
||||
? entry
|
||||
: Map<String, dynamic>.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<bool> _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<bool> _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<SyncItem> getAllItems() => _items.values.toList(growable: false);
|
||||
List<SyncItem> getPendingItems() => _items.values.where((i) => i.status.isPending).toList(growable: false);
|
||||
|
||||
List<SyncItem> getEligibleItems(DateTime now) => _items.values
|
||||
.where((item) => item.isEligibleForAttempt(now))
|
||||
.toList(growable: false);
|
||||
|
||||
Future<SyncItem> 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<SyncItem> getAllItems() {
|
||||
return _items.values.toList(growable: false);
|
||||
}
|
||||
|
||||
/// Returns all items requiring action or waiting for retry
|
||||
List<SyncItem> getPendingItems() {
|
||||
return _items.values.where((item) => item.status.isPending).toList(growable: false);
|
||||
}
|
||||
|
||||
/// Returns all items currently eligible for an immediate upload attempt
|
||||
List<SyncItem> 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<SyncItem> 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<void> update(SyncItem item) async {
|
||||
_items[item.key] = item;
|
||||
await _save();
|
||||
}
|
||||
|
||||
/// Removes an item from the outbox
|
||||
Future<void> 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<void> clear() async {
|
||||
_items.clear();
|
||||
_saveChain = Future<void>.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;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue