(WIP) New object table model, TODO: update API calls to use it

This commit is contained in:
Alex 2020-07-08 16:46:47 +02:00
parent 86bf4dedac
commit a5fa2a136b

View file

@ -11,6 +11,8 @@ use garage_table::*;
use crate::version_table::*; use crate::version_table::*;
use model010::object_table as prev;
#[derive(PartialEq, Clone, Debug, Serialize, Deserialize)] #[derive(PartialEq, Clone, Debug, Serialize, Deserialize)]
pub struct Object { pub struct Object {
// Primary key // Primary key
@ -59,39 +61,59 @@ pub struct ObjectVersion {
pub uuid: UUID, pub uuid: UUID,
pub timestamp: u64, pub timestamp: u64,
pub mime_type: String,
pub size: u64,
pub state: ObjectVersionState, pub state: ObjectVersionState,
pub data: ObjectVersionData,
} }
#[derive(PartialEq, Clone, Copy, Debug, Serialize, Deserialize)] #[derive(PartialEq, Clone, Debug, Serialize, Deserialize)]
pub enum ObjectVersionState { pub enum ObjectVersionState {
Uploading, Uploading,
Complete, Complete(ObjectVersionData),
Aborted, Aborted,
} }
impl ObjectVersionState { impl ObjectVersionState {
fn max(self, other: Self) -> Self { fn merge(&mut self, other: &Self) {
use ObjectVersionState::*; use ObjectVersionState::*;
if self == Aborted || other == Aborted { match other {
Aborted Aborted => {
} else if self == Complete || other == Complete { *self = Aborted;
Complete }
} else { Complete(b) => {
Uploading match self {
} Aborted => {},
Complete(a) => {
a.merge(b);
}
Uploading => {
*self = Complete(b.clone());
}
}
}
Uploading => {}
}
} }
} }
#[derive(PartialEq, Clone, Debug, Serialize, Deserialize)] #[derive(PartialEq, Clone, Debug, Serialize, Deserialize)]
pub enum ObjectVersionData { pub enum ObjectVersionData {
Uploading,
DeleteMarker, DeleteMarker,
Inline(#[serde(with = "serde_bytes")] Vec<u8>), Inline(ObjectVersionMeta, #[serde(with = "serde_bytes")] Vec<u8>),
FirstBlock(Hash), FirstBlock(ObjectVersionMeta, Hash),
}
#[derive(PartialEq, Clone, Debug, Serialize, Deserialize)]
pub struct ObjectVersionMeta {
pub mime_type: String,
pub size: u64,
pub etag: String,
}
impl ObjectVersionData {
fn merge(&mut self, b: &Self) {
if *self != *b {
warn!("Inconsistent object version data: {:?} (local) vs {:?} (remote)", self, b);
}
}
} }
impl ObjectVersion { impl ObjectVersion {
@ -99,10 +121,17 @@ impl ObjectVersion {
(self.timestamp, self.uuid) (self.timestamp, self.uuid)
} }
pub fn is_complete(&self) -> bool { pub fn is_complete(&self) -> bool {
self.state == ObjectVersionState::Complete match self.state {
ObjectVersionState::Complete(_) => true,
_ => false,
}
} }
pub fn is_data(&self) -> bool { pub fn is_data(&self) -> bool {
self.state == ObjectVersionState::Complete && self.data != ObjectVersionData::DeleteMarker match self.state {
ObjectVersionState::Complete(ObjectVersionData::DeleteMarker) => false,
ObjectVersionState::Complete(_) => true,
_ => false,
}
} }
} }
@ -121,14 +150,7 @@ impl Entry<String, String> for Object {
.binary_search_by(|v| v.cmp_key().cmp(&other_v.cmp_key())) .binary_search_by(|v| v.cmp_key().cmp(&other_v.cmp_key()))
{ {
Ok(i) => { Ok(i) => {
let mut v = &mut self.versions[i]; self.versions[i].state.merge(&other_v.state);
if other_v.size > v.size {
v.size = other_v.size;
}
v.state = v.state.max(other_v.state);
if v.data == ObjectVersionData::Uploading {
v.data = other_v.data.clone();
}
} }
Err(i) => { Err(i) => {
self.versions.insert(i, other_v.clone()); self.versions.insert(i, other_v.clone());
@ -195,4 +217,47 @@ impl TableSchema for ObjectTable {
fn matches_filter(entry: &Self::E, _filter: &Self::Filter) -> bool { fn matches_filter(entry: &Self::E, _filter: &Self::Filter) -> bool {
entry.versions.iter().any(|v| v.is_data()) entry.versions.iter().any(|v| v.is_data())
} }
fn try_migrate(bytes: &[u8]) -> Option<Self::E> {
let old = match rmp_serde::decode::from_read_ref::<_, prev::Object>(bytes) {
Ok(x) => x,
Err(_) => return None,
};
let new_v = old.versions().iter()
.map(migrate_version)
.collect::<Vec<_>>();
let new = Object::new(old.bucket.clone(), old.key.clone(), new_v);
Some(new)
}
}
fn migrate_version(old: &prev::ObjectVersion) -> ObjectVersion {
let meta = ObjectVersionMeta{
size: old.size,
mime_type: old.mime_type.clone(),
etag: "".to_string(),
};
let state = match old.state {
prev::ObjectVersionState::Uploading => ObjectVersionState::Uploading,
prev::ObjectVersionState::Aborted => ObjectVersionState::Aborted,
prev::ObjectVersionState::Complete => {
match &old.data {
prev::ObjectVersionData::Uploading => ObjectVersionState::Uploading,
prev::ObjectVersionData::DeleteMarker => ObjectVersionState::Complete(ObjectVersionData::DeleteMarker),
prev::ObjectVersionData::Inline(x) => ObjectVersionState::Complete(ObjectVersionData::Inline(meta, x.clone())),
prev::ObjectVersionData::FirstBlock(h) => {
let mut hash = [0u8; 32];
hash.copy_from_slice(h.as_ref());
ObjectVersionState::Complete(ObjectVersionData::FirstBlock(meta, Hash::from(hash)))
}
}
}
};
let mut uuid = [0u8; 32];
uuid.copy_from_slice(old.uuid.as_ref());
ObjectVersion{
uuid: UUID::from(uuid),
timestamp: old.timestamp,
state,
}
} }