Níveis de isolamento

O código desta página foi executado nas cinco linguagens contra um PostgreSQL 18. A matriz de anomalias do nível Conceito foi gerada por um script em Python que intercala duas conexões passo a passo, no PostgreSQL 18 e no MySQL 8.4.

A regra do plantão em serializable, em cada linguagem

A tabela do teste:

DROP TABLE IF EXISTS plantao;
CREATE TABLE plantao (medico text PRIMARY KEY, de_plantao boolean NOT NULL);
INSERT INTO plantao SELECT 'medico' || g, true FROM generate_series(1, 10) g;

Cada versão tem a mesma função: abre uma transação serializable, conta quem está de plantão, recusa se sobrar só um e, se não, tira o médico do plantão. Quando o banco recusa a transação com o erro 40001, a função espera um pouco e repete do começo, até dez vezes. O programa dispara dez pedidos ao mesmo tempo, um por médico.

import pg from "pg";

const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL, max: 20 });
const espera = (ms: number) => new Promise((r) => setTimeout(r, ms));
let novasTentativas = 0;

// Regra: pelo menos um médico continua de plantão
async function sairDoPlantao(medico: string): Promise<"saiu" | "recusado"> {
  for (let tentativa = 1; ; tentativa++) {
    const conexao = await pool.connect();
    try {
      await conexao.query("BEGIN ISOLATION LEVEL SERIALIZABLE");
      const { rows } = await conexao.query(
        "SELECT count(*)::int AS n FROM plantao WHERE de_plantao");
      if (rows[0].n <= 1) {
        await conexao.query("ROLLBACK");
        return "recusado";
      }
      await conexao.query("UPDATE plantao SET de_plantao = false WHERE medico = $1", [medico]);
      await conexao.query("COMMIT");
      return "saiu";
    } catch (erro: any) {
      await conexao.query("ROLLBACK");
      // 40001: o banco desfez a transação por conflito; repetir é seguro
      if (erro.code !== "40001" || tentativa === 10) throw erro;
      novasTentativas++;
      await espera(Math.random() * 20 * tentativa); // espera crescente e aleatória
    } finally {
      conexao.release();
    }
  }
}

const medicos = Array.from({ length: 10 }, (_, i) => `medico${i + 1}`);
const resultados = await Promise.all(medicos.map(sairDoPlantao));
const { rows } = await pool.query("SELECT count(*)::int AS n FROM plantao WHERE de_plantao");
console.log({
  saiu: resultados.filter((r) => r === "saiu").length,
  recusado: resultados.filter((r) => r === "recusado").length,
  novasTentativas,
  continuamDePlantao: rows[0].n,
});
await pool.end();

import os
import random
import time
from concurrent.futures import ThreadPoolExecutor

import psycopg
from psycopg import errors
from psycopg_pool import ConnectionPool

pool = ConnectionPool(os.environ["DATABASE_URL"], min_size=10, max_size=20, open=True)
novas_tentativas = 0


# Regra: pelo menos um médico continua de plantão
def sair_do_plantao(medico: str) -> str:
    global novas_tentativas
    for tentativa in range(1, 11):
        try:
            with pool.connection() as conexao:
                conexao.isolation_level = psycopg.IsolationLevel.SERIALIZABLE
                with conexao.transaction():
                    (n,) = conexao.execute(
                        "SELECT count(*) FROM plantao WHERE de_plantao").fetchone()
                    if n <= 1:
                        return "recusado"  # sair do bloco desfaz a transação de leitura
                    conexao.execute(
                        "UPDATE plantao SET de_plantao = false WHERE medico = %s", (medico,))
                return "saiu"
        except errors.SerializationFailure:
            # 40001: o banco desfez a transação por conflito; repetir é seguro
            if tentativa == 10:
                raise
            novas_tentativas += 1
            time.sleep(random.random() * 0.02 * tentativa)  # espera crescente e aleatória
    raise RuntimeError("inalcançável")


medicos = [f"medico{i}" for i in range(1, 11)]
with ThreadPoolExecutor(max_workers=10) as executor:
    resultados = list(executor.map(sair_do_plantao, medicos))
with pool.connection() as conexao:
    (restantes,) = conexao.execute("SELECT count(*) FROM plantao WHERE de_plantao").fetchone()
pool.close()
print({"saiu": resultados.count("saiu"), "recusado": resultados.count("recusado"),
       "novas_tentativas": novas_tentativas, "continuam_de_plantao": restantes})

package main

import (
	"context"
	"errors"
	"fmt"
	"math/rand/v2"
	"os"
	"sync"
	"sync/atomic"
	"time"

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

var novasTentativas atomic.Int64

// sairDoPlantao aplica a regra: pelo menos um médico continua de plantão
func sairDoPlantao(ctx context.Context, db *pgxpool.Pool, medico string) (string, error) {
	for tentativa := 1; ; tentativa++ {
		resultado, err := tentar(ctx, db, medico)
		var pgErr *pgconn.PgError
		// 40001: o banco desfez a transação por conflito; repetir é seguro
		if errors.As(err, &pgErr) && pgErr.Code == "40001" && tentativa < 10 {
			novasTentativas.Add(1)
			espera := time.Duration(rand.Float64()*20*float64(tentativa)) * time.Millisecond
			time.Sleep(espera) // espera crescente e aleatória
			continue
		}
		return resultado, err
	}
}

func tentar(ctx context.Context, db *pgxpool.Pool, medico string) (string, error) {
	tx, err := db.BeginTx(ctx, pgx.TxOptions{IsoLevel: pgx.Serializable})
	if err != nil {
		return "", err
	}
	defer tx.Rollback(ctx) // sem efeito depois do Commit

	var n int
	if err := tx.QueryRow(ctx, "SELECT count(*) FROM plantao WHERE de_plantao").Scan(&n); err != nil {
		return "", err
	}
	if n <= 1 {
		return "recusado", nil
	}
	_, err = tx.Exec(ctx, "UPDATE plantao SET de_plantao = false WHERE medico = $1", medico)
	if err != nil {
		return "", err
	}
	return "saiu", tx.Commit(ctx)
}

func main() {
	ctx := context.Background()
	db, err := pgxpool.New(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		panic(err)
	}
	defer db.Close()

	var wg sync.WaitGroup
	resultados := make([]string, 10)
	for i := range 10 {
		wg.Go(func() {
			r, err := sairDoPlantao(ctx, db, fmt.Sprintf("medico%d", i+1))
			if err != nil {
				r = err.Error()
			}
			resultados[i] = r
		})
	}
	wg.Wait()
	var restantes int
	db.QueryRow(ctx, "SELECT count(*) FROM plantao WHERE de_plantao").Scan(&restantes)
	fmt.Println(resultados, "novas tentativas:", novasTentativas.Load(),
		"continuam de plantão:", restantes)
}

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicInteger;

public class Plantao {
    static final String URL = System.getenv("JDBC_URL");
    static final AtomicInteger novasTentativas = new AtomicInteger();

    // Regra: pelo menos um médico continua de plantão
    static String sairDoPlantao(String medico) throws Exception {
        for (int tentativa = 1; ; tentativa++) {
            try (Connection c = DriverManager.getConnection(URL)) {
                c.setTransactionIsolation(Connection.TRANSACTION_SERIALIZABLE);
                c.setAutoCommit(false);
                try {
                    String resultado = tentar(c, medico);
                    c.commit();
                    return resultado;
                } catch (SQLException erro) {
                    c.rollback();
                    // 40001: o banco desfez a transação por conflito; repetir é seguro
                    if (!"40001".equals(erro.getSQLState()) || tentativa == 10) throw erro;
                }
            }
            novasTentativas.incrementAndGet();
            long espera = (long) (ThreadLocalRandom.current().nextDouble() * 20 * tentativa);
            Thread.sleep(espera); // espera crescente e aleatória
        }
    }

    static String tentar(Connection c, String medico) throws SQLException {
        try (var contar = c.createStatement();
             var linhas = contar.executeQuery("SELECT count(*) FROM plantao WHERE de_plantao")) {
            linhas.next();
            if (linhas.getInt(1) <= 1) return "recusado";
        }
        var sql = "UPDATE plantao SET de_plantao = false WHERE medico = ?";
        try (var sair = c.prepareStatement(sql)) {
            sair.setString(1, medico);
            sair.executeUpdate();
        }
        return "saiu";
    }

    public static void main(String[] args) throws Exception {
        var resultados = new ArrayList<Future<String>>();
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            for (int i = 1; i <= 10; i++) {
                String medico = "medico" + i;
                resultados.add(executor.submit(() -> sairDoPlantao(medico)));
            }
        }
        var saida = new ArrayList<String>();
        for (var r : resultados) saida.add(r.get());
        try (var c = DriverManager.getConnection(URL); var s = c.createStatement();
             var l = s.executeQuery("SELECT count(*) FROM plantao WHERE de_plantao")) {
            l.next();
            System.out.println(saida + " novas tentativas: " + novasTentativas
                + " continuam de plantão: " + l.getInt(1));
        }
    }
}

using System.Data;
using Npgsql;

await using var db = NpgsqlDataSource.Create(Environment.GetEnvironmentVariable("DATABASE_URL")!);
var novasTentativas = 0;

