Add exact resumable object source streaming

crates/remora-object-source/src/body.rs

@@ -0,0 +1,84 @@
+use bytes::Bytes;
+use futures_util::StreamExt;
+use opendal::FuturesBytesStream;
+use remora_stream::{SourceBody, StreamError, StreamSide};
+
+pub struct ObjectSourceBody {
+    stream: FuturesBytesStream,
+    pending: Bytes,
+    pending_offset: usize,
+    remaining: u64,
+    finished: bool,
+}
+
+impl ObjectSourceBody {
+    pub(crate) fn new(stream: FuturesBytesStream, expected_length: u64) -> Self {
+        Self {
+            stream,
+            pending: Bytes::new(),
+            pending_offset: 0,
+            remaining: expected_length,
+            finished: false,
+        }
+    }
+
+    fn copy_pending(&mut self, output: &mut [u8]) -> usize {
+        let available = &self.pending[self.pending_offset..];
+        let count = available.len().min(output.len());
+        output[..count].copy_from_slice(&available[..count]);
+        self.pending_offset += count;
+        self.remaining -= count as u64;
+        if self.pending_offset == self.pending.len() {
+            self.pending = Bytes::new();
+            self.pending_offset = 0;
+        }
+        count
+    }
+}
+
+impl SourceBody for ObjectSourceBody {
+    async fn read(&mut self, output: &mut [u8]) -> Result<usize, StreamError> {
+        if output.is_empty() {
+            return Ok(0);
+        }
+        if self.pending_offset < self.pending.len() {
+            return Ok(self.copy_pending(output));
+        }
+        if self.finished {
+            return Ok(0);
+        }
+        match self.stream.next().await {
+            Some(Ok(chunk)) if chunk.len() as u64 <= self.remaining => {
+                self.pending = chunk;
+                Ok(self.copy_pending(output))
+            }
+            Some(Ok(_)) => Err(StreamError::SourceChanged),
+            Some(Err(error)) => Err(map_stream_error(&error)),
+            None if self.remaining != 0 => Err(StreamError::transient_source(
+                "object source ended before its declared length",
+            )),
+            None => {
+                self.finished = true;
+                Ok(0)
+            }
+        }
+    }
+}
+
+fn map_stream_error(error: &std::io::Error) -> StreamError {
+    let kind = error
+        .get_ref()
+        .and_then(|source| source.downcast_ref::<opendal::Error>())
+        .map(opendal::Error::kind);
+    match kind {
+        Some(opendal::ErrorKind::PermissionDenied) => StreamError::Authorization {
+            side: StreamSide::Source,
+        },
+        Some(
+            opendal::ErrorKind::ConditionNotMatch
+            | opendal::ErrorKind::NotFound
+            | opendal::ErrorKind::RangeNotSatisfied,
+        ) => StreamError::SourceChanged,
+        _ => StreamError::transient_source("object source read failed"),
+    }
+}

crates/remora-object-source/src/config.rs

