Mesh Scheduling

Gossip-based worker discovery, consistent-hashing affinity, and work-stealing for distributed task dispatch.

Workers can form a decentralized mesh — SWIM gossip for peer discovery, a consistent-hash ring for task placement, and TCP work-stealing — so idle workers pull work from busy peers instead of every worker hammering storage on every poll.

Storage remains the source of truth. The mesh only changes how quickly a claimed job reaches an idle worker — if gossip fails entirely, a worker just falls back to standard polling — so it's safe to add to any worker without changing dispatch guarantees.

Not included in the published wheel

The package published to PyPI does not include the mesh cargo feature. MeshWorker is importable from it, but passing it to run_worker has no effect. To use mesh scheduling you must build the extension from source with the feature enabled: uv run maturin develop --features mesh,workflows

Requires the mesh cargo feature

pnpm build:native compiles the addon with postgres,redis,workflows,mesh enabled, and published prebuilt binaries ship with it too — so mesh works out of the box on install. If you build the native addon yourself with a narrower --features list, drop mesh and the mesh option on the worker is silently ignored.

Included in the published native library

The mesh cargo feature ships in the published native library, so MeshOptions and the worker builder's mesh(...) work out of the box — no separate build required.

An opt-in dependency, and not a worker option

mesh is a cargo feature on the flexiq crate — flexiq = { version = "2", features = ["mesh"] } — and it resolves an extra source crate, flexiq-mesh. Without it, flexiq::mesh does not exist and the code below will not compile.

Turning it on adds the module; it does not change what a worker does. WorkerBuilder has no mesh option, so a MeshNode is something you construct and drive yourself, running beside the worker rather than inside it.

Quick start

First worker — no seeds needed, it becomes the initial cluster node:

python

from flexiq import Queue, MeshWorker

queue = Queue(db_path="tasks.db")

@queue.task()
def process(item_id: int):
    ...

# First worker — no seeds needed, becomes the initial cluster node
mesh = MeshWorker(port=7946)
queue.run_worker(queues=["default"], mesh=mesh)

import { Queue } from "@byteveda/flexiq";

const queue = new Queue({ dbPath: "tasks.db" });

queue.task("process", (itemId: number) => {
  // ...
});

// First worker — no seeds needed, becomes the initial cluster node.
const worker = queue.runWorker({
  queues: ["default"],
  mesh: { port: 7946 },
});

java

// First worker — no seeds needed, becomes the initial cluster node
Worker worker = flexiq.worker()
        .handle(process, payload -> handle(payload))
        .mesh(MeshOptions.builder().port(7946).build())
        .start();

flexiq::mesh is a feature-gated re-export of the flexiq-mesh crate, not a worker option — WorkerBuilder has none. A node is built and spawned beside the worker, and you drive it:

toml

flexiq = { version = "2", features = ["mesh"] }

rust

use std::sync::Arc;

use flexiq::mesh::{MeshConfig, MeshNode};

// First node — no seeds, so it starts the cluster.
let node = Arc::new(MeshNode::new("worker-1".to_string(), MeshConfig::default()));
node.spawn_gossip(vec!["default".to_string()], 4);  // queues, and the capacity to advertise
node.spawn_steal_server();

Both spawn tokio tasks, so they need a runtime already running. should_prefetch, prefetch, pop_local and try_steal are the rest of the loop; nothing calls them for you.

In a second process (or on another machine), join the cluster by seeding from the first worker:

python

# Joins the cluster by seeding from the first worker
mesh = MeshWorker(
    port=7946,
    seeds=["first-worker-host:7946"],
)
queue.run_worker(queues=["default"], mesh=mesh)

// Joins the cluster by seeding from the first worker.
const worker = queue.runWorker({
  queues: ["default"],
  mesh: { port: 7946, seeds: ["first-worker-host:7946"] },
});

java

// Joins the cluster by seeding from the first worker
Worker worker = flexiq.worker()
        .handle(process, payload -> handle(payload))
        .mesh(MeshOptions.builder()
                .port(7946)
                .seed("first-worker-host:7946")
                .build())
        .start();

rust

// Joins the cluster by seeding from the first worker.
let config = MeshConfig {
    seeds: vec!["first-worker-host:7946".to_string()],
    // A bare IP, not host:port — `advertise_ip()` parses it as an `IpAddr` and
    // falls back to 0.0.0.0, which nobody can dial.
    advertise_addr: Some("10.0.0.7".to_string()),
    ..MeshConfig::default()
};
let node = Arc::new(MeshNode::new("worker-2".to_string(), config));
node.spawn_gossip(vec!["default".to_string()], 4);
node.spawn_steal_server();

MeshConfig::default() carries no seeds and no advertise address, so a node built from it gossips to nobody — that is the single-host shape.

The second worker discovers the first via gossip. Add a third and it only needs to seed from any existing member — gossip propagates the full membership list from there.

How it works

Mesh scheduling composes four primitives that work together:

Primitive	What it does
SWIM gossip	UDP protocol for peer discovery and failure detection (~1.5s vs a 30s DB heartbeat)
Consistent hashing	Hash ring with virtual nodes maps each task name to a preferred worker (soft affinity)
Local deque	In-memory buffer between DB polls, sorted by affinity — workers drain it before hitting storage again
Work-stealing	TCP protocol lets idle workers pull jobs from busy peers' deques

The database remains the source of truth — an atomic claim is what actually grants a worker a job. Gossip and stealing are a dispatch-locality optimization layered on top of that. See Architecture: Mesh Scheduling for the shared engine's internals, including adaptive prefetch sizing and poll jitter.

Configuration

All mesh settings live in one place, alongside the worker configuration:

python

mesh = MeshWorker(
    port=7946,              # gossip UDP port (steal port = port + 1)
    seeds=["host:7946"],    # seed nodes for cluster join
    steal=True,             # enable work-stealing
    affinity_weight=0.7,    # 0.0–1.0, how strongly tasks prefer their hashed worker
    local_buffer=64,        # local deque capacity
    steal_batch=4,          # max jobs stolen per request
    steal_threshold=2,      # steal when own deque ≤ this
    virtual_nodes=150,      # consistent-hash ring virtual nodes per worker
    bind_addr="0.0.0.0",   # network interface to bind
    encryption_key=None,    # base64-encoded key for gossip encryption
    steal_rate_limit=10,    # max steal requests per peer per second
)

queue.runWorker({
  mesh: {
    port: 7946, // gossip UDP port; steal TCP port is port + 1
    bindAddr: "0.0.0.0",
    seeds: ["10.0.0.2:7946"],
    steal: true,
    affinityWeight: 0.7,
    localBuffer: 64,
    stealBatch: 4,
    stealThreshold: 2,
    virtualNodes: 150,
    advertiseAddr: undefined,
    encryptionKey: undefined,
    stealRateLimit: 10,
  },
});

java

MeshOptions.builder()
        .port(7946)                 // gossip UDP port; steal TCP port is port + 1
        .seed("10.0.0.2:7946")      // repeatable; or seeds(List<String>)
        .bindAddr("0.0.0.0")
        .advertiseAddr(null)        // required when bindAddr is 0.0.0.0 across hosts
        .enableStealing(true)
        .affinityWeight(0.7)
        .localBufferCapacity(64)
        .maxStealBatch(4)
        .stealThreshold(2)
        .virtualNodes(150)
        .stealRateLimit(10)
        .encryptionKey(null)
        .build();

rust

MeshConfig {
    gossip_port: 7946,                        // SWIM gossip, UDP
    steal_port: 7947,                         // work-stealing, TCP
    bind_addr: "0.0.0.0".to_string(),
    seeds: vec!["10.0.0.2:7946".to_string()],
    advertise_addr: None,                     // a bare IP; required across hosts
    enable_stealing: true,
    affinity_weight: 0.7,
    local_buffer_capacity: 64,
    max_steal_batch: 4,
    steal_threshold: 2,
    virtual_nodes: 150,
    steal_rate_limit: 10,
    encryption_key: None,                     // base64, 32 bytes
    // protocol_period_ms, indirect_ping_count, suspicion_multiplier
    ..MeshConfig::default()
};

Two things differ from a settings object. steal_port is a field of its own and does not follow gossip_port — MeshConfig::default() pins 7947, so moving one means moving both. And the SWIM timings are ordinary public fields here rather than engine constants: protocol_period_ms (500), indirect_ping_count (3) and suspicion_multiplier (4) are yours to set.

Parameter reference

Each binding exposes these settings with its own names, defaults, and validation:

Parameter	Type	Default	Description
`port`	`int`	`7946`	Gossip UDP port. Steal TCP port is `port + 1`. Must be 1024–65535.
`seeds`	`list[str]`	`[]`	Addresses of existing cluster members (`host:port`).
`steal`	`bool`	`True`	Whether this worker can steal from and be stolen from.
`affinity_weight`	`float`	`0.7`	Task-to-worker affinity strength. `0.0` = no affinity, `1.0` = strong preference.
`local_buffer`	`int`	`64`	Max jobs buffered locally before back-pressuring prefetch.
`steal_batch`	`int`	`4`	Max jobs transferred per steal request.
`steal_threshold`	`int`	`2`	Trigger stealing when own buffer drops to this level.
`virtual_nodes`	`int`	`150`	Hash ring virtual nodes per worker. More = more even distribution.
`bind_addr`	`str`	`"0.0.0.0"`	Network interface for gossip and steal servers.
`encryption_key`	`str \| None`	`None`	Base64 key for XOR gossip encryption. All nodes must share the same key.
`steal_rate_limit`	`int`	`10`	Max steal requests accepted per peer per second. `0` = unlimited.

Option	Type	Default	Description
`port`	`number`	— (required)	Gossip UDP port. Steal TCP port is `port + 1`. Must be 1–65534 — `runWorker` throws otherwise.
`bindAddr`	`string`	`"0.0.0.0"`	Bind address for both the gossip and steal listeners.
`seeds`	`string[]`	`[]`	Existing members to join, as `"host:gossipPort"` strings. Empty means this node starts a new cluster.
`steal`	`boolean`	`true`	Whether this worker initiates steals when its deque runs low. See Work-stealing — it does not stop this worker from donating jobs to peers.
`affinityWeight`	`number`	`0.7`	Accepted for forward compatibility, but the current engine doesn't read it — affinity is a fixed owned/non-owned split (see Task affinity), not a graduated weight. Leave it at the default.
`localBuffer`	`number`	`64`	Local deque capacity before it stops accepting more prefetched jobs.
`stealBatch`	`number`	`4`	Max jobs requested per steal, capped by however many the peer actually has.
`stealThreshold`	`number`	`2`	This worker tries to steal once its own deque length drops to or below this.
`virtualNodes`	`number`	`150`	Virtual points per worker on the hash ring. Higher spreads task ownership more evenly across workers.
`advertiseAddr`	`string`	unset	IP address advertised to peers when this worker is behind NAT — a bare IP, not `host:port` (the advertised ports still come from `port`/`port + 1`). Falls back to `bindAddr`.
`encryptionKey`	`string`	unset	Base64 key that XOR-encrypts gossip datagrams. See Gossip encryption.
`stealRateLimit`	`number`	`10`	Max steal requests this worker will answer per peer per second. `0` disables the limit.

