lodestone_client/
cluster_control.rs

1//! Dedicated current-only cluster-control client.
2//!
3//! The client has no Collection profile, API-key provider, public session, or
4//! connection-pool reuse. Each call opens one TLS connection with the isolated
5//! ALPN and sends one bounded canonical cluster-control frame.
6
7use std::error::Error;
8use std::fmt::{self, Display, Formatter};
9use std::sync::Arc;
10use std::time::Duration;
11
12use lodestone_proto::{
13    CLUSTER_CONTROL_REQUEST_FORMAT_VERSION, CLUSTER_CONTROL_SESSION_FORMAT_VERSION,
14    COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION, ClusterControlActionV3, ClusterControlCredentialKindV1,
15    ClusterControlRequestV3, ClusterControlResponseDispositionV3, ClusterControlResponseOutcomeV3,
16    ClusterControlSessionV3, CollectionAuthorityObservationRequestV1, CollectionAuthorityObservationV1,
17    CollectionRetirementOutcomeV1, CollectionRetirementRequestV1, MAX_CLUSTER_CONTROL_FRAME_BYTES,
18    decode_cluster_control_response_v3, encode_cluster_control_request_v3,
19};
20use rustls::pki_types::{CertificateDer, ServerName};
21use serde::Serialize;
22use tokio::io::{AsyncReadExt, AsyncWriteExt};
23use tokio::net::TcpStream;
24use tokio_rustls::TlsConnector;
25
26pub const CLUSTER_CONTROL_ALPN: &[u8] = b"lodestone-cluster-control-v3";
27const MAX_CLUSTER_CONTROL_CALL_MILLIS: u64 = 120_000;
28const MAX_CLUSTER_CONTROL_ENDPOINT_BYTES: usize = 512;
29
30pub struct ClusterControlCredential {
31    kind: ClusterControlCredentialKindV1,
32    bytes: Vec<u8>,
33}
34
35impl ClusterControlCredential {
36    pub fn new(kind: ClusterControlCredentialKindV1, bytes: Vec<u8>) -> Result<Self, ClusterControlClientError> {
37        if lodestone_proto::validate_cluster_control_credential_v1(&bytes).is_err() {
38            return Err(ClusterControlClientError::InvalidConfiguration);
39        }
40        Ok(Self { kind, bytes })
41    }
42
43    #[must_use]
44    pub const fn kind(&self) -> ClusterControlCredentialKindV1 {
45        self.kind
46    }
47}
48
49impl fmt::Debug for ClusterControlCredential {
50    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
51        formatter
52            .debug_struct("ClusterControlCredential")
53            .field("kind", &self.kind)
54            .field("bytes", &"[REDACTED]")
55            .finish()
56    }
57}
58
59impl Drop for ClusterControlCredential {
60    fn drop(&mut self) {
61        self.bytes.fill(0);
62        std::hint::black_box(&mut self.bytes);
63    }
64}
65
66pub struct ClusterControlClientConfig {
67    endpoint: String,
68    server_name: String,
69    trust_certificates_der: Vec<Vec<u8>>,
70    credential: ClusterControlCredential,
71    timeout: Duration,
72}
73
74impl ClusterControlClientConfig {
75    pub fn new(
76        endpoint: String,
77        server_name: String,
78        trust_certificates_der: Vec<Vec<u8>>,
79        credential: ClusterControlCredential,
80        timeout: Duration,
81    ) -> Result<Self, ClusterControlClientError> {
82        if endpoint.is_empty()
83            || endpoint.len() > MAX_CLUSTER_CONTROL_ENDPOINT_BYTES
84            || endpoint.parse::<std::net::SocketAddr>().is_err()
85            || !matches!(ServerName::try_from(server_name.clone()), Ok(ServerName::DnsName(_)))
86            || trust_certificates_der.is_empty()
87            || timeout.is_zero()
88            || timeout > Duration::from_millis(MAX_CLUSTER_CONTROL_CALL_MILLIS)
89        {
90            return Err(ClusterControlClientError::InvalidConfiguration);
91        }
92        Ok(Self {
93            endpoint,
94            server_name,
95            trust_certificates_der,
96            credential,
97            timeout,
98        })
99    }
100}
101
102pub struct ClusterControlClient {
103    config: ClusterControlClientConfig,
104    tls: Arc<rustls::ClientConfig>,
105}
106
107impl ClusterControlClient {
108    pub fn new(config: ClusterControlClientConfig) -> Result<Self, ClusterControlClientError> {
109        let tls = build_tls(&config)?;
110        Ok(Self {
111            config,
112            tls: Arc::new(tls),
113        })
114    }
115
116    pub async fn initialize_collection(
117        &self,
118        request_id: &str,
119        body: impl Serialize,
120    ) -> Result<serde_json::Value, ClusterControlClientError> {
121        self.call(ClusterControlActionV3::InitializeCollectionAuthority, request_id, body)
122            .await
123    }
124
125    pub async fn activate_collection(
126        &self,
127        request_id: &str,
128        body: impl Serialize,
129    ) -> Result<serde_json::Value, ClusterControlClientError> {
130        self.call(ClusterControlActionV3::ActivateCollectionAuthority, request_id, body)
131            .await
132    }
133
134    pub async fn resolve_collection(
135        &self,
136        request_id: &str,
137        body: impl Serialize,
138    ) -> Result<serde_json::Value, ClusterControlClientError> {
139        self.call(ClusterControlActionV3::ResolveCollectionAuthority, request_id, body)
140            .await
141    }
142
143    /// Reads the current Collection-local authority without submitting or
144    /// resolving an operation. The response's canonical semantic digest and
145    /// exact Collection identity are validated before it is returned.
146    pub async fn observe_collection_authority(
147        &self,
148        request_id: &str,
149        request: &CollectionAuthorityObservationRequestV1,
150    ) -> Result<CollectionAuthorityObservationV1, ClusterControlClientError> {
151        request.validate().map_err(|_| ClusterControlClientError::Codec)?;
152        let body = self
153            .call(ClusterControlActionV3::ObserveCollectionAuthority, request_id, request)
154            .await?;
155        decode_collection_authority_observation(body, request)
156    }
157
158    pub async fn inspect_collection_retirement(
159        &self,
160        request_id: &str,
161        body: impl Serialize,
162    ) -> Result<serde_json::Value, ClusterControlClientError> {
163        self.call(ClusterControlActionV3::InspectCollectionRetirement, request_id, body)
164            .await
165    }
166
167    pub async fn retire_collection(
168        &self,
169        request_id: &str,
170        body: impl Serialize,
171    ) -> Result<serde_json::Value, ClusterControlClientError> {
172        self.call(ClusterControlActionV3::RetireCollectionAuthority, request_id, body)
173            .await
174    }
175
176    /// Submits or advances one exact provisioner-owned Collection retirement.
177    /// Only an authoritative `NotSubmitted` outcome is safe to submit again;
178    /// `Indeterminate` retains the exact request required by the resolver.
179    pub async fn submit_collection_retirement(
180        &self,
181        request_id: &str,
182        request: &CollectionRetirementRequestV1,
183    ) -> Result<CollectionRetirementOutcomeV1, ClusterControlClientError> {
184        request.validate().map_err(|_| ClusterControlClientError::Codec)?;
185        self.call_collection_retirement(
186            ClusterControlActionV3::RetireCollectionAuthority,
187            request_id,
188            request,
189            request,
190        )
191        .await
192    }
193
194    /// Resolves one exact Collection-retirement request without inventing a
195    /// replacement operation identity after response loss or process restart.
196    pub async fn resolve_collection_retirement(
197        &self,
198        request_id: &str,
199        request: &CollectionRetirementRequestV1,
200    ) -> Result<CollectionRetirementOutcomeV1, ClusterControlClientError> {
201        request.validate().map_err(|_| ClusterControlClientError::Codec)?;
202        let body = CollectionRetirementResolutionBodyV1::Retire {
203            request: request.clone(),
204        };
205        self.call_collection_retirement(
206            ClusterControlActionV3::ResolveCollectionAuthority,
207            request_id,
208            body,
209            request,
210        )
211        .await
212    }
213
214    async fn call_collection_retirement(
215        &self,
216        action: ClusterControlActionV3,
217        request_id: &str,
218        body: impl Serialize,
219        request: &CollectionRetirementRequestV1,
220    ) -> Result<CollectionRetirementOutcomeV1, ClusterControlClientError> {
221        classify_collection_retirement_call(self.call(action, request_id, body).await, request)
222    }
223
224    pub async fn observe_capacity(&self, request_id: &str) -> Result<serde_json::Value, ClusterControlClientError> {
225        self.call(
226            ClusterControlActionV3::ObserveClusterCapacity,
227            request_id,
228            serde_json::json!({}),
229        )
230        .await
231    }
232
233    pub async fn observe_resource_diagnostics(
234        &self,
235        request_id: &str,
236        body: impl Serialize,
237    ) -> Result<serde_json::Value, ClusterControlClientError> {
238        self.call(ClusterControlActionV3::ObserveResourceDiagnostics, request_id, body)
239            .await
240    }
241
242    pub async fn reserve_credential_rotation(
243        &self,
244        request_id: &str,
245        body: impl Serialize,
246    ) -> Result<serde_json::Value, ClusterControlClientError> {
247        self.call(
248            ClusterControlActionV3::ReserveClusterControlCredentialRotation,
249            request_id,
250            body,
251        )
252        .await
253    }
254
255    pub async fn activate_credential_rotation(
256        &self,
257        request_id: &str,
258        body: impl Serialize,
259    ) -> Result<serde_json::Value, ClusterControlClientError> {
260        self.call(
261            ClusterControlActionV3::ActivateClusterControlCredentialRotation,
262            request_id,
263            body,
264        )
265        .await
266    }
267
268    pub async fn revoke_credential(
269        &self,
270        request_id: &str,
271        body: impl Serialize,
272    ) -> Result<serde_json::Value, ClusterControlClientError> {
273        self.call(ClusterControlActionV3::RevokeClusterControlCredential, request_id, body)
274            .await
275    }
276
277    pub async fn resolve_credential(
278        &self,
279        request_id: &str,
280        body: impl Serialize,
281    ) -> Result<serde_json::Value, ClusterControlClientError> {
282        self.call(
283            ClusterControlActionV3::ResolveClusterControlCredential,
284            request_id,
285            body,
286        )
287        .await
288    }
289
290    pub async fn list_backups(&self, request_id: &str) -> Result<serde_json::Value, ClusterControlClientError> {
291        self.call(
292            ClusterControlActionV3::ListClusterBackups,
293            request_id,
294            serde_json::json!({}),
295        )
296        .await
297    }
298
299    pub async fn get_backup(
300        &self,
301        request_id: &str,
302        body: impl Serialize,
303    ) -> Result<serde_json::Value, ClusterControlClientError> {
304        self.call(ClusterControlActionV3::GetClusterBackup, request_id, body)
305            .await
306    }
307
308    pub async fn create_backup(
309        &self,
310        request_id: &str,
311        body: impl Serialize,
312    ) -> Result<serde_json::Value, ClusterControlClientError> {
313        self.call(ClusterControlActionV3::CreateClusterBackup, request_id, body)
314            .await
315    }
316
317    pub async fn resolve_backup(
318        &self,
319        request_id: &str,
320        body: impl Serialize,
321    ) -> Result<serde_json::Value, ClusterControlClientError> {
322        self.call(ClusterControlActionV3::ResolveClusterBackup, request_id, body)
323            .await
324    }
325
326    pub async fn delete_backup(
327        &self,
328        request_id: &str,
329        body: impl Serialize,
330    ) -> Result<serde_json::Value, ClusterControlClientError> {
331        self.call(ClusterControlActionV3::DeleteClusterBackup, request_id, body)
332            .await
333    }
334
335    pub async fn resolve_backup_deletion(
336        &self,
337        request_id: &str,
338        body: impl Serialize,
339    ) -> Result<serde_json::Value, ClusterControlClientError> {
340        self.call(ClusterControlActionV3::ResolveClusterBackupDeletion, request_id, body)
341            .await
342    }
343
344    pub async fn restore_readiness(&self, request_id: &str) -> Result<serde_json::Value, ClusterControlClientError> {
345        self.call(
346            ClusterControlActionV3::GetClusterRestoreReadiness,
347            request_id,
348            serde_json::json!({}),
349        )
350        .await
351    }
352
353    pub async fn prepare_offline_recovery(
354        &self,
355        request_id: &str,
356        body: impl Serialize,
357    ) -> Result<serde_json::Value, ClusterControlClientError> {
358        self.call(ClusterControlActionV3::PrepareOfflineRecovery, request_id, body)
359            .await
360    }
361
362    pub async fn call(
363        &self,
364        action: ClusterControlActionV3,
365        request_id: &str,
366        body: impl Serialize,
367    ) -> Result<serde_json::Value, ClusterControlClientError> {
368        if !self.config.credential.kind.admits(action) {
369            return Err(ClusterControlClientError::ActionDenied);
370        }
371        let body = serde_json::to_value(body).map_err(|_| ClusterControlClientError::Codec)?;
372        if !body.is_object() {
373            return Err(ClusterControlClientError::Codec);
374        }
375        let mut correlation = [0_u8; 16];
376        getrandom::fill(&mut correlation).map_err(|_| ClusterControlClientError::RandomUnavailable)?;
377        let mut request = ClusterControlRequestV3 {
378            format_version: CLUSTER_CONTROL_REQUEST_FORMAT_VERSION,
379            session: ClusterControlSessionV3 {
380                format_version: CLUSTER_CONTROL_SESSION_FORMAT_VERSION,
381                credential_kind: self.config.credential.kind,
382                credential: self.config.credential.bytes.clone(),
383                request_id: request_id.to_owned(),
384                action,
385                request_correlation_id: correlation,
386            },
387            body,
388        };
389        let encoded = encode_cluster_control_request_v3(&request);
390        request.session.credential.fill(0);
391        std::hint::black_box(&mut request.session.credential);
392        let mut payload = encoded.map_err(|_| ClusterControlClientError::Codec)?;
393        let result = tokio::time::timeout(self.config.timeout, self.call_once(&payload, request_id, correlation)).await;
394        payload.fill(0);
395        std::hint::black_box(&mut payload);
396        result.map_err(|_| ClusterControlClientError::DeadlineExceeded)?
397    }
398
399    async fn call_once(
400        &self,
401        payload: &[u8],
402        request_id: &str,
403        correlation: [u8; 16],
404    ) -> Result<serde_json::Value, ClusterControlClientError> {
405        let tcp = TcpStream::connect(&self.config.endpoint)
406            .await
407            .map_err(|_| ClusterControlClientError::Transport)?;
408        let server_name = ServerName::try_from(self.config.server_name.clone())
409            .map_err(|_| ClusterControlClientError::InvalidConfiguration)?;
410        let connector = TlsConnector::from(Arc::clone(&self.tls));
411        let mut tls = connector
412            .connect(server_name, tcp)
413            .await
414            .map_err(|_| ClusterControlClientError::Transport)?;
415        let length = u32::try_from(payload.len()).map_err(|_| ClusterControlClientError::Codec)?;
416        tls.write_u32(length)
417            .await
418            .map_err(|_| ClusterControlClientError::Transport)?;
419        tls.write_all(payload)
420            .await
421            .map_err(|_| ClusterControlClientError::Transport)?;
422        let length = usize::try_from(tls.read_u32().await.map_err(|_| ClusterControlClientError::Transport)?)
423            .map_err(|_| ClusterControlClientError::Codec)?;
424        if length == 0 || length > MAX_CLUSTER_CONTROL_FRAME_BYTES {
425            return Err(ClusterControlClientError::Codec);
426        }
427        let mut response = vec![0; length];
428        tls.read_exact(&mut response)
429            .await
430            .map_err(|_| ClusterControlClientError::Transport)?;
431        let response = decode_cluster_control_response_v3(&response).map_err(|_| ClusterControlClientError::Codec)?;
432        if response.request_id != request_id || response.request_correlation_id != correlation {
433            return Err(ClusterControlClientError::ResponseIdentityMismatch);
434        }
435        match response.outcome {
436            ClusterControlResponseOutcomeV3::Succeeded { body } => Ok(body),
437            ClusterControlResponseOutcomeV3::Refused { code, disposition } => {
438                Err(ClusterControlClientError::Refused(ClusterControlRefusal {
439                    code,
440                    disposition,
441                }))
442            }
443        }
444    }
445}
446
447fn decode_collection_authority_observation(
448    body: serde_json::Value,
449    request: &CollectionAuthorityObservationRequestV1,
450) -> Result<CollectionAuthorityObservationV1, ClusterControlClientError> {
451    let observation: CollectionAuthorityObservationV1 =
452        serde_json::from_value(body).map_err(|_| ClusterControlClientError::Codec)?;
453    observation.validate().map_err(|_| ClusterControlClientError::Codec)?;
454    if observation.collection_id != request.collection_id {
455        return Err(ClusterControlClientError::ResponseIdentityMismatch);
456    }
457    Ok(observation)
458}
459
460fn classify_collection_retirement_call(
461    result: Result<serde_json::Value, ClusterControlClientError>,
462    request: &CollectionRetirementRequestV1,
463) -> Result<CollectionRetirementOutcomeV1, ClusterControlClientError> {
464    match result {
465        Ok(body) => match serde_json::from_value::<CollectionRetirementOutcomeV1>(body) {
466            Ok(outcome)
467                if outcome.validate().is_ok() && collection_retirement_outcome_matches_request(&outcome, request)? =>
468            {
469                Ok(outcome)
470            }
471            Ok(_) | Err(_) => Ok(CollectionRetirementOutcomeV1::Indeterminate {
472                format_version: COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION,
473                request: Box::new(request.clone()),
474            }),
475        },
476        Err(ClusterControlClientError::Refused(refusal))
477            if refusal.disposition == ClusterControlResponseDispositionV3::RetrySameRequest =>
478        {
479            Ok(CollectionRetirementOutcomeV1::NotSubmitted {
480                format_version: COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION,
481                request_digest: request
482                    .canonical_request_digest()
483                    .map_err(|_| ClusterControlClientError::Codec)?,
484            })
485        }
486        Err(ClusterControlClientError::Transport)
487        | Err(ClusterControlClientError::DeadlineExceeded)
488        | Err(ClusterControlClientError::ResponseIdentityMismatch)
489        | Err(ClusterControlClientError::Codec)
490        | Err(ClusterControlClientError::Refused(ClusterControlRefusal {
491            disposition: ClusterControlResponseDispositionV3::ResolveOperation,
492            ..
493        })) => Ok(CollectionRetirementOutcomeV1::Indeterminate {
494            format_version: COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION,
495            request: Box::new(request.clone()),
496        }),
497        Err(error) => Err(error),
498    }
499}
500
501fn collection_retirement_outcome_matches_request(
502    outcome: &CollectionRetirementOutcomeV1,
503    request: &CollectionRetirementRequestV1,
504) -> Result<bool, ClusterControlClientError> {
505    let request_digest = request
506        .canonical_request_digest()
507        .map_err(|_| ClusterControlClientError::Codec)?;
508    let admitted_generation = request.expected_security_generation.get().checked_add(1);
509    Ok(match outcome {
510        CollectionRetirementOutcomeV1::NotSubmitted {
511            request_digest: observed,
512            ..
513        } => *observed == request_digest,
514        CollectionRetirementOutcomeV1::Blocked {
515            request_digest: observed,
516            admitted_security_generation,
517            ..
518        }
519        | CollectionRetirementOutcomeV1::InProgress {
520            request_digest: observed,
521            admitted_security_generation,
522            ..
523        } => *observed == request_digest && admitted_generation == Some(admitted_security_generation.get()),
524        CollectionRetirementOutcomeV1::Indeterminate { request: observed, .. } => **observed == *request,
525        CollectionRetirementOutcomeV1::Resolved { receipt, .. } => {
526            receipt.operation_id == request.operation_id
527                && receipt.request_digest == request_digest
528                && receipt.collection_id == request.collection_id
529                && admitted_generation == Some(receipt.admitted_security_generation.get())
530        }
531        CollectionRetirementOutcomeV1::KnownRetired {
532            request_digest: observed,
533            collection_id,
534            ..
535        } => *observed == request_digest && *collection_id == request.collection_id,
536        CollectionRetirementOutcomeV1::ExpiredUnknown {
537            request_digest: observed,
538            collection_id,
539            operation_id,
540            ..
541        } => {
542            *observed == request_digest
543                && *collection_id == request.collection_id
544                && *operation_id == request.operation_id
545        }
546    })
547}
548
549#[derive(Serialize)]
550#[serde(tag = "operation", rename_all = "snake_case")]
551enum CollectionRetirementResolutionBodyV1 {
552    Retire { request: CollectionRetirementRequestV1 },
553}
554
555fn build_tls(config: &ClusterControlClientConfig) -> Result<rustls::ClientConfig, ClusterControlClientError> {
556    let mut roots = rustls::RootCertStore::empty();
557    for certificate in &config.trust_certificates_der {
558        roots
559            .add(CertificateDer::from(certificate.clone()))
560            .map_err(|_| ClusterControlClientError::InvalidConfiguration)?;
561    }
562    let provider = Arc::new(rustls::crypto::aws_lc_rs::default_provider());
563    let mut tls = rustls::ClientConfig::builder_with_provider(provider)
564        .with_protocol_versions(&[&rustls::version::TLS13])
565        .map_err(|_| ClusterControlClientError::InvalidConfiguration)?
566        .with_root_certificates(roots)
567        .with_no_client_auth();
568    tls.alpn_protocols = vec![CLUSTER_CONTROL_ALPN.to_vec()];
569    Ok(tls)
570}
571
572#[derive(Clone, Debug, Eq, PartialEq)]
573pub struct ClusterControlRefusal {
574    pub code: String,
575    pub disposition: ClusterControlResponseDispositionV3,
576}
577
578#[derive(Debug)]
579pub enum ClusterControlClientError {
580    InvalidConfiguration,
581    ActionDenied,
582    RandomUnavailable,
583    Codec,
584    Transport,
585    DeadlineExceeded,
586    ResponseIdentityMismatch,
587    Refused(ClusterControlRefusal),
588}
589
590impl Display for ClusterControlClientError {
591    fn fmt(&self, formatter: &mut Formatter<'_>) -> fmt::Result {
592        formatter.write_str(match self {
593            Self::InvalidConfiguration => "cluster-control client configuration is invalid",
594            Self::ActionDenied => "cluster-control credential kind cannot invoke this action",
595            Self::RandomUnavailable => "cluster-control request identity entropy is unavailable",
596            Self::Codec => "cluster-control frame is invalid",
597            Self::Transport => "cluster-control transport failed",
598            Self::DeadlineExceeded => "cluster-control call deadline elapsed",
599            Self::ResponseIdentityMismatch => "cluster-control response identity differs from the request",
600            Self::Refused(_) => "cluster-control request was refused",
601        })
602    }
603}
604
605impl Error for ClusterControlClientError {}
606
607#[cfg(test)]
608mod tests {
609    use super::*;
610    use lodestone_proto::{
611        COLLECTION_AUTHORITY_OBSERVATION_FORMAT_VERSION, COLLECTION_AUTHORITY_OBSERVATION_REQUEST_FORMAT_VERSION,
612        COLLECTION_RETIREMENT_REQUEST_FORMAT_VERSION, CollectionAuthorityObservationStateV1, CollectionId,
613        CollectionRetirementClosurePolicyV1, CollectionSecurityGeneration, LogicalMillis, RecoveryEra,
614    };
615
616    fn retirement_request() -> Result<CollectionRetirementRequestV1, Box<dyn Error>> {
617        Ok(CollectionRetirementRequestV1 {
618            format_version: COLLECTION_RETIREMENT_REQUEST_FORMAT_VERSION,
619            operation_id: [19; 32],
620            collection_id: CollectionId::new(7)?,
621            expected_security_generation: CollectionSecurityGeneration::new(3),
622            closure_policy: CollectionRetirementClosurePolicyV1::RetireOwnedChildren,
623        })
624    }
625
626    #[test]
627    fn debug_never_exposes_cluster_credential() -> Result<(), ClusterControlClientError> {
628        let credential = ClusterControlCredential::new(
629            ClusterControlCredentialKindV1::CapacityObserver,
630            b"lcc1_AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA".to_vec(),
631        )?;
632        let debug = format!("{credential:?}");
633        assert!(!debug.contains("lcc1_AAA"));
634        assert!(debug.contains("REDACTED"));
635        Ok(())
636    }
637
638    #[test]
639    fn collection_authority_observation_decoder_refuses_substitution_and_malformed_digest() -> Result<(), Box<dyn Error>>
640    {
641        let request = CollectionAuthorityObservationRequestV1 {
642            format_version: COLLECTION_AUTHORITY_OBSERVATION_REQUEST_FORMAT_VERSION,
643            collection_id: CollectionId::new(7)?,
644        };
645        let mut observation = CollectionAuthorityObservationV1 {
646            format_version: COLLECTION_AUTHORITY_OBSERVATION_FORMAT_VERSION,
647            collection_id: request.collection_id,
648            recovery_era: RecoveryEra::new(1)?,
649            visible_root_digest: [8; 32],
650            observed_at: LogicalMillis::new(9),
651            authority: CollectionAuthorityObservationStateV1::Active {
652                security_generation: CollectionSecurityGeneration::new(2),
653            },
654            semantic_digest: [0; 32],
655        };
656        observation.semantic_digest = observation.canonical_semantic_digest()?;
657        assert_eq!(
658            decode_collection_authority_observation(serde_json::to_value(observation)?, &request)?,
659            observation
660        );
661
662        let mut substituted = observation;
663        substituted.collection_id = CollectionId::new(8)?;
664        substituted.semantic_digest = substituted.canonical_semantic_digest()?;
665        assert!(matches!(
666            decode_collection_authority_observation(serde_json::to_value(substituted)?, &request),
667            Err(ClusterControlClientError::ResponseIdentityMismatch)
668        ));
669        let mut malformed = observation;
670        malformed.semantic_digest[0] ^= 1;
671        assert!(matches!(
672            decode_collection_authority_observation(serde_json::to_value(malformed)?, &request),
673            Err(ClusterControlClientError::Codec)
674        ));
675        Ok(())
676    }
677
678    #[test]
679    fn collection_retirement_call_classification_preserves_submission_authority() -> Result<(), Box<dyn Error>> {
680        let request = retirement_request()?;
681        let not_submitted = classify_collection_retirement_call(
682            Err(ClusterControlClientError::Refused(ClusterControlRefusal {
683                code: "temporarily_unavailable".to_owned(),
684                disposition: ClusterControlResponseDispositionV3::RetrySameRequest,
685            })),
686            &request,
687        )?;
688        assert!(matches!(
689            not_submitted,
690            CollectionRetirementOutcomeV1::NotSubmitted { request_digest, .. }
691                if request_digest == request.canonical_request_digest()?
692        ));
693
694        for ambiguous in [
695            ClusterControlClientError::Transport,
696            ClusterControlClientError::DeadlineExceeded,
697            ClusterControlClientError::ResponseIdentityMismatch,
698            ClusterControlClientError::Refused(ClusterControlRefusal {
699                code: "submitted_indeterminate".to_owned(),
700                disposition: ClusterControlResponseDispositionV3::ResolveOperation,
701            }),
702        ] {
703            assert!(matches!(
704                classify_collection_retirement_call(Err(ambiguous), &request)?,
705                CollectionRetirementOutcomeV1::Indeterminate { request: retained, .. }
706                    if *retained == request
707            ));
708        }
709        assert!(matches!(
710            classify_collection_retirement_call(Ok(serde_json::json!({"state": "unknown"})), &request)?,
711            CollectionRetirementOutcomeV1::Indeterminate { .. }
712        ));
713        let mismatched = CollectionRetirementOutcomeV1::NotSubmitted {
714            format_version: COLLECTION_RETIREMENT_STATUS_FORMAT_VERSION,
715            request_digest: [99; 32],
716        };
717        assert!(matches!(
718            classify_collection_retirement_call(Ok(serde_json::to_value(mismatched)?), &request)?,
719            CollectionRetirementOutcomeV1::Indeterminate { request: retained, .. }
720                if *retained == request
721        ));
722        assert!(matches!(
723            classify_collection_retirement_call(
724                Err(ClusterControlClientError::Refused(ClusterControlRefusal {
725                    code: "conflict".to_owned(),
726                    disposition: ClusterControlResponseDispositionV3::CorrectRequest,
727                })),
728                &request,
729            ),
730            Err(ClusterControlClientError::Refused(_))
731        ));
732        Ok(())
733    }
734}
lodestone_client/cluster_control.rs

lodestone_client/
cluster_control.rs
