zmqcat

README

zmqcat

A ZMQ-style mailbox bus over Tailcat: WireGuard + NAT traversal, no Tailscale account, no root, no TUN device.

One side serves and prints a token. Everyone else joins with that token. Local processes (OpenResty, Python, Go, an AI harness, buzz) talk to a unix/tcp sidecar — they never see Tailcat.

  harness / openresty / pyzmq-shaped scripts
                 │
          unix socket / tcp
                 │
              zmqcat sidecar
                 │
         Tailcat (DERP → direct UDP)
                 │
              zmqcat hub
           mailboxes + topics

Install

go install github.com/pyrex41/zmqcat/cmd/zmqcat@latest

Or with Nix — nix run github:pyrex41/zmqcat -- serve --local to try it, or add the flake for the services.zmqcat module (below).

Just works

Hub (prints a tc… token):

zmqcat serve

Spoke, anywhere NAT’d:

zmqcat join tcXXXXXXXXX

Same machine, no Tailcat:

zmqcat serve --local

# durable jobs (atomic local state file)
zmqcat serve --local --mailbox ./zmqcat-mailbox.json

Then, from any process on that host:

zmqcat put jobs '{"task":"summarize","id":1}'
zmqcat take jobs
zmqcat pub harness.done '{"ok":true}'
zmqcat sub harness
zmqcat ping
zmqcat ready jobs
zmqcat req echo '{"hello":true}'

Default sidecar: unix:///tmp/zmqcat-<uid>.sock. Override with --listen tcp://127.0.0.1:5555 or ZMQCAT_LISTEN.

Run it as a service

{
  inputs.zmqcat.url = "github:pyrex41/zmqcat";
}

nixosModules.default and darwinModules.default give you services.zmqcat, which runs either role:

# the host that owns the bus
services.zmqcat = {
  enable = true;
  role = "serve";
  mailbox = "/var/lib/zmqcat/mailbox.json";  # jobs survive a restart
  allow = [ "nodekey:…" ];                   # else the token alone is enough
};

# every other host
services.zmqcat = {
  enable = true;
  role = "join";
  tokenFile = "/run/secrets/zmqcat-join-token";  # the tc… token, from a file
};

tokenFile is a path, never a literal: a string in your Nix config lands in the world-readable store.

If you are running this under huginn, its INSTALL.md covers both sides in order — one flake input pulls in this repo and re-exports these modules.

Patterns

ZMQ Guide	zmqcat
PUSH / PULL (jobs)	put / take on a named mailbox (FIFO, blocking take). Full mailbox rejects (ErrDropped); oldest is never dropped.
PUB / SUB (events)	pub / sub on a topic prefix (events. matches events.foo). Slow subscribers drop that delivery only. On sub, the last message per matching topic is replayed (last-value cache).
Majordomo-lite	ready <service> registers a competing consumer on that name; put / take / reserve on the same name share the queue. A ready delivery stays leased until the worker reps, acks, nacks, or disconnects, so a worker that dies holding a job gives it back.
Lazy Pirate	req / rep with a correlation id; the client retries with timeout and abandons after N attempts. A retry re-enqueues if the worker holding the request died, and is a no-op while the request is still queued or leased.
Heartbeats + leases	ping / pong on the same session socket; any inbound frame is liveness. Default interval ~5s, death after ~3 missed. reserve visibility leases expire and requeue; session close nacks that session's inflight.
identity	--name / hello from
trace	zmqcat serve --trace (or Config.Trace) logs frames quietly (op / id / name)

Queue cap is 1024 jobs. Pub/sub is intentionally lossy and ephemeral; mailboxes are not. The last-value cache holds at most 4096 topics and evicts arbitrarily past that — it is a convenience for late subscribers, not retention.

Existing ZeroMQ broker

If you already have libzmq on a port, punch it through the same tunnel:

# host with the real ZMQ bind
zmqcat serve --forward 5555

# elsewhere
zmqcat join "$TOKEN" --forward 5555
# localhost:5555 is now the remote broker

OpenResty

Sidecar on the nginx host (zmqcat serve or zmqcat join $TOKEN), then:

local zmqcat = require "zmqcat"  -- examples/openresty/zmqcat.lua
local c = assert(zmqcat.connect(os.getenv("ZMQCAT_LISTEN")))
c:put("jobs", '{"run":true}')
local msg = assert(c:take("jobs.out"))
ngx.print(msg.text)
c:close()

examples/openresty/nginx.conf exposes POST /mbox/:name and GET /mbox/:name.

