Merge tag 'v4.1.3' into lets-bump-hometown-to-mastodon-4.2

Conflict resolution: only the streaming API is interesting. The filtering for local statuses is done before the other filters even kick in to be sure.
2 years ago · cc9dc8d902
20 changed files with 448 additions and 266 deletions
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@ -3,6 +3,54 @@ Changelog

 All notable changes to this project will be documented in this file.

+## [4.1.3] - 2023-07-06
+
+### Added
+
+- Add fallback redirection when getting a webfinger query `LOCAL_DOMAIN@LOCAL_DOMAIN` ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/23600))
+
+### Changed
+
+- Change OpenGraph-based embeds to allow fullscreen ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25058))
+- Change AccessTokensVacuum to also delete expired tokens ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/24868))
+- Change profile updates to be sent to recently-mentioned servers ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/24852))
+- Change automatic post deletion thresholds and load detection ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/24614))
+- Change `/api/v1/statuses/:id/history` to always return at least one item ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25510))
+- Change auto-linking to allow carets in URL query params ([renchap](https://github.com/mastodon/mastodon/pull/25216))
+
+### Removed
+
+- Remove invalid `X-Frame-Options: ALLOWALL` ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25070))
+
+### Fixed
+
+- Fix wrong view being displayed when a webhook fails validation ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25464))
+- Fix soft-deleted post cleanup scheduler overwhelming the streaming server ([ThisIsMissEm](https://github.com/mastodon/mastodon/pull/25519))
+- Fix incorrect pagination headers in `/api/v2/admin/accounts` ([danielmbrasil](https://github.com/mastodon/mastodon/pull/25477))
+- Fix multiple inefficiencies in automatic post cleanup worker ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/24607), [ClearlyClaire](https://github.com/mastodon/mastodon/pull/24785), [ClearlyClaire](https://github.com/mastodon/mastodon/pull/24840))
+- Fix performance of streaming by parsing message JSON once ([ThisIsMissEm](https://github.com/mastodon/mastodon/pull/25278), [ThisIsMissEm](https://github.com/mastodon/mastodon/pull/25361))
+- Fix CSP headers when `S3_ALIAS_HOST` includes a path component ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25273))
+- Fix `tootctl accounts approve --number N` not aproving N earliest registrations ([danielmbrasil](https://github.com/mastodon/mastodon/pull/24605))
+- Fix reports not being closed when performing batch suspensions ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/24988))
+- Fix being able to vote on your own polls ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25015))
+- Fix race condition when reblogging a status ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25016))
+- Fix “Authorized applications” inefficiently and incorrectly getting last use date ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25060))
+- Fix “Authorized applications” crashing when listing apps with certain admin API scopes ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25713))
+- Fix multiple N+1s in ConversationsController ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25134), [ClearlyClaire](https://github.com/mastodon/mastodon/pull/25399), [ClearlyClaire](https://github.com/mastodon/mastodon/pull/25499))
+- Fix user archive takeouts when using OpenStack Swift ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/24431))
+- Fix searching for remote content by URL not working under certain conditions ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25637))
+- Fix inefficiencies in indexing content for search ([VyrCossont](https://github.com/mastodon/mastodon/pull/24285), [VyrCossont](https://github.com/mastodon/mastodon/pull/24342))
+
+### Security
+
+- Add finer permission requirements for managing webhooks ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25463))
+- Update dependencies
+- Add hardening headers for user-uploaded files ([ClearlyClaire](https://github.com/mastodon/mastodon/pull/25756))
+- Fix verified links possibly hiding important parts of the URL (CVE-2023-36462)
+- Fix timeout handling of outbound HTTP requests (CVE-2023-36461)
+- Fix arbitrary file creation through media processing (CVE-2023-36460)
+- Fix possible XSS in preview cards (CVE-2023-36459)
+
 ## [4.1.2] - 2023-04-04

 ### Fixed
--- a/Gemfile.lock
+++ b/Gemfile.lock
@ -10,40 +10,40 @@ GIT
 GEM
  remote: https://rubygems.org/
  specs:
-    actioncable (6.1.7.2)
-      actionpack (= 6.1.7.2)
-      activesupport (= 6.1.7.2)
+    actioncable (6.1.7.4)
+      actionpack (= 6.1.7.4)
+      activesupport (= 6.1.7.4)
      nio4r (~> 2.0)
      websocket-driver (>= 0.6.1)
-    actionmailbox (6.1.7.2)
-      actionpack (= 6.1.7.2)
-      activejob (= 6.1.7.2)
-      activerecord (= 6.1.7.2)
-      activestorage (= 6.1.7.2)
-      activesupport (= 6.1.7.2)
+    actionmailbox (6.1.7.4)
+      actionpack (= 6.1.7.4)
+      activejob (= 6.1.7.4)
+      activerecord (= 6.1.7.4)
+      activestorage (= 6.1.7.4)
+      activesupport (= 6.1.7.4)
      mail (>= 2.7.1)
-    actionmailer (6.1.7.2)
-      actionpack (= 6.1.7.2)
-      actionview (= 6.1.7.2)
-      activejob (= 6.1.7.2)
-      activesupport (= 6.1.7.2)
+    actionmailer (6.1.7.4)
+      actionpack (= 6.1.7.4)
+      actionview (= 6.1.7.4)
+      activejob (= 6.1.7.4)
+      activesupport (= 6.1.7.4)
      mail (~> 2.5, >= 2.5.4)
      rails-dom-testing (~> 2.0)
-    actionpack (6.1.7.2)
-      actionview (= 6.1.7.2)
-      activesupport (= 6.1.7.2)
+    actionpack (6.1.7.4)
+      actionview (= 6.1.7.4)
+      activesupport (= 6.1.7.4)
      rack (~> 2.0, >= 2.0.9)
      rack-test (>= 0.6.3)
      rails-dom-testing (~> 2.0)
      rails-html-sanitizer (~> 1.0, >= 1.2.0)
-    actiontext (6.1.7.2)
-      actionpack (= 6.1.7.2)
-      activerecord (= 6.1.7.2)
-      activestorage (= 6.1.7.2)
-      activesupport (= 6.1.7.2)
+    actiontext (6.1.7.4)
+      actionpack (= 6.1.7.4)
+      activerecord (= 6.1.7.4)
+      activestorage (= 6.1.7.4)
+      activesupport (= 6.1.7.4)
      nokogiri (>= 1.8.5)
-    actionview (6.1.7.2)
-      activesupport (= 6.1.7.2)
+    actionview (6.1.7.4)
+      activesupport (= 6.1.7.4)
      builder (~> 3.1)
      erubi (~> 1.4)
      rails-dom-testing (~> 2.0)
@ -54,22 +54,22 @@ GEM
      case_transform (>= 0.2)
      jsonapi-renderer (>= 0.1.1.beta1, < 0.3)
    active_record_query_trace (1.8)
-    activejob (6.1.7.2)
-      activesupport (= 6.1.7.2)
+    activejob (6.1.7.4)
+      activesupport (= 6.1.7.4)
      globalid (>= 0.3.6)
-    activemodel (6.1.7.2)
-      activesupport (= 6.1.7.2)
-    activerecord (6.1.7.2)
-      activemodel (= 6.1.7.2)
-      activesupport (= 6.1.7.2)
-    activestorage (6.1.7.2)
-      actionpack (= 6.1.7.2)
-      activejob (= 6.1.7.2)
-      activerecord (= 6.1.7.2)
-      activesupport (= 6.1.7.2)
+    activemodel (6.1.7.4)
+      activesupport (= 6.1.7.4)
+    activerecord (6.1.7.4)
+      activemodel (= 6.1.7.4)
+      activesupport (= 6.1.7.4)
+    activestorage (6.1.7.4)
+      actionpack (= 6.1.7.4)
+      activejob (= 6.1.7.4)
+      activerecord (= 6.1.7.4)
+      activesupport (= 6.1.7.4)
      marcel (~> 1.0)
      mini_mime (>= 1.1.0)
-    activesupport (6.1.7.2)
+    activesupport (6.1.7.4)
      concurrent-ruby (~> 1.0, >= 1.0.2)
      i18n (>= 1.6, < 2)
      minitest (>= 5.1)
@ -173,7 +173,7 @@ GEM
    cocoon (1.2.15)
    coderay (1.1.3)
    color_diff (0.1)
-    concurrent-ruby (1.2.0)
+    concurrent-ruby (1.2.2)
    connection_pool (2.3.0)
    cose (1.2.1)
      cbor (~> 0.5.9)
@ -206,7 +206,7 @@ GEM
    docile (1.4.0)
    domain_name (0.5.20190701)
      unf (>= 0.0.5, < 1.0.0)
-    doorkeeper (5.6.4)
+    doorkeeper (5.6.6)
      railties (>= 5)
    dotenv (2.8.1)
    dotenv-rails (2.8.1)
@ -388,7 +388,7 @@ GEM
    loofah (2.19.1)
      crass (~> 1.0.2)
      nokogiri (>= 1.5.9)
-    mail (2.8.0.1)
+    mail (2.8.1)
      mini_mime (>= 0.1.1)
      net-imap
      net-pop
@ -405,12 +405,12 @@ GEM
      mime-types-data (~> 3.2015)
    mime-types-data (3.2022.0105)
    mini_mime (1.1.2)
-    mini_portile2 (2.8.1)
+    mini_portile2 (2.8.2)
    minitest (5.17.0)
    msgpack (1.6.0)
    multi_json (1.15.0)
    multipart-post (2.1.1)
-    net-imap (0.3.4)
+    net-imap (0.3.6)
      date
      net-protocol
    net-ldap (0.17.1)
@ -423,8 +423,8 @@ GEM
    net-smtp (0.3.3)
      net-protocol
    net-ssh (7.0.1)
-    nio4r (2.5.8)
-    nokogiri (1.14.1)
+    nio4r (2.5.9)
+    nokogiri (1.14.5)
      mini_portile2 (~> 2.8.0)
      racc (~> 1.4)
    nsa (0.2.8)
@ -497,7 +497,7 @@ GEM
      activesupport (>= 3.0.0)
    raabro (1.4.0)
    racc (1.6.2)
-    rack (2.2.6.2)
+    rack (2.2.7)
    rack-attack (6.6.1)
      rack (>= 1.0, < 3)
    rack-cors (1.1.1)
@ -512,20 +512,20 @@ GEM
      rack
    rack-test (2.0.2)
      rack (>= 1.3)
-    rails (6.1.7.2)
-      actioncable (= 6.1.7.2)
-      actionmailbox (= 6.1.7.2)
-      actionmailer (= 6.1.7.2)
-      actionpack (= 6.1.7.2)
-      actiontext (= 6.1.7.2)
-      actionview (= 6.1.7.2)
-      activejob (= 6.1.7.2)
-      activemodel (= 6.1.7.2)
-      activerecord (= 6.1.7.2)
-      activestorage (= 6.1.7.2)
-      activesupport (= 6.1.7.2)
+    rails (6.1.7.4)
+      actioncable (= 6.1.7.4)
+      actionmailbox (= 6.1.7.4)
+      actionmailer (= 6.1.7.4)
+      actionpack (= 6.1.7.4)
+      actiontext (= 6.1.7.4)
+      actionview (= 6.1.7.4)
+      activejob (= 6.1.7.4)
+      activemodel (= 6.1.7.4)
+      activerecord (= 6.1.7.4)
+      activestorage (= 6.1.7.4)
+      activesupport (= 6.1.7.4)
      bundler (>= 1.15.0)
-      railties (= 6.1.7.2)
+      railties (= 6.1.7.4)
      sprockets-rails (>= 2.0.0)
    rails-controller-testing (1.0.5)
      actionpack (>= 5.0.1.rc1)
@ -541,9 +541,9 @@ GEM
      railties (>= 6.0.0, < 7)
    rails-settings-cached (0.6.6)
      rails (>= 4.2.0)
-    railties (6.1.7.2)
-      actionpack (= 6.1.7.2)
-      activesupport (= 6.1.7.2)
+    railties (6.1.7.4)
+      actionpack (= 6.1.7.4)
+      activesupport (= 6.1.7.4)
      method_source
      rake (>= 12.2)
      thor (~> 1.0)
@ -688,9 +688,9 @@ GEM
      unicode-display_width (>= 1.1.1, < 3)
    terrapin (0.6.0)
      climate_control (>= 0.0.3, < 1.0)
-    thor (1.2.1)
+    thor (1.2.2)
    tilt (2.0.11)
-    timeout (0.3.1)
+    timeout (0.3.2)
    tpm-key_attestation (0.11.0)
      bindata (~> 2.4)
      openssl (> 2.0, < 3.1)
@ -753,7 +753,7 @@ GEM
    xorcist (1.1.3)
    xpath (3.2.0)
      nokogiri (~> 1.8)
-    zeitwerk (2.6.6)
+    zeitwerk (2.6.8)

 PLATFORMS
  ruby
--- a/app/controllers/backups_controller.rb
+++ b/app/controllers/backups_controller.rb
@ -13,7 +13,7 @@ class BackupsController < ApplicationController
    when :s3
      redirect_to @backup.dump.expiring_url(10)
    when :fog
-      if Paperclip::Attachment.default_options.dig(:storage, :fog_credentials, :openstack_temp_url_key).present?
+      if Paperclip::Attachment.default_options.dig(:fog_credentials, :openstack_temp_url_key).present?
        redirect_to @backup.dump.expiring_url(Time.now.utc + 10)
      else
        redirect_to full_asset_url(@backup.dump.url)
--- a/app/controllers/well_known/webfinger_controller.rb
+++ b/app/controllers/well_known/webfinger_controller.rb
@ -18,7 +18,14 @@ module WellKnown
    private

    def set_account
-      @account = Account.find_local!(username_from_resource)
+      username = username_from_resource
+      @account = begin
+        if username == Rails.configuration.x.local_domain
+          Account.representative
+        else
+          Account.find_local!(username)
+        end
+      end
    end

    def username_from_resource
--- a/app/lib/request.rb
+++ b/app/lib/request.rb
@ -11,15 +11,7 @@ require 'resolv'
 # Also changes how the read timeout behaves so that it is cumulative (closer
 # to HTTP::Timeout::Global, but still having distinct timeouts for other
 # operation types)
-class PerOperationWithDeadline < HTTP::Timeout::PerOperation
-  READ_DEADLINE = 30
-
-  def initialize(*args)
-    super
-
-    @read_deadline = options.fetch(:read_deadline, READ_DEADLINE)
-  end
-
+class HTTP::Timeout::PerOperation
  def connect(socket_class, host, port, nodelay = false)
    @socket = socket_class.open(host, port)
    @socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1) if nodelay
@ -32,7 +24,7 @@ class PerOperationWithDeadline < HTTP::Timeout::PerOperation

  # Read data from the socket
  def readpartial(size, buffer = nil)
-    @deadline ||= Process.clock_gettime(Process::CLOCK_MONOTONIC) + @read_deadline
+    @deadline ||= Process.clock_gettime(Process::CLOCK_MONOTONIC) + @read_timeout

    timeout = false
    loop do
@ -41,8 +33,7 @@ class PerOperationWithDeadline < HTTP::Timeout::PerOperation
      return :eof if result.nil?

      remaining_time = @deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
-      raise HTTP::TimeoutError, "Read timed out after #{@read_timeout} seconds" if timeout
-      raise HTTP::TimeoutError, "Read timed out after a total of #{@read_deadline} seconds" if remaining_time <= 0
+      raise HTTP::TimeoutError, "Read timed out after #{@read_timeout} seconds" if timeout || remaining_time <= 0
      return result if result != :wait_readable

      # marking the socket for timeout. Why is this not being raised immediately?
@ -55,7 +46,7 @@ class PerOperationWithDeadline < HTTP::Timeout::PerOperation
      # timeout. Else, the first timeout was a proper timeout.
      # This hack has to be done because io/wait#wait_readable doesn't provide a value for when
      # the socket is closed by the server, and HTTP::Parser doesn't provide the limit for the chunks.
-      timeout = true unless @socket.to_io.wait_readable([remaining_time, @read_timeout].min)
+      timeout = true unless @socket.to_io.wait_readable(remaining_time)
    end
  end
 end
--- a/app/lib/scope_parser.rb
+++ b/app/lib/scope_parser.rb
@ -1,7 +1,7 @@
 # frozen_string_literal: true

 class ScopeParser < Parslet::Parser
-  rule(:term)      { match('[a-z]').repeat(1).as(:term) }
+  rule(:term)      { match('[a-z_]').repeat(1).as(:term) }
  rule(:colon)     { str(':') }
  rule(:access)    { (str('write') | str('read')).as(:access) }
  rule(:namespace) { str('admin').as(:namespace) }
--- a/app/lib/text_formatter.rb
+++ b/app/lib/text_formatter.rb
@ -60,7 +60,7 @@ class TextFormatter
      suffix      = url[prefix.length + 30..-1]
      cutoff      = url[prefix.length..-1].length > 30

-      <<~HTML.squish.html_safe # rubocop:disable Rails/OutputSafety
+      <<~HTML.squish
        <a href="#{h(url)}" target="_blank" rel="#{rel.join(' ')}"><span class="invisible">#{h(prefix)}</span><span class="#{cutoff ? 'ellipsis' : ''}">#{h(display_url)}</span><span class="invisible">#{h(suffix)}</span></a>
      HTML
    rescue Addressable::URI::InvalidURIError, IDN::Idna::IdnaError
--- a/app/lib/vacuum/access_tokens_vacuum.rb
+++ b/app/lib/vacuum/access_tokens_vacuum.rb
@ -9,10 +9,12 @@ class Vacuum::AccessTokensVacuum
  private

  def vacuum_revoked_access_tokens!
-    Doorkeeper::AccessToken.where.not(revoked_at: nil).where('revoked_at < NOW()').delete_all
+    Doorkeeper::AccessToken.where.not(expires_in: nil).where('created_at + make_interval(secs => expires_in) < NOW()').in_batches.delete_all
+    Doorkeeper::AccessToken.where.not(revoked_at: nil).where('revoked_at < NOW()').in_batches.delete_all
  end

  def vacuum_revoked_access_grants!
-    Doorkeeper::AccessGrant.where.not(revoked_at: nil).where('revoked_at < NOW()').delete_all
+    Doorkeeper::AccessGrant.where.not(expires_in: nil).where('created_at + make_interval(secs => expires_in) < NOW()').in_batches.delete_all
+    Doorkeeper::AccessGrant.where.not(revoked_at: nil).where('revoked_at < NOW()').in_batches.delete_all
  end
 end
--- a/app/models/concerns/attachmentable.rb
+++ b/app/models/concerns/attachmentable.rb
@ -24,7 +24,7 @@ module Attachmentable
    def self.has_attached_file(name, options = {}) # rubocop:disable Naming/PredicateName
      super(name, options)

-      send(:"before_#{name}_validate", prepend: true) do
+      send(:"before_#{name}_validate") do
        attachment = send(name)
        check_image_dimension(attachment)
        set_file_content_type(attachment)
--- a/app/models/form/account_batch.rb
+++ b/app/models/form/account_batch.rb
@ -123,7 +123,18 @@ class Form::AccountBatch
      account: current_account,
      action: :suspend
    )
+
    Admin::SuspensionWorker.perform_async(account.id)
+
+    # Suspending a single account closes their associated reports, so
+    # mass-suspending would be consistent.
+    Report.where(target_account: account).unresolved.find_each do |report|
+      authorize(report, :update?)
+      log_action(:resolve, report)
+      report.resolve!(current_account)
+    rescue Mastodon::NotPermittedError
+      # This should not happen, but just in case, do not fail early
+    end
  end

  def approve_account(account)
--- a/app/workers/scheduler/accounts_statuses_cleanup_scheduler.rb
+++ b/app/workers/scheduler/accounts_statuses_cleanup_scheduler.rb
@ -7,28 +7,30 @@ class Scheduler::AccountsStatusesCleanupScheduler
  # This limit is mostly to be nice to the fediverse at large and not
  # generate too much traffic.
  # This also helps limiting the running time of the scheduler itself.
-  MAX_BUDGET         = 150
+  MAX_BUDGET         = 300

-  # This is an attempt to spread the load across instances, as various
-  # accounts are likely to have various followers.
+  # This is an attempt to spread the load across remote servers, as
+  # spreading deletions across diverse accounts is likely to spread
+  # the deletion across diverse followers. It also helps each individual
+  # user see some effect sooner.
  PER_ACCOUNT_BUDGET = 5

  # This is an attempt to limit the workload generated by status removal
