kinship 0.2.0

Know which of your processes are alive, without a coordinator: SWIM membership and failure detection with Lifeguard, in Rust, for asyncio Python and tokio.

Source code: https://github.com/bytevader/kinship

Design and benchmarks: design, detection latency against memberlist

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).
• 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

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). No Rust toolchain is needed.

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:

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:

python demo.py 7946
python demo.py 7947 7946
python demo.py 7948 7946

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

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

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

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

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:

Events

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.EventsLost(count=n):
            ring.rebuild(cluster.members())

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.

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. Larger application state belongs in your own channel; use metadata to find it.

Configuration

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

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; 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.

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

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.

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, 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:

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

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.

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

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:

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 runs three apps in one process that way.

Examples

Runnable scripts in examples/, each a single file with nothing to install beyond kinship, except fastapi_app.py, which needs kinship[fastapi] and httpx:

python examples/sharded_keys.py

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.

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.

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.

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.

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 _:
                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

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::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(), keyring(), leave(), close(). 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 Seeds, 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:

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.

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.
let mut node = Node::new(cfg, Identity::new("a", my_addr)?, 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: the protocol state machine, wire format, threading model and failure modes.
• Benchmarks: detection latency against memberlist, and the timing preset sweep.
• Security: the threat model, the audit findings and how to report a vulnerability.
• Changelog.

License

Licensed under either of Apache License 2.0 or MIT, at your option.