sidekiq execute_in (задержка) изнутри рабочего игнорирует задержку - PullRequest
Новая
       93

sidekiq execute_in (задержка) изнутри рабочего игнорирует задержку

0 голосов
/ 16 декабря 2018

Пользователи в моем приложении создают Transactions, и мне нужно для этих транзакций (и связанных заданий, которые создаются для изменения состояния транзакции на ignored, когда пользователи не отвечают в течение определенного времени), чтобы отменить себя, если толькопользователь выполняет действие pay.

Метод, который я использую в одном примере, выполняет следующие вызовы с использованием perform_async после изменения состояния на approved, а затем отменяется, если на него не ответили вовремя: 

Class Transaction < ApplicationRecord
 #when approved
 def create_worker
  MyWorker.perform_async(self.id)
 end

 #if user responds in time, cancel the jobs and update the record to `paid` etc
 def cancel_worker
  jid = MyWorker.perform_async(self.id)
  MyWorker.cancel! jid
 end
end

Как предложено здесь и здесь , я добавляю дополнительную функциональность о том, когда отменять внутри рабочего.Это выглядит примерно так: 

class MyWorker
 include Sidekiq::Worker

 def perform(transaction_id)
  return if paid?
  transaction = Transaction.find transaction_id
  self.class.perform_in(1.minutes, transaction.ignore!)
 end

 def paid?
  Sidekiq.redis { |c| c.exists("paid-#{jid}") }
 end

 def self.cancel! jid
  Sidekiq.redis { |c| c.setex("paid-#{jid}", 86400, 1) }
 end
end

Этот код приводит к следующему выводу терминала: 

2018-12-16T01:40:50.645Z 30530 TID-oxm547nes MyWorker JID-6c97e448fe30998235dee95d INFO: start
Changing transaction 4 approved to ignored (event: ignore!)
2018-12-16T01:40:50.884Z 30530 TID-oxm547nes MyWorker JID-6c97e448fe30998235dee95d INFO: done: 0.239 sec
2018-12-16T01:41:56.122Z 30530 TID-oxm547oag MyWorker JID-b46bb3b002e00f480a04be16 INFO: start
2018-12-16T01:41:56.125Z 30530 TID-oxm547oag MyWorker JID-b46bb3b002e00f480a04be16 INFO: fail: 0.003 sec
2018-12-16T01:41:56.126Z 30530 TID-oxm547oag WARN: {"context":"Job raised exception","job":{"class":"MyWorker","args":[true],"retry":true,"queue":"default","jid":"b46bb3b002e00f480a04be16","created_at":1544924450.884224,"enqueued_at":1544924516.107598,"error_message":"Couldn't find Transaction with 'id'=true","error_class":"ActiveRecord::RecordNotFound","failed_at":1544924516.125679,"retry_count":0},"jobstr":"{\"class\":\"MyWorker\",\"args\":[true],\"retry\":true,\"queue\":\"default\",\"jid\":\"b46bb3b002e00f480a04be16\",\"created_at\":1544924450.884224,\"enqueued_at\":1544924516.107598}"}

Таким образом, создается два задания - одно с jid 6c97e448fe30998235dee95d и которое сразуустанавливает транзакцию на ignored, а затем одну с jid b46bb3b002e00f480a04be16, который дует сразу за ранним возвратом в функции perform работника (потому что он не использует тот же jid, что и в первой работе).

Одна причина, по которой я могу предположить, почему это не работает так, как я намереваюсь, состоит в том, что вызов MyWorker.cancel! не может получить jid работника. Я хочу отменить, не создав сначала миграцию БД.держать сказал Джид.

Является ли создание миграции базы данных, содержащей jid для работника, предпочтительным способом проверки доступности jid между действиями?И как id=true туда попал?Как сказано выше, ошибка: Couldn't find Transaction with 'id'=true"

1 Ответ

0 голосов
/ 16 декабря 2018

Хорошо, поехали по частям.

  1. Этот код: 

    self.class.perform_in(1.minute, transaction.ignore!)

    передает любое возвращаемое значение метода ignore! (в данном случае true) в качестве аргумента для задания, что вызывает исключение.

    Вы должны убедиться, что передали правильные аргументы: 

    self.class.perform_in(1.minute, transaction.tap(&:ignore!).id)

  2. Каждый раз, когда вы вызываете MyWorker.perform_async (или любой другой метод класса выполнения), вы создаетеновая работа, поэтому не удивительно, что вы не получаете тот же jid.

    В соответствии с предложением вы должны сохранить начальные значения jid в таблице транзакций, а затем при оплате получить их для отмены.В противном случае идентификатор работы теряется.Альтернатива состоит в том, чтобы фактически использовать тот же redis для хранения оплаченного флага, но вместо этого вводить ключ транзакцией.c.exists("paid-#{transaction.id}")

  3. Ваш код не ждет 1 минуту, чтобы проигнорировать транзакцию, он просто сразу игнорирует транзакцию и настраивается на повторное выполнение через 1 минуту.

    Возможно, вы хотите вызвать 

    jid = MyWorker.perform_in(1.minute, transaction.id)

    напрямую из метода create_worker.

ОБНОВЛЕНИЕ

Если, как я понимаю, вы используете какой-то постоянный конечный автомат, даже проще просто "игнорировать, если не завершено", и забытьоб отмене задания 

class Transaction
  # I'm inventing a DSL here
  include SomeStateMachine

  state :accepted do
    event :ignore, to: :ignored
    event :confirm, to: :confirmed
  end
  state :ignored
  state :confirmed

  def create_worker
    # no need to track it
    MyWorker.perform_in(1.minute, id)
  end
end

class MyWorker
  include Sidekiq::Worker

  def perform(id)
    transaction = Transaction.find(id)
    transaction.ignore! if transaction.can_ignore?
  end
end

Вы можете разрешить выполнение задания, и оно с радостью пропустит любую незаметную транзакцию.

...