Method step

crates/batcher/encoder/src/encoder.rs:551–689

(&mut self)

    }

    fn step(&mut self) -> Result<StepResult, StepError> {
        if let Some(error) = self.deferred_step_error.take() {
            return Err(error);
        }

        // Check for channel timeout first.
        if self.check_channel_timeout()? {
            return Ok(StepResult::ChannelClosed);
        }

        // If there are no blocks to encode, we're idle.
        if self.block_cursor >= self.blocks.len() {
            if self.config.batch_type == BatchType::Span
                && let Some(max_blocks_per_span_batch) = self.config.max_blocks_per_span_batch
                && self.span_accumulator.len() >= max_blocks_per_span_batch
            {
                if !self.try_flush_span_or_close("size_full")? {
                    return Ok(StepResult::ChannelClosed);
                }
                return Ok(StepResult::SpanFlushed);
            }

            return Ok(StepResult::Idle);
        }

        // Get the block at the cursor.
        let block = &self.blocks[self.block_cursor];
        let block_da_backlog_bytes = Self::block_da_backlog_bytes(block);

        // Convert block to a SingleBatch. Failure here is fatal: skipping the block
        // would produce a gap in the L2 block sequence submitted to L1.
        let (single_batch, l1_info) = BatchComposer::block_to_single_batch(block)
            .map_err(|source| StepError::CompositionFailed { cursor: self.block_cursor, source })?;

        match self.config.batch_type {
            BatchType::Span => {
                // In Span mode blocks are accumulated in memory; the span batch is
                // written to the channel only when a span-batch or channel boundary is reached.
                let seq_num = l1_info.sequence_number();
                // Maintain a running byte counter so the size check below is O(1) per
                // step rather than O(N·M) over the entire accumulator.
                let block_raw_bytes = Self::SPAN_BATCH_PER_BLOCK_OVERHEAD
                    + single_batch.transactions.iter().map(|tx| tx.len()).sum::<usize>();
                self.span_raw_bytes += block_raw_bytes;
                self.span_da_backlog_bytes += block_da_backlog_bytes;
                self.span_accumulator.push((single_batch, seq_num));
                self.block_cursor += 1;

                // Track the L1 head at which the first block of this span was accumulated.
                // `check_channel_timeout()` uses this to detect when the span has been open
                // too long even though `current_channel` is None between flushes.
                if self.span_opened_at_l1.is_none() {
                    self.span_opened_at_l1 = Some(self.l1_head);
                }

                // Estimate the compressed size of the accumulated span batch and close
                // the channel when it would exceed the configured size budget. This mirrors
                // the reference batcher's `SpanChannelOut`, which triggers closure based on estimated