Function ensure_source_offsets_table_on_conn 

pub fn ensure_source_offsets_table_on_conn(
    conn: &Connection,
) -> Result<(), ReplicaError>

source_offsets_table_ddl applied to a caller-owned connection (the initial-snapshot store writes it before any consumer exists).