resonatehq/resonate/integration

Building a Resonate Integration

What an integration is

An integration is an implementation of the ResonateWorker port that does not run the work itself. It makes one durable promise stand for one run in an external system:

promise pending  ──►  downstream run created  ──►  run monitored  ──►  promise settled

It lives in the server process, registered under its own URL scheme:

workers.insert("airflow".to_string(), Arc::new(AirflowWorker::new(...)));

To the rest of the application it is indistinguishable from any other remote function: someone creates a promise tagged resonate:target: airflow://prod, blocks on it, and gets back a value.

An integration is not a Resonate SDK application, and not a place for business logic. It maps a request to a run, and a run's state to a promise's state.

Why integrations look alike

There will be a lot of these, and they all work the same way: claim a task, start one run, watch two clocks, settle one promise. Because the mechanism is the same, the code should be recognisable — someone who has read one integration should open the next and find where it starts the run, where it decides, where it settles in the same place, under the same name.

They are not the same integration with the names changed. Addresses, config, params and what "done" means all differ, in whatever way makes sense for the system being wrapped — forcing those to match would make the integration lie about the system.

references/structure.md splits the two: the skeleton that is fixed everywhere, and the parts you decide for your system.

The two in-tree references

Read the Airflow worker before writing a new one. It is the complete worked example — address schema, idempotency key, error classification, poll loop, settlement — compiled and unit-tested in this repo.

One difference to expect: this skill writes protocol calls in their typed form, server.process(Request::…). Follow the skill for those; the surrounding lifecycle, classification and monitoring are what the in-tree worker is there to show.

The ports

Three traits in src/core/, which depends on nothing else in the crate. Verbatim:

/// The inbound port — src/core/server.rs
#[async_trait]
pub trait ResonateServer: Send + Sync {
    async fn process(&self, request: Request) -> Result<Response, Unavailable>;
}
 
/// The outbound port — src/core/worker.rs
#[async_trait]
pub trait ResonateWorker: Send + Sync {
    async fn send(&self, address: &str, msg: &Message) -> Result<(), Unavailable>;
}
 
/// The routing port — src/core/router.rs
#[async_trait]
pub trait ResonateRouter: Send + Sync {
    async fn route(&self, address: &str, msg: &Message) -> Result<(), Unavailable>;
}

Request and Response are the protocol's own unions — one variant per operation, the same ones the canonical types-raw.ts defines. A worker names the operation it wants and matches the answer: no envelope, no kind string, no JSON on this path.

Message is what a server emits toward a worker, and Unavailable is the only error any port returns:

#[derive(Debug, Serialize)]
#[serde(untagged)]                    // each variant carries its own `kind`
pub enum Message {
    Execute(ExecuteMsg),              // { kind, head: { serverUrl }, data: { task: { id, version } } }
    Unblock(UnblockMsg),              // { kind, head: {},            data: { promise } }
}
 
pub struct Unavailable { pub message: String }

The shape an integration implements

One trait, one method. Everything else is a free choice.

/// Everything the worker holds, behind one `Arc`. `server` is the inbound port:
/// how a run claims its task, heartbeats it, and settles its promise. `Server`
/// never holds the router, so this stays a DAG.
struct My { server: Arc<dyn ResonateServer>, client: reqwest::Client, /* config */ }
 
pub struct MyWorker { inner: Arc<My> }
 
/// One delivery in flight: the worker and the task. Nothing else — the
/// address, the param and the deadline all come off the promise once claimed.
struct RunContext { worker: Arc<My>, task: ExecuteMsgTask }
 
#[async_trait]
impl ResonateWorker for MyWorker {
    // `address` is unused: a *proxy* worker needs it (a URL to POST to, a group
    // to fan out to) because it has no promise in hand. This one claims the
    // task, so it reads its address off the promise's `resonate:target` tag.
    async fn send(&self, _address: &str, msg: &Message) -> Result<(), Unavailable> {
        let task = match msg {
            Message::Execute(e) => &e.data.task,   // { id, version }
            Message::Unblock(_) => return Ok(()),  // for workers that *wait*; not this one
        };
 
        // Hand off and do nothing else. The first real step is claiming the
        // task through `server`, and that is a round trip — the dispatch loop
        // awaits `send` sequentially over the batch. Validation waits too:
        // until the task is claimed its failures are unreportable.
        let ctx = RunContext { worker: Arc::clone(&self.inner), task: task.clone() };
        tokio::spawn(async move { ctx.run().await });
        Ok(())                                     // accepted for delivery, not executed
    }
}

