Unique list with sorted set (#114)

* Back UniqueList with a Redis Sorted Set

A sorted set is optimal for this type of functionality because we can
use fewer, more-performant Redis calls.

* Wrap unique list insert and trim in a transaction

* Fallback / migration mechanism for existing users of UniqueLists

In order to ease the transition from UniqueList being backed by
a Redis list, we can fallback to the legacy implementation for read
operations. For write operations we first migrate to a sorted set,
then retry.

* Update the Readme with sorted set backed unique list commands

* Test remaining methods with migration fallbacks

* Be more specific about which errors to raise

* Improve the fallback / migration system

* Improve unique list scoring to avoid collisions

Use the Redis server time as a basis. Then add the current process time.
Then add to that the index of the inserted elements as a microsecond
component.

* Update Readme with realistic Unique List scores

* Implement OrderedSet

Like a UniqueList, but backed by a Redis sorted set rather than a list

* Resolve pipeline deprecation warnings

* Document OrderedSet

* Remove UniqueListLegacy

* Revert fallback options

* Revert change to type_from

* Enforce limit is not negative

* Revert config proxying

* Fix bug in score ge
neration

* Support latest multi pipeline changes

Author: lewispb

README.md

@@ -62,6 +62,14 @@ unique_list << "5"                             # => LREM myuniquelist 0, "5" + R
 unique_list.remove(3)                          # => LREM myuniquelist 0, "3"
 [ "4", "2", "1", "5" ] == unique_list.elements # => LRANGE myuniquelist 0, -1
 
+ordered_set = Kredis.ordered_set "myorderedset"
+ordered_set.append(%w[ 2 3 4 ])                # => ZADD myorderedset 1646131025.4953232 2 1646131025.495326 3 1646131025.4953272 4
+ordered_set.prepend(%w[ 1 2 3 4 ])             # => ZADD myorderedset -1646131025.4957051 1 -1646131025.495707 2 -1646131025.4957082 3 -1646131025.4957092 4
+ordered_set.append([])
+ordered_set << "5"                             # => ZADD myorderedset 1646131025.4960442 5
+ordered_set.remove(3)                          # => ZREM myorderedset 3
+[ "4", "2", "1", "5" ] == ordered_set.elements # => ZRANGE myorderedset 0 -1
+
 set = Kredis.set "myset", typed: :datetime
 set.add(DateTime.tomorrow, DateTime.yesterday)           # => SADD myset "2021-02-03 00:00:00 +0100" "2021-02-01 00:00:00 +0100"
 set << DateTime.tomorrow                                 # => SADD myset "2021-02-03 00:00:00 +0100"

lib/kredis/types.rb

@@ -71,6 +71,10 @@ def set(key, typed: :string, config: :shared, after_change: nil)
     type_from(Set, config, key, after_change: after_change, typed: typed)
   end
 
+  def ordered_set(key, typed: :string, limit: nil, config: :shared, after_change: nil)
+    type_from(OrderedSet, config, key, after_change: after_change, typed: typed, limit: limit)
+  end
+
   def slot(key, config: :shared, after_change: nil)
     type_from(Slots, config, key, after_change: after_change, available: 1)
   end
@@ -99,4 +103,5 @@ def type_from(type_klass, config, key, after_change: nil, **options)
 require "kredis/types/list"
 require "kredis/types/unique_list"
 require "kredis/types/set"
+require "kredis/types/ordered_set"
 require "kredis/types/slots"

lib/kredis/types/ordered_set.rb

@@ -0,0 +1,71 @@
+class Kredis::Types::OrderedSet < Kredis::Types::Proxying
+  proxying :multi, :zrange, :zrem, :zadd, :zremrangebyrank, :zcard, :exists?, :del
+
+  attr_accessor :typed
+  attr_reader :limit
+
+  def elements
+    strings_to_types(zrange(0, -1) || [], typed)
+  end
+  alias to_a elements
+
+  def remove(*elements)
+    zrem(types_to_strings(elements, typed))
+  end
+
+  def prepend(elements)
+    insert(elements, prepending: true)
+  end
+
+  def append(elements)
+    insert(elements)
+  end
+  alias << append
+
+  def limit=(limit)
+    raise "Limit must be greater than 0" if limit && limit <= 0
+
+    @limit = limit
+  end
+
+  private
+    def insert(elements, prepending: false)
+      elements = Array(elements)
+      return if elements.empty?
+
+      elements_with_scores = types_to_strings(elements, typed).map.with_index do |element, index|
+        score = generate_base_score(negative: prepending) + (index / 100000)
+
+        [ score , element ]
+      end
+
+      multi do
+        zadd(elements_with_scores)
+        trim(from_beginning: prepending)
+      end
+    end
+
+    def generate_base_score(negative:)
+      current_time = process_start_time + process_uptime
+
+      negative ? -current_time : current_time
+    end
+
+    def process_start_time
+      @process_start_time ||= unproxied_redis.time.join(".").to_f - process_uptime
+    end
+
+    def process_uptime
+      Process.clock_gettime(Process::CLOCK_MONOTONIC)
+    end
+
+    def trim(from_beginning:)
+      return unless limit
+
+      if from_beginning
+        zremrangebyrank(limit, -1)
+      else
+        zremrangebyrank(0, -(limit + 1))
+      end
+    end
+end

lib/kredis/types/proxying.rb

@@ -8,6 +8,7 @@ def self.proxying(*commands)
   end
 
   def initialize(redis, key, **options)
+    @redis = redis
     @key = key
     @proxy = Kredis::Types::Proxy.new(redis, key)
     options.each { |key, value| send("#{key}=", value) }
@@ -17,6 +18,12 @@ def failsafe(returning: nil, &block)
     proxy.suppress_failsafe_with(returning: returning, &block)
   end
 
+  def unproxied_redis
+    # Generally, this should not be used. It's only here for the rare case where we need to
+    # call Redis commands that don't reference a key and don't want to be pipelined.
+    @redis
+  end
+
   private
     delegate :type_to_string, :string_to_type, :types_to_strings, :strings_to_types, to: :Kredis
 end

test/types/ordered_set_test.rb

@@ -0,0 +1,99 @@
+require "test_helper"
+
+class OrderedSetTest < ActiveSupport::TestCase
+  setup { @set = Kredis.ordered_set "ordered-set", limit: 5 }
+
+  test "append" do
+    @set.append(%w[ 1 2 3 ])
+    @set.append(%w[ 1 2 3 4 ])
+    assert_equal %w[ 1 2 3 4 ], @set.elements
+
+    @set << "5"
+    assert_equal %w[ 1 2 3 4 5 ], @set.elements
+  end
+
+  test "appending the same element re-appends it" do
+    @set.append(%w[ 1 2 3 ])
+    @set.append(%w[ 2 ])
+    assert_equal %w[ 1 3 2 ], @set.elements
+  end
+
+  test "mass append maintains ordering" do
+    @set = Kredis.ordered_set "ordered-set" # no limit
+
+    thousand_elements = 1000.times.map { [*"A".."Z"].sample(10).join }
+    @set.append(thousand_elements)
+    assert_equal thousand_elements, @set.elements
+
+    thousand_elements.each { |element| @set.append(element) }
+    assert_equal thousand_elements, @set.elements
+  end
+
+  test "prepend" do
+    @set.prepend(%w[ 1 2 3 ])
+    @set.prepend(%w[ 1 2 3 4 ])
+    assert_equal %w[ 4 3 2 1 ], @set.elements
+  end
+
+  test "append nothing" do
+    @set.append(%w[ 1 2 3 ])
+    @set.append([])
+    assert_equal %w[ 1 2 3 ], @set.elements
+  end
+
+  test "prepend nothing" do
+    @set.prepend(%w[ 1 2 3 ])
+    @set.prepend([])
+    assert_equal %w[ 3 2 1 ], @set.elements
+  end
+
+  test "typed as integers" do
+    @set = Kredis.ordered_set "mylist", typed: :integer
+
+    @set.append [ 1, 2 ]
+    @set << 2
+    assert_equal [ 1, 2 ], @set.elements
+
+    @set.remove(2)
+    assert_equal [ 1 ], @set.elements
+
+    @set.append [ "1-a", 2 ]
+
+    assert_equal [ 1, 2 ], @set.elements
+  end
+
+  test "exists?" do
+    assert_not @set.exists?
+
+    @set.append [ 1, 2 ]
+    assert @set.exists?
+  end
+
+  test "appending over limit" do
+    @set.append(%w[ 1 2 3 4 5 ])
+    @set.append(%w[ 6 7 8 ])
+    assert_equal %w[ 4 5 6 7 8 ], @set.elements
+  end
+
+  test "prepending over limit" do
+    @set.prepend(%w[ 1 2 3 4 5 ])
+    @set.prepend(%w[ 6 7 8 ])
+    assert_equal %w[ 8 7 6 5 4 ], @set.elements
+  end
+
+  test "appending array with duplicates" do
+    @set.append(%w[ 1 1 1 ])
+    assert_equal %w[ 1 ], @set.elements
+  end
+
+  test "prepending array with duplicates" do
+    @set.prepend(%w[ 1 1 1 ])
+    assert_equal %w[ 1 ], @set.elements
+  end
+
+  test "limit can't be 0 or less" do
+    assert_raises do
+      Kredis.ordered_set "ordered-set", limit: -1
+    end
+  end
+end