如何获取Npgsql逻辑复制中最后确认的LSN？

有两种靠谱的方式解决你的问题：

1. 直接查询PostgreSQL系统视图 #

PostgreSQL的pg_replication_slots系统视图里，restart_lsn字段就是复制槽最后被确认的LSN值——每次你调用Acknowledge确认LSN后，PostgreSQL会自动更新这个字段。

你可以用普通的Npgsql连接查询这个值，代码示例：

using var conn = new NpgsqlConnection("你的数据库连接字符串");
await conn.OpenAsync(cancellationToken);

var slotName = "你的复制槽名称";
using var cmd = new NpgsqlCommand(@"SELECT restart_lsn FROM pg_replication_slots WHERE slot_name = @slotName", conn);
cmd.Parameters.AddWithValue("slotName", slotName);

var result = await cmd.ExecuteScalarAsync(cancellationToken);
if (result is NpgsqlLogSequenceNumber lastConfirmedLsn)
{
    // 用这个LSN启动复制，就能从最后确认的位置开始
    await foreach (var message in replicationConnection.StartReplication(replicationSlot, replicationPublication, lastConfirmedLsn, cancellationToken))
    {
        // 处理变更消息
        await replicationConnection.Acknowledge(message.WalEnd);
    }
}

注意：执行这个查询的数据库用户需要有查询pg_replication_slots的权限。

2. 消费过程中自行记录并持久化LSN #

既然你是唯一消费者，完全可以自己在处理完消息后，把确认的LSN保存到本地（比如文件、轻量数据库），下次启动时直接读取这个值传入StartReplication。这种方式不需要依赖系统视图的权限，逻辑也更直接。

代码示例：

// 启动时读取本地存储的LSN
NpgsqlLogSequenceNumber? lastSavedLsn = null;
var lsnStoragePath = "last_processed_lsn.txt";
if (File.Exists(lsnStoragePath))
{
    var lsnStr = await File.ReadAllTextAsync(lsnStoragePath, cancellationToken);
    if (NpgsqlLogSequenceNumber.TryParse(lsnStr, out var parsedLsn))
    {
        lastSavedLsn = parsedLsn;
    }
}

// 从保存的LSN位置开始复制
await foreach (var message in replicationConnection.StartReplication(replicationSlot, replicationPublication, lastSavedLsn, cancellationToken))
{
    // 处理你的变更业务逻辑
    HandleChangeMessage(message);

    // 确认LSN并持久化到本地
    await replicationConnection.Acknowledge(message.WalEnd);
    await File.WriteAllTextAsync(lsnStoragePath, message.WalEnd.ToString(), cancellationToken);
}

两种方式各有优劣：第一种适合需要和数据库端状态对齐的场景；第二种更轻量，不需要额外权限，适合单一消费者的独立服务。

内容的提问来源于stack exchange，提问作者daniil_