Asynchronous Integration

Every number in the previous lesson was born from a single assumption: the caller waits until the successor’s work is done. Remove the wait, and both the round count and the dependency closure change. This lesson builds the same loan-issuing flow a second time; this time the notification and billing steps are pulled off the response path and turned into consumers of an event, and the two implementations are compared by the same measures.

Queue mechanics, delivery guarantees, back pressure, and competing consumers were built and measured in the Caching, Queues and Asynchronous Processing course; they are not repeated here. What is measured is not the carrier itself, but how the coupling on the two sides of the boundary changes.

SB3 — the carrier is modeled as an append log with fixed-interval polling. Rationale: since delivery and ordering guarantees are measured in a separate course, the carrier’s plainest form is enough here; the three quantities measured (units on the response path, units bound per field, the consistency window) are independent of the carrier’s type. The log is a real file, the polling runs in a real process; only the carrier’s plainness is a model.

Two Implementations, One Source

The loan service runs in two modes. In synchronous mode, it calls both its successors and waits for both responses. In event mode, it writes the same body to the log and returns. The difference is one line, but that line also changes the service’s configuration: in event mode the successor address list is empty.

// loan.mjs — <mode: synchronous|event> <port> <notification-port> <fee-port>; long-lived process
import { createServer } from "node:http";
import { appendFileSync } from "node:fs";
const [mode, port, pn, pf] = process.argv.slice(2);
if (!port) { console.log("usage: node loan.mjs <mode> <port> <notification-port> <fee-port>"); process.exit(1); }
const WORK = 25;                                   // duration of writing the loan record (ms)
const SUCCESSOR = mode === "synchronous" ? [pn, pf] : [];   // in event mode the publisher carries no successor address
let sequence = 0;

createServer(async (request, response) => {
  if (request.url === "/successors") { response.end(JSON.stringify({ successors: SUCCESSOR.length })); return; }
  await new Promise((c) => setTimeout(c, WORK));
  sequence += 1;
  const event = { type: "loan_issued", sequence: sequence, memberId: 7, copyId: 400 + sequence,
    overdueDays: 0, title: "Book loan issued" };
  const body = JSON.stringify(event);
  if (mode === "event") appendFileSync("events.log", `${body}\n`);
  else await Promise.all(SUCCESSOR.map((p) =>
    fetch(`http://127.0.0.1:${p}/work`, { method: "POST", body })));
  response.end(body);
}).listen(Number(port));

The notification and billing units also run from a single source and take on two roles at once: called as endpoints, they take the body from the request; run as consumers, they read the same body from the log. The work itself is the same function in both cases; only where the body comes from changes.

// unit.mjs — <name: notification|fee> <port> <poll: 0|1>; runs both as an endpoint and as a consumer
import { createServer } from "node:http";
import { appendFileSync, existsSync, readFileSync } from "node:fs";
const [name, port, poll] = process.argv.slice(2);
if (!port) { console.log("usage: node unit.mjs <name> <port> <poll>"); process.exit(1); }
const WORK = 25, INTERVAL = 200;                     // the unit's work and polling interval (ms)
const CONTRACT = {                               // the fields each unit reads from the event body
  notification: (o) => `notification "${o.title}" sent to member ${o.memberId}`,
  fee: (o) => `member ${o.memberId}, copy ${o.copyId}, ${o.overdueDays} day(s) overdue`,
};
const handle = async (o) => {
  await new Promise((c) => setTimeout(c, WORK));
  appendFileSync(`effect-${name}.log`, `${o.sequence} ${CONTRACT[name](o)}\n`);
};

createServer(async (request, response) => {
  let s = "";
  for await (const p of request) s += p;
  await handle(JSON.parse(s));
  response.end("done");
}).listen(Number(port));

let read = 0, busy = false;
if (poll === "1") setInterval(async () => {
  if (busy || !existsSync("events.log")) return;
  busy = true;
  const lines = readFileSync("events.log", "utf8").split("\n").filter(Boolean);
  while (read < lines.length) { await handle(JSON.parse(lines[read])); read += 1; }
  busy = false;
}, INTERVAL);

The Measurement Tool

The tool does two jobs. In run mode, it sends three requests to find the round count, then waits for effects to settle and sends one more request to see whether the new effect is visible the moment the response returns. It repeats the same observation 600 milliseconds later. In contract mode no process is up at all; only two source files are read, and how many deployment units each field in the event body binds is counted.

