Migrações de esquema

import pg from "pg";

const espera = (ms: number) => new Promise((r) => setTimeout(r, ms));

// Aplica um comando de migração sem formar fila: desiste rápido e tenta de novo
async function aplicar(sql: string, tentativas = 10): Promise<void> {
  const cliente = new pg.Client({ connectionString: process.env.DATABASE_URL });
  await cliente.connect();
  try {
    await cliente.query("SET lock_timeout = '1s'"); // não espera mais que 1 s pela trava
    for (let tentativa = 1; ; tentativa++) {
      try {
        await cliente.query(sql);
        console.log(`aplicada na tentativa ${tentativa}`);
        return;
      } catch (erro: any) {
        // 55P03: a trava não veio a tempo; quem esperava atrás da migração foi liberado
        if (erro.code !== "55P03" || tentativa === tentativas) throw erro;
        console.log(`tentativa ${tentativa}: tabela ocupada, tentando de novo`);
        await espera(500 * tentativa + Math.random() * 500);
      }
    }
  } finally {
    await cliente.end();
  }
}

await aplicar("ALTER TABLE pedidos ADD COLUMN IF NOT EXISTS canal text");

import os
import random
import time

import psycopg
from psycopg import errors


def aplicar(sql: str, tentativas: int = 10) -> None:
    """Aplica um comando de migração sem formar fila: desiste rápido e tenta de novo."""
    with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as conexao:
        conexao.execute("SET lock_timeout = '1s'")  # não espera mais que 1 s pela trava
        for tentativa in range(1, tentativas + 1):
            try:
                conexao.execute(sql)
                print(f"aplicada na tentativa {tentativa}")
                return
            except errors.LockNotAvailable:
                # 55P03: a trava não veio a tempo; quem esperava atrás da migração foi liberado
                if tentativa == tentativas:
                    raise
                print(f"tentativa {tentativa}: tabela ocupada, tentando de novo")
                time.sleep(0.5 * tentativa + random.random() * 0.5)


aplicar("ALTER TABLE pedidos ADD COLUMN IF NOT EXISTS canal text")

package main

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

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

// aplicar executa um comando de migração sem formar fila: desiste rápido e tenta de novo
func aplicar(ctx context.Context, sql string, tentativas int) error {
	conexao, err := pgx.Connect(ctx, os.Getenv("DATABASE_URL"))
	if err != nil {
		return err
	}
	defer conexao.Close(ctx)

	// Não espera mais que 1 s pela trava
	if _, err := conexao.Exec(ctx, "SET lock_timeout = '1s'"); err != nil {
		return err
	}
	for tentativa := 1; ; tentativa++ {
		_, err := conexao.Exec(ctx, sql)
		if err == nil {
			fmt.Println("aplicada na tentativa", tentativa)
			return nil
		}
		var pgErr *pgconn.PgError
		// 55P03: a trava não veio a tempo; quem esperava atrás da migração foi liberado
		if !errors.As(err, &pgErr) || pgErr.Code != "55P03" || tentativa == tentativas {
			return err
		}
		fmt.Printf("tentativa %d: tabela ocupada, tentando de novo\n", tentativa)
		pausa := time.Duration(500*tentativa+rand.IntN(500)) * time.Millisecond
		time.Sleep(pausa)
	}
}

func main() {
	err := aplicar(context.Background(), "ALTER TABLE pedidos ADD COLUMN IF NOT EXISTS canal text", 10)
	if err != nil {
		panic(err)
	}
}

import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.concurrent.ThreadLocalRandom;

public class Migrar {
    /** Aplica um comando de migração sem formar fila: desiste rápido e tenta de novo. */
    static void aplicar(String sql, int tentativas) throws Exception {
        try (var conexao = DriverManager.getConnection(System.getenv("JDBC_URL"));
             var comando = conexao.createStatement()) {
            comando.execute("SET lock_timeout = '1s'"); // não espera mais que 1 s pela trava
            for (int tentativa = 1; ; tentativa++) {
                try {
                    comando.execute(sql);
                    System.out.println("aplicada na tentativa " + tentativa);
                    return;
                } catch (SQLException erro) {
                    // 55P03: a trava não veio a tempo; quem esperava atrás foi liberado
                    if (!"55P03".equals(erro.getSQLState()) || tentativa == tentativas) throw erro;
                    System.out.println("tentativa " + tentativa + ": tabela ocupada");
                    Thread.sleep(500L * tentativa + ThreadLocalRandom.current().nextInt(500));
                }
            }
        }
    }

    public static void main(String[] args) throws Exception {
        aplicar("ALTER TABLE pedidos ADD COLUMN IF NOT EXISTS canal text", 10);
    }
}

using Npgsql;

// Aplica um comando de migração sem formar fila: desiste rápido e tenta de novo
async Task Aplicar(string sql, int tentativas = 10)
{
    var url = Environment.GetEnvironmentVariable("DATABASE_URL");
    await using var conexao = new NpgsqlConnection(url);
    await conexao.OpenAsync();
    await using (var limite = new NpgsqlCommand("SET lock_timeout = '1s'", conexao))
        await limite.ExecuteNonQueryAsync(); // não espera mais que 1 s pela trava

    for (var tentativa = 1; ; tentativa++)
    {
        try
        {
            await using var comando = new NpgsqlCommand(sql, conexao);
            await comando.ExecuteNonQueryAsync();
            Console.WriteLine($"aplicada na tentativa {tentativa}");
            return;
        }
        // 55P03: a trava não veio a tempo; quem esperava atrás da migração foi liberado
        catch (PostgresException erro) when (
            erro.SqlState == PostgresErrorCodes.LockNotAvailable && tentativa < tentativas)
        {
            Console.WriteLine($"tentativa {tentativa}: tabela ocupada, tentando de novo");
            await Task.Delay(500 * tentativa + Random.Shared.Next(500));
        }
    }
}

await Aplicar("ALTER TABLE pedidos ADD COLUMN IF NOT EXISTS canal text");