feat: add configurable RPC retries

fluxon_py/config.py

@@ -26,6 +26,7 @@ def debug_print(*args):
 _YAML_KEY_TYPES = (str, int, float, bool)
 _YAML_SCALAR_TYPES = (str, int, float, bool, type(None))
 _U64_MAX = 2**64 - 1
+_U32_MAX = 2**32 - 1
 _SIZE_BYTE_UNITS = {
     "": 1,
     "b": 1,
@@ -195,6 +196,7 @@ def _yaml_template():
 
 fluxonkv_spec:                        # fluxon kv specific config (dict(optional))
   etcd_addresses:                     # Etcd address list ((None|['{str}:{int}']))
+  etcd_rpc_max_retries:               # Retry count after the first etcd RPC attempt (int(optional))
   cluster_name:                       # Cluster name (str)
   share_mem_path:                     # Shared bundle path for mmap.file/shared.json/peer metadata (str)
   large_file_paths:                   # Owner-mode ordered large-file roots (['{str}'](optional))
@@ -535,6 +537,7 @@ def _validate_fluxonkv_contract(cfg: Dict[str, Any]) -> None:
     if is_zero_contribution:
         forbidden_spec_keys = [
             "etcd_addresses",
+            "etcd_rpc_max_retries",
             "redis_compat",
             "sub_cluster",
             "large_file_paths",
@@ -559,6 +562,17 @@ def _validate_fluxonkv_contract(cfg: Dict[str, Any]) -> None:
     if not isinstance(etcd_addresses, list) or len(etcd_addresses) == 0:
         raise ValueError("fluxonkv_spec.etcd_addresses must be a non-empty list")
 
+    if "etcd_rpc_max_retries" in spec:
+        etcd_rpc_max_retries = spec["etcd_rpc_max_retries"]
+        if isinstance(etcd_rpc_max_retries, bool) or not isinstance(
+            etcd_rpc_max_retries, int
+        ):
+            raise ValueError("fluxonkv_spec.etcd_rpc_max_retries must be an int")
+        if not 0 <= etcd_rpc_max_retries <= _U32_MAX:
+            raise ValueError(
+                f"fluxonkv_spec.etcd_rpc_max_retries must be in 0..={_U32_MAX}"
+            )
+
     if "sub_cluster" not in spec:
         raise ValueError("fluxonkv_spec.sub_cluster is required for owner mode")
     sub_cluster = spec.get("sub_cluster")

fluxon_py/tests/test_config.py

@@ -46,6 +46,7 @@ def _build_checks(selected_test_id: Optional[str]) -> List[Tuple[str, Callable[[
         ("load_from_file", _run_test_load_from_file),
         ("to_yaml_str_roundtrip", _run_test_to_yaml_str_roundtrip),
         ("fluxonkv_sub_cluster_config", test_fluxonkv_sub_cluster_config),
+        ("fluxonkv_etcd_rpc_max_retries", test_fluxonkv_etcd_rpc_max_retries),
         ("fluxonkv_owner_requires_sub_cluster", test_fluxonkv_owner_requires_sub_cluster),
         ("fluxonkv_owner_requires_large_file_paths", test_fluxonkv_owner_requires_large_file_paths),
         ("fluxonkv_large_limit_size_contract", test_fluxonkv_large_limit_size_contract),
@@ -184,6 +185,46 @@ def test_fluxonkv_sub_cluster_config():
         print(f"❌ FAIL: test_fluxonkv_sub_cluster_config - {e}")
 
 
+def test_fluxonkv_etcd_rpc_max_retries():
+    """Ensure owner retry overrides accept zero and external configs inherit the value."""
+    try:
+        owner = _owner_fluxonkv_base_config(tag="etcd_rpc_retries")
+        owner["fluxonkv_spec"]["etcd_rpc_max_retries"] = 0
+        loaded = yaml.safe_load(
+            FluxonKvClientConfig(owner).to_fluxon_kv_client_config_yaml_str()
+        )
+        assert loaded["fluxonkv_spec"]["etcd_rpc_max_retries"] == 0
+
+        for invalid_value in (None, True, -1, 2**32):
+            invalid = _owner_fluxonkv_base_config(tag="etcd_rpc_retries_invalid")
+            invalid["fluxonkv_spec"]["etcd_rpc_max_retries"] = invalid_value
+            try:
+                FluxonKvClientConfig(invalid)
+                raise AssertionError(
+                    f"invalid etcd_rpc_max_retries accepted: {invalid_value!r}"
+                )
+            except ValueError:
+                pass
+
+        external = {
+            "instance_key": "test_external",
+            "fluxonkv_spec": {
+                "cluster_name": "test_cluster",
+                "share_mem_path": "/tmp/kvcache_shared_memory/test",
+                "etcd_rpc_max_retries": 0,
+            },
+        }
+        try:
+            FluxonKvClientConfig(external)
+            raise AssertionError("external etcd retry override should be rejected")
+        except ValueError:
+            pass
+
+        print("✅ PASS: test_fluxonkv_etcd_rpc_max_retries")
+    except Exception as e:
+        print(f"❌ FAIL: test_fluxonkv_etcd_rpc_max_retries - {e}")
+
+
 def test_fluxon_pyo3_import_authority():
     """Ensure the PyO3 binding is imported only from the active venv authority."""
     try:

fluxon_release/closed_sdk/manifest.json

@@ -1,15 +1,15 @@
 {
   "object_kind": "FluxonCommuClosedSdk",
-  "schema_version": 7,
-  "abi_version": 9,
-  "sdk_version": "0.2.1",
-  "required_open_surface_version": "0.2.1",
+  "schema_version": 8,
+  "abi_version": 10,
+  "sdk_version": "0.2.2",
+  "required_open_surface_version": "0.2.2",
   "feature_contract": {
     "boundary_mode": "closed-sdk-consumer",
     "runtime_authority_crate": "fluxon_commu_private_impl",
     "transfer_link_snapshot_mode": "watch-v1"
   },
-  "public_workspace_input_sha256": "59c66ec1fe0dc849a3e2a9f8ce4c721cfee6a387286bd25d3d28bf6ca1705197",
+  "public_workspace_input_sha256": "5ddc15e2f78eb17c793ed1595f7071879ca5710c07a5058652f080a70f942e48",
   "layout": {
     "native": "native",
     "lib": "lib"

fluxon_release/release_notes/v0.2.2.md

@@ -1,13 +1,13 @@
 # 🚀 Fluxon v0.2.2
 
-`v0.2.2` rolls up the mainline work merged from July 14 through August 5, 2026, together with the new release pipeline in the tagged revision. The largest changes are a distributed SSD backing tier for Fluxon KV, hybrid S3 object writes, communication ABI 9, event-driven TCP reactors, stronger MQ and framework lifecycle handling, and expanded release-grade CI.
+`v0.2.2` rolls up the mainline work merged from July 14 through August 5, 2026, together with the new release pipeline in the tagged revision. The largest changes are a distributed SSD backing tier for Fluxon KV, hybrid S3 object writes, communication ABI 10, event-driven TCP reactors, stronger MQ and framework lifecycle handling, and expanded release-grade CI.
 
 ## ✨ Highlights
 
 - Added an owner-local SSD backing tier behind the existing Fluxon KV `put` / `get` / `delete` contract.
 - Added size-aware S3 writes, KV-backed write sessions, shared lease keepalive, and retryable temporary-key cleanup.
 - Added `FluxonFsVideoReader` and a pooled reader API for cached random access to video data.
-- Upgraded the closed communication boundary to ABI 9 and added event-driven TCP reactor support.
+- Upgraded the closed communication boundary to ABI 10 and added event-driven TCP reactor support.
 - Strengthened KV member cleanup, MQ close semantics, framework shutdown barriers, and background-task ownership.
 - Added rclone S3 integration coverage and a resource-bounded large-scale MPMC MQ CI scenario.
 - Unified GitHub Release, PyPI, and Docker Hub publication behind one parameter-free GitHub Actions entrypoint.
@@ -49,7 +49,7 @@ The documented single-node `rclone v1.60.1` comparison reports that FluxonFS led
 
 ## 🌐 Communication and KV Lifecycle
 
-- Updated the closed communication boundary to ABI 9 while retaining runtime ABI, open-surface, boundary-mode, and provider-anchor checks.
+- Updated the closed communication boundary to ABI 10 while retaining runtime ABI, open-surface, boundary-mode, and provider-anchor checks.
 - Added event-driven TCP reactor mode alongside the existing busy-poll path, plus bounded test controls for reactor shards and control/bulk lanes.
 - Added explicit KV member lifecycle indexes and cleanup paths for member departure, in-flight requests, replicas, holders, and allocation ownership.
 - Tightened configuration validation and separated stable network configuration from developer-only `test_spec_config` switches.
@@ -81,10 +81,11 @@ The documented single-node `rclone v1.60.1` comparison reports that FluxonFS led
 ## 🔐 Version and SDK Contract
 
 - Public Python package, Rust workspace, and Quick Start version: `0.2.2`.
-- Closed communication SDK version: `0.2.1`.
-- Closed SDK required open-surface contract version: `0.2.1`.
+- Closed communication SDK version: `0.2.2`.
+- Closed SDK required open-surface contract version: `0.2.2`.
+- Closed communication ABI: `10`; SDK manifest schema: `8`.
 
-The SDK version and open-surface contract version are independent from the public release version. The tagged runtime must still pass ABI, open-surface, boundary-mode, and provider-anchor validation.
+The SDK version and open-surface contract version are independent from the public release version. They advance in this release because the `ClusterManagerNewArg` bitcode payload changed incompatibly. The tagged runtime must still pass ABI, open-surface, boundary-mode, and provider-anchor validation.
 
 ## 📦 Release Artifacts
 
@@ -113,4 +114,5 @@ The Docker publication does not update `latest`.
 - #48, #49, #51, #52 — expand and refine the public Fluxon overview and AI-native distributed-data-plane positioning.
 - #47 — support communication ABI 9 and event-driven TCP reactors.
 - #50 — add hybrid S3 writes, shared lease keepalive, cleanup ownership, and shutdown barriers.
+- The tagged revision advances the communication boundary from ABI 9 to ABI 10 to carry the etcd unary RPC retry configuration through the open/closed payload.
 - The tagged `v0.2.2` revision also contains the unified GitHub Release / PyPI / Docker Hub release workflow and its deterministic readiness gates.

fluxon_rs/fluxon_commu/src/facade/closed_sdk.rs

@@ -45,6 +45,11 @@ pub enum CurrentProviderContractError {
         expected_by_sdk: String,
         actual_open_surface: String,
     },
+    VersionFieldMismatch {
+        field: &'static str,
+        expected_by_open: u32,
+        actual_sdk: u32,
+    },
     RuntimeAnchorMismatch {
         field: &'static str,
         expected_by_sdk: usize,
@@ -72,6 +77,15 @@ impl Display for CurrentProviderContractError {
                 "closed SDK requires open surface version {}, but the current open contract version is {}",
                 expected_by_sdk, actual_open_surface
             ),
+            Self::VersionFieldMismatch {
+                field,
+                expected_by_open,
+                actual_sdk,
+            } => write!(
+                f,
+                "closed SDK {} mismatch: open expects {}, sdk advertises {}",
+                field, expected_by_open, actual_sdk
+            ),
             Self::RuntimeAnchorMismatch {
                 field,
                 expected_by_sdk,
@@ -109,6 +123,20 @@ pub fn assert_current_provider_contract()
     assert_abi_compatible()?;
 
     let version_info = query_version()?;
+    if version_info.abi_version != FLUXON_COMMU_CLOSED_ABI_VERSION {
+        return Err(CurrentProviderContractError::VersionFieldMismatch {
+            field: "ABI version",
+            expected_by_open: FLUXON_COMMU_CLOSED_ABI_VERSION,
+            actual_sdk: version_info.abi_version,
+        });
+    }
+    if version_info.sdk_schema_version != FLUXON_COMMU_CLOSED_SDK_SCHEMA_VERSION {
+        return Err(CurrentProviderContractError::VersionFieldMismatch {
+            field: "schema version",
+            expected_by_open: FLUXON_COMMU_CLOSED_SDK_SCHEMA_VERSION,
+            actual_sdk: version_info.sdk_schema_version,
+        });
+    }
     let expected_boundary_mode = crate::provider::CURRENT_PROVIDER_BOUNDARY_MODE;
     if version_info.boundary_mode != expected_boundary_mode {
         return Err(CurrentProviderContractError::BoundaryModeMismatch {
@@ -225,7 +253,10 @@ pub(crate) fn spawn_deferred_drop_runtime_handle(
 
 #[cfg(test)]
 mod tests {
-    use super::assert_current_provider_contract;
+    use super::{
+        FLUXON_COMMU_CLOSED_ABI_VERSION, FLUXON_COMMU_CLOSED_SDK_SCHEMA_VERSION, abi_version,
+        assert_current_provider_contract, sdk_schema_version,
+    };
     use fluxon_commu_contract::FLUXON_COMMU_OPEN_SURFACE_VERSION;
 
     #[test]
@@ -236,6 +267,16 @@ mod tests {
             snapshot.version_info.required_open_surface_version,
             FLUXON_COMMU_OPEN_SURFACE_VERSION
         );
+        assert_eq!(
+            snapshot.version_info.abi_version,
+            FLUXON_COMMU_CLOSED_ABI_VERSION
+        );
+        assert_eq!(
+            snapshot.version_info.sdk_schema_version,
+            FLUXON_COMMU_CLOSED_SDK_SCHEMA_VERSION
+        );
+        assert_eq!(abi_version(), FLUXON_COMMU_CLOSED_ABI_VERSION);
+        assert_eq!(sdk_schema_version(), FLUXON_COMMU_CLOSED_SDK_SCHEMA_VERSION);
         assert_eq!(
             snapshot.sdk_runtime_anchor.cluster_manager_size,
             snapshot.provider_runtime_anchor.cluster_manager_size

fluxon_rs/fluxon_commu/src/facade/cluster_manager.rs

@@ -648,6 +648,7 @@ impl ClusterManager {
 
     pub async fn new(
         etcd_endpoints: Vec<String>,
+        etcd_rpc_max_retries: u32,
         cluster_name: String,
         instance_name: Option<String>,
         port: Option<u16>,
@@ -659,6 +660,7 @@ impl ClusterManager {
     ) -> crate::ClusterResult<Self> {
         Self::construct(ClusterManagerNewArg {
             etcd_endpoints,
+            etcd_rpc_max_retries,
             cluster_name,
             instance_name,
             port,

fluxon_rs/fluxon_commu/src/lib.rs

@@ -26,7 +26,7 @@ pub use closed_sdk::{
 pub use cluster::{
     ClusterError, ClusterEvent, ClusterMember, ClusterResult, ETCD_PREFIX_SCAN_PAGE_LIMIT,
     EtcdPrefixScanAction, EtcdPrefixScanError, NodeID, NodeIDStr, NodeIDString, NodeRole,
-    scan_etcd_prefix_paginated,
+    scan_etcd_prefix_paginated, scan_etcd_prefix_paginated_with_retry,
 };
 pub use cluster_manager::{
     ClusterManager, ClusterManagerNewArg, ClusterManagerRdmaControlInit,

fluxon_rs/fluxon_commu_closed_sdk_consumer/src/lib.rs

@@ -26,8 +26,8 @@ use fluxon_commu_contract::{
 
 pub mod rdma_probe;
 
-pub const FLUXON_COMMU_CLOSED_SDK_SCHEMA_VERSION: u32 = 7;
-pub const FLUXON_COMMU_CLOSED_ABI_VERSION: u32 = 9;
+pub const FLUXON_COMMU_CLOSED_SDK_SCHEMA_VERSION: u32 = 8;
+pub const FLUXON_COMMU_CLOSED_ABI_VERSION: u32 = 10;
 pub const FLUXON_COMMU_CLOSED_HOST_CALLBACKS_ABI_VERSION: u32 = 8;
 pub const FLUXON_COMMU_CLOSED_RUNTIME_RESULT_OK: i32 = 0;
 pub const FLUXON_COMMU_CLOSED_RUNTIME_RESULT_ERR: i32 = 1;

fluxon_rs/fluxon_commu_contract/src/cluster.rs

@@ -1,6 +1,7 @@
 use crate::config::NetworkConfig;
 use bitcode::{Decode, Encode};
 use etcd_client::{Client, GetOptions};
+use fluxon_util::etcd::retry_etcd_rpc;
 use fluxon_util::prefix_scan::{
     PrefixScanAction, prefix_scan_key_after, prefix_scan_range_end_exclusive,
 };
@@ -36,6 +37,23 @@ where
 pub async fn scan_etcd_prefix_paginated<E, F>(
     client: &mut Client,
     prefix: &str,
+    on_kv: F,
+) -> Result<(), EtcdPrefixScanError<E>>
+where
+    E: std::fmt::Display + std::fmt::Debug,
+    F: FnMut(&[u8], &[u8]) -> Result<EtcdPrefixScanAction, E>,
+{
+    scan_etcd_prefix_paginated_with_retry(client, prefix, 0, on_kv).await
+}
+
+/// Scans an etcd prefix page by page and retries transient page-range failures.
+///
+/// A page is delivered to `on_kv` only after its Range RPC succeeds, so retrying
+/// the RPC never invokes the callback twice for the same failed attempt.
+pub async fn scan_etcd_prefix_paginated_with_retry<E, F>(
+    client: &mut Client,
+    prefix: &str,
+    max_retries: u32,
     mut on_kv: F,
 ) -> Result<(), EtcdPrefixScanError<E>>
 where
@@ -46,21 +64,29 @@ where
     let mut start_key = prefix.as_bytes().to_vec();
 
     loop {
-        let resp = client
-            .get(
-                start_key.clone(),
-                Some(
-                    GetOptions::new()
-                        .with_range(range_end.clone())
-                        .with_limit(ETCD_PREFIX_SCAN_PAGE_LIMIT),
-                ),
-            )
-            .await
-            .map_err(|source| EtcdPrefixScanError::Get {
-                prefix: prefix.to_string(),
-                start_key: start_key.clone(),
-                source,
-            })?;
+        let resp = retry_etcd_rpc(max_retries, "prefix_range", || {
+            let mut attempt_client = client.clone();
+            let attempt_start_key = start_key.clone();
+            let attempt_range_end = range_end.clone();
+            async move {
+                attempt_client
+                    .get(
+                        attempt_start_key,
+                        Some(
+                            GetOptions::new()
+                                .with_range(attempt_range_end)
+                                .with_limit(ETCD_PREFIX_SCAN_PAGE_LIMIT),
+                        ),
+                    )
+                    .await
+            }
+        })
+        .await
+        .map_err(|source| EtcdPrefixScanError::Get {
+            prefix: prefix.to_string(),
+            start_key: start_key.clone(),
+            source,
+        })?;
 
         if resp.kvs().is_empty() {
             break;

fluxon_rs/fluxon_commu_contract/src/cluster_manager.rs

@@ -1,6 +1,7 @@
 pub use crate::cluster::{
     ClusterError, ClusterEvent, ClusterMember, ClusterResult, EtcdPrefixScanAction, NodeID,
     NodeIDStr, NodeIDString, NodeRole, scan_etcd_prefix_paginated,
+    scan_etcd_prefix_paginated_with_retry,
 };
 pub use crate::config::NetworkConfig;
 pub use crate::member_metadata::{
@@ -80,6 +81,7 @@ impl IpcBandwidthAttributorHandle {
 #[derive(Clone, Debug, Encode, Decode)]
 pub struct ClusterManagerNewArg {
     pub etcd_endpoints: Vec<String>,
+    pub etcd_rpc_max_retries: u32,
     pub cluster_name: String,
     pub instance_name: Option<String>,
     pub port: Option<u16>,

fluxon_rs/fluxon_commu_contract/src/lib.rs

@@ -1,7 +1,7 @@
 extern crate self as fluxon_commu_contract;
 
 /// Version of the open contract consumed by the closed communication SDK.
-pub const FLUXON_COMMU_OPEN_SURFACE_VERSION: &str = "0.2.1";
+pub const FLUXON_COMMU_OPEN_SURFACE_VERSION: &str = "0.2.2";
 
 pub mod closed_runtime;
 pub mod cluster;