feat(transaction): mempool observation, inclusion, 7d outcome and replacement models

Makefile

@@ -122,7 +122,7 @@ proto:
 	PASS=$${CLICKHOUSE_PASSWORD:-supersecret}; \
 	docker run --rm -v "$$(pwd):/workspace" \
 		--user "$$(id -u):$$(id -g)" \
-		--network xatu_xatu-net \
+		--network "$${XATU_CBT_PROJECT_NAME:-xatu-cbt-platform}_xatu-net" \
 		ethpandaops/clickhouse-proto-gen \
 		--dsn "clickhouse://$$USER:$$PASS@xatu-cbt-clickhouse-01:9000/$$NETWORK" \
 		--tables "admin_incremental,$$TABLES" \

migrations/101_int_transaction_mempool_observation_hourly.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS int_transaction_mempool_observation_hourly ON CLUSTER '{cluster}';
+DROP TABLE IF EXISTS int_transaction_mempool_observation_hourly_local ON CLUSTER '{cluster}';

migrations/101_int_transaction_mempool_observation_hourly.up.sql

@@ -0,0 +1,59 @@
+CREATE TABLE int_transaction_mempool_observation_hourly_local on cluster '{cluster}' (
+    `updated_date_time` DateTime COMMENT 'Timestamp when the record was last updated' CODEC(DoubleDelta, ZSTD(1)),
+    `hour_start_date_time` DateTime COMMENT 'The start of the hour bucket the sightings fall in' CODEC(DoubleDelta, ZSTD(1)),
+    `hash` FixedString(66) COMMENT 'The transaction hash' CODEC(ZSTD(1)),
+    `first_seen_date_time` DateTime64(3) COMMENT 'Earliest sighting of the transaction by any sentry within this hour' CODEC(DoubleDelta, ZSTD(1)),
+    `last_seen_date_time` DateTime64(3) COMMENT 'Latest sighting of the transaction by any sentry within this hour' CODEC(DoubleDelta, ZSTD(1)),
+    `sighting_count` UInt32 COMMENT 'Number of sightings of the transaction within this hour across all sentries' CODEC(ZSTD(1)),
+    `unique_sentries` UInt32 COMMENT 'Number of distinct sentries that sighted the transaction within this hour' CODEC(ZSTD(1)),
+    `from` FixedString(42) COMMENT 'The address of the account that sent the transaction' CODEC(ZSTD(1)),
+    `to` Nullable(FixedString(42)) COMMENT 'The address of the transaction recipient, null for contract creation' CODEC(ZSTD(1)),
+    `nonce` UInt64 COMMENT 'The nonce of the sender account at the time of the transaction' CODEC(ZSTD(1)),
+    `type` Nullable(UInt8) COMMENT 'The type of the transaction' CODEC(ZSTD(1)),
+    `gas` UInt64 COMMENT 'The maximum gas provided for the transaction execution' CODEC(ZSTD(1)),
+    `gas_price` UInt128 COMMENT 'The gas price of the transaction in wei' CODEC(ZSTD(1)),
+    `gas_tip_cap` Nullable(UInt128) COMMENT 'The priority fee (tip) the user has set for the transaction in wei' CODEC(ZSTD(1)),
+    `gas_fee_cap` Nullable(UInt128) COMMENT 'The max fee the user has set for the transaction in wei' CODEC(ZSTD(1)),
+    `value` UInt128 COMMENT 'The value transferred with the transaction in wei' CODEC(ZSTD(1)),
+    `size` UInt32 COMMENT 'The size of the transaction data in bytes' CODEC(ZSTD(1)),
+    `call_data_size` UInt32 COMMENT 'The size of the call data of the transaction in bytes' CODEC(ZSTD(1)),
+    `blob_gas` Nullable(UInt64) COMMENT 'The maximum gas provided for the blob transaction execution' CODEC(ZSTD(1)),
+    `blob_gas_fee_cap` Nullable(UInt128) COMMENT 'The max blob fee the user has set for the transaction in wei' CODEC(ZSTD(1)),
+    `blob_hashes` Array(String) COMMENT 'The hashes of the blob commitments for blob transactions' CODEC(ZSTD(1)),
+    `is_cancel_shape` Bool COMMENT 'Whether the transaction is a self-transfer of zero value, the common wallet cancellation pattern' CODEC(ZSTD(1)),
+    `first_sentry_name` LowCardinality(String) COMMENT 'Name of the sentry with the earliest sighting this hour, ties broken by sentry name' CODEC(ZSTD(1)),
+    `first_sentry_geo_city` LowCardinality(String) COMMENT 'City of the first-sighting sentry' CODEC(ZSTD(1)),
+    `first_sentry_geo_country` LowCardinality(String) COMMENT 'Country of the first-sighting sentry' CODEC(ZSTD(1)),
+    `first_sentry_geo_country_code` LowCardinality(String) COMMENT 'Country code of the first-sighting sentry' CODEC(ZSTD(1)),
+    `first_sentry_geo_continent_code` LowCardinality(String) COMMENT 'Continent code of the first-sighting sentry' CODEC(ZSTD(1))
+) ENGINE = ReplicatedReplacingMergeTree(
+    '/clickhouse/{installation}/{cluster}/tables/{shard}/{database}/{table}',
+    '{replica}',
+    `updated_date_time`
+) PARTITION BY toStartOfMonth(hour_start_date_time)
+ORDER BY
+    (`hour_start_date_time`, `hash`)
+SETTINGS
+    deduplicate_merge_projection_mode = 'rebuild'
+COMMENT 'One row per transaction hash per hour in which it was sighted in the public mempool by at least one sentry. Rows summarise the sightings within that hour only. A transaction sighted across multiple hours has multiple rows';
+
+CREATE TABLE int_transaction_mempool_observation_hourly ON CLUSTER '{cluster}' AS int_transaction_mempool_observation_hourly_local ENGINE = Distributed(
+    '{cluster}',
+    currentDatabase(),
+    int_transaction_mempool_observation_hourly_local,
+    cityHash64(`from`, `nonce`)
+);
+
+ALTER TABLE int_transaction_mempool_observation_hourly_local ON CLUSTER '{cluster}'
+ADD PROJECTION p_by_hash
+(
+    SELECT *
+    ORDER BY (`hash`, `hour_start_date_time`)
+);
+
+ALTER TABLE int_transaction_mempool_observation_hourly_local ON CLUSTER '{cluster}'
+ADD PROJECTION p_by_nonce_group
+(
+    SELECT *
+    ORDER BY (`from`, `nonce`, `hour_start_date_time`)
+);

