Streaming in Scala with FS2: bounded work, explicit effects

A stream is a program, not a collection already in memory

A List[A] contains values that have already been produced. An FS2 Stream[F, A] describes how zero or more A values can be produced while effects in F are evaluated. Constructing the stream does not run those effects; compiling it to an effect and running that effect does.

This distinction lets one program represent a small finite input, a file larger than memory or a source that continues until the application shuts down. The stream can transform values as they arrive without materialising the complete input.

Streaming is useful when work must remain bounded or incremental. It is unnecessary ceremony when the input is already a small collection and an ordinary map or fold communicates the operation more directly.

import cats.effect.IO
import fs2.Stream

val accounts: Stream[IO, Account] =
  Stream.emits(accountIds)
    .covary[IO]
    .evalMap(repository.find)
    .unNone

val count: IO[Long] = accounts.compile.count
val all: IO[List[Account]] = accounts.compile.toList

// Defining accounts performs no repository calls.
// Running count or all interprets the stream.

Demand keeps the pipeline bounded

FS2 is pull-based: downstream demand drives upstream production. If processing one value is slow, an ordinary sequential pipeline does not keep pulling and accumulating an unlimited number of values ahead of it.

That behaviour is often described as backpressure. It is a property of the complete pipeline, not a promise that every integration is automatically safe. An adapter that reads an entire response or database result before emitting its first element has already lost the memory benefit.

Inspect each boundary for buffering, batching and prefetching. Queue capacity and concurrent evaluation deliberately allow work ahead of demand, so give them limits based on measured throughput and failure behaviour rather than an arbitrary large number.

def nextAccount: IO[Option[Account]] = ???
def process(account: Account): IO[Unit] = ???

val program: Stream[IO, Unit] =
  Stream
    .repeatEval(nextAccount)
    .unNoneTerminate
    .evalMap(process)

val run: IO[Unit] = program.compile.drain

Use map for values and evalMap for effects

Stream.map applies a pure function to each emitted value. evalMap is for a function whose result is already wrapped in the stream effect, such as an IO database call or HTTP request. Keeping that distinction visible makes it easier to see where execution, failure and latency enter the pipeline.

A common mistake is calling an effectful API inside map and producing Stream[IO, IO[Result]]. The nested IO values are merely emitted; they are not automatically executed. evalMap sequences each returned effect and emits its successful result.

Keep parsing and domain decisions pure where possible. Lift only the operations that genuinely interact with the outside world. This produces smaller units that can be tested without running the complete stream.

def parse(line: String): Either[ParseError, Account] = ???
def save(account: Account): IO[Unit] = ???

val parsed: Stream[IO, Either[ParseError, Account]] =
  lines.map(parse)

val saved: Stream[IO, Unit] = parsed
  .collect { case Right(account) => account }
  .evalMap(save)

Chunks make batching explicit

FS2 processes values internally in chunks so it does not pay an allocation cost for every element at every stage. Application code can also introduce meaningful batches with chunkN when a database or remote API is more efficient with several records at once.

Batch size changes memory use, latency and failure scope. Larger batches can improve throughput but hold more data, delay the first result and create more work to retry when one request fails. The external system may impose its own payload or parameter limit.

Preserve the relationship between an input batch and its result. If a bulk operation partially succeeds, a return type containing only Unit cannot identify which records need retrying or reconciliation.

def insertBatch(accounts: List[Account]): IO[BatchResult] = ???

val results: Stream[IO, BatchResult] = accounts
  .chunkN(100, allowFewer = true)
  .evalMap(chunk => insertBatch(chunk.toList))

val completed: IO[List[BatchResult]] =
  results.compile.toList

Resource safety belongs in the stream lifecycle

Long-running pipelines open files, responses, connections and other resources whose lifetime must cover consumption rather than construction. FS2 integrates with Cats Effect Resource so acquisition and release follow success, failure and cancellation.

Stream.resource acquires a Resource when the stream runs and releases it when that scope finishes. Library stream constructors for files and HTTP bodies can provide the same guarantee when their documented scope is respected.

Do not return a stream that depends on a resource already closed by an outer use block. The resource scope must wrap the part of the stream that consumes it. Cancellation should be tested for custom adapters because finalisation is part of their contract.

import cats.effect.{IO, Resource}
import fs2.Stream

def openCursor: Resource[IO, AccountCursor] = ???

def rows(cursor: AccountCursor): Stream[IO, Account] = ???

val accounts: Stream[IO, Account] =
  Stream.resource(openCursor).flatMap(rows)

// The cursor remains open during rows and is released when
// consumption completes, fails or is cancelled.

Concurrency must stay bounded and intentional

evalMap runs one effect at a time. parEvalMap allows a fixed number to run concurrently while preserving output order. parEvalMapUnordered emits results as soon as they complete, which can reduce waiting when order has no meaning.

