mirror of
https://github.com/vitorpamplona/amethyst.git
synced 2026-10-05 19:28:25 +00:00
feat(blossom): fall back to upload when a sync/import target lacks /mirror
The File Sync / Import flow (and the mirror-on-upload fan-out) copy blobs across the user's Blossom servers with BUD-04 `PUT /mirror`, but not every server implements that endpoint. Blossom has no capability-discovery mechanism, so a target without /mirror just answered 404/405/501 and the whole copy was silently counted as failed. Detect the "endpoint absent" statuses (404/405/501) as a typed BlossomMirrorUnsupportedException — distinct from a mirror the server understood but rejected (400/403/413/…) — and add BlossomClient.mirrorOrUpload, which falls back to downloading the blob and re-uploading it (PUT /upload) when mirror is unsupported. The downloaded bytes are verified against the expected sha256 before re-upload, since a Blossom server is untrusted and could substitute content, and the same t=upload auth is reused. Wire every mirror path through mirrorOrUpload: the app-level BlossomMirrorQueue (sync-all + import sweep), the blob manager's per-blob mirror (including the paid-mirror retry), and UploadOrchestrator's mirror-on-upload. Task now carries the descriptor content-type so the fallback upload preserves the MIME. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0168TWLTgrMxUR6yjjCCiBLS
This commit is contained in:
+2
-1
@@ -297,11 +297,12 @@ class UploadOrchestrator {
|
||||
|
||||
if (targets.isEmpty()) return
|
||||
|
||||
val contentType = result.type ?: "application/octet-stream"
|
||||
targets.forEach { target ->
|
||||
try {
|
||||
val auth = account.createBlossomUploadAuth(hash, result.size ?: 0L, "Mirror $hash").toAuthorizationHeader()
|
||||
BlossomClient(Amethyst.instance.roleBasedHttpClientBuilder.okHttpClientForUploads(target))
|
||||
.mirror(sourceUrl, target, auth)
|
||||
.mirrorOrUpload(sourceUrl, hash, contentType, target, auth)
|
||||
} catch (e: Exception) {
|
||||
if (e is CancellationException) throw e
|
||||
Log.w("UploadOrchestrator", "Failed to mirror $hash to $target", e)
|
||||
|
||||
+8
-1
@@ -76,6 +76,8 @@ class BlossomMirrorQueue(
|
||||
val hash: HexKey,
|
||||
val sourceUrl: String,
|
||||
val size: Long?,
|
||||
/** Descriptor MIME, used when a target lacks `/mirror` and we fall back to `/upload`. */
|
||||
val contentType: String?,
|
||||
val targets: List<String>,
|
||||
)
|
||||
|
||||
@@ -137,7 +139,7 @@ class BlossomMirrorQueue(
|
||||
try {
|
||||
val auth = account.createBlossomUploadAuth(task.hash, task.size ?: 0L, "Mirror ${task.hash}").toAuthorizationHeader()
|
||||
BlossomClient(Amethyst.instance.roleBasedHttpClientBuilder.okHttpClientForUploads(target))
|
||||
.mirror(task.sourceUrl, target, auth)
|
||||
.mirrorOrUpload(task.sourceUrl, task.hash, task.contentType ?: DEFAULT_MIME_TYPE, target, auth)
|
||||
true
|
||||
} catch (e: CancellationException) {
|
||||
throw e
|
||||
@@ -157,4 +159,9 @@ class BlossomMirrorQueue(
|
||||
fun dismiss() {
|
||||
if (!isRunning) _state.value = null
|
||||
}
|
||||
|
||||
companion object {
|
||||
/** Fallback MIME for a `/upload` when the source descriptor carried no type. */
|
||||
private const val DEFAULT_MIME_TYPE = "application/octet-stream"
|
||||
}
|
||||
}
|
||||
|
||||
+9
-4
@@ -325,7 +325,7 @@ class BlossomBlobManagerViewModel : ViewModel() {
|
||||
for (target in targets) {
|
||||
setServerState(row.hash, target, PresenceState.PENDING)
|
||||
try {
|
||||
mirrorOne(source, row.hash, row.size, target, null)
|
||||
mirrorOne(source, row.hash, row.size, row.type, target, null)
|
||||
setServerState(row.hash, target, PresenceState.PRESENT)
|
||||
} catch (e: BlossomPaymentException) {
|
||||
setServerState(row.hash, target, PresenceState.MISSING)
|
||||
@@ -362,7 +362,7 @@ class BlossomBlobManagerViewModel : ViewModel() {
|
||||
val tasks =
|
||||
_blobs.value
|
||||
.filter { it.hasMissing && it.url != null }
|
||||
.map { BlossomMirrorQueue.Task(it.hash, it.url!!, it.size, it.missingServers) }
|
||||
.map { BlossomMirrorQueue.Task(it.hash, it.url!!, it.size, it.type, it.missingServers) }
|
||||
if (tasks.isEmpty()) return
|
||||
|
||||
_blobs.update { list ->
|
||||
@@ -381,11 +381,12 @@ class BlossomBlobManagerViewModel : ViewModel() {
|
||||
source: String,
|
||||
hash: HexKey,
|
||||
size: Long?,
|
||||
contentType: String?,
|
||||
target: String,
|
||||
proof: BlossomPaymentProof?,
|
||||
) {
|
||||
val auth = account.createBlossomUploadAuth(hash, size ?: 0L, "Mirror $hash").toAuthorizationHeader()
|
||||
clientFor(target).mirror(source, target, auth, proof)
|
||||
clientFor(target).mirrorOrUpload(source, hash, contentType ?: DEFAULT_MIME_TYPE, target, auth, proof)
|
||||
}
|
||||
|
||||
/** User confirmed the BUD-07 prompt: pay via the wallet, retry that server, then continue. */
|
||||
@@ -420,7 +421,8 @@ class BlossomBlobManagerViewModel : ViewModel() {
|
||||
}
|
||||
}
|
||||
try {
|
||||
mirrorOne(pending.sourceUrl, pending.hash, currentRow(pending.hash)?.size, pending.target, proof)
|
||||
val row = currentRow(pending.hash)
|
||||
mirrorOne(pending.sourceUrl, pending.hash, row?.size, row?.type, pending.target, proof)
|
||||
setServerState(pending.hash, pending.target, PresenceState.PRESENT)
|
||||
} catch (e: Exception) {
|
||||
setServerState(pending.hash, pending.target, PresenceState.MISSING)
|
||||
@@ -468,5 +470,8 @@ class BlossomBlobManagerViewModel : ViewModel() {
|
||||
companion object {
|
||||
/** Cap on concurrent HEAD probes during the /list backfill. */
|
||||
private const val MAX_HEAD_PROBES = 8
|
||||
|
||||
/** Fallback MIME for a `/upload` when a target lacks `/mirror` and the row has no type. */
|
||||
private const val DEFAULT_MIME_TYPE = "application/octet-stream"
|
||||
}
|
||||
}
|
||||
|
||||
+1
-1
@@ -377,7 +377,7 @@ class BlossomImportViewModel : ViewModel() {
|
||||
*/
|
||||
fun importSelected(): ImportStart {
|
||||
val candidates = _candidates.value
|
||||
val tasks = candidates.map { BlossomMirrorQueue.Task(it.hash, it.sourceUrl, it.size, it.missingTargets) }
|
||||
val tasks = candidates.map { BlossomMirrorQueue.Task(it.hash, it.sourceUrl, it.size, it.type, it.missingTargets) }
|
||||
if (tasks.isEmpty()) return ImportStart.Empty
|
||||
// start() itself atomically no-ops if a sweep is already running, so key off its return
|
||||
// rather than a separate isRunning check that could race with a sweep starting.
|
||||
|
||||
+67
-1
@@ -22,10 +22,12 @@ package com.vitorpamplona.amethyst.commons.service.upload
|
||||
|
||||
import com.vitorpamplona.quartz.nip01Core.core.HexKey
|
||||
import com.vitorpamplona.quartz.nip01Core.core.JsonMapper
|
||||
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
|
||||
import com.vitorpamplona.quartz.nipB7Blossom.BlossomPaymentProof
|
||||
import com.vitorpamplona.quartz.nipB7Blossom.BlossomPaymentRequired
|
||||
import com.vitorpamplona.quartz.nipB7Blossom.BlossomServerUrl
|
||||
import com.vitorpamplona.quartz.nipB7Blossom.BlossomUploadResult
|
||||
import com.vitorpamplona.quartz.utils.sha256.sha256
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.withContext
|
||||
import okhttp3.MediaType.Companion.toMediaType
|
||||
@@ -49,6 +51,20 @@ class BlossomPaymentException(
|
||||
val payment: BlossomPaymentRequired,
|
||||
) : RuntimeException("Payment required by $server: ${payment.reason ?: "402 Payment Required"}")
|
||||
|
||||
/**
|
||||
* Thrown by [BlossomClient.mirror] when a server does not implement the BUD-04
|
||||
* `/mirror` endpoint. Blossom has no capability-discovery mechanism (BUD-04 defines
|
||||
* none), so the only reliable signal is the status of the `PUT /mirror` itself:
|
||||
* `404 Not Found`, `405 Method Not Allowed`, or `501 Not Implemented` mean the
|
||||
* endpoint is absent — as opposed to a mirror the server understood but refused
|
||||
* (`400`/`403`/`413`/…, which stay a plain [RuntimeException]). Callers can catch
|
||||
* this to fall back to a direct download-and-upload (see [BlossomClient.mirrorOrUpload]).
|
||||
*/
|
||||
class BlossomMirrorUnsupportedException(
|
||||
val server: String,
|
||||
val status: Int,
|
||||
) : RuntimeException("$server does not support the /mirror endpoint (HTTP $status)")
|
||||
|
||||
/** Result of a BUD-06 `HEAD /upload` or `HEAD /media` preflight. */
|
||||
data class BlossomPreflightResult(
|
||||
val accepted: Boolean,
|
||||
@@ -111,6 +127,10 @@ open class BlossomClient(
|
||||
* BUD-04 mirror: ask [serverBaseUrl] to fetch and store the blob already at
|
||||
* [sourceUrl]. The server verifies the downloaded bytes hash to the `x` tag in
|
||||
* the (upload) auth token. Returns the mirrored blob's descriptor.
|
||||
*
|
||||
* Throws [BlossomMirrorUnsupportedException] when the server has no `/mirror`
|
||||
* endpoint (HTTP 404/405/501) so the caller can tell "can't mirror here" apart
|
||||
* from "mirror failed" — see [mirrorOrUpload] for the download-and-upload fallback.
|
||||
*/
|
||||
open suspend fun mirror(
|
||||
sourceUrl: String,
|
||||
@@ -129,7 +149,45 @@ open class BlossomClient(
|
||||
paymentProof?.headers()?.forEach { (name, value) -> addHeader(name, value) }
|
||||
}.put(body)
|
||||
.build()
|
||||
okHttpClient.newCall(request).execute().use { parseDescriptor(it, serverBaseUrl) }
|
||||
okHttpClient.newCall(request).execute().use { response ->
|
||||
if (response.code in MIRROR_UNSUPPORTED_CODES) {
|
||||
throw BlossomMirrorUnsupportedException(serverBaseUrl, response.code)
|
||||
}
|
||||
parseDescriptor(response, serverBaseUrl)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Mirror [sourceUrl] to [serverBaseUrl], falling back to a direct upload when the
|
||||
* server doesn't implement BUD-04 `/mirror`. First tries [mirror]; if that reports
|
||||
* [BlossomMirrorUnsupportedException], the blob is downloaded here, verified against
|
||||
* [expectedHash], then re-uploaded with [BlossomServerUrl.UPLOAD_PATH] — so a file
|
||||
* still lands on servers of every capability. Payment (`402`) and every other
|
||||
* failure propagate unchanged from [mirror].
|
||||
*
|
||||
* The downloaded bytes are hash-checked before re-upload: a Blossom server is
|
||||
* untrusted and could return substituted content, and the same `t=upload` auth
|
||||
* (whose `x` tag is the expected hash) is reused for the fallback `PUT /upload`.
|
||||
*/
|
||||
open suspend fun mirrorOrUpload(
|
||||
sourceUrl: String,
|
||||
expectedHash: HexKey,
|
||||
contentType: String,
|
||||
serverBaseUrl: String,
|
||||
authHeader: String?,
|
||||
paymentProof: BlossomPaymentProof? = null,
|
||||
): BlossomUploadResult =
|
||||
try {
|
||||
mirror(sourceUrl, serverBaseUrl, authHeader, paymentProof)
|
||||
} catch (e: BlossomMirrorUnsupportedException) {
|
||||
val bytes =
|
||||
download(sourceUrl)
|
||||
?: throw RuntimeException("Could not download $sourceUrl to upload to $serverBaseUrl")
|
||||
val actualHash = sha256(bytes).toHexKey()
|
||||
if (actualHash != expectedHash) {
|
||||
throw RuntimeException("$sourceUrl returned content ($actualHash) that does not match the expected $expectedHash")
|
||||
}
|
||||
upload(bytes, contentType, serverBaseUrl, authHeader)
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -340,4 +398,12 @@ open class BlossomClient(
|
||||
private data class MirrorRequest(
|
||||
val url: String,
|
||||
)
|
||||
|
||||
companion object {
|
||||
/**
|
||||
* `PUT /mirror` statuses that mean the endpoint isn't implemented (so the caller
|
||||
* should fall back to a direct upload) rather than a mirror the server rejected.
|
||||
*/
|
||||
private val MIRROR_UNSUPPORTED_CODES = setOf(404, 405, 501)
|
||||
}
|
||||
}
|
||||
|
||||
+167
@@ -21,21 +21,27 @@
|
||||
package com.vitorpamplona.amethyst.desktop.service.upload
|
||||
|
||||
import com.vitorpamplona.amethyst.commons.service.upload.BlossomClient
|
||||
import com.vitorpamplona.amethyst.commons.service.upload.BlossomMirrorUnsupportedException
|
||||
import com.vitorpamplona.quartz.nip01Core.core.toHexKey
|
||||
import com.vitorpamplona.quartz.utils.sha256.sha256
|
||||
import io.mockk.every
|
||||
import io.mockk.mockk
|
||||
import io.mockk.slot
|
||||
import kotlinx.coroutines.test.runTest
|
||||
import okhttp3.Call
|
||||
import okhttp3.Headers
|
||||
import okhttp3.MediaType.Companion.toMediaType
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Protocol
|
||||
import okhttp3.Request
|
||||
import okhttp3.Response
|
||||
import okhttp3.ResponseBody
|
||||
import okhttp3.ResponseBody.Companion.toResponseBody
|
||||
import java.io.File
|
||||
import kotlin.test.Test
|
||||
import kotlin.test.assertEquals
|
||||
import kotlin.test.assertFailsWith
|
||||
import kotlin.test.assertFalse
|
||||
import kotlin.test.assertTrue
|
||||
|
||||
class BlossomClientTest {
|
||||
@@ -63,6 +69,167 @@ class BlossomClientTest {
|
||||
return mockClient
|
||||
}
|
||||
|
||||
/** An OkHttpClient whose every call is answered by [handler], keyed off the request. */
|
||||
private fun dispatchingOkHttp(handler: (Request) -> Response): OkHttpClient {
|
||||
val mockClient = mockk<OkHttpClient>()
|
||||
every { mockClient.newCall(any()) } answers {
|
||||
val request = firstArg<Request>()
|
||||
val call = mockk<Call>()
|
||||
every { call.execute() } returns handler(request)
|
||||
call
|
||||
}
|
||||
return mockClient
|
||||
}
|
||||
|
||||
private fun response(
|
||||
request: Request,
|
||||
code: Int,
|
||||
body: ResponseBody = "".toResponseBody(),
|
||||
headers: Headers = Headers.headersOf(),
|
||||
): Response =
|
||||
Response
|
||||
.Builder()
|
||||
.request(request)
|
||||
.protocol(Protocol.HTTP_1_1)
|
||||
.code(code)
|
||||
.message(if (code in 200..299) "OK" else "Error")
|
||||
.headers(headers)
|
||||
.body(body)
|
||||
.build()
|
||||
|
||||
@Test
|
||||
fun mirrorThrowsUnsupportedOnMissingEndpoint() =
|
||||
runTest {
|
||||
// 404/405/501 on PUT /mirror means the endpoint is absent, not a rejected mirror.
|
||||
for (code in listOf(404, 405, 501)) {
|
||||
val client = BlossomClient(mockOkHttp(code))
|
||||
assertFailsWith<BlossomMirrorUnsupportedException> {
|
||||
client.mirror(
|
||||
sourceUrl = "https://source.example.com/abc",
|
||||
serverBaseUrl = "https://target.example.com",
|
||||
authHeader = "Nostr abc",
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun mirrorRejectionStaysPlainRuntimeException() =
|
||||
runTest {
|
||||
// A mirror the server understood but refused must NOT be read as "no /mirror".
|
||||
val client = BlossomClient(mockOkHttp(413, "", Headers.headersOf("X-Reason", "too big")))
|
||||
val ex =
|
||||
assertFailsWith<RuntimeException> {
|
||||
client.mirror("https://source.example.com/abc", "https://target.example.com", null)
|
||||
}
|
||||
assertFalse(ex is BlossomMirrorUnsupportedException)
|
||||
assertTrue(ex.message!!.contains("too big"))
|
||||
}
|
||||
|
||||
@Test
|
||||
fun mirrorOrUploadFallsBackToUploadWhenUnsupported() =
|
||||
runTest {
|
||||
val bytes = byteArrayOf(9, 8, 7, 6, 5)
|
||||
val expectedHash = sha256(bytes).toHexKey()
|
||||
val descriptor = """{"url":"https://target.example.com/$expectedHash","sha256":"$expectedHash","size":${bytes.size}}"""
|
||||
var uploadedTo: String? = null
|
||||
|
||||
val client =
|
||||
BlossomClient(
|
||||
dispatchingOkHttp { req ->
|
||||
when {
|
||||
req.method == "PUT" && req.url.encodedPath.endsWith("/mirror") -> response(req, 404)
|
||||
req.method == "GET" -> response(req, 200, bytes.toResponseBody("application/octet-stream".toMediaType()))
|
||||
req.method == "PUT" && req.url.encodedPath.endsWith("/upload") -> {
|
||||
uploadedTo = req.url.toString()
|
||||
response(req, 200, descriptor.toResponseBody())
|
||||
}
|
||||
else -> response(req, 500)
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
val result =
|
||||
client.mirrorOrUpload(
|
||||
sourceUrl = "https://source.example.com/$expectedHash",
|
||||
expectedHash = expectedHash,
|
||||
contentType = "image/png",
|
||||
serverBaseUrl = "https://target.example.com",
|
||||
authHeader = "Nostr abc",
|
||||
)
|
||||
|
||||
assertEquals("https://target.example.com/upload", uploadedTo)
|
||||
assertEquals(expectedHash, result.sha256)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun mirrorOrUploadRejectsHashMismatchAndDoesNotUpload() =
|
||||
runTest {
|
||||
val servedBytes = byteArrayOf(1, 2, 3)
|
||||
// Ask for a DIFFERENT blob than the source will serve — a substituted download.
|
||||
val expectedHash = sha256(byteArrayOf(4, 5, 6)).toHexKey()
|
||||
var uploaded = false
|
||||
|
||||
val client =
|
||||
BlossomClient(
|
||||
dispatchingOkHttp { req ->
|
||||
when {
|
||||
req.method == "PUT" && req.url.encodedPath.endsWith("/mirror") -> response(req, 405)
|
||||
req.method == "GET" -> response(req, 200, servedBytes.toResponseBody("application/octet-stream".toMediaType()))
|
||||
req.method == "PUT" && req.url.encodedPath.endsWith("/upload") -> {
|
||||
uploaded = true
|
||||
response(req, 200, "{}".toResponseBody())
|
||||
}
|
||||
else -> response(req, 500)
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
assertFailsWith<RuntimeException> {
|
||||
client.mirrorOrUpload(
|
||||
sourceUrl = "https://source.example.com/blob",
|
||||
expectedHash = expectedHash,
|
||||
contentType = "image/png",
|
||||
serverBaseUrl = "https://target.example.com",
|
||||
authHeader = null,
|
||||
)
|
||||
}
|
||||
assertFalse(uploaded)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun mirrorOrUploadUsesMirrorWhenSupported() =
|
||||
runTest {
|
||||
val descriptor = """{"url":"https://target.example.com/abc","sha256":"abc","size":3}"""
|
||||
var uploadCalled = false
|
||||
|
||||
val client =
|
||||
BlossomClient(
|
||||
dispatchingOkHttp { req ->
|
||||
when {
|
||||
req.method == "PUT" && req.url.encodedPath.endsWith("/mirror") -> response(req, 201, descriptor.toResponseBody())
|
||||
req.method == "PUT" && req.url.encodedPath.endsWith("/upload") -> {
|
||||
uploadCalled = true
|
||||
response(req, 200, descriptor.toResponseBody())
|
||||
}
|
||||
else -> response(req, 500)
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
val result =
|
||||
client.mirrorOrUpload(
|
||||
sourceUrl = "https://source.example.com/abc",
|
||||
expectedHash = "abc",
|
||||
contentType = "image/png",
|
||||
serverBaseUrl = "https://target.example.com",
|
||||
authHeader = "Nostr abc",
|
||||
)
|
||||
|
||||
assertFalse(uploadCalled)
|
||||
assertEquals("abc", result.sha256)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun uploadSuccessReturnsResult() =
|
||||
runTest {
|
||||
|
||||
Reference in New Issue
Block a user