Callbacks, Promises & async/await

import { readFile } from 'node:fs';

function countBooks(path, callback) {
  readFile(path, 'utf8', (err, text) => {
    if (err) {
      callback(err); // error first; no second argument
      return;
    }
    const rows = text.trim().split('\n').slice(1);
    callback(null, rows.length);
  });
}

countBooks('data.csv', (err, n) => {
  if (err) throw err;
  console.log('books:', n);
});

countBooks('missing.csv', (err, n) => {
  console.log(err.code, err.syscall, n);
});

// A try/catch around the call cannot see an error that is delivered later.
try {
  readFile('missing.csv', (err) => {
    console.log('delivered to the callback:', err.code);
  });
  console.log('the try block has already finished');
} catch {
  console.log('never reached');
}

the try block has already finished
ENOENT open undefined
delivered to the callback: ENOENT
books: 2

import { readFile, writeFile } from 'node:fs';

// Nested ("pyramid") style: every step is one level deeper
readFile('data.csv', 'utf8', (err, text) => {
  if (err) return console.error(err);
  const upper = text.toUpperCase();
  writeFile('out-upper.csv', upper, (err) => {
    if (err) return console.error(err);
    readFile('out-upper.csv', 'utf8', (err, copy) => {
      if (err) return console.error(err);
      console.log('nested:', copy.split('\n')[1]);
    });
  });
});

// Flattened with named functions: same behaviour, one level
function step1() {
  readFile('data.csv', 'utf8', step2);
}
function step2(err, text) {
  if (err) return console.error(err);
  writeFile('out-upper2.csv', text.toUpperCase(), step3);
}
function step3(err) {
  if (err) return console.error(err);
  readFile('out-upper2.csv', 'utf8', step4);
}
function step4(err, copy) {
  if (err) return console.error(err);
  console.log('flat:', copy.split('\n')[1]);
}
step1();

flat: DUNE,HERBERT,1965
nested: DUNE,HERBERT,1965

import { readFile } from 'node:fs';

const cache = new Map();

// Inconsistent: synchronous on a cache hit, asynchronous on a miss
function getTitleZalgo(path, callback) {
  if (cache.has(path)) {
    callback(null, cache.get(path)); // sync
    return;
  }
  readFile(path, 'utf8', (err, text) => {
    if (err) return callback(err);
    const title = text.split('\n')[1].split(',')[0];
    cache.set(path, title);
    callback(null, title); // async
  });
}

// Consistent: always asynchronous
function getTitle(path, callback) {
  if (cache.has(path)) {
    queueMicrotask(() => callback(null, cache.get(path)));
    return;
  }
  readFile(path, 'utf8', (err, text) => {
    if (err) return callback(err);
    const title = text.split('\n')[1].split(',')[0];
    cache.set(path, title);
    callback(null, title);
  });
}

function demo(label, fn, then) {
  let ready = false;
  fn('data.csv', () => {
    console.log(label, 'callback ran with ready =', ready);
  });
  ready = true; // the caller assumes the callback has not run yet
  if (then) setTimeout(then, 20);
}

demo('zalgo, first call (miss):', getTitleZalgo, () => {
  demo('zalgo, second call (hit): ', getTitleZalgo, () => {
    demo('fixed, third call (hit): ', getTitle);
  });
});

zalgo, first call (miss): callback ran with ready = true
zalgo, second call (hit):  callback ran with ready = false
fixed, third call (hit):  callback ran with ready = true

import { readFile } from 'node:fs/promises';

const p = readFile('data.csv', 'utf8');
console.log('right after the call:', p);

readFile('data.csv', 'utf8')
  .then((text) => text.trim().split('\n').slice(1))      // returns a value
  .then((rows) => rows.map((row) => row.split(',')))     // returns a value
  .then(([[title]]) => readFile(`${title.toLowerCase()}.txt`, 'utf8')) // returns a promise: the chain waits
  .then((never) => console.log('not reached', never))
  .catch((err) => {
    console.log('caught:', err.code);
    return 'recovered';                                  // the chain continues as fulfilled
  })
  .then((value) => console.log('after catch:', value))
  .finally(() => console.log('finally runs either way'));

// A promise settles once; later calls are ignored
new Promise((resolve, reject) => {
  resolve('first');
  resolve('second');
  reject(new Error('ignored'));
}).then((v) => console.log('settled with', v));

