From ead87e3727e63041fe3d68fe2f9bf6c4295b4662 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Pawe=C5=82=20Chmielowski?= Date: Tue, 16 Jul 2024 15:36:14 +0200 Subject: [PATCH] Add option update_sql_schema_timeout to allow schema update use longer timeouts This also makes batch of schema updates to single table use transaction, which should help in not leaving table in inconsistent state if some update steps fails (unless you use mysql where you can't rollback changes to table schemas). --- src/ejabberd_option.erl | 8 + src/ejabberd_options.erl | 3 + src/ejabberd_options_doc.erl | 6 + src/ejabberd_sql_schema.erl | 474 +++++++++++++++++------------------ 4 files changed, 254 insertions(+), 237 deletions(-) diff --git a/src/ejabberd_option.erl b/src/ejabberd_option.erl index e7dc04da0..bc0571e81 100644 --- a/src/ejabberd_option.erl +++ b/src/ejabberd_option.erl @@ -164,6 +164,7 @@ -export([sql_username/0, sql_username/1]). -export([trusted_proxies/0]). -export([update_sql_schema/0]). +-export([update_sql_schema_timeout/0, update_sql_schema_timeout/1]). -export([use_cache/0, use_cache/1]). -export([validate_stream/0]). -export([version/0]). @@ -1109,6 +1110,13 @@ trusted_proxies() -> update_sql_schema() -> ejabberd_config:get_option({update_sql_schema, global}). +-spec update_sql_schema_timeout() -> 'infinity' | pos_integer(). +update_sql_schema_timeout() -> + update_sql_schema_timeout(global). +-spec update_sql_schema_timeout(global | binary()) -> 'infinity' | pos_integer(). +update_sql_schema_timeout(Host) -> + ejabberd_config:get_option({update_sql_schema_timeout, Host}). + -spec use_cache() -> boolean(). use_cache() -> use_cache(global). diff --git a/src/ejabberd_options.erl b/src/ejabberd_options.erl index b6bb18d5a..c30cc072f 100644 --- a/src/ejabberd_options.erl +++ b/src/ejabberd_options.erl @@ -264,6 +264,8 @@ opt_type(new_sql_schema) -> econf:bool(); opt_type(update_sql_schema) -> econf:bool(); +opt_type(update_sql_schema_timeout) -> + econf:timeout(second, infinity); opt_type(oauth_access) -> econf:acl(); opt_type(oauth_cache_life_time) -> @@ -613,6 +615,7 @@ options() -> {net_ticktime, timer:seconds(60)}, {new_sql_schema, ?USE_NEW_SQL_SCHEMA_DEFAULT}, {update_sql_schema, true}, + {update_sql_schema_timeout, timer:minutes(5)}, {oauth_access, none}, {oauth_cache_life_time, fun(Host) -> ejabberd_config:get_option({cache_life_time, Host}) end}, diff --git a/src/ejabberd_options_doc.erl b/src/ejabberd_options_doc.erl index 061b1dccb..2e2313c97 100644 --- a/src/ejabberd_options_doc.erl +++ b/src/ejabberd_options_doc.erl @@ -929,6 +929,12 @@ doc() -> "This option was added in ejabberd 23.10, " "and enabled by default since 24.06. " "The default value is 'true'.")}}, + {update_sql_schema_timeout, + #{value => "timeout()", + note => "added in 24.07", + desc => + ?T("Time allocated to SQL schema update queries. " + "The default value is set to 5 minutes.")}}, {oauth_access, #{value => ?T("AccessName"), desc => ?T("By default creating OAuth tokens is not allowed. " diff --git a/src/ejabberd_sql_schema.erl b/src/ejabberd_sql_schema.erl index 91b8aab72..644844714 100644 --- a/src/ejabberd_sql_schema.erl +++ b/src/ejabberd_sql_schema.erl @@ -226,6 +226,13 @@ store_version(Host, Module, Version) -> ["!module=%(SModule)s", "version=%(Version)d"]). +store_version_t(Module, Version) -> + SModule = misc:atom_to_binary(Module), + ?SQL_UPSERT_T( + "schema_version", + ["!module=%(SModule)s", + "version=%(Version)d"]). + table_exists(Host, Table) -> ejabberd_sql:sql_query( Host, @@ -398,28 +405,25 @@ get_current_version(Host, Module, Schemas) -> Version end. -sqlite_table_copy(Host, SchemaInfo, Table) -> - ejabberd_sql:sql_transaction(Host, - fun() -> - TableName = Table#sql_table.name, - NewTableName = <<"new_", TableName/binary>>, - NewTable = Table#sql_table{name = NewTableName}, - create_table_t(SchemaInfo, NewTable), - SQL2 = <<"INSERT INTO ", NewTableName/binary, - " SELECT * FROM ", TableName/binary>>, - ?INFO_MSG("Copying table ~s to ~s:~n~s~n", - [TableName, NewTableName, SQL2]), - ejabberd_sql:sql_query_t(SQL2), - SQL3 = <<"DROP TABLE ", TableName/binary>>, - ?INFO_MSG("Droping old table ~s:~n~s~n", - [TableName, SQL2]), - ejabberd_sql:sql_query_t(SQL3), - SQL4 = <<"ALTER TABLE ", NewTableName/binary, - " RENAME TO ", TableName/binary>>, - ?INFO_MSG("Renameing table ~s to ~s:~n~s~n", - [NewTableName, TableName, SQL4]), - ejabberd_sql:sql_query_t(SQL4) - end). +sqlite_table_copy_t(SchemaInfo, Table) -> + TableName = Table#sql_table.name, + NewTableName = <<"new_", TableName/binary>>, + NewTable = Table#sql_table{name = NewTableName}, + create_table_t(SchemaInfo, NewTable), + SQL2 = <<"INSERT INTO ", NewTableName/binary, + " SELECT * FROM ", TableName/binary>>, + ?INFO_MSG("Copying table ~s to ~s:~n~s~n", + [TableName, NewTableName, SQL2]), + ejabberd_sql:sql_query_t(SQL2), + SQL3 = <<"DROP TABLE ", TableName/binary>>, + ?INFO_MSG("Droping old table ~s:~n~s~n", + [TableName, SQL2]), + ejabberd_sql:sql_query_t(SQL3), + SQL4 = <<"ALTER TABLE ", NewTableName/binary, + " RENAME TO ", TableName/binary>>, + ?INFO_MSG("Renameing table ~s to ~s:~n~s~n", + [NewTableName, TableName, SQL4]), + ejabberd_sql:sql_query_t(SQL4). format_type(#sql_schema_info{db_type = pgsql}, Column) -> case Column#sql_column.type of @@ -875,18 +879,18 @@ update_schema(Host, Module, RawSchemas) -> end. do_update_schema(Host, Module, SchemaInfo, Schema) -> - lists:foreach( - fun({add_column, TableName, ColumnName}) -> - {value, Table} = - lists:keysearch( - TableName, #sql_table.name, Schema#sql_schema.tables), - {value, Column} = - lists:keysearch( - ColumnName, #sql_column.name, Table#sql_table.columns), - Res = - ejabberd_sql:sql_query( - Host, - fun(DBType, _DBVersion) -> + F = fun() -> + lists:foreach( + fun({add_column, TableName, ColumnName}) -> + {value, Table} = + lists:keysearch( + TableName, #sql_table.name, Schema#sql_schema.tables), + {value, Column} = + lists:keysearch( + ColumnName, #sql_column.name, Table#sql_table.columns), + Res = + ejabberd_sql:sql_query_t( + fun(DBType, _DBVersion) -> Def = format_column_def(SchemaInfo, Column), Default = format_default(SchemaInfo, Column), SQLs = @@ -911,209 +915,205 @@ do_update_schema(Host, Module, SchemaInfo, Schema) -> ColumnName, SQLs]), lists:foreach( - fun(SQL) -> ejabberd_sql:sql_query_t(SQL) end, - SQLs) - end), - case Res of - {error, Error} -> - ?ERROR_MSG("Failed to update table ~s: ~p", - [TableName, Error]), - error(Error); - _ -> - ok - end; - ({drop_column, TableName, ColumnName}) -> - Res = - ejabberd_sql:sql_query( - Host, - fun(_DBType, _DBVersion) -> - SQL = [<<"ALTER TABLE ">>, - TableName, - <<" DROP COLUMN ">>, - ColumnName, - <<";">>], - ?INFO_MSG("Drop column ~s/~s:~n~s~n", - [TableName, - ColumnName, - SQL]), - ejabberd_sql:sql_query_t(SQL) - end), - case Res of - {error, Error} -> - ?ERROR_MSG("Failed to update table ~s: ~p", - [TableName, Error]), - error(Error); - _ -> - ok - end; - ({create_index, TableName, Columns1}) -> - Columns = - case ejabberd_sql:use_new_schema() of - true -> - Columns1; - false -> - lists:delete( - <<"server_host">>, Columns1) - end, - {value, Table} = - lists:keysearch( - TableName, #sql_table.name, Schema#sql_schema.tables), - {value, Index} = - lists:keysearch( - Columns, #sql_index.columns, Table#sql_table.indices), - case Index#sql_index.meta of - #{ignore := true} -> ok; - _ -> - Res = - ejabberd_sql:sql_query( - Host, - fun() -> - case Index#sql_index.meta of - #{primary_key := true} -> - SQL1 = format_add_primary_key( - SchemaInfo, Table, Index), - SQL = iolist_to_binary(SQL1), - ?INFO_MSG("Add primary key ~s/~p:~n~s~n", - [Table#sql_table.name, - Index#sql_index.columns, - SQL]), - ejabberd_sql:sql_query_t(SQL); - _ -> - SQL1 = format_create_index( - SchemaInfo, Table, Index), - SQL = iolist_to_binary(SQL1), - ?INFO_MSG("Create index ~s/~p:~n~s~n", - [Table#sql_table.name, - Index#sql_index.columns, - SQL]), - ejabberd_sql:sql_query_t(SQL) - end - end), - case Res of - {error, Error} -> - ?ERROR_MSG("Failed to update table ~s: ~p", - [TableName, Error]), - error(Error); - _ -> - ok - end - end; - ({update_primary_key, TableName, Columns1}) -> - Columns = - case ejabberd_sql:use_new_schema() of - true -> - Columns1; - false -> - lists:delete( - <<"server_host">>, Columns1) - end, - {value, Table} = - lists:keysearch( - TableName, #sql_table.name, Schema#sql_schema.tables), - {value, Index} = - lists:keysearch( - Columns, #sql_index.columns, Table#sql_table.indices), - Res = - case SchemaInfo#sql_schema_info.db_type of - sqlite -> - sqlite_table_copy(Host, SchemaInfo, Table); - pgsql -> - TableName = Table#sql_table.name, - SQL1 = [<<"ALTER TABLE ">>, TableName, <<" DROP CONSTRAINT ", - TableName/binary,"_pkey, ", - "ADD PRIMARY KEY (">>, - lists:join( - <<", ">>, - Index#sql_index.columns), - <<");">>], - SQL = iolist_to_binary(SQL1), - ?INFO_MSG("Update primary key ~s/~p:~n~s~n", - [Table#sql_table.name, - Index#sql_index.columns, - SQL]), - ejabberd_sql:sql_query( - Host, - fun(_DBType, _DBVersion) -> - ejabberd_sql:sql_query_t(SQL) - end); - mysql -> - TableName = Table#sql_table.name, - SQL1 = [<<"ALTER TABLE ">>, TableName, <<" DROP PRIMARY KEY, " - "ADD PRIMARY KEY (">>, - lists:join( - <<", ">>, - lists:map( - fun(Col) -> - format_mysql_index_column(Table, Col) - end, Index#sql_index.columns)), - <<");">>], - SQL = iolist_to_binary(SQL1), - ?INFO_MSG("Update primary key ~s/~p:~n~s~n", - [Table#sql_table.name, - Index#sql_index.columns, - SQL]), - ejabberd_sql:sql_query( - Host, - fun(_DBType, _DBVersion) -> - ejabberd_sql:sql_query_t(SQL) - end) - end, - case Res of - {error, Error} -> - ?ERROR_MSG("Failed to update table ~s: ~p", - [TableName, Error]), - error(Error); - _ -> - ok - end; - ({drop_index, TableName, Columns1}) -> - Columns = - case ejabberd_sql:use_new_schema() of - true -> - Columns1; - false -> - lists:delete( - <<"server_host">>, Columns1) - end, - case find_index_name(Host, TableName, Columns) of - false -> - ?ERROR_MSG("Can't find an index to drop for ~s/~p", - [TableName, Columns]); - {ok, IndexName} -> - Res = - ejabberd_sql:sql_query( - Host, - fun(DBType, _DBVersion) -> - SQL = - case DBType of - mysql -> - [<<"DROP INDEX ">>, - IndexName, - <<" ON ">>, - TableName, - <<";">>]; - _ -> - [<<"DROP INDEX ">>, - IndexName, <<";">>] - end, - ?INFO_MSG("Drop index ~s/~p:~n~s~n", - [TableName, - Columns, - SQL]), - ejabberd_sql:sql_query_t(SQL) - end), - case Res of - {error, Error} -> - ?ERROR_MSG("Failed to update table ~s: ~p", - [TableName, Error]), - error(Error); - _ -> - ok - end - end - end, Schema#sql_schema.update), - store_version(Host, Module, Schema#sql_schema.version). - + fun(SQL) -> ejabberd_sql:sql_query_t(SQL) end, + SQLs) + end), + case Res of + {error, Error} -> + ?ERROR_MSG("Failed to update table ~s: ~p", + [TableName, Error]), + error(Error); + _ -> + ok + end; + ({drop_column, TableName, ColumnName}) -> + Res = + ejabberd_sql:sql_query_t( + fun(_DBType, _DBVersion) -> + SQL = [<<"ALTER TABLE ">>, + TableName, + <<" DROP COLUMN ">>, + ColumnName, + <<";">>], + ?INFO_MSG("Drop column ~s/~s:~n~s~n", + [TableName, + ColumnName, + SQL]), + ejabberd_sql:sql_query_t(SQL) + end), + case Res of + {error, Error} -> + ?ERROR_MSG("Failed to update table ~s: ~p", + [TableName, Error]), + error(Error); + _ -> + ok + end; + ({create_index, TableName, Columns1}) -> + Columns = + case ejabberd_sql:use_new_schema() of + true -> + Columns1; + false -> + lists:delete( + <<"server_host">>, Columns1) + end, + {value, Table} = + lists:keysearch( + TableName, #sql_table.name, Schema#sql_schema.tables), + {value, Index} = + lists:keysearch( + Columns, #sql_index.columns, Table#sql_table.indices), + case Index#sql_index.meta of + #{ignore := true} -> ok; + _ -> + Res = + ejabberd_sql:sql_query_t( + fun() -> + case Index#sql_index.meta of + #{primary_key := true} -> + SQL1 = format_add_primary_key( + SchemaInfo, Table, Index), + SQL = iolist_to_binary(SQL1), + ?INFO_MSG("Add primary key ~s/~p:~n~s~n", + [Table#sql_table.name, + Index#sql_index.columns, + SQL]), + ejabberd_sql:sql_query_t(SQL); + _ -> + SQL1 = format_create_index( + SchemaInfo, Table, Index), + SQL = iolist_to_binary(SQL1), + ?INFO_MSG("Create index ~s/~p:~n~s~n", + [Table#sql_table.name, + Index#sql_index.columns, + SQL]), + ejabberd_sql:sql_query_t(SQL) + end + end), + case Res of + {error, Error} -> + ?ERROR_MSG("Failed to update table ~s: ~p", + [TableName, Error]), + error(Error); + _ -> + ok + end + end; + ({update_primary_key, TableName, Columns1}) -> + Columns = + case ejabberd_sql:use_new_schema() of + true -> + Columns1; + false -> + lists:delete( + <<"server_host">>, Columns1) + end, + {value, Table} = + lists:keysearch( + TableName, #sql_table.name, Schema#sql_schema.tables), + {value, Index} = + lists:keysearch( + Columns, #sql_index.columns, Table#sql_table.indices), + Res = + case SchemaInfo#sql_schema_info.db_type of + sqlite -> + sqlite_table_copy_t(SchemaInfo, Table); + pgsql -> + TableName = Table#sql_table.name, + SQL1 = [<<"ALTER TABLE ">>, TableName, <<" DROP CONSTRAINT ", + TableName/binary, "_pkey, ", + "ADD PRIMARY KEY (">>, + lists:join( + <<", ">>, + Index#sql_index.columns), + <<");">>], + SQL = iolist_to_binary(SQL1), + ?INFO_MSG("Update primary key ~s/~p:~n~s~n", + [Table#sql_table.name, + Index#sql_index.columns, + SQL]), + ejabberd_sql:sql_query_t( + fun(_DBType, _DBVersion) -> + ejabberd_sql:sql_query_t(SQL) + end); + mysql -> + TableName = Table#sql_table.name, + SQL1 = [<<"ALTER TABLE ">>, TableName, <<" DROP PRIMARY KEY, " + "ADD PRIMARY KEY (">>, + lists:join( + <<", ">>, + lists:map( + fun(Col) -> + format_mysql_index_column(Table, Col) + end, Index#sql_index.columns)), + <<");">>], + SQL = iolist_to_binary(SQL1), + ?INFO_MSG("Update primary key ~s/~p:~n~s~n", + [Table#sql_table.name, + Index#sql_index.columns, + SQL]), + ejabberd_sql:sql_query_t( + fun(_DBType, _DBVersion) -> + ejabberd_sql:sql_query_t(SQL) + end) + end, + case Res of + {error, Error} -> + ?ERROR_MSG("Failed to update table ~s: ~p", + [TableName, Error]), + error(Error); + _ -> + ok + end; + ({drop_index, TableName, Columns1}) -> + Columns = + case ejabberd_sql:use_new_schema() of + true -> + Columns1; + false -> + lists:delete( + <<"server_host">>, Columns1) + end, + case find_index_name(Host, TableName, Columns) of + false -> + ?ERROR_MSG("Can't find an index to drop for ~s/~p", + [TableName, Columns]); + {ok, IndexName} -> + Res = + ejabberd_sql:sql_query_t( + fun(DBType, _DBVersion) -> + SQL = + case DBType of + mysql -> + [<<"DROP INDEX ">>, + IndexName, + <<" ON ">>, + TableName, + <<";">>]; + _ -> + [<<"DROP INDEX ">>, + IndexName, <<";">>] + end, + ?INFO_MSG("Drop index ~s/~p:~n~s~n", + [TableName, + Columns, + SQL]), + ejabberd_sql:sql_query_t(SQL) + end), + case Res of + {error, Error} -> + ?ERROR_MSG("Failed to update table ~s: ~p", + [TableName, Error]), + error(Error); + _ -> + ok + end + end + end, Schema#sql_schema.update), + store_version_t(Module, Schema#sql_schema.version) + end, + ejabberd_sql:sql_transaction(Host, F, ejabberd_option:update_sql_schema_timeout(), 1). print_schema(SDBType, SDBVersion, SNewSchema) -> {DBType, DBVersion} =