worklease

package module
v1.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Oct 28, 2026 License: Apache-2.0 Imports: 8 Imported by: 0

README

worklease

CI Go Reference Go Report Card License

worklease is a Go library for lease-based work coordination in distributed systems.

It is not a distributed lock library.

Distributed locks answer: who owns this resource right now? worklease answers a different question: when ownership changes, what does the new owner need to continue the work the previous owner started?


Installation

go get github.com/aetomala/worklease

Requires Go 1.26 or later; CI tests Go 1.26 and 1.27. The PostgreSQL backend is tested on PostgreSQL 16 with github.com/lib/pq (see Backends).

Quick start

This program runs the same work ID twice on the in-memory backend. The second run reads what the first one left behind. It needs no database.

package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"github.com/aetomala/worklease"
	"github.com/aetomala/worklease/backend/memory"
	"github.com/aetomala/worklease/worker"
)

func main() {
	lease, err := worklease.New(memory.New(), worklease.Config{
		TTL:      30 * time.Second,
		HolderID: "worker-1", // required: identifies this process and must not be empty
	})
	if err != nil {
		log.Fatal(err)
	}

	runner, err := worker.NewRunner(worker.RunnerConfig{
		Lease: lease,
		WorkFn: func(ctx context.Context, token worklease.Token, prior worklease.Checkpoint) ([]byte, error) {
			fmt.Printf("previous exit: %s, saved state: %q\n", prior.PrevExit, prior.State)
			return []byte("step 1 done"), nil // the checkpoint the next holder reads
		},
	})
	if err != nil {
		log.Fatal(err)
	}

	for i := 0; i < 2; i++ {
		if err := runner.Run(context.Background(), "billing-run-2026-10"); err != nil {
			log.Fatal(err)
		}
	}
}

It prints:

previous exit: none, saved state: ""
previous exit: finished, saved state: "step 1 done"

For production, replace memory.New() with postgres.New(db) (see Backends). Usage shows the same lifecycle step by step, and Higher-level packages covers leader and pool.


The Problem

Worker A acquires a lease, begins processing a multi-step job, makes partial progress, then crashes. Worker B acquires the lease — and starts from scratch.

Depending on the operation, this means duplicated writes, lost progress, or inconsistent state. Idempotency helps only if restarting from the beginning is cheap. For long-running work — provider onboarding, async lifecycle management, batch processing with intermediate state — it isn't.

The correct fix is checkpointed lease handoff: the outgoing owner writes progress state atomically with its last lease renewal. The incoming owner reads that state before starting work. Fencing tokens prevent zombie writes from expired owners corrupting the checkpoint after they've been superseded.

Every Go distributed locking library (distlock, pglock, dynamodb-lock-go, etcd leases, client-go/leaderelection) solves presence — mutual exclusion, TTL expiry, heartbeat renewal. None solve continuity. worklease fills that gap.

Is worklease the right fit?

worklease is designed for a specific shape of work: stateful, multi-step operations where restarting from scratch is expensive or incorrect, and where adopting a full workflow engine is not the right trade-off.

It is a good fit if you are running Go workers with PostgreSQL already in the stack, doing long-running jobs with meaningful intermediate state, and need correct handoff semantics without restructuring your application.

It is not the right fit if your jobs are small enough to restart cheaply — a job queue (River, Asynq, FOR UPDATE SKIP LOCKED) is simpler and sufficient. If your work is complex enough to need replay semantics and activity scheduling, Temporal is the right tool.


Prior Art

This pattern is not new. AWS's Kinesis Client Library has implemented it since 2013 via its LeaseRefresher. KCL maintains a DynamoDB table where each lease entry holds an explicit checkpoint column and a leaseCounter for fencing. A worker atomically renews its lease while writing its progress checkpoint. When a worker's lease expires, the successor reads the checkpoint column and resumes from exactly that point. Fencing is enforced via conditional writes on leaseCounter — structurally identical to worklease's fencing_token.

The pattern is proven and has been running production Kinesis workloads at AWS scale for over a decade.

What KCL doesn't provide:

  • A Go-native implementation (KCL Go is a thin wrapper over a Java MultiLangDaemon process — not idiomatic)
  • General-purpose work coordination (its checkpoint is a stream sequence number; the entire model assumes Kinesis shard processing)
  • Portability outside AWS (DynamoDB backend, CloudWatch metrics, IAM — AWS-only)

worklease extracts the same semantics into a general-purpose Go primitive — no AWS dependency, no Kinesis assumption, pluggable backends.


What worklease does not solve

These are intentional scope boundaries, not gaps.

External side effects are not fenced. ErrFenced stops stale checkpoint writes to the lease store. It does not stop a zombie worker from calling Stripe, sending an email, or writing to S3 before it discovers it has been superseded. Callers must make every external mutation idempotent, use an outbox pattern, or enforce idempotency keys at downstream systems. This is an inherent property of any coordination primitive — the library fences what it owns.

The window between checkpoints is at-least-once. If Worker A executes a step and crashes before checkpointing, Worker B will re-execute that step from the previous checkpoint. worklease provides a resumable progress marker, not exactly-once step execution.

Recovery logic is yours. The library delivers the checkpoint bytes and how the previous holder exited (PrevExit). What those bytes mean, how to validate partial state, and how to reconcile external effects that happened before a crash are application concerns.

Release expires the lease immediately. Release records the exit mode you declare and sets the lease expiration to the past, making the work item immediately available. A successor using WithWaitForLease acquires on its next poll; a fail-fast successor can call Acquire immediately after. The TTL governs crash detection only — it does not add latency to handoffs.


Concepts

Lease — A time-limited claim on a named unit of work. Expires if not renewed. When it expires, another worker can acquire it. In code, a lease is the worklease.Lease interface: worklease.New(backend, cfg) returns one, and you call Acquire, ReadCheckpoint, Checkpoint, Renew, Release, Forget, and StartRenewal on it.

Holder ID — Config.HolderID names this process. It must not be empty: worklease.New returns an error that matches ErrConfigInvalid. Use a value that is unique per process, such as a pod name. A lease is not re-entrant: acquiring a lease that your own HolderID still holds returns ErrLeaseHeld, so a restarted process with a stable ID waits for the old lease to expire (up to the TTL) unless the previous run released it.

Fencing token — A monotonically incrementing integer issued on every acquisition. A worker's writes to the lease store are rejected if a higher token has been issued — preventing zombie workers from corrupting the checkpoint after their lease expires. Tokens are monotonic but not contiguous, so never assume consecutive values: gaps are expected and have no operational significance. On PostgreSQL every Acquire attempt draws from the sequence, including attempts that return ErrLeaseHeld. The in-memory backend advances one counter per instance, only on a successful Acquire, so the tokens of one work ID still skip whenever another work ID acquires in between.

Checkpoint — Progress state written atomically with lease renewal. If the worker is making progress, it proves liveness and saves state in one operation. The last checkpoint survives to the next owner. Checkpoint and Renew are distinct: Checkpoint writes state and extends the TTL atomically; Renew extends the TTL without updating state.

Handoff vs crash recovery — Two distinct acquisition paths. If the previous owner called Release explicitly, the checkpoint contains final state and the successor knows the work was cleanly surrendered. If the lease expired without a release, the checkpoint contains the last known progress and the successor resumes from partial state. These are different situations and callers handle them differently. The library makes the distinction explicit.


Usage

Acquiring a lease and checkpointing progress
b := memory.New() // in production: postgres.New(db), see Backends

lease, err := worklease.New(b, worklease.Config{
    TTL:      30 * time.Second,
    HolderID: os.Getenv("WORKER_ID"), // required and unique per process; New rejects an empty value
})
if err != nil {
    return err
}

