# Durable Streams

> **Warning:** Durable streams are in alpha and may change in future releases. To enable them
>   for your tenant, <a href="mailto:support@hatchet.run">contact us</a>.

> **Info:** Looking for the legacy run-based streaming API? See{" "}
>   <a href="/v1/streaming">Streaming (legacy)</a>.

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.

> **Info:** Messages of up to 64MB are supported. However, messages under 4MB are
>   significantly more performant. Slower performance should be expected when
>   publishing/consuming larger payloads.

#### Python

```python
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}
```

#### Typescript

```typescript
export type ProgressInput = {
  topic: string;
};

export const progressTask = hatchet.task({
  name: 'durable-streams-progress',
  fn: async (input: ProgressInput) => {
    const total = 5;

    for (let step = 1; step <= total; step += 1) {
      await sleep(500);
      await hatchet.streams.publish(input.topic, JSON.stringify({ step, total }));
    }

    return { steps: total };
  },
});
```

#### Go

```go
return client.NewStandaloneTask("durable-streams-progress", func(ctx hatchet.Context, input ProgressInput) (*ProgressOutput, error) {
	total := 5

	for step := 1; step <= total; step++ {
		time.Sleep(500 * time.Millisecond)

		message, err := json.Marshal(ProgressMessage{Step: step, Total: total})
		if err != nil {
			return nil, err
		}

		if err := client.Streams().Publish(ctx, input.Topic, message); err != nil {
			return nil, err
		}
	}

	return &ProgressOutput{Steps: total}, nil
})
```

#### Ruby

```ruby
DURABLE_STREAMS_PROGRESS_TASK = HATCHET.task(name: "durable-streams-progress") do |input, _ctx|
  total = 5

  (1..total).each do |step|
    sleep 0.5
    HATCHET.streams.publish(input["topic"], JSON.generate({ "step" => step, "total" => total }))
  end

  { "steps" => total }
end
```

## 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.

#### Python

```python
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())
```

#### Typescript

```typescript
let cursor: string | undefined;

for await (const event of hatchet.streams.subscribe(topic)) {
  console.log(decode(event.payload));
  ({ cursor } = event);
  break;
}

// picks up right after the last event read, even from another process
for await (const event of hatchet.streams.subscribe(topic, { cursor })) {
  console.log(decode(event.payload));
}
```

#### Go

```go
var cursor string

for event, err := range client.Streams().Subscribe(ctx, topic) {
	if err != nil {
		return err
	}

	fmt.Println(string(event.Payload))
	cursor = event.Cursor
	break
}

for event, err := range client.Streams().Subscribe(ctx, topic, features.WithStreamCursor(cursor)) {
	if err != nil {
		return err
	}

	fmt.Println(string(event.Payload))
}
```

#### Ruby

```ruby
cursor = nil

HATCHET.streams.subscribe(topic) do |event|
  puts event.payload
  cursor = event.cursor
  break
end

HATCHET.streams.subscribe(topic, cursor: cursor) do |event|
  puts event.payload
end
```

## 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.

#### Python

```python
await hatchet.streams.aio_reset("orders")

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

#### Typescript

```typescript
await hatchet.streams.reset('orders');

// starts after the reset; readers that were already iterating aren't affected
for await (const event of hatchet.streams.subscribe('orders')) {
  console.log(decode(event.payload));
}
```

#### Go

```go
if err := client.Streams().Reset(ctx, "orders"); err != nil {
	return err
}

for event, err := range client.Streams().Subscribe(ctx, "orders") {
	if err != nil {
		return err
	}

	fmt.Println(string(event.Payload))
}
```

#### Ruby

```ruby
HATCHET.streams.reset("orders")

HATCHET.streams.subscribe("orders") do |event|
  puts event.payload
end
```

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.

#### Python

```python
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())
```

#### Typescript

```typescript
let cursor: string | undefined;

for await (const event of hatchet.streams.subscribe('orders')) {
  ({ cursor } = event);
  break;
}

await hatchet.streams.reset('orders', { cursor });

// starts right after the event the cursor was read from
for await (const event of hatchet.streams.subscribe('orders')) {
  console.log(decode(event.payload));
}
```

#### Go

```go
var cursor string

for event, err := range client.Streams().Subscribe(ctx, "orders") {
	if err != nil {
		return err
	}

	cursor = event.Cursor
	break
}

if err := client.Streams().Reset(ctx, "orders", features.WithStreamCursor(cursor)); err != nil {
	return err
}

for event, err := range client.Streams().Subscribe(ctx, "orders") {
	if err != nil {
		return err
	}

	fmt.Println(string(event.Payload))
}
```

#### Ruby

```ruby
cursor = nil

HATCHET.streams.subscribe("orders") do |event|
  cursor = event.cursor
  break
end

HATCHET.streams.reset("orders", cursor: cursor)

HATCHET.streams.subscribe("orders") do |event|
  puts event.payload
end
```

## 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.

#### Python

```python
@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}
```

#### Typescript

```typescript
export const retryingProgressTask = hatchet.task({
  name: 'durable-streams-retrying-progress',
  retries: 1,
  fn: async (input: ProgressInput, ctx) => {
    // readers that subscribe from here on skip a failed attempt's messages
    await hatchet.streams.reset(input.topic);

    for (let chunk = 0; chunk < 3; chunk += 1) {
      await hatchet.streams.publish(
        input.topic,
        JSON.stringify({ attempt: ctx.retryCount(), chunk })
      );
    }

    if (ctx.retryCount() === 0) {
      throw new Error('intentional failure before the final message');
    }

    await hatchet.streams.publish(input.topic, JSON.stringify({ done: true }));

    return { chunks: 3 };
  },
});
```

#### Go

```go
return client.NewStandaloneTask("durable-streams-retrying-progress", func(ctx hatchet.Context, input ProgressInput) (*RetryingProgressOutput, error) {
	if err := client.Streams().Reset(ctx, input.Topic); err != nil {
		return nil, err
	}

	for chunk := 0; chunk < 3; chunk++ {
		attempt := ctx.RetryCount()

		message, err := json.Marshal(RetryingProgressMessage{Attempt: &attempt, Chunk: &chunk})
		if err != nil {
			return nil, err
		}

		if err := client.Streams().Publish(ctx, input.Topic, message); err != nil {
			return nil, err
		}
	}

	if ctx.RetryCount() == 0 {
		return nil, errors.New("intentional failure before the final message")
	}

	message, err := json.Marshal(RetryingProgressMessage{Done: true})
	if err != nil {
		return nil, err
	}

	if err := client.Streams().Publish(ctx, input.Topic, message); err != nil {
		return nil, err
	}

	return &RetryingProgressOutput{Chunks: 3}, nil
}, hatchet.WithRetries(1))
```

#### Ruby

```ruby
DURABLE_STREAMS_RETRYING_PROGRESS_TASK = HATCHET.task(name: "durable-streams-retrying-progress", retries: 1) do |input, ctx|
  HATCHET.streams.reset(input["topic"])

  3.times do |chunk|
    HATCHET.streams.publish(input["topic"], JSON.generate({ "attempt" => ctx.retry_count, "chunk" => chunk }))
  end

  raise "intentional failure before the final message" if ctx.retry_count.zero?

  HATCHET.streams.publish(input["topic"], JSON.generate({ "done" => true }))

  { "chunks" => 3 }
end
```