{"record":{"id":"fb97ac227124612d","repo":"zylon-ai/private-gpt","slug":"stream-with-correlation-id-correlation-id-alread-fb97ac","errorCode":null,"errorMessage":"Stream with correlation_id {correlation_id} already exists","messagePattern":"Stream with correlation_id (.+?) already exists","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"private_gpt/components/streaming/providers/redis_stream_service.py","lineNumber":118,"sourceCode":"        created = await cast(Any, self._client.eval)(\n            \"\"\"\n            if redis.call('exists', KEYS[1]) == 1 then\n                return 0\n            end\n            local fields = cjson.decode(ARGV[1])\n            for key, value in pairs(fields) do\n                redis.call('hset', KEYS[1], key, value)\n            end\n            redis.call('expire', KEYS[1], ARGV[2])\n            return 1\n            \"\"\",\n            1,\n            status_key,\n            json.dumps(mapping),\n            self._config.expiry_seconds,\n        )\n        if not created:\n            raise ValueError(\n                f\"Stream with correlation_id {correlation_id} already exists\"\n            )\n\n        return correlation_id\n\n    async def update_stream_status(\n        self,\n        correlation_id: str,\n        status: StreamStatus,\n        error_message: str | None = None,\n        metadata: dict[str, Any] | None = None,\n    ) -> None:\n        \"\"\"Update stream status and metadata.\"\"\"\n        status_key = self._get_status_key(correlation_id)\n\n        now = datetime.now(UTC)\n        updates = {\n            \"status\": status.value,","sourceCodeStart":100,"sourceCodeEnd":136,"githubUrl":"https://github.com/zylon-ai/private-gpt/blob/4a030776a31a901ad80b1bf4d7faa2c1a367efbb/private_gpt/components/streaming/providers/redis_stream_service.py#L100-L136","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Check await stream_service.stream_exists(correlation_id) first and reuse or delete the existing stream","Generate a fresh uuid correlation id per attempt instead of a stable business id","If the stale stream is garbage, delete it (or wait for expiry_seconds) before re-creating"],"exampleFix":"# before\nawait stream_service.create_stream(correlation_id=job_id, stream_type=\"ingestion\")\n# after\nif await stream_service.stream_exists(job_id):\n    await stream_service.delete_stream(job_id)\nawait stream_service.create_stream(correlation_id=job_id, stream_type=\"ingestion\")","handlingStrategy":"validation","validationCode":"if await stream_service.stream_exists(correlation_id):\n    await stream_service.delete_stream(correlation_id)\nawait stream_service.create_stream(correlation_id=correlation_id, stream_type=stream_type)","typeGuard":null,"tryCatchPattern":"try:\n    await stream_service.create_stream(correlation_id=cid, stream_type=st)\nexcept ValueError as e:\n    if \"already exists\" not in str(e):\n        raise\n    logger.info(\"stream %s exists in Redis; reusing\", cid)","preventionTips":["Use uuid correlation ids per attempt with Redis (state outlives the request)","Remember expiry_seconds governs how long duplicate protection lasts"],"tags":["streaming","redis","duplicate","idempotency","distributed"],"backgroundTag":null,"analyzedSha":"4a030776a31a901ad80b1bf4d7faa2c1a367efbb","analyzedAt":"2026-08-15T03:51:26.951Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}