// measure.mjs — <mode: synchronous|event|contract>; the first two modes run while the processes are up
import { existsSync, readFileSync } from "node:fs";
const WORK = 25, UNIT_PORT = { loan: 8951 }, UNIT = ["notification", "fee"];
const wait = (ms) => new Promise((c) => setTimeout(c, ms));
const effect = () => UNIT.reduce((s, a) => s + (existsSync(`effect-${a}.log`)
  ? readFileSync(`effect-${a}.log`, "utf8").split("\n").filter(Boolean).length : 0), 0);
const requestLoan = () => fetch(`http://127.0.0.1:${UNIT_PORT.loan}/work`).then((y) => y.text());
const [mode] = process.argv.slice(2);

if (mode === "synchronous" || mode === "event") {
  const { successors } = await (await fetch(`http://127.0.0.1:${UNIT_PORT.loan}/successors`)).json();
  const durations = [];
  for (let i = 0; i < 3; i += 1) {
    const t = performance.now();
    await requestLoan();
    durations.push(performance.now() - t);
  }
  const rounds = Math.floor(durations.sort((a, b) => a - b)[1] / WORK);
  await wait(600);
  const before = effect();
  await requestLoan();
  const immediate = effect() > before;                   // is the new effect visible the moment the response returns
  await wait(600);
  const after = effect() > before;
  console.log(`${mode.padEnd(14)}${String(rounds).padStart(18)}${String(1 + successors).padStart(24)}` +
    `${String(successors).padStart(28)}${(immediate ? "yes" : "no").padStart(20)}` +
    `${(after ? "yes" : "no").padStart(14)}${`${effect()}/8`.padStart(14)}`);
} else if (mode === "contract") {
  const eventBlock = readFileSync("loan.mjs", "utf8").match(/const event = \{([\s\S]*?)\};/)[1];
  const fields = [...eventBlock.matchAll(/(\w+):/g)].map((m) => m[1]);
  const source = readFileSync("unit.mjs", "utf8");
  const consumer = {};
  for (const m of source.matchAll(/^ {2}(\w+): \(o\) => (.*)$/gm)) {
    consumer[m[1]] = [...m[2].matchAll(/o\.(\w+)/g)].map((x) => x[1]);
  }
  const shared = [...source.replace(/^ {2}\w+: \(o\) => .*$/gm, "").matchAll(/o\.(\w+)/g)]
    .map((x) => x[1]);
  const consumerNames = Object.keys(consumer);
  console.log(`${"event field".padEnd(14)}${"publisher".padStart(11)}${"reading consumers".padStart(19)}` +
    `${"shared code".padStart(13)}${"bound deployment units".padStart(24)}`);
  let total = 0;
  for (const a of fields) {
    const reading = consumerNames.filter((t) => consumer[t].includes(a));
    const o = shared.includes(a) ? 1 : 0;
    const bound = 1 + (o ? consumerNames.length : reading.length);
    total += bound;
    console.log(`${a.padEnd(14)}${"1".padStart(11)}${String(reading.length).padStart(19)}` +
      `${String(o).padStart(13)}${String(bound).padStart(24)}`);
  }
  console.log(`${fields.length}-field event body, ${consumerNames.length} consumer units; ` +
    `average bound units per field ${(total / fields.length).toFixed(2)}`);
  console.log(`field no consumer reads: ` +
    `${fields.filter((a) => consumerNames.every((t) => !consumer[t].includes(a)) && !shared.includes(a)).join(", ")}`);
}

rm -f events.log effect-notification.log effect-fee.log
printf '%-14s%18s%24s%28s%20s%14s%14s\n' mode "response rounds" \
  "units on response path" "successors publisher knows" "effect at response" "600 ms later" "effect lines"
node unit.mjs notification 8952 0 &
node unit.mjs fee 8953 0 &
node loan.mjs synchronous 8951 8952 8953 &
sleep 1
node measure.mjs synchronous
pkill -f "node unit.mjs" ; pkill -f "node loan.mjs" ; sleep 0.3
rm -f effect-notification.log effect-fee.log

node unit.mjs notification 8952 1 &
node unit.mjs fee 8953 1 &
node loan.mjs event 8951 8952 8953 &
sleep 1
node measure.mjs event
pkill -f "node unit.mjs" ; pkill -f "node loan.mjs" ; sleep 0.3
echo
node measure.mjs contract

mode             response rounds  units on response path  successors publisher knows  effect at response  600 ms later  effect lines
synchronous                    2                       3                           2                 yes           yes           8/8
event                          1                       1                           0                  no           yes           8/8

