tonbo 0.4.0-a1

Embedded database for serverless and edge runtimes, storing data as Parquet on S3
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

//! Helpers that bridge WAL metadata into manifest structures.

use std::sync::Arc;

use crate::{
    manifest::WalSegmentRef,
    wal::{
        WalConfig, WalError,
        storage::{SegmentDescriptor, SegmentFrameBounds, WalStorage},
        wal_segment_file_id,
    },
};

/// Collect WAL segment references using the configuration supplied to the writer.
pub(crate) async fn collect_wal_segment_refs(
    cfg: &WalConfig,
    manifest_floor: Option<&WalSegmentRef>,
    live_frame_floor: Option<u64>,
) -> Result<Vec<WalSegmentRef>, WalError> {
    let storage = WalStorage::new(Arc::clone(&cfg.segment_backend), cfg.dir.clone());
    let wal_state_hint = storage
        .load_state_handle(cfg.state_store.as_ref())
        .await?
        .and_then(|handle| handle.state().last_segment_seq);

    let Some(tail) = storage.tail_metadata_with_hint(wal_state_hint).await? else {
        return Ok(Vec::new());
    };

    let mut refs = Vec::with_capacity(2);
    let manifest_cutoff = manifest_floor.map(|f| f.seq());
    for descriptor in tail
        .completed
        .iter()
        .filter(|descriptor| descriptor.bytes > 0)
    {
        let bounds = storage
            .segment_frame_bounds(&descriptor.path)
            .await?
            .ok_or({
                WalError::Corrupt("wal segment contained no frames despite non-zero length")
            })?;
        if segment_required(descriptor.seq, &bounds, manifest_cutoff, live_frame_floor) {
            refs.push(wal_segment_ref_from_descriptor(descriptor, bounds));
        }
    }

    if tail.active.bytes > 0 {
        let bounds = storage
            .segment_frame_bounds(&tail.active.path)
            .await?
            .ok_or({
                WalError::Corrupt("active wal segment contained no frames despite non-zero length")
            })?;
        if segment_required(tail.active.seq, &bounds, manifest_cutoff, live_frame_floor) {
            refs.push(wal_segment_ref_from_descriptor(&tail.active, bounds));
        }
    }

    // Note: We intentionally don't re-add the old floor when refs is empty.
    // If no segments are required (all data persisted), the floor should be cleared.
    // This ensures WAL replay doesn't replay already-persisted data.

    refs.sort_by_key(|segment| segment.seq());
    refs.dedup_by_key(|segment| segment.seq());

    Ok(refs)
}

fn wal_segment_ref_from_descriptor(
    descriptor: &SegmentDescriptor,
    bounds: SegmentFrameBounds,
) -> WalSegmentRef {
    let file_id = wal_segment_file_id(descriptor.seq);
    WalSegmentRef::new(descriptor.seq, file_id, bounds.first_seq, bounds.last_seq)
}

fn segment_required(
    seq: u64,
    bounds: &SegmentFrameBounds,
    manifest_cutoff: Option<u64>,
    live_frame_floor: Option<u64>,
) -> bool {
    if let Some(live_floor) = live_frame_floor
        && bounds.last_seq >= live_floor
    {
        return true;
    }

    match manifest_cutoff {
        Some(cutoff) => seq > cutoff,
        None => true,
    }
}

/// Prune WAL segments whose sequence is strictly below the manifest floor.
pub(crate) async fn prune_wal_segments(
    cfg: &WalConfig,
    floor: &WalSegmentRef,
) -> Result<usize, WalError> {
    let storage = WalStorage::new(Arc::clone(&cfg.segment_backend), cfg.dir.clone());
    if cfg.prune_dry_run {
        let segments = storage.list_segments_with_hint(None).await?;
        let removable = segments
            .into_iter()
            .filter(|descriptor| descriptor.seq < floor.seq())
            .count();
        Ok(removable)
    } else {
        storage.prune_below(floor.seq()).await
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn manifest_floor_filters_when_unpinned() {
        let bounds = SegmentFrameBounds {
            first_seq: 10,
            last_seq: 20,
        };
        assert!(!segment_required(5, &bounds, Some(5), None));
        assert!(segment_required(6, &bounds, Some(5), None));
    }

    #[test]
    fn live_floor_keeps_older_segments() {
        let bounds = SegmentFrameBounds {
            first_seq: 30,
            last_seq: 40,
        };
        assert!(segment_required(3, &bounds, Some(5), Some(35)));
        assert!(!segment_required(3, &bounds, Some(5), Some(45)));
    }
}