refetch cache and ttl

config-repo/pcp-stations-service.yaml

@@ -33,6 +33,17 @@ management:
     info:
       enabled: true
 
+classifier:
+  stations-url: ${CLASSIFIER_STATIONS_URL:https://ordinis.k8s.265.nstart.cloud/api/v1/ground_station/records}
+  # cache TTL
+  stations-cache-ttl: ${CLASSIFIER_STATIONS_CACHE_TTL:15m}
+  # MDM retries
+  stations-refresh-attempts: ${CLASSIFIER_STATIONS_REFRESH_ATTEMPTS:3}
+  # MDM retry delay
+  stations-refresh-retry-delay: ${CLASSIFIER_STATIONS_REFRESH_RETRY_DELAY:300ms}
+  # MDM failure cooldown
+  stations-refresh-failure-cooldown: ${CLASSIFIER_STATIONS_REFRESH_FAILURE_COOLDOWN:5s}
+
 ---
 spring:
   config:
@@ -89,13 +100,3 @@ spring:
 server:
   port: ${pcp.ports.stations:8080}
 
-classifier:
-  stations-url: ${CLASSIFIER_STATIONS_URL:https://ordinis.k8s.265.nstart.cloud/api/v1/ground_station/records}
-  # cache TTL
-  stations-cache-ttl: ${CLASSIFIER_STATIONS_CACHE_TTL:15m}
-  # MDM retries
-  stations-refresh-attempts: ${CLASSIFIER_STATIONS_REFRESH_ATTEMPTS:3}
-  # MDM retry delay
-  stations-refresh-retry-delay: ${CLASSIFIER_STATIONS_REFRESH_RETRY_DELAY:300ms}
-  # MDM failure cooldown
-  stations-refresh-failure-cooldown: ${CLASSIFIER_STATIONS_REFRESH_FAILURE_COOLDOWN:5s}

services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/GlobalExceptionHandle.kt

@@ -0,0 +1,43 @@
+package space.nstart.pcp.pcp_stations_service.service
+
+import org.springframework.http.HttpHeaders
+import org.springframework.http.HttpStatus
+import org.springframework.http.ResponseEntity
+import org.springframework.web.bind.annotation.ExceptionHandler
+import space.nstart.pcp.pcp_stations_service.integration.NsiStationsResponseException
+
+class GlobalExceptionHandle {
+    @ExceptionHandler(StationsDataUnavailableException::class)
+    fun handleStationsDataUnavailable(
+        ex: StationsDataUnavailableException
+    ): ResponseEntity<Map<String, String>> =
+        ResponseEntity
+            .status(HttpStatus.SERVICE_UNAVAILABLE)
+            .header(
+                HttpHeaders.RETRY_AFTER,
+                ex.retryAfterSeconds.toString()
+            )
+            .body(
+                mapOf(
+                    "error" to "data_unavailable",
+                    "message" to (
+                            ex.message
+                                ?: "Stations are temporarily unavailable"
+                            )
+                )
+            )
+
+    @ExceptionHandler(NsiStationsResponseException::class)
+    fun handleNsiStationsResponseException(
+        ex: NsiStationsResponseException
+    ): ResponseEntity<Map<String, String>> =
+        ResponseEntity
+            .status(HttpStatus.BAD_GATEWAY)
+            .body(
+                mapOf(
+                    "error" to "upstream_response_error",
+                    "message" to "НСИ вернула неожиданный HTTP-статус ${ex.statusCode}"
+                )
+            )
+
+}

services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/StationService.kt

@@ -1,32 +1,231 @@
 package space.nstart.pcp.pcp_stations_service.service
 
+
+import org.slf4j.LoggerFactory
 import org.springframework.beans.factory.annotation.Autowired
 import org.springframework.stereotype.Service
+import space.nstart.pcp.pcp_stations_service.configuration.ClassifierProperties
 import space.nstart.pcp.pcp_stations_service.configuration.CustomErrorException
 import space.nstart.pcp.pcp_stations_service.entity.StationEntity
 import space.nstart.pcp.pcp_stations_service.integration.NsiStationClient
+import space.nstart.pcp.pcp_stations_service.integration.NsiStationsUnavailableException
 import space.nstart.pcp.pcp_stations_service.repository.StationRepository
 import space.nstart.pcp.pcp_types_lib.dto.ballistics.StationDTO
 import java.util.UUID
+import java.time.Clock
+import java.time.Instant
 
 @Service
 class StationService(
+    private val stationRepository: StationRepository,
+    private val nsiStationClient: NsiStationClient,
+    private val classifierProperties: ClassifierProperties,
+    private val clock: Clock = Clock.systemUTC()
 ) {
 
-    @Autowired
-    private lateinit var stationRepository: StationRepository
+    private val log = LoggerFactory.getLogger(javaClass)
 
-    @Autowired
-    private lateinit var nsiStationClient: NsiStationClient
+    /*
+     * В каждый момент времени реальный refresh выполняет только один поток.
+     * Остальные запросы ожидают результат этого refresh.
+     */
+    private val cacheLock = Any()
 
-    fun all(): List<StationDTO> = nsiStationClient.fetchStations()
+    @Volatile
+    private var cachedStations: StationsCache? = null
+
+    /*
+     * После полного провала refresh временно не создаем нового лидера,
+     * чтобы очередь запросов не начала последовательно долбить лежащую НСИ.
+     */
+    @Volatile
+    private var refreshBlockedUntil: Instant = Instant.EPOCH
+
+
+    fun all(): List<StationDTO> =
+        loadStationsCache().stations
 
     fun byId(id: UUID): StationDTO =
-        nsiStationClient.fetchStations()
-            .firstOrNull { it.id == id }
+        loadStationsCache().byId[id]
             ?: throw CustomErrorException("Станция $id не найдена")
 
+
     fun add(station : StationDTO): StationDTO = stationRepository.save(StationEntity(station)).toDTO()
 
     fun delete(id : UUID): Int = stationRepository.deleteByStationId(id)
+
+    private fun loadStationsCache(): StationsCache {
+        val now = Instant.now(clock)
+
+        /*
+         * быстрый путь без блокировки
+         */
+        cachedStations
+            ?.takeIf { cache -> cache.isFresh(now) }
+            ?.let { cache ->
+                log.debug(
+                    "Stations cache hit. cachedCount={}, expiresAt={}",
+                    cache.stations.size,
+                    cache.expiresAt
+                )
+                return cache
             }
+
+        /*
+         * прошлый рефреш полностью провалился.
+         * во время колдауна НСИ повторно не вызываем.
+         */
+        if (now.isBefore(refreshBlockedUntil)) {
+            return cachedStations ?: throwDataUnavailable()
+        }
+
+        return synchronized(cacheLock) {
+            val lockedNow = Instant.now(clock)
+
+            /*
+             * двойная проверка пока запрос ожидал лок, другой поток мог
+             * уже успешно обновить кэш.
+             */
+            cachedStations
+                ?.takeIf { cache -> cache.isFresh(lockedNow) }
+                ?: if (lockedNow.isBefore(refreshBlockedUntil)) {
+                    cachedStations ?: throwDataUnavailable()
+                } else {
+                    refreshStationsOrFallback()
+                }
+        }
+    }
+
+    private fun refreshStationsOrFallback(): StationsCache {
+        return try {
+            val stations = fetchStationsWithRetry()
+            val now = Instant.now(clock)
+
+            val snapshotStations = stations.toList()
+
+            val refreshedCache = StationsCache(
+                stations = snapshotStations,
+                byId = snapshotStations
+                    .mapNotNull { station ->
+                        station.id?.let { id -> id to station }
+                    }
+                    .toMap(),
+                expiresAt = now.plus(classifierProperties.stationsCacheTtl)
+            )
+
+            cachedStations = refreshedCache
+            refreshBlockedUntil = Instant.EPOCH
+
+            log.info(
+                "Stations cache refreshed. cachedCount={}, expiresAt={}",
+                refreshedCache.stations.size,
+                refreshedCache.expiresAt
+            )
+
+            refreshedCache
+        } catch (ex: NsiStationsUnavailableException) {
+            refreshBlockedUntil = Instant.now(clock)
+                .plus(classifierProperties.stationsRefreshFailureCooldown)
+
+            val staleCache = cachedStations
+
+            if (staleCache != null) {
+                log.warn(
+                    "NSI stations are unavailable after refresh attempts. " +
+                            "Returning stale cache. cachedCount={}, cacheExpiredAt={}, " +
+                            "nextRefreshAllowedAt={}, status={}",
+                    staleCache.stations.size,
+                    staleCache.expiresAt,
+                    refreshBlockedUntil,
+                    ex.statusCode,
+                    ex
+                )
+
+                staleCache
+            } else {
+                log.warn(
+                    "NSI stations are unavailable and local cache is empty. " +
+                            "nextRefreshAllowedAt={}, status={}",
+                    refreshBlockedUntil,
+                    ex.statusCode,
+                    ex
+                )
+
+                throwDataUnavailable(ex)
+            }
+        }
+    }
+
+    private fun fetchStationsWithRetry(): List<StationDTO> {
+        val attempts = classifierProperties.stationsRefreshAttempts
+            .coerceAtLeast(1)
+
+        var lastFailure: NsiStationsUnavailableException? = null
+
+        for (attempt in 1..attempts) {
+            try {
+                log.info(
+                    "Refreshing stations from NSI. attempt={}/{}",
+                    attempt,
+                    attempts
+                )
+
+                return nsiStationClient.fetchStations()
+            } catch (ex: NsiStationsUnavailableException) {
+                lastFailure = ex
+
+                if (attempt == attempts) {
+                    break
+                }
+
+                log.warn(
+                    "NSI stations refresh attempt failed. " +
+                            "attempt={}/{}, status={}, retryDelay={}",
+                    attempt,
+                    attempts,
+                    ex.statusCode,
+                    classifierProperties.stationsRefreshRetryDelay
+                )
+
+                try {
+                    Thread.sleep(
+                        classifierProperties.stationsRefreshRetryDelay
+                            .toMillis()
+                            .coerceAtLeast(0)
+                    )
+                } catch (interrupted: InterruptedException) {
+                    Thread.currentThread().interrupt()
+                    throw ex
+                }
+            }
+        }
+
+        throw checkNotNull(lastFailure)
+    }
+
+    private fun throwDataUnavailable(
+        cause: Throwable? = null
+    ): Nothing {
+        val retryAfterSeconds = classifierProperties
+            .stationsRefreshFailureCooldown
+            .seconds
+            .coerceAtLeast(1)
+
+        throw StationsDataUnavailableException(
+            message = "Станции временно недоступны: НСИ не отвечает, локальный кэш пуст",
+            retryAfterSeconds = retryAfterSeconds,
+            cause = cause
+        )
+    }
+
+    private data class StationsCache(
+        val stations: List<StationDTO>,
+        val byId: Map<UUID, StationDTO>,
+        val expiresAt: Instant
+    ) {
+        fun isFresh(now: Instant): Boolean =
+            now.isBefore(expiresAt)
+    }
+
+}
+

services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/StationsDataUnavailableException.kt

@@ -0,0 +1,10 @@
+package space.nstart.pcp.pcp_stations_service.service
+
+/**
+ * Будет использоваться только при недоступности НСИ и без валидного кэша
+ * */
+class StationsDataUnavailableException(
+    message: String,
+    val retryAfterSeconds: Long,
+    cause: Throwable? = null
+) : RuntimeException(message, cause)