diff options
| -rw-r--r-- | src/model/object_table.rs | 119 |
1 files changed, 92 insertions, 27 deletions
diff --git a/src/model/object_table.rs b/src/model/object_table.rs index 01df70e6..2221c57e 100644 --- a/src/model/object_table.rs +++ b/src/model/object_table.rs @@ -11,6 +11,8 @@ use garage_table::*; use crate::version_table::*; +use model010::object_table as prev; + #[derive(PartialEq, Clone, Debug, Serialize, Deserialize)] pub struct Object { // Primary key @@ -59,39 +61,59 @@ pub struct ObjectVersion { pub uuid: UUID, pub timestamp: u64, - pub mime_type: String, - pub size: u64, pub state: ObjectVersionState, - - pub data: ObjectVersionData, } -#[derive(PartialEq, Clone, Copy, Debug, Serialize, Deserialize)] +#[derive(PartialEq, Clone, Debug, Serialize, Deserialize)] pub enum ObjectVersionState { Uploading, - Complete, + Complete(ObjectVersionData), Aborted, } impl ObjectVersionState { - fn max(self, other: Self) -> Self { + fn merge(&mut self, other: &Self) { use ObjectVersionState::*; - if self == Aborted || other == Aborted { - Aborted - } else if self == Complete || other == Complete { - Complete - } else { - Uploading - } + match other { + Aborted => { + *self = Aborted; + } + Complete(b) => { + match self { + Aborted => {}, + Complete(a) => { + a.merge(b); + } + Uploading => { + *self = Complete(b.clone()); + } + } + } + Uploading => {} + } } } #[derive(PartialEq, Clone, Debug, Serialize, Deserialize)] pub enum ObjectVersionData { - Uploading, DeleteMarker, - Inline(#[serde(with = "serde_bytes")] Vec<u8>), - FirstBlock(Hash), + Inline(ObjectVersionMeta, #[serde(with = "serde_bytes")] Vec<u8>), + 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 { @@ -99,10 +121,17 @@ impl ObjectVersion { (self.timestamp, self.uuid) } pub fn is_complete(&self) -> bool { - self.state == ObjectVersionState::Complete + match self.state { + ObjectVersionState::Complete(_) => true, + _ => false, + } } 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())) { Ok(i) => { - let mut v = &mut self.versions[i]; - 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(); - } + self.versions[i].state.merge(&other_v.state); } Err(i) => { 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 { 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, + } } |