Extension Points

Implement these traits to control where replicated data goes and how ETL state is stored.

Destination

Receives replicated data. This is the primary extension point for sending data to custom systems. ETL is at least once, so destinations must tolerate duplicate writes and concurrent calls.

pub trait Destination {
    fn name() -> &'static str;
    fn shutdown(&self) -> impl Future<Output = EtlResult<()>> + Send { async { Ok(()) } }
    fn startup(&self) -> impl Future<Output = EtlResult<()>> + Send { async { Ok(()) } }
    fn drop_table_for_copy(&self, replicated_table_schema: &ReplicatedTableSchema, async_result: DropTableForCopyResult<()>) -> impl Future<Output = EtlResult<()>> + Send;
    fn write_table_rows(&self, replicated_table_schema: &ReplicatedTableSchema, table_rows: Vec<TableRow>, async_result: WriteTableRowsResult) -> impl Future<Output = EtlResult<()>> + Send;
    fn write_events(&self, events: Vec<Event>, durability: WriteEventsDurability, async_result: WriteEventsResult) -> impl Future<Output = EtlResult<()>> + Send;
}

Methods

Implementation Notes

• drop_table_for_copy() should be idempotent. ETL calls it before clearing copy-scoped store state, so implementations can still use the supplied schema and existing destination metadata to locate the old object. Before returning success, it must also drain or invalidate writes accepted by an earlier copy attempt so stale work cannot mutate the recreated table.
• write_table_rows() is called even for empty source tables so destinations can prepare initial destination state before ongoing replication begins.
• An immediate write_table_rows() implementation returns DestinationWriteStatus::Durable after the batch is durable. A deferred implementation may return Accepted after taking ownership of a nonempty batch. It must bound its accepted-but-not-durable backlog and delay Accepted when no capacity is available. If any batch returns Accepted, ETL sends an empty batch after all copy workers finish; the destination must return Durable from that call only after all rows accepted during the current copy attempt are durable.
• WriteEventsDurability::MayDefer permits write_events() to return Accepted or Durable. ETL may issue write_events(Vec::new(), WriteEventsDurability::RequireDurable, ...) as a durability-only barrier, but never an empty MayDefer write. The empty vector carries no new replication events, but the call may flush or wait for earlier accepted work and must return Durable only after all writes covered by the destination's ordering state are durable. A destination may use a stronger barrier scope than the originating apply-loop stream.
• write_table_rows() and write_events() must tolerate duplicate delivery because ETL may retry or replay after failure.
• Handle concurrent calls safely, especially from parallel table sync workers.
• Preserve per-table event order. During initial sync and catch-up, transaction markers are not a reliable all-tables transaction boundary.
• Treat Event::Relation as an ordered schema transition, not a write_events() batch boundary. ETL batches ongoing replication events by size and time, so one call can contain multiple schema changes, including multiple relation events for the same table.
• Always complete the supplied async result handle. Dropping it reports a destination error to ETL.
• startup() runs after ETL has loaded destination metadata and table schemas from the store and purged tables removed from the publication, so destinations can compare active persisted ETL state with their physical objects before replication work starts.
• All three write-like methods use async results, but ETL waits differently. drop_table_for_copy() waits immediately before copy-scoped store cleanup. write_table_rows() also waits immediately, requesting the next batch only after the current one reports Accepted or Durable for that copy partition. write_events() is the method where ETL can keep processing other work while the destination finishes the current batch; ETL still waits for that result before handing over the next ongoing-replication batch.

See Events for details on the events received by write_events().

PipelineDestination is a blanket-implemented facade for destinations that also satisfy the pipeline runtime clone and thread-safety bounds. Pipeline runtime code uses this facade when it needs to move destinations across worker tasks, but custom destinations only implement Destination directly.

SchemaStore

Stores versioned table schema information (column names, types, primary keys, and snapshot IDs). A SnapshotId compares its commit LSN first and its message LSN second; store implementations should compare the type directly rather than its variable-width decimal display string.

pub trait SchemaStore {
    fn get_table_schema(&self, table_id: &TableId, snapshot_id: SnapshotId) -> impl Future<Output = EtlResult<Option<Arc<TableSchema>>>> + Send;
    fn get_table_schemas(&self) -> impl Future<Output = EtlResult<Vec<Arc<TableSchema>>>> + Send;
    fn load_table_schemas(&self) -> impl Future<Output = EtlResult<usize>> + Send;
    fn store_table_schema(&self, table_schema: TableSchema) -> impl Future<Output = EtlResult<Arc<TableSchema>>> + Send;
    fn prune_table_schemas(&self, retention_snapshot_ids: BTreeMap<TableId, SnapshotId>) -> impl Future<Output = EtlResult<u64>> + Send;
}