Python harness

from zmqcat import Client   # examples/python/zmqcat.py
c = Client(name="agent")
c.put("jobs", '{"task":"ping"}')
print(c.take("jobs"))

examples/python/agent.py is a blocking worker loop: take jobs, put jobs.out, pub harness.done.

Go library

n, err := zmqcat.Serve(ctx, zmqcat.Config{})
token := n.Token()

peer, err := zmqcat.Join(ctx, token, zmqcat.Config{})
c, err := zmqcat.Dial(peer.Listen())
c.Put("inbox", "hello", nil)

Wire protocol

Local and tunneled sessions are the same:

magic "ZMQC" | uint32be length | JSON

{"v":1,"op":"put","name":"jobs","from":"agent","text":"..."}

Ops: hello, put, take, pub, sub, unsub, ping, pong, reserve, ack, nack, ready, req, rep → ok / err / msg / pong / rep. Binary payloads use JSON body (standard base64). text is the UTF-8 convenience field. Correlation id is required for req/rep (retries reuse the same id).

An id on put or req also serves as the mailbox message id, which is what makes a retry idempotent. It must be unique per client — the shipped Go, Python, and Lua clients derive it from a random per-connection prefix. The hub additionally scopes ids by session, so two clients that pick the same id cannot deduplicate one another; ids are only compared within one connection. take and reserve block until a message is available.

Security

The Tailcat token is the capability. Anyone with it can join the bus.

• Ephemeral keys by default (new token every serve).
• --allow nodekey:… to pin client identities (from tailcat genkey --client).
• --forward is a hole to localhost; only expose ports you mean.

Public Tailcat DERP relays are rate-limited and have no SLA. Bring your own: tailcat --region=derp.example.com / DERPMapURL.

Tailcat itself has no API or CLI stability promise; zmqcat will track it.

Why not “just Tailcat + netcat”?

Tailcat is a pipe. zmqcat is a bus: many clients, named mailboxes, topics, a stable local socket so nginx workers and agents do not each bring up WireGuard.

Durable mailboxes (v2)

Set Config.MailboxPath when serving to persist queued and in-flight messages (JSON, rewritten in full and fsynced before an atomic rename; suitable for modest orchestrator traffic, not for high throughput — every put, take, ack, and nack costs one file rewrite under the bus lock). The Go client exposes Reserve, Ack, and Nack. A reservation uses a visibility lease: acknowledgement removes it, nack, lease expiry, or session disconnect redelivers it, providing at-least-once delivery. Delivery IDs are unique and message IDs are generated when omitted. Payloads and mailbox names retain the existing bounds. Put rejects when the mailbox is full (ErrDropped) and never drops the oldest job.

Delivery semantics differ by op. take is at-most-once: it acknowledges as soon as the frame is written, so a consumer that dies mid-processing loses the job. reserve and ready are at-least-once: the job stays leased until acknowledged, and a worker that dies holding one gives it back. Consumers must therefore tolerate duplicate deliveries and acknowledge only after successful processing.

Durability is local to the single serving hub; Tailcat provides transport encryption but mailbox-level identity/ACLs are not yet implemented. Anything that can reach the sidecar socket can read and write every mailbox. Pub/Sub remains intentionally lossy and ephemeral.

Documentation

Overview

Package zmqcat is a ZMQ-style mailbox bus over Tailcat.

One process serves (prints a tailcat token). Others join with that token. Local processes talk over a unix/tcp socket so OpenResty, Python, and AI harnesses do not need Tailcat or libzmq.

Index

• Constants
• Variables
• type Client
  • func Dial(listen string) (*Client, error)
  • func (c *Client) Ack(delivery string) error
  • func (c *Client) Close() error
  • func (c *Client) Hello(name string) error
  • func (c *Client) Nack(delivery string) error
  • func (c *Client) Ping() error
  • func (c *Client) Pub(topic, text string, body []byte) error
  • func (c *Client) Put(name, text string, body []byte) error
  • func (c *Client) Ready(service string) (wire.Frame, error)
  • func (c *Client) Recv() (wire.Frame, error)
  • func (c *Client) Rep(id, name, text string, body []byte) error
  • func (c *Client) Request(name, text string, body []byte, timeout time.Duration, attempts int) (wire.Frame, error)
  • func (c *Client) Reserve(name string, lease time.Duration) (wire.Frame, error)
  • func (c *Client) Sub(prefix string) error
  • func (c *Client) Take(name string) (wire.Frame, error)
