change: ETCM-12220 Support tx-out.value = "consumed" Db-Sync option (#949)

Author: AmbientTea

Cargo.lock

@@ -187,6 +187,7 @@ sqlx = { version = "0.8.6", default-features = false, features = [
 ] }
 derive-where = { version = "1.2.7", default-features = false }
 once_cell = { version = "1.21.3", default-features = false }
+paste = { version = "1.0.15" }
 
 # substrate dependencies
 frame-benchmarking = { default-features = false, git = "https://github.com/paritytech/polkadot-sdk.git", tag = "polkadot-stable2506" }

Cargo.toml

@@ -59,6 +59,8 @@ block producer rewards on-chain. Handling of the transfers is left to the chain
 * `delete_metadata` extrinsic in `pallet-block-producer-metadata`
 * Added litep2p networking stack support to pc demo node as default. libp2p can be set with explicitly `--network-backend` parameter
 * Added `CARDANO_DATA_SOURCE` env variable to determine data source for partner-chains. `db-sync` value is used by default
+* Added support for querying Db-Sync database when `tx-in.value` configuration field is set to `"consumed"`.
+The data sources will automatically detect this option on startup and adjust their queries accordingly.
 
 ## Fixed
 

changelog.md

@@ -1,4 +1,9 @@
 {
+  "insert_options": {
+    "tx_output": {
+      "value": "consumed"
+    }
+  },
   "EnableLogMetrics": false,
   "EnableLogging": true,
   "NetworkName": "Testnet",
@@ -176,4 +181,4 @@
       "scRotation": null
     }
   ]
-}
+}

dev/local-environment/configurations/db-sync/config.json

@@ -56,6 +56,7 @@ tokio-test = "0.4.3"
 ctor = "0.4.1"
 testcontainers-modules = { version = "0.1.3", features = ["postgres"] }
 pretty_assertions = { workspace = true }
+paste = { workspace = true }
 
 [features]
 default = []

toolkit/data-sources/db-sync/Cargo.toml

