{"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/9e0d92412ff2248f401a8ad12d535f2b5ac30912/crates/commitlog/src/stream/writer.rs#L215-L251","documentation":"StreamWriter::append_all consumes a raw commitlog byte stream that must contain segment headers before any commit data. When the destination commitlog is empty (no current segment, per StreamWriter::create finding no existing segments) and commit bytes arrive before any segment marker, the writer returns io::ErrorKind::InvalidData 'no current segment, expected segment header'.","triggerScenarios":"Replaying a stream captured starting mid-segment (producer skipped the segment header); a stream whose leading segment marker was truncated or lost in transport; feeding arbitrary or desynchronized data into StreamWriter.","commonSituations":"Restarting network mirroring from a byte offset inside a segment instead of a segment boundary; a proxy/transport that dropped the first bytes; hand-built test streams that omit the header.","solutions":["Ensure the source stream starts at a segment boundary and includes the segment header (start streaming from the beginning of a segment)","If resuming into an existing destination repo, verify the repo actually contains the prior segment so a current segment exists","Regenerate the stream from a known-good source rather than patching the desynchronized one","Validate the first bytes of the stream against the '(ds)^2' magic before starting the writer"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Verify the stream begins with a segment header before handing it to StreamWriter\nlet mut peek = [0u8; 6];\nlet n = stream.peek(&mut peek).await?; // e.g. via a BufReader fill/peek helper\nif n < 6 || &peek != b\"(ds)^2\" {\n    return Err(io::Error::new(\n        io::ErrorKind::InvalidData,\n        \"stream does not start with a segment header; restart the source at a segment boundary\",\n    ));\n}","typeGuard":"fn is_missing_segment_marker(err: &io::Error) -> bool {\n    err.kind() == io::ErrorKind::InvalidData\n        && err.to_string().contains(\"no current segment, expected segment header\")\n}","tryCatchPattern":"match writer.append_all(&mut stream, &mut progress).await {\n    Ok(done) => Ok(done),\n    Err(e) if is_missing_segment_marker(&e) => {\n        // restart the source from a segment boundary, then retry with a fresh StreamWriter\n        Err(e) // non-recoverable for this stream; caller must re-source it\n    }\n    Err(e) => Err(e),\n}","preventionTips":["Always start mirrored streams at a segment boundary so the header precedes commit data","On the producer side, emit the segment header before any commit bytes of that segment","Validate the leading magic bytes of the stream before starting the writer"],"tags":["rust","commitlog","streaming","mirror","invalid-data"],"backgroundTag":"invalid-stream-format","analyzedSha":"9e0d92412ff2248f401a8ad12d535f2b5ac30912","analyzedAt":"2026-08-20T06:08:37.179Z","contentChangedAt":"2026-08-20T06:08:37.179Z","schemaVersion":2},"datasetVersion":"2026-09-23T08:17:48.524Z"}