# restate API reference

> The `restate` package, from restate 0.4.1, as published on mojoshelf (commit e391b04b0de1), by `mojo doc` from Mojo 1.2.0.

`restate` — durable execution flows in Mojo, via Restate (restate.dev).

The Rust shim (ffi/ -> librestatemojo.dylib) embeds the Restate Rust SDK: it
owns the HTTP/2 endpoint, the event loop, and the journal. Mojo owns the
business logic, driven by a synchronous loop:

    from restate import App

    def main() raises:
        var stats = Stats(0)
        _ = App.run[Stats, __functions_in_module()]("Counter", stats)

That loop is single-threaded, and a handler that `call`s another handler served
by the same process therefore deadlocks. `App.serve` is the concurrent form —
the same loop on N threads, via threads-mojo's `TypedPool`:

    from restate import App, Invocation, OpaquePtr

    def handle(app: App, inv: Invocation, worker: Int, ctx: OpaquePtr) raises:
        app.complete(inv, "ok")

    def main() raises:
        var app = App("Counter", ["add", "get"], object=True)
        _ = app.serve[handle](num_workers=4)   # until app.stop()

With two or more workers one thread can wait for a call while another runs the
callee, so self-calls complete. See `HandlerFn` for the handler's obligations
and `App.stop` for how a worker parked in `next()` is released.

Semantics: on retry or resume, Restate re-invokes the handler and replays the
journal — ctx operations (`sleep`, `call`, `get_state`, `run_*`, ...) return
their journaled results instantly, so handler logic must be deterministic
between them; wrap non-deterministic work in `run_enter`/`run_exit`. When
Restate suspends an invocation (e.g. a long `sleep`), the pending operation
raises a suspension error: abandon the invocation and keep serving — Restate
re-delivers it later.

Payloads are raw bytes on the wire; the String helpers below treat them as
UTF-8 (use the `_bytes` variants for binary data).

## Structs

### `Invocation`

```mojo
struct Invocation
```

One in-flight invocation: which handler was called, the virtual-object key (empty for plain services), and the request payload.

Implements: `Copyable`, `Movable`

Fields:

- `handle: Int`
- `handler: String`
- `key: String`
- `id: String`
- `input: List[UInt8]`

#### `input_string`

```mojo
def input_string(self) -> String
```

### `App`

```mojo
struct App
```

A Restate endpoint serving one service with byte/string handlers, driven by `next()`.

Implements: `Movable`

Fields:

- `lib: OwnedDLHandle`

#### `__init__`

```mojo
def __init__(out self, service: String, handlers: List[String], object: Bool = True, workflow: Bool = False, host: String = "0.0.0.0", port: Int = Int(9080))
```

Start the endpoint (background thread in the shim). `object=True` registers a Virtual Object — keyed, with state and single-writer concurrency per key; `False` a plain (stateless) Service. `workflow=True` registers a Workflow: the handler named "run" is the workflow handler (runs once per key), the rest are shared and may use the promise operations to signal it.

#### `sleep_ms`

```mojo
def sleep_ms(self, inv: Invocation, millis: Int)
```

Durable sleep. Journaled; long sleeps suspend the invocation.

#### `get_state_bytes`

```mojo
def get_state_bytes(self, inv: Invocation, key: String) -> Optional[List[UInt8]]
```

#### `get_state`

```mojo
def get_state(self, inv: Invocation, key: String) -> Optional[String]
```

#### `get_state_int`

```mojo
def get_state_int(self, inv: Invocation, key: String, default: Int) -> Int
```

#### `set_state_bytes`

```mojo
def set_state_bytes(self, inv: Invocation, key: String, value: List[UInt8])
```

#### `set_state`

```mojo
def set_state(self, inv: Invocation, key: String, value: String)
```

#### `set_state_int`

```mojo
def set_state_int(self, inv: Invocation, key: String, value: Int)
```

#### `clear_state`

```mojo
def clear_state(self, inv: Invocation, key: String)
```

#### `call`

```mojo
def call(self, inv: Invocation, service: String, handler: String, payload: String, key: String = "") -> String
```

Durable request/response call to another handler. A non-empty `key` targets a virtual object; empty targets a plain service.

#### `send`

```mojo
def send(self, inv: Invocation, service: String, handler: String, payload: String, key: String = "", delay_ms: Int = Int(0))
```

