store/postgres: Make subgraph creation not block writers

Creating a subgraph can block all other subgraph writers if the creation
itself has to wait for its lock, e.g., because of an ongoing autovacuum of
event_meta_data. That wait can be long (an hour or more in the case of
autovacuum)

We now do subgraph creation in such a way that we wait for a lock only a
short time (2s) and then sleep for a bit so that other write activity can
proceed.

Author: lutter

store/postgres/src/store.rs

@@ -1026,13 +1026,69 @@ impl StoreTrait for Store {
         subgraph_id: &SubgraphDeploymentId,
         ops: Vec<EntityOperation>,
     ) -> Result<(), StoreError> {
+        // Various timing parameters, all in seconds
+        const INITIAL_DELAY: u64 = 2;
+        const MAX_DELAY: u64 = 64;
+        const LOCK_TIMEOUT: u64 = 2;
+
         let conn = self.get_conn().map_err(Error::from)?;
         let econn = e::Connection::new(&conn);
-
-        conn.transaction(|| {
-            self.apply_entity_operations_with_conn(&econn, ops, None)?;
-            crate::entities::create_schema(&conn, subgraph_id)
-        })
+        let mut delay = Duration::from_secs(INITIAL_DELAY);
+
+        // Creating a subgraph creates a table that references
+        // `event_meta_data`.  To validate that reference, Postgres takes a
+        // `share update exclusive` lock; for this lock, Postgres has to
+        // wait for all write activity that started before the lock was
+        // requested to finish, and it also has to hold all write activity
+        // that starts after the lock request. Usually, this is not a
+        // problem as the lock is only held for a very short amount of
+        // time.
+        //
+        // If there is other activity, like an autovacuum, happening
+        // already when the lock is requested though, we have to wait until
+        // that activity finishes, which in the case of an autovacuum can
+        // be an hour or longer. The autovacuum by itself is not a problem,
+        // as it still allows writes to happen, but once the lock request
+        // from this code gets into the lock queue, write activity also has
+        // to wait for the autovacuum to finish, effectively blocking all
+        // subgraph indexing until the autovacuum has finished.
+        //
+        // To avoid this, we set a lock timeout of 2s, which should be long
+        // enough to get the lock under normal conditions, but not so long
+        // that it materially impedes indexing in the above situation. If
+        // we can not get the lock within 2s, the subgraph creation fails,
+        // and we sleep an increasing amount of time (up to about a minute)
+        // and then retry the subgraph creation.
+        loop {
+            let start = Instant::now();
+            let result = conn.transaction(|| -> Result<(), StoreError> {
+                self.apply_entity_operations_with_conn(&econn, ops.clone(), None)?;
+                conn.batch_execute(&format!("set local lock_timeout to '{}s'", LOCK_TIMEOUT))?;
+                crate::entities::create_schema(&conn, subgraph_id)
+            });
+            if let Err(StoreError::Unknown(_)) = &result {
+                // There is no robust way to actually find out that we timed
+                // out on the lock from the error message; diesel shields us
+                // from these details too much. Rather than grep the error
+                // message, which would be very fragile, we assume that if a
+                // failure occurred after more than LOCK_TIMEOUT seconds that
+                // it was because we timed out on the lock and try again.
+                if start.elapsed() >= Duration::from_secs(LOCK_TIMEOUT) {
+                    debug!(
+                        self.logger,
+                        "could not acquire lock for creation of subgraph {}, trying again in {}s",
+                        &subgraph_id,
+                        delay.as_secs()
+                    );
+                    std::thread::sleep(delay);
+                    if delay.as_secs() < MAX_DELAY {
+                        delay *= 2;
+                    }
+                    continue;
+                }
+            }
+            break result;
+        }
     }
 
     fn start_subgraph_deployment(

store/postgres/tests/store.rs

@@ -2145,3 +2145,50 @@ fn subgraph_schema_types_have_subgraph_id_directive() {
         Ok(())
     })
 }
+
+#[test]
+fn create_subgraph_deployment_tolerates_locks() {
+    run_test(|store| -> Result<(), ()> {
+        use diesel::connection::SimpleConnection;
+        use diesel::connection::TransactionManager;
+        use std::sync::Barrier;
+
+        const BLOCK_TIME: u64 = 3;
+
+        let url = postgres_test_url();
+        let barrier = Arc::new(Barrier::new(2));
+        let blocker_barrier = barrier.clone();
+        // Start a thread that will take a lock for BLOCK_TIME seconds and
+        // therefore block subgraph creation during that time
+        std::thread::spawn(move || {
+            let blocker =
+                PgConnection::establish(url.as_str()).expect("Failed to connect to Postgres");
+            blocker
+                .transaction_manager()
+                .begin_transaction(&blocker)
+                .expect("Failed to set up transaction for blocker");
+            blocker
+                .batch_execute("lock table event_meta_data in share update exclusive mode")
+                .expect("Failed to lock event_meta_data");
+            blocker_barrier.wait();
+            std::thread::sleep(Duration::from_secs(BLOCK_TIME));
+            blocker
+                .transaction_manager()
+                .rollback_transaction(&blocker)
+                .expect("Failed to roll blocker transaction back");
+        });
+
+        // While we are blocking, try to create a subgraph. We don't really have
+        // a way to check from the outside that this does not block other write
+        // activity, but it is visible in the logs if this test is run with
+        // GRAPH_LOG=debug
+        let subgraph_id = SubgraphDeploymentId::new("DeploymentLocking").unwrap();
+        barrier.wait();
+        let start = std::time::Instant::now();
+        store
+            .create_subgraph_deployment(&subgraph_id, vec![])
+            .expect("Subgraph creation failed");
+        assert!(start.elapsed() >= Duration::from_secs(BLOCK_TIME));
+        Ok(())
+    })
+}