Applications read and write Flow state through typed RPCs. The Client does not expose direct Attribute reads or writes, Channel publication, or pending-message mutation. This keeps each external state transition behind a Flow-owned method.
An RPC can stage channel.delete(context, messageId). Annotate it with
@RPC(isTransactional = true) when a missing ID must abort every other RPC
write. Attribute locks already select transactional execution, but Channel
deletion does not do so implicitly.
RPCs receive ordinary Attributes and all Channel size metadata by default. AttributeMap entries and pending Channel messages are opt-in:
@RPC(
loadAttributeMapInstances = {"Items/tenant-a"},
loadChannels = {"Queued"},
loadChannelMaps = {"ByTenant"}
)
public RPCResult<Snapshot> snapshot(Context context) {
return RPCResult.of(new Snapshot(queued.pendingMessages(context)));
}Put a map name in loadAttributeMaps or loadChannelMaps to load every current instance. Use the singular instance options for the less common exact instance case. Text after the first slash is the logical instance and the SDK escapes it. A selected empty queue returns an empty list; reading an unselected map entry throws AttributeMapNotLoadedException. An unselected pending-message snapshot throws ChannelMessagesNotLoadedException. Pending messages preserve FIFO order and include the server-assigned message ID. The snapshot does not change after the handler stages a publish or deletion.
Loading controls which data reaches the Worker. isTransactional controls atomic commit and Channel deletion validation. Attribute locks add isolation only among cooperating Steps and RPCs using the same lock. Write-only publish and delete operations do not require loading.
@RPC remains the registration-time contract for timeout, transactionality,
locks, and fixed state loads. Use RPCInvokeOptions when the caller chooses an
exact map instance at runtime:
RPCInvokeOptions options = RPCInvokeOptions.newBuilder()
.addLockAttributeMapInstance(customerProfiles, partitionName)
.addLoadAttributeMapInstance(customerProfiles, partitionName)
.build();
CustomerProfile result = client.invokeRPC(
directoryRpc::upsertCustomerProfile,
profile,
options);The four function/procedure and input/no-input invokeRPC families accept these
options. Selections are additive, sorted, and deduplicated. Locking and loading
are independent; read-modify-write handlers need both for the same instance.
See the runnable sequential chunking and hash partitioning patterns for complete call sites.
StepOptions provides the same five selections independently for waitFor and
execute: all AttributeMap instances, exact AttributeMap instances, Channels,
all ChannelMap instances, and exact ChannelMap instances. The methods receive
independent snapshots. Execute reads after the winning Wait consumes messages;
retries of one logical method call reuse its first snapshot.
FlowTimeoutHandlerOptions provides Execute-style timeouts, heartbeat detection,
retry, durability, Attribute locks, and the same state selections. Set it on
StartFlowOptions or SubFlowOptions only for a positive timeout using HANDLER.
Exhausted retries may proceed to a registered Step<Void>; read the final
failure with Context.getRecoveryError().
Java SDK for Dex workflow engine
See samples for how to use this SDK to build your workflow.
Maven coordinates: io.superdurable:dex-sdk (namespace for domain superdurable.io).
The canonical rewrite API is under io.superdurable.dex. This phase includes
strongly typed workflow definitions, persistence handles, registry
validation, synchronous gRPC Client operations, and a Java pull Worker.
Rust is used only by the DXBC BlobCache JNI binding.
Java workflows and steps are interfaces, while RPCs keep the annotation and typed-stub model:
public final class IncrementStep implements Step<Long> {
@Override
public Class<Long> getInputType() {
return Long.class;
}
@Override
public StepDecision execute(Context context, Long input) {
return StepDecision.gracefulComplete(input + 1L);
}
}
public class CounterFlow implements Flow<Long> {
private final IncrementStep start = new IncrementStep();
@Override
public StepList<Long> getSteps() {
return StepList.startStep(start);
}
@RPC(name = "increment", timeoutSeconds = 10)
public RPCResult<Long> increment(Context context, Long amount) {
return RPCResult.of(amount + 1L);
}
}
CounterFlow stub = client.newRpcStub(CounterFlow.class, flowId, runId);
long value = client.invokeRPC(stub::increment, 1L);Flows exposing RPC methods and their annotated methods must not be final.
RPC stubs intercept those methods without invoking Flow constructors.
In Kotlin, declare the Flow class and its RPC methods with open.
Step<I> declares Class<I> getInputType(); parameterized Step inputs are not
supported. Steps, attributes, and channels declare Java classes, not codecs:
Attribute<String> status = Attribute.define("status", String.class);
Channel<Void> wakeup = Channel.define("wakeup", Void.class);Wait factories read from the domain nouns they create:
return Wait.until(
Timer.byDuration(Duration.ofSeconds(1)));StepOptions.waitForMethodTimeout(...) and executeMethodTimeout(...) bound
the two handler calls. Timer and channel conditions determine how long a Step
waits.
waitForRetry(...) and executeRetry(...) limit one logical handler
execution. Attempt numbers start at 1. The retry policy, method timeouts, and
heartbeat timeout apply to every attempt with either Step durability.
Dex defaults Step methods to SYNC durability, a four-hour retry duration, a
two-hour attempt timeout, and a one-minute heartbeat timeout. A Step method
durability override wins over FlowConfig.stepDurability(...), which wins over
SYNC. Failure policies do not change that order. Durability changes only when
Dex records a Step method result; it does not change retries, timeouts, or
heartbeats.
Dex does not heartbeat a Step automatically. Long-running Step handlers should emit progress before the configured heartbeat timeout:
@Override
public StepDecision execute(Context context, Batch batch) {
int start = context.hasLastHeartbeatValue()
? context.getLastHeartbeatValue(Integer.class)
: 0;
for (int index = start; index < batch.items().size(); index++) {
process(batch.items().get(index));
context.recordHeartbeat(index + 1);
progress.write(context, "processed " + (index + 1));
}
return StepDecision.gracefulComplete();
}recordHeartbeat(value) persists a nonnull checkpoint for a later attempt.
recordHeartbeat(null) sends a heartbeat without details and clears the current
checkpoint. Heartbeats behave the same with either Step durability.
A Stream write is an implicit heartbeat that preserves the latest explicit
heartbeat details.
Step Stream writes travel on the Worker response stream and are fire-and-forget.
One handler may write the same Stream any number of times. Dex assigns
#<stepExecutionID> as source metadata. Storage rejection does not fail the
handler, although local validation, serialization, and Worker transport errors
can still fail it.
For text deltas, create one invocation-managed writer and pass its write
method to the producer:
BufferedTextStream progress = BufferedTextStream.create(context, thinking);
progress.write(delta);It flushes after one second, at a soft 16 KiB UTF-8 threshold, or before the
final result or error. The configuration overload accepts a Duration and byte
threshold. It preserves chunks exactly, and an empty buffer does not emit a
message or heartbeat. Retry does not restore unsent text or deduplicate batches.
Flow timeout handlers and RPCs cannot send heartbeat or Stream progress.
Client.writeStream(flowId, stream, source, value) appends on every call.
Sources are required metadata, may repeat, and may contain #. Read them with
StreamMessage.getSource(); they do not provide deduplication.
Client.readStream moves forward one message at a time and can long-poll.
Client.listStreamMessages returns a non-blocking newest-first retained page.
Pass the registered typed Stream, a positive page size, and an empty token for
the first page. Pass StreamMessagesPage.getNextPageToken() unchanged to read
older messages until it is empty. The server caps page size at 1000 by default.
A trimmed anchor returns an empty page, and newer concurrent writes do not enter
an existing older-page chain.
Override Flow.handleTimeout to make a positive timeout use handler policy by
default. Dex invokes it as an Execute-only system Step after its durable timer
fires or is skipped:
@Override
public StepDecision handleTimeout(Context context) {
return StepDecision.forceComplete("expired");
}
StartFlowOptions options = StartFlowOptions.newBuilder()
.timeout(Duration.ofMinutes(30))
.timeoutPolicy(FlowTimeoutPolicy.HANDLER)
.build();FAIL produces FlowErrorType.FLOW_TIMEOUT and fails the Flow; CANCEL
cancels it. SubFlowOptions exposes the same timeoutPolicy(...)
override. The deadline and handler execution are preserved across Flow state
transfers. A zero or absent timeout disables
the feature.
A successful Step can cancel queued or active executions by registered Step type while still choosing its normal next or close action:
@Override
public StepDecision execute(Context context, Quote quote) {
return StepDecision.goTo(RecordQuote.class, quote)
.withCancelingSiblingSteps(QuoteCarrierA.class, QuoteCarrierB.class);
}withCancelingSteps(...) selects every queued or active execution of each Step
type in the current Flow. withCancelingSiblingSteps(...) selects only
executions whose Context.getFromStepExecutionId() is the same as the current
execution's source. Both methods are immutable: repeated calls form a union,
duplicate registered Step classes are removed, and a Flow-wide selector supersedes a
sibling selector for the same Step type. Passing an unregistered or null Step class
causes an invalid Step result.
Dex resolves selectors from a snapshot after the current execution succeeds. Executions already completed, already canceled, or absent are no-ops. Next Steps created by the same decision are outside that snapshot. Dex cancels the selected executions in workflow state and immediately applies the decision's next or close action; it does not wait for selected Java handlers to return. Their late decisions, Attribute and Channel writes, outputs, retry paths, and configured failure-proceed Steps are discarded.
An RPC can apply a Flow-wide Step type selector while returning its output and scheduling next Steps atomically:
@RPC(name = "acceptQuote")
public RPCResult<QuoteStatus> acceptQuote(Context context, Quote quote) {
return RPCResult.of(
QuoteStatus.ACCEPTED,
StepMovement.of(recordQuote, quote))
.withCancelingSteps(carrierA, carrierB, globalQuoteTimeout);
}RPCResult does not support sibling selection because separate invocations of
the same RPC name do not share an invocation-scoped Step execution lineage.
Attribute and Channel writes from the RPC commit before the cancellation
snapshot, while next Steps in the same result remain outside it.
ClientOptions and WorkerOptions contain a default Jackson ObjectMapper and
accept a configured mapper when needed. Java does not expose a public Codec API.
Workers use a builder so optional transport settings remain readable:
WorkerOptions options = WorkerOptions.newBuilder()
.serverAddress("localhost:8801")
.objectMapper(objectMapper)
.grpcErrorStatusMapping(GrpcErrorStatusMapping.newBuilder()
.map(PaymentDeclinedException.class, Status.Code.FAILED_PRECONDITION)
.build())
.build();serverAddress connects to Dex for startup synchronization, blob hydration,
and outbound task streams. Workers need only an outbound connection to the
Dex Server. Flow starts and SubFlows select pull Workers automatically.
Client.updateFlowConfig leaves the Worker target unset, preserving the Flow's
existing routing when other configuration fields change.
Every Worker receives tasks through outbound streams. Omit pull to use the
PullWorkerOptions defaults, or configure task capacity and shutdown behavior:
Worker worker = new Worker(registry, blobCache, WorkerOptions.newBuilder()
.serverAddress("dex:8801")
.pull(PullWorkerOptions.newBuilder()
.maxConcurrentTasks(8)
.build())
.build());
Client client = new Client(registry, blobCache,
new ClientOptions("dex:8801"));
new Thread(worker::start).start();Each Worker keeps exactly maxConcurrentTasks long-lived execution-slot streams.
Every stream serves the sticky Worker identity and every registered Flow type;
its registration has an empty queue. A stream is ready while idle and busy while
its task runs. The Worker executes every delivered task and never rejects for
capacity. It can reject an unknown Step or RPC before invoking its handler.
The default slot count is the available CPU count. Worker start fails if it
exceeds the Server's reported matching.maxConcurrentTasksCap per-Worker limit.
The deprecated idleStreamsPerQueue option is accepted but has no effect.
Heartbeat and Stream writes from cancelled or completed invocations are rejected
locally, including after stream reuse. A stream error cancels its running task;
that task's output is never sent over a later connection. The SDK sends a Pong
every 20 seconds during execution and stops before the final frame.
The default app Worker ID uses DEX_WORKER_INSTANCE, POD_NAME, HOSTNAME, or
the local hostname, plus eight random base32 characters. If the identity cannot
be read, it uses worker. A new Worker gets a new ID. An explicit appWorkerID
must have 1–63 ASCII letters, digits, periods, underscores, or hyphens and must
not equal a registered Flow type. A Flow type used for pull routing must be
unique within the Dex server.
stop() and close() start no new task. Idle streams immediately half-close
the request side and continue reading until the Server ends the stream. Busy
streams finish their handler and send its final output before half-closing.
Task frames arriving after stop begins are dropped without invoking a handler,
including after half-close. After stopDrainTimeout (30 seconds by default),
the Worker resets streams with running tasks and half-closed streams that the
Server has not ended, cancels handlers, and waits for them to return. A task the
Server sent on a reset stream may have run. A stream that has sent a final
output, but has not seen the Server end the stream or assign its next task,
instead gets a five-second flush grace before it is reset. This holds even when
its handler has returned or the stream has half-closed. So stop() and
close() can return up to five seconds after the timeout. A
handler that ignores cancellation can delay stop.
Reconnects and transient transport errors are retried automatically.
Call waitPullRegistered(Duration) after starting the Worker on another thread.
Readiness requires at least one STEP_AND_RPC registration and, when reserved
RPC slots are configured, at least one RPC_ONLY registration. Timeout throws
TimeoutException; interruption cancels only the wait. Startup failures and
uncovered permanent registration rejections reach the waiter. Stopping before
readiness throws IllegalStateException. Success latches after initial readiness,
including after stop.
getPullRegistrationState() provides immutable snapshots. The legacy queue
fields (PullRegistrationError.getQueue() and each snapshot's getQueue())
now hold POLL_WORKER_SLOT_KIND_STEP_AND_RPC or POLL_WORKER_SLOT_KIND_RPC_ONLY,
rather than a single routing queue. Entries expose hasRegisteredStream() and
getLastPermanentError(). Readiness stays true through transient disconnects
and becomes false when a required slot kind has a permanent error and no
registered stream. Re-registration clears that error.
stopOnPermanentStreamRejection(true) drains and fails start() when a permanent
rejection leaves a required slot kind without a registered stream. Another
registered stream of that kind prevents shutdown. By default such registration
errors retry; an excessive concurrency setting always fails startup without
background retry. Both stop() and close() remain safe after startup fails.
reservedRpcTasks selects slots that accept only RPCs; the remaining slots
accept Steps and RPCs. The Server routes Steps to the latter slots. The Worker
never declines a Step for reserved capacity. By default, one slot is reserved
when maxConcurrentTasks is at least two; otherwise none are reserved. Set zero
to opt out. The reserve must be nonnegative and less than the resolved slot count.
With one slot and no explicit reserve, startup warns about RPC starvation.
PullWorkerOptions pull = PullWorkerOptions.newBuilder()
.maxConcurrentTasks(8).reservedRpcTasks(2).build();The deprecated getPullReservedRpcDeclines() counter remains available and
always returns zero.
Set a Credential on Client and Worker options when Dex has authentication
enabled. The SDK asks for a current token before every FlowService call. A pull
Worker also asks when each WorkerTaskService stream opens, including reconnects.
The default null credential sends no authorization metadata.
Credential credential = new FileCredential("/var/run/secrets/dex/token");
ClientOptions clientOptions = ClientOptions.newBuilder()
.serverAddress("dex:8801")
.credential(credential)
.build();
WorkerOptions workerOptions = WorkerOptions.newBuilder()
.serverAddress("dex:8801")
.pull(PullWorkerOptions.newBuilder().build())
.credential(credential)
.build();FileCredential reads and trims the file at construction. It checks its
modification time and size at most once per second and rereads changed content.
An unreadable or empty file fails calls instead of sending a stale token.
Use ApiKeyCredential when the application holds a long-lived API key and the
token endpoint issues short-lived access tokens for a supplied Dex server ID:
ApiKeyCredential credential = ApiKeyCredential.builder(
URI.create("https://auth.example.com/token"), "dex-server-id", apiKey)
.exchangeTimeout(Duration.ofSeconds(10))
.build();
try (ApiKeyCredential ownedCredential = credential;
Client client = new Client(registry, blobCache,
ClientOptions.newBuilder().serverAddress("dex.example.com:8801")
.credential(ownedCredential).build())) {
client.searchFlows("", 10);
}The endpoint must use HTTPS; HTTP is allowed only for localhost, 127.0.0.0/8,
or ::1. The first call exchanges the key. A token call after 80% of the token
lifetime starts one background refresh, and callers keep using the current token until it
expires. Concurrent calls share an exchange. Failed refreshes are retried at
most every five seconds while the token remains valid. An expired token waits
for the exchange, bounded by the exchange timeout. When Dex rejects a token,
the credential discards only that token; unary service calls retry once, while
pull streams reconnect once immediately with a new token. A second rejection is a
permanent stream error. The application closes the credential after every Client and
Worker sharing it has closed. No timer runs between token calls. A custom
HttpConnectionFactory can open connections through an application-selected
proxy; the credential disconnects each connection after use.
Redirect responses fail without forwarding the API key. A call deadline or
cancellation bounds a token wait during a unary retry.
Credential descriptions contain no secrets. Exchange errors redact the API key,
cached and two most recently issued access tokens, the failed response's access token,
and JWT-shaped substrings before truncating each endpoint error field to 200 characters.
A JWT-shaped substring has three or more base64url segments of at least 10 characters
each. The rule also applies to the token URL and can redact other text of that shape,
such as a long dotted host name. Raw transport causes are omitted.
Custom Credential implementations must support concurrent calls and should
avoid unbounded blocking. An exception fails the call before it reaches Dex. A configured gRPC
deadline or cancellation ends a call even if token lookup blocks, but the lookup
continues to occupy a channel executor thread until it returns. Client calls
report credential failures as DexServiceException with UNAUTHENTICATED;
Worker startup reports an authentication error. Bearer
tokens travel over the same connection as other Dex calls. Configure TLS when
the network path requires encryption and server authentication.
Set flowServiceTls on both ClientOptions and WorkerOptions to connect to a
TLS endpoint, such as a TLS-terminating gateway. Omitting it keeps plaintext gRPC
for local dexcli dev connections. An empty FlowServiceTlsOptions builder enables
TLS 1.2 or newer with server certificate verification against the JDK default
trust store. The Worker uses it for startup, blob hydration, and pull task streams.
FileCredential, ApiKeyCredential, and custom credentials send bearer tokens as
call metadata on the same TLS connection.
import io.superdurable.dex.BlobCache;
import io.superdurable.dex.Client;
import io.superdurable.dex.ClientOptions;
import io.superdurable.dex.Credential;
import io.superdurable.dex.FileCredential;
import io.superdurable.dex.Flow;
import io.superdurable.dex.FlowServiceTlsOptions;
import io.superdurable.dex.Registry;
import io.superdurable.dex.Worker;
import io.superdurable.dex.WorkerOptions;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.time.Duration;
final class TlsExample {
static <I> void startOverTls(Registry registry, BlobCache cache,
Flow<I> flow, String flowId, I input) throws Exception {
FlowServiceTlsOptions tls = FlowServiceTlsOptions.newBuilder()
.additionalRootCAsPem(new String(Files.readAllBytes(
Paths.get("/etc/dex/root-ca.pem")), StandardCharsets.US_ASCII))
.build();
Credential credential = new FileCredential("/var/run/secrets/dex/token");
String serverAddress = "dex.example.com:443";
try (Worker worker = new Worker(registry, cache, WorkerOptions.newBuilder()
.serverAddress(serverAddress).flowServiceTls(tls)
.credential(credential).build());
Client client = new Client(registry, cache, ClientOptions.newBuilder()
.serverAddress(serverAddress).flowServiceTls(tls)
.credential(credential).build())) {
Thread workerThread = new Thread(worker::start);
workerThread.start();
try {
worker.waitPullRegistered(Duration.ofSeconds(30));
client.startFlow(flow, flowId, input);
client.waitForFlow(flowId, Duration.ofSeconds(30));
} finally {
worker.stop();
workerThread.join();
}
}
}
}additionalRootCAsPemadds PEM root certificates to every root in the active JDK default trust store, including a store selected withjavax.net.ssl.trustStore. Null or an empty string uses only default roots. A nonempty bundle must contain at least one parsable X.509 certificate. Other text and malformed blocks are ignored when another certificate parses. The bundle is read at construction; it is not watched for changes. The combined store uses the JVM's default KeyStore type. Adding roots fails if the default trust manager reports no accepted issuers, because the SDK cannot preserve its trust anchors. No new dependency is needed.serverNameoverrides the address host for certificate verification, SNI, and HTTP/2:authority. Omit it when the certificate covers the address host. Supply a DNS name or unbracketed IP address without a port. An explicit empty string is invalid. The override is sent as:authoritywithout a port, so the gateway must accept that value. Prefer a DNS override for IPv6 endpoints.- With TLS,
serverAddressmust behost:port,dns:///host:port, ordns://authority/host:port, with an unsigned port from 1 through 65535. IPv6 address hosts require brackets. The DNS resolver authority never becomes the certificate name or HTTP/2 authority. Java gRPC uses the JVM's DNS resolver. Other resolver schemes remain available only with plaintext connections. - The JDK trust manager provider must return
X509ExtendedTrustManagerfor both default and combined roots so the transport can verify hostnames. Providers returning onlyX509TrustManagerfail at Client/Worker construction. - The endpoint must negotiate HTTP/2 with ALPN
h2. HTTP/1.1-only endpoints and endpoints without ALPN are rejected. An IP endpoint requires a matching IP SAN. Netty omits SNI for IP addresses and single-label DNS names such aslocalhost. - Invalid addresses and TLS options throw
IllegalArgumentExceptionat options or Client/Worker construction. Connections open lazily, so certificate, hostname, and handshake failures appear asUNAVAILABLEon the first Client call or as a Worker startup failure. - These settings do not configure the
ApiKeyCredentialtoken endpoint, which uses its own HTTPS connection and allows HTTP only on loopback.
Serialized object values use wire encoding json for JSON and raw for raw bytes.
Internal Blob references are opaque. Hydration sends the owning Flow ID with
each reference, and the local cache is isolated by (flowId, blobRef).
Every Throwable escaping application Step waitFor, Step execute, or RPC
code becomes a structured Worker error. The default gRPC status is INTERNAL,
including when application code throws StatusException or
StatusRuntimeException. Configure GrpcErrorStatusMapping only for exception
classes with a stable meaning to the owner of the Worker application. The most
specific mapped superclass wins; an unmapped class remains INTERNAL. These
diagnostic statuses do not change Step retry or failure policies. Capacity and
other Worker transport failures retain their own SDK-selected statuses.
Java Worker errors include the original exception type, detail, and stack trace, including causes and suppressed exceptions. To keep the encoded worker failure within transport limits, Dex persists at most 256 bytes of type, 1 KiB of detail, and 4 KiB of UTF-8 stack trace. Each truncated field receives a marker without splitting a character. The Worker log keeps the complete local stack.
Failure-policy recovery exposes the previous method error through
context.getRecoveryError(). Wait-for recovery passes the error directly to
execute in the same run. Execute recovery passes it to both methods of the
configured recovery Step, including when that Step starts after
the Flow's state has been transferred to a new run:
RecoveryErrorInfo error = context.getRecoveryError();
if (error != null) {
logger.warn("Recovering from " + error.getErrorType() + ": " + error.getDetail());
}To override the next retry interval, throw a retry request with the current attempt failure as its cause:
throw RetryAfterException.after(Duration.ofSeconds(30), currentFailure);The delay must be a positive whole number of seconds in the signed 32-bit range. Retry limits and method timeouts remain unchanged. Recovery exposes only the final failure type and detail; stack traces remain in failure history, and retry delays are scheduling metadata.
Flow<StartInput>.getSteps() returns StepList<StartInput>. Start with
StepList.startStep(step) and append heterogeneous Steps with otherSteps(...).
Use StepList.empty() when a Flow has no Steps, or
StepList.withoutStartStep(...) when every Step is RPC-triggered. The generic
binding catches start-input mismatches during compilation; Registry also
validates the runtime classes with
inputType.isAssignableFrom(registeredType).
waitForFlow returns every output-bearing Step completion after hydrating all
blob-backed values. Decode a single-output Flow with the strict helper:
String output = client.waitForFlow("flow-123").getSingleOutput(String.class);For multiple outputs, select by Step type or Step execution ID and decode each completion independently:
FlowResult result = client.waitForFlow("flow-123");
for (StepCompletion completion : result.getCompletions()) {
if (completion.getStepType().equals("ChargeCard")) {
Receipt receipt = completion.getOutput(Receipt.class);
}
}The completion list is immutable and preserves server collection order, but
parallel branch order is not deterministic. getSingleOutput throws when the
Flow has zero or multiple outputs. A no-output Flow returns an empty list.
Step executions use StepExecutionId.of(stepType) or
StepExecutionId.of(stepType, executionNumber). Constructors are intentionally
not part of the public API.
Wait.until, Wait.allOf, and Wait.anyOf use unnamed Conditions by default.
Do not add Condition IDs merely because a Condition is nested in one of these
waits. Every Condition in Wait.anyCombinationOf must have a non-empty user ID;
the same Condition object may be reused across combinations.
SubFlow.run(ChildFlow.class, input) declares a normal Flow execution as a
durable condition. Read it during execute with
SubFlow.getConditionResults(context, index). allOf results are terminal;
an unfinished anyOf loser is a RUNNING snapshot. Obtain its deterministic
Flow ID with SubFlow.getFlowId(context, index) before passing it to
Client.stopFlow. A state transfer retains only each SubFlow Condition ID and
stable list position, not its starting input or options. SubFlows continue when
their parent closes.
client.waitForAttributeMatch(...) overloads cover singleton Attributes and
AttributeMap instances in the current run. Build a match with one of the six
AttributeMatch factories. The method returns the decoded matched value.
String and boolean support equality operators; integers and floating-point
values support every operator. Every AttributeMap and ChannelMap instance must
be nonblank and must not contain /. Objects, bytes, null, non-finite doubles,
and invalid ordering fail locally. AttributeMap.getMapSize/getAllInstanceKeys reflect
buffered sets and deletes. The corresponding ChannelMap methods are RPC-only,
include buffered publishes, and omit empty instances. Returned keys are decoded
and sorted. Conditional completion is
StepDecision.forceCompleteIfChannelsEmpty(...).
Request IDs are optional for both waits. The Server does not read them for these
waits. requestTimeout bounds the total SDK call across transport reattachments;
zero waits indefinitely. Request expiry throws RequestTimeoutException.
A caller disconnect releases its wait slot.
client.getAttributes(flowId, reads...) reads committed Attribute values
from outside the Flow without an RPC. Pass singleton Attribute definitions
and AttributeMap instances selected with AttributeMap.instance(key). A
List<AttributeRead<?>> overload accepts reads built at runtime. The Client
sends one request for all keys, so every value comes from the same committed
state, and no Worker runs. The call reads the current run, or the latest run
when none is open, so a closed run stays readable until Dex purges it. It does
not see writes that a running Step or RPC has not committed yet.
AttributeValues values = client.getAttributes(
"order-42",
orderFlow.orderStatus,
orderFlow.notes.instance("n1"));
String status = values.get(orderFlow.orderStatus).orElse("new");
Optional<Note> note = values.get(orderFlow.notes.instance("n1"));AttributeValues.get returns Optional.empty() for an unset Attribute or
instance, and also for a stored value that decodes to null. It throws
IllegalArgumentException for a read that was not part of the call. An equal
AttributeMapInstance (same AttributeMap object and instance key) works. Every
definition must be the exact object that a registered Flow's persistence
schema returns; another object with the same name fails with
FlowDefinitionException before any request. A missing Flow fails with
FlowNotFoundException. Hard-delete Attributes are readable, large values are
loaded through LoadBlobs and the BlobCache, and a value that cannot be decoded
fails with ValueMappingException. Every error names the Flow ID, and an error
about one read names its Attribute key.
Public exceptions live in io.superdurable.dex.exceptions. Catch concrete
types for expected outcomes instead of comparing ErrorSubStatus:
try {
client.invokeRPC(stub::updateOrder, input);
} catch (FlowNotActiveOrNotFoundException inactive) {
// The Flow never existed or is already closed.
} catch (RpcLockConflictException conflict) {
// Retry the RPC after lock contention.
} catch (WorkerInvocationException workerFailure) {
log.error(
"Worker {} failed: {}",
workerFailure.getWorkerErrorType(),
workerFailure.getWorkerErrorDetail());
}FlowNotFoundException is returned by read and history operations such as
describeFlow, getAttributes, waitForFlow, and timeTravel. These operations
can read a closed Flow, so failure means no matching execution was found.
FlowNotActiveOrNotFoundException is returned by RPC, stop, timer,
configuration, and step-wait operations that require an open Flow.
RPCs also use this error when the target Flow cannot be found, including query-only reads. A query-only RPC without locks, transactional execution, durable effects, or Server-forced Update routing can read a retained terminal execution. At that confirmed read boundary, this error means no readable target was found; return the application's missing or unavailable result directly, without a lifecycle probe, a short timeout, a retry, or historical Step-output decoding. Preserve other service and Worker failures. For mutations and active-only RPCs, the error can mean either missing or closed and does not prove that the requested action succeeded. RPC success does not prove that the Flow is active.
FlowAlreadyStartedException identifies duplicate starts.
LongPollTimeoutException identifies an expected long-poll timeout.
RequestTimeoutException identifies an exhausted caller-defined request
budget.
FlowClosingException identifies an update-class request rejected during Flow
close cleanup. Re-read the retained terminal Flow state before deciding whether
another action is needed.
Client.waitForFlow returns FlowResult for successful and unsuccessful
terminal statuses. Inspect getStatus, getErrorType, and getErrorMessage
first. Only a completed result carries outputs; an unsuccessful result has none.
WorkerInvocationException preserves the original Worker error type,
detail, gRPC code, and Java stack trace. Read the persisted trace with
getWorkerStackTrace(); it may be empty for Workers implemented by another SDK.
DexServiceException remains the generic service failure and exposes status
metadata for diagnostics.
Local definition and value failures use FlowDefinitionException,
InvalidStepResultException, and ValueMappingException.
Persistence schemas use factory methods instead of builders. Definitions may be passed directly, as one mixed list, or as separate attribute and channel lists:
PersistenceSchema.of(counter, commands);
PersistenceSchema.of(definitions);
PersistenceSchema.of(attributes, channels);Initial values bind each persistence definition to its Java value type directly
in StartFlowOptions.Builder:
StartFlowOptions options = StartFlowOptions.newBuilder()
.addAttribute(status, "created")
.addAttribute(items, "order-1", 1)
.build();Attribute Store synchronization is definition-level and immutable:
Attribute<String> email = Attribute.define("customer-email", String.class)
.syncToAttributeStore();
FlowConfig config = FlowConfig.newBuilder()
.attributeStoreNames("profiles", "audit")
.build();Stores are asynchronous latest-state projections. Every enabled Attribute write
is sent to every selected Store. Deletion writes SQL NULL, and projection
failures do not roll back Flow Attributes. Omitting attributeStoreNames
preserves current targets; attributeStoreNames() disables future
synchronization with protocol presence.
Hard-delete Attributes hold short-lived secrets, such as API keys:
Attribute<String> apiKey = Attribute.define("api-key", String.class)
.hardDeleteOnNull();
AttributeMap<String> tenantKeys = AttributeMap.define("tenant-keys", String.class)
.hardDeleteOnNull();Dex stores their values only in the blob store, never in Flow history. Deleting
the Attribute or setting it to null permanently deletes the stored value, and
the Attribute then reads as absent. Only string and object values, including
byte[], are supported. Numeric and Boolean value types fail at definition
time; broader types such as Object fail when a numeric or Boolean value is
written. The option cannot be combined with an index or
syncToAttributeStore(), and waitForAttributeMatch rejects it.
client.getAttributes reads its committed value. Steps and RPC
handlers receive values inline, and the SDK never caches them in the local
BlobCache. Writes require a Server with an enabled blob store and protocol
version 3 or later. See the Attribute.hardDeleteOnNull() Javadoc for retry,
concurrency, Flow ID reuse, reset, retention, and backup limitations.
The IWF integration inventory is implemented as real Dex E2E tests under
src/test/java/io/superdurable/dex/integ.
They run Java Client and Java Worker against dexcli dev, with Rust used only
for BlobCache. They cover flows, steps, RPCs, persistence, time travel, timers,
failure modes, and options.
Java owns Registry reflection, FlowService client transport, outbound task streams, invocation dispatch, and exception mapping. Worker requests never cross JNI. Synchronous Step and RPC handlers run on a bounded JVM executor so they do not block gRPC event-loop progress. Java failures are logged with their complete stack and returned with their Java type, message, mapped status, and bounded persisted stack.
Client and Worker accept the Java BlobCache interface. The default
BlobCache.open(...) implementation is the shared Rust DXBC cache. Blob-backed
values are hydrated in Java: cache hits are decoded locally, while misses use
FlowService LoadBlobs and populate the cache.
The Maven artifact includes BlobCache native libraries for Linux x86-64 and
ARM64, macOS x86-64 and ARM64, and Windows x86-64. The SDK extracts the matching
library from META-INF/native automatically. Use the absolute path in
-Ddex.blobCache.nativeLibrary=... only to override the packaged library.
Application callbacks, Registry, Client, and Worker remain usable with another
BlobCache implementation and do not load JNI.
Measure warm cache hits and identical-value puts across JNI with an optimized native library:
./gradlew blobCacheBenchmark -PblobCacheBenchmarkThreads=8Sustainable Use License 1.0, with legacy portions under their original terms as described in LEGACY_NOTICES.md.
- Java 1.8+
After publish, artifacts appear on Maven Central (and on MVN Repository with some delay). Javadoc: Java SDK API reference.
implementation 'io.superdurable:dex-sdk:0.0.3'<dependency>
<groupId>io.superdurable</groupId>
<artifactId>dex-sdk</artifactId>
<version>0.0.3</version>
</dependency>Applications implement Flow<I>
and Step<I>. A Flow declares
all registered Steps and the optional start marker in one getSteps() list,
plus its persistence schema and annotated RPC methods. Registry validates these
definitions before Client or Worker startup.
- Protobuf IDL lives in monorepo
protos/dex.proto(no submodule checkout needed). - In "Build, Execution, Deployment" -> "Gradle", choose "wrapper task in Gradle build script" for "Use gradle from".
- Open Gradle tab, click "build" under "build" to build the project
Edit protos/dex.proto, then run make -C ../protos proto to refresh checked-in stubs under src/main/java/io/superdurable/gen/.
Run the JVM/native integration tests and all E2E scenarios against a fresh
dexcli dev environment:
./run-integration-tests.shThe script builds dexcli from the current checkout and starts an isolated
Dex environment. Every E2E case starts an outbound polling Worker. The suite
also runs local JNI and pull dispatch integration tests. Each Worker synchronizes
its registered Indexed Attributes before serving tasks. Gradle builds the Rust
BlobCache JNI library. E2E cases use unique Flow IDs. A clean checkout also requires
Go 1.24+, Node.js 22+, and Rust 1.97+. The suite uses server-zd.
DEX_INTEG_ENGINE=zd remains accepted. Set DEX_SERVER_ZD_BINARY to reuse a
built server-zd binary.
The TLS transport tests generate certificates with the openssl CLI. The
dexDevTest suite also starts an isolated, token-authenticated server-zd behind a
TLS proxy and runs a Flow and pull Worker over TLS. The integration script exports
DEX_SERVER_ZD_BINARY for that fixture; set it explicitly when running the test
directly. Test certificates and private keys live only in temporary directories.
Run the same integration suite with JaCoCo coverage:
./run-integration-tests.sh --coverageThe report combines the local JNI and pull dispatch integration tests with all
dexcli dev E2E scenarios. It measures only production classes under
io.superdurable.dex; contract tests, other unit tests, test fixtures, and
generated protobuf classes are excluded. The XML report is written to
build/reports/jacoco/integration/jacoco.xml, and the browser report starts at
build/reports/jacoco/integration/html/index.html.
The validation command builds a Maven repository, verifies its JAR and POM, then compiles and runs independent Gradle and Maven consumers. Both consumers open the packaged Rust BlobCache library and perform a cache round trip.
./validate-publication.sh 0.0.3-localThe command requires JDK 17+ (to run Gradle 9), Rust 1.97+, and Maven. Build and test JDKs are Gradle toolchains downloaded automatically. Published classes still target Java 8.
Every hand-written public API under io.superdurable.dex must have detailed
Javadoc. The documentation check ignores package-private runtime code and
generated protobuf classes, validates parameter and return tags, and renders
the same documentation shipped in the Maven Javadoc JAR:
./gradlew checkstyleMain javadoc --no-daemonOpen build/docs/javadoc/index.html to review the generated documentation.
Maven Central publishing requires Portal user-token credentials for the
verified io.superdurable namespace and an ASCII-armored GPG private key. Set
the repository secrets MAVEN_CENTRAL_USERNAME, MAVEN_CENTRAL_PASSWORD, and
GPG_SECRET_KEY.
Publish a GitHub release whose tag is sdk-java/vX.Y.Z. The workflow derives
the immutable Maven version from that tag, builds all supported native
libraries, validates Gradle and Maven consumers, signs every artifact, and
closes and releases the Central staging repository. Manual dispatch requires
the same X.Y.Z version as an explicit input.
After the workflow succeeds, wait for the version and its artifacts to appear in the canonical Maven Central directory before updating examples or downstream dependencies. Secondary indexes such as MVN Repository and Javadoc may become available later.
If you'd like to test your changes to the SDK with the workflows in the samples repo, use the local publishing command:
- Run:
./gradlew publishToMavenLocal -PreleaseVersion=0.0.3-local
- In the samples
repo, depend on the same version passed with
-PreleaseVersion. Then run:
./gradlew --refresh-dependencies build
- Once you're done, to remove the locally published version, run:
./gradlew unpublishFromMavenLocal -PreleaseVersion=0.0.3-local
.github/workflows/: the GithubActions workflows- IDL source lives in monorepo
protos/dex.proto(seedocs/design/idl-renames.md) - Generated stubs:
src/main/java/io/superdurable/gen/ script/: some scripts for GithubActions and testingsrc/: Java source codemain/java/io/superdurable/dex/: public SDK and Java runtimemain/java/io/superdurable/gen/: checked-in protobuf and gRPC stubstest/java/io/superdurable/dex/: contract, transport, and integration tests
Worker startup calls GetServerInfo, requires protocol 4, synchronizes Attribute
indexes, and opens the reserved-slot streams. The Java SDK protocol interval is
[4,4]. This SDK requires a Dex Server with protocol version 4 or later
(server-zd newer than v0.1.16). Upgrade the Server before upgrading the SDK.
Absent RPCs, invalid intervals, and negotiation request failures also stop startup.
The compatibility error includes the actual reported values in this message:
This SDK requires a Dex Server with protocol version 4 or later (server-zd newer than v0.1.16). The Server reports protocol versions <min>-<current> (server version <server_version>). Upgrade the Server before upgrading the SDK.
Client construction stays lazy and does no network I/O. Before the first network request of any method, the Client negotiates protocol 4. It caches success and incompatibility; every subsequent request against an older Server returns the same compatibility error. Transient negotiation failures are retried on the next call. Protocol negotiation also establishes hard-delete Attribute support.
Step retry totalDuration starts when the Step is first handed to a Worker.
After an attempt fails it decides whether another retry can be scheduled; it
never ends a running attempt. In recovery, RecoveryErrorInfo.getAttemptProgress()
returns NOT_STARTED when no handler ran, NOT_COMPLETED when an attempt may
have run but none completed, or UNSPECIFIED for older records. Treat
UNSPECIFIED like NOT_COMPLETED when reconciling side effects.
Published JARs store the diagnostic SDK version in the
Implementation-Version manifest attribute. Direct class-directory builds use
dev. The artifact version never participates in the compatibility decision.
Searches return runs that match the query. Include FlowType when using shared
index slots, and keep the query and page size consistent across pages.