[Rust] Pin rustfmt for deterministic formatting

.github/scripts/rustfmt.sh

@@ -0,0 +1,6 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+readonly RUSTFMT_TOOLCHAIN="nightly-2025-08-07"
+
+exec cargo +"${RUSTFMT_TOOLCHAIN}" fmt "$@"

.github/workflows/ci-dotnet.yml

@@ -30,8 +30,9 @@ jobs:
             dotnet tool install -g dotnet-format
             echo "$HOME/.dotnet/tools" >> $GITHUB_PATH
 
-      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # nightly
         with:
+          toolchain: nightly-2025-08-07
           components: rustfmt
 
       - name: Format all files

.github/workflows/ci-go.yml

@@ -40,8 +40,9 @@ jobs:
         with:
           go-version: '1.25'
 
-      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # nightly
         with:
+          toolchain: nightly-2025-08-07
           components: rustfmt
 
       - name: Format all files
@@ -222,4 +223,4 @@ jobs:
 
       - name: Test Proto batch example builds
         working-directory: go/examples/proto/batch
-        run: go build -v
+        run: go build -v

.github/workflows/ci-python.yml

@@ -101,8 +101,9 @@ jobs:
         with:
           python-version: '3.12'
 
-      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # nightly
         with:
+          toolchain: nightly-2025-08-07
           components: rustfmt
 
       - name: Copy root LICENSE for maturin
@@ -140,4 +141,4 @@ jobs:
         run: cp ../LICENSE LICENSE
 
       - name: Lint code
-        run: make dev lint
+        run: make dev lint

.github/workflows/ci-rust.yml

@@ -19,8 +19,9 @@ jobs:
       - name: Configure Cargo registry
         shell: bash
         run: bash "$GITHUB_WORKSPACE/.github/scripts/configure-cargo.sh"
-      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+      - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # nightly
         with:
+          toolchain: nightly-2025-08-07
           components: rustfmt
       - name: Format all files
         run: make fmt
@@ -80,4 +81,4 @@ jobs:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
       - uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
       - name: Run tests
-        run: cargo test --workspace
+        run: cargo test --workspace

.rustfmt.toml

@@ -0,0 +1,13 @@
+# cargo fmt reads each crate's edition from Cargo.toml; this fallback keeps
+# direct rustfmt usable for the repository's Rust 2021 crates.
+edition = "2021"
+style_edition = "2021"
+newline_style = "Unix"
+group_imports = "StdExternalCrate"
+imports_layout = "HorizontalVertical"
+imports_granularity = "Module"
+
+max_width = 100
+fn_call_width = 100
+attr_fn_like_width = 100
+array_width = 100

dotnet/Makefile

@@ -50,7 +50,7 @@ fmt-dotnet:
 
 fmt-rust:
 	@echo "Formatting Rust code..."
-	cd ../rust/ffi && cargo fmt --all
+	cd ../rust/ffi && ../../.github/scripts/rustfmt.sh --all
 
 lint: lint-dotnet lint-rust
 

go/Makefile

@@ -85,7 +85,7 @@ fmt-go:
 
 fmt-rust:
 	@echo "Formatting Rust code..."
-	cd ../rust/ffi && cargo fmt --all
+	cd ../rust/ffi && ../../.github/scripts/rustfmt.sh --all
 
 lint: lint-go lint-rust
 

python/Makefile

@@ -72,7 +72,7 @@ fmt:
 	$(VENV) -m black zerobus examples tests
 	$(VENV) -m autoflake -ri --exclude '*_pb2*.py' zerobus examples tests
 	$(VENV) -m isort zerobus examples tests
-	cd rust && cargo fmt --all
+	cd rust && ../../.github/scripts/rustfmt.sh --all
 
 lint:
 	$(VENV) -m pycodestyle --exclude='*_pb2*.py' --max-line-length=120 --ignore=E203,W503 zerobus

python/rust/src/arrow.rs

@@ -8,14 +8,15 @@
 
 use std::sync::Arc;
 
-use pyo3::prelude::*;
-use pyo3::types::PyBytes;
-use tokio::sync::RwLock;
-
 use databricks_zerobus_ingest_sdk::{
-    StreamBuilder, ZerobusArrowStream as RustZerobusArrowStream, ZerobusError as RustError,
+    StreamBuilder,
+    ZerobusArrowStream as RustZerobusArrowStream,
+    ZerobusError as RustError,
     ZerobusSdk as RustSdk,
 };
+use pyo3::prelude::*;
+use pyo3::types::PyBytes;
+use tokio::sync::RwLock;
 
 use crate::auth::HeadersProviderWrapper;
 use crate::common::map_error;
