Add Rust domain kernel and wire contracts

crates/remora-domain/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "remora-domain"
+edition.workspace = true
+publish = false
+rust-version.workspace = true
+version.workspace = true
+
+[dependencies]
+serde.workspace = true
+serde_json_canonicalizer.workspace = true
+sha2.workspace = true
+unicode-general-category.workspace = true
+unicode-normalization.workspace = true
+
+[dev-dependencies]
+serde_json.workspace = true

crates/remora-domain/src/failure.rs

@@ -0,0 +1,68 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FailureDisposition {
+    Permanent,
+    Retryable,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum RuntimeFailureClass {
+    Authorization,
+    SourceNotFound,
+    SourceChanged,
+    TransientSource,
+    TransientTarget,
+    Integrity,
+    Capacity,
+    Internal,
+}
+
+impl RuntimeFailureClass {
+    #[must_use]
+    pub const fn disposition(self) -> FailureDisposition {
+        match self {
+            Self::Authorization | Self::SourceNotFound | Self::SourceChanged => {
+                FailureDisposition::Permanent
+            }
+            Self::TransientSource
+            | Self::TransientTarget
+            | Self::Integrity
+            | Self::Capacity
+            | Self::Internal => FailureDisposition::Retryable,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum FailureClass {
+    Validation,
+    Authorization,
+    SourceNotFound,
+    SourceChanged,
+    TransientSource,
+    TransientTarget,
+    Integrity,
+    Capacity,
+    Internal,
+    Cancelled,
+    LostAttempt,
+}
+
+impl From<RuntimeFailureClass> for FailureClass {
+    fn from(value: RuntimeFailureClass) -> Self {
+        match value {
+            RuntimeFailureClass::Authorization => Self::Authorization,
+            RuntimeFailureClass::SourceNotFound => Self::SourceNotFound,
+            RuntimeFailureClass::SourceChanged => Self::SourceChanged,
+            RuntimeFailureClass::TransientSource => Self::TransientSource,
+            RuntimeFailureClass::TransientTarget => Self::TransientTarget,
+            RuntimeFailureClass::Integrity => Self::Integrity,
+            RuntimeFailureClass::Capacity => Self::Capacity,
+            RuntimeFailureClass::Internal => Self::Internal,
+        }
+    }
+}

crates/remora-domain/src/fingerprint.rs

@@ -0,0 +1,45 @@
+use std::fmt;
+
+use serde::Serialize;
+use sha2::{Digest, Sha256};
+
+#[derive(Clone, Copy, Eq, Hash, PartialEq)]
+pub struct RequestFingerprint([u8; 32]);
+
+impl RequestFingerprint {
+    #[must_use]
+    pub fn to_hex(self) -> String {
+        const HEX: &[u8; 16] = b"0123456789abcdef";
+        let mut output = String::with_capacity(64);
+        for byte in self.0 {
+            output.push(char::from(HEX[usize::from(byte >> 4)]));
+            output.push(char::from(HEX[usize::from(byte & 0x0f)]));
+        }
+        output
+    }
+}
+
+impl fmt::Debug for RequestFingerprint {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_tuple("RequestFingerprint")
+            .field(&self.to_hex())
+            .finish()
+    }
+}
+
+pub fn fingerprint(value: &impl Serialize) -> Result<RequestFingerprint, FingerprintError> {
+    let canonical = serde_json_canonicalizer::to_vec(value).map_err(|_| FingerprintError)?;
+    Ok(RequestFingerprint(Sha256::digest(canonical).into()))
+}
+
+#[derive(Debug)]
+pub struct FingerprintError;
+
+impl fmt::Display for FingerprintError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str("request cannot be represented as canonical JSON")
+    }
+}
+
+impl std::error::Error for FingerprintError {}

crates/remora-domain/src/id.rs

@@ -0,0 +1,73 @@
+use std::{fmt, str::FromStr};
+
+use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error as _};
+
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct OpaqueId(String);
+
+impl OpaqueId {
+    pub fn parse(value: impl Into<String>) -> Result<Self, IdError> {
+        let value = value.into();
+        let mut chars = value.chars();
+        let Some(first) = chars.next() else {
+            return Err(IdError);
+        };
+        if !(8..=128).contains(&value.len())
+            || !first.is_ascii_alphanumeric()
+            || !chars.all(|character| {
+                character.is_ascii_alphanumeric() || matches!(character, '_' | '-')
+            })
+        {
+            return Err(IdError);
+        }
+        Ok(Self(value))
+    }
+
+    #[must_use]
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl fmt::Display for OpaqueId {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        self.0.fmt(formatter)
+    }
+}
+
+impl FromStr for OpaqueId {
+    type Err = IdError;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        Self::parse(value)
+    }
+}
+
+impl Serialize for OpaqueId {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        serializer.serialize_str(&self.0)
+    }
+}
+
+impl<'de> Deserialize<'de> for OpaqueId {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: Deserializer<'de>,
+    {
+        Self::parse(String::deserialize(deserializer)?).map_err(D::Error::custom)
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct IdError;
+
+impl fmt::Display for IdError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str("invalid opaque identifier")
+    }
+}
+
+impl std::error::Error for IdError {}

