postpone polling db-sync configuration

Author: AmbientTea

toolkit/data-sources/db-sync/src/candidates/mod.rs

@@ -1,8 +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,
-	TxInConfiguration,
+	self, Address, Asset, BlockNumber, DbSyncConfigurationProvider, EpochNumber, MainchainTxOutput,
+	StakePoolEntry,
 };
 use crate::metrics::McFollowerMetrics;
 use crate::observed_async_trait;
@@ -50,8 +50,8 @@ pub struct CandidatesDataSourceImpl {
 	pool: PgPool,
 	/// Prometheus metrics client
 	metrics_opt: Option<McFollowerMetrics>,
-	/// Transaction input configuration used by Db-Sync
-	tx_in_config: TxInConfiguration,
+	/// Configuration used by Db-Sync
+	db_sync_config: DbSyncConfigurationProvider,
 }
 
 observed_async_trait!(
@@ -129,8 +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?;
-		let tx_in_config = TxInConfiguration::from_connection(&pool).await?;
-		Ok(Self { pool, metrics_opt, tx_in_config })
+		Ok(Self {
+			pool: pool.clone(),
+			metrics_opt,
+			db_sync_config: DbSyncConfigurationProvider::new(pool),
+		})
 	}
 
 	/// Creates a new caching instance of the data source
@@ -159,8 +162,13 @@ impl CandidatesDataSourceImpl {
 		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, self.tx_in_config)
-					.await?
+				db_model::get_utxos_for_address(
+					&self.pool,
+					&address,
+					block,
+					self.db_sync_config.get_tx_in_config().await?,
+				)
+				.await?
 			},
 			None => vec![],
 		};

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

@@ -1,11 +1,13 @@
 use crate::candidates::CandidatesDataSourceImpl;
-use crate::db_model::{TxInConfiguration, index_exists_unsafe};
+use crate::db_model::{DbSyncConfigurationProvider, TxInConfiguration, index_exists_unsafe};
 use crate::metrics::mock::test_metrics;
 use authority_selection_inherents::AuthoritySelectionDataSource;
 use hex_literal::hex;
 use sidechain_domain::*;
 use sqlx::PgPool;
+use std::cell::OnceCell;
 use std::str::FromStr;
+use std::sync::Arc;
 use tokio_test::assert_err;
 
 const D_PARAM_POLICY: [u8; 28] = hex!("500000000000000000000000000000000000434845434b504f494e69");
@@ -290,7 +292,14 @@ mod candidate_caching {
 }
 
 fn make_source(pool: PgPool, tx_in_config: TxInConfiguration) -> CandidatesDataSourceImpl {
-	CandidatesDataSourceImpl { pool, metrics_opt: Some(test_metrics()), tx_in_config }
+	CandidatesDataSourceImpl {
+		pool: pool.clone(),
+		metrics_opt: Some(test_metrics()),
+		db_sync_config: DbSyncConfigurationProvider {
+			pool,
+			tx_in_config: Arc::new(tokio::sync::Mutex::new(OnceCell::from(tx_in_config))),
+		},
+	}
 }
 
 fn candidates_address() -> MainchainAddress {

toolkit/data-sources/db-sync/src/db_model.rs

@@ -1,5 +1,5 @@
-use crate::SqlxError;
 use crate::db_datum::DbDatum;
+use crate::{DataSourceError, SqlxError};
 use bigdecimal::ToPrimitive;
 use cardano_serialization_lib::PlutusData;
 use chrono::NaiveDateTime;
@@ -9,8 +9,11 @@ use sidechain_domain::{
 	MainchainBlock, McBlockHash, McBlockNumber, McEpochNumber, McSlotNumber, McTxHash, UtxoId,
 	UtxoIndex,
 };
-use sqlx::{Decode, Pool, Postgres, database::Database, error::BoxDynError, postgres::PgTypeInfo};
-use std::str::FromStr;
+use sqlx::{
+	Decode, PgPool, Pool, Postgres, database::Database, error::BoxDynError, postgres::PgTypeInfo,
+};
+use std::{cell::OnceCell, str::FromStr, sync::Arc};
+use tokio::sync::Mutex;
 
 /// Db-Sync `tx_in.value` configuration field
 #[derive(Debug, PartialEq, Copy, Clone)]
@@ -45,6 +48,37 @@ impl TxInConfiguration {
 	}
 }
 
