feat(lineage): Phase 2 — materialized view + throttled refresh + UI staleness
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<MV>
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 обычно.
This commit is contained in:
@@ -276,6 +276,16 @@ export type PerSourceSummary = {
|
|||||||
count: number
|
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 = {
|
export type RecordDependentsPage = {
|
||||||
targetDict: string
|
targetDict: string
|
||||||
targetBusinessKey: string
|
targetBusinessKey: string
|
||||||
@@ -284,6 +294,8 @@ export type RecordDependentsPage = {
|
|||||||
total: number
|
total: number
|
||||||
page: number
|
page: number
|
||||||
size: number
|
size: number
|
||||||
|
/** Phase 2 dict-relationships-v2: backing data source metadata + staleness. */
|
||||||
|
lineageMeta?: LineageMeta | null
|
||||||
}
|
}
|
||||||
|
|
||||||
export type WebhookTestPingResult = {
|
export type WebhookTestPingResult = {
|
||||||
|
|||||||
@@ -52,15 +52,38 @@ export const RecordDependentsPanel = ({ dictionaryName, businessKey }: Props) =>
|
|||||||
|
|
||||||
const totalPages = Math.max(1, Math.ceil(data.total / size))
|
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 (
|
return (
|
||||||
<section className="space-y-3">
|
<section className="space-y-3">
|
||||||
<div className="flex items-center justify-between">
|
<div className="flex items-center justify-between">
|
||||||
<h3 className="text-sm font-primary text-carbon">
|
<h3 className="text-sm font-primary text-carbon">
|
||||||
{t('lineage.refs.title')}
|
{t('lineage.refs.title')}
|
||||||
</h3>
|
</h3>
|
||||||
<span className="text-2xs text-carbon/60">
|
<div className="flex items-center gap-2">
|
||||||
{t('lineage.refs.total', { count: data.total })}
|
{stale !== null && stale >= 2 && (
|
||||||
</span>
|
<span
|
||||||
|
className="text-2xs text-carbon/60 px-2 py-0.5 rounded-sm border border-regolith"
|
||||||
|
title={t('lineage.refs.staleHint')}
|
||||||
|
>
|
||||||
|
{t('lineage.refs.stale', { minutes: stale })}
|
||||||
|
</span>
|
||||||
|
)}
|
||||||
|
<span className="text-2xs text-carbon/60">
|
||||||
|
{t('lineage.refs.total', { count: data.total })}
|
||||||
|
</span>
|
||||||
|
</div>
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
{/* Per-source summary chips */}
|
{/* Per-source summary chips */}
|
||||||
|
|||||||
@@ -249,6 +249,8 @@ i18n
|
|||||||
'lineage.refs.field': 'поле {{field}}',
|
'lineage.refs.field': 'поле {{field}}',
|
||||||
'lineage.refs.created': 'создана {{when}}',
|
'lineage.refs.created': 'создана {{when}}',
|
||||||
'lineage.refs.openSourceRecord': 'Открыть запись {{key}} в источнике',
|
'lineage.refs.openSourceRecord': 'Открыть запись {{key}} в источнике',
|
||||||
|
'lineage.refs.stale': 'обновлено {{minutes}} мин назад',
|
||||||
|
'lineage.refs.staleHint': 'Данные из materialized view; обновляются раз в минуту. Закрытые между обновлениями записи отфильтрованы по valid_to.',
|
||||||
'lineage.onClose.BLOCK': 'block',
|
'lineage.onClose.BLOCK': 'block',
|
||||||
'lineage.onClose.WARN': 'warn',
|
'lineage.onClose.WARN': 'warn',
|
||||||
'lineage.onClose.CASCADE': 'cascade',
|
'lineage.onClose.CASCADE': 'cascade',
|
||||||
@@ -552,6 +554,8 @@ i18n
|
|||||||
'lineage.refs.field': 'field {{field}}',
|
'lineage.refs.field': 'field {{field}}',
|
||||||
'lineage.refs.created': 'created {{when}}',
|
'lineage.refs.created': 'created {{when}}',
|
||||||
'lineage.refs.openSourceRecord': 'Open referencing record {{key}}',
|
'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.BLOCK': 'block',
|
||||||
'lineage.onClose.WARN': 'warn',
|
'lineage.onClose.WARN': 'warn',
|
||||||
'lineage.onClose.CASCADE': 'cascade',
|
'lineage.onClose.CASCADE': 'cascade',
|
||||||
|
|||||||
+110
@@ -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).
|
||||||
|
*
|
||||||
|
* <p>Производительность: один SQL по composite index
|
||||||
|
* {@code (target_dict_name, target_business_key_value)} вместо N JSONB
|
||||||
|
* lookups (по одному на каждый source dict).
|
||||||
|
*
|
||||||
|
* <p>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.
|
||||||
|
*
|
||||||
|
* <p>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<DependentRow> 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;
|
||||||
|
}
|
||||||
|
}
|
||||||
+160
@@ -0,0 +1,160 @@
|
|||||||
|
<?xml version="1.0" encoding="UTF-8"?>
|
||||||
|
<databaseChangeLog
|
||||||
|
xmlns="http://www.liquibase.org/xml/ns/dbchangelog"
|
||||||
|
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||||
|
xsi:schemaLocation="http://www.liquibase.org/xml/ns/dbchangelog
|
||||||
|
http://www.liquibase.org/xml/ns/dbchangelog/dbchangelog-4.27.xsd">
|
||||||
|
|
||||||
|
<!--
|
||||||
|
dict-relationships-v2 epic, Phase 2: precomputed reverse FK index.
|
||||||
|
|
||||||
|
Materialized view `record_dependents_index` сворачивает все active FK
|
||||||
|
relationships (source_record + source_field → target_dict + target_value)
|
||||||
|
в один таблично-индексированный snapshot. Даёт O(log n) lookup для
|
||||||
|
record-level dependents query — без JSONB предикатов в hot path.
|
||||||
|
|
||||||
|
Включение per request: feature flag `ordinis.lineage.mv.enabled`
|
||||||
|
(default false). Migration катится всегда — flag решает, использует
|
||||||
|
ли service path MV vs прямой JSONB query (Phase 1 implementation).
|
||||||
|
|
||||||
|
Refresh: REFRESH MATERIALIZED VIEW CONCURRENTLY (требует UNIQUE INDEX).
|
||||||
|
Throttled на app-side через `MaterializedViewRefreshService` — default
|
||||||
|
1 refresh / 60 sec, чтобы избежать storm при bulk close (R2 eng review).
|
||||||
|
|
||||||
|
Active record check сделан query-side (на valid_from/valid_to колонках),
|
||||||
|
чтобы записи закрытые между refresh'ами не появились в результатах
|
||||||
|
(stale window — "data may be stale: refreshed N min ago").
|
||||||
|
|
||||||
|
Initial population SLO: <5 min на 5k records (тест plan target). На
|
||||||
|
pure JSONB indexing (jsonb_each + ?-marker) — не ожидаем больше.
|
||||||
|
-->
|
||||||
|
<changeSet id="0016-record-dependents-index-create" author="ordinis">
|
||||||
|
<comment>MV record_dependents_index: precomputed reverse FK</comment>
|
||||||
|
<preConditions onFail="MARK_RAN">
|
||||||
|
<not>
|
||||||
|
<sqlCheck expectedResult="1">
|
||||||
|
SELECT COUNT(*) FROM pg_matviews WHERE matviewname = 'record_dependents_index'
|
||||||
|
</sqlCheck>
|
||||||
|
</not>
|
||||||
|
</preConditions>
|
||||||
|
<sql splitStatements="false"><![CDATA[
|
||||||
|
CREATE MATERIALIZED VIEW record_dependents_index AS
|
||||||
|
SELECT
|
||||||
|
r.id AS source_record_id,
|
||||||
|
r.dictionary_id AS source_dict_id,
|
||||||
|
src_def.name AS source_dict_name,
|
||||||
|
src_def.display_name AS source_display_name,
|
||||||
|
src_def.scope AS source_scope,
|
||||||
|
r.business_key AS source_business_key,
|
||||||
|
r.data_scope AS source_data_scope,
|
||||||
|
r.valid_from,
|
||||||
|
r.valid_to,
|
||||||
|
r.created_at,
|
||||||
|
prop.key AS source_field,
|
||||||
|
split_part(prop.value->>'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;
|
||||||
|
]]></sql>
|
||||||
|
<rollback>
|
||||||
|
<sql>DROP MATERIALIZED VIEW IF EXISTS record_dependents_index;</sql>
|
||||||
|
</rollback>
|
||||||
|
</changeSet>
|
||||||
|
|
||||||
|
<!--
|
||||||
|
UNIQUE INDEX (source_record_id, source_field) — REQUIRED для REFRESH
|
||||||
|
CONCURRENTLY. Composite потому что одна запись может иметь несколько
|
||||||
|
FK fields в schema (spacecraft.satTypeCode + spacecraft.operatorId).
|
||||||
|
-->
|
||||||
|
<changeSet id="0016-record-dependents-index-unique" author="ordinis">
|
||||||
|
<comment>UNIQUE INDEX для REFRESH CONCURRENTLY</comment>
|
||||||
|
<sql>
|
||||||
|
CREATE UNIQUE INDEX uniq_record_dependents_index_source
|
||||||
|
ON record_dependents_index (source_record_id, source_field);
|
||||||
|
</sql>
|
||||||
|
<rollback>
|
||||||
|
<sql>DROP INDEX IF EXISTS uniq_record_dependents_index_source;</sql>
|
||||||
|
</rollback>
|
||||||
|
</changeSet>
|
||||||
|
|
||||||
|
<!--
|
||||||
|
Main lookup index: (target_dict_name, target_business_key_value).
|
||||||
|
Includes valid_from/valid_to для query-side active filter без
|
||||||
|
дополнительного table scan.
|
||||||
|
-->
|
||||||
|
<changeSet id="0016-record-dependents-index-target-lookup" author="ordinis">
|
||||||
|
<comment>Composite index для record-level dependents lookup</comment>
|
||||||
|
<sql>
|
||||||
|
CREATE INDEX idx_record_dependents_target
|
||||||
|
ON record_dependents_index
|
||||||
|
(target_dict_name, target_business_key_value, valid_from, valid_to);
|
||||||
|
</sql>
|
||||||
|
<rollback>
|
||||||
|
<sql>DROP INDEX IF EXISTS idx_record_dependents_target;</sql>
|
||||||
|
</rollback>
|
||||||
|
</changeSet>
|
||||||
|
|
||||||
|
<!--
|
||||||
|
Schema-level lookup: (target_dict_name) для "какие словари ссылаются".
|
||||||
|
Не нужен composite — только префикс target_dict_name + DISTINCT
|
||||||
|
source_dict_name fingerprint берётся в SQL.
|
||||||
|
-->
|
||||||
|
<changeSet id="0016-record-dependents-index-schema-lookup" author="ordinis">
|
||||||
|
<comment>Index для schema-level reverse FK list</comment>
|
||||||
|
<sql>
|
||||||
|
CREATE INDEX idx_record_dependents_schema
|
||||||
|
ON record_dependents_index (target_dict_name);
|
||||||
|
</sql>
|
||||||
|
<rollback>
|
||||||
|
<sql>DROP INDEX IF EXISTS idx_record_dependents_schema;</sql>
|
||||||
|
</rollback>
|
||||||
|
</changeSet>
|
||||||
|
|
||||||
|
<!--
|
||||||
|
Initial population. Не блокирует existing reads/writes (REFRESH без
|
||||||
|
CONCURRENTLY на пустой view — fast path). На 5k records ожидание <5 min.
|
||||||
|
-->
|
||||||
|
<changeSet id="0016-record-dependents-index-populate" author="ordinis">
|
||||||
|
<comment>Initial REFRESH (no CONCURRENTLY на первый populate)</comment>
|
||||||
|
<sql>REFRESH MATERIALIZED VIEW record_dependents_index;</sql>
|
||||||
|
<rollback>
|
||||||
|
<!-- No-op: rollback drops view целиком. -->
|
||||||
|
</rollback>
|
||||||
|
</changeSet>
|
||||||
|
|
||||||
|
<!--
|
||||||
|
Refresh metadata table — single-row для tracking last refresh timestamp,
|
||||||
|
duration, row count. Используется service для exposure через API
|
||||||
|
(stale window UX), и admin endpoint для health check.
|
||||||
|
-->
|
||||||
|
<changeSet id="0016-record-dependents-refresh-meta" author="ordinis">
|
||||||
|
<comment>Metadata табличка для last MV refresh tracking</comment>
|
||||||
|
<createTable tableName="lineage_mv_meta">
|
||||||
|
<column name="id" type="SMALLINT" defaultValueNumeric="1">
|
||||||
|
<constraints primaryKey="true" nullable="false"/>
|
||||||
|
</column>
|
||||||
|
<column name="last_refreshed_at" type="TIMESTAMPTZ"/>
|
||||||
|
<column name="last_duration_ms" type="BIGINT"/>
|
||||||
|
<column name="last_row_count" type="BIGINT"/>
|
||||||
|
<column name="last_status" type="VARCHAR(32)" defaultValue="never"/>
|
||||||
|
<column name="last_error" type="TEXT"/>
|
||||||
|
</createTable>
|
||||||
|
<sql>
|
||||||
|
ALTER TABLE lineage_mv_meta
|
||||||
|
ADD CONSTRAINT lineage_mv_meta_singleton CHECK (id = 1);
|
||||||
|
</sql>
|
||||||
|
<insert tableName="lineage_mv_meta">
|
||||||
|
<column name="id" valueNumeric="1"/>
|
||||||
|
<column name="last_status" value="never"/>
|
||||||
|
</insert>
|
||||||
|
</changeSet>
|
||||||
|
|
||||||
|
</databaseChangeLog>
|
||||||
@@ -25,5 +25,6 @@
|
|||||||
<include file="changes/0013-webhook-subscriptions.xml" relativeToChangelogFile="true"/>
|
<include file="changes/0013-webhook-subscriptions.xml" relativeToChangelogFile="true"/>
|
||||||
<include file="changes/0014-webhook-deliveries.xml" relativeToChangelogFile="true"/>
|
<include file="changes/0014-webhook-deliveries.xml" relativeToChangelogFile="true"/>
|
||||||
<include file="changes/0015-backfill-latlon-geometry.xml" relativeToChangelogFile="true"/>
|
<include file="changes/0015-backfill-latlon-geometry.xml" relativeToChangelogFile="true"/>
|
||||||
|
<include file="changes/0016-record-dependents-index.xml" relativeToChangelogFile="true"/>
|
||||||
|
|
||||||
</databaseChangeLog>
|
</databaseChangeLog>
|
||||||
|
|||||||
+55
-7
@@ -9,6 +9,7 @@ import cloud.nstart.terravault.ordinis.domain.record.DictionaryRecordRepository;
|
|||||||
import com.fasterxml.jackson.databind.JsonNode;
|
import com.fasterxml.jackson.databind.JsonNode;
|
||||||
import org.slf4j.Logger;
|
import org.slf4j.Logger;
|
||||||
import org.slf4j.LoggerFactory;
|
import org.slf4j.LoggerFactory;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
import org.springframework.stereotype.Service;
|
import org.springframework.stereotype.Service;
|
||||||
import org.springframework.transaction.annotation.Transactional;
|
import org.springframework.transaction.annotation.Transactional;
|
||||||
|
|
||||||
@@ -53,14 +54,27 @@ public class LineageIndexService {
|
|||||||
private final DictionaryDefinitionRepository definitionRepository;
|
private final DictionaryDefinitionRepository definitionRepository;
|
||||||
private final DictionaryRecordRepository recordRepository;
|
private final DictionaryRecordRepository recordRepository;
|
||||||
private final DependentsQuery dependentsQuery;
|
private final DependentsQuery dependentsQuery;
|
||||||
|
/** Optional — присутствует только когда {@code ordinis.lineage.mv.enabled=true}. */
|
||||||
|
private final Optional<MaterializedViewRefreshService> mvRefresh;
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
public LineageIndexService(
|
||||||
|
DictionaryDefinitionRepository definitionRepository,
|
||||||
|
DictionaryRecordRepository recordRepository,
|
||||||
|
DependentsQuery dependentsQuery,
|
||||||
|
Optional<MaterializedViewRefreshService> mvRefresh) {
|
||||||
|
this.definitionRepository = definitionRepository;
|
||||||
|
this.recordRepository = recordRepository;
|
||||||
|
this.dependentsQuery = dependentsQuery;
|
||||||
|
this.mvRefresh = mvRefresh;
|
||||||
|
}
|
||||||
|
|
||||||
|
/** Test-friendly ctor без MV refresh service. */
|
||||||
public LineageIndexService(
|
public LineageIndexService(
|
||||||
DictionaryDefinitionRepository definitionRepository,
|
DictionaryDefinitionRepository definitionRepository,
|
||||||
DictionaryRecordRepository recordRepository,
|
DictionaryRecordRepository recordRepository,
|
||||||
DependentsQuery dependentsQuery) {
|
DependentsQuery dependentsQuery) {
|
||||||
this.definitionRepository = definitionRepository;
|
this(definitionRepository, recordRepository, dependentsQuery, Optional.empty());
|
||||||
this.recordRepository = recordRepository;
|
|
||||||
this.dependentsQuery = dependentsQuery;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -182,7 +196,8 @@ public class LineageIndexService {
|
|||||||
List<SchemaDependent> schemaRefs = findSchemaDependents(targetDictName, allowedScopes);
|
List<SchemaDependent> schemaRefs = findSchemaDependents(targetDictName, allowedScopes);
|
||||||
if (schemaRefs.isEmpty()) {
|
if (schemaRefs.isEmpty()) {
|
||||||
return new RecordDependentsPage(
|
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.
|
// Per-source: pull target value, count + paginate.
|
||||||
@@ -243,7 +258,26 @@ public class LineageIndexService {
|
|||||||
List<RecordDependent> slice = all.subList(from, to);
|
List<RecordDependent> slice = all.subList(from, to);
|
||||||
|
|
||||||
return new RecordDependentsPage(
|
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 ===
|
// === DTOs ===
|
||||||
@@ -277,7 +311,7 @@ public class LineageIndexService {
|
|||||||
OnCloseAction onClose,
|
OnCloseAction onClose,
|
||||||
long count) {}
|
long count) {}
|
||||||
|
|
||||||
/** Paginated page + per-source aggregates. */
|
/** Paginated page + per-source aggregates + lineage source metadata. */
|
||||||
public record RecordDependentsPage(
|
public record RecordDependentsPage(
|
||||||
String targetDict,
|
String targetDict,
|
||||||
String targetBusinessKey,
|
String targetBusinessKey,
|
||||||
@@ -285,5 +319,19 @@ public class LineageIndexService {
|
|||||||
List<PerSourceSummary> perSource,
|
List<PerSourceSummary> perSource,
|
||||||
long total,
|
long total,
|
||||||
int page,
|
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) {}
|
||||||
}
|
}
|
||||||
|
|||||||
+260
@@ -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).
|
||||||
|
*
|
||||||
|
* <p>Активируется когда {@code ordinis.lineage.mv.enabled=true}. Без MV
|
||||||
|
* сервис не нужен — {@link DependentsQueryJdbc} даёт прямой JSONB query
|
||||||
|
* без stale window'а.
|
||||||
|
*
|
||||||
|
* <p>Пути refresh'а:
|
||||||
|
* <ul>
|
||||||
|
* <li><b>Scheduled</b> — каждые
|
||||||
|
* {@code ordinis.lineage.mv.refresh-interval-sec} (default 60).
|
||||||
|
* Простейший throttle: если refresh уже запущен — пропускаем тик.</li>
|
||||||
|
* <li><b>Force</b> — {@link #refreshNow()} вызывается из bundle import
|
||||||
|
* или admin endpoint. Также throttled: refused если последний refresh
|
||||||
|
* был меньше {@code min-interval-sec} назад.</li>
|
||||||
|
* </ul>
|
||||||
|
*
|
||||||
|
* <p>REFRESH CONCURRENTLY — read concurrent с queries, требует UNIQUE INDEX
|
||||||
|
* на MV (создан в migration 0016). Если MV пуста (initial state до первого
|
||||||
|
* REFRESH) — CONCURRENTLY падает; fallback на non-CONCURRENTLY один раз.
|
||||||
|
*
|
||||||
|
* <p>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<Instant> 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<RefreshMeta> 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<RefreshMeta> 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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
+185
@@ -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 падает.
|
||||||
|
*
|
||||||
|
* <p>Использует 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();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user