feat(aerospike)!: rename total_timeout_ms -> batch_total_timeout_ms a…

…nd add socket_timeout_ms Review feedback: total_timeout_ms was ambiguous (users read it as a global/end-to-end timeout) and the timeout surface was missing socket_timeout, which is the per-attempt trigger that lets max_retries actually fire within the total budget. * total_timeout_ms -> batch_total_timeout_ms. Now explicitly named after the Aerospike batch policy it maps to, matches read_timeout_ms / write_timeout_ms in framing (each targets one policy scope). * Add socket_timeout_ms (optional). Applies uniformly to read, write and batch policies when set. Leaves the Aerospike client default in place when unset. BREAKING CHANGE: total_timeout_ms is renamed to batch_total_timeout_ms. Config files using the old name must be updated. No default value change. Docs updated (reference + perf-tuning guide) with a short explainer on the per-attempt vs total deadline distinction. Two new unit tests pin the policy wiring: socket_timeout_ms propagates to all three scopes, and is omitted (not injected as None) when unset. Signed-off-by: Valentyn Kahamlyk <valentin.kagamlyk@gmail.com>
ntkathole · Jul 2, 2026 · Apr 20, 2026 · Apr 20, 2026 · Apr 20, 2026 · Apr 20, 2026
commit 5553367066fc94fe6908ab8b95b172a7449ffbb7
@@ -480,18 +480,19 @@ online_store:
     - ["aerospike-1.internal", 3000]
     - ["aerospike-2.internal", 3000]
   namespace: feast
-  read_timeout_ms: 50
-  write_timeout_ms: 200
-  total_timeout_ms: 500
+  read_timeout_ms: 150           # hard deadline for a single-record get
+  write_timeout_ms: 300           # hard deadline for a single-record put/operate
+  batch_total_timeout_ms: 500     # hard deadline for online_read / online_write_batch
+  socket_timeout_ms: 50           # per-attempt deadline so max_retries can actually fire
   max_retries: 2
-  ttl_seconds: 86400            # record-level TTL; omit to use the namespace default
-  client_kwargs:                # escape hatch for any client-config field not surfaced above
+  ttl_seconds: 86400              # record-level TTL; omit to use the namespace default
+  client_kwargs:                  # escape hatch for any client-config field not surfaced above
     policies:
       batch:
-        concurrent_nodes: 0     # 0 = parallel to every node (lowest latency on multi-node clusters)
+        concurrent_nodes: 0       # 0 = parallel to every node (lowest latency on multi-node clusters)
 ```
 
-- **`read_timeout_ms` / `write_timeout_ms` / `total_timeout_ms`**: Per-call deadlines (in milliseconds). Tighter deadlines improve p99 by failing fast on misbehaving nodes; pair with `max_retries`.
+- **`*_timeout_ms` (total)** vs **`socket_timeout_ms` (per-attempt)**: `*_timeout_ms` is the hard deadline for a whole call *including* retries; `socket_timeout_ms` is the per-attempt deadline that allows `max_retries` to actually fire within that budget. Without `socket_timeout_ms`, a single slow attempt can consume the entire total deadline and retries never run.
 - **`hosts`**: List every seed node. The Aerospike client discovers the rest of the cluster automatically and opens one connection pool per node.
 - **`ttl_seconds: 0`** means "never expire"; omit the key to inherit the namespace's `default-ttl`. Expiry is enforced by the server's `nsup` thread — nothing to delete on the client side.
 - Co-locate the feature server in the **same availability zone / rack** as the Aerospike cluster; sub-millisecond reads are bandwidth- and RTT-sensitive.

@@ -58,14 +58,22 @@ online_store:
     - ["aerospike-2.internal", 3000]
     - ["aerospike-3.internal", 3000]
   namespace: feast
-  ttl_seconds: 86400            # 24h record-level TTL
-  read_timeout_ms: 50
-  write_timeout_ms: 200
-  total_timeout_ms: 500
+  ttl_seconds: 86400             # 24h record-level TTL
+  read_timeout_ms: 150            # hard deadline for a single-record get
+  write_timeout_ms: 300           # hard deadline for a single-record put/operate
+  batch_total_timeout_ms: 500     # hard deadline for online_read / online_write_batch
+  socket_timeout_ms: 50           # per-attempt deadline so max_retries can fire
   max_retries: 2
 ```
 {% endcode %}
 
+> **Timeout semantics.** The Aerospike client distinguishes per-attempt
+> (`socket_timeout`) from total (`total_timeout`) deadlines. `*_timeout_ms` map
+> to `total_timeout` — the overall budget for a call including retries. Set
+> `socket_timeout_ms` as well so each individual attempt has its own (shorter)
+> deadline; without it, `max_retries` effectively never fires because the
+> first attempt is allowed to consume the entire total deadline.
+
 ### Aerospike Enterprise with authentication
 
 > Requires Aerospike Enterprise Edition. The Community Edition server has no built-in user/security model and will reject these config keys.

