Combined pattern: advisory lock + savepoints

Lessons 02-05 covered advisory locks. Lesson 06 covered savepoints. This lesson combines the two into the canonical pattern for robust job runners: a global advisory lock to guarantee single-instance, a savepoint per batch item for fault tolerance. It's the pattern you'll use in any serious worker.

By the end you'll have the code that demonstrates integrated mastery of the module's two topics.

The conceptual pattern

A worker that processes a task queue:

1. Worker starts
2. Acquire an advisory lock (single instance)
   └─ If another worker holds the lock → exit
3. For each task in the queue:
   a. Create a savepoint
   b. Process the task
   c. If success → release savepoint, mark task as completed
   d. If it fails → rollback to savepoint, mark task as failed, continue
4. Release the advisory lock when done

Characteristics of the pattern:

• Single-instance: only one worker runs at a time (advisory lock).
• Fault tolerant: if one task fails, the rest keep processing (savepoint).
• Persistent: each processed task is committed independently (there's no global rollback).
• Observable: each task records its state (completed/failed).

Setup: the jobs table

# app/models/job.py
from datetime import datetime, timezone
import enum
import uuid
from sqlalchemy import String, DateTime, JSON, Index, Enum
from sqlalchemy.dialects.postgresql import UUID, JSONB
from sqlalchemy.orm import Mapped, mapped_column

from app.database import Base


class JobStatus(str, enum.Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"


class Job(Base):
    __tablename__ = "jobs"

    id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
    job_type: Mapped[str] = mapped_column(String(50))
    payload: Mapped[dict] = mapped_column(JSONB)
    status: Mapped[JobStatus] = mapped_column(Enum(JobStatus), default=JobStatus.PENDING)
    error_message: Mapped[str | None] = mapped_column(String(1000), nullable=True)
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)
    )
    completed_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)

    __table_args__ = (
        Index("idx_jobs_pending", "status", "created_at",
              postgresql_where="status = 'pending'"),
    )

Migration:

alembic revision --autogenerate -m "add jobs table"
alembic upgrade head

The job runner

# app/services/job_runner.py
import asyncio
from datetime import datetime, timezone
from typing import Any
from sqlalchemy import select, text
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import SessionLocal
from app.models import Job, JobStatus
from app.services.advisory_locks import advisory_session_lock, LockNamespace


JOB_RUNNER_LOCK_KEY = 9001  # Single instance lock


class JobRunner:
    """Worker that processes a job queue with single-instance + fault tolerance."""

    def __init__(self, batch_size: int = 10):
        self.batch_size = batch_size
        self.processed = 0
        self.failed = 0

    async def run(self) -> dict[str, Any]:
        """Process pending jobs. Returns stats."""
        async with SessionLocal() as session:
            # 1. Acquire the advisory lock (session-level because the worker lasts a long time)
            async with advisory_session_lock(session, JOB_RUNNER_LOCK_KEY) as got_lock:
                if not got_lock:
                    return {"status": "skipped", "reason": "another worker active"}

                # 2. Process jobs
                while True:
                    jobs = await self._fetch_pending_batch(session)
                    if not jobs:
                        break

                    for job in jobs:
                        await self._process_job_safely(session, job)

            # Lock released automatically here

        return {
            "status": "completed",
            "processed": self.processed,
            "failed": self.failed,
        }

    async def _fetch_pending_batch(self, session: AsyncSession) -> list[Job]:
        """Fetch next batch of pending jobs with row-level lock."""
        # SKIP LOCKED prevents two workers from taking the same job (extra
        # defense, even though the advisory lock already guarantees single instance)
        result = await session.execute(text(f"""
            SELECT id FROM jobs
            WHERE status = 'pending'
            ORDER BY created_at
            LIMIT {self.batch_size}
            FOR UPDATE SKIP LOCKED
        """))
        ids = [row[0] for row in result]

        if not ids:
            return []

        # Mark as processing
        await session.execute(text("""
            UPDATE jobs SET status = 'processing'
            WHERE id = ANY(:ids)
        """), {"ids": ids})
        await session.commit()  # commit the status change

        # Fetch full objects
        result = await session.execute(
            select(Job).where(Job.id.in_(ids))
        )
        return list(result.scalars())

    async def _process_job_safely(self, session: AsyncSession, job: Job):
        """Process a single job with savepoint for fault tolerance."""
        try:
            async with session.begin():  # Outer tx per job
                async with session.begin_nested():  # Savepoint
                    # Process inside savepoint
                    await self._do_job_work(session, job)

                # If reached here, job succeeded — mark completed
                job_db = await session.get(Job, job.id)
                job_db.status = JobStatus.COMPLETED
                job_db.completed_at = datetime.now(timezone.utc)
                self.processed += 1
                # Auto-commit at end of `async with session.begin()`

        except Exception as e:
            # Job failed — record error in separate transaction
            async with session.begin():
                job_db = await session.get(Job, job.id)
                if job_db:
                    job_db.status = JobStatus.FAILED
                    job_db.error_message = str(e)[:1000]
                    job_db.completed_at = datetime.now(timezone.utc)
            self.failed += 1

    async def _do_job_work(self, session: AsyncSession, job: Job):
        """Actual work for a job. Override per job type."""
        if job.job_type == "send_email":
            await self._send_email(session, job)
        elif job.job_type == "generate_report":
            await self._generate_report(session, job)
        else:
            raise ValueError(f"Unknown job type: {job.job_type}")

    async def _send_email(self, session: AsyncSession, job: Job):
        # Mock — in real production it would make external calls
        print(f"Sending email: {job.payload}")
        await asyncio.sleep(0.1)
        # If the email fails for some reason (template error, API down), the exception propagates
        # → savepoint rollback, job marked failed, other jobs continue

    async def _generate_report(self, session: AsyncSession, job: Job):
        print(f"Generating report: {job.payload}")
        await asyncio.sleep(0.5)

