dbt-labs/dbt-core · error

event should exist after peek

Error message

event should exist after peek

What it means

This is the `expect("event should exist after peek")` in `get_result_strict`. The method first confirms via `peek_next(node_id)` that a recorded event exists, then consumes it with `take_next(node_id)`. The expect asserts that peek-then-take is atomic — if it fires, the queue mutated between the two calls. This is a single-threaded internal invariant, so a panic here means a concurrency bug (shared &mut across threads) or a logic change in the Recording type, not bad user input.

Source

Thrown at crates/dbt-adapter/src/time_machine/engine.rs:285

        _args: &serde_json::Value,
        call_category: SemanticCategory,
    ) -> Result<&AdapterCallEvent, ReplayCallError> {
        if self.recording.peek_next(node_id).is_none() {
            return Err(ReplayCallError {
                message: format!(
                    "No recorded event for {} call on node '{}'. \
                     Recording may be incomplete or from a different code version.",
                    call_category, node_id
                ),
                recorded_error: None,
            });
        }

        // Consume and return the matching event for replay
        Ok(self
            .recording
            .take_next(node_id)
            .expect("event should exist after peek"))
    }

    /// Writes must match the next write barrier in sequence; reads can match any read in
    /// the current segment with matching args (and the same recorded read can satisfy
    /// multiple calls).
    fn get_result_semantic(
        &self,
        node_id: &str,
        method: &str,
        args: &serde_json::Value,
        call_category: SemanticCategory,
    ) -> Result<&AdapterCallEvent, ReplayCallError> {
        // Use semantic matching from the Recording
        self.recording
            .take_semantic_match(node_id, method, args, call_category)
            .ok_or_else(|| {
                let context = if call_category.is_mutating() {
                    "Write operations must match the next write barrier in sequence."

View on GitHub (pinned to 0267ce9170)

Solutions

1. Ensure the replay engine is used from a single thread per recording, or hold the lock across both peek and take (do the peek and take in one locked critical section).
2. Change get_result_strict to use a single atomic pop-and-check: match on take_next directly and return the existing ReplayCallError if None, removing the double-check pattern.
3. If the panic is intermittent, look for recently added concurrency in test harnesses (rayon, threads) around node execution and serialize access to the recording.
4. Verify no custom/overridden Recording implementation violates the peek/take contract.

Example fix

// before
if self.recording.peek_next(node_id).is_none() { return Err(...); }
Ok(self.recording.take_next(node_id).expect("event should exist after peek"))
// after
match self.recording.take_next(node_id) {
    Some(event) => Ok(event),
    None => Err(ReplayCallError { message: format!(
        "No recorded event for {} call on node '{}'.", call_category, node_id),
        recorded_error: None }),
}

// Before replaying, verify single-threaded access to each node's recording:
let seen = std::sync::atomic::AtomicBool::new(false);
if seen.swap(true, std::sync::atomic::Ordering::SeqCst) {
    panic!("node '{}' is being replayed concurrently", node_id);
}

fn can_replay(recording: &Recording, node_id: &str) -> bool {
    recording.peek_next(node_id).is_some()
        // plus architectural guarantee: no other thread holds this recording
}

let result = std::panic::catch_unwind(AssertUnwindSafe(||
    engine.get_result(node_id, method, &args)));
match result {
    Ok(ev) => ev,
    Err(_) => anyhow::bail!(
        "replay state corrupted for node '{}' — rerun single-threaded or regenerate recording",
        node_id),
}

When it happens

Trigger: Replaying an adapter recording when the event queue for a node_id is drained concurrently between `peek_next` and `take_next` — e.g. multiple threads calling the replay engine for the same node simultaneously, or a custom Recording implementation whose take_next can pop without a matching peek (such as a non-blocking queue returning None under contention).

Common situations: Running dbt replay tests with parallelized nodes sharing a recording; someone refactors Recording::take_next to evict entries or the recording is Arc<Mutex>-wrapped but a lock is released between peek and take; replaying a recording that another process is also consuming.

Understand the failure class

Background: "This is a bug, please report it": internal invariant violations, unreachable panics, and SNH errors explained — this error's family across 47 libraries.

Related errors

• inconsistent park state; actual = {actual}
• inconsistent park_timeout state; actual = {actual}
• inconsistent park_timeout state: {n}
• inconsistent state in unpark
• Schema not found for canonical FQN: {}

AI-assisted analysis of dbt-labs/dbt-core@0267ce9170 (2026-09-07). Data as JSON: /api/errors/11ca119082775aaa. Report an issue: GitHub.