StreamLineLabs · josedab · Aug 4, 2026 · Aug 4, 2026 · Aug 7, 2026 · Aug 7, 2026
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
@@ -15,7 +15,7 @@ jobs:
 
       - uses: dtolnay/rust-toolchain@stable
         with:
-          toolchain: "1.80"
+          toolchain: "1.88"
           components: rustfmt, clippy
 
       - run: cargo fmt --all -- --check

diff --git a/AUDIT.md b/AUDIT.md
@@ -0,0 +1,42 @@
+# Clean Code and SRP Audit
+
+## Summary
+
+- Controllers are already separated by Kubernetes resource actor; do not merge
+  their similar finalizer/status/requeue shapes into a generic controller.
+- **Highest-leverage future split:** isolate StatefulSet/Service/ConfigMap
+  rendering from cluster reconciliation once golden manifest tests cover every
+  option.
+- `main.rs` now correctly owns startup and shutdown of controller/health/metrics
+  tasks; further splitting would be small orchestration helpers, not new types.
+- CRD modules are public serialized contracts and remain intact.
+- Baseline formatting, strict Clippy, tests, integration gating, and auxiliary
+  task shutdown are green.
+
+## Findings
+
+| ID | Location | Category | Severity | Actors in conflict | Cost | Size | Behavior risk |
+|---|---|---|---|---|---|---|---|
+| OP-SRP-1 | `controllers/cluster.rs` | Mixed reconciliation/rendering | P2 | cluster product; Kubernetes workload API | Desired-state construction and reconciliation/status orchestration change for separate actors. | L | High |
+| OP-CC-1 | controller modules | Similar but actor-distinct code | P2 | cluster/topic/user/contract/branch/memory actors | Generic deduplication would couple independent CRD evolution and worsen SRP. | M | High |
+
+## Ordered Refactor Sequence
+
+1. Add golden tests for rendered cluster workloads across storage, auth, TLS,
+   replicas, resources, and service settings.
+2. Move rendering unchanged into `cluster_resources`.
+3. Keep finalizer/requeue/status orchestration in `ClusterController`.
+4. Do not create generic controller traits unless two CRDs share the same actor
+   and lifecycle contract.
+
+## Deferred
+
+- Cluster renderer extraction lacks complete golden coverage.
+- Live Kubernetes integration requires cluster credentials and a server image.
+- CRD changes require versioned schema/release decisions.
+
+## Out of Scope
+
+- Per-CRD controllers: actor-aligned.
+- CRD modules: serialization contracts.
+- Metrics and leader election: independent operational actors.
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
@@ -6,7 +6,7 @@ Thank you for your interest in contributing! Please review the [organization-wid
 
 ### Prerequisites
 
-- Rust 1.75+ (`rustup update stable`)
+- Rust 1.88+ (`rustup update stable`)
 - Docker (for integration tests)
 - kubectl + a Kubernetes cluster (for e2e tests, [kind](https://kind.sigs.k8s.io/) recommended)
 

diff --git a/Cargo.toml b/Cargo.toml
@@ -10,7 +10,7 @@ repository = "https://github.com/streamlinelabs/streamline-operator"
 readme = "README.md"
 keywords = ["kubernetes", "operator", "streamline", "kafka", "streaming"]
 categories = ["network-programming", "asynchronous"]
-rust-version = "1.80"
+rust-version = "1.88"
 
 [dependencies]
 # Kubernetes client and runtime
@@ -65,5 +65,3 @@ tokio-test = "0.4"
 [lints.clippy]
 unwrap_used = "warn"
 expect_used = "warn"
-
-
diff --git a/Makefile b/Makefile
@@ -1,14 +1,23 @@
-.PHONY: build test lint fmt clean help check
+.PHONY: build test lint fmt clean help check integration-up integration-down test-integration
 
 help: ## Show this help
 	@grep -E '^[a-zA-Z_-]+:.*?## .*$$' $(MAKEFILE_LIST) | sort | awk 'BEGIN {FS = ":.*?## "}; {printf "\033[36m%-15s\033[0m %s\n", $$1, $$2}'
 
 build: ## Build the operator
 	cargo build
 
-test: ## Run tests
+test: ## Run tests (hermetic — no Kubernetes or Streamline required)
 	cargo test
 
+integration-up: ## Start integration services (override with STREAMLINE_TEST_IMAGE)
+	docker compose -f docker-compose.test.yml up -d --wait
+
+integration-down: ## Stop and remove integration services
+	docker compose -f docker-compose.test.yml down -v
+
+test-integration: ## Run explicitly gated integration tests (needs integration-up)
+	cargo test --test integration -- --ignored --test-threads=1
+
 lint: ## Run clippy lints
 	cargo clippy --all-targets -- -D warnings
 

diff --git a/README.md b/README.md
@@ -3,7 +3,7 @@
 [![CI](https://github.com/streamlinelabs/streamline-operator/actions/workflows/ci.yml/badge.svg)](https://github.com/streamlinelabs/streamline-operator/actions/workflows/ci.yml)
 [![codecov](https://img.shields.io/codecov/c/github/streamlinelabs/streamline-operator?style=flat-square)](https://codecov.io/gh/streamlinelabs/streamline-operator)
 [![License](https://img.shields.io/badge/License-Apache_2.0-blue.svg)](https://opensource.org/licenses/Apache-2.0)
-[![Rust](https://img.shields.io/badge/Rust-1.80%2B-orange.svg)](https://www.rust-lang.org/)
+[![Rust](https://img.shields.io/badge/Rust-1.88%2B-orange.svg)](https://www.rust-lang.org/)
 [![Kubernetes](https://img.shields.io/badge/Kubernetes-1.26+-326CE5.svg)](https://kubernetes.io/)
 [![Docs](https://img.shields.io/badge/docs-streamlinelabs.dev-brightgreen)](https://streamlinelabs.dev/docs/operations/kubernetes)
 [![Release](https://img.shields.io/github/v/release/streamlinelabs/streamline-operator?label=release)](https://github.com/streamlinelabs/streamline-operator/releases)
@@ -289,7 +289,7 @@ spec:
 
 ### Prerequisites
 
-- Rust 1.80+
+- Rust 1.88+
 - Access to a Kubernetes cluster
 - `kubectl` configured
 
@@ -318,6 +318,36 @@ cargo run -p streamline-operator -- --metrics-port 8080 --health-port 8081
 cargo test
 ```
 
+The default test run is hermetic: it needs no Kubernetes cluster, no Streamline
+server, and no Docker.
+
+#### Integration tests (opt-in)
+
+Tests that need a live Streamline server live in `tests/integration.rs` and are
+`#[ignore]`d, so they never run as part of `cargo test`. Every networked
+assertion is bounded by a timeout so a missing backend fails fast.
+
+```bash
+make integration-up        # start the server from docker-compose.test.yml
+make test-integration      # cargo test --test integration -- --ignored
+make integration-down
+```
+
+The image and endpoints are configurable — see
+[`docs/ENVIRONMENT.md`](docs/ENVIRONMENT.md#integration-test-variables):
+
+```bash
+STREAMLINE_TEST_IMAGE=ghcr.io/streamlinelabs/streamline:0.3.0 \
+STREAMLINE_TEST_HTTP_PORT=19094 \
+STREAMLINE_TEST_KAFKA_PORT=19092 make integration-up
+
+STREAMLINE_TEST_HTTP_ENDPOINT=http://127.0.0.1:19094 \
+STREAMLINE_TEST_KAFKA_ENDPOINT=127.0.0.1:19092 make test-integration
+```
+
+A separate Helm/kubectl suite lives in `scripts/helm-integration-test.sh` and
+requires a configured `kubectl` context.
+
 ### Lint
 
 ```bash
@@ -403,4 +433,3 @@ See [CONTRIBUTING.md](CONTRIBUTING.md) for development setup and guidelines.
 
 Apache License 2.0 — see [LICENSE](LICENSE) for details.
 <!-- fix: 0cd8410e -->
-
diff --git a/docker-compose.test.yml b/docker-compose.test.yml
@@ -1,22 +1,34 @@
 # Docker Compose for operator integration testing
 #
-# Starts a Streamline server and a kind-compatible Kubernetes API.
-# The operator can be tested against these services locally.
+# Starts a Streamline server for the explicitly gated integration suite in
+# `tests/integration.rs`. Nothing in the default `cargo test` run depends on it.
 #
-# Usage: docker compose -f docker-compose.test.yml up -d
-version: "3.8"
+# Usage:
+#   make integration-up          # or: docker compose -f docker-compose.test.yml up -d --wait
+#   make test-integration        # cargo test --test integration -- --ignored
+#   make integration-down
+#
+# Configuration (all optional — defaults match tests/integration.rs):
+#   STREAMLINE_TEST_IMAGE        Server image         (default ghcr.io/streamlinelabs/streamline:0.2.0)
+#   STREAMLINE_TEST_KAFKA_PORT   Host Kafka port      (default 9092)
+#   STREAMLINE_TEST_HTTP_PORT    Host HTTP API port   (default 9094)
+#   STREAMLINE_TEST_LOG_LEVEL    Server log verbosity (default info)
+#
+# Example:
+#   STREAMLINE_TEST_IMAGE=ghcr.io/streamlinelabs/streamline:0.3.0 \
+#   STREAMLINE_TEST_HTTP_PORT=19094 make integration-up
 services:
   streamline:
-    image: ghcr.io/streamlinelabs/streamline:0.2.0
+    image: ${STREAMLINE_TEST_IMAGE:-ghcr.io/streamlinelabs/streamline:0.2.0}
     ports:
-      - "9092:9092"
-      - "9094:9094"
+      - "${STREAMLINE_TEST_KAFKA_PORT:-9092}:9092"
+      - "${STREAMLINE_TEST_HTTP_PORT:-9094}:9094"
     environment:
-      - STREAMLINE_LOG_LEVEL=info
+      - STREAMLINE_LOG_LEVEL=${STREAMLINE_TEST_LOG_LEVEL:-info}
       - STREAMLINE_PLAYGROUND=true
     healthcheck:
       test: ["CMD", "curl", "-f", "http://localhost:9094/health"]
       interval: 5s
       timeout: 5s
       retries: 10
-
+      start_period: 10s
diff --git a/docs/ENVIRONMENT.md b/docs/ENVIRONMENT.md
@@ -10,3 +10,27 @@ The Streamline Operator supports the following environment variables:
 | `STREAMLINE_OPERATOR_LEADER_ELECTION` | Enable leader election | `true` |
 | `STREAMLINE_DEFAULT_IMAGE` | Default Streamline image | `streamlinelabs/streamline:latest` |
 | `STREAMLINE_RECONCILE_INTERVAL` | Reconciliation interval | `30s` |
+
+## Integration Test Variables
+
+These apply only to the opt-in integration suite (`tests/integration.rs` and
+`docker-compose.test.yml`). They are never read by the operator at runtime, and
+the default `cargo test` run ignores them entirely.
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `STREAMLINE_TEST_IMAGE` | Streamline server image started by `docker-compose.test.yml` | `ghcr.io/streamlinelabs/streamline:0.2.0` |
+| `STREAMLINE_TEST_HTTP_PORT` | Host port mapped to the server HTTP API | `9094` |
+| `STREAMLINE_TEST_KAFKA_PORT` | Host port mapped to the server Kafka listener | `9092` |
+| `STREAMLINE_TEST_HTTP_ENDPOINT` | Full HTTP endpoint override for the tests | `http://127.0.0.1:$STREAMLINE_TEST_HTTP_PORT` |
+| `STREAMLINE_TEST_KAFKA_ENDPOINT` | Full `host:port` Kafka override for the tests | `127.0.0.1:$STREAMLINE_TEST_KAFKA_PORT` |
+| `STREAMLINE_TEST_TIMEOUT_SECS` | Upper bound on every networked assertion | `15` |
+| `STREAMLINE_TEST_LOG_LEVEL` | Log verbosity of the containerised server | `info` |
+
+Blank, zero, or unparseable values fall back to the defaults above.
+
+```bash
+make integration-up
+make test-integration
+make integration-down
+```
diff --git a/rustfmt.toml b/rustfmt.toml
@@ -3,4 +3,3 @@
 
 max_width = 100
 use_field_init_shorthand = true
-imports_granularity = "Crate"
diff --git a/src/conditions.rs b/src/conditions.rs
@@ -113,7 +113,12 @@ mod tests {
 
     #[test]
     fn test_build_condition() {
-        let cond = build_condition(CLUSTER_CONDITION_READY, CONDITION_TRUE, "AllReady", "All brokers ready");
+        let cond = build_condition(
+            CLUSTER_CONDITION_READY,
+            CONDITION_TRUE,
+            "AllReady",
+            "All brokers ready",
+        );
         assert_eq!(cond.condition_type, "Ready");
         assert_eq!(cond.status, "True");
         assert!(cond.last_transition_time.is_some());

diff --git a/src/controllers/autoscaling.rs b/src/controllers/autoscaling.rs
@@ -571,8 +571,7 @@ impl AutoScalingController {
                 if needed > recommended {
                     recommended = needed;
                     reasons.push(format!(
-                        "High throughput: {:.0} msg/s per broker. Target: {}",
-                        mps_per_broker, target_mps
+                        "High throughput: {mps_per_broker:.0} msg/s per broker. Target: {target_mps}"
                     ));
                     triggering.push("messages_per_second".to_string());
                 }
@@ -656,6 +655,8 @@ impl AutoScalingController {
 
 #[cfg(test)]
 mod tests {
+    // unwrap/expect are acceptable in tests; the crate-wide lint targets production code.
+    #![allow(clippy::unwrap_used, clippy::expect_used)]
     use super::*;
 
     #[test]

diff --git a/src/controllers/branch.rs b/src/controllers/branch.rs
@@ -2,12 +2,10 @@
 //! Manages branch lifecycle via the admin API.
 
 use crate::conditions::{
-    build_condition, set_condition, CONDITION_FALSE, CONDITION_TRUE, BRANCH_FINALIZER,
+    build_condition, set_condition, BRANCH_FINALIZER, CONDITION_FALSE, CONDITION_TRUE,
 };
 use crate::controllers::error_policy_backoff;
-use crate::crd::{
-    BranchCondition, BranchPhase, BranchStatus, StreamlineBranch, StreamlineCluster,
-};
+use crate::crd::{BranchCondition, BranchPhase, BranchStatus, StreamlineBranch, StreamlineCluster};
 use crate::error::{OperatorError, Result};
 use futures::StreamExt;
 use kube::api::{Api, Patch, PatchParams};
@@ -31,7 +29,10 @@ pub struct BranchController {
 impl BranchController {
     /// Create a new branch controller
     pub fn new(client: Client, http_client: reqwest::Client) -> Self {
-        Self { client, http_client }
+        Self {
+            client,
+            http_client,
+        }
     }
 
     /// Run the branch controller
@@ -147,11 +148,7 @@ impl BranchController {
     }
 
     /// Ensure the finalizer is present on the resource
-    async fn ensure_finalizer(
-        &self,
-        branch: &StreamlineBranch,
-        namespace: &str,
-    ) -> Result<()> {
+    async fn ensure_finalizer(&self, branch: &StreamlineBranch, namespace: &str) -> Result<()> {
         let finalizers = branch.metadata.finalizers.as_deref().unwrap_or_default();
         if finalizers.contains(&BRANCH_FINALIZER.to_string()) {
             return Ok(());
@@ -195,13 +192,10 @@ impl BranchController {
                 "http://{}-0.{}-headless.{}.svc:{}",
                 cluster_name, cluster_name, namespace, cluster.spec.http_port
             );
-            info!(
-                "Deleting branch {} from cluster at {}",
-                name, http_endpoint
-            );
+            info!("Deleting branch {} from cluster at {}", name, http_endpoint);
             if let Err(e) = self
                 .http_client
-                .delete(format!("{}/api/v1/branches/{}", http_endpoint, name))
+                .delete(format!("{http_endpoint}/api/v1/branches/{name}"))
                 .send()
                 .await
             {
@@ -232,11 +226,7 @@ impl BranchController {
             }
         });
         branches
-            .patch(
-                &name,
-                &PatchParams::default(),
-                &Patch::Merge(&patch),
-            )
+            .patch(&name, &PatchParams::default(), &Patch::Merge(&patch))
             .await
             .map_err(|e| OperatorError::KubeApi(e.to_string()))?;
 
@@ -267,7 +257,7 @@ impl BranchController {
 
         let response = self
             .http_client
-            .post(format!("{}/api/v1/branches", http_endpoint))
+            .post(format!("{http_endpoint}/api/v1/branches"))
             .json(&body)
             .send()
             .await
@@ -375,12 +365,7 @@ impl BranchController {
         let mut cond_fields = Vec::new();
         set_condition(
             &mut cond_fields,
-            build_condition(
-                BRANCH_CONDITION_ACTIVE,
-                CONDITION_FALSE,
-                "Pending",
-                message,
-            ),
+            build_condition(BRANCH_CONDITION_ACTIVE, CONDITION_FALSE, "Pending", message),
         );
 
         let conditions: Vec<BranchCondition> = cond_fields
Original file line number	Diff line number	Diff line change
Expand Up		@@ -3,4 +3,3 @@

		max_width = 100
		use_field_init_shorthand = true
		imports_granularity = "Crate"