| 232 | type SegmentReader = ReadOnlySegment; |
| 233 | |
| 234 | fn create_segment(&self, offset: u64, header: segment::Header) -> io::Result<Self::SegmentWriter> { |
| 235 | let path = self.segment_path(offset); |
| 236 | |
| 237 | // We need to check if the segment already exists, |
| 238 | // so use file locking to prevent a TOCTOU race. |
| 239 | // Using `flock` means we don't need to worry about stale lockfiles. |
| 240 | let lock_path = path.0.with_extension("lock"); |
| 241 | let _lock = scopeguard::guard( |
| 242 | lockfile::advisory::LockedFile::lock(&lock_path) |
| 243 | .map_err(|e| io::Error::new(e.source.kind(), format!("repo {}: {}: {}", self, e, e.source)))?, |
| 244 | |lockfile| { |
| 245 | if let Err(e) = lockfile.release(true) { |
| 246 | // It's ok if removing the file fails, but print a warning |
| 247 | // anyways. |
| 248 | warn!("repo {}: failed to remove {}: {}", self, lock_path.display(), e); |
| 249 | } |
| 250 | }, |
| 251 | ); |
| 252 | |
| 253 | // Check whether the segment already exists. |
| 254 | // Overwrite it if its length is zero. |
| 255 | match fs::metadata(&path) { |
| 256 | Ok(stat) => { |
| 257 | if stat.len() > 0 { |
| 258 | return Err(io::Error::new( |
| 259 | io::ErrorKind::AlreadyExists, |
| 260 | format!("repo {}: segment {} already exists and is non-empty", self, offset), |
| 261 | )); |
| 262 | } |
| 263 | } |
| 264 | Err(e) => { |
| 265 | if e.kind() != io::ErrorKind::NotFound { |
| 266 | return Err(io::Error::new( |
| 267 | e.kind(), |
| 268 | format!( |
| 269 | "repo {}: error getting file metadata for segment {}: {}", |
| 270 | self, offset, e |
| 271 | ), |
| 272 | )); |
| 273 | } |
| 274 | } |
| 275 | } |
| 276 | |
| 277 | // The segment file either does not exist, or is of length zero. |
| 278 | // Write the header to a temporary file and atomically move it into place. |
| 279 | let mut tmp = tempfile::Builder::new().make_in(&self.root.0, |tmp_path| { |
| 280 | File::options().read(true).write(true).create_new(true).open(tmp_path) |
| 281 | })?; |
| 282 | header.write(&mut tmp)?; |
| 283 | tmp.as_file_mut().sync_all()?; |
| 284 | let segment = tmp.persist(path)?; |
| 285 | |
| 286 | // Notify subscribers. |
| 287 | if let Some(on_new_segment) = self.on_new_segment.as_ref() { |
| 288 | on_new_segment(); |
| 289 | } |
| 290 | |
| 291 | Ok(segment) |