@@ -249,10 +250,7 @@ impl ArrowStreamConfigurationOptions {
             ("recovery_timeout_ms", self.recovery_timeout_ms),
             ("recovery_backoff_ms", self.recovery_backoff_ms),
             ("recovery_retries", self.recovery_retries as i64),
-            (
-                "server_lack_of_ack_timeout_ms",
-                self.server_lack_of_ack_timeout_ms,
-            ),
+            ("server_lack_of_ack_timeout_ms", self.server_lack_of_ack_timeout_ms),
             ("flush_timeout_ms", self.flush_timeout_ms),
             ("connection_timeout_ms", self.connection_timeout_ms),
         ] {

python/rust/src/async_wrapper.rs

@@ -2,20 +2,27 @@ use std::future::Future;
 use std::pin::Pin;
 use std::sync::{Arc, Mutex};
 
+use databricks_zerobus_ingest_sdk::{
+    StreamBuilder,
+    ZerobusSdk as RustSdk,
+    ZerobusStream as RustStream,
+};
 use pyo3::prelude::*;
 use pyo3_asyncio::tokio::future_into_py;
 use tokio::sync::RwLock;
 
-use databricks_zerobus_ingest_sdk::{
-    StreamBuilder, ZerobusSdk as RustSdk, ZerobusStream as RustStream,
-};
-
 use crate::arrow;
 use crate::arrow::{ArrowStreamConfigurationOptions, AsyncZerobusArrowStream};
 use crate::auth::HeadersProviderWrapper;
 use crate::common::{
-    apply_grpc_options, encoded_record_to_pybytes, extract_record_payload, extract_record_payloads,
-    map_error, StreamConfigurationOptions, TableProperties, SDK_IDENTIFIER_PREFIX,
+    apply_grpc_options,
+    encoded_record_to_pybytes,
+    extract_record_payload,
+    extract_record_payloads,
+    map_error,
+    StreamConfigurationOptions,
+    TableProperties,
+    SDK_IDENTIFIER_PREFIX,
 };
 
 // =============================================================================

python/rust/src/auth.rs

@@ -1,11 +1,13 @@
-use async_trait::async_trait;
-use pyo3::exceptions::PyNotImplementedError;
-use pyo3::prelude::*;
 use std::collections::HashMap;
 
+use async_trait::async_trait;
 use databricks_zerobus_ingest_sdk::{
-    HeadersProvider as RustHeadersProvider, ZerobusError as RustError, ZerobusResult as RustResult,
+    HeadersProvider as RustHeadersProvider,
+    ZerobusError as RustError,
+    ZerobusResult as RustResult,
 };
+use pyo3::exceptions::PyNotImplementedError;
+use pyo3::prelude::*;
 
 use crate::common::intern_header_name;
 
@@ -41,9 +43,7 @@ impl HeadersProvider {
     /// Returns:
     ///     List of (header_name, header_value) tuples
     fn get_headers(&self, _py: Python) -> PyResult<PyObject> {
-        Err(PyNotImplementedError::new_err(
-            "Subclasses must implement get_headers()",
-        ))
+        Err(PyNotImplementedError::new_err("Subclasses must implement get_headers()"))
     }
 }
 

python/rust/src/common.rs

@@ -1,15 +1,17 @@
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex, OnceLock};
 
+use databricks_zerobus_ingest_sdk::{
+    AckCallback as RustAckCallback,
+    EncodedRecord,
+    OffsetId,
+    StreamBuilder,
+};
 use prost::Message;
 use pyo3::exceptions::{PyException, PyValueError};
 use pyo3::prelude::*;
 use pyo3::types::{PyBytes, PyDict, PyList, PyTuple};
 
-use databricks_zerobus_ingest_sdk::{
-    AckCallback as RustAckCallback, EncodedRecord, OffsetId, StreamBuilder,
-};
-
 /// User-agent prefix emitted by this wrapper SDK. Combined with the wrapper
 /// crate version via `env!("CARGO_PKG_VERSION")` at the call site.
 pub(crate) const SDK_IDENTIFIER_PREFIX: &str = "zerobus-sdk-py";
