use std::fmt;
use std::time::Duration;
use crate::ModelError;
pub const MAX_SOURCE_PINS: usize = 32;
pub const INCARNATION_BYTES: usize = 32;
pub const DIGEST_BYTES: usize = 32;
pub const COMMIT_ROOT_BYTES: usize = 32;
pub const ATTESTATION_SIGNATURE_BYTES: usize = 64;
pub const ATTESTATION_SIGNER_BYTES: usize = 32;
fn push_len_prefixed(buffer: &mut Vec<u8>, bytes: &[u8]) {
buffer.extend_from_slice(&u64::try_from(bytes.len()).unwrap_or(u64::MAX).to_be_bytes());
buffer.extend_from_slice(bytes);
}
const fn non_empty(field: &'static str, value: &str) -> Result<(), ModelError> {
if value.is_empty() {
return Err(ModelError::Bounds(field));
}
Ok(())
}
#[derive(Clone, PartialEq, Eq)]
pub struct JournalSource {
partition: String,
incarnation: [u8; INCARNATION_BYTES],
}
impl JournalSource {
pub fn try_new(
partition: String,
incarnation: [u8; INCARNATION_BYTES],
) -> Result<Self, ModelError> {
non_empty("partition", &partition)?;
Ok(Self {
partition,
incarnation,
})
}
#[must_use]
pub fn partition(&self) -> &str {
&self.partition
}
#[must_use]
pub const fn incarnation(&self) -> &[u8; INCARNATION_BYTES] {
&self.incarnation
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct JournalAttestation {
root: [u8; COMMIT_ROOT_BYTES],
leaf_count: u64,
signature: [u8; ATTESTATION_SIGNATURE_BYTES],
signer: [u8; ATTESTATION_SIGNER_BYTES],
}
impl JournalAttestation {
#[must_use]
pub const fn new(
root: [u8; COMMIT_ROOT_BYTES],
leaf_count: u64,
signature: [u8; ATTESTATION_SIGNATURE_BYTES],
signer: [u8; ATTESTATION_SIGNER_BYTES],
) -> Self {
Self {
root,
leaf_count,
signature,
signer,
}
}
#[must_use]
pub const fn root(&self) -> &[u8; COMMIT_ROOT_BYTES] {
&self.root
}
#[must_use]
pub const fn leaf_count(&self) -> u64 {
self.leaf_count
}
#[must_use]
pub const fn signature(&self) -> &[u8; ATTESTATION_SIGNATURE_BYTES] {
&self.signature
}
#[must_use]
pub const fn signer(&self) -> &[u8; ATTESTATION_SIGNER_BYTES] {
&self.signer
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct SourceCheckpoint {
source: JournalSource,
feed_position: u64,
journal_position: u64,
evidence_leaf: u64,
covering_attestation: JournalAttestation,
}
impl SourceCheckpoint {
#[must_use]
pub const fn new(
source: JournalSource,
feed_position: u64,
journal_position: u64,
evidence_leaf: u64,
covering_attestation: JournalAttestation,
) -> Self {
Self {
source,
feed_position,
journal_position,
evidence_leaf,
covering_attestation,
}
}
#[must_use]
pub const fn source(&self) -> &JournalSource {
&self.source
}
#[must_use]
pub const fn feed_position(&self) -> u64 {
self.feed_position
}
#[must_use]
pub const fn journal_position(&self) -> u64 {
self.journal_position
}
#[must_use]
pub const fn evidence_leaf(&self) -> u64 {
self.evidence_leaf
}
#[must_use]
pub const fn covering_attestation(&self) -> &JournalAttestation {
&self.covering_attestation
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Retention {
For(Duration),
UntilReleased,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Classification {
Public,
Internal,
Confidential,
Restricted,
}
#[derive(Clone, PartialEq, Eq)]
pub struct ObjectDescriptor {
object: String,
generation: u64,
digest: [u8; DIGEST_BYTES],
owner: String,
classification: Classification,
retention: Retention,
byte_len: u64,
content_reference: String,
}
impl ObjectDescriptor {
#[allow(
clippy::too_many_arguments,
reason = "every field of the signed descriptor is named explicitly"
)]
pub fn try_new(
object: String,
generation: u64,
digest: [u8; DIGEST_BYTES],
owner: String,
classification: Classification,
retention: Retention,
byte_len: u64,
content_reference: String,
) -> Result<Self, ModelError> {
non_empty("object", &object)?;
non_empty("owner", &owner)?;
non_empty("content_reference", &content_reference)?;
Ok(Self {
object,
generation,
digest,
owner,
classification,
retention,
byte_len,
content_reference,
})
}
#[must_use]
pub fn object(&self) -> &str {
&self.object
}
#[must_use]
pub const fn generation(&self) -> u64 {
self.generation
}
#[must_use]
pub const fn digest(&self) -> &[u8; DIGEST_BYTES] {
&self.digest
}
#[must_use]
pub fn owner(&self) -> &str {
&self.owner
}
#[must_use]
pub const fn classification(&self) -> Classification {
self.classification
}
#[must_use]
pub const fn retention(&self) -> Retention {
self.retention
}
#[must_use]
pub const fn byte_len(&self) -> u64 {
self.byte_len
}
#[must_use]
pub fn content_reference(&self) -> &str {
&self.content_reference
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct ExactObjectRef {
namespace: String,
key: String,
backend_generation: u64,
}
impl ExactObjectRef {
pub fn try_new(
namespace: String,
key: String,
backend_generation: u64,
) -> Result<Self, ModelError> {
non_empty("namespace", &namespace)?;
non_empty("key", &key)?;
Ok(Self {
namespace,
key,
backend_generation,
})
}
#[must_use]
pub fn namespace(&self) -> &str {
&self.namespace
}
#[must_use]
pub fn key(&self) -> &str {
&self.key
}
#[must_use]
pub const fn backend_generation(&self) -> u64 {
self.backend_generation
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct ProjectionKey {
family: String,
source_partition: String,
}
impl ProjectionKey {
pub fn try_new(family: String, source_partition: String) -> Result<Self, ModelError> {
non_empty("family", &family)?;
non_empty("source_partition", &source_partition)?;
Ok(Self {
family,
source_partition,
})
}
#[must_use]
pub fn family(&self) -> &str {
&self.family
}
#[must_use]
pub fn source_partition(&self) -> &str {
&self.source_partition
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct PublisherFence {
key: ProjectionKey,
source_incarnation: [u8; INCARNATION_BYTES],
term: u64,
}
impl PublisherFence {
pub fn try_new(
key: ProjectionKey,
source_incarnation: [u8; INCARNATION_BYTES],
term: u64,
) -> Result<Self, ModelError> {
if term == 0 {
return Err(ModelError::Bounds("term"));
}
Ok(Self {
key,
source_incarnation,
term,
})
}
#[must_use]
pub const fn key(&self) -> &ProjectionKey {
&self.key
}
#[must_use]
pub const fn source_incarnation(&self) -> &[u8; INCARNATION_BYTES] {
&self.source_incarnation
}
#[must_use]
pub const fn term(&self) -> u64 {
self.term
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct ProjectionManifest {
key: ProjectionKey,
generation: u64,
checkpoint: SourceCheckpoint,
schema_version: u32,
fact_version: u32,
object: ObjectDescriptor,
artifact_object: ExactObjectRef,
publisher: String,
fence: PublisherFence,
}
impl ProjectionManifest {
#[allow(
clippy::too_many_arguments,
reason = "every field of the signed manifest is named explicitly"
)]
pub fn try_new(
key: ProjectionKey,
generation: u64,
checkpoint: SourceCheckpoint,
schema_version: u32,
fact_version: u32,
object: ObjectDescriptor,
artifact_object: ExactObjectRef,
publisher: String,
fence: PublisherFence,
) -> Result<Self, ModelError> {
if generation == 0 {
return Err(ModelError::Bounds("generation"));
}
if schema_version == 0 {
return Err(ModelError::Bounds("schema_version"));
}
if fact_version == 0 {
return Err(ModelError::Bounds("fact_version"));
}
non_empty("publisher", &publisher)?;
Ok(Self {
key,
generation,
checkpoint,
schema_version,
fact_version,
object,
artifact_object,
publisher,
fence,
})
}
#[must_use]
pub const fn key(&self) -> &ProjectionKey {
&self.key
}
#[must_use]
pub const fn generation(&self) -> u64 {
self.generation
}
#[must_use]
pub const fn checkpoint(&self) -> &SourceCheckpoint {
&self.checkpoint
}
#[must_use]
pub const fn schema_version(&self) -> u32 {
self.schema_version
}
#[must_use]
pub const fn fact_version(&self) -> u32 {
self.fact_version
}
#[must_use]
pub const fn object(&self) -> &ObjectDescriptor {
&self.object
}
#[must_use]
pub const fn artifact_object(&self) -> &ExactObjectRef {
&self.artifact_object
}
#[must_use]
pub fn publisher(&self) -> &str {
&self.publisher
}
#[must_use]
pub const fn fence(&self) -> &PublisherFence {
&self.fence
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct JournalAnchor {
source: JournalSource,
head: u64,
}
impl JournalAnchor {
#[must_use]
pub const fn new(source: JournalSource, head: u64) -> Self {
Self { source, head }
}
#[must_use]
pub const fn source(&self) -> &JournalSource {
&self.source
}
#[must_use]
pub const fn head(&self) -> u64 {
self.head
}
}
#[derive(Clone, PartialEq, Eq)]
pub enum SourcePin {
Projected(Box<ProjectionManifest>),
Journal(JournalAnchor),
Authoritative(u64),
}
impl SourcePin {
#[must_use]
pub fn identity_bytes(&self) -> Vec<u8> {
let mut bytes = Vec::new();
match self {
Self::Projected(manifest) => {
bytes.push(0);
push_len_prefixed(&mut bytes, manifest.key().family().as_bytes());
push_len_prefixed(&mut bytes, manifest.key().source_partition().as_bytes());
}
Self::Journal(anchor) => {
bytes.push(1);
push_len_prefixed(&mut bytes, anchor.source().partition().as_bytes());
}
Self::Authoritative(_) => bytes.push(2),
}
bytes
}
}
#[derive(Clone, PartialEq, Eq)]
pub struct SourceEvidence {
pins: Vec<SourcePin>,
}
impl SourceEvidence {
pub fn try_new(pins: Vec<SourcePin>) -> Result<Self, ModelError> {
if pins.len() > MAX_SOURCE_PINS {
return Err(ModelError::Bounds("source_pins"));
}
for pair in pins.windows(2) {
if pair[0].identity_bytes() >= pair[1].identity_bytes() {
return Err(ModelError::Order("source_pins"));
}
}
Ok(Self { pins })
}
#[must_use]
pub fn pins(&self) -> &[SourcePin] {
&self.pins
}
}
impl fmt::Debug for JournalSource {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("JournalSource")
}
}
impl fmt::Debug for JournalAttestation {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("JournalAttestation")
.field("leaf_count", &self.leaf_count)
.finish_non_exhaustive()
}
}
impl fmt::Debug for SourceCheckpoint {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("SourceCheckpoint")
.field("feed_position", &self.feed_position)
.field("journal_position", &self.journal_position)
.finish_non_exhaustive()
}
}
impl fmt::Debug for ObjectDescriptor {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ObjectDescriptor")
.field("generation", &self.generation)
.field("classification", &self.classification)
.field("byte_len", &self.byte_len)
.finish_non_exhaustive()
}
}
impl fmt::Debug for ExactObjectRef {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ExactObjectRef")
.field("backend_generation", &self.backend_generation)
.finish_non_exhaustive()
}
}
impl fmt::Debug for ProjectionKey {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str("ProjectionKey")
}
}
impl fmt::Debug for PublisherFence {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("PublisherFence")
.field("term", &self.term)
.finish_non_exhaustive()
}
}
impl fmt::Debug for ProjectionManifest {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ProjectionManifest")
.field("generation", &self.generation)
.field("schema_version", &self.schema_version)
.field("fact_version", &self.fact_version)
.finish_non_exhaustive()
}
}
impl fmt::Debug for JournalAnchor {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("JournalAnchor")
.field("head", &self.head)
.finish_non_exhaustive()
}
}
impl fmt::Debug for SourcePin {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(match self {
Self::Projected(_) => "projected",
Self::Journal(_) => "journal",
Self::Authoritative(_) => "authoritative",
})
}
}
impl fmt::Debug for SourceEvidence {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("SourceEvidence")
.field("pins", &self.pins.len())
.finish()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn journal(partition: &str) -> SourcePin {
SourcePin::Journal(JournalAnchor::new(
JournalSource::try_new(partition.to_owned(), [7; INCARNATION_BYTES]).unwrap(),
4,
))
}
#[test]
fn pins_must_be_strictly_ascending_and_bounded() {
assert!(SourceEvidence::try_new(vec![journal("a"), journal("b")]).is_ok());
assert_eq!(
SourceEvidence::try_new(vec![journal("b"), journal("a")]),
Err(ModelError::Order("source_pins"))
);
assert_eq!(
SourceEvidence::try_new(vec![journal("a"), journal("a")]),
Err(ModelError::Order("source_pins")),
"one partition may be anchored once"
);
let overflowing = (0..=MAX_SOURCE_PINS)
.map(|index| journal(&format!("p{index:04}")))
.collect();
assert_eq!(
SourceEvidence::try_new(overflowing),
Err(ModelError::Bounds("source_pins"))
);
}
#[test]
fn identity_ignores_everything_except_the_source_it_names() {
let first = SourcePin::Journal(JournalAnchor::new(
JournalSource::try_new("a".to_owned(), [1; INCARNATION_BYTES]).unwrap(),
1,
));
let second = SourcePin::Journal(JournalAnchor::new(
JournalSource::try_new("a".to_owned(), [2; INCARNATION_BYTES]).unwrap(),
9,
));
assert_eq!(first.identity_bytes(), second.identity_bytes());
}
#[test]
fn structural_bounds_refuse_empty_identifiers() {
assert_eq!(
JournalSource::try_new(String::new(), [0; INCARNATION_BYTES]),
Err(ModelError::Bounds("partition"))
);
assert_eq!(
ProjectionKey::try_new(String::new(), "p".to_owned()),
Err(ModelError::Bounds("family"))
);
assert_eq!(
ExactObjectRef::try_new("ns".to_owned(), String::new(), 1),
Err(ModelError::Bounds("key"))
);
}
}