Combining awaitables: awaitWithTimeout + sideEffect

[naquad]

Hello,

EDIT: digging to the Workflow::handle, I have discovered that the exception is never thrown (there's no $this->coroutine->throw() call for the failed promise). Now I'm confused: does that mean that awaitable can never fail?

I've got a big DIY FSM with lots of states and signals. Workflow seems to be a great thing that will simplify the code, make it clear, and managable.

The last puzzle piece I miss is a signal reactor with a mask. Sort of:

public function execute()
{
        // match only the signals in the mask or throw timeout exception
        $signal = yield $this->waitFor(signals: [
                MyEnumWithSignals::Signal1,
                MyEnumWithSignals::Signal2,
                MyEnumWithSignals::Signal3,
        ]);

        dump($signal); // output the received signal
}

My current implementation is the following:

    protected ?MyEnumWithSignals $lastSignal = null;

    // somewhy calling signal method signal() causes silent failure
    #[SignalMethod]
    public function emit(MyEnumWithSignals $signal): void
    {
        $this->lastSignal = $signal;
    }

    public function waitFor(
        array $signals = [],
        int $timeout = 0,
    ): PromiseInterface
    {
        $waiter = function() use ($signals) { // predicate to determine if the signal is received
            return $this->lastSignal !== null &&
                (!$signals || in_array($this->lastSignal, $signals, true));
        };

        if ($timeout > 0) { // handle the timeout
            // convenience method stands for WorkflowStub::awaitWithTimeout
            $resp = $this->awaitWithTimeout($timeout, $waiter);
        } else {
            // convenience method stands for WorkflowStub::await
            $resp = $this->await($waiter);
        }

        // That's the tricky part: after the signal is received (predicate() => true),
        // get the actual signal and resolve the awaitable with it.
        // originally, I've been trying to return sideEffect from the onResolved handler,
        // but that didn't work
        return $resp->then(function($value) {
            if (!$value) { // timeout
                return reject(new DomainException('Timeout'));
            }

            $signal = $this->lastSignal;
            $this->lastSignal = null;
            return resolve($signal);
        });
    }

This works, but I'm still not sure if I'm using the promise from await* correctly and if this code is reliable. I came up with it after reading what await() does and that there's a Promise instance under the hood that does its magic. I'm not sure if the event sourcing will properly handle this in all cases.

So am I getting it right or I shouldn't mess with the Promise instance?

Thank you.

[rmcdaniel]

Under the hood, a workflow execution from start to finish is not a single execution but several partial executions. After each partial execution, the workflow unloads from memory to wait for the activities to finish (which could be seconds or hours) and then reloads later.

https://github.com/laravel-workflow/laravel-workflow/blob/master/src/Workflow.php#L199

During a partial execution, the result of a promise represents the result of an activity. If the activity hasn't finished, then an unresolved promise is returned.

https://github.com/laravel-workflow/laravel-workflow/blob/master/src/ActivityStub.php#L53

Once you have this promise you can yield it to unload the workflow from memory and wait for the activity to complete. When the activity is complete, the workflow runs again and the same call to ActivityStub::make() is made but now it's immediately resolved.

https://github.com/laravel-workflow/laravel-workflow/blob/master/src/ActivityStub.php#L46

When you yield the resolved promise, the workflow doesn't unload. Instead, it unwraps the promise and sends the result to the coroutine.

https://github.com/laravel-workflow/laravel-workflow/blob/master/src/Workflow.php#L191

You can think of these promises as "logical" in the sense that they are made of several actual promises. Kind of like how a logical disk is made of many physical ones. I hope my explanation helped with understanding the inner workings more.

As for what you're trying to achieve, if I understand correctly, this is how you would do it.

use Workflow\ActivityStub;
use Workflow\SignalMethod;
use Workflow\Workflow;
use Workflow\WorkflowStub;

class MyWorkflow extends Workflow
{
    private bool $ready1 = false;
    private bool $ready2 = false;

    #[SignalMethod]
    public function setReady1($ready1)
    {
        $this->ready1 = $ready1;
    }

    #[SignalMethod]
    public function setReady2($ready2)
    {
        $this->ready2 = $ready2;
    }

    public function execute()
    {
        $signaled = yield from $this->awaitAnyCondition([
            fn() => $this->ready1,
            fn() => $this->ready2,
        ], 30);

        if (! $signaled) {
            // handle timeout logic here
            return 'timeout';
        }

        // handle normal logic here
        return 'ok';
    }

    private function awaitAnyCondition(array $conditions, ?int $timeout = null)
    {
        $conditionCheck = fn () => array_reduce($conditions, fn ($carry, $condition) => $carry || $condition(), false);

        if ($timeout) {
            return yield WorkflowStub::awaitWithTimeout($timeout, $conditionCheck);
        }

        return yield WorkflowStub::await($conditionCheck);
    }
}

[naquad]

Then a question: can Activity watch for signals in the flow? Or is there a way to forward the signal to the activity?

[rmcdaniel]

I'm not sure if this is what you mean?

use Workflow\Workflow;
use Workflow\WorkflowStub;
use Workflow\ActivityStub;
use Workflow\SignalMethod;

class MySignalWorkflow extends Workflow
{
    private $signalValue = null;

    #[SignalMethod]
    public function receiveSignal($value)
    {
        $this->signalValue = $value;
    }

    public function execute()
    {
        yield WorkflowStub::await(fn () => !is_null($this->signalValue));

        $result = yield ActivityStub::make(MySignalActivity::class, $this->signalValue);

        return $result;
    }
}

use Workflow\Activity;

class MySignalActivity extends Activity
{
    public function execute($signalValue)
    {
        return "Received signal with value: {$signalValue}";
    }
}

If not, maybe we can take a step back and look at the bigger picture of what kind of control flow you're trying to achieve? Could you diagram it out or list the steps perhaps?

[naquad]

The usage I'm trying to achieve is this:

public function execute()
{
        try {
            // match only the signals in the mask or throw timeout exception
            $signal = yield $this->waitFor(signals: [
                    MyEnumWithSignals::Signal1,
                    MyEnumWithSignals::Signal2,
                    MyEnumWithSignals::Signal3,
            ]);
        } catch (Timeout $e) {
                $this->cleanup();
                Log::warning(...);
                return;
        }

        dump($signal); // the flow continues
}

I want the waitFor to return when one of the signals is received or throws a Timeout exception. The problem is the exception part: from what we've figured out, the exceptions are not recorded and can't be replayed.

I'm not sure if the diagram of these lines makes sense, but here it is:

stateDiagram-v2
    [*] --> waitForSignal
    waitForSignal --> Exception: Timeout
    Exception --> ErrorHandling
    ErrorHandling --> Log
    Log --> [*]
    waitForSignal --> Signal: Some signal received
    Signal --> Signal1_Handling
    Signal --> Signal2_Handling
    Signal --> Signal3_Handling
    Signal1_Handling --> [*]
    Signal2_Handling --> [*]
    Signal3_Handling --> [*]

[rmcdaniel]

For this we can modify the awaitAnyCondition method from earlier.

use Exception;
use Workflow\ActivityStub;
use Workflow\SignalMethod;
use Workflow\Workflow;
use Workflow\WorkflowStub;

class TimeoutException extends Exception {}

class MySignalWorkflow extends Workflow
{
    private bool $ready1 = false;
    private bool $ready2 = false;

    #[SignalMethod]
    public function setReady1($ready1)
    {
        $this->ready1 = $ready1;
    }

    #[SignalMethod]
    public function setReady2($ready2)
    {
        $this->ready2 = $ready2;
    }

    public function execute()
    {
        try {
            $signal = yield from $this->awaitAnyCondition([
                'ready1' => fn () => $this->ready1,
                'ready2' => fn () => $this->ready2,
            ], 30);
            // $signal will be either 'ready1 or 'ready2'
        } catch (TimeoutException) {
        }
    }

    private function awaitAnyCondition(array $conditions, ?int $timeout = null)
    {
        $conditionCheck = fn () => array_reduce($conditions, fn ($carry, $condition) => $carry || $condition(), false);
        $signal = fn () => array_search(true, array_map(fn ($condition) => $condition(), $conditions));

        if ($timeout) {
            if (! yield WorkflowStub::awaitWithTimeout($timeout, $conditionCheck)) {
                throw new TimeoutException('Timeout waiting for signal');
            }

            return $signal();
        }

        yield WorkflowStub::await($conditionCheck);

        return $signal();
    }
}

[naquad]

Ah, the generator delegation instead of messing with the Promise. This way there are no rejected promises but the exception is thrown based on the value from the log.

Thank you for your help. This definitely solves the task at hand.

[rmcdaniel]

I'm glad to hear! Let me know if you have any other questions. Thanks!