Harden SyncManager auth recovery and startup behavior
This commit is contained in:
parent
48fa29aca2
commit
6d2d3494e3
|
|
@ -4,8 +4,7 @@ import 'sync_models.dart';
|
|||
import 'sync_outbox.dart';
|
||||
import 'sync_provider.dart';
|
||||
|
||||
/// Central, provider-agnostic synchronization manager.
|
||||
/// Coordinates queueing, deduplication, retry exponential backoff, and provider dispatch.
|
||||
/// Central provider-agnostic synchronization coordinator.
|
||||
class SyncManager extends ChangeNotifier {
|
||||
final SyncOutbox _outbox;
|
||||
final Map<String, SyncProvider> _providers = {};
|
||||
|
|
@ -13,185 +12,97 @@ class SyncManager extends ChangeNotifier {
|
|||
bool _isProcessing = false;
|
||||
DateTime? Function() _nowProvider;
|
||||
|
||||
SyncManager(
|
||||
this._outbox, {
|
||||
List<SyncProvider>? initialProviders,
|
||||
String? defaultProviderId,
|
||||
DateTime? Function()? nowProvider,
|
||||
}) : _nowProvider = nowProvider ?? (() => DateTime.now()) {
|
||||
if (initialProviders != null) {
|
||||
for (final p in initialProviders) {
|
||||
registerProvider(p);
|
||||
}
|
||||
}
|
||||
if (defaultProviderId != null) {
|
||||
_activeProviderId = defaultProviderId;
|
||||
}
|
||||
}
|
||||
|
||||
void setNowProvider(DateTime? Function() provider) {
|
||||
_nowProvider = provider;
|
||||
SyncManager(this._outbox, {List<SyncProvider>? initialProviders, String? defaultProviderId, DateTime? Function()? nowProvider}) : _nowProvider = nowProvider ?? (() => DateTime.now()) {
|
||||
if (initialProviders != null) for (final p in initialProviders) registerProvider(p);
|
||||
if (defaultProviderId != null) _activeProviderId = defaultProviderId;
|
||||
}
|
||||
|
||||
void setNowProvider(DateTime? Function() provider) => _nowProvider = provider;
|
||||
DateTime _now() => _nowProvider() ?? DateTime.now();
|
||||
|
||||
String get activeProviderId => _activeProviderId;
|
||||
set activeProviderId(String id) {
|
||||
if (_activeProviderId != id) {
|
||||
_activeProviderId = id;
|
||||
notifyListeners();
|
||||
}
|
||||
}
|
||||
|
||||
set activeProviderId(String id) { if (_activeProviderId != id) { _activeProviderId = id; notifyListeners(); } }
|
||||
bool get isProcessing => _isProcessing;
|
||||
int get outboxCount => _outbox.count;
|
||||
List<SyncItem> get allItems => _outbox.getAllItems();
|
||||
List<SyncItem> get pendingItems => _outbox.getPendingItems();
|
||||
|
||||
SyncManagerState get currentState {
|
||||
final items = _outbox.getAllItems();
|
||||
int pending = 0;
|
||||
int synced = 0;
|
||||
int failed = 0;
|
||||
int pending = 0, synced = 0, failed = 0;
|
||||
DateTime? lastSuccess;
|
||||
String? lastErr;
|
||||
|
||||
for (final item in items) {
|
||||
for (final item in _outbox.getAllItems()) {
|
||||
if (item.status == SyncStatus.synced) {
|
||||
synced++;
|
||||
if (item.uploadedAt != null) {
|
||||
if (lastSuccess == null || item.uploadedAt!.isAfter(lastSuccess)) {
|
||||
lastSuccess = item.uploadedAt;
|
||||
}
|
||||
}
|
||||
if (item.uploadedAt != null && (lastSuccess == null || item.uploadedAt!.isAfter(lastSuccess))) lastSuccess = item.uploadedAt;
|
||||
} else if (item.status == SyncStatus.failed || item.status == SyncStatus.cancelled) {
|
||||
failed++;
|
||||
if (item.lastError != null) lastErr = item.lastError;
|
||||
} else {
|
||||
pending++;
|
||||
}
|
||||
} else { pending++; }
|
||||
}
|
||||
|
||||
return SyncManagerState(
|
||||
isProcessing: _isProcessing,
|
||||
pendingCount: pending,
|
||||
syncedCount: synced,
|
||||
failedCount: failed,
|
||||
lastSuccessfulSync: lastSuccess,
|
||||
lastError: lastErr,
|
||||
activeProviderId: _activeProviderId,
|
||||
);
|
||||
return SyncManagerState(isProcessing: _isProcessing, pendingCount: pending, syncedCount: synced, failedCount: failed, lastSuccessfulSync: lastSuccess, lastError: lastErr, activeProviderId: _activeProviderId);
|
||||
}
|
||||
|
||||
/// Registers a synchronization provider implementation
|
||||
void registerProvider(SyncProvider provider) {
|
||||
_providers[provider.providerId] = provider;
|
||||
if (_activeProviderId == 'none' || _activeProviderId == 'fake') {
|
||||
_activeProviderId = provider.providerId;
|
||||
}
|
||||
if (_activeProviderId == 'none' || _activeProviderId == 'fake') _activeProviderId = provider.providerId;
|
||||
notifyListeners();
|
||||
}
|
||||
|
||||
/// Retrieves a registered provider by ID
|
||||
SyncProvider? getProvider(String providerId) => _providers[providerId];
|
||||
|
||||
/// Enqueues a local completed ride for synchronization
|
||||
Future<SyncItem> enqueueRide(
|
||||
int localRideId, {
|
||||
String? targetProviderId,
|
||||
bool autoProcess = true,
|
||||
}) async {
|
||||
Future<SyncItem> enqueueRide(int localRideId, {String? targetProviderId, bool autoProcess = true}) async {
|
||||
final providerId = targetProviderId ?? _activeProviderId;
|
||||
final item = await _outbox.enqueue(localRideId, providerId, now: _now());
|
||||
notifyListeners();
|
||||
|
||||
if (autoProcess) {
|
||||
// Fire-and-forget processing on microtask queue
|
||||
unawaited(processQueue());
|
||||
}
|
||||
|
||||
if (autoProcess) unawaited(processQueue());
|
||||
return item;
|
||||
}
|
||||
|
||||
/// Calculates exponential backoff duration based on attempt number:
|
||||
/// attempt 1 -> 5s
|
||||
/// attempt 2 -> 30s
|
||||
/// attempt 3 -> 2m (120s)
|
||||
/// attempt 4 -> 5m (300s)
|
||||
/// attempt 5 -> 15m (900s)
|
||||
/// attempt 6+ -> 30m (1800s) maximum
|
||||
/// Rearms only authentication-blocked work. Call this after a successful login/link.
|
||||
Future<void> retryAuthRequired() async {
|
||||
for (final item in _outbox.getAllItems().where((i) => i.status == SyncStatus.authRequired)) {
|
||||
await _outbox.update(item.copyWith(status: SyncStatus.queued, clearLastError: true, clearNextRetry: true));
|
||||
}
|
||||
await processQueue();
|
||||
}
|
||||
|
||||
static Duration calculateRetryDelay(int attemptCount) {
|
||||
switch (attemptCount) {
|
||||
case 1:
|
||||
return const Duration(seconds: 5);
|
||||
case 2:
|
||||
return const Duration(seconds: 30);
|
||||
case 3:
|
||||
return const Duration(minutes: 2);
|
||||
case 4:
|
||||
return const Duration(minutes: 5);
|
||||
case 5:
|
||||
return const Duration(minutes: 15);
|
||||
default:
|
||||
return const Duration(minutes: 30);
|
||||
case 1: return const Duration(seconds: 5);
|
||||
case 2: return const Duration(seconds: 30);
|
||||
case 3: return const Duration(minutes: 2);
|
||||
case 4: return const Duration(minutes: 5);
|
||||
case 5: return const Duration(minutes: 15);
|
||||
default: return const Duration(minutes: 30);
|
||||
}
|
||||
}
|
||||
|
||||
/// Processes eligible outbox items sequentially without blocking the caller or UI thread
|
||||
Future<void> processQueue() async {
|
||||
if (_isProcessing) return;
|
||||
|
||||
_isProcessing = true;
|
||||
notifyListeners();
|
||||
|
||||
try {
|
||||
final now = _now();
|
||||
// Snapshot eligible items once; each item is durably marked before network I/O.
|
||||
final eligibleItems = _outbox.getEligibleItems(now);
|
||||
|
||||
for (final item in eligibleItems) {
|
||||
final provider = _providers[item.providerId];
|
||||
if (provider == null) {
|
||||
final updated = item.copyWith(
|
||||
status: SyncStatus.failed,
|
||||
lastError: 'Provider "${item.providerId}" not registered',
|
||||
lastAttemptAt: now,
|
||||
);
|
||||
await _outbox.update(updated);
|
||||
await _outbox.update(item.copyWith(status: SyncStatus.failed, lastError: 'Provider "${item.providerId}" not registered', lastAttemptAt: now));
|
||||
continue;
|
||||
}
|
||||
if (!await provider.isAvailable()) {
|
||||
await _outbox.update(item.copyWith(status: SyncStatus.waitingForNetwork, lastAttemptAt: now));
|
||||
continue;
|
||||
}
|
||||
if (await provider.requiresAuthentication()) {
|
||||
await _outbox.update(item.copyWith(status: SyncStatus.authRequired, lastError: 'Authentication required for provider "${item.providerId}"', lastAttemptAt: now));
|
||||
continue;
|
||||
}
|
||||
|
||||
// Check provider availability / network
|
||||
final isAvailable = await provider.isAvailable();
|
||||
if (!isAvailable) {
|
||||
final updated = item.copyWith(
|
||||
status: SyncStatus.waitingForNetwork,
|
||||
lastAttemptAt: now,
|
||||
);
|
||||
await _outbox.update(updated);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Check authentication requirement
|
||||
final requiresAuth = await provider.requiresAuthentication();
|
||||
if (requiresAuth) {
|
||||
final updated = item.copyWith(
|
||||
status: SyncStatus.authRequired,
|
||||
lastError: 'Authentication required for provider "${item.providerId}"',
|
||||
lastAttemptAt: now,
|
||||
);
|
||||
await _outbox.update(updated);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Mark in-progress
|
||||
final uploadingItem = item.copyWith(
|
||||
status: SyncStatus.uploading,
|
||||
lastAttemptAt: now,
|
||||
);
|
||||
final uploadingItem = item.copyWith(status: SyncStatus.uploading, lastAttemptAt: now);
|
||||
await _outbox.update(uploadingItem);
|
||||
notifyListeners();
|
||||
|
||||
// Perform upload through provider
|
||||
SyncResult result;
|
||||
try {
|
||||
result = await provider.uploadRide(uploadingItem);
|
||||
|
|
@ -200,41 +111,14 @@ class SyncManager extends ChangeNotifier {
|
|||
}
|
||||
|
||||
if (result.success) {
|
||||
final syncedItem = uploadingItem.copyWith(
|
||||
status: SyncStatus.synced,
|
||||
remoteId: result.remoteId,
|
||||
uploadedAt: result.timestamp,
|
||||
clearLastError: true,
|
||||
clearNextRetry: true,
|
||||
);
|
||||
await _outbox.update(syncedItem);
|
||||
await _outbox.update(uploadingItem.copyWith(status: SyncStatus.synced, remoteId: result.remoteId, uploadedAt: result.timestamp, clearLastError: true, clearNextRetry: true));
|
||||
} else if (result.requiresAuth) {
|
||||
final authItem = uploadingItem.copyWith(
|
||||
status: SyncStatus.authRequired,
|
||||
lastError: result.errorMessage,
|
||||
);
|
||||
await _outbox.update(authItem);
|
||||
await _outbox.update(uploadingItem.copyWith(status: SyncStatus.authRequired, lastError: result.errorMessage));
|
||||
} else if (result.isRetryable) {
|
||||
final newAttempts = uploadingItem.attemptCount + 1;
|
||||
final delay = calculateRetryDelay(newAttempts);
|
||||
final nextRetry = now.add(delay);
|
||||
|
||||
final retryItem = uploadingItem.copyWith(
|
||||
status: SyncStatus.retryWait,
|
||||
attemptCount: newAttempts,
|
||||
nextRetryAt: nextRetry,
|
||||
lastError: result.errorMessage,
|
||||
);
|
||||
await _outbox.update(retryItem);
|
||||
final attempts = uploadingItem.attemptCount + 1;
|
||||
await _outbox.update(uploadingItem.copyWith(status: SyncStatus.retryWait, attemptCount: attempts, nextRetryAt: now.add(calculateRetryDelay(attempts)), lastError: result.errorMessage));
|
||||
} else {
|
||||
// Non-retryable permanent failure
|
||||
final failedItem = uploadingItem.copyWith(
|
||||
status: SyncStatus.failed,
|
||||
attemptCount: uploadingItem.attemptCount + 1,
|
||||
lastError: result.errorMessage,
|
||||
clearNextRetry: true,
|
||||
);
|
||||
await _outbox.update(failedItem);
|
||||
await _outbox.update(uploadingItem.copyWith(status: SyncStatus.failed, attemptCount: uploadingItem.attemptCount + 1, lastError: result.errorMessage, clearNextRetry: true));
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
|
|
@ -243,25 +127,17 @@ class SyncManager extends ChangeNotifier {
|
|||
}
|
||||
}
|
||||
|
||||
/// Cancels an active or pending item in the outbox
|
||||
Future<void> cancelSync(int localRideId, String providerId) async {
|
||||
final item = _outbox.getItem(localRideId, providerId);
|
||||
if (item != null && item.status != SyncStatus.synced) {
|
||||
final cancelled = item.copyWith(status: SyncStatus.cancelled);
|
||||
await _outbox.update(cancelled);
|
||||
if (item != null && !item.status.isTerminal) {
|
||||
await _outbox.update(item.copyWith(status: SyncStatus.cancelled));
|
||||
notifyListeners();
|
||||
}
|
||||
}
|
||||
|
||||
/// Manually retries all failed, waiting, or auth-required items immediately
|
||||
Future<void> retryAllPending() async {
|
||||
for (final item in _outbox.getPendingItems()) {
|
||||
final rearmed = item.copyWith(
|
||||
status: SyncStatus.queued,
|
||||
clearLastError: true,
|
||||
clearNextRetry: true,
|
||||
);
|
||||
await _outbox.update(rearmed);
|
||||
await _outbox.update(item.copyWith(status: SyncStatus.queued, clearLastError: true, clearNextRetry: true));
|
||||
}
|
||||
await processQueue();
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue