fix(glue,sagemaker,pipes,sns,scheduler): persist versions/status + honor DLQ + FIFO/EB params (bug-hunt)

Stubs cluster B from the 2026-07-15 bug hunt (1.31-1.36).

- glue (1.31): table versioning was faked — GetTableVersion(s) synthesized a
  single "1" from the live table and CreateTable/UpdateTable archived nothing.
  Now every Create/Update archives a real monotonic version into the
  table_versions store; Get/List read it (newest-first), DeleteTable purges it,
  with a synthesized fallback for pre-archive tables.
- sagemaker (1.32): Stop*/Start* lifecycle ops went through the generic
  action no-op and never advanced status, so Describe kept reporting the default.
  They now transition the target resource's {Family}Status (Stopped / InService /
  Scheduled / Cancelled / ... per family), inserting the member when a
  freshly-created record carries none.
- pipes (1.33): SQS-target FIFO params (MessageGroupId / MessageDeduplicationId)
  from TargetParameters.SqsQueueParameters were dropped at delivery, so a .fifo
  target rejected the send. Now forwarded via send_to_sqs_with_attrs.
- pipes (1.34): stream-source DLQ/retry was never applied — a poison record
  retried forever and wedged the stream. Now honors
  {Kinesis,DynamoDB}StreamParameters.MaximumRetryAttempts + DeadLetterConfig:
  once a window exceeds the cap it's routed to the DLQ and the checkpoint
  advances past it.
- sns (1.35): RedrivePolicy DLQ was wired for HTTP subscribers only; SQS and
  Lambda subscribers dropped the notification on failure. SQS now uses a
  fallible send and Lambda routes invoke failures to the subscription DLQ.
- scheduler (1.36): the EventBridge-bus target ignored
  EventBridgeParameters.{Source, DetailType} (hardcoded aws.scheduler /
  "Scheduled Event"), and the SageMaker-Pipeline target never fired. Both are
  now honored; sagemaker:pipeline targets start a real PipelineExecution via a
  new cross-service SageMakerPipelineDelivery hook.

Tests: e2e for glue versions, sagemaker stop/start, SNS SQS-subscriber DLQ; unit
for scheduler EB/pipeline delivery and pipes stream retry-config parsing.

Author: vieiralucas

crates/fakecloud-core/src/delivery.rs

