use super::*;
use crate::background::{CheckpointBytes, CheckpointWorkContext, CheckpointWorkError};
#[derive(Debug)]
pub(crate) enum CheckpointImmutableFileError {
Io(io::Error),
Work(CheckpointWorkError),
}
impl From<CheckpointWorkError> for CheckpointImmutableFileError {
fn from(error: CheckpointWorkError) -> Self {
Self::Work(error)
}
}
impl CheckpointImmutableFileError {
fn io(error: io::Error, work: &CheckpointWorkContext) -> Self {
match hawdb_core::error::file_descriptor_error(&error) {
Some(error) => {
Self::Work(work.record_failure(CheckpointWorkError::FileDescriptors(error)))
}
None => Self::Io(error),
}
}
}
impl ImmutableFileHandles {
fn try_checkpoint_opening(
&self,
work: &CheckpointWorkContext,
) -> Result<std::sync::MutexGuard<'_, ()>, CheckpointImmutableFileError> {
match self.opening.try_lock() {
Ok(opening) => Ok(opening),
Err(std::sync::TryLockError::Poisoned(error)) => Ok(error.into_inner()),
Err(std::sync::TryLockError::WouldBlock) => Err(work
.record_failure(CheckpointWorkError::Contended("immutable file validation"))
.into()),
}
}
pub(crate) fn checkpoint_get(
&self,
binding: &ImmutableFileBinding,
context: &FileOpenContext,
work: &CheckpointWorkContext,
) -> Result<Arc<File>, CheckpointImmutableFileError> {
{
let unit = work.start_unit()?;
let cached = self.cached(binding.reference);
unit.finish();
work.checkpoint()?;
if let Some(file) = cached {
self.state.record_cache_hit();
return Ok(file);
}
}
let _validation = {
let unit = work.start_unit()?;
let opening = self.try_checkpoint_opening(work)?;
let validation = match self.checkpoint_validation.try_read() {
Ok(validation) => validation,
Err(std::sync::TryLockError::Poisoned(error)) => error.into_inner(),
Err(std::sync::TryLockError::WouldBlock) => {
return Err(work
.record_failure(CheckpointWorkError::Contended("immutable file retirement"))
.into());
}
};
let cached = self.cached(binding.reference);
drop(opening);
unit.finish();
work.checkpoint()?;
if let Some(file) = cached {
self.state.record_cache_hit();
return Ok(file);
}
validation
};
self.state.record_cache_miss();
let scratch_length = binding.reference.byte_length.clamp(1, 64 * 1024) as usize;
let mut scratch = CheckpointBytes::zeroed(scratch_length, work)?;
let mut file = {
let unit = work.start_unit()?;
let _wave = work.io_wave()?;
let file = OpenOptions::new()
.read(true)
.descriptor_kind(DescriptorKind::ImmutableCache)
.open_with_context(&binding.object_path, context)
.map_err(|error| CheckpointImmutableFileError::io(error, work))?;
let metadata = file
.metadata()
.map_err(|error| CheckpointImmutableFileError::io(error, work))?;
if !metadata.is_file() || metadata.len() != binding.reference.byte_length {
return Err(CheckpointImmutableFileError::Io(io::Error::new(
io::ErrorKind::InvalidData,
"immutable handle identity length mismatch",
)));
}
unit.finish();
file
};
let mut hasher = crate::immutable_object::identity_hasher(
binding.reference.kind,
binding.reference.format_version,
binding.reference.byte_length,
);
let mut read_bytes = 0_u64;
loop {
let read = {
let unit = work.start_unit()?;
let _wave = work.io_wave()?;
let read = file
.read(scratch.as_mut_slice())
.map_err(|error| CheckpointImmutableFileError::io(error, work))?;
unit.finish();
read
};
if read == 0 {
break;
}
let unit = work.start_unit()?;
read_bytes = read_bytes.checked_add(read as u64).ok_or_else(|| {
CheckpointImmutableFileError::Io(io::Error::other("immutable file length overflow"))
})?;
if read_bytes > binding.reference.byte_length {
return Err(CheckpointImmutableFileError::Io(io::Error::new(
io::ErrorKind::InvalidData,
"immutable handle grew during validation",
)));
}
hasher.update(&scratch[..read]);
unit.finish();
}
{
let unit = work.start_unit()?;
if read_bytes != binding.reference.byte_length
|| hasher.finish().sha256 != binding.reference.sha256
{
return Err(CheckpointImmutableFileError::Io(io::Error::new(
io::ErrorKind::InvalidData,
"immutable handle identity digest mismatch",
)));
}
unit.finish();
}
work.checkpoint()?;
drop(scratch);
let unit = work.start_unit()?;
let mut candidate = Some(file);
let file = {
let _opening = self.try_checkpoint_opening(work)?;
let mut handles = self
.handles
.lock()
.unwrap_or_else(|error| error.into_inner());
if let Some(handle) = handles.get_mut(&binding.reference) {
handle.last_used = self.next_tick();
handle.file.clone()
} else {
let file = Arc::new(candidate.take().expect("validated candidate is present"));
handles.insert(
binding.reference,
CachedHandle {
file: file.clone(),
last_used: self.next_tick(),
},
);
file
}
};
drop(candidate);
unit.finish();
work.checkpoint()?;
Ok(file)
}
}