Module 8 — The Modern Hybrid Monolith: High-Throughput Modulith Architecture

The real problems with plain EventEmitter for this use case are different, and easy to miss:

• Unhandled rejections nobody catches. Because emit() never awaits what a listener returns, if that async listener throws after its first await, the rejection has no .catch() anywhere — it surfaces as an unhandledRejection with no obvious link back to the transaction that triggered it.
• No backpressure or completion tracking. EventEmitter has no API to ask "have all listeners for this event finished?" and no way to slow ingestion down if a listener falls behind. Every emission is fire-and-forget, by construction, not by choice.

These — not blocking — are the actual reasons to reach for emittery or a more structured event bus below.

Option 2: emittery for Async Event Handling

import Emittery from 'emittery';
 
// Fully async emitter — emit() returns immediately
// Listeners run asynchronously without blocking the emitter
const domainEvents = new Emittery<{
  'transaction:ingested': Transaction;
  'block:confirmed': Block;
  'balance:updated': BalanceUpdate;
}>();
 
export async function ingestTransaction(raw: RawTransaction) {
  const transaction = await parseAndValidate(raw);
  await repository.write(transaction);
 
  // Non-blocking: listeners are scheduled, not awaited
  void domainEvents.emit('transaction:ingested', transaction);
 
  return transaction;  // returns without waiting for listeners
}
 
// Analytics listener: runs asynchronously, does not block ingestion
domainEvents.on('transaction:ingested', async (transaction) => {
  await updateAggregations(transaction);
});
 
// Notification listener: runs asynchronously
domainEvents.on('transaction:ingested', async (transaction) => {
  await notifyWebSocketSubscribers(transaction);
});

The void on domainEvents.emit(...) is intentional — the ingestion module does not care when listeners complete, only that it fires the event. Ingestion latency is decoupled from analytics and notification latency.

Handling Backpressure on the Event Bus

If analytics is slow (processing 5,000 events/sec) and ingestion is fast (emitting 50,000 events/sec), the analytics listener builds up a queue:

// Add bounded queue with backpressure to async event listener
const analyticsQueue = new BoundedQueue(10_000);  // max 10K pending
 
domainEvents.on('transaction:ingested', (transaction) => {
  if (!analyticsQueue.push(transaction)) {
    // Queue full — analytics can't keep up
    // Options: drop, log, or apply backpressure upstream
    metrics.increment('analytics_queue_overflow');
  }
});
 
// Analytics worker drains from the bounded queue
async function analyticsWorker() {
  for await (const transaction of analyticsQueue) {
    await updateAggregations(transaction);
  }
}

---

Module Isolation: Own Your Database Connection Pool

Each domain module should manage its own database interactions. The shared connection pool is an infrastructure concern — the module's repository is the domain concern.

// shared/database.ts — single connection pool, shared infrastructure
import pg from 'pg';
 
export const pool = new pg.Pool({
  connectionString: process.env.DATABASE_URL,
  max: 50,           // total connections across all modules
  idleTimeoutMillis: 30_000,
});

// modules/ingestion/repository.ts — ingestion's private DB access
import { pool } from '../../shared/database';
 
export async function writeTransaction(tx: Transaction) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    await client.query(
      'INSERT INTO transactions (hash, block_height, sender, amount) VALUES ($1, $2, $3, $4)',
      [tx.hash, tx.blockHeight, tx.sender, tx.amount]
    );
    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

// modules/analytics/repository.ts — analytics' private DB access
import { pool } from '../../shared/database';
 
export async function updateAggregations(tx: Transaction) {
  await pool.query(
    'INSERT INTO hourly_stats (hour, total_volume, tx_count) VALUES (...) ON CONFLICT DO UPDATE ...',
    [...]
  );
}

Both modules share the pool for efficiency, but their SQL is encapsulated in their own repositories. Neither module can "accidentally" call the other module's queries.

---

Cross-Module Transactional Consistency: The Unit-of-Work Problem

