Surefold Core

Choose the part you need

Surefold Core is published today as Lark, under the group and artifact prefix io.github.matthewjones372:lark*. The code on this page uses the real artifact and package names (io.github.matthewjones372.lark), because that is what you will type.

The artifact named lark is the foundation. The wider Lark project supplies modules for the rest of a service’s life: how it starts, how it processes work, how it stores state and how it is tested. You do not need actors to use streams, or an application graph to use parZip.

How the pieces fit. An application graph can acquire a pool and start a worker; the worker can run a stream; a stream stage can make a blocking call or ask an actor. Each boundary still needs an owner and a shutdown path. The graph owns service resources, scopes own forks, and a stream backend runs the stream description. A typed error describes an expected failure; it does not turn every exception into a declared error.

Start with one dependency and a complete program

Virtual threads can park during supported blocking operations and release their carrier OS thread. Native calls and some filesystem operations can still occupy a carrier. A service can block on JDBC, a file or an HTTP client with no async API and still serve many requests at once. Writing that code as plain functions avoids suspend and a second calling convention, and stack traces, debuggers and profilers see ordinary threads.

What virtual threads do not give you is structure. Thread.ofVirtual().start does not know which request started it, does not stop when a sibling fails, and does not provide Lark’s scoped application context. Surefold Core adds four things:

• Structured concurrency. Forks belong to a scope. A scope returns only after every fork it started has ended, and the first failure interrupts the rest.
• Typed errors across forks. A branch calls fail(error) in Lark’s Fail<E> scope, and attempt { } returns Lark’s Either<E, A>. The combinators carry that failure across their own forks.
• Resources and policies. Acquire and release in reverse order on every exit path, and retry, time out, break or limit calls with composable values.
• Context that crosses forks. Log annotations, metric tags, trace context and the clock are inherited by every fork, so a correlation id or a test clock reaches the branches of a parMap.

Typed errors use Lark’s own Fail, attempt, fail and Either from io.github.matthewjones372.lark.fail. The separate lark-arrow module supplies interop for applications that already use Arrow. Core functions block on ordinary threads; coroutine callers need the integration boundary.

When to use it, and when not

Reach for it when:

• you write a JVM service in Kotlin and would rather block than write suspend code;
• your dependencies are blocking (JDBC, a blocking HTTP client, a file) and you want to call several at once with a deadline;
• you want expected failures represented as typed values, including across a fan-out;
• you want the wiring of a service checked before it runs, and tests that move a clock instead of sleeping.

Think twice when:

• you cannot run JDK 25. A service on JDK 21 cannot use it.
• you need a stable API. Releases are 0.x, any release can change or remove API, and there is no binary-compatibility check. No production use is known.
• your code is built on coroutines. Core functions block. Keep them off a dispatcher’s scarce worker threads; check the integration boundaries, including the future bridge available on current source.

Install

This guide uses Lark’s current declared-error API at 35ba39bc. Use JDK 25 and Kotlin 2.4.10. The verified Central release, 0.10.0, predates this API, so build the matching source locally before running these examples.

git clone https://github.com/matthewjones372/lark.git
cd lark
git checkout 35ba39bcccf0e6f69577f7bf035e09ae4c76add2
./gradlew :lark:publishToMavenLocal -Prelease.forceVersion=0.0.0-site

plugins {
    kotlin("jvm") version "2.4.10"
    application
}

repositories {
    mavenLocal()
    mavenCentral()
}

kotlin { jvmToolchain(25) }

dependencies {
    implementation("io.github.matthewjones372:lark:0.0.0-site-SNAPSHOT")
}

application { mainClass.set("MainKt") }

Your first program

Two blocking lookups run at the same time on virtual threads, inside a time limit, with a typed error. With only the lark dependency from above, this file compiles and runs as it is.

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.fail.fail
import io.github.matthewjones372.lark.fail.attempt
import io.github.matthewjones372.lark.parZip
import io.github.matthewjones372.lark.timeout
import kotlin.time.Duration.Companion.seconds

data class User(val id: Int, val name: String)

data class Order(val id: Int, val pence: Long)

data class Summary(val user: User, val orders: List<Order>)

data class UserNotFound(val id: Int)

// Stand-ins for a database and an HTTP client: each one blocks its thread.
fun findUser(id: Int): User? {
    Thread.sleep(100)
    return if (id == 1) User(1, "Ada") else null
}

fun recentOrders(id: Int): List<Order> {
    Thread.sleep(100)
    return listOf(Order(7, 1200))
}

fun summary(id: Int): Either<UserNotFound, Summary> = attempt {
    timeout(2.seconds) {
        parZip(
            { findUser(id) ?: fail(UserNotFound(id)) },
            { recentOrders(id) },
        ) { user, orders -> Summary(user, orders) }
    }
}

fun main() {
    println(summary(1)) // Right(value=Summary(user=User(id=1, name=Ada), orders=[Order(id=7, pence=1200)]))
    println(summary(2)) // Left(value=UserNotFound(id=2))
}

Save this as src/main/kotlin/Main.kt beside the build file above, then run ./gradlew run (or run main in your IDE). It prints:

Right(value=Summary(user=User(id=1, name=Ada), orders=[Order(id=7, pence=1200)]))
Left(value=UserNotFound(id=2))

The two lookups run at the same time, so summary(1) takes about 100 ms rather than 200.

What happens, case by case

1. Both branches succeed

   parZip forks each lambda onto its own virtual thread, waits for both, and passes the results to the final lambda. The answer is Right(Summary(…)).

2. A branch declares a failure

   When the user is missing, the first branch calls fail(UserNotFound(id)). Its sibling, the orders call, is interrupted and joined, and summary returns Left(UserNotFound(id)). The caller receives the declared error as a value.

3. A branch throws

   If either call throws, the other is interrupted and the exception propagates out of summary. A thrown exception is not a declared error, so it does not become a Left.

4. Time runs out

   If two seconds pass, both branches are interrupted and timeout throws java.util.concurrent.TimeoutException once the block has ended. This can take longer than two seconds when work ignores interruption. Use timeoutOrNull for null instead.

In every case summary returns only after both forks have ended, so no work is left running behind it.

Core concepts

Core operations are in io.github.matthewjones372.lark; declared-error types and helpers are in its fail package. Both are in the lark artifact.

Scopes and structured concurrency

A scope does not return until every fork it started has ended. When parZip, flock or timeout returns, normally or by declaring a failure or throwing, none of its work is still running. Four shapes cover most needs:

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.fail.bind
import io.github.matthewjones372.lark.fail.attempt
import io.github.matthewjones372.lark.flock
import io.github.matthewjones372.lark.parMap
import io.github.matthewjones372.lark.parZip
import io.github.matthewjones372.lark.raceN

data class Err(val why: String)

data class User(val id: Int)

data class Order(val id: Int)

data class Dashboard(val user: User, val orders: List<Order>)

// Blocking stand-ins for a repository and a client.
fun findUser(id: Int): Either<Err, User> = Either.Right(User(id))

fun ordersFor(id: Int): Either<Err, List<Order>> = Either.Right(listOf(Order(1)))

fun quoteFrom(source: String): Long = source.length.toLong()

// Two branches at once; up to nine are supported.
fun dashboard(id: Int): Either<Err, Dashboard> = attempt {
    parZip({ findUser(id).bind() }, { ordersFor(id).bind() }) { user, orders -> Dashboard(user, orders) }
}

// One fork per element, answered in the input's order.
fun users(ids: List<Int>): Either<Err, List<User>> = attempt { parMap(ids) { findUser(it).bind() } }

// The first branch to answer wins, and the loser is interrupted. Left is the first branch, Right the second.
fun quote(): Either<Long, Long> = raceN({ quoteFrom("primary") }, { quoteFrom("fallback") })

// async and await, for forks the combinators do not shape.
fun report(id: Int): Either<Err, Dashboard> = flock {
    val user = async { findUser(id).bind() }
    val orders = async { ordersFor(id).bind() }
    Dashboard(user.await(), orders.await())
}

• parZip takes two to nine branches. raceN takes two or three; with three it answers Either<A, Either<B, C>>.
• Outside a Fail, parZip, parMap and raceN take plain () -> A branches and return the combined value.
• flock { } is the scope async forks in. On its own it returns Either<E, A>; inside an existing Fail<E> it returns A and a failure propagates to the enclosing scope.
• A Deferred nobody awaits is still joined when its flock closes. If it failed, the declared error becomes the block's Left; if it threw, the exception is rethrown.
• async(start = Start.Lazy) { } does not start until something awaits it. A lazy fork that is never awaited never runs, so it cannot fail the scope.

By default each fork is a new virtual thread. Every forking combinator also has an overload that takes an executor first, such as parZip(on = pool, { … }, { … }) { … }, and flock(on = pool) { } and async(on = pool) { } take it as a defaulted parameter.

Typed errors across forks

Inside attempt { } (or any Fail<E>), a branch can bind() an Either or fail an error, and the first declared failure becomes the block's Left. If several branches fail on their own, the one started first wins; an exception thrown only because a branch was interrupted is not reported.

For validation you usually want every error, not the first. parZipOrAccumulate and parMapOrAccumulate run every branch to completion and return all the declared failures, in branch order:

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.fail.mapLeft
import io.github.matthewjones372.lark.fail.bind
import io.github.matthewjones372.lark.fail.NonEmptyList
import io.github.matthewjones372.lark.fail.attempt
import io.github.matthewjones372.lark.fail.ensure
import io.github.matthewjones372.lark.parMapOrAccumulate
import io.github.matthewjones372.lark.parZipOrAccumulate

data class Form(val name: String, val email: String)

data class Account(val name: String, val email: String)

data class Invalid(val field: String)

fun validate(form: Form): Either<NonEmptyList<Invalid>, Account> = attempt {
    parZipOrAccumulate(
        { form.name.also { ensure(it.isNotBlank()) { Invalid("name") } } },
        { form.email.also { ensure('@' in it) { Invalid("email") } } },
    ) { name, email -> Account(name, email) }
}

fun validateAll(forms: List<Form>): Either<NonEmptyList<Invalid>, List<Account>> = attempt {
    parMapOrAccumulate(forms) { form -> validate(form).mapLeft { it.first() }.bind() }
}

A thrown exception is never accumulated: it interrupts the other branches, as in parZip, and the same instance is rethrown. Each function also has a form taking combine: (E, E) -> E first, for a single error type that knows how to merge two.

Cancellation and interruption

The timeline illustrates interruptible I/O. Choose “Ignores interruption” to see why a deadline is not a bound on how long the scope can take to return.

Core requests cancellation by interrupting the fork. An operation that responds to interruption, such as Thread.sleep, Future.get or an interruptible lock acquisition, can then stop. A socket or JDBC call’s behaviour depends on the API and driver; configure their own timeouts too. CPU-bound work must check interruption explicitly if it needs to stop early.

deferred.cancel() interrupts one fork and returns once it has ended. A cancelled fork's outcome counts as handled, so the scope does not answer with it. A fork clears the interrupt flag as its body exits, so a pooled executor's next task does not start interrupted.

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.flock

data class Credit(val hasCredit: Boolean)

fun buildReport(id: Int): String {
    Thread.sleep(500) // interruptible, so cancel() stops it here
    return "report for $id"
}

fun creditFor(id: Int): Credit = Credit(hasCredit = id % 2 == 0)

// Start the expensive work early, and stop it if the cheap check says it is not wanted.
fun page(id: Int): Either<Nothing, String> = flock {
    val report = async { buildReport(id) }
    if (creditFor(id).hasCredit) {
        report.await()
    } else {
        report.cancel()
        "no report"
    }
}

Resources

resourceScope { } releases what was acquired inside it in reverse order when the block exits, whether by returning, declaring a failure, throwing or being interrupted. install takes an acquire and a release, and the release is told how the scope ended. A Resource<A> is a reusable value you bind() inside a scope, or use for one block.

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.fail.fail
import io.github.matthewjones372.lark.fail.bind
import io.github.matthewjones372.lark.fail.attempt
import io.github.matthewjones372.lark.ExitCase
import io.github.matthewjones372.lark.Resource
import io.github.matthewjones372.lark.resource
import io.github.matthewjones372.lark.resourceScope
import io.github.matthewjones372.lark.use

class Connection : AutoCloseable {
    fun rows(id: Int): List<String> = listOf("row-$id")

    fun commit() = Unit

    fun rollback() = Unit

    override fun close() = Unit
}

data class NotFound(val id: Int)

val connection: Resource<Connection> = resource {
    install({ Connection() }) { held, exit ->
        if (exit is ExitCase.Failure) held.rollback() else held.commit()
        held.close()
    }
}

fun report(id: Int): Either<NotFound, String> = attempt {
    resourceScope {
        val db = connection.bind()                                  // released when this scope ends
        val out = install({ StringBuilder() }) { _, _ -> }
        val rows = db.rows(id).ifEmpty { fail(NotFound(id)) }      // a declared failure releases both, in reverse
        out.append(rows.joinToString()).toString()
    }
}

// A single resource for the length of one block.
fun rows(id: Int): List<String> = connection use { it.rows(id) }

A declared failure releases with Completed, because it is a declared answer. Every release runs even if an earlier one throws. A release that throws when nothing else failed becomes the exception the caller sees; one that throws after another failure is added to it as suppressed.

Timeouts

A deadline requests cancellation; it does not forcibly terminate an underlying operation. The scope waits for that operation’s fork to end before returning.

timeout(d) { } runs its block on a fork and races it against a sleeper. A block that does not finish in time is interrupted, and timeout throws TimeoutException once the block has ended. timeoutOrNull answers null instead.

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.fail.bind
import io.github.matthewjones372.lark.fail.attempt
import io.github.matthewjones372.lark.timeout
import io.github.matthewjones372.lark.timeoutOrNull
import kotlin.time.Duration.Companion.seconds

data class Unavailable(val id: Int)

fun findRow(id: Int): Either<Unavailable, String> = Either.Right("row-$id")

fun quote(id: Int): Either<Unavailable, String> = attempt {
    timeout(2.seconds) { findRow(id).bind() }                       // throws TimeoutException
}

fun maybeQuote(id: Int): Either<Unavailable, String?> = attempt {
    timeoutOrNull(2.seconds) { findRow(id).bind() }                 // null instead
}

The sleeper waits on the inherited clock. That matters in tests: see the gotcha about fixedClock() in Gotchas and limits.

Schedules and retries

A Schedule<Input, Output> is a value that answers each input with a decision: continue after a delay, or stop. retry feeds it what the action threw, repeat what the action returned, and Fail.retry(schedule) { … } retries a declared error. Delays are spent on the calling thread through the inherited clock, so an interrupt ends the schedule where it is and a test can skip the waiting.

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.fail.bind
import io.github.matthewjones372.lark.Schedule
import io.github.matthewjones372.lark.retry
import io.github.matthewjones372.lark.fail.attempt
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds

data class Unavailable(val id: Int)

fun fetchRow(id: Int): String = "row-$id"

fun findRow(id: Int): Either<Unavailable, String> = Either.Right("row-$id")

fun pending(): Long = 0

// Exponential backoff with jitter, at most five retries.
val backoff = Schedule.exponential<Throwable>(100.milliseconds).jittered() and Schedule.recurs(5)

fun fetch(id: Int): String = backoff.retry { fetchRow(id) }          // rethrows the last failure

// Bounded by elapsed time rather than by attempts.
val patiently = Schedule.spaced<Throwable>(20.milliseconds) zipLeft Schedule.upTo(5.seconds)

fun drained(): Unit = patiently.retry { check(pending() == 0L) }

// Retrying a declared error: the last declared failure is the Left.
fun row(id: Int): Either<Unavailable, String> = attempt { retry(Schedule.recurs<Unavailable>(5)) { findRow(id).bind() } }

Builders: recurs, spaced, upTo, exponential, linear, fibonacci, forever, identity, doWhile, doUntil and collect. Combinators: and, andThen, zipLeft and zipRight (all infix), or, map, fold, delayed and jittered.

Policies: retries, breaker, rate limit, bulkhead

A policy combines a deadline, retries, guards and a per-attempt timeout into one value. It is checked when it is built and applied as a whole on every call.

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.fail.attempt
import io.github.matthewjones372.lark.Bulkhead
import io.github.matthewjones372.lark.CircuitBreaker
import io.github.matthewjones372.lark.RateLimiter
import io.github.matthewjones372.lark.Schedule
import io.github.matthewjones372.lark.guarded
import io.github.matthewjones372.lark.policy
import kotlin.time.Duration.Companion.milliseconds
import kotlin.time.Duration.Companion.seconds

class Partner(private val call: (String) -> String) {
    private val breaker = CircuitBreaker(
        name = "partner-api",
        maxFailures = 5,
        resetAfter = Schedule.exponential<Unit>(1.seconds).jittered(0.8, 1.2) zipLeft Schedule.recurs(6),
    )
    private val limiter = RateLimiter(name = "partner-api", rate = 50, per = 1.seconds, burst = 10, maxWait = 200.milliseconds)
    private val bulkhead = Bulkhead(name = "partner-api", maxConcurrent = 20, maxWait = 50.milliseconds)