migrations/102_fct_transaction_inclusion.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS fct_transaction_inclusion ON CLUSTER '{cluster}';
+DROP TABLE IF EXISTS fct_transaction_inclusion_local ON CLUSTER '{cluster}';

migrations/102_fct_transaction_inclusion.up.sql

@@ -0,0 +1,57 @@
+CREATE TABLE fct_transaction_inclusion_local on cluster '{cluster}' (
+    `updated_date_time` DateTime COMMENT 'Timestamp when the record was last updated' CODEC(DoubleDelta, ZSTD(1)),
+    `slot` UInt32 COMMENT 'The slot number of the beacon block containing the transaction' CODEC(DoubleDelta, ZSTD(1)),
+    `slot_start_date_time` DateTime COMMENT 'The wall clock time when the slot started' CODEC(DoubleDelta, ZSTD(1)),
+    `epoch` UInt32 COMMENT 'The epoch number containing the slot' CODEC(DoubleDelta, ZSTD(1)),
+    `epoch_start_date_time` DateTime COMMENT 'The wall clock time when the epoch started' CODEC(DoubleDelta, ZSTD(1)),
+    `block_root` FixedString(66) COMMENT 'The beacon block root hash' CODEC(ZSTD(1)),
+    `block_version` LowCardinality(String) COMMENT 'The beacon block version' CODEC(ZSTD(1)),
+    `block_number` Nullable(UInt64) COMMENT 'The execution block number, null when the block is not yet present in fct_block' CODEC(ZSTD(1)),
+    `position` UInt32 COMMENT 'The position of the transaction in the execution payload' CODEC(ZSTD(1)),
+    `hash` FixedString(66) COMMENT 'The transaction hash' CODEC(ZSTD(1)),
+    `from` FixedString(42) COMMENT 'The address of the account that sent the transaction' CODEC(ZSTD(1)),
+    `to` Nullable(FixedString(42)) COMMENT 'The address of the transaction recipient, null for contract creation' CODEC(ZSTD(1)),
+    `nonce` UInt64 COMMENT 'The nonce of the sender account at the time of the transaction' CODEC(ZSTD(1)),
+    `type` UInt8 COMMENT 'The type of the transaction' CODEC(ZSTD(1)),
+    `gas` UInt64 COMMENT 'The maximum gas provided for the transaction execution' CODEC(ZSTD(1)),
+    `gas_price` UInt128 COMMENT 'The gas price of the transaction in wei' CODEC(ZSTD(1)),
+    `gas_tip_cap` Nullable(UInt128) COMMENT 'The priority fee (tip) the user has set for the transaction in wei' CODEC(ZSTD(1)),
+    `gas_fee_cap` Nullable(UInt128) COMMENT 'The max fee the user has set for the transaction in wei' CODEC(ZSTD(1)),
+    `value` UInt128 COMMENT 'The value transferred with the transaction in wei' CODEC(ZSTD(1)),
+    `size` UInt32 COMMENT 'The size of the transaction data in bytes' CODEC(ZSTD(1)),
+    `call_data_size` UInt32 COMMENT 'The size of the call data of the transaction in bytes' CODEC(ZSTD(1)),
+    `blob_gas` Nullable(UInt64) COMMENT 'The maximum gas provided for the blob transaction execution' CODEC(ZSTD(1)),
+    `blob_gas_fee_cap` Nullable(UInt128) COMMENT 'The max blob fee the user has set for the transaction in wei' CODEC(ZSTD(1)),
+    `blob_hashes` Array(String) COMMENT 'The hashes of the blob commitments for blob transactions' CODEC(ZSTD(1)),
+    `is_cancel_shape` Bool COMMENT 'Whether the transaction is a self-transfer of zero value, the common wallet cancellation pattern' CODEC(ZSTD(1))
+) ENGINE = ReplicatedReplacingMergeTree(
+    '/clickhouse/{installation}/{cluster}/tables/{shard}/{database}/{table}',
+    '{replica}',
+    `updated_date_time`
+) PARTITION BY toStartOfMonth(slot_start_date_time)
+ORDER BY
+    (`slot_start_date_time`, `block_root`, `position`)
+SETTINGS
+    deduplicate_merge_projection_mode = 'rebuild'
+COMMENT 'One row per transaction included in a canonical beacon block execution payload. Carries no mempool or relay context so it can cover the full post-Merge history';
+
+CREATE TABLE fct_transaction_inclusion ON CLUSTER '{cluster}' AS fct_transaction_inclusion_local ENGINE = Distributed(
+    '{cluster}',
+    currentDatabase(),
+    fct_transaction_inclusion_local,
+    cityHash64(`slot_start_date_time`, `block_root`)
+);
+
+ALTER TABLE fct_transaction_inclusion_local ON CLUSTER '{cluster}'
+ADD PROJECTION p_by_hash
+(
+    SELECT *
+    ORDER BY (`hash`)
+);
+
+ALTER TABLE fct_transaction_inclusion_local ON CLUSTER '{cluster}'
+ADD PROJECTION p_by_nonce_group
+(
+    SELECT *
+    ORDER BY (`from`, `nonce`, `slot_start_date_time`)
+);

