Skip to content

refactor: DB snapshot copy specs into descriptors #7375

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
francesco-stacks wants to merge 7 commits into
base: develop
Choose a base branch
from
Open

refactor: DB snapshot copy specs into descriptors #7375

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
ee19ecb
unify snapshot copy specs behind DbSnapshotSpec trait with bound params
francesco-stacks Jul 2, 2026
72afda3
Merge branch 'develop' into feat/marf-snapshot-copy-spec-refactor
francesco-stacks Jul 2, 2026
2bd181c
crc: return 0 rows for schema only tables
francesco-stacks Jul 2, 2026
0ebf1d2
crc: generic per-DB bind + instance-independent classification
francesco-stacks Jul 13, 2026
5aeeaeb
crc: single spec source + memo bind tagging
francesco-stacks Jul 13, 2026
646f48e
Merge branch 'develop' into feat/marf-snapshot-copy-spec-refactor
francesco-stacks Jul 13, 2026
175bb09
crc: use classify_hint instead of hardcoding strings
francesco-stacks Jul 13, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 0 additions & 21 deletions stackslib/src/burnchains/db.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -943,27 +943,6 @@ impl BurnchainDB {
Ok(None)
}
}

/// Count the burn header hashes yielded by `canonical_sql` (a SELECT
/// producing one TEXT column) that have no `burnchain_db_block_headers`
/// row in `headers_schema`. Both arguments are interpolated into SQL;
/// pass only trusted fixed fragments.
pub(crate) fn count_canonical_burn_hashes_missing_from(
conn: &Connection,
headers_schema: &str,
canonical_sql: &str,
) -> Result<u64, DBError> {
conn.query_row(
&format!(
"SELECT COUNT(*) FROM ({canonical_sql}) \
WHERE burn_header_hash NOT IN \
(SELECT block_hash FROM {headers_schema}.burnchain_db_block_headers)"
),
NO_PARAMS,
|row| row.get(0),
)
.map_err(DBError::from)
}
}

// Raw-row test fixture writers. Each helper owns its table's column
Expand Down
100 changes: 0 additions & 100 deletions stackslib/src/chainstate/burn/db/sortdb.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4614,106 +4614,6 @@ impl SortitionDB {
query_row(conn, qry, NO_PARAMS).map(|opt| opt.expect("CORRUPTION: No burnchain tips"))
}

/// Get the distinct burn header hashes of all snapshots, forks included.
/// Only on a squashed sortition DB is this exactly the canonical
/// burnchain.
pub fn get_all_snapshot_burn_header_hashes(
conn: &Connection,
) -> Result<Vec<BurnchainHeaderHash>, db_error> {
let mut stmt = conn.prepare("SELECT DISTINCT burn_header_hash FROM snapshots")?;
let rows = stmt.query_map(NO_PARAMS, |row| row.get(0))?;
rows.collect::<Result<Vec<_>, _>>().map_err(db_error::from)
}

/// Get the burn header hash of the snapshot with the given sortition
/// ID, or `None` if no such snapshot exists. Returns the raw stored
/// TEXT so callers can preserve the value byte-for-byte.
pub fn get_snapshot_burn_header_hash(
conn: &Connection,
sortition_id: &SortitionId,
) -> Result<Option<String>, db_error> {
conn.prepare_cached("SELECT burn_header_hash FROM snapshots WHERE sortition_id = ?1")?
.query_row(params![sortition_id], |row| row.get(0))
.optional()
.map_err(db_error::from)
}

/// Source-projection SQL for copying a Stacks-tip memo table
/// (`stacks_chain_tips`, or `stacks_chain_tips_by_burn_view` with
/// `include_burn_view`) from the schema `from`, keeping rows whose
/// `sortition_id` is in `sortition_id_sql`. With a `boundary`, memo
/// rows above its Stacks height are rewritten down to the anchor (see
/// [`SortitionTipCopyBoundary`])
pub(crate) fn stacks_tip_memo_copy_sql(
table: &str,
from: &str,
sortition_id_sql: &str,
include_burn_view: bool,
boundary: Option<&SortitionTipCopyBoundary>,
) -> String {
let Some(boundary) = boundary else {
return format!(
"SELECT * FROM {from}.{table} WHERE sortition_id IN ({sortition_id_sql})"
);
};
let max_height = boundary.max_stacks_height;
let anchor_height = boundary.anchor_block_height;
let anchor_ch = &boundary.anchor_consensus_hash;
let anchor_bhh = &boundary.anchor_block_hash;
let anchor_burn_view_ch = &boundary.anchor_burn_view_consensus_hash;
let burn_view_column = if include_burn_view {
format!(
"CASE WHEN block_height > {max_height} THEN '{anchor_burn_view_ch}' \
ELSE burn_view_consensus_hash END, "
)
} else {
String::new()
};
let anchor_match = if include_burn_view {
format!(
"(consensus_hash = '{anchor_ch}' \
AND burn_view_consensus_hash = '{anchor_burn_view_ch}')"
)
} else {
format!("consensus_hash = '{anchor_ch}'")
};
format!(
"SELECT sortition_id, \
CASE WHEN block_height > {max_height} THEN '{anchor_ch}' ELSE consensus_hash END, \
{burn_view_column}\
CASE WHEN block_height > {max_height} THEN '{anchor_bhh}' ELSE block_hash END, \
CASE WHEN block_height > {max_height} THEN {anchor_height} ELSE block_height END \
FROM {from}.{table} \
WHERE sortition_id IN ({sortition_id_sql}) \
AND (block_height <= {max_height} OR {anchor_match})"
)
}

