diff --git a/sdk/cosmos/azure_data_cosmos/CHANGELOG.md b/sdk/cosmos/azure_data_cosmos/CHANGELOG.md index 38a3496ffb5..5c9f8837444 100644 --- a/sdk/cosmos/azure_data_cosmos/CHANGELOG.md +++ b/sdk/cosmos/azure_data_cosmos/CHANGELOG.md @@ -22,6 +22,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` that streams `FeedPage` 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::()` yields `ChangeFeedItem`, 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`, `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` 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 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 a69327a7543..2a015b6359e 100644 --- a/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs +++ b/sdk/cosmos/azure_data_cosmos/src/clients/container_client.rs @@ -8,10 +8,10 @@ use crate::{ models::{BatchResponse, ChangeFeedItem, ItemResponse, ResourceResponse}, models::{ContainerProperties, PatchInstructions, ThroughputProperties}, options::{ - BatchOptions, ChangeFeedOptions, ChangeFeedStartFrom, DeleteContainerOptions, - ItemReadOptions, ItemWriteOptions, PatchItemOptions, Precondition, QueryOptions, - ReadContainerOptions, ReadFeedRangesOptions, ReplaceContainerOptions, SessionToken, - ThroughputOptions, + BatchOptions, ChangeFeedMode, ChangeFeedOptions, ChangeFeedStartFrom, + DeleteContainerOptions, ItemReadOptions, ItemWriteOptions, PatchItemOptions, Precondition, + QueryOptions, ReadContainerOptions, ReadFeedRangesOptions, ReplaceContainerOptions, + SessionToken, ThroughputOptions, }, PartitionKey, Query, }; @@ -863,10 +863,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`](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`](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. @@ -875,12 +889,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; /// @@ -908,6 +942,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, + /// } + /// + /// # async fn example(container: ContainerClient) -> Result<(), Box> { + /// let options = ChangeFeedOptions::default().with_mode(ChangeFeedMode::AllVersionsAndDeletes); + /// let mut pages = container + /// .query_change_feed::( + /// 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( &self, scope: FeedScope, @@ -916,10 +989,29 @@ impl ContainerClient { ) -> crate::Result>> { 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` 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); diff --git a/sdk/cosmos/azure_data_cosmos/src/feed/change_feed_iterator.rs b/sdk/cosmos/azure_data_cosmos/src/feed/change_feed_iterator.rs index cacdfbc043c..92674e37211 100644 --- a/sdk/cosmos/azure_data_cosmos/src/feed/change_feed_iterator.rs +++ b/sdk/cosmos/azure_data_cosmos/src/feed/change_feed_iterator.rs @@ -151,8 +151,18 @@ impl LiveState { } } -/// Deserializes a change feed response body into the caller's item type `T` -/// (bound to [`ChangeFeedItem`](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`](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( response: CosmosResponse, ) -> crate::Result> { @@ -163,8 +173,9 @@ fn deserialize_change_feed_items( /// A stream of pages from a Cosmos DB change feed operation. /// /// Yields [`FeedPage`] instances where `T` is -/// [`ChangeFeedItem`](crate::models::ChangeFeedItem); read the -/// changed document via [`current()`](crate::models::ChangeFeedItem::current). +/// [`ChangeFeedItem`](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 @@ -176,9 +187,7 @@ fn deserialize_change_feed_items( /// # 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; /// @@ -267,7 +276,7 @@ impl Stream for ChangeFeedPageIterator #[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::{ @@ -325,6 +334,77 @@ mod tests { ); } + #[test] + fn deserializes_all_versions_and_deletes_page() { + // AllVersionsAndDeletes binds `T = ChangeFeedItem` 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> = + 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 { diff --git a/sdk/cosmos/azure_data_cosmos/src/models/container_properties.rs b/sdk/cosmos/azure_data_cosmos/src/models/container_properties.rs index 8c35445a073..be9079ad36c 100644 --- a/sdk/cosmos/azure_data_cosmos/src/models/container_properties.rs +++ b/sdk/cosmos/azure_data_cosmos/src/models/container_properties.rs @@ -2,6 +2,7 @@ // Licensed under the MIT License. use std::borrow::Cow; +use std::time::Duration; use azure_core::fmt::SafeDebug; use serde::{Deserialize, Deserializer, Serialize, Serializer}; @@ -121,6 +122,16 @@ pub struct ContainerProperties { #[serde(skip_serializing_if = "Option::is_none")] pub vector_embedding_policy: Option, + /// The change feed policy for the container. + /// + /// Configure a retention duration here to enable the + /// [`AllVersionsAndDeletes`](crate::options::ChangeFeedMode::AllVersionsAndDeletes) + /// change feed mode. Without it, only the default + /// [`LatestVersion`](crate::options::ChangeFeedMode::LatestVersion) mode is + /// available. + #[serde(skip_serializing_if = "Option::is_none")] + pub change_feed_policy: Option, + /// The time-to-live for items in the container. /// /// For more information see @@ -149,6 +160,7 @@ impl ContainerProperties { unique_key_policy: None, conflict_resolution_policy: None, vector_embedding_policy: None, + change_feed_policy: None, default_ttl: TimeToLive::Forever, analytical_storage_ttl: TimeToLive::Forever, system_properties: SystemProperties::default(), @@ -181,6 +193,16 @@ impl ContainerProperties { self } + /// Sets the change feed policy for the container. + /// + /// Use [`ChangeFeedPolicy::all_versions_and_deletes`] to enable the + /// [`AllVersionsAndDeletes`](crate::options::ChangeFeedMode::AllVersionsAndDeletes) + /// change feed mode with a retention window. + pub fn with_change_feed_policy(mut self, change_feed_policy: ChangeFeedPolicy) -> Self { + self.change_feed_policy = Some(change_feed_policy); + self + } + pub fn with_default_ttl(mut self, default_ttl: impl Into) -> Self { self.default_ttl = default_ttl.into(); self @@ -408,11 +430,87 @@ pub enum ConflictResolutionMode { Custom, } +/// The change feed policy for a container. +/// +/// Configuring a retention duration enables the +/// [`AllVersionsAndDeletes`](crate::options::ChangeFeedMode::AllVersionsAndDeletes) +/// change feed mode: intermediate versions and deletes are +/// retained for the configured window so they can be read back. Without a +/// retention duration only the default +/// [`LatestVersion`](crate::options::ChangeFeedMode::LatestVersion) mode is +/// available. +/// +/// The retention window has minute granularity. On the service it must fall +/// within the supported range (currently 1 hour to 30 days) when this mode +/// is enabled; this type does not enforce that range, leaving the service as the +/// source of truth. +/// +/// For more information see +#[derive(Clone, Default, SafeDebug, Deserialize, Serialize, PartialEq, Eq)] +#[safe(true)] +#[non_exhaustive] +pub struct ChangeFeedPolicy { + /// The all versions and deletes retention window, in minutes. + /// + /// A value of `0` (the default) disables all versions and deletes, leaving only + /// `LatestVersion` reads available. Kept private so it can only be set through + /// [`all_versions_and_deletes`](ChangeFeedPolicy::all_versions_and_deletes), + /// and clamped to a non-negative value on the wire. + #[serde( + rename = "retentionDuration", + default, + serialize_with = "serialize_retention_minutes" + )] + retention_duration_minutes: i32, +} + +/// Serializes the retention window, clamping any negative value to `0` so a +/// malformed (deserialized) policy can never send a negative window to the +/// service. +fn serialize_retention_minutes(value: &i32, serializer: S) -> Result +where + S: serde::Serializer, +{ + serializer.serialize_i32((*value).max(0)) +} + +impl ChangeFeedPolicy { + /// Creates a change feed policy that enables the + /// [`AllVersionsAndDeletes`](crate::options::ChangeFeedMode::AllVersionsAndDeletes) + /// mode with the given retention window. + /// + /// The window has minute granularity; any non-zero `retention` is rounded + /// **up** to the next whole minute so a sub-minute request never truncates + /// to `0` (which would disable the mode). A zero `retention` leaves the mode + /// disabled. + pub fn all_versions_and_deletes(retention: Duration) -> Self { + Self { + retention_duration_minutes: retention.as_secs().div_ceil(60).min(i32::MAX as u64) + as i32, + } + } + + /// Returns the all versions and deletes retention window, in whole minutes. + /// + /// A value of `0` means all versions and deletes is disabled. + pub fn retention_duration_minutes(&self) -> i32 { + self.retention_duration_minutes.max(0) + } + + /// Returns the all versions and deletes retention window. + /// + /// A zero duration means all versions and deletes is disabled. + pub fn retention_duration(&self) -> Duration { + Duration::from_secs(self.retention_duration_minutes.max(0) as u64 * 60) + } +} + #[cfg(test)] mod tests { use serde::{Deserialize, Serialize}; + use std::time::Duration; - use super::TimeToLive; + use super::{ChangeFeedPolicy, TimeToLive}; use crate::models::ContainerProperties; #[derive(Debug, Deserialize, Serialize)] @@ -549,4 +647,69 @@ mod tests { json ); } + + #[test] + fn change_feed_policy_serializes_retention_minutes() { + let policy = ChangeFeedPolicy::all_versions_and_deletes(Duration::from_secs(5 * 60)); + assert_eq!(5, policy.retention_duration_minutes()); + let json = serde_json::to_string(&policy).unwrap(); + assert_eq!(r#"{"retentionDuration":5}"#, json); + } + + #[test] + fn change_feed_policy_rounds_sub_minute_retention_up() { + // A sub-minute window must round up to 1 minute, not truncate to 0 + // (which would disable the mode). + let policy = ChangeFeedPolicy::all_versions_and_deletes(Duration::from_secs(30)); + assert_eq!(1, policy.retention_duration_minutes()); + assert_eq!(Duration::from_secs(60), policy.retention_duration()); + + // A non-whole-minute window rounds up to the next whole minute. + let policy = ChangeFeedPolicy::all_versions_and_deletes(Duration::from_secs(90)); + assert_eq!(2, policy.retention_duration_minutes()); + assert_eq!(Duration::from_secs(120), policy.retention_duration()); + + // A zero window leaves the mode disabled. + let policy = ChangeFeedPolicy::all_versions_and_deletes(Duration::from_secs(0)); + assert_eq!(0, policy.retention_duration_minutes()); + } + + #[test] + fn change_feed_policy_clamps_negative_retention_on_serialize() { + // A negative window can only arrive by deserializing a malformed policy; + // it must never be sent back to the service as a negative value. + let policy: ChangeFeedPolicy = serde_json::from_str(r#"{"retentionDuration":-5}"#).unwrap(); + assert_eq!(0, policy.retention_duration_minutes()); + assert_eq!(Duration::from_secs(0), policy.retention_duration()); + let json = serde_json::to_string(&policy).unwrap(); + assert_eq!(r#"{"retentionDuration":0}"#, json); + } + + #[test] + fn change_feed_policy_deserializes_retention_minutes() { + let policy: ChangeFeedPolicy = serde_json::from_str(r#"{"retentionDuration":10}"#).unwrap(); + assert_eq!(10, policy.retention_duration_minutes()); + assert_eq!(Duration::from_secs(10 * 60), policy.retention_duration()); + } + + #[test] + fn container_properties_with_change_feed_policy_round_trips() { + let properties = ContainerProperties::new("MyContainer", "/partitionKey".into()) + .with_change_feed_policy(ChangeFeedPolicy::all_versions_and_deletes( + Duration::from_secs(60 * 60), + )); + let json = serde_json::to_string(&properties).unwrap(); + assert!( + json.contains(r#""changeFeedPolicy":{"retentionDuration":60}"#), + "unexpected json: {json}" + ); + + let round_tripped: ContainerProperties = serde_json::from_str(&json).unwrap(); + assert_eq!( + Some(60), + round_tripped + .change_feed_policy + .map(|p| p.retention_duration_minutes()) + ); + } } diff --git a/sdk/cosmos/azure_data_cosmos/src/models/mod.rs b/sdk/cosmos/azure_data_cosmos/src/models/mod.rs index 546d70b99c5..0dbd2369713 100644 --- a/sdk/cosmos/azure_data_cosmos/src/models/mod.rs +++ b/sdk/cosmos/azure_data_cosmos/src/models/mod.rs @@ -18,9 +18,9 @@ pub use change_feed_item::{ ChangeFeedItem, ChangeFeedMetadata, ChangeFeedOperationType, LogicalSequenceNumber, }; pub use container_properties::{ - ConflictResolutionMode, ConflictResolutionPolicy, ContainerProperties, TimeToLive, UniqueKey, - UniqueKeyPolicy, VectorDataType, VectorDistanceFunction, VectorEmbedding, - VectorEmbeddingPolicy, + ChangeFeedPolicy, ConflictResolutionMode, ConflictResolutionPolicy, ContainerProperties, + TimeToLive, UniqueKey, UniqueKeyPolicy, VectorDataType, VectorDistanceFunction, + VectorEmbedding, VectorEmbeddingPolicy, }; pub use database_properties::DatabaseProperties; pub use indexing_policy::{ diff --git a/sdk/cosmos/azure_data_cosmos/src/options/change_feed.rs b/sdk/cosmos/azure_data_cosmos/src/options/change_feed.rs index 9074bd09e7a..b2d4b63b879 100644 --- a/sdk/cosmos/azure_data_cosmos/src/options/change_feed.rs +++ b/sdk/cosmos/azure_data_cosmos/src/options/change_feed.rs @@ -23,15 +23,32 @@ pub use azure_data_cosmos_driver::models::ChangeFeedStartFrom; /// Selects which change feed mode to read. /// -/// Only [`LatestVersion`](Self::LatestVersion) is supported today. The enum is -/// `#[non_exhaustive]`, so additional modes can be added in a future release -/// without breaking callers. +/// * [`LatestVersion`](Self::LatestVersion) (default) returns the latest +/// version of each created or replaced item. +/// * [`AllVersionsAndDeletes`](Self::AllVersionsAndDeletes) returns every +/// intermediate version plus deletes. In this mode +/// `query_change_feed::()` still yields the +/// [`ChangeFeedItem`](crate::models::ChangeFeedItem) envelope, so the +/// whole change envelope is preserved. See +/// [`ContainerClient::query_change_feed()`](crate::clients::ContainerClient::query_change_feed) +/// for its start-position restrictions. #[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] #[non_exhaustive] pub enum ChangeFeedMode { /// Returns the latest version of each changed item ("incremental" feed). #[default] LatestVersion, + + /// Returns every intermediate version plus deletes. + /// + /// Each change is delivered as a + /// [`ChangeFeedItem`](crate::models::ChangeFeedItem) envelope + /// carrying the post-change document (`current`), the pre-change document + /// (`previous`, when available), and change + /// [`metadata`](crate::models::ChangeFeedMetadata). Reading from the + /// beginning is not supported for this mode because intermediate versions + /// and deletes are only retained within the container's retention window. + AllVersionsAndDeletes, } /// Options for change feed operations. diff --git a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_change_feed.rs b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_change_feed.rs index c9fd85d54d1..a7def79de0f 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_change_feed.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/emulator_tests/cosmos_change_feed.rs @@ -16,6 +16,16 @@ //! occurred after the captured position. //! * Resuming a partially-polled `StartFrom::Now` feed does not replay history //! on the partitions that were never polled before the checkpoint. +//! +//! A second group exercises the `AllVersionsAndDeletes` ("full fidelity") mode +//! against a container configured with a change feed retention policy: create, +//! replace, and delete each surface as a distinct [`ChangeFeedItem`] envelope, +//! and the mode reads correctly across a cross-partition fan-out. These AVAD +//! tests are gated on `test_category = "emulator"` only — the vnext (Linux) +//! emulator does not yet support full-fidelity reads. Full-fidelity also +//! requires the account to be provisioned for continuous backup, which the +//! weekly live-test public accounts are not; against those the AVAD tests +//! soft-skip (see [`create_avad_container`]) rather than fail. use super::framework; @@ -23,12 +33,21 @@ use std::error::Error; use std::num::NonZeroU32; use std::time::Duration; +use azure_core::http::StatusCode; +use azure_data_cosmos::clients::{ContainerClient, DatabaseClient}; use azure_data_cosmos::feed::{ChangeFeedPageIterator, ContinuationToken, FeedScope}; -use azure_data_cosmos::models::{ChangeFeedItem, ThroughputProperties}; -use azure_data_cosmos::options::{ChangeFeedOptions, ChangeFeedStartFrom, MaxItemCountHint}; -use framework::{test_data, MockItem, TestClient, TestOptions}; +use azure_data_cosmos::models::{ + ChangeFeedItem, ChangeFeedOperationType, ChangeFeedPolicy, ContainerProperties, + ThroughputProperties, +}; +use azure_data_cosmos::options::{ + ChangeFeedMode, ChangeFeedOptions, ChangeFeedStartFrom, CreateContainerOptions, + MaxItemCountHint, +}; +use framework::{test_data, MockItem, TestClient, TestOptions, TestRunContext}; use futures::StreamExt; use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; use time::OffsetDateTime; /// Maximum number of page polls a drain loop will perform before giving up. @@ -614,3 +633,408 @@ pub async fn change_feed_max_item_count_pages_backlog() -> Result<(), Box, + #[serde(default, skip_serializing_if = "Option::is_none")] + partition_key: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + description: Option, +} + +impl AvadItem { + fn doc(id: &str, partition_key: &str, description: &str) -> Self { + Self { + id: Some(id.to_string()), + partition_key: Some(partition_key.to_string()), + description: Some(description.to_string()), + } + } +} + +/// Creates a container whose change feed policy enables full-fidelity +/// (`AllVersionsAndDeletes`) reads with the given retention window. +/// +/// Returns `Ok(None)` when the target account does not support full-fidelity +/// change feed. Full-fidelity requires the account to be provisioned for +/// continuous backup; accounts without it reject a container carrying a +/// full-fidelity change feed policy at creation time with `400 BadRequest`. +/// The weekly live-test pipeline runs against standard (periodic-backup) +/// public accounts, so on those the AVAD tests soft-skip rather than fail. The +/// policy's wire format is still verified unconditionally by the +/// `container_properties_serialize_change_feed_policy` unit test in the +/// required PR gate, so this skip cannot mask a serialization regression. +async fn create_avad_container( + run_context: &TestRunContext, + db_client: &DatabaseClient, + name: &str, + retention: Duration, + throughput: Option, +) -> azure_data_cosmos::Result> { + let properties = ContainerProperties::new(name.to_string(), "/partitionKey".into()) + .with_change_feed_policy(ChangeFeedPolicy::all_versions_and_deletes(retention)); + let options = throughput.map(|t| CreateContainerOptions::default().with_throughput(t)); + match run_context + .create_container(db_client, properties, options) + .await + { + Ok(container) => Ok(Some(container)), + Err(err) if err.status().status_code() == StatusCode::BadRequest => Ok(None), + Err(err) => Err(err), + } +} + +/// Message logged when an AVAD test soft-skips because the target account does +/// not support full-fidelity change feed (no continuous backup). +const AVAD_UNSUPPORTED_SKIP: &str = + "skipping AllVersionsAndDeletes test: account does not support full-fidelity \ + change feed (continuous backup not enabled)"; + +/// Polls an AVAD change feed, accumulating every envelope seen, until +/// `is_complete` is satisfied by the collection so far or a deadline elapses. +/// +/// Full-fidelity changes — deletes especially — can take a little while to +/// materialize on the emulator, so (unlike the incremental [`drain_changes`] +/// helper) this keeps polling across empty 304 pages rather than stopping at the +/// first empty streak. +async fn drain_avad_until( + iterator: &mut ChangeFeedPageIterator>, + deadline: std::time::Instant, + mut is_complete: F, +) -> Result>, Box> +where + F: FnMut(&[ChangeFeedItem]) -> bool, +{ + let mut collected = Vec::new(); + + while std::time::Instant::now() < deadline { + if is_complete(&collected) { + break; + } + + match iterator.next().await { + Some(page) => { + let page = page?; + if page.items().is_empty() { + tokio::time::sleep(Duration::from_secs(1)).await; + } else { + collected.extend(page.into_items()); + } + } + None => break, + } + } + + Ok(collected) +} + +/// AllVersionsAndDeletes surfaces a create, a replace, and a delete of the same +/// document as three distinct full-fidelity envelopes. +/// +/// This mirrors the .NET SDK's emulator coverage: with only a retention policy +/// configured the service does not return a pre-image, so `previous` is absent +/// on every envelope; the replace's `previousImageLsn` instead chains back to +/// the create's LSN. The delete carries a delete `operationType` and metadata +/// but a minimal `current`. +/// +/// Gated on `test_category = "emulator"` only: full-fidelity reads are not +/// supported by the vnext (Linux) emulator. +#[tokio::test] +#[cfg_attr( + not(test_category = "emulator"), + ignore = "requires test_category 'emulator' (the vnext emulator does not support full-fidelity change feed)" +)] +pub async fn all_versions_and_deletes_surfaces_create_replace_delete() -> Result<(), Box> +{ + TestClient::run_with_unique_db( + async |run_context, db_client| { + let container = create_avad_container( + &run_context, + db_client, + "AvadCreateReplaceDelete", + // The emulator accepts a short (5 minute) full-fidelity retention. + Duration::from_secs(5 * 60), + None, + ) + .await?; + let Some(container) = container else { + eprintln!("{AVAD_UNSUPPORTED_SKIP}"); + return Ok(()); + }; + let pk = "1"; + + let mut iterator = container + .query_change_feed::( + FeedScope::partition(pk), + ChangeFeedStartFrom::Now, + Some( + ChangeFeedOptions::default() + .with_mode(ChangeFeedMode::AllVersionsAndDeletes), + ), + ) + .await?; + + // Prime the `Now` position: the first poll (before any writes) is an + // empty 304 that establishes where the feed starts. + let primed = iterator + .next() + .await + .expect("change feed stream always yields a page")?; + assert!( + primed.items().is_empty(), + "StartFrom::Now must start empty, got {}", + primed.items().len() + ); + + // Create, then replace, then delete the same document. + container + .create_item(pk, "1", &AvadItem::doc("1", pk, "original test"), None) + .await?; + container + .replace_item(pk, "1", &AvadItem::doc("1", pk, "test after replace"), None) + .await?; + container.delete_item(pk, "1", None).await?; + + // Full-fidelity deletes can lag, so poll until the delete arrives. + let deadline = std::time::Instant::now() + Duration::from_secs(180); + let envelopes = drain_avad_until(&mut iterator, deadline, |seen| { + seen.iter() + .any(|e| e.operation_type() == Some(ChangeFeedOperationType::Delete)) + }) + .await?; + + let find_op = |op| { + envelopes + .iter() + .find(move |e| e.operation_type() == Some(op)) + }; + let create = + find_op(ChangeFeedOperationType::Create).expect("a create envelope must surface"); + let replace = + find_op(ChangeFeedOperationType::Replace).expect("a replace envelope must surface"); + let delete = + find_op(ChangeFeedOperationType::Delete).expect("a delete envelope must surface"); + + // Create: `current` holds the original document and there is no + // pre-image. + assert_eq!( + create.current().and_then(|c| c.description.as_deref()), + Some("original test") + ); + assert!( + create.previous().is_none(), + "a create has no previous image" + ); + let create_lsn = create.metadata().and_then(|m| m.lsn()); + assert!(create_lsn.is_some(), "create metadata must carry an LSN"); + + // Replace: `current` holds the new document; the pre-image is not + // returned, but `previousImageLsn` chains back to the create's LSN. + assert_eq!( + replace.current().and_then(|c| c.description.as_deref()), + Some("test after replace") + ); + assert!( + replace.previous().is_none(), + "the retention policy alone does not enable replace pre-images" + ); + assert_eq!( + replace.metadata().and_then(|m| m.previous_image_lsn()), + create_lsn, + "the replace's previousImageLsn must chain to the create's LSN" + ); + + // Delete: a delete `operationType` with metadata; the emulator does + // not return a delete pre-image without an explicit opt-in. + assert_eq!( + delete.operation_type(), + Some(ChangeFeedOperationType::Delete) + ); + assert!( + delete.metadata().and_then(|m| m.lsn()).is_some(), + "delete metadata must carry an LSN" + ); + + Ok(()) + }, + // The internal drain deadline (180s) must sit below the framework's + // per-test timeout, otherwise a lagging full-fidelity delete gets the + // test force-killed mid-poll instead of awaited. + Some(TestOptions::for_emulator().with_timeout(Duration::from_secs(210))), + ) + .await +} + +/// AllVersionsAndDeletes reads fan out across every physical partition: a create +/// on each of many logical partitions surfaces exactly once as a full-fidelity +/// create envelope. The container is provisioned with enough throughput to force +/// multiple physical partitions so the cross-partition merge path is exercised. +/// +/// Gated on `test_category = "emulator"` only: full-fidelity reads are not +/// supported by the vnext (Linux) emulator. +#[tokio::test] +#[cfg_attr( + not(test_category = "emulator"), + ignore = "requires test_category 'emulator' (the vnext emulator does not support full-fidelity change feed)" +)] +pub async fn all_versions_and_deletes_fans_out_creates_across_partitions( +) -> Result<(), Box> { + const PK_COUNT: usize = 10; + + TestClient::run_with_unique_db( + async |run_context, db_client| { + // 11000 RU/s forces the service to create at least 2 physical + // partitions, so the full-container read must fan out. + let container = create_avad_container( + &run_context, + db_client, + "AvadFanOut", + Duration::from_secs(5 * 60), + Some(ThroughputProperties::manual(11000)), + ) + .await?; + let Some(container) = container else { + eprintln!("{AVAD_UNSUPPORTED_SKIP}"); + return Ok(()); + }; + let mut iterator = container + .query_change_feed::( + FeedScope::full_container(), + ChangeFeedStartFrom::Now, + Some( + ChangeFeedOptions::default() + .with_mode(ChangeFeedMode::AllVersionsAndDeletes), + ), + ) + .await?; + + // Prime the per-range `Now` positions before writing. + let primed = iterator + .next() + .await + .expect("change feed stream always yields a page")?; + assert!( + primed.items().is_empty(), + "StartFrom::Now must start empty, got {}", + primed.items().len() + ); + + let mut expected_ids: Vec = Vec::new(); + for p in 0..PK_COUNT { + let partition_key = format!("pk{p}"); + let id = format!("doc{p}"); + expected_ids.push(id.clone()); + container + .create_item( + partition_key.clone(), + &id, + &AvadItem::doc(&id, &partition_key, "created"), + None, + ) + .await?; + } + + let deadline = std::time::Instant::now() + Duration::from_secs(180); + let envelopes = drain_avad_until(&mut iterator, deadline, |seen| { + seen.iter() + .filter(|e| e.operation_type() == Some(ChangeFeedOperationType::Create)) + .count() + >= PK_COUNT + }) + .await?; + + let mut seen_ids: Vec = envelopes + .iter() + .filter(|e| e.operation_type() == Some(ChangeFeedOperationType::Create)) + .filter_map(|e| e.current().and_then(|c| c.id.clone())) + .collect(); + seen_ids.sort(); + expected_ids.sort(); + + assert_eq!( + seen_ids, expected_ids, + "every create must surface exactly once across the fan-out" + ); + Ok(()) + }, + // Raise the per-test timeout above the internal drain deadline (180s) so + // fanning the creates out across every physical partition has room to + // complete instead of being force-killed at the default 80s. + Some(TestOptions::for_emulator().with_timeout(Duration::from_secs(210))), + ) + .await +} + +/// AllVersionsAndDeletes rejects a `PointInTime` start. +/// +/// Full-fidelity reads can only start from "now" or resume from a continuation +/// token within the container's retention / continuous-backup window; the +/// service does not support reading from an arbitrary point in time in this +/// mode. This mirrors the .NET and Java SDKs, which reject +/// `ChangeFeedStartFrom.Time` for all-versions-and-deletes. The client issues +/// the request and the service returns a `BadRequest`. +/// +/// Gated on `test_category = "emulator"` only: full-fidelity reads are not +/// supported by the vnext (Linux) emulator. +#[tokio::test] +#[cfg_attr( + not(test_category = "emulator"), + ignore = "requires test_category 'emulator' (the vnext emulator does not support full-fidelity change feed)" +)] +pub async fn all_versions_and_deletes_rejects_point_in_time_start() -> Result<(), Box> { + TestClient::run_with_unique_db( + async |run_context, db_client| { + let container = create_avad_container( + &run_context, + db_client, + "AvadRejectPointInTime", + Duration::from_secs(5 * 60), + None, + ) + .await?; + let Some(container) = container else { + eprintln!("{AVAD_UNSUPPORTED_SKIP}"); + return Ok(()); + }; + let mut pages = container + .query_change_feed::( + FeedScope::partition("1"), + ChangeFeedStartFrom::PointInTime(OffsetDateTime::now_utc()), + Some( + ChangeFeedOptions::default() + .with_mode(ChangeFeedMode::AllVersionsAndDeletes), + ), + ) + .await?; + + // The client issues the request lazily; the service rejects the + // unsupported start on the first page poll. + let err = pages + .next() + .await + .expect("the change feed should yield a page") + .expect_err("PointInTime start must be rejected for AllVersionsAndDeletes"); + assert_eq!( + StatusCode::BadRequest, + err.status().status_code(), + "expected BadRequest (400) for AVAD + PointInTime, got {:?}", + err.status().status_code() + ); + + Ok(()) + }, + Some(TestOptions::for_emulator()), + ) + .await +} diff --git a/sdk/cosmos/azure_data_cosmos/tests/in_memory_emulator_tests/change_feed.rs b/sdk/cosmos/azure_data_cosmos/tests/in_memory_emulator_tests/change_feed.rs new file mode 100644 index 00000000000..142152c1c1b --- /dev/null +++ b/sdk/cosmos/azure_data_cosmos/tests/in_memory_emulator_tests/change_feed.rs @@ -0,0 +1,280 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +//! In-memory emulator end-to-end tests for the change feed pull API, covering +//! both [`ChangeFeedMode::LatestVersion`] and +//! [`ChangeFeedMode::AllVersionsAndDeletes`]. +//! +//! These drive the full SDK pipeline (`CosmosClient` → `ContainerClient` → +//! driver → in-memory emulator) so they exercise, end-to-end: +//! +//! * `A-IM` header emission and propagation (`Incremental Feed` vs +//! `Full-Fidelity Feed`), +//! * `query_change_feed` mode dispatch, +//! * the `ChangeFeedStartFrom::Beginning` and `ChangeFeedStartFrom::PointInTime` +//! rejections for AllVersionsAndDeletes (gated by the emulator, mirroring the +//! service), and +//! * decoding per mode: callers bind the plain document type `T` and read +//! `ChangeFeedItem` envelopes in both modes. +//! +//! ## What the in-memory emulator models (and what it does not) +//! +//! The in-memory store only retains the *latest* state of each document — it +//! keeps no change log — so it cannot replay historical versions, deletes, or +//! pre-images. For a full-fidelity read it therefore synthesizes a minimal +//! `create` envelope per current document. That is enough to validate the AVAD +//! code path (headers, dispatch, envelope deserialization) deterministically. +//! Delete / `previous`-image envelopes remain covered by the model unit tests +//! in `src/models/change_feed_item.rs`, and true replay is a documented +//! follow-up against a full-fidelity-capable emulator or live account. +//! +//! Because the emulator returns the current state on every poll (no incremental +//! filtering, no 304), these tests poll a **single page** rather than draining +//! to an empty-page streak (which would loop forever collecting duplicates). + +use std::error::Error; + +use azure_data_cosmos::options::{ChangeFeedMode, ChangeFeedOptions, ChangeFeedStartFrom, Region}; +use azure_data_cosmos::{ + AccountEndpoint, AccountReference, ContainerClient, CosmosClient, CosmosClientBuilder, + CosmosRuntimeBuilder, FeedScope, RoutingStrategy, +}; +use azure_data_cosmos_driver::in_memory_emulator::{ + ConsistencyLevel, InMemoryEmulatorHttpClient, VirtualAccountConfig, VirtualRegion, +}; +use futures::StreamExt; +use serde::{Deserialize, Serialize}; + +const EMULATOR_GATEWAY_URL: &str = "https://eastus.emulator.local"; +const DB_NAME: &str = "changefeed-db"; +const CONTAINER_NAME: &str = "changefeed-coll"; +const PK_PATH: &str = "/pk"; +const PARTITION_KEY: &str = "pk1"; + +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +struct TestItem { + id: String, + pk: String, + value: i64, +} + +impl TestItem { + fn new(id: &str, value: i64) -> Self { + Self { + id: id.to_string(), + pk: PARTITION_KEY.to_string(), + value, + } + } +} + +/// Builds an emulator-backed [`CosmosClient`], provisions a single-partition +/// container, and returns a ready-to-use [`ContainerClient`]. +async fn setup() -> Result> { + let config = VirtualAccountConfig::new(vec![VirtualRegion::new( + "East US", + azure_core::http::Url::parse(EMULATOR_GATEWAY_URL)?, + )])? + .with_consistency(ConsistencyLevel::Session); + + let emulator = std::sync::Arc::new(InMemoryEmulatorHttpClient::new(config)); + let store = emulator.store(); + + store.create_database(DB_NAME); + store.create_container( + DB_NAME, + CONTAINER_NAME, + serde_json::from_value(serde_json::json!({ + "paths": [PK_PATH], + "kind": "Hash", + "version": 2 + }))?, + ); + + let account = AccountReference::with_authentication_key( + EMULATOR_GATEWAY_URL.parse::()?, + azure_core::credentials::Secret::new("dGVzdGtleQ=="), + ); + + let client: CosmosClient = CosmosClientBuilder::new() + .with_runtime( + CosmosRuntimeBuilder::from(emulator.runtime_builder()) + .build() + .await?, + ) + .build(account, RoutingStrategy::ProximityTo(Region::EAST_US)) + .await?; + + let container = client + .database_client(DB_NAME) + .container_client(CONTAINER_NAME) + .await?; + + Ok(container) +} + +/// Inserts the given items under [`PARTITION_KEY`]. +async fn insert_items(container: &ContainerClient, items: &[TestItem]) { + for item in items { + container + .create_item(PARTITION_KEY, &item.id, item, None) + .await + .expect("create_item should succeed against the in-memory emulator"); + } +} + +/// LatestVersion reads against the in-memory emulator yield envelopes whose +/// `current` carries the full document. +#[tokio::test] +async fn latest_version_returns_current_documents() { + let container = setup().await.unwrap(); + let items = vec![ + TestItem::new("item-1", 1), + TestItem::new("item-2", 2), + TestItem::new("item-3", 3), + ]; + insert_items(&container, &items).await; + + let mut pages = container + .query_change_feed::( + FeedScope::partition(PARTITION_KEY), + ChangeFeedStartFrom::Beginning, + None, + ) + .await + .unwrap(); + + let page = pages + .next() + .await + .expect("the change feed should yield a page") + .expect("the page should not be an error"); + + let mut returned: Vec = page + .items() + .iter() + .map(|envelope| { + envelope + .current() + .cloned() + .expect("LatestVersion envelopes should carry current documents") + }) + .collect(); + returned.sort_by(|a, b| a.id.cmp(&b.id)); + assert_eq!(returned, items); +} + +/// AllVersionsAndDeletes reads yield full-fidelity envelopes: the whole +/// document is preserved under `current` and the change metadata is populated. +#[tokio::test] +async fn all_versions_and_deletes_returns_envelopes() { + let container = setup().await.unwrap(); + let items = vec![TestItem::new("item-1", 10), TestItem::new("item-2", 20)]; + insert_items(&container, &items).await; + + let options = ChangeFeedOptions::default().with_mode(ChangeFeedMode::AllVersionsAndDeletes); + let mut pages = container + .query_change_feed::( + FeedScope::partition(PARTITION_KEY), + ChangeFeedStartFrom::Now, + Some(options), + ) + .await + .unwrap(); + + let page = pages + .next() + .await + .expect("the change feed should yield a page") + .expect("the page should not be an error"); + + let mut envelopes = page.items().to_vec(); + assert_eq!(envelopes.len(), items.len()); + + // Each envelope must preserve the whole document under `current` and carry + // change metadata. The in-memory emulator synthesizes `create` envelopes. + envelopes.sort_by(|a, b| { + a.current() + .map(|d| d.id.clone()) + .cmp(&b.current().map(|d| d.id.clone())) + }); + for (envelope, expected) in envelopes.iter().zip(items.iter()) { + use azure_data_cosmos::models::ChangeFeedOperationType; + + assert_eq!( + envelope.operation_type(), + Some(ChangeFeedOperationType::Create) + ); + assert_eq!(envelope.current(), Some(expected)); + assert!( + envelope.previous().is_none(), + "the in-memory emulator does not synthesize pre-images" + ); + } +} + +/// AllVersionsAndDeletes rejects `ChangeFeedStartFrom::Beginning`. The client +/// no longer validates this itself; the request is issued and the emulator +/// (standing in for the service) returns `400 Bad Request` on the first poll. +#[tokio::test] +async fn all_versions_and_deletes_rejects_beginning() { + let container = setup().await.unwrap(); + + let options = ChangeFeedOptions::default().with_mode(ChangeFeedMode::AllVersionsAndDeletes); + let mut pages = container + .query_change_feed::( + FeedScope::partition(PARTITION_KEY), + ChangeFeedStartFrom::Beginning, + Some(options), + ) + .await + .unwrap(); + + // A `Beginning` start sends no `If-None-Match`, which the emulator rejects + // for a full-fidelity read with a 400 BadRequest on the first page poll. + let err = pages + .next() + .await + .expect("the change feed should yield a page") + .expect_err("AllVersionsAndDeletes must reject ChangeFeedStartFrom::Beginning"); + + assert_eq!( + u16::from(err.status().status_code()), + 400, + "the rejection must be a 400 BadRequest, got {err:?}" + ); +} + +/// AllVersionsAndDeletes rejects `ChangeFeedStartFrom::PointInTime`. As with +/// `Beginning`, the client defers to the emulator (service stand-in), which +/// returns `400 Bad Request` on the first poll. +#[tokio::test] +async fn all_versions_and_deletes_rejects_point_in_time() { + use time::OffsetDateTime; + + let container = setup().await.unwrap(); + + let options = ChangeFeedOptions::default().with_mode(ChangeFeedMode::AllVersionsAndDeletes); + let mut pages = container + .query_change_feed::( + FeedScope::partition(PARTITION_KEY), + ChangeFeedStartFrom::PointInTime(OffsetDateTime::now_utc()), + Some(options), + ) + .await + .unwrap(); + + // A `PointInTime` start sends `If-Modified-Since`, which the emulator + // rejects for a full-fidelity read with a 400 BadRequest on the first poll. + let err = pages + .next() + .await + .expect("the change feed should yield a page") + .expect_err("AllVersionsAndDeletes must reject ChangeFeedStartFrom::PointInTime"); + + assert_eq!( + u16::from(err.status().status_code()), + 400, + "the rejection must be a 400 BadRequest, got {err:?}" + ); +} diff --git a/sdk/cosmos/azure_data_cosmos/tests/in_memory_emulator_tests/mod.rs b/sdk/cosmos/azure_data_cosmos/tests/in_memory_emulator_tests/mod.rs index e2b905900f5..60e540fac2c 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/in_memory_emulator_tests/mod.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/in_memory_emulator_tests/mod.rs @@ -6,8 +6,8 @@ //! azure_data_cosmos client surface and (optionally) compare against a //! real Cosmos DB account. +pub mod change_feed; use std::time::Duration; - pub mod driver_end_to_end; #[cfg(feature = "preview_dtx")] pub mod dtx_live_comparison; diff --git a/sdk/cosmos/azure_data_cosmos/tests/split_tests/cosmos_change_feed_split.rs b/sdk/cosmos/azure_data_cosmos/tests/split_tests/cosmos_change_feed_split.rs index fc89238848c..7c1ce8d908c 100644 --- a/sdk/cosmos/azure_data_cosmos/tests/split_tests/cosmos_change_feed_split.rs +++ b/sdk/cosmos/azure_data_cosmos/tests/split_tests/cosmos_change_feed_split.rs @@ -1,13 +1,15 @@ // Copyright (c) Microsoft Corporation. All rights reserved. // Licensed under the MIT License. -//! Live split test for the change feed pull API. +//! Live split tests for the change feed pull API. //! //! A continuation token captured before a partition split must resume cleanly //! across the split: every change written after the captured position is //! delivered exactly once — with no replay of pre-token history and no losses — //! even on the post-split child partitions whose parent-partition token is -//! reused. +//! reused. This is covered for both the default `LatestVersion` mode and the +//! `AllVersionsAndDeletes` ("full fidelity") mode, the latter also asserting +//! that post-split deletes surface as full-fidelity delete envelopes. //! //! Forcing a real split is expensive, so this reuses //! [`force_split_and_wait`](super::cosmos_query_split::force_split_and_wait) @@ -21,11 +23,18 @@ use std::error::Error; use std::sync::Arc; use std::time::Duration; +use azure_core::http::StatusCode; use azure_data_cosmos::feed::{ChangeFeedPageIterator, ContinuationToken, FeedScope}; -use azure_data_cosmos::models::{ChangeFeedItem, ContainerProperties, ThroughputProperties}; -use azure_data_cosmos::options::{ChangeFeedOptions, ChangeFeedStartFrom, CreateContainerOptions}; +use azure_data_cosmos::models::{ + ChangeFeedItem, ChangeFeedOperationType, ChangeFeedPolicy, ContainerProperties, + ThroughputProperties, +}; +use azure_data_cosmos::options::{ + ChangeFeedMode, ChangeFeedOptions, ChangeFeedStartFrom, CreateContainerOptions, +}; use framework::{MockItem, TestClient, TestOptions}; use futures::StreamExt; +use serde::{Deserialize, Serialize}; /// Maximum page polls a drain loop performs before giving up, guarding the /// change feed's intentionally infinite stream against looping forever. @@ -202,3 +211,269 @@ pub async fn change_feed_resume_across_split() -> Result<(), Box> { ) .await } + +/// A full-fidelity change feed document tolerant of the minimal `current` a +/// delete envelope carries: both fields are optional so a delete — whose +/// `current` omits (or nulls out) the document fields — still deserializes +/// instead of failing the whole page. +#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +struct AvadDoc { + #[serde(default, skip_serializing_if = "Option::is_none")] + id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + partition_key: Option, +} + +impl AvadDoc { + fn new(id: &str, partition_key: &str) -> Self { + Self { + id: Some(id.to_string()), + partition_key: Some(partition_key.to_string()), + } + } +} + +/// Polls a full-fidelity change feed, accumulating every envelope seen, until +/// `is_complete` is satisfied by the collection so far or a deadline elapses. +/// +/// Full-fidelity changes — deletes especially — can materialize a little while +/// after an empty 304 page, so (unlike the incremental [`drain_changes`] +/// helper) this keeps polling across empty pages, sleeping between them, rather +/// than stopping at the first empty streak and finishing before post-split +/// events arrive. +async fn drain_avad_until( + iterator: &mut ChangeFeedPageIterator>, + deadline: std::time::Instant, + mut is_complete: F, +) -> Result>, Box> +where + F: FnMut(&[ChangeFeedItem]) -> bool, +{ + let mut collected = Vec::new(); + + while std::time::Instant::now() < deadline { + if is_complete(&collected) { + break; + } + + match iterator.next().await { + Some(page) => { + let page = page?; + if page.items().is_empty() { + tokio::time::sleep(Duration::from_secs(1)).await; + } else { + collected.extend(page.into_items()); + } + } + None => break, + } + } + + Ok(collected) +} + +/// The `AllVersionsAndDeletes` ("full fidelity") change feed resumes correctly +/// across a partition split: a token captured before the split, from a container +/// with a change feed retention policy, yields exactly the post-split creates +/// and deletes — each surfacing as a full-fidelity envelope — with no replay of +/// the pre-token baseline. +/// +/// This exercises the AVAD variant of the cross-partition resume path: the mode +/// (and its `Full-Fidelity Feed` A-IM header) must be re-applied on resume so the +/// pre-split parent token maps onto the post-split children while partitions +/// without a saved continuation re-apply the feed's original `Now` position. +#[tokio::test] +#[cfg_attr( + not(test_category = "split"), + ignore = "requires test_category 'split'" +)] +pub async fn change_feed_all_versions_and_deletes_resume_across_split() -> Result<(), Box> +{ + const PK_COUNT: usize = 30; + const BASELINE_PER_PK: usize = 2; + const NEW_PER_PK: usize = 2; + + TestClient::run_with_unique_db( + async |run_context, db_client| { + let properties = ContainerProperties::new( + "ChangeFeedAvadResumeAcrossSplit", + "/partitionKey".into(), + ) + .with_change_feed_policy(ChangeFeedPolicy::all_versions_and_deletes( + Duration::from_secs(60 * 60), + )); + let throughput = ThroughputProperties::manual(1000); + let container_client = match run_context + .create_container( + db_client, + properties, + Some(CreateContainerOptions::default().with_throughput(throughput)), + ) + .await + { + Ok(container) => Arc::new(container), + // Full-fidelity change feed requires the account to be provisioned + // for continuous backup; standard (periodic-backup) public test + // accounts reject a container carrying an AVAD change feed policy at + // creation with `400 BadRequest`. Soft-skip there rather than fail. + // The policy's wire format is still verified unconditionally by the + // `container_properties_serialize_change_feed_policy` unit test in the + // required PR gate, so this skip cannot mask a serialization regression. + Err(err) if err.status().status_code() == StatusCode::BadRequest => { + eprintln!( + "skipping AllVersionsAndDeletes split test: account does not \ + support full-fidelity change feed (continuous backup not enabled)" + ); + return Ok(()); + } + Err(err) => return Err(err.into()), + }; + + // Seed the baseline (pre-split, pre-token) creates. AVAD `Now` must + // exclude these, and they must never replay after the split. + for p in 0..PK_COUNT { + let partition_key = format!("pk{p}"); + for i in 0..BASELINE_PER_PK { + let id = format!("baseline-{p}-{i}"); + container_client + .create_item(partition_key.clone(), &id, &AvadDoc::new(&id, &partition_key), None) + .await?; + } + } + + let partitions_before = container_client.read_feed_ranges(None).await?.len(); + assert!( + partitions_before >= 1, + "expected at least one physical partition before the split" + ); + + // Open an AVAD feed at `Now` (Beginning is rejected for AVAD) and + // drain to a caught-up state to capture a resume token. `Now` + // excludes the baseline, so nothing should be collected here. + let mut iterator = container_client + .query_change_feed::( + FeedScope::full_container(), + ChangeFeedStartFrom::Now, + Some( + ChangeFeedOptions::default() + .with_mode(ChangeFeedMode::AllVersionsAndDeletes), + ), + ) + .await?; + let pre_split = drain_avad_until( + &mut iterator, + std::time::Instant::now() + Duration::from_secs(15), + |_| false, + ) + .await?; + assert!( + pre_split.is_empty(), + "AVAD StartFrom::Now must exclude the pre-existing baseline, saw {}", + pre_split.len() + ); + + let token = iterator.to_continuation_token()?; + let token = ContinuationToken::from_string(token.as_str().to_owned()); + drop(iterator); + + // Force a real split AFTER the token is captured so the resume must + // map the pre-split parent token onto the post-split children. + let partitions_after = + force_split_and_wait(&container_client, partitions_before).await?; + assert!( + partitions_after > partitions_before, + "split must increase partition count: before={partitions_before}, after={partitions_after}" + ); + + // After the split: create new docs per pk and delete one baseline doc + // per pk. Both are changes AFTER the token, so both must resume. + let mut expected_creates: Vec = Vec::new(); + let mut expected_deletes = 0usize; + for p in 0..PK_COUNT { + let partition_key = format!("pk{p}"); + for i in 0..NEW_PER_PK { + let id = format!("post-{p}-{i}"); + expected_creates.push(id.clone()); + container_client + .create_item(partition_key.clone(), &id, &AvadDoc::new(&id, &partition_key), None) + .await?; + } + let delete_id = format!("baseline-{p}-0"); + container_client + .delete_item(partition_key.clone(), &delete_id, None) + .await?; + expected_deletes += 1; + } + + // Resume from the pre-split token. The mode must be re-applied so the + // resumed request also reads full fidelity. + let mut resumed = container_client + .query_change_feed::( + FeedScope::full_container(), + // Ignored on resume: the token carries its own position. + ChangeFeedStartFrom::Now, + Some( + ChangeFeedOptions::default() + .with_mode(ChangeFeedMode::AllVersionsAndDeletes) + .with_continuation_token(token), + ), + ) + .await?; + // Resume until every post-split create and delete has surfaced (or a + // deadline expires), sleeping between empty pages so full-fidelity + // deletes that lag behind their creates are not missed. + let want_creates = expected_creates.len(); + let want_deletes = expected_deletes; + let resumed_envelopes = drain_avad_until( + &mut resumed, + std::time::Instant::now() + Duration::from_secs(5 * 60), + |seen| { + let creates = seen + .iter() + .filter(|e| e.operation_type() == Some(ChangeFeedOperationType::Create)) + .count(); + let deletes = seen + .iter() + .filter(|e| e.operation_type() == Some(ChangeFeedOperationType::Delete)) + .count(); + creates >= want_creates && deletes >= want_deletes + }, + ) + .await?; + + // Exactly the post-split creates must surface, once each. + let mut got_creates: Vec = resumed_envelopes + .iter() + .filter(|e| e.operation_type() == Some(ChangeFeedOperationType::Create)) + .filter_map(|e| e.current().and_then(|c| c.id.clone())) + .collect(); + got_creates.sort(); + expected_creates.sort(); + assert_eq!( + got_creates, expected_creates, + "resume across split must deliver exactly the post-split creates once each" + ); + + // Every post-split delete must surface as a full-fidelity delete + // envelope. + let got_deletes = resumed_envelopes + .iter() + .filter(|e| e.operation_type() == Some(ChangeFeedOperationType::Delete)) + .count(); + assert_eq!( + got_deletes, expected_deletes, + "every post-split delete must surface as a full-fidelity delete envelope" + ); + + // No baseline create may replay: they were committed before the token. + assert!( + !got_creates.iter().any(|id| id.starts_with("baseline-")), + "baseline creates must not replay after resume" + ); + Ok(()) + }, + Some(TestOptions::new().with_timeout(Duration::from_secs(40 * 60))), + ) + .await +} diff --git a/sdk/cosmos/azure_data_cosmos_driver/CHANGELOG.md b/sdk/cosmos/azure_data_cosmos_driver/CHANGELOG.md index fa1d74748a4..6949b035fef 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/CHANGELOG.md +++ b/sdk/cosmos/azure_data_cosmos_driver/CHANGELOG.md @@ -21,6 +21,7 @@ - Added per-partition hub region caching for read operations on PPAF-enabled single-master accounts. When the `x-ms-cosmos-hub-region-processing-only` latch is set on a retry and a partition key range ID has been resolved, the driver now consults the per-partition failover cache (`PartitionEndpointState::failover_overrides` — the same map already used by per-partition automatic failover for writes) and routes directly to the cached hub endpoint, skipping the `403/3 (WriteForbidden)` discovery chain on subsequent operations for the same partition. ([#4555](https://github.com/Azure/azure-sdk-for-rust/pull/4555)) - Added preview distributed transaction driver models, request serialization, response parsing/reordering, strict session-token merge, DTX retry handling, diagnostics, and a baseline in-memory emulator `/operations/dtc` path behind the disabled-by-default `preview_dtx` feature. ([#4702](https://github.com/Azure/azure-sdk-for-rust/pull/4702)) - Added change feed support in the dataflow pipeline: a new `UnorderedMerge` node fans a change feed read out across physical partitions and round-robins their pages, and `CosmosOperation::change_feed` builds incremental-feed operations with the appropriate wire headers. A new public `ChangeFeedStartFrom` enum (`Beginning`, `Now`, `PointInTime`) records the feed's original start position and is persisted in the continuation token so partitions never polled before a checkpoint re-apply it on resume instead of replaying history; set it via `CosmosOperation::with_change_feed_start`. ([#4621](https://github.com/Azure/azure-sdk-for-rust/pull/4621)) +- Added full-fidelity (AllVersionsAndDeletes) change feed support: a new `CosmosOperation::change_feed_all_versions_and_deletes` factory and a `CosmosRequestHeaders::full_fidelity_feed` flag emit `A-IM: Full-Fidelity Feed` (in place of `Incremental Feed`) so the service returns every intermediate version and delete as an envelope with `current`, `previous`, and `metadata`. The continuation token records the feed mode so a resume cannot silently switch between incremental and full-fidelity, and the in-memory emulator rejects the unsupported `Beginning` and `PointInTime` starts for this mode with `400 Bad Request`. ([#4706](https://github.com/Azure/azure-sdk-for-rust/pull/4706)) - Added `TlsBackend` (currently `TlsBackend::Rustls`, the default) and a `tls_backend` option on `ConnectionPoolOptions` (`ConnectionPoolOptionsBuilder::with_tls_backend` / `ConnectionPoolOptions::tls_backend`), available under the `rustls` feature. The driver asserts the selected backend on the `reqwest` transport, giving a supported way to pin the TLS backend without direct transport access. This is additive and changes no behavior for the default (rustls-only) build, where reqwest already negotiates rustls; it only has an effect in builds that compile in multiple reqwest TLS backends (e.g. `rustls` plus `native_tls`, absent reqwest's `http3` feature), 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 diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/integration_tests/change_feed_resume.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/integration_tests/change_feed_resume.rs index 87697597a17..3397fea4a4a 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/integration_tests/change_feed_resume.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/integration_tests/change_feed_resume.rs @@ -83,6 +83,19 @@ fn change_feed_operation(start_from: Option) -> Arc) -> Arc { + let mut op = CosmosOperation::change_feed_all_versions_and_deletes( + test_container(), + Some(FeedRange::full()), + ); + if let Some(marker) = start_from { + op = op.with_change_feed_start(marker); + } + Arc::new(op) +} + fn resolved(min: &str, max: &str, pk_range_id: &str) -> ResolvedRange { ResolvedRange { partition_key_range_id: pk_range_id.to_string(), @@ -319,3 +332,220 @@ async fn change_feed_resume_across_merge_reads_each_parent_subrange() { "each merged leaf must carry its parent's EPK sub-range", ); } + +/// Guards the AllVersionsAndDeletes lossless-`Now` contract: a fresh +/// full-fidelity feed must pin **every** range to a concrete starting ETag +/// before any checkpoint, so a range that is never served before a checkpoint +/// still resumes from its true starting position instead of a resume-time +/// `Now` (which would silently drop the versions and deletes in the gap). +/// +/// Two partitions exist, but session 1 pulls only a single page — round-robin +/// alone would poll just the left range. Priming must poll the right range too, +/// so the checkpoint carries an ETag for both. On resume both ranges re-send +/// their pinned ETags rather than restarting. +/// +/// The LatestVersion (incremental) feed deliberately does **not** prime: a +/// never-polled range there simply re-reads from the persisted start on resume, +/// which is benign because incremental only surfaces the latest version. +#[tokio::test] +async fn all_versions_and_deletes_pins_every_range_before_checkpoint() { + let op = avad_change_feed_operation(Some(ChangeFeedStartFrom::Now)); + + // Session 1: two partitions. Priming polls left then right (each a + // start-from-`Now` 304 carrying only an ETag); the round-robin then serves + // the left range's next poll. Three polls total for one served page. + let mut topology1 = MockTopologyProvider::new(vec![Ok(vec![ + resolved("", "80", "pk-left"), + resolved("80", "FF", "pk-right"), + ])]); + let mut executor1 = MockRequestExecutor::new(vec![ + Ok(cf_page(b"", "lsn-left-0")), // prime left + Ok(cf_page(b"", "lsn-right-0")), // prime right (never served) + Ok(cf_page(b"left-1", "lsn-left-1")), // served page + ]); + + let mut pipeline1 = build_unordered_merge(&FeedRange::full(), &mut topology1, &op, None) + .await + .unwrap(); + let pages1 = drain_pages(&mut pipeline1, &mut executor1, 1).await; + assert_eq!(pages1, vec![b"left-1".to_vec()]); + assert_eq!( + executor1.continuation_calls, + vec![None, None, Some("lsn-left-0".to_owned())], + "priming starts both ranges fresh; the served left poll continues from its primed ETag", + ); + + // The snapshot must record an ETag for BOTH ranges — including the right + // range that was only primed, never served — so neither resumes from `Now`. + let state = pipeline1.snapshot_state().unwrap(); + match &state { + PipelineNodeState::UnorderedMerge { + active_tokens, + start_from, + } => { + assert_eq!( + active_tokens.len(), + 2, + "both ranges must be pinned, got {active_tokens:?}", + ); + assert_eq!(active_tokens[0].max_epk, "80"); + assert_eq!(active_tokens[0].server_continuation, "lsn-left-1"); + assert_eq!(active_tokens[1].min_epk, "80"); + assert_eq!( + active_tokens[1].server_continuation, "lsn-right-0", + "the never-served right range must still carry its primed ETag", + ); + assert_eq!(*start_from, Some(ChangeFeedStartFrom::Now)); + } + other => panic!("expected UnorderedMerge snapshot, got {other:?}"), + } + drop(pipeline1); + + // Session 2: resume. Resume does not prime (every range already carries a + // saved ETag). Both ranges must re-send their pinned ETags — the right one + // in particular must NOT restart from `Now` and drop its gap. + let resumed_state = round_trip_state(state, &op); + let mut topology2 = MockTopologyProvider::new(vec![Ok(vec![ + resolved("", "80", "pk-left"), + resolved("80", "FF", "pk-right"), + ])]); + let mut executor2 = MockRequestExecutor::new(vec![ + Ok(cf_page(b"left-2", "lsn-left-2")), + Ok(cf_page(b"right-2", "lsn-right-2")), + ]); + + let mut pipeline2 = + build_unordered_merge(&FeedRange::full(), &mut topology2, &op, Some(resumed_state)) + .await + .unwrap(); + let pages2 = drain_pages(&mut pipeline2, &mut executor2, 2).await; + assert_eq!(pages2, vec![b"left-2".to_vec(), b"right-2".to_vec()]); + assert_eq!( + executor2.continuation_calls, + vec![ + Some("lsn-left-1".to_owned()), + Some("lsn-right-0".to_owned()) + ], + "resume must re-send each range's pinned ETag, not restart from Now", + ); +} + +/// The LatestVersion feed must **not** prime: a single page pull polls only one +/// range, and the never-polled range relies on the persisted `start_from` on +/// resume. This is the counterpart to +/// [`all_versions_and_deletes_pins_every_range_before_checkpoint`] and pins the +/// mode-gated behavior so priming can't accidentally leak into incremental. +#[tokio::test] +async fn latest_version_does_not_prime_ranges() { + let op = change_feed_operation(Some(ChangeFeedStartFrom::Now)); + + let mut topology1 = MockTopologyProvider::new(vec![Ok(vec![ + resolved("", "80", "pk-left"), + resolved("80", "FF", "pk-right"), + ])]); + // Only one response is available: if priming were (incorrectly) enabled the + // mock would be polled twice and panic on the missing second response. + let mut executor1 = MockRequestExecutor::new(vec![Ok(cf_page(b"left-1", "lsn-left-1"))]); + + let mut pipeline1 = build_unordered_merge(&FeedRange::full(), &mut topology1, &op, None) + .await + .unwrap(); + let pages1 = drain_pages(&mut pipeline1, &mut executor1, 1).await; + assert_eq!(pages1, vec![b"left-1".to_vec()]); + assert_eq!( + executor1.continuation_calls, + vec![None], + "incremental polls a single range per page — no priming", + ); + + // Only the polled range is pinned; the other relies on `start_from`. + let state = pipeline1.snapshot_state().unwrap(); + match &state { + PipelineNodeState::UnorderedMerge { active_tokens, .. } => { + assert_eq!(active_tokens.len(), 1, "got {active_tokens:?}"); + assert_eq!(active_tokens[0].server_continuation, "lsn-left-1"); + } + other => panic!("expected UnorderedMerge snapshot, got {other:?}"), + } +} + +/// Guards the pre-first-page window of the AllVersionsAndDeletes lossless-`Now` +/// contract: a checkpoint taken **before** the first page is pulled snapshots an +/// empty token set (nothing has been polled yet). Resuming from that empty set +/// must still prime every range, otherwise every range would fall back to a +/// resume-time `Now` and drop the versions/deletes in the gap — the same data +/// loss the priming fix closes, relocated to the pre-first-page window. +/// +/// A fully drained feed instead snapshots `PipelineNodeState::Drained`, so an +/// empty `UnorderedMerge` token set unambiguously means "not yet polled" and is +/// safe to treat as still-needs-priming. +#[tokio::test] +async fn all_versions_and_deletes_pins_ranges_when_resumed_before_first_page() { + let op = avad_change_feed_operation(Some(ChangeFeedStartFrom::Now)); + + // Session 1: build the pipeline and checkpoint immediately, before pulling + // any page. No range has been polled, so the snapshot carries no tokens. + let mut topology1 = MockTopologyProvider::new(vec![Ok(vec![ + resolved("", "80", "pk-left"), + resolved("80", "FF", "pk-right"), + ])]); + let pipeline1 = build_unordered_merge(&FeedRange::full(), &mut topology1, &op, None) + .await + .unwrap(); + let state = pipeline1.snapshot_state().unwrap(); + match &state { + PipelineNodeState::UnorderedMerge { + active_tokens, + start_from, + } => { + assert!( + active_tokens.is_empty(), + "no range polled yet, expected no tokens, got {active_tokens:?}", + ); + assert_eq!(*start_from, Some(ChangeFeedStartFrom::Now)); + } + other => panic!("expected UnorderedMerge snapshot, got {other:?}"), + } + drop(pipeline1); + + // Session 2: resume from the empty-token checkpoint. Priming must still run + // (the token set is empty), pinning BOTH ranges before the first page. + let resumed_state = round_trip_state(state, &op); + let mut topology2 = MockTopologyProvider::new(vec![Ok(vec![ + resolved("", "80", "pk-left"), + resolved("80", "FF", "pk-right"), + ])]); + let mut executor2 = MockRequestExecutor::new(vec![ + Ok(cf_page(b"", "lsn-left-0")), // prime left + Ok(cf_page(b"", "lsn-right-0")), // prime right (never served) + Ok(cf_page(b"left-1", "lsn-left-1")), // served page + ]); + + let mut pipeline2 = + build_unordered_merge(&FeedRange::full(), &mut topology2, &op, Some(resumed_state)) + .await + .unwrap(); + let pages2 = drain_pages(&mut pipeline2, &mut executor2, 1).await; + assert_eq!(pages2, vec![b"left-1".to_vec()]); + assert_eq!( + executor2.continuation_calls, + vec![None, None, Some("lsn-left-0".to_owned())], + "resume-before-first-page must prime both ranges fresh, not skip priming", + ); + + // The follow-up checkpoint now pins both ranges — including the right range + // that was only primed — so a subsequent resume is lossless. + let state2 = pipeline2.snapshot_state().unwrap(); + match &state2 { + PipelineNodeState::UnorderedMerge { active_tokens, .. } => { + assert_eq!( + active_tokens.len(), + 2, + "both ranges must be pinned after priming, got {active_tokens:?}", + ); + assert_eq!(active_tokens[1].min_epk, "80"); + assert_eq!(active_tokens[1].server_continuation, "lsn-right-0"); + } + other => panic!("expected UnorderedMerge snapshot, got {other:?}"), + } +} diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/mocks.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/mocks.rs index 029665dd9a1..43a961ad19f 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/mocks.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/mocks.rs @@ -5,7 +5,7 @@ use std::{collections::VecDeque, sync::Arc}; -use azure_core::http::StatusCode; +use azure_core::http::{Etag, StatusCode}; use futures::future::BoxFuture; use super::{ @@ -263,6 +263,25 @@ pub(crate) fn response_with_continuation( ) } +/// Creates a test response with the given body and an `ETag`, mirroring the +/// change feed contract where every poll (including a start-from-`Now` 304) +/// carries an ETag continuation. +pub(crate) fn response_with_etag(body: &[u8], etag: &str) -> CosmosResponse { + let mut diagnostics = DiagnosticsContextBuilder::new( + ActivityId::new_uuid(), + Arc::new(DiagnosticsOptions::default()), + ); + diagnostics.set_operation_status(StatusCode::Ok, None); + let mut headers = CosmosResponseHeaders::new(); + headers.etag = Some(Etag::from(etag.to_owned())); + CosmosResponse::new( + body.to_vec(), + headers, + CosmosStatus::new(StatusCode::Ok), + Arc::new(diagnostics.complete()), + ) +} + /// Creates a 410 Gone error with a partition topology change substatus. pub(crate) fn gone_error() -> crate::error::CosmosError { crate::error::CosmosError::builder() diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/planner.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/planner.rs index 716476653c4..ffb2d987e26 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/planner.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/planner.rs @@ -262,6 +262,20 @@ pub(crate) async fn build_unordered_merge( operation.change_feed_start().cloned() }; + // Full-fidelity (AllVersionsAndDeletes) feeds must pin every range to a + // concrete starting continuation before the first checkpoint, so a range + // that is never polled can't resume from a stale `Now` and drop the + // versions/deletes in the gap. Priming is needed whenever no range has yet + // recorded a continuation: on a fresh start, and also on resume from a + // checkpoint taken *before* the first page was pulled (an empty token set). + // A fully drained resume arrives as `PipelineNodeState::Drained` and is + // handled earlier, so an empty `UnorderedMerge` token set here only ever + // means "nothing has been polled yet". A resume that already carries saved + // continuations must NOT prime: those ranges would re-poll from their real + // ETags and the discarded primed page would lose real data. + let prime_on_first_drain = operation.request_headers().full_fidelity_feed + && saved_tokens.as_ref().is_none_or(|tokens| tokens.is_empty()); + // On resume the operation rebuilt by the SDK no longer carries the original // start headers (the caller only passed the continuation token). Re-derive // them from the persisted marker so partitions with no saved continuation @@ -345,7 +359,11 @@ pub(crate) async fn build_unordered_merge( .build()); } - let root = Box::new(UnorderedMerge::new(request_nodes).with_start_marker(start_marker)); + let root = Box::new( + UnorderedMerge::new(request_nodes) + .with_start_marker(start_marker) + .with_prime_on_first_drain(prime_on_first_drain), + ); Ok(Pipeline::new(root)) } diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/unordered_merge.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/unordered_merge.rs index 5a79703b467..ef9c9843dd6 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/unordered_merge.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/dataflow/unordered_merge.rs @@ -35,6 +35,22 @@ pub(crate) struct UnorderedMerge { /// polled re-apply it on resume instead of reading from the beginning. /// `None` means no start position was recorded. start_marker: Option, + /// When set, poll every child once before serving the first page so each + /// range records a concrete starting continuation (ETag) up front. + /// + /// Enabled only for a fresh AllVersionsAndDeletes feed. Without it, a range + /// that is never polled before the first checkpoint has no saved + /// continuation and resumes from the persisted `start_marker` (`Now`), + /// which re-evaluates "now" at resume time and silently drops the + /// intermediate versions and deletes that occurred in the gap. Priming + /// pins every range to its actual starting position so resume is lossless. + /// + /// A fresh AllVersionsAndDeletes feed can only start from `Now` + /// (`If-None-Match: *`; `Beginning`/`PointInTime` are rejected by the + /// service), whose first poll is always a `304 Not Modified` carrying no + /// items — only an ETag. Priming therefore captures each range's starting + /// ETag without dropping any change. Cleared after the priming pass runs. + prime_on_first_drain: bool, } impl UnorderedMerge { @@ -44,6 +60,7 @@ impl UnorderedMerge { children: children.into(), cursor: 0, start_marker: None, + prime_on_first_drain: false, } } @@ -52,6 +69,92 @@ impl UnorderedMerge { self.start_marker = start_marker; self } + + /// Enables priming every child once before the first page is served. + /// + /// See [`prime_on_first_drain`](Self::prime_on_first_drain). Set for a fresh + /// AllVersionsAndDeletes feed so no range can resume from a stale `Now`. + pub(crate) fn with_prime_on_first_drain(mut self, prime: bool) -> Self { + self.prime_on_first_drain = prime; + self + } + + /// Polls every child once so each records its starting continuation. + /// + /// Only invoked for a fresh AllVersionsAndDeletes feed, whose first + /// (start-from-`Now`) poll per range is a `304` with no items. The primed + /// page is discarded; only the ETag it advanced — now held by the child — + /// is retained, so the next real poll of that range resumes from its true + /// starting position rather than from a resume-time `Now`. + /// + /// # Cost + /// + /// Priming front-loads one poll per range before the first page is served, + /// so a wide fan-out AVAD feed pays N sequential round-trips (and N request + /// units) of extra first-page latency. This is the price of the lossless- + /// `Now` guarantee; LatestVersion feeds don't prime and don't pay it. Each + /// priming poll goes through the same per-request path as a normal poll, so + /// transient failures are retried by the pipeline's retry policy beneath + /// this call. A non-retryable error on any single range aborts the whole + /// first page (the `?` below) rather than starting the feed with an + /// unpinned range, keeping the lossless guarantee all-or-nothing. + async fn prime_children( + &mut self, + context: &mut PipelineContext<'_>, + ) -> crate::error::Result<()> { + let mut idx = 0; + while idx < self.children.len() { + let mut split_retries = 0; + loop { + match self.children[idx].next_page(context).await? { + PageResult::Page { response, .. } => { + // Start-from-`Now` first poll: a 304 carrying only an + // ETag, which the child has now recorded. Discarding the + // (item-less) page loses nothing; advance to the next + // child. The ETag is what makes priming worthwhile — a + // 304 with no ETag would leave the child unpinned and + // silently resume from `Now`, so assert its presence to + // catch a service/transport contract violation in debug + // builds. + debug_assert!( + response.headers().etag.is_some(), + "priming poll returned a page without an ETag; the range \ + cannot record its start position and would resume from `Now`" + ); + idx += 1; + break; + } + PageResult::Drained => { + // A fully drained child contributes nothing; drop it and + // stay at the same index (the next child shifts in). + self.children.remove(idx); + break; + } + PageResult::SplitRequired { replacement_nodes } => { + split_retries += 1; + if split_retries > MAX_SPLIT_RETRIES { + return Err(crate::error::CosmosError::builder() + .with_status( + crate::error::CosmosStatus::CLIENT_SPLIT_RETRIES_EXHAUSTED, + ) + .with_message(format!( + "exceeded maximum split retries ({MAX_SPLIT_RETRIES}) \ + while priming UnorderedMerge children" + )) + .build()); + } + // Splice the replacement ranges in place and re-prime + // from the first replacement (same index). + self.children.remove(idx); + for (i, node) in replacement_nodes.into_iter().enumerate() { + self.children.insert(idx + i, node); + } + } + } + } + } + Ok(()) + } } #[async_trait] @@ -64,6 +167,17 @@ impl PipelineNode for UnorderedMerge { return Ok(PageResult::Drained); } + // For a fresh AllVersionsAndDeletes feed, poll every range once up front + // so each records its concrete starting continuation before any + // checkpoint can be taken. This runs exactly once. + if self.prime_on_first_drain { + self.prime_on_first_drain = false; + self.prime_children(context).await?; + if self.children.is_empty() { + return Ok(PageResult::Drained); + } + } + let mut split_retries = 0; let children_count = self.children.len(); // Try up to `children_count` children to find one with data. @@ -338,4 +452,93 @@ mod tests { other => panic!("expected Page, got {other:?}"), } } + + #[tokio::test] + async fn prime_on_first_drain_polls_every_child_once() { + // Child 0 has a second page; children 1 and 2 have one each. Priming + // must poll every child once (consuming a1, b1, c1) before the + // round-robin serves a page, so the first served page is child 0's + // *second* poll (a2) rather than a1. + let child_a = MockLeaf::with_pages(vec![ + Ok(PageResult::Page { + response: response_with_etag(b"a1", "etag-a1"), + is_terminal: true, + }), + Ok(PageResult::Page { + response: response(b"a2"), + is_terminal: false, + }), + ]); + let child_b = MockLeaf::with_pages(vec![Ok(PageResult::Page { + response: response_with_etag(b"b1", "etag-b1"), + is_terminal: true, + })]); + let child_c = MockLeaf::with_pages(vec![Ok(PageResult::Page { + response: response_with_etag(b"c1", "etag-c1"), + is_terminal: true, + })]); + + let mut merge = UnorderedMerge::new(vec![ + Box::new(child_a), + Box::new(child_b), + Box::new(child_c), + ]) + .with_prime_on_first_drain(true); + let mut executor = NoopRequestExecutor; + let mut topology = NoopTopologyProvider; + let mut ctx = PipelineContext::new(&mut executor, Some(&mut topology)); + + let r = merge.next_page(&mut ctx).await.unwrap(); + match r { + PageResult::Page { response, .. } => { + assert_eq!( + response.body_bytes(), + b"a2", + "priming must consume every child's first poll before serving a page" + ); + } + other => panic!("expected Page, got {other:?}"), + } + } + + #[tokio::test] + async fn prime_on_first_drain_handles_split() { + // A range that has split before its first poll must be primed through + // its replacements: priming polls each replacement once before the + // round-robin serves a page. + let split_child = MockLeaf::with_pages(vec![Ok(PageResult::SplitRequired { + replacement_nodes: vec![ + Box::new(MockLeaf::with_pages(vec![ + Ok(PageResult::Page { + response: response_with_etag(b"ra1", "etag-ra1"), + is_terminal: true, + }), + Ok(PageResult::Page { + response: response(b"ra2"), + is_terminal: false, + }), + ])), + Box::new(MockLeaf::with_pages(vec![Ok(PageResult::Page { + response: response_with_etag(b"rb1", "etag-rb1"), + is_terminal: true, + })])), + ], + })]); + + let mut merge = + UnorderedMerge::new(vec![Box::new(split_child)]).with_prime_on_first_drain(true); + let mut executor = NoopRequestExecutor; + let mut topology = NoopTopologyProvider; + let mut ctx = PipelineContext::new(&mut executor, Some(&mut topology)); + + // Priming splices in the two replacements and polls each once (ra1, + // rb1); the served page is the first replacement's second poll (ra2). + let r = merge.next_page(&mut ctx).await.unwrap(); + match r { + PageResult::Page { response, .. } => { + assert_eq!(response.body_bytes(), b"ra2"); + } + other => panic!("expected Page, got {other:?}"), + } + } } 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 b02fdb947ed..291f206e057 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 @@ -1269,6 +1269,7 @@ fn resolve_endpoint( && is_operation_supported_by_gateway_v2( operation.resource_type(), operation.operation_type(), + operation.request_headers().full_fidelity_feed, ); let transport_mode = if use_gateway_v2 { TransportMode::GatewayV2 @@ -1297,6 +1298,7 @@ fn resolve_endpoint( && is_operation_supported_by_gateway_v2( operation.resource_type(), operation.operation_type(), + operation.request_headers().full_fidelity_feed, ); let ep_url = ep.selected_url(ep_use_gw_v2).clone(); let ep_endpoint_key = if ep_use_gw_v2 { @@ -5412,6 +5414,72 @@ mod tests { assert_eq!(routing.selected_url, *endpoint.url()); } + #[test] + fn resolve_endpoint_falls_back_to_gateway_for_full_fidelity_change_feed() { + // A full-fidelity (AllVersionsAndDeletes) change feed is a + // `Document`/`ReadFeed` op — otherwise Gateway 2.0 eligible — but must + // route through the standard gateway because Gateway 2.0 does not + // forward the `A-IM` header. An incremental change feed on the same + // endpoint stays on Gateway 2.0. + let full_fidelity = CosmosOperation::change_feed_all_versions_and_deletes( + test_container(), + Some(FeedRange::full()), + ); + let incremental = CosmosOperation::change_feed(test_container(), Some(FeedRange::full())); + let endpoint = CosmosEndpoint::regional_with_gateway_v2( + "westus2".into(), + Url::parse("https://test-westus2.documents.azure.com:443/").unwrap(), + Url::parse("https://test-westus2-thin.documents.azure.com:444/").unwrap(), + ); + + let location = LocationSnapshot::for_tests(Arc::new(AccountEndpointState { + generation: 0, + preferred_read_endpoints: vec![endpoint.clone()].into(), + preferred_write_endpoints: vec![endpoint.clone()].into(), + account_write_endpoints: vec![endpoint.clone()].into(), + unavailable_endpoints: Default::default(), + multiple_write_locations_enabled: false, + default_endpoint: endpoint.clone(), + })); + + let retry_state = crate::driver::pipeline::components::OperationRetryState::initial( + 0, + false, + Vec::new(), + 3, + 2, + ); + + let full_fidelity_routing = super::resolve_endpoint( + &full_fidelity, + &retry_state, + &location, + true, + true, + Duration::from_secs(60), + ); + assert_eq!( + full_fidelity_routing.transport_mode, + TransportMode::Gateway, + "full-fidelity change feed must fall back to the standard gateway" + ); + assert_eq!(full_fidelity_routing.selected_url, *endpoint.url()); + + let incremental_routing = super::resolve_endpoint( + &incremental, + &retry_state, + &location, + true, + true, + Duration::from_secs(60), + ); + assert_eq!( + incremental_routing.transport_mode, + TransportMode::GatewayV2, + "incremental change feed remains Gateway 2.0 eligible" + ); + } + #[test] fn resolve_endpoint_falls_back_to_gateway_when_account_name_unparseable() { let operation = CosmosOperation::read_item(ItemReference::from_name( diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/driver/transport/gateway_v2_eligibility.rs b/sdk/cosmos/azure_data_cosmos_driver/src/driver/transport/gateway_v2_eligibility.rs index 9b4a95073fa..e797255e723 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/driver/transport/gateway_v2_eligibility.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/driver/transport/gateway_v2_eligibility.rs @@ -21,7 +21,12 @@ use crate::models::{OperationType, ResourceType}; pub(crate) fn is_operation_supported_by_gateway_v2( resource_type: ResourceType, operation_type: OperationType, + is_full_fidelity_change_feed: bool, ) -> bool { + if is_full_fidelity_change_feed { + // Excluded by the Gateway 2.0 contract; see docs/GATEWAY_V2_SPEC.md. + return false; + } // Both arms of this match are intentionally exhaustive (no wildcard `_` arm) so // that adding a new variant to either enum is a compile-time error, forcing an // explicit eligibility decision rather than a silent fail-closed default. @@ -139,11 +144,28 @@ mod tests { for resource_type in all_resource_types() { for operation_type in all_operation_types() { assert_eq!( - is_operation_supported_by_gateway_v2(resource_type, operation_type), + is_operation_supported_by_gateway_v2(resource_type, operation_type, false), expected_gateway_v2_eligibility(resource_type, operation_type), "unexpected Gateway 2.0 eligibility for {resource_type:?} {operation_type:?}" ); } } } + + #[test] + fn full_fidelity_change_feed_read_feed_is_ineligible() { + // A `Document`/`ReadFeed` is otherwise eligible, but a full-fidelity + // (AllVersionsAndDeletes) change feed must route through the standard + // gateway because Gateway 2.0 does not forward the `A-IM` header. + assert!(is_operation_supported_by_gateway_v2( + ResourceType::Document, + OperationType::ReadFeed, + false, + )); + assert!(!is_operation_supported_by_gateway_v2( + ResourceType::Document, + OperationType::ReadFeed, + true, + )); + } } diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/in_memory_emulator/dispatch.rs b/sdk/cosmos/azure_data_cosmos_driver/src/in_memory_emulator/dispatch.rs index 6e98df40b1a..c203e67ee1c 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/in_memory_emulator/dispatch.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/in_memory_emulator/dispatch.rs @@ -61,6 +61,11 @@ pub(crate) struct ParsedRequest { pub partition_key_header: Option, pub if_match: Option, pub if_none_match: Option, + /// Value of the `If-Modified-Since` request header, when present. The change + /// feed maps a `PointInTime` start position to this header, so the read-feed + /// handler uses its presence to detect (and, for AllVersionsAndDeletes, + /// reject) point-in-time starts. + pub if_modified_since: Option, pub session_token: Option, pub activity_id: Option, pub content_response_on_write: bool, @@ -87,6 +92,13 @@ pub(crate) struct ParsedRequest { pub is_batch: bool, #[allow(dead_code)] pub is_upsert: bool, // used during dispatch resolution + /// Value of the change-feed `A-IM` request header, when present. + /// + /// `"Incremental Feed"` selects the LatestVersion change feed and + /// `"Full-Fidelity Feed"` selects AllVersionsAndDeletes. The emulator's + /// read-feed handler consults this to decide whether to return flat + /// documents or full-fidelity envelopes. + pub a_im: Option, } // Header name constants for request parsing @@ -94,6 +106,7 @@ static IS_UPSERT: HeaderName = HeaderName::from_static("x-ms-documentdb-is-upser static PARTITION_KEY: HeaderName = HeaderName::from_static("x-ms-documentdb-partitionkey"); static IF_MATCH: HeaderName = HeaderName::from_static("if-match"); static IF_NONE_MATCH: HeaderName = HeaderName::from_static("if-none-match"); +static IF_MODIFIED_SINCE: HeaderName = HeaderName::from_static("if-modified-since"); static SESSION_TOKEN: HeaderName = HeaderName::from_static("x-ms-session-token"); static ACTIVITY_ID: HeaderName = HeaderName::from_static("x-ms-activity-id"); static CONTENT_RESPONSE: HeaderName = @@ -114,6 +127,7 @@ static IS_BATCH_REQUEST: HeaderName = HeaderName::from_static("x-ms-cosmos-is-ba static OFFER_THROUGHPUT: HeaderName = HeaderName::from_static("x-ms-offer-throughput"); static OFFER_AUTOPILOT_SETTINGS: HeaderName = HeaderName::from_static("x-ms-cosmos-offer-autopilot-settings"); +static A_IM: HeaderName = HeaderName::from_static("a-im"); /// Parses an HTTP request into a `ParsedRequest`. pub(crate) fn parse_request(request: &Request) -> ParsedRequest { @@ -128,6 +142,9 @@ pub(crate) fn parse_request(request: &Request) -> ParsedRequest { let if_none_match = headers .get_optional_str(&IF_NONE_MATCH) .map(|s| s.to_string()); + let if_modified_since = headers + .get_optional_str(&IF_MODIFIED_SINCE) + .map(|s| s.to_string()); let session_token = headers .get_optional_str(&SESSION_TOKEN) .map(|s| s.to_string()); @@ -178,6 +195,7 @@ pub(crate) fn parse_request(request: &Request) -> ParsedRequest { let offer_autopilot_settings = headers .get_optional_str(&OFFER_AUTOPILOT_SETTINGS) .map(|s| s.to_string()); + let a_im = headers.get_optional_str(&A_IM).map(|s| s.to_string()); let path = url.path(); // Reject trailing slashes after the leading `/`. `/dbs/mydb/colls/mycoll/docs/` @@ -242,6 +260,7 @@ pub(crate) fn parse_request(request: &Request) -> ParsedRequest { partition_key_header, if_match, if_none_match, + if_modified_since, session_token, activity_id, content_response_on_write, @@ -255,6 +274,7 @@ pub(crate) fn parse_request(request: &Request) -> ParsedRequest { is_query_plan, is_batch, is_upsert, + a_im, } } diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/in_memory_emulator/operations.rs b/sdk/cosmos/azure_data_cosmos_driver/src/in_memory_emulator/operations.rs index 63987b1f76a..b4dd7527917 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/in_memory_emulator/operations.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/in_memory_emulator/operations.rs @@ -645,6 +645,7 @@ pub(crate) async fn handle_operation( partition_key_header: Some(operation.partition_key.to_string()), if_match: operation.if_match.clone(), if_none_match: operation.if_none_match.clone(), + if_modified_since: None, session_token: operation.session_token.clone(), activity_id: None, content_response_on_write: true, @@ -658,6 +659,7 @@ pub(crate) async fn handle_operation( is_query_plan: false, is_batch: false, is_upsert: matches!(operation_type, OperationType::Upsert), + a_im: None, }; match operation_type { @@ -1449,6 +1451,7 @@ pub(crate) async fn handle_operation( partition_key_header: Some(operation.partition_key.to_string()), if_match: operation.if_match.clone(), if_none_match: operation.if_none_match.clone(), + if_modified_since: None, session_token: operation.session_token.clone(), activity_id: None, content_response_on_write: true, @@ -1462,6 +1465,7 @@ pub(crate) async fn handle_operation( is_query_plan: false, is_batch: false, is_upsert: false, + a_im: None, } } @@ -3114,8 +3118,34 @@ fn handle_read_feed_items( parsed: &ParsedRequest, start: Instant, ) -> AsyncRawResponse { + // The service only supports the AllVersionsAndDeletes (full-fidelity) change + // feed starting from `Now` or resuming from a continuation. A `Beginning` + // start (no `If-None-Match`) or a `PointInTime` start (`If-Modified-Since`) + // is rejected with 400. The SDK relies on the service to gate these, so + // mirror that here. + if is_full_fidelity_feed(parsed.a_im.as_deref()) { + if let Some(response) = reject_unsupported_full_fidelity_start(parsed, start) { + return response; + } + } match collect_item_documents(store, region_name, parsed, start) { Ok((rid, docs, token, mut headers)) => { + // Full-fidelity (AllVersionsAndDeletes) change feed reads carry + // `A-IM: Full-Fidelity Feed`. The in-memory store only retains the + // latest state of each document (no change log), so it cannot replay + // historical versions, deletes, or pre-images. It therefore + // synthesizes a minimal `create` envelope per current document so the + // SDK's full-fidelity code path (header emission, mode dispatch, and + // the iterator's raw `ChangeFeedItem` deserialization) can be + // exercised end-to-end. Deletes / `previous` images remain covered by + // unit tests and are a documented follow-up. Incremental + // (`A-IM: Incremental Feed`) and plain read-feed requests are + // unchanged and return flat documents. + let docs = if is_full_fidelity_feed(parsed.a_im.as_deref()) { + docs.into_iter().map(full_fidelity_envelope).collect() + } else { + docs + }; headers.session_token = token; success_document_feed_response( "Documents", @@ -3130,6 +3160,70 @@ fn handle_read_feed_items( } } +/// Returns `true` when the `A-IM` header selects the full-fidelity +/// (AllVersionsAndDeletes) change feed. +fn is_full_fidelity_feed(a_im: Option<&str>) -> bool { + a_im.is_some_and(|value| value.eq_ignore_ascii_case("Full-Fidelity Feed")) +} + +/// Rejects AllVersionsAndDeletes change-feed reads that start from an +/// unsupported position. +/// +/// Returns `Some(400)` for a `Beginning` start (no `If-None-Match`) or a +/// `PointInTime` start (`If-Modified-Since` present), and `None` for `Now` +/// (`If-None-Match: *`) or a resume (`If-None-Match: `). +fn reject_unsupported_full_fidelity_start( + parsed: &ParsedRequest, + start: Instant, +) -> Option { + let reason = if parsed.if_modified_since.is_some() { + "a point-in-time start" + } else if parsed.if_none_match.is_none() { + "a start from the beginning" + } else { + return None; + }; + + Some( + error_response( + StatusCode::BadRequest, + None, + "BadRequest", + &format!( + "The AllVersionsAndDeletes change feed mode does not support {reason}; \ + start from Now or resume from a continuation token." + ), + 0.0, + "", + start, + ) + .build(), + ) +} + +/// Wraps a current document body in a minimal full-fidelity change envelope. +/// +/// The emulator has no change log, so every retained document is surfaced as a +/// `create`. `crts` is taken from the document's `_ts` when available; `lsn` and +/// `previous` are omitted because the store does not track them. +fn full_fidelity_envelope(doc: DocumentFeedItem) -> DocumentFeedItem { + let crts = doc + .body + .get("_ts") + .cloned() + .unwrap_or(serde_json::Value::Null); + DocumentFeedItem { + body: serde_json::json!({ + "current": doc.body, + "metadata": { + "operationType": "create", + "crts": crts, + }, + }), + cursor: doc.cursor, + } +} + fn handle_query_items( store: &Arc, region_name: &str, diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/continuation_token.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/continuation_token.rs index 7c0386de6c5..1e787e12698 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/continuation_token.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/continuation_token.rs @@ -64,10 +64,21 @@ impl ContinuationToken { let container = operation.container().ok_or_else(|| { crate::error::CosmosError::builder().with_status(crate::error::CosmosStatus::new(azure_core::http::StatusCode::BadRequest)).with_message("client-side continuation tokens require a query or change feed operation targeting a container").build() })?; + // Change feed continuations are mode-specific: resuming a feed in a + // different mode (LatestVersion vs AllVersionsAndDeletes) is not + // supported, so record the mode and validate it on resume. Only the + // full-fidelity marker is stored; LatestVersion (and query) tokens carry + // no marker and are treated as LatestVersion, keeping their payload + // unchanged and backward-compatible. + let change_feed_full_fidelity = operation + .is_change_feed() + .then(|| operation.request_headers().full_fidelity_feed) + .filter(|&full_fidelity| full_fidelity); let state = TokenState { operation: token_operation, rid: container.rid().to_string(), root: root_state.clone(), + change_feed_full_fidelity, }; let json = serde_json::to_vec(&state).map_err(|e| { @@ -172,6 +183,14 @@ pub struct TokenState { /// The root node's state at the point of snapshotting. root: PipelineNodeState, + + /// For change feed tokens, `true` when the feed was issued in the + /// AllVersionsAndDeletes (full-fidelity) mode and `false` for LatestVersion. + /// `None` for query tokens (and for legacy change feed tokens that predate + /// mode encoding, which are treated as LatestVersion). Validated on resume: + /// a change feed cannot switch modes across continuations. + #[serde(rename = "cfm", default, skip_serializing_if = "Option::is_none")] + change_feed_full_fidelity: Option, } impl TokenState { @@ -190,6 +209,32 @@ impl TokenState { )) .build()); } + if expected == TokenOperation::ChangeFeed { + // Legacy tokens without an encoded mode are treated as LatestVersion. + let token_full_fidelity = self.change_feed_full_fidelity.unwrap_or(false); + let operation_full_fidelity = operation.request_headers().full_fidelity_feed; + if token_full_fidelity != operation_full_fidelity { + let mode_name = |full_fidelity| { + if full_fidelity { + "AllVersionsAndDeletes" + } else { + "LatestVersion" + } + }; + return Err(crate::error::CosmosError::builder() + .with_status(crate::error::CosmosStatus::new( + azure_core::http::StatusCode::BadRequest, + )) + .with_message(format!( + "change feed continuation token was issued for {token_mode} mode but this \ + request uses {operation_mode} mode; a change feed cannot switch modes \ + across continuations. Start a new change feed to change mode.", + token_mode = mode_name(token_full_fidelity), + operation_mode = mode_name(operation_full_fidelity), + )) + .build()); + } + } let container = operation.container().ok_or_else(|| { crate::error::CosmosError::builder().with_status(crate::error::CosmosStatus::new(azure_core::http::StatusCode::BadRequest)).with_message("client-side continuation tokens require a query operation targeting a container").build() })?; @@ -301,6 +346,16 @@ mod tests { ) } + /// Builds a single-partition AllVersionsAndDeletes (full-fidelity) change + /// feed operation against `test_container()` (rid `coll_rid`). + fn change_feed_avad_op() -> CosmosOperation { + let def: PartitionKeyDefinition = serde_json::from_str(r#"{"paths":["/pk"]}"#).unwrap(); + CosmosOperation::change_feed_all_versions_and_deletes( + test_container(), + Some(FeedRange::for_partition(PartitionKey::from("pk1"), &def)), + ) + } + /// Decodes the base64url-no-pad payload of a `c1.`-prefixed token into /// its raw JSON bytes for inspection. fn decode_v1_payload(token: &ContinuationToken) -> String { @@ -458,6 +513,7 @@ mod tests { operation: TokenOperation::Query, rid: "coll_rid".to_string(), root: PipelineNodeState::Drained, + change_feed_full_fidelity: None, }; let err = state.is_valid_for_operation(&change_feed_op()).unwrap_err(); assert!(err.to_string().contains("ChangeFeed")); @@ -469,11 +525,66 @@ mod tests { operation: TokenOperation::ChangeFeed, rid: "coll_rid".to_string(), root: PipelineNodeState::Drained, + change_feed_full_fidelity: None, }; let err = state.is_valid_for_operation(&query_op()).unwrap_err(); assert!(err.to_string().contains("Query")); } + #[test] + fn is_valid_for_operation_rejects_change_feed_mode_switch() { + // A LatestVersion token (no mode marker) cannot resume an + // AllVersionsAndDeletes feed, and vice versa. + let latest = TokenState { + operation: TokenOperation::ChangeFeed, + rid: "coll_rid".to_string(), + root: PipelineNodeState::Drained, + change_feed_full_fidelity: None, + }; + let err = latest + .is_valid_for_operation(&change_feed_avad_op()) + .unwrap_err(); + assert!(err.to_string().contains("AllVersionsAndDeletes")); + assert!(err.to_string().contains("LatestVersion")); + + let avad = TokenState { + operation: TokenOperation::ChangeFeed, + rid: "coll_rid".to_string(), + root: PipelineNodeState::Drained, + change_feed_full_fidelity: Some(true), + }; + let err = avad.is_valid_for_operation(&change_feed_op()).unwrap_err(); + assert!(err.to_string().contains("AllVersionsAndDeletes")); + assert!(err.to_string().contains("LatestVersion")); + } + + #[test] + fn is_valid_for_operation_accepts_matching_change_feed_mode() { + let avad = TokenState { + operation: TokenOperation::ChangeFeed, + rid: "coll_rid".to_string(), + root: PipelineNodeState::Drained, + change_feed_full_fidelity: Some(true), + }; + avad.is_valid_for_operation(&change_feed_avad_op()) + .expect("AllVersionsAndDeletes token resumes an AllVersionsAndDeletes operation"); + } + + #[test] + fn encode_v1_records_all_versions_and_deletes_mode() { + let token = + ContinuationToken::encode_v1(&change_feed_avad_op(), &PipelineNodeState::Drained) + .unwrap(); + assert!(decode_v1_payload(&token).contains(r#""cfm":true"#)); + } + + #[test] + fn encode_v1_omits_mode_marker_for_latest_version() { + let token = + ContinuationToken::encode_v1(&change_feed_op(), &PipelineNodeState::Drained).unwrap(); + assert!(!decode_v1_payload(&token).contains("cfm")); + } + // ── Deserialization ───────────────────────────────────────────────── #[test] @@ -584,6 +695,7 @@ mod tests { operation: TokenOperation::Query, rid: "coll_rid".to_string(), root: PipelineNodeState::Drained, + change_feed_full_fidelity: None, }; state.is_valid_for_operation(&query_op()).unwrap(); } @@ -594,6 +706,7 @@ mod tests { operation: TokenOperation::Query, rid: "different_rid".to_string(), root: PipelineNodeState::Drained, + change_feed_full_fidelity: None, }; let err = state.is_valid_for_operation(&query_op()).unwrap_err(); assert!(err.to_string().contains("different_rid")); @@ -606,6 +719,7 @@ mod tests { operation: TokenOperation::Query, rid: "coll_rid".to_string(), root: PipelineNodeState::Drained, + change_feed_full_fidelity: None, }; let item = ItemReference::from_name(&test_container(), PartitionKey::from("pk1"), "doc1"); let read = CosmosOperation::read_item(item); diff --git a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_headers.rs b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_headers.rs index 34e5235e18a..70a04046d48 100644 --- a/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_headers.rs +++ b/sdk/cosmos/azure_data_cosmos_driver/src/models/cosmos_headers.rs @@ -49,6 +49,13 @@ pub(crate) mod request_header_names { /// Change-feed indicator ("Incremental Feed"). HTTP standard name `a-im`. pub const A_IM: &str = "a-im"; pub const INCREMENTAL_FEED: &str = "Incremental Feed"; + /// Full-fidelity change-feed indicator ("Full-Fidelity Feed"). + /// + /// Selects the AllVersionsAndDeletes change feed, which returns every + /// intermediate version and delete as an envelope with `current`, + /// `previous`, and `metadata`. Sent as the `a-im` value in place of + /// [`INCREMENTAL_FEED`]. + pub const FULL_FIDELITY_FEED: &str = "Full-Fidelity Feed"; /// Wire format version for change feed responses. pub const CHANGEFEED_WIRE_FORMAT_VERSION: &str = "x-ms-cosmos-changefeed-wire-format-version"; /// The wire format version value used by this SDK. @@ -236,8 +243,22 @@ pub struct CosmosRequestHeaders { /// When `true`, the driver emits the standard change-feed indicator /// header. Combine with [`Precondition::if_none_match`] to pass a /// continuation token. + /// + /// Mutually exclusive with [`full_fidelity_feed`](Self::full_fidelity_feed); + /// if both are set, full-fidelity takes precedence when emitting `a-im`. pub incremental_feed: bool, + /// Requests a full-fidelity change feed read (`a-im: Full-Fidelity Feed`). + /// + /// When `true`, the driver emits the full-fidelity change-feed indicator, + /// selecting the AllVersionsAndDeletes mode where every intermediate + /// version and delete is returned inside an envelope + /// (`{ current, previous, metadata }`). + /// + /// Mutually exclusive with [`incremental_feed`](Self::incremental_feed); + /// full-fidelity takes precedence when emitting `a-im`. + pub full_fidelity_feed: bool, + /// When `true`, emits the change-feed wire format version header /// (`x-ms-cosmos-changefeed-wire-format-version: 2021-09-15`). /// @@ -339,7 +360,16 @@ impl CosmosRequestHeaders { HeaderValue::from(wire), ); } - if self.incremental_feed { + // `a-im` selects the change feed mode. Full-fidelity + // (AllVersionsAndDeletes) takes precedence over incremental + // (LatestVersion); the two are mutually exclusive, but guard the + // invariant and emit exactly one value. + if self.full_fidelity_feed { + headers.insert( + request_header_names::A_IM, + HeaderValue::from_static(request_header_names::FULL_FIDELITY_FEED), + ); + } else if self.incremental_feed { headers.insert( request_header_names::A_IM, HeaderValue::from_static(request_header_names::INCREMENTAL_FEED), @@ -1357,6 +1387,45 @@ mod tests { ); } + #[test] + fn write_to_headers_emits_incremental_a_im() { + let cosmos_headers = CosmosRequestHeaders { + incremental_feed: true, + ..Default::default() + }; + let mut headers = Headers::new(); + cosmos_headers.write_to_headers(&mut headers); + assert_eq!( + headers.get_optional_str(&HeaderName::from_static("a-im")), + Some("Incremental Feed") + ); + } + + #[test] + fn write_to_headers_emits_full_fidelity_a_im() { + let cosmos_headers = CosmosRequestHeaders { + full_fidelity_feed: true, + ..Default::default() + }; + let mut headers = Headers::new(); + cosmos_headers.write_to_headers(&mut headers); + assert_eq!( + headers.get_optional_str(&HeaderName::from_static("a-im")), + Some("Full-Fidelity Feed") + ); + } + + #[test] + fn write_to_headers_omits_a_im_when_no_feed_mode() { + let cosmos_headers = CosmosRequestHeaders::default(); + let mut headers = Headers::new(); + cosmos_headers.write_to_headers(&mut headers); + assert_eq!( + headers.get_optional_str(&HeaderName::from_static("a-im")), + None + ); + } + /// Round-trips a fully-populated [`CosmosResponseHeaders`] through /// [`to_raw_headers`](CosmosResponseHeaders::to_raw_headers) followed /// by [`from_headers`](CosmosResponseHeaders::from_headers) and 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 c3905213be9..161f5aee88b 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 @@ -13,6 +13,19 @@ use serde::{Deserialize, Serialize}; use std::borrow::Cow; use time::OffsetDateTime; +/// Which change feed mode a factory should configure. +/// +/// Private helper shared by [`CosmosOperation::change_feed`] and +/// [`CosmosOperation::change_feed_all_versions_and_deletes`]; the only +/// difference between the two is the `A-IM` header value they emit. +#[derive(Clone, Copy)] +enum ChangeFeedFactoryMode { + /// LatestVersion: `A-IM: Incremental Feed`. + Incremental, + /// AllVersionsAndDeletes: `A-IM: Full-Fidelity Feed`. + FullFidelity, +} + /// The position a change feed is started from. /// /// Passed explicitly when starting a change feed read and persisted inside the @@ -39,9 +52,12 @@ pub enum ChangeFeedStartFrom { /// "Now" is evaluated when the request is sent, so on resume a never-polled /// partition starts from resume time rather than the original start time. /// This is acceptable for LatestVersion (it still converges to the latest - /// state under at-least-once delivery), but AllVersionsAndDeletes must - /// resolve "Now" to a concrete start position before persisting so resume - /// does not drop intermediate versions/deletes. + /// state under at-least-once delivery). For AllVersionsAndDeletes it is a + /// documented limitation: a range that is never polled before a checkpoint + /// can drop the intermediate versions and deletes that occurred between the + /// original start and the resume. "Now" is deliberately not pinned to a + /// concrete start position before persisting, because that would change its + /// semantics; lossless per-range "Now" resolution is a future improvement. Now, /// Start from a specific point in time (wire header `If-Modified-Since`). @@ -757,11 +773,51 @@ impl CosmosOperation { /// `target` scopes the change feed to a specific partition or EPK range. /// Pass `None` or `Some(FeedRange::full())` to read the entire container. pub fn change_feed(container: ContainerReference, target: Option) -> Self { + Self::change_feed_with_mode(container, target, ChangeFeedFactoryMode::Incremental) + } + + /// Creates a full-fidelity (AllVersionsAndDeletes) change feed read + /// operation for a container. + /// + /// Identical to [`change_feed`](Self::change_feed) except it sets the + /// `A-IM` header to `Full-Fidelity Feed` instead of `Incremental Feed`. + /// This selects the AllVersionsAndDeletes mode, in which every intermediate + /// version and delete is returned inside an envelope carrying `current` + /// (post-image), `previous` (pre-image, when enabled), and `metadata` + /// (operation type, LSN, timestamps). The SDK does **not** unwrap + /// `current`; the caller deserializes each item into a `ChangeFeedItem`. + /// + /// Like [`change_feed`](Self::change_feed) this also sets the + /// `x-ms-cosmos-changefeed-wire-format-version` header and marks the + /// operation as a change feed read. The start position is set via + /// [`with_change_feed_start`](Self::with_change_feed_start). + /// + /// `target` scopes the change feed to a specific partition or EPK range. + /// Pass `None` or `Some(FeedRange::full())` to read the entire container. + pub fn change_feed_all_versions_and_deletes( + container: ContainerReference, + target: Option, + ) -> Self { + Self::change_feed_with_mode(container, target, ChangeFeedFactoryMode::FullFidelity) + } + + /// Shared constructor for the change feed factories. The only difference + /// between LatestVersion and AllVersionsAndDeletes is which `A-IM` value is + /// emitted; everything else (resource shape, wire-format-version header, + /// change-feed marking) is identical. + fn change_feed_with_mode( + container: ContainerReference, + target: Option, + mode: ChangeFeedFactoryMode, + ) -> Self { let resource_ref: CosmosResourceReference = CosmosResourceReference::from(container) .with_resource_type(ResourceType::Document) .into_feed_reference(); let mut headers = CosmosRequestHeaders::new(); - headers.incremental_feed = true; + match mode { + ChangeFeedFactoryMode::Incremental => headers.incremental_feed = true, + ChangeFeedFactoryMode::FullFidelity => headers.full_fidelity_feed = true, + } headers.changefeed_wire_format_version = true; let mut operation = Self::new(OperationType::ReadFeed, resource_ref, target).with_request_headers(headers); @@ -1055,6 +1111,23 @@ mod tests { assert!(op.is_change_feed()); assert!(op.request_headers().incremental_feed); + assert!(!op.request_headers().full_fidelity_feed); + assert!(op.request_headers().changefeed_wire_format_version); + } + + /// The full-fidelity (AllVersionsAndDeletes) factory sets the + /// full-fidelity indicator instead of the incremental one, while keeping + /// the wire-format-version header and change-feed marking. + #[test] + fn change_feed_all_versions_and_deletes_sets_full_fidelity_header() { + let op = CosmosOperation::change_feed_all_versions_and_deletes( + test_container(), + Some(FeedRange::full()), + ); + + assert!(op.is_change_feed()); + assert!(op.request_headers().full_fidelity_feed); + assert!(!op.request_headers().incremental_feed); assert!(op.request_headers().changefeed_wire_format_version); }