Builder method	Type	Default	Description
`port(int)`	`int`	`7946`	Gossip UDP port. Steal TCP port is always `port + 1`. Must be `1..65534`.
`seed(String)` / `seeds(List<String>)`	`String` / `List<String>`	none	`host:port` of an existing cluster member; call repeatedly to add more.
`bindAddr(String)`	`String`	`"0.0.0.0"`	Listen address for both the gossip and steal servers.
`advertiseAddr(String)`	`String`	`null`	Address advertised to peers; required behind NAT or when `bindAddr` is `0.0.0.0` and workers span hosts.
`enableStealing(boolean)`	`boolean`	`true`	Whether this worker can steal from, and be stolen from by, peers.
`affinityWeight(double)`	`double`	`0.7`	Hash-ring affinity strength. `0.0` ignores affinity, `1.0` is strict.
`localBufferCapacity(int)`	`int`	`64`	Max jobs buffered in the local deque before back-pressuring prefetch.
`maxStealBatch(int)`	`int`	`4`	Max jobs transferred to a thief per steal request.
`stealThreshold(int)`	`int`	`2`	Trigger stealing once the local deque drops to this length.
`virtualNodes(int)`	`int`	`150`	Hash-ring virtual nodes per worker; higher spreads placement more evenly.
`stealRateLimit(int)`	`int`	`10`	Max steal requests served per peer per second; `0` disables the limit.
`encryptionKey(String)`	`String`	`null`	Base64 32-byte key XOR-applied to gossip datagrams. Obfuscation, not encryption.

SWIM protocol timing (500ms period, 3 indirect-probe peers, suspicion multiplier 4) is fixed by the core engine and isn't exposed on MeshOptions.Builder.

MeshConfig is a plain struct of public fields — no builder, no setters, and nothing validates a value on the way in. Fill in the ones you care about and let ..MeshConfig::default() supply the rest.

Field	Type	Default	Description
`gossip_port`	`u16`	`7946`	UDP port the SWIM gossip listener binds.
`steal_port`	`u16`	`7947`	TCP port the steal server binds. A field of its own — see the note below.
`bind_addr`	`String`	`"0.0.0.0"`	Interface both listeners bind.
`seeds`	`Vec<String>`	`vec![]`	`host:port` of existing members to join. Empty starts a fresh cluster.
`protocol_period_ms`	`u64`	`500`	One SWIM round. Also the upper bound of `poll_jitter_ms()`.
`indirect_ping_count`	`usize`	`3`	Peers asked to probe on your behalf when a direct ping goes unanswered.
`suspicion_multiplier`	`u32`	`4`	Applied to `ln(N+1) × protocol_period` before a suspect is declared dead.
`virtual_nodes`	`usize`	`150`	Points per worker on the hash ring; more spreads ownership more evenly.
`local_buffer_capacity`	`usize`	`64`	Local deque capacity. `push_sorted` truncates a batch that would overflow it.
`max_steal_batch`	`usize`	`4`	Jobs asked for per steal request.
`steal_threshold`	`usize`	`2`	`should_steal()` is true once the deque is at or below this.
`affinity_weight`	`f64`	`0.7`	Carried on the struct, but nothing in the engine reads it — see Task affinity.
`enable_stealing`	`bool`	`true`	Gates `should_steal()`, so it stops this node initiating a steal. The steal server still answers peers.
`advertise_addr`	`Option<String>`	`None`	A bare IP to advertise. Falls back to `bind_addr`.
`encryption_key`	`Option<String>`	`None`	Base64 32-byte key XOR-applied to gossip datagrams.
`steal_rate_limit`	`u32`	`10`	Steal requests answered per peer per second. `0` is unlimited.

steal_port does not follow gossip_port

The field's own doc comment reads "default: gossip_port + 1", but MeshConfig::default() hardcodes 7947. Move gossip_port and steal_port stays where it was — set both, or two nodes on one host will fight over the same steal port.

Two fields are quietly forgiving in a way worth knowing. advertise_addr goes through advertise_ip(), which parses it as an IpAddr — so "10.0.0.7:7946" or a hostname does not parse, and the unwrap_or hands back 0.0.0.0, which no peer can dial. And decoded_encryption_key() answers None for anything that is not valid base64, leaving the node gossiping in the clear. Assert on both at startup.

Cluster topology

Single-machine (development)

Run multiple workers on one host with different ports — the steal port is always port + 1, so pick gossip ports far enough apart:

python

# worker_a.py
from flexiq import Queue, MeshWorker

queue = Queue(db_path="tasks.db")

@queue.task()
def send_email(to: str, subject: str):
    ...

mesh = MeshWorker(port=7946)
queue.run_worker(mesh=mesh)

python

# worker_b.py — seeds from worker A
from myapp import queue  # same queue, different process

mesh = MeshWorker(port=7948, seeds=["127.0.0.1:7946"])
queue.run_worker(mesh=mesh)

// worker-a.ts
import { Queue } from "@byteveda/flexiq";

const queue = new Queue({ dbPath: "tasks.db" });
queue.task("sendEmail", (to: string, subject: string) => {
  /* ... */
});

queue.runWorker({ mesh: { port: 7946 } });

// worker-b.ts — same db file, second process
import { Queue } from "@byteveda/flexiq";

const queue = new Queue({ dbPath: "tasks.db" });
queue.task("sendEmail", (to: string, subject: string) => {
  /* ... */
});

queue.runWorker({ mesh: { port: 7948, seeds: ["127.0.0.1:7946"] } });

java

// Worker A — first node, no seeds
Worker workerA = flexiq.worker()
        .handle(process, payload -> handle(payload))
        .mesh(MeshOptions.builder().port(7946).build())
        .start();

// Worker B — same storage, seeds from A
Worker workerB = flexiq.worker()
        .handle(process, payload -> handle(payload))
        .mesh(MeshOptions.builder()
                .port(7948)                       // steal port becomes 7949
                .seed("127.0.0.1:7946")
                .build())
        .start();

rust

// Node A — first node, no seeds. Defaults are 7946 gossip / 7947 steal.
let a = Arc::new(MeshNode::new("worker-a".to_string(), MeshConfig::default()));
a.spawn_gossip(vec!["default".to_string()], 4);
a.spawn_steal_server();

// Node B — seeds from A. Both ports move, because neither follows the other.
let b = Arc::new(MeshNode::new(
    "worker-b".to_string(),
    MeshConfig {
        gossip_port: 7948,
        steal_port: 7949,
        seeds: vec!["127.0.0.1:7946".to_string()],
        ..MeshConfig::default()
    },
));
b.spawn_gossip(vec!["default".to_string()], 4);
b.spawn_steal_server();

Multi-machine (production)

Point seeds at a couple of known-stable nodes — you don't need to list every node, gossip propagates membership from there:

python

import os

SEEDS = ["scheduler-1.internal:7946", "scheduler-2.internal:7946"]

mesh = MeshWorker(
    port=7946,
    seeds=SEEDS,
    encryption_key=os.environ["MESH_ENCRYPTION_KEY"],
    steal_rate_limit=20,
)
queue.run_worker(
    queues=["default", "emails"],
    mesh=mesh,
)

const SEEDS = ["scheduler-1.internal:7946", "scheduler-2.internal:7946"];

queue.runWorker({
  queues: ["default", "emails"],
  mesh: {
    port: 7946,
    seeds: SEEDS,
    encryptionKey: process.env.MESH_ENCRYPTION_KEY,
    stealRateLimit: 20,
  },
});

java

List<String> seeds = List.of(
        "scheduler-1.internal:7946",
        "scheduler-2.internal:7946");

Worker worker = flexiq.worker()
        .handle(process, payload -> handle(payload))
        .queues("default", "emails")
        .mesh(MeshOptions.builder()
                .seeds(seeds)
                .advertiseAddr(System.getenv("HOST_ADDR") + ":7946")
                .encryptionKey(System.getenv("MESH_KEY"))
                .stealRateLimit(20)
                .build())
        .start();

rust

let config = MeshConfig {
    seeds: vec![
        "scheduler-1.internal:7946".to_string(),
        "scheduler-2.internal:7946".to_string(),
    ],
    advertise_addr: Some(std::env::var("HOST_ADDR")?),      // the pod's own IP
    encryption_key: Some(std::env::var("MESH_ENCRYPTION_KEY")?),
    steal_rate_limit: 20,
    ..MeshConfig::default()
};
let node = Arc::new(MeshNode::new(worker_id, config));
node.spawn_gossip(vec!["default".to_string(), "emails".to_string()], 4);
node.spawn_steal_server();

The queues a node gossips are the ones you pass to spawn_gossip; nothing reads them off a worker, so they have to match the queues that worker actually pulls.

Set advertiseAddr so peers dial back the right host when bindAddr stays 0.0.0.0.

Across hosts, advertise_addr is the field that has to be right. advertise_ip() prefers it and falls back to bind_addr, and a 0.0.0.0 bind is not an address a peer can dial — so leave it unset on a multi-host rollout and the cluster forms in name only. Give it the pod's or host's own IP, bare:

rust

let advertise = std::env::var("POD_IP")?;
advertise.parse::<std::net::IpAddr>()?;  // fail here, not silently at 0.0.0.0

The two ports are attached to that IP for you — spawn_gossip builds the advertised gossip_addr and steal_addr from advertise_ip() plus gossip_port and steal_port, which is why the field takes an IP and not a host:port.

The threads argument to spawn_gossip is gossiped as this node's capacity, and peers divide the prefetch budget by it — pass the worker's real concurrency, not a placeholder. Its queues argument is likewise the list peers see, and nothing cross-checks it against .queues([...]) on the builder: they are two separate lists you have to keep equal.

Firewall rules

Workers need UDP (gossip) and TCP (steal) access to each other. Open both the gossip port and the steal port (gossip + 1) between all mesh workers.

No CLI flags for mesh

The flexiq run ./app.js CLI command has no mesh options — mesh is only enabled by passing mesh to runWorker() in your own code. Read the config from the environment in your entrypoint instead:

// worker-entry.ts — container entrypoint, reads mesh config from env vars
import { Queue, type MeshWorkerConfig } from "@byteveda/flexiq";

const queue = new Queue({ dbPath: "tasks.db" });
queue.task("sendEmail", (to: string, subject: string) => {
  /* ... */
});

function meshFromEnv(): MeshWorkerConfig | undefined {
  const port = process.env.MESH_PORT;
  if (!port) return undefined;
  const seeds = (process.env.MESH_SEEDS ?? "")
    .split(",")
    .map((s) => s.trim())
    .filter(Boolean);
  return { port: Number(port), seeds, encryptionKey: process.env.MESH_KEY };
}

queue.runWorker({ mesh: meshFromEnv() });

bash

node worker-entry.js  # not `flexiq run` — the CLI has no mesh flags

Docker Compose example

yaml

