{"record":{"id":"c724841be81fd022","repo":"zylon-ai/private-gpt","slug":"stream-with-correlation-id-correlation-id-alread","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/in_memory_stream_service.py","lineNumber":54,"sourceCode":"        metadata: dict[str, Any] | None = None,\n    ) -> str:\n        \"\"\"Create a new stream and return correlation ID.\"\"\"\n        if correlation_id is None:\n            correlation_id = str(uuid.uuid4())\n\n        now = datetime.now(UTC)\n        stream_metadata = StreamMetadata(\n            correlation_id=correlation_id,\n            status=StreamStatus.PENDING,\n            created_at=now,\n            updated_at=now,\n            stream_type=stream_type,\n            metadata=metadata or {},\n        )\n\n        async with self._lock:\n            if correlation_id in self._metadata:\n                raise ValueError(\n                    f\"Stream with correlation_id {correlation_id} already exists\"\n                )\n            self._metadata[correlation_id] = stream_metadata\n            self._events[correlation_id] = []\n            self._event_counters[correlation_id] = 0\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        async with self._lock:\n            if correlation_id not in self._metadata:","sourceCodeStart":36,"sourceCodeEnd":72,"githubUrl":"https://github.com/zylon-ai/private-gpt/blob/4a030776a31a901ad80b1bf4d7faa2c1a367efbb/private_gpt/components/streaming/providers/in_memory_stream_service.py#L36-L72","documentation":"Raised by InMemoryStreamService.create_stream when a stream with the same correlation_id is registered twice. The service keys metadata, events and counters by correlation_id, so duplicate creation would overwrite state; the check runs under the async lock to guarantee uniqueness. The offending correlation_id is included in the message.","triggerScenarios":"Calling create_stream(correlation_id=X) twice without delete_stream(X) in between; retrying an ingestion/job request that reuse the same client-supplied correlation id; concurrent workers registering the same id.","commonSituations":"At-least-once job queues redelivering a task that already created its stream; client retries after a timeout where the first create actually succeeded; idempotency keys reused across runs.","solutions":["Before creating, check with await stream_service.stream_exists(correlation_id) and reuse or delete the existing stream","Use a fresh correlation id (e.g. uuid4) for each new stream instead of reusing a business-level id","On retry paths, catch the ValueError and resume consuming the existing stream rather than re-creating it"],"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    # reuse the existing stream","preventionTips":["Generate a uuid correlation_id per attempt instead of reusing business ids","Treat create as non-idempotent: always pair with exists/delete on retry paths"],"tags":["streaming","in-memory","duplicate","idempotency"],"backgroundTag":null,"analyzedSha":"4a030776a31a901ad80b1bf4d7faa2c1a367efbb","analyzedAt":"2026-08-15T03:51:26.951Z","schemaVersion":2},"datasetVersion":"2026-08-15T22:17:37.221Z"}