# Files, streams & processes

> Use Node's core modules on the server: promise-based files and paths, streams and pipeline for large data, events, cancellation, and child processes.

Backend with Node.js roadmap · Foundations · 17 min read · https://levelcode.org/roadmaps/node-backend/node-core-apis

Web frameworks hide a lot, but sooner or later every backend reads an uploaded file, writes an export, streams a large response, or runs another program. Node's standard library has solid tools for all of it, and they share a few ideas: promises for one-off operations, **streams** for data too large to hold in memory, `AbortSignal` for cancellation. This lesson covers the core modules you'll use most, with a focus on streams, which are what let a Node server handle gigabytes with a few megabytes of memory.

## Files and paths

Use the promise-based API from `node:fs/promises` in server code, so the event loop stays free while the disk works:

```ts
import { mkdir, readFile, readdir, rm, stat, writeFile } from 'node:fs/promises'
import path from 'node:path'

await mkdir('exports/2026', { recursive: true })
await writeFile('exports/2026/report.json', JSON.stringify({ total: 42 }, null, 2))

const text = await readFile('exports/2026/report.json', 'utf8')
const info = await stat('exports/2026/report.json')
console.log(info.size, info.mtime)

for (const entry of await readdir('exports', { withFileTypes: true, recursive: true })) {
  if (entry.isFile()) console.log(path.join(entry.parentPath, entry.name))
}

await rm('exports', { recursive: true, force: true })
```

Handle errors by their `code`:

```ts
try {
  return JSON.parse(await readFile(file, 'utf8'))
} catch (err) {
  if ((err as NodeJS.ErrnoException).code === 'ENOENT') return {} // file doesn't exist yet
  throw err
}
```

Build paths with `node:path` instead of string concatenation, so separators are right on every OS. Relative paths resolve from the **current working directory** of the process, which isn't necessarily where your code lives. For files next to your module, use `import.meta.dirname` (Node 20.11+):

```ts
const templatesDir = path.join(import.meta.dirname, 'templates')
```

> [!WARNING]
> Never build a file path directly from user input. `path.join('uploads', userInput)` with `userInput = '../../etc/passwd'` escapes your folder. Generate your own file names (a random ID), or resolve the path and check it still starts with the directory you intended.

## Why streams

`readFile` loads the whole file into memory. That's fine for a config file and a disaster for a 5 GB log: the process needs 5 GB of RAM, and nothing happens until the last byte arrives.

A **stream** processes data in chunks (typically 64 KB) as it flows. Node has four kinds:

| Kind | Produces or consumes | Examples |
|---|---|---|
| Readable | Produces chunks | `fs.createReadStream`, an incoming HTTP request body, `process.stdin` |
| Writable | Consumes chunks | `fs.createWriteStream`, an HTTP response, `process.stdout` |
| Duplex | Both, independently | A TCP socket |
| Transform | Consumes, changes, produces | `zlib.createGzip()`, `crypto.createHash()`, a CSV parser |

Readable streams are **async iterables**, so the simplest way to consume one is `for await`:

```ts
import { createReadStream } from 'node:fs'

let bytes = 0
for await (const chunk of createReadStream('huge.log')) {
  bytes += chunk.length // chunk is a Buffer of up to 64 KB
}
console.log(`${bytes} bytes read with constant memory`)
```

To process **lines**, wrap the stream with `node:readline`:

```ts
import { createReadStream } from 'node:fs'
import { createInterface } from 'node:readline'

const lines = createInterface({ input: createReadStream('access.log'), crlfDelay: Infinity })
let errors = 0
for await (const line of lines) {
  if (line.includes('" 500 ')) errors++
}
console.log(`${errors} server errors`)
```

## pipeline and backpressure

Connecting streams together, such as reading a file, compressing it and writing it out, is done with `pipeline` from `node:stream/promises`:

```ts
import { createReadStream, createWriteStream } from 'node:fs'
import { pipeline } from 'node:stream/promises'
import { createGzip } from 'node:zlib'

await pipeline(
  createReadStream('export.csv'),
  createGzip(),
  createWriteStream('export.csv.gz'),
)
```

`pipeline` does three jobs you'd otherwise get wrong:

1. **Backpressure.** If the destination is slower than the source (a slow disk, a slow client on a mobile connection), the writable's buffer fills up and `pipeline` pauses the source until it drains. Memory stays bounded however mismatched the speeds are.
2. **Error handling.** An error in any stage rejects the promise and destroys every stream in the chain, closing file handles.
3. **Completion.** The promise resolves only when all data has been flushed to the destination.

The older `a.pipe(b).pipe(c)` style does backpressure but not error propagation or cleanup, so a failed stage can leave file descriptors open. Use `pipeline`.

You can put an async generator function in a pipeline as a transform stage, which is the easiest way to write custom processing:

```ts
await pipeline(
  createReadStream('orders.ndjson'),
  async function* (source) {
    let buffer = ''
    for await (const chunk of source) {
      buffer += chunk
      const lines = buffer.split('\n')
      buffer = lines.pop() ?? ''
      for (const line of lines) {
        if (!line) continue
        const order = JSON.parse(line)
        if (order.total > 100) yield `${order.id},${order.total}\n`
      }
    }
  },
  createWriteStream('big-orders.csv'),
)
```

In a web server, the same idea serves large downloads: pipe a file stream into the response instead of reading the file first. Frameworks such as Hono accept a stream as a response body.

## Events

Many core objects are **event emitters**: they announce things happening, and you subscribe with `on`:

```ts
import { EventEmitter, once } from 'node:events'

const orders = new EventEmitter<{ created: [orderId: string] }>()
orders.on('created', (id) => console.log(`send confirmation for ${id}`))
orders.emit('created', 'o_1')

// Wait for a single event as a promise:
const [id] = await once(orders, 'created')
```

Streams, servers, child processes and sockets all use this API (`'data'`, `'error'`, `'close'`). In your own application code, a direct function call or a job queue is usually clearer than an in-process event bus, but you'll read emitter-based code constantly.

An `'error'` event with no listener crashes the process. Streams used through `pipeline` or `for await` have their errors handled for you; streams you wire up manually need an `'error'` handler.

## Cancellation with AbortSignal

Node uses the same `AbortController` and `AbortSignal` as the browser. Many APIs accept a `signal` and stop when it's aborted, which gives you timeouts and cancellation everywhere:

```ts
import { readFile } from 'node:fs/promises'

// Give up on a slow upstream after 3 seconds
const response = await fetch('https://api.example.com/rates', { signal: AbortSignal.timeout(3000) })

// Cancel several operations together
const controller = new AbortController()
setTimeout(() => controller.abort(), 5000)
const data = await readFile('big.json', { signal: controller.signal, encoding: 'utf8' })
```

An aborted operation rejects with an error whose `name` is `'AbortError'` (or `'TimeoutError'` for `AbortSignal.timeout`). Pass signals down through your own functions so that, when a client disconnects or a deadline passes, the work underneath stops too.

## Running other programs

`node:child_process` runs other programs, for example `ffmpeg` to process video or `git` in a CI tool. Prefer `execFile` and `spawn`, which take the program and its arguments separately:

```ts
import { execFile } from 'node:child_process'
import { promisify } from 'node:util'

const execFileAsync = promisify(execFile)

const { stdout } = await execFileAsync('git', ['log', '-1', '--format=%H'], { timeout: 5000 })
console.log('current commit:', stdout.trim())
```

Avoid `exec` with user input. `exec` runs a **shell** command line, so `exec(\`convert ${filename} out.png\`)` with a filename of `x.png; rm -rf /` runs both commands. `execFile` passes `filename` as a single argument, with no shell to interpret it. This is command injection, a classic and severe vulnerability.

For long-running processes or large output, use `spawn`, whose `stdout` and `stderr` are streams you can pipe, so output never has to fit in memory.

## Practice

Write `summarise-log.ts`, a command-line tool that summarises a gzip-compressed web server log of any size:

1. Usage: `node summarise-log.ts access.log.gz summary.json`.
2. Stream the file through `zlib.createGunzip()` and `readline`, never loading it whole.
3. Each line looks like `203.0.113.9 - - [01/Nov/2026:10:00:00 +0000] "GET /orders/42 HTTP/1.1" 200 512`. Count requests per status code, and the 5 paths with the most 5xx responses.
4. Write the summary as JSON to the output file, and exit with code 1 and a clear message if the input doesn't exist.
5. Generate a test file with a million lines using a stream and `pipeline` (through `createGzip`) and check memory stays low (`process.memoryUsage().rss`).

**Solution**

```ts
// summarise-log.ts
import { createReadStream } from 'node:fs'
import { access, writeFile } from 'node:fs/promises'
import { createInterface } from 'node:readline'
import { createGunzip } from 'node:zlib'

const LINE = /"(?:GET|POST|PUT|PATCH|DELETE|HEAD|OPTIONS) (\S+) [^"]*" (\d{3}) /

async function summarise(input: string) {
  const byStatus: Record<string, number> = {}
  const serverErrors = new Map<string, number>()
  let lines = 0

  const gunzipped = createReadStream(input).pipe(createGunzip())
  for await (const line of createInterface({ input: gunzipped, crlfDelay: Infinity })) {
    const match = LINE.exec(line)
    if (!match) continue
    const [, rawPath, status] = match as unknown as [string, string, string]
    lines++
    byStatus[status] = (byStatus[status] ?? 0) + 1
    if (status.startsWith('5')) {
      const path = rawPath.split('?')[0] ?? rawPath
      serverErrors.set(path, (serverErrors.get(path) ?? 0) + 1)
    }
  }

  const topServerErrors = [...serverErrors]
    .sort((a, b) => b[1] - a[1])
    .slice(0, 5)
    .map(([path, count]) => ({ path, count }))

  return { lines, byStatus, topServerErrors, rssMb: Math.round(process.memoryUsage().rss / 1e6) }
}

const [input, output] = process.argv.slice(2)
if (!input || !output) {
  console.error('usage: node summarise-log.ts <access.log.gz> <summary.json>')
  process.exit(1)
}
try {
  await access(input)
} catch {
  console.error(`input file not found: ${input}`)
  process.exit(1)
}

const summary = await summarise(input)
await writeFile(output, JSON.stringify(summary, null, 2))
console.log(`summarised ${summary.lines} lines, peak RSS about ${summary.rssMb} MB`)
```

```ts
// make-log.ts: generates a test file
import { createWriteStream } from 'node:fs'
import { Readable } from 'node:stream'
import { pipeline } from 'node:stream/promises'
import { createGzip } from 'node:zlib'

const paths = ['/orders', '/orders/42', '/checkout', '/health', '/search?q=x']
const statuses = ['200', '200', '200', '201', '404', '500', '503']

function* lines(n: number) {
  for (let i = 0; i < n; i++) {
    const path = paths[i % paths.length]
    const status = statuses[(i * 3) % statuses.length]
    yield `203.0.113.${i % 255} - - [01/Nov/2026:10:00:00 +0000] "GET ${path} HTTP/1.1" ${status} 512\n`
  }
}

await pipeline(Readable.from(lines(1_000_000)), createGzip(), createWriteStream('access.log.gz'))
```

```bash
node make-log.ts
node summarise-log.ts access.log.gz summary.json
# summarised 1000000 lines, peak RSS about 80 MB
```

Most of that RSS is Node itself, and it stays the same whether the uncompressed log is 80 MB, as here, or 80 GB, because only one chunk and one line are in memory at a time. `Readable.from` turns a generator into a stream, so the generator side gets backpressure too: it only produces lines as fast as gzip and the disk accept them. In `summarise-log.ts`, `.pipe()` is acceptable for the gunzip step because the `for await` loop surfaces errors from the readline interface; for anything with a writable destination, use `pipeline`.

## Key takeaways

- Use `node:fs/promises` and `node:path` in servers; resolve module-relative files with `import.meta.dirname`, and never trust user input in paths.
- Streams process data in chunks with constant memory; consume readables with `for await` and lines with `readline`.
- Connect streams with `pipeline` from `node:stream/promises`, which handles backpressure, errors, cleanup and completion; async generators make easy transforms.
- Use `AbortSignal` (including `AbortSignal.timeout`) to cancel fetches, file reads and your own work.
- Run programs with `execFile` or `spawn`, never `exec` with user input, to avoid command injection.

## Quiz

1. Why prefer pipeline(source, transform, destination) over source.pipe(transform).pipe(destination)?

   A. pipeline is the only way to get backpressure
   B. pipeline propagates errors from any stage and destroys every stream, closing file handles
   C. pipe() doesn't work with gzip
   D. pipeline runs on a separate thread

   **Answer: B.** pipe() handles backpressure but not errors: a failing stage can leave other streams open. pipeline cleans up everything and gives you one promise to await.

2. What does backpressure do when a client downloads a large file slowly?

   A. Drops chunks the client can't keep up with
   B. Buffers the whole file in memory until the client catches up
   C. Pauses reading from the disk until the response buffer drains, keeping memory bounded
   D. Closes the connection

   **Answer: C.** When the writable's buffer is full, the readable is paused. Data flows only as fast as the slowest stage can accept it.

3. A handler runs exec(`convert ${req.query.file} out.png`). What's the risk?

   A. None, if the file exists
   B. The output image may be corrupted
   C. It blocks the event loop
   D. Command injection: a crafted file name can run any shell command

   **Answer: D.** exec passes a string to a shell, so characters like ; and && start new commands. execFile with an argument array passes the name as data, with no shell.

4. How do you stop a fetch call that takes longer than 3 seconds?

   A. fetch(url, { signal: AbortSignal.timeout(3000) })
   B. fetch(url, { timeout: 3 })
   C. setTimeout(() => fetch.cancel(), 3000)
   D. Wrap it in try/catch

   **Answer: A.** fetch accepts an AbortSignal. AbortSignal.timeout(ms) aborts it automatically, and the promise rejects with a TimeoutError.