Venture allows you to create and manage complex, async workflows in your Laravel apps.

I recently started having concerns about the payload size when saving dependantJobs with full job payloads and today I got my first exception for workflow_jobs.job column being too small. Average size of the job column for my project while testing is almost 20kb, with jobs regularly creeping up towards 40-50kb and this is just for the small workflows/jobs in the project. This PR reduces the size of the job column by 85% on average for me, but upwards of 95% on my current test data.

What are your thoughts on this?

I took the liberty of trying to solve the limitation of duplicate jobs, as I need it for a project.

I'm guessing there's a few areas in the code you might want to have changed or adjusted, so consider this more of a jumping off point for discussion :)

This PR changes the following;

• Instead of relying on jobClassName, switch to using stepId in the graph
• Keep a classMap for lookup, instead of having to inspect the graph constantly
• Add getDependenciesAsJobs for similar functionality to getDependantJobs but backwards compatible by keeping current method
• Loosen visibility of resolveDependencies so WorkflowDefinition can use it
• Added hasWorkflow method to easier test if workflows were merged instead of having to assert all of the individual jobs were added
• Added getWorkflow method to easier make assertions on added workflows

I wanted to test out the nested workflows, so I did the following. Its a typical scenario of document signing.

I have a bunch of dummy jobs that does nothing else than write a line in the log. just to see that it works.

return Workflow::define('Sign a document')
    ->addJob(new PrepareDocumentForSigning($this->document))
    ->addWorkflow(new FirstOrderSigning($this->document), [
        PrepareDocumentForSigning::class,
    ])
    ->addWorkflow(new SecondOrderSigning($this->document), [
        FirstOrderSigning::class,
    ])
    ->addJob(new ExecuteDocument($this->document), [
        SecondOrderSigning::class,
    ])
    ->addJob(new SendExecutedDocument($this->document), [
        ExecuteDocument::class,
    ])
    ->catch(function (\Sassnowski\Venture\Models\Workflow $workflow, $job, Throwable $e) {
        info("Workflow cancelled " .  $e);
        $workflow->cancel();
    });

FirstOrderSigning and SecondOrderSigning are identical and its content is:

class FirstOrderSigning extends AbstractWorkflow
{
    private Document $document;

    public function __construct(Document $document)
    {
        $this->document = $document;
    }

    public function definition(): WorkflowDefinition
    {
        return Workflow::define('Sign a document')
            ->addJob(new SendSigningEmailInvite($this->document))
            ->addJob(new CaptureSignature($this->document), [
                SendSigningEmailInvite::class,
            ]);
    }
}

I run

SignDocumentWorkflow::start($document);

but Horizon outputs only 1 job executed: (the one before the nested workflow)

However, if I remove those workflows and only use addJobs - then add 3 jobs get executed.

return Workflow::define('Sign a document')
    ->addJob(new PrepareDocumentForSigning($this->document))
    ->addJob(new ExecuteDocument($this->document), [
        PrepareDocumentForSigning::class,
    ])
    ->addJob(new SendExecutedDocument($this->document), [
        ExecuteDocument::class,
    ])
    ->catch(function (\Sassnowski\Venture\Models\Workflow $workflow, $job, Throwable $e) {
        info("Workflow cancelled " .  $e);
        $workflow->cancel();
    });

What am I missing here?

The development of version 4.0 happens in the 4.x branch of this repository.

Todo

