Sasedev/khadhroony-solana-project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

0.3.16-pre.3.fix.1

Browse Source
This commit is contained in:
SinuS von SifriduS
2026-09-21 10:33:06 +02:00
parent 05cdb4e2ab
commit 540b1bc305
36 changed files with 1852 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/001_pk_ksp_raw_transaction_variants.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/002_ck_ksp_raw_transaction_variants_identity.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/003_ck_ksp_raw_transaction_variants_origin.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/004_ck_ksp_raw_transaction_variants_slot_time.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/005_ck_ksp_raw_transaction_variants_format.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/006_ck_ksp_raw_transaction_variants_payload.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/007_fk_ksp_raw_transaction_variants_transaction.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/008_pk_ksp_raw_transaction_canonical_selectors.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/009_ck_ksp_raw_transaction_canonical_selectors_bounds.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/010_fk_ksp_raw_transaction_canonical_selectors_variant.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/011_pk_ksp_raw_transaction_observation_variants.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/012_ck_ksp_raw_transaction_observation_variants_bounds.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/013_fk_ksp_raw_transaction_observation_variants_observation.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/014_fk_ksp_raw_transaction_observation_variants_variant.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/015_pk_ksp_raw_transaction_conflicts.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/016_ck_ksp_raw_transaction_conflicts_bounds.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/017_fk_ksp_raw_transaction_conflicts_canonical_variant.sql Normal file
View File

@@ -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$;

12
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/constraints/018_fk_ksp_raw_transaction_conflicts_incoming_variant.sql Normal file
View File

@@ -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$;

2
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/001_ix_ksp_raw_transaction_variants_signature_hash_variant.sql Normal file
View File

@@ -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);

2
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/002_ix_ksp_raw_transaction_observation_variants_variant.sql Normal file
View File

@@ -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);

2
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/indexes/003_ix_ksp_raw_transaction_conflicts_status_updated_signature.sql Normal file
View File

@@ -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);

24
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/001_ksp_raw_transaction_variants.sql Normal file
View File

@@ -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;

10
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/002_ksp_raw_transaction_canonical_selectors.sql Normal file
View File

@@ -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;

10
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/003_ksp_raw_transaction_observation_variants.sql Normal file
View File

@@ -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;

20
crates/ksp-store-postgres-lib/migrations/v003_raw_transaction_variants/tables/004_ksp_raw_transaction_conflicts.sql Normal file
View File

@@ -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;

66
crates/ksp-store-postgres-lib/src/lib.rs
View File

@@ -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;

525
crates/ksp-store-postgres-lib/src/migration_v003.rs Normal file
View File

@@ -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::<usize, bool>(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::<usize, i16>(0);
let network_result = row.try_get::<usize, std::string::String>(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<bool, crate::PostgresBackendError> {
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::<usize, bool>(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<std::option::Option<i64>, 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::<usize, std::option::Option<i64>>(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<std::vec::Vec<AppliedMigration>, 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::<usize, i64>(0);
let name = row.try_get::<usize, std::string::String>(1);
let checksum = row.try_get::<usize, std::string::String>(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<i64, crate::PostgresBackendError> {
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<i64>) -> 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<i64>) -> 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;

666
crates/ksp-store-postgres-lib/src/schema_v003.rs Normal file
View File

@@ -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<i32>,
numeric_scale: std::option::Option<i32>,
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<i32>,
numeric_scale: std::option::Option<i32>,
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<V003SchemaResourceState, crate::PostgresBackendError> {
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<V003SchemaResourceState, crate::PostgresBackendError> {
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::<usize, std::string::String>(0);
let validated = row.try_get::<usize, bool>(1);
let deferrable = row.try_get::<usize, bool>(2);
let deferred = row.try_get::<usize, bool>(3);
let definition = row.try_get::<usize, std::string::String>(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<V003SchemaResourceState, crate::PostgresBackendError> {
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::<usize, bool>(0);
let access_method = row.try_get::<usize, std::string::String>(1);
let definition = row.try_get::<usize, std::string::String>(2);
let predicate = row.try_get::<usize, std::option::Option<std::string::String>>(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<V003SchemaResourceState, crate::PostgresBackendError> {
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::<usize, std::string::String>(0);
let row_security = table_row.try_get::<usize, bool>(1);
let force_row_security = table_row.try_get::<usize, bool>(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::<ActualColumn>::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<std::string::String>,
generated: std::string::String,
identity: std::string::String,
name: std::string::String,
nullable: bool,
numeric_precision: std::option::Option<i32>,
numeric_scale: std::option::Option<i32>,
udt_name: std::string::String,
}
impl ActualColumn {
fn from_row(row: &tokio_postgres::Row) -> std::result::Result<Self, crate::PostgresBackendError> {
let name = row.try_get::<usize, std::string::String>(0);
let udt_name = row.try_get::<usize, std::string::String>(1);
let nullable = row.try_get::<usize, bool>(2);
let numeric_precision = row.try_get::<usize, std::option::Option<i32>>(3);
let numeric_scale = row.try_get::<usize, std::option::Option<i32>>(4);
let default = row.try_get::<usize, std::option::Option<std::string::String>>(5);
let identity = row.try_get::<usize, std::string::String>(6);
let generated = row.try_get::<usize, std::string::String>(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<std::string::String> {
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::<std::string::String>()
.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<T>(phase: &'static str) -> std::result::Result<T, crate::PostgresBackendError> {
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationMismatch, phase));
}
fn schema_query_error<T>(phase: &'static str) -> std::result::Result<T, crate::PostgresBackendError> {
return std::result::Result::Err(crate::PostgresBackendError::new(crate::PostgresBackendErrorKind::MigrationFailed, phase));
}
#[cfg(test)]
#[path = "../unit_tests/schema_v003.rs"]
mod tests;

56
crates/ksp-store-postgres-lib/tests/v003_migration_completeness.rs Normal file
View File

@@ -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;
}

4
crates/ksp-store-postgres-lib/unit_tests/migration.rs
View File

@@ -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()),

87
crates/ksp-store-postgres-lib/unit_tests/migration_v003.rs Normal file
View File

@@ -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;
}

54
crates/ksp-store-postgres-lib/unit_tests/schema_v003.rs Normal file
View File

@@ -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::<std::vec::Vec<_>>();
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::<std::vec::Vec<_>>().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::<std::vec::Vec<_>>().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;
}