Retrying Serialisation Failures

# psycopg 3, with the pre-built binary package that bundles libpq.
psycopg[binary]>=3.2,<4

{
  "name": "retry-transactions",
  "private": true,
  "type": "module",
  "scripts": {
    "build": "tsc",
    "start": "node dist/main.js"
  },
  "dependencies": {
    "pg": "^8.23.1"
  },
  "devDependencies": {
    "@types/node": "^22.0.0",
    "@types/pg": "^8.23.1",
    "typescript": "^5.9.0"
  }
}

<dependencies>
  <!-- The PostgreSQL JDBC driver (pgJDBC). -->
  <dependency>
    <groupId>org.postgresql</groupId>
    <artifactId>postgresql</artifactId>
    <version>42.7.14</version>
  </dependency>
</dependencies>

<ItemGroup>
  <!-- Npgsql, the ADO.NET data provider for PostgreSQL. -->
  <PackageReference Include="Npgsql" Version="10.0.3" />
</ItemGroup>

// pgx is a pure Go driver and toolkit for PostgreSQL.
require github.com/jackc/pgx/v5 v5.12.0

require (
    github.com/jackc/pgpassfile v1.0.0 // indirect
    github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect
    golang.org/x/text v0.41.0 // indirect
)

[dependencies]
# The synchronous client; tokio-postgres is the asynchronous equivalent.
postgres = "0.19"

source "https://rubygems.org"

# The pg gem wraps libpq, so libpq and its headers must be installed.
gem "pg", "~> 1.6"

# Build with libpq, using pg_config to find the headers and library.
CFLAGS += -Wall -Wextra -I$(shell pg_config --includedir)
LDFLAGS += -L$(shell pg_config --libdir)
LDLIBS += -lpq

main: main.c

import os
import sys
import time

import psycopg

import { setTimeout as sleep } from "node:timers/promises";

import pg from "pg";

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Properties;
import java.util.Set;

using System.Data;
using Npgsql;

import (
    "context"
    "errors"
    "fmt"
    "net"
    "net/url"
    "os"
    "time"

    "github.com/jackc/pgx/v5"
    "github.com/jackc/pgx/v5/pgconn"
)

use std::env;
use std::error::Error;
use std::process::ExitCode;
use std::thread::sleep;
use std::time::Duration;

use postgres::error::SqlState;
use postgres::{Client, Config, IsolationLevel, NoTls, Transaction};

require "pg"

#include <stdbool.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>

#include <libpq-fe.h>

# The SQLSTATE codes that mean "run the transaction again": 40001 is a
# serialisation failure and 40P01 is a deadlock.
RETRYABLE = {"40001", "40P01"}
MAX_ATTEMPTS = 5


def connection_settings() -> dict:
    """Reads the connection settings from the standard libpq environment
    variables, with defaults suitable for a local development server.

    The or operator treats an empty variable as unset, as libpq does, and
    a password of None is simply not sent.
    """
    return {
        "host": os.environ.get("PGHOST") or "localhost",
        "port": os.environ.get("PGPORT") or "5432",
        "dbname": os.environ.get("PGDATABASE") or "postgres",
        "user": os.environ.get("PGUSER") or "postgres",
        "password": os.environ.get("PGPASSWORD") or None,
    }

// The SQLSTATE codes that mean "run the transaction again": 40001 is a
// serialisation failure and 40P01 is a deadlock.
const RETRYABLE = new Set(["40001", "40P01"]);
const MAX_ATTEMPTS = 5;

// Read the connection settings from the standard libpq environment
// variables, with defaults suitable for a local development server.
// The || operator, unlike ??, also treats an empty variable as unset, as
// libpq does.
const config: pg.ClientConfig = {
  host: process.env.PGHOST || "localhost",
  port: Number(process.env.PGPORT || "5432"),
  database: process.env.PGDATABASE || "postgres",
  user: process.env.PGUSER || "postgres",
  password: process.env.PGPASSWORD || undefined,
};

/**
 * The SQLSTATE codes that mean "run the transaction again": 40001 is
 * a serialisation failure and 40P01 is a deadlock.
 */
static final Set<String> RETRYABLE = Set.of("40001", "40P01");
static final int MAX_ATTEMPTS = 5;

/** Returns the value of an environment variable, or a fallback. */
static String envOr(String name, String fallback) {
    String value = System.getenv(name);
    return (value != null && !value.isEmpty()) ? value : fallback;
}

/**
 * Returns the JDBC URL that identifies the server and database, built
 * from the standard libpq environment variables.
 */
static String connectionUrl() {
    return "jdbc:postgresql://"
            + envOr("PGHOST", "localhost") + ":"
            + envOr("PGPORT", "5432") + "/"
            + envOr("PGDATABASE", "postgres");
}

/**
 * Returns the user name and password as connection properties; the
 * password is omitted if PGPASSWORD is unset or empty.
 */
static Properties connectionProperties() {
    Properties props = new Properties();
    props.setProperty("user", envOr("PGUSER", "postgres"));
    String password = envOr("PGPASSWORD", null);
    if (password != null) {
        props.setProperty("password", password);
    }
    return props;
}

const int MaxAttempts = 5;

// Returns true if a SQLSTATE code means "run the transaction again": 40001
// is a serialisation failure and 40P01 is a deadlock.
static bool IsRetryable(string? sqlState) => sqlState is "40001" or "40P01";

// Returns the value of an environment variable, or null if the variable
// is unset or empty; libpq also treats an empty variable as unset.
static string? EnvValue(string name) =>
    Environment.GetEnvironmentVariable(name) is { Length: > 0 } value
        ? value
        : null;

// Returns the value of an environment variable, or a fallback if unset.
static string EnvOr(string name, string fallback) =>
    EnvValue(name) ?? fallback;

// Builds a connection string from the standard libpq environment
// variables, with defaults suitable for a local development server. The
// builder takes care of quoting any special characters.
static string ConnectionString() =>
    new NpgsqlConnectionStringBuilder
    {
        Host = EnvOr("PGHOST", "localhost"),
        Port = int.Parse(EnvOr("PGPORT", "5432")),
        Database = EnvOr("PGDATABASE", "postgres"),
        Username = EnvOr("PGUSER", "postgres"),
        Password = EnvValue("PGPASSWORD"),
    }.ConnectionString;

// maxAttempts is the number of times the transaction runs before the
// program gives up.
const maxAttempts = 5

// isRetryable reports whether err carries a SQLSTATE code that means "run
// the transaction again": 40001 is a serialisation failure and 40P01 is a
// deadlock.
func isRetryable(err error) bool {
    var pgErr *pgconn.PgError
    return errors.As(err, &pgErr) &&
        (pgErr.Code == "40001" || pgErr.Code == "40P01")
}

// envOr returns the value of an environment variable, or a fallback if
// the variable is unset or empty.
func envOr(name, fallback string) string {
    if value := os.Getenv(name); value != "" {
        return value
    }
    return fallback
}

// connString builds a postgres:// URL from the standard libpq environment
// variables. net/url takes care of escaping any special characters in
// the user name, password, or database name.
func connString() string {
    u := url.URL{
        Scheme: "postgres",
        Host:   net.JoinHostPort(envOr("PGHOST", "localhost"), envOr("PGPORT", "5432")),
        Path:   "/" + envOr("PGDATABASE", "postgres"),
    }
    if password := os.Getenv("PGPASSWORD"); password != "" {
        u.User = url.UserPassword(envOr("PGUSER", "postgres"), password)
    } else {
        u.User = url.User(envOr("PGUSER", "postgres"))
    }
    return u.String()
}

const MAX_ATTEMPTS: u32 = 5;

/// Returns true if the error carries a SQLSTATE code that means "run the
/// transaction again": a serialisation failure or a deadlock.
fn is_retryable(error: &postgres::Error) -> bool {
    matches!(
        error.code(),
        Some(code) if *code == SqlState::T_R_SERIALIZATION_FAILURE
            || *code == SqlState::T_R_DEADLOCK_DETECTED
    )
}

/// Returns the value of an environment variable, if it is set and not
/// empty; libpq also treats an empty variable as unset.
fn env_value(name: &str) -> Option<String> {
    env::var(name).ok().filter(|value| !value.is_empty())
}

/// Returns the value of an environment variable, or a fallback if the
/// variable is unset or empty.
fn env_or(name: &str, fallback: &str) -> String {
    env_value(name).unwrap_or_else(|| fallback.to_string())
}

/// Builds the connection configuration from the standard libpq
/// environment variables, with defaults suitable for a local server.
fn config() -> Config {
    let mut config = Config::new();
    config
        .host(&env_or("PGHOST", "localhost"))
        .port(env_or("PGPORT", "5432").parse().expect("PGPORT must be a number"))
        .dbname(&env_or("PGDATABASE", "postgres"))
        .user(&env_or("PGUSER", "postgres"));
    if let Some(password) = env_value("PGPASSWORD") {
        config.password(&password);
    }
    config
}

# The SQLSTATE codes that mean "run the transaction again": 40001 is a
# serialisation failure and 40P01 is a deadlock.
RETRYABLE = %w[40001 40P01].freeze
MAX_ATTEMPTS = 5

# Returns the value of an environment variable, or a fallback if the
# variable is unset or empty; libpq also treats an empty variable as unset.
def env_or(name, fallback)
  value = ENV[name]
  value.nil? || value.empty? ? fallback : value
end

# Returns the connection settings from the standard libpq environment
# variables, with defaults suitable for a local development server.
# compact removes the password if PGPASSWORD is not set.
def connection_params
  {
    host: env_or("PGHOST", "localhost"),
    port: env_or("PGPORT", "5432"),
    dbname: env_or("PGDATABASE", "postgres"),
    user: env_or("PGUSER", "postgres"),
    password: env_or("PGPASSWORD", nil)
  }.compact
end

#define MAX_ATTEMPTS 5

/*
 * Return true if a SQLSTATE code means "run the transaction again": 40001
 * is a serialisation failure and 40P01 is a deadlock.
 */
static bool
is_retryable(const char *sqlstate)
{
    return strcmp(sqlstate, "40001") == 0 || strcmp(sqlstate, "40P01") == 0;
}

/* Return the value of an environment variable, or a fallback if unset. */
static const char *
env_or(const char *name, const char *fallback)
{
    const char *value = getenv(name);

    return (value != NULL && value[0] != '\0') ? value : fallback;
}

/*
 * Connection parameters are passed to libpq as two parallel,
 * NULL-terminated arrays of keywords and values. A NULL value is ignored,
 * so the password is only sent if PGPASSWORD is set.
 */
static const char *const keywords[] = {
    "host", "port", "dbname", "user", "password", NULL
};

/* Fill in the value for each keyword from the environment. */
static void
read_settings(const char *values[])
{
    values[0] = env_or("PGHOST", "localhost");
    values[1] = env_or("PGPORT", "5432");
    values[2] = env_or("PGDATABASE", "postgres");
    values[3] = env_or("PGUSER", "postgres");
    values[4] = env_or("PGPASSWORD", NULL);
    values[5] = NULL;
}

def connect() -> psycopg.Connection:
    """Opens a connection with the settings from the environment.

    Autocommit mode means that the program, rather than the driver,
    decides where each transaction begins and ends.
    """
    return psycopg.connect(**connection_settings(), autocommit=True)

// Opens a connection with the settings from the environment. The driver
// runs each statement in its own transaction unless the program sends
// BEGIN, so the program decides where each transaction starts and ends.
async function connect(): Promise<pg.Client> {
  const client = new pg.Client(config);
  try {
    await client.connect();
  } catch (err) {
    // Release the client before passing the failure to the caller.
    await client.end();
    throw err;
  }
  return client;
}

/**
 * Opens a connection with the settings from the environment. The
 * connection starts in autocommit mode, where each statement is its
 * own transaction.
 */
static Connection connect() throws SQLException {
    return DriverManager.getConnection(
            connectionUrl(), connectionProperties());
}

// Creates a data source with the settings from the environment. A data
// source manages a pool of physical connections, and each call to
// OpenConnectionAsync borrows one from the pool, which also gives the
// program a second connection when it needs one. A connection runs each
// statement in its own transaction unless the program begins one.
static NpgsqlDataSource CreateDataSource() =>
    NpgsqlDataSource.Create(ConnectionString());

// connect opens a connection with the settings from the environment.
func connect(ctx context.Context) (*pgx.Conn, error) {
    conn, err := pgx.Connect(ctx, connString())
    if err != nil {
        return nil, fmt.Errorf("connection failed: %w", err)
    }
    return conn, nil
}

/// Opens a connection with the settings from the environment. NoTls
/// disables TLS, which keeps this example free of certificate set-up;
/// connections over an untrusted network should use TLS. The server runs
/// each statement in its own transaction unless the program begins one, so
/// the program decides where each transaction starts and ends.
fn connect() -> Result<Client, postgres::Error> {
    config().connect(NoTls)
}

# Opens a connection with the settings from the environment. The server
# runs each statement in its own transaction unless the program sends
# BEGIN, so the program decides where each transaction starts and ends.
def connect
  PG.connect(connection_params)
end

/*
 * Open a connection with the settings from the environment. libpq always
 * returns a connection object, which the caller must check with PQstatus()
 * and release with PQfinish(), even if the connection attempt failed. The
 * server runs each statement in its own transaction unless the program
 * sends BEGIN, so the program decides where each transaction starts and
 * ends.
 */
static PGconn *
connect_to_server(void)
{
    const char *values[sizeof(keywords) / sizeof(keywords[0])];

    read_settings(values);
    return PQconnectdbParams(keywords, values, 0);
}

def create_table(conn: psycopg.Connection) -> None:
    """Creates the shared counter table, with a single row set to zero.

    The table must be visible to a second connection, so it cannot be
    temporary; the name is specific to this recipe, and any earlier copy
    is dropped first.
    """
    drop_table(conn)
    conn.execute(
        "CREATE TABLE cookbook_retry_transactions "
        "(id int PRIMARY KEY, value int NOT NULL)"
    )
    conn.execute(
        "INSERT INTO cookbook_retry_transactions (id, value) VALUES (%s, %s)",
        (1, 0),
    )


def drop_table(conn: psycopg.Connection) -> None:
    """Drops the shared counter table if it exists."""
    conn.execute("DROP TABLE IF EXISTS cookbook_retry_transactions")

// Drops the shared counter table if it exists.
async function dropTable(client: pg.Client): Promise<void> {
  await client.query("DROP TABLE IF EXISTS cookbook_retry_transactions");
}

// Creates the shared counter table, with a single row set to zero. The
// table must be visible to a second connection, so it cannot be
// temporary; the name is specific to this recipe, and any earlier copy is
// dropped first.
async function createTable(client: pg.Client): Promise<void> {
  await dropTable(client);
  await client.query(
    "CREATE TABLE cookbook_retry_transactions " +
      "(id int PRIMARY KEY, value int NOT NULL)",
  );
  await client.query(
    "INSERT INTO cookbook_retry_transactions (id, value) VALUES ($1, $2)",
    [1, 0],
  );
}

/** Drops the shared counter table if it exists. */
static void dropTable(Connection conn) throws SQLException {
    try (Statement stmt = conn.createStatement()) {
        stmt.execute("DROP TABLE IF EXISTS cookbook_retry_transactions");
    }
}

/**
 * Creates the shared counter table, with a single row set to zero.
 * The table must be visible to a second connection, so it cannot be
 * temporary; the name is specific to this recipe, and any earlier
 * copy is dropped first.
 */
static void createTable(Connection conn) throws SQLException {
    dropTable(conn);
    try (Statement stmt = conn.createStatement()) {
        stmt.execute("CREATE TABLE cookbook_retry_transactions "
                + "(id int PRIMARY KEY, value int NOT NULL)");
    }
    try (PreparedStatement stmt = conn.prepareStatement(
            "INSERT INTO cookbook_retry_transactions (id, value) "
            + "VALUES (?, ?)")) {
        stmt.setInt(1, 1);
        stmt.setInt(2, 0);
        stmt.executeUpdate();
    }
}

// Drops the shared counter table if it exists.
static async Task DropTableAsync(NpgsqlConnection conn)
{
    await using var cmd = new NpgsqlCommand(
        "DROP TABLE IF EXISTS cookbook_retry_transactions", conn);
    await cmd.ExecuteNonQueryAsync();
}

// Creates the shared counter table, with a single row set to zero. The
// table must be visible to a second connection, so it cannot be
// temporary; the name is specific to this recipe, and any earlier copy is
// dropped first.
static async Task CreateTableAsync(NpgsqlConnection conn)
{
    await DropTableAsync(conn);

    await using var create = new NpgsqlCommand(
        "CREATE TABLE cookbook_retry_transactions " +
        "(id int PRIMARY KEY, value int NOT NULL)", conn);
    await create.ExecuteNonQueryAsync();

    await using var insert = new NpgsqlCommand(
        "INSERT INTO cookbook_retry_transactions (id, value) " +
        "VALUES ($1, $2)", conn)
    {
        Parameters = { new() { Value = 1 }, new() { Value = 0 } },
    };
    await insert.ExecuteNonQueryAsync();
}

// dropTable drops the shared counter table if it exists.
func dropTable(ctx context.Context, conn *pgx.Conn) error {
    _, err := conn.Exec(ctx, "DROP TABLE IF EXISTS cookbook_retry_transactions")
    return err
}

// createTable creates the shared counter table, with a single row set to
// zero. The table must be visible to a second connection, so it cannot be
// temporary; the name is specific to this recipe, and any earlier copy is
// dropped first.
func createTable(ctx context.Context, conn *pgx.Conn) error {
    if err := dropTable(ctx, conn); err != nil {
        return err
    }
    _, err := conn.Exec(ctx,
        "CREATE TABLE cookbook_retry_transactions "+
            "(id int PRIMARY KEY, value int NOT NULL)")
    if err != nil {
        return err
    }
    _, err = conn.Exec(ctx,
        "INSERT INTO cookbook_retry_transactions (id, value) VALUES ($1, $2)",
        1, 0)
    return err
}

/// Drops the shared counter table if it exists.
fn drop_table(client: &mut Client) -> Result<(), postgres::Error> {
    client.batch_execute("DROP TABLE IF EXISTS cookbook_retry_transactions")
}

/// Creates the shared counter table, with a single row set to zero. The
/// table must be visible to a second connection, so it cannot be
/// temporary; the name is specific to this recipe, and any earlier copy is
/// dropped first.
fn create_table(client: &mut Client) -> Result<(), postgres::Error> {
    drop_table(client)?;
    client.batch_execute(
        "CREATE TABLE cookbook_retry_transactions \
         (id int PRIMARY KEY, value int NOT NULL)",
    )?;
    client.execute(
        "INSERT INTO cookbook_retry_transactions (id, value) VALUES ($1, $2)",
        &[&1_i32, &0_i32],
    )?;
    Ok(())
}

# Drops the shared counter table if it exists.
def drop_table(conn)
  conn.exec("DROP TABLE IF EXISTS cookbook_retry_transactions")
end

# Creates the shared counter table, with a single row set to zero. The
# table must be visible to a second connection, so it cannot be
# temporary; the name is specific to this recipe, and any earlier copy is
# dropped first.
def create_table(conn)
  drop_table(conn)
  conn.exec("CREATE TABLE cookbook_retry_transactions " \
            "(id int PRIMARY KEY, value int NOT NULL)")
  conn.exec_params(
    "INSERT INTO cookbook_retry_transactions (id, value) VALUES ($1, $2)",
    [1, 0]
  )
end

/*
 * Run a statement with optional text parameters, and return true if it
 * succeeded. The five-character SQLSTATE code of a failure is copied into
 * sqlstate, which is empty if the statement succeeded or the failure has
 * no code (for example, a lost connection). If value is not NULL, the
 * first column of the first row is copied into it, up to size bytes.
 */
static bool
run_sql(PGconn *conn, const char *sql, int nparams, const char *const *params,
        char *sqlstate, char *value, size_t size)
{
    PGresult *res = PQexecParams(conn, sql, nparams, NULL, params, NULL,
                                 NULL, 0);
    ExecStatusType status = PQresultStatus(res);
    bool ok = status == PGRES_COMMAND_OK || status == PGRES_TUPLES_OK;
    const char *code = PQresultErrorField(res, PG_DIAG_SQLSTATE);

    snprintf(sqlstate, 6, "%s", code != NULL ? code : "");
    if (ok && value != NULL)
        snprintf(value, size, "%s", PQgetvalue(res, 0, 0));

    /* PQclear() frees the result, and accepts NULL. */
    PQclear(res);
    return ok;
}

/* Drop the shared counter table if it exists. */
static bool
drop_table(PGconn *conn)
{
    char sqlstate[6];

    return run_sql(conn, "DROP TABLE IF EXISTS cookbook_retry_transactions",
                   0, NULL, sqlstate, NULL, 0);
}

/*
 * Create the shared counter table, with a single row set to zero. The
 * table must be visible to a second connection, so it cannot be
 * temporary; the name is specific to this recipe, and any earlier copy is
 * dropped first.
 */
static bool
create_table(PGconn *conn)
{
    const char *params[] = {"1", "0"};
    char sqlstate[6];

    return drop_table(conn) &&
        run_sql(conn,
                "CREATE TABLE cookbook_retry_transactions "
                "(id int PRIMARY KEY, value int NOT NULL)",
                0, NULL, sqlstate, NULL, 0) &&
        run_sql(conn,
                "INSERT INTO cookbook_retry_transactions (id, value) "
                "VALUES ($1, $2)",
                2, params, sqlstate, NULL, 0);
}

def increment_counter(conn: psycopg.Connection, attempt: int) -> None:
    """Reads the counter and writes back the value plus one."""
    value = conn.execute(
        "SELECT value FROM cookbook_retry_transactions WHERE id = %s",
        (1,),
    ).fetchone()[0]

    # Simulate another client by updating the row from a second
    # connection, after this transaction has read the row but before it
    # writes. This happens on the first attempt only.
    if attempt == 1:
        with connect() as other:
            other.execute(
                "UPDATE cookbook_retry_transactions "
                "SET value = value + 1 WHERE id = %s",
                (1,),
            )

    # Under REPEATABLE READ, updating a row that another transaction
    # changed and committed since this transaction's snapshot was taken
    # fails with SQLSTATE 40001.
    conn.execute(
        "UPDATE cookbook_retry_transactions SET value = %s WHERE id = %s",
        (value + 1, 1),
    )


def read_counter(conn: psycopg.Connection) -> int:
    """Returns the current value of the counter."""
    row = conn.execute(
        "SELECT value FROM cookbook_retry_transactions WHERE id = %s",
        (1,),
    ).fetchone()
    return row[0]

// Reads the counter and writes back the value plus one.
async function incrementCounter(
  client: pg.Client,
  attempt: number,
): Promise<void> {
  const result = await client.query<{ value: number }>(
    "SELECT value FROM cookbook_retry_transactions WHERE id = $1",
    [1],
  );
  const value = result.rows[0].value;

  // Simulate another client by updating the row from a second
  // connection, after this transaction has read the row but before it
  // writes. This happens on the first attempt only.
  if (attempt === 1) {
    const other = await connect();
    try {
      await other.query(
        "UPDATE cookbook_retry_transactions SET value = value + 1 " +
          "WHERE id = $1",
        [1],
      );
    } finally {
      await other.end();
    }
  }

  // Under REPEATABLE READ, updating a row that another transaction
  // changed and committed since this transaction's snapshot was taken
  // fails with SQLSTATE 40001.
  await client.query(
    "UPDATE cookbook_retry_transactions SET value = $1 WHERE id = $2",
    [value + 1, 1],
  );
}

// Returns the current value of the counter.
async function readCounter(client: pg.Client): Promise<number> {
  const result = await client.query<{ value: number }>(
    "SELECT value FROM cookbook_retry_transactions WHERE id = $1",
    [1],
  );
  return result.rows[0].value;
}

/** The work done inside one transaction. */
interface Work {
    void run(Connection conn, int attempt) throws SQLException;
}

/** Reads the counter and writes back the value plus one. */
static void incrementCounter(Connection conn, int attempt)
        throws SQLException {
    int value = readCounter(conn);

    // Simulate another client by updating the row from a second
    // connection, after this transaction has read the row but before
    // it writes. This happens on the first attempt only.
    if (attempt == 1) {
        try (Connection other = connect();
             PreparedStatement stmt = other.prepareStatement(
                "UPDATE cookbook_retry_transactions "
                + "SET value = value + 1 WHERE id = ?")) {
            stmt.setInt(1, 1);
            stmt.executeUpdate();
        }
    }

    // Under REPEATABLE READ, updating a row that another transaction
    // changed and committed since this transaction's snapshot was
    // taken fails with SQLSTATE 40001.
    try (PreparedStatement stmt = conn.prepareStatement(
            "UPDATE cookbook_retry_transactions SET value = ? "
            + "WHERE id = ?")) {
        stmt.setInt(1, value + 1);
        stmt.setInt(2, 1);
        stmt.executeUpdate();
    }
}

/** Returns the current value of the counter. */
static int readCounter(Connection conn) throws SQLException {
    try (PreparedStatement stmt = conn.prepareStatement(
            "SELECT value FROM cookbook_retry_transactions "
            + "WHERE id = ?")) {
        stmt.setInt(1, 1);
        try (ResultSet rs = stmt.executeQuery()) {
            rs.next();
            return rs.getInt(1);
        }
    }
}

// Reads the counter and writes back the value plus one.
static async Task IncrementCounterAsync(
    NpgsqlDataSource dataSource, NpgsqlConnection conn, int attempt)
{
    var value = await ReadCounterAsync(conn);

    // Simulate another client by updating the row from a second
    // connection, after this transaction has read the row but before it
    // writes. This happens on the first attempt only.
    if (attempt == 1)
    {
        await using var other = await dataSource.OpenConnectionAsync();
        await using var cmd = new NpgsqlCommand(
            "UPDATE cookbook_retry_transactions SET value = value + 1 " +
            "WHERE id = $1", other)
        {
            Parameters = { new() { Value = 1 } },
        };
        await cmd.ExecuteNonQueryAsync();
    }

    // Under REPEATABLE READ, updating a row that another transaction
    // changed and committed since this transaction's snapshot was taken
    // fails with SQLSTATE 40001.
    await using var update = new NpgsqlCommand(
        "UPDATE cookbook_retry_transactions SET value = $1 WHERE id = $2",
        conn)
    {
        Parameters = { new() { Value = value + 1 }, new() { Value = 1 } },
    };
    await update.ExecuteNonQueryAsync();
}

// Returns the current value of the counter.
static async Task<int> ReadCounterAsync(NpgsqlConnection conn)
{
    await using var cmd = new NpgsqlCommand(
        "SELECT value FROM cookbook_retry_transactions WHERE id = $1", conn)
    {
        Parameters = { new() { Value = 1 } },
    };
    return (int)(await cmd.ExecuteScalarAsync())!;
}

// incrementCounter reads the counter and writes back the value plus one.
func incrementCounter(ctx context.Context, tx pgx.Tx, attempt int) error {
    var value int
    err := tx.QueryRow(ctx,
        "SELECT value FROM cookbook_retry_transactions WHERE id = $1",
        1).Scan(&value)
    if err != nil {
        return err
    }

    // Simulate another client by updating the row from a second
    // connection, after this transaction has read the row but before it
    // writes. This happens on the first attempt only.
    if attempt == 1 {
        if err := updateFromOtherConnection(ctx); err != nil {
            return err
        }
    }

    // Under REPEATABLE READ, updating a row that another transaction
    // changed and committed since this transaction's snapshot was taken
    // fails with SQLSTATE 40001.
    _, err = tx.Exec(ctx,
        "UPDATE cookbook_retry_transactions SET value = $1 WHERE id = $2",
        value+1, 1)
    return err
}

// updateFromOtherConnection increments the counter on a second connection,
// which commits straight away because it has no open transaction.
func updateFromOtherConnection(ctx context.Context) error {
    other, err := connect(ctx)
    if err != nil {
        return err
    }
    defer other.Close(ctx)

    _, err = other.Exec(ctx,
        "UPDATE cookbook_retry_transactions SET value = value + 1 "+
            "WHERE id = $1", 1)
    return err
}

// readCounter returns the current value of the counter.
func readCounter(ctx context.Context, conn *pgx.Conn) (int, error) {
    var value int
    err := conn.QueryRow(ctx,
        "SELECT value FROM cookbook_retry_transactions WHERE id = $1",
        1).Scan(&value)
    return value, err
}

/// Reads the counter and writes back the value plus one.
fn increment_counter(
    tx: &mut Transaction,
    attempt: u32,
) -> Result<(), postgres::Error> {
    let row = tx.query_one(
        "SELECT value FROM cookbook_retry_transactions WHERE id = $1",
        &[&1_i32],
    )?;
    let value: i32 = row.get(0);

    // Simulate another client by updating the row from a second
    // connection, after this transaction has read the row but before it
    // writes. This happens on the first attempt only.
    if attempt == 1 {
        let mut other = connect()?;
        other.execute(
            "UPDATE cookbook_retry_transactions SET value = value + 1 \
             WHERE id = $1",
            &[&1_i32],
        )?;
    }

    // Under REPEATABLE READ, updating a row that another transaction
    // changed and committed since this transaction's snapshot was taken
    // fails with SQLSTATE 40001.
    tx.execute(
        "UPDATE cookbook_retry_transactions SET value = $1 WHERE id = $2",
        &[&(value + 1), &1_i32],
    )?;
    Ok(())
}

/// Returns the current value of the counter.
fn read_counter(client: &mut Client) -> Result<i32, postgres::Error> {
    let row = client.query_one(
        "SELECT value FROM cookbook_retry_transactions WHERE id = $1",
        &[&1_i32],
    )?;
    Ok(row.get(0))
}

# Reads the counter and writes back the value plus one.
def increment_counter(conn, attempt)
  value = read_counter(conn)

  # Simulate another client by updating the row from a second connection,
  # after this transaction has read the row but before it writes. This
  # happens on the first attempt only.
  if attempt == 1
    other = connect
    begin
      other.exec_params(
        "UPDATE cookbook_retry_transactions SET value = value + 1 " \
        "WHERE id = $1",
        [1]
      )
    ensure
      other.close
    end
  end

  # Under REPEATABLE READ, updating a row that another transaction
  # changed and committed since this transaction's snapshot was taken
  # fails with SQLSTATE 40001.
  conn.exec_params(
    "UPDATE cookbook_retry_transactions SET value = $1 WHERE id = $2",
    [value + 1, 1]
  )
end

# Returns the current value of the counter. Results arrive as strings, so
# the value is converted to an Integer.
def read_counter(conn)
  conn.exec_params(
    "SELECT value FROM cookbook_retry_transactions WHERE id = $1", [1]
  ).getvalue(0, 0).to_i
end

/* Read the current value of the counter into *counter. */
static bool
read_counter(PGconn *conn, int *counter, char *sqlstate)
{
    const char *params[] = {"1"};
    char text[32];

    if (!run_sql(conn,
                 "SELECT value FROM cookbook_retry_transactions "
                 "WHERE id = $1",
                 1, params, sqlstate, text, sizeof(text)))
        return false;

    /* Results arrive as text, so convert the value to a number. */
    *counter = atoi(text);
    return true;
}

/*
 * Simulate another client by incrementing the counter on a second
 * connection, which commits straight away because it has no open
 * transaction. On failure, the SQLSTATE code and error message of the
 * second connection are copied into the caller's buffers.
 */
static bool
update_from_other_connection(char *sqlstate, char *message, size_t size)
{
    const char *params[] = {"1"};
    PGconn *other = connect_to_server();
    bool ok = PQstatus(other) == CONNECTION_OK &&
        run_sql(other,
                "UPDATE cookbook_retry_transactions SET value = value + 1 "
                "WHERE id = $1",
                1, params, sqlstate, NULL, 0);

    if (!ok)
        snprintf(message, size, "%s", PQerrorMessage(other));
    PQfinish(other);
    return ok;
}

/*
 * Read the counter and write back the value plus one. A failure on conn
 * leaves message empty, because PQerrorMessage(conn) describes it.
 */
static bool
increment_counter(PGconn *conn, int attempt, char *sqlstate,
                  char *message, size_t size)
{
    int counter;
    char next[32];
    const char *params[2];

    if (!read_counter(conn, &counter, sqlstate))
        return false;

    /*
     * Simulate another client by updating the row from a second
     * connection, after this transaction has read the row but before it
     * writes. This happens on the first attempt only.
     */
    if (attempt == 1 &&
        !update_from_other_connection(sqlstate, message, size))
        return false;

    /*
     * Under REPEATABLE READ, updating a row that another transaction
     * changed and committed since this transaction's snapshot was taken
     * fails with SQLSTATE 40001.
     */
    snprintf(next, sizeof(next), "%d", counter + 1);
    params[0] = next;
    params[1] = "1";
    return run_sql(conn,
                   "UPDATE cookbook_retry_transactions SET value = $1 "
                   "WHERE id = $2",
                   2, params, sqlstate, NULL, 0);
}

def run_with_retry(conn: psycopg.Connection, work) -> None:
    """Runs work(conn, attempt) in a REPEATABLE READ transaction, and runs
    it again if the server reports a serialisation failure or a deadlock.
    """
    conn.isolation_level = psycopg.IsolationLevel.REPEATABLE_READ

    for attempt in range(1, MAX_ATTEMPTS + 1):
        try:
            # The transaction block commits when it exits normally, and
            # rolls back if an exception escapes it.
            with conn.transaction():
                work(conn, attempt)
        except psycopg.Error as e:
            # Any other error, or running out of attempts, is a real
            # failure that the caller must handle.
            if e.sqlstate not in RETRYABLE or attempt == MAX_ATTEMPTS:
                raise
            print(f"Attempt {attempt}: serialisation failure, retrying")
            # A short, growing pause gives the competing transaction time
            # to finish.
            time.sleep(0.05 * attempt)
        else:
            print(f"Attempt {attempt}: committed")
            return

// Runs work(client, attempt) in a REPEATABLE READ transaction, and runs
// it again if the server reports a serialisation failure or a deadlock.
async function runWithRetry(
  client: pg.Client,
  work: (client: pg.Client, attempt: number) => Promise<void>,
): Promise<void> {
  for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
    try {
      await client.query("BEGIN ISOLATION LEVEL REPEATABLE READ");
      await work(client, attempt);
      await client.query("COMMIT");
      console.log(`Attempt ${attempt}: committed`);
      return;
    } catch (err) {
      // End the failed transaction so that the session can start another.
      // If the connection is lost the rollback fails too, and that error
      // must not hide the original one.
      await client.query("ROLLBACK").catch(() => undefined);

      // Any other error, or running out of attempts, is a real failure
      // that the caller must handle.
      const code = (err as pg.DatabaseError).code;
      if (code === undefined || !RETRYABLE.has(code) ||
          attempt === MAX_ATTEMPTS) {
        throw err;
      }
      console.log(`Attempt ${attempt}: serialisation failure, retrying`);
      // A short, growing pause gives the competing transaction time to
      // finish.
      await sleep(50 * attempt);
    }
  }
}

/**
 * Runs the work in a REPEATABLE READ transaction, and runs it again
 * if the server reports a serialisation failure or a deadlock.
 */
static void runWithRetry(Connection conn, Work work)
        throws SQLException, InterruptedException {
    // Turning autocommit off makes the driver start a transaction
    // with the first statement, and end it at commit() or rollback().
    conn.setAutoCommit(false);
    conn.setTransactionIsolation(Connection.TRANSACTION_REPEATABLE_READ);
    try {
        for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
            try {
                work.run(conn, attempt);
                conn.commit();
                System.out.println("Attempt " + attempt + ": committed");
                return;
            } catch (SQLException e) {
                // End the failed transaction so that the session can
                // start another. If the connection is lost the
                // rollback fails too, and that error must not hide
                // the original one, so it is attached to it instead.
                try {
                    conn.rollback();
                } catch (SQLException rollbackError) {
                    e.addSuppressed(rollbackError);
                }

                // Any other error, or running out of attempts, is a
                // real failure that the caller must handle.
                if (!RETRYABLE.contains(e.getSQLState())
                        || attempt == MAX_ATTEMPTS) {
                    throw e;
                }
                System.out.println("Attempt " + attempt
                        + ": serialisation failure, retrying");
                // A short, growing pause gives the competing
                // transaction time to finish.
                Thread.sleep(50L * attempt);
            }
        }
    } finally {
        // Restoring autocommit fails only if the connection is already
        // broken. That error must not replace the one in flight, and
        // closing the connection discards the session anyway.
        try {
            conn.setAutoCommit(true);
        } catch (SQLException ignored) {
            // Nothing useful can be done with a broken connection.
        }
    }
}

// Runs work(conn, attempt) in a REPEATABLE READ transaction, and runs it
// again if the server reports a serialisation failure or a deadlock.
static async Task RunWithRetryAsync(
    NpgsqlConnection conn, Func<NpgsqlConnection, int, Task> work)
{
    for (var attempt = 1; attempt <= MaxAttempts; attempt++)
    {
        try
        {
            // Disposing the transaction without committing rolls it
            // back, so an error from the work leaves nothing behind.
            await using var tx = await conn.BeginTransactionAsync(
                IsolationLevel.RepeatableRead);
            await work(conn, attempt);
            await tx.CommitAsync();

            Console.WriteLine($"Attempt {attempt}: committed");
            return;
        }
        // Any other error, or running out of attempts, is a real failure
        // that the caller must handle, so the filter lets it through.
        catch (PostgresException e)
            when (IsRetryable(e.SqlState) && attempt < MaxAttempts)
        {
            Console.WriteLine(
                $"Attempt {attempt}: serialisation failure, retrying");
            // A short, growing pause gives the competing transaction time
            // to finish.
            await Task.Delay(50 * attempt);
        }
    }
}

// runOnce runs work in a single REPEATABLE READ transaction.
func runOnce(ctx context.Context, conn *pgx.Conn, attempt int,
    work func(context.Context, pgx.Tx, int) error) error {
    tx, err := conn.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.RepeatableRead})
    if err != nil {
        return err
    }
    // The deferred rollback undoes the work if it fails; after a
    // successful commit it does nothing.
    defer tx.Rollback(ctx)

    if err := work(ctx, tx, attempt); err != nil {
        return err
    }
    return tx.Commit(ctx)
}

// runWithRetry runs work in a REPEATABLE READ transaction, and runs it
// again if the server reports a serialisation failure or a deadlock.
func runWithRetry(ctx context.Context, conn *pgx.Conn,
    work func(context.Context, pgx.Tx, int) error) error {
    for attempt := 1; attempt <= maxAttempts; attempt++ {
        err := runOnce(ctx, conn, attempt, work)
        if err == nil {
            fmt.Printf("Attempt %d: committed\n", attempt)
            return nil
        }

        // Any other error, or running out of attempts, is a real failure
        // that the caller must handle.
        if !isRetryable(err) || attempt == maxAttempts {
            return err
        }
        fmt.Printf("Attempt %d: serialisation failure, retrying\n", attempt)
        // A short, growing pause gives the competing transaction time to
        // finish.
        time.Sleep(time.Duration(attempt) * 50 * time.Millisecond)
    }
    return nil
}

/// The type of a function that does the work of one transaction.
type Work = fn(&mut Transaction, u32) -> Result<(), postgres::Error>;

/// Runs the work in a single REPEATABLE READ transaction. Dropping the
/// transaction without committing rolls it back, so an error from the
/// work or from the commit leaves nothing behind.
fn run_once(
    client: &mut Client,
    attempt: u32,
    work: Work,
) -> Result<(), postgres::Error> {
    let mut tx = client
        .build_transaction()
        .isolation_level(IsolationLevel::RepeatableRead)
        .start()?;
    work(&mut tx, attempt)?;
    tx.commit()
}

