如何在PostgreSQL事务回滚时记录缺失的序列ID

下面是几种可行的方案，根据你的业务场景选择：

1. 预生成序列ID并手动管控 #

这是最直接的方案，核心是在事务开始时先拿到序列ID，提前写入日志，再执行主表操作，通过事务结果更新日志状态：

首先创建用于记录的日志表：

CREATE TABLE id_gap_log (
    log_id SERIAL PRIMARY KEY,
    target_table TEXT NOT NULL, -- 关联的业务表名
    sequence_id BIGINT NOT NULL, -- 分配的序列ID
    payload JSONB NOT NULL, -- 请求payload
    error_message TEXT, -- 错误信息
    status VARCHAR(20) NOT NULL DEFAULT 'pending', -- pending/used/failed
    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
);

应用层事务示例（伪代码） #

# 伪代码示例，以Python为例
conn = psycopg2.connect(...)
try:
    with conn.cursor() as cur:
        # 1. 获取序列ID
        cur.execute("SELECT nextval('your_table_id_seq')")
        seq_id = cur.fetchone()[0]
        # 2. 写入待处理日志
        cur.execute(
            "INSERT INTO id_gap_log (target_table, sequence_id, payload) VALUES (%s, %s, %s)",
            ("your_table", seq_id, json.dumps(request_payload))
        )
        # 3. 执行主表插入
        cur.execute(
            "INSERT INTO your_table (id, col1, col2) VALUES (%s, %s, %s)",
            (seq_id, val1, val2)
        )
    conn.commit()
    # 提交后更新日志为已使用
    with conn.cursor() as cur:
        cur.execute("UPDATE id_gap_log SET status = 'used' WHERE sequence_id = %s", (seq_id,))
    conn.commit()
except Exception as e:
    conn.rollback()
    # 回滚后记录错误
    with conn.cursor() as cur:
        cur.execute(
            "INSERT INTO id_gap_log (target_table, sequence_id, payload, error_message, status) VALUES (%s, %s, %s, %s, %s)",
            ("your_table", seq_id, json.dumps(request_payload), str(e), "failed")
        )
    conn.commit()
    raise

数据库层PL/pgSQL函数封装 #

如果不想在应用层写太多逻辑，可以把整个逻辑封装成数据库函数：

CREATE OR REPLACE FUNCTION insert_your_table_with_log(p_payload JSONB, p_col1 TEXT, p_col2 INT) RETURNS VOID AS $$
DECLARE
    v_seq_id BIGINT;
BEGIN
    -- 获取序列ID并写入日志
    SELECT nextval('your_table_id_seq') INTO v_seq_id;
    INSERT INTO id_gap_log (target_table, sequence_id, payload)
    VALUES ('your_table', v_seq_id, p_payload);
    
    -- 执行主表插入
    INSERT INTO your_table (id, col1, col2) VALUES (v_seq_id, p_col1, p_col2);
    
    -- 标记日志为已使用
    UPDATE id_gap_log SET status = 'used' WHERE sequence_id = v_seq_id;
EXCEPTION
    WHEN OTHERS THEN
        -- 捕获异常，更新日志为失败并记录错误
        UPDATE id_gap_log 
        SET status = 'failed', error_message = SQLERRM
        WHERE sequence_id = v_seq_id;
        RAISE; -- 重新抛出异常，让应用层感知错误
END;
$$ LANGUAGE plpgsql;

2. 利用逻辑复制追踪序列操作 #

如果不想修改业务代码，可以通过PostgreSQL的逻辑复制功能，解析事务日志（WAL）来追踪序列的分配和主表的插入：

• 开启逻辑复制，创建一个发布者，包含目标业务表和序列的变更日志
• 编写一个订阅者程序，解析WAL日志：
  • 记录所有nextval调用产生的序列ID
  • 匹配对应的主表插入操作，如果某个序列ID没有对应的插入记录（事务回滚），则将其记录到日志表，并关联请求payload（需要payload能从其他日志来源关联）
• 这种方案对业务侵入性低，但需要额外维护逻辑复制和日志解析程序，适合对代码修改敏感的场景

3. 自定义序列函数+定时清理 #

通过自定义函数替代原生nextval，每次分配ID时自动写入日志，再用定时任务清理超时未使用的ID：

创建自定义序列函数 #

CREATE OR REPLACE FUNCTION custom_nextval(seq_name TEXT) RETURNS BIGINT AS $$
DECLARE
    v_seq_id BIGINT;
    v_table_name TEXT;
BEGIN
    -- 从序列名推导业务表名（假设序列命名规则为{table}_id_seq）
    v_table_name := regexp_replace(seq_name, '_id_seq$', '', 'i');
    -- 获取序列ID
    SELECT nextval(seq_name) INTO v_seq_id;
    -- 写入日志，标记为已分配
    INSERT INTO id_gap_log (target_table, sequence_id, status)
    VALUES (v_table_name, v_seq_id, 'allocated');
    RETURN v_seq_id;
END;
$$ LANGUAGE plpgsql;

修改业务表主键默认值 #

ALTER TABLE your_table ALTER COLUMN id SET DEFAULT custom_nextval('your_table_id_seq');

定时任务清理超时记录 #

用pg_cron（需要先安装扩展）创建定时任务，把超过一定时间仍处于allocated状态的记录标记为失败：

-- 安装pg_cron
CREATE EXTENSION IF NOT EXISTS pg_cron;

-- 每天凌晨1点清理超时1小时的记录
SELECT cron.schedule(
    'cleanup-id-gap-log',
    '0 1 * * *',
    $$
        UPDATE id_gap_log 
        SET status = 'failed', error_message = '事务超时回滚'
        WHERE status = 'allocated' AND created_at < NOW() - INTERVAL '1 hour'
    $$
);

这种方案的缺点是无法获取具体的错误信息，只能推测是事务回滚，需要结合应用日志来补全细节。

内容的提问来源于stack exchange，提问作者Chếnhếng Seang