{"record":{"id":"51c2caf395fa9bf2","repo":"clockworklabs/SpacetimeDB","slug":"expected-commit-offset-but-encountered","errorCode":null,"errorMessage":"expected commit offset {} but encountered {}","messagePattern":"expected commit offset (.+?) but encountered (.+?)","errorType":"exception","errorClass":"io::Error","httpStatus":null,"severity":"error","filePath":"crates/commitlog/src/stream/writer.rs","lineNumber":327,"sourceCode":"            // Read the rest of the commit.\n            self.commit_buf.body.resize(\n                commit_header.len as usize + CHECKSUM_LEN[current_segment.header.checksum_algorithm as usize],\n                0,\n            );\n            stream.read_exact(&mut self.commit_buf.body).await?;\n            // Decode the commit and verify its checksum.\n            let commit = StoredCommit::decode(&mut self.commit_buf.as_reader(bytes_written))\n                .inspect_err(|e| warn!(\"failed to decode commit: {e}\"))?\n                .expect(\"commit decode cannot return `None` because we already decoded the header\");\n\n            // Check that the commit offset is what we expect.\n            let expected_offset = self\n                .last_written_tx_range\n                .as_ref()\n                .map(|range| range.end)\n                .unwrap_or_default();\n            if commit.min_tx_offset != expected_offset {\n                return Err(io::Error::new(\n                    io::ErrorKind::InvalidData,\n                    format!(\n                        \"expected commit offset {} but encountered {}\",\n                        expected_offset, commit.min_tx_offset\n                    ),\n                ));\n            }\n            trace!(\"received commit {commit:?}\");\n\n            // Write the commit and report progress.\n            current_segment\n                .segment\n                .write_all_buf(&mut self.commit_buf.as_buf())\n                .await?;\n            let written_range = commit.min_tx_offset..(commit.min_tx_offset + commit.n as u64);\n            self.last_written_tx_range = Some(written_range.clone());\n            progress.range_written(written_range);\n","sourceCodeStart":309,"sourceCodeEnd":345,"githubUrl":"https://github.com/clockworklabs/SpacetimeDB/blob/9e0d92412ff2248f401a8ad12d535f2b5ac30912/crates/commitlog/src/stream/writer.rs#L309-L345","documentation":"StreamWriter enforces commit-offset contiguity across the whole stream: each commit's min_tx_offset must equal the end of the last written tx range (0 when nothing was written yet). A gap or duplicate returns io::ErrorKind::InvalidData 'expected commit offset X but encountered Y', identifying exactly where the stream and destination diverged.","triggerScenarios":"Starting the source stream at an offset that does not continue the destination's last written range; dropped or duplicated data in the network transport; the producer restarting from a different point mid-stream; resuming a mirror job with a stale source offset.","commonSituations":"Reconnecting replication with a wrong 'resume from' offset; source and destination out of sync after a partial failure; replaying an old stream snippet over newer data.","solutions":["Query the destination's actual position (StreamWriter::create_and_metadata returns the last segment's Metadata; use its tx_range.end) and request the source stream from exactly that offset","On mismatch, do not skip forward blindly: restart the mirror from a verified common offset into a clean state","Verify the transport (retry buffering, dedup) is not dropping or duplicating stream chunks","Log the expected/actual pair to identify whether data was lost (gap) or repeated (duplicate)"],"exampleFix":null,"handlingStrategy":"validation","validationCode":"// Align the source stream start with the destination before writing\nlet (_, meta) = StreamWriter::create_and_metadata(repo.clone(), opts.clone(), OnTrailingData::Trim)?;\nlet resume_from = meta.as_ref().map(|m| m.tx_range.end).unwrap_or(0);\n// request the source to stream commits starting exactly at `resume_from`\nlet stream = source.stream_from_tx(resume_from).await?;","typeGuard":"fn is_stream_offset_gap(err: &io::Error) -> bool {\n    err.kind() == io::ErrorKind::InvalidData && err.to_string().contains(\"expected commit offset\")\n}","tryCatchPattern":"match writer.append_all(&mut stream, &mut progress).await {\n    Ok(done) => Ok(done),\n    Err(e) if is_stream_offset_gap(&e) => {\n        // destination and source diverged: tear down and restart the mirror aligned\n        restart_mirror_aligned(&repo, &source).await\n    }\n    Err(e) => Err(e),\n}","preventionTips":["Derive the mirror resume offset from the destination's metadata, never from a config constant","Make the transport at-least-once with dedup at the writer, so retries cannot create gaps","Persist the last written tx range on the destination after each progress report"],"tags":["rust","commitlog","streaming","contiguity","replication"],"backgroundTag":"sequence-number-gap","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"}