diff --git a/docs/etl.md b/docs/etl.md index d7b0ec96..c0f5a9a7 100644 --- a/docs/etl.md +++ b/docs/etl.md @@ -116,12 +116,27 @@ node scripts/load-eop.mjs --from=YYYY-MM-DD --to=YYYY-MM-DD --no-ocds Първоначалният backfill и ежедневният refresh ползват едни и същи staging таблици, mapper-и и SQL. Различават се по прозореца от дати и режима на derive: -- **голямо или първоначално догонване:** CLI прозорец + пълен derive; -- **малък steady-state refresh:** gap-aware прозорец + slice derive. +- **първоначално зареждане или пълно презареждане:** прозорец от началото на емисията + пълен derive; +- **догонване и steady-state refresh:** gap-aware прозорец + slice derive. `--derive=full` пуска amendment rollup, FX, NUTS, пълна нормализация и precompute. `--derive=slice` -пуска scoped refresh SQL-а. По подразбиране catch-up логиката избира full derive за големи -празнини и slice derive за малки. +пуска scoped refresh SQL-а. + +Пълният derive **презижда** доменните таблици от staging — `normalize-raw.sql` започва с +`DELETE FROM contracts` — тоест всичко извън заредения прозорец отпада и не се връща. Затова той е +допустим само когато прозорецът стига до началото на емисията (или когато още няма корпус). Понеже +gap-aware прозорецът по устройство покрива само опашката, `--catchup` върху **вече зареден** корпус +прави slice derive, колкото и голяма да е празнината. Изключението е първото пускане: когато няма +никакви заредени дни, догонването взима прозорец от началото на емисията и пълен derive — там няма +какво да се загуби. + +Пълен derive с частичен прозорец върху вече зареден корпус `import.mjs` отказва да продължи — преди +зареждането — вместо да изтрие историята. Отказът важи за **действащия** режим, не само за изрично +подадения: без `--catchup` подразбиращият се derive е `full`, тъй че и `import.mjs --from=2026-06-01` +получава същия отказ. Проверката пита за всяка таблица, която `normalize-raw.sql` изпразва (виж +`@full-clear` там), не само за `contracts` — корпус без договори, но с попълнени `tenders` или +`bidders` е точно състоянието, което половинчат пробег оставя. Ако някоя от тях не може да бъде +прочетена, отказът пак важи: проверка, която не може да провери, не бива да пуска нататък. ## Доменна нормализация (derive) diff --git a/packages/ingest/src/ocds.test.ts b/packages/ingest/src/ocds.test.ts index 014f65f9..fd42274a 100644 --- a/packages/ingest/src/ocds.test.ts +++ b/packages/ingest/src/ocds.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest'; import { classifyBucketKey, computeCatchupWindow, + fullDeriveIsSafe, releaseToAmendments, releaseToContracts, releaseToLots, @@ -352,6 +353,27 @@ describe('bucket key and catchup helpers', () => { computeCatchupWindow({ maxLoadedDate: '2026-06-01', today: '2026-06-07', lookbackDays: 3 }), ).toEqual({ from: '2026-05-29', to: '2026-06-07' }); }); + + it('allows a full derive only when the window reaches the start of the feed', () => { + // Initial backfill: nothing to lose, any window is safe. + expect( + fullDeriveIsSafe({ windowFrom: '2026-06-10', feedStart: '2020-01-01', hasCorpus: false }), + ).toBe(true); + // Whole feed reloaded into staging — the rebuild is complete. + expect( + fullDeriveIsSafe({ windowFrom: '2020-01-01', feedStart: '2020-01-01', hasCorpus: true }), + ).toBe(true); + // A catch-up window over an existing corpus would drop everything before it. + expect( + fullDeriveIsSafe({ windowFrom: '2026-06-10', feedStart: '2020-01-01', hasCorpus: true }), + ).toBe(false); + }); + + it('rejects malformed days rather than silently allowing a full derive', () => { + expect(() => + fullDeriveIsSafe({ windowFrom: '10-06-2026', feedStart: '2020-01-01', hasCorpus: true }), + ).toThrow(/windowFrom/); + }); }); describe('splitSqlStatements', () => { diff --git a/packages/ingest/src/ocds.ts b/packages/ingest/src/ocds.ts index fb460cfa..1133671e 100644 --- a/packages/ingest/src/ocds.ts +++ b/packages/ingest/src/ocds.ts @@ -452,6 +452,27 @@ export function computeCatchupWindow({ return { from: from > today ? today : from, to: today }; } +/** + * A full derive rebuilds the domain from whatever the staging tables hold — `scripts/normalize-raw.sql` + * opens with `DELETE FROM contracts` — so every contract outside the loaded window is dropped. It is + * only sound when the window reaches back to the first day the feed is loaded from, or when there is + * no corpus yet (the initial backfill). A gap-aware catch-up window never does, which is why + * `--catchup` derives a slice. + */ +export function fullDeriveIsSafe({ + windowFrom, + feedStart, + hasCorpus, +}: { + windowFrom: string; + feedStart: string; + hasCorpus: boolean; +}): boolean { + validateDay(windowFrom, 'windowFrom'); + validateDay(feedStart, 'feedStart'); + return !hasCorpus || windowFrom <= feedStart; +} + export function daysInWindow(from: string, to: string): number { validateDay(from, 'from'); validateDay(to, 'to'); diff --git a/packages/ingest/src/refresh-full-clear.test.ts b/packages/ingest/src/refresh-full-clear.test.ts new file mode 100644 index 00000000..c90f9f4a --- /dev/null +++ b/packages/ingest/src/refresh-full-clear.test.ts @@ -0,0 +1,77 @@ +/// +import { readFileSync } from 'node:fs'; +import { dirname, resolve } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import { describe, expect, it } from 'vitest'; +import { fullClearTables } from './refresh'; + +// fullClearTables answers one question for scripts/import.mjs: which tables does a full derive empty, +// and therefore what does a partial window destroy? The guard that used to ask it named `contracts` +// alone while the clear had reached fourteen tables — so the tests that matter here are the ones that +// keep the answer tied to the SQL rather than to a copy of it. + +const root = resolve(dirname(fileURLToPath(import.meta.url)), '../../..'); +const normalizeRaw = readFileSync(resolve(root, 'scripts/normalize-raw.sql'), 'utf8'); + +describe('fullClearTables', () => { + it('reads the block out of the real normalize-raw.sql', () => { + const tables = fullClearTables(normalizeRaw); + // The base domain tables: whatever else the block grows, losing any of these is losing the corpus. + expect(tables).toEqual( + expect.arrayContaining(['contracts', 'lots', 'tenders', 'bidders', 'authorities']), + ); + }); + + it('stops before the per-run metadata resets further down the file', () => { + // normalize-raw.sql also clears data_freshness and pipeline_stats, which are rewritten every run. + // Counting them as corpus would make the guard refuse EVERY full derive, initial backfill included + // — a guard that always refuses gets deleted, so this boundary is load-bearing. + const tables = fullClearTables(normalizeRaw); + expect(tables).not.toContain('data_freshness'); + expect(tables).not.toContain('pipeline_stats'); + }); + + it('keeps the marker and the block adjacent', () => { + // If the marker is dropped or drifts away from the DELETEs, this returns [] and import.mjs throws + // rather than silently deciding the corpus is empty and letting the destructive path through. + expect(fullClearTables(normalizeRaw).length).toBeGreaterThanOrEqual(5); + }); + + it('takes only the marked block, and only unqualified deletes', () => { + const sql = [ + 'DELETE FROM before_the_marker;', + '-- @full-clear', + 'DROP TABLE IF EXISTS scratch;', + 'DELETE FROM contracts;', + "DELETE FROM lots WHERE id = 'x';", // scoped: not a full clear + 'DELETE FROM authorities;', + '', + 'DELETE FROM after_the_block;', + ].join('\n'); + expect(fullClearTables(sql)).toEqual(['contracts', 'authorities']); + }); + + it('sees a table however it is quoted', () => { + // `DELETE FROM "search_index";` is valid SQLite and reads as pure formatting. A bare-identifier + // matcher dropped it from the list, which silently reopened the data-loss hole: the guard would + // then wave through a corpus whose only populated table was the re-quoted one. + const sql = [ + '-- @full-clear', + 'DELETE FROM "search_index";', + 'DELETE FROM `flow_pairs`;', + 'DELETE FROM [home_totals];', + 'delete from contracts;', + '', + ].join('\n'); + expect(fullClearTables(sql)).toEqual([ + 'search_index', + 'flow_pairs', + 'home_totals', + 'contracts', + ]); + }); + + it('returns nothing when the marker is absent', () => { + expect(fullClearTables('DELETE FROM contracts;\nDELETE FROM lots;\n')).toEqual([]); + }); +}); diff --git a/packages/ingest/src/refresh.ts b/packages/ingest/src/refresh.ts index 4159da90..45808bf3 100644 --- a/packages/ingest/src/refresh.ts +++ b/packages/ingest/src/refresh.ts @@ -108,6 +108,45 @@ export function transientStagingStatements(workStagingSchemaSql: string): string ); } +const FULL_CLEAR_MARKER = /^--\s*@full-clear\b/i; +// All three SQLite quoting styles, not just the bare identifier. Rewriting one line as +// `DELETE FROM "search_index";` is a valid, invisible formatting change — and with a bare-only +// matcher it would drop that table out of the guard's list and quietly reopen the hole this parser +// exists to close. +const DELETE_FROM = + /^DELETE\s+FROM\s+(?:"([^"]+)"|`([^`]+)`|\[([^\]]+)\]|([A-Za-z_][A-Za-z0-9_]*))\s*;?\s*$/i; + +/** + * The tables `scripts/normalize-raw.sql` empties before rebuilding the domain from staging, read out + * of the SQL rather than restated in JS. The guard that consumes this list used to ask about + * `contracts` alone while the clear had grown to fourteen tables — a hardcoded copy of a destructive + * list is a data-loss bug on a timer, so the list has exactly one home. + * + * Scoped to the `@full-clear` block on purpose: the same file later resets `data_freshness` and + * `pipeline_stats`, which are per-run metadata. Counting those as corpus would make the guard refuse + * every full derive, including the initial backfill it is supposed to let through. + */ +export function fullClearTables(normalizeRawSql: string): string[] { + const tables: string[] = []; + let inBlock = false; + for (const line of normalizeRawSql.split(/\r?\n/)) { + const trimmed = line.trim(); + if (FULL_CLEAR_MARKER.test(trimmed)) { + inBlock = true; + continue; + } + if (!inBlock) continue; + const hit = trimmed.match(DELETE_FROM); + if (hit) { + tables.push((hit[1] ?? hit[2] ?? hit[3] ?? hit[4])!); + continue; + } + // Comments and the DROP TABLEs share the block; a blank line ends it. + if (trimmed === '') break; + } + return tables; +} + export function dropTransientStagingStatements(): string[] { return [...TRANSIENT_STAGING_TABLES, ...LEGACY_TRANSIENT_STAGING_TABLES] .reverse() diff --git a/scripts/import-guard.test.mjs b/scripts/import-guard.test.mjs new file mode 100644 index 00000000..f82c8b8c --- /dev/null +++ b/scripts/import-guard.test.mjs @@ -0,0 +1,201 @@ +// The --derive=full window guard, exercised through the real scripts/import.mjs. +// +// Testing the predicate alone is what let the first version ship: fullDeriveIsSafe() was green while +// the call site asked `SELECT COUNT(*) FROM contracts` and the clear it was protecting emptied +// fourteen tables. So this drives the actual script as a subprocess, with a fake `wrangler` and a +// fake `node` first on PATH, and asserts on what the script DID: whether it refused, which tables it +// named, whether it cleaned up after itself, and whether the load ever started. +// +// Run: node --test scripts/import-guard.test.mjs +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { spawnSync } from 'node:child_process'; +import { chmodSync, mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { dirname, join, resolve } from 'node:path'; +import { fileURLToPath } from 'node:url'; + +const HERE = dirname(fileURLToPath(import.meta.url)); +const ROOT = resolve(HERE, '..'); +const SCRIPT = resolve(HERE, 'import.mjs'); + +// `wrangler` answers the probe from GUARD_FAKE_POPULATED and records every call; `node` stands in for +// the child scripts import.mjs shells out to (load-eop and friends), so a run that gets PAST the guard +// stops here instead of hitting the network. Launching the script under test with process.execPath +// keeps the real node for the parent while the child's PATH lookup finds the stub. +// The shebangs name the real interpreter outright. `#!/usr/bin/env node` would resolve through the +// very PATH these stubs sit at the front of, so the fake `node` would end up interpreting the fake +// `wrangler` and every answer would come back empty. +const FAKE_WRANGLER = `#!${process.execPath} +import { appendFileSync } from 'node:fs'; +const argv = process.argv.slice(2); +appendFileSync(process.env.GUARD_FAKE_LOG, JSON.stringify(argv) + '\\n'); +const ci = argv.indexOf('--command'); +if (argv.includes('--json') && ci !== -1) { + const sql = argv[ci + 1]; + const populated = (process.env.GUARD_FAKE_POPULATED || '').split(',').filter(Boolean); + const missing = (process.env.GUARD_FAKE_MISSING || '').split(',').filter(Boolean); + const aliases = [...sql.matchAll(/AS "([^"]+)"/g)].map((m) => m[1]); + if (aliases.length) { + const row = {}; + for (const a of aliases) if (!missing.includes(a)) row[a] = populated.includes(a) ? 1 : 0; + process.stdout.write(JSON.stringify([{ results: [row], success: true }])); + process.exit(0); + } + process.stdout.write(JSON.stringify([{ results: [], success: true }])); +} +process.exit(0); +`; + +const FAKE_NODE = `#!${process.execPath} +import { appendFileSync } from 'node:fs'; +appendFileSync(process.env.GUARD_FAKE_LOG, JSON.stringify(['node', ...process.argv.slice(2)]) + '\\n'); +process.exit(0); +`; + +function bin(dir, name, source) { + const file = join(dir, name); + writeFileSync(file, source); + chmodSync(file, 0o755); +} + +/** Runs the real import.mjs with the fakes in front, and returns what it did. */ +function runImport(args, { populated = '', missing = '' } = {}) { + const dir = mkdtempSync(join(tmpdir(), 'guard-')); + try { + const binDir = join(dir, 'bin'); + mkdirSync(binDir); + writeFileSync(join(binDir, 'package.json'), '{"type":"module"}'); + bin(binDir, 'wrangler', FAKE_WRANGLER); + bin(binDir, 'node', FAKE_NODE); + const log = join(dir, 'calls.log'); + writeFileSync(log, ''); + const res = spawnSync(process.execPath, [SCRIPT, ...args], { + cwd: ROOT, + encoding: 'utf8', + timeout: 30_000, + env: { + ...process.env, + PATH: `${binDir}:${process.env.PATH}`, + GUARD_FAKE_LOG: log, + GUARD_FAKE_POPULATED: populated, + GUARD_FAKE_MISSING: missing, + SIGMA_D1_NAME: 'sigma-test-local', + }, + }); + const calls = readFileSync(log, 'utf8').trim().split('\n').filter(Boolean).map(JSON.parse); + return { + status: res.status, + stderr: res.stderr ?? '', + calls, + refused: /refusing --derive=full/.test(res.stderr ?? ''), + loadStarted: calls.some((c) => c[0] === 'node' && String(c[1]).includes('load-eop')), + // The guard's own query, not merely any --json call: the derive paths issue plenty of their own. + probe: calls.find((c) => /EXISTS\(SELECT 1 FROM contracts\)/.test(String(c[c.length - 1]))), + }; + } finally { + rmSync(dir, { recursive: true, force: true }); + } +} + +const PARTIAL = ['--from=2026-06-01', '--derive=full']; + +test('refuses a partial-window full derive over a populated corpus, before the load', () => { + const r = runImport(PARTIAL, { populated: 'contracts,lots,tenders,bidders,authorities' }); + assert.equal(r.refused, true, r.stderr); + assert.equal(r.status, 1); + assert.equal(r.loadStarted, false, 'the load must not start after a refusal'); + assert.match(r.stderr, /contracts/); +}); + +test('a corpus with NO contracts but populated tenders still refuses', () => { + // The regression this guard exists for. The previous call site asked COUNT(*) FROM contracts, so + // exactly this state - the one a half-failed run leaves behind - was waved through, and the rebuild + // would then drop tenders, bidders and authorities with nothing to reload them from. + const r = runImport(PARTIAL, { populated: 'tenders,bidders,authorities' }); + assert.equal(r.refused, true, r.stderr); + assert.equal(r.status, 1); + assert.match(r.stderr, /tenders/); + assert.doesNotMatch(r.stderr, /including [^\n]*\bcontracts\b/); +}); + +test('refuses when the probe cannot answer for every cleared table', () => { + // A missing table makes safeD1 return nothing at all, which would otherwise read as "no corpus". + const r = runImport(PARTIAL, { populated: 'contracts', missing: 'facet_counts' }); + assert.equal(r.status, 1); + assert.match(r.stderr, /could not read the corpus/); +}); + +test('an empty corpus is the initial backfill and passes', () => { + const r = runImport(PARTIAL, { populated: '' }); + assert.equal(r.refused, false, r.stderr); + assert.equal(r.loadStarted, true, 'the load should start when there is nothing to lose'); +}); + +test('a window reaching the start of the feed passes even over a full corpus', () => { + const r = runImport(['--from=2020-01-01', '--derive=full'], { + populated: 'contracts,authorities', + }); + assert.equal(r.refused, false, r.stderr); + assert.equal(r.loadStarted, true); +}); + +test('a slice derive is never probed at all', () => { + const r = runImport(['--from=2026-06-01', '--derive=slice'], { populated: 'contracts' }); + assert.equal(r.refused, false, r.stderr); + assert.equal(r.probe, undefined, 'slice derives must not pay for the corpus probe'); +}); + +// Written out by hand ON PURPOSE. The first version of this test re-derived the list from +// normalize-raw.sql using its own copy of the parser's regex — so the oracle inherited the parser's +// blind spot, and rewriting one line as `DELETE FROM "search_index";` (valid SQLite, invisible in +// review) dropped that table out of the probe with both suites still green. A list that shares the +// implementation's assumptions cannot test them. +const CLEARED = [ + 'search_index', + 'flow_pairs', + 'company_totals', + 'authority_joint_participation', + 'authority_totals', + 'sector_totals', + 'facet_counts', + 'home_totals', + 'contract_co_authorities', + 'contracts', + 'lots', + 'tenders', + 'bidders', + 'authorities', +]; + +test('the probe covers every table the SQL clears, not a subset', () => { + const r = runImport(PARTIAL, { populated: 'contracts' }); + const sql = String(r.probe[r.probe.indexOf('--command') + 1]); + for (const table of CLEARED) { + assert.match(sql, new RegExp(`FROM ${table}\\)`), `probe is missing ${table}`); + } +}); + +test('the hand-written list still matches what the SQL clears', () => { + // The other half of the cross-check: the list above holds the parser to account, and this holds the + // list to account. Either one drifting is caught here instead of in production. The matcher is + // deliberately loose about quoting so that a re-quoted table shows up as a MISMATCH rather than + // vanishing the way it did from the first version. + const sql = readFileSync(resolve(ROOT, 'scripts/normalize-raw.sql'), 'utf8'); + const marker = sql.indexOf('-- @full-clear'); + assert.notEqual(marker, -1, 'normalize-raw.sql lost its @full-clear marker'); + const block = sql.slice(marker).split(/\r?\n\s*\r?\n/)[0]; + const deletes = [...block.matchAll(/DELETE\s+FROM\s+(.+?)\s*;/gi)].map((m) => + m[1].replace(/^["`[]/, '').replace(/["`\]]$/, ''), + ); + assert.deepEqual(deletes, CLEARED); +}); + +test('a refusal leaves no transient staging behind', () => { + const r = runImport(PARTIAL, { populated: 'contracts' }); + const files = r.calls.filter((c) => c.includes('--file')).map((c) => c[c.indexOf('--file') + 1]); + assert.ok( + files.some((f) => /drop-transient-staging\.sql$/.test(String(f))), + 'the guard should tear down the staging it found created', + ); +}); diff --git a/scripts/import.mjs b/scripts/import.mjs index 578a1132..43ec75c5 100644 --- a/scripts/import.mjs +++ b/scripts/import.mjs @@ -14,9 +14,14 @@ import { } from 'node:fs'; import { basename, dirname, resolve } from 'node:path'; import { fileURLToPath } from 'node:url'; -import { computeCatchupWindow, daysInWindow } from '../packages/ingest/src/ocds.ts'; +import { + computeCatchupWindow, + daysInWindow, + fullDeriveIsSafe, +} from '../packages/ingest/src/ocds.ts'; import { dropTransientStagingStatements, + fullClearTables, refreshSliceStatementGroups, } from '../packages/ingest/src/refresh.ts'; import { assertIntegrity } from './integrity-checks.mjs'; @@ -37,7 +42,6 @@ function reportAnomalies(runner, label) { const root = resolve(dirname(fileURLToPath(import.meta.url)), '..'); const apiDir = resolve(root, 'apps/web'); const DEFAULT_FROM = '2020-01-01'; -const LARGE_GAP_DAYS = 14; const DEFAULT_LOOKBACK_DAYS = 3; const remote = process.argv.includes('--remote'); @@ -215,12 +219,11 @@ function resolveCatchupPlan() { const to = String(arg('to') || window.to); const gapDays = daysInWindow(from, to); const requestedDerive = arg('derive'); - const derive = - requestedDerive && requestedDerive !== true - ? String(requestedDerive) - : gapDays > LARGE_GAP_DAYS - ? 'full' - : 'slice'; + // The catch-up window is gap-aware, so it only ever covers the tail of the feed. A full derive + // rebuilds `contracts` from staging (normalize-raw.sql opens with DELETE FROM contracts), which + // would drop every contract older than the window — so catch-up always derives a slice, however + // wide the gap. An operator who really has loaded the whole feed can still pass --derive=full. + const derive = requestedDerive && requestedDerive !== true ? String(requestedDerive) : 'slice'; return { from, to, maxLoadedDate, gapDays, derive }; } @@ -229,6 +232,57 @@ function validateDeriveMode(mode) { throw new Error(`unknown --derive=${mode}; expected full|slice`); } +// Refuse the one combination that silently destroys data: a full derive (which rebuilds the domain +// from staging) driven by a window that does not reach back to the start of the feed. Anything the +// window misses is deleted and never reloaded. Checked before the load, and the refusal tears the +// transient staging back down: it cannot run any earlier, because the catch-up plan reads +// raw_contracts, but it must not leave a half-built schema behind for a run that never started. +// +// The question is asked of EVERY table the full clear empties, not of `contracts` alone: a corpus +// with no contracts but populated tenders, bidders or authorities is exactly the state a half-failed +// run leaves behind, and the narrow-window rebuild would then wipe those too while the guard waved +// it through. The list comes out of normalize-raw.sql itself (see @full-clear there). +function assertDeriveWindowSafe(mode, from) { + if (mode !== 'full') return; + const tables = fullClearTables(readFileSync(resolve(root, 'scripts/normalize-raw.sql'), 'utf8')); + if (tables.length === 0) + throw new Error( + 'normalize-raw.sql has no @full-clear block — refusing to guess what it clears', + ); + // EXISTS per table, not COUNT(*) over the union: this runs on every full derive and must stay + // cheap on a corpus of hundreds of thousands of rows. + const probe = tables.map((t) => `(SELECT EXISTS(SELECT 1 FROM ${t})) AS "${t}"`).join(', '); + const row = safeD1(`SELECT ${probe}`)[0]; + // safeD1 turns a missing table into an empty result, which would otherwise read as "no corpus" and + // wave the destructive path through — one absent table blinding the guard about the other thirteen. + // A probe that could not answer is not an answer: fail closed. + if (!row || tables.some((table) => !(table in row))) { + console.error( + `!! refusing --derive=full: could not read the corpus. Every table normalize-raw.sql clears ` + + `(${tables.join(', ')}) must be answerable before a rebuild may drop it.`, + ); + execSqlStatements(dropTransientStagingStatements(), 'drop-transient-staging'); + process.exit(1); + } + const populated = Object.entries(row) + .filter(([, present]) => Number(present) > 0) + .map(([table]) => table); + if ( + fullDeriveIsSafe({ windowFrom: from, feedStart: DEFAULT_FROM, hasCorpus: populated.length > 0 }) + ) + return; + console.error( + `!! refusing --derive=full: the load window starts ${from}, but the corpus is already ` + + `populated back to ${DEFAULT_FROM}.\n` + + ` A full derive rebuilds the domain from staging, so everything before ${from} would be ` + + `dropped and not reloaded — including ${populated.join(', ')}.\n` + + ` Use --derive=slice for an incremental refresh, or reload the whole feed with ` + + `--from=${DEFAULT_FROM}.`, + ); + execSqlStatements(dropTransientStagingStatements(), 'drop-transient-staging'); + process.exit(1); +} + async function runFullDerive() { execSql(resolve(root, 'scripts/derive-amendments.sql')); run('node', ['scripts/load-fx.mjs', '--apply', ...passthru]); @@ -370,19 +424,24 @@ if (arg('work-db') !== undefined) { console.log(`==> Sigma import (${remote ? 'REMOTE' : 'local'})`); run('wrangler', ['d1', 'migrations', 'apply', d1Name, loc, ...d1PersistArgs], apiDir); execSqlStatements(dropTransientStagingStatements(), 'drop-stale-transient-staging'); +// Must precede resolveCatchupPlan(): latestLoadedDate() reads raw_contracts, which lives here. execSql(resolve(root, 'scripts/work-staging-schema.sql')); let deriveMode = String(arg('derive') || 'full'); let loadFlags = explicitRangeFlags(); +// Mirrors load-eop.mjs, which also falls back to DEFAULT_FROM when no --from is given. +let windowFrom = String(arg('from') || DEFAULT_FROM); if (catchup) { const plan = resolveCatchupPlan(); deriveMode = plan.derive; loadFlags = rangeFlags(plan.from, plan.to); + windowFrom = plan.from; console.log( `==> catchup window ${plan.from}..${plan.to} (${plan.gapDays} days, latest=${plan.maxLoadedDate || 'none'}, derive=${deriveMode})`, ); } validateDeriveMode(deriveMode); +assertDeriveWindowSafe(deriveMode, windowFrom); run('node', ['scripts/load-eop.mjs', '--apply', ...loadFlags, ...passthru]); if (deriveMode === 'slice') await runSliceDerive(); diff --git a/scripts/normalize-raw.sql b/scripts/normalize-raw.sql index ebfef975..68c3dbf5 100644 --- a/scripts/normalize-raw.sql +++ b/scripts/normalize-raw.sql @@ -31,6 +31,10 @@ -- every contract has a parent. bids stays empty (the data has a bid COUNT, not bids). -- Full clear in child→parent order (D1 enforces FKs). +-- @full-clear — everything emptied between this marker and the blank line below is rebuilt from +-- staging alone, so a full derive driven by a partial window loses whatever the window misses. +-- scripts/import.mjs reads the list from here to decide whether refusing is warranted; keep new +-- DELETEs inside the block so the guard picks them up on its own. DROP TABLE IF EXISTS joint_tender_leads; DROP TABLE IF EXISTS unp_prefix_authorities; DROP TABLE IF EXISTS joint_authority_members;