Add Dex to Existing Code

View as MarkdownView as Markdown

This page turns code you already run, such as a function, a queue consumer, a cron job, a webhook handler, or a script, into a Flow that continues after a crash, a restart, or a long wait.

TODAY · ONE FUNCTION

1. charge()
2. crash: half done
3. wait for approval
4. ship()
5. refund()

split into Steps

split into Steps

AS A FLOW

Charge→Shipapproval or 24 h→Done↓ failsRefund

Do it with a coding agent

Turn <function> in <file> into a Dex Flow.

1. Read https://docs.superdurable.io/intro/add-to-existing-code.md and treat it
   as the source of truth.
2. Give each call that changes another system its own Step, move waits into
   WaitFor, and keep state in Attributes.
3. Add an idempotency key to every external write.
4. Add an integration test that runs against dexcli dev.

Rules: keep the existing behavior. Do not invent APIs. Ask me before changing
code outside this function.

Install the SDK

Add the SDK and a Worker to the service that runs this code. Add Dex to an existing service shows the install command for each language and where the Worker and Client go.

Start from what you have

Take an order process that runs as one function today:

handle_order(order):
    charge the buyer through the payment API
    save status "charged" in the orders table
    wait until the seller approves (poll the approvals table, remind after 24 hours)
    ship the item through the shipping API
    save status "shipped"
    if shipping keeps failing: refund the buyer, save status "refunded"

A crash between any two lines leaves the order in the middle. A restart either repeats the charge or forgets the order.

Split it into Steps

Give each call that changes another system its own Step. A Step is the unit that Dex saves, so a crash repeats at most the Step that was running.

In your function	In the Flow
Charge the buyer	A Charge Step whose Execute calls the payment API and goes to Ship
Wait for approval, remind after 24 hours	Ship's WaitFor: wait for a message on a Channel or a 24-hour Timer, whichever comes first
Send the reminder	Ship's Execute when the Timer fired: send the reminder, then go to Ship again
Ship the item	Ship's Execute when the approval arrived
Refund after repeated failures	A failure route from Ship to a Refund Step, used when Ship's retries run out
The status column	An Attribute, such as order-status
The approval endpoint	An RPC that publishes the approval to the Channel
The polling loop and the reminder job	Nothing. Dex holds the wait and fires the Timer.

How Ship knows which wait ended

Ship's WaitFor returns both conditions, and Dex wakes the Step when the first one is met. Execute then asks the context whether the Timer fired. If it did, Ship sends a reminder and goes to itself, which waits again. Otherwise the approval arrived, so Ship ships the order and completes the Flow. Order Processing shows the whole Step, with its retry and failure route.

order-processingPython

123456789101112131415161718192021

def wait_for(self, context: Context, input: OrderRequest) -> Wait:
    return Wait.any_of(
        seller_ok.for_one(),
        Timer.by_duration(timedelta(hours=24)),
    )

def execute(self, context: Context, input: OrderRequest) -> StepDecision:
    if context.has_timer_fired():
        self.service.send_email(
            input.email,
            "Reminder: approve shipment",
            "Please approve or provide a tracking number.",
        )
        return go_to(Ship, input)
    self.service.ship_item(
        input.order_id,
        input.test_fail_at_shipping,
        idempotency_key=f"ship-{input.order_id}",
    )
    order_status.set(context, "shipped")
    return graceful_complete(f"shipped:{input.order_id}")

Example: examples/python/dex_examples/products/order-processing

order-processingGo

123456789101112131415161718192021222324252627282930

func (shipStep) WaitFor(
	dex.Context,
	OrderRequest,
) (*dex.Wait, error) {
	return dex.AnyOf(
		SellerOK.ForOne(),
		dex.Timer(24*time.Hour),
	), nil
}

func (step shipStep) Execute(
	ctx dex.Context,
	order OrderRequest,
) (*dex.StepDecision, error) {
	if ctx.HasTimerFired() {
		step.service.SendEmail(
			order.Email,
			"Reminder: approve shipment",
			"Please approve or provide a tracking number.",
		)
		return dex.GoTo(shipStep{}, order), nil
	}
	if err := step.service.ShipItemWithIdempotencyKey(order.OrderID, order.TestFailAtShipping, "ship-"+order.OrderID); err != nil {
		return nil, err
	}
	if err := OrderStatus.Set(ctx, "shipped"); err != nil {
		return nil, err
	}
	return dex.GracefulComplete("shipped:" + order.OrderID), nil
}