@@ -9,7 +9,7 @@ use std::str::FromStr;
 const BLOCK_4_TS_MILLIS: u64 = 1650561570000;
 const BLOCK_5_TS_MILLIS: u64 = 1650562570000;
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn get_latest_block_info(pool: PgPool) {
 	let irrelevant_security_parameter = 1000;
 	let source = mk_datasource(pool, irrelevant_security_parameter);
@@ -29,7 +29,7 @@ async fn get_latest_block_info(pool: PgPool) {
 	assert_eq!(block, expected)
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_latest_stable_block(pool: PgPool) {
 	let security_parameter = 2;
 	let source = mk_datasource(pool, security_parameter);
@@ -43,7 +43,7 @@ async fn test_get_latest_stable_block(pool: PgPool) {
 	assert_eq!(block, Some(expected))
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_latest_stable_block_at_filters_out_by_max_slot_boundary(pool: PgPool) {
 	let security_parameter = 2;
 	let slots_distance_between_blocks = block_4().slot.0 - block_2().slot.0;
@@ -58,7 +58,7 @@ async fn test_get_latest_stable_block_at_filters_out_by_max_slot_boundary(pool:
 	assert_eq!(block, None)
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_stable_block_at(pool: PgPool) {
 	let security_parameter = 2;
 	let source = mk_datasource(pool, security_parameter);
@@ -70,7 +70,7 @@ async fn test_get_stable_block_at(pool: PgPool) {
 	assert_eq!(block, Some(block_2()));
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_stable_block_at_returns_block_that_dont_have_k_blocks_on_them_at_given_timestamp(
 	pool: PgPool,
 ) {
@@ -85,7 +85,7 @@ async fn test_get_stable_block_at_returns_block_that_dont_have_k_blocks_on_them_
 	assert_eq!(block, Some(block_2()));
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_stable_block_at_filters_out_by_min_slots_boundary(pool: PgPool) {
 	let security_parameter = 3;
 	let slots_distance_between_blocks = block_4().slot.0 - block_2().slot.0;
@@ -104,7 +104,7 @@ async fn test_get_stable_block_at_filters_out_by_min_slots_boundary(pool: PgPool
 	assert_eq!(block, None);
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_stable_block_at_filters_out_by_max_slots_boundary(pool: PgPool) {
 	let security_parameter = 2;
 	let slots_distance_between_blocks = block_4().slot.0 - block_2().slot.0;
@@ -123,7 +123,7 @@ async fn test_get_stable_block_at_filters_out_by_max_slots_boundary(pool: PgPool
 	assert_eq!(block, None);
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_stable_block_info_by_hash_for_unknown_hash(pool: PgPool) {
 	let source = mk_datasource(pool, 2);
 	let unknown_hash =
@@ -135,7 +135,7 @@ async fn test_get_stable_block_info_by_hash_for_unknown_hash(pool: PgPool) {
 	assert_eq!(result, None)
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_latest_stable_block_with_stability_margin(pool: PgPool) {
 	let security_parameter = 2;
 	let stability_margin = 1;
@@ -164,7 +164,7 @@ async fn test_get_latest_stable_block_with_stability_margin(pool: PgPool) {
 	assert_eq!(block, None);
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_latest_stable_block_with_stability_margin_2(pool: PgPool) {
 	let security_parameter = 2;
 	let stability_margin = 0;
@@ -190,7 +190,7 @@ async fn test_get_latest_stable_block_with_stability_margin_2(pool: PgPool) {
 	assert_eq!(block, Some(block_0()));
 }
 
-#[sqlx::test(migrations = "./testdata/migrations")]
+#[sqlx::test(migrations = "./testdata/migrations-tx-in-consumed")]
 async fn test_get_stable_block_caching(pool: PgPool) {
 	fn dummy_hash(n: u8) -> McBlockHash {
 		McBlockHash([n; 32])

toolkit/data-sources/db-sync/src/block/tests.rs

@@ -1,7 +1,8 @@
 //! Db-Sync data source used by Partner Chain committee selection
 use crate::DataSourceError::*;
 use crate::db_model::{
-	self, Address, Asset, BlockNumber, EpochNumber, MainchainTxOutput, StakePoolEntry,
+	self, Address, Asset, BlockNumber, DbSyncConfigurationProvider, EpochNumber, MainchainTxOutput,
+	StakePoolEntry,
 };
 use crate::metrics::McFollowerMetrics;
 use crate::observed_async_trait;
@@ -49,6 +50,8 @@ pub struct CandidatesDataSourceImpl {
 	pool: PgPool,
 	/// Prometheus metrics client
 	metrics_opt: Option<McFollowerMetrics>,
+	/// Configuration used by Db-Sync
+	db_sync_config: DbSyncConfigurationProvider,
 }
 
 observed_async_trait!(
@@ -126,7 +129,11 @@ impl CandidatesDataSourceImpl {
 	) -> Result<CandidatesDataSourceImpl, Box<dyn std::error::Error + Send + Sync>> {
 		db_model::create_idx_ma_tx_out_ident(&pool).await?;
 		db_model::create_idx_tx_out_address(&pool).await?;
-		Ok(Self { pool, metrics_opt })
+		Ok(Self {
+			pool: pool.clone(),
+			metrics_opt,
+			db_sync_config: DbSyncConfigurationProvider::new(pool),
+		})
 	}
 
 	/// Creates a new caching instance of the data source
@@ -154,7 +161,15 @@ impl CandidatesDataSourceImpl {
 		let registrations_block_for_epoch = self.get_last_block_for_epoch(epoch).await?;
 		let address: Address = Address(committee_candidate_address.to_string());
 		let active_utxos = match registrations_block_for_epoch {
-			Some(block) => db_model::get_utxos_for_address(&self.pool, &address, block).await?,
+			Some(block) => {
+				db_model::get_utxos_for_address(
+					&self.pool,
+					&address,
+					block,
+					self.db_sync_config.get_tx_in_config().await?,
+				)
+				.await?
+			},
 			None => vec![],
 		};
 		self.convert_utxos_to_candidates(&active_utxos)