Surefold core

What it is

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.

Since JDK 21, a blocked virtual thread parks without holding an OS thread. 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 carries no 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 can raise a declared error from Arrow's Raise, and the combinator returns it as the scope's Left. Code already inside either { } needs no new scope.
• 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.

The core module depends only on arrow-core. Its combinators have the same names and shapes as arrow-fx-coroutines, so moving a function across usually means removing suspend and changing an import. No published module uses coroutines, starts an event loop or depends on kotlinx-coroutines.

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 already keep errors in Arrow's Either and want them to survive 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. There is no coroutine interop layer. You can call these functions from coroutine code as you would any blocking call, on a dispatcher that tolerates blocking, but nothing more.

Install

Every published module shares the group io.github.matthewjones372 and one version, and is on Maven Central. The latest release is 0.10.1. You need JDK 25 or newer at runtime (the libraries are compiled for 25), and the libraries are built with Kotlin 2.4.10.

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

repositories { mavenCentral() }

kotlin { jvmToolchain(25) }

dependencies {
    // brings arrow-core and nothing else
    implementation("io.github.matthewjones372:lark:0.10.1")

    // add only what the service uses, for example:
    implementation("io.github.matthewjones372:lark-app:0.10.1")
    testImplementation("io.github.matthewjones372:lark-test:0.10.1")
}

Modules

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. These are the published modules.

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.

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 arrow.core.Either
import arrow.core.raise.either
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> = either {
    timeout(2.seconds) {
        parZip(
            { findUser(id) ?: raise(UserNotFound(id)) },
            { recentOrders(id) },
        ) { user, orders -> Summary(user, orders) }
    }
}

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

Run main from your IDE, or apply Gradle's application plugin and run ./gradlew run. It prints:

Either.Right(Summary(user=User(id=1, name=Ada), orders=[Order(id=7, pence=1200)]))
Either.Left(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 raises

   When the user is missing, the first branch calls raise(UserNotFound(id)). Its sibling, the orders call, is interrupted and joined, and summary returns Left(UserNotFound(id)). No exception is thrown.

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. 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

Everything in this section is in the package io.github.matthewjones372.lark, 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 raising or throwing, none of its work is still running. Four shapes cover most needs:

import arrow.core.Either
import arrow.core.raise.either
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> = either {
    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>> = either { 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 Raise, 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 Raise<E> it returns A and a raise propagates to the enclosing scope.
• A Deferred nobody awaits is still joined when its flock closes. If it raised, the raise 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 either { } (or any Raise<E>), a branch can bind() an Either or raise an error, and the first raise 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 raises, in branch order:

import arrow.core.Either
import arrow.core.NonEmptyList
import arrow.core.raise.either
import arrow.core.raise.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> = either {
    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>> = either {
    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 JDK has no cancellation other than interruption, so a cancelled branch stops at its next interruptible blocking call: a sleep, a lock, a socket read, Future.get. A branch that computes without blocking runs on until it reaches one.

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 arrow.core.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, raising, 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 arrow.core.Either
import arrow.core.raise.either
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> = either {
    resourceScope {
        val db = connection.bind()                                  // released when this scope ends
        val out = install({ StringBuilder() }) { _, _ -> }
        val rows = db.rows(id).ifEmpty { raise(NotFound(id)) }      // a raise 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 raise 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

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 arrow.core.Either
import arrow.core.raise.either
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> = either {
    timeout(2.seconds) { findRow(id).bind() }                       // throws TimeoutException
}

fun maybeQuote(id: Int): Either<Unavailable, String?> = either {
    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 retryRaise 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 arrow.core.Either
import io.github.matthewjones372.lark.Schedule
import io.github.matthewjones372.lark.retry
import io.github.matthewjones372.lark.retryRaise
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 raise is the Left.
fun row(id: Int): Either<Unavailable, String> = Schedule.recurs<Unavailable>(5).retryRaise { 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 arrow.core.Either
import arrow.core.raise.either
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> = either {
        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 raise 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 raise 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 {
    // brings lark, and through it arrow-core
    implementation("io.github.matthewjones372:lark-app:0.10.1")
}

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.

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. It is on Maven Central, not the Gradle Plugin Portal, so the plugin repositories must include Maven Central:

pluginManagement {
    repositories {
        mavenCentral()
        gradlePluginPortal()
    }
}

plugins {
    kotlin("jvm") version "2.4.10"
    id("io.github.matthewjones372.lark.wiring") version "0.10.1"
}

repositories { 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

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

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.10.1")
    // time a test owns
    testImplementation("io.github.matthewjones372:lark-stream-test:0.10.1")
}

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 ?: raise(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 Raise<E>:

import arrow.core.Either
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

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 arrow.core.Either
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: Either<Nothing, Either<AskFailure, Int>> = flock {
        val till = spawn("till", till())
        till.tell(Ring(250))
        till.tell(Ring(120))
        till.ask(1.seconds) { Total(it) }
    }
    println(total) // Either.Right(Either.Right(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>, and raises it. Raised 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 arrow.core.Either
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: Either<Nothing, Either<AskFailure, Long>> = flock {
        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) // Either.Right(Either.Right(300))
}

• Conflicts. An append that finds another writer got there first is raised 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

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.10.1")       // two nodes
    implementation("io.github.matthewjones372:lark-cluster:0.10.1")            // membership, sharding, delivery, topics
    implementation("io.github.matthewjones372:lark-actor-journal-jdbc:0.10.1") // a journal every node reaches
    implementation("io.github.matthewjones372:lark-actor-projection:0.10.1")   // 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

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.10.1")
}

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.

Observability

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.10.1")
    implementation("io.github.matthewjones372:lark-micrometer:0.10.1")
    implementation("io.github.matthewjones372:lark-otel:0.10.1")
}

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

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 arrow.core.right
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, String>(0) { _, total, message ->
    when (message) {
        is Ring -> if (message.pence > 0) become(total + message.pence) else raise("rang ${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) } shouldBe 250.right()
        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<String>>(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

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

Add io.github.matthewjones372:lark-app-typesafe:0.10.1. 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

Add io.github.matthewjones372:lark-app-liquibase:0.10.1. 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

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 Raise 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.
• main is ahead of 0.10.1. Unreleased so far: the default run on LarkApp, choice for a setting that picks a module, and lark-app-kotlinx.

Where next

The project's own guides go deeper on every section above.