chio_settle/channel/projection/
dispatch.rs

1use chio_core::economic_continuity::{
2    economic_effect_slot_from_head, EconomicEffectSlotV1, EconomicEffectStateV1,
3    EconomicEffectTargetV1, EconomicRequestBindingV1, EconomicRequestReplayV1,
4    EconomicTransitionProofVerifier, VerifiedEconomicStateBatchAdvance, VerifiedEconomicStateView,
5};
6use serde::Serialize;
7
8use super::super::validation::digest;
9use super::super::{
10    derive_channel_service_dispatch_idempotency_key, verify_channel_lifecycle_snapshot,
11    ChannelError, VerifiedAdmittedChannelReservationV1, CHANNEL_SERVICE_DISPATCH_EFFECT_KIND,
12};
13use super::{
14    head_digest, successor_head, transition, ChannelLifecycleBatchVerifier,
15    ChannelLifecycleProjectionV1, SuccessorHeadBinding,
16};
17
18const CHANNEL_DISPATCH_TRANSITION_PROOF_SCHEMA: &str = "chio.channel.dispatch-transition-proof.v1";
19const CHANNEL_DISPATCH_TRANSITION_PROOF_DOMAIN: &[u8] =
20    b"chio.channel.dispatch-transition-proof.digest.v1\0";
21
22#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
23#[serde(rename_all = "camelCase")]
24struct ChannelDispatchTransitionProofV1 {
25    schema: String,
26    reservation_proposal_digest: String,
27    reservation_digest: String,
28    operation_id: String,
29    request: EconomicRequestBindingV1,
30    request_replay: EconomicRequestReplayV1,
31    provider: EconomicEffectTargetV1,
32    source_checkpoint_sequence: u64,
33    source_checkpoint_digest: String,
34    ready_effect_digest: String,
35    ready_effect_head_digest: String,
36    dispatch_effect_digest: String,
37    dispatch_effect_head_digest: String,
38    issued_at: u64,
39}
40
41impl ChannelDispatchTransitionProofV1 {
42    fn digest(&self) -> Result<String, ChannelError> {
43        digest(CHANNEL_DISPATCH_TRANSITION_PROOF_DOMAIN, self)
44    }
45}
46
47pub fn compose_channel_dispatch_transition(
48    reservation: &VerifiedAdmittedChannelReservationV1,
49    current: &VerifiedEconomicStateView,
50    issued_at: u64,
51) -> Result<ChannelLifecycleProjectionV1, ChannelError> {
52    let body = &reservation.artifact().body;
53    let ready_effect = reservation.ready_effect();
54    let admitted = reservation.snapshot();
55    let snapshot = verify_channel_lifecycle_snapshot(
56        current,
57        admitted.settlement_authority_scope_id(),
58        &body.channel_id,
59    )?;
60    let effect_key = ready_effect.resource_head_key();
61    let ready_head = current
62        .view()
63        .head(&effect_key)
64        .ok_or(ChannelError::AuthorityVerification)?;
65    let retained_effect = economic_effect_slot_from_head(ready_head)
66        .map_err(|_| ChannelError::AuthorityVerification)?;
67    let reservation_digest = reservation.artifact().digest()?;
68    let expected_idempotency_key = derive_channel_service_dispatch_idempotency_key(
69        &body.operation_id,
70        &body.reservation_id,
71        body.next_sequence,
72    )?;
73    let request_replay = EconomicRequestReplayV1 {
74        request: ready_effect.request.clone(),
75        operation_id: body.operation_id.clone(),
76        effect_slot_ids: vec![ready_effect.slot_id.clone()],
77    };
78    request_replay
79        .validate()
80        .map_err(|_| ChannelError::AuthorityVerification)?;
81    let retained_replay = current
82        .view()
83        .request_replay(&ready_effect.request.key())
84        .ok_or(ChannelError::AuthorityVerification)?;
85    let ready_head_digest = head_digest(ready_head)?;
86    let exact_checkpoint = current.view().checkpoint_sequence == admitted.checkpoint_sequence()
87        && current.view().checkpoint_digest == admitted.checkpoint_digest();
88    let later_checkpoint = current.view().checkpoint_sequence > admitted.checkpoint_sequence()
89        && current.view().checkpoint_digest != admitted.checkpoint_digest();
90    if !(exact_checkpoint || later_checkpoint)
91        || current.view().observed_at < admitted.observed_at_unix_ms()
92        || snapshot.lifecycle() != admitted.lifecycle()
93        || snapshot.escrow() != admitted.escrow()
94        || snapshot.channel_head_digest() != admitted.channel_head_digest()
95        || snapshot.escrow_head_digest() != admitted.escrow_head_digest()
96        || snapshot.channel_head() != admitted.channel_head()
97        || snapshot.escrow_head() != admitted.escrow_head()
98        || &retained_effect != ready_effect
99        || retained_replay != &request_replay
100        || ready_head_digest != reservation.ready_effect_head_digest()
101        || ready_effect.operation_id != body.operation_id
102        || ready_effect.request.request_id != body.request_id
103        || ready_effect.effect_kind != CHANNEL_SERVICE_DISPATCH_EFFECT_KIND
104        || ready_effect.parameters_digest != reservation_digest
105        || ready_effect.idempotency_key != expected_idempotency_key
106        || ready_effect.state != EconomicEffectStateV1::Ready
107        || ready_effect.terminal.is_some()
108        || ready_effect.frost.is_some()
109        || ready_head.head_version != 1
110        || ready_head.resource_version != 1
111        || ready_head.lifecycle_fence != 1
112        || ready_head.lifecycle_state != "ready"
113        || ready_head.operation_id.as_deref() != Some(body.operation_id.as_str())
114        || ready_head.effect_idempotency_key.as_deref() != Some(expected_idempotency_key.as_str())
115        || ready_head.frost.is_some()
116        || ready_head.terminal_result.is_some()
117        || ready_head.trusted_clock_high_water != admitted.observed_at_unix_ms()
118        || ready_head.trusted_clock_high_water > current.view().observed_at
119        || ready_head.predecessor_digest.is_some()
120        || issued_at < reservation.accepted_at_unix_ms()
121        || issued_at < current.view().observed_at
122        || issued_at >= body.expires_at_unix_ms
123    {
124        return Err(ChannelError::AuthorityVerification);
125    }
126    let mut dispatch_effect = ready_effect.clone();
127    dispatch_effect.state = EconomicEffectStateV1::DispatchCommitted;
128    ready_effect
129        .validate_successor(&dispatch_effect)
130        .map_err(|_| ChannelError::AuthorityVerification)?;
131    let dispatch_head = successor_head(
132        ready_head,
133        &dispatch_effect,
134        SuccessorHeadBinding {
135            resource_version: 2,
136            lifecycle_fence: 2,
137            lifecycle_state: "dispatch_committed",
138            operation_id: Some(body.operation_id.clone()),
139            effect_idempotency_key: Some(expected_idempotency_key),
140            terminal_result: None,
141        },
142        issued_at,
143    )?;
144    let proof = ChannelDispatchTransitionProofV1 {
145        schema: CHANNEL_DISPATCH_TRANSITION_PROOF_SCHEMA.to_owned(),
146        reservation_proposal_digest: body.proposal_digest()?,
147        reservation_digest,
148        operation_id: body.operation_id.clone(),
149        request: ready_effect.request.clone(),
150        request_replay,
151        provider: ready_effect.target.clone(),
152        source_checkpoint_sequence: current.view().checkpoint_sequence,
153        source_checkpoint_digest: current.view().checkpoint_digest.clone(),
154        ready_effect_digest: ready_effect
155            .digest()
156            .map_err(|_| ChannelError::AuthorityVerification)?,
157        ready_effect_head_digest: ready_head_digest.clone(),
158        dispatch_effect_digest: dispatch_effect
159            .digest()
160            .map_err(|_| ChannelError::AuthorityVerification)?,
161        dispatch_effect_head_digest: head_digest(&dispatch_head)?,
162        issued_at,
163    };
164    let proof_digest = proof.digest()?;
165    Ok(ChannelLifecycleProjectionV1 {
166        current: current.clone(),
167        proof_digest: proof_digest.clone(),
168        transitions: vec![transition(
169            effect_key,
170            ready_head_digest,
171            dispatch_head,
172            &proof_digest,
173        )],
174        effect_slots: Vec::new(),
175        request_replays: Vec::new(),
176        operation_id: body.operation_id.clone(),
177        issued_at,
178        not_after_unix_ms: Some(body.expires_at_unix_ms),
179    })
180}
181
182#[derive(Debug, Clone)]
183pub struct VerifiedChannelDispatchAdvanceV1 {
184    effect_slot: EconomicEffectSlotV1,
185    request_replay: EconomicRequestReplayV1,
186    reservation_digest: String,
187    previous_checkpoint_digest: String,
188}
189
190impl VerifiedChannelDispatchAdvanceV1 {
191    #[must_use]
192    pub const fn effect_slot(&self) -> &EconomicEffectSlotV1 {
193        &self.effect_slot
194    }
195
196    #[must_use]
197    pub const fn request_replay(&self) -> &EconomicRequestReplayV1 {
198        &self.request_replay
199    }
200
201    #[must_use]
202    pub fn reservation_digest(&self) -> &str {
203        &self.reservation_digest
204    }
205
206    #[must_use]
207    pub fn previous_checkpoint_digest(&self) -> &str {
208        &self.previous_checkpoint_digest
209    }
210}
211
212pub fn verify_channel_dispatch_advance(
213    reservation: &VerifiedAdmittedChannelReservationV1,
214    advance: &VerifiedEconomicStateBatchAdvance,
215) -> Result<VerifiedChannelDispatchAdvanceV1, ChannelError> {
216    let projection = compose_channel_dispatch_transition(
217        reservation,
218        advance.current(),
219        advance.batch().issued_at,
220    )?;
221    let verifier = ChannelLifecycleBatchVerifier::new(projection);
222    verifier
223        .verify_batch(advance.current(), advance.batch())
224        .map_err(|_| ChannelError::AuthorityVerification)?;
225    let transition = advance
226        .batch()
227        .transitions
228        .first()
229        .ok_or(ChannelError::AuthorityVerification)?;
230    let effect_slot = economic_effect_slot_from_head(&transition.next_head)
231        .map_err(|_| ChannelError::AuthorityVerification)?;
232    let request_replay = advance
233        .current()
234        .view()
235        .request_replay(&effect_slot.request.key())
236        .ok_or(ChannelError::AuthorityVerification)?
237        .clone();
238    Ok(VerifiedChannelDispatchAdvanceV1 {
239        effect_slot,
240        request_replay,
241        reservation_digest: reservation.artifact().digest()?,
242        previous_checkpoint_digest: advance.current().view().checkpoint_digest.clone(),
243    })
244}
chio_settle/channel/projection/dispatch.rs

chio_settle/channel/projection/
dispatch.rs
