refactor: move product image resolution from dereference to validate

Image resolution is a pure computation, not an I/O dereference, so it
belongs in validate_cluster alongside the other config validation. This
also aligns with the pattern used by the trino operator.

The dereference error variants were renamed to drop the `Invalid` prefix
(e.g. InvalidAuthenticationConfig → AuthenticationConfig) because
removing ResolveProductImage left all remaining variants sharing the
same prefix, triggering clippy::enum_variant_names.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: adwk67

rust/operator-binary/src/airflow_controller.rs

@@ -325,15 +325,9 @@ pub async fn reconcile_airflow(
 
     let client = &ctx.client;
 
-    let dereferenced = crate::controller::dereference::dereference(
-        client,
-        airflow,
-        CONTAINER_IMAGE_BASE_NAME,
-        &ctx.operator_environment.image_repository,
-        crate::built_info::PKG_VERSION,
-    )
-    .await
-    .context(DereferenceSnafu)?;
+    let dereferenced = crate::controller::dereference::dereference(client, airflow)
+        .await
+        .context(DereferenceSnafu)?;
 
     let cluster_operation_cond_builder =
         ClusterOperationsConditionBuilder::new(&airflow.spec.cluster_operation);
@@ -375,9 +369,14 @@ pub async fn reconcile_airflow(
         None
     };
 
-    let validated =
-        crate::controller::validate::validate_cluster(airflow, &dereferenced, &ctx.product_config)
-            .context(ValidateSnafu)?;
+    let validated = crate::controller::validate::validate_cluster(
+        airflow,
+        CONTAINER_IMAGE_BASE_NAME,
+        &ctx.operator_environment.image_repository,
+        crate::built_info::PKG_VERSION,
+        &ctx.product_config,
+    )
+    .context(ValidateSnafu)?;
 
     let mut cluster_resources = ClusterResources::new(
         APP_NAME,

rust/operator-binary/src/controller/dereference.rs

@@ -1,8 +1,5 @@
 use snafu::{ResultExt, Snafu};
-use stackable_operator::commons::{
-    product_image_selection::{self, ResolvedProductImage},
-    random_secret_creation,
-};
+use stackable_operator::commons::random_secret_creation;
 
 use crate::crd::{
     authentication::AirflowClientAuthenticationDetailsResolved,
@@ -13,61 +10,46 @@ use crate::crd::{
 
 #[derive(Snafu, Debug)]
 pub enum Error {
-    #[snafu(display("failed to resolve product image"))]
-    ResolveProductImage {
-        source: product_image_selection::Error,
-    },
-
     #[snafu(display("failed to apply authentication configuration"))]
-    InvalidAuthenticationConfig {
+    AuthenticationConfig {
         source: crate::crd::authentication::Error,
     },
 
     #[snafu(display("invalid authorization config"))]
-    InvalidAuthorizationConfig {
+    AuthorizationConfig {
         source: stackable_operator::commons::opa::Error,
     },
 
     #[snafu(display("failed to create internal secret"))]
-    InvalidInternalSecret {
+    InternalSecret {
         source: random_secret_creation::Error,
     },
 }
 
 /// External references resolved during the dereference step.
 pub struct DereferencedObjects {
-    pub resolved_product_image: ResolvedProductImage,
     pub authentication_config: AirflowClientAuthenticationDetailsResolved,
     pub authorization_config: AirflowAuthorizationResolved,
 }
 
 pub async fn dereference(
     client: &stackable_operator::client::Client,
     airflow: &v1alpha2::AirflowCluster,
-    image_base_name: &str,
-    image_repository: &str,
-    pkg_version: &str,
 ) -> Result<DereferencedObjects, Error> {
-    let resolved_product_image = airflow
-        .spec
-        .image
-        .resolve(image_base_name, image_repository, pkg_version)
-        .context(ResolveProductImageSnafu)?;
-
     let authentication_config = AirflowClientAuthenticationDetailsResolved::from(
         &airflow.spec.cluster_config.authentication,
         client,
     )
     .await
-    .context(InvalidAuthenticationConfigSnafu)?;
+    .context(AuthenticationConfigSnafu)?;
 
     let authorization_config = AirflowAuthorizationResolved::from_authorization_config(
         client,
         airflow,
         &airflow.spec.cluster_config.authorization,
     )
     .await
-    .context(InvalidAuthorizationConfigSnafu)?;
+    .context(AuthorizationConfigSnafu)?;
 
     random_secret_creation::create_random_secret_if_not_exists(
         &airflow.shared_internal_secret_secret_name(),
@@ -77,7 +59,7 @@ pub async fn dereference(
         client,
     )
     .await
-    .context(InvalidInternalSecretSnafu)?;
+    .context(InternalSecretSnafu)?;
 
     random_secret_creation::create_random_secret_if_not_exists(
         &airflow.shared_jwt_secret_secret_name(),
@@ -87,7 +69,7 @@ pub async fn dereference(
         client,
     )
     .await
-    .context(InvalidInternalSecretSnafu)?;
+    .context(InternalSecretSnafu)?;
 
     // https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/fernet.html#security-fernet
     // does not document how long the fernet key should be, but recommends using
@@ -101,10 +83,9 @@ pub async fn dereference(
         client,
     )
     .await
-    .context(InvalidInternalSecretSnafu)?;
+    .context(InternalSecretSnafu)?;
 
     Ok(DereferencedObjects {
-        resolved_product_image,
         authentication_config,
         authorization_config,
     })

rust/operator-binary/src/controller/validate.rs

@@ -6,17 +6,21 @@ use std::{
 use product_config::{ProductConfigManager, types::PropertyNameKind};
 use snafu::{ResultExt, Snafu};
 use stackable_operator::{
-    commons::product_image_selection::ResolvedProductImage,
+    commons::product_image_selection::{self, ResolvedProductImage},
     product_config_utils::{transform_all_roles_to_config, validate_all_roles_and_groups_config},
     role_utils::RoleGroupRef,
 };
 use strum::IntoEnumIterator;
 
-use super::dereference::DereferencedObjects;
 use crate::crd::{AIRFLOW_CONFIG_FILENAME, AirflowConfig, AirflowExecutor, AirflowRole, v1alpha2};
 
 #[derive(Snafu, Debug)]
 pub enum Error {
+    #[snafu(display("failed to resolve product image"))]
+    ResolveProductImage {
+        source: product_image_selection::Error,
+    },
+
     #[snafu(display("invalid product config"))]
     InvalidProductConfig {
         source: stackable_operator::product_config_utils::Error,
@@ -64,9 +68,17 @@ pub struct ValidatedAirflowCluster {
 
 pub fn validate_cluster(
     airflow: &v1alpha2::AirflowCluster,
-    dereferenced: &DereferencedObjects,
+    image_base_name: &str,
+    image_repository: &str,
+    pkg_version: &str,
     product_config_manager: &ProductConfigManager,
 ) -> Result<ValidatedAirflowCluster, Error> {
+    let resolved_product_image = airflow
+        .spec
+        .image
+        .resolve(image_base_name, image_repository, pkg_version)
+        .context(ResolveProductImageSnafu)?;
+
     let mut roles = HashMap::new();
 
     // if the kubernetes executor is specified there will be no worker role as the pods
@@ -88,7 +100,7 @@ pub fn validate_cluster(
 
     let role_config = transform_all_roles_to_config(airflow, &roles);
     let validated_role_config = validate_all_roles_and_groups_config(
-        &dereferenced.resolved_product_image.product_version,
+        &resolved_product_image.product_version,
         &role_config.context(GenerateProductConfigSnafu)?,
         product_config_manager,
         false,
@@ -143,7 +155,7 @@ pub fn validate_cluster(
     }
 
     Ok(ValidatedAirflowCluster {
-        image: dereferenced.resolved_product_image.clone(),
+        image: resolved_product_image,
         role_groups,
         role_configs,
         executor: airflow.spec.executor.clone(),