fix(eventhubs): fold checkpoint blob key to lowercase ASCII

sdk/eventhubs/azure_messaging_eventhubs/CHANGELOG.md

@@ -17,6 +17,10 @@
 
 - The `default` feature now selects `fe2o3_amqp_rustls`, so AMQP framed directly on TCP (`amqps://`, port 5671) runs on rustls with the aws-lc-rs provider where it ran on native-tls. Both stacks read the trust store of the operating system, so a namespace behind a private or an enterprise certificate authority keeps working. The stacks read that store through different platform APIs, and a deployment that tunes native-tls directly, such as one that sets OpenSSL environment variables, can still see a difference. To keep native-tls, turn off the default features, name `fe2o3_amqp`, and take a direct dependency on `fe2o3-amqp` with its `native-tls` feature. ([#4189](https://github.com/Azure/azure-sdk-for-rust/issues/4189))
 - On the receive path, the `amqp:link:stolen` AMQP condition is no longer auto-retried. A receiver displaced by a higher-or-equal-epoch attacher now surfaces the error (translated to `EventHubsError::ConsumerDisconnected` by `EventReceiver::stream_events`) instead of silently re-attaching. Sender, CBS, and management operations retain the historical retry-on-stolen behavior.
+- Event Hubs treats the consumer group as case insensitive, so one deployment that spelled the group `$Default` on one run and `$default` on the next built two disjoint key sets and reprocessed events. ([#5099](https://github.com/Azure/azure-sdk-for-rust/issues/5099))
+- The checkpoint and ownership blob key that `Checkpoint::get_checkpoint_blob_prefix_name`, `Checkpoint::get_checkpoint_blob_name`, `Ownership::get_ownership_prefix_name`, and `Ownership::get_ownership_name` build now folds the fully qualified namespace, the event hub name, and the consumer group to lowercase ASCII. The partition id keeps its case. The .NET, JavaScript, and Python clients fold these three names; the Go and Java clients do not, so this change moves the Rust crate from the second group into the first.
+- The old key `NS.ServiceBus.Windows.Net/My-Hub/$Default/checkpoint/0` becomes `ns.servicebus.windows.net/my-hub/$default/checkpoint/0`. The ownership key changes in the same way.
+- Migration: records that an older Rust client wrote stay at the old key and become unreachable. The change adds no dual read and no fallback lookup. A processor that starts against an existing container resumes from its configured start position.
 
 ### Bugs Fixed
 

sdk/eventhubs/azure_messaging_eventhubs/src/event_processor/models.rs

@@ -4,7 +4,8 @@
 // Note that this module returns azure_core errors, *not* eventhub errors. That is because these structures are used by checkpoint stores which always return azure_core errors.
 use crate::StartPosition;
 use azure_core::{
-    error::ErrorKind as AzureErrorKind, http::Etag, time::OffsetDateTime, Error, Result,
+    error::ErrorKind as AzureErrorKind, fmt::to_ascii_lowercase, http::Etag, time::OffsetDateTime,
+    Error, Result,
 };
 use std::collections::HashMap;
 
@@ -42,6 +43,13 @@ macro_rules! check_non_empty_parameter(
 
 impl Checkpoint {
     /// Returns the prefix for the checkpoint blob name.
+    ///
+    /// The layout is `{namespace}/{event hub}/{consumer group}/checkpoint/`.
+    ///
+    /// The namespace, the event hub name, and the consumer group fold to
+    /// lowercase. The fold applies to ASCII letters only, and it leaves all
+    /// other characters unchanged. Event Hubs treats these three names as
+    /// case insensitive, so the fold keeps one prefix for one Event Hub.
     pub fn get_checkpoint_blob_prefix_name(
         fully_qualified_namespace: &str,
         event_hub_name: &str,
@@ -50,15 +58,22 @@ impl Checkpoint {
         check_non_empty_parameter!(fully_qualified_namespace);
         check_non_empty_parameter!(event_hub_name);
         check_non_empty_parameter!(consumer_group);
-        Ok(fully_qualified_namespace.to_string()
-            + "/"
-            + event_hub_name
-            + "/"
-            + consumer_group
-            + "/checkpoint/")
+        Ok(format!(
+            "{}/{}/{}/checkpoint/",
+            to_ascii_lowercase(fully_qualified_namespace),
+            to_ascii_lowercase(event_hub_name),
+            to_ascii_lowercase(consumer_group)
+        ))
     }
 
     /// Returns the full name of the checkpoint blob.
+    ///
+    /// The layout is
+    /// `{namespace}/{event hub}/{consumer group}/checkpoint/{partition id}`.
+    ///
+    /// The namespace, the event hub name, and the consumer group fold to
+    /// lowercase. The fold applies to ASCII letters only, and it leaves all
+    /// other characters unchanged. The partition id keeps its case.
     pub fn get_checkpoint_blob_name(
         fully_qualified_namespace: &str,
         event_hub_name: &str,
@@ -100,6 +115,13 @@ pub struct Ownership {
 
 impl Ownership {
     /// Returns the prefix for the ownership blob name.
+    ///
+    /// The layout is `{namespace}/{event hub}/{consumer group}/ownership/`.
+    ///
+    /// The namespace, the event hub name, and the consumer group fold to
+    /// lowercase. The fold applies to ASCII letters only, and it leaves all
+    /// other characters unchanged. Event Hubs treats these three names as
+    /// case insensitive, so the fold keeps one prefix for one Event Hub.
     pub fn get_ownership_prefix_name(
         fully_qualified_namespace: &str,
         event_hub_name: &str,
@@ -108,15 +130,22 @@ impl Ownership {
         check_non_empty_parameter!(fully_qualified_namespace);
         check_non_empty_parameter!(event_hub_name);
         check_non_empty_parameter!(consumer_group);
-        Ok(fully_qualified_namespace.to_string()
-            + "/"
-            + event_hub_name
-            + "/"
-            + consumer_group
-            + "/ownership/")
+        Ok(format!(
+            "{}/{}/{}/ownership/",
+            to_ascii_lowercase(fully_qualified_namespace),
+            to_ascii_lowercase(event_hub_name),
+            to_ascii_lowercase(consumer_group)
+        ))
     }
 
     /// Returns the full name of the ownership blob.
+    ///
+    /// The layout is
+    /// `{namespace}/{event hub}/{consumer group}/ownership/{partition id}`.
+    ///
+    /// The namespace, the event hub name, and the consumer group fold to
+    /// lowercase. The fold applies to ASCII letters only, and it leaves all
+    /// other characters unchanged. The partition id keeps its case.
     pub fn get_ownership_name(
         fully_qualified_namespace: &str,
         event_hub_name: &str,
@@ -155,3 +184,198 @@ pub struct StartPositions {
     /// or the latest event.
     pub default: StartPosition,
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use azure_core::error::ErrorKind;
+
+    const NS_MIXED: &str = "NS-Test.ServiceBus.Windows.Net";
+    const NS_LOWER: &str = "ns-test.servicebus.windows.net";
+    const HUB_MIXED: &str = "My-EventHub";
+    const HUB_LOWER: &str = "my-eventhub";
+    const GROUP_MIXED: &str = "$Default";
+    const GROUP_LOWER: &str = "$default";
+    const PARTITION: &str = "Partition-A";
+
+    /// Each of the three key fields folds to lowercase on its own, so a
+    /// difference in one field cannot move the blob key.
+    #[test]
+    fn key_fields_fold_to_lowercase_independently() {
+        let rows = [
+            ("namespace only", NS_MIXED, HUB_LOWER, GROUP_LOWER),
+            ("event hub name only", NS_LOWER, HUB_MIXED, GROUP_LOWER),
+            ("consumer group only", NS_LOWER, HUB_LOWER, GROUP_MIXED),
+            ("all three", NS_MIXED, HUB_MIXED, GROUP_MIXED),
+        ];
+        let expected_checkpoint = format!("{NS_LOWER}/{HUB_LOWER}/{GROUP_LOWER}/checkpoint/");
+        let expected_ownership = format!("{NS_LOWER}/{HUB_LOWER}/{GROUP_LOWER}/ownership/");
+
+        // One run reports every row, so a partial fold shows all of its
+        // damage at once.
+        let mut mismatches = Vec::new();
+        for (label, namespace, event_hub_name, consumer_group) in rows {
+            let checkpoint = Checkpoint::get_checkpoint_blob_prefix_name(
+                namespace,
+                event_hub_name,
+                consumer_group,
+            )
+            .unwrap();
+            if checkpoint != expected_checkpoint {
+                mismatches.push(format!(
+                    "{label}: checkpoint prefix is {checkpoint:?}, expected {expected_checkpoint:?}"
+                ));
+            }
+
+            let ownership =
+                Ownership::get_ownership_prefix_name(namespace, event_hub_name, consumer_group)
+                    .unwrap();
+            if ownership != expected_ownership {
+                mismatches.push(format!(
+                    "{label}: ownership prefix is {ownership:?}, expected {expected_ownership:?}"
+                ));
+            }
+        }
+
+        assert!(
+            mismatches.is_empty(),
+            "the key fields did not fold to lowercase:\n{}",
+            mismatches.join("\n")
+        );
+    }
+
+    #[test]
+    fn checkpoint_blob_name_folds_key_and_keeps_partition_id_case() {
+        let name =
+            Checkpoint::get_checkpoint_blob_name(NS_MIXED, HUB_MIXED, GROUP_MIXED, PARTITION);
+        assert_eq!(
+            name.unwrap(),
+            "ns-test.servicebus.windows.net/my-eventhub/$default/checkpoint/Partition-A"
+        );
+    }
+
+    #[test]
+    fn ownership_name_folds_key_and_keeps_partition_id_case() {
+        let name = Ownership::get_ownership_name(NS_MIXED, HUB_MIXED, GROUP_MIXED, PARTITION);
+        assert_eq!(
+            name.unwrap(),
+            "ns-test.servicebus.windows.net/my-eventhub/$default/ownership/Partition-A"
+        );
+    }
+
+    /// Two callers that spell the same Event Hub with a different case must
+    /// land on one key set. This test does not pin the fold direction, which
+    /// is the job of the tests above.
+    #[test]
+    fn key_is_stable_across_input_case() {
+        assert_eq!(
+            Checkpoint::get_checkpoint_blob_prefix_name(NS_LOWER, HUB_LOWER, GROUP_MIXED).unwrap(),
+            Checkpoint::get_checkpoint_blob_prefix_name(NS_LOWER, HUB_LOWER, GROUP_LOWER).unwrap(),
+            "the consumer group case moved the checkpoint prefix"
+        );
+        assert_eq!(
+            Ownership::get_ownership_prefix_name(NS_LOWER, HUB_LOWER, GROUP_MIXED).unwrap(),
+            Ownership::get_ownership_prefix_name(NS_LOWER, HUB_LOWER, GROUP_LOWER).unwrap(),
+            "the consumer group case moved the ownership prefix"
+        );
+        assert_eq!(
+            Checkpoint::get_checkpoint_blob_name(NS_LOWER, HUB_LOWER, GROUP_MIXED, PARTITION)
+                .unwrap(),
+            Checkpoint::get_checkpoint_blob_name(NS_LOWER, HUB_LOWER, GROUP_LOWER, PARTITION)
+                .unwrap(),
+            "the consumer group case moved the checkpoint blob name"
+        );
+        assert_eq!(
+            Ownership::get_ownership_name(NS_LOWER, HUB_LOWER, GROUP_MIXED, PARTITION).unwrap(),
+            Ownership::get_ownership_name(NS_LOWER, HUB_LOWER, GROUP_LOWER, PARTITION).unwrap(),
+            "the consumer group case moved the ownership name"
+        );
+
+        assert_eq!(
+            Checkpoint::get_checkpoint_blob_name(NS_MIXED, HUB_LOWER, GROUP_LOWER, PARTITION)
+                .unwrap(),
+            Checkpoint::get_checkpoint_blob_name(NS_LOWER, HUB_LOWER, GROUP_LOWER, PARTITION)
+                .unwrap(),
+            "the namespace case moved the checkpoint blob name"
+        );
+        assert_eq!(
+            Checkpoint::get_checkpoint_blob_name(NS_LOWER, HUB_MIXED, GROUP_LOWER, PARTITION)
+                .unwrap(),
+            Checkpoint::get_checkpoint_blob_name(NS_LOWER, HUB_LOWER, GROUP_LOWER, PARTITION)
+                .unwrap(),
+            "the event hub name case moved the checkpoint blob name"
+        );
+    }
+
+    /// An empty key field stays an error, and the message keeps the name of
+    /// the field that is empty. Each case leaves exactly one field empty,
+    /// because the two `*_name` functions check the fields in a different
+    /// order.
+    #[test]
+    fn key_functions_reject_empty_parameters() {
+        let cases: Vec<(Result<String>, &str)> = vec![
+            (
+                Checkpoint::get_checkpoint_blob_prefix_name("", "hub", "group"),
+                "Required field fully_qualified_namespace is empty",
+            ),
+            (
+                Checkpoint::get_checkpoint_blob_prefix_name(NS_LOWER, "", "group"),
+                "Required field event_hub_name is empty",
+            ),
+            (
+                Checkpoint::get_checkpoint_blob_prefix_name(NS_LOWER, "hub", ""),
+                "Required field consumer_group is empty",
+            ),
+            (
+                Ownership::get_ownership_prefix_name("", "hub", "group"),
+                "Required field fully_qualified_namespace is empty",
+            ),
+            (
+                Ownership::get_ownership_prefix_name(NS_LOWER, "", "group"),
+                "Required field event_hub_name is empty",
+            ),
+            (
+                Ownership::get_ownership_prefix_name(NS_LOWER, "hub", ""),
+                "Required field consumer_group is empty",
+            ),
+            (
+                Checkpoint::get_checkpoint_blob_name(NS_LOWER, "hub", "group", ""),
+                "Required field partition_id is empty",
+            ),
+            (
+                Ownership::get_ownership_name(NS_LOWER, "hub", "group", ""),
+                "Required field partition_id is empty",
+            ),
+        ];
+
+        for (result, expected_message) in cases {
+            let error = result.expect_err(expected_message);
+            assert_eq!(
+                *error.kind(),
+                ErrorKind::Other,
+                "wrong error kind for {expected_message:?}"
+            );
+            assert_eq!(error.to_string(), expected_message);
+        }
+    }
+
+    /// The fold is an ASCII fold, not a Unicode fold. This test guards the
+    /// implementation choice. A non-ASCII identifier is not a supported
+    /// Event Hubs scenario, because the service limits these names to ASCII.
+    #[test]
+    fn fold_is_ascii_only() {
+        // A trailing capital sigma folds to the final sigma form under
+        // `str::to_lowercase`, because that rule depends on the position in
+        // the word. An ASCII fold leaves both Greek letters alone.
+        let name = Checkpoint::get_checkpoint_blob_name(
+            NS_LOWER,
+            HUB_LOWER,
+            "$Default-\u{0391}\u{03A3}",
+            "0",
+        );
+        assert_eq!(
+            name.unwrap(),
+            "ns-test.servicebus.windows.net/my-eventhub/$default-\u{0391}\u{03A3}/checkpoint/0"
+        );
+    }
+}

sdk/eventhubs/azure_messaging_eventhubs/tests/eventhubs_checkpoint_store.rs

@@ -201,3 +201,67 @@ async fn checkpoints() -> azure_core::Result<()> {
 
     Ok(())
 }
+
+/// A checkpoint that one caller stores with a mixed case consumer group must
+/// come back to a caller that lists with a lowercase consumer group.
+#[tokio::test]
+async fn test_checkpoint_key_survives_consumer_group_case_change() {
+    common::setup();
+    let store = InMemoryCheckpointStore::new();
+    let checkpoint = Checkpoint {
+        fully_qualified_namespace: "NS-Test.ServiceBus.Windows.Net".to_string(),
+        event_hub_name: "My-EventHub".to_string(),
+        consumer_group: "$Default".to_string(),
+        partition_id: "Partition-A".to_string(),
+        ..Default::default()
+    };
+    store.update_checkpoint(checkpoint).await.unwrap();
+
+    let checkpoints = store
+        .list_checkpoints("ns-test.servicebus.windows.net", "my-eventhub", "$default")
+        .await
+        .unwrap();
+    assert_eq!(
+        checkpoints.len(),
+        1,
+        "the lowercase listing did not find the mixed case checkpoint"
+    );
+
+    // The store returns a clone of the stored record, so the fields keep the
+    // case of the caller that stored them. No folded value leaks into a field.
+    assert_eq!(checkpoints[0].partition_id, "Partition-A");
+    assert_eq!(checkpoints[0].consumer_group, "$Default");
+    assert_eq!(checkpoints[0].event_hub_name, "My-EventHub");
+    assert_eq!(
+        checkpoints[0].fully_qualified_namespace,
+        "NS-Test.ServiceBus.Windows.Net"
+    );
+}
+
+/// The load balancer drives the ownership path, so it needs the same
+/// stability across the case of the consumer group.
+#[tokio::test]
+async fn test_ownership_key_survives_consumer_group_case_change() {
+    common::setup();
+    let store = InMemoryCheckpointStore::new();
+    let ownership = Ownership {
+        fully_qualified_namespace: "NS-Test.ServiceBus.Windows.Net".to_string(),
+        event_hub_name: "My-EventHub".to_string(),
+        consumer_group: "$Default".to_string(),
+        partition_id: "Partition-A".to_string(),
+        owner_id: Some("owner_id".to_string()),
+        ..Default::default()
+    };
+    store.claim_ownership(&[ownership]).await.unwrap();
+
+    let ownerships = store
+        .list_ownerships("ns-test.servicebus.windows.net", "my-eventhub", "$default")
+        .await
+        .unwrap();
+    assert_eq!(
+        ownerships.len(),
+        1,
+        "the lowercase listing did not find the mixed case ownership"
+    );
+    assert_eq!(ownerships[0].partition_id, "Partition-A");
+}

sdk/eventhubs/azure_messaging_eventhubs_checkpointstore_blob/CHANGELOG.md

@@ -6,6 +6,11 @@
 
 ### Breaking Changes
 
+- Event Hubs treats the consumer group as case insensitive, so one deployment that spelled the group `$Default` on one run and `$default` on the next built two disjoint key sets and reprocessed events. ([#5099](https://github.com/Azure/azure-sdk-for-rust/issues/5099))
+- The checkpoint and ownership blob key that `BlobCheckpointStore` reads and writes now folds the fully qualified namespace, the event hub name, and the consumer group to lowercase ASCII. The partition id keeps its case. The .NET, JavaScript, and Python clients fold these three names; the Go and Java clients do not, so this change moves the Rust crate from the second group into the first.
+- The old key `NS.ServiceBus.Windows.Net/My-Hub/$Default/checkpoint/0` becomes `ns.servicebus.windows.net/my-hub/$default/checkpoint/0`. The ownership key changes in the same way.
+- Migration: records that an older Rust client wrote stay at the old key and become unreachable. The change adds no dual read and no fallback lookup. A processor that starts against an existing container resumes from its configured start position.
+
 ### Bugs Fixed
 
 ### Other Changes

sdk/eventhubs/azure_messaging_eventhubs_checkpointstore_blob/assets.json

@@ -2,5 +2,5 @@
   "AssetsRepo": "Azure/azure-sdk-assets",
   "AssetsRepoPrefixPath": "rust",
   "TagPrefix": "rust/eventhubs/azure_messaging_eventhubs_checkpointstore_blob",
-  "Tag": "rust/eventhubs/azure_messaging_eventhubs_checkpointstore_blob_d7273c4b84"
+  "Tag": "rust/eventhubs/azure_messaging_eventhubs_checkpointstore_blob_a1ab5fd8f3"
 }