Durable Streams

Durable streams allow you to stream results from a Hatchet task to a consumer in real-time, with durability and ordering guarantees. Because of its durable nature, it can also be used to stream results across multiple Hatchet tasks, as well as consume messages well after a workflow run has started.

Guarantees

Durable streams have the following guarantees:

1. All messages from a single publisher will be persisted, and streamed, in the exact order they were published. This guarantee does not extend across multiple publishers.
2. All messages in a durable stream topic will be persisted, and consumers can resume from any message within the topic's retention.

Publishing

Publishing messages to a durable stream topic is as simple as calling the publish method. It takes the topic name, a namespace, and the message payload.

class ProgressInput(BaseModel):    topic: str@hatchet.task(input_validator=ProgressInput)async def durable_streams_progress(    input: ProgressInput, ctx: Context) -> dict[str, int]:    total = 5    for step in range(1, total + 1):        await asyncio.sleep(0.5)        await hatchet.streams.publish(            input.topic, json.dumps({"step": step, "total": total})        )    return {"steps": total}

class ProgressInput(BaseModel):
    topic: str


@hatchet.task(input_validator=ProgressInput)
async def durable_streams_progress(
    input: ProgressInput, ctx: Context
) -> dict[str, int]:
    total = 5

    for step in range(1, total + 1):
        await asyncio.sleep(0.5)
        await hatchet.streams.publish(
            input.topic, json.dumps({"step": step, "total": total})
        )

    return {"steps": total}

Consuming

Consumption, by default, starts from the first message published to the topic. Each event object that is returned will contain both the message payload, as well as a cursor. This cursor is opaque, but can be saved, and allows resumption from that point in the stream. For example, if you wanted to make sure that a stream was only ever reading unprocessed messages, you could save the cursor, and use that to resume.

cursor = Noneasync for event in hatchet.streams.subscribe(topic):    print(event.payload.decode())    cursor = event.cursor    breakasync for event in hatchet.streams.subscribe(topic, cursor=cursor):    print(event.payload.decode())

cursor = None

async for event in hatchet.streams.subscribe(topic):
    print(event.payload.decode())
    cursor = event.cursor
    break

async for event in hatchet.streams.subscribe(topic, cursor=cursor):
    print(event.payload.decode())

Resetting a Topic

Resetting a topic sets a new high watermark, from which all new subscriptions will start. Calling reset does not change the messages that existing subscriptions, or subscriptions from a cursor will receive.

await hatchet.streams.aio_reset("orders")async for event in hatchet.streams.subscribe("orders"):    print(event.payload.decode())

await hatchet.streams.aio_reset("orders")

async for event in hatchet.streams.subscribe("orders"):
    print(event.payload.decode())

You can also supply a cursor to Reset, which will set the high watermark to the cursor supplied rather than the most recently published message.

cursor = Noneasync for event in hatchet.streams.subscribe("orders"):    cursor = event.cursor    breakawait hatchet.streams.aio_reset("orders", cursor=cursor)async for event in hatchet.streams.subscribe("orders"):    print(event.payload.decode())

cursor = None

async for event in hatchet.streams.subscribe("orders"):
    cursor = event.cursor
    break

await hatchet.streams.aio_reset("orders", cursor=cursor)

async for event in hatchet.streams.subscribe("orders"):
    print(event.payload.decode())

Resetting on Retry

Resetting a topic is useful for when there is a singular topic that is streaming messages from a task run with a retry policy. On successive retries, rather than using a separate topic for each individual run, just reset the topic before publishing, and new subscribers will automatically read only the messages from the most recent retry.

@hatchet.task(input_validator=ProgressInput, retries=1)async def durable_streams_retrying_progress(    input: ProgressInput, ctx: Context) -> dict[str, int]:    await hatchet.streams.aio_reset(input.topic)    for chunk in range(3):        await hatchet.streams.publish(            input.topic, json.dumps({"attempt": ctx.retry_count, "chunk": chunk})        )    if ctx.retry_count == 0:        raise RuntimeError("intentional failure before the final message")    await hatchet.streams.publish(input.topic, json.dumps({"done": True}))    return {"chunks": 3}

@hatchet.task(input_validator=ProgressInput, retries=1)
async def durable_streams_retrying_progress(
    input: ProgressInput, ctx: Context
) -> dict[str, int]:
    await hatchet.streams.aio_reset(input.topic)

    for chunk in range(3):
        await hatchet.streams.publish(
            input.topic, json.dumps({"attempt": ctx.retry_count, "chunk": chunk})
        )

    if ctx.retry_count == 0:
        raise RuntimeError("intentional failure before the final message")

    await hatchet.streams.publish(input.topic, json.dumps({"done": True}))

    return {"chunks": 3}