Job nodes can't be connected directly to the Start nodes

[boakley]

In trying to debug why my celery task doesn't seem to do anything I realized that my code was hitting this block in flow_task_decorator:

except flow_task.flow_cls.task_cls.DoesNotExist:
    # There was rollback on job task created transaction,
    # we don't need to do the job
    return

This causes the body of the celery task to do nothing. There is no output, no log message, no indication whatsoever that there's a problem. The celery log shows that it received, accepted and completed the task, but because the decorator returned nothing, nothing actually executed.

Also, that comment means absolutely nothing to me as a user of viewflow. No rollback has been done to my task (at least, not intentionally), and I have no idea how to fix this problem.

I'm guessing this is related to the fact I'm creating the task from the following custom view:

# flows.py
class DistributorAuditFlow(Flow):
    process_title = "Distributor Audit"
    process_cls = DistributorAuditProcess
    summary_template = '{{ process.title }}'
    lock_impl = lock.select_for_update_lock

    start = (
        # flow.Start(StartProcessView,
        #          fields=["distributor", "days_ago_modified"],
        flow.Start(distributor_audit_view,
                   task_title="Distributor Audit",
                   task_description="Perform an audit of one of our distributors")
        .Permission(auto_create=True)
        .Next(this.run_audit)
    )

    run_audit = (viewflow.contrib.celery.Job(send_hello_world_request)
                 .Next(this.view_results)
    )
    ...

# views.py
@flow_start_view()
def distributor_audit_view(request, activation):
    activation.prepare(request.POST or None, user=request.user)
    form = DistributorAuditForm(request.POST or None, request.FILES or None)

    if form.is_valid():
        slug = form.cleaned_data["distributor"].slug
        report_file = form.cleaned_data["report_file"]
        days_ago_modified = form.cleaned_data["days_ago_modified"]
        distributor = form.cleaned_data["distributor"]
        audit_worker = AUDIT_WORKERS[slug]

        activation.process.days_ago_modified = days_ago_modified
        activation.process.distributor = distributor
        activation.process.report_file = report_file
        activation.done()

        # FIXME: redirect to the current view, and modify the html to either show
        # a link to the generated file, or show "processing..." or something to that effect
        return HttpResponse("form was processed")

    return render(request, 'viewflow/flow/task.html',{
                  'form': form,
                  'activation': activation
                  })

[kmmbvnr]

What database do you use? Does it have 'select .. for update' functionality?

Generally task.DoesNotExist could happens b/c celery task started before view code finish and commit. That could happens if lock is not working.

[kmmbvnr]

Sorry, that seems a bug in viewflow. Start nodes do no locking at all, so celery task could start before start view commit the process and task instances

[kmmbvnr]

-pro: 0.10.5 released. Due design flaws in 0.10 the only case for (StartView -> Job) could be fixed.

[kmmbvnr]

btw @boakley you can simplify your view by implementing DistributorAuditForm as model form

class DistributorAuditForm(forms.ModelForm):
    class Meta:
        model = DistributorAuditProcess
        fields = ['report_file', 'days_ago_modified', 'distributor']

@flow_start_view()
def distributor_audit_view(request, activation):
    activation.prepare(request.POST or None, user=request.user)
    form = DistributorAuditForm(
            request.POST or None,
            request.FILES or None,
            instance=activation.process)
    if form.is_valid():
        form.save(commit=False)
        activation.done()
        return redirect(...)
   return render(...)

[boakley]

This doesn't seem to have solved my problem. I upgraded to viewflow-pro 0.10.5, but my tasks still don't do anything. They are still hitting this block of code in job.py:

        except flow_task.flow_cls.task_cls.DoesNotExist:
            # There was rollback on job task created transaction,
            # we don't need to do the job
            return

[kmmbvnr]

I've got the problem.

At the time when flow task is schedulled, the process instance are not commuted. So select_for_update lock wont work in the celery task,

The only solution for now as i see - is to switch to redis based locks

[kmmbvnr]

I fixed the issue with select_for_update lock 0.10.6.pro released

[kmmbvnr]

0.10.7.pro released with fix for cache_lock