commit a160446c6caa0bf12a69c4f70ee3e0f6e140f45c from: mtmn date: Sun Aug 9 09:39:07 2026 UTC feat: add timeouts and metadata improvements commit - 38217ef3c221c9f7763be487075f91ec7d3755f0 commit + a160446c6caa0bf12a69c4f70ee3e0f6e140f45c blob - e9bb2af8657ebf2f0ca58c1ecc990f3460ae65d3 blob + 6d2e424a29701f96420eb7c4aaaad1ccac9f904a --- src/Cosine.purs +++ src/Cosine.purs @@ -19,6 +19,7 @@ import Effect.Exception as Exception import Fetch (fetch, Method(GET)) import Foreign.Object as Object import JSURI (encodeURIComponent) +import Http (apiTimeout, withTimeout) import Log as Log import Metrics as Metrics import Node.Encoding (Encoding(UTF8)) @@ -46,7 +47,7 @@ fetchCosineSimilar slug cfg query = do let headers = { "User-Agent": "corpus/1.0 +https://sr.ht/~mtmn/corpus", "Authorization": "Bearer " <> apiKey } let searchUrl = "https://cosine.club/api/v1/search?q=" <> (fromMaybe "" $ encodeURIComponent query) <> "&limit=1" Log.info $ "Cosine Club: searching for: " <> query - searchResult <- try $ fetch searchUrl { method: GET, headers: headers } + searchResult <- try $ withTimeout apiTimeout "Cosine Club search" $ fetch searchUrl { method: GET, headers: headers } case searchResult of Left err -> do liftEffect $ Metrics.incCosineRequest slug "error" @@ -61,7 +62,7 @@ fetchCosineSimilar slug cfg query = do liftEffect $ Metrics.incCosineRequest slug "error" throwError (Exception.error "Search API error") else do - searchBody <- searchRes.text + searchBody <- withTimeout apiTimeout "Cosine Club search response" searchRes.text let mTrackId = do json <- hush $ parseJson searchBody @@ -78,7 +79,7 @@ fetchCosineSimilar slug cfg query = do Just trackId -> do let similarUrl = "https://cosine.club/api/v1/tracks/" <> trackId <> "/similar?limit=10" Log.info $ "Cosine Club: fetching similar for ID " <> trackId - similarResult <- try $ fetch similarUrl { method: GET, headers: headers } + similarResult <- try $ withTimeout apiTimeout "Cosine Club similar lookup" $ fetch similarUrl { method: GET, headers: headers } case similarResult of Left err -> do liftEffect $ Metrics.incCosineRequest slug "error" @@ -94,7 +95,7 @@ fetchCosineSimilar slug cfg query = do throwError (Exception.error "Similar API error") else do liftEffect $ Metrics.incCosineRequest slug "success" - similarRes.text + withTimeout apiTimeout "Cosine Club similar response" similarRes.text serveSimilar :: (Response -> String -> Effect Unit) blob - 699c1e8ca1c24a9f1533766ddf24ee97bf73ae90 blob + ce457b60afff7dd07f43ed8a04ea99c11c599986 --- src/Cover.purs +++ src/Cover.purs @@ -1,6 +1,9 @@ module Cover ( CoverSource , sanitizeKey + , cacheJobKey + , tryStartCacheFill + , finishCacheFill , fetchLastfmCoverUrl , fetchDiscogsCoverUrl , coverSources @@ -14,9 +17,9 @@ import Config (UserConfig, s3ConfigFromUser) import S3 (existsInS3, getPresignedUrl, uploadToS3) import Data.Array ((!!), find) import Data.Either (Either(..), hush) -import Data.Foldable (foldM) +import Data.Foldable (foldM, for_) import Data.Map as Map -import Data.Maybe (Maybe(..), fromMaybe) +import Data.Maybe (Maybe(..), fromMaybe, isJust) import Data.String.CaseInsensitive (CaseInsensitiveString(..)) import Data.String.Regex (Regex, regex, replace, parseFlags) import Effect (Effect) @@ -27,7 +30,8 @@ import Fetch (fetch, Method(GET)) import Fetch.Argonaut.Json (fromJson) import Foreign.Object as Object import Data.Argonaut.Core (toArray, toObject, toString, toBoolean) -import Image (convertToAvif) +import Image (arrayBufferByteLength, convertToAvif) +import Http (apiTimeout, imageTimeout, withTimeout) import JSURI (encodeURIComponent) import Log as Log import Metrics as Metrics @@ -40,6 +44,15 @@ import Web.URL (URL) import Web.URL as URL import Web.URL.URLSearchParams as URLSearchParams +foreign import tryStartCacheFillImpl :: String -> Effect Boolean +foreign import finishCacheFillImpl :: String -> Effect Unit + +tryStartCacheFill :: String -> Effect Boolean +tryStartCacheFill = tryStartCacheFillImpl + +finishCacheFill :: String -> Effect Unit +finishCacheFill = finishCacheFillImpl + type Response = ServerResponse type CoverSource = @@ -62,6 +75,12 @@ sanitizeKeyRe1 = hush $ regex "[^a-z0-9.-]" (parseFlag sanitizeKeyRe2 :: Maybe Regex sanitizeKeyRe2 = hush $ regex "_{2,}" (parseFlags "g") +cacheJobKey :: String -> String -> String +cacheJobKey bucket s3Key = bucket <> ":" <> s3Key + +maxCoverBytes :: Number +maxCoverBytes = 10_000_000.0 + getQueryParam :: String -> URL -> Maybe String getQueryParam key url = URLSearchParams.get key (URL.searchParams url) @@ -69,7 +88,7 @@ fetchCaaCoverUrl :: String -> Aff (Maybe String) fetchCaaCoverUrl mbid = do let url = "https://coverartarchive.org/release/" <> mbid let headers = { "User-Agent": "corpus/1.0 (+https://sr.ht/~mtmn/corpus)" } - result <- try $ fetch url { method: GET, headers } + result <- try $ withTimeout apiTimeout "Cover Art Archive lookup" $ fetch url { method: GET, headers } case result of Right fr | fr.status == 200 -> do json <- fromJson fr.json @@ -104,7 +123,7 @@ fetchLastfmCoverUrl cfg artist release = case cfg.last <> k Log.info $ "Searching Last.fm for: " <> artist <> " - " <> release let headers = { "User-Agent": "corpus/1.0 (+https://sr.ht/~mtmn/corpus)" } - result <- try $ fetch searchUrl { method: GET, headers } + result <- try $ withTimeout apiTimeout "Last.fm cover lookup" $ fetch searchUrl { method: GET, headers } case result of Right fr | fr.status == 200 -> do json <- fromJson fr.json @@ -129,7 +148,7 @@ fetchDiscogsCoverUrl cfg artist release = case cfg.dis <> (fromMaybe "" $ encodeURIComponent queryStr) <> "&type=release&per_page=1" Log.info $ "Searching Discogs for: " <> queryStr - result <- try $ fetch searchUrl { method: GET, headers: { "User-Agent": "corpus/1.0 (+https://sr.ht/~mtmn/corpus)", "Authorization": "Discogs token=" <> t } } + result <- try $ withTimeout apiTimeout "Discogs cover lookup" $ fetch searchUrl { method: GET, headers: { "User-Agent": "corpus/1.0 (+https://sr.ht/~mtmn/corpus)", "Authorization": "Discogs token=" <> t } } case result of Right fr | fr.status == 200 -> do json <- fromJson fr.json @@ -177,71 +196,93 @@ serveCover serveNotFound cfg slug url res = do artist = fromMaybe "" (getQueryParam "artist" url) release = fromMaybe "" (getQueryParam "release" url) s3cfg = s3ConfigFromUser cfg + cacheEnabled = cfg.coverCacheEnabled && isJust s3cfg.bucket - served <- foldM (trySource s3cfg) false (coverSources mbid artist release cfg) + served <- foldM (trySource s3cfg cacheEnabled) false (coverSources mbid artist release cfg) unless served $ liftEffect $ serveNotFound res where - trySource _ true _ = pure true - trySource s3cfg false { name, s3Key, findUrl } = do - cached <- checkS3 s3cfg s3Key + trySource _ _ true _ = pure true + trySource s3cfg canCache false { name, s3Key, findUrl } = do + cached <- checkS3 canCache s3cfg s3Key if cached then do Log.info $ "Serving " <> name <> " cover from cache: " <> s3Key liftEffect $ Metrics.incCoverRequest slug name "s3_hit" - serveS3Redirect s3cfg s3Key res - pure true + presigned <- try $ getPresignedUrl s3cfg s3Key + case presigned of + Right presignedUrl -> do + serveRedirect presignedUrl Nothing res + pure true + Left err -> do + Log.warn $ "Failed to sign cached cover " <> s3Key <> ": " <> Exception.message err + tryUpstream false else do - mUrl <- findUrl - case mUrl of - Nothing -> + tryUpstream canCache + + where + tryUpstream shouldCache = do + found <- try findUrl + case found of + Left err -> do + Log.warn $ "Cover lookup failed for " <> name <> ": " <> Exception.message err pure false - Just urlStr -> do - redirectAndCache s3cfg urlStr s3Key res + Right Nothing -> + pure false + Right (Just urlStr) -> do + serveRedirect urlStr (Just "public, max-age=3600") res liftEffect $ Metrics.incCoverRequest slug name "fetch" + when shouldCache $ startCacheFill s3cfg urlStr s3Key pure true - checkS3 s3cfg s3Key - | not cfg.coverCacheEnabled = pure false + checkS3 canCache s3cfg s3Key + | not canCache = pure false | otherwise = do result <- try $ existsInS3 s3cfg s3Key pure $ case result of Right b -> b Left _ -> false - serveS3Redirect s3cfg s3Key response = do - presignedUrl <- getPresignedUrl s3cfg s3Key + serveRedirect redirectUrl mCacheControl response = liftEffect $ do setStatusCode 302 response - setHeader "Location" presignedUrl (toOutgoingMessage response) + setHeader "Location" redirectUrl (toOutgoingMessage response) + for_ mCacheControl \cacheControl -> + setHeader "Cache-Control" cacheControl (toOutgoingMessage response) end (toWriteable (toOutgoingMessage response)) - -- Redirect the client immediately to the upstream URL, then fetch+convert+cache in background. - -- This avoids blocking the response on AVIF conversion (which can take hundreds of ms). - redirectAndCache s3cfg urlStr s3Key response = do - liftEffect $ do - setStatusCode 302 response - setHeader "Location" urlStr (toOutgoingMessage response) - setHeader "Cache-Control" "public, max-age=3600" (toOutgoingMessage response) - end (toWriteable (toOutgoingMessage response)) - when cfg.coverCacheEnabled $ void $ forkAff do - let headers = { "User-Agent": "corpus/1.0 (+https://sr.ht/~mtmn/corpus)" } - fetchResult <- try $ fetch urlStr { method: GET, headers } - case fetchResult of - Right fr | fr.status == 200 -> do - let - contentType = Map.lookup (CaseInsensitiveString "content-type") fr.headers - isAvif = case contentType of - Just ct | ct == "image/avif" -> true - _ -> false - Log.info $ "Caching image: " <> urlStr - ab <- fr.arrayBuffer - avifAb <- if isAvif then pure ab else convertToAvif ab - avifBuf <- liftEffect $ fromArrayBuffer avifAb - uploadResult <- try $ uploadToS3 s3cfg s3Key avifBuf "image/avif" - case uploadResult of - Right _ -> Log.info $ "Uploaded " <> s3Key - Left err -> Log.error $ "Failed to upload " <> s3Key <> ": " <> Exception.message err - Right fr -> - Log.warn $ "Background fetch failed for " <> urlStr <> " with status " <> show fr.status - Left err -> - Log.error $ "Background fetch error for " <> urlStr <> ": " <> Exception.message err + -- Only one request may populate a given S3 object at a time. Concurrent + -- misses still redirect immediately; later requests use the cached object. + startCacheFill s3cfg urlStr s3Key = do + let jobKey = cacheJobKey (fromMaybe "" s3cfg.bucket) s3Key + started <- liftEffect $ tryStartCacheFill jobKey + when started $ void $ forkAff do + _ <- try do + let headers = { "User-Agent": "corpus/1.0 (+https://sr.ht/~mtmn/corpus)" } + fetchResult <- try $ withTimeout imageTimeout "Cover image request" $ fetch urlStr { method: GET, headers } + case fetchResult of + Right fr | fr.status == 200 -> do + let + contentType = Map.lookup (CaseInsensitiveString "content-type") fr.headers + isAvif = case contentType of + Just ct | ct == "image/avif" -> true + _ -> false + Log.info $ "Caching image: " <> urlStr + bodyResult <- try $ withTimeout imageTimeout "Cover image download" fr.arrayBuffer + case bodyResult of + Left err -> Log.error $ "Cover image download failed for " <> urlStr <> ": " <> Exception.message err + Right ab -> do + bytes <- liftEffect $ arrayBufferByteLength ab + if bytes > maxCoverBytes then + Log.warn $ "Skipping oversized cover (" <> show bytes <> " bytes): " <> urlStr + else do + avifAb <- if isAvif then pure ab else convertToAvif ab + avifBuf <- liftEffect $ fromArrayBuffer avifAb + uploadResult <- try $ uploadToS3 s3cfg s3Key avifBuf "image/avif" + case uploadResult of + Right _ -> Log.info $ "Uploaded " <> s3Key + Left err -> Log.error $ "Failed to upload " <> s3Key <> ": " <> Exception.message err + Right fr -> + Log.warn $ "Background fetch failed for " <> urlStr <> " with status " <> show fr.status + Left err -> + Log.error $ "Background fetch error for " <> urlStr <> ": " <> Exception.message err + liftEffect $ finishCacheFill jobKey blob - /dev/null blob + c55141c712da902d1cd6af1bdf6588d2339d14e5 (mode 644) --- /dev/null +++ src/Cover.js @@ -0,0 +1,13 @@ +const activeCacheFills = new Set(); +const maxConcurrentCacheFills = 4; + +export const tryStartCacheFillImpl = (key) => () => { + if (activeCacheFills.has(key)) return false; + if (activeCacheFills.size >= maxConcurrentCacheFills) return false; + activeCacheFills.add(key); + return true; +}; + +export const finishCacheFillImpl = (key) => () => { + activeCacheFills.delete(key); +}; blob - 6f66406c1dd8beb44ca5ccbf6a1b9d057321904b blob + 281ad4ed71931c1828fe60de91b96b894973cca2 --- src/Image.js +++ src/Image.js @@ -2,7 +2,7 @@ import sharp from "sharp"; export const convertToAvifImpl = (buffer) => () => new Promise((resolve, reject) => - sharp(buffer) + sharp(buffer, { limitInputPixels: 40_000_000, failOn: "error" }) .avif() .toBuffer((err, data) => { if (err) return reject(err); @@ -11,3 +11,5 @@ export const convertToAvifImpl = (buffer) => () => ); }), ); + +export const arrayBufferByteLengthImpl = (buffer) => () => buffer.byteLength; blob - 303ae4adcd4814fe0dade35640fcd99992f6cf82 blob + a16e514877a824d82e83e27ad342da0773ffa37e --- src/Image.purs +++ src/Image.purs @@ -1,4 +1,4 @@ -module Image (convertToAvif) where +module Image (arrayBufferByteLength, convertToAvif) where import Prelude import Control.Promise (Promise, toAffE) @@ -7,6 +7,10 @@ import Effect.Aff (Aff) import Data.ArrayBuffer.Types (ArrayBuffer) foreign import convertToAvifImpl :: ArrayBuffer -> Effect (Promise ArrayBuffer) +foreign import arrayBufferByteLengthImpl :: ArrayBuffer -> Effect Number convertToAvif :: ArrayBuffer -> Aff ArrayBuffer convertToAvif = toAffE <<< convertToAvifImpl + +arrayBufferByteLength :: ArrayBuffer -> Effect Number +arrayBufferByteLength = arrayBufferByteLengthImpl blob - /dev/null blob + e2c99105dfc0dcbd6b5d63e92b850b9760be1ddc (mode 644) --- /dev/null +++ src/Http.purs @@ -0,0 +1,33 @@ +module Http + ( apiTimeout + , imageTimeout + , withTimeout + ) where + +import Prelude + +import Control.Alt ((<|>)) +import Control.Monad.Error.Class (throwError) +import Control.Parallel (parallel, sequential) +import Data.Either (Either(..)) +import Data.Time.Duration (Milliseconds(..)) +import Effect.Aff (Aff, delay) +import Effect.Exception (error) + +-- External API calls should fail promptly so the caller can use its fallback +-- or retry policy. Cancelling Fetch's Aff aborts its underlying request. +apiTimeout :: Milliseconds +apiTimeout = Milliseconds 10_000.0 + +-- Image responses are larger than API payloads, but must still be bounded. +imageTimeout :: Milliseconds +imageTimeout = Milliseconds 20_000.0 + +withTimeout :: forall a. Milliseconds -> String -> Aff a -> Aff a +withTimeout timeout label action = do + result <- sequential $ + parallel (Right <$> action) + <|> parallel (delay timeout *> pure (Left $ error $ label <> " timed out")) + case result of + Left err -> throwError err + Right value -> pure value blob - 46b5aea6a228edaa63fa95609bf10af96a4e0d49 blob + 3df3d5ddc7ad951e6b39d5a6a815ce30dbee0c91 --- src/Main.purs +++ src/Main.purs @@ -234,7 +234,9 @@ serveValidateToken contextsParam reqParam resParam = d -- Extract a ListenBrainz API token from an `Authorization: Token ` header value. parseAuthToken :: Maybe String -> Maybe String -parseAuthToken mAuth = mAuth >>= stripPrefix (Pattern "Token ") +parseAuthToken mAuth = do + token <- mAuth >>= stripPrefix (Pattern "Token ") + if token == "" then Nothing else Just token -- Build the `validate-token` response body. `Just displayName` means the token -- resolved to a user (valid); `Nothing` means the token is unknown (invalid). @@ -276,7 +278,9 @@ serveConflict res message = respond "text/plain" 409 m -- Extract an admin secret from an `Authorization: Bearer ` header value. parseBearer :: Maybe String -> Maybe String -parseBearer mAuth = mAuth >>= stripPrefix (Pattern "Bearer ") +parseBearer mAuth = do + token <- mAuth >>= stripPrefix (Pattern "Bearer ") + if token == "" then Nothing else Just token -- Gates admin endpoints behind ADMIN_TOKEN: unset -> 404 (disabled), mismatch -> 401. withAdmin :: ServerEnv -> Request -> Response -> Aff Unit -> Aff Unit blob - 4f3adff5afee9b0baf4185d123d2c1c0c2623c57 blob + 3400d5f40aa7df418812239b9f8cfbbf27b3385c --- src/Metadata.purs +++ src/Metadata.purs @@ -29,6 +29,7 @@ import Fetch (fetch, Method(GET)) import Fetch.Argonaut.Json (fromJson) import Foreign.Object as Object import JSURI (encodeURIComponent) +import Http (apiTimeout, withTimeout) import Log as Log import Metrics as Metrics @@ -43,7 +44,7 @@ type GenreSource = fetchMusicBrainzRelease :: String -> Aff (Maybe MbData) fetchMusicBrainzRelease mbid = do let url = "https://musicbrainz.org/ws/2/release/" <> mbid <> "?inc=genres+labels+release-groups&fmt=json" - result <- try $ fetch url { method: GET, headers: { "User-Agent": "corpus/1.0 +https://sr.ht/~mtmn/corpus" } } + result <- try $ withTimeout apiTimeout "MusicBrainz metadata lookup" $ fetch url { method: GET, headers: { "User-Agent": "corpus/1.0 +https://sr.ht/~mtmn/corpus" } } case result of Left err -> do Log.error $ "MusicBrainz fetch error for " <> mbid <> ": " <> Exception.message err @@ -91,7 +92,7 @@ fetchLastfmGenre Nothing _ _ = do fetchLastfmGenre (Just k) artist release = do let searchUrl = "https://ws.audioscrobbler.com/2.0/?method=album.getinfo&artist=" <> (fromMaybe "" $ encodeURIComponent artist) <> "&album=" <> (fromMaybe "" $ encodeURIComponent release) <> "&format=json" <> "&api_key=" <> k Log.info $ "Fetching Last.fm genre for: " <> artist <> " - " <> release - result <- try $ fetch searchUrl { method: GET } + result <- try $ withTimeout apiTimeout "Last.fm genre lookup" $ fetch searchUrl { method: GET } case result of Right fetchRes | fetchRes.status == 200 -> do jsonResult <- try $ fromJson fetchRes.json @@ -122,7 +123,7 @@ fetchDiscogsGenre (Just t) artist release = do let queryStr = artist <> " " <> release let searchUrl = "https://api.discogs.com/database/search?q=" <> (fromMaybe "" $ encodeURIComponent queryStr) <> "&type=release&per_page=1" Log.info $ "Fetching Discogs genre for: " <> queryStr - result <- try $ fetch searchUrl { method: GET, headers: { "User-Agent": "corpus/1.0 +https://sr.ht/~mtmn/corpus", "Authorization": "Discogs token=" <> t } } + result <- try $ withTimeout apiTimeout "Discogs genre lookup" $ fetch searchUrl { method: GET, headers: { "User-Agent": "corpus/1.0 +https://sr.ht/~mtmn/corpus", "Authorization": "Discogs token=" <> t } } case result of Right fetchRes | fetchRes.status == 200 -> do jsonResult <- try $ fromJson fetchRes.json @@ -179,8 +180,10 @@ enrichMetadata conn cfg slug = forever do liftEffect $ Metrics.incEnrichmentFetch slug "musicbrainz" "error" Right Nothing -> do liftEffect $ Metrics.incEnrichmentFetch slug "musicbrainz" "retry" - upsertReleaseMetadata conn mbid Nothing Nothing Nothing - touchGenreCheckedAt conn mbid + -- Nothing represents a transient request failure. Do not create an + -- empty metadata row or mark the release checked: both would hide it + -- from the enrichment queue and turn a retry into a week-long skip. + Log.warn $ "Deferring metadata enrichment for " <> mbid <> " after transient MusicBrainz failure" Right (Just mbdata) -> do liftEffect $ Metrics.incEnrichmentFetch slug "musicbrainz" "success" case mbdata.genre of blob - dc1b000dce1a6e0c3885056f80ef6400f6aa4a52 blob + 444a12315483446cb27475bcb291225cb594ab56 --- src/Sync.purs +++ src/Sync.purs @@ -21,28 +21,18 @@ import Data.Maybe (Maybe(..), fromMaybe) import Data.String (Pattern(..), stripPrefix) import Data.Time.Duration (Milliseconds(..)) import Db (Connection, checkExists, getOldestTs, upsertScrobble, withTransaction) -import Effect.Aff (Aff, delay, launchAff_, makeAff, nonCanceler, throwError, try) +import Effect.Aff (Aff, delay, throwError, try) import Effect.Aff.AVar (AVar) import Effect.Aff.Retry (RetryStatus(..), exponentialBackoff, limitRetries, recovering) -import Effect (Effect) import Effect.Class (liftEffect) -import Effect.Exception (Error, error, message) -import Effect.Ref as Ref -import Effect.Uncurried (mkEffectFn1) +import Effect.Exception (error, message) import Fetch (fetch, Method(GET)) import Fetch.Argonaut.Json (fromJson) import JSURI (encodeURIComponent) +import Http (apiTimeout, withTimeout) import Log as Log import Metrics as Metrics -import Node.EventEmitter (EventHandle(..), on_) -import Node.HTTP.ClientRequest as Client -import Node.HTTP.IncomingMessage as IM -import Node.HTTPS as HTTPS -import Node.Stream.Aff (readableToStringUtf8) import Types (Listen(..), ListenBrainzResponse(..), MbidMapping(..), Payload(..), TrackMetadata(..), LastfmResponse(..), LastfmRecentTracks(..), LastfmAttr(..), LastfmTracks(..), LastfmArtist(..), LastfmAlbum(..), LastfmDate(..), LastfmTrack(..)) --- | Cast required due to a gap in the node-http FFI bindings: --- | the request object lacks an "error" event emitter type. -import Unsafe.Coerce (unsafeCoerce) -- | Result of processing a batch of listens/scrobbles. type SyncResult = @@ -71,11 +61,6 @@ processTracks conn listens trackMinTs = do Log.warn "Skipping scrobble without timestamp" pure s --- | Wraps unsafeCoerce for attaching error handlers to HTTP requests. --- | This is the only legitimate use of unsafeCoerce in this module. -castRequestForError :: forall a. a -> a -castRequestForError = unsafeCoerce - listenBrainzUrl :: String -> String listenBrainzUrl username = "https://api.listenbrainz.org/1/user/" <> username <> "/listens" @@ -101,44 +86,14 @@ withRetry label action = recovering policy [ \_ err -> Nothing -> false fetchListenBrainzUrl :: String -> Aff String -fetchListenBrainzUrl url = withRetry "ListenBrainz fetch" $ makeAff \callback -> do - -- One-shot gate: the response handler and the error handler both race to - -- resolve this Aff. The first invocation wins; any subsequent (late) event - -- is logged so silent socket errors after a successful response aren't lost. - done <- Ref.new false - let - once :: Either Error String -> Effect Unit - once result = do - already <- Ref.read done - if already then - case result of - Left err -> - Log.warn $ "ListenBrainz: dropped late error after response: " <> message err - Right _ -> - pure unit - else do - Ref.write true done - callback result +fetchListenBrainzUrl url = withRetry "ListenBrainz fetch" do + let headers = { "User-Agent": "corpus/1.0 (+https://sr.ht/~mtmn/corpus)" } + fr <- withTimeout apiTimeout "ListenBrainz fetch" $ fetch url { method: GET, headers } + if fr.status == 200 then + withTimeout apiTimeout "ListenBrainz response" fr.text + else + throwError $ error $ "ListenBrainz API returned status " <> show fr.status - req <- HTTPS.get url - - req # on_ Client.responseH \res -> do - launchAff_ do - result <- try $ readableToStringUtf8 (IM.toReadable res) - liftEffect $ case result of - Left err -> - once (Left err) - Right body -> - if IM.statusCode res == 200 then - once (Right body) - else - once (Left $ error $ "ListenBrainz API returned status " <> show (IM.statusCode res)) - - let errorH = EventHandle "error" mkEffectFn1 - on_ errorH (\err -> once (Left err)) (castRequestForError req) - - pure nonCanceler - fetchLastfmPage :: String -> String -> Int -> Maybe Int -> Aff { tracks :: Array Json, totalPages :: Int } fetchLastfmPage apiKey lfmUser page mTo = withRetry "Last.fm fetch" do let @@ -152,7 +107,7 @@ fetchLastfmPage apiKey lfmUser page mTo = withRetry "L <> toParam url = baseUrl <> "&api_key=" <> apiKey let headers = { "User-Agent": "corpus/1.0 (+https://sr.ht/~mtmn/corpus)" } - fr <- fetch url { method: GET, headers } + fr <- withTimeout apiTimeout "Last.fm sync fetch" $ fetch url { method: GET, headers } if fr.status == 200 then do json <- fromJson fr.json case parseLastfmResponse json of blob - 91527af7dfb631601a1f91925bf4cbd314125b64 blob + cfabe57c38ef9cc81526584ec02d3955c098cf55 --- test/Main.purs +++ test/Main.purs @@ -7,7 +7,11 @@ import Data.Array (length) import Data.Either (Either(..), isRight) import Data.Maybe (Maybe(..)) import Effect (Effect) +import Effect.Aff (delay, try) import Effect.Aff.AVar as Avar +import Effect.Class (liftEffect) +import Effect.Exception (message) +import Data.Time.Duration (Milliseconds(..)) import Test.Spec (describe, it) import Test.Spec.Assertions (shouldEqual, fail) import Data.String.Regex (regex, parseFlags) @@ -19,12 +23,19 @@ import Data.Argonaut.Core (Json, toBoolean, toNumber, import Foreign.Object as Object import Main (submitListenToListen, findUserByToken, sanitizeDate, parseAuthToken, parseBearer, validateTokenJson) import Registrations (getById, initRegistrations, insertRegistration, isReservedSlug, listByStatus, setStatus, slugTaken, validSlugFormat) -import Cover (sanitizeKey) +import Cover (sanitizeKey, cacheJobKey, coverSources, finishCacheFill, tryStartCacheFill) import Sync (listenBrainzUrl, lastfmTrackToListen, parseLastfmResponse) +import Http (withTimeout) main :: Effect Unit main = runSpecAndExitProcess [ consoleReporter ] do describe "Corpus Main Utils" do + it "cancels operations that exceed their fetch timeout" do + result <- try $ withTimeout (Milliseconds 1.0) "test request" (delay (Milliseconds 20.0)) + case result of + Left err -> message err `shouldEqual` "test request timed out" + Right _ -> fail "Expected timeout" + it "should build ListenBrainz URLs correctly" do listenBrainzUrl "user1" `shouldEqual` "https://api.listenbrainz.org/1/user/user1/listens" @@ -49,6 +60,70 @@ main = runSpecAndExitProcess [ consoleReporter ] do sanitizeKey "a...b" `shouldEqual` "a...b" sanitizeKey "UPPER lower" `shouldEqual` "UPPER_lower" + it "scopes concurrent cover-cache jobs to their S3 bucket and key" do + cacheJobKey "covers-a" "covers/caa/release.avif" `shouldEqual` "covers-a:covers/caa/release.avif" + cacheJobKey "covers-b" "covers/caa/release.avif" `shouldEqual` "covers-b:covers/caa/release.avif" + + it "coalesces duplicate cover-cache jobs and releases their key on completion" do + let key = cacheJobKey "test-bucket" "covers/caa/release.avif" + first <- liftEffect $ tryStartCacheFill key + second <- liftEffect $ tryStartCacheFill key + liftEffect $ finishCacheFill key + third <- liftEffect $ tryStartCacheFill key + liftEffect $ finishCacheFill key + first `shouldEqual` true + second `shouldEqual` false + third `shouldEqual` true + + it "limits background cover-cache fills across distinct keys" do + one <- liftEffect $ tryStartCacheFill "capacity-1" + two <- liftEffect $ tryStartCacheFill "capacity-2" + three <- liftEffect $ tryStartCacheFill "capacity-3" + four <- liftEffect $ tryStartCacheFill "capacity-4" + five <- liftEffect $ tryStartCacheFill "capacity-5" + liftEffect $ finishCacheFill "capacity-1" + liftEffect $ finishCacheFill "capacity-2" + liftEffect $ finishCacheFill "capacity-3" + liftEffect $ finishCacheFill "capacity-4" + one `shouldEqual` true + two `shouldEqual` true + three `shouldEqual` true + four `shouldEqual` true + five `shouldEqual` false + + describe "cover sources" do + let + coverConfig = + { listenbrainzUser: Nothing + , lastfmUser: Nothing + , lastfmApiKey: Nothing + , discogsToken: Nothing + , cosineApiKey: Nothing + , databaseFile: "" + , s3Bucket: Nothing + , s3Region: "" + , awsAccessKeyId: Nothing + , awsSecretAccessKey: Nothing + , awsEndpointUrl: Nothing + , awsS3AddressingStyle: Nothing + , coverCacheEnabled: false + , backupEnabled: false + , backupIntervalHours: 0 + } + + it "prefers CAA and uses source-specific cache keys" do + let sources = coverSources "release/id" "The Artist" "The Release" coverConfig + map _.name sources `shouldEqual` [ "caa", "discogs", "lastfm" ] + map _.s3Key sources `shouldEqual` + [ "covers/caa/release_id.avif" + , "covers/discogs/The_Artist-The_Release.avif" + , "covers/lastfm/The_Artist-The_Release.avif" + ] + + it "omits CAA when no release MBID is available" do + let sources = coverSources "" "The Artist" "The Release" coverConfig + map _.name sources `shouldEqual` [ "discogs", "lastfm" ] + describe "fromString" do it "maps all valid field names" do fromString "artist" `shouldEqual` Just FilterArtist @@ -56,6 +131,7 @@ main = runSpecAndExitProcess [ consoleReporter ] do fromString "label" `shouldEqual` Just FilterLabel fromString "year" `shouldEqual` Just FilterYear fromString "genre" `shouldEqual` Just FilterGenre + fromString "track" `shouldEqual` Just FilterTrack it "returns Nothing for unknown or empty input" do fromString "unknown" `shouldEqual` Nothing @@ -200,6 +276,7 @@ main = runSpecAndExitProcess [ consoleReporter ] do parseAuthToken Nothing `shouldEqual` Nothing parseAuthToken (Just "Bearer abc-123") `shouldEqual` Nothing parseAuthToken (Just "token abc-123") `shouldEqual` Nothing + parseAuthToken (Just "Token ") `shouldEqual` Nothing it "builds a valid-token response with the user name" do let body = validateTokenJson (Just "User One") @@ -548,6 +625,33 @@ main = runSpecAndExitProcess [ consoleReporter ] do listensNone <- getScrobbles conn 10 0 (Just { field: FilterYear, value: "1999" }) Nothing length listensNone `shouldEqual` 0 + it "filters by track and searches across track, artist, release, and label" do + conn <- connect ":memory:" + initDb conn + initReleaseMetadata conn + upsertScrobble conn + ( Listen + { listenedAt: Just 300 + , trackMetadata: TrackMetadata + { trackName: Just "Neon Skyline" + , artistName: Just "Andy Shauf" + , releaseName: Just "The Neon Skyline" + , mbidMapping: Just (MbidMapping { releaseMbid: Just "search-mbid", caaReleaseMbid: Nothing }) + , genre: Nothing + , label: Nothing + } + } + ) + upsertReleaseMetadata conn "search-mbid" Nothing (Just "Anti-") (Just 2020) + exact <- getScrobbles conn 10 0 (Just { field: FilterTrack, value: "Neon Skyline" }) Nothing + byArtist <- getScrobbles conn 10 0 Nothing (Just "andy") + byRelease <- getScrobbles conn 10 0 Nothing (Just "the neon") + byLabel <- getScrobbles conn 10 0 Nothing (Just "anti-") + length exact `shouldEqual` 1 + length byArtist `shouldEqual` 1 + length byRelease `shouldEqual` 1 + length byLabel `shouldEqual` 1 + describe "getOldestTs" do let listenAt ts = Listen @@ -774,6 +878,11 @@ main = runSpecAndExitProcess [ consoleReporter ] do Nothing -> do fail "Should have parsed numeric totalPages" + it "rejects malformed responses" do + case parseLastfmResponse (parseTrack "{ \"recenttracks\": {} }") of + Nothing -> pure unit + Just _ -> fail "Malformed response should not parse" + describe "lastfmTrackToListen" do it "parses a valid track with MBID" do let @@ -897,6 +1006,7 @@ main = runSpecAndExitProcess [ consoleReporter ] do parseBearer (Just "Bearer secret") `shouldEqual` Just "secret" parseBearer (Just "Token secret") `shouldEqual` Nothing parseBearer Nothing `shouldEqual` Nothing + parseBearer (Just "Bearer ") `shouldEqual` Nothing describe "Registrations" do it "validates slug format" do