fix(sync): стабилизировал синхронизацию, Yandex I/O и вёрстку карточки storage

Добавил TRASH вместо DELETE для moveToTrash, компакцию журналов и отчёт об ошибках apply.
Исправил проброс ошибок upload Yandex при close, CAS lock и загрузку OAuth-токена.
Упростил совместимость sync-групп (только encInfo), поправил растягивание StorageTree при недоступных meta.
This commit is contained in:
2026-05-21 18:46:03 +03:00
parent ef40aa9e73
commit 51e6f40587
18 changed files with 268 additions and 89 deletions

View File

@@ -42,6 +42,7 @@ class StorageSyncBootstrap @Inject constructor(
merge(*triggers.toTypedArray()) merge(*triggers.toTypedArray())
.debounce(DEBOUNCE_AFTER_CHANGE_MS) .debounce(DEBOUNCE_AFTER_CHANGE_MS)
.collect { .collect {
// RunStorageSyncUseCase.enqueue отбрасывает повтор, пока sync уже в очереди/в работе.
syncRunner.enqueue( syncRunner.enqueue(
displayTitle = uiStrings(R.string.task_title_storage_sync_background), displayTitle = uiStrings(R.string.task_title_storage_sync_background),
logReason = "debounce", logReason = "debounce",

View File

@@ -54,28 +54,26 @@ class YandexDiskApiFactory(
} }
/** /**
* Провайдер токена читает [YandexAccountRepository] на IO-диспетчере (как и остальной data-слой). * OAuth-токен загружается один раз при создании API (не в OkHttp interceptor).
* При 401 см. [YandexDiskAuthException] и повторную привязку vault.
*/ */
fun createApiForVault(vaultUuid: UUID): YandexDiskApi { fun createApiForVault(vaultUuid: UUID): YandexDiskApi {
val id = vaultUuid.toString() val id = vaultUuid.toString()
val token = runBlocking(ioDispatcher) {
accountRepository.getByVaultUuid(id)?.oauthToken
} ?: throw java.io.IOException("Yandex OAuth token is missing")
oauthTokenCache[id] = System.currentTimeMillis() to token
return createAuthenticatedApi { return createAuthenticatedApi {
val now = System.currentTimeMillis() oauthTokenCache[id]?.second ?: token
val hit = oauthTokenCache[id]
if (hit != null && now - hit.first < OAUTH_TOKEN_CACHE_TTL_MS) {
return@createAuthenticatedApi hit.second
}
val token = runBlocking(ioDispatcher) {
accountRepository.getByVaultUuid(id)?.oauthToken
} ?: throw java.io.IOException("Yandex OAuth token is missing")
oauthTokenCache[id] = now to token
token
} }
} }
fun invalidateTokenCache(vaultUuid: UUID) {
oauthTokenCache.remove(vaultUuid.toString())
}
companion object { companion object {
const val BASE_URL = "https://cloud-api.yandex.net/" const val BASE_URL = "https://cloud-api.yandex.net/"
private const val OAUTH_TOKEN_CACHE_TTL_MS = 120_000L
fun createRepositoryWithToken( fun createRepositoryWithToken(
oauthToken: String, oauthToken: String,
ioDispatcher: CoroutineDispatcher, ioDispatcher: CoroutineDispatcher,

View File

@@ -46,6 +46,7 @@ class EncryptedStorageAccessor(
private val scope: CoroutineScope private val scope: CoroutineScope
) : IStorageAccessor, DisposableHandle { ) : IStorageAccessor, DisposableHandle {
private val syncActorId = UUID.randomUUID().toString() private val syncActorId = UUID.randomUUID().toString()
private val syncLockMutex = Mutex()
private val _size = MutableStateFlow<Long?>(null) private val _size = MutableStateFlow<Long?>(null)
override val size: StateFlow<Long?> = _size override val size: StateFlow<Long?> = _size
@@ -63,7 +64,6 @@ class EncryptedStorageAccessor(
private val dataEncryptor = Encryptor(key.toAesKey()) private val dataEncryptor = Encryptor(key.toAesKey())
private val pathEncryptor: EncryptorWithStaticIv? = if(pathIv != null) EncryptorWithStaticIv(key.toAesKey(), pathIv) else null private val pathEncryptor: EncryptorWithStaticIv? = if(pathIv != null) EncryptorWithStaticIv(key.toAesKey(), pathIv) else null
private val syncLockMutex = Mutex()
private var systemHiddenFilesIsActual = false private var systemHiddenFilesIsActual = false
init { init {
@@ -266,7 +266,7 @@ class EncryptedStorageAccessor(
override suspend fun moveToTrash(path: String) { override suspend fun moveToTrash(path: String) {
source.moveToTrash(encryptPath(path)) source.moveToTrash(encryptPath(path))
appendSyncEntry(path = path, operation = StorageSyncOperation.DELETE) appendSyncEntry(path = path, operation = StorageSyncOperation.TRASH)
} }
override fun dispose() { override fun dispose() {

View File

@@ -547,7 +547,7 @@ class LocalStorageAccessor(
?: throw WallencException.Storage.FileNotFound() ?: throw WallencException.Storage.FileNotFound()
val newMeta = pair.meta.copy(isDeleted = true) val newMeta = pair.meta.copy(isDeleted = true)
writeMeta(pair.metaFile, newMeta) writeMeta(pair.metaFile, newMeta)
appendSyncEntry(path = path, operation = StorageSyncOperation.DELETE) appendSyncEntry(path = path, operation = StorageSyncOperation.TRASH)
} }
override suspend fun openReadSystemFile(name: String): InputStream = withContext(ioDispatcher) { override suspend fun openReadSystemFile(name: String): InputStream = withContext(ioDispatcher) {

View File

@@ -532,34 +532,8 @@ class YandexStorageAccessor(
val tmp = File.createTempFile("wallenc-yandex-", ".upload") val tmp = File.createTempFile("wallenc-yandex-", ".upload")
val fos = FileOutputStream(tmp) val fos = FileOutputStream(tmp)
fos.onClosed { fos.onClosed {
runBlocking(ioDispatcher) { runCommitAfterStreamClosed {
try { commitUploadedFile(path, tmp)
val diskPath = toDiskPath(path)
val prior = guard { repo.getOrNull(diskPath) }
if (prior?.type == "dir") {
throw WallencException.Storage.CannotWriteOverDirectory()
}
val hadFile = prior?.type == "file"
val priorSize = if (prior?.type == "file") prior.size ?: 0L else 0L
guard { repo.uploadFile(diskPath, tmp, overwrite = true) }
val after = guard { getMetadataAfterWrite(diskPath) }
if (after.type != "file") {
throw WallencException.Storage.UnexpectedState()
}
val newSize = after.size ?: 0L
_size.value = ((_size.value ?: 0L) + newSize - priorSize).coerceAtLeast(0L)
if (!hadFile) {
_numberOfFiles.value = (_numberOfFiles.value ?: 0) + 1
}
persistStatsImmediate()
val info = runCatching { after.toCommonFile(path) }.getOrNull()
info?.let {
_filesUpdates.emit(DataPage(listOf(it), pageLength = 1, pageIndex = 0))
}
appendSyncEntry(path = path, operation = StorageSyncOperation.UPSERT)
} finally {
tmp.delete()
}
} }
} }
} }
@@ -570,7 +544,7 @@ class YandexStorageAccessor(
override suspend fun moveToTrash(path: String) = withContext(ioDispatcher) { override suspend fun moveToTrash(path: String) = withContext(ioDispatcher) {
patchCustomProps(path, mapOf(PROP_DELETED to "true")) patchCustomProps(path, mapOf(PROP_DELETED to "true"))
appendSyncEntry(path = path, operation = StorageSyncOperation.DELETE) appendSyncEntry(path = path, operation = StorageSyncOperation.TRASH)
} }
override suspend fun openReadSystemFile(name: String): InputStream = withContext(ioDispatcher) { override suspend fun openReadSystemFile(name: String): InputStream = withContext(ioDispatcher) {
@@ -589,10 +563,10 @@ class YandexStorageAccessor(
val rel = "/$SYSTEM_HIDDEN_DIRNAME/$name" val rel = "/$SYSTEM_HIDDEN_DIRNAME/$name"
val uploadBuffer = ByteArrayOutputStream() val uploadBuffer = ByteArrayOutputStream()
uploadBuffer.onClosed { uploadBuffer.onClosed {
runBlocking(ioDispatcher) { val bytes = uploadBuffer.toByteArray()
guard { val diskPath = toDiskPath(rel)
repo.uploadBytes(toDiskPath(rel), uploadBuffer.toByteArray(), overwrite = true) runCommitAfterStreamClosed {
} guard { repo.uploadBytes(diskPath, bytes, overwrite = true) }
} }
} }
} }
@@ -641,8 +615,25 @@ class YandexStorageAccessor(
}.getOrNull() }.getOrNull()
} }
/**
* Best-effort lock на Диске (read-modify-write без CAS на стороне API).
* Межустройственная координация опирается на [StorageSyncEngine] mutex в процессе.
*/
override suspend fun tryAcquireSyncLock(holderId: String, leaseUntil: Instant): Boolean = withContext(ioDispatcher) { override suspend fun tryAcquireSyncLock(holderId: String, leaseUntil: Instant): Boolean = withContext(ioDispatcher) {
return@withContext syncLockMutex.withLock { repeat(SYNC_LOCK_CAS_RETRIES) { attempt ->
val acquired = tryAcquireSyncLockOnce(holderId, leaseUntil)
if (acquired) {
return@withContext true
}
if (attempt < SYNC_LOCK_CAS_RETRIES - 1) {
delay(SYNC_LOCK_CAS_DELAY_MS)
}
}
return@withContext false
}
private suspend fun tryAcquireSyncLockOnce(holderId: String, leaseUntil: Instant): Boolean {
return syncLockMutex.withLock {
val current = readSyncLock() val current = readSyncLock()
val now = Instant.now() val now = Instant.now()
val foreignLockActive = current != null && val foreignLockActive = current != null &&
@@ -684,6 +675,45 @@ class YandexStorageAccessor(
} }
} }
/**
* Выполняется из [OutputStream.close]; ошибки upload пробрасываются вызывающему коду.
*/
private fun runCommitAfterStreamClosed(block: suspend () -> Unit) {
runBlocking(ioDispatcher) {
block()
}
}
private suspend fun commitUploadedFile(path: String, tmp: File) {
try {
val diskPath = toDiskPath(path)
val prior = guard { repo.getOrNull(diskPath) }
if (prior?.type == "dir") {
throw WallencException.Storage.CannotWriteOverDirectory()
}
val hadFile = prior?.type == "file"
val priorSize = if (prior?.type == "file") prior.size ?: 0L else 0L
guard { repo.uploadFile(diskPath, tmp, overwrite = true) }
val after = guard { getMetadataAfterWrite(diskPath) }
if (after.type != "file") {
throw WallencException.Storage.UnexpectedState()
}
val newSize = after.size ?: 0L
_size.value = ((_size.value ?: 0L) + newSize - priorSize).coerceAtLeast(0L)
if (!hadFile) {
_numberOfFiles.value = (_numberOfFiles.value ?: 0) + 1
}
persistStatsImmediate()
val info = runCatching { after.toCommonFile(path) }.getOrNull()
info?.let {
_filesUpdates.emit(DataPage(listOf(it), pageLength = 1, pageIndex = 0))
}
appendSyncEntry(path = path, operation = StorageSyncOperation.UPSERT)
} finally {
tmp.delete()
}
}
private suspend fun appendSyncEntry(path: String, operation: StorageSyncOperation) { private suspend fun appendSyncEntry(path: String, operation: StorageSyncOperation) {
val cleanedPath = if (path.startsWith("/")) path else "/$path" val cleanedPath = if (path.startsWith("/")) path else "/$path"
val entries = readSyncJournal() val entries = readSyncJournal()
@@ -726,6 +756,8 @@ class YandexStorageAccessor(
private const val SYNC_JOURNAL_FILENAME = "sync-journal.json" private const val SYNC_JOURNAL_FILENAME = "sync-journal.json"
private const val SYNC_LOCK_FILENAME = "sync-lock.json" private const val SYNC_LOCK_FILENAME = "sync-lock.json"
private const val SYNC_LOCK_STALE_TIMEOUT_SECONDS: Long = 60 * 60 private const val SYNC_LOCK_STALE_TIMEOUT_SECONDS: Long = 60 * 60
private const val SYNC_LOCK_CAS_RETRIES = 3
private const val SYNC_LOCK_CAS_DELAY_MS = 80L
private const val STATS_DEBOUNCE_MS = 450L private const val STATS_DEBOUNCE_MS = 450L
private const val DATA_PAGE_LENGTH = 10 private const val DATA_PAGE_LENGTH = 10
private const val API_LIST_LIMIT = 1000 private const val API_LIST_LIMIT = 1000

View File

@@ -27,11 +27,19 @@ private class CloseHandledOutputStream(
override fun close() { override fun close() {
onClosing() onClosing()
var streamFailure: Throwable? = null
try { try {
stream.close() stream.close()
} finally { } catch (t: Throwable) {
onClose() streamFailure = t
} }
try {
onClose()
} catch (afterCloseFailure: Throwable) {
streamFailure?.let { afterCloseFailure.addSuppressed(it) }
throw afterCloseFailure
}
streamFailure?.let { throw it }
} }
} }

View File

@@ -5,6 +5,9 @@ import java.util.UUID
enum class StorageSyncOperation { enum class StorageSyncOperation {
UPSERT, UPSERT,
/** Soft-delete (корзина): на peer вызывается [IStorageAccessor.moveToTrash]. */
TRASH,
/** Жёсткое удаление файла с носителя. */
DELETE, DELETE,
} }

View File

@@ -20,6 +20,7 @@ sealed class TaskProgressLabel {
data class SyncGroupProcessingEntries(val groupId: String, val count: Int) : TaskProgressLabel() data class SyncGroupProcessingEntries(val groupId: String, val count: Int) : TaskProgressLabel()
data class SyncGroupEntryProgress(val groupId: String, val current: Int, val total: Int) : TaskProgressLabel() data class SyncGroupEntryProgress(val groupId: String, val current: Int, val total: Int) : TaskProgressLabel()
data class SyncGroupCompleted(val groupId: String) : TaskProgressLabel() data class SyncGroupCompleted(val groupId: String) : TaskProgressLabel()
data class SyncGroupEntriesFailed(val groupId: String, val failedCount: Int) : TaskProgressLabel()
data class SyncGroupRenewingLocks(val groupId: String) : TaskProgressLabel() data class SyncGroupRenewingLocks(val groupId: String) : TaskProgressLabel()
data class SyncGroupLockRenewalFailed(val groupId: String) : TaskProgressLabel() data class SyncGroupLockRenewalFailed(val groupId: String) : TaskProgressLabel()

View File

@@ -5,14 +5,12 @@ import androidx.compose.foundation.clickable
import androidx.compose.foundation.interaction.MutableInteractionSource import androidx.compose.foundation.interaction.MutableInteractionSource
import androidx.compose.foundation.layout.Box import androidx.compose.foundation.layout.Box
import androidx.compose.foundation.layout.Column import androidx.compose.foundation.layout.Column
import androidx.compose.foundation.layout.IntrinsicSize
import androidx.compose.foundation.layout.Row import androidx.compose.foundation.layout.Row
import androidx.compose.foundation.layout.Spacer
import androidx.compose.foundation.layout.fillMaxSize
import androidx.compose.foundation.layout.fillMaxWidth import androidx.compose.foundation.layout.fillMaxWidth
import androidx.compose.foundation.layout.height
import androidx.compose.foundation.layout.offset import androidx.compose.foundation.layout.offset
import androidx.compose.foundation.layout.padding import androidx.compose.foundation.layout.padding
import androidx.compose.foundation.layout.widthIn
import androidx.compose.foundation.layout.wrapContentHeight
import androidx.compose.material.icons.Icons import androidx.compose.material.icons.Icons
import androidx.compose.material.icons.filled.Lock import androidx.compose.material.icons.filled.Lock
import androidx.compose.material.icons.filled.LockOpen import androidx.compose.material.icons.filled.LockOpen
@@ -90,17 +88,16 @@ fun StorageTree(
Column(modifier) { Column(modifier) {
Box( Box(
modifier = Modifier modifier = Modifier
.height(IntrinsicSize.Min) .fillMaxWidth()
.wrapContentHeight()
.zIndex(100f), .zIndex(100f),
) { ) {
val interactionSource = remember { MutableInteractionSource() } val interactionSource = remember { MutableInteractionSource() }
Box( Box(
modifier = Modifier modifier = Modifier
.clip( .matchParentSize()
CardDefaults.shape, .padding(end = 16.dp)
) .clip(CardDefaults.shape)
.padding(0.dp, 0.dp, 16.dp, 0.dp)
.fillMaxSize()
.background(borderColor) .background(borderColor)
.clickable( .clickable(
interactionSource = interactionSource, interactionSource = interactionSource,
@@ -112,8 +109,9 @@ fun StorageTree(
Card( Card(
interactionSource = interactionSource, interactionSource = interactionSource,
modifier = Modifier modifier = Modifier
.padding(8.dp, 0.dp, 0.dp, 0.dp) .padding(start = 8.dp)
.fillMaxWidth(), .fillMaxWidth()
.wrapContentHeight(),
elevation = CardDefaults.cardElevation( elevation = CardDefaults.cardElevation(
defaultElevation = 4.dp, defaultElevation = 4.dp,
), ),
@@ -124,8 +122,16 @@ fun StorageTree(
} }
}, },
) { ) {
Row(modifier = Modifier.height(IntrinsicSize.Min)) { Row(
Column(modifier = Modifier.padding(8.dp)) { modifier = Modifier
.fillMaxWidth()
.wrapContentHeight(),
) {
Column(
modifier = Modifier
.weight(1f)
.padding(8.dp),
) {
Text(metaInfo.name ?: stringResource(R.string.no_name)) Text(metaInfo.name ?: stringResource(R.string.no_name))
Text( Text(
text = stringResource( text = stringResource(
@@ -170,7 +176,9 @@ fun StorageTree(
} }
} }
Column( Column(
modifier = Modifier, modifier = Modifier
.widthIn(min = 112.dp)
.padding(end = 4.dp),
horizontalAlignment = Alignment.End, horizontalAlignment = Alignment.End,
) { ) {
var expanded by remember { mutableStateOf(false) } var expanded by remember { mutableStateOf(false) }
@@ -368,7 +376,6 @@ fun StorageTree(
) )
} }
} }
Spacer(modifier = Modifier.weight(1f))
if (isEncrypted) { if (isEncrypted) {
IconButton( IconButton(
onClick = { showLockDialog = true }, onClick = { showLockDialog = true },
@@ -382,18 +389,14 @@ fun StorageTree(
} }
Text( Text(
modifier = Modifier modifier = Modifier
.fillMaxWidth() .padding(top = 4.dp, end = 8.dp),
.padding(0.dp, 0.dp, 12.dp, 0.dp)
.align(Alignment.End),
text = stringResource(getStatusTextRes(tree)), text = stringResource(getStatusTextRes(tree)),
textAlign = TextAlign.End, textAlign = TextAlign.End,
fontSize = 11.sp, fontSize = 11.sp,
) )
Text( Text(
modifier = Modifier modifier = Modifier
.fillMaxWidth() .padding(end = 8.dp, bottom = 8.dp),
.padding(0.dp, 0.dp, 12.dp, 8.dp)
.align(Alignment.End),
text = cur.uuid.toString(), text = cur.uuid.toString(),
textAlign = TextAlign.End, textAlign = TextAlign.End,
fontSize = 8.sp, fontSize = 8.sp,

View File

@@ -37,6 +37,8 @@ fun TaskProgressLabel.resolve(resolver: UiStringResolver): String = when (this)
resolver(R.string.sync_progress_group_entry, groupId, current, total) resolver(R.string.sync_progress_group_entry, groupId, current, total)
is TaskProgressLabel.SyncGroupCompleted -> is TaskProgressLabel.SyncGroupCompleted ->
resolver(R.string.sync_progress_group_completed, groupId) resolver(R.string.sync_progress_group_completed, groupId)
is TaskProgressLabel.SyncGroupEntriesFailed ->
resolver.plurals(R.plurals.sync_progress_group_entries_failed, failedCount, groupId, failedCount)
is TaskProgressLabel.SyncGroupRenewingLocks -> is TaskProgressLabel.SyncGroupRenewingLocks ->
resolver(R.string.sync_progress_group_renewing_locks, groupId) resolver(R.string.sync_progress_group_renewing_locks, groupId)
is TaskProgressLabel.SyncGroupLockRenewalFailed -> is TaskProgressLabel.SyncGroupLockRenewalFailed ->

View File

@@ -383,7 +383,6 @@ class StorageSyncViewModel @Inject constructor(
!isStorageCompatibleWithGroup( !isStorageCompatibleWithGroup(
storage = storage, storage = storage,
group = group, group = group,
resolveStorageKey = vaultsManager.unlockManager::getOpenedStorageKey,
) )
} }
StorageSyncGroupUi( StorageSyncGroupUi(

View File

@@ -12,4 +12,10 @@
<item quantity="many">Синхронизация: группа «%1$s» — обработка %2$d записей</item> <item quantity="many">Синхронизация: группа «%1$s» — обработка %2$d записей</item>
<item quantity="other">Синхронизация: группа «%1$s» — обработка %2$d записей</item> <item quantity="other">Синхронизация: группа «%1$s» — обработка %2$d записей</item>
</plurals> </plurals>
<plurals name="sync_progress_group_entries_failed">
<item quantity="one">Синхронизация: группа «%1$s» — не применена %2$d запись</item>
<item quantity="few">Синхронизация: группа «%1$s» — не применены %2$d записи</item>
<item quantity="many">Синхронизация: группа «%1$s» — не применено %2$d записей</item>
<item quantity="other">Синхронизация: группа «%1$s» — не применено %2$d записей</item>
</plurals>
</resources> </resources>

View File

@@ -8,4 +8,8 @@
<item quantity="one">Storage sync: group "%1$s" processing %2$d entry</item> <item quantity="one">Storage sync: group "%1$s" processing %2$d entry</item>
<item quantity="other">Storage sync: group "%1$s" processing %2$d entries</item> <item quantity="other">Storage sync: group "%1$s" processing %2$d entries</item>
</plurals> </plurals>
<plurals name="sync_progress_group_entries_failed">
<item quantity="one">Storage sync: group "%1$s" — %2$d entry failed</item>
<item quantity="other">Storage sync: group "%1$s" — %2$d entries failed</item>
</plurals>
</resources> </resources>

View File

@@ -1,17 +1,14 @@
package com.github.nullptroma.wallenc.usecases package com.github.nullptroma.wallenc.usecases
import com.github.nullptroma.wallenc.domain.datatypes.EncryptKey
import com.github.nullptroma.wallenc.domain.interfaces.IStorage import com.github.nullptroma.wallenc.domain.interfaces.IStorage
import com.github.nullptroma.wallenc.domain.interfaces.StorageSyncGroup import com.github.nullptroma.wallenc.domain.interfaces.StorageSyncGroup
import com.github.nullptroma.wallenc.domain.interfaces.StorageSyncGroupEncryptionKind import com.github.nullptroma.wallenc.domain.interfaces.StorageSyncGroupEncryptionKind
import java.util.UUID
/** Совместим, если у storage нет активного шифрования ([encInfo] == null). */
fun isStorageCompatibleWithGroup( fun isStorageCompatibleWithGroup(
storage: IStorage, storage: IStorage,
group: StorageSyncGroup, group: StorageSyncGroup,
resolveStorageKey: (UUID) -> EncryptKey?,
): Boolean { ): Boolean {
// Режим упрощён: в sync-группах допускаются только незашифрованные storage.
if (storage.metaInfo.value.encInfo != null) { if (storage.metaInfo.value.encInfo != null) {
return false return false
} }

View File

@@ -19,6 +19,11 @@ import java.util.concurrent.atomic.AtomicLong
import javax.inject.Inject import javax.inject.Inject
import javax.inject.Singleton import javax.inject.Singleton
/**
* Синхронизация по журналам storage в группе.
* Блокировка на Yandex Disk — best-effort (см. [IStorageAccessor.tryAcquireSyncLock]);
* сериализация внутри процесса — [groupMutexes].
*/
@Singleton @Singleton
class StorageSyncEngine @Inject constructor( class StorageSyncEngine @Inject constructor(
private val vaultsManager: IVaultsManager, private val vaultsManager: IVaultsManager,
@@ -81,7 +86,6 @@ class StorageSyncEngine @Inject constructor(
isStorageCompatibleWithGroup( isStorageCompatibleWithGroup(
storage = storage, storage = storage,
group = group, group = group,
resolveStorageKey = vaultsManager.unlockManager::getOpenedStorageKey,
) )
} }
if (incompatible.isNotEmpty()) { if (incompatible.isNotEmpty()) {
@@ -148,6 +152,7 @@ class StorageSyncEngine @Inject constructor(
null, null,
TaskProgressLabel.SyncGroupProcessingEntries(groupId, mergedEntries.size), TaskProgressLabel.SyncGroupProcessingEntries(groupId, mergedEntries.size),
) )
var applyFailures = 0
for ((pathIndex, merged) in mergedEntries.withIndex()) { for ((pathIndex, merged) in mergedEntries.withIndex()) {
leaseUntil = renewLocksIfNeeded( leaseUntil = renewLocksIfNeeded(
groupId = groupId, groupId = groupId,
@@ -166,7 +171,9 @@ class StorageSyncEngine @Inject constructor(
return return
} }
val sourceStorage = findSourceStorage(storages, entriesByStorage, path, winnerEntry) val sourceStorage = findSourceStorage(storages, entriesByStorage, path, winnerEntry)
if (sourceStorage == null && winnerEntry.operation == StorageSyncOperation.UPSERT) { if (sourceStorage == null &&
winnerEntry.operation == StorageSyncOperation.UPSERT
) {
continue continue
} }
@@ -185,9 +192,18 @@ class StorageSyncEngine @Inject constructor(
) )
if (applied) { if (applied) {
target.accessor.appendSyncJournal(listOf(winnerEntry)) target.accessor.appendSyncJournal(listOf(winnerEntry))
} else {
applyFailures++
} }
} }
} }
compactSyncJournals(storages)
if (applyFailures > 0) {
reportProgress(
null,
TaskProgressLabel.SyncGroupEntriesFailed(groupId, applyFailures),
)
}
reportProgress(null, TaskProgressLabel.SyncGroupCompleted(groupId)) reportProgress(null, TaskProgressLabel.SyncGroupCompleted(groupId))
} finally { } finally {
for (accessor in lockedAccessors) { for (accessor in lockedAccessors) {
@@ -244,8 +260,12 @@ class StorageSyncEngine @Inject constructor(
path: String, path: String,
winnerEntry: StorageSyncJournalEntry, winnerEntry: StorageSyncJournalEntry,
): IStorage? { ): IStorage? {
if (winnerEntry.operation == StorageSyncOperation.DELETE) { if (winnerEntry.operation == StorageSyncOperation.DELETE ||
return storages.firstOrNull() winnerEntry.operation == StorageSyncOperation.TRASH
) {
return storages.firstOrNull { storage ->
entriesByStorage[storage.uuid]?.get(path) != null
} ?: storages.firstOrNull()
} }
return storages.firstOrNull { storage -> return storages.firstOrNull { storage ->
val entry = entriesByStorage[storage.uuid]?.get(path) ?: return@firstOrNull false val entry = entriesByStorage[storage.uuid]?.get(path) ?: return@firstOrNull false
@@ -253,29 +273,52 @@ class StorageSyncEngine @Inject constructor(
} }
} }
private suspend fun compactSyncJournals(storages: List<IStorage>) {
for (storage in storages) {
val entries = storage.accessor.readSyncJournal()
val compacted = latestByPath(entries).values.toList()
if (compacted.size < entries.size) {
storage.accessor.rewriteSyncJournal(compacted)
}
}
}
private suspend fun applyEntry( private suspend fun applyEntry(
source: IStorage?, source: IStorage?,
target: IStorage, target: IStorage,
entry: StorageSyncJournalEntry, entry: StorageSyncJournalEntry,
): Boolean { ): Boolean {
when (entry.operation) { val result = when (entry.operation) {
StorageSyncOperation.DELETE -> { StorageSyncOperation.DELETE -> {
return runCatching { runCatching {
target.accessor.delete(entry.path) target.accessor.delete(entry.path)
}.isSuccess }
}
StorageSyncOperation.TRASH -> {
runCatching {
target.accessor.moveToTrash(entry.path)
}
} }
StorageSyncOperation.UPSERT -> { StorageSyncOperation.UPSERT -> {
val sourceAccessor = source?.accessor ?: return false val sourceAccessor = source?.accessor ?: return false
return runCatching { runCatching {
sourceAccessor.openRead(entry.path).use { input -> sourceAccessor.openRead(entry.path).use { input ->
target.accessor.openWrite(entry.path).use { output -> target.accessor.openWrite(entry.path).use { output ->
input.copyTo(output) input.copyTo(output)
} }
} }
}.isSuccess }
} }
} }
result.exceptionOrNull()?.let { error ->
System.err.println(
"StorageSyncEngine: apply ${entry.operation} ${entry.path} " +
"target=${target.uuid}: ${error.message}",
)
}
return result.isSuccess
} }
private fun compareEntries(a: StorageSyncJournalEntry, b: StorageSyncJournalEntry): Int { private fun compareEntries(a: StorageSyncJournalEntry, b: StorageSyncJournalEntry): Int {

View File

@@ -0,0 +1,41 @@
package com.github.nullptroma.wallenc.usecases
import com.github.nullptroma.wallenc.domain.datatypes.StorageEncryptionInfo
import com.github.nullptroma.wallenc.domain.interfaces.StorageSyncGroup
import com.github.nullptroma.wallenc.domain.interfaces.StorageSyncGroupEncryptionKind
import com.github.nullptroma.wallenc.usecases.fakes.FakeMetaInfo
import com.github.nullptroma.wallenc.usecases.fakes.FakeStorage
import org.junit.Assert.assertFalse
import org.junit.Assert.assertTrue
import org.junit.Test
import java.util.UUID
class StorageSyncEncryptionCompatTest {
@Test
fun storageWithoutEncInfoIsCompatible() {
val storage = FakeStorage(uuid = UUID.randomUUID(), meta = FakeMetaInfo(encInfo = null))
val group = StorageSyncGroup(
id = "g1",
storageUuids = setOf(storage.uuid),
encryptionKind = StorageSyncGroupEncryptionKind.NONE,
)
assertTrue(isStorageCompatibleWithGroup(storage = storage, group = group))
}
@Test
fun storageWithEncInfoIsIncompatible() {
val storage = FakeStorage(
uuid = UUID.randomUUID(),
meta = FakeMetaInfo(
encInfo = StorageEncryptionInfo(encryptedTestData = "x", pathIv = ByteArray(16)),
),
)
val group = StorageSyncGroup(
id = "g1",
storageUuids = setOf(storage.uuid),
encryptionKind = StorageSyncGroupEncryptionKind.NONE,
)
assertFalse(isStorageCompatibleWithGroup(storage = storage, group = group))
}
}

View File

@@ -7,6 +7,7 @@ import com.github.nullptroma.wallenc.domain.interfaces.StorageSyncGroup
import com.github.nullptroma.wallenc.domain.interfaces.StorageSyncGroupEncryptionKind import com.github.nullptroma.wallenc.domain.interfaces.StorageSyncGroupEncryptionKind
import com.github.nullptroma.wallenc.domain.tasks.TaskProgressLabel import com.github.nullptroma.wallenc.domain.tasks.TaskProgressLabel
import com.github.nullptroma.wallenc.usecases.fakes.FakeStorage import com.github.nullptroma.wallenc.usecases.fakes.FakeStorage
import com.github.nullptroma.wallenc.usecases.fakes.FakeStorageAccessor
import com.github.nullptroma.wallenc.usecases.fakes.FakeStorageSyncGroupStore import com.github.nullptroma.wallenc.usecases.fakes.FakeStorageSyncGroupStore
import com.github.nullptroma.wallenc.usecases.fakes.FakeVaultsManager import com.github.nullptroma.wallenc.usecases.fakes.FakeVaultsManager
import kotlinx.coroutines.runBlocking import kotlinx.coroutines.runBlocking
@@ -112,6 +113,41 @@ class StorageSyncEngineTest {
assertNull(target.fileBytes(path)) assertNull(target.fileBytes(path))
} }
@Test
fun syncGroupTrashSoftDeletesOnTarget() = runBlocking {
val source = FakeStorage()
val target = FakeStorage()
val path = "trashed/doc.txt"
val payload = "keep-in-trash".encodeToByteArray()
source.putFile(path, payload)
target.putFile(path, payload)
val entry = StorageSyncJournalEntry(
path = path,
operation = StorageSyncOperation.TRASH,
revision = StorageSyncRevision(
sequence = 3L,
actorId = "actor-trash",
createdAt = Instant.parse("2024-07-01T00:00:00Z"),
),
)
source.addSyncJournalEntry(entry)
val group = StorageSyncGroup(
id = "trash-group",
storageUuids = setOf(source.uuid, target.uuid),
encryptionKind = StorageSyncGroupEncryptionKind.NONE,
)
val engine = createEngine(
storages = listOf(source, target),
groups = listOf(group),
)
engine.syncGroup(group.id) { _, _ -> }
assertArrayEquals(payload, target.fileBytes(path))
assertTrue(path in (target.accessor as FakeStorageAccessor).trashedPaths)
}
@Test @Test
fun syncGroupStopsWhenLockCannotBeAcquired() = runBlocking { fun syncGroupStopsWhenLockCannotBeAcquired() = runBlocking {
val first = FakeStorage() val first = FakeStorage()

View File

@@ -21,6 +21,7 @@ import java.time.Instant
class FakeStorageAccessor : IStorageAccessor { class FakeStorageAccessor : IStorageAccessor {
val dataFiles: MutableMap<String, ByteArray> = mutableMapOf() val dataFiles: MutableMap<String, ByteArray> = mutableMapOf()
val trashedPaths: MutableSet<String> = mutableSetOf()
private val systemFiles: MutableMap<String, ByteArray> = mutableMapOf() private val systemFiles: MutableMap<String, ByteArray> = mutableMapOf()
private val _filesUpdates = MutableSharedFlow<DataPage<IFile>>(extraBufferCapacity = 16) private val _filesUpdates = MutableSharedFlow<DataPage<IFile>>(extraBufferCapacity = 16)
@@ -84,7 +85,11 @@ class FakeStorageAccessor : IStorageAccessor {
return ByteArrayInputStream(bytes) return ByteArrayInputStream(bytes)
} }
override suspend fun moveToTrash(path: String) = Unit override suspend fun moveToTrash(path: String) {
if (path in dataFiles) {
trashedPaths.add(path)
}
}
override suspend fun openReadSystemFile(name: String): InputStream { override suspend fun openReadSystemFile(name: String): InputStream {
val bytes = systemFiles[name] ?: ByteArray(0) val bytes = systemFiles[name] ?: ByteArray(0)