What causes "cannot get truncated epoch" in risingwavelabs/risingwave?

Find the upstream failure that terminated the reader task; this error is a consequence.

cannot get truncated epoch

Error message

cannot get truncated epoch

What it means

After a checkpoint barrier, the writer waits for the reader to report the truncated epoch back over an unbounded channel. `recv()` returning None means every `truncated_epoch_tx` sender was dropped, i.e. the reader is gone, so the writer can never learn whether the epoch was consumed. The error therefore signals a dead reader at checkpoint time.

Source

Thrown at src/stream/src/common/log_store_impl/in_mem.rs:330

                next_epoch,
                options,
            })
            .instrument_await("in_mem_send_item_barrier")
            .await
            .map_err(|_| anyhow!("unable to send barrier"))?;

        let prev_epoch = self
            .curr_epoch
            .replace(next_epoch)
            .expect("should have epoch");

        if is_checkpoint {
            let truncated_epoch = self
                .truncated_epoch_rx
                .recv()
                .instrument_await("in_mem_recv_truncated_epoch")
                .await
                .ok_or_else(|| anyhow!("cannot get truncated epoch"))?;
            assert_eq!(truncated_epoch, prev_epoch);
        }

        Ok(LogWriterPostFlushCurrentEpoch::new(move || {
            async move { Ok(()) }.boxed()
        }))
    }

    fn pause(&mut self) -> LogStoreResult<()> {
        // no-op when decouple is not enabled
        Ok(())
    }

    fn resume(&mut self) -> LogStoreResult<()> {
        // no-op when decouple is not enabled
        Ok(())
    }
}

View on GitHub (pinned to 6469eb736d)

Solutions

Find the upstream failure that terminated the reader task; this error is a consequence.
Ensure the reader runs a consume loop that calls truncate after each checkpoint epoch so the ack is sent.
Treat as terminal: fail the writing actor and let recovery rebuild the log store pair.
In tests, keep the truncated_epoch_tx alive (keep the reader instance) until flush completes.

Example fix

// before: reader dropped before ack
let (mut writer, reader) = factory.create();
drop(reader);
writer.flush_current_epoch(epoch, opts).await?;

// after: reader alive and truncating
let mut reader = reader;
reader.truncate(offset)?;
writer.flush_current_epoch(epoch, opts).await?;

Defensive patterns

Strategy: try-catch

Try / catch

if let Err(e) = writer.flush_current_epoch(epoch, opts).await {
    if e.to_string().contains("cannot get truncated epoch") {
        // reader dropped before acking; fail actor for recovery
        fail_actor(e);
    }
    return Err(e);
}

Prevention

Reader must consume and truncate each checkpoint epoch to send its ack.
Never drop the reader before the writer finishes flushing the current epoch.
Add supervision/restart on the reader task to avoid silent drops.

When it happens

Trigger: Calling `flush_current_epoch` with a checkpoint (is_checkpoint=true) after the reader was dropped, so `truncated_epoch_rx.recv()` returns None instead of the expected prev_epoch.

Common situations: Reader actor crashed or was cancelled before it could ack the previous checkpoint epoch; job restart/scale-in terminating the consumer; test harness dropping the reader early.

Related errors

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