    private val partner = policy("partner-api") {
        deadline(2.seconds)
        retry(Schedule.exponential<Throwable>(100.milliseconds).jittered() zipLeft Schedule.recurs(3))
        guard(breaker)
        guard(limiter, cost = 1)
        guard(bulkhead)
        attemptTimeout(500.milliseconds)
    }

    fun quote(id: String): String = partner { call(id) }

    // A refusal by any guard, as a declared error rather than an exception.
    fun quoted(id: String): Either<String, String> = attempt {
        guarded(partner, ifRejected = { "${it.guard} refused: ${it.message}" }) { call(id) }
    }
}

• deadline(total) is one time budget for the whole call, retries included. Retrying stops before a delay the budget cannot cover.
• retry(schedule) retries what the call throws. A fail is the call's own answer and is never retried.
• A CircuitBreaker opens after maxFailures failures in a row and refuses calls with Rejected.CircuitOpen until the next delay of resetAfter has passed; then one trial call goes through. A declared failure never counts as a failure.
• A RateLimiter is a token bucket refilled at rate every per, holding at most burst. A call that would wait longer than maxWait is refused with Rejected.RateLimited.
• A Bulkhead lets at most maxConcurrent callers in at once; the rest wait up to maxWait and are then refused with Rejected.BulkheadFull.
• attemptTimeout(each) limits each attempt.

Context that crosses forks

A plain virtual thread inherits nothing from the thread that started it. A LarkLocal does cross: it is read on the thread that forks and bound again inside the fork, whichever executor the fork runs on.

import io.github.matthewjones372.lark.larkLocal
import io.github.matthewjones372.lark.parMap

val requestId = larkLocal { "none" }

fun tagged(rows: List<String>): List<String> =
    requestId.locally("abc-123") {
        parMap(rows) { row -> "${requestId.get()}: $row" }   // every branch reads abc-123
    }

The library's own context uses the same mechanism:

• The clock. clock is a LarkLocal<Clock>. Schedule, timeout, probes, actor timers and stream operators all wait on it, so binding fixedClock() or a TestClock changes how they wait.
• Log annotations and metric tags. logAnnotated and metricTagged bind values that every log line and measurement inside the block carries, including on forks.
• Trace context. With lark-otel on the classpath, OpenTelemetry's Context is kept in a LarkLocal.

The same context also travels with an actor message, local or remote, and is stored with journal events.

Building a service

lark-app describes an application as a value: a dependency graph you can check, draw, subset, override and start. There are no annotations, no reflection-driven container and no effect type. A node is a function, and a module is an ordinary Kotlin expression. Everything here is in io.github.matthewjones372.lark.app.

dependencies {
    // application graph and lifecycle, on the same Core source version
    implementation("io.github.matthewjones372:lark-app:0.0.0-site-SNAPSHOT")
}

Nodes, and what they need

• A node is keyed by the type it provides, and is built once however many nodes depend on it.
• Its parameters are its dependencies. A node can take up to nine; past that, group them into a node of their own.
• singleOf(::Constructor) reads the key and the dependencies off a constructor reference. single { … } takes a lambda for anything else.
• boundTo<Interface>() keys a node as the type its dependents ask for. Without it, singleOf(::PgUserRepo) is keyed PgUserRepo.
• closingOf builds an AutoCloseable and closes it on release. singleOf(::Server, Server::stop) gives a release with another name, and install(acquire) { held, exit -> … } inside single gives one that sees how the application ended.
• plus combines modules. Order means nothing, because matching is by type, except where two modules provide the same key: then the right-hand node wins.
• refuse(reason) inside a node ends the start with that reason rather than a stack trace.

val persistence: Module =
    closingOf { cfg: DbConfig -> HikariDataSource(cfg) }.boundTo<DataSource>() +
        singleOf(::PgUserRepo).boundTo<UserRepo>() +
        single<OrderRepo, DataSource> { ds -> PgOrderRepo(ds) }

val domain: Module =
    singleOf(::RegisterUser) +
        single { users: UserRepo, orders: OrderRepo -> PlaceOrder(users, orders) }

Start order, readiness and shutdown

• Start order. The graph starts one topological layer at a time. Nodes in the same layer start in parallel on forks, so the first refusal or failure interrupts its siblings, and everything already acquired is released before runApp returns.
• Readiness. .probe(name, timeout, attempts, interval) { … } makes "started" mean "ready": dependents wait for the probe, not just the constructor. Each attempt runs inside its own timeout, and a probe that throws counts as failed. HealthRegistry, a node the running application provides, answers readiness() (Up, Degraded or Down) and liveness() later.
• Shutdown. runApp installs a JVM shutdown hook that requests shutdown and waits up to 30 seconds for the releases. Releases run in reverse topological order, not reverse acquisition order.
• Exit code. runApp returns an ExitCode and prints what went wrong to stderr; it never ends the process itself. Call exitProcess with it, since a stray non-daemon thread would otherwise keep the JVM alive.

A worked example

A small service: configuration from the environment, a pool that is closed on release and probed for readiness, a repository bound to its interface, a use case, and a server stopped on shutdown.

1. Write the classes as plain Kotlin

   Nothing here knows about the graph. Dependencies are constructor parameters.

   data class DbConfig(val url: String)
   
   data class HttpConfig(val port: Int)
   
   interface Pool { fun healthy(): Boolean }
   
   class Hikari(val config: DbConfig) : Pool, AutoCloseable {
       override fun healthy() = true
       override fun close() = println("pool closed")
   }
   
   interface UserRepo { fun find(id: Long): String? }
   
   class PgUserRepo(private val pool: Pool) : UserRepo {
       override fun find(id: Long) = "user-$id"
   }
   
   class RegisterUser(private val users: UserRepo) {
       fun register(id: Long): String = users.find(id) ?: "unknown"
   }
   
   class HttpServer(private val config: HttpConfig, private val register: RegisterUser) {
       fun start() = println("listening on ${config.port}")
       fun stop() = println("server stopped")
   }

2. Describe the graph

   Sys is the process environment as a node, so a test can swap in FakeSys(mapOf("DB_URL" to "jdbc:…")) instead of setting a JVM-wide variable.

   import io.github.matthewjones372.lark.app.AppScope
   import io.github.matthewjones372.lark.app.LarkApp
   import io.github.matthewjones372.lark.app.Module
   import io.github.matthewjones372.lark.app.RealSys
   import io.github.matthewjones372.lark.app.Sys
   import io.github.matthewjones372.lark.app.boundTo
   import io.github.matthewjones372.lark.app.closingOf
   import io.github.matthewjones372.lark.app.probe
   import io.github.matthewjones372.lark.app.runApp
   import io.github.matthewjones372.lark.app.single
   import io.github.matthewjones372.lark.app.singleOf
   import kotlin.system.exitProcess
   import kotlin.time.Duration.Companion.milliseconds
   import kotlin.time.Duration.Companion.seconds
   
   val config: Module =
       single<Sys> { RealSys } +
           single { sys: Sys -> DbConfig(sys.env("DB_URL") ?: refuse("DB_URL is not set")) } +
           single { sys: Sys -> HttpConfig(sys.env("PORT")?.toIntOrNull() ?: refuse("PORT is not a number")) }
   
   val persistence: Module =
       closingOf(::Hikari).boundTo<Pool>()
           .probe("db", timeout = 2.seconds, attempts = 10, interval = 500.milliseconds) { pool: Pool -> pool.healthy() } +
           singleOf(::PgUserRepo).boundTo<UserRepo>()
   
   val domain: Module = singleOf(::RegisterUser)
   
   val web: Module = singleOf(::HttpServer, HttpServer::stop)
   
   val app: Module = config + persistence + domain + web

3. Declare the application and its root

   The root is a type argument of LarkApp so that the build can read it from the class file without running main.

   object Shop : LarkApp<HttpServer>() {
       override val module: Module = app
   
       override fun AppScope.run(root: HttpServer) {
           root.start()
           awaitShutdown()   // parks until SIGTERM or Ctrl-C, then the graph is released
       }
   }
   
   fun main(): Unit = exitProcess(runApp(Shop).code)

4. Run it

   With DB_URL set and PORT=8080, the config nodes start first, then the pool (which waits for its probe), then the repository, the use case and the server, and run prints listening on 8080. On Ctrl-C the server is stopped and the pool closed, in that order. With DB_URL unset, nothing past the config is built: stderr shows lark-app: DbConfig refused to start: DB_URL is not set and the exit code is 1.

For a test or a one-file program, runApp(module) { root: HttpServer -> … } does the same without a LarkApp. Testing shows how to start part of this graph with a fake.

Checks before anything runs

app.validate() runs no node. It answers Either<NonEmptyList<WiringError>, Plan> with every missing key at once, then checks for cycles. findings(root) also reports keys provided twice and nodes the root does not reach, and render() draws the graph as Mermaid in a stable order, so the drawing can be kept in a golden file.

