Refactor `apply_block`

[Mirko-von-Leipzig]

apply_block is a bit of a mess at the moment. It works (probably), but its a huge function, instrumentation is missing, and things happen both concurrently and then sequentially again.

As a short summary:

It performs some checks, spawns some tasks to store the raw block data and update the sqlite database, and then carries on with updating the forests inline. Since we consider the sqlite database as the master, that task waits for confirmation that other portions have succeeded before it commits the transaction.

There is further incidental complexity with synchronizing the in-memory structures with the sqlite database commit, while minimizing the time we hold the write locks. This is a sort of separate problem, and I'll have to dig in further to really understand how we can solve this more coherently.

WIP

This is still undercooked; and I need to do some additional thinking/digging but I want to write it down so long for comments and feedback.

Proposal

This is inspired by the deferred proving PR, where we now have multiple database writers. It would be great to follow the rust model of single writer only, this simplifies things ito understanding and consistency. Its also a concern e.g. we often have issues with block writes taking 10s+ due to bugs, or poor performance, missing index etc, and this would cause the proof writers to error out, causing a crash (?) most likely.

I think we can solve this, and the complexity of apply_block by slightly shifting what we consider the master. This can be done in several phases, improving ergonomics and reducing footguns as we go.

Core principle

Introduce a new chain_tip: watch::Receiver<BlockHeader> to the store state, which specifies the current chain tip, regardless of what the database contains.

apply_block changes slightly, it now spawns tasks for every update part, and then waits for them all to complete. If all complete successfully, then the chain tip is updated. This should simplify each task, and also the instrumentation of them since they can't be missed now.

On startup we'll need to find the latest complete block, i.e. check databases, block store, and ignore the latest block if its only partial (i.e. apply_block failed to complete). This is used to init the chain_tip.

Proofs & single writer

The idea is that apply_block should perform the database updates that are done by the proof co-ordinator. At present, this means deleting the proof inputs up until the proven tip.

We use a similar mechanism to indicate the latest proven chain tip. The proof co-ordinator updates the proven_tip: watch::Receiver<BlockNumber> whenever it moves, without deleting the proof inputs. apply_block reads the latest proven tip and performs the required database deletions as part of its database update.

Database readers

Database readers should use these chain tip watch channels as a "snapshot" to have a database transaction-like property with the state. This can be forced by updating open-bounded sql queries (i.e. block_to is always capped by the chain tip).

We can further make things more ergonomic by making state opaque, e.g.

struct ReadOnlyStore {
   ...
}

impl ReadOnlyStore {
    async fn begin_transaction(&self) -> ReadOnlyTransaction {...}
}

struct ReadOnlyTransaction {
    chain_tip: BlockHeader,
    proven_tip: BlockNumber,
    tx_sql: SqliteTransaction,
    forests: Arc<Forest>,
    ...
}

Unknowns

Something that is going to be a concern is in-memory state and pruning. Ideally we would enable concurrent reads and a writer, similar to how sqlite does the WAL file. I'm unsure how things currently stand, but conceptually we could change the in-memory structs to be reference counted by block. This would mean a reader can hold the block's data and once the last reader that snapshotted that block ends, the data is automatically cleaned up.

We could change the chain_tip watch channel, to include more than just the header. Readers then snapshot this, while the writer simply updates the snapshot pointer.

However I think unfortunately, the current implementation of things doesn't allow for this. I need to investigate further.

chain_tip: watch::Receiver<Arc<(BlockHeader, History)>>

/// Holds the last `N` in-memory historial data.
///
/// Each block is also arc'd so it can be shared and re-used.
#[derive(Clone)]
struct History {
   blocks: VecDequeue<Arc<Block>>,
}

impl History {
    fn push(&mut self, block: Arc<Block>) {
        self.blocks.push_back(block);

        if self.blocks.len() > N {
            self.blocks.pop_front();
        }
    }
}

fn apply_block(block: Block, mut tx_chain_tip: watch::Sender<..>) {
    // ....
    
    // Get the current chain tip and update it.
    let mut history = tx_chain_tip.subscribe().borrow().history.clone();
    
    let header = block.header();
    history.push(Arc::new(Block));

    // This doesn't impact existing reader snapshots, only new ones will see this value.
    tx_chain_tip.send((header, history));
}

Relates to #1840, #1539, #1541, #1592, #1544, #1537

[Mirko-von-Leipzig]

A more concrete pruning explanation. Its (imo) difficult to understand which data is available to readers - some is in-memory, some is in the database but is also pruned. Readers therefore need to be cognizant of what may or may not be swept out under their feet e.g. is in-memory aligned with database.. it depends on which order the locks were acquired, both in the reader and writer. This kinda sucks.

What I'd like to incorporate into my proposal is a deterministic way of defining what is and isn't available to readers, and a way in which the writer knows what is safe to prune.

Something like the following.

The writer maintains the current history in a watch channel as per the original description. Readers acquire this, and each block's data is shared as an Arc<BlockState> which defines all the data required for accessing that specific block's data safely -- and is guaranteed to not be pruned.

The writer keeps the history of this pruned to a certain length N. Previous reads will still hold older blocks until they complete. The writer wants to prune these, but must wait until the old readers are done. This is communicated via the Arc as follows.

/// aka `apply_block` state
struct Writer {
    /// The latest `N` blocks of state. Readers contain the receiver for this.
    chain_tip: watch::Sender<Arc<VecDequeue<Arc<BlockState>>>>,
    /// Blocks older than `N` ago, for each block committed, `chain_tip` loses one and it
    /// gets appended here, to be pruned when possible.
    to_prune: VecDequeue<Arc<BlockState>>,
}

async fn apply_block(writer: &mut Writer, block: Block) {
    // .. do block checks etc, apply etc

    // Update chain tip with new block, removing oldest one.
    let mut history = writer.chain_tip.subscribe().borrow().clone();
    history.push_back(Arc::new(BlockState));

    if history.len() > N {
        writer.to_prune.push_back(history.pop_front().unwrap());
    }

    // This doesn't impact existing reader snapshots, only new ones will see this value.
    tx_chain_tip.send((header, history));
}

/// Prunes old state if it has no more readers.
///
/// A simple implementation would call this after every apply_block.
/// Smarter scheduling is possible to try account for block periods etc.
async fn prune_history(writer: &mut Writer) {
    // Prune each block until we hit one that has a reader still.
    while let Some(block) = writer.to_prune.pop_front() {
        // This call succeeds only if this Arc is the only owner,
        // meaning it is safe to prune this block since there are no other readers.
        match block.try_unwrap() {
            // Err returns the original Arc, re-insert it and we try again later.
            Err(block) => {
                writer.to_prune.push_front(block);
                return;
            },
            Ok(inner_block) => todo!("Prune the block"),
        }
    }
}