SAVEPOINT 与嵌套事务

概念定义

事务在顶层是全有或全无的。savepoint 在内部刻出检查点：失败的子步骤退回检查点，外围工作得以幸存。这就是“导入一万行、跳过一行坏数据而不丢九千九百九十九行”的实现方式。

工作机制

SAVEPOINT 建立命名标记，ROLLBACK TO SAVEPOINT 撤销其后工作、保留之前获得的锁，RELEASE SAVEPOINT 释放名字。PostgreSQL 出错只中止当前语句，捕获后继续很自然；Oracle 与 SQL Server 有类似机制但事务域细节不同。

使用场景

逐条容错的批量装载、后续步骤可跳过的复合操作、把每个用例包在回滚里的测试框架。建议逐条编号 savepoint 并定期 release 以约束状态规模。

常见误区

savepoint 是同一事务内的位置，不是独立事务，提交语义仍是整事务级的；它也只影响创建它的那条连接。部分驱动或严格模式在出错时直接中止整个事务，回退点形同虚设——需按栈验证。savepoint 之后获取的锁要到外层事务结束才释放；深层嵌套不释放会劣化性能与内存。

代码示例

from collections.abc import Iterator
from contextlib import contextmanager
from dataclasses import dataclass, field
from typing import Any


class TransactionAborted(RuntimeError):
    pass


@dataclass
class Store:
    data: dict[str, Any] = field(default_factory=dict)
    commits: int = 0
    rollbacks: int = 0

    @contextmanager
    def transaction(self) -> Iterator["Transaction"]:
        snapshot = dict(self.data)
        tx = Transaction(self)
        try:
            yield tx
        except BaseException:
            self.data = snapshot
            self.rollbacks += 1
            raise
        else:
            self.commits += 1


@dataclass
class Transaction:
    store: Store

    def set(self, key: str, value: Any) -> None:
        self.store.data[key] = value

    def delete(self, key: str) -> None:
        self.store.data.pop(key, None)


def main() -> None:
    store = Store(data={"balance": 100})
    with store.transaction() as tx:
        tx.set("balance", 80)
        tx.set("currency", "USD")
    print(store.data, store.commits, store.rollbacks)

    try:
        with store.transaction() as tx:
            tx.set("balance", 0)
            raise TransactionAborted("payment declined")
    except TransactionAborted as error:
        print("rolled back:", error)
    print(store.data, store.commits, store.rollbacks)


if __name__ == "__main__":
    main()

package main

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"time"
)

var ErrConflict = errors.New("serialization conflict")

func WithTx(
	ctx context.Context,
	db *sql.DB,
	attempts int,
	work func(*sql.Tx) error,
) error {
	if attempts < 1 {
		attempts = 1
	}
	var last error
	for attempt := 0; attempt < attempts; attempt++ {
		tx, err := db.BeginTx(ctx, nil)
		if err != nil {
			return err
		}
		err = work(tx)
		if err == nil {
			if commitErr := tx.Commit(); commitErr == nil {
				return nil
			} else {
				last = commitErr
			}
		} else {
			_ = tx.Rollback()
			last = err
		}
		if !errors.Is(last, ErrConflict) {
			return last
		}
		select {
		case <-ctx.Done():
			return ctx.Err()
		case <-time.After(time.Duration(attempt+1) * 50 * time.Millisecond):
		}
	}
	return fmt.Errorf("transaction failed after %d attempts: %w", attempts, last)
}

func transfer(ctx context.Context, db *sql.DB, from, to, amount int64) error {
	return WithTx(ctx, db, 3, func(tx *sql.Tx) error {
		result, err := tx.ExecContext(
			ctx,
			"UPDATE accounts SET balance = balance - ? WHERE id = ? AND balance >= ?",
			amount, from, amount,
		)
		if err != nil {
			return err
		}
		affected, err := result.RowsAffected()
		if err != nil {
			return err
		}
		if affected == 0 {
			return errors.New("insufficient funds")
		}
		_, err = tx.ExecContext(
			ctx,
			"UPDATE accounts SET balance = balance + ? WHERE id = ?",
			amount, to,
		)
		return err
	})
}

func main() {
	fmt.Println(transfer != nil, ErrConflict)
}