Skip to content

Commit b8f3430

Browse files
feat(durable): wait for callbacks with ctx.wait_for_callback (#400)
* chore(openapi): sync the durable callback contract Co-authored-by: Cursor <cursoragent@cursor.com> * feat(durable): wait for callbacks with ctx.wait_for_callback Co-authored-by: Cursor <cursoragent@cursor.com> * chore(openapi): sync the per-token callback rate limit Co-authored-by: Cursor <cursoragent@cursor.com> * fix(durable): refuse an async callback submitter The runtime calls the submitter on a thread without an event loop, and its result was discarded, so an async def submitter returned a coroutine that never ran: the wait opened and nothing received the token. Refuse an async def submitter with a TypeError before the callback opens, and fail the hand-off when a submitter returns any other awaitable, closing a coroutine so it does not warn later. The submitter stays typed as returning object, as in the JavaScript SDK, so one written as an expression that returns what it started still type-checks. The token fixture now reads as the fake it is. Co-authored-by: Cursor <cursoragent@cursor.com> * chore(openapi): sync the callback registration's operation id Co-authored-by: Cursor <cursoragent@cursor.com> * fix(durable): name the callback's operation when a replay registers it Co-authored-by: Cursor <cursoragent@cursor.com> --------- Co-authored-by: Cursor <cursoragent@cursor.com>
1 parent ec49a8d commit b8f3430

66 files changed

Lines changed: 6407 additions & 84 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎README.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -388,6 +388,7 @@ the clock or a random value.
388388
| `ctx.wait_until(check, options, name=None)` | Polls your own state until `options.until` holds, suspending between checks. `options.initial_state` is required. |
389389
| `ctx.wait_for_approval(name, title=..., description=None, details=None, timeout=None)` | Suspends until a person approves or denies, and returns an `ApprovalDecision`. A timeout returns `status="expired"` rather than raising. |
390390
| `ctx.wait_for_message(name, timeout=None)` | Suspends until a message with that name is sent to the execution, and returns a `ReceivedMessage`, or `None` once the timeout passes. |
391+
| `ctx.wait_for_callback(name, submitter, timeout=None, heartbeat_timeout=None)` | Hands a token to another system and suspends until it succeeds or fails the callback. Returns the result, or `None` once a timeout passes; a failure raises `CallbackFailedError`. |
391392
| `ctx.publish(name, data=None)` | Records a message for applications to list or receive live, and returns a `PublishedMessage` with `id` and `seq`. |
392393
| `ctx.map(items, fn, name=None, options=None)` | Runs the same work over every item, each in its own child context. |
393394
| `ctx.parallel(branches, name=None, options=None)` | Runs independent branches at the same time. |

‎docs/functions.md‎

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,55 @@ def handler(event, ctx):
209209

210210
Names follow the approval rules: up to 237 printable ASCII characters, checked before anything is recorded, as is the timeout. Both need `VOLCANO_PLATFORM_API_URL`. A wait Volcano does not answer, cannot see yet, throttles, or fails on is retried for the same 30 seconds an approval gets, then raises `CallbackSubmitterError`. A wait Volcano refuses, for example with `too_many_message_waits` when 100 are already open, raises it at once. A refused publish fails its step without a retry, which fails the execution unless the handler catches it.
211211

212+
## Wait for another system with a callback token
213+
214+
`ctx.wait_for_callback()` hands a token to another system and pauses the execution until that system answers with it. Use it when the work happens somewhere you do not run code, such as a payment provider, a job queue, or a partner's API, and that system can call back once it is done:
215+
216+
```python
217+
from volcano_sdk.durable_authoring import CallbackFailedError, durable
218+
219+
220+
@durable
221+
def handler(event, ctx):
222+
try:
223+
payment = ctx.wait_for_callback(
224+
"payment",
225+
lambda token: payments.start(
226+
order_id=event["order_id"], callback_token=token
227+
),
228+
timeout="1h",
229+
heartbeat_timeout="10m",
230+
)
231+
except CallbackFailedError as error:
232+
return {"status": "declined", "reason": error.type, "message": error.message}
233+
234+
if payment is None:
235+
return {"status": "abandoned"}
236+
return ctx.step(
237+
"fulfil", lambda scope: fulfil(event["order_id"], payment["charge_id"])
238+
)
239+
```
240+
241+
The submitter receives a `vcb_` token. The system holding it answers with the token alone and needs no Volcano credential:
242+
243+
```bash
244+
curl -X POST "https://api.volcano.dev/durable-callbacks/succeed" \
245+
-H 'Content-Type: application/json' \
246+
-d '{"token":"vcb_...","result":{"charge_id":"ch_1"}}'
247+
248+
curl -X POST "https://api.volcano.dev/durable-callbacks/fail" \
249+
-H 'Content-Type: application/json' \
250+
-d '{"token":"vcb_...","error":{"type":"CardDeclined","message":"Insufficient funds"}}'
251+
252+
curl -X POST "https://api.volcano.dev/durable-callbacks/heartbeat" \
253+
-H 'Content-Type: application/json' \
254+
-d '{"token":"vcb_..."}'
255+
```
256+
257+
`wait_for_callback(name, submitter, timeout=None, heartbeat_timeout=None)` returns the JSON value the token was succeeded with, or `None` once `timeout` passes, or `heartbeat_timeout` passes without a heartbeat. A result is never `None`, so the two cannot be confused. A failure raises `CallbackFailedError` with the `type` and `message` it was failed with. A token takes one answer: a second gets `409 callback_completed`, and one after the wait timed out or the execution ended gets `410`. It works only for this wait, so treat it as a secret.
258+
259+
The submitter runs inside a checkpointed step that is not retried, so a replay does not run it again. An attempt interrupted before the step is recorded can run it twice, with the same token both times, so make the hand-off idempotent; if the token was answered or the wait timed out in between, it is not run again. When it raises, `wait_for_callback()` raises the runtime's `CallbackSubmitterError`. It must be synchronous: an `async def` submitter raises `TypeError` before the callback opens, and one that returns an awaitable fails the wait instead of handing off. Getting the token needs `VOLCANO_PLATFORM_API_URL` and is retried for the same 30 seconds an approval gets. See [Durable callbacks](/platform/functions/durable-callbacks) for every request, response, and limit.
260+
212261
## Decide approvals from a backend
213262

214263
`client.durable.approvals` lists, reads, and decides a project's approvals:
@@ -322,4 +371,6 @@ Approvals work the same locally. Decide one with `volcano durable approvals appr
322371

323372
Messages work the same locally too. Send one with `volcano durable messages send`, or point the client at the local server. Message wait timeouts run in real time.
324373

374+
Callback tokens work locally as well: send the same requests to the local server. A callback's timeout and heartbeat timeout run in real time.
375+
325376
Running a decorated handler directly in a Python process still needs the optional test runtime: `python -m pip install 'volcano-sdk-python[durable]'`.

0 commit comments

Comments
 (0)