lock

README

db/lock

Postgres advisory-lock примитив — обёртка над pg_try_advisory_lock (non-blocking) и pg_advisory_lock (blocking) на dedicated pool conn. Имена хешатся в int64 ключ через первые 8 байт sha256, так что одинаковый name всегда мапится в тот же lock-key через все реплики.

Импорт: github.com/theizzatbek/gokit/db/lock Зависит от: gokit/db + pgx

Зачем это нужно

pg_advisory_lock — нативный distributed-mutex Postgres'а, удобный там, где сервис уже подключён к БД. Кит уже использует его внутри service.WithSingletonCron для leader-election'а cron-job'ов; этот пакет выставляет тот же примитив для app-кода — race-free init одноразовых задач, distributed mutex для редко-выполняемых критических секций, ad-hoc leader-election помимо cron'а.

Quickstart

import "github.com/theizzatbek/gokit/db/lock"

// One-liner: запустить fn ровно один раз через все реплики.
if err := lock.RunOnce(ctx, svc.DB, "orders.daily-rollup",
    func(ctx context.Context) error {
        return rollups.Run(ctx, svc.DB)
    }); err != nil {
    return err
}

// Низкоуровневый Try/release — для случаев, где успех/skip нужно
// разделить (логирование, метрики на выигравшую реплику).
lk := lock.New(svc.DB, "session.warmup")
acquired, release, err := lk.TryAcquire(ctx)
if err != nil { return err }
if !acquired { return nil /* другая реплика держит */ }
defer release()
// критическая секция

API-поверхность

Символ	Смысл
lock.New(d, name, opts...) *Lock	Конструирует. Panic'ит на nil d / пустом name (программерская ошибка). Опции — WithLogger, WithMetrics.
(*Lock).TryAcquire(ctx) (acquired, release, err)	Non-blocking. acquired=false — нормальный skip-путь.
(*Lock).Acquire(ctx) (release, err)	Blocking — ждёт лока или ctx-cancel.
(*Lock).TryAcquireXact(ctx, tx) (acquired, err)	Tx-scope (pg_try_advisory_xact_lock). Авто-release на commit/rollback — нет manual release.
(*Lock).IsHeld(ctx) (held, err)	Диагностика: TryAcquire → если contended, snapshot=held. Для /admin, не для control flow.
(*Lock).Name() string / (*Lock).Key() int64	Доступ к имени + derived int64-ключу для ops-запросов вроде SELECT pid FROM pg_locks WHERE objid = $1.
lock.RunOnce(ctx, d, name, fn, opts...)	Convenience: TryAcquire + run + release. Skip-fn'а на не-acquire.
lock.RunBlocking(ctx, d, name, fn, opts...)	Convenience: Acquire + run + release. Ждёт ctx-cancel.

Observability

lk := lock.New(svc.DB, "orders.daily-rollup",
    lock.WithLogger(svc.Logger),
    lock.WithMetrics(svc.Metrics),
)

WithLogger пишет (name=<lock-name> в каждом attr):

• Debug на lock: acquired, lock: contended, lock: released (held_ms=...).
• Warn на lock: acquire failed (conn drop, server down).

WithMetrics регистрирует:

• lock_acquires_total{name, outcome=acquired|contended|error} — counter.
• lock_hold_duration_seconds{name} — histogram (buckets от 1ms до 10min).

Один collector на Lock-name. Второй lock.New(d, "same-name", WithMetrics(reg)) против того же registry уйдёт в duplicate metric panic — wire каждое имя в одну New (та же конвенция, что у breaker/bulkhead).

Transaction-scope: TryAcquireXact

Когда критическая секция уже обёрнута в *db.Tx, pg_try_advisory_xact_lock снимает необходимость в manual release — лок уходит вместе с tx commit/rollback:

err := svc.DB.Tx(ctx, func(tx *db.Tx) error {
    lk := lock.New(svc.DB, "orders.batch-charge")
    ok, err := lk.TryAcquireXact(ctx, tx)
    if err != nil { return err }
    if !ok { return nil /* другая tx держит */ }

    // ... read + UPDATE rows ...
    return nil
})

Namespace — общий с session-level TryAcquire; смешивать оба варианта на одном имени безопасно. tx == nil → *errs.Error{Code: lock_acquire_failed}.

Семантика

• Session-level lock: держится на dedicated pool conn до release() или до возврата conn в пул (panic-safe).
• Авто-release на conn-close: даже если release never called (paniccing handler), session-lock уходит сам, когда conn реап'ится пулом — не зависает forever.
• Skip — не error: TryAcquire возвращает (false, nil, nil) когда другой holder; ошибка только когда что-то реально сломалось (pool exhausted, DB down).
• Deterministic key: Key() — это int64(big-endian(sha256(name)[:8])). Два одинаковых имени → один ключ; namespace per-service через префикс ("orders.daily-rollup").

Error codes

Code	Смысл
lock_acquire_failed	pg_advisory_lock errored для non-cancel причин (conn drop, server down).
lock_release_failed	pg_advisory_unlock errored. Best-effort — session-level лок уходит сам.
lock_nil_db	New(nil, …) — programmer-error.
lock_empty_name	New(d, "") — programmer-error, пустые имена коллидировали бы.

См. также

• db — обёртка над пулом
• service.WithSingletonCron — использует этот примитив внутри

Documentation

Overview

Package lock is the kit's Postgres advisory-lock primitive.

Built around `pg_try_advisory_lock` (non-blocking) and `pg_advisory_lock` (blocking) on a dedicated pool connection. Names hash to int64 via the first 8 bytes of sha256, so identical `name` strings always map to the same lock key across replicas.

Lifecycle:

lk := lock.New(svc.DB, "daily-rollup")

acquired, release, err := lk.TryAcquire(ctx)
if err != nil { return err }
if !acquired { return nil /* another replica holds it */ }
defer release()

// critical section

Convenience wrappers fold the acquire/release dance into one call:

if err := lock.RunOnce(ctx, svc.DB, "daily-rollup", fn); err != nil {
    return err
}

Used internally by

service.WithSingletonCron wraps the supplied JobFn with a `TryAcquire + run + release` block keyed by the job name — only ONE replica per multi-replica deployment runs the job per tick. Exposing the primitive lets app code do the same for any race- sensitive one-shot.

Index

• Constants
• func RunBlocking(ctx context.Context, d *db.DB, name string, fn func(context.Context) error, ...) error
• func RunOnce(ctx context.Context, d *db.DB, name string, fn func(context.Context) error, ...) error
• type Lock
  • func New(d *db.DB, name string, opts ...Option) *Lock
  • func (l *Lock) Acquire(ctx context.Context) (ReleaseFunc, error)
  • func (l *Lock) AcquireConn(ctx context.Context) (*pgxpool.Conn, ReleaseFunc, error)
  • func (l *Lock) IsHeld(ctx context.Context) (bool, error)
  • func (l *Lock) Key() int64
  • func (l *Lock) Name() string
  • func (l *Lock) TryAcquire(ctx context.Context) (bool, ReleaseFunc, error)
  • func (l *Lock) TryAcquireXact(ctx context.Context, tx *db.Tx) (bool, error)
• type Option
  • func WithLogger(l *slog.Logger) Option
  • func WithMetrics(reg prometheus.Registerer) Option
• type ReleaseFunc

Constants

const (
	// CodeAcquireFailed — pg_advisory_lock / pg_try_advisory_lock
	// errored for non-cancel reasons (conn drop, server down).
	CodeAcquireFailed = "lock_acquire_failed"

	// CodeReleaseFailed — pg_advisory_unlock errored. Session-level
	// locks auto-release when the conn returns to the pool, so this
	// is usually log-only.
	CodeReleaseFailed = "lock_release_failed"

	// CodeNilDB — Lock constructed with nil *db.DB.
	CodeNilDB = "lock_nil_db"

	// CodeEmptyName — Lock constructed with empty name. The key is
	// derived from the name, so empty would collide with every
	// other empty-name lock.
	CodeEmptyName = "lock_empty_name"
)

Stable error Code constants for lock operations.

Functions

func RunBlocking

func RunBlocking(ctx context.Context, d *db.DB, name string, fn func(context.Context) error, opts ...Option) error

RunBlocking is RunOnce's blocking sibling: waits for the lock (via Acquire), runs fn, releases. Ctx cancellation aborts the wait.

func RunOnce

func RunOnce(ctx context.Context, d *db.DB, name string, fn func(context.Context) error, opts ...Option) error

RunOnce is the convenience wrapper around TryAcquire: if the lock is free, run fn and release on return; otherwise skip silently and return nil. Errors from fn propagate unchanged so the caller's error-handling stays straightforward.

if err := lock.RunOnce(ctx, svc.DB, "daily-rollup", func(ctx context.Context) error {
    return rollup.Run(ctx)
}); err != nil { return err }

Variadic Options (WithLogger, WithMetrics) tune the internally-constructed Lock when observability is needed.

Types

type Lock

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

Lock identifies an advisory lock by name + the underlying pool. Built once at startup; thread-safe — TryAcquire / Acquire may be called from multiple goroutines, each gets its own conn from the pool.

func New

func New(d *db.DB, name string, opts ...Option) *Lock

New constructs a Lock. Panics on nil d or empty name — both are programmer errors caught at startup. The key derives from sha256(name)[:8] interpreted as big-endian int64, so two different services using the same name will fight over the same lock. Namespace per-service via prefix: "orders.daily-rollup".

Optional Options (WithLogger, WithMetrics) wire observability.

func (*Lock) Acquire

func (l *Lock) Acquire(ctx context.Context) (ReleaseFunc, error)

