Phantom Workflows

[johntrickett86]

Good morning!

Several months ago I refactored my application into a Domain Driven architecture as the application grew larger. I took the opportunity to refactor my workflows and activities as part of the process.

However, I am still getting workflows starting from the old class (which no longer exists):

There is no reference to this class anywhere in my codebase anymore, yet they keep on running. Any idea how I can find the source of this?

[johntrickett86]

There is not much data in Waterline for the workflows to go on

[rmcdaniel]

Have you made sure to restart all your queue workers? If you don’t restart them, they will still run using the old code, regardless of file changes.

[johntrickett86]

Have you made sure to restart all your queue workers? If you don’t restart them, they will still run using the old code, regardless of file changes.

Thanks for the quick reply! Yes, I restart Horizon with each deployment. I've also had to take the server down during the time the change was made and restarted everything again when I brought it back online.

[rmcdaniel]

It sounds like some workflows still have serialized payloads referencing the old class. When making breaking changes to a workflow, you'll need to version it to avoid conflicts with already-running instances. One solution is to put the updated workflow into a version subfolder, like Workflows/V2/ or Workflows/V20250521/.

To resolve the immediate issue, create a new version of ProcessSimproDataWorkflow inside a version subfolder, and update your code to use that path. Then, locate any old workflows still running under the original path and set their status to failed.

[travisaustin]

One way I've worked around versioning is to use versioned queue names. I have a class that defines the queue names as const values. I use these const values to name the queues in queues.php and horizon.php, as well as in each dispatched job/workflow.

Then, anytime I make a breaking change (to how things are serialized, etc), I modify the queue names in that 1 central file and deploy my update.

Using this pattern, I can leave an instance of the old version of my stack running for a while until all of the old processes are complete. The old stack will use the old queue names, and the new stack will use the new queue names.

Here's the class I use to define queue names. When it's time to introduce a breaking change, I simply modify QUEUE_SUFFIX (usually to the current month and year).

<?php

// File: QueueConstants.php

use Illuminate\Support\Collection;

class QueueConstants
{
    // Change this prior to deploying breaking changes in job structure. Leave an instance of the old running until 
    // the old queues are empty and nothing else is being dispatched to those old queues, and then terminate 
    // that remaining old instance.
    const QUEUE_SUFFIX = '202406';

    /** @var string DEFAULT_QUEUE
     *
     * 'default' is hard-coded as the queue name of any jobs that do not specify a name. If changing this, it is imperative to
     * ensure that every single job dispatched for this app has a queue name set. Since that's not practical 
     * (dispatchable events, mailables, etc), it is easier to just leave this as 'default'. That's probably safe so long as 
     * we don't change the signature of any mission-critical jobs on this queue (during a deploy, an old version of the job 
     * may be used to process a newly serialized object, or vice versa).
     */
    const DEFAULT = 'default_' . self::QUEUE_SUFFIX;

    const MY_QUEUE = 'my_queue_' . self::QUEUE_SUFFIX;
}

<?php

// File: queue.php

return [

    'connections' => [
        'redis' => [
            'driver' => 'redis',
            'connection' => 'default',
            // the default queue for all new jobs on this connection (only for jobs that don't specify their queue)
            'queue' => env('REDIS_QUEUE', QueueConstants::DEFAULT),
            'retry_after' => (60 * 15), // Run for max 15 minutes
            'block_for' => 1,
            'after_commit' => false,
        ],

        'my_queue' => [
            'driver' => 'redis',
            'connection' => 'default',
            'queue' => QueueConstants::MY_QUEUE,
            'retry_after' => (60 * 65), // Run for max 65 minutes
            'block_for' => 1,
        ],
    ],

];