The concurrency number is not merely a performance setting. It controls pressure on connection pools, rate-limited APIs, CPU and downstream services. Increasing it can reduce throughput when contention, throttling or retries dominate.

Choose ordering from the domain. Reordering independent image inspections may be harmless; reordering account updates or audit records may not be. If several events for one entity must remain sequential, partitioning and per-key ordering require an explicit design rather than a global concurrency increase.

def enrich(account: Account): IO[EnrichedAccount] = ???

val ordered: Stream[IO, EnrichedAccount] =
  accounts.parEvalMap(maxConcurrent = 8)(enrich)

val completionOrder: Stream[IO, EnrichedAccount] =
  accounts.parEvalMapUnordered(maxConcurrent = 8)(enrich)

Model expected rejection separately from stream failure

An effect failure normally terminates the stream and runs its finalisers. That is appropriate for an unavailable database, a broken input connection or another condition that prevents useful continuation. A malformed row in an import may instead be an expected value that should be recorded while later rows continue.

Represent recoverable outcomes with Either or a domain result type inside the stream. Use attempt when the policy genuinely treats an effect failure as data, and recover only errors the current boundary understands. Catching every Throwable per element can hide an outage behind a successful-looking job.

Retry is also a policy, not a generic decoration. Retrying an idempotent read differs from retrying a write whose first result is unknown. Bound attempts, preserve the final cause and expose retry activity through metrics or logs.

sealed trait ImportResult
final case class Imported(account: Account) extends ImportResult
final case class Rejected(line: String, error: ParseError) extends ImportResult

def importLine(line: String): IO[ImportResult] =
  parse(line) match {
    case Left(error) => IO.pure(Rejected(line, error))
    case Right(account) =>
      repository.save(account).as(Imported(account))
  }

// A repository failure still fails the stream.
val results: Stream[IO, ImportResult] = lines.evalMap(importLine)

Keep stream boundaries visible in the application

A Stream return type is useful when the caller benefits from incremental consumption, cancellation or composition. It is less useful when a service immediately compiles it internally and returns a collection, or when a domain method exposes FS2 despite every caller needing one small finite result.

Place decoding, transport framing and persistence adapters near their respective boundaries. Convert bytes into validated application values before business rules, and translate domain outcomes into output framing afterwards. This prevents a pipeline from becoming one long expression mixing protocol, business and storage decisions.

In an http4s application, streaming request and response bodies fit the same Cats Effect runtime naturally. A Play application may use a different streaming integration. Bridging models is possible, but the ownership of execution, cancellation and materialisation must be explicit.

• Does the caller need incremental values or only one final result?
• Which layer owns decoding and validation?
• Where is the stream compiled and run?
• Which runtime owns cancellation and shutdown?
• Does an adapter buffer the complete input before emitting?
• Is a framework bridge adding another execution model?

Test emitted values, effects and termination

A finite stream can be compiled to List, Vector, a count or a final fold in a test. This is useful for transformation and ordering assertions, but it does not by itself prove resource release, cancellation or behaviour under slow demand.

Keep pure parsing and domain transitions covered as ordinary function tests. For the pipeline, use controlled sources and small fakes to assert emitted results and externally visible effects. Add integration tests for real file, database or HTTP adapters because buffering and failure semantics live there.

Avoid tests that sleep and hope concurrent work has finished. Use effect coordination such as Deferred or Ref when completion order is the behaviour under test, and apply a test timeout so a stream that never terminates produces a useful failure.

val source: Stream[IO, String] = Stream.emits(List(
  "valid-account",
  "invalid-account"
)).covary[IO]

val result: IO[List[ImportResult]] =
  source.evalMap(importLine).compile.toList

result.map { values =>
  assert(values.length == 2)
  assert(values.exists(_.isInstanceOf[Rejected]))
}

Operate the pipeline as a production process

A stream that runs correctly on a laptop still needs an operational contract. Record input and output counts, rejection and failure counts, processing latency, queue or batch utilisation and the last successfully committed position where work can resume.

Shutdown matters for continuous streams. Decide whether interruption should finish in-flight work, abandon it safely or persist a checkpoint. Ensure the application runtime waits for finalisers and that deployment termination windows match the work required for a clean stop.

Performance tests should vary input rate, downstream latency and failure behaviour rather than measure only the happy path. Watch heap, connection pools and queue depth alongside throughput. Streaming succeeds when resource use stays bounded, failures remain diagnosable and the system can resume without guessing what was processed.

• Measure received, processed, rejected and failed elements
• Use bounded queues and report saturation
• Correlate batches or records without unbounded metric labels
• Define retry, duplicate and checkpoint behaviour
• Exercise cancellation and graceful shutdown
• Load-test the slowest real boundary, not only pure transformations

• FS2 guide ↗
• Cats Effect Resource documentation ↗