Module 5: Ownership and cleanup

The concrete friction

Lexical try...finally fails across asynchronous boundaries and interruptions

Operating system resources (file handles, database connections, locks) are strictly finite. In synchronous code, try...finally guarantees cleanup on block exit. But in async and concurrent code, a resource may be acquired in one helper, used in another, or passed to a concurrent fiber. If a fiber is cancelled, interrupted midway, or encounters an asynchronous race, standard finally blocks either run prematurely or fail to run at all, leaking locks and exhausting connection pools.

First-principles consequence

Asynchronous cancellation has three terminal states: Success, Failure, and Interruption. Lexical finally does not handle interruption during async races. Leaked locks cause deadlocks; leaked connections cause server starvation.

const openImportFile = async (path: string) => ({ path });
const importIssues = async (_file: { path: string }) => undefined;
const closeImportFile = async (_file: { path: string }) => undefined;

const file = await openImportFile("issues.ndjson");
try {
  await importIssues(file);
} finally {
  await closeImportFile(file);
}

The mental model

Scope tracks an atomic, LIFO stack of release finalizers

Scope is an explicit runtime capability. Effect.acquireRelease(acquire, release) is atomic: if acquire succeeds, the runtime guarantees that release will be registered to the active Scope before any other code can run. When the scope closes, finalizers run in reverse order (LIFO) across all exit conditions: Success, Failure, Defect, and Interruption. Effect.scoped defines the scope boundary.

Scope Opened
  │
  ├─ Effect.acquireRelease ───────────> "opened issue-import" registered
  │
  ├─ Import logic executes
  │    └─ Yields Effect.fail(ImportFailed)  <-- FAILS!
  │
Scope Closes ─────────────────────────> Runtime runs the finalizer:
                                        "closed issue-import" executes

Minimal working example

Executable source

This is the checked source file used by the course. Read it before you run it.

import { Console, Effect } from "effect"

const events: Array<string> = []

const openImportSession = Effect.acquireRelease(
  Effect.sync(() => {
    events.push("opened issue-import")
    return "issue-import"
  }),
  (session) =>
    Effect.sync(() => {
      events.push(`closed ${session}`)
    })
)

export const processIssueImport = Effect.scoped(
  Effect.gen(function* () {
    const session = yield* openImportSession
    events.push(`processing issues in ${session}`)
    return yield* Effect.fail({ _tag: "ImportFailed" as const })
  })
)

await Effect.runPromise(
  Effect.gen(function* () {
    yield* processIssueImport.pipe(Effect.exit)
    yield* Console.log(events.join(" -> "))
  })
)

Execute and verify

Predict the result, run the command, and compare the output with the model.

Run this command

pnpm exec tsx src/course/examples/scope.ts

Before you run it

The import fails midway. Why does closed issue-import still execute?

Reveal expected output

opened issue-import -> processing issues in issue-import -> closed issue-import

Error anatomy and edge cases

Leaking the Scope requirement to the execution boundary

Acquiring a scoped resource or calling Effect.addFinalizer without establishing an enclosing Scope lifetime boundary.

Compiler or runtime diagnostic

The Effect type contains Scope.Scope in its R channel: Effect<T, E, Scope.Scope>. Calling Effect.runPromise triggers: "TS2345: Argument of type 'Effect<T, E, Scope.Scope>' is not assignable to parameter of type 'Effect<T, E, never>'. Type 'Scope' is not assignable to type 'never'".

Remedy

Wrap the resource-consuming pipeline in Effect.scoped(...) and keep the scope around the complete operation lifetime.

Active lab exercise

src/exercises/exercise5/starter.ts

Exercise: extend cleanup with a report finalizer

Extension: register a finalizer so buffered import entries are written after success and failure.

ObjectiveFix starter.ts with Effect.addFinalizer so pending issue-import entries are flushed and cleared.

pnpm exec tsx src/exercises/exercise5/exercise.test.ts starter

The harness should fail against the starter. Inspect the assertion, change the starter implementation, and run it again.

Inspect starter code

Broken starter implementation

import { Context, Effect, Layer, Ref } from "effect"

export interface ImportReportSink {
  readonly writeBatch: (
    entries: ReadonlyArray<string>
  ) => Effect.Effect<void>
}

export const ImportReportSink = Context.Service<ImportReportSink>(
  "ImportReportSink"
)

export interface IssueImportSession {
  readonly record: (message: string) => Effect.Effect<void>
}

export const IssueImportSession = Context.Service<IssueImportSession>(
  "IssueImportSession"
)

export const IssueImportSessionLive = Layer.effect(
  IssueImportSession,
  Effect.gen(function* () {
    const sink = yield* ImportReportSink
    const bufferRef = yield* Ref.make<ReadonlyArray<string>>([])

    const record = (message: string) =>
      Ref.update(bufferRef, (entries) => [...entries, message])

    // TODO: register a finalizer that flushes the import report to the sink.
    // The starter loses entries when the scope closes.
    void sink

    return { record }
  })
)

export const performIssueImport = (
  batchId: string
): Effect.Effect<string, never, IssueImportSession> =>
  Effect.gen(function* () {
    const session = yield* IssueImportSession
    yield* session.record(`Import started: ${batchId}`)
    yield* session.record(`Import finished: ${batchId}`)
    return `Imported: ${batchId}`
  })

export const performFailingIssueImport = (
  batchId: string
): Effect.Effect<
  never,
  { readonly _tag: "ImportFailed"; readonly message: string },
  IssueImportSession
> =>
  Effect.gen(function* () {
    const session = yield* IssueImportSession
    yield* session.record(`Import started: ${batchId}`)
    return yield* Effect.fail({
      _tag: "ImportFailed" as const,
      message: `Import failed: ${batchId}`
    })
  })

Reveal reference solution

Target reference solution

import { Context, Effect, Layer, Ref } from "effect"

export interface ImportReportSink {
  readonly writeBatch: (
    entries: ReadonlyArray<string>
  ) => Effect.Effect<void>
}

export const ImportReportSink = Context.Service<ImportReportSink>(
  "ImportReportSink"
)

export interface IssueImportSession {
  readonly record: (message: string) => Effect.Effect<void>
}

export const IssueImportSession = Context.Service<IssueImportSession>(
  "IssueImportSession"
)

export const IssueImportSessionLive = Layer.effect(
  IssueImportSession,
  Effect.gen(function* () {
    const sink = yield* ImportReportSink
    const bufferRef = yield* Ref.make<ReadonlyArray<string>>([])

    const record = (message: string) =>
      Ref.update(bufferRef, (entries) => [...entries, message])

    const flush = Effect.gen(function* () {
      const pending = yield* Ref.getAndSet(bufferRef, [])
      if (pending.length > 0) {
        yield* sink.writeBatch(pending)
      }
    })

    yield* Effect.addFinalizer(() => flush)

    return { record }
  })
)

export const performIssueImport = (
  batchId: string
): Effect.Effect<string, never, IssueImportSession> =>
  Effect.gen(function* () {
    const session = yield* IssueImportSession
    yield* session.record(`Import started: ${batchId}`)
    yield* session.record(`Import finished: ${batchId}`)
    return `Imported: ${batchId}`
  })

type ImportFailed = {
  readonly _tag: "ImportFailed"
  readonly message: string
}

const ImportFailed = (batchId: string): ImportFailed => ({
  _tag: "ImportFailed",
  message: `Import failed: ${batchId}`
})

export const performFailingIssueImport = (
  batchId: string
): Effect.Effect<never, ImportFailed, IssueImportSession> =>
  Effect.gen(function* () {
    const session = yield* IssueImportSession
    yield* session.record(`Import started: ${batchId}`)
    return yield* Effect.fail(ImportFailed(batchId))
  })

Why this solution works

Define a flush effect that atomically grabs pending entries with Ref.getAndSet(bufferRef, []) and calls sink.writeBatch(pending). Register it with yield* Effect.addFinalizer(() => flush) inside the scoped session layer.