fix: Add workaround for multiprocess engine forking (#6499)

app/__init__.py

@@ -31,7 +31,7 @@
 
 import stripe
 from app.settings import get_settings
-from app.models import db
+from app.models import db, add_engine_pidguard
 from app.api.helpers.jwt import jwt_user_loader
 from app.api.helpers.cache import cache
 from werkzeug.middleware.profiler import ProfilerMiddleware
@@ -168,6 +168,8 @@ def create_app():
         app.register_blueprint(alipay_blueprint)
         app.register_blueprint(admin_misc_routes)
 
+        add_engine_pidguard(db.engine)
+
     sa.orm.configure_mappers()
 
     if app.config['SERVE_STATIC']:

app/models/__init__.py

@@ -1,3 +1,6 @@
+import os
+import warnings
+
 from flask_sqlalchemy import SQLAlchemy
 from sqlalchemy_continuum import make_versioned
 from sqlalchemy_continuum.plugins import FlaskPlugin
@@ -7,3 +10,37 @@
 })
 
 db = SQLAlchemy()
+
+
+# https://docs.sqlalchemy.org/en/13/faq/connections.html#how-do-i-use-engines-connections-sessions-with-python-multiprocessing-or-os-fork
+def add_engine_pidguard(engine):
+    from sqlalchemy import event, exc # placed here for import conflict resolution
+    """Add multiprocessing guards.
+
+    Forces a connection to be reconnected if it is detected
+    as having been shared to a sub-process.
+
+    """
+
+    @event.listens_for(engine, "connect")
+    def connect(dbapi_connection, connection_record):
+        print(f'connected to pid { os.getpid() }')
+        connection_record.info['pid'] = os.getpid()
+
+    @event.listens_for(engine, "checkout")
+    def checkout(dbapi_connection, connection_record, connection_proxy):
+        pid = os.getpid()
+        print(f'checkout pid {pid}')
+        if connection_record.info['pid'] != pid:
+            # substitute log.debug() or similar here as desired
+            warnings.warn(
+                "Parent process %(orig)s forked (%(newproc)s) with an open "
+                "database connection, "
+                "which is being discarded and recreated." %
+                {"newproc": pid, "orig": connection_record.info['pid']})
+            connection_record.connection = connection_proxy.connection = None
+            raise exc.DisconnectionError(
+                "Connection record belongs to pid %s, "
+                "attempting to check out in pid %s" %
+                (connection_record.info['pid'], pid)
+            )