@@ -0,0 +1,84 @@
+use std::fmt;
+
+use remora_outbound::CanonicalUrl;
+use remora_target::TargetStoreConfig;
+
+use crate::ObjectSourceError;
+
+#[derive(Clone, Eq, PartialEq)]
+pub struct ObjectSourceConfig {
+    endpoint: CanonicalUrl,
+    bucket: String,
+    region: String,
+    access_key_id: String,
+    secret_access_key: String,
+}
+
+impl ObjectSourceConfig {
+    pub fn new(
+        endpoint: &str,
+        bucket: impl Into<String>,
+        region: impl Into<String>,
+        access_key_id: impl Into<String>,
+        secret_access_key: impl Into<String>,
+    ) -> Result<Self, ObjectSourceError> {
+        let endpoint =
+            CanonicalUrl::parse(endpoint).map_err(|_| ObjectSourceError::InvalidRequest)?;
+        let bucket = bucket.into();
+        let region = region.into();
+        let access_key_id = access_key_id.into();
+        let secret_access_key = secret_access_key.into();
+        if bucket.is_empty() || access_key_id.is_empty() || secret_access_key.is_empty() {
+            return Err(ObjectSourceError::InvalidRequest);
+        }
+        Ok(Self {
+            endpoint,
+            bucket,
+            region,
+            access_key_id,
+            secret_access_key,
+        })
+    }
+
+    pub fn from_target(config: &TargetStoreConfig) -> Result<Self, ObjectSourceError> {
+        Self::new(
+            &config.endpoint,
+            config.bucket.clone(),
+            config.region.clone(),
+            config.access_key_id.clone(),
+            config.secret_access_key.clone(),
+        )
+    }
+
+    #[must_use]
+    pub fn endpoint(&self) -> &CanonicalUrl {
+        &self.endpoint
+    }
+
+    #[must_use]
+    pub fn bucket(&self) -> &str {
+        &self.bucket
+    }
+
+    pub(crate) fn region(&self) -> &str {
+        &self.region
+    }
+
+    pub(crate) fn access_key_id(&self) -> &str {
+        &self.access_key_id
+    }
+
+    pub(crate) fn secret_access_key(&self) -> &str {
+        &self.secret_access_key
+    }
+}
+
+impl fmt::Debug for ObjectSourceConfig {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("ObjectSourceConfig")
+            .field("bucket", &self.bucket)
+            .field("region", &self.region)
+            .finish_non_exhaustive()
+    }
+}

crates/remora-object-source/src/error.rs

@@ -0,0 +1,83 @@
+use remora_stream::{StreamError, StreamSide};
+use remora_target::TargetError;
+
+#[derive(Clone, Copy, Debug, Eq, thiserror::Error, PartialEq)]
+pub enum ObjectSourceError {
+    #[error("object source request is invalid")]
+    InvalidRequest,
+    #[error("object source access is denied")]
+    SourceAuthorization,
+    #[error("destination access is denied")]
+    DestinationAuthorization,
+    #[error("object source was not found")]
+    SourceNotFound,
+    #[error("object source changed")]
+    SourceChanged,
+    #[error("destination resume state changed")]
+    DestinationChanged,
+    #[error("object source is temporarily unavailable")]
+    TransientSource,
+    #[error("destination is temporarily unavailable")]
+    TransientDestination,
+    #[error("object source adapter failed")]
+    Internal,
+}
+
+impl ObjectSourceError {
+    #[must_use]
+    pub fn into_stream_error(self) -> StreamError {
+        match self {
+            Self::SourceAuthorization => StreamError::Authorization {
+                side: StreamSide::Source,
+            },
+            Self::DestinationAuthorization => StreamError::Authorization {
+                side: StreamSide::Target,
+            },
+            Self::SourceNotFound => StreamError::SourceNotFound,
+            Self::SourceChanged => StreamError::SourceChanged,
+            Self::TransientSource => StreamError::transient_source("object source unavailable"),
+            Self::TransientDestination => StreamError::transient_target("destination unavailable"),
+            Self::DestinationChanged => {
+                StreamError::internal("destination resume proof is no longer reusable")
+            }
+            Self::InvalidRequest | Self::Internal => {
+                StreamError::internal("object source contract violation")
+            }
+        }
+    }
+}
+
+pub(crate) fn map_open_error(error: opendal::Error) -> ObjectSourceError {
+    use opendal::ErrorKind;
+
+    match error.kind() {
+        ErrorKind::NotFound => ObjectSourceError::SourceNotFound,
+        ErrorKind::PermissionDenied => ObjectSourceError::SourceAuthorization,
+        ErrorKind::ConditionNotMatch | ErrorKind::RangeNotSatisfied => {
+            ObjectSourceError::SourceChanged
+        }
+        ErrorKind::RateLimited => ObjectSourceError::TransientSource,
+        ErrorKind::Unsupported | ErrorKind::ConfigInvalid => ObjectSourceError::InvalidRequest,
+        _ => ObjectSourceError::TransientSource,
+    }
+}
+
+pub(crate) fn map_resume_error(error: opendal::Error) -> ObjectSourceError {
+    match map_open_error(error) {
+        ObjectSourceError::SourceNotFound => ObjectSourceError::SourceChanged,
+        other => other,
+    }
+}
+
+pub(crate) fn map_destination_error(error: TargetError) -> ObjectSourceError {
+    match error {
+        TargetError::Authorization => ObjectSourceError::DestinationAuthorization,
+        TargetError::NotFound | TargetError::AlreadyExists | TargetError::Integrity(_) => {
+            ObjectSourceError::DestinationChanged
+        }
+        TargetError::Capacity | TargetError::Transient(_) => {
+            ObjectSourceError::TransientDestination
+        }
+        TargetError::Validation(_) | TargetError::Internal(_) => ObjectSourceError::Internal,
+    }
+}

