阅读中文版本

Build a Change Data Capture Pipeline

Change Data Capture Pipeline — a complete walkthrough of change data capture, covering log tail reader, change decoder, schema projector, and testing.

change_data_capture/
├── src/index.ts
├── src/log-tail-reader.ts
├── src/change-decoder.ts
├── tests/change_data_capture.test.ts
└── package.json

// Log tail reader: change data capture core
export function logtailreader(input: LogtailreaderInput): LogtailreaderResult {
  // Build the log tail reader that anchors change data capture and defines the core data shape.
  return processLogtailreader(input);
}

// Change decoder — handles change data capture Add the change decoder so
export class Changedecoder {
  constructor(private opts: ChangedecoderOptions) {}
  async run(ctx: Context): Promise<ChangedecoderOutput> {
    return this.execute(ctx);
  }
}

// Schema projector: change data capture stage
export const schemaprojector = (next: Next) => async (req: Request) => {
  const out = await transform(req);
  return next(out);
};

// Sink writer for change data capture
export async function sinkwriter(stream: AsyncIterable<Chunk>): Promise<Aggregate> {
  let acc = init();
  for await (const c of stream) acc = merge(acc, c);
  return acc;
}

// Offset tracker: change data capture finalization
export function offsettracker(state: State): Result {
  validate(state);
  return finalize(state);
}

Project overview

This project builds a change data capture system whose reason for existing is streaming row changes from a transaction log so downstream systems stay in sync without polling. The temptation in change data capture is to bolt the interesting part onto a working prototype and discover the hard cases in production; this build inverts that order and treats a schema change in the source that the projector does not know about and that corrupts the sink as the first thing designed, not the last.

You will implement Log tail reader, Change decoder, Schema projector, Sink writer, and Offset tracker, each as a module with one responsibility and a test that proves it in isolation. The change data capture pipeline is then the wiring that composes them, and the integration test is where the composition is shown to survive a schema change in the source that the projector does not know about and that corrupts the sink.

The walkthrough is written for a developer who has used a change data capture system but has not had to make one honest. It favors explicit data shapes over clever abstractions, and it names the failure each design choice prevents rather than presenting the choice as a style preference.

Each phase ends with something observable — a log tail reader that returns the right shape, a change decoder that rejects bad input, an integration test that goes green — so you can stop at any point and still have a change data capture system that works for the cases covered so far.

What you will build

The finished change data capture system exposes five capabilities, each with a focused test:

• Log tail reader
• Change decoder
• Schema projector
• Sink writer
• Offset tracker

Log tail reader is the foundation that every later stage assumes; Change decoder is where change data capture correctness lives; Schema projector handles a schema change in the source that the projector does not know about and that corrupts the sink; Sink writer makes the system observable; and Offset tracker is the graceful-degradation layer that keeps overload from becoming corruption.

The build keeps the change data capture logic in ordinary modules you can read in one sitting, and pulls in dependencies only for the parts that are genuinely tedious to hand-roll. That keeps the diff between the tutorial and a real codebase small, and keeps the failure modes visible in code rather than hidden behind a framework.

Architecture and data model

The system is organized around Log tail reader, Change decoder, and Schema projector, with Sink writer tying them together. Data flows as a stream of change events read from a transaction log, projected against a schema and written to a sink with an offset checkpoint.

Each component is a module with a pure interface where practical: the same input produces the same output, and side effects are pushed to the edges. That discipline is what makes the phases independently testable, because a test for Schema projector does not need to stand up Log tail reader and Change decoder to exercise the interesting branch.

The data model is deliberately explicit. A stream of change events read from a transaction log, projected against a schema and written to a sink with an offset checkpoint Keeping the shape written down — in a schema, a type, or a fixture — means a change to the shape is a visible diff rather than a silent runtime shift, and it means the integration test can assert against a concrete shape instead of guessing.

Configuration and code are kept apart: the configuration says what change data capture should do, the modules say how. A reviewer can change a threshold or a policy without reading the implementation, which keeps the risky changes small and lets the diff stand as the review.

The failure path through Schema projector is designed before the happy path is finished, because a schema change in the source that the projector does not know about and that corrupts the sink is the case that distinguishes a change data capture system from a demo. Every component declares what it does when its inputs are wrong, and the integration test asserts each of those declarations.

Prerequisites and dependencies

Have these ready before you install anything, so the change data capture setup is not blocked by a missing tool:

