ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-27 22:34:51 -06:00
Code Issues Packages Projects Releases Wiki Activity

docs: extract HashRing into reference design document

Browse Source 
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).
This commit is contained in:
Patrick Buckley
2026-03-30 16:08:01 -07:00
committed by Patrick Buckley
parent c2750de7a4
commit 0cfe521ce7
5 changed files with 211 additions and 216 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/design/consistent-hash-ring.md
+168
View File
@@ -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
}
}
```
tests/test_hash_ring.py
+1 -131
View File
@@ -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)
turnstone/console/router.py
+13
View File
@@ -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] = {
turnstone/console/server.py
+21
View File
@@ -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,
turnstone/core/hash_ring.py
+8 -85
View File
@@ -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)."""