{"record":{"id":"76ad9ba7b74afaac","repo":"quickwit-oss/quickwit","slug":"trying-to-replace-in-progress-message","errorCode":null,"errorMessage":"trying to replace in progress message","messagePattern":"trying to replace in progress message","errorType":"exception","errorClass":null,"httpStatus":null,"severity":"error","filePath":"quickwit/quickwit-indexing/src/source/queue_sources/local_state.rs","lineNumber":114,"sourceCode":"                in_progress.partition_id.clone(),\n                in_progress.visibility_handle.ack_id().to_string(),\n            );\n            in_progress\n                .visibility_handle\n                .request_last_extension()\n                .await?;\n        }\n        Ok(())\n    }\n\n    /// Tries to set the message that is currently being read. Returns an error\n    /// if there is already a message being read.\n    pub fn set_currently_read(\n        &mut self,\n        in_progress: Option<InProgressMessage>,\n    ) -> anyhow::Result<()> {\n        if self.read_in_progress.is_some() {\n            bail!(\"trying to replace in progress message\");\n        }\n        self.read_in_progress = in_progress;\n        Ok(())\n    }\n\n    /// Returns the ack_id if that message was awaiting_commit\n    pub fn mark_completed(&mut self, partition_id: PartitionId) -> Option<String> {\n        let ack_id_opt = self.awaiting_commit.remove(&partition_id);\n        self.completed.insert(partition_id);\n        ack_id_opt\n    }\n}\n","sourceCodeStart":96,"sourceCodeEnd":127,"githubUrl":"https://github.com/quickwit-oss/quickwit/blob/a39730c5cdcd1a4fe798403737ae293999ea21f8/quickwit/quickwit-indexing/src/source/queue_sources/local_state.rs#L96-L127","documentation":"QueueSourceState tracks at most one InProgressMessage at a time. set_currently_read refuses to overwrite an existing in-progress message with a new one, since doing so would lose the pending message's ack state. It only accepts a replacement when the current slot is empty (the argument may be None to clear/commit the slot).","triggerScenarios":"Calling set_currently_read(Some(msg)) while self.read_in_progress is already Some(...) — i.e., starting to read a new queue message before the previous in-progress message was acknowledged/cleared.","commonSituations":"Source actor logic bugs where try_read/acknowledge ordering is wrong; a message loop that calls set_currently_read twice without emitting/acknowledging the first message; double-polling a queue source.","solutions":["Fix the source loop to clear the in-progress slot (ack/commit or set_currently_read(None)) before reading the next message.","Review the sequencing in the queue source actor so exactly one message is in flight at a time.","If reproducible, file a bug with logs — this indicates an internal protocol violation in the source pipeline."],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":"// Only call set_currently_read when the slot is free\nif state.currently_in_progress().is_some() {\n    return Err(anyhow::anyhow!(\"previous in-progress message must be acked/cleared first\"));\n}","typeGuard":null,"tryCatchPattern":"match state.set_currently_read(Some(msg)) {\n    Ok(()) => { /* proceed */ }\n    Err(e) => { /* source loop bug: clear slot via ack/None before polling again */ log::error!(\"in-progress slot occupied: {e}\"); }\n}","preventionTips":["Always acknowledge or clear the in-progress message before polling the next one.","Keep one-message-at-a-time discipline in queue source actor loops.","Add debug assertions/logs around the read/ack lifecycle during development."],"tags":["state-machine","queue-source","internal-invariant","rust"],"backgroundTag":"invalid-state-transition","analyzedSha":"a39730c5cdcd1a4fe798403737ae293999ea21f8","analyzedAt":"2026-09-08T13:19:37.784Z","contentChangedAt":"2026-09-08T13:19:37.784Z","schemaVersion":2},"datasetVersion":"2026-09-14T11:17:12.474Z"}