• Node.js 20 LTS or later
• Familiarity with change data capture fundamentals
• A local environment to run the test suite

The runtime dependencies are chosen to keep the change data capture datapath legible:

• pg (^8.13.0) — Postgres logical decoding
• redis (^4.7.0) — Offset store

# setup for change-data-capture
pnpm init
pnpm add pg redis
pnpm add -D typescript tsx vitest

Versions are pinned to a major range so the change data capture install is reproducible; commit the lockfile. The log tail reader reads its operational parameters from the environment and validates them at startup, so a missing value fails the boot rather than the first change data capture request.

Directory structure

Keep each change data capture component in its own file so it can be tested without starting the others:

change_data_capture/
├── src/index.ts
├── src/log-tail-reader.ts
├── src/change-decoder.ts
├── tests/change_data_capture.test.ts
└── package.json

Phase 1: Log tail reader

Build the log tail reader that anchors change data capture and defines the core data shape.

// Log tail reader: change data capture core
export function logtailreader(input: LogtailreaderInput): LogtailreaderResult {
  // Build the log tail reader that anchors change data capture and defines the core data shape.
  return processLogtailreader(input);
}

After this phase the log tail reader works in isolation and is covered by a unit test. The log tail reader defines the data shape that every later stage assumes, so getting it right first means the later phases compile against a stable contract.

Treat the log tail reader as the single source of truth for the change data capture data shape. Validate inputs at the boundary and reject early, so a malformed value never reaches Change decoder or Schema projector where its effect would be harder to localize.

Phase 2: Change decoder

Add the change decoder so change data capture can streaming row changes from correctly.

// Change decoder — handles change data capture Add the change decoder so
export class Changedecoder {
  constructor(private opts: ChangedecoderOptions) {}
  async run(ctx: Context): Promise<ChangedecoderOutput> {
    return this.execute(ctx);
  }
}

After this phase the change decoder works in isolation and is covered by a unit test. This is where change data capture correctness lives: the change decoder is what makes streaming row changes from a transaction log so downstream systems stay in sync without polling true, and its test is the proof.

Keep the change decoder pure where you can. A function that takes an input and returns an output is trivial to test and impossible to mock wrong; a class with hidden state is the opposite. Push side effects to the caller and keep the change decoder about the transformation.

Phase 3: Schema projector

Wire the schema projector to handle a schema change in without losing data.

// Schema projector: change data capture stage
export const schemaprojector = (next: Next) => async (req: Request) => {
  const out = await transform(req);
  return next(out);
};

After this phase the schema projector works in isolation and is covered by a unit test. The schema projector is the difference between a change data capture system that works on the happy path and one that survives a schema change in the source that the projector does not know about and that corrupts the sink.

Design the failure before the success. Decide what schema projector does when a schema change in the source that the projector does not know about and that corrupts the sink: reject, retry, shed, or degrade — and write that test first. A change data capture system whose failure behavior is unspecified will pick one at random under pressure, and it is usually the wrong one.

Phase 4: Sink writer

Integrate the sink writer to make change data capture observable and testable end to end.

// Sink writer for change data capture
export async function sinkwriter(stream: AsyncIterable<Chunk>): Promise<Aggregate> {
  let acc = init();
  for await (const c of stream) acc = merge(acc, c);
  return acc;
}

After this phase the sink writer works in isolation and is covered by a unit test. Observability is not a finishing touch; it is how you know the change data capture system is doing what it claims in production.

Emit one structured event per meaningful action, with enough context to localize a failure without a second round-trip. The sink writer is what turns “something is slow” into “the change decoder is slow for inputs matching this shape,” which is the difference between an alert and an investigation.

Phase 5: Offset tracker

Finalize the offset tracker so the change data capture system degrades gracefully under overload.

// Offset tracker: change data capture finalization
export function offsettracker(state: State): Result {
  validate(state);
  return finalize(state);
}

After this phase the offset tracker works in isolation and is covered by a unit test. The offset tracker is what keeps overload from becoming corruption: when the change data capture system is asked for more than it can do, it degrades in a defined way instead of failing in an undefined one.

Pick a degradation strategy for offset tracker — shed, queue, or backpressure — and make it explicit in code and in a test. Leaving the behavior to chance is how a change data capture traffic spike becomes a data loss incident.

Testing

Each phase has a focused test: log tail reader, change decoder, and schema projector are exercised in isolation, then an integration test drives the full change data capture pipeline and asserts the a schema change in case is handled.

