Skip to content

Move persist into async part of the sweeper #3819

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
wants to merge 2 commits into
base: main
Choose a base branch
from
+77 −54
Open

Move persist into async part of the sweeper #3819

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c09d41c
Separate sweeper persistent state
joostjager Jun 9, 2025
84ce2f2
Persist sweeper state as part of background process
joostjager Jun 9, 2025
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
131 changes: 77 additions & 54 deletions lightning/src/util/sweep.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -382,7 +382,10 @@ where
output_spender: O, change_destination_source: D, kv_store: K, logger: L,
) -> Self {
let outputs = Vec::new();
let sweeper_state = Mutex::new(SweeperState { outputs, best_block });
let sweeper_state = Mutex::new(SweeperState {
persistent: PersistentSweeperState { outputs, best_block },
dirty: false,
});
Comment on lines +385 to +388
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

IMO adding the wrapper struct causes us to litter the code with .persistent in a lot of places where it's not relevant what's written to disk and what's not, and I'm not sure the concrete benefit besides the principle. Not worth holding up the PR though.

Self {
sweeper_state,
pending_sweep: AtomicBool::new(false),
Expand Down Expand Up @@ -437,27 +440,30 @@ where
},
};

if state_lock.outputs.iter().find(|o| o.descriptor == output_info.descriptor).is_some()
{
let mut outputs = state_lock.persistent.outputs.iter();
if outputs.find(|o| o.descriptor == output_info.descriptor).is_some() {
continue;
}

state_lock.outputs.push(output_info);
state_lock.persistent.outputs.push(output_info);
}
self.persist_state(&*state_lock).map_err(|e| {
self.persist_state(&state_lock.persistent).map_err(|e| {
log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
})
})?;
state_lock.dirty = false;

Ok(())
}

/// Returns a list of the currently tracked spendable outputs.
pub fn tracked_spendable_outputs(&self) -> Vec<TrackedSpendableOutput> {
self.sweeper_state.lock().unwrap().outputs.clone()
self.sweeper_state.lock().unwrap().persistent.outputs.clone()
}

/// Gets the latest best block which was connected either via the [`Listen`] or
/// [`Confirm`] interfaces.
pub fn current_best_block(&self) -> BestBlock {
self.sweeper_state.lock().unwrap().best_block
self.sweeper_state.lock().unwrap().persistent.best_block
}

/// Regenerates and broadcasts the spending transaction for any outputs that are pending. This method will be a
Expand Down Expand Up @@ -503,11 +509,20 @@ where

