laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-08-09 08:04:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: harden Blossom sync concurrency and cancellation

Browse Source 
Audit fixes for the Blossom client:

- BlossomBlobManagerViewModel: use StateFlow.update{} for the presence
  matrix so the Main-thread sync collector and IO-thread delete/mirror
  actions can't lose each other's writes; add refreshJob de-dup so two
  quick refreshes can't interleave; rethrow CancellationException; bound
  the /list HEAD-probe backfill with a Semaphore(8).
- BlossomClient.has(): rethrow CancellationException instead of
  swallowing it as 'not found'.
- BlossomSyncForegroundService: drop the stale 'running' de-dup guard so
  a fresh sweep always gets foreground protection.
- CLI mirror: strip query/fragment before extracting the sha256.
- DisplayBlossomSyncProgress: retain the last state so the slide-out exit
  animation still has content to draw after the state clears.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Ckbnz1N94W1hnNC9xpsCNP
This commit is contained in:
Claude
2026-07-18 13:46:06 +00:00
parent 84640a5890
commit a25c5f9518
5 changed files with 70 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
amethyst/src/main/java/com/vitorpamplona/amethyst/service/uploads/blossom/BlossomSyncForegroundService.kt
+6 -15
View File
@@ -46,16 +46,6 @@ class BlossomSyncForegroundService : FlowProgressForegroundService<BlossomSyncSt
override val cancelAction = ACTION_CANCEL
override val cancelLabelRes = R.string.blossom_sync_cancel
override fun onCreate() {
super.onCreate()
running = true
}
override fun onDestroy() {
running = false
super.onDestroy()
}
override fun state() = Amethyst.instance.blossomMirrorQueue.state
override fun isActive(value: BlossomSyncState?) = value?.running == true
@@ -81,12 +71,13 @@ class BlossomSyncForegroundService : FlowProgressForegroundService<BlossomSyncSt
private const val NOTIFICATION_ID = 0x424C4F // "BLO"
private const val ACTION_CANCEL = "com.vitorpamplona.amethyst.blossom.SYNC_CANCEL"
@Volatile
private var running = false
/** Started when a sweep begins (from the foreground); stops itself when it finishes. */
/**
* Started when a sweep begins (from the foreground); stops itself when it finishes.
* No `running` de-dup: a sweep starts this exactly once (via [BlossomMirrorQueue.onActive]),
* and a redundant start would just route to a cheap onStartCommand — whereas a stale
* "already running" flag could leave a fresh sweep with no foreground protection.
*/
fun start(context: Context) {
if (running) return
FlowProgressForegroundService.start(context, BlossomSyncForegroundService::class.java, TAG)
}
}
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/actions/mediaServers/BlossomBlobManagerViewModel.kt
+44 -20
View File
@@ -40,13 +40,18 @@ import com.vitorpamplona.quartz.nipB7Blossom.BlossomServerUrl
import com.vitorpamplona.quartz.nipB7Blossom.BlossomUploadResult
import com.vitorpamplona.quartz.utils.Log
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.update
import kotlinx.coroutines.launch
import kotlinx.coroutines.sync.Semaphore
import kotlinx.coroutines.sync.withPermit
import kotlinx.coroutines.withContext
import kotlin.coroutines.cancellation.CancellationException
/** Whether one of the user's servers holds a blob, or an operation on it is in flight. */
enum class PresenceState {
@@ -147,19 +152,27 @@ class BlossomBlobManagerViewModel : ViewModel() {
account.blossomServers.flow.value
.distinct()
private var refreshJob: Job? = null
fun refresh() {
viewModelScope.launch(Dispatchers.IO) {
_isLoading.value = true
_error.value = null
try {
loadMatrix()
} catch (e: Exception) {
Log.w("BlossomBlobManager", "Failed to load blob list", e)
_error.value = e.message?.ifBlank { null } ?: e.javaClass.simpleName
} finally {
_isLoading.value = false
// Cancel any in-flight load so two quick refreshes can't interleave their writes
// or fight over _isLoading.
refreshJob?.cancel()
refreshJob =
viewModelScope.launch(Dispatchers.IO) {
_isLoading.value = true
_error.value = null
try {
loadMatrix()
} catch (e: CancellationException) {
throw e
} catch (e: Exception) {
Log.w("BlossomBlobManager", "Failed to load blob list", e)
_error.value = e.message?.ifBlank { null } ?: e.javaClass.simpleName
} finally {
_isLoading.value = false
}
}
}
}
private suspend fun loadMatrix() {
@@ -225,15 +238,18 @@ class BlossomBlobManagerViewModel : ViewModel() {
_blobs.value = rows()
_isLoading.value = false
// Phase 2 — HEAD-probe ONLY the non-list servers, for the known hashes, in parallel.
// Phase 2 — HEAD-probe ONLY the non-list servers, for the known hashes. Bounded so a
// user with hundreds of blobs on a non-/list server doesn't spawn hundreds of probes
// at once; OkHttp still caps per-host, this caps the coroutine/allocation breadth.
val nonListServers = servers.filter { it !in listCapable }
if (nonListServers.isNotEmpty() && allHashes.isNotEmpty()) {
val limiter = Semaphore(MAX_HEAD_PROBES)
val head =
coroutineScope {
nonListServers
.flatMap { server ->
allHashes.map { hash ->
async { (server to hash) to clientFor(server).has(hash, server) }
async { limiter.withPermit { (server to hash) to clientFor(server).has(hash, server) } }
}
}.awaitAll()
}.toMap()
@@ -248,14 +264,18 @@ class BlossomBlobManagerViewModel : ViewModel() {
server: String,
state: PresenceState,
) {
_blobs.value =
_blobs.value.map { row ->
// Atomic read-modify-write: the app-level sync results collector (Main) and the
// per-row delete/mirror actions (IO) both mutate _blobs concurrently, so a plain
// `_blobs.value = _blobs.value.map{}` would lose updates.
_blobs.update { list ->
list.map { row ->
if (row.hash != hash) {
row
} else {
row.copy(servers = row.servers.map { if (it.server == server) it.copy(state = state) else it })
}
}
}
}
/** BUD-02 delete: remove [hash] from a single [server]; the pill spins then goes grey. */
@@ -275,9 +295,7 @@ class BlossomBlobManagerViewModel : ViewModel() {
}
setServerState(hash, server, if (ok) PresenceState.MISSING else PresenceState.PRESENT)
// Drop the row entirely once it's gone from every server.
if (currentRow(hash)?.hasPresent == false) {
_blobs.value = _blobs.value.filter { it.hash != hash }
}
_blobs.update { list -> if (list.firstOrNull { it.hash == hash }?.hasPresent == false) list.filter { it.hash != hash } else list }
}
}
@@ -322,14 +340,15 @@ class BlossomBlobManagerViewModel : ViewModel() {
.map { BlossomMirrorQueue.Task(it.hash, it.url!!, it.size, it.missingServers) }
if (tasks.isEmpty()) return
_blobs.value =
_blobs.value.map { row ->
_blobs.update { list ->
list.map { row ->
if (!row.hasMissing) {
row
} else {
row.copy(servers = row.servers.map { if (it.state == PresenceState.MISSING) it.copy(state = PresenceState.PENDING) else it })
}
}
}
Amethyst.instance.blossomMirrorQueue.start(account, tasks)
}
@@ -400,4 +419,9 @@ class BlossomBlobManagerViewModel : ViewModel() {
val size: Long?,
val type: String?,
)
companion object {
/** Cap on concurrent HEAD probes during the /list backfill. */
private const val MAX_HEAD_PROBES = 8
}
}
amethyst/src/main/java/com/vitorpamplona/amethyst/ui/actions/mediaServers/DisplayBlossomSyncProgress.kt
+11 -1
View File
@@ -41,7 +41,11 @@ import androidx.compose.material3.MaterialTheme
import androidx.compose.material3.Surface
import androidx.compose.material3.Text
import androidx.compose.runtime.Composable
import androidx.compose.runtime.LaunchedEffect
import androidx.compose.runtime.getValue
import androidx.compose.runtime.mutableStateOf
import androidx.compose.runtime.remember
import androidx.compose.runtime.setValue
import androidx.compose.ui.Alignment
import androidx.compose.ui.Modifier
import androidx.compose.ui.text.style.TextOverflow
@@ -51,6 +55,7 @@ import com.vitorpamplona.amethyst.Amethyst
import com.vitorpamplona.amethyst.R
import com.vitorpamplona.amethyst.commons.icons.symbols.Icon
import com.vitorpamplona.amethyst.commons.icons.symbols.MaterialSymbols
import com.vitorpamplona.amethyst.service.uploads.blossom.BlossomSyncState
import com.vitorpamplona.amethyst.ui.stringRes
import com.vitorpamplona.amethyst.ui.theme.allGoodColor
import com.vitorpamplona.amethyst.ui.theme.grayText
@@ -65,6 +70,11 @@ fun DisplayBlossomSyncProgress() {
val queue = Amethyst.instance.blossomMirrorQueue
val state by queue.state.collectAsStateWithLifecycle()
// Retain the last non-null value so the slide-out exit still has content to draw when
// state clears to null on cancel/dismiss.
var lastShown by remember { mutableStateOf<BlossomSyncState?>(null) }
LaunchedEffect(state) { state?.let { lastShown = it } }
Box(modifier = Modifier.fillMaxSize(), contentAlignment = Alignment.BottomCenter) {
AnimatedVisibility(
visible = state != null,
@@ -76,7 +86,7 @@ fun DisplayBlossomSyncProgress() {
.padding(start = 12.dp, end = 12.dp, bottom = 116.dp)
.widthIn(max = 560.dp),
) {
val s = state ?: return@AnimatedVisibility
val s = lastShown ?: return@AnimatedVisibility
Surface(
shape = RoundedCornerShape(18.dp),
color = MaterialTheme.colorScheme.surfaceContainer,
cli/src/main/kotlin/com/vitorpamplona/amethyst/cli/commands/BlossomCommands.kt
+6 -1
View File
@@ -116,7 +116,12 @@ object BlossomCommands {
val args = Args(rest)
val server = args.flag("server") ?: return Output.error("bad_args", "blossom mirror requires --server URL")
val sourceUrl = args.positional(0, "source-url")
val hash = sourceUrl.substringAfterLast('/').substringBefore('.')
val hash =
sourceUrl
.substringBefore('?')
.substringBefore('#')
.substringAfterLast('/')
.substringBefore('.')
if (hash.length != 64 || hash.any { it !in "0123456789abcdef" }) {
return Output.error("bad_args", "could not extract a sha256 from the source url '$sourceUrl'")
}
commons/src/jvmAndroid/kotlin/com/vitorpamplona/amethyst/commons/service/upload/BlossomClient.kt
+3
View File
@@ -37,6 +37,7 @@ import okhttp3.Response
import okio.BufferedSink
import okio.source
import java.io.File
import kotlin.coroutines.cancellation.CancellationException
/**
* Thrown when a Blossom server answers with `402 Payment Required` (BUD-07). The
@@ -198,6 +199,8 @@ open class BlossomClient(
.build()
try {
okHttpClient.newCall(request).execute().use { it.isSuccessful }
} catch (e: CancellationException) {
throw e
} catch (_: Exception) {
false
}