Non-Serial ("Snowflake") IDs

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
 

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 @@ def filter?(timeline_type, status, receiver_id)
   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 @@ def merge_into_timeline(from_account, into_account)
       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,21 +85,29 @@ def unmerge_from_timeline(from_account, into_account)
     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
 
   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 @@ def filter_from_mentions?(status, receiver_id)
 
     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

app/models/feed.rb

@@ -19,7 +19,7 @@ def get(limit, max_id = nil, since_id = nil)
   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
 

app/services/batched_remove_status_service.rb

@@ -29,7 +29,7 @@ def call(statuses)
     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 @@ def batch_activity_json(account, statuses)
     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 @@ def batch_salmon_slaps(status)
     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

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

app/services/remove_status_service.rb

@@ -102,13 +102,7 @@ def remove_reblogs
   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

db/migrate/20170920024819_status_ids_to_timestamp_ids.rb

@@ -0,0 +1,36 @@
+class StatusIdsToTimestampIds < ActiveRecord::Migration[5.1]
+  def up
+    # Prepare the function we will use to generate IDs.
+    Rake::Task['db:define_timestamp_id'].execute
+
+    # Set up the statuses.id column to use our timestamp-based IDs.
+    ActiveRecord::Base.connection.execute(
+      <<~SQL
+        ALTER TABLE statuses
+        ALTER COLUMN id
+        SET DEFAULT timestamp_id('statuses')
+      SQL
+    )
+
+    # Make sure we have a sequence to use.
+    Rake::Task['db:ensure_id_sequences_exist'].execute
+  end
+
+  def down
+    # Revert the column to the old method of just using the sequence
+    # value for new IDs. Set the current ID sequence to the maximum
+    # existing ID, such that the next sequence will be one higher.
+
+    # We lock the table during this so that the ID won't get clobbered,
+    # but ID is indexed, so this should be a fast operation.
+    ActiveRecord::Base.connection.execute(
+      <<~SQL
+        LOCK statuses;
+        SELECT setval('statuses_id_seq', (SELECT MAX(id) FROM statuses));
+        ALTER TABLE statuses
+          ALTER COLUMN id
+          SET DEFAULT nextval('statuses_id_seq');"
+      SQL
+    )
+  end
+end