scriptling.net.gossip

Gossip protocol cluster membership and messaging, with automatic failure detection, metadata propagation, tag-based routing, node groups, leader election, encryption, and compression.

In this section

• Node Groups & Leader Election: create_node_group(), create_leader_election(), and the objects they return.

Overview

The scriptling.net.gossip library implements a gossip protocol for decentralized cluster management. Nodes automatically discover each other, detect failures, and propagate metadata across the cluster. It supports both unreliable (UDP) and reliable (TCP) messaging, with optional AES encryption and Snappy compression. Advanced features include request/reply messaging, metadata-criteria-based node groups, and quorum-based leader election with optional metadata filtering.

Handler concurrency

All registered callbacks: handle(), handle_with_reply(), on_state_change(), on_metadata_change(), on_gossip_interval(), node-group on_node_added/on_node_removed, and leader-election on_event(): fire automatically as messages and events arrive. They run on gossip’s internal goroutines under the per-environment interpreter lock (GIL). There is no event pump and no wait() loop: handlers invoke themselves.

Because the GIL serializes script execution, a handler never runs concurrently with the rest of your script, so you do not need locks around shared state. Handlers interleave with your script only at its blocking points (or yield_now()), which is memory-safe.

A long-running node just stays alive: do work, or sleep: and handlers fire as traffic arrives:

cluster.handle(gossip.MSG_USER, on_message)

while True:
    time.sleep(1)   # keep the process alive; handlers fire as messages arrive

Ordering: gossip messages carry Hybrid Logical Clock (HLC) timestamps, so application-level ordering is resolved at the data layer: handler delivery order is not significant and may differ from arrival order.

A single script can both call send_request() and serve its own handle_with_reply() responder: send_request() releases the GIL while it blocks, letting an incoming request’s handler run and reply.

Available Functions

Constants

Cluster Object

The create() function returns a cluster object with the following methods.

create_node_group() and create_leader_election(), and the objects they return, are documented on Node Groups & Leader Election.

Functions

create(bind_addr="127.0.0.1:8000", ...)

Creates a gossip cluster node.

Parameters:

• bind_addr (str, optional): Address to bind to. Default: "127.0.0.1:8000".
• node_id (str, optional): Unique node ID. Default: "" (auto-generated).
• advertise_addr (str, optional): Address to advertise to peers. Default: same as bind_addr.
• encryption_key (str, optional): AES encryption key, 16, 24, or 32 bytes. Default: "" (no encryption). Socket transport only; the "http" transport relies on HTTPS instead and rejects a key.
• tags (list, optional): Tags for tag-based message routing. Default: [].
• compression (bool, optional): Enable Snappy compression. Default: False.
• bearer_token (str, optional): Authentication bearer token. Default: "".
• app_version (str, optional): Application version for compatibility checks. Default: "".
• transport (str, optional): Transport type, "socket" or "http". Default: "socket". With "http", the node serves the gossip endpoint over HTTP on bind_addr; advertise_addr defaults to http://<bind_addr> (set it to an https:// URL behind a TLS proxy; a value without a scheme gets http://), and peers join it by host:port as with sockets. When bind_addr listens on all interfaces (for example 0.0.0.0:8000), set advertise_addr to the address peers should use.
• compress_min_size (int, optional): Minimum message size for compression. Default: 256.
• gossip_interval (str, optional): Gossip interval duration. Default: "5s".
• gossip_max_interval (str, optional): Maximum gossip interval. Default: "20s".
• metadata_gossip_interval (str, optional): Metadata gossip interval. Default: "500ms".
• state_gossip_interval (str, optional): State exchange interval. Default: "45s".
• fan_out_multiplier (float, optional): Fan-out scaling factor. Default: 1.0.
• ttl_multiplier (float, optional): TTL scaling factor. Default: 1.0.
• state_exchange_multiplier (float, optional): State exchange scaling. Default: 0.8.
• force_reliable_transport (bool, optional): Force TCP for all messages. Default: False.
• prefer_ipv6 (bool, optional): Prefer IPv6 for DNS resolution. Default: False.
• node_cleanup_interval (str, optional): Dead node cleanup interval. Default: "20s".
• node_retention_time (str, optional): How long to keep dead nodes. Default: "1h".
• leaving_node_timeout (str, optional): Timeout before moving a leaving node to dead. Default: "30s".
• health_check_interval (str, optional): Health check interval. Default: "2s".
• suspect_timeout (str, optional): Time before marking a node suspect. Default: "1.5s".
• suspect_retry_interval (str, optional): Suspect node retry interval. Default: "1s".
• dead_node_timeout (str, optional): Time before marking a suspect node dead. Default: "15s".
• peer_recovery_interval (str, optional): Peer recovery check interval. Default: "30s".
• insecure_skip_verify (bool, optional): Skip TLS verification for the "http" transport. Default: False.

