{"record":{"id":"c31db009ad8fdb6e","repo":"clockworklabs/SpacetimeDB","slug":"no-current-segment-expected-segment-header","errorCode":null,"errorMessage":"no current segment, expected segment header","messagePattern":"no current segment, expected segment header","errorType":"exception","errorClass":"io::Error","httpStatus":null,"severity":"error","filePath":"crates/commitlog/src/stream/writer.rs","lineNumber":233,"sourceCode":"                    let last_written_tx_range = self.last_written_tx_range.clone();\n                    let commitlog_options = self.commitlog_options;\n                    move || create_segment(repo, last_written_tx_range, commitlog_options, header)\n                })\n                .await\n                .unwrap()\n                .map(|(segment, index)| (segment.into_async_writer(), index))?;\n                stream.consume(segment::Header::LEN as _);\n\n                CurrentSegment {\n                    header,\n                    segment,\n                    offset_index: index,\n                }\n            } else {\n                match self.current_segment.take() {\n                    Some(current_segment) => current_segment,\n                    _ => {\n                        return Err(io::Error::new(\n                            io::ErrorKind::InvalidData,\n                            \"no current segment, expected segment header\",\n                        ));\n                    }\n                }\n            };\n\n            // What follows is commits to be written to `current_segment`,\n            // until we encounter EOF or a segment marker.\n            let res = self\n                .append_all_inner(&mut stream, &mut current_segment, &mut progress)\n                .await;\n            // Ensure we flush application buffers (BufWriter).\n            current_segment.segment.flush().await?;\n            let maybe_eof = res?;\n            // Put back segment, so it is available for syncing or closing.\n            self.current_segment = Some(current_segment);\n            match maybe_eof {","sourceCodeStart":215,"sourceCodeEnd":251,"githubUrl":"https://github.com/clockworklabs/SpacetimeDB/blob/524b4487d949b61a07d4f39c862d1290259dfd20/crates/commitlog/src/stream/writer.rs#L215-L251","documentation":"StreamWriter::append_all ingests a replication/bootstrap stream into the local commitlog. A segment is only set up when the stream yields a buffer starting with the segment magic (a segment header); otherwise data is appended to the latest segment found when the writer was created. This error means neither was available: the stream's next chunk was not a segment header AND the writer has no current segment to append to (the target repo was empty at create time).","triggerScenarios":"Calling StreamWriter::append_all with a stream that begins mid-segment (no leading segment header) against an empty or newly created commitlog repo; resuming a stream from a mid-segment offset after the local log was wiped; a producer that skips the header, or a framing/protocol version mismatch between producer and consumer.","commonSituations":"Follower re-bootstrap into a cleared data directory while the leader starts streaming from a non-zero offset; test harnesses feeding hand-crafted commit streams; producer and consumer built from different commitlog format versions.","solutions":["Start the stream at offset 0 (or at a segment boundary) so a segment header precedes the commits.","Re-create the writer on a repo that still contains the segment being continued, so the current segment is populated.","Align producer and consumer on the same commitlog crate/format version."],"exampleFix":"// before: feeding a mid-segment stream into a fresh repo\nlet writer = StreamWriter::create(repo, opts, OnTrailingData::Error)?;\nlet writer = writer.append_all(stream_from_offset(mid_segment_tx), progress).await?; // Err: no current segment\n\n// after: begin at the segment start so the header arrives first\nlet writer = StreamWriter::create(repo, opts, OnTrailingData::Error)?;\nlet writer = writer.append_all(stream_from_offset(0), progress).await?;","handlingStrategy":"validation","validationCode":"use tokio::io::{AsyncBufRead, AsyncBufReadExt};\n\nasync fn stream_starts_with_segment_header<S: AsyncBufRead + Unpin>(\n    stream: &mut S,\n) -> std::io::Result<bool> {\n    let buf = stream.fill_buf().await?;\n    Ok(buf.starts_with(&segment::MAGIC))\n}","typeGuard":null,"tryCatchPattern":"match writer.append_all(stream, progress).await {\n    Err(e) if e.kind() == io::ErrorKind::InvalidData && e.to_string().contains(\"expected segment header\") => {\n        // Restart the feed from offset 0 / a segment boundary and re-create the writer.\n    }\n    r => r,\n}","preventionTips":["Always begin replication streams with the segment header of the first segment sent.","Do not clear a follower's commitlog while resuming from a non-zero offset.","Pin producer and consumer to the same commitlog crate version."],"tags":["commitlog","streaming","replication","rust","spacetimedb"],"backgroundTag":"stream-framing-error","analyzedSha":"524b4487d949b61a07d4f39c862d1290259dfd20","analyzedAt":"2026-08-16T23:58:54.611Z","schemaVersion":2},"datasetVersion":"2026-08-17T04:17:16.089Z"}