We use cookies

We use cookies to ensure you get the best experience on our website. For more information on how we use cookies, please see our cookie policy.

By clicking "Accept", you agree to our use of cookies.
Learn more.

Durable Events

Durable events are a feature of durable tasks which allow tasks to wait for an event to occur before continuing. This is useful in cases where a task needs to wait for a long time for an external action. Durable events are useful, because even if your task is interrupted and requeued while waiting for an event, the event will still be processed. When the task is resumed, it will read the event from the durable event log and continue processing.

Declaring durable events

Durable events are declared using the context method WaitFor (or utility method WaitForEvent) on the DurableContext object.

examples/durable_event/worker.py

@hatchet.durable_task(name="DurableEventTask")
async def durable_event_task(input: EmptyModel, ctx: DurableContext) -> None:
    res = await ctx.aio_wait_for(
        "event",
        UserEventCondition(event_key="user:update"),
    )

    print("got event", res)

src/v1/examples/durable-event/workflow.ts

export const durableEvent = hatchet.durableTask({
  name: 'durable-event',
  executionTimeout: '10m',
  fn: async (_, ctx) => {
    const res = ctx.waitFor({
      eventKey: 'user:update',
    });

    console.log('res', res);

    return {
      Value: 'done',
    };
  },
});

examples/v1/workflows/durable-event.go

durableEventTask := factory.NewDurableTask(
	create.StandaloneTask{
		Name: "durable-event",
	},
	func(ctx worker.DurableHatchetContext, input DurableEventInput) (*DurableEventOutput, error) {
		eventData, err := ctx.WaitForEvent("user:update", "")

		if err != nil {
			return nil, err
		}

		v := EventData{}
		err = eventData.Unmarshal(&v)

		if err != nil {
			return nil, err
		}

		return &DurableEventOutput{
			Data: v,
		}, nil
	},
	hatchet,
)

Durable event filters

Durable events can be filtered using CEL expressions. For example, to only receive user:update events for a specific user, you can use the following filter:

examples/durable_event/worker.py

res = await ctx.aio_wait_for(
    "event",
    UserEventCondition(
        event_key="user:update", expression="input.user_id == '1234'"
    ),
)

src/v1/examples/durable-event/workflow.ts

const res = ctx.waitFor({
  eventKey: 'user:update',
  expression: "input.userId == '1234'",
});

examples/v1/workflows/durable-event.go

eventData, err := ctx.WaitForEvent("user:update", "input.user_id == '1234'")

Durable Execution Durable Sleep