Why this structure

1. A session-level lock outside the loop:

async with advisory_session_lock(session, JOB_RUNNER_LOCK_KEY) as got_lock:
    while True:
        jobs = await fetch()
        for job in jobs:
            ...

The lock lasts the whole run of the worker, not just one transaction. If the worker processes 1000 jobs in 30 minutes, the lock is held that entire time.

2. A transaction per job (not per batch):

async with session.begin():  # Tx per job
    async with session.begin_nested():  # Savepoint
        await do_work(job)
    # Mark completed
# Commit

Each job is its own tx. If it commits, the job is persisted. If there's an error, the job is marked failed. Other jobs are not affected.

Without this (everything in one big tx): a failed job could roll back the previous ones. Job persistence → tx per job.

3. A savepoint inside the job's tx:

async with session.begin_nested():
    await do_work(job)

If do_work raises an exception, the savepoint rolls back the operation's changes. The outer tx stays valid so it can run the UPDATE jobs SET status = 'failed' after the catch.

Without a savepoint, the outer tx would be aborted when do_work fails and you couldn't mark the job as failed.

4. FOR UPDATE SKIP LOCKED as additional defense:

Even though the advisory lock guarantees a single instance, SKIP LOCKED is additional defense against race conditions:

SELECT id FROM jobs WHERE status = 'pending'
FOR UPDATE SKIP LOCKED

If for some reason another process ALSO tries to take jobs, SKIP LOCKED prevents both from taking the same one. Defense in depth.

Variant: periodic cron

For crons (not daemons), simplify:

# app/tasks/cron_runner.py
async def run_cron_with_lock(
    cron_name: str,
    work_func,
    lock_namespace: int = LockNamespace.CRON,
):
    """Run cron with advisory lock to prevent concurrent runs."""
    lock_key = hash(cron_name) & 0x7FFFFFFF

    async with SessionLocal() as session:
        async with session.begin():  # Single tx for crons
            async with advisory_xact_lock_ns(
                session, lock_namespace, lock_key
            ) as got_lock:
                if not got_lock:
                    print(f"[{cron_name}] Already running, skip")
                    return

                await work_func(session)
                print(f"[{cron_name}] Completed")
        # Lock released on commit


# Usage
async def reindex_posts(session: AsyncSession):
    await session.execute(text("""
        UPDATE posts
        SET search_vector = to_tsvector('spanish', title || ' ' || content)
        WHERE search_vector IS NULL
    """))


if __name__ == "__main__":
    asyncio.run(run_cron_with_lock("reindex_posts", reindex_posts))

For crons that process many items with savepoints:

async def reindex_with_partial_failure(session: AsyncSession):
    """Reindex with savepoints — a post that fails doesn't affect others."""
    posts_to_reindex = await fetch_pending_reindex(session)

    success = 0
    failed = 0

    for post in posts_to_reindex:
        try:
            async with session.begin_nested():
                await recalculate_tsvector(session, post)
            success += 1
        except Exception as e:
            failed += 1
            print(f"Failed post {post.id}: {e}")

    print(f"Reindexed: {success} ok, {failed} failed")

Cron + advisory lock + savepoints = the canonical pattern.

Real-world case: FTS re-indexing in BlogPlatform

