select * from pg_stat_subscription
  pg_stat_get_subscription()
    memcpy(&worker, &LogicalRepCtx->workers[i],sizeof(LogicalRepWorker));
    values[6] = LSNGetDatum(worker.reply_lsn);

ApplyLauncherMain(Datum main_arg)
  logicalrep_worker_launch(sub->dbid, sub->oid, sub->name, sub->owner, InvalidOid);
	/* Find unused worker slot. */
	for (i = 0; i < max_logical_replication_workers; i++)
	{
		LogicalRepWorker *w = &LogicalRepCtx->workers[i];

		if (!w->in_use)
		{
			worker = w;
			slot = i;
			break;
		}
	}
	bgw.bgw_main_arg = Int32GetDatum(slot);
	RegisterDynamicBackgroundWorker(&bgw, &bgw_handle)

LogicalRepApplyLoop(XLogRecPtr last_received)
  
	for (;;)
	{
	...
		len = walrcv_receive(wrconn, &buf, &fd);
		if (len != 0)
		{
		
			if (c == 'w')
			{
				XLogRecPtr	start_lsn;
				XLogRecPtr	end_lsn;
				TimestampTz send_time;

				start_lsn = pq_getmsgint64(&s);
				end_lsn = pq_getmsgint64(&s);
				send_time = pq_getmsgint64(&s);

				if (last_received < start_lsn)
					last_received = start_lsn;

				if (last_received < end_lsn)
					last_received = end_lsn;

				UpdateWorkerStats(last_received, send_time, false);//更新pg_stat_subscription.received_lsn

				apply_dispatch(&s);
			}
			else if (c == 'k')
			{
				XLogRecPtr	end_lsn;
				TimestampTz timestamp;
				bool		reply_requested;

				end_lsn = pq_getmsgint64(&s);
				timestamp = pq_getmsgint64(&s);
				reply_requested = pq_getmsgbyte(&s);

				if (last_received < end_lsn)
					last_received = end_lsn;

				send_feedback(last_received, reply_requested, false);//反馈订阅端的write/flush/reply lsn
				UpdateWorkerStats(last_received, timestamp, true);//更新pg_stat_subscription.received_lsn和pg_stat_subscription.latest_end_lsn
			}
		}
		send_feedback(last_received, false, false);//反馈订阅端的write/flush/reply lsn

send_feedback(XLogRecPtr recvpos, bool force, bool requestReply)
  	get_flush_position(&writepos, &flushpos, &have_pending_txes);
	/*
	 * No outstanding transactions to flush, we can report the latest received
	 * position. This is important for synchronous replication.
	 */
	if (!have_pending_txes)
		flushpos = writepos = recvpos;
	...
	pq_sendbyte(reply_message, 'r');
	pq_sendint64(reply_message, recvpos);	/* write */
	pq_sendint64(reply_message, flushpos);	/* flush */
	pq_sendint64(reply_message, writepos);	/* apply */
	pq_sendint64(reply_message, now);	/* sendTime */
	pq_sendbyte(reply_message, requestReply);	/* replyRequested */


static void
get_flush_position(XLogRecPtr *write, XLogRecPtr *flush,
				   bool *have_pending_txes)
{
	dlist_mutable_iter iter;
	XLogRecPtr	local_flush = GetFlushRecPtr();

	*write = InvalidXLogRecPtr;
	*flush = InvalidXLogRecPtr;

	dlist_foreach_modify(iter, &lsn_mapping)//lsn_mapping 在应用commit日志时更新
	{
		FlushPosition *pos =
		dlist_container(FlushPosition, node, iter.cur);

		*write = pos->remote_end;

		if (pos->local_end <= local_flush)
		{
			*flush = pos->remote_end;
			dlist_delete(iter.cur);//从lsn_mapping中移除已经本地刷盘的记录
			pfree(pos);
		}
		else
		{
			/*
			 * Don't want to uselessly iterate over the rest of the list which
			 * could potentially be long. Instead get the last element and
			 * grab the write position from there.
			 */
			pos = dlist_tail_element(FlushPosition, node,
									 &lsn_mapping);
			*write = pos->remote_end;
			*have_pending_txes = true;
			return;
		}
	}

	*have_pending_txes = !dlist_is_empty(&lsn_mapping);
}

ApplyWorkerMain
  LogicalRepApplyLoop(origin_startpos);
    apply_dispatch(&s);
      apply_handle_commit(StringInfo s)
        replorigin_session_origin_lsn = commit_data.end_lsn; //更新pg_replication_origin_status
        replorigin_session_origin_timestamp = commit_data.committime;
        CommitTransactionCommand();
        store_flush_position(commit_data.end_lsn);
            /* Track commit lsn  */
            flushpos = (FlushPosition *) palloc(sizeof(FlushPosition));
            flushpos->local_end = XactLastCommitEnd;
            flushpos->remote_end = remote_lsn;
            dlist_push_tail(&lsn_mapping, &flushpos->node);