crates/remora-object-source/src/identity.rs

@@ -0,0 +1,312 @@
+use std::fmt;
+
+use remora_domain::MemberPath;
+use remora_outbound::CanonicalUrl;
+use remora_stream::{IntegrityDigest, Sha256Accountant};
+use remora_target::{CompletedPart, ContentSha256, MultipartIdentity};
+
+use crate::ObjectSourceError;
+
+#[derive(Clone, Eq, Hash, PartialEq)]
+pub struct ObjectValidator(String);
+
+impl ObjectValidator {
+    pub fn parse(value: &str) -> Result<Self, ObjectSourceError> {
+        if value.is_empty() || value.len() > 1024 || value.chars().any(char::is_control) {
+            return Err(ObjectSourceError::InvalidRequest);
+        }
+        Ok(Self(value.to_owned()))
+    }
+
+    #[must_use]
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl fmt::Debug for ObjectValidator {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str("ObjectValidator([redacted])")
+    }
+}
+
+#[derive(Clone, Eq, PartialEq)]
+pub struct ObjectSourceIdentity {
+    endpoint: CanonicalUrl,
+    bucket: String,
+    key: String,
+    size: u64,
+    version_id: Option<String>,
+    validator: Option<ObjectValidator>,
+}
+
+impl ObjectSourceIdentity {
+    pub fn from_parts(
+        endpoint: CanonicalUrl,
+        bucket: impl Into<String>,
+        key: impl Into<String>,
+        size: u64,
+        version_id: Option<String>,
+        validator: Option<ObjectValidator>,
+    ) -> Result<Self, ObjectSourceError> {
+        let bucket = bucket.into();
+        let key = key.into();
+        let version_id = version_id.filter(|value| !value.is_empty() && value != "null");
+        if bucket.is_empty()
+            || bucket.chars().any(char::is_control)
+            || key.is_empty()
+            || key.starts_with('/')
+            || key.chars().any(char::is_control)
+            || version_id
+                .as_deref()
+                .is_some_and(|value| value.chars().any(char::is_control))
+        {
+            return Err(ObjectSourceError::InvalidRequest);
+        }
+        Ok(Self {
+            endpoint,
+            bucket,
+            key,
+            size,
+            version_id,
+            validator,
+        })
+    }
+
+    #[must_use]
+    pub fn endpoint(&self) -> &CanonicalUrl {
+        &self.endpoint
+    }
+
+    #[must_use]
+    pub fn bucket(&self) -> &str {
+        &self.bucket
+    }
+
+    #[must_use]
+    pub fn key(&self) -> &str {
+        &self.key
+    }
+
+    #[must_use]
+    pub fn size(&self) -> u64 {
+        self.size
+    }
+
+    #[must_use]
+    pub fn version_id(&self) -> Option<&str> {
+        self.version_id.as_deref()
+    }
+
+    #[must_use]
+    pub fn validator(&self) -> Option<&ObjectValidator> {
+        self.validator.as_ref()
+    }
+
+    #[must_use]
+    pub fn is_resumable(&self) -> bool {
+        self.version_id.is_some() || self.validator.is_some()
+    }
+}
+
+impl fmt::Debug for ObjectSourceIdentity {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("ObjectSourceIdentity")
+            .field("size", &self.size)
+            .field("versioned", &self.version_id.is_some())
+            .field("has_validator", &self.validator.is_some())
+            .finish_non_exhaustive()
+    }
+}
+
+#[derive(Clone, Eq, PartialEq)]
+pub struct DestinationResumeProof {
+    endpoint: CanonicalUrl,
+    identity: MultipartIdentity,
+    completed_part_count: u32,
+    expected_content_sha256: Option<ContentSha256>,
+    completed_bytes: u64,
+    completed_parts_integrity: IntegrityDigest,
+}
+
+impl DestinationResumeProof {
+    pub fn new(
+        endpoint: CanonicalUrl,
+        identity: MultipartIdentity,
+        completed_parts: Vec<CompletedPart>,
+        expected_content_sha256: Option<ContentSha256>,
+    ) -> Result<Self, ObjectSourceError> {
+        if identity.target_store_id.is_empty()
+            || identity.bucket.is_empty()
+            || identity.key.is_empty()
+            || identity.upload_id.is_empty()
+        {
+            return Err(ObjectSourceError::InvalidRequest);
+        }
+        let (completed_part_count, completed_bytes, completed_parts_integrity) =
+            summarize_parts(&completed_parts)?;
+        Ok(Self {
+            endpoint,
+            identity,
+            completed_part_count,
+            expected_content_sha256,
+            completed_bytes,
+            completed_parts_integrity,
+        })
+    }
+
+    #[must_use]
+    pub fn identity(&self) -> &MultipartIdentity {
+        &self.identity
+    }
+
+    #[must_use]
+    pub fn endpoint(&self) -> &CanonicalUrl {
+        &self.endpoint
+    }
+
+    #[must_use]
+    pub fn completed_part_count(&self) -> u32 {
+        self.completed_part_count
+    }
+
+    #[must_use]
+    pub fn expected_content_sha256(&self) -> Option<&ContentSha256> {
+        self.expected_content_sha256.as_ref()
+    }
+
+    #[must_use]
+    pub fn completed_bytes(&self) -> u64 {
+        self.completed_bytes
+    }
+
+    pub(crate) fn matches_parts(&self, parts: &[CompletedPart]) -> bool {
+        summarize_parts(parts).is_ok_and(|(count, bytes, integrity)| {
+            count == self.completed_part_count
+                && bytes == self.completed_bytes
+                && integrity == self.completed_parts_integrity
+        })
+    }
+}
+
+impl fmt::Debug for DestinationResumeProof {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("DestinationResumeProof")
+            .field("completed_part_count", &self.completed_part_count)
+            .field("completed_bytes", &self.completed_bytes)
+            .finish_non_exhaustive()
+    }
+}
+
+fn summarize_parts(
+    parts: &[CompletedPart],
+) -> Result<(u32, u64, IntegrityDigest), ObjectSourceError> {
+    if parts.is_empty() || parts.len() > u32::MAX as usize {
+        return Err(ObjectSourceError::InvalidRequest);
+    }
+    let mut previous = None;
+    let mut completed_bytes = 0_u64;
+    let mut accountant = Sha256Accountant::new();
+    for part in parts {
+        let size = part
+            .size
+            .filter(|size| *size > 0)
+            .ok_or(ObjectSourceError::InvalidRequest)?;
+        if part.part_number == 0
+            || previous.is_some_and(|number| part.part_number <= number)
+            || part.etag.is_empty()
+        {
+            return Err(ObjectSourceError::InvalidRequest);
+        }
+        previous = Some(part.part_number);
+        completed_bytes = completed_bytes
+            .checked_add(size)
+            .ok_or(ObjectSourceError::InvalidRequest)?;
+        let etag_length =
+            u64::try_from(part.etag.len()).map_err(|_| ObjectSourceError::InvalidRequest)?;
+        accountant.update(&part.part_number.to_be_bytes());
+        accountant.update(&size.to_be_bytes());
+        accountant.update(&etag_length.to_be_bytes());
+        accountant.update(part.etag.as_bytes());
+    }
+    let count = u32::try_from(parts.len()).map_err(|_| ObjectSourceError::InvalidRequest)?;
+    let integrity = accountant
+        .finalize()
+        .digest()
+        .ok_or(ObjectSourceError::Internal)?
+        .clone();
+    Ok((count, completed_bytes, integrity))
+}
+
+#[derive(Clone, Eq, PartialEq)]
+pub struct ObjectResumeCheckpoint {
+    source: ObjectSourceIdentity,
+    result_path: MemberPath,
+    offset: u64,
+    prefix_integrity: IntegrityDigest,
+    destination: DestinationResumeProof,
+}
+
+impl ObjectResumeCheckpoint {
+    pub fn new(
+        source: ObjectSourceIdentity,
+        result_path: MemberPath,
+        offset: u64,
+        prefix_integrity: IntegrityDigest,
+        destination: DestinationResumeProof,
+    ) -> Result<Self, ObjectSourceError> {
+        if !source.is_resumable()
+            || offset == 0
+            || offset >= source.size
+            || destination.completed_bytes != offset
+        {
+            return Err(ObjectSourceError::InvalidRequest);
+        }
+        Ok(Self {
+            source,
+            result_path,
+            offset,
+            prefix_integrity,
+            destination,
+        })
+    }
+
+    #[must_use]
+    pub fn source(&self) -> &ObjectSourceIdentity {
+        &self.source
+    }
+
+    #[must_use]
+    pub fn result_path(&self) -> &MemberPath {
+        &self.result_path
+    }
+
+    #[must_use]
+    pub fn offset(&self) -> u64 {
+        self.offset
+    }
+
+    #[must_use]
+    pub fn prefix_integrity(&self) -> &IntegrityDigest {
+        &self.prefix_integrity
+    }
+
+    #[must_use]
+    pub fn destination(&self) -> &DestinationResumeProof {
+        &self.destination
+    }
+}
+
+impl fmt::Debug for ObjectResumeCheckpoint {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("ObjectResumeCheckpoint")
+            .field("source", &self.source)
+            .field("result_path", &self.result_path.canonical())
+            .field("offset", &self.offset)
+            .field("destination", &self.destination)
+            .finish_non_exhaustive()
+    }
+}

