Say when media bytes go missing, and show it on /admin/health (#4926)

.github/workflows/ci.yml

@@ -314,6 +314,23 @@ jobs:
       # - PersistentMediaDirCheckSpec (#4925): user-uploaded media must resolve outside the application directory,
       #   which a deploy rebuilds and deletes. Pure logic, no DB — and the failure it guards against is invisible to
       #   every other test, since the broken config behaves correctly right up until the next release.
+      # - MediaIntegritySpec (#4926): the media-storage panel's calls on what counts as data loss. Pure logic, no DB.
+      #   Its false-alarm cases matter as much as its real ones — a monitor that cries loss over an unreadable
+      #   directory gets ignored, and an ignored monitor leaves us exactly where #4925 found us.
+      # - LostMediaLogSpec (#4926): the log line that is the only signal a media file has been destroyed. Both ways
+      #   it can fail are silent — announcing every request buries the event, announcing nothing reads as health —
+      #   so the lines themselves are asserted. Pure logic, no DB.
+      # - SingleFlightGateSpec (#4926): keeps a stuck media scan from being joined by a fresh copy every poll until
+      #   the blocking-io pool is gone. Its whole job is invisible until storage hangs, and by then the panel that
+      #   would have reported it is the thing that went dark. Pure logic, no DB.
+      # - HealthMediaPayloadSpec (#4926): the field names the health page reads the media panel by. A renamed case
+      #   class field emits a different key, every value goes undefined, and a monitor reporting nothing looks
+      #   exactly like a monitor reporting nothing wrong. Pure logic, no DB.
+      # - ImageControllerSpec (#4415, #4726, #4926): the crop write path and its share-preview invalidation, plus the
+      #   loss lines a signed crop/pano URL emits when the bytes it was signed for are gone. Nothing else asserts the
+      #   controller calls LostMediaLog at all, so without this the tripwire could be deleted with CI still green.
+      # - StoryUploadWindowSpec (#4926): the grace window that stops the loss tripwire firing at its own upload path.
+      #   Reporting is once per media id, so a false alarm spends the line a real loss would have needed. Pure logic.
       # - PanoExpiredAtSpec / BackgroundJobRunTableSpec / JobRunServiceSpec / StreetLifecycleServiceSpec (#4928): the
       #   transition records the admin trend and nightly-job panels read. Each seeds its own rows rather than hunting
       #   for them, so they assert real behavior against an empty seed: the expiry stamp only lands on the false ->
@@ -333,7 +350,7 @@ jobs:
       #   the second as a success is how a rotated-out API key ends the re-audit signal behind a permanently green
       #   badge.
       - name: Run gating/auth tests (health dashboard + route auth posture + geodesic distances)
-        run: sbt 'set Test / parallelExecution := false' 'testOnly controllers.HealthDashboardSpec service.HealthServiceSpec controllers.RouteAuthPostureSpec models.street.GeodesicDistanceSpec service.ExploreTutorialRouteSpec controllers.MobileDetectionSpec service.PanoDataServiceSpec service.ActivityBreakdownSpec service.ConfigServiceTrendSpec models.utils.CityScorecardSpec models.utils.ConfigTableVoidedArchiveSpec controllers.api.StatsApiSpec controllers.ExploreSubmissionSpec controllers.ValidateSubmissionSpec controllers.ExploreNoImageryRateLimitSpec models.pano.PanoExpiredAtSpec models.utils.BackgroundJobRunTableSpec service.JobRunServiceSpec service.NightlyJobStatusSpec service.StreetLifecycleServiceSpec models.audit.OutdatedImageryFlagSyncSpec models.audit.OutdatedImageryRoutingSpec models.street.UpToDateCoverageSpec models.street.NoImageryReportsSpec models.utils.EnumTypeParitySpec actor.ScheduledJobsSpec service.ImageryPollOutcomeSpec modules.PersistentMediaDirCheckSpec'
+        run: sbt 'set Test / parallelExecution := false' 'testOnly controllers.HealthDashboardSpec service.HealthServiceSpec controllers.RouteAuthPostureSpec models.street.GeodesicDistanceSpec service.ExploreTutorialRouteSpec controllers.MobileDetectionSpec service.PanoDataServiceSpec service.ActivityBreakdownSpec service.ConfigServiceTrendSpec models.utils.CityScorecardSpec models.utils.ConfigTableVoidedArchiveSpec controllers.api.StatsApiSpec controllers.ExploreSubmissionSpec controllers.ValidateSubmissionSpec controllers.ExploreNoImageryRateLimitSpec models.pano.PanoExpiredAtSpec models.utils.BackgroundJobRunTableSpec service.JobRunServiceSpec service.NightlyJobStatusSpec service.StreetLifecycleServiceSpec models.audit.OutdatedImageryFlagSyncSpec models.audit.OutdatedImageryRoutingSpec models.street.UpToDateCoverageSpec models.street.NoImageryReportsSpec models.utils.EnumTypeParitySpec actor.ScheduledJobsSpec service.ImageryPollOutcomeSpec modules.PersistentMediaDirCheckSpec service.MediaIntegritySpec service.LostMediaLogSpec service.SingleFlightGateSpec service.HealthMediaPayloadSpec controllers.ImageControllerSpec controllers.StoryUploadWindowSpec'
         env:
           DATABASE_URL: jdbc:postgresql://localhost:5432/sidewalk
           DATABASE_USER: sidewalk

app/controllers/ImageController.scala

@@ -8,7 +8,7 @@ import models.label.LabelTypeEnum
 import play.api.libs.json._
 import play.api.mvc.{AnyContent, Request, RequestHeader}
 import play.api.{Configuration, Logger}
-import service.ImageSigningService
+import service.{ImageSigningService, LostMediaLog}
 
 import java.awt.Image
 import java.awt.image.BufferedImage
@@ -25,6 +25,7 @@ class ImageController @Inject() (
     signingService: ImageSigningService,
     shareImageCache: service.ShareImageCache,
     config: Configuration,
+    lostMediaLog: LostMediaLog,
     cpuEc: CpuIntensiveExecutionContext
 )(implicit ec: ExecutionContext)
     extends CustomBaseController(cc) {
@@ -137,6 +138,15 @@ class ImageController @Inject() (
             val contentType = if (file.getName.toLowerCase.endsWith(".png")) "image/png" else "image/jpeg"
             Future.successful(Ok.sendFile(file, inline = true).as(contentType))
           case None =>
+            // Reaching here means the file was on disk when this URL was signed (backupImageUrl and
+            // getBackupImageMetadata both check first) and is gone within the signature's ~75-minute life. For an
+            // pano the provider no longer serves nothing can re-fetch it, so say so (#4926) — the 404 stays bare.
+            lostMediaLog.reportMissing(
+              "pano",
+              panoId,
+              s"${panoDataService.backupImageDir(panoId).getAbsolutePath}/$panoId.{jpg,jpeg,png}",
+              irreplaceable = true
+            )
             Future.successful(NotFound(s"Pano image not found: $panoId"))
         }
     }
@@ -188,6 +198,10 @@ class ImageController @Inject() (
         if (file.exists()) {
           Future.successful(Ok.sendFile(file, inline = true).as("image/png"))
         } else {
+          // Same signed-URL reasoning as serveBackupImage above: cropUrl only signs a crop it just saw on disk, so a
+          // miss here is a file that vanished — and nothing recreates it: the crop was captured in the labeler's
+          // browser at labeling time, from a pano the provider may no longer serve.
+          lostMediaLog.reportMissing("crop", s"$labelType/$labelId", file.getAbsolutePath, irreplaceable = true)
           Future.successful(NotFound("Crop image not found"))
         }
     }

app/controllers/StoryController.scala

@@ -9,11 +9,10 @@ import models.story.{Story, StoryMedia, StoryPhotoUpload, StoryRejection}
 import play.api.libs.json.{JsBoolean, Json}
 import play.api.{Configuration, Logger}
 import play.silhouette.api.Silhouette
-import service.{ConfigService, ImageSigningService, RateLimiter, StoryService}
+import service.{ConfigService, ImageSigningService, LostMediaLog, RateLimiter, StoryService}
 
 import java.io.File
 import java.time.{Duration, OffsetDateTime}
-import java.util.concurrent.ConcurrentHashMap
 import javax.inject.{Inject, Singleton}
 import scala.concurrent.{ExecutionContext, Future}
 import scala.util.Try
@@ -33,6 +32,7 @@ class StoryController @Inject() (
     storyService: StoryService,
     signingService: ImageSigningService,
     rateLimiter: RateLimiter,
+    lostMediaLog: LostMediaLog,
     implicit val ec: ExecutionContext
 ) extends CustomBaseController(cc) {
   private val logger = Logger(this.getClass)
@@ -228,27 +228,16 @@ class StoryController @Inject() (
     }
   }
 
-  // serveStoryMedia's data-loss tripwire (#4925): one entry per media id already reported, so a busy page
-  // re-requesting one lost file reads as one loss in the log, not hundreds.
-  private val lostMediaLogged = ConcurrentHashMap.newKeySet[Int]()
-
-  // The upload flow commits the media row before the file move lands (StoryService's place-before-commit windows), so
-  // a row this young with no bytes is almost certainly mid-upload, not loss. The window is sub-second; a minute is
-  // generous slack.
-  private val lostMediaGrace = Duration.ofMinutes(1)
-
   /**
-   * Logs a media row whose bytes are missing from disk. This error is a post-#4925 tripwire someone is expected to
-   * investigate, so it is kept high-signal: skipped inside the upload grace window (a false alarm has real cost) and
-   * logged once per media id per instance.
+   * Reports a media row whose bytes are missing from disk (#4925). Kept high-signal by skipping the upload grace
+   * window, since a false alarm on this tripwire has real cost; `LostMediaLog` handles the rest.
    *
    * @param media The media row whose file is missing.
    * @param file  Where the bytes should have been.
    */
   private def logLostMedia(media: StoryMedia, file: File): Unit = {
-    val inUploadWindow = media.createdAt.isAfter(OffsetDateTime.now.minus(lostMediaGrace))
-    if (!inUploadWindow && lostMediaLogged.add(media.storyMediaId)) {
-      logger.error(s"story_media ${media.storyMediaId} has no file on disk at ${file.getAbsolutePath}")
+    if (!StoryController.withinUploadWindow(media.createdAt, OffsetDateTime.now)) {
+      lostMediaLog.reportMissing("story_media", media.storyMediaId.toString, file.getAbsolutePath, irreplaceable = true)
     }
   }
 
@@ -312,4 +301,22 @@ object StoryController {
 
   /** Cap on the stories the /stories page renders, so the page stays bounded as a city's story count grows. */
   val ListingMax: Int = 500
+
+  // The upload flow commits the media row before the file move lands (StoryService's place-before-commit windows),
+  // so a row this young with no bytes is almost certainly mid-upload. The window is sub-second; a minute is generous
+  // slack.
+  private val lostMediaGrace = Duration.ofMinutes(1)
+
+  /**
+   * Whether a media row is young enough that missing bytes read as an upload in progress rather than as loss.
+   *
+   * The data-loss log deduplicates per item, so a false alarm doesn't merely add noise — it spends the one line that
+   * media id will ever get, and a real loss discovered later says nothing at all.
+   *
+   * @param createdAt When the media row was committed.
+   * @param now       The instant to judge it against.
+   * @return          True while the row is inside the upload window and its file may still be landing.
+   */
+  private[controllers] def withinUploadWindow(createdAt: OffsetDateTime, now: OffsetDateTime): Boolean =
+    createdAt.isAfter(now.minus(lostMediaGrace))
 }

app/executors/CustomExecutionContexts.scala

@@ -27,3 +27,17 @@ object CpuIntensiveExecutionContext {
       extends CustomExecutionContext(system, "cpu-intensive")
       with CpuIntensiveExecutionContext
 }
+
+/**
+ * Execution context for blocking filesystem operations, isolated so that a call that never returns — a stat or a
+ * directory listing against an unreachable network mount — can only exhaust this pool and not the ones serving
+ * requests or running streams.
+ */
+trait BlockingIoExecutionContext extends ExecutionContext
+
+object BlockingIoExecutionContext {
+  @Singleton
+  class PekkoBased @Inject() (system: ActorSystem)
+      extends CustomExecutionContext(system, "blocking-io")
+      with BlockingIoExecutionContext
+}

app/models/utils/HealthTable.scala

@@ -54,6 +54,12 @@ class HealthTable @Inject() (protected val dbConfigProvider: DatabaseConfigProvi
   implicit private val grDbEnvInfo: GetResult[DbEnvInfo] =
     GetResult(r => DbEnvInfo(r.nextString(), r.nextString(), r.nextBoolean()))
 
+  implicit private val grSchemaRowCount: GetResult[SchemaRowCount] =
+    GetResult(r => SchemaRowCount(r.nextString(), r.nextInt()))
+
+  implicit private val grSchemaMediaId: GetResult[SchemaMediaId] =
+    GetResult(r => SchemaMediaId(r.nextString(), r.nextInt()))
+
   /**
    * Caps a health read so it can never hold a pool connection for long. A monitoring query must not add load — least
    * of all when the database is already stressed, which is exactly when this dashboard gets opened. `statement_timeout`
@@ -275,4 +281,63 @@ class HealthTable @Inject() (protected val dbConfigProvider: DatabaseConfigProvi
       LEFT JOIN pano_data pd ON pd.pano_id = labeled.pano_id
     """.as[PanoBackupStats].head
   }
+
+  /**
+   * The schema this instance actually reads and writes, which is what the media scan resolves its own city's
+   * directory from. Config alone can't answer it: `city-id` and the connection's schema are separate settings and a
+   * misconfigured instance has them disagreeing, which would make the scan look for files somewhere the app never
+   * writes them.
+   */
+  def getCurrentSchema: DBIO[String] = bounded { sql"""SELECT current_schema()""".as[String].head }
+
+  /**
+   * Schemas holding a readable `story_media` table, for the cross-city media-integrity scan (#4926).
+   *
+   * Same shape as [[getEvolutionSchemas]], including its trap: the table-privilege check takes the `pg_class` oid
+   * rather than a name, because the planner may evaluate the predicates in any order and the name-based form raises
+   * an error (instead of returning false) on a relation that the other filters would have excluded.
+   */
+  def getStoryMediaSchemas: DBIO[Seq[String]] = bounded {
+    sql"""SELECT nspname FROM pg_catalog.pg_class
+          JOIN pg_catalog.pg_namespace ON pg_namespace.oid = pg_class.relnamespace
+          WHERE relname = 'story_media' AND relkind = 'r'
+            AND has_schema_privilege(nspname, 'USAGE')
+            AND has_table_privilege(pg_class.oid, 'SELECT')
+          ORDER BY nspname""".as[String]
+  }
+
+  /**
+   * How many `story_media` rows each schema holds, in ONE query — the same `UNION ALL` construction and the same
+   * reason as [[getStuckEvolutionsForSchemas]]: a per-schema fan-out would want ~50 pool connections per poll on
+   * prod. Read before the ids so the scan can decline an implausibly large result rather than pulling it into heap.
+   *
+   * Every `schema` MUST already be validated as a bare identifier by the caller; the names are spliced literally.
+   *
+   * @param schemas Validated schema names to count. Must be non-empty.
+   * @return        One row per schema, including the schemas that hold no media at all.
+   */
+  def getStoryMediaCounts(schemas: Seq[String]): DBIO[Seq[SchemaRowCount]] = {
+    require(schemas.nonEmpty, "getStoryMediaCounts requires at least one schema (empty input builds invalid SQL)")
+    val union = schemas
+      .map(schema => s"""SELECT text '$schema' AS schema, count(*)::int AS media_rows FROM "$schema".story_media""")
+      .mkString("\nUNION ALL\n")
+    bounded(sql"""#$union""".as[SchemaRowCount])
+  }
+
+  /**
+   * Every `story_media` id in the given schemas, in ONE `UNION ALL` query. The ids themselves are needed — a count
+   * comparison would call a city with one lost photo and one orphaned file clean.
+   *
+   * Every `schema` MUST already be validated as a bare identifier by the caller; the names are spliced literally.
+   *
+   * @param schemas Validated schema names known to hold at least one row. Must be non-empty.
+   * @return        One row per media id, tagged with its schema.
+   */
+  def getStoryMediaIds(schemas: Seq[String]): DBIO[Seq[SchemaMediaId]] = {
+    require(schemas.nonEmpty, "getStoryMediaIds requires at least one schema (empty input builds invalid SQL)")
+    val union = schemas
+      .map(schema => s"""SELECT text '$schema' AS schema, story_media_id FROM "$schema".story_media""")
+      .mkString("\nUNION ALL\n")
+    bounded(sql"""#$union""".as[SchemaMediaId])
+  }
 }

app/modules/ExecutorsModule.scala

@@ -9,5 +9,6 @@ import executors._
 class ExecutorsModule extends AbstractModule {
   override def configure(): Unit = {
     bind(classOf[CpuIntensiveExecutionContext]).to(classOf[CpuIntensiveExecutionContext.PekkoBased]).asEagerSingleton()
+    bind(classOf[BlockingIoExecutionContext]).to(classOf[BlockingIoExecutionContext.PekkoBased]).asEagerSingleton()
   }
 }