The work lives in run, and it starts by claiming the task through the server handle — server.process(Request::TaskAcquire(..)), the same operation a remote worker would put on the wire. That claim is the gate: before it the only way to report a failure is Err(Unavailable), which the dispatch loop logs and drops; after it, every failure can settle the promise. So validation, config lookup and the downstream call all belong on the far side of it. Full stub in references/registration.md.

Registering it is one line, and nothing in core changes:

workers.insert("my".to_string(), Arc::new(MyWorker::new(...)));

Four consequences that shape every integration:

1. `send` means accepted for delivery, not executed. A DAG run takes hours; the

dispatch loop must not block on it. Spawn and return Ok(()).

1. The address arrives unparsed, and the worker owns everything past the scheme.

core::address::is_valid_address only checks that the string is a URI — deliberately, so admission is a pure function of the string and identical on every deployment. Syntax errors past the scheme surface at delivery, not at promise.create.

1. A new scheme is a registration, never a change to `core`. That is the whole cost of

adding an integration.

1. `Unavailable` is the only error. Everything the peer can say — not found, conflict,

forbidden — comes back as an Ok response. Unavailable means the exchange did not complete, and the request may already have been applied. Retries must be idempotent.

An in-process worker holds Arc<dyn ResonateServer> and calls process directly: no HTTP hop, no auth, but the same validation and settlement path a remote worker's calls take.

The fact that determines the design

`execute` is at-least-once and is re-sent until the promise settles. The server re-enqueues it every tasks.retry_timeout (default 30 s) while a task is pending, and again whenever a lease expires. The message carries {task: {id, version}} and nothing else — no delivery counter, no redelivery flag. A worker cannot tell a first delivery from a redelivery, and task.version counts acquisitions, not downstream side effects.

So the integration must not ask "have I already done this?". Build it so asking is unnecessary:

**Fire the request again, every time, under a key derived from the promise id, and let

the downstream system reject the duplicate.**

Airflow's 409 on a duplicate dag_run_id is then the recovery path: an earlier attempt already triggered the run, so re-attach and monitor it. If the downstream system offers no such key, stop and discuss with the user — see references/idempotency.md. There is no safe default.

Scope of this skill

For push, see references/push-vs-poll.md — the questions to settle before building it.

The five things every integration defines

1. Address schema — the scheme it owns and the syntax past it.

references/schemas.md

1. Param schema — the request, in promise.param.data. The protocol carries opaque

bytes, so the encoding is yours to pick and yours to document; validate against it before the first side effect and reject on violation.

1. Value schema — the outcome, in promise.value.data, for both branches.
2. Create — idempotently start or re-attach to the downstream run.
3. Check — map the downstream state to pending / succeeded / failed.

All five go in the integration's README, which is the contract callers write against — references/readme-template.md.

The three phases

Phase 0 is task.acquire; a 409 means another attempt owns the task, so drop the message and never touch the downstream system.

1. Create — runs on every delivery. Issue the create with a deterministic key derived

from promise.id. Treat "already exists" as success and fall through to monitoring. Never a UUID, a clock read, or task.version.

1. Monitor — two independent clocks. The lease clock heartbeats task.heartbeat

at a third of the lease TTL. The downstream clock asks the external system for the run's state on a backing-off interval sized for its cost and latency. They answer to different authorities and run at different frequencies; keep the heartbeat in its own task so the poll interval can back off past the lease TTL without the lease lapsing. Both stop at promise.timeoutAt — the server settles rejected_timedout there regardless.

1. Settle. task.fulfill, which completes task and promise in one transaction and is

what finally stops redelivery.

Every downstream error is permanent (reject the promise), transient (do not settle; let redelivery re-run the idempotent create), or ambiguous — and ambiguous is transient. references/lifecycle.md has the full table.

Workflow for building one