speedup full scanning

modules/astra-library-scanner/android/src/androidTest/java/expo/modules/astralibraryscanner/data/LocalScanFastPathTest.kt

@@ -4,17 +4,124 @@ import android.content.Context
 import androidx.test.core.app.ApplicationProvider
 import androidx.test.ext.junit.runners.AndroidJUnit4
 import androidx.test.filters.LargeTest
+import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicInteger
 import kotlinx.coroutines.runBlocking
 import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
 import org.junit.Assert.assertNotEquals
 import org.junit.Assert.assertNotNull
+import org.junit.Assert.assertTrue
+import org.junit.Assert.fail
 import org.junit.Test
 import org.junit.runner.RunWith
 
 @LargeTest
 @RunWith(AndroidJUnit4::class)
 class LocalScanFastPathTest {
+  @Test
+  fun fullScanFlushesOrderedWindowsAndRetainsCatalogOnCancellationOrFailure() = runBlocking {
+    val context = ApplicationProvider.getApplicationContext<Context>()
+    val repository = AstraLibraryRepository.get(context)
+    val suffix = System.nanoTime().toString()
+    val treeUri =
+      "content://com.android.externalstorage.documents/tree/primary%3AMusic%2Fpipeline-$suffix"
+    val parentUri =
+      "$treeUri/document/primary%3AMusic%2Fpipeline-$suffix"
+    val files = (0 until 101).map { index ->
+      file(
+        uri = "$parentUri%2Ftrack-${index.toString().padStart(3, '0')}.flac",
+        parentUri = parentUri,
+        size = 1_024L + index,
+        mtime = 100L + index,
+      )
+    }
+    val folder = repository.registerFolder(treeUri, "Pipeline scan test")
+    val folderId = (folder.getValue("id") as Number).toLong()
+
+    try {
+      val firstProgress = mutableListOf<Int>()
+      val first = repository.scanLocalFolder(
+        folderId = folderId,
+        full = true,
+        discover = { files },
+        extract = { discovered -> metadata(discovered.name) },
+        onProgress = { phase, processed, _, _ ->
+          if (phase == "extracting") firstProgress += processed
+        },
+      )
+      assertEquals(101, first.added)
+      assertEquals(0, first.updated)
+      assertEquals(0, first.removed)
+      assertEquals(101, first.total)
+      assertEquals(listOf(96, 101), firstProgress)
+      assertTrue(firstProgress.zipWithNext().all { (before, after) -> before < after })
+      val firstTitles = files.map { discovered ->
+        repository.getTrack(discovered.uri)?.get("title")
+      }
+      assertEquals(files.map(LocalAudioFile::name), firstTitles)
+
+      val secondProgress = mutableListOf<Int>()
+      val second = repository.scanLocalFolder(
+        folderId = folderId,
+        full = true,
+        discover = { files },
+        extract = { discovered -> metadata(discovered.name) },
+        onProgress = { phase, processed, _, _ ->
+          if (phase == "extracting") secondProgress += processed
+        },
+      )
+      assertNotEquals(first.revision, second.revision)
+      assertEquals(0, second.added)
+      assertEquals(101, second.updated)
+      assertEquals(0, second.removed)
+      assertEquals(listOf(96, 101), secondProgress)
+      assertEquals(
+        firstTitles,
+        files.map { discovered -> repository.getTrack(discovered.uri)?.get("title") },
+      )
+
+      val cancelFlag = AtomicBoolean(false)
+      val extractionCount = AtomicInteger()
+      val cancelled = repository.scanLocalFolder(
+        folderId = folderId,
+        full = true,
+        discover = { files },
+        extract = { discovered ->
+          if (extractionCount.incrementAndGet() == 3) cancelFlag.set(true)
+          metadata("Cancelled ${discovered.name}")
+        },
+        onProgress = { _, _, _, _ -> },
+        isCancelled = cancelFlag::get,
+      )
+      assertTrue(cancelled.cancelled)
+      assertEquals(second.revision, cancelled.revision)
+      assertEquals(second.revision, repository.status().catalogRevision)
+      assertEquals(firstTitles[0], repository.getTrack(files[0].uri)?.get("title"))
+
+      try {
+        repository.scanLocalFolder(
+          folderId = folderId,
+          full = true,
+          discover = { files },
+          extract = { discovered ->
+            if (discovered == files[3]) error("parser failure")
+            metadata("Failed ${discovered.name}")
+          },
+          onProgress = { _, _, _, _ -> },
+        )
+        fail("Expected parser failure")
+      } catch (error: IllegalStateException) {
+        assertEquals("parser failure", error.message)
+      }
+      assertEquals(second.revision, repository.status().catalogRevision)
+      assertEquals(firstTitles[0], repository.getTrack(files[0].uri)?.get("title"))
+      assertFalse(repository.getTrack(files.last().uri).isNullOrEmpty())
+    } finally {
+      repository.removeFolder(folderId)
+    }
+  }
+
   @Test
   fun unchangedIncrementalScanKeepsRevisionWhileChangesAndFullScansPublish() = runBlocking {
     val context = ApplicationProvider.getApplicationContext<Context>()
@@ -125,7 +232,7 @@ class LocalScanFastPathTest {
     mtime: Long,
   ): LocalAudioFile = LocalAudioFile(
     uri = uri,
-    name = "song.flac",
+    name = uri.substringAfterLast("%2F").substringAfterLast('/'),
     size = size,
     lastModified = mtime,
     mimeType = "audio/flac",

modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/AstraLibraryScannerModule.kt

@@ -2,6 +2,7 @@ package expo.modules.astralibraryscanner
 
 import android.content.Context
 import android.content.Intent
+import android.content.pm.ApplicationInfo
 import android.graphics.Bitmap
 import android.graphics.BitmapFactory
 import android.media.AudioFormat
@@ -15,7 +16,9 @@ import android.net.Uri
 import android.os.Build
 import android.os.Handler
 import android.os.HandlerThread
+import android.os.SystemClock
 import android.provider.DocumentsContract
+import android.util.Log
 import com.google.android.exoplayer2.MediaItem
 import com.google.android.exoplayer2.MetadataRetriever
 import com.google.android.exoplayer2.metadata.id3.BinaryFrame
@@ -59,6 +62,86 @@ import java.security.MessageDigest
 import java.util.concurrent.ConcurrentHashMap
 import kotlin.math.roundToInt
 
+private const val LIBRARY_SCAN_LOG_TAG = "AstraLibraryScan"
+
+private data class MetadataStageTiming(
+  val androidMetadataNanos: Long,
+  val multiArtistNanos: Long,
+  val technicalFormatNanos: Long,
+  val artworkNanos: Long,
+  val totalNanos: Long,
+)
+
+private data class MetadataTimingSnapshot(
+  val count: Int,
+  val androidMetadataNanos: Long,
+  val multiArtistNanos: Long,
+  val technicalFormatNanos: Long,
+  val artworkNanos: Long,
+  val totalNanos: Long,
+  val maximumTotalNanos: Long,
+)
+
+private class MetadataTimingAccumulator {
+  private var count = 0
+  private var androidMetadataNanos = 0L
+  private var multiArtistNanos = 0L
+  private var technicalFormatNanos = 0L
+  private var artworkNanos = 0L
+  private var totalNanos = 0L
+  private var maximumTotalNanos = 0L
+
+  @Synchronized
+  fun record(timing: MetadataStageTiming) {
+    count += 1
+    androidMetadataNanos += timing.androidMetadataNanos
+    multiArtistNanos += timing.multiArtistNanos
+    technicalFormatNanos += timing.technicalFormatNanos
+    artworkNanos += timing.artworkNanos
+    totalNanos += timing.totalNanos
+    maximumTotalNanos = maxOf(maximumTotalNanos, timing.totalNanos)
+  }
+
+  @Synchronized
+  private fun snapshot(): MetadataTimingSnapshot = MetadataTimingSnapshot(
+    count = count,
+    androidMetadataNanos = androidMetadataNanos,
+    multiArtistNanos = multiArtistNanos,
+    technicalFormatNanos = technicalFormatNanos,
+    artworkNanos = artworkNanos,
+    totalNanos = totalNanos,
+    maximumTotalNanos = maximumTotalNanos,
+  )
+
+  fun logIfDebuggable(context: Context, folderId: Long) {
+    val debuggable =
+      context.applicationInfo.flags and ApplicationInfo.FLAG_DEBUGGABLE != 0
+    if (!debuggable) return
+    val timing = snapshot()
+    Log.d(
+      LIBRARY_SCAN_LOG_TAG,
+      "folderId=$folderId metadataFiles=${timing.count}" +
+        " androidMetadataTotalMs=${nanosToMs(timing.androidMetadataNanos)}" +
+        " androidMetadataAvgMs=${averageMs(timing.androidMetadataNanos, timing.count)}" +
+        " multiArtistTotalMs=${nanosToMs(timing.multiArtistNanos)}" +
+        " multiArtistAvgMs=${averageMs(timing.multiArtistNanos, timing.count)}" +
+        " technicalFormatTotalMs=${nanosToMs(timing.technicalFormatNanos)}" +
+        " technicalFormatAvgMs=${averageMs(timing.technicalFormatNanos, timing.count)}" +
+        " artworkTotalMs=${nanosToMs(timing.artworkNanos)}" +
+        " artworkAvgMs=${averageMs(timing.artworkNanos, timing.count)}" +
+        " trackExtractionTotalMs=${nanosToMs(timing.totalNanos)}" +
+        " trackExtractionAvgMs=${averageMs(timing.totalNanos, timing.count)}" +
+        " trackExtractionMaxMs=${nanosToMs(timing.maximumTotalNanos)}",
+    )
+  }
+
+  private fun averageMs(totalNanos: Long, count: Int): Double =
+    if (count == 0) 0.0 else nanosToMs(totalNanos / count)
+
+  private fun nanosToMs(nanos: Long): Double =
+    (nanos / 10_000.0).roundToInt() / 100.0
+}
+
 class FileRequest : Record {
   @Field val uri: String = ""
   @Field val coverUri: String? = null
@@ -143,11 +226,13 @@ class AstraLibraryScannerModule : Module() {
         mode: String,
         extensions: List<String>,
       ->
+      val context = requireContext().applicationContext
       val cancelFlag = AtomicBoolean(false)
+      val metadataTimings = MetadataTimingAccumulator()
       activeScans.add(cancelFlag)
       try {
         withContext(Dispatchers.IO) {
-          val repository = AstraLibraryRepository.get(requireContext())
+          val repository = AstraLibraryRepository.get(context)
           repository.withUserRecovery { scanLocalFolder(
             folderId = folderId.toLong(),
             full = mode == "full",
@@ -172,7 +257,7 @@ class AstraLibraryScannerModule : Module() {
               }
             },
             extract = { file ->
-              extractOne(file.uri, file.coverUri).toLocalAudioMetadata()
+              extractOne(file.uri, file.coverUri, metadataTimings::record).toLocalAudioMetadata()
             },
             onProgress = { phase, processed, total, folderName ->
               sendEvent(
@@ -189,6 +274,7 @@ class AstraLibraryScannerModule : Module() {
           ).toMap() }
         }
       } finally {
+        runCatching { metadataTimings.logIfDebuggable(context, folderId.toLong()) }
         activeScans.remove(cancelFlag)
       }
     }
@@ -625,104 +711,145 @@ class AstraLibraryScannerModule : Module() {
     return extractOne(request.uri, request.coverUri)
   }
 
-  private fun extractOne(uriString: String, coverUri: String?): Map<String, Any?> {
+  private fun extractOne(
+    uriString: String,
+    coverUri: String?,
+    timingRecorder: ((MetadataStageTiming) -> Unit)? = null,
+  ): Map<String, Any?> {
+    val totalStartedNanos = SystemClock.elapsedRealtimeNanos()
+    var androidMetadataNanos = 0L
+    var multiArtistNanos = 0L
+    var technicalFormatNanos = 0L
+    var artworkNanos = 0L
     val context = requireContext()
     val uri = Uri.parse(uriString)
     val result = mutableMapOf<String, Any?>("uri" to uriString, "ok" to true)
 
-    var embeddedPicture: ByteArray? = null
-    val retriever = MediaMetadataRetriever()
     try {
-      retriever.setDataSource(context, uri)
+      var embeddedPicture: ByteArray? = null
-
+      var androidMetadataError: Throwable? = null
-      result["title"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_TITLE)
+      val androidMetadataStartedNanos = SystemClock.elapsedRealtimeNanos()
-      result["artist"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ARTIST)
+      val retriever = MediaMetadataRetriever()
-      result["album"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ALBUM)
-      result["albumArtist"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ALBUMARTIST)
-      result["genre"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_GENRE)
-      result["mimeType"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_MIMETYPE)
-      result["durationMs"] =
-        retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DURATION)?.toLongOrNull()
-      result["bitrate"] =
-        retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_BITRATE)?.toIntOrNull()
-      result["trackNumber"] = parseTagNumber(
-        retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_CD_TRACK_NUMBER)
-      )
-      result["discNumber"] = parseTagNumber(
-        retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DISC_NUMBER)
-      )
-      result["year"] = parseYear(
-        retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_YEAR),
-        retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DATE)
-      )
-      if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.S) {
-        result["sampleRate"] =
-          retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_SAMPLERATE)?.toIntOrNull()
-      }
-      embeddedPicture = retriever.embeddedPicture
-    } catch (t: Throwable) {
-      return mapOf(
-        "uri" to uriString,
-        "ok" to false,
-        "error" to (t.message ?: t.javaClass.simpleName)
-      )
-    } finally {
       try {
-        retriever.release()
+        retriever.setDataSource(context, uri)
-      } catch (_: Throwable) {}
-    }
 