token, err := lease.Acquire(ctx, "billing-run-2026-10")
if err != nil {
    return err // worklease.ErrLeaseHeld: another holder has it
}

prior, err := lease.ReadCheckpoint(ctx, token)
if err != nil {
    return err
}

switch prior.PrevExit {
case worklease.ExitNone:
    // First run: start from scratch.
case worklease.ExitFinished:
    // The previous run completed; prior.State is its final state.
case worklease.ExitRetired:
    // This work is done for good. Hand the lease back unchanged.
    return lease.Release(ctx, token, worklease.ExitRetired)
default:
    // ExitAbandoned, ExitExpired, or a mode this code does not know:
    // prior.State is partial. Validate it before resuming.
}

renewCtx, stopRenewal := lease.StartRenewal(ctx, token)
defer stopRenewal()

state, workErr := run(renewCtx, token, prior.State)
stopRenewal()

if errors.Is(workErr, worklease.ErrFenced) || errors.Is(context.Cause(renewCtx), worklease.ErrFenced) {
    return worklease.ErrFenced // a successor holds the lease or its record was removed; record nothing
}

if errors.Is(context.Cause(renewCtx), worklease.ErrLeaseWindowExhausted) {
    // The lease lapsed before a renewal succeeded. Do not Checkpoint or Release it:
    // the successor sees ExitExpired and the last good checkpoint.
    return errors.Join(workErr, worklease.ErrLeaseWindowExhausted)
}

// Final writes must survive cancellation of ctx without hanging shutdown.
cleanupCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 5*time.Second)
defer cancel()

mode := worklease.ExitFinished // use worklease.ExitRetired for one-shot work
if workErr != nil {
    mode = worklease.ExitAbandoned
}
if state != nil {
    if err := lease.Checkpoint(cleanupCtx, token, state); err != nil {
        return errors.Join(workErr, err)
    }
}
if err := lease.Release(cleanupCtx, token, mode); err != nil {
    return errors.Join(workErr, err) // ErrLeaseExpired: the successor sees ExitExpired
}
return workErr

Rows released with ExitFinished are kept as resume points for work that runs again, and Vacuum.Sweep never deletes them. Release one-shot work with ExitRetired; otherwise its row is never eligible for cleanup. worker.Runner, leader.Elect, and pool do all of the above for you.

Renewing without checkpointing

If the worker is healthy but not at a checkpointable boundary yet:

if err := lease.Renew(ctx, token); err != nil {
    // ErrFenced: a higher token has been issued or the record is gone — stop working.
    return err
}

Checkpoint and Renew are separate because they mean different things. Renew says "I'm alive." Checkpoint says "I'm alive and here's where I am."

Fresh acquisition

ReadCheckpoint returns a nil state (not an error) when this is the first worker to acquire this lease, and also after a Checkpoint with a nil or empty state, on every backend:

prior, err := lease.ReadCheckpoint(ctx, token)
if err != nil {
    return err
}
if prior.State == nil {
    // First acquisition — start from the beginning.
    progress = OnboardingProgress{}
}
Managing renewal directly

StartRenewal starts a managed renewal goroutine and returns a derived context and a stop function. worker.Runner and leader.Elect handle this lifecycle automatically; use StartRenewal directly only when you need fine-grained control.

Two requirements apply when calling StartRenewal directly:

  1. Call stopRenewal() before Release — the goroutine must exit before ownership is surrendered. Register defer stopRenewal() immediately as a panic-safety net, then call it explicitly before Release (the defer becomes a no-op on the clean path).
  2. Pass the original ctx to Release, not renewCtx — renewCtx may already be cancelled (by fencing or window exhaustion) when Release is called; using it causes Release to fail with a context error.
renewCtx, stopRenewal := lease.StartRenewal(ctx, token)
defer stopRenewal() // panic-safety net

if err := doWork(renewCtx, ...); err != nil {
    return err // stopRenewal fires via defer; do not Release if fenced
}

stopRenewal()             // explicit stop before Release
lease.Release(ctx, token, worklease.ExitFinished) // use the original ctx, not renewCtx

stopRenewal returns without waiting for an in-flight Renew call, and the renewal context is cancelled when the lease window closes, even if a Renew call is stalled. Work code can read the window with worklease.LeaseWindow:

if deadline, ok := worklease.LeaseWindow(renewCtx); ok && time.Until(deadline) < 5*time.Second {
    return errNotEnoughLease // too little guaranteed lease left for this step
}

LeaseWindow works on renewCtx and on any context derived from it, advances with each successful renewal, and, under a constant clock offset, never exceeds the true expiry. It returns ok == false outside a renewal context and once renewCtx is done. Token.ExpiresAt() is the expiry recorded at acquisition and never moves.


Error Reference

Error Returned by Meaning
ErrFenced Checkpoint, Renew, Release, ReadCheckpoint, Forget The token no longer matches the stored lease, or no record exists for the work ID (also for a zero Token). This worker has been superseded. Stop writing to the lease store.
ErrLeaseHeld Acquire The lease is currently held, by another worker or by your own HolderID.
ErrLeaseExpired Renew, Release, Checkpoint Renew and Release: the lease expired before the call. Release also returns it for a second Release with the same token, and Checkpoint returns it once this holder has declared an exit with Release.
ErrInvalidExitMode Release The exit mode is not ExitFinished, ExitAbandoned, or ExitRetired.
ErrLeaseWindowExhausted — No Lease method returns it: it is the cause of renewCtx when the lease window closes without a successful renewal. Read it with context.Cause(renewCtx). worker.Runner.Run and leader.Elect return it wrapped when the work function returned nil.
ErrRetire a work function Return it, or wrap it, to release with ExitRetired (see Retiring work).
ErrConfigInvalid — The parent of the configuration errors. Match it with errors.Is; see Row lifecycle for the members.
ErrNilBackend New, NewVacuum, Vacuum.Sweep The backend is nil, or the Vacuum was not built by NewVacuum. Matches ErrConfigInvalid.
ErrRetentionRequired Vacuum.Sweep SweepOptions.Retention is zero or negative. Matches ErrConfigInvalid.

The four backend sentinels (ErrFenced, ErrLeaseHeld, ErrLeaseExpired, ErrInvalidExitMode) are defined in package backend and re-exported from worklease as the same values, so errors.Is matches either name. Acquire with an empty work ID returns a plain error that matches no sentinel.

Match sentinels with errors.Is. The text of an error message is not part of the compatibility promise and may change in a minor release; such changes are listed under ### Changed in CHANGELOG.md.

ErrFenced is the critical one. When Checkpoint, Renew, Release, ReadCheckpoint, or Forget returns ErrFenced, a successor has already acquired the lease or its record was removed, and this worker is a zombie. Stop immediately — any further writes to the lease store will be rejected. Note that ErrFenced does not cancel in-flight external calls the worker may have already initiated; those must be made idempotent at the application layer.


Upgrading

UPGRADING.md has the steps for each version, including the schema migrations that v0.5.0 and v0.6.0 required. From v0.8.0 to v1.0.0 there is no schema change, and the module requires Go 1.26.


Higher-level packages

worker — Lifecycle management

worker.Runner handles the acquire / ReadCheckpoint / StartRenewal / WorkFn / Release lifecycle so callers implement only the work function:

import "github.com/aetomala/worklease/worker"

r, err := worker.NewRunner(worker.RunnerConfig{
    Lease:  lease,
    WorkFn: func(ctx context.Context, token worklease.Token, prior worklease.Checkpoint) ([]byte, error) {
        // prior.State is the last checkpoint; prior.PrevExit says how the previous holder exited (ExitNone on first acquisition).
        // Return updated checkpoint bytes, or nil to leave the checkpoint unchanged.
        return processWork(ctx, prior)
    },
})
if err != nil { ... }
if err := r.Run(ctx, "onboarding:tenant-abc"); err != nil { ... }

