{"record":{"id":"11ca119082775aaa","repo":"dbt-labs/dbt-core","slug":"event-should-exist-after-peek","errorCode":null,"errorMessage":"event should exist after peek","messagePattern":"event should exist after peek","errorType":"panic","errorClass":null,"httpStatus":null,"severity":"error","filePath":"crates/dbt-adapter/src/time_machine/engine.rs","lineNumber":285,"sourceCode":"        _args: &serde_json::Value,\n        call_category: SemanticCategory,\n    ) -> Result<&AdapterCallEvent, ReplayCallError> {\n        if self.recording.peek_next(node_id).is_none() {\n            return Err(ReplayCallError {\n                message: format!(\n                    \"No recorded event for {} call on node '{}'. \\\n                     Recording may be incomplete or from a different code version.\",\n                    call_category, node_id\n                ),\n                recorded_error: None,\n            });\n        }\n\n        // Consume and return the matching event for replay\n        Ok(self\n            .recording\n            .take_next(node_id)\n            .expect(\"event should exist after peek\"))\n    }\n\n    /// Writes must match the next write barrier in sequence; reads can match any read in\n    /// the current segment with matching args (and the same recorded read can satisfy\n    /// multiple calls).\n    fn get_result_semantic(\n        &self,\n        node_id: &str,\n        method: &str,\n        args: &serde_json::Value,\n        call_category: SemanticCategory,\n    ) -> Result<&AdapterCallEvent, ReplayCallError> {\n        // Use semantic matching from the Recording\n        self.recording\n            .take_semantic_match(node_id, method, args, call_category)\n            .ok_or_else(|| {\n                let context = if call_category.is_mutating() {\n                    \"Write operations must match the next write barrier in sequence.\"","sourceCodeStart":267,"sourceCodeEnd":303,"githubUrl":"https://github.com/dbt-labs/dbt-core/blob/0267ce9170576975b76b64ce856b2e5848e96617/crates/dbt-adapter/src/time_machine/engine.rs#L267-L303","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["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).","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.","If the panic is intermittent, look for recently added concurrency in test harnesses (rayon, threads) around node execution and serialize access to the recording.","Verify no custom/overridden Recording implementation violates the peek/take contract."],"exampleFix":"// before\nif self.recording.peek_next(node_id).is_none() { return Err(...); }\nOk(self.recording.take_next(node_id).expect(\"event should exist after peek\"))\n// after\nmatch self.recording.take_next(node_id) {\n    Some(event) => Ok(event),\n    None => Err(ReplayCallError { message: format!(\n        \"No recorded event for {} call on node '{}'.\", call_category, node_id),\n        recorded_error: None }),\n}","handlingStrategy":"try-catch","validationCode":"// Before replaying, verify single-threaded access to each node's recording:\nlet seen = std::sync::atomic::AtomicBool::new(false);\nif seen.swap(true, std::sync::atomic::Ordering::SeqCst) {\n    panic!(\"node '{}' is being replayed concurrently\", node_id);\n}","typeGuard":"fn can_replay(recording: &Recording, node_id: &str) -> bool {\n    recording.peek_next(node_id).is_some()\n        // plus architectural guarantee: no other thread holds this recording\n}","tryCatchPattern":"let result = std::panic::catch_unwind(AssertUnwindSafe(||\n    engine.get_result(node_id, method, &args)));\nmatch result {\n    Ok(ev) => ev,\n    Err(_) => anyhow::bail!(\n        \"replay state corrupted for node '{}' — rerun single-threaded or regenerate recording\",\n        node_id),\n}","preventionTips":["Never share one Recording across threads performing replay for the same node_id.","Keep peek and take in the same critical section if Recording is behind a Mutex.","Treat 'No recorded event' errors and this panic as signals the recording is stale or mismatched; regenerate recordings after code changes.","Review any new parallelism (rayon/threads) added around node execution in replay tests."],"tags":["rust","panic","concurrency","replay","time-machine"],"backgroundTag":"internal-invariant-violation","analyzedSha":"0267ce9170576975b76b64ce856b2e5848e96617","analyzedAt":"2026-09-07T21:53:39.732Z","contentChangedAt":"2026-09-07T21:53:39.732Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}