| | | 1 | | using MySqlConnector; |
| | | 2 | | |
| | | 3 | | namespace Services.HangfireService; |
| | | 4 | | |
| | | 5 | | /// <summary> |
| | | 6 | | /// Перенос зависших задач из очереди default (воркеры её не слушают). |
| | | 7 | | /// Выполняется в БД Hangfire (ST2_Hangfire), не в ST2. |
| | | 8 | | /// </summary> |
| | | 9 | | public static class HangfireQueueRepair |
| | | 10 | | { |
| | | 11 | | /// <summary> |
| | | 12 | | /// Блокировка таблиц внутреннего хранилища Hangfire кратковременна: штатный планировщик |
| | | 13 | | /// повторяет свои операции, а миграция очереди повторится при следующем старте. |
| | | 14 | | /// </summary> |
| | | 15 | | public static bool IsLockWaitTimeout(Exception exception) => |
| | 2 | 16 | | exception.ToString().Contains("Lock wait timeout exceeded", StringComparison.OrdinalIgnoreCase); |
| | | 17 | | |
| | | 18 | | public static async Task RepairDefaultQueuesAsync( |
| | | 19 | | HangfireStorageConnection hangfireStorage, |
| | | 20 | | CancellationToken cancellationToken = default) |
| | | 21 | | { |
| | 0 | 22 | | await using var connection = new MySqlConnection(hangfireStorage.ConnectionString); |
| | 0 | 23 | | await connection.OpenAsync(cancellationToken); |
| | | 24 | | |
| | 0 | 25 | | if (!await HangfireTablesExistAsync(connection, cancellationToken)) |
| | | 26 | | { |
| | | 27 | | return; |
| | | 28 | | } |
| | | 29 | | |
| | 0 | 30 | | await ExecuteNonQueryAsync( |
| | 0 | 31 | | connection, |
| | 0 | 32 | | """ |
| | 0 | 33 | | UPDATE HangfireHash |
| | 0 | 34 | | SET Value = @queue |
| | 0 | 35 | | WHERE Field = 'Queue' |
| | 0 | 36 | | AND Value = 'default' |
| | 0 | 37 | | """, |
| | 0 | 38 | | new MySqlParameter("@queue", HangfireQueueNames.Main), |
| | 0 | 39 | | cancellationToken); |
| | | 40 | | |
| | 0 | 41 | | await ExecuteNonQueryAsync( |
| | 0 | 42 | | connection, |
| | 0 | 43 | | """ |
| | 0 | 44 | | UPDATE HangfireJobQueue q |
| | 0 | 45 | | INNER JOIN HangfireJob j ON j.Id = q.JobId |
| | 0 | 46 | | SET q.Queue = @queue |
| | 0 | 47 | | WHERE q.Queue = 'default' |
| | 0 | 48 | | AND ( |
| | 0 | 49 | | j.InvocationData LIKE '%IOzonWebhookService%' |
| | 0 | 50 | | OR j.InvocationData LIKE '%OzonWebhookService%' |
| | 0 | 51 | | ) |
| | 0 | 52 | | """, |
| | 0 | 53 | | new MySqlParameter("@queue", HangfireQueueNames.OzonWebhook), |
| | 0 | 54 | | cancellationToken); |
| | | 55 | | |
| | 0 | 56 | | await ExecuteNonQueryAsync( |
| | 0 | 57 | | connection, |
| | 0 | 58 | | """ |
| | 0 | 59 | | UPDATE HangfireJobQueue q |
| | 0 | 60 | | INNER JOIN HangfireJob j ON j.Id = q.JobId |
| | 0 | 61 | | SET q.Queue = @queue |
| | 0 | 62 | | WHERE q.Queue = 'default' |
| | 0 | 63 | | """, |
| | 0 | 64 | | new MySqlParameter("@queue", HangfireQueueNames.Main), |
| | 0 | 65 | | cancellationToken); |
| | 0 | 66 | | } |
| | | 67 | | |
| | | 68 | | private static async Task<bool> HangfireTablesExistAsync( |
| | | 69 | | MySqlConnection connection, |
| | | 70 | | CancellationToken cancellationToken) |
| | | 71 | | { |
| | 0 | 72 | | await using var cmd = connection.CreateCommand(); |
| | 0 | 73 | | cmd.CommandText = """ |
| | 0 | 74 | | SELECT COUNT(*) |
| | 0 | 75 | | FROM information_schema.tables |
| | 0 | 76 | | WHERE table_schema = DATABASE() |
| | 0 | 77 | | AND table_name = 'HangfireHash' |
| | 0 | 78 | | """; |
| | 0 | 79 | | var count = Convert.ToInt32(await cmd.ExecuteScalarAsync(cancellationToken)); |
| | 0 | 80 | | return count > 0; |
| | 0 | 81 | | } |
| | | 82 | | |
| | | 83 | | private static async Task ExecuteNonQueryAsync( |
| | | 84 | | MySqlConnection connection, |
| | | 85 | | string sql, |
| | | 86 | | MySqlParameter parameter, |
| | | 87 | | CancellationToken cancellationToken) |
| | | 88 | | { |
| | 0 | 89 | | await using var cmd = connection.CreateCommand(); |
| | 0 | 90 | | cmd.CommandText = sql; |
| | 0 | 91 | | cmd.Parameters.Add(parameter); |
| | 0 | 92 | | await cmd.ExecuteNonQueryAsync(cancellationToken); |
| | 0 | 93 | | } |
| | | 94 | | } |