use super::{
core_resource_failure, fmt, invalid_resource, new_deferred_device_cleanup_domain, watch, Arc,
AtomicBool, AtomicU8, BTreeMap, BTreeSet, BufferDescriptor, CoreOwnedAllocation,
DeviceCapacityClaim, DriverCommitAcknowledgement, DynamicPoolMaintenanceController,
DynamicPoolSet, ErasedPlanStaticDriver, Mutex, Ordering, PhantomData, PlanRuntimeResources,
PlanRuntimeStatic, PlanStaticResources, RefCell, ResourceAbandonSignal, ResourceActionCursor,
ResourceCommitView, ResourceCompensationRecord, ResourceDriverFailure, ResourceFailureId,
ResourceFailurePoint, ResourceFailureReceipt, ResourceId, ResourceLeaseAction,
ResourceLeaseState, ResourceLeaseTransitionReceipt, ResourceLeaseValidationContext,
ResourceLedgerEntrySnapshot, ResourceLedgerSnapshot, ResourceOwnedBuffer,
ResourceOwnershipReason, ResourcePoolOwnership, ResourceRecoveryStrategy,
ResourceReservationBatch, ResourceTransactionAction, ResourceTransactionContext,
ResourceTransactionDriver, ResourceTransactionIdentity, ResourceTransactionState,
ResourceTransitionReceipt, ResourceTransitionRecord, ResourceTransitionValidationContext,
RwLock, StaticProvisioningBinding, StaticProvisioningLease, StaticProvisioningPermit,
VNextError, PLAN_RUNTIME_OPEN,
};
mod sealed {
pub trait Sealed {}
}
pub trait TransactionStage: sealed::Sealed {
const STATE: ResourceTransactionState;
const TERMINAL: bool;
}
pub struct TransactionNew;
pub struct TransactionReserved;
pub struct TransactionCommitted;
pub struct TransactionRolledBack;
pub struct TransactionReleased;
pub struct TransactionQuarantined;
impl sealed::Sealed for TransactionNew {}
impl sealed::Sealed for TransactionReserved {}
impl sealed::Sealed for TransactionCommitted {}
impl sealed::Sealed for TransactionRolledBack {}
impl sealed::Sealed for TransactionReleased {}
impl sealed::Sealed for TransactionQuarantined {}
impl TransactionStage for TransactionNew {
const STATE: ResourceTransactionState = ResourceTransactionState::New;
const TERMINAL: bool = false;
}
impl TransactionStage for TransactionReserved {
const STATE: ResourceTransactionState = ResourceTransactionState::Reserved;
const TERMINAL: bool = false;
}
impl TransactionStage for TransactionCommitted {
const STATE: ResourceTransactionState = ResourceTransactionState::Committed;
const TERMINAL: bool = false;
}
impl TransactionStage for TransactionRolledBack {
const STATE: ResourceTransactionState = ResourceTransactionState::RolledBack;
const TERMINAL: bool = true;
}
impl TransactionStage for TransactionReleased {
const STATE: ResourceTransactionState = ResourceTransactionState::Released;
const TERMINAL: bool = true;
}
impl TransactionStage for TransactionQuarantined {
const STATE: ResourceTransactionState = ResourceTransactionState::Quarantined;
const TERMINAL: bool = true;
}
struct PendingSubsetRelease {
target_orders: Vec<usize>,
failure: ResourceFailureReceipt,
}
#[must_use = "dropping a nonterminal resource transaction emits an abandon signal"]
pub struct ResourceTransaction<D: ResourceTransactionDriver, S: TransactionStage> {
driver: Option<D>,
maintenance_controller: Option<DynamicPoolMaintenanceController<D::Runtime>>,
dynamic_pools: Option<Arc<DynamicPoolSet<D::Runtime>>>,
identity: ResourceTransactionIdentity,
admission: StaticProvisioningBinding,
reservations: ResourceReservationBatch,
capacity_claim: Option<DeviceCapacityClaim>,
allocation_issued: Vec<AtomicBool>,
pending_allocations: Vec<RefCell<Option<CoreOwnedAllocation<D::Buffer>>>>,
states: Vec<ResourceTransactionState>,
lease: Option<StaticProvisioningLease<D::Runtime>>,
receipts: Vec<ResourceTransitionReceipt>,
lease_receipts: Vec<ResourceLeaseTransitionReceipt>,
recovery_history: Vec<ResourceFailureReceipt>,
latest_transition_context: Option<ResourceTransitionValidationContext>,
latest_lease_context: Option<ResourceLeaseValidationContext>,
pending_failure: Option<ResourceFailureReceipt>,
pending_subset_release: Option<PendingSubsetRelease>,
next_failure_id: u64,
finalized: bool,
stage: PhantomData<S>,
}
impl<D: ResourceTransactionDriver, S: TransactionStage> ResourceTransaction<D, S> {
pub fn maintenance_controller(&self) -> &DynamicPoolMaintenanceController<D::Runtime> {
self.maintenance_controller
.as_ref()
.expect("live resource transaction owns dynamic-pool maintenance authority")
}
pub fn identity(&self) -> &ResourceTransactionIdentity {
&self.identity
}
pub fn admission(&self) -> &StaticProvisioningBinding {
&self.admission
}
pub fn reservations(&self) -> &ResourceReservationBatch {
&self.reservations
}
pub fn receipts(&self) -> &[ResourceTransitionReceipt] {
&self.receipts
}
pub fn lease_receipts(&self) -> &[ResourceLeaseTransitionReceipt] {
&self.lease_receipts
}
pub fn recovery_history(&self) -> &[ResourceFailureReceipt] {
&self.recovery_history
}
pub const fn state(&self) -> ResourceTransactionState {
S::STATE
}
pub fn actual_states(&self) -> &[ResourceTransactionState] {
&self.states
}
pub fn ledger_snapshot(&self) -> ResourceLedgerSnapshot {
ResourceLedgerSnapshot {
identity: self.identity.clone(),
admission: self.admission.clone(),
entries: self.ledger_snapshot_entries(),
}
}
pub fn latest_transition_validation_context(
&self,
) -> Option<&ResourceTransitionValidationContext> {
self.latest_transition_context.as_ref()
}
pub fn latest_lease_validation_context(&self) -> Option<&ResourceLeaseValidationContext> {
self.latest_lease_context.as_ref()
}
fn ledger_snapshot_entries(&self) -> Vec<ResourceLedgerEntrySnapshot> {
let lease = self
.lease
.as_ref()
.expect("every resource transaction owns its batch lease ledger");
lease
.slots
.iter()
.zip(&self.pending_allocations)
.zip(&self.states)
.map(|((slot, pending), &transaction_state)| {
let pending = pending.borrow();
ResourceLedgerEntrySnapshot {
entry: slot.entry.clone(),
transaction_state,
buffer_present: slot.buffer.is_some() || pending.is_some(),
actual_resource_id: slot
.actual_resource_id
.clone()
.or_else(|| pending.as_ref().map(|value| value.resource_id.clone())),
actual_generation: slot
.actual_generation
.or_else(|| pending.as_ref().map(|value| value.generation)),
actual_descriptor: slot
.descriptor
.clone()
.or_else(|| pending.as_ref().map(|value| value.descriptor.clone())),
}
})
.collect()
}
fn driver_and_context(
&mut self,
order: usize,
action: ResourceTransactionAction,
) -> (&mut D, ResourceTransactionContext<'_, D::Runtime>) {
let before = self.states[order];
let driver = self
.driver
.as_mut()
.expect("live transaction owns its resource driver");
let context = ResourceTransactionContext {
runtime: &self
.lease
.as_ref()
.expect("transaction owns lease ledger")
.runtime,
identity: &self.identity,
binding: &self.admission,
reservations: &self.reservations,
cursor: Some(ResourceActionCursor {
order,
action,
before,
allocation_authorized: action == ResourceTransactionAction::Commit,
}),
allocation_authority: (action == ResourceTransactionAction::Commit)
.then_some(&self.allocation_issued[order]),
pending_allocation: (action == ResourceTransactionAction::Commit)
.then_some(&self.pending_allocations[order]),
};
(driver, context)
}
fn make_record(
&self,
order: usize,
action: ResourceTransactionAction,
before: ResourceTransactionState,
after: ResourceTransactionState,
) -> ResourceTransitionRecord {
ResourceTransitionRecord::from_reservation(
&self.identity,
&self.admission,
&self.reservations.reservations[order],
action,
before,
after,
order,
)
}
fn transition_receipt(
&mut self,
action: ResourceTransactionAction,
before: Vec<ResourceLedgerEntrySnapshot>,
records: Vec<ResourceTransitionRecord>,
) -> Result<ResourceTransitionReceipt, VNextError> {
let context = ResourceTransitionValidationContext {
identity: self.identity.clone(),
admission: self.admission.clone(),
action,
before,
after: self.ledger_snapshot_entries(),
};
let receipt = ResourceTransitionReceipt::from_context(&context, records)?;
self.latest_transition_context = Some(context);
Ok(receipt)
}
fn lease_transition(
&mut self,
resource_ids: &[ResourceId],
action: ResourceLeaseAction,
) -> Result<ResourceLeaseTransitionReceipt, VNextError> {
if self.pending_subset_release.is_some() {
return Err(invalid_resource(
"lease policy cannot change during pending release recovery",
));
}
let orders = self.orders_for_ids(resource_ids)?;
if orders
.iter()
.any(|&order| self.states[order] != ResourceTransactionState::Committed)
{
return Err(invalid_resource(
"lease policy targets a resource that is not actually committed",
));
}
let before_ledger = self.ledger_snapshot_entries();
let (before, after, entries) = self
.lease
.as_mut()
.expect("transaction owns lease ledger")
.transition_subset(&orders, action)?;
let context = ResourceLeaseValidationContext {
identity: self.identity.clone(),
admission: self.admission.clone(),
action,
before: before_ledger,
after: self.ledger_snapshot_entries(),
};
let receipt =
ResourceLeaseTransitionReceipt::from_context(&context, before, after, entries)?;
self.latest_lease_context = Some(context);
self.lease_receipts.push(receipt.clone());
Ok(receipt)
}
fn orders_for_ids(&self, resource_ids: &[ResourceId]) -> Result<Vec<usize>, VNextError> {
if resource_ids.is_empty() {
return Err(invalid_resource("resource subset must not be empty"));
}
let mut unique = BTreeSet::new();
let by_id = self
.reservations
.reservations()
.iter()
.enumerate()
.map(|(order, reservation)| (reservation.resource_id(), order))
.collect::<BTreeMap<_, _>>();
let mut orders = Vec::with_capacity(resource_ids.len());
for resource_id in resource_ids {
if !unique.insert(resource_id) {
return Err(invalid_resource("resource subset contains duplicates"));
}
orders.push(
*by_id
.get(resource_id)
.ok_or_else(|| invalid_resource("resource subset is not admitted"))?,
);
}
orders.sort_unstable();
Ok(orders)
}
fn set_pending_failure(&mut self, failure: &ResourceFailureReceipt) {
self.pending_failure = Some(failure.clone());
}
fn clear_pending_failure(&mut self) {
self.pending_failure = None;
}
fn issue_failure_id(&mut self) -> ResourceFailureId {
let current = self.next_failure_id;
self.next_failure_id = current
.checked_add(1)
.expect("resource transaction failure id space is exhausted");
ResourceFailureId::try_from(current).expect("transaction failure ids start at one")
}
fn advance<T: TransactionStage>(
mut self,
receipt: Option<ResourceTransitionReceipt>,
) -> ResourceTransaction<D, T> {
self.finalized = true;
if let Some(receipt) = receipt {
self.receipts.push(receipt);
}
ResourceTransaction {
driver: self.driver.take(),
maintenance_controller: self.maintenance_controller.take(),
dynamic_pools: self.dynamic_pools.take(),
identity: self.identity.clone(),
admission: self.admission.clone(),
reservations: self.reservations.clone(),
capacity_claim: if T::TERMINAL {
if let Some(mut claim) = self.capacity_claim.take() {
claim.release();
}
None
} else {
self.capacity_claim.take()
},
allocation_issued: std::mem::take(&mut self.allocation_issued),
pending_allocations: std::mem::take(&mut self.pending_allocations),
states: std::mem::take(&mut self.states),
lease: self.lease.take(),
receipts: std::mem::take(&mut self.receipts),
lease_receipts: std::mem::take(&mut self.lease_receipts),
recovery_history: std::mem::take(&mut self.recovery_history),
latest_transition_context: self.latest_transition_context.take(),
latest_lease_context: self.latest_lease_context.take(),
pending_failure: None,
pending_subset_release: None,
next_failure_id: self.next_failure_id,
finalized: T::TERMINAL,
stage: PhantomData,
}
}
fn abandon_signal(&self) -> ResourceAbandonSignal {
ResourceAbandonSignal {
identity: self.identity.clone(),
admission: self.admission.clone(),
state: S::STATE,
pending_action: self
.pending_failure
.as_ref()
.map(ResourceFailureReceipt::action),
ledger: self.ledger_snapshot_entries(),
active_sequence_slots: Vec::new(),
poisoned_sequence_slots: Vec::new(),
undrained_sequence_slots: Vec::new(),
failure: self
.pending_failure
.as_ref()
.map(|receipt| receipt.failure.clone()),
}
}
fn take_all_owned_buffers(&mut self) -> Vec<ResourceOwnedBuffer<D::Buffer>> {
let mut buffers = self
.lease
.as_mut()
.expect("transaction owns lease ledger")
.take_owned_buffers(&self.reservations);
for (order, pending) in self.pending_allocations.iter_mut().enumerate() {
let Some(allocation) = pending.get_mut().take() else {
continue;
};
let reservation = &self.reservations.reservations[order];
buffers.push(ResourceOwnedBuffer {
order,
expected_resource_id: reservation.resource_id.clone(),
actual_resource_id: allocation.resource_id,
expected_generation: reservation.generation,
actual_generation: allocation.generation,
expected_descriptor: BufferDescriptor {
resource_id: reservation.resource_id.clone(),
size_bytes: reservation.size_bytes,
alignment_bytes: reservation.alignment_bytes,
usage: reservation.usage,
element_type: reservation.element_type,
},
actual_descriptor: allocation.descriptor,
buffer: allocation.buffer,
});
}
buffers.sort_by_key(|buffer| buffer.order);
buffers
}
fn quarantine_live(&mut self) -> Result<Vec<ResourceTransitionRecord>, ResourceDriverFailure> {
let buffers = self.take_all_owned_buffers();
let ownership = ResourcePoolOwnership {
runtime: Arc::clone(
&self
.lease
.as_ref()
.expect("transaction owns lease ledger")
.runtime,
),
pool_identity: self.admission.pool_identity.clone(),
reason: ResourceOwnershipReason::Quarantine,
signal: None,
buffers,
capacity_claim: self.capacity_claim.take(),
};
let result = {
let driver = self
.driver
.as_mut()
.expect("live transaction owns its resource driver");
let context = ResourceTransactionContext {
runtime: &self
.lease
.as_ref()
.expect("transaction owns lease ledger")
.runtime,
identity: &self.identity,
binding: &self.admission,
reservations: &self.reservations,
cursor: None,
allocation_authority: None,
pending_allocation: None,
};
driver.quarantine_transaction(&context, ownership)
};
if let Err(failure) = result {
let (failure, mut ownership) = failure.into_parts();
let expected_claimed_bytes = self
.states
.iter()
.zip(self.reservations.reservations())
.filter(|(state, _)| state.is_live())
.map(|(_, reservation)| reservation.size_bytes())
.sum::<u64>();
if ownership.pool_identity != self.admission.pool_identity
|| ownership.claimed_bytes() != expected_claimed_bytes
{
std::mem::forget(ownership);
return Err(ResourceDriverFailure::new(core_resource_failure(
"ownership_transfer_identity_mismatch",
"quarantine failure returned ownership for a different resource pool",
false,
))
.expect("core failure has resource domain"));
}
self.lease
.as_mut()
.expect("transaction owns lease ledger")
.restore_owned_buffers(std::mem::take(&mut ownership.buffers));
self.capacity_claim = ownership.capacity_claim.take();
return Err(failure);
}
let mut records = Vec::new();
for order in 0..self.states.len() {
let before = self.states[order];
if before.is_live() {
self.states[order] = ResourceTransactionState::Quarantined;
records.push(self.make_record(
order,
ResourceTransactionAction::Quarantine,
before,
ResourceTransactionState::Quarantined,
));
}
}
Ok(records)
}
}
impl<D: ResourceTransactionDriver, S: TransactionStage> Drop for ResourceTransaction<D, S> {
fn drop(&mut self) {
if self.finalized || S::TERMINAL {
return;
}
let signal = self.abandon_signal();
drop(self.maintenance_controller.take());
drop(self.dynamic_pools.take());
let buffers = self.take_all_owned_buffers();
let ownership = ResourcePoolOwnership {
runtime: Arc::clone(
&self
.lease
.as_ref()
.expect("transaction owns lease ledger")
.runtime,
),
pool_identity: self.admission.pool_identity.clone(),
reason: ResourceOwnershipReason::Abandon,
signal: Some(signal),
buffers,
capacity_claim: self.capacity_claim.take(),
};
if let Some(driver) = self.driver.as_mut() {
let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
driver.abandon_transaction(ownership);
}));
} else {
std::mem::forget(ownership);
}
self.finalized = true;
}
}
impl<D: ResourceTransactionDriver> ResourceTransaction<D, TransactionNew> {
pub fn begin(
driver: D,
identity: ResourceTransactionIdentity,
permit: StaticProvisioningPermit<D::Runtime>,
) -> Result<Self, VNextError> {
if identity.pool_id != permit.binding.pool_id()
|| identity.request_id != permit.binding.request_id
|| permit.binding.pool_identity.plan_id != permit.binding.plan_id
|| permit.binding.pool_identity.plan_hash != permit.binding.plan_hash
|| permit.binding.pool_identity.device_id != permit.binding.device_id
|| permit
.binding
.pool_identity
.device_runtime_implementation_fingerprint
!= permit.binding.device_runtime_implementation_fingerprint
|| permit.binding.pool_identity.admission_generation
!= permit.binding.admission_generation
|| permit.binding.request_id != *permit.reservations.request_id()
|| permit.binding.admitted_bytes != permit.reservations.total_size_bytes()
|| permit.binding.plan_static_bytes != permit.reservations.plan_static_size_bytes()
|| permit.binding.admitted_bytes != permit.binding.plan_static_bytes
|| permit.binding.plan_static_bytes > permit.binding.usable_capacity_bytes
|| permit.binding.usable_capacity_bytes > permit.binding.device_capacity_bytes
|| driver.device_id() != &permit.binding.device_id
|| driver.device_runtime_implementation_fingerprint()
!= permit.binding.device_runtime_implementation_fingerprint
|| driver.device_capacity_bytes() != permit.binding.device_capacity_bytes
|| !Arc::ptr_eq(driver.runtime(), &permit.runtime)
{
return Err(invalid_resource(
"transaction identity, driver device, or admitted capacity does not match its permit",
));
}
let _ = permit.seal;
if permit.capacity_claim.bytes() != permit.binding.admitted_bytes {
return Err(invalid_resource(
"device capacity claim differs from admitted bytes",
));
}
let states = vec![ResourceTransactionState::New; permit.reservations.reservations.len()];
let allocation_issued = (0..states.len()).map(|_| AtomicBool::new(false)).collect();
let pending_allocations = (0..states.len()).map(|_| RefCell::new(None)).collect();
let lease = StaticProvisioningLease::new(
Arc::clone(&permit.runtime),
&identity,
&permit.binding,
&permit.reservations,
);
Ok(Self {
driver: Some(driver),
maintenance_controller: Some(permit.maintenance_controller),
dynamic_pools: Some(permit.dynamic_pools),
identity,
admission: permit.binding,
reservations: permit.reservations,
capacity_claim: Some(permit.capacity_claim),
allocation_issued,
pending_allocations,
states,
lease: Some(lease),
receipts: Vec::new(),
lease_receipts: Vec::new(),
recovery_history: Vec::new(),
latest_transition_context: None,
latest_lease_context: None,
pending_failure: None,
pending_subset_release: None,
next_failure_id: 1,
finalized: false,
stage: PhantomData,
})
}
pub fn reserve(
mut self,
) -> Result<
ResourceTransaction<D, TransactionReserved>,
ResourcePrepareTransitionError<D, TransactionNew>,
> {
debug_assert!(
self.states
.iter()
.all(|state| *state == ResourceTransactionState::New),
"new transaction ledger must be uniformly new"
);
let ledger_before = self.ledger_snapshot_entries();
let mut completed = Vec::new();
for order in 0..self.states.len() {
let result = {
let reservation = self.reservations.reservations[order].clone();
let (driver, context) =
self.driver_and_context(order, ResourceTransactionAction::Reserve);
driver.reserve_resource(&context, &reservation)
};
match result {
Ok(()) => {
let before = self.states[order];
let after = before
.transition(
&self.reservations.reservations[order].resource_id,
ResourceTransactionAction::Reserve,
)
.expect("reserve was preflight validated");
self.states[order] = after;
completed.push(self.make_record(
order,
ResourceTransactionAction::Reserve,
before,
after,
));
}
Err(failure) => {
let failure_id = self.issue_failure_id();
let receipt = ResourceFailureReceipt::new(
failure_id,
&self.identity,
&self.admission,
ResourceTransactionAction::Reserve,
failure.into_failure(),
Some(ResourceFailurePoint::new(
&self.reservations.reservations[order],
order,
self.states[order],
)),
completed,
ResourceRecoveryStrategy::ReverseCompensation,
ledger_before,
self.ledger_snapshot_entries(),
);
self.set_pending_failure(&receipt);
return Err(ResourcePrepareTransitionError {
transaction: Some(self),
failure: receipt,
});
}
}
}
let receipt = self
.transition_receipt(ResourceTransactionAction::Reserve, ledger_before, completed)
.expect("core reserve journal must validate");
Ok(self.advance(Some(receipt)))
}
}
#[derive(Debug)]
pub enum ResourceCommitTransitionError<D: ResourceTransactionDriver> {
Recoverable(ResourcePrepareTransitionError<D, TransactionReserved>),
Poisoned(ResourcePoisonedTransaction<D>),
}
impl<D: ResourceTransactionDriver> ResourceTransaction<D, TransactionReserved> {
pub fn commit(
mut self,
) -> Result<ResourceTransaction<D, TransactionCommitted>, ResourceCommitTransitionError<D>>
{
debug_assert!(
self.states
.iter()
.all(|state| *state == ResourceTransactionState::Reserved),
"reserved transaction must be uniformly reserved before commit"
);
let ledger_before = self.ledger_snapshot_entries();
let mut completed = Vec::new();
for order in 0..self.states.len() {
let driver_result = {
let reservation = self.reservations.reservations[order].clone();
let (driver, context) =
self.driver_and_context(order, ResourceTransactionAction::Commit);
driver
.commit_resource(&context, &reservation)
.map(|receipt| DriverCommitAcknowledgement::from_receipt(&receipt))
};
let allocation = self.pending_allocations[order].get_mut().take();
match (driver_result, allocation) {
(Ok(acknowledgement), Some(allocation))
if acknowledgement.matches(&allocation)
&& allocation.matches(&self.reservations.reservations[order]) =>
{
self.lease
.as_mut()
.expect("transaction owns lease ledger")
.install(order, allocation);
let before = self.states[order];
let after = before
.transition(
&self.reservations.reservations[order].resource_id,
ResourceTransactionAction::Commit,
)
.expect("commit was preflight validated");
self.states[order] = after;
completed.push(self.make_record(
order,
ResourceTransactionAction::Commit,
before,
after,
));
}
(driver_result, Some(poisoned)) => {
let actual_resource_id = poisoned.resource_id.clone();
let actual_generation = poisoned.generation;
let failure_envelope = match driver_result {
Ok(_) => core_resource_failure(
"invalid_commit_outcome",
format!(
"core-owned allocation `{}` generation {} does not match allocation `{}` generation {}",
actual_resource_id,
actual_generation,
self.reservations.reservations[order].resource_id(),
self.reservations.reservations[order].generation(),
),
false,
),
Err(failure) => failure.into_failure(),
};
self.lease
.as_mut()
.expect("transaction owns lease ledger")
.install(order, poisoned);
let failure_id = self.issue_failure_id();
let failure = ResourceFailureReceipt::new(
failure_id,
&self.identity,
&self.admission,
ResourceTransactionAction::Commit,
failure_envelope,
Some(ResourceFailurePoint::new(
&self.reservations.reservations[order],
order,
self.states[order],
)),
completed,
ResourceRecoveryStrategy::ReconcileOrQuarantine,
ledger_before,
self.ledger_snapshot_entries(),
);
self.set_pending_failure(&failure);
return Err(ResourceCommitTransitionError::Poisoned(
ResourcePoisonedTransaction {
transaction: Some(self),
failure,
poisoned_order: order,
},
));
}
(driver_result, None) => {
let failure = match driver_result {
Ok(_) => core_resource_failure(
"commit_acknowledged_without_allocation",
"driver acknowledged commit without a core-owned allocation",
false,
),
Err(failure) => failure.into_failure(),
};
let failure_id = self.issue_failure_id();
let receipt = ResourceFailureReceipt::new(
failure_id,
&self.identity,
&self.admission,
ResourceTransactionAction::Commit,
failure,
Some(ResourceFailurePoint::new(
&self.reservations.reservations[order],
order,
self.states[order],
)),
completed,
ResourceRecoveryStrategy::ReverseCompensation,
ledger_before,
self.ledger_snapshot_entries(),
);
self.set_pending_failure(&receipt);
return Err(ResourceCommitTransitionError::Recoverable(
ResourcePrepareTransitionError {
transaction: Some(self),
failure: receipt,
},
));
}
}
}
let receipt = self
.transition_receipt(ResourceTransactionAction::Commit, ledger_before, completed)
.expect("core commit journal must validate");
Ok(self.advance(Some(receipt)))
}
pub fn rollback(
mut self,
) -> Result<ResourceTransaction<D, TransactionRolledBack>, ResourceRollbackTransitionError<D>>
{
debug_assert!(
self.states
.iter()
.all(|state| *state == ResourceTransactionState::Reserved),
"rollback starts from a uniformly reserved transaction"
);
let ledger_before = self.ledger_snapshot_entries();
let mut completed = Vec::new();
for order in 0..self.states.len() {
let result = {
let reservation = self.reservations.reservations[order].clone();
let (driver, context) =
self.driver_and_context(order, ResourceTransactionAction::Rollback);
driver.rollback_resource(&context, &reservation)
};
match result {
Ok(()) => {
let before = self.states[order];
self.states[order] = ResourceTransactionState::RolledBack;
completed.push(self.make_record(
order,
ResourceTransactionAction::Rollback,
before,
ResourceTransactionState::RolledBack,
));
}
Err(failure) => {
let failure_id = self.issue_failure_id();
let receipt = ResourceFailureReceipt::new(
failure_id,
&self.identity,
&self.admission,
ResourceTransactionAction::Rollback,
failure.into_failure(),
Some(ResourceFailurePoint::new(
&self.reservations.reservations[order],
order,
self.states[order],
)),
completed,
ResourceRecoveryStrategy::ForwardCompletion,
ledger_before,
self.ledger_snapshot_entries(),
);
self.set_pending_failure(&receipt);
return Err(ResourceRollbackTransitionError {
transaction: Some(self),
failure: receipt,
});
}
}
}
let receipt = self
.transition_receipt(
ResourceTransactionAction::Rollback,
ledger_before,
completed,
)
.expect("core rollback journal must validate");
Ok(self.advance(Some(receipt)))
}
}
impl<D: ResourceTransactionDriver> ResourceTransaction<D, TransactionCommitted> {
pub fn into_plan_runtime(
mut self,
) -> Result<Arc<PlanRuntimeResources<D::Runtime>>, PlanRuntimeHandoffError<D>>
where
D: 'static,
{
let valid = self
.states
.iter()
.all(|state| *state == ResourceTransactionState::Committed)
&& self.pending_failure.is_none()
&& self.pending_subset_release.is_none()
&& self
.pending_allocations
.iter()
.all(|allocation| allocation.borrow().is_none())
&& self.lease.as_ref().is_some_and(|lease| {
!lease.slots.is_empty()
&& lease.slots.iter().all(|slot| {
slot.buffer.is_some()
&& slot.descriptor.is_some()
&& slot.entry.state == ResourceLeaseState::Active
})
})
&& self.driver.is_some()
&& self.maintenance_controller.is_some()
&& self.dynamic_pools.is_some()
&& self.capacity_claim.is_some();
if !valid {
return Err(PlanRuntimeHandoffError {
error: invalid_resource(
"plan runtime handoff requires one complete active committed transaction",
),
transaction: Some(self),
});
}
let lease = self
.lease
.take()
.expect("validated plan runtime handoff owns its static lease");
let runtime = Arc::clone(&lease.runtime);
let dynamic_pools = self
.dynamic_pools
.take()
.expect("validated plan runtime handoff owns its dynamic pools");
let driver: Box<dyn ErasedPlanStaticDriver<D::Runtime>> = Box::new(
self.driver
.take()
.expect("validated plan runtime handoff owns its driver"),
);
let static_resources = PlanStaticResources {
driver: Mutex::new(Some(driver)),
identity: self.identity.clone(),
admission: self.admission.clone(),
reservations: self.reservations.clone(),
states: std::mem::take(&mut self.states),
capacity_claim: self.capacity_claim.take(),
lease: Some(lease),
finalized: false,
};
let maintenance_controller = self
.maintenance_controller
.take()
.expect("validated plan runtime handoff owns maintenance authority");
let (lifecycle_tx, _) = watch::channel(PLAN_RUNTIME_OPEN);
self.finalized = true;
Ok(Arc::new(PlanRuntimeResources {
lifecycle: RwLock::new(()),
phase: AtomicU8::new(PLAN_RUNTIME_OPEN),
lifecycle_tx,
maintenance_controller,
dynamic_pools,
static_resources: PlanRuntimeStatic::Static(static_resources),
runtime,
deferred_cleanup_domain: new_deferred_device_cleanup_domain(),
}))
}
pub fn lease(&self) -> &StaticProvisioningLease<D::Runtime> {
self.lease
.as_ref()
.expect("committed transaction owns its batch lease")
}
fn defer_lease(
&mut self,
resource_ids: &[ResourceId],
) -> Result<ResourceLeaseTransitionReceipt, VNextError> {
self.lease_transition(resource_ids, ResourceLeaseAction::Defer)
}
fn resume_lease(
&mut self,
resource_ids: &[ResourceId],
) -> Result<ResourceLeaseTransitionReceipt, VNextError> {
self.lease_transition(resource_ids, ResourceLeaseAction::Resume)
}
fn cancel_lease(
&mut self,
resource_ids: &[ResourceId],
) -> Result<ResourceLeaseTransitionReceipt, VNextError> {
self.lease_transition(resource_ids, ResourceLeaseAction::Cancel)
}
pub fn defer_all(&mut self) -> Result<ResourceLeaseTransitionReceipt, VNextError> {
let ids = self
.reservations
.resource_ids()
.cloned()
.collect::<Vec<_>>();
self.defer_lease(&ids)
}
pub fn resume_all(&mut self) -> Result<ResourceLeaseTransitionReceipt, VNextError> {
let ids = self
.reservations
.resource_ids()
.cloned()
.collect::<Vec<_>>();
self.resume_lease(&ids)
}
pub fn cancel_all(&mut self) -> Result<ResourceLeaseTransitionReceipt, VNextError> {
let ids = self
.reservations
.resource_ids()
.cloned()
.collect::<Vec<_>>();
self.cancel_lease(&ids)
}
fn release_all_resources(
&mut self,
) -> Result<ResourceTransitionReceipt, ResourceFailureReceipt> {
if let Some(pending) = &self.pending_subset_release {
return Err(pending.failure.clone());
}
let orders = (0..self.states.len()).collect::<Vec<_>>();
if let Some(&order) = orders
.iter()
.find(|&&order| self.states[order] != ResourceTransactionState::Committed)
{
return Err(self.local_release_failure(format!(
"resource `{}` is not actually committed",
self.reservations.reservations[order].resource_id()
)));
}
let ledger_before = self.ledger_snapshot_entries();
let mut completed = Vec::new();
for &order in &orders {
let result = {
let reservation = self.reservations.reservations[order].clone();
let buffer = self
.lease
.as_ref()
.expect("committed transaction owns lease")
.buffer(order)
.expect("committed ledger entry owns its buffer");
let driver = self.driver.as_mut().expect("live transaction owns driver");
let context = ResourceTransactionContext {
runtime: &self.lease.as_ref().expect("transaction owns lease").runtime,
identity: &self.identity,
binding: &self.admission,
reservations: &self.reservations,
cursor: Some(ResourceActionCursor {
order,
action: ResourceTransactionAction::Release,
before: self.states[order],
allocation_authorized: false,
}),
allocation_authority: None,
pending_allocation: None,
};
driver.release_resource(&context, &reservation, buffer)
};
match result {
Ok(()) => {
let before = self.states[order];
self.lease
.as_mut()
.expect("transaction owns lease ledger")
.clear(order);
self.states[order] = ResourceTransactionState::Released;
self.capacity_claim
.as_mut()
.expect("live transaction owns its static capacity claim")
.release_bytes(self.reservations.reservations[order].size_bytes());
completed.push(self.make_record(
order,
ResourceTransactionAction::Release,
before,
ResourceTransactionState::Released,
));
}
Err(failure) => {
let failure_id = self.issue_failure_id();
let receipt = ResourceFailureReceipt::new(
failure_id,
&self.identity,
&self.admission,
ResourceTransactionAction::Release,
failure.into_failure(),
Some(ResourceFailurePoint::new(
&self.reservations.reservations[order],
order,
self.states[order],
)),
completed,
ResourceRecoveryStrategy::ForwardCompletion,
ledger_before,
self.ledger_snapshot_entries(),
);
self.set_pending_failure(&receipt);
self.pending_subset_release = Some(PendingSubsetRelease {
target_orders: orders,
failure: receipt.clone(),
});
return Err(receipt);
}
}
}
let receipt = self
.transition_receipt(ResourceTransactionAction::Release, ledger_before, completed)
.expect("core subset release journal must validate");
self.receipts.push(receipt.clone());
Ok(receipt)
}
pub fn complete_pending_release(
&mut self,
) -> Result<ResourceTransitionReceipt, ResourceFailureReceipt> {
let Some(mut pending) = self.pending_subset_release.take() else {
return Err(self.local_release_failure(
"there is no pending subset release to complete".to_owned(),
));
};
for &order in &pending.target_orders {
match self.states[order] {
ResourceTransactionState::Released => continue,
ResourceTransactionState::Committed => {}
actual => {
let failure = ResourceDriverFailure::new(core_resource_failure(
"release_ledger_diverged",
format!(
"resource `{}` has unexpected actual state {}",
self.reservations.reservations[order].resource_id(),
actual.as_str()
),
false,
))
.expect("core failure has resource domain");
pending.failure.record_recovery_failure(
failure,
Some(ResourceFailurePoint::new(
&self.reservations.reservations[order],
order,
actual,
)),
);
pending.failure.ledger_after = self.ledger_snapshot_entries();
self.set_pending_failure(&pending.failure);
self.pending_subset_release = Some(pending);
return Err(self
.pending_subset_release
.as_ref()
.expect("pending release was restored")
.failure
.clone());
}
}
let result = {
let reservation = self.reservations.reservations[order].clone();
let buffer = self
.lease
.as_ref()
.expect("committed transaction owns lease")
.buffer(order)
.expect("outstanding release owns its buffer");
let driver = self.driver.as_mut().expect("live transaction owns driver");
let context = ResourceTransactionContext {
runtime: &self.lease.as_ref().expect("transaction owns lease").runtime,
identity: &self.identity,
binding: &self.admission,
reservations: &self.reservations,
cursor: Some(ResourceActionCursor {
order,
action: ResourceTransactionAction::Release,
before: self.states[order],
allocation_authorized: false,
}),
allocation_authority: None,
pending_allocation: None,
};
driver.release_resource(&context, &reservation, buffer)
};
match result {
Ok(()) => {
let before = self.states[order];
self.lease
.as_mut()
.expect("transaction owns lease ledger")
.clear(order);
self.states[order] = ResourceTransactionState::Released;
self.capacity_claim
.as_mut()
.expect("live transaction owns its static capacity claim")
.release_bytes(self.reservations.reservations[order].size_bytes());
pending.failure.completed.push(self.make_record(
order,
ResourceTransactionAction::Release,
before,
ResourceTransactionState::Released,
));
}
Err(failure) => {
pending.failure.record_recovery_failure(
failure,
Some(ResourceFailurePoint::new(
&self.reservations.reservations[order],
order,
self.states[order],
)),
);
pending.failure.ledger_after = self.ledger_snapshot_entries();
self.set_pending_failure(&pending.failure);
let snapshot = pending.failure.clone();
self.pending_subset_release = Some(pending);
return Err(snapshot);
}
}
}
pending.failure.recovery_complete = true;
pending.failure.ledger_after = self.ledger_snapshot_entries();
let receipt = self
.transition_receipt(
ResourceTransactionAction::Release,
pending.failure.ledger_before.clone(),
pending.failure.completed.clone(),
)
.expect("forward-completed release journal must validate");
self.clear_pending_failure();
self.recovery_history.push(pending.failure);
self.receipts.push(receipt.clone());
Ok(receipt)
}
fn local_release_failure(&mut self, message: String) -> ResourceFailureReceipt {
let ledger = self.ledger_snapshot_entries();
let failure_id = self.issue_failure_id();
ResourceFailureReceipt::new(
failure_id,
&self.identity,
&self.admission,
ResourceTransactionAction::Release,
core_resource_failure("invalid_release_request", message, false),
None,
Vec::new(),
ResourceRecoveryStrategy::ForwardCompletion,
ledger.clone(),
ledger,
)
}
pub fn release(
mut self,
) -> Result<ResourceTransaction<D, TransactionReleased>, ResourceReleaseTransitionError<D>>
{
if self.pending_subset_release.is_some() {
let failure = self
.pending_subset_release
.as_ref()
.expect("pending release exists")
.failure
.clone();
return Err(ResourceReleaseTransitionError {
transaction: Some(self),
failure,
});
}
if self
.states
.iter()
.all(|state| *state == ResourceTransactionState::Released)
{
return Ok(self.advance(None));
}
if !self
.states
.iter()
.all(|state| *state == ResourceTransactionState::Committed)
{
let failure = self.local_release_failure(
"transaction contains non-releasable actual states".to_owned(),
);
return Err(ResourceReleaseTransitionError {
transaction: Some(self),
failure,
});
}
match self.release_all_resources() {
Ok(_) => Ok(self.advance(None)),
Err(failure) => Err(ResourceReleaseTransitionError {
transaction: Some(self),
failure,
}),
}
}
}
#[must_use = "failed plan runtime handoff retains the committed transaction"]
pub struct PlanRuntimeHandoffError<D>
where
D: ResourceTransactionDriver,
{
error: VNextError,
transaction: Option<ResourceTransaction<D, TransactionCommitted>>,
}
impl<D> PlanRuntimeHandoffError<D>
where
D: ResourceTransactionDriver,
{
pub fn error(&self) -> &VNextError {
&self.error
}
pub fn into_transaction(mut self) -> ResourceTransaction<D, TransactionCommitted> {
self.transaction
.take()
.expect("plan runtime handoff error owns its transaction")
}
}
#[must_use = "reverse recovery must complete or ownership must be quarantined"]
pub struct ResourcePrepareTransitionError<D: ResourceTransactionDriver, S: TransactionStage> {
transaction: Option<ResourceTransaction<D, S>>,
failure: ResourceFailureReceipt,
}
impl<D: ResourceTransactionDriver, S: TransactionStage> fmt::Debug
for ResourcePrepareTransitionError<D, S>
{
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ResourcePrepareTransitionError")
.field("stage", &S::STATE)
.field("failure", &self.failure)
.finish_non_exhaustive()
}
}
impl<D: ResourceTransactionDriver, S: TransactionStage> ResourcePrepareTransitionError<D, S> {
pub fn failure(&self) -> &ResourceFailureReceipt {
&self.failure
}
pub fn recover(mut self) -> Result<ResourceTransaction<D, S>, Self> {
while self.failure.compensation.len() < self.failure.completed.len() {
let record_index = self.failure.completed.len() - 1 - self.failure.compensation.len();
let attempted = self.failure.completed[record_index].clone();
let order = attempted.order as usize;
let transaction = self
.transaction
.as_mut()
.expect("prepare recovery owns transaction");
if transaction.states[order] != attempted.after {
let failure = ResourceDriverFailure::new(core_resource_failure(
"compensation_ledger_diverged",
format!(
"resource `{}` expected actual state {}, found {}",
attempted.resource_id,
attempted.after.as_str(),
transaction.states[order].as_str()
),
false,
))
.expect("core failure has resource domain");
self.failure.record_recovery_failure(
failure,
Some(ResourceFailurePoint::new(
&transaction.reservations.reservations[order],
order,
transaction.states[order],
)),
);
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.set_pending_failure(&self.failure);
return Err(self);
}
let result = match self.failure.action {
ResourceTransactionAction::Reserve => {
let reservation = transaction.reservations.reservations[order].clone();
let driver = transaction
.driver
.as_mut()
.expect("live transaction owns driver");
let context = ResourceTransactionContext {
runtime: &transaction
.lease
.as_ref()
.expect("transaction owns lease")
.runtime,
identity: &transaction.identity,
binding: &transaction.admission,
reservations: &transaction.reservations,
cursor: Some(ResourceActionCursor {
order,
action: ResourceTransactionAction::Reserve,
before: transaction.states[order],
allocation_authorized: false,
}),
allocation_authority: None,
pending_allocation: None,
};
driver.compensate_reserve_resource(&context, &reservation)
}
ResourceTransactionAction::Commit => {
let reservation = transaction.reservations.reservations[order].clone();
let buffer = transaction
.lease
.as_ref()
.expect("transaction owns lease ledger")
.buffer(order)
.expect("uncompensated commit owns its buffer");
let driver = transaction
.driver
.as_mut()
.expect("live transaction owns driver");
let context = ResourceTransactionContext {
runtime: &transaction
.lease
.as_ref()
.expect("transaction owns lease")
.runtime,
identity: &transaction.identity,
binding: &transaction.admission,
reservations: &transaction.reservations,
cursor: Some(ResourceActionCursor {
order,
action: ResourceTransactionAction::Commit,
before: transaction.states[order],
allocation_authorized: false,
}),
allocation_authority: None,
pending_allocation: None,
};
driver.compensate_commit_resource(&context, &reservation, buffer)
}
_ => unreachable!("prepare recovery handles only reserve and commit"),
};
match result {
Ok(()) => {
transaction.states[order] = attempted.before;
if self.failure.action == ResourceTransactionAction::Commit {
transaction
.lease
.as_mut()
.expect("transaction owns lease ledger")
.clear(order);
}
self.failure
.compensation
.push(ResourceCompensationRecord::from_transition(
&attempted,
self.failure.compensation.len(),
));
self.failure.ledger_after = transaction.ledger_snapshot_entries();
}
Err(failure) => {
self.failure.record_recovery_failure(
failure,
Some(ResourceFailurePoint::new(
&transaction.reservations.reservations[order],
order,
transaction.states[order],
)),
);
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.set_pending_failure(&self.failure);
return Err(self);
}
}
}
self.failure.recovery_complete = true;
let mut transaction = self
.transaction
.take()
.expect("prepare recovery owns transaction");
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.clear_pending_failure();
if self.failure.action == ResourceTransactionAction::Commit {
for authority in &transaction.allocation_issued {
authority.store(false, Ordering::Release);
}
}
transaction.recovery_history.push(self.failure);
Ok(transaction)
}
pub fn quarantine(mut self) -> Result<ResourceTransaction<D, TransactionQuarantined>, Self> {
let transaction = self
.transaction
.as_mut()
.expect("prepare recovery owns transaction");
let ledger_before = transaction.ledger_snapshot_entries();
match transaction.quarantine_live() {
Ok(records) => {
self.failure.recovery_complete = true;
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.clear_pending_failure();
transaction.recovery_history.push(self.failure.clone());
let receipt = transaction
.transition_receipt(
ResourceTransactionAction::Quarantine,
ledger_before,
records,
)
.expect("core quarantine journal must validate");
let transaction = self
.transaction
.take()
.expect("prepare recovery owns transaction");
Ok(transaction.advance(Some(receipt)))
}
Err(failure) => {
self.failure.record_recovery_failure(failure, None);
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.set_pending_failure(&self.failure);
Err(self)
}
}
}
}
#[must_use = "poisoned commit ownership must be reconciled or quarantined"]
pub struct ResourcePoisonedTransaction<D: ResourceTransactionDriver> {
transaction: Option<ResourceTransaction<D, TransactionReserved>>,
failure: ResourceFailureReceipt,
poisoned_order: usize,
}
impl<D: ResourceTransactionDriver> fmt::Debug for ResourcePoisonedTransaction<D> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ResourcePoisonedTransaction")
.field("failure", &self.failure)
.field("poisoned_order", &self.poisoned_order)
.finish_non_exhaustive()
}
}
impl<D: ResourceTransactionDriver> ResourcePoisonedTransaction<D> {
pub fn failure(&self) -> &ResourceFailureReceipt {
&self.failure
}
pub fn reconcile(
mut self,
) -> Result<ResourcePrepareTransitionError<D, TransactionReserved>, Self> {
let transaction = self
.transaction
.as_mut()
.expect("poisoned owner owns transaction");
let reservation = transaction.reservations.reservations[self.poisoned_order].clone();
let result = {
let lease = transaction
.lease
.as_ref()
.expect("poisoned transaction owns lease ledger");
let slot = &lease.slots[self.poisoned_order];
let actual = ResourceCommitView {
resource_id: slot
.actual_resource_id
.as_ref()
.expect("poisoned allocation records actual resource identity"),
generation: slot
.actual_generation
.expect("poisoned allocation records actual generation"),
descriptor: slot
.descriptor
.as_ref()
.expect("poisoned allocation records actual descriptor"),
buffer: slot
.buffer
.as_ref()
.expect("poisoned allocation remains core-owned"),
};
let driver = transaction
.driver
.as_mut()
.expect("live transaction owns driver");
let context = ResourceTransactionContext {
runtime: &lease.runtime,
identity: &transaction.identity,
binding: &transaction.admission,
reservations: &transaction.reservations,
cursor: Some(ResourceActionCursor {
order: self.poisoned_order,
action: ResourceTransactionAction::Commit,
before: transaction.states[self.poisoned_order],
allocation_authorized: false,
}),
allocation_authority: None,
pending_allocation: None,
};
driver.reconcile_commit_outcome(&context, &reservation, actual)
};
match result {
Ok(()) => {
transaction
.lease
.as_mut()
.expect("poisoned transaction owns lease ledger")
.clear(self.poisoned_order);
self.failure.recovery_strategy = ResourceRecoveryStrategy::ReverseCompensation;
self.failure.ledger_after = transaction.ledger_snapshot_entries();
let transaction = self
.transaction
.take()
.expect("poisoned owner owns transaction");
Ok(ResourcePrepareTransitionError {
transaction: Some(transaction),
failure: self.failure,
})
}
Err(failure) => {
self.failure.record_recovery_failure(
failure,
Some(ResourceFailurePoint::new(
&reservation,
self.poisoned_order,
transaction.states[self.poisoned_order],
)),
);
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.set_pending_failure(&self.failure);
Err(self)
}
}
}
pub fn quarantine(mut self) -> Result<ResourceTransaction<D, TransactionQuarantined>, Self> {
let transaction = self
.transaction
.as_mut()
.expect("poisoned owner owns transaction");
let ledger_before = transaction.ledger_snapshot_entries();
match transaction.quarantine_live() {
Ok(records) => {
self.failure.recovery_complete = true;
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.clear_pending_failure();
transaction.recovery_history.push(self.failure.clone());
let receipt = transaction
.transition_receipt(
ResourceTransactionAction::Quarantine,
ledger_before,
records,
)
.expect("poison quarantine journal must validate");
let transaction = self
.transaction
.take()
.expect("poisoned owner owns transaction");
Ok(transaction.advance(Some(receipt)))
}
Err(failure) => {
self.failure.record_recovery_failure(failure, None);
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.set_pending_failure(&self.failure);
Err(self)
}
}
}
}
#[must_use = "rollback forward completion owns the transaction"]
pub struct ResourceRollbackTransitionError<D: ResourceTransactionDriver> {
transaction: Option<ResourceTransaction<D, TransactionReserved>>,
failure: ResourceFailureReceipt,
}
impl<D: ResourceTransactionDriver> fmt::Debug for ResourceRollbackTransitionError<D> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ResourceRollbackTransitionError")
.field("failure", &self.failure)
.finish_non_exhaustive()
}
}
impl<D: ResourceTransactionDriver> ResourceRollbackTransitionError<D> {
pub fn failure(&self) -> &ResourceFailureReceipt {
&self.failure
}
pub fn complete(mut self) -> Result<ResourceTransaction<D, TransactionRolledBack>, Self> {
let transaction = self
.transaction
.as_mut()
.expect("rollback recovery owns transaction");
for order in 0..transaction.states.len() {
match transaction.states[order] {
ResourceTransactionState::RolledBack => continue,
ResourceTransactionState::Reserved => {}
actual => {
let failure = ResourceDriverFailure::new(core_resource_failure(
"rollback_ledger_diverged",
format!("rollback found actual state {}", actual.as_str()),
false,
))
.expect("core failure has resource domain");
self.failure.record_recovery_failure(
failure,
Some(ResourceFailurePoint::new(
&transaction.reservations.reservations[order],
order,
actual,
)),
);
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.set_pending_failure(&self.failure);
return Err(self);
}
}
let result = {
let reservation = transaction.reservations.reservations[order].clone();
let driver = transaction
.driver
.as_mut()
.expect("live transaction owns driver");
let context = ResourceTransactionContext {
runtime: &transaction
.lease
.as_ref()
.expect("transaction owns lease")
.runtime,
identity: &transaction.identity,
binding: &transaction.admission,
reservations: &transaction.reservations,
cursor: Some(ResourceActionCursor {
order,
action: ResourceTransactionAction::Rollback,
before: transaction.states[order],
allocation_authorized: false,
}),
allocation_authority: None,
pending_allocation: None,
};
driver.rollback_resource(&context, &reservation)
};
match result {
Ok(()) => {
let before = transaction.states[order];
transaction.states[order] = ResourceTransactionState::RolledBack;
self.failure.completed.push(transaction.make_record(
order,
ResourceTransactionAction::Rollback,
before,
ResourceTransactionState::RolledBack,
));
}
Err(failure) => {
self.failure.record_recovery_failure(
failure,
Some(ResourceFailurePoint::new(
&transaction.reservations.reservations[order],
order,
transaction.states[order],
)),
);
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.set_pending_failure(&self.failure);
return Err(self);
}
}
}
self.failure.recovery_complete = true;
self.failure.ledger_after = transaction.ledger_snapshot_entries();
let receipt = transaction
.transition_receipt(
ResourceTransactionAction::Rollback,
self.failure.ledger_before.clone(),
self.failure.completed.clone(),
)
.expect("forward-completed rollback journal must validate");
transaction.clear_pending_failure();
transaction.recovery_history.push(self.failure);
let transaction = self
.transaction
.take()
.expect("rollback recovery owns transaction");
Ok(transaction.advance(Some(receipt)))
}
pub fn quarantine(mut self) -> Result<ResourceTransaction<D, TransactionQuarantined>, Self> {
let transaction = self
.transaction
.as_mut()
.expect("rollback recovery owns transaction");
let ledger_before = transaction.ledger_snapshot_entries();
match transaction.quarantine_live() {
Ok(records) => {
self.failure.recovery_complete = true;
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.clear_pending_failure();
transaction.recovery_history.push(self.failure.clone());
let receipt = transaction
.transition_receipt(
ResourceTransactionAction::Quarantine,
ledger_before,
records,
)
.expect("rollback quarantine journal must validate");
let transaction = self
.transaction
.take()
.expect("rollback recovery owns transaction");
Ok(transaction.advance(Some(receipt)))
}
Err(failure) => {
self.failure.record_recovery_failure(failure, None);
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.set_pending_failure(&self.failure);
Err(self)
}
}
}
}
#[must_use = "release forward completion owns the transaction"]
pub struct ResourceReleaseTransitionError<D: ResourceTransactionDriver> {
transaction: Option<ResourceTransaction<D, TransactionCommitted>>,
failure: ResourceFailureReceipt,
}
impl<D: ResourceTransactionDriver> fmt::Debug for ResourceReleaseTransitionError<D> {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter
.debug_struct("ResourceReleaseTransitionError")
.field("failure", &self.failure)
.finish_non_exhaustive()
}
}
impl<D: ResourceTransactionDriver> ResourceReleaseTransitionError<D> {
pub fn failure(&self) -> &ResourceFailureReceipt {
&self.failure
}
pub fn complete(mut self) -> Result<ResourceTransaction<D, TransactionReleased>, Self> {
let transaction = self
.transaction
.as_mut()
.expect("release recovery owns transaction");
match transaction.complete_pending_release() {
Ok(_) => {
self.failure = transaction
.recovery_history
.last()
.cloned()
.unwrap_or_else(|| self.failure.clone());
let transaction = self
.transaction
.take()
.expect("release recovery owns transaction");
transaction.release().map_err(|next| Self {
transaction: next.transaction,
failure: next.failure,
})
}
Err(failure) => {
self.failure = failure;
Err(self)
}
}
}
pub fn quarantine(mut self) -> Result<ResourceTransaction<D, TransactionQuarantined>, Self> {
let transaction = self
.transaction
.as_mut()
.expect("release recovery owns transaction");
let ledger_before = transaction.ledger_snapshot_entries();
match transaction.quarantine_live() {
Ok(records) => {
self.failure.recovery_complete = true;
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.clear_pending_failure();
transaction.pending_subset_release = None;
transaction.recovery_history.push(self.failure.clone());
let receipt = transaction
.transition_receipt(
ResourceTransactionAction::Quarantine,
ledger_before,
records,
)
.expect("release quarantine journal must validate");
let transaction = self
.transaction
.take()
.expect("release recovery owns transaction");
Ok(transaction.advance(Some(receipt)))
}
Err(failure) => {
self.failure.record_recovery_failure(failure, None);
self.failure.ledger_after = transaction.ledger_snapshot_entries();
transaction.set_pending_failure(&self.failure);
Err(self)
}
}
}
}