-    val credits = ArtistCreditMetadataReader.read(context, uri, metadataTimeoutMs)
+        result["title"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_TITLE)
-    val artistNames = credits.artists.takeIf { it.size > 1 }.orEmpty()
+        result["artist"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ARTIST)
-    val albumArtistNames = credits.albumArtists.takeIf { it.size > 1 }.orEmpty()
+        result["album"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ALBUM)
-    result["artistNames"] = artistNames
+        result["albumArtist"] =
-    result["albumArtistNames"] = albumArtistNames
+          retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_ALBUMARTIST)
-    if (artistNames.isNotEmpty()) {
+        result["genre"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_GENRE)
-      result["artist"] = formatArtistNames(artistNames)
+        result["mimeType"] = retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_MIMETYPE)
-    } else if (result["artist"] == null && credits.artists.size == 1) {
+        result["durationMs"] =
-      result["artist"] = credits.artists[0]
+          retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DURATION)?.toLongOrNull()
-    }
+        result["bitrate"] =
-    if (albumArtistNames.isNotEmpty()) {
+          retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_BITRATE)?.toIntOrNull()
-      result["albumArtist"] = formatArtistNames(albumArtistNames)
+        result["trackNumber"] = parseTagNumber(
-    } else if (result["albumArtist"] == null && credits.albumArtists.size == 1) {
+          retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_CD_TRACK_NUMBER)
-      result["albumArtist"] = credits.albumArtists[0]
+        )
-    }
+        result["discNumber"] = parseTagNumber(
-
+          retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DISC_NUMBER)
-    // Header-level facts MMR can't provide (channels, bit depth) or only on
+        )
-    // API 31+ (sample rate). Failure here is non-fatal — keep the tag data.
+        result["year"] = parseYear(
-    val extractor = MediaExtractor()
+          retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_YEAR),
-    try {
+          retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_DATE)
-      extractor.setDataSource(context, uri, null)
+        )
-      for (i in 0 until extractor.trackCount) {
+        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.S) {
-        val format = extractor.getTrackFormat(i)
+          result["sampleRate"] =
-        val trackMime = format.getString(MediaFormat.KEY_MIME) ?: continue
+            retriever.extractMetadata(MediaMetadataRetriever.METADATA_KEY_SAMPLERATE)?.toIntOrNull()
-        if (!trackMime.startsWith("audio/")) continue
-
-        result["codecMime"] = trackMime
-        if (format.containsKey(MediaFormat.KEY_CHANNEL_COUNT)) {
-          result["channels"] = format.getInteger(MediaFormat.KEY_CHANNEL_COUNT)
         }
-        if (result["sampleRate"] == null && format.containsKey(MediaFormat.KEY_SAMPLE_RATE)) {
+        embeddedPicture = retriever.embeddedPicture
-          result["sampleRate"] = format.getInteger(MediaFormat.KEY_SAMPLE_RATE)
+      } catch (error: Throwable) {
-        }
+        androidMetadataError = error
-        result["bitsPerSample"] = readBitsPerSample(format)
+      } finally {
-        break
+        try {
+          retriever.release()
+        } catch (_: Throwable) {}
+        androidMetadataNanos = SystemClock.elapsedRealtimeNanos() - androidMetadataStartedNanos
       }