Example: examples/go/products/order-processing

orderprocessingJava

1234567891011121314151617181920

@Override
public Wait waitFor(final Context context, final OrderRequest order) {
    return Wait.anyOf(
            sellerOk.forOne(),
            Timer.byDuration(Duration.ofHours(24)));
}

@Override
public StepDecision execute(final Context context, final OrderRequest order) {
    if (context.hasTimerFired()) {
        service.sendEmail(
                order.email,
                "Reminder: approve shipment",
                "Please approve or provide a tracking number.");
        return StepDecision.goTo(Ship.class, order);
    }
    service.shipItem(order.orderId, order.testFailAtShipping, "ship-" + order.orderId);
    orderStatus.set(context, "shipped");
    return StepDecision.gracefulComplete("shipped:" + order.orderId);
}

Example: examples/java/src/main/java/io/superdurable/dex/products/orderprocessing

order-processingTypeScript

1234567891011121314151617

public waitFor(_context: Context, _input: OrderRequest): Wait {
  return Wait.anyOf(sellerOk.forOne(), Timer.byDuration(DAY_MS));
}

public execute(context: Context, input: OrderRequest): StepDecision {
  if (context.hasTimerFired()) {
    this.service.sendEmail(
      input.email,
      "Reminder: approve shipment",
      "Please approve or provide a tracking number.",
    );
    return goTo(Ship, input);
  }
  this.service.shipItem(input.orderId, input.testFailAtShipping, `ship-${input.orderId}`);
  orderStatus.set(context, "shipped");
  return gracefulComplete(`shipped:${input.orderId}`);
}

Example: examples/typescript/src/products/order-processing

order_processingRust

1234567891011121314151617181920212223242526272829

