Resume data replication in Postgres and Node.js

How to resume replication from the point where the Node.js application was stopped

Rectangular blue blocks rest on top of one another and ascend up and to the right.

Manuel Spigolon 19 Jul 2024

This article is a continuation of Real-time data Replication in Postgres and Node.js. Before reading this article, I recommend you read the previous one because it provides essential context to the points I cover.

In our previous article, we discussed how to replicate data from a Postgres database to a Node.js application in real-time using logical replication. However, if the Node.js application crashes or stops for some reason, the replication will cease, and we risk losing the data that our system produces in the meantime via another microservice or application.

In this article, I discuss how to resume replication from the last point where the Node.js application stopped by using a persistent replication slot in the Postgres database. This ensures that our application doesn't lose events produced by other microservices or applications during downtime.

Editor’s note: This is a cross-post written by Senior Software Developer, Manuel Spigolon. Manuel has his own blog at backend.cafe where you can subscribe for updates and find more great posts. Some of the links in the article point to Manuel’s personal GitHub account.

Creating a replication slot

To resume replication, we need to create a replication slot in the Postgres database. A replication slot is a logical entity that keeps track of changes happening in the database and sends them to the subscriber. The postgres package we used in the previous article automatically created a replication slot for us, but it was not persistent, it was a `TEMPORARY` replication slot that was removed when the subscriber disconnected.

Since we want to resume replication from the last point where the Node.js application was stopped, we need to create a persistent replication slot. Let's create one in a new setup-replication.js file:

import pg from 'pg'
const { Client } = pg

const client = new Client({ user: 'postgres', password: 'foopsw' })
await client.connect()

await createReplicationSlotIfNotExists('foo_slot')

await client.end()

async function createReplicationSlotIfNotExists (slotName) {
  const slots = await client.query('SELECT * FROM pg_replication_slots WHERE slot_name = $1', [slotName])

  if (!slots.rows.length) {
    const newSlot = await client.query("SELECT * FROM pg_create_logical_replication_slot($1, 'pgoutput')", [slotName])
    console.log('Created replication slot', newSlot.rows[0])
  } else {
    console.log('Slot already exists', slots.rows[0])
  }
}

We are using the `pgreplicationslots` view to check whether a replication slot with the given name already exists. If it doesn't exist then we create a new replication slot using the `pgcreatelogicalreplicationslot` function.

Note that we specified the `pgoutput` plugin in the function to decode the changes in the replication slot. This is the default plugin for logical replication, and it ships with Postgres. Be aware that there are other plugins, such as:

`test_decoding` plugin is the simplest plugin that ships with Postgres to start building your own custom plugin.
`wal2json`, which must be installed separately in the Postgres database, allowing you to use them in the pgcreatelogicalreplicationslot function.

Note that each plugin has its own advantages and disadvantages, so choose the one that best fits your use case. The biggest difference if you try to use test_decoding versus pgoutput is that the former does not accept a publication name as a parameter while the latter does. This means that you can use pgoutput to filter the changes you want to replicate, while test_decoding will replicate all changes in the database without filtering them!

Now, run the setup-replication.js file to create a replication slot!

Configuring the consumer

In the previous article, we already created the setup-consumer.js that creates the publications that our application is interested in. So, we can reuse the same file and just run it — if we haven't already.

As a reminder: you will first need to start the Postgres server and create the foo database.

Resuming the replication

We are ready to create a new consumer-resume.js file that will resume replication from the last point where the Node.js application was stopped, so let's jump into it:

import { LogicalReplicationService, PgoutputPlugin } from 'pg-logical-replication'

const client = new LogicalReplicationService({
  user: 'postgres',
  password: 'foopsw'
}, { acknowledge: { auto: false } })

client.on('data', async (lsn, log) => {
  if (log.tag === 'insert') {
    console.log(`${lsn}) Received insert: ${log.relation.schema}.${log.relation.name} ${log.new.id}`)
  } else if (log.relation) {
    console.log(`${lsn}) Received log: ${log.relation.schema}.${log.relation.name} ${log.tag}`)
  }

  await client.acknowledge(lsn)
})

const eventDecoder = new PgoutputPlugin({
  // Get a complete list of available options at:
  // https://www.postgresql.org/docs/16/protocol-logical-replication.html
  protoVersion: 4,
  binary: true,
  publicationNames: [
    'foo_odd',
    'foo_update_only'
  ]
})

console.log('Listening for changes...')
process.on('SIGINT', async () => {
  console.log('Stopping client...')
  await client.stop()
})

await client.subscribe(eventDecoder, 'foo_slot')

This time, we are using the `pg-logical-replication package` to demonstrate the resuming of replication. Its low-level API provides us more control over the replication process, otherwise we would not be able to configure the Plugin to receive only the changes we are interested in.

The code can be explained as follows:

1: We are creating a new LogicalReplicationService instance and passing the connection options to it. Note that we are setting the acknowledge.auto option to false to manually acknowledge the changes; otherwise, they would be automatically acknowledged. By setting this option to false, we gain even more control over the process.
2: We are listening to the data event to receive changes from the replication slot.
- At this point, you should process the log and apply your business logic. In this case, we're just logging the changes to the console.
- After processing the changes, you must acknowledge them using the acknowledge method; otherwise, the slot will not advance. The lsn (Log Sequence Number) is the unique identifier for each change in the database and is used to track changes in the replication slot.
3: We are creating a new PgoutputPlugin instance and passing it to the subscribe method to establish a connection with the replication slot.

To start the application, run the node consumer-resume.js file, and it will begin receiving changes from the replication slot. If we did all the steps correctly, you can start the node producer.js file that we wrote in the previous article to produce changes in the database and see the changes in the consumer application.

If you stop the consumer application by pressing Ctrl+C, the replication will stop, and the slot will not move forward. However, if you start the consumer-resume.js application again, it will resume replication from the last point where it was stopped! 🎉

Moreover, we can see that the output shows only the changes from the foo_odd and fooupdateonly publications, which we configured in the PgoutputPlugin instance so we will see updates and inserts with odd id numbers only:

text

0/15648E0) Received insert: public.foo 18
0/15649A0) Received log: public.foo update
0/1564AF0) Received log: public.foo update
0/1564B80) Received insert: public.foo 20
0/1564C40) Received log: public.foo update
0/1564D90) Received log: public.foo update
0/1564E20) Received insert: public.foo 22
0/1564EE0) Received log: public.foo update
0/1565030) Received log: public.foo update
0/15650C0) Received insert: public.foo 24

Conclusion

In this article, we discussed how to resume replication from the last point where the Node.js application was stopped.

We created a persistent replication slot in the Postgres database and used the pg-logical-replication package to demonstrate resuming replication. This ensures that our application doesn't lose data produced by other microservices or applications during downtime.

In doing so, we did not change the producer.js file, which means that the producer can continue to produce changes in the database without any issues and the previous Publications setup is still valid: we just configured manually the replication slot and the new consumer.

Remember, the replication slot retains changes in the database until the slot is dropped or the changes are acknowledged by the subscriber. If not managed properly, this can lead to high disk usage because Postgres will keep the changes in the WAL logs indefinitely instead of removing them.

I hope you enjoyed this article and learned something new! Does it deserve a comment and a share? 🚀

Real-time data replication in Postgres and Node.js

9 mins
Manuel Spigolon
23 May 2024

The complete guide to the Fastify plugin system

7 mins
Manuel Spigolon
23 Nov 2022

Insight, imagination and expertly engineered solutions to accelerate and sustain progress.

Contact