-  # jobs to something the particular instance can handle.
-  PER_THREAD_BUDGET  = 6
-
-  # Those avoid loading an instance that is already under load
-  MAX_DEFAULT_SIZE    = 200
-  MAX_DEFAULT_LATENCY = 5
-  MAX_PUSH_SIZE       = 500
-  MAX_PUSH_LATENCY    = 10
-
-  # 'pull' queue has lower priority jobs, and it's unlikely that pushing
-  # deletes would cause much issues with this queue if it didn't cause issues
-  # with default and push. Yet, do not enqueue deletes if the instance is
-  # lagging behind too much.
-  MAX_PULL_SIZE       = 10_000
-  MAX_PULL_LATENCY    = 5.minutes.to_i
+  # jobs to something the particular server can handle.
+  PER_THREAD_BUDGET  = 5
+
+  # These are latency limits on various queues above which a server is
+  # considered to be under load, causing the auto-deletion to be entirely
+  # skipped for that run.
+  LOAD_LATENCY_THRESHOLDS = {
+    default: 5,
+    push: 10,
+    # The `pull` queue has lower priority jobs, and it's unlikely that
+    # pushing deletes would cause much issues with this queue if it didn't
+    # cause issues with `default` and `push`. Yet, do not enqueue deletes
+    # if the instance is lagging behind too much.
+    pull: 5.minutes.to_i,
+  }.freeze

  sidekiq_options retry: 0, lock: :until_executed, lock_ttl: 1.day.to_i

@ -36,17 +38,37 @@ class Scheduler::AccountsStatusesCleanupScheduler
    return if under_load?

    budget = compute_budget
-    first_policy_id = last_processed_id
+
+    # If the budget allows it, we want to consider all accounts with enabled
+    # auto cleanup at least once.
+    #
+    # We start from `first_policy_id` (the last processed id in the previous
+    # run) and process each policy until we loop to `first_policy_id`,
+    # recording into `affected_policies` any policy that caused posts to be
+    # deleted.
+    #
+    # After that, we set `full_iteration` to `false` and continue looping on
+    # policies from `affected_policies`.
+    first_policy_id   = last_processed_id || 0
+    first_iteration   = true
+    full_iteration    = true
+    affected_policies = []

    loop do
      num_processed_accounts = 0

-      scope = AccountStatusesCleanupPolicy.where(enabled: true)
-      scope.where(Account.arel_table[:id].gt(first_policy_id)) if first_policy_id.present?
+      scope = cleanup_policies(first_policy_id, affected_policies, first_iteration, full_iteration)
      scope.find_each(order: :asc) do |policy|
        num_deleted = AccountStatusesCleanupService.new.call(policy, [budget, PER_ACCOUNT_BUDGET].min)
-        num_processed_accounts += 1 unless num_deleted.zero?
        budget -= num_deleted
+
+        unless num_deleted.zero?
+          num_processed_accounts += 1
+          affected_policies << policy.id if full_iteration
+        end
+
+        full_iteration = false if !first_iteration && policy.id >= first_policy_id
+
        if budget.zero?
          save_last_processed_id(policy.id)
          break
@ -55,36 +77,55 @@ class Scheduler::AccountsStatusesCleanupScheduler

      # The idea here is to loop through all policies at least once until the budget is exhausted
      # and start back after the last processed account otherwise
-      break if budget.zero? || (num_processed_accounts.zero? && first_policy_id.nil?)
-      first_policy_id = nil
+      break if budget.zero? || (num_processed_accounts.zero? && !full_iteration)
+
+      full_iteration  = false unless first_iteration
+      first_iteration = false
    end
  end

  def compute_budget
-    threads = Sidekiq::ProcessSet.new.select { |x| x['queues'].include?('push') }.map { |x| x['concurrency'] }.sum
+    # Each post deletion is a `RemovalWorker` job (on `default` queue), each
+    # potentially spawning many `ActivityPub::DeliveryWorker` jobs (on the `push` queue).
+    threads = Sidekiq::ProcessSet.new.select { |x| x['queues'].include?('push') }.pluck('concurrency').sum
    [PER_THREAD_BUDGET * threads, MAX_BUDGET].min
  end

  def under_load?
-    queue_under_load?('default', MAX_DEFAULT_SIZE, MAX_DEFAULT_LATENCY) || queue_under_load?('push', MAX_PUSH_SIZE, MAX_PUSH_LATENCY) || queue_under_load?('pull', MAX_PULL_SIZE, MAX_PULL_LATENCY)
+    LOAD_LATENCY_THRESHOLDS.any? { |queue, max_latency| queue_under_load?(queue, max_latency) }
  end

  private

-  def queue_under_load?(name, max_size, max_latency)
-    queue = Sidekiq::Queue.new(name)
-    queue.size > max_size || queue.latency > max_latency
+  def cleanup_policies(first_policy_id, affected_policies, first_iteration, full_iteration)
+    scope = AccountStatusesCleanupPolicy.where(enabled: true)
+
+    if full_iteration
+      # If we are doing a full iteration, examine all policies we have not examined yet
+      if first_iteration
+        scope.where(id: first_policy_id...)
+      else
+        scope.where(id: ..first_policy_id).or(scope.where(id: affected_policies))
+      end
+    else
+      # Otherwise, examine only policies that previously yielded posts to delete
+      scope.where(id: affected_policies)
+    end
+  end
+
+  def queue_under_load?(name, max_latency)
+    Sidekiq::Queue.new(name).latency > max_latency
  end

  def last_processed_id
-    redis.get('account_statuses_cleanup_scheduler:last_account_id')
+    redis.get('account_statuses_cleanup_scheduler:last_policy_id')&.to_i
  end

  def save_last_processed_id(id)
    if id.nil?
-      redis.del('account_statuses_cleanup_scheduler:last_account_id')
+      redis.del('account_statuses_cleanup_scheduler:last_policy_id')
    else
