diff --git a/tests/bugs/core_2668_test.py b/tests/bugs/core_2668_test.py index e98a8921..a812742e 100644 --- a/tests/bugs/core_2668_test.py +++ b/tests/bugs/core_2668_test.py @@ -5,92 +5,214 @@ ID: issue-3072 ISSUE: 3072 TITLE: Write note into log when automatic sweep is started DESCRIPTION: + Test changes sweep interval to some low value (see SWEEP_GAP) and runs TX_COUNT transactions which + lead difference between OST and OIT to exceed given sweep interval. These transactions are performed + by ISQL which is launched as child process. SQL script uses table with record that is locked at the + beginning of script and execute block with loop of TX_COUNT statements which insert new records. + After this loop finish, we make ISQL to hang by forcing it to update first record (see LOCKED_ROW). + Then we change DB state to full shutdown and wait until ISQL will be terminated. + At this point database has sweep gap that is enough to run auto sweep at first connection to DB. + Finally, we bring DB online and start trace with log_sweep = true and log_transactions = true. + Doing connection and wait about 2..3 seconds cause auto sweep to be started and completed. + This must be reflected in the trace. + + If ServerMode = 'Super' and ParallelWorkers >= 2 and MaxParallelWorkers >= ParallelWorkers + then trace log will contain folloing five lines related to worker(s) activity: + (...) START_TRANSACTION + (ATT_..., , NONE, ) --------------------- [ 1 ] + (TRA_..., READ_COMMITTED | REC_VERSION | WAIT | READ_ONLY) + (...) COMMIT_TRANSACTION + (ATT_..., , NONE, ) --------------------- [ 2 ] + + This is the only difference that can be observed for snapshots before and after fix + (i.e. BEFORE fix trace had no such lines but all other data about sweep *did* present). + Test checks that trace log contains TWO lines with '', see above [ 1 ] and [ 2 ]. + JIRA: CORE-2668 FBTEST: bugs.core_2668 +NOTES: + [07.11.2024] pzotov + Confirmed absense of lines marked as ',\s+NONE,\s+\)', re.IGNORECASE) +################ -test_script = """ -recreate table test(s varchar(36) unique); -insert into test(s) values('LOCKED_FOR_PAUSE'); -commit; - -set transaction read committed WAIT; - -update test set s = s where s = 'LOCKED_FOR_PAUSE'; - -set term ^; -execute block as - declare n int = 150; - declare v_role varchar(31); -begin - while (n > 0) do - in autonomous transaction do - insert into test(s) values( rpad('', 36, uuid_to_char(gen_uuid()) ) ) - returning :n-1 into n; - - v_role = left(replace( uuid_to_char(gen_uuid()), '-', ''), 31); - - begin - execute statement ('update test set s = s where s = ?') ('LOCKED_FOR_PAUSE') - on external - 'localhost:' || rdb$get_context('SYSTEM', 'DB_NAME') - as user 'SYSDBA' password 'masterkey' role v_role - with autonomous transaction; - when any do - begin - end - end - -end -^ -set term ;^ -set heading off; -select '-- shutdown me now --' from rdb$database; -""" - -tmp_script = temp_file('work_script.sql') +tmp_sql = temp_file('tmp_2668.sql') +tmp_log = temp_file('tmp_2668.log') @pytest.mark.es_eds -@pytest.mark.version('>=3') -def test_1(act: Action, tmp_script: Path): - tmp_script.write_text(test_script) +@pytest.mark.version('>=5.0.0') +def test_1(act: Action, tmp_sql: Path, tmp_log: Path, capsys): + + if act.vars['server-arch'] != 'SuperServer': + pytest.skip("Applies only to SuperServer") + + with act.db.connect() as con: + cur = con.cursor() + sql = """ + select + cast(max(iif(g.rdb$config_name = 'ParallelWorkers', g.rdb$config_value, null)) as int) as cfg_par_workers + ,cast(max(iif(g.rdb$config_name = 'MaxParallelWorkers', g.rdb$config_value, null)) as int) as cfg_max_par_workers + from rdb$database + left join rdb$config g on g.rdb$config_name in ('ParallelWorkers', 'MaxParallelWorkers') + """ + cur.execute(sql) + cfg_par_workers, cfg_max_par_workers = cur.fetchone() + + assert cfg_par_workers >=2 and cfg_max_par_workers >= cfg_par_workers, "Server must be configured for parallel work. Check values of ParallelWorkers and MaxParallelWorkers" + + test_script = f""" + set echo on; + set bail on; + connect '{act.db.dsn}' user {act.db.user} password '{act.db.password}'; + + recreate table test(id int primary key); + insert into test(id) values({LOCKED_ROW}); + commit; + + set transaction read committed WAIT; + + update test set id = id where id = {LOCKED_ROW}; + set term ^; + execute block as + declare n int = {TX_COUNT}; + declare v_role varchar(31); + begin + while (n > 0) do + in autonomous transaction do + insert into test(id) values(:n) + returning :n-1 into n; + + v_role = left(replace( uuid_to_char(gen_uuid()), '-', ''), 31); + + begin + execute statement ('update test /* ' || ascii_char(65) || ' */ set id = id where id = ?') ({LOCKED_ROW}) + on external + 'localhost:' || rdb$get_context('SYSTEM', 'DB_NAME') + as user '{act.db.user}' password '{act.db.password}' role v_role + with autonomous transaction; + when any do + begin + end + end + + end + ^ + set term ;^ + set heading off; + select '-- shutdown me now --' from rdb$database; + """ + + tmp_sql.write_text(test_script) with act.connect_server() as srv: - srv.database.set_sweep_interval(database=act.db.db_path, interval=100) + ############################## + ### reduce SWEEEP interval ### + ############################## + srv.database.set_sweep_interval(database = act.db.db_path, interval = SWEEP_GAP) srv.database.set_write_mode(database=act.db.db_path, mode=DbWriteMode.ASYNC) - p_work_sql = subprocess.Popen([act.vars['isql'], '-i', str(tmp_script), - '-user', act.db.user, - '-password', act.db.password, act.db.dsn], - stderr = subprocess.STDOUT) - time.sleep(3) - try: - srv.database.shutdown(database=act.db.db_path, mode=ShutdownMode.FULL, - method=ShutdownMethod.FORCED, timeout=0) - finally: - p_work_sql.terminate() + + with open(tmp_log,'w') as f_log: + p_work_sql = subprocess.Popen([act.vars['isql'], '-q', '-i', str(tmp_sql)], stdout = f_log, stderr = subprocess.STDOUT) + + chk_mon_sql = """ + select 1 + from mon$attachments a + join mon$statements s + using (mon$attachment_id) + where + a.mon$attachment_id <> current_connection + and cast(s.mon$sql_text as varchar(8192)) containing '/* A */' + """ + + found_in_mon_tables = False + with act.db.connect() as con_watcher: + + custom_tpb = tpb(isolation = Isolation.SNAPSHOT, lock_timeout = -1) + tx_watcher = con_watcher.transaction_manager(custom_tpb) + cur_watcher = tx_watcher.cursor() + + ps = cur_watcher.prepare(chk_mon_sql) + + i = 0 + da = dt.now() + while True: + cur_watcher.execute(ps) + mon_result = -1 + for r in cur_watcher: + mon_result = r[0] + + tx_watcher.commit() + db = dt.now() + diff_ms = (db-da).seconds*1000 + (db-da).microseconds//1000 + if mon_result == 1: + found_in_mon_tables = True + break + elif diff_ms > MAX_WAIT_FOR_ISQL_PID_APPEARS_MS: + break + + time.sleep(0.1) + + ps.free() + + assert found_in_mon_tables, f'Could not find attachment in mon$ tables for {MAX_WAIT_FOR_ISQL_PID_APPEARS_MS} ms.' + + try: + ############################################## + ### f u l l s h u t d o w n D B ### + ############################################## + srv.database.shutdown(database=act.db.db_path, mode=ShutdownMode.FULL, + method=ShutdownMethod.FORCED, timeout=0) + finally: + p_work_sql.terminate() + # < with open(tmp_log,'w') as f_log + srv.database.bring_online(database=act.db.db_path) - srv.info.get_log() - fblog_before = srv.readlines() + + trace_options = \ + [ + 'time_threshold = 0' + ,'log_initfini = false' + ,'log_connections = true' + ,'log_transactions = true' + ,'log_errors = true' + ,'log_sweep = true' + ] + + with act.trace(db_events = trace_options, encoding='utf8', encoding_errors='utf8'): with act.db.connect() as con_for_sweep_start: - con_for_sweep_start.begin() time.sleep(2) - srv.info.get_log() - fblog_after = srv.readlines() - pattern = re.compile('Sweep\\s+.*SWEEPER', re.IGNORECASE) - success = False - for line in unified_diff(fblog_before, fblog_after): - if line.startswith('+') and pattern.search(line): - success = True - assert success + + + for line in act.trace_log: + if WATCH_FOR_PTN.search(line): + print(WATCH_FOR_PTN.search(line).group()) + + act.expected_stdout = """ + (ATT_N, , NONE, ) + (ATT_N, , NONE, ) + """ + act.stdout = capsys.readouterr().out + assert act.clean_stdout == act.clean_expected_stdout