Systems

vcluster

Distributed computing across processes and machines over TCP, written in Venutian.

Released v1.0.0 · 2026-09-28

Import

use "stdlib/cluster.ven"

Distributed computing across processes and machines over TCP, with framed messages, heartbeats, failure detection and cluster wide locks, written in Venutian.

vcluster.ven

print default_chunk_bytes()
print heartbeat_interval_seconds()

Reference (65 functions)

Functions
Wire framing
Message shapes
Functions
Remote tasks
Distributed shared state
Distributed memory (AI-agent memory, across real machines)

API reference

`hex_digits()`

The sixteen hexadecimal digit characters, lowercase, in index order.

hex_digits()[10:11]

`default_chunk_bytes()`

The chunk size, in characters, used for each tcp_receive call while reassembling a frame.

default_chunk_bytes()

`length_header_chars()`

The fixed width, in hex characters, of the frame length header this module's wire protocol prepends to every message (see this file's header comment, "Wire protocol").

length_header_chars()

`max_frame_payload_bytes()`

max_frame_payload_bytes()

`heartbeat_interval_seconds()`

How often heartbeat_loop sends a fresh heartbeat message to every known peer.

heartbeat_interval_seconds()

`suspected_after_seconds()`

How long a member can go without a heartbeat before nodes_view reports it as "suspected" rather than "up".

suspected_after_seconds()

`down_after_seconds()`

How long a member can go without a heartbeat before nodes_view reports it as "down".

down_after_seconds()

`lock_poll_seconds()`

The backoff between retries while acquire_local_lock spin-waits for a distributed lock already held by someone else.

lock_poll_seconds()

`transient_retry_count()`

transient_retry_count()

`transient_retry_backoff_seconds()`

The sleep between each transient-failure retry round_trip_attempts/ send_only_attempts make.

transient_retry_backoff_seconds()

Wire framing

`hex_digit_char(d)`

A single hex digit character for the nibble value d.

hex_digit_char(10)

`hex_char_value(c)`

The nibble value of a single hex digit character.

hex_char_value("a")

`byte_to_hex2(n)`

A byte value rendered as two lowercase hex characters.

byte_to_hex2(255)

`hex2_to_byte(text2)`

hex2_to_byte("ff")

`pack_length_header(n)`

pack_length_header(12)

`unpack_length_header(header_text)`

unpack_length_header("0000000c")

`pack_hex4(code)`

pack_hex4(65)

`surrogate_pair_escape(code)`

surrogate_pair_escape(128512)

`ascii_escape_json(text)`

ascii_escape_json(json_encode(new_map()))

`send_all(sock, text)`

send_all(tcp_connect("127.0.0.1", 9100), "hello")

`send_message(sock, message_map)`

send_message(tcp_connect("127.0.0.1", 9100), new_heartbeat_message("node-1"))

`receive_exact(sock, n, max_chunk)`

receive_exact(tcp_connect("127.0.0.1", 9100), 8, default_chunk_bytes())

`receive_frame(sock, max_chunk)`

receive_frame(tcp_connect("127.0.0.1", 9100), default_chunk_bytes())

`round_trip_attempts(host, port, message_map, retries_left)`

round_trip_attempts("127.0.0.1", 9100, new_heartbeat_message("node-1"), transient_retry_count())

`round_trip(host, port, message_map)`

Sends message_map to host:port and gives back the decoded reply, retrying a transient connection failure up to transient_retry_count times with transient_retry_backoff_seconds between attempts (see round_trip_attemptss own comment for exactly which failures this retries and why).

round_trip("127.0.0.1", 9100, new_heartbeat_message("node-1"))

`send_only_attempts(host, port, message_map, retries_left)`

See round_trip_attemptss own comment just above: the identical recursion-to-loop fix, for the identical reason (this is send_onlys own retry loop, not round_trips).

send_only_attempts("127.0.0.1", 9100, new_heartbeat_message("node-1"), transient_retry_count())

`send_only(host, port, message_map)`

Sends message_map to host:port and closes the connection without waiting for any reply, retrying a transient connection failure the same way round_trip does (see send_only_attempts and round_trip_attemptss own comment).

send_only("127.0.0.1", 9100, new_heartbeat_message("node-1"))

Message shapes

`new_call_message(id, callee, args, kwargs)`

Builds a "call" message: a request to invoke callee remotely with args and kwargs.

new_call_message(uuid4(), "add_one", [41], new_map())

`new_result_ok(id, value)`

Builds a successful "result" reply to a "call" message.

new_result_ok("some-call-id", 42)

`new_result_fault(id, message_text)`

Builds a failed "result" reply to a "call" message.

new_result_fault("some-call-id", "division by zero")

`new_heartbeat_message(node_id)`

Builds a "heartbeat" message announcing that node_id is alive right now.

new_heartbeat_message("node-1")

`new_join_message(node_id, host, port)`

new_join_message(uuid4(), "127.0.0.1", 9101)

`new_members_message(members_list)`

new_members_message([])

`new_leave_message(node_id)`

Builds a "leave" message announcing that node_id is departing the cluster gracefully.

new_leave_message("node-1")

`new_cancel_message(id)`

Builds a "cancel" message for a call already in flight.

new_cancel_message("some-call-id")

`new_lock_request_message(node_id, name)`

Builds a "lock_request" message asking the owner of a distributed lock named name to grant it to node_id.

