From 540b1bc305e7979c4708e02d6420915a6bdbd5fd Mon Sep 17 00:00:00 2001 From: SinuS Von SifriduS Date: Mon, 21 Sep 2026 10:33:06 +0200 Subject: [PATCH] 0.3.16-pre.3.fix.1 --- Cargo.toml | 4 +- .../001_pk_ksp_raw_transaction_variants.sql | 12 + ..._ksp_raw_transaction_variants_identity.sql | 12 + ...ck_ksp_raw_transaction_variants_origin.sql | 12 + ...ksp_raw_transaction_variants_slot_time.sql | 12 + ...ck_ksp_raw_transaction_variants_format.sql | 12 + ...k_ksp_raw_transaction_variants_payload.sql | 12 + ...p_raw_transaction_variants_transaction.sql | 12 + ...sp_raw_transaction_canonical_selectors.sql | 12 + ...transaction_canonical_selectors_bounds.sql | 12 + ...ransaction_canonical_selectors_variant.sql | 12 + ...p_raw_transaction_observation_variants.sql | 12 + ...ransaction_observation_variants_bounds.sql | 12 + ...ction_observation_variants_observation.sql | 12 + ...ansaction_observation_variants_variant.sql | 12 + .../015_pk_ksp_raw_transaction_conflicts.sql | 12 + ...k_ksp_raw_transaction_conflicts_bounds.sql | 12 + ...ransaction_conflicts_canonical_variant.sql | 12 + ...transaction_conflicts_incoming_variant.sql | 12 + ...action_variants_signature_hash_variant.sql | 2 + ...ansaction_observation_variants_variant.sql | 2 + ...ion_conflicts_status_updated_signature.sql | 2 + .../001_ksp_raw_transaction_variants.sql | 24 + ...sp_raw_transaction_canonical_selectors.sql | 10 + ...p_raw_transaction_observation_variants.sql | 10 + .../004_ksp_raw_transaction_conflicts.sql | 20 + crates/ksp-store-postgres-lib/src/lib.rs | 66 +- .../src/migration_v003.rs | 525 ++++++++++++++ .../ksp-store-postgres-lib/src/schema_v003.rs | 666 ++++++++++++++++++ .../tests/v003_migration_completeness.rs | 56 ++ .../unit_tests/migration.rs | 4 +- .../unit_tests/migration_v003.rs | 87 +++ .../unit_tests/schema_v003.rs | 54 ++ deltas/0.3.16/pre.003-fix.001.md | 68 ++ deltas/0.3.16/pre.003.md | 75 ++ ...38-V0_3_16_RAW_RESILIENCE_CONFLICT_PLAN.md | 23 +- 36 files changed, 1852 insertions(+), 62 deletions(-) create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/001_pk_ksp_raw_transaction_variants.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/002_ck_ksp_raw_transaction_variants_identity.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/003_ck_ksp_raw_transaction_variants_origin.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/004_ck_ksp_raw_transaction_variants_slot_time.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/005_ck_ksp_raw_transaction_variants_format.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/006_ck_ksp_raw_transaction_variants_payload.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/007_fk_ksp_raw_transaction_variants_transaction.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/008_pk_ksp_raw_transaction_canonical_selectors.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/009_ck_ksp_raw_transaction_canonical_selectors_bounds.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/010_fk_ksp_raw_transaction_canonical_selectors_variant.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/011_pk_ksp_raw_transaction_observation_variants.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/012_ck_ksp_raw_transaction_observation_variants_bounds.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/013_fk_ksp_raw_transaction_observation_variants_observation.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/014_fk_ksp_raw_transaction_observation_variants_variant.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/015_pk_ksp_raw_transaction_conflicts.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/016_ck_ksp_raw_transaction_conflicts_bounds.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/017_fk_ksp_raw_transaction_conflicts_canonical_variant.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/018_fk_ksp_raw_transaction_conflicts_incoming_variant.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/001_ix_ksp_raw_transaction_variants_signature_hash_variant.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/002_ix_ksp_raw_transaction_observation_variants_variant.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/003_ix_ksp_raw_transaction_conflicts_status_updated_signature.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/001_ksp_raw_transaction_variants.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/002_ksp_raw_transaction_canonical_selectors.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/003_ksp_raw_transaction_observation_variants.sql create mode 100644 crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/004_ksp_raw_transaction_conflicts.sql create mode 100644 crates/ksp-store-postgres-lib/src/migration_v003.rs create mode 100644 crates/ksp-store-postgres-lib/src/schema_v003.rs create mode 100644 crates/ksp-store-postgres-lib/tests/v003_migration_completeness.rs create mode 100644 crates/ksp-store-postgres-lib/unit_tests/migration_v003.rs create mode 100644 crates/ksp-store-postgres-lib/unit_tests/schema_v003.rs create mode 100644 deltas/0.3.16/pre.003-fix.001.md create mode 100644 deltas/0.3.16/pre.003.md diff --git a/Cargo.toml b/Cargo.toml index e2c3cdc..eab6612 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,12 +1,12 @@ # file: Cargo.toml -# version: 640 +# version: 642 [workspace] resolver = "3" members = ["crates/ksp-app-backfill-desk", "crates/ksp-app-config-desk", "crates/ksp-app-raw-transaction-ingest-desk", "crates/ksp-app-solprices-desk", "crates/ksp-app-store-desk", "crates/ksp-app-wallet-desk", "crates/ksp-config-lib", "crates/ksp-core-lib", "crates/ksp-interface-lib", "crates/ksp-job-api", "crates/ksp-job-backfill-lib", "crates/ksp-logging-lib", "crates/ksp-offchain-transport-lib", "crates/ksp-onchain-transport-lib", "crates/ksp-program-api", "crates/ksp-raw-transaction-lib", "crates/ksp-store-api", "crates/ksp-store-lib", "crates/ksp-store-postgres-lib", "crates/ksp-wallet-lib", "crates/ksp-worker-api", "crates/ksp-worker-raw-transaction-ingest-lib"] [workspace.package] -version = "0.3.16-pre.2.fix.1" +version = "0.3.16-pre.3.fix.1" edition = "2024" license = "MIT" repository = "https://git.sasedev.com/Sasedev/khadhroony-solana-project" diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/001_pk_ksp_raw_transaction_variants.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/001_pk_ksp_raw_transaction_variants.sql new file mode 100644 index 0000000..589b7be --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/001_pk_ksp_raw_transaction_variants.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'pk_ksp_raw_transaction_variants' + AND conrelid = to_regclass('ksp_raw_transaction_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_variants ADD CONSTRAINT pk_ksp_raw_transaction_variants PRIMARY KEY (transaction_signature, variant_id); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/002_ck_ksp_raw_transaction_variants_identity.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/002_ck_ksp_raw_transaction_variants_identity.sql new file mode 100644 index 0000000..79ce17f --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/002_ck_ksp_raw_transaction_variants_identity.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'ck_ksp_raw_transaction_variants_identity' + AND conrelid = to_regclass('ksp_raw_transaction_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_variants ADD CONSTRAINT ck_ksp_raw_transaction_variants_identity CHECK (octet_length(transaction_signature) = 64 AND variant_id >= 1 AND variant_id <= 18446744073709551615); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/003_ck_ksp_raw_transaction_variants_origin.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/003_ck_ksp_raw_transaction_variants_origin.sql new file mode 100644 index 0000000..2186f80 --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/003_ck_ksp_raw_transaction_variants_origin.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'ck_ksp_raw_transaction_variants_origin' + AND conrelid = to_regclass('ksp_raw_transaction_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_variants ADD CONSTRAINT ck_ksp_raw_transaction_variants_origin CHECK (origin_kind = 'native' OR origin_kind = 'synthetic'); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/004_ck_ksp_raw_transaction_variants_slot_time.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/004_ck_ksp_raw_transaction_variants_slot_time.sql new file mode 100644 index 0000000..cd6699a --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/004_ck_ksp_raw_transaction_variants_slot_time.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'ck_ksp_raw_transaction_variants_slot_time' + AND conrelid = to_regclass('ksp_raw_transaction_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_variants ADD CONSTRAINT ck_ksp_raw_transaction_variants_slot_time CHECK (slot >= 0 AND slot <= 18446744073709551615 AND (block_time_unix_millis IS NULL OR (block_time_unix_millis >= 0 AND block_time_unix_millis <= 253402300799999)) AND created_at_unix_millis >= 0 AND created_at_unix_millis <= 253402300799999); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/005_ck_ksp_raw_transaction_variants_format.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/005_ck_ksp_raw_transaction_variants_format.sql new file mode 100644 index 0000000..632b5c5 --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/005_ck_ksp_raw_transaction_variants_format.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'ck_ksp_raw_transaction_variants_format' + AND conrelid = to_regclass('ksp_raw_transaction_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_variants ADD CONSTRAINT ck_ksp_raw_transaction_variants_format CHECK (char_length(format_id) >= 1 AND char_length(format_id) <= 128 AND format_version >= 1 AND format_version <= 4294967295 AND octet_length(content_hash) = 32); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/006_ck_ksp_raw_transaction_variants_payload.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/006_ck_ksp_raw_transaction_variants_payload.sql new file mode 100644 index 0000000..50fb4ba --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/006_ck_ksp_raw_transaction_variants_payload.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'ck_ksp_raw_transaction_variants_payload' + AND conrelid = to_regclass('ksp_raw_transaction_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_variants ADD CONSTRAINT ck_ksp_raw_transaction_variants_payload CHECK ((retention_state = 'full' AND payload IS NOT NULL AND octet_length(payload) >= 1 AND octet_length(payload) <= 16777216) OR ((retention_state = 'archived' OR retention_state = 'purged') AND payload IS NULL)); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/007_fk_ksp_raw_transaction_variants_transaction.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/007_fk_ksp_raw_transaction_variants_transaction.sql new file mode 100644 index 0000000..3fdb49f --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/007_fk_ksp_raw_transaction_variants_transaction.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'fk_ksp_raw_transaction_variants_transaction' + AND conrelid = to_regclass('ksp_raw_transaction_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_variants ADD CONSTRAINT fk_ksp_raw_transaction_variants_transaction FOREIGN KEY (transaction_signature) REFERENCES ksp_raw_transactions(signature) ON DELETE RESTRICT; + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/008_pk_ksp_raw_transaction_canonical_selectors.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/008_pk_ksp_raw_transaction_canonical_selectors.sql new file mode 100644 index 0000000..d95cf86 --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/008_pk_ksp_raw_transaction_canonical_selectors.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'pk_ksp_raw_transaction_canonical_selectors' + AND conrelid = to_regclass('ksp_raw_transaction_canonical_selectors') + ) THEN + ALTER TABLE ksp_raw_transaction_canonical_selectors ADD CONSTRAINT pk_ksp_raw_transaction_canonical_selectors PRIMARY KEY (transaction_signature); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/009_ck_ksp_raw_transaction_canonical_selectors_bounds.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/009_ck_ksp_raw_transaction_canonical_selectors_bounds.sql new file mode 100644 index 0000000..1ee50de --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/009_ck_ksp_raw_transaction_canonical_selectors_bounds.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'ck_ksp_raw_transaction_canonical_selectors_bounds' + AND conrelid = to_regclass('ksp_raw_transaction_canonical_selectors') + ) THEN + ALTER TABLE ksp_raw_transaction_canonical_selectors ADD CONSTRAINT ck_ksp_raw_transaction_canonical_selectors_bounds CHECK (octet_length(transaction_signature) = 64 AND canonical_variant_id >= 1 AND canonical_variant_id <= 18446744073709551615 AND canonical_revision >= 1 AND canonical_revision <= 18446744073709551615 AND updated_at_unix_millis >= 0 AND updated_at_unix_millis <= 253402300799999); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/010_fk_ksp_raw_transaction_canonical_selectors_variant.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/010_fk_ksp_raw_transaction_canonical_selectors_variant.sql new file mode 100644 index 0000000..3af30b2 --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/010_fk_ksp_raw_transaction_canonical_selectors_variant.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'fk_ksp_raw_transaction_canonical_selectors_variant' + AND conrelid = to_regclass('ksp_raw_transaction_canonical_selectors') + ) THEN + ALTER TABLE ksp_raw_transaction_canonical_selectors ADD CONSTRAINT fk_ksp_raw_transaction_canonical_selectors_variant FOREIGN KEY (transaction_signature, canonical_variant_id) REFERENCES ksp_raw_transaction_variants(transaction_signature, variant_id) ON DELETE RESTRICT; + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/011_pk_ksp_raw_transaction_observation_variants.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/011_pk_ksp_raw_transaction_observation_variants.sql new file mode 100644 index 0000000..673dec9 --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/011_pk_ksp_raw_transaction_observation_variants.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'pk_ksp_raw_transaction_observation_variants' + AND conrelid = to_regclass('ksp_raw_transaction_observation_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_observation_variants ADD CONSTRAINT pk_ksp_raw_transaction_observation_variants PRIMARY KEY (observation_key); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/012_ck_ksp_raw_transaction_observation_variants_bounds.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/012_ck_ksp_raw_transaction_observation_variants_bounds.sql new file mode 100644 index 0000000..0892770 --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/012_ck_ksp_raw_transaction_observation_variants_bounds.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'ck_ksp_raw_transaction_observation_variants_bounds' + AND conrelid = to_regclass('ksp_raw_transaction_observation_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_observation_variants ADD CONSTRAINT ck_ksp_raw_transaction_observation_variants_bounds CHECK (octet_length(observation_key) = 32 AND octet_length(transaction_signature) = 64 AND variant_id >= 1 AND variant_id <= 18446744073709551615 AND linked_at_unix_millis >= 0 AND linked_at_unix_millis <= 253402300799999); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/013_fk_ksp_raw_transaction_observation_variants_observation.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/013_fk_ksp_raw_transaction_observation_variants_observation.sql new file mode 100644 index 0000000..6727b8d --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/013_fk_ksp_raw_transaction_observation_variants_observation.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'fk_ksp_raw_transaction_observation_variants_observation' + AND conrelid = to_regclass('ksp_raw_transaction_observation_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_observation_variants ADD CONSTRAINT fk_ksp_raw_transaction_observation_variants_observation FOREIGN KEY (observation_key) REFERENCES ksp_raw_transaction_observations(observation_key) ON DELETE CASCADE; + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/014_fk_ksp_raw_transaction_observation_variants_variant.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/014_fk_ksp_raw_transaction_observation_variants_variant.sql new file mode 100644 index 0000000..3afe4cc --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/014_fk_ksp_raw_transaction_observation_variants_variant.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'fk_ksp_raw_transaction_observation_variants_variant' + AND conrelid = to_regclass('ksp_raw_transaction_observation_variants') + ) THEN + ALTER TABLE ksp_raw_transaction_observation_variants ADD CONSTRAINT fk_ksp_raw_transaction_observation_variants_variant FOREIGN KEY (transaction_signature, variant_id) REFERENCES ksp_raw_transaction_variants(transaction_signature, variant_id) ON DELETE RESTRICT; + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/015_pk_ksp_raw_transaction_conflicts.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/015_pk_ksp_raw_transaction_conflicts.sql new file mode 100644 index 0000000..1bfb404 --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/015_pk_ksp_raw_transaction_conflicts.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'pk_ksp_raw_transaction_conflicts' + AND conrelid = to_regclass('ksp_raw_transaction_conflicts') + ) THEN + ALTER TABLE ksp_raw_transaction_conflicts ADD CONSTRAINT pk_ksp_raw_transaction_conflicts PRIMARY KEY (transaction_signature); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/016_ck_ksp_raw_transaction_conflicts_bounds.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/016_ck_ksp_raw_transaction_conflicts_bounds.sql new file mode 100644 index 0000000..5b8e1ee --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/016_ck_ksp_raw_transaction_conflicts_bounds.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'ck_ksp_raw_transaction_conflicts_bounds' + AND conrelid = to_regclass('ksp_raw_transaction_conflicts') + ) THEN + ALTER TABLE ksp_raw_transaction_conflicts ADD CONSTRAINT ck_ksp_raw_transaction_conflicts_bounds CHECK (octet_length(transaction_signature) = 64 AND (status = 'open' OR status = 'resolved') AND revision >= 1 AND revision <= 18446744073709551615 AND canonical_variant_id >= 1 AND canonical_variant_id <= 18446744073709551615 AND incoming_variant_id >= 1 AND incoming_variant_id <= 18446744073709551615 AND canonical_variant_id <> incoming_variant_id AND (latest_relation = 'conflict' OR latest_relation = 'incomparable') AND ((latest_relation = 'conflict' AND (latest_reason_code = 'slot_mismatch' OR latest_reason_code = 'block_time_mismatch' OR latest_reason_code = 'canonical_payload_conflict' OR latest_reason_code = 'content_hash_collision')) OR (latest_relation = 'incomparable' AND (latest_reason_code = 'payload_format_mismatch' OR latest_reason_code = 'unsupported_canonical_difference'))) AND created_at_unix_millis >= 0 AND created_at_unix_millis <= 253402300799999 AND updated_at_unix_millis >= created_at_unix_millis AND updated_at_unix_millis <= 253402300799999); + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/017_fk_ksp_raw_transaction_conflicts_canonical_variant.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/017_fk_ksp_raw_transaction_conflicts_canonical_variant.sql new file mode 100644 index 0000000..ed514bc --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/017_fk_ksp_raw_transaction_conflicts_canonical_variant.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'fk_ksp_raw_transaction_conflicts_canonical_variant' + AND conrelid = to_regclass('ksp_raw_transaction_conflicts') + ) THEN + ALTER TABLE ksp_raw_transaction_conflicts ADD CONSTRAINT fk_ksp_raw_transaction_conflicts_canonical_variant FOREIGN KEY (transaction_signature, canonical_variant_id) REFERENCES ksp_raw_transaction_variants(transaction_signature, variant_id) ON DELETE RESTRICT; + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/018_fk_ksp_raw_transaction_conflicts_incoming_variant.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/018_fk_ksp_raw_transaction_conflicts_incoming_variant.sql new file mode 100644 index 0000000..f11822e --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/018_fk_ksp_raw_transaction_conflicts_incoming_variant.sql @@ -0,0 +1,12 @@ +DO $ksp$ +BEGIN + IF NOT EXISTS ( + SELECT 1 + FROM pg_constraint + WHERE conname = 'fk_ksp_raw_transaction_conflicts_incoming_variant' + AND conrelid = to_regclass('ksp_raw_transaction_conflicts') + ) THEN + ALTER TABLE ksp_raw_transaction_conflicts ADD CONSTRAINT fk_ksp_raw_transaction_conflicts_incoming_variant FOREIGN KEY (transaction_signature, incoming_variant_id) REFERENCES ksp_raw_transaction_variants(transaction_signature, variant_id) ON DELETE RESTRICT; + END IF; +END +$ksp$; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/001_ix_ksp_raw_transaction_variants_signature_hash_variant.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/001_ix_ksp_raw_transaction_variants_signature_hash_variant.sql new file mode 100644 index 0000000..1625eb5 --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/001_ix_ksp_raw_transaction_variants_signature_hash_variant.sql @@ -0,0 +1,2 @@ +CREATE INDEX IF NOT EXISTS ix_ksp_raw_transaction_variants_signature_hash_variant + ON ksp_raw_transaction_variants (transaction_signature, content_hash, variant_id); diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/002_ix_ksp_raw_transaction_observation_variants_variant.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/002_ix_ksp_raw_transaction_observation_variants_variant.sql new file mode 100644 index 0000000..3c0c3d3 --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/002_ix_ksp_raw_transaction_observation_variants_variant.sql @@ -0,0 +1,2 @@ +CREATE INDEX IF NOT EXISTS ix_ksp_raw_transaction_observation_variants_variant + ON ksp_raw_transaction_observation_variants (transaction_signature, variant_id, observation_key); diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/003_ix_ksp_raw_transaction_conflicts_status_updated_signature.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/003_ix_ksp_raw_transaction_conflicts_status_updated_signature.sql new file mode 100644 index 0000000..7bc9a9f --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/003_ix_ksp_raw_transaction_conflicts_status_updated_signature.sql @@ -0,0 +1,2 @@ +CREATE INDEX IF NOT EXISTS ix_ksp_raw_transaction_conflicts_status_updated_signature + ON ksp_raw_transaction_conflicts (status, updated_at_unix_millis, transaction_signature); diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/001_ksp_raw_transaction_variants.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/001_ksp_raw_transaction_variants.sql new file mode 100644 index 0000000..c2f6a7b --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/001_ksp_raw_transaction_variants.sql @@ -0,0 +1,24 @@ +CREATE TABLE IF NOT EXISTS ksp_raw_transaction_variants ( + transaction_signature BYTEA NOT NULL, + variant_id NUMERIC(20, 0) NOT NULL, + origin_kind TEXT NOT NULL, + slot NUMERIC(20, 0) NOT NULL, + block_time_unix_millis BIGINT NULL, + format_id TEXT NOT NULL, + format_version BIGINT NOT NULL, + content_hash BYTEA NOT NULL, + payload BYTEA NULL, + retention_state TEXT NOT NULL, + created_at_unix_millis BIGINT NOT NULL +); +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS transaction_signature BYTEA NOT NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS variant_id NUMERIC(20, 0) NOT NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS origin_kind TEXT NOT NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS slot NUMERIC(20, 0) NOT NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS block_time_unix_millis BIGINT NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS format_id TEXT NOT NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS format_version BIGINT NOT NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS content_hash BYTEA NOT NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS payload BYTEA NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS retention_state TEXT NOT NULL; +ALTER TABLE ksp_raw_transaction_variants ADD COLUMN IF NOT EXISTS created_at_unix_millis BIGINT NOT NULL; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/002_ksp_raw_transaction_canonical_selectors.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/002_ksp_raw_transaction_canonical_selectors.sql new file mode 100644 index 0000000..dead7df --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/002_ksp_raw_transaction_canonical_selectors.sql @@ -0,0 +1,10 @@ +CREATE TABLE IF NOT EXISTS ksp_raw_transaction_canonical_selectors ( + transaction_signature BYTEA NOT NULL, + canonical_variant_id NUMERIC(20, 0) NOT NULL, + canonical_revision NUMERIC(20, 0) NOT NULL, + updated_at_unix_millis BIGINT NOT NULL +); +ALTER TABLE ksp_raw_transaction_canonical_selectors ADD COLUMN IF NOT EXISTS transaction_signature BYTEA NOT NULL; +ALTER TABLE ksp_raw_transaction_canonical_selectors ADD COLUMN IF NOT EXISTS canonical_variant_id NUMERIC(20, 0) NOT NULL; +ALTER TABLE ksp_raw_transaction_canonical_selectors ADD COLUMN IF NOT EXISTS canonical_revision NUMERIC(20, 0) NOT NULL; +ALTER TABLE ksp_raw_transaction_canonical_selectors ADD COLUMN IF NOT EXISTS updated_at_unix_millis BIGINT NOT NULL; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/003_ksp_raw_transaction_observation_variants.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/003_ksp_raw_transaction_observation_variants.sql new file mode 100644 index 0000000..f3c7d8f --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/003_ksp_raw_transaction_observation_variants.sql @@ -0,0 +1,10 @@ +CREATE TABLE IF NOT EXISTS ksp_raw_transaction_observation_variants ( + observation_key BYTEA NOT NULL, + transaction_signature BYTEA NOT NULL, + variant_id NUMERIC(20, 0) NOT NULL, + linked_at_unix_millis BIGINT NOT NULL +); +ALTER TABLE ksp_raw_transaction_observation_variants ADD COLUMN IF NOT EXISTS observation_key BYTEA NOT NULL; +ALTER TABLE ksp_raw_transaction_observation_variants ADD COLUMN IF NOT EXISTS transaction_signature BYTEA NOT NULL; +ALTER TABLE ksp_raw_transaction_observation_variants ADD COLUMN IF NOT EXISTS variant_id NUMERIC(20, 0) NOT NULL; +ALTER TABLE ksp_raw_transaction_observation_variants ADD COLUMN IF NOT EXISTS linked_at_unix_millis BIGINT NOT NULL; diff --git a/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/004_ksp_raw_transaction_conflicts.sql b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/004_ksp_raw_transaction_conflicts.sql new file mode 100644 index 0000000..cdbca6b --- /dev/null +++ b/crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/004_ksp_raw_transaction_conflicts.sql @@ -0,0 +1,20 @@ +CREATE TABLE IF NOT EXISTS ksp_raw_transaction_conflicts ( + transaction_signature BYTEA NOT NULL, + status TEXT NOT NULL, + revision NUMERIC(20, 0) NOT NULL, + canonical_variant_id NUMERIC(20, 0) NOT NULL, + incoming_variant_id NUMERIC(20, 0) NOT NULL, + latest_relation TEXT NOT NULL, + latest_reason_code TEXT NOT NULL, + created_at_unix_millis BIGINT NOT NULL, + updated_at_unix_millis BIGINT NOT NULL +); +ALTER TABLE ksp_raw_transaction_conflicts ADD COLUMN IF NOT EXISTS transaction_signature BYTEA NOT NULL; +ALTER TABLE ksp_raw_transaction_conflicts ADD COLUMN IF NOT EXISTS status TEXT NOT NULL; +ALTER TABLE ksp_raw_transaction_conflicts ADD COLUMN IF NOT EXISTS revision NUMERIC(20, 0) NOT NULL; +ALTER TABLE ksp_raw_transaction_conflicts ADD COLUMN IF NOT EXISTS canonical_variant_id NUMERIC(20, 0) NOT NULL; +ALTER TABLE ksp_raw_transaction_conflicts ADD COLUMN IF NOT EXISTS incoming_variant_id NUMERIC(20, 0) NOT NULL; +ALTER TABLE ksp_raw_transaction_conflicts ADD COLUMN IF NOT EXISTS latest_relation TEXT NOT NULL; +ALTER TABLE ksp_raw_transaction_conflicts ADD COLUMN IF NOT EXISTS latest_reason_code TEXT NOT NULL; +ALTER TABLE ksp_raw_transaction_conflicts ADD COLUMN IF NOT EXISTS created_at_unix_millis BIGINT NOT NULL; +ALTER TABLE ksp_raw_transaction_conflicts ADD COLUMN IF NOT EXISTS updated_at_unix_millis BIGINT NOT NULL; diff --git a/crates/ksp-store-postgres-lib/src/lib.rs b/crates/ksp-store-postgres-lib/src/lib.rs index 401bd3e..fce60d9 100644 --- a/crates/ksp-store-postgres-lib/src/lib.rs +++ b/crates/ksp-store-postgres-lib/src/lib.rs @@ -1,10 +1,8 @@ // file: crates/ksp-store-postgres-lib/src/lib.rs -// version: 24 - +// version: 26 #![warn(missing_docs)] #![deny(unreachable_pub)] #![forbid(unsafe_code)] - //! Official PostgreSQL backend implementation for KSP Store. //! //! The backend owns the physical `tokio-postgres` connection, bounded Deadpool @@ -38,112 +36,76 @@ //! with exact counts while preserving both keyset traversal families unchanged. //! `0.3.8-pre.009` adds safe random-access observation inspection for both RAW //! families without changing the physical schema or read-by-key contracts. +//! `0.3.16-pre.003` layers additive V003 transaction-variant resources over the +//! frozen V000-V002 registry without changing any prior migration resource bytes. //! //! This crate depends on `ksp-store-api` and never on `ksp-store-lib`. The //! common facade consumes only this crate's narrow backend bridge and never //! exposes PostgreSQL pool, client, row or statement types. - mod constants; mod error; mod health; mod migration; +mod migration_v003; mod raw_account; mod raw_transaction; mod runtime; mod schema; +mod schema_v003; -/// Stable KSP error code for unsupported PostgreSQL retention compaction. pub use self::error::ERROR_CODE_POSTGRES_RETENTION_COMPACTION_UNSUPPORTED; -/// Safe backend-local error returned to the common Store facade. pub use self::error::PostgresBackendError; -/// Safe backend-local error classification used by the common Store facade. pub use self::error::PostgresBackendErrorKind; -/// Opaque physical PostgreSQL backend owning its connection pool. pub use self::runtime::PostgresBackend; -/// Safe PostgreSQL readiness projection returned through the backend bridge. pub use self::runtime::PostgresBackendHealthSnapshot; -/// Safe PostgreSQL pool counter projection returned through the backend bridge. pub use self::runtime::PostgresBackendRuntimeSnapshot; -/// Physical PostgreSQL settings bridge consumed only by the backend crate. pub use self::runtime::PostgresBackendSettings; -/// TLS mode accepted by the physical PostgreSQL settings bridge. pub use self::runtime::PostgresBackendTlsMode; -/// Crate-owned tracing target for PostgreSQL backend behavior. pub(crate) use self::constants::TRACING_TARGET; -/// Private bounded health probe consumed by the physical backend runtime. pub(crate) use self::health::probe_health; -/// Private migration/bootstrap runner consumed by the physical backend runtime. -pub(crate) use self::migration::bootstrap; -/// Current embedded migration version consumed by the private health probe. -pub(crate) use self::migration::current_migration_version; -/// Private RAW account cursor decoder consumed by the physical RAW account module. +pub(crate) use self::migration::bootstrap as legacy_bootstrap; +pub(crate) use self::migration::current_migration_version as legacy_current_migration_version; +pub(crate) use self::migration_v003::bootstrap; +pub(crate) use self::migration_v003::current_migration_version; pub(crate) use self::raw_account::cursor::decode_raw_account_cursor; -/// Private RAW account cursor encoder consumed by the physical RAW account module. pub(crate) use self::raw_account::cursor::encode_raw_account_cursor; -/// Private physical account page-limit converter consumed by the physical RAW account module. pub(crate) use self::raw_account::cursor::raw_account_physical_page_limit; -/// Private RAW account observation reader consumed by the physical backend runtime. pub(crate) use self::raw_account::get_raw_account_observation; -/// Private RAW account state reader consumed by the physical backend runtime. pub(crate) use self::raw_account::get_raw_account_state; -/// Private safe RAW account-observation inspection reader consumed by the physical backend runtime. pub(crate) use self::raw_account::inspect_raw_account_observations; -/// Private data-free RAW account-state inspection reader consumed by the physical backend runtime. pub(crate) use self::raw_account::inspect_raw_account_states; -/// Private RAW account-state list reader consumed by the physical backend runtime. pub(crate) use self::raw_account::list_raw_account_states; -/// Private atomic RAW account acquisition writer consumed by the physical backend runtime. pub(crate) use self::raw_account::persist_raw_account_acquisition; -/// Private additional RAW account observation writer consumed by the physical backend runtime. pub(crate) use self::raw_account::record_raw_account_observation; -/// Private RAW transaction cursor decoder consumed by the physical RAW module. pub(crate) use self::raw_transaction::cursor::decode_raw_transaction_cursor; -/// Private RAW transaction cursor encoder consumed by the physical RAW module. pub(crate) use self::raw_transaction::cursor::encode_raw_transaction_cursor; -/// Private physical page-limit converter consumed by the physical RAW module. pub(crate) use self::raw_transaction::cursor::raw_transaction_physical_page_limit; -/// Private RAW transaction reader consumed by the physical backend runtime. pub(crate) use self::raw_transaction::get_raw_transaction; -/// Private RAW transaction observation reader consumed by the physical backend runtime. pub(crate) use self::raw_transaction::get_raw_transaction_observation; -/// Private RAW transaction retention-state reader consumed by the physical backend runtime. pub(crate) use self::raw_transaction::get_raw_transaction_retention_state; -/// Private RAW transaction tombstone reader consumed by the physical backend runtime. pub(crate) use self::raw_transaction::get_raw_transaction_tombstone; -/// Private safe RAW transaction-observation inspection reader consumed by the physical backend runtime. pub(crate) use self::raw_transaction::inspect_raw_transaction_observations; -/// Private payload-free RAW transaction inspection reader consumed by the physical backend runtime. pub(crate) use self::raw_transaction::inspect_raw_transactions; -/// Private RAW transaction list reader consumed by the physical backend runtime. pub(crate) use self::raw_transaction::list_raw_transactions; -/// Private atomic RAW transaction acquisition writer consumed by the physical backend runtime. pub(crate) use self::raw_transaction::persist_raw_transaction_acquisition; -/// Private additional RAW transaction observation writer consumed by the physical backend runtime. pub(crate) use self::raw_transaction::record_raw_transaction_observation; -/// Private RAW transaction retention transition writer consumed by the physical backend runtime. pub(crate) use self::raw_transaction::transition_raw_transaction_retention; -/// Private Deadpool error mapper shared with the health probe. pub(crate) use self::runtime::map_pool_error; -/// Private Deadpool status projector shared with the health probe. pub(crate) use self::runtime::runtime_snapshot_from_status; -/// Private physical schema resource descriptor consumed by the migration engine. pub(crate) use self::schema::SchemaResource; -/// Private physical schema resource compatibility state consumed by the migration engine. pub(crate) use self::schema::SchemaResourceState; -/// Private V000 schema resource inventory consumed by the migration engine. pub(crate) use self::schema::V000_RESOURCES; -/// Private V001 schema resource inventory consumed by the migration engine. pub(crate) use self::schema::V001_RESOURCES; -/// Private V002 schema resource inventory consumed by the migration engine. pub(crate) use self::schema::V002_RESOURCES; -/// Private physical schema resource inspector consumed by the migration engine. pub(crate) use self::schema::inspect_resource; -/// Private managed-schema adoption probe consumed by the migration engine. pub(crate) use self::schema::managed_schema_objects_exist; -/// Private V001 external-schema compatibility gate consumed by the migration engine. pub(crate) use self::schema::verify_v001_external_compatibility; -/// Private V002 external-schema compatibility gate consumed by the migration engine. pub(crate) use self::schema::verify_v002_external_compatibility; +pub(crate) use self::schema_v003::V003SchemaResource; +pub(crate) use self::schema_v003::V003SchemaResourceState; +pub(crate) use self::schema_v003::V003_RESOURCES; +pub(crate) use self::schema_v003::inspect_v003_resource; +pub(crate) use self::schema_v003::verify_v003_external_compatibility; const _: &str = crate::TRACING_TARGET; diff --git a/crates/ksp-store-postgres-lib/src/migration_v003.rs b/crates/ksp-store-postgres-lib/src/migration_v003.rs new file mode 100644 index 0000000..dcf38d4 --- /dev/null +++ b/crates/ksp-store-postgres-lib/src/migration_v003.rs @@ -0,0 +1,525 @@ +// file: crates/ksp-store-postgres-lib/src/migration_v003.rs +// version: 2 + +use sha2::Digest; // rust-rules: trait-import + +const ADVISORY_LOCK_KEY: i64 = 0x4b53_5053_544f_5245; +const HISTORY_INSERT_SQL: &str = "INSERT INTO ksp_store_schema_migrations (version, name, checksum, applied_at) VALUES ($1, $2, $3, CURRENT_TIMESTAMP)"; +const HISTORY_LOAD_SQL: &str = "SELECT version, name, checksum FROM ksp_store_schema_migrations ORDER BY version"; +const IDENTITY_LOAD_SQL: &str = "SELECT singleton, network FROM ksp_store_identity ORDER BY singleton LIMIT 2"; +const LATEST_VERSION_SQL: &str = "SELECT MAX(version) FROM ksp_store_schema_migrations"; +const LOCK_POLL_INTERVAL_MS: u64 = 25; +const METADATA_EXISTS_SQL: &str = r#"SELECT EXISTS ( + SELECT 1 FROM information_schema.tables + WHERE table_schema = current_schema() + AND table_name = 'ksp_store_schema_migrations' + AND table_type = 'BASE TABLE' +)"#; +const SET_STATEMENT_TIMEOUT_SQL: &str = "SELECT set_config('statement_timeout', $1, true)"; +const V000_CHECKSUM: &str = "d29068b8c13b9dc0cc9ef6aaadd0fa12d41e0fe4c56541a1118c4bfc846a1450"; +const V001_CHECKSUM: &str = "31488cda2f08f3f46c4cdbdbb6c18c243662fada02eac4487040c8735d72cc51"; +const V002_CHECKSUM: &str = "ff21605ed45f7ab4c0f92bbb692700b4118a9488b04d50a31d259ac59bdb550e"; +const V002_VERSION: i64 = 2; +const V003_NAME: &str = "raw_transaction_variants"; +const V003_VERSION: i64 = 3; + +struct AppliedMigration { + checksum: std::string::String, + name: std::string::String, + version: i64, +} + +/// Returns the latest migration version embedded by the physical PostgreSQL backend. +#[must_use] +pub(crate) const fn current_migration_version() -> i64 { + return V003_VERSION; +} + +/// Runs the frozen V000-V002 bootstrap followed by additive V003 under the KSP schema lock. +pub(crate) async fn bootstrap( + client: &mut deadpool_postgres::Client, + network: &ksp_store_api::RawNetworkId, + schema_autocreate: bool, + schema_autoupdate: bool, + migration_timeout: std::time::Duration, + migration_lock_timeout: std::time::Duration, +) -> std::result::Result<(), crate::PostgresBackendError> { + if crate::legacy_current_migration_version() != 2 { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationMismatch, + "v003_legacy_registry_version", + )); + } + let metadata_result = metadata_exists_client(client).await; + let metadata_preexisting = match metadata_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let latest_before = if metadata_preexisting { + let latest_result = latest_version_client(client).await; + match latest_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + } + } else { + std::option::Option::None + }; + if let std::option::Option::Some(version) = latest_before + && version > V003_VERSION + { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::SchemaNewer, "history_newer")); + } + if legacy_bootstrap_required(latest_before) { + let legacy_result = crate::legacy_bootstrap( + client, + network, + schema_autocreate, + schema_autoupdate, + migration_timeout, + migration_lock_timeout, + ) + .await; + if let std::result::Result::Err(error) = legacy_result { + if error.kind() != crate::PostgresBackendErrorKind::SchemaNewer { + return std::result::Result::Err(error); + } + let latest_result = latest_version_client(client).await; + let latest = match latest_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(latest_error) => return std::result::Result::Err(latest_error), + }; + if !legacy_schema_newer_is_completed_v003(latest) { + return std::result::Result::Err(error); + } + } + } + let allow_v003_apply = v003_apply_allowed(metadata_preexisting, schema_autoupdate); + let bounded = tokio::time::timeout( + migration_timeout, + bootstrap_v003_inner(client, network, schema_autoupdate, allow_v003_apply, migration_timeout, migration_lock_timeout), + ) + .await; + return match bounded { + std::result::Result::Ok(result) => result, + std::result::Result::Err(_) => { + std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_timeout")) + }, + }; +} + +async fn bootstrap_v003_inner( + client: &mut deadpool_postgres::Client, + network: &ksp_store_api::RawNetworkId, + schema_autoupdate: bool, + allow_v003_apply: bool, + migration_timeout: std::time::Duration, + migration_lock_timeout: std::time::Duration, +) -> std::result::Result<(), crate::PostgresBackendError> { + let transaction_result = client.transaction().await; + let transaction = match transaction_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_begin")); + }, + }; + let lock_result = acquire_advisory_lock(&transaction, migration_lock_timeout).await; + if let std::result::Result::Err(error) = lock_result { + return std::result::Result::Err(error); + } + let timeout_result = set_statement_timeout(&transaction, migration_timeout).await; + if let std::result::Result::Err(error) = timeout_result { + return std::result::Result::Err(error); + } + let history_result = load_history(&transaction).await; + let history = match history_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let next_version_result = validate_history(history.as_slice()); + let next_version = match next_version_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + let legacy_result = verify_or_repair_legacy_resources(&transaction, schema_autoupdate).await; + if let std::result::Result::Err(error) = legacy_result { + return std::result::Result::Err(error); + } + let identity_result = validate_store_identity(&transaction, network).await; + if let std::result::Result::Err(error) = identity_result { + return std::result::Result::Err(error); + } + let v003_applied = next_version > V003_VERSION; + let v003_result = verify_or_apply_v003_resources(&transaction, schema_autoupdate, allow_v003_apply, v003_applied).await; + if let std::result::Result::Err(error) = v003_result { + return std::result::Result::Err(error); + } + let compatibility_result = crate::verify_v003_external_compatibility(&transaction).await; + if let std::result::Result::Err(error) = compatibility_result { + return std::result::Result::Err(error); + } + if !v003_applied { + let checksum = migration_checksum_v003(); + let insert_result = transaction.execute(HISTORY_INSERT_SQL, &[&V003_VERSION, &V003_NAME, &checksum]).await; + match insert_result { + std::result::Result::Ok(1) => {}, + std::result::Result::Ok(_) | std::result::Result::Err(_) => { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_insert")); + }, + } + } + let commit_result = transaction.commit().await; + return match commit_result { + std::result::Result::Ok(()) => std::result::Result::Ok(()), + std::result::Result::Err(_) => { + std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_commit")) + }, + }; +} + +async fn acquire_advisory_lock( + transaction: &deadpool_postgres::Transaction<'_>, + timeout: std::time::Duration, +) -> std::result::Result<(), crate::PostgresBackendError> { + let started = tokio::time::Instant::now(); + let deadline = match started.checked_add(timeout) { + std::option::Option::Some(value) => value, + std::option::Option::None => { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock_timeout")); + }, + }; + loop { + let row_result = transaction.query_one("SELECT pg_try_advisory_xact_lock($1)", &[&ADVISORY_LOCK_KEY]).await; + let row = match row_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock")); + }, + }; + let acquired_result = row.try_get::(0); + let acquired = match acquired_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationFailed, + "migration_lock_decode", + )); + }, + }; + if acquired { + return std::result::Result::Ok(()); + } + let now = tokio::time::Instant::now(); + if now >= deadline { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "migration_lock_timeout")); + } + let candidate = now + std::time::Duration::from_millis(LOCK_POLL_INTERVAL_MS); + let wake = if candidate < deadline { candidate } else { deadline }; + tokio::time::sleep_until(wake).await; + } +} + +async fn verify_or_repair_legacy_resources( + transaction: &deadpool_postgres::Transaction<'_>, + schema_autoupdate: bool, +) -> std::result::Result<(), crate::PostgresBackendError> { + for resources in [crate::V000_RESOURCES, crate::V001_RESOURCES, crate::V002_RESOURCES] { + for resource in resources { + let state_result = crate::inspect_resource(transaction, resource).await; + let state = match state_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + match state { + crate::SchemaResourceState::Compatible => {}, + crate::SchemaResourceState::Incompatible => { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationMismatch, + "schema_resource_incompatible", + )); + }, + crate::SchemaResourceState::Missing => { + if !schema_autoupdate { + return std::result::Result::Err(schema_autoupdate_disabled_error()); + } + if !resource.repair_existing { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationMismatch, + "schema_resource_missing", + )); + } + let execute_result = transaction.batch_execute(resource.sql).await; + if execute_result.is_err() { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationFailed, + "schema_resource_apply", + )); + } + let verify_result = crate::inspect_resource(transaction, resource).await; + match verify_result { + std::result::Result::Ok(crate::SchemaResourceState::Compatible) => {}, + std::result::Result::Ok(crate::SchemaResourceState::Missing | crate::SchemaResourceState::Incompatible) => { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationMismatch, + "schema_resource_post_apply", + )); + }, + std::result::Result::Err(error) => return std::result::Result::Err(error), + } + }, + } + } + } + let v001_result = crate::verify_v001_external_compatibility(transaction).await; + if let std::result::Result::Err(error) = v001_result { + return std::result::Result::Err(error); + } + return crate::verify_v002_external_compatibility(transaction).await; +} + +async fn verify_or_apply_v003_resources( + transaction: &deadpool_postgres::Transaction<'_>, + schema_autoupdate: bool, + allow_v003_apply: bool, + applied_history: bool, +) -> std::result::Result<(), crate::PostgresBackendError> { + for resource in crate::V003_RESOURCES { + let state_result = crate::inspect_v003_resource(transaction, resource).await; + let state = match state_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + match state { + crate::V003SchemaResourceState::Compatible => {}, + crate::V003SchemaResourceState::Incompatible => { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationMismatch, + "schema_v003_resource_incompatible", + )); + }, + crate::V003SchemaResourceState::Missing => { + if applied_history && !schema_autoupdate { + return std::result::Result::Err(schema_autoupdate_disabled_error()); + } + if !applied_history && !allow_v003_apply { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationFailed, + "migration_pending", + )); + } + if applied_history && !resource.repair_existing { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationMismatch, + "schema_v003_resource_missing", + )); + } + let execute_result = transaction.batch_execute(resource.sql).await; + if execute_result.is_err() { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationFailed, + "schema_v003_resource_apply", + )); + } + let verify_result = crate::inspect_v003_resource(transaction, resource).await; + match verify_result { + std::result::Result::Ok(crate::V003SchemaResourceState::Compatible) => {}, + std::result::Result::Ok(crate::V003SchemaResourceState::Missing | crate::V003SchemaResourceState::Incompatible) => { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationMismatch, + "schema_v003_resource_post_apply", + )); + }, + std::result::Result::Err(error) => return std::result::Result::Err(error), + } + }, + } + } + return std::result::Result::Ok(()); +} + +async fn validate_store_identity( + transaction: &deadpool_postgres::Transaction<'_>, + network: &ksp_store_api::RawNetworkId, +) -> std::result::Result<(), crate::PostgresBackendError> { + let rows_result = transaction.query(IDENTITY_LOAD_SQL, &[]).await; + let rows = match rows_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "store_identity_read")); + }, + }; + if rows.len() != 1 { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "store_identity_count")); + } + let row = &rows[0]; + let singleton_result = row.try_get::(0); + let network_result = row.try_get::(1); + let (singleton, stored_network) = match (singleton_result, network_result) { + (std::result::Result::Ok(singleton), std::result::Result::Ok(stored_network)) => (singleton, stored_network), + _ => { + return std::result::Result::Err(crate::PostgresBackendError::new( + crate::PostgresBackendErrorKind::MigrationMismatch, + "store_identity_decode", + )); + }, + }; + if singleton != 1 || stored_network != network.as_str() { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "store_identity_network")); + } + return std::result::Result::Ok(()); +} + +async fn metadata_exists_client(client: &deadpool_postgres::Client) -> std::result::Result { + let result = client.query_one(METADATA_EXISTS_SQL, &[]).await; + let row = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_probe")); + }, + }; + return match row.try_get::(0) { + std::result::Result::Ok(value) => std::result::Result::Ok(value), + std::result::Result::Err(_) => { + std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "metadata_probe_decode")) + }, + }; +} + +async fn latest_version_client(client: &deadpool_postgres::Client) -> std::result::Result, crate::PostgresBackendError> { + let result = client.query_one(LATEST_VERSION_SQL, &[]).await; + let row = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_latest")); + }, + }; + return match row.try_get::>(0) { + std::result::Result::Ok(value) => std::result::Result::Ok(value), + std::result::Result::Err(_) => { + std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_latest_decode")) + }, + }; +} + +async fn load_history( + transaction: &deadpool_postgres::Transaction<'_>, +) -> std::result::Result, crate::PostgresBackendError> { + let result = transaction.query(HISTORY_LOAD_SQL, &[]).await; + let rows = match result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_load")); + }, + }; + let mut history = std::vec::Vec::with_capacity(rows.len()); + for row in rows { + let version = row.try_get::(0); + let name = row.try_get::(1); + let checksum = row.try_get::(2); + match (version, name, checksum) { + (std::result::Result::Ok(version), std::result::Result::Ok(name), std::result::Result::Ok(checksum)) => { + history.push(AppliedMigration { checksum, name, version }); + }, + _ => { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "history_decode")); + }, + } + } + return std::result::Result::Ok(history); +} + +fn validate_history(history: &[AppliedMigration]) -> std::result::Result { + if history.len() < 3 { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_missing")); + } + let expected = [ + (0_i64, "bootstrap", V000_CHECKSUM), + (1_i64, "raw_transaction", V001_CHECKSUM), + (2_i64, "raw_account_state", V002_CHECKSUM), + ]; + for (index, (version, name, checksum)) in expected.iter().enumerate() { + let applied = &history[index]; + if applied.version != *version || applied.name != *name || applied.checksum != *checksum { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_diverged")); + } + } + if history.len() == 3 { + return std::result::Result::Ok(3); + } + let v003 = &history[3]; + if v003.version > V003_VERSION { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::SchemaNewer, "history_newer")); + } + if v003.version != V003_VERSION || v003.name != V003_NAME || v003.checksum != migration_checksum_v003() { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "history_diverged")); + } + if history.len() > 4 { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::SchemaNewer, "history_newer")); + } + return std::result::Result::Ok(4); +} + +fn v003_apply_allowed(metadata_preexisting: bool, schema_autoupdate: bool) -> bool { + return !metadata_preexisting || schema_autoupdate; +} + +fn legacy_bootstrap_required(latest_before: std::option::Option) -> bool { + let legacy_version = match latest_before { + std::option::Option::Some(value) => value, + std::option::Option::None => -1, + }; + return legacy_version < V002_VERSION; +} + +fn legacy_schema_newer_is_completed_v003(latest: std::option::Option) -> bool { + return latest == std::option::Option::Some(V003_VERSION); +} + +fn migration_checksum_v003() -> std::string::String { + let mut hasher = sha2::Sha256::new(); + hasher.update(b"ksp-migration-resources-v1\0"); + for resource in crate::V003_RESOURCES { + update_v003_checksum(&mut hasher, resource); + } + return encode_digest(hasher.finalize().as_slice()); +} + +fn update_v003_checksum(hasher: &mut sha2::Sha256, resource: &crate::V003SchemaResource) { + hasher.update(resource.id.as_bytes()); + hasher.update([0]); + hasher.update(resource.sql.as_bytes()); + hasher.update([0]); + return; +} + +fn encode_digest(bytes: &[u8]) -> std::string::String { + const HEX_LOWER: &[u8; 16] = b"0123456789abcdef"; + let mut encoded = std::string::String::with_capacity(bytes.len() * 2); + for byte in bytes { + let value = *byte; + encoded.push(char::from(HEX_LOWER[(value >> 4) as usize])); + encoded.push(char::from(HEX_LOWER[(value & 0x0f) as usize])); + } + return encoded; +} + +async fn set_statement_timeout( + transaction: &deadpool_postgres::Transaction<'_>, + timeout: std::time::Duration, +) -> std::result::Result<(), crate::PostgresBackendError> { + let timeout_value = format!("{}ms", timeout.as_millis()); + let result = transaction.query_one(SET_STATEMENT_TIMEOUT_SQL, &[&timeout_value]).await; + return match result { + std::result::Result::Ok(_) => std::result::Result::Ok(()), + std::result::Result::Err(_) => { + std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, "statement_timeout")) + }, + }; +} + +fn schema_autoupdate_disabled_error() -> crate::PostgresBackendError { + return crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, "schema_autoupdate_disabled"); +} + +#[cfg(test)] +#[path = "../unit_tests/migration_v003.rs"] +mod tests; diff --git a/crates/ksp-store-postgres-lib/src/schema_v003.rs b/crates/ksp-store-postgres-lib/src/schema_v003.rs new file mode 100644 index 0000000..1354ff9 --- /dev/null +++ b/crates/ksp-store-postgres-lib/src/schema_v003.rs @@ -0,0 +1,666 @@ +// file: crates/ksp-store-postgres-lib/src/schema_v003.rs +// version: 2 + +const COLUMN_LOAD_SQL: &str = r#"SELECT column_name, udt_name, is_nullable = 'YES', numeric_precision, numeric_scale, column_default, is_identity, is_generated +FROM information_schema.columns +WHERE table_schema = current_schema() AND table_name = $1 +ORDER BY ordinal_position"#; +const CONSTRAINT_LOAD_SQL: &str = r#"SELECT c.contype::TEXT, c.convalidated, c.condeferrable, c.condeferred, pg_get_constraintdef(c.oid, true) +FROM pg_constraint c +JOIN pg_class t ON t.oid = c.conrelid +JOIN pg_namespace n ON n.oid = t.relnamespace +WHERE n.nspname = current_schema() AND t.relname = $1 AND c.conname = $2"#; +const INDEX_LOAD_SQL: &str = r#"SELECT i.indisunique, am.amname, pg_get_indexdef(i.indexrelid), pg_get_expr(i.indpred, i.indrelid) +FROM pg_index i +JOIN pg_class idx ON idx.oid = i.indexrelid +JOIN pg_class tbl ON tbl.oid = i.indrelid +JOIN pg_namespace n ON n.oid = tbl.relnamespace +JOIN pg_am am ON am.oid = idx.relam +WHERE n.nspname = current_schema() AND tbl.relname = $1 AND idx.relname = $2 AND i.indisvalid"#; +const TABLE_LOAD_SQL: &str = r#"SELECT c.relkind::TEXT, c.relrowsecurity, c.relforcerowsecurity +FROM pg_class c +JOIN pg_namespace n ON n.oid = c.relnamespace +WHERE n.nspname = current_schema() AND c.relname = $1"#; + +/// Compatibility state of one additive V003 physical schema resource. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum V003SchemaResourceState { + /// Resource does not exist or is missing one additive expected column. + Missing, + /// Resource exactly satisfies the owned V003 physical contract. + Compatible, + /// Existing resource contradicts the owned V003 physical contract. + Incompatible, +} + +/// Additive V003 physical schema resource descriptor consumed by the V003 migration extension. +#[derive(Clone, Copy)] +pub(crate) struct V003SchemaResource { + /// Stable resource identifier included in the V003 migration checksum domain. + pub(crate) id: &'static str, + object: V003SchemaObjectContract, + /// Whether an already-applied history may repair a missing additive resource. + pub(crate) repair_existing: bool, + /// Immutable SQL bytes owned by this V003 physical resource. + pub(crate) sql: &'static str, +} + +#[derive(Clone, Copy)] +enum V003SchemaObjectContract { + Constraint(ConstraintContract), + Index(IndexContract), + Table(TableContract), +} + +#[derive(Clone, Copy)] +struct ConstraintContract { + kind: &'static str, + name: &'static str, + table: &'static str, +} + +#[derive(Clone, Copy)] +struct IndexContract { + access_method: &'static str, + key_fragment: &'static str, + name: &'static str, + predicate_fragment: std::option::Option<&'static str>, + table: &'static str, + unique: bool, +} + +#[derive(Clone, Copy)] +struct TableContract { + columns: &'static [ColumnContract], + name: &'static str, +} + +#[derive(Clone, Copy)] +struct ColumnContract { + name: &'static str, + nullable: bool, + numeric_precision: std::option::Option, + numeric_scale: std::option::Option, + udt_name: &'static str, +} + +const VARIANT_COLUMNS: &[ColumnContract] = &[ + column("transaction_signature", false, None, None, "bytea"), + column("variant_id", false, Some(20), Some(0), "numeric"), + column("origin_kind", false, None, None, "text"), + column("slot", false, Some(20), Some(0), "numeric"), + column("block_time_unix_millis", true, None, None, "int8"), + column("format_id", false, None, None, "text"), + column("format_version", false, None, None, "int8"), + column("content_hash", false, None, None, "bytea"), + column("payload", true, None, None, "bytea"), + column("retention_state", false, None, None, "text"), + column("created_at_unix_millis", false, None, None, "int8"), +]; +const SELECTOR_COLUMNS: &[ColumnContract] = &[ + column("transaction_signature", false, None, None, "bytea"), + column("canonical_variant_id", false, Some(20), Some(0), "numeric"), + column("canonical_revision", false, Some(20), Some(0), "numeric"), + column("updated_at_unix_millis", false, None, None, "int8"), +]; +const OBSERVATION_VARIANT_COLUMNS: &[ColumnContract] = &[ + column("observation_key", false, None, None, "bytea"), + column("transaction_signature", false, None, None, "bytea"), + column("variant_id", false, Some(20), Some(0), "numeric"), + column("linked_at_unix_millis", false, None, None, "int8"), +]; +const CONFLICT_COLUMNS: &[ColumnContract] = &[ + column("transaction_signature", false, None, None, "bytea"), + column("status", false, None, None, "text"), + column("revision", false, Some(20), Some(0), "numeric"), + column("canonical_variant_id", false, Some(20), Some(0), "numeric"), + column("incoming_variant_id", false, Some(20), Some(0), "numeric"), + column("latest_relation", false, None, None, "text"), + column("latest_reason_code", false, None, None, "text"), + column("created_at_unix_millis", false, None, None, "int8"), + column("updated_at_unix_millis", false, None, None, "int8"), +]; + +/// Immutable V003 physical resource inventory. +pub(crate) const V003_RESOURCES: &[V003SchemaResource] = &[ + table_resource( + "tables/001_ksp_raw_transaction_variants.sql", + "ksp_raw_transaction_variants", + VARIANT_COLUMNS, + include_str!("../migrations/v003_raw_transaction_variants/tables/001_ksp_raw_transaction_variants.sql"), + ), + table_resource( + "tables/002_ksp_raw_transaction_canonical_selectors.sql", + "ksp_raw_transaction_canonical_selectors", + SELECTOR_COLUMNS, + include_str!("../migrations/v003_raw_transaction_variants/tables/002_ksp_raw_transaction_canonical_selectors.sql"), + ), + table_resource( + "tables/003_ksp_raw_transaction_observation_variants.sql", + "ksp_raw_transaction_observation_variants", + OBSERVATION_VARIANT_COLUMNS, + include_str!("../migrations/v003_raw_transaction_variants/tables/003_ksp_raw_transaction_observation_variants.sql"), + ), + table_resource( + "tables/004_ksp_raw_transaction_conflicts.sql", + "ksp_raw_transaction_conflicts", + CONFLICT_COLUMNS, + include_str!("../migrations/v003_raw_transaction_variants/tables/004_ksp_raw_transaction_conflicts.sql"), + ), + constraint_resource( + "constraints/001_pk_ksp_raw_transaction_variants.sql", + "p", + "pk_ksp_raw_transaction_variants", + "ksp_raw_transaction_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/001_pk_ksp_raw_transaction_variants.sql"), + ), + constraint_resource( + "constraints/002_ck_ksp_raw_transaction_variants_identity.sql", + "c", + "ck_ksp_raw_transaction_variants_identity", + "ksp_raw_transaction_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/002_ck_ksp_raw_transaction_variants_identity.sql"), + ), + constraint_resource( + "constraints/003_ck_ksp_raw_transaction_variants_origin.sql", + "c", + "ck_ksp_raw_transaction_variants_origin", + "ksp_raw_transaction_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/003_ck_ksp_raw_transaction_variants_origin.sql"), + ), + constraint_resource( + "constraints/004_ck_ksp_raw_transaction_variants_slot_time.sql", + "c", + "ck_ksp_raw_transaction_variants_slot_time", + "ksp_raw_transaction_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/004_ck_ksp_raw_transaction_variants_slot_time.sql"), + ), + constraint_resource( + "constraints/005_ck_ksp_raw_transaction_variants_format.sql", + "c", + "ck_ksp_raw_transaction_variants_format", + "ksp_raw_transaction_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/005_ck_ksp_raw_transaction_variants_format.sql"), + ), + constraint_resource( + "constraints/006_ck_ksp_raw_transaction_variants_payload.sql", + "c", + "ck_ksp_raw_transaction_variants_payload", + "ksp_raw_transaction_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/006_ck_ksp_raw_transaction_variants_payload.sql"), + ), + constraint_resource( + "constraints/007_fk_ksp_raw_transaction_variants_transaction.sql", + "f", + "fk_ksp_raw_transaction_variants_transaction", + "ksp_raw_transaction_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/007_fk_ksp_raw_transaction_variants_transaction.sql"), + ), + constraint_resource( + "constraints/008_pk_ksp_raw_transaction_canonical_selectors.sql", + "p", + "pk_ksp_raw_transaction_canonical_selectors", + "ksp_raw_transaction_canonical_selectors", + include_str!("../migrations/v003_raw_transaction_variants/constraints/008_pk_ksp_raw_transaction_canonical_selectors.sql"), + ), + constraint_resource( + "constraints/009_ck_ksp_raw_transaction_canonical_selectors_bounds.sql", + "c", + "ck_ksp_raw_transaction_canonical_selectors_bounds", + "ksp_raw_transaction_canonical_selectors", + include_str!("../migrations/v003_raw_transaction_variants/constraints/009_ck_ksp_raw_transaction_canonical_selectors_bounds.sql"), + ), + constraint_resource( + "constraints/010_fk_ksp_raw_transaction_canonical_selectors_variant.sql", + "f", + "fk_ksp_raw_transaction_canonical_selectors_variant", + "ksp_raw_transaction_canonical_selectors", + include_str!("../migrations/v003_raw_transaction_variants/constraints/010_fk_ksp_raw_transaction_canonical_selectors_variant.sql"), + ), + constraint_resource( + "constraints/011_pk_ksp_raw_transaction_observation_variants.sql", + "p", + "pk_ksp_raw_transaction_observation_variants", + "ksp_raw_transaction_observation_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/011_pk_ksp_raw_transaction_observation_variants.sql"), + ), + constraint_resource( + "constraints/012_ck_ksp_raw_transaction_observation_variants_bounds.sql", + "c", + "ck_ksp_raw_transaction_observation_variants_bounds", + "ksp_raw_transaction_observation_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/012_ck_ksp_raw_transaction_observation_variants_bounds.sql"), + ), + constraint_resource( + "constraints/013_fk_ksp_raw_transaction_observation_variants_observation.sql", + "f", + "fk_ksp_raw_transaction_observation_variants_observation", + "ksp_raw_transaction_observation_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/013_fk_ksp_raw_transaction_observation_variants_observation.sql"), + ), + constraint_resource( + "constraints/014_fk_ksp_raw_transaction_observation_variants_variant.sql", + "f", + "fk_ksp_raw_transaction_observation_variants_variant", + "ksp_raw_transaction_observation_variants", + include_str!("../migrations/v003_raw_transaction_variants/constraints/014_fk_ksp_raw_transaction_observation_variants_variant.sql"), + ), + constraint_resource( + "constraints/015_pk_ksp_raw_transaction_conflicts.sql", + "p", + "pk_ksp_raw_transaction_conflicts", + "ksp_raw_transaction_conflicts", + include_str!("../migrations/v003_raw_transaction_variants/constraints/015_pk_ksp_raw_transaction_conflicts.sql"), + ), + constraint_resource( + "constraints/016_ck_ksp_raw_transaction_conflicts_bounds.sql", + "c", + "ck_ksp_raw_transaction_conflicts_bounds", + "ksp_raw_transaction_conflicts", + include_str!("../migrations/v003_raw_transaction_variants/constraints/016_ck_ksp_raw_transaction_conflicts_bounds.sql"), + ), + constraint_resource( + "constraints/017_fk_ksp_raw_transaction_conflicts_canonical_variant.sql", + "f", + "fk_ksp_raw_transaction_conflicts_canonical_variant", + "ksp_raw_transaction_conflicts", + include_str!("../migrations/v003_raw_transaction_variants/constraints/017_fk_ksp_raw_transaction_conflicts_canonical_variant.sql"), + ), + constraint_resource( + "constraints/018_fk_ksp_raw_transaction_conflicts_incoming_variant.sql", + "f", + "fk_ksp_raw_transaction_conflicts_incoming_variant", + "ksp_raw_transaction_conflicts", + include_str!("../migrations/v003_raw_transaction_variants/constraints/018_fk_ksp_raw_transaction_conflicts_incoming_variant.sql"), + ), + index_resource( + "indexes/001_ix_ksp_raw_transaction_variants_signature_hash_variant.sql", + "(transaction_signature, content_hash, variant_id)", + "ix_ksp_raw_transaction_variants_signature_hash_variant", + "ksp_raw_transaction_variants", + include_str!("../migrations/v003_raw_transaction_variants/indexes/001_ix_ksp_raw_transaction_variants_signature_hash_variant.sql"), + ), + index_resource( + "indexes/002_ix_ksp_raw_transaction_observation_variants_variant.sql", + "(transaction_signature, variant_id, observation_key)", + "ix_ksp_raw_transaction_observation_variants_variant", + "ksp_raw_transaction_observation_variants", + include_str!("../migrations/v003_raw_transaction_variants/indexes/002_ix_ksp_raw_transaction_observation_variants_variant.sql"), + ), + index_resource( + "indexes/003_ix_ksp_raw_transaction_conflicts_status_updated_signature.sql", + "(status, updated_at_unix_millis, transaction_signature)", + "ix_ksp_raw_transaction_conflicts_status_updated_signature", + "ksp_raw_transaction_conflicts", + include_str!("../migrations/v003_raw_transaction_variants/indexes/003_ix_ksp_raw_transaction_conflicts_status_updated_signature.sql"), + ), +]; + +const fn column( + name: &'static str, + nullable: bool, + numeric_precision: std::option::Option, + numeric_scale: std::option::Option, + udt_name: &'static str, +) -> ColumnContract { + return ColumnContract { name, nullable, numeric_precision, numeric_scale, udt_name }; +} + +const fn table_resource(id: &'static str, name: &'static str, columns: &'static [ColumnContract], sql: &'static str) -> V003SchemaResource { + return V003SchemaResource { id, object: V003SchemaObjectContract::Table(TableContract { columns, name }), repair_existing: true, sql }; +} + +const fn constraint_resource( + id: &'static str, + kind: &'static str, + name: &'static str, + table: &'static str, + sql: &'static str, +) -> V003SchemaResource { + return V003SchemaResource { + id, + object: V003SchemaObjectContract::Constraint(ConstraintContract { kind, name, table }), + repair_existing: true, + sql, + }; +} + +const fn index_resource( + id: &'static str, + key_fragment: &'static str, + name: &'static str, + table: &'static str, + sql: &'static str, +) -> V003SchemaResource { + return V003SchemaResource { + id, + object: V003SchemaObjectContract::Index(IndexContract { + access_method: "btree", + key_fragment, + name, + predicate_fragment: std::option::Option::None, + table, + unique: false, + }), + repair_existing: true, + sql, + }; +} + +/// Inspects one V003 resource against the active PostgreSQL schema. +pub(crate) async fn inspect_v003_resource( + transaction: &deadpool_postgres::Transaction<'_>, + resource: &V003SchemaResource, +) -> std::result::Result { + return match resource.object { + V003SchemaObjectContract::Constraint(contract) => inspect_constraint(transaction, contract, resource.sql).await, + V003SchemaObjectContract::Index(contract) => inspect_index(transaction, contract).await, + V003SchemaObjectContract::Table(contract) => inspect_table(transaction, contract).await, + }; +} + +/// Verifies that every owned V003 object is compatible after migration/repair. +pub(crate) async fn verify_v003_external_compatibility( + transaction: &deadpool_postgres::Transaction<'_>, +) -> std::result::Result<(), crate::PostgresBackendError> { + for resource in V003_RESOURCES { + let state_result = inspect_v003_resource(transaction, resource).await; + let state = match state_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(error) => return std::result::Result::Err(error), + }; + if state != V003SchemaResourceState::Compatible { + return schema_incompatible("schema_v003_external_contract"); + } + } + return std::result::Result::Ok(()); +} + +async fn inspect_constraint( + transaction: &deadpool_postgres::Transaction<'_>, + contract: ConstraintContract, + resource_sql: &str, +) -> std::result::Result { + let expected = match expected_constraint_definition(resource_sql, contract.name) { + std::option::Option::Some(value) => value, + std::option::Option::None => return schema_query_error("schema_v003_constraint_contract"), + }; + let result = transaction.query_opt(CONSTRAINT_LOAD_SQL, &[&contract.table, &contract.name]).await; + let row = match result { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Ok(V003SchemaResourceState::Missing), + std::result::Result::Err(_) => return schema_query_error("schema_v003_constraint_probe"), + }; + let kind = row.try_get::(0); + let validated = row.try_get::(1); + let deferrable = row.try_get::(2); + let deferred = row.try_get::(3); + let definition = row.try_get::(4); + let (kind, validated, deferrable, deferred, definition) = match (kind, validated, deferrable, deferred, definition) { + ( + std::result::Result::Ok(kind), + std::result::Result::Ok(validated), + std::result::Result::Ok(deferrable), + std::result::Result::Ok(deferred), + std::result::Result::Ok(definition), + ) => (kind, validated, deferrable, deferred, definition), + _ => return schema_query_error("schema_v003_constraint_probe_decode"), + }; + if kind != contract.kind || !validated || deferrable || deferred || normalize_catalog_sql(definition.as_str()) != expected { + return std::result::Result::Ok(V003SchemaResourceState::Incompatible); + } + return std::result::Result::Ok(V003SchemaResourceState::Compatible); +} + +async fn inspect_index( + transaction: &deadpool_postgres::Transaction<'_>, + contract: IndexContract, +) -> std::result::Result { + let result = transaction.query_opt(INDEX_LOAD_SQL, &[&contract.table, &contract.name]).await; + let row = match result { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Ok(V003SchemaResourceState::Missing), + std::result::Result::Err(_) => return schema_query_error("schema_v003_index_probe"), + }; + let unique = row.try_get::(0); + let access_method = row.try_get::(1); + let definition = row.try_get::(2); + let predicate = row.try_get::>(3); + let (unique, access_method, definition, predicate) = match (unique, access_method, definition, predicate) { + (std::result::Result::Ok(unique), std::result::Result::Ok(access_method), std::result::Result::Ok(definition), std::result::Result::Ok(predicate)) => { + (unique, access_method, definition, predicate) + }, + _ => return schema_query_error("schema_v003_index_probe_decode"), + }; + let definition = normalize_catalog_sql(definition.as_str()); + let predicate = predicate.map(|value| return normalize_catalog_sql(value.as_str())); + let predicate_matches = match (contract.predicate_fragment, predicate.as_deref()) { + (std::option::Option::Some(expected), std::option::Option::Some(value)) => value.contains(expected), + (std::option::Option::None, std::option::Option::None) => true, + _ => false, + }; + if unique != contract.unique + || access_method != contract.access_method + || !definition.contains(&normalize_catalog_sql(contract.key_fragment)) + || !predicate_matches + { + return std::result::Result::Ok(V003SchemaResourceState::Incompatible); + } + return std::result::Result::Ok(V003SchemaResourceState::Compatible); +} + +async fn inspect_table( + transaction: &deadpool_postgres::Transaction<'_>, + contract: TableContract, +) -> std::result::Result { + let table_result = transaction.query_opt(TABLE_LOAD_SQL, &[&contract.name]).await; + let table_row = match table_result { + std::result::Result::Ok(std::option::Option::Some(value)) => value, + std::result::Result::Ok(std::option::Option::None) => return std::result::Result::Ok(V003SchemaResourceState::Missing), + std::result::Result::Err(_) => return schema_query_error("schema_v003_table_probe"), + }; + let relkind = table_row.try_get::(0); + let row_security = table_row.try_get::(1); + let force_row_security = table_row.try_get::(2); + let (relkind, row_security, force_row_security) = match (relkind, row_security, force_row_security) { + (std::result::Result::Ok(relkind), std::result::Result::Ok(row_security), std::result::Result::Ok(force_row_security)) => { + (relkind, row_security, force_row_security) + }, + _ => return schema_query_error("schema_v003_table_probe_decode"), + }; + if relkind != "r" || row_security || force_row_security { + return std::result::Result::Ok(V003SchemaResourceState::Incompatible); + } + let rows_result = transaction.query(COLUMN_LOAD_SQL, &[&contract.name]).await; + let rows = match rows_result { + std::result::Result::Ok(value) => value, + std::result::Result::Err(_) => return schema_query_error("schema_v003_column_probe"), + }; + let mut actual = std::vec::Vec::::with_capacity(rows.len()); + for row in rows { + let decoded = ActualColumn::from_row(&row); + match decoded { + std::result::Result::Ok(value) => actual.push(value), + std::result::Result::Err(error) => return std::result::Result::Err(error), + } + } + let mut missing = false; + for expected in contract.columns { + let found = actual.iter().find(|candidate| return candidate.name == expected.name); + match found { + std::option::Option::Some(value) if value.matches(expected) => {}, + std::option::Option::Some(_) => return std::result::Result::Ok(V003SchemaResourceState::Incompatible), + std::option::Option::None => missing = true, + } + } + for column in &actual { + if contract.columns.iter().any(|expected| return expected.name == column.name) { + continue; + } + if !column.is_non_blocking_extra() { + return std::result::Result::Ok(V003SchemaResourceState::Incompatible); + } + } + if missing { + return std::result::Result::Ok(V003SchemaResourceState::Missing); + } + return std::result::Result::Ok(V003SchemaResourceState::Compatible); +} + +struct ActualColumn { + default: std::option::Option, + generated: std::string::String, + identity: std::string::String, + name: std::string::String, + nullable: bool, + numeric_precision: std::option::Option, + numeric_scale: std::option::Option, + udt_name: std::string::String, +} + +impl ActualColumn { + fn from_row(row: &tokio_postgres::Row) -> std::result::Result { + let name = row.try_get::(0); + let udt_name = row.try_get::(1); + let nullable = row.try_get::(2); + let numeric_precision = row.try_get::>(3); + let numeric_scale = row.try_get::>(4); + let default = row.try_get::>(5); + let identity = row.try_get::(6); + let generated = row.try_get::(7); + return match (name, udt_name, nullable, numeric_precision, numeric_scale, default, identity, generated) { + ( + std::result::Result::Ok(name), + std::result::Result::Ok(udt_name), + std::result::Result::Ok(nullable), + std::result::Result::Ok(numeric_precision), + std::result::Result::Ok(numeric_scale), + std::result::Result::Ok(default), + std::result::Result::Ok(identity), + std::result::Result::Ok(generated), + ) => std::result::Result::Ok(Self { default, generated, identity, name, nullable, numeric_precision, numeric_scale, udt_name }), + _ => schema_query_error("schema_v003_column_probe_decode"), + }; + } + + fn is_non_blocking_extra(&self) -> bool { + return self.nullable && self.default.is_none() && self.identity == "NO" && self.generated == "NEVER"; + } + + fn matches(&self, expected: &ColumnContract) -> bool { + let numeric_matches = if expected.udt_name == "numeric" { + self.numeric_precision == expected.numeric_precision && self.numeric_scale == expected.numeric_scale + } else { + true + }; + return self.udt_name == expected.udt_name + && self.nullable == expected.nullable + && numeric_matches + && self.identity == "NO" + && self.generated == "NEVER"; + } +} + +fn expected_constraint_definition(resource_sql: &str, name: &str) -> std::option::Option { + let marker = std::format!("ADD CONSTRAINT {name} "); + let marker_start = match resource_sql.find(marker.as_str()) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let start = match marker_start.checked_add(marker.len()) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let tail = match resource_sql.get(start..) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let end = match tail.find(';') { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + let definition = match tail.get(..end) { + std::option::Option::Some(value) => value, + std::option::Option::None => return std::option::Option::None, + }; + return std::option::Option::Some(normalize_catalog_sql(definition)); +} + +fn normalize_catalog_sql(value: &str) -> std::string::String { + let value = normalize_quoted_integral_cast_literals(value); + return value + .chars() + .filter(|character| return !character.is_whitespace() && *character != '"' && *character != '(' && *character != ')') + .collect::() + .replace("::text", "") + .replace("::numeric", "") + .replace("::bigint", "") + .replace("::smallint", "") + .replace("::integer", "") + .replace("::int8", "") + .replace("::int4", "") + .replace("::int2", "") + .to_ascii_lowercase(); +} + +fn normalize_quoted_integral_cast_literals(value: &str) -> std::string::String { + const INTEGRAL_CASTS: &[&str] = &["::bigint", "::int2", "::int4", "::int8", "::integer", "::numeric", "::smallint"]; + let mut normalized = std::string::String::with_capacity(value.len()); + let mut remaining = value; + loop { + let (before_quote, after_quote) = match remaining.split_once('\'') { + std::option::Option::Some(value) => value, + std::option::Option::None => { + normalized.push_str(remaining); + break; + }, + }; + normalized.push_str(before_quote); + let (literal, after_literal) = match after_quote.split_once('\'') { + std::option::Option::Some(value) => value, + std::option::Option::None => { + normalized.push('\''); + normalized.push_str(after_quote); + break; + }, + }; + let has_digit = literal.chars().any(|character| return character.is_ascii_digit()); + let integral_literal = !literal.is_empty() + && has_digit + && literal + .chars() + .enumerate() + .all(|(offset, character)| return character.is_ascii_digit() || (offset == 0 && (character == '+' || character == '-'))); + let mut after_integral_cast = std::option::Option::None; + if integral_literal { + for cast in INTEGRAL_CASTS { + if let std::option::Option::Some(value) = after_literal.strip_prefix(*cast) { + after_integral_cast = std::option::Option::Some(value); + break; + } + } + } + if let std::option::Option::Some(value) = after_integral_cast { + normalized.push_str(literal); + remaining = value; + continue; + } + normalized.push('\''); + normalized.push_str(literal); + normalized.push('\''); + remaining = after_literal; + } + return normalized; +} + +fn schema_incompatible(phase: &'static str) -> std::result::Result { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, phase)); +} + +fn schema_query_error(phase: &'static str) -> std::result::Result { + return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, phase)); +} + +#[cfg(test)] +#[path = "../unit_tests/schema_v003.rs"] +mod tests; diff --git a/crates/ksp-store-postgres-lib/tests/v003_migration_completeness.rs b/crates/ksp-store-postgres-lib/tests/v003_migration_completeness.rs new file mode 100644 index 0000000..018e328 --- /dev/null +++ b/crates/ksp-store-postgres-lib/tests/v003_migration_completeness.rs @@ -0,0 +1,56 @@ +// file: crates/ksp-store-postgres-lib/tests/v003_migration_completeness.rs +// version: 1 + +#![warn(missing_docs)] +#![deny(unreachable_pub)] +#![forbid(unsafe_code)] + +//! V003 additive migration and frozen-history canaries. + +#[test] +fn pre_003_v003_extension_is_wired_without_modifying_frozen_registry_sources() { + let crate_root = include_str!("../src/lib.rs"); + let legacy_migration = include_str!("../src/migration.rs"); + let legacy_schema = include_str!("../src/schema.rs"); + let extension = include_str!("../src/migration_v003.rs"); + assert!(crate_root.contains("mod migration_v003;")); + assert!(crate_root.contains("mod schema_v003;")); + assert!(crate_root.contains("pub(crate) use self::migration_v003::bootstrap;")); + assert!(crate_root.contains("pub(crate) use self::migration_v003::current_migration_version;")); + assert!(legacy_migration.starts_with("// file: crates/ksp-store-postgres-lib/src/migration.rs\n// version: 10\n")); + assert!(legacy_schema.starts_with("// file: crates/ksp-store-postgres-lib/src/schema.rs\n// version: 9\n")); + assert!(!legacy_migration.contains("V003_RESOURCES")); + assert!(!legacy_schema.contains("v003_raw_transaction_variants")); + for checksum in [ + "d29068b8c13b9dc0cc9ef6aaadd0fa12d41e0fe4c56541a1118c4bfc846a1450", + "31488cda2f08f3f46c4cdbdbb6c18c243662fada02eac4487040c8735d72cc51", + "ff21605ed45f7ab4c0f92bbb692700b4118a9488b04d50a31d259ac59bdb550e", + ] { + assert!(extension.contains(checksum), "missing frozen history checksum canary: {checksum}"); + } + return; +} + +#[test] +fn pre_003_v003_sources_remain_backend_private_secret_safe_and_without_facade_edges() { + let migration = include_str!("../src/migration_v003.rs"); + let schema = include_str!("../src/schema_v003.rs"); + let combined = std::format!("{migration}\n{schema}"); + for forbidden in ["std::env", "dotenv", "KSP_SECRET_", "PGPASSWORD", "ksp_store_lib", "ksp_config_lib", "sqlx::", "unwrap()", "expect(", "unsafe "] { + assert!(!combined.contains(forbidden), "forbidden V003 migration dependency/material detected: {forbidden}"); + } + return; +} + +#[test] +fn pre_003_v003_migration_is_schema_only_and_defers_business_backfill() { + let schema = include_str!("../src/schema_v003.rs"); + assert!(schema.contains("ksp_raw_transaction_variants")); + assert!(schema.contains("ksp_raw_transaction_canonical_selectors")); + assert!(schema.contains("ksp_raw_transaction_observation_variants")); + assert!(schema.contains("ksp_raw_transaction_conflicts")); + for forbidden in ["INSERT INTO ksp_raw_transaction_variants", "INSERT INTO ksp_raw_transaction_observation_variants", "UPDATE ksp_raw_transactions"] { + assert!(!schema.contains(forbidden), "V003 source unexpectedly performs business backfill: {forbidden}"); + } + return; +} diff --git a/crates/ksp-store-postgres-lib/unit_tests/migration.rs b/crates/ksp-store-postgres-lib/unit_tests/migration.rs index 6522eec..2bbd566 100644 --- a/crates/ksp-store-postgres-lib/unit_tests/migration.rs +++ b/crates/ksp-store-postgres-lib/unit_tests/migration.rs @@ -1,5 +1,5 @@ // file: crates/ksp-store-postgres-lib/unit_tests/migration.rs -// version: 8 +// version: 9 fn applied(version: i64, name: &str, checksum: &str) -> super::AppliedMigration { return super::AppliedMigration { checksum: checksum.to_owned(), name: name.to_owned(), version }; @@ -51,7 +51,7 @@ fn pre_003_v002_registry_is_complete_and_keeps_v000_v001_checksums_stable() { assert_eq!(v002.resources.len(), 32); assert_eq!(super::migration_checksum(v002), "ff21605ed45f7ab4c0f92bbb692700b4118a9488b04d50a31d259ac59bdb550e"); assert!(super::validate_embedded_registry(super::EMBEDDED_MIGRATIONS).is_ok()); - assert_eq!(crate::current_migration_version(), 2); + assert_eq!(crate::current_migration_version(), 3); let full = [ applied(0, v000.name, super::migration_checksum(v000).as_str()), applied(1, v001.name, super::migration_checksum(v001).as_str()), diff --git a/crates/ksp-store-postgres-lib/unit_tests/migration_v003.rs b/crates/ksp-store-postgres-lib/unit_tests/migration_v003.rs new file mode 100644 index 0000000..d783fb6 --- /dev/null +++ b/crates/ksp-store-postgres-lib/unit_tests/migration_v003.rs @@ -0,0 +1,87 @@ +// file: crates/ksp-store-postgres-lib/unit_tests/migration_v003.rs +// version: 2 + +fn applied(version: i64, name: &str, checksum: &str) -> super::AppliedMigration { + return super::AppliedMigration { checksum: checksum.to_owned(), name: name.to_owned(), version }; +} + +#[test] +fn pre_003_v003_extension_preserves_frozen_history_and_advances_current_version() { + assert_eq!(crate::current_migration_version(), 3); + assert_eq!(super::V000_CHECKSUM, "d29068b8c13b9dc0cc9ef6aaadd0fa12d41e0fe4c56541a1118c4bfc846a1450"); + assert_eq!(super::V001_CHECKSUM, "31488cda2f08f3f46c4cdbdbb6c18c243662fada02eac4487040c8735d72cc51"); + assert_eq!(super::V002_CHECKSUM, "ff21605ed45f7ab4c0f92bbb692700b4118a9488b04d50a31d259ac59bdb550e"); + let prefix = [ + applied(0, "bootstrap", super::V000_CHECKSUM), + applied(1, "raw_transaction", super::V001_CHECKSUM), + applied(2, "raw_account_state", super::V002_CHECKSUM), + ]; + assert_eq!(super::validate_history(&prefix).ok(), std::option::Option::Some(3)); + return; +} + +#[test] +fn pre_003_v003_resource_checksum_and_history_are_stable() { + let checksum = super::migration_checksum_v003(); + assert_eq!(checksum, "3c9cf41877c96944a378abb41ee7a8d8ab7f99fc6dc2bb7a26c295cf8f9fc914"); + let full = [ + applied(0, "bootstrap", super::V000_CHECKSUM), + applied(1, "raw_transaction", super::V001_CHECKSUM), + applied(2, "raw_account_state", super::V002_CHECKSUM), + applied(3, super::V003_NAME, checksum.as_str()), + ]; + assert_eq!(super::validate_history(&full).ok(), std::option::Option::Some(4)); + return; +} + +#[test] +fn pre_003_v003_history_rejects_divergence_and_future_versions() { + let checksum = super::migration_checksum_v003(); + let divergent = [ + applied(0, "bootstrap", super::V000_CHECKSUM), + applied(1, "raw_transaction", super::V001_CHECKSUM), + applied(2, "raw_account_state", super::V002_CHECKSUM), + applied(3, super::V003_NAME, "changed"), + ]; + let divergent_result = super::validate_history(&divergent); + assert_eq!(divergent_result.err().map(|value| return value.kind()), std::option::Option::Some(crate::PostgresBackendErrorKind::MigrationMismatch)); + let future = [ + applied(0, "bootstrap", super::V000_CHECKSUM), + applied(1, "raw_transaction", super::V001_CHECKSUM), + applied(2, "raw_account_state", super::V002_CHECKSUM), + applied(3, super::V003_NAME, checksum.as_str()), + applied(4, "future", "future"), + ]; + let future_result = super::validate_history(&future); + assert_eq!(future_result.err().map(|value| return value.kind()), std::option::Option::Some(crate::PostgresBackendErrorKind::SchemaNewer)); + let direct_future = [ + applied(0, "bootstrap", super::V000_CHECKSUM), + applied(1, "raw_transaction", super::V001_CHECKSUM), + applied(2, "raw_account_state", super::V002_CHECKSUM), + applied(4, "future", "future"), + ]; + let direct_future_result = super::validate_history(&direct_future); + assert_eq!( + direct_future_result.err().map(|value| return value.kind()), + std::option::Option::Some(crate::PostgresBackendErrorKind::SchemaNewer), + ); + return; +} + +#[test] +fn pre_003_v003_bootstrap_uses_legacy_engine_only_before_v002_and_absorbs_completed_v003_race() { + assert!(super::legacy_bootstrap_required(std::option::Option::None)); + assert!(super::legacy_bootstrap_required(std::option::Option::Some(0))); + assert!(super::legacy_bootstrap_required(std::option::Option::Some(1))); + assert!(!super::legacy_bootstrap_required(std::option::Option::Some(2))); + assert!(!super::legacy_bootstrap_required(std::option::Option::Some(3))); + assert!(super::legacy_schema_newer_is_completed_v003(std::option::Option::Some(3))); + assert!(!super::legacy_schema_newer_is_completed_v003(std::option::Option::Some(2))); + assert!(!super::legacy_schema_newer_is_completed_v003(std::option::Option::Some(4))); + assert!(super::v003_apply_allowed(false, false)); + assert!(super::v003_apply_allowed(false, true)); + assert!(super::v003_apply_allowed(true, true)); + assert!(!super::v003_apply_allowed(true, false)); + return; +} + diff --git a/crates/ksp-store-postgres-lib/unit_tests/schema_v003.rs b/crates/ksp-store-postgres-lib/unit_tests/schema_v003.rs new file mode 100644 index 0000000..4f128e0 --- /dev/null +++ b/crates/ksp-store-postgres-lib/unit_tests/schema_v003.rs @@ -0,0 +1,54 @@ +// file: crates/ksp-store-postgres-lib/unit_tests/schema_v003.rs +// version: 2 + +#[test] +fn pre_003_v003_resource_inventory_is_exact_and_ordered() { + let resources = crate::V003_RESOURCES; + assert_eq!(resources.len(), 25); + let ids = resources.iter().map(|resource| return resource.id).collect::>(); + assert_eq!(ids.iter().filter(|id| return id.starts_with("tables/")).count(), 4); + assert_eq!(ids.iter().filter(|id| return id.starts_with("constraints/")).count(), 18); + assert_eq!(ids.iter().filter(|id| return id.starts_with("indexes/")).count(), 3); + assert!(ids[..4].iter().all(|id| return id.starts_with("tables/"))); + assert!(ids[4..22].iter().all(|id| return id.starts_with("constraints/"))); + assert!(ids[22..].iter().all(|id| return id.starts_with("indexes/"))); + let mut unique = std::collections::BTreeSet::<&str>::new(); + for id in ids { + assert!(unique.insert(id), "duplicate V003 resource id: {id}"); + } + return; +} + +#[test] +fn pre_003_v003_schema_is_additive_payload_bounded_and_hash_non_unique() { + let sql = crate::V003_RESOURCES.iter().map(|resource| return resource.sql).collect::>().concat(); + for required in [ + "CREATE TABLE IF NOT EXISTS ksp_raw_transaction_variants", + "CREATE TABLE IF NOT EXISTS ksp_raw_transaction_canonical_selectors", + "CREATE TABLE IF NOT EXISTS ksp_raw_transaction_observation_variants", + "CREATE TABLE IF NOT EXISTS ksp_raw_transaction_conflicts", + "PRIMARY KEY (transaction_signature, variant_id)", + "FOREIGN KEY (transaction_signature) REFERENCES ksp_raw_transactions(signature)", + "FOREIGN KEY (observation_key) REFERENCES ksp_raw_transaction_observations(observation_key)", + "octet_length(payload) <= 16777216", + "origin_kind = 'native' OR origin_kind = 'synthetic'", + "latest_relation = 'conflict' OR latest_relation = 'incomparable'", + "latest_reason_code = 'content_hash_collision'", + "latest_reason_code = 'unsupported_canonical_difference'", + ] { + assert!(sql.contains(required), "missing V003 schema invariant: {required}"); + } + for forbidden in ["UNIQUE (content_hash", "UNIQUE(content_hash", "BIGSERIAL", "INSERT INTO", "UPDATE ksp_raw_transactions"] { + assert!(!sql.contains(forbidden), "forbidden V003 migration behavior: {forbidden}"); + } + return; +} + +#[test] +fn pre_003_v003_observation_mapping_does_not_fabricate_legacy_rows() { + let sql = crate::V003_RESOURCES.iter().map(|resource| return resource.sql).collect::>().concat(); + assert!(sql.contains("ksp_raw_transaction_observation_variants")); + assert!(!sql.contains("SELECT observation_key FROM ksp_raw_transaction_observations")); + assert!(!sql.contains("INSERT INTO ksp_raw_transaction_observation_variants")); + return; +} diff --git a/deltas/0.3.16/pre.003-fix.001.md b/deltas/0.3.16/pre.003-fix.001.md new file mode 100644 index 0000000..0cde84f --- /dev/null +++ b/deltas/0.3.16/pre.003-fix.001.md @@ -0,0 +1,68 @@ + + + +# Delta `0.3.16-pre.003-fix.001` — conformité Rust de V003 + +## Base requise + +```text +0.3.16-pre.003 appliquée +workspace.package.version = 0.3.16-pre.3 +``` + +Le gate opérateur de `pre.003` confirme que `cargo check --workspace` et les suites ciblées `ksp-store-postgres-lib` passent, y compris les 82 tests unitaires, les tests de frontière, de hardening et les trois canaris V003. Le gate reste toutefois rouge avant ce fix : 31 violations de l'audit Rust et un `unused import` transformé en erreur par Clippy `-D warnings`. + +## Version + +Cette correction modifie du Rust : + +```text +workspace.package.version = 0.3.16-pre.3.fix.1 +``` + +## Corrections + +Le fix est strictement structurel et ne modifie ni le schéma SQL V003 ni son comportement métier : + +- les blocs `pub use` et `pub(crate) use` de `ksp-store-postgres-lib` sont séparés et ordonnés selon les règles du workspace ; +- les fonctions legacy `migration::bootstrap` et `migration::current_migration_version` sont réexportées à la racine sous les alias privés `crate::legacy_bootstrap` et `crate::legacy_current_migration_version` ; +- `migration_v003.rs` ne traverse plus directement le module privé `crate::migration` ; +- `V003SchemaResource` reste réexporté à la racine et est réellement utilisé via `crate::V003SchemaResource` dans le calcul du checksum, au lieu d'être supprimé ou masqué par un `allow` ; +- les tests V003 utilisent `crate::current_migration_version` et `crate::V003_RESOURCES` lorsqu'ils ciblent des items `pub(crate)` ; +- les champs `pub(crate)` de `V003SchemaResource` reçoivent leur rustdoc adjacente ; +- les blocs homogènes de constantes sont remis au format exact attendu par `RUST-FMT-112` ; +- aucune annotation `#[allow(unused_imports)]` et aucun marquage artificiel `#[cfg(test)]` ne sont ajoutés. + +## Invariants conservés + +Aucune ressource sous `migrations/v003_raw_transaction_variants/` n'est incluse ou modifiée par ce fix. + +Le checksum V003 reste donc : + +```text +3c9cf41877c96944a378abb41ee7a8d8ab7f99fc6dc2bb7a26c295cf8f9fc914 +``` + +Les checksums V000/V001/V002 et les sources historiques `src/migration.rs` / `src/schema.rs` restent inchangés. + +Le fix ne modifie pas les quatre tables, les dix-huit contraintes, les trois index, la politique de backfill, la sémantique `content_hash`, ni le comportement de concurrence V002 -> V003. + +## Validation demandée + +```bash +cargo fmt --all +cargo fmt --all -- --check + +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas + +cargo check --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test -p ksp-store-postgres-lib --all-targets --all-features +``` + +Les trois tests PostgreSQL live restent opt-in comme dans `pre.003`. + +## Suite + +Après gate propre : `0.3.16-pre.004` — backend PostgreSQL V003, bootstrap paresseux de l'identité, ledger de variantes, rattachement exact des nouvelles observations, idempotence et concurrence d'insertion. diff --git a/deltas/0.3.16/pre.003.md b/deltas/0.3.16/pre.003.md new file mode 100644 index 0000000..1f2179d --- /dev/null +++ b/deltas/0.3.16/pre.003.md @@ -0,0 +1,75 @@ + + + +# Delta `0.3.16-pre.003` — PostgreSQL V003 RAW transaction variants + +## Base requise + +```text +0.3.16-pre.002-fix.001 appliquée +workspace.package.version = 0.3.16-pre.2.fix.1 +``` + +Le gate opérateur de la base est confirmé propre : audits Rust/export/KSP/Markdown, `cargo check --workspace`, Clippy `-D warnings` et toutes les suites `ksp-store-api` passent. + +## Version + +Cette tranche contient du Rust et du schéma PostgreSQL : + +```text +workspace.package.version = 0.3.16-pre.3 +``` + +## Portée + +Cette tranche livre uniquement la fondation physique V003. Elle ne branche encore aucune écriture métier multi-variante. + +V003 ajoute : + +- `ksp_raw_transaction_variants` ; +- `ksp_raw_transaction_canonical_selectors` ; +- `ksp_raw_transaction_observation_variants` ; +- `ksp_raw_transaction_conflicts` ; +- dix-huit contraintes physiques et trois index non uniques ; +- un checksum V003 stable calculé par le même domaine `ksp-migration-resources-v1` ; +- un bootstrap d'extension privé qui conserve le moteur V000-V002 et applique V003 avec la même clé et le même protocole d'advisory-lock ; +- une garde de concurrence qui n'appelle plus le moteur legacy lorsqu'une base est déjà V002 et absorbe uniquement la course où un pair vient de terminer V003 ; +- une politique `schema_autoupdate=false` qui valide d'abord l'historique existant avant de rendre `migration_pending`, tout en autorisant la création complète d'une base fraîche ; +- des canaris unitaires et d'intégration pour la version courante, le checksum, l'ordre des resources et la frontière de migration. + +## Invariants + +`src/migration.rs`, `src/schema.rs` et les ressources SQL V000/V001/V002 ne sont pas modifiés par ce delta. + +Les checksums historiques restent : + +```text +V000 d29068b8c13b9dc0cc9ef6aaadd0fa12d41e0fe4c56541a1118c4bfc846a1450 +V001 31488cda2f08f3f46c4cdbdbb6c18c243662fada02eac4487040c8735d72cc51 +V002 ff21605ed45f7ab4c0f92bbb692700b4118a9488b04d50a31d259ac59bdb550e +V003 3c9cf41877c96944a378abb41ee7a8d8ab7f99fc6dc2bb7a26c295cf8f9fc914 +``` + +`content_hash` n'est jamais `UNIQUE` et ne constitue pas une preuve d'égalité métier. + +La migration V003 est schema-only : aucun `INSERT` de variantes, aucun rattachement artificiel des observations historiques et aucun `UPDATE` de la projection V001 n'est réalisé. Le bootstrap métier des identités existantes appartient à `pre.004`. + +## Validation demandée + +```bash +cargo fmt --all +cargo fmt --all -- --check + +python3 scripts/audit_rust_workspace_rules.py +python3 scripts/audit_markdown_tables.py README.md RULES.md ROADMAP.md CHANGELOG.md docs prompts crates deltas + +cargo check --workspace +cargo clippy --workspace --all-targets --all-features -- -D warnings +cargo test -p ksp-store-postgres-lib --all-targets --all-features +``` + +Les tests PostgreSQL live restent opt-in et ne sont pas requis pour ce gate de schéma statique ; `pre.004` ajoutera les preuves backend de bootstrap/écriture sur base réelle. + +## Suite + +Après gate propre : `0.3.16-pre.004` — backend PostgreSQL V003, bootstrap paresseux de l'identité, ledger de variantes, rattachement exact des nouvelles observations, idempotence et concurrence d'insertion. diff --git a/docs/plans/038-V0_3_16_RAW_RESILIENCE_CONFLICT_PLAN.md b/docs/plans/038-V0_3_16_RAW_RESILIENCE_CONFLICT_PLAN.md index 39815ef..c9a7536 100644 --- a/docs/plans/038-V0_3_16_RAW_RESILIENCE_CONFLICT_PLAN.md +++ b/docs/plans/038-V0_3_16_RAW_RESILIENCE_CONFLICT_PLAN.md @@ -1,5 +1,5 @@ - + # Plan `0.3.16` -> `0.3.18` — résilience RAW, variantes, conflits et récupération @@ -405,9 +405,18 @@ same hash Aucune contrainte d'unicité ne doit transformer automatiquement `(signature, content_hash)` en preuve métier. La sérialisation par verrou de l'identité permet de prévenir les doubles insertions concurrentes après comparaison exacte. -## 7. Structures V003 conceptuelles retenues +## 7. Structures V003 retenues -Les noms SQL exacts seront figés dans la tranche de migration, mais les rôles sont désormais décidés. +`0.3.16-pre.003` fige les quatre surfaces physiques minimales suivantes : + +```text +ksp_raw_transaction_variants +ksp_raw_transaction_canonical_selectors +ksp_raw_transaction_observation_variants +ksp_raw_transaction_conflicts +``` + +Le journal append-only complet, les participants de conflit et les parents de variantes synthétiques restent volontairement différés aux tranches `0.3.17` prévues ; V003 réserve ici uniquement les surfaces nécessaires à la vertical slice `0.3.16`. ### 7.1 Variant ledger @@ -519,7 +528,9 @@ Elle ne réécrit jamais A ou B et ne se présente jamais comme une observation V000/V001/V002 restent byte-identiques. -V003 ajoute des ressources nouvelles et, si nécessaire, des index/contraintes nouvelles via une migration indépendante enregistrée avec son propre checksum. +V003 ajoute des ressources nouvelles via une migration indépendante enregistrée avec son propre checksum. `0.3.16-pre.003` conserve en outre `src/migration.rs`, `src/schema.rs` et toutes les ressources V000/V001/V002 inchangés : le bootstrap V003 est une extension privée qui délègue d'abord au moteur historique V000-V002 puis applique/valide V003 avec la même clé et le même protocole d'advisory-lock KSP. Cette séparation fournit un canari structurel supplémentaire contre toute dérive des anciens checksums. + +La délégation au moteur historique est limitée aux historiques antérieurs à V002. Une base déjà V002 passe directement sous le verrou V003 ; cela évite qu'un second processus concurrent relise V003 comme une version future du moteur legacy. Si une course sur une base plus ancienne se termine par `SchemaNewer`, elle n'est absorbée que si le registre est alors exactement V003, puis tous les resources et l'identité sont revérifiés. ### 8.2 Bootstrap des identités existantes @@ -955,11 +966,11 @@ Gate d'audit, architecture, sizing, plan et validation. Cette tranche est déjà #### `0.3.16-pre.002` -Contrats Store API backend-neutral : variant identity, relation de qualité, outcomes de persistance, reason codes et contrats minimaux nécessaires au conflit durable. Implémentation livrée par cette tranche ; gate opérateur requis avant `pre.003`. +Contrats Store API backend-neutral : variant identity, relation de qualité, outcomes de persistance, reason codes et contrats minimaux nécessaires au conflit durable. Implémentation livrée et gate opérateur confirmé propre après `pre.002-fix.001`. #### `0.3.16-pre.003` -Migration V003 : registre, resources et schéma multi-variantes/selector/conflict minimal ; canaris garantissant l'immuabilité byte/checksum de V000/V001/V002. +Migration V003 : registre d'extension, resources et schéma multi-variantes/selector/conflict minimal ; canaris garantissant l'immuabilité byte/checksum de V000/V001/V002. Implémentation livrée par cette tranche ; gate opérateur requis avant `pre.004`. #### `0.3.16-pre.004`