Returns: Cluster: a cluster node object.

import scriptling.net.gossip as gossip

cluster = gossip.create(
    bind_addr="127.0.0.1:8000",
    tags=["web"],
    encryption_key="0123456789abcdef"
)

cluster.start()

Starts the cluster node. Begins the transport, health monitoring, and gossip routines.

Parameters: None

Returns: None

cluster.start()

cluster.join(peers)

Joins an existing cluster by connecting to known peers.

Parameters:

• peers (str or list): One or more peer addresses to join.

Returns: None

cluster.join("127.0.0.1:8001")
cluster.join(["127.0.0.1:8001", "127.0.0.1:8002"])

cluster.leave()

Gracefully leaves the cluster. Other nodes are notified.

Parameters: None

Returns: None

cluster.leave()

cluster.stop()

Stops the cluster and cleans up all resources. Streams this node opened with open_stream() are closed.

Parameters: None

Returns: None

cluster.stop()

cluster.send(message_type, data, reliable=False)

Broadcasts a message to all cluster nodes.

Parameters:

• message_type (int): Message type. Must be >= 128 (see MSG_USER).
• data (str, int, float, list, or dict): Message payload.
• reliable (bool, optional): Use reliable TCP transport instead of UDP. Default: False.

Returns: None

cluster.send(128, "Hello cluster!")
cluster.send(128, {"key": "value"}, reliable=True)

cluster.send_tagged(tag, message_type, data, reliable=False)

Sends a tagged message. Only delivered to nodes that have the matching tag.

Parameters:

• tag (str): Tag for routing.
• message_type (int): Message type. Must be >= 128.
• data (str, int, float, list, or dict): Message payload.
• reliable (bool, optional): Use reliable transport. Default: False.

Returns: None

cluster.send_tagged("web", 128, "Hello web nodes!")

cluster.send_to(node_id, message_type, data, reliable=False)

Sends a direct message to a specific node.

Parameters:

• node_id (str): Target node UUID.
• message_type (int): Message type. Must be >= 128.
• data (str, int, float, list, or dict): Message payload.
• reliable (bool, optional): Use reliable transport. Default: False.

Returns: None

target = cluster.nodes()[0]
cluster.send_to(target["id"], 128, "Direct message!")

cluster.send_request(node_id, message_type, data)

Sends a request to a specific node and waits for a reply. Releases the interpreter lock while waiting, so the calling script’s own handle_with_reply() responders can still run.

Parameters:

• node_id (str): Target node UUID.
• message_type (int): Message type. Must be >= 128.
• data (str, int, float, list, or dict): Message payload.

Returns: the reply payload from the target node (type depends on what the remote handler returned).

reply = cluster.send_request(target_id, 128, {"cmd": "ping"})
print(reply)

cluster.handle(message_type, handler)

Registers a handler for a specific message type.

Parameters:

• message_type (int): Message type to handle. Must be >= 128.
• handler (callable): Function called with a message dict containing type (int), sender (dict with id, addr, state, metadata, tags), and payload (the decoded message payload).

Returns: None

def on_message(msg):
    print(f"From {msg['sender']['id']}: {msg['payload']}")

cluster.handle(128, on_message)

cluster.handle_with_reply(message_type, handler)

Registers a request/reply handler. The handler’s return value is sent back as the reply.

Parameters:

• message_type (int): Message type to handle. Must be >= 128.
• handler (callable): Function called with the same message dict as handle(); its return value becomes the reply payload.

Returns: None

def on_request(msg):
    return {"status": "ok", "echo": msg["payload"]}

cluster.handle_with_reply(128, on_request)

cluster.open_stream(node_id, message_type, data)