-    } catch (_: Throwable) {
+      androidMetadataError?.let { error ->
-      // Container not supported by MediaExtractor — tag data already collected.
+        return mapOf(
-    } finally {
+          "uri" to uriString,
+          "ok" to false,
+          "error" to (error.message ?: error.javaClass.simpleName),
+        )
+      }
+
+      val multiArtistStartedNanos = SystemClock.elapsedRealtimeNanos()
+      val credits = try {
+        ArtistCreditMetadataReader.read(context, uri, metadataTimeoutMs)
+      } finally {
+        multiArtistNanos = SystemClock.elapsedRealtimeNanos() - multiArtistStartedNanos
+      }
+      val artistNames = credits.artists.takeIf { it.size > 1 }.orEmpty()
+      val albumArtistNames = credits.albumArtists.takeIf { it.size > 1 }.orEmpty()
+      result["artistNames"] = artistNames
+      result["albumArtistNames"] = albumArtistNames
+      if (artistNames.isNotEmpty()) {
+        result["artist"] = formatArtistNames(artistNames)
+      } else if (result["artist"] == null && credits.artists.size == 1) {
+        result["artist"] = credits.artists[0]
+      }
+      if (albumArtistNames.isNotEmpty()) {
+        result["albumArtist"] = formatArtistNames(albumArtistNames)
+      } else if (result["albumArtist"] == null && credits.albumArtists.size == 1) {
+        result["albumArtist"] = credits.albumArtists[0]
+      }
+
+      // Header-level facts MMR can't provide (channels, bit depth) or only on
+      // API 31+ (sample rate). Failure here is non-fatal — keep the tag data.
+      val technicalFormatStartedNanos = SystemClock.elapsedRealtimeNanos()
+      val extractor = MediaExtractor()
       try {
-        extractor.release()
+        extractor.setDataSource(context, uri, null)
-      } catch (_: Throwable) {}
+        for (i in 0 until extractor.trackCount) {
-    }
+          val format = extractor.getTrackFormat(i)
+          val trackMime = format.getString(MediaFormat.KEY_MIME) ?: continue
+          if (!trackMime.startsWith("audio/")) continue
 
-    try {
+          result["codecMime"] = trackMime
-      result["artworkHash"] = resolveArtwork(embeddedPicture, coverUri)
+          if (format.containsKey(MediaFormat.KEY_CHANNEL_COUNT)) {
-    } catch (_: Throwable) {
+            result["channels"] = format.getInteger(MediaFormat.KEY_CHANNEL_COUNT)
-      // Artwork failure never fails the track.
+          }
-    }
+          if (result["sampleRate"] == null && format.containsKey(MediaFormat.KEY_SAMPLE_RATE)) {
+            result["sampleRate"] = format.getInteger(MediaFormat.KEY_SAMPLE_RATE)
+          }
+          result["bitsPerSample"] = readBitsPerSample(format)
+          break
+        }
+      } catch (_: Throwable) {
+        // Container not supported by MediaExtractor — tag data already collected.
+      } finally {
+        try {
+          extractor.release()
+        } catch (_: Throwable) {}
+        technicalFormatNanos =
+          SystemClock.elapsedRealtimeNanos() - technicalFormatStartedNanos
+      }
 