-      redis.set('account_statuses_cleanup_scheduler:last_account_id', id, ex: 1.hour.seconds)
+      redis.set('account_statuses_cleanup_scheduler:last_policy_id', id, ex: 1.hour.seconds)
    end
  end
 end
--- a/app/workers/scheduler/indexing_scheduler.rb
+++ b/app/workers/scheduler/indexing_scheduler.rb
@ -9,6 +9,9 @@ class Scheduler::IndexingScheduler
  IMPORT_BATCH_SIZE = 1000
  SCAN_BATCH_SIZE = 10 * IMPORT_BATCH_SIZE

+  IMPORT_BATCH_SIZE = 1000
+  SCAN_BATCH_SIZE = 10 * IMPORT_BATCH_SIZE
+
  def perform
    return unless Chewy.enabled?

--- a/config/initializers/content_security_policy.rb
+++ b/config/initializers/content_security_policy.rb
@ -3,7 +3,7 @@
 # https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Security-Policy

 def host_to_url(str)
-  "http#{Rails.configuration.x.use_https ? 's' : ''}://#{str.split('/').first}" if str.present?
+  "http#{Rails.configuration.x.use_https ? 's' : ''}://#{str}".split('/').first if str.present?
 end

 base_host = Rails.configuration.x.web_domain
--- a/lib/mastodon/version.rb
+++ b/lib/mastodon/version.rb
@ -13,7 +13,7 @@ module Mastodon
    end

    def patch
-      2
+      3
    end

    def flags
--- a/lib/public_file_server_middleware.rb
+++ b/lib/public_file_server_middleware.rb
@ -32,6 +32,11 @@ class PublicFileServerMiddleware
      end
    end

+    # Override the default CSP header set by the CSP middleware
+    headers['Content-Security-Policy'] = "default-src 'none'; form-action 'none'" if request_path.start_with?(paperclip_root_url)
+
+    headers['X-Content-Type-Options'] = 'nosniff'
+
    [status, headers, response]
  end

--- a/spec/controllers/well_known/webfinger_controller_spec.rb
+++ b/spec/controllers/well_known/webfinger_controller_spec.rb
@ -4,6 +4,10 @@ describe WellKnown::WebfingerController, type: :controller do
  render_views

  describe 'GET #show' do
+    subject(:perform_show!) do
+      get :show, params: { resource: resource }, format: :json
+    end
+
    let(:alternate_domains) { [] }
    let(:alice) { Fabricate(:account, username: 'alice') }
    let(:resource) { nil }
@ -15,10 +19,6 @@ describe WellKnown::WebfingerController, type: :controller do
      Rails.configuration.x.alternate_domains = tmp
    end

-    subject do
-      get :show, params: { resource: resource }, format: :json
-    end
-
    shared_examples 'a successful response' do
      it 'returns http success' do
        expect(response).to have_http_status(200)
@ -43,7 +43,7 @@ describe WellKnown::WebfingerController, type: :controller do
      let(:resource) { alice.to_webfinger_s }

      before do
-        subject
+        perform_show!
      end

      it_behaves_like 'a successful response'
@ -54,7 +54,7 @@ describe WellKnown::WebfingerController, type: :controller do

      before do
        alice.suspend!
-        subject
+        perform_show!
      end

      it_behaves_like 'a successful response'
@ -66,7 +66,7 @@ describe WellKnown::WebfingerController, type: :controller do
      before do
        alice.suspend!
        alice.deletion_request.destroy
-        subject
+        perform_show!
      end

      it 'returns http gone' do
@ -78,7 +78,7 @@ describe WellKnown::WebfingerController, type: :controller do
      let(:resource) { 'acct:not@existing.com' }

      before do
-        subject
+        perform_show!
      end

      it 'returns http not found' do
@ -90,7 +90,7 @@ describe WellKnown::WebfingerController, type: :controller do
      let(:alternate_domains) { ['foo.org'] }

      before do
-        subject
+        perform_show!
      end

      context 'when an account exists' do
@ -114,11 +114,39 @@ describe WellKnown::WebfingerController, type: :controller do
      end
    end

+    context 'when the old name scheme is used to query the instance actor' do
+      let(:resource) do
+        "#{Rails.configuration.x.local_domain}@#{Rails.configuration.x.local_domain}"
+      end
+
+      before do
+        perform_show!
+      end
+
+      it 'returns http success' do
+        expect(response).to have_http_status(200)
+      end
+
+      it 'does not set a Vary header' do
+        expect(response.headers['Vary']).to be_nil
+      end
+
+      it 'returns application/jrd+json' do
+        expect(response.media_type).to eq 'application/jrd+json'
+      end
+
+      it 'returns links for the internal account' do
+        json = body_as_json
+        expect(json[:subject]).to eq 'acct:mastodon.internal@cb6e6126.ngrok.io'
+        expect(json[:aliases]).to eq ['https://cb6e6126.ngrok.io/actor']
+      end
+    end
+
    context 'with no resource parameter' do
      let(:resource) { nil }

      before do
-        subject
+        perform_show!
      end

      it 'returns http bad request' do
@ -130,7 +158,7 @@ describe WellKnown::WebfingerController, type: :controller do
      let(:resource) { 'df/:dfkj' }

      before do
-        subject
+        perform_show!
      end

      it 'returns http bad request' do
--- a/spec/lib/vacuum/access_tokens_vacuum_spec.rb
+++ b/spec/lib/vacuum/access_tokens_vacuum_spec.rb
@ -5,9 +5,11 @@ RSpec.describe Vacuum::AccessTokensVacuum do

  describe '#perform' do
    let!(:revoked_access_token) { Fabricate(:access_token, revoked_at: 1.minute.ago) }
+    let!(:expired_access_token) { Fabricate(:access_token, expires_in: 59.minutes.to_i, created_at: 1.hour.ago) }
    let!(:active_access_token) { Fabricate(:access_token) }

    let!(:revoked_access_grant) { Fabricate(:access_grant, revoked_at: 1.minute.ago) }
+    let!(:expired_access_grant) { Fabricate(:access_grant, expires_in: 59.minutes.to_i, created_at: 1.hour.ago) }
    let!(:active_access_grant) { Fabricate(:access_grant) }

    before do
@ -18,10 +20,18 @@ RSpec.describe Vacuum::AccessTokensVacuum do
      expect { revoked_access_token.reload }.to raise_error ActiveRecord::RecordNotFound
    end

+    it 'deletes expired access tokens' do
+      expect { expired_access_token.reload }.to raise_error ActiveRecord::RecordNotFound
+    end
+
    it 'deletes revoked access grants' do
      expect { revoked_access_grant.reload }.to raise_error ActiveRecord::RecordNotFound
    end

+    it 'deletes expired access grants' do
+      expect { expired_access_grant.reload }.to raise_error ActiveRecord::RecordNotFound
+    end
+
    it 'does not delete active access tokens' do
      expect { active_access_token.reload }.to_not raise_error
    end
--- a/spec/models/form/account_batch_spec.rb
+++ b/spec/models/form/account_batch_spec.rb
@ -0,0 +1,63 @@
+# frozen_string_literal: true
+
+require 'rails_helper'
+
+RSpec.describe Form::AccountBatch do
+  let(:account_batch) { described_class.new }
+
+  describe '#save' do
+    subject           { account_batch.save }
+
+    let(:account)     { Fabricate(:user, role: UserRole.find_by(name: 'Admin')).account }
+    let(:account_ids) { [] }
+    let(:query)       { Account.none }
+
+    before do
+      account_batch.assign_attributes(
+        action: action,
+        current_account: account,
+        account_ids: account_ids,
+        query: query,
+        select_all_matching: select_all_matching
+      )
+    end
+
+    context 'when action is "suspend"' do
+      let(:action) { 'suspend' }
+
+      let(:target_account)  { Fabricate(:account) }
+      let(:target_account2) { Fabricate(:account) }
+
+      before do
+        Fabricate(:report, target_account: target_account)
+        Fabricate(:report, target_account: target_account2)
+      end
+
+      context 'when accounts are passed as account_ids' do
+        let(:select_all_matching) { '0' }
+        let(:account_ids)         { [target_account.id, target_account2.id] }
+
+        it 'suspends the expected users' do
+          expect { subject }.to change { [target_account.reload.suspended?, target_account2.reload.suspended?] }.from([false, false]).to([true, true])
+        end
+
+        it 'closes open reports targeting the suspended users' do
+          expect { subject }.to change { Report.unresolved.where(target_account: [target_account, target_account2]).count }.from(2).to(0)
+        end
+      end
+
+      context 'when accounts are passed as a query' do
+        let(:select_all_matching) { '1' }
+        let(:query)               { Account.where(id: [target_account.id, target_account2.id]) }
+
+        it 'suspends the expected users' do
+          expect { subject }.to change { [target_account.reload.suspended?, target_account2.reload.suspended?] }.from([false, false]).to([true, true])
+        end
+
+        it 'closes open reports targeting the suspended users' do
+          expect { subject }.to change { Report.unresolved.where(target_account: [target_account, target_account2]).count }.from(2).to(0)
+        end
+      end
+    end
+  end
+end
--- a/spec/workers/scheduler/accounts_statuses_cleanup_scheduler_spec.rb
+++ b/spec/workers/scheduler/accounts_statuses_cleanup_scheduler_spec.rb
@ -7,11 +7,13 @@ describe Scheduler::AccountsStatusesCleanupScheduler do
  let!(:account2)  { Fabricate(:account, domain: nil) }
  let!(:account3)  { Fabricate(:account, domain: nil) }
  let!(:account4)  { Fabricate(:account, domain: nil) }
+  let!(:account5)  { Fabricate(:account, domain: nil) }
  let!(:remote)    { Fabricate(:account) }

  let!(:policy1)   { Fabricate(:account_statuses_cleanup_policy, account: account1) }
  let!(:policy2)   { Fabricate(:account_statuses_cleanup_policy, account: account3) }
  let!(:policy3)   { Fabricate(:account_statuses_cleanup_policy, account: account4, enabled: false) }
+  let!(:policy4)   { Fabricate(:account_statuses_cleanup_policy, account: account5) }

  let(:queue_size)       { 0 }
  let(:queue_latency)    { 0 }
@ -40,6 +42,7 @@ describe Scheduler::AccountsStatusesCleanupScheduler do
      Fabricate(:status, account: account2, created_at: 3.years.ago)
      Fabricate(:status, account: account3, created_at: 3.years.ago)
      Fabricate(:status, account: account4, created_at: 3.years.ago)
+      Fabricate(:status, account: account5, created_at: 3.years.ago)
      Fabricate(:status, account: remote, created_at: 3.years.ago)
    end

@ -70,7 +73,7 @@ describe Scheduler::AccountsStatusesCleanupScheduler do
    end
  end

-  describe '#get_budget' do
+  describe '#compute_budget' do
    context 'on a single thread' do
      let(:process_set_stub) { [ { 'concurrency' => 1, 'queues' => ['push', 'default'] } ] }

@ -109,8 +112,48 @@ describe Scheduler::AccountsStatusesCleanupScheduler do
        expect { subject.perform }.to_not change { account4.statuses.count }
      end

-      it 'eventually deletes every deletable toot' do
-        expect { subject.perform; subject.perform; subject.perform; subject.perform }.to change { Status.count }.by(-20)
+      it 'eventually deletes every deletable toot given enough runs' do
+        stub_const 'Scheduler::AccountsStatusesCleanupScheduler::MAX_BUDGET', 4
+
+        expect { 10.times { subject.perform } }.to change { Status.count }.by(-30)
+      end
+
+      it 'correctly round-trips between users across several runs' do
+        stub_const 'Scheduler::AccountsStatusesCleanupScheduler::MAX_BUDGET', 3
+        stub_const 'Scheduler::AccountsStatusesCleanupScheduler::PER_ACCOUNT_BUDGET', 2
+
+        expect { 3.times { subject.perform } }
+          .to change { Status.count }.by(-3 * 3)
+          .and change { account1.statuses.count }
+          .and change { account3.statuses.count }
+          .and change { account5.statuses.count }
+      end
+
+      context 'when given a big budget' do
+        let(:process_set_stub) { [{ 'concurrency' => 400, 'queues' => %w(push default) }] }
+
+        before do
+          stub_const 'Scheduler::AccountsStatusesCleanupScheduler::MAX_BUDGET', 400
+        end
+
+        it 'correctly handles looping in a single run' do
+          expect(subject.compute_budget).to eq(400)
+          expect { subject.perform }.to change { Status.count }.by(-30)
+        end
+      end
+
+      context 'when there is no work to be done' do
+        let(:process_set_stub) { [{ 'concurrency' => 400, 'queues' => %w(push default) }] }
+
+        before do
+          stub_const 'Scheduler::AccountsStatusesCleanupScheduler::MAX_BUDGET', 400
+          subject.perform
+        end
+
+        it 'does not get stuck' do
+          expect(subject.compute_budget).to eq(400)
+          expect { subject.perform }.to_not change { Status.count }
+        end
      end
    end
  end
--- a/streaming/index.js
+++ b/streaming/index.js
@ -226,15 +226,9 @@ const startWorker = async (workerId) => {
    callbacks.forEach(callback => callback(json));
  };

-  /**
-   * @callback SubscriptionListener
-   * @param {ReturnType<parseJSON>} json of the message
-   * @returns void
-   */
-
  /**
   * @param {string} channel
-   * @param {SubscriptionListener} callback
+   * @param {function(string): void} callback
   */
  const subscribe = (channel, callback) => {
    log.silly(`Adding listener for ${channel}`);
@ -251,7 +245,7 @@ const startWorker = async (workerId) => {

  /**
   * @param {string} channel
-   * @param {SubscriptionListener} callback
+   * @param {function(Object<string, any>): void} callback
   */
  const unsubscribe = (channel, callback) => {
    log.silly(`Removing listener for ${channel}`);
@ -629,29 +623,27 @@ const startWorker = async (workerId) => {
   * @param {string[]} ids
   * @param {any} req
   * @param {function(string, string): void} output
-   * @param {undefined | function(string[], SubscriptionListener): void} attachCloseHandler
+   * @param {function(string[], function(string): void): void} attachCloseHandler
   * @param {boolean=} needsFiltering
-   * @returns {SubscriptionListener}
+   * @returns {function(object): void}
   */
  const streamFrom = (ids, req, output, attachCloseHandler, needsFiltering = false) => {
    const accountId = req.accountId || req.remoteAddress;

    log.verbose(req.requestId, `Starting stream from ${ids.join(', ')} for ${accountId}`);

-    const transmit = (event, payload) => {
-      // TODO: Replace "string"-based delete payloads with object payloads:
-      const encodedPayload = typeof payload === 'object' ? JSON.stringify(payload) : payload;
+    // Currently message is of type string, soon it'll be Record<string, any>
+    const listener = message => {
+      const { event, payload, queued_at } = message;

-      log.silly(req.requestId, `Transmitting for ${accountId}: ${event} ${encodedPayload}`);
-      output(event, encodedPayload);
-    };
+      const transmit = () => {
+        const now = new Date().getTime();
+        const delta = now - queued_at;
+        const encodedPayload = typeof payload === 'object' ? JSON.stringify(payload) : payload;

-    // The listener used to process each message off the redis subscription,
-    // message here is an object with an `event` and `payload` property. Some
-    // events also include a queued_at value, but this is being removed shortly.
-    /** @type {SubscriptionListener} */
-    const listener = message => {
-      const { event, payload } = message;
+        log.silly(req.requestId, `Transmitting for ${accountId}: ${event} ${encodedPayload} Delay: ${delta}ms`);
+        output(event, encodedPayload);
+      };

      // Only send local-only statuses to logged-in users
      if (payload.local_only && !req.accountId) {
@ -659,42 +651,29 @@ const startWorker = async (workerId) => {
        return;
      }

-      // Streaming only needs to apply filtering to some channels and only to
-      // some events. This is because majority of the filtering happens on the
-      // Ruby on Rails side when producing the event for streaming.
-      //
-      // The only events that require filtering from the streaming server are
-      // `update` and `status.update`, all other events are transmitted to the
-      // client as soon as they're received (pass-through).
-      //
-      // The channels that need filtering are determined in the function
-      // `channelNameToIds` defined below:
-      if (!needsFiltering || (event !== 'update' && event !== 'status.update')) {
-        transmit(event, payload);
+      // Only messages that may require filtering are statuses, since notifications
+      // are already personalized and deletes do not matter
+      if (!needsFiltering || event !== 'update') {
+        transmit();
        return;
      }

-      // The rest of the logic from here on in this function is to handle
-      // filtering of statuses:
+      const unpackedPayload = payload;
+      const targetAccountIds = [unpackedPayload.account.id].concat(unpackedPayload.mentions.map(item => item.id));
+      const accountDomain = unpackedPayload.account.acct.split('@')[1];

-      // Filter based on language:
-      if (Array.isArray(req.chosenLanguages) && payload.language !== null && req.chosenLanguages.indexOf(payload.language) === -1) {
-        log.silly(req.requestId, `Message ${payload.id} filtered by language (${payload.language})`);
+      if (Array.isArray(req.chosenLanguages) && unpackedPayload.language !== null && req.chosenLanguages.indexOf(unpackedPayload.language) === -1) {
+        log.silly(req.requestId, `Message ${unpackedPayload.id} filtered by language (${unpackedPayload.language})`);
        return;
      }

      // When the account is not logged in, it is not necessary to confirm the block or mute
      if (!req.accountId) {
-        transmit(event, payload);
+        transmit();
        return;
      }

-      // Filter based on domain blocks, blocks, mutes, or custom filters:
-      const targetAccountIds = [payload.account.id].concat(payload.mentions.map(item => item.id));
-      const accountDomain = payload.account.acct.split('@')[1];
-
-      // TODO: Move this logic out of the message handling loop
-      pgPool.connect((err, client, releasePgConnection) => {
+      pgPool.connect((err, client, done) => {
        if (err) {
          log.error(err);
          return;
@ -709,57 +688,40 @@ const startWorker = async (workerId) => {
                        SELECT 1
                        FROM mutes
                        WHERE account_id = $1
-                          AND target_account_id IN (${placeholders(targetAccountIds, 2)})`, [req.accountId, payload.account.id].concat(targetAccountIds)),
+                          AND target_account_id IN (${placeholders(targetAccountIds, 2)})`, [req.accountId, unpackedPayload.account.id].concat(targetAccountIds)),
        ];

        if (accountDomain) {
          queries.push(client.query('SELECT 1 FROM account_domain_blocks WHERE account_id = $1 AND domain = $2', [req.accountId, accountDomain]));
        }

-        if (!payload.filtered && !req.cachedFilters) {
+        if (!unpackedPayload.filtered && !req.cachedFilters) {
          queries.push(client.query('SELECT filter.id AS id, filter.phrase AS title, filter.context AS context, filter.expires_at AS expires_at, filter.action AS filter_action, keyword.keyword AS keyword, keyword.whole_word AS whole_word FROM custom_filter_keywords keyword JOIN custom_filters filter ON keyword.custom_filter_id = filter.id WHERE filter.account_id = $1 AND (filter.expires_at IS NULL OR filter.expires_at > NOW())', [req.accountId]));
        }

        Promise.all(queries).then(values => {
-          releasePgConnection();
+          done();

-          // Handling blocks & mutes and domain blocks: If one of those applies,
-          // then we don't transmit the payload of the event to the client
          if (values[0].rows.length > 0 || (accountDomain && values[1].rows.length > 0)) {
            return;
          }

-          // If the payload already contains the `filtered` property, it means
-          // that filtering has been applied on the ruby on rails side, as
-          // such, we don't need to construct or apply the filters in streaming:
-          if (Object.prototype.hasOwnProperty.call(payload, "filtered")) {
-            transmit(event, payload);
-            return;
-          }
-
-          // Handling for constructing the custom filters and caching them on the request
-          // TODO: Move this logic out of the message handling lifecycle
-          if (!req.cachedFilters) {
+          if (!unpackedPayload.filtered && !req.cachedFilters) {
            const filterRows = values[accountDomain ? 2 : 1].rows;

-            req.cachedFilters = filterRows.reduce((cache, filter) => {
-              if (cache[filter.id]) {
-                cache[filter.id].keywords.push([filter.keyword, filter.whole_word]);
+            req.cachedFilters = filterRows.reduce((cache, row) => {
+              if (cache[row.id]) {
+                cache[row.id].keywords.push([row.keyword, row.whole_word]);
              } else {
-                cache[filter.id] = {
-                  keywords: [[filter.keyword, filter.whole_word]],
-                  expires_at: filter.expires_at,
-                  filter: {
-                    id: filter.id,
-                    title: filter.title,
-                    context: filter.context,
-                    expires_at: filter.expires_at,
-                    // filter.filter_action is the value from the
-                    // custom_filters.action database column, it is an integer
-                    // representing a value in an enum defined by Ruby on Rails:
-                    //
-                    // enum { warn: 0, hide: 1 }
-                    filter_action: ['warn', 'hide'][filter.filter_action],
+                cache[row.id] = {
+                  keywords: [[row.keyword, row.whole_word]],
+                  expires_at: row.expires_at,
+                  repr: {
+                    id: row.id,
+                    title: row.title,
+                    context: row.context,
+                    expires_at: row.expires_at,
+                    filter_action: ['warn', 'hide'][row.filter_action],
                  },
                };
              }
@ -767,10 +729,6 @@ const startWorker = async (workerId) => {
              return cache;
            }, {});

-            // Construct the regular expressions for the custom filters: This
-            // needs to be done in a separate loop as the database returns one
-            // filterRow per keyword, so we need all the keywords before
-            // constructing the regular expression
            Object.keys(req.cachedFilters).forEach((key) => {
              req.cachedFilters[key].regexp = new RegExp(req.cachedFilters[key].keywords.map(([keyword, whole_word]) => {
                let expr = keyword.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
@ -790,58 +748,31 @@ const startWorker = async (workerId) => {
            });
          }

-          // Apply cachedFilters against the payload, constructing a
-          // `filter_results` array of FilterResult entities
-          if (req.cachedFilters) {
-            const status = payload;
-            // TODO: Calculate searchableContent in Ruby on Rails:
-            const searchableContent = ([status.spoiler_text || '', status.content].concat((status.poll && status.poll.options) ? status.poll.options.map(option => option.title) : [])).concat(status.media_attachments.map(att => att.description)).join('\n\n').replace(/<br\s*\/?>/g, '\n').replace(/<\/p><p>/g, '\n\n');
-            const searchableTextContent = JSDOM.fragment(searchableContent).textContent;
+          // Check filters
+          if (req.cachedFilters && !unpackedPayload.filtered) {
+            const status = unpackedPayload;
+            const searchContent = ([status.spoiler_text || '', status.content].concat((status.poll && status.poll.options) ? status.poll.options.map(option => option.title) : [])).concat(status.media_attachments.map(att => att.description)).join('\n\n').replace(/<br\s*\/?>/g, '\n').replace(/<\/p><p>/g, '\n\n');
+            const searchIndex = JSDOM.fragment(searchContent).textContent;

            const now = new Date();
-            const filter_results = Object.values(req.cachedFilters).reduce((results, cachedFilter) => {
-              // Check the filter hasn't expired before applying:
-              if (cachedFilter.expires_at !== null && cachedFilter.expires_at < now) {
-                return results;
-              }
-
-              // Just in-case JSDOM fails to find textContent in searchableContent
-              if (!searchableTextContent) {
-                return results;
-              }
-
-              const keyword_matches = searchableTextContent.match(cachedFilter.regexp);
-              if (keyword_matches) {
-                // results is an Array of FilterResult; status_matches is always
-                // null as we only are only applying the keyword-based custom
-                // filters, not the status-based custom filters.
-                // https://docs.joinmastodon.org/entities/FilterResult/
-                results.push({
-                  filter: cachedFilter.filter,
-                  keyword_matches,
-                  status_matches: null
-                });
+            payload.filtered = [];
+            Object.values(req.cachedFilters).forEach((cachedFilter) => {
+              if ((cachedFilter.expires_at === null || cachedFilter.expires_at > now)) {
+                const keyword_matches = searchIndex.match(cachedFilter.regexp);
+                if (keyword_matches) {
+                  payload.filtered.push({
+                    filter: cachedFilter.repr,
+                    keyword_matches,
+                  });
+                }
              }
-
-              return results;
-            }, []);
-
-            // Send the payload + the FilterResults as the `filtered` property
-            // to the streaming connection. To reach this code, the `event` must
-            // have been either `update` or `status.update`, meaning the
-            // `payload` is a Status entity, which has a `filtered` property:
-            //
-            // filtered: https://docs.joinmastodon.org/entities/Status/#filtered
-            transmit(event, {
-              ...payload,
-              filtered: filter_results
            });
-          } else {
-            transmit(event, payload);
          }
+
+          transmit();
        }).catch(err => {
-          releasePgConnection();
          log.error(err);
+          done();
        });
      });
    };
@ -850,7 +781,7 @@ const startWorker = async (workerId) => {
      subscribe(`${redisPrefix}${id}`, listener);
    });

-    if (typeof attachCloseHandler === 'function') {
+    if (attachCloseHandler) {
      attachCloseHandler(ids.map(id => `${redisPrefix}${id}`), listener);
    }

@ -887,13 +818,12 @@ const startWorker = async (workerId) => {
  /**
   * @param {any} req
   * @param {function(): void} [closeHandler]
-   * @returns {function(string[], SubscriptionListener): void}
+   * @return {function(string[]): void}
   */
-
-  const streamHttpEnd = (req, closeHandler = undefined) => (ids, listener) => {
+  const streamHttpEnd = (req, closeHandler = undefined) => (ids) => {
    req.on('close', () => {
      ids.forEach(id => {
-        unsubscribe(id, listener);
+        unsubscribe(id);
      });

      if (closeHandler) {
@ -1153,7 +1083,7 @@ const startWorker = async (workerId) => {
   * @typedef WebSocketSession
   * @property {any} socket
   * @property {any} request
-   * @property {Object.<string, { listener: SubscriptionListener, stopHeartbeat: function(): void }>} subscriptions
+   * @property {Object.<string, { listener: function(string): void, stopHeartbeat: function(): void }>} subscriptions
   */

  /**