mononoke: passing LiveCommitSyncConfig all the way to CommitSyncer

Summary:
CommitSyncer is a struct that we use to remap a commit from one repo to another. It uses commit sync map to figure out which paths need to be changed. Commit sync mapping might change, and each commit sync mapping has a version associated with it.

At the moment CommitSyncer doesn't work correctly if a commit sync mapping is changed. Consider the following DAG

```
large repo

A' <- remapped with mapping V1
|
O  B' <- remapped with mapping V1
|  /
...

small repo

A
|
O  B
|  /
...
```

We have commit A and B from a small repo remapped into a large repo into commits A' and B'. They were remapped with commit sync mapping V1, which for example remaps files in "dir/" into "smallrepo/dir".

Now let's say we start to use a new mapping v2 which remaps "dir/" into "otherdir/". After this point every commit will be created with new mapping. But this is incorrect - if we create a commit on top of B in a small repo that touches "dir/file.txt" then it will be remapped into "otherdir/file.txt" in the large repo, even though every other file is still in "smallrepo/dir"!

The fix for this issue is to always use the same mapping as commit parent was using (there are a few tricky cases with merge commits and commits with no parents, but those will be dealt with separately).

This diff is the first step - it threads through LiveCommitSyncConfig all the way to the CommitSyncer object, so that CommitSyncer can always fetch whatever mapping it needs.

Reviewed By: ikostia

Differential Revision: D23845720

fbshipit-source-id: 555cc31fd4ce09f0a6fa2869bfcee2c7cdfbcc61

eden/mononoke/cmds/admin/crossrepo.rs

@@ -26,6 +26,7 @@ use metaconfig_types::{CommitSyncConfig, RepoConfig};
 use mononoke_types::RepositoryId;
 use movers::{get_large_to_small_mover, get_small_to_large_mover};
 use slog::{info, warn, Logger};
+use std::sync::Arc;
 use synced_commit_mapping::{SqlSyncedCommitMapping, SyncedCommitMapping};
 
 use crate::error::SubcommandError;
@@ -51,6 +52,10 @@ pub async fn subcommand_crossrepo<'a>(
     matches: &'a ArgMatches<'_>,
     sub_m: &'a ArgMatches<'_>,
 ) -> Result<(), SubcommandError> {
+    let config_store = args::maybe_init_config_store(fb, &logger, &matches)
+        .ok_or_else(|| format_err!("Failed initializing ConfigStore"))?;
+    let live_commit_sync_config = CfgrLiveCommitSyncConfig::new(&logger, &config_store)?;
+
     args::init_cachelib(fb, &matches, None);
     let ctx = CoreContext::new_with_logger(fb, logger.clone());
     match sub_m.subcommand() {
@@ -74,7 +79,11 @@ pub async fn subcommand_crossrepo<'a>(
                     target_repo,
                     &source_repo_config,
                 )?;
-                CommitSyncer::new(mapping, commit_sync_repos)
+                CommitSyncer::new(
+                    mapping,
+                    commit_sync_repos,
+                    Arc::new(live_commit_sync_config),
+                )
             };
 
             let large_hash = {
@@ -106,11 +115,12 @@ pub async fn subcommand_crossrepo<'a>(
                 target_repo,
                 mapping,
                 update_large_repo_bookmarks,
+                Arc::new(live_commit_sync_config),
             )
             .await
         }
         (SUBCOMMAND_CONFIG, Some(sub_sub_m)) => {
-            run_config_sub_subcommand(fb, ctx, logger, matches, sub_sub_m).await
+            run_config_sub_subcommand(fb, ctx, matches, sub_sub_m, live_commit_sync_config).await
         }
         _ => Err(SubcommandError::InvalidArgs),
     }
@@ -119,14 +129,11 @@ pub async fn subcommand_crossrepo<'a>(
 async fn run_config_sub_subcommand<'a>(
     fb: FacebookInit,
     ctx: CoreContext,
-    logger: Logger,
     matches: &'a ArgMatches<'_>,
     config_subcommand_matches: &'a ArgMatches<'a>,