event field     publisher  reading consumers  shared code  bound deployment units
type                    1                  0            0                       1
sequence                1                  0            1                       3
memberId                1                  2            0                       3
copyId                  1                  1            0                       2
overdueDays             1                  1            0                       2
title                   1                  1            0                       2
6-field event body, 2 consumer units; average bound units per field 2.17
field no consumer reads: type

Work That Drops Off the Response Path

The first two columns give the gain. In the synchronous implementation, the response takes two rounds: writing the loan record and the successors’ work. In the event-based implementation, the same request returns in one round, because all that is left on the response path is appending one line to the log. The last column shows that the work is not lost: in both implementations, four requests produce eight effect lines.

The third column gives the real size of the gain. The number of deployment units on the response path drops from three to one. Read against the previous lesson’s measure, the meaning is clear: the notification and billing units have left the loan entry point’s dependency closure. In the synchronous implementation, either of these two units going down drops the loan request; in the event-based implementation, that outage never shows up on the response path at all.

The Contract Spreads

The fourth column gives the first half of the cost. In the synchronous implementation, the publisher carries two successor addresses: it knows whom it calls, and that is readable at the call site. In event mode this number is zero. The publisher no longer knows its consumers; this is exactly the loose coupling that was wanted, and exactly the source of the contract burden.

The second table turns this burden into a number. The event body has six fields, and each field binds an average of 2.17 deployment units. Both consumers read the member identifier: when the field’s name or meaning changes, three units at once must be touched. The sequence number does not appear in any consumer’s contract line, but it is read in the shared handling code, so it, too, binds three units.

The last row is the characteristic problem of asynchronous integration: no consumer reads the type field. The publisher has to defend every field it puts in the body forever, because it cannot see which fields are actually read. In a synchronous call, this information was readable at the call site; on the other side of a log, it is not. Contract evolution and consumer protection are addressed in this course’s final topic.

The Consistency Window

The fifth and sixth columns show the new failure mode. In the synchronous implementation, the effect is already in place the moment the response returns: the caller returns knowing the fee row has been written. In the event-based implementation, there is no effect the moment the response returns, and there is one 600 milliseconds later. The gap between the two is the consistency window, and in this setup its upper bound is known: the polling interval plus the consumer’s work duration, that is, 225 milliseconds. The measurement confirms this bound from both ends — no effect at the start of the window, an effect at the end.

Inside the window, the system is inconsistent, and the inconsistency is observable: a request reading the member’s account during that interval sees the loan record but not the fee row. This is a state that does not exist in the synchronous implementation, and it cannot be removed — only narrowed and accommodated in the interface. The second new failure mode comes from the same place: while a consumer is down, events pile up in the log, response time is never disturbed, and the failure never shows up on the response path. The problem is noticed only if someone is watching the accumulating events.

The triple resolves like this. What got cheaper: the response round dropped from 2 to 1, and the units on the response path dropped from 3 to 1; two consumer units left the dependency closure. What got more expensive: the contract spread across three deployment units, an average of 2.17 units are bound per field, and because the publisher’s known successor count fell to zero, which field is actually read is no longer visible from the code. The failure mode that was born: the consistency window, and a consumer failure that is invisible on the response path.

Summary

• When the same workflow is built as event-based, the response round drops from 2 to 1 and the deployment units on the response path drop from 3 to 1; four requests produce 8 effect lines in both implementations.
• The publisher’s known successor count drops from 2 to 0; this number is both the source of loose coupling and the source of the contract burden.
• In the six-field event body, an average of 2.17 deployment units are bound per field; the member identifier binds three units at once, and no consumer reads the type field.
• The consistency window was confirmed by measurement from both ends: no effect at response time, an effect 600 ms later; the upper bound is the polling interval plus the consumer’s work.
• Asynchronous integration removes consumer failure from the response path; the failure no longer shows up in response time — it shows up in accumulating events.

Next Step

Over two lessons, both sides of the boundary have been worked: where the boundary should pass, how to call across it, and turning some calls into events instead of making them at all. One side has never been addressed in any of these arrangements — the outside. The client still knows every service’s address individually, each service carries its own identity check in its own code, and the same rate-limit rule sits written in more than one place. The next lesson counts this repetition: how many lines in how many services one edge responsibility is repeated as, how many files and endpoints change when these responsibilities move to a single place, and how many hops are added per request.