Function trim_garbage

crates/commitlog/tests/streaming/mod.rs:105–159

()

#[tokio::test]
async fn trim_garbage() {
    enable_logging();

    let root = tempdir().unwrap();
    let (src, dst) = create_dirs(root.path()).await;
    fill_log(src.clone()).await;

    {
        let writer = create_writer(dst.clone())
            .await
            .expect("failed to create stream writer");
        let reader = create_reader(&src, ..);
        pin!(reader);
        writer.append_all(reader, |_| ()).await.unwrap();
        assert_equal_dirs(&src, &dst).await
    }

    // Truncate the destination log so the last commit is broken.
    spawn_blocking({
        let repo = repo(&dst);
        move || {
            let last_segment_offset = repo.existing_offsets().unwrap().pop().unwrap();
            let mut segment = repo.open_segment_writer(last_segment_offset).unwrap();
            let len = segment.segment_len().unwrap();
            segment.set_len(len - 128).unwrap();
        }
    })
    .await
    .unwrap();
    // The default is to return an error.
    pretty_assertions::assert_matches!(
        create_writer(dst.clone()).await.map(drop),
        Err(e) if e.kind() == io::ErrorKind::InvalidData
    );

    // With `Trim`, we can retry from commit 99.
    let writer = spawn_blocking({
        let path = dst.clone();
        move || StreamWriter::create(repo(&path), default_options(), OnTrailingData::Trim)
    })
    .await
    .unwrap()
    .expect("failed to create stream writer");
    let reader = create_reader(&src, 99..);
    pin!(reader);
    writer
        .append_all(reader, |_| ())
        .await
        .unwrap()
        .sync_all()
        .await
        .unwrap();

    assert_equal_dirs(&src, &dst).await
}

async fn assert_equal_dirs(src: &Path, dst: &Path) {
    let mut src_dir = fs::read_dir(src).await.map(ReadDirStream::new).unwrap();