Sometimes a single business operation must update two modules' data atomically. A payment that debits the ledger and writes an ingestion record either both happen or neither does — there is no acceptable partial state where the ledger is debited but the transaction was never recorded, or vice versa.

Firing two independent transactions — one against the ledger repository, one against the ingestion repository — leaves exactly that gap: if the process crashes between them, you have a debited account with no corresponding ingested transaction. Since both modules share the same physical PostgreSQL instance, the fix doesn't require distributed transactions — it requires a shared unit-of-work that both modules' repositories can write through:

// shared/unit-of-work.ts — a transactional context passed across module boundaries
import { pool } from './database';
 
export async function withTransaction<T>(work: (client: PoolClient) => Promise<T>): Promise<T> {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    const result = await work(client);
    await client.query('COMMIT');
    return result;
  } catch (err) {
    await client.query('ROLLBACK');
    throw err;
  } finally {
    client.release();
  }
}

// modules/ledger/index.ts — public interface accepts an externally-managed client
export async function debit(client: PoolClient, accountId: string, amount: bigint) {
  await client.query(
    'UPDATE accounts SET balance = balance - $1 WHERE id = $2',
    [amount, accountId]
  );
}

// modules/ingestion/index.ts — same shape: accepts the shared client
export async function writeWithClient(client: PoolClient, tx: Transaction) {
  await client.query(
    'INSERT INTO transactions (hash, block_height, sender, amount) VALUES ($1, $2, $3, $4)',
    [tx.hash, tx.blockHeight, tx.sender, tx.amount]
  );
}

// app.ts (composition root) — the only place allowed to coordinate two modules' writes
import { withTransaction } from './shared/unit-of-work';
import * as ledger from './modules/ledger';
import * as ingestion from './modules/ingestion';
 
export async function processDebitedIngestion(tx: RawTransaction) {
  return withTransaction(async (client) => {
    await ledger.debit(client, tx.senderAccountId, tx.amount);
    await ingestion.writeWithClient(client, tx);
    // Both writes commit together, or neither does — no partial state
  });
}

Each module still owns its own SQL — the boundary isn't broken, only the transaction is shared, and only the composition root (never another module) is allowed to coordinate across two modules' public interfaces this way.

This only works because both modules live in the same database and process. Once a module is extracted to its own datastore or service (Module 9), atomic two-writes-in-one-transaction is no longer available — that's when you move to a saga: one module's write is the transaction of record, it emits a domain event, and the second module's write happens in an event handler with an explicit compensating action if it fails. Module 11 covers this exact pattern for the Payment aggregate — a single transaction should only ever mutate one aggregate directly, with any second aggregate updated via a domain event.

Propagating Context Across Module Boundaries with AsyncLocalStorage

A single incoming transaction flows through several modules — ingestion validates and writes it, analytics aggregates it off an event, notifications push it to subscribers. Threading a requestId or trace ID through every function signature so your logs correlate across modules is tedious, and it's the kind of thing that silently rots the first time someone forgets to pass it along. Node's built-in AsyncLocalStorage propagates that context implicitly through the async call chain, without touching every function signature.

// shared/context.ts
import { AsyncLocalStorage } from 'node:async_hooks';
 
interface RequestContext {
  requestId: string;
  traceId: string;
}
 
export const requestContext = new AsyncLocalStorage<RequestContext>();
 
export function getRequestId(): string | undefined {
  return requestContext.getStore()?.requestId;
}

// modules/ingestion/index.ts — enter the context once, at the edge
import { requestContext } from '../../shared/context';
import { randomUUID } from 'node:crypto';
 
export async function ingestTransaction(raw: RawTransaction) {
  return requestContext.run({ requestId: randomUUID(), traceId: raw.traceId }, async () => {
    const transaction = await parseAndValidate(raw);
    await repository.write(transaction);
    domainEvents.emit(EVENTS.TRANSACTION_INGESTED, transaction);
    return transaction;
  });
}

// modules/analytics/index.ts — the context survives the event bus hop and any awaited I/O
import { getRequestId } from '../../shared/context';
 
