Documentation
¶
Overview ¶
Package distributedlock provides a pessimistic mutual-exclusion atom for coordinating exclusive access to a named resource across processes. Provider implementations live in subpackages and are selected at runtime via distributedlock/config.
The interface is intentionally narrow: Acquire/Release/Refresh, with no built-in retry loop or queueing. Callers compose Acquire with platform/retry, the platform circuit breaker, or their own backoff strategy. Higher-level concerns such as leader election, distributed cron, and exactly-once batch execution are compositions on top of this atom and live in consuming applications, not in platform.
For the common run-fn-while-held shape, ScopedLocker (WithLock/TryWithLock) removes the handle entirely: the lock is released when fn returns, panics included. The postgres provider implements it natively with transaction-scoped advisory locks — waiters queue server-side, a crashed holder's lock dies with its connection, and no session or connection is pinned beyond fn's duration. Any other Locker gains the same surface through the NewScopedLocker adapter, which polls a contended WithLock on a configurable interval that backs off exponentially and is jittered, so a crowd of waiters on one key does not hammer the underlying store. Prefer ScopedLocker unless the hold genuinely must outlive a function scope (e.g. a lock held across asynchronous work), where the raw Acquire/Release handle remains the right tool.
Both ScopedLocker implementations emit the same shape of telemetry, so a dashboard survives a provider swap: a span covering the whole acquire-run-release operation, plus acquire, contention, and error counters and a latency histogram. Contention is counted once per call in both, even though the generic adapter reaches it by polling and postgres by waiting server-side; the adapter additionally reports how long it waited (scoped_lock_wait_ms) and how many polls that took. Watch scoped_lock_release_failures in particular: it means a lock's TTL elapsed while fn was still running, so mutual exclusion was not actually held for the whole call.
Provider semantics differ in one important respect: the redis and memory providers enforce TTLs natively, while the postgres provider's TTL is advisory only — the underlying pg_advisory_lock is held until either Release is called or the dedicated session is closed. See distributedlock/postgres for details.
Index ¶
- Constants
- Variables
- type Lock
- type Locker
- type ScopedLocker
- type ScopedOption
- func WithLogger(logger logging.Logger) ScopedOption
- func WithMetricsProvider(metricsProvider metrics.Provider) ScopedOption
- func WithScopedClock(c clock.Clock) ScopedOption
- func WithScopedJitter(fn func() float64) ScopedOption
- func WithScopedLockTTL(ttl time.Duration) ScopedOption
- func WithScopedPollBackoff(factor float64, maxInterval time.Duration) ScopedOption
- func WithScopedPollInterval(interval time.Duration) ScopedOption
- func WithTracerProvider(tracerProvider tracing.TracerProvider) ScopedOption
Examples ¶
Constants ¶
const ( // DefaultScopedLockTTL is the TTL the generic scoped adapter passes to // Acquire when WithScopedLockTTL is not supplied. DefaultScopedLockTTL = 30 * time.Second // DefaultScopedPollInterval is how long the generic scoped adapter waits // before its first re-try of a contended WithLock acquisition, when // WithScopedPollInterval is not supplied. Subsequent waits grow from here. DefaultScopedPollInterval = 100 * time.Millisecond // DefaultScopedPollBackoff is the factor each successive contended wait is // multiplied by, when WithScopedPollBackoff is not supplied. DefaultScopedPollBackoff = 2.0 // DefaultScopedMaxPollInterval caps the grown wait, when // WithScopedPollBackoff is not supplied. It bounds how long a waiter can // sit idle after the lock actually frees, which is the cost backoff trades // against reduced load on the underlying store. DefaultScopedMaxPollInterval = time.Second )
Variables ¶
var ( // ErrLockNotAcquired indicates Acquire could not obtain the lock immediately // because another caller currently holds it. Callers that want to wait should // compose Acquire with a retry/backoff loop themselves — the Locker interface // does not retry internally. ErrLockNotAcquired = platformerrors.New("lock not acquired") // ErrLockNotHeld indicates Release or Refresh was called on a lock the caller // no longer owns. Reasons include TTL expiration, the lock being stolen by // another caller after expiration, double-release, or — for the postgres // provider — the underlying connection having been closed out from under us. ErrLockNotHeld = platformerrors.New("lock not held") // ErrNilConfig indicates a nil provider config was passed to a constructor. ErrNilConfig = platformerrors.New("nil distributedlock config") // ErrInvalidTTL indicates a non-positive TTL was supplied to Acquire or Refresh. ErrInvalidTTL = platformerrors.New("invalid lock TTL") // ErrEmptyKey indicates an empty key was supplied to Acquire. ErrEmptyKey = platformerrors.New("empty lock key") // ErrNilDatabaseClient indicates a nil database.Client was passed to a postgres- // backed provider. ErrNilDatabaseClient = platformerrors.New("nil database client") )
Functions ¶
This section is empty.
Types ¶
type Lock ¶
type Lock interface {
// Key returns the lock name this handle owns.
Key() string
// TTL returns the configured expiration for this lock at the time it was
// last acquired or refreshed. It is not adjusted as time passes.
TTL() time.Duration
// Release releases the lock. Returns ErrLockNotHeld if the caller no longer
// owns the lock (expiration, theft after expiration, double-release).
Release(ctx context.Context) error
// Refresh extends the lock's TTL to the supplied value. Returns
// ErrLockNotHeld if the caller no longer owns the lock.
Refresh(ctx context.Context, ttl time.Duration) error
}
Lock is the handle returned from Acquire. It carries the ownership token internally and is the only way to release or refresh the lock. Lock handles are owned by a single goroutine — they must not be shared.
type Locker ¶
type Locker interface {
// Acquire attempts to acquire the lock named `key` with the supplied TTL.
// Returns ErrLockNotAcquired immediately if the lock is currently held by
// another caller. There is no internal retry — callers wrap with
// retry/backoff themselves.
Acquire(ctx context.Context, key string, ttl time.Duration) (Lock, error)
// Ping verifies the underlying backend is reachable.
Ping(ctx context.Context) error
// Close releases any backend resources held by the Locker. Outstanding Lock
// handles obtained from this Locker may become invalid after Close.
Close() error
}
Locker is the manager atom. It hands out Lock handles keyed by string. Locker implementations must be safe for concurrent use; the Lock handles they return are owned by the goroutine that called Acquire and are NOT goroutine-safe.
type ScopedLocker ¶
type ScopedLocker interface {
// WithLock blocks until the lock named key is acquired (or ctx is done),
// runs fn while holding it, and releases on return. fn's error is
// returned to the caller.
WithLock(ctx context.Context, key string, fn func(ctx context.Context) error) error
// TryWithLock never waits: if the lock is currently held elsewhere it
// returns (false, nil) without running fn. Otherwise it runs fn under the
// lock and returns (true, fn's error). An acquisition-infrastructure
// failure returns (false, err).
TryWithLock(ctx context.Context, key string, fn func(ctx context.Context) error) (bool, error)
}
ScopedLocker runs a function while holding a named lock, releasing the lock when the function returns — including on panic. It is the surface most lock consumers actually want (singleton chores, janitor election, migration serialization): there is no handle to carry, no TTL bookkeeping, and no way to forget Release.
Obtain one natively from a provider that supports scoped execution (the postgres provider's transaction-scoped implementation), or wrap any Locker with NewScopedLocker.
func NewScopedLocker ¶
func NewScopedLocker(locker Locker, opts ...ScopedOption) (ScopedLocker, error)
NewScopedLocker wraps any Locker in scoped execution. WithLock waits for a contended lock by polling Acquire (the Locker atom deliberately has no queueing of its own); providers with native waiting (postgres) ship their own ScopedLocker and don't need this adapter.
The scoped surface emits the same telemetry whichever Locker backs it — attach the observability deps via WithLogger, WithTracerProvider, and WithMetricsProvider: the wrapped Locker's own Acquire/Release instrumentation describes individual attempts, while scoped_lock_* describes the whole acquire-run-release operation, including fn's duration and the time spent waiting.
Example ¶
ExampleNewScopedLocker wraps a plain Locker in the scoped surface: acquire, run, release — including on panic. Nothing is left for the caller to carry or to forget.
package main
import (
"context"
"fmt"
"github.com/primandproper/platform-go/v9/distributedlock"
"github.com/primandproper/platform-go/v9/distributedlock/memory"
)
func main() {
ctx := context.Background()
locker, err := memory.NewLocker()
if err != nil {
panic(err)
}
defer func() { _ = locker.Close() }()
// nil logger, tracer provider, and metrics provider fall back to noop
// implementations; a real service passes the ones it built at startup.
scoped, err := distributedlock.NewScopedLocker(locker)
if err != nil {
panic(err)
}
if err = scoped.WithLock(ctx, "nightly-compaction", func(context.Context) error {
fmt.Println("compacting")
return nil
}); err != nil {
panic(err)
}
// The lock was released the moment fn returned, so the next caller gets it
// without waiting.
ran, err := scoped.TryWithLock(ctx, "nightly-compaction", func(context.Context) error {
fmt.Println("compacting again")
return nil
})
if err != nil {
panic(err)
}
fmt.Println("ran:", ran)
}
Output: compacting compacting again ran: true
type ScopedOption ¶
type ScopedOption func(*scopedLocker)
ScopedOption configures the generic scoped adapter returned by NewScopedLocker.
func WithLogger ¶
func WithLogger(logger logging.Logger) ScopedOption
WithLogger attaches a logger. An absent logger logs nowhere.
func WithMetricsProvider ¶
func WithMetricsProvider(metricsProvider metrics.Provider) ScopedOption
WithMetricsProvider attaches a metrics provider for the scoped_lock_* counters and histograms. An absent provider records nothing.
func WithScopedClock ¶
func WithScopedClock(c clock.Clock) ScopedOption
WithScopedClock swaps the clock used for contention polling. Tests generally do not need it: under testing/synctest the default clock already runs on bubble time, so WithLock's waiting is deterministic and instant.
func WithScopedJitter ¶
func WithScopedJitter(fn func() float64) ScopedOption
WithScopedJitter replaces the source of randomness that spreads contended waiters apart. fn must return a value in [0,1]; the adapter sleeps for half the current interval plus that fraction of the other half, so waiters that started together do not re-collide on every round.
The default draws from math/rand/v2 and needs no seeding. Tests wanting a fixed schedule can pass func() float64 { return 1 }, which yields exactly the un-jittered interval. A nil fn is ignored.
func WithScopedLockTTL ¶
func WithScopedLockTTL(ttl time.Duration) ScopedOption
WithScopedLockTTL sets the TTL the adapter passes to Acquire. The TTL must comfortably exceed fn's worst-case duration: if the underlying lock expires while fn is still running, mutual exclusion is no longer guaranteed, and the implicit release will surface ErrLockNotHeld in the returned error — and increment the scoped_lock_release_failures counter, which is the signal to alert on.
func WithScopedPollBackoff ¶
func WithScopedPollBackoff(factor float64, maxInterval time.Duration) ScopedOption
WithScopedPollBackoff sets how the contended wait grows: each successive wait is multiplied by factor and clamped to maxInterval.
Backoff exists because a fixed poll interval makes N waiters on one key a thundering herd against the underlying store — N/interval requests per second for as long as the holder runs, none of which can succeed. Growing the interval trades a bounded amount of post-release latency (at most maxInterval, and on average less because of the jitter) for a large drop in that load.
factor must be at least 1 and maxInterval at least the poll interval; NewScopedLocker rejects anything else. A factor of exactly 1 disables growth and restores a fixed interval, still jittered.
func WithScopedPollInterval ¶
func WithScopedPollInterval(interval time.Duration) ScopedOption
WithScopedPollInterval sets the first wait WithLock takes after a contended acquisition; later waits grow from it per WithScopedPollBackoff. It must be positive — a non-positive interval would turn the wait into a spin, since clock.Sleep returns immediately for a non-positive duration — and NewScopedLocker rejects it. TryWithLock never polls.
func WithTracerProvider ¶
func WithTracerProvider(tracerProvider tracing.TracerProvider) ScopedOption
WithTracerProvider attaches a tracer provider, enabling spans on every scoped-lock operation. An absent tracer provider traces nowhere.
Source Files
Directories
¶
| Path | Synopsis |
|---|---|
|
Package distributedlockmock provides moq-generated mock implementations of the distributedlock package's interfaces.
|
Package distributedlockmock provides moq-generated mock implementations of the distributedlock package's interfaces. |
|
Package postgres implements distributedlock.Locker against PostgreSQL session- scoped advisory locks (pg_try_advisory_lock).
|
Package postgres implements distributedlock.Locker against PostgreSQL session- scoped advisory locks (pg_try_advisory_lock). |