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:
@@ -88,3 +99,4 @@ spring:
     max-in-memory-size: 20MB
 server:
   port: ${pcp.ports.stations:8080}
+

deploy/local-bundle/docker-compose.yml

@@ -64,7 +64,7 @@ services:
     volumes:
       - kafka-data:/var/lib/kafka/data
     healthcheck:
-      test: ["CMD-SHELL", "kafka-topics.sh --bootstrap-server kafka:9092 --list >/dev/null 2>&1"]
+      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --list >/dev/null 2>&1"]
       interval: 10s
       timeout: 10s
       retries: 30

deploy/local-jar/run-services.sh

@@ -192,7 +192,7 @@ wait_for_postgres() {
 wait_for_kafka() {
   local attempt
   for attempt in {1..60}; do
-    if compose exec -T kafka kafka-topics.sh --bootstrap-server kafka:9092 --list >/dev/null 2>&1; then
+    if compose exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server kafka:9092 --list >/dev/null 2>&1; then
       echo "Kafka is ready."
       return 0
     fi

services/pcp-stations-service/build.gradle.kts

@@ -50,6 +50,11 @@ dependencies {
 
 	implementation("org.locationtech.jts:jts-core:1.19.0")
 
+	// in‑memory cache с поддержкой TTL, позволяет задать TTL и автоматически удалять устаревшие записи.
+	implementation("com.github.ben-manes.caffeine:caffeine:3.1.8")
+
+	// обёртка для «circuit breaker», «retry» и т.п. защищает от падения при ошибках сети (circuit‑breaker + fallback).
+	implementation("org.springframework.cloud:spring-cloud-starter-circuitbreaker-resilience4j")
 
 	testImplementation("junit:junit")
 	testImplementation("org.jetbrains.kotlin:kotlin-test-junit5")
@@ -65,7 +70,7 @@ dependencyManagement {
 	imports {
 		mavenBom("org.springframework.cloud:spring-cloud-dependencies:${property("versions.spring.cloud")}")
 		mavenBom("org.testcontainers:testcontainers-bom:${property("versions.testcontainers")}")
-	}
+    }
 }
 
 tasks.withType<Jar> {

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

@@ -1,9 +1,11 @@
 package space.nstart.pcp.pcp_stations_service
 
 import org.springframework.boot.autoconfigure.SpringBootApplication
+import org.springframework.boot.context.properties.ConfigurationPropertiesScan
 import org.springframework.boot.runApplication
 
 @SpringBootApplication
+@ConfigurationPropertiesScan
 class PcpStationsServiceApplication
 
 fun main(args: Array<String>) {

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

@@ -0,0 +1,13 @@
+package space.nstart.pcp.pcp_stations_service.configuration
+
+import org.springframework.boot.context.properties.ConfigurationProperties
+import java.time.Duration
+
+@ConfigurationProperties(prefix = "classifier")
+data class ClassifierProperties(
+    val stationsUrl: String = "https://ordinis.k8s.265.nstart.cloud/api/v1/ground_station/records",
+    val stationsCacheTtl: Duration = Duration.ofMinutes(15),
+    val stationsRefreshAttempts: Int = 3,
+    val stationsRefreshRetryDelay: Duration = Duration.ofMillis(300),
+    val stationsRefreshFailureCooldown: Duration = Duration.ofSeconds(5)
+)

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

@@ -0,0 +1,12 @@
+package space.nstart.pcp.pcp_stations_service.configuration
+
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Configuration
+import org.springframework.web.reactive.function.client.WebClient
+
+@Configuration
+class WebClientConfig {
+
+    @Bean
+    fun webClientBuilder(): WebClient.Builder = WebClient.builder()
+}

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

@@ -0,0 +1,159 @@
+package space.nstart.pcp.pcp_stations_service.integration
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties
+import com.fasterxml.jackson.annotation.JsonProperty
+import org.slf4j.LoggerFactory
+import org.springframework.http.MediaType
+import org.springframework.stereotype.Component
+import org.springframework.web.reactive.function.client.WebClient
+import org.springframework.web.reactive.function.client.WebClientRequestException
+import org.springframework.web.reactive.function.client.WebClientResponseException
+import space.nstart.pcp.pcp_stations_service.configuration.ClassifierProperties
+import space.nstart.pcp.pcp_types_lib.dto.ballistics.PositionDTO
+import space.nstart.pcp.pcp_types_lib.dto.ballistics.StationDTO
+import java.time.Duration
+import java.util.UUID
+
+@Component
+class NsiStationClient(
+    webClientBuilder: WebClient.Builder,
+    private val classifierProperties: ClassifierProperties,
+) {
+    private val log = LoggerFactory.getLogger(javaClass)
+
+    private val webClient: WebClient = webClientBuilder.build()
+
+    private fun classifyResponseException(
+        ex: WebClientResponseException
+    ): NsiStationsException {
+        val statusCode = ex.statusCode.value()
+
+        return if (
+            statusCode == 408 ||
+            statusCode == 429 ||
+            statusCode in 500..599
+        ) {
+            NsiStationsUnavailableException(
+                classifierUrl = classifierProperties.stationsUrl,
+                statusCode = statusCode,
+                cause = ex
+            )
+        } else {
+            NsiStationsResponseException(
+                classifierUrl = classifierProperties.stationsUrl,
+                statusCode = statusCode,
+                cause = ex
+            )
+        }
+    }
+
+    fun fetchStations(): List<StationDTO> {
+        log.info(
+            "Fetching stations from NSI. url={}",
+            classifierProperties.stationsUrl
+        )
+
+        val response: List<NsiStationRecord> = try {
+            webClient.get()
+                .uri(classifierProperties.stationsUrl)
+                .accept(MediaType.APPLICATION_JSON)
+                .retrieve()
+                .bodyToMono(Array<NsiStationRecord>::class.java)
+                .block(Duration.ofSeconds(10))
+                ?.toList()
+                ?: throw NsiStationsUnavailableException(
+                    classifierUrl = classifierProperties.stationsUrl,
+                    statusCode = null,
+                    cause = IllegalStateException("NSI response body is absent")
+                )
+        } catch (ex: NsiStationsException) {
+            throw ex
+        } catch (ex: WebClientResponseException) {
+            throw classifyResponseException(ex)
+        } catch (ex: WebClientRequestException) {
+            throw NsiStationsUnavailableException(
+                classifierUrl = classifierProperties.stationsUrl,
+                statusCode = null,
+                cause = ex
+            )
+        } catch (ex: IllegalStateException) {
+            /*
+             * Сюда, в частности, попадает timeout от block(Duration).
+             */
+            throw NsiStationsUnavailableException(
+                classifierUrl = classifierProperties.stationsUrl,
+                statusCode = null,
+                cause = ex
+            )
+        }
+
+        return response.mapIndexed { index, station ->
+            station.toStationDTO(number = index + 1)
+        }
+    }
+
+    private fun NsiStationRecord.toStationDTO(number: Int): StationDTO {
+        return StationDTO(
+            id = id.toUuidOrNull(),
+            number = number,
+            name = data?.name ?: businessKey ?: "",
+            position = PositionDTO(
+                lat = data?.lat ?: 0.0,
+                long = data?.lon ?: 0.0,
+                height = data?.elevationM ?: 0.0
+            ),
+            elevationMin = data?.elevationMinDeg ?: 0.0,
+            elevationMax = 90.0
+        )
+    }
+
+    private fun String?.toUuidOrNull(): UUID? =
+        this?.let {
+            runCatching { UUID.fromString(it) }.getOrNull()
+        }
+}
+
+sealed class NsiStationsException(
+    message: String,
+    cause: Throwable?
+) : RuntimeException(message, cause)
+
+class NsiStationsUnavailableException(
+    val classifierUrl: String,
+    val statusCode: Int?,
+    cause: Throwable?
+) : NsiStationsException(
+    message = "NSI stations classifier is temporarily unavailable: $classifierUrl",
+    cause = cause
+)
+
+class NsiStationsResponseException(
+    val classifierUrl: String,
+    val statusCode: Int,
+    cause: Throwable?
+) : NsiStationsException(
+    message = "NSI stations classifier returned HTTP $statusCode: $classifierUrl",
+    cause = cause
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class NsiStationRecord(
+    val id: String? = null,
+    val businessKey: String? = null,
+    val data: NsiStationData? = null
+)
+
+@JsonIgnoreProperties(ignoreUnknown = true)
+data class NsiStationData(
+    val lat: Double? = null,
+    val lon: Double? = null,
+    val code: String? = null,
+    val name: String? = null,
+    val status: String? = null,
+
+    @JsonProperty("elevation_m")
+    val elevationM: Double? = null,
+
+    @JsonProperty("elevation_min_deg")
+    val elevationMinDeg: Double? = null
+)

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,30 +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()
+) {
+
+    private val log = LoggerFactory.getLogger(javaClass)
+
+    /*
+     * В каждый момент времени реальный refresh выполняет только один поток.
+     * Остальные запросы ожидают результат этого refresh.
+     */
+    private val cacheLock = Any()
+
+    @Volatile
+    private var cachedStations: StationsCache? = null
+
+    /*
+     * После полного провала refresh временно не создаем нового лидера,
+     * чтобы очередь запросов не начала последовательно долбить лежащую НСИ.
+     */
+    @Volatile
+    private var refreshBlockedUntil: Instant = Instant.EPOCH
 
 
-    @Autowired
-    private lateinit var stationRepository: StationRepository
+    fun all(): List<StationDTO> =
+        loadStationsCache().stations
 
+    fun byId(id: UUID): StationDTO =
+        loadStationsCache().byId[id]
+            ?: throw CustomErrorException("Станция $id не найдена")
 
 
-    fun all() = stationRepository.findAll().map { it.toDTO() }
-
     fun add(station : StationDTO): StationDTO = stationRepository.save(StationEntity(station)).toDTO()
 
-    fun byId(id : UUID): StationDTO =
-        stationRepository.findByStationId(id)
-            .map { it.toDTO() }
-            .orElseThrow { CustomErrorException("Станция $id не найдена") }
-
     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)

services/pcp-stations-service/src/test/kotlin/space/nstart/pcp/pcp_stations_service/controller/StationControllerTest.kt

@@ -51,6 +51,12 @@ class StationControllerTest {
         verify(stationService).add(anyStation())
     }
 
+    @Test
+    fun `get all stations`() {
+        val stations = stationService.getStations();
+        assertEquals(8, stations.size)
+    }
+
     @Test
     fun `station get by id endpoint returns dto without optional wrapper`() {
         val station = sampleStation()