Runner picks the exit mode from your work function's result: nil releases with ExitFinished, an error wrapping worklease.ErrRetire releases with ExitRetired and Run returns nil, and any other error releases with ExitAbandoned. It does not release when fenced or when the lease window ran out; the next holder sees ExitExpired. The final Checkpoint and Release run on a context that survives cancellation of yours, bounded by RunnerConfig.CleanupTimeout (default 5s). leader.Elect applies the same mapping to fn, and calls OnRelinquished only after ExitFinished or ExitRetired.

leader — Simplified leadership

leader.Elect acquires a work ID, starts managed lease renewal, calls fn under the renewal context, stops renewal, and releases. Suitable for single-leader patterns where one process should be active at a time:

import "github.com/aetomala/worklease/leader"

err := leader.Elect(ctx, lease, "scheduler:primary", leader.Config{}, func(ctx context.Context) error {
    // ctx is cancelled if the lease is fenced or renewal fails.
    // Return when the work is done; leader.Elect will release.
    return runScheduler(ctx)
})

Without worklease.WithWaitForLease(), Elect returns ErrLeaseHeld when another process leads, after waiting leader.Config.IdleInterval (default 1s, up to 20% jitter), so a follower can call Elect in a loop without spinning against the backend. Any other Acquire error, such as a database outage, is returned after BackoffInterval, and a context error at once. BackoffInterval also applies after every non-fencing return that follows a successful Acquire, including a clean finish. Its zero value means no sleep, so a loop that retries on every error must set it: with the database unreachable, a loop of Elect calls with leader.Config{} spins (tens of thousands of Acquire calls per second on a development machine), and with BackoffInterval: time.Second it makes one call per second. Pass worklease.WithWaitForLease() in leader.Config.AcquireOptions to block until leadership is available instead.

leader.Config also accepts optional lifecycle callbacks — OnElected, OnLost, and OnRelinquished — for observing leadership transitions; none of them fires when Acquire fails:

cfg := leader.Config{
    BackoffInterval: time.Second, // pace retries after errors other than ErrLeaseHeld
    OnElected: func(ctx context.Context, token worklease.Token) {
        log.Printf("elected: %v", token)
    },
    OnLost: func(ctx context.Context, token worklease.Token) {
        log.Printf("lost the lease: %v", token) // fenced, or the lease window ran out
    },
    OnRelinquished: func(ctx context.Context, token worklease.Token) {
        log.Printf("released: %v", token) // after ExitFinished or ExitRetired
    },
}
pool — Distributed work distribution

pool.Pool distributes a fixed set of work IDs across competing processes. Multiple Pool instances — one per process — collectively cover the full work ID set. Rebalancing is emergent from lease acquisition races:

import "github.com/aetomala/worklease/pool"

p, err := pool.New(lease, pool.Config{
    WorkIDs: []string{"shard-0", "shard-1", "shard-2", "shard-3"},
}, func(ctx context.Context, workID string, token worklease.Token, prior worklease.Checkpoint) ([]byte, error) {
    return processShard(ctx, workID, prior)
})
if err != nil { ... }

// Blocks until ctx is cancelled; all active slots complete before Run returns.
if err := p.Run(ctx); err != nil { ... }

Return worklease.ErrRetire (or wrap it) from the work function to retire the work ID: the slot releases with ExitRetired, calls OnSlotRetired, and stops. Return a PermanentError to drop a slot without reacquisition — implement the interface on a custom type, or wrap an error with pool.Permanent(err); the slot releases with ExitAbandoned. When every slot exits through a PermanentError, Run returns pool.ErrAllSlotsDead; if any slot retired, it returns nil. ErrAllSlotsDead requires every slot to be counted dead: a slot that observes cancellation of ctx before its PermanentError check exits uncounted, so if ctx is cancelled at the same moment the last slot fails, Run may return nil instead. A slot whose work ID another process holds waits Config.IdleInterval (default 1s, up to 20% jitter); a slot whose work function returned nil waits Config.RerunInterval (default 1s, up to 20% jitter) before running again; a slot whose work function failed waits Config.BackoffInterval (default 1s) and calls OnSlotBackoff. Set pool.Config.Observer to a pool.Observer to receive slot lifecycle events (OnSlotAcquired / OnSlotLost / OnSlotBackoff / OnSlotDead / OnSlotRetired), and call ActiveSlots() for a point-in-time view of slots currently executing their work function. Construction errors are distinct sentinels (ErrLeaseRequired / ErrEmptyWorkIDs / ErrEmptyWorkID / ErrDuplicateWorkID / ErrWithWaitForLeaseProhibited / ErrWorkFnRequired) that all satisfy errors.Is(err, pool.ErrConfigInvalid) and errors.Is(err, worklease.ErrConfigInvalid); every work ID must be non-empty and unique.

checkpoint — Typed serialization helpers

checkpoint.Codec and the generic Encode[T] / Decode[T] helpers add typed serialization on top of the raw []byte checkpoint layer. checkpoint.JSON() returns a JSONCodec backed by encoding/json. The codec must not be nil. Decode returns the zero value for nil data without calling the codec, and a backend returns a nil State when there is no checkpoint, so a first acquisition decodes to the zero value.

Observability — LeaseObserver

Set worklease.Config.Observer to a LeaseObserver to receive a synchronous callback after every lease operation — OnAcquire, OnCheckpoint, OnRenew, OnRelease, OnReadCheckpoint, and OnForget — each with an event struct carrying the Token, the error, and a Duration for the backend call. OnFenced fires in addition when Checkpoint, Renew, Release, or Forget is fenced. nil installs a no-op, so observability is fully opt-in. Vacuum is observed separately: set worklease.VacuumConfig.Observer to a VacuumObserver to receive OnSweep with each sweep's options, rows deleted, duration, and error. See examples/observability for a stdlib-only implementation of both.

Embedding a no-op observer

worklease.NoopObserver, worklease.NoopVacuumObserver, and pool.NoopObserver discard every event. Embed one in your own observer to implement only the methods you need; your type keeps compiling when a later release adds an observer method. The same pattern applies to all three: embed the no-op type that matches the interface you implement.

type acquireCounter struct {
	worklease.NoopObserver
	n atomic.Int64
}

func (c *acquireCounter) OnAcquire(_ context.Context, _ worklease.AcquireEvent) {
	c.n.Add(1)
}

Pass &acquireCounter{} as worklease.Config.Observer. A runnable version is ExampleNoopObserver in the package documentation. FencedEvent.Operation prints as Checkpoint, Renew, Release, or Forget when formatted with %v.


Exit modes and handoff

Every holder declares how it leaves; the next holder reads it as Checkpoint.PrevExit.

PrevExit How it is set What the next holder should do
ExitNone Inferred: the work ID was never acquired, or its row was removed by Forget or Sweep Start fresh
ExitFinished Release(ctx, token, worklease.ExitFinished) Continue from the final state
ExitAbandoned Release(ctx, token, worklease.ExitAbandoned) — an error, a cancellation, a shutdown Validate the partial state, then resume
ExitExpired Inferred: the lease expired with no recorded exit — a crash, a partition, or a renewal window that ran out Validate the partial state; external effects may have happened after the last checkpoint
ExitRetired Release(ctx, token, worklease.ExitRetired) Do not redo the work

