use crate::envelope::EnvelopeCodecError;
use crate::WriterEpoch;
use crate::{
wal_segment_id_start_seq, ChangeSeq, CheckpointId, Checksum, ChecksumAlgorithm, CommitId,
ContentId, ContentRef, ContentRefKind, ContentStoreId, InodeId, ManifestNo, ManifestObjectId,
MetadataCompactionId, NamespaceId, UploadId, WalSegmentId,
};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Deserializer, Serialize};
use std::fmt;
use std::num::NonZeroU64;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ControlObjectKind {
WalHead,
WalFloor,
MetadataRoot,
CheckpointRecord,
UploadSession,
CompactionLease,
}
impl ControlObjectKind {
pub const ALL: [Self; 6] = [
Self::WalHead,
Self::WalFloor,
Self::MetadataRoot,
Self::CheckpointRecord,
Self::UploadSession,
Self::CompactionLease,
];
pub const fn format_version(self) -> u32 {
match self {
Self::WalHead => 1,
Self::WalFloor => 1,
Self::MetadataRoot => 1,
Self::CheckpointRecord => 1,
Self::UploadSession => 1,
Self::CompactionLease => 1,
}
}
pub const fn as_str(self) -> &'static str {
match self {
Self::WalHead => "wal_head",
Self::WalFloor => "wal_floor",
Self::MetadataRoot => "metadata_root",
Self::CheckpointRecord => "checkpoint_record",
Self::UploadSession => "upload_session",
Self::CompactionLease => "compaction_lease",
}
}
pub fn parse(value: &str) -> Option<Self> {
Self::ALL.into_iter().find(|kind| kind.as_str() == value)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WalFloorState {
pub namespace_id: NamespaceId,
pub floor_seq: ChangeSeq,
pub updated_at_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ManifestRef {
pub owner_namespace_id: NamespaceId,
pub manifest_no: ManifestNo,
pub manifest_object_id: ManifestObjectId,
pub manifest_head_seq: ChangeSeq,
pub manifest_payload_checksum: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MetadataRootState {
pub namespace_id: NamespaceId,
pub manifest: ManifestRef,
pub updated_at_ms: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CompactionLeaseStatus {
Active {},
Reaping {},
}
impl fmt::Display for CompactionLeaseStatus {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(match self {
Self::Active {} => "active",
Self::Reaping {} => "reaping",
})
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MetadataCompactionLeaseState {
pub job_id: MetadataCompactionId,
pub namespace_id: NamespaceId,
pub writer_id: String,
pub status: CompactionLeaseStatus,
pub started_at_ms: u64,
pub heartbeat_at_ms: u64,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CheckpointStatus {
Active {},
Released {
released_at_ms: u64,
},
}
impl std::fmt::Display for CheckpointStatus {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let status = match self {
Self::Active {} => "active",
Self::Released { .. } => "released",
};
formatter.write_str(status)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum CheckpointOwner {
User {
name: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
expires_at_ms: Option<u64>,
},
Fork {
target_namespace_id: NamespaceId,
expires_at_ms: u64,
},
Snapshot {
name: String,
expires_at_ms: u64,
},
}
impl CheckpointOwner {
pub fn expires_at_ms(&self) -> Option<u64> {
match self {
Self::User { expires_at_ms, .. } => *expires_at_ms,
Self::Fork { expires_at_ms, .. } => Some(*expires_at_ms),
Self::Snapshot { expires_at_ms, .. } => Some(*expires_at_ms),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct CheckpointRecordState {
pub checkpoint_id: CheckpointId,
pub namespace_id: NamespaceId,
pub manifest: ManifestRef,
pub head_commit_id: CommitId,
pub created_at_ms: u64,
pub owner: CheckpointOwner,
pub status: CheckpointStatus,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct WalSegmentPointer {
pub segment_id: WalSegmentId,
pub start_seq: ChangeSeq,
pub end_seq: ChangeSeq,
pub payload_checksum: String,
}
impl<'de> Deserialize<'de> for WalSegmentPointer {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
#[derive(Deserialize)]
struct StoredWalSegmentPointer {
segment_id: WalSegmentId,
start_seq: ChangeSeq,
end_seq: ChangeSeq,
payload_checksum: String,
}
let stored = StoredWalSegmentPointer::deserialize(deserializer)?;
validated_wal_segment_pointer(Self {
segment_id: stored.segment_id,
start_seq: stored.start_seq,
end_seq: stored.end_seq,
payload_checksum: stored.payload_checksum,
})
}
}
pub(crate) fn validate_wal_segment_start_seq(
segment_id: &WalSegmentId,
start_seq: ChangeSeq,
) -> Result<(), String> {
if wal_segment_id_start_seq(segment_id.as_str()) == Some(start_seq) {
return Ok(());
}
Err(format!(
"wal segment id `{segment_id}` does not encode start seq `{start_seq}`"
))
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct StrictWalSegmentPointer {
segment_id: WalSegmentId,
start_seq: ChangeSeq,
end_seq: ChangeSeq,
payload_checksum: String,
}
impl From<StrictWalSegmentPointer> for WalSegmentPointer {
fn from(pointer: StrictWalSegmentPointer) -> Self {
Self {
segment_id: pointer.segment_id,
start_seq: pointer.start_seq,
end_seq: pointer.end_seq,
payload_checksum: pointer.payload_checksum,
}
}
}
fn validated_wal_segment_pointer<E>(pointer: WalSegmentPointer) -> Result<WalSegmentPointer, E>
where
E: serde::de::Error,
{
validate_wal_segment_start_seq(&pointer.segment_id, pointer.start_seq).map_err(E::custom)?;
Ok(pointer)
}
fn strict_wal_segment_pointer<'de, D>(
deserializer: D,
) -> Result<Option<WalSegmentPointer>, D::Error>
where
D: Deserializer<'de>,
{
Option::<StrictWalSegmentPointer>::deserialize(deserializer)?
.map(|pointer| validated_wal_segment_pointer(pointer.into()))
.transpose()
}
fn strict_wal_segment_pointers<'de, D>(deserializer: D) -> Result<Vec<WalSegmentPointer>, D::Error>
where
D: Deserializer<'de>,
{
Vec::<StrictWalSegmentPointer>::deserialize(deserializer)?
.into_iter()
.map(|pointer| validated_wal_segment_pointer(pointer.into()))
.collect()
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WriterBlock {
pub writer_id: String,
pub acquired_at_ms: u64,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct AcquiredWriter {
pub writer_id: String,
pub writer_epoch: WriterEpoch,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum NamespaceStatus {
Active {},
Deleted {},
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct ForkBasis {
pub manifest: ManifestRef,
pub source_checkpoint_id: CheckpointId,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct HeadState {
pub namespace_id: NamespaceId,
pub content_store_id: ContentStoreId,
pub created_at_ms: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub fork_basis: Option<ForkBasis>,
pub seq: ChangeSeq,
pub head_commit_id: CommitId,
pub writer_epoch: WriterEpoch,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub writer: Option<WriterBlock>,
pub next_inode_id: InodeId,
#[serde(
default,
skip_serializing_if = "Option::is_none",
deserialize_with = "strict_wal_segment_pointer"
)]
pub visible_wal_tip: Option<WalSegmentPointer>,
#[serde(deserialize_with = "strict_wal_segment_pointers")]
pub recent_segments: Vec<WalSegmentPointer>,
pub status: NamespaceStatus,
}
const GENESIS_COMMIT_ID: &str = "c_00000000000000000000000000000000";
pub fn genesis_commit_id() -> CommitId {
CommitId::parse(GENESIS_COMMIT_ID).expect("genesis commit id is valid")
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct HeadIdentityDrift {
pub field: String,
}
impl fmt::Display for HeadIdentityDrift {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
write!(
formatter,
"successor head changes the namespace's immutable `{}`",
self.field
)
}
}
impl HeadState {
pub fn initial(
namespace_id: NamespaceId,
content_store_id: ContentStoreId,
created_at_ms: u64,
) -> Self {
Self {
namespace_id,
content_store_id,
created_at_ms,
fork_basis: None,
seq: ChangeSeq(0),
head_commit_id: CommitId::parse(GENESIS_COMMIT_ID).expect("genesis commit id is valid"),
writer_epoch: WriterEpoch(0),
writer: None,
next_inode_id: crate::FIRST_ALLOCATABLE_INODE_ID,
visible_wal_tip: None,
recent_segments: Vec::new(),
status: NamespaceStatus::Active {},
}
}
pub fn ensure_successor_identity(
&self,
successor: &HeadState,
) -> Result<(), HeadIdentityDrift> {
let drift = |field: &str| {
Err(HeadIdentityDrift {
field: field.to_owned(),
})
};
if successor.namespace_id != self.namespace_id {
return drift("namespace_id");
}
if successor.content_store_id != self.content_store_id {
return drift("content_store_id");
}
if successor.created_at_ms != self.created_at_ms {
return drift("created_at_ms");
}
if successor.fork_basis != self.fork_basis {
return drift("fork_basis");
}
Ok(())
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ProxiedStaging {
Idle,
Claimed,
Staged(ContentRef),
}
impl Serialize for ProxiedStaging {
fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
where
S: serde::Serializer,
{
#[derive(Serialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
enum Shape<'a> {
Idle {},
Claimed {},
Staged { content_ref: &'a ContentRef },
}
match self {
Self::Idle => Shape::Idle {}.serialize(serializer),
Self::Claimed => Shape::Claimed {}.serialize(serializer),
Self::Staged(content_ref) => Shape::Staged { content_ref }.serialize(serializer),
}
}
}
impl<'de> Deserialize<'de> for ProxiedStaging {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
StrictProxiedStaging::deserialize(deserializer).map(Into::into)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum UploadSessionMode {
ServiceProxied {
staging: ProxiedStaging,
},
DirectPut {
checksum_algorithm: ChecksumAlgorithm,
},
DirectMultipart {
provider_upload_id: String,
part_size_bytes: NonZeroU64,
checksum_algorithm: ChecksumAlgorithm,
},
}
impl UploadSessionMode {
fn content_ref(&self) -> Option<&ContentRef> {
match self {
Self::ServiceProxied {
staging: ProxiedStaging::Staged(content_ref),
} => Some(content_ref),
Self::ServiceProxied { .. } | Self::DirectPut { .. } | Self::DirectMultipart { .. } => {
None
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum UploadSessionRecordStatus {
Open {
expires_at_ms: u64,
},
Completed {
completed_at_ms: u64,
content_ref: ContentRef,
},
Aborted {
aborted_at_ms: u64,
},
}
impl UploadSessionRecordStatus {
fn content_ref(&self) -> Option<&ContentRef> {
match self {
Self::Open { .. } => None,
Self::Completed { content_ref, .. } => Some(content_ref),
Self::Aborted { .. } => None,
}
}
}
impl std::fmt::Display for UploadSessionRecordStatus {
fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
let status = match self {
Self::Open { .. } => "open",
Self::Completed { .. } => "completed",
Self::Aborted { .. } => "aborted",
};
formatter.write_str(status)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
pub struct UploadSessionState {
pub namespace_id: NamespaceId,
pub upload_id: UploadId,
pub content_id: ContentId,
pub created_at_ms: u64,
pub mode: UploadSessionMode,
pub status: UploadSessionRecordStatus,
}
impl UploadSessionState {
fn validate(&self) -> Result<(), String> {
for content_ref in self
.mode
.content_ref()
.into_iter()
.chain(self.status.content_ref())
{
content_ref.validate().map_err(|error| {
format!(
"upload session `{}` holds an invalid content ref: {error}",
self.upload_id
)
})?;
if content_ref.content_id != self.content_id {
return Err(format!(
"upload session `{}` owns content `{}` but holds a reference to `{}`",
self.upload_id, self.content_id, content_ref.content_id
));
}
}
if let (
UploadSessionMode::DirectPut { checksum_algorithm },
UploadSessionRecordStatus::Completed { content_ref, .. },
) = (&self.mode, &self.status)
{
if content_ref.checksum.algorithm != *checksum_algorithm {
return Err(format!(
"upload session `{}` requires `{checksum_algorithm}` but its completed \
content uses `{}`",
self.upload_id, content_ref.checksum.algorithm
));
}
}
Ok(())
}
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct StrictUploadSessionState {
namespace_id: NamespaceId,
upload_id: UploadId,
content_id: ContentId,
created_at_ms: u64,
mode: StrictUploadSessionMode,
status: StrictUploadSessionRecordStatus,
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum StrictUploadSessionMode {
ServiceProxied {
staging: StrictProxiedStaging,
},
DirectPut {
checksum_algorithm: ChecksumAlgorithm,
},
DirectMultipart {
provider_upload_id: String,
part_size_bytes: NonZeroU64,
checksum_algorithm: ChecksumAlgorithm,
},
}
impl From<StrictUploadSessionMode> for UploadSessionMode {
fn from(mode: StrictUploadSessionMode) -> Self {
match mode {
StrictUploadSessionMode::ServiceProxied { staging } => Self::ServiceProxied {
staging: staging.into(),
},
StrictUploadSessionMode::DirectPut { checksum_algorithm } => {
Self::DirectPut { checksum_algorithm }
}
StrictUploadSessionMode::DirectMultipart {
provider_upload_id,
part_size_bytes,
checksum_algorithm,
} => Self::DirectMultipart {
provider_upload_id,
part_size_bytes,
checksum_algorithm,
},
}
}
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum StrictProxiedStaging {
Idle {},
Claimed {},
Staged { content_ref: StrictContentRef },
}
impl From<StrictProxiedStaging> for ProxiedStaging {
fn from(staging: StrictProxiedStaging) -> Self {
match staging {
StrictProxiedStaging::Idle {} => Self::Idle,
StrictProxiedStaging::Claimed {} => Self::Claimed,
StrictProxiedStaging::Staged { content_ref } => Self::Staged(content_ref.into()),
}
}
}
#[derive(Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
enum StrictUploadSessionRecordStatus {
Open {
expires_at_ms: u64,
},
Completed {
completed_at_ms: u64,
content_ref: StrictContentRef,
},
Aborted {
aborted_at_ms: u64,
},
}
impl From<StrictUploadSessionRecordStatus> for UploadSessionRecordStatus {
fn from(status: StrictUploadSessionRecordStatus) -> Self {
match status {
StrictUploadSessionRecordStatus::Open { expires_at_ms } => Self::Open { expires_at_ms },
StrictUploadSessionRecordStatus::Completed {
completed_at_ms,
content_ref,
} => Self::Completed {
completed_at_ms,
content_ref: content_ref.into(),
},
StrictUploadSessionRecordStatus::Aborted { aborted_at_ms } => {
Self::Aborted { aborted_at_ms }
}
}
}
}
#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct StrictContentRef {
kind: MutableContentRefKind,
content_id: ContentId,
size_bytes: u64,
checksum: Checksum,
}
#[derive(Deserialize)]
#[serde(rename_all = "snake_case")]
enum MutableContentRefKind {
BlobV1,
}
impl From<StrictContentRef> for ContentRef {
fn from(content_ref: StrictContentRef) -> Self {
let kind = match content_ref.kind {
MutableContentRefKind::BlobV1 => ContentRefKind::BlobV1,
};
Self {
kind,
content_id: content_ref.content_id,
size_bytes: content_ref.size_bytes,
checksum: content_ref.checksum,
}
}
}
impl<'de> Deserialize<'de> for UploadSessionState {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: Deserializer<'de>,
{
let record = StrictUploadSessionState::deserialize(deserializer)?;
let session = Self {
namespace_id: record.namespace_id,
upload_id: record.upload_id,
content_id: record.content_id,
created_at_ms: record.created_at_ms,
mode: record.mode.into(),
status: record.status.into(),
};
session.validate().map_err(serde::de::Error::custom)?;
Ok(session)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ControlObjectEnvelope<T> {
pub kind: ControlObjectKind,
pub format_version: u32,
pub payload_checksum: String,
pub state: T,
}
impl<T> ControlObjectEnvelope<T>
where
T: Serialize,
{
pub fn from_state(kind: ControlObjectKind, state: T) -> Result<Self, EnvelopeCodecError> {
Ok(Self {
kind,
format_version: kind.format_version(),
payload_checksum: control_payload_checksum(&state)?,
state,
})
}
}
pub type HeadStateEnvelope = ControlObjectEnvelope<HeadState>;
pub type UploadSessionEnvelope = ControlObjectEnvelope<UploadSessionState>;
pub type MetadataRootEnvelope = ControlObjectEnvelope<MetadataRootState>;
pub type WalFloorEnvelope = ControlObjectEnvelope<WalFloorState>;
pub type CheckpointRecordEnvelope = ControlObjectEnvelope<CheckpointRecordState>;
pub type MetadataCompactionLeaseEnvelope = ControlObjectEnvelope<MetadataCompactionLeaseState>;
pub fn control_payload_checksum<T>(state: &T) -> Result<String, EnvelopeCodecError>
where
T: Serialize,
{
crate::envelope::json_payload_checksum(state)
}
pub fn encode_control_object<T>(
envelope: &ControlObjectEnvelope<T>,
) -> Result<Vec<u8>, EnvelopeCodecError>
where
T: Serialize,
{
crate::envelope::encode_json_envelope(
envelope.kind.as_str(),
envelope.format_version,
envelope.kind.format_version(),
&envelope.payload_checksum,
&envelope.state,
)
}
pub fn encode_control_state<T: Serialize>(
kind: ControlObjectKind,
state: &T,
) -> Result<Vec<u8>, EnvelopeCodecError> {
let envelope = ControlObjectEnvelope::from_state(kind, state)?;
encode_control_object(&envelope)
}
pub fn decode_control_object<T>(
bytes: &[u8],
expected_kind: ControlObjectKind,
) -> Result<ControlObjectEnvelope<T>, EnvelopeCodecError>
where
T: DeserializeOwned,
{
let decoded = crate::envelope::decode_strict_json_envelope(
bytes,
expected_kind.format_version(),
|found| match ControlObjectKind::parse(found) {
None => Err(EnvelopeCodecError::UnknownKind {
found: found.to_owned(),
}),
Some(kind) if kind != expected_kind => Err(EnvelopeCodecError::KindMismatch {
expected: expected_kind.as_str().to_owned(),
found: found.to_owned(),
}),
Some(_) => Ok(()),
},
)?;
Ok(ControlObjectEnvelope {
kind: expected_kind,
format_version: decoded.format_version,
payload_checksum: decoded.payload_checksum,
state: decoded.payload,
})
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn control_object_kind_strings_round_trip_and_match_serde() {
for kind in ControlObjectKind::ALL {
assert_eq!(ControlObjectKind::parse(kind.as_str()), Some(kind));
let serialized = serde_json::to_value(kind).expect("serialize kind");
assert_eq!(serialized, serde_json::Value::from(kind.as_str()));
}
assert_eq!(ControlObjectKind::parse("not_a_kind"), None);
}
fn sample_head() -> HeadState {
HeadState::initial(
NamespaceId::parse("demo").expect("valid namespace id"),
ContentStoreId::parse("cs_0123456789abcdef0123456789abcdef")
.expect("valid content store id"),
1_000,
)
}
#[test]
fn head_without_content_store_is_rejected() {
let mut missing = head_json(None, Vec::new());
missing
.as_object_mut()
.expect("head payload object")
.remove("content_store_id");
let error = serde_json::from_value::<HeadState>(missing)
.expect_err("head without its immutable identity must be rejected");
assert!(
error.to_string().contains("content_store_id"),
"the rejection should name the missing field: {error}"
);
}
fn wal_pointer_json(segment_id: &str, start_seq: u64, end_seq: u64) -> serde_json::Value {
serde_json::json!({
"segment_id": segment_id,
"start_seq": start_seq,
"end_seq": end_seq,
"payload_checksum": format!("sha256:{}", "b".repeat(64)),
})
}
fn head_json(
visible_wal_tip: Option<serde_json::Value>,
recent_segments: Vec<serde_json::Value>,
) -> serde_json::Value {
let mut head = serde_json::json!({
"namespace_id": "demo",
"content_store_id": "cs_0123456789abcdef0123456789abcdef",
"created_at_ms": 1_000,
"seq": 2,
"head_commit_id": GENESIS_COMMIT_ID,
"writer_epoch": 0,
"next_inode_id": 2,
"status": { "kind": "active" }
});
if let Some(tip) = visible_wal_tip {
head["visible_wal_tip"] = tip;
}
head["recent_segments"] = serde_json::Value::Array(recent_segments);
head
}
#[test]
fn a_head_decodes_its_tip_with_and_without_predecessor_hints() {
let tip = wal_pointer_json("wal_00000000000000000002-fedcba9876543210", 2, 2);
let older = wal_pointer_json("wal_00000000000000000001-0123456789abcdef", 1, 1);
let head = serde_json::from_value::<HeadState>(head_json(Some(tip.clone()), Vec::new()))
.expect("the first published segment has no predecessor hints");
assert!(head.visible_wal_tip.is_some());
assert!(head.recent_segments.is_empty());
let head = serde_json::from_value::<HeadState>(head_json(Some(tip), vec![older.clone()]))
.expect("predecessor hints decode independently of the authoritative tip");
assert_eq!(
head.recent_segments,
vec![serde_json::from_value(older).expect("valid predecessor pointer")]
);
}
#[test]
fn head_rejects_a_pointer_field_it_does_not_define() {
let mut tip = wal_pointer_json("wal_00000000000000000002-fedcba9876543210", 2, 2);
tip["object_key"] = serde_json::json!(
"namespaces/demo/wal/segments/wal_00000000000000000002-fedcba9876543210.wal.zst"
);
serde_json::from_value::<HeadState>(head_json(Some(tip.clone()), Vec::new()))
.expect_err("the head rejects a field its tip pointer does not define");
let older = wal_pointer_json("wal_00000000000000000001-0123456789abcdef", 1, 1);
serde_json::from_value::<HeadState>(head_json(Some(older), vec![tip]))
.expect_err("the head rejects a field a predecessor hint does not define");
}
#[test]
fn wal_pointers_reject_an_id_that_disagrees_with_its_start_seq() {
let agreeing = wal_pointer_json("wal_00000000000000000002-fedcba9876543210", 2, 2);
serde_json::from_value::<WalSegmentPointer>(agreeing)
.expect("a pointer whose id encodes its start seq decodes");
let disagreeing = wal_pointer_json("wal_00000000000000000003-fedcba9876543210", 2, 2);
let error = serde_json::from_value::<WalSegmentPointer>(disagreeing)
.expect_err("a pointer whose id disagrees with its start seq is corruption");
let message = error.to_string();
assert!(
message.contains("`wal_00000000000000000003-fedcba9876543210`")
&& message.contains("start seq `2`"),
"the rejection should name both values: {message}"
);
}
#[test]
fn the_head_rejects_a_pointer_whose_id_disagrees_with_its_start_seq() {
let tip = wal_pointer_json("wal_00000000000000000003-aaaaaaaaaaaaaaaa", 3, 3);
let older = wal_pointer_json("wal_00000000000000000002-fedcba9876543210", 2, 2);
serde_json::from_value::<HeadState>(head_json(Some(tip.clone()), vec![older.clone()]))
.expect("pointers whose ids encode their start seqs decode");
let drifted_tip = wal_pointer_json("wal_00000000000000000004-aaaaaaaaaaaaaaaa", 3, 3);
let error = serde_json::from_value::<HeadState>(head_json(Some(drifted_tip), vec![older]))
.expect_err("the head rejects a tip that disagrees with its start seq");
let message = error.to_string();
assert!(
message.contains("`wal_00000000000000000004-aaaaaaaaaaaaaaaa`")
&& message.contains("start seq `3`"),
"the rejection should name both values: {message}"
);
let drifted_hint = wal_pointer_json("wal_00000000000000000001-fedcba9876543210", 2, 2);
serde_json::from_value::<HeadState>(head_json(Some(tip), vec![drifted_hint]))
.expect_err("the head rejects a hint that disagrees with its start seq");
}
#[test]
fn genesis_head_decodes_without_a_tip_or_hints() {
let genesis = serde_json::from_value::<HeadState>(head_json(None, Vec::new()))
.expect("a head with no visible tip decodes");
assert_eq!(genesis.visible_wal_tip, None);
assert!(genesis.recent_segments.is_empty());
}
#[test]
fn a_head_that_omits_its_predecessor_hints_does_not_decode() {
let mut head = head_json(None, Vec::new());
assert_eq!(head["recent_segments"], serde_json::json!([]));
head.as_object_mut()
.expect("the head is a JSON object")
.remove("recent_segments");
let error = serde_json::from_value::<HeadState>(head)
.expect_err("a head without `recent_segments` is corruption");
assert!(
error.to_string().contains("recent_segments"),
"the rejection should name the field: {error}"
);
}
#[test]
fn control_object_codec_round_trips_and_validates() {
let envelope = HeadStateEnvelope::from_state(ControlObjectKind::WalHead, sample_head())
.expect("envelope");
let encoded = encode_control_object(&envelope).expect("encode");
let decoded: HeadStateEnvelope =
decode_control_object(&encoded, ControlObjectKind::WalHead).expect("decode");
assert_eq!(decoded, envelope);
let mismatch =
decode_control_object::<MetadataRootState>(&encoded, ControlObjectKind::MetadataRoot)
.expect_err("kind mismatch");
assert!(matches!(mismatch, EnvelopeCodecError::KindMismatch { .. }));
}
#[test]
fn successor_head_must_carry_the_namespace_identity_forward() {
let head = sample_head();
let mut successor = head.clone();
successor.seq = ChangeSeq(4);
head.ensure_successor_identity(&successor)
.expect("advancing the sequence keeps the identity");
let mut drifted = head.clone();
drifted.content_store_id = ContentStoreId::parse("cs_fedcba9876543210fedcba9876543210")
.expect("valid content store id");
assert_eq!(
head.ensure_successor_identity(&drifted)
.expect_err("content store drift is rejected")
.field,
"content_store_id"
);
let mut forked = head.clone();
forked.fork_basis = Some(ForkBasis {
manifest: ManifestRef {
owner_namespace_id: NamespaceId::parse("source").expect("valid namespace id"),
manifest_no: ManifestNo(7),
manifest_object_id: ManifestObjectId::parse(
"man_00000000000000000007-0123456789abcdef",
)
.expect("valid manifest object id"),
manifest_head_seq: ChangeSeq(7),
manifest_payload_checksum: "sha256:test".to_owned(),
},
source_checkpoint_id: CheckpointId::parse("chk_00000000000000000000000000000002")
.expect("valid checkpoint id"),
});
assert_eq!(
head.ensure_successor_identity(&forked)
.expect_err("gaining a fork basis is rejected")
.field,
"fork_basis"
);
}
}