pub mod postgres;
mod rows;
pub mod schema;
use async_trait::async_trait;
use super::{BackendFailure, BackendKind, Capabilities, FailureCategory};
use crate::desired_state::{
AuditEvent, ExpectedRevision, IdempotencyKey, IntegrityError, LoadedRevision, ResourceRef,
RevisionCandidate, RevisionId, RevisionManifest, ValidationError,
};
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum ControlPlaneBackend {
#[default]
Postgres,
}
impl<'de> serde::Deserialize<'de> for ControlPlaneBackend {
fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
let name = <std::borrow::Cow<'de, str>>::deserialize(deserializer)?;
Self::parse(&name).map_err(serde::de::Error::custom)
}
}
impl ControlPlaneBackend {
pub const fn kind(self) -> BackendKind {
match self {
Self::Postgres => BackendKind::Postgres,
}
}
pub fn parse(name: &str) -> Result<Self, UnsupportedControlPlaneBackend> {
match name {
"postgres" => Ok(Self::Postgres),
"postgresql" | "pg" => Err(UnsupportedControlPlaneBackend::Unknown {
name: name.to_owned(),
}),
"redis" => Err(UnsupportedControlPlaneBackend::HotStateOnly {
name: "redis".to_owned(),
}),
"memory" | "in-memory" => Err(UnsupportedControlPlaneBackend::NotDurable {
name: name.to_owned(),
}),
other => Err(UnsupportedControlPlaneBackend::Unknown {
name: other.to_owned(),
}),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum UnsupportedControlPlaneBackend {
#[error(
"`{name}` holds loss-tolerant hot state and cannot own durable control-plane state; \
the only durable control-plane backend is `postgres`"
)]
HotStateOnly { name: String },
#[error("`{name}` is not durable and cannot own control-plane state")]
NotDurable { name: String },
#[error("unknown control-plane backend `{name}`; the only durable backend is `postgres`")]
Unknown { name: String },
}
#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
pub enum ControlPlaneError {
#[error("control-plane store `{backend}` unavailable: {message}")]
Unavailable {
backend: &'static str,
message: String,
},
#[error("expected {expected} to be current, but the newest is {actual:?}")]
Conflict {
expected: ExpectedRevision,
actual: Option<RevisionId>,
},
#[error("revision {0} is not retained")]
RevisionNotFound(RevisionId),
#[error("invalid candidate revision: {0}")]
Invalid(#[from] ValidationError),
#[error("{reference} is already published with different content; publish a new version")]
ImmutableResourceVersion { reference: ResourceRef },
#[error(
"idempotency key `{key}` already published revision {published} with different desired state"
)]
IdempotencyKeyReused {
key: IdempotencyKey,
published: RevisionId,
},
#[error("control-plane store `{backend}` refused the operation: {message}")]
Denied {
backend: &'static str,
message: String,
},
#[error("stored revision {revision} is unreadable: {source}")]
Corrupt {
revision: RevisionId,
source: Box<IntegrityError>,
},
#[error("control-plane storage is unreadable: {detail}")]
CorruptStorage { detail: String },
}
impl ControlPlaneError {
pub fn corrupt(revision: RevisionId, source: IntegrityError) -> Self {
Self::Corrupt {
revision,
source: Box::new(source),
}
}
}
impl BackendFailure for ControlPlaneError {
fn category(&self) -> FailureCategory {
match self {
Self::Unavailable { .. } => FailureCategory::Unavailable,
Self::Conflict { .. } => FailureCategory::Conflict,
Self::RevisionNotFound(_) => FailureCategory::NotFound,
Self::Invalid(_)
| Self::ImmutableResourceVersion { .. }
| Self::IdempotencyKeyReused { .. } => FailureCategory::Invalid,
Self::Denied { .. } => FailureCategory::Denied,
Self::Corrupt { .. } | Self::CorruptStorage { .. } => FailureCategory::Corrupt,
}
}
}
#[async_trait]
pub trait ControlPlaneStore: Send + Sync {
fn name(&self) -> &'static str;
fn capabilities(&self) -> Capabilities;
async fn health(&self) -> Result<(), ControlPlaneError>;
async fn desired_revision(&self) -> Result<Option<RevisionId>, ControlPlaneError>;
async fn load_manifest(&self, id: RevisionId) -> Result<RevisionManifest, ControlPlaneError>;
async fn load_revision(&self, id: RevisionId) -> Result<LoadedRevision, ControlPlaneError>;
async fn publish_revision(
&self,
candidate: RevisionCandidate,
) -> Result<RevisionManifest, ControlPlaneError>;
async fn audit_trail(&self, id: RevisionId) -> Result<Vec<AuditEvent>, ControlPlaneError>;
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn redis_cannot_be_selected_as_a_control_plane_backend() {
let error = ControlPlaneBackend::parse("redis").expect_err("redis must be refused");
assert!(matches!(
error,
UnsupportedControlPlaneBackend::HotStateOnly { .. }
));
assert!(error.to_string().contains("hot state"));
assert!(matches!(
ControlPlaneBackend::parse("in-memory"),
Err(UnsupportedControlPlaneBackend::NotDurable { .. })
));
assert!(matches!(
ControlPlaneBackend::parse("sqlite"),
Err(UnsupportedControlPlaneBackend::Unknown { .. })
));
}
#[test]
fn every_selectable_control_plane_backend_is_durable() {
let backend = ControlPlaneBackend::parse("postgres").expect("durable backend");
assert!(backend.kind().durable_control_plane());
assert_eq!(
ControlPlaneBackend::default(),
ControlPlaneBackend::Postgres
);
}
#[test]
fn deserialization_resolves_through_parse() {
assert_eq!(
serde_json::from_str::<ControlPlaneBackend>("\"postgres\"").unwrap(),
ControlPlaneBackend::Postgres
);
for name in ["redis", "in-memory", "postgresql", "sqlite"] {
let refusal = serde_json::from_str::<ControlPlaneBackend>(&format!("\"{name}\""))
.expect_err("only postgres is a durable control plane")
.to_string();
let expected = ControlPlaneBackend::parse(name).unwrap_err().to_string();
assert!(
refusal.contains(&expected),
"`{name}` was refused as `{refusal}` instead of `{expected}`"
);
}
}
#[test]
fn a_callers_mistake_and_unreadable_storage_are_different_categories() {
let invalid = ControlPlaneError::Invalid(ValidationError::Empty);
assert_eq!(invalid.category(), FailureCategory::Invalid);
assert!(!invalid.retryable());
let corrupt = ControlPlaneError::corrupt(
crate::desired_state::RevisionId::new(
crate::desired_state::Uuid7::from_parts(1, 0, 1).unwrap(),
),
IntegrityError::Invalid(ValidationError::Empty),
);
assert_eq!(corrupt.category(), FailureCategory::Corrupt);
assert!(!corrupt.retryable());
assert!(corrupt.to_string().contains("unreadable"));
let outage = ControlPlaneError::Unavailable {
backend: "postgres",
message: "connection refused".to_owned(),
};
assert_eq!(outage.category(), FailureCategory::Unavailable);
assert!(outage.retryable());
}
}