Release accepts only ExitFinished, ExitAbandoned, and ExitRetired; anything else returns worklease.ErrInvalidExitMode. Once a lease has expired, Release returns worklease.ErrLeaseExpired and records nothing, so a holder that lost its lease cannot vouch for its work. Treat any mode your code does not recognize as ExitExpired; later releases may add modes. "Crash recovery" in these docs means handling ExitExpired.

prior, err := lease.ReadCheckpoint(ctx, token)
if err != nil {
    return err
}
log.Printf("previous holder %q exited %s", prior.PrevHolderID, prior.PrevExit)
Retiring work

Return worklease.ErrRetire from a worker, leader, or pool work function when its work ID is complete for good. The lease is released with ExitRetired, Run and Elect report success, and a pool slot stops. A later holder of the same work ID sees ExitRetired.

WorkFn: func(ctx context.Context, token worklease.Token, prior worklease.Checkpoint) ([]byte, error) {
    if prior.PrevExit == worklease.ExitRetired {
        return prior.State, worklease.ErrRetire // already done: keep it retired
    }
    state, err := migrateTenant(ctx, prior.State)
    if err != nil {
        return state, err // ExitAbandoned: the next holder resumes from state
    }
    return state, worklease.ErrRetire // ExitRetired: Run returns nil
},
Row lifecycle: Forget and Vacuum

Lease rows outlive their holders so that successors can read them. Two calls remove them:

// Delete one row now. Forget matches the holder and the token, so it also works
// after your own Release.
if err := lease.Forget(ctx, token); err != nil {
    return err // ErrFenced: the lease moved on, or the row is already gone
}

// Periodically delete retired rows last updated more than a week ago.
vacuum, err := worklease.NewVacuum(b, worklease.VacuumConfig{})
if err != nil {
    return err // ErrNilBackend
}
n, err := vacuum.Sweep(ctx, worklease.SweepOptions{
    Retention: 7 * 24 * time.Hour,
})
if err != nil {
    return err
}
log.Printf("deleted %d retired rows", n)

Sweep deletes rows released with ExitRetired once they are older than Retention. With IncludeExpired: true it also deletes rows whose lease expired with no declared exit, which discards the partial state a successor would have recovered from. Rows released with ExitFinished or ExitAbandoned, and rows currently held, are never deleted. Retention is how long a retired or expired row is kept after its last update; it is not tied to the TTL. After Forget or Sweep, the next holder sees ExitNone, and a holder whose row was removed gets ErrFenced. Do not run Sweep while v0.5 processes still share the table (see UPGRADING.md).

Forget is reported to LeaseObserver.OnForget, and each Sweep that reaches the backend to VacuumConfig.Observer.

Configuration errors are matchable: every error worklease.New returns, ErrNilBackend from NewVacuum and Vacuum.Sweep, and ErrRetentionRequired from Vacuum.Sweep satisfy errors.Is(err, worklease.ErrConfigInvalid). So do the pool configuration sentinels (including pool.ErrConfigInvalid), worker.ErrLeaseRequired, worker.ErrWorkFnRequired, leader.ErrLeaseRequired, and leader.ErrFnRequired.

Backends

PostgreSQL
import (
    "database/sql"

    _ "github.com/lib/pq" // any database/sql PostgreSQL driver; the backend does not register one

    "github.com/aetomala/worklease/backend/postgres"
)

db, _ := sql.Open("postgres", dsn)
backend, _ := postgres.New(db)

lease, _ := worklease.New(backend, worklease.Config{
    TTL:      30 * time.Second,
    HolderID: os.Getenv("WORKER_ID"),
})

Run the migration before first use (canonical source: backend/postgres/schema.sql). Without it, the first Acquire fails with an error that wraps the driver's relation "worklease_leases" does not exist and matches no worklease sentinel; the statements are idempotent, so run them and retry:

-- canonical: backend/postgres/schema.sql
CREATE SEQUENCE IF NOT EXISTS worklease_fencing_seq;

CREATE TABLE IF NOT EXISTS worklease_leases (
    work_id         TEXT        PRIMARY KEY,
    holder_id       TEXT        NOT NULL,
    fencing_token   BIGINT      NOT NULL DEFAULT nextval('worklease_fencing_seq'),
    expires_at      TIMESTAMPTZ NOT NULL,
    checkpoint      BYTEA,
    exit_mode       TEXT,
    prev_exit_mode  TEXT        NOT NULL DEFAULT 'none',
    prev_holder_id  TEXT,
    acquired_at     TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    CONSTRAINT worklease_leases_exit_mode_check
        CHECK (exit_mode IS NULL OR exit_mode IN ('finished', 'abandoned', 'retired')),
    CONSTRAINT worklease_leases_prev_exit_mode_check
        CHECK (prev_exit_mode IN ('none', 'expired', 'finished', 'abandoned', 'retired'))
);

CREATE INDEX IF NOT EXISTS idx_worklease_leases_updated_at
    ON worklease_leases (updated_at);

The atomic checkpoint operation is a single UPDATE ... WHERE fencing_token = $n. Zero rows updated means the token is stale (ErrFenced) or that this holder already declared an exit with Release (ErrLeaseExpired); a follow-up query tells them apart. There is no Lua script, no optimistic retry loop, no distributed clock dependency. The checkpoint column is NULL until state is written, and again after a Checkpoint of a nil or empty state; ReadCheckpoint returns a nil State for it.

