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
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
use crate::error::Result;
use std::fs::{File, OpenOptions};
use std::io::{Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
const FILENAME_DIGITS: usize = 20;
const FILENAME_EXTENSION: &str = "log";
/// One on-disk `.log` segment file plus its in-memory append cursor.
///
/// Segments are owned exclusively by their partition's writer task; the
/// `cursor` optimization (see field doc) depends on that single-writer
/// invariant. Callers should not construct or operate on `Segment` directly —
/// it's exposed for downstream tools that want to read raw segments.
pub struct Segment {
base_offset: u64,
path: PathBuf,
file: File,
size: u64,
// Tracked cursor position to avoid redundant seeks. None = unknown, force seek.
//
// This optimization is sound ONLY because the partition writer task owns this
// Segment exclusively (the single-writer-per-partition invariant). With one
// owner, no other task can change `file`'s position between our seek and the
// following read/write — so we can skip the seek when we already know we're
// at the requested position. If a second concurrent caller were ever to touch
// `file` (currently impossible by construction), this field would become a
// correctness hazard: the cached cursor could disagree with the kernel's view
// and a "no seek needed" branch would read or write at the wrong offset.
cursor: Option<u64>,
}
impl Segment {
/// Creates a fresh empty segment file at `dir/<base_offset>.log`. Errors
/// if a file at that path already exists.
pub async fn create(dir: &Path, base_offset: u64) -> Result<Self> {
let path = segment_path(dir, base_offset);
let file = OpenOptions::new()
.create_new(true)
.read(true)
.write(true)
.open(&path)?;
Ok(Self {
base_offset,
path,
file,
size: 0,
cursor: Some(0),
})
}
/// Opens an existing segment file at `dir/<base_offset>.log`. Errors if
/// the file is missing.
pub async fn open(dir: &Path, base_offset: u64) -> Result<Self> {
let path = segment_path(dir, base_offset);
let metadata = std::fs::metadata(&path)?;
let file = OpenOptions::new().read(true).write(true).open(&path)?;
Ok(Self {
base_offset,
path,
file,
size: metadata.len(),
cursor: Some(0),
})
}
/// Returns the absolute offset of the first record in this segment.
pub fn base_offset(&self) -> u64 {
self.base_offset
}
/// Returns the path of the underlying `.log` file.
pub fn path(&self) -> &Path {
&self.path
}
/// Returns the current size of the segment file in bytes.
pub fn size(&self) -> u64 {
self.size
}
/// Returns true if appending `additional` bytes would push this segment
/// past `threshold`. Used by [`Log`] to decide when to rotate.
///
/// An empty segment always accepts the first write, even if `additional`
/// already exceeds `threshold` — otherwise rotation would try to create a
/// new segment at the same base offset and fail. The size cap is a soft
/// target: a single oversized append produces one oversized segment, and
/// rotation kicks in on the next append.
///
/// [`Log`]: crate::Log
pub fn would_overflow(&self, additional: usize, threshold: u64) -> bool {
if self.size == 0 {
return false;
}
self.size + additional as u64 > threshold
}
/// Appends `bytes` at the end of the segment file and returns the file
/// position where the write started (i.e. the segment-relative offset).
#[cfg_attr(feature = "hotpath", hotpath::measure)]
pub async fn append(&mut self, bytes: &[u8]) -> Result<u64> {
let file_pos = self.size;
self.ensure_cursor(file_pos)?;
self.file.write_all(bytes)?;
self.cursor = Some(file_pos + bytes.len() as u64);
self.size = file_pos + bytes.len() as u64;
Ok(file_pos)
}
/// Reads bytes from `file_pos` into `into`, returning the number of bytes
/// actually read. Short reads are possible at EOF.
pub async fn read_at(&mut self, file_pos: u64, into: &mut [u8]) -> Result<usize> {
self.ensure_cursor(file_pos)?;
let n = self.file.read(into)?;
self.cursor = Some(file_pos + n as u64);
Ok(n)
}
/// Truncates the segment to `new_size` bytes and fsyncs. Used by the
/// recovery path to drop torn or corrupted records at the tail.
pub async fn truncate(&mut self, new_size: u64) -> Result<()> {
self.file.set_len(new_size)?;
self.file.sync_data()?;
self.size = new_size;
// truncate may leave cursor past EOF on some platforms; force re-seek next op
self.cursor = None;
Ok(())
}
fn ensure_cursor(&mut self, pos: u64) -> Result<()> {
if self.cursor != Some(pos) {
self.file.seek(SeekFrom::Start(pos))?;
self.cursor = Some(pos);
}
Ok(())
}
/// Fsyncs the segment file to disk via `sync_data` (durable data, metadata
/// may lag).
#[cfg_attr(feature = "hotpath", hotpath::measure)]
pub async fn sync(&mut self) -> Result<()> {
self.file.sync_data()?;
Ok(())
}
/// Returns the segment file's last-modified time in Unix epoch milliseconds.
/// Used by age-based retention.
pub async fn last_modified_ms(&self) -> Result<i64> {
let metadata = std::fs::metadata(&self.path)?;
let modified = metadata.modified()?;
let ms = modified
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_millis() as i64)
.unwrap_or(0);
Ok(ms)
}
}
fn segment_path(dir: &Path, base_offset: u64) -> PathBuf {
dir.join(format!(
"{:0width$}.{ext}",
base_offset,
width = FILENAME_DIGITS,
ext = FILENAME_EXTENSION
))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn segment_path_is_zero_padded_20_digits() {
let path = segment_path(Path::new("data"), 42);
assert_eq!(
path.file_name().unwrap().to_str().unwrap(),
"00000000000000000042.log"
);
let path = segment_path(Path::new("data"), 0);
assert_eq!(
path.file_name().unwrap().to_str().unwrap(),
"00000000000000000000.log"
);
let path = segment_path(Path::new("data"), 10_000_000);
assert_eq!(
path.file_name().unwrap().to_str().unwrap(),
"00000000000010000000.log"
);
}
}