/// Whether every Stacks-tip memo row (in both memo tables) sits at or
/// below the boundary's Stacks height. A `None` boundary trivially
/// passes. Validation counterpart of
/// [`Self::stacks_tip_memo_copy_sql`]'s height rewrite.
pub(crate) fn stacks_tip_memos_within_boundary(
conn: &Connection,
boundary: Option<&SortitionTipCopyBoundary>,
) -> Result<bool, db_error> {
let Some(boundary) = boundary else {
return Ok(true);
};
let max_height = u64_to_sql(boundary.max_stacks_height)?;
// Check if ANY above-boundary memo row exists.
conn.query_row(
"SELECT NOT EXISTS( \
SELECT 1 FROM stacks_chain_tips WHERE block_height > ?1 \
UNION ALL \
SELECT 1 FROM stacks_chain_tips_by_burn_view WHERE block_height > ?1 \
)",
params![max_height],
|row| row.get(0),
)
.map_err(db_error::from)
}

/// Get the canonical burn chain tip -- the tip of the longest burn chain we know about.
/// Break ties deterministically by ordering on burnchain block hash.
pub fn get_canonical_chain_tip_bhh(conn: &Connection) -> Result<BurnchainHeaderHash, db_error> {
Expand Down
100 changes: 54 additions & 46 deletions stackslib/src/chainstate/stacks/db/snapshot/blocks.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,13 @@ use std::collections::HashSet;
use std::fs;
use std::path::Path;

use rusqlite::types::Value;
use rusqlite::{params, Connection, OpenFlags};
use stacks_common::types::chainstate::{BlockHeaderHash, ConsensusHash, StacksBlockId};

use super::common::{
assert_source_schema, clone_schemas_from_source, copied_rows, execute_copy_specs,
with_offline_write_session, TableCopySpec,
classify_hint, clone_schemas_from_source, copied_rows, execute_copy_specs,
with_offline_write_session, DbSnapshotSpec, NoBind, TableCopySpec,
};
use crate::chainstate::stacks::db::StacksChainState;
use crate::chainstate::stacks::index::Error;
Expand Down Expand Up @@ -53,26 +54,36 @@ pub struct NakamotoBlockCopyStats {
pub total_blob_bytes: u64,
}

/// Tables copied from the source Nakamoto staging-blocks DB. The index-side
/// The Nakamoto staging (`nakamoto.sqlite`) snapshot spec. The index-side
/// staging tables (`staging_microblocks*`) come from the index DB and are
/// classified in `index.rs`.
pub(super) const NAKAMOTO_STAGING_TABLES: &[&str] = &["nakamoto_staging_blocks", "db_version"];
pub(super) struct NakamotoStagingDbSnapshotSpec;

/// Every table the Nakamoto staging snapshot accounts for. nakamoto.sqlite is not
/// MARF-backed, so unlike the other slices no MARF infra tables are exempted.
fn known_nakamoto_staging_tables() -> Vec<&'static str> {
NAKAMOTO_STAGING_TABLES.to_vec()
impl DbSnapshotSpec for NakamotoStagingDbSnapshotSpec {
type Bind = NoBind;

fn copy_spec_list() -> &'static [TableCopySpec<NoBind>] {
nakamoto_copy_specs()
}

fn db_label() -> &'static str {
"Nakamoto staging DB"
}

fn classify_hint() -> &'static str {
classify_hint!(nakamoto_copy_specs)
}

fn bind_params(&self, bind: NoBind) -> Result<Vec<Value>, Error> {
match bind {}
}
}

