store: Make target duration for batch operations configurable

Also lower the default to 3 minutes; 5 minutes is too much on busy systems

Author: lutter

docs/environment-variables.md

@@ -200,3 +200,7 @@ those.
   identified as unused, `graph-node` will wait at least this long before
   actually deleting the data (value is in minutes, defaults to 360, i.e. 6
   hours)
+- `GRAPH_STORE_BATCH_TARGET_DURATION`: How long batch operations during
+  copying or grafting should take. This limits how long transactions for
+  such long running operations will be, and therefore helps control bloat
+  in other tables. Value is in seconds and defaults to 180s.

graph/src/env/store.rs

@@ -86,6 +86,11 @@ pub struct EnvVarsStore {
     /// Setting this to `0` disables pipelined writes, and writes will be
     /// done synchronously.
     pub write_queue_size: usize,
+
+    /// How long batch operations during copying or grafting should take.
+    /// Set by `GRAPH_STORE_BATCH_TARGET_DURATION` (expressed in seconds).
+    /// The default is 180s.
+    pub batch_target_duration: Duration,
 }
 
 // This does not print any values avoid accidentally leaking any sensitive env vars
@@ -122,6 +127,7 @@ impl From<InnerStore> for EnvVarsStore {
             connection_min_idle: x.connection_min_idle,
             connection_idle_timeout: Duration::from_secs(x.connection_idle_timeout_in_secs),
             write_queue_size: x.write_queue_size,
+            batch_target_duration: Duration::from_secs(x.batch_target_duration_in_secs),
         }
     }
 }
@@ -165,4 +171,6 @@ pub struct InnerStore {
     connection_idle_timeout_in_secs: u64,
     #[envconfig(from = "GRAPH_STORE_WRITE_QUEUE", default = "5")]
     write_queue_size: usize,
+    #[envconfig(from = "GRAPH_STORE_BATCH_TARGET_DURATION", default = "180")]
+    batch_target_duration_in_secs: u64,
 }

store/postgres/src/copy.rs

@@ -34,7 +34,7 @@ use diesel::{
 use graph::{
     components::store::EntityType,
     constraint_violation,
-    prelude::{info, o, warn, BlockNumber, BlockPtr, Logger, StoreError},
+    prelude::{info, o, warn, BlockNumber, BlockPtr, Logger, StoreError, ENV_VARS},
 };
 
 use crate::{
@@ -51,7 +51,7 @@ const INITIAL_BATCH_SIZE: i64 = 10_000;
 /// arrays can be large and large arrays will slow down copying a lot. We
 /// therefore tread lightly in that case
 const INITIAL_BATCH_SIZE_LIST: i64 = 100;
-const TARGET_DURATION: Duration = Duration::from_secs(5 * 60);
+
 const LOG_INTERVAL: Duration = Duration::from_secs(3 * 60);
 
 /// If replicas are lagging by more than this, the copying code will pause
@@ -308,8 +308,9 @@ impl AdaptiveBatchSize {
     pub fn adapt(&mut self, duration: Duration) {
         // Avoid division by zero
         let duration = duration.as_millis().max(1);
-        let new_batch_size =
-            self.size as f64 * TARGET_DURATION.as_millis() as f64 / duration as f64;
+        let new_batch_size = self.size as f64
+            * ENV_VARS.store.batch_target_duration.as_millis() as f64
+            / duration as f64;
         self.size = (2 * self.size).min(new_batch_size.round() as i64);
     }
 }