feat(durable-streams): Go, Python, Ruby SDKs by juliusgeo · Pull Request #5348 · hatchet-dev/hatchet

juliusgeo · 2026-10-27T19:30:41Z

Description

Fixes # (issue)

Type of change

Checklist

Changes have been:

Documented (where applicable)
Added to CHANGELOG (where applicable) -- see Keep a Changelog

🤖 AI Disclosure

I acknowledge that an LLM was used in the creation of this Pull Request, in accordance with Hatchet's AI_POLICY.md.

Details: [e.g. generating tests, writing docs]

vercel · 2026-10-27T19:30:45Z

The latest updates on your projects. Learn more about Vercel for GitHub.

Project	Deployment	Actions	Updated
hatchet-docs	Ready	Preview	Oct 29, 2026 4:17pm UTC

mrkaye97

left a couple comments on the public API, I think if we make this async-only in Python it'll be dramatically simpler. I wish we'd made more stuff async-only early on...

mrkaye97 · 2026-10-27T20:40:19Z

+def _iterate_in_thread(agen: AsyncGenerator[T, None]) -> Generator[T, None, None]:
+    """Iterates an async generator from sync code on an event loop of its own."""
+    loop = asyncio.new_event_loop()
+    thread = threading.Thread(target=loop.run_forever, daemon=True)
+    thread.start()
+
+    reading: asyncio.Task[T] | None = None
+
+    async def next_item() -> T:
+        nonlocal reading
+        reading = asyncio.current_task()
+        return await agen.__anext__()
+
+    async def close() -> None:
+        ## interrupted mid-read: a running generator can't be closed until the read unwinds
+        if reading is not None and not reading.done():
+            reading.cancel()
+            await asyncio.wait([reading])
+        await agen.aclose()
+
+    try:
+        while True:
+            try:
+                item = asyncio.run_coroutine_threadsafe(next_item(), loop).result()
+            except StopAsyncIteration:
+                return
+            yield item
+    finally:
+        asyncio.run_coroutine_threadsafe(close(), loop).result()
+        loop.call_soon_threadsafe(loop.stop)
+        thread.join()


I this is probably a bad idea - we've tried this before. IMO, we should probably make this async-only and not suffer from event loop blocked nightmares. WDYT?

mrkaye97 · 2026-10-27T20:41:33Z

+        await asyncio.wrap_future(
+            self._enqueue_publish(topic, message, namespace, batch_timeout)
+        )


it'd be ideal to have an async version of this because doing it this way will cause a ton of thread / event loop churn

mrkaye97 · 2026-10-27T20:42:42Z

