quickwit-oss/quickwit · error

trying to replace in progress message

Error message

trying to replace in progress message

What it means

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).

Source

Thrown at quickwit/quickwit-indexing/src/source/queue_sources/local_state.rs:114

                in_progress.partition_id.clone(),
                in_progress.visibility_handle.ack_id().to_string(),
            );
            in_progress
                .visibility_handle
                .request_last_extension()
                .await?;
        }
        Ok(())
    }

    /// Tries to set the message that is currently being read. Returns an error
    /// if there is already a message being read.
    pub fn set_currently_read(
        &mut self,
        in_progress: Option<InProgressMessage>,
    ) -> anyhow::Result<()> {
        if self.read_in_progress.is_some() {
            bail!("trying to replace in progress message");
        }
        self.read_in_progress = in_progress;
        Ok(())
    }

    /// Returns the ack_id if that message was awaiting_commit
    pub fn mark_completed(&mut self, partition_id: PartitionId) -> Option<String> {
        let ack_id_opt = self.awaiting_commit.remove(&partition_id);
        self.completed.insert(partition_id);
        ack_id_opt
    }
}

View on GitHub (pinned to a39730c5cd)

Solutions

1. Fix the source loop to clear the in-progress slot (ack/commit or set_currently_read(None)) before reading the next message.
2. Review the sequencing in the queue source actor so exactly one message is in flight at a time.
3. If reproducible, file a bug with logs — this indicates an internal protocol violation in the source pipeline.

// Only call set_currently_read when the slot is free
if state.currently_in_progress().is_some() {
    return Err(anyhow::anyhow!("previous in-progress message must be acked/cleared first"));
}

match state.set_currently_read(Some(msg)) {
    Ok(()) => { /* proceed */ }
    Err(e) => { /* source loop bug: clear slot via ack/None before polling again */ log::error!("in-progress slot occupied: {e}"); }
}

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

• node not found in pending
• OTP logs or traces do not support VRL transforms
• position of a Kafka partition should never be EOF
• position of a Kinesis shard should never be EOF
• Partition is owned by this indexing pipeline but is not at t

AI-assisted analysis of quickwit-oss/quickwit@a39730c5cd (2026-09-08). Data as JSON: /api/errors/76ad9ba7b74afaac. Report an issue: GitHub.