right after the call: Promise { <pending> }
settled with first
caught: ENOENT
after catch: recovered
finally runs either way

import { promisify, callbackify } from 'node:util';
import { readFile } from 'node:fs';
import { setTimeout as sleep } from 'node:timers/promises';
import { execFile } from 'node:child_process';
import timers from 'node:timers';

// 1. promisify a standard error-first function
const readFileP = promisify(readFile);
console.log((await readFileP('data.csv', 'utf8')).split('\n')[0]);

// 2. functions with a custom promisified variant return something more useful
console.log(promisify(timers.setTimeout) === sleep);
const execFileP = promisify(execFile);
const { stdout } = await execFileP(process.execPath, ['-p', '1 + 1']);
console.log('execFile resolves an object:', stdout.trim());

// 3. a hand-written custom variant for a callback with a non-standard shape
function legacyLookup(id, onOk, onFail) {
  setImmediate(() => (id === 1 ? onOk({ id, title: 'Dune' }) : onFail(new Error('no such id'))));
}
legacyLookup[promisify.custom] = (id) =>
  new Promise((resolve, reject) => legacyLookup(id, resolve, reject));
const lookup = promisify(legacyLookup);
console.log(await lookup(1));
await lookup(2).catch((e) => console.log(e.message));

// 4. callbackify: an async function for code that wants a callback
async function titleOf(id) {
  if (id !== 1) throw new Error('no such id');
  return 'Dune';
}
const titleOfCb = callbackify(titleOf);
titleOfCb(1, (err, title) => console.log('callbackify ok:', err, title));
titleOfCb(2, (err, title) => console.log('callbackify err:', err.message, title));

// 5. callbackify wraps a falsy rejection in ERR_FALSY_VALUE_REJECTION
callbackify(() => Promise.reject(null))((err) => console.log(err.code, err.reason));

title,author,year
true
execFile resolves an object: 2
{ id: 1, title: 'Dune' }
no such id
callbackify ok: null Dune
callbackify err: no such id undefined
ERR_FALSY_VALUE_REJECTION null

import { promisify } from 'node:util';
import { readFile } from 'node:fs/promises';

// DEP0174: promisify() of a function that already returns a promise
const bad = promisify(readFile);
const p = bad('data.csv', 'utf8');
console.log(await Promise.race([p, new Promise((r) => setTimeout(r, 200, 'still pending after 200 ms'))]));

(node:12345) [DEP0174] DeprecationWarning: Calling promisify on a function that returns a Promise is likely a mistake.
still pending after 200 ms

import { readFile, stat } from 'node:fs/promises';
import { setTimeout as sleep } from 'node:timers/promises';
import { pipeline } from 'node:stream/promises';
import { lookup } from 'node:dns/promises';
import { createInterface } from 'node:readline/promises';
import { Readable, Writable } from 'node:stream';

console.log((await stat('data.csv')).size, 'bytes');
console.log(await sleep(20, 'timers/promises resolves with the value'));

const chunks = [];
await pipeline(
  Readable.from(['a', 'b', 'c']),
  new Writable({ write(chunk, _enc, done) { chunks.push(chunk.toString()); done(); } }),
);
console.log('stream/promises pipeline done:', chunks.join(''));

console.log('dns/promises:', (await lookup('localhost', { family: 4 })).address);
console.log('readline/promises:', typeof createInterface);

53 bytes
timers/promises resolves with the value
stream/promises pipeline done: abc
dns/promises: 127.0.0.1
readline/promises: function

import { setTimeout as sleep } from 'node:timers/promises';

const ok = (ms, v) => sleep(ms, v);
const fail = (ms, msg) => sleep(ms).then(() => { throw new Error(msg); });

// all: fulfils with an array in input order; rejects with the first rejection
console.log(await Promise.all([ok(30, 'a'), ok(10, 'b'), 3]));
await Promise.all([ok(30, 'a'), fail(10, 'b failed'), fail(20, 'c failed')])
  .catch((e) => console.log('all rejected with:', e.message));

// allSettled: never rejects; one record per input
const settled = await Promise.allSettled([ok(10, 'a'), fail(10, 'b failed')]);
console.log(settled.map((r) => (r.status === 'fulfilled' ? r.value : r.reason.message)), settled.map((r) => r.status));

