Metadata-Version: 2.4
Name: kinship
Version: 0.4.0
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Rust
Classifier: Topic :: System :: Distributed Computing
Classifier: Typing :: Typed
Requires-Dist: fastapi>=0.110 ; extra == 'fastapi'
Provides-Extra: fastapi
License-File: LICENSE-APACHE
License-File: LICENSE-MIT
Summary: SWIM membership and failure detection with Lifeguard, for Python
Keywords: swim,gossip,membership,failure-detection,cluster,asyncio
Author: Nikola Stankovic
License-Expression: Apache-2.0 OR MIT
Requires-Python: >=3.11
Description-Content-Type: text/markdown; charset=UTF-8; variant=GFM
Project-URL: Changelog, https://github.com/bytevader/kinship/blob/main/CHANGELOG.md
Project-URL: Issues, https://github.com/bytevader/kinship/issues
Project-URL: Repository, https://github.com/bytevader/kinship

<p align="center">
  <picture>
    <source media="(prefers-color-scheme: dark)" srcset="https://raw.githubusercontent.com/bytevader/kinship/main/docs/assets/banner-dark.png">
    <img src="https://raw.githubusercontent.com/bytevader/kinship/main/docs/assets/banner-light.png" alt="kinship" width="560">
  </picture>
</p>
<p align="center">
  <em>Know which of your processes are alive, without a coordinator: SWIM membership and failure detection with Lifeguard, in Rust, for asyncio Python and tokio.</em>
</p>
<p align="center">
  <a href="https://github.com/bytevader/kinship/actions/workflows/ci.yml?query=branch%3Amain"><img src="https://github.com/bytevader/kinship/actions/workflows/ci.yml/badge.svg?branch=main" alt="Test"></a>
  <a href="https://github.com/bytevader/kinship/actions/workflows/ci.yml?query=branch%3Amain"><img src="https://img.shields.io/endpoint?url=https%3A%2F%2Fraw.githubusercontent.com%2Fbytevader%2Fkinship%2Fbadges%2Fcoverage.json" alt="Coverage"></a>
  <a href="https://pypi.org/project/kinship/"><img src="https://img.shields.io/pypi/v/kinship?color=%2334D058&label=pypi%20package" alt="Package version"></a>
  <a href="https://pypi.org/project/kinship/"><img src="https://img.shields.io/badge/python-3.11%2B%20%C2%B7%203.14t-%2334D058" alt="Supported Python versions"></a>
  <a href="https://crates.io/crates/kinship"><img src="https://img.shields.io/crates/v/kinship?color=%2334D058" alt="crates.io"></a>
  <a href="https://docs.rs/kinship"><img src="https://img.shields.io/docsrs/kinship" alt="docs.rs"></a>
  <a href="https://github.com/bytevader/kinship/blob/main/LICENSE-APACHE"><img src="https://img.shields.io/badge/license-Apache--2.0%20OR%20MIT-blue" alt="License: Apache-2.0 OR MIT"></a>
</p>

---

**Source code**: <a href="https://github.com/bytevader/kinship" target="_blank">https://github.com/bytevader/kinship</a>

**Design and benchmarks**: <a href="https://github.com/bytevader/kinship/blob/main/docs/design.md" target="_blank">design</a>, <a href="https://github.com/bytevader/kinship/blob/main/docs/results/detection.md" target="_blank">detection latency against memberlist</a>

---

kinship tells every Python process in a cluster which peers are alive, suspect, dead or gone, within seconds and with no coordinator to run. It is SWIM with the full Lifeguard extensions, the protocol behind HashiCorp's memberlist, Serf and Consul.

Use it when you are building your own distributed piece (a sharded cache, a job scheduler, a WebSocket fleet, a pool of edge boxes) and you need membership, not consensus. kinship does not elect leaders, store data or give you agreement during a partition.

