python - 失败的 ndb 事务尝试不回滚所有更改？

Question

我在理解导致我的应用程序中出现错误的一系列事件时遇到了一些困难，这些错误只能在部署在 GAE 上的应用程序中间歇性出现，而在使用本地 devserver 运行时永远不会出现。 py.

下面的所有相关代码片段(针对 MCV 进行了修剪，希望我没有丢失任何重要内容)都是在处理相同任务队列请求期间执行的。

入口点:

def job_completed_task(self, _):

    # running outside transaction as query is made
    if not self.all_context_jobs_completed(self.context.db_key, self):
        # this will transactionally enqueue another task
        self.trigger_job_mark_completed_transaction()
    else:
        # this is transactional
        self.context.jobs_completed(self)

对应的self.context.jobs_completed(self)是:

@ndb.transactional(xg=True)
def jobs_completed(self, job):

    if self.status == QAStrings.status_done:
        logging.debug('%s jobs_completed %s NOP' % (self.lid, job.job_id))
        return

    # some logic computing step_completed here

    if step_completed:
        self.status = QAStrings.status_done  # includes self.db_data.put()

    # this will transactionally enqueue another task
    job.trigger_job_mark_completed_transaction()

self.status setter，被黑客攻击以获得用于调试此场景的回溯:

@status.setter
def status(self, new_status):
    assert ndb.in_transaction()

    status = getattr(self, self.attr_status)
    if status != new_status:
        traceback.print_stack()
        logging.info('%s status change %s -> %s' % (self.name, status, new_status))
        setattr(self, self.attr_status, new_status)

job.trigger_job_mark_completed_transaction() 最终将这样的新任务排入队列:

    task = taskqueue.add(queue_name=self.task_queue_name, url=url, params=params,
                         transactional=ndb.in_transaction(), countdown=delay)

出现的 GAE 日志，由于它不适合单个屏幕而拆分:

我对 jobs_completed 事务的期望是要么看到 ... jobs_completed ... NOP 调试消息并且没有任务排队，要么至少看到 status change running -> done 信息消息和由 job.trigger_job_mark_completed_transaction() 排队的任务。

我实际看到的是消息，没有排队的任务。

日志似乎表明事务已尝试两次:

• 它第一次发现状态不是done，所以它执行逻辑，将状态设置为done(并显示回溯和信息消息)并且应该以事务方式将新任务排入队列——但它没有

• 第二次它找到状态done 并只打印调试消息

我的问题是 - 如果第一次交易尝试失败，难道不应该回滚状态更改吗？我错过了什么？

Answer 1

我找到了一个解决方法:指定不重试 jobs_completed() 事务:

@ndb.transactional(xg=True, retries=0)
def jobs_completed(self, job):

这会阻止自动重复执行，而是导致异常:

TransactionFailedError(The transaction could not be committed. Please try again.)

这是可以接受的，因为我已经为整个 job_completed_task() 设置了退避/重试安全网。现在一切正常。

至于为什么回滚没有发生，我唯一想到的是在进入事务之前以某种方式读取了实体(并缓存在我的对象属性中)，因此不被视为(相同的)的一部分) 交易。但我找不到可以做到这一点的代码路径，所以这只是猜测。