fn wait_for(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult<Wait> {
    Ok(Wait::any_of([
        SELLER_OK.for_one(),
        Timer::by_duration(Duration::from_secs(24 * 60 * 60)),
    ]))
}

fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult<StepDecision> {
    if context.has_any_timer_fired() {
        self.service.send_email(
            &input.email,
            "Reminder: approve shipment",
            "Please approve or provide a tracking number.",
        );
        context.record_event("shipment-reminder", input.order_id.clone())?;
        return Ok(StepDecision::go_to(&Ship::default(), input));
    }
    self.service.ship_item_with_idempotency_key(
        &input.order_id,
        input.test_fail_at_shipping,
        &format!("ship-{}", input.order_id),
    )?;
    context.record_event("ship", input.order_id.clone())?;
    ORDER_STATUS.set(context, "shipped".to_string())?;
    Ok(StepDecision::graceful_complete(format!(
        "shipped:{}",
        input.order_id
    )))
}

Example: examples/rust/src/products/order_processing

The reminder email passes no idempotency key, because a second reminder does no harm. Ship's ship call does pass one.

Keep two calls in one Step only when they succeed or fail together. A Step with two external calls repeats both when it runs again.

Route a failure to a recovery Step

When the retries of a Step run out, its failure route can move the Flow to a recovery Step instead of failing it. In Order Processing, Ship retries, and then the Flow goes to Refund:

order-processingPython

123456789101112131415

def get_step_options(self) -> StepOptions:
    return StepOptions(
        execute_retry=RetryPolicy(
            # One hour in production; 3 seconds keeps the demo short.
            total_duration=timedelta(seconds=3),
        )
    ).on_execute_failure_proceed_to(
        Refund,
        StepOptions(
            execute_retry=RetryPolicy(
                # One hour in production; 3 seconds keeps the demo short.
                total_duration=timedelta(seconds=3),
            )
        ),
    )

Example: examples/python/dex_examples/products/order-processing

order-processingGo

1234567891011121314151617

func (step shipStep) GetStepOptions() *dex.StepOptions {
	return &dex.StepOptions{
		ExecuteRetry: &dex.RetryPolicy{
			// One hour in production; 3 seconds keeps the demo short.
			TotalDuration: 3 * time.Second,
		},
		ExecuteFailure: dex.ProceedToOnExecuteFailure(
			refundStep{},
			&dex.StepOptions{
				ExecuteRetry: &dex.RetryPolicy{
					// One hour in production; 3 seconds keeps the demo short.
					TotalDuration: 3 * time.Second,
				},
			},
		),
	}
}

Example: examples/go/products/order-processing

orderprocessingJava

12345678910111213141516

public StepOptions getStepOptions() {
    return StepOptions.newBuilder()
            .executeRetry(RetryPolicy.newBuilder()
                    // One hour in production; 3 seconds keeps the demo short.
                    .totalDuration(Duration.ofSeconds(3))
                    .build())
            .onExecuteFailureProceedTo(
                    Refund.class,
                    StepOptions.newBuilder()
                            .executeRetry(RetryPolicy.newBuilder()
                                    // One hour in production; 3 seconds keeps the demo short.
                                    .totalDuration(Duration.ofSeconds(3))
                                    .build())
                            .build())
            .build();
}

Example: examples/java/src/main/java/io/superdurable/dex/products/orderprocessing

order-processingTypeScript

1234567891011121314

public getStepOptions() {
  return {
    executeRetry: {
      // One hour in production; 3 seconds keeps the demo short.
      totalDurationMs: 3_000,
    },
    executeFailure: ExecuteFailure.proceedTo(Refund, {
      executeRetry: {
        // One hour in production; 3 seconds keeps the demo short.
        totalDurationMs: 3_000,
      },
    }),
  };
}

Example: examples/typescript/src/products/order-processing

order_processingRust

123456789

fn options(&self) -> StepOptions<Self::Input> {
    StepOptions::new()
        .execute_retry(
            RetryPolicy::new()
                // One hour in production; 3 seconds keeps the demo short.
                .total_duration(Duration::from_secs(3)),
        )
        .on_execute_failure_proceed_to(&Refund::default())
}

Example: examples/rust/src/products/order_processing

The default retry policy keeps retrying a Step for 4 hours and then fails the Flow; a failure route catches that failure instead. To let a person decide, route the failure to a Step that waits for them, as Manual Recovery shows. Execute Failure Recovery shows the automatic route. After you fix the cause of a failed run, Time Travel can run it again from an earlier point.

Place the idempotency keys

Give each external call an idempotency key. Here the order ID is the Flow ID, so charge- plus the order ID works for the Charge call.

Make a keyless API call safe

For an API that takes no key, a repeated call has a second effect. Two approaches make that window small:

• Check first. Send your own reference with the call, such as the order ID in a shipment's reference field. At the start of the Step, look the effect up by that reference. If it already exists, skip the call.
• Record the attempt. If the API has no lookup, write a row keyed like an idempotency key, such as ship-order-1042, to your own database in the same Step, just before the call. On a rerun, a row that is already there means the call may have happened. Stop and check by hand, or go to a recovery Step, instead of calling again.

A crash after the call and before Dex saves the Step still leaves one case where you cannot tell whether the call happened. Each approach turns a silent repeat into a case you can see and handle.

Keep your database writes

You can keep the orders table that support and reporting read. Write to it from a Step with an upsert keyed by the order ID, so a repeat writes the same row. Start the Flow from the handler that creates the order, with the order ID as the Flow ID. Dex and Your Database covers both, including what happens when one of the two writes fails.

Checklist

• Each call that changes another system is in its own Step, or shares a Step only with calls that succeed or fail together.
• Each such call passes an idempotency key built from the Flow ID, not from the time, a random value, or the attempt number.
• Each wait is a WaitFor on a Channel, a Timer, or both, not a loop that polls.
• Each command from outside, such as an approval, is an RPC.
• Status that other code reads is an Attribute, or an upsert into your own table.
• Each Step's retry policy fits the call, not the default.
• Each failure that needs a business answer, such as a refund, has a failure route to a recovery Step.
• Only when two Steps, or a Step and an RPC, can run at the same time and read then write the same Attribute, they declare a lock on it. A sequential Flow needs no lock.
• The handler starts the Flow with a stable Flow ID and a request ID, so a retried request does not start a second Flow. A minimal service shows the handler.

Next

• Run Dex in Production: what to set up before the Flow runs in production.
• Order Processing: this process as runnable code in every SDK.
• Concepts: each building block in detail.
• Change Flow Code: how to change a Flow while some of its runs are still open.