Sends data to the node’s handle_stream() handler and returns a stream to read the reply from. Unlike send_request(), the reply is not limited to one packet: use streams for files, state transfers and other bulk data.

The stream has read(size=-1) and readline(), which return bytes (b'' at the end), and close(). Use it in a with statement to close it automatically. An error raised by the handler is raised by the read that reaches it.

cluster.handle_stream(message_type, handler)

Registers handler(msg, writer) for open_stream() requests. msg is the same dict handle() receives. Write the reply with writer.write(data), where data is str (sent as UTF-8) or bytes; returning ends the reply, and raising sends the error to the caller. The writer is only valid while the handler runs.

import time
import scriptling.net.gossip as gossip

server = gossip.create(bind_addr="127.0.0.1:17946")

def send_report(msg, writer):
    for i in range(msg["payload"]["rows"]):
        writer.write(f"row {i}\n")

server.handle_stream(300, send_report)
server.start()

client = gossip.create(bind_addr="127.0.0.1:17947")
client.start()
client.join(["127.0.0.1:17946"])
while client.get_node(server.node_id()) is None:
    time.sleep(0.05)

with client.open_stream(server.node_id(), 300, {"rows": 3}) as stream:
    print(stream.readline())   # b'row 0\n'
    print(stream.read())       # b'row 1\nrow 2\n'

client.stop()
server.stop()

cluster.unhandle(message_type)

Removes a previously registered message handler.

Parameters:

• message_type (int): Message type to unregister.

Returns: bool: True if a handler was removed.

cluster.unhandle(128)

cluster.on_state_change(handler)

Registers a handler called when any node changes state.

Parameters:

• handler (callable): Function called as handler(node_id, new_state). new_state is one of "alive", "suspect", "dead", "leaving".

Returns: None

def on_change(node_id, state):
    print(f"Node {node_id} is now {state}")

cluster.on_state_change(on_change)

cluster.on_metadata_change(handler)

Registers a handler called when any remote node’s metadata changes.

Parameters:

• handler (callable): Function called as handler(node_dict).

Returns: None

def on_meta(node):
    print(f"Node {node['id']} metadata: {node['metadata']}")

cluster.on_metadata_change(on_meta)

cluster.on_gossip_interval(handler)

Registers a handler called at every gossip interval.

Parameters:

• handler (callable): Function called with no arguments at each interval.

Returns: None

def on_tick():
    print(f"Alive: {cluster.num_alive()}")

cluster.on_gossip_interval(on_tick)

cluster.nodes()

Gets all known nodes in the cluster.

Parameters: None

Returns: list: list of node dicts, each with id, addr, state, metadata, tags.

for node in cluster.nodes():
    print(f"{node['id']}: {node['state']} at {node['addr']}")

cluster.alive_nodes()

Gets all nodes currently in the alive state.

Parameters: None

Returns: list: list of node dicts.

cluster.nodes_by_tag(tag)

Gets all nodes that have a specific tag.

Parameters:

• tag (str): Tag to filter by.

Returns: list: list of node dicts with the matching tag.

web_nodes = cluster.nodes_by_tag("web")

cluster.get_node(node_id)

Gets a specific node by ID.

Parameters:

• node_id (str): Node UUID.

Returns: dict: node dict, or None if not found.

node = cluster.get_node("some-uuid")
if node:
    print(node["state"])

cluster.local_node()

Gets the local node’s information.

Parameters: None

Returns: dict: node dict with id, addr, state, metadata, tags.

cluster.num_nodes()

Gets the total number of known nodes.

Parameters: None

Returns: int

cluster.num_alive()

Gets the number of alive nodes.

Parameters: None

Returns: int

cluster.num_suspect()

Gets the number of suspect nodes.

Parameters: None

Returns: int

cluster.num_dead()

Gets the number of dead nodes.

Parameters: None

Returns: int

cluster.node_id()

Gets the local node’s unique UUID.

Parameters: None

Returns: str

cluster.is_local(node_id)

Checks if a node ID refers to the local node.

Parameters:

• node_id (str): Node UUID to check.

Returns: bool

if cluster.is_local(node["id"]):
    print("That's me!")

cluster.candidates()

Gets a random subset of nodes for gossiping.

Parameters: None

Returns: list: list of node dicts.

cluster.set_metadata(key, value)

