incorrect lastLsn

[exedealer]

.endLsn of PrimaryKeepaliveMessage is greater than XLogData messages lsn. This causes replication slot moved to position before XLogData messages actually consumed

https://github.com/kagis/pgwire/blob/12826adc830fbb2a501d8b605d5d249441aab90d/mod.js#L1611

[exedealer]

postgres does not send empty transactions anymore, so we need to use endLsn of PrimaryKeepaliveMessage to move slot position

[exedealer]

endLsn of PrimaryKeepaliveMessage goes ahead of XLogData lsn, but not further than commitLsn. So acking endLsn does not cause message loss, because replication restarts from transaction start. My problem is that I mistakenly thought that I can use confirmed_flush_lsn of slot to filter already handled messages in case of big transactions. But message lsn is not monotonic and not unique.

[rkistner]

I'm also running in the issue where endLsn from PrimaryKeepaliveMessage is not available anymore.

On an idle instance on AWS RDS, the WAL grows by 64MB every 5 minutes. When it's idle, there are no messages received on the WAL other than the PrimaryKeepaliveMessage. This means I can't ack anything, and the WAL just keeps on growing until the instance runs out of space.

I've worked around this by periodically using pg_logical_emit_message, but this only works on Postgres 14+.

[rkistner]

Thanks, the change in the commit above solves this issue for me.

FWIW, I only use this LSN when not inside a transaction, just to be safe. Something like the following:

      let inTx = false;
      let ackedLsn = lastPersistedLsn;

      for await (const chunk of replicationStream.pgoutputDecode()) {
        await touch();

        const { messages, lastLsn } = chunk;

        for (const msg of messages) {
          if (msg.tag == 'begin') {
            inTx = true;
          } else if (msg.tag == 'commit') {
            inTx = false;
            // (flush data here)
            ackedLsn = msg.lsn!;
            await replicationStream.ack(msg.lsn!);
          } else {
            // (write data here)
          }
        }

        if (!inTx && lastLsn > ackedLsn) {
          ackedLsn = lastLsn;
          await replicationStream.ack(lastLsn);
        }
      }

[jmealo]

Is this issue addressed by #30? 9532a39

[rkistner]

Yes, the issue is fixed for me.