@@ -19,6 +19,8 @@ pub struct DeliveryBus {
     kinesis_sender: Option<Arc<dyn KinesisDelivery>>,
     /// Start Step Functions executions.
     stepfunctions_starter: Option<Arc<dyn StepFunctionsDelivery>>,
+    /// Start SageMaker pipeline executions (Scheduler sagemaker target).
+    sagemaker_pipeline_starter: Option<Arc<dyn SageMakerPipelineDelivery>>,
     /// Write objects to S3 buckets.
     s3_writer: Option<Arc<dyn S3Delivery>>,
     /// Put records into Firehose delivery streams.
@@ -196,6 +198,14 @@ pub trait StepFunctionsDelivery: Send + Sync {
     fn start_execution(&self, state_machine_arn: &str, input: &str);
 }
 
+/// Cross-service SageMaker pipeline dispatch used by EventBridge Scheduler's
+/// `sagemaker:pipeline` target. The pipeline is identified by ARN
+/// (`arn:aws:sagemaker:<region>:<account>:pipeline/<name>`); `parameters` is the
+/// target's `PipelineParameterList`.
+pub trait SageMakerPipelineDelivery: Send + Sync {
+    fn start_pipeline_execution(&self, pipeline_arn: &str, parameters: &serde_json::Value);
+}
+
 /// Cross-service Kinesis Data Firehose dispatch used by services
 /// (CloudWatch Logs subscription filters, EventBridge targets) that
 /// route records into a delivery stream without depending on the
@@ -385,6 +395,7 @@ impl DeliveryBus {
             lambda_invoker: None,
             kinesis_sender: None,
             stepfunctions_starter: None,
+            sagemaker_pipeline_starter: None,
             s3_writer: None,
             firehose_sender: None,
             ses_dispatcher: None,
@@ -679,6 +690,18 @@ impl DeliveryBus {
         }
     }
 
+    pub fn with_sagemaker_pipeline(mut self, starter: Arc<dyn SageMakerPipelineDelivery>) -> Self {
+        self.sagemaker_pipeline_starter = Some(starter);
+        self
+    }
+
+    /// Start a SageMaker pipeline execution if a starter is wired.
+    pub fn start_sagemaker_pipeline(&self, pipeline_arn: &str, parameters: &serde_json::Value) {
+        if let Some(ref starter) = self.sagemaker_pipeline_starter {
+            starter.start_pipeline_execution(pipeline_arn, parameters);
+        }
+    }
+
     pub fn with_stepfunctions(mut self, starter: Arc<dyn StepFunctionsDelivery>) -> Self {
         self.stepfunctions_starter = Some(starter);
         self

crates/fakecloud-e2e/tests/glue.rs

@@ -190,6 +190,95 @@ async fn table_lifecycle() {
     assert!(err.into_service_error().is_entity_not_found_exception());
 }
 
+#[tokio::test]
+async fn table_versions_are_archived_on_create_and_update() {
+    let server = TestServer::start().await;
+    let glue = server.glue_client().await;
+
+    glue.create_database()
+        .database_input(DatabaseInput::builder().name("versdb").build().unwrap())
+        .send()
+        .await
+        .expect("create db");
+
+    glue.create_table()
+        .database_name("versdb")
+        .table_input(table_input("t"))
+        .send()
+        .await
+        .expect("create table");
+
+    // A create archives version 1.
+    let v1 = glue
+        .get_table_versions()
+        .database_name("versdb")
+        .table_name("t")
+        .send()
+        .await
+        .expect("get versions after create");
+    assert_eq!(v1.table_versions().len(), 1);
+    assert_eq!(v1.table_versions()[0].version_id(), Some("1"));
+
+    // An update archives a second version with the mutated table.
+    glue.update_table()
+        .database_name("versdb")
+        .table_input(
+            TableInput::builder()
+                .name("t")
+                .description("updated description")
+                .build()
+                .unwrap(),
+        )
+        .send()
+        .await
+        .expect("update table");
+
+    let v2 = glue
+        .get_table_versions()
+        .database_name("versdb")
+        .table_name("t")
+        .send()
+        .await
+        .expect("get versions after update");
+    assert_eq!(
+        v2.table_versions().len(),
+        2,
+        "update must archive a new version"
+    );
+
+    // GetTableVersion with no VersionId returns the latest (2), reflecting the
+    // updated description rather than a synthesized default.
+    let latest = glue
+        .get_table_version()
+        .database_name("versdb")
+        .table_name("t")
+        .send()
+        .await
+        .expect("get latest version");
+    let tv = latest.table_version().expect("table version");
+    assert_eq!(tv.version_id(), Some("2"));
+    assert_eq!(
+        tv.table().and_then(|t| t.description()),
+        Some("updated description")
+    );
+
+    // Deleting the table purges its version archive.
+    glue.delete_table()
+        .database_name("versdb")
+        .name("t")
+        .send()
+        .await
+        .expect("delete");
+    let after = glue
+        .get_table_versions()
+        .database_name("versdb")
+        .table_name("t")
+        .send()
+        .await
+        .expect("get versions after delete");
+    assert!(after.table_versions().is_empty());
+}
+
 #[tokio::test]
 async fn partition_lifecycle() {
     let server = TestServer::start().await;

crates/fakecloud-e2e/tests/sagemaker.rs

@@ -193,6 +193,59 @@ async fn sagemaker_pipeline_execution_round_trip() {
     );
 }
 
+/// Stop*/Start* lifecycle ops must advance the target resource's status so a
+/// subsequent Describe reflects the transition instead of the default state.
+#[tokio::test]
+async fn sagemaker_notebook_instance_stop_start_transitions_status() {
+    let server = TestServer::start().await;
+    let client = sagemaker_client(&server).await;
+
+    client
+        .create_notebook_instance()
+        .notebook_instance_name("nb-1")
+        .instance_type(aws_sdk_sagemaker::types::InstanceType::MlT2Medium)
+        .role_arn("arn:aws:iam::000000000000:role/SageMakerRole")
+        .send()
+        .await
+        .expect("create_notebook_instance");
+
+    client
+        .stop_notebook_instance()
+        .notebook_instance_name("nb-1")
+        .send()
+        .await
+        .expect("stop_notebook_instance");
+    let stopped = client
+        .describe_notebook_instance()
+        .notebook_instance_name("nb-1")
+        .send()
+        .await
+        .expect("describe after stop");
+    assert_eq!(
+        stopped.notebook_instance_status(),
+        Some(&aws_sdk_sagemaker::types::NotebookInstanceStatus::Stopped),
+        "StopNotebookInstance must move status to Stopped"
+    );
+
+    client
+        .start_notebook_instance()
+        .notebook_instance_name("nb-1")
+        .send()
+        .await
+        .expect("start_notebook_instance");
+    let started = client
+        .describe_notebook_instance()
+        .notebook_instance_name("nb-1")
+        .send()
+        .await
+        .expect("describe after start");
+    assert_eq!(
+        started.notebook_instance_status(),
+        Some(&aws_sdk_sagemaker::types::NotebookInstanceStatus::InService),
+        "StartNotebookInstance must move status to InService"
+    );
+}
+
 /// `AddAssociation` is the only creation path for a lineage edge; the edge must
 /// then be listed and deletable.
 #[tokio::test]

crates/fakecloud-e2e/tests/sns.rs

@@ -264,6 +264,114 @@ async fn sns_publish_delivers_to_sqs_subscriber() {
     assert_eq!(envelope["TopicArn"], topic_arn);
 }
 
+/// An SQS subscriber whose target queue is gone must route the notification to
+/// the subscription's RedrivePolicy DLQ instead of silently dropping it — the
+/// DLQ was previously honored for HTTP subscribers only.
+#[tokio::test]
+async fn sns_sqs_subscriber_failure_routes_to_dlq() {
+    let server = TestServer::start().await;
+    let sns = server.sns_client().await;
+    let sqs = server.sqs_client().await;
+
+    async fn queue_arn(sqs: &aws_sdk_sqs::Client, url: &str) -> String {
+        sqs.get_queue_attributes()
+            .queue_url(url)
+            .attribute_names(QueueAttributeName::QueueArn)
+            .send()
+            .await
+            .unwrap()
+            .attributes()
+            .unwrap()
+            .get(&QueueAttributeName::QueueArn)
+            .unwrap()
+            .to_string()
+    }
+
+    // Main queue (will be deleted) + DLQ.
+    let main_url = sqs
+        .create_queue()
+        .queue_name("dlq-main")
+        .send()
+        .await
+        .unwrap()
+        .queue_url()
+        .unwrap()
+        .to_string();
+    let main_arn = queue_arn(&sqs, &main_url).await;
+    let dlq_url = sqs
+        .create_queue()
+        .queue_name("dlq-target")
+        .send()
+        .await
+        .unwrap()
+        .queue_url()
+        .unwrap()
+        .to_string();
+    let dlq_arn = queue_arn(&sqs, &dlq_url).await;
+
+    let topic_arn = sns
+        .create_topic()
+        .name("dlq-topic")
+        .send()
+        .await
+        .unwrap()
+        .topic_arn()
+        .unwrap()
+        .to_string();
+    let sub_arn = sns
+        .subscribe()
+        .topic_arn(&topic_arn)
+        .protocol("sqs")
+        .endpoint(&main_arn)
+        .return_subscription_arn(true)
+        .send()
+        .await
+        .unwrap()
+        .subscription_arn()
+        .unwrap()
+        .to_string();
+    sns.set_subscription_attributes()
+        .subscription_arn(&sub_arn)
+        .attribute_name("RedrivePolicy")
+        .attribute_value(format!(r#"{{"deadLetterTargetArn":"{dlq_arn}"}}"#))
+        .send()
+        .await
+        .unwrap();
+
+    // Delete the subscribed queue so delivery fails.
+    sqs.delete_queue()
+        .queue_url(&main_url)
+        .send()
+        .await
+        .unwrap();
+
+    sns.publish()
+        .topic_arn(&topic_arn)
+        .message("to-dlq")
+        .send()
+        .await
+        .unwrap();
+
+    // The notification must land in the DLQ.
+    let msgs = sqs
+        .receive_message()
+        .queue_url(&dlq_url)
+        .max_number_of_messages(10)
+        .wait_time_seconds(1)
+        .send()
+        .await
+        .unwrap();
+    assert_eq!(
+        msgs.messages().len(),
+        1,
+        "failed delivery must route to DLQ"
+    );
+    assert!(
+        msgs.messages()[0].body().unwrap().contains("to-dlq"),
+        "DLQ envelope should wrap the original message"
+    );
+}
+
 /// Subscribing Lambda/email/sms protocols should succeed (stub delivery).
 #[tokio::test]
 async fn sns_subscribe_lambda_email_sms() {

crates/fakecloud-glue/src/service.rs

@@ -1101,26 +1101,27 @@ impl GlueService {
         if db.tables.contains_key(&name) {
             return Err(already_exists(format!("Table {name} already exists")));
         }
-        db.tables.insert(
-            name.clone(),
-            Table {
-                name,
-                database_name: db_name,
-                description: input["Description"].as_str().map(|s| s.to_string()),
-                owner: input["Owner"].as_str().map(|s| s.to_string()),
-                create_time: now,
-                update_time: now,
-                last_access_time: None,
-                retention: input["Retention"].as_i64().unwrap_or(0),
-                storage_descriptor: parse_storage_descriptor(&input["StorageDescriptor"]),
-                partition_keys: parse_columns(&input["PartitionKeys"]),
-                view_original_text: input["ViewOriginalText"].as_str().map(|s| s.to_string()),
-                view_expanded_text: input["ViewExpandedText"].as_str().map(|s| s.to_string()),
-                table_type: input["TableType"].as_str().map(|s| s.to_string()),
-                parameters: parse_string_map(&input["Parameters"]),
-                partitions: BTreeMap::new(),
-            },
-        );
+        let table = Table {
+            name: name.clone(),
+            database_name: db_name.clone(),
+            description: input["Description"].as_str().map(|s| s.to_string()),
+            owner: input["Owner"].as_str().map(|s| s.to_string()),
+            create_time: now,
+            update_time: now,
+            last_access_time: None,
+            retention: input["Retention"].as_i64().unwrap_or(0),
+            storage_descriptor: parse_storage_descriptor(&input["StorageDescriptor"]),
+            partition_keys: parse_columns(&input["PartitionKeys"]),
+            view_original_text: input["ViewOriginalText"].as_str().map(|s| s.to_string()),
+            view_expanded_text: input["ViewExpandedText"].as_str().map(|s| s.to_string()),
+            table_type: input["TableType"].as_str().map(|s| s.to_string()),
+            parameters: parse_string_map(&input["Parameters"]),
+            partitions: BTreeMap::new(),
+        };
+        let tv_json = table_json(&table);
+        db.tables.insert(name.clone(), table);
+        // Archive the initial version so GetTableVersion(s) return real data.
+        state.archive_table_version(&db_name, &name, tv_json);
         Ok(AwsResponse::ok_json(json!({})))
     }
 
@@ -1224,6 +1225,10 @@ impl GlueService {
         if let Some(r) = input["Retention"].as_i64() {
             t.retention = r;
         }
+        // Snapshot the mutated table as a new archived version (Glue bumps the
+        // VersionId on every UpdateTable).
+        let tv_json = table_json(t);
+        state.archive_table_version(db_name, name, tv_json);
         Ok(AwsResponse::ok_json(json!({})))
     }
 
@@ -1242,6 +1247,7 @@ impl GlueService {
         if db.tables.remove(name).is_none() {
             return Err(entity_not_found(format!("Table {name} not found")));
         }
+        state.purge_table_versions(db_name, name);
         Ok(AwsResponse::ok_json(json!({})))
     }