risingwavelabs/risingwave · error · anyhow::Error

log_store_rewind_start_epoch

Error message

log_store_rewind_start_epoch {} not later than first_epoch {}

What it means

After rewinding the sink's log store reader to log_store_rewind_start_epoch, the coordinator expects the first log item's epoch to be strictly later than the rewind start epoch. If the first item's epoch is >= the rewind start, the rewind semantics are violated and the log would be consumed from a point at or before the requested epoch, so it bails.

Solutions

1. Check the log store to confirm the rewind start epoch is strictly less than the first retained epoch.
2. Retrigger recovery with a corrected rewind start epoch (usually the last checkpoint epoch).
3. Inspect meta log store cleanup/retention to ensure it did not trim past the requested epoch.

if let Some(start) = log_store_rewind_start_epoch {
    if start >= first_epoch_in_log {
        // adjust rewind point to last checkpoint epoch before retrying
    }
}

match result {
    Err(e) if e.to_string().contains("log_store_rewind_start_epoch") => retry_with_last_checkpoint_epoch(),
    other => other,
}

When it happens

Trigger: consume_log_and_sink is called with a rewind start epoch, and the log reader's first returned item carries first_epoch such that log_store_rewind_start_epoch >= first_epoch (rewind point not strictly before the first available item).

Common situations: Recovery after failover where the requested rewind epoch equals the first epoch still present in the log store; stale log store state after migration or incorrect epoch bookkeeping in the meta log store.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

• failed to start for handle
• initial epoch greater than aligned initial epoch
• a stream has reached the end but some other stream has not…
• cannot find StreamActor of actor
• corrupt MATCH_RECOGNIZE state row: seq column must be a…

AI-assisted analysis of risingwavelabs/risingwave@6469eb736d (2026-09-11). Data as JSON: /api/errors/5d5077780d30338c. Report an issue: GitHub.

Appendix: source

Thrown at src/connector/src/sink/coordinate.rs:87

        })
    }
}

#[async_trait]
impl<W: SinkWriter<CommitMetadata = Option<SinkMetadata>>> LogSinker for CoordinatedLogSinker<W> {
    async fn consume_log_and_sink(self, mut log_reader: impl SinkLogReader) -> Result<!> {
        let (mut coordinator_stream_handle, log_store_rewind_start_epoch) = self
            .sink_coordinate_client
            .new_stream_handle(&self.param, self.vnode_bitmap)
            .await?;
        let mut sink_writer = self.writer;
        log_reader.start_from(log_store_rewind_start_epoch).await?;
        let mut first_item = log_reader.next_item().await?;
        if let (Some(log_store_rewind_start_epoch), (first_epoch, _)) =
            (log_store_rewind_start_epoch, &first_item)
        {
            if log_store_rewind_start_epoch >= *first_epoch {
                bail!(
                    "log_store_rewind_start_epoch {} not later than first_epoch {}",
                    log_store_rewind_start_epoch,
                    first_epoch
                );
            }
        } else {
            let &(initial_epoch, _) = &first_item;
            let aligned_initial_epoch = coordinator_stream_handle
                .align_initial_epoch(initial_epoch)
                .await?;
            if initial_epoch != aligned_initial_epoch {
                warn!(
                    initial_epoch,
                    aligned_initial_epoch,
                    sink_id = %self.param.sink_id,
                    "initial epoch not matched aligned initial epoch"
                );
                let mut peeked_first = Some(first_item);

View on GitHub (pinned to 6469eb736d)