crates/remora-domain/src/lib.rs

@@ -0,0 +1,16 @@
+mod failure;
+mod fingerprint;
+mod id;
+pub mod ports;
+mod target;
+mod validation;
+mod vocabulary;
+
+pub use failure::{FailureClass, FailureDisposition, RuntimeFailureClass};
+pub use fingerprint::{FingerprintError, RequestFingerprint, fingerprint};
+pub use id::{IdError, OpaqueId};
+pub use target::{StorageScope, Target, TargetStoreId, TargetStoreIdError};
+pub use validation::{BucketName, MemberPath, TargetPrefix, ValidationError};
+pub use vocabulary::{
+    AttemptState, JobLifecycle, JobPhase, LeaseState, OutcomeRequest, StageState,
+};

crates/remora-domain/src/ports.rs

@@ -0,0 +1,90 @@
+use std::future::Future;
+
+use crate::{OpaqueId, RequestFingerprint};
+
+pub trait Clock: Send + Sync {
+    type Instant: Copy + Ord + Send;
+
+    fn now(&self) -> Self::Instant;
+}
+
+pub trait DurableMutationStore: Send + Sync {
+    type Error;
+    type Request;
+    type Response;
+
+    fn execute(
+        &self,
+        scope: MutationScope<'_>,
+        request: &Self::Request,
+        request_fingerprint: RequestFingerprint,
+    ) -> impl Future<Output = Result<Self::Response, Self::Error>> + Send;
+}
+
+pub trait CoordinationStore: Send + Sync {
+    type Error;
+    type Claim;
+
+    fn claim(
+        &self,
+        worker_incarnation: &OpaqueId,
+        operation_id: &OpaqueId,
+    ) -> impl Future<Output = Result<Option<Self::Claim>, Self::Error>> + Send;
+}
+
+pub trait IdentityProvider: Send + Sync {
+    type Error;
+    type Principal;
+
+    fn authenticate(
+        &self,
+        bearer_token: &str,
+    ) -> impl Future<Output = Result<Self::Principal, Self::Error>> + Send;
+}
+
+pub trait CredentialResolver: Send + Sync {
+    type Error;
+    type Secret;
+
+    fn resolve(
+        &self,
+        credential_id: &OpaqueId,
+        version_id: &OpaqueId,
+    ) -> impl Future<Output = Result<Self::Secret, Self::Error>> + Send;
+}
+
+pub trait SourceAdapter: Send + Sync {
+    type Error;
+    type Request;
+    type ResolvedSource;
+
+    fn resolve(
+        &self,
+        request: &Self::Request,
+    ) -> impl Future<Output = Result<Self::ResolvedSource, Self::Error>> + Send;
+}
+
+pub trait StorageAdapter: Send + Sync {
+    type Error;
+    type Request;
+    type StoredObject;
+
+    fn create(
+        &self,
+        request: &Self::Request,
+    ) -> impl Future<Output = Result<Self::StoredObject, Self::Error>> + Send;
+}
+
+pub trait EventSink: Send + Sync {
+    type Error;
+    type Event;
+
+    fn publish(&self, event: Self::Event) -> impl Future<Output = Result<(), Self::Error>> + Send;
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct MutationScope<'a> {
+    pub subject: &'a OpaqueId,
+    pub operation: &'a str,
+    pub key: &'a str,
+}

crates/remora-domain/src/target.rs

@@ -0,0 +1,55 @@
+use std::fmt;
+
+use crate::{BucketName, TargetPrefix};
+
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct TargetStoreId(String);
+
+impl TargetStoreId {
+    pub fn parse(value: impl Into<String>) -> Result<Self, TargetStoreIdError> {
+        let value = value.into();
+        if value.is_empty() || value.chars().count() > 128 {
+            return Err(TargetStoreIdError);
+        }
+        Ok(Self(value))
+    }
+
+    #[must_use]
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct Target {
+    pub store: TargetStoreId,
+    pub bucket: BucketName,
+    pub prefix: TargetPrefix,
+}
+
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub struct StorageScope {
+    pub store: TargetStoreId,
+    pub bucket: BucketName,
+    pub prefix: TargetPrefix,
+}
+
+impl StorageScope {
+    #[must_use]
+    pub fn covers(&self, target: &Target) -> bool {
+        self.store == target.store
+            && self.bucket == target.bucket
+            && target.prefix.segments().starts_with(self.prefix.segments())
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub struct TargetStoreIdError;
+
+impl fmt::Display for TargetStoreIdError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.write_str("invalid target store identifier")
+    }
+}
+
+impl std::error::Error for TargetStoreIdError {}

crates/remora-domain/src/validation.rs

@@ -0,0 +1,161 @@
+use std::{collections::BTreeSet, fmt, str::FromStr};
+
+use unicode_general_category::{GeneralCategory, get_general_category};
+use unicode_normalization::UnicodeNormalization;
+
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct BucketName(String);
+
+impl BucketName {
+    pub fn parse(value: impl Into<String>) -> Result<Self, ValidationError> {
+        let value = value.into();
+        let bytes = value.as_bytes();
+        let valid_character = |byte: &u8| {
+            byte.is_ascii_lowercase() || byte.is_ascii_digit() || matches!(byte, b'.' | b'-')
+        };
+        if !(3..=63).contains(&bytes.len())
+            || !bytes.first().is_some_and(u8::is_ascii_alphanumeric)
+            || !bytes.last().is_some_and(u8::is_ascii_alphanumeric)
+            || !bytes.iter().all(valid_character)
+            || value.contains("..")
+        {
+            return Err(ValidationError::Bucket);
+        }
+        Ok(Self(value))
+    }
+
+    #[must_use]
+    pub fn as_str(&self) -> &str {
+        &self.0
+    }
+}
+
+impl FromStr for BucketName {
+    type Err = ValidationError;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        Self::parse(value)
+    }
+}
+
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct TargetPrefix(Vec<String>);
+
+impl TargetPrefix {
+    pub fn parse<'a>(
+        value: &str,
+        reserved_segments: impl IntoIterator<Item = &'a str>,
+    ) -> Result<Self, ValidationError> {
+        if value == "/" || value.starts_with('/') || value.contains("//") || value.contains('\\') {
+            return Err(ValidationError::TargetPrefix);
+        }
+        let value = value.strip_suffix('/').unwrap_or(value);
+        if value.is_empty() {
+            return Ok(Self(Vec::new()));
+        }
+        let segments = value
+            .split('/')
+            .map(|segment| normalize_segment(segment, ValidationError::TargetPrefix))
+            .collect::<Result<Vec<_>, _>>()?;
+        let reserved_segments = reserved_segments
+            .into_iter()
+            .map(|segment| normalize_segment(segment, ValidationError::ReservedTargetPrefix))
+            .collect::<Result<BTreeSet<_>, _>>()?;
+        if segments
+            .iter()
+            .any(|segment| reserved_segments.contains(segment))
+        {
+            return Err(ValidationError::ReservedTargetPrefix);
+        }
+        Ok(Self(segments))
+    }
+
+    #[must_use]
+    pub fn segments(&self) -> &[String] {
+        &self.0
+    }
+
+    #[must_use]
+    pub fn canonical(&self) -> String {
+        self.0.join("/")
+    }
+}
+
+#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
+pub struct MemberPath(Vec<String>);
+
+impl MemberPath {
+    pub fn parse(value: &str) -> Result<Self, ValidationError> {
+        if value.is_empty()
+            || value.starts_with('/')
+            || value.ends_with('/')
+            || value.contains("//")
+            || value.contains('\\')
+        {
+            return Err(ValidationError::MemberPath);
+        }
+        let segments = value
+            .split('/')
+            .map(|segment| normalize_segment(segment, ValidationError::MemberPath))
+            .collect::<Result<Vec<_>, _>>()?;
+        Ok(Self(segments))
+    }
+
+    #[must_use]
+    pub fn segments(&self) -> &[String] {
+        &self.0
+    }
+
+    #[must_use]
+    pub fn canonical(&self) -> String {
+        self.0.join("/")
+    }
+}
+
+impl FromStr for MemberPath {
+    type Err = ValidationError;
+
+    fn from_str(value: &str) -> Result<Self, Self::Err> {
+        Self::parse(value)
+    }
+}
+
+fn normalize_segment(segment: &str, error: ValidationError) -> Result<String, ValidationError> {
+    if segment.is_empty() || matches!(segment, "." | "..") || segment.chars().any(is_category_c) {
+        return Err(error);
+    }
+    Ok(segment.nfc().collect())
+}
+
+fn is_category_c(character: char) -> bool {
+    matches!(
+        get_general_category(character),
+        GeneralCategory::Control
+            | GeneralCategory::Format
+            | GeneralCategory::PrivateUse
+            | GeneralCategory::Surrogate
+            | GeneralCategory::Unassigned
+    )
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum ValidationError {
+    Bucket,
+    TargetPrefix,
+    ReservedTargetPrefix,
+    MemberPath,
+}
+
+impl fmt::Display for ValidationError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        let message = match self {
+            Self::Bucket => "invalid bucket name",
+            Self::TargetPrefix => "invalid target prefix",
+            Self::ReservedTargetPrefix => "target prefix uses a reserved segment",
+            Self::MemberPath => "invalid result member path",
+        };
+        formatter.write_str(message)
+    }
+}
+
+impl std::error::Error for ValidationError {}

