all: Make number of copy workers configurable

Author: lutter

docs/environment-variables.md

@@ -231,6 +231,10 @@ those.
   timeout is hit, the batch size is reset to 1 so we can be sure that
   batches stay below `GRAPH_STORE_BATCH_TARGET_DURATION` and the smaller
   batch is retried. Value is in seconds and defaults to unlimited.
+- `GRAPH_STORE_BATCH_WORKERS`: The number of workers to use for batch
+    operations. If there are idle connectiosn, each subgraph copy operation
+    will use up to this many workers to copy tables in parallel. Defaults
+    to 1 and must be at least 1
 - `GRAPH_START_BLOCK`: block hash:block number where the forked subgraph will start indexing at.
 - `GRAPH_FORK_BASE`: api url for where the graph node will fork from, use `https://api.thegraph.com/subgraphs/id/`
   for the hosted service.

graph/src/env/store.rs

@@ -85,6 +85,12 @@ pub struct EnvVarsStore {
     /// this. Set by `GRAPH_STORE_BATCH_TIMEOUT`. Unlimited by default
     pub batch_timeout: Option<Duration>,
 
+    /// The number of workers to use for batch operations. If there are idle
+    /// connectiosn, each subgraph copy operation will use up to this many
+    /// workers to copy tables in parallel. Defaults to 1 and must be at
+    /// least 1
+    pub batch_workers: usize,
+
     /// Prune tables where we will remove at least this fraction of entity
     /// versions by rebuilding the table. Set by
     /// `GRAPH_STORE_HISTORY_REBUILD_THRESHOLD`. The default is 0.5
@@ -175,6 +181,7 @@ impl TryFrom<InnerStore> for EnvVarsStore {
             write_queue_size: x.write_queue_size,
             batch_target_duration: Duration::from_secs(x.batch_target_duration_in_secs),
             batch_timeout: x.batch_timeout_in_secs.map(Duration::from_secs),
+            batch_workers: x.batch_workers,
             rebuild_threshold: x.rebuild_threshold.0,
             delete_threshold: x.delete_threshold.0,
             history_slack_factor: x.history_slack_factor.0,
@@ -194,6 +201,9 @@ impl TryFrom<InnerStore> for EnvVarsStore {
                 );
             }
         }
+        if vars.batch_workers < 1 {
+            bail!("GRAPH_STORE_BATCH_WORKERS must be at least 1");
+        }
         Ok(vars)
     }
 }
@@ -239,6 +249,8 @@ pub struct InnerStore {
     batch_target_duration_in_secs: u64,
     #[envconfig(from = "GRAPH_STORE_BATCH_TIMEOUT")]
     batch_timeout_in_secs: Option<u64>,
+    #[envconfig(from = "GRAPH_STORE_BATCH_WORKERS", default = "1")]
+    batch_workers: usize,
     #[envconfig(from = "GRAPH_STORE_HISTORY_REBUILD_THRESHOLD", default = "0.5")]
     rebuild_threshold: ZeroToOneF64,
     #[envconfig(from = "GRAPH_STORE_HISTORY_DELETE_THRESHOLD", default = "0.05")]

store/postgres/src/copy.rs

@@ -840,7 +840,7 @@ impl Connection {
             logger,
             conn,
             pool,
-            workers: 5,
+            workers: ENV_VARS.store.batch_workers,
             src,
             dst,
             target_block,
@@ -945,7 +945,7 @@ impl Connection {
         // and there are more tables to be copied, we can start more
         // workers, up to `self.workers` many
         let mut workers = Vec::new();
-        while !state.unfinished.is_empty() && !workers.is_empty() {
+        while !state.unfinished.is_empty() || !workers.is_empty() {
             // We usually add at least one job here, except if we are out of
             // tables to copy. In that case, we go through the `while` loop
             // every time one of the tables we are currently copying