0001-sql-lock-wcs_meta-during-schema-updates-57017.patch

Benjamin Dauvergne, 16 septembre 2021 23:14

Voir les différences: en ligne côte à côte

Subject: [PATCH] sql: lock wcs_meta during schema updates (#57017)

 wcs/sql.py | 114 +++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 114 insertions(+)

     # You should have received a copy of the GNU General Public License
     # along with this program; if not, see <http://www.gnu.org/licenses/>.
     import contextvars
     import copy
     import datetime
     import functools
     import io
     import json
     import re
-...
         return get_publisher().pgconn
     def lock_wcs_meta():
         conn, cur = get_connection_and_cursor()
         do_meta_table(conn, cur, insert_current_sql_level=False)
         cur.execute('LOCK wcs_meta')
     def cleanup_connection():
         if hasattr(get_publisher(), 'pgconn') and get_publisher().pgconn is not None:
             get_publisher().pgconn.close()
-...
         return f
     class RedoWithLock(Exception):
         pass
     locked = contextvars.ContextVar('locked', default=False)
     guard_lock_depth = contextvars.ContextVar('guard_lock_depth', default=0)
     def lock():
         if not locked.get():
             raise RedoWithLock
     def guard_lock(func):
         @functools.wraps(func)
         def wrapper(*args, **kwargs):
             depth = guard_lock_depth.get()
             depth_token = guard_lock_depth.set(depth + 1)
             try:
                 try:
                     return func(*args, **kwargs)
                 except RedoWithLock:
                     if depth > 0:
                         raise
                     token = locked.set(True)
                     try:
                         lock_wcs_meta()
                         return func(*args, **kwargs)
                     finally:
                         locked.reset(token)
             finally:
                 guard_lock_depth.reset(depth_token)
         return wrapper
     @guard_postgres
     @guard_lock
     def do_formdef_tables(formdef, conn=None, cur=None, rebuild_views=False, rebuild_global_views=True):
         if formdef.id is None:
             return []
-...
             (table_name,),
+        )
         if cur.fetchone()[0] == 0:
             lock()
             cur.execute(
                 '''CREATE TABLE %s (id serial PRIMARY KEY,
                                         user_id varchar,
-...
         # migrations
         if 'fts' not in existing_fields:
             # full text search
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN fts tsvector''' % table_name)
             cur.execute('''CREATE INDEX %s_fts ON %s USING gin(fts)''' % (table_name, table_name))
         if 'workflow_roles' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN workflow_roles bytea''' % table_name)
             cur.execute('''ALTER TABLE %s ADD COLUMN workflow_roles_array text[]''' % table_name)
         if 'concerned_roles_array' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN concerned_roles_array text[]''' % table_name)
         if 'actions_roles_array' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN actions_roles_array text[]''' % table_name)
         if 'page_no' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN page_no varchar''' % table_name)
         if 'anonymised' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN anonymised timestamptz''' % table_name)
         if 'tracking_code' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN tracking_code varchar''' % table_name)
         if 'backoffice_submission' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN backoffice_submission boolean''' % table_name)
         if 'submission_context' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN submission_context bytea''' % table_name)
         if 'submission_agent_id' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN submission_agent_id varchar''' % table_name)
         if 'submission_channel' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN submission_channel varchar''' % table_name)
         if 'criticality_level' not in existing_fields:
             lock()
             cur.execute(
                 '''ALTER TABLE %s ADD COLUMN criticality_level integer NOT NULL DEFAULT(0)''' % table_name
+            )
         if 'last_update_time' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN last_update_time timestamp''' % table_name)
         if 'digests' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN digests jsonb''' % table_name)
         if 'user_label' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN user_label varchar''' % table_name)
         if 'prefilling_data' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN prefilling_data bytea''' % table_name)
         # add new fields