@@ -135,13 +135,27 @@ class AerospikeOnlineStoreConfig(FeastConfigBaseModel):
     No per-feature-view override in v1."""
 
     write_timeout_ms: int = 1_000
-    """Per-call write timeout in milliseconds."""
+    """Per-call write total timeout in milliseconds. This is the hard deadline
+    the client gives a single ``put`` / ``operate`` — including any retries —
+    to return a response, after which the call fails."""
 
     read_timeout_ms: int = 250
-    """Per-call read timeout in milliseconds."""
-
-    total_timeout_ms: int = 2_000
-    """Total (including retries) timeout in milliseconds."""
+    """Per-call read total timeout in milliseconds. Hard deadline for a
+    single-record ``get`` — including any retries — after which the call
+    fails."""
+
+    batch_total_timeout_ms: int = 2_000
+    """Total timeout in milliseconds for a whole ``batch_write`` /
+    ``batch_operate`` call, including retries. Applies to every batch
+    operation ``online_read`` and ``online_write_batch`` issue."""
+
+    socket_timeout_ms: Optional[int] = None
+    """Per-attempt socket timeout in milliseconds. This is the per-retry
+    trigger that lets ``max_retries`` actually fire within the caller's
+    overall ``*_timeout_ms`` budget — without it, a single attempt can
+    consume the whole deadline and retries never run. Applied uniformly
+    to ``read``, ``write`` and ``batch`` policies. ``None`` leaves the
+    client default in place."""
 
     max_retries: int = 2
     """Maximum number of automatic retries on transient errors."""
@@ -188,18 +202,30 @@ def _get_client(self, config: RepoConfig) -> aerospike.Client:
             raise RuntimeError(f"{config.online_store.type = }. It must be aerospike.")
         store_cfg = config.online_store
 
+        read_policy: Dict[str, Any] = {
+            "total_timeout": store_cfg.read_timeout_ms,
+            "max_retries": store_cfg.max_retries,
+        }
+        write_policy: Dict[str, Any] = {
+            "total_timeout": store_cfg.write_timeout_ms,
+            "max_retries": store_cfg.max_retries,
+        }
+        batch_policy: Dict[str, Any] = {
+            "total_timeout": store_cfg.batch_total_timeout_ms,
+        }
+        if store_cfg.socket_timeout_ms is not None:
+            # socket_timeout is the per-attempt deadline; without it,
+            # total_timeout is the whole budget and retries never fire.
+            read_policy["socket_timeout"] = store_cfg.socket_timeout_ms
+            write_policy["socket_timeout"] = store_cfg.socket_timeout_ms
+            batch_policy["socket_timeout"] = store_cfg.socket_timeout_ms
+
         client_config: Dict[str, Any] = {
             "hosts": [tuple(h) for h in store_cfg.hosts],
             "policies": {
-                "read": {
-                    "total_timeout": store_cfg.read_timeout_ms,
-                    "max_retries": store_cfg.max_retries,
-                },
-                "write": {
-                    "total_timeout": store_cfg.write_timeout_ms,
-                    "max_retries": store_cfg.max_retries,
-                },
-                "batch": {"total_timeout": store_cfg.total_timeout_ms},
+                "read": read_policy,
+                "write": write_policy,
+                "batch": batch_policy,
             },
             **store_cfg.client_kwargs,
         }

@@ -120,6 +120,61 @@ def test_resolve_ttl_sentinels():
     assert _resolve_ttl(3600) == 3600
 
 
+def test_socket_timeout_ms_propagates_to_read_write_and_batch_policies(monkeypatch):
+    """``socket_timeout_ms`` must apply uniformly to read, write and batch
+    policies — otherwise ``max_retries`` can't fire within the total budget
+    (the first attempt alone consumes the whole deadline)."""
+    captured: dict = {}
+
+    def fake_client(cfg):
+        captured["cfg"] = cfg
+        fake = MagicMock()
+        fake.connect.return_value = fake
+        return fake
+
+    monkeypatch.setattr(aerospike, "client", fake_client)
+
+    config = _aerospike_repo_config(
+        batch_total_timeout_ms=1500,
+        socket_timeout_ms=40,
+        read_timeout_ms=100,
+        write_timeout_ms=200,
+    )
+    store = AerospikeOnlineStore()
+    store._get_client(config)
+
+    policies = captured["cfg"]["policies"]
+    assert policies["read"]["total_timeout"] == 100
+    assert policies["read"]["socket_timeout"] == 40
+    assert policies["write"]["total_timeout"] == 200
+    assert policies["write"]["socket_timeout"] == 40
+    assert policies["batch"]["total_timeout"] == 1500
+    assert policies["batch"]["socket_timeout"] == 40
+
+
+def test_socket_timeout_ms_omitted_when_not_set(monkeypatch):
+    """Unset ``socket_timeout_ms`` must leave the client's default in place,
+    not inject ``None`` into the policy dicts."""
+    captured: dict = {}
+
+    def fake_client(cfg):
+        captured["cfg"] = cfg
+        fake = MagicMock()
+        fake.connect.return_value = fake
+        return fake
+
+    monkeypatch.setattr(aerospike, "client", fake_client)
+
+    store = AerospikeOnlineStore()
+    store._get_client(_aerospike_repo_config())
+
+    policies = captured["cfg"]["policies"]
+    for scope in ("read", "write", "batch"):
+        assert "socket_timeout" not in policies[scope], (
+            f"socket_timeout leaked into {scope} policy with no config override"
+        )
+
+
 # ---------------------------------------------------------------------------
 # _convert_raw_docs_to_proto — same contract as MongoDB's helper
 # ---------------------------------------------------------------------------