---
title: "Data primitive semantics"
description: "This page holds the detailed rules for Attributes, RPCs, Channels, Streams, and"
canonical_url: "https://docs.superdurable.io/zh-Hans/reference/server/runtime-data/"
source: "docs/reference/server/runtime-data.mdx"
locale: "zh-Hans"
---
# Data primitive semantics
This page holds the detailed rules for Attributes, RPCs, Channels, Streams, and
SubFlows: parameters, defaults, limits, errors, and edge cases. The concept
pages explain what each primitive is and when to use it:
[Attribute](https://docs.superdurable.io/zh-Hans/primitives/attribute.md), [RPC](https://docs.superdurable.io/zh-Hans/primitives/rpc.md),
[Channel](https://docs.superdurable.io/zh-Hans/primitives/channel.md), [Stream](https://docs.superdurable.io/zh-Hans/primitives/stream.md),
[Timer](https://docs.superdurable.io/zh-Hans/primitives/timer.md), and [SubFlow](https://docs.superdurable.io/zh-Hans/primitives/subflow.md). The SDK doc
comments describe each API in its own language.
## Attributes
### Initial values and later writes
**StartFlowOptions** can set initial Attributes when the Flow is created.
Initial values and Flow creation are one operation, so the first Step sees them
immediately. See [Attributes](https://docs.superdurable.io/zh-Hans/reference/server/runtime-steps.md#attributes) in
StartFlowOptions.
Applications use Flow RPCs to write Attributes after a Flow starts. A Client can
also use **WaitForAttributeMatch** to wait until a singleton Attribute or one
AttributeMap instance satisfies a scalar comparison. The call returns the
matched current value. See
[Wait for an Attribute match](https://docs.superdurable.io/zh-Hans/reference/server/runtime-client.md#wait-for-an-attribute-match).
### Read Attributes from outside a Flow
Application code outside the Flow reads Attribute values with the Client's
**GetAttributes** call. One call reads any mix of singleton Attributes and
AttributeMap instances.
The read has these semantics:
- **One commit.** The Client sends every read in one server request. All values
come from the same committed state.
- **Current or latest run.** The call reads the current run, or the latest run
when no run is open. A closed run stays readable until Dex purges it.
- **No Worker.** The server answers the read itself. It dispatches nothing to a
Worker, uses no Worker capacity, and works while no Worker is running.
- **Committed state only.** The call does not see writes that a running Step or
RPC has made but not yet committed. An RPC handler, by contrast, also sees
the writes it made earlier in the same invocation.
- **No locks.** **GetAttributes** takes no Attribute locks. Use an RPC for a
read-modify-write.
- **Large and hard-delete values.** The server resolves
[hard-delete](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#hard-delete-attributes) values before it
responds, and a deleted value reads as unset. The Client loads other large
values from blob storage, then decodes every value with its definition. The
call returns values only after all of them decode.
An RPC remains the right choice when the read needs Flow-side validation,
authorization, a derived view, or Attribute locks. See
[Client requests: Read Attributes](https://docs.superdurable.io/zh-Hans/reference/server/runtime-client.md#read-attributes)
for the example, the unset value in each SDK, and the errors.
### Indexes
An indexed Attribute can be used in **SearchFlows** queries to find the Flow
executions that match the query.
| Index type | Value |
| ----------------------------- | ------------------------------- |
| **KEYWORD** | One exact string. |
| **FULL_TEXT** | Text for tokenized search. |
| **KEYWORD_ARRAY** | A list of exact strings. |
| **INT**, **DOUBLE**, **BOOL** | The corresponding scalar value. |
| **DATETIME** | A date and time value. |
An **index key** is the key in the search schema and the key you use in a query.
Index keys consume a finite, namespace-level Search Attribute pool and are
shared across Flow types.
The SDK defaults a regular Attribute's index key to its Attribute name. For
application indexes, always override that default with an explicit generic
numbered key. Keep business names on Attributes and share one typed slot pool
across every Flow type in the namespace. Two logical indexed fields in the same
Flow need different slots. Reuse a slot across Flow types only with the same
index type; its business meaning can differ. Dex system indexes keep their
reserved names.
| Index type | Application index keys |
| ----------------- | ------------------------------------- |
| **KEYWORD** | **keyword1**, **keyword2**, … |
| **KEYWORD_ARRAY** | **keywordList1**, **keywordList2**, … |
| **FULL_TEXT** | **text1**, **text2**, … |
| **INT** | **int1**, **int2**, … |
| **BOOL** | **bool1**, **bool2**, … |
| **DOUBLE** | **double1**, **double2**, … |
| **DATETIME** | **datetime1**, **datetime2**, … |
Renaming an index does not backfill existing Flow executions. Validate changed
example schemas in a fresh local **dexcli dev** instance. Plan any existing
deployment migration separately.
A generic key can have a different business meaning in each Flow type.
Application run queries must constrain **FlowType**. Keep the query unchanged
across pages. Dex Web already scopes searches to the selected Flow type.
Parenthesize caller filters before combining them with the application scope so
an OR cannot bypass it. Do not index sensitive data or PII. The customer-email
value in the Customer Refund example is synthetic data only.
**Python**
```python
status = Attribute(
"primitive-attribute-status",
str,
index=AttributeIndex(IndexType.KEYWORD, "keyword1"),
)
```
例子: [examples/python/dex_examples/primitives/attribute/attribute_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/attribute/attribute_flow.py)
**Go**
```go
var Status = dex.DefineAttribute[string](
"primitive-attribute-status",
dex.Indexed(dex.AttributeIndex{Type: dex.IndexKeyword, IndexKey: "keyword1"}),
)
```
例子: [examples/go/primitives/attribute/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/attribute/workflow.go)
**Java**
```java
private final Attribute status = Attribute.define(
"primitive-attribute-status",
String.class,
new AttributeIndex(AttributeIndex.Type.KEYWORD, "keyword1"));
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java)
**TypeScript**
```typescript
public readonly status = new Attribute("primitive-attribute-status", stringCodec, {
type: IndexType.KEYWORD,
indexKey: "keyword1",
});
```
例子: [examples/typescript/src/primitives/attribute/attribute-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/attribute/attribute-flow.ts)
**Rust**
```rust
let status = Attribute::new("primitive-attribute-status")
.indexed(AttributeIndex::keyword().with_key("keyword1"));
```
例子: [examples/rust/src/primitives/attribute/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/attribute/flow.rs)
For example, the status Attribute above uses **keyword1** as its index key. To
find completed **AttributeFlow** executions, use this query:
```text
FlowType = 'AttributeFlow' AND keyword1 = 'completed'
```
Pass the query to **SearchFlows**:
**Python**
```python
page = await app_state.client.search_flows(
"FlowType = 'AttributeFlow' AND keyword1 = 'completed'",
20,
"",
)
```
例子: [examples/python/dex_examples/primitives/client-apis/controller.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/client-apis/controller.py)
**Go**
```go
page, err := controller.client.SearchFlows(
request.Request.Context(),
"FlowType = 'AttributeFlow' AND keyword1 = 'completed'",
20,
"",
)
```
例子: [examples/go/primitives/client-apis/controller.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/client-apis/controller.go)
**Java**
```java
final SearchFlowsPage page = client.searchFlows("FlowType = 'AttributeFlow' AND keyword1 = 'completed'", 20, "");
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/clientapis/ClientApisController.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/clientapis/ClientApisController.java)
**TypeScript**
```typescript
const page = await client.searchFlows("FlowType = 'AttributeFlow' AND keyword1 = 'completed'", 20, "");
```
例子: [examples/typescript/src/primitives/client-apis/controller.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/client-apis/controller.ts)
**Rust**
```rust
client.search_flows("FlowType = 'AttributeFlow' AND keyword1 = 'completed'", 20)
```
例子: [examples/rust/src/primitives/client_apis/controller.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/client_apis/controller.rs)
#### Indexed AttributeMap
An indexed **AttributeMap** must use an explicit index key. It is not an index
over all Map instances. Each instance write updates the same search field, so
one instance can replace the value written by another instance. The Map
instance key is not searchable.
This does not turn the Map into a **KEYWORD_ARRAY**. **KEYWORD_ARRAY** indexes
the string array from one write; Dex does not combine the values from all Map
instances. After the example below writes the payment instance with value
authorized, you can query the current search value with:
```text
FlowType = 'AttributeFlow' AND keyword2 = 'authorized'
```
If another Map instance later writes a different value, that value replaces the
search value for **keyword2**. Use a regular Attribute with **KEYWORD_ARRAY**
when you need to search a collection of values. Do not use an AttributeMap index
to find every instance with a value.
**Python**
```python
progress = AttributeMap(
"primitive-attribute-progress",
str,
index=AttributeIndex(IndexType.KEYWORD, "keyword2"),
)
```
例子: [examples/python/dex_examples/primitives/attribute/attribute_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/attribute/attribute_flow.py)
**Go**
```go
var Progress = dex.DefineAttributeMap[string](
"primitive-attribute-progress",
dex.Indexed(dex.AttributeIndex{Type: dex.IndexKeyword, IndexKey: "keyword2"}),
)
```
例子: [examples/go/primitives/attribute/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/attribute/workflow.go)
**Java**
```java
private final AttributeMap progress =
AttributeMap.define(
"primitive-attribute-progress",
String.class,
new AttributeIndex(AttributeIndex.Type.KEYWORD, "keyword2"));
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java)
**TypeScript**
```typescript
public readonly progress = new AttributeMap("primitive-attribute-progress", stringCodec, {
type: IndexType.KEYWORD,
indexKey: "keyword2",
});
```
例子: [examples/typescript/src/primitives/attribute/attribute-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/attribute/attribute-flow.ts)
**Rust**
```rust
let progress = AttributeMap::new("primitive-attribute-progress")
.indexed(AttributeIndex::keyword().with_key("keyword2"));
```
例子: [examples/rust/src/primitives/attribute/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/attribute/flow.rs)
### Hard-delete Attributes
Dex stores every value of a hard-delete Attribute only in object storage,
whatever the value size. Flow history keeps only a reference. Deleting the
Attribute, or setting it to null, permanently deletes the stored value. After
that, neither Dex APIs nor Flow history can recover it.
Step and RPC handlers receive the value directly. The SDKs never write it to
their local blob cache, and Dex Server never writes it to its blob cache.
Enable it on the definition with **HardDeleteOnNull** in Go,
**hardDeleteOnNull** in Java and TypeScript, and **hard_delete_on_null** in
Python and Rust. The option works for both Attributes and AttributeMaps.
**Python**
```python
class AttributeFlow(Flow[str]):
model_api_key = Attribute(
"primitive-attribute-model-api-key",
str,
hard_delete_on_null=True,
)
class AttributeStep(Step[str]):
def execute(self, context: Context, input: str) -> StepDecision:
self.flow.status.set(context, "completed")
# The Flow no longer needs the short-lived key, so delete it permanently.
self.flow.model_api_key.delete(context)
return graceful_complete(input)
```
例子: [examples/python/dex_examples/primitives/attribute/attribute_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/attribute/attribute_flow.py)
**Go**
```go
ModelAPIKey = dex.DefineAttribute[string](
"primitive-attribute-model-api-key",
dex.HardDeleteOnNull(),
)
func (attributeStep) Execute(ctx dex.Context, input string) (*dex.StepDecision, error) {
if err := Status.Set(ctx, "completed"); err != nil {
return nil, err
}
// The Flow no longer needs the short-lived key, so delete it permanently.
if err := ModelAPIKey.Delete(ctx); err != nil {
return nil, err
}
return dex.GracefulComplete(input), nil
}
```
例子: [examples/go/primitives/attribute/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/attribute/workflow.go)
**Java**
```java
private final Attribute modelApiKey = Attribute.define(
"primitive-attribute-model-api-key",
String.class).hardDeleteOnNull();
@Override
public StepDecision execute(final Context context, final String input) {
status.set(context, "completed");
// The Flow no longer needs the short-lived key, so delete it permanently.
modelApiKey.delete(context);
return StepDecision.gracefulComplete(input);
}
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java)
**TypeScript**
```typescript
const modelApiKey = new Attribute(
"primitive-attribute-model-api-key",
stringCodec,
).hardDeleteOnNull();
public execute(context: Context, input: string): StepDecision {
this.flow.status.set(context, "completed");
// The Flow no longer needs the short-lived key, so delete it permanently.
this.flow.modelApiKey.delete(context);
return gracefulComplete(input);
}
```
例子: [examples/typescript/src/primitives/attribute/attribute-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/attribute/attribute-flow.ts)
**Rust**
```rust
static MODEL_API_KEY: LazyLock> =
LazyLock::new(|| Attribute::new("primitive-attribute-model-api-key").hard_delete_on_null());
fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult {
STATUS.set(context, "completed".to_owned())?;
// The Flow no longer needs the short-lived key, so delete it permanently.
MODEL_API_KEY.delete(context)?;
Ok(StepDecision::graceful_complete(input))
}
```
例子: [examples/rust/src/primitives/attribute/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/attribute/flow.rs)
A hard-delete Attribute has these requirements:
- Only string and object values are supported, including bytes. Numeric and
Boolean value types are rejected.
- It cannot be indexed or synced to an Attribute Store.
**WaitForAttributeMatch** and Action conditions reject it.
- A server that supports hard-delete Attributes, such as server-zd. The SDKs
refuse to write the Attribute to an older server.
Each Flow keeps one stored value per Attribute key and overwrites it in place.
Plan for these consequences:
- A Step or RPC attempt writes or deletes the stored value before its result is
committed. A retried attempt, other Steps and RPCs, and Attribute reads can
observe a write or deletion made by an attempt that later failed. Make
handlers safe to repeat instead of relying on a read-modify-write of the
Attribute.
- When concurrent Steps or RPCs write the same key, the final value may not be
the last committed write.
- A Flow start or SubFlow start that fails because the Flow ID is already
running still overwrites the running Flow's value. A null initial value
deletes it.
- Flows that reuse a Flow ID share the stored value.
- A deleted value always reads as absent. Time travel does not restore it. A
restarted SubFlow starts from the original initial value, so it reads as
absent when an earlier run deleted that value.
- Removing the option from a definition leaves the stored value in place until
Dex removes the Flow. Delete the value with the hard-delete definition first.
A value remains stored until it is deleted, or until Dex removes the Flow after
its retention period. Object storage backups and bucket versioning outside Dex
can keep copies. See [Object storage](https://docs.superdurable.io/zh-Hans/reference/server/object-storage.md).
#### Erase after the run closes
Since server-zd 0.1.5, a direct call to the FlowService **SetAttributes** API
can erase a hard-delete Attribute after its Flow run closes and before the run
is purged. The SDK Clients do not expose this API, so send it with
**dexcli api call SetAttributes** or another FlowService client. Set every
Attribute in the request to null. Each Attribute must still have a stored
hard-delete value. The erase leaves the run closed and permanently deletes its
stored objects. Other writes to a closed run return NOT_FOUND.
### Locks
Dex Server lets only one Step execution or RPC invocation hold an Attribute lock
at a time. A Step declares locks separately for **WaitFor** and **Execute** in
its Step options. An RPC declares its locks when it is registered. The
[Attribute](https://docs.superdurable.io/zh-Hans/primitives/attribute.md#locking) page shows an **Execute** lock.
A Step attempt takes its locks when Dex hands it to a Worker. A first attempt
that no Worker took yet holds no locks, but later Steps that need the same lock
wait behind it. A returned attempt, for example after a Worker declined its Step
type, and a retry attempt keep their locks while they wait in the queue. When no
Worker can run them, they hold the locks until the Flow times out or stops.
Transactional RPCs on those keys get the lock conflict until the attempt runs
and finishes.
Lock **WaitFor**:
**Python**
```python
def get_step_options(self) -> StepOptions:
return StepOptions(
wait_for_lock_attributes=(
self.flow.status.lock(),
self.flow.progress.lock("payment"),
),
)
```
例子: [examples/python/dex_examples/primitives/attribute/attribute_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/attribute/attribute_flow.py)
**Go**
```go
func (attributeStep) GetStepOptions() *dex.StepOptions {
return &dex.StepOptions{
WaitForLockAttributes: []dex.AttributeLock{
dex.LockAttribute(Status),
dex.LockAttributeMap(Progress, "payment"),
},
}
}
```
例子: [examples/go/primitives/attribute/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/attribute/workflow.go)
**Java**
```java
return StepOptions.newBuilder()
.addWaitForLock(AttributeLock.of(status))
.addWaitForLock(AttributeLock.of(progress, "payment"))
.build();
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java)
**TypeScript**
```typescript
return {
waitForLockAttributes: [this.flow.status.lock(), this.flow.progress.lock("payment")],
};
```
例子: [examples/typescript/src/primitives/attribute/attribute-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/attribute/attribute-flow.ts)
**Rust**
```rust
StepOptions::new()
.wait_for_lock(self.status.lock())
.wait_for_lock(self.progress.lock("payment"))
```
例子: [examples/rust/src/primitives/attribute/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/attribute/flow.rs)
Lock an RPC:
**Python**
```python
@rpc(lock_attributes=(status.lock(), progress.lock("payment")))
def update_status(self, context: Context, input: str) -> RPCResult[str]:
self.status.set(context, input)
self.progress.set(context, "payment", input)
return RPCResult(input)
```
例子: [examples/python/dex_examples/primitives/attribute/attribute_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/attribute/attribute_flow.py)
**Go**
```go
func (flow *AttributeFlow) GetRPCs() []dex.RPCDef {
return []dex.RPCDef{
dex.DefineRPC(flow.UpdateStatus, &dex.RPCOptions{
LockAttributes: []dex.AttributeLock{
dex.LockAttribute(Status),
dex.LockAttributeMap(Progress, "payment"),
},
}),
}
}
```
例子: [examples/go/primitives/attribute/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/attribute/workflow.go)
**Java**
```java
@RPC(
lockAttributes = {"primitive-attribute-status"},
lockAttributeMaps = {
@RPCAttributeMapLock(
attribute = "primitive-attribute-progress",
instance = "payment")
})
public RPCResult updateStatus(final Context context, final String input) {
status.set(context, input);
progress.set(context, "payment", input);
return RPCResult.of(input);
}
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/attribute/AttributeFlow.java)
**TypeScript**
```typescript
@rpc({
inputCodec: stringCodec,
outputCodec: stringCodec,
lockAttributes: [status.lock(), progress.lock("payment")],
})
public updateStatus(context: Context, input: string): RPCResult {
this.status.set(context, input);
this.progress.set(context, "payment", input);
return { output: input };
}
```
例子: [examples/typescript/src/primitives/attribute/attribute-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/attribute/attribute-flow.ts)
**Rust**
```rust
fn rpcs(&self) -> RpcList {
RpcList::new().function(
UPDATE_STATUS
.lock(self.status.lock())
.lock(self.progress.lock("payment")),
Self::update_status,
)
}
```
例子: [examples/rust/src/primitives/attribute/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/attribute/flow.rs)
#### Coordinate writes across the whole AttributeMap
An AttributeMap lock protects one exact instance. When an invariant spans
several instances or the whole collection, register a separate singleton bool
Attribute as its coordination key. For example, register **currentMessages** as
an AttributeMap and **currentMessagesLock** as a non-map, non-indexed bool
Attribute in the same **PersistenceSchema**.
Every Step, timeout handler, and RPC that writes, deletes, archives, or
otherwise changes the protected map must request the same singleton Attribute
lock in its method-specific lock options. RPCs declare this fixed lock at
registration time. A writer that requests only a map-instance lock does not
conflict with the singleton lock. All participating writers serialize within
that Flow execution; other Flow executions have their own locks.
The bool value is not the lock state. Do not acquire or release the lock by
reading or toggling the value. Dex acquires and releases it around the
invocation; the coordination Attribute does not need a persisted value. Locking
also does not load map entries. Declare exact instance loads, or a whole-map
load when the operation must enumerate instances. RPC lock acquisition can
return a lock-conflict error rather than queueing the caller.
Keep per-instance locks for independent partition updates. Use the shared
singleton lock when all affected writers must preserve a collection-wide
invariant.
#### RPC lock conflicts
RPC lock acquisition does not queue callers. When another Step execution or RPC
holds a requested lock, the call fails with the error "RPC lock conflict:
attribute key \ is locked by a Step". The SDK exposes it as
**RPCLockConflictError** in Go, **RpcLockConflictError** in Python and
TypeScript, **RpcLockConflictException** in Java, and
**SdkError::RpcLockConflict** in Rust.
Retry specifically after that error, with application-appropriate backoff. Make
the RPC idempotent so that a retry is safe. Successful calls serialize through
the same locks, so concurrent requests do not lose or duplicate updates.
## RPCs
Language-specific RPC rules are in the SDK references: Java Flow classes and
[Kotlin](https://docs.superdurable.io/zh-Hans/reference/java.md#rpc-stubs-and-kotlin), and
[TypeScript codecs](https://docs.superdurable.io/zh-Hans/reference/typescript.md#rpc-codecs).
### Cancel running Steps
An RPC result can cancel active Steps of its Flow. Select the Steps by Step
type: **with_canceling_steps** in Python, **CancelSteps** in Go,
**withCancelingSteps** in Java, **cancelingSteps** in TypeScript, and
**cancel_step** in Rust. The selection covers the whole Flow. An RPC has no
parent Step execution, so it cannot cancel sibling Steps. A Step can cancel
other Steps the same way from its
[StepDecision](https://docs.superdurable.io/zh-Hans/reference/server/runtime-steps.md#cancellation-on-stepdecision).
In the example, **GenerateReport** writes a report section by section. The
**stop_report** RPC cancels it and starts **FinishReportStop**.
**Python**
```python
@rpc(timeout=timedelta(seconds=30))
def stop_report(self, context: Context, input: None) -> RPCResult[bool]:
if REPORT_STATUS.get(context) in (REPORT_READY, REPORT_STOPPED):
return RPCResult(False)
return RPCResult(
True,
next_steps=(StepMovement.of(FinishReportStop, None),),
).with_canceling_steps(GenerateReport)
```
例子: [examples/python/dex_examples/primitives/rpc_canceling_steps](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/rpc_canceling_steps)
**Go**
```go
func (*ReportGenerationFlow) StopReport(ctx dex.Context, _ dex.None) (*dex.RPCResult[bool], error) {
status, err := getReportStatus(ctx)
if err != nil {
return nil, err
}
if status == ReportReady || status == ReportStopped {
return &dex.RPCResult[bool]{Output: false}, nil
}
result := &dex.RPCResult[bool]{
Output: true,
NextSteps: []dex.StepMovement{dex.MovementOf(FinishReportStop{}, nil)},
}
return result.CancelSteps(GenerateReport{}), nil
}
```
例子: [examples/go/primitives/rpc-canceling-steps](https://github.com/superdurable/dex/tree/main/examples/go/primitives/rpc-canceling-steps)
**Java**
```java
@RPC(timeoutSeconds = 30)
public RPCResult stopReport(final Context context) {
final String status = reportStatus.get(context);
if (REPORT_READY.equals(status) || REPORT_STOPPED.equals(status)) {
return RPCResult.of(false);
}
return RPCResult.of(true, StepMovement.of(FinishReportStop.class, null))
.withCancelingSteps(GenerateReport.class);
}
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/rpccancelingsteps](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/rpccancelingsteps)
**TypeScript**
```typescript
@rpc({ inputCodec: voidCodec, outputCodec: booleanCodec, timeoutMs: 30_000 })
public stopReport(context: Context, _input: void): RPCResult {
const status = reportStatus.get(context);
if (status === REPORT_READY || status === REPORT_STOPPED) {
return { output: false };
}
return {
output: true,
nextSteps: [StepMovement.of(FinishReportStop, undefined)],
cancelingSteps: [GenerateReport],
};
}
```
例子: [examples/typescript/src/primitives/rpc-canceling-steps](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/rpc-canceling-steps)
**Rust**
```rust
fn stop_report(&self, context: &mut Context, _input: ()) -> HandlerResult> {
let status = REPORT_STATUS.get(context)?;
if matches!(status.as_deref(), Some(REPORT_READY | REPORT_STOPPED)) {
return Ok(RpcResult::new(false));
}
Ok(RpcResult::new(true)
.then(StepMovement::to(&self.finish_report_stop, ()))
.cancel_step(&self.generate_report))
}
```
例子: [examples/rust/src/primitives/rpc/canceling_steps](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/rpc/canceling_steps)
Dex applies the cancellation when it commits the RPC. It commits the RPC's
Attribute writes and Channel publications first, then cancels, then starts the
RPC's next Steps. All of it commits together, or none of it does. The RPC's own
next Steps are not canceled, even when they have a selected Step type. A
selected Step that has already finished is skipped, and so is a type with no
active Step.
Cancellation ends every active Step of a selected type: one that waits, one that
waits for a retry, and one whose attempt runs on a Worker. For a running
attempt, Dex sends a cancellation to that Worker. A canceled Step is not
retried, and its failure route does not run. Dex discards anything the canceled
attempt returns later, including its StepDecision and Attribute writes.
#### Resolve the race from committed state
The RPC handler reads a snapshot of the Flow when it starts. Dex commits its
result later. The Steps to cancel are chosen at commit time, not from the
snapshot. Between the two, a selected Step can finish and commit its own result.
Then the cancellation finds nothing to cancel. A new Step of a selected type can
also start, and the cancellation ends it too. Dex does not reject the RPC in
either case. So the handler cannot assume that the Step it saw is still running,
or that its work did not commit.
Do not decide the outcome in the RPC handler. Start a follow-up Step from the
RPC, and let it read the committed state. In the example,
**FinishReportStop** checks whether the report committed before the
cancellation. If it did, the Stop came too late, and the Flow delivers the
report. Otherwise, it records that the report stopped.
**Python**
```python
def execute(self, context: Context, input: None) -> StepDecision:
# The report can commit after stop_report read its snapshot.
if REPORT_STATUS.get(context) == REPORT_READY:
return dead_end()
REPORT_STATUS.set(context, REPORT_STOPPED)
return graceful_complete(REPORT_STOPPED)
```
例子: [examples/python/dex_examples/primitives/rpc_canceling_steps](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/rpc_canceling_steps)
**Go**
```go
func (FinishReportStop) Execute(ctx dex.Context, _ dex.None) (*dex.StepDecision, error) {
status, err := getReportStatus(ctx)
if err != nil {
return nil, err
}
// The report can commit after StopReport read its snapshot.
if status == ReportReady {
return dex.DeadEnd(), nil
}
if err := ReportStatus.Set(ctx, ReportStopped); err != nil {
return nil, err
}
return dex.GracefulComplete(ReportStopped), nil
}
```
例子: [examples/go/primitives/rpc-canceling-steps](https://github.com/superdurable/dex/tree/main/examples/go/primitives/rpc-canceling-steps)
**Java**
```java
public StepDecision execute(final Context context, final Void input) {
// The report can commit after stopReport read its snapshot.
if (REPORT_READY.equals(reportStatus.get(context))) {
return StepDecision.deadEnd();
}
reportStatus.set(context, REPORT_STOPPED);
return StepDecision.gracefulComplete(REPORT_STOPPED);
}
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/rpccancelingsteps](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/rpccancelingsteps)
**TypeScript**
```typescript
public execute(context: Context, _input: void): StepDecision {
// The report can commit after stopReport read its snapshot.
if (reportStatus.get(context) === REPORT_READY) {
return deadEnd();
}
reportStatus.set(context, REPORT_STOPPED);
return gracefulComplete(REPORT_STOPPED);
}
```
例子: [examples/typescript/src/primitives/rpc-canceling-steps](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/rpc-canceling-steps)
**Rust**
```rust
fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult {
// The report can commit after stop_report read its snapshot.
if REPORT_STATUS.get(context)?.as_deref() == Some(REPORT_READY) {
return Ok(StepDecision::dead_end());
}
REPORT_STATUS.set(context, REPORT_STOPPED.to_string())?;
Ok(StepDecision::graceful_complete(REPORT_STOPPED.to_string()))
}
```
例子: [examples/rust/src/primitives/rpc/canceling_steps](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/rpc/canceling_steps)
[Attribute locks](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#attribute-locks) do not close this gap on their own. They
serialize the RPC only with Step methods that declare the same locks.
#### Make the Step stop
Dex cannot stop code that ignores cancellation. The Step code must check for it
and return:
- Python: with **AsyncWorker**, Dex cancels the handler's task, so the next
await raises **CancelledError**. With **Worker**, the handler runs on a thread
and must check **is_cancellation_requested**.
- Go: Dex cancels the handler's **Context**. Pass it to calls that accept a
context, or select on **Done**.
- Java: Dex interrupts the handler thread. Blocking calls throw
**InterruptedException**, and **isCancellationRequested** returns true.
- TypeScript: Dex aborts **context.cancellationSignal**. Pass it to calls that
accept an **AbortSignal**.
- Rust: check **is_cancelled**, or block in **wait_for_cancellation**.
**Python**
```python
for section in range(1, SECTION_COUNT + 1):
# A Stop cancels this task, so the await raises CancelledError.
await asyncio.sleep(SECTION_WORK.total_seconds())
REPORT_PROGRESS.write(context, section)
```
例子: [examples/python/dex_examples/primitives/rpc_canceling_steps](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/rpc_canceling_steps)
**Go**
```go
for section := 1; section <= sectionCount; section++ {
select {
case <-ctx.Done():
// A Stop cancels ctx; the result of this attempt is discarded.
return nil, ctx.Err()
case <-time.After(sectionWork):
}
if err := ReportProgress.Write(ctx, section); err != nil {
return nil, err
}
}
```
例子: [examples/go/primitives/rpc-canceling-steps](https://github.com/superdurable/dex/tree/main/examples/go/primitives/rpc-canceling-steps)
**Java**
```java
for (int section = 1; section <= SECTION_COUNT; section++) {
try {
Thread.sleep(SECTION_WORK.toMillis());
} catch (final InterruptedException interrupted) {
// A Stop interrupts this thread; the result of this attempt is discarded.
Thread.currentThread().interrupt();
throw new IllegalStateException("report generation canceled", interrupted);
}
reportProgress.write(context, section);
}
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/rpccancelingsteps](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/rpccancelingsteps)
**TypeScript**
```typescript
for (let section = 1; section <= SECTION_COUNT; section++) {
// A Stop aborts the signal, so this await rejects with an AbortError.
await setTimeout(SECTION_WORK_MS, undefined, { signal: context.cancellationSignal });
reportProgress.write(context, section);
}
```
例子: [examples/typescript/src/primitives/rpc-canceling-steps](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/rpc-canceling-steps)
**Rust**
```rust
for section in 1..=SECTION_COUNT {
// A Stop cancels this attempt, and Dex discards its result.
if context.is_cancelled() {
return Ok(StepDecision::dead_end());
}
thread::sleep(SECTION_WORK);
REPORT_PROGRESS.write(context, section)?;
}
```
例子: [examples/rust/src/primitives/rpc/canceling_steps](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/rpc/canceling_steps)
Dex waits about one second for the canceled attempt to end. If the handler is
still running, Dex closes the Worker stream that carried it. The handler keeps
running inside the Worker process until it returns, and its external calls
still take effect. A canceled attempt may also have done part of its work before
it stopped. Keep its external effects idempotent. See
[Failure and redelivery](https://docs.superdurable.io/zh-Hans/reference/server/runtime-steps.md#failure-and-redelivery).
### Read-only RPCs and terminal Flows
To read stored Attribute values from outside the Flow, an application does not
need a read-only RPC. The Client's **GetAttributes** reads committed values
directly, without a Worker. See
[Read Attributes from outside a Flow](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#read-attributes-from-outside-a-flow). Use
a read-only RPC when the read needs Flow-side logic, such as validation,
authorization, Channel messages, or a derived view.
An RPC without Attribute locks and without transactional execution starts with a
read of the Flow. If its handler returns only an output and no durable effects,
Dex returns that output without changing the Flow. That read can reach a
retained terminal execution. A successful read-only RPC therefore does not prove
that the Flow is active.
Attribute locks and explicit transactional execution use an active execution. A
non-transactional handler that returns Attribute writes, Channel operations,
events, or Step movements also needs an active execution because Dex commits
those effects to the Flow after the handler returns. A terminal Flow rejects
that commit even though the Worker handler may already have run. A Server can
also be configured to run every RPC transactionally, which requires an active
execution.
Use a Flow lifecycle API when the response needs current execution status. An
application snapshot RPC reads the durable application state captured by the
Flow; it does not infer terminal status from RPC success.
**FlowNotActiveOrNotFoundError** means the target was not found, or the selected
operation cannot use its closed execution. Java uses
**FlowNotActiveOrNotFoundException**; Rust uses
**SdkError::FlowNotActiveOrNotFound**. The error alone does not distinguish
missing from closed Flows.
For a confirmed read-only RPC with no locks, transactionality, durable effects,
or Server-forced transactional execution, a retained terminal execution remains
readable. At that read boundary, handle this error directly as no readable
target found. Preserve the application's missing or retention-unavailable
outcome. Do not add a lifecycle lookup, **WaitForFlow** with a short timeout, a
retry, or historical Step-output decoding solely to distinguish missing from
closed. Keep other service and Worker failures distinct.
For mutations and active-only RPCs, the same error may mean the Flow is closed
or missing. Apply the operation's business policy; the error does not prove that
the requested action succeeded.
Callers do not need to search for a Run ID before a normal read-only RPC. When
the SDK exposes an optional Run ID, omit it to let the Server resolve and pin the
current execution for that invocation. Supply one only when the application must
address an exact execution across Flow ID reuse.
### Selective state loading
Every RPC receives regular Attribute values and Channel size metadata. Reading
AttributeMap entries or pending Channel messages requires an explicit load. This
keeps large collections out of RPC requests that do not use them.
Load a whole AttributeMap or ChannelMap when the handler needs all current
instances. Load exact map instances when it needs only known keys. Load a
Channel when the handler needs its pending message IDs or Values. Channel size
and ChannelMap keys and sizes remain available without loading pending messages.
Every SDK declares loads with the RPC. Go places them in **RPCOptions** passed
to **DefineRPC** from **GetRPCs**. Dex Server validates each requested definition
against the Flow persistence schema and returns the loaded scope with the RPC
snapshot. The SDK checks that scope when the handler reads a selectively loaded
collection.
For an **AttributeMap** entry or instance enumeration outside that scope, the SDK
raises **AttributeMapNotLoadedError**. For pending messages outside that scope,
it raises **ChannelMessagesNotLoadedError**. Java uses
**AttributeMapNotLoadedException** and **ChannelMessagesNotLoadedException**.
For Go, **dexcli visualize** reports an RPC that reads unselected state. See
[State loads](https://docs.superdurable.io/zh-Hans/reference/web/flow-visualization.md#state-loads).
An explicitly loaded collection can be empty. That is not an error. An empty
pending-message snapshot returns an empty list. An absent **AttributeMap** entry
uses the SDK's normal absent-value behavior. Loading creates one input snapshot.
It does not consume messages, make the RPC transactional, or isolate the
handler from concurrent work.
#### Runtime-selected map instances
Use the RPC definition options for fixed behavior shared by every invocation:
timeout, ordinary Attribute locks, transactionality, whole-map loads, singleton
Channel loads, and fixed exact-instance loads. Go names this type
**RPCOptions**.
Use **RPCInvokeOptions** when the application adapter chooses an AttributeMap or
ChannelMap instance from request data. Rust uses **RpcInvokeOptions**.
Invocation options can add exact AttributeMap loads, exact ChannelMap loads, and
AttributeMap instance locks. Dex unions, sorts, and deduplicates them with the
registered definition. They cannot replace or clear registered settings.
Loading and locking are independent. A read-only lookup needs only the exact
load. A read-modify-write RPC must add both the load and the lock for the same
AttributeMap instance. The SDK validates the Flow ownership, definition kind,
and non-empty slash-free instance before making the network call.
**Python**
```python
options=RPCInvokeOptions(
lock_attribute_map_instances=(
flow.customer_profiles_by_email_partition.lock(partition_name),
),
load_attribute_map_instances=(
flow.customer_profiles_by_email_partition.load(partition_name),
),
)
```
例子: [examples/python/dex_examples/patterns/hash-partitioned-attribute-map/controller.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/patterns/hash-partitioned-attribute-map/controller.py)
**Go**
```go
dex.RPCInvokeOptions{
LockAttributeMapInstances: []dex.AttributeLock{
dex.LockAttributeMap(CustomerProfilesByEmailPartition, partitionName),
},
LoadAttributeMapInstances: []dex.AttributeMapLoad{
CustomerProfilesByEmailPartition.Load(partitionName),
},
}
```
例子: [examples/go/patterns/hash-partitioned-attribute-map/controller.go](https://github.com/superdurable/dex/tree/main/examples/go/patterns/hash-partitioned-attribute-map/controller.go)
**Java**
```java
final RPCInvokeOptions options = RPCInvokeOptions.newBuilder()
.addLockAttributeMapInstance(
flow.customerProfilesByEmailPartition,
partition.partitionName)
.addLoadAttributeMapInstance(
flow.customerProfilesByEmailPartition,
partition.partitionName)
.build();
```
例子: [examples/java/src/main/java/io/superdurable/dex/patterns/hashpartitionedattributemap/HashPartitionedAttributeMapController.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/patterns/hashpartitionedattributemap/HashPartitionedAttributeMapController.java)
**TypeScript**
```typescript
{
lockAttributeMapInstances: [profiles.lock(partitionName)],
loadAttributeMapInstances: [profiles.load(partitionName)],
}
```
例子: [examples/typescript/src/patterns/hash-partitioned-attribute-map/controller.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/patterns/hash-partitioned-attribute-map/controller.ts)
**Rust**
```rust
RpcInvokeOptions::new()
.lock_attribute_map_instance(
CUSTOMER_PROFILES_BY_EMAIL_PARTITION.lock(&partition.partition_name),
)
.load_attribute_map_instance(
CUSTOMER_PROFILES_BY_EMAIL_PARTITION.load(partition.partition_name),
)
```
例子: [examples/rust/src/patterns/hash_partitioned_attribute_map/controller.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/patterns/hash_partitioned_attribute_map/controller.rs)
See [Hash-Partitioned AttributeMap](https://docs.superdurable.io/zh-Hans/design-patterns/hash-partitioned-attribute-map.md)
for the complete lookup and upsert example, and
[Sequentially Chunked AttributeMap](https://docs.superdurable.io/zh-Hans/design-patterns/sequentially-chunked-attribute-map.md)
for dynamic pagination loads.
### Transactional RPCs
A transactional RPC performs its reads and writes as one atomic operation. An
RPC that holds an Attribute lock is already transactional implicitly. You can
also make an RPC explicitly transactional by setting **is_transactional=true**.
Transactional execution matters most for Channel message deletion. Dex validates
that every deleted message is still pending before it commits any write. If a
message is gone, the RPC fails with the Channel-message-not-found error
(**NOT_FOUND**) and commits nothing. A non-transactional RPC treats a missing
deletion as a no-op and still commits its other successful effects.
In the example, the caller passes only a pending message ID. The RPC loads the
source Channel, reads the original Value from its snapshot, then stages the
deletion and destination publication. A concurrent consumer makes deletion
validation fail, so Dex commits neither effect:
**Python**
```python
@rpc(is_transactional=True, load_channels=(queued_messages,))
def move_queued_message_to_prioritized_messages(
self, context: Context, queued_message: QueuedMessageReference
) -> None:
message_to_prioritize = self.queued_messages.find_pending_message(
context, queued_message.message_id
)
self.queued_messages.delete(context, queued_message.message_id)
if message_to_prioritize is not None:
self.prioritized_messages.publish(context, message_to_prioritize.value)
```
例子: [examples/python/dex_examples/primitives/channel/channel_flow.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/channel/channel_flow.py)
**Go**
```go
func (flow *ChannelFlow) GetRPCs() []dex.RPCDef {
return []dex.RPCDef{
dex.DefineRPC(flow.PublishApprovalMessage, nil),
dex.DefineRPC(flow.EnqueueChannelMessage, nil),
dex.DefineRPC(flow.GetQueuedMessages, &dex.RPCOptions{
LoadChannels: []dex.ChannelDef{QueuedMessages},
}),
dex.DefineRPC(flow.DeleteQueuedMessage, &dex.RPCOptions{
IsTransactional: true,
LoadChannels: []dex.ChannelDef{QueuedMessages},
}),
dex.DefineRPC(flow.GetPrioritizedMessages, &dex.RPCOptions{
LoadChannels: []dex.ChannelDef{PrioritizedMessages},
}),
dex.DefineRPC(flow.MoveQueuedMessageToPrioritizedMessages, &dex.RPCOptions{
IsTransactional: true,
LoadChannels: []dex.ChannelDef{QueuedMessages},
}),
}
}
```
例子: [examples/go/primitives/channel/workflow.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/channel/workflow.go)
**Java**
```java
@RPC(isTransactional = true, loadChannels = {"QueuedMessages"})
public void moveQueuedMessageToPrioritizedMessages(
final Context context,
final QueuedMessageReference queuedMessageReference) {
final ChannelMessage messageToPrioritize =
queuedMessages.findPendingMessage(context, queuedMessageReference.messageId);
queuedMessages.delete(context, queuedMessageReference.messageId);
if (messageToPrioritize != null) {
prioritizedMessages.publish(context, messageToPrioritize.getValue());
}
}
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/channel/ChannelFlow.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/channel/ChannelFlow.java)
**TypeScript**
```typescript
@rpc({
isTransactional: true,
loadChannels: [queuedMessages],
inputCodec: queuedMessageReferenceCodec,
})
public moveQueuedMessageToPrioritizedMessages(
context: Context,
queuedMessage: QueuedMessageReference,
): void {
const messageToPrioritize = queuedMessages.findPendingMessage(
context,
queuedMessage.messageId,
);
queuedMessages.delete(context, queuedMessage.messageId);
if (messageToPrioritize !== undefined) {
prioritizedMessages.publish(context, messageToPrioritize.value);
}
}
```
例子: [examples/typescript/src/primitives/channel/channel-flow.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/channel/channel-flow.ts)
**Rust**
```rust
fn rpcs(&self) -> RpcList {
RpcList::new()
.procedure_without_input(PUBLISH_APPROVAL_MESSAGE, Self::publish_approval_message)
.procedure(
MOVE_QUEUED_MESSAGE_TO_PRIORITIZED_MESSAGES
.is_transactional()
.load_channel(&QUEUED_MESSAGES),
Self::move_queued_message_to_prioritized_messages,
)
}
```
例子: [examples/rust/src/primitives/channel/flow.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/channel/flow.rs)
A transactional commit does not isolate the handler while it runs. If a decision
depends on the whole loaded Channel, ChannelMap, or AttributeMap remaining
unchanged, every cooperating Step and RPC writer must use the same Attribute
lock. The lock provides that isolation and implicitly makes the RPC
transactional; it does not load collection contents.
### RPC timeout
An RPC timeout limits one Worker handler invocation.
Every SDK declares the timeout with the RPC. Go places it in the static
**RPCOptions** passed to **DefineRPC**. A Client invocation cannot override it,
although the caller's context can stop waiting sooner. When the application
does not set a timeout, Dex Server uses its default of 60 seconds, set by
**api.defaultRpcTimeout**. An explicit timeout has no maximum.
**Python**
```python
@rpc(timeout=timedelta(seconds=30))
```
例子: [examples/python/dex_examples/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/rpc)
**Go**
```go
func (flow *RpcFlow) GetRPCs() []dex.RPCDef {
return []dex.RPCDef{
dex.DefineRPC(flow.Trigger, &dex.RPCOptions{Timeout: 30 * time.Second}),
}
}
```
例子: [examples/go/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/go/primitives/rpc)
**Java**
```java
@RPC(timeoutSeconds = 30)
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/rpc)
**TypeScript**
```typescript
@rpc({ inputCodec: stringCodec, outputCodec: stringCodec, timeoutMs: 30_000 })
```
例子: [examples/typescript/src/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/rpc)
**Rust**
```rust
RpcList::new().function(TRIGGER_RPC.timeout(Duration::from_secs(30)), Self::trigger)
```
例子: [examples/rust/src/primitives/rpc](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/rpc)
## Channels
### Wait conditions
**AtMost** does not wait on its own for messages to accumulate. When the
surrounding Wait completes, it consumes up to the limit from the queue at that
time. To wait for at least one message and consume up to a limit, use
**AtLeastAtMost** with a lower bound of 1.
### Pending messages
Dex assigns a UUIDv7 message ID when it accepts each publication. A Step or RPC
can list every pending message in FIFO order and delete one by ID. Listing does
not consume messages.
Deleting a message that is no longer pending depends on the caller:
| Caller | Missing message |
| ------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| Transactional RPC | The RPC fails with the Channel-message-not-found error and commits nothing. See [Transactional RPCs](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#transactional-reads-and-writes). |
| Non-transactional RPC | The deletion is a no-op. The other effects commit. |
| Step method or Flow timeout handler | The deletion is a no-op. The other effects commit. |
| Direct **DeleteChannelMessage** (dexcli, Dex Web) | The call fails with **NOT_FOUND** and sub-status **CHANNEL_MESSAGE_NOT_FOUND**. |
Pending messages are queue state, not Flow history. After a Step consumes a
message, it disappears from the Channel even though the publication and
consumption remain observable in recent
[Flow history](https://docs.superdurable.io/zh-Hans/reference/server/configuration.md#history-window).
Every Step method, timeout handler, and RPC always receives Channel size
metadata. It must explicitly load pending messages before reading their IDs or
Values. Deleting a message whose ID is already known does not require loading
it. **StepOptions** selects state separately for **WaitFor** and **Execute**;
**FlowTimeoutHandlerOptions** selects state for the timeout handler. Loading a
Channel creates one invocation snapshot and does not consume messages. A loaded
empty Channel returns an empty list; reading pending messages without loading
the Channel is a usage error.
Treat every pending-message snapshot read inside a Step or RPC as potentially
stale. Other Steps and RPCs can consume, delete, or publish messages while the
handler is running. Transactional execution validates selected deletions and
commits the handler's writes atomically, but it does not lock the whole Channel
snapshot. Read and write pending messages directly only when the operation
explicitly tolerates that concurrency. If a decision requires the queue to
remain unchanged, every cooperating Step and RPC writer must use the same
Attribute lock.
A Step or timeout handler can stage deletion with its other side effects. Dex
applies successful handler effects in this order: Attribute writes, best-effort
Channel deletions, then Channel publications, followed by the Step decision.
**StepDecision** describes only control flow; it does not contain deletions.
An RPC can combine deletion with other durable writes. The move example in
[Transactional RPCs](https://docs.superdurable.io/zh-Hans/reference/server/runtime-data.md#transactional-reads-and-writes) loads **QueuedMessages**,
finds the original Value by message ID, then stages its deletion and
republishes that Value. The caller sends only the ID.
### ChannelMap
RPCs can always read the current ChannelMap keys and sizes. Load the entire
ChannelMap when the handler needs every instance's pending messages. Load exact
instances when it needs only known keys. Instance keys cannot contain **/**
because Dex uses that character between the map name and instance key.
## Streams
Server settings for Streams, with their defaults, are in
[Streams](https://docs.superdurable.io/zh-Hans/reference/server/configuration.md#streams).
### Write semantics
A direct Step write has these semantics:
- Dex sends it immediately during **WaitFor** or **Execute**. It is not buffered
until the Step completes.
- A later Step failure does not roll the message back.
- One method invocation can write any number of messages to the same or
different Streams. Each write is an implicit heartbeat and preserves the last
explicit heartbeat value. The [Polling](https://docs.superdurable.io/zh-Hans/design-patterns/polling.md) pattern uses
these writes to report progress.
- The SDK only confirms local encoding and delivery to the Worker output stream.
A disabled, unavailable, full, or rejecting Stream Store drops the message
without failing the Step.
- Every Step message has source **#StepExecutionID**. Attempts and messages from
the same Step execution share that source. Source is informational, not
unique, and does not deduplicate retries. After a Dex Server node crashes, an
ASYNC Step can also run the same attempt number again, and its batches can
repeat in the same source.
A Client can write even when the Flow instance does not exist or is no longer
active. Supply a non-empty source for each write. A source may be repeated and
may contain **#**; every call appends another message. RPC handlers cannot write
through their Context, but they can send Stream messages through an injected
Client dependency.
Python coroutine Steps call **Stream.write** without **await**. Python
synchronous Steps must yield the returned **StepOutput** from a generator. Other
SDKs send the frame through their Step Context and return after local
enqueueing.
### Buffer text chunks
Use the buffered text writer when many small chunks form one text value, such as
LLM deltas. The default flush interval is one second and the soft threshold is
16 KiB of UTF-8 data. A timer, the size threshold, or invocation completion
emits the current non-empty batch. The helper preserves text exactly, never
splits a chunk, and ignores empty chunks.
Go, Java, Python async, TypeScript, and Rust stop the timer and flush the tail
before sending the final result or error. Python sync generators use cooperative
elapsed-time checks and must yield from an explicit final flush. An empty buffer
does not produce a Stream message or implicit heartbeat. Retry does not restore
unsent text, and batches sent before a retry may appear again.
### Resumable reads
**ReadStream** returns one message with its value, resume token, creation time,
and source.
- An empty token starts at the current retained head.
- A token older than the retained head also starts at that head.
- Pass the returned token unchanged to read the next message.
- When no next message exists, the call long-polls until a message arrives or
the wait expires.
Resume is best effort. A slow reader can miss messages trimmed before its next
read.
### List messages
Use **ListStreamMessages** to inspect retained messages from newest to oldest.
The call is non-blocking. It returns immediately with up to the requested page
size.
An empty before-page token starts at the retained tail. Pass the returned
next-page token unchanged to read the next older page. The token is exclusive
and bound to the Flow type, Flow ID, and Stream name. An empty next-page token
marks the final page.
**Python**
```python
page = await app_state.client.list_stream_messages(
required_query("workflowId"),
app_state.stream.progress,
required_int_query("pageSize"),
optional_query("beforePageToken", ""),
)
```
例子: [examples/python/dex_examples/primitives/stream/controller.py](https://github.com/superdurable/dex/tree/main/examples/python/dex_examples/primitives/stream/controller.py)
**Go**
```go
var page sdk.StreamMessagesPage[string]
err = controller.client.ListStreamMessages(
request.Request.Context(),
flowID,
Progress,
int32(pageSize),
request.Query("beforePageToken"),
&page,
)
```
例子: [examples/go/primitives/stream/controller.go](https://github.com/superdurable/dex/tree/main/examples/go/primitives/stream/controller.go)
**Java**
```java
final StreamMessagesPage page = client.listStreamMessages(
workflowId,
flow.progress,
pageSize,
beforePageToken);
```
例子: [examples/java/src/main/java/io/superdurable/dex/primitives/stream/StreamController.java](https://github.com/superdurable/dex/tree/main/examples/java/src/main/java/io/superdurable/dex/primitives/stream/StreamController.java)
**TypeScript**
```typescript
const page = await client.listStreamMessages(
String(request.query.workflowId ?? ""),
progress,
Number(request.query.pageSize),
String(request.query.beforePageToken ?? ""),
);
```
例子: [examples/typescript/src/primitives/stream/controller.ts](https://github.com/superdurable/dex/tree/main/examples/typescript/src/primitives/stream/controller.ts)
**Rust**
```rust
client
.list_stream_messages(
&query.workflow_id,
&PROGRESS,
query.page_size,
&query.before_page_token,
)
```
例子: [examples/rust/src/primitives/stream/controller.rs](https://github.com/superdurable/dex/tree/main/examples/rust/src/primitives/stream/controller.rs)
Listing is a best-effort retained-message snapshot, not a transactional
snapshot. New messages appended after the first page do not enter that
older-page chain, but trimming can remove messages between calls. If trimming
moves the retained head past the page anchor, Dex returns an empty page instead
of restarting from the tail.
The page size must be between 1 and **streamStore.maxReadMessages**, which
defaults to 1000. The server rejects a page size outside that range with
**INVALID_ARGUMENT**; it does not reduce a larger value.
Use **ReadStream** for forward, one-message-at-a-time consumption that can
long-poll and resume. Use **ListStreamMessages** for reverse, non-blocking
pagination over currently retained messages.
### Capacity and trimming
**streamCapacityBytes** is an approximate capacity shared by all Flow executions
with the same Flow type and Stream name. It is not a per-Flow-ID limit.
The Dex server calculates the estimated size from each message's serialized
value, Flow ID, source, and configured per-message overhead.
When usage reaches the trim trigger threshold, background trimming removes the
oldest messages across all instances until usage reaches the trim target
threshold. A server runs several trim workers, but each Stream is trimmed by one
job at a time. If trimming is too slow and usage reaches the capacity limit
(**streamCapacityBytes**), the write request is rejected with
**RESOURCE_EXHAUSTED**. The rejection also asks trimming to free room for that
message, so a Client can retry the write. A message larger than the Stream's
capacity is rejected the same way, but it can never fit, so do not retry it.
The server also limits the size of each Stream message, the capacity of one
Stream, and the memory of all Streams on a server. A Client write past the
server-wide total fails with **RESOURCE_EXHAUSTED**, and background trimming
then removes the oldest messages of the largest Streams until the message fits.
Retry the write after a short delay. See
[Streams](https://docs.superdurable.io/zh-Hans/reference/server/configuration.md#streams) for each limit and its
default.
### Best-effort storage
The Stream Store keeps messages in the memory of the active server process.
**streamStore.backend** defaults to **memory** for both **dexcli dev** and a
production server. Memory contents disappear when the server process restarts or
a standby takes over. Trimming or capacity pressure can also remove or reject
messages. An operator can set the backend to **disabled**. Client Stream writes
and reads then fail with **FAILED_PRECONDITION**, and Step writes are dropped
without failing the Step.
## SubFlows
### SubFlowOptions
**SubFlowOptions** are the equivalent of **StartFlowOptions** on the
**StartFlow** API. Most options are the same, except for
**SubFlowReusePolicy**.
### SubFlowReusePolicy
Dex Server generates a SubFlow ID from the parent Flow ID, Step execution ID,
and condition index. When a parent reuses a Flow ID with **IDReusePolicy**, or
time travels, it can enter the same Step execution and try to start the same
SubFlow ID again. **SubFlowReusePolicy** controls what happens in that case.
The default **RESTART_IF_PREVIOUS_EXITS_ABNORMALLY** is usually what a parent
wants: continue waiting for a running child, use a successful child result, and
restart a child that ended abnormally.
| Existing child Flow | **ATTACH** | **RESTART_IF_PREVIOUS_EXITS_ABNORMALLY** | **ALWAYS_RESTART** |
| ------------------------------------------ | ----------------- | ---------------------------------------- | ------------------- |
| No existing Flow | Start | Start | Start |
| Running | Attach | Attach | Terminate and start |
| Completed | Return its result | Return its result | Start |
| Failed, canceled, timed out, or terminated | Return its result | Start | Start |
Dex Server first compares the Request ID it generated for this child start. A
matching Request ID always attaches to a running child or returns its terminal
result, regardless of **ReusePolicy**. This makes a retry safe when Dex Server
accepted the original start but the parent did not receive the response. For a
different Request ID, Dex Server applies the selected **ReusePolicy** from the
table.