pub mod constants;
use anyhow::{bail, Context, Result};
use constants::*;
use std::collections::{HashMap, HashSet};
use crate::orchestrator::BLOB_SIZE_C_BIT;
use std::time::{SystemTime, UNIX_EPOCH};
use crate::piv::PivBackend;
#[derive(Debug, Clone)]
pub struct Object {
pub(crate) index: u8,
pub(crate) object_size: usize,
pub(crate) yblob_magic: u32,
pub(crate) object_count: u8,
pub(crate) store_key_slot: u8,
pub(crate) age: u32,
pub(crate) chunk_pos: u8,
pub(crate) next_chunk: u8,
pub blob_mtime: u32,
pub blob_size: u32,
pub blob_key_slot: u8,
pub blob_plain_size: u32,
pub is_compressed: bool,
pub blob_name: String,
pub(crate) payload: Vec<u8>,
pub(crate) dirty: bool,
}
impl Object {
pub fn from_bytes(index: u8, data: &[u8]) -> Result<Self> {
let object_size = data.len();
if !(OBJECT_MIN_SIZE..=OBJECT_MAX_SIZE).contains(&object_size) {
bail!(
"object {index}: invalid size {object_size} \
(expected {OBJECT_MIN_SIZE}..={OBJECT_MAX_SIZE})"
);
}
let magic = read_u32_le(data, MAGIC_O)?;
if magic != YBLOB_MAGIC {
bail!("object {index}: bad magic 0x{magic:08x} (expected 0x{YBLOB_MAGIC:08x})");
}
let object_count = data[OBJECT_COUNT_O];
let store_key_slot = data[STORE_KEY_SLOT_O];
let age = read_u24_le(data, OBJECT_AGE_O)?;
if age == 0 {
return Ok(Self {
index,
object_size,
yblob_magic: magic,
object_count,
store_key_slot,
age: 0,
chunk_pos: 0,
next_chunk: 0,
blob_mtime: 0,
blob_size: 0,
blob_key_slot: 0,
blob_plain_size: 0,
is_compressed: false,
blob_name: String::new(),
payload: Vec::new(),
dirty: false,
});
}
let chunk_pos = data[CHUNK_POS_O];
let next_chunk = data[NEXT_CHUNK_O];
let (
blob_mtime,
blob_size,
blob_key_slot,
blob_plain_size,
is_compressed,
blob_name,
payload_start,
) = if chunk_pos == 0 {
let mtime = read_u32_le(data, BLOB_MTIME_O)?;
let bsize = read_u24_le(data, BLOB_SIZE_O)?;
let bkslot = data[BLOB_KEY_SLOT_O];
let raw_plain = read_u24_le(data, BLOB_PLAIN_SIZE_O)?;
let compressed = raw_plain & BLOB_SIZE_C_BIT as u32 != 0;
let plain = raw_plain & !(BLOB_SIZE_C_BIT as u32);
let nlen = data[BLOB_NAME_LEN_O] as usize;
if BLOB_NAME_O + nlen > object_size {
bail!("object {index}: name length {nlen} overflows object");
}
let name = std::str::from_utf8(&data[BLOB_NAME_O..BLOB_NAME_O + nlen])
.with_context(|| format!("object {index}: blob name is not valid UTF-8"))?
.to_owned();
(
mtime,
bsize,
bkslot,
plain,
compressed,
name,
BLOB_NAME_O + nlen,
)
} else {
(0, 0, 0, 0, false, String::new(), CONTINUATION_PAYLOAD_O)
};
let raw_payload = &data[payload_start..];
let payload = raw_payload.to_vec();
Ok(Self {
index,
object_size,
yblob_magic: magic,
object_count,
store_key_slot,
age,
chunk_pos,
next_chunk,
blob_mtime,
blob_size,
blob_key_slot,
blob_plain_size,
is_compressed,
blob_name,
payload,
dirty: false,
})
}
pub fn to_bytes(&self) -> Vec<u8> {
if self.age == 0 {
let mut buf = vec![0u8; OBJECT_MIN_SIZE];
write_u32_le(&mut buf, MAGIC_O, self.yblob_magic);
buf[OBJECT_COUNT_O] = self.object_count;
buf[STORE_KEY_SLOT_O] = self.store_key_slot;
return buf;
}
if self.chunk_pos == 0 {
let name_bytes = self.blob_name.as_bytes();
let total = (BLOB_NAME_O + name_bytes.len() + self.payload.len())
.clamp(OBJECT_MIN_SIZE, OBJECT_MAX_SIZE);
let mut buf = vec![0u8; total];
write_u32_le(&mut buf, MAGIC_O, self.yblob_magic);
buf[OBJECT_COUNT_O] = self.object_count;
buf[STORE_KEY_SLOT_O] = self.store_key_slot;
write_u24_le(&mut buf, OBJECT_AGE_O, self.age);
buf[CHUNK_POS_O] = self.chunk_pos;
buf[NEXT_CHUNK_O] = self.next_chunk;
write_u32_le(&mut buf, BLOB_MTIME_O, self.blob_mtime);
write_u24_le(&mut buf, BLOB_SIZE_O, self.blob_size);
buf[BLOB_KEY_SLOT_O] = self.blob_key_slot;
write_u24_le(
&mut buf,
BLOB_PLAIN_SIZE_O,
self.blob_plain_size | self.c_bit(),
);
buf[BLOB_NAME_LEN_O] = name_bytes.len() as u8;
buf[BLOB_NAME_O..BLOB_NAME_O + name_bytes.len()].copy_from_slice(name_bytes);
let payload_start = BLOB_NAME_O + name_bytes.len();
let payload_end = (payload_start + self.payload.len()).min(total);
buf[payload_start..payload_end]
.copy_from_slice(&self.payload[..payload_end - payload_start]);
buf
} else {
let total = (CONTINUATION_PAYLOAD_O + self.payload.len())
.clamp(OBJECT_MIN_SIZE, OBJECT_MAX_SIZE);
let mut buf = vec![0u8; total];
write_u32_le(&mut buf, MAGIC_O, self.yblob_magic);
buf[OBJECT_COUNT_O] = self.object_count;
buf[STORE_KEY_SLOT_O] = self.store_key_slot;
write_u24_le(&mut buf, OBJECT_AGE_O, self.age);
buf[CHUNK_POS_O] = self.chunk_pos;
buf[NEXT_CHUNK_O] = self.next_chunk;
let payload_end = (CONTINUATION_PAYLOAD_O + self.payload.len()).min(total);
buf[CONTINUATION_PAYLOAD_O..payload_end]
.copy_from_slice(&self.payload[..payload_end - CONTINUATION_PAYLOAD_O]);
buf
}
}
pub fn reset(&mut self) {
let index = self.index;
let object_count = self.object_count;
let store_key_slot = self.store_key_slot;
*self = Self {
index,
object_size: OBJECT_MIN_SIZE, yblob_magic: crate::store::constants::YBLOB_MAGIC,
object_count,
store_key_slot,
age: 0,
chunk_pos: 0,
next_chunk: 0,
blob_mtime: 0,
blob_size: 0,
blob_key_slot: 0,
blob_plain_size: 0,
is_compressed: false,
blob_name: String::new(),
payload: Vec::new(),
dirty: true,
};
}
pub fn head_payload_capacity(_object_size: usize, name_len: usize) -> usize {
OBJECT_MAX_SIZE.saturating_sub(BLOB_NAME_O + name_len)
}
pub fn continuation_payload_capacity(_object_size: usize) -> usize {
OBJECT_MAX_SIZE.saturating_sub(CONTINUATION_PAYLOAD_O)
}
fn c_bit(&self) -> u32 {
BLOB_SIZE_C_BIT as u32 * self.is_compressed as u32
}
pub fn is_empty(&self) -> bool {
self.age == 0
}
pub fn is_head(&self) -> bool {
self.age != 0 && self.chunk_pos == 0
}
pub fn is_encrypted(&self) -> bool {
self.blob_key_slot != 0
}
pub fn set_payload(&mut self, payload: Vec<u8>) {
self.payload = payload;
}
pub fn index(&self) -> u8 {
self.index
}
pub fn age(&self) -> u32 {
self.age
}
pub fn chunk_pos(&self) -> u8 {
self.chunk_pos
}
pub fn next_chunk(&self) -> u8 {
self.next_chunk
}
pub fn payload(&self) -> &[u8] {
&self.payload
}
pub fn payload_len(&self) -> usize {
self.payload.len()
}
pub fn object_size(&self) -> usize {
self.object_size
}
}
pub struct ObjectParams {
pub index: u8,
pub age: u32,
pub chunk_pos: u8,
pub next_chunk: u8,
}
pub struct Store {
pub reader: String,
pub object_size: usize,
pub object_count: u8,
pub store_key_slot: u8,
pub objects: Vec<Object>,
pub store_age: u32,
}
impl Store {
pub fn from_device(reader: &str, piv: &dyn PivBackend) -> Result<Self> {
let first_id = OBJECT_ID_ZERO;
let raw = piv
.read_object(reader, first_id)
.with_context(|| format!("reading object 0x{first_id:06x}"))?;
let first =
Object::from_bytes(0, &raw).with_context(|| "parsing object 0 (store header)")?;
let object_count = first.object_count;
let object_size = raw.len();
let store_key_slot = first.store_key_slot;
let mut objects = vec![first];
for i in 1..object_count {
let id = OBJECT_ID_ZERO + i as u32;
let raw = piv
.read_object(reader, id)
.with_context(|| format!("reading object 0x{id:06x}"))?;
let obj = Object::from_bytes(i, &raw).with_context(|| format!("parsing object {i}"))?;
objects.push(obj);
}
let store_age = objects.iter().map(|o| o.age).max().unwrap_or(0);
Ok(Self {
reader: reader.to_owned(),
object_size,
object_count,
store_key_slot,
objects,
store_age,
})
}
pub fn format(
reader: &str,
piv: &dyn PivBackend,
object_count: u8,
store_key_slot: u8,
management_key: &str,
) -> Result<Self> {
let mut store = Self {
reader: reader.to_owned(),
object_size: OBJECT_MIN_SIZE,
object_count,
store_key_slot,
objects: Vec::with_capacity(object_count as usize),
store_age: 0,
};
for i in 0..object_count {
store.objects.push(store.make_object(ObjectParams {
index: i,
age: 0,
chunk_pos: 0,
next_chunk: 0,
}));
}
store.write_dirty(piv, management_key, false)?;
Ok(store)
}
pub fn sanitize(&mut self) {
let mut seen: HashMap<String, (u8, u32)> = HashMap::new();
let mut to_reset: Vec<u8> = Vec::new();
for obj in self.objects.iter().filter(|o| o.is_head()) {
if let Some(&(prev_idx, prev_age)) = seen.get(&obj.blob_name) {
if obj.age > prev_age {
to_reset.push(prev_idx);
seen.insert(obj.blob_name.clone(), (obj.index, obj.age));
} else {
to_reset.push(obj.index);
}
} else {
seen.insert(obj.blob_name.clone(), (obj.index, obj.age));
}
}
let mut reachable: HashSet<u8> = HashSet::new();
for (head_idx, _) in seen.values() {
let mut idx = *head_idx;
loop {
reachable.insert(idx);
let next = self.objects[idx as usize].next_chunk;
if next == idx {
break;
}
idx = next;
}
}
for obj in self.objects.iter().filter(|o| !o.is_empty()) {
if !reachable.contains(&obj.index) {
to_reset.push(obj.index);
}
}
for idx in to_reset {
self.objects[idx as usize].reset();
}
}
pub fn sync(&mut self, piv: &dyn PivBackend, management_key: &str) -> Result<()> {
self.write_dirty(piv, management_key, true)
}
fn write_dirty(
&mut self,
piv: &dyn PivBackend,
management_key: &str,
note_partial: bool,
) -> Result<()> {
use indicatif::{ProgressBar, ProgressStyle};
let dirty: Vec<u8> = self
.objects
.iter()
.filter(|o| o.dirty)
.map(|o| o.index)
.collect();
let pb = ProgressBar::new(dirty.len() as u64);
pb.set_style(
ProgressStyle::with_template("Writing objects: [{bar:30}] {pos}/{len}")
.unwrap()
.progress_chars("=>-"),
);
for (written, idx) in dirty.iter().enumerate() {
let obj = &mut self.objects[*idx as usize];
let id = OBJECT_ID_ZERO + obj.index as u32;
let data = obj.to_bytes();
let result = piv.write_object(&self.reader, id, &data, management_key);
if note_partial && written > 0 {
result.context("the store may be partially updated; run `yb fsck`")?;
} else {
result?;
}
obj.dirty = false;
pb.inc(1);
}
pb.finish_and_clear();
Ok(())
}
pub fn alloc_free(&self) -> Option<u8> {
self.objects.iter().find(|o| o.is_empty()).map(|o| o.index)
}
pub fn free_count(&self) -> usize {
self.objects.iter().filter(|o| o.is_empty()).count()
}
pub fn find_head(&self, name: &str) -> Option<&Object> {
self.objects
.iter()
.find(|o| o.is_head() && o.blob_name == name)
}
pub fn chunk_chain(&self, head_index: u8) -> Vec<u8> {
let mut chain = vec![head_index];
let mut seen: HashSet<u8> = HashSet::from([head_index]);
let mut idx = head_index;
loop {
let next = self.objects[idx as usize].next_chunk;
if next == idx {
break;
}
if seen.contains(&next) {
break;
}
seen.insert(next);
chain.push(next);
idx = next;
}
chain
}
pub fn now_unix() -> u32 {
SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|d| d.as_secs() as u32)
.unwrap_or(0)
}
pub fn next_age(&mut self) -> u32 {
self.store_age += 1;
self.store_age
}
pub fn make_object(&self, p: ObjectParams) -> Object {
Object {
index: p.index,
object_size: self.object_size,
yblob_magic: crate::store::constants::YBLOB_MAGIC,
object_count: self.object_count,
store_key_slot: self.store_key_slot,
age: p.age,
chunk_pos: p.chunk_pos,
next_chunk: p.next_chunk,
blob_mtime: 0,
blob_size: 0,
blob_key_slot: 0,
blob_plain_size: 0,
is_compressed: false,
blob_name: String::new(),
payload: Vec::new(),
dirty: true,
}
}
}
pub(crate) fn read_u32_le(buf: &[u8], offset: usize) -> Result<u32> {
let end = offset + 4;
if end > buf.len() {
bail!(
"read_u32_le: offset {offset}+4 out of bounds (buf len {})",
buf.len()
);
}
Ok(u32::from_le_bytes(buf[offset..end].try_into().unwrap()))
}
pub(crate) fn read_u24_le(buf: &[u8], offset: usize) -> Result<u32> {
let end = offset + 3;
if end > buf.len() {
bail!(
"read_u24_le: offset {offset}+3 out of bounds (buf len {})",
buf.len()
);
}
let b = &buf[offset..end];
Ok(b[0] as u32 | ((b[1] as u32) << 8) | ((b[2] as u32) << 16))
}
pub(crate) fn write_u32_le(buf: &mut [u8], offset: usize, v: u32) {
buf[offset..offset + 4].copy_from_slice(&v.to_le_bytes());
}
pub(crate) fn write_u24_le(buf: &mut [u8], offset: usize, v: u32) {
buf[offset] = (v & 0xff) as u8;
buf[offset + 1] = ((v >> 8) & 0xff) as u8;
buf[offset + 2] = ((v >> 16) & 0xff) as u8;
}
#[cfg(test)]
mod tests;