SHIFT HEAVY
WORKLOADS.

00Install & run

loadshift is built with the Kotlin Toolchain. Clone it, build it, run the demo workflow on the in-process backend.

git clone https://github.com/it-atelier-gn/loadshift-kotlin.git
cd loadshift-kotlin
./kotlin test
./kotlin run -m loadshift-demo

== Dry run plan ==
  cleanup
  flag-missing
== Execute on LocalBackend ==
  flag-missing cust-1-b
  cleanup cust-1-a <cust-1@example.com>
done=3 failed=0 deadLetters=3
  DLQ topic=cleanup error=invalid email for cust-1-c

Or pull it into an existing project from Maven Central (io.github.it-atelier-gn, version 0.5.0):

dependencies {
    implementation("io.github.it-atelier-gn:loadshift-core:0.5.0")
    implementation("io.github.it-atelier-gn:loadshift-local:0.5.0")
    implementation("io.github.it-atelier-gn:loadshift-camunda-7:0.5.0")
    implementation("io.github.it-atelier-gn:loadshift-camunda-8:0.5.0")
    implementation("io.github.it-atelier-gn:loadshift-web:0.5.0")
    implementation("io.github.it-atelier-gn:loadshift-sqlite:0.5.0")
}

dependencies {
    implementation 'io.github.it-atelier-gn:loadshift-core:0.5.0'
    implementation 'io.github.it-atelier-gn:loadshift-local:0.5.0'
    implementation 'io.github.it-atelier-gn:loadshift-camunda-7:0.5.0'
    implementation 'io.github.it-atelier-gn:loadshift-camunda-8:0.5.0'
    implementation 'io.github.it-atelier-gn:loadshift-web:0.5.0'
    implementation 'io.github.it-atelier-gn:loadshift-sqlite:0.5.0'
}

<dependency>
    <groupId>io.github.it-atelier-gn</groupId>
    <artifactId>loadshift-core</artifactId>
    <version>0.5.0</version>
</dependency>
<dependency>
    <groupId>io.github.it-atelier-gn</groupId>
    <artifactId>loadshift-local</artifactId>
    <version>0.5.0</version>
</dependency>
<dependency>
    <groupId>io.github.it-atelier-gn</groupId>
    <artifactId>loadshift-camunda-7</artifactId>
    <version>0.5.0</version>
</dependency>
<dependency>
    <groupId>io.github.it-atelier-gn</groupId>
    <artifactId>loadshift-camunda-8</artifactId>
    <version>0.5.0</version>
</dependency>
<dependency>
    <groupId>io.github.it-atelier-gn</groupId>
    <artifactId>loadshift-web</artifactId>
    <version>0.5.0</version>
</dependency>
<dependency>
    <groupId>io.github.it-atelier-gn</groupId>
    <artifactId>loadshift-sqlite</artifactId>
    <version>0.5.0</version>
</dependency>

Pick the modules you need; see Modules for what each one contains.

01Work items

A work item is the unit that travels through a workflow. Implement WorkItem with a @Serializable data class; fields are plain constructor properties. The Camunda backends use kotlinx.serialization to convert them to and from process variables. Override key to give items a stable identity for tracing and dead letters.

@Serializable
data class Customer(var id: String, var attempts: Int = 0) : WorkItem {
    override val key get() = id
}

@Serializable
data class Contact(var id: String, var email: String? = null) : WorkItem {
    override val key get() = id
}

val vip = Customer("cust-1")

@Serializable is required on any work item type used as a workflow input or fanOut child when running on a Camunda backend. The local backend passes objects in memory directly and never calls the codec.

02Workflows & tasks

A workflow is a named sequence of steps over one work-item type. input(...) seeds it with a fixed list and task declares a unit of work under a topic name.

val onboarding = workflow<Customer>("onboarding") {
    input(customers)

    task("validate") { println("validating ${it.id}") }
    task("provision") { provision(it) }
    task("welcome-mail") { sendWelcome(it) }
}

Topic names become BPMN external-task topics on Camunda, and trace entries on the local backend, so keep them short and stable.