-...
                 continue
             needed_fields.add(get_field_id(field))
             if get_field_id(field) not in existing_fields:
                 lock()
                 cur.execute('''ALTER TABLE %s ADD COLUMN %s %s''' % (table_name, get_field_id(field), sql_type))
             if field.store_display_value:
                 needed_fields.add('%s_display' % get_field_id(field))
                 if '%s_display' % get_field_id(field) not in existing_fields:
                     lock()
                     cur.execute(
                         '''ALTER TABLE %s ADD COLUMN %s varchar'''
                         % (table_name, '%s_display' % get_field_id(field))
-...
             if field.store_structured_value:
                 needed_fields.add('%s_structured' % get_field_id(field))
                 if '%s_structured' % get_field_id(field) not in existing_fields:
                     lock()
                     cur.execute(
                         '''ALTER TABLE %s ADD COLUMN %s bytea'''
                         % (table_name, '%s_structured' % get_field_id(field))
-...
             column_name = 'geoloc_%s' % field
             needed_fields.add(column_name)
             if column_name not in existing_fields:
                 lock()
                 cur.execute('ALTER TABLE %s ADD COLUMN %s %s' '' % (table_name, column_name, 'POINT'))
         # delete obsolete fields
         for field in existing_fields - needed_fields:
             lock()
             cur.execute('''ALTER TABLE %s DROP COLUMN %s CASCADE''' % (table_name, field))
         # migrations on _evolutions table
-...
+        )
         evo_existing_fields = {x[0] for x in cur.fetchall()}
         if 'last_jump_datetime' not in evo_existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s_evolutions ADD COLUMN last_jump_datetime timestamp''' % table_name)
         if rebuild_views or len(existing_fields - needed_fields):
             # views may have been dropped when dropping columns, so we recreate
             # them even if not asked to.
             lock()
             redo_views(conn, cur, formdef, rebuild_global_views=rebuild_global_views)
         if own_conn:
-...
         return actions
     @guard_lock
     def do_formdef_indexes(formdef, created, conn, cur, concurrently=False):
         table_name = get_formdef_table_name(formdef)
         evolutions_table_name = table_name + '_evolutions'
-...
             create_index = 'CREATE INDEX CONCURRENTLY'
         if evolutions_table_name + '_fid' not in existing_indexes:
             lock()
             cur.execute(
                 '''%s %s_fid ON %s (formdata_id)''' % (create_index, evolutions_table_name, evolutions_table_name)
+            )
         for attr in ('receipt_time', 'anonymised', 'user_id', 'status'):
             if table_name + '_' + attr + '_idx' not in existing_indexes:
                 lock()
                 cur.execute(
                     '%(create_index)s %(table_name)s_%(attr)s_idx ON %(table_name)s (%(attr)s)'
                     % {'create_index': create_index, 'table_name': table_name, 'attr': attr}
-...
     @guard_postgres
     @guard_lock
     def do_user_table():
         conn, cur = get_connection_and_cursor()
         table_name = 'users'
-...
             (table_name,),
+        )
         if cur.fetchone()[0] == 0:
             lock()
             cur.execute(
                 '''CREATE TABLE %s (id serial PRIMARY KEY,
                                         name varchar,
-...
                 continue
             needed_fields.add(get_field_id(field))
             if get_field_id(field) not in existing_fields:
                 lock()
                 cur.execute('''ALTER TABLE %s ADD COLUMN %s %s''' % (table_name, get_field_id(field), sql_type))
             if field.store_display_value:
                 needed_fields.add('%s_display' % get_field_id(field))
                 if '%s_display' % get_field_id(field) not in existing_fields:
                     lock()
                     cur.execute(
                         '''ALTER TABLE %s ADD COLUMN %s varchar'''
                         % (table_name, '%s_display' % get_field_id(field))
-...
             if field.store_structured_value:
                 needed_fields.add('%s_structured' % get_field_id(field))
                 if '%s_structured' % get_field_id(field) not in existing_fields:
                     lock()
                     cur.execute(
                         '''ALTER TABLE %s ADD COLUMN %s bytea'''
                         % (table_name, '%s_structured' % get_field_id(field))
-...
         # migrations
         if 'fts' not in existing_fields:
             # full text search
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN fts tsvector''' % table_name)
             cur.execute('''CREATE INDEX %s_fts ON %s USING gin(fts)''' % (table_name, table_name))
         if 'verified_fields' not in existing_fields:
             lock()
             cur.execute('ALTER TABLE %s ADD COLUMN verified_fields text[]' % table_name)
         if 'ascii_name' not in existing_fields:
             lock()
             cur.execute('ALTER TABLE %s ADD COLUMN ascii_name varchar' % table_name)
         if 'deleted_timestamp' not in existing_fields:
             lock()
             cur.execute('ALTER TABLE %s ADD COLUMN deleted_timestamp timestamp' % table_name)
         if 'is_active' not in existing_fields:
             lock()
             cur.execute('ALTER TABLE %s ADD COLUMN is_active bool DEFAULT TRUE' % table_name)
             cur.execute('UPDATE %s SET is_active = FALSE WHERE deleted_timestamp IS NOT NULL' % table_name)
         # delete obsolete fields
         for field in existing_fields - needed_fields:
             lock()
             cur.execute('''ALTER TABLE %s DROP COLUMN %s''' % (table_name, field))
         conn.commit()
         try:
             lock()
             cur.execute('CREATE INDEX users_name_idx ON users (name)')
             conn.commit()
         except psycopg2.ProgrammingError:
-...
         cur.close()
     @guard_lock
     def do_role_table(concurrently=False):
         conn, cur = get_connection_and_cursor()
         table_name = 'roles'
-...
             (table_name,),
+        )
         if cur.fetchone()[0] == 0:
             lock()
             cur.execute(
                 '''CREATE TABLE %s (id VARCHAR PRIMARY KEY,
                                     name VARCHAR,
-...
         # delete obsolete fields
         for field in existing_fields - needed_fields:
             lock()
             cur.execute('''ALTER TABLE %s DROP COLUMN %s''' % (table_name, field))
         conn.commit()
-...
             role.store()
     @guard_lock
     def do_tracking_code_table():
         conn, cur = get_connection_and_cursor()
         table_name = 'tracking_codes'
-...
             (table_name,),
+        )
         if cur.fetchone()[0] == 0:
             lock()
             cur.execute(
                 '''CREATE TABLE %s (id varchar PRIMARY KEY,
                                         formdef_id varchar,
-...
         # delete obsolete fields
         for field in existing_fields - needed_fields:
             lock()
             cur.execute('''ALTER TABLE %s DROP COLUMN %s''' % (table_name, field))
         conn.commit()
         cur.close()
     @guard_lock
     def do_session_table():
         conn, cur = get_connection_and_cursor()
         table_name = 'sessions'
-...
             (table_name,),
+        )
         if cur.fetchone()[0] == 0:
             lock()
             cur.execute(
                 '''CREATE TABLE %s (id varchar PRIMARY KEY,
                                             session_data bytea,
-...
         # migrations
         if 'last_update_time' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN last_update_time timestamp DEFAULT NOW()''' % table_name)
             cur.execute('''CREATE INDEX %s_ts ON %s (last_update_time)''' % (table_name, table_name))
         # delete obsolete fields
         for field in existing_fields - needed_fields:
             lock()
             cur.execute('''ALTER TABLE %s DROP COLUMN %s''' % (table_name, field))
         conn.commit()
         cur.close()
     @guard_lock
     def do_custom_views_table():
         conn, cur = get_connection_and_cursor()
         table_name = 'custom_views'
-...
             (table_name,),
+        )
         if cur.fetchone()[0] == 0:
             lock()
             cur.execute(
                 '''CREATE TABLE %s (id varchar PRIMARY KEY,
                                             title varchar,
-...
         # migrations
         if 'is_default' not in existing_fields:
             lock()
             cur.execute('''ALTER TABLE %s ADD COLUMN is_default boolean DEFAULT FALSE''' % table_name)
         # delete obsolete fields
         for field in existing_fields - needed_fields:
             lock()
             cur.execute('''ALTER TABLE %s DROP COLUMN %s''' % (table_name, field))
         conn.commit()
         cur.close()
     @guard_lock
     def do_snapshots_table():
         conn, cur = get_connection_and_cursor()
         table_name = 'snapshots'
-...
             (table_name,),
+        )
         if cur.fetchone()[0] == 0:
             lock()
             cur.execute(
                 '''CREATE TABLE %s (id SERIAL,
                                             object_type VARCHAR,
-...
         # delete obsolete fields
         for field in existing_fields - needed_fields:
             lock()
             cur.execute('''ALTER TABLE %s DROP COLUMN %s''' % (table_name, field))
         conn.commit()
         cur.close()
     @guard_lock
     def do_loggederrors_table(concurrently=False):
         conn, cur = get_connection_and_cursor()
         table_name = 'loggederrors'
-...
             (table_name,),
+        )
         if cur.fetchone()[0] == 0:
             lock()
             cur.execute(
                 '''CREATE TABLE %s (id SERIAL PRIMARY KEY,
                                             tech_id VARCHAR UNIQUE,
-...
         # delete obsolete fields
         for field in existing_fields - needed_fields:
             lock()
             cur.execute('''ALTER TABLE %s DROP COLUMN %s''' % (table_name, field))
         create_index = 'CREATE INDEX'
-...
         for attr in ('formdef_id', 'workflow_id'):
             if table_name + '_' + attr + '_idx' not in existing_indexes:
                 lock()
                 cur.execute(
                     '%(create_index)s %(table_name)s_%(attr)s_idx ON %(table_name)s (%(attr)s)'
                     % {'create_index': create_index, 'table_name': table_name, 'attr': attr}
-...
     @guard_postgres
     @guard_lock
     def drop_views(formdef, conn, cur):
         # remove the global views
         drop_global_views(conn, cur)
-...
             view_names.append(row[0])
         for view_name in view_names:
             lock()
             cur.execute('''DROP VIEW IF EXISTS %s''' % view_name)
-...
     @guard_postgres
     @guard_lock
     def do_views(formdef, conn, cur, rebuild_global_views=True):
         # create new view
         table_name = get_formdef_table_name(formdef)
-...
         fields_list = ', '.join(['%s AS %s' % (force_text(x), force_text(y)) for (x, y) in view_fields])
         lock()
         cur.execute('''CREATE VIEW %s AS SELECT %s FROM %s''' % (view_name, fields_list, table_name))
         if rebuild_global_views:
             do_global_views(conn, cur)  # recreate global views
     @guard_lock
     def drop_global_views(conn, cur):
         lock()
         cur.execute(
             '''SELECT table_name FROM information_schema.views
                         WHERE table_schema = 'public'
-...
         cur.execute('''DROP VIEW IF EXISTS wcs_all_forms''')
     @guard_lock
     def do_global_views(conn, cur):
         # recreate global views
         from wcs.formdef import FormDef
-...
         if not view_names:
             return
         lock()
         cur.execute('''DROP VIEW IF EXISTS wcs_all_forms CASCADE''')
         fake_formdef = FormDef()
-...
     @guard_postgres
     @guard_lock
     def migrate():
         conn, cur = get_connection_and_cursor()
         sql_level = get_sql_level(conn, cur)
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-sql-lock-wcs_meta-during-schema-updates-57017.patch