# BLK-008: Write Pipeline (`put_pipelined`)

## Summary

A high-throughput write pipeline that accepts blocks via a bounded `mpsc` channel, batches them, and issues a single RocksDB `WriteBatch` per batch. This amortizes write overhead during bulk ingestion (e.g., initial sync) and should yield 5-10x throughput improvement over individual `put` calls.

## Specification

```rust
/// Configuration for the write pipeline.
pub struct WritePipelineConfig {
    /// Maximum blocks per WriteBatch (default: 64).
    pub batch_size: usize,
    /// Maximum time to wait before flushing an incomplete batch (default: 100ms).
    pub flush_ms: u64,
    /// Bounded channel capacity.
    pub channel_capacity: usize,
}

impl Default for WritePipelineConfig {
    fn default() -> Self {
        Self {
            batch_size: 64,
            flush_ms: 100,
            channel_capacity: 256,
        }
    }
}

impl BlockStore {
    /// Submit a block to the write pipeline. Blocks until channel has capacity.
    pub fn put_pipelined(&self, block: L2Block, canonical: bool) -> Result<()> {
        self.pipeline_tx
            .send((block, canonical))
            .map_err(|_| Error::PipelineClosed)?;
        Ok(())
    }

    /// Background task that drains the pipeline channel and writes batches.
    async fn run_write_pipeline(
        store: Arc<BlockStoreInner>,
        mut rx: mpsc::Receiver<(L2Block, bool)>,
        config: WritePipelineConfig,
    ) {
        let mut batch_buf: Vec<(L2Block, bool)> = Vec::with_capacity(config.batch_size);
        let flush_interval = Duration::from_millis(config.flush_ms);

        loop {
            let deadline = tokio::time::sleep(flush_interval);
            tokio::pin!(deadline);

            // Collect up to batch_size items or until flush timeout
            loop {
                tokio::select! {
                    item = rx.recv() => {
                        match item {
                            Some(entry) => {
                                batch_buf.push(entry);
                                if batch_buf.len() >= config.batch_size {
                                    break;
                                }
                            }
                            None => {
                                // Channel closed; flush remaining and exit
                                if !batch_buf.is_empty() {
                                    store.flush_batch(&mut batch_buf).unwrap();
                                }
                                return;
                            }
                        }
                    }
                    _ = &mut deadline => {
                        break; // Flush timeout
                    }
                }
            }

            if !batch_buf.is_empty() {
                store.flush_batch(&mut batch_buf).unwrap();
            }
        }
    }

    /// Flush a collected batch as a single RocksDB WriteBatch.
    fn flush_batch(&self, batch_buf: &mut Vec<(L2Block, bool)>) -> Result<()> {
        let mut wb = WriteBatch::default();

        for (block, canonical) in batch_buf.drain(..) {
            let hash = block.header.hash();

            // Idempotency: skip if already exists
            if self.db.get_cf(&self.cf_blocks, hash.as_ref())?.is_some() {
                continue;
            }

            let block_bytes = bincode::serialize(&block)?;
            let compressed = zstd::encode_all(&block_bytes[..], self.compression_level)?;
            wb.put_cf(&self.cf_blocks, hash.as_ref(), &compressed);

            let header_bytes = bincode::serialize(&block.header)?;
            wb.put_cf(&self.cf_headers, hash.as_ref(), &header_bytes);

            let record = BlockRecord::from_header(&block.header);
            self.record_cache.insert(hash, record);

            if canonical {
                let height_key = block.header.height.to_be_bytes();
                wb.put_cf(&self.cf_canonical, &height_key, hash.as_ref());
            }
        }

        self.db.write(wb)?;
        Ok(())
    }
}
```

## Acceptance Criteria

1. `put_pipelined` accepts a block into a bounded `mpsc` channel without directly writing to RocksDB.
2. The background pipeline task batches up to `write_pipeline_batch_size` blocks (default: 64) before flushing.
3. An incomplete batch is flushed after `write_pipeline_flush_ms` (default: 100 ms).
4. Each flush issues a single RocksDB `WriteBatch` containing all blocks in the batch.
5. Idempotency is maintained: duplicate block hashes within or across batches do not cause overwrites.
6. The record cache is populated for each block in the batch.
7. Canonical mappings are written for blocks flagged as canonical.
8. When the channel is closed, any remaining buffered blocks are flushed before the pipeline task exits.

## Implementation Notes

- The bounded channel provides natural backpressure: producers block when the channel is full.
- `WriteBatch` is atomic in RocksDB, so either all entries in a batch are written or none are.
- The idempotency check inside `flush_batch` uses a point read per block; for very large batches, consider maintaining a `HashSet` of hashes seen in the current batch.
- Error handling in the background task should log failures and optionally notify producers.

## Test Plan

- **Integration: bulk store** -- Send 256 blocks via `put_pipelined`, verify all are retrievable via `get_block`.
- **Integration: batch size** -- Send exactly `batch_size` blocks, verify they are flushed in a single `WriteBatch` (instrument or count write operations).
- **Integration: flush timeout** -- Send fewer than `batch_size` blocks, wait >100ms, verify they are flushed.
- **Integration: idempotency** -- Send the same block twice via pipeline, verify it is stored once.
- **Benchmark: throughput** -- Compare throughput of `put_pipelined` vs serial `put` for 10,000 blocks; target 5-10x improvement.
- **Unit: backpressure** -- Fill the channel, verify `put_pipelined` blocks (does not drop blocks).
- **Unit: graceful shutdown** -- Drop the sender, verify remaining buffered blocks are flushed.

## Expected Test Files

- `tests/blk_008_tests.rs`
