[Cosmos] Add additional response headers (#3960)

## Summary

Adds type-safe response types for Cosmos DB operations with access to
additional response headers for diagnostics and metrics. Fixes #3905.

## Design

Uses dedicated wrapper types for each operation category. A design spec
is at `sdk/cosmos/azure_data_cosmos/docs/RESPONSE_METADATA_SPEC.md`.

### Response Types

| Type | Operations | Extra Fields |
|------|-----------|-------------|
| `ItemResponse<T>` | create/read/replace/upsert/delete item | `etag()
-> Option<&Etag>` |
| `ResourceResponse<T>` | create/read/delete database/container,
throughput | (future-proof placeholder) |
| `BatchResponse` | execute_transactional_batch | `etag() ->
Option<&Etag>` |
| `QueryFeedPage<T>` | query_items, query_containers, query_databases |
`index_metrics()`, `query_metrics()` |
| `FeedPage<T>` | generic feed page (reusable for future read-many,
change feed) | common fields only |
| `CosmosDiagnostics` | all operations (via `diagnostics()`) |
`activity_id()`, `server_duration_ms()` |

### Principle

Each operation category gets a dedicated wrapper type. Common accessors
(`request_charge()`, `session_token()`, `diagnostics()`) are on all
types. Operation-specific fields are only on the relevant type.
`QueryFeedPage<T>` composes over `FeedPage<T>`, adding query-specific
metadata. `CosmosResponse<T>` is internal (`pub(crate)`).

## Driver Changes (`azure_data_cosmos_driver`)

- Added `index_metrics`, `query_metrics`, `server_duration_ms`, and
`lsn` fields to `CosmosResponseHeaders`.
- `index_metrics` is base64-decoded (matching Java/.NET) with
`tracing::warn!` on decode failure.
- `server_duration_ms` filters non-finite and negative values.
- Added `server_duration_ms` to `RequestDiagnostics`, populated from
response headers in transport pipeline.
- Made `from_headers()` public for cross-crate access.

## SDK Changes (`azure_data_cosmos`)

- `ItemResponse<T>` wraps `CosmosResponse<T>` with `etag()` using
`azure_core::http::Etag`.
- `ResourceResponse<T>` wraps `CosmosResponse<T>` for resource
management operations.
- `BatchResponse` wraps `CosmosResponse<TransactionalBatchResponse>`
with batch-level `etag()`.
- `QueryFeedPage<T>` composes over `FeedPage<T>`, adding
`index_metrics()` and `query_metrics()`.
- `FeedPage<T>` is a public generic type with common feed fields,
reusable for future read-many and change feed.
- `CosmosDiagnostics` provides `activity_id()` and
`server_duration_ms()` on all response types.
- SDK delegates all header parsing to the driver's
`CosmosResponseHeaders`.

## Tests

- Driver unit tests: base64 decode, invalid base64, NaN/inf/negative
filtering, JSON serialization.
- SDK unit tests: wrapper types, diagnostics accessors, etag with `Etag`
type, edge cases.
- Integration tests: `assert_response` helper validates
`diagnostics().activity_id()` and `diagnostics().server_duration_ms()`
on all point operations; `query_returns_index_and_query_metrics`
validates query-specific metadata with opt-in headers.

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: tvaron3

Cargo.lock

@@ -115,6 +115,8 @@
         "pksysdocs",
         "pluggable",
         "pointee",
+        "populateindexmetrics",
+        "populatequerymetrics",
         "PPAF",
         "PPCB",
         "purgeable",

sdk/cosmos/.cspell.json

@@ -8,6 +8,8 @@
 
 ### Breaking Changes
 
+- Client methods now return dedicated response types instead of `CosmosResponse<T>`: `ItemResponse<T>` for point operations, `ResourceResponse<T>` for resource management, `BatchResponse` for transactional batch, and `QueryFeedPage<T>` for query pages. `etag()` returns `Option<&Etag>` instead of `Option<&str>`, and `activity_id()` / `server_duration_ms()` are accessed via `response.diagnostics()`. ([#3960](https://github.com/Azure/azure-sdk-for-rust/pull/3960))
+- `FeedPage::deconstruct()` has been removed. Use `into_items()`, `continuation()`, `headers()`, and `diagnostics()` instead. ([#3960](https://github.com/Azure/azure-sdk-for-rust/pull/3960))
 - Replaced `CosmosClientBuilder::with_application_region()` with a mandatory `RoutingStrategy` parameter on `build()`. Use `RoutingStrategy::ProximityTo(region)` to specify the application region. Also removed `CosmosClientOptions::with_application_region()`. ([#3889](https://github.com/Azure/azure-sdk-for-rust/pull/3889))
 - Changed `default_ttl` and `analytical_storage_ttl` fields on `ContainerProperties` from `Option<Duration>` to `TimeToLive`, a new enum with variants `Forever`, `NoDefault`, and `Seconds(u32)`, to correctly handle the `-1` wire value (TTL enabled with no default expiration).
 - `DatabaseClient::container_client()` now returns `azure_core::Result<ContainerClient>`, eagerly resolving container metadata (RID, partition key definition) at construction time. ([#4005](https://github.com/Azure/azure-sdk-for-rust/pull/4005))
@@ -19,7 +21,6 @@
 
 ### Other Changes
 
-
 ## 0.31.0 (2026-02-25)
 
 ### Features Added

sdk/cosmos/azure_data_cosmos/CHANGELOG.md

@@ -0,0 +1,90 @@
+# Response Metadata Type System Design Spec
+
+## Motivation
+
+The SDK needs a type-safe way to expose operation-specific response metadata.
+Different operations return different headers (e.g. queries return index metrics,
+point operations return ETags). The type system should prevent callers from
+accessing headers that don't apply to their operation.
+
+## Design (Wrapper Types with Composition)
+
+### Point operation wrappers
+
+Each operation category gets its own response wrapper that composes over
+the internal `CosmosResponse<T>`:
+
+```rust
+/// Point item operations (create, read, replace, upsert, delete).
+pub struct ItemResponse<T> {
+    response: CosmosResponse<T>,
+    etag: Option<Etag>,           // azure_core::http::Etag
+}
+
+/// Resource management operations (databases, containers, throughput).
+pub struct ResourceResponse<T> {
+    response: CosmosResponse<T>,
+}
+
+/// Transactional batch operations.
+pub struct BatchResponse {
+    response: CosmosResponse<TransactionalBatchResponse>,
+    etag: Option<Etag>,
+}
+```
+
+### Feed page composition
+
+`FeedPage<T>` is a generic, reusable page type for any feed operation.
+`QueryFeedPage<T>` composes over it, adding query-specific fields:
+
+```rust
+/// Generic feed page — used for any feed operation (queries, read-many,
+/// change feed, offers, etc.).
+pub struct FeedPage<T> {
+    items: Vec<T>,
+    continuation: Option<String>,
+    raw_headers: Headers,
+    headers: CosmosResponseHeaders,
+    diagnostics: CosmosDiagnostics,
+}
+
+/// Query-specific feed page — wraps FeedPage and adds query metadata.
+pub struct QueryFeedPage<T> {
+    page: FeedPage<T>,
+    index_metrics: Option<String>,
+    query_metrics: Option<String>,
+}
+```
+
+### Common accessors (on all wrapper types)
+
+- `status()` / `headers()` — HTTP response data
+- `request_charge()` — RU consumption
+- `session_token()` — session token for consistency
+- `diagnostics()` — `CosmosDiagnostics` (activity ID, server duration)
+- `into_body()` / `into_model()` — consume response body
+
+### Operation-specific accessors
+
+| Type | Extra methods |
+|------|--------------|
+| `ItemResponse<T>` | `etag() -> Option<&Etag>` |
+| `ResourceResponse<T>` | (none — future-proof) |
+| `BatchResponse` | `etag() -> Option<&Etag>` |
+| `FeedPage<T>` | (common methods only) |
+| `QueryFeedPage<T>` | `index_metrics() -> Option<&str>`, `query_metrics() -> Option<&str>` |
+
+### Decisions made
+
+1. **Wrapper types** preferred over generic metadata params.
+2. **`CosmosResponse<T>` is `pub(crate)`** — wrapper types are the public API.
+3. **`QueryFeedPage<T>` composes over `FeedPage<T>`** — query-specific fields on the wrapper.
+4. **`FeedPage<T>` is public** — reusable for future read-many, change feed, etc.
+5. **`FeedItemIterator<T>` and `FeedPageIterator<T>`** yield `QueryFeedPage<T>` for queries.
+6. **ETags use `azure_core::http::Etag`** — not raw strings.
+7. **`CosmosDiagnostics`** is universal, available on all response types via `diagnostics()`.
+
+### Open questions
+
+None — all design decisions resolved.

sdk/cosmos/azure_data_cosmos/docs/RESPONSE_METADATA_SPEC.md

@@ -3,11 +3,14 @@
 
 use crate::{
     clients::OffersClient,
-    models::{ContainerProperties, CosmosResponse, ThroughputProperties},
+    models::{
+        BatchResponse, ContainerProperties, CosmosResponse, ItemResponse, ResourceResponse,
+        ThroughputProperties,
+    },
     options::{BatchOptions, QueryOptions, ReadContainerOptions},
     pipeline::GatewayPipeline,
     resource_context::{ResourceLink, ResourceType},
-    transactional_batch::{TransactionalBatch, TransactionalBatchResponse},
+    transactional_batch::TransactionalBatch,
     DeleteContainerOptions, FeedItemIterator, ItemOptions, PartitionKey, Query,
     ReplaceContainerOptions, ThroughputOptions,
 };
@@ -106,15 +109,15 @@ impl ContainerClient {
             reason = "The 'options' parameter may be used in the future"
         )]
         options: Option<ReadContainerOptions>,
-    ) -> azure_core::Result<CosmosResponse<ContainerProperties>> {
+    ) -> azure_core::Result<ResourceResponse<ContainerProperties>> {
         let cosmos_request =
             CosmosRequest::builder(OperationType::Read, self.link.clone()).build()?;
         let response: CosmosResponse<ContainerProperties> = self
             .container_connection
             .send(cosmos_request, Context::default())
             .await?;
 
-        Ok(response)
+        Ok(ResourceResponse::new(response))
     }
 
     /// Updates the indexing policy of the container.
@@ -150,13 +153,14 @@ impl ContainerClient {
             reason = "The 'options' parameter may be used in the future"
         )]
         options: Option<ReplaceContainerOptions>,
-    ) -> azure_core::Result<CosmosResponse<ContainerProperties>> {
+    ) -> azure_core::Result<ResourceResponse<ContainerProperties>> {
         let cosmos_request = CosmosRequest::builder(OperationType::Replace, self.link.clone())
             .json(&properties)
             .build()?;
         self.container_connection
             .send(cosmos_request, Context::default())
             .await
+            .map(ResourceResponse::new)
     }
 
     /// Reads container throughput properties, if any.
@@ -199,7 +203,7 @@ impl ContainerClient {
         &self,
         throughput: ThroughputProperties,
         options: Option<ThroughputOptions>,
-    ) -> azure_core::Result<CosmosResponse<ThroughputProperties>> {
+    ) -> azure_core::Result<ResourceResponse<ThroughputProperties>> {
         #[allow(
             unused_variables,
             reason = "The 'options' variable may be used in the future"
@@ -214,7 +218,10 @@ impl ContainerClient {
             .expect("service should always return a '_rid' for a container");
 
         let offers_client = OffersClient::new(self.pipeline.clone(), resource_id);
-        offers_client.replace(Context::default(), throughput).await
+        offers_client
+            .replace(Context::default(), throughput)
+            .await
+            .map(ResourceResponse::new)
     }
 
     /// Deletes this container.
@@ -231,12 +238,13 @@ impl ContainerClient {
             reason = "The 'options' parameter may be used in the future"
         )]
         options: Option<DeleteContainerOptions>,
-    ) -> azure_core::Result<CosmosResponse<()>> {
+    ) -> azure_core::Result<ResourceResponse<()>> {
         let cosmos_request =
             CosmosRequest::builder(OperationType::Delete, self.link.clone()).build()?;
         self.container_connection
             .send(cosmos_request, Context::default())
             .await
+            .map(ResourceResponse::new)
     }
 
     /// Creates a new item in the container.
@@ -274,7 +282,7 @@ impl ContainerClient {
     ///
     /// By default, the newly created item is *not* returned in the HTTP response.
     /// If you want the new item to be returned, set the [`ItemOptions::with_content_response_on_write_enabled()`] option to `true`.
-    /// You can deserialize the returned item by retrieving the [`ResponseBody`](azure_core::http::response::ResponseBody) using [`CosmosResponse::into_body`] and then calling [`ResponseBody::json`](azure_core::http::response::ResponseBody::json), like this:
+    /// You can deserialize the returned item by retrieving the [`ResponseBody`](azure_core::http::response::ResponseBody) using [`ItemResponse::into_body`] and then calling [`ResponseBody::json`](azure_core::http::response::ResponseBody::json), like this:
     ///
     /// ```rust,no_run
     /// use azure_data_cosmos::ItemOptions;
@@ -307,7 +315,7 @@ impl ContainerClient {
         partition_key: impl Into<PartitionKey>,
         item: T,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<CosmosResponse<()>> {
+    ) -> azure_core::Result<ItemResponse<()>> {
         let options = options.clone().unwrap_or_default();
         let excluded_regions = options.excluded_regions.clone();
         let mut cosmos_request =
@@ -321,6 +329,7 @@ impl ContainerClient {
         self.container_connection
             .send(cosmos_request, Context::default())
             .await
+            .map(ItemResponse::new)
     }
 
     /// Replaces an existing item in the container.
@@ -359,7 +368,7 @@ impl ContainerClient {
     ///
     /// By default, the replaced item is *not* returned in the HTTP response.
     /// If you want the replaced item to be returned, set the [`ItemOptions::with_content_response_on_write_enabled()`] option to `true`.
-    /// You can deserialize the returned item by retrieving the [`ResponseBody`](azure_core::http::response::ResponseBody) using [`CosmosResponse::into_body`] and then calling [`ResponseBody::json`](azure_core::http::response::ResponseBody::json), like this:
+    /// You can deserialize the returned item by retrieving the [`ResponseBody`](azure_core::http::response::ResponseBody) using [`ItemResponse::into_body`] and then calling [`ResponseBody::json`](azure_core::http::response::ResponseBody::json), like this:
     ///
     /// ```rust,no_run
     /// use azure_data_cosmos::ItemOptions;
@@ -392,7 +401,7 @@ impl ContainerClient {
         item_id: &str,
         item: T,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<CosmosResponse<()>> {
+    ) -> azure_core::Result<ItemResponse<()>> {
         let link = self.items_link.item(item_id);
         let options = options.clone().unwrap_or_default();
         let excluded_regions = options.excluded_regions.clone();
@@ -406,6 +415,7 @@ impl ContainerClient {
         self.container_connection
             .send(cosmos_request, Context::default())
             .await
+            .map(ItemResponse::new)
     }
 
     /// Creates or replaces an item in the container.
@@ -447,7 +457,7 @@ impl ContainerClient {
     ///
     /// By default, the created/replaced item is *not* returned in the HTTP response.
     /// If you want the created/replaced item to be returned, set the [`ItemOptions::with_content_response_on_write_enabled()`] option to `true`.
-    /// You can deserialize the returned item by retrieving the [`ResponseBody`](azure_core::http::response::ResponseBody) using [`CosmosResponse::into_body`] and then calling [`ResponseBody::json`](azure_core::http::response::ResponseBody::json), like this:
+    /// You can deserialize the returned item by retrieving the [`ResponseBody`](azure_core::http::response::ResponseBody) using [`ItemResponse::into_body`] and then calling [`ResponseBody::json`](azure_core::http::response::ResponseBody::json), like this:
     ///
     /// ```rust,no_run
     /// use azure_data_cosmos::ItemOptions;
@@ -479,7 +489,7 @@ impl ContainerClient {
         partition_key: impl Into<PartitionKey>,
         item: T,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<CosmosResponse<()>> {
+    ) -> azure_core::Result<ItemResponse<()>> {
         let options = options.clone().unwrap_or_default();
         let excluded_regions = options.excluded_regions.clone();
         let mut cosmos_request =
@@ -493,7 +503,8 @@ impl ContainerClient {
         return self
             .container_connection
             .send(cosmos_request, Context::default())
-            .await;
+            .await
+            .map(ItemResponse::new);
     }
 
     /// Reads a specific item from the container.
@@ -532,7 +543,7 @@ impl ContainerClient {
         partition_key: impl Into<PartitionKey>,
         item_id: &str,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<CosmosResponse<T>> {
+    ) -> azure_core::Result<ItemResponse<T>> {
         let mut options = options.unwrap_or_default();
 
         // Read APIs should always return the item, ignoring whatever the user set.
@@ -549,6 +560,7 @@ impl ContainerClient {
         self.container_connection
             .send(cosmos_request, Context::default())
             .await
+            .map(|r| ItemResponse::new(r))
     }
 
     /// Deletes an item from the container.
@@ -577,7 +589,7 @@ impl ContainerClient {
         partition_key: impl Into<PartitionKey>,
         item_id: &str,
         options: Option<ItemOptions>,
-    ) -> azure_core::Result<CosmosResponse<()>> {
+    ) -> azure_core::Result<ItemResponse<()>> {
         let link = self.items_link.item(item_id);
         let options = options.clone().unwrap_or_default();
         let excluded_regions = options.excluded_regions.clone();
@@ -590,6 +602,7 @@ impl ContainerClient {
         self.container_connection
             .send(cosmos_request, Context::default())
             .await
+            .map(ItemResponse::new)
     }
 
     /// Executes a single-partition query against items in the container.
@@ -726,7 +739,7 @@ impl ContainerClient {
         &self,
         batch: TransactionalBatch,
         options: Option<BatchOptions>,
-    ) -> azure_core::Result<CosmosResponse<TransactionalBatchResponse>> {
+    ) -> azure_core::Result<BatchResponse> {
         let options = options.unwrap_or_default();
         let partition_key = batch.partition_key().clone();
 
@@ -740,5 +753,6 @@ impl ContainerClient {
         self.container_connection
             .send(cosmos_request, Context::default())
             .await
+            .map(BatchResponse::new)
     }
 }