import { Pool } from "pg";
async function keyFor(pool: Pool, name: string): Promise<string> {
const { rows } = await pool.query<{ key: string }>(
"SELECT hashtextextended($1, 0) AS key",
[name]
);
return rows[0].key;
}
export async function withAdvisoryLock<T>(
pool: Pool,
name: string,
fn: () => Promise<T>
): Promise<{ ran: false } | { ran: true; result: T }> {
const key = await keyFor(pool, name);
const client = await pool.connect();
try {
const { rows } = await client.query<{ locked: boolean }>(
"SELECT pg_try_advisory_lock($1) AS locked",
[key]
);
if (!rows[0].locked) {
return { ran: false };
}
try {
const result = await fn();
return { ran: true, result };
} finally {
await client.query("SELECT pg_advisory_unlock($1)", [key]);
}
} finally {
client.release();
}
}
import { Pool } from "pg";
import { withAdvisoryLock } from "./advisoryLock";
export class ReportJob {
constructor(private readonly pool: Pool) {}
async run(): Promise<void> {
const outcome = await withAdvisoryLock(this.pool, "reports:daily", () =>
this.generateDailyReport()
);
if (!outcome.ran) {
console.log("[reports:daily] lock held elsewhere, skipping this run");
return;
}
console.log(`[reports:daily] finished, rows=${outcome.result}`);
}
private async generateDailyReport(): Promise<number> {
const { rowCount } = await this.pool.query(
`INSERT INTO daily_reports (day, total_orders, total_revenue)
SELECT current_date, count(*), coalesce(sum(amount_cents), 0)
FROM orders
WHERE created_at >= current_date
ON CONFLICT (day) DO UPDATE
SET total_orders = excluded.total_orders,
total_revenue = excluded.total_revenue`
);
return rowCount ?? 0;
}
}
import cron from "node-cron";
import { Pool } from "pg";
import { ReportJob } from "./ReportJob";
const pool = new Pool({
connectionString: process.env.DATABASE_URL,
max: 10,
});
const reportJob = new ReportJob(pool);
// Every instance registers this; the advisory lock ensures only one runs it.
cron.schedule("*/5 * * * *", async () => {
try {
await reportJob.run();
} catch (err) {
console.error("[reports:daily] failed", err);
}
});
process.on("SIGTERM", async () => {
await pool.end();
process.exit(0);
});
When multiple instances of a service run the same scheduled task, only one of them should actually do the work at any given moment. This snippet shows how to enforce that with a Postgres session-level advisory lock, which acts as a distributed mutex without needing a dedicated locking table or an external system like Redis.
The withAdvisoryLock helper in the advisoryLock util tab is the core. Advisory locks in Postgres are keyed by one 64-bit integer (or two 32-bit ints), so the helper first hashes a human-readable name into a stable bigint via pg_advisory_lock_key, which uses hashtextextended. It then calls pg_try_advisory_lock, the non-blocking variant that returns immediately with true or false instead of waiting. This matters: a cron job that blocks waiting for a lock will pile up and eventually exhaust the connection pool, so try semantics let a losing instance simply skip the run.
A crucial detail is that the lock must be acquired and released on the same physical connection. The helper checks out a dedicated client with pool.connect() rather than using pool.query, because pool queries can land on different backends. The try/finally guarantees pg_advisory_unlock runs even if the callback throws, and the client is always released back to the pool. Session-level locks are also auto-released if the connection dies, which is the safety net for crashes.
The ReportJob service tab wraps a real unit of work, generateDailyReport, in withAdvisoryLock under the key reports:daily. When the lock is not obtained, it logs and returns cleanly rather than erroring, so a skipped run is a normal outcome.
The scheduler tab wires it into node-cron and shows the payoff: every instance schedules the same job, but only the one that wins the lock executes the body. This pattern is ideal for idempotency-sensitive tasks — report generation, cache warming, cleanup sweeps — where double execution is wasteful or unsafe. The main trade-off is that advisory locks are advisory: nothing forces unrelated code to respect them, so the key naming convention must be shared and disciplined across the codebase.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
package dbutil
import (
"context"
"github.com/jackc/pgconn"Retry Postgres serialization failures with bounded attempts
require "csv"
class PeopleCsvStream
include Enumerable
HEADERS = %w[id full_name email signed_up_at plan].freezeResilient CSV Export as a Streamed Response
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
Share this code
Here's the card — post it anywhere.
