Add ChangeFeed AllVersionsAndDeletes (full-fidelity) mode

Implements the AllVersionsAndDeletes (AVAD / full-fidelity) change feed
mode, stacked on the LatestVersion change-feed work (#4621).

Driver (azure_data_cosmos_driver):
- Add FULL_FIDELITY_FEED = "Full-Fidelity Feed" and a full_fidelity_feed
  flag on CosmosRequestHeaders; write_to_headers emits A-IM: Full-Fidelity
  Feed for AVAD (mutually exclusive with Incremental Feed).
- Add CosmosOperation::change_feed_all_versions_and_deletes factory sharing
  a private helper with change_feed; both set the changefeed wire-format
  version. Unit tests cover both header values and factories.

SDK (azure_data_cosmos):
- New public models ChangeFeedItem<T>, ChangeFeedMetadata, and
  ChangeFeedOperationType describing the full-fidelity envelope
  (current/previous/metadata), with accessors and thorough unit tests.
- Thread an unwrap_current flag through ChangeFeedPageIterator so AVAD
  pages deserialize the whole envelope (T = ChangeFeedItem<Doc>) instead of
  stripping current as LatestVersion does.
- Wire AVAD into ContainerClient::query_change_feed, dispatched on
  options.mode, and reject ChangeFeedStartFrom::Beginning for AVAD since
  intermediate versions/deletes are only retained within the container's
  retention window.
- Update ChangeFeedMode docs and both CHANGELOGs.

Deferred (documented as follow-ups): emulator/live AVAD end-to-end tests,
lossless per-range Now resume, and persisting the feed mode in the
continuation token.

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

Author: yumnahussain

sdk/cosmos/azure_data_cosmos/CHANGELOG.md

@@ -5,6 +5,7 @@
 ### Features Added
 
 - 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` (`LatestVersion` default; `AllVersionsAndDeletes` is reserved for a future release and currently returns an error); 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))
+- Added `ChangeFeedMode::AllVersionsAndDeletes` ("full fidelity") support to `ContainerClient::query_change_feed()`. In this mode each change is delivered as a `ChangeFeedItem<T>` envelope exposing the post-change document (`current`), the pre-change document (`previous`, when available), and a `ChangeFeedMetadata` describing the change (`ChangeFeedOperationType` — `Create`/`Replace`/`Delete` — plus `lsn`, commit timestamp, previous-image LSN, and a time-to-live-expiry flag). New public model types `ChangeFeedItem<T>`, `ChangeFeedMetadata`, and `ChangeFeedOperationType`. `ChangeFeedStartFrom::Beginning` is rejected for this mode because intermediate versions and deletes are only retained within the container's retention window; use `Now` or an in-retention `PointInTime`. The continuation token does not persist the feed mode, so callers must re-pass `AllVersionsAndDeletes` on resume.
 - 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))
 
 ### Breaking Changes

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

@@ -857,8 +857,18 @@ 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 in LatestVersion mode) made to items in the container.
+    /// The change feed provides an ordered list of changes made to items in the
+    /// container. The [`mode`](crate::options::ChangeFeedOptions::mode) selects
+    /// what each change carries and how the item type `T` is bound:
+    ///
+    /// * [`ChangeFeedMode::LatestVersion`] (default) — returns the latest
+    ///   version of each created or replaced item. Bind `T` to your document
+    ///   type; the SDK unwraps the wire-format envelope for you.
+    /// * [`ChangeFeedMode::AllVersionsAndDeletes`] — returns every intermediate
+    ///   version plus deletes ("full fidelity"). Bind `T` to
+    ///   [`ChangeFeedItem<YourDoc>`](crate::models::ChangeFeedItem); the SDK
+    ///   preserves the whole envelope so you can inspect `current`, `previous`,
+    ///   and the change [`metadata`](crate::models::ChangeFeedMetadata).
     ///
     /// # Arguments
     /// * `scope` - Determines which partitions to read changes from.
@@ -867,8 +877,28 @@ impl ContainerClient {
     ///   the token holds its own position.
     /// * `options` - Optional parameters controlling mode, session token, and paging.
     ///
+    /// # AllVersionsAndDeletes limitations
+    ///
+    /// * [`ChangeFeedStartFrom::Beginning`] is **rejected** for this mode:
+    ///   intermediate versions and deletes are only retained within the
+    ///   container's retention / continuous-backup window, so there is no
+    ///   "beginning of all history" to read from. Use
+    ///   [`ChangeFeedStartFrom::Now`] or a [`ChangeFeedStartFrom::PointInTime`]
+    ///   within the retention window.
+    /// * [`ChangeFeedStartFrom::Now`] is re-evaluated per range the first time a
+    ///   range is polled, so a range that is never polled before a checkpoint
+    ///   resumes from resume-time and can drop the intermediate versions and
+    ///   deletes that occurred between the original start and the resume. `Now`
+    ///   is deliberately **not** converted to a concrete `PointInTime`, because
+    ///   that would change its semantics; lossless per-range `Now` resolution is
+    ///   a future improvement.
+    /// * The continuation token does not persist the feed mode, so callers must
+    ///   re-pass [`ChangeFeedMode::AllVersionsAndDeletes`] on resume.
+    ///
     /// # 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 futures::StreamExt;
@@ -894,6 +924,46 @@ impl ContainerClient {
     /// # Ok(())
     /// # }
     /// ```
+    ///
+    /// Read every version and delete ("full fidelity"):
+    ///
+    /// ```rust,no_run
+    /// use azure_data_cosmos::{
+    ///     clients::ContainerClient,
+    ///     feed::FeedScope,
+    ///     models::ChangeFeedItem,
+    ///     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::<ChangeFeedItem<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,
@@ -902,39 +972,45 @@ impl ContainerClient {
     ) -> crate::Result<ChangeFeedPageIterator<T>> {
         let options = options.unwrap_or_default();
 
-        // AllVersionsAndDeletes is reserved for a future release. Reject it
-        // explicitly rather than silently behaving like LatestVersion.
-        //
-        // TODO: When enabling AllVersionsAndDeletes, revisit start/resume
-        // semantics — they differ from LatestVersion:
-        //   - `Beginning` is invalid under AVAD (bounded retention window);
-        //     require `Now` or a `PointInTime` within retention and reject
-        //     `Beginning` for this mode.
-        //   - `Now` is re-evaluated per request, so a range never polled before
-        //     a checkpoint resumes from resume-time and would drop the
-        //     intermediate versions/deletes between the original start and the
-        //     resume. Resolve `Now` to a concrete per-range start position (or
-        //     snapshot the start time as a `PointInTime`) so AVAD resume is
-        //     lossless.
-        //   - The continuation token does not persist the feed mode; callers
-        //     must re-pass AllVersionsAndDeletes on resume.
-        if options.mode == ChangeFeedMode::AllVersionsAndDeletes {
-            return Err(crate::DriverCosmosError::builder()
-                .with_status(crate::error::CosmosStatus::new(
-                    azure_core::http::StatusCode::NotImplemented,
-                ))
-                .with_message(
-                    "AllVersionsAndDeletes change feed mode is not yet supported; \
-                     use ChangeFeedMode::LatestVersion",
+        let feed_range = scope.into_feed_range(self.container_ref.partition_key_definition());
+
+        // The mode selects the base operation (which `A-IM` header is sent) and
+        // whether the iterator unwraps `current` (LatestVersion) or preserves
+        // the whole envelope (AllVersionsAndDeletes → `ChangeFeedItem<T>`).
+        let (mut initial_operation, unwrap_current) = match options.mode {
+            ChangeFeedMode::AllVersionsAndDeletes => {
+                // AllVersionsAndDeletes can only read within the container's
+                // retention window, so "from the beginning of all history" is
+                // not a valid start. Reject it with a clear client error rather
+                // than issuing a request the service would refuse.
+                if start_from == ChangeFeedStartFrom::Beginning {
+                    return Err(crate::DriverCosmosError::builder()
+                        .with_status(crate::error::CosmosStatus::new(
+                            azure_core::http::StatusCode::BadRequest,
+                        ))
+                        .with_message(
+                            "ChangeFeedStartFrom::Beginning is not supported for \
+                             ChangeFeedMode::AllVersionsAndDeletes because intermediate versions \
+                             and deletes are only retained within the container's retention / \
+                             continuous-backup window; use ChangeFeedStartFrom::Now or a \
+                             ChangeFeedStartFrom::PointInTime within the retention window",
+                        )
+                        .build()
+                        .into());
+                }
+                (
+                    CosmosOperation::change_feed_all_versions_and_deletes(
+                        self.container_ref.clone(),
+                        Some(feed_range),
+                    ),
+                    false,
                 )
-                .build()
-                .into());
-        }
-
-        let mut initial_operation = CosmosOperation::change_feed(
-            self.container_ref.clone(),
-            Some(scope.into_feed_range(self.container_ref.partition_key_definition())),
-        );
+            }
+            ChangeFeedMode::LatestVersion => (
+                CosmosOperation::change_feed(self.container_ref.clone(), Some(feed_range)),
+                true,
+            ),
+        };
 
         if let Some(token) = options.session_token {
             initial_operation = initial_operation.with_session_token(token);
@@ -966,6 +1042,7 @@ impl ContainerClient {
             Some(self.container_ref.clone()),
             plan,
             options.operation,
+            unwrap_current,
         ))
     }
 

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

@@ -27,6 +27,13 @@ struct LiveState {
     plan: Option<OperationPlan>,
     in_flight: Option<DriverPageFuture>,
     errored: bool,
+    /// Controls how each page's items are decoded.
+    ///
+    /// `true` (LatestVersion): strip the `current` field from each enveloped
+    /// item so `T` is the caller's plain document type. `false`
+    /// (AllVersionsAndDeletes): deserialize the whole envelope so
+    /// `T = ChangeFeedItem<Doc>` binds `current`, `previous`, and `metadata`.
+    unwrap_current: bool,
 }
 
 impl LiveState {
@@ -35,6 +42,7 @@ impl LiveState {
         container: Option<ContainerReference>,
         plan: OperationPlan,
         options: OperationOptions,
+        unwrap_current: bool,
     ) -> Self {
         Self {
             driver,
@@ -43,6 +51,7 @@ impl LiveState {
             plan: Some(plan),
             in_flight: None,
             errored: false,
+            unwrap_current,
         }
     }
 
@@ -123,7 +132,15 @@ impl LiveState {
                     return task::Poll::Ready(Some(Ok(page)));
                 }
 
-                match unwrap_change_feed_items::<T>(response) {
+                // LatestVersion strips `current` so `T` is the plain document;
+                // AllVersionsAndDeletes keeps the whole envelope so
+                // `T = ChangeFeedItem<Doc>`.
+                let items = if *this.unwrap_current {
+                    unwrap_change_feed_items::<T>(response)
+                } else {
+                    deserialize_change_feed_items::<T>(response)
+                };
+                match items {
                     Ok(items) => {
                         let page = FeedPage::new(items, headers, diagnostics);
                         task::Poll::Ready(Some(Ok(page)))
@@ -178,6 +195,21 @@ fn unwrap_change_feed_items<T: DeserializeOwned>(
         .collect()
 }
 
+/// Deserializes a change feed response body into the caller's item type
+/// **without** unwrapping the wire-format envelope.
+///
+/// Used for AllVersionsAndDeletes (full-fidelity) mode, where the caller binds
+/// `T = ChangeFeedItem<Doc>` so the whole `{ current, previous, metadata }`
+/// envelope is preserved. Each entry in the feed body is deserialized directly
+/// into `T` rather than having its `current` field stripped (contrast
+/// [`unwrap_change_feed_items`], used for LatestVersion).
+fn deserialize_change_feed_items<T: DeserializeOwned>(
+    response: CosmosResponse,
+) -> crate::Result<Vec<T>> {
+    let body: FeedBody<T> = response.into_model()?;
+    Ok(body.items)
+}
+
 /// Unwraps the `current` document from a single change feed wire-format
 /// envelope, or returns the value unchanged when it is not enveloped.
 ///
@@ -205,11 +237,20 @@ fn unwrap_change_feed_item(item: serde_json::Value) -> serde_json::Value {
 
 /// A stream of pages from a Cosmos DB change feed operation.
 ///
-/// Yields [`FeedPage<T>`] instances, where `T` is the user's document type
-/// (for LatestVersion mode). The stream is conceptually infinite: when a
-/// partition has no new changes (304 Not Modified), an empty page is
-/// returned instead of terminating the stream. The consumer decides when
-/// to stop polling.
+/// Yields [`FeedPage<T>`] instances. The binding of `T` depends on the change
+/// feed mode:
+///
+/// * [`ChangeFeedMode::LatestVersion`](crate::options::ChangeFeedMode::LatestVersion):
+///   `T` is the user's document type. Each item's wire-format envelope is
+///   unwrapped so the caller sees the plain document.
+/// * [`ChangeFeedMode::AllVersionsAndDeletes`](crate::options::ChangeFeedMode::AllVersionsAndDeletes):
+///   `T` is [`ChangeFeedItem<YourDoc>`](crate::models::ChangeFeedItem). The full
+///   envelope is preserved so the caller can inspect `current`, `previous`, and
+///   the change [`metadata`](crate::models::ChangeFeedMetadata).
+///
+/// The stream is conceptually infinite: when a partition has no new changes
+/// (304 Not Modified), an empty page is returned instead of terminating the
+/// stream. The consumer decides when to stop polling.
 ///
 /// Use [`to_continuation_token()`](Self::to_continuation_token) to capture
 /// the current position for later resumption.
@@ -257,14 +298,26 @@ pub struct ChangeFeedPageIterator<T: Send> {
 
 impl<T: Send + DeserializeOwned + 'static> ChangeFeedPageIterator<T> {
     /// Creates a new `ChangeFeedPageIterator` backed by the given operation plan.
+    ///
+    /// `unwrap_current` selects how each item is decoded: `true` for
+    /// LatestVersion (strip `current` into the caller's document type), `false`
+    /// for AllVersionsAndDeletes (keep the whole envelope, `T =
+    /// ChangeFeedItem<Doc>`).
     pub(crate) fn new(
         driver: Arc<CosmosDriver>,
         container: Option<ContainerReference>,
         plan: OperationPlan,
         options: OperationOptions,
+        unwrap_current: bool,
     ) -> Self {
         Self {
-            state: Box::pin(LiveState::new(driver, container, plan, options)),
+            state: Box::pin(LiveState::new(
+                driver,
+                container,
+                plan,
+                options,
+                unwrap_current,
+            )),
             _marker: PhantomData,
         }
     }
@@ -301,8 +354,8 @@ impl<T: Send + DeserializeOwned + 'static> Stream for ChangeFeedPageIterator<T>
 
 #[cfg(test)]
 mod tests {
-    use super::{unwrap_change_feed_item, unwrap_change_feed_items};
-    use crate::models::CosmosResponse;
+    use super::{deserialize_change_feed_items, unwrap_change_feed_item, unwrap_change_feed_items};
+    use crate::models::{ChangeFeedItem, ChangeFeedOperationType, CosmosResponse};
     use azure_core::http::StatusCode;
     use azure_data_cosmos_driver::diagnostics::DiagnosticsContext;
     use azure_data_cosmos_driver::models::{
@@ -365,6 +418,53 @@ mod tests {
         assert_eq!(items, vec![Doc { id: "1".into() }, Doc { id: "2".into() }]);
     }
 
+    #[test]
+    fn deserializes_all_versions_and_deletes_page_without_unwrapping() {
+        // AllVersionsAndDeletes mode 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 }
+                },
+                {
+                    "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(), ChangeFeedOperationType::Create);
+        assert_eq!(items[0].current(), Some(&Doc { id: "1".into() }));
+        assert!(items[0].previous().is_none());
+        assert_eq!(items[0].metadata().lsn(), Some(10));
+
+        // Replace: both current and previous present.
+        assert_eq!(items[1].operation_type(), 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().previous_image_lsn(), Some(10));
+
+        // Delete: current absent, previous (pre-image) preserved, TTL flag set.
+        assert_eq!(items[2].operation_type(), ChangeFeedOperationType::Delete);
+        assert!(items[2].current().is_none());
+        assert_eq!(items[2].previous(), Some(&Doc { id: "3".into() }));
+        assert_eq!(items[2].metadata().time_to_live_expired(), Some(true));
+    }
+
     /// Builds an SDK [`CosmosResponse`] wrapping the given JSON body so the
     /// unwrap helper can be exercised end to end.
     fn make_response(body: serde_json::Value) -> CosmosResponse {