/// Runs the work in a REPEATABLE READ transaction, and runs it again if
/// the server reports a serialisation failure or a deadlock.
fn run_with_retry(client: &mut Client, work: Work) -> Result<(), postgres::Error> {
    for attempt in 1..=MAX_ATTEMPTS {
        match run_once(client, attempt, work) {
            Ok(()) => {
                println!("Attempt {attempt}: committed");
                return Ok(());
            }
            // Any other error, or running out of attempts, is a real
            // failure that the caller must handle.
            Err(e) if !is_retryable(&e) || attempt == MAX_ATTEMPTS => {
                return Err(e);
            }
            Err(_) => {
                println!("Attempt {attempt}: serialisation failure, retrying");
                // A short, growing pause gives the competing transaction
                // time to finish.
                sleep(Duration::from_millis(50 * u64::from(attempt)));
            }
        }
    }
    unreachable!("the final attempt returns above")
}

# Returns the SQLSTATE code of an error reported by the server, or nil for
# errors such as a lost connection that have no result.
def sqlstate(error)
  error.result&.error_field(PG::PG_DIAG_SQLSTATE)
end

# Yields the connection and the attempt number inside a REPEATABLE READ
# transaction, and runs the block again if the server reports a
# serialisation failure or a deadlock.
def run_with_retry(conn)
  (1..MAX_ATTEMPTS).each do |attempt|
    begin
      conn.exec("BEGIN ISOLATION LEVEL REPEATABLE READ")
      yield conn, attempt
      conn.exec("COMMIT")
      puts "Attempt #{attempt}: committed"
      return
    rescue PG::Error => e
      # End the failed transaction so that the session can start another.
      # If the connection is lost the rollback fails too, and that error
      # must not hide the original one.
      begin
        conn.exec("ROLLBACK")
      rescue PG::Error
        nil
      end

      # Any other error, or running out of attempts, is a real failure
      # that the caller must handle.
      raise unless RETRYABLE.include?(sqlstate(e)) && attempt < MAX_ATTEMPTS

      puts "Attempt #{attempt}: serialisation failure, retrying"
      # A short, growing pause gives the competing transaction time to
      # finish.
      sleep(0.05 * attempt)
    end
  end
end

/* The type of a function that does the work of one transaction. */
typedef bool (*work_fn)(PGconn *conn, int attempt, char *sqlstate,
                        char *message, size_t size);

/* Sleep for the given number of milliseconds. */
static void
pause_ms(long ms)
{
    struct timespec ts = {ms / 1000, (ms % 1000) * 1000000L};

    nanosleep(&ts, NULL);
}

/*
 * Run work() in a REPEATABLE READ transaction, and run it again if the
 * server reports a serialisation failure or a deadlock. Return true if a
 * transaction committed. On failure, print the saved error message to
 * stderr, because the caller cannot recover it from the connection.
 */
static bool
run_with_retry(PGconn *conn, work_fn work)
{
    char sqlstate[6];
    char message[512];

    for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++)
    {
        bool ok;

        message[0] = '\0';
        ok = run_sql(conn, "BEGIN ISOLATION LEVEL REPEATABLE READ",
                     0, NULL, sqlstate, NULL, 0) &&
            work(conn, attempt, sqlstate, message, sizeof(message)) &&
            run_sql(conn, "COMMIT", 0, NULL, sqlstate, NULL, 0);

        if (ok)
        {
            printf("Attempt %d: committed\n", attempt);
            return true;
        }

        /*
         * End the failed transaction so that the session can start
         * another. The error message is kept, because the ROLLBACK
         * would otherwise replace it; the work may already have supplied
         * a message from another connection.
         */
        char ignored[6];

        if (message[0] == '\0')
            snprintf(message, sizeof(message), "%s", PQerrorMessage(conn));
        run_sql(conn, "ROLLBACK", 0, NULL, ignored, NULL, 0);

        /*
         * Any other error, or running out of attempts, is a real failure
         * that the caller must report.
         */
        if (!is_retryable(sqlstate) || attempt == MAX_ATTEMPTS)
        {
            fprintf(stderr, "Database error: %s", message);
            return false;
        }

        printf("Attempt %d: serialisation failure, retrying\n", attempt);
        /* A short, growing pause gives the competing transaction time. */
        pause_ms(50L * attempt);
    }
    return false;
}

def main() -> int:
    try:
        # The with block closes the connection when it exits, whether or
        # not an error occurred.
        with connect() as conn:
            try:
                create_table(conn)
                run_with_retry(conn, increment_counter)
                print(f"Counter value: {read_counter(conn)}")
            finally:
                # Drop the shared table even if the work failed.
                drop_table(conn)
    # psycopg.Error is the base class for every error the driver raises,
    # including connection failures and errors reported by the server.
    except psycopg.Error as e:
        print(f"Database error: {e}", file=sys.stderr)
        return 1

    return 0


if __name__ == "__main__":
    sys.exit(main())

// Returns a readable description of an error. When a host name resolves
// to several addresses, Node reports a failure to reach all of them as an
// AggregateError, whose own message is empty.
function describe(err: unknown): string {
  if (err instanceof AggregateError) {
    return err.errors.map(describe).join("; ");
  }
  return err instanceof Error ? err.message : String(err);
}

async function main(): Promise<number> {
  let client: pg.Client | undefined;
  try {
    client = await connect();
    try {
      await createTable(client);
      await runWithRetry(client, incrementCounter);
      console.log(`Counter value: ${await readCounter(client)}`);
    } finally {
      // Drop the shared table even if the work failed.
      await dropTable(client);
    }
    return 0;
  } catch (err) {
    // Connection failures and errors reported by the server both arrive
    // here as rejected promises.
    console.error(`Database error: ${describe(err)}`);
    return 1;
  } finally {
    // The finally block runs whether or not an error occurred; client is
    // undefined if the connection attempt failed.
    await client?.end();
  }
}

process.exitCode = await main();