diff --git a/apps/android/app/src/main/java/video/crumb/app/data/CrumbApi.kt b/apps/android/app/src/main/java/video/crumb/app/data/CrumbApi.kt index 14d6cf39..33997ced 100644 --- a/apps/android/app/src/main/java/video/crumb/app/data/CrumbApi.kt +++ b/apps/android/app/src/main/java/video/crumb/app/data/CrumbApi.kt @@ -80,6 +80,23 @@ interface CrumbApi { @Query("buckets") buckets: Int = 240, ): IntensityResponse + /** + * Batched form of [timelineIntensity]: one request for many cameras instead + * of one per camera (#599). Drives the multi-camera playback wall's combined + * motion overlay, which previously fired N parallel per-camera requests on + * every scrub re-center and tripped the server's shared rate limiter. The + * server caps `camera_ids` at `MAX_INTENSITY_BATCH` (services/api/src/timeline.rs, + * currently 64) and 400s above it; a server predating this route 404s. See + * [CrumbRepository.timelineIntensityCombined] for the chunking + fallback. + */ + @GET("timeline/intensity/batch") + suspend fun timelineIntensityBatch( + @Query("camera_ids") cameraIds: String, + @Query("start") start: String, + @Query("end") end: String, + @Query("buckets") buckets: Int = 240, + ): IntensityBatchResponse + /** * The leading edge of the next/previous merged motion EVENT relative to * [from], searched across ALL recorded history (server-side; same diff --git a/apps/android/app/src/main/java/video/crumb/app/data/CrumbRepository.kt b/apps/android/app/src/main/java/video/crumb/app/data/CrumbRepository.kt index 5f8c31fd..accf371b 100644 --- a/apps/android/app/src/main/java/video/crumb/app/data/CrumbRepository.kt +++ b/apps/android/app/src/main/java/video/crumb/app/data/CrumbRepository.kt @@ -11,8 +11,20 @@ import kotlinx.coroutines.coroutineScope import kotlinx.coroutines.delay import retrofit2.HttpException import java.io.IOException +import java.util.concurrent.ConcurrentHashMap import javax.net.ssl.SSLHandshakeException +/** + * Max cameras accepted by `GET /timeline/intensity/batch` per request — must + * match the server's `MAX_INTENSITY_BATCH` (services/api/src/timeline.rs), + * which 400s above this. The multi-camera playback wall can exceed it (e.g. + * "All Cameras"), so [CrumbRepository.timelineIntensityCombined] splits into + * chunks of this size, mirroring the desktop + * (`apps/desktop-flutter/lib/api/motion_timeline_api.dart`) and iOS + * (`apps/ios/Crumb/Features/Playback/PlaybackViewModel.swift`) clients (#599). + */ +private const val MAX_INTENSITY_BATCH = 64 + /** * Like [runCatching] but re-throws [CancellationException] instead of capturing * it as a `Result.failure`. @@ -47,6 +59,15 @@ class CrumbRepository(private val container: AppContainer) { private val api: CrumbApi get() = container.api val store: SecureStore get() = container.store + /** + * Server base URLs whose API 404'd `/timeline/intensity/batch` — i.e. older + * than the batch endpoint. Keyed by base (not a lone flag) so switching to a + * different, newer server later in the same process isn't wrongly demoted + * to the per-camera fallback (#599). Thread-safe: reads/writes can happen + * from concurrent scrub-triggered loads. + */ + private val batchUnsupportedBases = ConcurrentHashMap.newKeySet() + fun mediaUrls(): MediaUrls = container.mediaUrls() /** @@ -176,28 +197,88 @@ class CrumbRepository(private val container: AppContainer) { /** Combined motion histogram across MANY cameras: the per-bucket MAX of each * camera's intensity, so a multi-camera wall timeline shows "the busiest * camera at that moment" and only goes quiet when EVERY camera is quiet. - * Fetches each camera in parallel; a camera that errors contributes nothing - * (no bar) rather than failing the whole overlay. */ + * Uses the batched `/timeline/intensity/batch` endpoint (chunked, #599) + * rather than one request per camera; a camera that errors contributes + * nothing (no bar) rather than failing the whole overlay. */ suspend fun timelineIntensityCombined( cameraIds: List, startIso: String, endIso: String, buckets: Int = 240, ): Result> = runCatchingCancellable { - coroutineScope { - val perCamera = cameraIds.map { id -> + val byId = fetchIntensityBatched(cameraIds, startIso, endIso, buckets) + combineIntensityMax(byId.values, buckets) + } + + /** + * Fetch per-camera intensity buckets for [cameraIds] via the batched + * `GET /timeline/intensity/batch`, split into <=[MAX_INTENSITY_BATCH]-camera + * chunks (the server 400s above that; a large "All Cameras" wall can exceed + * it). This replaces the old N-simultaneous-requests-per-scrub fan-out that + * tripped the server's shared rate limiter (#599). + * + * Falls back to the pre-batch per-camera fan-out ([fetchIntensityPerCamera]) + * on 404 (server predates the batch route) or 400 (unexpected rejection, + * e.g. a stricter cap on that server), and remembers a 404 per server base so + * later loads skip the doomed batch attempt for the rest of the session — + * mirrors the desktop (`motion_timeline_api.dart`) and iOS + * (`PlaybackViewModel.swift`) clients. Any other per-chunk failure (network + * blip, 5xx) just leaves that chunk's cameras out of the map rather than + * failing the whole load, matching the old per-camera error tolerance. + */ + private suspend fun fetchIntensityBatched( + cameraIds: List, + startIso: String, + endIso: String, + buckets: Int, + ): Map> { + val base = store.serverUrl + if (base in batchUnsupportedBases) { + return fetchIntensityPerCamera(cameraIds, startIso, endIso, buckets) + } + val merged = mutableMapOf>() + for (chunk in cameraIds.chunked(MAX_INTENSITY_BATCH)) { + try { + merged.putAll( + api.timelineIntensityBatch(chunk.joinToString(","), startIso, endIso, buckets).cameras, + ) + } catch (e: CancellationException) { + throw e + } catch (e: HttpException) { + if (e.code() == 404 || e.code() == 400) { + if (e.code() == 404) batchUnsupportedBases.add(base) + // Older/incompatible server: fall back to the per-camera fan-out + // for every requested camera, not just the remaining chunks. + return fetchIntensityPerCamera(cameraIds, startIso, endIso, buckets) + } + // Any other HTTP failure is transient for this chunk only; leave + // its cameras out of the map and keep going. + } catch (e: IOException) { + // Network failure for this chunk only; same tolerance as above. + } + } + return merged + } + + /** Pre-batch per-camera fan-out — the fallback for a server older than the + * batch route (or one that unexpectedly rejects it). One request per + * camera, in parallel, same as before #599; a camera that errors + * contributes nothing rather than failing the whole overlay. */ + private suspend fun fetchIntensityPerCamera( + cameraIds: List, + startIso: String, + endIso: String, + buckets: Int, + ): Map> = coroutineScope { + cameraIds + .map { id -> async { - runCatchingCancellable { api.timelineIntensity(id, startIso, endIso, buckets).buckets } + id to runCatchingCancellable { api.timelineIntensity(id, startIso, endIso, buckets).buckets } .getOrDefault(emptyList()) } - }.awaitAll() - val combined = FloatArray(buckets) - for (cam in perCamera) { - val n = minOf(cam.size, buckets) - for (i in 0 until n) if (cam[i] > combined[i]) combined[i] = cam[i] } - combined.asList() - } + .awaitAll() + .toMap() } suspend fun resolveSegment(cameraId: String, tsIso: String, stream: String = "main"): Result = @@ -511,6 +592,23 @@ class CrumbRepository(private val container: AppContainer) { runCatchingCancellable { api.updatesLatest(if (refresh) "1" else null) } } +/** + * Per-bucket MAX across a set of per-camera intensity arrays, sized to + * [buckets]. Pulled out of [CrumbRepository.timelineIntensityCombined] as a + * pure function so the merge math (the part most likely to regress) is + * testable without standing up the full [CrumbRepository]/[AppContainer] + * dependency chain (#599). A camera's array shorter than [buckets] (e.g. an + * error-tolerant fallback entry) contributes only over its own length. + */ +internal fun combineIntensityMax(perCamera: Collection>, buckets: Int): List { + val combined = FloatArray(buckets) + for (cam in perCamera) { + val n = minOf(cam.size, buckets) + for (i in 0 until n) if (cam[i] > combined[i]) combined[i] = cam[i] + } + return combined.asList() +} + /** * True when a repository call failed specifically with HTTP 404. * diff --git a/apps/android/app/src/main/java/video/crumb/app/data/Models.kt b/apps/android/app/src/main/java/video/crumb/app/data/Models.kt index 991e814a..48788e73 100644 --- a/apps/android/app/src/main/java/video/crumb/app/data/Models.kt +++ b/apps/android/app/src/main/java/video/crumb/app/data/Models.kt @@ -272,6 +272,19 @@ data class IntensityResponse( val buckets: List = emptyList(), ) +/** + * Response for `GET /timeline/intensity/batch` — the multi-camera form of + * [IntensityResponse]: one [IntensityResponse.buckets]-shaped array per + * requested camera, keyed by camera id (#599). Every requested camera is + * present in [cameras] (all-zero buckets for one with no footage or outside + * the caller's scope), so the caller gets a complete map in one request + * instead of N. Mirrors the desktop/iOS clients' `IntensityBatchResponse`. + */ +@Serializable +data class IntensityBatchResponse( + val cameras: Map> = emptyMap(), +) + /** * Wire envelope for `GET /timeline/motion` — the leading edge of the next/ * previous merged motion event relative to a reference time, searched across diff --git a/apps/android/app/src/test/java/video/crumb/app/data/CombineIntensityMaxTest.kt b/apps/android/app/src/test/java/video/crumb/app/data/CombineIntensityMaxTest.kt new file mode 100644 index 00000000..3c0f65aa --- /dev/null +++ b/apps/android/app/src/test/java/video/crumb/app/data/CombineIntensityMaxTest.kt @@ -0,0 +1,48 @@ +// SPDX-License-Identifier: AGPL-3.0-or-later + +package video.crumb.app.data + +import org.junit.Assert.assertEquals +import org.junit.Test + +/** + * Regression guard for the [combineIntensityMax] merge math behind + * [CrumbRepository.timelineIntensityCombined] (#599). This is the part of the + * batched-intensity rewrite most likely to regress silently — a wrong merge + * would show a flat or misleading combined motion strip on the multi-camera + * playback wall without throwing anything. + */ +class CombineIntensityMaxTest { + + @Test + fun `takes the per-bucket max across cameras`() { + val perCamera = listOf( + listOf(0.1f, 0.9f, 0.0f), + listOf(0.5f, 0.2f, 0.0f), + listOf(0.0f, 0.0f, 0.3f), + ) + assertEquals(listOf(0.5f, 0.9f, 0.3f), combineIntensityMax(perCamera, buckets = 3)) + } + + @Test + fun `no cameras yields all-zero buckets`() { + assertEquals(listOf(0.0f, 0.0f, 0.0f, 0.0f), combineIntensityMax(emptyList(), buckets = 4)) + } + + @Test + fun `a camera array shorter than buckets only contributes over its own length`() { + // Mirrors a fallback entry from an error-tolerant per-camera fetch that + // returned fewer buckets than requested. + val perCamera = listOf( + listOf(0.7f), // short — e.g. a degraded/partial response + listOf(0.1f, 0.1f, 0.1f), + ) + assertEquals(listOf(0.7f, 0.1f, 0.1f), combineIntensityMax(perCamera, buckets = 3)) + } + + @Test + fun `a camera array longer than buckets is truncated to buckets`() { + val perCamera = listOf(listOf(0.2f, 0.4f, 0.6f, 0.8f)) + assertEquals(listOf(0.2f, 0.4f), combineIntensityMax(perCamera, buckets = 2)) + } +}