-    return result
+      val artworkStartedNanos = SystemClock.elapsedRealtimeNanos()
+      try {
+        result["artworkHash"] = resolveArtwork(embeddedPicture, coverUri)
+      } catch (_: Throwable) {
+        // Artwork failure never fails the track.
+      } finally {
+        artworkNanos = SystemClock.elapsedRealtimeNanos() - artworkStartedNanos
+      }
+
+      return result
+    } finally {
+      runCatching {
+        timingRecorder?.invoke(
+          MetadataStageTiming(
+            androidMetadataNanos = androidMetadataNanos,
+            multiArtistNanos = multiArtistNanos,
+            technicalFormatNanos = technicalFormatNanos,
+            artworkNanos = artworkNanos,
+            totalNanos = SystemClock.elapsedRealtimeNanos() - totalStartedNanos,
+          ),
+        )
+      }
+    }
   }
 
   private fun Map<String, Any?>.toLocalAudioMetadata(): LocalAudioMetadata =

modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/AstraLibraryRepository.kt

@@ -20,9 +20,6 @@ import java.util.concurrent.ConcurrentHashMap
 import java.util.concurrent.CopyOnWriteArraySet
 import java.util.concurrent.atomic.AtomicInteger
 import kotlin.random.Random
-import kotlinx.coroutines.async
-import kotlinx.coroutines.awaitAll
-import kotlinx.coroutines.coroutineScope
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -62,6 +59,8 @@ private class LocalScanTiming(
   var readModelMs: Long = 0
   var publishMs: Long = 0
   var files: Int = 0
+  var workers: Int = 0
+  var windowSize: Int = 0
   var outcome: String = "failed"
 
   fun beginLockWait() {
@@ -84,6 +83,7 @@ private class LocalScanTiming(
     Log.d(
       SCAN_LOG_TAG,
       "folderId=$folderId outcome=$outcome files=$files" +
+        " workers=$workers windowSize=$windowSize" +
         " lockWaitMs=$lockWaitMs discoveryMs=$discoveryMs" +
         " comparisonMs=$comparisonMs extractionAndStagingMs=$extractionAndStagingMs" +
         " readModelMs=$readModelMs publishMs=$publishMs" +
@@ -552,75 +552,80 @@ class AstraLibraryRepository private constructor(
         var added = 0
         var updated = 0
         var errors = 0
-        var processed = 0
+        val workerCount = localScanWorkerCount(
+          availableProcessors = Runtime.getRuntime().availableProcessors(),
+          itemCount = files.size,
+        )
+        timing.workers = workerCount
+        timing.windowSize = LOCAL_SCAN_WINDOW_SIZE
 
-        for (batch in files.chunked(24)) {
+        runBoundedLocalScanPipeline(
-          throwIfScanCancelled(isCancelled)
+          items = files,
-          val rows = coroutineScope {
+          workerCount = workerCount,
-            batch.map { file ->
+          process = { file ->
-              async(Dispatchers.IO) {
+            throwIfScanCancelled(isCancelled)
-                throwIfScanCancelled(isCancelled)
+            val old = existingByPath[file.uri]
-                val old = existingByPath[file.uri]
+            val unchanged = !effectiveFull &&
-                val unchanged = !effectiveFull &&
+              old != null &&
-                  old != null &&
+              old.mtime == file.lastModified &&
-                  old.mtime == file.lastModified &&
+              old.size == file.size
-                  old.size == file.size
+            if (unchanged) {
-                if (unchanged) {
+              StagedLocalTrack(
-                  StagedLocalTrack(
+                row = old!!.copy(
-                    row = old!!.copy(
+                  id = 0,
-                      id = 0,
+                  generationId = generationId,
-                      generationId = generationId,
+                  sourceKey = sourceKey,
-                      sourceKey = sourceKey,
+                  titleSortKey = SortKeys.forText(old.title),
-                      titleSortKey = SortKeys.forText(old.title),
+                  artistSortKey = SortKeys.forText(old.artist),
-                      artistSortKey = SortKeys.forText(old.artist),
+                  albumSortKey = SortKeys.forText(old.album),
-                      albumSortKey = SortKeys.forText(old.album),
+                  fileNameSortKey = SortKeys.forText(old.fileName),
-                      fileNameSortKey = SortKeys.forText(old.fileName),
+                  sectionLabel = SortKeys.sectionLabel(old.title),
-                      sectionLabel = SortKeys.sectionLabel(old.title),
+                ),
-                    ),
+                failed = false,
-                    failed = false,
+                metadataChanged = false,
-                    metadataChanged = false,
+              )
-                  )
+            } else {
-                } else {
+              val metadata = extract(file)
-                  val metadata = extract(file)
+              throwIfScanCancelled(isCancelled)
-                  throwIfScanCancelled(isCancelled)
+              if (!metadata.ok) {
-                  if (!metadata.ok) {
+                StagedLocalTrack(
-                    StagedLocalTrack(
+                  row = old?.copy(id = 0, generationId = generationId, sourceKey = sourceKey),
-                      row = old?.copy(id = 0, generationId = generationId, sourceKey = sourceKey),
+                  failed = true,
-                      failed = true,
+                  metadataChanged = false,
-                      metadataChanged = false,
+                )
-                    )
+              } else {
-                  } else {
+                StagedLocalTrack(
-                    StagedLocalTrack(
+                  row = trackFromMetadata(
-                      row = trackFromMetadata(
+                    generationId = generationId,
-                        generationId = generationId,
+                    sourceKey = sourceKey,
-                        sourceKey = sourceKey,
+                    folderId = folderId,
-                        folderId = folderId,
+                    file = file,
-                        file = file,
+                    metadata = metadata,
-                        metadata = metadata,
+                    addedAt = old?.addedAt ?: startedAt,
-                        addedAt = old?.addedAt ?: startedAt,
+                  ),
-                      ),
+                  failed = false,
-                      failed = false,
+                  metadataChanged = true,
-                      metadataChanged = true,
+                )
-                    )
-                  }
-                }
               }
-            }.awaitAll()
-          }
-          throwIfScanCancelled(isCancelled)
-          val insertRows = ArrayList<TrackEntity>(rows.size)
-          for (staged in rows) {
-            if (staged.failed) errors += 1
-            val row = staged.row ?: continue
-            insertRows += row
-            if (staged.metadataChanged) {
-              if (existingByPath.containsKey(row.path)) updated += 1 else added += 1
             }
-          }
+          },
-          if (insertRows.isNotEmpty()) dao.putTracks(insertRows)
+          writeWindow = { rows ->
-          processed += batch.size
+            throwIfScanCancelled(isCancelled)
-          onProgress("extracting", processed, files.size, folder.displayName)
+            val insertRows = ArrayList<TrackEntity>(rows.size)
-        }
+            for (staged in rows) {
+              if (staged.failed) errors += 1
+              val row = staged.row ?: continue
+              insertRows += row
+              if (staged.metadataChanged) {
+                if (existingByPath.containsKey(row.path)) updated += 1 else added += 1
+              }
+            }
+            if (insertRows.isNotEmpty()) dao.putTracks(insertRows)
+          },
+          onWindowCommitted = { processed, total ->
+            onProgress("extracting", processed, total, folder.displayName)
+          },
+        )
         timing.extractionAndStagingMs = timing.elapsedMs(extractionAndStagingStarted)
 
         throwIfScanCancelled(isCancelled)

modules/astra-library-scanner/android/src/main/java/expo/modules/astralibraryscanner/data/LocalScanPipeline.kt

@@ -0,0 +1,108 @@
+package expo.modules.astralibraryscanner.data
+
+import java.util.concurrent.atomic.AtomicInteger
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.coroutineScope
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.ensureActive
+import kotlinx.coroutines.joinAll
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.sync.Semaphore
+
+internal const val LOCAL_SCAN_WINDOW_SIZE = 96
+private const val LOCAL_SCAN_WORKERS_PER_PROCESSOR = 3
+private const val LOCAL_SCAN_MIN_WORKERS = 8
+private const val LOCAL_SCAN_MAX_WORKERS = 24
+private const val LOCAL_SCAN_CHANNEL_CAPACITY = 1
+private const val LOCAL_SCAN_RETAINED_WINDOWS = LOCAL_SCAN_CHANNEL_CAPACITY + 1
+
+internal fun localScanWorkerCount(
+  availableProcessors: Int,
+  itemCount: Int,
+): Int {
+  if (itemCount <= 0) return 0
+  val processors = availableProcessors.coerceAtLeast(1)
+  val target = (processors * LOCAL_SCAN_WORKERS_PER_PROCESSOR)
+    .coerceIn(LOCAL_SCAN_MIN_WORKERS, LOCAL_SCAN_MAX_WORKERS)
+  return minOf(target, itemCount)
+}
+
+/**
+ * Extracts ordered windows in parallel while a single consumer commits them.
+ *
+ * The capacity-one channel plus the two-permit window guard allows one window to be
+ * written while the next is extracted, without starting a third retained window.
+ */
+internal suspend fun <Input, Output : Any> runBoundedLocalScanPipeline(
+  items: List<Input>,
+  workerCount: Int,
+  windowSize: Int = LOCAL_SCAN_WINDOW_SIZE,
+  process: suspend (Input) -> Output,
+  writeWindow: suspend (List<Output>) -> Unit,
+  onWindowCommitted: (processed: Int, total: Int) -> Unit = { _, _ -> },
+) {
+  if (items.isEmpty()) return
+  require(workerCount > 0) { "workerCount must be positive" }
+  require(windowSize > 0) { "windowSize must be positive" }
+
+  coroutineScope {
+    val completedWindows = Channel<List<Output>>(capacity = LOCAL_SCAN_CHANNEL_CAPACITY)
+    val retainedWindowSlots = Semaphore(LOCAL_SCAN_RETAINED_WINDOWS)
+    val producer = launch(Dispatchers.IO) {
+      try {
+        for (inputWindow in items.chunked(windowSize)) {
+          currentCoroutineContext().ensureActive()
+          retainedWindowSlots.acquire()
+          var handedToWriter = false
+          try {
+            val outputWindow = processOrderedWindow(inputWindow, workerCount, process)
+            completedWindows.send(outputWindow)
+            handedToWriter = true
+          } finally {
+            if (!handedToWriter) retainedWindowSlots.release()
+          }
+        }
+      } finally {
+        completedWindows.close()
+      }
+    }
+
+    var processed = 0
+    try {
+      for (window in completedWindows) {
+        try {
+          writeWindow(window)
+        } finally {
+          retainedWindowSlots.release()
+        }
+        processed += window.size
+        onWindowCommitted(processed, items.size)
+      }
+      producer.join()
+    } finally {
+      completedWindows.cancel()
+    }
+  }
+}
+
+private suspend fun <Input, Output : Any> processOrderedWindow(
+  items: List<Input>,
+  requestedWorkerCount: Int,
+  process: suspend (Input) -> Output,
+): List<Output> = coroutineScope {
+  val nextIndex = AtomicInteger()
+  val results = MutableList<Output?>(items.size) { null }
+  val workers = List(minOf(requestedWorkerCount, items.size)) {
+    launch {
+      while (true) {
+        currentCoroutineContext().ensureActive()
+        val index = nextIndex.getAndIncrement()
+        if (index >= items.size) break
+        results[index] = process(items[index])
+      }
+    }
+  }
+  workers.joinAll()
+  results.map { checkNotNull(it) }
+}

modules/astra-library-scanner/android/src/test/java/expo/modules/astralibraryscanner/data/LocalScanPipelineTest.kt

@@ -0,0 +1,163 @@
+package expo.modules.astralibraryscanner.data
+
+import java.util.Collections
+import java.util.concurrent.atomic.AtomicInteger
+import kotlinx.coroutines.CompletableDeferred
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.withTimeout
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertFalse
+import org.junit.Assert.assertTrue
+import org.junit.Assert.fail
+import org.junit.Test
+
+class LocalScanPipelineTest {
+  @Test
+  fun workerCountClampsToProcessorsLimitsAndItems() {
+    assertEquals(0, localScanWorkerCount(16, 0))
+    assertEquals(1, localScanWorkerCount(16, 1))
+    assertEquals(8, localScanWorkerCount(1, 20))
+    assertEquals(12, localScanWorkerCount(4, 20))
+    assertEquals(20, localScanWorkerCount(32, 20))
+    assertEquals(24, localScanWorkerCount(32, 100))
+    assertEquals(3, localScanWorkerCount(8, 3))
+  }
+
+  @Test
+  fun outputStaysOrderedAndWritesFullAndPartialWindows() = runBlocking {
+    val writes = mutableListOf<List<Int>>()
+    val progress = mutableListOf<Int>()
+
+    runBoundedLocalScanPipeline(
+      items = (0 until 205).toList(),
+      workerCount = 8,
+      process = { value ->
+        delay(((7 - value % 8) + 1).toLong())
+        value
+      },
+      writeWindow = { writes += it },
+      onWindowCommitted = { processed, total ->
+        assertEquals(205, total)
+        progress += processed
+      },
+    )
+
+    assertEquals(listOf(96, 96, 13), writes.map(List<Int>::size))
+    assertEquals((0 until 205).toList(), writes.flatten())
+    assertEquals(listOf(96, 192, 205), progress)
+  }
+
+  @Test
+  fun activeExtractionNeverExceedsWorkerCount() = runBlocking {
+    val active = AtomicInteger()
+    val maximum = AtomicInteger()
+
+    runBoundedLocalScanPipeline(
+      items = (0 until 24).toList(),
+      workerCount = 3,
+      windowSize = 24,
+      process = { value ->
+        val current = active.incrementAndGet()
+        maximum.getAndUpdate { previous -> maxOf(previous, current) }
+        try {
+          delay(10)
+          value
+        } finally {
+          active.decrementAndGet()
+        }
+      },
+      writeWindow = {},
+    )
+
+    assertEquals(3, maximum.get())
+  }
+
+  @Test
+  fun extractionOfNextWindowOverlapsCurrentWrite() = runBlocking {
+    val secondWindowStarted = CompletableDeferred<Unit>()
+    val thirdWindowStarted = CompletableDeferred<Unit>()
+    val writes = mutableListOf<List<Int>>()
+
+    withTimeout(2_000) {
+      runBoundedLocalScanPipeline(
+        items = (0 until 6).toList(),
+        workerCount = 2,
+        windowSize = 2,
+        process = { value ->
+          if (value >= 2) secondWindowStarted.complete(Unit)
+          if (value >= 4) thirdWindowStarted.complete(Unit)
+          value
+        },
+        writeWindow = { window ->
+          if (writes.isEmpty()) {
+            secondWindowStarted.await()
+            delay(50)
+            assertFalse(thirdWindowStarted.isCompleted)
+          }
+          writes += window
+        },
+      )
+    }
+
+    assertEquals(listOf(listOf(0, 1), listOf(2, 3), listOf(4, 5)), writes)
+  }
+
+  @Test
+  fun extractionFailureCancelsPipelineAndSkipsUncommittedWindows() = runBlocking {
+    val processed = Collections.synchronizedList(mutableListOf<Int>())
+    val writes = mutableListOf<List<Int>>()
+
+    try {
+      runBoundedLocalScanPipeline(
+        items = (0 until 12).toList(),
+        workerCount = 4,
+        windowSize = 6,
+        process = { value ->
+          processed += value
+          if (value == 3) error("parser failure")
+          delay(20)
+          value
+        },
+        writeWindow = { writes += it },
+      )
+      fail("Expected parser failure")
+    } catch (error: IllegalStateException) {
+      assertEquals("parser failure", error.message)
+    }
+
+    assertTrue(writes.isEmpty())
+    assertTrue(processed.size < 12)
+  }
+
+  @Test
+  fun writerFailureCancelsFurtherExtraction() = runBlocking {
+    val active = AtomicInteger()
+    val completed = AtomicInteger()
+
+    try {
+      runBoundedLocalScanPipeline(
+        items = (0 until 30).toList(),
+        workerCount = 2,
+        windowSize = 5,
+        process = { value ->
+          active.incrementAndGet()
+          try {
+            delay(15)
+            completed.incrementAndGet()
+            value
+          } finally {
+            active.decrementAndGet()
+          }
+        },
+        writeWindow = { error("database failure") },
+      )
+      fail("Expected database failure")
+    } catch (error: IllegalStateException) {
+      assertEquals("database failure", error.message)
+    }
+
+    assertEquals(0, active.get())
+    assertTrue(completed.get() < 30)
+  }
+}