Mesh Scheduling

The mesh layer sits between the scheduler and the storage backend, forming a decentralized overlay network that reduces DB contention and enables sub-millisecond load awareness.

For usage and configuration, see the Mesh Scheduling guide.

Live demonstration

Five workers run in different regions, linked into one mesh over HTTP. Jobs hash-route into the cluster — but enqueues are deliberately skewed toward us-east-1/us-west-2, so their queues build up. Watch idle regions steal batches from the busiest peer across the network (cyan, with real inter-region latency), SWIM gossip flicker along the mesh links, and the counters track throughput. Hover a region to see its links and latencies; hit Add burst to flood the queue.

Crate structure

The mesh is implemented as a standalone Rust crate (crates/flexiq-mesh/) with no PyO3 dependency. Feature-gated via mesh cargo feature on flexiq-python. Depends on flexiq-core for the Job type only.

SWIM gossip protocol

Each mesh worker runs a SWIM-based gossip loop on a UDP socket. SWIM was chosen over full-broadcast gossip (like Serf) because its protocol overhead is O(1) per period regardless of cluster size.

Protocol period

Every 500ms (configurable via protocol_period_ms in MeshConfig), a node:

1. Picks a random alive peer and sends a Ping (with sequence number).
2. If no Ack within protocol_period / 2 → sends PingReq to indirect_ping_count (3) random intermediaries to probe the target on its behalf.
3. If still no response → marks the target as Suspect.
4. After suspicion_multiplier × ln(N+1) × protocol_period → declares Dead and removes from the hash ring.

Message types

All messages are bincode-encoded and fit within a single UDP datagram (<1400 bytes):

Membership states

Piggybacked dissemination

Every gossip message (Ping, Ack, PingReq) carries piggybacked MemberUpdate payloads — new members, state changes, incarnation bumps. This achieves O(log N) convergence without dedicated protocol rounds.

ACK responses include the full list of known alive peers, ensuring that even nodes with no direct connection converge quickly. Example: node C seeds from node A only, but learns about node B through A's ACK — without B and C ever communicating directly.

The pending update queue holds up to 64 updates, with a maximum of 8 piggybacked per message to stay within UDP datagram limits.

Incarnation numbers

When a node sees itself reported as Suspect, it increments its own incarnation number and broadcasts a refutation. The rule is simple: higher incarnation always wins. At the same incarnation, higher-severity state wins (Alive < Suspect < Dead < Left). This prevents false-positive failure detection from cascading across the cluster.

Encryption

Optional XOR symmetric encryption with a shared key (base64-encoded). Applied to every UDP datagram before send, reversed on receive. All cluster nodes must share the same key. See Gossip encryption for setup.

Consistent-hashing ring

Implemented in ring.rs using a BTreeMap<u64, String> that maps xxh3 hash values to worker IDs:

• Each worker gets virtual_nodes (default 150) entries in the ring, hashed as xxh3_64("{worker_id}-{i}") for i in 0..virtual_nodes
• Lookup: preferred_worker(key) computes xxh3_64(key) and walks clockwise (via BTreeMap::range) to find the first entry

Verified properties

The ring's properties are validated by unit tests in crates/flexiq-mesh/src/ring.rs:

• Even distribution: 150 virtual nodes across 3 workers distributes 3000 keys within 700–1300 each (within ±2× of theoretical 1000)
• Minimal migration: removing one of 3 workers migrates <60% of keys (theoretical minimum is 33%)
• Deterministic: same ring state → same placement across all nodes, verified by inserting/removing/re-inserting workers

The ring recalculates on every membership change (join, leave, death). The ring_recalculations metric tracks frequency.

Local deque

Each worker maintains a Mutex<VecDeque<Job>> as a local job buffer with a configurable capacity (local_buffer_capacity, default 64):