// See if there is anything to sweep before requesting a change address.
{
let sweeper_state = self.sweeper_state.lock().unwrap();
let mut sweeper_state = self.sweeper_state.lock().unwrap();

let cur_height = sweeper_state.best_block.height;
let has_respends = sweeper_state.outputs.iter().any(|o| filter_fn(o, cur_height));
let cur_height = sweeper_state.persistent.best_block.height;
let has_respends =
sweeper_state.persistent.outputs.iter().any(|o| filter_fn(o, cur_height));
if !has_respends {
// If there is nothing to sweep, we still persist the state if it is dirty.
if sweeper_state.dirty {
self.persist_state(&sweeper_state.persistent).map_err(|e| {
log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
})?;
sweeper_state.dirty = false;
}

return Ok(());
}
}
Expand All @@ -520,23 +535,29 @@ where
{
let mut sweeper_state = self.sweeper_state.lock().unwrap();

let cur_height = sweeper_state.best_block.height;
let cur_hash = sweeper_state.best_block.block_hash;
let cur_height = sweeper_state.persistent.best_block.height;
let cur_hash = sweeper_state.persistent.best_block.block_hash;

let respend_descriptors: Vec<&SpendableOutputDescriptor> = sweeper_state
.persistent
.outputs
.iter()
.filter(|o| filter_fn(*o, cur_height))
.map(|o| &o.descriptor)
.collect();

if respend_descriptors.is_empty() {
// It could be that a tx confirmed and there is now nothing to sweep anymore.
// It could be that a tx confirmed and there is now nothing to sweep anymore. If there is dirty state,
// we'll persist it in the next cycle.
return Ok(());
}

let spending_tx = self
.spend_outputs(&sweeper_state, &respend_descriptors, change_destination_script)
.spend_outputs(
&sweeper_state.persistent,
&respend_descriptors,
change_destination_script,
)
.map_err(|e| {
log_error!(self.logger, "Error spending outputs: {:?}", e);
})?;
Expand All @@ -550,7 +571,7 @@ where
// As we didn't modify the state so far, the same filter_fn yields the same elements as
// above.
let respend_outputs =
sweeper_state.outputs.iter_mut().filter(|o| filter_fn(&**o, cur_height));
sweeper_state.persistent.outputs.iter_mut().filter(|o| filter_fn(&**o, cur_height));
for output_info in respend_outputs {
if let Some(filter) = self.chain_data_source.as_ref() {
let watched_output = output_info.to_watched_output(cur_hash);
Expand All @@ -560,9 +581,10 @@ where
output_info.status.broadcast(cur_hash, cur_height, spending_tx.clone());
}

self.persist_state(&sweeper_state).map_err(|e| {
self.persist_state(&sweeper_state.persistent).map_err(|e| {
log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
})?;
sweeper_state.dirty = false;

self.broadcaster.broadcast_transactions(&[&spending_tx]);
}
Expand All @@ -571,10 +593,10 @@ where
}

fn prune_confirmed_outputs(&self, sweeper_state: &mut SweeperState) {
let cur_height = sweeper_state.best_block.height;
let cur_height = sweeper_state.persistent.best_block.height;

// Prune all outputs that have sufficient depth by now.
sweeper_state.outputs.retain(|o| {
sweeper_state.persistent.outputs.retain(|o| {
if let Some(confirmation_height) = o.status.confirmation_height() {
// We wait at least `PRUNE_DELAY_BLOCKS` as before that
// `Event::SpendableOutputs` from lingering monitors might get replayed.
Expand All @@ -588,9 +610,11 @@ where
}
true
});

sweeper_state.dirty = true;
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems this is redundant, but probably also can't hurt.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Leaving as is, to at least flag dirty in the functions that change state.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also not doing the RAII pattern. There are just a few places where dirty needs to be cleared, and seems fine for now.

}

fn persist_state(&self, sweeper_state: &SweeperState) -> Result<(), io::Error> {
fn persist_state(&self, sweeper_state: &PersistentSweeperState) -> Result<(), io::Error> {
self.kv_store
.write(
OUTPUT_SWEEPER_PERSISTENCE_PRIMARY_NAMESPACE,
Expand All @@ -612,7 +636,7 @@ where
}

fn spend_outputs(
&self, sweeper_state: &SweeperState, descriptors: &[&SpendableOutputDescriptor],
&self, sweeper_state: &PersistentSweeperState, descriptors: &[&SpendableOutputDescriptor],
change_destination_script: ScriptBuf,
) -> Result<Transaction, ()> {
let tx_feerate =
Expand All @@ -635,19 +659,23 @@ where
) {
let confirmation_hash = header.block_hash();
for (_, tx) in txdata {
for output_info in sweeper_state.outputs.iter_mut() {
for output_info in sweeper_state.persistent.outputs.iter_mut() {
if output_info.is_spent_in(*tx) {
output_info.status.confirmed(confirmation_hash, height, (*tx).clone())
}
}
}

sweeper_state.dirty = true;
}

fn best_block_updated_internal(
&self, sweeper_state: &mut SweeperState, header: &Header, height: u32,
) {
sweeper_state.best_block = BestBlock::new(header.block_hash(), height);
sweeper_state.persistent.best_block = BestBlock::new(header.block_hash(), height);
self.prune_confirmed_outputs(sweeper_state);

sweeper_state.dirty = true;
}
}

Expand All @@ -666,17 +694,13 @@ where
&self, header: &Header, txdata: &chain::transaction::TransactionData, height: u32,
) {
let mut state_lock = self.sweeper_state.lock().unwrap();
assert_eq!(state_lock.best_block.block_hash, header.prev_blockhash,
assert_eq!(state_lock.persistent.best_block.block_hash, header.prev_blockhash,
"Blocks must be connected in chain-order - the connected header must build on the last connected header");
assert_eq!(state_lock.best_block.height, height - 1,
assert_eq!(state_lock.persistent.best_block.height, height - 1,
"Blocks must be connected in chain-order - the connected block height must be one greater than the previous height");

self.transactions_confirmed_internal(&mut *state_lock, header, txdata, height);
self.best_block_updated_internal(&mut *state_lock, header, height);

let _ = self.persist_state(&*state_lock).map_err(|e| {
log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
});
self.transactions_confirmed_internal(&mut state_lock, header, txdata, height);
self.best_block_updated_internal(&mut state_lock, header, height);
}

fn block_disconnected(&self, header: &Header, height: u32) {
Expand All @@ -685,22 +709,20 @@ where
let new_height = height - 1;
let block_hash = header.block_hash();

assert_eq!(state_lock.best_block.block_hash, block_hash,
assert_eq!(state_lock.persistent.best_block.block_hash, block_hash,
"Blocks must be disconnected in chain-order - the disconnected header must be the last connected header");
assert_eq!(state_lock.best_block.height, height,
assert_eq!(state_lock.persistent.best_block.height, height,
"Blocks must be disconnected in chain-order - the disconnected block must have the correct height");
state_lock.best_block = BestBlock::new(header.prev_blockhash, new_height);
state_lock.persistent.best_block = BestBlock::new(header.prev_blockhash, new_height);

for output_info in state_lock.outputs.iter_mut() {
for output_info in state_lock.persistent.outputs.iter_mut() {
if output_info.status.confirmation_hash() == Some(block_hash) {
debug_assert_eq!(output_info.status.confirmation_height(), Some(height));
output_info.status.unconfirmed();
}
}

self.persist_state(&*state_lock).unwrap_or_else(|e| {
log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
});
state_lock.dirty = true;
}
}

Expand All @@ -720,16 +742,14 @@ where
) {
let mut state_lock = self.sweeper_state.lock().unwrap();
self.transactions_confirmed_internal(&mut *state_lock, header, txdata, height);
self.persist_state(&*state_lock).unwrap_or_else(|e| {
log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
});
}

fn transaction_unconfirmed(&self, txid: &Txid) {
let mut state_lock = self.sweeper_state.lock().unwrap();

// Get what height was unconfirmed.
let unconf_height = state_lock
.persistent
.outputs
.iter()
.find(|o| o.status.latest_spending_tx().map(|tx| tx.compute_txid()) == Some(*txid))
Expand All @@ -738,28 +758,25 @@ where
if let Some(unconf_height) = unconf_height {
// Unconfirm all >= this height.
state_lock
.persistent
.outputs
.iter_mut()
.filter(|o| o.status.confirmation_height() >= Some(unconf_height))
.for_each(|o| o.status.unconfirmed());

self.persist_state(&*state_lock).unwrap_or_else(|e| {
log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
});
state_lock.dirty = true;
}
}

fn best_block_updated(&self, header: &Header, height: u32) {
let mut state_lock = self.sweeper_state.lock().unwrap();
self.best_block_updated_internal(&mut *state_lock, header, height);
let _ = self.persist_state(&*state_lock).map_err(|e| {
log_error!(self.logger, "Error persisting OutputSweeper: {:?}", e);
});
self.best_block_updated_internal(&mut state_lock, header, height);
}

fn get_relevant_txids(&self) -> Vec<(Txid, u32, Option<BlockHash>)> {
let state_lock = self.sweeper_state.lock().unwrap();
state_lock
.persistent
.outputs
.iter()
.filter_map(|o| match o.status {
Expand All @@ -779,13 +796,19 @@ where
}
}

#[derive(Debug, Clone)]
#[derive(Debug)]
struct SweeperState {
persistent: PersistentSweeperState,
dirty: bool,
}

#[derive(Debug, Clone)]
struct PersistentSweeperState {
outputs: Vec<TrackedSpendableOutput>,
best_block: BestBlock,
}

impl_writeable_tlv_based!(SweeperState, {
impl_writeable_tlv_based!(PersistentSweeperState, {
(0, outputs, required_vec),
(2, best_block, required),
});
Expand Down Expand Up @@ -831,7 +854,7 @@ where
kv_store,
logger,
) = args;
let state = SweeperState::read(reader)?;
let state = PersistentSweeperState::read(reader)?;
let best_block = state.best_block;

if let Some(filter) = chain_data_source.as_ref() {
Expand All @@ -841,7 +864,7 @@ where
}
}

let sweeper_state = Mutex::new(state);
let sweeper_state = Mutex::new(SweeperState { persistent: state, dirty: false });
Ok(Self {
sweeper_state,
pending_sweep: AtomicBool::new(false),
Expand Down Expand Up @@ -880,7 +903,7 @@ where
kv_store,
logger,
) = args;
let state = SweeperState::read(reader)?;
let state = PersistentSweeperState::read(reader)?;
let best_block = state.best_block;

if let Some(filter) = chain_data_source.as_ref() {
Expand All @@ -890,7 +913,7 @@ where
}
}

let sweeper_state = Mutex::new(state);
let sweeper_state = Mutex::new(SweeperState { persistent: state, dirty: false });
Ok((
best_block,
OutputSweeper {
Expand Down
Loading