use crate::{
ExtractionExpectedAuthority, ExtractionPlan, ExtractionPlanInputs, ExtractionPlanPhaseKind,
ExtractionScaffold, ExtractionScaffoldApplyResult, ExtractionWorkloadRole,
ensure_extraction_plan_fresh, extraction_input_digest, extraction_plan_integrity_is_valid,
extraction_scaffold_integrity_is_valid, validate_extraction_scaffold,
};
use async_trait::async_trait;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
pub const EXTRACTION_RUN_PROTOCOL: &str = "lenso.extraction-run.v1";
pub const EXTRACTION_OPERATION_RECEIPT_PROTOCOL: &str = "lenso.extraction-operation-receipt.v1";
pub const DESTINATION_EXPANSION_PHASE_ID: &str = "03-destination-expansion";
const EXTRACTION_RUN_SCHEMA_ID: &str =
"https://contracts.lenso.local/extraction/lenso.extraction-run.v1.schema.json";
#[derive(
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionRunMode {
Apply,
DryRun,
}
#[derive(
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionRunStatus {
Planned,
InProgress,
Blocked,
Succeeded,
}
#[derive(
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionExpansionOperationKind {
CreateIsolatedStore,
ApplyExpandMigration,
VerifyMigrationWorkload,
VerifyCandidateHealth,
}
impl ExtractionExpansionOperationKind {
fn is_mutating(self) -> bool {
matches!(self, Self::CreateIsolatedStore | Self::ApplyExpandMigration)
}
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionMigrationArtifact {
pub migration_id: String,
pub source_reference: String,
pub source_digest: String,
pub sql: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionExpandMigration {
pub migration_id: String,
pub source_reference: String,
pub source_digest: String,
pub sql: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionExpansionOperation {
pub operation_id: String,
pub operation_digest: String,
pub order: u16,
pub kind: ExtractionExpansionOperationKind,
pub workload_id: String,
pub candidate_service_id: String,
pub destination_store_id: String,
pub mutating: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub migration: Option<ExtractionExpandMigration>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionRunExpectedState {
pub plan_id: String,
pub plan_digest: String,
pub scaffold_id: String,
pub scaffold_digest: String,
pub phase_id: String,
pub source_authority: ExtractionExpectedAuthority,
pub linked_authority_remains_authoritative: bool,
pub source_store_remains_unchanged: bool,
pub candidate_service_id: String,
pub destination_store_id: String,
pub destination_store_engine: String,
pub destination_store_isolated: bool,
pub migration_workload_id: String,
pub api_workload_id: String,
pub ordered_operations_digest: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionRunPhase {
pub phase_id: String,
pub kind: ExtractionPlanPhaseKind,
pub status: ExtractionRunStatus,
#[serde(default)]
pub completed_operation_ids: Vec<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub next_operation_id: Option<String>,
}
#[derive(
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionRunEvidenceKind {
StoreIsolation,
MigrationApplied,
MigrationWorkloadHealth,
CandidateHealth,
SourceAuthority,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionRunEvidence {
pub kind: ExtractionRunEvidenceKind,
pub subject: String,
pub digest: String,
pub detail: String,
}
#[derive(
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionOperationOutcome {
Created,
AlreadyExists,
Applied,
AlreadyApplied,
Healthy,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionOperationReceipt {
pub protocol: String,
pub receipt_id: String,
pub receipt_digest: String,
pub run_id: String,
pub plan_id: String,
pub plan_digest: String,
pub expected_state_digest: String,
pub operation_id: String,
pub operation_digest: String,
pub operation_kind: ExtractionExpansionOperationKind,
pub workload_id: String,
pub candidate_service_id: String,
pub destination_store_id: String,
pub outcome: ExtractionOperationOutcome,
pub source_authority: ExtractionExpectedAuthority,
pub source_store_unchanged: bool,
pub linked_authority_remains_authoritative: bool,
pub candidate_authoritative: bool,
#[serde(default)]
pub evidence: Vec<ExtractionRunEvidence>,
}
#[derive(
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionRunErrorCode {
PlanStale,
AuthorityChanged,
SourceMutationReported,
ReceiptInvalid,
StoreProvisioningFailed,
MigrationFailed,
MigrationWorkloadUnhealthy,
CandidateUnhealthy,
WorkloadUnavailable,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionRunError {
pub sequence: u32,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub operation_id: Option<String>,
pub code: ExtractionRunErrorCode,
pub message: String,
#[serde(default)]
pub evidence: Vec<ExtractionRunEvidence>,
#[serde(default)]
pub next_actions: Vec<String>,
pub resolved: bool,
}
#[allow(clippy::struct_excessive_bools)]
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionRunEffects {
pub creates_destination_store: bool,
pub applies_destination_schema: bool,
pub invokes_candidate_workload_behavior: bool,
pub copies_service_data: bool,
pub mutates_source_store: bool,
pub mutates_linked_implementation: bool,
pub changes_authority: bool,
pub performs_destructive_cleanup: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionRun {
pub protocol: String,
pub run_id: String,
pub run_digest: String,
pub revision: u64,
pub mode: ExtractionRunMode,
pub plan: ExtractionPlan,
pub current_phase: ExtractionRunPhase,
pub expected_state: ExtractionRunExpectedState,
pub expected_state_digest: String,
pub ordered_operations: Vec<ExtractionExpansionOperation>,
#[serde(default)]
pub receipts: Vec<ExtractionOperationReceipt>,
#[serde(default)]
pub evidence: Vec<ExtractionRunEvidence>,
#[serde(default)]
pub errors: Vec<ExtractionRunError>,
#[serde(default)]
pub next_actions: Vec<String>,
pub effects: ExtractionRunEffects,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ExtractionRunInputs {
pub plan: ExtractionPlan,
pub current_plan_inputs: ExtractionPlanInputs,
pub scaffold: ExtractionScaffold,
pub scaffold_apply_result: ExtractionScaffoldApplyResult,
pub migrations: Vec<ExtractionMigrationArtifact>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionRunStartErrorCode {
PlanInvalid,
PlanStale,
PhaseInvalid,
ScaffoldInvalid,
ScaffoldNotApplied,
UnsupportedStoreEngine,
StoreNotIsolated,
WorkloadMissing,
MigrationArtifactMissing,
MigrationArtifactUnexpected,
MigrationArtifactChanged,
MigrationNotExpandFirst,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionRunStartError {
pub code: ExtractionRunStartErrorCode,
pub message: String,
pub next_actions: Vec<String>,
pub effects: ExtractionRunEffects,
}
impl fmt::Display for ExtractionRunStartError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.message)
}
}
impl std::error::Error for ExtractionRunStartError {}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionRunAdvanceErrorCode {
RunInvalid,
DryRunCannotAdvance,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionRunAdvanceError {
pub code: ExtractionRunAdvanceErrorCode,
pub message: String,
pub next_actions: Vec<String>,
}
impl fmt::Display for ExtractionRunAdvanceError {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.message)
}
}
impl std::error::Error for ExtractionRunAdvanceError {}
#[derive(
Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, JsonSchema,
)]
#[serde(rename_all = "snake_case")]
pub enum ExtractionWorkloadFailureCode {
Unavailable,
StoreProvisioningFailed,
MigrationFailed,
MigrationWorkloadUnhealthy,
CandidateUnhealthy,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "camelCase")]
pub struct ExtractionWorkloadFailure {
pub code: ExtractionWorkloadFailureCode,
pub message: String,
#[serde(default)]
pub evidence: Vec<ExtractionRunEvidence>,
#[serde(default)]
pub next_actions: Vec<String>,
}
impl fmt::Display for ExtractionWorkloadFailure {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(&self.message)
}
}
impl std::error::Error for ExtractionWorkloadFailure {}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExtractionWorkloadRequest {
pub run_id: String,
pub plan_id: String,
pub plan_digest: String,
pub expected_state: ExtractionRunExpectedState,
pub expected_state_digest: String,
pub operation: ExtractionExpansionOperation,
}
#[async_trait]
pub trait ExtractionExpansionWorkload: fmt::Debug + Send + Sync {
async fn inspect_receipt(
&self,
request: &ExtractionWorkloadRequest,
) -> Result<Option<ExtractionOperationReceipt>, ExtractionWorkloadFailure>;
async fn execute(
&self,
request: &ExtractionWorkloadRequest,
) -> Result<ExtractionOperationReceipt, ExtractionWorkloadFailure>;
}
pub fn start_destination_expansion(
inputs: &ExtractionRunInputs,
) -> Result<ExtractionRun, ExtractionRunStartError> {
build_destination_expansion(inputs, ExtractionRunMode::Apply)
}
pub fn dry_run_destination_expansion(
inputs: &ExtractionRunInputs,
) -> Result<ExtractionRun, ExtractionRunStartError> {
build_destination_expansion(inputs, ExtractionRunMode::DryRun)
}
fn build_destination_expansion(
inputs: &ExtractionRunInputs,
mode: ExtractionRunMode,
) -> Result<ExtractionRun, ExtractionRunStartError> {
validate_start_inputs(inputs)?;
let plan = &inputs.plan;
let (operations, migration_workload_id, api_workload_id) =
destination_expansion_operations(inputs)?;
let ordered_operations_digest = digest_serializable(&operations)?;
let expected_state = ExtractionRunExpectedState {
plan_id: plan.plan_id.clone(),
plan_digest: plan.plan_digest.clone(),
scaffold_id: inputs.scaffold.scaffold_id.clone(),
scaffold_digest: inputs.scaffold.scaffold_digest.clone(),
phase_id: DESTINATION_EXPANSION_PHASE_ID.to_owned(),
source_authority: plan.expected_authority.clone(),
linked_authority_remains_authoritative: true,
source_store_remains_unchanged: true,
candidate_service_id: plan.proposed_service.service_id.clone(),
destination_store_id: plan.proposed_service.store.store_id.clone(),
destination_store_engine: plan.proposed_service.store.engine.clone(),
destination_store_isolated: true,
migration_workload_id,
api_workload_id,
ordered_operations_digest,
};
let expected_state_digest = digest_serializable(&expected_state)?;
let run_identity_digest = digest_serializable(&(
plan.plan_id.as_str(),
DESTINATION_EXPANSION_PHASE_ID,
expected_state_digest.as_str(),
))?;
let mut run = ExtractionRun {
protocol: EXTRACTION_RUN_PROTOCOL.to_owned(),
run_id: format!("extraction-run:{run_identity_digest}"),
run_digest: String::new(),
revision: 1,
mode,
plan: plan.clone(),
current_phase: ExtractionRunPhase {
phase_id: DESTINATION_EXPANSION_PHASE_ID.to_owned(),
kind: ExtractionPlanPhaseKind::DestinationExpansion,
status: ExtractionRunStatus::Planned,
completed_operation_ids: Vec::new(),
next_operation_id: operations.first().map(|operation| operation.operation_id.clone()),
},
expected_state,
expected_state_digest,
ordered_operations: operations,
receipts: Vec::new(),
evidence: Vec::new(),
errors: Vec::new(),
next_actions: match mode {
ExtractionRunMode::Apply => vec![
"Persist this Run, then advance exactly one destination operation through the public Workload behavior."
.to_owned(),
],
ExtractionRunMode::DryRun => vec![
"Review these exact ordered operations before starting an apply Run."
.to_owned(),
],
},
effects: ExtractionRunEffects::default(),
};
refresh_run_digest(&mut run);
Ok(run)
}
fn destination_expansion_operations(
inputs: &ExtractionRunInputs,
) -> Result<(Vec<ExtractionExpansionOperation>, String, String), ExtractionRunStartError> {
let plan = &inputs.plan;
let migration_workload_id = workload_id(plan, ExtractionWorkloadRole::Migration)?;
let api_workload_id = workload_id(plan, ExtractionWorkloadRole::Api)?;
let mut operations = vec![operation(
1,
ExtractionExpansionOperationKind::CreateIsolatedStore,
&migration_workload_id,
plan,
None,
)?];
for (index, mapping) in plan.data_mapping.migrations.iter().enumerate() {
let artifact = inputs
.migrations
.iter()
.find(|artifact| artifact.migration_id == mapping.source_migration)
.expect("validated migration artifact");
let order = u16::try_from(index + 2).map_err(|_| too_many_operations_error())?;
operations.push(operation(
order,
ExtractionExpansionOperationKind::ApplyExpandMigration,
&migration_workload_id,
plan,
Some(ExtractionExpandMigration {
migration_id: artifact.migration_id.clone(),
source_reference: artifact.source_reference.clone(),
source_digest: artifact.source_digest.clone(),
sql: artifact.sql.clone(),
}),
)?);
}
let next_order =
u16::try_from(operations.len() + 1).map_err(|_| too_many_operations_error())?;
let health_order = next_order
.checked_add(1)
.ok_or_else(too_many_operations_error)?;
operations.push(operation(
next_order,
ExtractionExpansionOperationKind::VerifyMigrationWorkload,
&migration_workload_id,
plan,
None,
)?);
operations.push(operation(
health_order,
ExtractionExpansionOperationKind::VerifyCandidateHealth,
&api_workload_id,
plan,
None,
)?);
Ok((operations, migration_workload_id, api_workload_id))
}
fn too_many_operations_error() -> ExtractionRunStartError {
start_error(
ExtractionRunStartErrorCode::MigrationArtifactUnexpected,
"Too many destination operations were supplied for one Extraction Run.",
"Split the Module migration set into a reviewable Extraction Plan.",
)
}
pub async fn advance_destination_expansion(
mut run: ExtractionRun,
current_inputs: &ExtractionPlanInputs,
workload: &dyn ExtractionExpansionWorkload,
) -> Result<ExtractionRun, ExtractionRunAdvanceError> {
if !extraction_run_integrity_is_valid(&run) {
return Err(advance_error(
ExtractionRunAdvanceErrorCode::RunInvalid,
"Extraction Run integrity validation failed before Workload behavior was invoked.",
"Discard the changed Run and resume from the last integrity-valid revision.",
));
}
if run.mode == ExtractionRunMode::DryRun {
return Err(advance_error(
ExtractionRunAdvanceErrorCode::DryRunCannotAdvance,
"A dry-run Extraction Run cannot execute destination operations.",
"Start an apply Run from the same fresh inputs after review.",
));
}
if run.current_phase.status == ExtractionRunStatus::Succeeded {
return Ok(run);
}
if let Err(rejection) = ensure_extraction_plan_fresh(&run.plan, current_inputs) {
block_run(
&mut run,
None,
ExtractionRunErrorCode::PlanStale,
rejection.message,
Vec::new(),
rejection.next_actions,
);
return Ok(run);
}
let Some(operation) = next_unreceipted_operation(&run).cloned() else {
finish_run(&mut run);
return Ok(run);
};
for error in &mut run.errors {
if error.operation_id.as_deref() == Some(operation.operation_id.as_str()) {
error.resolved = true;
}
}
let request = ExtractionWorkloadRequest {
run_id: run.run_id.clone(),
plan_id: run.plan.plan_id.clone(),
plan_digest: run.plan.plan_digest.clone(),
expected_state: run.expected_state.clone(),
expected_state_digest: run.expected_state_digest.clone(),
operation: operation.clone(),
};
let inspected = match workload.inspect_receipt(&request).await {
Ok(receipt) => receipt,
Err(failure) => {
record_workload_failure(&mut run, &operation, failure);
return Ok(run);
}
};
let receipt = match inspected {
Some(receipt) => receipt,
None => match workload.execute(&request).await {
Ok(receipt) => receipt,
Err(failure) => {
record_workload_failure(&mut run, &operation, failure);
return Ok(run);
}
},
};
record_destination_expansion_receipt(run, receipt)
}
pub fn record_destination_expansion_receipt(
mut run: ExtractionRun,
mut receipt: ExtractionOperationReceipt,
) -> Result<ExtractionRun, ExtractionRunAdvanceError> {
if !extraction_run_integrity_is_valid(&run) {
return Err(advance_error(
ExtractionRunAdvanceErrorCode::RunInvalid,
"Extraction Run integrity validation failed before recording a receipt.",
"Resume from the last integrity-valid Run revision.",
));
}
let Some(operation) = next_unreceipted_operation(&run).cloned() else {
return Ok(run);
};
receipt.evidence.sort();
receipt.evidence.dedup();
if let Err((code, message, next_action)) = validate_receipt(&run, &operation, &receipt) {
block_run(
&mut run,
Some(operation.operation_id),
code,
message,
Vec::new(),
vec![next_action],
);
return Ok(run);
}
run.effects.invokes_candidate_workload_behavior = true;
match (operation.kind, receipt.outcome) {
(
ExtractionExpansionOperationKind::CreateIsolatedStore,
ExtractionOperationOutcome::Created,
) => run.effects.creates_destination_store = true,
(
ExtractionExpansionOperationKind::ApplyExpandMigration,
ExtractionOperationOutcome::Applied,
) => run.effects.applies_destination_schema = true,
_ => {}
}
run.evidence.extend(receipt.evidence.iter().cloned());
run.evidence.sort();
run.evidence.dedup();
run.current_phase
.completed_operation_ids
.push(operation.operation_id.clone());
run.receipts.push(receipt);
run.current_phase.status = ExtractionRunStatus::InProgress;
run.current_phase.next_operation_id =
next_unreceipted_operation(&run).map(|operation| operation.operation_id.clone());
run.next_actions = if let Some(next) = &run.current_phase.next_operation_id {
vec![format!(
"Persist this Run revision, then advance operation `{next}`."
)]
} else {
vec![
"Persist the successful destination expansion evidence before starting backfill."
.to_owned(),
]
};
run.revision += 1;
if run.current_phase.next_operation_id.is_none() {
run.current_phase.status = ExtractionRunStatus::Succeeded;
}
refresh_run_digest(&mut run);
Ok(run)
}
pub fn build_extraction_operation_receipt(
request: &ExtractionWorkloadRequest,
outcome: ExtractionOperationOutcome,
mut evidence: Vec<ExtractionRunEvidence>,
) -> Result<ExtractionOperationReceipt, ExtractionRunStartError> {
evidence.sort();
evidence.dedup();
let mut receipt = ExtractionOperationReceipt {
protocol: EXTRACTION_OPERATION_RECEIPT_PROTOCOL.to_owned(),
receipt_id: String::new(),
receipt_digest: String::new(),
run_id: request.run_id.clone(),
plan_id: request.plan_id.clone(),
plan_digest: request.plan_digest.clone(),
expected_state_digest: request.expected_state_digest.clone(),
operation_id: request.operation.operation_id.clone(),
operation_digest: request.operation.operation_digest.clone(),
operation_kind: request.operation.kind,
workload_id: request.operation.workload_id.clone(),
candidate_service_id: request.expected_state.candidate_service_id.clone(),
destination_store_id: request.expected_state.destination_store_id.clone(),
outcome,
source_authority: request.expected_state.source_authority.clone(),
source_store_unchanged: true,
linked_authority_remains_authoritative: true,
candidate_authoritative: false,
evidence,
};
let digest = receipt_digest(&receipt)?;
receipt.receipt_id = format!("extraction-operation-receipt:{digest}");
receipt.receipt_digest = digest;
Ok(receipt)
}
#[must_use]
pub fn extraction_operation_receipt_integrity_is_valid(
receipt: &ExtractionOperationReceipt,
) -> bool {
receipt.protocol == EXTRACTION_OPERATION_RECEIPT_PROTOCOL
&& receipt.receipt_id == format!("extraction-operation-receipt:{}", receipt.receipt_digest)
&& receipt_digest(receipt).is_ok_and(|digest| digest == receipt.receipt_digest)
}
#[must_use]
pub fn extraction_run_integrity_is_valid(run: &ExtractionRun) -> bool {
if run.protocol != EXTRACTION_RUN_PROTOCOL
|| !extraction_plan_integrity_is_valid(&run.plan)
|| run.current_phase.phase_id != DESTINATION_EXPANSION_PHASE_ID
|| run.current_phase.kind != ExtractionPlanPhaseKind::DestinationExpansion
|| run.expected_state.plan_id != run.plan.plan_id
|| run.expected_state.plan_digest != run.plan.plan_digest
|| run.expected_state.source_authority != run.plan.expected_authority
|| run.expected_state.candidate_service_id != run.plan.proposed_service.service_id
|| run.expected_state.destination_store_id != run.plan.proposed_service.store.store_id
|| run.expected_state.destination_store_engine != run.plan.proposed_service.store.engine
|| !run.expected_state.destination_store_isolated
|| !run.expected_state.linked_authority_remains_authoritative
|| !run.expected_state.source_store_remains_unchanged
|| run.effects.copies_service_data
|| run.effects.mutates_source_store
|| run.effects.mutates_linked_implementation
|| run.effects.changes_authority
|| run.effects.performs_destructive_cleanup
|| !digest_serializable(&run.expected_state)
.is_ok_and(|digest| digest == run.expected_state_digest)
|| !digest_serializable(&run.ordered_operations)
.is_ok_and(|digest| digest == run.expected_state.ordered_operations_digest)
{
return false;
}
let identity = digest_serializable(&(
run.plan.plan_id.as_str(),
DESTINATION_EXPANSION_PHASE_ID,
run.expected_state_digest.as_str(),
));
if !identity.is_ok_and(|digest| run.run_id == format!("extraction-run:{digest}")) {
return false;
}
let operation_ids = run
.ordered_operations
.iter()
.map(|operation| operation.operation_id.as_str())
.collect::<BTreeSet<_>>();
if operation_ids.len() != run.ordered_operations.len()
|| run
.ordered_operations
.iter()
.enumerate()
.any(|(index, operation)| {
operation.order != u16::try_from(index + 1).unwrap_or(u16::MAX)
|| operation.mutating != operation.kind.is_mutating()
|| !operation_digest(operation)
.is_ok_and(|digest| digest == operation.operation_digest)
})
{
return false;
}
if run.receipts.iter().any(|receipt| {
let operation = run
.ordered_operations
.iter()
.find(|operation| operation.operation_id == receipt.operation_id);
operation.is_none_or(|operation| validate_receipt(run, operation, receipt).is_err())
}) {
return false;
}
let completed = run
.receipts
.iter()
.map(|receipt| receipt.operation_id.as_str())
.collect::<Vec<_>>();
let expected_completed = run
.ordered_operations
.iter()
.take(completed.len())
.map(|operation| operation.operation_id.as_str())
.collect::<Vec<_>>();
if completed
!= run
.current_phase
.completed_operation_ids
.iter()
.map(String::as_str)
.collect::<Vec<_>>()
|| completed != expected_completed
{
return false;
}
let expected_next = run
.ordered_operations
.get(completed.len())
.map(|operation| operation.operation_id.as_str());
if run.current_phase.next_operation_id.as_deref() != expected_next
|| (expected_next.is_none() && run.current_phase.status != ExtractionRunStatus::Succeeded)
|| (expected_next.is_some() && run.current_phase.status == ExtractionRunStatus::Succeeded)
{
return false;
}
run_digest(run).is_ok_and(|digest| digest == run.run_digest)
}
pub fn extraction_run_json(run: &ExtractionRun) -> Result<String, serde_json::Error> {
serde_json::to_string_pretty(run).map(|value| format!("{value}\n"))
}
#[must_use]
pub fn extraction_run_schema() -> Value {
let mut schema = serde_json::to_value(schemars::schema_for!(ExtractionRun))
.expect("Extraction Run schema must serialize");
schema["$id"] = Value::String(EXTRACTION_RUN_SCHEMA_ID.to_owned());
schema["title"] = Value::String("Lenso Extraction Run v1".to_owned());
schema["properties"]["protocol"] = json!({
"type": "string",
"const": EXTRACTION_RUN_PROTOCOL
});
schema["properties"]["runId"] = json!({
"type": "string",
"pattern": "^extraction-run:sha256:[0-9a-f]{64}$"
});
schema["properties"]["runDigest"] = json!({
"type": "string",
"pattern": "^sha256:[0-9a-f]{64}$"
});
schema
}
#[must_use]
pub fn render_extraction_run(run: &ExtractionRun) -> String {
let mut output = vec![
format!("Extraction Run: {}", run.run_id),
format!("Plan: {}", run.plan.plan_id),
format!("Mode: {:?}", run.mode).to_lowercase(),
format!(
"Phase: {} ({:?})",
run.current_phase.phase_id, run.current_phase.status
)
.to_lowercase(),
format!(
"Authority: {:?}:{}@{} (unchanged)",
run.expected_state.source_authority.kind,
run.expected_state.source_authority.owner_id,
run.expected_state.source_authority.revision
)
.to_lowercase(),
format!(
"Candidate Store: {} ({}, isolated={})",
run.expected_state.destination_store_id,
run.expected_state.destination_store_engine,
run.expected_state.destination_store_isolated
),
String::new(),
"Ordered operations:".to_owned(),
];
output.extend(run.ordered_operations.iter().map(|operation| {
let completed = run
.current_phase
.completed_operation_ids
.contains(&operation.operation_id);
format!(
"- {:02} [{}] {:?}: {}",
operation.order,
if completed { "done" } else { "pending" },
operation.kind,
operation.operation_id
)
.to_lowercase()
}));
if !run.errors.is_empty() {
output.push(String::new());
output.push("Errors:".to_owned());
output.extend(run.errors.iter().map(|error| {
format!(
"- {:?}: {}{}",
error.code,
error.message,
if error.resolved { " (resolved)" } else { "" }
)
.to_lowercase()
}));
}
output.push(String::new());
output.push("Next actions:".to_owned());
output.extend(run.next_actions.iter().map(|action| format!("- {action}")));
format!("{}\n", output.join("\n"))
}
#[must_use]
pub fn validate_expand_first_postgres_sql(sql: &str) -> bool {
let mut source = String::new();
for line in sql.lines() {
let code = line.split("--").next().unwrap_or_default();
if code.contains("/*") || code.contains("*/") {
return false;
}
source.push_str(code);
source.push('\n');
}
let statements = source
.split(';')
.map(|statement| statement.split_whitespace().collect::<Vec<_>>().join(" "))
.filter(|statement| !statement.is_empty())
.map(|statement| statement.to_ascii_lowercase())
.collect::<Vec<_>>();
!statements.is_empty()
&& statements.iter().all(|statement| {
let forbidden = [
" drop ",
" truncate ",
" delete ",
" update ",
" insert ",
" rename ",
" alter column ",
" set schema ",
];
let padded = format!(" {statement} ");
!forbidden.iter().any(|token| padded.contains(token))
&& (statement.starts_with("create schema ")
|| statement.starts_with("create table ")
|| statement.starts_with("create index ")
|| statement.starts_with("create unique index ")
|| statement.starts_with("create type ")
|| statement.starts_with("create extension ")
|| statement.starts_with("comment on ")
|| (statement.starts_with("alter table ")
&& (statement.contains(" add column ")
|| statement.contains(" add constraint "))))
})
}
fn validate_start_inputs(inputs: &ExtractionRunInputs) -> Result<(), ExtractionRunStartError> {
let plan = &inputs.plan;
if !extraction_plan_integrity_is_valid(plan) {
return Err(start_error(
ExtractionRunStartErrorCode::PlanInvalid,
"Extraction Plan integrity validation failed before destination expansion.",
"Regenerate the exact content-addressed Extraction Plan.",
));
}
ensure_extraction_plan_fresh(plan, &inputs.current_plan_inputs).map_err(|rejection| {
ExtractionRunStartError {
code: ExtractionRunStartErrorCode::PlanStale,
message: rejection.message,
next_actions: rejection.next_actions,
effects: ExtractionRunEffects::default(),
}
})?;
let phase = plan
.phases
.iter()
.find(|phase| phase.phase_id == DESTINATION_EXPANSION_PHASE_ID);
if phase.is_none_or(|phase| phase.kind != ExtractionPlanPhaseKind::DestinationExpansion) {
return Err(start_error(
ExtractionRunStartErrorCode::PhaseInvalid,
"The exact Extraction Plan does not contain the destination expansion phase.",
"Regenerate the plan with the supported ordered phase protocol.",
));
}
if !extraction_scaffold_integrity_is_valid(&inputs.scaffold)
|| !validate_extraction_scaffold(&inputs.scaffold).is_empty()
|| inputs.scaffold.plan_id != plan.plan_id
|| inputs.scaffold.plan_digest != plan.plan_digest
{
return Err(start_error(
ExtractionRunStartErrorCode::ScaffoldInvalid,
"The candidate scaffold does not match the exact Extraction Plan.",
"Regenerate and apply the identity-preserving scaffold from this plan.",
));
}
let mut applied = inputs
.scaffold_apply_result
.created_files
.iter()
.chain(&inputs.scaffold_apply_result.unchanged_files)
.cloned()
.collect::<Vec<_>>();
applied.sort();
applied.dedup();
let mut expected = inputs
.scaffold
.files
.iter()
.map(|file| file.path.clone())
.collect::<Vec<_>>();
expected.sort();
if inputs.scaffold_apply_result.protocol != "lenso.extraction-scaffold-apply.v1"
|| inputs.scaffold_apply_result.scaffold_id != inputs.scaffold.scaffold_id
|| inputs.scaffold_apply_result.plan_id != plan.plan_id
|| !inputs
.scaffold_apply_result
.linked_authority_remains_authoritative
|| inputs.scaffold_apply_result.effects.starts_workloads
|| inputs.scaffold_apply_result.effects.copies_data
|| inputs.scaffold_apply_result.effects.changes_authority
|| inputs.scaffold_apply_result.effects.changes_provider_path
|| applied != expected
{
return Err(start_error(
ExtractionRunStartErrorCode::ScaffoldNotApplied,
"The complete candidate scaffold has not been applied idempotently.",
"Apply every plan-owned scaffold file without changing linked authority.",
));
}
if plan.proposed_service.store.engine != "postgres"
|| plan.data_mapping.store_engine != "postgres"
{
return Err(start_error(
ExtractionRunStartErrorCode::UnsupportedStoreEngine,
"Destination expansion currently supports Postgres Service Stores only.",
"Use Postgres or block the phase until the Store has equivalent safety semantics.",
));
}
if !plan.proposed_service.store.isolated
|| plan.proposed_service.store.store_id != plan.data_mapping.destination_store
{
return Err(start_error(
ExtractionRunStartErrorCode::StoreNotIsolated,
"The candidate destination Store is not isolated and plan-owned.",
"Generate one isolated Store owned only by the candidate Autonomous Service.",
));
}
validate_migration_artifacts(inputs)
}
fn validate_migration_artifacts(
inputs: &ExtractionRunInputs,
) -> Result<(), ExtractionRunStartError> {
let mappings = inputs
.plan
.data_mapping
.migrations
.iter()
.map(|mapping| (mapping.source_migration.as_str(), mapping))
.collect::<BTreeMap<_, _>>();
let artifacts = inputs
.migrations
.iter()
.map(|artifact| (artifact.migration_id.as_str(), artifact))
.collect::<BTreeMap<_, _>>();
if mappings.len() != inputs.plan.data_mapping.migrations.len()
|| artifacts.len() != inputs.migrations.len()
{
return Err(start_error(
ExtractionRunStartErrorCode::MigrationArtifactUnexpected,
"Migration identities must be unique within one destination expansion phase.",
"Rename or deduplicate the migrations, then generate a new Extraction Plan.",
));
}
let missing = mappings
.keys()
.filter(|migration| !artifacts.contains_key(**migration))
.copied()
.collect::<Vec<_>>();
if !missing.is_empty() {
return Err(start_error(
ExtractionRunStartErrorCode::MigrationArtifactMissing,
format!(
"Plan-owned migration artifacts are missing: {}.",
missing.join(", ")
),
"Supply the exact digest-pinned source migrations from the target Module.",
));
}
let unexpected = artifacts
.keys()
.filter(|migration| !mappings.contains_key(**migration))
.copied()
.collect::<Vec<_>>();
if !unexpected.is_empty() {
return Err(start_error(
ExtractionRunStartErrorCode::MigrationArtifactUnexpected,
format!(
"Unplanned migration artifacts were supplied: {}.",
unexpected.join(", ")
),
"Remove every migration not pinned by the exact Extraction Plan.",
));
}
for (migration_id, mapping) in mappings {
let artifact = artifacts[migration_id];
if artifact.source_reference != mapping.source_reference
|| artifact.source_digest != mapping.source_digest
|| extraction_input_digest(artifact.sql.as_bytes()) != artifact.source_digest
{
return Err(start_error(
ExtractionRunStartErrorCode::MigrationArtifactChanged,
format!("Migration `{migration_id}` changed after plan approval."),
"Regenerate readiness evidence and the Extraction Plan from the current migration content.",
));
}
if !validate_expand_first_postgres_sql(&artifact.sql) {
return Err(start_error(
ExtractionRunStartErrorCode::MigrationNotExpandFirst,
format!(
"Migration `{migration_id}` contains a non-expand or data-mutating Postgres statement."
),
"Split the migration so destination expansion contains only additive schema statements.",
));
}
}
Ok(())
}
fn workload_id(
plan: &ExtractionPlan,
role: ExtractionWorkloadRole,
) -> Result<String, ExtractionRunStartError> {
let matches = plan
.proposed_service
.workloads
.iter()
.filter(|workload| workload.role == role)
.collect::<Vec<_>>();
let [workload] = matches.as_slice() else {
return Err(start_error(
ExtractionRunStartErrorCode::WorkloadMissing,
format!("The candidate must declare exactly one {role:?} Workload."),
"Regenerate the candidate Service with API, Worker, and Migration Workloads.",
));
};
Ok(workload.workload_id.clone())
}
fn operation(
order: u16,
kind: ExtractionExpansionOperationKind,
workload_id: &str,
plan: &ExtractionPlan,
migration: Option<ExtractionExpandMigration>,
) -> Result<ExtractionExpansionOperation, ExtractionRunStartError> {
let label = match kind {
ExtractionExpansionOperationKind::CreateIsolatedStore => "create-isolated-store".to_owned(),
ExtractionExpansionOperationKind::ApplyExpandMigration => format!(
"apply-expand-migration-{}",
migration
.as_ref()
.map_or("missing", |migration| migration.migration_id.as_str())
),
ExtractionExpansionOperationKind::VerifyMigrationWorkload => {
"verify-migration-workload".to_owned()
}
ExtractionExpansionOperationKind::VerifyCandidateHealth => {
"verify-candidate-health".to_owned()
}
};
let mut operation = ExtractionExpansionOperation {
operation_id: format!("{DESTINATION_EXPANSION_PHASE_ID}/{order:02}-{label}"),
operation_digest: String::new(),
order,
kind,
workload_id: workload_id.to_owned(),
candidate_service_id: plan.proposed_service.service_id.clone(),
destination_store_id: plan.proposed_service.store.store_id.clone(),
mutating: kind.is_mutating(),
migration,
};
operation.operation_digest = operation_digest(&operation)?;
Ok(operation)
}
fn validate_receipt(
run: &ExtractionRun,
operation: &ExtractionExpansionOperation,
receipt: &ExtractionOperationReceipt,
) -> Result<(), (ExtractionRunErrorCode, String, String)> {
if !extraction_operation_receipt_integrity_is_valid(receipt)
|| receipt.run_id != run.run_id
|| receipt.plan_id != run.plan.plan_id
|| receipt.plan_digest != run.plan.plan_digest
|| receipt.expected_state_digest != run.expected_state_digest
|| receipt.operation_id != operation.operation_id
|| receipt.operation_digest != operation.operation_digest
|| receipt.operation_kind != operation.kind
|| receipt.workload_id != operation.workload_id
|| receipt.candidate_service_id != run.expected_state.candidate_service_id
|| receipt.destination_store_id != run.expected_state.destination_store_id
|| receipt.evidence.is_empty()
|| !outcome_matches(operation.kind, receipt.outcome)
{
return Err((
ExtractionRunErrorCode::ReceiptInvalid,
format!(
"Operation `{}` returned a receipt that is not bound to the exact plan and expected state.",
operation.operation_id
),
"Inspect the candidate Workload receipt store and retry only with the exact operation identity."
.to_owned(),
));
}
if receipt.source_authority != run.expected_state.source_authority {
return Err((
ExtractionRunErrorCode::AuthorityChanged,
"Linked source authority changed during destination expansion.".to_owned(),
"Stop preparation, refresh authority evidence, and generate a new Extraction Plan."
.to_owned(),
));
}
if !receipt.source_store_unchanged
|| !receipt.linked_authority_remains_authoritative
|| receipt.candidate_authoritative
{
return Err((
ExtractionRunErrorCode::SourceMutationReported,
"A Workload receipt did not preserve the source Store and linked authority invariants."
.to_owned(),
"Stop extraction and inspect the source before any further candidate operation."
.to_owned(),
));
}
Ok(())
}
fn outcome_matches(
kind: ExtractionExpansionOperationKind,
outcome: ExtractionOperationOutcome,
) -> bool {
matches!(
(kind, outcome),
(
ExtractionExpansionOperationKind::CreateIsolatedStore,
ExtractionOperationOutcome::Created | ExtractionOperationOutcome::AlreadyExists
) | (
ExtractionExpansionOperationKind::ApplyExpandMigration,
ExtractionOperationOutcome::Applied | ExtractionOperationOutcome::AlreadyApplied
) | (
ExtractionExpansionOperationKind::VerifyMigrationWorkload
| ExtractionExpansionOperationKind::VerifyCandidateHealth,
ExtractionOperationOutcome::Healthy
)
)
}
fn record_workload_failure(
run: &mut ExtractionRun,
operation: &ExtractionExpansionOperation,
failure: ExtractionWorkloadFailure,
) {
let code = match failure.code {
ExtractionWorkloadFailureCode::Unavailable => ExtractionRunErrorCode::WorkloadUnavailable,
ExtractionWorkloadFailureCode::StoreProvisioningFailed => {
ExtractionRunErrorCode::StoreProvisioningFailed
}
ExtractionWorkloadFailureCode::MigrationFailed => ExtractionRunErrorCode::MigrationFailed,
ExtractionWorkloadFailureCode::MigrationWorkloadUnhealthy => {
ExtractionRunErrorCode::MigrationWorkloadUnhealthy
}
ExtractionWorkloadFailureCode::CandidateUnhealthy => {
ExtractionRunErrorCode::CandidateUnhealthy
}
};
block_run(
run,
Some(operation.operation_id.clone()),
code,
failure.message,
failure.evidence,
failure.next_actions,
);
}
fn block_run(
run: &mut ExtractionRun,
operation_id: Option<String>,
code: ExtractionRunErrorCode,
message: impl Into<String>,
evidence: Vec<ExtractionRunEvidence>,
next_actions: Vec<String>,
) {
let sequence = u32::try_from(run.errors.len() + 1).unwrap_or(u32::MAX);
run.errors.push(ExtractionRunError {
sequence,
operation_id,
code,
message: message.into(),
evidence,
next_actions: next_actions.clone(),
resolved: false,
});
run.current_phase.status = ExtractionRunStatus::Blocked;
run.next_actions = next_actions;
run.revision += 1;
refresh_run_digest(run);
}
fn finish_run(run: &mut ExtractionRun) {
run.current_phase.status = ExtractionRunStatus::Succeeded;
run.current_phase.next_operation_id = None;
run.next_actions = vec![
"Persist the successful destination expansion evidence before starting backfill."
.to_owned(),
];
run.revision += 1;
refresh_run_digest(run);
}
fn next_unreceipted_operation(run: &ExtractionRun) -> Option<&ExtractionExpansionOperation> {
let completed = run
.receipts
.iter()
.map(|receipt| receipt.operation_id.as_str())
.collect::<BTreeSet<_>>();
run.ordered_operations
.iter()
.find(|operation| !completed.contains(operation.operation_id.as_str()))
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct OperationContent<'a> {
operation_id: &'a str,
order: u16,
kind: ExtractionExpansionOperationKind,
workload_id: &'a str,
candidate_service_id: &'a str,
destination_store_id: &'a str,
mutating: bool,
migration: &'a Option<ExtractionExpandMigration>,
}
fn operation_digest(
operation: &ExtractionExpansionOperation,
) -> Result<String, ExtractionRunStartError> {
digest_serializable(&OperationContent {
operation_id: &operation.operation_id,
order: operation.order,
kind: operation.kind,
workload_id: &operation.workload_id,
candidate_service_id: &operation.candidate_service_id,
destination_store_id: &operation.destination_store_id,
mutating: operation.mutating,
migration: &operation.migration,
})
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct ReceiptContent<'a> {
protocol: &'a str,
run_id: &'a str,
plan_id: &'a str,
plan_digest: &'a str,
expected_state_digest: &'a str,
operation_id: &'a str,
operation_digest: &'a str,
operation_kind: ExtractionExpansionOperationKind,
workload_id: &'a str,
candidate_service_id: &'a str,
destination_store_id: &'a str,
outcome: ExtractionOperationOutcome,
source_authority: &'a ExtractionExpectedAuthority,
source_store_unchanged: bool,
linked_authority_remains_authoritative: bool,
candidate_authoritative: bool,
evidence: &'a [ExtractionRunEvidence],
}
fn receipt_digest(receipt: &ExtractionOperationReceipt) -> Result<String, ExtractionRunStartError> {
digest_serializable(&ReceiptContent {
protocol: &receipt.protocol,
run_id: &receipt.run_id,
plan_id: &receipt.plan_id,
plan_digest: &receipt.plan_digest,
expected_state_digest: &receipt.expected_state_digest,
operation_id: &receipt.operation_id,
operation_digest: &receipt.operation_digest,
operation_kind: receipt.operation_kind,
workload_id: &receipt.workload_id,
candidate_service_id: &receipt.candidate_service_id,
destination_store_id: &receipt.destination_store_id,
outcome: receipt.outcome,
source_authority: &receipt.source_authority,
source_store_unchanged: receipt.source_store_unchanged,
linked_authority_remains_authoritative: receipt.linked_authority_remains_authoritative,
candidate_authoritative: receipt.candidate_authoritative,
evidence: &receipt.evidence,
})
}
#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
struct RunContent<'a> {
protocol: &'a str,
run_id: &'a str,
revision: u64,
mode: ExtractionRunMode,
plan: &'a ExtractionPlan,
current_phase: &'a ExtractionRunPhase,
expected_state: &'a ExtractionRunExpectedState,
expected_state_digest: &'a str,
ordered_operations: &'a [ExtractionExpansionOperation],
receipts: &'a [ExtractionOperationReceipt],
evidence: &'a [ExtractionRunEvidence],
errors: &'a [ExtractionRunError],
next_actions: &'a [String],
effects: ExtractionRunEffects,
}
fn run_digest(run: &ExtractionRun) -> Result<String, ExtractionRunStartError> {
digest_serializable(&RunContent {
protocol: &run.protocol,
run_id: &run.run_id,
revision: run.revision,
mode: run.mode,
plan: &run.plan,
current_phase: &run.current_phase,
expected_state: &run.expected_state,
expected_state_digest: &run.expected_state_digest,
ordered_operations: &run.ordered_operations,
receipts: &run.receipts,
evidence: &run.evidence,
errors: &run.errors,
next_actions: &run.next_actions,
effects: run.effects,
})
}
fn refresh_run_digest(run: &mut ExtractionRun) {
run.run_digest = run_digest(run).expect("Extraction Run content must serialize");
}
fn digest_serializable(value: &impl Serialize) -> Result<String, ExtractionRunStartError> {
serde_json::to_vec(value)
.map(extraction_input_digest)
.map_err(|error| {
start_error(
ExtractionRunStartErrorCode::PlanInvalid,
format!("Extraction Run content could not be serialized: {error}"),
"Correct the public artifact input and retry without mutation.",
)
})
}
fn start_error(
code: ExtractionRunStartErrorCode,
message: impl Into<String>,
next_action: impl Into<String>,
) -> ExtractionRunStartError {
ExtractionRunStartError {
code,
message: message.into(),
next_actions: vec![next_action.into()],
effects: ExtractionRunEffects::default(),
}
}
fn advance_error(
code: ExtractionRunAdvanceErrorCode,
message: impl Into<String>,
next_action: impl Into<String>,
) -> ExtractionRunAdvanceError {
ExtractionRunAdvanceError {
code,
message: message.into(),
next_actions: vec![next_action.into()],
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::{
CommonContextRequirement, DIRECT_HTTP_OPENAPI_V1_FIXTURE_YAML,
EXTRACTION_READINESS_ANALYZER_VERSION, EXTRACTION_READINESS_REPORT_PROTOCOL,
ExtractionAuthorityKind, ExtractionContractArtifactFormat, ExtractionContractDirection,
ExtractionContractKind, ExtractionEvidenceDigest, ExtractionPlanContractVersion,
ExtractionReadinessEffects, ExtractionReadinessReport, ExtractionReadinessSurfaceSummary,
ExtractionScaffoldArtifact, ExtractionScaffoldEffects, ExtractionScaffoldInputs,
ExtractionServiceDataEvidence, ServiceTenancyMode, generate_extraction_plan,
generate_extraction_scaffold,
};
use lenso_contracts::{
ModuleHttpMethod, ModuleHttpRoute, ModuleManifest, ServiceOperationMetadata,
};
use std::sync::Mutex;
#[derive(Debug, Default)]
struct FakeWorkload {
receipts: Mutex<BTreeMap<String, ExtractionOperationReceipt>>,
executions: Mutex<Vec<String>>,
}
impl FakeWorkload {
fn execution_count(&self) -> usize {
self.executions.lock().unwrap().len()
}
}
#[async_trait]
impl ExtractionExpansionWorkload for FakeWorkload {
async fn inspect_receipt(
&self,
request: &ExtractionWorkloadRequest,
) -> Result<Option<ExtractionOperationReceipt>, ExtractionWorkloadFailure> {
Ok(self
.receipts
.lock()
.unwrap()
.get(&request.operation.operation_id)
.cloned())
}
async fn execute(
&self,
request: &ExtractionWorkloadRequest,
) -> Result<ExtractionOperationReceipt, ExtractionWorkloadFailure> {
self.executions
.lock()
.unwrap()
.push(request.operation.operation_id.clone());
let (outcome, kind) = match request.operation.kind {
ExtractionExpansionOperationKind::CreateIsolatedStore => (
ExtractionOperationOutcome::Created,
ExtractionRunEvidenceKind::StoreIsolation,
),
ExtractionExpansionOperationKind::ApplyExpandMigration => (
ExtractionOperationOutcome::Applied,
ExtractionRunEvidenceKind::MigrationApplied,
),
ExtractionExpansionOperationKind::VerifyMigrationWorkload => (
ExtractionOperationOutcome::Healthy,
ExtractionRunEvidenceKind::MigrationWorkloadHealth,
),
ExtractionExpansionOperationKind::VerifyCandidateHealth => (
ExtractionOperationOutcome::Healthy,
ExtractionRunEvidenceKind::CandidateHealth,
),
};
let evidence = vec![ExtractionRunEvidence {
kind,
subject: request.operation.operation_id.clone(),
digest: extraction_input_digest(request.operation.operation_digest.as_bytes()),
detail: "verified through candidate Workload behavior".to_owned(),
}];
let receipt = build_extraction_operation_receipt(request, outcome, evidence)
.expect("fake receipt must build");
self.receipts
.lock()
.unwrap()
.insert(request.operation.operation_id.clone(), receipt.clone());
Ok(receipt)
}
}
fn module() -> ModuleManifest {
ModuleManifest::builder("acme/support-ticket")
.capabilities(vec!["support.tickets.read".to_owned()])
.http_routes(vec![ModuleHttpRoute {
method: ModuleHttpMethod::Get,
path: "/v1/tickets/{ticket_id}".to_owned(),
capability: Some("support.tickets.read".to_owned()),
display_name: Some("Get ticket".to_owned()),
story_title: Some("Support ticket opened".to_owned()),
operation: Some(ServiceOperationMetadata {
operation_id: Some("getTicket".to_owned()),
summary: Some("Get ticket".to_owned()),
..ServiceOperationMetadata::default()
}),
}])
.build()
}
#[allow(clippy::too_many_lines)]
fn run_inputs(sql: &str) -> ExtractionRunInputs {
let module = module();
let migration_reference = "modules/support-ticket/migrations/0001_tickets.sql";
let migration_digest = extraction_input_digest(sql.as_bytes());
let report = ExtractionReadinessReport {
protocol: EXTRACTION_READINESS_REPORT_PROTOCOL.to_owned(),
analyzer_version: EXTRACTION_READINESS_ANALYZER_VERSION.to_owned(),
target_module: module.module_id.clone(),
system_id: Some("support-system".to_owned()),
target_owner: Some("support-host".to_owned()),
classification: crate::CompatibilityCategory::Safe,
ready: true,
issue_codes: Vec::new(),
contract_evidence: Vec::new(),
active_consumers: Vec::new(),
surfaces: ExtractionReadinessSurfaceSummary::default(),
service_data: ExtractionServiceDataEvidence {
complete: true,
migrations: vec![crate::ExtractionMigrationEvidence {
migration: "0001_create_support_tickets".to_owned(),
owner_module: Some("support-ticket".to_owned()),
source: crate::ExtractionDataEvidenceSource::StaticDeclaration,
evidence_references: vec![migration_reference.to_owned()],
}],
..ExtractionServiceDataEvidence::default()
},
findings: Vec::new(),
effects: ExtractionReadinessEffects::default(),
};
let current_plan_inputs = ExtractionPlanInputs {
readiness_report: report,
module: module.clone(),
system: json!({
"protocol": "lenso.system.v2",
"systemId": "support-system",
"host": { "hostId": "support-host", "modules": ["acme/support-ticket"] },
"providers": [{
"providerId": "notification-provider",
"modules": ["notification-gateway"]
}],
"autonomousServices": [{
"serviceId": "support-sla-service",
"modules": ["support-sla"],
"workloads": [{ "workloadId": "support-sla-api", "role": "api" }]
}],
"contracts": [{
"contractId": "support.sla-updated.v1",
"version": "v1",
"producerKind": "autonomous_service",
"producerId": "support-sla-service",
"artifact": {
"format": "json_schema",
"path": "contracts/events/support.sla-updated.v1.schema.json"
},
"tenancyMode": "required"
}],
"consumers": [{
"consumerId": "support-ticket-sla-updates",
"ownerKind": "host",
"ownerId": "support-host",
"contractId": "support.sla-updated.v1",
"tenancyMode": "required"
}]
}),
contract_versions: vec![ExtractionPlanContractVersion {
contract_id: "support-ticket-http.v1".to_owned(),
version: "v1".to_owned(),
kind: ExtractionContractKind::Service,
direction: ExtractionContractDirection::Provides,
artifact_reference: "contracts/openapi/support.v1.yaml".to_owned(),
artifact_digest: extraction_input_digest(
DIRECT_HTTP_OPENAPI_V1_FIXTURE_YAML.as_bytes(),
),
artifact_format: ExtractionContractArtifactFormat::Openapi,
tenancy_mode: ServiceTenancyMode::Required,
required_context: vec![CommonContextRequirement::Tenant],
producer_id: None,
consumer_ids: Vec::new(),
}],
expected_authority: ExtractionExpectedAuthority {
kind: ExtractionAuthorityKind::LinkedHost,
owner_id: "support-host".to_owned(),
revision: "support-authority-r7".to_owned(),
},
evidence_digests: vec![ExtractionEvidenceDigest {
reference: migration_reference.to_owned(),
digest: migration_digest.clone(),
}],
};
let plan = generate_extraction_plan(¤t_plan_inputs).expect("plan");
let scaffold = generate_extraction_scaffold(&ExtractionScaffoldInputs {
plan: plan.clone(),
module,
artifacts: vec![ExtractionScaffoldArtifact {
contract_id: "support-ticket-http.v1".to_owned(),
version: "v1".to_owned(),
contents: DIRECT_HTTP_OPENAPI_V1_FIXTURE_YAML.to_owned(),
protobuf_descriptor: None,
}],
})
.expect("scaffold");
let unchanged_files = scaffold
.files
.iter()
.map(|file| file.path.clone())
.collect();
let scaffold_apply_result = ExtractionScaffoldApplyResult {
protocol: "lenso.extraction-scaffold-apply.v1".to_owned(),
scaffold_id: scaffold.scaffold_id.clone(),
plan_id: plan.plan_id.clone(),
created_files: Vec::new(),
unchanged_files,
linked_authority_remains_authoritative: true,
effects: ExtractionScaffoldEffects::default(),
};
ExtractionRunInputs {
plan,
current_plan_inputs,
scaffold,
scaffold_apply_result,
migrations: vec![ExtractionMigrationArtifact {
migration_id: "0001_create_support_tickets".to_owned(),
source_reference: migration_reference.to_owned(),
source_digest: migration_digest,
sql: sql.to_owned(),
}],
}
}
fn safe_sql() -> &'static str {
"create schema if not exists support;\ncreate table if not exists support.tickets (id text primary key);\n"
}
fn request_for(
run: &ExtractionRun,
operation: ExtractionExpansionOperation,
) -> ExtractionWorkloadRequest {
ExtractionWorkloadRequest {
run_id: run.run_id.clone(),
plan_id: run.plan.plan_id.clone(),
plan_digest: run.plan.plan_digest.clone(),
expected_state: run.expected_state.clone(),
expected_state_digest: run.expected_state_digest.clone(),
operation,
}
}
#[test]
fn dry_run_reports_the_exact_apply_operations_without_effects() {
let inputs = run_inputs(safe_sql());
let apply = start_destination_expansion(&inputs).expect("apply run");
let dry_run = dry_run_destination_expansion(&inputs).expect("dry run");
assert_eq!(dry_run.run_id, apply.run_id);
assert_eq!(dry_run.expected_state, apply.expected_state);
assert_eq!(dry_run.ordered_operations, apply.ordered_operations);
assert_eq!(dry_run.effects, ExtractionRunEffects::default());
assert!(extraction_run_integrity_is_valid(&dry_run));
assert_eq!(
dry_run
.ordered_operations
.iter()
.map(|operation| operation.kind)
.collect::<Vec<_>>(),
vec![
ExtractionExpansionOperationKind::CreateIsolatedStore,
ExtractionExpansionOperationKind::ApplyExpandMigration,
ExtractionExpansionOperationKind::VerifyMigrationWorkload,
ExtractionExpansionOperationKind::VerifyCandidateHealth,
]
);
}
#[test]
fn destructive_or_data_mutating_sql_is_rejected_before_workload_behavior() {
let inputs = run_inputs("drop table support.tickets;");
let error = start_destination_expansion(&inputs).expect_err("drop must fail closed");
assert_eq!(
error.code,
ExtractionRunStartErrorCode::MigrationNotExpandFirst
);
assert_eq!(error.effects, ExtractionRunEffects::default());
assert!(!validate_expand_first_postgres_sql(
"alter table support.tickets drop column title;"
));
assert!(!validate_expand_first_postgres_sql(
"insert into support.tickets values ('x');"
));
}
#[tokio::test]
async fn interrupted_run_recovers_the_workload_receipt_without_repeating_effects() {
let inputs = run_inputs(safe_sql());
let workload = FakeWorkload::default();
let mut run = start_destination_expansion(&inputs).expect("run");
run = advance_destination_expansion(run, &inputs.current_plan_inputs, &workload)
.await
.expect("create Store");
assert_eq!(workload.execution_count(), 1);
let migration = run
.ordered_operations
.iter()
.find(|operation| {
operation.kind == ExtractionExpansionOperationKind::ApplyExpandMigration
})
.cloned()
.unwrap();
workload
.execute(&request_for(&run, migration))
.await
.expect("commit effect and durable receipt before simulated crash");
assert_eq!(workload.execution_count(), 2);
run = advance_destination_expansion(run, &inputs.current_plan_inputs, &workload)
.await
.expect("recover receipt");
assert_eq!(workload.execution_count(), 2, "migration must not repeat");
while run.current_phase.status != ExtractionRunStatus::Succeeded {
run = advance_destination_expansion(run, &inputs.current_plan_inputs, &workload)
.await
.expect("advance remaining health checks");
}
assert_eq!(run.receipts.len(), run.ordered_operations.len());
assert!(run.effects.creates_destination_store);
assert!(run.effects.applies_destination_schema);
assert!(!run.effects.copies_service_data);
assert!(!run.effects.mutates_source_store);
assert!(!run.effects.mutates_linked_implementation);
assert!(!run.effects.changes_authority);
assert!(!run.effects.performs_destructive_cleanup);
assert!(extraction_run_integrity_is_valid(&run));
}
#[tokio::test]
async fn stale_plan_blocks_before_workload_behavior() {
let inputs = run_inputs(safe_sql());
let workload = FakeWorkload::default();
let run = start_destination_expansion(&inputs).expect("run");
let mut changed = inputs.current_plan_inputs.clone();
changed.expected_authority.revision = "support-authority-r8".to_owned();
let blocked = advance_destination_expansion(run, &changed, &workload)
.await
.expect("stale plan becomes blocked evidence");
assert_eq!(blocked.current_phase.status, ExtractionRunStatus::Blocked);
assert_eq!(blocked.errors[0].code, ExtractionRunErrorCode::PlanStale);
assert_eq!(workload.execution_count(), 0);
assert!(extraction_run_integrity_is_valid(&blocked));
}
#[test]
fn public_schema_accepts_a_versioned_run() {
let run = dry_run_destination_expansion(&run_inputs(safe_sql())).expect("dry run");
let value = serde_json::to_value(&run).unwrap();
let validator = jsonschema::validator_for(&extraction_run_schema()).unwrap();
assert!(validator.is_valid(&value));
}
}