Consume `stream` and append it to the local commitog. The `stream` should be the suffix after the commitlog already present in the local [`Repo`]. The method checks that commit offsets are contiguous. Segments are created whenever the stream yields a segment header. If the stream doesn't start with a segment header, the data is appended to the latest segment found when the writer was created. A
(
mut self,
mut stream: impl AsyncBufRead + Unpin,
mut progress: impl Progress,
)
| 188 | /// The caller should use [`Self::sync_all`] to ensure that if a segment |
| 189 | /// remains open after `append_all`, it is synced to disk. |
| 190 | pub async fn append_all( |
| 191 | mut self, |
| 192 | mut stream: impl AsyncBufRead + Unpin, |
| 193 | mut progress: impl Progress, |
| 194 | ) -> io::Result<Self> { |
| 195 | loop { |
| 196 | let Some(buf) = peek_buf(&mut stream).await? else { |
| 197 | break; |
| 198 | }; |
| 199 | |
| 200 | let mut current_segment = if buf.starts_with(&segment::MAGIC) { |
| 201 | // Ensure the previous segment, if any, is fsync'ed. |
| 202 | self.close_current_segment().await?; |
| 203 | // Ensure we actually have a valid segment header. |
| 204 | let header = |
| 205 | segment::Header::decode(buf).inspect_err(|e| warn!("failed to decode segment header: {e}"))?; |
| 206 | trace!( |
| 207 | "create segment at {}", |
| 208 | self.last_written_tx_range |
| 209 | .as_ref() |
| 210 | .map(|range| range.end) |
| 211 | .unwrap_or_default() |
| 212 | ); |
| 213 | let (segment, index) = spawn_blocking({ |
| 214 | let repo = self.repo.clone(); |
| 215 | let last_written_tx_range = self.last_written_tx_range.clone(); |
| 216 | let commitlog_options = self.commitlog_options; |
| 217 | move || create_segment(repo, last_written_tx_range, commitlog_options, header) |
| 218 | }) |
| 219 | .await |
| 220 | .unwrap() |
| 221 | .map(|(segment, index)| (segment.into_async_writer(), index))?; |
| 222 | stream.consume(segment::Header::LEN as _); |
| 223 | |
| 224 | CurrentSegment { |
| 225 | header, |
| 226 | segment, |
| 227 | offset_index: index, |
| 228 | } |
| 229 | } else { |
| 230 | match self.current_segment.take() { |
| 231 | Some(current_segment) => current_segment, |
| 232 | _ => { |
| 233 | return Err(io::Error::new( |
| 234 | io::ErrorKind::InvalidData, |
| 235 | "no current segment, expected segment header", |
| 236 | )); |
| 237 | } |
| 238 | } |
| 239 | }; |
| 240 | |
| 241 | // What follows is commits to be written to `current_segment`, |
| 242 | // until we encounter EOF or a segment marker. |
| 243 | let res = self |
| 244 | .append_all_inner(&mut stream, &mut current_segment, &mut progress) |
| 245 | .await; |
| 246 | // Ensure we flush application buffers (BufWriter). |
| 247 | current_segment.segment.flush().await?; |