fix(android): не принимать разгруженное превью за новую версию элемента

- роль локального файла определяется меткой провенанса, а не состоянием
  индекса: собственное превью выправляет запись (MarkPreview) вместо
  повторной загрузки, постер видео связывается с самим видео;
- изменённое содержимое заменяет серверную копию только после сверки
  original_checksum (ReplaceUpload): прежняя копия удаляется мягко, а при
  расхождении принятая загрузка отбрасывается и элемент не разгружается;
- файл не разгружается, пока по нему есть незавершённая загрузка;
- восстановленный оригинал помечается UPLOADED, чтобы следующий проход не
  разгрузил его снова;
- миграция 8→9 добавляет upload_jobs.replacesPhotoId.
This commit is contained in:
iu
2026-10-20 20:32:20 +03:00
parent 97f8f4e768
commit 2e3728d83d
16 changed files with 611 additions and 28 deletions
@@ -28,7 +28,7 @@ import com.galo.android.data.db.entity.UploadJobEntity
DeviceAlbumMetaEntity::class,
DeletedAssetEntity::class,
],
version = 8,
version = 9,
exportSchema = false,
)abstract class GaloDatabase : RoomDatabase() {
@@ -45,7 +45,7 @@ import com.galo.android.data.db.entity.UploadJobEntity
fun create(context: Context): GaloDatabase =
Room.databaseBuilder(context.applicationContext, GaloDatabase::class.java, NAME)
.addMigrations(MIGRATION_6_7, MIGRATION_7_8)
.addMigrations(MIGRATION_6_7, MIGRATION_7_8, MIGRATION_8_9)
.fallbackToDestructiveMigration()
.build()
@@ -83,5 +83,16 @@ import com.galo.android.data.db.entity.UploadJobEntity
db.execSQL("ALTER TABLE device_album_meta ADD COLUMN syncDisabledAt INTEGER")
}
}
/**
* Records which earlier photo version an upload replaces, so the stale
* copy is deleted only after the new version is confirmed by the server.
* Existing jobs read as "no replacement".
*/
val MIGRATION_8_9 = object : Migration(8, 9) {
override fun migrate(db: SupportSQLiteDatabase) {
db.execSQL("ALTER TABLE upload_jobs ADD COLUMN replacesPhotoId TEXT")
}
}
}
}
@@ -87,6 +87,17 @@ interface SyncDao {
@Query("UPDATE device_assets SET displayName = :name WHERE serverPhotoId = :photoId")
suspend fun updateDisplayNameByServerPhoto(photoId: String, name: String)
/**
* Records the identity the local file really has: an offloaded preview belongs
* to the original's photo and hash, replacing values that drifted (for example
* a version uploaded from the preview itself).
*/
@Query(
"UPDATE device_assets SET serverPhotoId = :serverPhotoId, contentHash = :originalHash " +
"WHERE assetKey = :assetKey",
)
suspend fun linkOriginal(assetKey: String, serverPhotoId: String, originalHash: String?)
@Query("UPDATE device_assets SET state = :state WHERE assetKey = :assetKey")
suspend fun markState(assetKey: String, state: String)
@@ -28,4 +28,7 @@ data class UploadJobEntity(
val assetKey: String? = null,
val contentHash: String? = null,
val syncMode: String? = null,
// Set when this upload replaces an earlier version of the same element: the
// stale photo is deleted only after this upload is confirmed by the server.
val replacesPhotoId: String? = null,
)
@@ -29,14 +29,17 @@ class DeviceLocalAssetProvider(
override suspend fun assets(deviceAlbumId: String): List<LocalAsset> = withContext(Dispatchers.IO) {
deviceSource.items(deviceAlbumId).map { media ->
val described = describe(media.uri, media.size, media.dateModified)
LocalAsset(
assetKey = DeviceAssetEntity.key(deviceAlbumId, media.id),
mediaStoreId = media.id,
displayName = media.displayName,
mimeType = media.mimeType,
kind = media.kind,
contentHash = hashCached(media.uri, media.size, media.dateModified),
contentHash = described.hash,
uri = media.uri,
stampedOriginalHash = described.stamp?.originalHash,
stampedAssetKey = described.stamp?.assetKey,
)
}
}
@@ -47,23 +50,34 @@ class DeviceLocalAssetProvider(
override fun openStream(uri: String): InputStream? =
runCatching { context.contentResolver.openInputStream(Uri.parse(uri)) }.getOrNull()
private data class CachedHash(val size: Long, val dateModified: Long, val hash: String?)
private data class CachedHash(
val size: Long,
val dateModified: Long,
val hash: String?,
val stamp: Provenance?,
)
// Hashing a large album every pass is expensive; MediaStore updates
// DATE_MODIFIED on any write, so an unchanged (size, dateModified) pair lets
// us reuse the previous sha256 instead of re-reading the whole file.
// us reuse the previous sha256 instead of re-reading the whole file. The
// provenance marker is read together with the bytes: it matters only when the
// file was read anyway, and it tells a preview written by this client apart
// from a new version of the original.
private val hashCache = ConcurrentHashMap<String, CachedHash>()
private fun hashCached(uri: String, size: Long, dateModified: Long): String? {
private fun describe(uri: String, size: Long, dateModified: Long): CachedHash {
val cached = hashCache[uri]
if (cached != null && cached.size == size && cached.dateModified == dateModified) {
return cached.hash
return cached
}
val hash = hash(uri)
hashCache[uri] = CachedHash(size, dateModified, hash)
return hash
val fresh = CachedHash(size, dateModified, hash(uri), stamp(uri))
hashCache[uri] = fresh
return fresh
}
private fun hash(uri: String): String? =
runCatching { openStream(uri)?.use { ContentHash.sha256(it) } }.getOrNull()
private fun stamp(uri: String): Provenance? =
runCatching { Uri.parse(uri) }.getOrNull()?.let { ProvenanceStamp.read(context, it) }
}
@@ -76,7 +76,10 @@ class RestoreManager(
}
try {
writeInPlace(Uri.parse(asset.uri), data)
dao.markState(asset.assetKey, LocalAssetState.LOCAL.name)
// The original is back on the device and the server copy exists, so
// the element is uploaded-and-local again: recording LOCAL would let
// the next move-mode pass offload it away immediately.
dao.markState(asset.assetKey, LocalAssetState.UPLOADED.name)
restored++
bytes += data.size
} catch (e: SecurityException) {
@@ -335,6 +335,9 @@ class SyncCoordinator(
for (entity in dao.getAssets(deviceAlbumId)) {
if (entity.serverPhotoId == null) continue
if (entity.assetKey !in localKeys) continue
// Never rewrite a file while an upload still has to read it: the upload
// would send the preview instead of the original.
if (uploadQueue.isPending(entity.assetKey)) continue
// Anything linked to a server photo that still has its original on the
// device (uploaded, awaiting free, or otherwise) is offloaded; only the
// already-offloaded preview/lost/server-only states are skipped.
@@ -370,6 +373,16 @@ class SyncCoordinator(
is SyncAction.Link ->
record(deviceAlbumId, action.asset, action.photoId, mode)
is SyncAction.MarkPreview -> {
// The bytes are the preview this client offloaded: record the role the
// file actually has (the index may have drifted back to an original)
// and, when the original's copy is known, its identity.
dao.markState(action.assetKey, LocalAssetState.PREVIEW.name)
if (action.serverPhotoId != null) {
dao.linkOriginal(action.assetKey, action.serverPhotoId, action.originalIdentity)
}
}
is SyncAction.Upload ->
// Skip if an upload for this asset is already queued/running, so
// overlapping passes do not duplicate it.
@@ -377,6 +390,30 @@ class SyncCoordinator(
uploadQueue.enqueue(deviceAlbumId, serverAlbumId, action.asset, mode)
}
is SyncAction.ReplaceUpload -> {
// Journal the stale copy so it is not materialized back while the
// replacement is in flight; the worker removes it once the server
// confirms the new version, and the journal entry is cleared when the
// photo is gone from the album.
dao.upsertDeletedAsset(
DeletedAssetEntity(
photoId = action.stalePhotoId,
deviceAlbumId = deviceAlbumId,
status = DeletedAssetEntity.PENDING,
createdAt = System.currentTimeMillis(),
),
)
if (!uploadQueue.isPending(action.asset.assetKey)) {
uploadQueue.enqueue(
deviceAlbumId,
serverAlbumId,
action.asset,
mode,
action.stalePhotoId,
)
}
}
is SyncAction.RenameOnServer -> {
runCatching { remote.renamePhoto(action.photoId, action.newName) }
dao.updateDisplayNameByServerPhoto(action.photoId, action.newName)
@@ -13,8 +13,29 @@ data class LocalAsset(
val kind: MediaKind,
val contentHash: String?,
val uri: String = "",
/** Hash of the original recorded in the file's own provenance marker, if any. */
val stampedOriginalHash: String? = null,
/** Asset the marker was written for; distinguishes this element's preview from another's. */
val stampedAssetKey: String? = null,
)
/**
* The element this local file is the offloaded preview of, or null when the file
* is not a preview written by this client. The provenance marker names the element:
* a photo preview names itself, while a video poster names the video it replaced —
* a different MediaStore row, because offloading a video deletes the video and
* inserts a poster. Mirrored files carry no asset key in their marker and are never
* matched here (their role comes from the recorded state instead).
*/
fun LocalAsset.offloadedPreviewTarget(): String? {
val stampedAsset = stampedAssetKey ?: return null
val stampedHash = stampedOriginalHash ?: return null
val hash = contentHash ?: return null
// The file is the very original the marker was written from.
if (hash == stampedHash) return null
return stampedAsset
}
/** What the local index remembers about a device asset. */
data class IndexedAsset(
val assetKey: String,
@@ -32,6 +53,25 @@ sealed interface SyncAction {
data class Link(val asset: LocalAsset, val photoId: String) : SyncAction
data class DeleteOnServer(val photoId: String, val assetKey: String) : SyncAction
data class RenameOnServer(val photoId: String, val newName: String) : SyncAction
/**
* Uploads a new version of an element while remembering the earlier photo it
* replaces. The stale copy is removed only once this upload is confirmed by the
* server, so a failed replacement never leaves the album without a live copy.
*/
data class ReplaceUpload(val asset: LocalAsset, val stalePhotoId: String) : SyncAction
/**
* The bytes on the device are the preview this client wrote when offloading
* the element, so the element is not holding an original. Records the role the
* file actually has instead of acting on the drifted index state; when the
* original's copy is known on the server, also repairs the recorded identity.
*/
data class MarkPreview(
val assetKey: String,
val serverPhotoId: String? = null,
val originalIdentity: String? = null,
) : SyncAction
}
/**
@@ -44,11 +84,14 @@ sealed interface SyncAction {
*
* Rules:
* - content already on the server is linked instead of uploaded (dedup);
* - an indexed asset whose content changed is re-uploaded and its old upload is
* removed;
* - an indexed asset whose content really changed is re-uploaded and its old
* upload is replaced once the new version is confirmed;
* - a preview this client wrote itself is never a content change: the recorded
* role is repaired instead (see [offloadedPreviewTarget]);
* - a rename is propagated when only the filename changed;
* - a photo is deleted on the server only when it was uploaded by this client and
* its content is gone from the device and from every other linked album.
* its content is gone from the device and from every other linked album; a video
* that was offloaded into a poster is not "gone", so it is not deleted.
*/
object SyncPlanner {
@@ -64,14 +107,33 @@ object SyncPlanner {
for (asset in local) {
val indexed = index.firstOrNull { it.assetKey == asset.assetKey }
// An offloaded asset keeps only a preview locally, so its content hash
// no longer matches the original: it must not be re-uploaded.
// An offloaded element keeps only a preview locally; the marker names the
// element it belongs to (itself for a photo, the replaced video for a
// poster), so a preview is never read as an original or as a new file.
val previewOf = asset.offloadedPreviewTarget()
if (previewOf != null) {
val target = index.firstOrNull { it.assetKey == previewOf }
val identity = asset.stampedOriginalHash
val livePhotoId = identity?.let { serverChecksums[it] }
val alreadyConsistent = target != null &&
target.state == LocalAssetState.PREVIEW.name &&
identity != null && identity == target.contentHash &&
livePhotoId != null && livePhotoId == target.serverPhotoId
if (!alreadyConsistent && target?.state != LocalAssetState.ORPHANED.name) {
actions += SyncAction.MarkPreview(
previewOf,
livePhotoId ?: target?.serverPhotoId?.takeIf { it in livePhotoIds },
identity,
)
}
continue
}
if (indexed?.state == LocalAssetState.PREVIEW.name) continue
if (indexed?.state == LocalAssetState.ORPHANED.name) {
// The server copy is gone. The offloaded preview must not be
// re-uploaded, but the original coming back to the device is a new
// upload (deduplicated against the server as usual).
val originalHash = indexed.contentHash
val originalHash = asset.stampedOriginalHash ?: indexed.contentHash
if (originalHash != null && asset.contentHash == originalHash) {
if (serverChecksums.containsKey(originalHash)) {
actions += SyncAction.Link(asset, serverChecksums.getValue(originalHash))
@@ -88,8 +150,7 @@ object SyncPlanner {
indexed.contentHash != null &&
asset.contentHash != indexed.contentHash
if (changed) {
actions += SyncAction.Upload(asset)
actions += SyncAction.DeleteOnServer(liveIndexedPhotoId, asset.assetKey)
actions += SyncAction.ReplaceUpload(asset, liveIndexedPhotoId)
} else if (asset.displayName != indexed.displayName) {
actions += SyncAction.RenameOnServer(liveIndexedPhotoId, asset.displayName)
}
@@ -104,9 +165,13 @@ object SyncPlanner {
}
val localKeys = local.mapTo(HashSet()) { it.assetKey }
// Elements still represented by a local offloaded preview (a video replaced
// by its poster) have left their MediaStore row but are not "gone".
val offloaded = local.mapNotNullTo(HashSet()) { it.offloadedPreviewTarget() }
for (indexed in index) {
val photoId = indexed.serverPhotoId ?: continue
if (indexed.assetKey in localKeys) continue
if (indexed.assetKey in offloaded) continue
// A materialized (server-origin) photo is deleted on the server only on
// an explicit in-app deletion (a tombstone). A merely missing local copy
// is not an intent to delete: the coordinator recreates it instead.
@@ -5,11 +5,16 @@ import com.galo.android.domain.model.SyncMode
/** Hands uploads to the persistent queue instead of uploading inline. */
interface SyncUploadQueue {
/**
* [replacesPhotoId] names the earlier server photo this upload supersedes: the
* worker removes it only after the server confirms the new version.
*/
suspend fun enqueue(
deviceAlbumId: String,
serverAlbumId: String,
asset: LocalAsset,
mode: SyncMode,
replacesPhotoId: String? = null,
)
/** True when this asset already has a queued/running upload. */
@@ -27,6 +32,7 @@ class RepositorySyncUploadQueue(
serverAlbumId: String,
asset: LocalAsset,
mode: SyncMode,
replacesPhotoId: String?,
) {
uploads.enqueueSync(
deviceAlbumId = deviceAlbumId,
@@ -38,6 +44,7 @@ class RepositorySyncUploadQueue(
contentHash = asset.contentHash,
mode = mode.name,
wifiOnly = wifiOnly(),
replacesPhotoId = replacesPhotoId,
)
}
@@ -90,6 +90,7 @@ class UploadRepository(
contentHash: String?,
mode: String,
wifiOnly: Boolean,
replacesPhotoId: String? = null,
) {
val job = UploadJobEntity(
id = UUID.randomUUID().toString(),
@@ -109,6 +110,7 @@ class UploadRepository(
assetKey = assetKey,
contentHash = contentHash,
syncMode = mode,
replacesPhotoId = replacesPhotoId,
)
dao.upsert(job)
UploadScheduler.enqueueDrain(context)
@@ -0,0 +1,31 @@
package com.galo.android.data.upload
/**
* Compares the content checksum the server reports for a finished upload with the
* hash this client sent.
*
* The queue records the hash computed while planning, but the bytes are read from
* the file only when the request runs. If the file changed in between — the client
* itself rewrote it, another app edited it — the two disagree, and accepting such
* an upload as the element's new version would replace a good original with
* whatever happened to be on disk.
*
* Only a *reported* checksum can confirm or contradict: a server that reports none
* leaves the upload unverified, which never makes it invalid — it only forbids
* dropping an earlier version on the strength of it.
*/
object UploadVerification {
/** Whether the job carries a content claim at all (manual uploads do not). */
fun hasClaim(sentHash: String?): Boolean = !sentHash.isNullOrBlank()
/** The server reported exactly the content this client sent. */
fun matches(sentHash: String?, acceptedChecksum: String?): Boolean =
hasClaim(sentHash) && !acceptedChecksum.isNullOrBlank() &&
sentHash.equals(acceptedChecksum, ignoreCase = true)
/** The server reported different content than the one this client sent. */
fun contradicts(sentHash: String?, acceptedChecksum: String?): Boolean =
hasClaim(sentHash) && !acceptedChecksum.isNullOrBlank() &&
!sentHash.equals(acceptedChecksum, ignoreCase = true)
}
@@ -88,8 +88,30 @@ class UploadWorker(
if (prepared.tempPath != null) tempPath = prepared.tempPath
val photo = container.api.upload(job.albumId, prepared.filename, prepared.body)
if (UploadVerification.contradicts(job.contentHash, photo.originalChecksum)) {
// The server reported storing something other than what this client
// hashed, so nothing may be trusted about it: drop the just-uploaded
// copy (soft, recoverable) and keep the element as it was, for the
// next pass with a fresh hash.
android.util.Log.w(
TAG,
"upload ${job.id}: accepted checksum differs, dropping the new copy",
)
container.photoRepository.remove(photo.id)
runCatching { container.api.deletePhoto(photo.id) }
repository.updateState(
jobId,
UploadStatus.ERROR,
job.progress,
"Сервер принял другое содержимое",
attempt,
tempPath,
)
tempPath?.let { runCatching { File(it).delete() } }
return Outcome.DONE
}
container.photoRepository.upsert(photo)
recordProvenance(container, job, photo.id)
recordProvenance(container, job, photo.id, photo.originalChecksum)
repository.updateState(jobId, UploadStatus.SUCCESS, 100, null, attempt, tempPath)
tempPath?.let { runCatching { File(it).delete() } }
Outcome.DONE
@@ -107,12 +129,21 @@ class UploadWorker(
/**
* Records the created server photo as provenance of the source asset (sync
* jobs) and, in mode C, offloads the original right after the upload.
* jobs) and, in mode C, offloads the original right after the upload. A version
* this upload supersedes is removed first: losing that step would only leave an
* extra copy, while deleting before the new identity is recorded means a crash
* in between is repaired by linking the content that is already on the server.
*/
private suspend fun recordProvenance(container: AppContainer, job: UploadJobEntity, photoId: String) {
private suspend fun recordProvenance(
container: AppContainer,
job: UploadJobEntity,
photoId: String,
acceptedChecksum: String?,
) {
val deviceAlbumId = job.deviceAlbumId ?: return
val assetKey = job.assetKey ?: return
val mode = SyncMode.from(job.syncMode)
dropSupersededCopy(container, job, acceptedChecksum)
val state = if (mode == SyncMode.MOVE) LocalAssetState.AWAITING_FREE else LocalAssetState.UPLOADED
container.syncRepository.recordUploaded(
deviceAlbumId = deviceAlbumId,
@@ -132,6 +163,23 @@ class UploadWorker(
}
}
/**
* Removes the earlier version this upload replaces, and only when the new
* version is positively confirmed by the server: a refused or unverified upload
* never leaves the album without a live copy. The removal is soft — the trash
* keeps the bytes.
*/
private suspend fun dropSupersededCopy(
container: AppContainer,
job: UploadJobEntity,
acceptedChecksum: String?,
) {
val stalePhotoId = job.replacesPhotoId ?: return
if (!UploadVerification.matches(job.contentHash, acceptedChecksum)) return
runCatching { container.api.deletePhoto(stalePhotoId) }
.onSuccess { container.photoRepository.remove(stalePhotoId) }
}
private data class PreparedBody(
val filename: String,
val body: StreamingRequestBody,
@@ -232,6 +280,7 @@ class UploadWorker(
if (total > 0) ((sent * 100) / total).toInt().coerceIn(0, 100) else 0
private companion object {
const val TAG = "UploadWorker"
const val MAX_ATTEMPTS = 3
const val NOTIFICATION_ID = 42
}