let producers: Vec<_> = sources
            .into_iter()
            .map(|(name, reader)| {
                let tx = tx.clone();
                s.spawn(move || -> Result<()> {
                    for record in reader.iter()? {
                        let bytes = record?;
                        let fti = FeatureToIndex::decode(bytes.as_slice()).with_context(|| {
                            format!("failed to decode a FeatureToIndex record from {name}")
                        })?;
                        tx.send(fti)?;
                        progress_bar.inc(1);
                    }
                    Ok(())
                })
            })
            .collect();