Deuxfleurs/garage
48
56
Fork 73
Code Issues 100 Pull requests 13 Releases 30 Wiki Activity

Ensure increasing version timestamps when writing new object versions #543

Merged
lx merged 6 commits from increasing-timestamps into main 2023-10-24 10:07:17 +00:00
Conversation 0 Commits 6 Files changed 3 +18 -6
1 changed files with 18 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit 45b0453d0f - Show all commits

24
src/api/s3/put.rs
View file

@ -3,6 +3,7 @@ use std::sync::Arc;
use base64::prelude::*; use base64::prelude::*;
use futures::prelude::*; use futures::prelude::*;
use futures::try_join;
use hyper::body::{Body, Bytes}; use hyper::body::{Body, Bytes};
use hyper::header::{HeaderMap, HeaderValue}; use hyper::header::{HeaderMap, HeaderValue};
use hyper::{Request, Response}; use hyper::{Request, Response};
@ -35,7 +36,7 @@ pub async fn handle_put(
garage: Arc<Garage>, garage: Arc<Garage>,
req: Request<Body>, req: Request<Body>,
bucket: &Bucket, bucket: &Bucket,
key: &str, key: &String,
content_sha256: Option<Hash>, content_sha256: Option<Hash>,
) -> Result<Response<Body>, Error> { ) -> Result<Response<Body>, Error> {
// Retrieve interesting headers from request // Retrieve interesting headers from request
@ -68,16 +69,27 @@ pub(crate) async fn save_stream<S: Stream<Item = Result<Bytes, Error>> + Unpin>(
headers: ObjectVersionHeaders, headers: ObjectVersionHeaders,
body: S, body: S,
bucket: &Bucket, bucket: &Bucket,
key: &str, key: &String,
content_md5: Option<String>, content_md5: Option<String>,
content_sha256: Option<FixedBytes32>, content_sha256: Option<FixedBytes32>,
) -> Result<(Uuid, String), Error> { ) -> Result<(Uuid, String), Error> {
let mut chunker = StreamChunker::new(body, garage.config.block_size);
let (first_block_opt, existing_object) = try_join!(
chunker.next(),
garage
.object_table
.get(&bucket.id, key)
.map_err(Error::from),
)?;
let first_block = first_block_opt.unwrap_or_default();
// Generate identity of new version // Generate identity of new version
let version_uuid = gen_uuid(); let version_uuid = gen_uuid();
let version_timestamp = now_msec(); let version_timestamp = existing_object
.and_then(|obj| obj.versions().iter().map(|v| v.timestamp).max())
let mut chunker = StreamChunker::new(body, garage.config.block_size); .map(|t| std::cmp::max(t + 1, now_msec()))
let first_block = chunker.next().await?.unwrap_or_default(); .unwrap_or_else(now_msec);
// If body is small enough, store it directly in the object table // If body is small enough, store it directly in the object table
// as "inline data". We can then return immediately. // as "inline data". We can then return immediately.