ViewFlow自定义视图中原子事务未回滚的问题排查

我自定义了流程创建视图，用于将登录用户的邮箱添加到流程对象中，同时添加了事务块，期望当self.activation_done()执行失败时，能回滚form_valid()中的所有数据库变更，不保存新流程实例并在表单上显示错误信息。

初始实现代码：

from django.core.exceptions import ValidationError
from django.db import transaction
from django.http import HttpResponseRedirect
from viewflow.flow.views import CreateProcessView, StartFlowMixin


class StarterEmailCreateProcessView(CreateProcessView):
    """
    为流程模型设置用户邮箱
    """
    def form_valid(self, form, *args, **kwargs):
        """表单验证通过时，保存关联模型并完成任务"""
        try:
            with transaction.atomic():
                super(StartFlowMixin, self).form_valid(form, *args, **kwargs)
                # 为流程实例设置提交者邮箱
                self.object.requester_email = self.request.user.email
                self.object.save()
                self.activation_done(form, *args, **kwargs)
        except Exception as ex:
            form.add_error(None, ValidationError(str(ex)))
            return self.form_invalid(form, *args, **kwargs)
        return HttpResponseRedirect(self.get_success_url())

按Django文档说明，事务内抛出异常会触发回滚，但实际测试时，表单显示错误信息的同时，新流程实例依然被保存，事务并未回滚。

更新1：
如果在self.activation_done之前主动抛出异常，事务可以正常回滚，完全搞不懂self.activation_done的内部逻辑为什么会阻止回滚？

测试代码：

class StarterEmailCreateProcessView(CreateProcessView):
    """
    为流程模型设置用户邮箱
    """
    def form_valid(self, form, *args, **kwargs):
        """表单验证通过时，保存关联模型并完成任务"""
        try:
            with transaction.atomic():
                super(StartFlowMixin, self).form_valid(form, *args, **kwargs)
                # 为流程实例设置提交者邮箱
                self.object.requester_email = self.request.user.email
                self.object.save()
                raise Exception('11111111111')
                self.activation_done(form, *args, **kwargs)
                
        except Exception as ex:
            form.add_error(None, ValidationError(str(ex)))
            return self.form_invalid(form, *args, **kwargs)
        return HttpResponseRedirect(self.get_success_url())

我还尝试了嵌套捕获异常的方式，虽然能捕获到异常，但事务仍然没有回滚：

def form_valid(self, form, *args, **kwargs):
        """表单验证通过时，保存关联模型并完成任务"""
        try:
            with transaction.atomic():
                super(StartFlowMixin, self).form_valid(form, *args, **kwargs)
                # 为流程实例设置提交者邮箱
                self.object.requester_email = self.request.user.email
                self.object.save()
                try:
                    self.activation_done(form, *args, **kwargs)
                except:
                    raise Exception(1111111)
        except Exception as ex:
            form.add_error(None, ValidationError(str(ex)))
            return self.form_invalid(form, *args, **kwargs)
        return HttpResponseRedirect(self.get_success_url())

更新2：
查看activation.done的源码后，我注释掉其中的锁逻辑，事务就能正常回滚了：

@Activation.status.transition(source=STATUS.PREPARED, target=STATUS.DONE)
    def done(self):
        """
        创建并启动新的流程实例。

        .. seealso::
            :data:`viewflow.signals.task_started`

        .. seealso::
             :data:`viewflow.signals.task_finished`

        .. seealso::
            :data:`viewflow.signals.flow_started`

        """
        with transaction.atomic(savepoint=True):
            signals.task_started.send(sender=self.flow_class, process=self.process, task=self.task)

            self.process.save()

            # lock_impl = self.flow_class.lock_impl(self.flow_class.instance)
            # self.lock = lock_impl(self.flow_class, self.process.pk)
            # self.lock.__enter__()

            self.task.process = self.process
            self.task.finished = now()
            self.task.save()

            signals.task_finished.send(sender=self.flow_class, process=self.process, task=self.task)
            signals.flow_started.send(sender=self.flow_class, process=self.process, task=self.task)

            self.activate_next()

对应的锁实现代码如下，目前还没弄明白它阻止事务回滚的机制：

class NoLock(object):
    """
    无悲观锁，仅在事务中执行流程任务。
    不适用于包含Join节点的流程。
    """

    def __call__(self, flow):
        @contextmanager
        def lock(flow_class, process_pk):
            with transaction.atomic():
                yield
        return lock

no_lock = NoLock()

内容的提问来源于stack exchange，提问作者James Lin