spiral/
registration.rs

1//! Idempotent registration of already persisted table versions.
2use spiral_contract::call::CallError;
3
4use crate::api::tables::objects::TableId;
5use crate::api::tables::{CreateVersionRequest, TablesError};
6use crate::resources::arbor::Client as ArborClient;
7use crate::scandal::types::{PreparedWriteReceipt, TableWriteResult};
8use crate::{Client, Error, Table};
9
10impl Client {
11    /// Retry resource registration without replaying a consumed Arrow upload.
12    ///
13    /// The receipt must come from this project's authorized scandal write.
14    /// Registration alone does not publish a table in Catalog.
15    pub async fn register_prepared(&self, receipt: PreparedWriteReceipt) -> Result<Table, Error> {
16        let result = register(&self.connection.arbor, receipt).await?;
17        Ok(self.table_by_id(result.resource_id, result.version_id))
18    }
19}
20
21pub(crate) async fn register(
22    client: &ArborClient,
23    receipt: PreparedWriteReceipt,
24) -> Result<TableWriteResult, Error> {
25    let result = client
26        .create_version(
27            TableId(receipt.resource_id),
28            &CreateVersionRequest {
29                base_version_id: receipt.base_version,
30                version_id: receipt.version_id,
31            },
32        )
33        .await
34        .map_err(|error| match &error {
35            Error::Arbor(failure) if registration_may_have_committed(failure) => {
36                Error::RegistrationUnknown {
37                    receipt: receipt.clone(),
38                    message: error.to_string(),
39                }
40            }
41            _ => error,
42        })?;
43    if result.version_id != receipt.version_id
44        || result.parent_version_id != receipt.base_version
45        || result.seq_no.0 != receipt.sequence.0
46    {
47        return Err(Error::RegistrationUnknown {
48            receipt,
49            message: "registration did not match the prepared receipt".into(),
50        });
51    }
52    Ok(TableWriteResult {
53        resource_id: receipt.resource_id,
54        version_id: receipt.version_id,
55    })
56}
57
58fn registration_may_have_committed(error: &CallError<TablesError>) -> bool {
59    match error {
60        CallError::Service(failure) => failure.status() >= 500,
61        CallError::UnknownRemote(failure) => failure.status() >= 500,
62        CallError::Transport(failure) => failure.observed_status().is_none_or(uncertain_status),
63        CallError::Protocol(failure) => failure.observed_status().is_none_or(uncertain_status),
64        CallError::Internal { .. } => true,
65    }
66}
67
68fn uncertain_status(status: u16) -> bool {
69    (200..300).contains(&status) || status >= 500
70}
71
72#[cfg(test)]
73mod tests {
74    use serde_json::json;
75    use uuid::Uuid;
76    use wiremock::matchers::{body_json, method, path};
77    use wiremock::{Mock, MockServer, ResponseTemplate};
78
79    use super::*;
80    use crate::auth::TokenSource;
81
82    #[tokio::test]
83    async fn registration_retains_the_receipt_when_the_outcome_is_uncertain() {
84        let receipt = PreparedWriteReceipt {
85            base_version: Some(Uuid::now_v7()),
86            resource_id: Uuid::new_v4(),
87            sequence: 2.into(),
88            version_id: Uuid::now_v7(),
89        };
90        for status in [200, 409, 500] {
91            let server = MockServer::start().await;
92            Mock::given(method("POST"))
93                .and(path(format!(
94                    "/api/tables/{}/versions",
95                    receipt.resource_id
96                )))
97                .and(body_json(json!({
98                    "baseVersionId": receipt.base_version,
99                    "versionId": receipt.version_id,
100                })))
101                .respond_with(ResponseTemplate::new(status).set_body_json(json!({
102                    "versionId": receipt.version_id,
103                    "parentVersionId": receipt.base_version,
104                    "seqNo": receipt.sequence,
105                })))
106                .expect(1)
107                .mount(&server)
108                .await;
109            let result = register(
110                &ArborClient::new(&server.uri())
111                    .unwrap()
112                    .with_token_source(TokenSource::fixed("test".into())),
113                receipt.clone(),
114            )
115            .await;
116            match status {
117                200 => assert_eq!(result.unwrap().version_id, receipt.version_id),
118                409 => assert!(matches!(result, Err(Error::Arbor(_)))),
119                _ => assert!(matches!(
120                    result,
121                    Err(Error::RegistrationUnknown { receipt: retained, .. }) if retained == receipt
122                )),
123            }
124        }
125
126        let unavailable = ArborClient::new("http://127.0.0.1:9")
127            .unwrap()
128            .with_token_source(TokenSource::fixed("test".into()));
129        assert!(matches!(
130            register(&unavailable, receipt.clone()).await,
131            Err(Error::RegistrationUnknown { receipt: retained, .. }) if retained == receipt
132        ));
133    }
134}
spiral/registration.rs

spiral/
registration.rs