• type Config
• type Node
  • func Join(ctx context.Context, token string, cfg Config) (*Node, error)
  • func Serve(ctx context.Context, cfg Config) (*Node, error)
  • func (n *Node) Close() error
  • func (n *Node) Listen() string
  • func (n *Node) Token() string

Constants

const MailboxPort uint16 = 7

MailboxPort is the TCP port on the Tailcat server that speaks zmqcat.

Variables

var (
	// ErrAbandoned is returned by Request after retries are exhausted.
	ErrAbandoned = errors.New("zmqcat: request abandoned")
	// ErrDesync means a read was interrupted part way through a frame, so the
	// remaining bytes of that frame are still queued. The session cannot be
	// resynchronized; redial instead.
	ErrDesync = errors.New("zmqcat: session desynchronized, redial required")
)

Types

type Client

type Client struct {
	Conn net.Conn
	Name string
	// contains filtered or unexported fields
}

Client is a single session on a hub (local unix/tcp or a spliced tunnel).

func Dial

func Dial(listen string) (*Client, error)

Dial opens a session to a local sidecar listen address.

func (*Client) Ack

func (c *Client) Ack(delivery string) error

func (*Client) Close

func (c *Client) Close() error

func (*Client) Hello

func (c *Client) Hello(name string) error

func (*Client) Nack

func (c *Client) Nack(delivery string) error

func (*Client) Ping

func (c *Client) Ping() error

func (*Client) Pub

func (c *Client) Pub(topic, text string, body []byte) error

func (*Client) Put

func (c *Client) Put(name, text string, body []byte) error

func (*Client) Ready

func (c *Client) Ready(service string) (wire.Frame, error)

Ready registers as a competing consumer for service and waits for one job.

func (*Client) Recv

func (c *Client) Recv() (wire.Frame, error)

func (*Client) Rep

func (c *Client) Rep(id, name, text string, body []byte) error

func (*Client) Request

func (c *Client) Request(name, text string, body []byte, timeout time.Duration, attempts int) (wire.Frame, error)

Request is Lazy Pirate: req/rep with a correlation id, timeout, and retries. Duplicate delivery is possible; the same id is reused across attempts.

func (*Client) Reserve

func (c *Client) Reserve(name string, lease time.Duration) (wire.Frame, error)

func (*Client) Sub

func (c *Client) Sub(prefix string) error

func (*Client) Take

func (c *Client) Take(name string) (wire.Frame, error)

type Config

type Config struct {
	// MailboxPath enables durable at-least-once mailboxes on the serving node.
	// Empty keeps the historical in-memory behavior.
	MailboxPath string
	// Heartbeat is the session liveness interval. Zero means 5s; negative disables.
	Heartbeat time.Duration
	// Trace logs each ZMQC frame (session, direction, op/id/name).
	Trace bool
	// Listen is the local sidecar address (unix:// or tcp://). Empty uses
	// unix:///tmp/zmqcat-<uid>.sock.
	Listen string
	// Name is this node's hello identity.
	Name string
	// Logf logs diagnostics. Nil uses log.Printf. Set to a no-op to hush.
	Logf func(string, ...any)
	// Quiet suppresses Tailcat's own chatter.
	Quiet bool
	// AllowedClients, if non-empty, is a Tailcat nodekey allowlist.
	AllowedClients []key.NodePublic
	// DERPMapURL overrides Tailcat's default DERP map.
	DERPMapURL string
	// LocalOnly skips Tailcat (tests, same-host bus).
	LocalOnly bool
	// ForwardPorts are extra TCP ports on localhost to expose through the
	// tunnel (a real libzmq bind, buzz, whatever). Serve dials
	// 127.0.0.1:port; Join listens on 127.0.0.1:port and dials the server.
	ForwardPorts []uint16
}

Config is Serve/Join options.

type Node

type Node struct {
	// contains filtered or unexported fields
}

Node is a running serve or join sidecar.

func Join

func Join(ctx context.Context, token string, cfg Config) (*Node, error)

Join connects to a Serve token, then exposes the same local sidecar. Each local connection is a Tailcat TCP session to MailboxPort.

func Serve

func Serve(ctx context.Context, cfg Config) (*Node, error)

Serve starts a mailbox hub, a local sidecar, and (unless LocalOnly) a Tailcat server. The token is available from Node.Token after return.

func (*Node) Close

func (n *Node) Close() error

func (*Node) Listen

func (n *Node) Listen() string

func (*Node) Token

func (n *Node) Token() string