pg-logical-replication

- PostgreSQL Logical Replication client for node.js(
>=16.9.0)
- Supported plugins
- pgoutput (Native to
PostgreSQL, Recommended)
- Use the pgoutput plugin to process huge transactions.
- wal2json
- decoderbufs
- test_decoding (Not recommended)
- Document for old version(1.x)


| PostgreSQL Versions | on Node.js 16, 18, 20, 22, 24 |
|---------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| PostgreSQL 14 |  |
| PostgreSQL 15 |  |
| PostgreSQL 16 |  |
| PostgreSQL 17 |  |

1. Install

- pg-logical-replication depends on pg(node-postgres) >= 6.2.2
and eventemitter2

``sh
$ npm install pg-logical-replication
`

2. Usage

- This is an example using wal2json. A replication slot(test_slot_wal2json) must be created on the PostgreSQL
server.
- SELECT * FROM pg_create_logical_replication_slot('test_slot_wal2json', 'wal2json')

`typescript
const slotName = 'test_slot_wal2json';

const service = new LogicalReplicationService(
/**
* node-postgres Client options for connection
* https://github.com/DefinitelyTyped/DefinitelyTyped/blob/master/types/pg/index.d.ts#L16
*/
{
database: 'playground',
// ...
},
/**
* Logical replication service config
* https://github.com/kibae/pg-logical-replication/blob/main/src/logical-replication-service.ts#L9
*/
{
acknowledge: {
auto: true,
timeoutSeconds: 10
}
}
)

// TestDecodingPlugin for test_decoding and ProtocolBuffersPlugin for decoderbufs are also available.
const plugin = new Wal2JsonPlugin({
/**
* Plugin options for wal2json
* https://github.com/kibae/pg-logical-replication/blob/main/src/output-plugins/wal2json/wal2json-plugin-options.type.ts
*/
//...
});

/**
* Wal2Json.Output
* https://github.com/kibae/pg-logical-replication/blob/main/src/output-plugins/wal2json/wal2json-plugin-output.type.ts
*/
service.on('data', (lsn: string, log: Wal2Json.Output) => {
// Do something what you want.
// log.change.filter((change) => change.kind === 'insert').length;
});

// Start subscribing to data change events.
(function proc() {
service.subscribe(plugin, slotName)
.catch((e) => {
console.error(e);
})
.then(() => {
setTimeout(proc, 100);
});
})();
`

----

3. LogicalReplicationService

$3

`typescript
const service = new LogicalReplicationService(
/**
* node-postgres Client options for connection
* https://github.com/DefinitelyTyped/DefinitelyTyped/blob/master/types/pg/index.d.ts#L16
*/
clientConfig
:
{
user ? : string | undefined;
database ? : string | undefined;
password ? : string | (() => string | Promise) | undefined;
port ? : number | undefined;
host ? : string | undefined;
connectionString ? : string | undefined;
keepAlive ? : boolean | undefined;
stream ? : stream.Duplex | undefined;
statement_timeout ? : false | number | undefined;
parseInputDatesAsUTC ? : boolean | undefined;
ssl ? : boolean | ConnectionOptions | undefined;
query_timeout ? : number | undefined;
keepAliveInitialDelayMillis ? : number | undefined;
idle_in_transaction_session_timeout ? : number | undefined;
application_name ? : string | undefined;
connectionTimeoutMillis ? : number | undefined;
types ? : CustomTypesConfig | undefined;
options ? : string | undefined;
}
,
/**
* Logical replication service config
* https://github.com/kibae/pg-logical-replication/blob/main/src/logical-replication-service.ts#L9
*/
config ? : Partial<{
acknowledge?: {
/**
* If the value is false, acknowledge must be done manually.
* Default: true
*/
auto: boolean;
/**
* Acknowledge is performed every set time (sec). If 0, do not do it.
* Default: 10
*/
timeoutSeconds: 0 | 10 | number;
};
flowControl?: {
/**
* If true, pause the stream until the data handler completes.
* This enables backpressure support for async handlers.
* Default: false
*/
enabled: boolean;
};
}>
)
`

$3

- Receive changes from the server.
- plugin output plugins.
- slotName Logical replication slot name. You can create slot
via pg_create_logical_replication_slot
function.
- uptoLsn (optional) The starting point of the data to be streamed.

$3

- After processing the data, it signals the PostgreSQL server that it is OK to clear the WAL log.
- Usually this is done automatically.
- Manually use only when new LogicalReplicationService({}, {acknowledge: {auto: false}}).

$3

When processing messages takes longer than the rate at which PostgreSQL sends them, the internal buffer can grow
indefinitely, leading to memory issues (OOM). The flowControl option enables backpressure support to prevent this.

`typescript
const service = new LogicalReplicationService(clientConfig, {
acknowledge: { auto: true, timeoutSeconds: 10 },
flowControl: { enabled: true } // Enable backpressure support
});

// Now async handlers are fully supported - the stream pauses until processing completes
service.on('data', async (lsn: string, log: Pgoutput.Message) => {
await someSlowAsyncOperation(log); // Safe: next message waits for this to complete
});
`

How it works:
- When flowControl.enabled is true, the stream is paused while processing each message
- Messages are queued and processed sequentially
- The stream resumes only after the handler (including async operations) completes
- This prevents memory overflow when handlers are slower than the incoming message rate

Default behavior:
- flowControl.enabled defaults to false for backward compatibility
- When disabled, messages are emitted immediately without waiting for handler completion

$3

- on(event: 'start', listener: () => Promise | void)
- Emitted when replication starts.
- on(event: 'data', listener: (lsn: string, log: any) => Promise | void)
- Emitted when PostgreSQL data changes. The log value type varies depending on the plugin.
- on(event: 'error', listener: (err: Error) => void)
- on(event: 'acknowledge', listener: (lsn: string) => Promise | void)
- Emitted when acknowledging automatically.
- on(event: 'heartbeat', listener: (lsn: string, timestamp: number, shouldRespond: boolean) => Promise | void)
- A heartbeat check signal has been received from the server. You may need to run service.acknowledge().

$3

- stop(): Promise
- Terminate the server's connection and stop replication.
- isStop(): boolean
- Returns false when replication starts from the server.
- lastLsn(): string
- Returns the last LSN(Log Sequence Number) received
from the server.

----

4. Output Plugins

$3


PgoutputPlugin for pgoutput (Native to PostgreSQL)

- Use the pgoutput plugin to process large-scale transactions.

$3

$3

$3

TestDecodingPlugin` for test_decoding (Not recommended)

Contributors