03Branching & loops

Control flow is part of the definition, not buried in task bodies, so it compiles to real BPMN gateways. condition / otherwise branches, loop repeats while a guard holds, parallel runs branches concurrently.

condition({ it.email == null }) {
    task("flag-missing") { flag(it) }
} otherwise {
    task("cleanup") { clean(it) }
}

loop({ it.attempts < 3 }) {
    task("poll-status") { it.attempts = it.attempts + 1 }
}

parallel {
    branch { task("index") { index(it) } }
    branch { task("notify") { notify(it) } }
}

wait(15.minutes)
task("settle") { settle(it) }

timeout(10.minutes) {
    task("call-vendor") { callVendor(it) }
}

awaitMessage("payment-confirmed")
task("fulfil") { fulfil(it) }

Runaway loops are cut off by RunConfig(maxLoopIterations = …) and turned into dead letters. wait pauses the flow for a fixed duration — a real timer catch event on Camunda, a suspending delay locally. timeout wraps a block in an interrupting boundary timer (an embedded subprocess on Camunda); overruns become dead letters.

awaitMessage parks an item until an event arrives. The run handle delivers it with send(message, key) (one item) or broadcast(message) (all waiters, local). On Camunda, send correlates through the engine's message API — a catch event keyed by item — and is the verified path on both Camunda 7 and 8.

task("charge") { charge(it) } compensate { refund(it) }
task("reserve") { reserve(it) } compensate { release(it) }
task("ship") { ship(it) }   // if this fails…
// …release(it) then refund(it) run automatically, in reverse

A failing step rolls back the completed steps before it: each compensate block runs in reverse order, so a half-done item is undone instead of left inconsistent. Enforced at runtime on the local backend and worker-side on Camunda 7 and 8.

04Fan-out

Expand one parent item into many children and process them with bounded concurrency. The child block is a full workflow scope: tasks, branching, even nested fan-out. On Camunda each fan-out level becomes its own process, invoked as a call activity.

val cleanup = workflow<Customer>("contact-cleanup") {
    input(customers)

    fanOut(expand = { fetchContacts(it.id) }, concurrency = 4) {
        condition({ it.email == null }) {
            task("flag-missing") { flag(it) }
        } otherwise {
            task("scrub") { scrub(it) }
        }
    }
}

Tasks inside a fan-out can read a parent-derived value through context(). Opt in by passing a context transform; without it context() is null. Pass context = { it } to expose the parent item itself.

fanOut(
    expand = { customer -> fetchOrders(customer.id) },
    context = { customer -> customer.address },
) {
    task("ship") { order ->
        ship(order, context())
    }
}

context() resolves against the level it is called in and exposes the nearest parent. To reach further up, a nested fanOut receives the enclosing context value as a second argument, so you fold the chain into your own value as you descend. Use the built-in Parent(item, parent) carrier to keep both levels without declaring a type, or compose your own.

fanOut(
    expand = { customer -> fetchOrders(customer.id) },
    context = { customer -> customer },
) {
    fanOut(
        expand = { order -> order.lines },
        context = { order, customer -> Parent(order, customer) },
    ) {
        task("ship") { line ->
            val (order, customer) = context()
            ship(line, order, customer)
        }
    }
}

To aggregate children back into the parent, chain a terminal onto fanOut. join() just waits for every child; collect gathers the finished children into a list; reduce folds them through initial/combine into one aggregate (streaming, so it never materializes the children). Each runs its callback on the parent once the fan joins.

fanOut(expand = { customer -> fetchOrders(customer.id) }, concurrency = 8) {
    task("charge") { order -> charge(order) }
}.reduce(0.0, combine = { sum, order -> sum + order.amount }) { customer, total ->
    customer.billed = total
}

reduce and collect fold over the children as produced by expand — identically on the local and Camunda backends. The child body runs per child for its side effects; aggregate the values you need from the expanded items rather than from in-task mutations.

05Streaming sources

For workloads too big for a list, seed from a Flow or from a paginated API. paginate walks a cursor until it returns null; items stream through the workflow without ever materializing the full set.

