diff --git a/Cargo.lock b/Cargo.lock index e8744f65..d85df38d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4636,21 +4636,6 @@ version = "0.57.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" -[[package]] -name = "write-amplification-probe" -version = "0.1.0" -dependencies = [ - "aws-config", - "aws-sdk-cloudwatch", - "clickhouse", - "lambda_runtime 0.13.0", - "prices-clickhouse", - "serde", - "serde_json", - "tokio", - "tracing", -] - [[package]] name = "writeable" version = "0.6.3" diff --git a/Cargo.toml b/Cargo.toml index f2b8f73b..292b3279 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,7 +19,6 @@ members = [ "packages/prices-api", "packages/backfill-freshness-probe", "packages/mtls-notafter-probe", - "packages/write-amplification-probe", "packages/pool-registry-seed", ] diff --git a/infra/envs/production.json b/infra/envs/production.json index 4feaba5a..b7123c8c 100644 --- a/infra/envs/production.json +++ b/infra/envs/production.json @@ -18,14 +18,12 @@ "cleanup": "cron(0 3 * * ? *)", "enrichment": "rate(1 hour)", "backfillFreshnessProbe": "rate(15 minutes)", - "mtlsNotafterProbe": "rate(1 day)", - "writeAmplificationProbe": "rate(1 hour)" + "mtlsNotafterProbe": "rate(1 day)" }, "opsAlarms": { "sdexPushFreshnessSeconds": 604800, "mtlsNotAfterDaysThreshold": 30, "ledgerProcessorLagSeconds": 120, - "writeAmplificationRowsPerHour": 50000000, "slack": { "workspaceIdSsmParam": "/prices/production/slack-workspace-id", "channelIdSsmParam": "/prices/production/slack-channel-id" diff --git a/infra/src/lib/stacks/eventbridge-stack.ts b/infra/src/lib/stacks/eventbridge-stack.ts index 8e122780..3eeffd85 100644 --- a/infra/src/lib/stacks/eventbridge-stack.ts +++ b/infra/src/lib/stacks/eventbridge-stack.ts @@ -57,11 +57,6 @@ const MTLS_NOTAFTER_PROBE_ASSET_DIR = process.env['MTLS_NOTAFTER_PROBE_ASSET_DIR'] ?? '../target/lambda/mtls-notafter-probe'; -/** Cargo-lambda build output for the `write-amplification-probe` (task 0133). */ -const WRITE_AMPLIFICATION_PROBE_ASSET_DIR = - process.env['WRITE_AMPLIFICATION_PROBE_ASSET_DIR'] ?? - '../target/lambda/write-amplification-probe'; - export interface EventBridgeStackProps extends cdk.StackProps { readonly config: EnvironmentConfig; } @@ -89,7 +84,6 @@ export class EventBridgeStack extends cdk.Stack { public readonly enrichmentRule: events.Rule; public readonly backfillFreshnessProbeRule: events.Rule; public readonly mtlsNotafterProbeRule: events.Rule; - public readonly writeAmplificationProbeRule: events.Rule; public readonly assetDiscoveryFunction: lambda.Function; public readonly cleanupFunction: lambda.Function; public readonly supplyFunction: lambda.Function; @@ -97,7 +91,6 @@ export class EventBridgeStack extends cdk.Stack { public readonly enrichmentFunction: lambda.Function; public readonly backfillFreshnessProbeFunction: lambda.Function; public readonly mtlsNotafterProbeFunction: lambda.Function; - public readonly writeAmplificationProbeFunction: lambda.Function; constructor(scope: Construct, id: string, props: EventBridgeStackProps) { super(scope, id, props); @@ -176,16 +169,6 @@ export class EventBridgeStack extends cdk.Stack { }, ); - this.writeAmplificationProbeRule = new events.Rule( - this, - 'WriteAmplificationProbeRule', - { - ruleName: `prices-${env}-write-amplification-probe`, - description: `Publishes max rows-written/hour per prices table → Prices/Ingest MaxRowsWrittenPerHour (${env})`, - schedule: events.Schedule.expression(schedules.writeAmplificationProbe), - }, - ); - // ----------------------------------------------------------------- // Asset Discovery worker Lambda (task 0054) + its rule target. // No VPC (ADR 0007 §6); mTLS to ClickHouse + S3 read on BE's ledger @@ -577,56 +560,12 @@ export class EventBridgeStack extends cdk.Stack { }), ); - // ----------------------------------------------------------------- - // Write-amplification probe (task 0133) + its rate(1h) target. CH-only - // (no S3, no VPC): reads rows-written-per-hour per prices table from - // system.part_log and republishes the max as Prices/Ingest - // MaxRowsWrittenPerHour — the guardrail that would have caught the 0132 - // egress bug (9,413× re-emit) in minutes instead of weeks. Reads as the - // `api` (prices_reader) identity, which is granted SELECT on system.part_log - // (task 0133 prerequisite) alongside its prices.* read — NOT the ingestion - // identity, keeping the probe read-only. - // ----------------------------------------------------------------- - const writeAmplification = createWorkerLambda(this, { - config, - accountId, - mtlsSecretName: apiMtlsSecretName, - idPrefix: 'WriteAmplificationProbe', - name: 'write-amplification-probe', - assetDir: WRITE_AMPLIFICATION_PROBE_ASSET_DIR, - memorySize: 256, - // One aggregate SELECT over system.part_log + one PutMetricData; fast. - timeout: cdk.Duration.minutes(1), - secretsExtensionLayer, - chDomain, - rule: this.writeAmplificationProbeRule, - alarmDescription: - 'Write-amplification probe invocation errors — the rows-written metric may be stale, blinding the write-amplification alarm.', - alarmPeriod: cdk.Duration.hours(1), - errorAlarmActions: [opsAlarmAction], - }); - this.writeAmplificationProbeFunction = writeAmplification.function; - - writeAmplification.role.addToPolicy( - new iam.PolicyStatement({ - sid: 'PublishIngestMetrics', - actions: ['cloudwatch:PutMetricData'], - resources: ['*'], - conditions: { - StringEquals: { 'cloudwatch:namespace': 'Prices/Ingest' }, - }, - }), - ); - new cdk.CfnOutput(this, 'BackfillFreshnessProbeFunctionName', { value: this.backfillFreshnessProbeFunction.functionName, }); new cdk.CfnOutput(this, 'MtlsNotafterProbeFunctionName', { value: this.mtlsNotafterProbeFunction.functionName, }); - new cdk.CfnOutput(this, 'WriteAmplificationProbeFunctionName', { - value: this.writeAmplificationProbeFunction.functionName, - }); cdk.Tags.of(this).add('Project', 'stellar-prices-api'); cdk.Tags.of(this).add('ManagedBy', 'cdk'); diff --git a/infra/src/lib/stacks/observability-stack.ts b/infra/src/lib/stacks/observability-stack.ts index 8555042a..e1a172af 100644 --- a/infra/src/lib/stacks/observability-stack.ts +++ b/infra/src/lib/stacks/observability-stack.ts @@ -231,9 +231,6 @@ export class ObservabilityStack extends cdk.Stack { public readonly sdexPushFreshnessAlarm: cloudwatch.Alarm; /** mTLS client-cert expiry alarm (§7 / §11.4). */ public readonly mtlsNotAfterAlarm: cloudwatch.Alarm; - - /** Write-amplification guardrail (task 0133). */ - public readonly writeAmplificationAlarm: cloudwatch.Alarm; /** Live ledger-processor ingestion-lag alarm (task 0056 finding B). */ public readonly ledgerProcessorLagAlarm: cloudwatch.Alarm; /** Live ledger-processor invocation-error alarm (task 0056 finding B). */ @@ -485,61 +482,6 @@ export class ObservabilityStack extends cdk.Stack { this.mtlsNotAfterAlarm.addAlarmAction(snsAction); this.mtlsNotAfterAlarm.addOkAction(snsAction); - // Write amplification (task 0133 — the guardrail for the 0132 egress bug). - // The write-amplification-probe publishes the max rows-written-per-hour - // across all prices tables as Prices/Ingest MaxRowsWrittenPerHour; alarm - // when it stays above the operator-tuned threshold. A quiet hour publishes a - // real 0 (healthy), so missing data is non-breaching — probe-down is covered - // by the probe's own error alarm. - // - // ⚠️ system.part_log counts ALL writes to prices.* — including legitimate - // BULK loads (the 0088 backfill, coarse pre-rolls, enrichment bursts), which - // an absolute row count cannot distinguish from a re-emit amplification by - // *magnitude*: a 14-day part_log measurement (task 0133) found a legit - // one-hour `_bak` copy at 154M rows/hour — HIGHER than the 0132 bug's ~130M. - // What separates them is DURATION: legit bulk is bursty (the `_bak` was one - // hour; the sustained legit peak is price_ohlcv_1m at ~16M/hour on a backfill - // day), while a 0132-class runaway persists for days. Hence: (1) the alarm - // requires a *sustained* 3-hour breach (datapointsToAlarm=3), which clears the - // one-hour spikes; (2) the threshold (default 50M, measured to sit ~3× above - // the ~16M sustained legit peak and ~2.6× below the ~130M bug) is set so no - // legit event in the measured window breaches it for 3 sustained hours. An - // operator running a known heavy migration should still expect a possible fire - // and ack it or raise config.opsAlarms.writeAmplificationRowsPerHour for the - // window. A true "written vs deduplicated real rows" ratio (which a legit bulk - // load keeps ~1× while a re-emit pushes high) is the robust future enhancement. - // - // The 1h metric period is coupled to the probe's SQL window (INTERVAL 1 HOUR) - // and its rate(1 hour) schedule — see WINDOW_HOURS in the probe crate. All - // three must change together. - this.writeAmplificationAlarm = new cloudwatch.Alarm( - this, - 'WriteAmplificationAlarm', - { - alarmName: `prices-${config.envName}-write-amplification`, - alarmDescription: - 'A prices table has been written far above any legitimate steady-state rate for 3 consecutive hours (rows-written/hour above config.opsAlarms.writeAmplificationRowsPerHour). Likely a write-amplification regression like task 0132 (full-registry re-emit) — but a sustained heavy backfill/pre-roll can also trip it. Check system.part_log per table to find the offender; if it is a known bulk load, ack or raise the threshold for that window.', - metric: new cloudwatch.Metric({ - namespace: 'Prices/Ingest', - metricName: 'MaxRowsWrittenPerHour', - dimensionsMap: { Environment: config.envName }, - statistic: 'Maximum', - period: cdk.Duration.hours(1), - }), - threshold: config.opsAlarms.writeAmplificationRowsPerHour, - // Sustained 3-hour breach, not a single anomalous hour (task 0133 - // review): the guardrail targets a persistent runaway, not a one-off - // legit burst. - evaluationPeriods: 3, - datapointsToAlarm: 3, - comparisonOperator: - cloudwatch.ComparisonOperator.GREATER_THAN_THRESHOLD, - treatMissingData: cloudwatch.TreatMissingData.NOT_BREACHING, - }, - ); - this.writeAmplificationAlarm.addAlarmAction(snsAction); - this.writeAmplificationAlarm.addOkAction(snsAction); - // ----------------------------------------------------------------- // Live ledger-processor health (task 0056 finding B). The core ingestion // Lambda shipped unmonitored: `prices.ledger_processor.lag_seconds` existed diff --git a/infra/src/lib/types.ts b/infra/src/lib/types.ts index d6418438..20f4aedd 100644 --- a/infra/src/lib/types.ts +++ b/infra/src/lib/types.ts @@ -113,20 +113,6 @@ export interface EnvironmentConfig { * Daily is ample for a 30-day threshold. */ readonly mtlsNotafterProbe: string; - /** - * Write-amplification probe (task 0133). Reads rows-written-per-hour per - * `prices.*` table from `system.part_log` (as `prices_reader`, which is - * granted `SELECT ON system.part_log`) and republishes the max as the - * `Prices/Ingest` `MaxRowsWrittenPerHour` metric the write-amplification - * alarm watches. Hourly is ample — the guardrail catches a sustained - * runaway (0132 bled for weeks), not a sub-hour spike. - * - * ⚠️ Must stay `rate(1 hour)`: it is coupled to the probe's trailing SQL - * window (`INTERVAL 1 HOUR`, `WINDOW_HOURS`) and the alarm's 1h metric - * period. Changing the cadence without changing the SQL window makes - * consecutive runs overlap (double-count) or gap the window. - */ - readonly writeAmplificationProbe: string; }; // Ops alarms + notification (consumed by ObservabilityStack — task 0056) @@ -151,20 +137,6 @@ export interface EnvironmentConfig { readonly sdexPushFreshnessSeconds: number; /** Days-to-NotAfter below which the mTLS cert-expiry alarm fires (30). */ readonly mtlsNotAfterDaysThreshold: number; - /** - * Rows-written-per-hour to any single `prices.*` table above which the - * write-amplification alarm fires — for the required 3-hour sustained window - * (task 0133; see the alarm in observability-stack.ts). Default 50,000,000, - * set from a 14-day `system.part_log` measurement, NOT a guess: the highest - * *sustained* legitimate load is `price_ohlcv_1m` at ~16M rows/hour during a - * backfill/reprice day (07-26, ~18h). One-hour bulk spikes go higher - * (rollup-rework `_bak` copies hit 154M/hour) but clear within the 3-hour - * window. Task 0132 ran ~130M/hour for days. 50M sits ~3× above the sustained - * legit peak (headroom for backfill growth) and ~2.6× below a 0132-class - * runaway, and no legit event in the measured window breaches it for 3 - * sustained hours. Operator-tunable; raise it during a known heavy migration. - */ - readonly writeAmplificationRowsPerHour: number; /** * Ingestion-lag threshold (seconds) for the live ledger-processor alarm * (task 0056 finding B). Watches the `prices-ingest-{env}` SQS queue's @@ -358,7 +330,6 @@ export function validateConfig(config: EnvironmentConfig): void { 'enrichment', 'backfillFreshnessProbe', 'mtlsNotafterProbe', - 'writeAmplificationProbe', ] as const; for (const key of expectedKeys) { const value = schedules[key]; @@ -400,14 +371,6 @@ export function validateConfig(config: EnvironmentConfig): void { `opsAlarms.ledgerProcessorLagSeconds must be a positive integer (seconds), got: ${ops.ledgerProcessorLagSeconds}`, ); } - if ( - !Number.isInteger(ops.writeAmplificationRowsPerHour) || - ops.writeAmplificationRowsPerHour < 1 - ) { - errors.push( - `opsAlarms.writeAmplificationRowsPerHour must be a positive integer (rows/hour), got: ${ops.writeAmplificationRowsPerHour}`, - ); - } if (ops.slack !== undefined) { const isSsmName = (v: unknown): boolean => typeof v === 'string' && v.startsWith('/') && v.length > 1; diff --git a/lore/1-tasks/active/0133_FEATURE_live-pipeline-egress-write-volume-alarm.md b/lore/1-tasks/archive/0133_FEATURE_live-pipeline-egress-write-volume-alarm.md similarity index 62% rename from lore/1-tasks/active/0133_FEATURE_live-pipeline-egress-write-volume-alarm.md rename to lore/1-tasks/archive/0133_FEATURE_live-pipeline-egress-write-volume-alarm.md index f36ee710..2a574142 100644 --- a/lore/1-tasks/active/0133_FEATURE_live-pipeline-egress-write-volume-alarm.md +++ b/lore/1-tasks/archive/0133_FEATURE_live-pipeline-egress-write-volume-alarm.md @@ -2,7 +2,7 @@ id: "0133" title: "Guardrail: egress / write-volume alarm on the live pipeline so amplification shows on a dashboard, not a bill" type: FEATURE -status: active +status: completed related_adr: [] related_tasks: ["0132", "0039", "0056"] tags: [observability, cost, clickhouse, egress, perf, priority-medium, effort-small, phase-future] @@ -25,10 +25,48 @@ history: (writes 21.7M/10min → 0). The guardrail is the direct lesson of 0132: the amplification ran undetected for weeks. Prioritised so the next one hits a dashboard, not a bill. + - date: 2026-07-29 + status: completed + who: okarcz + note: > + Completed via a different solution after the BE response. Our own + write-amplification-probe (PR #156 — new Lambda + Prices/Ingest metric + + alarm) was built, reviewed, threshold-tuned from a 14-day part_log + measurement, and merged — but the deploy hit a hard blocker: the prices CH + users are XML-managed in BE's users_xml (readonly to SQL), so the required + `SELECT ON system.part_log` grant for prices_reader could not be applied + (Code 495 ACCESS_STORAGE_READONLY) and would need a BE services.xml change. + On raising it, BE opted to cover this at the shared-infra layer instead: a + transfer-cost alarm they own. That satisfies the task goal (a guardrail so + the next amplification hits an alarm, not a bill) without a prices-owned + probe, so **PR #156 was reverted** (unused code) and this task is closed. + Guardrail responsibility now sits with BE's shared infra alarm. --- # Egress / write-volume alarm on the live pipeline +## ✅ Resolution (2026-07-29) — solved by BE shared-infra alarm; our probe reverted + +**Goal met, but not with our code.** We built the prices-owned probe (PR #156: new +`write-amplification-probe` Lambda → `Prices/Ingest MaxRowsWrittenPerHour` metric → +CloudWatch alarm → Slack), reviewed it, and tuned the threshold to **50M/hour over a +3-hour sustained window** from a 14-day `system.part_log` measurement (the measurement +also surfaced that a legit one-hour `_bak` copy hit 154M/hour — *above* the 0132 bug's +130M — so only a sustained window separates legit bulk from a runaway). + +**Blocker that forced the pivot:** the deploy needs `prices_reader` to read +`system.part_log`, but the prices CH users are **XML-managed in BE's `users_xml`** +(readonly to SQL) — a SQL `GRANT` fails with `Code 495 ACCESS_STORAGE_READONLY`, so the +grant would require a change to BE's `soroban-block-explorer/.../users.d/services.xml`. +On raising it with BE, they chose to cover it at the **shared-infra layer instead: a +transfer-cost alarm they own**. That satisfies the task goal (a guardrail so the next +amplification hits an alarm, not a bill) without a prices-owned probe or a CH grant. + +**Outcome:** PR #156 **reverted** (unused code removed from `develop`); guardrail +responsibility now sits with **BE's shared-infra transfer-cost alarm**. The design +below is retained for the record (and if a prices-owned probe is ever wanted, the +`system.part_log`-grant path and the measured threshold are documented here). + ## Summary Task 0132 (live processor re-emitting the whole asset registry every reconcile, @@ -124,54 +162,15 @@ sister `default.assets` ran 4.6×); 0132 ran 9,413×. A threshold of **~50×** c runaway with wide margin above normal churn. Also consider an absolute floor (e.g. `RowsWrittenPerHour` per table) so a low-real-row table can't hide a large absolute write. -## Implementation Notes (built 2026-07-29) - -Mirrors the 0056 probe pattern end-to-end. New crate **`write-amplification-probe`** -(`src/lib.rs` pure logic + tests, `src/main.rs` `lambda`-gated entrypoint), added to the -workspace members. Infra: `eventbridge-stack.ts` (rule + `createWorkerLambda` on the `api`/ -`prices_reader` bundle + `PutMetricData` scoped to `Prices/Ingest`), `observability-stack.ts` -(alarm → existing SNS/Slack), `types.ts` + `envs/production.json` (schedule + threshold). -Verified: 4 unit tests pass, clippy clean, arm64 bootstrap builds, infra `tsc` build passes, -full app `cdk synth` succeeds, and the alarm/probe/IAM land correctly in the synthesized CFN. - -## Issues Encountered (code review, PR #156) - -- **Finding 1 (false-positive on legit bulk loads) — REAL, RESOLVED with measured tuning.** - `system.part_log` counts *all* writes to `prices.*`, so legit bulk (0088 backfill, pre-rolls, - enrichment) is indistinguishable from a re-emit by *magnitude*. A 14-day `part_log` measurement - proved this is not hypothetical: a legit one-hour `price_ohlcv_15m_bak` copy (07-17 rollup rework) - wrote **154M rows/hour — higher than the 0132 bug's ~130M**. So no absolute threshold works on - value alone. **But the discriminator is duration:** the `_bak` spike was **one hour**; the highest - *sustained* legit load is `price_ohlcv_1m` at **~16M/hour** (07-26 backfill day, ~18h); the 0132 - bug ran ~130M/hour for **days**. Resolution: (a) **sustained 3-hour breach** clears the one-hour - spikes; (b) **threshold set to 50M** (measured: ~3× above the 16M sustained legit peak, ~2.6× - below the 130M bug) — no legit event in the 14-day window breaches it for 3 sustained hours, and a - 0132-class runaway still does. Threshold operator-tunable / ack-able during known migrations. The - written-vs-real *ratio* (legit bulk ~1×, re-emit high) remains the robust future enhancement (needs - a deduplicated-count denominator — `count() FINAL` heavy, `system.parts` diluted by un-merged parts). -- **Finding 2 (single-hour trigger) — FIXED.** `evaluationPeriods`/`datapointsToAlarm` 1→3, so a - lone anomalous hour no longer pages; matches the "sustained runaway" intent. -- **Finding 3 (window coupled across 3 files) — FIXED.** Added `WINDOW_HOURS` const + explicit - coupling notes on the SQL query, the schedule config doc, and the alarm period. -- **Finding 4 (unused `thiserror` dep) — FIXED.** Removed (copy-paste from the freshness probe). - ## Design Decisions ### Emerged -1. **General sweep, not an assets-specific watch** — the point of a guardrail is the *next* - unknown regression, not re-watching the one we already fixed. The probe reads every - `prices.*` table's write volume. -2. **`prices_reader`, not a dedicated probe user** — reuse the existing `api` mTLS bundle (no - new cert to provision); the trade-off (the read-API identity gains one metadata read) is - accepted (user decision). -3. **Absolute `MaxRowsWrittenPerHour`, not a written÷real *ratio*** — the ratio would need - `system.parts` too, but `prices_reader` is granted only `system.part_log` (+ `prices.*`). - For a guardrail, absolute rows/hour is sufficient: the busiest legit table is <1M/h and - 0132 was ~130M/h, so a 10M threshold has wide margin both ways. Publishing a single scalar - (max across tables) rather than a per-`Table` metric also keeps CloudWatch dimensions - bounded and the alarm a trivial threshold. A true ratio is a documented future enhancement - that would add a `system.parts` read grant. +1. **Amplification factor, not raw rows** — a table legitimately grows; what's pathological + is writing many multiples of the real row count. The ratio normalises table size and is + what made 0132 obvious (9,413× vs a 60 MiB table). +2. **General per-table sweep, not an assets-specific watch** — the point of a guardrail is + the *next* unknown regression, not re-watching the one we already fixed. ## Open Questions / Risks @@ -185,21 +184,17 @@ full app `cdk synth` succeeds, and the alarm/probe/IAM land correctly in the syn ## Acceptance Criteria - [ ] Prerequisite grant applied on ch-prod-01: `GRANT SELECT ON system.part_log TO prices_reader` - (one-time admin op at **deploy time**; verify the probe can then read `system.part_log`). -- [x] `write-amplification-probe` crate: pure metric-shaping (`max_rows_written`) + query-shaping - unit-tested (4 tests); CH fetch (as `prices_reader` over mTLS) + `PutMetricData` gated behind - `lambda`/`aws-mtls` (mirrors 0056 probes). Clippy clean; arm64 bootstrap builds. -- [x] Publishes `MaxRowsWrittenPerHour` (see Emerged #3 — scalar max, not per-`Table` ratio) - under `Prices/Ingest`; IAM `PutMetricData` scoped by the namespace condition. -- [x] EventBridge rule schedules it `rate(1 hour)`, with `errorAlarmActions` so a dead probe alarms. -- [x] CloudWatch alarm on the metric breaching the threshold → 0056 SNS/Slack; `OkAction` set. - Verified in synthesized CFN (threshold 10,000,000, GreaterThanThreshold, notBreaching). -- [x] Threshold operator-tunable via `config.opsAlarms.writeAmplificationRowsPerHour`; set to - **50M from a 14-day part_log measurement** (sustained legit peak ~16M/h, bug ~130M/h) with a - 3-hour sustained window — validated against real backfill/pre-roll peaks, not a guess. -- [~] Runbook: the alarm *description* embeds "check `system.part_log` per table to find the - offender"; a fuller runbook note (links 0132's part_log day-slice + anti-join) is a small - follow-up. + (one-time admin op; verify the probe can then read `system.part_log` as `prices_reader`). +- [ ] `write-amplification-probe` crate: pure factor-math + query-shaping unit-tested; + CH fetch (as `prices_reader` over mTLS) + `PutMetricData` gated behind `lambda`/`aws-mtls` + (mirrors 0056 probes). +- [ ] Publishes `WriteAmplificationFactor` (per-`Table`) under `Prices/Ingest`; IAM + `PutMetricData` scoped by namespace condition. +- [ ] EventBridge rule schedules it (hourly), with `errorAlarmActions` so a dead probe alarms. +- [ ] CloudWatch alarm on the metric breaching the threshold → 0056 SNS/Slack; `OkAction` set. +- [ ] Threshold operator-tunable via `config`; documented (legit few× vs 0132's 9,413×). +- [ ] Brief runbook note: what a breach means + "what to check first" (links 0132's + `part_log` day-slice + anti-join queries). - [x] `system.part_log` read-access question resolved — Plan A (grant to `prices_reader`); see Prerequisite. Grant *application* tracked as the first AC above. diff --git a/packages/write-amplification-probe/Cargo.toml b/packages/write-amplification-probe/Cargo.toml deleted file mode 100644 index 925fa195..00000000 --- a/packages/write-amplification-probe/Cargo.toml +++ /dev/null @@ -1,40 +0,0 @@ -[package] -name = "write-amplification-probe" -version = "0.1.0" -edition = "2024" -description = "Publishes max rows-written/hour per prices table to CloudWatch for the write-amplification alarm (task 0133)" - -[lib] -name = "write_amplification_probe" -path = "src/lib.rs" - -# EventBridge rate(1 hour) Lambda entrypoint (task 0133). Behind `lambda` so the -# default build/test exercises the pure metric-shaping logic without the AWS -# runtime / mTLS stack. Deployable: -# cargo lambda build -p write-amplification-probe --release --arm64 --features lambda -[[bin]] -name = "write-amplification-probe" -path = "src/main.rs" -required-features = ["lambda"] - -[features] -default = [] -aws-mtls = ["prices-clickhouse/aws-mtls"] -lambda = [ - "aws-mtls", - "dep:lambda_runtime", - "dep:aws-config", - "dep:aws-sdk-cloudwatch", -] - -[dependencies] -prices-clickhouse = { path = "../prices-clickhouse" } -clickhouse = { workspace = true } -serde = { workspace = true } -serde_json = { workspace = true } -tokio = { workspace = true } -tracing = { workspace = true } - -lambda_runtime = { workspace = true, optional = true } -aws-config = { workspace = true, optional = true } -aws-sdk-cloudwatch = { workspace = true, optional = true } diff --git a/packages/write-amplification-probe/src/lib.rs b/packages/write-amplification-probe/src/lib.rs deleted file mode 100644 index 07ac530c..00000000 --- a/packages/write-amplification-probe/src/lib.rs +++ /dev/null @@ -1,210 +0,0 @@ -//! Write-amplification probe (task 0133). -//! -//! The guardrail that would have caught task 0132 in minutes instead of weeks. -//! 0132 was a 9,413× write amplification: the live ledger-processor re-emitted -//! the whole `prices.assets` registry every reconcile, writing ~130M rows/hour -//! to a ~200k-row table and billing ~$337/mo of AWS→Hetzner egress. Nothing -//! watched write volume, so it ran undetected until the BE team found it in -//! `system.part_log`. -//! -//! This scheduled Lambda closes that gap. Once an hour it asks prod ClickHouse -//! how many rows were written to each `prices.*` table in the trailing hour -//! (from `system.part_log`), and republishes the **maximum across all tables** -//! as the custom metric [`METRIC_NAME`] under the [`METRIC_NAMESPACE`] namespace. -//! A CloudWatch alarm fires when that max exceeds an operator-tunable threshold -//! (default well above the busiest legitimate table, far below a 0132-class -//! runaway) → the existing 0056 SNS topic → Slack `#stellar-prices-api-bot`. -//! -//! The window is evaluated **server-side** (`event_time >= now() - INTERVAL 1 -//! HOUR`), so it is immune to clock skew between the Lambda and ClickHouse. -//! -//! ## Why "max rows written", not a written÷real *ratio* -//! -//! Amplification is conceptually `rows_written / real_rows`, but the probe reads -//! as `prices_reader`, which is granted `SELECT` on `system.part_log` (the write -//! log) and `prices.*` — **not** `system.parts` (the real-row snapshot). A -//! per-table ratio would need that second system grant, so v1 alarms on the -//! absolute rows-written-per-hour instead: the busiest legitimate table writes -//! well under 1M rows/hour, while 0132 wrote ~130M/hour, so an absolute -//! threshold with wide margin is a perfectly good guardrail. A true ratio is a -//! documented future enhancement (see the task) that would add a `system.parts` -//! read grant. -//! -//! ## A quiet window is healthy -//! -//! [`WRITE_VOLUME_QUERY`] returns a row only for a table actually written to in -//! the last hour. After the 0132 fix, `assets` writes nothing on an idle hour, -//! so it simply produces no row; [`max_rows_written`] returns `0.0` and the -//! alarm stays OK. That is the correct healthy steady state, not missing data. -//! -//! ## Real tables only -//! -//! The `table NOT LIKE '.%'` filter drops ClickHouse's internal storage tables -//! (`.inner_id.*` materialized-view targets, `.tmp.*` merge scratch) so the -//! metric tracks the real `prices.*` surface and its `Table`-space stays bounded -//! (no UUID-named dimensions). -//! -//! Split for testability: the pure metric-shaping ([`max_rows_written`]) and the -//! query text are compiled in every build and unit-tested without the AWS SDK; -//! the actual CloudWatch publish ([`publish`]) is gated behind the `lambda` -//! feature. - -/// CloudWatch namespace for the write-volume metric. Must match the -/// `cloudwatch:namespace` condition on the Lambda role's `PutMetricData` grant -/// and the alarm wiring in `infra/`. -pub const METRIC_NAMESPACE: &str = "Prices/Ingest"; - -/// Custom metric name: the maximum rows-written-per-hour across all real -/// `prices.*` tables. A single scalar per run (dimensioned only by -/// `Environment`) so the alarm is a trivial threshold and never fans out over a -/// growing set of table dimensions. -pub const METRIC_NAME: &str = "MaxRowsWrittenPerHour"; - -/// The trailing window (hours) summed by [`WRITE_VOLUME_QUERY`]. It is baked into -/// the SQL literal (`INTERVAL 1 HOUR`), so this const is the single documented -/// source of truth for the value and the anchor for the coupling note on the -/// query: the EventBridge schedule and the alarm metric period must both equal -/// this. If it ever needs to change, update the SQL, the schedule -/// (`scheduleExpressions.writeAmplificationProbe`), and the alarm period together. -pub const WINDOW_HOURS: u32 = 1; - -/// One table's rows-written total over the trailing window, as read from -/// `system.part_log`. `sum(rows)` over `NewPart` events is a `UInt64`, so this -/// deserializes into a plain `u64`; a table not written to in the window simply -/// produces no row (see [`WRITE_VOLUME_QUERY`]). -#[derive(Debug, Clone, PartialEq, Eq, clickhouse::Row, serde::Deserialize)] -pub struct TableWrite { - pub table: String, - pub rows_written: u64, -} - -/// The metric value for one run: the maximum rows-written across all reported -/// tables. `0.0` when the window was quiet (no rows returned) — the healthy -/// post-0132 steady state, not missing data. -pub fn max_rows_written(rows: &[TableWrite]) -> f64 { - rows.iter().map(|r| r.rows_written).max().unwrap_or(0) as f64 -} - -/// SQL that sums rows written to each real `prices.*` table over the trailing -/// hour from `system.part_log`. -/// -/// - **`event_type = 'NewPart'`** — count only part *creations* (the inserts / -/// merged outputs that actually cross the wire and land as writes), not -/// merges/mutations/removals. -/// - **`event_time >= now() - INTERVAL 1 HOUR`** — evaluated server-side, so the -/// window is immune to Lambda↔CH clock skew (mirrors the freshness probe). -/// - **`table NOT LIKE '.%'`** — drop ClickHouse-internal tables (`.inner_id.*` -/// MV storage, `.tmp.*` merge scratch) so the metric tracks the real surface -/// and the `Table` space stays bounded. -/// -/// Ordered by volume so the caller can log the top writers for forensics when -/// the alarm fires. -/// -/// ⚠️ **This trailing-hour window is coupled to two infra values and they must -/// stay in lockstep — there is no compile-time link:** -/// 1. the EventBridge schedule `scheduleExpressions.writeAmplificationProbe` -/// (must be `rate(1 hour)`), so runs neither overlap (double-count) nor gap -/// the window, and -/// 2. the alarm metric `period` in `observability-stack.ts` (must be 1 hour), -/// so the threshold is compared against a whole window's writes. -/// -/// Changing the window here **requires** changing both. See [`WINDOW_HOURS`]. -pub const WRITE_VOLUME_QUERY: &str = "SELECT \ - table, \ - sum(rows) AS rows_written \ - FROM system.part_log \ - WHERE database = 'prices' \ - AND event_type = 'NewPart' \ - AND event_time >= now() - INTERVAL 1 HOUR \ - AND table NOT LIKE '.%' \ - GROUP BY table \ - ORDER BY rows_written DESC"; - -/// Publish the max rows-written value to CloudWatch under [`METRIC_NAMESPACE`] -/// as [`METRIC_NAME`], tagged with an `Environment` dimension. One -/// `PutMetricData` call. Always publishes (including `0.0`) so the alarm has a -/// fresh datum every run and a quiet hour reads as a real zero, not missing -/// data. -#[cfg(feature = "lambda")] -pub async fn publish( - client: &aws_sdk_cloudwatch::Client, - environment: &str, - max_rows: f64, -) -> Result<(), aws_sdk_cloudwatch::Error> { - use aws_sdk_cloudwatch::types::{Dimension, MetricDatum, StandardUnit}; - - let datum = MetricDatum::builder() - .metric_name(METRIC_NAME) - .value(max_rows) - .unit(StandardUnit::Count) - .dimensions( - Dimension::builder() - .name("Environment") - .value(environment) - .build(), - ) - .build(); - - client - .put_metric_data() - .namespace(METRIC_NAMESPACE) - .metric_data(datum) - .send() - .await?; - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::*; - - fn w(table: &str, rows: u64) -> TableWrite { - TableWrite { - table: table.to_string(), - rows_written: rows, - } - } - - #[test] - fn max_picks_the_largest_writer() { - let rows = vec![ - w("price_ohlcv_15m", 715_000), - w("assets", 130_000_000), // a 0132-class runaway - w("price_ohlcv_1m", 140_000), - ]; - assert_eq!(max_rows_written(&rows), 130_000_000.0); - } - - #[test] - fn quiet_window_is_zero_not_missing() { - // Post-0132 healthy steady state: nothing written → no rows → 0.0, so the - // alarm reads a real zero rather than treating-missing-data. - assert_eq!(max_rows_written(&[]), 0.0); - } - - #[test] - fn legitimate_traffic_stays_well_under_a_runaway() { - // The busiest legit table (the 15m rollup) is ~0.7M/hour — orders of - // magnitude below a 0132-class ~130M/hour, so any reasonable threshold - // between them separates healthy from pathological. - let rows = vec![w("price_ohlcv_15m", 715_000), w("price_ohlcv_1m", 140_000)]; - let max = max_rows_written(&rows); - assert!( - max < 5_000_000.0, - "legit max {max} should be under a guardrail threshold" - ); - } - - #[test] - fn query_targets_the_write_log_correctly() { - // Reads the write log, counts only part creations, over a server-side - // trailing hour, for the real prices surface only. - assert!(WRITE_VOLUME_QUERY.contains("system.part_log")); - assert!(WRITE_VOLUME_QUERY.contains("database = 'prices'")); - assert!(WRITE_VOLUME_QUERY.contains("event_type = 'NewPart'")); - assert!(WRITE_VOLUME_QUERY.contains("now() - INTERVAL 1 HOUR")); - // Internal storage tables (.inner_id.* / .tmp.*) are excluded so the - // Table space stays bounded and the metric tracks the real surface. - assert!(WRITE_VOLUME_QUERY.contains("table NOT LIKE '.%'")); - } -} diff --git a/packages/write-amplification-probe/src/main.rs b/packages/write-amplification-probe/src/main.rs deleted file mode 100644 index 07a6ee20..00000000 --- a/packages/write-amplification-probe/src/main.rs +++ /dev/null @@ -1,87 +0,0 @@ -//! Write-amplification probe Lambda entrypoint (task 0133). -//! -//! EventBridge `rate(1 hour)` → this binary. Each run reads the rows written to -//! every `prices.*` table in the trailing hour from `system.part_log` over the -//! 0052 mTLS ClickHouse client, and republishes the max as the `Prices/Ingest` -//! `MaxRowsWrittenPerHour` CloudWatch metric that the write-amplification alarm -//! watches. -//! -//! cargo lambda build -p write-amplification-probe --release --arm64 --features lambda -//! -//! Reads as the **`prices_reader`** identity (the `api` mTLS bundle, set on the -//! Function env by infra), which is granted `SELECT ON system.part_log` (task -//! 0133 prerequisite) in addition to its `prices.*` read. Requires the `lambda` -//! feature (the default build/test exercises the pure metric-shaping in `lib.rs` -//! without the AWS runtime / mTLS stack). - -#[cfg(feature = "lambda")] -#[tokio::main] -async fn main() -> Result<(), lambda_runtime::Error> { - use lambda_runtime::{LambdaEvent, run, service_fn}; - use std::sync::Arc; - use write_amplification_probe::{TableWrite, WRITE_VOLUME_QUERY, max_rows_written, publish}; - - prices_clickhouse::observability::init_tracing(); - - // Cold start: build the CH + CloudWatch clients once. A bad secret/endpoint - // or a missing part_log grant surfaces on the first query (the invocation - // fails and the probe's own `-errors` alarm fires) — no separate liveness - // probe needed. Reads as `prices_reader` (the `api` bundle, wired by infra), - // which has the task-0133 `SELECT ON system.part_log` grant. - let ch = Arc::new(prices_clickhouse::mtls::client_from_lambda_env("prices").await?); - - let aws_cfg = aws_config::defaults(aws_config::BehaviorVersion::latest()) - .load() - .await; - let cw = Arc::new(aws_sdk_cloudwatch::Client::new(&aws_cfg)); - let environment = Arc::new(prices_clickhouse::env::env_or("ENV_NAME", "unknown")); - tracing::info!(environment = %environment, "write-amplification-probe cold start ready"); - - run(service_fn(move |_event: LambdaEvent| { - let ch = ch.clone(); - let cw = cw.clone(); - let environment = environment.clone(); - async move { - let rows = ch - .query(WRITE_VOLUME_QUERY) - .fetch_all::() - .await?; - let max_rows = max_rows_written(&rows); - - // Propagate a publish failure so the invocation errors (mirrors the - // freshness / notafter probes). The alarm treats missing data as - // NOT_BREACHING, so a swallowed PutMetricData failure would blind it; - // failing instead trips the probe's own `-errors` alarm — the - // intended dead-probe signal. A transient blip self-heals next hour. - publish(&cw, &environment, max_rows).await?; - - // Log the top writers so a breach is immediately diagnosable from the - // invocation log without a manual part_log query. - let top: Vec<_> = rows - .iter() - .take(5) - .map(|r| serde_json::json!({ "table": r.table, "rows_written": r.rows_written })) - .collect(); - tracing::info!( - max_rows, - tables = rows.len(), - top = %serde_json::Value::Array(top), - "write-amplification-probe run complete" - ); - Ok::(serde_json::json!({ - "max_rows_written": max_rows, - "tables_written": rows.len(), - })) - } - })) - .await -} - -#[cfg(not(feature = "lambda"))] -fn main() { - eprintln!( - "write-amplification-probe: build with `--features lambda` (or `cargo lambda build -p \ - write-amplification-probe --release --arm64 --features lambda`) for the AWS Lambda \ - entrypoint." - ); -}