Durable one-way message (optionally delayed).

#### `awakeable_create`

```mojo
def awakeable_create(self, inv: Invocation) -> String
```

Create an awakeable — a durable promise resolvable from outside (another service, or the ingress HTTP API). Returns its id: hand it to the outside world (inside `run_enter`/`run_exit` or via `call`), then `awakeable_await` it.

#### `awakeable_await`

```mojo
def awakeable_await(self, inv: Invocation, id: String) -> String
```

Await an awakeable created in this invocation; returns its resolved payload. Suspends if unresolved.

#### `awakeable_resolve`

```mojo
def awakeable_resolve(self, inv: Invocation, id: String, value: String)
```

Resolve another invocation's awakeable with a payload.

#### `awakeable_reject`

```mojo
def awakeable_reject(self, inv: Invocation, id: String, message: String)
```

Reject another invocation's awakeable with a terminal error.

#### `promise_await`

```mojo
def promise_await(self, inv: Invocation, name: String) -> String
```

Await a named workflow promise (workflow services only).

#### `promise_peek`

```mojo
def promise_peek(self, inv: Invocation, name: String) -> Optional[String]
```

Peek a named workflow promise: its value, or None if unresolved.

#### `promise_resolve`

```mojo
def promise_resolve(self, inv: Invocation, name: String, value: String)
```

Resolve a named workflow promise (from a shared handler).

#### `promise_reject`

```mojo
def promise_reject(self, inv: Invocation, name: String, message: String)
```

Reject a named workflow promise with a terminal error.

#### `cancel_invocation`

```mojo
def cancel_invocation(self, inv: Invocation, invocation_id: String)
```

Cancel another invocation by id.

#### `run_enter`

```mojo
def run_enter(self, inv: Invocation) -> Optional[String]
```

Enter a journaled side-effect block. Returns the journaled value on replay; None means: execute the side effect now, then close the block with `run_exit` (it succeeded) or `run_fail` (it did not).

Exactly one of those must follow. Leaving the block open makes the
invocation unreplayable: every later attempt raises "protocol error:
expected rst_run_exit".

#### `run_exit`

```mojo
def run_exit(self, inv: Invocation, value: String) -> String
```

Journal the side-effect result produced after `run_enter` returned None. Returns the journaled value.

#### `run_enter_policy`

```mojo
def run_enter_policy(self, inv: Invocation, initial_delay_ms: Int = Int(0), factor: Float64 = 0, max_delay_ms: Int = Int(0), max_attempts: Int = Int(0), max_duration_ms: Int = Int(0)) -> Optional[String]
```

`run_enter`, with a retry policy for this block's non-terminal failures.

Plain `run_enter` leaves retries to Restate's server-side invoker
policy, which retries indefinitely. Set any of these to bound it; 0
means "leave that knob alone", and `max_attempts` or `max_duration_ms`
of 0 mean unbounded.

When attempts or duration run out the SDK fails the block terminally,
which arrives here the same way `run_fail(terminal=True)` does — as a
raised error the handler can compensate for.

#### `step`

```mojo
def step[F: def() raises -> String](self, inv: Invocation, f: F, initial_delay_ms: Int = Int(0), factor: Float64 = 0, max_delay_ms: Int = Int(0), max_attempts: Int = Int(0), max_duration_ms: Int = Int(0)) -> String
```

One journaled step: replay it, or run `f` and journal what it returned.

def charge() raises {imm reservation_id} -> String:
        return charge_card(reservation_id)

    var charge_id = app.step(inv, charge)

`f` is passed by value with an explicit capture list, which keeps
what it captures alive until `step` returns. Its type is a parameter,
so each closure still gets its own monomorphised `step`.

This is `run_enter` / `run_exit` / `run_fail` with the protocol
handled. It matters less for brevity than for the fact that a block
cannot be left open: if `f` raises, the step is closed as a
non-terminal failure and Restate runs it again, which is what the
hand-written form has to remember to do every time.

The retry arguments are `run_enter_policy`'s, and bound the retries of
`f` itself. Passing none leaves them to Restate's invoker policy,
which does not give up.

A step that should fail *terminally* — a downstream that is not coming
back, where the handler compensates and finishes — wants the explicit
protocol instead: `run_enter`, then `run_fail(terminal=True)`.

