Merge pull request #3991 from vitorpamplona/claude/slow-image-loading-feed-9b5leo

Move Blossom read-auth signing off OkHttp threads
This commit is contained in:
Vitor Pamplona
2026-08-26 11:37:29 -04:00
committed by GitHub
14 changed files with 982 additions and 219 deletions
@@ -208,6 +208,19 @@ class AppModules(
val applicationIOScope = CoroutineScope(Dispatchers.IO + SupervisorJob() + exceptionHandler)
/**
* Mints and caches BUD-01 read-auth tokens for auth-gated Blossom hosts.
* Shared by the OkHttp interceptor (which only reads the cache) and Coil's
* [com.vitorpamplona.amethyst.service.images.BlossomReadAuthFetcher] (which
* awaits a signature), so both see one token and one in-flight signature per
* host. Signing runs on [applicationIOScope], never on an OkHttp thread.
*/
val blossomReadAuthTokens =
BlossomReadAuthTokenProvider(
signerProvider = { sessionManager.loggedInAccount()?.signer },
scope = applicationIOScope,
)
private val _trimLevelEvents = MutableSharedFlow<Int>(extraBufferCapacity = 1, onBufferOverflow = BufferOverflow.DROP_OLDEST)
val trimLevelEvents = _trimLevelEvents.asSharedFlow()
@@ -448,9 +461,8 @@ class AppModules(
// tracks the logged-in account.
blossomReadAuth =
BlossomReadAuthInterceptor(
BlossomReadAuthTokenProvider(
signerProvider = { sessionManager.loggedInAccount()?.signer },
)::authHeader,
cachedHeaderProvider = blossomReadAuthTokens::cachedHeader,
onAuthRequired = blossomReadAuthTokens::warm,
),
)
@@ -1083,6 +1095,7 @@ class AppModules(
callFactory = { roleBasedHttpClientBuilder.okHttpClientForImage(it) },
thumbnailCache = thumbnailDiskCache,
backgroundScope = applicationIOScope,
readAuth = blossomReadAuthTokens,
)
}
@@ -27,11 +27,12 @@ import coil3.annotation.ExperimentalCoilApi
import coil3.fetch.FetchResult
import coil3.fetch.Fetcher
import coil3.network.CacheStrategy
import coil3.network.ConcurrentRequestStrategy
import coil3.network.ConnectivityChecker
import coil3.network.DeDupeConcurrentRequestStrategy
import coil3.network.NetworkFetcher
import coil3.network.okhttp.asNetworkClient
import coil3.request.Options
import com.vitorpamplona.amethyst.service.okhttp.BlossomReadAuthTokenProvider
import com.vitorpamplona.amethyst.service.uploads.blossom.bud10.BlossomServerResolver
import com.vitorpamplona.quartz.utils.startsWithIgnoreCase
import okhttp3.Call
@@ -57,8 +58,15 @@ class BlossomFetcher(
class Factory(
val blossomServerResolver: () -> BlossomServerResolver,
val networkClient: (url: String) -> Call.Factory,
// Shared with every other network-backed factory on this ImageLoader --
// see the note in ImageLoaderSetup.setup(): the de-dupe only works when
// all fetchers coordinate through the same instance.
concurrentRequestStrategy: ConcurrentRequestStrategy,
private val readAuth: BlossomReadAuthTokenProvider? = null,
) : Fetcher.Factory<Uri> {
private val cacheStrategyLazy = lazy { CacheStrategy.DEFAULT }
private val connectivityCheckerLazy = singleParameterLazy(::ConnectivityChecker)
private val concurrentRequestStrategyLazy = lazyOf(concurrentRequestStrategy)
override fun create(
data: Uri,
@@ -66,16 +74,20 @@ class BlossomFetcher(
imageLoader: ImageLoader,
): Fetcher? {
if (!isApplicable(data)) return null
// Wrapped per resolved url (not per Factory) because the server the
// blob actually lives on is only known once the resolver has run.
return BlossomFetcher(options, data, blossomServerResolver) { url ->
NetworkFetcher(
url = url,
options = options,
networkClient = lazy { networkClient(url).asNetworkClient() },
diskCache = lazy { imageLoader.diskCache },
cacheStrategy = lazy { CacheStrategy.DEFAULT },
connectivityChecker = lazy { connectivityCheckerLazy.get(options.context) },
concurrentRequestStrategy = lazy { DeDupeConcurrentRequestStrategy() },
)
readAuthAware(url, readAuth) { authHeader ->
NetworkFetcher(
url = url,
options = options.withAuthHeader(authHeader),
networkClient = lazy { networkClient(url).asNetworkClient() },
diskCache = lazy { imageLoader.diskCache },
cacheStrategy = cacheStrategyLazy,
connectivityChecker = lazy { connectivityCheckerLazy.get(options.context) },
concurrentRequestStrategy = concurrentRequestStrategyLazy,
)
}
}
}
@@ -0,0 +1,109 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.amethyst.service.images
import coil3.Extras
import coil3.fetch.FetchResult
import coil3.fetch.Fetcher
import coil3.network.HttpException
import coil3.network.httpHeaders
import coil3.request.Options
import com.vitorpamplona.amethyst.service.okhttp.BlossomReadAuthInterceptor
import com.vitorpamplona.amethyst.service.okhttp.BlossomReadAuthTokenProvider
import okhttp3.HttpUrl.Companion.toHttpUrlOrNull
/**
* Retries an auth-gated Blossom blob with a signed BUD-01 `t=get` token when the
* anonymous fetch comes back `401`.
*
* This is the half of the read-auth flow that has to wait for a signature.
* `BlossomReadAuthInterceptor` cannot: it runs on an OkHttp dispatcher thread,
* so waiting there holds one of the 16 per-host slots and stalls every other
* image from the same host. `Fetcher.fetch()` is `suspend`, so the wait costs a
* suspended coroutine and nothing else.
*
* [build] produces the underlying network fetcher, optionally carrying an
* `Authorization` header. Deliberately a `(String?) -> Fetcher` lambda rather
* than taking Coil's [Options] directly — the retry decision is then testable
* without an Android `Context` to construct [Options] with.
*/
class BlossomReadAuthFetcher(
private val url: String,
private val auth: BlossomReadAuthTokenProvider,
private val build: (authHeader: String?) -> Fetcher,
) : Fetcher {
override suspend fun fetch(): FetchResult? {
try {
return build(null).fetch()
} catch (e: HttpException) {
// Coil's NetworkFetcher throws HttpException for any non-2xx/304,
// which is how a 401 reaches us with its code intact.
if (e.response.code != HTTP_UNAUTHORIZED) throw e
val httpUrl = url.toHttpUrlOrNull() ?: throw e
// Gate only: read-auth applies to Blossom blob URLs, but the token is
// scoped to the host (BUD-11 `server` tag) and carries no `x` tag.
BlossomReadAuthInterceptor.blossomHashOrNull(httpUrl.encodedPath) ?: throw e
val header = auth.header(httpUrl.host) ?: throw e
return build(header).fetch()
}
}
companion object {
private const val HTTP_UNAUTHORIZED = 401
}
}
/**
* Wraps [build] in read-auth handling when a token provider is configured, and
* returns the plain fetcher when it isn't (tests, pre-configuration call sites).
*/
fun readAuthAware(
url: String,
auth: BlossomReadAuthTokenProvider?,
build: (authHeader: String?) -> Fetcher,
): Fetcher =
if (auth == null) {
build(null)
} else {
BlossomReadAuthFetcher(url, auth, build)
}
/**
* Copy of these options carrying [header] as `Authorization`, or the same
* options when there is no header. Coil's `NetworkFetcher` builds its request
* from `options.httpHeaders`, so this is how the retry gets authenticated.
*/
fun Options.withAuthHeader(header: String?): Options =
if (header == null) {
this
} else {
copy(
extras =
extras
.newBuilder()
.set(
Extras.Key.httpHeaders,
httpHeaders.newBuilder().set("Authorization", header).build(),
).build(),
)
}
@@ -33,6 +33,7 @@ import coil3.gif.AnimatedImageDecoder
import coil3.gif.GifDecoder
import coil3.memory.MemoryCache
import coil3.network.CacheStrategy
import coil3.network.ConcurrentRequestStrategy
import coil3.network.ConnectivityChecker
import coil3.network.DeDupeConcurrentRequestStrategy
import coil3.network.NetworkFetcher
@@ -43,6 +44,7 @@ import coil3.svg.SvgDecoder
import coil3.util.Logger
import coil3.video.VideoFrameDecoder
import com.vitorpamplona.amethyst.isDebug
import com.vitorpamplona.amethyst.service.okhttp.BlossomReadAuthTokenProvider
import com.vitorpamplona.amethyst.service.uploads.blossom.bud10.BlossomServerResolver
import com.vitorpamplona.quartz.utils.Log
import kotlinx.coroutines.CoroutineScope
@@ -60,7 +62,7 @@ class ImageLoaderSetup {
val debugLogger = if (isDebug) MyDebugLogger() else null
@OptIn(DelicateCoilApi::class)
@OptIn(DelicateCoilApi::class, ExperimentalCoilApi::class)
fun setup(
app: Context,
diskCache: () -> DiskCache,
@@ -69,7 +71,19 @@ class ImageLoaderSetup {
callFactory: (url: String) -> Call.Factory,
thumbnailCache: ThumbnailDiskCache,
backgroundScope: CoroutineScope,
// Signs the BUD-01 retry when a gated host answers 401. Null keeps every
// fetch anonymous (tests, pre-configuration call sites).
readAuth: BlossomReadAuthTokenProvider? = null,
) {
// ONE strategy for the whole ImageLoader. DeDupeConcurrentRequestStrategy
// coordinates through a map of in-flight fetches that it owns, so it only
// works when every fetcher shares the same instance -- a fresh one per
// request can never see anybody else's fetch and the de-dupe silently
// no-ops. Shared across all three network-backed factories so a feed
// image and the same blob reached through `blossom:` (or a profile
// picture) still collapse onto one download.
val concurrentRequests = DeDupeConcurrentRequestStrategy()
SingletonImageLoader.setUnsafe(
ImageLoader
.Builder(app)
@@ -87,13 +101,13 @@ class ImageLoaderSetup {
add(Base64Fetcher.Factory)
add(BlurHashFetcher.Factory)
add(ThumbHashFetcher.Factory)
add(BlossomFetcher.Factory(blossomServerResolver, callFactory))
add(ProfilePictureFetcher.Factory(thumbnailCache, callFactory, backgroundScope))
add(BlossomFetcher.Factory(blossomServerResolver, callFactory, concurrentRequests, readAuth))
add(ProfilePictureFetcher.Factory(thumbnailCache, callFactory, backgroundScope, concurrentRequests, readAuth))
add(Base64Fetcher.BKeyer)
add(BlurHashFetcher.BKeyer)
add(ThumbHashFetcher.TKeyer)
add(ProfilePictureFetcher.BKeyer)
add(OkHttpFactory(callFactory))
add(OkHttpFactory(callFactory, concurrentRequests, readAuth))
}.build(),
)
}
@@ -132,9 +146,12 @@ class MyDebugLogger(
@OptIn(ExperimentalCoilApi::class)
class OkHttpFactory(
val networkClient: (url: String) -> Call.Factory,
concurrentRequestStrategy: ConcurrentRequestStrategy,
private val readAuth: BlossomReadAuthTokenProvider? = null,
) : Fetcher.Factory<Uri> {
private val cacheStrategyLazy = lazy { CacheStrategy.DEFAULT }
private val connectivityCheckerLazy = singleParameterLazy(::ConnectivityChecker)
private val concurrentRequestStrategyLazy = lazyOf(concurrentRequestStrategy)
override fun create(
data: Uri,
@@ -145,15 +162,17 @@ class OkHttpFactory(
val url = data.toString()
return NetworkFetcher(
url = url,
options = options,
networkClient = lazy { networkClient(url).asNetworkClient() },
diskCache = lazy { imageLoader.diskCache },
cacheStrategy = cacheStrategyLazy,
connectivityChecker = lazy { connectivityCheckerLazy.get(options.context) },
concurrentRequestStrategy = lazy { DeDupeConcurrentRequestStrategy() },
)
return readAuthAware(url, readAuth) { authHeader ->
NetworkFetcher(
url = url,
options = options.withAuthHeader(authHeader),
networkClient = lazy { networkClient(url).asNetworkClient() },
diskCache = lazy { imageLoader.diskCache },
cacheStrategy = cacheStrategyLazy,
connectivityChecker = lazy { connectivityCheckerLazy.get(options.context) },
concurrentRequestStrategy = concurrentRequestStrategyLazy,
)
}
}
private fun isApplicable(data: Uri): Boolean = data.scheme == "http" || data.scheme == "https"
@@ -30,12 +30,13 @@ import coil3.fetch.Fetcher
import coil3.fetch.ImageFetchResult
import coil3.key.Keyer
import coil3.network.CacheStrategy
import coil3.network.ConcurrentRequestStrategy
import coil3.network.ConnectivityChecker
import coil3.network.DeDupeConcurrentRequestStrategy
import coil3.network.NetworkFetcher
import coil3.network.okhttp.asNetworkClient
import coil3.request.Options
import com.vitorpamplona.amethyst.commons.ui.components.ProfilePictureUrl
import com.vitorpamplona.amethyst.service.okhttp.BlossomReadAuthTokenProvider
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.launch
import okhttp3.Call
@@ -95,8 +96,16 @@ class ProfilePictureFetcher(
private val thumbnailCache: ThumbnailDiskCache,
private val networkClient: (url: String) -> Call.Factory,
private val backgroundScope: CoroutineScope,
// Shared with every other network-backed factory on this ImageLoader --
// see the note in ImageLoaderSetup.setup(). Avatars repeat constantly
// down a feed, so this is where a per-request instance cost the most:
// every row holding the same author's picture downloaded it again.
concurrentRequestStrategy: ConcurrentRequestStrategy,
private val readAuth: BlossomReadAuthTokenProvider? = null,
) : Fetcher.Factory<ProfilePictureUrl> {
private val cacheStrategyLazy = lazy { CacheStrategy.DEFAULT }
private val connectivityCheckerLazy = singleParameterLazy(::ConnectivityChecker)
private val concurrentRequestStrategyLazy = lazyOf(concurrentRequestStrategy)
override fun create(
data: ProfilePictureUrl,
@@ -106,15 +115,17 @@ class ProfilePictureFetcher(
val diskCacheLazy = lazy { imageLoader.diskCache }
val netFetcher =
NetworkFetcher(
url = data.url,
options = options,
networkClient = lazy { networkClient(data.url).asNetworkClient() },
diskCache = diskCacheLazy,
cacheStrategy = lazy { CacheStrategy.DEFAULT },
connectivityChecker = lazy { connectivityCheckerLazy.get(options.context) },
concurrentRequestStrategy = lazy { DeDupeConcurrentRequestStrategy() },
)
readAuthAware(data.url, readAuth) { authHeader ->
NetworkFetcher(
url = data.url,
options = options.withAuthHeader(authHeader),
networkClient = lazy { networkClient(data.url).asNetworkClient() },
diskCache = diskCacheLazy,
cacheStrategy = cacheStrategyLazy,
connectivityChecker = lazy { connectivityCheckerLazy.get(options.context) },
concurrentRequestStrategy = concurrentRequestStrategyLazy,
)
}
return ProfilePictureFetcher(
data.url,
@@ -49,22 +49,27 @@ import java.util.concurrent.ConcurrentHashMap
* - and at most one retry (an application interceptor's second `chain.proceed`
* runs the downstream chain again, it does not re-enter this interceptor).
*
* [authHeaderProvider] is `(host, sha256) -> header?`. It is synchronous by
* contract (the caller bridges the suspend signer), returns `null` when no
* signer is available or signing times out, and is only consulted on a real
* `401`, so an unauthenticated user simply keeps seeing the broken image
* rather than paying any signing cost.
* This interceptor never signs and never waits. [cachedHeaderProvider] is a
* pure cache read and [onAuthRequired] is fire-and-forget: `intercept` runs on
* an OkHttp dispatcher thread, where blocking would hold one of the 16 per-host
* slots for the whole signing window and stall every other image from that
* host. The signed *retry* therefore lives one layer up, in
* `BlossomReadAuthFetcher`, which is `suspend` and can await the signature
* without occupying a slot.
*
* The first blob from an auth-gated host costs an extra round trip (anonymous
* `GET` → `401` → signed retry), but that host is then remembered in
* [knownAuthHosts] so every later blob from it is signed **up front** — one
* round trip, not two. This matters on a Buzz community feed where nearly every
* image comes from the same gated host: without it each image would keep paying
* the wasted 401 probe. The learned host also short-circuits to anonymous when
* no signer is available, so a logged-out user never re-probes needlessly.
* The first blob from an auth-gated host still costs an extra round trip
* (anonymous `GET` -> `401` -> signed retry by the fetcher), but the host is
* then remembered in [knownAuthHosts] so every later blob from it is signed
* **up front** from the cache — one round trip, not two. This matters on a Buzz
* community feed where nearly every image comes from the same gated host.
* Callers that cannot retry (e.g. the media3 video datasource) get the token on
* their next request, once [onAuthRequired] has landed it in the cache.
*/
class BlossomReadAuthInterceptor(
private val authHeaderProvider: (host: String, sha256: HexKey) -> String?,
/** Pure cache read — must not sign, must not block. */
private val cachedHeaderProvider: (host: String) -> String?,
/** Fire-and-forget: starts a signature for a host we just learned is gated. */
private val onAuthRequired: (host: String) -> Unit,
) : Interceptor {
// Hosts observed to answer 401 to an anonymous Blossom GET. Small (a user
// follows a handful of auth-gated servers at most) and shared across all
@@ -81,14 +86,16 @@ class BlossomReadAuthInterceptor(
return chain.proceed(request)
}
val sha256 = blossomHashOrNull(request.url.encodedPath) ?: return chain.proceed(request)
// The hash is a gate, not an input: read-auth applies only to Blossom
// blob URLs. The token itself is host-scoped and carries no `x` tag.
blossomHashOrNull(request.url.encodedPath) ?: return chain.proceed(request)
val host = request.url.host
// Known-gated host: skip the anonymous probe and sign the first attempt.
// Falls through to anonymous only when we can't produce a token (no
// signer / timeout) — the server would 401 either way.
if (host in knownAuthHosts) {
authHeaderProvider(host, sha256)?.let { header ->
cachedHeaderProvider(host)?.let { header ->
return chain.proceed(request.withAuth(header))
}
}
@@ -99,12 +106,12 @@ class BlossomReadAuthInterceptor(
// Learn the host so its next blob is signed up front.
knownAuthHosts.add(host)
val header = authHeaderProvider(host, sha256) ?: return response
// Start the signature but do not wait for it: this thread holds a
// per-host dispatcher slot. BlossomReadAuthFetcher performs the signed
// retry for this very request from a coroutine.
onAuthRequired(host)
// Close the 401 body before replaying so the connection can be reused.
response.close()
return chain.proceed(request.withAuth(header))
return response
}
private fun Request.withAuth(header: String) =
@@ -21,29 +21,49 @@
package com.vitorpamplona.amethyst.service.okhttp
import com.vitorpamplona.amethyst.commons.service.upload.BlossomAuth
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.launch
import kotlinx.coroutines.withTimeoutOrNull
import java.util.concurrent.ConcurrentHashMap
import kotlin.coroutines.cancellation.CancellationException
/**
* Signs and caches BUD-01 read-auth headers for [BlossomReadAuthInterceptor].
* Signs and caches BUD-01 read-auth headers for auth-gated Blossom hosts.
*
* The interceptor is synchronous (it runs on an OkHttp dispatcher thread) but
* signing is `suspend`, so [authHeader] bridges with [runBlocking] guarded by a
* timeout: an internal key signs instantly, while a remote (NIP-46) or external
* (NIP-55) signer that hangs or needs user interaction simply yields `null` and
* the download stays unauthenticated instead of pinning the thread.
* Signing never blocks a caller's thread. It runs on [scope]; callers either
* `suspend` on [header] or fire [warm] and pick the token up later. That
* matters because the consumer used to be [BlossomReadAuthInterceptor], which
* runs on an OkHttp dispatcher thread — bridging the suspend signer with
* `runBlocking` there held one of the 16 per-host dispatcher slots for as long
* as the signer took (up to the timeout), so a feed's first burst against a
* gated host could occupy every slot and stall every other image from it.
*
* Tokens are cached per host, not per blob. A BUD-11 `server`-scoped token
* grants reads for every blob on the host (thumbnails included), so one signed
* event covers a whole feed's worth of images from an auth-gated host for the
* life of the token. The blob hash of the request that first triggered signing
* is still included as the `x` tag for BUD-01 servers that check it.
* One signature per host, however many callers. The token cache alone couldn't
* provide that: it is only populated *after* a signature returns, so a cold
* burst of N images all missed it and all signed concurrently — with a NIP-55
* external signer that meant N IPC round trips (and potentially N prompts).
* [inFlight] is what collapses them; the leader signs and every follower awaits
* the same [CompletableDeferred].
*
* Tokens are cached per host, not per blob, and are therefore minted with a
* BUD-11 `server` tag and **no** `x` tag. BUD-11 lists `x` as optional for
* `GET /<sha256>` but is strict about what including one means: "When `x` tags
* are present, the token is only valid for operations on the specified blob
* hashes." A token carrying the hash of whichever blob happened to trigger
* signing would therefore be invalid for every other blob it was reused for.
* Server-scoped and hash-free, one signed event legitimately covers a whole
* feed's worth of images from the host for the life of the token.
*
* The tradeoff that buys: the token authorizes reading any blob on that host
* until it expires, rather than one. It is only ever sent to that host, over
* TLS, and BUD-11 sanctions the shape — but it is a wider grant than a
* per-blob token, which is the price of caching at all.
*/
class BlossomReadAuthTokenProvider(
private val signerProvider: () -> NostrSigner?,
private val scope: CoroutineScope,
private val clock: () -> Long = { System.currentTimeMillis() },
) {
private class CachedToken(
@@ -52,31 +72,82 @@ class BlossomReadAuthTokenProvider(
)
private val cache = ConcurrentHashMap<String, CachedToken>()
private val inFlight = ConcurrentHashMap<String, CompletableDeferred<String?>>()
fun authHeader(
host: String,
sha256: HexKey,
): String? {
val now = clock()
/**
* The token already held for [host], or null. Pure map read — safe to call
* from an OkHttp interceptor, and never signs.
*/
fun cachedHeader(host: String): String? = cache[host]?.takeIf { it.expiresAtMs > clock() }?.header
cache[host]?.let { if (it.expiresAtMs > now) return it.header }
/**
* The token for [host], signing one if none is cached. Suspends rather than
* blocking, so the caller must already be in a coroutine — on the image path
* that is Coil's `Fetcher.fetch()`.
*/
suspend fun header(host: String): String? {
cachedHeader(host)?.let { return it }
return signOnce(host)?.await()
}
/**
* Starts a signature for [host] without waiting for it. For callers that
* cannot suspend (the interceptor) and only need the token to exist by the
* time some later request needs it.
*/
fun warm(host: String) {
if (cachedHeader(host) != null) return
signOnce(host)
}
/**
* Returns the in-flight signature for [host], starting one if this caller
* wins the race. Null when there is no signer to sign with.
*
* Leader/follower over [ConcurrentHashMap.putIfAbsent] rather than
* `computeIfAbsent`: the completion handler removes the map entry, and a job
* that finishes immediately would run that removal *inside* the mapping
* function, which `ConcurrentHashMap` forbids.
*/
private fun signOnce(host: String): CompletableDeferred<String?>? {
inFlight[host]?.let { return it }
val signer = signerProvider() ?: return null
val header =
runBlocking {
withTimeoutOrNull(SIGN_TIMEOUT_MS) {
BlossomAuth.createGetAuth(
hash = sha256,
alt = "Downloading media from $host",
signer = signer,
servers = listOf(host),
)
}
} ?: return null
val fresh = CompletableDeferred<String?>()
inFlight.putIfAbsent(host, fresh)?.let { return it }
cache[host] = CachedToken(header, now + CACHE_TTL_MS)
return header
scope
.launch {
val header =
try {
withTimeoutOrNull(SIGN_TIMEOUT_MS) {
BlossomAuth.createGetAuth(
// No `x` tag: this token is reused for every blob
// on the host. See the class kdoc.
hash = null,
alt = "Downloading media from $host",
signer = signer,
servers = listOf(host),
)
}
} catch (e: Exception) {
if (e is CancellationException) throw e
null
}
if (header != null) {
cache[host] = CachedToken(header, clock() + CACHE_TTL_MS)
}
fresh.complete(header)
}.invokeOnCompletion {
inFlight.remove(host, fresh)
// No-op when the job completed normally; releases followers when
// it was cancelled (scope torn down) instead of hanging them.
fresh.complete(null)
}
return fresh
}
companion object {
@@ -84,7 +155,8 @@ class BlossomReadAuthTokenProvider(
// refresh a little early to avoid handing over a token that dies mid-flight.
private const val CACHE_TTL_MS = 55L * 60L * 1000L
// Bounds how long an image download may block waiting on a slow signer.
// Bounds how long an image may wait on a slow signer. No thread is held
// for this window any more — only the waiting coroutine.
private const val SIGN_TIMEOUT_MS = 8_000L
}
}
@@ -0,0 +1,171 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.amethyst.service.images
import coil3.fetch.FetchResult
import coil3.fetch.Fetcher
import coil3.network.HttpException
import coil3.network.NetworkResponse
import com.vitorpamplona.amethyst.service.okhttp.BlossomReadAuthTokenProvider
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.cancel
import kotlinx.coroutines.runBlocking
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNull
import org.junit.Assert.assertThrows
import org.junit.Assert.assertTrue
import org.junit.Test
/**
* The signed retry that used to live in `BlossomReadAuthInterceptor` (behind a
* `runBlocking`) now lives here, where `fetch()` is already `suspend`.
*/
class BlossomReadAuthFetcherTest {
private val sha = "2c5287a55cc550c9d6bc4206a4663900e083315f4a544ea3bc189e43dc330af6"
private val host = "nosfabrica.communities.buzz.xyz"
private val url = "https://nosfabrica.communities.buzz.xyz/media/2c5287a55cc550c9d6bc4206a4663900e083315f4a544ea3bc189e43dc330af6.png"
private val signer = NostrSignerInternal(KeyPair())
private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
@After
fun tearDown() = scope.cancel()
private fun unauthorized() = HttpException(NetworkResponse(code = 401))
private fun notFound() = HttpException(NetworkResponse(code = 404))
/** Records the header each attempt carried, and fails the attempts in [failWith]. */
private class RecordingBuilder(
private val failWith: List<HttpException?>,
) {
val headers = mutableListOf<String?>()
fun build(authHeader: String?): Fetcher {
val attempt = headers.size
headers.add(authHeader)
return Fetcher {
failWith.getOrNull(attempt)?.let { throw it }
null
}
}
}
private fun Fetcher(block: suspend () -> FetchResult?): Fetcher =
object : Fetcher {
override suspend fun fetch(): FetchResult? = block()
}
@Test
fun anonymousSuccessNeverSigns() =
runBlocking {
val builder = RecordingBuilder(failWith = listOf(null))
val provider = BlossomReadAuthTokenProvider({ signer }, scope)
BlossomReadAuthFetcher(url, provider, builder::build).fetch()
assertEquals("one attempt only", 1, builder.headers.size)
assertNull("and it must be anonymous", builder.headers.single())
assertNull("no token minted for a host that never 401'd", provider.cachedHeader(host))
}
@Test
fun signsAndRetriesOn401() =
runBlocking {
val builder = RecordingBuilder(failWith = listOf(unauthorized(), null))
val provider = BlossomReadAuthTokenProvider({ signer }, scope)
BlossomReadAuthFetcher(url, provider, builder::build).fetch()
assertEquals("anonymous attempt then signed retry", 2, builder.headers.size)
assertNull(builder.headers[0])
assertTrue(
"retry must carry a Nostr token, got ${builder.headers[1]}",
builder.headers[1]!!.startsWith("Nostr "),
)
}
@Test
fun nonAuthFailuresPropagateUnretried() =
runBlocking {
val builder = RecordingBuilder(failWith = listOf(notFound()))
val provider = BlossomReadAuthTokenProvider({ signer }, scope)
assertThrows(HttpException::class.java) {
runBlocking { BlossomReadAuthFetcher(url, provider, builder::build).fetch() }
}
assertEquals("a 404 must not be retried", 1, builder.headers.size)
}
@Test
fun without401CapableSignerThe401Propagates() =
runBlocking {
val builder = RecordingBuilder(failWith = listOf(unauthorized()))
val provider = BlossomReadAuthTokenProvider({ null }, scope)
assertThrows(HttpException::class.java) {
runBlocking { BlossomReadAuthFetcher(url, provider, builder::build).fetch() }
}
assertEquals("no signer means no retry", 1, builder.headers.size)
}
@Test
fun nonBlossomUrlIsNotRetried() =
runBlocking {
val builder = RecordingBuilder(failWith = listOf(unauthorized()))
val provider = BlossomReadAuthTokenProvider({ signer }, scope)
assertThrows(HttpException::class.java) {
runBlocking {
BlossomReadAuthFetcher("https://example.com/media/avatar.png", provider, builder::build).fetch()
}
}
assertEquals(1, builder.headers.size)
}
/**
* The feed shape: a burst of images from a gated host all 401 at once. Each
* retries, but they share one signature via the provider's single-flight.
*/
@Test
fun aBurstOf401sSharesOneSignature() =
runBlocking {
val provider = BlossomReadAuthTokenProvider({ signer }, scope)
val builders = (1..16).map { RecordingBuilder(failWith = listOf(unauthorized(), null)) }
builders
.map { b ->
async(Dispatchers.Default) {
BlossomReadAuthFetcher(url, provider, b::build).fetch()
}
}.awaitAll()
val retryHeaders = builders.map { it.headers[1] }
assertTrue("every retry must be signed", retryHeaders.all { it != null })
assertEquals("all 16 retries must reuse one token", 1, retryHeaders.toSet().size)
}
}
@@ -20,7 +20,13 @@
*/
package com.vitorpamplona.amethyst.service.okhttp
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.cancel
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import okhttp3.HttpUrl.Companion.toHttpUrl
import okhttp3.Interceptor
import okhttp3.Protocol
@@ -47,7 +53,6 @@ class BlossomReadAuthInterceptorTest {
@Test
fun hashParsedFromThumbnailVariant() {
// Buzz thumbnails use the dot form <hash>.thumb.jpg — the base is still the hash.
assertEquals(sha, BlossomReadAuthInterceptor.blossomHashOrNull("/media/$sha.thumb.jpg"))
}
@@ -58,126 +63,113 @@ class BlossomReadAuthInterceptorTest {
@Test
fun hashLowercasedFromUppercaseSegment() {
assertEquals(sha, BlossomReadAuthInterceptor.blossomHashOrNull("/media/${sha.uppercase()}.png"))
assertEquals(sha, BlossomReadAuthInterceptor.blossomHashOrNull("/${sha.uppercase()}.png"))
}
@Test
fun nonBlobPathsReturnNull() {
assertNull(BlossomReadAuthInterceptor.blossomHashOrNull("/media/avatar.png"))
assertNull(BlossomReadAuthInterceptor.blossomHashOrNull("/media/nostr.build_$sha.jpg"))
assertNull(BlossomReadAuthInterceptor.blossomHashOrNull("/media/${sha}_thumb.jpg"))
// 65 hex chars: isHex64 checks only the first 64, so the length guard must reject it.
assertNull(BlossomReadAuthInterceptor.blossomHashOrNull("/media/${sha}a.png"))
assertNull(BlossomReadAuthInterceptor.blossomHashOrNull("/"))
}
// --- intercept behavior ----------------------------------------------
//
// The interceptor no longer performs the signed retry: waiting for a
// signature on an OkHttp dispatcher thread held one of the 16 per-host
// slots. It now returns the 401 and asks for a token to be minted
// off-thread; BlossomReadAuthFetcher does the retry from a coroutine.
@Test
fun retriesWithAuthOn401() {
val provider = RecordingProvider(header = "Nostr token")
val chain = fakeChain("https://$host/media/$sha.png", codes = listOf(401, 200))
fun learnsHostAndAsksForATokenOn401() {
val provider = RecordingProvider(cached = null)
val chain = fakeChain("https://$host/media/$sha.png", codes = listOf(401))
val response = BlossomReadAuthInterceptor(provider::header).intercept(chain.asChain())
val response = provider.interceptor().intercept(chain.asChain())
assertEquals(200, response.code)
assertEquals(2, chain.requests.size)
assertNull("first attempt is anonymous", chain.requests[0].header("Authorization"))
assertEquals("Nostr token", chain.requests[1].header("Authorization"))
assertEquals(host to sha, provider.calls.single())
assertEquals("the 401 is surfaced for the fetcher to retry", 401, response.code)
assertEquals("the interceptor must not retry itself", 1, chain.requests.size)
assertNull("the only attempt is anonymous", chain.requests[0].header("Authorization"))
assertEquals(host, provider.warmed.single())
response.close()
}
@Test
fun thumbnailUrlAlsoRetries() {
val provider = RecordingProvider(header = "Nostr token")
val chain = fakeChain("https://$host/media/$sha.thumb.jpg", codes = listOf(401, 200))
fun thumbnailUrlAlsoAsksForAToken() {
val provider = RecordingProvider(cached = null)
val chain = fakeChain("https://$host/media/$sha.thumb.jpg", codes = listOf(401))
val response = BlossomReadAuthInterceptor(provider::header).intercept(chain.asChain())
provider.interceptor().intercept(chain.asChain()).close()
assertEquals(200, response.code)
assertEquals(sha, provider.calls.single().second)
response.close()
assertEquals(host, provider.warmed.single())
}
@Test
fun successfulRequestNeverSigns() {
val provider = RecordingProvider(header = "Nostr token")
val provider = RecordingProvider(cached = "Nostr token")
val chain = fakeChain("https://blossom.example.com/$sha.png", codes = listOf(200))
val response = BlossomReadAuthInterceptor(provider::header).intercept(chain.asChain())
val response = provider.interceptor().intercept(chain.asChain())
assertEquals(200, response.code)
assertEquals(1, chain.requests.size)
assertTrue("public host must not be signed", provider.calls.isEmpty())
response.close()
}
@Test
fun keepsThe401WhenNoSignerAvailable() {
val provider = RecordingProvider(header = null)
val chain = fakeChain("https://$host/media/$sha.png", codes = listOf(401))
val response = BlossomReadAuthInterceptor(provider::header).intercept(chain.asChain())
assertEquals(401, response.code)
assertEquals(1, chain.requests.size)
assertEquals(host to sha, provider.calls.single())
assertTrue("public host must not be signed", provider.warmed.isEmpty())
response.close()
}
@Test
fun nonBlobUrlNeverSignsEvenOn401() {
val provider = RecordingProvider(header = "Nostr token")
val provider = RecordingProvider(cached = "Nostr token")
val chain = fakeChain("https://example.com/media/avatar.png", codes = listOf(401))
val response = BlossomReadAuthInterceptor(provider::header).intercept(chain.asChain())
val response = provider.interceptor().intercept(chain.asChain())
assertEquals(401, response.code)
assertEquals(1, chain.requests.size)
assertTrue(provider.calls.isEmpty())
assertTrue(provider.warmed.isEmpty())
response.close()
}
@Test
fun requestWithExistingAuthPassesThrough() {
val provider = RecordingProvider(header = "Nostr token")
val provider = RecordingProvider(cached = "Nostr token")
val chain = fakeChain("https://$host/media/$sha.png", codes = listOf(401), preAuthHeader = "Nostr existing")
val response = BlossomReadAuthInterceptor(provider::header).intercept(chain.asChain())
val response = provider.interceptor().intercept(chain.asChain())
assertEquals(401, response.code)
assertEquals(1, chain.requests.size)
assertTrue(provider.calls.isEmpty())
assertTrue(provider.warmed.isEmpty())
response.close()
}
@Test
fun nonGetRequestPassesThrough() {
val provider = RecordingProvider(header = "Nostr token")
val provider = RecordingProvider(cached = "Nostr token")
val chain = fakeChain("https://$host/media/$sha.png", codes = listOf(401), method = "PUT")
val response = BlossomReadAuthInterceptor(provider::header).intercept(chain.asChain())
val response = provider.interceptor().intercept(chain.asChain())
assertEquals(401, response.code)
assertEquals(1, chain.requests.size)
assertTrue(provider.calls.isEmpty())
assertTrue(provider.warmed.isEmpty())
response.close()
}
@Test
fun learnsHostThenSignsSubsequentBlobsUpFront() {
val provider = RecordingProvider(header = "Nostr token")
val interceptor = BlossomReadAuthInterceptor(provider::header)
fun learnsHostThenSignsSubsequentBlobsUpFrontFromCache() {
val provider = RecordingProvider(cached = null)
val interceptor = provider.interceptor()
val otherSha = "b1674191a88ec5cdd733e4240a81803105dc412d6c6708d53ab94fc248f4f553"
// First blob learns the host via the 401 probe + signed retry.
val first = fakeChain("https://$host/media/$sha.png", codes = listOf(401, 200))
// First blob learns the host from the 401 and asks for a token.
val first = fakeChain("https://$host/media/$sha.png", codes = listOf(401))
interceptor.intercept(first.asChain()).close()
assertEquals(2, first.requests.size)
assertEquals(1, first.requests.size)
// Second, different blob on the same host is signed on the first attempt —
// no anonymous probe, so a single request.
// Once that token has landed, the next blob on the same host is signed
// on its first attempt — no anonymous probe.
provider.cached = "Nostr token"
val second = fakeChain("https://$host/media/$otherSha.png", codes = listOf(200))
val response = interceptor.intercept(second.asChain())
@@ -188,15 +180,15 @@ class BlossomReadAuthInterceptorTest {
}
@Test
fun learnedHostWithoutSignerDoesNotLoop() {
val provider = RecordingProvider(header = null)
val interceptor = BlossomReadAuthInterceptor(provider::header)
fun learnedHostWithoutTokenDoesNotLoop() {
val provider = RecordingProvider(cached = null)
val interceptor = provider.interceptor()
val first = fakeChain("https://$host/media/$sha.png", codes = listOf(401))
interceptor.intercept(first.asChain()).close()
// Host is now known, but with no signer the preemptive path must fall
// back to a single anonymous request rather than retrying endlessly.
// Host is known but no token was minted (no signer). The preemptive path
// must fall back to a single anonymous request rather than looping.
val second = fakeChain("https://$host/media/$sha.png", codes = listOf(401))
val response = interceptor.intercept(second.asChain())
@@ -205,20 +197,84 @@ class BlossomReadAuthInterceptorTest {
response.close()
}
private class RecordingProvider(
private val header: String?,
) {
val calls = mutableListOf<Pair<String, HexKey>>()
/**
* The point of the whole split: `intercept` runs on an OkHttp dispatcher
* thread and holds one of the 16 per-host slots for as long as it stays
* there, so it must return without waiting for a signature.
*
* Uses the real provider and a deliberately slow signer rather than a fake
* `warm`: what is being measured is that the production wiring hands the
* signing off, not that a stub returns quickly.
*/
@Test
fun interceptDoesNotWaitForTheSignature() {
val slowSigner = DelayingTestSigner(delayMs = SIGN_MS)
val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
try {
val provider = BlossomReadAuthTokenProvider({ slowSigner }, scope)
val interceptor = BlossomReadAuthInterceptor(provider::cachedHeader, provider::warm)
val chain = fakeChain("https://$host/media/$sha.png", codes = listOf(401))
fun header(
host: String,
sha256: HexKey,
): String? {
calls.add(host to sha256)
return header
// What the previous design cost: the interceptor bridged the suspend
// signer with runBlocking, so the calling thread wore the full
// signing latency. Same signer, same host, measured on this thread.
val blockingScope = CoroutineScope(Dispatchers.Default + SupervisorJob())
val blockingProbe = BlossomReadAuthTokenProvider({ DelayingTestSigner(delayMs = SIGN_MS) }, blockingScope)
val beforeAt = System.nanoTime()
runBlocking { blockingProbe.header(host) }
val beforeMs = (System.nanoTime() - beforeAt) / 1_000_000
blockingScope.cancel()
val startedAt = System.nanoTime()
interceptor.intercept(chain.asChain()).close()
val elapsedMs = (System.nanoTime() - startedAt) / 1_000_000
println(
"[measure] signature=${SIGN_MS}ms waiting-for-it=${beforeMs}ms " +
"intercept-now=${elapsedMs}ms",
)
assertTrue(
"intercept must not wait out the ${SIGN_MS}ms signature; it took ${elapsedMs}ms",
elapsedMs < SIGN_MS / 2,
)
assertNull(
"returning before the token exists is exactly the point",
provider.cachedHeader(host),
)
// ...and the signature it kicked off does still complete.
runBlocking {
withTimeout(SIGN_MS * 20) {
while (provider.cachedHeader(host) == null) delay(10)
}
}
assertEquals(1, slowSigner.signatures)
} finally {
scope.cancel()
}
}
private companion object {
// Long enough that a blocking implementation could not possibly pass the
// assertion above, short enough to keep the suite quick.
const val SIGN_MS = 2_000L
}
private class RecordingProvider(
var cached: String?,
) {
val warmed = mutableListOf<String>()
fun cachedHeader(host: String): String? = cached
fun warm(host: String) {
warmed.add(host)
}
fun interceptor() = BlossomReadAuthInterceptor(::cachedHeader, ::warm)
}
/**
* Records every [Request] it is asked to proceed and answers each with the
* next code from [codes], so `[401, 200]` models "anonymous fails,
@@ -21,7 +21,18 @@
package com.vitorpamplona.amethyst.service.okhttp
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.jackson.JacksonMapper
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
import com.vitorpamplona.quartz.nipB7Blossom.BlossomAuthorizationEvent
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async
import kotlinx.coroutines.awaitAll
import kotlinx.coroutines.cancel
import kotlinx.coroutines.runBlocking
import kotlinx.coroutines.withTimeout
import org.junit.After
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotEquals
import org.junit.Assert.assertNull
@@ -33,59 +44,170 @@ class BlossomReadAuthTokenProviderTest {
private val host = "nosfabrica.communities.buzz.xyz"
private val signer = NostrSignerInternal(KeyPair())
@Test
fun signsAndFormatsHeader() {
val provider = BlossomReadAuthTokenProvider(signerProvider = { signer })
val header = provider.authHeader(host, sha)
assertTrue("expected a Nostr auth header, got $header", header!!.startsWith("Nostr "))
}
// A real dispatcher, not runTest's virtual clock: the concurrency test below
// measures wall time, which virtual time would collapse to zero.
private val scope = CoroutineScope(Dispatchers.Default + SupervisorJob())
@After
fun tearDown() = scope.cancel()
@Test
fun returnsNullWhenNoSigner() {
val provider = BlossomReadAuthTokenProvider(signerProvider = { null })
assertNull(provider.authHeader(host, sha))
}
fun signsAndFormatsHeader() =
runBlocking {
val provider = BlossomReadAuthTokenProvider({ signer }, scope)
val header = provider.header(host)
assertTrue("expected a Nostr auth header, got $header", header!!.startsWith("Nostr "))
}
@Test
fun cachesPerHostWithinTtl() {
// signerProvider is consulted only on a cache miss, so its invocation
// count is the number of times a fresh token was signed.
var lookups = 0
val provider =
BlossomReadAuthTokenProvider(
signerProvider = {
fun returnsNullWhenNoSigner() =
runBlocking {
val provider = BlossomReadAuthTokenProvider({ null }, scope)
assertNull(provider.header(host))
}
@Test
fun cachedHeaderNeverSigns() =
runBlocking {
var lookups = 0
val provider =
BlossomReadAuthTokenProvider({
lookups++
signer
},
clock = { 0L },
}, scope)
// Nothing minted yet, so the pure read must miss without signing.
assertNull(provider.cachedHeader(host))
assertEquals(0, lookups)
val minted = provider.header(host)
assertEquals(minted, provider.cachedHeader(host))
}
@Test
fun cachesPerHostWithinTtl() =
runBlocking {
var lookups = 0
val provider =
BlossomReadAuthTokenProvider({
lookups++
signer
}, scope, clock = { 0L })
val first = provider.header(host)
val second = provider.header(host)
assertEquals("second call must be served from cache", first, second)
assertEquals("signer must be resolved only once for the same host", 1, lookups)
}
@Test
fun differentHostSignsSeparately() =
runBlocking {
val provider = BlossomReadAuthTokenProvider({ signer }, scope, clock = { 0L })
val a = provider.header(host)
val b = provider.header("other.example.com")
assertNotEquals(a, b)
}
@Test
fun refreshesAfterExpiry() =
runBlocking {
var now = 0L
val provider = BlossomReadAuthTokenProvider({ signer }, scope, clock = { now })
val first = provider.header(host)
now += 56L * 60L * 1000L
assertNull("token must be gone from the pure read once expired", provider.cachedHeader(host))
val second = provider.header(host)
assertNotEquals("an expired token must be re-signed", first, second)
}
@Test
fun warmPopulatesTheCacheWithoutTheCallerWaiting() =
runBlocking {
val provider = BlossomReadAuthTokenProvider({ signer }, scope)
provider.warm(host)
// warm() returns immediately; the token lands shortly after.
withTimeout(5_000) {
while (provider.cachedHeader(host) == null) {
kotlinx.coroutines.delay(5)
}
}
assertTrue(provider.cachedHeader(host)!!.startsWith("Nostr "))
}
/**
* End-to-end BUD-11 check on the token this path actually mints: reused
* across every blob on the host, so it must be `server`-scoped and carry no
* `x` tag ("When `x` tags are present, the token is only valid for
* operations on the specified blob hashes"), and be Base64url without
* padding.
*/
@Test
fun mintedTokenIsAReusableBud11GetToken() =
runBlocking {
val provider = BlossomReadAuthTokenProvider({ signer }, scope)
val token =
provider.header(host)!!.removePrefix(BlossomAuthorizationEvent.AUTH_HEADER_SCHEME)
assertTrue("token must be base64url without padding, got: $token", token.none { it == '=' || it == '+' || it == '/' })
val event = BlossomAuthorizationEvent.BASE64URL.decode(token).decodeToString()
val parsed = JacksonMapper.fromJson(event) as BlossomAuthorizationEvent
assertEquals(BlossomAuthorizationEvent.KIND, parsed.kind)
assertEquals("get", parsed.tags.first { it[0] == "t" }[1])
assertEquals(host, parsed.tags.first { it[0] == "server" }[1])
assertTrue("a host-cached token must not be blob-scoped", parsed.tags.none { it[0] == "x" })
assertTrue(
"BUD-11 requires an expiration in the future",
parsed.tags.first { it[0] == "expiration" }[1].toLong() > parsed.createdAt,
)
}
/**
* The single-flight guarantee. Before it existed the token cache was only
* populated *after* a signature returned, so a cold burst of N images from a
* gated host all missed and all signed — N signatures, and with a NIP-55
* external signer N IPC round trips.
*/
@Test
fun concurrentCallersShareOneSignature() =
runBlocking {
val slow = DelayingTestSigner(delayMs = SIGN_MS)
val provider = BlossomReadAuthTokenProvider({ slow }, scope)
val startedAt = System.nanoTime()
val results =
(1..CONCURRENT_CALLERS)
.map { async(Dispatchers.Default) { provider.header(host) } }
.awaitAll()
val elapsedMs = (System.nanoTime() - startedAt) / 1_000_000
println(
"[measure] callers=$CONCURRENT_CALLERS signatures=${slow.signatures} " +
"one-signature=${SIGN_MS}ms all-callers-done=${elapsedMs}ms",
)
val first = provider.authHeader(host, sha)
val second = provider.authHeader(host, sha)
assertEquals("exactly one signature for $CONCURRENT_CALLERS callers", 1, slow.signatures)
assertEquals("every caller must get the same token", 1, results.toSet().size)
assertTrue("first token must be non-null", results.first() != null)
assertTrue(
"$CONCURRENT_CALLERS callers should share one ~${SIGN_MS}ms signature, took ${elapsedMs}ms",
elapsedMs < SIGN_MS * 3,
)
}
assertEquals("second call must be served from cache", first, second)
assertEquals("signer must run only once for the same host", 1, lookups)
}
private companion object {
// Matches OkHttpClientFactory's maxRequestsPerHost: the worst realistic
// burst is one gated host filling every per-host dispatcher slot.
const val CONCURRENT_CALLERS = 16
@Test
fun differentHostSignsSeparately() {
val provider = BlossomReadAuthTokenProvider(signerProvider = { signer }, clock = { 0L })
val a = provider.authHeader(host, sha)
val b = provider.authHeader("other.example.com", sha)
assertNotEquals(a, b)
}
@Test
fun refreshesAfterExpiry() {
var now = 0L
val provider = BlossomReadAuthTokenProvider(signerProvider = { signer }, clock = { now })
val first = provider.authHeader(host, sha)
now += 60L * 60L * 1000L // one hour later — past the 55-min TTL
val second = provider.authHeader(host, sha)
assertNotEquals("expired token must be re-signed", first, second)
// How long one signature takes in the concurrency test.
const val SIGN_MS = 300L
}
}
@@ -0,0 +1,86 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.amethyst.service.okhttp
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.core.HexKey
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.signers.NostrSigner
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
import com.vitorpamplona.quartz.nip57Zaps.LnZapRequestEvent
import kotlinx.coroutines.delay
/**
* Signs with a real key, but only after [delayMs] — standing in for a NIP-55
* IPC round trip or a NIP-46 relay hop. Counts signatures so a test can assert
* how many actually happened.
*/
class DelayingTestSigner(
private val delayMs: Long,
private val delegate: NostrSignerInternal = NostrSignerInternal(KeyPair()),
) : NostrSigner(delegate.pubKey) {
@Volatile
var signatures = 0
private set
override fun isWriteable() = true
override suspend fun <T : Event> sign(
createdAt: Long,
kind: Int,
tags: Array<Array<String>>,
content: String,
): T {
delay(delayMs)
synchronized(this) { signatures++ }
return delegate.sign(createdAt, kind, tags, content)
}
override suspend fun nip04Encrypt(
plaintext: String,
toPublicKey: HexKey,
) = unsupported()
override suspend fun nip04Decrypt(
ciphertext: String,
fromPublicKey: HexKey,
) = unsupported()
override suspend fun nip44Encrypt(
plaintext: String,
toPublicKey: HexKey,
) = unsupported()
override suspend fun nip44Decrypt(
ciphertext: String,
fromPublicKey: HexKey,
) = unsupported()
override suspend fun decryptZapEvent(event: LnZapRequestEvent) = unsupported()
override suspend fun deriveKey(nonce: HexKey) = unsupported()
override suspend fun signPsbt(psbtHex: String) = unsupported()
override fun hasForegroundSupport() = false
private fun unsupported(): Nothing = throw UnsupportedOperationException("test signer")
}
@@ -26,14 +26,18 @@ import com.vitorpamplona.quartz.nipB7Blossom.BlossomAuthorizationEvent
object BlossomAuth {
/**
* BUD-01 read auth (`t=get`). Servers that gate downloads (e.g. Buzz's
* private media relay) require this on `GET /<sha256>`. The [servers] list
* adds BUD-11 `server` tags so a single token can be scoped to a whole host
* (which also covers derived blobs like `.thumb.jpg` whose hash differs
* from [hash]).
* BUD-11 read auth (`t=get`). Servers that gate downloads (e.g. Buzz's
* private media relay) require this on `GET /<sha256>`.
*
* [servers] adds BUD-11 `server` tags, scoping the token to those hosts.
* [hash] adds an `x` tag, scoping it to that one blob — pass null to leave
* it off, which is what makes a token reusable for every blob on the host
* (derived blobs like `.thumb.jpg` included). BUD-11 allows either for
* `GET`, but a token that carries `x` is valid *only* for that hash. See
* [BlossomAuthorizationEvent.createGetAuth].
*/
suspend fun createGetAuth(
hash: HexKey,
hash: HexKey?,
alt: String,
signer: NostrSigner,
servers: List<String> = emptyList(),
@@ -36,25 +36,48 @@ class BlossomAuthorizationEvent(
content: String,
sig: HexKey,
) : Event(id, pubKey, createdAt, KIND, tags, content, sig) {
/** Base64 of this event's JSON, as carried in the `Authorization` header value. */
fun rawToken() = Base64.encode(toJson().encodeToByteArray())
/**
* This event's JSON as Base64url without padding, per BUD-11: "the
* authorization token MUST be encoded as Base64 URL-safe without padding
* (Base64url, as used by JWTs)".
*
* Deliberately NOT the same encoder as NIP-98's
* [com.vitorpamplona.quartz.nip98HttpAuth.HTTPAuthorizationEvent.rawToken],
* which stays on standard Base64 because NIP-98 does not specify a variant.
* In practice the alphabets coincide here — a token's JSON is printable
* ASCII, and a sextet can only reach 62/63 when the third byte of its group
* is `>`, `~`, `?` or DEL — so the observable change is the dropped `=`.
*/
fun rawToken() = BASE64URL.encode(toJson().encodeToByteArray())
/**
* The full `Authorization` header value for a BUD-01/BUD-02 request:
* `Nostr <base64-event>`. Mirrors NIP-98's
* [com.vitorpamplona.quartz.nip98HttpAuth.HTTPAuthorizationEvent.toAuthToken],
* which Blossom auth reuses.
* The full `Authorization` header value for a Blossom request:
* `Nostr <base64url-event>` (BUD-11, HTTP Authorization Header).
*/
fun toAuthorizationHeader() = "$AUTH_HEADER_SCHEME${rawToken()}"
companion object {
const val KIND = 24242
/** Scheme prefix for the `Authorization` header value (BUD-01). */
/** Scheme prefix for the `Authorization` header value (BUD-11). */
const val AUTH_HEADER_SCHEME = "Nostr "
/** BUD-11's required token encoding: URL-safe alphabet, no `=` padding. */
val BASE64URL = Base64.UrlSafe.withPadding(Base64.PaddingOption.ABSENT)
/**
* BUD-11 `t=get` read authorization.
*
* [hash] is optional because BUD-11 lists the `x` tag as *optional* for
* `GET /<sha256>`, and its Tag scoping rule is strict about what adding
* one means: "When `x` tags are present, the token is only valid for
* operations on the specified blob hashes." So pass a hash only for a
* token used to fetch that one blob; pass null for a token that will be
* reused across blobs on a host, and let the `server` tag scope it.
* A hash-scoped token replayed for a different blob is invalid.
*/
suspend fun createGetAuth(
hash: HexKey,
hash: HexKey?,
alt: String,
signer: NostrSigner,
servers: List<String> = emptyList(),
@@ -23,7 +23,6 @@ package com.vitorpamplona.quartz.nipB7Blossom
import com.vitorpamplona.quartz.nip01Core.crypto.KeyPair
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerInternal
import kotlinx.coroutines.test.runTest
import kotlin.io.encoding.Base64
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
@@ -96,7 +95,66 @@ class BlossomAuthorizationEventTest {
val header = event.toAuthorizationHeader()
assertTrue(header.startsWith(BlossomAuthorizationEvent.AUTH_HEADER_SCHEME))
val decoded = Base64.decode(header.removePrefix(BlossomAuthorizationEvent.AUTH_HEADER_SCHEME)).decodeToString()
val token = header.removePrefix(BlossomAuthorizationEvent.AUTH_HEADER_SCHEME)
val decoded = BlossomAuthorizationEvent.BASE64URL.decode(token).decodeToString()
assertEquals(event.toJson(), decoded)
}
/**
* BUD-11: "the authorization token MUST be encoded as Base64 URL-safe
* without padding (Base64url, as used by JWTs)". Padded standard Base64 was
* what this produced before, so both halves are worth pinning. Several
* lengths, because whether padding appears at all depends on the JSON
* length mod 3 — a single sample passes by luck about half the time.
*/
@Test
fun authorizationTokenIsBase64UrlWithoutPadding() =
runTest {
listOf("a", "List", "List blobs", "List all of the blobs", "List blobs \u00e1\u00e9\u00ed")
.forEach { alt ->
val header = BlossomAuthorizationEvent.createListAuth(signer, alt).toAuthorizationHeader()
val token = header.removePrefix(BlossomAuthorizationEvent.AUTH_HEADER_SCHEME)
assertTrue(token.none { it == '=' }, "padding must be absent for `$alt`, got: $token")
assertTrue(
token.none { it == '+' || it == '/' },
"standard-alphabet chars must not appear for `$alt`, got: $token",
)
assertTrue(
token.all { it.isLetterOrDigit() || it == '-' || it == '_' },
"token must be base64url for `$alt`, got: $token",
)
}
}
/**
* BUD-11 lists `x` as optional for `GET /<sha256>`, and its Tag scoping rule
* makes the omission load-bearing: "When `x` tags are present, the token is
* only valid for operations on the specified blob hashes." A token cached
* per host and reused across blobs must therefore carry no `x`.
*/
@Test
fun getAuthOmitsTheBlobScopeWhenNoHashIsGiven() =
runTest {
val event =
BlossomAuthorizationEvent.createGetAuth(
hash = null,
alt = "Downloading media from cdn.example.com",
signer = signer,
servers = listOf("https://cdn.example.com"),
)
assertEquals("get", event.tags.first { it[0] == "t" }[1])
assertTrue(event.tags.none { it[0] == "x" }, "a reusable get token must not be blob-scoped")
assertEquals("cdn.example.com", event.tags.first { it[0] == "server" }[1])
}
@Test
fun getAuthKeepsTheBlobScopeWhenAHashIsGiven() =
runTest {
val event = BlossomAuthorizationEvent.createGetAuth(hash, "Downloading one blob", signer)
assertEquals("get", event.tags.first { it[0] == "t" }[1])
assertEquals(hash, event.tags.first { it[0] == "x" }[1])
}
}