Processamento de Registros em Lote com PostgreSQL e Go

Este é um padrão interessante para processamento de registros em lote no Go com PostgreSQL. Ele utiliza transações e bloqueios para garantir que os registros sejam processados de forma segura e eficiente, evitando conflito entre múltiplas instâncias do processo.

O truque é combinar o uso de FOR UPDATE SKIP LOCKED para evitar que múltiplas instâncias do processo leiam e processem o mesmo registro ao mesmo tempo, e o uso do índice (id) para evitar que algum erro de processamento faça com que o registro seja processado novamente porque não houve uma mudança de status.

Se houver um erro durante o processamento, o registro não é atualizado e permanece com o status ‘pendente’. Sem o WHERE id > $1, o loop poderia entrar em um ciclo infinito tentando processar o mesmo registro repetidamente.

	var id int
	for {
		done := func() bool {

			const sqlStatement = `SELECT
                id, ...
            FROM ...
            WHERE id > $1
            AND status = 'pendente'
            AND ...
            ORDER BY id
            FOR UPDATE SKIP LOCKED LIMIT 1;`

            // aqui inicia uma transação
            tx := ...
            defer func() {
                // mecanismo de rollback
            }()

            // então executa a consulta
            err := tx.QueryRow(sqlStatement, id).Scan(&id, ...)
            if err != nil {
                if errors.Is(err, sql.ErrNoRows) {
                    return true // não há mais registros
                }
                // trata outros erros
                ...
                return false
            }

            // aqui faz o processamento do registro e atualiza o status
            // pode ser fail, paid, processed, etc.
            // ...

            // se tudo estiver ok, confirma a transação
            err = tx.Commit()
            if err != nil {
                // trata erro de commit geralmente log.Fatal(...)
            }
            return false // indica que o loop deve continuar
        }()
        if done {
            break // não há mais registros para processar
        }
    }

Este é um exemplo bom para processamento de registros em lote e pode ser tranquilamente colocado para rodar de tempos em tempos, como um cron job, para processar registros pendentes em um banco de dados PostgreSQL.

Uma vantagem interessante é que, se o processamento demorar muito, automaticamente o cron vai carregar outras instâncias do processo, que vão pegar os registros que ainda não foram processados e não estão bloqueados (SKIP LOCKED). Isso permite que o sistema escale horizontalmente, processando mais registros em paralelo sem risco de conflitos.