The backend requires the READ COMMITTED isolation level, the PostgreSQL default. Under REPEATABLE READ or SERIALIZABLE, a contended Acquire fails with a serialization error (SQLSTATE 40001) instead of returning ErrLeaseHeld. If your database or role defaults to another level, pin it for the role worklease connects as: ALTER ROLE worklease_role SET default_transaction_isolation = 'read committed';. The schema drops the deprecated clean_handoff column as of v0.7; existing databases drop it with the step in UPGRADING.md. A statement cancelled or timed out by its context returns an error that wraps both the driver error and ctx.Err() (#113); the Backend contract requires custom backends to match ctx.Err() the same way.

The backend is tested on PostgreSQL 16 with github.com/lib/pq. The backend suite also passes on pgx v5 (github.com/jackc/pgx/v5/stdlib), apart from two specs that look for lib/pq's *pq.Error type.

In-memory (testing)
import "github.com/aetomala/worklease/backend/memory"

backend := memory.New()
lease, _ := worklease.New(backend, worklease.Config{
    TTL:      5 * time.Second,
    HolderID: "test-worker",
})

Suitable for unit tests within a single process. It passes the same conformance suite as the PostgreSQL backend. Its fencing tokens come from a per-instance counter that advances only on a successful Acquire, while PostgreSQL uses one sequence per database that every attempt advances. Not safe for concurrent use across processes.


Comparison

Tool Layer Solves Doesn't solve
distlock / pglock Library Mutual exclusion with TTL No checkpoint, no handoff semantics
dynamodb-lock-go Library DynamoDB-backed locking No progress state, no write fencing
etcd leases Infrastructure Kubernetes-native presence No work coordination layer
client-go/leaderelection Library Single leader election Not per-work-item ownership
KCL LeaseRefresher Library Checkpointed lease handoff at AWS scale Java only, Kinesis-specific, AWS-only
Temporal Platform Durable workflow execution with replay Requires full application rewrite as Workflows + Activities; run a server
Inngest Platform Managed durable step execution Requires platform adoption — not a drop-in library
DBOS Framework DB-backed durable execution Requires full application restructure; limited Go support
Watermill Library Go message routing Checkpointing is broker offset tracking, not general work state
River / Asynq / job queues Library Work assignment with retry No stateful handoff — successor restarts from scratch
worklease Library Work handoff with continuity, idiomatic Go, pluggable backends Not a general-purpose lock; does not fence external side effects
Why not Temporal, Inngest, or DBOS?

These tools sit at the high-guarantee, high-adoption-cost end of the spectrum. They solve the failure mode — but the adoption unit is your entire application architecture.

Temporal requires rewriting business logic as Workflows and Activities, running a Temporal server, and accepting deterministic replay constraints. Inngest is a fully managed platform with its own execution model. DBOS restructures your entire application around a framework.

Teams already running Go workers with PostgreSQL don't need to restructure their application to get correct lease handoff semantics. They need a library primitive they can drop into the system they already have.

Distributed locks sit at the opposite end — low adoption cost, wrong abstraction. They tell you who holds the resource; they don't help the new holder continue what the old holder started.

The gap between these two ends is where worklease lives. The only prior art in that gap is KCL, which is Java-only and Kinesis-specific.


Stability

From v1.0.0 worklease follows semantic versioning, and ADR-0020 records what that covers:

  • A change to an exported identifier outside internal/ is breaking, with exceptions for implementers: Lease and the observer interfaces may gain methods in a minor release (embed worklease.NoopObserver, worklease.NoopVacuumObserver, or pool.NoopObserver, or embed the Lease interface in a wrapper). The backend.Backend method set is fixed for all of v1.
  • Match sentinels with errors.Is. Error message text is not promised.
  • A default marked tunable in ADR-0020 may change in a minor release when the CHANGELOG lists it; set a value explicitly if you depend on it.
  • A minor release may add tables, nullable columns, and indexes to the PostgreSQL schema, never drop or rename a column. Applying and migrating the schema stays your job.
  • The supported Go versions are the two latest releases, and the go directive is raised only in a minor release.
  • An identifier is deprecated for at least one minor release before a major release removes it.
  • Nothing under internal/ is promised, and backend/conformance tightens the backend contract only in a major release.

Status

CHANGELOG.md lists every release, and UPGRADING.md has the steps between versions. The core lease model (Lease, Token, fencing, options, sentinels) and the worker, leader, pool, and checkpoint packages are covered by the promise above.

Roadmap

A Redis backend and an etcd backend are possible additions (unscheduled).


Examples

Runnable examples covering crash recovery, checkpoint resume, cluster leadership, partition processing, observability, and the renewal and acquire lifecycle. No infrastructure required — all examples run against the in-memory backend.

See examples/ for the full list with descriptions and run instructions.


License

Apache 2.0 — see LICENSE and NOTICE.

Documentation

Overview

Package worklease provides lease-based work coordination for distributed systems.

worklease solves a different problem than distributed locking. Distributed locks answer: who owns this resource right now? worklease answers: when ownership changes, what does the new owner need to continue the work the previous owner started?

The library is built around three primitives:

  • A Lease — a time-limited, named claim on a unit of work.
  • A fencing token — a monotonically incrementing integer that prevents zombie writes from a worker that was slow, not dead.
  • A checkpoint — progress state written atomically with lease renewal, so the last known state survives to the next owner.

Entry point

Construct a Lease with New, backed by a github.com/aetomala/worklease/backend.Backend:

backend, err := postgres.New(db)
lease, err := worklease.New(backend, worklease.Config{
    TTL:      30 * time.Second,
    HolderID: workerID,
})

See the Lease interface for the full API.

What worklease is not

worklease is not a distributed lock library, a workflow engine, or a general-purpose job queue. For mutual exclusion alone, use distlock or pglock. For durable workflow execution, use Temporal. For Kubernetes-native leader election, use client-go/leaderelection. The leader subpackage provides simplified single-work-item leadership, and the pool subpackage distributes a fixed set of work IDs across competing processes — both built on worklease's own Lease primitives.

Backends

Two backends are included:

Index

Examples

Constants

View Source
const (
	// ExitNone means there was no previous holder: the work ID was never
	// acquired, or its row was removed by Forget or Sweep. Inferred; Release
	// never accepts it.
	ExitNone = backend.ExitNone

	// ExitFinished means the run completed and the checkpoint is final state
	// for that run. The work ID may be acquired again. Sweep never deletes a
	// row whose last exit is ExitFinished, so release one-shot work with
	// ExitRetired instead.
	ExitFinished = backend.ExitFinished

	// ExitAbandoned means the holder stopped deliberately without completing,
	// on error, cancellation, or shutdown. The checkpoint is partial state.
	ExitAbandoned = backend.ExitAbandoned

	// ExitRetired means the work ID is complete permanently. A successor
	// should not redo the work. The row is eligible for Sweep.
	ExitRetired = backend.ExitRetired

	// ExitExpired means the lease expired with no recorded exit: a crash, a
	// partition, or a renewal window that ran out. The checkpoint is partial
	// state, and external effects may have happened after it. Inferred;
	// Release never accepts it.
	ExitExpired = backend.ExitExpired
)

Exit modes. Callers must treat any value they do not recognize as ExitExpired, because modes may be added in later releases.

Variables

View Source
var (
	// ErrFenced is returned by Checkpoint, Renew, Release, ReadCheckpoint, and
	// Forget when the token no longer matches the stored lease or when no record
	// exists for the work ID. It is the same value as backend.ErrFenced.
	ErrFenced = backend.ErrFenced

	// ErrLeaseHeld is returned by Acquire when the lease for the work ID is held
	// and has not expired. It is the same value as backend.ErrLeaseHeld.
	ErrLeaseHeld = backend.ErrLeaseHeld

	// ErrLeaseExpired is returned by Renew when the lease has already expired, by
	// Release when the lease has expired or this holder already declared an exit,
	// and by Checkpoint only once this holder has declared an exit with Release.
	// It is the same value as backend.ErrLeaseExpired.
	ErrLeaseExpired = backend.ErrLeaseExpired

	// ErrLeaseWindowExhausted is set as the cancel cause of the renewal context
	// when the renewal goroutine exhausts the remaining lease window without a
	// successful renewal (goroutine lifecycle path 3).
	// Inspect via context.Cause(renewCtx) — not returned directly from any method.
	ErrLeaseWindowExhausted = errors.New(msgLeaseWindowExhausted)

	// ErrConfigInvalid is the parent sentinel for configuration errors: every
	// error worklease.New returns, ErrNilBackend wherever it is returned,
	// ErrRetentionRequired, and the configuration sentinels of the pool, worker,
	// and leader packages. Errors from other packages, such as the nil database
	// check of postgres.New, are not members. Match it with errors.Is.
	ErrConfigInvalid = errors.New("worklease: invalid configuration")

	// ErrNilBackend is returned by New and NewVacuum when the backend is nil,
	// and by Vacuum.Sweep on a Vacuum that NewVacuum did not build. It matches
	// ErrConfigInvalid.
	ErrNilBackend error = &configError{msg: msgNilBackend}
)

Sentinel errors for Lease operations. ErrFenced, ErrLeaseHeld, and ErrLeaseExpired are the identical values of the sentinels of the same name in package backend, so errors.Is matches either name.

View Source
var (
	// ErrInvalidExitMode is returned by Release when mode is not ExitFinished,
	// ExitAbandoned, or ExitRetired. It is the same value as
	// backend.ErrInvalidExitMode.
	ErrInvalidExitMode = backend.ErrInvalidExitMode

	// ErrRetire is returned, or wrapped, by a work function to declare that its
	// work ID is complete permanently. Runner.Run in package worker, Elect in
	// package leader, and Pool in package pool release the lease with
	// ExitRetired and report success. Pool also stops the slot.
	ErrRetire = errors.New(msgRetire)
)
View Source
var ErrRetentionRequired error = &configError{msg: msgRetentionRequired}

ErrRetentionRequired is returned by Vacuum.Sweep when opts.Retention <= 0. It matches ErrConfigInvalid.

Functions

func HasWaitForLease added in v0.3.0

func HasWaitForLease(opts []AcquireOption) bool

HasWaitForLease reports whether opts includes WithWaitForLease. Pool.New uses this to enforce that WithWaitForLease is not passed in Config.AcquireOptions at construction time.

func LeaseWindow added in v0.7.0

func LeaseWindow(ctx context.Context) (deadline time.Time, ok bool)

LeaseWindow returns the end of the lease window tracked by the renewal goroutine whose renewal context is ctx or the nearest ancestor of ctx. The window starts as the local start of the acquiring backend call plus TTL, and advances to the local start of each successful renewal plus TTL. It is measured on the local monotonic clock, so under a constant offset between the local and backend clocks it never exceeds the true expiry. The ok result is false when ctx does not descend from a renewal context, and once that renewal context is done: after fencing, an exhausted window, or cancellation of its parent. After stopRenewal the window stays valid and stops advancing.

Types

type AcquireEvent added in v0.4.0

type AcquireEvent struct {
	WorkID   string
	Token    Token
	Duration time.Duration
	Err      error
}

AcquireEvent carries the result of one backend acquire attempt. Duration is the duration of that backend call only — not the wait loop. Token is the zero value of Token if Err is non-nil.

type AcquireOption

type AcquireOption func(*acquireConfig)

AcquireOption is a functional option for Acquire.

func WithPollInterval

func WithPollInterval(d time.Duration) AcquireOption

WithPollInterval sets the interval at which Acquire polls when WithWaitForLease is active. If d is zero or negative, the default (2 * time.Second) is preserved.

func WithWaitForLease

func WithWaitForLease() AcquireOption

WithWaitForLease configures Acquire to block and retry until a lease becomes available, rather than returning ErrLeaseHeld immediately.

type Checkpoint added in v0.6.0

type Checkpoint = backend.Checkpoint

Checkpoint is the last checkpointed state plus how the immediately previous holder exited. It is an alias for backend.Checkpoint.

type CheckpointEvent added in v0.4.0

type CheckpointEvent struct {
	Token    Token
	Size     int
	Duration time.Duration
	Err      error
}

CheckpointEvent carries the result of a Checkpoint call. Size is len(state) from the call site.

type Config

type Config struct {
	// TTL is the time-to-live for acquired leases. Required; New returns an
	// error for zero, for a negative value, and for any value below 1ms.
	TTL time.Duration

	// HolderID is the identifier of the entity that will hold leases. Required; empty returns an error.
	HolderID string

	// Observer is the LeaseObserver that will be notified of lease events. Optional; nil defaults to NoopObserver.
	Observer LeaseObserver
}

Config holds configuration for a Lease instance.

type ExitMode added in v0.6.0

type ExitMode = backend.ExitMode

ExitMode records how a lease holder left a lease. It is an alias for backend.ExitMode; the canonical definition lives in package backend because package backend cannot import package worklease.

type FencedEvent added in v0.4.0

type FencedEvent struct {
	Token     Token
	Operation Operation
}

FencedEvent carries the context of a fencing event. Called in addition to the operation-specific event — not instead of.

type ForgetEvent added in v0.7.0

type ForgetEvent struct {
	Token    Token
	Duration time.Duration
	Err      error
}

ForgetEvent carries the result of a Forget call. Duration is the duration of the backend call.

type Lease

type Lease interface {
	// Acquire attempts to acquire a lease for the given workID. Returns ErrLeaseHeld
	// if the lease for this workID is held and has not expired. If WithWaitForLease
	// is set, blocks until the lease is available, polling at the configured interval.
	// An empty workID returns an error that matches no sentinel, including
	// ErrConfigInvalid.
	Acquire(ctx context.Context, workID string, opts ...AcquireOption) (Token, error)

	// Checkpoint persists state associated with the current lease. The caller must
	// pass a valid Token obtained from Acquire. Returns ErrFenced if the
	// token's holder ID or fencing token no longer matches the stored lease, including when the
	// lease lapsed and another holder has since acquired it.
	// Checkpoint returns ErrLeaseExpired, without writing, only once this holder has
	// declared an exit with Release; a declared exit cannot be undone by a late
	// write. Stop any background checkpoint loop when it sees ErrLeaseExpired.
	// A lease whose TTL lapsed with no declared exit and no successor is not an
	// error: the write succeeds and revives the lease.
	// A nil or empty state clears the stored state: ReadCheckpoint then returns a
	// nil State on every backend.
	Checkpoint(ctx context.Context, token Token, state []byte) error

	// Renew extends the lease expiration time. Returns ErrFenced if the token's
	// holder ID or fencing token no longer matches the stored lease, or ErrLeaseExpired if the
	// lease has already expired.
	Renew(ctx context.Context, token Token) error

	// Release records mode as this holder's exit and expires the lease
	// immediately, so a successor can acquire without waiting for the TTL. The
	// successor reads mode as Checkpoint.PrevExit. Mode must be ExitFinished,
	// ExitAbandoned, or ExitRetired; any other value returns ErrInvalidExitMode
	// without calling the backend or the observer. Returns ErrFenced if the
	// lease was acquired by another holder or its record was removed, and
	// ErrLeaseExpired if the lease has expired or this holder already declared an
	// exit, even when no successor has acquired it. A lapsed holder cannot
	// record an exit, so its successor sees ExitExpired. Call Release with the
	// original ctx or a cleanup context, never the renewal context.
	Release(ctx context.Context, token Token, mode ExitMode) error

	// ReadCheckpoint returns the last checkpointed state and how the
	// immediately previous holder exited. On a first acquisition State is nil
	// and PrevExit is ExitNone. State is also nil after a Checkpoint with a nil
	// or empty state, and is never a non-nil empty slice. PrevExit and
	// PrevHolderID are fixed for the life of the lease. Returns ErrFenced if the
	// token no longer matches the stored lease or if no row exists for the work
	// ID, for example after Forget or Sweep.
	ReadCheckpoint(ctx context.Context, token Token) (Checkpoint, error)

	// StartRenewal begins automatic renewal of the lease at regular intervals.
	// Returns a derived context and a stop function. The renewal context is
	// cancelled if the parent context is cancelled, if the lease is fenced, or
	// if the lease window closes before a renewal succeeds. Calling the stop
	// function ends renewal without cancelling the renewal context, and it
	// returns without waiting for an in-flight Renew call to finish.
	// If Renew returns ErrLeaseExpired, storage has found the lease lapsed:
	// renewal stops at once without retrying, and the renewal context's cause
	// matches both ErrLeaseWindowExhausted and ErrLeaseExpired.
	// LeaseWindow reports the current lease window from the renewal context.
	StartRenewal(ctx context.Context, token Token, opts ...RenewalOption) (renewCtx context.Context, stopRenewal func())

	// Forget permanently deletes the lease record for token's workID. Returns
	// ErrFenced if the holder ID or fencing token no longer matches the stored lease, or if no
	// record exists for token's workID. Unlike Release, the record is not left
	// behind for a future ReadCheckpoint — it is gone. Use only when a work ID is
	// permanently retired. Every call is reported to LeaseObserver.OnForget, and
	// a fenced call also to OnFenced.
	Forget(ctx context.Context, token Token) error
}

Lease defines the contract for acquiring, managing, and renewing leases on distributed work. Implementations are responsible for handling backend storage, fencing, and expiration logic. All methods are safe for concurrent use.

Lease is not a stable target for implementations outside this module: a minor release may add a method to it. A wrapper or fake that must satisfy Lease embeds the Lease interface so that it keeps compiling.

func New

func New(b backend.Backend, cfg Config) (Lease, error)

New returns a new Lease instance backed by the provided Backend. Every error it returns matches ErrConfigInvalid with errors.Is:

  • ErrNilBackend if b is nil.
  • "worklease: New: TTL is required" if cfg.TTL is zero.
  • "worklease: New: HolderID is required" if cfg.HolderID is empty.
  • "worklease: New: TTL must be positive, got <TTL>" if cfg.TTL is negative.
  • "worklease: New: TTL must be at least 1ms, got <TTL>" if cfg.TTL is below 1ms.

The checks run in that order, and the first failing check is returned.

type LeaseObserver added in v0.2.0

type LeaseObserver interface {
	// OnAcquire is called after every backend acquire attempt, successful or not —
	// once per poll when WithWaitForLease is set. e.Duration is the duration of
	// that attempt's backend call only — not the wait loop.
	// e.Token is the zero value of Token if e.Err is non-nil.
	OnAcquire(ctx context.Context, e AcquireEvent)

	// OnCheckpoint is called after every Checkpoint attempt.
	// If e.Err is ErrFenced, OnFenced is also called after this method returns.
	OnCheckpoint(ctx context.Context, e CheckpointEvent)

	// OnRenew is called after every Renew attempt.
	// If e.Err is ErrFenced, OnFenced is also called after this method returns.
	OnRenew(ctx context.Context, e RenewEvent)

	// OnRelease is called after every Release attempt.
	// If e.Err is ErrFenced, OnFenced is also called after this method returns.
	OnRelease(ctx context.Context, e ReleaseEvent)

	// OnReadCheckpoint is called after every ReadCheckpoint attempt.
	// OnFenced is NOT called on ReadCheckpoint — ErrFenced surfaces via e.Err only.
	OnReadCheckpoint(ctx context.Context, e ReadCheckpointEvent)

	// OnForget is called after every Forget attempt.
	// If e.Err is ErrFenced, OnFenced is also called after this method returns.
	OnForget(ctx context.Context, e ForgetEvent)

	// OnFenced is called when Checkpoint, Renew, Release, or Forget returns
	// ErrFenced — not for ReadCheckpoint. Called in addition to the
	// operation-specific callback — not instead of.
	// e.Operation identifies which operation triggered the fencing event.
	OnFenced(ctx context.Context, e FencedEvent)
}

LeaseObserver receives callbacks after each Lease operation. All methods are called synchronously. Implementations must not block or panic. The zero value of Config.Observer is nil; the library substitutes a no-op observer.

A minor release may add a method to LeaseObserver. Embed NoopObserver in an observer type so that it keeps compiling.

type NoopObserver added in v0.8.0

type NoopObserver struct{}

NoopObserver is a LeaseObserver that discards every event. The library substitutes it when Config.Observer is nil. Embed it in an observer type to implement only the methods of interest; the embedding type keeps compiling when a later release adds a method to LeaseObserver.

Example
package main

import (
	"context"
	"fmt"
	"sync/atomic"
	"time"

	"github.com/aetomala/worklease"
	"github.com/aetomala/worklease/backend/memory"
)

// The acquireCounter type counts Acquire attempts and ignores every other event.
// Embedding worklease.NoopObserver supplies the other LeaseObserver methods.
type acquireCounter struct {
	worklease.NoopObserver
	n atomic.Int64
}

func (c *acquireCounter) OnAcquire(_ context.Context, _ worklease.AcquireEvent) {
	c.n.Add(1)
}

func main() {
	obs := &acquireCounter{}
	lease, err := worklease.New(memory.New(), worklease.Config{
		TTL:      30 * time.Second,
		HolderID: "worker-1",
		Observer: obs,
	})
	if err != nil {
		fmt.Println("error:", err)
		return
	}
	if _, err := lease.Acquire(context.Background(), "work-1"); err != nil {
		fmt.Println("error:", err)
		return
	}
	fmt.Println("acquires:", obs.n.Load())
}
Output:
acquires: 1

func (NoopObserver) OnAcquire added in v0.8.0

func (NoopObserver) OnAcquire(_ context.Context, _ AcquireEvent)

OnAcquire discards the event.

func (NoopObserver) OnCheckpoint added in v0.8.0

func (NoopObserver) OnCheckpoint(_ context.Context, _ CheckpointEvent)

OnCheckpoint discards the event.

func (NoopObserver) OnFenced added in v0.8.0

func (NoopObserver) OnFenced(_ context.Context, _ FencedEvent)

OnFenced discards the event.

func (NoopObserver) OnForget added in v0.8.0

func (NoopObserver) OnForget(_ context.Context, _ ForgetEvent)

OnForget discards the event.

func (NoopObserver) OnReadCheckpoint added in v0.8.0

func (NoopObserver) OnReadCheckpoint(_ context.Context, _ ReadCheckpointEvent)

OnReadCheckpoint discards the event.

func (NoopObserver) OnRelease added in v0.8.0

func (NoopObserver) OnRelease(_ context.Context, _ ReleaseEvent)

OnRelease discards the event.

func (NoopObserver) OnRenew added in v0.8.0

func (NoopObserver) OnRenew(_ context.Context, _ RenewEvent)

OnRenew discards the event.

type NoopVacuumObserver added in v0.8.0

type NoopVacuumObserver struct{}

NoopVacuumObserver is a VacuumObserver that discards every event. NewVacuum substitutes it when VacuumConfig.Observer is nil. Embed it in an observer type so that the embedding type keeps compiling when a later release adds a method to VacuumObserver.

func (NoopVacuumObserver) OnSweep added in v0.8.0

OnSweep discards the event.

type Operation added in v0.4.0

type Operation uint8

Operation identifies the Lease operation that triggered a fencing event. The zero value is not a valid Operation.

const (
	// OperationCheckpoint identifies a Checkpoint call.
	OperationCheckpoint Operation = iota + 1

	// OperationRenew identifies a Renew call.
	OperationRenew

	// OperationRelease identifies a Release call.
	OperationRelease

	// OperationForget identifies a Forget call.
	OperationForget
)

Operation values for fencing events. The values start at 1 so that the zero value is not a valid Operation.

func (Operation) String added in v0.8.0

func (o Operation) String() string

String returns "Checkpoint", "Renew", "Release", or "Forget". Any other value formats as "Operation(<n>)".

type ReadCheckpointEvent added in v0.4.0

type ReadCheckpointEvent struct {
	Token        Token
	Duration     time.Duration
	PrevExit     ExitMode
	PrevHolderID string
	Size         int
	Err          error
}

ReadCheckpointEvent carries the result of a ReadCheckpoint call. Size is the length of the returned state; 0 if nil. PrevExit is ExitNone and PrevHolderID is empty when Err is non-nil.

type ReleaseEvent added in v0.4.0

type ReleaseEvent struct {
	Token    Token
	Duration time.Duration
	Mode     ExitMode
	Err      error
}

ReleaseEvent carries the result of a Release call. Mode is the exit the caller declared. OnRelease is not called when Release rejects the mode with ErrInvalidExitMode.

type RenewEvent added in v0.4.0

type RenewEvent struct {
	Token    Token
	Duration time.Duration
	Attempt  int
	Err      error
}

RenewEvent carries the result of a Renew call. Attempt is 1 on the first attempt. The renewal goroutine increments Attempt on each retry attempt within the backoff loop. Direct calls to Renew from caller code always produce Attempt: 1. When the renewal goroutine abandons an in-flight attempt, Err is the error of the abandonment: context.Canceled on stopRenewal, the parent context's error on its cancellation, and context.DeadlineExceeded when the lease window closes. Duration then runs to the moment of abandonment.

type RenewalOption

type RenewalOption func(*renewalConfig)

RenewalOption is a functional option for StartRenewal.

func WithRenewalBackoff added in v0.5.0

func WithRenewalBackoff(initial, maxInterval time.Duration, jitter float64) RenewalOption

WithRenewalBackoff configures the exponential backoff policy used by the renewal goroutine when Renew returns a non-fencing, non-nil error. The initial parameter is the first retry interval; maxInterval caps the interval after growth. The jitter parameter is a fraction in [0, 1]; the actual wait is drawn from [base, base+jitter*base]. Defaults: initial=100ms, max=5s, jitter=0.20. Clamping (in order): initial floors to 1ms if <= 0; max floors to 1ms if <= 0; jitter clamps to [0, 1]; finally initial caps at max if initial > max.

func WithRenewalInterval

func WithRenewalInterval(d time.Duration) RenewalOption

WithRenewalInterval sets the time between renewal attempts. Default: TTL/2. Zero or negative values are ignored — the default is used silently.

type SweepEvent added in v0.7.0

type SweepEvent struct {
	Options  SweepOptions
	Deleted  int64
	Duration time.Duration
	Err      error
}

SweepEvent carries the result of one Sweep backend call. Options holds the options passed to Sweep, Deleted the number of rows the backend reported deleted, and Duration the duration of the backend call.

type SweepOptions added in v0.6.0

type SweepOptions = backend.SweepOptions

SweepOptions configures a Vacuum.Sweep call. It is a type alias for backend.SweepOptions — the canonical definition lives in package backend to avoid an import cycle (package backend cannot import package worklease).

type Token

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

Token represents a currently held lease. It is returned by Acquire and must be passed back to Checkpoint, Renew, Release, ReadCheckpoint, Forget, and StartRenewal. Renew and Checkpoint do not return a new Token — the same Token stays valid for the life of the lease. All fields are unexported; use accessor methods to read them. A Token is a value and is safe for concurrent use.

func NewToken added in v0.8.0

func NewToken(workID, holderID string, fencingToken uint64, expiresAt time.Time) Token

NewToken returns a Token holding the given fields, for tests and fakes that need a Token value without acquiring a lease. NewToken does not call a backend and does not validate its arguments. A Token it returns has no local lease window, so the window that StartRenewal bounds renewal by is already closed. Use Acquire to obtain a Token for a real lease.

Example
package main

import (
	"fmt"
	"time"

	"github.com/aetomala/worklease"
)

func main() {
	expires := time.Date(2026, 1, 1, 12, 0, 0, 0, time.UTC)
	token := worklease.NewToken("work-1", "worker-1", 7, expires)
	fmt.Println(token.WorkID(), token.HolderID(), token.FencingToken(), token.ExpiresAt().Format(time.RFC3339))
}
Output:
work-1 worker-1 7 2026-01-01T12:00:00Z

func (Token) ExpiresAt

func (t Token) ExpiresAt() time.Time

ExpiresAt returns the lease expiry recorded at acquisition, as reported by the backend's clock. It is not updated by Renew or Checkpoint. For a bound that reflects renewals, call LeaseWindow with the renewal context.

func (Token) FencingToken

func (t Token) FencingToken() uint64

FencingToken returns the monotonically increasing token that prevents stale operations on the lease. If the lease is reassigned, the token changes.

func (Token) HolderID

func (t Token) HolderID() string

HolderID returns the identifier of the entity holding the lease.

func (Token) String

func (t Token) String() string

String returns a string representation of the Token.

func (Token) WorkID

func (t Token) WorkID() string

WorkID returns the identifier for the unit of work being leased.

type Vacuum added in v0.6.0

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

Vacuum performs age-based bulk cleanup of lease rows. Sweep deletes rows released with ExitRetired once they are older than SweepOptions.Retention and, with SweepOptions.IncludeExpired, rows whose lease expired with no declared exit. Rows released with ExitFinished or ExitAbandoned, and rows currently held, are never deleted. A deleted row's checkpoint is gone: the next holder of that work ID sees ExitNone and empty state, and a holder whose row was deleted gets ErrFenced. Build a Vacuum with NewVacuum. A Vacuum is safe for concurrent use if its Backend and its VacuumObserver are.

func NewVacuum added in v0.6.0

func NewVacuum(b backend.Backend, cfg VacuumConfig) (*Vacuum, error)

NewVacuum returns a Vacuum backed by b. Returns ErrNilBackend if b is nil. The cfg argument may be the zero value.

func (*Vacuum) Sweep added in v0.6.0

func (v *Vacuum) Sweep(ctx context.Context, opts SweepOptions) (int64, error)

Sweep deletes rows that are not currently held and were last updated more than opts.Retention ago: rows released with ExitRetired and, with opts.IncludeExpired, rows whose lease expired with no declared exit. It returns the number of rows the backend reported deleted. Sweep returns ErrNilBackend, without calling the backend or the observer, on a Vacuum that NewVacuum did not build, including a nil *Vacuum. It returns ErrRetentionRequired, also without either call, if opts.Retention <= 0. Otherwise it calls the backend, reports the outcome to the configured VacuumObserver, and returns the backend's result unchanged.

type VacuumConfig added in v0.7.0

type VacuumConfig struct {
	// Observer receives the outcome of every Sweep that reaches the backend.
	// Optional; nil installs a no-op observer.
	Observer VacuumObserver
}

VacuumConfig configures a Vacuum.

type VacuumObserver added in v0.7.0

type VacuumObserver interface {
	// OnSweep is called after every Sweep backend call, successful or not. It
	// is not called when Sweep returns ErrNilBackend or ErrRetentionRequired,
	// since those return before the backend is called.
	OnSweep(ctx context.Context, e SweepEvent)
}

VacuumObserver receives a callback after each Vacuum.Sweep backend call. It is called synchronously. Implementations must not block or panic.

A minor release may add a method to VacuumObserver. Embed NoopVacuumObserver in an observer type so that it keeps compiling.

Directories

Path Synopsis
Package backend defines the storage contract that worklease backends implement, the types shared across it, and the sentinel errors that backends return.
Package backend defines the storage contract that worklease backends implement, the types shared across it, and the sentinel errors that backends return.
conformance
Package conformance provides a backend-agnostic Ginkgo specification that any worklease backend must satisfy.
Package conformance provides a backend-agnostic Ginkgo specification that any worklease backend must satisfy.
memory
Package memory implements the worklease Backend in process memory.
Package memory implements the worklease Backend in process memory.
postgres
Package postgres implements the worklease Backend on PostgreSQL.
Package postgres implements the worklease Backend on PostgreSQL.
Package checkpoint provides typed serialization helpers for worklease checkpoints: the Codec interface, a JSON codec, and the generic Encode and Decode functions.
Package checkpoint provides typed serialization helpers for worklease checkpoints: the Codec interface, a JSON codec, and the generic Encode and Decode functions.
internal
testutil
Package testutil is a generated GoMock package.
Package testutil is a generated GoMock package.
Package leader runs a function while holding a worklease lease, so that one process at a time acts as the leader for a work ID.
Package leader runs a function while holding a worklease lease, so that one process at a time acts as the leader for a work ID.
Package pool runs a work function for each of a set of work IDs, holding a worklease lease while each one runs.
Package pool runs a work function for each of a set of work IDs, holding a worklease lease while each one runs.
Package worker runs a work function under a worklease lease: it acquires the lease, passes the prior checkpoint to the function, and renews the lease while the function runs.
Package worker runs a work function under a worklease lease: it acquires the lease, passes the prior checkpoint to the function, and renews the lease while the function runs.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL