lodestone_types/

replayable_source.rs

//! P13.1 replayable-source authority and current-format contracts.

use std::collections::{BTreeMap, BTreeSet};
use std::error::Error;
use std::fmt::{self, Display, Formatter};
use std::future::Future;
use std::num::{NonZeroU32, NonZeroU64, NonZeroU128};
use std::pin::Pin;
use std::time::Instant;

use serde::de::{DeserializeOwned, Visitor};
use serde::{Deserialize, Deserializer, Serialize, Serializer};

use crate::{CollectionIndexRef, FenceToken, IndexMode, ManifestGeneration, ShardKey};

pub const REPLAYABLE_SOURCE_FORMAT_VERSION: u16 = 1;
pub const MAX_SOURCE_PARTITIONS: usize = 128;
pub const MAX_SOURCE_RESUME_TOKEN_BYTES: usize = 4_096;
pub const MAX_SOURCE_RECORD_BYTES: usize = 1_048_576;
pub const MAX_SOURCE_BATCH_RECORDS: usize = 4_096;
pub const MAX_SOURCE_BATCH_BYTES: usize = 16_777_216;
pub const MAX_SOURCE_CONFIGURATION_BYTES: usize = 262_144;
pub const MAX_ATOMIC_SOURCE_SHARDS: usize = 512;
pub const MAX_ATOMIC_SOURCE_COMMAND_BYTES: usize = 1_048_576;

macro_rules! nonzero_number {
    ($name:ident, $inner:ty, $primitive:ty, $description:literal) => {
        #[doc = $description]
        #[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
        #[serde(transparent)]
        pub struct $name($inner);

        impl $name {
            pub fn new(value: $primitive) -> Result<Self, SourceContractError> {
                <$inner>::new(value)
                    .map(Self)
                    .ok_or(SourceContractError::ZeroIdentity(stringify!($name)))
            }

            #[must_use]
            pub const fn get(self) -> $primitive {
                self.0.get()
            }
        }

        impl Display for $name {
            fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
                Display::fmt(&self.get(), formatter)
            }
        }
    };
}

// INVARIANT: JSON cannot portably represent the complete u128 domain. Source
// identities are therefore canonical nonzero decimal strings on every durable
// and network boundary, while the Rust API retains its semantic numeric type.
macro_rules! nonzero_u128_decimal {
    ($name:ident, $description:literal) => {
        #[doc = $description]
        #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
        pub struct $name(NonZeroU128);

        impl $name {
            pub fn new(value: u128) -> Result<Self, SourceContractError> {
                NonZeroU128::new(value)
                    .map(Self)
                    .ok_or(SourceContractError::ZeroIdentity(stringify!($name)))
            }

            #[must_use]
            pub const fn get(self) -> u128 {
                self.0.get()
            }
        }

        impl Display for $name {
            fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
                Display::fmt(&self.get(), formatter)
            }
        }

        impl Serialize for $name {
            fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
            where
                S: Serializer,
            {
                serializer.serialize_str(&self.get().to_string())
            }
        }

        impl<'de> Deserialize<'de> for $name {
            fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
            where
                D: Deserializer<'de>,
            {
                struct CanonicalNonzeroDecimalVisitor;

                impl Visitor<'_> for CanonicalNonzeroDecimalVisitor {
                    type Value = $name;

                    fn expecting(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
                        formatter.write_str("a canonical nonzero u128 decimal string")
                    }

                    fn visit_str<E>(self, value: &str) -> Result<Self::Value, E>
                    where
                        E: serde::de::Error,
                    {
                        let bytes = value.as_bytes();
                        if bytes.is_empty()
                            || !matches!(bytes[0], b'1'..=b'9')
                            || !bytes[1..].iter().all(u8::is_ascii_digit)
                        {
                            return Err(E::custom("expected a canonical nonzero u128 decimal string"));
                        }
                        let parsed = value
                            .parse::<u128>()
                            .map_err(|_| E::custom("source identity exceeds u128"))?;
                        NonZeroU128::new(parsed)
                            .map($name)
                            .ok_or_else(|| E::custom("source identity cannot be zero"))
                    }
                }

                deserializer.deserialize_str(CanonicalNonzeroDecimalVisitor)
            }
        }
    };
}

nonzero_number!(
    SourceIncarnation,
    NonZeroU64,
    u64,
    "A never-reused lifetime of one exact external source binding."
);
/// A stable, zero-based partition identity inside one source incarnation.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct SourcePartitionId(u32);

impl SourcePartitionId {
    #[must_use]
    pub const fn new(value: u32) -> Self {
        Self(value)
    }

    #[must_use]
    pub const fn get(self) -> u32 {
        self.0
    }
}

impl Display for SourcePartitionId {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        Display::fmt(&self.0, formatter)
    }
}
nonzero_number!(
    SourcePartitionIncarnation,
    NonZeroU64,
    u64,
    "A never-reused lifetime of one source partition identity."
);
nonzero_u128_decimal!(
    SourceOperationId,
    "A stable operation identity reproduced by every replay."
);
nonzero_u128_decimal!(
    SourceTransactionId,
    "A stable source transaction identity reproduced by every replay."
);
nonzero_u128_decimal!(SourceBatchId, "A stable identity for one bounded provider batch.");
nonzero_number!(
    SourceControlGeneration,
    NonZeroU64,
    u64,
    "The monotonic generation of one replayable-source lifecycle state."
);

/// An exclusive per-partition source position: the next ordinal to return.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(transparent)]
pub struct SourceOrdinal(u64);

impl SourceOrdinal {
    #[must_use]
    pub const fn new(value: u64) -> Self {
        Self(value)
    }

    #[must_use]
    pub const fn get(self) -> u64 {
        self.0
    }

    #[must_use]
    pub const fn checked_next(self) -> Option<Self> {
        match self.0.checked_add(1) {
            Some(value) => Some(Self(value)),
            None => None,
        }
    }
}

impl Display for SourceOrdinal {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        Display::fmt(&self.0, formatter)
    }
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SourceProviderKindV1 {
    Kafka,
    ObjectLog,
    HttpReplay,
    GrpcReplay,
}

#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SourceDecoderKindV1 {
    NativeDocumentJson,
    DebeziumPostgresJson,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SourceReplayGuaranteesV1 {
    pub durable_replay: bool,
    pub stable_partition_identity: bool,
    pub stable_operation_identity: bool,
    pub ordered_positions: bool,
    pub bounded_reads: bool,
    pub explicit_retention: bool,
    pub explicit_rewind: bool,
    pub minimum_replay_window_millis: u64,
}

impl SourceReplayGuaranteesV1 {
    pub fn validate(&self) -> Result<(), SourceContractError> {
        if !self.durable_replay
            || !self.stable_partition_identity
            || !self.stable_operation_identity
            || !self.ordered_positions
            || !self.bounded_reads
            || !self.explicit_retention
            || !self.explicit_rewind
            || self.minimum_replay_window_millis == 0
        {
            return Err(SourceContractError::Violation(
                "WAL elision requires every replay guarantee and a positive replay window",
            ));
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SourceCursorV1 {
    /// Exclusive position: the next record the provider must return.
    pub next_ordinal: SourceOrdinal,
    /// Provider-owned opaque resume material. It is metadata, never a secret.
    pub resume_token: Vec<u8>,
}

impl SourceCursorV1 {
    pub fn validate(&self) -> Result<(), SourceContractError> {
        if self.resume_token.len() > MAX_SOURCE_RESUME_TOKEN_BYTES {
            return Err(SourceContractError::Budget {
                field: "source resume token bytes",
                observed: self.resume_token.len(),
                maximum: MAX_SOURCE_RESUME_TOKEN_BYTES,
            });
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SourcePartitionCheckpointV1 {
    pub partition_id: SourcePartitionId,
    pub partition_incarnation: SourcePartitionIncarnation,
    pub cursor: SourceCursorV1,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SourceCutV1 {
    pub format_version: u16,
    pub source_incarnation: SourceIncarnation,
    pub partitions: Vec<SourcePartitionCheckpointV1>,
}

impl SourceCutV1 {
    pub fn validate(&self) -> Result<(), SourceContractError> {
        validate_format(self.format_version)?;
        if self.partitions.is_empty() || self.partitions.len() > MAX_SOURCE_PARTITIONS {
            return Err(SourceContractError::Budget {
                field: "source partitions",
                observed: self.partitions.len(),
                maximum: MAX_SOURCE_PARTITIONS,
            });
        }
        let mut prior = None;
        for checkpoint in &self.partitions {
            if prior.is_some_and(|value| value >= checkpoint.partition_id) {
                return Err(SourceContractError::Violation(
                    "partition checkpoints must be strictly ordered and unique",
                ));
            }
            checkpoint.cursor.validate()?;
            prior = Some(checkpoint.partition_id);
        }
        Ok(())
    }

    pub fn advances_to(&self, replacement: &Self) -> Result<(), SourceContractError> {
        self.validate()?;
        replacement.validate()?;
        if self.source_incarnation != replacement.source_incarnation
            || self.partitions.len() != replacement.partitions.len()
        {
            return Err(SourceContractError::Violation(
                "a source cut cannot change source incarnation or partition membership",
            ));
        }
        let mut advanced = false;
        for (expected, next) in self.partitions.iter().zip(&replacement.partitions) {
            if expected.partition_id != next.partition_id
                || expected.partition_incarnation != next.partition_incarnation
                || next.cursor.next_ordinal < expected.cursor.next_ordinal
                || (next.cursor.next_ordinal == expected.cursor.next_ordinal
                    && next.cursor.resume_token != expected.cursor.resume_token)
            {
                return Err(SourceContractError::Violation(
                    "replacement source cut must preserve partition identities and move forward",
                ));
            }
            advanced |= next.cursor.next_ordinal > expected.cursor.next_ordinal;
        }
        if !advanced {
            return Err(SourceContractError::Violation(
                "replacement source cut must advance at least one partition",
            ));
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SourcePartitionBindingV1 {
    pub partition_id: SourcePartitionId,
    pub partition_incarnation: SourcePartitionIncarnation,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SourceBindingV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub source_incarnation: SourceIncarnation,
    pub source_identity_digest: [u8; 32],
    pub provider: SourceProviderKindV1,
    pub decoder: SourceDecoderKindV1,
    pub partitions: Vec<SourcePartitionBindingV1>,
    pub guarantees: SourceReplayGuaranteesV1,
}

impl SourceBindingV1 {
    pub fn validate(&self) -> Result<(), SourceContractError> {
        validate_format(self.format_version)?;
        if self.source_identity_digest == [0; 32] {
            return Err(SourceContractError::ZeroIdentity("source_identity_digest"));
        }
        if self.partitions.is_empty() || self.partitions.len() > MAX_SOURCE_PARTITIONS {
            return Err(SourceContractError::Budget {
                field: "source partitions",
                observed: self.partitions.len(),
                maximum: MAX_SOURCE_PARTITIONS,
            });
        }
        let mut prior = None;
        for partition in &self.partitions {
            if prior.is_some_and(|value| value >= partition.partition_id) {
                return Err(SourceContractError::Violation(
                    "source partition bindings must be strictly ordered and unique",
                ));
            }
            prior = Some(partition.partition_id);
        }
        self.guarantees.validate()
    }

    pub fn validate_cut(&self, cut: &SourceCutV1) -> Result<(), SourceContractError> {
        self.validate()?;
        cut.validate()?;
        if self.source_incarnation != cut.source_incarnation || self.partitions.len() != cut.partitions.len() {
            return Err(SourceContractError::Violation(
                "source cut does not belong to the catalog source binding",
            ));
        }
        for (binding, checkpoint) in self.partitions.iter().zip(&cut.partitions) {
            if binding.partition_id != checkpoint.partition_id
                || binding.partition_incarnation != checkpoint.partition_incarnation
            {
                return Err(SourceContractError::Violation(
                    "source cut partition identity differs from the catalog binding",
                ));
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum IngestAuthorityV1 {
    PushWithWal {
        format_version: u16,
    },
    ReplayableSource {
        format_version: u16,
        binding: SourceBindingV1,
    },
}

impl IngestAuthorityV1 {
    #[must_use]
    pub const fn push_with_wal() -> Self {
        Self::PushWithWal {
            format_version: REPLAYABLE_SOURCE_FORMAT_VERSION,
        }
    }

    #[must_use]
    pub const fn replayable_source(binding: SourceBindingV1) -> Self {
        Self::ReplayableSource {
            format_version: REPLAYABLE_SOURCE_FORMAT_VERSION,
            binding,
        }
    }

    pub fn validate_for_mode(&self, mode: IndexMode) -> Result<(), SourceContractError> {
        match self {
            Self::PushWithWal { format_version } => validate_format(*format_version),
            Self::ReplayableSource {
                format_version,
                binding,
            } if mode.is_state() => {
                validate_format(*format_version)?;
                let _ = binding;
                Err(SourceContractError::UnsupportedMode(mode))
            }
            Self::ReplayableSource {
                format_version,
                binding,
            } => {
                validate_format(*format_version)?;
                binding.validate()
            }
        }
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum SourceTransactionMembershipV1 {
    Standalone,
    Transaction {
        transaction_id: SourceTransactionId,
        member_index: u32,
        member_count: NonZeroU32,
    },
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct RawSourceRecordV1 {
    pub source_incarnation: SourceIncarnation,
    pub partition_id: SourcePartitionId,
    pub partition_incarnation: SourcePartitionIncarnation,
    pub position: SourceCursorV1,
    pub next_position: SourceCursorV1,
    pub operation_id: SourceOperationId,
    pub transaction: SourceTransactionMembershipV1,
    pub key: Vec<u8>,
    pub payload: Vec<u8>,
}

impl RawSourceRecordV1 {
    pub fn validate(&self) -> Result<(), SourceContractError> {
        self.position.validate()?;
        self.next_position.validate()?;
        let bytes = self
            .key
            .len()
            .checked_add(self.payload.len())
            .ok_or(SourceContractError::ArithmeticOverflow)?;
        if bytes > MAX_SOURCE_RECORD_BYTES {
            return Err(SourceContractError::Budget {
                field: "source record bytes",
                observed: bytes,
                maximum: MAX_SOURCE_RECORD_BYTES,
            });
        }
        if self.next_position.next_ordinal <= self.position.next_ordinal {
            return Err(SourceContractError::Violation(
                "a source record must advance its exclusive cursor",
            ));
        }
        if let SourceTransactionMembershipV1::Transaction {
            member_index,
            member_count,
            ..
        } = self.transaction
            && member_index >= member_count.get()
        {
            return Err(SourceContractError::Violation(
                "source transaction member index is outside member count",
            ));
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SourceBatchV1 {
    pub format_version: u16,
    pub batch_id: SourceBatchId,
    pub expected_cut: SourceCutV1,
    pub replacement_cut: SourceCutV1,
    pub records: Vec<RawSourceRecordV1>,
}

impl SourceBatchV1 {
    pub fn validate(&self) -> Result<(), SourceContractError> {
        validate_format(self.format_version)?;
        self.expected_cut.advances_to(&self.replacement_cut)?;
        if self.records.is_empty() || self.records.len() > MAX_SOURCE_BATCH_RECORDS {
            return Err(SourceContractError::Budget {
                field: "source batch records",
                observed: self.records.len(),
                maximum: MAX_SOURCE_BATCH_RECORDS,
            });
        }
        let encoded = serde_json::to_vec(self).map_err(SourceContractError::serde)?;
        if encoded.len() > MAX_SOURCE_BATCH_BYTES {
            return Err(SourceContractError::Budget {
                field: "source batch bytes",
                observed: encoded.len(),
                maximum: MAX_SOURCE_BATCH_BYTES,
            });
        }

        let mut cursors = self
            .expected_cut
            .partitions
            .iter()
            .map(|checkpoint| {
                (
                    checkpoint.partition_id,
                    (checkpoint.partition_incarnation, checkpoint.cursor.clone()),
                )
            })
            .collect::<BTreeMap<_, _>>();
        let mut operations = BTreeSet::new();
        let mut transactions = BTreeMap::<SourceTransactionId, (u32, BTreeSet<u32>)>::new();
        let mut prior_record_key = None;
        for record in &self.records {
            record.validate()?;
            let record_key = (record.partition_id, record.position.next_ordinal, record.operation_id);
            if prior_record_key.is_some_and(|prior| prior >= record_key) {
                return Err(SourceContractError::Violation(
                    "source records must be strictly ordered by partition, position, and operation",
                ));
            }
            prior_record_key = Some(record_key);
            if record.source_incarnation != self.expected_cut.source_incarnation {
                return Err(SourceContractError::Violation("source record incarnation mismatch"));
            }
            let (partition_incarnation, cursor) = cursors
                .get_mut(&record.partition_id)
                .ok_or(SourceContractError::Violation("source record has an unknown partition"))?;
            if *partition_incarnation != record.partition_incarnation || *cursor != record.position {
                return Err(SourceContractError::Violation(
                    "source records must form a gap-free cursor chain",
                ));
            }
            *cursor = record.next_position.clone();
            if !operations.insert(record.operation_id) {
                return Err(SourceContractError::Violation(
                    "source operation identity was duplicated",
                ));
            }
            if let SourceTransactionMembershipV1::Transaction {
                transaction_id,
                member_index,
                member_count,
            } = record.transaction
            {
                let entry = transactions
                    .entry(transaction_id)
                    .or_insert_with(|| (member_count.get(), BTreeSet::new()));
                if entry.0 != member_count.get() || !entry.1.insert(member_index) {
                    return Err(SourceContractError::Violation(
                        "source transaction membership is inconsistent or duplicated",
                    ));
                }
            }
        }
        for (_, (member_count, members)) in transactions {
            if members.len() != member_count as usize || members.iter().copied().ne(0..member_count) {
                return Err(SourceContractError::Violation(
                    "a source batch must contain every member of each admitted transaction",
                ));
            }
        }
        for replacement in &self.replacement_cut.partitions {
            let observed = cursors
                .get(&replacement.partition_id)
                .ok_or(SourceContractError::Violation(
                    "replacement cut has an unknown partition",
                ))?;
            if observed.0 != replacement.partition_incarnation || observed.1 != replacement.cursor {
                return Err(SourceContractError::Violation(
                    "replacement cut must equal the exact end of the record chains",
                ));
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
pub struct SourceShardPublicationV1 {
    pub shard_key: ShardKey,
    pub fence_token: FenceToken,
    pub expected_generation: ManifestGeneration,
    pub replacement_generation: ManifestGeneration,
    pub replacement_root_digest: [u8; 32],
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(deny_unknown_fields)]
/// The sole bounded multi-shard publication primitive for one replayable
/// source batch. It is internal authority state, not a client transaction.
pub struct AtomicSourcePublicationV1 {
    pub format_version: u16,
    pub collection_index_ref: CollectionIndexRef,
    pub source_incarnation: SourceIncarnation,
    pub batch_id: SourceBatchId,
    pub expected_cut: SourceCutV1,
    pub replacement_cut: SourceCutV1,
    pub operation_count: u32,
    pub operation_set_digest: [u8; 32],
    pub committed_transactions: Vec<SourceTransactionId>,
    pub shard_publications: Vec<SourceShardPublicationV1>,
}

impl AtomicSourcePublicationV1 {
    pub fn validate(&self) -> Result<(), SourceContractError> {
        validate_format(self.format_version)?;
        self.expected_cut.advances_to(&self.replacement_cut)?;
        if self.source_incarnation != self.expected_cut.source_incarnation
            || self.operation_count == 0
            || self.operation_set_digest == [0; 32]
        {
            return Err(SourceContractError::Violation(
                "atomic source publication identity, operation count, or operation digest is invalid",
            ));
        }
        if self.shard_publications.is_empty() || self.shard_publications.len() > MAX_ATOMIC_SOURCE_SHARDS {
            return Err(SourceContractError::Budget {
                field: "atomic source shards",
                observed: self.shard_publications.len(),
                maximum: MAX_ATOMIC_SOURCE_SHARDS,
            });
        }
        let mut prior_shard = None;
        for publication in &self.shard_publications {
            if publication.shard_key.collection_index_ref() != self.collection_index_ref
                || prior_shard.is_some_and(|prior| prior >= publication.shard_key)
                || publication.expected_generation.checked_next() != Some(publication.replacement_generation)
                || publication.replacement_root_digest == [0; 32]
            {
                return Err(SourceContractError::Violation(
                    "source shard publications must be scoped, ordered, unique, and one-generation successors",
                ));
            }
            prior_shard = Some(publication.shard_key);
        }
        if self.committed_transactions.windows(2).any(|pair| pair[0] >= pair[1]) {
            return Err(SourceContractError::Violation(
                "committed source transactions must be strictly ordered and unique",
            ));
        }
        let encoded = serde_json::to_vec(self).map_err(SourceContractError::serde)?;
        if encoded.len() > MAX_ATOMIC_SOURCE_COMMAND_BYTES {
            return Err(SourceContractError::Budget {
                field: "atomic source command bytes",
                observed: encoded.len(),
                maximum: MAX_ATOMIC_SOURCE_COMMAND_BYTES,
            });
        }
        Ok(())
    }

    pub fn validate_against_batch(&self, batch: &SourceBatchV1) -> Result<(), SourceContractError> {
        self.validate()?;
        batch.validate()?;
        if self.source_incarnation != batch.expected_cut.source_incarnation
            || self.batch_id != batch.batch_id
            || self.expected_cut != batch.expected_cut
            || self.replacement_cut != batch.replacement_cut
            || usize::try_from(self.operation_count).ok() != Some(batch.records.len())
            || self.operation_set_digest != source_batch_operation_set_digest_v1(batch)?
            || self.committed_transactions != source_batch_transaction_ids_v1(batch)?
        {
            return Err(SourceContractError::Violation(
                "atomic source publication does not exactly summarize its validated source batch",
            ));
        }
        Ok(())
    }
}

pub fn source_batch_operation_set_digest_v1(batch: &SourceBatchV1) -> Result<[u8; 32], SourceContractError> {
    batch.validate()?;
    let mut hasher = blake3::Hasher::new();
    hasher.update(b"lodestone:source-operation-set:v1\0");
    for record in &batch.records {
        let encoded = serde_json::to_vec(record).map_err(SourceContractError::serde)?;
        let length = u64::try_from(encoded.len()).map_err(|_| SourceContractError::ArithmeticOverflow)?;
        hasher.update(&length.to_be_bytes());
        hasher.update(&encoded);
    }
    Ok(*hasher.finalize().as_bytes())
}

pub fn source_batch_transaction_ids_v1(batch: &SourceBatchV1) -> Result<Vec<SourceTransactionId>, SourceContractError> {
    batch.validate()?;
    Ok(batch
        .records
        .iter()
        .filter_map(|record| match record.transaction {
            SourceTransactionMembershipV1::Standalone => None,
            SourceTransactionMembershipV1::Transaction { transaction_id, .. } => Some(transaction_id),
        })
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect())
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SourceFetchRequestV1 {
    pub cut: SourceCutV1,
    pub maximum_records: usize,
    pub maximum_bytes: usize,
    pub deadline: Instant,
}

impl SourceFetchRequestV1 {
    pub fn validate(&self, now: Instant) -> Result<(), SourceContractError> {
        self.cut.validate()?;
        if self.maximum_records == 0 || self.maximum_records > MAX_SOURCE_BATCH_RECORDS {
            return Err(SourceContractError::Budget {
                field: "fetch records",
                observed: self.maximum_records,
                maximum: MAX_SOURCE_BATCH_RECORDS,
            });
        }
        if self.maximum_bytes == 0 || self.maximum_bytes > MAX_SOURCE_BATCH_BYTES {
            return Err(SourceContractError::Budget {
                field: "fetch bytes",
                observed: self.maximum_bytes,
                maximum: MAX_SOURCE_BATCH_BYTES,
            });
        }
        if self.deadline <= now {
            return Err(SourceContractError::DeadlineElapsed);
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SourceHighWatermarkV1 {
    Known { cursor: SourceCursorV1 },
    Unknown,
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub struct SourceFetchResponseV1 {
    pub batch: SourceBatchV1,
    pub high_watermarks: Vec<(SourcePartitionId, SourceHighWatermarkV1)>,
    pub retention_floors: Vec<(SourcePartitionId, SourceCursorV1)>,
}

impl SourceFetchResponseV1 {
    pub fn validate(&self) -> Result<(), SourceContractError> {
        self.batch.validate()?;
        if self.high_watermarks.len() != self.batch.replacement_cut.partitions.len() {
            return Err(SourceContractError::Violation(
                "source response must report one watermark state per partition",
            ));
        }
        if self.retention_floors.len() != self.batch.expected_cut.partitions.len() {
            return Err(SourceContractError::Violation(
                "source response must report one retention floor per partition",
            ));
        }
        for (reported, checkpoint) in self.high_watermarks.iter().zip(&self.batch.replacement_cut.partitions) {
            if reported.0 != checkpoint.partition_id {
                return Err(SourceContractError::Violation(
                    "source watermarks must follow the complete ordered partition cut",
                ));
            }
            if let SourceHighWatermarkV1::Known { cursor } = &reported.1 {
                cursor.validate()?;
                if cursor.next_ordinal < checkpoint.cursor.next_ordinal
                    || (cursor.next_ordinal == checkpoint.cursor.next_ordinal
                        && cursor.resume_token != checkpoint.cursor.resume_token)
                {
                    return Err(SourceContractError::Violation(
                        "known source watermark cannot precede the fetched replacement cut",
                    ));
                }
            }
        }
        for (reported, checkpoint) in self.retention_floors.iter().zip(&self.batch.expected_cut.partitions) {
            reported.1.validate()?;
            if reported.0 != checkpoint.partition_id
                || reported.1.next_ordinal > checkpoint.cursor.next_ordinal
                || (reported.1.next_ordinal == checkpoint.cursor.next_ordinal
                    && reported.1.resume_token != checkpoint.cursor.resume_token)
            {
                return Err(SourceContractError::Violation(
                    "source retention floors must cover the complete requested cut",
                ));
            }
        }
        Ok(())
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum DecodedSourceMutationV1 {
    /// A source control record whose position must advance atomically with the
    /// surrounding batch but which produces no document mutation.
    Control { operation_id: SourceOperationId },
    Put {
        operation_id: SourceOperationId,
        document_key: Vec<u8>,
        source: Vec<u8>,
    },
    Delete {
        operation_id: SourceOperationId,
        document_key: Vec<u8>,
    },
}

pub type SourceProviderFuture<'a> =
    Pin<Box<dyn Future<Output = Result<SourceFetchResponseV1, SourceProviderError>> + Send + 'a>>;

/// Provider owns replay position and stable raw-record identity; it never decodes documents.
pub trait ReplayableSourceProvider: Send {
    fn provider_kind(&self) -> SourceProviderKindV1;
    fn fetch<'a>(&'a mut self, request: &'a SourceFetchRequestV1) -> SourceProviderFuture<'a>;
}

/// Decoder owns strict envelope-to-mutation semantics; it never owns source position.
pub trait SourceRecordDecoder: Send + Sync {
    fn decoder_kind(&self) -> SourceDecoderKindV1;
    fn decode(&self, record: &RawSourceRecordV1) -> Result<Vec<DecodedSourceMutationV1>, SourceDecodeError>;
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SourceProviderError {
    Contract(SourceContractError),
    TemporarilyUnavailable,
    AuthenticationRefused,
    SourceIdentityChanged,
    AmbiguousPosition(SourcePartitionId),
    RetentionExpired {
        partition_id: SourcePartitionId,
        requested: SourceCursorV1,
        earliest_available: SourceCursorV1,
    },
}

impl Display for SourceProviderError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::Contract(error) => Display::fmt(error, formatter),
            Self::TemporarilyUnavailable => formatter.write_str("replay source is temporarily unavailable"),
            Self::AuthenticationRefused => formatter.write_str("replay source authentication was refused"),
            Self::SourceIdentityChanged => formatter.write_str("replay source identity changed"),
            Self::AmbiguousPosition(partition_id) => {
                write!(
                    formatter,
                    "replay source position is ambiguous for partition {partition_id}"
                )
            }
            Self::RetentionExpired {
                partition_id,
                requested,
                earliest_available,
            } => write!(
                formatter,
                "replay source retention expired for partition {partition_id}: requested {}, earliest available {}",
                requested.next_ordinal, earliest_available.next_ordinal
            ),
        }
    }
}

impl Error for SourceProviderError {}

impl From<SourceContractError> for SourceProviderError {
    fn from(error: SourceContractError) -> Self {
        Self::Contract(error)
    }
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SourceDecodeError {
    Contract(SourceContractError),
    UnsupportedEnvelopeVersion(u32),
    MissingTransactionMetadata(SourceOperationId),
    PoisonRecord(SourceOperationId),
    SchemaRefused(SourceOperationId),
}

impl Display for SourceDecodeError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::Contract(error) => Display::fmt(error, formatter),
            Self::UnsupportedEnvelopeVersion(version) => {
                write!(formatter, "unsupported source envelope version {version}")
            }
            Self::MissingTransactionMetadata(operation_id) => {
                write!(
                    formatter,
                    "source operation {operation_id} is missing transaction metadata"
                )
            }
            Self::PoisonRecord(operation_id) => write!(formatter, "source operation {operation_id} is malformed"),
            Self::SchemaRefused(operation_id) => {
                write!(
                    formatter,
                    "source operation {operation_id} was refused by the index schema"
                )
            }
        }
    }
}

impl Error for SourceDecodeError {}

impl From<SourceContractError> for SourceDecodeError {
    fn from(error: SourceContractError) -> Self {
        Self::Contract(error)
    }
}

pub fn encode_source_binding_v1(binding: &SourceBindingV1) -> Result<Vec<u8>, SourceContractError> {
    binding.validate()?;
    encode_with_budget(binding, "source configuration bytes", MAX_SOURCE_CONFIGURATION_BYTES)
}

pub fn encode_ingest_authority_v1(authority: &IngestAuthorityV1) -> Result<Vec<u8>, SourceContractError> {
    authority.validate_for_mode(IndexMode::Mutable)?;
    encode_with_budget(authority, "source configuration bytes", MAX_SOURCE_CONFIGURATION_BYTES)
}

pub fn decode_ingest_authority_v1(bytes: &[u8]) -> Result<IngestAuthorityV1, SourceContractError> {
    decode_current(
        bytes,
        "source configuration bytes",
        MAX_SOURCE_CONFIGURATION_BYTES,
        |authority: &IngestAuthorityV1| authority.validate_for_mode(IndexMode::Mutable),
    )
}

pub fn decode_source_binding_v1(bytes: &[u8]) -> Result<SourceBindingV1, SourceContractError> {
    decode_current(
        bytes,
        "source configuration bytes",
        MAX_SOURCE_CONFIGURATION_BYTES,
        SourceBindingV1::validate,
    )
}

pub fn encode_source_batch_v1(batch: &SourceBatchV1) -> Result<Vec<u8>, SourceContractError> {
    batch.validate()?;
    serde_json::to_vec(batch).map_err(SourceContractError::serde)
}

pub fn decode_source_batch_v1(bytes: &[u8]) -> Result<SourceBatchV1, SourceContractError> {
    decode_current(
        bytes,
        "source batch bytes",
        MAX_SOURCE_BATCH_BYTES,
        SourceBatchV1::validate,
    )
}

pub fn encode_atomic_source_publication_v1(commit: &AtomicSourcePublicationV1) -> Result<Vec<u8>, SourceContractError> {
    commit.validate()?;
    serde_json::to_vec(commit).map_err(SourceContractError::serde)
}

pub fn decode_atomic_source_publication_v1(bytes: &[u8]) -> Result<AtomicSourcePublicationV1, SourceContractError> {
    decode_current(
        bytes,
        "atomic source command bytes",
        MAX_ATOMIC_SOURCE_COMMAND_BYTES,
        AtomicSourcePublicationV1::validate,
    )
}

fn decode_current<T>(
    bytes: &[u8],
    budget_field: &'static str,
    maximum_bytes: usize,
    validate: impl FnOnce(&T) -> Result<(), SourceContractError>,
) -> Result<T, SourceContractError>
where
    T: DeserializeOwned + Serialize,
{
    #[derive(Deserialize)]
    struct FormatProbe {
        format_version: Option<u16>,
    }

    validate_encoded_budget(bytes.len(), budget_field, maximum_bytes)?;
    let probe: FormatProbe = serde_json::from_slice(bytes).map_err(SourceContractError::serde)?;
    let version = probe.format_version.ok_or(SourceContractError::MissingFormatVersion)?;
    validate_format(version)?;
    // INVARIANT: decode the original bytes directly. An intermediate map
    // would collapse duplicate keys before serde can refuse them.
    let decoded: T = serde_json::from_slice(bytes).map_err(SourceContractError::serde)?;
    validate(&decoded)?;
    let canonical = serde_json::to_vec(&decoded).map_err(SourceContractError::serde)?;
    if canonical != bytes {
        return Err(SourceContractError::NonCanonical);
    }
    Ok(decoded)
}

fn encode_with_budget<T: Serialize>(
    value: &T,
    budget_field: &'static str,
    maximum_bytes: usize,
) -> Result<Vec<u8>, SourceContractError> {
    let encoded = serde_json::to_vec(value).map_err(SourceContractError::serde)?;
    validate_encoded_budget(encoded.len(), budget_field, maximum_bytes)?;
    Ok(encoded)
}

fn validate_encoded_budget(observed: usize, field: &'static str, maximum: usize) -> Result<(), SourceContractError> {
    if observed > maximum {
        return Err(SourceContractError::Budget {
            field,
            observed,
            maximum,
        });
    }
    Ok(())
}

fn validate_format(version: u16) -> Result<(), SourceContractError> {
    if version != REPLAYABLE_SOURCE_FORMAT_VERSION {
        return Err(SourceContractError::UnsupportedFormat(version));
    }
    Ok(())
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub enum SourceContractError {
    UnsupportedFormat(u16),
    MissingFormatVersion,
    NonCanonical,
    ZeroIdentity(&'static str),
    UnsupportedMode(IndexMode),
    Budget {
        field: &'static str,
        observed: usize,
        maximum: usize,
    },
    Violation(&'static str),
    DeadlineElapsed,
    ArithmeticOverflow,
    Serde(String),
}

impl SourceContractError {
    fn serde(error: serde_json::Error) -> Self {
        Self::Serde(error.to_string())
    }
}

impl Display for SourceContractError {
    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
        match self {
            Self::UnsupportedFormat(version) => write!(formatter, "unsupported replayable-source format {version}"),
            Self::MissingFormatVersion => formatter.write_str("replayable-source format version is missing"),
            Self::NonCanonical => formatter.write_str("replayable-source bytes are not canonical current format"),
            Self::ZeroIdentity(kind) => write!(formatter, "{kind} cannot use the reserved zero identity"),
            Self::UnsupportedMode(mode) => write!(
                formatter,
                "replayable-source authority is unavailable for {mode:?} indexes"
            ),
            Self::Budget {
                field,
                observed,
                maximum,
            } => {
                write!(formatter, "{field} {observed} is outside 1..={maximum}")
            }
            Self::Violation(message) => formatter.write_str(message),
            Self::DeadlineElapsed => formatter.write_str("source fetch deadline elapsed"),
            Self::ArithmeticOverflow => formatter.write_str("source contract arithmetic overflowed"),
            Self::Serde(message) => write!(formatter, "replayable-source codec refused input: {message}"),
        }
    }
}

impl Error for SourceContractError {}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{CollectionId, Epoch, IndexId, RecoveryEra, ShardId};

    fn scope() -> Result<CollectionIndexRef, Box<dyn Error>> {
        Ok(CollectionIndexRef::new(CollectionId::new(1)?, IndexId::new(2)?))
    }

    fn cursor(ordinal: u64) -> SourceCursorV1 {
        SourceCursorV1 {
            next_ordinal: SourceOrdinal::new(ordinal),
            resume_token: ordinal.to_be_bytes().to_vec(),
        }
    }

    fn cut(ordinal: u64) -> Result<SourceCutV1, Box<dyn Error>> {
        Ok(SourceCutV1 {
            format_version: REPLAYABLE_SOURCE_FORMAT_VERSION,
            source_incarnation: SourceIncarnation::new(1)?,
            partitions: vec![SourcePartitionCheckpointV1 {
                partition_id: SourcePartitionId::new(0),
                partition_incarnation: SourcePartitionIncarnation::new(1)?,
                cursor: cursor(ordinal),
            }],
        })
    }

    fn batch() -> Result<SourceBatchV1, Box<dyn Error>> {
        Ok(SourceBatchV1 {
            format_version: REPLAYABLE_SOURCE_FORMAT_VERSION,
            batch_id: SourceBatchId::new(1)?,
            expected_cut: cut(0)?,
            replacement_cut: cut(1)?,
            records: vec![RawSourceRecordV1 {
                source_incarnation: SourceIncarnation::new(1)?,
                partition_id: SourcePartitionId::new(0),
                partition_incarnation: SourcePartitionIncarnation::new(1)?,
                position: cursor(0),
                next_position: cursor(1),
                operation_id: SourceOperationId::new(1)?,
                transaction: SourceTransactionMembershipV1::Standalone,
                key: b"doc-1".to_vec(),
                payload: br#"{"value":1}"#.to_vec(),
            }],
        })
    }

    #[test]
    fn current_batch_round_trips_exactly() -> Result<(), Box<dyn Error>> {
        let expected = batch()?;
        let bytes = encode_source_batch_v1(&expected)?;
        assert_eq!(decode_source_batch_v1(&bytes)?, expected);
        Ok(())
    }

    #[test]
    fn full_width_source_identities_use_canonical_decimal_strings() -> Result<(), Box<dyn Error>> {
        let operation_id = SourceOperationId::new(u128::MAX)?;
        assert_eq!(serde_json::to_string(&operation_id)?, format!("\"{}\"", u128::MAX));
        assert_eq!(
            serde_json::from_str::<SourceOperationId>(&format!("\"{}\"", u128::MAX))?,
            operation_id
        );

        for refused in ["0", "01", "+1", " 1", "1 ", "340282366920938463463374607431768211456"] {
            assert!(serde_json::from_str::<SourceBatchId>(&format!("\"{refused}\"")).is_err());
        }
        assert!(serde_json::from_str::<SourceTransactionId>("1").is_err());

        let mut expected = batch()?;
        expected.batch_id = SourceBatchId::new(u128::from(u64::MAX) + 1)?;
        expected.records[0].operation_id = SourceOperationId::new(u128::MAX)?;
        let bytes = encode_source_batch_v1(&expected)?;
        assert_eq!(decode_source_batch_v1(&bytes)?, expected);
        Ok(())
    }

    #[test]
    fn newer_unknown_and_noncanonical_inputs_refuse() -> Result<(), Box<dyn Error>> {
        let bytes = encode_source_batch_v1(&batch()?)?;
        let mut value: serde_json::Value = serde_json::from_slice(&bytes)?;
        value["format_version"] = serde_json::Value::from(2);
        let newer = serde_json::to_vec(&value)?;
        assert_eq!(
            decode_source_batch_v1(&newer),
            Err(SourceContractError::UnsupportedFormat(2))
        );

        let mut older: serde_json::Value = serde_json::from_slice(&bytes)?;
        older["format_version"] = serde_json::Value::from(0);
        assert_eq!(
            decode_source_batch_v1(&serde_json::to_vec(&older)?),
            Err(SourceContractError::UnsupportedFormat(0))
        );

        let mut missing: serde_json::Value = serde_json::from_slice(&bytes)?;
        missing
            .as_object_mut()
            .ok_or("source batch must encode as an object")?
            .remove("format_version");
        assert_eq!(
            decode_source_batch_v1(&serde_json::to_vec(&missing)?),
            Err(SourceContractError::MissingFormatVersion)
        );

        let mut value: serde_json::Value = serde_json::from_slice(&bytes)?;
        value["unknown"] = serde_json::Value::Bool(true);
        let unknown = serde_json::to_vec(&value)?;
        assert!(matches!(
            decode_source_batch_v1(&unknown),
            Err(SourceContractError::Serde(_))
        ));

        let spaced = [b" ".as_slice(), bytes.as_slice()].concat();
        assert_eq!(decode_source_batch_v1(&spaced), Err(SourceContractError::NonCanonical));

        let text = String::from_utf8(bytes)?;
        let duplicate = text.replacen("\"batch_id\":\"1\"", "\"batch_id\":\"1\",\"batch_id\":\"1\"", 1);
        if duplicate == text {
            return Err("batch id was not present in canonical bytes".into());
        }
        assert!(matches!(
            decode_source_batch_v1(duplicate.as_bytes()),
            Err(SourceContractError::Serde(_))
        ));
        Ok(())
    }

    #[test]
    fn incomplete_transaction_and_duplicate_operation_refuse() -> Result<(), Box<dyn Error>> {
        let mut incomplete = batch()?;
        incomplete.records[0].transaction = SourceTransactionMembershipV1::Transaction {
            transaction_id: SourceTransactionId::new(1)?,
            member_index: 0,
            member_count: NonZeroU32::new(2).ok_or("nonzero member count")?,
        };
        assert!(matches!(incomplete.validate(), Err(SourceContractError::Violation(_))));

        let mut duplicate = batch()?;
        let mut second = duplicate.records[0].clone();
        second.position = cursor(1);
        second.next_position = cursor(2);
        duplicate.replacement_cut = cut(2)?;
        duplicate.records.push(second);
        assert_eq!(
            duplicate.validate(),
            Err(SourceContractError::Violation(
                "source operation identity was duplicated"
            ))
        );
        Ok(())
    }

    #[test]
    fn replay_authority_mode_matrix_is_explicit() -> Result<(), Box<dyn Error>> {
        let binding = SourceBindingV1 {
            format_version: REPLAYABLE_SOURCE_FORMAT_VERSION,
            collection_index_ref: scope()?,
            source_incarnation: SourceIncarnation::new(1)?,
            source_identity_digest: [1; 32],
            provider: SourceProviderKindV1::Kafka,
            decoder: SourceDecoderKindV1::DebeziumPostgresJson,
            partitions: vec![SourcePartitionBindingV1 {
                partition_id: SourcePartitionId::new(0),
                partition_incarnation: SourcePartitionIncarnation::new(1)?,
            }],
            guarantees: SourceReplayGuaranteesV1 {
                durable_replay: true,
                stable_partition_identity: true,
                stable_operation_identity: true,
                ordered_positions: true,
                bounded_reads: true,
                explicit_retention: true,
                explicit_rewind: true,
                minimum_replay_window_millis: 86_400_000,
            },
        };
        binding.validate_cut(&cut(0)?)?;
        let authority = IngestAuthorityV1::replayable_source(binding);
        assert_eq!(authority.validate_for_mode(IndexMode::Mutable), Ok(()));
        assert_eq!(authority.validate_for_mode(IndexMode::AppendLog), Ok(()));
        assert_eq!(
            authority.validate_for_mode(IndexMode::State),
            Err(SourceContractError::UnsupportedMode(IndexMode::State))
        );
        assert_eq!(
            IngestAuthorityV1::push_with_wal().validate_for_mode(IndexMode::State),
            Ok(())
        );
        let encoded = encode_ingest_authority_v1(&authority)?;
        assert_eq!(decode_ingest_authority_v1(&encoded)?, authority);
        Ok(())
    }

    #[test]
    fn same_ordinal_token_drift_is_not_progress() -> Result<(), Box<dyn Error>> {
        let expected = cut(1)?;
        let mut replacement = cut(1)?;
        replacement.partitions[0].cursor.resume_token = b"different".to_vec();
        assert!(matches!(
            expected.advances_to(&replacement),
            Err(SourceContractError::Violation(_))
        ));
        Ok(())
    }

    #[test]
    fn fetch_response_requires_exact_watermark_and_retention_partition_sets() -> Result<(), Box<dyn Error>> {
        let mut response = SourceFetchResponseV1 {
            batch: batch()?,
            high_watermarks: vec![(
                SourcePartitionId::new(0),
                SourceHighWatermarkV1::Known { cursor: cursor(2) },
            )],
            retention_floors: vec![(SourcePartitionId::new(0), cursor(0))],
        };
        response.validate()?;
        response.retention_floors[0].1 = cursor(1);
        assert!(matches!(response.validate(), Err(SourceContractError::Violation(_))));
        Ok(())
    }

    #[test]
    fn atomic_cut_requires_one_scoped_successor_per_shard() -> Result<(), Box<dyn Error>> {
        let scope = scope()?;
        let commit = AtomicSourcePublicationV1 {
            format_version: REPLAYABLE_SOURCE_FORMAT_VERSION,
            collection_index_ref: scope,
            source_incarnation: SourceIncarnation::new(1)?,
            batch_id: SourceBatchId::new(1)?,
            expected_cut: cut(0)?,
            replacement_cut: cut(1)?,
            operation_count: 1,
            operation_set_digest: source_batch_operation_set_digest_v1(&batch()?)?,
            committed_transactions: Vec::new(),
            shard_publications: vec![SourceShardPublicationV1 {
                shard_key: ShardKey::new(scope, ShardId::new(1)),
                fence_token: FenceToken::new(RecoveryEra::new(1)?, Epoch::new(1))?,
                expected_generation: ManifestGeneration::new(4),
                replacement_generation: ManifestGeneration::new(5),
                replacement_root_digest: [3; 32],
            }],
        };
        let bytes = encode_atomic_source_publication_v1(&commit)?;
        assert_eq!(decode_atomic_source_publication_v1(&bytes)?, commit);
        commit.validate_against_batch(&batch()?)?;
        Ok(())
    }

    #[test]
    fn oversized_inputs_refuse_before_json_decode() {
        let oversized_configuration = vec![b' '; MAX_SOURCE_CONFIGURATION_BYTES + 1];
        assert_eq!(
            decode_source_binding_v1(&oversized_configuration),
            Err(SourceContractError::Budget {
                field: "source configuration bytes",
                observed: MAX_SOURCE_CONFIGURATION_BYTES + 1,
                maximum: MAX_SOURCE_CONFIGURATION_BYTES,
            })
        );

        let oversized_batch = vec![b' '; MAX_SOURCE_BATCH_BYTES + 1];
        assert_eq!(
            decode_source_batch_v1(&oversized_batch),
            Err(SourceContractError::Budget {
                field: "source batch bytes",
                observed: MAX_SOURCE_BATCH_BYTES + 1,
                maximum: MAX_SOURCE_BATCH_BYTES,
            })
        );

        let oversized_command = vec![b' '; MAX_ATOMIC_SOURCE_COMMAND_BYTES + 1];
        assert_eq!(
            decode_atomic_source_publication_v1(&oversized_command),
            Err(SourceContractError::Budget {
                field: "atomic source command bytes",
                observed: MAX_ATOMIC_SOURCE_COMMAND_BYTES + 1,
                maximum: MAX_ATOMIC_SOURCE_COMMAND_BYTES,
            })
        );
    }
}