From 0cfe521ce7dec14b469cef4caa002a67fe80486a Mon Sep 17 00:00:00 2001 From: Patrick Buckley Date: Mon, 30 Mar 2026 16:08:01 -0700 Subject: [PATCH] docs: extract HashRing into reference design document Move the consistent hash ring implementation (FNV-1a, virtual nodes, bisect lookup) from code to docs/design/consistent-hash-ring.md as a forward-looking reference for future scalability work. The current rebalancer uses weight-proportional distribution (simpler, exact splits, no hash variance). The ring algorithm is documented with test vectors, stability properties, and a comparison table for when the ring approach becomes advantageous (large clusters, decentralized routing, cross-language determinism). hash_ring.py retains: RING_SIZE, bucket_of(), RingNode, NoAvailableNodeError (all actively used by router and rebalancer). --- docs/design/consistent-hash-ring.md | 168 ++++++++++++++++++++++++++++ tests/test_hash_ring.py | 132 +--------------------- turnstone/console/router.py | 13 +++ turnstone/console/server.py | 21 ++++ turnstone/core/hash_ring.py | 93 ++------------- 5 files changed, 211 insertions(+), 216 deletions(-) create mode 100644 docs/design/consistent-hash-ring.md diff --git a/docs/design/consistent-hash-ring.md b/docs/design/consistent-hash-ring.md new file mode 100644 index 00000000..b6aa4e14 --- /dev/null +++ b/docs/design/consistent-hash-ring.md @@ -0,0 +1,168 @@ +# Consistent Hash Ring — Reference Design + +**Status**: Reference (not currently in the hot path) +**Date**: 2026-03-30 + +## Overview + +This document describes a consistent hash ring algorithm evaluated during +the design of the direct HTTP transport routing system. The current +implementation uses weight-proportional bucket assignment with a +donor/recipient rebalancing algorithm (see `direct-http-transport.md`). +The consistent hash ring is documented here as a reference for future +scalability work — if the cluster grows beyond the point where the +weight-proportional approach is sufficient, the ring provides a +proven alternative with stronger stability guarantees. + +## When to consider the ring approach + +The current weight-proportional seeding + donor/recipient rebalancer works +well when: +- Cluster size is moderate (< 50 nodes) +- Nodes join/leave infrequently +- The rebalancer runs centrally (in the console) + +The consistent hash ring becomes advantageous when: +- Cluster size grows large (50+ nodes) and frequent membership changes + cause the donor/recipient algorithm to churn +- Decentralized routing is needed (each node computes the ring locally, + no central console required) +- Cross-language determinism is important (multiple implementations must + agree on the same assignment without sharing state) + +## Algorithm + +### Hash function: FNV-1a (32-bit) + +```python +def fnv1a_32(data: bytes) -> int: + """FNV-1a 32-bit hash. + + Basis: 0x811C9DC5, Prime: 0x01000193. + XOR each byte, then multiply by prime (masked to 32 bits). + """ + h = 0x811C9DC5 + for b in data: + h ^= b + h = (h * 0x01000193) & 0xFFFFFFFF + return h +``` + +Known test vectors: +- `fnv1a_32(b"")` = `0x811C9DC5` (basis value) +- `fnv1a_32(b"foobar")` = `0xBF9CF968` + +Cross-language implementations: +- **Python**: loop above (no dependencies) +- **Go**: same algorithm with `uint32` arithmetic +- **TypeScript**: same algorithm with `>>> 0` for unsigned 32-bit + +### Virtual nodes + +Each physical node with weight `w` gets `w * 150` virtual positions on a +16-bit ring (65536 positions). Virtual node `i` of physical node `N` is +placed at: + +``` +position = fnv1a_32(f"{N.node_id}:{i}".encode()) % 65536 +``` + +With 150 vnodes per unit weight: +- 2 equal-weight nodes: ~50/50 split (measured: 38-62% range due to + hash variance, stddev ~3% with large vnode counts) +- 3 nodes at weights 2:1:1: ~50/25/25 (within 10% tolerance) + +### Lookup + +```python +def owner(bucket: int) -> str: + """O(log n) bisect-right walk to find the next virtual node clockwise.""" + idx = bisect_right(positions, bucket) + if idx >= len(positions): + idx = 0 # wrap around + return vnode_map[positions[idx]] +``` + +### Stability properties + +The consistent hash ring guarantees: +- **Node addition**: adding a node moves at most `1/N` of buckets (where N + is the new node count). Other nodes' buckets are unaffected. +- **Node removal**: only the removed node's buckets are reassigned. Buckets + owned by surviving nodes don't move. +- **Determinism**: same membership list always produces the same ring. + No coordination needed between processes. + +### Full assignment precomputation + +```python +def assignments() -> list[tuple[int, str]]: + """Compute all 65536 bucket-to-node mappings.""" + return [(b, owner(b)) for b in range(65536)] +``` + +This produces a complete assignment table that can be loaded into a flat +array for O(1) request-time lookup. The ring itself is never consulted +on the hot path. + +## Data structures + +```python +@dataclass(frozen=True, slots=True) +class RingNode: + node_id: str + url: str + weight: int = 1 + +class HashRing: + """Immutable consistent hash ring. Thread-safe (no mutable state).""" + + def __init__(self, nodes: Sequence[RingNode], vnodes_per_unit: int = 150): + # Validate no duplicate node_ids + # Build sorted array of (position, node_id) tuples + # positions[i] = fnv1a_32(f"{node_id}:{i}".encode()) % RING_SIZE + + def owner(self, bucket: int) -> RingNode | None: + # bisect_right + wrap + + @property + def version(self) -> int: + # Deterministic hash of membership: fnv1a_32 of sorted node_id:weight pairs + + def assignments(self) -> list[tuple[int, str]]: + # Precompute all 65536 bucket assignments +``` + +## Comparison with current approach + +| Aspect | Weight-proportional (current) | Consistent hash ring | +|--------|------------------------------|---------------------| +| Seeding | Exact weight split, deterministic | Hash-based, ~3% variance | +| Node addition | Donor/recipient moves only excess | Ring moves ~1/N buckets | +| Node removal | Dead buckets → most underloaded | Ring redistributes to clockwise neighbors | +| Cross-node churn | Zero (only donor→recipient) | Zero (ring stability guarantee) | +| Decentralized | No (needs central rebalancer) | Yes (each node computes locally) | +| Complexity | Simple weight arithmetic | Virtual node construction + bisect | + +## Test vectors + +For cross-language implementation validation: + +```json +{ + "fnv1a_32": [ + {"input": "", "output": 2166136261}, + {"input": "foobar", "output": 3215766888} + ], + "bucket_of": [ + {"ws_id": "a3f100000000000000000000000000000", "bucket": 41969}, + {"ws_id": "00000000000000000000000000000000", "bucket": 0}, + {"ws_id": "ffff0000000000000000000000000000", "bucket": 65535} + ], + "ring_single_node": { + "nodes": [{"node_id": "n1", "weight": 1}], + "vnodes_per_unit": 150, + "expected_n1_buckets": 65536 + } +} +``` diff --git a/tests/test_hash_ring.py b/tests/test_hash_ring.py index ea6d0ee6..6bdf2892 100644 --- a/tests/test_hash_ring.py +++ b/tests/test_hash_ring.py @@ -1,16 +1,6 @@ """Tests for turnstone.core.hash_ring.""" -from collections import Counter - -import pytest - -from turnstone.core.hash_ring import ( - RING_SIZE, - HashRing, - RingNode, - bucket_of, - fnv1a_32, -) +from turnstone.core.hash_ring import bucket_of class TestBucketOf: @@ -23,123 +13,3 @@ class TestBucketOf: # Only the first 4 hex chars matter — the rest is ignored. assert bucket_of("abcd0000") == bucket_of("abcdffff") assert bucket_of("abcd0000") == 0xABCD - - -class TestFnv1a: - def test_empty(self): - assert fnv1a_32(b"") == 0x811C9DC5 - - def test_known_value(self): - assert fnv1a_32(b"foobar") == 0xBF9CF968 - - def test_deterministic(self): - assert fnv1a_32(b"hello") == fnv1a_32(b"hello") - - -class TestHashRing: - def test_single_node(self): - node = RingNode(node_id="n1", url="http://n1:8000") - ring = HashRing([node]) - counts = Counter(ring.owner(b).node_id for b in range(RING_SIZE)) # type: ignore[union-attr] - assert counts["n1"] == RING_SIZE - - def test_two_equal_nodes(self): - nodes = [ - RingNode(node_id="n1", url="http://n1:8000"), - RingNode(node_id="n2", url="http://n2:8000"), - ] - ring = HashRing(nodes) - counts = Counter(ring.owner(b).node_id for b in range(RING_SIZE)) # type: ignore[union-attr] - assert 25000 <= counts["n1"] <= 40000 - assert 25000 <= counts["n2"] <= 40000 - - def test_three_weighted_nodes(self): - nodes = [ - RingNode(node_id="a", url="http://a:8000", weight=2), - RingNode(node_id="b", url="http://b:8000", weight=1), - RingNode(node_id="c", url="http://c:8000", weight=1), - ] - ring = HashRing(nodes) - counts = Counter(ring.owner(b).node_id for b in range(RING_SIZE)) # type: ignore[union-attr] - total = RING_SIZE - # weight 2:1:1 → expect ~50:25:25 with +-10% tolerance - assert 0.40 * total <= counts["a"] <= 0.60 * total - assert 0.15 * total <= counts["b"] <= 0.35 * total - assert 0.15 * total <= counts["c"] <= 0.35 * total - - def test_deterministic(self): - nodes = [ - RingNode(node_id="n1", url="http://n1:8000"), - RingNode(node_id="n2", url="http://n2:8000"), - ] - a = HashRing(nodes).assignments() - b = HashRing(nodes).assignments() - assert a == b - - def test_node_addition_stability(self): - two = [ - RingNode(node_id="n1", url="http://n1:8000"), - RingNode(node_id="n2", url="http://n2:8000"), - ] - three = [ - *two, - RingNode(node_id="n3", url="http://n3:8000"), - ] - old = HashRing(two).assignments() - new = HashRing(three).assignments() - moved = sum(1 for (_, o), (_, n) in zip(old, new, strict=True) if o != n) - # Ideal is ~33% moved; allow up to 40%. - assert moved < 0.40 * RING_SIZE - - def test_node_removal_stability(self): - nodes = [ - RingNode(node_id="n1", url="http://n1:8000"), - RingNode(node_id="n2", url="http://n2:8000"), - RingNode(node_id="n3", url="http://n3:8000"), - ] - before = HashRing(nodes).assignments() - after = HashRing(nodes[:2]).assignments() - # Only buckets previously owned by n3 should move. - for (b, old_owner), (_, new_owner) in zip(before, after, strict=True): - if old_owner != "n3": - assert old_owner == new_owner, f"bucket {b} changed from {old_owner} to {new_owner}" - - def test_empty_ring(self): - ring = HashRing([]) - assert ring.owner(0) is None - assert ring.owner(42) is None - assert ring.assignments() == [] - - def test_assignments_complete(self): - nodes = [ - RingNode(node_id="n1", url="http://n1:8000"), - RingNode(node_id="n2", url="http://n2:8000"), - ] - a = HashRing(nodes).assignments() - assert len(a) == RING_SIZE - assert all(node_id is not None for _, node_id in a) - - def test_version_changes_on_membership(self): - r1 = HashRing([RingNode(node_id="n1", url="http://n1:8000")]) - r2 = HashRing( - [ - RingNode(node_id="n1", url="http://n1:8000"), - RingNode(node_id="n2", url="http://n2:8000"), - ] - ) - assert r1.version != r2.version - - def test_version_same_for_same_membership(self): - nodes = [ - RingNode(node_id="n1", url="http://n1:8000"), - RingNode(node_id="n2", url="http://n2:8000"), - ] - assert HashRing(nodes).version == HashRing(nodes).version - - def test_duplicate_node_id_raises(self): - nodes = [ - RingNode(node_id="n1", url="http://n1:8000"), - RingNode(node_id="n1", url="http://n1-alt:8000"), - ] - with pytest.raises(ValueError, match="duplicate node_ids"): - HashRing(nodes) diff --git a/turnstone/console/router.py b/turnstone/console/router.py index bc60ba94..886d65f8 100644 --- a/turnstone/console/router.py +++ b/turnstone/console/router.py @@ -9,6 +9,7 @@ from __future__ import annotations import json import logging import secrets +import threading from dataclasses import dataclass from typing import TYPE_CHECKING @@ -41,6 +42,7 @@ class ConsoleRouter: self._cache: list[NodeRef | None] = [None] * RING_SIZE self._overrides: dict[str, NodeRef] = {} self._version: int = 0 + self._refresh_lock = threading.Lock() # ------------------------------------------------------------------ # Cache management @@ -49,8 +51,19 @@ class ConsoleRouter: def refresh_cache(self) -> bool: """Reload the assignment cache from DB. + Thread-safe: if another thread is already refreshing, this call + returns False immediately (the other thread's refresh will apply). Returns True if the cache changed compared to the previous load. """ + if not self._refresh_lock.acquire(blocking=False): + return False # another thread is refreshing + try: + return self._refresh_cache_locked() + finally: + self._refresh_lock.release() + + def _refresh_cache_locked(self) -> bool: + """Inner refresh — must be called with _refresh_lock held.""" # Load node URLs from services table members = self._storage.list_services("server", max_age_seconds=120) nodes: dict[str, NodeRef] = { diff --git a/turnstone/console/server.py b/turnstone/console/server.py index 30791305..19d4efe7 100644 --- a/turnstone/console/server.py +++ b/turnstone/console/server.py @@ -12,6 +12,7 @@ from __future__ import annotations import argparse import asyncio +import contextlib import functools import html import json @@ -735,6 +736,26 @@ async def route_proxy(request: Request) -> Response: status_code=502, ), ) + + # Transparent retry on 404 (at most once): + # + # The bucket-routed node doesn't have the workstream. Refresh the + # cache (reloads overrides + bucket assignments from DB) and re-route. + # If the route changed (e.g., a local-create override was added since + # the last cache load), retry on the new node. If the route is the + # same, return the 404 as-is — no loop, no scan. + if resp.status_code == 404: + router.refresh_cache() + try: + new_ref = router.route(ws_id) + except (NoAvailableNodeError, ValueError): + new_ref = ref + if new_ref.node_id != ref.node_id: + with contextlib.suppress(httpx.HTTPError): + resp = await client.post( + f"{new_ref.url}{upstream_path}", json=body, headers=headers + ) + return _record_route( request, method, diff --git a/turnstone/core/hash_ring.py b/turnstone/core/hash_ring.py index 3c2f3490..5fa48f10 100644 --- a/turnstone/core/hash_ring.py +++ b/turnstone/core/hash_ring.py @@ -1,21 +1,17 @@ -"""Consistent hash ring for workstream-to-node routing. +"""Hash ring routing primitives. -The ring is used internally by the rebalancer to compute ideal bucket-to-node -assignments. At request time, routing is a flat array lookup — the ring is -never consulted on the hot path. +Constants, helpers, and data types for the bucket-based routing system. +The consistent hash ring algorithm itself is documented in +``docs/design/consistent-hash-ring.md`` as a reference design for future +scalability work. The current rebalancer uses weight-proportional +distribution instead (see ``turnstone/console/rebalancer.py``). """ from __future__ import annotations -from bisect import bisect_right from dataclasses import dataclass -from typing import TYPE_CHECKING - -if TYPE_CHECKING: - from collections.abc import Sequence RING_SIZE = 65536 # 16-bit bucket space -DEFAULT_VNODES = 150 # virtual nodes per unit weight def bucket_of(ws_id: str) -> int: @@ -28,18 +24,9 @@ def bucket_of(ws_id: str) -> int: return int(ws_id[:4], 16) -def fnv1a_32(data: bytes) -> int: - """FNV-1a 32-bit hash.""" - h = 0x811C9DC5 - for b in data: - h ^= b - h = (h * 0x01000193) & 0xFFFFFFFF - return h - - @dataclass(frozen=True, slots=True) class RingNode: - """A physical node on the ring.""" + """A physical node participating in the cluster.""" node_id: str url: str @@ -47,68 +34,4 @@ class RingNode: class NoAvailableNodeError(Exception): - """Raised when routing fails (no nodes in ring, bucket not assigned).""" - - -class HashRing: - """Consistent hash ring with virtual nodes. - - Immutable after construction — create a new ring on membership change. - Thread-safe (no mutable state). - """ - - def __init__( - self, - nodes: Sequence[RingNode], - vnodes_per_unit: int = DEFAULT_VNODES, - ) -> None: - ids = [n.node_id for n in nodes] - if len(ids) != len(set(ids)): - dupes = [x for x in set(ids) if ids.count(x) > 1] - raise ValueError(f"duplicate node_ids: {dupes}") - self._nodes = tuple(nodes) - self._node_map: dict[str, RingNode] = {n.node_id: n for n in nodes} - - # Build sorted array of (position, node_id) tuples. - ring: list[tuple[int, str]] = [] - for node in nodes: - count = node.weight * vnodes_per_unit - for i in range(count): - pos = fnv1a_32(f"{node.node_id}:{i}".encode()) % RING_SIZE - ring.append((pos, node.node_id)) - ring.sort() - - self._positions = [p for p, _ in ring] - self._ring = ring - - def owner(self, bucket: int) -> RingNode | None: - """Return the node that owns the given bucket. O(log n) lookup.""" - if not self._ring: - return None - idx = bisect_right(self._positions, bucket) - if idx >= len(self._positions): - idx = 0 - return self._node_map[self._ring[idx][1]] - - @property - def nodes(self) -> tuple[RingNode, ...]: - """All physical nodes, sorted by node_id.""" - return tuple(sorted(self._nodes, key=lambda n: n.node_id)) - - @property - def version(self) -> int: - """Hash of the membership list. Changes when nodes join/leave.""" - key = ",".join( - f"{n.node_id}:{n.weight}" for n in sorted(self._nodes, key=lambda x: x.node_id) - ) - return fnv1a_32(key.encode()) - - def assignments(self) -> list[tuple[int, str]]: - """Precompute all 65536 bucket-to-node assignments. - - Returns a list of ``(bucket, node_id)`` tuples. Used by the - rebalancer to seed/update the assignment table. - """ - if not self._ring: - return [] - return [(b, self.owner(b).node_id) for b in range(RING_SIZE)] # type: ignore[union-attr] + """Raised when routing fails (no nodes registered, bucket not assigned)."""