# app/tasks/reindex_fts.py
import asyncio
from sqlalchemy import select, text, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.database import SessionLocal
from app.models import Post
from app.services.advisory_locks import advisory_xact_lock_ns, LockNamespace


REINDEX_LOCK = hash("cron:reindex_fts") & 0x7FFFFFFF


async def reindex_fts():
    """Periodic cron that recalculates tsvector on modified posts."""
    async with SessionLocal() as session:
        async with session.begin():
            async with advisory_xact_lock_ns(
                session, LockNamespace.CRON, REINDEX_LOCK
            ) as got:
                if not got:
                    print("Another reindex active")
                    return

                # Fetch posts to reindex (NULL search_vector or stale)
                posts = await session.scalars(
                    select(Post).where(Post.search_vector.is_(None)).limit(1000)
                )
                posts = list(posts)

                if not posts:
                    print("No posts to reindex")
                    return

                # Process each with savepoint
                success = 0
                failed = []

                for post in posts:
                    try:
                        async with session.begin_nested():
                            # Recalculate tsvector
                            new_vector = await session.scalar(
                                text("""
                                    SELECT to_tsvector('spanish',
                                        coalesce(:title, '') || ' ' || coalesce(:content, ''))
                                """),
                                {"title": post.title, "content": post.content}
                            )
                            post.search_vector = new_vector
                        success += 1
                    except Exception as e:
                        failed.append({"id": str(post.id), "error": str(e)})

                print(f"Reindexed {success} posts, {len(failed)} failed")
                if failed:
                    print(f"Failures: {failed}")


if __name__ == "__main__":
    asyncio.run(reindex_fts())

Crontab:

*/15 * * * * cd /app && python -m app.tasks.reindex_fts

If they overlap → the second run skips. If a post fails on invalid content → savepoint rollback, the other 999 get processed. A robust cron.

Tests for the combined pattern

# tests/test_job_runner.py
import pytest
import asyncio
from app.services.job_runner import JobRunner
from app.models import Job, JobStatus


@pytest.mark.asyncio
async def test_only_one_runner_active(db_session):
    """Only one runner should run at a time."""
    # Create jobs
    for i in range(5):
        db_session.add(Job(
            job_type="send_email",
            payload={"to": f"user-{i}@test.com"}
        ))
    await db_session.commit()

    # Run two workers in parallel
    runner1 = JobRunner()
    runner2 = JobRunner()

    results = await asyncio.gather(runner1.run(), runner2.run())

    # One processes, the other skips
    skipped = [r for r in results if r["status"] == "skipped"]
    completed = [r for r in results if r["status"] == "completed"]

    assert len(skipped) == 1
    assert len(completed) == 1


@pytest.mark.asyncio
async def test_failed_job_doesnt_affect_others(db_session, monkeypatch):
    """A job that fails should not affect others."""
    # Create jobs, one with a "bad" payload that triggers an error
    db_session.add(Job(job_type="send_email", payload={"to": "ok-1@test.com"}))
    db_session.add(Job(job_type="send_email", payload={"to": "BAD"}))  # will trigger an error
    db_session.add(Job(job_type="send_email", payload={"to": "ok-2@test.com"}))
    await db_session.commit()

    # Mock _send_email to fail on the "BAD" payload
    async def mock_send(session, job):
        if job.payload.get("to") == "BAD":
            raise ValueError("Bad payload")

    monkeypatch.setattr(JobRunner, "_send_email", mock_send)

    runner = JobRunner()
    result = await runner.run()

    assert result["processed"] == 2
    assert result["failed"] == 1

    # Verify final status in the DB
    jobs = await db_session.scalars(select(Job).order_by(Job.created_at))
    statuses = [j.status for j in jobs]
    assert JobStatus.COMPLETED in statuses
    assert JobStatus.FAILED in statuses


@pytest.mark.asyncio
async def test_runner_is_idempotent(db_session):
    """Running multiple times doesn't duplicate work."""
    db_session.add(Job(job_type="send_email", payload={"to": "test@test.com"}))
    await db_session.commit()

    runner = JobRunner()
    result1 = await runner.run()
    assert result1["processed"] == 1

    # Second run — no pending jobs
    result2 = await runner.run()
    assert result2["processed"] == 0

Traps and common mistakes

1. A session-level lock with PgBouncer transaction mode.

A reminder from lesson 03: with PgBouncer transaction mode, session-level locks are NOT released when you "return" the conn to the pool. For job runners, connect bypassing PgBouncer (port 5432 directly), not via PgBouncer.

2. A tx per job without committing it.

async with session.begin():
    async with session.begin_nested():
        await do_work()
    job.status = COMPLETED