services:
  worker-1:
    build: .
    command: flexiq worker --app myapp:queue
    environment:
      MESH_PORT: "7946"
      MESH_SEEDS: ""  # first node, no seeds
      MESH_KEY: ${MESH_ENCRYPTION_KEY}
    ports:
      - "7946:7946/udp"  # gossip
      - "7947:7947/tcp"  # steal

  worker-2:
    build: .
    command: flexiq worker --app myapp:queue
    environment:
      MESH_PORT: "7946"
      MESH_SEEDS: "worker-1:7946"
      MESH_KEY: ${MESH_ENCRYPTION_KEY}
    ports:
      - "7948:7946/udp"
      - "7949:7947/tcp"
    depends_on:
      - worker-1

Illustrative — the CLI can't consume MESH_* env vars

This compose file is illustrative, not runnable as-is: the flexiq worker CLI has no mesh flags and never reads MESH_PORT/MESH_SEEDS/MESH_KEY. Mesh is only enabled by calling queue.run_worker(mesh=...) in your own Python code. Replace command: flexiq worker --app myapp:queue with command: python -m myapp, where myapp is a module like the one below — its __main__ block starts the worker with the env-derived mesh config.

python

# myapp.py — reads mesh config from environment
import os
from flexiq import Queue, MeshWorker

queue = Queue(db_path="tasks.db")

def get_mesh() -> MeshWorker | None:
    port = os.environ.get("MESH_PORT")
    if not port:
        return None
    seeds_raw = os.environ.get("MESH_SEEDS", "")
    seeds = [s.strip() for s in seeds_raw.split(",") if s.strip()]
    return MeshWorker(
        port=int(port),
        seeds=seeds,
        encryption_key=os.environ.get("MESH_KEY"),
    )

if __name__ == "__main__":
    queue.run_worker(mesh=get_mesh())

Task affinity

The consistent-hash ring maps each task name to a preferred worker. When a worker prefetches a batch of jobs, it sorts them before pushing them into the local deque:

Owned tasks (hashed to this worker) go to the back of the deque — popped first
Non-owned tasks go to the front — left there for peers to steal

Any worker can still run any task — affinity only biases which worker gets to it first. This improves cache locality: when a task consistently lands on the same worker, its open connections, warm caches, and worker-scoped resources stay hot instead of being rebuilt everywhere.

Watch a routing key pick its pool

Each job carries a key that decides which pool of workers is built to run it. Drop a machine and its work reroutes across the rest of that pool rather than stalling.

Loading demo…

The affinity_weight parameter controls how aggressively the ring biases dispatch — 0.0 disables affinity entirely, 1.0 is a strict preference:

python

# High affinity — tasks strongly prefer their hashed worker
mesh = MeshWorker(affinity_weight=1.0)

# No affinity — pure load-balancing, no task-to-worker preference
mesh = MeshWorker(affinity_weight=0.0)

affinityWeight on the mesh config is reserved but currently unused — the sort above is a hard owned/non-owned split, not a tunable strength. There's no option today to dial affinity up or down.

affinityWeight controls the same tradeoff:

java

// Strict — tasks strongly prefer their hashed worker
MeshOptions.builder().affinityWeight(1.0).build();

// None — pure load-balancing, no task-to-worker preference
MeshOptions.builder().affinityWeight(0.0).build();

MeshConfig::affinity_weight exists and defaults to 0.7, but the engine never reads it — LocalDeque::push_sorted partitions a batch into owned and non-owned and nothing consults a weight. Setting it to 0.0 or 1.0 changes nothing today.

The split itself is a function you can call:

rust

if node.is_affinity_owner("process_order") {
    // this node is the ring owner for that task name
}

prefetch calls it once per job and pushes accordingly: owned to the back, which pop_local() takes first, and non-owned to the front, which is the end give_jobs() hands to a thief. The knob that actually moves placement is virtual_nodes — it controls how evenly names spread over the ring, not how hard the deque prefers them.

Work-stealing

When a worker's local buffer drops to its steal threshold, it looks at gossip-reported buffer lengths across the cluster, picks the busiest peer, and opens a TCP connection to that peer's steal port requesting a batch of jobs.

The victim pops jobs from the cold end of its deque (non-owned tasks first) and sends them back. The thief executes them locally — no DB round-trip needed, since the jobs are already claimed as running in storage. Stealing is rate-limited per peer to prevent thundering-herd effects.

The worker only steals from peers with enough surplus (steal threshold + steal batch). The connect attempt times out after 500ms and the response after 2s, so a stuck peer doesn't stall dispatch.

Nothing drives that loop for you — MeshNode::try_steal is an async method you call, and it returns how many jobs came back:

rust

if node.should_steal() {
    let stolen = node.try_steal().await;  // already pushed into the local deque
}

It bails early unless should_steal() holds, then asks best_steal_target for the busiest alive peer whose gossiped local_buffer_len is strictly greater than steal_threshold + max_steal_batch. Below that surplus there is no target and the call returns 0. Connecting is capped at 500ms and the reply at 2s, so a wedged peer costs a round, not the loop.

One counter to read with care: steals_initiated is bumped before the connection, so a failure to connect, write, read or decode is counted as an attempt. A gap from steals_succeeded means attempts are not paying off — it does not specifically mean peers were empty.

Work-stealing is on by default — workers automatically balance load across the cluster. When one worker gets a burst of jobs, idle peers steal the overflow within milliseconds. Disabling it stops this worker from participating in steals:

python

mesh = MeshWorker(steal=False)

Workers still benefit from gossip discovery, consistent hashing, and local deque prefetch — but won't steal from or donate jobs to peers. Useful when tasks have strong per-worker state dependencies.

queue.runWorker({ mesh: { port: 7946, steal: false } });

steal: false is one-directional

Setting steal: false only stops this worker from initiating steals — the TCP steal server always runs, so this worker will still hand jobs to any peer that asks (subject to stealRateLimit). Use it on a worker you don't want draining its own queue to help others, not to shield a worker from ever donating work.

java

MeshOptions.builder().enableStealing(false).build();

Gossip discovery, consistent hashing, and local-deque prefetch keep running — the worker just won't steal from or donate jobs to peers. Useful when tasks have strong per-worker state.

rust

MeshConfig { enable_stealing: false, ..MeshConfig::default() };

The flag is read by should_steal() only, so it stops the steals this node initiates. spawn_steal_server() still serves peers that ask, subject to steal_rate_limit — skip the call entirely if this node must never donate.

Tuning work-stealing

Scenario	Recommended settings
Bursty, short tasks	`steal_batch=8, steal_threshold=4` — steal more, earlier
Long-running tasks	`steal_batch=1, steal_threshold=1` — steal conservatively
Many workers (10+)	`steal_rate_limit=5` — prevent steal storms
Few workers (2–3)	`steal_rate_limit=0` — unlimited, low contention

Scenario	Recommended settings
Bursty, short tasks	`stealBatch: 8, stealThreshold: 4` — steal more, earlier
Long-running tasks	`stealBatch: 1, stealThreshold: 1` — steal conservatively
Many workers (10+)	`stealRateLimit: 5` — prevent steal storms
Few workers (2–3)	`stealRateLimit: 0` — unlimited, contention is low anyway

Scenario	Recommended settings
Bursty, short tasks	`maxStealBatch(8).stealThreshold(4)` — steal more, earlier
Long-running tasks	`maxStealBatch(1).stealThreshold(1)` — steal conservatively
Many workers (10+)	`stealRateLimit(5)` — avoid steal storms
Few workers (2-3)	`stealRateLimit(0)` — unlimited, low contention

Scenario	`MeshConfig` fields
Bursty, short tasks	`max_steal_batch: 8, steal_threshold: 4` — steal more, earlier
Long-running tasks	`max_steal_batch: 1, steal_threshold: 1` — steal conservatively
Many nodes (10+)	`steal_rate_limit: 5` — cap what each peer will answer
Few nodes (2–3)	`steal_rate_limit: 0` — unlimited, contention is low anyway

Raising the pair also raises the surplus a peer must show before it counts as a target: try_steal requires local_buffer_len > steal_threshold + max_steal_batch, so 8/4 needs a peer holding thirteen buffered jobs. Push both high enough and no peer ever qualifies.

enable_stealing: false only gates should_steal(). The steal server still runs once you have called spawn_steal_server, so the node keeps donating jobs while refusing to take any — skip the spawn to opt out of both.

Gossip encryption

Enable gossip encryption with a shared, base64-encoded key:

python

import base64
import os

# Generate a key (share this across all workers)
key = base64.b64encode(os.urandom(32)).decode()
print(key)  # store in environment variable or secret manager

mesh = MeshWorker(encryption_key=key)

import { randomBytes } from "node:crypto";

const key = randomBytes(32).toString("base64"); // share across every mesh worker
queue.runWorker({ mesh: { port: 7946, encryptionKey: key } });

java

MeshOptions.builder()
        .port(7946)
        .seed("10.0.0.2:7946")
        .encryptionKey(System.getenv("MESH_KEY")) // 32-byte key, base64-encoded
        .build();

rust

let config = MeshConfig {
    seeds: vec!["10.0.0.2:7946".to_string()],
    encryption_key: Some(std::env::var("MESH_KEY")?),  // 32 bytes, base64
    ..MeshConfig::default()
};
assert!(config.decoded_encryption_key().is_some(), "MESH_KEY is not base64");

decoded_encryption_key() is where the string becomes bytes, and it answers None for anything it cannot decode rather than failing — check it at startup, or a typo leaves this node gossiping in the clear while its peers do not.

All nodes in the cluster must use the same key. Nodes with mismatched keys fail to decode each other's gossip messages and never join the cluster.

Encryption scope

Gossip encryption XOR-obfuscates the UDP membership protocol only — it deters casual sniffing, not a determined attacker. Work-stealing traffic over TCP is never encrypted. For real transport security, put mesh traffic on a private network (WireGuard, VPN, or a service mesh).

Monitoring

Mesh activity appears in Rust log output at debug and info levels:

bash

# See all mesh activity
RUST_LOG=flexiq_mesh=debug flexiq worker --app myapp:queue

# See only gossip membership changes
RUST_LOG=flexiq_mesh::swim=info flexiq worker --app myapp:queue

Key log messages:

Level	Message	Meaning
`info`	`gossip listening on ...`	Gossip server started
`info`	`discovered peer X at Y`	New cluster member joined
`info`	`member X declared dead`	Failure detector confirmed crash
`info`	`leave broadcast sent`	Graceful shutdown leave
`debug`	`giving N jobs to thief X`	Responded to steal request
`debug`	`ack from X resolved probe`	Healthy ping-ack cycle

If you use Prometheus, mesh metrics flow through the same worker observability pipeline.

Mesh workers register and heartbeat exactly like any other worker, so they show up in listWorkers() and on the dashboard Workers panel:

await queue.listWorkers();
// [{ workerId, hostname, pid, queues, status, lastHeartbeat, ... }]

There's no separate JS-facing API for mesh-only signals yet (peer count, steal counters, ring membership) — those live inside the Rust engine but aren't exposed through the addon. See Architecture: Mesh Scheduling for what the engine tracks internally.

Inspecting the cluster

Worker.meshClusterInfo() returns an Optional<MeshClusterInfo> — empty unless the worker was started with .mesh(...):

java

worker.meshClusterInfo().ifPresent(cluster ->
        log.info("peers={} load={} buffered={} prefetch={}",
                cluster.peerCount(), cluster.totalLoad(),
                cluster.totalBuffered(), cluster.adaptivePrefetch()));

`MeshClusterInfo` field	Description
`peerCount()`	Alive peers discovered via gossip (excludes this node).
`totalCapacity()`	Summed advertised capacity across those peers.
`totalLoad()`	Summed in-flight jobs across those peers.
`totalBuffered()`	Summed local-deque length across those peers.
`localBufferLen()`	This worker's own local-deque length.
`adaptivePrefetch()`	This worker's current prefetch budget.

Because you hold the MeshNode, the counters are a method call rather than an exported endpoint. Two snapshots, both Serialize, so either drops straight into whatever you already scrape:

rust

let m = node.metrics();      // MetricsSnapshot
let c = node.cluster_info(); // ClusterInfo

log::info!(
    "peers={} buffered={} prefetch={} stolen_in={} stolen_out={}",
    c.peer_count, c.local_buffer_len, c.adaptive_prefetch,
    m.jobs_stolen_in, m.jobs_stolen_out,
);

`MetricsSnapshot` field	What it counts
`prefetch_count`	Batches handed to the deque — how often you topped up, not how much you fetched.
`prefetch_jobs`	Jobs those batches actually buffered. Short of what was offered once capacity is reached.
`local_pops`	Jobs taken off the hot end to run here. The mesh's useful output.
`steals_initiated`	Attempts started, counted before the connection — failures are in here.
`steals_succeeded`	Attempts that returned at least one job.
`jobs_stolen_in`	Jobs received from peers. Rises when this node is the idle one.
`jobs_stolen_out`	Jobs surrendered to thieves. Rises when this node is the buffered-up one.
`ring_recalculations`	Ring rebuilds. A fast climb is affinity thrashing.

ClusterInfo carries peer_count, total_capacity, total_load, total_buffered, local_buffer_len and adaptive_prefetch. Only local_buffer_len is exact — the peer sums are as of each peer's last gossip, so they are at least one protocol_period_ms stale. The counters are free-running atomics loaded Relaxed and never reset, so a rate needs two snapshots and a clock.

For the narrative, flexiq-mesh logs through the log facade, so the output appears once your binary installs a logger:

bash

RUST_LOG=flexiq_mesh=debug ./my-worker          # everything the mesh does
RUST_LOG=flexiq_mesh::swim=info ./my-worker     # membership changes only
RUST_LOG=flexiq_mesh::steal=debug ./my-worker   # steal attempts and donations

Combining with other features

Mesh composes with everything else a worker already does. Each mesh worker still manages its own resource pool and prefetch batch independently:

python

from flexiq import Queue, MeshWorker
from flexiq.resources import ResourceDefinition, Scope

queue = Queue(db_path="tasks.db")

# Worker resources work normally — each mesh worker manages its own pool
@queue.worker_resource(scope=Scope.WORKER)
def db_pool():
    return create_connection_pool()

# Rate limits and concurrency are per-task, enforced in the DB layer
@queue.task(max_concurrent=5, rate_limit="100/m")
def process_order(order_id: int, db_pool=None):
    ...

# Batch dequeue works with mesh — prefetch fills the local deque
queue_config = Queue(db_path="tasks.db", scheduler_batch_size=10)

mesh = MeshWorker(seeds=["peer:7946"])
queue.run_worker(mesh=mesh)

import { Queue } from "@byteveda/flexiq";

const queue = new Queue({ dbPath: "tasks.db" });

// Worker resources work normally — each mesh worker builds its own pool.
queue.resource("db", () => createConnectionPool());

queue.task(
  "processOrder",
  async (orderId: string, deps: { db: Pool }) => {
    await deps.db.query("...", [orderId]);
  },
  { inject: ["db"], maxConcurrent: 5, rateLimit: "100/m" },
);

queue.runWorker({
  queues: ["default"],
  batchSize: 10, // scheduler pulls more per poll; the mesh sorts the batch by affinity
  mesh: { port: 7946, seeds: ["peer:7946"] },
});

java

flexiq.resource("db", ResourceScope.WORKER, ctx -> openPool(), pool -> pool.close());

try (Worker worker = flexiq.worker()
        .handle(processOrder, orderId -> {
            DataSource db = Resources.use("db");
            return process(db, orderId);
        })
        .batchSize(10) // scheduler claims 10 jobs per poll; mesh drains them into the local deque
        .mesh(MeshOptions.builder().seed("peer-a:7946").build())
        .start()) {
    worker.awaitShutdown();
}

rust

use flexiq::SchedulerConfig;

// Caps and rate limits are task attributes, enforced by the scheduler.
#[flexiq::task(max_concurrent = 5, rate_limit = "100/m")]
fn process_order(order_id: i64) -> flexiq::Outcome<()> {
    let _ = order_id;
    Ok(())
}

let worker = queue
    .worker()
    .register::<process_order>()
    // The scheduler claims 10 jobs per round; the mesh node, if you run one,
    // is what would sort them.
    .scheduler_config(SchedulerConfig { batch_size: 10, ..Default::default() })
    .spawn()?;

A Rust task takes its declared arguments and nothing else — the Rust SDK has no resource injection, so a shared pool is an ordinary OnceLock. And the worker never learns the mesh exists: a MeshNode you spawn runs beside it, not inside it.

Mixing mesh and non-mesh workers

A mesh worker and a plain worker can consume the same storage at once — the atomic claim in storage is what guarantees a job dispatches exactly once, regardless of whether the worker that wins it is meshed:

python

# Standard worker — no mesh, polls DB directly
queue.run_worker(queues=["default"])

