diff --git a/tests/adapters/persistence/test_migration_and_drift.py b/tests/adapters/persistence/test_migration_and_drift.py index e0aa8ca..9a2a312 100644 --- a/tests/adapters/persistence/test_migration_and_drift.py +++ b/tests/adapters/persistence/test_migration_and_drift.py @@ -27,6 +27,7 @@ _EXPECTED_TABLES = { "snapshot_pause", # BAT-008 / REQ-032 "broker_account_snapshot", # dcy-041 / GitHub #45 "ledger_skip", # GitHub #68 skipped-ledger durable audit + "reconcile_notification_state", # req-0007 / ac-0024 } diff --git a/tests/adapters/portfolio/test_reconcile_notification_store.py b/tests/adapters/portfolio/test_reconcile_notification_store.py new file mode 100644 index 0000000..5e93511 --- /dev/null +++ b/tests/adapters/portfolio/test_reconcile_notification_store.py @@ -0,0 +1,213 @@ +"""PostgreSQLReconcileNotificationStore integration tests (req-0007, ac-0024). + +Verifies migration 026 schema creation (idempotent replay), full CRUD +round-trip, and lifecycle semantics against a real PostgreSQL 16 instance. + +Gated on TREX_POSTGRESQL_TEST_DSN — skipped in unit-test-only CI. +""" +from __future__ import annotations + +import pytest + +from trex_core.application.reconciliation import _c_identity_key + + +@pytest.fixture +def pg_notify_store(pg_clean): + from trex_core.adapters.portfolio.reconcile_notification_store import ( + PostgreSQLReconcileNotificationStore, + ) + return PostgreSQLReconcileNotificationStore(pg_clean) + + +# --------------------------------------------------------------------------- +# Migration 026 schema +# --------------------------------------------------------------------------- + +class TestMigration026: + def test_table_exists_after_migration(self, pg_clean): + """Migration 026 creates reconcile_notification_state (IF NOT EXISTS).""" + import psycopg + with psycopg.connect(pg_clean) as conn, conn.cursor() as cur: + cur.execute( + "SELECT table_name FROM information_schema.tables " + "WHERE table_schema = 'public' AND table_name = 'reconcile_notification_state'" + ) + assert cur.fetchone() is not None, "reconcile_notification_state table not found" + + def test_table_has_expected_columns(self, pg_clean): + """Migration 026 creates all required columns.""" + import psycopg + with psycopg.connect(pg_clean) as conn, conn.cursor() as cur: + cur.execute( + "SELECT column_name FROM information_schema.columns " + "WHERE table_schema = 'public' AND table_name = 'reconcile_notification_state'" + ) + cols = {r[0] for r in cur.fetchall()} + expected = { + "identity_key", "case_code", "instance_key", "symbol", + "quantity_a", "quantity_b", "lifecycle_state", "created_at", "updated_at", + } + assert expected <= cols, f"Missing columns: {expected - cols}" + + def test_migration_is_idempotent(self, pg_clean): + """Running migration 026 twice does not fail (IF NOT EXISTS guard).""" + from trex_core.adapters.persistence.migrate import apply_migrations + # pg_clean already ran apply_migrations once; run again to verify idempotency. + apply_migrations(pg_clean) # must not raise + + +# --------------------------------------------------------------------------- +# CRUD round-trip +# --------------------------------------------------------------------------- + +class TestCRUDRoundTrip: + def test_set_and_get_state_b_case(self, pg_notify_store): + """set_state → get_state round-trip for B-case (slot_id identity).""" + pg_notify_store.set_state( + "slot-uuid-001", + case_code="B", + instance_key="inst-A", + symbol="005930", + quantity_a=10, + quantity_b=7, + lifecycle_state="notified", + ) + state = pg_notify_store.get_state("slot-uuid-001") + assert state is not None + assert state["case_code"] == "B" + assert state["instance_key"] == "inst-A" + assert state["symbol"] == "005930" + assert state["quantity_a"] == 10 + assert state["quantity_b"] == 7 + assert state["lifecycle_state"] == "notified" + + def test_set_and_get_state_c_case(self, pg_notify_store): + """set_state → get_state round-trip for C-case (SHA-256 digest identity).""" + c_key = _c_identity_key("inst-A", "005930") + pg_notify_store.set_state( + c_key, + case_code="C", + instance_key="inst-A", + symbol="005930", + quantity_a=50, + quantity_b=42, + lifecycle_state="notified", + ) + state = pg_notify_store.get_state(c_key) + assert state is not None + assert state["case_code"] == "C" + assert state["symbol"] == "005930" + assert state["quantity_a"] == 50 + + def test_set_state_upsert(self, pg_notify_store): + """set_state is an upsert — second call with same key updates the row.""" + pg_notify_store.set_state( + "slot-uuid-002", + case_code="B", + instance_key="inst-A", + symbol="005930", + quantity_a=10, + quantity_b=7, + lifecycle_state="notified", + ) + pg_notify_store.set_state( + "slot-uuid-002", + case_code="B", + instance_key="inst-A", + symbol="005930", + quantity_a=10, + quantity_b=9, # quantity changed + lifecycle_state="notified", + ) + state = pg_notify_store.get_state("slot-uuid-002") + assert state["quantity_b"] == 9 + + def test_get_state_returns_none_for_missing_key(self, pg_notify_store): + assert pg_notify_store.get_state("does-not-exist") is None + + def test_clear_state_removes_row(self, pg_notify_store): + """clear_state deletes the row; subsequent get_state returns None.""" + pg_notify_store.set_state( + "slot-uuid-003", + case_code="B", + instance_key="inst-A", + symbol="005930", + quantity_a=10, + quantity_b=7, + lifecycle_state="notified", + ) + pg_notify_store.clear_state("slot-uuid-003") + assert pg_notify_store.get_state("slot-uuid-003") is None + + def test_list_states_by_lifecycle(self, pg_notify_store): + """list_states_by_lifecycle returns only rows with the given state.""" + pg_notify_store.set_state( + "slot-A", + case_code="B", instance_key="inst-A", symbol="005930", + quantity_a=10, quantity_b=7, lifecycle_state="notified", + ) + pg_notify_store.set_state( + "slot-B", + case_code="B", instance_key="inst-A", symbol="005930", + quantity_a=20, quantity_b=15, lifecycle_state="pending_resolved", + ) + notified = pg_notify_store.list_states_by_lifecycle("notified") + pending = pg_notify_store.list_states_by_lifecycle("pending_resolved") + assert len(notified) == 1 + assert notified[0]["identity_key"] == "slot-A" + assert len(pending) == 1 + assert pending[0]["identity_key"] == "slot-B" + + def test_durability_across_store_instances(self, pg_clean): + """State survives a simulated process restart (new store instance, same DSN).""" + from trex_core.adapters.portfolio.reconcile_notification_store import ( + PostgreSQLReconcileNotificationStore, + ) + a = PostgreSQLReconcileNotificationStore(pg_clean) + a.set_state( + "slot-restart", + case_code="B", instance_key="inst-A", symbol="005930", + quantity_a=10, quantity_b=7, lifecycle_state="notified", + ) + # "restart" — fresh store instance + b = PostgreSQLReconcileNotificationStore(pg_clean) + state = b.get_state("slot-restart") + assert state is not None + assert state["lifecycle_state"] == "notified" + + def test_c_identity_keys_for_distinct_inputs_are_distinct(self, pg_notify_store): + """Two distinct (instance_key, symbol) pairs must yield distinct DB rows.""" + key_a = _c_identity_key("inst-A", "005930") + key_b = _c_identity_key("inst-A", "005931") + assert key_a != key_b + pg_notify_store.set_state( + key_a, case_code="C", instance_key="inst-A", symbol="005930", + quantity_a=10, quantity_b=8, lifecycle_state="notified", + ) + pg_notify_store.set_state( + key_b, case_code="C", instance_key="inst-A", symbol="005931", + quantity_a=20, quantity_b=15, lifecycle_state="notified", + ) + state_a = pg_notify_store.get_state(key_a) + state_b = pg_notify_store.get_state(key_b) + assert state_a["symbol"] == "005930" + assert state_b["symbol"] == "005931" + + def test_crafted_separator_collision_absent(self, pg_notify_store): + """Crafted-separator inputs ('a\\x1fC','b') vs ('a','C\\x1fb') map to distinct rows.""" + key_a = _c_identity_key("a\x1fC", "b") + key_b = _c_identity_key("a", "C\x1fb") + assert key_a != key_b + pg_notify_store.set_state( + key_a, case_code="C", instance_key="a\x1fC", symbol="b", + quantity_a=1, quantity_b=2, lifecycle_state="notified", + ) + pg_notify_store.set_state( + key_b, case_code="C", instance_key="a", symbol="C\x1fb", + quantity_a=3, quantity_b=4, lifecycle_state="notified", + ) + state_a = pg_notify_store.get_state(key_a) + state_b = pg_notify_store.get_state(key_b) + assert state_a["instance_key"] == "a\x1fC" + assert state_b["instance_key"] == "a" diff --git a/tests/application/test_reconcile_notification.py b/tests/application/test_reconcile_notification.py index a96e7f3..1c1de0d 100644 --- a/tests/application/test_reconcile_notification.py +++ b/tests/application/test_reconcile_notification.py @@ -845,16 +845,25 @@ def test_publish_alarm_returns_bool(): # Regression tests for ac-0024 blocker fixes # --------------------------------------------------------------------------- -def test_c_identity_key_contains_no_nul_byte(): - """_c_identity_key must not contain NUL (\\x00) — PostgreSQL TEXT bind would fail.""" +def test_c_identity_key_is_injective_and_postgres_safe(): + """_c_identity_key must be injective (no collisions) and PostgreSQL TEXT-safe. + + Specifically verifies the crafted-separator collision that bit the \\x1f + implementation: ('a\\x1fC', 'b') and ('a', 'C\\x1fb') previously produced + the same key and suppressed the second detected event. + """ key = _c_identity_key("my-instance", "005930") assert "\x00" not in key, f"identity key contains NUL: {key!r}" - # Must still be collision-resistant (different symbols must produce different keys) - key2 = _c_identity_key("my-instance", "005931") - assert key != key2 - # Must still be collision-resistant (different instances must produce different keys) - key3 = _c_identity_key("other-instance", "005930") - assert key != key3 + # Different symbols → different keys + assert key != _c_identity_key("my-instance", "005931") + # Different instances → different keys + assert key != _c_identity_key("other-instance", "005930") + # Crafted-separator collision must not exist + crafted_a = _c_identity_key("a\x1fC", "b") + crafted_b = _c_identity_key("a", "C\x1fb") + assert crafted_a != crafted_b, ( + f"Separator-crafted collision: both produce {crafted_a!r}" + ) def test_retry_pending_resolved_ignores_other_instance_states(): diff --git a/tests/conftest.py b/tests/conftest.py index 250933a..264e656 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,6 +26,7 @@ _PG_TABLES = ( "slot_transition_log", "daily_symbol_entries", "daily_slot", + "reconcile_notification_state", ) diff --git a/trex_core/adapters/persistence/sql/026_create_reconcile_notification_state.sql b/trex_core/adapters/persistence/sql/026_create_reconcile_notification_state.sql index 7593e74..ea54aaa 100644 --- a/trex_core/adapters/persistence/sql/026_create_reconcile_notification_state.sql +++ b/trex_core/adapters/persistence/sql/026_create_reconcile_notification_state.sql @@ -1,5 +1,5 @@ -- req-0007 / ac-0024: durable mismatch notification lifecycle state. --- PK: identity_key (slot_id TEXT for B-case; "{instance_key}\x00C\x00{symbol}" for C-case). +-- PK: identity_key (slot_id TEXT for B-case; "C:" length-prefix-digest for C-case). -- lifecycle_state: 'notified' | 'pending_resolved'. -- ON CONFLICT DO UPDATE: replay-safe; re-running migration is a no-op. CREATE TABLE IF NOT EXISTS reconcile_notification_state ( diff --git a/trex_core/adapters/portfolio/reconcile_notification_store.py b/trex_core/adapters/portfolio/reconcile_notification_store.py index fc2b0cd..eb176ba 100644 --- a/trex_core/adapters/portfolio/reconcile_notification_store.py +++ b/trex_core/adapters/portfolio/reconcile_notification_store.py @@ -4,7 +4,7 @@ Tracks per-identity B/C-case mismatch notification fingerprints so that duplicate alarms are suppressed across scheduler ticks and process restarts, and pending resolved transitions can be retried. -PK: identity_key (slot_id for B-case; "{instance_key}\\x1fC\\x1f{symbol}" for C-case). +PK: identity_key (slot_id TEXT for B-case; "C:" length-prefix-digest for C-case). lifecycle_state: 'notified' | 'pending_resolved'. """ from __future__ import annotations diff --git a/trex_core/application/reconciliation.py b/trex_core/application/reconciliation.py index 42fa628..aca2c5d 100644 --- a/trex_core/application/reconciliation.py +++ b/trex_core/application/reconciliation.py @@ -25,6 +25,7 @@ published via AlarmEmitter with durable deduplication through ReconcileNotificat """ from __future__ import annotations +import hashlib import logging import math import re @@ -105,12 +106,23 @@ def _normalize_legacy_cost_basis(value: object) -> tuple[float | None, str, str # --------------------------------------------------------------------------- def _c_identity_key(instance_key: str, symbol: str) -> str: - """Stable C-case notification identity key (PostgreSQL TEXT-safe, collision-resistant). + """Return an injective, PostgreSQL TEXT-safe key for a (instance_key, symbol) C-case pair. - Uses ASCII Unit Separator (\\x1f) which PostgreSQL TEXT accepts (only \\x00 is - forbidden) and which cannot appear in well-formed instance_key / symbol values. + Uses a length-prefixed SHA-256 hex digest so that *any* inputs — including + those containing separator characters — produce distinct keys with no + collision risk. The prefix ``C:`` makes the key case-code-scoped and + human-scannable in the DB; the digest is 64 hex chars (256-bit). + + Injectivity: the raw string fed to SHA-256 is ``"::"`` + where ``N = len(instance_key)``. The leading ``N:`` uniquely delimits + ``instance_key`` from the rest, so two distinct ``(instance_key, symbol)`` + pairs always produce different raw strings (and thus different digests). + + Example: _c_identity_key("a\\x1fC", "b") != _c_identity_key("a", "C\\x1fb") """ - return f"{instance_key}\x1fC\x1f{symbol}" + raw = f"{len(instance_key)}:{instance_key}:{symbol}" + digest = hashlib.sha256(raw.encode()).hexdigest() + return f"C:{digest}" def _build_alarm_fields(