+        """
+        return _iterate_in_thread(self.aio_events(topic, namespace, cursor))
+
+    async def aio_events(


should we call this consume or subscribe? I tend to believe that function names should be verbs, and those both describe this better than events does IMO

greptile-apps · 2026-10-29T15:14:20Z

Confidence Score: 2/5

[Medium impact] The PR adds durable-stream workflows across SDK clients and serverless task code.

The PR is not ready to merge because the earlier Go ordering, Python cancellation, and Ruby 3.1 failures remain.

Summary

Adds durable stream clients for Go, Python, and Ruby.

Go, Python, and Ruby callers can publish and read durable topic messages.
TypeScript callers use subscribe() to read streams, and serverless tasks can read through the operator socket.

_{Reviews (4) · Last reviewed commit: "generate" · Reviewed by Greptile}

greptile-apps · 2026-10-29T15:14:24Z

+	if c.streams == nil {
+		tenantId := c.legacyClient.TenantId()
+		c.streams = features.NewStreamsClient(c.legacyClient.Streams(), c.legacyClient.API(), tenantId)


Shared publishes can lose order

Two worker goroutines calling Streams() for the first time can create different StreamsClient instances. Each instance has its own queues and producer IDs, so publishes to the same topic can run independently and arrive out of order. Create the shared client during setup or guard its creation with sync.Once.

greptile-apps · 2026-10-29T15:14:24Z

+            ## a failed upload fails only its own publish, as the rest never depended on it
+            try:
+                ref = await p.uploaded
+            except Exception as e:
+                p.settle(e)
+                continue
+
+            entries.append(streams_protos.StreamPublishEntry(payload_ref=ref))
+            sent.append(p)
+
+        if entries:
+            await self._send_entries(sent, entries)


Cancelled messages still get published

_send_batch does not check cancellation again after waiting for uploads. If a caller cancels while an upload in the selected batch is running, the drain still sends that message when the upload finishes. This breaks the documented promise to drop publishes cancelled before their request goes out. Remove cancelled entries immediately before sending, including entries collected before a later upload wait.

greptile-apps · 2026-10-29T15:14:24Z

+      # an upload's ref, which a batch carries in place of its payload, rounded up
+      PAYLOAD_REF_BYTES = 256
+
+      Config = Struct.new(:payload_request_attempts, :resubscribe_min_delay_seconds, :resubscribe_max_delay_seconds) do


Ruby 3.1 recovery fails

Config lacks keyword_init: true, but the gem supports Ruby 3.1. On that version, the bare super passes the keyword arguments as one positional hash. payload_request_attempts becomes that hash, and the delay fields remain nil. Reconnects then fail when calculating a delay, and failed payload requests cannot retry. Add keyword_init: true or pass the three values positionally.

Suggested change

Config = Struct.new(:payload_request_attempts, :resubscribe_min_delay_seconds, :resubscribe_max_delay_seconds) do

Config = Struct.new(:payload_request_attempts, :resubscribe_min_delay_seconds, :resubscribe_max_delay_seconds, keyword_init: true) do

greptile-apps · 2026-10-29T15:14:24Z

+  deadline = Time.now + 10
+
+  HATCHET.streams.subscribe(topic) do |event|
+    puts event.payload
+    break if Time.now > deadline


Quiet reads ignore the deadline

stop_reading checks its ten-second deadline only after receiving a message. On a quiet topic, it keeps waiting until another message arrives or the server closes the idle stream. Someone copying this example will not get the advertised timeout. Use a timeout that interrupts the blocked read instead of checking time inside the block.

greptile-apps · 2026-10-29T15:14:24Z

+	hatchet "github.com/hatchet-dev/hatchet/sdks/go"
+)
+
+type ProgressInput struct {


Exported examples lack required comments

The new exported Go types and functions have no godoc comments. The repository requires comments on exported types and functions, so this must be fixed before merging. Add them to ProgressInput, ProgressTask, and the other exported declarations in the shared example, then regenerate its mirror.

Context Used: Comment and logging conventions for the Hatchet codebase (source)

Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!

mrkaye97 · 2026-10-29T15:17:41Z

+    def events(
+        self, topic: str, namespace: str = "", cursor: str | None = None
+    ) -> AsyncGenerator[StreamEvent, None]:
+        """
+        An alias of `subscribe`, which replaces it; `events` will be removed in a future release.
+
+        :param topic: The topic to read from.
+        :param namespace: The topic's namespace; the empty namespace by default.
+        :param cursor: A cursor read off a previous event, to resume right after it.
+
+        :return: An async iterator of the topic's events.
+        """
+        return self.subscribe(topic, namespace, cursor)


prob don't need this since it was never released

true lol, forgot about that

mrkaye97 · 2026-10-29T15:18:28Z

+                            payload = (
+                                await asyncio.to_thread(
+                                    self._fetch_payload, entry.payload_ref
+                                )


ideally we'd be able to do this as native-async without needing to_thread. just worried this is going to cause blocked event loops really easily

we have to do this because the REST API isn't asynchronous :/ unless I'm missing something

argh alright - is this just for large payloads?

yep, <4mb payloads go through gRPC path

initial commit

b0e3c92

github-actions Bot added documentation Improvements or additions to documentation github_actions Pull requests that update GitHub Actions code sdk-go Related to the Go SDK sdk-py Related to the Python sdk engine Related to the core Hatchet engine sdk-ruby Related to the Ruby SDK labels Oct 27, 2026

vercel Bot deployed to Preview October 27, 2026 19:32

View deployment

merge

467c645

vercel Bot deployed to Preview October 27, 2026 20:30

View deployment

mrkaye97 reviewed Oct 27, 2026

View reviewed changes

get rid of sync in python

a864906

vercel Bot deployed to Preview October 28, 2026 15:14

View deployment

add subscribe alias

7414e35

github-actions Bot added the sdk-ts Related to the Typescript SDK label Oct 28, 2026

vercel Bot deployed to Preview October 28, 2026 16:29

View deployment

use asyncio rather than threads

4050af7

vercel Bot deployed to Preview October 28, 2026 19:05

View deployment

another refactor

036935d

vercel Bot deployed to Preview October 28, 2026 20:36

View deployment

merge

8509c49

juliusgeo marked this pull request as ready for review October 29, 2026 15:07

vercel Bot deployed to Preview October 29, 2026 15:08

View deployment

greptile-apps Bot reviewed Oct 29, 2026

View reviewed changes

mrkaye97 reviewed Oct 29, 2026

View reviewed changes

fix some issues

857f73e

mrkaye97 approved these changes Oct 29, 2026

View reviewed changes

vercel Bot deployed to Preview October 29, 2026 15:29

View deployment

fix ruby

539a475

vercel Bot deployed to Preview October 29, 2026 15:40

View deployment

generate

efead67

vercel Bot deployed to Preview October 29, 2026 16:17

View deployment

juliusgeo merged commit 7b086a9 into main Oct 29, 2026
65 of 66 checks passed

juliusgeo deleted the durable_streams_other_sdks branch October 29, 2026 18:59

	Config = Struct.new(:payload_request_attempts, :resubscribe_min_delay_seconds, :resubscribe_max_delay_seconds) do
	Config = Struct.new(:payload_request_attempts, :resubscribe_min_delay_seconds, :resubscribe_max_delay_seconds, keyword_init: true) do

Conversation

juliusgeo commented Oct 27, 2026

Description

Type of change

Checklist

Uh oh!

vercel Bot commented Oct 27, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mrkaye97 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot commented Oct 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Confidence Score: 2/5

Summary

Uh oh!

greptile-apps Bot Oct 29, 2026

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Oct 29, 2026

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Oct 29, 2026

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Oct 29, 2026

Choose a reason for hiding this comment

Uh oh!

greptile-apps Bot Oct 29, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

This branch was successfully deployed

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

vercel Bot commented Oct 27, 2026 •

edited

Loading

greptile-apps Bot commented Oct 29, 2026 •

edited

Loading