Struct SharedLsnFeedback

Source

pub struct SharedLsnFeedback { /* private fields */ }

Expand description

Thread-safe tracker for LSN positions used in replication feedback

This tracker is designed to be shared between the producer (which sends feedback to PostgreSQL) and the consumer (which commits transactions to the destination).

§Example

use pg_walstream::lsn::SharedLsnFeedback;
use std::sync::Arc;

let feedback = SharedLsnFeedback::new_shared();

// Consumer updates LSN after flushing data
feedback.update_flushed_lsn(1000);

// Consumer updates LSN after committing transaction  
feedback.update_applied_lsn(1000);

// Producer reads LSN values for feedback to PostgreSQL
let (flushed, applied) = feedback.get_feedback_lsn();

Implementations§

Source §

impl SharedLsnFeedback

Source

pub fn new() -> Self

Create a new shared LSN feedback tracker

Initializes both flushed_lsn and applied_lsn to 0. Use new_shared() if you need an Arc-wrapped instance for sharing between threads.

§Example

use pg_walstream::lsn::SharedLsnFeedback;

let feedback = SharedLsnFeedback::new();
assert_eq!(feedback.get_flushed_lsn(), 0);
assert_eq!(feedback.get_applied_lsn(), 0);

Source

pub fn new_shared() -> Arc<Self> ⓘ

Create a new shared LSN feedback tracker wrapped in Arc for sharing

This is the preferred way to create a feedback tracker that will be shared between the producer (reading from PostgreSQL) and consumer (writing to destination).

§Returns

An Arc<SharedLsnFeedback> that can be cloned and shared across threads safely.

§Example

use pg_walstream::lsn::SharedLsnFeedback;
use std::sync::Arc;

let feedback = SharedLsnFeedback::new_shared();
let consumer_feedback = Arc::clone(&feedback);

// Consumer thread updates LSN after commit
consumer_feedback.update_applied_lsn(1000);

// Producer thread reads LSN for PostgreSQL feedback
let (flushed, applied) = feedback.get_feedback_lsn();

Source

pub fn update_flushed_lsn(&self, lsn: impl Into<Lsn>)

Update the flushed LSN if the new value is greater

This should be called when data has been written/flushed to the destination database, but not yet committed (e.g., during batch writes).

Accepts anything convertible to Lsn, so event.lsn and a raw u64 both work without an explicit .value().

Source

pub fn update_applied_lsn(&self, lsn: impl Into<Lsn>)

Update the applied/replayed LSN if the new value is greater

This should be called when a transaction has been successfully committed to the destination database. This is the most important LSN as PostgreSQL uses it to determine which WAL can be recycled.

Accepts anything convertible to Lsn, so event.lsn and a raw u64 both work without an explicit .value().

Source

pub fn get_flushed_lsn(&self) -> XLogRecPtr

Get the current flushed LSN

Returns the last LSN value that was flushed to the destination database. This represents data that has been written but may not yet be committed.

§Returns

The current flushed LSN as a u64 value (XLogRecPtr).

Source

pub fn get_applied_lsn(&self) -> XLogRecPtr

Get the current applied LSN

Returns the last LSN value that was successfully committed to the destination database. This is the most important LSN as PostgreSQL uses it to determine which WAL segments can be safely recycled.

§Returns

The current applied LSN as a u64 value (XLogRecPtr).

Source

pub fn get_feedback_lsn(&self) -> (XLogRecPtr, XLogRecPtr)

Get both LSN values for feedback, preserving applied <= flushed.

The two loads are separate atomics, so they can straddle a concurrent update_applied_lsn, which publishes applied before bumping flushed. A reader landing in that window would otherwise observe flushed < applied and ship a Standby Status Update whose flush_lsn trails its replay_lsn — which PostgreSQL uses to advance confirmed_flush_lsn and to release SYNC_REP_WAIT_FLUSH waiters, so the inversion is not cosmetic.

Reordering the two loads does not fix this (the reader can still land between the applied CAS and the flushed fetch_max); clamping does. applied is the conservative side: it is the position the consumer claims to have durably applied, and update_applied_lsn folds the implicit flush bump in, so applied <= flushed is the steady-state invariant being restored here, not a value being invented.

§Returns

A tuple of (flushed_lsn, applied_lsn) as u64 values.

§Example

use pg_walstream::lsn::SharedLsnFeedback;

let feedback = SharedLsnFeedback::new();
feedback.update_applied_lsn(1000);

let (flushed, applied) = feedback.get_feedback_lsn();
assert_eq!(flushed, 1000);
assert_eq!(applied, 1000);
assert!(applied <= flushed, "feedback must never report replay ahead of flush");