diff --git a/.planning/workstreams/whole-source-census/TICKET.md b/.planning/workstreams/whole-source-census/TICKET.md index a97c1606e..8af8a2033 100644 --- a/.planning/workstreams/whole-source-census/TICKET.md +++ b/.planning/workstreams/whole-source-census/TICKET.md @@ -289,3 +289,42 @@ CI on 54b8bcd0 passed; these reduction changes require their own full CI gate. - [x] Make PR886 ready at operator request and save self-contained Grok handoff, upstream/CI receipts and broader original-goal backlog; `gh pr view` verifies `isDraft=false`, receipt JSON and handoff checklist inspected. The original-source comparison is recorded on the 16:40 ET bullets above. Deferred acceptance and retirement remain incomplete. 2026-10-09 13:42 ET - [x] Write the handoff for Claude or Codex. `.planning/workstreams/whole-source-census/HANDOFF-20261010.md`, 2026-10-10 06:42 ET. The installed proof stays open. + +## Operator handoff to Claude — 2026-10-10 + +Operator, 2026-10-10 07:08 ET: "merge and take over the goals from grok handover including the checklist". Claude owns this goal from here, on `claude/whole-source-census` (worktree `edgartools-platform-claude-census`). Source: `HANDOFF-20261010.md` (Grok, 06:42 ET). Its guards stay: no further full `source.read` execute; the checkout replay is not the installed proof; no live SEC request; `sec_client.py` stays until a live-SEC proof is allowed; `READING.md`/`COMBINING.md` edited only with the guard; new prose says Name Frequency, code names stay. + +- [x] Take over: handoff read; evidence present (2026-10-10 07:08 ET): saved reading, separate verify, reduction, checkout counts (`/private/tmp/codex-worker-replay-proof/`, `/private/tmp/counts-pin-20261009.json`), engine build `/private/tmp/codex-census-eof-native-site-v3`, seven bundles (`proving/cm27/bundles`, 6,726 records) +- [x] Installed empty PostgreSQL 16 proof: 6,414 Companies, 3,052 CIK+LEI bindings, an unchanged second pass, and recovery; a separate worker and verifier outside the checkout (driver: `installed_proof.py`). Passed 2026-10-10 10:36 ET in 33 min 46 s; report `evidence/installed-company-proof-20261010.json` sha256 `fa85e0d2…` + - [x] Before the long run: at least 3 GB free for the engine build. 01d worktree removed after Trial A; 4.7 GB free, the engine build used under 1 GB. 2026-10-10 08:09 ET + - [x] Before the long run: the installed tool loads GLEIF `level1-json.yaml` (the "object names fields" refusal was removed by #857, `220b8a49`; the older installed build still had it). `stream_policy` on the installed bundle's own rules, 2026-10-10 08:22 ET + - [x] SEC half alone (`COMPANY_PROOF_ONLY=sec`): plumbing, recovery and the unchanged second pass. 6,414 Companies, 6,414 SEC records bound (all by `company-cik`), 312 classification reviews deferred, second pass unchanged (bindings `67749419…`). The first try killed a worker that had already finished (exit 0): the merge writes an item in one transaction, so rows appear only on commit. The kill now lands while the item is `running` and the proof fails otherwise (`8882f9d6`). Rerun: killed with SIGKILL holding `sec-first-chunk1` attempt 1, no rows committed, resumed after the lease, same counts; 7 min 13 s. 2026-10-10 08:29 ET + - [x] Full proof: SEC and GLEIF, counts equal to the checkout bindings. Bundle installed from this branch by `uv tool install`; every step by `edgar-warehouse workers work` and checked by `workers verify` under a separate login. First pass: 6,414 Companies; SEC 6,414 bound (`company-cik`); GLEIF Level 1 3,716 records, 3,052 bound (2,861 `sec-gleif-name-jurisdiction`, 191 `sec-gleif-name-postal`); open reviews 664 binding required, 312 classification deferred, 39 unsupported identity kind. Bindings `9216cc18…`, equal to the checkout replay. Second pass changed nothing. Recovery: SIGKILL while holding `sec-first-chunk1` attempt 1, no rows committed, finished after the lease. GLEIF `source.read` execute and its verifier: 20 min. 2026-10-10 10:36 ET + - [x] Two failed runs, fixed: (1) both GLEIF passes named one consumer, so the second pass's checkpoint 0→1 was refused as stale; each unit now names its own (`mdm_prepare.py`: "a unit's consumer is its own") (`b142560e`), 09:28 ET. (2) The disk guard stopped the rerun at 870 MB free while the second pass re-read the 13 GB archive; the second GLEIF pass now prepares and merges the first pass's verified reading again (`gleif-replay` target, `424e5b09`), 09:59 ET. Leftovers of the killed run (its PostgreSQL, 652 MB) removed + - [x] Review (Standards, Spec, GoF), fixes in `2e982a2b` and `34fe52dc`: bindings equal to the checkout replay are asserted; the SEC second pass gives each copied batch its own id and consumer, so it is mastered again, not skipped; the resumed item must be verified on its next attempt; `pg16.server()` yields only once the server answers `SELECT 1`. 92 integration tests pass on pgserver. SEC half on `2e982a2b`, 10:47 ET: killed after the item's MDM rows committed (919 rows) and before it was reported; resumed on attempt 2, verified; second pass re-mastered and unchanged (`evidence/installed-sec-recovery-after-commit-20261010.json`). 2026-10-10 10:47 ET + - [x] Full run on the final commit `fdc020f9` (operator, 2026-10-10: "Delete Codex /tmp caches"; `codex-calendar-target` and `codex-streaming-uv-cache` removed, unused by any process). Passed 13:57 ET in 36 min 50 s: 6,414 Companies; 3,052 GLEIF bound; bindings `9216cc18…` asserted equal to the checkout replay; the killed item verified on attempt 2; both sources' second passes mastered again under their own batches and consumers and changed nothing. Report `evidence/installed-company-proof-final-20261010.json` sha256 `04f108b8…`. 2026-10-10 13:57 ET +- [ ] Replace the active Name Frequency command (`cli.py` still calls `write_name_census`). Only after the installed proof. Blueprint: `crates/source-contract/tests/qualification/whole_source_frequency.py` builds the document from the configured reading and `cascade.assign`. Done test: on the same pinned inputs, the new command writes a file byte-for-byte equal (same sha256) to the old command's + - [x] One function assembles the document: `name_census.document()`, called by the old `build()` (GoF consult: extract it so the old and new paths share it; no pattern). 68 census, Company source and matching tests pass. Branch `claude/name-frequency-assembly` `3a3c1210`. 2026-10-10 08:19 ET + - [x] GLEIF half from the saved configured reading (`census-complete-stream.yaml`, read back with `source_readings.iter_load`): legal holders and last update, other-name holders, matching-pass candidates merged per LEI by source position, and global address counts; then `document()`. `name_census.fold_reading`; two tests read one fixture through `build()` and through the configured stream, documents equal, names and seven-pass answers alike (`4b6c0ec1`). 2026-10-10 08:31 ET + - [x] SEC half from configured readings: `name_frequency.write_name_frequency` reads each capture through `census-landed-company.yaml`, `census-landed-former.yaml` and the business address of `landed-preparation.yaml` with the `source.read` worker and verifier. Matching-pass candidates still go through `company_source.cascade_filer` (old adapters): moving them is part of L4. Fixture: the same bytes as `write_name_census` over two captures; refuses another population's reading, a delta or short publication, a filer in two captures (`a3c7ed5d`). 2026-10-10 08:35 ET + - [x] `mdm name-census` takes `--gleif-reading` and `--gleif-reading-sha256` in place of `--gleif-archive` and `--gleif-sha256` (`604775b6`). Full population: 69 captures and the saved 3,428,477-record reading, 43,245 entries in 174 s, sha256 `84522f67…`, byte-for-byte the `census.json` the seven cm27 bundles pin (the old command re-reads the Golden Copy for about an hour). 2026-10-10 08:39 ET + - [ ] The seven-pass answers: the pinned census has none (the policy's passes are off); when switched on, compare with `complete-name-frequency.json` from the whole-source run +- [ ] Delete the old construction code once no caller remains (`gleif_source.py`: `source_publications.py`, `native_consumption.py`, `company_source.py`, `gleif_publication.py`; `adapters.py`: `mdm_merge.py`, `name_census.py`, `cli.py`, `company_source.py`, `gleif_source.py`, `quality.py`, `store.py`) +- [ ] Issue 20 L4: whole-capture Company equivalence, then delete `company_source.py` (14 original columns match on 1,000 pinned submissions; the current read also carries `last_sync_run_id`, `last_synced_at`) +- [ ] Issue 20 L6: Golden Copy equivalence of the older GLEIF reader, then delete `gleif_source.py` (52 in-memory fixture tests passed 2026-10-09 20:37 ET) + - [x] Found (2026-10-10 10:55 ET): the older reader is reachable only through a manifest's `native_source` (`cli.execute_manifest` → `native_consumption.prepare_native` → `gleif_source.record_evidence`), and nothing in production or tests writes one. Its checks are not all in the configured GLEIF rules: the configured Level 1 reading selects by approved scope and maps through the contract, but the deletion-flag refusal, the relationship checks (one RELATIONSHIP_PERIOD, LEI endpoints, a valid interval, status against the end date) and the reporting-exception category and reason checks live only in `record_evidence` (24 tests, `tests/mdm/test_clean_gleif_source.py`). Out-of-scope records are dropped by the configured reading, where the older reader deferred them. The configured relationship and reporting-exception field readings map no fields (`relationships-fields.yaml`, `reporting-exceptions-fields.yaml`: `fields: {}`), so only `record_evidence` builds the `start`, `end`, `relationship_type`, `valid_from`, `valid_to`, `status` the relationship contract reads: GLEIF relationships have no configured path yet + - [ ] Express those checks in the GLEIF MDM contract or quality rules (ruling above), the relationship and reporting-exception readings projecting the fields the contracts read, with each of the 24 behaviours tested on the configured path + - [x] A contract's record checks set a record aside first (`source_mapping.record_refusal`, `adapters._refuse` at the top of `normalize` and `mapped_values`; GoF consult: one refusal step, not a side effect of reading fields). Branch `claude/gleif-reader-retirement` `027f6b41`, 5 tests, 252 affected pass. 2026-10-10 13:30 ET + - [x] Level 1: LEI shape and check digits, a filled deletion flag (blocking), a category outside GLEIF's six, in the older reader's order, each read through a literal-path column (`00580b18`). Contract digest 9305c467… awaiting the operator's approval; peeled, it is the approved f53c9c42…. 284 GLEIF, census and digest tests pass. 2026-10-10 13:37 ET + - [x] Relationships read by their contract alone: generic `find ... one`, `time_format: instant`, `interval` and `status` link rules (GoF: leave the structure, one `_refuse_link`); the record checks take the LEI ends, endpoint types and deletion flag; the record's own time and whole record kept. 14 cases equal to the older reader under the approved contract b7493a71…; digest 78cb4dd5… awaiting approval (`a01d5b5c`). 2026-10-10 14:31 ET + - [x] Level 1 per-record fidelity (found 2026-10-10 14:05 ET, one fixture both ways; fixed `4da44047`: a contract may name `effective_at`, a provenance path `.` is the whole record; the Level 1 contract uses both, digest fc104ef9… approved by the operator 2026-10-10 14:13 ET: "Approved", on the 20,000-record evidence with the whole-file comparison running; peeled f53c9c42…): the configured path stores no effective time (the older reader stamps each record with its `Registration.LastUpdateDate`) and no native record in provenance (`native_record: None`; the older reader keeps the raw GLEIF record). Bindings are unaffected (the proof matched `9216cc18…`), but assertion ids differ. Needs the contract to name the record's effective time and keep the whole record as its native evidence + - [x] Reporting exceptions read by their contract alone: record checks for the LEI, deletion flag and category; generic `every_in` (every reason in GLEIF's list) and `retained_as` (`reported_parent_exception`). 13 cases equal to the older reader under the approved contract 77bb3a8d…; digest 8ec18c6d… awaiting approval (`8fa2f8f2`). 2026-10-10 14:40 ET + - [ ] Whole-file equivalence for relationships (487,721 records) and reporting exceptions (6,351,397), as for Level 1; then ask approval of 78cb4dd5… and 8ec18c6d… + - [x] Decision (operator, 2026-10-10 13:20 ET: "MDM contract (Recommended)"): where a GLEIF record-level refusal lives on the configured path. The engine's record checks (`lei`, `count`, `absent`, `before`, `required`, `in_set`, `in_lookup`, each with a `reason`) can express every `record_evidence` check, but a record a reading check sets aside stays in the reading (`mdm_prepare.py`) and never reaches MDM, so no steward review is opened and a blocking reason (a filled deletion flag) no longer blocks the run. Options: (a) checks in the readings, with set-aside records reported for review by `mdm.prepare`; (b) checks in the MDM contract or the GLEIF quality rules, so they defer in the Stage as today. Ruled (b): the checks go in the GLEIF MDM contract or its quality rules; the readings only select and project + - [ ] Golden Copy equivalence (`crates/source-contract/tests/qualification/gleif_level1_equivalence.py`, every record as if in scope; first 20,000 all equal, 2026-10-10 14:20 ET; whole file running in 3 shards, about 2 hours): each in-scope record reaches the same outcome (assertion digest, or deferral and its reason) both ways, all three members + - [ ] Delete `record_evidence`, `native_consumption.py` and the `native_source` branch; move `dataset_contract`, `validate_metadata`, `FORMATS`, `release_sequence` to where they belong; delete `gleif_source.py` +- [ ] Issue 20 L8: a configured `provider.capture` worker, then delete `sec_client.py` (needs the operator's permission for a live SEC proof) +- [ ] Issue 20 20c: Company through the engine only (the four Company-named steps renamed to registered profiles, or removed) +- [ ] Issue 20 20d: Person through the engine only (blocked by 20c) +- [ ] Skill audit remainder: installed separate-process qualification, full CI, independent review, a reviewable PR for any skill edit (`READING.md` and `COMBINING.md` still use the old wording) + diff --git a/.planning/workstreams/whole-source-census/evidence/installed-company-proof-20261010.json b/.planning/workstreams/whole-source-census/evidence/installed-company-proof-20261010.json new file mode 100644 index 000000000..f74d329d8 --- /dev/null +++ b/.planning/workstreams/whole-source-census/evidence/installed-company-proof-20261010.json @@ -0,0 +1,46 @@ +{ + "after_first_pass": { + "companies_open": 6414, + "decisions_by_rule": { + "company-cik": 6414, + "sec-gleif-name-jurisdiction": 2861, + "sec-gleif-name-postal": 191 + }, + "open_reviews": { + "binding_required": 664, + "classification_deferred": 312, + "unsupported_identity_kind": 39 + }, + "stage_bound": { + "gleif.level1.v1": 3052, + "sec.submissions.company.v1": 6414 + }, + "stage_records": { + "gleif.level1.v1:company": 3716, + "sec.submissions.company.v1:company": 6414 + } + }, + "bindings": "9216cc18292f1c683f4767dfb86ae89c14a72f7df050df08e3800ce1e305b85b", + "bindings_equal_checkout_replay": true, + "gleif_scope_leis": 3755, + "goal": "whole-source census: installed empty-store Company proof", + "minutes": 30.8, + "only": null, + "policy": "34f548776040bb44451f0ac17e496ae30e1ce20568ad199bfa03616a6645b2e2", + "recovery": { + "item_at_kill": { + "attempt": "b9985863-9240-4cf4-8a6c-463bff57b4da", + "attempts": 1, + "checks": {}, + "error": null, + "ordinal": 0, + "receipt": null, + "state": "running", + "step": "merge", + "unit_key": "sec-first-chunk1" + }, + "killed_worker_exit": -9, + "rows_at_kill": {} + }, + "second_pass_changed_nothing": true +} diff --git a/.planning/workstreams/whole-source-census/evidence/installed-company-proof-20261010.log b/.planning/workstreams/whole-source-census/evidence/installed-company-proof-20261010.log new file mode 100644 index 000000000..245f3af36 --- /dev/null +++ b/.planning/workstreams/whole-source-census/evidence/installed-company-proof-20261010.log @@ -0,0 +1,12 @@ +{"at": "10:05:03", "stage": "provisioned", "policy": "34f548776040bb44451f0ac17e496ae30e1ce20568ad199bfa03616a6645b2e2"} +{"at": "10:05:17", "stage": "worker killed", "killed_worker_exit": -9, "item_at_kill": {"attempt": "b9985863-9240-4cf4-8a6c-463bff57b4da", "attempts": 1, "checks": {}, "error": null, "ordinal": 0, "receipt": null, "state": "running", "step": "merge", "unit_key": "sec-first-chunk1"}, "rows_at_kill": {}} +{"at": "10:11:27", "run": "5a9445b5-d6c9-4492-b6ca-d2b9cceb2851", "step": "mdm.merge", "verified": true, "minutes": 6.4} +{"at": "10:31:33", "run": "b9a79d46-6b83-42b7-a1e6-c1e690f97c23", "step": "source.read", "verified": true, "minutes": 26.5} +{"at": "10:31:38", "run": "b9a79d46-6b83-42b7-a1e6-c1e690f97c23", "step": "mdm.prepare", "verified": true, "minutes": 26.6} +{"at": "10:33:38", "run": "b9a79d46-6b83-42b7-a1e6-c1e690f97c23", "step": "mdm.merge", "verified": true, "minutes": 28.6} +{"at": "10:33:39", "stage": "first pass", "companies_open": 6414, "stage_records": {"gleif.level1.v1:company": 3716, "sec.submissions.company.v1:company": 6414}, "stage_bound": {"gleif.level1.v1": 3052, "sec.submissions.company.v1": 6414}, "open_reviews": {"binding_required": 664, "classification_deferred": 312, "unsupported_identity_kind": 39}, "decisions_by_rule": {"company-cik": 6414, "sec-gleif-name-postal": 191, "sec-gleif-name-jurisdiction": 2861}} +{"at": "10:35:00", "run": "614b7219-da85-4869-9c07-38bdfb264abf", "step": "mdm.merge", "verified": true, "minutes": 30.0} +{"at": "10:35:07", "run": "171d5f4e-5063-4efb-b816-bcaaa55b988c", "step": "mdm.prepare", "verified": true, "minutes": 30.1} +{"at": "10:35:50", "run": "171d5f4e-5063-4efb-b816-bcaaa55b988c", "step": "mdm.merge", "verified": true, "minutes": 30.8} +{"at": "10:35:52", "stage": "report", "goal": "whole-source census: installed empty-store Company proof", "only": null, "policy": "34f548776040bb44451f0ac17e496ae30e1ce20568ad199bfa03616a6645b2e2", "gleif_scope_leis": 3755, "after_first_pass": {"companies_open": 6414, "stage_records": {"gleif.level1.v1:company": 3716, "sec.submissions.company.v1:company": 6414}, "stage_bound": {"gleif.level1.v1": 3052, "sec.submissions.company.v1": 6414}, "open_reviews": {"binding_required": 664, "classification_deferred": 312, "unsupported_identity_kind": 39}, "decisions_by_rule": {"company-cik": 6414, "sec-gleif-name-postal": 191, "sec-gleif-name-jurisdiction": 2861}}, "bindings": "9216cc18292f1c683f4767dfb86ae89c14a72f7df050df08e3800ce1e305b85b", "second_pass_changed_nothing": true, "recovery": {"killed_worker_exit": -9, "item_at_kill": {"attempt": "b9985863-9240-4cf4-8a6c-463bff57b4da", "attempts": 1, "checks": {}, "error": null, "ordinal": 0, "receipt": null, "state": "running", "step": "merge", "unit_key": "sec-first-chunk1"}, "rows_at_kill": {}}, "bindings_equal_checkout_replay": true, "minutes": 30.8} +1 passed in 2026.17s (0:33:46) diff --git a/.planning/workstreams/whole-source-census/evidence/installed-company-proof-final-20261010.json b/.planning/workstreams/whole-source-census/evidence/installed-company-proof-final-20261010.json new file mode 100644 index 000000000..b7ec5275b --- /dev/null +++ b/.planning/workstreams/whole-source-census/evidence/installed-company-proof-final-20261010.json @@ -0,0 +1,47 @@ +{ + "after_first_pass": { + "companies_open": 6414, + "decisions_by_rule": { + "company-cik": 6414, + "sec-gleif-name-jurisdiction": 2861, + "sec-gleif-name-postal": 191 + }, + "open_reviews": { + "binding_required": 664, + "classification_deferred": 312, + "unsupported_identity_kind": 39 + }, + "stage_bound": { + "gleif.level1.v1": 3052, + "sec.submissions.company.v1": 6414 + }, + "stage_records": { + "gleif.level1.v1:company": 3716, + "sec.submissions.company.v1:company": 6414 + } + }, + "bindings": "9216cc18292f1c683f4767dfb86ae89c14a72f7df050df08e3800ce1e305b85b", + "bindings_equal_checkout_replay": true, + "gleif_scope_leis": 3755, + "goal": "whole-source census: installed empty-store Company proof", + "minutes": 33.9, + "only": null, + "policy": "34f548776040bb44451f0ac17e496ae30e1ce20568ad199bfa03616a6645b2e2", + "recovery": { + "item_at_kill": { + "attempt": "7049568c-80ae-43cc-997c-320e3f76b433", + "attempts": 1, + "checks": {}, + "error": null, + "ordinal": 0, + "receipt": null, + "state": "running", + "step": "merge", + "unit_key": "sec-first-chunk1" + }, + "killed_worker_exit": -9, + "resumed_attempts": 2, + "rows_at_kill": {} + }, + "second_pass_changed_nothing": true +} diff --git a/.planning/workstreams/whole-source-census/evidence/installed-company-proof-final-20261010.log b/.planning/workstreams/whole-source-census/evidence/installed-company-proof-final-20261010.log new file mode 100644 index 000000000..c5e77aa49 --- /dev/null +++ b/.planning/workstreams/whole-source-census/evidence/installed-company-proof-final-20261010.log @@ -0,0 +1,12 @@ +{"at": "13:23:25", "stage": "provisioned", "policy": "34f548776040bb44451f0ac17e496ae30e1ce20568ad199bfa03616a6645b2e2"} +{"at": "13:23:37", "stage": "worker killed", "killed_worker_exit": -9, "item_at_kill": {"attempt": "7049568c-80ae-43cc-997c-320e3f76b433", "attempts": 1, "checks": {}, "error": null, "ordinal": 0, "receipt": null, "state": "running", "step": "merge", "unit_key": "sec-first-chunk1"}, "rows_at_kill": {}} +{"at": "13:26:52", "run": "c8911413-ea08-418f-a45c-cde8c980d00a", "step": "mdm.merge", "verified": true, "minutes": 3.5} +{"at": "13:52:33", "run": "135d5acb-52b9-477e-af8e-ac516343ec84", "step": "source.read", "verified": true, "minutes": 29.1} +{"at": "13:52:39", "run": "135d5acb-52b9-477e-af8e-ac516343ec84", "step": "mdm.prepare", "verified": true, "minutes": 29.2} +{"at": "13:54:33", "run": "135d5acb-52b9-477e-af8e-ac516343ec84", "step": "mdm.merge", "verified": true, "minutes": 31.1} +{"at": "13:54:34", "stage": "first pass", "companies_open": 6414, "stage_records": {"gleif.level1.v1:company": 3716, "sec.submissions.company.v1:company": 6414}, "stage_bound": {"gleif.level1.v1": 3052, "sec.submissions.company.v1": 6414}, "open_reviews": {"binding_required": 664, "classification_deferred": 312, "unsupported_identity_kind": 39}, "decisions_by_rule": {"company-cik": 6414, "sec-gleif-name-postal": 191, "sec-gleif-name-jurisdiction": 2861}} +{"at": "13:56:31", "run": "74e2ee56-5fe0-459c-9d91-4e85effce97b", "step": "mdm.merge", "verified": true, "minutes": 33.1} +{"at": "13:56:37", "run": "5695fd2d-f65d-4b62-a1c9-953b4e58ea13", "step": "mdm.prepare", "verified": true, "minutes": 33.2} +{"at": "13:57:17", "run": "5695fd2d-f65d-4b62-a1c9-953b4e58ea13", "step": "mdm.merge", "verified": true, "minutes": 33.9} +{"at": "13:57:19", "stage": "report", "goal": "whole-source census: installed empty-store Company proof", "only": null, "policy": "34f548776040bb44451f0ac17e496ae30e1ce20568ad199bfa03616a6645b2e2", "gleif_scope_leis": 3755, "after_first_pass": {"companies_open": 6414, "stage_records": {"gleif.level1.v1:company": 3716, "sec.submissions.company.v1:company": 6414}, "stage_bound": {"gleif.level1.v1": 3052, "sec.submissions.company.v1": 6414}, "open_reviews": {"binding_required": 664, "classification_deferred": 312, "unsupported_identity_kind": 39}, "decisions_by_rule": {"company-cik": 6414, "sec-gleif-name-postal": 191, "sec-gleif-name-jurisdiction": 2861}}, "bindings": "9216cc18292f1c683f4767dfb86ae89c14a72f7df050df08e3800ce1e305b85b", "second_pass_changed_nothing": true, "recovery": {"killed_worker_exit": -9, "item_at_kill": {"attempt": "7049568c-80ae-43cc-997c-320e3f76b433", "attempts": 1, "checks": {}, "error": null, "ordinal": 0, "receipt": null, "state": "running", "step": "merge", "unit_key": "sec-first-chunk1"}, "rows_at_kill": {}, "resumed_attempts": 2}, "bindings_equal_checkout_replay": true, "minutes": 33.9} +1 passed in 2210.28s (0:36:50) diff --git a/.planning/workstreams/whole-source-census/evidence/installed-sec-recovery-after-commit-20261010.json b/.planning/workstreams/whole-source-census/evidence/installed-sec-recovery-after-commit-20261010.json new file mode 100644 index 000000000..34f813edd --- /dev/null +++ b/.planning/workstreams/whole-source-census/evidence/installed-sec-recovery-after-commit-20261010.json @@ -0,0 +1,43 @@ +{ + "after_first_pass": { + "companies_open": 6414, + "decisions_by_rule": { + "company-cik": 6414 + }, + "open_reviews": { + "classification_deferred": 312 + }, + "stage_bound": { + "sec.submissions.company.v1": 6414 + }, + "stage_records": { + "sec.submissions.company.v1:company": 6414 + } + }, + "bindings": "67749419118e9df5b13b0ad4fc277c9612ec07f8e1434e7078cc6cb4a1270af8", + "bindings_equal_checkout_replay": null, + "gleif_scope_leis": 0, + "goal": "whole-source census: installed empty-store Company proof", + "minutes": 5.0, + "only": "sec", + "policy": "34f548776040bb44451f0ac17e496ae30e1ce20568ad199bfa03616a6645b2e2", + "recovery": { + "item_at_kill": { + "attempt": "1fa91227-75bd-481c-b8f5-17f4a62aba18", + "attempts": 1, + "checks": {}, + "error": null, + "ordinal": 0, + "receipt": null, + "state": "running", + "step": "merge", + "unit_key": "sec-first-chunk1" + }, + "killed_worker_exit": -9, + "resumed_attempts": 2, + "rows_at_kill": { + "sec.submissions.company.v1:company": 919 + } + }, + "second_pass_changed_nothing": true +} diff --git a/.planning/workstreams/whole-source-census/installed_proof.py b/.planning/workstreams/whole-source-census/installed_proof.py new file mode 100644 index 000000000..1fda9df34 --- /dev/null +++ b/.planning/workstreams/whole-source-census/installed_proof.py @@ -0,0 +1,296 @@ +"""Whole-source census goal: the installed empty-store Company proof. + +One empty PostgreSQL 16 MDM store, loaded only by the installed bundle's +workers, each step checked by a verifier in a separate process with its own +logins; the checkout is never imported by either. The harness (this file) +only provisions: it creates the stores, registers the Company policy and the +two Dataset Contracts, and saves, proves and switches on the proof pipeline in +the disposable Rules database, as the installed-bundle acceptance test (G3) +does. + +Inputs, on this machine only (no provider request): +- SEC: the seven prepared Company bundles of company mastering ticket 27 + (`/bundles/chunk1..7`, version-2 manifests, 6,414 Companies and 586 + controls), each a unit of the `mdm.merge` step; +- GLEIF: the Golden Copy of 2026-09-11 16:00 (Level 1, 927,550,946 bytes, + 3,428,477 records), read by `rules/sources/gleif/level1-json.yaml` with its + approved scope filled with the 3,755 LEIs the Name Frequency names for the + cohort, then `mdm.prepare` and `mdm.merge`. + +Passes: the SEC target with a recovery (its first worker is killed mid-run and +a later worker finishes the same run), then the GLEIF target, then both again +as new runs: the SEC bundles again, and the GLEIF reading the first pass +verified, prepared and merged again under its own consumer. Expected: 6,414 Companies, 3,052 GLEIF Level 1 records bound, the +second pass changes nothing, and the bindings equal the checkout replay's +(`/private/tmp/counts-pin-20261009.json`, 9216cc18…). + + PG16_SERVER=pgserver COMPANY_PROOF_OUT= PATH="$HOME/.cargo/bin:$PATH" \\ + uv run --extra mdm --with pgserver pytest -q -s -p no:randomly \\ + .planning/workstreams/whole-source-census/installed_proof.py + +Optional: COMPANY_PROOF_WORK (default ~/.local/share/edgartools/clean-mdm/proving/cm27), +COMPANY_PROOF_ONLY=sec (the SEC target alone, to check the plumbing). +""" + +from __future__ import annotations + +import hashlib +import json +import os +import shutil +import signal +import subprocess +import time +from pathlib import Path + +from sqlalchemy import create_engine, text + +from edgar_warehouse.bookkeeping.clean.artifacts import Artifacts +from edgar_warehouse.bookkeeping.clean.database import grant_profile +from edgar_warehouse.bookkeeping.clean.destinations import migrate_guard +from edgar_warehouse.mdm.clean.store import digest, migrate, register_policy +from edgar_warehouse.rules import files +from tests.engine.test_data_skill_bundle_postgres import _run, _stores, installed # noqa: F401 +from tests.integration.test_configured_bookkeeping_postgres import databases # noqa: F401 +from tests.support.rules_approval import approve +from tests.support.rules_authority import register_dataset + +WORK = Path(os.environ.get("COMPANY_PROOF_WORK", "~/.local/share/edgartools/clean-mdm/proving/cm27")).expanduser() +GOLDEN = Path("~/.local/share/edgartools/clean-mdm/research/gleif-20260911-1600/" + "01-20260911-1600-gleif-goldencopy-lei2-golden-copy.json.zip").expanduser() +GOLDEN_SHA256 = "1b6cd9cda3f94269fd406ee481842ea042b699e95eb5b8124b1496d4fda36a6a" +PUBLICATION_COUNT = 3428477 +SEC, GLEIF = "sec.submissions.company.v1", "gleif.level1.v1" +CHECKOUT_BINDINGS = "9216cc18292f1c683f4767dfb86ae89c14a72f7df050df08e3800ce1e305b85b" +AS_OF = "2026-09-29T15:00:00+00:00" +LEASE_SECONDS = 30 +TARGET = lambda steps: {"lease_seconds": LEASE_SECONDS, "heartbeat_seconds": 10, # noqa: E731 + "retry": {"attempts": 5, "base_ms": 100, "cap_ms": 5000}, "allow_zero_work": False, + "steps": steps, "checks": ["manifest.hash", "work.accounting", "journal.delivered"]} +MERGE = {"name": "merge", "operation": "mdm.merge", "requires": [], "key": "{batch_id}", + "leases": ["mdm:consumer:{consumer}"], "checks": ["input.hash", "output.receipt", "mdm.committed"]} +PIPELINE = {"pipeline": "installed-company-proof", "bookkeeping": {"version": 1, "targets": { + "sec": TARGET([MERGE]), + "gleif": TARGET([ + {"name": "read", "operation": "source.read", "requires": [], "key": "{batch_id}", + "leases": ["source-output:{batch_id}"], "checks": ["input.hash", "output.receipt", "source.output"]}, + {"name": "prepare", "operation": "mdm.prepare", "requires": ["read"], "key": "{batch_id}", + "leases": ["mdm-prepare:{batch_id}"], "checks": ["input.hash", "output.receipt", "mdm.prepared"]}, + {**MERGE, "requires": ["prepare"]}]), + # The second GLEIF pass: the same verified reading prepared and merged again. + "gleif-replay": TARGET([ + {"name": "prepare", "operation": "mdm.prepare", "requires": [], "key": "{batch_id}", + "leases": ["mdm-prepare:{batch_id}"], "checks": ["input.hash", "output.receipt", "mdm.prepared"]}, + {**MERGE, "requires": ["prepare"]}])}}} + + +def counts(conn) -> dict: + one = lambda sql: conn.execute(text(sql)).scalar() # noqa: E731 + group = lambda sql: {str(k): v for k, v in conn.execute(text(sql)).all()} # noqa: E731 + return { + "companies_open": one("SELECT count(*) FROM mdm.current_entity WHERE kind='company' AND status<>'alias'"), + "stage_records": group("SELECT source_code||':'||kind, count(*) FROM mdm.stage_record GROUP BY 1"), + "stage_bound": group("SELECT source_code, count(*) FROM mdm.stage_record WHERE entity_id IS NOT NULL GROUP BY 1"), + "open_reviews": group("SELECT coalesce(body->>'reason','(none)'), count(*) FROM mdm.current_record " + "WHERE object_type='review' AND body->'open'='true'::jsonb GROUP BY 1"), + "decisions_by_rule": group("SELECT coalesce(body->>'rule_id', body->'rule'->>'rule_id', '(none)'), count(*) " + "FROM mdm.decision GROUP BY 1"), + } + + +def bindings(conn) -> str: + """Which record is bound to which master, as one digest (masters renumber per run).""" + groups: dict[str, list[str]] = {} + for entity, subject in conn.execute(text( + "SELECT entity_id::text, subject FROM mdm.stage_record WHERE entity_id IS NOT NULL")): + groups.setdefault(entity, []).append(subject) + return digest(sorted(sorted(g) for g in groups.values())) + + +def say(**fields) -> None: + print(json.dumps({"at": time.strftime("%H:%M:%S"), **fields}), flush=True) + + +def test_installed_company_population(installed, databases, tmp_path): # noqa: F811 + python, root = installed + store = Artifacts() + out = Path(os.environ["COMPANY_PROOF_OUT"]) + only = os.environ.get("COMPANY_PROOF_ONLY") + started = time.monotonic() + + # Provisioning (harness): an empty MDM store with its fence, the policy and the two contracts. + for profile in ("source.read", "mdm.prepare", "mdm.merge"): + grant_profile(databases.admin, profile=profile, worker="bk_runtime", verifier="bk_verifier") + with databases.admin.connect().execution_options(isolation_level="AUTOCOMMIT") as conn: + conn.exec_driver_sql("CREATE DATABASE mdm_proof") + mdm_admin = create_engine(databases.admin.url.set(database="mdm_proof")) + mdm_app = create_engine(mdm_admin.url.set(username="clean_application", password="test")) + migrate(mdm_admin, application_role="clean_application") + migrate_guard(mdm_admin, runtime_role="clean_application") + with mdm_admin.begin() as conn: + policy = register_policy(conn, files.policy()) + register_dataset(conn, SEC, "1", files.mdm_contract("sec.submissions.company", SEC)) + register_dataset(conn, GLEIF, "1", files.mdm_contract("gleif", GLEIF)) + conn.exec_driver_sql("GRANT USAGE ON SCHEMA mdm TO bk_verifier") + conn.exec_driver_sql("GRANT SELECT ON ALL TABLES IN SCHEMA mdm TO bk_verifier") + assert conn.execute(text("SELECT count(*) FROM mdm.stage_record")).scalar() == 0 + mdm_reader = create_engine(mdm_admin.url.set(username="bk_verifier", password="test")) + say(stage="provisioned", policy=policy) + + # SEC: each bundle a copy naming today's policy (a manifest reads only files inside its folder). + def sec_units(tag: str) -> dict: + merge = [] + for n in range(1, 8): + source, target = WORK / "bundles" / f"chunk{n}", tmp_path / tag / f"chunk{n}" + target.mkdir(parents=True) + for item in source.iterdir(): + if item.name != "manifest.json": + shutil.copy2(item, target / item.name) + manifest = json.loads((source / "manifest.json").read_text()) + manifest["policy_digest"] = policy + # Each pass's batches are its own, as a fresh capture's would be: a repeat of the + # first pass's batch ids would be skipped as already merged, never mastered again. + for batch in manifest["batches"]: + batch["batch_id"], batch["consumer"] = f"{batch['batch_id']}:{tag}", f"{batch['consumer']}:{tag}" + ref = store.put_bytes((target / "manifest.json").as_uri(), json.dumps(manifest).encode()) + merge.append({"keys": {"batch_id": f"{tag}-chunk{n}", "consumer": "sec.submissions.company.v1"}, + "input": ref, "output": (tmp_path / tag / f"merged-chunk{n}.json").as_uri(), "cursor": {}}) + return store.put(tmp_path.as_uri(), {"version": 2, "steps": {"merge": merge}}) + + # GLEIF: the Level 1 reading with its approved scope filled; one artifact, the whole archive. + def gleif_units(tag: str) -> tuple[dict, int]: + (cache,) = WORK.glob("gleif-rows-*.jsonl") + leis = sorted({json.loads(line)[1]["LEI"]["$"] for line in cache.open()}) + contract = files.load(files.ROOT / "sources/gleif/level1-json.yaml") + contract["read"]["references"]["approved_scope"] = {lei: {"selected": True} for lei in leis} + contract_ref = store.put(tmp_path.as_uri(), contract) + raw = {"uri": GOLDEN.as_uri(), "sha256": GOLDEN_SHA256} + context = store.put(tmp_path.as_uri(), {"version": 1, "input": raw, + "values": {"publication_count": PUBLICATION_COUNT}}) + reading = store.put(tmp_path.as_uri(), {"version": 2, "contract": contract_ref, + "artifacts": [{"input": raw, "context": context}]}) + # Each unit's consumer is its own, from checkpoint 0 (`mdm_prepare.py`): the second + # pass re-applies the same records as a new consumer, never rewinds the first one's. + keys = {"batch_id": f"{tag}-level1", "consumer": f"gleif.level1.v1:{tag}"} + base = tmp_path / tag + return store.put(tmp_path.as_uri(), {"version": 2, "steps": { + "read": [{"keys": keys, "input": reading, "output": (base / "reading.json").as_uri(), "cursor": {}}], + "prepare": [{"keys": {**keys, "table": "level1", "dataset": GLEIF, "record_column": "record", + "policy": policy, "as_of": AS_OF}, + "input": {"from": {"step": "read", "key": keys["batch_id"]}}, + "output": (base / "mdm" / "manifest.json").as_uri(), "cursor": {}}], + "merge": [{"keys": keys, "input": {"from": {"step": "prepare", "key": keys["batch_id"]}}, + "output": (base / "merged.json").as_uri(), "cursor": {}}]}}), len(leis) + + def replay_units(tag: str, first: str) -> dict: + """The first pass's verified Golden Copy reading, prepared and merged again under its own consumer. + Reading the 13 GB archive again would only repeat what `source.read`'s verifier recomputed.""" + reading = tmp_path / first / "reading.json" + ref = {"uri": reading.as_uri(), "sha256": hashlib.sha256(reading.read_bytes()).hexdigest()} + keys = {"batch_id": f"{tag}-level1", "consumer": f"gleif.level1.v1:{tag}"} + base = tmp_path / tag + return store.put(tmp_path.as_uri(), {"version": 2, "steps": { + "prepare": [{"keys": {**keys, "table": "level1", "dataset": GLEIF, "record_column": "record", + "policy": policy, "as_of": AS_OF}, + "input": ref, "output": (base / "mdm" / "manifest.json").as_uri(), "cursor": {}}], + "merge": [{"keys": keys, "input": {"from": {"step": "prepare", "key": keys["batch_id"]}}, + "output": (base / "merged.json").as_uri(), "cursor": {}}]}}) + + sec_first = sec_units("sec-first") + gleif_first, scope = gleif_units("gleif-first") if only != "sec" else (None, 0) + saved = databases.rules.save("pipeline", PIPELINE["pipeline"], "1", PIPELINE) + databases.rules.prove("pipeline", PIPELINE["pipeline"], "1", + {"digest": saved["digest"], "batch_hash": sec_first["sha256"], "passed": True}) + approve(databases.approver, "pipeline", PIPELINE["pipeline"], "1") + databases.rules.activate("pipeline", PIPELINE["pipeline"], "1") + + url = lambda engine: engine.url.render_as_string(hide_password=False) # noqa: E731 + env = {**_stores(databases), "BOOKKEEPING_MANIFEST_ROOT": (tmp_path / "control").as_uri()} + worker = {**env, "MDM_DATABASE_URL": url(mdm_app), "MDM_APPLICATION_ROLE": "clean_application"} + verifier = {**env, "BOOKKEEPING_CLEAN_DATABASE_URL": url(databases.verifier), "MDM_DATABASE_URL": url(mdm_reader)} + entry = ("-c", "import sys; from edgar_warehouse.cli import main; sys.exit(main(sys.argv[1:]))") + + def cli(*args, env=env): + done = _run(python, "-m", "edgar_warehouse.cli", *args, env=env, cwd=root) + assert done.returncode == 0, done.stderr[-4000:] + return done.stdout + + def submit(target: str, units: dict) -> str: + return json.loads(cli("rules", "run", "--pipeline", PIPELINE["pipeline"], "--target", target, + "--input-manifest", units["uri"], "--input-sha256", units["sha256"]))["run"]["run_id"] + + def finish(run_id: str, profiles: tuple[str, ...]) -> None: + """Work and verify each step until its units are verified; then finalize.""" + for profile in profiles: + deadline = time.monotonic() + 4 * 3600 + while True: + cli("workers", "work", profile, run_id, "--limit", "100", env=worker) + cli("workers", "verify", profile, run_id, "--reports", (tmp_path / "reports").as_uri(), + "--limit", "100", env=verifier) + status = json.loads(cli("bookkeeping", "status", run_id)) + mine = [i for i in status["items"] if i["step"] == profile.split(".")[-1] or i.get("operation") == profile] + if mine and all(i["state"] == "verified" for i in mine): + break + assert time.monotonic() < deadline, status + time.sleep(5) + say(run=run_id, step=profile, verified=True, minutes=round((time.monotonic() - started) / 60, 1)) + cli("bookkeeping", "finalize", run_id) + assert json.loads(cli("bookkeeping", "status", run_id))["run"]["state"] == "complete" + + def snapshot() -> tuple[dict, str]: + with mdm_reader.connect() as conn: + return counts(conn), bindings(conn) + + # First SEC pass, with a recovery: the first worker is killed while it holds an item. + # Where the kill lands inside the item (before or after its one MDM transaction commits) + # is timing; either way the item must be verified on its next attempt. + run_id = submit("sec", sec_first) + killed = subprocess.Popen([str(python), "-I", *entry, "workers", "work", "mdm.merge", run_id, "--limit", "100"], + env={**os.environ, **worker}, cwd=root, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + deadline = time.monotonic() + 600 + held = None + while killed.poll() is None and time.monotonic() < deadline: + items = json.loads(cli("bookkeeping", "status", run_id))["items"] + held = next((i for i in items if i["state"] == "running"), None) + if held: + killed.send_signal(signal.SIGKILL) + break + time.sleep(0.2) + killed.wait() + recovery = {"killed_worker_exit": killed.returncode, "item_at_kill": held, + "rows_at_kill": snapshot()[0]["stage_records"]} + say(stage="worker killed", **recovery) + assert killed.returncode == -signal.SIGKILL, "the worker finished before it was killed" + time.sleep(LEASE_SECONDS + 5) # its lease lapses; the run resumes under a new worker + finish(run_id, ("mdm.merge",)) + resumed = next(i for i in json.loads(cli("bookkeeping", "status", run_id))["items"] + if i["unit_key"] == held["unit_key"]) + recovery["resumed_attempts"] = resumed["attempts"] + assert resumed["state"] == "verified" and resumed["attempts"] == held["attempts"] + 1, resumed + if only != "sec": + finish(submit("gleif", gleif_first), ("source.read", "mdm.prepare", "mdm.merge")) + first, first_bindings = snapshot() + say(stage="first pass", **first) + + # Second pass: new runs over the same inputs. + finish(submit("sec", sec_units("sec-second")), ("mdm.merge",)) + if only != "sec": + finish(submit("gleif-replay", replay_units("gleif-second", "gleif-first")), ("mdm.prepare", "mdm.merge")) + second, second_bindings = snapshot() + + report = { + "goal": "whole-source census: installed empty-store Company proof", + "only": only, "policy": policy, "gleif_scope_leis": scope, + "after_first_pass": first, "bindings": first_bindings, + "second_pass_changed_nothing": first == second and first_bindings == second_bindings, + "recovery": recovery, + "bindings_equal_checkout_replay": first_bindings == CHECKOUT_BINDINGS if only != "sec" else None, + "minutes": round((time.monotonic() - started) / 60, 1), + } + out.write_text(json.dumps(report, sort_keys=True, indent=1) + "\n") + say(stage="report", **report) + assert report["second_pass_changed_nothing"] + assert first["stage_bound"].get(SEC) == 6414 + assert report["bindings_equal_checkout_replay"] in (True, None) + if only != "sec": + assert first["companies_open"] == 6414 and first["stage_bound"].get(GLEIF) == 3052 diff --git a/tests/integration/test_clean_mdm_postgres.py b/tests/integration/test_clean_mdm_postgres.py index dc24266a2..4c9612ca4 100644 --- a/tests/integration/test_clean_mdm_postgres.py +++ b/tests/integration/test_clean_mdm_postgres.py @@ -4,7 +4,6 @@ import json import subprocess -import time from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass from pathlib import Path @@ -20,6 +19,7 @@ migrate, register_policy, ) +from tests.support import pg16 from tests.support.rules_authority import register_dataset IMAGE = "postgres:16-alpine" @@ -41,53 +41,25 @@ class Database: @pytest.fixture(scope="module") def postgres(): - docker("image", "inspect", IMAGE) - name = f"clean-mdm-test-{uuid4().hex[:10]}" - docker( - "run", - "-d", - "--rm", - "--name", - name, - "-p", - "127.0.0.1::5432", - "-e", - "POSTGRES_PASSWORD=test", - IMAGE, - ) - port = None - try: - port = docker("port", name, "5432/tcp").rsplit(":", 1)[1] - admin = create_engine( - f"postgresql+psycopg2://postgres:test@127.0.0.1:{port}/postgres" - ) - deadline = time.monotonic() + 30 - while time.monotonic() < deadline: - try: - with admin.connect() as conn: - conn.execute(text("SELECT 1")) - break - except DBAPIError: - time.sleep(0.1) - else: - pytest.fail("PostgreSQL did not become ready") - with admin.begin() as conn: - conn.execute( - text( - "CREATE ROLE clean_application LOGIN PASSWORD 'test' NOSUPERUSER NOCREATEDB NOCREATEROLE" + # Docker in CI; PG16_SERVER=pgserver runs on a local PostgreSQL 16 (tests/support/pg16.py). + with pg16.server() as server: + admin = create_engine(server.url()) + app = None + try: + with admin.begin() as conn: + conn.execute( + text( + "CREATE ROLE clean_application LOGIN PASSWORD 'test' NOSUPERUSER NOCREATEDB NOCREATEROLE" + ) ) - ) - app = create_engine( - f"postgresql+psycopg2://clean_application:test@127.0.0.1:{port}/postgres" - ) - yield admin, app - app.dispose() - admin.dispose() - finally: - docker("stop", name) - # The next module's container may be given the same port. - if port: - _templates.pop(int(port), None) + app = create_engine(server.url(user="clean_application")) + yield admin, app + finally: + if app is not None: + app.dispose() + admin.dispose() + # The next module's server may be given the same port. + _templates.pop(admin.url.port, None) # Each module's first test migrates a template database once (checking a diff --git a/tests/integration/test_configured_bookkeeping_postgres.py b/tests/integration/test_configured_bookkeeping_postgres.py index e50ed2c01..ea2f2d9d1 100644 --- a/tests/integration/test_configured_bookkeeping_postgres.py +++ b/tests/integration/test_configured_bookkeeping_postgres.py @@ -9,7 +9,6 @@ from dataclasses import dataclass from pathlib import Path import json -import shutil import subprocess import time from uuid import uuid4 @@ -27,6 +26,7 @@ from edgar_warehouse.bookkeeping.clean.engine import Bookkeeping from edgar_warehouse.rules.db import Rules, migrate as migrate_rules from edgar_warehouse.workers import copy +from tests.support import pg16 from tests.support.bookkeeping_protocol import RUNTIME, complete, drive, verification_for, verification_report from tests.support.rules_approval import approve @@ -50,75 +50,52 @@ class Databases: @pytest.fixture(scope="module") def databases(): - assert shutil.which("docker"), "PG16 acceptance requires Docker (Colima on macOS)" - name = f"bookkeeping-acceptance-{uuid4().hex[:10]}" - subprocess.run(["docker", "image", "inspect", "postgres:16-alpine"], capture_output=True, check=True) - subprocess.run(["docker", "run", "--rm", "-d", "--name", name, "-p", "127.0.0.1::5432", - "-e", "POSTGRES_PASSWORD=test", "postgres:16-alpine"], capture_output=True, check=True) - engines = [] - try: - for _ in range(80): - ready = subprocess.run(["docker", "exec", name, "pg_isready", "-U", "postgres"], capture_output=True) - if ready.returncode == 0: - break - time.sleep(0.1) - else: - pytest.fail("PostgreSQL 16 failed to start") - port = subprocess.run(["docker", "port", name, "5432/tcp"], capture_output=True, text=True, check=True).stdout.strip().rsplit(":", 1)[1] - - def engine(db, user="postgres"): - value = create_engine(f"postgresql://{user}:test@127.0.0.1:{port}/{db}", connect_args={"connect_timeout": 5}) - engines.append(value) - return value - - setup = engine("postgres") - deadline = time.monotonic() + 30 - while time.monotonic() < deadline: - try: - with setup.connect() as conn: - conn.execute(text("SELECT 1")) - break - except DBAPIError: - time.sleep(0.1) - else: - pytest.fail("PostgreSQL host connection did not become ready") - with setup.connect().execution_options(isolation_level="AUTOCOMMIT") as conn: - for role in ("bk_runtime", "bk_verifier", "rules_agent", "operator", "ledger_runtime", "destination_runtime", "clean_application"): - conn.exec_driver_sql(f"CREATE ROLE {role} LOGIN PASSWORD 'test' NOSUPERUSER NOCREATEDB NOCREATEROLE") - # Control's grants go to one group; the worker (bk_runtime) and the - # verifier (bk_verifier) log in separately (mastering to-do 20b). - conn.exec_driver_sql("CREATE ROLE bk_control NOLOGIN") - conn.exec_driver_sql("GRANT bk_control TO bk_runtime, bk_verifier") - conn.exec_driver_sql("CREATE ROLE rules_approver NOLOGIN") - conn.exec_driver_sql("GRANT rules_approver TO operator") - for db in ("bookkeeping_clean", "rules", "change_journal_clean", "destination"): - conn.exec_driver_sql(f"CREATE DATABASE {db}") - admin = engine("bookkeeping_clean") - runtime = engine("bookkeeping_clean", "bk_runtime") - rules_admin = engine("rules") - ledger_admin = engine("change_journal_clean") - destination_admin = engine("destination") - with pytest.raises(Blocked, match="not initialized"): - migrate(admin, runtime_role="bk_control", existing_only=True) - with pytest.raises(Blocked, match="not initialized"): - migrate_journal(ledger_admin, runtime_role="ledger_runtime", existing_only=True) - migrate(admin, runtime_role="bk_control") - for profile in PROFILES: - grant_profile(admin, profile=profile, worker="bk_runtime", verifier="bk_verifier") - migrate_rules(rules_admin) - migrate_journal(ledger_admin, runtime_role="ledger_runtime") - migrate_guard(destination_admin, runtime_role="destination_runtime") - with destination_admin.begin() as conn: - conn.exec_driver_sql("CREATE TABLE effects(key text PRIMARY KEY, body jsonb NOT NULL)") - conn.exec_driver_sql("GRANT SELECT,INSERT ON effects TO destination_runtime") - migrate_guard(destination_admin, runtime_role="clean_application") - yield Databases(admin, runtime, engine("bookkeeping_clean", "bk_verifier"), Rules(engine("rules", "rules_agent")), Rules(engine("rules", "operator")), - ChangeJournal(engine("change_journal_clean", "ledger_runtime")), ledger_admin, - engine("destination", "destination_runtime"), destination_admin, engine("destination", "clean_application")) - finally: - for value in engines: - value.dispose() - subprocess.run(["docker", "stop", "--time", "1", name], capture_output=True, check=True) + # Docker in CI; PG16_SERVER=pgserver runs on a local PostgreSQL 16 (tests/support/pg16.py). + with pg16.server() as server: + engines = [] + try: + def engine(db, user="postgres"): + value = create_engine(server.url(db, user, driver="postgresql"), connect_args={"connect_timeout": 5}) + engines.append(value) + return value + + setup = engine("postgres") + with setup.connect().execution_options(isolation_level="AUTOCOMMIT") as conn: + for role in ("bk_runtime", "bk_verifier", "rules_agent", "operator", "ledger_runtime", "destination_runtime", "clean_application"): + conn.exec_driver_sql(f"CREATE ROLE {role} LOGIN PASSWORD 'test' NOSUPERUSER NOCREATEDB NOCREATEROLE") + # Control's grants go to one group; the worker (bk_runtime) and the + # verifier (bk_verifier) log in separately (mastering to-do 20b). + conn.exec_driver_sql("CREATE ROLE bk_control NOLOGIN") + conn.exec_driver_sql("GRANT bk_control TO bk_runtime, bk_verifier") + conn.exec_driver_sql("CREATE ROLE rules_approver NOLOGIN") + conn.exec_driver_sql("GRANT rules_approver TO operator") + for db in ("bookkeeping_clean", "rules", "change_journal_clean", "destination"): + conn.exec_driver_sql(f"CREATE DATABASE {db}") + admin = engine("bookkeeping_clean") + runtime = engine("bookkeeping_clean", "bk_runtime") + rules_admin = engine("rules") + ledger_admin = engine("change_journal_clean") + destination_admin = engine("destination") + with pytest.raises(Blocked, match="not initialized"): + migrate(admin, runtime_role="bk_control", existing_only=True) + with pytest.raises(Blocked, match="not initialized"): + migrate_journal(ledger_admin, runtime_role="ledger_runtime", existing_only=True) + migrate(admin, runtime_role="bk_control") + for profile in PROFILES: + grant_profile(admin, profile=profile, worker="bk_runtime", verifier="bk_verifier") + migrate_rules(rules_admin) + migrate_journal(ledger_admin, runtime_role="ledger_runtime") + migrate_guard(destination_admin, runtime_role="destination_runtime") + with destination_admin.begin() as conn: + conn.exec_driver_sql("CREATE TABLE effects(key text PRIMARY KEY, body jsonb NOT NULL)") + conn.exec_driver_sql("GRANT SELECT,INSERT ON effects TO destination_runtime") + migrate_guard(destination_admin, runtime_role="clean_application") + yield Databases(admin, runtime, engine("bookkeeping_clean", "bk_verifier"), Rules(engine("rules", "rules_agent")), Rules(engine("rules", "operator")), + ChangeJournal(engine("change_journal_clean", "ledger_runtime")), ledger_admin, + engine("destination", "destination_runtime"), destination_admin, engine("destination", "clean_application")) + finally: + for value in engines: + value.dispose() def config(steps=1, *, seconds=120, zero=False, resource="output:{destination}"): @@ -949,25 +926,16 @@ def test_source_configs_use_same_control_contract(): def test_provisioning_uses_nologin_owners_and_empty_stores(monkeypatch): import runpy - name = f"bookkeeping-provision-{uuid4().hex[:10]}" - subprocess.run(["docker", "run", "--rm", "-d", "--name", name, "-p", "127.0.0.1::5432", - "-e", "POSTGRES_PASSWORD=test", "postgres:16-alpine"], capture_output=True, check=True) + with pg16.server() as server: + _provision(server, monkeypatch, runpy) + + +def _provision(server, monkeypatch, runpy): engines = [] try: - port = subprocess.run(["docker", "port", name, "5432/tcp"], capture_output=True, text=True, check=True).stdout.strip().rsplit(":", 1)[1] - url = f"postgresql://postgres:test@127.0.0.1:{port}/postgres" + url = server.url(driver="postgresql") admin = create_engine(url, connect_args={"connect_timeout": 3}) engines.append(admin) - deadline = time.monotonic() + 30 - while time.monotonic() < deadline: - try: - with admin.connect() as conn: - conn.execute(text("SELECT 1")) - break - except DBAPIError: - time.sleep(0.1) - else: - pytest.fail("Provisioning acceptance PostgreSQL unavailable") monkeypatch.setenv("BOOKKEEPING_CLEAN_RUNTIME_PASSWORD", "test") monkeypatch.setenv("RULES_AGENT_PASSWORD", "test") script = Path(__file__).parents[2] / "infra/scripts/provision-clean-bookkeeping.py" @@ -978,13 +946,13 @@ def test_provisioning_uses_nologin_owners_and_empty_stores(monkeypatch): assert provision(url, rules=True, journal=True) == result for database, role, table in (("bookkeeping_clean", "bookkeeping_clean_runtime", "bookkeeping.pipeline_run"), ("rules", "rules_agent", "rules.rule_version")): - runtime = create_engine(f"postgresql://{role}:test@127.0.0.1:{port}/{database}") + runtime = create_engine(server.url(database, role, driver="postgresql")) engines.append(runtime) with runtime.connect() as conn: assert conn.scalar(text(f"SELECT count(*) FROM {table}")) == 0 with pytest.raises(DBAPIError), runtime.begin() as conn: conn.execute(text(f"DELETE FROM {table}")) - journal_admin = create_engine(f"postgresql://postgres:test@127.0.0.1:{port}/change_journal_clean") + journal_admin = create_engine(server.url("change_journal_clean", driver="postgresql")) engines.append(journal_admin) with journal_admin.connect() as conn: assert conn.scalar(text("SELECT count(*) FROM journal.event")) == 0 @@ -994,4 +962,3 @@ def test_provisioning_uses_nologin_owners_and_empty_stores(monkeypatch): finally: for engine in engines: engine.dispose() - subprocess.run(["docker", "stop", "--time", "1", name], capture_output=True, check=True) diff --git a/tests/support/pg16.py b/tests/support/pg16.py new file mode 100644 index 000000000..ff9d83486 --- /dev/null +++ b/tests/support/pg16.py @@ -0,0 +1,89 @@ +"""An empty PostgreSQL 16 server for the Postgres acceptance tests. + +Docker by default (`postgres:16-alpine`, superuser `postgres` / `test`, a +loopback port), as CI runs them. With `PG16_SERVER=pgserver`, a local +PostgreSQL 16 from the `pgserver` package instead, on its own socket in a +temporary folder that is deleted afterwards: for a machine with no Docker +(`uv run --with pgserver ...`). Either way the test gets `url(database, user)`; +every role a test creates logs in with password `test`. The server is yielded +only once it answers a query. +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import tempfile +import time +import uuid +import warnings +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass + +IMAGE = "postgres:16-alpine" + + +@dataclass(frozen=True) +class Server: + """Where the server answers: a loopback port, or a socket folder.""" + port: str | None = None + socket: str | None = None + + def url(self, database: str = "postgres", user: str = "postgres", driver: str = "postgresql+psycopg2") -> str: + if self.socket is not None: + return f"{driver}://{user}:test@/{database}?host={self.socket}" + return f"{driver}://{user}:test@127.0.0.1:{self.port}/{database}" + + +def _docker(*args: str) -> str: + return subprocess.run(["docker", *args], text=True, capture_output=True, check=True).stdout.strip() + + +def _ready(found: Server, seconds: float = 30) -> Server: + """The server once it answers `SELECT 1`; the image restarts once after it initializes, + so a ready port alone is not enough.""" + from sqlalchemy import create_engine, text + from sqlalchemy.exc import DBAPIError + + engine = create_engine(found.url(driver="postgresql"), connect_args={"connect_timeout": 3}) + deadline = time.monotonic() + seconds + try: + while True: + try: + with engine.connect() as conn: + conn.execute(text("SELECT 1")) + return found + except DBAPIError: + if time.monotonic() > deadline: + raise RuntimeError("PostgreSQL 16 did not become ready") from None + time.sleep(0.1) + finally: + engine.dispose() + + +@contextmanager +def server() -> Iterator[Server]: + if os.environ.get("PG16_SERVER") == "pgserver": + import pgserver + + folder = tempfile.mkdtemp(prefix="pg16-") + try: + local = pgserver.get_server(folder, cleanup_mode="delete") + try: + yield _ready(Server(socket=local.get_uri().rsplit("host=", 1)[1])) + finally: + local.cleanup() + finally: + shutil.rmtree(folder, ignore_errors=True) + return + _docker("image", "inspect", IMAGE) + name = f"pg16-test-{uuid.uuid4().hex[:10]}" + _docker("run", "-d", "--rm", "--name", name, "-p", "127.0.0.1::5432", "-e", "POSTGRES_PASSWORD=test", IMAGE) + try: + yield _ready(Server(port=_docker("port", name, "5432/tcp").rsplit(":", 1)[1])) + finally: + stopped = subprocess.run(["docker", "stop", "--time", "1", name], capture_output=True, text=True) + if stopped.returncode: + warnings.warn(f"Test container {name} was not stopped: {stopped.stderr.strip()}", stacklevel=2)