Node.js: File System, Streams, Async I/O, Uploads and Pipes

Use case: Globomantics — SaaS document processing system

Table of Contents

1. Module 1 — Modern File System Operations
   • Introduction to Node.js File System APIs
   • The Three Approaches: sync, callback, promises
   • Local File System vs Cloud Object Storage
   • File Metadata and Lifecycle Management
2. Module 2 — Streams for Efficient Data Processing
   • Understanding Node.js Streams
   • Processing Large Files with Streams
   • Backpressure and Flow Control
3. Module 3 — Data Piping and Transformation
   • Data Pipelines
   • Composition with pipe() and pipeline()
   • Fan-out and Side Effects
   • File Uploads and Stream Transformations
   • Error Handling in Stream Pipelines
4. Module 4 — Advanced Operations and Automation
   • File Processing API Design
   • Hardening Uploads with pipeline() and AbortController
   • Status and Content Endpoints
   • Automation with File Watching (chokidar)
   • Security and Scalability Guardrails
   • Metadata, Ownership, and Access Control
   • Logs, Operational Health, and Metrics
5. Reference Tables
   • Main fs/promises Methods
   • Stream Types
   • Stream Events
   • HTTP Error Codes in the Upload Pipeline
6. Architecture Diagrams

Module 1 — Modern File System Operations

Introduction to Node.js File System APIs

Globomantics is a SaaS document processing company. Its existing file management system was causing performance issues and security vulnerabilities: users experienced slow uploads, memory crashes when processing large files, and occasional data corruption. When a user uploaded a 100 MB PDF, the entire server became unresponsive — because the system read the whole file into memory synchronously.

The course objective is to rebuild this file management system from scratch using modern Node.js patterns.

The Three Approaches: sync, callback, promises

Node.js provides three main approaches for file operations:

// ❌ Synchronous approach — blocks the event loop
import fs from 'node:fs';
const content = fs.readFileSync('./report.txt', 'utf-8');

// ✅ Modern approach with fs/promises
import { readFile, writeFile, mkdir } from 'node:fs/promises';
import { join, dirname } from 'node:path';
import { fileURLToPath } from 'node:url';

// Reconstructing __dirname in ES Modules
const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);

async function readDocument(filename) {
  try {
    const content = await readFile(join(__dirname, filename), 'utf-8');
    return content;
  } catch (err) {
    if (err.code === 'ENOENT') throw new Error(`File not found: ${filename}`);
    throw err;
  }
}

Key rule: Synchronous operations block the entire event loop. Never use them in HTTP request handlers in production.

package.json configuration for ES Modules:

{
  "name": "globo-systems",
  "type": "module",
  "scripts": {
    "start": "node src/index.js",
    "dev": "node --watch src/index.js"
  },
  "engines": {
    "node": ">=22"
  }
}

Local File System vs Cloud Object Storage

Both storage types play complementary roles:

In practice, both are often used together: files arrive locally first (temporary upload), then are transferred to the cloud.

// S3 access with AWS SDK (cloud example)
import { S3Client, GetObjectCommand } from '@aws-sdk/client-s3';

const s3 = new S3Client({ region: 'us-east-1' });
const response = await s3.send(new GetObjectCommand({
  Bucket: 'globomantics-docs',
  Key: 'uploads/invoice.pdf',
}));

File Metadata and Lifecycle Management

Files carry essential metadata: size, creation/modification/access dates, permissions, and MIME type.

import { stat, chmod } from 'node:fs/promises';

async function getFileMetadata(filePath) {
  const info = await stat(filePath);
  return {
    sizeBytes: info.size,
    createdAt: info.birthtime,
    modifiedAt: info.mtime,
    accessedAt: info.atime,
    isFile: info.isFile(),
    isDirectory: info.isDirectory(),
  };
}

// Modifying permissions (Unix style, octal)
// 0o644 = rw-r--r-- (owner: rw, group: r, other: r)
await chmod('./report.txt', 0o644);

Reading Unix permissions:

-rw-r--r--
│ │  │  └─ Other : r (read)
│ │  └──── Group : r (read)
│ └──────── Owner : rw (read + write)
└────────── Type  : - (file), d (directory)

Module 2 — Streams for Efficient Data Processing

Understanding Node.js Streams

The problem: reading an entire file into memory (the “bucket” approach) does not scale. A 500 MB file can saturate a server’s RAM.

The solution: Streams (the “tap” approach) — process data in small chunks as they arrive.

Streams vs Buffering Advantages

flowchart LR
    subgraph Buffering["❌ Buffer Approach"]
        B1[500 MB File] -->|All into RAM| B2[RAM saturated]
        B2 -->|Processing blocked| B3[Response]
    end
    subgraph Streaming["✅ Stream Approach"]
        S1[500 MB File] -->|64KB chunk| S2[Process chunk]
        S2 -->|next chunk| S2
        S2 -->|Progressive response| S3[Client]
    end

Processing Large Files with Streams

import fs from 'node:fs';

// Create a Readable Stream — nothing is read yet
const readable = fs.createReadStream('./large-dataset.txt');

let chunkCount = 0;
let totalBytes = 0;

// Asynchronous iteration over chunks
for await (const chunk of readable) {
  chunkCount++;
  totalBytes += chunk.length;

  // Log every 1000 iterations to avoid console flooding
  if (chunkCount % 1000 === 0) {
    const memMb = process.memoryUsage().rss / 1024 / 1024;
    console.log(`Chunks: ${chunkCount}, Bytes: ${totalBytes}, RAM: ${memMb.toFixed(1)} MB`);
  }
}

console.log(`Total: ${chunkCount} chunks, ${totalBytes} bytes`);

Key observation: Memory usage (RSS) stays constant even for a 1 GB file, because Node.js only keeps one chunk in memory at a time.

Backpressure and Flow Control

Backpressure is the mechanism that allows a slow destination to signal a fast source to slow down. It is the feature that prevents memory overflows.

sequenceDiagram
    participant OS as OS / Disk
    participant RS as Readable Stream
    participant TS as Transform Stream
    participant WS as Writable Stream

    OS->>RS: data chunk
    RS->>TS: push(chunk)
    TS->>WS: write(chunk)
    WS-->>TS: return false (buffer full)
    TS-->>RS: pause()
    Note over RS: ⏸ Paused — reading stopped
    WS-->>TS: emit('drain')
    TS-->>RS: resume()
    Note over RS: ▶ Reading resumed

Explicit form of backpressure:

import { createReadStream, createWriteStream } from 'node:fs';

const readable = createReadStream('./source.txt');
const writable = createWriteStream('./destination.txt');

readable.on('data', (chunk) => {
  // write() returns false if the internal buffer is full
  const canContinue = writable.write(chunk);
  if (!canContinue) {
    readable.pause(); // ⏸ Stop the source
    writable.once('drain', () => {
      readable.resume(); // ▶ Resume when destination is ready
    });
  }
});

readable.on('end', () => writable.end());

highWaterMark parameter: controls the internal buffer size before backpressure triggers.

// Default: 64KB for byte-mode streams
const rs = fs.createReadStream('./file.txt', { highWaterMark: 128 * 1024 }); // 128 KB

Three layers of flow control:

1. highWaterMark — threshold triggering backpressure
2. Async iteration (for await) — Node pauses/resumes automatically
3. OS pacing — the operating system regulates the source itself

Module 3 — Data Piping and Transformation

Data Pipelines

A data pipeline always has three parts: a source, one or more transformations, and a destination. A Transform stream runs once per chunk.

flowchart LR
    SRC["📂 Source\n(Readable)"]
    T1["⚙️ Transform 1\n(e.g., uppercase)"]
    T2["⚙️ Transform 2\n(e.g., gzip)"]
    DST["💾 Destination\n(Writable)"]

    SRC -->|chunks| T1
    T1 -->|transformed chunks| T2
    T2 -->|compressed chunks| DST

Creating a Transform stream:

import { Transform } from 'node:stream';

const upperCaseTransform = new Transform({
  transform(chunk, encoding, callback) {
    try {
      // Convert chunk Buffer to string, transform, return
      const transformed = chunk.toString('utf-8').toUpperCase();
      callback(null, transformed); // null = no error
    } catch (err) {
      callback(err); // Propagate the error
    }
  }
});

Composition with pipe() and pipeline()

pipe() — basic composition

import { createReadStream, createWriteStream } from 'node:fs';
import { Transform } from 'node:stream';

const readable = createReadStream('./input.txt');
const upperCase = new Transform({ /* ... */ });
const writable = createWriteStream('./output.txt');

// Chaining with pipe()
readable
  .pipe(upperCase)
  .pipe(writable);

Limitation of pipe(): errors do not propagate automatically between streams. You must attach .on('error') handlers on each stream manually.

pipeline() — robust composition (recommended in production)

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

// ✅ pipeline(): single error handler, automatic cleanup
await pipeline(
  createReadStream('./input.txt'),
  createGzip(),
  createWriteStream('./output.txt.gz')
);
// If an error occurs at any stage, all streams are closed

flowchart LR
    R["Readable\n(source)"] -->|pipe| T["Transform\n(uppercase)"] -->|pipe| W["Writable\n(dest)"]
    style R fill:#4CAF50,color:#fff
    style T fill:#2196F3,color:#fff
    style W fill:#FF5722,color:#fff

Fan-out and Side Effects

Fan-out means sending the same data to multiple destinations from a single source.

flowchart TD
    SRC["📂 Source\n(Readable)"]
    PT["🔀 PassThrough\n(tee)"]
    B1["📝 Branch 1\nTransform + Write (main file)"]
    B2["📊 Branch 2\nTransform + Metrics / audit"]

    SRC --> PT
    PT --> B1
    PT --> B2

Implementation with PassThrough:

import { PassThrough } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createReadStream, createWriteStream } from 'node:fs';

const source = createReadStream('./input.txt');
const tee = new PassThrough(); // Acts like a plumbing "T"

// Pipe source → tee (only once)
source.pipe(tee);

// Branch 1: main write
const branch1 = pipeline(tee, createWriteStream('./output-main.txt'));

// Branch 2: audit/metrics (read from the same source)
let totalBytes = 0;
tee.on('data', (chunk) => { totalBytes += chunk.length; });
tee.on('end', () => console.log(`Total bytes processed: ${totalBytes}`));

await branch1;

Note: with fan-out, the slowest path controls the speed of the entire chain.

File Uploads and Stream Transformations

Multipart upload flow:

flowchart LR
    C["Client\n(HTTP POST)"]
    MP["Multipart\nParser"]
    RS["Readable Stream\n(file)"]
    T1["Transform 1\nValidation / Limit"]
    T2["Transform 2\nHashing / Transform"]
    FS["💾 File Store\n(disk / S3)"]

    C -->|multipart form| MP
    MP -->|file part| RS
    RS --> T1
    T1 --> T2
    T2 --> FS

The multipart parser splits the HTTP body and exposes the file part as a Readable Stream. From that point, everything is handled with standard stream primitives.

Buffer vs Stream for uploads:

Error Handling in Stream Pipelines

Sources of errors in a pipeline:

1. Source errors — file deleted, socket cut, client aborts upload
2. Transform errors — invalid JSON, corrupted data, exception in code
3. Destination errors — disk full, permissions denied, remote service rejected
4. Side-effect errors — logs, metrics, DB writes

With .pipe() (verbose, fragile):

// ❌ Fragile approach — each stream has its own error handler
readable.on('error', handleError);
transform1.on('error', handleError);
transform2.on('error', handleError);
writable.on('error', handleError);

With pipeline() (recommended):

import { pipeline } from 'node:stream/promises';

try {
  await pipeline(
    sourceStream,        // Readable
    transform1,          // Transform
    transform2,          // Transform
    destinationStream    // Writable
  );
} catch (err) {
  // ✅ All errors arrive here
  // All streams are automatically closed
  console.error('Pipeline failed:', err);
  // Cleanup partial files, etc.
}

Module 4 — Advanced Operations and Automation

File Processing API Design

The Globomantics system architecture follows a clear processing pipeline:

flowchart LR
    subgraph Ingestion["Ingestion"]
        HTTP["📡 HTTP Upload\n(multipart)"]
        WATCH["👀 File Watcher\n(incoming directory)"]
    end

    INGEST["🔀 Ingest\n(ID + context + stream)"]
    PIPE["⚙️ Stream Pipeline\n(limits, validation,\nhashing, metrics)"]
    COMMIT["✅ Commit\n(success / failure / abort)"]

    subgraph Storage["Storage"]
        FS["💾 File Store\n(staging → processed)"]
        META["🗄️ Metadata Store\n(status, size, MIME,\nowners, timestamps)"]
    end

    HTTP --> INGEST
    WATCH --> INGEST
    INGEST --> PIPE
    PIPE --> COMMIT
    COMMIT --> FS
    COMMIT --> META

    GET_STATUS["GET /files/:id\n(status)"] --> META
    GET_CONTENT["GET /files/:id/content\n(stream)"] --> FS

File lifecycle stages:

Hardening Uploads with pipeline() and AbortController

Hardening principles:

1. Staged writes — write to a .tmp file (staging), rename to the final directory only on success
2. Abort handling — AbortController to stop the pipeline if the client disconnects
3. Predictable failure — cleanup of partial files, mapping errors to clear HTTP codes

// src/lib/stream-pipeline.js
import { pipeline } from 'node:stream/promises';
import fs from 'node:fs';
import { Transform } from 'node:stream';

// Transform that limits file size
export function createMaxBytesTransform(maxBytes) {
  let seen = 0;
  return new Transform({
    transform(chunk, _enc, cb) {
      seen += chunk.length;
      if (seen > maxBytes) {
        const err = new Error(`File too large (>${maxBytes} bytes)`);
        err.code = 'LIMIT_FILE_SIZE';
        cb(err);
        return;
      }
      cb(null, chunk);
    },
  });
}

// "tap" Transform — side effect without modifying data
export function createTapTransform(sideEffect) {
  return new Transform({
    transform(chunk, _enc, cb) {
      try {
        sideEffect(chunk);
        cb(null, chunk);
      } catch (err) {
        cb(err);
      }
    },
  });
}

// Main upload pipeline
export async function runUploadPipeline({ sourceStream, destPath, maxBytes, signal }) {
  let totalBytes = 0;

  const tap = createTapTransform((chunk) => {
    totalBytes += chunk.length;
  });

  await pipeline(
    sourceStream,
    createMaxBytesTransform(maxBytes),
    tap,
    fs.createWriteStream(destPath),
    { signal } // AbortController signal
  );

  return { bytesWritten: totalBytes };
}

Upload route with staging and abort:

// src/routes.js (excerpt)
app.post('/upload', async (req, reply) => {
  const ingested = await ingestFirstFileFromMultipart(req);
  if (!ingested) {
    return reply.code(400).send({ ok: false, error: 'No file part found' });
  }

  const finalPath = path.join(config.processedDir, `${ingested.id}.bin`);
  const tmpPath   = path.join(config.stagingDir,   `${ingested.id}.tmp`);

  // AbortController to stop the pipeline if the client disconnects
  const ac = new AbortController();
  req.raw.on('close', () => ac.abort());

  try {
    const { bytesWritten } = await runUploadPipeline({
      sourceStream: ingested.stream,
      destPath: tmpPath,        // ← write to staging
      maxBytes: config.maxFileBytes,
      signal: ac.signal,
    });

    // ✅ Success — move to final directory
    await fsp.rename(tmpPath, finalPath);
    updateRecord(ingested.id, { status: 'succeeded', bytesStored: bytesWritten, storedPath: finalPath });

    return reply.send({ ok: true, id: ingested.id, bytesWritten });

  } catch (err) {
    // ❌ Failure — cleanup partial file
    await fsp.rm(tmpPath, { force: true }).catch(() => {});
    updateRecord(ingested.id, { status: 'failed' });

    const mapped = mapUploadError(err);
    return reply.code(mapped.status).send(mapped.body);
  }
});

Error mapping:

function mapUploadError(err) {
  if (err?.name === 'AbortError' || err?.code === 'ABORT_ERR') {
    return { status: 499, body: { ok: false, error: 'Upload aborted' } };
    // 499 = "Client Closed Request" (nginx convention)
  }
  if (err?.code === 'LIMIT_FILE_SIZE') {
    return { status: 413, body: { ok: false, error: 'File too large' } };
  }
  return { status: 500, body: { ok: false, error: 'Upload failed' } };
}

Status and Content Endpoints

Key rule:

• GET /files/:id → reads only metadata (never the file)
• GET /files/:id/content → streams the content only if status === 'succeeded'

// Status endpoint — metadata only
app.get('/files/:id', async (req, reply) => {
  const rec = getById(req.params.id);
  if (!rec) return reply.code(404).send({ ok: false, error: 'Not found' });
  assertOwner({ req, rec }); // Ownership check
  return reply.send({ ok: true, file: rec });
});

// Content endpoint — streaming the file
app.get('/files/:id/content', async (req, reply) => {
  const rec = getById(req.params.id);
  if (!rec) return reply.code(404).send({ ok: false, error: 'Not found' });
  assertOwner({ req, rec });

  // Check that the file is ready
  if (rec.status !== 'succeeded') {
    return reply.code(409).send({
      ok: false,
      error: 'Not ready',
      status: rec.status, // 'uploading', 'processing', 'failed', 'aborted'
    });
  }

  // Stream the file with the right headers
  const stream = fs.createReadStream(rec.storedPath);
  reply.header('Content-Type', rec.mime ?? 'application/octet-stream');
  reply.header('Content-Disposition', `attachment; filename="${rec.originalName}"`);
  return reply.send(stream);
});

Automation with File Watching (chokidar)

File watchers monitor a directory and trigger the pipeline on new files — a second ingestion source in addition to HTTP upload.

Caveat: Node.js native fs.watch() is unreliable (may fire multiple times, may fire during write). chokidar resolves these issues.

// src/lib/watcher.js
import chokidar from 'chokidar';
import fs from 'node:fs';
import fsp from 'node:fs/promises';
import path from 'node:path';
import crypto from 'node:crypto';
import { createRecord, updateRecord } from '../store.js';
import { runUploadPipeline } from './stream-pipeline.js';
import { config } from '../config.js';

// Set to prevent double-processing (idempotency)
const inFlight = new Set();

export function startIncomingWatcher({ logger }) {
  const incomingDir = config.incomingDir;

  const watcher = chokidar.watch(incomingDir, {
    ignoreInitial: true, // Ignore files already present at startup
    awaitWriteFinish: {
      stabilityThreshold: 1000, // Wait 1s of stability before processing
      pollInterval: 100,
    },
  });

  watcher.on('add', (filePath) => {
    void handleIncomingFile({ filePath, logger });
  });

  watcher.on('error', (err) => {
    logger.error({ err }, 'Incoming watcher error');
  });

  logger.info({ incomingDir, stabilityMs: 1000 }, 'Incoming watcher started');
  return watcher;
}

async function handleIncomingFile({ filePath, logger }) {
  const stat = await fsp.stat(filePath);
  if (!stat?.isFile()) return;

  const abs = path.resolve(filePath);

  // Guard against duplicates (watcher can fire multiple times)
  if (inFlight.has(abs)) return;
  inFlight.add(abs);

  const id = crypto.randomUUID();
  const originalName = path.basename(filePath);

  createRecord({ id, originalName, mime: 'application/octet-stream', source: 'incoming' });
  updateRecord(id, { status: 'processing' });

  try {
    const tmpPath   = path.join(config.stagingDir,   `${id}.tmp`);
    const finalPath = path.join(config.processedDir, `${id}.bin`);

    const { bytesWritten } = await runUploadPipeline({
      sourceStream: fs.createReadStream(abs),
      destPath: tmpPath,
      maxBytes: config.maxFileBytes,
      signal: undefined,
    });

    await fsp.rename(tmpPath, finalPath);
    updateRecord(id, { status: 'succeeded', bytesStored: bytesWritten, storedPath: finalPath });
    logger.info({ id, originalName, bytesWritten }, 'Incoming file processed');

  } catch (err) {
    updateRecord(id, { status: 'failed' });
    logger.error({ id, err }, 'Incoming file processing failed');
  } finally {
    inFlight.delete(abs);
  }
}

Watcher vs HTTP flow:

flowchart TD
    subgraph Sources
        HTTP["📡 HTTP Multipart\n(POST /upload)"]
        WATCHER["👀 chokidar\n(incoming/ directory)"]
    end

    INGEST["🔀 Ingest\n(ID + stream)"]
    PIPELINE["⚙️ runUploadPipeline()\n(max bytes, tap, pipeline)"]
    STAGING["📁 Staging\n(.tmp)"]
    FINAL["💾 Processed\n(.bin)"]
    META["🗄️ Metadata Store"]

    HTTP -->|ingested.stream| INGEST
    WATCHER -->|"fs.createReadStream()"| INGEST
    INGEST --> PIPELINE
    PIPELINE --> STAGING
    STAGING -->|"fsp.rename()"| FINAL
    FINAL --> META

Security and Scalability Guardrails

The 6 security gates of the system:

// ✅ Safe path — always server IDs, never user-supplied filenames
const id = crypto.randomUUID(); // UUID v4, non-predictable
const safePath = path.join(config.processedDir, `${id}.bin`);

// ❌ NEVER — path traversal possible
const unsafePath = path.join(config.processedDir, req.body.filename);
// Attack example: filename = "../../etc/passwd"

Configuration via environment variables:

// src/config.js
export const config = {
  maxFileBytes: parseInt(process.env.MAX_FILE_BYTES ?? String(10 * 1024 * 1024)), // 10 MB
  stagingDir:   process.env.STAGING_DIR   ?? './data/staging',
  processedDir: process.env.PROCESSED_DIR ?? './data/processed',
  failedDir:    process.env.FAILED_DIR    ?? './data/failed',
  incomingDir:  process.env.INCOMING_DIR  ?? './data/incoming',
};

Metadata, Ownership, and Access Control

In production, metadata is the system’s control plane: it decides whether a file exists, whether it is ready, and who has the right to access it.

flowchart LR
    REQ["📡 Request\n(untrusted)"] -->|x-owner-id header| AUTH["🔐 Auth\npreHandler hook"]
    AUTH -->|req.user.id| ROUTE["🛣️ Route Handler"]
    ROUTE -->|id lookup| META["🗄️ Metadata Store\n(ownerId)"]
    META -->|"assertOwner()"| FS["💾 File Store\n(streaming)"]

    style AUTH fill:#FF5722,color:#fff
    style META fill:#2196F3,color:#fff

// src/lib/auth.js
export function registerAuth(app) {
  app.addHook('preHandler', async (req, reply) => {
    const ownerId = req.headers['x-owner-id']; // Conventional custom header (x- prefix)
    if (!ownerId) {
      return reply.code(401).send({ ok: false, error: 'Missing x-owner-id' });
    }
    req.user = { id: ownerId }; // Attached to the request for routes
  });
}

// Ownership check in routes
function assertOwner({ req, rec }) {
  if (rec.ownerId !== req.user.id) {
    const err = new Error('Forbidden');
    err.code = 'ERR_FORBIDDEN';
    throw err;
  }
}

Note: In production, x-owner-id would be replaced by a JWT token or a server-verified session.

Logs, Operational Health, and Metrics

Operational checklist:

// Health endpoint — checks that directories are writable
app.get('/health', async () => {
  const [stagingOk, processedOk, failedOk] = await Promise.all([
    canWrite(config.stagingDir),
    canWrite(config.processedDir),
    canWrite(config.failedDir),
  ]);

  return {
    ok: stagingOk && processedOk && failedOk,
    checks: {
      stagingWritable: stagingOk,
      processedWritable: processedOk,
      failedWritable: failedOk,
    },
  };
});

// Write check
async function canWrite(dir) {
  const testFile = path.join(dir, '.healthcheck');
  try {
    await fsp.mkdir(dir, { recursive: true });
    await fsp.writeFile(testFile, 'ok');
    await fsp.rm(testFile, { force: true });
    return true;
  } catch {
    return false;
  }
}

Useful metrics for this pipeline:

// src/lib/metrics.js (simplified)
const counters = {};

export function inc(name, labels = {}) {
  const key = `${name}:${JSON.stringify(labels)}`;
  counters[key] = (counters[key] ?? 0) + 1;
}

export function snapshot() {
  return { ...counters };
}

// Usage
inc('uploads_started_total', { source: 'http' });
inc('uploads_succeeded_total');
inc('uploads_failed_total', { reason: 'file_too_large' });

Metrics to monitor:

• uploads_started_total — uploads started
• uploads_succeeded_total — uploads completed successfully
• uploads_failed_total — failed uploads (by reason)
• uploads_aborted_total — uploads abandoned by the client
• bytes_processed_total — total volume processed
• upload_duration_ms — upload duration (p50, p95, p99)
• inflight_count — work in progress (queue depth)

Reference Tables

Main fs/promises Methods

Stream Types

Stream Events

HTTP Error Codes in the Upload Pipeline

Architecture Diagrams

Complete Node.js Streams Architecture

classDiagram
    class EventEmitter {
        +on(event, listener)
        +emit(event, ...args)
        +removeListener(event, listener)
    }

    class Stream {
        +pipe(destination)
        +destroy(err)
    }

    class Readable {
        +read(size)
        +pause()
        +resume()
        +pipe(dest)
        +unpipe(dest)
        +setEncoding(encoding)
        -highWaterMark
        +event: data
        +event: end
        +event: readable
        +event: error
        +event: close
    }

    class Writable {
        +write(chunk, encoding, cb)
        +end(chunk)
        +cork()
        +uncork()
        -highWaterMark
        +event: drain
        +event: finish
        +event: error
        +event: close
    }

    class Duplex {
        +read(size)
        +write(chunk, encoding, cb)
    }

    class Transform {
        +_transform(chunk, encoding, cb)
        +_flush(cb)
    }

    class PassThrough {
        +_transform(chunk, encoding, cb)
    }

    EventEmitter <|-- Stream
    Stream <|-- Readable
    Stream <|-- Writable
    Readable <|-- Duplex
    Writable <|-- Duplex
    Duplex <|-- Transform
    Transform <|-- PassThrough

Backpressure Flow

flowchart TD
    A["📂 Readable Stream\n(source)"] -->|chunk| B["⚙️ Transform Stream\n(processing)"]
    B -->|transformed chunk| C["💾 Writable Stream\n(destination)"]

    C -->|"write() returns false\n(buffer full)"| D{{"⏸ Backpressure"}}
    D -->|"pause()"| A
    C -->|"emit('drain')\n(buffer drained)"| E{{"▶ Resume"}}
    E -->|"resume()"| A

    style D fill:#FF5722,color:#fff
    style E fill:#4CAF50,color:#fff

Complete Upload Pipeline with Staging

sequenceDiagram
    participant C as Client
    participant R as Route /upload
    participant P as Multipart Parser
    participant PL as runUploadPipeline()
    participant S as Staging (.tmp)
    participant F as Processed (.bin)
    participant M as Metadata Store

    C->>R: POST /upload (multipart)
    R->>P: ingestFirstFileFromMultipart(req)
    P->>R: { id, stream, mime, originalName }
    R->>M: createRecord({ id, status: 'uploading' })
    R->>PL: runUploadPipeline({ stream, tmpPath, maxBytes, signal })
    PL->>S: pipeline(source → maxBytes → tap → writeStream)
    Note over PL,S: Write to staging only

    alt Success
        PL->>R: { bytesWritten }
        R->>F: fsp.rename(tmpPath, finalPath)
        R->>M: updateRecord({ status: 'succeeded' })
        R->>C: 200 { ok: true, id, bytesWritten }
    else File too large (LIMIT_FILE_SIZE)
        PL->>R: throw err (code: LIMIT_FILE_SIZE)
        R->>S: fsp.rm(tmpPath, { force: true })
        R->>M: updateRecord({ status: 'failed' })
        R->>C: 413 { ok: false, error: 'File too large' }
    else Client disconnected (AbortError)
        C-->>R: connection close
        R->>PL: ac.abort()
        PL->>R: throw AbortError
        R->>S: fsp.rm(tmpPath, { force: true })
        R->>M: updateRecord({ status: 'aborted' })
        R->>C: 499 { ok: false, error: 'Upload aborted' }
    end

Key Concepts — Summary

Search Terms

node.js · system · streams · async · uploads · pipes · apis · microservices · backend · full-stack · web · pipeline · stream · composition · data · processing · architecture · automation · backpressure · control · error · flow · metadata · operations