Acquire blocks until pg_advisory_lock succeeds OR ctx is cancelled. Use when the critical section MUST run on this replica (vs. TryAcquire's "run if I'm the winner" semantics).

Returns the release func on success; ctx errors propagate. Other underlying errors map to *errs.Error{Code: CodeAcquireFailed}.

func (*Lock) AcquireConn

func (l *Lock) AcquireConn(ctx context.Context) (*pgxpool.Conn, ReleaseFunc, error)

AcquireConn is Lock.Acquire that also hands back the pool connection the advisory lock lives on. Session-level advisory locks are bound to their connection, so callers whose critical section must not draw a SECOND conn from the pool (hash-chained audit appends on a small pool, MaxConns=1 test pools) run their queries directly on conn instead of going back to the pool.

Do NOT call conn.Release() yourself — the returned ReleaseFunc unlocks AND returns the conn to the pool, exactly like Acquire's. The conn is only valid until release is called.

func (*Lock) IsHeld

func (l *Lock) IsHeld(ctx context.Context) (bool, error)

IsHeld reports whether any backend currently holds the lock. Implementation: optimistically TryAcquires and immediately releases — if the acquire was contended, the lock is held; if it succeeded, no one was holding it at the instant of the check.

Intended for /admin observability and operator runbook scripts, NOT for control flow — by the time the caller reads the boolean, another backend may have grabbed the lock. Treat the result as a snapshot.

Returns (false, nil) when nobody holds the lock OR an error occurred during TryAcquire (the err is non-nil in the latter case, so callers can disambiguate via the err return).

func (*Lock) Key

func (l *Lock) Key() int64

Key returns the int64 advisory-lock key derived from the name. Useful for ops queries like `SELECT pid FROM pg_locks WHERE objid = $1`.

func (*Lock) Name

func (l *Lock) Name() string

Name returns the human-readable lock name.

func (*Lock) TryAcquire

func (l *Lock) TryAcquire(ctx context.Context) (bool, ReleaseFunc, error)

TryAcquire attempts pg_try_advisory_lock on a freshly-acquired pool conn. Returns (acquired=true, release, nil) when the lock is now held — caller MUST call release in a defer. Returns (false, nil, nil) when another holder has it; this is the expected "skip" path, not an error.

Acquire errors from the underlying pool / pgx surface as *errs.Error{Code: CodeAcquireFailed}.

func (*Lock) TryAcquireXact

func (l *Lock) TryAcquireXact(ctx context.Context, tx *db.Tx) (bool, error)

TryAcquireXact attempts `pg_try_advisory_xact_lock` inside tx. Unlike Lock.TryAcquire, no manual release is required — the lock is automatically released on tx commit OR rollback.

Use when the critical section is already wrapped in a *db.Tx:

err := svc.DB.Tx(ctx, func(tx *db.Tx) error {
    lk := lock.New(svc.DB, "orders.batch-charge")
    ok, err := lk.TryAcquireXact(ctx, tx)
    if err != nil { return err }
    if !ok { return nil /* another tx holds it */ }

    // ... critical section: read + UPDATE rows ...
    return nil
})

Returns (true, nil) when the lock is now held by this transaction; (false, nil) when another backend holds it. Errors from the underlying query surface as *errs.Error{Code: CodeAcquireFailed}.

Locks acquired this way:

• share the namespace with session-level Lock.TryAcquire — mixing the two on the same name interlocks correctly.
• record an `acquired` / `contended` metric outcome the same way as TryAcquire, but observeHold is NOT called (the release point is implicit, owned by Postgres tx end).

tx MUST be the *db.Tx the surrounding fn received; passing a tx from a different transaction defeats the auto-release.

type Option

type Option func(*Lock)

Option tunes New — wires optional observability.

func WithLogger

func WithLogger(l *slog.Logger) Option

WithLogger installs a *slog.Logger that records lifecycle events:

• Debug on TryAcquire/Acquire success (name, hold start).
• Debug on release (name, hold duration).
• Warn on acquire / unlock errors (name, err).
• Debug on TryAcquire returning false (contended path).

Without this option the lock runs silently.

func WithMetrics

func WithMetrics(reg prometheus.Registerer) Option

WithMetrics registers Prometheus collectors on reg, tagged with the lock's name as a const label:

• lock_acquires_total{name, outcome=acquired|contended|error} (counter)
• lock_hold_duration_seconds{name} (histogram)

One lock_* series per Lock name. The kit follows the same "one collector per instance" convention as breaker / bulkhead — constructing two Lock instances with the same name AND the same registry will panic on duplicate registration, so wire each distinct lock name to its own New.

Without this option no collectors are created (zero Prometheus footprint).

type ReleaseFunc

type ReleaseFunc func()

ReleaseFunc releases the underlying conn back to the pool AND calls `pg_advisory_unlock`. Idempotent — calling twice is a no-op. Session-level lock semantics: even if Release is never called, the lock auto-releases when the pool closes the conn, so a panicking handler doesn't leak the lock forever.