The graph it produces

For the worked service above, print the graph without constructing a pool or starting a server:

import io.github.matthewjones372.lark.app.render

println(app.render())

Arrows point from a dependency to the node that needs it. Sys supplies DbConfig and HttpConfig. The database branch continues through Pool, UserRepo and RegisterUser. HttpServer needs both HttpConfig and RegisterUser, so it starts only when both are ready. Independent nodes in a layer can start together; shutdown follows the reverse dependency order. Labels name the provided types, so Hikari appears as Pool and PgUserRepo as UserRepo.

Download Mermaid source · Open the SVG

graph TD
    DbConfig[DbConfig]
    HttpConfig[HttpConfig]
    HttpServer[HttpServer]
    Pool[Pool]
    RegisterUser[RegisterUser]
    UserRepo[UserRepo]
    Sys[Sys]
    Sys --> DbConfig
    Sys --> HttpConfig
    HttpConfig --> HttpServer
    RegisterUser --> HttpServer
    DbConfig --> Pool
    UserRepo --> RegisterUser
    Pool --> UserRepo

lark-app wiring

❯ error: missing DataSource
❯     for OrderRepo          Wiring.kt:42

❯ warning: Tracer provided twice
❯     Telemetry.kt:14        shadowed
❯     Local.kt:9             wins

❯ warning: nothing reaches KafkaProducer    Kafka.kt:9

The Gradle plugin runs the same check on every build. Released versions use Maven Central rather than the Gradle Plugin Portal. Publish :lark-app-gradle:publishToMavenLocal with the same local version option, then include Maven Local first:

pluginManagement {
    repositories {
        mavenLocal() // current source built locally, as described above
        mavenCentral()
        gradlePluginPortal()
    }
}

plugins {
    kotlin("jvm") version "2.4.10"
    id("io.github.matthewjones372.lark.wiring") version "0.0.0-site-SNAPSHOT"
}

repositories {
    mavenLocal()
    mavenCentral()
}

larkWiring {
    failOn = "WARN"                                  // "FAIL" (the default) or "WARN"
    diagrams = layout.buildDirectory.dir("wiring")   // default build/reports/lark
}

The larkWiring task runs after classes and is a dependency of check, so it runs whenever the IDE builds the project. It finds every LarkApp in the project, runs the expression that assembles its graph (but none of the nodes), checks it against the declared root, and writes a .mmd diagram for each. Faults are printed in the compiler's format, so the IDE links them to the node's source line.

Applying the plugin also applies lark-app-compiler, a K2 checker that reports the same faults in the editor as you type. It reads the source rather than running it, so for shapes it does not know it reports the warning LARK_APP_UNREAD and leaves that application to the Gradle task. It also warns about Thread.sleep in tests (LARK_SLEEP_IN_TEST), hand-rolled schedulers (LARK_SCHEDULER_BY_HAND), a secret read as a string (LARK_SECRET_AS_STRING), and a run that never calls awaitShutdown() (LARK_RUN_NEVER_RETURNS_ON_SHUTDOWN).

Streams, actors, persistence, clusters and Kafka

Use the same current-source build and local version; publish the modules named below before adding them.

Each of these areas is a separate set of modules built on the core. The core, lark-app and lark-stream are the most established; the actor, cluster and Kafka modules are newer.

Streams: a long-running workload with a shape

Use parMap for a finite collection whose work can all be started. Use a stream when input keeps arriving, parallelism must be bounded, or processing needs buffering, batching, merging and a named sink. A stream can describe the same domain pipeline while the application chooses its backend at the edge.

lark-stream provides Stream<E, A>, a pipeline whose declared failure is part of its type. Elements can never be null. A stream is a description: building one runs nothing and names no backend. The backend is a value passed in where the run starts, and every run answers an Exit:

• Exit.Done(value): the run finished;
• Exit.Failed(error): the pipeline ended with an E it declared;
• Exit.Died(cause): something nobody declared was thrown. A Died is also logged at error level, naming the operator, the element and the line that built it.

dependencies {
    // a pull loop on a virtual thread: brings lark-stream and nothing else
    implementation("io.github.matthewjones372:lark-stream-forks:0.0.0-site-SNAPSHOT")
    // time a test owns
    testImplementation("io.github.matthewjones372:lark-stream-test:0.0.0-site-SNAPSHOT")
}

import io.github.matthewjones372.lark.stream.Exit
import io.github.matthewjones372.lark.stream.Forks
import io.github.matthewjones372.lark.stream.Run
import io.github.matthewjones372.lark.stream.Stream
import io.github.matthewjones372.lark.stream.filter
import io.github.matthewjones372.lark.stream.from
import io.github.matthewjones372.lark.stream.mapOrFail
import io.github.matthewjones372.lark.stream.run
import io.github.matthewjones372.lark.stream.runFold

data class Order(val id: Int, val pence: Long?)

data class Unpriced(val id: Int)

val orders = listOf(Order(1, 1200), Order(2, 800), Order(3, 4500))

// What happens to an order, and the failure it can end with. No backend is named here.
val takings: Run<Unpriced, Long> =
    Stream.from(orders)
        .mapOrFail { order -> order.pence ?: fail(Unpriced(order.id)) }
        .filter { pence -> pence >= 1000 }
        .runFold(0L) { total, pence -> total + pence }

fun main() {
    val exit: Exit<Unpriced, Long> = takings.run(Forks()).toCompletableFuture().join()
    println(exit) // Done(value=5700)
}

run returns a CompletionStage of the Exit. start returns a Running handle whose exit is the same stage, whose stop() ends the run now as Done with what the sink has so far, and whose close() stops it and waits, so Running::close works as the release of an application node.

For a body that must block (JDBC, a blocking client), use mapPar, or mapParOrFail on a stream that has not declared a failure yet. Each element in flight gets its own virtual thread, at most parallelism at a time, and the body runs in the stream's Fail<E>:

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.fail.bind
import io.github.matthewjones372.lark.stream.Stream
import io.github.matthewjones372.lark.stream.from
import io.github.matthewjones372.lark.stream.mapParOrFail

data class Row(val id: Int, val customer: String?)

data class Customer(val id: Int, val name: String)

data class NoCustomer(val id: Int)

class Directory {
    fun lookup(id: Int): Either<NoCustomer, Customer> = Either.Right(Customer(id, "ada"))
}

val directory = Directory()
val rows = listOf(Row(1, "ada"), Row(2, "grace"))

val customers: Stream<NoCustomer, Customer> =
    Stream.from(rows).mapParOrFail(4) { row -> directory.lookup(row.id).bind() }   // a virtual thread per element in flight

Key guarantees and choices:

• Backends. Forks() pulls on one virtual thread and is the usual choice; PekkoStreams(system) compiles each operator to the Pekko stage it names; Actors(flock) runs on lark-actor; TestStreams(clock) waits only on a test's clock. All of them pass one shared parity suite.
• No silent drops. There is no supervision strategy and no resume. An element leaves a pipeline through a named sink, a filter, or not at all. Recover with catchAll, orElse or mapError, and retry a defect with restartOnDefect(schedule).
• Absence is a failure. Stream.single refuses a nullable at compile time, Stream.fromStage(stage, ifNull = …) names a null completion, and orFailIfEmpty turns zero elements into a failure.
• Pekko-only operators (anything taking a Pekko Source, Flow or Sink) run only on PekkoStreams; on another backend the run answers Died before anything is pulled.

Actors: one owner for changing state

An actor is useful when several callers must update the same state without sharing its internals: a shop catalogue, an account or a workflow. Its mailbox serialises messages. This is different from a stream, which describes how elements move through a pipeline; an actor’s behaviour decides what its next state will be after each message.

lark-actor is an actor as a value: a state, and a step that takes one message at a time. Actors run on virtual threads inside a flock, the same scope the core forks in, so there is no actor system to start and stop, and a step may block. When the flock's block ends, every actor in it is stopped, and the flock returns only once they have.

import io.github.matthewjones372.lark.actor.AskFailure
import io.github.matthewjones372.lark.actor.Reply
import io.github.matthewjones372.lark.actor.ask
import io.github.matthewjones372.lark.actor.become
import io.github.matthewjones372.lark.actor.behaviour
import io.github.matthewjones372.lark.actor.spawn
import io.github.matthewjones372.lark.actor.stay
import io.github.matthewjones372.lark.actor.stop
import io.github.matthewjones372.lark.actor.unhandled
import io.github.matthewjones372.lark.flock
import kotlin.time.Duration.Companion.seconds

sealed interface Till

data class Ring(val pence: Int) : Till

data class Total(val reply: Reply<Int>) : Till

