Struct ResilientWatchConfig 

pub struct ResilientWatchConfig {
    pub backoff: Backoff,
    pub cursor_store: Arc<dyn CursorStore>,
    pub from_cursor: Option<Cursor>,
    /* private fields */
}

Knobs for StreamClient::resilient_watch. Reuses the Backoff and CursorStore from the ResilientSubscription resilience layer.

Fields

backoff: Backoff

Reconnect (and re-anchor-retry) backoff schedule.

cursor_store: Arc<dyn CursorStore>

Where the resume cursor is persisted across reconnects and restarts.

from_cursor: Option<Cursor>

Initial resume anchor used on the first connect when the store is empty.

Implementations

impl ResilientWatchConfig

pub fn new() -> Self

Start from the defaults (forever-retry backoff, no persistence, no initial cursor).

pub fn cursor_store(self, store: Arc<dyn CursorStore>) -> Self

Persist the resume cursor through store.

pub fn backoff(self, backoff: Backoff) -> Self

Override the reconnect / re-anchor-retry backoff.

pub fn from_cursor(self, cursor: Cursor) -> Self

Seed the first connect’s resume anchor (used only when the store has no persisted cursor).

pub fn watch_set_loader<F, Fut>(self, loader: F) -> Self

where F: Fn(WatchSetBuilder) -> Fut + Send + Sync + 'static, Fut: Future<Output = Result<(), Box<dyn Error + Send + Sync>>> + Send + 'static,

Install a canonical watch-set loader for integrators whose watch-set has a durable source-of-truth (a DB, a config file, an upstream service) outside the wrapper.

Without a loader, ResilientWatch re-registers its in-memory mirror of the add_* / remove_* calls made through it — correct when the watch-set is built once at startup and never drifts, but the mirror is empty after a process restart and goes stale if the truth changes while the stream is down. With a loader, the mirror is treated as a cache of the external truth:

• The loader runs once after every successful (re)connect, before the consumer’s event stream resumes — so the first events pumped after a reconnect already land on a fully-populated subscription. It receives a fresh WatchSetBuilder and declares the canonical set into it (typically by querying its truth with its own awaits between calls).
• On return, the builder’s set replaces the mirror: it is canonical. In-process add_* / remove_* calls still mutate the mirror and send live, but the next reconnect re-derives the set from the loader, so the integrator’s truth — not the accumulated in-process edits — is the record across reconnects (this is what closes the restart-loses-mirror and truth-drift-during-disconnect gaps).
• A loader error is transient: it maps to StreamError::WatchSetLoader, which the reconnect loop backs off and retries on the next connect rather than surfacing — a momentary failure of the integrator’s truth must not crash a consumer whose contract is at-least-once. A permanent loader error (a config typo, a closure that always fails) is indistinguishable from a transient one and so is retried indefinitely: with the default backoff (max_retries: None) the stream never resumes and next simply never yields. Set Backoff::max_retries if you need a permanently-failing loader to surface a terminal error instead of retrying forever. The loader should return Err rather than panic — a panic unwinds through the reconnect and aborts the current next(); it is not caught.

The resume cursor is independent of the watch-set: it still comes from the cursor_store / from_cursor and the re-anchor runs after the loaded set is registered, exactly as without a loader.

let config = ResilientWatchConfig::new().watch_set_loader(|builder: WatchSetBuilder| async move {
    // Query the durable source-of-truth and declare the canonical set.
    let scripts = rows().await?;
    builder.add_scripts(scripts);
    Ok(())
});

Trait Implementations

impl Debug for ResilientWatchConfig

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Default for ResilientWatchConfig

fn default() -> Self

Returns the “default value” for a type.