Preserve file wake lock ownership and isolate cleanup errors
This commit is contained in:
1 parent
4f06b4bca9
commit
5290a2868d
2 files changed
+76
-11
No files matched your search
+66
-11
@@ -14,16 +14,35 @@ import kotlinx.coroutines.flow.StateFlow
|
||||
import java.io.File
|
||||
import java.io.IOException
|
||||
|
||||
/** Адрес каталога: u64-ID группы/хранителя/владельца и относительный UTF-8 path.
|
||||
* owner=0 — виртуальный корень группы, допустим только с пустым path. */
|
||||
data class FileLocation(val group: Long, val server: Long, val owner: Long, val path: String)
|
||||
/** Состояние одной очереди/экспорта: token — локальная операция, index/count — текущий файл/длина очереди.
|
||||
* done/total — байты прогресса native, stage — ID строки UI, status — MF-ошибка либо null.
|
||||
* inactive не гарантирует успех: дополнительно проверять status/cancelled; uploaded/skipped относятся к текущему запуску очереди. */
|
||||
data class FileOperation(val token: Long, val location: FileLocation, val name: String = "", val index: Int = 1,
|
||||
val count: Int = 1, val done: Long = 0, val total: Long = 0, val stage: Int = R.string.files_preparing,
|
||||
val status: Int? = null, val active: Boolean = true, val cancelled: Boolean = false, val retryable: Boolean = true,
|
||||
val upload: Boolean = false, val uploaded: Int = 0, val skipped: Int = 0)
|
||||
/** Конфликт загрузки: true — заменить, false — пропустить, null — отменить оставшуюся очередь. */
|
||||
data class FileConflict(val token: Long, val name: String, val decision: CompletableDeferred<Boolean?>)
|
||||
/** Готовая локальная копия для OPEN/SAVE/SHARE; location — исходный родительский каталог, не путь файла. */
|
||||
data class FileExport(val token: Long, val location: FileLocation, val name: String, val file: File, val action: Int)
|
||||
/** Статус общего C-протокола; timeout ожидания Kotlin также переводится в MF_TIMEOUT=13. */
|
||||
class FileStatusException(val status: Int) : IOException("member_files status=$status")
|
||||
|
||||
/** Владеет передачами независимо от экранов. Все состояния/ожидания изменяются только на main. */
|
||||
/**
|
||||
* App-scoped владелец файловых передач; публичные вызовы и обработка событий — main thread.
|
||||
* Одновременно выполняется одна upload-очередь, download или сохранение URI. Непотреблённый export
|
||||
* также блокирует следующий запуск; отдельные request для LIST/QUOTA/мутаций живут независимо.
|
||||
* AppEventHandler передаёт события 49/50 напрямую без UI replay; контекст ответа сверяется с ожидающим ID.
|
||||
*
|
||||
* URI копируется на Dispatchers.IO в стабильный upload-source. При отмене transfer ждёт terminal
|
||||
* native-событие (закрытый handle) до удаления своего source/destination; остановка core завершает ожидания.
|
||||
* Работа удерживает wake lock и CoreActivity.transfer, ожидание решения о замене снимает оба удержания.
|
||||
* Ошибки cleanup логируются независимо; неосвобождённый wake lock сохраняется для следующей попытки.
|
||||
* Скачанная копия остаётся в cacheDir/member_files для внешнего приложения/повторного SAVE.
|
||||
*/
|
||||
class MemberFilesRepository(context: Context) {
|
||||
companion object { const val OPEN = 0; const val SAVE = 1; const val SHARE = 2 }
|
||||
private val app = context.applicationContext
|
||||
@@ -44,10 +63,14 @@ class MemberFilesRepository(context: Context) {
|
||||
private val _revision = MutableStateFlow(0L)
|
||||
val revision: StateFlow<Long> = _revision
|
||||
private fun log(level: String, message: String) = LogManager.addLog(level, "Files", message)
|
||||
private fun main() = check(Looper.myLooper() == Looper.getMainLooper())
|
||||
private fun main() = check(Looper.myLooper() == Looper.getMainLooper()) {
|
||||
val message = "MemberFilesRepository requires main thread: caller=${Thread.currentThread().name}"
|
||||
log("ERROR", message)
|
||||
message
|
||||
}
|
||||
|
||||
init {
|
||||
// URI для внешнего приложения сохраняется сутки; старые экспорты и брошенные источники очищаются при запуске.
|
||||
// При создании репозитория удаляются экспорты/источники старше суток; периодического TTL-таймера здесь нет.
|
||||
scope.launch(Dispatchers.IO) {
|
||||
try {
|
||||
val root = File(app.cacheDir, "member_files")
|
||||
@@ -61,15 +84,29 @@ class MemberFilesRepository(context: Context) {
|
||||
}
|
||||
|
||||
private fun protect(active: Boolean) {
|
||||
if (active && wakeLock == null) {
|
||||
val manager = app.getSystemService(Context.POWER_SERVICE) as PowerManager
|
||||
wakeLock = manager.newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "utun:member_files").apply { acquire() }
|
||||
} else if (!active) {
|
||||
wakeLock?.let { if (it.isHeld) it.release() }; wakeLock = null
|
||||
if (active) {
|
||||
val lock = wakeLock ?: (app.getSystemService(Context.POWER_SERVICE) as PowerManager)
|
||||
.newWakeLock(PowerManager.PARTIAL_WAKE_LOCK, "utun:member_files").also { wakeLock = it }
|
||||
if (!lock.isHeld) {
|
||||
lock.acquire()
|
||||
log("DEBUG", "wake lock acquired token=${_operation.value?.token}")
|
||||
}
|
||||
} else {
|
||||
try {
|
||||
wakeLock?.let { if (it.isHeld) it.release() }
|
||||
wakeLock = null
|
||||
} catch (e: Exception) {
|
||||
log("ERROR", "wake lock release failed token=${_operation.value?.token}; retained for retry: $e")
|
||||
}
|
||||
}
|
||||
try { CoreActivity.transfer(active) }
|
||||
catch (e: Exception) {
|
||||
if (active) throw e
|
||||
log("ERROR", "activity release failed token=${_operation.value?.token}: $e")
|
||||
}
|
||||
CoreActivity.transfer(active)
|
||||
}
|
||||
|
||||
/** Main: ответ 49, прогресс/завершение 50 или остановка 13; чужой ID игнорируется, неверный контекст логируется. */
|
||||
fun onEvent(type: Int, bytes: ByteArray?) {
|
||||
main()
|
||||
if (type == 13) { coreStopped(); return }
|
||||
@@ -94,6 +131,9 @@ class MemberFilesRepository(context: Context) {
|
||||
} catch (e: Exception) { log("ERROR", "Rejected event type=$type bytes=${bytes.size}: $e") }
|
||||
}
|
||||
|
||||
/** Main suspend: одна команда с новым ID и ожиданием до 60 с; ненулевой MF-status возвращается без исключения.
|
||||
* Ошибка enqueue бросает IOException, timeout — FileStatusException(13), отмена — CancellationException.
|
||||
* Незавершённый native-запрос отменяется; уже применённую серверную мутацию отмена не откатывает. */
|
||||
suspend fun request(location: FileLocation, command: Int, target: String = "", cursor: Long = 0,
|
||||
version: ByteArray = ByteArray(16)): MemberFilesProtocol.Reply {
|
||||
main()
|
||||
@@ -103,7 +143,10 @@ class MemberFilesRepository(context: Context) {
|
||||
try {
|
||||
log("DEBUG", "request id=$id command=$command location=$location cursor=$cursor")
|
||||
val wire = MemberFilesProtocol.request(id, location.owner, command, location.path, target, cursor, version)
|
||||
if (!NativeLib.fileRequest(location.group, location.server, wire)) throw IOException("request enqueue failed id=$id")
|
||||
if (!NativeLib.fileRequest(location.group, location.server, wire)) {
|
||||
log("ERROR", "request enqueue failed id=$id command=$command location=$location")
|
||||
throw IOException("request enqueue failed id=$id")
|
||||
}
|
||||
return withTimeout(60_000) { result.await() }.also { finished = true }
|
||||
} catch (e: TimeoutCancellationException) {
|
||||
log("WARN", "request timed out id=$id command=$command location=$location")
|
||||
@@ -114,6 +157,7 @@ class MemberFilesRepository(context: Context) {
|
||||
}
|
||||
}
|
||||
|
||||
/** Вернуть успешный reply либо бросить FileStatusException с его MF-status; контекст ошибки логирует caller. */
|
||||
fun check(reply: MemberFilesProtocol.Reply): MemberFilesProtocol.Reply {
|
||||
if (reply.status != 0) throw FileStatusException(reply.status)
|
||||
return reply
|
||||
@@ -175,11 +219,14 @@ class MemberFilesRepository(context: Context) {
|
||||
} finally {
|
||||
_conflict.value = null
|
||||
protect(false)
|
||||
log("DEBUG", "operation released token=$token wake=false")
|
||||
log("DEBUG", "operation cleanup token=$token wakePending=${wakeLock != null}")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/** Main: последовательно загрузить уникальные URI в собственный каталог вне avatars.
|
||||
* При EXISTS запрашивается решение через conflict; замена использует версию STAT.
|
||||
* Retry начинает с текущего неуспешного URI, уже опубликованные/пропущенные элементы не повторяются. */
|
||||
fun upload(location: FileLocation, uris: List<Uri>) {
|
||||
if (uris.isEmpty()) return
|
||||
val sources = uris.distinct()
|
||||
@@ -249,6 +296,8 @@ class MemberFilesRepository(context: Context) {
|
||||
}
|
||||
private fun remove(file: File) { if (file.exists() && !file.delete()) log("WARN", "Cannot remove temporary file=$file") }
|
||||
|
||||
/** Main: скачать файл из родительского location по версии entry в отдельный каталог cacheDir.
|
||||
* action=OPEN/SAVE/SHARE задаёт намерение UI; успешный native finished создаёт export, не запускает Android Intent. */
|
||||
fun download(location: FileLocation, entry: MemberFilesProtocol.Entry, action: Int) {
|
||||
launch(location, 1) { token ->
|
||||
publish(FileOperation(token, location, entry.name))
|
||||
@@ -270,16 +319,19 @@ class MemberFilesRepository(context: Context) {
|
||||
}
|
||||
}
|
||||
|
||||
/** Снять совпадающий export с ожидания UI, сохранив его файл; caller может удержать FileExport для SAVE. */
|
||||
fun consumeExport(token: Long) { if (_export.value?.token == token) _export.value = null }
|
||||
fun decideReplace(token: Long, replace: Boolean?) {
|
||||
_conflict.value?.takeIf { it.token == token }?.decision?.complete(replace)
|
||||
}
|
||||
/** Main: отменить текущую очередь/экспорт; запросы каталога отменяет их вызывающая корутина. Cleanup асинхронный. */
|
||||
fun cancel() {
|
||||
main()
|
||||
log("INFO", "cancel requested token=${_operation.value?.token}")
|
||||
task?.cancel()
|
||||
}
|
||||
fun retry() { if (_operation.value?.active == false) retry?.invoke() }
|
||||
/** Main после остановки core: отменить работу и разблокировать все pending/transfer ожидания исключением. */
|
||||
fun coreStopped() {
|
||||
main(); cancel()
|
||||
val error = CancellationException("core stopped")
|
||||
@@ -289,6 +341,9 @@ class MemberFilesRepository(context: Context) {
|
||||
}
|
||||
fun dismiss() { if (_operation.value?.active == false) _operation.value = null; HeadlessService.refreshFiles() }
|
||||
|
||||
/** Main: скопировать готовый export в URI CreateDocument, после consumeExport и при свободном репозитории.
|
||||
* При ошибке/отмене пытается удалить неполный документ provider; исходная копия сохраняется.
|
||||
* Эта операция не имеет автоматического Retry: новый SAVE требует нового URI от UI. */
|
||||
fun save(export: FileExport, uri: Uri) {
|
||||
launch(export.location, 1, retryable = false) { token ->
|
||||
publish(FileOperation(token, export.location, export.name, stage = R.string.files_saving, retryable = false))
|
||||
|
||||
+10
@@ -71,6 +71,16 @@ class MemberFilesFailureTest {
|
||||
putLong(0); putLong(0); putShort(status.toShort()); put(2)
|
||||
}.array()
|
||||
|
||||
@Test fun backgroundCallerIsRejectedWithADiagnostic() {
|
||||
val error = java.util.concurrent.atomic.AtomicReference<Throwable?>()
|
||||
val worker = Thread({ try { repository.cancel() } catch (e: Throwable) { error.set(e) } }, "files-test-worker")
|
||||
worker.start(); worker.join(2_000)
|
||||
assertFalse(worker.isAlive)
|
||||
assertTrue(error.get() is IllegalStateException)
|
||||
diagnostic("requires main thread: caller=files-test-worker")
|
||||
assertNull(repository.operation.value)
|
||||
}
|
||||
|
||||
@Test fun rejectedRequestAndTimeoutAreExplicitAndCancelNativeWaiter() = runTest(dispatcher) {
|
||||
NativeLib.fileRequest = { _, _, _ -> false }
|
||||
assertTrue(runCatching { repository.request(location, MemberFilesProtocol.LIST) }.exceptionOrNull() is IOException)
|
||||
|
||||
Reference in new issue
Block a user