crates/remora-domain/src/vocabulary.rs

@@ -0,0 +1,60 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+pub enum JobPhase {
+    Queued,
+    Running,
+    RetryWait,
+    Cancelling,
+    Succeeded,
+    Failed,
+    Cancelled,
+}
+
+impl JobPhase {
+    #[must_use]
+    pub const fn is_terminal(self) -> bool {
+        matches!(self, Self::Succeeded | Self::Failed | Self::Cancelled)
+    }
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+pub enum AttemptState {
+    AttemptUnused,
+    AttemptAuthorized,
+    AttemptExecuting,
+    AttemptStopped,
+    AttemptCommitted,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+pub enum LeaseState {
+    NoLease,
+    LeaseLive,
+    LeaseExpired,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+pub enum StageState {
+    StageNone,
+    StagePartial,
+    StageComplete,
+    StageCleaned,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+pub enum OutcomeRequest {
+    NoOutcome,
+    WantsSuccess,
+    WantsRetryableFailure,
+    WantsPermanentFailure,
+}
+
+#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
+#[serde(rename_all = "snake_case")]
+pub enum JobLifecycle {
+    Visible,
+    SoftDeleted,
+    HardDeleting,
+    HardDeleted,
+}

crates/remora-domain/tests/golden.rs

@@ -0,0 +1,137 @@
+use remora_domain::{
+    BucketName, FailureDisposition, JobPhase, MemberPath, RuntimeFailureClass, StorageScope,
+    Target, TargetPrefix, TargetStoreId, fingerprint,
+};
+use serde::Deserialize;
+use serde_json::Value;
+
+#[derive(Deserialize)]
+struct Corpus {
+    job_phases: Vec<JobPhase>,
+    runtime_failures: Vec<FailureCase>,
+    buckets: ValidationCases<String>,
+    target_prefixes: ValidationCases<CanonicalCase>,
+    reserved_target_prefixes: Vec<ReservedPrefixCase>,
+    member_paths: ValidationCases<CanonicalCase>,
+    scope_coverage: Vec<ScopeCase>,
+    fingerprints: Vec<FingerprintCase>,
+}
+
+#[derive(Deserialize)]
+struct FailureCase {
+    class: RuntimeFailureClass,
+    disposition: FailureDisposition,
+}
+
+#[derive(Deserialize)]
+struct ValidationCases<T> {
+    valid: Vec<T>,
+    invalid: Vec<String>,
+}
+
+#[derive(Deserialize)]
+struct CanonicalCase {
+    input: String,
+    canonical: String,
+}
+
+#[derive(Deserialize)]
+struct FingerprintCase {
+    value: Value,
+    canonical: String,
+    sha256: String,
+}
+
+#[derive(Deserialize)]
+struct ScopeCase {
+    scope: [String; 3],
+    target: [String; 3],
+    covers: bool,
+}
+
+#[derive(Deserialize)]
+struct ReservedPrefixCase {
+    input: String,
+    reserved: Vec<String>,
+}
+
+fn corpus() -> Corpus {
+    serde_json::from_str(include_str!("../../../contracts/golden/domain.json")).unwrap()
+}
+
+#[test]
+fn protocol_phase_vocabulary_is_exact() {
+    let phases = corpus().job_phases;
+    assert_eq!(phases.len(), 7);
+    assert!(phases[..4].iter().all(|phase| !phase.is_terminal()));
+    assert!(phases[4..].iter().all(|phase| phase.is_terminal()));
+}
+
+#[test]
+fn runtime_failure_dispositions_match_the_architecture() {
+    for case in corpus().runtime_failures {
+        assert_eq!(case.class.disposition(), case.disposition);
+    }
+}
+
+#[test]
+fn domain_values_validate_and_normalize() {
+    let corpus = corpus();
+    for value in corpus.buckets.valid {
+        assert_eq!(BucketName::parse(&value).unwrap().as_str(), value);
+    }
+    for value in corpus.buckets.invalid {
+        assert!(BucketName::parse(value).is_err());
+    }
+    for case in corpus.target_prefixes.valid {
+        assert_eq!(
+            TargetPrefix::parse(&case.input, []).unwrap().canonical(),
+            case.canonical
+        );
+    }
+    for value in corpus.target_prefixes.invalid {
+        assert!(TargetPrefix::parse(&value, []).is_err());
+    }
+    for case in corpus.reserved_target_prefixes {
+        assert!(
+            TargetPrefix::parse(&case.input, case.reserved.iter().map(String::as_str)).is_err()
+        );
+    }
+    for case in corpus.member_paths.valid {
+        assert_eq!(
+            MemberPath::parse(&case.input).unwrap().canonical(),
+            case.canonical
+        );
+    }
+    for value in corpus.member_paths.invalid {
+        assert!(MemberPath::parse(&value).is_err());
+    }
+}
+
+#[test]
+fn request_fingerprints_match_the_shared_corpus() {
+    for case in corpus().fingerprints {
+        assert_eq!(
+            serde_json_canonicalizer::to_string(&case.value).unwrap(),
+            case.canonical
+        );
+        assert_eq!(fingerprint(&case.value).unwrap().to_hex(), case.sha256);
+    }
+}
+
+#[test]
+fn storage_scope_coverage_uses_exact_segment_tuples() {
+    for case in corpus().scope_coverage {
+        let scope = StorageScope {
+            store: TargetStoreId::parse(&case.scope[0]).unwrap(),
+            bucket: BucketName::parse(&case.scope[1]).unwrap(),
+            prefix: TargetPrefix::parse(&case.scope[2], []).unwrap(),
+        };
+        let target = Target {
+            store: TargetStoreId::parse(&case.target[0]).unwrap(),
+            bucket: BucketName::parse(&case.target[1]).unwrap(),
+            prefix: TargetPrefix::parse(&case.target[2], []).unwrap(),
+        };
+        assert_eq!(scope.covers(&target), case.covers);
+    }
+}