Methods

StateStore

Tracks table states, persisted replication checkpoints, and destination table metadata.

pub trait StateStore {
    // Table state
    fn get_table_state(&self, table_id: TableId) -> impl Future<Output = EtlResult<Option<TableState>>> + Send;
    fn get_table_states(&self) -> impl Future<Output = EtlResult<TableStates>> + Send;
    fn load_table_states(&self) -> impl Future<Output = EtlResult<usize>> + Send;
    fn update_table_states(&self, updates: Vec<(TableId, TableState)>) -> impl Future<Output = EtlResult<()>> + Send;
    fn update_table_state(&self, table_id: TableId, state: TableState) -> impl Future<Output = EtlResult<()>> + Send;
    fn rollback_table_state(&self, table_id: TableId) -> impl Future<Output = EtlResult<TableState>> + Send;

    // Persisted replication checkpoints
    fn get_replication_checkpoint(&self, worker_type: WorkerType) -> impl Future<Output = EtlResult<Option<PgLsn>>> + Send;
    fn upsert_replication_checkpoint(&self, worker_type: WorkerType, checkpoint_lsn: PgLsn) -> impl Future<Output = EtlResult<PgLsn>> + Send;
    fn delete_replication_checkpoint(&self, worker_type: WorkerType) -> impl Future<Output = EtlResult<()>> + Send;

    // Destination table metadata
    fn get_destination_table_metadata(&self, table_id: TableId) -> impl Future<Output = EtlResult<Option<DestinationTableMetadata>>> + Send;
    fn get_applied_destination_table_metadata(&self, table_id: TableId) -> impl Future<Output = EtlResult<Option<AppliedDestinationTableMetadata>>> + Send;
    fn load_destination_tables_metadata(&self) -> impl Future<Output = EtlResult<usize>> + Send;
    fn store_destination_table_metadata(&self, table_id: TableId, metadata: DestinationTableMetadata) -> impl Future<Output = EtlResult<()>> + Send;
}

Table State Methods

Replication Checkpoint Methods

A persisted replication checkpoint records a safe replay frontier for the apply worker or a table-sync worker. ETL can select a durably flushed commit boundary or, when the apply loop is fully idle, its last received LSN. The checkpoint lets the worker resume safely after a restart.

Destination Metadata Methods

Destination table metadata connects source table IDs to destination state, including the destination table identifier, the schema snapshot under management, the schema status (Applying or Applied), and the replication mask. Only AppliedDestinationTableMetadata guarantees the destination schema is ready for normal reads and writes.

Table States

Tables progress through these states:

TableStateLifecycleStore

Coordinates ETL table-state lifecycle operations across state, schema, destination metadata, persisted checkpoints, and any store caches.

pub trait TableStateLifecycleStore {
    fn apply_table_state_operation(
        &self,
        operation: TableStateOperation,
    ) -> impl Future<Output = EtlResult<usize>> + Send;

    fn prepare_table_state_for_copy(
        &self,
        table_id: TableId,
    ) -> impl Future<Output = EtlResult<()>> + Send;

    fn reset_table_states_for_resync(
        &self,
    ) -> impl Future<Output = EtlResult<usize>> + Send;

    fn delete_table_state(
        &self,
        table_id: TableId,
    ) -> impl Future<Output = EtlResult<()>> + Send;
}

Combining Traits

A single type typically implements all store traits:

pub struct MyStore { /* ... */ }

impl SchemaStore for MyStore { /* ... */ }
impl StateStore for MyStore { /* ... */ }
impl TableStateLifecycleStore for MyStore { /* ... */ }

PipelineStore is a blanket-implemented facade for stores that satisfy the full pipeline runtime store bounds. Pipeline runtime code uses this facade, while code that only needs one capability should depend on the narrower trait directly.

DestinationStore is a blanket-implemented facade for stores that satisfy the destination runtime store bounds. Destination implementations use this when they need schema and state metadata but do not need lifecycle reset/removal operations. SharedStateStore covers state-only users with the corresponding worker-safe bounds.

ETL provides two built-in implementations:

• MemoryStore: In-memory storage, not persistent across restarts
• PostgresStore: Persistent storage backed by PostgreSQL

PostgresStore::new() runs only the Postgres-backed state-store migrations. Pipeline::start() runs the source migrations required by ETL itself, including the schema helper functions and DDL event trigger, regardless of which store implementation you use.

Thread Safety

All trait implementations must be thread-safe. ETL calls these methods concurrently from:

• Multiple table sync workers (parallel initial sync)
• Apply worker (ongoing replication)
• Pipeline coordination

Use Arc<Mutex<_>>, RwLock, or similar synchronization primitives for shared state.