Function lock_and_prepare_snapshot 

async fn lock_and_prepare_snapshot(
    config: &RawSourceCreationConfig,
    connection_config: &Config,
    task_name: &str,
    tables: &BTreeMap<MySqlTableName, Vec<SourceOutputInfo>>,
    metrics: &MySqlSnapshotMetrics,
) -> Result<(SnapshotInfo, BTreeMap<MySqlTableName, u64>, MySqlConn), TransientError>

Leader-only snapshot setup: sample PK bounds, lock the tables READ, and read the snapshot GTID frontier. All fallible work happens here so the caller can always broadcast a result. A dropped snapshot_cap_set with no broadcast deadlocks the other workers waiting on the feedback loop.