feldera · gz · Aug 5, 2026 · Aug 2, 2026 · Aug 2, 2026 · Aug 2, 2026
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/crates/pipeline-manager/Cargo.toml b/crates/pipeline-manager/Cargo.toml
@@ -5,7 +5,7 @@ keywords = ["DBSP", "streaming", "analytics", "database", "webui"]
 categories = ["database", "gui"]
 publish = false
 
-edition = { workspace = true }
+edition = "2024"
 version = { workspace = true }
 homepage = { workspace = true }
 repository = { workspace = true }
@@ -151,6 +151,7 @@ wiremock = { workspace = true }
 feldera-types = { workspace = true, features = ["testing"] }
 itertools = { workspace = true }
 tokio = { workspace = true, features = ["sync"] }
+libc = { workspace = true }  # For: asserting errno-to-ErrorKind mapping
 
 [package.metadata.cargo-machete]
 ignored = ["static-files", "compare", "tikv-jemallocator"]
diff --git a/crates/pipeline-manager/build.rs b/crates/pipeline-manager/build.rs
@@ -1,5 +1,5 @@
 use change_detection::ChangeDetection;
-use static_files::{resource_dir, NpmBuild};
+use static_files::{NpmBuild, resource_dir};
 use std::env;
 use std::path::{Path, PathBuf};
 use vergen_gitcl::*;