Start with unit tests that pin each component’s contract: the log tail reader returns the documented shape, the change decoder transforms correctly, and the schema projector fails in the documented way. These tests are fast and they localize regressions to the component that broke.

Then add an integration test that composes the components and drives a realistic change data capture scenario end to end. Assert the happy path first, then deliberately trigger a schema change in the source that the projector does not know about and that corrupts the sink and assert the system degrades rather than corrupts. Because the components are pure where practical, the integration test stays deterministic rather than flaky.

A change data capture suite must also prove the negative: bad input is rejected at the boundary, overload is shed rather than queued indefinitely, and a partial failure does not leave the system in an inconsistent state. These are the tests that catch the regressions a happy-path suite never would.

The entire change data capture suite runs with no external service; every dependency is either in-process or stubbed, so every case is reproducible in CI. That is what lets the schema projector failure-mode assertions live as tests rather than as a manual checklist that drifts out of date.

# run the change data capture test suite
pnpm vitest run

Deployment and operations

Run at least two change data capture instances behind a load balancer and share state in Redis. Set resource limits so change data capture overload surfaces as backpressure rather than failure.

Run at least two stateless instances behind a load balancer, and move any shared state into a shared store so the change data capture behavior is global rather than per instance. Size limits so the system sheds load before it runs out of memory, because a change data capture system that OOMs under load fails in a way that is hard to reason about.

Drain on deploy: stop accepting new work, let in-flight change data capture operations finish, then exit. Because the system holds no state that cannot be reconstructed, a rollout is a matter of cycling instances one at a time while watching the error rate.

When a dependency is unavailable, decide explicitly whether the change data capture system fails closed or degrades. Leaving that choice to chance is how a dependency outage becomes a silent data loss, so make the choice visible in code and in a test.

Security considerations

Validate all change data capture inputs at the boundary, reject oversized payloads before buffering them, and keep secrets in the environment rather than the repository.

The change data capture system is a choke point, so it is the right place to reject malformed input before it reaches change decoder or schema projector. Validate the shape and the size at the edge, and let the inner components assume what they receive is well-formed.

Never log secrets or full payloads; log enough to localize a failure without recording data that should not be on disk. Treat any external call as untrusted and validate its response before using it, because a change data capture system that trusts its inputs blindly is one malformed response away from corruption.

Keep dependencies pinned and run an audit in CI. A transitive dependency with a known vulnerability is still a vulnerability in your change data capture system, and the lockfile is what makes the audit meaningful.

Observability

Emit one structured event per change data capture operation with the inputs’ shape, the outcome, and a correlation id. Track error rate and latency per component and alert on a step change rather than a single spike.

The correlation id is the thread that ties a user-visible symptom to the change data capture decision and to the component log entry for the same operation. Without it, an investigation starts from a timestamp and a guess.

Watch the schema projector failure rate separately from the overall error rate. A jump in schema projector failures usually means a schema change in the source that the projector does not know about and that corrupts the sink, not a bug in change decoder, and the two have different owners and different fixes.

Chart the share of change data capture operations that shed, queue, or succeed. A change in the mix points at the cause faster than an aggregate latency can, because each class has a different root.

Troubleshooting

A schema change in the source that the projector does not know about and that corrupts the sink. Reproduce with a minimal fixture, then add a guard test so the change data capture system rejects the condition loudly.

Performance regresses under change data capture load. Profile with the sampling analyzer, locate the hot path in change decoder, and add a targeted benchmark to lock the regression.

Keep the change data capture dashboards separate from dependency dashboards. When an operation fails, the correlation id lets you decide in seconds whether the fault is log tail reader, change decoder, or schema projector, instead of guessing from a timeout.

Limitations and next steps

• Single-region; multi-region replication is out of scope.
• The log tail reader is in-process and not horizontally shared by default.
• No auth layer; assume deployment behind an edge that enforces identity.

The change data capture system is single-region by design; extending to multi-region means making log tail reader shareable across regions, which is a real change to the consistency model rather than a config flag. The schema projector is in-process, which is correct for one instance but must move to a shared store before scaling out.

There is no authentication or authorization layer; the system assumes it is deployed behind an edge that enforces identity. Adding one is straightforward but out of scope for this build, and doing it here would distract from the change data capture correctness that is the point.

The natural next step, once the change data capture pipeline is trusted in production, is to add a short-TTL cache for idempotent reads and a richer sink writer that exports histograms rather than counters. Both are additive and neither changes the contracts this build establishes.