Introducing Taskturbine

Through most of 2025, I had been working on Sentry’s task platform. That system operates with pretty high throughput (~200,000 tasks/second), so I’m familiar with the challenges in operating big systems in this space. In November 2025, Armin published a post on Absurd and I was intrigued by the operational simplicity that ‘just Postgres’ could provide. After seeing Armin’s work, I wanted to explore this space further. I thought it would be a good opportunity to learn more Rust, get experience using Py03 to build a Python package, optimize some SQL queries, and potentially also build extensions for PHP and JavaScript with a common Rust library. Using the schema, and general API from Absurd as a starting point, I wanted to build a ‘core’ library in Rust, and then extend that core library with SDKs for a variety of languages starting with Rust and Python. The end result is taskturbine

Durable tasks

Durable tasks (or workflows) are a powerful tool for building reliable application logic that can survive crashes, application restarts, and network failures without losing or duplicating work. Durable tasks are composed of ‘steps’. As steps are completed ‘checkpoints’ are stored with the result of each step. If a task fails or is interrupted, it can continue where it left off on the next attempt by skipping any steps that already checkpoint state stored. By thoughtfully structuring your application logic you can build fault-tolerant applications with greater reliability and consistency than traditional fire-and-forget task systems.

Getting started with Taskturbine in Python

Once you’ve installed the taskturbine package, you can create an application. The `app` needs to be an importable from a module as the Python worker uses multiprocessing:

Show Plain Text
  1. from taskturbine import Config, TaskturbineApp
  2.  
  3. config = Config(
  4.     app_module="demoapp.tasks:app",
  5.     database_url="postgres://app:password@localhost:5432/demoapp",
  6.     usecase="demoappp"
  7. )
  8.  
  9. # Build an app that can have tasks attached.
  10. app = TaskturbineApp(config=config)
  11.  
  12. # Install/update schema with migrations
  13. app.update_schema()

The update_schema() method will create a Postgres schema where taskturbine will store all of its tables. With an application created, you can define your first task:

Show Plain Text
  1. from taskturbine import TaskContext
  2.  
  3. @app.register_task("hello-world")
  4. def hello_world(ctx: TaskContext) -> None:
  5.     print('Hello {ctx.params["name"]}')

We can spawn tasks using our app as well:

Show Plain Text
  1. app.spawn_task("hello-world", {"name": "Mark"}, retry_seconds=30)

To run our tasks, we need to create a Worker and run it:

Show Plain Text
  1. worker = app.worker("worker-1")
  2. worker.run()

A hello-world example isn’t overly interesting, but if we were building a workflow like a user deletion or account cancellation we run into more interesting problems like coordinating cleanup operations across multiple systems:

Show Plain Text
  1. @app.register_task("account-cancellation")
  2. def account_cancellation(ctx: TaskContext) -> None:
  3.  
  4.     # Define all the steps for our cleanup operation
  5.     @ctx.step("unsubscribe-mailing-list")
  6.     def unsubscribe_mailing_list(user_id: int) -> dict[str, str]:
  7.         logger.info("starting unsubscribe")
  8.         subscription_id = mailing_list_service.unsubscribe_user(user_id)
  9.         return {"subscription_id": subscription_id}
  10.  
  11.     @ctx.step("remove-data-from-blob-storage")
  12.     def cleanup_blob_storage(user_id: int) -> dict[str, list[str]]:
  13.         logger.info("starting blob cleanup")
  14.         removed = []
  15.         for blob_id in blob_storage.list_for_user(user_id):
  16.             blob_storage.remove(blob_id)
  17.             removed.append(blob_id)
  18.         return {"blobs": removed}
  19.  
  20.     @ctx.step("remove-database-data")
  21.     def cleanup_database(user_id: int) -> bool:
  22.         logger.info("starting db cleanup")
  23.         cleanup_service.remove_user(user_id)
  24.         return true
  25.  
  26.     # Run the steps in the order you want
  27.     user_id = ctx.params["user_id"]
  28.  
  29.     subscription = unsubscribe_mailing_list(user_id)
  30.     logger.info(f"removed mailing list data {subscription}")
  31.  
  32.     blob_ids = cleanup_blob_storage(user_id)
  33.     logger.info(f"removed blobs {blob_ids}")
  34.  
  35.     cleanup_database(user_id)
  36.     logger.info("completed user cleanup")
  37.  
  38.     return None

Each step will store its results as a ‘checkpoint’. Should the task or one of its steps encounter an error and crash, taskturbine can resume from the most recently completed step on the next attempt. Logic outside of steps is not durable, and will be re-run each time the task executes.

External events and waits

Like many other durable workflow systems, taskturbine also includes support for external events, and sleeps. External events are a great way to incorporate webhooks. For example, in a checkout flow you may need to wait for a payment intent to be completed before you mark the order as paid and make update the order status.

Show Plain Text
  1. @app.register_task("process-checkout-order")
  2. def process_checkout_order(ctx: TaskContext) -> None:
  3.  
  4.     @ctx.step(name="reserve-inventory")
  5.     def reserve_inventory(order: Order) -> str:
  6.         reservation_id = inventory_service.reserve_for_order(order)
  7.         return {"inventory_reservation": reservation_id}
  8.  
  9.     @ctx.step(name="release-inventory")
  10.     def release_inventory(reservation_id: str) -> dict[str, Any]:
  11.         reservation_id = inventory_service.release_reservation(reservation_id)
  12.         return {"released": True}
  13.  
  14.     @ctx.step(name="cancel-order")
  15.     def cancel_order(order: Order) -> dict[str, Any]:
  16.         order_service.cancel(order)
  17.         return {"cancelled": True}
  18.  
  19.     @ctx.step(name="process-order")
  20.     def process_order(order: Order) -> str:
  21.         status = update_order_status(order, OrderStatus.PAID)
  22.         return {"status": status}
  23.  
  24.     order = order_service.get(ctx.params["order_id"])
  25.     reservation = reserve_inventory(order)
  26.  
  27.     # Wait for the payment to be completed (via async webhook)
  28.     # Each event must have a unique id
  29.     payment = ctx.await_event(f"order:payment:{order_id}", timeout=timedelta(minutes=10))
  30.  
  31.     # Once our payment event has been received, the task will re-run with the data
  32.     # captured by the event.
  33.     if payment["state"] == PaymentStatus.COMPLETE:
  34.         process_order(order)
  35.     else:
  36.         release_inventory(reservation["reservation"])
  37.         cancel_order(order)

In the HTTP endpoint that handles our payment provider webhooks we could capture a task event with:

Show Plain Text
  1. payment_state = PaymentStatus.from_webhook(webhook_data["payment"])
  2. app.emit_event(f"order:payment:{order_id}", {"state": payment_state})

When events are captured, any tasks waiting on the event will attempt another run.

What’s next?

This project is currently in a development preview stage. I still have to finish off writing documentation, and doing more extensive testing for performance and failure mode scenarios. I’m also planning on adding support for PHP and Javascript as there are good Rust bindings for each of these languages. The CLI tool is also very early on and could use a more effort to polish it off and extend the number of usage scenarios that it can cover. Lastly, I’d love to make a simple monitoring dashboard that gives a web-based GUI that lets an operator view the current state of tasks, and cancel/retry/start tasks. If this project is something that you’re interested in contributing to, open a discussion or issue on GitHub.

Comments

There are no comments, be the first!

Have your say: