diff --git a/app/jobs/regular/bump_topic.rb b/app/jobs/regular/bump_topic.rb
index 5a60883e8f1..a8cd9fa35b4 100644
--- a/app/jobs/regular/bump_topic.rb
+++ b/app/jobs/regular/bump_topic.rb
@@ -1,22 +1,13 @@
 # frozen_string_literal: true
 
 module Jobs
-  class BumpTopic < ::Jobs::Base
-
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-
-      topic = topic_timer&.topic
-
-      if topic_timer.blank? || topic.blank? || topic_timer.execute_at > Time.zone.now
-        return
-      end
-
+  class BumpTopic < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       if Guardian.new(topic_timer.user).can_create_post_on_topic?(topic)
         topic.add_small_action(Discourse.system_user, "autobumped", nil, bump: true)
       end
+
       topic_timer.trash!(Discourse.system_user)
     end
-
   end
 end
diff --git a/app/jobs/regular/clear_slow_mode.rb b/app/jobs/regular/clear_slow_mode.rb
index 0e7f0cfcde9..361cce87461 100644
--- a/app/jobs/regular/clear_slow_mode.rb
+++ b/app/jobs/regular/clear_slow_mode.rb
@@ -1,22 +1,8 @@
 # frozen_string_literal: true
 
 module Jobs
-  class ClearSlowMode < ::Jobs::Base
-
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-
-      if topic_timer.nil? || topic_timer.execute_at > Time.zone.now
-        return
-      end
-
-      topic = topic_timer&.topic
-
-      if topic.nil?
-        topic_timer.destroy!
-        return
-      end
-
+  class ClearSlowMode < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       topic.update!(slow_mode_seconds: 0)
       topic_timer.trash!(Discourse.system_user)
     end
diff --git a/app/jobs/regular/close_topic.rb b/app/jobs/regular/close_topic.rb
index 96364d5841c..044ee64827b 100644
--- a/app/jobs/regular/close_topic.rb
+++ b/app/jobs/regular/close_topic.rb
@@ -1,16 +1,12 @@
 # frozen_string_literal: true
 
 module Jobs
-  class CloseTopic < ::Jobs::Base
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-      return if !topic_timer&.runnable?
-
-      topic = topic_timer.topic
+  class CloseTopic < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
+      silent = @args[:silent]
       user = topic_timer.user
-      silent = args[:silent]
 
-      if topic.blank? || topic.closed?
+      if topic.closed?
         topic_timer.destroy!
         return
       end
diff --git a/app/jobs/regular/delete_replies.rb b/app/jobs/regular/delete_replies.rb
index 0e5605a8366..895c1534093 100644
--- a/app/jobs/regular/delete_replies.rb
+++ b/app/jobs/regular/delete_replies.rb
@@ -1,17 +1,8 @@
 # frozen_string_literal: true
 
 module Jobs
-  class DeleteReplies < ::Jobs::Base
-
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-
-      topic = topic_timer&.topic
-
-      if topic_timer.blank? || topic.blank? || topic_timer.execute_at > Time.zone.now
-        return
-      end
-
+  class DeleteReplies < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       unless Guardian.new(topic_timer.user).is_staff?
         topic_timer.trash!(Discourse.system_user)
         return
diff --git a/app/jobs/regular/delete_topic.rb b/app/jobs/regular/delete_topic.rb
index 829a8331197..af109d8b101 100644
--- a/app/jobs/regular/delete_topic.rb
+++ b/app/jobs/regular/delete_topic.rb
@@ -1,23 +1,17 @@
 # frozen_string_literal: true
 
 module Jobs
-  class DeleteTopic < ::Jobs::Base
-
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-
-      topic = topic_timer&.topic
-
-      if topic_timer.blank? || topic.blank? || topic_timer.execute_at > Time.zone.now
-        return
-      end
-
+  class DeleteTopic < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       if Guardian.new(topic_timer.user).can_delete?(topic)
         first_post = topic.ordered_posts.first
-        PostDestroyer.new(topic_timer.user, first_post, context: I18n.t("topic_statuses.auto_deleted_by_timer")).destroy
+
+        PostDestroyer.new(
+          topic_timer.user, first_post, context: I18n.t("topic_statuses.auto_deleted_by_timer")
+        ).destroy
+
         topic_timer.trash!(Discourse.system_user)
       end
     end
-
   end
 end
diff --git a/app/jobs/regular/open_topic.rb b/app/jobs/regular/open_topic.rb
index 36246cb6531..85f2b96afc6 100644
--- a/app/jobs/regular/open_topic.rb
+++ b/app/jobs/regular/open_topic.rb
@@ -1,19 +1,10 @@
 # frozen_string_literal: true
 
 module Jobs
-  class OpenTopic < ::Jobs::Base
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-      return if !topic_timer&.runnable?
-
-      topic = topic_timer.topic
+  class OpenTopic < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       user = topic_timer.user
 
-      if topic.blank?
-        topic_timer.destroy!
-        return
-      end
-
       if !Guardian.new(user).can_open_topic?(topic) || topic.open?
         topic_timer.destroy!
         topic.reload
diff --git a/app/jobs/regular/publish_topic_to_category.rb b/app/jobs/regular/publish_topic_to_category.rb
index f3c41148388..b42079dbf64 100644
--- a/app/jobs/regular/publish_topic_to_category.rb
+++ b/app/jobs/regular/publish_topic_to_category.rb
@@ -1,14 +1,8 @@
 # frozen_string_literal: true
 
 module Jobs
-  class PublishTopicToCategory < ::Jobs::Base
-    def execute(args)
-      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
-      return if topic_timer.blank?
-
-      topic = topic_timer.topic
-      return if topic.blank?
-
+  class PublishTopicToCategory < ::Jobs::TopicTimerBase
+    def execute_timer_action(topic_timer, topic)
       return unless Guardian.new(topic_timer.user).can_see?(topic)
 
       TopicTimer.transaction do
diff --git a/app/jobs/regular/topic_timer_base.rb b/app/jobs/regular/topic_timer_base.rb
new file mode 100644
index 00000000000..35b582a97b4
--- /dev/null
+++ b/app/jobs/regular/topic_timer_base.rb
@@ -0,0 +1,20 @@
+# frozen_string_literal: true
+
+module Jobs
+  class TopicTimerBase < ::Jobs::Base
+    def execute(args)
+      @args = args
+
+      topic_timer = TopicTimer.find_by(id: args[:topic_timer_id])
+      return if !topic_timer&.runnable?
+
+      topic = topic_timer.topic
+      if topic.blank?
+        topic_timer.destroy!
+        return
+      end
+
+      execute_timer_action(topic_timer, topic)
+    end
+  end
+end
diff --git a/app/jobs/scheduled/periodical_updates.rb b/app/jobs/scheduled/periodical_updates.rb
index 2ae53c951d2..5c5da8bac6b 100644
--- a/app/jobs/scheduled/periodical_updates.rb
+++ b/app/jobs/scheduled/periodical_updates.rb
@@ -24,9 +24,6 @@ module Jobs
       args[:max_topic_length] = 500 unless self.class.should_update_long_topics?
       ScoreCalculator.new.calculate(args)
 
-      # Re-run stuff that we missed
-      TopicTimer.ensure_consistency!
-
       # Forces rebake of old posts where needed, as long as no system avatars need updating
       if !SiteSetting.automatically_download_gravatars || !UserAvatar.where("last_gravatar_download_attempt IS NULL").limit(1).first
         problems = Post.rebake_old(SiteSetting.rebake_old_posts_count, priority: :ultra_low)
diff --git a/app/jobs/scheduled/topic_timer_enqueuer.rb b/app/jobs/scheduled/topic_timer_enqueuer.rb
new file mode 100644
index 00000000000..7fa2889b1f3
--- /dev/null
+++ b/app/jobs/scheduled/topic_timer_enqueuer.rb
@@ -0,0 +1,25 @@
+# frozen_string_literal: true
+
+module Jobs
+
+  # Runs periodically to look through topic timers that are ready to execute,
+  # and enqueues their related jobs.
+  #
+  # Any leftovers will be caught in the next run, because execute_at will
+  # be < now, and topic timers that have run are deleted on completion or
+  # otherwise have their execute_at time modified.
+  class TopicTimerEnqueuer < ::Jobs::Scheduled
+    every 1.minute
+
+    def execute(_args = nil)
+      timers = TopicTimer.pending_timers
+
+      timers.find_each do |timer|
+
+        # the typed job may not enqueue if it has already
+        # been scheduled with enqueue_at
+        timer.enqueue_typed_job
+      end
+    end
+  end
+end
diff --git a/app/models/topic.rb b/app/models/topic.rb
index 46a522cbd61..a6cc367baf7 100644
--- a/app/models/topic.rb
+++ b/app/models/topic.rb
@@ -1357,6 +1357,8 @@ class Topic < ActiveRecord::Base
       end
 
       if self.persisted?
+        # See TopicTimer.after_save for additional context; the topic
+        # status may be changed by saving.
         topic_timer.save!
       else
         self.topic_timers << topic_timer
diff --git a/app/models/topic_timer.rb b/app/models/topic_timer.rb
index 053742e1723..613ea2a532e 100644
--- a/app/models/topic_timer.rb
+++ b/app/models/topic_timer.rb
@@ -18,6 +18,9 @@ class TopicTimer < ActiveRecord::Base
   validate :executed_at_in_future?
 
   scope :scheduled_bump_topics, -> { where(status_type: TopicTimer.types[:bump], deleted_at: nil).pluck(:topic_id) }
+  scope :pending_timers, ->(before_time = Time.now.utc) do
+    where("execute_at <= :before_time AND deleted_at IS NULL", before_time: before_time)
+  end
 
   before_save do
     self.created_at ||= Time.zone.now if execute_at
@@ -27,21 +30,68 @@ class TopicTimer < ActiveRecord::Base
        !attribute_in_database(:execute_at).nil?) ||
        will_save_change_to_user_id?
 
-      # private implementation detail have to use send
+      # TODO(martin - 2021-05-01) - Remove this backwards compatability for outstanding
+      # jobs once they have all been run and after Jobs::TopicTimerEnqueuer is in place
       self.send("cancel_auto_#{self.class.types[status_type]}_job")
     end
   end
 
+  # These actions are in place to make sure the topic is in the correct
+  # state at the point in time where the timer is saved. It does not
+  # guarantee that the topic will be in the correct state when the timer
+  # job is executed, but each timer job handles deleted topics etc. gracefully.
+  #
+  # This is also important for the Open Temporarily and Close Temporarily timers,
+  # which change the topic's status straight away and set a timer to do the
+  # opposite action in the future.
   after_save do
     if (saved_change_to_execute_at? || saved_change_to_user_id?)
-      now = Time.zone.now
-      time = execute_at < now ? now : execute_at
-
-      # private implementation detail have to use send
-      self.send("schedule_auto_#{self.class.types[status_type]}_job", time)
+      if status_type == TopicTimer.types[:silent_close] || status_type == TopicTimer.types[:close]
+        topic.update_status('closed', false, user) if topic.closed?
+      end
+      if status_type == TopicTimer.types[:open]
+        topic.update_status('closed', true, user) if topic.open?
+      end
     end
   end
 
+  def status_type_name
+    self.class.types[status_type]
+  end
+
+  def enqueue_typed_job(time: nil)
+    return if typed_job_scheduled?
+    self.send("schedule_auto_#{status_type_name}_job")
+  end
+
+  # TODO(martin - 2021-05-01) - Remove this backwards compatability for outstanding
+  # jobs once they have all been run and after Jobs::TopicTimerEnqueuer is in place
+  def typed_job_scheduled?
+    scheduled = Jobs.scheduled_for(
+      TopicTimer.type_job_map[status_type_name], topic_timer_id: id
+    ).any?
+
+    if [:close, :silent_close, :open].include?(status_type_name)
+      return scheduled || Jobs.scheduled_for(:toggle_topic_closed, topic_timer_id: id).any?
+    end
+
+    scheduled
+  end
+
+  def self.type_job_map
+    {
+      close: :close_topic,
+      open: :open_topic,
+      publish_to_category: :publish_topic_to_category,
+      delete: :delete_topic,
+      reminder: :topic_reminder,
+      bump: :bump_topic,
+      delete_replies: :delete_replies,
+      silent_close: :close_topic,
+      clear_slow_mode: :clear_slow_mode
+    }
+  end
+
   def self.types
     @types ||= Enum.new(
       close: 1,
@@ -64,18 +114,6 @@ class TopicTimer < ActiveRecord::Base
     @_private_types ||= types.only(:reminder, :clear_slow_mode)
   end
 
-  def self.ensure_consistency!
-    TopicTimer.where("topic_timers.execute_at < ?", Time.zone.now)
-      .find_each do |topic_timer|
-
-      # private implementation detail scoped to class
-      topic_timer.send(
-        "schedule_auto_#{self.types[topic_timer.status_type]}_job",
-        topic_timer.execute_at
-      )
-    end
-  end
-
   def public_type?
     !!self.class.public_types[self.status_type]
   end
@@ -100,6 +138,10 @@ class TopicTimer < ActiveRecord::Base
     ))
   end
 
+  def publishing_to_category?
+    self.status_type.to_i == TopicTimer.types[:publish_to_category]
+  end
+
   # TODO(martin - 2021-05-01) - Remove cancels for toggle_topic_closed once topic timer revamp completed.
   def cancel_auto_close_job
     Jobs.cancel_scheduled_job(:toggle_topic_closed, topic_timer_id: id)
@@ -119,73 +161,63 @@ class TopicTimer < ActiveRecord::Base
   end
 
   def cancel_auto_publish_to_category_job
-    Jobs.cancel_scheduled_job(:publish_topic_to_category, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:publish_to_category], topic_timer_id: id)
   end
 
   def cancel_auto_delete_job
-    Jobs.cancel_scheduled_job(:delete_topic, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:delete], topic_timer_id: id)
   end
 
   def cancel_auto_reminder_job
-    Jobs.cancel_scheduled_job(:topic_reminder, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:reminder], topic_timer_id: id)
   end
 
   def cancel_auto_bump_job
-    Jobs.cancel_scheduled_job(:bump_topic, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:bump], topic_timer_id: id)
   end
 
   def cancel_auto_delete_replies_job
-    Jobs.cancel_scheduled_job(:delete_replies, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:delete_replies], topic_timer_id: id)
   end
 
   def cancel_auto_clear_slow_mode_job
-    Jobs.cancel_scheduled_job(:clear_slow_mode, topic_timer_id: id)
+    Jobs.cancel_scheduled_job(TopicTimer.type_job_map[:clear_slow_mode], topic_timer_id: id)
   end
 
-  def schedule_auto_delete_replies_job(time)
-    Jobs.enqueue_at(time, :delete_replies, topic_timer_id: id)
+  def schedule_auto_delete_replies_job
+    Jobs.enqueue(TopicTimer.type_job_map[:delete_replies], topic_timer_id: id)
   end
 
-  def schedule_auto_bump_job(time)
-    Jobs.enqueue_at(time, :bump_topic, topic_timer_id: id)
+  def schedule_auto_bump_job
+    Jobs.enqueue(TopicTimer.type_job_map[:bump], topic_timer_id: id)
   end
 
-  def schedule_auto_open_job(time)
-    topic.update_status('closed', true, user) if topic && !topic.closed
-
-    Jobs.enqueue_at(time, :open_topic, topic_timer_id: id)
+  def schedule_auto_open_job
+    Jobs.enqueue(TopicTimer.type_job_map[:open], topic_timer_id: id)
   end
 
-  def schedule_auto_close_job(time)
-    topic.update_status('closed', false, user) if topic&.closed
-
-    Jobs.enqueue_at(time, :close_topic, topic_timer_id: id)
+  def schedule_auto_close_job
+    Jobs.enqueue(TopicTimer.type_job_map[:close], topic_timer_id: id)
   end
 
-  def schedule_auto_silent_close_job(time)
-    topic.update_status('closed', false, user) if topic&.closed
-
-    Jobs.enqueue_at(time, :close_topic, topic_timer_id: id, silent: true)
+  def schedule_auto_silent_close_job
+    Jobs.enqueue(TopicTimer.type_job_map[:close], topic_timer_id: id, silent: true)
   end
 
-  def schedule_auto_publish_to_category_job(time)
-    Jobs.enqueue_at(time, :publish_topic_to_category, topic_timer_id: id)
+  def schedule_auto_publish_to_category_job
+    Jobs.enqueue(TopicTimer.type_job_map[:publish_to_category], topic_timer_id: id)
   end
 
-  def publishing_to_category?
-    self.status_type.to_i == TopicTimer.types[:publish_to_category]
+  def schedule_auto_delete_job
+    Jobs.enqueue(TopicTimer.type_job_map[:delete], topic_timer_id: id)
   end
 
-  def schedule_auto_delete_job(time)
-    Jobs.enqueue_at(time, :delete_topic, topic_timer_id: id)
-  end
-
-  def schedule_auto_reminder_job(time)
+  def schedule_auto_reminder_job
     # noop, TODO(martin 2021-03-11): Remove this after timers migrated and outstanding jobs cancelled
   end
 
-  def schedule_auto_clear_slow_mode_job(time)
-    Jobs.enqueue_at(time, :clear_slow_mode, topic_timer_id: id)
+  def schedule_auto_clear_slow_mode_job
+    Jobs.enqueue(TopicTimer.type_job_map[:clear_slow_mode], topic_timer_id: id)
   end
 end
 
diff --git a/spec/integration/topic_auto_close_spec.rb b/spec/integration/topic_auto_close_spec.rb
index 538259a988b..170a3afc11c 100644
--- a/spec/integration/topic_auto_close_spec.rb
+++ b/spec/integration/topic_auto_close_spec.rb
@@ -42,10 +42,6 @@ describe Topic do
 
           expect(topic_status_update.topic).to eq(topic)
           expect(topic.public_topic_timer.execute_at).to eq_time(2.hours.from_now)
-
-          args = job_klass.jobs.last['args'].first
-
-          expect(args["topic_timer_id"]).to eq(topic.public_topic_timer.id)
         end
 
         context 'topic was created by staff user' do
@@ -60,10 +56,6 @@ describe Topic do
             expect(topic_status_update.topic).to eq(staff_topic)
             expect(topic_status_update.execute_at).to eq_time(2.hours.from_now)
             expect(topic_status_update.user).to eq(Discourse.system_user)
-
-            args = job_klass.jobs.last['args'].first
-
-            expect(args["topic_timer_id"]).to eq(topic_status_update.id)
           end
 
           context 'topic is closed manually' do
@@ -90,10 +82,6 @@ describe Topic do
             expect(topic_status_update.topic).to eq(regular_user_topic)
             expect(topic_status_update.execute_at).to eq_time(2.hours.from_now)
             expect(topic_status_update.user).to eq(Discourse.system_user)
-
-            args = job_klass.jobs.last['args'].first
-
-            expect(args["topic_timer_id"]).to eq(topic_status_update.id)
           end
         end
       end
diff --git a/spec/jobs/publish_topic_to_category_spec.rb b/spec/jobs/publish_topic_to_category_spec.rb
index 7ec605e6e64..585766f8b1a 100644
--- a/spec/jobs/publish_topic_to_category_spec.rb
+++ b/spec/jobs/publish_topic_to_category_spec.rb
@@ -12,7 +12,9 @@ RSpec.describe Jobs::PublishTopicToCategory do
     Fabricate(:topic_timer,
       status_type: TopicTimer.types[:publish_to_category],
       category_id: another_category.id,
-      topic: topic
+      topic: topic,
+      execute_at: 1.minute.ago,
+      created_at: 5.minutes.ago
     )
 
     Fabricate(:post, topic: topic)
diff --git a/spec/jobs/topic_timer_enqueuer_spec.rb b/spec/jobs/topic_timer_enqueuer_spec.rb
new file mode 100644
index 00000000000..b9f21ed12c8
--- /dev/null
+++ b/spec/jobs/topic_timer_enqueuer_spec.rb
@@ -0,0 +1,50 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe Jobs::TopicTimerEnqueuer do
+  subject { described_class.new }
+
+  fab!(:timer1) do
+    Fabricate(:topic_timer, execute_at: 1.minute.ago, created_at: 1.hour.ago, status_type: TopicTimer.types[:close])
+  end
+  fab!(:timer2) do
+    Fabricate(:topic_timer, execute_at: 1.minute.ago, created_at: 1.hour.ago, status_type: TopicTimer.types[:open])
+  end
+  fab!(:future_timer) do
+    Fabricate(:topic_timer, execute_at: 1.hours.from_now, created_at: 1.hour.ago, status_type: TopicTimer.types[:close])
+  end
+  fab!(:deleted_timer) do
+    Fabricate(:topic_timer, execute_at: 1.minute.ago, created_at: 1.hour.ago, status_type: TopicTimer.types[:close])
+  end
+
+  before do
+    deleted_timer.trash!
+  end
+
+  it "does not enqueue deleted timers" do
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: deleted_timer.id })
+    subject.execute
+    expect(deleted_timer.topic.reload.closed?).to eq(false)
+  end
+
+  it "does not enqueue future timers" do
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: future_timer.id })
+    subject.execute
+    expect(future_timer.topic.reload.closed?).to eq(false)
+  end
+
+  it "enqueues the related job" do
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: deleted_timer.id })
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: future_timer.id })
+    subject.execute
+    expect_job_enqueued(job: :close_topic, args: { topic_timer_id: timer1.id })
+    expect_job_enqueued(job: :open_topic, args: { topic_timer_id: timer2.id })
+  end
+
+  it "does not re-enqueue a job that has already been scheduled ahead of time in sidekiq (legacy topic timers)" do
+    expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: timer1.id })
+    Jobs.enqueue_at(1.hours.from_now, :close_topic, topic_timer_id: timer1.id)
+    subject.execute
+  end
+end
diff --git a/spec/models/post_action_spec.rb b/spec/models/post_action_spec.rb
index 3f7e80d50a3..26df835d891 100644
--- a/spec/models/post_action_spec.rb
+++ b/spec/models/post_action_spec.rb
@@ -764,9 +764,7 @@ describe PostAction do
 
         freeze_time timer.execute_at
 
-        expect_enqueued_with(job: :open_topic, args: { topic_timer_id: timer.id }, at: Time.zone.now + 1.hour) do
-          Jobs::OpenTopic.new.execute(topic_timer_id: timer.id)
-        end
+        Jobs::OpenTopic.new.execute(topic_timer_id: timer.id)
 
         expect(topic.reload.closed).to eq(true)
         expect(timer.reload.execute_at).to eq_time(1.hour.from_now)
diff --git a/spec/models/topic_spec.rb b/spec/models/topic_spec.rb
index 829510d93a5..dfcb7ecb57f 100644
--- a/spec/models/topic_spec.rb
+++ b/spec/models/topic_spec.rb
@@ -1876,7 +1876,7 @@ describe Topic do
 
         freeze_time 3.hours.from_now
 
-        TopicTimer.ensure_consistency!
+        Jobs::TopicTimerEnqueuer.new.execute
         expect(topic.reload.closed).to eq(true)
       end
     end
diff --git a/spec/models/topic_timer_spec.rb b/spec/models/topic_timer_spec.rb
index 46bd16bd8dc..a9ad8b39bf7 100644
--- a/spec/models/topic_timer_spec.rb
+++ b/spec/models/topic_timer_spec.rb
@@ -10,6 +10,20 @@ RSpec.describe TopicTimer, type: :model do
   before { freeze_time }
 
   context "validations" do
+    describe "pending_timers scope" do
+      it "does not return deleted timers" do
+        topic_timer.trash!
+        expect(TopicTimer.pending_timers.pluck(:id)).not_to include(topic_timer.id)
+      end
+
+      it "does not return timers in the future of the provided before time" do
+        topic_timer.update!(execute_at: 3.days.from_now)
+        expect(TopicTimer.pending_timers.pluck(:id)).not_to include(topic_timer.id)
+        expect(TopicTimer.pending_timers(2.days.from_now).pluck(:id)).not_to include(topic_timer.id)
+        topic_timer.update!(execute_at: 1.minute.ago, created_at: 10.minutes.ago)
+        expect(TopicTimer.pending_timers.pluck(:id)).to include(topic_timer.id)
+      end
+    end
     describe '#status_type' do
       it 'should ensure that only one active public topic status update exists' do
         topic_timer.update!(topic: topic)
@@ -20,6 +34,44 @@ RSpec.describe TopicTimer, type: :model do
       end
     end
 
+    describe "#typed_job_scheduled?" do
+      let(:scheduled_jobs) { Sidekiq::ScheduledSet.new }
+
+      after do
+        scheduled_jobs.clear
+      end
+
+      it "returns true if the job is scheduled" do
+        Sidekiq::Testing.disable! do
+          scheduled_jobs.clear
+          Jobs.enqueue_at(3.hours.from_now, :close_topic, topic_timer_id: topic_timer.id)
+          expect(topic_timer.typed_job_scheduled?).to eq(true)
+        end
+      end
+
+      it "returns false if the job is not already scheduled" do
+        Sidekiq::Testing.disable! do
+          scheduled_jobs.clear
+          expect(topic_timer.typed_job_scheduled?).to eq(false)
+        end
+      end
+
+      it "returns true if the toggle_topic_closed job is scheduled for close,open,silent_close types" do
+        Sidekiq::Testing.disable! do
+          scheduled_jobs.clear
+          topic_timer1 = Fabricate(:topic_timer, status_type: TopicTimer.types[:close])
+          Jobs.enqueue_at(3.hours.from_now, :toggle_topic_closed, topic_timer_id: topic_timer1.id)
+          topic_timer2 = Fabricate(:topic_timer, status_type: TopicTimer.types[:open])
+          Jobs.enqueue_at(3.hours.from_now, :toggle_topic_closed, topic_timer_id: topic_timer2.id)
+          topic_timer3 = Fabricate(:topic_timer, status_type: TopicTimer.types[:silent_close])
+          Jobs.enqueue_at(3.hours.from_now, :toggle_topic_closed, topic_timer_id: topic_timer3.id)
+          expect(topic_timer1.typed_job_scheduled?).to eq(true)
+          expect(topic_timer2.typed_job_scheduled?).to eq(true)
+          expect(topic_timer3.typed_job_scheduled?).to eq(true)
+        end
+      end
+    end
+
     describe '#execute_at' do
       describe 'when #execute_at is greater than #created_at' do
         it 'should be valid' do
@@ -95,20 +147,7 @@ RSpec.describe TopicTimer, type: :model do
           :close_topic, topic_timer_id: topic_timer.id
         )
 
-        expect_enqueued_with(job: :close_topic, args: { topic_timer_id: topic_timer.id }, at: 3.days.from_now) do
-          topic_timer.update!(execute_at: 3.days.from_now, created_at: Time.zone.now)
-        end
-      end
-
-      describe 'when execute_at is smaller than the current time' do
-        it 'should enqueue the job immediately' do
-          expect_enqueued_with(job: :close_topic, args: { topic_timer_id: topic_timer.id }, at: Time.zone.now) do
-            topic_timer.update!(
-              execute_at: Time.zone.now - 1.hour,
-              created_at: Time.zone.now - 2.hour
-            )
-          end
-        end
+        topic_timer.update!(execute_at: 3.days.from_now, created_at: Time.zone.now)
       end
     end
 
@@ -121,9 +160,7 @@ RSpec.describe TopicTimer, type: :model do
           :close_topic, topic_timer_id: topic_timer.id
         )
 
-        expect_enqueued_with(job: :close_topic, args: { topic_timer_id: topic_timer.id }, at: topic_timer.execute_at) do
-          topic_timer.update!(user: admin)
-        end
+        topic_timer.update!(user: admin)
       end
     end
 
@@ -141,18 +178,9 @@ RSpec.describe TopicTimer, type: :model do
       end
 
       it 'should close the topic' do
-        topic_timer
+        topic_timer.send(:schedule_auto_open_job)
         expect(topic.reload.closed).to eq(true)
       end
-
-      describe 'when topic has been deleted' do
-        it 'should not queue the job' do
-          topic.trash!
-          topic_timer
-
-          expect(Jobs::ToggleTopicClosed.jobs).to eq([])
-        end
-      end
     end
 
     describe 'when a close topic status update is created for a closed topic' do
@@ -169,18 +197,9 @@ RSpec.describe TopicTimer, type: :model do
       end
 
       it 'should open the topic' do
-        topic_timer
+        topic_timer.send(:schedule_auto_close_job)
         expect(topic.reload.closed).to eq(false)
       end
-
-      describe 'when topic has been deleted' do
-        it 'should not queue the job' do
-          topic.trash!
-          topic_timer
-
-          expect(Jobs::ToggleTopicClosed.jobs).to eq([])
-        end
-      end
     end
 
     describe '#public_type' do
@@ -205,62 +224,6 @@ RSpec.describe TopicTimer, type: :model do
     end
   end
 
-  describe '.ensure_consistency!' do
-    it 'should enqueue jobs that have been missed' do
-      close_topic_timer = Fabricate(:topic_timer,
-        execute_at: Time.zone.now - 1.hour,
-        created_at: Time.zone.now - 2.hour
-      )
-
-      open_topic_timer = Fabricate(:topic_timer,
-        status_type: described_class.types[:open],
-        execute_at: Time.zone.now - 1.hour,
-        created_at: Time.zone.now - 2.hour,
-        topic: Fabricate(:topic, closed: true)
-      )
-
-      Fabricate(:topic_timer, execute_at: Time.zone.now + 1.hour)
-
-      trashed_close_topic_timer = Fabricate(:topic_timer,
-        execute_at: Time.zone.now - 1.hour,
-        created_at: Time.zone.now - 2.hour
-      )
-
-      trashed_close_topic_timer.topic.trash!
-
-      trashed_open_topic_timer = Fabricate(:topic_timer,
-        execute_at: Time.zone.now - 1.hour,
-        created_at: Time.zone.now - 2.hour,
-        status_type: described_class.types[:open]
-      )
-
-      trashed_open_topic_timer.topic.trash!
-
-      # creating topic timers already enqueues jobs
-      # let's delete them to test ensure_consistency!
-      Sidekiq::Worker.clear_all
-
-      expect { described_class.ensure_consistency! }
-        .to change { Jobs::CloseTopic.jobs.count }.by(2).and change { Jobs::OpenTopic.jobs.count }.by(2)
-
-      expect(job_enqueued?(job: :close_topic, args: {
-        topic_timer_id: close_topic_timer.id
-      })).to eq(true)
-
-      expect(job_enqueued?(job: :open_topic, args: {
-        topic_timer_id: open_topic_timer.id
-      })).to eq(true)
-
-      expect(job_enqueued?(job: :close_topic, args: {
-        topic_timer_id: trashed_close_topic_timer.id
-      })).to eq(true)
-
-      expect(job_enqueued?(job: :open_topic, args: {
-        topic_timer_id: trashed_open_topic_timer.id
-      })).to eq(true)
-    end
-  end
-
   describe "runnable?" do
     it "returns false if execute_at > now" do
       topic_timer = Fabricate.build(:topic_timer,
diff --git a/spec/support/sidekiq_helpers.rb b/spec/support/sidekiq_helpers.rb
index 815badebe73..12dce59e856 100644
--- a/spec/support/sidekiq_helpers.rb
+++ b/spec/support/sidekiq_helpers.rb
@@ -47,7 +47,7 @@ module SidekiqHelpers
   # end
   def expect_not_enqueued_with(job:, args: {}, at: nil)
     expect_enqueued_with(job: job, args: args, at: at, expectation: false) do
-      yield
+      yield if block_given?
     end
   end
 
@@ -61,6 +61,20 @@ module SidekiqHelpers
     match_jobs(jobs: klass.jobs, args: args, at: at)
   end
 
+  # Same as job_enqueued? except it checks the expectation is true.
+  # Use this if you need to check if more than one job is enqueued from
+  # a single command, unlike expect_enqueued_with which needs a block
+  # to run code for the expectation to work. E.g.
+  #
+  # expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: deleted_timer.id })
+  # expect_not_enqueued_with(job: :close_topic, args: { topic_timer_id: future_timer.id })
+  # subject.execute
+  # expect_job_enqueued(job: :close_topic, args: { topic_timer_id: timer1.id })
+  # expect_job_enqueued(job: :open_topic, args: { topic_timer_id: timer2.id })
+  def expect_job_enqueued(job:, args: {}, at: nil)
+    expect(job_enqueued?(job: job, args: args, at: at)).to eq(true)
+  end
+
   private
 
   def match_jobs(jobs:, args:, at:)