diff options
Diffstat (limited to 'app')
| -rw-r--r-- | app/controllers/api/v1/accounts/relationships_controller.rb | 5 | ||||
| -rw-r--r-- | app/lib/feed_manager.rb | 128 | ||||
| -rw-r--r-- | app/models/feed.rb | 2 | ||||
| -rw-r--r-- | app/services/batched_remove_status_service.rb | 37 | ||||
| -rw-r--r-- | app/services/precompute_feed_service.rb | 38 | ||||
| -rw-r--r-- | app/services/remove_status_service.rb | 8 |
6 files changed, 118 insertions, 100 deletions
diff --git a/app/controllers/api/v1/accounts/relationships_controller.rb b/app/controllers/api/v1/accounts/relationships_controller.rb index a88cf2021..91a942d75 100644 --- a/app/controllers/api/v1/accounts/relationships_controller.rb +++ b/app/controllers/api/v1/accounts/relationships_controller.rb @@ -7,7 +7,10 @@ class Api::V1::Accounts::RelationshipsController < Api::BaseController respond_to :json def index - @accounts = Account.where(id: account_ids).select('id') + accounts = Account.where(id: account_ids).select('id') + # .where doesn't guarantee that our results are in the same order + # we requested them, so return the "right" order to the requestor. + @accounts = accounts.index_by(&:id).values_at(*account_ids) render json: @accounts, each_serializer: REST::RelationshipSerializer, relationships: relationships end diff --git a/app/lib/feed_manager.rb b/app/lib/feed_manager.rb index b1ae11084..c509c5702 100644 --- a/app/lib/feed_manager.rb +++ b/app/lib/feed_manager.rb @@ -7,8 +7,13 @@ class FeedManager MAX_ITEMS = 400 - def key(type, id) - "feed:#{type}:#{id}" + # Must be <= MAX_ITEMS or the tracking sets will grow forever + REBLOG_FALLOFF = 40 + + def key(type, id, subtype = nil) + return "feed:#{type}:#{id}" unless subtype + + "feed:#{type}:#{id}:#{subtype}" end def filter?(timeline_type, status, receiver_id) @@ -22,23 +27,36 @@ class FeedManager end def push(timeline_type, account, status) - timeline_key = key(timeline_type, account.id) + return false unless add_to_feed(timeline_type, account, status) - if status.reblog? - # If the original status is within 40 statuses from top, do not re-insert it into the feed - rank = redis.zrevrank(timeline_key, status.reblog_of_id) - return if !rank.nil? && rank < 40 - redis.zadd(timeline_key, status.id, status.reblog_of_id) - else - redis.zadd(timeline_key, status.id, status.id) - trim(timeline_type, account.id) - end + trim(timeline_type, account.id) PushUpdateWorker.perform_async(account.id, status.id) if push_update_required?(timeline_type, account.id) + + true + end + + def unpush(timeline_type, account, status) + return false unless remove_from_feed(timeline_type, account, status) + + payload = Oj.dump(event: :delete, payload: status.id.to_s) + Redis.current.publish("timeline:#{account.id}", payload) + + true end def trim(type, account_id) - redis.zremrangebyrank(key(type, account_id), '0', (-(FeedManager::MAX_ITEMS + 1)).to_s) + timeline_key = key(type, account_id) + reblog_key = key(type, account_id, 'reblogs') + # Remove any items past the MAX_ITEMS'th entry in our feed + redis.zremrangebyrank(timeline_key, '0', (-(FeedManager::MAX_ITEMS + 1)).to_s) + + # Get the score of the REBLOG_FALLOFF'th item in our feed, and stop + # tracking anything after it for deduplication purposes. + falloff_rank = FeedManager::REBLOG_FALLOFF - 1 + falloff_range = redis.zrevrange(timeline_key, falloff_rank, falloff_rank, with_scores: true) + falloff_score = falloff_range&.first&.last&.to_i || 0 + redis.zremrangebyscore(reblog_key, 0, falloff_score) end def push_update_required?(timeline_type, account_id) @@ -54,11 +72,9 @@ class FeedManager query = query.where('id > ?', oldest_home_score) end - redis.pipelined do - query.each do |status| - next if status.direct_visibility? || filter?(:home, status, into_account) - redis.zadd(timeline_key, status.id, status.id) - end + query.each do |status| + next if status.direct_visibility? || filter?(:home, status, into_account) + add_to_feed(:home, into_account, status) end trim(:home, into_account.id) @@ -69,11 +85,8 @@ class FeedManager oldest_home_score = redis.zrange(timeline_key, 0, 0, with_scores: true)&.first&.last&.to_i || 0 from_account.statuses.select('id').where('id > ?', oldest_home_score).reorder(nil).find_in_batches do |statuses| - redis.pipelined do - statuses.each do |status| - redis.zrem(timeline_key, status.id) - redis.zremrangebyscore(timeline_key, status.id, status.id) - end + statuses.each do |status| + unpush(:home, into_account, status) end end end @@ -81,9 +94,20 @@ class FeedManager def clear_from_timeline(account, target_account) timeline_key = key(:home, account.id) timeline_status_ids = redis.zrange(timeline_key, 0, -1) - target_status_ids = Status.where(id: timeline_status_ids, account: target_account).ids + target_statuses = Status.where(id: timeline_status_ids, account: target_account) - redis.zrem(timeline_key, target_status_ids) if target_status_ids.present? + target_statuses.each do |status| + unpush(:home, account, status) + end + end + + def populate_feed(account) + prepopulate_limit = FeedManager::MAX_ITEMS / 4 + statuses = Status.as_home_timeline(account).order(account_id: :desc).limit(prepopulate_limit) + statuses.reverse_each do |status| + next if filter_from_home?(status, account) + add_to_feed(:home, account, status) + end end private @@ -131,4 +155,58 @@ class FeedManager should_filter end + + # Adds a status to an account's feed, returning true if a status was + # added, and false if it was not added to the feed. Note that this is + # an internal helper: callers must call trim or push updates if + # either action is appropriate. + def add_to_feed(timeline_type, account, status) + timeline_key = key(timeline_type, account.id) + reblog_key = key(timeline_type, account.id, 'reblogs') + + if status.reblog? + # If the original status or a reblog of it is within + # REBLOG_FALLOFF statuses from the top, do not re-insert it into + # the feed + rank = redis.zrevrank(timeline_key, status.reblog_of_id) + return false if !rank.nil? && rank < FeedManager::REBLOG_FALLOFF + + reblog_rank = redis.zrevrank(reblog_key, status.reblog_of_id) + return false unless reblog_rank.nil? + + redis.zadd(timeline_key, status.id, status.id) + redis.zadd(reblog_key, status.id, status.reblog_of_id) + else + redis.zadd(timeline_key, status.id, status.id) + end + + true + end + + # Removes an individual status from a feed, correctly handling cases + # with reblogs, and returning true if a status was removed. As with + # `add_to_feed`, this does not trigger push updates, so callers must + # do so if appropriate. + def remove_from_feed(timeline_type, account, status) + timeline_key = key(timeline_type, account.id) + reblog_key = key(timeline_type, account.id, 'reblogs') + + if status.reblog? + # 1. If the reblogging status is not in the feed, stop. + status_rank = redis.zrevrank(timeline_key, status.id) + return false if status_rank.nil? + + # 2. Remove the reblogged status from the `:reblogs` zset. + redis.zrem(reblog_key, status.reblog_of_id) + + # 3. Add the reblogged status to the feed using the reblogging + # status' ID as its score, and the reblogged status' ID as its + # value. + redis.zadd(timeline_key, status.id, status.reblog_of_id) + + # 4. Remove the reblogging status from the feed (as normal) + end + + redis.zrem(timeline_key, status.id) + end end diff --git a/app/models/feed.rb b/app/models/feed.rb index beb4a8de3..5f7b7877a 100644 --- a/app/models/feed.rb +++ b/app/models/feed.rb @@ -19,7 +19,7 @@ class Feed def from_redis(limit, max_id, since_id) max_id = '+inf' if max_id.blank? since_id = '-inf' if since_id.blank? - unhydrated = redis.zrevrangebyscore(key, "(#{max_id}", "(#{since_id}", limit: [0, limit], with_scores: true).map(&:last).map(&:to_i) + unhydrated = redis.zrevrangebyscore(key, "(#{max_id}", "(#{since_id}", limit: [0, limit], with_scores: true).map(&:first).map(&:to_i) Status.where(id: unhydrated).cache_ids end diff --git a/app/services/batched_remove_status_service.rb b/app/services/batched_remove_status_service.rb index 2fd623922..5d83771c9 100644 --- a/app/services/batched_remove_status_service.rb +++ b/app/services/batched_remove_status_service.rb @@ -29,7 +29,7 @@ class BatchedRemoveStatusService < BaseService statuses.group_by(&:account_id).each do |_, account_statuses| account = account_statuses.first.account - unpush_from_home_timelines(account_statuses) + unpush_from_home_timelines(account, account_statuses) if account.local? batch_stream_entries(account, account_statuses) @@ -72,14 +72,15 @@ class BatchedRemoveStatusService < BaseService end end - def unpush_from_home_timelines(statuses) - account = statuses.first.account - recipients = account.followers.local.pluck(:id) + def unpush_from_home_timelines(account, statuses) + recipients = account.followers.local.to_a - recipients << account.id if account.local? + recipients << account if account.local? - recipients.each do |follower_id| - unpush(follower_id, statuses) + recipients.each do |follower| + statuses.each do |status| + FeedManager.instance.unpush(:home, follower, status) + end end end @@ -109,28 +110,6 @@ class BatchedRemoveStatusService < BaseService end end - def unpush(follower_id, statuses) - key = FeedManager.instance.key(:home, follower_id) - - originals = statuses.reject(&:reblog?) - reblogs = statuses.select(&:reblog?) - - # Quickly remove all originals - redis.pipelined do - originals.each do |status| - redis.zremrangebyscore(key, status.id, status.id) - redis.publish("timeline:#{follower_id}", @json_payloads[status.id]) - end - end - - # For reblogs, re-add original status to feed, unless the reblog - # was not in the feed in the first place - reblogs.each do |status| - redis.zadd(key, status.reblog_of_id, status.reblog_of_id) unless redis.zscore(key, status.reblog_of_id).nil? - redis.publish("timeline:#{follower_id}", @json_payloads[status.id]) - end - end - def redis Redis.current end diff --git a/app/services/precompute_feed_service.rb b/app/services/precompute_feed_service.rb index 85635a008..36aabaa00 100644 --- a/app/services/precompute_feed_service.rb +++ b/app/services/precompute_feed_service.rb @@ -1,43 +1,7 @@ # frozen_string_literal: true class PrecomputeFeedService < BaseService - LIMIT = FeedManager::MAX_ITEMS / 4 - def call(account) - @account = account - populate_feed - end - - private - - attr_reader :account - - def populate_feed - pairs = statuses.reverse_each.lazy.reject(&method(:status_filtered?)).map(&method(:process_status)).to_a - - redis.pipelined do - redis.zadd(account_home_key, pairs) if pairs.any? - redis.del("account:#{@account.id}:regeneration") - end - end - - def process_status(status) - [status.id, status.reblog? ? status.reblog_of_id : status.id] - end - - def status_filtered?(status) - FeedManager.instance.filter?(:home, status, account.id) - end - - def account_home_key - FeedManager.instance.key(:home, account.id) - end - - def statuses - Status.as_home_timeline(account).order(account_id: :desc).limit(LIMIT) - end - - def redis - Redis.current + FeedManager.instance.populate_feed(account) end end diff --git a/app/services/remove_status_service.rb b/app/services/remove_status_service.rb index 14f24908c..96d9208cc 100644 --- a/app/services/remove_status_service.rb +++ b/app/services/remove_status_service.rb @@ -102,13 +102,7 @@ class RemoveStatusService < BaseService end def unpush(type, receiver, status) - if status.reblog? && !redis.zscore(FeedManager.instance.key(type, receiver.id), status.reblog_of_id).nil? - redis.zadd(FeedManager.instance.key(type, receiver.id), status.reblog_of_id, status.reblog_of_id) - else - redis.zremrangebyscore(FeedManager.instance.key(type, receiver.id), status.id, status.id) - end - - Redis.current.publish("timeline:#{receiver.id}", @payload) + FeedManager.instance.unpush(type, receiver, status) end def remove_from_hashtags |