From 452a67a343253bb5945c1ca4bcf0391259501a40 Mon Sep 17 00:00:00 2001 From: Mattias Jansson Date: Mon, 3 Aug 2026 08:49:46 +0200 Subject: [PATCH] lore-aws: Deduplicate S3 payloads across partitions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Writing a fragment uploaded its payload whenever the exact (partition, context, hash) row was absent, even when the payload for that hash was already stored. The S3 object key is the bare content hash, so that upload replaced content other partitions reference, and the object and the metadata describing it were written as two independent, unordered operations. That permitted two corruption cases. Two writers storing the same content in different representations on different partitions at once are written to S3 and DynamoDB independently, so the writer that wins S3 need not be the writer that wins DynamoDB. Every write succeeds and the stored blob ends up described by the other writer's fragment. More likely, a second partition storing content that is already stored has no exact row, so it uploads and replaces the object other partitions reference. Losing the metadata write that follows — DynamoDB throttling, or termination between the two writes, which the code already anticipated — leaves the blob and the published metadata no longer describing each other. The affected partition cannot repair that: its own re-put matches on the association and returns without writing. A put now probes the association and the payload's durability together: two strongly consistent single-item reads against different tables, one round trip. The metadata table is keyed by hash alone, so the probe does not grow with the number of partitions referencing the content, and S3 is not consulted. Content that is already stored is referenced rather than uploaded again, which removes the second case entirely — there is no upload to replace the blob and no metadata write to lose. Deduplication adopts the stored fragment, never the incoming one, and still requires the payload, so presenting bytes remains proof of possession. Content that is genuinely absent is uploaded conditionally on the key being free. That makes the object write-once, so exactly one writer can create the bytes for a hash and S3 itself arbitrates between competing writers with no external lock. Metadata is published only after that writer's own upload created the object, so the S3 winner and the metadata winner are the same writer by construction, which removes the first case. A key that is already taken means the object belongs to a writer that has not published. Its age decides what to do: younger than the abandonment threshold the caller backs off, older it is reclaimed conditionally on the entity tag just observed, making reclaim single-winner. Only a writer that displaced bytes may overwrite another's metadata after a rejected publish; a writer whose upload was conditional stands down, since its bytes may since have been reclaimed. Order obliteration so associations need no transaction Obliteration marks the metadata row, removes the reference it was asked to remove, waits for puts that read the row beforehand to finish writing, and counts what remains — clearing the mark if anything does, and otherwise deleting the payload and leaving a tombstone. A put that reads the mark backs off, so no reference can appear while the payload is being removed, and the mark precedes the removal so the partition being obliterated cannot write its own reference back. An association is therefore written plainly rather than inside a transaction conditioned on the metadata row, and no new DynamoDB permission is required. Association reads are strongly consistent, so the wait is not for consistency: it is for in-flight writes to land, and is sized above the DynamoDB request timeout. Removing the reference is what obliteration has to do; the payload and the metadata are cleanup that only applies once nothing references the content. Ordering it that way also stops sub-fragments being obliterated before the parent is known to be going away. An obliteration finding the mark already set removes its own reference and leaves the payload and the metadata to whichever obliteration holds the mark, rather than reporting success without doing anything. Only a tombstone refuses a read: a row that is merely marked still has its payload, so refusing reads there would hide live content from every partition holding it. Published metadata is authoritative and deliberately not verified on the read or deduplication paths, since checking would cost a request per put and give back what deduplication buys. A read that finds published metadata with no object behind it is counted and logged rather than surfacing as an ordinary not-found, because that state is an operational failure the store cannot repair on its own. The fragment metadata table is unchanged: no new attributes, no migration. Requires an S3 endpoint honouring If-None-Match, If-Match and Last-Modified. Both corruption cases are reproduced as failing tests on main in repro/fragment-dedup-corruption, and their counterparts here run the same scenarios with the same fault injected. Signed-off-by: Mattias Jansson --- lore-aws/src/s3.rs | 92 + lore-aws/src/store/immutable_store.rs | 3062 ++++++++++++++++++++++--- 2 files changed, 2850 insertions(+), 304 deletions(-) diff --git a/lore-aws/src/s3.rs b/lore-aws/src/s3.rs index cfa86190..884d0730 100644 --- a/lore-aws/src/s3.rs +++ b/lore-aws/src/s3.rs @@ -34,6 +34,22 @@ use tracing::warn; use crate::aws_error::AwsError; use crate::observe_aws_operation_callback; +/// HTTP status S3 returns when a conditional write's precondition is not met. +const PRECONDITION_FAILED: u16 = 412; + +/// Whether an S3 error is the precondition failure a conditional write returns when the key is +/// already present, rather than a genuine failure. +/// +/// Used to tell "someone else's object is already there" apart from "the upload broke", which +/// are handled very differently: the first is a race to recover from, the second is an error. +pub fn is_precondition_failed(error: &AwsError>) -> bool { + matches!( + error, + AwsError::AwsSdkError(SdkError::ServiceError(err)) + if err.raw().status().as_u16() == PRECONDITION_FAILED + ) +} + #[derive(Clone)] struct S3InstrumentProvider; @@ -218,6 +234,82 @@ impl S3Impl { .map_err(AwsError::AwsSdkError) } + /// Store an object only if the key is absent. + /// + /// Makes the object write-once, so exactly one writer can ever create the bytes under a key. + /// That writer is the only one whose put succeeds, which makes S3 itself the mutual exclusion + /// between competing writers — no external lock required. A successful put is therefore proof + /// the stored bytes are this writer's, and the only point at which it is safe to publish + /// metadata describing them. + /// + /// A key that already exists fails with HTTP 412; see [`is_precondition_failed`]. + #[tracing::instrument(name = "S3Impl::put_object_if_absent", skip_all)] + pub async fn put_object_if_absent( + &self, + bucket: &str, + key: &str, + body: T, + ) -> Result>> + where + T: Into> + 'static, + { + self.client + .put_object() + .bucket(bucket) + .key(key) + .if_none_match("*") + .body(ByteStream::from(Into::>::into(body))) + .send() + .observe( + self.instruments.operation_latency_histogram.clone(), + self.instruments + .instrument_provider + .get_labels_for_operation_context("put_object_if_absent"), + observe_aws_operation_callback(self.slow_operation_duration), + ) + .await + .output + .map_err(AwsError::AwsSdkError) + } + + /// Replace an object only if it is still exactly the one identified by `etag`. + /// + /// Used to reclaim an object abandoned by a writer that never published it. Conditioning on + /// the entity tag observed a moment earlier makes reclaiming single-winner: the first reclaim + /// changes the tag, so a second writer racing for the same object is rejected rather than + /// quietly replacing bytes the first one is about to publish metadata for. + /// + /// A tag that no longer matches fails with HTTP 412; see [`is_precondition_failed`]. + #[tracing::instrument(name = "S3Impl::put_object_if_match", skip_all)] + pub async fn put_object_if_match( + &self, + bucket: &str, + key: &str, + body: T, + etag: &str, + ) -> Result>> + where + T: Into> + 'static, + { + self.client + .put_object() + .bucket(bucket) + .key(key) + .if_match(etag) + .body(ByteStream::from(Into::>::into(body))) + .send() + .observe( + self.instruments.operation_latency_histogram.clone(), + self.instruments + .instrument_provider + .get_labels_for_operation_context("put_object_if_match"), + observe_aws_operation_callback(self.slow_operation_duration), + ) + .await + .output + .map_err(AwsError::AwsSdkError) + } + pub async fn list_versions( &self, bucket: &str, diff --git a/lore-aws/src/store/immutable_store.rs b/lore-aws/src/store/immutable_store.rs index 333b6138..6a7fb81f 100644 --- a/lore-aws/src/store/immutable_store.rs +++ b/lore-aws/src/store/immutable_store.rs @@ -6,6 +6,9 @@ use std::fmt::Formatter; use std::string::ToString; use std::sync::Arc; use std::sync::LazyLock; +use std::time::Duration; +use std::time::SystemTime; +use std::time::UNIX_EPOCH; use async_trait::async_trait; use aws_sdk_dynamodb::error::SdkError; @@ -14,6 +17,7 @@ use aws_sdk_dynamodb::primitives::Blob; use aws_sdk_dynamodb::types::AttributeValue; use aws_sdk_dynamodb::types::Select; use aws_sdk_s3::operation::get_object::GetObjectError; +use aws_smithy_types::DateTime; use bytes::Bytes; use bytes::BytesMut; use lore_base::error::AddressNotFound; @@ -44,6 +48,7 @@ use lore_telemetry::timer::TimedResult; use lore_telemetry::tracing::fields::ADDRESS; use lore_telemetry::tracing::fields::REPOSITORY_ID; use opentelemetry::KeyValue; +use opentelemetry::metrics::Counter; use opentelemetry::metrics::Histogram; use serde::Deserialize; use serde::Serialize; @@ -104,6 +109,67 @@ impl FragmentsEntry { } } +/// Lower bound on how long to wait for an in-flight upload to publish before treating the object +/// it left behind as abandoned. +const MIN_ABANDONED_GRACE_MILLIS: u64 = 100; + +/// Lower bound on the obliteration drain, regardless of how the `DynamoDB` timeout is configured. +const MIN_OBLITERATION_DRAIN_MILLIS: u64 = 100; + +fn now_millis() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |since| since.as_millis() as u64) +} + +/// Whether an object stored at `last_modified` has gone unpublished long enough that the writer +/// which stored it must be gone rather than still finishing. +/// +/// Deliberately one-sided: without a usable timestamp, or with one that looks like the future, +/// the object is treated as live. Concluding too early is not a correctness problem — reclaiming +/// is single-winner and only a reclaimer may overwrite metadata — but it does waste another +/// writer's upload, so the doubt is resolved in their favour. +/// +/// The age is measured against the local clock, so the threshold wants to stay comfortably above +/// any skew between this host and S3. +fn is_abandoned(last_modified: Option<&DateTime>, threshold_millis: u64) -> bool { + let Some(stored_at_millis) = last_modified.and_then(datetime_millis) else { + // An object that cannot be dated can never be judged abandoned, so it will never be + // reclaimed and every put for this hash backs off indefinitely. Real S3 always reports + // this; an implementation that does not turns a recoverable orphan into an unwritable + // hash, which is worth saying out loud rather than leaving to look like contention. + error!( + "Stored object reports no usable last-modified time, so it can never be reclaimed \ + if its writer abandoned it" + ); + + return false; + }; + + let now = now_millis(); + + if stored_at_millis > now { + warn!( + ahead_millis = stored_at_millis - now, + "Stored object is dated in the future; check this host's clock against S3, as \ + reclaiming abandoned uploads depends on the two agreeing" + ); + + return false; + } + + now - stored_at_millis >= threshold_millis +} + +/// Milliseconds since the epoch, or `None` for a time that predates it. +/// +/// Keeps sub-second precision, so a threshold below a second means what it says. +fn datetime_millis(time: &DateTime) -> Option { + let seconds = u64::try_from(time.secs()).ok()?; + + Some(seconds.saturating_mul(1_000) + u64::from(time.subsec_nanos()) / 1_000_000) +} + #[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] struct FragmentMetadataEntry { hash: Hash, @@ -203,6 +269,15 @@ pub struct AwsImmutableStoreSettings { pub force_write: bool, #[serde(default = "default_submission_limit")] pub batch_exist_submission_limit: usize, + /// How long to keep waiting for another writer to publish an object it has already uploaded + /// before treating that object as abandoned and recovering it. Defaults to a multiple of the + /// S3 request timeout, so a merely slow writer is given time to finish rather than having its + /// upload adopted. + pub abandoned_upload_grace_millis: Option, + /// How long an obliteration waits, after marking a row, for puts that read it beforehand to + /// finish writing their association. Sized above the `DynamoDB` request timeout so such a write + /// has either landed or failed by the time the references are counted again. + pub obliteration_drain_millis: Option, } impl AwsImmutableStoreSettings { @@ -216,13 +291,118 @@ impl AwsImmutableStoreSettings { dynamodb, force_write, batch_exist_submission_limit: default_submission_limit(), + abandoned_upload_grace_millis: None, + obliteration_drain_millis: None, } } + + /// Resolve how long a writer waits for someone else's upload to be published. Sized above the + /// S3 request timeout so a live writer has finished by the time its object is treated as + /// abandoned; recovery is correct either way, this only avoids doing it needlessly. + fn abandoned_grace_millis(&self) -> u64 { + self.abandoned_upload_grace_millis + .unwrap_or_else(|| self.s3.timeout_millis.saturating_mul(4)) + // `max` raises anything below the floor up to it. + .max(MIN_ABANDONED_GRACE_MILLIS) + } + + /// Resolve the obliteration drain. Obliteration is rare and not latency sensitive, so this is + /// sized generously: an association write that had already begun must have completed or timed + /// out before the references are counted again. + fn obliteration_drain_millis(&self) -> u64 { + self.obliteration_drain_millis + .unwrap_or_else(|| self.dynamodb.timeout_millis.saturating_mul(4)) + // `max` raises anything below the floor up to it. + .max(MIN_OBLITERATION_DRAIN_MILLIS) + } } +/// Counts payloads whose metadata says they are stored but whose object is not in S3. +/// +/// Should always be zero. A non-zero value means content has been lost underneath the store — +/// see [`AwsImmutableStore`] for why nothing repairs it automatically. +pub const METRICS_MISSING_PAYLOAD_METRIC_NAME: &str = "store.immutable.missing_payload"; + pub const FRAGMENTS_DYNAMO_PARTITION_KEY_ATTRIBUTE: &str = "hash"; pub const FRAGMENTS_DYNAMO_SORT_KEY_ATTRIBUTE: &str = "repository_context"; +/// How many times `put` re-probes after losing a race. Each retry is one extra metadata read; +/// the bound stops a contended hash from spinning here instead of returning to the caller, which +/// can retry with its own backoff. +const PUT_MAX_ATTEMPTS: usize = 3; + +/// What a `put` should do, given what its probe observed. +#[derive(Debug, PartialEq)] +enum PutAction { + /// The exact association already exists against durable content; nothing to write. + Done, + /// The payload is already durable, in this or another partition. Record the association and + /// skip the upload. + Deduplicate(Fragment), + /// Store the bytes and publish them, with the publish conditioned on this. + Upload(MetadataWriteCondition), + /// The caller did not supply the bytes and nothing here entitles it to skip them. + PayloadRequired, + /// The payload is marked for obliteration. The mark is transient, so this is a back-off + /// rather than a failure. + Obliterating, + /// Different content is already stored under this hash. + Collision, +} + +/// Decide what a put should do from the probed association and metadata state. +/// +/// Deduplication hinges on [`MetadataState::Committed`] meaning the payload is durable in S3: +/// rows are only committed after their upload succeeded, so a committed row lets this writer +/// record a reference instead of re-uploading bytes the server already holds. Crucially the +/// *stored* fragment is what gets referenced, never the incoming one — the S3 object is whatever +/// the original writer stored, so adopting an incoming fragment that described a different +/// representation would leave the metadata contradicting the bytes. +fn decide_put( + fragment: Fragment, + associated: bool, + stored: Option, + force_write: bool, + has_payload: bool, +) -> PutAction { + match stored { + // Nothing stored, so this writer has to supply the bytes. Note this is also how a lost + // metadata row heals: the association may well exist already, but without metadata the + // payload is not readable, so it is stored and published again. + None => PutAction::Upload(MetadataWriteCondition::Absent), + + Some(stored) => { + // Checked ahead of `force_write`, because an obliteration in progress holds a lock on + // this row. Overwriting it would release that lock underneath the obliteration and + // resurrect content that is being deleted. + if stored.flags & FragmentFlags::PayloadObliterating + == FragmentFlags::PayloadObliterating + { + PutAction::Obliterating + } else if force_write { + PutAction::Upload(MetadataWriteCondition::Unchanged(stored)) + } else if stored.flags & FragmentFlags::PayloadObliterated + == FragmentFlags::PayloadObliterated + { + // A tombstone: the payload was deleted, so the bytes must be stored again. + PutAction::Upload(MetadataWriteCondition::Unchanged(stored)) + } else if fragment.size_content != stored.size_content { + PutAction::Collision + } else if associated { + PutAction::Done + } else if has_payload { + // Attaching content the caller does not already reference means presenting the + // bytes for it. The upload is what deduplication skips, not that requirement: + // a hash on its own is not evidence the caller holds the content, and treating + // it as such would let one be attached to a partition that never had it. + PutAction::Deduplicate(stored) + } else { + PutAction::PayloadRequired + } + } + } +} + #[derive(Debug, Clone, PartialEq)] enum FragmentsQuery { Repository(Hash, Context), @@ -295,32 +475,80 @@ impl DynamoDbQuery for FragmentsQuery { } } +/// Condition parts asserting that a metadata row still describes exactly `expected`. +fn committed_metadata_condition_parts(expected: Fragment) -> ConditionParts { + ConditionParts { + condition_expression: + "#flags = :flags AND #size_payload = :size_payload AND #size_content = :size_content" + .to_string(), + expression_names: HashMap::from([ + ("#flags".to_string(), "flags".to_string()), + ("#size_payload".to_string(), "size_payload".to_string()), + ("#size_content".to_string(), "size_content".to_string()), + ]), + expression_values: HashMap::from([ + ( + ":flags".to_string(), + AttributeValue::N(expected.flags.to_string()), + ), + ( + ":size_payload".to_string(), + AttributeValue::N(expected.size_payload.to_string()), + ), + ( + ":size_content".to_string(), + AttributeValue::N(expected.size_content.to_string()), + ), + ]), + } +} + +/// Whether a conditional put failed its condition, as opposed to failing outright. Callers treat +/// this as "another writer won the race", not as an error. +fn is_conditional_check_failed(error: &AwsError>) -> bool { + let AwsError::AwsSdkError(sdk_error) = error else { + return false; + }; + + sdk_error + .as_service_error() + .is_some_and(PutItemError::is_conditional_check_failed_exception) +} + #[derive(Debug, PartialEq)] struct UpdateMetadataCondition(Fragment); impl DynamoDbPutCondition for UpdateMetadataCondition { fn into_parts(self) -> ConditionParts { - ConditionParts { - condition_expression: "#flags = :flags AND #size_payload = :size_payload AND #size_content = :size_content".to_string(), - expression_names: HashMap::from([ - ("#flags".to_string(), "flags".to_string()), - ("#size_payload".to_string(), "size_payload".to_string()), - ("#size_content".to_string(), "size_content".to_string()), - ]), - expression_values: HashMap::from([ - ( - ":flags".to_string(), - AttributeValue::N(self.0.flags.to_string()), - ), - ( - ":size_payload".to_string(), - AttributeValue::N(self.0.size_payload.to_string()), - ), - ( - ":size_content".to_string(), - AttributeValue::N(self.0.size_content.to_string()), - ), - ]), + committed_metadata_condition_parts(self.0) + } +} + +/// Guards publishing a payload's metadata. +/// +/// Publishing is only ever done by the writer whose own conditional upload created the object, +/// or by one that recovered the object's representation from the object itself. The condition +/// stops that publish from overwriting a row that changed in the meantime — most importantly an +/// obliteration lock, which must not be cleared by a concurrent write. +#[derive(Debug, PartialEq)] +enum MetadataWriteCondition { + /// No metadata row exists for this hash at all. + Absent, + /// The row still describes exactly this fragment. + Unchanged(Fragment), +} + +impl DynamoDbPutCondition for MetadataWriteCondition { + fn into_parts(self) -> ConditionParts { + match self { + MetadataWriteCondition::Absent => ConditionParts { + condition_expression: "attribute_not_exists(#hash)".to_string(), + expression_names: HashMap::from([("#hash".to_string(), "hash".to_string())]), + expression_values: HashMap::new(), + }, + MetadataWriteCondition::Unchanged(expected) => { + committed_metadata_condition_parts(expected) + } } } } @@ -330,11 +558,105 @@ static STORE_ATTRIBUTES: LazyLock<[KeyValue; 1]> = type BatchTaskResult = Result<(usize, StoreMatch), (usize, StoreError)>; +/// Result of an upload: whether this writer is the one that created the object. +#[derive(Debug, PartialEq, Eq)] +enum UploadOutcome { + /// This writer's bytes are now the object's contents. + Stored, + /// An object this writer did not create is already under the key. + AlreadyPresent, +} + +/// What became of an object found under a key this writer does not own. +#[derive(Debug, PartialEq, Eq)] +enum UnpublishedObject { + /// A published row accounts for it, so it is durable and described. + Published, + /// An obliteration owns the hash, so this object should not exist at all. A tombstone is a + /// row, but it does not account for an object. + ObliteratedRemnant, + /// Stored too recently to conclude anything: its writer is most likely still finishing. + InFlight, + /// Stored long enough ago that its writer would have published by now. The tag identifies + /// the object as observed, so reclaiming it can be made single-winner. + Abandoned(Option), +} + +/// Whether this writer's bytes displaced someone else's. +/// +/// A conditional upload cannot displace anything, so its bytes stay canonical until an +/// obliteration or a reclaim removes them. An unconditional one replaces whatever was there, +/// which is what decides who may overwrite whose metadata when a publish is rejected. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum WriteMode { + /// Written only because the key was free. + Exclusive, + /// Written over whatever was there. + Displacing, +} + +/// How far a publish got before the row moved underneath it. +#[derive(Debug, PartialEq, Eq)] +enum PublishOutcome { + Published, + /// An obliteration took the row. It deletes the payload itself, so there is nothing to do. + Obliterating, + /// Another writer owns the hash now, and these bytes are no longer known to be the stored + /// ones, so publishing over it would be a guess. + Superseded, + /// The row kept changing; the publish never landed. + Contended, +} + +/// How many times a publish re-reads and re-conditions before giving up. Only the metadata write +/// is retried: the bytes are already stored, so re-uploading them buys nothing. +const PUBLISH_MAX_ATTEMPTS: usize = 3; + +/// Classify a failed upload, keeping the underlying error attached so the cause survives into +/// the error chain rather than only into the log line. +fn upload_error(key: &str, error: AwsError) -> StoreError +where + AwsError: std::error::Error + Send + Sync + 'static, +{ + warn!(%key, ?error, "Failed to write payload"); + + if matches!(error, AwsError::AwsSdkError(_)) { + StoreError::from(SlowDown) + } else { + StoreError::internal_with_context(error, "S3 put object failed") + } +} + struct GetS3objectContentsOutput { read: usize, bytes: BytesMut, } +/// Fragment storage backed by S3 for payloads and `DynamoDB` for the metadata describing them. +/// +/// # Published metadata is authoritative +/// +/// A published metadata row is taken as proof that the payload it describes is in S3. That is +/// what lets a put deduplicate — recording a reference to content another partition already +/// stored, without re-uploading it or consulting S3 at all — and it holds because metadata is +/// only ever published after the upload that stored those bytes succeeded. +/// +/// The assumption is not verified on the read or deduplication paths, deliberately: checking it +/// would cost an S3 request per put and give back exactly what deduplication buys. So if an +/// object is removed from S3 while its metadata row survives — a lifecycle rule, a direct +/// deletion, or S3 and `DynamoDB` being restored to different points in time — the store cannot +/// tell, and: +/// +/// - reads of that hash fail from every partition referencing it, not only the one that wrote it; +/// - further puts of the same content deduplicate onto it, spreading references to a payload that +/// cannot be read; +/// - nothing repairs it. Re-writing the content does not, because a published row makes the put +/// deduplicate rather than upload. +/// +/// This is an operational failure to be recovered by whatever restores the object, not a state +/// the store resolves on its own. It is reported rather than hidden: see +/// [`METRICS_MISSING_PAYLOAD_METRIC_NAME`], which counts reads that find published metadata with +/// no object behind it and should never be non-zero. pub struct AwsImmutableStore { s3: S3, dynamodb: DynamoDb, @@ -343,7 +665,11 @@ pub struct AwsImmutableStore { fragments_table_name: Arc, metadata_table_name: Arc, force_write: bool, + abandoned_grace_millis: u64, + obliteration_drain_millis: u64, latency_histogram: Histogram, + missing_payload_counter: Counter, + labels_missing_payload: LabelArray, labels_get: LabelArray, labels_put: LabelArray, labels_exist: LabelArray, @@ -359,6 +685,8 @@ impl AwsImmutableStore { let latency_histogram = provider.latency_histogram_ms(METRICS_OPERATION_LATENCY_METRIC_NAME); + let missing_payload_counter = provider.counter(METRICS_MISSING_PAYLOAD_METRIC_NAME); + let labels_missing_payload = provider.get_labels_for_operation_context("get"); let labels_exist = provider.get_labels_for_operation_context("exist"); let labels_get = provider.get_labels_for_operation_context("get"); let labels_put = provider.get_labels_for_operation_context("put"); @@ -382,7 +710,11 @@ impl AwsImmutableStore { fragments_table_name: Arc::from(settings.dynamodb.fragments_table_name.clone()), metadata_table_name: Arc::from(settings.dynamodb.metadata_table_name.clone()), force_write: settings.force_write, + abandoned_grace_millis: settings.abandoned_grace_millis(), + obliteration_drain_millis: settings.obliteration_drain_millis(), latency_histogram, + missing_payload_counter, + labels_missing_payload, labels_get, labels_put, labels_exist, @@ -742,30 +1074,6 @@ impl AwsImmutableStore { } } - async fn write_metadata( - &self, - repository: Context, - address: Address, - fragment: Fragment, - ) -> Result<(), StoreError> { - let metadata = FragmentMetadataEntry::new(address.hash).with_fragment(fragment); - let item = serde_dynamo::to_item(&metadata).map_err(|e| { - warn!("Failed to serialize metadata entry for repository: {repository:?} and address: {address:?} to dynamo av map: {e:?}"); - StoreError::internal_with_context(e, "Failed to serialize metadata for DynamoDB write") - })?; - - self.dynamodb.put_item(&self.metadata_table_name, item).await.map_err(|e| { - warn!("Failed to save metadata entry for repository: {repository:?} and address: {address:?}: {e:?}"); - if matches!(&e, AwsError::AwsSdkError(_)) { - StoreError::from(SlowDown) - } else { - StoreError::internal_with_context(e, "DynamoDB metadata write failed") - } - })?; - - Ok(()) - } - async fn update_metadata( &self, address: Address, @@ -832,6 +1140,97 @@ impl AwsImmutableStore { } } + /// Publish the metadata describing a payload that is already in S3, making it visible. + /// + /// Only ever called by a writer whose own upload put the bytes under the key — a conditional + /// upload that created the object, a reclaim that replaced an abandoned one, or a + /// `force_write` that replaced it deliberately — so the fragment published here always + /// describes the bytes actually stored. + /// + /// `Ok(false)` means the condition rejected the write and the caller should re-probe. + async fn publish_metadata( + &self, + hash: Hash, + fragment: Fragment, + condition: MetadataWriteCondition, + ) -> Result { + let item = serde_dynamo::to_item(FragmentMetadataEntry::new(hash).with_fragment(fragment)) + .map_err(|e| { + warn!(%hash, ?e, "Failed to serialize metadata entry"); + StoreError::internal_with_context(e, "Failed to serialize metadata for DynamoDB") + })?; + + match self + .dynamodb + .put_item_conditional(&self.metadata_table_name, item, condition) + .await + { + Ok(_) => Ok(true), + Err(e) if is_conditional_check_failed(&e) => { + debug!(%hash, "Metadata changed before this write could publish"); + Ok(false) + } + Err(e) => { + warn!(%hash, ?e, "Failed to publish payload metadata"); + if matches!(&e, AwsError::AwsSdkError(_)) { + Err(StoreError::from(SlowDown)) + } else { + Err(StoreError::internal_with_context( + e, + "DynamoDB metadata publish failed", + )) + } + } + } + } + + /// Publish, re-reading and re-conditioning if the row moves underneath us. + /// + /// A rejected publish means the row changed since it was probed, not that the bytes are + /// wrong, so the right response is to re-condition against what is there now rather than + /// unwind and store the payload again. + async fn publish_converging( + &self, + hash: Hash, + fragment: Fragment, + mut condition: MetadataWriteCondition, + mode: WriteMode, + ) -> Result { + for _ in 0..PUBLISH_MAX_ATTEMPTS { + if self.publish_metadata(hash, fragment, condition).await? { + return Ok(PublishOutcome::Published); + } + + let current = self.metadata_lookup(hash).await?; + + if matches!(current, Some(f) if f.flags & FragmentFlags::PayloadObliteration != 0) { + return Ok(PublishOutcome::Obliterating); + } + + // Only a writer that displaced what was under the key may overwrite another writer's + // metadata, because only it knows the stored bytes are its own. A writer whose upload + // was conditional cannot know that: its bytes may since have been reclaimed, and + // publishing over the reclaimer would describe their object with this writer's + // fragment. + if mode == WriteMode::Exclusive { + return Ok(PublishOutcome::Superseded); + } + + condition = match current { + Some(current) => MetadataWriteCondition::Unchanged(current), + None => MetadataWriteCondition::Absent, + }; + } + + Ok(PublishOutcome::Contended) + } + + /// Record that `(repository, context)` references a payload, without guarding the metadata + /// row. + /// + /// Obliteration marks the metadata row before removing a payload and counts references again + /// afterwards, so a put that saw no mark is either counted by that second pass or has already + /// backed off. Nothing here needs to be conditional on the row. async fn associate_fragment( &self, repository: Context, @@ -847,13 +1246,18 @@ impl AwsImmutableStore { ) })?; - self.dynamodb.put_item(&self.fragments_table_name, item).await + self.dynamodb + .put_item(&self.fragments_table_name, item) + .await .map_err(|e| { warn!({REPOSITORY_ID} = %repository, {ADDRESS} = %address, error = ?e, "Failed to put item while storing fragment association"); if matches!(&e, AwsError::AwsSdkError(_)) { StoreError::from(SlowDown) } else { - StoreError::internal_with_context(e, "DynamoDB fragment association write failed") + StoreError::internal_with_context( + e, + "DynamoDB fragment association write failed", + ) } })?; @@ -910,13 +1314,153 @@ impl AwsImmutableStore { Ok(()) } + /// Store a fragment, deduplicating the payload against content the server already holds. + /// + /// The probe is two strongly consistent single-item reads on different tables, issued + /// together so they cost one round trip. The metadata table is keyed by hash alone, so + /// asking whether a payload is already durable is a point lookup whose cost does not grow + /// with how many partitions reference it — and S3 is never consulted to answer it. + /// + /// ```text + /// OK returned success RE-PROBE loop back to the probe (max 3, then SLOWDOWN) + /// ERROR returned failure SLOWDOWN returned to the caller to retry later + /// + /// PROBE GetItem fragments PK=hash SK=repo|ctx ┐ concurrent + /// GetItem metadata PK=hash ┘ no S3 + /// │ + /// ├─ no metadata row ...................................... UPLOAD cond: absent + /// ├─ flags: Obliterating .................................. SLOWDOWN + /// ├─ force_write .......................................... UPLOAD cond: unchanged + /// ├─ flags: Obliterated ................................... UPLOAD cond: unchanged + /// ├─ size_content differs ................................. ERROR hash collision + /// ├─ association present .................................. OK (no writes) + /// ├─ payload supplied ..................................... DEDUPLICATE + /// └─ no payload ........................................... ERROR payload required + /// + /// DEDUPLICATE PutItem fragments ............................ OK + /// No S3 request: a published row already means the payload is durable. The stored + /// fragment is referenced, never the incoming one. + /// + /// UPLOAD PutObject S3 If-None-Match (or unconditional under force_write) + /// ├─ stored ............................................... PUBLISH + /// └─ 412, the key is taken ................................ RESOLVE + /// + /// RESOLVE GetItem metadata + /// ├─ published ............................................ RE-PROBE + /// ├─ obliteration flags → discard the object .............. RE-PROBE + /// └─ absent → HeadObject S3 ETag + Last-Modified + /// ├─ younger than the threshold ...................... SLOWDOWN + /// └─ older → PutObject S3 If-Match (single winner) + /// ├─ 412, lost the reclaim ...................... RE-PROBE + /// └─ stored ..................................... PUBLISH + /// + /// PUBLISH PutItem metadata conditional + /// ├─ accepted → PutItem fragments ......................... OK + /// └─ rejected → GetItem metadata + /// ├─ obliteration flags → discard the object ......... RE-PROBE + /// ├─ Exclusive (our upload was conditional) .......... RE-PROBE + /// └─ Displacing → re-condition and retry (max 3) + /// ├─ accepted ................................... OK + /// └─ exhausted .................................. ERROR (logged loudly) + /// ``` + /// + /// `SLOWDOWN` means nothing will change within this request — another writer owns the + /// outcome, so the caller has to come back. `RE-PROBE` means the state changed underneath + /// us and the correct action is now a different one, which re-reading resolves without + /// troubling the caller. + /// + /// The two `WriteMode` cases decide who may overwrite whose metadata after a rejected + /// publish. `Exclusive` means the upload was conditional, so these bytes may since have been + /// reclaimed and this writer cannot vouch for what is stored; it stands down. `Displacing` + /// means a reclaim or a `force_write` replaced whatever was there, so the stored bytes are + /// known to be this writer's. + async fn put_deduplicated( + &self, + repository: Context, + address: Address, + fragment: Fragment, + payload: Option, + ) -> Result<(), StoreError> { + for _ in 0..PUT_MAX_ATTEMPTS { + // Both reads are single-item and hit different tables, so issuing them together costs + // one round trip rather than two. The metadata table is keyed by hash alone, which is + // what keeps the deduplication probe independent of how many partitions reference the + // content: no query, no scan, and S3 is never consulted. + let association = FragmentsEntry::new(repository, address); + let (associated, state) = tokio::join!( + self.exists_exact(&association), + self.metadata_lookup(address.hash), + ); + + match decide_put( + fragment, + associated?, + state?, + self.force_write, + payload.is_some(), + ) { + PutAction::Done => return Ok(()), + + PutAction::PayloadRequired => { + return Err(StoreError::internal("Payload buffer required")); + } + + // Backing off is what keeps a new reference from appearing while the payload is + // being removed, which is what lets the obliteration count references without a + // transaction to serialise against. + PutAction::Obliterating => { + debug!(%address, "Payload is marked for obliteration; backing off"); + return Err(StoreError::from(SlowDown)); + } + + PutAction::Collision => return Err(StoreError::internal("Hash collision")), + + // The payload is already durable, so this only has to record the reference. + // Obliteration marks the row before removing a payload and counts references + // again afterwards, so a put that saw no mark is either counted or backed off. + PutAction::Deduplicate(_) => { + self.associate_fragment(repository, address).await?; + + return Ok(()); + } + + PutAction::Upload(condition) => { + let Some(payload) = payload.clone() else { + return Err(StoreError::internal("Payload buffer required")); + }; + + if self + .write_payload(repository, address, fragment, payload, condition) + .await? + { + return Ok(()); + } + } + } + } + + debug!(%address, "Gave up storing after repeatedly losing the race for the payload"); + Err(StoreError::from(SlowDown)) + } + + /// Store a payload and publish the metadata describing it. + /// + /// The upload is conditional on the key being absent, so exactly one writer can ever create + /// the bytes for a hash. That is the whole of the mutual exclusion: no lock is needed, + /// because S3 itself arbitrates. Metadata is published only after *this* writer's own upload + /// created the object, so the published fragment always describes the bytes that are there — + /// which is what stops two writers holding the same content in different representations + /// (different compression, say) from leaving the pair disagreeing. + /// + /// `Ok(false)` means another writer got there first and the caller should re-probe. async fn write_payload( &self, repository: Context, address: Address, fragment: Fragment, payload: Bytes, - ) -> Result<(), StoreError> { + publish_with: MetadataWriteCondition, + ) -> Result { if payload.len() != fragment.size_payload as usize { warn!( "Failed to write fragment to immutable store for address: {address}, payload size invalid (expected {} bytes, but got {})", @@ -930,31 +1474,237 @@ impl AwsImmutableStore { } let mut dst = [0u8; 64]; - let hash = lore_revision::util::to_hex_str(address.hash.data(), &mut dst); + let key = lore_revision::util::to_hex_str(address.hash.data(), &mut dst); - self.s3 - .put_object(self.bucket.as_str(), hash, payload.to_vec()) + // `force_write` deliberately replaces whatever is stored, so it writes unconditionally; + // it is an operator override rather than part of the concurrent write protocol. Replacing + // in place keeps it from ever leaving the key empty, which deleting first would. + let (stored, mut mode) = if self.force_write { + (self.upload(key, &payload).await?, WriteMode::Displacing) + } else { + ( + self.upload_if_absent(key, &payload).await?, + WriteMode::Exclusive, + ) + }; + + let (published_fragment, condition) = match stored { + UploadOutcome::Stored => (fragment, publish_with), + + // The key is taken by an object this writer did not create, so its bytes are not the + // ones `fragment` describes and publishing `fragment` would tear the pair. + UploadOutcome::AlreadyPresent => { + match self.resolve_unpublished_object(address.hash).await? { + // Another writer published while we were uploading. Nothing to store. + UnpublishedObject::Published => return Ok(false), + + // The object outlived the content it belongs to. Nothing references it and + // nothing can adopt it, so removing it both unwedges the hash — the next + // attempt's conditional upload can finally succeed — and stops obliterated + // bytes from sitting in S3 indefinitely. + UnpublishedObject::ObliteratedRemnant => { + self.discard_orphaned_object(address.hash).await; + return Ok(false); + } + + // Another writer stored this recently and has not published yet. Its + // representation is the one that will win, so this writer backs off and lets + // the caller come back rather than holding the request or racing it. + UnpublishedObject::InFlight => { + debug!(%address, "Upload is not published yet; backing off"); + return Err(StoreError::from(SlowDown)); + } + + // Long enough unpublished that the writer that stored it is gone. Take the + // key over with bytes this writer can vouch for, rather than trying to work + // out what the abandoned ones were. + UnpublishedObject::Abandoned(etag) => { + if !self.reclaim_object(key, &payload, etag.as_deref()).await? { + debug!(%address, "Lost the race to reclaim the abandoned object"); + return Ok(false); + } + + info!(%address, "Reclaimed an abandoned object"); + mode = WriteMode::Displacing; + + (fragment, MetadataWriteCondition::Absent) + } + } + } + }; + + match self + .publish_converging(address.hash, published_fragment, condition, mode) + .await? + { + PublishOutcome::Published => {} + + // Every path that reaches here stored this writer's bytes: either the conditional + // upload created the object, or the reclaim replaced it. Anything else returned + // earlier. So the object is always this writer's to withdraw, and leaving it would + // put content an obliteration is deleting back into S3. + PublishOutcome::Obliterating => { + self.discard_orphaned_object(address.hash).await; + + return Ok(false); + } + + // An unconditional overwrite already replaced the stored bytes, so giving up here + // leaves them described by another writer's metadata and nothing will repair it. + // That has to surface as a failure an operator can see, not as a retry hint. + PublishOutcome::Contended if mode == WriteMode::Displacing => { + error!( + "Storing {address} replaced the stored payload but could not publish its \ + metadata; the stored bytes and the published metadata may disagree" + ); + return Err(StoreError::internal(format!( + "Failed to publish metadata for replaced payload {address}" + ))); + } + + // Someone else owns the hash now, or the row would not settle. Either way the + // conditional upload left any existing object untouched, so there is nothing to + // repair and the caller can simply re-probe. + PublishOutcome::Superseded | PublishOutcome::Contended => return Ok(false), + } + + self.associate_fragment(repository, address) + .await + .map(|()| true) + } + + /// Remove an object this writer created but could not publish, when an obliteration owns the + /// hash. + /// + /// Obliteration marks the row before it deletes the payload, so a put that slips in between + /// that delete and the final tombstone would otherwise leave the content back in S3 — + /// unreferenced and unreadable, but present, which is not what "permanently delete" should + /// mean. Re-reading the row is what makes removing it safe: it only proceeds while an + /// obliteration holds the hash, which is exactly the case where nothing else can have adopted + /// these bytes. + async fn discard_orphaned_object(&self, hash: Hash) { + let obliterating = matches!( + self.metadata_lookup(hash).await, + Ok(Some(fragment)) if fragment.flags & FragmentFlags::PayloadObliteration != 0 + ); + + if !obliterating { + return; + } + + info!(%hash, "Discarding an orphaned object left behind by an obliteration"); + + if let Err(e) = self.delete_payload(hash).await { + warn!(%hash, ?e, "Failed to discard the unpublished object"); + } + } + + /// Decide what to do about an object that exists under a key this writer does not own. + /// + /// The overwhelmingly likely explanation is a writer that is merely slow, so an object is + /// only treated as abandoned once it has gone unpublished for longer than the threshold. + /// Concluding that too early is not a correctness problem — reclaiming is single-winner and + /// only a reclaimer may overwrite another writer's metadata — but it throws away a live + /// writer's upload, so the doubt is resolved in their favour. + async fn resolve_unpublished_object( + &self, + hash: Hash, + ) -> Result { + match self.metadata_lookup(hash).await? { + // A published row accounts for these bytes. + Some(fragment) if fragment.flags & FragmentFlags::PayloadObliteration == 0 => { + return Ok(UnpublishedObject::Published); + } + // A tombstone is a row, but it does not account for an object: obliteration deletes + // the payload, so anything still under the key outlived its content. + Some(_) => return Ok(UnpublishedObject::ObliteratedRemnant), + None => {} + } + + let mut dst = [0u8; 64]; + let key = lore_revision::util::to_hex_str(hash.data(), &mut dst); + + let object = self + .s3 + .head_object(self.bucket.as_str(), key) .await - .map(|_| ()) .map_err(|e| { - warn!("Failed to write payload for hash: {}: {e:?}", address.hash); - if matches!(&e, AwsError::AwsSdkError(_)) { - StoreError::from(SlowDown) - } else { - StoreError::internal_with_context(e, "S3 put object failed") - } + warn!(%hash, ?e, "Failed to inspect the unpublished object"); + StoreError::from(SlowDown) })?; - // Writing metadata is not tied to writing the payload to S3, which means that over time - // we'll likely wind up in a scenario where some fragments exist in S3, but their associated - // metadata does not exist in Dynamo. In this scenario, a later query and/or read for the - // fragment would treat it as not found, prompting clients to resend the fragment. This - // means that whenever we land in this scenario, we should be self-healing. - self.write_metadata(repository, address, fragment).await?; + // How long the object has been sitting unpublished is what separates a writer that has + // died from one that is still finishing, and S3 already knows it. Asking is what keeps + // this decision out of the request: waiting here would hold the caller for the whole + // threshold to learn something a timestamp answers immediately. + if !is_abandoned(object.last_modified(), self.abandoned_grace_millis) { + return Ok(UnpublishedObject::InFlight); + } - self.associate_fragment(repository, address).await?; + debug!(%hash, "Object has gone unpublished long enough to be abandoned"); - Ok(()) + Ok(UnpublishedObject::Abandoned( + object.e_tag().map(ToString::to_string), + )) + } + + /// Take over an object whose writer never published it. + /// + /// Conditioned on the object still being the one just inspected, so exactly one writer can + /// reclaim a given object: a second one racing for it is rejected and re-probes instead of + /// replacing bytes the first is about to publish. + /// + /// `Ok(false)` means the reclaim was lost. + async fn reclaim_object( + &self, + key: &str, + payload: &Bytes, + etag: Option<&str>, + ) -> Result { + let Some(etag) = etag else { + // Nothing to condition on. Only reachable if S3 returned no entity tag, and the + // exposure is the one `force_write` already accepts. + warn!(%key, "Reclaiming unconditionally: the stored object reported no entity tag"); + self.upload(key, payload).await?; + + return Ok(true); + }; + + match self + .s3 + .put_object_if_match(self.bucket.as_str(), key, payload.to_vec(), etag) + .await + { + Ok(_) => Ok(true), + Err(e) if crate::s3::is_precondition_failed(&e) => Ok(false), + Err(e) => Err(upload_error(key, e)), + } + } + + /// A single conditional upload. + async fn upload_if_absent( + &self, + key: &str, + payload: &Bytes, + ) -> Result { + match self + .s3 + .put_object_if_absent(self.bucket.as_str(), key, payload.to_vec()) + .await + { + Ok(_) => Ok(UploadOutcome::Stored), + Err(e) if crate::s3::is_precondition_failed(&e) => Ok(UploadOutcome::AlreadyPresent), + Err(e) => Err(upload_error(key, e)), + } + } + + /// An unconditional upload, replacing whatever is stored under the key. + async fn upload(&self, key: &str, payload: &Bytes) -> Result { + self.s3 + .put_object(self.bucket.as_str(), key, payload.to_vec()) + .await + .map(|_| UploadOutcome::Stored) + .map_err(|e| upload_error(key, e)) } /// Permanently delete a payload from S3 by removing *ALL* versions from the bucket. @@ -1026,7 +1776,11 @@ impl AwsImmutableStore { async fn metadata_with_load_validation(&self, hash: Hash) -> Result { let metadata = self.metadata_with_size_validation(hash).await?; - if (metadata.flags & FragmentFlags::PayloadObliteration) != 0 { + // Only a tombstone means the payload is gone. A row that is merely marked still has its + // payload — it is removed after the references are counted — so refusing reads there + // would hide content from every partition holding it, and would hide it permanently if + // the obliteration that set the mark never finished. + if (metadata.flags & FragmentFlags::PayloadObliterated) != 0 { return Err(StoreError::from(AddressNotFound::from( Address::zero_context_hash(hash), ))); @@ -1035,8 +1789,13 @@ impl AwsImmutableStore { Ok(metadata) } - async fn load_metadata(&self, hash: Hash) -> Result { - let item = serde_dynamo::to_item(FragmentMetadataEntry::new(hash)).map_err(|e| { + /// Read the raw metadata row for `hash`. A missing row is `Ok(None)` rather than an error, so + /// callers can distinguish "nothing stored" from "the read failed". + async fn metadata_entry( + &self, + hash: Hash, + ) -> Result, StoreError> { + let key = serde_dynamo::to_item(FragmentMetadataEntry::new(hash)).map_err(|e| { warn!("Failed to serialize fragment metadata entry for {hash}: {e:?}"); StoreError::internal_with_context( e, @@ -1044,11 +1803,11 @@ impl AwsImmutableStore { ) })?; - let metadata: FragmentMetadataEntry = if let Some(av_map) = self + let item = self .dynamodb .get_item( &self.metadata_table_name, - item, + key, true, /* consistent read */ ) .await @@ -1062,21 +1821,41 @@ impl AwsImmutableStore { StoreError::from(AddressNotFound::from(Address::zero_context_hash(hash))) } })? - .item - { - serde_dynamo::from_item(av_map).map_err(|e| { + .item; + + match item { + Some(av_map) => serde_dynamo::from_item(av_map).map(Some).map_err(|e| { warn!("Failed to deserialize fragment metadata: {e:?}"); StoreError::from(AddressNotFound::from(Address::zero_context_hash(hash))) - }) - } else { + }), + None => Ok(None), + } + } + + /// Read the metadata row for `hash` without treating absence as an error. This is the + /// deduplication probe: a published row proves the payload is durable in S3, because + /// metadata is only ever published after the upload that stored those bytes succeeded. + /// + /// One strongly consistent `GetItem` against a table keyed by hash alone, so the cost does + /// not grow with how many partitions reference the content, and S3 is never consulted. + async fn metadata_lookup(&self, hash: Hash) -> Result, StoreError> { + Ok(self + .metadata_entry(hash) + .await? + .and_then(|entry| entry.fragment)) + } + + async fn load_metadata(&self, hash: Hash) -> Result { + let Some(entry) = self.metadata_entry(hash).await? else { warn!("Failed to get metadata for fragment, no item found"); - Err(StoreError::from(AddressNotFound::from( + + return Err(StoreError::from(AddressNotFound::from( Address::zero_context_hash(hash), - ))) - }?; + ))); + }; - metadata.fragment.ok_or_else(|| { - warn!("No fragment found on metadata from store: {metadata:?}"); + entry.fragment.ok_or_else(|| { + warn!("No fragment found on metadata from store: {entry:?}"); StoreError::internal("Fragment metadata entry missing fragment field") }) } @@ -1186,12 +1965,28 @@ impl AwsImmutableStore { let fragment = metadata_result?; let s3_contents = match s3_result { - Some(r) => r?, - None => s3_fut.await?, + Some(r) => r, + None => s3_fut.await, }; - let payload = self.read_payload(s3_contents, hash, fragment)?; - Ok((fragment, payload)) + // Metadata resolved, so the store believes this payload is durable, but S3 does not have + // it. Reads of this hash will fail from every partition referencing it, and because a + // missing object is reported as a plain not-found, it is otherwise indistinguishable + // from content that was never stored. Say so explicitly and count it. + let s3_contents = s3_contents.inspect_err(|error| { + if error.is_address_not_found() { + self.missing_payload_counter + .add(1, &self.labels_missing_payload); + error!( + %hash, + "Payload is published in metadata but absent from S3; content for this hash \ + has been lost and will not be repaired by writing it again" + ); + } + })?; + + let payload = self.read_payload(s3_contents, hash, fragment)?; + Ok((fragment, payload)) } } @@ -1323,75 +2118,11 @@ impl ImmutableStoreTrait for AwsImmutableStore { lore_storage::validate_fragment_size(&fragment)?; } let repository: Context = partition.into(); - timed!( - self.latency_histogram, - &self.labels_put, - { - let query = self.do_query( - repository, - address, - StoreMatch::MatchFull, - false, /* hide obliterates */ - ) - .await; - - let match_made = if !self.force_write && query.is_ok() { - let query = query?; - - if (query.fragment.flags & FragmentFlags::PayloadObliterating) == FragmentFlags::PayloadObliterating - { - info!("Received request to put fragment at {address} that is in the process of being obliterated"); - return Err(StoreError::internal(format!("Failed to obliterate immutable {address}"))); - } - - if query.match_made != StoreMatch::MatchNone - && fragment.size_content != query.fragment.size_content - && (query.fragment.flags & FragmentFlags::PayloadObliterated) != FragmentFlags::PayloadObliterated - { - return Err(StoreError::internal("Hash collision")); - } - - query.match_made - } else { - // If we're in this branch because the query failed, we should log the error. - if let Err(e) = query { - warn!("Query failed for address: {address:?} in repository: {repository}: {e:?}"); - } - - StoreMatch::MatchNone - }; - - match match_made { - // If the fragment exists with the same context, there's nothing to do. - StoreMatch::MatchFull => Ok(()), - - // If we matched on hash + repo, then we need to associate the fragment with the new - // context. Does not need the payload as it already exist in repository. - StoreMatch::MatchPartition => { - self.associate_fragment(repository, address).await - } - - // If we were only able to match on hash, the payload must have been provided. - // If so, associate the fragment. - StoreMatch::MatchHash if payload.is_some() => { - self.associate_fragment(repository, address).await - } - - // If no match, the payload must have been provided. Write it to S3 and store fragment. - StoreMatch::MatchNone if payload.is_some() => { - self.write_payload(repository, address, fragment, payload.unwrap()) - .await - } - - // If we were only able to match on hash, or were not able to match at all, and no - // payload was provided, that's an error. - StoreMatch::MatchHash | StoreMatch::MatchNone => { - Err(StoreError::internal("Payload buffer required")) - } - } - } - ) - .into() + timed!(self.latency_histogram, &self.labels_put, { + self.put_deduplicated(repository, address, fragment, payload) + .await + }) + .into() } #[lore_macro::lore_instrument] @@ -1408,36 +2139,84 @@ impl ImmutableStoreTrait for AwsImmutableStore { // expect this to be invoked, the log output in this method is intentionally very verbose. let span = tracing::Span::current(); - let original_metadata = self - .metadata_with_size_validation(address.hash) - .instrument(span.clone()) - .await?; + let original_metadata = self.metadata_with_size_validation(address.hash).await?; - info!("Original metadata: {original_metadata:?}"); + info!(?original_metadata, "Loaded metadata"); - // Acquire the lock on the fragment. - let updated_metadata = if original_metadata.flags & FragmentFlags::PayloadObliteration == 0 - { - let mut updated_metadata = original_metadata; - updated_metadata.flags |= FragmentFlags::PayloadObliterating; + if original_metadata.flags & FragmentFlags::PayloadObliterated != 0 { + info!("Fragment has already been obliterated"); + return Ok(()); + } + + // Another obliteration owns the mark. Removing this partition's reference is what + // this call has to do and is safe whoever owns the mark, since it is a single row + // keyed by this partition and context. The payload and the metadata are the mark + // owner's to decide: it counts the references after its own drain and will see this + // one gone, whereas racing it here could delete a payload it has decided to keep. + if original_metadata.flags & FragmentFlags::PayloadObliterating != 0 { + info!("Another obliteration holds the mark; removing this association only"); + + self.delete_association(repository, address).await?; + stats + .num_fragments + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - self.update_metadata(address, updated_metadata, original_metadata) - .instrument(span.clone()) - .await?; - info!("Acquired obliteration lock, updated metadata: {updated_metadata:?}"); - updated_metadata - } else { - info!("Fragment metadata indicates fragment is already being (or has previously been) obliterated"); return Ok(()); + } + + // Removing this partition's reference is the part that has to happen. Everything after + // it — deleting the shared payload, marking the metadata — is cleanup that only + // applies once nothing references the content any more. + // Mark before touching the association. A put that reads the mark backs off, which + // is what stops a reference appearing while this runs — including the reference this + // obliteration is about to delete being written straight back by the very partition + // it is being deleted for. + let considered_metadata = { + let mut considered = original_metadata; + considered.flags |= FragmentFlags::PayloadObliterating; + + self.update_metadata(address, considered, original_metadata) + .await?; + info!(?considered, "Marked for obliteration"); + considered }; - if updated_metadata.flags & FragmentFlags::PayloadFragmented != 0 { + self.delete_association(repository, address).await?; + stats + .num_fragments + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + + // A put that read the row before the mark went on may still be about to write its + // association. Association reads are strongly consistent, so this is not waiting for + // consistency: it is waiting for those writes to land, so the count below sees them. + // Afterwards no further reference can appear, which is what lets a single count + // decide, and lets an association be written plainly rather than in a transaction. + tokio::time::sleep(Duration::from_millis(self.obliteration_drain_millis)).await; + + info!("Association deleted, counting remaining associations..."); + if self.has_associations(address.hash).await? { + info!("Fragment still associated, clearing the mark"); + return self + .update_metadata(address, original_metadata, considered_metadata) + .await + .inspect_err(|e| { + warn!("Failed to reset metadata back to original state: {e:?}"); + }); + } + + // Only now is the content definitely going away, so its sub-fragments can go with it. + if considered_metadata.flags & FragmentFlags::PayloadFragmented != 0 { info!("Fragment is fragmented"); - // There's no reason we couldn't use the `updated_metadata` here, since `read_payload` - // only cares about the size fields (which haven't changed), but it feels wrong given it - // doesn't explicitly match the metadata for what's currently in S3. + // There's no reason we couldn't use the `considered_metadata` here, since + // `read_payload` only cares about the size fields (which haven't changed), but it + // feels wrong given it doesn't explicitly match the metadata for what's currently + // in S3. let payload = self - .read_payload(self.get_s3_object_contents(address.hash).await?, address.hash, original_metadata)? + .read_payload( + self.get_s3_object_contents(address.hash).await?, + address.hash, + original_metadata, + )? .to_aligned::(); let sub_fragments = payload.as_type_slice::(); @@ -1484,57 +2263,28 @@ impl ImmutableStoreTrait for AwsImmutableStore { if failures { warn!("Obliteration failed for at least one sub-fragment."); - return Err(StoreError::internal(format!("Failed to obliterate immutable {address}"))); + return Err(StoreError::internal(format!( + "Failed to obliterate immutable {address}" + ))); } info!("Done obliterating sub-fragments"); } - self.delete_association(repository, address) - .instrument(span.clone()) - .await?; - stats - .num_fragments - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - - // TODO(jcohen): Assuming we always lock the fragment regardless of the association count - // then this process of re-checking the count after removing the association should - // theoretically not be necessary since no one else should have been able to add a new - // fragment association while we maintain the lock. - info!("Association deleted, re-checking for other association..."); - let remain_associated = self - .has_associations(address.hash) - .instrument(span.clone()) - .await?; - - // If the association count is still >= 1 after we deleted, other references remain, so - // there's nothing left to do... - if remain_associated { - info!("Fragment still associated, nothing more to do"); - return self - .update_metadata(address, original_metadata, updated_metadata) - .instrument(span.clone()) - .await - .inspect_err(|e| { - warn!("Failed to reset metadata back to original state: {e:?}"); - }); - } - - self.delete_payload(address.hash) - .instrument(span.clone()) - .await?; + self.delete_payload(address.hash).await?; stats .num_payloads .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - let mut obliterated_metadata = updated_metadata; + let mut obliterated_metadata = considered_metadata; obliterated_metadata.flags = FragmentFlags::PayloadObliterated.bits(); obliterated_metadata.size_payload = 0; obliterated_metadata.size_content = 0; - // Final metadata update to clear out the sizes and set the flags to `Obliterated`. - self.update_metadata(address, obliterated_metadata, updated_metadata) + // Leave a tombstone rather than removing the row: it keeps a repeat obliteration + // idempotent, and is what a policy refusing re-upload would key on. + self.update_metadata(address, obliterated_metadata, considered_metadata) .await .inspect_err(|e| { // At this point we've already deleted the underlying payload, so there's not any @@ -1542,7 +2292,8 @@ impl ImmutableStoreTrait for AwsImmutableStore { // broken. warn!("Failed to finalize obliterate for {address}: {e:?}"); }) - }).into() + }) + .into() } #[lore_macro::lore_instrument] @@ -1642,6 +2393,7 @@ impl InstrumentProvider for AwsImmutableStoreInstrumentProvider { #[cfg(test)] mod test { use std::collections::HashMap; + use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use aws_sdk_dynamodb::operation::delete_item::DeleteItemError; @@ -1658,16 +2410,21 @@ mod test { use aws_sdk_s3::error::ErrorMetadata; use aws_sdk_s3::operation::delete_object::DeleteObjectError; use aws_sdk_s3::operation::delete_object::DeleteObjectOutput; + use aws_sdk_s3::operation::get_object::GetObjectError; use aws_sdk_s3::operation::get_object::GetObjectOutput; + use aws_sdk_s3::operation::head_object::HeadObjectOutput; use aws_sdk_s3::operation::list_object_versions::ListObjectVersionsError; use aws_sdk_s3::operation::list_object_versions::ListObjectVersionsOutput; + use aws_sdk_s3::operation::put_object::PutObjectError; use aws_sdk_s3::operation::put_object::PutObjectOutput; use aws_sdk_s3::primitives::SdkBody; use aws_sdk_s3::types::ObjectVersion; + use aws_sdk_s3::types::error::NoSuchKey; use aws_smithy_runtime_api::client::orchestrator::HttpResponse; use aws_smithy_runtime_api::client::result::SdkError; use aws_smithy_runtime_api::client::result::ServiceError; use aws_smithy_runtime_api::client::result::TimeoutError; + use aws_smithy_types::DateTime; use lore_base::runtime::LORE_CONTEXT; use lore_base::types::FragmentFlags; use lore_revision::fragment; @@ -1687,6 +2444,7 @@ mod test { const BUCKET: &str = "test-bucket"; const FRAGMENTS_TABLE_NAME: &str = "fragments"; const METADATA_TABLE_NAME: &str = "metadata"; + const ABANDONED_ETAG: &str = "\"abandoned-object\""; fn mock_lookup_fragments( dynamodb_mock: &mut MockDynamoDb, @@ -1763,17 +2521,134 @@ mod test { } } + /// Mock an association write. Obliteration marks the metadata row before removing a payload + /// and counts references afterwards, so a put that saw no mark is either counted by that pass + /// or has already backed off, and nothing about this write is conditional. fn mock_associate_fragment(dynamodb_mock: &mut MockDynamoDb, entry: &FragmentsEntry) { let item: HashMap = serde_dynamo::to_item(entry).unwrap(); dynamodb_mock .expect_put_item() - .with(eq(Arc::::from(FRAGMENTS_TABLE_NAME)), eq(item.clone())) - .return_once(move |_, _| { - Ok(PutItemOutput::builder().set_attributes(Some(item)).build()) + .withf(move |table, written| table.as_ref() == FRAGMENTS_TABLE_NAME && *written == item) + .returning(|_, _| Ok(PutItemOutput::builder().build())); + } + + /// Mock the metadata probe `put` issues, resolving to the supplied row (or to nothing). + fn mock_metadata_probe( + dynamodb_mock: &mut MockDynamoDb, + hash: Hash, + entry: Option, + ) { + let key: HashMap = + serde_dynamo::to_item(FragmentMetadataEntry::new(hash)).unwrap(); + let item = entry.map(|e| serde_dynamo::to_item(e).unwrap()); + + dynamodb_mock + .expect_get_item() + .with(eq(Arc::::from(METADATA_TABLE_NAME)), eq(key), eq(true)) + .return_once(move |_, _, _| Ok(GetItemOutput::builder().set_item(item).build())); + } + + /// Mock a fragments-table probe that answers every attempt with a miss. `put` re-probes on + /// each attempt, and a `return_once` expectation would be re-matched rather than fall + /// through. + fn mock_fragments_probe_repeated(dynamodb_mock: &mut MockDynamoDb, entry: &FragmentsEntry) { + let key: HashMap = serde_dynamo::to_item(entry).unwrap(); + + dynamodb_mock + .expect_get_item() + .with( + eq(Arc::::from(FRAGMENTS_TABLE_NAME)), + eq(key), + eq(true), + ) + .returning(|_, _, _| Ok(GetItemOutput::builder().set_item(None).build())); + } + + /// Mock a metadata probe whose answer changes between reads, so a test can model the row + /// moving underneath a writer. The last answer is repeated once the sequence is exhausted. + fn mock_metadata_probe_sequence( + dynamodb_mock: &mut MockDynamoDb, + hash: Hash, + answers: Vec>, + ) { + let key: HashMap = + serde_dynamo::to_item(FragmentMetadataEntry::new(hash)).unwrap(); + let items: Vec>> = answers + .into_iter() + .map(|entry| entry.map(|e| serde_dynamo::to_item(e).unwrap())) + .collect(); + let reads = AtomicUsize::new(0); + + dynamodb_mock + .expect_get_item() + .with(eq(Arc::::from(METADATA_TABLE_NAME)), eq(key), eq(true)) + .returning(move |_, _, _| { + let index = reads.fetch_add(1, Ordering::Relaxed).min(items.len() - 1); + + Ok(GetItemOutput::builder() + .set_item(items[index].clone()) + .build()) }); } + /// Mock a metadata probe that answers every read, as the recovery poll issues several. + fn mock_metadata_probe_repeated( + dynamodb_mock: &mut MockDynamoDb, + hash: Hash, + entry: Option, + ) { + let key: HashMap = + serde_dynamo::to_item(FragmentMetadataEntry::new(hash)).unwrap(); + let item = entry.map(|e| serde_dynamo::to_item(e).unwrap()); + + dynamodb_mock + .expect_get_item() + .with(eq(Arc::::from(METADATA_TABLE_NAME)), eq(key), eq(true)) + .returning(move |_, _, _| Ok(GetItemOutput::builder().set_item(item.clone()).build())); + } + + /// Mock the conditional publish that follows an upload, asserting it writes exactly + /// `fragment` and nothing more. + fn mock_publish_metadata(dynamodb_mock: &mut MockDynamoDb, hash: Hash, fragment: Fragment) { + let published: HashMap = + serde_dynamo::to_item(FragmentMetadataEntry::new(hash).with_fragment(fragment)) + .unwrap(); + + dynamodb_mock + .expect_put_item_conditional::() + .times(1) + .withf(move |table, item, _| { + table.as_ref() == METADATA_TABLE_NAME && *item == published + }) + .returning(|_, _, _| Ok(PutItemOutput::builder().build())); + } + + async fn initialize_immutable_store_with_grace( + s3: S3, + dynamodb: DynamoDb, + grace_millis: u64, + ) -> AwsImmutableStore { + let settings = AwsImmutableStoreSettings { + s3: S3StoreSettings::new(BUCKET.to_string()), + dynamodb: DynamoDbImmutableStoreSettings::new( + FRAGMENTS_TABLE_NAME.to_string(), + METADATA_TABLE_NAME.to_string(), + ), + force_write: false, + batch_exist_submission_limit: 1000, + abandoned_upload_grace_millis: Some(grace_millis), + obliteration_drain_millis: Some(0), + }; + + let execution = setup_execution("test".to_string()); + LORE_CONTEXT + .scope(execution.clone(), async move { + AwsImmutableStore::new(s3, dynamodb, &settings) + }) + .await + } + async fn initialize_immutable_store(s3: S3, dynamodb: DynamoDb) -> AwsImmutableStore { let settings = AwsImmutableStoreSettings { s3: S3StoreSettings::new(BUCKET.to_string()), @@ -1783,6 +2658,8 @@ mod test { ), force_write: false, batch_exist_submission_limit: 1000, + abandoned_upload_grace_millis: None, + obliteration_drain_millis: Some(0), }; let execution = setup_execution("test".to_string()); @@ -2126,21 +3003,13 @@ mod test { StoreMatch::MatchNone, ); - let item: HashMap = - serde_dynamo::to_item(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)) - .unwrap(); - - dynamodb_mock - .expect_put_item() - .with(eq(Arc::::from(METADATA_TABLE_NAME)), eq(item.clone())) - .return_once(move |_, _| { - Ok(PutItemOutput::builder().set_attributes(Some(item)).build()) - }); + mock_metadata_probe(&mut dynamodb_mock, address.hash, None); + mock_publish_metadata(&mut dynamodb_mock, address.hash, fragment); mock_associate_fragment(&mut dynamodb_mock, &entry); s3mock - .expect_put_object() + .expect_put_object_if_absent() .with( eq(BUCKET), eq(address.hash.to_string()), @@ -2253,8 +3122,8 @@ mod test { store .put(repository.into(), address, fragment, Some(payload), false) .await - .expect_err("expected put to fail") - .is_internal() + .expect_err("expected put to back off") + .is_slow_down() ); } @@ -2282,40 +3151,20 @@ mod test { size_content: 0, }; - let metadata_entry = FragmentMetadataEntry::new(address.hash); - let av_map: HashMap = - serde_dynamo::to_item(metadata_entry.clone()).unwrap(); - let full_entry = metadata_entry.with_fragment(obliterated_fragment); - let full_entry_av_map = serde_dynamo::to_item(full_entry.clone()).unwrap(); - - dynamodb_mock - .expect_get_item() - .with( - eq(Arc::::from(METADATA_TABLE_NAME)), - eq(av_map), - eq(true), - ) - .return_once(move |_, _, _| { - Ok(GetItemOutput::builder() - .set_item(Some(full_entry_av_map)) - .build()) - }); - - let item: HashMap = - serde_dynamo::to_item(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)) - .unwrap(); + mock_metadata_probe( + &mut dynamodb_mock, + address.hash, + Some(FragmentMetadataEntry::new(address.hash).with_fragment(obliterated_fragment)), + ); - dynamodb_mock - .expect_put_item() - .with(eq(Arc::::from(METADATA_TABLE_NAME)), eq(item.clone())) - .return_once(move |_, _| { - Ok(PutItemOutput::builder().set_attributes(Some(item)).build()) - }); + // A tombstone is not a deduplication source: the payload was deleted, so the bytes have + // to be stored again and the row republished. + mock_publish_metadata(&mut dynamodb_mock, address.hash, fragment); mock_associate_fragment(&mut dynamodb_mock, &entry); s3mock - .expect_put_object() + .expect_put_object_if_absent() .with( eq(BUCKET), eq(address.hash.to_string()), @@ -2404,24 +3253,9 @@ mod test { StoreMatch::MatchHash, ); - let metadata_entry = FragmentMetadataEntry::new(address.hash); - let av_map: HashMap = - serde_dynamo::to_item(metadata_entry.clone()).unwrap(); - let full_entry = metadata_entry.with_fragment(fragment); - let full_entry_av_map = serde_dynamo::to_item(full_entry.clone()).unwrap(); - - dynamodb_mock - .expect_get_item() - .with( - eq(Arc::::from(METADATA_TABLE_NAME)), - eq(av_map), - eq(true), - ) - .return_once(move |_, _, _| { - Ok(GetItemOutput::builder() - .set_item(Some(full_entry_av_map)) - .build()) - }); + // Nothing is stored for this hash, so there is nothing to deduplicate against and the + // caller has to supply the bytes. + mock_metadata_probe(&mut dynamodb_mock, address.hash, None); let store = initialize_immutable_store(s3mock, dynamodb_mock).await; let store = Arc::new(store); @@ -2435,50 +3269,851 @@ mod test { ); } + /// `force_write` is an operator override, but it must still not tear an obliteration's lock + /// off the row: doing so would resurrect content that is midway through being deleted. #[tokio::test] - async fn test_put_immutable_extra_data() { + async fn test_put_immutable_force_write_respects_obliteration() { let repository = random::(); let (fragment, address, payload) = fragment::generate_random(); - let mut s3mock = MockS3Impl::default(); + // Bare mocks: any upload or metadata write here is a test failure. + let s3mock = MockS3Impl::default(); let mut dynamodb_mock = MockDynamoDb::default(); let entry = FragmentsEntry::new(repository, address); + let mut obliterating = fragment; + obliterating.flags |= FragmentFlags::PayloadObliterating; - // Mock the list objects calls that `put_immutable` makes when querying for an object. mock_lookup_fragments( &mut dynamodb_mock, - entry.clone(), + entry, StoreMatch::MatchFull, StoreMatch::MatchNone, ); + mock_metadata_probe( + &mut dynamodb_mock, + address.hash, + Some(FragmentMetadataEntry::new(address.hash).with_fragment(obliterating)), + ); - let item: HashMap = - serde_dynamo::to_item(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)) - .unwrap(); + let settings = AwsImmutableStoreSettings { + s3: S3StoreSettings::new(BUCKET.to_string()), + dynamodb: DynamoDbImmutableStoreSettings::new( + FRAGMENTS_TABLE_NAME.to_string(), + METADATA_TABLE_NAME.to_string(), + ), + force_write: true, + batch_exist_submission_limit: 1000, + abandoned_upload_grace_millis: None, + obliteration_drain_millis: Some(0), + }; - dynamodb_mock - .expect_put_item() - .with(eq(Arc::::from(METADATA_TABLE_NAME)), eq(item.clone())) - .return_once(move |_, _| { - Ok(PutItemOutput::builder().set_attributes(Some(item)).build()) - }); + let execution = setup_execution("test".to_string()); + let store = LORE_CONTEXT + .scope(execution, async move { + AwsImmutableStore::new(s3mock, dynamodb_mock, &settings) + }) + .await; - mock_associate_fragment(&mut dynamodb_mock, &entry); + assert!( + Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect_err("force_write must not override an obliteration") + .is_slow_down() + ); + } - let mut body = vec![]; - body.extend_from_slice(payload.as_ref()); + /// Build a `force_write` store, the operator override that replaces whatever is stored. + async fn initialize_force_write_store(s3: S3, dynamodb: DynamoDb) -> AwsImmutableStore { + let settings = AwsImmutableStoreSettings { + s3: S3StoreSettings::new(BUCKET.to_string()), + dynamodb: DynamoDbImmutableStoreSettings::new( + FRAGMENTS_TABLE_NAME.to_string(), + METADATA_TABLE_NAME.to_string(), + ), + force_write: true, + batch_exist_submission_limit: 1000, + abandoned_upload_grace_millis: Some(100), + obliteration_drain_millis: Some(0), + }; - let real_len = body.len(); + let execution = setup_execution("test".to_string()); + LORE_CONTEXT + .scope(execution, async move { + AwsImmutableStore::new(s3, dynamodb, &settings) + }) + .await + } - let extra = random::<[u8; 32]>(); - body.extend_from_slice(extra.as_slice()); + fn conditional_check_failed() -> AwsError> { + aws_error( + PutItemError::ConditionalCheckFailedException( + ConditionalCheckFailedException::builder().build(), + ), + 400u16, + ) + } - // Ensure we only write bytes equal to the actual payload size, regardless of how much extra - // was sent. + fn precondition_failed() -> AwsError> { + aws_error( + PutObjectError::generic(ErrorMetadata::builder().code("PreconditionFailed").build()), + 412u16, + ) + } + + /// A publish rejected because the row moved is not a reason to store the payload again: the + /// bytes are already there, so only the metadata write is retried, against what is now stored. + #[tokio::test] + async fn test_put_immutable_publish_reconditions_without_reuploading() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + let mut moved = fragment; + moved.flags |= FragmentFlags::PayloadStoredDurable; + + mock_lookup_fragments( + &mut dynamodb_mock, + entry.clone(), + StoreMatch::MatchFull, + StoreMatch::MatchNone, + ); + // Probed as `fragment`, but by publish time the row reads as `moved`. + mock_metadata_probe_sequence( + &mut dynamodb_mock, + address.hash, + vec![ + Some(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)), + Some(FragmentMetadataEntry::new(address.hash).with_fragment(moved)), + ], + ); + + // Exactly one upload, however many times the publish is retried. + s3mock + .expect_put_object::>() + .times(1) + .returning(|_, _, _| Ok(PutObjectOutput::builder().build())); + + let mut seq = mockall::Sequence::default(); + dynamodb_mock + .expect_put_item_conditional::() + .times(1) + .in_sequence(&mut seq) + .withf(move |_, _, condition| *condition == MetadataWriteCondition::Unchanged(fragment)) + .returning(|_, _, _| Err(conditional_check_failed())); + dynamodb_mock + .expect_put_item_conditional::() + .times(1) + .in_sequence(&mut seq) + .withf(move |_, _, condition| *condition == MetadataWriteCondition::Unchanged(moved)) + .returning(|_, _, _| Ok(PutItemOutput::builder().build())); + + mock_associate_fragment(&mut dynamodb_mock, &entry); + + let store = initialize_force_write_store(s3mock, dynamodb_mock).await; + + Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect("the publish should have re-conditioned and succeeded"); + } + + /// A `force_write` that replaced the stored bytes but never managed to publish has left them + /// described by someone else's metadata, and nothing will repair that. It must surface as a + /// failure an operator can see, not as a retry hint. + #[tokio::test] + async fn test_put_immutable_force_write_publish_exhaustion_is_loud() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + + mock_lookup_fragments( + &mut dynamodb_mock, + entry, + StoreMatch::MatchFull, + StoreMatch::MatchNone, + ); + // The row never settles. + mock_metadata_probe_repeated( + &mut dynamodb_mock, + address.hash, + Some(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)), + ); + + s3mock + .expect_put_object::>() + .times(1) + .returning(|_, _, _| Ok(PutObjectOutput::builder().build())); + + dynamodb_mock + .expect_put_item_conditional::() + .times(PUBLISH_MAX_ATTEMPTS) + .returning(|_, _, _| Err(conditional_check_failed())); + + let store = initialize_force_write_store(s3mock, dynamodb_mock).await; + + let error = Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect_err("an unrepairable force_write must not look like a retry"); + + assert!( + error.is_internal() && !error.is_slow_down(), + "expected a visible failure, got {error:?}" + ); + } + + /// An object left under a key whose content has been obliterated outlived that content: a + /// tombstone is a row, but it does not account for an object. It must be discarded rather + /// than mistaken for a published payload, which would wedge the hash forever and leave + /// deleted bytes in S3. + #[tokio::test] + async fn test_put_immutable_discards_object_left_by_obliteration() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + let tombstone = Fragment { + flags: FragmentFlags::PayloadObliterated.bits(), + size_payload: 0, + size_content: 0, + }; + + // Two attempts: the first discards the remnant, the second stores the content. + mock_fragments_probe_repeated(&mut dynamodb_mock, &entry); + mock_metadata_probe_repeated( + &mut dynamodb_mock, + address.hash, + Some(FragmentMetadataEntry::new(address.hash).with_fragment(tombstone)), + ); + + let mut seq = mockall::Sequence::default(); + + // The remnant blocks the first upload... + s3mock + .expect_put_object_if_absent::>() + .times(1) + .in_sequence(&mut seq) + .returning(|_, _, _| Err(precondition_failed())); + + // ...so it is deleted... + s3mock + .expect_list_versions() + .times(1) + .in_sequence(&mut seq) + .returning(|_, _| Ok(ListObjectVersionsOutput::builder().build())); + s3mock + .expect_delete_object() + .times(1) + .in_sequence(&mut seq) + .returning(|_, _, _| Ok(DeleteObjectOutput::builder().build())); + + // ...and the retry stores this writer's bytes in its place. + s3mock + .expect_put_object_if_absent::>() + .times(1) + .in_sequence(&mut seq) + .returning(|_, _, _| Ok(PutObjectOutput::builder().build())); + + dynamodb_mock + .expect_put_item_conditional::() + .times(1) + .withf(move |_, _, condition| { + *condition == MetadataWriteCondition::Unchanged(tombstone) + }) + .returning(|_, _, _| Ok(PutItemOutput::builder().build())); + + mock_associate_fragment(&mut dynamodb_mock, &entry); + + let store = initialize_immutable_store_with_grace(s3mock, dynamodb_mock, 100).await; + + Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect("should have discarded the remnant and stored the content"); + } + + /// Reclaiming stores this writer's bytes, so losing the publish to an obliteration must + /// withdraw them. Otherwise content an obliteration is deleting comes back into S3 through + /// the reclaim path — the same hole the tombstone handling closes on the upload path. + #[tokio::test] + async fn test_put_immutable_discards_reclaimed_bytes_when_obliteration_wins() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + let mut obliterating = fragment; + obliterating.flags |= FragmentFlags::PayloadObliterating; + + mock_fragments_probe_repeated(&mut dynamodb_mock, &entry); + mock_metadata_probe_sequence( + &mut dynamodb_mock, + address.hash, + vec![ + // Nothing stored when the put probes, nor when the object is resolved... + None, + None, + // ...but an obliteration owns the hash by the time it is published. + Some(FragmentMetadataEntry::new(address.hash).with_fragment(obliterating)), + ], + ); + + s3mock + .expect_put_object_if_absent::>() + .times(1) + .returning(|_, _, _| Err(precondition_failed())); + s3mock.expect_head_object().times(1).returning(|_, _| { + Ok(HeadObjectOutput::builder() + .e_tag(ABANDONED_ETAG) + .last_modified(DateTime::from_secs(1)) + .build()) + }); + s3mock + .expect_put_object_if_match::>() + .times(1) + .returning(|_, _, _, _| Ok(PutObjectOutput::builder().build())); + + dynamodb_mock + .expect_put_item_conditional::() + .times(1) + .returning(|_, _, _| Err(conditional_check_failed())); + + // The reclaimed bytes must be withdrawn. + s3mock + .expect_list_versions() + .times(1) + .returning(|_, _| Ok(ListObjectVersionsOutput::builder().build())); + s3mock + .expect_delete_object() + .times(1) + .returning(|_, _, _| Ok(DeleteObjectOutput::builder().build())); + + let store = initialize_immutable_store_with_grace(s3mock, dynamodb_mock, 100).await; + + assert!( + Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect_err("an obliterated hash must not accept a put") + .is_slow_down() + ); + } + + /// `is_abandoned` is the only gate on reclaiming another writer's object, and the only + /// clock-dependent logic here, so its edges are worth pinning directly. + #[test] + fn abandonment_is_decided_conservatively() { + let now = now_millis(); + let recent = DateTime::from_millis((now - 1_000) as i64); + let old = DateTime::from_millis((now - 60_000) as i64); + let future = DateTime::from_millis((now + 60_000) as i64); + + assert!( + !is_abandoned(None, 10_000), + "an object that cannot be dated must never be reclaimed" + ); + assert!( + !is_abandoned(Some(&future), 10_000), + "an object dated in the future means clock skew, not abandonment" + ); + assert!( + !is_abandoned(Some(&recent), 10_000), + "an object younger than the threshold belongs to a writer still finishing" + ); + assert!( + is_abandoned(Some(&old), 10_000), + "an object older than the threshold has been left behind" + ); + } + + /// Sub-second thresholds cannot mean what they say if the stored time is truncated to whole + /// seconds, and the truncation error is up to a second — larger than the smallest threshold + /// the settings allow. + #[test] + fn stored_time_keeps_sub_second_precision() { + assert_eq!(datetime_millis(&DateTime::from_millis(1_500)), Some(1_500)); + assert_eq!(datetime_millis(&DateTime::from_millis(999)), Some(999)); + assert_eq!( + datetime_millis(&DateTime::from_secs(-1)), + None, + "a time before the epoch is not a usable age" + ); + } + + /// A payload that is published but missing from S3 is otherwise indistinguishable from + /// content that was never stored, because a missing object surfaces as a plain not-found. It + /// has to be called out, since nothing repairs it and every partition referencing the hash is + /// affected. + #[tokio::test] + #[traced_test] + async fn test_get_immutable_reports_published_metadata_with_no_object() { + let repository = random::(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let (fragment, address) = + mock_load_fragment_metadata(&mut dynamodb_mock, None, false /* fail */); + + let entry = FragmentsEntry::new(repository, address); + mock_lookup_fragments( + &mut dynamodb_mock, + entry, + StoreMatch::MatchFull, + StoreMatch::MatchFull, + ); + + s3mock + .expect_get_object() + .with(eq(BUCKET), eq(address.hash.to_string()), eq(None)) + .return_once(move |_, _, _| { + Err(aws_error( + GetObjectError::NoSuchKey(NoSuchKey::builder().build()), + 404u16, + )) + }); + + let store = initialize_immutable_store(s3mock, dynamodb_mock).await; + + let error = Arc::new(store) + .get(repository.into(), address, StoreMatch::MatchFull) + .await + .expect_err("a published payload with no object cannot be read"); + + assert!(error.is_address_not_found()); + assert!( + logs_contain("published in metadata but absent from S3"), + "losing a payload underneath the store must not look like an ordinary miss" + ); + + let _ = fragment; + } + + /// The metadata table holds rows written long before any of this, and this design keeps its + /// shape untouched so no table change or backfill is needed. Guards against a future addition + /// quietly changing what a published row looks like. + #[test] + fn published_rows_keep_their_original_shape() { + let (fragment, address, _) = fragment::generate_random(); + + let published: HashMap = + serde_dynamo::to_item(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)) + .unwrap(); + + let mut attributes = published.keys().cloned().collect::>(); + attributes.sort(); + assert_eq!( + attributes, + vec!["flags", "hash", "size_content", "size_payload"], + "a published row must carry exactly the attributes existing rows already have" + ); + + let parsed: FragmentMetadataEntry = serde_dynamo::from_item(published).unwrap(); + assert_eq!(parsed.fragment, Some(fragment)); + } + + /// A row that is only marked for obliteration still has its payload — it is removed after the + /// references are counted, and the mark is cleared again if any remain. Refusing reads there + /// would hide content from every partition holding it, and hide it permanently if the + /// obliteration that set the mark never finished. + #[tokio::test] + async fn test_get_immutable_marked_for_obliteration_is_still_readable() { + let repository = random::(); + let (mut fragment, address, payload) = fragment::generate_random(); + fragment.flags |= FragmentFlags::PayloadObliterating; + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + mock_lookup_fragments( + &mut dynamodb_mock, + entry, + StoreMatch::MatchFull, + StoreMatch::MatchFull, + ); + + let key: HashMap = + serde_dynamo::to_item(FragmentMetadataEntry::new(address.hash)).unwrap(); + let row = + serde_dynamo::to_item(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)) + .unwrap(); + dynamodb_mock + .expect_get_item() + .with(eq(Arc::::from(METADATA_TABLE_NAME)), eq(key), eq(true)) + .return_once(move |_, _, _| Ok(GetItemOutput::builder().set_item(Some(row)).build())); + + let stored = payload.to_vec(); + s3mock + .expect_get_object() + .with(eq(BUCKET), eq(address.hash.to_string()), eq(None)) + .return_once(move |_, _, _| { + Ok(GetObjectOutput::builder() + .set_body(Some(stored.into())) + .build()) + }); + + let store = initialize_immutable_store(s3mock, dynamodb_mock).await; + + let (read_fragment, read_payload) = Arc::new(store) + .get(repository.into(), address, StoreMatch::MatchFull) + .await + .expect("a marked payload is still stored and must still be readable"); + + assert_eq!(read_fragment, fragment); + assert_eq!(read_payload, payload); + } + + /// Deduplication skips the upload, not the requirement to present the bytes. A caller that + /// does not already reference the content cannot attach it by naming its hash alone. + #[tokio::test] + async fn test_put_immutable_payload_required_to_deduplicate() { + let repository = random::(); + let (fragment, address, _) = fragment::generate_random(); + + let s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + + // This partition holds no reference to the content... + mock_lookup_fragments( + &mut dynamodb_mock, + entry, + StoreMatch::MatchFull, + StoreMatch::MatchNone, + ); + + // ...and although the payload is durable elsewhere, that alone does not entitle this + // caller to a reference to it. No association write is mocked: making one is a failure. + mock_metadata_probe( + &mut dynamodb_mock, + address.hash, + Some(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)), + ); + + let store = initialize_immutable_store(s3mock, dynamodb_mock).await; + + assert!( + Arc::new(store) + .put(repository.into(), address, fragment, None, false) + .await + .expect_err("a hash alone should not attach content to a partition") + .is_internal() + ); + } + + /// An object left behind by a writer that died before publishing must not wedge the hash. + /// After waiting out the grace period, the key is taken over with bytes this writer can + /// vouch for — rather than trying to work out what the abandoned ones were. + #[tokio::test] + async fn test_put_immutable_reclaims_abandoned_object() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + + mock_fragments_probe_repeated(&mut dynamodb_mock, &entry); + // Nothing is ever published by the writer that left the object behind. + mock_metadata_probe_repeated(&mut dynamodb_mock, address.hash, None); + + // The conditional upload refuses to replace the abandoned object. + s3mock + .expect_put_object_if_absent::>() + .times(1) + .returning(|_, _, _| Err(precondition_failed())); + + // It is identified, then replaced only if it is still that same object. + s3mock.expect_head_object().times(1).returning(|_, _| { + Ok(HeadObjectOutput::builder() + .e_tag(ABANDONED_ETAG) + .last_modified(DateTime::from_secs(1)) + .build()) + }); + s3mock + .expect_put_object_if_match::>() + .times(1) + .withf(|_, _, _, etag| etag == ABANDONED_ETAG) + .returning(|_, _, _, _| Ok(PutObjectOutput::builder().build())); + + // This writer's own fragment is published, because these are now its own bytes. + mock_publish_metadata(&mut dynamodb_mock, address.hash, fragment); + mock_associate_fragment(&mut dynamodb_mock, &entry); + + let store = initialize_immutable_store_with_grace(s3mock, dynamodb_mock, 100).await; + + Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect("should have reclaimed the abandoned object"); + } + + /// An object stored moments ago belongs to a writer that is most likely still finishing. + /// Its representation is the one that will win, so this writer must leave it alone and hand + /// the wait back to the caller rather than reclaiming it or holding the request. + #[tokio::test] + async fn test_put_immutable_backs_off_from_an_unpublished_upload() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + + mock_fragments_probe_repeated(&mut dynamodb_mock, &entry); + mock_metadata_probe_repeated(&mut dynamodb_mock, address.hash, None); + + s3mock + .expect_put_object_if_absent::>() + .times(1) + .returning(|_, _, _| Err(precondition_failed())); + + // Stored just now. No reclaim is mocked: taking it over would be a test failure. + let stored_at = DateTime::from_millis(now_millis() as i64); + s3mock.expect_head_object().times(1).returning(move |_, _| { + Ok(HeadObjectOutput::builder() + .e_tag(ABANDONED_ETAG) + .last_modified(stored_at) + .build()) + }); + + let store = initialize_immutable_store_with_grace(s3mock, dynamodb_mock, 60_000).await; + + let error = Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect_err("should have backed off rather than reclaiming a live upload"); + + assert!(error.is_slow_down(), "expected a back-off, got {error:?}"); + } + + /// Two writers can reach recovery for the same abandoned object. Conditioning the reclaim on + /// the object being unchanged makes it single-winner: the loser must publish nothing, or it + /// would describe the winner's bytes with its own fragment. + #[tokio::test] + async fn test_put_immutable_losing_the_reclaim_publishes_nothing() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + + mock_fragments_probe_repeated(&mut dynamodb_mock, &entry); + mock_metadata_probe_repeated(&mut dynamodb_mock, address.hash, None); + + // No publish is mocked: writing metadata after losing the reclaim would describe another + // writer's bytes, so any conditional put here is a test failure. + s3mock + .expect_put_object_if_absent::>() + .times(PUT_MAX_ATTEMPTS) + .returning(|_, _, _| Err(precondition_failed())); + s3mock + .expect_head_object() + .times(PUT_MAX_ATTEMPTS) + .returning(|_, _| { + Ok(HeadObjectOutput::builder() + .e_tag(ABANDONED_ETAG) + .last_modified(DateTime::from_secs(1)) + .build()) + }); + + // Another writer reclaimed it first every time, so the entity tag never matches. + s3mock + .expect_put_object_if_match::>() + .times(PUT_MAX_ATTEMPTS) + .returning(|_, _, _, _| Err(precondition_failed())); + + let store = initialize_immutable_store_with_grace(s3mock, dynamodb_mock, 100).await; + + let error = Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect_err("losing every reclaim should not report success"); + + assert!(error.is_slow_down(), "expected a back-off, got {error:?}"); + } + + /// A writer whose upload was conditional cannot know its bytes are still the stored ones — + /// they may have been reclaimed while it was stalled. So a rejected publish must make it + /// stand down and re-probe, never overwrite the other writer's metadata with its own + /// fragment. + #[tokio::test] + async fn test_put_immutable_conditional_writer_does_not_publish_over_a_reclaimer() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + let reclaimer = Fragment { + flags: FragmentFlags::PayloadCompressedZstd.bits(), + size_payload: fragment.size_payload / 2, + size_content: fragment.size_content, + }; + + mock_fragments_probe_repeated(&mut dynamodb_mock, &entry); + // Nothing stored at probe time; by publish time a reclaimer owns the hash. + mock_metadata_probe_sequence( + &mut dynamodb_mock, + address.hash, + vec![ + None, + Some(FragmentMetadataEntry::new(address.hash).with_fragment(reclaimer)), + ], + ); + + s3mock + .expect_put_object_if_absent::>() + .times(1) + .returning(|_, _, _| Ok(PutObjectOutput::builder().build())); + + // Exactly one publish attempt. A second would be this writer overwriting the reclaimer's + // metadata with a fragment describing bytes that are no longer stored. + dynamodb_mock + .expect_put_item_conditional::() + .times(1) + .withf(|_, _, condition| *condition == MetadataWriteCondition::Absent) + .returning(|_, _, _| Err(conditional_check_failed())); + + // Standing down leads to deduplicating against what the reclaimer published. + mock_associate_fragment(&mut dynamodb_mock, &entry); + + let store = initialize_immutable_store(s3mock, dynamodb_mock).await; + + Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect("should have stood down and deduplicated"); + } + + /// The payload is already durable under a different partition, so this put records a + /// reference to it instead of uploading anything. No S3 call and no metadata write: the + /// stored representation is adopted as-is. + #[tokio::test] + async fn test_put_immutable_deduplicates_across_partitions() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + // A bare S3 mock with no expectations: any upload here is a test failure. + let s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + + // This partition has no association yet... + mock_lookup_fragments( + &mut dynamodb_mock, + entry.clone(), + StoreMatch::MatchFull, + StoreMatch::MatchNone, + ); + + // ...but the payload is already committed, so it is durable in S3. + mock_metadata_probe( + &mut dynamodb_mock, + address.hash, + Some(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)), + ); + + mock_associate_fragment(&mut dynamodb_mock, &entry); + + let store = initialize_immutable_store(s3mock, dynamodb_mock).await; + + Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect("failed to deduplicate against durable content"); + } + + /// A put whose exact association already exists against committed content writes nothing at + /// all: no S3, no metadata, not even an association. + #[tokio::test] + async fn test_put_immutable_full_match_writes_nothing() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + + mock_lookup_fragments( + &mut dynamodb_mock, + entry, + StoreMatch::MatchFull, + StoreMatch::MatchFull, + ); + mock_metadata_probe( + &mut dynamodb_mock, + address.hash, + Some(FragmentMetadataEntry::new(address.hash).with_fragment(fragment)), + ); + + let store = initialize_immutable_store(s3mock, dynamodb_mock).await; + + Arc::new(store) + .put(repository.into(), address, fragment, Some(payload), false) + .await + .expect("failed to put an already stored fragment"); + } + + #[tokio::test] + async fn test_put_immutable_extra_data() { + let repository = random::(); + let (fragment, address, payload) = fragment::generate_random(); + + let mut s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let entry = FragmentsEntry::new(repository, address); + + // Mock the list objects calls that `put_immutable` makes when querying for an object. + mock_lookup_fragments( + &mut dynamodb_mock, + entry.clone(), + StoreMatch::MatchFull, + StoreMatch::MatchNone, + ); + + mock_metadata_probe(&mut dynamodb_mock, address.hash, None); + mock_publish_metadata(&mut dynamodb_mock, address.hash, fragment); + + mock_associate_fragment(&mut dynamodb_mock, &entry); + + let mut body = vec![]; + body.extend_from_slice(payload.as_ref()); + + let real_len = body.len(); + + let extra = random::<[u8; 32]>(); + body.extend_from_slice(extra.as_slice()); + + // Ensure we only write bytes equal to the actual payload size, regardless of how much extra + // was sent. let expected = body[..real_len].to_vec(); s3mock - .expect_put_object() + .expect_put_object_if_absent() .with(eq(BUCKET), eq(address.hash.to_string()), eq(expected)) .return_once(move |_, _, _: Vec| Ok(PutObjectOutput::builder().build())); @@ -2670,7 +4305,7 @@ mod test { let (_, address, payload) = fragment::generate_random(); let repository = random::(); let fragment = Fragment { - flags: FragmentFlags::PayloadObliterating.bits(), + flags: FragmentFlags::PayloadObliterated.bits(), size_payload: 0, size_content: 0, }; @@ -2889,13 +4524,30 @@ mod test { false, /* fail */ ); + // Another obliteration owns the mark, so the payload and the metadata are its to decide. + // Removing this partition's own reference is still this call's job, and neither S3 nor a + // metadata write is mocked: touching either here would be a test failure. + mock_remove_association( + &mut dynamodb_mock, + repository, + address, + false, /* fail */ + ); + let store = initialize_immutable_store(s3mock, dynamodb_mock).await; - let stats = Default::default(); + let stats: Arc = Default::default(); Arc::new(store) - .obliterate(repository.into(), address, stats) + .obliterate(repository.into(), address, stats.clone()) .await .expect("obliterate failed"); + + assert_eq!( + stats.num_fragments.load(Ordering::Relaxed), + 1, + "the reference this call was asked to remove must still be removed" + ); + assert_eq!(stats.num_payloads.load(Ordering::Relaxed), 0); } #[tokio::test] @@ -3152,6 +4804,65 @@ mod test { }); } + /// The reference being obliterated must not be removable while a put could write it straight + /// back. Marking the row first is what prevents that: a put reading the mark backs off, so + /// the partition the content is being deleted for cannot restore its own reference and leave + /// the obliteration reporting success over content that is still referenced. + /// + /// Pins the ordering rather than the race, since the race is what the ordering rules out. + #[tokio::test] + async fn test_obliterate_marks_before_removing_the_association() { + let repository = random::(); + + let s3mock = MockS3Impl::default(); + let mut dynamodb_mock = MockDynamoDb::default(); + + let (fragment, address) = + mock_load_fragment_metadata(&mut dynamodb_mock, None, false /* fail */); + + let mut marked = fragment; + marked.flags |= FragmentFlags::PayloadObliterating; + let mut order = mockall::Sequence::default(); + + dynamodb_mock + .expect_put_item_conditional::() + .times(1) + .in_sequence(&mut order) + .withf(move |table, _, condition| { + table.as_ref() == METADATA_TABLE_NAME + && *condition == UpdateMetadataCondition(fragment) + }) + .returning(|_, _, _| Ok(PutItemOutput::builder().build())); + + dynamodb_mock + .expect_delete_item() + .times(1) + .in_sequence(&mut order) + .withf(move |table, _| table.as_ref() == FRAGMENTS_TABLE_NAME) + .returning(|_, _| Ok(DeleteItemOutput::builder().build())); + + // A reference remains, so the mark is cleared and nothing else happens. Reaching this at + // all means the mark was already in place when the association was removed. + mock_count_associations(&mut dynamodb_mock, address.hash, 1, false /* fail */); + + dynamodb_mock + .expect_put_item_conditional::() + .times(1) + .in_sequence(&mut order) + .withf(move |_, _, condition| *condition == UpdateMetadataCondition(marked)) + .returning(|_, _, _| Ok(PutItemOutput::builder().build())); + + let store = initialize_immutable_store(s3mock, dynamodb_mock).await; + + let stats: Arc = Default::default(); + Arc::new(store) + .obliterate(repository.into(), address, stats.clone()) + .await + .expect("obliterate should leave a still-referenced payload alone"); + + assert_eq!(stats.num_payloads.load(Ordering::Relaxed), 0); + } + #[tokio::test] #[traced_test] async fn test_obliterate_single_fragment() { @@ -3864,6 +5575,16 @@ mod test { .build()) }); + // The parent's own association is now removed before its sub-fragments are visited, so + // that they are only obliterated once the parent is known to be going away. + mock_remove_association( + &mut dynamodb_mock, + repository, + address, + false, /* fail */ + ); + mock_count_associations(&mut dynamodb_mock, address.hash, 0, false /* fail */); + mock_acquire_obliterate_lock( &mut dynamodb_mock, fragment, @@ -3889,8 +5610,8 @@ mod test { // mocks. assert_eq!( stats.num_fragments.load(Ordering::Relaxed), - // We deleted associations for both sub-fragments, but not the parent fragment - SUB_FRAGMENT_COUNT as usize + // Associations for both sub-fragments and the parent + (SUB_FRAGMENT_COUNT + 1) as usize ); assert_eq!( stats.num_payloads.load(Ordering::Relaxed), @@ -4029,4 +5750,737 @@ mod test { .await .expect("copy should succeed"); } + + /// Concurrency tests that drive the real store against an in-memory stand-in for `DynamoDB` + /// and S3. + /// + /// The fake implements the parts of the semantics the protocol actually leans on: per-object + /// and per-item atomicity, conditional puts, and write-once objects. That is what lets these + /// tests answer the question they exist for — whether concurrent writers holding the same + /// content in *different* representations can leave the stored blob and the metadata + /// describing it disagreeing, which would make every later read of that hash fail. + mod concurrency { + use std::collections::HashSet; + use std::sync::Mutex; + + use super::*; + + /// Number of writers racing for the same hash. + const WRITERS: u8 = 6; + /// Base compressed length; each writer's representation is a different length. + const BASE_PAYLOAD_LEN: usize = 64; + /// Uncompressed size, shared by every representation because it is the same content. + const CONTENT_SIZE: u64 = 4096; + + /// The stored state for a single hash, which is all these races need. + /// + /// Guarded by one mutex, so each operation is atomic with respect to the others — the + /// property S3 gives per object and `DynamoDB` per item, and the only one these tests + /// need, since the corruption they cover comes from two separately atomic writes to two + /// independent stores rather than from a torn individual write. + #[derive(Default)] + struct FakeState { + metadata: Option>, + object: Option>, + associations: HashSet, + uploads: usize, + /// Bumped on every write, so the entity tag changes exactly when the object does. + generation: u64, + stored_at_millis: u64, + /// When set, the next metadata write fails, standing in for the throttling that + /// corrupts the store before this change. + fail_next_metadata_write: bool, + /// When set, conditional metadata writes always succeed and uploads overwrite freely, + /// modelling the unguarded write this store used before S3 arbitrated between + /// writers. Used by the negative control to show what the conditions actually buy. + unprotected: bool, + } + + impl FakeState { + fn etag(&self) -> String { + format!("\"generation-{}\"", self.generation) + } + + fn store(&mut self, body: Vec) { + self.object = Some(body); + self.generation += 1; + self.stored_at_millis = now_millis(); + self.uploads += 1; + } + + /// Evaluate a publish condition against the current row. + fn permits(&self, condition: &MetadataWriteCondition) -> bool { + if self.unprotected { + return true; + } + + match condition { + MetadataWriteCondition::Absent => self.metadata.is_none(), + MetadataWriteCondition::Unchanged(expected) => { + self.metadata.as_ref().is_some_and(|row| { + row.get("flags") == Some(&AttributeValue::N(expected.flags.to_string())) + && row.get("size_payload") + == Some(&AttributeValue::N(expected.size_payload.to_string())) + && row.get("size_content") + == Some(&AttributeValue::N(expected.size_content.to_string())) + }) + } + } + } + } + + /// Identity of an association, taken from the sort key of a fragments row. + fn association_key(item: &HashMap) -> String { + format!("{:?}", item.get(FRAGMENTS_DYNAMO_SORT_KEY_ATTRIBUTE)) + } + + /// Writer `index`'s representation of the shared content: a distinct codec, a distinct + /// compressed length, and bytes that identify which writer produced them. + fn representation(index: u8) -> (Fragment, Bytes) { + let len = BASE_PAYLOAD_LEN + usize::from(index); + let codec = match index % 3 { + 0 => FragmentFlags::PayloadCompressedLZ4, + 1 => FragmentFlags::PayloadCompressedZstd, + _ => FragmentFlags::PayloadCompressedOodle2, + }; + + let fragment = Fragment { + flags: codec.bits(), + size_payload: len as u32, + size_content: CONTENT_SIZE, + }; + + (fragment, Bytes::from(vec![index; len])) + } + + /// Build a store whose `DynamoDB` and S3 calls are served by `state`. + async fn fake_store(state: Arc>) -> Arc { + let mut dynamodb = MockDynamoDb::default(); + let mut s3 = MockS3Impl::default(); + + let get_state = state.clone(); + dynamodb + .expect_get_item() + .returning(move |table, key, _consistent| { + let state = get_state.lock().unwrap(); + let item = if table.as_ref() == METADATA_TABLE_NAME { + state.metadata.clone() + } else if state.associations.contains(&association_key(&key)) { + Some(key) + } else { + None + }; + + Ok(GetItemOutput::builder().set_item(item).build()) + }); + + let put_state = state.clone(); + dynamodb.expect_put_item().returning(move |table, item| { + assert_eq!( + table.as_ref(), + FRAGMENTS_TABLE_NAME, + "metadata must only ever be written conditionally" + ); + put_state + .lock() + .unwrap() + .associations + .insert(association_key(&item)); + + Ok(PutItemOutput::builder().build()) + }); + + let conditional_state = state.clone(); + dynamodb + .expect_put_item_conditional::() + .returning(move |table, item, condition| { + assert_eq!(table.as_ref(), METADATA_TABLE_NAME); + let mut state = conditional_state.lock().unwrap(); + + if state.fail_next_metadata_write { + state.fail_next_metadata_write = false; + + return Err(aws_error( + PutItemError::ProvisionedThroughputExceededException( + ProvisionedThroughputExceededException::builder().build(), + ), + 400u16, + )); + } + + if !state.permits(&condition) { + return Err(aws_error( + PutItemError::ConditionalCheckFailedException( + ConditionalCheckFailedException::builder() + .set_item(state.metadata.clone()) + .build(), + ), + 400u16, + )); + } + + state.metadata = Some(item); + + Ok(PutItemOutput::builder().build()) + }); + + let upload_state = state.clone(); + s3.expect_put_object_if_absent::>() + .returning(move |_bucket, _key, body| { + let mut state = upload_state.lock().unwrap(); + + // Write-once, as S3 enforces for a conditional put: an existing object is + // never silently replaced. + if state.object.is_some() && !state.unprotected { + return Err(aws_error( + PutObjectError::generic( + ErrorMetadata::builder().code("PreconditionFailed").build(), + ), + 412u16, + )); + } + + state.store(body); + + Ok(PutObjectOutput::builder().build()) + }); + + let read_state = state.clone(); + s3.expect_get_object() + .returning(move |_bucket, _key, _range| { + let object = read_state.lock().unwrap().object.clone(); + + Ok(GetObjectOutput::builder() + .set_body(object.map(Into::into)) + .build()) + }); + + let head_state = state.clone(); + s3.expect_head_object().returning(move |_bucket, _key| { + let state = head_state.lock().unwrap(); + + Ok(HeadObjectOutput::builder() + .e_tag(state.etag()) + .last_modified(DateTime::from_millis(state.stored_at_millis as i64)) + .build()) + }); + + // Reclaiming succeeds only while the object is still the one that was inspected, + // which is what makes it single-winner. + let reclaim_state = state.clone(); + s3.expect_put_object_if_match::>().returning( + move |_bucket, _key, body, etag| { + let mut state = reclaim_state.lock().unwrap(); + + if state.etag() != etag { + return Err(aws_error( + PutObjectError::generic( + ErrorMetadata::builder().code("PreconditionFailed").build(), + ), + 412u16, + )); + } + + state.store(body); + + Ok(PutObjectOutput::builder().build()) + }, + ); + + let delete_state = state.clone(); + s3.expect_delete_object() + .returning(move |_bucket, _key, _version| { + delete_state.lock().unwrap().object = None; + + Ok(DeleteObjectOutput::builder().build()) + }); + + Arc::new(initialize_immutable_store(s3, dynamodb).await) + } + + /// Describe how the stored blob and the published metadata disagree, or `None` when they + /// are a matched pair. + /// + /// This is the invariant everything here exists to protect: whatever metadata is + /// published must describe exactly the bytes sitting in S3, because a reader fetches the + /// two independently and fails if they do not line up. + fn cohesion_violation(state: &FakeState) -> Option { + let row = state.metadata.as_ref()?; + let entry: FragmentMetadataEntry = serde_dynamo::from_item(row.clone()).unwrap(); + + let published = entry.fragment?; + + let Some(object) = state.object.as_ref() else { + return Some("metadata was published with no blob in S3".to_string()); + }; + + if published.size_payload as usize != object.len() { + return Some(format!( + "published size_payload {} does not match the stored blob length {}", + published.size_payload, + object.len() + )); + } + + // Every writer fills its payload with its own index, so the blob names the writer + // whose bytes actually survived. The published metadata must be that writer's. + let writer = object[0]; + if !object.iter().all(|byte| *byte == writer) { + return Some("the stored blob mixes bytes from more than one writer".to_string()); + } + + let (expected, _) = representation(writer); + if published.flags != expected.flags { + return Some(format!( + "writer {writer}'s blob was published with flags {:#x} instead of {:#x}", + published.flags, expected.flags + )); + } + if published.size_content != expected.size_content { + return Some("published size_content does not match the stored blob".to_string()); + } + + None + } + + fn assert_blob_and_metadata_agree(state: &FakeState, context: &str) { + if let Some(violation) = cohesion_violation(state) { + panic!("{context}: {violation}"); + } + } + + /// Many writers race to store the same content in different representations. Whatever + /// ends up published has to describe the bytes that actually landed in S3. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn concurrent_writers_with_different_representations_stay_cohesive() { + for iteration in 0..64 { + let state = Arc::new(Mutex::new(FakeState::default())); + let store = fake_store(state.clone()).await; + let (_, shared, _) = fragment::generate_random(); + + let mut writers = JoinSet::new(); + for index in 0..WRITERS { + let store = store.clone(); + // Different partitions and contexts, same content: exactly the case + // deduplication is meant to collapse. + let partition = random::(); + let address = address_with_random_context(shared); + let (fragment, payload) = representation(index); + + lore_base::lore_spawn!(writers, async move { + store + .put(partition, address, fragment, Some(payload), false) + .await + }); + } + + let mut stored = 0; + while let Some(writer) = writers.join_next().await { + match writer.expect("writer panicked") { + Ok(()) => stored += 1, + // Backing off is a legitimate outcome: another writer's upload is in + // flight, and the caller retries rather than racing a second + // representation into the same object. + Err(error) => assert!( + error.is_slow_down(), + "iteration {iteration}: unexpected error {error:?}" + ), + } + } + + let state = state.lock().unwrap(); + assert_blob_and_metadata_agree(&state, &format!("iteration {iteration}")); + assert!( + stored >= 1, + "iteration {iteration}: every writer backed off, so none made progress" + ); + assert_eq!( + state.uploads, 1, + "iteration {iteration}: the payload should be uploaded exactly once no \ + matter how many writers raced for it" + ); + } + } + + /// Negative control for the race test above: the same writers, against a store whose + /// metadata writes always succeed and whose objects can be overwritten — which is how + /// this store behaved before S3 arbitrated between writers. + /// + /// With nothing serialising them, writers that probed before anyone published all go on + /// to upload their own representation of the hash. That is precisely the precondition + /// for tearing: once more than one writer stores bytes, which representation wins S3 is + /// decided independently of which one wins the metadata table, so the two can disagree. + /// Tearing itself is a race and does not surface on every run, so the assertion here is + /// on the collision; [`legacy_write_order_tears_blob_and_metadata`] pins down the torn + /// outcome deterministically. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn without_metadata_conditions_writers_upload_over_each_other() { + let mut collisions = 0; + let mut tears = 0; + + for _ in 0..64 { + let state = Arc::new(Mutex::new(FakeState { + unprotected: true, + ..Default::default() + })); + let store = fake_store(state.clone()).await; + let (_, shared, _) = fragment::generate_random(); + + let mut writers = JoinSet::new(); + for index in 0..WRITERS { + let store = store.clone(); + let partition = random::(); + let address = address_with_random_context(shared); + let (fragment, payload) = representation(index); + + lore_base::lore_spawn!(writers, async move { + store + .put(partition, address, fragment, Some(payload), false) + .await + }); + } + + while let Some(writer) = writers.join_next().await { + let _ = writer.expect("writer panicked"); + } + + let state = state.lock().unwrap(); + if state.uploads > 1 { + collisions += 1; + } + if cohesion_violation(&state).is_some() { + tears += 1; + } + } + + assert!( + collisions > 0, + "without the metadata conditions nothing serialises the writers, so competing \ + uploads for a single hash were expected (collisions: {collisions}, torn: {tears})" + ); + } + + /// One partition stores the content; another later stores the *same* content in a + /// different representation. The second put must leave the stored blob and its metadata + /// completely alone, because overwriting the blob would strand the first partition's + /// content behind metadata that no longer describes it. + /// + /// This needs no concurrency at all — it is two sequential puts from different + /// partitions. + #[tokio::test] + async fn cross_partition_put_of_another_representation_leaves_the_blob_alone() { + let state = Arc::new(Mutex::new(FakeState::default())); + let store = fake_store(state.clone()).await; + let (_, shared, _) = fragment::generate_random(); + + // The first partition stores the content as one representation. + let (first, first_payload) = representation(0); + store + .clone() + .put( + random::(), + address_with_random_context(shared), + first, + Some(first_payload.clone()), + false, + ) + .await + .expect("the first partition should store the content"); + + let (blob, metadata) = { + let state = state.lock().unwrap(); + (state.object.clone(), state.metadata.clone()) + }; + assert_eq!( + blob.as_deref(), + Some(first_payload.as_ref()), + "the first partition's bytes should be the ones stored" + ); + + // A second partition stores the same content compressed differently. + let (second, second_payload) = representation(1); + assert_ne!( + first.size_payload, second.size_payload, + "the two partitions must hold genuinely different representations" + ); + + store + .put( + random::(), + address_with_random_context(shared), + second, + Some(second_payload), + false, + ) + .await + .expect("the second partition should deduplicate against the stored content"); + + let state = state.lock().unwrap(); + assert_eq!( + state.object, blob, + "the stored blob must be untouched by the second partition" + ); + assert_eq!( + state.metadata, metadata, + "the published metadata must be untouched by the second partition" + ); + assert_blob_and_metadata_agree(&state, "after a cross-partition put"); + assert_eq!( + state.uploads, 1, + "the second partition must not upload its own representation over the first's" + ); + } + + /// The sequence that corrupts the store without this change, end to end. + /// + /// One partition stores content and reads it back. A second partition stores the same + /// content compressed differently, with the next metadata write armed to fail — the + /// throttling that previously left the first partition reading a blob its own metadata + /// no longer described. + /// + /// Nothing fails now, because nothing is written: the second partition records a + /// reference to content that is already stored, so there is no upload to replace the + /// blob and no metadata write to lose. + #[tokio::test] + async fn a_failed_cross_partition_write_cannot_corrupt_the_first_partition() { + let state = Arc::new(Mutex::new(FakeState::default())); + let store = fake_store(state.clone()).await; + let (_, shared, _) = fragment::generate_random(); + + let first_partition = random::(); + let first_address = address_with_random_context(shared); + let (first_fragment, first_payload) = representation(0); + + store + .clone() + .put( + first_partition, + first_address, + first_fragment, + Some(first_payload.clone()), + false, + ) + .await + .expect("the first partition should store the content"); + + store + .clone() + .get(first_partition, first_address, StoreMatch::MatchFull) + .await + .expect("the first partition should be able to read what it stored"); + + // Arm the write that used to be lost, then have another partition store the same + // content in its own representation. + state.lock().unwrap().fail_next_metadata_write = true; + + let (second_fragment, second_payload) = representation(1); + store + .clone() + .put( + random::(), + address_with_random_context(shared), + second_fragment, + Some(second_payload), + false, + ) + .await + .expect("the second partition should deduplicate against the stored content"); + + { + let state = state.lock().unwrap(); + assert!( + state.fail_next_metadata_write, + "the armed metadata write was consumed, so the second partition published \ + metadata it had no business publishing" + ); + assert_eq!( + state.uploads, 1, + "the second partition uploaded content that was already stored" + ); + } + + store + .get(first_partition, first_address, StoreMatch::MatchFull) + .await + .expect("the first partition must still be able to read what it stored"); + } + + /// Negative control for the test above, modelling the write order this store used before + /// deduplication: a put with no exact match uploaded unconditionally, then published its + /// own metadata. A second partition storing the same content in a different + /// representation therefore replaced the blob while the first partition's metadata still + /// described it. + /// + /// The pair is inconsistent for the whole gap between those two writes — every read of + /// the hash fails, including reads from the partition that did not write — and stays that + /// way permanently if the writer dies before publishing. + #[tokio::test] + async fn legacy_cross_partition_overwrite_tears_blob_and_metadata() { + let state = Arc::new(Mutex::new(FakeState::default())); + let (first, first_payload) = representation(0); + let (second, second_payload) = representation(1); + let (_, shared, _) = fragment::generate_random(); + + // The first partition's consistent pair. + { + let mut state = state.lock().unwrap(); + state.object = Some(first_payload.to_vec()); + state.metadata = Some( + serde_dynamo::to_item( + FragmentMetadataEntry::new(shared.hash).with_fragment(first), + ) + .unwrap(), + ); + } + assert!( + cohesion_violation(&state.lock().unwrap()).is_none(), + "the first partition should start out consistent" + ); + + // The second partition uploads its representation over the top, as the old + // unconditional put did. + state.lock().unwrap().object = Some(second_payload.to_vec()); + + let torn = cohesion_violation(&state.lock().unwrap()); + assert!( + torn.is_some(), + "overwriting the blob should have left it disagreeing with the published metadata" + ); + + // Publishing the second representation restores consistency — so the damage lasts + // only as long as the gap, unless the writer never gets here. + state.lock().unwrap().metadata = Some( + serde_dynamo::to_item( + FragmentMetadataEntry::new(shared.hash).with_fragment(second), + ) + .unwrap(), + ); + assert!( + cohesion_violation(&state.lock().unwrap()).is_none(), + "publishing the second representation should have restored the pair" + ); + } + + /// Negative control for [`assert_blob_and_metadata_agree`]. + /// + /// Replays the write order this store used before this change — upload, then publish + /// metadata unconditionally — with the interleaving that made it unsafe. Both + /// writers upload and both publish, and because S3 and the metadata table are updated + /// independently, the writer that wins one is not the writer that wins the other. The + /// result is a hash whose published metadata describes bytes that are no longer there, + /// which fails every subsequent read. + /// + /// Without this, a cohesion assertion that never fires would look like a passing test. + #[tokio::test] + async fn legacy_write_order_tears_blob_and_metadata() { + let state = Arc::new(Mutex::new(FakeState::default())); + let (_, shared, _) = fragment::generate_random(); + + let (fragment_a, payload_a) = representation(0); + let (fragment_b, payload_b) = representation(1); + assert_ne!( + fragment_a.size_payload, fragment_b.size_payload, + "the two writers must hold genuinely different representations" + ); + + { + let mut state = state.lock().unwrap(); + + // Both uploads land, B's last... + state.object = Some(payload_a.to_vec()); + state.object = Some(payload_b.to_vec()); + + // ...but the unconditional metadata writes land in the opposite order, so the + // published fragment is A's while the stored bytes are B's. + for fragment in [fragment_b, fragment_a] { + state.metadata = Some( + serde_dynamo::to_item( + FragmentMetadataEntry::new(shared.hash).with_fragment(fragment), + ) + .unwrap(), + ); + } + } + + let state = state.lock().unwrap(); + let entry: FragmentMetadataEntry = + serde_dynamo::from_item(state.metadata.clone().unwrap()).unwrap(); + let published = entry + .fragment + .expect("the legacy order publishes a row describing a fragment"); + let object = state.object.as_ref().expect("a blob was uploaded"); + + // This is the exact condition `assert_blob_and_metadata_agree` checks first, so the + // cohesion assertion used by the tests above does detect this state. + assert_ne!( + published.size_payload as usize, + object.len(), + "the legacy write order should leave the published size describing another \ + writer's blob" + ); + } + + /// Once the content is durable, writers holding a *different* representation of it must + /// deduplicate against what is stored rather than republishing their own description of + /// it — otherwise the metadata would stop matching the blob. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn deduplicating_writers_adopt_the_stored_representation() { + let state = Arc::new(Mutex::new(FakeState::default())); + let store = fake_store(state.clone()).await; + let (_, shared, _) = fragment::generate_random(); + + // One writer stores the content. + let (fragment, payload) = representation(0); + store + .clone() + .put( + random::(), + address_with_random_context(shared), + fragment, + Some(payload), + false, + ) + .await + .expect("initial write should succeed"); + + let published = state.lock().unwrap().metadata.clone(); + + // Now everyone else stores the same content in their own representation. + let mut writers = JoinSet::new(); + for index in 1..WRITERS { + let store = store.clone(); + let partition = random::(); + let address = address_with_random_context(shared); + let (fragment, payload) = representation(index); + + lore_base::lore_spawn!(writers, async move { + store + .put(partition, address, fragment, Some(payload), false) + .await + }); + } + + while let Some(writer) = writers.join_next().await { + writer + .expect("writer panicked") + .expect("deduplicating against durable content should succeed"); + } + + let state = state.lock().unwrap(); + assert_eq!( + state.uploads, 1, + "deduplicating writers must not upload their own representation" + ); + assert_eq!( + state.metadata, published, + "deduplicating writers must not republish the payload metadata" + ); + assert_eq!( + state.associations.len(), + usize::from(WRITERS), + "every writer should have recorded its own association" + ); + assert_blob_and_metadata_agree(&state, "after deduplication"); + } + } }