{"record":{"id":"7aac65433178037e","repo":"risingwavelabs/risingwave","slug":"end-of-log-stream","errorCode":null,"errorMessage":"end of log stream","messagePattern":"end of log stream","errorType":"exception","errorClass":"anyhow::Error","httpStatus":null,"severity":"error","filePath":"src/stream/src/common/log_store_impl/in_mem.rs","lineNumber":226,"sourceCode":"                            };\n                        } else {\n                            self.epoch_progress = Consuming(next_epoch);\n                        }\n                        self.latest_offset = TruncateOffset::Barrier {\n                            epoch: current_epoch,\n                        };\n                        Ok((\n                            current_epoch,\n                            LogStoreReadItem::Barrier {\n                                is_checkpoint: options.is_checkpoint,\n                                new_vnode_bitmap: options.new_vnode_bitmap,\n                                is_stop: options.is_stop,\n                                schema_change: options.schema_change,\n                            },\n                        ))\n                    }\n                },\n                None => Err(anyhow!(\"end of log stream\")),\n            },\n            AwaitingTruncate { .. } => std::future::pending().await,\n        }\n    }\n\n    fn truncate(&mut self, offset: TruncateOffset) -> LogStoreResult<()> {\n        // check the truncate offset is higher than prev truncate offset\n        if self.truncate_offset >= offset {\n            return Err(anyhow!(\n                \"truncate offset {:?} but prev truncate offset is {:?}\",\n                offset,\n                self.truncate_offset\n            ));\n        }\n\n        // check the truncate offset does not exceed the latest possible offset\n        if offset > self.latest_offset {\n            return Err(anyhow!(","sourceCodeStart":208,"sourceCodeEnd":244,"githubUrl":"https://github.com/risingwavelabs/risingwave/blob/6469eb736d691e8e9b8a419a57edd6429ca77417/src/stream/src/common/log_store_impl/in_mem.rs#L208-L244","documentation":"The in-memory log store's reader attempted to fetch the next log item but its buffer was exhausted (None from the queue), so it reports the log stream has ended. Reading past the end of an in-memory log stream is treated as a fatal error rather than a normal EOF.","triggerScenarios":"next_item called after all appended items were consumed and the stream was not extended; occurs when a stream consumer outpaces producers or after the log was truncated/reset.","commonSituations":"Actor/barrier reader advancing beyond the latest appended entry, a crashed or stalled producer leaving the stream short, or stream recovery logic reading a closed in-memory log.","solutions":["Check why the producer stopped appending (upstream actor failure) and restart the source","Verify barrier/stream recovery logic does not read past latest_offset","Re-run/recover the streaming job so the log store is rebuilt with correct offsets"],"exampleFix":null,"handlingStrategy":"retry","validationCode":null,"typeGuard":null,"tryCatchPattern":"match log_reader.next_item().await {\n    Err(e) if e.to_string().contains(\"end of log stream\") => {\n        // wait for producer or trigger recovery/restart of the actor\n        tokio::time::sleep(Duration::from_millis(100)).await;\n    }\n    other => other?,\n}","preventionTips":["Ensure producers append before consumers advance","Add liveness monitoring for stream producer actors","Model recovery so readers never read past latest_offset"],"tags":["stream","log-store","eof"],"backgroundTag":"stream-ended-unexpectedly","analyzedSha":"6469eb736d691e8e9b8a419a57edd6429ca77417","analyzedAt":"2026-09-11T21:06:21.487Z","contentChangedAt":"2026-09-11T21:06:21.487Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}