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

lodestone_client/
cluster_control.rs
