Node.js、PostgreSQL 中的事务冲突、乐观并发控制和事务重试

Posted 2023-03-08

【中文标题】Node.js、PostgreSQL 中的事务冲突、乐观并发控制和事务重试

【英文标题】：Node.js, transaction coflicts in PostgreSQL, optimistic concurrency control and transaction retries

【发布时间】：2020-06-05 22:00:45

【问题描述】：

我想使用PostgreSQL transaction isolation 来确保数据的正确性，optimistic concurrency control 模式会自动重试冲突的事务，而不是我的应用程序预先锁定数据库行和表。

实现这一点的一种常用方法是，Web 应用程序 retries the transaction 在代码块内的特定次数或通过中间件层 replays the HTTP request，也称为 HTTP 请求重放。 Here is an example of such a middleware for Pyramid and Python web applications web。

我没有找到任何关于 Node.js 及其 PostgreSQL 驱动程序如何处理正在进行的两个并发事务并且一个由于读写冲突而无法通过的情况的好的信息。 PostgreSQL 将回滚其中一个事务，但这是如何向应用程序发出信号的呢？在 Python 中，PSQL 驱动程序会在这种情况下引发psycopg2.extensions.TransactionRollbackError。 For other SQL database drivers here are some exceptions they will raise.

当您将 SQL 事务隔离级别设置为 SERIALIZABLE 时，这种行为更为常见，因为您往往会在负载下遇到更多冲突，因此我想优雅地处理它，而不是向用户提供 HTTP 500。

我的问题是：

如果需要特殊处理且重试库不能独立，如何使用 PostgreSQL 和一些常见的 ORM 框架（如 TypeORM）检测脏读回滚？

是否有中间件（NestJS/Express.js/others）来处理这个问题，并在数据库驱动程序发生事务回滚时自动尝试重播 HTTP 请求 N 次？

【问题讨论】：

"如何检测脏读" - 这很简单：Postgres 不支持脏读，所以没有什么可以“检测”

删除了脏读位，所以希望这个问题现在更有意义。我只是想演示一下什么是与图像的事务冲突。

如果 Postgres 中没有脏读，我仍然不明白“脏读回滚”是什么意思

在链接中有更多信息，描述了其他编程语言的问题和解决方案。

【参考方案1】：

以下是您在使用使用 pg library 的库（例如 TypeORM）时如何处理并发：

/**
 * Check error code to determine if we should retry a transaction.
 * 
 * See https://www.postgresql.org/docs/10/errcodes-appendix.html and
 * https://***.com/a/16409293/749644
 */
function shouldRetryTransaction(err: unknown) 
  const code = typeof err === 'object' ? String((err as any).code) : null
  return code === '40001' || code === '40P01';


/**
 * Using a repeatable read transaction throws an error with the code 40001 
 * "serialization failure due to concurrent update" if the user was 
 * updated by another concurrent transaction.
 */
async function updateUser(data: unknown) 
  try 
    return await this.userRepo.manager.transaction(
      'REPEATABLE READ',
      async manager => 
        const user = manager.findOne(User, id);
    
        // Modify user
        // ...
    
        // Save the user
        await manager.save(user);
      
    );
   catch (err) 
    if (shouldRetryTransaction(err)) 
      // retry logic     
     else 
      throw err;
    
  

对于重试事务，我建议使用诸如 async-retry 这样抽象重试逻辑的库。

你会注意到这种模式非常适合简单的东西，但如果你想传递manager（例如，这样事务可以在其他服务中重用），那么这将变得非常麻烦。我建议使用typeorm-transactional-cls-hooked 库，它利用持续本地存储来传播事务。

您可以通过以下方式重放快递应用的交易：

/**
 * Request replay middleware
 */

import retry from 'async-retry';

function replayOnTransactionError(fn: (req, res, next) => unknown) 
  return (req, res, next) => 
    retry(bail => 
      try 
        // Call the actual handler
        await fn(req, res, next);
       catch (err) 
        if (!shouldRetryTransaction(err)) 
          // Bail out if we're not supposed to retry anymore
          return bail(err);
        

        // Rethrow error to continue retrying
        throw err;
      
    , 
      factor: 2,
      retries: 3,
      minTimeout: 30,
    );
  


app.put('/users/:id', replayOnTransactionError(async (req, res, next) => 
  // ...
))

【讨论】：

谢谢米哈伊尔。这是一个好的开始。通常已经存在较长时间的其他框架（Django、Rails）在中间件对等 HTTP 请求中执行此操作，因此不需要为每个函数编写手动更新逻辑。这就是所谓的 HTTP 请求重放。

@MikkoOhtamaa 我添加了一个中间件，用于重放快速应用程序的请求。