+    live_commit_sync_config: CfgrLiveCommitSyncConfig,
 ) -> Result<(), SubcommandError> {
     let repo_id = args::get_repo_id(fb, matches)?;
-    let config_store = args::maybe_init_config_store(fb, &logger, &matches)
-        .expect("failed to instantiate ConfigStore");
-    let live_commit_sync_config = CfgrLiveCommitSyncConfig::new(ctx.logger(), &config_store)?;
 
     match config_subcommand_matches.subcommand() {
         (SUBCOMMAND_BY_VERSION, Some(sub_m)) => {
@@ -312,13 +319,15 @@ async fn subcommand_verify_bookmarks(
     target_repo: BlobRepo,
     mapping: SqlSyncedCommitMapping,
     should_update_large_repo_bookmarks: bool,
+    live_commit_sync_config: Arc<dyn LiveCommitSyncConfig>,
 ) -> Result<(), SubcommandError> {
     let commit_sync_repos = get_large_to_small_commit_sync_repos(
         source_repo.clone(),
         target_repo.clone(),
         &source_repo_config,
     )?;
-    let commit_syncer = CommitSyncer::new(mapping.clone(), commit_sync_repos);
+    let commit_syncer =
+        CommitSyncer::new(mapping.clone(), commit_sync_repos, live_commit_sync_config);
 
     let large_repo = commit_syncer.get_large_repo();
     let small_repo = commit_syncer.get_small_repo();
@@ -609,6 +618,7 @@ mod test {
     use cross_repo_sync::validation::find_bookmark_diff;
     use fixtures::{linear, set_bookmark};
     use futures_old::stream::Stream;
+    use live_commit_sync_config::TestLiveCommitSyncConfig;
     use maplit::{hashmap, hashset};
     use metaconfig_types::{
         CommitSyncConfig, CommitSyncConfigVersion, CommitSyncDirection,
@@ -814,6 +824,7 @@ mod test {
             },
         };
 
-        Ok(CommitSyncer::new(mapping, repos))
+        let live_commit_sync_config = Arc::new(TestLiveCommitSyncConfig::new_empty());
+        Ok(CommitSyncer::new(mapping, repos, live_commit_sync_config))
     }
 }

eden/mononoke/commit_rewriting/backsyncer/src/main.rs

@@ -37,6 +37,7 @@ use stats::prelude::*;
 use std::fs::File;
 use std::io::{BufRead, BufReader};
 use std::str::FromStr;
+use std::sync::Arc;
 use std::time::Duration;
 use synced_commit_mapping::{SqlSyncedCommitMapping, SyncedCommitMapping};
 
@@ -113,6 +114,7 @@ where
 {
     let target_repo_id = commit_syncer_args.get_target_repo_id();
 
+    let live_commit_sync_config = Arc::new(live_commit_sync_config);
     loop {
         // We only care about public pushes because draft pushes are not in the bookmark
         // update log at all.
@@ -131,7 +133,7 @@ where
 
                 let commit_syncer = commit_syncer_args
                     .clone()
-                    .try_into_commit_syncer(&commit_sync_config)?;
+                    .try_into_commit_syncer(&commit_sync_config, live_commit_sync_config.clone())?;
 
                 backsync_latest(
                     ctx.clone(),
@@ -289,7 +291,8 @@ fn main(fb: FacebookInit) -> Result<(), Error> {
             let ctx = session_container.new_context(logger.clone(), scuba_sample);
             let commit_sync_config =
                 live_commit_sync_config.get_current_commit_sync_config(&ctx, target_repo_id)?;
-            let commit_syncer = commit_syncer_args.try_into_commit_syncer(&commit_sync_config)?;
+            let commit_syncer = commit_syncer_args
+                .try_into_commit_syncer(&commit_sync_config, Arc::new(live_commit_sync_config))?;
 
             let db_config = target_repo_config.storage_config.metadata;
             let target_repo_dbs = runtime.block_on_std(
@@ -355,7 +358,8 @@ fn main(fb: FacebookInit) -> Result<(), Error> {
             let ctx = session_container.new_context(logger, ScubaSampleBuilder::with_discard());
             let commit_sync_config =
                 live_commit_sync_config.get_current_commit_sync_config(&ctx, target_repo_id)?;
-            let commit_syncer = commit_syncer_args.try_into_commit_syncer(&commit_sync_config)?;
+            let commit_syncer = commit_syncer_args
+                .try_into_commit_syncer(&commit_sync_config, Arc::new(live_commit_sync_config))?;
 
             let inputfile = sub_m
                 .value_of(ARG_INPUT_FILE)

eden/mononoke/commit_rewriting/backsyncer/src/tests.rs

@@ -26,6 +26,7 @@ use futures::{
 };
 use futures_ext::spawn_future;
 use futures_old::{future, stream::Stream as OldStream};
+use live_commit_sync_config::TestLiveCommitSyncConfig;
 use manifest::{Entry, ManifestOps};
 use maplit::btreemap;
 use mercurial_types::HgChangesetId;
@@ -904,8 +905,9 @@ async fn init_repos(
     )
     .await;
 
+    let live_commit_sync_config = Arc::new(TestLiveCommitSyncConfig::new_empty());
     // Sync first commit manually
-    let commit_syncer = CommitSyncer::new(mapping.clone(), repos);
+    let commit_syncer = CommitSyncer::new(mapping.clone(), repos, live_commit_sync_config);
     let initial_bcs_id = source_repo
         .get_bonsai_from_hg(
             ctx.clone(),
@@ -1197,7 +1199,8 @@ async fn init_merged_repos(
             version_name: CommitSyncConfigVersion("TEST_VERSION_NAME".to_string()),
         };
 
-        let commit_syncer = CommitSyncer::new(mapping.clone(), repos);
+        let live_commit_sync_config = Arc::new(TestLiveCommitSyncConfig::new_empty());
+        let commit_syncer = CommitSyncer::new(mapping.clone(), repos, live_commit_sync_config);
         output.push((commit_syncer, small_repo_dbs));
 
         let filename = format!("file_in_smallrepo{}", small_repo.get_repoid().id());
@@ -1433,7 +1436,8 @@ async fn preserve_premerge_commit(
             version_name: CommitSyncConfigVersion("TEST_VERSION_NAME".to_string()),
         };
 
-        CommitSyncer::new(mapping.clone(), repos)
+        let live_commit_sync_config = Arc::new(TestLiveCommitSyncConfig::new_empty());
+        CommitSyncer::new(mapping.clone(), repos, live_commit_sync_config)
     };
 
     small_to_large_sync_config

eden/mononoke/commit_rewriting/bookmarks_validator/src/main.rs

@@ -106,10 +106,10 @@ fn get_commit_syncer<'a>(
     let target_repo_id = args::get_target_repo_id(ctx.fb, &matches)?;
     let config_store = args::maybe_init_config_store(ctx.fb, &logger, &matches)
         .ok_or_else(|| format_err!("Failed initializing ConfigStore"))?;
-    let live_commit_sync_config = CfgrLiveCommitSyncConfig::new(&logger, &config_store)?;
+    let live_commit_sync_config = Arc::new(CfgrLiveCommitSyncConfig::new(&logger, &config_store)?);
     let commit_sync_config =
         live_commit_sync_config.get_current_commit_sync_config(&ctx, target_repo_id)?;
-    commit_syncer_args.try_into_commit_syncer(&commit_sync_config)
+    commit_syncer_args.try_into_commit_syncer(&commit_sync_config, live_commit_sync_config)
 }
 
 async fn loop_forever<M: SyncedCommitMapping + Clone + 'static>(

eden/mononoke/commit_rewriting/commit_validator/src/validation.rs

@@ -662,6 +662,7 @@ impl ValidationHelpers {
             self.large_repo.0.clone(),
             &commit_sync_config,
             self.mapping.clone(),
+            Arc::new(self.live_commit_sync_config.clone()),
         )
     }
 }

eden/mononoke/commit_rewriting/cross_repo_sync/Cargo.toml

@@ -21,6 +21,7 @@ blobsync = { path = "../../blobrepo/blobsync" }
 bookmark_renaming = { path = "../bookmark_renaming" }
 bookmarks = { path = "../../bookmarks" }
 context = { path = "../../server/context" }
+live_commit_sync_config = { path = "../live_commit_sync_config" }
 manifest = { path = "../../manifest" }
 mercurial_types = { path = "../../mercurial/types" }
 metaconfig_types = { path = "../../metaconfig/types" }

eden/mononoke/commit_rewriting/cross_repo_sync/src/commit_syncer_args.rs

@@ -7,9 +7,11 @@
 
 use anyhow::Error;
 use blobrepo::BlobRepo;
+use live_commit_sync_config::LiveCommitSyncConfig;
 use metaconfig_types::CommitSyncConfig;
 use mononoke_types::RepositoryId;
 use std::fmt;
+use std::sync::Arc;
 use synced_commit_mapping::SyncedCommitMapping;
 
 use crate::{CommitSyncRepos, CommitSyncer};
@@ -50,6 +52,8 @@ impl<T: SyncedCommitMapping + Clone + 'static> CommitSyncerArgs<T> {
     pub fn try_into_commit_syncer(
         self,
         commit_sync_config: &CommitSyncConfig,
+        // TODO(stash): remove commit_sync_config and use just live_commit_sync_config
+        live_commit_sync_config: Arc<dyn LiveCommitSyncConfig>,
     ) -> Result<CommitSyncer<T>, Error> {
         let Self {
             source_repo,
@@ -58,7 +62,11 @@ impl<T: SyncedCommitMapping + Clone + 'static> CommitSyncerArgs<T> {
         } = self;
 
         let commit_sync_repos = CommitSyncRepos::new(source_repo, target_repo, commit_sync_config)?;
-        Ok(CommitSyncer::new(mapping, commit_sync_repos))
+        Ok(CommitSyncer::new(
+            mapping,
+            commit_sync_repos,
+            live_commit_sync_config,
+        ))
     }
 }
 

eden/mononoke/commit_rewriting/cross_repo_sync/src/lib.rs

@@ -28,6 +28,7 @@ use futures::{
 };
 use futures_old::Future;
 use futures_old::Stream as StreamOld;
+use live_commit_sync_config::LiveCommitSyncConfig;
 use manifest::get_implicit_deletes;
 use maplit::{hashmap, hashset};
 use mercurial_types::HgManifestId;
@@ -42,6 +43,7 @@ use movers::{get_movers, Movers};
 use pushrebase::{do_pushrebase_bonsai, PushrebaseError};
 use slog::info;
 use std::collections::{BTreeMap, HashMap};
+use std::sync::Arc;
 use std::{collections::VecDeque, fmt};
 use synced_commit_mapping::{
     EquivalentWorkingCopyEntry, SyncedCommitMapping, SyncedCommitMappingEntry,
@@ -485,6 +487,7 @@ pub struct CommitSyncer<M> {
     // TODO: Finish refactor and remove pub
     pub mapping: M,
     pub repos: CommitSyncRepos,
+    live_commit_sync_config: Arc<dyn LiveCommitSyncConfig>,
 }
 
 impl<M> fmt::Debug for CommitSyncer<M>
@@ -502,8 +505,16 @@ impl<M> CommitSyncer<M>
 where
     M: SyncedCommitMapping + Clone + 'static,
 {
-    pub fn new(mapping: M, repos: CommitSyncRepos) -> Self {
-        Self { mapping, repos }
+    pub fn new(
+        mapping: M,
+        repos: CommitSyncRepos,
+        live_commit_sync_config: Arc<dyn LiveCommitSyncConfig>,
+    ) -> Self {
+        Self {
+            mapping,
+            repos,
+            live_commit_sync_config,
+        }
     }
 
     pub fn get_source_repo(&self) -> &BlobRepo {
@@ -1253,7 +1264,7 @@ where
         source_bcs_id: ChangesetId,
         maybe_target_bcs_id: Option<ChangesetId>,
     ) -> Result<(), Error> {
-        let CommitSyncer { repos, mapping } = self.clone();
+        let CommitSyncer { repos, mapping, .. } = self.clone();
         let (source_repo, target_repo, source_is_large) = match repos {
             CommitSyncRepos::LargeToSmall {
                 large_repo,
@@ -1488,6 +1499,7 @@ pub fn create_commit_syncers<M>(
     large_repo: BlobRepo,
     commit_sync_config: &CommitSyncConfig,
     mapping: M,
+    live_commit_sync_config: Arc<dyn LiveCommitSyncConfig>,
 ) -> Result<Syncers<M>, Error>
 where
     M: SyncedCommitMapping + Clone + 'static,
@@ -1523,10 +1535,12 @@ where
     let large_to_small_commit_syncer = CommitSyncer {
         mapping: mapping.clone(),
         repos: large_to_small_commit_sync_repos,
+        live_commit_sync_config: live_commit_sync_config.clone(),
     };
     let small_to_large_commit_syncer = CommitSyncer {
         mapping,
         repos: small_to_large_commit_sync_repos,
+        live_commit_sync_config: live_commit_sync_config.clone(),
     };
 
     Ok(Syncers {

eden/mononoke/commit_rewriting/cross_repo_sync/src/validation.rs

@@ -654,6 +654,7 @@ mod test {
     use fbinit::FacebookInit;
     use fixtures::linear;
     use futures_old::stream::Stream;
+    use live_commit_sync_config::TestLiveCommitSyncConfig;
     use metaconfig_types::{CommitSyncConfigVersion, CommitSyncDirection};
     use mononoke_types::{MPath, RepositoryId};
     use revset::AncestorsNodeStream;
@@ -863,6 +864,7 @@ mod test {
             },
         };
 
-        Ok(CommitSyncer::new(mapping, repos))
+        let live_commit_sync_config = Arc::new(TestLiveCommitSyncConfig::new_empty());
+        Ok(CommitSyncer::new(mapping, repos, live_commit_sync_config))
     }
 }

eden/mononoke/commit_rewriting/cross_repo_sync/test/main.rs

@@ -11,7 +11,6 @@
 
 use bytes::Bytes;
 use fbinit::FacebookInit;
-use maplit::{btreemap, hashmap};
 use std::collections::BTreeMap;
 use std::str::FromStr;
 use std::sync::Arc;
@@ -28,6 +27,8 @@ use cross_repo_sync_test_utils::rebase_root_on_master;
 
 use fixtures::{linear, many_files_dirs};
 use futures::compat::Future01CompatExt;
+use live_commit_sync_config::TestLiveCommitSyncConfig;
+use maplit::{btreemap, hashmap};
 use mercurial_types::HgChangesetId;
 use metaconfig_types::{
     CommitSyncConfig, CommitSyncConfigVersion, CommitSyncDirection,
@@ -239,7 +240,8 @@ fn create_small_to_large_commit_syncer(
 
     let commit_sync_config = create_commit_sync_config(small_repo_id, large_repo_id, prefix)?;
     let repos = CommitSyncRepos::new(small_repo, large_repo, &commit_sync_config)?;
-    Ok(CommitSyncer::new(mapping, repos))
+    let live_commit_sync_config = Arc::new(TestLiveCommitSyncConfig::new_empty());
+    Ok(CommitSyncer::new(mapping, repos, live_commit_sync_config))
 }
 
 fn create_large_to_small_commit_syncer(
@@ -253,7 +255,8 @@ fn create_large_to_small_commit_syncer(
 
     let commit_sync_config = create_commit_sync_config(small_repo_id, large_repo_id, prefix)?;
     let repos = CommitSyncRepos::new(large_repo, small_repo, &commit_sync_config)?;
-    Ok(CommitSyncer::new(mapping, repos))
+    let live_commit_sync_config = Arc::new(TestLiveCommitSyncConfig::new_empty());
+    Ok(CommitSyncer::new(mapping, repos, live_commit_sync_config))
 }
 
 #[fbinit::compat_test]

eden/mononoke/commit_rewriting/cross_repo_sync/test_utils/Cargo.toml

@@ -17,6 +17,7 @@ blobstore = { path = "../../../blobstore" }
 bookmarks = { path = "../../../bookmarks" }
 context = { path = "../../../server/context" }
 cross_repo_sync = { path = ".." }
+live_commit_sync_config = { path = "../../live_commit_sync_config" }
 megarepolib = { path = "../../megarepo" }
 metaconfig_types = { path = "../../../metaconfig/types" }
 mononoke_types = { path = "../../../mononoke_types" }