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

The Distributed Systems Fallacy for Single-Database Systems

A blockchain indexer has one primary constraint: write throughput to PostgreSQL. The bottleneck is always the database. Every architectural decision that adds latency between an incoming event and a database write is a decision that hurts throughput.

Consider what happens when you split a monolithic indexer into microservices:

Monolith: 
  TCP socket → parse → validate → write DB
  Latency: 2–5ms per transaction

Microservices:
  TCP socket → Ingestion Service → HTTP call → Validation Service
           → HTTP call → Transform Service → HTTP call → Write Service → write DB
  Latency: 2ms + 3ms + 3ms + 3ms + 2ms = 13ms per transaction
  Plus: 4× network hops, 4× serialization/deserialization

The throughput penalty: at 5ms vs 13ms per transaction, the microservice architecture has 2.6× lower throughput ceiling on the same hardware, before considering the operational complexity of 4 deployable services.

The Modulith is not a step backward. It is the architecturally correct choice when:

1. Your system writes to a single datastore
2. Your bottleneck is that datastore, not CPU
3. The subdomains need to coordinate at sub-millisecond latency
4. Your team can enforce module boundaries through code conventions rather than network boundaries

Domain Module Structure

A Modulith divides the application into domain modules with strict boundaries, but keeps them in the same process. The key discipline: a module may not import from another module's internals. It may only call the other module's public interface.

src/
├── modules/
│   ├── ingestion/
│   │   ├── index.ts          ← PUBLIC interface (only file other modules can import)
│   │   ├── parser.ts         ← PRIVATE
│   │   ├── validator.ts      ← PRIVATE
│   │   └── repository.ts     ← PRIVATE
│   ├── analytics/
│   │   ├── index.ts          ← PUBLIC
│   │   ├── aggregator.ts     ← PRIVATE
│   │   └── repository.ts     ← PRIVATE
│   ├── notifications/
│   │   ├── index.ts          ← PUBLIC
│   │   ├── websocket.ts      ← PRIVATE
│   │   └── emailer.ts        ← PRIVATE
│   └── ledger/
│       ├── index.ts          ← PUBLIC
│       ├── balance.ts        ← PRIVATE
│       └── repository.ts     ← PRIVATE
├── shared/
│   ├── events.ts             ← Domain event types (shared contracts)
│   ├── database.ts           ← Shared DB pool (single connection pool)
│   └── logger.ts             ← Shared logger
└── app.ts                    ← Wires modules together

Enforcing boundaries with ESLint:

javascript
// .eslintrc.js — enforce module boundaries
module.exports = {
  rules: {
    'no-restricted-imports': ['error', {
      patterns: [
        // Ingestion module cannot import from analytics internals
        { group: ['*/analytics/aggregator*'], message: 'Use analytics/index.ts public API' },
        { group: ['*/analytics/repository*'], message: 'Use analytics/index.ts public API' },
        // etc for each module
      ]
    }]
  }
};

This is the key discipline that makes a Modulith work: the network boundary in microservices is replaced by a code review + linting boundary. The cost of violating it is a failed lint check, not a runtime error — much cheaper to enforce.

The In-Process Event Bus: Zero-Latency Cross-Domain Communication

When the ingestion module writes a transaction, the analytics module needs to update its aggregations, the notification module needs to push WebSocket updates, and the ledger module needs to update balances. In microservices, this is three HTTP calls. In a Modulith, it is three in-process event emissions.

Option 1: Node.js EventEmitter (Synchronous)

javascript
// shared/events.ts — domain event definitions
export const domainEvents = new EventEmitter();

export const EVENTS = {
  TRANSACTION_INGESTED: 'transaction:ingested',
  BLOCK_CONFIRMED: 'block:confirmed',
  BALANCE_UPDATED: 'balance:updated',
} as const;

javascript
// modules/ingestion/index.ts
import { domainEvents, EVENTS } from '../../shared/events';

export async function ingestTransaction(raw: RawTransaction) {
  const transaction = await parseAndValidate(raw);
  await repository.write(transaction);

  // Synchronous in-process emission — no network, no serialization
  // All listeners run before this line returns
  domainEvents.emit(EVENTS.TRANSACTION_INGESTED, transaction);

  return transaction;
}

javascript
// modules/analytics/index.ts
import { domainEvents, EVENTS } from '../../shared/events';

// Register handler at startup
domainEvents.on(EVENTS.TRANSACTION_INGESTED, async (transaction) => {
  await updateAggregations(transaction);
});

The synchronous emission problem: EventEmitter.emit() runs all listeners synchronously before returning. If analytics and notifications are async (they issue database writes), your ingestion function waits for all of them. This couples ingestion latency to analytics latency.

Option 2: emittery for Async Event Handling

javascript
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:

javascript
// 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.

javascript
// 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,
});

javascript
// 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();
  }
}

javascript
// 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.

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.

javascript
// 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);

javascript
// 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.

javascript
// 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
# Dockerfile
FROM node:22-alpine
WORKDIR /app
COPY package*.json ./
RUN npm ci --production
COPY dist/ ./dist/
EXPOSE 3000
CMD ["node", "dist/app.js"]

yaml
# 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.

javascript
// 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:

javascript
// 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 →