fix: incremental sync + unit tests for decide() logic
Sync change detection (3rd attempt — now correct): - After UPLOAD: save null remote metadata (server mtime unknown until next listing); decide() treats null remoteModifiedAt as "not changed" - After DOWNLOAD: read actual local mtime via accessor.lastModifiedMs() so the stored value matches what walkFiles() sees on next scan - SKIP reconciliation: if known state has null timestamps and both sides exist, fill them in — stabilises state within 2 syncs after first transfer - Extract syncDecide() as internal top-level function for testability Unit tests (14 cases covering all key scenarios): - First sync decisions (upload/download/conflict) - Second sync after upload with null remote metadata → SKIP - Second sync after download with recorded local mtime → SKIP - Epoch-millis precision: same ms = SKIP, +1ms = change detected - Regression: epoch-second stored value would have differed → now correct - Delete behaviour (MIRROR vs KEEP) - Direction filters (UPLOAD_ONLY, DOWNLOAD_ONLY) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -8,6 +8,7 @@ import com.syncflow.data.db.SyncFileStateDao
|
||||
import com.syncflow.data.db.SyncPairDao
|
||||
import com.syncflow.data.db.entities.SyncConflictEntity
|
||||
import com.syncflow.data.db.entities.SyncEventEntity
|
||||
import com.syncflow.data.db.entities.SyncFileStateEntity
|
||||
import com.syncflow.data.providers.CloudProvider
|
||||
import com.syncflow.domain.model.ConflictStrategy
|
||||
import com.syncflow.domain.model.DeleteBehavior
|
||||
@@ -76,7 +77,7 @@ class SyncEngine @Inject constructor(
|
||||
val uploaded: Int = 0, val downloaded: Int = 0, val deleted: Int = 0,
|
||||
val skipped: Int = 0, val failed: Int = 0, val conflicts: Int = 0,
|
||||
val bytesTransferred: Long = 0L,
|
||||
val newState: com.syncflow.data.db.entities.SyncFileStateEntity? = null,
|
||||
val newState: SyncFileStateEntity? = null,
|
||||
)
|
||||
|
||||
val outcomes: List<FileOutcome> = coroutineScope {
|
||||
@@ -86,7 +87,7 @@ class SyncEngine @Inject constructor(
|
||||
val local = localFiles[rel]
|
||||
val remote = remoteFiles[rel]
|
||||
val known = knownStates[rel]
|
||||
val decision = decide(pair.syncDirection, pair.conflictStrategy, pair.deleteBehavior, local, remote, known)
|
||||
val decision = syncDecide(pair.syncDirection, pair.conflictStrategy, pair.deleteBehavior, local, remote, known)
|
||||
|
||||
when (decision) {
|
||||
SyncDecision.UPLOAD -> {
|
||||
@@ -101,7 +102,10 @@ class SyncEngine @Inject constructor(
|
||||
return@withPermit FileOutcome(failed = 1)
|
||||
}
|
||||
logEvent(pair.id, SyncEventType.FILE_UPLOADED, rel, null, bytes)
|
||||
FileOutcome(uploaded = 1, bytesTransferred = bytes, newState = buildState(pair.id, rel, local!!, remote))
|
||||
// Remote metadata is unknown until the next listing; save null so
|
||||
// decide() treats it as "not changed" and reconciles on next SKIP.
|
||||
FileOutcome(uploaded = 1, bytesTransferred = bytes,
|
||||
newState = buildState(pair.id, rel, local!!, remoteAfterTransfer = null))
|
||||
}
|
||||
SyncDecision.DOWNLOAD -> {
|
||||
val bytes = runCatching {
|
||||
@@ -114,8 +118,14 @@ class SyncEngine @Inject constructor(
|
||||
logEvent(pair.id, SyncEventType.FILE_SKIPPED, rel, e.message, 0)
|
||||
return@withPermit FileOutcome(failed = 1)
|
||||
}
|
||||
// Read the actual local mtime written by the OS/SAF after download.
|
||||
val localMtime = runCatching { accessor.lastModifiedMs(rel) }
|
||||
.getOrDefault(System.currentTimeMillis()).takeIf { it > 0L }
|
||||
?: System.currentTimeMillis()
|
||||
logEvent(pair.id, SyncEventType.FILE_DOWNLOADED, rel, null, bytes)
|
||||
FileOutcome(downloaded = 1, bytesTransferred = bytes, newState = buildState(pair.id, rel, null, remote))
|
||||
FileOutcome(downloaded = 1, bytesTransferred = bytes,
|
||||
newState = buildState(pair.id, rel,
|
||||
LocalFileInfo(rel, remote!!.sizeBytes, localMtime), remoteAfterTransfer = remote))
|
||||
}
|
||||
SyncDecision.DELETE_LOCAL -> {
|
||||
accessor.delete(rel)
|
||||
@@ -143,7 +153,19 @@ class SyncEngine @Inject constructor(
|
||||
logEvent(pair.id, SyncEventType.CONFLICT_DETECTED, rel, null, 0)
|
||||
FileOutcome(conflicts = 1)
|
||||
}
|
||||
SyncDecision.SKIP -> FileOutcome(skipped = 1)
|
||||
SyncDecision.SKIP -> {
|
||||
// Reconcile: if the known state is missing remote or local metadata
|
||||
// (saved right after an upload before we had the server's response),
|
||||
// fill it in now that we have the full listing.
|
||||
val needsReconcile = known != null &&
|
||||
(known.remoteModifiedAt == null || known.localModifiedAt == null) &&
|
||||
local != null && remote != null
|
||||
if (needsReconcile) {
|
||||
FileOutcome(skipped = 1, newState = buildState(pair.id, rel, local, remoteAfterTransfer = remote))
|
||||
} else {
|
||||
FileOutcome(skipped = 1)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -162,86 +184,20 @@ class SyncEngine @Inject constructor(
|
||||
)
|
||||
}
|
||||
|
||||
private fun decide(
|
||||
direction: SyncDirection,
|
||||
conflictStrategy: ConflictStrategy,
|
||||
deleteBehavior: DeleteBehavior,
|
||||
local: LocalFileInfo?,
|
||||
remote: RemoteFile?,
|
||||
known: com.syncflow.data.db.entities.SyncFileStateEntity?,
|
||||
): SyncDecision {
|
||||
val localExists = local != null
|
||||
val remoteExists = remote != null
|
||||
|
||||
val localChanged = known == null || (localExists && local!!.lastModifiedMs != known.localModifiedAt?.toEpochMilli())
|
||||
val remoteChanged = known == null || (remoteExists && (remote!!.etag != known.remoteEtag || remote.modifiedAt != known.remoteModifiedAt))
|
||||
|
||||
return when {
|
||||
!localExists && !remoteExists -> SyncDecision.SKIP
|
||||
|
||||
localExists && !remoteExists -> when {
|
||||
known == null -> when (direction) {
|
||||
SyncDirection.UPLOAD_ONLY, SyncDirection.TWO_WAY -> SyncDecision.UPLOAD
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
else -> when {
|
||||
deleteBehavior == DeleteBehavior.KEEP -> SyncDecision.SKIP
|
||||
direction == SyncDirection.DOWNLOAD_ONLY || direction == SyncDirection.TWO_WAY -> SyncDecision.DELETE_LOCAL
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
}
|
||||
|
||||
!localExists && remoteExists -> when {
|
||||
known == null -> when (direction) {
|
||||
SyncDirection.DOWNLOAD_ONLY, SyncDirection.TWO_WAY -> SyncDecision.DOWNLOAD
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
else -> when {
|
||||
deleteBehavior == DeleteBehavior.KEEP -> SyncDecision.SKIP
|
||||
direction == SyncDirection.UPLOAD_ONLY || direction == SyncDirection.TWO_WAY -> SyncDecision.DELETE_REMOTE
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
}
|
||||
|
||||
localChanged && remoteChanged -> when (direction) {
|
||||
SyncDirection.UPLOAD_ONLY -> SyncDecision.UPLOAD
|
||||
SyncDirection.DOWNLOAD_ONLY -> SyncDecision.DOWNLOAD
|
||||
SyncDirection.TWO_WAY -> when (conflictStrategy) {
|
||||
ConflictStrategy.KEEP_LOCAL -> SyncDecision.UPLOAD
|
||||
ConflictStrategy.KEEP_REMOTE -> SyncDecision.DOWNLOAD
|
||||
ConflictStrategy.KEEP_NEWEST -> if ((local?.lastModifiedMs ?: 0L) >= (remote?.modifiedAt?.toEpochMilli() ?: 0L)) SyncDecision.UPLOAD else SyncDecision.DOWNLOAD
|
||||
ConflictStrategy.KEEP_LARGEST -> if ((local?.sizeBytes ?: 0L) >= (remote?.sizeBytes ?: 0L)) SyncDecision.UPLOAD else SyncDecision.DOWNLOAD
|
||||
ConflictStrategy.KEEP_BOTH -> SyncDecision.CONFLICT
|
||||
ConflictStrategy.ASK -> SyncDecision.CONFLICT
|
||||
}
|
||||
}
|
||||
|
||||
localChanged -> when (direction) {
|
||||
SyncDirection.UPLOAD_ONLY, SyncDirection.TWO_WAY -> SyncDecision.UPLOAD
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
remoteChanged -> when (direction) {
|
||||
SyncDirection.DOWNLOAD_ONLY, SyncDirection.TWO_WAY -> SyncDecision.DOWNLOAD
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
}
|
||||
|
||||
private fun buildState(
|
||||
pairId: Long,
|
||||
rel: String,
|
||||
local: LocalFileInfo?,
|
||||
remote: RemoteFile?,
|
||||
) = com.syncflow.data.db.entities.SyncFileStateEntity(
|
||||
remoteAfterTransfer: RemoteFile?,
|
||||
) = SyncFileStateEntity(
|
||||
syncPairId = pairId,
|
||||
relativePath = rel,
|
||||
localModifiedAt = local?.lastModifiedMs?.let { Instant.ofEpochMilli(it) },
|
||||
localSizeBytes = local?.sizeBytes ?: 0L,
|
||||
localHash = null,
|
||||
remoteModifiedAt = remote?.modifiedAt,
|
||||
remoteSizeBytes = remote?.sizeBytes ?: 0L,
|
||||
remoteEtag = remote?.etag,
|
||||
remoteModifiedAt = remoteAfterTransfer?.modifiedAt,
|
||||
remoteSizeBytes = remoteAfterTransfer?.sizeBytes ?: 0L,
|
||||
remoteEtag = remoteAfterTransfer?.etag,
|
||||
lastSyncedAt = Instant.now(),
|
||||
syncedHash = null,
|
||||
)
|
||||
@@ -251,6 +207,79 @@ class SyncEngine @Inject constructor(
|
||||
}
|
||||
}
|
||||
|
||||
// Top-level so unit tests can call it directly without instantiating SyncEngine.
|
||||
internal fun syncDecide(
|
||||
direction: SyncDirection,
|
||||
conflictStrategy: ConflictStrategy,
|
||||
deleteBehavior: DeleteBehavior,
|
||||
local: LocalFileInfo?,
|
||||
remote: RemoteFile?,
|
||||
known: SyncFileStateEntity?,
|
||||
): SyncDecision {
|
||||
val localExists = local != null
|
||||
val remoteExists = remote != null
|
||||
|
||||
// Treat null known timestamps as "not yet recorded" — don't treat as changed.
|
||||
// The SKIP reconciliation pass will fill them in on the next sync.
|
||||
val localChanged = known == null ||
|
||||
(localExists && known.localModifiedAt != null &&
|
||||
local!!.lastModifiedMs != known.localModifiedAt.toEpochMilli())
|
||||
val remoteChanged = known == null ||
|
||||
(remoteExists && known.remoteModifiedAt != null &&
|
||||
(remote!!.etag != known.remoteEtag || remote.modifiedAt != known.remoteModifiedAt))
|
||||
|
||||
return when {
|
||||
!localExists && !remoteExists -> SyncDecision.SKIP
|
||||
|
||||
localExists && !remoteExists -> when {
|
||||
known == null -> when (direction) {
|
||||
SyncDirection.UPLOAD_ONLY, SyncDirection.TWO_WAY -> SyncDecision.UPLOAD
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
else -> when {
|
||||
deleteBehavior == DeleteBehavior.KEEP -> SyncDecision.SKIP
|
||||
direction == SyncDirection.DOWNLOAD_ONLY || direction == SyncDirection.TWO_WAY -> SyncDecision.DELETE_LOCAL
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
}
|
||||
|
||||
!localExists && remoteExists -> when {
|
||||
known == null -> when (direction) {
|
||||
SyncDirection.DOWNLOAD_ONLY, SyncDirection.TWO_WAY -> SyncDecision.DOWNLOAD
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
else -> when {
|
||||
deleteBehavior == DeleteBehavior.KEEP -> SyncDecision.SKIP
|
||||
direction == SyncDirection.UPLOAD_ONLY || direction == SyncDirection.TWO_WAY -> SyncDecision.DELETE_REMOTE
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
}
|
||||
|
||||
localChanged && remoteChanged -> when (direction) {
|
||||
SyncDirection.UPLOAD_ONLY -> SyncDecision.UPLOAD
|
||||
SyncDirection.DOWNLOAD_ONLY -> SyncDecision.DOWNLOAD
|
||||
SyncDirection.TWO_WAY -> when (conflictStrategy) {
|
||||
ConflictStrategy.KEEP_LOCAL -> SyncDecision.UPLOAD
|
||||
ConflictStrategy.KEEP_REMOTE -> SyncDecision.DOWNLOAD
|
||||
ConflictStrategy.KEEP_NEWEST -> if ((local?.lastModifiedMs ?: 0L) >= (remote?.modifiedAt?.toEpochMilli() ?: 0L)) SyncDecision.UPLOAD else SyncDecision.DOWNLOAD
|
||||
ConflictStrategy.KEEP_LARGEST -> if ((local?.sizeBytes ?: 0L) >= (remote?.sizeBytes ?: 0L)) SyncDecision.UPLOAD else SyncDecision.DOWNLOAD
|
||||
ConflictStrategy.KEEP_BOTH -> SyncDecision.CONFLICT
|
||||
ConflictStrategy.ASK -> SyncDecision.CONFLICT
|
||||
}
|
||||
}
|
||||
|
||||
localChanged -> when (direction) {
|
||||
SyncDirection.UPLOAD_ONLY, SyncDirection.TWO_WAY -> SyncDecision.UPLOAD
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
remoteChanged -> when (direction) {
|
||||
SyncDirection.DOWNLOAD_ONLY, SyncDirection.TWO_WAY -> SyncDecision.DOWNLOAD
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
else -> SyncDecision.SKIP
|
||||
}
|
||||
}
|
||||
|
||||
enum class SyncDecision { UPLOAD, DOWNLOAD, DELETE_LOCAL, DELETE_REMOTE, CONFLICT, SKIP }
|
||||
|
||||
data class SyncResult(
|
||||
|
||||
Reference in New Issue
Block a user