+/// Structure that queries, caches and provides Db-Sync configuration
+pub struct DbSyncConfigurationProvider {
+	/// Postgres connection pool
+	pub(crate) pool: PgPool,
+	/// Transaction input configuration used by Db-Sync
+	pub(crate) tx_in_config: Arc<Mutex<OnceCell<TxInConfiguration>>>,
+}
+
+impl DbSyncConfigurationProvider {
+	pub(crate) fn new(pool: PgPool) -> Self {
+		Self { tx_in_config: Arc::new(Mutex::new(OnceCell::new())), pool }
+	}
+
+	pub(crate) async fn get_tx_in_config(
+		&self,
+	) -> std::result::Result<TxInConfiguration, DataSourceError> {
+		let lock = self.tx_in_config.lock().await;
+		if let Some(tx_in_config) = lock.get() {
+			return Ok(*tx_in_config);
+		} else {
+			let tx_in_config = TxInConfiguration::from_connection(&self.pool).await?;
+			lock.set(tx_in_config).map_err(|_| {
+				DataSourceError::InternalDataSourceError(
+					"Failed to set tx_in_config in GovernedMapDataSourceImpl".into(),
+				)
+			})?;
+			return Ok(tx_in_config);
+		}
+	}
+}
+
 #[derive(Debug, Clone, sqlx::FromRow, PartialEq)]
 pub(crate) struct Block {
 	pub block_no: BlockNumber,

toolkit/data-sources/db-sync/src/governed_map/mod.rs

@@ -2,7 +2,7 @@
 use crate::DataSourceError::ExpectedDataNotFound;
 use crate::Result;
 use crate::block::BlockDataSourceImpl;
-use crate::db_model::{GovernedMapAction, TxInConfiguration};
+use crate::db_model::{DbSyncConfigurationProvider, GovernedMapAction, TxInConfiguration};
 use crate::{metrics::McFollowerMetrics, observed_async_trait};
 use db_sync_sqlx::{Asset, BlockNumber};
 use itertools::Itertools;
@@ -26,8 +26,8 @@ pub struct GovernedMapDataSourceImpl {
 	pub pool: PgPool,
 	/// Prometheus metrics client
 	pub metrics_opt: Option<McFollowerMetrics>,
-	/// Transaction input configuration used by Db-Sync
-	tx_in_config: TxInConfiguration,
+	/// Configuration used by Db-Sync
+	db_sync_config: DbSyncConfigurationProvider,
 }
 
 impl GovernedMapDataSourceImpl {
@@ -37,8 +37,11 @@ impl GovernedMapDataSourceImpl {
 		metrics_opt: Option<McFollowerMetrics>,
 	) -> std::result::Result<Self, Box<dyn std::error::Error + Send + Sync>> {
 		crate::db_model::create_idx_tx_out_address(&pool).await?;
-		let tx_in_config = TxInConfiguration::from_connection(&pool).await?;
-		Ok(Self { pool, metrics_opt, tx_in_config })
+		Ok(Self {
+			pool: pool.clone(),
+			metrics_opt,
+			db_sync_config: DbSyncConfigurationProvider::new(pool),
+		})
 	}
 }
 
@@ -50,7 +53,12 @@ impl GovernedMapDataSource for GovernedMapDataSourceImpl {
 		scripts: MainChainScriptsV1,
 	) -> std::result::Result<BTreeMap<String, ByteString>, Box<dyn std::error::Error + Send + Sync>>
 	{
-		Ok(get_mappings_entries(&self.pool, mc_block, scripts, self.tx_in_config).await?.into())
+		Ok(get_mappings_entries(
+			&self.pool,
+			mc_block,
+			scripts,
+			self.db_sync_config.get_tx_in_config().await?
+		).await?.into())
 	}
 
 	async fn get_mapping_changes(
@@ -131,8 +139,8 @@ pub struct GovernedMapDataSourceCachedImpl {
 	cache: Arc<Mutex<Cache>>,
 	/// [BlockDataSourceImpl] instance shared with other data sources for cache reuse.
 	blocks: Arc<BlockDataSourceImpl>,
-	/// Transaction input configuration used by Db-Sync
-	tx_in_config: TxInConfiguration,
+	/// Configuration used by Db-Sync
+	db_sync_config: DbSyncConfigurationProvider,
 }
 
 impl GovernedMapDataSourceCachedImpl {
@@ -145,8 +153,14 @@ impl GovernedMapDataSourceCachedImpl {
 	) -> std::result::Result<Self, Box<dyn std::error::Error + Send + Sync>> {
 		crate::db_model::create_idx_tx_out_address(&pool).await?;
 		let cache = Default::default();
-		let tx_in_config = TxInConfiguration::from_connection(&pool).await?;
-		Ok(Self { pool, metrics_opt, cache_size, cache, blocks, tx_in_config })
+		Ok(Self {
+			pool: pool.clone(),
+			metrics_opt,
+			cache_size,
+			cache,
+			blocks,
+			db_sync_config: DbSyncConfigurationProvider::new(pool),
+		})
 	}
 }
 
@@ -158,7 +172,7 @@ impl GovernedMapDataSource for GovernedMapDataSourceCachedImpl {
 		scripts: MainChainScriptsV1,
 	) -> std::result::Result<BTreeMap<String, ByteString>, Box<dyn std::error::Error + Send + Sync>>
 	{
-		Ok(get_mappings_entries(&self.pool, mc_block, scripts, self.tx_in_config).await?.into())
+		Ok(get_mappings_entries(&self.pool, mc_block, scripts, self.db_sync_config.get_tx_in_config().await?).await?.into())
 	}
 
 	async fn get_mapping_changes(
@@ -244,7 +258,7 @@ impl GovernedMapDataSourceCachedImpl {
 			since_block,
 			up_to_block,
 			Asset::new(scripts.asset_policy_id),
-			self.tx_in_config,
+			self.db_sync_config.get_tx_in_config().await?,
 		)
 		.await?;
 

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

@@ -1,6 +1,6 @@
 use super::{Cache, GovernedMapDataSourceCachedImpl, GovernedMapDataSourceImpl};
 use crate::block::{BlockDataSourceImpl, DbSyncBlockDataSourceConfig};
-use crate::db_model::TxInConfiguration;
+use crate::db_model::{DbSyncConfigurationProvider, TxInConfiguration};
 use crate::metrics::mock::test_metrics;
 use hex_literal::hex;
 use pretty_assertions::assert_eq;
@@ -9,6 +9,7 @@ use sidechain_domain::mainchain_epoch::{Duration, MainchainEpochConfig, Timestam
 use sidechain_domain::*;
 use sp_governed_map::{GovernedMapDataSource, MainChainScriptsV1};
 use sqlx::PgPool;
+use std::cell::OnceCell;
 use std::str::FromStr;
 use std::sync::{Arc, Mutex};
 use tokio_test::assert_err;
@@ -150,7 +151,14 @@ fn scripts() -> MainChainScriptsV1 {
 }
 
 fn make_source(pool: PgPool, tx_in_config: TxInConfiguration) -> GovernedMapDataSourceImpl {
-	GovernedMapDataSourceImpl { pool, metrics_opt: Some(test_metrics()), tx_in_config }
+	GovernedMapDataSourceImpl {
+		pool: pool.clone(),
+		metrics_opt: Some(test_metrics()),
+		db_sync_config: DbSyncConfigurationProvider {
+			pool,
+			tx_in_config: Arc::new(tokio::sync::Mutex::new(OnceCell::from(tx_in_config))),
+		},
+	}
 }
 
 async fn make_cached_source(
@@ -163,15 +171,18 @@ async fn make_cached_source(
 		cache_size: 10u16,
 		cache: Arc::new(Mutex::new(Cache::default())),
 		blocks: Arc::new(BlockDataSourceImpl::from_config(
-			pool,
+			pool.clone(),
 			DbSyncBlockDataSourceConfig {
 				cardano_security_parameter: 432,
 				cardano_active_slots_coeff: 0.05,
 				block_stability_margin: 0,
 			},
 			&mainchain_epoch_config(),
 		)),
-		tx_in_config,
+		db_sync_config: DbSyncConfigurationProvider {
+			pool,
+			tx_in_config: Arc::new(tokio::sync::Mutex::new(OnceCell::from(tx_in_config))),
+		},
 	}
 }