Engine

open/openSafe and the Engine class, covering reads, writes, transactions, declared boards, watch, and operational helpers.

import { open, openSafe, Engine } from "lakefront";

One DuckDB instance holding the lake plus every attached board replica, fronted by a connection pool and an LRU of attached files. Every fallible method returns a Result from better-result.

Functions

open

function open(options: OpenOptions): Promise<Engine>;

Returns the opened Engine.

Attaches the lake, starts the WAL and flusher (under writePath: "wal"), and starts the head-cache poller. open throws LakeUnavailable on a boot failure; use openSafe to receive that failure as a Result. It creates the cache directory and, for a local (non-s3://) data, the data directory itself, so there is nothing to mkdir before calling it.

await using engine = await open({
  postgres: "postgres://lakefront:lakefront@127.0.0.1:5432/lakefront",
  data: "./data",
});

Two required fields. Everything else (where the cache lives, how many rows stay inlined, how the cluster is shaped) has a default. Engine implements AsyncDisposable, so await using engine = await open(...) closes it (connection pool, WAL, flusher) when the scope ends; close() is idempotent.

openSafe

function openSafe(options: OpenOptions): Promise<Result<Engine, LakeUnavailable>>;

Returns a boot failure as a Result instead of throwing.

resetLake

function resetLake(options: Pick<OpenOptions, "postgres" | "data" | "namespace">): Promise<void>;

Removes both halves of a lake: drops the DuckLake catalog schema and deletes the data directory.

dropCatalogSchema

function dropCatalogSchema(postgres: string, namespace: string): Promise<void>;

Drops the DuckLake catalog schema alone, without deleting the data directory. Safe to call when the schema does not exist. resetLake calls this and then removes data too.

toKeywordDsn

function toKeywordDsn(input: string): string;

Converts a standard Postgres URL or the keyword/value form into the keyword/value DSN DuckLake’s ATTACH accepts. What open() does internally to postgres before use.

pgOptionsFromDsn

function pgOptionsFromDsn(input: string): {
  hostname: string;
  port: number;
  database: string;
  username: string;
  password: string;
  max: number;
};

Converts either input form into the options shape Bun’s SQL client accepts.

messageOf

function messageOf(cause: unknown): string;

Normalizes a thrown value into a message without swallowing its type: cause instanceof Error ? cause.message : String(cause). Also exported from lakefront/contract, where it is defined; re-exported here so engine code keeps one import path.

Engine

Instances come from open/openSafe; there is no public constructor. Engine implements AsyncDisposable.

Reads

query

query(
  boardId: string,
  sqlTemplate: string,
  options?: QueryOptions,
): Promise<Result<QueryOutcome, EngineError>>

Returns Result<QueryOutcome, EngineError>.

Sync-then-query: the board’s replica catches up to the head (or to minWatermark, if given) before the SQL runs against it. sync runs that catch-up alone, without a query. See the consistency guide for the watermark model and read-your-write semantics.

sync

sync(boardId: string): Promise<Result<SyncResult, EngineError>>

Runs the sync-then-query catch-up for boardId without executing a query. Returns the SyncResult it produced.

watermarkOf

watermarkOf(boardId: string): Promise<Result<number, EngineError>>

The board’s replica watermark, without a sync.

rowCount

rowCount(boardId: string): Promise<Result<number, EngineError>>

Row count on the board’s attached replica.

head

head(): number

The cached head, with no round-trip. headFromCatalog() pays for the authoritative one.

headFromCatalog

headFromCatalog(): Promise<Result<number, EngineError>>

The authoritative head, read from the catalog. One round-trip; head() is the cached answer most callers want.

withReplica

withReplica<T>(
  boardId: string,
  fn: (ctx: { conn: DuckDBConnection; replica: LocalReplica; lake: Lake; head: HeadCache }) => Promise<T>,
): Promise<Result<T, EngineError>>

Low-level access to an attached replica on a pooled connection. Attaching and busy-tracking are handled; sync is not, so the caller decides.

Writes

All writes return Result<number, EngineError>, where the number is the watermark the write committed at: hand it back as minWatermark on a subsequent query to read your own write.

insert

insert(boardId: string, row: Row, options?: WriteOptions): Promise<Result<number, EngineError>>

Inserts one row. Sugar over insertMany.

insertMany

insertMany(
  boardId: string,
  rows: readonly Row[],
  options?: WriteOptions,
): Promise<Result<number, EngineError>>

Inserts rows as one write.

upsert

upsert(boardId: string, row: Row, options?: WriteOptions): Promise<Result<number, EngineError>>

Upserts one complete row. Sugar over upsertMany.

upsertMany

upsertMany(
  boardId: string,
  rows: readonly Row[],
  options?: WriteOptions,
): Promise<Result<number, EngineError>>

Append-based upsert: writes a new version of each row instead of issuing an UPDATE. An UPDATE on a DuckLake table costs about 2.5x an INSERT, because it has to locate the existing rows and mark them deleted first; appending skips that. Rows must be complete, and superseded versions accumulate until compact() reclaims them. The winner is chosen by version stamp, so across nodes it is last-writer-wins to whatever precision the clocks agree on, which is why update/updateMany remain the default and this is opt-in.

update

update(
  boardId: string,
  id: string,
  patch: Row,
  options?: WriteOptions,
): Promise<Result<number, EngineError>>

Patches one row by id.

updateMany

updateMany(
  boardId: string,
  updates: ReadonlyArray<{ readonly id: string; readonly patch: Row }>,
  options?: WriteOptions,
): Promise<Result<number, EngineError>>

Applies many single-row patches as one atomic write, the batch shape OLTP loops actually produce, without paying a lake commit or a WAL append per row. Entries apply in array order, so two patches to the same id stack exactly as two update() calls would.

delete

delete(boardId: string, id: string, options?: WriteOptions): Promise<Result<number, EngineError>>

Deletes one row by id. Sugar over deleteMany.

deleteMany

deleteMany(
  boardId: string,
  ids: readonly string[],
  options?: WriteOptions,
): Promise<Result<number, EngineError>>

Deletes rows by id.

compact

compact(boardId: string): Promise<Result<number, EngineError>>

Collapses superseded versions left behind by upsert.

Transactions

transaction

transaction<T>(
  boards: readonly string[],
  fn: (tx: TransactionOps) => Promise<T>,
  options?: WriteOptions,
): Promise<Result<{ value: T; watermark: number }, EngineError>>

Runs several writes as one atomic commit spanning every board named in boards. The whole block runs on one connection (or as one WAL unit, under writePath: "wal"), and the returned watermark is the read-your-write watermark for everything the block wrote.

const r = await engine.transaction(["accounts", "history"], async (tx) => {
  await tx.update("accounts", id, { balance: next });
  await tx.insert("history", { id: crypto.randomUUID(), delta });
});

Lifecycle

createBoard

createBoard(boardId: string, options?: CreateBoardOptions): Promise<Result<number, EngineError>>

Fails with BoardExists if the board is already there. It is the explicit counterpart to write-time auto-create, and the only place a board’s columns can be declared rather than inferred:

await engine.createBoard("billing", {
  columns: { amount: "double", currency: "text", paid: "boolean", due: "timestamp" },
  strict: true,
});

Base columns (id, updated_at, state) exist on every board already and cannot be redeclared, neither can a column under the reserved _ prefix. Declaring a name that collides with either fails with ReservedColumnName.

dropBoard

dropBoard(boardId: string): Promise<Result<number, EngineError>>

Removes the board and every row, everywhere. Other nodes need no message: the drop advances the snapshot log, so their next sync discovers the table gone.

truncateBoard

truncateBoard(boardId: string): Promise<Result<number, EngineError>>

Empties a board, keeping it and its columns.

boards

boards(): Promise<Result<string[], EngineError>>

Every board in the lake, whether or not this node has it cached.

hasBoard

hasBoard(boardId: string): Promise<Result<boolean, EngineError>>

Whether boardId exists in the lake.

addColumn

addColumn(boardId: string, column: string, type: ColumnType): Promise<Result<number, EngineError>>

Declares a column on a board, without waiting for data to imply it. Required under columns: "declared" (engine-wide) or a board created with strict: true, where inference is off. Useful under inference too: it fixes a column’s type up front instead of letting the first value decide.

widenColumn

widenColumn(boardId: string, column: string, to: WidenTarget): Promise<Result<number, EngineError>>

Widens one column’s type, deliberately and irreversibly. Column types are inferred once, when a write first mentions the column, and fixed after that: a later value that does not fit is refused with ColumnTypeConflict rather than silently reshaping the board. This is the escape hatch, and it only ever widens: a request that would narrow a column is refused, so nothing already written can stop fitting.

await engine.widenColumn("acme_roadmap", "priority", "VARCHAR");

Feed

watch

watch(
  boardId: string,
  options?: { after?: number; signal?: AbortSignal; pollMs?: number },
): AsyncGenerator<ChangesPage, void, undefined>

Follows a board’s change feed as an async iterable: catch-up from after, then live.

for await (const event of engine.watch("tasks", { after })) {
  if (event.resync) {
    // re-read current state, then continue -- the iterator already
    // moved its cursor to event.from
  } else {
    // event.records
  }
}

Each yielded event is a non-empty ChangesPage. Local writes wake the loop immediately; writes landing on other nodes are picked up by the poll fallback, so the feed is complete either way, and the wakeup only sharpens latency. resync: true is yielded as an ordinary value and the iterator continues from the new floor; it is not an error. A read failure is thrown (the same EngineError changes() would return as a Result). Design, retention, and what resync means for a consumer are covered in the watch guide.

changes

changes(boardId: string, options: { after: number; limit?: number }): Promise<Result<ChangesPage, EngineError>>

One page of the change feed for a board above a watermark this engine handed out. watch is a loop built on top of it, and is the surface most callers want directly.

onChange

onChange(listener: (event: ChangeEvent) => void): () => void

Registers a local trigger, called once per touched board after each of this node’s own write acks; fires on a microtask, so a slow or throwing listener can neither delay nor fail the write. It sees only this node’s own writes (that local signal is what wakes watch between polls) and is not itself a substitute for watch or changes. Returns the unsubscribe function.

Operations

maintain

maintain(olderThanHours?: number): Promise<Result<void, EngineError>>

Runs lake maintenance now (snapshot expiry, file cleanup, small-file merge); under writePath: "wal" the flusher also runs it hourly.

flush

flush(boardId?: string): Promise<Result<FlushStats[], EngineError>>

Drains the WAL into the lake immediately: one board, or every board with pending records when boardId is omitted. A no-op under writePath: "lake".

evict

evict(boardId: string, deleteFile?: boolean): Promise<void>

Drops a replica from the cache. The file stays on disk unless deleteFile is set, which simulates losing the node’s storage entirely.

stats

stats(): EngineStats

Returns the EngineStats snapshot.

cachedBoards

cachedBoards(): readonly string[]

Boards currently attached in this node’s replica cache.

close

close(): Promise<void>

Closes the connection pool, WAL, and flusher. Idempotent: a second call returns immediately rather than deadlocking on connections the first close already released. await using engine = await open(...) calls this at scope exit via Symbol.asyncDispose.

Types

OpenOptions

What open() and openSafe() take: two required fields (where the catalog Postgres is, where the data lives) and defaults for everything else.

A few of these carry rationale too long for a cell:

• topology: "clustered" cross-board reads are stale by at most one interval, bounded by the head-refresh poll. "single-node" trades that staleness for exact cross-board reads. Running two "single-node" engines against one lake produces incorrect reads.
• columns: "infer" is what makes a board schemaless. A single board can opt into declared-only columns on its own via createBoard(id, { strict: true }), independent of this engine-wide setting.
• writePath: "lake" is simpler, but caps throughput at double-digit tps per node. "wal" trades that ceiling for the durability contract described under write paths.

S3Config

QueryOptions

QueryOutcome

SyncResult

WriteOptions

TransactionOps

The handle transaction’s body runs against. Methods here throw instead of returning Result: a failure aborts the whole transaction rather than being handled statement by statement, and Engine.transaction converts the throw back into the outer Result at the boundary.

CreateBoardOptions

ColumnType

The public column-type vocabulary uses plain lowercase names, mapped onto the DuckDB type underneath:

COLUMN_TYPES is the object above, exported as the runtime source of truth; ColumnType is keyof typeof COLUMN_TYPES. See the schema guide for how inference and widening behave in practice.

WidenTarget

Exclude<ColumnType, "timestamp">: every widenColumn destination except "timestamp", since that type sits outside the lattice a column’s inferred type moves up.

EngineStats

ChangeEvent

{ boardId: string, watermark: number, ops: readonly WalRecordOp[] }. Fired to onChange listeners after a write’s ack, once per touched board.

WalRecordOp

{ kind: WalKind, payload: WalPayload }. A change record’s operational core.

ChangesPage

Discriminated on resync: { resync: false, records: ChangeRecord[], next: number } | { resync: true, from: number }. resync: true means the requested range is beyond retention; re-read current state and continue from from.

FlushStats

interface FlushStats {
  readonly boardId: string;
  readonly from: number;
  readonly to: number;
  readonly records: number;
  readonly skipped: boolean;
}

What one flush() call did for one board.

ParkedBoard

interface ParkedBoard {
  readonly boardId: string;
  readonly failures: number;
  readonly since: number;
  readonly retryAt: number;
  readonly message: string;
}

A board the periodic flusher has given up on for now: its WAL records are untouched (durability never depended on flushing), but they accumulate until the cause is repaired. stats() itself returns the narrower wire shape ParkedBoardStat ({ boardId, failures, since }); ParkedBoard is the richer local shape, with the retry backoff and message, that the flusher tracks internally.

WalKind

"insert" | "upsert" | "update" | "delete" | "drop".

WalOp

{ boardId: string, kind: WalKind, payload: WalPayload }.

WalPayload

A union keyed by shape: { rows: Row[] } (insert/upsert), { id: string, patch: Row } (update), { ids: string[] } (delete), or { drop: true } (a whole-board tombstone).

Row

Readonly<Record<string, Cell>>, defined once in lakefront/contract and re-exported here.

Cell

string | number | boolean | null, defined once in lakefront/contract and re-exported here.

QueryResult

{ columns: string[], rows: Cell[][] }, defined once in lakefront/contract and re-exported here.

EngineError

The union of every tagged error class the engine can raise: InvalidBoardId, EntityNotAllowed, PlaceholderUnbound, LakeUnavailable, QueryFailed, SyncFailed, ReplicaUnavailable, WriteFailed, ReservedColumnName, ColumnTypeConflict, BoardNotFound, BoardExists, UnknownColumn, each also exported by name. Full catalog, fields, and when each is raised: the error catalog.

Constants

COLUMN_TYPES

The runtime source of truth for the ColumnType table above: { boolean: "BOOLEAN", tinyint: "TINYINT", ..., text: "VARCHAR", timestamp: "TIMESTAMP" }.

BASE_COLUMNS

BASE_COLUMNS = { id: "VARCHAR", updated_at: "TIMESTAMP", state: "VARCHAR" };

The fixed columns every board table has; their types are fixed by the system and never inferred from data.

RESERVED_COLUMN_PREFIX

RESERVED_COLUMN_PREFIX = "_";

The namespace user columns may not write into: a column named _v decides upsert ordering, compaction, and version-scoped deletes.

ENGINE_ERRORS

The lookup object every derivation in the system uses: keyed by tag, values are the error classes themselves. WIRE_TAGS, the engine’s own error-wrapping, and the reference catalog are all built from it, so a class added here is everywhere at once. See the error catalog.