@@ -273,34 +275,24 @@ impl StreamConfigurationOptions {
     /// Validate that all numeric fields are non-negative before casting to unsigned types.
     pub fn validate(&self) -> PyResult<()> {
         if self.max_inflight_records < 0 {
-            return Err(PyValueError::new_err(
-                "max_inflight_records must be non-negative",
-            ));
+            return Err(PyValueError::new_err("max_inflight_records must be non-negative"));
         }
         if self.recovery_timeout_ms < 0 {
-            return Err(PyValueError::new_err(
-                "recovery_timeout_ms must be non-negative",
-            ));
+            return Err(PyValueError::new_err("recovery_timeout_ms must be non-negative"));
         }
         if self.recovery_backoff_ms < 0 {
-            return Err(PyValueError::new_err(
-                "recovery_backoff_ms must be non-negative",
-            ));
+            return Err(PyValueError::new_err("recovery_backoff_ms must be non-negative"));
         }
         if self.recovery_retries < 0 {
-            return Err(PyValueError::new_err(
-                "recovery_retries must be non-negative",
-            ));
+            return Err(PyValueError::new_err("recovery_retries must be non-negative"));
         }
         if self.server_lack_of_ack_timeout_ms < 0 {
             return Err(PyValueError::new_err(
                 "server_lack_of_ack_timeout_ms must be non-negative",
             ));
         }
         if self.flush_timeout_ms < 0 {
-            return Err(PyValueError::new_err(
-                "flush_timeout_ms must be non-negative",
-            ));
+            return Err(PyValueError::new_err("flush_timeout_ms must be non-negative"));
         }
         if let Some(v) = self.stream_paused_max_wait_time_ms {
             if v < 0 {
@@ -454,9 +446,7 @@ pub(crate) fn extract_record_payloads(payloads: &PyAny) -> PyResult<Vec<EncodedR
             out.push(EncodedRecord::Json(json));
         }
     } else {
-        return Err(PyErr::new::<pyo3::exceptions::PyTypeError, _>(
-            "Payloads must be a list",
-        ));
+        return Err(PyErr::new::<pyo3::exceptions::PyTypeError, _>("Payloads must be a list"));
     }
 
     Ok(out)

python/rust/src/lib.rs

@@ -32,14 +32,8 @@ fn _zerobus_core(py: Python, m: &PyModule) -> PyResult<()> {
     m.add_class::<common::AckCallback>()?;
 
     // Add exception types
-    m.add(
-        "ZerobusException",
-        py.get_type::<common::ZerobusException>(),
-    )?;
-    m.add(
-        "NonRetriableException",
-        py.get_type::<common::NonRetriableException>(),
-    )?;
+    m.add("ZerobusException", py.get_type::<common::ZerobusException>())?;
+    m.add("NonRetriableException", py.get_type::<common::NonRetriableException>())?;
 
     // Add authentication classes
     m.add_class::<auth::HeadersProvider>()?;

python/rust/src/sync_wrapper.rs

@@ -1,20 +1,27 @@
 use std::sync::Arc;
 
+use databricks_zerobus_ingest_sdk::{
+    StreamBuilder,
+    ZerobusSdk as RustSdk,
+    ZerobusStream as RustStream,
+};
 use pyo3::prelude::*;
 use pyo3::types::PyBytes;
 use tokio::runtime::Runtime;
 use tokio::sync::RwLock;
 
-use databricks_zerobus_ingest_sdk::{
-    StreamBuilder, ZerobusSdk as RustSdk, ZerobusStream as RustStream,
-};
-
 use crate::arrow;
 use crate::arrow::{ArrowStreamConfigurationOptions, ZerobusArrowStream};
 use crate::auth::HeadersProviderWrapper;
 use crate::common::{
-    apply_grpc_options, encoded_record_to_pybytes, extract_record_payload, extract_record_payloads,
-    map_error, StreamConfigurationOptions, TableProperties, SDK_IDENTIFIER_PREFIX,
+    apply_grpc_options,
+    encoded_record_to_pybytes,
+    extract_record_payload,
+    extract_record_payloads,
+    map_error,
+    StreamConfigurationOptions,
+    TableProperties,
+    SDK_IDENTIFIER_PREFIX,
 };
 
 // =============================================================================

rust/CONTRIBUTING.md

@@ -10,6 +10,7 @@ This document covers Rust-specific development setup and workflow.
 
 - Git
 - Rust 1.70+ (stable toolchain)
+- Pinned formatter: `rustup toolchain install nightly-2025-08-07 --profile minimal --component rustfmt`
 - Protocol Buffers compiler (`protoc`)
 
 ### Setting Up Your Development Environment
@@ -37,7 +38,7 @@ Code style is enforced by a formatter check in your pull request. We use `rustfm
 make fmt
 ```
 
-This runs `cargo fmt --all` to format all crates in the workspace.
+This runs `cargo fmt --all` with the same pinned rustfmt used by CI. Compilation, linting, and tests continue to use stable Rust.
 
 ### Running Linters
 

rust/Makefile

@@ -33,7 +33,7 @@ clean:
 	cargo clean
 
 fmt:
-	cargo fmt --all
+	../.github/scripts/rustfmt.sh --all
 
 lint:
 	cargo clippy --all -- -D warnings

rust/examples/arrow/src/main.rs

@@ -2,7 +2,11 @@ use std::error::Error;
 use std::sync::Arc;
 
 use arrow_array::{
-    Float64Array, Int32Array, LargeStringArray, RecordBatch, TimestampMicrosecondArray,
+    Float64Array,
+    Int32Array,
+    LargeStringArray,
+    RecordBatch,
+    TimestampMicrosecondArray,
 };
 use arrow_ipc::CompressionType;
 use databricks_zerobus_ingest_sdk::{ArrowSchema, DataType, Field, TimeUnit, ZerobusSdk};
@@ -128,11 +132,7 @@ async fn main() -> Result<(), Box<dyn Error>> {
 
         if (i + 1) % WAIT_EVERY == 0 {
             stream.wait_for_offset(offset_id).await?;
-            println!(
-                "Acknowledged through batch {} (offset ID {})",
-                i + 1,
-                offset_id
-            );
+            println!("Acknowledged through batch {} (offset ID {})", i + 1, offset_id);
         }
     }