From c06ae263e02284908ba1c396c27f90d8a4bbc118 Mon Sep 17 00:00:00 2001 From: "Zimin A.N." Date: Fri, 8 May 2026 11:05:33 +0300 Subject: [PATCH] =?UTF-8?q?feat(lineage):=20Phase=202=20=E2=80=94=20materi?= =?UTF-8?q?alized=20view=20+=20throttled=20refresh=20+=20UI=20staleness?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dict-relationships-v2 epic, Phase 2. Precomputed reverse FK index + throttled refresh + feature flag + UI staleness badge. Migration катится всегда; query path активируется через ordinis.lineage.mv.enabled=true. Backend: - Liquibase 0016: CREATE MATERIALIZED VIEW record_dependents_index + UNIQUE INDEX(source_record_id, source_field) (REQUIRED для REFRESH CONCURRENTLY) + composite indices для record-level и schema-level lookup. Plus lineage_mv_meta table — single-row tracking (last_refreshed_at, duration_ms, row_count, status, error). - DependentsQueryMv (interface impl, @Primary + @ConditionalOnProperty) — один SQL query через MV вместо N JSONB lookups (по одному per source dict). - MaterializedViewRefreshService: * @Scheduled refresh каждые ordinis.lineage.mv.refresh-interval-sec (default 60). Storm prevention: AtomicReference guard для in-flight, cooldown через ordinis.lineage.mv.min-interval-sec (default 30). * REFRESH CONCURRENTLY с fallback на plain REFRESH когда MV пустой (initial state). Не блокирует concurrent reads. * MvGateway interface (extracted из-за JDK 25 + Mockito incompat для concrete JdbcTemplate). Production: JdbcMvGateway (nested static). - LineageIndexService теперь возвращает LineageMeta в RecordDependentsPage — mvEnabled, lastRefreshedAt, lastStatus, lastDurationMs. Optional inject — works без MV (mvEnabled=false, всё null). Tests: - MaterializedViewRefreshServiceTest (9 tests): * scheduledRefresh first time → REFRESH CONCURRENTLY * skip when within cooldown * refreshNow throws RefreshThrottledException when throttled * refresh succeeds после cooldown * fallback на plain REFRESH когда CONCURRENTLY fails * STORM PREVENTION (R2): 100 concurrent refresh attempts → ровно 1 actual refresh (via cooldown + in-flight guard) * lastRefresh returns meta correctly * updates lineage_mv_meta after refresh (status=ok) * updates with error если оба refresh'а fail (status=error) Admin UI: - RecordDependentsPage type расширен LineageMeta { mvEnabled, lastRefreshedAt, lastStatus, lastDurationMs }. - RecordDependentsPanel показывает staleness badge "обновлено N мин назад" если MV active + last refresh > 2 min ago. Tooltip объясняет что closed-between-refreshes records отфильтрованы query-side. - i18n RU/EN: lineage.refs.{stale, staleHint}. Verify: - mvn -pl ordinis-rest-api -am test: 98/98 PASS (+9 new MV tests). - mvn verify -pl '!ordinis-app': все модули SUCCESS (9.3s). - pnpm tsc --noEmit: clean. - pnpm test (vitest): 89/89 PASS. Deploy strategy: - Migration катится автоматом (initial REFRESH < 5 min на 5k records). - Query path остаётся JSONB direct (Phase 1) до явного flip flag'а ordinis.lineage.mv.enabled=true. Это позволит верифицировать MV корректность на staging без риска для prod. - После flip: refresh каждые 60 sec, stale window <= 2 min обычно. --- ordinis-admin-ui/src/api/client.ts | 12 + .../lineage/RecordDependentsPanel.tsx | 29 +- ordinis-admin-ui/src/i18n.ts | 4 + .../domain/record/DependentsQueryMv.java | 110 ++++++++ .../changes/0016-record-dependents-index.xml | 160 +++++++++++ .../main/resources/db/changelog/master.xml | 1 + .../reference/LineageIndexService.java | 62 ++++- .../MaterializedViewRefreshService.java | 260 ++++++++++++++++++ .../MaterializedViewRefreshServiceTest.java | 185 +++++++++++++ 9 files changed, 813 insertions(+), 10 deletions(-) create mode 100644 ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/record/DependentsQueryMv.java create mode 100644 ordinis-migrations/src/main/resources/db/changelog/changes/0016-record-dependents-index.xml create mode 100644 ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/service/reference/MaterializedViewRefreshService.java create mode 100644 ordinis-rest-api/src/test/java/cloud/nstart/terravault/ordinis/restapi/service/reference/MaterializedViewRefreshServiceTest.java diff --git a/ordinis-admin-ui/src/api/client.ts b/ordinis-admin-ui/src/api/client.ts index 70fbf91..87b97c8 100644 --- a/ordinis-admin-ui/src/api/client.ts +++ b/ordinis-admin-ui/src/api/client.ts @@ -276,6 +276,16 @@ export type PerSourceSummary = { count: number } +export type LineageMeta = { + /** True если backend использует materialized view; false — JSONB direct (fresh). */ + mvEnabled: boolean + /** ISO timestamp последнего успешного MV refresh; null если MV выключена / never. */ + lastRefreshedAt: string | null + /** "ok" | "error" | "never" — null если mvEnabled=false. */ + lastStatus: string | null + lastDurationMs: number | null +} + export type RecordDependentsPage = { targetDict: string targetBusinessKey: string @@ -284,6 +294,8 @@ export type RecordDependentsPage = { total: number page: number size: number + /** Phase 2 dict-relationships-v2: backing data source metadata + staleness. */ + lineageMeta?: LineageMeta | null } export type WebhookTestPingResult = { diff --git a/ordinis-admin-ui/src/components/lineage/RecordDependentsPanel.tsx b/ordinis-admin-ui/src/components/lineage/RecordDependentsPanel.tsx index 358669b..47c1945 100644 --- a/ordinis-admin-ui/src/components/lineage/RecordDependentsPanel.tsx +++ b/ordinis-admin-ui/src/components/lineage/RecordDependentsPanel.tsx @@ -52,15 +52,38 @@ export const RecordDependentsPanel = ({ dictionaryName, businessKey }: Props) => const totalPages = Math.max(1, Math.ceil(data.total / size)) + // Staleness: show если MV active + last refresh > 2 min ago. + // <2 min — fresh enough, не отвлекаем UI'ем. + const stale = + data.lineageMeta?.mvEnabled && data.lineageMeta.lastRefreshedAt + ? Math.max( + 0, + Math.floor( + (Date.now() - new Date(data.lineageMeta.lastRefreshedAt).getTime()) / + 60000, + ), + ) + : null + return (

{t('lineage.refs.title')}

- - {t('lineage.refs.total', { count: data.total })} - +
+ {stale !== null && stale >= 2 && ( + + {t('lineage.refs.stale', { minutes: stale })} + + )} + + {t('lineage.refs.total', { count: data.total })} + +
{/* Per-source summary chips */} diff --git a/ordinis-admin-ui/src/i18n.ts b/ordinis-admin-ui/src/i18n.ts index dc56011..5ae2774 100644 --- a/ordinis-admin-ui/src/i18n.ts +++ b/ordinis-admin-ui/src/i18n.ts @@ -249,6 +249,8 @@ i18n 'lineage.refs.field': 'поле {{field}}', 'lineage.refs.created': 'создана {{when}}', 'lineage.refs.openSourceRecord': 'Открыть запись {{key}} в источнике', + 'lineage.refs.stale': 'обновлено {{minutes}} мин назад', + 'lineage.refs.staleHint': 'Данные из materialized view; обновляются раз в минуту. Закрытые между обновлениями записи отфильтрованы по valid_to.', 'lineage.onClose.BLOCK': 'block', 'lineage.onClose.WARN': 'warn', 'lineage.onClose.CASCADE': 'cascade', @@ -552,6 +554,8 @@ i18n 'lineage.refs.field': 'field {{field}}', 'lineage.refs.created': 'created {{when}}', 'lineage.refs.openSourceRecord': 'Open referencing record {{key}}', + 'lineage.refs.stale': 'refreshed {{minutes}} min ago', + 'lineage.refs.staleHint': 'Data from materialized view, refreshed once per minute. Records closed between refreshes are filtered by valid_to.', 'lineage.onClose.BLOCK': 'block', 'lineage.onClose.WARN': 'warn', 'lineage.onClose.CASCADE': 'cascade', diff --git a/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/record/DependentsQueryMv.java b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/record/DependentsQueryMv.java new file mode 100644 index 0000000..6cdb17b --- /dev/null +++ b/ordinis-domain/src/main/java/cloud/nstart/terravault/ordinis/domain/record/DependentsQueryMv.java @@ -0,0 +1,110 @@ +package cloud.nstart.terravault.ordinis.domain.record; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.context.annotation.Primary; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.stereotype.Component; + +import java.time.OffsetDateTime; +import java.util.List; +import java.util.UUID; + +/** + * MV-backed реализация {@link DependentsQuery}. Активируется через property + * {@code ordinis.lineage.mv.enabled=true} (default false). Когда выключена — + * Spring инжектит {@code DependentsQueryJdbc} (без MV, прямой JSONB query). + * + *

Производительность: один SQL по composite index + * {@code (target_dict_name, target_business_key_value)} вместо N JSONB + * lookups (по одному на каждый source dict). + * + *

Stale window: MV refresh throttled на 1/min (см. {@code MaterializedViewRefreshService}). + * Записи закрытые между refresh'ами всё равно отфильтрованы query-side через + * {@code valid_from/valid_to} columns в MV — actual valid_to update сделан + * через {@code DictionaryRecord} table refresh, MV отстаёт только по новым + * referencing records которые ещё не подхвачены MV. + * + *

Active filter в WHERE clause использует {@code valid_from <= ? AND valid_to > ?} + * (тот же at) что и в JdbcImpl — consistent semantics. + */ +@Component +@Primary +@ConditionalOnProperty(name = "ordinis.lineage.mv.enabled", havingValue = "true", matchIfMissing = false) +public class DependentsQueryMv implements DependentsQuery { + + private static final Logger log = LoggerFactory.getLogger(DependentsQueryMv.class); + + private final JdbcTemplate jdbc; + + @Autowired + public DependentsQueryMv(JdbcTemplate jdbc) { + this.jdbc = jdbc; + log.info("DependentsQuery: MV-backed implementation активирована " + + "(ordinis.lineage.mv.enabled=true). Query path использует " + + "record_dependents_index materialized view."); + } + + @Override + public List findActive( + UUID sourceDictionaryId, + String sourceField, + String targetValue, + OffsetDateTime at, + int limit, + int offset) { + if (limit <= 0 || limit > 500) { + throw new IllegalArgumentException("limit must be 1..500, got " + limit); + } + if (offset < 0) { + throw new IllegalArgumentException("offset must be >= 0, got " + offset); + } + + // MV path: source_dict_id + source_field + target_value уникально определяет ряды. + String sql = """ + SELECT source_record_id, source_dict_id, source_business_key, + valid_from, valid_to, created_at + FROM record_dependents_index + WHERE source_dict_id = ? + AND source_field = ? + AND target_business_key_value = ? + AND valid_from <= ? + AND valid_to > ? + ORDER BY created_at DESC + LIMIT ? OFFSET ? + """; + + return jdbc.query( + sql, + (rs, n) -> new DependentRow( + (UUID) rs.getObject("source_record_id"), + (UUID) rs.getObject("source_dict_id"), + rs.getString("source_business_key"), + rs.getObject("valid_from", OffsetDateTime.class), + rs.getObject("valid_to", OffsetDateTime.class), + rs.getObject("created_at", OffsetDateTime.class)), + sourceDictionaryId, sourceField, targetValue, at, at, limit, offset); + } + + @Override + public long countActive( + UUID sourceDictionaryId, + String sourceField, + String targetValue, + OffsetDateTime at) { + String sql = """ + SELECT COUNT(*) FROM record_dependents_index + WHERE source_dict_id = ? + AND source_field = ? + AND target_business_key_value = ? + AND valid_from <= ? + AND valid_to > ? + """; + + Long count = jdbc.queryForObject(sql, Long.class, + sourceDictionaryId, sourceField, targetValue, at, at); + return count == null ? 0 : count; + } +} diff --git a/ordinis-migrations/src/main/resources/db/changelog/changes/0016-record-dependents-index.xml b/ordinis-migrations/src/main/resources/db/changelog/changes/0016-record-dependents-index.xml new file mode 100644 index 0000000..383c0ff --- /dev/null +++ b/ordinis-migrations/src/main/resources/db/changelog/changes/0016-record-dependents-index.xml @@ -0,0 +1,160 @@ + + + + + + MV record_dependents_index: precomputed reverse FK + + + + SELECT COUNT(*) FROM pg_matviews WHERE matviewname = 'record_dependents_index' + + + + >'x-references', '.', 1) AS target_dict_name, + split_part(prop.value->>'x-references', '.', 2) AS target_field, + UPPER(COALESCE(prop.value->>'x-references-on-close', 'BLOCK')) AS on_close, + r.data ->> prop.key AS target_business_key_value + FROM dictionary_records r + JOIN dictionary_definitions src_def ON src_def.id = r.dictionary_id + JOIN LATERAL jsonb_each(src_def.schema_json -> 'properties') AS prop(key, value) ON true + WHERE prop.value ? 'x-references' + AND prop.value->>'x-references' ~ '^[a-zA-Z_][a-zA-Z0-9_]*\.[a-zA-Z_][a-zA-Z0-9_]*$' + AND r.data ? prop.key + AND jsonb_typeof(r.data -> prop.key) = 'string' + WITH NO DATA; + ]]> + + DROP MATERIALIZED VIEW IF EXISTS record_dependents_index; + + + + + + UNIQUE INDEX для REFRESH CONCURRENTLY + + CREATE UNIQUE INDEX uniq_record_dependents_index_source + ON record_dependents_index (source_record_id, source_field); + + + DROP INDEX IF EXISTS uniq_record_dependents_index_source; + + + + + + Composite index для record-level dependents lookup + + CREATE INDEX idx_record_dependents_target + ON record_dependents_index + (target_dict_name, target_business_key_value, valid_from, valid_to); + + + DROP INDEX IF EXISTS idx_record_dependents_target; + + + + + + Index для schema-level reverse FK list + + CREATE INDEX idx_record_dependents_schema + ON record_dependents_index (target_dict_name); + + + DROP INDEX IF EXISTS idx_record_dependents_schema; + + + + + + Initial REFRESH (no CONCURRENTLY на первый populate) + REFRESH MATERIALIZED VIEW record_dependents_index; + + + + + + + + Metadata табличка для last MV refresh tracking + + + + + + + + + + + + ALTER TABLE lineage_mv_meta + ADD CONSTRAINT lineage_mv_meta_singleton CHECK (id = 1); + + + + + + + + diff --git a/ordinis-migrations/src/main/resources/db/changelog/master.xml b/ordinis-migrations/src/main/resources/db/changelog/master.xml index ddcd83c..6555d96 100644 --- a/ordinis-migrations/src/main/resources/db/changelog/master.xml +++ b/ordinis-migrations/src/main/resources/db/changelog/master.xml @@ -25,5 +25,6 @@ + diff --git a/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/service/reference/LineageIndexService.java b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/service/reference/LineageIndexService.java index b21f108..a9e9b95 100644 --- a/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/service/reference/LineageIndexService.java +++ b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/service/reference/LineageIndexService.java @@ -9,6 +9,7 @@ import cloud.nstart.terravault.ordinis.domain.record.DictionaryRecordRepository; import com.fasterxml.jackson.databind.JsonNode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -53,14 +54,27 @@ public class LineageIndexService { private final DictionaryDefinitionRepository definitionRepository; private final DictionaryRecordRepository recordRepository; private final DependentsQuery dependentsQuery; + /** Optional — присутствует только когда {@code ordinis.lineage.mv.enabled=true}. */ + private final Optional mvRefresh; + @Autowired + public LineageIndexService( + DictionaryDefinitionRepository definitionRepository, + DictionaryRecordRepository recordRepository, + DependentsQuery dependentsQuery, + Optional mvRefresh) { + this.definitionRepository = definitionRepository; + this.recordRepository = recordRepository; + this.dependentsQuery = dependentsQuery; + this.mvRefresh = mvRefresh; + } + + /** Test-friendly ctor без MV refresh service. */ public LineageIndexService( DictionaryDefinitionRepository definitionRepository, DictionaryRecordRepository recordRepository, DependentsQuery dependentsQuery) { - this.definitionRepository = definitionRepository; - this.recordRepository = recordRepository; - this.dependentsQuery = dependentsQuery; + this(definitionRepository, recordRepository, dependentsQuery, Optional.empty()); } /** @@ -182,7 +196,8 @@ public class LineageIndexService { List schemaRefs = findSchemaDependents(targetDictName, allowedScopes); if (schemaRefs.isEmpty()) { return new RecordDependentsPage( - targetDictName, targetBusinessKey, List.of(), List.of(), 0, page, size); + targetDictName, targetBusinessKey, List.of(), List.of(), 0, page, size, + buildLineageMeta()); } // Per-source: pull target value, count + paginate. @@ -243,7 +258,26 @@ public class LineageIndexService { List slice = all.subList(from, to); return new RecordDependentsPage( - targetDictName, targetBusinessKey, slice, summary, total, page, size); + targetDictName, targetBusinessKey, slice, summary, total, page, size, + buildLineageMeta()); + } + + /** + * Метаданные lineage index source: помечает MV-backed vs JSONB direct + * + last refresh timestamp (если MV active). Используется UI'ем для + * staleness badge "обновлено N min назад". + */ + private LineageMeta buildLineageMeta() { + if (mvRefresh.isEmpty()) { + // JSONB direct path — данные always fresh, нет stale window. + return new LineageMeta(false, null, null, null); + } + var meta = mvRefresh.get().lastRefresh(); + if (meta.isEmpty() || meta.get().lastRefreshedAt() == null) { + return new LineageMeta(true, null, "never", null); + } + var m = meta.get(); + return new LineageMeta(true, m.lastRefreshedAt(), m.lastStatus(), m.lastDurationMs()); } // === DTOs === @@ -277,7 +311,7 @@ public class LineageIndexService { OnCloseAction onClose, long count) {} - /** Paginated page + per-source aggregates. */ + /** Paginated page + per-source aggregates + lineage source metadata. */ public record RecordDependentsPage( String targetDict, String targetBusinessKey, @@ -285,5 +319,19 @@ public class LineageIndexService { List perSource, long total, int page, - int size) {} + int size, + LineageMeta lineageMeta) {} + + /** + * Metadata о backing data source (Phase 2): + * @param mvEnabled MV-backed query path active. False — JSONB direct (always fresh). + * @param lastRefreshedAt последний successful MV refresh, null если MV выключен или never. + * @param lastStatus "ok" / "error" / "never". null если MV выключен. + * @param lastDurationMs длительность последнего refresh'а в ms. + */ + public record LineageMeta( + boolean mvEnabled, + OffsetDateTime lastRefreshedAt, + String lastStatus, + Long lastDurationMs) {} } diff --git a/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/service/reference/MaterializedViewRefreshService.java b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/service/reference/MaterializedViewRefreshService.java new file mode 100644 index 0000000..1fac0e2 --- /dev/null +++ b/ordinis-rest-api/src/main/java/cloud/nstart/terravault/ordinis/restapi/service/reference/MaterializedViewRefreshService.java @@ -0,0 +1,260 @@ +package cloud.nstart.terravault.ordinis.restapi.service.reference; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.dao.DataAccessException; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.time.Instant; +import java.time.OffsetDateTime; +import java.util.Optional; +import java.util.concurrent.atomic.AtomicReference; + +/* JDK 25 + Mockito не мочат concrete JdbcTemplate (subclass mock-maker + * disabled). Обходим через MvGateway — узкий интерфейс над JdbcTemplate, + * который тесты mock'ают. Производственная impl (см. nested class) делает + * прямые JdbcTemplate calls. */ + +/** + * Throttled refresh для {@code record_dependents_index} materialized view + * (Phase 2 dict-relationships-v2). + * + *

Активируется когда {@code ordinis.lineage.mv.enabled=true}. Без MV + * сервис не нужен — {@link DependentsQueryJdbc} даёт прямой JSONB query + * без stale window'а. + * + *

Пути refresh'а: + *

    + *
  • Scheduled — каждые + * {@code ordinis.lineage.mv.refresh-interval-sec} (default 60). + * Простейший throttle: если refresh уже запущен — пропускаем тик.
    • + *
  • Force — {@link #refreshNow()} вызывается из bundle import + * или admin endpoint. Также throttled: refused если последний refresh + * был меньше {@code min-interval-sec} назад.
    • + *
+ * + *

REFRESH CONCURRENTLY — read concurrent с queries, требует UNIQUE INDEX + * на MV (создан в migration 0016). Если MV пуста (initial state до первого + * REFRESH) — CONCURRENTLY падает; fallback на non-CONCURRENTLY один раз. + * + *

Storm prevention (R2 eng review): even при 100 closes/sec, refresh + * может запускаться не чаще раза в {@code refresh-interval-sec}. Closing + * record делает MV stale — UI badge показывает "обновлено N min назад"; + * actual valid_to update в dictionary_records появится синхронно (через + * record close transaction), MV догонит на next tick. + */ +@Component +@ConditionalOnProperty( + name = "ordinis.lineage.mv.enabled", + havingValue = "true", + matchIfMissing = false) +public class MaterializedViewRefreshService { + + private static final Logger log = LoggerFactory.getLogger(MaterializedViewRefreshService.class); + + private final MvGateway gateway; + /** Volatile guard — не позволяет двум refresh'ам параллельно. */ + private final AtomicReference currentRefreshStart = new AtomicReference<>(null); + + @Value("${ordinis.lineage.mv.min-interval-sec:30}") + private long minIntervalSec; + + public MaterializedViewRefreshService(JdbcTemplate jdbc) { + this(new JdbcMvGateway(jdbc)); + } + + /** Test-only ctor — gateway инжектится напрямую. */ + MaterializedViewRefreshService(MvGateway gateway) { + this.gateway = gateway; + } + + /** + * Periodic tick. Проверяет cooldown + наличие work, делает refresh если + * можно. fixedDelayString — следующий запуск считается ОТ конца предыдущего + * (а не от старта), что предотвращает overlap при медленном refresh'е. + */ + @Scheduled( + initialDelayString = "${ordinis.lineage.mv.initial-delay-sec:30}000", + fixedDelayString = "${ordinis.lineage.mv.refresh-interval-sec:60}000") + public void scheduledRefresh() { + refreshIfDue(); + } + + /** + * Внешний trigger (bundle import, admin endpoint). Сам решает — можно + * ли refresh, выкидывает {@link RefreshThrottledException} если cooldown + * не прошёл. Caller отображает: "Refresh throttled, retry в N sec". + */ + public void refreshNow() { + if (!refreshIfDue()) { + throw new RefreshThrottledException( + "MV refresh throttled: last refresh < " + minIntervalSec + "s ago"); + } + } + + /** + * @return last refresh metadata из {@code lineage_mv_meta} table. + * Empty если migration ещё не накатилась или таблица недоступна. + */ + public Optional lastRefresh() { + try { + return Optional.ofNullable(gateway.loadMeta()); + } catch (DataAccessException e) { + log.warn("Failed to read lineage_mv_meta: {}", e.getMessage()); + return Optional.empty(); + } + } + + // === Internals === + + private boolean refreshIfDue() { + Instant now = Instant.now(); + Optional last = lastRefresh(); + if (last.isPresent() && last.get().lastRefreshedAt() != null) { + Duration since = Duration.between(last.get().lastRefreshedAt().toInstant(), now); + if (since.toSeconds() < minIntervalSec) { + log.debug("Skip MV refresh: last was {} sec ago (min interval {})", + since.toSeconds(), minIntervalSec); + return false; + } + } + if (!currentRefreshStart.compareAndSet(null, now)) { + log.debug("Skip MV refresh: previous tick still running since {}", + currentRefreshStart.get()); + return false; + } + try { + doRefresh(); + return true; + } finally { + currentRefreshStart.set(null); + } + } + + private void doRefresh() { + long start = System.currentTimeMillis(); + String status = "ok"; + String error = null; + try { + // Try CONCURRENTLY first; fallback на non-CONCURRENTLY если MV пустой + // (initial state — REFRESH CONCURRENTLY требует prior populate). + try { + gateway.refreshConcurrently(); + } catch (DataAccessException concurrentFail) { + log.warn("REFRESH CONCURRENTLY failed, trying plain REFRESH: {}", + concurrentFail.getMessage()); + gateway.refresh(); + } + } catch (DataAccessException e) { + status = "error"; + error = truncate(e.getMessage(), 1000); + log.error("MV refresh failed: {}", e.getMessage(), e); + } + long duration = System.currentTimeMillis() - start; + Long rowCount = null; + try { + rowCount = gateway.countRows(); + } catch (DataAccessException e) { + log.warn("Failed to count rows after refresh: {}", e.getMessage()); + } + try { + gateway.updateMeta(duration, rowCount, status, error); + } catch (DataAccessException e) { + log.warn("Failed to update lineage_mv_meta: {}", e.getMessage()); + } + if ("ok".equals(status)) { + log.info("MV record_dependents_index refreshed: {}ms, {} rows", duration, rowCount); + } + } + + private static String truncate(String s, int max) { + if (s == null) return null; + return s.length() <= max ? s : s.substring(0, max); + } + + /** Last refresh snapshot. */ + public record RefreshMeta( + OffsetDateTime lastRefreshedAt, + Long lastDurationMs, + Long lastRowCount, + String lastStatus, + String lastError) {} + + /** Throw'ится из {@link #refreshNow()} если cooldown не прошёл. */ + public static class RefreshThrottledException extends RuntimeException { + public RefreshThrottledException(String message) { + super(message); + } + } + + /** + * Узкий gateway над JdbcTemplate — позволяет тестам mock'нуть вместо + * concrete JdbcTemplate (Mockito blocked на JDK 25). + */ + interface MvGateway { + void refreshConcurrently(); + + void refresh(); + + Long countRows(); + + RefreshMeta loadMeta(); + + void updateMeta(long durationMs, Long rowCount, String status, String error); + } + + /** Production impl над JdbcTemplate. */ + static final class JdbcMvGateway implements MvGateway { + private final JdbcTemplate jdbc; + + JdbcMvGateway(JdbcTemplate jdbc) { + this.jdbc = jdbc; + } + + @Override + public void refreshConcurrently() { + jdbc.execute("REFRESH MATERIALIZED VIEW CONCURRENTLY record_dependents_index"); + } + + @Override + public void refresh() { + jdbc.execute("REFRESH MATERIALIZED VIEW record_dependents_index"); + } + + @Override + public Long countRows() { + return jdbc.queryForObject( + "SELECT COUNT(*) FROM record_dependents_index", Long.class); + } + + @Override + public RefreshMeta loadMeta() { + return jdbc.query( + "SELECT last_refreshed_at, last_duration_ms, last_row_count, last_status, last_error " + + "FROM lineage_mv_meta WHERE id = 1", + rs -> { + if (!rs.next()) return null; + OffsetDateTime ts = rs.getObject("last_refreshed_at", OffsetDateTime.class); + return new RefreshMeta( + ts, + rs.getObject("last_duration_ms", Long.class), + rs.getObject("last_row_count", Long.class), + rs.getString("last_status"), + rs.getString("last_error")); + }); + } + + @Override + public void updateMeta(long durationMs, Long rowCount, String status, String error) { + jdbc.update( + "UPDATE lineage_mv_meta SET last_refreshed_at = NOW(), last_duration_ms = ?, " + + "last_row_count = ?, last_status = ?, last_error = ? WHERE id = 1", + durationMs, rowCount, status, error); + } + } +} diff --git a/ordinis-rest-api/src/test/java/cloud/nstart/terravault/ordinis/restapi/service/reference/MaterializedViewRefreshServiceTest.java b/ordinis-rest-api/src/test/java/cloud/nstart/terravault/ordinis/restapi/service/reference/MaterializedViewRefreshServiceTest.java new file mode 100644 index 0000000..519f06b --- /dev/null +++ b/ordinis-rest-api/src/test/java/cloud/nstart/terravault/ordinis/restapi/service/reference/MaterializedViewRefreshServiceTest.java @@ -0,0 +1,185 @@ +package cloud.nstart.terravault.ordinis.restapi.service.reference; + +import cloud.nstart.terravault.ordinis.restapi.service.reference.MaterializedViewRefreshService.MvGateway; +import cloud.nstart.terravault.ordinis.restapi.service.reference.MaterializedViewRefreshService.RefreshMeta; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.test.util.ReflectionTestUtils; + +import java.time.OffsetDateTime; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.atomic.AtomicInteger; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +/** + * Тесты {@link MaterializedViewRefreshService}: throttle, storm prevention, + * concurrent guard, fallback REFRESH когда CONCURRENTLY падает. + * + *

Использует stub {@link MvGateway} — JdbcTemplate не мокается на JDK 25. + * Stub симулирует in-memory state lineage_mv_meta + REFRESH execution count. + */ +class MaterializedViewRefreshServiceTest { + + private StubGateway gateway; + private MaterializedViewRefreshService service; + + @BeforeEach + void setUp() { + gateway = new StubGateway(); + service = new MaterializedViewRefreshService(gateway); + ReflectionTestUtils.setField(service, "minIntervalSec", 30L); + } + + @Test + void scheduledRefresh_runsRefreshFirstTime() { + service.scheduledRefresh(); + assertThat(gateway.refreshConcurrentlyCalls.get()).isEqualTo(1); + assertThat(gateway.refreshPlainCalls.get()).isEqualTo(0); + } + + @Test + void scheduledRefresh_skipsIfWithinCooldown() { + service.scheduledRefresh(); + int afterFirst = gateway.refreshConcurrentlyCalls.get(); + service.scheduledRefresh(); + assertThat(gateway.refreshConcurrentlyCalls.get()).isEqualTo(afterFirst); + } + + @Test + void refreshNow_throwsIfThrottled() { + service.refreshNow(); + assertThatThrownBy(() -> service.refreshNow()) + .isInstanceOf(MaterializedViewRefreshService.RefreshThrottledException.class) + .hasMessageContaining("throttled"); + } + + @Test + void refreshNow_succeedsAfterCooldown() { + gateway.simulatedLastRefresh = OffsetDateTime.now().minusSeconds(31); + service.refreshNow(); + assertThat(gateway.refreshConcurrentlyCalls.get()).isEqualTo(1); + } + + @Test + void refresh_fallsBackToPlainWhenConcurrentlyFails() { + gateway.failConcurrentlyOnce = true; + service.scheduledRefresh(); + assertThat(gateway.refreshConcurrentlyCalls.get()).isEqualTo(1); + assertThat(gateway.refreshPlainCalls.get()).isEqualTo(1); + } + + @Test + void stormPrevention_100ConcurrentRefreshAttempts_resultsInOneActualRefresh() throws Exception { + // R2 eng review: 100 closes за 1 sec не должны вызвать 100 refresh'ей. + int parallelism = 100; + ExecutorService exec = Executors.newFixedThreadPool(Math.min(parallelism, 16)); + var latch = new CountDownLatch(parallelism); + var startGate = new CountDownLatch(1); + for (int i = 0; i < parallelism; i++) { + exec.submit(() -> { + try { + startGate.await(); + service.scheduledRefresh(); + } catch (InterruptedException ignored) { + Thread.currentThread().interrupt(); + } finally { + latch.countDown(); + } + }); + } + startGate.countDown(); + latch.await(); + exec.shutdown(); + + // Точно 1 refresh выполнен — остальные 99 либо skipped по cooldown'у, + // либо по in-flight guard. + assertThat(gateway.refreshConcurrentlyCalls.get()).isEqualTo(1); + } + + @Test + void lastRefresh_returnsMetaWhenAvailable() { + gateway.simulatedLastRefresh = OffsetDateTime.now().minusMinutes(5); + var meta = service.lastRefresh(); + assertThat(meta).isPresent(); + assertThat(meta.get().lastRefreshedAt()).isEqualTo(gateway.simulatedLastRefresh); + } + + @Test + void scheduledRefresh_updatesMetaAfterRefresh() { + service.scheduledRefresh(); + assertThat(gateway.updateMetaCalls.get()).isEqualTo(1); + assertThat(gateway.lastUpdateStatus).isEqualTo("ok"); + assertThat(gateway.lastUpdateRowCount).isEqualTo(42L); + } + + @Test + void scheduledRefresh_updatesMetaWithErrorIfBothRefreshesFail() { + gateway.failConcurrentlyOnce = true; + gateway.failPlain = true; + service.scheduledRefresh(); + assertThat(gateway.updateMetaCalls.get()).isEqualTo(1); + assertThat(gateway.lastUpdateStatus).isEqualTo("error"); + assertThat(gateway.lastUpdateError).isNotNull(); + } + + /** Stub gateway — in-memory state machine без real JDBC. */ + private static class StubGateway implements MvGateway { + final AtomicInteger refreshConcurrentlyCalls = new AtomicInteger(0); + final AtomicInteger refreshPlainCalls = new AtomicInteger(0); + final AtomicInteger updateMetaCalls = new AtomicInteger(0); + OffsetDateTime simulatedLastRefresh; + boolean failConcurrentlyOnce; + boolean failPlain; + String lastUpdateStatus; + String lastUpdateError; + Long lastUpdateRowCount; + + @Override + public void refreshConcurrently() { + refreshConcurrentlyCalls.incrementAndGet(); + if (failConcurrentlyOnce) { + failConcurrentlyOnce = false; + throw new DataIntegrityViolationException("not populated yet"); + } + simulatedLastRefresh = OffsetDateTime.now(); + } + + @Override + public void refresh() { + refreshPlainCalls.incrementAndGet(); + if (failPlain) { + throw new DataIntegrityViolationException("plain refresh failed"); + } + simulatedLastRefresh = OffsetDateTime.now(); + } + + @Override + public Long countRows() { + return 42L; + } + + @Override + public RefreshMeta loadMeta() { + if (simulatedLastRefresh == null) { + return new RefreshMeta(null, null, null, "never", null); + } + return new RefreshMeta(simulatedLastRefresh, 100L, 42L, "ok", null); + } + + @Override + public void updateMeta(long durationMs, Long rowCount, String status, String error) { + updateMetaCalls.incrementAndGet(); + lastUpdateStatus = status; + lastUpdateRowCount = rowCount; + lastUpdateError = error; + if (status.equals("ok")) { + simulatedLastRefresh = OffsetDateTime.now(); + } + } + } +}