use liminal_protocol::lifecycle::{
AggregateOperationDecision, AttachCommit, AttachCommitParameters, AttachFrontierCharges,
AttachSecretProof, AttachTransition, AttachedRecordPosition, BindingSlotDecision, BindingState,
ClosureState, CommittedBindingTerminalPosition, CredentialAttachLiveReceipt,
CredentialAttachLookupResult, LiveFrontierOwner, PresentedIdentity, RetainedRecordCharge,
SemanticConnectionCapacityDecision, apply_attach_frontier, commit_attach,
decide_attached_operation, lookup_credential_attach, select_credential_attach_binding_slot,
};
use liminal_protocol::wire::{
AttachBound, AttachEnvelope, AttachSecret, BindingEpoch, CredentialAttachRequest,
CredentialAttachResponse, Generation, ReceiptExpiryReason,
};
use super::barrier::{ArmOutcome, CommitMode, OperationFacts, commit_through_barrier};
use super::capacity::ServerCapacity;
use super::facts::{self, Digest};
use super::frontier;
use super::log::{StoredAttachAllocation, StoredAttachRequest, StoredOperation};
use super::ops_attach_capacity::AttachStage8;
use super::ops_attach_lookup::{credential_attach_refusal, marker_bearing_attach_refusal};
use super::state::{
AttachProvenanceRecord, AttachReceiptState, ConversationAuthority, DurableAppend, StateError,
};
impl ConversationAuthority {
pub(super) fn apply_credential_attach(
&mut self,
request: &CredentialAttachRequest,
operation_facts: &OperationFacts,
server_capacity: &ServerCapacity,
appender: &dyn DurableAppend,
) -> Result<ArmOutcome, StateError> {
let envelope = attach_envelope(request);
let now = u128::from(operation_facts.now_ms);
self.prune_expired_provenance(now);
let Some(slot) = self.slots.get(&request.participant_id) else {
return Ok(ArmOutcome::respond(
CredentialAttachResponse::participant_unknown(envelope).into_server_value(),
));
};
let (token_phase, secret_proof) = slot.attach_token_phase(request, now);
let lookup = lookup_credential_attach(
token_phase,
PresentedIdentity::Live(&slot.member),
&slot.binding,
request,
secret_proof,
);
if !matches!(lookup, CredentialAttachLookupResult::AuthorizedFresh { .. }) {
return credential_attach_refusal(&lookup, envelope, slot).map(ArmOutcome::respond);
}
let capacity = match operation_facts.semantic_connection_capacity() {
SemanticConnectionCapacityDecision::Commit(value) => value,
SemanticConnectionCapacityDecision::Respond { limit } => {
return Ok(ArmOutcome::respond(
CredentialAttachResponse::connection_conversation_capacity_exceeded(
envelope, limit,
)
.into_server_value(),
));
}
};
if let BindingSlotDecision::Respond(response) = select_credential_attach_binding_slot(
request,
self.binding_slot_occupancy(operation_facts.receiving_incarnation),
) {
return Ok(ArmOutcome::respond(response.into_server_value()));
}
if request.accept_marker_delivery_seq.is_some() {
return marker_bearing_attach_refusal(request, slot, operation_facts)
.map(ArmOutcome::respond);
}
let deadlines = operation_facts.deadlines()?;
let (reservation, retire) = match self.attach_stage8(
request,
slot,
operation_facts,
server_capacity,
&deadlines,
)? {
AttachStage8::Refused(response) => {
return Ok(ArmOutcome::respond(response.into_server_value()));
}
AttachStage8::Reserved {
reservation,
retire,
} => (reservation, retire),
};
let superseding = match &slot.binding {
BindingState::Detached => false,
BindingState::Bound(_) => true,
BindingState::PendingFinalization(_) => {
return Err(StateError::invariant(
"pending finalization observed in a binding that commits detaches immediately",
));
}
};
let next_generation = request
.capability_generation
.get()
.checked_add(1)
.and_then(Generation::new)
.ok_or(StateError::AllocationExhausted {
domain: "capability generation",
})?;
let (attached_order, superseded_terminal_seq, attached_seq) = if superseding {
let (order, terminal_seq, attached_seq) = self.allocate_supersession_position()?;
(order, Some(terminal_seq), attached_seq)
} else {
let (order, seq) = self.allocate_position()?;
(order, None, seq)
};
let allocation = StoredAttachAllocation {
binding_epoch: BindingEpoch::new(
operation_facts.receiving_incarnation,
next_generation,
)
.into(),
attach_secret: facts::mint_secret_bytes()?,
attached_order,
attached_seq,
receipt_expires_at: deadlines.receipt_expires_at().into(),
provenance_expires_at: deadlines.provenance_expires_at().into(),
admitted_now_ms: operation_facts.now_ms,
superseded_terminal_seq,
};
let outcome = self.attach_commit(request, &allocation, CommitMode::Live(appender))?;
reservation.confirm(&retire);
Ok(ArmOutcome::committed(
CredentialAttachResponse::attach_bound(outcome).into_server_value(),
capacity,
))
}
pub(super) fn replay_attached(
&mut self,
request: StoredAttachRequest,
allocation: &StoredAttachAllocation,
stored_event: &[u8],
sequence: u64,
) -> Result<(), StateError> {
let request = request.to_request()?;
self.attach_commit(
&request,
allocation,
CommitMode::Replay {
stored_event,
sequence,
},
)?;
Ok(())
}
fn attach_commit(
&mut self,
request: &CredentialAttachRequest,
allocation: &StoredAttachAllocation,
mode: CommitMode<'_>,
) -> Result<AttachBound, StateError> {
let (participant_id, mut slot) = self
.slots
.remove_entry(&request.participant_id)
.ok_or_else(|| {
StateError::invariant("attach commit requires an enrolled participant slot")
})?;
let binding_epoch = allocation.binding_epoch.to_epoch()?;
let result_generation = binding_epoch.capability_generation;
let parameters = AttachCommitParameters {
binding: liminal_protocol::lifecycle::ActiveBinding {
participant_id: request.participant_id,
conversation_id: request.conversation_id,
binding_epoch,
},
attach_secret: AttachSecret::new(allocation.attach_secret),
attached_position: AttachedRecordPosition::new(
allocation.attached_order,
allocation.attached_seq,
),
receipt_expires_at: allocation.receipt_expires_at.get(),
provenance_expires_at: allocation.provenance_expires_at.get(),
};
let verified =
verify_attach_mode(slot.member, slot.binding, request, allocation, parameters)?;
let committed = commit_attach(verified, slot.cell).map_err(|error| {
StateError::invariant(format!("protocol attach transition failed: {error:?}"))
})?;
let observer_projection = committed.observer_progress_projection();
let (committed, frontier_owner) =
transition_attach_frontier(self.take_frontier()?, committed, request, allocation)?;
let shell = self.take_shell()?;
let barrier = match decide_attached_operation(shell, committed) {
AggregateOperationDecision::Commit(barrier) => barrier,
AggregateOperationDecision::Refused(refusal) => {
return Err(StateError::ShellRefused {
reason: refusal.reason(),
});
}
};
let make_operation = |event: Vec<u8>| StoredOperation::Attached {
request: request.into(),
secret_verified: true,
allocation: *allocation,
event,
};
let (shell, committed) =
commit_through_barrier(barrier, mode, self.next_log_sequence, &make_operation)?;
self.shell = Some(shell);
self.install_frontier(frontier_owner);
self.advance_log_head()?;
let outcome = committed.outcome.clone();
slot.member = committed.member;
slot.binding = committed.binding_state;
slot.cell = committed.detach_cell;
slot.attach_secret = AttachSecret::new(allocation.attach_secret);
if let Some(previous) = slot.attach.take() {
let reason = if u128::from(allocation.admitted_now_ms) < previous.receipt_expires_at {
ReceiptExpiryReason::Superseded
} else {
ReceiptExpiryReason::Deadline
};
slot.attach_provenance.insert(
previous.token.into_bytes(),
AttachProvenanceRecord {
result_generation: previous.result_generation,
reason,
provenance_expires_at: previous.provenance_expires_at,
},
);
}
if slot.enrollment_receipt_ended.is_none() {
slot.enrollment_receipt_ended = Some(
if u128::from(allocation.admitted_now_ms) < slot.enrollment_receipt_expires_at {
ReceiptExpiryReason::Superseded
} else {
ReceiptExpiryReason::Deadline
},
);
}
slot.attach = Some(AttachReceiptState {
token: request.attach_attempt_token,
receipt: CredentialAttachLiveReceipt::from_commit(outcome.clone()),
outcome: committed.outcome,
verifier: request.attach_secret.into_bytes(),
result_generation,
receipt_expires_at: allocation.receipt_expires_at.get(),
provenance_expires_at: allocation.provenance_expires_at.get(),
});
self.slots.insert(participant_id, slot);
if let Some(projection) = observer_projection {
self.record_observer_progress_projection(projection);
}
self.observe_replayed_position(allocation.attached_order, allocation.attached_seq);
Ok(outcome)
}
}
fn transition_attach_frontier(
owner: LiveFrontierOwner,
committed: AttachCommit<Digest, Digest>,
request: &CredentialAttachRequest,
allocation: &StoredAttachAllocation,
) -> Result<(AttachCommit<Digest, Digest>, LiveFrontierOwner), StateError> {
let attached_encoded = frontier::credential_attached_charge(
request.conversation_id,
request.participant_id,
allocation,
)?;
let attached_charge = RetainedRecordCharge::new(
committed.attached.delivery_seq(),
committed.attached.admission_order(),
attached_encoded,
);
let terminal = match committed.transition {
AttachTransition::Detached => None,
AttachTransition::Superseded { terminal } => Some(terminal.into()),
AttachTransition::FencedRecovery {
composed_terminal, ..
} => composed_terminal,
};
let terminal_charge = terminal
.map(|terminal| {
frontier::terminal_charge(
terminal.conversation_id(),
terminal.participant_id(),
terminal.binding_epoch(),
terminal.admission_order().transaction_order(),
terminal.delivery_seq(),
)
.map(|encoded| {
RetainedRecordCharge::new(
terminal.delivery_seq(),
terminal.admission_order(),
encoded,
)
})
})
.transpose()?;
apply_attach_frontier(
owner,
committed,
AttachFrontierCharges::new(terminal_charge, attached_charge),
)
.map_err(|failure| {
StateError::invariant(format!(
"attach frontier transition failed: {:?}",
failure.error()
))
})
.map(liminal_protocol::lifecycle::LiveFrontierCommit::into_parts)
}
pub(super) const fn attach_envelope(request: &CredentialAttachRequest) -> AttachEnvelope {
AttachEnvelope {
conversation_id: request.conversation_id,
participant_id: request.participant_id,
capability_generation: request.capability_generation,
attach_attempt_token: request.attach_attempt_token,
accept_marker_delivery_seq: request.accept_marker_delivery_seq,
}
}
fn verify_attach_mode(
member: liminal_protocol::lifecycle::LiveMember<Digest>,
binding: BindingState,
request: &CredentialAttachRequest,
allocation: &StoredAttachAllocation,
parameters: AttachCommitParameters,
) -> Result<liminal_protocol::lifecycle::VerifiedAttachCommit<'static, Digest>, StateError> {
match (binding, allocation.superseded_terminal_seq) {
(BindingState::Detached, None) => {
let closure_admission = ClosureState::Clear
.ordinary_detached_attach_admission()
.map_err(|error| {
StateError::invariant(format!(
"clear closure refused detached attach admission: {error:?}"
))
})?;
member.verify_detached_attach(
BindingState::Detached,
closure_admission,
request.clone(),
AttachSecretProof::Verified,
parameters,
)
}
(BindingState::Bound(active), Some(terminal_seq)) => member.verify_superseding_attach(
active,
request.clone(),
AttachSecretProof::Verified,
CommittedBindingTerminalPosition::new(allocation.attached_order, terminal_seq),
parameters,
),
(_, _) => {
return Err(StateError::invariant(
"attach allocation mode does not match the slot's binding authority",
));
}
}
.map_err(|error| {
StateError::invariant(format!("protocol attach verification failed: {error:?}"))
})
}