- **Fast, honest failure detection.** A kill -9 at 100 nodes is detected in 9.7 s at the median, against 9.8 s for HashiCorp memberlist on the same network ([benchmark](https://github.com/bytevader/kinship/blob/main/docs/results/detection.md)).
- **No false deaths from a busy process.** The protocol runs in Rust on its own thread, and Lifeguard stretches a slow node's own timeouts, so a blocked event loop or a long GIL hold does not get a healthy node declared dead.
- **Encrypted by default.** XChaCha20-Poly1305 on every packet, replay protection, and key rotation with no downtime.
- **Python and Rust.** An asyncio API and a blocking one for Flask, Django and Celery, abi3 wheels plus free-threaded 3.14t; a tokio `Cluster` handle and a sans-IO protocol core for Rust.
- **Reproducible.** Every run of the protocol core replays from a seed; CI sweeps 1,000 simulated seeds per scenario, fuzzes the decoders and kills nodes on real sockets under packet loss.

## Install

```bash
pip install kinship
```

Wheels cover CPython 3.11 and later (abi3) plus free-threaded 3.14t, on Linux (x86_64, aarch64), macOS (arm64, x86_64) and Windows (x86_64). Linux armv7 has abi3 wheels for glibc 2.28+ and musl 1.2+. No Rust toolchain is needed.

```bash
cargo add kinship        # tokio runtime and Cluster handle
cargo add kinship-core   # the sans-IO protocol state machine only
```

The Rust crates need Rust 1.85 or later.

## Quickstart

Save this as `demo.py`:

```python
import asyncio, sys, kinship

async def main(port: int, seeds: list[str]) -> None:
    cfg = kinship.Config.local(bind=f"127.0.0.1:{port}", seeds=seeds)
    async with kinship.Cluster(cfg) as cluster:
        async for event in cluster.events():
            print(event, sorted(m.name for m in cluster.members()))

port, *seeds = sys.argv[1:]
asyncio.run(main(int(port), [f"127.0.0.1:{s}" for s in seeds]))
```

Run it in three terminals, then press Ctrl-C in one of them:

```bash
python demo.py 7946
python demo.py 7947 7946
python demo.py 7948 7946
```

```text
MemberJoined(laptop-4be1c9) ['laptop-4be1c9', 'laptop-a07f3e', 'laptop-d2913b']
MemberLeft(laptop-d2913b) ['laptop-4be1c9', 'laptop-a07f3e']
```

Kill a process with `kill -9` instead and you see `MemberSuspect`, then `MemberDead` a few seconds later. `Config.local()` runs without encryption because it only binds loopback. Anything that leaves the host needs a key, as below.

## Going beyond one machine

```bash
python -m kinship keygen     # prints a base64 32-byte key; give the same key to every node
```

```python
import os, kinship

cfg = kinship.Config.lan(
    name="cache-07",                     # stable names survive restarts
    keys=[os.environ["KINSHIP_KEY"]],
    seeds=["10.0.0.5:7946", "10.0.0.6:7946"],
    cluster="cache-prod",                # clusters with other labels ignore you
)
```

Every node can share the same seed list, including itself. A seed that is down or is the node itself is skipped, and a node that reaches no seed starts alone and keeps retrying in the background.

A seed's host can be a DNS name, such as a docker compose service (`seeds=["node-1:7946"]`) or a Kubernetes headless Service (`seeds=["kinship.default.svc.cluster.local:7946"]`). kinship looks each name up again at every join and every background retry, and every address it returns, A and AAAA, is a seed, so one name can stand for the whole cluster and follows pods to new addresses. A name that does not resolve is logged and looked up again; only when no seed resolves within `resolve_timeout` (30 s) does starting the cluster raise `JoinError`. A headless Service lists only ready pods by default, so set `publishNotReadyAddresses: true` on it if kinship starts before your readiness probe passes, or the first pod finds no seed.

## How membership works

Each node probes one peer per round. A peer that misses its probe becomes **suspect**, and the suspicion spreads by gossip. The suspect node hears the rumour and refutes it if it is alive; otherwise it becomes **dead** a few seconds later. A node that shuts down cleanly becomes **left** instead of dead. Views are eventually consistent: two nodes can disagree for a few gossip rounds, and a partition gives you two groups that each think the other is dead until it heals.

## Cluster

```python
async with kinship.Cluster(cfg) as cluster:   # binds, then joins cfg.seeds
    ...
# on exit: leave(timeout=5.0), then close()
```

| Call | Returns | What it does |
| --- | --- | --- |
| `await cluster.join(seeds=None)` | `int` | Push-pull with each seed until one answers; returns how many answered. Raises `JoinError` if none did, or if no seed name resolved within `resolve_timeout`. With no argument it uses `cfg.seeds`. |
| `cluster.members()` | `list[Member]` | Alive and suspect members, the local node included. Never waits on the network. |
| `cluster.member(name)` | `Member \| None` | One member by name, including dead and left tombstones. |
| `cluster.local` | `Member` | This node as the cluster sees it. |
| `cluster.events()` | `EventStream` | A new async iterator of events; see below. |
| `await cluster.set_meta(meta)` | `None` | Replace this node's metadata and gossip it. |
| `await cluster.update_meta(**changes)` | `None` | Change some metadata keys; a value of `None` deletes the key. |
| `await cluster.send(member, data)` | `None` | Send bytes to one member; see Messages. |
| `await cluster.broadcast(data)` | `None` | Send bytes to every other member; see Messages. |
| `cluster.keyring` | `Keyring` | Runtime key rotation on this node; see Encryption. |
| `cluster.stats()` | `Stats` | Counters: local health score, probes, suspicions raised and refuted, restarts seen, `decrypt_failures`, `decode_errors`, `replays_dropped`, and the user messages sent, received and dropped. |
| `await cluster.leave(timeout=5.0)` | `None` | Tell the cluster this node is leaving and wait for it to spread. |
| `await cluster.close()` | `None` | Stop the node and close sockets. Without `leave()` first, peers see a crash. |

If you cannot use `async with`, call `await cluster.start()` yourself and `await cluster.close()` in a `finally`.

A `Member` is frozen and hashable by name:

| Field | Type | Meaning |
| --- | --- | --- |
| `name` | `str` | Unique node name, 1 to 64 UTF-8 bytes |
| `addr` | `str` | Gossip address, `"10.0.0.5:7946"` or `"[fd00::5]:7946"` |
| `state` | `kinship.State` | `ALIVE`, `SUSPECT`, `DEAD` or `LEFT` |
| `incarnation` | `int` | Raised only by the member itself, on refutation or metadata change |
| `meta` | `Mapping[str, str]` | Metadata tags, read-only |
| `epoch` | `int` | Which run of the node this is: the milliseconds of its wall clock when the process started. `0` for a node on kinship older than 0.3, which sends none |

The epoch tells one run of a node from the next. A node that restarts under the same name comes back with a higher epoch, and every node replaces the old run with the new one at once, whatever their incarnations: the new process counts those from 0 again. Rumours about the old run, such as its death or a recording of its Alive, change nothing about the new one, and cannot bring it back once it has died. If the clock of the restarted machine reads behind the old run's epoch, the node raises its epoch past it when the cluster tells it about the old run, during its join. Use `(member.name, member.epoch)` wherever you need to know that you still talk to the same process, or `ring.fence(key)` for writes.

## Events

```python
async for event in cluster.events():
    match event:
        case kinship.MemberJoined(member=m):
            ring.add(m)
        case kinship.MemberSuspect(member=m):
            log.warning("%s is not answering", m.name)
        case kinship.MemberDead(member=m) | kinship.MemberLeft(member=m):
            ring.remove(m)
        case kinship.MemberUpdated(member=m, previous_meta=old):
            ring.update(m)
        case kinship.UserMessage(sender=m, data=data):
            await handle(m, data)
        case kinship.EventsLost(count=n):
            ring.rebuild(cluster.members())
```

| Event | Fields | When |
| --- | --- | --- |
| `MemberJoined` | `member` | A node is alive that was unknown, dead or left |
| `MemberSuspect` | `member` | A node missed its probes and is suspected |
| `MemberRecovered` | `member` | A suspect node refuted the suspicion and is alive again |
| `MemberDead` | `member` | A suspicion expired without a refutation |
| `MemberLeft` | `member` | A node left on purpose, or restarted: its old run left, and `MemberJoined` follows with the new `epoch` |
| `MemberUpdated` | `member`, `previous_meta` | A node changed its metadata |
| `NameConflict` | `member`, `other_addr` | Two nodes claim the same name, a live one or one that recently died or left; the one at `other_addr` is ignored |
| `UserMessage` | `sender`, `data`, `broadcast` | Another member sent this node `data`, or sent it to every member if `broadcast`; see Messages |
| `EventsLost` | `count` | Your consumer fell behind and `count` events were dropped |

Events never describe the local node, except `NameConflict`. Each call to `events()` is an independent subscription that starts at the moment of the call, not at the first `await`, so nothing is missed between `events()` and the loop. Every subscription buffers up to `event_buffer` events (1,024). When a slow consumer fills it, kinship drops the oldest events and your next read is `EventsLost(n)`; call `cluster.members()` and resync. The protocol never waits for you.

The iterator ends when the cluster closes. If the node stopped because of an internal error, it raises `KinshipClosed` instead.

## Metadata

Every node carries a few string tags that all other nodes can read, for facts such as role, zone, version, an application port or a load hint.

```python
cfg = kinship.Config.lan(..., meta={"role": "cache", "zone": "eu-1", "http": "10.0.0.7:8080"})

await cluster.update_meta(role="draining")       # merge
await cluster.set_meta({"role": "cache"})        # replace
zone = cluster.member("cache-03").meta["zone"]  # read anyone's
```

Encoded tags must fit in 512 bytes; a larger value raises `kinship.MetaTooLarge`, a `ValueError`. A change reaches every node in a few gossip rounds, under a second on `lan()` at 100 nodes, and other nodes see it as `MemberUpdated`. Ten quick updates cost one broadcast, because only the newest value is gossiped. Metadata is for facts about a node that every node should see; to tell other nodes something, send a message.

## Messages

Members can send each other bytes through the cluster itself, sealed with the cluster's key, so your nodes need no HTTP endpoint, port or credentials of their own between them to tell each other small things: a cache key to drop, a job handed over, a request whose answer is another message.

```python
await cluster.broadcast(b"invalidate user:42")    # every other member, once each
peer = cluster.member("cache-03")
await cluster.send(peer, b"rebuild shard 7")      # this run of cache-03
await cluster.send("cache-03", b"ping")           # whichever run of cache-03 is live

async for event in cluster.events():
    match event:
        case kinship.UserMessage(data=data, broadcast=True):
            cache.drop(data.removeprefix(b"invalidate "))
        case kinship.UserMessage(sender=m, data=data):
            await handle(m.name, data)
```

| Call | Data | How it travels |
| --- | --- | --- |
| `await cluster.send(member, data)` | up to `max_user_message`, 64 KiB | In one datagram when it fits, about 1.2 KB with the longest names; over a TCP connection above that |
| `await cluster.broadcast(data)` | up to `max_user_broadcast`, 256 bytes | Gossiped beside the membership rumours, which always go first, with the same retransmit rule |

Both return once the message is on its way, and the receiving nodes read it from `events()` as `UserMessage(sender, data, broadcast)`. A message is never delivered to the node that sent it, and `sender` is the member as the receiving node holds it, so its state can be suspect, or left for a message sent just before a leave.

Messages are best effort. kinship never acknowledges or retries one: a direct message in a datagram can be lost, one over TCP arrives unless its connection fails, and a broadcast reaches each member with the probability a membership rumour does. In the simulator, at 5 to 16 nodes with up to 2% packet loss, a broadcast reached 99.8% of the members and a datagram 99.0%; nothing repairs a miss, and a member that joins after a broadcast has spread never sees it. If a message must arrive, have the receiver answer and retry when no answer comes. Each message arrives at most once with encryption, in no particular order, except that direct messages to one member over TCP keep their order.

A `Member` names one run of a node, so `send(member, ...)` raises `kinship.NotAMember` once that node has restarted, died or left, and a restarted node drops anything meant for its previous run; a name alone goes to whichever run is live. Data over the limit raises `kinship.MessageTooLarge`, a `ValueError`, and after `leave()` both raise `KinshipClosed`.

Messages share `events()` with membership events, so a consumer that falls behind loses them as `EventsLost(n)`. Broadcasts share the gossip bandwidth with membership, so a cluster spreads only a handful of full-size broadcasts a second: each node holds at most `user_broadcast_queue` (256) to send and drops the rest. `stats()` counts `user_messages_sent`, `user_messages_received`, `user_messages_dropped` (for another node or run, from an unknown sender, or a failed connection), `user_messages_too_large` and `user_queue_overflow`.

Messages are authenticated to the cluster, not to their sender: any node that holds the key can send as any name, so treat `data` as input from the network. Nodes on kinship 0.3 or older ignore messages: a `send()` to one succeeds and nothing arrives, and they do not pass broadcasts on. Bulk data, and anything that must arrive, still belongs on a channel of your own; a message can say where to find it.

## Configuration

Pick a preset and override fields by name. Presets differ only in timing.

| Preset | Use it for | Changes from `lan()` |
| --- | --- | --- |
| `Config.lan()` | One datacenter or VPC | none |
| `Config.wan()` | Regions, edge sites, flaky links | probes every 5 s with a 3 s timeout, slower suspicion, push-pull every 60 s |
| `Config.local()` | Tests and demos | probes every 200 ms, loopback only, plaintext allowed, no TCP fallback ping |

```python
cfg = kinship.Config.wan(keys=[key], probe_interval=4.0)
cfg2 = cfg.replace(name="edge-12")    # configs are frozen
```

Durations are seconds as `float`, or a `datetime.timedelta`. Every field is validated when the config is built, so `kinship.ConfigError` (a `ValueError` that names the field) fires before any socket opens. The full field list lives in the [configuration reference](https://github.com/bytevader/kinship/blob/main/docs/design.md#configuration-surface); the ones you are most likely to set are `name`, `bind`, `advertise`, `cluster`, `seeds`, `meta`, `keys` and `event_buffer`. Use `bind="0.0.0.0:0"` to let the OS pick a port when several processes share a host, and read it back from `cluster.local.addr`.

A node holds at most `max_members` members (8,192, itself and recently departed members included) and learns at most `max_new_members_per_packet` new ones (1,024) from any one packet, so a member cannot flood the cluster with invented names. Names refused either way are counted in `cluster.stats().members_refused`. For a larger cluster, raise `max_members` on every node, and raise `max_new_members_per_packet` to the cluster's size if a node should learn all of it from its join rather than over its next few push-pulls.

## Encryption and keys

Every packet and stream is sealed with XChaCha20-Poly1305 and bound to your cluster label. A node without the key cannot read, join or inject anything, and packets it recorded are refused when it plays them back. Keys are 32 bytes, given as `bytes` or base64 text, and never logged. The threat model and what kinship does not protect against are in [SECURITY.md](https://github.com/bytevader/kinship/blob/main/SECURITY.md).

The first key in `keys` encrypts; every key in the list decrypts. Rotate with no downtime in three steps, each run on every node:

```python
await cluster.keyring.install(new_key)   # 1. every node can now read the new key
await cluster.keyring.use(new_key)       # 2. every node sends with it
await cluster.keyring.remove(old_key)    # 3. the old key is gone
print(cluster.keyring.key_ids())         # ['9f3a01c2', ...], safe to log
```

Finish each step on every node before starting the next. A node that falls behind looks dead to the others until it catches up, and `cluster.stats().decrypt_failures` counts what it dropped. Changing `keys` in your config and restarting nodes one by one does the same thing.

Inside the node, keys live in one place and are wiped from memory when removed or when the node closes. Python cannot wipe a `bytes` or `str`, so a key you hold as one stays in memory until Python reuses it: read it from a file or the environment just before you build the config, pass it straight in, and drop your references to it afterwards. The config keeps its keys until it is freed, and a cluster keeps its config.

Every node that holds the key is trusted with membership: it can report any member dead, and claim names and metadata. kinship keeps a member that died or left at its address until its record expires, so another node cannot take its name meanwhile, but it does not authenticate who a member is. If your application routes requests, assigns shards or sends data to members by name, check that a member's address, and any address you read from its metadata, is one you expect from your own configuration or service discovery, and escape names and metadata before you log or display them. Remove a node you no longer trust by rotating the key on every other node.

To run without encryption beyond loopback, set `insecure_plaintext=True`; kinship logs a warning at startup. Without encryption anyone who can reach the port can forge membership, so use it only where everything that can reach the port is trusted. A plaintext node sends only to members it knows, so a forged packet cannot make it send traffic to a third party.

## Logs, fork and threads

kinship runs its protocol on one background thread per process. Logs go to stderr until you call `kinship.log_to_python()`, which routes them to the `kinship` logger. Create clusters after `os.fork()`, never before; using a cluster in a forked child raises `KinshipClosed`. The API is safe to share across threads and works on free-threaded Python.

## Sync code: Flask, Django, Celery

`kinship.blocking` has the same API without `async`. The protocol still runs on kinship's own thread, so a slow request handler cannot make the node look dead.

```python
from kinship.blocking import Cluster

cluster = Cluster(cfg).start()           # binds and joins cfg.seeds, then returns
atexit.register(cluster.close)           # close() runs leave() first

owner = rendezvous.owner(key, cluster.members())
for event in cluster.events(timeout=1.0):   # yields None on each timeout tick
    ...
```

Methods that wait on the network (`join`, `set_meta`, `update_meta`, `send`, `broadcast`, `leave`, `close` and the keyring calls) block the calling thread and accept `timeout=`. Read events from a dedicated thread. Under Celery or gunicorn's prefork model, start the cluster in the worker after the fork (`worker_process_init`, `post_fork`), and give each worker its own port with `bind="0.0.0.0:0"`.

## Ownership with kinship.ring

`kinship.ring.Ring` answers "which node owns this key?" from the cluster's live view, over the members you pick, and tells you when the answer changes. It uses `kinship.rendezvous`, so every process that sees the same members agrees, and only the keys of a node that joins or goes move. Here is the sharded cache from the patterns below, built on it:

```python
import kinship
from kinship.ring import Ring

class ShardedCache:
    def __init__(self, cluster: kinship.Cluster, http: aiohttp.ClientSession) -> None:
        self.ring = Ring(cluster, where={"role": "cache"})   # or a predicate on Member
        self.cluster, self.http, self.data = cluster, http, {}

    async def get(self, key: str) -> bytes | None:
        node = self.ring.owner(key)
        if node.name == self.cluster.local.name:
            return self.data.get(key)
        async with self.http.get(f"http://{node.meta['http']}/cache/{key}") as resp:
            return await resp.read() if resp.status == 200 else None

    async def drop_moved_keys(self) -> None:
        me, before = self.cluster.local.name, self.ring.members()
        async for change in self.ring.changes():
            for key, (old, new) in self.ring.moved(self.data, before, change.members).items():
                if new is None or new.name != me:
                    del self.data[key]
            before = change.members
```

| Call | Returns | What it does |
| --- | --- | --- |
| `Ring(cluster, *, where=None)` | `Ring` | A ring over the members of a `kinship.Cluster` or `kinship.blocking.Cluster` whose metadata matches `where`, a dict such as `{"role": "api"}`, or for which `where(member)` is true. |
| `ring.owner(key, *, new_work=False)` | `Member` | The member that owns `key`, a `str` or `bytes`. Raises `LookupError` if no member matches. |
| `ring.replicas(key, n, *, new_work=False)` | `list[Member]` | The `n` best members for `key`, owner first: where to put replicas. |
| `ring.members()` | `tuple[Member, ...]` | The members on the ring now, sorted by name. |
| `ring.fence(key, *, new_work=False)` | `tuple[str, int]` | `(owner_name, epoch)` of the member that owns `key`, as `owner()` picks it: a fencing token for writes, below. Raises `LookupError` if no member matches. |
| `ring.changes()` | iterator of `RingChanged` | A new subscription to changes in the ring, built on `events()`. |
| `ring.moved(keys, before, after)` | `dict` | `{key: (old_owner, new_owner)}` for the keys whose owner differs between two `members()` snapshots, a restarted owner included; keys that stayed put are left out. |

Suspect members keep their keys, so a node that only stalls causes no reshuffle; dead and departed members own nothing. To route new work, such as a new session or a write, pass `new_work=True`: a suspect owner is skipped, and the work goes to the member that would take over its keys if it died. A member that restarts leaves the ring and joins it again under the same name, and its keys count as moved, since the new run has none of the old one's state.

### Fencing writes

A process can keep writing after the cluster has moved on: it stalled in a long pause, or it is the old run of a node that has already restarted. `ring.fence(key)` gives every write a token that such a writer cannot fake, because a restart raises the epoch:

```python
token = ring.fence(key, new_work=True)     # ("cache-03", 1790000000000), as writes route
if token[0] == cluster.local.name:
    await store.write(key, value, token)   # send the token with the write

# On a member that applies writes, a replica or the owner itself:
if ring.fence(key, new_work=True) != token:
    raise StaleOwner(token)                # another owner, or another run of it, in our view
```

A store outside the cluster keeps the token of the last write it accepted for each key and refuses a write with the same owner name and a lower epoch, in the same conditional write (compare-and-set) as the data, so that two writers cannot both pass the check. The token is only as current as membership: until views converge, after a partition for instance, two nodes can each hold a token that their own view confirms, so kinship fences off stale and restarted writers but is not a lock or consensus. Members on kinship older than 0.3 have epoch 0, which tells their runs apart only by name.

`changes()` yields `RingChanged(added, removed, suspect, recovered, members)` whenever a member on the ring joins or goes, becomes suspect or recovers, including a metadata update that makes a member start or stop matching `where`. `members` is the ring's new snapshot, ready for `moved()`. After `EventsLost` it re-reads the members and always yields, because lost events may have moved keys. On `kinship.Cluster` it is an async iterator; on `kinship.blocking.Cluster`, `for change in ring.changes(timeout=1.0)` blocks and yields `None` on each quiet tick, like `events()`. Events never describe the local node, so a change to this node's own metadata shows up with the next event.

## FastAPI

```bash
pip install "kinship[fastapi]"
```

`kinship.contrib.fastapi` runs a node for as long as your app runs, and hands it, or a ring over it, to your routes. The sharded cache as a FastAPI service:

```python
import os
from typing import Annotated
from urllib.parse import quote

import kinship
from fastapi import Depends, FastAPI, HTTPException
from fastapi.responses import RedirectResponse
from kinship.contrib.fastapi import cluster_lifespan, cluster_router, get_ring
from kinship.ring import Ring

def config() -> kinship.Config:   # runs in each worker, at startup
    return kinship.Config.lan(
        bind="0.0.0.0:0",
        keys=[os.environ["KINSHIP_KEY"]],
        seeds=["cache.default.svc.cluster.local:7946"],
        meta={"role": "cache", "url": os.environ["PUBLIC_URL"]},   # "https://cache-7.example.com"
    )

URLS = set(os.environ["CACHE_URLS"].split())   # the URLs you expect your nodes to have

app = FastAPI(lifespan=cluster_lifespan(config))
app.include_router(cluster_router(), dependencies=[Depends(admin_only)])
CacheRing = Annotated[Ring, Depends(get_ring(where={"role": "cache"}))]
data: dict[str, str] = {}

@app.get("/cache/{key}", response_model=None)
async def get(key: str, ring: CacheRing) -> dict[str, str] | RedirectResponse:
    node = ring.owner(key)
    if node.name != ring.cluster.local.name:
        if node.meta.get("url") not in URLS:   # members write their own metadata
            raise HTTPException(503)
        return RedirectResponse(node.meta["url"] + "/cache/" + quote(key, safe=""), 307)
    if key not in data:
        raise HTTPException(404)
    return {"key": key, "value": data[key]}
```

- `cluster_lifespan(config, *, leave_timeout=5.0)` goes in `FastAPI(lifespan=...)`. At startup it builds the config, calling `config` if it is a function, starts a `kinship.Cluster` and stores it on `app.state.cluster`; at shutdown it calls `leave()`, then `close()`. To run it next to a lifespan of your own, enter it from yours: `async with cluster_lifespan(config)(app): yield`.
- `get_cluster` and `get_ring(where=...)` are dependencies, for HTTP and WebSocket routes. `get_ring` builds its `Ring` once per `where`, not per request.
- `cluster_router(prefix="/cluster")` serves `GET /members` (name, address, state and metadata of each member) and `GET /stats` (this node's counters). They expose member addresses and metadata, so protect them, as above, or leave them off in production.
- Every server worker process runs its own node. With several workers (`uvicorn --workers 4`, gunicorn), each needs its own gossip port and name: pass a function as above, so each worker builds its own config after the fork, with `bind="0.0.0.0:0"` and no `name` (the default adds a random suffix to the host name) or one that includes the process id. The workers on one host share its HTTP port and so its `url`, and a request sent there reaches any of them, so keep per-node state such as this cache to one worker per host.
- kinship's own traffic is not HTTP: gossip runs over UDP and TCP, encrypted with XChaCha20-Poly1305. Whether your app serves HTTP or HTTPS is up to your server or proxy; put the scheme in the URL you publish in metadata, as above.
- `httpx.ASGITransport` does not run the lifespan; in tests, enter `app.router.lifespan_context(app)` around the client. [`examples/fastapi_app.py`](https://github.com/bytevader/kinship/blob/main/examples/fastapi_app.py) runs three apps in one process that way.

## Metrics

`kinship.metrics.prometheus_text(cluster)` exports Prometheus text without extra dependencies,
from either `kinship.Cluster` or `kinship.blocking.Cluster`:

```python
from kinship.metrics import prometheus_text

print(prometheus_text(cluster))  # a started cluster
```

Every `Stats` field is a `kinship_<field>_total` counter, except `local_health`, which is
the `kinship_local_health` gauge. All samples have a `node` label with the local member's
name. Counters reset when the node restarts. `kinship_members` gauges have a `state` label
(`alive`, `suspect`, `dead`, `left`), counted from `cluster.members()`. That API excludes
dead and left tombstones, so those two gauges are zero.

With the FastAPI integration, opt into an HTTP endpoint (disabled by default):

```python
app.include_router(cluster_router(metrics=True))  # GET /cluster/metrics
# Use cluster_router(prefix="", metrics=True) for GET /metrics.
```

Configure Prometheus to scrape each node's HTTP server, for example:

```yaml
scrape_configs:
  - job_name: kinship
    metrics_path: /cluster/metrics
    static_configs:
      - targets: ["node-1:8000", "node-2:8000"]
```

Expose this endpoint only to trusted scrapers. Give each node its own scrape target;
with several server workers sharing an HTTP port, a scrape can reach a different node
each time.

The [five-node Compose demo](deploy/compose/README.md#prometheus-and-grafana) includes
an optional Prometheus and Grafana override with a provisioned cluster dashboard:

```sh
cd deploy/compose
docker compose -f docker-compose.yml -f docker-compose.monitoring.yml up --build -d
```

Open Grafana at <http://localhost:3000/d/kinship-cluster> and the demo at
<http://localhost:8080> to watch metrics while injecting faults.

## Examples

Runnable scripts in [examples/](https://github.com/bytevader/kinship/blob/main/examples), each a single file with nothing to install beyond kinship, except `fastapi_app.py`, which needs `kinship[fastapi]` and httpx:

| Script | What it shows |
| --- | --- |
| [`quickstart.py`](https://github.com/bytevader/kinship/blob/main/examples/quickstart.py) | Three terminals, one cluster: nodes joining, leaving and crashing, with timestamps |
| [`sharded_keys.py`](https://github.com/bytevader/kinship/blob/main/examples/sharded_keys.py) | Rendezvous ownership of keys, and that only a departed node's keys move, after a leave and after a crash |
| [`failover_job.py`](https://github.com/bytevader/kinship/blob/main/examples/failover_job.py) | A job that runs on exactly one node and moves on its own when that node dies |
| [`sync_workers.py`](https://github.com/bytevader/kinship/blob/main/examples/sync_workers.py) | The blocking API and metadata: routing around a node that marks itself as draining |
| [`messages.py`](https://github.com/bytevader/kinship/blob/main/examples/messages.py) | A broadcast every other node hears once, and a request answered with a direct message too large for a datagram |
| [`fastapi_app.py`](https://github.com/bytevader/kinship/blob/main/examples/fastapi_app.py) | A sharded cache as three FastAPI apps: `cluster_lifespan`, `get_ring`, redirects to a key's owner, and the keys that move when a node leaves |
| [`deploy/compose/`](https://github.com/bytevader/kinship/tree/main/deploy/compose) | Five FastAPI cache nodes and a live dashboard: key ownership, load, stalls, crashes, draining and opt-in Docker partitions |

```bash
python examples/sharded_keys.py
```

```text
node-c crashes
  node-a sees MemberSuspect(node-c) after 0.5 s
  node-a sees MemberDead(node-c) after 1.1 s
```

The patterns below show the same building blocks inside real services. Three of the four decide which node owns a key with `kinship.rendezvous.owner(key, members)`. It uses rendezvous (highest random weight) hashing with a stable hash, so every process agrees on the owner, and only the keys of a node that joins or leaves move. Never use Python's built-in `hash()` for this: it differs between processes.

### Sharded cache

Every node holds the keys it owns and fetches the rest from their owner over HTTP. The owner's HTTP address travels in metadata.

```python
import kinship
from kinship.rendezvous import owner

class ShardedCache:
    def __init__(self, cluster: kinship.Cluster, http: aiohttp.ClientSession) -> None:
        self.cluster, self.http, self.data = cluster, http, {}

    async def get(self, key: str) -> bytes | None:
        node = owner(key, self.cluster.members())
        if node.name == self.cluster.local.name:
            return self.data.get(key)
        async with self.http.get(f"http://{node.meta['http']}/cache/{key}") as resp:
            return await resp.read() if resp.status == 200 else None

    async def drop_moved_keys(self) -> None:
        async for event in self.cluster.events():
            if isinstance(event, (kinship.MemberJoined, kinship.EventsLost)):
                me, members = self.cluster.local.name, self.cluster.members()
                for key in [k for k in self.data if owner(k, members).name != me]:
                    del self.data[key]
```

`members()` includes suspect nodes on purpose. A node that is only slow keeps its keys, so a brief stall does not reshuffle the cache; its keys move only once it is dead or has left.

### Single-run cron

Each scheduled job runs on the one alive node that owns the job's name, so jobs spread across the fleet and move when their node dies.

```python
MIN_NODES = 3   # below this, assume we have not found the cluster yet

async def nightly(cluster: kinship.Cluster) -> None:
    while True:
        await sleep_until(hour=2)
        alive = [m for m in cluster.members() if m.state is kinship.State.ALIVE]
        if len(alive) < MIN_NODES:
            continue
        if owner("nightly-report", alive).name == cluster.local.name:
            await build_report(run_id=today())   # must be idempotent on run_id
```

This is best effort, not a lock. During a partition, or for the second or two while views disagree, two nodes can both believe they own a job, so the job must be safe to run twice or take a database lock keyed on `run_id`. The `MIN_NODES` check stops a node that started cut off from the cluster from running every job alone.

### WebSocket room routing

All sockets for one chat room live on the same server, so a message to the room never crosses the network. A client that lands on the wrong server is told where to go.

```python
rooms: dict[str, set[ServerConnection]] = defaultdict(set)

async def handler(ws: ServerConnection) -> None:
    room = ws.request.path.removeprefix("/rooms/")
    node = owner(room, cluster.members())
    if node.name != cluster.local.name:
        await ws.close(4001, node.meta["ws"])   # client reconnects to this URL
        return
    rooms[room].add(ws)
    try:
        async for message in ws:
            broadcast(rooms[room], message)
    finally:
        rooms[room].discard(ws)

async def rebalance() -> None:
    async for event in cluster.events():
        if isinstance(event, (kinship.MemberJoined, kinship.MemberDead,
                              kinship.MemberLeft, kinship.EventsLost)):
            members = cluster.members()
            for room, sockets in list(rooms.items()):
                target = owner(room, members)
                if target.name != cluster.local.name:
                    for ws in list(sockets):
                        await ws.close(4001, target.meta["ws"])
```

Each server advertises its public URL with `meta={"ws": "wss://ws-3.example.com"}`. When a server joins, only the rooms it now owns move to it; when one dies, its clients reconnect and land on the new owners.

### Edge fleet over flaky links

Store gateways join a cluster whose seeds are two regional hubs. A dashboard on each hub shows which stores are up, degraded or down.

```python
cfg = kinship.Config.wan(
    name=os.environ["STORE_ID"],                         # "store-0412", stable across reboots
    bind="100.64.12.7:7946",                             # overlay address, see below
    keys=[Path("/etc/kinship/key").read_text().strip()],
    seeds=["100.64.0.1:7946", "100.64.0.2:7946"],
    cluster="stores-prod",
    meta={"site": "lisbon-2", "sw": VERSION},
)

status: dict[str, kinship.State] = {}

async def track(cluster: kinship.Cluster) -> None:
    events = cluster.events()                      # subscribe first, then read the snapshot
    status.update({m.name: m.state for m in cluster.members()})
    async for event in events:
        match event:
            case kinship.EventsLost():
                status.update({m.name: m.state for m in cluster.members()})
            case kinship.NameConflict(member=m, other_addr=addr):
                alert(f"two boxes claim {m.name}: {m.addr} and {addr}")
            case kinship.UserMessage():
                pass
            case _:
                status[event.member.name] = event.member.state   # SUSPECT shows as degraded
```

`wan()` probes less often and waits longer before declaring a store dead, and Lifeguard keeps a store whose CPU is pegged from being reported dead. A store that drops off for an hour shows as dead; when its link returns it keeps retrying its seeds in the background and rejoins without a restart. Every node must reach every other node's advertised address over UDP and TCP, so stores behind NAT should run on an overlay network such as WireGuard or Tailscale and bind to that address.

## Rust

The Python package is a thin layer over two Rust crates you can use directly.

### kinship: the tokio handle

```rust
use kinship::{Cluster, Config, Event, Key};
use std::time::Duration;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let cfg = Config::lan()
        .with_name("worker-17")
        .with_keys([Key::from_base64(&std::env::var("KINSHIP_KEY")?)?])
        .with_seeds(["10.0.0.5:7946".parse()?])
        .with_seed_hosts(["kinship.default.svc.cluster.local:7946".parse()?]);
    let cluster = Cluster::start(cfg).await?;     // binds, joins seeds
    let mut events = cluster.events();
    while let Some(event) = events.recv().await {
        match event {
            Event::MemberDead(m) | Event::MemberLeft(m) => println!("{} is gone", m.name),
            Event::UserMessage { sender, data, .. } => println!("{}: {data:?}", sender.name),
            Event::EventsLost(_) => println!("resync: {} members", cluster.members().len()),
            _ => {}
        }
    }
    cluster.leave(Duration::from_secs(5)).await?;
    Ok(())
}
```

The Rust and Python APIs share names: `Cluster`, `Config::lan()`, `members()`, `events()`, `set_meta()`, `send()`, `broadcast()`, `keyring()`, `leave()`, `close()`. Rust's `send` takes a `&Member` and `send_to` a name. The Rust keyring has `install`, `use_key` (`use` is a keyword), `remove` and `key_ids`. `Config` is built with `with_*` setters so new fields can be added without a breaking release, and it is validated by `Cluster::start`. `Event` is `#[non_exhaustive]`. `with_seeds` takes addresses and `with_seed_hosts` takes `Seed`s, parsed from `host:port` with a DNS name or an IP address; `join_seeds` is `join` for them.

The `kinship-agent` example joins a cluster from seeds on the command line and logs every event until Ctrl-C, when it leaves:

```bash
cargo run -p kinship --example kinship-agent -- --bind 127.0.0.1:7946
cargo run -p kinship --example kinship-agent -- --bind 127.0.0.1:7947 localhost:7946
```

### kinship-core: sans-IO SWIM

`kinship-core` is the whole protocol with no sockets, clocks, threads or randomness of its own. You feed it bytes and the time, and it tells you what to send and when to wake it. Use it to run SWIM on your own runtime, an embedded event loop or a simulator.

```rust
use kinship_core::{Command, Config, Event, Identity, Instant, Node, StreamEvent, Transmit};

// Both from the OS RNG: the seed for protocol choices, 32 bytes for the nonces' ChaCha20 key.
// The epoch is the wall clock's milliseconds at start, so a restart reads as a new life.
let me = Identity::new("a", my_addr)?.with_epoch(unix_millis);
let mut node = Node::new(cfg, me, Instant::ZERO, seed, &nonce_key)?;
let join = node.command(now, Command::Join { seeds: vec![seed_addr] });   // a CommandId

// Feed every input with the current time:
node.handle_datagram(now, from, &buf);
node.handle_stream(now, conn, StreamEvent::Frame(&frame));
node.handle_stream(now, conn, StreamEvent::Failed);   // connect refused, timeout, reset
node.handle_timeout(now);

// Then drain the outputs:
while let Some(t) = node.poll_transmit() {
    match t {
        Transmit::Datagram { to, payload } => udp_send(to, &payload),
        Transmit::Connect { conn, to } => tcp_connect(conn, to),
        Transmit::Stream { conn, frame } => tcp_send(conn, &frame),
        Transmit::Close { conn } => tcp_close(conn),
    }
}
while let Some(event) = node.poll_event() {
    match event {
        Event::CommandDone { id, result } if id == join => println!("joined: {result:?}"),
        other => handle(other),
    }
}
let wake_at: Option<Instant> = node.poll_timeout();
```

The same seed, nonce key and inputs always give the same outputs, byte for byte, which is how `kinship-sim` replays any failure from its seed, from which it derives both. Encryption happens inside the core, so the bytes in `payload` and `frame` are already sealed.

## Learn more

- [Design](https://github.com/bytevader/kinship/blob/main/docs/design.md): the protocol state machine, wire format, threading model and failure modes.
- [Benchmarks](https://github.com/bytevader/kinship/blob/main/docs/results/detection.md): detection latency against memberlist, and the [timing preset sweep](https://github.com/bytevader/kinship/blob/main/docs/results/tuning.md).
- [Security](https://github.com/bytevader/kinship/blob/main/SECURITY.md): the threat model, the audit findings and how to report a vulnerability.
- [Changelog](https://github.com/bytevader/kinship/blob/main/CHANGELOG.md).

## License

Licensed under either of [Apache License 2.0](https://github.com/bytevader/kinship/blob/main/LICENSE-APACHE) or [MIT](https://github.com/bytevader/kinship/blob/main/LICENSE-MIT), at your option.