#### `run_fail`

```mojo
def run_fail(self, inv: Invocation, message: String, terminal: Bool = True) -> Optional[String]
```

Abort the run block opened by `run_enter`, rather than closing it with `run_exit`.

Every `run_enter` that returns None must be closed by exactly one of
`run_exit` or `run_fail`. Leaving one open makes the invocation
unreplayable: the next attempt raises "protocol error: expected
rst_run_exit" and keeps doing so forever.

`terminal=True` journals the failure. The step is recorded as failed,
replay reproduces the failure instead of executing again, and this
raises so the handler can compensate and finish.

`terminal=False` journals nothing and lets Restate re-run the block
under its own retry policy. The return value is then the next
execute slot: `None` means "run the side effect again", and a value
means the journal answered in the meantime.

#### `complete`

```mojo
def complete(self, inv: Invocation, output: String)
```

Finish the invocation successfully with `output`.

#### `fail`

```mojo
def fail(self, inv: Invocation, message: String)
```

Finish the invocation with a terminal error (not retried).

#### `abandon`

```mojo
def abandon(self, inv: Invocation)
```

Drop an invocation without completing it (after suspension or an internal error). Restate retries/resumes it with journal replay.

#### `stop`

```mojo
def stop(self)
```

Unblock every `next()` — in this thread or any other — so driver loops can end.

Setting a flag in Mojo cannot reach a thread parked inside the shim's
blocking receiver, so this goes through the shim (`rst_stop`), which
sets a process-wide flag its receive loop rechecks. Idempotent, and
safe to call from inside a handler: that is how a service gives itself
a `shutdown` endpoint.

The endpoint itself keeps listening — this ends the Mojo driver, not
the HTTP server. There is no un-stop.

#### `is_stopping`

```mojo
def is_stopping(self) -> Bool
```

Whether `stop()` has been called.

#### `run`

```mojo
def run[T: AnyType, funcs: Tuple[funcs.Ts], /](service: String, ref state: T, object: Bool = True, workflow: Bool = False, host: String = "0.0.0.0", port: Int = Int(9080), num_workers: Int = Int(0)) -> Int
```

Discover the module's handlers, serve them, and block until stopped.

The whole driver:

    @fieldwise_init
    struct Stats(Copyable, Movable):
        var seen: Int64

    def handle_add(app: App, inv: Invocation, w: Int, ctx: Ctx[Stats]) raises:
        app.complete(inv, String("ok"))

    def main() raises:
        var stats = Stats(0)
        _ = App.run[Stats, __functions_in_module()]("Counter", stats)

Every `handle_*` in the calling module is registered under its name
without the prefix and dispatched to by that name. There is no list to
keep in step with the handlers, because there is no list.

Pass `__functions_in_module()` for `funcs`. `state` is process-local
and shared by every worker — see `Ctx`.

Returns how many invocations completed without raising.

#### `serve_with`

```mojo
def serve_with[T: AnyType, handler: def(Self, Invocation, Int, Ctx[T]) raises thin -> None](self, ref state: T, num_workers: Int = Int(0)) -> Int
```

`serve`, with the process-local state carried as a typed `Ctx[T]`.

The same driver; the only difference is that the handler receives
`Ctx[T]` instead of a raw `OpaquePtr`, so it can say `ctx[].field`
rather than computing offsets. `state` must outlive the call, which it
does: `serve` joins every worker before returning.

Prefer this over `serve`. The untyped form remains for handlers whose
state is genuinely not a single struct.

#### `serve`

```mojo
def serve[handler: def(Self, Invocation, Int, Pointer[UInt8, MutUntrackedOrigin]) raises thin -> None](self, num_workers: Int = Int(0), ctx: Pointer[UInt8, MutUntrackedOrigin] = opaque_ptr(Int(0))) -> Int
```

Run `handler` on `num_workers` threads until `stop()` is called.

The concurrent counterpart to the `while True: app.next()` loop, and
the reason a handler may now `call` another handler served by this same
process: with two or more workers, one thread waits for the callee
while another picks it up. With one worker that is still a deadlock,
exactly as before.

Blocks until every worker has left its loop, then returns.

Parameters:

- `handler`: The handler body — `def(App, Invocation, Int, OpaquePtr)
    thin raises -> None`, receiving the app, the invocation, its
    worker index, and `ctx`. Thin (non-capturing), because it is
    reached through a thread start routine; see the module
    docstring for the shape and the obligations.

Args:

- `num_workers`: How many driver threads. `0` (the default) means
    `num_cpus()`. **Use at least 2** if any handler calls back into
    this process.
- `ctx`: Process-local state shared by every handler invocation,
    passed straight through. Must outlive the `serve` call — which
    it does automatically, since `serve` joins before returning.
    Defaults to a null pointer for handlers that need none.

Raises: Error: If the workers cannot be started or joined.

### `Unit`

```mojo
struct Unit
```

No process-local state.

A service that keeps everything in Restate state still has to name a type
for `Ctx`. `Unit` is that type, and saying it out loud is worth more than
it costs: it tells a reader the handlers share nothing across invocations.

Implements: `Copyable`, `Movable`

### `Ctx`

```mojo
struct Ctx[T: AnyType]
```

A typed view of the process-local state `serve` hands to each handler.

Handlers must be thin, so state cannot be captured — it travels through a
pointer instead. `Ctx` is that pointer with its type still attached, so a
handler writes `ctx[].live` rather than reckoning byte offsets into an
untyped block.

**It does not make the state safe to share.** Every worker thread gets the
same `T`, concurrently. Fields that more than one handler touches have to
be atomics, exactly as they would through a raw pointer.

When Mojo can carry captures across a thread boundary this whole parameter
disappears from the signature and the state is simply captured; the body
keeps working with `ctx[].x` renamed to `x`.

Implements: `Copyable`, `Movable`

#### `__getitem__`

```mojo
def __getitem__(self) -> ref[MutUntrackedOrigin] T
```

#### `to`

```mojo
def to(ref state: T) -> Self
```

A `Ctx` over `state`, the one erasure served mode still performs: `state` is an argument of `serve_with`, which joins every worker before it returns, so it outlives the workers by construction even though the pointer no longer carries the origin that would say so.

#### `from_opaque`

```mojo
def from_opaque(ptr: Pointer[UInt8, MutUntrackedOrigin]) -> Self
```

Rebuild the typed view inside a worker, from what `serve` passed.

#### `opaque`

```mojo
def opaque(self) -> OpaquePtr
```

## Functions

### `is_suspended`

```mojo
def is_suspended(e: Error) -> Bool
```

True if `e` is the suspension signal (abandon the invocation and keep serving).

### `is_stopped`

```mojo
def is_stopped(e: Error) -> Bool
```

True if `e` means `stop()` was called: `next()` will not deliver anything more, so leave the loop.

## Aliases

### `STATUS_OK`

```mojo
comptime STATUS_OK = 0
```

### `STATUS_GONE`

```mojo
comptime STATUS_GONE = 1
```

### `STATUS_TERMINAL`

```mojo
comptime STATUS_TERMINAL = 2
```

### `STATUS_ERROR`

```mojo
comptime STATUS_ERROR = 3
```

### `STATUS_EMPTY`

```mojo
comptime STATUS_EMPTY = 4
```

### `STATUS_EXECUTE`

```mojo
comptime STATUS_EXECUTE = 5
```

### `SUSPENDED_MSG`

```mojo
comptime SUSPENDED_MSG = "restate: invocation suspended"
```

### `STOPPED_MSG`

```mojo
comptime STOPPED_MSG = "restate: driver stopped"
```

### `HandlerFn`

```mojo
comptime HandlerFn = def(App, Invocation, Int, Pointer[UInt8, MutUntrackedOrigin]) raises thin -> None
```

What `App.serve` runs for each invocation: `(app, inv, worker, ctx)`.

**Thin**, i.e. non-capturing — it is reached through a pthread start routine,
which takes a C function pointer and one `void *`. Anything the handler needs
beyond the app and the invocation travels through `ctx`, exactly as in
`threads.parallel_for`.

It *may* raise, unlike a raw thread body: `serve` wraps each call, so a raise
is caught, the invocation is `abandon`ed (Restate re-delivers it), and the
worker carries on. Suspension arrives this way too — `is_suspended(e)` — and is
not logged, because it is normal.

A handler must finish its invocation: `complete`, `fail`, or raise. Returning
without doing any of those leaks the invocation handle in the shim and leaves
Restate waiting.