// any: first fulfilment; AggregateError when every input rejects
console.log(await Promise.any([fail(10, 'x'), ok(30, 'y'), ok(50, 'z')]));
await Promise.any([fail(10, 'x'), fail(20, 'y')]).catch((e) =>
  console.log(e.constructor.name, e.message, e.errors.map((x) => x.message)));

// race: first settlement wins, fulfilled or rejected
console.log(await Promise.race([ok(30, 'slow'), ok(10, 'fast')]));
await Promise.race([ok(30, 'slow'), fail(10, 'fast failure')])
  .catch((e) => console.log('race rejected with:', e.message));

// the empty cases
console.log(await Promise.all([]), await Promise.allSettled([]));
Promise.any([]).catch((e) => console.log('any([]) ->', e.constructor.name));

[ 'a', 'b', 3 ]
all rejected with: b failed
[ 'a', 'b failed' ] [ 'fulfilled', 'rejected' ]
y
AggregateError All promises were rejected [ 'x', 'y' ]
fast
race rejected with: fast failure
[] []
any([]) -> AggregateError

import { readFile } from 'node:fs/promises';
import { setTimeout as sleep } from 'node:timers/promises';

// An async function always returns a promise, even when it returns a plain value
async function answer() { return 42; }
console.log(answer() instanceof Promise, await answer());

// A throw inside becomes a rejection; try/catch around await sees it
async function load(path) {
  try {
    return JSON.parse(await readFile(path, 'utf8'));
  } catch (err) {
    console.log(`load(${path}) failed:`, err.code ?? err.name);
    return null;
  }
}
console.log(await load('missing.json'));

// Sequential vs concurrent: same work, different elapsed time
const fetchBook = async (id) => { await sleep(100); return { id }; };
const time = async (label, fn) => {
  const t = performance.now();
  const r = await fn();
  console.log(label, Math.round((performance.now() - t) / 50) * 50, 'ms ->', r.length, 'books');
};
const ids = [1, 2, 3, 4];

await time('sequential:', async () => {
  const out = [];
  for (const id of ids) out.push(await fetchBook(id));
  return out;
});
await time('concurrent:', () => Promise.all(ids.map(fetchBook)));

// forEach does not wait for async callbacks
const seen = [];
ids.forEach(async (id) => { await fetchBook(id); seen.push(id); });
console.log('after forEach:', seen.length, 'done');
await sleep(150);
console.log('later:', seen.length, 'done');

true 42
load(missing.json) failed: ENOENT
null
sequential: 400 ms -> 4 books
concurrent: 100 ms -> 4 books
after forEach: 0 done
later: 4 done

import { setTimeout as sleep } from 'node:timers/promises';

async function inner() { await sleep(5); throw new Error('boom'); }

async function withoutAwait() {
  try { return inner(); }            // the promise escapes the try block
  catch { return 'caught here'; }
}
async function withAwait() {
  try { return await inner(); }      // the rejection is thrown inside the try block
  catch { return 'caught here'; }
}
console.log(await withAwait());
console.log(await withoutAwait().catch((e) => 'caught by the caller: ' + e.message));

caught here
caught by the caller: boom

import { createReadStream } from 'node:fs';
import { createInterface } from 'node:readline';
import { setInterval as ticks } from 'node:timers/promises';
import { once, on, EventEmitter } from 'node:events';

// 1. a Readable stream is an async iterable of chunks
let bytes = 0;
for await (const chunk of createReadStream('data.csv')) bytes += chunk.length;
console.log('bytes read:', bytes);

// 2. readline gives lines; crlfDelay: Infinity treats \r\n as one line break
const rl = createInterface({ input: createReadStream('data.csv'), crlfDelay: Infinity });
for await (const line of rl) console.log('line:', line);

// 3. an async generator is the producer side of for await
async function* books() {
  for await (const line of createInterface({ input: createReadStream('data.csv') })) {
    if (line.startsWith('title')) continue;
    const [title, author, year] = line.split(',');
    yield { title, author, year: Number(year) };
  }
}
for await (const b of books()) console.log(b.title, b.year);
console.log(await Array.fromAsync(books(), (b) => b.title));