diff --git a/crates/pipeline-manager/src/api/demo.rs b/crates/pipeline-manager/src/api/demo.rs
@@ -250,7 +250,7 @@ pub fn read_demos_from_directories(demos_dir: &Vec<String>) -> Vec<Demo> {
 
 #[cfg(test)]
 mod test {
-    use super::{parse_demo, read_demos_from_directories, Demo, DemoError};
+    use super::{Demo, DemoError, parse_demo, read_demos_from_directories};
     use std::fs;
     use std::fs::File;
     use std::io::Write;

diff --git a/crates/pipeline-manager/src/api/endpoints/api_key.rs b/crates/pipeline-manager/src/api/endpoints/api_key.rs
@@ -9,11 +9,10 @@ use crate::db::types::tenant::TenantId;
 use crate::error::ManagerError;
 use crate::{api::examples, db::storage::Storage};
 use actix_web::{
-    delete, get,
+    HttpRequest, HttpResponse, delete, get,
     http::header::{CacheControl, CacheDirective},
     post,
     web::{self, Data as WebData, ReqData},
-    HttpRequest, HttpResponse,
 };
 use serde::{Deserialize, Serialize};
 use tracing::info;

diff --git a/crates/pipeline-manager/src/api/endpoints/cluster.rs b/crates/pipeline-manager/src/api/endpoints/cluster.rs
@@ -7,7 +7,7 @@ use crate::{
     error::ManagerError,
 };
 use actix_web::http::header::{CacheControl, CacheDirective};
-use actix_web::{get, web, web::Data as WebData, HttpResponse};
+use actix_web::{HttpResponse, get, web, web::Data as WebData};
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use std::str::FromStr;

diff --git a/crates/pipeline-manager/src/api/endpoints/config.rs b/crates/pipeline-manager/src/api/endpoints/config.rs
@@ -1,8 +1,7 @@
 // Configuration API to retrieve the current authentication configuration and list of demos
 use actix_web::{
-    get,
+    HttpRequest, HttpResponse, get,
     web::{Data as WebData, ReqData},
-    HttpRequest, HttpResponse,
 };
 use feldera_cloud1_client::license::DisplaySchedule;
 use serde::Serialize;

diff --git a/crates/pipeline-manager/src/api/endpoints/metrics.rs b/crates/pipeline-manager/src/api/endpoints/metrics.rs
@@ -2,10 +2,9 @@ use crate::api::main::ServerState;
 use crate::db::{storage::Storage as _, types::tenant::TenantId};
 use crate::error::ManagerError;
 use actix_web::{
-    get,
+    HttpResponse, get,
     http::Method,
     web::{Data as WebData, ReqData},
-    HttpResponse,
 };
 use awc::body::MessageBody as _;
 use feldera_types::runtime_status::RuntimeStatus;
@@ -45,10 +44,9 @@ pub(crate) async fn get_metrics(
     let mut result = Vec::new();
 
     for pipeline in pipelines {
-        if pipeline.deployment_runtime_status == Some(RuntimeStatus::Running)
-            || pipeline.deployment_runtime_status == Some(RuntimeStatus::Paused)
-        {
-            if let Ok(res) = state
+        if (pipeline.deployment_runtime_status == Some(RuntimeStatus::Running)
+            || pipeline.deployment_runtime_status == Some(RuntimeStatus::Paused))
+            && let Ok(res) = state
                 .runner
                 .forward_http_request_to_pipeline_by_name(
                     client.as_ref(),
@@ -61,14 +59,11 @@ pub(crate) async fn get_metrics(
                     None,
                 )
                 .await
-            {
-                if res.status().is_success() {
-                    if let Ok(bytes) = res.into_body().try_into_bytes() {
-                        result.extend(bytes);
-                        result.push(NEWLINE);
-                    }
-                }
-            }
+            && res.status().is_success()
+            && let Ok(bytes) = res.into_body().try_into_bytes()
+        {
+            result.extend(bytes);
+            result.push(NEWLINE);
         }
     }
 

diff --git a/crates/pipeline-manager/src/api/endpoints/oidc_trust.rs b/crates/pipeline-manager/src/api/endpoints/oidc_trust.rs
@@ -19,11 +19,10 @@ use crate::db::types::role::{MemberRole, Role};
 use crate::db::types::tenant::TenantId;
 use crate::error::ManagerError;
 use actix_web::{
-    delete, get,
+    HttpRequest, HttpResponse, delete, get,
     http::header::{CacheControl, CacheDirective},
     post,
     web::{self, Data as WebData, ReqData},
-    HttpRequest, HttpResponse,
 };
 use serde::{Deserialize, Serialize};
 use tracing::info;

diff --git a/crates/pipeline-manager/src/api/endpoints/pipeline_interaction.rs b/crates/pipeline-manager/src/api/endpoints/pipeline_interaction.rs
@@ -9,11 +9,10 @@ use crate::db::types::tenant::TenantId;
 use crate::error::ManagerError;
 use actix_http::StatusCode;
 use actix_web::{
-    get,
-    http::{header, Method},
+    HttpRequest, HttpResponse, get,
+    http::{Method, header},
     post,
     web::{self, Data as WebData, ReqData},
-    HttpRequest, HttpResponse,
 };
 #[allow(unused_imports)]
 use feldera_types::checkpoint::RemoteCheckpoint;

diff --git a/crates/pipeline-manager/src/api/endpoints/pipeline_interaction/support_bundle.rs b/crates/pipeline-manager/src/api/endpoints/pipeline_interaction/support_bundle.rs
@@ -9,14 +9,13 @@ use crate::api::main::ServerState;
 use crate::api::support_data_collector::{
     CollectionSummary, SupportBundleData, SupportBundleParameters,
 };
-use crate::db::types::combined_status::{combine_since, CombinedDesiredStatus, CombinedStatus};
+use crate::db::types::combined_status::{CombinedDesiredStatus, CombinedStatus, combine_since};
 use crate::db::types::pipeline::ExtendedPipelineDescrMonitoring;
 use crate::db::{storage::Storage, types::tenant::TenantId};
 use crate::error::ManagerError;
 use actix_web::{
-    get,
+    HttpResponse, get,
     web::{self, Data as WebData, ReqData},
-    HttpResponse,
 };
 use chrono::{DateTime, Utc};
 use serde::Serialize;
@@ -459,11 +458,13 @@ mod tests {
         assert_eq!(collections.len(), 2);
         for c in collections {
             assert!(dirs.contains(c["directory"].as_str().unwrap()));
-            assert!(c["collected"]
-                .as_array()
-                .unwrap()
-                .iter()
-                .any(|f| f == "pipeline_events.json"));
+            assert!(
+                c["collected"]
+                    .as_array()
+                    .unwrap()
+                    .iter()
+                    .any(|f| f == "pipeline_events.json")
+            );
         }
     }
 

diff --git a/crates/pipeline-manager/src/api/endpoints/pipeline_management.rs b/crates/pipeline-manager/src/api/endpoints/pipeline_management.rs
@@ -7,7 +7,7 @@ use crate::compiler::{ProgramValidationRequest, ValidateProgramResponse};
 use crate::config::CommonConfig;
 use crate::db::error::DBError;
 use crate::db::storage::Storage;
-use crate::db::types::combined_status::{combine_since, CombinedDesiredStatus, CombinedStatus};
+use crate::db::types::combined_status::{CombinedDesiredStatus, CombinedStatus, combine_since};
 use crate::db::types::pipeline::{
     ClientMetadata, ExtendedPipelineDescr, ExtendedPipelineDescrMonitoring, PatchClientMetadata,
     PipelineDescr, PipelineId,
@@ -24,17 +24,16 @@ use crate::has_unstable_feature;
 #[cfg(feature = "feldera-enterprise")]
 use actix_web::http::Method;
 use actix_web::{
-    delete, get,
+    HttpResponse, delete, get,
     http::header::{CacheControl, CacheDirective},
     patch, post, put,
     web::{self, Data as WebData, ReqData},
-    HttpResponse,
 };
 use chrono::{DateTime, Utc};
 use feldera_types::adapter_stats::PipelineStatsErrorsResponse;
 use feldera_types::config::{InputEndpointConfig, OutputEndpointConfig, RuntimeConfig};
 use feldera_types::error::ErrorResponse;
-use feldera_types::pipeline_diff::{compute_pipeline_diff, PipelineDiff};
+use feldera_types::pipeline_diff::{PipelineDiff, compute_pipeline_diff};
 use feldera_types::program_schema::ProgramSchema;
 use feldera_types::runtime_status::{
     BootstrapConfig, BootstrapPolicy, ConnectorStats, RuntimeDesiredStatus, RuntimeStatus,
@@ -878,12 +877,11 @@ async fn fetch_connector_error_stats(
     let details = backward_compatible_runtime_status_details(
         pipeline.deployment_runtime_status_details.clone(),
     );
-    if let Some(value) = details {
-        if let Ok(details) = serde_json::from_value::<RuntimeStatusDetails>(value) {
-            if let Some(connector_stats) = details.connector_stats {
-                return Some(connector_stats);
-            }
-        }
+    if let Some(value) = details
+        && let Ok(details) = serde_json::from_value::<RuntimeStatusDetails>(value)
+        && let Some(connector_stats) = details.connector_stats
+    {
+        return Some(connector_stats);
     };
 
     // Only forward the request if the pipeline is in a valid runtime status

diff --git a/crates/pipeline-manager/src/api/endpoints/pipeline_management/pipeline_events.rs b/crates/pipeline-manager/src/api/endpoints/pipeline_management/pipeline_events.rs
@@ -10,10 +10,9 @@ use crate::db::types::storage::StorageStatus;
 use crate::db::types::tenant::TenantId;
 use crate::error::ManagerError;
 use actix_web::{
-    get,
+    HttpResponse, get,
     http::header::{CacheControl, CacheDirective},
     web::{self, Data as WebData, ReqData},
-    HttpResponse,
 };
 use chrono::{DateTime, Utc};
 use feldera_types::error::ErrorResponse;

diff --git a/crates/pipeline-manager/src/api/endpoints/tenant.rs b/crates/pipeline-manager/src/api/endpoints/tenant.rs
@@ -15,11 +15,10 @@ use crate::db::types::tenant::TenantId;
 use crate::db::types::user::{TenantInfo, UserId};
 use crate::error::ManagerError;
 use actix_web::{
-    delete, get,
+    HttpRequest, HttpResponse, delete, get,
     http::header::{CacheControl, CacheDirective},
     patch, post, put,
     web::{self, Data as WebData, ReqData},
-    HttpRequest, HttpResponse,
 };
 use serde::{Deserialize, Serialize};
 use tracing::info;

diff --git a/crates/pipeline-manager/src/api/error.rs b/crates/pipeline-manager/src/api/error.rs
@@ -1,5 +1,5 @@
 use actix_web::{
-    body::BoxBody, http::StatusCode, HttpResponse, HttpResponseBuilder, ResponseError,
+    HttpResponse, HttpResponseBuilder, ResponseError, body::BoxBody, http::StatusCode,
 };
 use feldera_types::error::{DetailedError, ErrorResponse};
 use serde::Serialize;

diff --git a/crates/pipeline-manager/src/api/examples.rs b/crates/pipeline-manager/src/api/examples.rs
@@ -14,13 +14,13 @@ use crate::db::types::program::{CompilationProfile, ProgramConfig, ProgramError,
 use crate::db::types::resources_status::{ResourcesDesiredStatus, ResourcesStatus};
 use crate::db::types::storage::StorageStatus;
 use crate::db::types::utils::{
-    validate_program_config, validate_program_info, validate_runtime_config,
     PATTERN_KUBERNETES_LABEL_VALUE, PATTERN_KUBERNETES_LABEL_VALUE_DESCRIPTION,
+    validate_program_config, validate_program_info, validate_runtime_config,
 };
 use crate::db::types::version::Version;
 use crate::runner::error::RunnerError;
 use crate::runner::interaction::{
-    format_disconnected_error_message, format_timeout_error_message, RunnerInteraction,
+    RunnerInteraction, format_disconnected_error_message, format_timeout_error_message,
 };
 use feldera_types::config::{DevTweaks, FtConfig, ResourceConfig, StorageOptions};
 use feldera_types::runtime_status::{RuntimeStatusDetails, StorageStatusDetails};

diff --git a/crates/pipeline-manager/src/api/main.rs b/crates/pipeline-manager/src/api/main.rs
@@ -1,4 +1,4 @@
-use crate::api::demo::{read_demos_from_directories, Demo};
+use crate::api::demo::{Demo, read_demos_from_directories};
 use crate::api::endpoints;
 use crate::api::support_data_collector::SupportDataCollector;
 use crate::auth::{IssuerJwkCache, JwkCache};
@@ -9,17 +9,16 @@ use crate::error::ManagerError;
 use crate::license::LicenseCheck;
 use crate::runner::interaction::RunnerInteraction;
 use crate::unstable_features;
-use actix_http::body::BoxBody;
 use actix_http::StatusCode;
+use actix_http::body::BoxBody;
+use actix_web::Scope;
 use actix_web::body::MessageBody;
 use actix_web::dev::{Service, ServiceResponse};
-use actix_web::http::{header, Method};
-use actix_web::Scope;
+use actix_web::http::{Method, header};
 use actix_web::{
-    get, middleware,
+    App, HttpResponse, HttpServer, get, middleware,
     web::Data as WebData,
     web::{self},
-    App, HttpResponse, HttpServer,
 };
 use actix_web_httpauth::middleware::HttpAuthentication;
 use actix_web_static_files::ResourceFiles;
@@ -29,11 +28,11 @@ use futures_util::FutureExt;
 use std::io::Write;
 use std::time::Duration;
 use std::{env, io, net::TcpListener, sync::Arc};
-use termbg::{theme, Theme};
+use termbg::{Theme, theme};
 use tokio::signal;
 use tokio::sync::watch;
 use tokio::sync::{Mutex, RwLock};
-use tracing::{error, info, Level};
+use tracing::{Level, error, info};
 use utoipa::openapi::security::{HttpAuthScheme, HttpBuilder, SecurityScheme};
 use utoipa::{Modify, OpenApi};
 use utoipa_swagger_ui::SwaggerUi;
@@ -608,13 +607,16 @@ fn build_app(
     api_config: &ApiServerConfig,
     auth_configuration: &Option<crate::auth::AuthConfiguration>,
 ) -> App<
+    // `use<>`: the returned app owns everything it needs, so it must not
+    // capture the argument lifetimes that edition 2024 would capture by
+    // default. Callers pass short-lived config to an app that outlives it.
     impl actix_web::dev::ServiceFactory<
         actix_web::dev::ServiceRequest,
         Config = (),
-        Response = actix_web::dev::ServiceResponse<impl MessageBody>,
+        Response = actix_web::dev::ServiceResponse<impl MessageBody + use<>>,
         Error = actix_web::Error,
         InitError = (),
-    >,
+    > + use<>,
 > {
     let cors = api_config.cors();
     let app = App::new()
@@ -1119,7 +1121,7 @@ Version: {} v{}{}
     });
     #[cfg(unix)]
     {
-        use tokio::signal::unix::{signal as unix_signal, SignalKind};
+        use tokio::signal::unix::{SignalKind, signal as unix_signal};
         let mut term_stream = unix_signal(SignalKind::terminate()).expect("SIGTERM handler");
         let server_handle_term = server.handle();
         tokio::spawn(async move {
@@ -1221,10 +1223,11 @@ mod tests {
             "public, max-age=86400",
         );
         // Only the immutable branch emits ACAO/EXPIRES.
-        assert!(res
-            .headers()
-            .get(header::ACCESS_CONTROL_ALLOW_ORIGIN)
-            .is_none());
+        assert!(
+            res.headers()
+                .get(header::ACCESS_CONTROL_ALLOW_ORIGIN)
+                .is_none()
+        );
         assert!(res.headers().get(header::EXPIRES).is_none());
     }
 
@@ -1288,7 +1291,9 @@ mod tests {
         let res = test::call_service(&app, req).await;
 
         assert!(
-            res.headers().get(header::ACCESS_CONTROL_ALLOW_CREDENTIALS).is_none(),
+            res.headers()
+                .get(header::ACCESS_CONTROL_ALLOW_CREDENTIALS)
+                .is_none(),
             "static asset leaked Access-Control-Allow-Credentials — actix-cors regressed onto the static scope",
         );
         if let Some(vary) = res.headers().get(header::VARY) {