Struct ReplayState 

pub(crate) struct ReplayState {
    pub(crate) guard: ReplayGuard,
    pub(crate) path: Option<PathBuf>,
    pub(crate) stream: String,
    pub(crate) stream_created_unix_ns: i64,
    pub(crate) subject: String,
    pub(crate) saved: u64,
    pub(crate) last_refresh: Instant,
}

Replay protection for one channel writer.

JetStream can lose a durable consumer’s progress (an empty or missing consumer state file after a power cut) and then deliver the whole stream again. This keeps the archiver’s own checkpoint of the consumer’s ack floor per stream and subject (see replay_guard) and acks messages at or below it without writing them.

Fields

guard: ReplayGuard

Skips messages at or below the trusted checkpoint.

path: Option<PathBuf>

Checkpoint file; None when the guard is off.

stream: String

Stream name, stored in the checkpoint.

stream_created_unix_ns: i64

Stream creation time in Unix nanoseconds, stored in the checkpoint.

subject: String

Channel subject, stored in the checkpoint.

saved: u64

Floor last written to the checkpoint file.

last_refresh: Instant

When the checkpoint was last refreshed.

Implementations

impl ReplayState

pub(crate) fn attach( state_dir: Option<&Path>, stream: &Info, subject: &str, consumer_name: &str, consumer_ack_floor: u64, ) -> Self

Loads the checkpoint for a channel and compares it with the consumer.

A checkpoint that does not match the stream as it is now is ignored with a log line (see replay_guard::trusted_floor). If the consumer’s ack floor is below a trusted checkpoint, JetStream has lost the consumer’s progress; that is logged and the messages up to the checkpoint will be skipped.

Arguments

• state_dir - Checkpoint directory, or None when the guard is off.
• stream - Stream info as fetched when the channel started.
• subject - Channel subject.
• consumer_name - Durable consumer name, for log lines.
• consumer_ack_floor - The consumer’s ack floor stream sequence.

pub(crate) fn note_skip(&self, channel: u8, stream_seq: u64)

Logs the first skipped message of a run and then every SKIP_LOG_EVERY.

pub(crate) fn end_skip_run(&mut self, channel: u8)

Logs the end of a run of skipped messages, if one was in progress.

pub(crate) async fn checkpoint( &mut self, consumer: &mut PullConsumer, channel: u8, force: bool, )

Saves the consumer’s ack floor as the new checkpoint if it has moved up.

Without force this runs at most once per CHECKPOINT_INTERVAL. Failures are logged; the old checkpoint stays valid, only less recent.

Arguments

• consumer - The channel’s durable consumer.
• channel - Channel number, for log lines.
• force - Refresh now, regardless of the interval.