Celestia Backpressure and Heuristics

[l-monninger]

Summary

• We ought to try a number of strategies for dealing with Celestia backpressure.
• These strategies have a general grouping heuristic.
• Adding an API to cover this heuristic will make it easier to test.
• Here's a starter API

Celestia

Celestia has a finality and block time of 15 seconds. It also has a maximum blob size and a maximum transaction size which will be thrown as different errors, but may have the same upper bound (this needs clarification). Hence you submit multiple blobs in a single API call.

To better handle this backpressure from Celestia, we need to maximally batch block submissions. That is, we thus need to compute (or test with API call rejection) the grouping of blocks that will produce the maximum acceptable Celestia transaction.

Approach

1. If we can indeed rely on information from before the API call to guarantee block submission, e.g., block size in bytes, we can simply compute the groups with a Multiple Knapsack Problem algorithm.
2. However, we likely also want to handle retries dynamically for one tentative and two definite reasons:
   1. Tentatively, there may be information relating to failure and backpressure that would be difficult to ascertain without running the exact requests.
   2. Definitely, we will need to make sure the weight of the Knapsack is dynamic, because the max blob size is variable is determined by governance.
   3. We definitely want to retry in case of general network availability or synchronization issues.

Provisional API

I've created a grouping_heuristic API, which can be most easy reviewed here: #277

The main utility of this API is to expose a method which runs a loop-until-done generalization that can be run over requests:

 /// Runs the grouping heuristic asynchronously, but in a sequential manner.
    pub async fn run_async_sequential<F, Fut>(
        &mut self, 
        mut distribution: Vec<GroupingOutcome<T>>,
        func: F
    ) -> Result<Vec<GroupingOutcome<T>>, anyhow::Error>
    where
        F: Fn(GroupingOutcome<T>) -> Fut + Send + Sync,
        Fut: std::future::Future<Output = Result<GroupingOutcome<T>, anyhow::Error>> + Send,
    {
        loop {
            // distribute
            distribution = self.distribute(distribution)?;
    
            // run the function asynchronously
            let mut new_distribution = Vec::new();
            for outcome in distribution {
                new_distribution.push(func(outcome).await?);
            }
    
            // check if we're done
            if new_distribution.iter().all(|outcome| outcome.all_done()) {
                return Ok(new_distribution);
            }
    
            // update the distribution
            distribution = new_distribution;
        }
    }

Create a GroupingHeuristic

pub struct Chunking {
    pub size: usize
}

impl Chunking {
    pub fn new(size: usize) -> Self {
        Self { size }
    }
}

impl <T> GroupingHeuristic<T> for Chunking {
    
    fn distribute(&mut self, distribution: Vec<GroupingOutcome<T>>) -> Result<Vec<GroupingOutcome<T>>, anyhow::Error> {

        // flatten the distribution
        let mut distribution = distribution.into_iter().flat_map(|outcome| outcome.into_inner()).collect::<Vec<_>>();

        // chunk the distribution
        let mut chunks = Vec::new();
        while !distribution.is_empty() {
            let chunk = distribution.drain(0..self.size.min(distribution.len())).collect();
            chunks.push(GroupingOutcome::new(chunk));
        }

        Ok(chunks)
        
    }

}

You can see this in action here:

pub async fn submit_heuristic(&self, blocks : Vec<Block>) -> Result<(), anyhow::Error> {

		let mut heuristic : GroupingHeuristicStack<Block> = GroupingHeuristicStack::new(vec![
			Box::new(DropSuccess),
			Box::new(ToApply),
			Box::new(Splitting::new(2)),
			Box::new(LinearlyDecreasingChunking::new(4, 1))
		]);

		let _failed_blocks = heuristic.run_async_sequential(
			GroupingOutcome::new_apply_distribution(blocks),
			|grouping| async move {
				let blocks = grouping.into_original();
				match self.submit_blocks(&blocks).await {
					Ok(_) => Ok(GroupingOutcome::new_all_success(blocks.len())),
					Err(_) => Ok(GroupingOutcome::new_apply(blocks)),
				}
			},
		);

		Ok(())

	}

[l-monninger]

So, I'd like to play with this a bit. For me the OOO I was thinking would be:

1. Check in with Celestia folks on the transactions size vs. blob size distinction.
2. Implement the Knapsacking for blob size.
3. Play around with Heuristics.

We may also want to try compress our Celestia blobs.