new_lock_request_message("node-1", "counter-lock")

`new_lock_granted_message(node_id, name)`

Builds a "lock_granted" reply confirming that a distributed lock has been acquired.

new_lock_granted_message("owner-node", "counter-lock")

`new_lock_release_message(node_id, name)`

Builds a "lock_release" message releasing a distributed lock.

new_lock_release_message("node-1", "counter-lock")

`new_value_get_message(name)`

Builds a "value_get" message requesting the current value of a distributed value named name from its owner.

new_value_get_message("counter")

`new_value_set_message(name, value)`

Builds a "value_set" message asking the owner of a distributed value named name to overwrite it with value.

new_value_set_message("counter", 1)

`new_value_reply_message(value)`

Builds a "value_reply" message carrying a distributed value's current contents back to whoever asked for it.

new_value_reply_message(1)

`new_value_ack_message()`

Builds a "value_ack" message: a plain, valueless acknowledgement that a write succeeded.

new_value_ack_message()

`new_remember_get_message(key)`

Builds a "remember_get" message requesting the current value stored under key in a distributed memory owner's local remember/recall store.

new_remember_get_message("last-seen")

`new_remember_set_message(key, value)`

Builds a "remember_set" message asking a distributed memory owner to store value under key via its own local remember.

new_remember_set_message("last-seen", "2026-09-13")

`new_member_record(node_id, host, port)`

new_member_record("node-1", "127.0.0.1", 9100)

`cluster_join(my_port, seed_host, seed_port)`

node_a = cluster_join(9101, nothing, nothing)
node_b = cluster_join(9102, "127.0.0.1", 9101)

`cluster_register_function(my_node, name, fn)`

node = cluster_join(9101, nothing, nothing)
cluster_register_function(node, "add_one", make(n) give n + 1)

`cluster_nodes(my_node)`

node = cluster_join(9101, nothing, nothing)
cluster_nodes(node)

`cluster_diagnostics(my_node)`

node = cluster_join(9101, nothing, nothing)
cluster_diagnostics(node)

`cluster_leave(my_node)`

node = cluster_join(9101, nothing, nothing)
cluster_leave(node)

`cluster_is_cancelled(my_node, call_id)`

node = cluster_join(9101, nothing, nothing)
cluster_is_cancelled(node, "some-call-id")

Remote tasks

`remote_spawn_with_id(node, call_id, callee_name, args_list, kwargs_map)`

Connects to node, sends a "call" message under call_id, and returns a RemoteTask handle immediately, without waiting for a reply.

task = remote_spawn_with_id(node_b, uuid4(), "add_one", [41], new_map())
remote_await(task)

`remote_spawn(node, callee_name, args_list, kwargs_map)`

task = remote_spawn(node_b, "add_one", [41], new_map())
remote_await(task)

`remote_await(remote_task)`

task = remote_spawn(node_b, "add_one", [41], new_map())
remote_await(task)

`remote_spawn_retry(node_list, callee_name, args_list, kwargs_map, max_retries)`

task = remote_spawn_retry([node_b, node_c], "add_one", [41], new_map(), 4)
remote_await(task)

`cluster_cancel(remote_task)`

task = remote_spawn(node_b, "slow_job", [], new_map())
cluster_cancel(task)

Distributed shared state

`distributed_handle(name, owner_node)`

Builds a DistributedShared handle for a value or lock named name, owned by owner_node.

handle = distributed_handle("counter", node_a)
distributed_get(node_b, handle)

`share_distributed(my_node, name, initial_value)`

node = cluster_join(9101, nothing, nothing)
counter = share_distributed(node, "counter", 0)

`distributed_get(my_node, handle)`

Reads a distributed value's current contents.

node = cluster_join(9101, nothing, nothing)
counter = share_distributed(node, "counter", 0)
distributed_get(node, counter)

`distributed_set(my_node, handle, value)`

Overwrites a distributed value's contents.

node = cluster_join(9101, nothing, nothing)
counter = share_distributed(node, "counter", 0)
distributed_set(node, counter, 1)

`lock_acquire_distributed(my_node, handle)`

Blocks until the distributed lock handle is acquired by my_node.

node = cluster_join(9101, nothing, nothing)
counter = share_distributed(node, "counter", 0)
lock_acquire_distributed(node, counter)
lock_release_distributed(node, counter)

`lock_release_distributed(my_node, handle)`

Releases a distributed lock previously acquired with lock_acquire_distributed.

node = cluster_join(9101, nothing, nothing)
counter = share_distributed(node, "counter", 0)
lock_acquire_distributed(node, counter)
lock_release_distributed(node, counter)

`protect_distributed(my_node, handle, body_fn)`

node = cluster_join(9101, nothing, nothing)
counter = share_distributed(node, "counter", 0)
protect_distributed(node, counter, make() give distributed_set(node, counter, distributed_get(node, counter) + 1))

Distributed memory (AI-agent memory, across real machines)

`remember_distributed(my_node, owner_node, key, value)`

node_a = cluster_join(9101, nothing, nothing)
remember_distributed(node_a, node_a, "last-seen", "2026-09-13")

`recall_distributed(my_node, owner_node, key)`

Reads a value previously stored with remember_distributed.

node_a = cluster_join(9101, nothing, nothing)
remember_distributed(node_a, node_a, "last-seen", "2026-09-13")
recall_distributed(node_a, node_a, "last-seen")