domainEvents.on(EVENTS.TRANSACTION_INGESTED, async (transaction) => {
  logger.info({ requestId: getRequestId() }, 'updating aggregations');
  await updateAggregations(transaction);
});

The one gap: AsyncLocalStorage propagates through the continuation chain of promises, timers, and EventEmitter callbacks that Node's async_hooks track — but if a listener hands work off to a separate structure like the BoundedAsyncQueue below, the context does not automatically travel with the queued item. Capture it explicitly at push time (const store = requestContext.getStore(); queue.push({ item, store })) and re-enter it with requestContext.run(store, () => ...) when the worker dequeues it.

---

Memory Boundary Management: Preventing Unbounded Queues

In a Modulith, the biggest risk is an async event listener that cannot keep up with the emitter, causing the intermediate queue to grow without bound. This is the in-process equivalent of the external OOM from Module 4.

// Monitoring module queue depths in production
const queueDepths = {
  analytics: 0,
  notifications: 0,
  ledger: 0,
};
 
// Register high-water-mark gauges
setInterval(() => {
  for (const [module, depth] of Object.entries(queueDepths)) {
    queueDepthGauge.labels(module).set(depth);
    if (depth > 50_000) {
      logger.warn(`${module} queue is backing up: ${depth} events pending`);
    }
  }
}, 5_000);

// Bounded async queue implementation
class BoundedAsyncQueue<T> {
  #queue: T[] = [];
  #maxSize: number;
  #resolvers: Array<(value: T) => void> = [];
 
  constructor(maxSize: number) {
    this.#maxSize = maxSize;
  }
 
