Method requeue

crates/batcher/encoder/src/encoder.rs:853–888

(&mut self, id: SubmissionId)

    }

    fn requeue(&mut self, id: SubmissionId) {
        // Invariant: each `ReadyChannel` owns its own frame cursor. This
        // encoder keeps at most one `current_channel` open at a time; when it
        // closes (by size or timeout) it moves to `ready_channels` as the
        // newest entry. `pending_ref.channel_idx` therefore always points to
        // a specific, independent slot in `ready_channels`. Resetting the
        // cursor on that slot does not affect any other channel and FIFO
        // ordering across channels is preserved by construction.
        let Some(pending_ref) = self.pending.remove(&id) else {
            warn!(id = ?id, "requeue called for unknown submission id");
            return;
        };

        let chan_idx = pending_ref.channel_idx;
        if chan_idx >= self.ready_channels.len() {
            warn!(id = ?id, chan_idx = %chan_idx, "requeue: channel index out of bounds; submission lost");
            return;
        }

        let channel = &mut self.ready_channels[chan_idx];
        channel.pending_confirmations = channel.pending_confirmations.saturating_sub(1);
        // Rewind cursor to the first frame of the requeued submission so all frames
        // in the batch are retried together.
        if pending_ref.frame_start < channel.cursor {
            channel.cursor = pending_ref.frame_start;
        }
        // Frames are back in pending state; re-increment the gauge.
        BatcherMetrics::pending_frames().increment(pending_ref.frame_count as f64);

        debug!(
            id = ?id,
            frame_start = %pending_ref.frame_start,
            frame_count = %pending_ref.frame_count,
            "requeued submission frames back to pending"
        );
    }

    fn force_close_channel(&mut self) {
        debug!("force-closing current channel");