# dista.frame

> A module of Dista's API.

Length-prefixed messages between drivers, the head and workers.

Every frame is a 4-byte little-endian length, then that many bytes:

| offset | field |
|---|---|
| 0 | kind (below) |
| 1 | object ID, u64 |
| 9 | task ID, u64 (0 unless the kind uses it) |
| 17 | payload |

| kind | sent by | payload |
|---|---|---|
| 1 task | driver to head, head to worker | the encoded argument |
| 2 result | worker to head, head to driver | the encoded result |
| 3 error | worker to head, head to driver | a UTF-8 message |
| 4 hello | driver, worker or observer to head | role byte (`ROLE_*`), the registry hash u64 (an observer's is ignored); a worker adds its object server's address |
| 5 welcome | head to driver | the driver's number, u64 |
| 6 task-after | driver to head | flags byte (`GATHER`, `WITH_VALUE`), count u64, then that many object IDs; with `WITH_VALUE`, a length u64 and the encoded value |
| 7 stored | worker to head | the result's size, u64: it stayed on the worker |
| 8 located | head to driver | the address of the worker holding the result |
| 9 fetch | driver or worker to a worker | none: send that object back as a result frame |
| 10 task-inputs | head to worker | flags byte, count u64, then per input a tag byte, its object ID, then for tag 0 a length u64 and the bytes, or for tag 1 its location; with `WITH_VALUE`, a length u64 and the value |
| 11 release | driver to head | none: the driver is done with this object |
| 12 free | head to worker | none: delete this object |
| 13 stats | driver to head, head to driver | none; the reply is five u64 counts (see `HeadStats`) |
| 14 holding | worker to head, on rejoining | count u64, then per object its ID u64 and its location |
| 15 claim | driver to head, on resuming | none: this driver still holds this object, which lives on a worker |
| 16 resumed | driver to head | none: this driver has restored, claimed or resubmitted everything |
| 17 span | worker to head, before the result | three u64 wall-clock nanoseconds: when the task arrived, started and ended |
| 18 timeline | driver to head, head to driver | none; the reply is a count u64, then per finished task its object ID, task ID, the four head stamps (submitted, queued, dispatched, settled), the three worker stamps, and the worker's address |
| 19 actor-create | driver to head, head to worker | from the driver, `checkpoint_every` u64 first; then the encoded constructor argument. The object ID names the actor, the task ID its type |
| 20 actor-call | driver to head, head to worker | the actor's ID u64, then the encoded argument; the task ID is the method's |
| 21 actor-kill | driver to head, head to worker | none: destroy the actor the object ID names |
| 22 checkpoint | driver to head, head to worker | from the driver, the actor's ID u64, and the object ID is the checkpoint's own; from the head, the calls covered by the checkpoint it would restore from, i64 (-1 for none), the object ID names the actor and the task ID is the number of calls the new one covers |
| 23 checkpointed | worker to head | none if the checkpoint is written, else why not; the object and task IDs are the checkpoint frame's |
| 24 actor-restore | head to worker | the number of calls the checkpoint covers, u64; the object ID names the actor, the task ID its type |
| 25 covered | head to driver | the newest of this driver's calls to the actor a checkpoint now includes, u64; the object ID names the actor |
| 26 status | observer to head, head to observer | none; the reply is the head's state as UTF-8 text |
| 27 stop | observer to head, head to worker | none: shut down; a worker exits instead of rejoining |
| 28 task-many | driver to head, head to worker | the number of results u64, then the encoded argument; the results are this object ID and the ones after it |
| 29 drain | worker to head | none: the worker was told to stop; it takes no more work, and leaves once the head says stop |
| 30 adopt | head to worker | a draining worker's location of the object: fetch it and keep it |
| 31 adopted | worker to head | 1 and the object's new location, or 0 and why not |

An observer (`dista status`, `timeline`, `stop`) says hello without a
registry and owns no objects. Its timeline frame gets every driver's
finished tasks, and leaves them for the drivers too.

After a head restart, a driver's hello carries its old number as a u64 after
the registry hash, so its object IDs stay valid. It then sends, in order of
submission, each object it still needs: a small result or error back as a
result or error frame, a result on a worker as claim, and an unfinished task
as the task or task-after frame it first sent; then resumed. A rejoining
worker sends holding.

An actor lives on one worker, which runs nothing else while it hosts it. The
worker answers actor-create with an empty result, or the constructor's
error, and each call like a task: a result, stored or error frame for the
call's object ID. Calls run in the order the driver sent them. A checkpoint
runs between calls too: the worker writes the actor's state to
`DISTA_CHECKPOINT_URI` and answers checkpointed. Actor-restore builds an
actor from its checkpoint instead of its constructor, and is answered like
actor-create.

Actors survive a head restart through their checkpoints. Every caller keeps
its calls to an actor until covered says a checkpoint includes them. A
resuming driver sends each live actor's actor-create again, in order with
everything else, then every call it still keeps: those answered too, after
their answers, since the actor's state hasn't seen them. The new head
restores the actor from its newest checkpoint; a call that checkpoint
already includes isn't run again. A rejoining worker destroys the actor it
hosted: its state may hold calls no caller would resend.

A task-after frame queues a task whose argument is built from other results:
the single result as is, or with gather set, a `Seq` of all of them; with
`WITH_VALUE` set, after an encoded value (a `TaskWith`). With `MANY` set it's
a `TaskMany`, whose results, the object ID and the ones after it, number the
u64 that follows the flags byte; the task-inputs frame carries it on.

A task-many frame runs a `TaskMany`: its worker answers each of its results
with its own result, stored or error frame, as if each were a task. When
some of those results live on workers, the head sends the task as
task-inputs, and the worker fetches them before it runs the task. An address
is 12 bytes: the IPv4 address, then the port as a u64.

The head counts references to every object: its driver's, until a release,
and one per submitted task that takes it as input, until that task settles.
At zero it forgets the object, and sends free to the worker holding it.

## Structs

### `Frame`

```mojo
struct Frame
```

Implements: `Movable`

Fields:

- `kind: UInt8`
- `object_id: UInt64`
- `task_id: UInt64`
- `payload: List[UInt8]`

#### `__init__`

```mojo
def __init__(out self, kind: UInt8, object_id: UInt64, task_id: UInt64, var payload: List[UInt8])
```

#### `take_payload`

```mojo
def take_payload(deinit self) -> List[UInt8]
```

Consumes the frame and returns its payload without copying it.

## Functions

### `message`

```mojo
def message(payload: Span[UInt8]) -> String
```

An error frame's text. It comes from another process, so bytes that aren't UTF-8 become replacement characters instead of a broken String.

### `append_frame`

```mojo
def append_frame(mut out: List[UInt8], kind: UInt8, object_id: UInt64, task_id: UInt64, payload: Span[UInt8])
```

Appends one whole frame to `out`.

### `write_frame`

```mojo
def write_frame(s: Stream, kind: UInt8, object_id: UInt64, task_id: UInt64, payload: Span[UInt8])
```

### `frame_length`

```mojo
def frame_length(head: Span[UInt8]) -> Int
```

The body length a 4-byte frame prefix announces, checked.

### `parse_body`

```mojo
def parse_body(body: Span[UInt8]) -> Frame
```

A frame from its body: everything after the length prefix.

### `read_frame`

```mojo
def read_frame(s: Stream) -> Optional[Frame]
```

Reads the next frame, or returns `None` if the peer closed cleanly.

### `encode_addr`

```mojo
def encode_addr(ip: Array[UInt8, Int(4)], port: Int) -> List[UInt8]
```

An address as frames carry it: the IPv4 address, then the port, u64.

### `decode_addr`

```mojo
def decode_addr(buf: Span[UInt8], mut pos: Int) -> Tuple[Array[UInt8, Int(4)], Int]
```

Reads an address written by `encode_addr` and advances `pos`.

## Aliases

### `KIND_TASK`

```mojo
comptime KIND_TASK = UInt8(1)
```

### `KIND_RESULT`

```mojo
comptime KIND_RESULT = UInt8(2)
```

### `KIND_ERROR`

```mojo
comptime KIND_ERROR = UInt8(3)
```

### `KIND_HELLO`

```mojo
comptime KIND_HELLO = UInt8(4)
```

### `KIND_WELCOME`

```mojo
comptime KIND_WELCOME = UInt8(5)
```

### `KIND_TASK_AFTER`

```mojo
comptime KIND_TASK_AFTER = UInt8(6)
```

### `KIND_STORED`

```mojo
comptime KIND_STORED = UInt8(7)
```

### `KIND_LOCATED`

```mojo
comptime KIND_LOCATED = UInt8(8)
```

### `KIND_FETCH`

```mojo
comptime KIND_FETCH = UInt8(9)
```

### `KIND_TASK_INPUTS`

```mojo
comptime KIND_TASK_INPUTS = UInt8(10)
```

### `KIND_RELEASE`

```mojo
comptime KIND_RELEASE = UInt8(11)
```

### `KIND_FREE`

```mojo
comptime KIND_FREE = UInt8(12)
```

### `KIND_STATS`

```mojo
comptime KIND_STATS = UInt8(13)
```

### `KIND_HOLDING`

```mojo
comptime KIND_HOLDING = UInt8(14)
```

### `KIND_CLAIM`

```mojo
comptime KIND_CLAIM = UInt8(15)
```

### `KIND_RESUMED`

```mojo
comptime KIND_RESUMED = UInt8(16)
```

### `KIND_SPAN`

```mojo
comptime KIND_SPAN = UInt8(17)
```

### `KIND_TIMELINE`

```mojo
comptime KIND_TIMELINE = UInt8(18)
```

### `KIND_ACTOR_CREATE`

```mojo
comptime KIND_ACTOR_CREATE = UInt8(19)
```

### `KIND_ACTOR_CALL`

```mojo
comptime KIND_ACTOR_CALL = UInt8(20)
```

### `KIND_ACTOR_KILL`

```mojo
comptime KIND_ACTOR_KILL = UInt8(21)
```

### `KIND_CHECKPOINT`

```mojo
comptime KIND_CHECKPOINT = UInt8(22)
```

### `KIND_CHECKPOINTED`

```mojo
comptime KIND_CHECKPOINTED = UInt8(23)
```

### `KIND_ACTOR_RESTORE`

```mojo
comptime KIND_ACTOR_RESTORE = UInt8(24)
```

### `KIND_COVERED`

```mojo
comptime KIND_COVERED = UInt8(25)
```

### `KIND_STATUS`

```mojo
comptime KIND_STATUS = UInt8(26)
```

### `KIND_STOP`

```mojo
comptime KIND_STOP = UInt8(27)
```

### `KIND_TASK_MANY`

```mojo
comptime KIND_TASK_MANY = UInt8(28)
```

### `KIND_DRAIN`

```mojo
comptime KIND_DRAIN = UInt8(29)
```

Worker to head: it was told to stop (SIGTERM), so send it no more work and move what it holds; the head says `KIND_STOP` once it may go.

### `KIND_ADOPT`

```mojo
comptime KIND_ADOPT = UInt8(30)
```

Head to worker: fetch object `object_id` from the location in the payload, a draining worker's, and keep it as its own.

### `KIND_ADOPTED`

```mojo
comptime KIND_ADOPTED = UInt8(31)
```

Worker to head: 1 and the object's new location, or 0 and why it couldn't take it.

### `GATHER`

```mojo
comptime GATHER = UInt8(1)
```

Task-after and task-inputs flag: the inputs gather into a `Seq`.

### `WITH_VALUE`

```mojo
comptime WITH_VALUE = UInt8(2)
```

Task-after and task-inputs flag: an encoded value goes before them.

### `MANY`

```mojo
comptime MANY = UInt8(4)
```

Task-after and task-inputs flag: the task is a `TaskMany`, and the number of its results, a u64, follows the flags byte.

### `ADDR_SIZE`

```mojo
comptime ADDR_SIZE = 12
```

### `ROLE_DRIVER`

```mojo
comptime ROLE_DRIVER = UInt8(1)
```

### `ROLE_WORKER`

```mojo
comptime ROLE_WORKER = UInt8(2)
```

### `ROLE_CLIENT`

```mojo
comptime ROLE_CLIENT = UInt8(3)
```

### `ROLE_OBSERVER`

```mojo
comptime ROLE_OBSERVER = UInt8(4)
```

A driver inside a worker, for a task that calls actors: the head treats it as a driver, except that it doesn't keep a local-mode head alive.

### `HEADER`

```mojo
comptime HEADER = 17
```

### `MAX_FRAME`

```mojo
comptime MAX_FRAME = 1073741824
```