Server

The server export turns a Sirannon instance into a network service. Applications call it over HTTP, and WebSocket connections add real-time change subscriptions and live queries. This page builds one file, server.ts, step by step. Start a minimal loopback server, give it the operations callers may run, identify each caller, and then set the limits that shape how it accepts connections.

Sirannon writes the database file into a directory you create yourself, so make one before you run any of the code below.

mkdir -p data

Start a server

Open a database on the Sirannon instance, pass that instance to createServer, and call listen. Start server.ts with the driver, the instance, an open database, and a server on the default port.

import { Sirannon } from '@delali/sirannon-db'
import { betterSqlite3 } from '@delali/sirannon-db/driver/better-sqlite3'
import { createServer } from '@delali/sirannon-db/server'
 
const driver = betterSqlite3()
const sirannon = new Sirannon({ driver })
await sirannon.open('orders', './data/orders.db')
 
const server = createServer(sirannon)
await server.listen()

With no options, the server binds 127.0.0.1 on port 9876. That default keeps it on the loopback interface, which puts it out of reach of everything outside the host until you decide to expose it. The server caps HTTP request bodies and WebSocket messages at 1 MB by default, which bounds the memory one request can claim. maxBodyBytes raises or lowers that single limit. A remote error returns a machine-readable code, and the server strips the stack trace and the internal details out of it.

This server also accepts no SQL from the network. acceptSql defaults to false, and the five statement routes and their WebSocket messages therefore answer SQL_NOT_ACCEPTED until you turn them on. The next step gives the server the reads and writes it may run.

Register what callers may run

The default path is registered operations, where server-side code names each statement the server may run and a caller invokes it by name with arguments. Add a registry keyed by database identifier, with one read and one write for the orders database.

import { Sirannon } from '@delali/sirannon-db'
import { betterSqlite3 } from '@delali/sirannon-db/driver/better-sqlite3'
import { createServer } from '@delali/sirannon-db/server'
 
const driver = betterSqlite3()
const sirannon = new Sirannon({ driver })
await sirannon.open('orders', './data/orders.db')
 
const server = createServer(sirannon) 
const server = createServer(sirannon, {
  operations: {
    orders: {
      reads: {
        ordersByStatus: {
          args: ['status'],
          columns: ['id', 'total', 'status'],
          statement: ({ status }) => ({
            sql: 'SELECT id, total, status FROM orders WHERE status = ? ORDER BY id',
            params: [status],
          }),
        },
      },
      writes: {
        placeOrder: {
          args: ['total'],
          statements: ({ total }) => [
            { sql: 'INSERT INTO orders (total, status) VALUES (?, ?)', params: [total, 'pending'] },
          ],
        },
      },
    },
  },
})
await server.listen()

A caller now reaches two data routes, POST /db/orders/query/ordersByStatus and POST /db/orders/execute/placeOrder, and every other route that touches data stays shut. The server holds every statement, so the tables and columns it never registered stay out of reach. The registered operations guide covers argument declaration, filling an argument from the caller's identity, the WebSocket form, and the error codes each refusal uses.

Choose a port and identify the caller

createServer(sirannon, options) accepts a port, a cors origin, a maxBodyBytes cap, a maxWebSocketBackpressureBytes bound, the operations registry above, and an authenticate hook that runs before every database route and every WebSocket upgrade. The hook returns the caller's identity, and registered operations read that identity through fromIdentity. Throw a RequestDeniedError to refuse the request with a status and code of your own. Health and capability endpoints skip the hook.

Beyond those, the server takes eight more options.

• host sets the bind address.
• acceptSql opens the statement routes.
• acceptBackupRestore opens the route that rebuilds a database from its backups.
• cdcRetentionMs sets how long change history stays available to replay.
• deviceCursorRetentionMs sets how long an idle device's cursor lasts before that device has to resync from a snapshot, and it defaults to 30 days.
• maxUnacknowledgedChanges sets the device sync delivery window.
• authorizeClusterStatus gates GET /db/:id/cluster.
• resolveExecutionTarget, getReplicationStatus, and getClusterStatus are the hooks distributed replication uses.

Set an explicit port and turn away requests that present no valid bearer token.

import { Sirannon } from '@delali/sirannon-db'
import { betterSqlite3 } from '@delali/sirannon-db/driver/better-sqlite3'
import { createServer } from '@delali/sirannon-db/server'
import { RequestDeniedError, Sirannon } from '@delali/sirannon-db'
import { betterSqlite3 } from '@delali/sirannon-db/driver/better-sqlite3'
import { createServer } from '@delali/sirannon-db/server'
 
interface Identity {
  userId: string
}
 
const SESSIONS = new Map<string, Identity>([['Bearer local-dev-token', { userId: 'u_5f31' }]])
 
const verifyBearerToken = (header: string | undefined): Identity | undefined =>
  header === undefined ? undefined : SESSIONS.get(header)
 
const driver = betterSqlite3()
const sirannon = new Sirannon({ driver })
await sirannon.open('orders', './data/orders.db')
 
const server = createServer(sirannon, { 
const server = createServer<Identity>(sirannon, {
  port: 9876,
  operations: {
    orders: {
      reads: {
        ordersByStatus: {
          args: ['status'],
          columns: ['id', 'total', 'status'],
          statement: ({ status }) => ({
            sql: 'SELECT id, total, status FROM orders WHERE status = ? ORDER BY id',
            params: [status],
          }),
        },
      },
      writes: {
        placeOrder: {
          args: ['total'],
          statements: ({ total }) => [
            { sql: 'INSERT INTO orders (total, status) VALUES (?, ?)', params: [total, 'pending'] },
          ],
        },
      },
    },
  },
  authenticate: ({ headers }) => {
    const identity = verifyBearerToken(headers.authorization)
    if (!identity) throw new RequestDeniedError(401, 'UNAUTHORIZED', 'Invalid or missing token')
    return identity
  },
})
await server.listen()

The SESSIONS map stands in for whatever turns a header into an identity in your application, such as a signed token you verify or a session row you read.

Every value the hook returns becomes the caller's identity, including an object shaped like a refusal. A hook that returns { status, code, message } fails the request with HOOK_ERROR, because the server reads that shape as a mistake. Refuse by throwing RequestDeniedError. The status, code, and message you pass then reach the caller unchanged. Any other error the hook throws produces a 500 with code HOOK_ERROR.

The hook reads headers.authorization the same way on an HTTP route and on a WebSocket upgrade, because a Node client attaches its headers to both. A browser attaches none to a socket and sends its credential in Sec-WebSocket-Protocol, which the security guide covers. No socket client can read the status of a refused handshake. A refusal you throw with status 401 or 403 therefore closes the connection with code 4401 or 4403, and the close reason states your code and message.

The security guide covers bearer headers, WebSocket credentials over Sec-WebSocket-Protocol, TLS termination, and what operations do with the identity once they read it. Read it before exposing the server beyond localhost.

Cap the request size

maxBodyBytes sets the largest HTTP request body and the largest WebSocket message the server accepts, in bytes. One value governs both transports. It must be a positive integer and defaults to 1_048_576 (1 MB). Set it to match the biggest write you expect, and keep it small enough that a single request cannot exhaust memory.

import { RequestDeniedError, Sirannon } from '@delali/sirannon-db'
import { betterSqlite3 } from '@delali/sirannon-db/driver/better-sqlite3'
import { createServer } from '@delali/sirannon-db/server'
 
interface Identity {
  userId: string
}
 
const SESSIONS = new Map<string, Identity>([['Bearer local-dev-token', { userId: 'u_5f31' }]])
 
const verifyBearerToken = (header: string | undefined): Identity | undefined =>
  header === undefined ? undefined : SESSIONS.get(header)
 
const driver = betterSqlite3()
const sirannon = new Sirannon({ driver })
await sirannon.open('orders', './data/orders.db')
 
const server = createServer<Identity>(sirannon, {
  port: 9876,
  maxBodyBytes: 4 * 1024 * 1024, 
  operations: {
    orders: {
      reads: {
        ordersByStatus: {
          args: ['status'],
          columns: ['id', 'total', 'status'],
          statement: ({ status }) => ({
            sql: 'SELECT id, total, status FROM orders WHERE status = ? ORDER BY id',
            params: [status],
          }),
        },
      },
      writes: {
        placeOrder: {
          args: ['total'],
          statements: ({ total }) => [
            { sql: 'INSERT INTO orders (total, status) VALUES (?, ?)', params: [total, 'pending'] },
          ],
        },
      },
    },
  },
  authenticate: ({ headers }) => {
    const identity = verifyBearerToken(headers.authorization)
    if (!identity) throw new RequestDeniedError(401, 'UNAUTHORIZED', 'Invalid or missing token')
    return identity
  },
})
await server.listen()

createServer throws a SirannonError with code INVALID_MAX_BODY_BYTES when the value is not a positive integer. Both this cap and the WebSocket buffer bound below must stay at or under 4_294_967_295 bytes, because uWebSockets.js stores each limit as an unsigned 32-bit integer and would silently wrap a larger value modulo 2^32. Once the server is running, it rejects a request or message over the cap with PAYLOAD_TOO_LARGE before it reads that body into memory. Split a bulk load that exceeds the cap into several sequential loads.

Bound the WebSocket buffer

maxBodyBytes caps what comes in; maxWebSocketBackpressureBytes caps what waits to go out. When a WebSocket client reads slower than the server sends, the replies, change events, and live-query updates queue in that connection's outbound buffer. This option caps how many bytes the server buffers per connection before it stops absorbing the lag. It defaults to the larger of 16 MB and maxBodyBytes. A single outbound frame can be as large as maxBodyBytes, and the server therefore raises the resolved value to at least that. An explicit value below it throws. Give a subscriber of large rows more headroom.

import { RequestDeniedError, Sirannon } from '@delali/sirannon-db'
import { betterSqlite3 } from '@delali/sirannon-db/driver/better-sqlite3'
import { createServer } from '@delali/sirannon-db/server'
 
interface Identity {
  userId: string
}
 
const SESSIONS = new Map<string, Identity>([['Bearer local-dev-token', { userId: 'u_5f31' }]])
 
const verifyBearerToken = (header: string | undefined): Identity | undefined =>
  header === undefined ? undefined : SESSIONS.get(header)
 
const driver = betterSqlite3()
const sirannon = new Sirannon({ driver })
await sirannon.open('orders', './data/orders.db')
 
const server = createServer<Identity>(sirannon, {
  port: 9876,
  maxBodyBytes: 4 * 1024 * 1024,
  maxWebSocketBackpressureBytes: 64 * 1024 * 1024, 
  operations: {
    orders: {
      reads: {
        ordersByStatus: {
          args: ['status'],
          columns: ['id', 'total', 'status'],
          statement: ({ status }) => ({
            sql: 'SELECT id, total, status FROM orders WHERE status = ? ORDER BY id',
            params: [status],
          }),
        },
      },
      writes: {
        placeOrder: {
          args: ['total'],
          statements: ({ total }) => [
            { sql: 'INSERT INTO orders (total, status) VALUES (?, ?)', params: [total, 'pending'] },
          ],
        },
      },
    },
  },
  authenticate: ({ headers }) => {
    const identity = verifyBearerToken(headers.authorization)
    if (!identity) throw new RequestDeniedError(401, 'UNAUTHORIZED', 'Invalid or missing token')
    return identity
  },
})
await server.listen()

createServer throws INVALID_WS_BACKPRESSURE when the value is not a positive integer, exceeds the 32-bit ceiling, or is smaller than maxBodyBytes. When a connection's buffer would exceed the bound, the server closes that connection with application close code 4290, echoing HTTP 429, which is how the client recognises an overload and reconnects. The client SDK does this on its own when autoReconnect is on, and it restores active subscriptions after the reconnect.

Four write shapes

The server accepts writes in four shapes over both transports. Reach for each one when the write in front of you matches it.

• Use a registered write for the ordinary case. It names statements the server already holds, and the caller therefore sends a name and arguments. The server then runs every statement of that write in one transaction and replies with one result per statement.
• Use a transaction to run several different statements that must all succeed or all fail together, such as a debit on one row and a credit on another. It sends statement text, and it therefore needs acceptSql: true.
• Use a batch to run one statement many times with different values, such as inserting a thousand rows into the same table. It does less work than a transaction of a thousand near-identical statements, and it stays all-or-nothing. It needs acceptSql: true.
• Use a load for a large, from-scratch import. It relaxes durability while the rows go in and restores it afterwards, which trades power-loss safety during the load for speed. Where the process dies mid-load, run the load again. It needs acceptSql: true, and the bulk load guide covers the durability levels and the summary it returns.

Turn SQL back on

Some callers are trusted service code that composes its own statements, and registering every shape would add work without adding safety. Set acceptSql: true for those. The five statement routes and their WebSocket messages then start serving.

import { RequestDeniedError, Sirannon } from '@delali/sirannon-db'
import { betterSqlite3 } from '@delali/sirannon-db/driver/better-sqlite3'
import { createServer } from '@delali/sirannon-db/server'
 
interface Identity {
  userId: string
}
 
const SESSIONS = new Map<string, Identity>([['Bearer local-dev-token', { userId: 'u_5f31' }]])
 
const verifyBearerToken = (header: string | undefined): Identity | undefined =>
  header === undefined ? undefined : SESSIONS.get(header)
 
const driver = betterSqlite3()
const sirannon = new Sirannon({ driver })
await sirannon.open('orders', './data/orders.db')
 
const server = createServer<Identity>(sirannon, {
  port: 9876,
  maxBodyBytes: 4 * 1024 * 1024,
  maxWebSocketBackpressureBytes: 64 * 1024 * 1024,
  acceptSql: true, 
  operations: {
    orders: {
      reads: {
        ordersByStatus: {
          args: ['status'],
          columns: ['id', 'total', 'status'],
          statement: ({ status }) => ({
            sql: 'SELECT id, total, status FROM orders WHERE status = ? ORDER BY id',
            params: [status],
          }),
        },
      },
      writes: {
        placeOrder: {
          args: ['total'],
          statements: ({ total }) => [
            { sql: 'INSERT INTO orders (total, status) VALUES (?, ?)', params: [total, 'pending'] },
          ],
        },
      },
    },
  },
  authenticate: ({ headers }) => {
    const identity = verifyBearerToken(headers.authorization)
    if (!identity) throw new RequestDeniedError(401, 'UNAUTHORIZED', 'Invalid or missing token')
    return identity
  },
})
await server.listen()

Such a server runs any statement a caller sends. Authenticate every request, and keep it off the public internet. Registered operations stay available either way, because acceptSql never governs them. GET /capabilities announces query.sql once you set it, which is what tells the client SDK whether to send a statement; the capabilities section covers that handshake.

Serve the backup routes

Seven routes put a database's backups on the network. Between them an operator can start a turn, read how far the cycle has got, list the chain the destination holds, verify one stored backup, name the pieces that are safe to delete, rebuild the database from a moment, and read how that rebuild is going. The server calls authenticate before each of them, as it does before every other /db/:id route.

Reserve them for an operator credential. Sirannon defines no separate hook for them, while RequestContext states method and path, which lets one authenticate hook admit your application on the data routes and refuse it here. A hook without such a check lets every identity it accepts call all seven, including the one that replaces the database. The file below builds such a hook, starts a server, and calls each route in turn from the same process.

import { type BackupDestination, type BackupPiece, RequestDeniedError, Sirannon } from '@delali/sirannon-db'
import { betterSqlite3 } from '@delali/sirannon-db/driver/better-sqlite3'
import { createServer } from '@delali/sirannon-db/server'
 
const stored = new Map<string, Uint8Array>()
const pieceKey = (name: string, index: number): string => `${name}#${index}`
 
const memory: BackupDestination = {
  async writePiece(name, index, bytes) {
    stored.set(pieceKey(name, index), Uint8Array.from(bytes))
  },
  async writePieceIfAbsent(name, index, bytes) {
    if (stored.has(pieceKey(name, index))) return false
    stored.set(pieceKey(name, index), Uint8Array.from(bytes))
    return true
  },
  async readPiece(name, index) {
    const bytes = stored.get(pieceKey(name, index))
    if (!bytes) throw new Error(`No piece ${index} of '${name}'`)
    return bytes
  },
  async listPieces(name) {
    const pieces: BackupPiece[] = []
    for (const [key, bytes] of stored) {
      if (key.startsWith(`${name}#`)) {
        pieces.push({ index: Number(key.slice(name.length + 1)), byteLength: bytes.byteLength })
      }
    }
    return pieces
  },
}
 
const OPERATOR_TOKEN = 'operator-token'
const APPLICATION_TOKEN = 'application-token'
 
const sirannon = new Sirannon({ driver: betterSqlite3() })
 
const db = await sirannon.open('orders', './data/orders.db', {
  backups: { destination: memory, intervalMs: 0, onError: (err) => console.error('Backup cycle failed:', err.message) },
})
 
await db.execute('CREATE TABLE IF NOT EXISTS orders (id INTEGER PRIMARY KEY, customer TEXT NOT NULL, total REAL NOT NULL)')
await db.execute('INSERT INTO orders (customer, total) VALUES (?, ?)', ['Amara Okonkwo', 128.5])
 
const server = createServer(sirannon, {
  port: 9876,
  acceptBackupRestore: true,
  authenticate: (ctx) => {
    const token = ctx.headers.authorization?.replace('Bearer ', '')
    if (token !== OPERATOR_TOKEN && token !== APPLICATION_TOKEN) {
      throw new RequestDeniedError(401, 'IDENTITY_REQUIRED', 'Present a bearer token')
    }
    const operator = token === OPERATOR_TOKEN
    if (ctx.path.includes('/backup') && !operator) {
      throw new RequestDeniedError(403, 'HOOK_DENIED', 'Only an operator may reach the backups')
    }
    return { operator }
  },
})
 
await server.listen()
 
const base = 'http://127.0.0.1:9876/db/orders'
const asOperator = { authorization: `Bearer ${OPERATOR_TOKEN}`, 'content-type': 'application/json' }
const asApplication = { authorization: `Bearer ${APPLICATION_TOKEN}`, 'content-type': 'application/json' }
 
const call = async (method: string, path: string, headers: Record<string, string>, body?: unknown) => {
  const response = await fetch(`${base}${path}`, {
    method,
    headers,
    ...(body === undefined ? {} : { body: JSON.stringify(body) }),
  })
  return { status: response.status, body: (await response.json()) as Record<string, unknown> }
}
 
const untilTheTurnFinishes = async () => {
  for (;;) {
    const answer = await call('GET', '/backup', asOperator)
    if (answer.body.running === false && answer.body.lastRun !== undefined) return answer
    await new Promise((resolve) => setTimeout(resolve, 25))
  }
}
 
const untilTheRestoreFinishes = async () => {
  for (;;) {
    const answer = await call('GET', '/backup/restore', asOperator)
    if (answer.body.state !== 'running') return answer
    await new Promise((resolve) => setTimeout(resolve, 25))
  }
}
 
const refused = await call('POST', '/backup', asApplication)
const triggered = await call('POST', '/backup', asOperator)
const status = await untilTheTurnFinishes()
const chain = await call('GET', '/backup/chain', asOperator)
 
const chains = chain.body.chains as { base?: { name: string } }[]
const verified = await call('POST', '/backup/verify', asOperator, { name: chains[0].base?.name })
const safeToDelete = await call('POST', '/backup/safe-to-delete', asOperator, {})
 
const restoreAccepted = await call('POST', '/backup/restore', asOperator, {})
const restored = await untilTheRestoreFinishes()
 
console.log(
  JSON.stringify(
    {
      refusedForTheApplication: { status: refused.status, code: (refused.body.error as { code: string }).code },
      triggered: { status: triggered.status, body: triggered.body },
      status: {
        status: status.status,
        running: status.body.running,
        lastRunKind: (status.body.lastRun as { kind: string })?.kind,
      },
      chain: { status: chain.status, chains: chains.length, hasFullCopy: chains[0].base !== undefined },
      verified: { status: verified.status, kind: verified.body.kind, pieceCount: verified.body.pieceCount },
      safeToDelete: { status: safeToDelete.status, records: (safeToDelete.body.records as unknown[]).length },
      restoreAccepted: { status: restoreAccepted.status, body: restoreAccepted.body },
      restored: {
        status: restored.status,
        state: restored.body.state,
        changesApplied: (restored.body.report as { changesApplied: number })?.changesApplied,
      },
    },
    null,
    2,
  ),
)
 
await server.close()
await sirannon.shutdown()

Both write routes answer 202 Accepted at once, because a full copy of a large database can continue past the deadline any proxy between the caller and the server allows. The matching GET reports the outcome, which is why the file above polls it. A turn triggered while another is under way queues behind it, and every trigger after that joins the queued turn, which leaves at most one turn waiting.

Four statuses tell the failures apart. A database that is not open answers 404 DATABASE_NOT_FOUND. A database opened without the backups option answers 501 BACKUP_UNSUPPORTED on every one of these routes but GET /db/:id/backup/restore, which reads the server's own record and touches no database. A chain that cannot reach the moment asked for answers 409 BACKUP_CHAIN_BROKEN. A destination that refuses a call or fails a check answers 502 BACKUP_DESTINATION_ERROR, which separates your storage failing from Sirannon failing.

Restore a database over the network

POST /db/:id/backup/restore stays shut until you set acceptBackupRestore: true, because a restore replaces the database that is serving your traffic and no server should open such a route by default. While it stays false, the server answers 403 BACKUP_RESTORE_NOT_ACCEPTED before it looks the database up, which is why a server that never opened the route answers 403 for a database opened without backups as well.

That option also makes the authenticate hook compulsory. A server built with acceptBackupRestore: true and no hook refuses to start with INVALID_BACKUP_RESTORE, because that hook is the only check in front of a route that replaces a database, and every request would otherwise reach it anonymously.

The body takes an optional moment, in epoch milliseconds, and an optional batchSize, which defaults to 16. Leave the moment out for the newest backup the destination holds. A missing name on the verify route, a moment or restorableFrom that is negative or fractional, a batchSize outside one to 4,096, and a body that parses as JSON but is not an object each answer 400 INVALID_REQUEST.

Sirannon rebuilds the database at the path it already occupies, in this order:

1. It closes the database, and that close captures its log a final time.
2. It discards the cycle state and the staged captures of the chain that database was extending.
3. It rebuilds the file from that database's own backups, applying the destination deadline the database was opened with.
4. It opens the database again under the same identifier, with the settings it had.

Every route answers 404 DATABASE_NOT_FOUND for that identifier while the rebuild proceeds, which is why the status route reads the server's own record. The first turn of the cycle after the reopen copies the whole database and starts a fresh chain, since the rebuilt file's log continues none of the old one. Discarding the cycle state first is why the next turn copies the whole database, and that order is what keeps the rebuild safe. A process that died between a finished rebuild and a later cleanup would otherwise resume capturing onto the chain the restore had replaced, appending a piece cut from one timeline to a chain built from another.

A second restore of the same database while one is under way answers 409 BACKUP_RESTORE_IN_PROGRESS. A rebuild that fails still opens the database again, and the status route states the code it stopped with. A close that fails leaves nothing open under that identifier, because a second runtime over a file the old connections may still hold would put two writers on one database. A reopen that fails after a rebuild succeeded reports done with the report and a separate reopenError, since Sirannon replaced the data either way and only the process needs restarting.

HTTP routes

The server reads :id and :name as URL-encoded values. The five statement routes need acceptSql: true and answer SQL_NOT_ACCEPTED without it, and the server always serves the rest.

A read body takes an optional readConcern, and a write body takes an optional writeConcern; distributed replication defines both. A path matching no route answers NOT_FOUND, which lets a caller tell a refused capability from a wrong address.

WebSocket protocol

Connect to ws://host:port/db/:id and send JSON messages. Every message states a type and a client-chosen id, and every reply echoes that id. The server sends sequence numbers as decimal strings, and a value beyond the safe integer range therefore keeps every digit through JSON.

The server supports one subprotocol, the plain identifier sirannon.v1. An upgrade that offers no subprotocol connects, which is what a hand-written client with its credential in a header does. An upgrade that offers any must include sirannon.v1, because the server selects that one and answers 400 UNSUPPORTED_SUBPROTOCOL to anything else. Selecting the plain identifier keeps a browser's credential out of the handshake response.

A query or an execute message that supplies name runs the registered operation of that name and sends no SQL, which leaves acceptSql with nothing to govern. Without acceptSql, a query, execute, transaction, batch, or load message that sends statement text answers SQL_NOT_ACCEPTED.

Subscribe with a single table, or with a tables array of up to 500 names to receive them in one ascending stream. A tables subscription requires a deviceId. sinceSeq and epoch resume a feed after a disconnect, which subscription resumption covers in full. The deviceId, schemaVersion, stagedStream, and ack fields drive device sync. Acknowledgements pace a subscription that presents a deviceId. The server holds delivery once that device runs more than maxUnacknowledgedChanges past its acknowledged sequence, and it resumes on the next ack. A subscription naming a registered read opens a live query, and the server answers it with the rows themselves.

Beyond the subscribed and unsubscribed replies above, the server sends five kinds of message back.

• A change delivers one change event.
• A changes delivers several in ascending sequence order, and it reaches only a subscription that asked for stagedStream.
• A live delivers the ops, rows, or revalidating update of a live query.
• A result answers a query, execute, transaction, batch, load, or ack.
• An error reports { code, message }.

The transaction, batch, and load messages run every statement server-side in one transaction and reply once, as a registered write does. The server never holds the write lock across a network round-trip, which is why it accepts no interactive transaction where the client sends BEGIN, then more statements, then COMMIT over separate messages. A single slow or dead client would otherwise freeze every write to the database.