val allCustomers = paginate { cursor ->
    val page = crm.customers(after = cursor)
    Page(page.items, page.nextCursor)
}

val sync = workflow<Customer>("sync-all") {
    input(allCustomers)
    task("sync") { push(it) }
}

Fan-out can paginate per parent too, with paginateWith:

fanOut(
    paginateWith { customer, cursor -> crm.contacts(customer.id, cursor) },
    concurrency = 8,
) {
    task("scrub") { scrub(it) }
}

06Reliability

Every task takes retry, timeout and rateLimit directly. Retries back off exponentially with jitter; rate limits are enforced per topic.

task(
    "enrich",
    retry = RetryPolicy(maxAttempts = 5, baseDelay = 200.milliseconds),
    timeout = 10.seconds,
    rateLimit = perSecond(50),
) { enrich(it) }

A task's retry policy applies wherever it runs, including inside a parallel branch: a failing branch task is retried in full (e.g. all 3 attempts of the default policy) before the item is handed to the run's ErrorPolicy.

What happens when a task still fails is a run-level decision. ErrorPolicy.DeadLetter records the failure and keeps going, Fail aborts the run, Skip counts and continues.

val result = LocalBackend().run(
    cleanup,
    RunConfig(
        maxConcurrency = 32,
        retry = RetryPolicy(maxAttempts = 3),
        onError = ErrorPolicy.DeadLetter,
    ),
).await()

result.deadLetters.forEach { println("${it.topic}: ${it.error}") }

07Backends

The definition is backend-agnostic. LocalBackend interprets it in-process, ideal for tests and development. The Camunda backends compile the same definition to BPMN, deploy it, seed instances, and work the external-task topics with your Kotlin task bodies.

LocalBackend().dryRun(cleanup).forEach(::println)

LocalBackend().run(cleanup).await()

Camunda7Backend("http://localhost:8080/engine-rest").run(cleanup).await()

Camunda8Backend("http://localhost:8080", token = zeebeToken).run(cleanup).await()

dryRun walks the plan without executing task bodies and returns the topics in visit order: cheap workflow shape tests.

08Run control

run() returns a RunHandle: live progress counters, pause, cancel, and an awaitable result. RunConfig.start picks one of Start.Now, Start.Manual, Start.At, or Start.Cron.

val handle = LocalBackend().run(sync, RunConfig(start = Start.Manual))

handle.start()
println(handle.progress())

handle.pause()
handle.cancel()

val result = handle.await()
println("done=${result.done} failed=${result.failed} skipped=${result.skipped}")

Start.Manual is uniform across backends: the run is registered (process definitions deployed on Camunda, worker pool started) and then waits for handle.start() before seeding via input(). If the process that called run() isn't the one that should trigger it, loadshift-web's ControlServer exposes the same controls over HTTP: POST /api/runs/{id}/start, /pause, and /cancel, backed by Control.start/pause/cancel on the tracked RunHandle.

LocalBackend().run(sync, RunConfig(start = Start.At(Clock.System.now() + 5.minutes)))

LocalBackend().run(sync, RunConfig(start = Start.Cron("0 9 * * *")))

Start.At waits until the given instant, then seeds and runs once. Start.Cron takes a standard 5-field cron expression (minute hour day-of-month month day-of-week) and repeats indefinitely: on every tick it calls input() again and runs that batch to completion before waiting for the next tick. Because each tick re-reads input() (including paginated sources), this works the same whether a tick seeds one item or millions.

09Logging

Task bodies can call log() to record structured entries that carry the execution tree context: run id, workflow name, the ancestor item-key path through nested fanOuts, the current item key, and the current task's topic.

task("charge") { order ->
    log("charging customer", "orderId" to order.id, "amount" to order.amount)
}

Logging is opt-in via RunConfig.logSink (default NoopLogSink, which discards everything). loadshift-sqlite provides SqliteLogSink, which persists entries to a SQLite database.

val sink = SqliteLogSink("logs.db")
backend.run(workflow, RunConfig(logSink = sink)).await()

10Modules