/// The blocks snapshot's source-schema guard (see [`assert_source_schema`]);
/// The blocks snapshot's source-schema guard (see
/// [`DbSnapshotSpec::assert_source_classified`]);
/// `test_no_unclassified_nakamoto_staging_tables` runs it against a fresh schema.
pub(super) fn assert_source_tables_classified(src_conn: &Connection) -> Result<(), Error> {
assert_source_schema(
src_conn,
&known_nakamoto_staging_tables(),
"Nakamoto staging DB",
"NAKAMOTO_STAGING_TABLES in snapshot/blocks.rs",
)
NakamotoStagingDbSnapshotSpec::assert_source_classified(src_conn)
}

/// Return the `(sequence, microblock_hash)` rows of processed,
Expand Down Expand Up @@ -207,23 +218,22 @@ fn populate_microblock_temp_tables(

/// Copy specs for the confirmed-microblock tables, filtered by the temp
/// tables [`populate_microblock_temp_tables`] builds.
fn microblock_copy_specs() -> Vec<TableCopySpec> {
vec![
TableCopySpec {
table: "staging_microblocks",
source_sql: "SELECT s.* FROM src.staging_microblocks s \
fn microblock_copy_specs() -> &'static [TableCopySpec<NoBind>] {
static SPECS: &[TableCopySpec<NoBind>] = &[
TableCopySpec::sql(
"staging_microblocks",
"SELECT s.* FROM src.staging_microblocks s \
WHERE s.microblock_hash IN (SELECT hash FROM temp.selected_microblocks) \
AND s.index_block_hash IN (SELECT ibh FROM temp.selected_parents) \
AND s.orphaned = 0"
.into(),
},
TableCopySpec {
table: "staging_microblocks_data",
source_sql: "SELECT s.* FROM src.staging_microblocks_data s \
WHERE s.block_hash IN (SELECT hash FROM temp.selected_microblocks)"
.into(),
},
]
AND s.orphaned = 0",
),
TableCopySpec::sql(
"staging_microblocks_data",
"SELECT s.* FROM src.staging_microblocks_data s \
WHERE s.block_hash IN (SELECT hash FROM temp.selected_microblocks)",
),
];
SPECS
}

/// Copy confirmed canonical epoch-2 microblock streams into the squashed index.
Expand All @@ -243,7 +253,7 @@ pub fn copy_confirmed_epoch2_microblocks(
if !selected_hashes.is_empty() {
populate_microblock_temp_tables(conn, &selected_hashes, &selected_parents)?;

let results = execute_copy_specs(conn, &microblock_copy_specs())?;
let results = execute_copy_specs(conn, microblock_copy_specs(), |bind| match bind {})?;
stats.microblock_rows_copied = copied_rows(&results, "staging_microblocks");

stats.microblock_bytes_copied = conn.query_row(
Expand Down Expand Up @@ -322,21 +332,18 @@ pub fn copy_epoch2_block_files(
}

/// Copy specs for the Nakamoto staging DB.
pub(super) fn nakamoto_copy_specs() -> Vec<TableCopySpec> {
vec![
TableCopySpec {
table: "db_version",
source_sql: "SELECT * FROM src.db_version".into(),
},
TableCopySpec {
table: "nakamoto_staging_blocks",
source_sql: "SELECT s.* FROM src.nakamoto_staging_blocks s \
pub(super) fn nakamoto_copy_specs() -> &'static [TableCopySpec<NoBind>] {
static SPECS: &[TableCopySpec<NoBind>] = &[
TableCopySpec::sql("db_version", "SELECT * FROM src.db_version"),
TableCopySpec::sql(
"nakamoto_staging_blocks",
"SELECT s.* FROM src.nakamoto_staging_blocks s \
WHERE s.orphaned = 0 \
AND s.index_block_hash IN \
(SELECT index_block_hash FROM idx.nakamoto_block_headers)"
.into(),
},
]
(SELECT index_block_hash FROM idx.nakamoto_block_headers)",
),
];
SPECS
}

/// Create and populate `nakamoto.sqlite` with canonical `nakamoto_staging_blocks` rows.
Expand Down Expand Up @@ -369,9 +376,10 @@ pub fn copy_nakamoto_staging_blocks(
&[("src", src_nakamoto_path), ("idx", squashed_index_path)],
"",
|conn| {
clone_schemas_from_source(conn, NAKAMOTO_STAGING_TABLES)?;
let spec = NakamotoStagingDbSnapshotSpec;
clone_schemas_from_source(conn, &NakamotoStagingDbSnapshotSpec::table_names())?;

let results = execute_copy_specs(conn, &nakamoto_copy_specs())?;
let results = spec.run_copy(conn)?;

let total_blob_bytes: i64 = conn.query_row(
"SELECT COALESCE(SUM(LENGTH(data)), 0) FROM nakamoto_staging_blocks",
Expand Down
Loading
Loading