latch is a Go package that provides distributed coordination primitives — advisory locking and pub/sub signaling — backed by PostgreSQL. It lets multiple processes safely coordinate work and exchange events using a database they already have.
go get github.com/supazonic/latchUse advisory locks to ensure only one process runs a critical section at a time.
import (
"context""database/sql"
_ "github.com/lib/pq""github.com/supazonic/latch/postgres"
)
dsn:=os.Getenv("DATABASE_URL")
db, _:=sql.Open("postgres", dsn)
l:=postgres.New(db, dsn)
deferl.Close()
acquired, err:=l.AcquireLock(ctx, 42)
iferr!=nil {
// handle error
}
ifacquired {
deferl.ReleaseLock(ctx, 42)
// only one process reaches here at a time
}A PostgreSQL advisory lock belongs to a session, not to a process, and it is released the instant that session ends. If the connection holding it dies, the server hands the key to whoever asks next while this process still believes it holds it — which is the one thing the lock existed to prevent. latch checks the connection behind every held lock and reports the loss:
l:=postgres.New(db, dsn, postgres.WithLockLost(func(keyint64, errerror) {
log.Printf("latch: lost lock %d: %v", key, err)
// another process can hold this key now; stop the work it was guarding
}))A connection that stops answering counts as a lost lock even though its session
may still be alive: giving up a lock you still hold costs progress, whereas
keeping one you have already lost is a correctness bug. Close releases every
lock still held.
Checks run every DefaultLockPingInterval (15s) and one may go unanswered for
DefaultLockPingTimeout (5s) before the lock is given up, so a lost lock is
reported within 20 seconds. Tune both with WithLockPing:
// tolerate a database that stalls for up to a minute under loadpostgres.New(db, dsn, postgres.WithLockPing(15*time.Second, time.Minute))Raise the timeout if your database stalls and the work a lock guards is expensive to redo; lower the interval to hear about a genuinely lost lock sooner, at the cost of more round trips.
Send and receive events across processes using PostgreSQL channels. Payloads
are []byte, so you can send whatever encoding you like.
// Register handlers. Whatever a handler returns is the response for the// round trip — handlers never send it back themselves.handlers:=map[latch.Event]latch.Handler{
"jobs": func(ctx context.Context, payload []byte) ([]byte, error) {
fmt.Println("received:", string(payload))
return []byte("done"), nil
},
}
err:=l.Listen(ctx, handlers) // runs in background until ctx is cancelled// Send a notification and block until the pod running the handler returns.resp, err:=l.Notify(ctx, "jobs", []byte("payload-here"))
iferr!=nil {
// handle error — includes any error the remote handler returned
}
fmt.Println("response:", string(resp)) // "done"Listen returns immediately and dispatches notifications in a goroutine until
the context is cancelled.
New takes both a *sql.DB and the connection string, because LISTEN needs a
connection to itself for as long as the subscription lives and database/sql
reclaims connections for its pool. Listen and Subscribe therefore open their
own, outside the pool; everything else — Notify, locks, the payload store —
uses the *sql.DB you pass. Close releases the connection shared by
Subscribe; listeners started by Listen end with their context.
A listener is the one part of latch that no query will tell you about: if its
connection dies, it simply stops hearing anything. Lost connections are
re-established automatically and every channel re-subscribed, but PostgreSQL
does not queue notifications for a listener that is not there, so anything sent
during the outage is gone. Register WithEvents to hear about it and resync
whatever state the notifications were driving.
l:=postgres.New(db, dsn, postgres.WithEvents(func(s postgres.ConnState, errerror) {
log.Printf("latch listener: %s: %v", s, err)
ifs==postgres.Reconnected {
// notifications sent while the connection was down were not queued
}
}))A server that stops answering without dropping the connection is detected by
ping and the connection replaced, which takes up to 15 seconds; that reports as
postgres.Stalled.
Notify is a round trip: it blocks until the handler on the receiving process
returns, then delivers that result as the response. Use ctx to bound the
wait. If the handler returns an error, Notify returns it on the calling side.
To wait for a single event without sending one, use Subscribe:
sub, err:=l.Subscribe(ctx, "jobs")
iferr!=nil {
// handle error
}
defersub.Close()
payload, err:=sub.Wait(ctx) // blocks until a notification arrives or ctx is doneBecause pg_notify carries text, round-trip payloads are wrapped in a JSON
envelope and base64 encoded. PostgreSQL caps a notification at 7999 bytes,
which leaves about 5.9 KB for your payload. Past that, Notify fails.
Rather than making callers work around this, hand latch a PayloadStore and it
sends oversized payloads by reference: the bytes go to the store, only the key
travels over the channel, and the receiving process loads them back before the
handler runs. Callers and handlers see the payload either way.
store:=postgres.NewPayloadTable(db)
store.CreateTable(ctx) // or run store.Schema() through your migration tooll:=postgres.New(db, dsn, postgres.WithStore(store))
// 1 MB payload: stored as a row, notification carries an ~80 byte reference.resp, err:=l.Notify(ctx, "jobs", bigPayload)Payloads of postgres.DefaultInlineLimit (4096) bytes or fewer still go inline
with no extra query; tune with postgres.WithInlineLimit. Replies are offloaded
by the same rule.
PayloadTable expires rows on a TTL (default 1 hour) instead of deleting them
on read, because every process listening on a channel receives the notification
and any number of them may resolve the same key. Call DeleteExpired on a timer
to reclaim space.
To store payloads somewhere else — object storage, a cache, another table shape — implement the interface:
typePayloadStoreinterface {
Put(ctx context.Context, payload []byte) (string, error)
Get(ctx context.Context, keystring) ([]byte, error)
}It must be reachable by every process that might receive the notification, so per-process memory will not do.
The latch.Coordinator interface (combining Locker and Signaler) lets you swap in any backend:
typeCoordinatorinterface {
AcquireLock(ctx context.Context, keyint64) (bool, error)
ReleaseLock(ctx context.Context, keyint64) errorNotify(ctx context.Context, eventEvent, payload []byte) ([]byte, error)
Subscribe(ctx context.Context, eventEvent) (Subscription, error)
Listen(ctx context.Context, handlersmap[Event]Handler) error
}