We're looking to add locking capabilities to our workflows(mostly for efficiency gains) like so:

@workflow
def locking_wf() -> None:
    acquired, lock_handle = acquire_lock(lock_handle="my_lock", duration_in_seconds=10)

    cond = (
        conditional("test lock")
        .if_(acquired.is_true())
        .then(task1(lock_handle))
        .else_()
        .then(task2())
    )

    cond > release_lock(lock_handle=lock_handle)

Where

acquire_lock

and

release_lock

are Agents responsible for maintaining the lock. Curious if others have similar use-cases and or implementations?

I think theirs no similar in OSS

Sure. From a high level, we have downstreams that can run if any number of upstream triggers it. If something kicks it off, it just shouldn't run again while the other is running. We are perfectly fine with eventual consistency of running the downstream again on the next scheduled execution.

Please use on_failure as well

yea, was just looking at that 🙂