crates/remora-object-source/src/lib.rs

@@ -0,0 +1,15 @@
+//! S3-compatible object source adapter with exact source and destination resume proof.
+
+mod body;
+mod config;
+mod error;
+mod identity;
+mod source;
+
+pub use body::ObjectSourceBody;
+pub use config::ObjectSourceConfig;
+pub use error::ObjectSourceError;
+pub use identity::{
+    DestinationResumeProof, ObjectResumeCheckpoint, ObjectSourceIdentity, ObjectValidator,
+};
+pub use source::{ObjectSource, ObjectSourceRequest, OpenedObjectSource, ValidatedObjectResume};

crates/remora-object-source/src/source.rs

@@ -0,0 +1,261 @@
+use std::fmt;
+
+use opendal::{Operator, services::S3};
+use remora_domain::MemberPath;
+use remora_stream::IntegrityDigest;
+use remora_target::{MultipartSession, TargetStore};
+
+use crate::{
+    ObjectResumeCheckpoint, ObjectSourceBody, ObjectSourceConfig, ObjectSourceError,
+    ObjectSourceIdentity, ObjectValidator,
+    error::{map_destination_error, map_open_error, map_resume_error},
+};
+
+pub struct ObjectSourceRequest {
+    key: String,
+    result_path: MemberPath,
+}
+
+impl ObjectSourceRequest {
+    pub fn new(key: impl Into<String>, result_path: &str) -> Result<Self, ObjectSourceError> {
+        let key = key.into();
+        if key.is_empty() || key.starts_with('/') || key.chars().any(char::is_control) {
+            return Err(ObjectSourceError::InvalidRequest);
+        }
+        Ok(Self {
+            key,
+            result_path: MemberPath::parse(result_path)
+                .map_err(|_| ObjectSourceError::InvalidRequest)?,
+        })
+    }
+}
+
+impl fmt::Debug for ObjectSourceRequest {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("ObjectSourceRequest")
+            .field("result_path", &self.result_path.canonical())
+            .finish_non_exhaustive()
+    }
+}
+
+pub struct OpenedObjectSource {
+    result_path: MemberPath,
+    identity: ObjectSourceIdentity,
+    body: ObjectSourceBody,
+}
+
+impl OpenedObjectSource {
+    #[must_use]
+    pub fn result_path(&self) -> &MemberPath {
+        &self.result_path
+    }
+
+    #[must_use]
+    pub fn identity(&self) -> &ObjectSourceIdentity {
+        &self.identity
+    }
+
+    #[must_use]
+    pub fn into_body(self) -> ObjectSourceBody {
+        self.body
+    }
+}
+
+impl fmt::Debug for OpenedObjectSource {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("OpenedObjectSource")
+            .field("result_path", &self.result_path.canonical())
+            .field("identity", &self.identity)
+            .finish_non_exhaustive()
+    }
+}
+
+pub struct ValidatedObjectResume {
+    source: OpenedObjectSource,
+    destination: MultipartSession,
+}
+
+impl fmt::Debug for ValidatedObjectResume {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("ValidatedObjectResume")
+            .field("source", &self.source)
+            .field("destination_validated", &true)
+            .finish_non_exhaustive()
+    }
+}
+
+impl ValidatedObjectResume {
+    #[must_use]
+    pub fn source(&self) -> &OpenedObjectSource {
+        &self.source
+    }
+
+    #[must_use]
+    pub fn destination(&self) -> &MultipartSession {
+        &self.destination
+    }
+
+    #[must_use]
+    pub fn into_parts(self) -> (OpenedObjectSource, MultipartSession) {
+        (self.source, self.destination)
+    }
+}
+
+pub struct ObjectSource {
+    config: ObjectSourceConfig,
+    operator: Operator,
+}
+
+impl ObjectSource {
+    pub fn connect(config: ObjectSourceConfig) -> Result<Self, ObjectSourceError> {
+        opendal::install_default();
+        let builder = S3::default()
+            .bucket(config.bucket())
+            .endpoint(config.endpoint().as_url().as_str())
+            .region(config.region())
+            .access_key_id(config.access_key_id())
+            .secret_access_key(config.secret_access_key())
+            .disable_config_load()
+            .disable_ec2_metadata();
+        let operator = Operator::new(builder).map_err(map_open_error)?;
+        Ok(Self { config, operator })
+    }
+
+    pub async fn open(
+        &self,
+        request: ObjectSourceRequest,
+    ) -> Result<OpenedObjectSource, ObjectSourceError> {
+        let metadata = self
+            .operator
+            .stat(&request.key)
+            .await
+            .map_err(map_open_error)?;
+        let identity = ObjectSourceIdentity::from_parts(
+            self.config.endpoint().clone(),
+            self.config.bucket().to_owned(),
+            request.key,
+            metadata.content_length(),
+            normalized_version(metadata.version()),
+            metadata.etag().map(ObjectValidator::parse).transpose()?,
+        )?;
+        let body = self.open_body(&identity, 0, false).await?;
+        Ok(OpenedObjectSource {
+            result_path: request.result_path,
+            identity,
+            body,
+        })
+    }
+
+    pub async fn resume(
+        &self,
+        checkpoint: &ObjectResumeCheckpoint,
+        observed_prefix: &IntegrityDigest,
+        target: &TargetStore,
+    ) -> Result<ValidatedObjectResume, ObjectSourceError> {
+        if checkpoint.prefix_integrity() != observed_prefix {
+            return Err(ObjectSourceError::SourceChanged);
+        }
+        self.ensure_identity(checkpoint.source())?;
+        let proof = checkpoint.destination();
+        let target_endpoint = remora_outbound::CanonicalUrl::parse(&target.config().endpoint)
+            .map_err(|_| ObjectSourceError::DestinationChanged)?;
+        if proof.endpoint() != &target_endpoint {
+            return Err(ObjectSourceError::DestinationChanged);
+        }
+        let (destination, live_parts) = target
+            .resume_multipart(
+                proof.identity().clone(),
+                proof.expected_content_sha256().cloned(),
+            )
+            .await
+            .map_err(map_destination_error)?;
+        if !proof.matches_parts(&live_parts) {
+            return Err(ObjectSourceError::DestinationChanged);
+        }
+        self.validate_current(checkpoint.source()).await?;
+        let body = self
+            .open_body(checkpoint.source(), checkpoint.offset(), true)
+            .await?;
+        Ok(ValidatedObjectResume {
+            source: OpenedObjectSource {
+                result_path: checkpoint.result_path().clone(),
+                identity: checkpoint.source().clone(),
+                body,
+            },
+            destination,
+        })
+    }
+
+    async fn validate_current(
+        &self,
+        identity: &ObjectSourceIdentity,
+    ) -> Result<(), ObjectSourceError> {
+        let mut stat = self.operator.stat_with(identity.key());
+        if let Some(version) = identity.version_id() {
+            stat = stat.version(version);
+        } else if let Some(validator) = identity.validator() {
+            stat = stat.if_match(validator.as_str());
+        } else {
+            return Err(ObjectSourceError::SourceChanged);
+        }
+        let metadata = stat.await.map_err(map_resume_error)?;
+        if metadata.content_length() != identity.size() {
+            return Err(ObjectSourceError::SourceChanged);
+        }
+        if identity.version_id().is_none()
+            && metadata.etag() != identity.validator().map(ObjectValidator::as_str)
+        {
+            return Err(ObjectSourceError::SourceChanged);
+        }
+        Ok(())
+    }
+
+    async fn open_body(
+        &self,
+        identity: &ObjectSourceIdentity,
+        offset: u64,
+        resume: bool,
+    ) -> Result<ObjectSourceBody, ObjectSourceError> {
+        let mut reader = self.operator.reader_with(identity.key()).chunk(256 * 1024);
+        if let Some(version) = identity.version_id() {
+            reader = reader.version(version);
+        } else if let Some(validator) = identity.validator() {
+            reader = reader.if_match(validator.as_str());
+        }
+        let reader = if resume {
+            reader.await.map_err(map_resume_error)?
+        } else {
+            reader.await.map_err(map_open_error)?
+        };
+        let stream = if resume {
+            reader
+                .into_bytes_stream(offset..identity.size())
+                .await
+                .map_err(map_resume_error)?
+        } else {
+            reader
+                .into_bytes_stream(offset..identity.size())
+                .await
+                .map_err(map_open_error)?
+        };
+        Ok(ObjectSourceBody::new(stream, identity.size() - offset))
+    }
+
+    fn ensure_identity(&self, identity: &ObjectSourceIdentity) -> Result<(), ObjectSourceError> {
+        if identity.endpoint() != self.config.endpoint()
+            || identity.bucket() != self.config.bucket()
+        {
+            return Err(ObjectSourceError::SourceChanged);
+        }
+        Ok(())
+    }
+}
+
+fn normalized_version(version: Option<&str>) -> Option<String> {
+    version
+        .filter(|value| !value.is_empty() && *value != "null")
+        .map(str::to_owned)
+}