# Mesh worker — same DB, but prefetches + steals
mesh = MeshWorker(seeds=["other-mesh-node:7946"])
queue.run_worker(queues=["default"], mesh=mesh)

// Standard worker — polls storage directly, no gossip
queue.runWorker({ queues: ["default"] });

// Mesh worker — same storage, but prefetches into a local deque and can steal
queue.runWorker({ queues: ["default"], mesh: { port: 7946, seeds: ["10.0.0.2:7946"] } });

java

// Standard worker — no mesh, polls storage directly
flexiq.worker().handle(process, this::handle).start();

// Mesh worker — same storage, but prefetches + steals
flexiq.worker().handle(process, this::handle)
        .mesh(MeshOptions.builder().seed("other-mesh-node:7946").build())
        .start();

rust

// A Rust worker is always the plain side — it polls storage directly.
let worker = queue.worker().register::<process>().spawn()?;

WorkerBuilder has no mesh option, so a Rust worker never joins the ring and is invisible to gossip. It still consumes the same storage as a meshed fleet, and the atomic claim is what keeps dispatch exactly-once across both.

Non-mesh workers are invisible to the mesh (no gossip, no stealing) but still process jobs normally — roll mesh out to part of a fleet gradually, without downtime.

Graceful shutdown

When a mesh worker shuts down, it broadcasts a Leave message to all known peers. They remove it from the ring immediately — no suspicion timeout needed.

Triggered via Ctrl+C, SIGTERM, or programmatic shutdown.

process.on("SIGTERM", () => worker.stop());

stop() broadcasts the leave message before the gossip loop exits.

worker.close() (and stop()) signal the mesh node before the native worker tears down, which broadcasts the leave message.

MeshNode::request_shutdown() is the signal. The gossip loop's shutdown arm is where broadcast_leave runs, so the Leave only goes out if that loop is the one that wakes:

rust

let gossip = node.spawn_gossip(vec!["default".to_string()], 4);
let stealer = node.spawn_steal_server();

// … later, on SIGTERM
node.request_shutdown();
node.request_shutdown();  // one signal per spawned loop — see below
let _ = tokio::join!(gossip, stealer);

One signal wakes one loop

request_shutdown calls Notify::notify_one on a handle both loops await. Two tasks are waiting and one call wakes one of them, so a single call leaves the other running — and if the steal server is the one that wakes, no leave message is ever broadcast and peers fall back to the suspicion timeout. Signal once per loop you spawned, then join the handles.

Awaiting the JoinHandles is what makes shutdown ordered: both are plain tokio::spawns, so dropping the runtime instead just cancels them mid-flight.

If a worker crashes without leaving, the SWIM failure detector kicks in: direct ping fails → indirect ping via intermediaries → suspicion → declared dead after suspicion_multiplier × ln(N+1) × protocol_period. With default settings (multiplier 4, 500ms period), a crashed worker in a 3-node cluster is typically detected in about 4 × ln(4) × 500ms ≈ 2.8s. These SWIM timing knobs are fixed by the core engine and aren't exposed as configuration options.

When to use mesh

Good fit:

Multiple workers processing the same queues (2+ workers)
High job throughput where DB polling is a bottleneck
Tasks that benefit from cache locality (imports, connections, warm state)
Environments where sub-second failure detection matters
Horizontal scaling where workers come and go frequently

Not needed:

Single worker setups
Low throughput (< 100 jobs/second)
Workers processing completely different queues (no overlap = no stealing)
Already using bare-metal autoscaler or KEDA for process-level scaling (mesh is within-process optimization, not a replacement for autoscaling)

Good fit:

Two or more workers consuming the same queue(s)
High throughput where DB polling latency is the bottleneck
Tasks that benefit from cache locality — warm connections, open clients — especially paired with worker-scoped resources
Fleets that scale workers up and down often and need sub-second peer detection

Not needed:

A single worker process
Low throughput
Workers consuming entirely disjoint queues — little to steal without overlap
You're already scaling worker processes up and down at the infrastructure level — mesh optimizes dispatch within a running fleet, it isn't a process autoscaler

Good fit:

Multiple workers processing the same queues (2+ workers)
High job throughput where DB polling is a bottleneck
Tasks that benefit from cache locality (connections, warm state)
Environments where sub-second failure detection matters
Horizontal scaling where workers come and go frequently

Not needed:

Single-worker setups
Low throughput (< 100 jobs/second)
Workers processing entirely disjoint queues (no overlap, no stealing)
Already scaling replicas with autoscaling / KEDA — mesh is a within-process dispatch optimization, not a replacement for it

Start from what it costs here. WorkerBuilder has no mesh option, so a MeshNode does not feed queue.worker() — the worker keeps polling storage and never looks at the local deque. Running both gives you gossip and a buffer that nothing drains. Reaching for the mesh means writing the dispatch loop yourself against queue.storage(), with should_prefetch, adaptive_prefetch_size, prefetch, pop_local and try_steal as its parts.

Worth that:

You already run a custom dispatch loop, and polling latency is the bottleneck
Several Rust nodes on the same queues, where sub-second peer detection beats a 30-second heartbeat
Work that gains from landing on the same node — warm connections, a populated cache, anything a OnceLock static holds
A fleet whose nodes come and go often enough that ring churn is cheaper than waiting out a heartbeat

Not worth it:

One node, or nodes on disjoint queues — there is nothing to steal
Low throughput, where an extra UDP protocol buys nothing
You want the stock queue.worker() — it will not use the deque, so the mesh is pure overhead
Scaling replicas is the actual goal; reach for autoscaling instead, since mesh moves work between running nodes rather than adding them