diff --git a/rust/lance-namespace-impls/src/dir.rs b/rust/lance-namespace-impls/src/dir.rs index bd5fa70f89d..dc2d83cf278 100644 --- a/rust/lance-namespace-impls/src/dir.rs +++ b/rust/lance-namespace-impls/src/dir.rs @@ -48,13 +48,14 @@ use crate::context::DynamicContextProvider; use lance_namespace::models::{ AlterTableAddColumnsRequest, AlterTableAddColumnsResponse, AlterTableAlterColumnsRequest, AlterTableAlterColumnsResponse, AlterTableDropColumnsRequest, AlterTableDropColumnsResponse, - AnalyzeTableQueryPlanRequest, BatchDeleteTableVersionsRequest, - BatchDeleteTableVersionsResponse, BranchContents as ModelBranchContents, CountTableRowsRequest, - CreateNamespaceRequest, CreateNamespaceResponse, CreateTableBranchRequest, - CreateTableBranchResponse, CreateTableIndexRequest, CreateTableIndexResponse, - CreateTableRequest, CreateTableResponse, CreateTableScalarIndexResponse, CreateTableTagRequest, - CreateTableTagResponse, CreateTableVersionRequest, CreateTableVersionResponse, - DeclareTableRequest, DeclareTableResponse, DeleteFromTableRequest, DeleteFromTableResponse, + AlterTransactionRequest, AlterTransactionResponse, AnalyzeTableQueryPlanRequest, + BatchDeleteTableVersionsRequest, BatchDeleteTableVersionsResponse, + BranchContents as ModelBranchContents, CountTableRowsRequest, CreateNamespaceRequest, + CreateNamespaceResponse, CreateTableBranchRequest, CreateTableBranchResponse, + CreateTableIndexRequest, CreateTableIndexResponse, CreateTableRequest, CreateTableResponse, + CreateTableScalarIndexResponse, CreateTableTagRequest, CreateTableTagResponse, + CreateTableVersionRequest, CreateTableVersionResponse, DeclareTableRequest, + DeclareTableResponse, DeleteFromTableRequest, DeleteFromTableResponse, DeleteTableBranchRequest, DeleteTableBranchResponse, DeleteTableTagRequest, DeleteTableTagResponse, DescribeNamespaceRequest, DescribeNamespaceResponse, DescribeTableIndexStatsRequest, DescribeTableIndexStatsResponse, DescribeTableRequest, @@ -913,6 +914,74 @@ struct TableDeleteEntry { ranges: Vec<(i64, i64)>, } +/// Persistent record of `alter_transaction` outcomes for a single transaction. +/// +/// Lance's transaction file is immutable once written, so we record any +/// modifications (status transitions, extra properties, tombstoned properties) +/// in a namespace-owned sidecar file. The sidecar is then merged into the +/// response of subsequent `describe_transaction` / `alter_transaction` calls. +/// +/// Serialization is implemented manually via `serde_json::Value` to avoid +/// pulling in `serde`'s `derive` feature for this crate. +#[derive(Debug, Clone, Default)] +struct TransactionAlteration { + /// The most recently applied status, if any. + status: Option, + /// User-defined properties layered on top of the immutable transaction + /// properties. Values here take precedence over the transaction's own + /// properties when both are present. + properties: HashMap, + /// Names of transaction properties that have been tombstoned via + /// `unset_property_action`. A tombstoned key is hidden from the response + /// even when the immutable transaction still carries it. + removed_properties: std::collections::HashSet, +} + +impl TransactionAlteration { + /// JSON field names used for the sidecar on-disk representation. + const F_STATUS: &'static str = "status"; + const F_PROPERTIES: &'static str = "properties"; + const F_REMOVED_PROPERTIES: &'static str = "removed_properties"; + + /// Serialize this alteration to a JSON byte vector. + /// + /// Uses the same pattern as `dir/manifest.rs`: rely on the built-in + /// `Serialize` impls for `Option`, `HashMap` and + /// `HashSet` provided by the `serde` crate (transitively pulled in + /// by `serde_json`), so no `serde` derive nor extra dependency is needed. + fn to_json_bytes(&self) -> serde_json::Result> { + serde_json::to_vec(&serde_json::json!({ + Self::F_STATUS: self.status, + Self::F_PROPERTIES: self.properties, + Self::F_REMOVED_PROPERTIES: self.removed_properties, + })) + } + + /// Deserialize an alteration from JSON bytes, mirroring the + /// `serde_json::from_slice::>(...)` idiom already + /// used in `dir/manifest.rs`. Missing / null fields fall back to defaults + /// so that the sidecar format stays forward-compatible. + fn from_json_slice(bytes: &[u8]) -> serde_json::Result { + let mut obj: serde_json::Map = serde_json::from_slice(bytes)?; + Ok(Self { + status: serde_json::from_value( + obj.remove(Self::F_STATUS) + .unwrap_or(serde_json::Value::Null), + )?, + properties: serde_json::from_value( + obj.remove(Self::F_PROPERTIES) + .unwrap_or(serde_json::Value::Null), + ) + .unwrap_or_default(), + removed_properties: serde_json::from_value( + obj.remove(Self::F_REMOVED_PROPERTIES) + .unwrap_or(serde_json::Value::Null), + ) + .unwrap_or_default(), + }) + } +} + impl DirectoryNamespace { /// Apply pagination to a list of table names /// @@ -2064,12 +2133,29 @@ impl DirectoryNamespace { fn transaction_response( version: u64, transaction: &Transaction, + alteration: Option, ) -> DescribeTransactionResponse { let mut properties = transaction .transaction_properties .as_ref() .map(|properties| (**properties).clone()) .unwrap_or_default(); + + // Apply persisted alterations on top of the immutable transaction + // properties so callers see the current effective state. + let mut effective_status = "SUCCEEDED".to_string(); + if let Some(alteration) = alteration { + for key in &alteration.removed_properties { + properties.remove(key); + } + for (key, value) in alteration.properties { + properties.insert(key, value); + } + if let Some(status) = alteration.status { + effective_status = status; + } + } + properties.insert("uuid".to_string(), transaction.uuid.clone()); properties.insert("version".to_string(), version.to_string()); properties.insert( @@ -2085,7 +2171,7 @@ impl DirectoryNamespace { } DescribeTransactionResponse { - status: "SUCCEEDED".to_string(), + status: effective_status, properties: Some(properties), } } @@ -2174,6 +2260,84 @@ impl DirectoryNamespace { .into()) } + /// Relative directory (under a table's Lance root) used to persist + /// alter_transaction outcomes. The Lance transaction file itself is + /// immutable, so we keep alterations in a namespace-owned sidecar. + const TRANSACTION_ALTERATIONS_DIR: &'static str = "_alter_transactions"; + + fn transaction_alteration_path(&self, table_uri: &str, txn_uuid: &str) -> Result { + let table_path = self.object_store_path_from_uri(table_uri)?; + Ok(table_path + .join(Self::TRANSACTION_ALTERATIONS_DIR) + .join(format!("{}.json", txn_uuid).as_str())) + } + + async fn load_transaction_alteration( + &self, + table_uri: &str, + txn_uuid: &str, + ) -> Result> { + let path = self.transaction_alteration_path(table_uri, txn_uuid)?; + match self.object_store.inner.get(&path).await { + Ok(get_result) => { + let bytes = get_result.bytes().await.map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to read alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; + let alteration = TransactionAlteration::from_json_slice(&bytes).map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to parse alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; + Ok(Some(alteration)) + } + Err(ObjectStoreError::NotFound { .. }) => Ok(None), + Err(e) => Err(lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to load alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + })), + } + } + + async fn save_transaction_alteration( + &self, + table_uri: &str, + txn_uuid: &str, + alteration: &TransactionAlteration, + ) -> Result<()> { + let path = self.transaction_alteration_path(table_uri, txn_uuid)?; + let bytes = alteration.to_json_bytes().map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to serialize alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; + self.object_store + .inner + .put(&path, bytes.into()) + .await + .map_err(|e| { + lance_core::Error::from(NamespaceError::Internal { + message: format!( + "Failed to persist alter_transaction sidecar for '{}': {}", + txn_uuid, e + ), + }) + })?; + Ok(()) + } + fn table_full_uri(&self, table_name: &str) -> String { format!("{}/{}.lance", &self.root, table_name) } @@ -3761,7 +3925,212 @@ impl LanceNamespace for DirectoryNamespace { .await?; let (version, transaction) = self.find_transaction(&dataset, &id).await?; - Ok(Self::transaction_response(version, &transaction)) + // Merge any persisted alter_transaction changes stored in the sidecar + // so that describe_transaction reflects the latest altered state. + let sidecar = self + .load_transaction_alteration(&table_uri, &transaction.uuid) + .await?; + + Ok(Self::transaction_response(version, &transaction, sidecar)) + } + + async fn alter_transaction( + &self, + request: AlterTransactionRequest, + ) -> Result { + self.record_op("alter_transaction"); + + // Parse the request ID: must include table id and transaction identifier + let mut request_id = request.id.ok_or_else(|| { + lance_core::Error::from(NamespaceError::InvalidInput { + message: "Transaction id must include table id and transaction identifier" + .to_string(), + }) + })?; + if request_id.len() < 2 { + return Err(NamespaceError::InvalidInput { + message: format!( + "Transaction request id must include table id and transaction identifier, got {:?}", + request_id + ), + } + .into()); + } + + let txn_id = request_id.pop().expect("request_id len checked above"); + let table_id = Some(request_id); + let table_uri = self.resolve_table_location(&table_id).await?; + let dataset = self + .load_dataset(&table_uri, None, "alter_transaction") + .await?; + let (version, transaction) = self.find_transaction(&dataset, &txn_id).await?; + + // Reserved keys are derived from the immutable Transaction metadata and + // must not be modified via alter_transaction. They are only surfaced in + // the response for the caller's convenience. + const RESERVED_KEYS: &[&str] = &["uuid", "version", "read_version", "operation", "tag"]; + let is_reserved = |key: &str| RESERVED_KEYS.contains(&key); + + // Load the existing sidecar (if any) so alterations accumulate across + // successive alter_transaction calls. + let mut sidecar = self + .load_transaction_alteration(&table_uri, &transaction.uuid) + .await? + .unwrap_or_default(); + + for action in &request.actions { + if let Some(ref set_status) = action.set_status_action + && let Some(ref status) = set_status.status + { + // Validate the status value (case-insensitive) + let normalized = status.to_lowercase().replace('_', ""); + match normalized.as_str() { + "queued" | "running" | "succeeded" | "failed" | "canceled" => { + sidecar.status = Some(status.clone()); + } + _ => { + return Err(NamespaceError::InvalidInput { + message: format!( + "Invalid transaction status '{}'. Valid values are: Queued, Running, Succeeded, Failed, Canceled", + status + ), + } + .into()); + } + } + } + + if let Some(ref set_property) = action.set_property_action + && let (Some(key), Some(value)) = (&set_property.key, &set_property.value) + { + if is_reserved(key) { + return Err(NamespaceError::InvalidInput { + message: format!("Property '{}' is reserved and cannot be modified", key), + } + .into()); + } + let mode = set_property + .mode + .as_deref() + .unwrap_or("Overwrite") + .to_lowercase(); + match mode.as_str() { + "overwrite" => { + sidecar.properties.insert(key.clone(), value.clone()); + } + "fail" => { + // Consider both the immutable transaction properties + // and any values previously written to the sidecar. + let exists = sidecar.properties.contains_key(key) + || transaction + .transaction_properties + .as_ref() + .is_some_and(|props| props.contains_key(key)); + if exists { + return Err(NamespaceError::ConcurrentModification { + message: format!( + "Property '{}' already exists and mode is 'Fail'", + key + ), + } + .into()); + } + sidecar.properties.insert(key.clone(), value.clone()); + } + "skip" => { + let exists = sidecar.properties.contains_key(key) + || transaction + .transaction_properties + .as_ref() + .is_some_and(|props| props.contains_key(key)); + if !exists { + sidecar.properties.insert(key.clone(), value.clone()); + } + } + _ => { + return Err(NamespaceError::InvalidInput { + message: format!( + "Invalid set_property mode '{}'. Valid values are: Overwrite, Fail, Skip", + mode + ), + } + .into()); + } + } + } + + if let Some(ref unset_property) = action.unset_property_action + && let Some(ref key) = unset_property.key + { + if is_reserved(key) { + return Err(NamespaceError::InvalidInput { + message: format!("Property '{}' is reserved and cannot be modified", key), + } + .into()); + } + let mode = unset_property + .mode + .as_deref() + .unwrap_or("Skip") + .to_lowercase(); + let exists_in_transaction = transaction + .transaction_properties + .as_ref() + .is_some_and(|props| props.contains_key(key)); + match mode.as_str() { + "skip" => { + sidecar.properties.remove(key); + if exists_in_transaction { + // Track a tombstone so describe_transaction can + // hide the immutable property from the response. + sidecar.removed_properties.insert(key.clone()); + } + } + "fail" => { + if !sidecar.properties.contains_key(key) && !exists_in_transaction { + return Err(NamespaceError::InvalidInput { + message: format!( + "Property '{}' does not exist and mode is 'Fail'", + key + ), + } + .into()); + } + sidecar.properties.remove(key); + if exists_in_transaction { + sidecar.removed_properties.insert(key.clone()); + } + } + _ => { + return Err(NamespaceError::InvalidInput { + message: format!( + "Invalid unset_property mode '{}'. Valid values are: Skip, Fail", + mode + ), + } + .into()); + } + } + } + } + + // Persist the accumulated alterations so subsequent calls observe + // them. The transaction file itself is immutable in Lance, so we + // record alter_transaction outcomes in a namespace-owned sidecar. + self.save_transaction_alteration(&table_uri, &transaction.uuid, &sidecar) + .await?; + + // Assemble the response by merging the immutable transaction metadata + // with the persisted alterations. + let final_status = sidecar + .status + .clone() + .unwrap_or_else(|| "SUCCEEDED".to_string()); + let response = Self::transaction_response(version, &transaction, Some(sidecar)); + Ok(AlterTransactionResponse { + status: final_status, + properties: response.properties, + }) } async fn create_table_scalar_index( @@ -12968,4 +13337,341 @@ mod tests { .expect("reopen branch failed"); assert_eq!(scan_id_column(&reopened).await, vec![1, 2]); } + + #[tokio::test] + async fn test_alter_transaction_set_status() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetStatus, + DescribeTransactionRequest, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); + + // First verify the transaction exists + let describe_resp = namespace + .describe_transaction(DescribeTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(describe_resp.status, "SUCCEEDED"); + + // Alter the transaction status + let response = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(response.status, "Canceled"); + assert!(response.properties.is_some()); + let props = response.properties.unwrap(); + assert_eq!(props.get("uuid"), Some(&txn_id)); + assert_eq!(props.get("operation"), Some(&"CreateIndex".to_string())); + } + + #[tokio::test] + async fn test_alter_transaction_set_property() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetProperty, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); + + let response = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("custom_key".to_string()), + value: Some("custom_value".to_string()), + mode: None, + })), + unset_property_action: None, + }], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(response.status, "SUCCEEDED"); + let props = response.properties.unwrap(); + assert_eq!(props.get("custom_key"), Some(&"custom_value".to_string())); + } + + #[tokio::test] + async fn test_alter_transaction_set_property_fail_mode() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetProperty, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); + + // First, set a non-reserved property so it exists in the sidecar. + namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("custom_key".to_string()), + value: Some("initial_value".to_string()), + mode: None, + })), + unset_property_action: None, + }], + ..Default::default() + }) + .await + .unwrap(); + + // Now try to set the same property again with Fail mode, which must + // exercise the mode='Fail' branch (not the reserved-key guard). + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("custom_key".to_string()), + value: Some("new_value".to_string()), + mode: Some("Fail".to_string()), + })), + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn test_alter_transaction_unset_property() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetProperty, + AlterTransactionUnsetProperty, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); + + // First set a custom property, then unset it + let response = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![ + AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("temp_key".to_string()), + value: Some("temp_value".to_string()), + mode: None, + })), + unset_property_action: None, + }, + AlterTransactionAction { + set_status_action: None, + set_property_action: None, + unset_property_action: Some(Box::new(AlterTransactionUnsetProperty { + key: Some("temp_key".to_string()), + mode: None, + })), + }, + ], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(response.status, "SUCCEEDED"); + let props = response.properties.unwrap(); + assert!(!props.contains_key("temp_key")); + } + + #[tokio::test] + async fn test_alter_transaction_invalid_status() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetStatus, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let txn_id = create_scalar_index(&namespace, "users", "users_id_idx") + .await + .expect("create_scalar_index should return a transaction id"); + + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("InvalidStatus".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn test_alter_transaction_not_found() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetStatus, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + + // Try to alter a non-existent transaction + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), "non_existent_txn".to_string()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn test_alter_transaction_missing_id() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetStatus, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + + // Try with missing id + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: None, + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + + // Try with insufficient id parts + let result = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string()]), + actions: vec![AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }], + ..Default::default() + }) + .await; + assert!(result.is_err()); + } + + #[tokio::test] + async fn test_alter_transaction_persists_changes() { + use lance_namespace::models::{ + AlterTransactionAction, AlterTransactionRequest, AlterTransactionSetProperty, + AlterTransactionSetStatus, DescribeTransactionRequest, + }; + + let (namespace, _temp_dir) = create_test_namespace().await; + create_scalar_table(&namespace, "users").await; + let transaction_id = create_scalar_index(&namespace, "users", "users_id_idx").await; + + let txn_id = transaction_id.expect("scalar index should produce a transaction id"); + + // Alter status and set a custom property. + namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![ + AlterTransactionAction { + set_status_action: Some(Box::new(AlterTransactionSetStatus { + status: Some("Canceled".to_string()), + })), + set_property_action: None, + unset_property_action: None, + }, + AlterTransactionAction { + set_status_action: None, + set_property_action: Some(Box::new(AlterTransactionSetProperty { + key: Some("owner".to_string()), + value: Some("alice".to_string()), + mode: None, + })), + unset_property_action: None, + }, + ], + ..Default::default() + }) + .await + .unwrap(); + + // The changes must survive across a fresh describe_transaction call, + // proving the alteration was persisted to the transaction file. + let describe_resp = namespace + .describe_transaction(DescribeTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + ..Default::default() + }) + .await + .unwrap(); + let props = describe_resp.properties.expect("properties should be set"); + assert_eq!(props.get("owner"), Some(&"alice".to_string())); + // The internal `_status` marker should not leak into the response but + // must be present on disk so subsequent alter_transaction calls can + // observe the previously set status. + assert!(!props.contains_key("_status")); + + let follow_up = namespace + .alter_transaction(AlterTransactionRequest { + id: Some(vec!["users".to_string(), txn_id.clone()]), + actions: vec![], + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(follow_up.status, "Canceled"); + let follow_up_props = follow_up.properties.unwrap(); + assert_eq!(follow_up_props.get("owner"), Some(&"alice".to_string())); + } }