  push(item: T): boolean {
    if (this.#resolvers.length > 0) {
      const resolve = this.#resolvers.shift()!;
      resolve(item);
      return true;
    }
    if (this.#queue.length >= this.#maxSize) return false;  // bounded
    this.#queue.push(item);
    return true;
  }
 
  async pop(): Promise<T> {
    if (this.#queue.length > 0) return this.#queue.shift()!;
    return new Promise(resolve => this.#resolvers.push(resolve));
  }
 
  get size() { return this.#queue.length; }
 
  async *[Symbol.asyncIterator]() {
    while (true) yield await this.pop();
  }
}

---

Startup Wiring: The Composition Root

The app.ts file is the only place where modules are wired together. It knows about all modules; modules know nothing about each other except through the event bus.

// app.ts — composition root
import Fastify from 'fastify';
import { pool } from './shared/database';
import { domainEvents } from './shared/events';
 
// Import each module's public interface
import { createIngestionModule } from './modules/ingestion';
import { createAnalyticsModule } from './modules/analytics';
import { createNotificationModule } from './modules/notifications';
import { createLedgerModule } from './modules/ledger';
 
async function bootstrap() {
  const app = Fastify({ logger: true });
 
  // Initialize modules — each registers its own event listeners
  const ingestion = createIngestionModule({ pool, events: domainEvents });
  const analytics = createAnalyticsModule({ pool, events: domainEvents });
  const notifications = createNotificationModule({ events: domainEvents });
  const ledger = createLedgerModule({ pool, events: domainEvents });
 
  // Register HTTP routes from each module
  await app.register(ingestion.routes, { prefix: '/api/v1/ingest' });
  await app.register(analytics.routes, { prefix: '/api/v1/analytics' });
  await app.register(notifications.routes, { prefix: '/api/v1/notify' });
 
  // Graceful shutdown — drain in-flight events before closing
  process.on('SIGTERM', async () => {
    await app.close();
    await ingestion.drain();
    await analytics.drain();
    await pool.end();
    process.exit(0);
  });
 
  await app.listen({ port: 3000 });
}
 
bootstrap().catch(console.error);

---

Deploying a Modulith: One Binary, One Pod

The deployment simplicity is the Modulith's most underrated advantage:

# Dockerfile
FROM node:22-alpine
WORKDIR /app
COPY package*.json ./
RUN npm ci --production
COPY dist/ ./dist/
EXPOSE 3000
CMD ["node", "dist/app.js"]

# kubernetes/deployment.yaml
apiVersion: apps/v1
kind: Deployment
spec:
  replicas: 8          # 8 identical pods, each running the full modulith
  template:
    spec:
      containers:
      - name: indexer
        image: indexer:latest
        resources:
          requests: { memory: "512Mi", cpu: "500m" }
          limits: { memory: "2Gi", cpu: "2000m" }

Compare to 8 pods per service × 5 services = 40 pods to manage, monitor, and scale. The modulith runs the same workload in 8 pods.

---

The Migration Path: Modulith → Microservice

When a single module genuinely needs to scale independently (the analytics module needs 10× the compute of ingestion), the modulith's clean boundaries make extraction straightforward:

Step 1: The analytics module already has:
  - A clean public interface (index.ts)
  - Its own repository (no shared SQL with other modules)
  - Its own event listener (no direct function calls from ingestion)
 
Step 2: Wrap the analytics module in a separate Node.js process
  - Create analytics-service/ with its own app.ts
  - Replace in-process event listener with a Kafka consumer
  - Replace domain event emission in ingestion with Kafka producer
 
Step 3: Replace the in-process event emission in app.ts
  - Before: void domainEvents.emit('transaction:ingested', tx)
  - After: await kafkaProducer.send({ topic: 'transactions', value: tx })
 
Step 4: Deploy analytics-service as a separate pod

The domain boundary was already clean — extraction is mechanical, not architectural. This is the Modulith's long-term value: it lets you defer the operational complexity of microservices until the performance need actually justifies it.

---

Production Incident: Unbounded Event Queue OOM

Context: A blockchain indexer Modulith processing 8,000 transactions/second. The notification module sent WebSocket updates to 12,000 connected subscribers.

What happened:

At peak load during a token launch, WebSocket subscriber count jumped to 180,000. The notification module's event listener became the slowest module — sending to 180,000 subscribers took ~200ms per transaction. Ingestion was emitting 8,000 events/sec; notifications could only process 5 events/sec.

// The broken pattern
domainEvents.on('transaction:ingested', async (transaction) => {
  // Sending to 180,000 subscribers takes 200ms
  await Promise.all(
    subscribers.map(sub => sub.send(JSON.stringify(transaction)))
  );
  // 8,000 events queued per second, 5 processed per second
  // After 1 minute: 480,000 events in memory = ~960MB
  // After 3 minutes: OOM
});

The fix — bounded queue with subscriber batching:

// Step 1: bound the notification queue
const notifQueue = new BoundedAsyncQueue(5_000);
 
domainEvents.on('transaction:ingested', (transaction) => {
  if (!notifQueue.push(transaction)) {
    // Queue full — skip notification for this transaction
    // Subscribers get the next one; no data loss for the indexer
    metrics.increment('notification_skipped');
  }
});
 
// Step 2: batch subscribers into groups of 1,000
async function notificationWorker() {
  for await (const transaction of notifQueue) {
    const BATCH_SIZE = 1_000;
    const payload = JSON.stringify(transaction);
 
    // Send in parallel batches to avoid blocking for 200ms
    for (let i = 0; i < subscribers.size; i += BATCH_SIZE) {
      const batch = Array.from(subscribers).slice(i, i + BATCH_SIZE);
      await Promise.all(batch.map(sub => sub.send(payload).catch(() => {})));
    }
  }
}

After the fix: queue depth stabilized at 200–800. Memory: constant 150MB. 12,000 skipped notifications over 10 minutes of peak load — acceptable for a "best-effort" notification system.

---

Summary

---

The Modulith works until a single module needs to scale 10× independently, or until you need geographic distribution. Module 9 covers the precise moment to split — how to identify the correct seam, execute the extraction without data loss, and manage the operational split-plane.

Next: Module 9 — Pragmatic Microservice Deconstruction: Splitting Ingestion from Analytics →

---