# If you forget `await session.commit()` and the session closes, it doesn't persist

async with session.begin() auto-commits on exit IF there was no exception. Verify.

3. Catching errors that are too generic.

try:
    async with session.begin_nested():
        await do_work()
except Exception:
    pass  # Silences real bugs

A specific catch (IntegrityError, DataError, custom errors) and log the rest.

4. A worker that takes too long holding the lock.

If an individual job takes 1 hour, the worker holds the lock that entire time. Other workers wait. If there are timeouts or a crash, recovery is slow.

Mitigation:

• Short jobs (<1min ideally).
• If a job is necessarily long, consider splitting it.

5. Not detecting dead workers.

A worker dies from OOM. The session-level lock is held until TCP keepalive detects it (hours). Meanwhile, no jobs are processed.

Mitigation:

• Aggressive TCP keepalive.
• A heartbeat-based liveness check.
• Fallback: a cron monitor that detects a dead worker and releases the lock manually.

6. FOR UPDATE SKIP LOCKED without committing the status update.

# ❌
result = await session.execute(text("""
    SELECT id FROM jobs WHERE status = 'pending'
    FOR UPDATE SKIP LOCKED LIMIT 10
"""))
# Without committing the status, other workers can take the same jobs

After SELECT FOR UPDATE, run UPDATE status and commit before processing.

7. Mixing transaction-level and session-level locks for the same key.

As in lesson 03 — don't mix them. One piece of code uses session-level, another transaction-level → unexpected conflicts.

8. No retry logic.

If a job fails for a temporary reason (network, deadlock), it should be able to retry. Implement retry_count on Job and retry logic with backoff.

Exercise: implement the complete pattern

Setup: Job + JobStatus models + the table.

Step 1: implement JobRunner following the code from the section.

Step 2: implement 2-3 job types:

async def _send_email(self, session, job):
    print(f"Email to {job.payload['to']}")

async def _generate_report(self, session, job):
    print(f"Report: {job.payload['report_id']}")

async def _process_payment(self, session, job):
    # more interesting: does an UPDATE on another table
    user_id = job.payload["user_id"]
    amount = job.payload["amount"]
    await session.execute(
        text("UPDATE users SET balance = balance + :amt WHERE id = :uid"),
        {"amt": amount, "uid": user_id}
    )

Step 3: seed jobs and run.

# Seed
for i in range(20):
    job = Job(
        job_type="send_email",
        payload={"to": f"user-{i}@test.com"}
    )
    session.add(job)
await session.commit()

# Run
runner = JobRunner()
result = await runner.run()
print(result)

Step 4: test concurrency.

Run two workers in parallel:

async def main():
    await asyncio.gather(
        JobRunner().run(),
        JobRunner().run(),
    )

Expect only one to do the work.

Step 5: test fault tolerance.

Force errors on some jobs and verify that the others get processed.

# Job with a bad payload
session.add(Job(job_type="send_email", payload={"to": None}))  # force an error

Verify afterward: ok jobs = completed, the bad job = failed with an error message.

{"status": "completed", "processed": 20, "failed": 0}
{"status": "skipped", "reason": "another worker active"}

{"status": "completed", "processed": 19, "failed": 1}

SELECT status, COUNT(*) FROM jobs GROUP BY status;
-- completed | 19
-- failed    | 1

Summary and next step

What you learned:

• The canonical pattern: a global advisory lock (single-instance) + a savepoint per item (fault tolerance).
• Structure: an outer lock, a loop with tx-per-job + a savepoint inside.
• FOR UPDATE SKIP LOCKED as extra defense against races.
• Variant for crons: simplification with advisory_xact_lock_ns.
• Tests: simulate concurrency and item errors.
• Traps: PgBouncer + session-level, held locks, dead workers.

Before moving on, you should be able to:

• Implement a complete job runner from scratch.
• Combine advisory lock + savepoint in any worker.
• Decide between a daemon runner vs a periodic cron.
• Apply the pattern to the capstone project.

In the next lesson we close the module with the mini-project: a complete job runner with a task queue, throughput metrics, retry logic with exponential backoff, and benchmarks. It's the deliverable that demonstrates mastery of the module and connects with the guide's final project.

Resources

1. PostgreSQL — FOR UPDATE SKIP LOCKED — reference.
2. Brandur Leach — Job queues with PostgreSQL — deep dive.
3. Sidekiq — Worker uniqueness — an applicable Ruby pattern.
4. Better job queue with SKIP LOCKED — real-world case.
5. Citus Data — Coordination patterns — real-world patterns.

Lesson 07 of 08 — Module 6 — Advanced PostgreSQL for Backend Guide