Skip to content

fix: add explicit cloud conflict resolution #150

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
SunkenInTime wants to merge 1 commit into from
Closed

fix: add explicit cloud conflict resolution #150

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions lib/providers/collab/active_page_live_sync_provider.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,7 @@ class ActivePageLiveSyncNotifier extends Notifier<ActivePageLiveSyncState> {
// Live reads can advance while local work blocks rehydration. Outbound diffs
// must stay based on the server state that was actually loaded into canvas.
final Map<EntitySyncKey, _NormalizedEntity> _hydratedBaseByEntityKey = {};
final Set<EntitySyncKey> _remoteAdoptionPending = {};

@override
ActivePageLiveSyncState build() {
Expand All @@ -81,6 +82,7 @@ class ActivePageLiveSyncNotifier extends Notifier<ActivePageLiveSyncState> {

void reset() {
_hydratedBaseByEntityKey.clear();
_remoteAdoptionPending.clear();
state = const ActivePageLiveSyncState();
}

Expand All @@ -97,6 +99,7 @@ class ActivePageLiveSyncNotifier extends Notifier<ActivePageLiveSyncState> {
activePageId != state.activePageId;
if (strategyChanged) {
_hydratedBaseByEntityKey.clear();
_remoteAdoptionPending.clear();
}
state = state.copyWith(
strategyPublicId: strategyPublicId,
Expand Down Expand Up @@ -138,6 +141,7 @@ class ActivePageLiveSyncNotifier extends Notifier<ActivePageLiveSyncState> {
: _normalizedRemoteEntities(snapshot, pageId);
_hydratedBaseByEntityKey.removeWhere((key, _) => key.pageId == pageId);
_hydratedBaseByEntityKey.addAll(remoteEntities);
_remoteAdoptionPending.removeWhere((key) => key.pageId == pageId);
final remoteRevisions = Map<EntitySyncKey, int>.from(
state.remoteBaseRevisionByEntity,
)..removeWhere((key, _) => key.pageId == pageId);
Expand All @@ -159,6 +163,22 @@ class ActivePageLiveSyncNotifier extends Notifier<ActivePageLiveSyncState> {
state = state.copyWith(lastAckBatch: intents);
}

/// Stops local projection and reconciliation for explicitly discarded work
/// until the affected page has loaded the authoritative remote snapshot.
void adoptRemoteForEntities(Set<EntitySyncKey> entityKeys) {
if (entityKeys.isEmpty) return;
final overlays = Map<EntitySyncKey, ActivePageOverlayEntry>.from(
state.overlayByEntityKey,
);
for (final key in entityKeys) {
overlays.remove(key);
if (key.pageId != null) {
_remoteAdoptionPending.add(key);
}
}
state = state.copyWith(overlayByEntityKey: overlays);
}

Map<EntitySyncKey, StrategyOp>? syncLocalPage({
required String strategyPublicId,
required String pageId,
Expand Down Expand Up @@ -194,6 +214,7 @@ class ActivePageLiveSyncNotifier extends Notifier<ActivePageLiveSyncState> {
.where((key) => key.pageId == pageId),
...queueState.successorByEntityKey.keys
.where((key) => key.pageId == pageId),
..._remoteAdoptionPending.where((key) => key.pageId == pageId),
};

final nextOverlay = Map<EntitySyncKey, ActivePageOverlayEntry>.from(
Expand All @@ -202,6 +223,11 @@ class ActivePageLiveSyncNotifier extends Notifier<ActivePageLiveSyncState> {
final retainedDesiredOps = <EntitySyncKey, StrategyOp>{};

for (final key in pageKeys) {
if (_remoteAdoptionPending.contains(key)) {
nextOverlay.remove(key);
_debugLog('overlay.remove $key reason=adopting_remote');
continue;
}
final remote = remoteEntities[key];
final local = localEntities[key];
final hydratedBase = _hydratedBaseByEntityKey[key];
Expand Down
88 changes: 83 additions & 5 deletions lib/providers/collab/strategy_op_queue_provider.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,6 +112,7 @@ class StrategyOpQueueNotifier extends Notifier<StrategyOpQueueState> {
int _offlineRetryCount = 0;
late DurableStrategyOutboxStore _store;
late Map<String, DurableOutboxRecord> _recordsByStorageKey;
final Set<EntitySyncKey> _awaitingRemoteAdoption = {};
Future<void> _writeTail = Future<void>.value();

ConvexStrategyRepository get _repo =>
Expand Down Expand Up @@ -147,6 +148,7 @@ class StrategyOpQueueNotifier extends Notifier<StrategyOpQueueState> {

_debounceTimer?.cancel();
_retryTimer?.cancel();
_awaitingRemoteAdoption.clear();
_offlineRetryCount = 0;
final matching = accountId == null || strategyPublicId == null
? const <DurableOutboxRecord>[]
Expand Down Expand Up @@ -323,6 +325,7 @@ class StrategyOpQueueNotifier extends Notifier<StrategyOpQueueState> {
var changed = false;
try {
for (final key in keys) {
if (_awaitingRemoteAdoption.contains(key)) continue;
final desired = desiredOps[key];
final existing = queued[key];
final inFlightIntent = state.inFlightByEntityKey[key];
Expand Down Expand Up @@ -647,6 +650,79 @@ class StrategyOpQueueNotifier extends Notifier<StrategyOpQueueState> {
});
}

/// Discards selected server-rejected intents after an explicit user choice.
///
/// Each durable record contains both the rejected predecessor and any newer
/// successor for that entity. Removing the record discards both, without
/// changing unrelated queued, in-flight, paused, or rejected work.
Future<Set<EntitySyncKey>> discardRejected(
Set<EntitySyncKey> entityKeys,
) {
return _serializeWrite(() async {
final attention = Map<EntitySyncKey, QueuedEntityIntent>.from(
state.attentionByEntityKey,
);
final successors = Map<EntitySyncKey, QueuedEntityIntent>.from(
state.successorByEntityKey,
);
final discarded = <EntitySyncKey>{};
Object? persistenceError;
StackTrace? persistenceStackTrace;

for (final key in entityKeys) {
final rejected = attention[key];
final record = _recordForActiveKey(key);
if (rejected == null ||
record == null ||
record.status != DurableOutboxStatus.attention ||
record.pending.op.opId != rejected.pending.op.opId) {
continue;
}
try {
await _store.remove(record.storageKey);
_recordsByStorageKey.remove(record.storageKey);
attention.remove(key);
successors.remove(key);
_awaitingRemoteAdoption.add(key);
discarded.add(key);
} catch (error, stackTrace) {
persistenceError = error;
persistenceStackTrace = stackTrace;
break;
}
}

if (persistenceError != null) {
log(
'Durable outbox persistence failed: $persistenceError',
name: 'strategy_outbox',
error: persistenceError,
stackTrace: persistenceStackTrace,
);
}
final attentionMessage = _loadedAttentionMessage(
loadIssues: state.loadIssues,
paused: state.pausedByEntityKey,
attention: attention,
);
final errorMessage = persistenceError == null
? attentionMessage
: 'Cloud work could not be removed from the durable outbox: '
'$persistenceError';
state = state.copyWith(
attentionByEntityKey: attention,
successorByEntityKey: successors,
lastError: errorMessage,
clearError: errorMessage == null,
);
return Set<EntitySyncKey>.unmodifiable(discarded);
});
}

void completeRemoteAdoption(Set<EntitySyncKey> entityKeys) {
_awaitingRemoteAdoption.removeAll(entityKeys);
}

Future<void> flushNow() async {
await _writeTail;
if (state.isFlushing) return;
Expand Down Expand Up @@ -968,12 +1044,14 @@ class StrategyOpQueueNotifier extends Notifier<StrategyOpQueueState> {
_recordsByStorageKey.remove(record.storageKey);
}

Future<void> _serializeWrite(Future<void> Function() action) {
Future<T> _serializeWrite<T>(Future<T> Function() action) {
final next = _writeTail.then((_) => action());
_writeTail = next.catchError((Object error, StackTrace stackTrace) {
log('Outbox write failed: $error',
name: 'strategy_outbox', error: error, stackTrace: stackTrace);
});
_writeTail = next.then<void>((_) {}).catchError(
(Object error, StackTrace stackTrace) {
log('Outbox write failed: $error',
name: 'strategy_outbox', error: error, stackTrace: stackTrace);
},
);
return next;
}

Expand Down
78 changes: 77 additions & 1 deletion lib/providers/strategy_page_session_provider.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,6 +110,7 @@ final strategyPageSessionProvider =
class StrategyPageSessionNotifier extends Notifier<StrategyPageSessionState> {
_RemotePageHydrationKey? _lastHydratedRemotePageKey;
bool _pendingRemoteReapply = false;
bool _isResolvingConflicts = false;

@override
StrategyPageSessionState build() {
Expand Down Expand Up @@ -421,6 +422,70 @@ class StrategyPageSessionNotifier extends Notifier<StrategyPageSessionState> {
}
}

/// Adopts the cloud version for every current conflict in this strategy.
///
/// The snapshot refresh happens before any local intent is discarded. Each
/// entity is removed from local projection only after its durable outbox
/// record has been deleted.
Future<bool> useCloudVersionsForRejected() async {
final strategyState = ref.read(strategyProvider);
final strategyId = strategyState.strategyId;
if (strategyState.source != StrategySource.cloud || strategyId == null) {
return false;
}

_isResolvingConflicts = true;
try {
await _resolvePageSource(strategyId, StrategySource.cloud)
.flushCurrentPage();
final strategyNotifier = ref.read(strategyProvider.notifier);
strategyNotifier.consumeScheduledCloudPageSync();
strategyNotifier.consumeScheduledCloudStrategySync();
final rejected = Map<EntitySyncKey, QueuedEntityIntent>.from(
ref.read(strategyOpQueueProvider).attentionByEntityKey,
);
if (rejected.isEmpty) return false;

await ref.read(remoteEditorSnapshotProvider.notifier).refresh();
final snapshot = ref.read(remoteEditorSnapshotProvider).valueOrNull;
if (snapshot == null || snapshot.header.publicId != strategyId) {
return false;
}

final discarded = await ref
.read(strategyOpQueueProvider.notifier)
.discardRejected(rejected.keys.toSet());
if (discarded.isEmpty) return false;

ref
.read(activePageLiveSyncProvider.notifier)
.adoptRemoteForEntities(discarded);
for (final entry in rejected.entries) {
if (discarded.contains(entry.key)) {
ref
.read(strategyConflictProvider.notifier)
.clear(entry.value.pending.op.opId);
}
}

final targetPageId = _resolveHydrationTargetPage(snapshot);
if (targetPageId != null) {
await _rehydrateActivePageFromSource(
targetPageId,
hydrationKey: _buildRemotePageHydrationKey(snapshot, targetPageId),
preserveTextDrafts: true,
);
ref
.read(strategyOpQueueProvider.notifier)
.completeRemoteAdoption(discarded);
}
_pendingRemoteReapply = false;
return true;
} finally {
_isResolvingConflicts = false;
}
}

bool get isApplyingPage => state.isApplyingPage;

void setStateForTest(StrategyPageSessionState newState) {
Expand All @@ -436,6 +501,7 @@ class StrategyPageSessionNotifier extends Notifier<StrategyPageSessionState> {
);
_lastHydratedRemotePageKey = null;
_pendingRemoteReapply = false;
_isResolvingConflicts = false;
ref.read(activePageLiveSyncProvider.notifier).reset();
}

Expand Down Expand Up @@ -486,6 +552,7 @@ class StrategyPageSessionNotifier extends Notifier<StrategyPageSessionState> {
Future<void> _rehydrateActivePageFromSource(
String pageId, {
_RemotePageHydrationKey? hydrationKey,
bool preserveTextDrafts = false,
}) async {
final strategyState = ref.read(strategyProvider);
final strategyId = strategyState.strategyId;
Expand All @@ -511,6 +578,7 @@ class StrategyPageSessionNotifier extends Notifier<StrategyPageSessionState> {
strategyId: strategyId,
source: source,
hydrationKey: hydrationKey,
preserveTextDrafts: preserveTextDrafts,
);
}

Expand All @@ -519,6 +587,7 @@ class StrategyPageSessionNotifier extends Notifier<StrategyPageSessionState> {
required String strategyId,
required StrategySource source,
_RemotePageHydrationKey? hydrationKey,
bool preserveTextDrafts = false,
}) async {
final preserveHistory = source == StrategySource.cloud &&
_lastHydratedRemotePageKey?.strategyPublicId == strategyId &&
Expand All @@ -534,6 +603,9 @@ class StrategyPageSessionNotifier extends Notifier<StrategyPageSessionState> {
await _resolvePageSource(strategyId, source).listPageIds(),
);

final retainedTextDrafts = preserveTextDrafts
? Map<String, String>.from(ref.read(textDraftProvider))
: const <String, String>{};
try {
await applyStrategyEditorPageData(
ref,
Expand All @@ -542,6 +614,9 @@ class StrategyPageSessionNotifier extends Notifier<StrategyPageSessionState> {
themeOverridePalette: themeOverridePalette,
preserveHistory: preserveHistory,
);
for (final entry in retainedTextDrafts.entries) {
ref.read(textDraftProvider.notifier).setDraft(entry.key, entry.value);
}
if (source == StrategySource.cloud) {
ref.read(activePageLiveSyncProvider.notifier).markPageHydrated(
strategyPublicId: strategyId,
Expand Down Expand Up @@ -622,7 +697,8 @@ class StrategyPageSessionNotifier extends Notifier<StrategyPageSessionState> {

bool _canSafelyReapplyRemotePage() {
final saveState = ref.read(strategySaveStateProvider);
return !state.isApplyingPage &&
return !_isResolvingConflicts &&
!state.isApplyingPage &&
state.transitionState == PageTransitionState.idle &&
ref.read(textDraftProvider).isEmpty &&
!saveState.isDirty &&
Expand Down
4 changes: 4 additions & 0 deletions lib/providers/strategy_provider.dart
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -413,6 +413,10 @@ class StrategyProvider extends Notifier<StrategyState> {
_cloudMutationSyncScheduled = false;
}

void consumeScheduledCloudStrategySync() {
_cloudStrategyMutationSyncScheduled = false;
}

void _scheduleCloudStrategySync() {
if (_cloudStrategyMutationSyncScheduled) {
return;
Expand Down
Loading
Loading