From be5fc08ee7ac9be8cf484415fcb4d86e0532c35e Mon Sep 17 00:00:00 2001 From: Fabian Meiswinkel Date: Mon, 27 Apr 2026 11:21:04 +0000 Subject: [PATCH 1/7] Move create_item and upsert_item back to accepting only ItemReference --- Cargo.lock | 154 ++++++++++++++---- .../examples/cosmos/create.rs | 7 +- .../examples/cosmos/upsert.rs | 6 +- .../src/clients/container_client.rs | 36 ++-- .../tests/emulator_tests/cosmos_batch.rs | 6 +- .../emulator_tests/cosmos_fault_injection.rs | 28 ++-- .../tests/emulator_tests/cosmos_items.rs | 43 +++-- .../tests/framework/test_data.rs | 3 +- .../multi_write_tests/cosmos_multi_write.rs | 1 + .../cosmos_multi_write_fault_injection.rs | 26 +-- .../cosmos_multi_write_retry_policies.rs | 17 +- .../azure_data_cosmos_benchmarks/src/lib.rs | 12 +- .../src/driver/pipeline/operation_pipeline.rs | 26 ++- .../src/models/cosmos_operation.rs | 32 ++-- .../src/models/cosmos_resource_reference.rs | 27 +++ .../emulator_tests/driver_fault_injection.rs | 10 +- .../emulator_tests/driver_item_operations.rs | 6 +- .../tests/framework/test_client.rs | 5 +- .../src/operations/create_item.rs | 2 +- .../src/operations/upsert_item.rs | 2 +- 20 files changed, 307 insertions(+), 142 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c6be8ad7523..8e92035b9e0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -266,6 +266,30 @@ dependencies = [ "azure_canary", ] +[[package]] +name = "azure_core" +version = "0.35.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9966f75417a6779ea221531960d9d46fe630879606501aa5277312ad4ae38587" +dependencies = [ + "async-lock", + "async-trait", + "azure_core_macros 0.8.0", + "bytes", + "futures", + "hmac", + "openssl", + "pin-project", + "rustc_version", + "serde", + "serde_json", + "sha2", + "tokio", + "tracing", + "typespec 0.14.0", + "typespec_client_core 0.14.0", +] + [[package]] name = "azure_core" version = "0.36.0" @@ -274,7 +298,7 @@ dependencies = [ "async-trait", "azure_core_macros 0.8.0", "azure_core_test", - "azure_identity", + "azure_identity 0.36.0", "azure_security_keyvault_certificates", "azure_security_keyvault_secrets", "bytes", @@ -296,7 +320,7 @@ dependencies = [ "tracing", "tracing-subscriber", "typespec 0.14.0", - "typespec_client_core", + "typespec_client_core 0.15.0", "ureq", ] @@ -305,7 +329,7 @@ name = "azure_core_amqp" version = "0.15.0" dependencies = [ "async-trait", - "azure_core", + "azure_core 0.36.0", "fe2o3-amqp", "fe2o3-amqp-cbs", "fe2o3-amqp-ext", @@ -350,10 +374,10 @@ dependencies = [ name = "azure_core_opentelemetry" version = "0.10.0" dependencies = [ - "azure_core", + "azure_core 0.36.0", "azure_core_test", "azure_core_test_macros", - "azure_identity", + "azure_identity 0.36.0", "opentelemetry", "opentelemetry_sdk", "tokio", @@ -366,9 +390,9 @@ name = "azure_core_test" version = "0.1.0" dependencies = [ "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_test_macros", - "azure_identity", + "azure_identity 0.36.0", "azure_security_keyvault_secrets", "clap", "dotenvy", @@ -391,7 +415,7 @@ dependencies = [ name = "azure_core_test_macros" version = "0.1.0" dependencies = [ - "azure_core", + "azure_core 0.36.0", "azure_core_test", "proc-macro2", "quote", @@ -405,10 +429,10 @@ version = "0.33.0" dependencies = [ "async-lock", "async-trait", - "azure_core", + "azure_core 0.35.0", "azure_core_test", "azure_data_cosmos_driver", - "azure_identity", + "azure_identity 0.35.0", "base64 0.22.1", "clap", "futures", @@ -428,7 +452,7 @@ name = "azure_data_cosmos_benchmarks" version = "0.1.0" dependencies = [ "async-trait", - "azure_core", + "azure_core 0.35.0", "azure_data_cosmos_driver", "criterion", "tokio", @@ -442,9 +466,9 @@ dependencies = [ "arc-swap", "async-lock", "async-trait", - "azure_core", + "azure_core 0.35.0", "azure_data_cosmos_macros", - "azure_identity", + "azure_identity 0.35.0", "base64 0.22.1", "bytes", "crossbeam-epoch", @@ -478,9 +502,9 @@ name = "azure_data_cosmos_perf" version = "0.1.0" dependencies = [ "async-trait", - "azure_core", + "azure_core 0.35.0", "azure_data_cosmos", - "azure_identity", + "azure_identity 0.35.0", "clap", "console-subscriber", "futures", @@ -496,13 +520,31 @@ dependencies = [ "uuid", ] +[[package]] +name = "azure_identity" +version = "0.35.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "385e4426c01965149a002a72c54c43d6f1b9d2244179e70647df48e3b28f8c32" +dependencies = [ + "async-lock", + "async-trait", + "azure_core 0.35.0", + "futures", + "pin-project", + "serde", + "serde_json", + "time", + "tracing", + "url", +] + [[package]] name = "azure_identity" version = "0.36.0" dependencies = [ "async-lock", "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_test", "azure_security_keyvault_secrets", "clap", @@ -528,10 +570,10 @@ dependencies = [ "async-lock", "async-stream", "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_amqp", "azure_core_test", - "azure_identity", + "azure_identity 0.36.0", "azure_messaging_eventhubs", "criterion", "fe2o3-amqp", @@ -549,10 +591,10 @@ name = "azure_messaging_eventhubs_checkpointstore_blob" version = "0.9.0" dependencies = [ "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_opentelemetry", "azure_core_test", - "azure_identity", + "azure_identity 0.36.0", "azure_messaging_eventhubs", "azure_storage_blob", "futures", @@ -575,10 +617,10 @@ dependencies = [ "async-lock", "async-stream", "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_amqp", "azure_core_test", - "azure_identity", + "azure_identity 0.36.0", "azure_messaging_servicebus", "futures", "rand 0.10.1", @@ -597,9 +639,9 @@ version = "0.14.0" dependencies = [ "async-lock", "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_test", - "azure_identity", + "azure_identity 0.36.0", "azure_security_keyvault_keys", "azure_security_keyvault_test", "futures", @@ -619,9 +661,9 @@ version = "0.15.0" dependencies = [ "async-lock", "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_test", - "azure_identity", + "azure_identity 0.36.0", "azure_security_keyvault_test", "criterion", "futures", @@ -642,9 +684,9 @@ version = "0.15.0" dependencies = [ "async-lock", "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_test", - "azure_identity", + "azure_identity 0.36.0", "azure_security_keyvault_test", "futures", "include-file", @@ -670,10 +712,10 @@ version = "0.13.0" dependencies = [ "async-stream", "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_opentelemetry", "azure_core_test", - "azure_identity", + "azure_identity 0.36.0", "azure_storage_blob_test", "bytes", "clap", @@ -698,7 +740,7 @@ name = "azure_storage_blob_test" version = "0.1.0" dependencies = [ "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_test", "azure_storage_blob", "bytes", @@ -711,10 +753,10 @@ name = "azure_storage_queue" version = "0.7.0" dependencies = [ "async-trait", - "azure_core", + "azure_core 0.36.0", "azure_core_opentelemetry", "azure_core_test", - "azure_identity", + "azure_identity 0.36.0", "clap", "futures", "opentelemetry-stdout", @@ -1857,6 +1899,22 @@ dependencies = [ "tower-service", ] +[[package]] +name = "hyper-tls" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "70206fc6890eaca9fde8a0bf71caa2ddfc9fe045ac9e5c70df101a7dbde866e0" +dependencies = [ + "bytes", + "http-body-util", + "hyper", + "hyper-util", + "native-tls", + "tokio", + "tokio-native-tls", + "tower-service", +] + [[package]] name = "hyper-util" version = "0.1.20" @@ -2968,9 +3026,11 @@ dependencies = [ "http-body-util", "hyper", "hyper-rustls", + "hyper-tls", "hyper-util", "js-sys", "log", + "native-tls", "percent-encoding", "pin-project-lite", "quinn", @@ -2979,6 +3039,7 @@ dependencies = [ "rustls-platform-verifier", "sync_wrapper", "tokio", + "tokio-native-tls", "tokio-rustls", "tokio-util", "tower 0.5.3", @@ -3997,6 +4058,31 @@ dependencies = [ "url", ] +[[package]] +name = "typespec_client_core" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4714a935f6aa74724775f07291390e8f07df0a1804544cd4106ac3b75c0478d2" +dependencies = [ + "async-trait", + "base64 0.22.1", + "bytes", + "dyn-clone", + "futures", + "pin-project", + "rand 0.10.1", + "reqwest", + "serde", + "serde_json", + "time", + "tokio", + "tracing", + "typespec 0.14.0", + "typespec_macros 0.13.0", + "url", + "uuid", +] + [[package]] name = "typespec_client_core" version = "0.15.0" @@ -4047,7 +4133,7 @@ dependencies = [ "serde_json", "syn 2.0.117", "tokio", - "typespec_client_core", + "typespec_client_core 0.15.0", ] [[package]] diff --git a/sdk/cosmos/azure_data_cosmos/examples/cosmos/create.rs b/sdk/cosmos/azure_data_cosmos/examples/cosmos/create.rs index 6c0b3a43372..090a93e9797 100644 --- a/sdk/cosmos/azure_data_cosmos/examples/cosmos/create.rs +++ b/sdk/cosmos/azure_data_cosmos/examples/cosmos/create.rs @@ -33,6 +33,10 @@ pub enum Subcommands { #[arg(long, short)] partition_key: String, + /// The id of the new item. + #[arg(long, short)] + item_id: String, + /// The JSON of the new item. #[arg(long, short)] json: String, @@ -77,6 +81,7 @@ impl CreateCommand { database, container, partition_key, + item_id, json, show_updated, } => { @@ -94,7 +99,7 @@ impl CreateCommand { None }; - let response = container_client.create_item(pk, item, options).await?; + let response = container_client.create_item(pk, &item_id, item, options).await?; println!("Created item successfully"); diff --git a/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs b/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs index 4e77d466d97..1879d1f02f6 100644 --- a/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs +++ b/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs @@ -21,6 +21,10 @@ pub struct UpsertCommand { #[arg(long, short)] partition_key: String, + /// The id of the new item. + #[arg(long, short)] + item_id: String, + /// The JSON of the new item. #[arg(long, short)] json: String, @@ -46,7 +50,7 @@ impl UpsertCommand { None }; - let response = container_client.upsert_item(pk, item, options).await?; + let response = container_client.upsert_item(pk, &self.item_id, item, options).await?; println!("Item updated successfully"); if self.show_updated { diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs index fc6fb790ea9..108e2c16e17 100644 --- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs +++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs @@ -276,7 +276,7 @@ impl ContainerClient { /// }; /// # let container_client: azure_data_cosmos::clients::ContainerClient = panic!("this is a non-running example"); /// container_client - /// .create_item("category1", p, None) + /// .create_item("category1", "product1", p, None) /// .await?; /// # } /// ``` @@ -308,7 +308,7 @@ impl ContainerClient { /// operation.content_response_on_write = Some(ContentResponseOnWrite::Enabled); /// let options = ItemWriteOptions::default().with_operation_options(operation); /// let created_item = container_client - /// .create_item("category1", p, Some(options)) + /// .create_item("category1", "product1", p, Some(options)) /// .await? /// .into_body().json::(); /// # Ok(()) @@ -317,16 +317,22 @@ impl ContainerClient { pub async fn create_item( &self, partition_key: impl Into, + item_id: &str, item: T, options: Option, ) -> azure_core::Result> { let options = options.unwrap_or_default(); let body = serde_json::to_vec(&item)?; - let driver_pk = partition_key.into().into_driver_partition_key(); + + // Build the driver's item reference from our stored container metadata. + let item_ref = ItemReference::from_name( + &self.container_ref, + partition_key.into().into_driver_partition_key(), + item_id.to_owned(), + ); // Create the driver operation and apply ItemWriteOptions fields. - let operation = - CosmosOperation::create_item(self.container_ref.clone(), driver_pk).with_body(body); + let operation = CosmosOperation::create_item(item_ref).with_body(body); let operation = apply_item_options(operation, options.session_token, options.precondition); // Execute through the driver. @@ -469,7 +475,7 @@ impl ContainerClient { /// }; /// # let container_client: azure_data_cosmos::clients::ContainerClient = panic!("this is a non-running example"); /// container_client - /// .upsert_item("category1", p, None) + /// .upsert_item("category1", "product1", p, None) /// .await?; /// # Ok(()) /// # } @@ -502,7 +508,7 @@ impl ContainerClient { /// operation.content_response_on_write = Some(ContentResponseOnWrite::Enabled); /// let options = ItemWriteOptions::default().with_operation_options(operation); /// let updated_product = container_client - /// .upsert_item("category1", p, Some(options)) + /// .upsert_item("category1", "product1", p, Some(options)) /// .await? /// .into_body().json::()?; /// Ok(()) @@ -510,16 +516,22 @@ impl ContainerClient { pub async fn upsert_item( &self, partition_key: impl Into, + item_id: &str, item: T, options: Option, ) -> azure_core::Result> { let options = options.unwrap_or_default(); let body = serde_json::to_vec(&item)?; - let driver_pk = partition_key.into().into_driver_partition_key(); + + // Build the driver's item reference from our stored container metadata. + let item_ref = ItemReference::from_name( + &self.container_ref, + partition_key.into().into_driver_partition_key(), + item_id.to_owned(), + ); // Create the driver operation and apply ItemWriteOptions fields. - let operation = - CosmosOperation::upsert_item(self.container_ref.clone(), driver_pk).with_body(body); + let operation = CosmosOperation::upsert_item(item_ref).with_body(body); let operation = apply_item_options(operation, options.session_token, options.precondition); // Execute through the driver. @@ -954,9 +966,9 @@ mod tests { assert_send(client.read_throughput(todo!())); assert_send(client.begin_replace_throughput(todo!(), todo!())); assert_send(client.delete(todo!())); - assert_send(client.create_item::("", todo!(), todo!())); + assert_send(client.create_item::("", todo!(), todo!(), todo!())); assert_send(client.replace_item::("", todo!(), todo!(), todo!())); - assert_send(client.upsert_item::("", todo!(), todo!())); + assert_send(client.upsert_item::("", todo!(), todo!(), todo!())); assert_send(client.read_item::("", todo!(), todo!())); assert_send(client.delete_item("", todo!(), todo!())); assert_send(client.execute_transactional_batch(todo!(), todo!())); diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_batch.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_batch.rs index 44aa4c91967..79b61e116a0 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_batch.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_batch.rs @@ -132,10 +132,10 @@ pub async fn batch_mixed_operations() -> Result<(), Box> { }; container_client - .create_item(&partition_key, &item1, None) + .create_item(&partition_key, &item1.id, &item1, None) .await?; container_client - .create_item(&partition_key, &item2, None) + .create_item(&partition_key, &item2.id, &item2, None) .await?; // Now execute a batch with mixed operations @@ -206,7 +206,7 @@ pub async fn batch_atomicity_on_failure() -> Result<(), Box> { }; container_client - .create_item(&partition_key, &item1, None) + .create_item(&partition_key, &item1.id, &item1, None) .await?; // Try to create a batch that will fail (trying to delete a non-existent item) diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs index ebb2908a68c..b6eb42b3364 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs @@ -87,7 +87,7 @@ pub async fn fault_injection_probability_zero_never_fails() -> Result<(), Box Result<(), Box Result<(), Box Result<(), Box let item_id = format!("Item-{}", unique_id); // Create using normal client - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -315,7 +315,7 @@ pub async fn fault_injection_delete_item_fault_crud_succeeds() -> Result<(), Box let mut updated_item = item.clone(); updated_item.value = 100; let upsert_result = fault_container_client - .upsert_item(&pk, &updated_item, None) + .upsert_item(&pk, &item_id, &updated_item, None) .await; assert!( upsert_result.is_ok(), @@ -381,7 +381,7 @@ pub async fn fault_injection_container_specific() -> Result<(), Box> let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -482,7 +482,7 @@ pub async fn fault_injection_multiple_rules_priority() -> Result<(), Box Result<( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -638,7 +638,7 @@ pub async fn fault_injection_first_rule_expired_due_to_end_time() -> Result<(), let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -709,7 +709,7 @@ pub async fn fault_injection_hit_limit_behavior() -> Result<(), Box> let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -775,7 +775,7 @@ pub async fn fault_injection_empty_rules() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -871,7 +871,7 @@ pub async fn fault_injection_metadata_fault_item_ops_succeed() -> Result<(), Box let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - let create_result = fault_container_client.create_item(&pk, &item, None).await; + let create_result = fault_container_client.create_item(&pk, &item_id, &item, None).await; assert!( create_result.is_ok(), "create item should succeed: {:?}", @@ -892,7 +892,7 @@ pub async fn fault_injection_metadata_fault_item_ops_succeed() -> Result<(), Box let mut updated_item = item.clone(); updated_item.value = 999; let upsert_result = fault_container_client - .upsert_item(&pk, &updated_item, None) + .upsert_item(&pk, &item_id, &updated_item, None) .await; assert!( upsert_result.is_ok(), @@ -952,7 +952,7 @@ pub async fn fault_injection_enable_disable_rule() -> Result<(), Box> let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs index 2bad01200be..de34b9ea265 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs @@ -135,7 +135,7 @@ pub async fn item_crud() -> Result<(), Box> { let pk = format!("Partition@1-{}", unique_id); let item_id = format!("Item@1-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client.create_item(&pk, &item_id, &item, None).await?; assert_response( &response, StatusCode::Created, @@ -257,7 +257,7 @@ pub async fn item_read_system_properties() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let create_response = container_client.create_item(&pk, &item, None).await?; + let create_response = container_client.create_item(&pk, &item_id, &item, None).await?; assert_response( &create_response, StatusCode::Created, @@ -316,7 +316,7 @@ pub async fn item_upsert_new() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let upsert_response = container_client.upsert_item(&pk, &item, None).await?; + let upsert_response = container_client.upsert_item(&pk, &item_id, &item, None).await?; assert_response( &upsert_response, StatusCode::Created, @@ -365,8 +365,9 @@ pub async fn item_upsert_existing() -> Result<(), Box> { }; let pk = format!("Partition1-{}", unique_id); + let item_id = format!("Item1-{}", unique_id); - let create_response = container_client.create_item(&pk, &item, None).await?; + let create_response = container_client.create_item(&pk, &item_id, &item, None).await?; assert_response( &create_response, StatusCode::Created, @@ -378,7 +379,7 @@ pub async fn item_upsert_existing() -> Result<(), Box> { item.nested.nested_value = "Updated".into(); let upsert_response = container_client - .upsert_item(&pk, &item, { + .upsert_item(&pk, &item_id, &item, { let mut operation = OperationOptions::default(); operation.content_response_on_write = Some(ContentResponseOnWrite::Enabled); Some(ItemWriteOptions::default().with_operation_options(operation)) @@ -424,7 +425,7 @@ pub async fn item_null_partition_key() -> Result<(), Box> { let item_id = format!("Item1-{}", unique_id); let create_response = container_client - .create_item(PartitionKey::NULL, &item, None) + .create_item(PartitionKey::NULL, &item_id, &item, None) .await?; assert_response( &create_response, @@ -437,7 +438,7 @@ pub async fn item_null_partition_key() -> Result<(), Box> { item.nested.nested_value = "Updated".into(); let upsert_response = container_client - .upsert_item(PartitionKey::NULL, &item, None) + .upsert_item(PartitionKey::NULL, &item_id, &item, None) .await?; assert_response( &upsert_response, @@ -520,7 +521,7 @@ pub async fn item_replace_if_match_etag() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client.create_item(&pk, &item_id, &item, None).await?; assert_response( &response, StatusCode::Created, @@ -610,8 +611,9 @@ pub async fn item_upsert_if_match_etag() -> Result<(), Box> { }; let pk = format!("Partition1-{}", unique_id); + let item_id = format!("Item1-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client.create_item(&pk, &item_id, &item, None).await?; assert_response( &response, StatusCode::Created, @@ -633,6 +635,7 @@ pub async fn item_upsert_if_match_etag() -> Result<(), Box> { let upsert_response = container_client .upsert_item( &pk, + &item_id, &item, Some( ItemWriteOptions::default() @@ -654,6 +657,7 @@ pub async fn item_upsert_if_match_etag() -> Result<(), Box> { let response = container_client .upsert_item( &pk, + &item_id, &item, Some( ItemWriteOptions::default() @@ -701,7 +705,7 @@ pub async fn item_delete_if_match_etag() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client.create_item(&pk, &item_id, &item, None).await?; assert_response( &response, StatusCode::Created, @@ -735,7 +739,7 @@ pub async fn item_delete_if_match_etag() -> Result<(), Box> { ); //Add item again for second delete test - let create_response = container_client.create_item(&pk, &item, None).await?; + let create_response = container_client.create_item(&pk, &item_id, &item, None).await?; assert_response( &create_response, StatusCode::Created, @@ -803,7 +807,7 @@ pub async fn item_undefined_partition_key() -> Result<(), Box> { let item_no_pk_id = format!("Item-NoPK-{}", unique_id); let response = container_client - .create_item(PartitionKey::UNDEFINED, &item_no_pk, None) + .create_item(PartitionKey::UNDEFINED, &item_no_pk_id, &item_no_pk, None) .await?; assert_response( &response, @@ -825,7 +829,7 @@ pub async fn item_undefined_partition_key() -> Result<(), Box> { let item_null_pk_id = format!("Item-NullPK-{}", unique_id); let response = container_client - .create_item(PartitionKey::NULL, &item_null_pk, None) + .create_item(PartitionKey::NULL, &item_null_pk_id, &item_null_pk, None) .await?; assert_response( &response, @@ -844,7 +848,7 @@ pub async fn item_undefined_partition_key() -> Result<(), Box> { let item_with_pk_id = format!("Item-WithPK-{}", unique_id); let response = container_client - .create_item(&pk_value, &item_with_pk, None) + .create_item(&pk_value, &item_with_pk_id, &item_with_pk, None) .await?; assert_response( &response, @@ -955,9 +959,10 @@ pub async fn create_item_duplicate_returns_conflict() -> Result<(), Box Result<(), Box Result<(), Box> { bool_value: false, }; let pk = format!("pk-{}", unique_id); + let item_id = format!("cr-{}", unique_id); let mut operation = OperationOptions::default(); operation.content_response_on_write = Some(ContentResponseOnWrite::Enabled); let options = ItemWriteOptions::default().with_operation_options(operation); let response = container_client - .create_item(&pk, &item, Some(options)) + .create_item(&pk, &item_id, &item, Some(options)) .await?; assert_response( &response, @@ -1053,8 +1059,9 @@ pub async fn create_item_response_metadata() -> Result<(), Box> { bool_value: true, }; let pk = format!("pk-{}", unique_id); + let item_id = format!("meta-{}", unique_id); - let response = container_client.create_item(&pk, &item, None).await?; + let response = container_client.create_item(&pk, &item_id, &item, None).await?; assert_eq!(response.status(), StatusCode::Created); // Session token must be present for session consistency. diff --git a/sdk/cosmos/azure_data_cosmos/tests/framework/test_data.rs b/sdk/cosmos/azure_data_cosmos/tests/framework/test_data.rs index 1a47368f9e6..99ec7849347 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/framework/test_data.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/framework/test_data.rs @@ -65,8 +65,9 @@ pub async fn create_container_with_items( let container_client = db.container_client("TestContainer").await?; for item in items { + let item_id = item.id.clone(); container_client - .create_item(item.partition_key.clone(), item, None) + .create_item(item.partition_key.clone(), &item_id, item, None) .await?; } diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs index 86afec06471..ffb40180cb2 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write.rs @@ -90,6 +90,7 @@ fn create_container_and_write_item<'a>( // This upsert operation triggers a routing decision log in the driver container_client .upsert_item( + "item1", "item1", &serde_json::json!({"id": "item1", "value": "test"}), None, diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs index ad151e362c1..b5fb39f8d25 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs @@ -80,7 +80,7 @@ async fn verify_read_fails_with_injected_error( let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -231,7 +231,7 @@ pub async fn item_read_succeeds_when_fault_targets_create_item() -> Result<(), B let item_id = format!("Item1-{}", unique_id); // Create the item using the normal client (this should succeed) - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -314,7 +314,7 @@ pub async fn fault_injection_read_region_retry_503() -> Result<(), Box Result<() bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); // Transport-generated 503 on a non-idempotent write (upsert) should NOT // be retried — the driver cannot know if the server processed the request. - let result = fault_container_client.upsert_item(&pk, &item, None).await; + let result = fault_container_client.upsert_item(&pk, &item_id, &item, None).await; assert!( result.is_err(), @@ -476,7 +477,7 @@ pub async fn fault_injection_read_region_retry_404_1002() -> Result<(), Box Result<(), Box bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); let response = fault_container_client - .create_item(&pk, &item, None) + .create_item(&pk, &item_id, &item, None) .await .expect("write should succeed via failover to satellite"); @@ -648,7 +650,7 @@ pub async fn fault_injection_read_connection_error_failover() -> Result<(), Box< let item_id = format!("Item-{}", unique_id); // Create item with the normal client - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -739,8 +741,9 @@ pub async fn fault_injection_write_response_timeout_does_not_retry() -> Result<( bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); - let result = fault_container_client.create_item(&pk, &item, None).await; + let result = fault_container_client.create_item(&pk, &item_id, &item, None).await; assert!( result.is_err(), @@ -807,7 +810,7 @@ pub async fn fault_injection_read_response_timeout_retries_to_satellite( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -898,9 +901,10 @@ pub async fn fault_injection_connection_error_reverse_failover() -> Result<(), B bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); let response = fault_container_client - .create_item(&pk, &item, None) + .create_item(&pk, &item_id, &item, None) .await .expect("write should succeed via reverse failover to hub"); @@ -971,7 +975,7 @@ pub async fn fault_injection_connection_error_local_retry_succeeds() -> Result<( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs index 838147408c6..c38dcdfcc65 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs @@ -93,7 +93,7 @@ pub async fn read_cross_region_retry_on_408() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -186,9 +186,10 @@ pub async fn write_no_cross_region_retry_on_408() -> Result<(), Box> bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); // Write should fail with 408 — no cross-region retry for writes - let result = fault_container_client.create_item(&pk, &item, None).await; + let result = fault_container_client.create_item(&pk, &item_id, &item, None).await; let err = result.expect_err("write should fail with 408 and not retry across regions"); assert_eq!( @@ -261,9 +262,10 @@ pub async fn upsert_no_cross_region_retry_on_408() -> Result<(), Box> bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); // Upsert should fail with 408 — no cross-region retry for writes - let result = fault_container_client.upsert_item(&pk, &item, None).await; + let result = fault_container_client.upsert_item(&pk, &item_id, &item, None).await; let err = result.expect_err("upsert should fail with 408 and not retry across regions"); assert_eq!( @@ -333,8 +335,9 @@ pub async fn query_cross_region_retry_on_408() -> Result<(), Box> { bool_value: true, }; let pk = format!("Partition-{}", unique_id); + let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -417,7 +420,7 @@ pub async fn read_cross_region_retry_on_500() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -507,7 +510,7 @@ pub async fn replace_no_cross_region_retry_on_408() -> Result<(), Box let item_id = format!("Item-{}", unique_id); // Create the item first via the non-fault client - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() @@ -599,7 +602,7 @@ pub async fn delete_no_cross_region_retry_on_408() -> Result<(), Box> let item_id = format!("Item-{}", unique_id); // Create the item first via the non-fault client - container_client.create_item(&pk, &item, None).await?; + container_client.create_item(&pk, &item_id, &item, None).await?; let fault_client = run_context .fault_client() diff --git a/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs b/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs index 6487f1b5040..f594cb05f6e 100644 --- a/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs +++ b/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs @@ -303,14 +303,16 @@ pub async fn setup_live() -> (Arc, ItemReference) { // Create the benchmark item if it doesn't already exist. let item_body = format!(r#"{{"id": "{}", "pk": "{}"}}"#, item_id, pk_value); + let item_ref = ItemReference::from_name( + &container_ref, + PartitionKey::from(pk_value.clone()), + item_id.clone(), + ); ignore_conflict( driver .execute_operation( - CosmosOperation::create_item( - container_ref.clone(), - PartitionKey::from(pk_value.clone()), - ) - .with_body(item_body.into_bytes()), + CosmosOperation::create_item(item_ref) + .with_body(item_body.into_bytes()), OperationOptions::default(), ) .await, diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs index d4cd938e5c0..4d2db2efcf4 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs @@ -20,7 +20,8 @@ use crate::{ }, models::{ request_header_names, AccountEndpoint, ActivityId, CosmosOperation, CosmosResponse, - Credential, DefaultConsistencyLevel, OperationType, SessionToken, SubStatusCode, + Credential, DefaultConsistencyLevel, OperationType, ResourceType, SessionToken, + SubStatusCode, }, options::{OperationOptionsView, ReadConsistencyStrategy, ThroughputControlGroupSnapshot}, }; @@ -445,8 +446,19 @@ fn build_transport_request( ctx: &TransportRequestContext<'_>, ) -> azure_core::Result { let resource_ref = operation.resource_reference(); - // Compute both paths in a single pass with a single allocation. - let paths = resource_ref.compute_paths(); + // Create and Upsert are POSTed to the collection feed even though the + // operation carries an ItemReference with the document id. Use the + // feed-path variant so the request URL targets the collection and the + // signing link uses the parent resource. + let paths = if matches!( + operation.operation_type(), + OperationType::Create | OperationType::Upsert + ) && operation.resource_type() == ResourceType::Document + { + resource_ref.compute_feed_paths() + } else { + resource_ref.compute_paths() + }; let url = { let mut base = ctx.routing.selected_url.clone(); let request_path = paths.request_path(); @@ -1249,7 +1261,9 @@ mod tests { #[test] fn build_transport_request_sets_is_upsert_header() { - let operation = CosmosOperation::upsert_item(test_container(), PartitionKey::from("pk1")) + let container = test_container(); + let item = ItemReference::from_name(&container, PartitionKey::from("pk1"), "doc1"); + let operation = CosmosOperation::upsert_item(item) .with_body(b"{}".to_vec()); let routing = test_routing(); @@ -1274,7 +1288,9 @@ mod tests { #[test] fn build_transport_request_omits_is_upsert_header_for_create() { - let operation = CosmosOperation::create_item(test_container(), PartitionKey::from("pk1")) + let container = test_container(); + let item = ItemReference::from_name(&container, PartitionKey::from("pk1"), "doc1"); + let operation = CosmosOperation::create_item(item) .with_body(b"{}".to_vec()); let routing = test_routing(); diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs index a474d2061d1..f5e9af08382 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs @@ -376,9 +376,10 @@ impl CosmosOperation { // ===== Data Plane Factory Methods ===== - /// Creates an item (document) in a container. + /// Creates a new item (document) in a container. /// - /// The `container` and `partition_key` identify where to create the document. + /// The `ItemReference` contains the container, partition key, and item identifier, + /// providing all the information needed for the operation. /// Use `with_body()` to provide the document JSON. /// /// # Example @@ -386,7 +387,7 @@ impl CosmosOperation { /// ```no_run /// use azure_data_cosmos_driver::driver::CosmosDriverRuntime; /// use azure_data_cosmos_driver::models::{ - /// AccountReference, CosmosOperation, ContainerReference, PartitionKey, + /// AccountReference, CosmosOperation, ItemReference, PartitionKey, /// }; /// use azure_data_cosmos_driver::options::OperationOptions; /// use url::Url; @@ -400,10 +401,10 @@ impl CosmosOperation { /// let driver = runtime.get_or_create_driver(account, None).await?; /// let container = driver.resolve_container("my-database", "my-container").await?; /// - /// let pk = PartitionKey::from("pk-value"); + /// let item = ItemReference::from_name(&container, PartitionKey::from("pk-value"), "doc1"); /// let result = driver /// .execute_operation( - /// CosmosOperation::create_item(container, pk) + /// CosmosOperation::create_item(item) /// .with_body(br#"{"id": "doc1", "pk": "pk-value", "data": "hello"}"#.to_vec()), /// OperationOptions::default(), /// ) @@ -411,11 +412,9 @@ impl CosmosOperation { /// # Ok(()) /// # } /// ``` - pub fn create_item(container: ContainerReference, partition_key: PartitionKey) -> Self { - let resource_ref: CosmosResourceReference = CosmosResourceReference::from(container) - .with_resource_type(ResourceType::Document) - .into_feed_reference(); - Self::new(OperationType::Create, resource_ref).with_partition_key(partition_key) + pub fn create_item(item: ItemReference) -> Self { + let partition_key = item.partition_key().clone(); + Self::new(OperationType::Create, item).with_partition_key(partition_key) } /// Reads an item (document) from a container. @@ -466,16 +465,13 @@ impl CosmosOperation { /// Upserts (creates or replaces) an item (document) in a container. /// - /// The Cosmos DB REST API treats upsert as a `POST` to the collection feed - /// (same as create), so this takes a `ContainerReference` and `PartitionKey` - /// rather than an `ItemReference`. + /// The `ItemReference` contains the container, partition key, and item identifier, + /// providing all the information needed for the operation. /// Use `with_body()` to provide the document JSON. /// If an item with the same ID exists, it will be replaced; otherwise, a new item is created. - pub fn upsert_item(container: ContainerReference, partition_key: PartitionKey) -> Self { - let resource_ref: CosmosResourceReference = CosmosResourceReference::from(container) - .with_resource_type(ResourceType::Document) - .into_feed_reference(); - Self::new(OperationType::Upsert, resource_ref).with_partition_key(partition_key) + pub fn upsert_item(item: ItemReference) -> Self { + let partition_key = item.partition_key().clone(); + Self::new(OperationType::Upsert, item).with_partition_key(partition_key) } /// Replaces an existing item (document) in a container. diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs index 8f909d89a01..d6f45de0c91 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs @@ -158,6 +158,33 @@ impl CosmosResourceReference { self } + /// Returns the item identifier (name), if this reference has one. + pub(crate) fn item_id(&self) -> Option<&str> { + self.id.as_ref().and_then(|id| id.name()) + } + + /// Computes paths treating this reference as a feed operation. + /// + /// Used by Create and Upsert which carry an [`ItemReference`] (with an + /// item id) but still POST to the parent (collection) URL and sign + /// against the parent resource. + pub(crate) fn compute_feed_paths(&self) -> ResourcePaths { + // Temporarily treat the reference as a feed for path computation. + let parent = self.parent_link_cow(); + let segment = self.resource_type.path_segment(); + let buf = if parent.is_empty() { + format!("/{}", segment) + } else { + format!("{}/{}", parent, segment) + }; + let signing_end = if parent.is_empty() { 1 } else { parent.len() }; + ResourcePaths { + buf, + signing_end, + signing_override: None, + } + } + /// Returns the resource link used for authorization signing. /// /// For feed operations this is the **parent** resource's link (because diff --git a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs index de3810bc294..1e8affc5620 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs @@ -42,7 +42,7 @@ pub async fn fault_injection_probability_zero_never_fails() -> Result<(), Box Result<(), .await?; let item_json = br#"{"id": "item1", "pk": "pk1", "value": "test"}"#; - context.create_item(&container, "pk1", item_json).await?; + context.create_item(&container, "item1", "pk1", item_json).await?; // With probability 1.0, the read should fail let read_result = context.read_item(&container, "item1", "pk1").await; @@ -163,7 +163,7 @@ pub async fn fault_injection_operation_type_filter() -> Result<(), Box Result<(), Box< .await?; let item_json = br#"{"id": "item1", "pk": "pk1", "value": "test"}"#; - context.create_item(&container, "pk1", item_json).await?; + context.create_item(&container, "item1", "pk1", item_json).await?; // Execute reads to consume the hit limit. // Due to internal retries, the limit may be exhausted within fewer @@ -305,7 +305,7 @@ pub async fn fault_injection_connection_error() -> Result<(), Box> { .await?; let item_json = br#"{"id": "item1", "pk": "pk1", "value": "test"}"#; - context.create_item(&container, "pk1", item_json).await?; + context.create_item(&container, "item1", "pk1", item_json).await?; // With a connection error injected, the read should fail let read_result = context.read_item(&container, "item1", "pk1").await; diff --git a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_item_operations.rs b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_item_operations.rs index dca95e9ab52..36e83165271 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_item_operations.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_item_operations.rs @@ -54,7 +54,7 @@ pub async fn create_and_read_item() -> Result<(), Box> { // Create the item let create_result = context - .create_item(&container, item.pk.clone(), &item_json) + .create_item(&container, &item.id, item.pk.clone(), &item_json) .await?; // Validate create diagnostics @@ -128,7 +128,7 @@ pub async fn control_plane_uses_metadata_pipeline() -> Result<(), Box let item_json = serde_json::to_vec(&test_item)?; let result = context - .create_item(&container, test_item.pk.clone(), &item_json) + .create_item(&container, &test_item.id, test_item.pk.clone(), &item_json) .await?; // Verify item creation succeeded @@ -164,7 +164,7 @@ pub async fn diagnostics_contain_expected_fields() -> Result<(), Box> let item_json = serde_json::to_vec(&item)?; let result = context - .create_item(&container, item.pk.clone(), &item_json) + .create_item(&container, &item.id, item.pk.clone(), &item_json) .await?; let diagnostics = result.diagnostics(); diff --git a/sdk/cosmos/azure_data_cosmos_driver/tests/framework/test_client.rs b/sdk/cosmos/azure_data_cosmos_driver/tests/framework/test_client.rs index db1f5bb65fa..6657157d9e2 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/tests/framework/test_client.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/tests/framework/test_client.rs @@ -338,6 +338,7 @@ impl DriverTestRunContext { pub async fn create_item( &self, container: &ContainerReference, + item_id: &str, partition_key: impl Into, body: &[u8], ) -> Result> { @@ -348,8 +349,8 @@ impl DriverTestRunContext { .await?; let pk = partition_key.into(); - let operation = - CosmosOperation::create_item(container.clone(), pk).with_body(body.to_vec()); + let item_ref = ItemReference::from_name(container, pk, item_id.to_owned()); + let operation = CosmosOperation::create_item(item_ref).with_body(body.to_vec()); let result = driver .execute_operation(operation, OperationOptions::default()) diff --git a/sdk/cosmos/azure_data_cosmos_perf/src/operations/create_item.rs b/sdk/cosmos/azure_data_cosmos_perf/src/operations/create_item.rs index 54602c223fa..7af0a13ee8b 100644 --- a/sdk/cosmos/azure_data_cosmos_perf/src/operations/create_item.rs +++ b/sdk/cosmos/azure_data_cosmos_perf/src/operations/create_item.rs @@ -49,7 +49,7 @@ impl Operation for CreateItemOperation { }; container - .create_item(&item.partition_key, &item, self.options.clone()) + .create_item(&item.partition_key, &id, &item, self.options.clone()) .await?; self.items.push(SeededItem { id, partition_key }); diff --git a/sdk/cosmos/azure_data_cosmos_perf/src/operations/upsert_item.rs b/sdk/cosmos/azure_data_cosmos_perf/src/operations/upsert_item.rs index 6338a08c9bb..0b804483d3f 100644 --- a/sdk/cosmos/azure_data_cosmos_perf/src/operations/upsert_item.rs +++ b/sdk/cosmos/azure_data_cosmos_perf/src/operations/upsert_item.rs @@ -44,7 +44,7 @@ impl Operation for UpsertItemOperation { }; container - .upsert_item(&item.partition_key, &item, self.options.clone()) + .upsert_item(&item.partition_key, &seeded.id, &item, self.options.clone()) .await?; Ok(()) } From 6e6c61481ad6318171e74d20aef0efab68eb2f7e Mon Sep 17 00:00:00 2001 From: Fabian Meiswinkel Date: Mon, 27 Apr 2026 11:55:12 +0000 Subject: [PATCH 2/7] Fix formatting --- .../examples/cosmos/create.rs | 4 +- .../examples/cosmos/upsert.rs | 4 +- .../emulator_tests/cosmos_fault_injection.rs | 48 ++++++++++++++----- .../tests/emulator_tests/cosmos_items.rs | 44 ++++++++++++----- .../cosmos_multi_write_fault_injection.rs | 36 ++++++++++---- .../cosmos_multi_write_retry_policies.rs | 28 ++++++++--- .../src/driver/pipeline/operation_pipeline.rs | 6 +-- .../src/models/cosmos_resource_reference.rs | 5 -- .../emulator_tests/driver_fault_injection.rs | 8 +++- .../azure_data_cosmos_perf/src/runner.rs | 7 ++- sdk/cosmos/azure_data_cosmos_perf/src/seed.rs | 2 +- 11 files changed, 137 insertions(+), 55 deletions(-) diff --git a/sdk/cosmos/azure_data_cosmos/examples/cosmos/create.rs b/sdk/cosmos/azure_data_cosmos/examples/cosmos/create.rs index 090a93e9797..c2b77538909 100644 --- a/sdk/cosmos/azure_data_cosmos/examples/cosmos/create.rs +++ b/sdk/cosmos/azure_data_cosmos/examples/cosmos/create.rs @@ -99,7 +99,9 @@ impl CreateCommand { None }; - let response = container_client.create_item(pk, &item_id, item, options).await?; + let response = container_client + .create_item(pk, &item_id, item, options) + .await?; println!("Created item successfully"); diff --git a/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs b/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs index 1879d1f02f6..dc033906874 100644 --- a/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs +++ b/sdk/cosmos/azure_data_cosmos/examples/cosmos/upsert.rs @@ -50,7 +50,9 @@ impl UpsertCommand { None }; - let response = container_client.upsert_item(pk, &self.item_id, item, options).await?; + let response = container_client + .upsert_item(pk, &self.item_id, item, options) + .await?; println!("Item updated successfully"); if self.show_updated { diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs index b6eb42b3364..75213af2319 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_fault_injection.rs @@ -87,7 +87,9 @@ pub async fn fault_injection_probability_zero_never_fails() -> Result<(), Box Result<(), Box Result<(), Box Result<(), Box let item_id = format!("Item-{}", unique_id); // Create using normal client - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -381,7 +389,9 @@ pub async fn fault_injection_container_specific() -> Result<(), Box> let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -482,7 +492,9 @@ pub async fn fault_injection_multiple_rules_priority() -> Result<(), Box Result<( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -638,7 +652,9 @@ pub async fn fault_injection_first_rule_expired_due_to_end_time() -> Result<(), let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -709,7 +725,9 @@ pub async fn fault_injection_hit_limit_behavior() -> Result<(), Box> let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -775,7 +793,9 @@ pub async fn fault_injection_empty_rules() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -871,7 +891,9 @@ pub async fn fault_injection_metadata_fault_item_ops_succeed() -> Result<(), Box let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - let create_result = fault_container_client.create_item(&pk, &item_id, &item, None).await; + let create_result = fault_container_client + .create_item(&pk, &item_id, &item, None) + .await; assert!( create_result.is_ok(), "create item should succeed: {:?}", @@ -952,7 +974,9 @@ pub async fn fault_injection_enable_disable_rule() -> Result<(), Box> let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs index de34b9ea265..528a156363e 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_items.rs @@ -135,7 +135,9 @@ pub async fn item_crud() -> Result<(), Box> { let pk = format!("Partition@1-{}", unique_id); let item_id = format!("Item@1-{}", unique_id); - let response = container_client.create_item(&pk, &item_id, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &response, StatusCode::Created, @@ -257,7 +259,9 @@ pub async fn item_read_system_properties() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let create_response = container_client.create_item(&pk, &item_id, &item, None).await?; + let create_response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &create_response, StatusCode::Created, @@ -316,7 +320,9 @@ pub async fn item_upsert_new() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let upsert_response = container_client.upsert_item(&pk, &item_id, &item, None).await?; + let upsert_response = container_client + .upsert_item(&pk, &item_id, &item, None) + .await?; assert_response( &upsert_response, StatusCode::Created, @@ -367,7 +373,9 @@ pub async fn item_upsert_existing() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let create_response = container_client.create_item(&pk, &item_id, &item, None).await?; + let create_response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &create_response, StatusCode::Created, @@ -521,7 +529,9 @@ pub async fn item_replace_if_match_etag() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let response = container_client.create_item(&pk, &item_id, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &response, StatusCode::Created, @@ -613,7 +623,9 @@ pub async fn item_upsert_if_match_etag() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let response = container_client.create_item(&pk, &item_id, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &response, StatusCode::Created, @@ -705,7 +717,9 @@ pub async fn item_delete_if_match_etag() -> Result<(), Box> { let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - let response = container_client.create_item(&pk, &item_id, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &response, StatusCode::Created, @@ -739,7 +753,9 @@ pub async fn item_delete_if_match_etag() -> Result<(), Box> { ); //Add item again for second delete test - let create_response = container_client.create_item(&pk, &item_id, &item, None).await?; + let create_response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_response( &create_response, StatusCode::Created, @@ -962,7 +978,9 @@ pub async fn create_item_duplicate_returns_conflict() -> Result<(), Box Result<(), Box Result<(), Box> { let pk = format!("pk-{}", unique_id); let item_id = format!("meta-{}", unique_id); - let response = container_client.create_item(&pk, &item_id, &item, None).await?; + let response = container_client + .create_item(&pk, &item_id, &item, None) + .await?; assert_eq!(response.status(), StatusCode::Created); // Session token must be present for session consistency. diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs index b5fb39f8d25..95de25d2c72 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_fault_injection.rs @@ -80,7 +80,9 @@ async fn verify_read_fails_with_injected_error( let pk = format!("Partition1-{}", unique_id); let item_id = format!("Item1-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -231,7 +233,9 @@ pub async fn item_read_succeeds_when_fault_targets_create_item() -> Result<(), B let item_id = format!("Item1-{}", unique_id); // Create the item using the normal client (this should succeed) - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -314,7 +318,9 @@ pub async fn fault_injection_read_region_retry_503() -> Result<(), Box Result<() // Transport-generated 503 on a non-idempotent write (upsert) should NOT // be retried — the driver cannot know if the server processed the request. - let result = fault_container_client.upsert_item(&pk, &item_id, &item, None).await; + let result = fault_container_client + .upsert_item(&pk, &item_id, &item, None) + .await; assert!( result.is_err(), @@ -477,7 +485,9 @@ pub async fn fault_injection_read_region_retry_404_1002() -> Result<(), Box Result<(), Box< let item_id = format!("Item-{}", unique_id); // Create item with the normal client - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -743,7 +755,9 @@ pub async fn fault_injection_write_response_timeout_does_not_retry() -> Result<( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - let result = fault_container_client.create_item(&pk, &item_id, &item, None).await; + let result = fault_container_client + .create_item(&pk, &item_id, &item, None) + .await; assert!( result.is_err(), @@ -810,7 +824,9 @@ pub async fn fault_injection_read_response_timeout_retries_to_satellite( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -975,7 +991,9 @@ pub async fn fault_injection_connection_error_local_retry_succeeds() -> Result<( let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() diff --git a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs index c38dcdfcc65..7cba26d28f1 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/multi_write_tests/cosmos_multi_write_retry_policies.rs @@ -93,7 +93,9 @@ pub async fn read_cross_region_retry_on_408() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -189,7 +191,9 @@ pub async fn write_no_cross_region_retry_on_408() -> Result<(), Box> let item_id = format!("Item-{}", unique_id); // Write should fail with 408 — no cross-region retry for writes - let result = fault_container_client.create_item(&pk, &item_id, &item, None).await; + let result = fault_container_client + .create_item(&pk, &item_id, &item, None) + .await; let err = result.expect_err("write should fail with 408 and not retry across regions"); assert_eq!( @@ -265,7 +269,9 @@ pub async fn upsert_no_cross_region_retry_on_408() -> Result<(), Box> let item_id = format!("Item-{}", unique_id); // Upsert should fail with 408 — no cross-region retry for writes - let result = fault_container_client.upsert_item(&pk, &item_id, &item, None).await; + let result = fault_container_client + .upsert_item(&pk, &item_id, &item, None) + .await; let err = result.expect_err("upsert should fail with 408 and not retry across regions"); assert_eq!( @@ -337,7 +343,9 @@ pub async fn query_cross_region_retry_on_408() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -420,7 +428,9 @@ pub async fn read_cross_region_retry_on_500() -> Result<(), Box> { let pk = format!("Partition-{}", unique_id); let item_id = format!("Item-{}", unique_id); - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -510,7 +520,9 @@ pub async fn replace_no_cross_region_retry_on_408() -> Result<(), Box let item_id = format!("Item-{}", unique_id); // Create the item first via the non-fault client - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() @@ -602,7 +614,9 @@ pub async fn delete_no_cross_region_retry_on_408() -> Result<(), Box> let item_id = format!("Item-{}", unique_id); // Create the item first via the non-fault client - container_client.create_item(&pk, &item_id, &item, None).await?; + container_client + .create_item(&pk, &item_id, &item, None) + .await?; let fault_client = run_context .fault_client() diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs index 4d2db2efcf4..0f0358e3bee 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs @@ -1263,8 +1263,7 @@ mod tests { fn build_transport_request_sets_is_upsert_header() { let container = test_container(); let item = ItemReference::from_name(&container, PartitionKey::from("pk1"), "doc1"); - let operation = CosmosOperation::upsert_item(item) - .with_body(b"{}".to_vec()); + let operation = CosmosOperation::upsert_item(item).with_body(b"{}".to_vec()); let routing = test_routing(); let activity_id = ActivityId::from_string("default-activity".to_string()); @@ -1290,8 +1289,7 @@ mod tests { fn build_transport_request_omits_is_upsert_header_for_create() { let container = test_container(); let item = ItemReference::from_name(&container, PartitionKey::from("pk1"), "doc1"); - let operation = CosmosOperation::create_item(item) - .with_body(b"{}".to_vec()); + let operation = CosmosOperation::create_item(item).with_body(b"{}".to_vec()); let routing = test_routing(); let activity_id = ActivityId::from_string("default-activity".to_string()); diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs index d6f45de0c91..7900bd0a838 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs @@ -158,11 +158,6 @@ impl CosmosResourceReference { self } - /// Returns the item identifier (name), if this reference has one. - pub(crate) fn item_id(&self) -> Option<&str> { - self.id.as_ref().and_then(|id| id.name()) - } - /// Computes paths treating this reference as a feed operation. /// /// Used by Create and Upsert which carry an [`ItemReference`] (with an diff --git a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs index 1e8affc5620..439377273ab 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/tests/emulator_tests/driver_fault_injection.rs @@ -104,7 +104,9 @@ pub async fn fault_injection_service_unavailable_causes_failure() -> Result<(), .await?; let item_json = br#"{"id": "item1", "pk": "pk1", "value": "test"}"#; - context.create_item(&container, "item1", "pk1", item_json).await?; + context + .create_item(&container, "item1", "pk1", item_json) + .await?; // With probability 1.0, the read should fail let read_result = context.read_item(&container, "item1", "pk1").await; @@ -305,7 +307,9 @@ pub async fn fault_injection_connection_error() -> Result<(), Box> { .await?; let item_json = br#"{"id": "item1", "pk": "pk1", "value": "test"}"#; - context.create_item(&container, "item1", "pk1", item_json).await?; + context + .create_item(&container, "item1", "pk1", item_json) + .await?; // With a connection error injected, the read should fail let read_result = context.read_item(&container, "item1", "pk1").await; diff --git a/sdk/cosmos/azure_data_cosmos_perf/src/runner.rs b/sdk/cosmos/azure_data_cosmos_perf/src/runner.rs index fdda04c78f8..1d24329782e 100644 --- a/sdk/cosmos/azure_data_cosmos_perf/src/runner.rs +++ b/sdk/cosmos/azure_data_cosmos_perf/src/runner.rs @@ -404,7 +404,7 @@ async fn upsert_results( }; if let Err(e) = container - .upsert_item(&result.partition_key, &result, None) + .upsert_item(&result.partition_key, &result.id, &result, None) .await { eprintln!("Warning: failed to upsert perf result: {e}"); @@ -439,7 +439,10 @@ async fn upsert_error( error_message: format!("{error}"), source_message: error_source_chain(error), }; - if let Err(e) = container.upsert_item(&doc.partition_key, &doc, None).await { + if let Err(e) = container + .upsert_item(&doc.partition_key, &id, &doc, None) + .await + { eprintln!("Warning: failed to upsert error result: {e}"); } } diff --git a/sdk/cosmos/azure_data_cosmos_perf/src/seed.rs b/sdk/cosmos/azure_data_cosmos_perf/src/seed.rs index 917e8b22af4..f3ab9844836 100644 --- a/sdk/cosmos/azure_data_cosmos_perf/src/seed.rs +++ b/sdk/cosmos/azure_data_cosmos_perf/src/seed.rs @@ -103,7 +103,7 @@ pub async fn seed_container( }; let result = container - .upsert_item(&item.partition_key, &item, None) + .upsert_item(&item.partition_key, &item.id, &item, None) .await; if result.is_ok() { From a9617c3acfe6cf4aa715ba212efc56e8cd636d3e Mon Sep 17 00:00:00 2001 From: Fabian Meiswinkel Date: Mon, 27 Apr 2026 12:08:19 +0000 Subject: [PATCH 3/7] Update lib.rs --- sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs b/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs index f594cb05f6e..a52b728396b 100644 --- a/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs +++ b/sdk/cosmos/azure_data_cosmos_benchmarks/src/lib.rs @@ -311,8 +311,7 @@ pub async fn setup_live() -> (Arc, ItemReference) { ignore_conflict( driver .execute_operation( - CosmosOperation::create_item(item_ref) - .with_body(item_body.into_bytes()), + CosmosOperation::create_item(item_ref).with_body(item_body.into_bytes()), OperationOptions::default(), ) .await, From 66a3bed202dec4eaf1bf78b73e4a498ec35e7888 Mon Sep 17 00:00:00 2001 From: Fabian Meiswinkel Date: Mon, 27 Apr 2026 12:30:15 +0000 Subject: [PATCH 4/7] Update cosmos_resource_reference.rs --- .../src/models/cosmos_resource_reference.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs index 7900bd0a838..245942ebf0e 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs @@ -35,8 +35,8 @@ pub(crate) struct ResourcePaths { buf: String, /// Byte index in `buf` where the signing link ends (exclusive). /// - /// For non-feed: `buf.len()` → signing link = `buf[1..]` - /// For feed: `parent.len()` → signing link = `buf[1..signing_end]` + /// For non-feed: `buf.len()` → signing link = `buf[1..]` + /// For feed: `parent.len()` → signing link = `buf[1..signing_end]` /// Always `>= 1` when `buf` is non-empty (skips the leading `/`). signing_end: usize, /// Signing link override for offer resources. From 2001d59b63e0f7264f817467bf07f4840870e538 Mon Sep 17 00:00:00 2001 From: Fabian Meiswinkel Date: Mon, 27 Apr 2026 17:34:12 +0000 Subject: [PATCH 5/7] Update CHANGELOG.md --- sdk/cosmos/azure_data_cosmos/CHANGELOG.md | 1 + 1 file changed, 1 insertion(+) diff --git a/sdk/cosmos/azure_data_cosmos/CHANGELOG.md b/sdk/cosmos/azure_data_cosmos/CHANGELOG.md index f0e787e5e92..c8483df62a7 100644 --- a/sdk/cosmos/azure_data_cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure_data_cosmos/CHANGELOG.md @@ -14,6 +14,7 @@ ### Breaking Changes +- `ContainerClient::create_item()` and `ContainerClient::upsert_item()` now require an `item_id: &str` parameter (same pattern as `replace_item` and `read_item`). The item id is passed to the driver via `ItemReference` so the body never needs to be parsed to extract the document id. - Renamed `replace_throughput` to `begin_replace_throughput` on `ContainerClient` and `DatabaseClient`. The return type changed from `ResourceResponse` to `ThroughputPoller`. ([#4096](https://github.com/Azure/azure-sdk-for-rust/pull/4096)) - Removed `CreateDatabaseOptions::with_throughput()`. Database-level shared throughput provisioning is no longer supported through the SDK. Use container-level throughput instead. ([#4147](https://github.com/Azure/azure-sdk-for-rust/pull/4147)) From 03ea5cb69f7716ca9bdbe7f8e88d5ab262a536fe Mon Sep 17 00:00:00 2001 From: Fabian Meiswinkel Date: Mon, 27 Apr 2026 18:07:42 +0000 Subject: [PATCH 6/7] Code review comments --- sdk/cosmos/.cspell.json | 1 + .../src/clients/container_client.rs | 2 ++ .../src/driver/pipeline/operation_pipeline.rs | 32 +++++++++---------- .../src/models/cosmos_operation.rs | 18 +++++++++++ .../src/models/cosmos_resource_reference.rs | 29 +++++++++++++++++ 5 files changed, 66 insertions(+), 16 deletions(-) diff --git a/sdk/cosmos/.cspell.json b/sdk/cosmos/.cspell.json index 3cd11f77289..1eff6010f47 100644 --- a/sdk/cosmos/.cspell.json +++ b/sdk/cosmos/.cspell.json @@ -96,6 +96,7 @@ "PPAF", "PPCB", "pushback", + "pyroscope", "qname", "RAII", "readfeed", diff --git a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs index 108e2c16e17..f3fdf385f4f 100644 --- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs +++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs @@ -254,6 +254,7 @@ impl ContainerClient { /// /// # Arguments /// * `partition_key` - The partition key of the new item. + /// * `item_id` - The id of the new item. /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`](serde::Deserialize) /// * `options` - Optional parameters for the request /// @@ -453,6 +454,7 @@ impl ContainerClient { /// /// # Arguments /// * `partition_key` - The partition key of the item to create or replace. + /// * `item_id` - The id of the item to create or replace. /// * `item` - The item to create. The type must implement [`Serialize`] and [`Deserialize`](serde::Deserialize) /// * `options` - Optional parameters for the request /// diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs index 0f0358e3bee..e179abd85ed 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/pipeline/operation_pipeline.rs @@ -20,8 +20,7 @@ use crate::{ }, models::{ request_header_names, AccountEndpoint, ActivityId, CosmosOperation, CosmosResponse, - Credential, DefaultConsistencyLevel, OperationType, ResourceType, SessionToken, - SubStatusCode, + Credential, DefaultConsistencyLevel, OperationType, SessionToken, SubStatusCode, }, options::{OperationOptionsView, ReadConsistencyStrategy, ThroughputControlGroupSnapshot}, }; @@ -445,20 +444,7 @@ fn build_transport_request( custom_headers: Option<&std::collections::HashMap>, ctx: &TransportRequestContext<'_>, ) -> azure_core::Result { - let resource_ref = operation.resource_reference(); - // Create and Upsert are POSTed to the collection feed even though the - // operation carries an ItemReference with the document id. Use the - // feed-path variant so the request URL targets the collection and the - // signing link uses the parent resource. - let paths = if matches!( - operation.operation_type(), - OperationType::Create | OperationType::Upsert - ) && operation.resource_type() == ResourceType::Document - { - resource_ref.compute_feed_paths() - } else { - resource_ref.compute_paths() - }; + let paths = operation.compute_resource_paths(); let url = { let mut base = ctx.routing.selected_url.clone(); let request_path = paths.request_path(); @@ -1283,6 +1269,13 @@ mod tests { .get_optional_str(&HeaderName::from_static("x-ms-documentdb-is-upsert")) .expect("is-upsert header should be set"); assert_eq!(is_upsert, "true"); + + // Upsert targets the collection feed URL, not the individual document. + assert_eq!( + request.url.path(), + "/dbs/testdb/colls/testcontainer/docs", + "upsert should POST to the collection feed, not /docs/doc1" + ); } #[test] @@ -1311,6 +1304,13 @@ mod tests { .is_none(), "is-upsert header should not be set for create" ); + + // Create targets the collection feed URL, not the individual document. + assert_eq!( + request.url.path(), + "/dbs/testdb/colls/testcontainer/docs", + "create should POST to the collection feed, not /docs/doc1" + ); } #[test] diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs index f5e9af08382..0df26573147 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_operation.rs @@ -86,6 +86,24 @@ impl CosmosOperation { &self.resource_reference } + /// Computes the request path and signing link for this operation. + /// + /// Create and Upsert document operations use feed-style paths (targeting + /// the collection URL) even though they carry an item id, because the + /// Cosmos DB REST API POSTs these to the collection feed. All other + /// operations use the standard resource paths. + pub(crate) fn compute_resource_paths(&self) -> crate::models::ResourcePaths { + if matches!( + self.operation_type, + OperationType::Create | OperationType::Upsert + ) && self.resource_type == ResourceType::Document + { + self.resource_reference.compute_feed_paths() + } else { + self.resource_reference.compute_paths() + } + } + /// Returns the container for this operation, if applicable. /// /// Returns `None` for account-level and database-level operations. diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs index 245942ebf0e..cb6fc035ac2 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_resource_reference.rs @@ -784,6 +784,35 @@ mod tests { assert_compute_paths_consistent(&r); } + #[test] + fn compute_feed_paths_item_reference() { + // An ItemReference carries the document id, but compute_feed_paths + // must produce the same feed-style paths as compute_paths on a + // feed reference (without the item id in the URL). + let item = ItemReference::from_name(&test_container(), PartitionKey::from("pk1"), "doc1"); + let r: CosmosResourceReference = item.into(); + + let feed_paths = r.compute_feed_paths(); + assert_eq!( + feed_paths.request_path(), + "/dbs/testdb/colls/testcontainer/docs", + "request path should target the collection feed, not the individual document" + ); + assert_eq!( + feed_paths.signing_link(), + "dbs/testdb/colls/testcontainer", + "signing link should be the parent container path" + ); + + // Verify consistency with compute_paths on an equivalent feed reference. + let feed_ref = CosmosResourceReference::from(test_container()) + .with_resource_type(ResourceType::Document) + .into_feed_reference(); + let expected = feed_ref.compute_paths(); + assert_eq!(feed_paths.request_path(), expected.request_path()); + assert_eq!(feed_paths.signing_link(), expected.signing_link()); + } + #[test] fn compute_paths_offer_uses_signing_override() { let account = test_account(); From ad18e629da4702e4725bca67ab3373740ffe2911 Mon Sep 17 00:00:00 2001 From: Fabian Meiswinkel Date: Mon, 27 Apr 2026 19:06:48 +0000 Subject: [PATCH 7/7] Fixes build issues --- sdk/cosmos/.cspell.json | 20 ++++++++++++++++++-- sdk/cosmos/azure_data_cosmos/README.md | 2 +- 2 files changed, 19 insertions(+), 3 deletions(-) diff --git a/sdk/cosmos/.cspell.json b/sdk/cosmos/.cspell.json index 1eff6010f47..117ff7d1c0e 100644 --- a/sdk/cosmos/.cspell.json +++ b/sdk/cosmos/.cspell.json @@ -13,6 +13,7 @@ "autoscale", "azurecosmos", "backoff", + "backpressure", "bytearray", "canadacentral", "canadaeast", @@ -26,9 +27,11 @@ "cloneable", "colls", "cosmosclient", + "cutover", "Daad", "dedicatedgateway", "derefs", + "dhat", "Dmaster", "documentdb", "dotproduct", @@ -40,10 +43,12 @@ "enablepreview", "EUAP", "euclidian", + "evals", "fabianm", "failback", "failovers", "FILETIME", + "flamegraph", "fract", "francecentral", "francesouth", @@ -54,6 +59,7 @@ "germanywestcentral", "hostnames", "hotfixes", + "idents", "IMDS", "inclusivity", "isquery", @@ -62,7 +68,10 @@ "keepalive", "koreacentral", "koreasouth", + "Kusto", "linearizability", + "LLZA", + "memcheck", "MEMORYSTATUSEX", "moka", "multihash", @@ -86,6 +95,11 @@ "othercoll", "partitionkey", "partitionkeyrangeid", + "perfcontainer", + "perfdb", + "perfresults", + "PIDS", + "Pigw", "Pkrange", "pkranges", "pksysdocs", @@ -98,16 +112,18 @@ "pushback", "pyroscope", "qname", + "qself", "RAII", "readfeed", "replicaset", - "reqs", "Replicaset", + "reqs", "Retriable", "retryable", "rfind", "RNTBD", "roundtrips", + "RUPM", "rwcache", "sess", "southafricanorth", @@ -180,4 +196,4 @@ ] } ] -} +} \ No newline at end of file diff --git a/sdk/cosmos/azure_data_cosmos/README.md b/sdk/cosmos/azure_data_cosmos/README.md index c5c7c4066e7..8d68b68acd5 100644 --- a/sdk/cosmos/azure_data_cosmos/README.md +++ b/sdk/cosmos/azure_data_cosmos/README.md @@ -108,7 +108,7 @@ async fn example(cosmos_client: CosmosClient) -> Result<(), Box