From b895a1682f6b116d777982e8f146b5f040d035a1 Mon Sep 17 00:00:00 2001 From: "Ivan I. Ovchinnikov" Date: Thu, 16 Jul 2026 20:23:50 +0300 Subject: [PATCH] refetch cache and ttl --- config-repo/pcp-stations-service.yaml | 21 +- .../service/GlobalExceptionHandle.kt | 43 ++++ .../service/StationService.kt | 217 +++++++++++++++++- .../StationsDataUnavailableException.kt | 10 + 4 files changed, 272 insertions(+), 19 deletions(-) create mode 100644 services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/GlobalExceptionHandle.kt create mode 100644 services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/StationsDataUnavailableException.kt diff --git a/config-repo/pcp-stations-service.yaml b/config-repo/pcp-stations-service.yaml index 21530ea..eab6560 100644 --- a/config-repo/pcp-stations-service.yaml +++ b/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} diff --git a/services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/GlobalExceptionHandle.kt b/services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/GlobalExceptionHandle.kt new file mode 100644 index 0000000..57de6d8 --- /dev/null +++ b/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> = + 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> = + ResponseEntity + .status(HttpStatus.BAD_GATEWAY) + .body( + mapOf( + "error" to "upstream_response_error", + "message" to "НСИ вернула неожиданный HTTP-статус ${ex.statusCode}" + ) + ) + +} \ No newline at end of file diff --git a/services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/StationService.kt b/services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/StationService.kt index 6f7a01d..f2eae86 100644 --- a/services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/StationService.kt +++ b/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 ( -){ +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 = nsiStationClient.fetchStations() + @Volatile + private var cachedStations: StationsCache? = null + + /* + * После полного провала refresh временно не создаем нового лидера, + * чтобы очередь запросов не начала последовательно долбить лежащую НСИ. + */ + @Volatile + private var refreshBlockedUntil: Instant = Instant.EPOCH + + + fun all(): List = + 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 { + 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, + val byId: Map, + val expiresAt: Instant + ) { + fun isFresh(now: Instant): Boolean = + now.isBefore(expiresAt) + } + } + diff --git a/services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/StationsDataUnavailableException.kt b/services/pcp-stations-service/src/main/kotlin/space/nstart/pcp/pcp_stations_service/service/StationsDataUnavailableException.kt new file mode 100644 index 0000000..219a4ef --- /dev/null +++ b/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)