Add ChangeFeed AllVersionsAndDeletes mode

sdk/cosmos/azure_data_cosmos/CHANGELOG.md

@@ -33,6 +33,7 @@
 - Added preview distributed transaction SDK builders, patch operations, diagnostics, and response accessors behind the disabled-by-default `preview_dtx` feature. ([#4702](https://github.com/Azure/azure-sdk-for-rust/pull/4702))
 - Added change feed pull support via `ContainerClient::query_change_feed()`, which takes a required `ChangeFeedStartFrom` start position (`Beginning`, `Now`, `PointInTime`) and returns a `ChangeFeedPageIterator<T>` that streams `FeedPage<T>` results. New `feed` types `ChangeFeedPageIterator`, `FeedScope`, and `ContinuationToken`, plus `options` types `ChangeFeedOptions` and `ChangeFeedMode` (currently `LatestVersion`); supports single-partition, per-partition-key, and full-container (cross-partition fan-out) reads with continuation-token resumption that persists the original start position so never-polled partitions don't replay history on resume. ([#4621](https://github.com/Azure/azure-sdk-for-rust/pull/4621))
 - Change feed items are now surfaced as an envelope. `ContainerClient::query_change_feed::<YourDoc>()` yields `ChangeFeedItem<YourDoc>`, binding the envelope into the return type so the post-change document is read via `ChangeFeedItem::current()` and cannot be silently deserialized away. The envelope also exposes the pre-change document (`previous()`) and per-change `metadata()` (populated by full-fidelity reads; absent for `LatestVersion`). A full-fidelity delete returns an empty `current` object, which maps to `None` so callers with strict document types still deserialize the delete; the deleted item's identity is available via `ChangeFeedMetadata::id()` and `ChangeFeedMetadata::partition_key()`. `ChangeFeedOperationType` includes an `Unknown` catch-all so a future operation type cannot fail a page and stall the feed. A backend that does not envelope change feed items (such as the Cosmos emulator) returns the bare document, which is mapped onto `current()` so no data is lost. Added the `models` types `ChangeFeedItem<T>`, `ChangeFeedMetadata`, `ChangeFeedOperationType`, and `LogicalSequenceNumber`. ([#4723](https://github.com/Azure/azure-sdk-for-rust/pull/4723))
+- Added `ChangeFeedMode::AllVersionsAndDeletes` ("full fidelity") to `ContainerClient::query_change_feed()`. In this mode the service returns every intermediate version and delete, and each `ChangeFeedItem<T>` envelope carries per-change `metadata()` (operation type, LSN, conflict resolution timestamp, previous-image LSN, and a time-to-live-expiry flag); the pre-change document (`previous()`) is populated only when the container also retains pre-images. `ChangeFeedStartFrom::Beginning` and `ChangeFeedStartFrom::PointInTime` are rejected by the service for this mode (with `400 Bad Request`) because intermediate versions and deletes are only retained within the container's retention window; use `Now` or resume from a continuation token. The continuation token records the feed mode, so a resume that switches modes is rejected. Added the `models` type `ChangeFeedPolicy` and `ContainerProperties::with_change_feed_policy()` to configure a container's full-fidelity retention window (required to read in this mode). ([#4706](https://github.com/Azure/azure-sdk-for-rust/pull/4706))
 - Added `TlsBackend` (re-exported) and a `tls_backend` option on `ConnectionPoolOptions` (`ConnectionPoolOptionsBuilder::with_tls_backend`), defaulting to `TlsBackend::Rustls`, available under the `rustls` feature, to pin the TLS backend used by the transport. This is additive and changes no behavior for the default (rustls) build; it only has an effect in builds that compile in multiple reqwest TLS backends, where reqwest would otherwise default to native-tls and the driver now pins rustls instead. ([#4649](https://github.com/Azure/azure-sdk-for-rust/pull/4649))
 
 ### Bugs Fixed

sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs

@@ -9,10 +9,10 @@ use crate::{
     models::{BatchResponse, ChangeFeedItem, ItemResponse, ResourceResponse},
     models::{ContainerProperties, PatchInstructions, ThroughputProperties},
     options::{
-        BatchOptions, BinaryEncodingOptions, ChangeFeedOptions, ChangeFeedStartFrom,
-        DeleteContainerOptions, ItemReadOptions, ItemWriteOptions, OperationOptions,
-        PatchItemOptions, Precondition, QueryOptions, ReadContainerOptions, ReadFeedRangesOptions,
-        ReplaceContainerOptions, SessionToken, ThroughputOptions,
+        BatchOptions, BinaryEncodingOptions, ChangeFeedMode, ChangeFeedOptions,
+        ChangeFeedStartFrom, DeleteContainerOptions, ItemReadOptions, ItemWriteOptions,
+        OperationOptions, PatchItemOptions, Precondition, QueryOptions, ReadContainerOptions,
+        ReadFeedRangesOptions, ReplaceContainerOptions, SessionToken, ThroughputOptions,
     },
     PartitionKey, Query,
 };
@@ -900,10 +900,24 @@ impl ContainerClient {
 
     /// Queries the change feed for a container, returning a stream of pages.
     ///
-    /// The change feed provides an ordered list of changes (creates and
-    /// replaces) made to items in the container. Each change is yielded as a
-    /// [`ChangeFeedItem<T>`](crate::models::ChangeFeedItem); see that type for
-    /// how to read the changed document.
+    /// The change feed provides an ordered list of changes made to items in the
+    /// container. Every change is returned as a
+    /// [`ChangeFeedItem<T>`](crate::models::ChangeFeedItem) wire-format
+    /// envelope, so bind `T = YourDoc` and read the post-change
+    /// document via [`current()`](crate::models::ChangeFeedItem::current).
+    ///
+    /// The [`mode`](crate::options::ChangeFeedOptions::mode) selects what each
+    /// change carries:
+    ///
+    /// * [`ChangeFeedMode::LatestVersion`] (default) — the latest version of
+    ///   each created or replaced item. `current()` holds the item and
+    ///   `metadata()` may also be present (for example `lsn` and the commit
+    ///   timestamp), while `previous()` is not populated.
+    /// * [`ChangeFeedMode::AllVersionsAndDeletes`] — every intermediate version
+    ///   plus deletes. The envelope additionally exposes the
+    ///   pre-change document
+    ///   ([`previous()`](crate::models::ChangeFeedItem::previous)) and change
+    ///   [`metadata()`](crate::models::ChangeFeedItem::metadata).
     ///
     /// # Arguments
     /// * `scope` - Determines which partitions to read changes from.
@@ -912,12 +926,32 @@ impl ContainerClient {
     ///   the token holds its own position.
     /// * `options` - Optional parameters controlling mode, session token, and paging.
     ///
+    /// # AllVersionsAndDeletes limitations
+    ///
+    /// * Only [`ChangeFeedStartFrom::Now`] or resuming from a continuation token
+    ///   is supported. [`ChangeFeedStartFrom::Beginning`] and
+    ///   [`ChangeFeedStartFrom::PointInTime`] are **not** supported, because
+    ///   intermediate versions and deletes are only retained within the
+    ///   container's retention / continuous-backup window. The service gates
+    ///   this and rejects such a request with `400 Bad Request`.
+    /// * When starting from [`ChangeFeedStartFrom::Now`], every range is pinned
+    ///   to its concrete starting position before the first page is returned, so
+    ///   a range that is never served before a checkpoint still resumes from its
+    ///   true start rather than resume-time. No intermediate versions or deletes
+    ///   are dropped across a resume. `Now` is deliberately **not** rewritten to
+    ///   a concrete [`ChangeFeedStartFrom::PointInTime`], which would change its
+    ///   semantics; each range instead captures its own server continuation.
+    /// * The feed mode is encoded in the continuation token, so a token issued in
+    ///   one mode cannot be used to resume in another; attempting to do so is
+    ///   rejected. Re-pass [`ChangeFeedMode::AllVersionsAndDeletes`] on resume to
+    ///   match the original mode.
+    ///
     /// # Examples
     ///
+    /// Read the latest version of each change from the beginning:
+    ///
     /// ```rust,no_run
-    /// use azure_data_cosmos::{
-    ///     clients::ContainerClient, feed::FeedScope, options::ChangeFeedStartFrom,
-    /// };
+    /// use azure_data_cosmos::{clients::ContainerClient, feed::FeedScope, options::ChangeFeedStartFrom};
     /// use futures::StreamExt;
     /// use serde::Deserialize;
     ///
@@ -945,6 +979,45 @@ impl ContainerClient {
     /// # Ok(())
     /// # }
     /// ```
+    ///
+    /// Read every version and delete:
+    ///
+    /// ```rust,no_run
+    /// use azure_data_cosmos::{
+    ///     clients::ContainerClient,
+    ///     feed::FeedScope,
+    ///     options::{ChangeFeedMode, ChangeFeedOptions, ChangeFeedStartFrom},
+    /// };
+    /// use futures::StreamExt;
+    /// use serde::Deserialize;
+    ///
+    /// // A delete envelope may omit non-key fields, so keep them optional.
+    /// #[derive(Debug, Deserialize)]
+    /// struct MyItem {
+    ///     id: String,
+    ///     #[serde(default)]
+    ///     value: Option<i64>,
+    /// }
+    ///
+    /// # async fn example(container: ContainerClient) -> Result<(), Box<dyn std::error::Error>> {
+    /// let options = ChangeFeedOptions::default().with_mode(ChangeFeedMode::AllVersionsAndDeletes);
+    /// let mut pages = container
+    ///     .query_change_feed::<MyItem>(
+    ///         FeedScope::full_container(),
+    ///         ChangeFeedStartFrom::Now,
+    ///         Some(options),
+    ///     )
+    ///     .await?;
+    ///
+    /// while let Some(page) = pages.next().await {
+    ///     for item in page?.items() {
+    ///         println!("{:?}: current={:?} previous={:?}",
+    ///             item.operation_type(), item.current(), item.previous());
+    ///     }
+    /// }
+    /// # Ok(())
+    /// # }
+    /// ```
     pub async fn query_change_feed<T: DeserializeOwned + Send + 'static>(
         &self,
         scope: FeedScope,
@@ -953,10 +1026,29 @@ impl ContainerClient {
     ) -> crate::Result<ChangeFeedPageIterator<ChangeFeedItem<T>>> {
         let options = options.unwrap_or_default();
 
-        let mut initial_operation = CosmosOperation::change_feed(
-            self.container_ref.clone(),
-            Some(scope.into_feed_range(self.container_ref.partition_key_definition())),
-        );
+        let feed_range = scope.into_feed_range(self.container_ref.partition_key_definition());
+
+        // The mode selects the base operation, i.e. which `A-IM` header is sent.
+        // Both modes return `ChangeFeedItem<T>` envelopes; AllVersionsAndDeletes
+        // additionally populates `previous` and `metadata`.
+        let mut initial_operation = match options.mode {
+            ChangeFeedMode::AllVersionsAndDeletes => {
+                // AllVersionsAndDeletes can only read within the container's
+                // retention / continuous-backup window, so it supports starting
+                // only from "now" or by resuming a continuation token. Reading
+                // from the beginning of the container or from an arbitrary point
+                // in time is not supported in this mode; the service gates this
+                // and returns a `400 Bad Request`, so it is not re-validated
+                // client-side here.
+                CosmosOperation::change_feed_all_versions_and_deletes(
+                    self.container_ref.clone(),
+                    Some(feed_range),
+                )
+            }
+            ChangeFeedMode::LatestVersion => {
+                CosmosOperation::change_feed(self.container_ref.clone(), Some(feed_range))
+            }
+        };
 
         if let Some(token) = options.session_token {
             initial_operation = initial_operation.with_session_token(token);

sdk/cosmos/azure_data_cosmos/src/feed/change_feed_iterator.rs

@@ -187,8 +187,18 @@ impl LiveState {
     }
 }
 
-/// Deserializes a change feed response body into the caller's item type `T`
-/// (bound to [`ChangeFeedItem<Doc>`](crate::models::ChangeFeedItem)).
+/// Deserializes a change feed response body into the caller's item type.
+///
+/// Every change feed item is returned as a wire-format envelope
+/// (`{ current, ... }`) because the SDK always sends the
+/// `x-ms-cosmos-changefeed-wire-format-version` header (see
+/// [`CosmosOperation::change_feed`]). Each entry in the feed body is
+/// deserialized directly into `T` — which
+/// [`ContainerClient::query_change_feed`](crate::clients::ContainerClient::query_change_feed)
+/// binds to [`ChangeFeedItem<Doc>`](crate::models::ChangeFeedItem) — so the whole
+/// envelope is preserved rather than stripped.
+///
+/// [`CosmosOperation::change_feed`]: azure_data_cosmos_driver::models::CosmosOperation
 fn deserialize_change_feed_items<T: DeserializeOwned>(
     response: CosmosResponse,
 ) -> crate::Result<Vec<T>> {
@@ -199,8 +209,9 @@ fn deserialize_change_feed_items<T: DeserializeOwned>(
 /// A stream of pages from a Cosmos DB change feed operation.
 ///
 /// Yields [`FeedPage<T>`] instances where `T` is
-/// [`ChangeFeedItem<YourDoc>`](crate::models::ChangeFeedItem); read the
-/// changed document via [`current()`](crate::models::ChangeFeedItem::current).
+/// [`ChangeFeedItem<YourDoc>`](crate::models::ChangeFeedItem): every item is
+/// returned as the wire-format envelope, so the caller reads the post-change
+/// document via [`current()`](crate::models::ChangeFeedItem::current).
 ///
 /// The stream is conceptually infinite: when a partition has no new changes
 /// (304 Not Modified), an empty page is returned instead of terminating the
@@ -212,9 +223,7 @@ fn deserialize_change_feed_items<T: DeserializeOwned>(
 /// # Examples
 ///
 /// ```rust,no_run
-/// use azure_data_cosmos::{
-///     clients::ContainerClient, feed::FeedScope, options::ChangeFeedStartFrom,
-/// };
+/// use azure_data_cosmos::{clients::ContainerClient, feed::FeedScope, options::ChangeFeedStartFrom};
 /// use futures::StreamExt;
 /// use serde::Deserialize;
 ///
@@ -312,7 +321,7 @@ impl<T: Send + DeserializeOwned + 'static> Stream for ChangeFeedPageIterator<T>
 #[cfg(test)]
 mod tests {
     use super::deserialize_change_feed_items;
-    use crate::models::{ChangeFeedItem, CosmosResponse};
+    use crate::models::{ChangeFeedItem, ChangeFeedOperationType, CosmosResponse};
     use azure_core::http::StatusCode;
     use azure_data_cosmos_driver::diagnostics::DiagnosticsContext;
     use azure_data_cosmos_driver::models::{
@@ -370,6 +379,77 @@ mod tests {
         );
     }
 
+    #[test]
+    fn deserializes_all_versions_and_deletes_page() {
+        // AllVersionsAndDeletes binds `T = ChangeFeedItem<Doc>` and keeps the
+        // whole envelope: create + replace + delete, with metadata and a
+        // pre-image preserved rather than stripped.
+        let body = json!({
+            "Documents": [
+                {
+                    "current": { "id": "1" },
+                    "metadata": { "operationType": "create", "lsn": 10, "crts": 1720322460 }
+                },
+                {
+                    "current": { "id": "2" },
+                    "previous": { "id": "2" },
+                    "metadata": { "operationType": "replace", "lsn": 11, "previousImageLSN": 10 }
+                },
+                {
+                    "previous": { "id": "3" },
+                    "metadata": { "operationType": "delete", "lsn": 12, "timeToLiveExpired": true }
+                }
+            ],
+            "_count": 3
+        });
+
+        let items: Vec<ChangeFeedItem<Doc>> =
+            deserialize_change_feed_items(make_response(body)).unwrap();
+        assert_eq!(items.len(), 3);
+
+        // Create: current present, no previous.
+        assert_eq!(
+            items[0].operation_type(),
+            Some(ChangeFeedOperationType::Create)
+        );
+        assert_eq!(items[0].current(), Some(&Doc { id: "1".into() }));
+        assert!(items[0].previous().is_none());
+        assert_eq!(
+            items[0].metadata().and_then(|m| m.lsn()),
+            Some(crate::models::LogicalSequenceNumber::from(10))
+        );
+        assert_eq!(
+            items[0]
+                .metadata()
+                .and_then(|m| m.conflict_resolution_timestamp()),
+            Some(std::time::Duration::from_secs(1720322460))
+        );
+
+        // Replace: both current and previous present.
+        assert_eq!(
+            items[1].operation_type(),
+            Some(ChangeFeedOperationType::Replace)
+        );
+        assert_eq!(items[1].current(), Some(&Doc { id: "2".into() }));
+        assert_eq!(items[1].previous(), Some(&Doc { id: "2".into() }));
+        assert_eq!(
+            items[1].metadata().and_then(|m| m.previous_image_lsn()),
+            Some(crate::models::LogicalSequenceNumber::from(10))
+        );
+
+        // Delete: current absent, previous (pre-image) preserved, TTL flag set.
+        assert_eq!(
+            items[2].operation_type(),
+            Some(ChangeFeedOperationType::Delete)
+        );
+        assert!(items[2].current().is_none());
+        assert_eq!(items[2].previous(), Some(&Doc { id: "3".into() }));
+        assert_eq!(
+            items[2].metadata().and_then(|m| m.time_to_live_expired()),
+            Some(true)
+        );
+    }
+
     /// Builds an SDK [`CosmosResponse`] wrapping the given JSON body so the
     /// deserialize helper can be exercised end to end.
     fn make_response(body: serde_json::Value) -> CosmosResponse {