• [x] Implement Plugin System
• [x] Replace vimeo/psalm with nunomaduro/larastan
• [x] Define WorkflowStepInterface for workflow steps
• [x] Add gated jobs (see #32 for reference)
• [x] Add closure-based jobs (see #24) (968c26354d638fa5a8b3e7842a822344a803df67)
• [x] Add testing helpers to run the then and catch callbacks of a workflow (42231332e4d0504f4c01cfd14980750ab39550b8)
• [x] Add option to run all jobs of a workflow on a specific connection (804193ec60352110cd783e6a18897c0523376832)
• [x] Add plugin to make workflows entity aware
• [x] Add to workflow definition of existing workflow instance
• [x] Write migration guide
  • [x] Migrating the database
  • [x] Minimum Laravel Version
  • [x] Steps should use WorkflowStepInterface
  • [x] Replace Workflow::define(...) with $this->define(...)
• [x] Update documentation
  • [x] Update all examples to the new API
  • [x] Writing Plugins
  • [x] Optional EntityAwareWorkflows plugin
  • [x] Testing the then and catch callbacks of a workflow
  • [x] Define queue connection when starting a workflow
  • [x] Asserting that a workflow was started on a specific connection
  • [x] Adding jobs to an existing workflow instance
  • [x] Adding conditional jobs
  • [x] Depending on conditional jobs
  • [x] The callback passed to assertStarted now takes the job's queue connection as a second parameter
  • [x] Adding a closure to a workflow
  • [x] Gated jobs
  • [x] Staring a workflow synchronously
  • [x] Defining the queue connection for all jobs of a workflow
  • [x] Defining the queue for all jobs of a workflow
  • [x] Update section about testing workflows to use WorkflowTester
  • [x] Update section about inspecting workflow models to use state
  • [x] Customizing model states

BC Breaks

This section is used to keep track of all BC breaks this version introduces. While I will try to keep them to a minimum and provide sensible migration paths, there are a few warts in the current implementation that require a BC break to really fix.

Drop Laravel 8 support

Starting with 4.0, Venture will require Laravel 9.

Jobs need to implement WorkflowStepInterface

~Since we can't type hint against traits, we currently have to resort to the object type whenever we're dealing with workflow jobs. Version 4.0 adds a proper interface and extends the existing WorkflowStep trait so it automatically implements the interface.~

~Since it's necessary to add implements WorkflowStepInterface to all existing workflow job classes, this is a BC break. We should be able to automate this with some sort of artisan command, however (e.g. venture:migrate-jobs). We basically have to look for classes that use the WorkflowStep trait and perform some kind of string replace on the class declaration.~

I found a way to make this backwards compatible for now. Instead, adding a job that doesn't implement the interface to a workflow will trigger a deprecation warning in version 4.

Workflow::define requires the workflow instance as the first parameter

Up until now, the Workflow facade's define method only took a $name parameter. In version 4, this method takes an instance of AbstractWorkflow as a parameter.

- Workflow::define('My workflow');
+ Workflow::define($this, 'My workflow');

This is so we can acces the actual workflow class inside plugins, instead of only dealing with the WorkflowDefinition class. This allows users to define extra properties/methods on their workflow classes which can then be used by various plugins.

As a convenience, the AbstractWorkflow class now provides a define method which automatically passes the current workflow instance to the Workflow::define method.

- Workflow::define('My workflow');
+ $this->define('My workflow');

Again, this is something we should be able to automate since it should be a rather straight-forward string replace.

Changes to WorkflowManagerInterface::startWorkflow signature

The WorkflowManagerInterface::startWorkflow method now takes a second parameter $connection which specifies the connection to use for all jobs of the workflow. If $connection is null, it will use whatever default connection has been set on the job.

This is a fairly low-impact change as a I think it's unlikely that people will implement their own workflow manager or extend the default one. It is still a breaking change, however.

StepIdGenerator needs to account for WorkflowStepAdapter

Jobs that don’t implement WorkflowStepInterface directly now get wrapped in a WorkflowStepAdapter class. This means that when implementing a custom StepIdGenerator, you now need to account for this class and potentially unwrap it before generating the id.

Not sure if I am being stupid but none of the job/steps ever get marked as finished (no update to finished_at column). Is this done automatically? Do I need to do it manually or is anything required to be returned from the handle() method in the Jobs?

The workflow runs fine and all the jobs do complete, they are just not being marked as finished.

Hi, I recently started using this package but stumbled upon an issue around non-public fields on jobs.

If a job has a non-public field PHP \unserialize function places a \0\0 value in front of it. When this serialized string is transferred into DB and it is accepted as a null-terminated string and as expected half of it is missing. So when this package tries to unserialize the string, it does not work.

Currently, I made them public to overcome the issue but another solution might be possible. I thought about SerializesModels but, trait usage can not be checked against. So don't really know about it.

Currently, I don't have the repro but if requested I can whip something up.

laravel/framework v8.83.4 ksassnowski/venture 3.5.0 php8.1 Postgres 12

It's possible to conditionally add jobs and workflows in the definition method, however, I'd like to have all my jobs added and instead early return from them before they do any work. In order to do so I've added a $shouldSkip property to my jobs and workflows.

It works fine for jobs, but I'm having a little difficulty when merging workflows.

Consider this example;

Workflow::define('My workflow')
    ->addJob(new MyJob())
    ->addJob((new MyOtherJob())->skip($myCondition), [
        MyJob::class,
    ])
    ->addWorkflow((new MyConditionalWorkflow())->skip($myCondition), [
        MyOtherJob::class,
    ]);

MyOtherJob is all good, but MyConditionalWorkflow currently needs a defintion like;

Workflow::define('My conditional workflow')
    ->addJob(
        (new MyNestedJob())->skip($this->shouldSkip)
    )

This gets kinda cumbersome very quickly when you have a few jobs in the nested workflow.

WorkflowDefinition::addWorkflow could be modified with the following hook;

        $definition = $workflow->definition();

        if (method_exists($workflow, 'beforeConnecting')) {
            $workflow->beforeConnecting($definition->jobs);
        }

-- or just call it regardless, if a no-op method was added to AbstractWorkflow like beforeCreate currently is.

What's your thoughts? Open to a PR with tests for this, or for another implementation?

For reference, my skip is basically just the following and similar for Jobs

abstract class AbstractWorkflow extends \Sassnowski\Venture\AbstractWorkflow
{
    protected bool $shouldSkip = false;

    public function skip(bool $shouldSkip = true)
    {
        return tap($this, fn () => $this->shouldSkip = $shouldSkip);
    }
}

Venture looks super interesting and very suitable for an upcoming project of mine. Could be super sweet if you could nest workflows to increase reusability of common workflows.

For instance, the following example;

class EncodeWorkflow extends AbstractWorkflow
{
    public function definition(): WorkflowDefinition
    {
        return Workflow::define('Encode Podcast')
            ->addJob(new EncodePodcast($this->podcast, 'mp3'))
            ->addJob(new EncodePodcast($this->podcast, 'wav'));
    }
}

class OptimizeWorkflow extends AbstractWorkflow
{
    public function definition(): WorkflowDefinition
    {
        return Workflow::define('Optimize Podcast')
            ->addJob(new AnalyzePodcast($this->podcast))
            ->addJob(new OptimizePodcast($this->podcast), [AnalyzePodcast::class])
            ->addWorkflow((new EncodeWorkflow($this->podcast))->definition(), [OptimizePodcast::class]);
    }
}

class PublishWorkflow extends AbstractWorkflow
{
    public function definition(): WorkflowDefinition
    {
        return $publishWorkflow = Workflow::define('Publish Podcast')
            ->addJob(new ReleaseOnTransistorFM($this->podcast))
            ->addJob(new ReleaseOnApplePodcasts($this->podcast));
    }
}


$processWorkflow = Workflow::define('Release Podcast')
    ->addJob(new ProcessPodcast($this->podcast))
    ->addWorkflow((new OptimizeWorkflow($this->podcast))->definition(), [ProcessPodcast::class])
    ->addWorkflow((new PublishWorkflow($this->podcast))->definition(), [OptimizeWorkflow::class]);

Have you considered something like this for the package? If so, do you already have an idea for such an implementation, or would you be open to a PR when I start on the project that would benefit from such a feature?

Sometimes workflows have users involved to perform an action before the workflow step is complete and the overall workflow can continue. An example would be a user that needs to review and approve a blog post before it gets published. It would be beneficial to see support for user tasks within this package.

A user task can be completed in e.g. a controller and the application should be able to update a specific workflow step to complete or cancelled.

I believe we would need:

• Jobs that do not automatically completes when run - but instead is in progress
• workflow API to complete a step

Description

This PR adds the ability to easily fetch the entire completion status of a workflow. One that may have processed and failed jobs, but is actually done processing.

Purpose

In a workflow, some jobs may pass and some may fail, leaving the finished_at and cancelled_at timestamps empty. This simple helper method allows me to check if the workflow is actually "done", and then perform further analysis afterwards.

Let me know if you'd like more tests added. Thanks so much for this package and your time! ❤️

The current stable version of opis/closure is not yet compatible with PHP 8.1 and produces various deprecation warnings. While version 4 aims to fix this, it is still in beta. So in the meantime, we're switching to laravel/serializable-closure.

I attempted to add the package and use it on one process, but I am now unable to run any of my other non-venture jobs.

So I started commenting out the events in the venture provider, then ran my test for my regular jobs, which are Laravel Actions Job, and it passed. All stim from the JobProcessed event appears to be attempting to unserialize a base64, which fails because regular jobs do not have base64 as far as I know?

Step 1: https://github.com/ksassnowski/venture/blob/40a5bb1367cd61d8ac9da203a7b8b67134043e4b/src/VentureServiceProvider.php#L51

Step 2: https://github.com/ksassnowski/venture/blob/40a5bb1367cd61d8ac9da203a7b8b67134043e4b/src/WorkflowEventSubscriber.php#L96

Step 3: https://github.com/ksassnowski/venture/blob/40a5bb1367cd61d8ac9da203a7b8b67134043e4b/src/UnserializeJobExtractor.php#L27

Step 4: https://github.com/ksassnowski/venture/blob/40a5bb1367cd61d8ac9da203a7b8b67134043e4b/src/Serializer/Base64WorkflowSerializer.php#L38

So seems to be incompatible with Laravel Jobs and it's venture all or nothing.

Afternoon,

I'm currently doing research for work around using the Venture package for one of our projects. Something I noticed and wanted to see if you were interested in as a future feature/enhancement is making workflow jobs aware of their parent workflows state. For example, say you use a workflow to model an email campaign where follow-up emails are sent. But if any point in time, the receiver replies, we don't want to send any of the remaining follow-up emails:

<?php declare(strict_types=1);

namespace App;

use Sassnowski\Venture\AbstractWorkflow;
use Sassnowski\Venture\WorkflowDefinition;

class WorkflowTest extends AbstractWorkflow
{
    public function definition(): WorkflowDefinition
    {
        $this->define('Workflow Name')
            ->addJob(id: 'send-first-followup', (new SendFollowUp())->delay($this->settings->first_delay))
            ->addJob(id: 'send-second-followup', (new SendFollowUp())->delay($this->settings->second_delay))
            ->addJob(id: 'send-third-followup', (new SendFollowUp())->delay($this->settings->third_delay));
    }
}

At the moment, I think this is possible with some logic on the end users' side, doing something like this in the Job class

<?php declare(strict_types=1);

namespace App\Jobs;

use Sassnowski\Venture\WorkflowableJob;
use Sassnowski\Venture\WorkflowStep;

class SendFollowup implements WorkflowableJob
{
    use WorkflowStep;

    public function handle(): void
    {
        if ($this->workflow()->isCancelled() || $this->workflow()->isFinished()) {
            return;
        }

        // send email
    }
}

And then, in our business logic, when we receive an inbound email from the user (or another event like an unsubscribe), we can manually reference the workflow and change its state to canceled or finished.

What are your thoughts on making WorkflowableJob's state aware so that little to no code is needed to make workflow jobs aware of their parent's state and not execute or dispatch jobs if the workflow has already finished?

Would you be open to the introduction of a repository to make the persistence of models abstract?

I'm not thinking any massively crazy refactor, but basically just a repository that persists (and fetches) the eloquent models, instead of calling save() directly on the model like here

My motivation for this is

1. Be able to use an in memory repository in my current project during tests
2. Have an easy way to hook into loading/saving the models for extra metadata