/** Takes nothing back: the till answers `unhandled()`, and a refund is a dead letter. */
data class Refund(val pence: Int) : Till

data object Close : Till

fun till() = behaviour<Till, Int>(0) { _, total, message ->
    when (message) {
        is Ring -> become(total + message.pence)
        is Total -> stay().also { message.reply(total) }
        is Refund -> unhandled()
        Close -> stop()
    }
}

fun main() {
    val total = flock<Nothing, _> {
        val till = spawn("till", till())
        till.tell(Ring(250))
        till.tell(Ring(120))
        till.ask(1.seconds) { Total(it) }
    }
    println(total.getOrNull()?.getOrNull()) // 370
}

• A step answers what happens next: stay(), become(state), stop() or unhandled(). One message at a time, so the state needs no lock.
• tell and ask. A mailbox holds 1,024 messages by default (capacity on spawn). ask blocks the caller up to a timeout and answers an Either: an actor that stopped, took too long or could not be reached is an AskFailure value, not an exception.
• Failure and supervision. A behaviour declares its failure type as a third type parameter, behaviour<M, S, E>, for declared failures. Whether declared or thrown, the actor's restart schedule (an ordinary Schedule) decides: restart from the initial state after a delay, or stop.
• More. Timers (ctx.timers.after, every), receive timeouts, signals and watching, ServiceKey registration, group and pool routers, and entities for one actor per id on a single node.

Persistence

A persistent actor is remembered by its events. A command answers an effect instead of a next state: persist(events) writes them to the journal then applies them, none() writes nothing, and .then { } runs once the events are written, which is where a reply goes. On start, and after every restart, the actor replays its events before its first command.

import io.github.matthewjones372.lark.actor.AskFailure
import io.github.matthewjones372.lark.actor.EventCodec
import io.github.matthewjones372.lark.actor.InMemoryJournal
import io.github.matthewjones372.lark.actor.InMemorySnapshots
import io.github.matthewjones372.lark.actor.PersistenceId
import io.github.matthewjones372.lark.actor.Reply
import io.github.matthewjones372.lark.actor.StateCodec
import io.github.matthewjones372.lark.actor.ask
import io.github.matthewjones372.lark.actor.every
import io.github.matthewjones372.lark.actor.journal
import io.github.matthewjones372.lark.actor.persistent
import io.github.matthewjones372.lark.actor.snapshots
import io.github.matthewjones372.lark.actor.spawn
import io.github.matthewjones372.lark.actor.stop
import io.github.matthewjones372.lark.flock
import java.nio.ByteBuffer
import kotlin.time.Duration.Companion.seconds

sealed interface Account

data class Deposit(val pence: Long) : Account

data class Withdraw(val pence: Long, val reply: Reply<Boolean>) : Account

data class Balance(val reply: Reply<Long>) : Account

/** What happened to an account: a change to its balance, in pence. */
data class Moved(val pence: Long)

object MovedCodec : EventCodec<Moved> {
    override fun encode(event: Moved): ByteArray = ByteBuffer.allocate(Long.SIZE_BYTES).putLong(event.pence).array()

    override fun decode(bytes: ByteArray): Moved = Moved(ByteBuffer.wrap(bytes).long)
}

object BalanceCodec : StateCodec<Long> {
    override fun encode(state: Long): ByteArray = ByteBuffer.allocate(Long.SIZE_BYTES).putLong(state).array()

    override fun decode(bytes: ByteArray): Long = ByteBuffer.wrap(bytes).long
}

fun account(id: String) = persistent<Account, Moved, Long>(
    id = PersistenceId("account", id),
    empty = 0,
    codec = MovedCodec,
    command = { _, balance, command ->
        when (command) {
            is Deposit -> persist(Moved(command.pence))
            is Withdraw ->
                if (command.pence > balance) {
                    none().then { command.reply(false) }
                } else {
                    persist(Moved(-command.pence)).then { command.reply(true) }
                }
            is Balance -> none().then { command.reply(balance) }
        }
    },
    event = { balance, moved -> balance + moved.pence },
    snapshots = every(100, BalanceCodec),
)

fun main() {
    val balance = flock<Nothing, _> {
        journal(InMemoryJournal())
        snapshots(InMemorySnapshots())
        val first = spawn("a-1", account("a-1"))
        first.tell(Deposit(500))
        first.ask<Account, Boolean>(1.seconds) { Withdraw(200, it) }
        stop(first).await()

        // The same id, started again: it replays its events and carries on where the first left off.
        val again = spawn("a-1-again", account("a-1"))
        again.ask(1.seconds) { Balance(it) }
    }
    println(balance.getOrNull()?.getOrNull()) // 300
}

• Conflicts. An append that finds another writer got there first is reported as a JournalConflict, so supervision decides.
• Snapshots. With snapshots = every(n, codec) the actor saves its state every n events, and a start replays only what came after the newest save.
• Changing an event. versioned(current, codec, upgrades) wraps a codec so a replay reads events written in an older shape; versionedState does the same for snapshots.
• Durable storage. InMemoryJournal is for tests and trying things out. lark-actor-journal-jdbc provides JdbcJournal, JdbcSnapshots and JdbcOffsets over a DataSource, and ships its tables as a Liquibase changelog (lark/journal/jdbc/postgres.sql) for your changelog to include. Nothing creates tables at start.
• Read models. lark-actor-projection's Projection.follow is every event of one kind as a stream; runProjecting() saves each offset once its work is done. Delivery is at least once, so write handlers that are safe to repeat.

Remoting and clusters

Start with local actors. Remoting adds addresses and codecs; clustering adds membership and placement of entities across nodes. Persistence supplies the journal from which an entity can recover. Reliable delivery and idempotent handling address a separate question: what happens when a command or its acknowledgement is lost.

lark-actor-remote makes a flock a node that listens on a port (loopback unless you give it a host). You expose an actor with a codec, and another node reaches it through a ref made from its address. lark-cluster adds gossip membership, downing, sharded entities, singletons, reliable delivery and topics.

dependencies {
    implementation("io.github.matthewjones372:lark-actor-remote:0.0.0-site-SNAPSHOT")       // two nodes
    implementation("io.github.matthewjones372:lark-cluster:0.0.0-site-SNAPSHOT")            // membership, sharding, delivery, topics
    implementation("io.github.matthewjones372:lark-actor-journal-jdbc:0.0.0-site-SNAPSHOT") // a journal every node reaches
    implementation("io.github.matthewjones372:lark-actor-projection:0.0.0-site-SNAPSHOT")   // read models
}

A cluster node that finds its seeds through DNS and logs membership changes:

import io.github.matthewjones372.lark.Flock
import io.github.matthewjones372.lark.actor.behaviour
import io.github.matthewjones372.lark.actor.remote.node
import io.github.matthewjones372.lark.actor.spawn
import io.github.matthewjones372.lark.actor.stay
import io.github.matthewjones372.lark.cluster.Cluster
import io.github.matthewjones372.lark.cluster.Discovery
import io.github.matthewjones372.lark.cluster.Downing
import io.github.matthewjones372.lark.cluster.MemberEvent
import io.github.matthewjones372.lark.cluster.cluster
import io.github.matthewjones372.lark.logInfo
import kotlin.time.Duration.Companion.seconds

/** A node of the shop's cluster, whose seeds are every address the headless service's name resolves to. */
fun Flock<Nothing>.joinShop(): Cluster {
    val node = node("shop-1", 25520, host = "0.0.0.0")
    val cluster = cluster(
        node,
        Discovery.dns("shop.default.svc.cluster.local", port = 25520),
        downing = Downing.keepMajority(stableAfter = 20.seconds),
    )
    val watcher = spawn(
        "membership-log",
        behaviour<MemberEvent, Unit>(Unit) { _, _, event ->
            // isSelf, not the address: a node restarted where it was hears its earlier life downed at its own.
            if (event is MemberEvent.Downed && cluster.isSelf(event.member)) logInfo("cluster: this node was downed")
            stay().also { logInfo("cluster: $event") }
        },
    )
    cluster.subscribe(watcher)
    return cluster
}

Entities spread across the cluster by id. Every node calls sharding with the same arguments, and entity(id) is a ref that stays good while the entity moves (BasketCodec is a MessageCodec<Basket> you write field by field, or generate with one of the codec modules):

val baskets = cluster.sharding("basket", BasketCodec, passivateAfter = 10.minutes, role = "baskets") { basket() }
val report = cluster.singleton("daily-report", BasketCodec) { basket() }

baskets.entity("customer-42").run {
    tell(Put("sku-1"))
    ask(5.seconds) { Contents(it) }
}

Key guarantees:

• A remote tell is at most once. Frames for a peer queue up to 8,192; past that, or while the peer is unreachable, they become dead letters. Order holds for each sender, receiver and lane.
• No serialisation library is chosen for you. Write a MessageCodec, or use lark-actor-remote-protobuf, -avro or -kotlinx. A Reply or ActorRef inside a message crosses as an address, so an ask made on one node is answered from another.
• One owner per entity. Ids map to 256 shards by a hash every node computes the same way, with no coordinator. When a shard moves, the new owner waits until the old one has stopped it, so an entity never runs on two nodes at once.
• Partitions. Nothing moves on while a member is unreachable, until it comes back or is downed. Downing.keepMajority is the default; staticQuorum and lease suit a fixed size or an even split.
• Commands that must arrive. sharded.reliable(id) numbers, keeps and resends each command until the entity confirms it, and a persistent entity wrapped in delivered(…) drops duplicates. durable = true keeps unconfirmed commands in the journal so a crash loses nothing.
• Mutual TLS. Give each node a Tls.mutual(keys, password, trusted) and every connection is TLS 1.3 with a certificate on each side, checked against the node's name.
• Leaving cleanly. A node whose flock closes leaves the cluster first (up to leaveWithin, 30 seconds by default), so the others remove it rather than downing it. In a lark-app service, start the cluster through Actors.within so SIGTERM leaves the cluster before the actors stop.

Kafka: processing and commit progress together

The integration keeps “I have processed this record” attached to the record as it passes through the pipeline. That makes commit ordering part of the API, rather than a second loop a caller must keep in sync. Database side effects still need their own duplicate handling; a Kafka transaction does not make an arbitrary external write exactly once.

lark-kafka reads a topic as a Stream of Committed elements, each carrying its record's offset, and the only way to run one is runCommitting. A record is committed only after the work on it is done; forgetting to commit, or committing before the work, does not compile.

dependencies {
    // brings lark-stream and kafka-clients; pick a stream backend as well
    implementation("io.github.matthewjones372:lark-kafka:0.0.0-site-SNAPSHOT")
}

import io.github.matthewjones372.lark.kafka.Kafka
import io.github.matthewjones372.lark.kafka.Topic
import io.github.matthewjones372.lark.kafka.consume
import io.github.matthewjones372.lark.kafka.mapRecordOrFail
import io.github.matthewjones372.lark.kafka.runCommitting
import io.github.matthewjones372.lark.stream.Forks
import io.github.matthewjones372.lark.stream.start
import org.apache.kafka.common.serialization.StringDeserializer

val placing = Kafka.consume(consumerProperties, Topic("orders"), key = StringDeserializer(), value = StringDeserializer())
    .mapRecordOrFail { record -> shop.place(record.value()).bind() }
    .runCommitting()

val running = placing.start(Forks())   // one virtual thread polls, handles and commits

Write handlers to be idempotent wherever the guarantee is at least once. For exactly-once output, consume, transform and write in one transaction:

val orders = Kafka.transactionalProducer(producerProperties, transactionalId = "checkout-1", StringSerializer(), orderSerializer)

Kafka.consume(consumerProperties, Topic("carts"), key = StringDeserializer(), value = carts)
    .mapRecord { cart -> cart.value().checkout() }
    .transacted(orders) { order -> listOf(Topic("orders").record(order.id, order)) }
    .restartOnDefect(Schedule.exponential(100.milliseconds))
    .runFold(0L, Long::plus)

• Records are batched up to 500 or what arrives within 100 ms, and each batch is one transaction. A failure aborts it, and restartOnDefect reads again from the last committed offsets.
• deadLetters(topic) with divertLefts sends an unreadable record to a dead-letter topic, and returns only once the letter is acknowledged.
• A Decoder is told which throws are transient, so a schema registry that is down is retried rather than sent to the dead-letter topic.
• lark-kafka brings kafka-clients 4.3, which needs a broker at 2.1 or later. lark-kafka-pekko offers the same through Pekko's connector.

Futures and coroutine callers

You can adopt Core inside an application whose edges are already asynchronous. The direction matters: waiting for an existing future and offering new work as a future have different ownership and cancellation rules.

larkFuture is not in the verified 0.10.0 release or the v0.10.1 tag. The current source guide contains the complete boundary examples and explains cancellation. Core itself does not acquire a dependency on kotlinx-coroutines.

Observability

Use the same current-source build and local version; publish the modules named below before adding them.

Logging and metrics calls are part of the core lark module, so library and service code can log or count without taking a logger or registry as a dependency. Where the output goes depends on what is on the classpath; each integration registers itself through ServiceLoader, so adding the dependency is the whole setup.

dependencies {
    implementation("io.github.matthewjones372:lark-slf4j:0.0.0-site-SNAPSHOT")
    implementation("io.github.matthewjones372:lark-micrometer:0.0.0-site-SNAPSHOT")
    implementation("io.github.matthewjones372:lark-otel:0.0.0-site-SNAPSHOT")
}

Logs

logDebug, logInfo, logWarn and logError(message, cause) write to the bound Logger. logAnnotated binds key-value pairs that every line inside the block carries, including lines written on forks, and logSpan(name) adds <name>_ms.

import io.github.matthewjones372.lark.logAnnotated
import io.github.matthewjones372.lark.logInfo
import io.github.matthewjones372.lark.logSpan
import io.github.matthewjones372.lark.parMap

fun register(id: String, items: List<Int>): List<Unit> =
    logAnnotated("correlation_id" to id) {
        logSpan("register") {
            parMap(items) { item -> logInfo("checking $item") }   // each line: correlation_id=… register_ms=…
        }
    }

With lark-slf4j, each line goes to the SLF4J logger named after the top-level class that wrote it, and annotations are put in the MDC for the call, so %X{correlation_id} works. It compiles against slf4j-api 1.7 and works with 2.x.

Metrics

Meters are looked up by name when used, so nothing is declared up front:

import io.github.matthewjones372.lark.counter
import io.github.matthewjones372.lark.gauge
import io.github.matthewjones372.lark.increment
import io.github.matthewjones372.lark.metricTagged
import io.github.matthewjones372.lark.timed

class Adoptions(private val waiting: MutableList<String>) {
    fun adopt(name: String): String = timed("petshop.adopt.duration.seconds") {
        counter("petshop.adoptions").increment()
        gauge("petshop.queue.depth").set(waiting.size.toDouble())
        waiting.remove(name)
        name
    }

    fun adoptTortoise(): Unit = metricTagged("species" to "tortoise") {
        counter("petshop.adoptions").increment()
    }
}

timed records in seconds, whether the block returns or throws. With lark-micrometer, bind MicrometerMetrics(registry) with metrics.locally(…) { } to use a registry other than the global one. The library's own meters start with lark.: policies (lark.policy.calls, lark.breaker.state, and so on), stream stages run with measured, and actors, remoting, sharding and delivery.

Traces

lark-otel keeps OpenTelemetry's Context in a LarkLocal, so a span opened before a parMap is the parent of each branch's spans. It registers a ContextStorageProvider for the whole JVM, which is why it is a separate module.

import io.github.matthewjones372.lark.logInfo
import io.github.matthewjones372.lark.otel.span
import io.github.matthewjones372.lark.otel.tracedSpan
import io.github.matthewjones372.lark.parMap
import io.opentelemetry.api.trace.Tracer

fun pricing(tracer: Tracer, orders: List<Order>): List<Price> =
    tracer.span("price-all") {                                   // a throw is recorded with ERROR status
        parMap(orders) { order -> tracer.span("price") { price(order) } }
    }

fun registering(tracer: Tracer) =
    tracer.tracedSpan("register") { logInfo("started") }        // the line carries trace_id and span_id

Context also travels with actor messages (locally and across nodes) and is stored with journal events. For Kafka it is explicit: ProducerRecord.carrying(Carriers.capture()) writes it into headers, and ConsumerRecord.within { } restores it.

Testing

Use the same current-source build and local version; publish the modules named below before adding them.

Time and context are values bound for a block, so a test can replace the clock, the logger or the metrics registry without touching the code under test, and nothing has to sleep.

Time

fixedClock() never advances, and a sleep on it returns at once, so a five-retry backoff runs in microseconds. TestClock() moves only when the test calls adjust(by) or setTime(to); each adjust releases every sleeper due by the new time, in order. adjustWhenBlocked(by) first waits until every sleeping fork has reached its sleep.

import io.github.matthewjones372.lark.Schedule
import io.github.matthewjones372.lark.clock
import io.github.matthewjones372.lark.fixedClock
import io.github.matthewjones372.lark.retry
import kotlin.time.Duration.Companion.seconds

val backoff = Schedule.exponential<Throwable>(1.seconds) and Schedule.recurs(5)

var attempts = 0

fun flakyRow(): String = if (++attempts < 4) error("not yet") else "row"

fun withoutWaiting(): String = clock.locally(fixedClock()) { backoff.retry { flakyRow() } }   // no real waiting

Logs and metrics

import io.github.matthewjones372.lark.LogLine
import io.github.matthewjones372.lark.capturingLogs
import io.github.matthewjones372.lark.capturingMetrics
import io.github.matthewjones372.lark.counter
import io.github.matthewjones372.lark.increment
import io.github.matthewjones372.lark.logAnnotated
import io.github.matthewjones372.lark.logInfo

fun linesLogged(): List<LogLine> = capturingLogs { logs ->
    logAnnotated("correlation_id" to "abc-123") { logInfo("registered") }
    logs.all()
}

fun adoptionsCounted(): Double = capturingMetrics { measured ->
    counter("petshop.adoptions").increment()
    measured.counter("petshop.adoptions")
}

Stories and eventually

lark-test is for tests of a whole service. story { } writes a test as Given, When, Then, And and But steps that return values; when a step fails, the message shows the story up to that step and what it threw. eventually(within) { } retries a block every 20 ms by default until it stops throwing, and gives up with GaveUp, an AssertionError, once within has passed on the inherited clock.

import io.github.matthewjones372.lark.test.story
import io.kotest.matchers.shouldBe
import org.junit.jupiter.api.Test
import kotlin.time.Duration.Companion.seconds

class AdoptionTest {

    // `PetShop` belongs to the service
    @Test
    fun `an adopted pet is listed as taken`() = story {
        val shop = Given("a shop with three pets") { PetShop(listOf("Nibbles", "Barnaby", "Mrs Peel")) }
        val taken = When("Ada adopts the first") { shop.adoptFirst("Ada") }
        Then("Ada has Nibbles") { taken shouldBe "Nibbles" }
        And("the listing says so").eventually(within = 5.seconds) { shop.status("Nibbles") shouldBe "taken" }
    }
}

lark-test brings no assertion library. The example uses Kotest's matchers, but any assertion that throws works.

Part of an application

testApp(module) { root: Root -> … } starts a graph, runs the block with its root, and releases the graph however the block ends, a failed assertion included. overriding replaces nodes, and subgraph<Root>() keeps only the nodes the root is reached through. Using the worked example from Building a service:

import io.github.matthewjones372.lark.app.overriding
import io.github.matthewjones372.lark.app.single
import io.github.matthewjones372.lark.app.subgraph
import io.github.matthewjones372.lark.app.testApp
import io.kotest.matchers.shouldBe
import org.junit.jupiter.api.Test

class InMemoryUsers : UserRepo {
    override fun find(id: Long) = "fake-$id"
}

class RegisterUserTest {

    @Test
    fun `registering reads the user repository`() {
        val underTest = app
            .overriding(single<UserRepo> { InMemoryUsers() })
            .subgraph<RegisterUser>()

        testApp(underTest) { register: RegisterUser ->
            register.register(7) shouldBe "fake-7"
        }
    }
}

Overriding first means the fake has no dependencies, so subgraph then drops the pool, the config and Sys, and the test needs no DB_URL.

Actors and streams

behaviour.test() runs one actor on the calling thread, and testActors { } runs several. A send returns once that message and everything it caused has been handled, restart delays are recorded rather than waited out, and advance moves the test's clock and delivers due timers.

import io.github.matthewjones372.lark.Schedule
import io.github.matthewjones372.lark.actor.Failure
import io.github.matthewjones372.lark.actor.Reply
import io.github.matthewjones372.lark.actor.become
import io.github.matthewjones372.lark.actor.behaviour
import io.github.matthewjones372.lark.actor.stay
import io.github.matthewjones372.lark.actor.test
import io.github.matthewjones372.lark.actor.unhandled
import io.kotest.matchers.shouldBe
import org.junit.jupiter.api.Test
import kotlin.time.Duration.Companion.milliseconds

sealed interface Till

data class Ring(val pence: Int) : Till

data class Total(val reply: Reply<Int>) : Till

data class Refund(val pence: Int) : Till

fun till() = behaviour<Till, Int>(0) { _, total, message ->
    when (message) {
        is Ring -> {
            require(message.pence > 0) { "rang ${message.pence}" }
            become(total + message.pence)
        }
        is Total -> stay().also { message.reply(total) }
        is Refund -> unhandled()
    }
}

class TillTest {

    @Test
    fun `a till totals what it rang, and leaves a refund unhandled`() {
        val till = till().test()

        till.send(Ring(250))
        till.send(Refund(100))

        till.ask { Total(it) }.getOrNull() shouldBe 250
        till.unhandled shouldBe listOf(Refund(100))
    }

    @Test
    fun `a till that fails restarts empty, and each back-off is recorded rather than waited`() {
        val backingOff = Schedule.exponential<Failure<Nothing>>(100.milliseconds) zipLeft Schedule.recurs(5)
        val till = till().test(restart = backingOff)

        till.send(Ring(250))
        till.send(Ring(-1))
        till.send(Ring(-2))

        till.state shouldBe 0
        till.delays shouldBe listOf(100.milliseconds, 200.milliseconds)
    }
}

testActors(order = Ordering.seeded(42)) replays message orderings from a seed, and Faults(drop = …, duplicate = …, delay = …) injects lost, doubled and late messages between actors. For streams, TestStreams(clock) makes tick, groupedWithin and restartOnDefect wait only on the test's clock:

val clock = TestClock()
val running = Stream.tick(1.minutes, "t").runCollect().start(TestStreams(clock))

clock.adjust(1.hours)
running.emitted().size shouldBe 60

Recipes

Use the same current-source build and local version; publish the modules named below before adding them.

Short how-tos from the application cookbook, for things a new service meets in its first week. Each assumes the types from the worked example.

Read an environment variable, or refuse to start

val fromTheProcess: Module =
    single<Sys> { RealSys } +
        single { sys: Sys -> DbConfig(sys.env("DB_URL") ?: refuse("DB_URL is not set")) }

A node calling System.getenv directly is a node no test can configure. In a test, provide FakeSys(mapOf("DB_URL" to "jdbc:postgresql://db/app")).

Read a HOCON section

For the local source build, add io.github.matthewjones372:lark-app-typesafe:0.0.0-site-SNAPSHOT. A failed read records its fault and carries on, so a bad file produces one message listing every fault.

import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory
import io.github.matthewjones372.lark.app.typesafe.config

data class DbSettings(val url: String, val poolSize: Int, val idle: kotlin.time.Duration)

val database: Module =
    single<Config> { ConfigFactory.load() } +
    config<DbSettings>("database") { DbSettings(string("url"), int("poolSize"), duration("idle")) } +
    closingOf { db: DbSettings -> Hikari(DbConfig(db.url)) }.boundTo<Pool>()

The readers are string, int, long, boolean, duration, bytes, strings and secret. Use secret("path") for passwords and tokens: a Secret prints as a mask and gives its value only to reveal().

Wait for something slow to start

val probed: Module =
    closingOf(::Hikari).boundTo<Pool>()
        .probe("db", timeout = 2.seconds, attempts = 10, interval = 500.milliseconds) { pool: Pool ->
            pool.healthy()
        }

Dependents wait for the probe, not the constructor. critical = false degrades readiness rather than downing it.

Answer /ready and /live

class Routes(private val health: HealthRegistry) {
    fun ready() = health.readiness()
    fun live() = health.liveness()
}

val routed: Module = probed + single { health: HealthRegistry -> Routes(health) }

readiness() never throws; liveness() answers Down with the reason starting until the graph is running.

Two values of the same type

One key, one instance. A @JvmInline value class costs nothing at runtime and says which one a dependent wants:

@JvmInline
value class Replica(val pool: Pool)

val replicated: Module =
    single { db: DbConfig -> Hikari(db) as Pool } +
    single { db: DbConfig -> Replica(Hikari(DbConfig(db.url + "-replica"))) } +
    single { primary: Pool, replica: Replica -> PgUserRepo(replica.pool) as UserRepo }

A node built by a Java factory

A Java method returns a platform type such as Tracer!, which matches no node asking for Tracer; the graph then reports it missing. Name the key with boundTo:

val tracing: Module =
    single { sdk: OpenTelemetrySdk -> sdk.getTracer("petshop") }.boundTo<Tracer>()

A release that knows how the application ended

val closing: Module = single { db: DbConfig ->
    install({ Hikari(db) }) { pool, exit ->
        if (exit is ExitCase.Failure) logWarn("closing the pool after a failure")
        pool.close()
    } as Pool
}

Run migrations before anything reads the database

For the local source build, add io.github.matthewjones372:lark-app-liquibase:0.0.0-site-SNAPSHOT. Anything that reads the database takes a Migrated, which turns "after the migrations" into an edge the graph enforces:

import io.github.matthewjones372.lark.app.liquibase.Migrated
import io.github.matthewjones372.lark.app.liquibase.migrations

val database: Module =
    closingOf { cfg: DbConfig -> HikariDataSource(cfg) }.boundTo<DataSource>() +
    migrations("db/changelog.xml") +
    single { db: DataSource, _: Migrated -> PgUserRepo(db) as UserRepo }

Check the wiring in a unit test

fun theApplicationWires() = app.validate()                         // Either<NonEmptyList<WiringError>, Plan>

fun theWiringIsWhatWeThink(): String = app.render()                // Mermaid, stable order: keep it in a golden file

fun everyFault(): String = Shop.module.findings(Shop.root).report()  // errors and warnings, as text

Current source additions

Lark’s own declared-error API

Current source has its own Fail<E> scope and Either<E, A>. attempt opens the scope; fail(error) exits it with a declared error; bind() consumes another result. Parallel combinators carry that failure scope across their own forks. lark-arrow supplies conversions at integration boundaries. See typed errors across forks for accumulation and installation for source-version requirements.

import io.github.matthewjones372.lark.fail.Either
import io.github.matthewjones372.lark.fail.attempt
import io.github.matthewjones372.lark.fail.fail

data class MissingUser(val id: Int)

fun userName(id: Int): Either<MissingUser, String> = attempt {
    if (id != 1) fail(MissingUser(id))
    "Ada"
}

A thrown exception remains an exception. A failure scope must not escape its lifetime or be invoked from an arbitrary thread: use Core’s combinators to cross forks. See the failure-scope tests.

One concurrency budget for a dependency

A fixed per-pipeline limit can still overload a database when several pipelines run together. Current source lets them share a Limit. Limit.fixed(n) fixes the budget; Limit.adaptive adjusts it using measured call durations, within a configured maximum. Share the same value between callers whose work competes for the same capacity.

Streams can use that budget with mapPar(limit) or mapParUnordered(limit). The ordered form preserves input order; the unordered form can emit a finished result while an earlier call is still waiting. Adaptive concurrency is feedback from this workload, not proof of a dependency’s safe capacity. See the worked example and metrics.

Configuration with an explanation

A module can own and validate its own HOCON section. Current source reports where each setting came from and, when layers were preserved, what it overrode. A startup fault can name all bad settings together. lark-app-kotlinx also derives section readers from serializable settings classes, with defaults and naming rules.

Use Secret fields or explicit redaction paths for sensitive values: a name-based mask cannot recognise every credential. Configuration can also select a module, so an in-process implementation and a broker-backed implementation are graph choices rather than branches scattered through the service. See configuration ownership and module selection.

JUnit suites with a shared resource

lark-test supplies stories and waiting helpers for an existing test framework. lark-junit goes further: it is a JUnit Platform engine. A Suite<R> owns one Resource<R>, acquires it once, passes it to the tests and releases it after the last test. This fits an expensive test server or fixture whose ownership should be explicit.

import io.github.matthewjones372.lark.Resource
import io.github.matthewjones372.lark.junit.Suite
import io.github.matthewjones372.lark.junit.expect
import io.github.matthewjones372.lark.resource
import java.util.concurrent.ConcurrentHashMap

object PetShopSuite : Suite<ConcurrentHashMap<String, String>>() {
    override val shared: Resource<ConcurrentHashMap<String, String>> = resource {
        install({ ConcurrentHashMap<String, String>() }) { pets, _ -> pets.clear() }
    }

    init {
        test("a listed pet is available") { pets ->
            pets["Nibbles"] = "available"
            expect(pets["Nibbles"] == "available")
        }
        test("an unlisted pet has no status") { pets ->
            expect(pets["Mrs Peel"] == null)
        }
    }
}

Tests run as virtual-thread forks under the suite’s concurrency limit; suites run sequentially. Use alone for a test that must wait for earlier tests and finish before later ones start. Sharing a fixture does not isolate mutable state: use separate records or run conflicting tests alone.

A test returns Expectations, so several checks can be combined as values and a body returning only Unit does not compile. Failed tests attach their captured logs; passing tests stay quiet. Results use the JUnit Platform’s normal reports, alongside Jupiter tests. The suite guide includes a shared-resource example, expectation composition and aspects.

Gotchas and limits

• Cancellation is interruption. Work that never reaches an interruptible call, or swallows InterruptedException, cannot be stopped early, by cancel(), a failing sibling or a timeout.
• timeout under fixedClock() gives up at once. The sleeper waits on the inherited clock, and a sleep on fixedClock() returns immediately. Test a timeout with a TestClock.
• parMap has no concurrency limit. Cap it with a Bulkhead or use a stream's mapPar.
• A Fail scope does not cross threads by itself. Work handed to a callback or another thread should answer an Either for the handler to bind.
• An MDC entry does not survive a fork, because it lives in a ThreadLocal. Use logAnnotated, which is bound in a LarkLocal.
• Policy steps have a fixed order, checked when the policy is built.
• No partial type-argument inference. single<UserRepo, DataSource> { ds -> … } has to name the dependency as well as the key; boundTo avoids that. A node with no dependencies needs its type argument: single<Sys> { RealSys }.
• One instance per key, with no scopes or named instances. Use a wrapper type for a second.
• The editor check reads a subset of shapes. For anything else, including a module defined in another Gradle module, it reports LARK_APP_UNREAD; the larkWiring Gradle task is the authoritative check.
• Remote tells are at most once. Use reliable delivery for commands that must arrive.
• Pekko-specific stream operators run only on PekkoStreams.

Status and versioning

The library is experimental. Everything described in its guides is implemented and tested, the build enforces 90% line coverage, and the guides' examples are compiled by tests. But:

• The API is not stable. Releases are 0.x, any release can change or remove API, and there is no binary-compatibility check. Pin an exact version, and expect to update code when you upgrade.
• No production use is known. Beyond its tests, the code is exercised by the example application lark-bank and by related projects such as petshop and tweet-street.
• Maturity varies. The core, lark-app and lark-stream are the most established. The actor, cluster and Kafka modules are newer.
• Some modules wrap experimental APIs. lark-actor-remote-kotlinx uses kotlinx.serialization's ProtoBuf format, which kotlinx marks experimental.
• Source and published artifacts differ. Central resolves 0.10.0; the tag v0.10.1 also contains closingOf. Current main is further ahead. Unreleased changes include Lark’s own error types and Arrow interop, lark-junit, the default run on LarkApp, choice for a setting that picks a module, and lark-app-kotlinx.

Applications in practice

These excerpts show the libraries working together. Both applications track snapshots; use their own build files and READMEs when running them.

Petshop: wiring and tests

Petshop wires the service as a checked graph. Its JVM service is separate from the hosted fake shop.

actor<Shop, Outbox, ShopWindow>("shop") { outbox, window ->
    shop(outbox, window, outbox.shelf(opening).associateBy { it.id })
} +
closingOf { ref: ActorRef<Shop>, registry: ChipRegistry,
            window: ShopWindow, streams: StreamBackend ->
    Adoption(ref, registry, window, streams)
} +
single { window: ShopWindow, adoption: Adoption, tracer: Tracer ->
    ObservedShop(ShopFront(window, adoption), tracer)
}.boundTo<PetShop>()

docker compose -f demo/docker-compose.yml up -d postgres registry
./gradlew :app:run     # API docs at /api-docs
./gradlew test         # containers shared per run; fresh schema per graph
./gradlew loadTest     # 200 requests a second

tweet-street: a transfer saga

tweet-street records transfers as events and drives the next step of each saga. Run the chaos script from lark-bank/.

fun Ctx<TransferMessage>.drive(written: Transfer) {
    val state = heard ?: written
    when (val step = state.next(id, screens)) {
        is Step.DebitSource       -> sendLeg(step, step.account, step.command)
        is Step.CreditDestination -> sendLeg(step, step.account, step.command)
        is Step.RefundSource      -> sendLeg(step, step.account, step.command)
        // Screening and completion branches omitted; see Transfers.kt.
        // …
    }
}

scripts/chaos-docker.sh   # 200 transfers a second while it breaks things
                          # then checks the ledger still balances

Module inventory

Each module brings the modules it builds on and, at most, one external library. A NoOtherDependenciesTest in every module checks its runtime classpath against that list, so a new transitive dependency fails the build. The table includes published modules and marks additions available only on main.

A good starting set: lark for any service; lark-app for wiring and lifecycle; lark-stream plus lark-stream-forks for streams (unless you already run Pekko Streams); lark-actor for actors; lark-test for tests.

Where next

The links below pin the same source revision as this guide. Some upstream prose still describes the earlier error API; the declared-error examples here use Lark’s current API. Follow the latest source guides for subsequent changes.