{"record":{"id":"943ca5a294954ffd","repo":"zylon-ai/private-gpt","slug":"stream-with-correlation-id-event-correlation-id","errorCode":null,"errorMessage":"Stream with correlation_id {event.correlation_id} not found","messagePattern":"Stream with correlation_id (.+?) not found","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"private_gpt/components/streaming/providers/in_memory_stream_service.py","lineNumber":119,"sourceCode":"        async with self._lock:\n            if correlation_id not in self._metadata:\n                raise ValueError(\n                    f\"Stream with correlation_id {correlation_id} not found\"\n                )\n\n            # Generate sequential message ID like Redis\n            self._event_counters[correlation_id] += 1\n            message_id = f\"{int(datetime.now(UTC).timestamp() * 1000)}-{self._event_counters[correlation_id]}\"\n\n            self._events[correlation_id].append((message_id, event_data))\n            for waiter in list(self._waiters.get(correlation_id, set())):\n                waiter.set()\n            return message_id\n\n    async def push_event_batch(self, events: list[Event]) -> dict[str, str]:\n        for event in events:\n            if event.correlation_id not in self._metadata:\n                raise ValueError(\n                    f\"Stream with correlation_id {event.correlation_id} not found\"\n                )\n        grouped: dict[str, list[str]] = defaultdict(list)\n        for event in events:\n            grouped[event.correlation_id].append(event.event_data)\n\n        result: dict[str, str] = {}\n        for correlation_id, event_datas in grouped.items():\n            last_id = None\n            for event_data in event_datas:\n                last_id = await self.push_event(correlation_id, event_data)\n            if last_id:\n                result[correlation_id] = last_id\n        return result\n\n    async def read_events(\n        self,\n        correlation_id: str,","sourceCodeStart":101,"sourceCodeEnd":137,"githubUrl":"https://github.com/zylon-ai/private-gpt/blob/4a030776a31a901ad80b1bf4d7faa2c1a367efbb/private_gpt/components/streaming/providers/in_memory_stream_service.py#L101-L137","documentation":"Raised by InMemoryStreamService.push_event_batch during its pre-flight validation pass: it iterates all events and verifies every event.correlation_id exists in the metadata map before writing anything. This makes the batch operation fail atomically — no partial writes — if any single event references an unknown stream.","triggerScenarios":"Calling push_event_batch with a list where at least one Event has a correlation_id that was never created (or was deleted). The remaining events in the batch are not pushed.","commonSituations":"Batching events for multiple jobs where one job's stream was cleaned up mid-batch; mixed creation paths where one subsystem forgot to create its stream; reprocessing old buffered events after a restart.","solutions":["Pre-validate all correlation ids: missing = {e.correlation_id for e in events} - existing ids; create missing streams before the batch call","Split the batch per correlation_id and handle failures per stream so one bad id does not block the rest","Wrap the call in try/except ValueError, log the offending events, and drop/requeue only the invalid ones"],"exampleFix":"# before\nawait stream_service.push_event_batch(events)\n# after\nmissing = {\n    e.correlation_id\n    for e in events\n    if not await stream_service.stream_exists(e.correlation_id)\n}\nfor cid in missing:\n    await stream_service.create_stream(correlation_id=cid, stream_type=\"ingestion\")\nawait stream_service.push_event_batch(events)","handlingStrategy":"validation","validationCode":"existing = {\n    e.correlation_id\n    for e in events\n    if await stream_service.stream_exists(e.correlation_id)\n}\nmissing = {e.correlation_id for e in events} - existing\nfor cid in missing:\n    await stream_service.create_stream(correlation_id=cid, stream_type=\"ingestion\")\nawait stream_service.push_event_batch(events)","typeGuard":null,"tryCatchPattern":"try:\n    await stream_service.push_event_batch(events)\nexcept ValueError as e:\n    if \"not found\" not in str(e):\n        raise\n    # split per correlation_id and handle the bad id individually","preventionTips":["Remember push_event_batch is atomic — one bad id blocks the whole batch","Pre-validate all correlation ids when mixing streams from multiple jobs"],"tags":["streaming","in-memory","batch","not-found","atomicity"],"backgroundTag":null,"analyzedSha":"4a030776a31a901ad80b1bf4d7faa2c1a367efbb","analyzedAt":"2026-08-15T03:51:26.951Z","schemaVersion":2},"datasetVersion":"2026-08-15T17:31:12.345Z"}