fix: Kinesis event-source region, EventBridge archive rule ARN, Logs group ARN form, cached Bedrock regexes

crates/fakecloud-eventbridge/src/helpers.rs

@@ -1123,7 +1123,7 @@ pub(crate) fn deliver_to_logs(
         .entry(group_name.to_string())
         .or_insert_with(|| fakecloud_logs::LogGroup {
             name: group_name.to_string(),
-            arn: fakecloud_logs::log_group_arn(&region, &account_id, group_name),
+            arn: fakecloud_logs::log_group_stored_arn(&region, &account_id, group_name),
             creation_time: ts_millis,
             retention_in_days: None,
             kms_key_id: None,
@@ -1145,14 +1145,15 @@ pub(crate) fn deliver_to_logs(
         return;
     }
 
+    let stream_arn = group.stream_arn("events");
     let stream = group
         .log_streams
         .entry(stream_name.clone())
         .or_insert_with(|| fakecloud_logs::LogStream {
             persistence_id: uuid::Uuid::new_v4().to_string(),
             last_sequence: 0,
             name: stream_name,
-            arn: format!("{}:log-stream:events", group.arn),
+            arn: stream_arn,
             creation_time: ts_millis,
             first_event_timestamp: None,
             last_event_timestamp: None,
@@ -1577,5 +1578,15 @@ mod logs_persist_tests {
             "group ARN must carry the target region+account: {}",
             group.arn
         );
+        // Stored in the same `:*` form CreateLogGroup stores, and the stream
+        // ARN hangs off the suffix-less group ARN.
+        assert_eq!(
+            group.arn,
+            "arn:aws:logs:eu-central-1:999999999999:log-group:/eb/xacct:*"
+        );
+        assert_eq!(
+            group.log_streams["events"].arn,
+            "arn:aws:logs:eu-central-1:999999999999:log-group:/eb/xacct:log-stream:events"
+        );
     }
 }

crates/fakecloud-eventbridge/src/service_archives_replays.rs

@@ -89,7 +89,7 @@ impl EventBridgeService {
 
         // Create the archive rule
         let rule_name = format!("Events-Archive-{name}");
-        let rule_arn = rule_arn(&req.region, &state.account_id, "default", &rule_name);
+        let rule_arn = rule_arn(&req.region, &state.account_id, &bus_name, &rule_name);
         // Merge archive event pattern with replay-name filter
         let rule_event_pattern = {
             let mut merged = if let Some(ref ep) = event_pattern {

crates/fakecloud-eventbridge/src/service_tests.rs

@@ -3563,6 +3563,47 @@ fn create_archive_missing_source_arn_errors() {
     assert!(svc.create_archive(&req).is_err());
 }
 
+#[test]
+fn archive_rule_on_custom_bus_carries_bus_in_arn() {
+    let svc = make_service();
+    create_event_bus(&svc, "orders");
+    svc.create_archive(&make_request(
+        "CreateArchive",
+        json!({
+            "ArchiveName": "orders-arc",
+            "EventSourceArn": "arn:aws:events:us-east-1:123456789012:event-bus/orders"
+        }),
+    ))
+    .unwrap();
+    create_archive(&svc, "default-arc");
+
+    let resp = svc
+        .describe_rule(&make_request(
+            "DescribeRule",
+            json!({"Name": "Events-Archive-orders-arc", "EventBusName": "orders"}),
+        ))
+        .unwrap();
+    let body: Value = serde_json::from_slice(resp.body.expect_bytes()).unwrap();
+    assert_eq!(
+        body["Arn"],
+        "arn:aws:events:us-east-1:123456789012:rule/orders/Events-Archive-orders-arc"
+    );
+    assert_eq!(body["EventBusName"], "orders");
+
+    // An archive on the default bus keeps the bus-less rule ARN.
+    let resp = svc
+        .describe_rule(&make_request(
+            "DescribeRule",
+            json!({"Name": "Events-Archive-default-arc"}),
+        ))
+        .unwrap();
+    let body: Value = serde_json::from_slice(resp.body.expect_bytes()).unwrap();
+    assert_eq!(
+        body["Arn"],
+        "arn:aws:events:us-east-1:123456789012:rule/Events-Archive-default-arc"
+    );
+}
+
 #[test]
 fn delete_archive_missing_errors() {
     let svc = make_service();

crates/fakecloud-logs/src/ingest.rs

@@ -41,7 +41,7 @@ pub fn append_events(
         .entry(group_name.to_string())
         .or_insert_with(|| LogGroup {
             name: group_name.to_string(),
-            arn: crate::state::log_group_arn(region, account_id, group_name),
+            arn: crate::state::log_group_stored_arn(region, account_id, group_name),
             creation_time: now,
             retention_in_days: None,
             kms_key_id: None,
@@ -55,14 +55,15 @@ pub fn append_events(
             deletion_protection: false,
             log_group_class: Some("STANDARD".into()),
         });
+    let stream_arn = group.stream_arn(stream_name);
     let stream = group
         .log_streams
         .entry(stream_name.to_string())
         .or_insert_with(|| LogStream {
             persistence_id: uuid::Uuid::new_v4().to_string(),
             last_sequence: 0,
             name: stream_name.to_string(),
-            arn: format!("{}:log-stream:{}", group.arn, stream_name),
+            arn: stream_arn,
             creation_time: now,
             first_event_timestamp: None,
             last_event_timestamp: None,

crates/fakecloud-logs/src/lib.rs

@@ -8,9 +8,9 @@ pub(crate) mod validation;
 
 pub use service::{infer_delivery_destination_type, save_logs_state, LogsService};
 pub use state::{
-    log_group_arn, Delivery, DeliveryDestination, DeliverySource, Destination, LogAnomaly,
-    LogEvent, LogGroup, LogStream, LogsSnapshot, LogsState, MetricFilter, MetricTransformation,
-    QueryDefinition, ResourcePolicy, SharedLogsState, SubscriptionFilter,
+    log_group_arn, log_group_stored_arn, Delivery, DeliveryDestination, DeliverySource,
+    Destination, LogAnomaly, LogEvent, LogGroup, LogStream, LogsSnapshot, LogsState, MetricFilter,
+    MetricTransformation, QueryDefinition, ResourcePolicy, SharedLogsState, SubscriptionFilter,
     LOGS_SNAPSHOT_SCHEMA_VERSION, VALID_RETENTION_DAYS,
 };
 

crates/fakecloud-logs/src/service/groups.rs

@@ -66,10 +66,7 @@ impl LogsService {
             ));
         }
 
-        let arn = format!(
-            "{}:*",
-            crate::state::log_group_arn(&req.region, &state.account_id, &name)
-        );
+        let arn = crate::state::log_group_stored_arn(&req.region, &state.account_id, &name);
         let now = Utc::now().timestamp_millis();
 
         let tags = body["tags"]
@@ -208,15 +205,15 @@ impl LogsService {
         let log_groups: Vec<Value> = page
             .iter()
             .map(|g| {
-                let log_group_arn = g.arn.trim_end_matches(":*").to_string();
+                let log_group_arn = g.log_group_arn();
                 let metric_filter_count = state
                     .metric_filters
                     .iter()
                     .filter(|mf| mf.log_group_name == g.name)
                     .count();
                 let mut obj = json!({
                     "logGroupName": g.name,
-                    "arn": g.arn,
+                    "arn": g.wildcard_arn(),
                     "logGroupArn": log_group_arn,
                     "creationTime": g.creation_time,
                     "storedBytes": g.stored_bytes,
@@ -706,7 +703,7 @@ impl LogsService {
         let log_groups: Vec<Value> = page
             .iter()
             .map(|g| {
-                let log_group_arn = g.arn.trim_end_matches(":*").to_string();
+                let log_group_arn = g.log_group_arn();
                 json!({
                     "logGroupName": g.name,
                     "logGroupArn": log_group_arn,
@@ -764,6 +761,75 @@ mod tests {
         assert!(names.contains(&"/app/api"));
     }
 
+    #[test]
+    fn created_and_implicit_log_groups_describe_identically() {
+        let svc = make_service();
+        create_group(&svc, "explicit");
+        let event = [crate::ingest::IngestEvent {
+            timestamp_ms: chrono::Utc::now().timestamp_millis(),
+            message: "m".into(),
+        }];
+        // Ingest into the created group, and into a group ingest creates.
+        for group in ["explicit", "implicit"] {
+            crate::ingest::append_events(
+                &svc.state,
+                "123456789012",
+                "us-east-1",
+                group,
+                "s",
+                &event,
+            );
+        }
+
+        let resp = svc
+            .describe_log_groups(&make_request("DescribeLogGroups", json!({})))
+            .unwrap();
+        let body: Value = serde_json::from_slice(resp.body.expect_bytes()).unwrap();
+        let groups = body["logGroups"].as_array().unwrap();
+        assert_eq!(groups.len(), 2);
+        for g in groups {
+            let name = g["logGroupName"].as_str().unwrap();
+            let base = format!("arn:aws:logs:us-east-1:123456789012:log-group:{name}");
+            assert_eq!(g["arn"], format!("{base}:*"), "{name}");
+            assert_eq!(g["logGroupArn"], base, "{name}");
+
+            let resp = svc
+                .describe_log_streams(&make_request(
+                    "DescribeLogStreams",
+                    json!({ "logGroupName": name }),
+                ))
+                .unwrap();
+            let streams: Value = serde_json::from_slice(resp.body.expect_bytes()).unwrap();
+            assert_eq!(
+                streams["logStreams"][0]["arn"],
+                format!("{base}:log-stream:s"),
+                "{name}"
+            );
+        }
+        let strip = |g: &Value| {
+            let mut g = g.clone();
+            let obj = g.as_object_mut().unwrap();
+            for k in [
+                "logGroupName",
+                "arn",
+                "logGroupArn",
+                "creationTime",
+                "storedBytes",
+            ] {
+                obj.remove(k);
+            }
+            g
+        };
+        assert_eq!(strip(&groups[0]), strip(&groups[1]));
+
+        let mas = svc.state.read();
+        let state = mas.default_ref();
+        assert_eq!(
+            state.log_groups["implicit"].arn,
+            "arn:aws:logs:us-east-1:123456789012:log-group:implicit:*"
+        );
+    }
+
     #[test]
     fn describe_log_groups_pattern_empty_returns_all() {
         let svc = make_service();

crates/fakecloud-logs/src/service/policies.rs

@@ -566,7 +566,7 @@ impl LogsService {
         // the index-policy `logGroupIdentifier` reports it without the suffix so
         // the resource's `log_group_name` round-trips (otherwise the read sees a
         // `:*`-suffixed identifier and forces replacement).
-        let log_group_identifier = group.arn.trim_end_matches(":*").to_string();
+        let log_group_identifier = group.log_group_arn().to_string();
         let result = json!({
             "indexPolicy": {
                 "policyName": policy_name,
@@ -609,7 +609,7 @@ impl LogsService {
                 id.to_string()
             };
             if let Some(group) = state.log_groups.get(&group_name) {
-                let log_group_identifier = group.arn.trim_end_matches(":*");
+                let log_group_identifier = group.log_group_arn();
                 for p in &group.index_policies {
                     policies.push(json!({
                         "policyName": p.policy_name,
@@ -744,7 +744,7 @@ impl LogsService {
                     for f in fields {
                         if let Some(name) = f.as_str() {
                             field_indexes.push(json!({
-                                "logGroupIdentifier": group.arn,
+                                "logGroupIdentifier": group.log_group_arn(),
                                 "fieldIndexName": name,
                                 "lastScanTime": p.last_updated_time,
                                 "firstEventTime": p.last_updated_time,
@@ -855,7 +855,7 @@ impl LogsService {
         })?;
 
         let mut result = json!({
-            "logGroupIdentifier": group.arn,
+            "logGroupIdentifier": group.log_group_arn(),
         });
         if let Some(ref t) = group.transformer {
             result["transformerConfig"] = t.transformer_config.clone();

crates/fakecloud-logs/src/service/streams.rs

@@ -546,14 +546,18 @@ impl LogsService {
 
         // Collect delivery pipeline info: find active deliveries whose source
         // resource ARN matches this log group's ARN.
-        let group_arn = group.arn.clone();
+        let group_arn = group.log_group_arn().to_string();
         let delivery_targets: Vec<String> = state
             .deliveries
             .values()
             .filter_map(|d| {
                 // Check if the delivery source references this log group
                 if let Some(source) = state.delivery_sources.get(&d.delivery_source_name) {
-                    if source.resource_arns.contains(&group_arn) {
+                    if source
+                        .resource_arns
+                        .iter()
+                        .any(|a| a.strip_suffix(":*").unwrap_or(a) == group_arn)
+                    {
                         // Find the destination's S3 bucket configuration
                         if let Some(dest) = state
                             .delivery_destinations

crates/fakecloud-logs/src/service/syslog.rs

@@ -67,7 +67,7 @@ impl LogsService {
                 format!("The specified log group does not exist: {group_name}"),
             )
         })?;
-        let log_group_arn = group.arn.clone();
+        let log_group_arn = group.log_group_arn().to_string();
 
         let created_at = state
             .syslog_configurations

crates/fakecloud-logs/src/state.rs

@@ -181,6 +181,15 @@ pub fn log_group_arn(region: &str, account_id: &str, name: &str) -> String {
         .to_string()
 }
 
+/// The ARN stored on [`LogGroup::arn`]: [`log_group_arn`] plus the trailing
+/// `:*`, the form `DescribeLogGroups` reports on `arn`. Every path that creates
+/// a log group (CreateLogGroup, implicit creation on ingest or EventBridge
+/// delivery) stores this form so the group describes the same way regardless
+/// of how it came to exist.
+pub fn log_group_stored_arn(region: &str, account_id: &str, name: &str) -> String {
+    format!("{}:*", log_group_arn(region, account_id, name))
+}
+
 #[derive(Clone, serde::Serialize, serde::Deserialize)]
 pub struct LogGroup {
     pub name: String,
@@ -545,6 +554,23 @@ impl LogsState {
 }
 
 impl LogGroup {
+    /// The group's ARN without the trailing `:*` (the `logGroupArn` form, and
+    /// the prefix of its log streams' ARNs). Tolerates a stored ARN that lacks
+    /// the suffix, as snapshots from before the stored form was unified do.
+    pub fn log_group_arn(&self) -> &str {
+        self.arn.strip_suffix(":*").unwrap_or(&self.arn)
+    }
+
+    /// The group's ARN with the trailing `:*` (the `DescribeLogGroups` `arn`).
+    pub fn wildcard_arn(&self) -> String {
+        format!("{}:*", self.log_group_arn())
+    }
+
+    /// The ARN of log stream `stream_name` in this group.
+    pub fn stream_arn(&self, stream_name: &str) -> String {
+        format!("{}:log-stream:{stream_name}", self.log_group_arn())
+    }
+
     pub(crate) fn metadata(&self) -> Self {
         Self {
             name: self.name.clone(),