• Prefetch: the scheduler dequeues a batch from the DB and pushes into the deque via push_sorted()
• Affinity sorting: push_sorted() partitions the batch — non-owned tasks go to the front (stealable), owned tasks to the back (hot)
• Pop (LIFO): the owner calls pop() which takes from the back — executing affinity-matched tasks first
• Steal (FIFO): thieves call steal(n) which takes from the front — preferring tasks that don't benefit from local affinity

Why not crossbeam::deque?

crossbeam::deque::Worker is !Send. Since the mesh bridge runs inside tokio::spawn (which requires Send futures), Mutex<VecDeque> is used instead. Lock contention is minimal — only the owner thread, the prefetch path, and occasional steal requests touch the deque, and critical sections are short (pop/push, no I/O under lock).

Work-stealing protocol

Wire format

TCP with length-prefixed bincode framing:

Each frame is a 4-byte big-endian length prefix followed by the bincode payload (max 1 MB). The StealRequest carries thief_id and max_count. The StealResponse carries Vec<Job> — full job structs including payload, metadata, and scheduling info.

Sequence diagram

Timeouts: 500ms connect, 2s response read. On any failure, returns empty vec — stealing is best-effort.

Target selection

try_steal() queries the gossip state for alive peers, sorts by local_buffer_len descending, and picks the busiest. This is eventually consistent — the buffer length may have changed since the last gossip update — but convergence is fast enough (<1s) that the target is usually still the best choice.

Rate limiting

The steal server (steal/server.rs) maintains a per-peer sliding window rate limiter. For each thief_id, it tracks request timestamps in the last 1 second. If the count exceeds steal_rate_limit (default 10), the request gets an empty response (no error, no disconnect). Setting steal_rate_limit=0 disables limiting.

Adaptive load balancing

Two mechanisms tune dispatch without a central coordinator:

Prefetch sizing

adaptive_prefetch_size() scales the batch size based on peer count:

base_size = local_buffer_capacity / 4
scale     = 1.0 / (peer_count + 1)
prefetch  = max(1, base_size × scale)

With default capacity 64: standalone worker prefetches 16, in a 3-node cluster each prefetches ~4. This distributes DB load evenly.

Poll jitter

poll_jitter_ms() uses the worker's hash-ring position to stagger DB polls:

jitter = xxh3_64(worker_id) % (protocol_period_ms / 2)

The jitter is deterministic per worker ID, so it's stable across restarts. This prevents thundering herd when multiple workers poll simultaneously after a period of queue emptiness.

Integration with the scheduler

The mesh does not modify the Scheduler struct or the WorkerDispatcher trait. Instead, the per-SDK worker entrypoint spawns a mesh bridge — an intermediate tokio::sync::mpsc channel between the scheduler and dispatcher:

The run_mesh_bridge() function:

1. recv from job_rx (scheduler channel) — push into the local deque.
2. pop from the local deque — send to dispatch_tx (dispatcher channel).
3. On idle (no jobs from scheduler, deque low) → try_steal() from the busiest peer.
4. Loop until shutdown.

This keeps the scheduler and dispatcher completely unaware of mesh logic. Without the mesh feature flag, the scheduler sends directly to the dispatcher as before.

Metrics

MeshMetrics tracks eight AtomicU64 counters, all lock-free:

Access via MeshNode::metrics() which returns a MetricsSnapshot (all values cloned atomically).

Failure modes

In every failure case, the database remains the source of truth. The worst outcome is reduced performance (more DB polls), never data loss or two workers running the same job at once — the storage layer's atomic claim mechanism gives exactly-once dispatch. Execution itself is at-least-once: a crash mid-task is retried, so the same task body can run more than once. Design tasks to be idempotent — see the failure model.

See also

• Mesh Scheduling guide — configuration, cluster topology, Docker Compose example
• Scheduler — the poll loop that feeds jobs into the mesh bridge
• Worker Pool — the dispatcher that receives jobs from the mesh bridge
• Storage — atomic claim mechanism that guarantees exactly-once dispatch
• Failure Model — how flexiq handles crashes and timeouts