migrations/103_fct_transaction_mempool_outcome_7d.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS fct_transaction_mempool_outcome_7d ON CLUSTER '{cluster}';
+DROP TABLE IF EXISTS fct_transaction_mempool_outcome_7d_local ON CLUSTER '{cluster}';

migrations/103_fct_transaction_mempool_outcome_7d.up.sql

@@ -0,0 +1,65 @@
+CREATE TABLE fct_transaction_mempool_outcome_7d_local on cluster '{cluster}' (
+    `updated_date_time` DateTime COMMENT 'Timestamp when the record was last updated' CODEC(DoubleDelta, ZSTD(1)),
+    `first_seen_date_time` DateTime64(3) COMMENT 'Earliest sighting of the transaction within its observation episode, the anchor of the 7 day horizon' CODEC(DoubleDelta, ZSTD(1)),
+    `hash` FixedString(66) COMMENT 'The transaction hash' CODEC(ZSTD(1)),
+    `from` FixedString(42) COMMENT 'The address of the account that sent the transaction' CODEC(ZSTD(1)),
+    `to` Nullable(FixedString(42)) COMMENT 'The address of the transaction recipient, null for contract creation' CODEC(ZSTD(1)),
+    `nonce` UInt64 COMMENT 'The nonce of the sender account at the time of the transaction' CODEC(ZSTD(1)),
+    `type` Nullable(UInt8) COMMENT 'The type of the transaction' CODEC(ZSTD(1)),
+    `gas` UInt64 COMMENT 'The maximum gas provided for the transaction execution' CODEC(ZSTD(1)),
+    `gas_price` UInt128 COMMENT 'The gas price of the transaction in wei' CODEC(ZSTD(1)),
+    `gas_tip_cap` Nullable(UInt128) COMMENT 'The priority fee (tip) the user has set for the transaction in wei' CODEC(ZSTD(1)),
+    `gas_fee_cap` Nullable(UInt128) COMMENT 'The max fee the user has set for the transaction in wei' CODEC(ZSTD(1)),
+    `value` UInt128 COMMENT 'The value transferred with the transaction in wei' CODEC(ZSTD(1)),
+    `size` UInt32 COMMENT 'The size of the transaction data in bytes' CODEC(ZSTD(1)),
+    `call_data_size` UInt32 COMMENT 'The size of the call data of the transaction in bytes' CODEC(ZSTD(1)),
+    `blob_gas` Nullable(UInt64) COMMENT 'The maximum gas provided for the blob transaction execution' CODEC(ZSTD(1)),
+    `blob_gas_fee_cap` Nullable(UInt128) COMMENT 'The max blob fee the user has set for the transaction in wei' CODEC(ZSTD(1)),
+    `blob_hashes` Array(String) COMMENT 'The hashes of the blob commitments for blob transactions' CODEC(ZSTD(1)),
+    `is_cancel_shape` Bool COMMENT 'Whether the transaction is a self-transfer of zero value, the common wallet cancellation pattern' CODEC(ZSTD(1)),
+    `last_seen_date_time` DateTime64(3) COMMENT 'Latest sighting of the transaction within the 7 day horizon' CODEC(DoubleDelta, ZSTD(1)),
+    `sighting_count` UInt32 COMMENT 'Number of sightings within the 7 day horizon across all sentries' CODEC(ZSTD(1)),
+    `peak_hourly_unique_sentries` UInt32 COMMENT 'Peak number of distinct sentries sighting the transaction within any single hour of the horizon' CODEC(ZSTD(1)),
+    `outcome` LowCardinality(String) COMMENT 'Outcome within 7 days of first sighting: included, nonce_consumed (a different transaction with the same from and nonce was included) or unincluded. The statement is fixed to the horizon and remains true even if the transaction is included later' CODEC(ZSTD(1)),
+    `resolution_date_time` DateTime64(3) COMMENT 'When the outcome was determined: the inclusion slot start for included, the winner inclusion slot start for nonce_consumed, or first_seen_date_time plus 7 days for unincluded' CODEC(DoubleDelta, ZSTD(1)),
+    `included_slot` Nullable(UInt32) COMMENT 'The slot of the beacon block that included the transaction, when outcome is included' CODEC(ZSTD(1)),
+    `included_slot_start_date_time` Nullable(DateTime) COMMENT 'The wall clock time when the inclusion slot started, when outcome is included' CODEC(ZSTD(1)),
+    `included_block_root` Nullable(FixedString(66)) COMMENT 'The beacon block root of the including block, when outcome is included' CODEC(ZSTD(1)),
+    `included_position` Nullable(UInt32) COMMENT 'The position of the transaction in the execution payload, when outcome is included' CODEC(ZSTD(1)),
+    `wait_ms` Nullable(Int64) COMMENT 'Milliseconds from first sighting to the inclusion slot start, when outcome is included. Negative when the transaction was first sighted after its inclusion slot started' CODEC(ZSTD(1)),
+    `included_via_known_relay` Nullable(Bool) COMMENT 'Whether the including block matched a known relay payload-delivered record, when outcome is included. False is not proof the block was locally built' CODEC(ZSTD(1)),
+    `winner_hash` Nullable(FixedString(66)) COMMENT 'The hash of the transaction that consumed the nonce, when outcome is nonce_consumed' CODEC(ZSTD(1)),
+    `winner_slot_start_date_time` Nullable(DateTime) COMMENT 'The wall clock time when the winner inclusion slot started, when outcome is nonce_consumed' CODEC(ZSTD(1)),
+    `observed_after_nonce_consumed` Bool COMMENT 'Whether the transaction was first sighted after its nonce had already been consumed by another transaction' CODEC(ZSTD(1)),
+    `in_mempool_at_deadline` Bool COMMENT 'Whether a sighting occurred in the final hour before the 7 day deadline. Sensor evidence of continued circulation, only meaningful when outcome is unincluded' CODEC(ZSTD(1))
+) ENGINE = ReplicatedReplacingMergeTree(
+    '/clickhouse/{installation}/{cluster}/tables/{shard}/{database}/{table}',
+    '{replica}',
+    `updated_date_time`
+) PARTITION BY toStartOfMonth(first_seen_date_time)
+ORDER BY
+    (`first_seen_date_time`, `hash`)
+SETTINGS
+    deduplicate_merge_projection_mode = 'rebuild'
+COMMENT 'One immutable row per transaction hash observed in the public mempool, stating its fixed-horizon outcome within 7 days of first sighting. An observation episode starts when a hash is sighted with no sightings in the preceding 7 days. A hash re-sighted after more than 7 days of silence starts a new episode and a new row';
+
+CREATE TABLE fct_transaction_mempool_outcome_7d ON CLUSTER '{cluster}' AS fct_transaction_mempool_outcome_7d_local ENGINE = Distributed(
+    '{cluster}',
+    currentDatabase(),
+    fct_transaction_mempool_outcome_7d_local,
+    cityHash64(`from`, `nonce`)
+);
+
+ALTER TABLE fct_transaction_mempool_outcome_7d_local ON CLUSTER '{cluster}'
+ADD PROJECTION p_by_hash
+(
+    SELECT *
+    ORDER BY (`hash`)
+);
+
+ALTER TABLE fct_transaction_mempool_outcome_7d_local ON CLUSTER '{cluster}'
+ADD PROJECTION p_by_nonce_group
+(
+    SELECT *
+    ORDER BY (`from`, `nonce`, `first_seen_date_time`)
+);