// 4. timers/promises setInterval is an async iterator; break stops it
let n = 0;
for await (const _ of ticks(20)) { if (++n === 3) break; }
console.log('ticks:', n);

// 5. events.on turns an emitter into an async iterator
const ee = new EventEmitter();
setImmediate(() => { ee.emit('book', 'Dune'); ee.emit('book', 'Emma'); ee.emit('end'); });
const ac = new AbortController();
once(ee, 'end').then(() => ac.abort());
try {
  for await (const [title] of on(ee, 'book', { signal: ac.signal })) console.log('event:', title);
} catch (e) {
  console.log(e.name);
}

bytes read: 53
line: title,author,year
line: Dune,Herbert,1965
line: Emma,Austen,1815
Dune 1965
Emma 1815
[ 'Dune', 'Emma' ]
ticks: 3
event: Dune
event: Emma
AbortError

// tla.mjs
import { readFile } from 'node:fs/promises';
const csv = await readFile(new URL('./data.csv', import.meta.url), 'utf8');
export const titles = csv.trim().split('\n').slice(1).map((r) => r.split(',')[0]);
console.log('tla module evaluated:', titles);

// tla-main.mjs
console.log('main starts');
const { titles } = await import('./tla.mjs');
console.log('main resumes:', titles.length);

// tla-req.cjs
try {
  require('./tla.mjs');
} catch (e) {
  console.log(e.code);
}

// unsettled.mjs
console.log("waiting"); await new Promise(() => {}); console.log("never");

main starts
tla module evaluated: [ 'Dune', 'Emma' ]
main resumes: 2

ERR_REQUIRE_ASYNC_MODULE

waiting
Warning: Detected unsettled top-level await at file:///.../unsettled.mjs:1

import { setTimeout as sleep } from 'node:timers/promises';
import { Readable } from 'node:stream';

let running = 0;
let peak = 0;
const job = async (id) => {
  running++;
  peak = Math.max(peak, running);
  await sleep(40);
  running--;
  return id * 10;
};
const ids = [1, 2, 3, 4, 5, 6, 7, 8];

async function timed(label, fn) {
  running = 0; peak = 0;
  const t = performance.now();
  const out = await fn();
  console.log(`${label.padEnd(10)} ${String(Math.round((performance.now() - t) / 40) * 40).padStart(3)} ms  peak=${peak}  ${JSON.stringify(out)}`);
}

// Series: one at a time, results in order
await timed('series', async () => {
  const out = [];
  for (const id of ids) out.push(await job(id));
  return out;
});

// Parallel: everything at once (unbounded)
await timed('parallel', () => Promise.all(ids.map(job)));

// Limited concurrency: a fixed number of workers pulling from a shared iterator
async function mapLimit(items, limit, fn) {
  const results = new Array(items.length);
  const iterator = items.entries();
  async function worker() {
    for (const [index, item] of iterator) {
      results[index] = await fn(item);
    }
  }
  await Promise.all(Array.from({ length: Math.min(limit, items.length) }, worker));
  return results;
}
await timed('limit 3', () => mapLimit(ids, 3, job));

// The built-in alternative (Experimental): Readable.map with the concurrency option
await timed('stream x3', () => Readable.from(ids).map(job, { concurrency: 3 }).toArray());

// Fail fast vs collect everything
const flaky = async (id) => { await sleep(10); if (id % 4 === 0) throw new Error(`job ${id}`); return id; };
console.log(await Promise.allSettled(ids.map(flaky)).then((r) => r.filter((x) => x.status === 'rejected').map((x) => x.reason.message)));

series     320 ms  peak=1  [10,20,30,40,50,60,70,80]
parallel    40 ms  peak=8  [10,20,30,40,50,60,70,80]
limit 3    120 ms  peak=3  [10,20,30,40,50,60,70,80]
stream x3  120 ms  peak=3  [10,20,30,40,50,60,70,80]
[ 'job 4', 'job 8' ]

import { setTimeout as sleep } from 'node:timers/promises';

Promise.reject(new Error('nobody handled me'));
await sleep(50);
console.log('still running after the rejection');

file:///.../ur.mjs:3
Promise.reject(new Error('nobody handled me'));
               ^

Error: nobody handled me
    at file:///.../ur.mjs:3:16

Node.js v26.10.0