From 6185dbfdb0752db7d715f1061d74baab3c43122e Mon Sep 17 00:00:00 2001 From: b0txec Date: Sat, 22 Aug 2026 22:41:35 +0300 Subject: [PATCH] =?UTF-8?q?Harden=20LV=C4=A2MC=20warning=20fetching=20agai?= =?UTF-8?q?nst=20partial=20failures=20and=20type=20drift?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../scala/fetch/warnings/WarningService.scala | 74 +++++++++++++------ 1 file changed, 50 insertions(+), 24 deletions(-) diff --git a/src/main/scala/fetch/warnings/WarningService.scala b/src/main/scala/fetch/warnings/WarningService.scala index f3394c3..30ed83f 100644 --- a/src/main/scala/fetch/warnings/WarningService.scala +++ b/src/main/scala/fetch/warnings/WarningService.scala @@ -55,7 +55,12 @@ final class WarningService private (logger: Logger[IO], cache: Ref[IO, Option[(I fetchFresh.attempt.flatMap { case Right(value) => cache.set(Some(now -> value)).as(value) case Left(error) => stale match { - case Some((_, value)) => logger.warn(error)("Using stale LVĢMC warning data") *> IO.pure(value) + // Re-stamp the cache with `now` even though the value is unchanged, so a + // failing upstream is retried at most once per cache lifetime instead of + // on every single incoming request. + case Some((_, value)) => + logger.warn(error)("Using stale LVĢMC warning data; will not retry until the cache lifetime elapses") *> + cache.set(Some(now -> value)).as(value) case None => IO.raiseError(error) } } @@ -70,8 +75,22 @@ final class WarningService private (logger: Logger[IO], cache: Ref[IO, Option[(I metadataBody <- client.expect[String](metadataUri) metadata <- parseRecords(metadataBody, "warning metadata") eventIds = metadata.flatMap(record => stringValue(record, "WEATHER_WARNING_EV_ID")).distinct - polygonRows <- eventIds.traverse(eventId => fetchPolygonRows(client, polygonsBaseUri, eventId)).map(_.flatten) - polygons = buildPolygons(polygonRows) + // Each event's polygons are fetched independently and concurrently; one event + // failing (network error, malformed page, ...) must not take down every other + // warning, so a failure is logged and treated as "no polygons for this event" + // rather than aborting the whole refresh. + polygonRows <- eventIds.parTraverse { eventId => + fetchPolygonRows(client, polygonsBaseUri, eventId).attempt.flatMap { + case Right(rows) => IO.pure(rows) + case Left(error) => + logger.warn(error)(s"Failed to fetch warning polygons for event $eventId; it will be shown without a shape or dropped") *> + IO.pure(Vector.empty[Json]) + } + }.map(_.flatten) + (polygons, unresolvedOrderCount) = buildPolygons(polygonRows) + _ <- IO.whenA(unresolvedOrderCount > 0)( + logger.warn(s"$unresolvedOrderCount warning polygon point(s) had a missing/invalid NPK order value and were defaulted to position 0") + ) warnings = metadata.flatMap(record => buildWarning(record, polygons)) .filter(_.polygons.nonEmpty) _ <- logger.info(s"Loaded ${warnings.size} LVĢMC warnings") @@ -103,7 +122,13 @@ final class WarningService private (logger: Logger[IO], cache: Ref[IO, Option[(I offset: Int = 0, accumulated: Vector[Json] = Vector.empty ): IO[Vector[Json]] = { - val filterValue = eventId.toLongOption.map(Json.fromLong).getOrElse(Json.fromString(eventId)) + // The polygons resource's WEATHER_WARNING_EV_ID column type isn't guaranteed, so + // match either JSON representation via CKAN's list-filter OR semantics instead of + // guessing one and silently returning zero rows if the guess is wrong. + val filterValue = eventId.toLongOption match { + case Some(numeric) => Json.arr(Json.fromString(eventId), Json.fromLong(numeric)) + case None => Json.fromString(eventId) + } val filters = Json.obj("WEATHER_WARNING_EV_ID" -> filterValue).noSpaces val uri = baseUri .withQueryParam("limit", polygonPageSize.toString) @@ -119,31 +144,32 @@ final class WarningService private (logger: Logger[IO], cache: Ref[IO, Option[(I } } + // Returns the polygons grouped by event, plus a count of points whose NPK vertex-order + // field was missing/invalid (those points are kept, defaulted to order 0, rather than + // dropped — the caller logs the count so a garbled shape doesn't fail silently). private def buildPolygons( rows: Vector[Json] - ): Map[String, Vector[WarningPolygon]] = - rows - .flatMap { row => - for { - eventId <- stringValue(row, "WEATHER_WARNING_EV_ID") - polygonId <- stringValue(row, "POLIGON_ID") - lat <- doubleValue(row, "LAT") - lon <- doubleValue(row, "LON") - } yield ((eventId, polygonId), intValue(row, "NPK").getOrElse(0), WarningPoint(lat, lon)) - } - .groupBy(_._1) + ): (Map[String, Vector[WarningPolygon]], Int) = { + val points = rows.flatMap { row => + for { + eventId <- stringValue(row, "WEATHER_WARNING_EV_ID") + polygonId <- stringValue(row, "POLIGON_ID") + lat <- doubleValue(row, "LAT") + lon <- doubleValue(row, "LON") + } yield (eventId, polygonId, intValue(row, "NPK"), WarningPoint(lat, lon)) + } + val unresolvedOrderCount = points.count(_._3.isEmpty) + val polygons = points + .groupMap(point => (point._1, point._2))(point => (point._3.getOrElse(0), point._4)) .toVector - .groupBy(_._1._1) - .view - .mapValues { polygonGroups => - polygonGroups - .map { case ((_, polygonId), points) => - WarningPolygon(polygonId, points.sortBy(_._2).map(_._3)) - } - .sortBy(_.id) - .toVector + .map { case ((eventId, polygonId), ordered) => + eventId -> WarningPolygon(polygonId, ordered.sortBy(_._1).map(_._2)) } + .groupMap(_._1)(_._2) + .view.mapValues(_.sortBy(_.id)) .toMap + (polygons, unresolvedOrderCount) + } private def buildWarning( record: Json,