migrations/104_fct_transaction_replacement.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS fct_transaction_replacement ON CLUSTER '{cluster}';
+DROP TABLE IF EXISTS fct_transaction_replacement_local ON CLUSTER '{cluster}';

migrations/104_fct_transaction_replacement.up.sql

@@ -0,0 +1,55 @@
+CREATE TABLE fct_transaction_replacement_local on cluster '{cluster}' (
+    `updated_date_time` DateTime COMMENT 'Timestamp when the record was last updated' CODEC(DoubleDelta, ZSTD(1)),
+    `group_first_seen_date_time` DateTime64(3) COMMENT 'Earliest sighting of any attempt in the nonce group, the anchor of the group window' CODEC(DoubleDelta, ZSTD(1)),
+    `from` FixedString(42) COMMENT 'The address of the account that sent the attempts' CODEC(ZSTD(1)),
+    `nonce` UInt64 COMMENT 'The nonce the attempts compete for' CODEC(ZSTD(1)),
+    `hash` FixedString(66) COMMENT 'The transaction hash of this attempt' CODEC(ZSTD(1)),
+    `attempt_index` UInt16 COMMENT 'The 1-based rank of this attempt within the group, ordered by first sighting with ties broken by hash' CODEC(ZSTD(1)),
+    `group_attempt_count` UInt16 COMMENT 'Total number of publicly observed attempts in the group window' CODEC(ZSTD(1)),
+    `first_seen_date_time` DateTime64(3) COMMENT 'Earliest sighting of this attempt' CODEC(DoubleDelta, ZSTD(1)),
+    `previous_hash` Nullable(FixedString(66)) COMMENT 'The hash of the attempt immediately before this one in the group, null for the first attempt' CODEC(ZSTD(1)),
+    `to` Nullable(FixedString(42)) COMMENT 'The address of the attempt recipient, null for contract creation' CODEC(ZSTD(1)),
+    `type` Nullable(UInt8) COMMENT 'The type of the transaction' CODEC(ZSTD(1)),
+    `gas_price` UInt128 COMMENT 'The gas price of the attempt in wei' CODEC(ZSTD(1)),
+    `gas_tip_cap` Nullable(UInt128) COMMENT 'The priority fee (tip) of the attempt in wei' CODEC(ZSTD(1)),
+    `gas_fee_cap` Nullable(UInt128) COMMENT 'The max fee of the attempt in wei' CODEC(ZSTD(1)),
+    `value` UInt128 COMMENT 'The value transferred with the attempt in wei' CODEC(ZSTD(1)),
+    `gas_tip_cap_delta` Nullable(Int128) COMMENT 'This attempt gas_tip_cap minus the previous attempt gas_tip_cap, null for the first attempt or when either side is null' CODEC(ZSTD(1)),
+    `gas_fee_cap_delta` Nullable(Int128) COMMENT 'This attempt gas_fee_cap minus the previous attempt gas_fee_cap, null for the first attempt or when either side is null' CODEC(ZSTD(1)),
+    `is_cancel_shape` Bool COMMENT 'Whether the attempt is a self-transfer of zero value, the common wallet cancellation pattern' CODEC(ZSTD(1)),
+    `is_winner` Bool COMMENT 'Whether this attempt is the one that was included on chain' CODEC(ZSTD(1)),
+    `group_outcome` LowCardinality(String) COMMENT 'Outcome of the nonce group within 7 days of the group anchor: included or unincluded' CODEC(ZSTD(1)),
+    `winner_hash` Nullable(FixedString(66)) COMMENT 'The hash of the included attempt, null when the group resolved unincluded' CODEC(ZSTD(1)),
+    `resolution_date_time` DateTime64(3) COMMENT 'When the group resolved: the winner inclusion slot start, or the group anchor plus 7 days when unincluded' CODEC(DoubleDelta, ZSTD(1)),
+    `observed_after_resolution` Bool COMMENT 'Whether this attempt was first sighted after the group had already resolved' CODEC(ZSTD(1))
+) ENGINE = ReplicatedReplacingMergeTree(
+    '/clickhouse/{installation}/{cluster}/tables/{shard}/{database}/{table}',
+    '{replica}',
+    `updated_date_time`
+) PARTITION BY toStartOfMonth(group_first_seen_date_time)
+ORDER BY
+    (`group_first_seen_date_time`, `from`, `nonce`, `hash`)
+SETTINGS
+    deduplicate_merge_projection_mode = 'rebuild'
+COMMENT 'One row per publicly observed attempt in nonce groups with at least two attempts, emitted when the group resolves. A nonce group is all transactions sharing a from address and nonce whose first sightings fall within 7 days of the group anchor';
+
+CREATE TABLE fct_transaction_replacement ON CLUSTER '{cluster}' AS fct_transaction_replacement_local ENGINE = Distributed(
+    '{cluster}',
+    currentDatabase(),
+    fct_transaction_replacement_local,
+    cityHash64(`from`, `nonce`)
+);
+
+ALTER TABLE fct_transaction_replacement_local ON CLUSTER '{cluster}'
+ADD PROJECTION p_by_hash
+(
+    SELECT *
+    ORDER BY (`hash`)
+);
+
+ALTER TABLE fct_transaction_replacement_local ON CLUSTER '{cluster}'
+ADD PROJECTION p_by_nonce_group
+(
+    SELECT *
+    ORDER BY (`from`, `nonce`, `group_first_seen_date_time`)
+);

migrations/105_fct_transaction_inclusion_hourly.down.sql

@@ -0,0 +1,2 @@
+DROP TABLE IF EXISTS fct_transaction_inclusion_hourly ON CLUSTER '{cluster}';
+DROP TABLE IF EXISTS fct_transaction_inclusion_hourly_local ON CLUSTER '{cluster}';