zylon-ai/private-gpt · error · ValueError

Stream with correlation_id {correlation_id} already exists

Error message

Stream with correlation_id {correlation_id} already exists

What it means

Raised by RedisStreamService.create_stream when the Lua creation script reports the status hash already exists. Creation uses HSET NX-style semantics (the script returns 0 when the key exists), so creating the same correlation_id twice is rejected atomically across all clients talking to the same Redis DB. This is the distributed equivalent of the in-memory duplicate-stream guard.

Source

Thrown at private_gpt/components/streaming/providers/redis_stream_service.py:118

        created = await cast(Any, self._client.eval)(
            """
            if redis.call('exists', KEYS[1]) == 1 then
                return 0
            end
            local fields = cjson.decode(ARGV[1])
            for key, value in pairs(fields) do
                redis.call('hset', KEYS[1], key, value)
            end
            redis.call('expire', KEYS[1], ARGV[2])
            return 1
            """,
            1,
            status_key,
            json.dumps(mapping),
            self._config.expiry_seconds,
        )
        if not created:
            raise ValueError(
                f"Stream with correlation_id {correlation_id} already exists"
            )

        return correlation_id

    async def update_stream_status(
        self,
        correlation_id: str,
        status: StreamStatus,
        error_message: str | None = None,
        metadata: dict[str, Any] | None = None,
    ) -> None:
        """Update stream status and metadata."""
        status_key = self._get_status_key(correlation_id)

        now = datetime.now(UTC)
        updates = {
            "status": status.value,

View on GitHub (pinned to 4a030776a3)

Solutions

1. Check await stream_service.stream_exists(correlation_id) first and reuse or delete the existing stream
2. Generate a fresh uuid correlation id per attempt instead of a stable business id
3. If the stale stream is garbage, delete it (or wait for expiry_seconds) before re-creating

Example fix

# before
await stream_service.create_stream(correlation_id=job_id, stream_type="ingestion")
# after
if await stream_service.stream_exists(job_id):
    await stream_service.delete_stream(job_id)
await stream_service.create_stream(correlation_id=job_id, stream_type="ingestion")

if await stream_service.stream_exists(correlation_id):
    await stream_service.delete_stream(correlation_id)
await stream_service.create_stream(correlation_id=correlation_id, stream_type=stream_type)

try:
    await stream_service.create_stream(correlation_id=cid, stream_type=st)
except ValueError as e:
    if "already exists" not in str(e):
        raise
    logger.info("stream %s exists in Redis; reusing", cid)

When it happens

Trigger: Calling create_stream for a correlation_id whose status hash still exists in Redis (created earlier and not yet expired via expiry_seconds). Typical with job retries, at-least-once queues, or rerunning a job with a fixed id.

Common situations: Queue redelivery after a worker timeout where the first attempt already created the stream; short expiry_seconds hiding state in tests but long expiry in prod exposing duplicates; manually rerunning a failed ingestion with the same correlation id.

Related errors

• Stream with correlation_id {correlation_id} already exists
• Redis streaming dependencies are not installed. Install with
• Stream with this message_id already exists
• scheduler.chat.mode={self.scheduler.chat.mode!r} requires st
• Handler must define function handle(input, context).

AI-assisted analysis of zylon-ai/private-gpt@4a030776a3 (2026-08-15). Data as JSON: /api/errors/fb97ac227124612d. Report an issue: GitHub.