contractless/Contractless
contractless
/
Contractless
1
1
Fork 0
Code Issues Pull Requests Packages Projects Releases 1 Wiki Activity

removed unused params

This commit is contained in:
viraladmin 2026-06-23 09:17:12 -06:00
parent ef3f4ed3c4
commit bdca5ed1ff
13 changed files with 256 additions and 268 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

59
src/records/memory/mempool/mod.rs
View File

@ -18,7 +18,8 @@ use anyhow::{anyhow, Result};
use once_cell::sync::OnceCell; use once_cell::sync::OnceCell;
use std::fs::File; use std::fs::File;
use std::io::Write; use std::io::Write;
use tokio_postgres::Client; use tokio_postgres::types::ToSql;
use tokio_postgres::{Client, Row};
lazy_static! { lazy_static! {
pub static ref BASECOIN: String = { pub static ref BASECOIN: String = {
@ -215,7 +216,9 @@ pub use processing::{
restore_processed_by_signatures, restore_selected_transactions_processed, restore_processed_by_signatures, restore_selected_transactions_processed,
spawn_processed_cleanup, spawn_processed_cleanup,
}; };
pub use schema::{clear_mempool, db_client, ensure_db_connection, init_db, setup_mempool}; pub use schema::{
clear_mempool, db_client, ensure_db_connection, init_db, pg_execute, pg_query, setup_mempool,
};
pub use selection::{ pub use selection::{
apply_selected_transaction_math, clear_selected_transaction_sql, delete_selected_transactions, apply_selected_transaction_math, clear_selected_transaction_sql, delete_selected_transactions,
select_transactions_for_block, stream_selected_transaction_originals, select_transactions_for_block, stream_selected_transaction_originals,
@ -308,23 +311,19 @@ fn ids_for_table(batch: &SelectedMempoolBatch, table: &str) -> Vec<i64> {
.collect() .collect()
} }
async fn mark_rows_by_ids( async fn mark_rows_by_ids(table: &str, ids: &[i64], block_number: i32) -> Result<()> {
client: &Client,
table: &str,
ids: &[i64],
block_number: i32,
) -> Result<()> {
if ids.is_empty() { if ids.is_empty() {
return Ok(()); return Ok(());
} }
let statement = let statement =
format!("UPDATE {table} SET processed=true, processed_block_number=$1 WHERE id = ANY($2)"); format!("UPDATE {table} SET processed=true, processed_block_number=$1 WHERE id = ANY($2)");
client.execute(&statement, &[&block_number, &ids]).await?; let params: [&(dyn ToSql + Sync); 2] = [&block_number, &ids];
pg_execute(&statement, &params).await?;
Ok(()) Ok(())
} }
async fn unmark_rows_by_ids(client: &Client, table: &str, ids: &[i64]) -> Result<()> { async fn unmark_rows_by_ids(table: &str, ids: &[i64]) -> Result<()> {
if ids.is_empty() { if ids.is_empty() {
return Ok(()); return Ok(());
} }
@ -332,22 +331,23 @@ async fn unmark_rows_by_ids(client: &Client, table: &str, ids: &[i64]) -> Result
let statement = format!( let statement = format!(
"UPDATE {table} SET processed=false, processed_block_number=NULL WHERE id = ANY($1)" "UPDATE {table} SET processed=false, processed_block_number=NULL WHERE id = ANY($1)"
); );
client.execute(&statement, &[&ids]).await?; let params: [&(dyn ToSql + Sync); 1] = [&ids];
pg_execute(&statement, &params).await?;
Ok(()) Ok(())
} }
async fn delete_rows(client: &Client, table: &str, ids: &[i64]) -> Result<()> { async fn delete_rows(table: &str, ids: &[i64]) -> Result<()> {
if ids.is_empty() { if ids.is_empty() {
return Ok(()); return Ok(());
} }
let statement = format!("DELETE FROM {table} WHERE id = ANY($1)"); let statement = format!("DELETE FROM {table} WHERE id = ANY($1)");
client.execute(&statement, &[&ids]).await?; let params: [&(dyn ToSql + Sync); 1] = [&ids];
pg_execute(&statement, &params).await?;
Ok(()) Ok(())
} }
async fn unmark_by_signatures( async fn unmark_by_signatures(
client: &Client,
table: &str, table: &str,
signature_column: &str, signature_column: &str,
signatures: &[String], signatures: &[String],
@ -355,33 +355,31 @@ async fn unmark_by_signatures(
let statement = format!( let statement = format!(
"UPDATE {table} SET processed=false, processed_block_number=NULL WHERE {signature_column} = ANY($1) AND processed = true" "UPDATE {table} SET processed=false, processed_block_number=NULL WHERE {signature_column} = ANY($1) AND processed = true"
); );
Ok(client.execute(&statement, &[&signatures]).await?) let params: [&(dyn ToSql + Sync); 1] = [&signatures];
pg_execute(&statement, &params).await
} }
async fn delete_processed_before_or_at(block_number: u32, limit: i64) -> Result<()> { async fn delete_processed_before_or_at(block_number: u32, limit: i64) -> Result<()> {
// Periodic cleanup deletes processed mempool rows in bounded batches // Periodic cleanup deletes processed mempool rows in bounded batches
// so long-lived nodes do not accumulate infinite processed history. // so long-lived nodes do not accumulate infinite processed history.
let client_handle = db_client().await?;
let client = client_handle.as_ref();
let bn = block_number as i32; let bn = block_number as i32;
delete_processed_rows_limited(client, "transfer", bn, limit).await?; delete_processed_rows_limited("transfer", bn, limit).await?;
delete_processed_rows_limited(client, "token", bn, limit).await?; delete_processed_rows_limited("token", bn, limit).await?;
delete_processed_rows_limited(client, "issue_token", bn, limit).await?; delete_processed_rows_limited("issue_token", bn, limit).await?;
delete_processed_rows_limited(client, "burn", bn, limit).await?; delete_processed_rows_limited("burn", bn, limit).await?;
delete_processed_rows_limited(client, "nft", bn, limit).await?; delete_processed_rows_limited("nft", bn, limit).await?;
delete_processed_rows_limited(client, "marketing", bn, limit).await?; delete_processed_rows_limited("marketing", bn, limit).await?;
delete_processed_rows_limited(client, "vanity_address", bn, limit).await?; delete_processed_rows_limited("vanity_address", bn, limit).await?;
delete_processed_rows_limited(client, "swap", bn, limit).await?; delete_processed_rows_limited("swap", bn, limit).await?;
delete_processed_rows_limited(client, "loan_contract", bn, limit).await?; delete_processed_rows_limited("loan_contract", bn, limit).await?;
delete_processed_rows_limited(client, "loan_payment", bn, limit).await?; delete_processed_rows_limited("loan_payment", bn, limit).await?;
delete_processed_rows_limited(client, "collateral_claim", bn, limit).await?; delete_processed_rows_limited("collateral_claim", bn, limit).await?;
Ok(()) Ok(())
} }
async fn delete_processed_rows_limited( async fn delete_processed_rows_limited(
client: &Client,
table: &str, table: &str,
block_number: i32, block_number: i32,
limit: i64, limit: i64,
@ -401,5 +399,6 @@ async fn delete_processed_rows_limited(
"# "#
); );
Ok(client.execute(&statement, &[&block_number, &limit]).await?) let params: [&(dyn ToSql + Sync); 2] = [&block_number, &limit];
pg_execute(&statement, &params).await
} }

364
src/records/memory/mempool/processing.rs
View File

@ -1,4 +1,19 @@
use super::*; use super::*;
use tokio_postgres::types::ToSql;
async fn pg_execute_block_signatures(
statement: &str,
block_number: i32,
signatures: &[String],
) -> Result<u64> {
let params: [&(dyn ToSql + Sync); 2] = [&block_number, &signatures];
pg_execute(statement, &params).await
}
async fn pg_execute_signatures(statement: &str, signatures: &[String]) -> Result<u64> {
let params: [&(dyn ToSql + Sync); 1] = [&signatures];
pg_execute(statement, &params).await
}
pub async fn mark_selected_transactions_processed( pub async fn mark_selected_transactions_processed(
batch: &SelectedMempoolBatch, batch: &SelectedMempoolBatch,
@ -6,48 +21,26 @@ pub async fn mark_selected_transactions_processed(
) -> Result<()> { ) -> Result<()> {
// Mark each selected mempool row as processed under the saved block // Mark each selected mempool row as processed under the saved block
// number so it can be cleaned up or restored later if needed. // number so it can be cleaned up or restored later if needed.
let client_handle = db_client().await?;
let client = client_handle.as_ref();
let bn = block_number as i32; let bn = block_number as i32;
// Selected batches are grouped by table, then marked with one UPDATE per // Selected batches are grouped by table, then marked with one UPDATE per
// table instead of touching rows one at a time. // table instead of touching rows one at a time.
mark_rows_by_ids(client, "transfer", &ids_for_table(batch, "transfer"), bn).await?; mark_rows_by_ids("transfer", &ids_for_table(batch, "transfer"), bn).await?;
mark_rows_by_ids(client, "token", &ids_for_table(batch, "token"), bn).await?; mark_rows_by_ids("token", &ids_for_table(batch, "token"), bn).await?;
mark_rows_by_ids("issue_token", &ids_for_table(batch, "issue_token"), bn).await?;
mark_rows_by_ids("burn", &ids_for_table(batch, "burn"), bn).await?;
mark_rows_by_ids("nft", &ids_for_table(batch, "nft"), bn).await?;
mark_rows_by_ids("marketing", &ids_for_table(batch, "marketing"), bn).await?;
mark_rows_by_ids( mark_rows_by_ids(
&client,
"issue_token",
&ids_for_table(batch, "issue_token"),
bn,
)
.await?;
mark_rows_by_ids(client, "burn", &ids_for_table(batch, "burn"), bn).await?;
mark_rows_by_ids(client, "nft", &ids_for_table(batch, "nft"), bn).await?;
mark_rows_by_ids(client, "marketing", &ids_for_table(batch, "marketing"), bn).await?;
mark_rows_by_ids(
&client,
"vanity_address", "vanity_address",
&ids_for_table(batch, "vanity_address"), &ids_for_table(batch, "vanity_address"),
bn, bn,
) )
.await?; .await?;
mark_rows_by_ids(client, "swap", &ids_for_table(batch, "swap"), bn).await?; mark_rows_by_ids("swap", &ids_for_table(batch, "swap"), bn).await?;
mark_rows_by_ids("loan_contract", &ids_for_table(batch, "loan_contract"), bn).await?;
mark_rows_by_ids("loan_payment", &ids_for_table(batch, "loan_payment"), bn).await?;
mark_rows_by_ids( mark_rows_by_ids(
&client,
"loan_contract",
&ids_for_table(batch, "loan_contract"),
bn,
)
.await?;
mark_rows_by_ids(
&client,
"loan_payment",
&ids_for_table(batch, "loan_payment"),
bn,
)
.await?;
mark_rows_by_ids(
&client,
"collateral_claim", "collateral_claim",
&ids_for_table(batch, "collateral_claim"), &ids_for_table(batch, "collateral_claim"),
bn, bn,
@ -60,36 +53,29 @@ pub async fn mark_selected_transactions_processed(
pub async fn restore_selected_transactions_processed(batch: &SelectedMempoolBatch) -> Result<()> { pub async fn restore_selected_transactions_processed(batch: &SelectedMempoolBatch) -> Result<()> {
// If block commit fails after selected rows were marked processed, // If block commit fails after selected rows were marked processed,
// restore them before the chain height can acknowledge the block. // restore them before the chain height can acknowledge the block.
let client_handle = db_client().await?; unmark_rows_by_ids("transfer", &ids_for_table(batch, "transfer")).await?;
let client = client_handle.as_ref(); unmark_rows_by_ids("token", &ids_for_table(batch, "token")).await?;
unmark_rows_by_ids("issue_token", &ids_for_table(batch, "issue_token")).await?;
unmark_rows_by_ids(client, "transfer", &ids_for_table(batch, "transfer")).await?; unmark_rows_by_ids("burn", &ids_for_table(batch, "burn")).await?;
unmark_rows_by_ids(client, "token", &ids_for_table(batch, "token")).await?; unmark_rows_by_ids("nft", &ids_for_table(batch, "nft")).await?;
unmark_rows_by_ids(client, "issue_token", &ids_for_table(batch, "issue_token")).await?; unmark_rows_by_ids("marketing", &ids_for_table(batch, "marketing")).await?;
unmark_rows_by_ids(client, "burn", &ids_for_table(batch, "burn")).await?;
unmark_rows_by_ids(client, "nft", &ids_for_table(batch, "nft")).await?;
unmark_rows_by_ids(client, "marketing", &ids_for_table(batch, "marketing")).await?;
unmark_rows_by_ids( unmark_rows_by_ids(
&client,
"vanity_address", "vanity_address",
&ids_for_table(batch, "vanity_address"), &ids_for_table(batch, "vanity_address"),
) )
.await?; .await?;
unmark_rows_by_ids(client, "swap", &ids_for_table(batch, "swap")).await?; unmark_rows_by_ids("swap", &ids_for_table(batch, "swap")).await?;
unmark_rows_by_ids( unmark_rows_by_ids(
&client,
"loan_contract", "loan_contract",
&ids_for_table(batch, "loan_contract"), &ids_for_table(batch, "loan_contract"),
) )
.await?; .await?;
unmark_rows_by_ids( unmark_rows_by_ids(
&client,
"loan_payment", "loan_payment",
&ids_for_table(batch, "loan_payment"), &ids_for_table(batch, "loan_payment"),
) )
.await?; .await?;
unmark_rows_by_ids( unmark_rows_by_ids(
&client,
"collateral_claim", "collateral_claim",
&ids_for_table(batch, "collateral_claim"), &ids_for_table(batch, "collateral_claim"),
) )
@ -105,25 +91,23 @@ pub async fn restore_processed_by_signatures(signatures: &[String]) -> Result<bo
return Ok(false); return Ok(false);
} }
let client_handle = db_client().await?;
let client = client_handle.as_ref();
let mut restored = 0_u64; let mut restored = 0_u64;
// Each table keeps its own signature columns, so rollback unmarks every // Each table keeps its own signature columns, so rollback unmarks every
// column that could contain one of the rolled-back signatures. // column that could contain one of the rolled-back signatures.
restored += unmark_by_signatures(client, "transfer", "signature", signatures).await?; restored += unmark_by_signatures("transfer", "signature", signatures).await?;
restored += unmark_by_signatures(client, "token", "signature", signatures).await?; restored += unmark_by_signatures("token", "signature", signatures).await?;
restored += unmark_by_signatures(client, "issue_token", "signature", signatures).await?; restored += unmark_by_signatures("issue_token", "signature", signatures).await?;
restored += unmark_by_signatures(client, "burn", "signature", signatures).await?; restored += unmark_by_signatures("burn", "signature", signatures).await?;
restored += unmark_by_signatures(client, "nft", "signature", signatures).await?; restored += unmark_by_signatures("nft", "signature", signatures).await?;
restored += unmark_by_signatures(client, "marketing", "signature", signatures).await?; restored += unmark_by_signatures("marketing", "signature", signatures).await?;
restored += unmark_by_signatures(client, "vanity_address", "signature", signatures).await?; restored += unmark_by_signatures("vanity_address", "signature", signatures).await?;
restored += unmark_by_signatures(client, "swap", "signature1", signatures).await?; restored += unmark_by_signatures("swap", "signature1", signatures).await?;
restored += unmark_by_signatures(client, "swap", "signature2", signatures).await?; restored += unmark_by_signatures("swap", "signature2", signatures).await?;
restored += unmark_by_signatures(client, "loan_contract", "signature1", signatures).await?; restored += unmark_by_signatures("loan_contract", "signature1", signatures).await?;
restored += unmark_by_signatures(client, "loan_contract", "signature2", signatures).await?; restored += unmark_by_signatures("loan_contract", "signature2", signatures).await?;
restored += unmark_by_signatures(client, "loan_payment", "signature", signatures).await?; restored += unmark_by_signatures("loan_payment", "signature", signatures).await?;
restored += unmark_by_signatures(client, "collateral_claim", "signature", signatures).await?; restored += unmark_by_signatures("collateral_claim", "signature", signatures).await?;
Ok(restored > 0) Ok(restored > 0)
} }
@ -167,90 +151,88 @@ pub async fn mark_processed_by_signatures(signatures: &[String], block_number: u
return Ok(()); return Ok(());
} }
let client_handle = db_client().await?;
let client = client_handle.as_ref();
let bn = block_number as i32; let bn = block_number as i32;
// Remote/synced blocks do not know local row IDs, so they mark by // Remote/synced blocks do not know local row IDs, so they mark by
// transaction signatures instead. // transaction signatures instead.
client pg_execute_block_signatures(
.execute( "UPDATE transfer SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)",
"UPDATE transfer SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE token SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)",
"UPDATE token SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE issue_token SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)",
"UPDATE issue_token SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE burn SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)",
"UPDATE burn SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE nft SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)",
"UPDATE nft SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE marketing SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)",
"UPDATE marketing SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE vanity_address SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)",
"UPDATE vanity_address SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE swap SET processed=true, processed_block_number=$1 WHERE signature1 = ANY($2)",
"UPDATE swap SET processed=true, processed_block_number=$1 WHERE signature1 = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE swap SET processed=true, processed_block_number=$1 WHERE signature2 = ANY($2)",
"UPDATE swap SET processed=true, processed_block_number=$1 WHERE signature2 = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE loan_contract SET processed=true, processed_block_number=$1 WHERE signature1 = ANY($2)",
"UPDATE loan_contract SET processed=true, processed_block_number=$1 WHERE signature1 = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE loan_contract SET processed=true, processed_block_number=$1 WHERE signature2 = ANY($2)",
"UPDATE loan_contract SET processed=true, processed_block_number=$1 WHERE signature2 = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE loan_payment SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)",
"UPDATE loan_payment SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
client pg_execute_block_signatures(
.execute( "UPDATE collateral_claim SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)",
"UPDATE collateral_claim SET processed=true, processed_block_number=$1 WHERE signature = ANY($2)", bn,
&[&bn, &signatures], signatures,
) )
.await?; .await?;
Ok(()) Ok(())
} }
@ -262,71 +244,47 @@ pub async fn delete_by_signatures(signatures: &[String]) -> Result<()> {
return Ok(()); return Ok(());
} }
let client_handle = db_client().await?;
let client = client_handle.as_ref();
// Failed validation removes every matching pending row no matter which // Failed validation removes every matching pending row no matter which
// transaction table currently owns the signature. // transaction table currently owns the signature.
client pg_execute_signatures("DELETE FROM transfer WHERE signature = ANY($1)", signatures).await?;
.execute( pg_execute_signatures("DELETE FROM token WHERE signature = ANY($1)", signatures).await?;
"DELETE FROM transfer WHERE signature = ANY($1)", pg_execute_signatures(
&[&signatures], "DELETE FROM issue_token WHERE signature = ANY($1)",
) signatures,
.await?; )
client .await?;
.execute( pg_execute_signatures("DELETE FROM burn WHERE signature = ANY($1)", signatures).await?;
"DELETE FROM token WHERE signature = ANY($1)", pg_execute_signatures("DELETE FROM nft WHERE signature = ANY($1)", signatures).await?;
&[&signatures], pg_execute_signatures(
) "DELETE FROM marketing WHERE signature = ANY($1)",
.await?; signatures,
client )
.execute( .await?;
"DELETE FROM issue_token WHERE signature = ANY($1)", pg_execute_signatures(
&[&signatures], "DELETE FROM vanity_address WHERE signature = ANY($1)",
) signatures,
.await?; )
client .await?;
.execute("DELETE FROM burn WHERE signature = ANY($1)", &[&signatures]) pg_execute_signatures(
.await?; "DELETE FROM swap WHERE signature1 = ANY($1) OR signature2 = ANY($1)",
client signatures,
.execute("DELETE FROM nft WHERE signature = ANY($1)", &[&signatures]) )
.await?; .await?;
client pg_execute_signatures(
.execute( "DELETE FROM loan_contract WHERE signature1 = ANY($1) OR signature2 = ANY($1)",
"DELETE FROM marketing WHERE signature = ANY($1)", signatures,
&[&signatures], )
) .await?;
.await?; pg_execute_signatures(
client "DELETE FROM loan_payment WHERE signature = ANY($1)",
.execute( signatures,
"DELETE FROM vanity_address WHERE signature = ANY($1)", )
&[&signatures], .await?;
) pg_execute_signatures(
.await?; "DELETE FROM collateral_claim WHERE signature = ANY($1)",
client signatures,
.execute( )
"DELETE FROM swap WHERE signature1 = ANY($1) OR signature2 = ANY($1)", .await?;
&[&signatures],
)
.await?;
client
.execute(
"DELETE FROM loan_contract WHERE signature1 = ANY($1) OR signature2 = ANY($1)",
&[&signatures],
)
.await?;
client
.execute(
"DELETE FROM loan_payment WHERE signature = ANY($1)",
&[&signatures],
)
.await?;
client
.execute(
"DELETE FROM collateral_claim WHERE signature = ANY($1)",
&[&signatures],
)
.await?;
Ok(()) Ok(())
} }

37
src/records/memory/mempool/schema.rs
View File

@ -31,6 +31,17 @@ pub async fn db_client() -> Result<Arc<Client>> {
Ok(slot.read().await.clone()) Ok(slot.read().await.clone())
} }
fn is_closed_postgres_connection(err: &tokio_postgres::Error) -> bool {
if err.is_closed() {
return true;
}
let message = err.to_string().to_ascii_lowercase();
message.contains("connection closed")
|| message.contains("connection reset")
|| message.contains("broken pipe")
}
pub async fn reconnect_db() -> Result<()> { pub async fn reconnect_db() -> Result<()> {
let client = connect_client().await?; let client = connect_client().await?;
@ -55,6 +66,32 @@ pub async fn ensure_db_connection() -> Result<()> {
reconnect_db().await reconnect_db().await
} }
pub async fn pg_execute(statement: &str, params: &[&(dyn ToSql + Sync)]) -> Result<u64> {
let client_handle = db_client().await?;
match client_handle.execute(statement, params).await {
Ok(count) => Ok(count),
Err(err) if is_closed_postgres_connection(&err) => {
reconnect_db().await?;
let client_handle = db_client().await?;
Ok(client_handle.execute(statement, params).await?)
}
Err(err) => Err(err.into()),
}
}
pub async fn pg_query(statement: &str, params: &[&(dyn ToSql + Sync)]) -> Result<Vec<Row>> {
let client_handle = db_client().await?;
match client_handle.query(statement, params).await {
Ok(rows) => Ok(rows),
Err(err) if is_closed_postgres_connection(&err) => {
reconnect_db().await?;
let client_handle = db_client().await?;
Ok(client_handle.query(statement, params).await?)
}
Err(err) => Err(err.into()),
}
}
pub async fn init_db() -> Result<()> { pub async fn init_db() -> Result<()> {
// Initialize the shared Postgres client used by the mempool tables. // Initialize the shared Postgres client used by the mempool tables.
if DB.get().is_some() { if DB.get().is_some() {

31
src/records/memory/mempool/selection.rs
View File

@ -18,10 +18,8 @@ fn index_selected_wallet_transaction(
pub async fn select_transactions_for_block(limit: i64) -> Result<SelectedMempoolBatch> { pub async fn select_transactions_for_block(limit: i64) -> Result<SelectedMempoolBatch> {
// Pull the highest-priority unprocessed rows across all mempool // Pull the highest-priority unprocessed rows across all mempool
// tables, keeping the original bytes for block-file streaming later. // tables, keeping the original bytes for block-file streaming later.
let client_handle = db_client().await?; let rows =
let client = client_handle.as_ref(); pg_query(
let rows = client
.query(
r#" r#"
SELECT * FROM ( SELECT * FROM (
SELECT SELECT
@ -1016,9 +1014,6 @@ pub async fn stream_selected_transaction_originals(
} }
pub async fn delete_selected_transactions(batch: &SelectedMempoolBatch) -> Result<()> { pub async fn delete_selected_transactions(batch: &SelectedMempoolBatch) -> Result<()> {
let client_handle = db_client().await?;
let client = client_handle.as_ref();
// Each transaction kind still lives in a separate SQL table, so deletion // Each transaction kind still lives in a separate SQL table, so deletion
// groups the selected IDs by table after the block has been written. // groups the selected IDs by table after the block has been written.
let transfer_ids = ids_for_table(batch, "transfer"); let transfer_ids = ids_for_table(batch, "transfer");
@ -1033,17 +1028,17 @@ pub async fn delete_selected_transactions(batch: &SelectedMempoolBatch) -> Resul
let borrower_ids = ids_for_table(batch, "loan_payment"); let borrower_ids = ids_for_table(batch, "loan_payment");
let collateral_ids = ids_for_table(batch, "collateral_claim"); let collateral_ids = ids_for_table(batch, "collateral_claim");
delete_rows(client, "transfer", &transfer_ids).await?; delete_rows("transfer", &transfer_ids).await?;
delete_rows(client, "token", &token_ids).await?; delete_rows("token", &token_ids).await?;
delete_rows(client, "issue_token", &issue_token_ids).await?; delete_rows("issue_token", &issue_token_ids).await?;
delete_rows(client, "burn", &burn_ids).await?; delete_rows("burn", &burn_ids).await?;
delete_rows(client, "nft", &nft_ids).await?; delete_rows("nft", &nft_ids).await?;
delete_rows(client, "marketing", &marketing_ids).await?; delete_rows("marketing", &marketing_ids).await?;
delete_rows(client, "vanity_address", &vanity_ids).await?; delete_rows("vanity_address", &vanity_ids).await?;
delete_rows(client, "swap", &swap_ids).await?; delete_rows("swap", &swap_ids).await?;
delete_rows(client, "loan_contract", &lender_ids).await?; delete_rows("loan_contract", &lender_ids).await?;
delete_rows(client, "loan_payment", &borrower_ids).await?; delete_rows("loan_payment", &borrower_ids).await?;
delete_rows(client, "collateral_claim", &collateral_ids).await?; delete_rows("collateral_claim", &collateral_ids).await?;
Ok(()) Ok(())
} }

2
src/rpc/commands/receive_torrent.rs
View File

@ -256,7 +256,7 @@ pub async fn torrent_submission(
} }
}; };
let torrent_hash = skein_128_hash_bytes(&torrent_bytes); let torrent_hash = skein_128_hash_bytes(&torrent_bytes);
if has_recent_torrent(&torrent_hash, height).await { if has_recent_torrent(&torrent_hash).await {
// Recent-torrent cache prevents rebroadcast loops from repeatedly // Recent-torrent cache prevents rebroadcast loops from repeatedly
// staging and verifying the same metadata. // staging and verifying the same metadata.
let msg = "Torrent already seen.".to_string().as_bytes().to_vec(); let msg = "Torrent already seen.".to_string().as_bytes().to_vec();

2
src/startup/daemonize.rs
View File

@ -233,7 +233,7 @@ pub fn handle_control_command() -> Result<bool, Box<dyn std::error::Error>> {
} }
#[cfg(not(unix))] #[cfg(not(unix))]
pub fn install_shutdown_cleanup(_db: crate::sled::Db) {} pub fn install_shutdown_cleanup() {}
#[cfg(not(unix))] #[cfg(not(unix))]
pub fn remove_registered_pid_file() {} pub fn remove_registered_pid_file() {}

3
src/startup/node_runtime.rs
View File

@ -119,7 +119,10 @@ pub async fn run_unlocked_node(wallet: Arc<Wallet>, install_shutdown: bool) -> R
if install_shutdown { if install_shutdown {
// Console/daemon mode owns signal cleanup; Windows service shutdown is handled separately. // Console/daemon mode owns signal cleanup; Windows service shutdown is handled separately.
#[cfg(unix)]
install_shutdown_cleanup(db.clone()); install_shutdown_cleanup(db.clone());
#[cfg(not(unix))]
install_shutdown_cleanup();
} }
if let Err(e) = clear_ip_scores(&db) { if let Err(e) = clear_ip_scores(&db) {

4
src/startup/windows_service.rs
View File

@ -419,7 +419,9 @@ fn stop_service_if_running(
} }
#[cfg(windows)] #[cfg(windows)]
fn contractless_service_main(_arguments: Vec<OsString>) { fn contractless_service_main(arguments: Vec<OsString>) {
drop(arguments);
// SCM enters here instead of the normal console main path. // SCM enters here instead of the normal console main path.
if let Err(err) = run_service() { if let Err(err) = run_service() {
eprintln!("Windows service error: {err}"); eprintln!("Windows service error: {err}");

2
src/torrent/torrenting_system/torrent_cache.rs
View File

@ -8,7 +8,7 @@ lazy_static! {
Mutex::new(HashMap::new()); Mutex::new(HashMap::new());
} }
pub async fn has_recent_torrent(hash: &str, _block_height: u32) -> bool { pub async fn has_recent_torrent(hash: &str) -> bool {
// The hash alone identifies duplicate torrent bytes in the recent in-memory cache. // The hash alone identifies duplicate torrent bytes in the recent in-memory cache.
let map = RECENT_TORRENTS.lock().await; let map = RECENT_TORRENTS.lock().await;
map.contains_key(hash) map.contains_key(hash)

2
src/verifications/async_funcs/transactions.rs
View File

@ -73,7 +73,7 @@ async fn verify_transaction(
balance_tracker: Arc<Mutex<BlockBalanceTracker>>, balance_tracker: Arc<Mutex<BlockBalanceTracker>>,
) -> Result<String, String> { ) -> Result<String, String> {
if let Transaction::Genesis(genesis_tx) = &transaction { if let Transaction::Genesis(genesis_tx) = &transaction {
match genesis_tx.verify(miner, db).await { match genesis_tx.verify().await {
Ok(value) => { Ok(value) => {
reserve_verified_transaction(db, &transaction, balance_tracker).await?; reserve_verified_transaction(db, &transaction, balance_tracker).await?;
return Ok(value); return Ok(value);

4
src/verifications/async_funcs/verify_genesis.rs
View File

@ -1,9 +1,8 @@
use crate::blocks::genesis::GenesisTransaction; use crate::blocks::genesis::GenesisTransaction;
use crate::common::check_genesis::genesis_checkup; use crate::common::check_genesis::genesis_checkup;
use crate::sled::Db;
impl GenesisTransaction { impl GenesisTransaction {
pub async fn verify(&self, _miner: String, db: &Db) -> Result<String, String> { pub async fn verify(&self) -> Result<String, String> {
// Genesis is pinned to a fixed message and only valid before the // Genesis is pinned to a fixed message and only valid before the
// chain already has a committed genesis block. // chain already has a committed genesis block.
let msg = &self.unsigned.message; let msg = &self.unsigned.message;
@ -11,7 +10,6 @@ impl GenesisTransaction {
return Err("Not a valid Genesis block").map_err(|s| s.to_string())?; return Err("Not a valid Genesis block").map_err(|s| s.to_string())?;
} }
let _ = db;
if genesis_checkup().await { if genesis_checkup().await {
return Err("Genesis block already exists.".to_string()); return Err("Genesis block already exists.".to_string());
} }

6
src/wallets/create_keys.rs
View File

@ -3,7 +3,7 @@ use crate::FnDsaKeyPair;
use crate::{decode, encode}; use crate::{decode, encode};
impl Wallet { impl Wallet {
pub fn generate_keypair(_network_byte: u8) -> (Vec<u8>, String) { pub fn generate_keypair() -> (Vec<u8>, String) {
loop { loop {
// Generate a new FN-DSA key pair using the wallet security parameter. // Generate a new FN-DSA key pair using the wallet security parameter.
let keypair = FnDsaKeyPair::generate(Self::FN_DSA_LOGN) let keypair = FnDsaKeyPair::generate(Self::FN_DSA_LOGN)
@ -25,9 +25,6 @@ impl Wallet {
} }
pub fn regenerate_public_key(private_key_hex: &str) -> Result<Vec<u8>, String> { pub fn regenerate_public_key(private_key_hex: &str) -> Result<Vec<u8>, String> {
// Regenerated public keys always use the network selected at compile time.
let network_byte = Self::current_network_byte();
// Decode the stored private key from its wallet hex representation. // Decode the stored private key from its wallet hex representation.
let private_key_bytes = match decode(private_key_hex) { let private_key_bytes = match decode(private_key_hex) {
Ok(bytes) => bytes, Ok(bytes) => bytes,
@ -47,7 +44,6 @@ impl Wallet {
let keypair = FnDsaKeyPair::from_private_key(&private_key_bytes) let keypair = FnDsaKeyPair::from_private_key(&private_key_bytes)
.map_err(|e| format!("Failed to decode the provided FN-DSA private key: {e}"))?; .map_err(|e| format!("Failed to decode the provided FN-DSA private key: {e}"))?;
let _ = network_byte;
let public_key_bytes = keypair.public_key().to_vec(); let public_key_bytes = keypair.public_key().to_vec();
if !Self::has_valid_public_key_bytes(&public_key_bytes) { if !Self::has_valid_public_key_bytes(&public_key_bytes) {
return Err("FN-DSA public key does not satisfy this network's wallet key rule." return Err("FN-DSA public key does not satisfy this network's wallet key rule."

6
src/wallets/create_wallet.rs
View File

@ -5,9 +5,9 @@ use crate::Path;
use crate::{create_img, encrypts}; use crate::{create_img, encrypts};
impl Wallet { impl Wallet {
pub fn create_wallet(&self, network_byte: u8) -> Wallet { pub fn create_wallet(&self) -> Wallet {
// Generate a fresh FN-DSA key pair for the selected network. // Generate a fresh FN-DSA key pair for the selected network.
let (public_key, private_key) = Self::generate_keypair(network_byte); let (public_key, private_key) = Self::generate_keypair();
// Hash the public key body into the fixed 22-byte short-address format. // Hash the public key body into the fixed 22-byte short-address format.
let short_address_bytes = Self::public_key_bytes_to_short_address_bytes(&public_key) let short_address_bytes = Self::public_key_bytes_to_short_address_bytes(&public_key)
@ -77,7 +77,7 @@ impl Wallet {
}; };
// Generate the wallet for the active network. // Generate the wallet for the active network.
let wallet = Wallet::create_wallet(&wallet, Self::current_network_byte()); let wallet = Wallet::create_wallet(&wallet);
// Save the generated wallet file before loading and decrypting it through the normal path. // Save the generated wallet file before loading and decrypting it through the normal path.
wallet.saved.save_the_wallet(wallet_path).await; wallet.saved.save_the_wallet(wallet_path).await;