// Regra: pelo menos um médico continua de plantão
async Task<string> SairDoPlantao(string medico)
{
    for (var tentativa = 1; ; tentativa++)
    {
        await using var conexao = await db.OpenConnectionAsync();
        await using var tx = await conexao.BeginTransactionAsync(IsolationLevel.Serializable);
        try
        {
            await using var contar = new NpgsqlCommand(
                "SELECT count(*) FROM plantao WHERE de_plantao", conexao, tx);
            if ((long)(await contar.ExecuteScalarAsync())! <= 1) return "recusado";

            await using var sair = new NpgsqlCommand(
                "UPDATE plantao SET de_plantao = false WHERE medico = $1", conexao, tx);
            sair.Parameters.Add(new NpgsqlParameter { Value = medico });
            await sair.ExecuteNonQueryAsync();
            await tx.CommitAsync();
            return "saiu";
        }
        // 40001: o banco desfez a transação por conflito; repetir é seguro
        catch (PostgresException erro) when (
            erro.SqlState == PostgresErrorCodes.SerializationFailure && tentativa < 10)
        {
            // Sem RollbackAsync: se o erro veio do Commit, a transação já terminou e o
            // Npgsql lança InvalidOperationException; o await using desfaz o que restar
            Interlocked.Increment(ref novasTentativas);
            // espera crescente e aleatória
            await Task.Delay((int)(Random.Shared.NextDouble() * 20 * tentativa));
        }
    }
}

var resultados = await Task.WhenAll(
    Enumerable.Range(1, 10).Select(i => SairDoPlantao($"medico{i}")));
await using var total = db.CreateCommand("SELECT count(*) FROM plantao WHERE de_plantao");
Console.WriteLine($"{string.Join(", ", resultados)} | novas tentativas: {novasTentativas} | " +
                  $"continuam de plantão: {await total.ExecuteScalarAsync()}");

O resultado, nas cinco linguagens

	Saíram	Recusados	Novas tentativas	Continuam de plantão
TypeScript	9	1	12	1
Python	9	1	10	1
Go	9	1	20	1
Java	9	1	18	1
C#	9	1	12 e 13, em duas rodadas	1
TypeScript em read committed	10	0	0	0

A última linha é a mesma função TypeScript, trocando só o nível para read committed: os dez médicos saíram, e a regra foi violada sem nenhum erro. O número de novas tentativas varia de uma execução para outra, porque depende da ordem em que as transações se cruzam. O resultado final não varia.

O que muda de uma linguagem para outra

	Como pedir serializable	Como reconhecer a recusa
TypeScript	`BEGIN ISOLATION LEVEL SERIALIZABLE`	`erro.code === "40001"`
Python	`conexao.isolation_level = IsolationLevel.SERIALIZABLE`	`errors.SerializationFailure`
Go	`pgx.TxOptions{IsoLevel: pgx.Serializable}`	`*pgconn.PgError` com `Code == "40001"`
Java	`setTransactionIsolation(TRANSACTION_SERIALIZABLE)`	`getSQLState()` igual a `"40001"`
C#	`BeginTransactionAsync(IsolationLevel.Serializable)`	`PostgresErrorCodes.SerializationFailure`

A recusa pode chegar em qualquer comando da transação, inclusive no COMMIT. Por isso o laço de repetição envolve a transação inteira, e não um comando.

Em frameworks, a repetição costuma ser um decorador ou uma configuração: @Retryable no Spring, políticas de repetição do EF Core (EnableRetryOnFailure), ou uma função auxiliar como a das versões acima. O cuidado é o mesmo: repetir a transação inteira e só nos códigos de conflito.

Com MySQL

O mesmo código funciona no MySQL trocando os códigos: repita nos erros 1213 (impasse) e 1205 (espera por trava esgotada). No teste da matriz, o serializable do MySQL resolveu a escrita enviesada com um impasse: as duas transações travaram as linhas lidas, cada uma esperou a outra, e o banco escolheu uma para desfazer.

Lista de revisão

Nível

O time sabe qual é o nível padrão do banco em uso?
As operações que leem, calculam e gravam têm gravação condicional, FOR UPDATE ou versão?
As regras sobre várias linhas usam serializable, uma trava num registro comum ou uma restrição?

Repetição

A transação inteira é repetida, desde as leituras?
Só os códigos de conflito (40001 e 40P01 no PostgreSQL, 1213 e 1205 no MySQL) são repetidos?
Há limite de tentativas e espera crescente com variação aleatória?
O código não chama ROLLBACK numa transação que o driver já encerrou?

Escopo

As transações são curtas?
E-mails, filas e chamadas a outras APIs acontecem depois do COMMIT, e não dentro da transação?
Há um teste com transações simultâneas para cada regra sensível?

A regra do plantão em serializable, em cada linguagem

O resultado, nas cinco linguagens

O que muda de uma linguagem para outra

Com MySQL

Lista de revisão

Prompts para aprofundar

Reproduzir as anomalias no meu banco

Proteger uma regra com serializable

Revisar o tratamento de concorrência