Sets a local node metadata value. Metadata is automatically gossiped to other nodes.

Parameters:

• key (str): Metadata key.
• value (str, int, float, or bool): Metadata value.

Returns: None

cluster.set_metadata("role", "worker")
cluster.set_metadata("version", 2)

cluster.get_metadata(key)

Gets a local metadata value.

Parameters:

• key (str): Metadata key.

Returns: str: the value, or None if not set.

cluster.all_metadata()

Gets all local metadata.

Parameters: None

Returns: dict

cluster.delete_metadata(key)

Deletes a metadata key.

Parameters:

• key (str): Metadata key to delete.

Returns: None

Security Considerations

This is an extended library, requiring registration in Go, see Library Registration.

scriptling.net.gossip opens raw UDP/TCP sockets to bind, join, and exchange traffic with other cluster nodes, and can both send and receive arbitrary script-supplied payloads over the network. The library itself does not restrict which hosts or ports a script can bind to or contact: that is the embedder’s responsibility, typically enforced with OS-level firewalling, network namespacing, or by controlling which addresses are reachable from the process. It is also not wired to --network-policy, so a configured policy does not constrain the peers join() contacts. Use encryption_key and bearer_token to protect traffic and authenticate peers when running across untrusted networks. See Security Considerations for a full breakdown of network-enabled libraries.

Examples

Basic Cluster

import scriptling.net.gossip as gossip

cluster = gossip.create(bind_addr="127.0.0.1:8000")
cluster.start()
cluster.join(["127.0.0.1:8001"])

cluster.handle(128, lambda msg: print(msg["payload"]))
cluster.send(128, "Hello!")

cluster.stop()

Three-Node Cluster with Tags

import scriptling.net.gossip as gossip

# Node 1 - web server
web = gossip.create(bind_addr="127.0.0.1:8000", tags=["web"])
web.start()

# Node 2 - worker
worker = gossip.create(bind_addr="127.0.0.1:8001", tags=["worker"])
worker.start()
worker.join(["127.0.0.1:8000"])

# Node 3 - both
hybrid = gossip.create(bind_addr="127.0.0.1:8002", tags=["web", "worker"])
hybrid.start()
hybrid.join(["127.0.0.1:8000"])

# Only web-tagged nodes receive this
web.send_tagged("web", 128, "Hello web nodes!")

Metadata and State Monitoring

import scriptling.net.gossip as gossip

cluster = gossip.create(bind_addr="127.0.0.1:8000")
cluster.set_metadata("role", "leader")
cluster.set_metadata("version", 2)
cluster.start()

cluster.on_state_change(lambda node_id, state:
    print(f"Node {node_id} -> {state}")
)

cluster.on_metadata_change(lambda node:
    print(f"Node {node['id']} metadata changed")
)

cluster.join(["127.0.0.1:8001"])

# Check other nodes' metadata
for node in cluster.alive_nodes():
    print(f"{node['id']}: {node['metadata']}")

Request/Reply Messaging

import scriptling.net.gossip as gossip

cluster = gossip.create(bind_addr="127.0.0.1:8000")
cluster.start()

# Register a handler that returns a reply
cluster.handle_with_reply(128, lambda msg: {"echo": msg["payload"]})

# Send a request and wait for the reply
for node in cluster.alive_nodes():
    reply = cluster.send_request(node["id"], 128, "ping")
    print(f"Reply from {node['id']}: {reply}")

Encrypted Cluster

import scriptling.net.gossip as gossip

cluster = gossip.create(
    bind_addr="0.0.0.0:8000",
    encryption_key="32-byte-key-here-1234567890ab",
    compression=True,
    bearer_token="secret-token"
)
cluster.start()
cluster.join(["10.0.0.1:8000"])

Notes

• Message types 0-127 are reserved for internal protocol use; user message types must be >= 128 (use the MSG_USER constant).
• reliable=True uses TCP for guaranteed delivery.
• Metadata is eventually consistent across the cluster.
• Always call stop() to properly clean up resources.

See Also

• scriptling.net.unicast: direct point-to-point UDP/TCP messaging
• scriptling.net.multicast: one-to-many UDP group messaging
• scriptling.net.resolve: DNS and SRV record resolution
• Security Guide: full risk breakdown across all libraries