rchan

Remote Go channels, backed by Redis Streams.

Two processes on different machines share what looks and behaves like a Go channel: one side runs sc <- v, the other runs for v := range ch.C(), and everything you expect from a channel carries across the network — blocking sends on a full buffer, true rendezvous on unbuffered channels, close propagation, select (including non-blocking try-send), len/cap, and panic-on-send-after-close.

// process A (producer)                 // process B (consumer)
ch, _ := rchan.New[Job](ctx, rdb, "jobs",    ch, _ := rchan.New[Job](ctx, rdb, "jobs",
    rchan.WithCapacity(16))                 rchan.WithCapacity(16))
sc := ch.SendC()                        for job := range ch.C() {
sc <- Job{ID: 1}   // blocks when 16        process(job) // exactly one
sc <- Job{ID: 2}   // are in flight     }
close(sc)          // close(ch)         // loop ends: closed + drained

Experiment-grade software. Built and tested like a real library (race detector, exactly-once stress tests, adversarial reviews), but the design goal is semantic fidelity, not production hardening.

---

Why

Go channels are the best concurrency API ever shipped. They stop at the process boundary. Message queues cross the boundary but hand you a different mental model — consumers, acks, visibility timeouts, redelivery.

rchan keeps the channel model and pays the distributed-systems tax under the hood: the ack machinery, crash recovery, and backpressure signaling all exist, but the API surface stays make / <- / close / select.

Quick start

go get github.com/ndyakov/rchan

rdb := redis.NewClient(&redis.Options{Addr: "localhost:6379"})

ch, err := rchan.New[string](ctx, rdb, "greetings", rchan.WithCapacity(8))
if err != nil { ... }
defer ch.Stop()

// Method API — explicit contexts and errors.
err = ch.Send(ctx, "hello")     // blocks while the buffer is full
v, ok, err := ch.Recv(ctx)      // ok=false once closed and drained
err = ch.Close(ctx)             // like close(ch), cluster-wide

// Operator API — real Go chans, pumped by background goroutines.
sc := ch.SendC()                // chan<- string
rc := ch.C()                    // <-chan string
sc <- "hi"
v = <-rc
close(sc)                       // maps to ch.Close()

Run the demos:

make up                              # redis:8.2 via docker compose
go run ./example -mode demo          # select over two channels + rendezvous ping-pong
go run ./example -mode produce -n 10 # cross-process: watch send #5 block...
go run ./example -mode consume       # ...until this drains it (other terminal)

Semantics scorecard

Go	rchan	notes
make(chan T, n)	New[T](ctx, rdb, name, WithCapacity(n))	shared by name across processes; capacity/codec validated against existing channel
make(chan T)	New[T](ctx, rdb, name) — capacity 0	true rendezvous: sender parks until pickup
ch <- v	Send(ctx, v) or sc <- v via SendC()	blocks on full buffer / until rendezvous pickup
v, ok := <-ch	Recv(ctx) or <-ch.C()	ok=false = closed and drained
for v := range ch	for v := range ch.C()	ends after close + drain
close(ch)	Close(ctx) or close(sc)	remaining values drainable everywhere
select receive	select over multiple C()s	native
select send + default	select { case sc <- v: default: }	default fires while buffer full / no receiver present
len(ch), cap(ch)	Len(ctx), Cap()	Len counts delivered-but-unacked values too
send on closed → panic	native panic (local); pump panic (remote close race)	Go's close-under-active-senders doctrine, networked
exactly-once handoff	at-least-once (default) or at-most-once	the one physics gap; see Delivery guarantees

How it works

One channel = one Redis stream + a consumer group

Everything for channel name lives under hash-tagged keys (one cluster slot):

key	type	role
rchan:{name}:s	stream	the buffer; every value is one entry (v = codec bytes, rz = rendezvous reply id for capacity-0 sends)
rchan:{name}:meta	hash	cap, codec, closed — validated by New, gate for every send
rchan:{name}:rz:<id>	list	rendezvous reply: receiver LPUSHes on pickup, sender BLPOPs
rchan:{name}:rd	zset	receiver-presence beacons (server-time deadlines) for the native-send reader gate
rchan:{name}:sig	pub/sub	wakeups: space (slot freed), reader (receiver arrived), close

All multi-step state transitions are Lua scripts, so every gate is atomic:

• send gate — closed? (missing meta counts as closed, so a Send racing Destroy can't resurrect the channel) → XLEN < cap? → XADD. Returns CLOSED, FULL, or the entry id.
• ack — XACK + XDEL for one or many ids, then PUBLISH space only on the full→not-full transition. Capacity is freed by acking, so XLEN is the occupancy count and len(ch)-style accounting falls out for free.
• close — flip closed, broadcast. No sentinel entry (a consumer group would deliver it to exactly one consumer — the wrong semantics).
• drained — closed == 1 && XLEN == 0. Since entries are only deleted on ack, XLEN == 0 also proves no value is in flight anywhere.
• retract — remove a canceled rendezvous entry only if no consumer has it (XPENDING check); otherwise the send counts as completed.

Receiving: competing consumers, exactly one delivery

Every handle joins consumer group recv with a unique consumer name. Redis guarantees each entry is delivered to exactly one group member. Recv loops XREADGROUP BLOCK 1s, checking the closed flag between blocks.

At-least-once (default): the value is acked after hand-off — Recv acks on return, C()'s pump acks after your goroutine actually reads from the proxy chan, RecvAck/Ack let you ack after processing. A consumer that crashes mid-processing leaves the entry in its pending list (PEL); every receiving handle runs a reaper (XAUTOCLAIM every visibility/2) that steals entries idle past the visibility timeout and redelivers them. Nothing is lost; duplicates are possible exactly there.

At-most-once (WithAckMode(AtMostOnce)): acked at read, before your code sees the value. A crash after the ack loses that value. No duplicates.

Rendezvous (capacity 0)

Send on an unbuffered channel:

1. XADD the value with a reply id (rz), no capacity gate — each blocked sender owns exactly one parked entry, like N goroutines blocked on an unbuffered chan.
2. BLPOP the reply key in 1s slices (go-redis blocking commands don't abort on ctx cancellation mid-read, so the loop checks ctx and channel existence between slices).
3. The receiver that picks the entry up LPUSHes the reply — that unblocks the sender. Rendezvous completes at pickup, like Go: what the receiving app does afterwards is not the sender's business.
4. Cancel while parked → atomic retract (unless a receiver already has it — then the send counts as delivered). Crash of the receiver between pickup and reply → the reaper reclaims the entry and the reply still fires: a parked sender survives receiver crashes.

Native operators: pumps and gates

C() and SendC() return real Go chans serviced by per-handle goroutines. The trick that keeps the operators honest is gated listening: a pump only performs the proxy-chan operation when the remote side could make progress.

• C()'s pump claims one value (or up to WithRecvBatch(n)) and blocks pushing it into an unbuffered proxy — your <-rc is the hand-off, and in at-least-once mode the ack fires right after it.
• SendC()'s pump listens on the send proxy only while: the buffer has space (cap > 0, woken by the space transition publish), or a live receiver beacon exists (cap 0 — receivers advertise themselves in the rd zset, scored by Redis server time, every read cycle). Not listening is what makes select { case sc <- v: default: } take default — and what makes a rendezvous sc <- v block until a reader exists somewhere.
• close(sc) maps to Close. Send-after-close panics: locally Go's runtime does it for you; a remote Close/Destroy racing an in-flight forward panics the pump — the same doctrine (closing while senders are active is the closer's bug), extended across the network.

Lifecycle

• New — idempotent; creates stream/group/meta or validates against them.
• Stop — detaches this handle (pumps, reaper, gates die). Not Close.
• Close — the channel-wide close(ch): no new sends, drain then ok=false.
• Destroy — deletes every key. Parked senders notice within ~1s and return ErrClosed; a late Send cannot resurrect the channel.

Delivery guarantees, honestly

scenario	at-least-once (default)	at-most-once
consumer crash mid-processing	value redelivered (duplicate possible)	value lost
producer crash mid-Send	value either in the stream or not; sender never got a false success	same
Recv returned, ack lost (network)	ok=false, err — value redelivers	n/a (acked at read)
async-ack tail at crash (WithAsyncAck)	unflushed acks redeliver	n/a (combo rejected)
rendezvous receiver crashes after pickup	sender already unblocked; value redelivers to another receiver	combo rejected
reader-gate staleness (SendC, cap 0)	sc <- v can complete up to ~3.5s after the last reader died; value waits for the next receiver	—
try-send accuracy	default reflects remote state within one gate slice; spurious "full" while pump is mid-forward (~1 RTT)	—

Ordering: FIFO per channel through one consumer, except redeliveries arrive late — exactly the caveat every at-least-once system carries.

Not offered: exactly-once (impossible over a network without cooperative dedup), broadcast (every receiver sees every value — use plain pub/sub or one channel per subscriber).

Performance

Defaults are correctness-first: ~3 round trips per value (send gate, read, ack). Two opt-in knobs amortize that while keeping the at-least-once contract:

• WithRecvBatch(n) — claim up to n entries per read, hand them out locally. Keep WithVisibilityTimeout above the expected buffer dwell.
• WithAsyncAck() — acks flow through a self-clocking background batcher; a crash redelivers the unflushed tail instead of losing it.

M4 Max, redis:8.2 under Docker Desktop (go test -bench=. -benchtime=2s; PING floor ~215µs on this setup):

benchmark	per value
RedisPing (floor)	~215µs
SendRecvSerial	~660µs
Pipelined1P (defaults)	~530µs (~1.9k msg/s)
Pipelined8PBatch64	~480µs
Pipelined8PBatch64Async	~87µs (~11.5k msg/s)
RendezvousHandoff	~820µs

Batching + async acks push per-value cost below one round trip. Absolute numbers are dominated by the VM network boundary; native-Linux localhost is several times faster, real networks slower. Remaining ideas live in TODO.md.

Use cases

Work queue with real backpressure. The classic. Capacity is enforced at the source: when consumers fall behind, producers block (or see try-send default), instead of an unbounded queue quietly growing. Competing consumers scale horizontally; crashes redeliver.

Cross-service hand-off with rendezvous. Capacity 0 means "this call returns when the other service has the value in hand" — a synchronization point, not a mailbox. Deploy gates, migration steps, lock-step batch pipelines: sc <- batch returns exactly when the downstream picked it up.

Fan-in / fan-out pipelines with select. Each stage is a process; stages select over multiple channels natively:

for {
    select {
    case job := <-jobs.C():      // work-stealing across processes
        results.SendC() <- run(job)
    case cmd := <-control.C():   // control plane on a second channel
        reconfigure(cmd)
    }
}

Graceful drain on shutdown. Close is cluster-wide and drain-correct: producers stop (send → ErrClosed / native panic per Go doctrine), consumers finish the backlog, range loops end everywhere. One call.

Bounded job admission. Len/Cap + try-send give load shedding:

select {
case sc <- req:
default:
    http.Error(w, "busy", 503)   // buffer full: shed instead of queue
}

Migrating a channel-shaped monolith. Code already structured around channels splits into processes without changing the concurrency model — replace make(chan T, n) with rchan.New, keep the select loops.

Poor fits. Broadcast/event-sourcing (use streams or pub/sub directly); financial exactly-once (needs idempotency keys and dedup regardless of transport); >100k msg/s firehoses (use raw batched streams — the channel abstraction's per-value semantics are the cost); values that must survive Redis loss (streams are as durable as your Redis persistence).

Options reference

option	default	effect
WithCapacity(n)	0 (rendezvous)	buffer size; enforced at send
WithAckMode(m)	AtLeastOnce	AtMostOnce = ack at read
WithCodec(c)	JSON	any Marshal/Unmarshal/Name impl; validated across handles
WithVisibilityTimeout(d)	30s	idle time before the reaper redelivers a claimed value
WithRecvBatch(n)	1	entries claimed per read (at-least-once, cap > 0)
WithAsyncAck()	off	batched background acks (at-least-once)
WithPollInterval(d)	1s	jittered wakeup fallback for blocked senders and gates
WithLogger(l)	log.Default()	background goroutine diagnostics

Development

make up      # redis:8.2 on localhost:6395 (docker compose)
make test    # full suite, race detector
make down

Requires Go 1.25+ and Redis ≥ 6.2 (streams + XAUTOCLAIM); tested against 8.2. Standalone Redis; keys are hash-tagged so cluster should work, but it is untested (Destroy's SCAN is standalone-only).

Credits

The Go gopher in the logo was designed by Renee French; the vector version is by Takuya Ueda (CC BY 3.0).