blockworks-foundation
/
mango-feeds
mirror of https://github.com/blockworks-foundation/mango-feeds.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add config flag for deleting old data 

Keeping it is very useful for testing.
This commit is contained in:
Christian Kamm 2022-01-03 10:33:26 +01:00
parent f0c26bb8fb
commit d5395a3b75
4 changed files with 49 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
connector-mango/example-config.toml
View File

@ -24,3 +24,4 @@ retry_query_sleep_secs = 5
retry_connection_sleep_secs = 30
fatal_connection_timeout_secs = 600
allow_invalid_certs = false
delete_old_data = true

1
connector-raw/example-config.toml
View File

@ -24,3 +24,4 @@ retry_query_sleep_secs = 5
retry_connection_sleep_secs = 30
fatal_connection_timeout_secs = 600
allow_invalid_certs = false
delete_old_data = true

2
lib/src/lib.rs
View File

@ -82,6 +82,8 @@ pub struct PostgresConfig {
pub fatal_connection_timeout_secs: u64,
/// Allow invalid TLS certificates, passed to native_tls danger_accept_invalid_certs
pub allow_invalid_certs: bool,
/// Delete old data automatically, keeping only the current data snapshot
pub delete_old_data: bool,
}
#[derive(Clone, Debug, Deserialize)]

85
lib/src/postgres_target.rs
View File

@ -170,48 +170,53 @@ struct SlotsProcessing {
}
impl SlotsProcessing {
fn new(tables: &Vec<String>) -> Self {
// Delete:
// 1. account writes that came before the newest rooted write
// 2. account writes that came after the newest rooted write but before
// the newest rooted slot (like processed writes that never confirmed)
let mut cleanup_table_sql: Vec<String> = tables
.iter()
.map(|table_name| {
format!(
"DELETE FROM {table} AS data
USING (
SELECT DISTINCT ON(pubkey_id) pubkey_id, slot, write_version
FROM {table}
LEFT JOIN slot USING(slot)
WHERE slot <= $newest_final_slot AND (status = 'Rooted' OR status is NULL)
ORDER BY pubkey_id, slot DESC, write_version DESC
) latest_write
WHERE data.pubkey_id = latest_write.pubkey_id
AND (
(data.slot < latest_write.slot
OR (data.slot = latest_write.slot
AND data.write_version < latest_write.write_version
)
)
OR
(
data.slot < $newest_final_slot
AND
(data.slot > latest_write.slot
fn new(tables: &Vec<String>, delete_old_data: bool) -> Self {
let cleanup_table_sql = Vec::<String>::new();
if delete_old_data {
// Delete:
// 1. account writes that came before the newest rooted write
// 2. account writes that came after the newest rooted write but before
// the newest rooted slot (like processed writes that never confirmed)
let mut cleanup_table_sql: Vec<String> = tables
.iter()
.map(|table_name| {
format!(
"DELETE FROM {table} AS data
USING (
SELECT DISTINCT ON(pubkey_id) pubkey_id, slot, write_version
FROM {table}
LEFT JOIN slot USING(slot)
WHERE slot <= $newest_final_slot AND (status = 'Rooted' OR status is NULL)
ORDER BY pubkey_id, slot DESC, write_version DESC
) latest_write
WHERE data.pubkey_id = latest_write.pubkey_id
AND (
(data.slot < latest_write.slot
OR (data.slot = latest_write.slot
AND data.write_version > latest_write.write_version
AND data.write_version < latest_write.write_version
)
)
)
)",
table = table_name
)
})
.collect();
OR
(
data.slot < $newest_final_slot
AND
(data.slot > latest_write.slot
OR (data.slot = latest_write.slot
AND data.write_version > latest_write.write_version
)
)
)
)",
table = table_name
)
})
.collect();
// Delete old slots
cleanup_table_sql.push("DELETE FROM slot WHERE slot + 100000 < $newest_final_slot".into());
// Delete old slots
cleanup_table_sql
.push("DELETE FROM slot WHERE slot + 100000 < $newest_final_slot".into());
}
Self { cleanup_table_sql }
}
@ -264,7 +269,7 @@ impl SlotsProcessing {
.context("updating preceding non-rooted slots")?;
// Keep only the newest rooted account write and also
// wipe old slots
// wipe old slots (if configured)
for cleanup_sql in &self.cleanup_table_sql {
let query = query_dyn!(cleanup_sql, newest_final_slot = update.slot)?;
let _ = query
@ -407,7 +412,7 @@ pub async fn init(
.iter()
.map(|table| table.table_name().to_string())
.collect();
let slots_processing = SlotsProcessing::new(&table_names);
let slots_processing = SlotsProcessing::new(&table_names, config.delete_old_data);
for _ in 0..config.slot_update_connection_count {
let postgres_slot =
postgres_connection(config, metric_con_retries.clone(), metric_con_live.clone())