risingwavelabs/risingwave · error · anyhow::Error

log_store_rewind_start_epoch {} not later than first_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.

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)

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

• initial epoch {} greater than aligned initial epoch {}
• failed to start {:?} for handle {}
• Invalid time: {value} {unit} is out of range for a time of d
• Invalid jsonb encoding
• Invalid variant encoding

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