ref(api): Update the update image endpoint behavior (supabase#336)

Author: iambriccardo

etl-api/src/routes/pipelines.rs

@@ -95,8 +95,8 @@ pub enum PipelineError {
     #[error("A pipeline already exists for this source and destination combination")]
     DuplicatePipeline,
 
-    #[error("The specified image with id {0} was not found")]
-    ImageNotFoundById(i64),
+    #[error("The specified image id {0} does not match the default image id")]
+    ImageIdNotDefault(i64),
 
     #[error("There was an error while looking up table information in the source database: {0}")]
     TableLookup(#[from] TableLookupError),
@@ -152,13 +152,14 @@ impl ResponseError for PipelineError {
             | PipelineError::TableLookup(_)
             | PipelineError::InvalidTableReplicationState(_)
             | PipelineError::MissingTableReplicationState => StatusCode::INTERNAL_SERVER_ERROR,
-            PipelineError::NotRollbackable(_) => StatusCode::BAD_REQUEST,
             PipelineError::PipelineNotFound(_)
-            | PipelineError::ImageNotFoundById(_)
-            | PipelineError::EtlStateNotInitialized => StatusCode::NOT_FOUND,
-            PipelineError::TenantId(_)
-            | PipelineError::SourceNotFound(_)
-            | PipelineError::DestinationNotFound(_) => StatusCode::BAD_REQUEST,
+            | PipelineError::EtlStateNotInitialized
+            | PipelineError::ImageIdNotDefault(_)
+            | PipelineError::DestinationNotFound(_)
+            | PipelineError::SourceNotFound(_) => StatusCode::NOT_FOUND,
+            PipelineError::TenantId(_) | PipelineError::NotRollbackable(_) => {
+                StatusCode::BAD_REQUEST
+            }
             PipelineError::DuplicatePipeline => StatusCode::CONFLICT,
         }
     }
@@ -237,9 +238,9 @@ pub struct ReadPipelinesResponse {
 }
 
 #[derive(Debug, Serialize, Deserialize, ToSchema)]
-pub struct UpdatePipelineImageRequest {
-    #[schema(example = 1)]
-    pub image_id: Option<i64>,
+pub struct UpdatePipelineVersionRequest {
+    #[schema(example = 1, required = true)]
+    pub version_id: i64,
 }
 
 #[derive(Debug, Serialize, Deserialize, ToSchema)]
@@ -1044,29 +1045,29 @@ pub async fn rollback_table_state(
 }
 
 #[utoipa::path(
-    summary = "Update pipeline image",
-    description = "Updates the pipeline's container image while preserving its state.",
-    request_body = UpdatePipelineImageRequest,
+    summary = "Update pipeline version",
+    description = "Updates the pipeline's version while preserving its state.",
+    request_body = UpdatePipelineVersionRequest,
     params(
         ("pipeline_id" = i64, Path, description = "Unique ID of the pipeline"),
         ("tenant_id" = String, Header, description = "Tenant ID used to scope the request")
     ),
     responses(
-        (status = 200, description = "Pipeline image updated successfully"),
+        (status = 200, description = "Pipeline version updated successfully"),
         (status = 400, description = "Bad request or pipeline not running", body = ErrorMessage),
-        (status = 404, description = "Pipeline or image not found", body = ErrorMessage),
+        (status = 404, description = "Pipeline or version not found", body = ErrorMessage),
         (status = 500, description = "Internal server error", body = ErrorMessage)
     ),
     tag = "Pipelines"
 )]
-#[post("/pipelines/{pipeline_id}/update-image")]
-pub async fn update_pipeline_image(
+#[post("/pipelines/{pipeline_id}/update-version")]
+pub async fn update_pipeline_version(
     req: HttpRequest,
     pool: Data<PgPool>,
     encryption_key: Data<EncryptionKey>,
     k8s_client: Data<dyn K8sClient>,
     pipeline_id: Path<i64>,
-    update_request: Json<UpdatePipelineImageRequest>,
+    update_request: Json<UpdatePipelineVersionRequest>,
 ) -> Result<impl Responder, PipelineError> {
     let tenant_id = extract_tenant_id(&req)?;
     let pipeline_id = pipeline_id.into_inner();
@@ -1077,14 +1078,18 @@ pub async fn update_pipeline_image(
     let (pipeline, replicator, current_image, source, destination) =
         read_all_required_data(&mut txn, tenant_id, pipeline_id, &encryption_key).await?;
 
-    let target_image = match update_request.image_id {
-        Some(image_id) => db::images::read_image(txn.deref_mut(), image_id)
-            .await?
-            .ok_or(PipelineError::ImageNotFoundById(image_id))?,
-        None => db::images::read_default_image(txn.deref_mut())
-            .await?
-            .ok_or(PipelineError::NoDefaultImageFound)?,
-    };
+    // Only allow updating to the current default image. The client must provide the version id and
+    // it must match the default version id. If it does not, we consider this a race condition and we
+    // fail the update.
+    let default_image = db::images::read_default_image(txn.deref_mut())
+        .await?
+        .ok_or(PipelineError::NoDefaultImageFound)?;
+
+    if update_request.version_id != default_image.id {
+        return Err(PipelineError::ImageIdNotDefault(update_request.version_id));
+    }
+
+    let target_image = default_image;
 
     // If the image ids are different, we change the database entry.
     if target_image.id != current_image.id {

etl-api/src/startup.rs

@@ -41,11 +41,11 @@ use crate::{
             CreatePipelineRequest, CreatePipelineResponse, GetPipelineReplicationStatusResponse,
             GetPipelineStatusResponse, GetPipelineVersionResponse, ReadPipelineResponse,
             ReadPipelinesResponse, SimpleTableReplicationState, TableReplicationStatus,
-            UpdatePipelineImageRequest, UpdatePipelineRequest, create_pipeline, delete_pipeline,
+            UpdatePipelineRequest, UpdatePipelineVersionRequest, create_pipeline, delete_pipeline,
             get_pipeline_replication_status, get_pipeline_status, get_pipeline_version,
             read_all_pipelines, read_pipeline, rollback_table_state, start_pipeline,
             stop_all_pipelines, stop_pipeline, update_pipeline, update_pipeline_config,
-            update_pipeline_image,
+            update_pipeline_version,
         },
         sources::{
             CreateSourceRequest, CreateSourceResponse, ReadSourceResponse, ReadSourcesResponse,
@@ -183,7 +183,7 @@ pub async fn run(
             ReadPipelineResponse,
             ReadPipelinesResponse,
             GetPipelineVersionResponse,
-            UpdatePipelineImageRequest,
+            UpdatePipelineVersionRequest,
             GetPipelineStatusResponse,
             GetPipelineReplicationStatusResponse,
             TableReplicationStatus,
@@ -236,7 +236,7 @@ pub async fn run(
         crate::routes::pipelines::get_pipeline_version,
         crate::routes::pipelines::get_pipeline_replication_status,
         crate::routes::pipelines::rollback_table_state,
-        crate::routes::pipelines::update_pipeline_image,
+        crate::routes::pipelines::update_pipeline_version,
         crate::routes::tenants::create_tenant,
         crate::routes::tenants::create_or_update_tenant,
         crate::routes::tenants::read_tenant,
@@ -321,7 +321,7 @@ pub async fn run(
                     .service(get_pipeline_version)
                     .service(get_pipeline_replication_status)
                     .service(rollback_table_state)
-                    .service(update_pipeline_image)
+                    .service(update_pipeline_version)
                     .service(update_pipeline_config)
                     //tables
                     .service(read_table_names)

etl-api/tests/pipelines.rs

@@ -3,7 +3,7 @@ use etl_api::routes::pipelines::{
     GetPipelineVersionResponse, ReadPipelineResponse, ReadPipelinesResponse,
     RollbackTableStateRequest, RollbackTableStateResponse, RollbackType,
     SimpleTableReplicationState, UpdatePipelineConfigRequest, UpdatePipelineConfigResponse,
-    UpdatePipelineImageRequest, UpdatePipelineRequest,
+    UpdatePipelineRequest, UpdatePipelineVersionRequest,
 };
 use etl_config::shared::{BatchConfig, PgConnectionConfig};
 use etl_postgres::sqlx::test_utils::drop_pg_database;
@@ -250,7 +250,7 @@ async fn pipeline_with_another_tenants_source_cannot_be_created() {
     let response = app.create_pipeline(tenant1_id, &pipeline).await;
 
     // Assert
-    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(response.status(), StatusCode::NOT_FOUND);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -283,7 +283,7 @@ async fn pipeline_with_another_tenants_destination_cannot_be_created() {
     let response = app.create_pipeline(tenant1_id, &pipeline).await;
 
     // Assert
-    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(response.status(), StatusCode::NOT_FOUND);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -432,7 +432,7 @@ async fn pipeline_with_another_tenants_source_cannot_be_updated() {
         .await;
 
     // Assert
-    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(response.status(), StatusCode::NOT_FOUND);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -480,7 +480,7 @@ async fn pipeline_with_another_tenants_destination_cannot_be_updated() {
         .await;
 
     // Assert
-    assert_eq!(response.status(), StatusCode::BAD_REQUEST);
+    assert_eq!(response.status(), StatusCode::NOT_FOUND);
 }
 
 #[tokio::test(flavor = "multi_thread")]
@@ -666,11 +666,11 @@ async fn updating_pipeline_to_duplicate_source_destination_combination_fails() {
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn pipeline_image_can_be_updated_with_specific_image() {
+async fn pipeline_version_can_be_updated() {
     init_test_tracing();
     // Arrange
     let app = spawn_test_app().await;
-    create_default_image(&app).await;
+    let default_image_id = create_default_image(&app).await;
     let tenant_id = &create_tenant(&app).await;
     let source_id = create_source(&app, tenant_id).await;
     let destination_id = create_destination(&app, tenant_id).await;
@@ -685,65 +685,36 @@ async fn pipeline_image_can_be_updated_with_specific_image() {
     .await;
 
     // Act
-    let update_request = UpdatePipelineImageRequest {
-        image_id: Some(1), // Use the default image ID
+    let update_request = UpdatePipelineVersionRequest {
+        version_id: default_image_id,
     };
     let response = app
-        .update_pipeline_image(tenant_id, pipeline_id, &update_request)
+        .update_pipeline_version(tenant_id, pipeline_id, &update_request)
         .await;
 
     // Assert
     assert!(response.status().is_success());
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn pipeline_image_can_be_updated_to_default_image() {
-    init_test_tracing();
-    // Arrange
-    let app = spawn_test_app().await;
-    create_default_image(&app).await;
-    let tenant_id = &create_tenant(&app).await;
-    let source_id = create_source(&app, tenant_id).await;
-    let destination_id = create_destination(&app, tenant_id).await;
-
-    let pipeline_id = create_pipeline_with_config(
-        &app,
-        tenant_id,
-        source_id,
-        destination_id,
-        new_pipeline_config(),
-    )
-    .await;
-
-    // Act - update to default image (no image_id specified)
-    let update_request = UpdatePipelineImageRequest { image_id: None };
-    let response = app
-        .update_pipeline_image(tenant_id, pipeline_id, &update_request)
-        .await;
-
-    // Assert
-    assert!(response.status().is_success());
-}
-
-#[tokio::test(flavor = "multi_thread")]
-async fn update_image_fails_for_non_existing_pipeline() {
+async fn update_version_fails_for_non_existing_pipeline() {
     init_test_tracing();
     // Arrange
     let app = spawn_test_app().await;
     let tenant_id = &create_tenant(&app).await;
 
     // Act
-    let update_request = UpdatePipelineImageRequest { image_id: None };
+    let update_request = UpdatePipelineVersionRequest { version_id: 1 };
     let response = app
-        .update_pipeline_image(tenant_id, 42, &update_request)
+        .update_pipeline_version(tenant_id, 42, &update_request)
         .await;
 
     // Assert
     assert_eq!(response.status(), StatusCode::NOT_FOUND);
 }
 
 #[tokio::test(flavor = "multi_thread")]
-async fn update_image_fails_for_non_existing_image() {
+async fn update_version_fails_when_version_is_not_default() {
     init_test_tracing();
     // Arrange
     let app = spawn_test_app().await;
@@ -761,63 +732,19 @@ async fn update_image_fails_for_non_existing_image() {
     )
     .await;
 
-    // Act
-    let update_request = UpdatePipelineImageRequest {
-        image_id: Some(999), // Non-existing image ID
-    };
-    let response = app
-        .update_pipeline_image(tenant_id, pipeline_id, &update_request)
-        .await;
+    // Create a non-default image
+    let non_default_image_id =
+        create_image_with_name(&app, "another/image".to_string(), false).await;
 
-    // Assert
-    assert_eq!(response.status(), StatusCode::NOT_FOUND);
-}
-
-#[tokio::test(flavor = "multi_thread")]
-async fn update_image_fails_for_pipeline_from_another_tenant() {
-    init_test_tracing();
-    // Arrange
-    let app = spawn_test_app().await;
-    create_default_image(&app).await;
-    let tenant1_id = &create_tenant(&app).await;
-
-    let source1_id = create_source(&app, tenant1_id).await;
-    let destination1_id = create_destination(&app, tenant1_id).await;
-
-    let pipeline_id = create_pipeline_with_config(
-        &app,
-        tenant1_id,
-        source1_id,
-        destination1_id,
-        new_pipeline_config(),
-    )
-    .await;
-
-    // Act - Try to update image using wrong tenant credentials
-    let update_request = UpdatePipelineImageRequest { image_id: None };
-    let response = app
-        .update_pipeline_image("wrong-tenant-id", pipeline_id, &update_request)
-        .await;
-
-    // Assert
-    assert_eq!(response.status(), StatusCode::NOT_FOUND);
-}
-
-#[tokio::test(flavor = "multi_thread")]
-async fn update_image_fails_when_no_default_image_exists() {
-    init_test_tracing();
-    // Arrange
-    let app = spawn_test_app().await;
-    // Don't create default image
-    let tenant_id = &create_tenant(&app).await;
-
-    // Act - Try to update to default image when none exists
-    let update_request = UpdatePipelineImageRequest { image_id: None };
+    // Act - attempt update with a non-default image id
+    let update_request = UpdatePipelineVersionRequest {
+        version_id: non_default_image_id,
+    };
     let response = app
-        .update_pipeline_image(tenant_id, 1, &update_request)
+        .update_pipeline_version(tenant_id, pipeline_id, &update_request)
         .await;
 
-    // Assert
+    // Assert - mismatching image id should be rejected
     assert_eq!(response.status(), StatusCode::NOT_FOUND);
 }
 
@@ -906,6 +833,7 @@ async fn update_config_fails_for_pipeline_from_another_tenant() {
     init_test_tracing();
     // Arrange
     let app = spawn_test_app().await;
+    create_default_image(&app).await;
     let tenant1_id = &create_tenant(&app).await;
 
     let source1_id = create_source(&app, tenant1_id).await;

etl-api/tests/support/mocks.rs

@@ -24,6 +24,7 @@ pub async fn create_image_with_name(app: &TestApp, name: String, is_default: boo
         .json()
         .await
         .expect("failed to deserialize response");
+
     response.id
 }
 
@@ -258,18 +259,19 @@ pub mod pipelines {
         destination_id: i64,
         config: FullApiPipelineConfig,
     ) -> i64 {
-        // Ensure there is a default image available.
-        super::create_default_image(app).await;
         let pipeline = CreatePipelineRequest {
             source_id,
             destination_id,
             config,
         };
         let response = app.create_pipeline(tenant_id, &pipeline).await;
+        assert!(response.status().is_success(), "failed to created pipeline");
+
         let response: CreatePipelineResponse = response
             .json()
             .await
             .expect("failed to deserialize response");
+
         response.id
     }
 }