Implement StorageTables::Service::MirrorService (#71)

* Implement StorageTables::Service::MirrorService for synchronized file uploads across primary and mirror services

* Implement asynchronous mirroring for uploaded blobs and refactor MirrorService

* Refactor StorageTables module and improve MirrorJob implementation

* Implement StorageTables::Downloader for file downloading and integrity verification

* Add downloader requirement to StorageTables service

* Refactor mirror service tests to use checksum instead of key for downloads and uploads

* Update comments in StorageTables service for clarity and accuracy

* Refactor comments and method signatures in StorageTables service for clarity and consistency

* Remove unnecessary EOF assertion in mirror service tests for cleaner code

* Update comments in mirror service for accuracy and consistency

* Remove commented-out test for URL generation in mirror service tests

Author: bobvanoorschot

app/jobs/storage_tables/mirror_job.rb

@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+require "active_support/core_ext/object/try"
+
+module StorageTables
+  # Provides asynchronous mirroring of directly-uploaded blobs.
+  class MirrorJob < ActiveJob::Base # rubocop:disable Rails/ApplicationJob
+    queue_as { StorageTables.queues[:mirror] }
+
+    discard_on StorageTables::FileNotFoundError
+    retry_on StorageTables::IntegrityError, attempts: 10, wait: :polynomially_longer
+
+    def perform(checksum)
+      StorageTables::Blob.service.try(:mirror, checksum)
+    end
+  end
+end

lib/storage_tables.rb

@@ -26,6 +26,8 @@ module StorageTables
   mattr_accessor :routes_prefix, default: "/rails/storage_tables"
   mattr_accessor :draw_routes, default: true
 
+  mattr_accessor :queues, default: {}
+
   def self.deprecator
     @deprecator ||= ActiveSupport::Deprecation.new("0.2.0", "StorageTables")
   end

lib/storage_tables/service.rb

@@ -10,9 +10,9 @@ class Service
     attr_accessor :name
 
     class << self
-      # Configure an Active Storage service by name from a set of configurations,
-      # typically loaded from a YAML file. The Active Storage engine uses this
-      # to set the global Active Storage service when the app boots.
+      # Configure an Storage Tables service by name from a set of configurations,
+      # typically loaded from a YAML file. The Storage Tables engine uses this
+      # to set the global Storage Tables service when the app boots.
       def configure(service_name, configurations)
         Configurator.build(service_name, configurations)
       end
@@ -23,7 +23,7 @@ def configure(service_name, configurations)
       # Passes the configurator and all of the service's config as keyword args.
       #
       # See MirrorService for an example.
-      def build(name:, **service_config) # :nodoc:
+      def build(configurator:, name:, **service_config) # rubocop:disable Lint/UnusedMethodArgument
         new(**service_config).tap do |service_instance|
           service_instance.name = name
         end

lib/storage_tables/service/configurator.rb

@@ -20,7 +20,7 @@ def initialize(configurations)
       def build(service_name)
         config = config_for(service_name.to_sym)
         resolve(config.fetch(:service)).build(
-          **config.except(:service), name: service_name
+          **config.except(:service), name: service_name, configurator: self
         )
       end
 

lib/storage_tables/service/disk_service.rb

@@ -11,7 +11,7 @@ class Service
     class DiskService < Service
       attr_accessor :root
 
-      def initialize(root:, public: false)
+      def initialize(root:, public: false, **)
         @root = root
         @public = public
 

lib/storage_tables/service/mirror_service.rb

@@ -0,0 +1,92 @@
+# frozen_string_literal: true
+
+require "active_support/core_ext/module/delegation"
+
+module StorageTables
+  class Service
+    # = Storage Tables Mirror \Service
+    #
+    # Wraps a set of mirror services and provides a single StorageTables::Service object that will all
+    # have the files uploaded to them. A +primary+ service is designated to answer calls to:
+    # * +download+
+    # * +exists?+
+    # * +url+
+    # * +url_for_direct_upload+
+    # * +headers_for_direct_upload+
+    class MirrorService < Service
+      attr_reader :primary, :mirrors
+
+      delegate :download, :download_chunk, :exist?, :url,
+               :url_for_direct_upload, :headers_for_direct_upload, :path_for, :compose, to: :primary
+
+      # Stitch together from named services.
+      def self.build(primary:, mirrors:, name:, configurator:, **) # :nodoc:
+        new(
+          primary: configurator.build(primary),
+          mirrors: mirrors.collect { |mirror_name| configurator.build mirror_name }
+        ).tap do |service_instance|
+          service_instance.name = name
+        end
+      end
+
+      def initialize(primary:, mirrors:) # rubocop:disable Lint/MissingSuper
+        @primary = primary
+        @mirrors = mirrors
+        @executor = Concurrent::ThreadPoolExecutor.new(
+          min_threads: 1,
+          max_threads: mirrors.size,
+          max_queue: 0,
+          fallback_policy: :caller_runs,
+          idle_time: 60
+        )
+      end
+
+      # Upload the +io+ to the +checksum+ specified to all services. The upload to the primary service is done
+      # synchronously whereas the upload to the mirrors is done asynchronously. If a +checksum+ is provided, all
+      # services will ensure a match when the upload has completed or raise an StorageTables::IntegrityError.
+      def upload(checksum, io, **)
+        io.rewind
+        primary.upload(checksum, io, **)
+        mirror_later checksum
+      end
+
+      # Delete the file at the +checksum+ on all services.
+      def delete(checksum)
+        perform_across_services :delete, checksum
+      end
+
+      def mirror_later(checksum)
+        StorageTables::MirrorJob.perform_later checksum
+      end
+
+      # Copy the file at the +checksum+ from the primary service to each of the mirrors where it doesn't already exist.
+      def mirror(checksum)
+        instrument(:mirror, checksum) do
+          if (mirrors_in_need_of_mirroring = mirrors.reject { |service| service.exist?(checksum) }).any?
+            primary.open(checksum) do |io|
+              mirrors_in_need_of_mirroring.each do |service|
+                io.rewind
+                service.upload checksum, io
+              end
+            end
+          end
+        end
+      end
+
+      private
+
+      def each_service(&)
+        [primary, *mirrors].each(&)
+      end
+
+      def perform_across_services(method, *args)
+        tasks = each_service.collect do |service|
+          Concurrent::Promise.execute(executor: @executor) do
+            service.public_send method, *args
+          end
+        end
+        tasks.each(&:value!)
+      end
+    end
+  end
+end

test/service/mirror_service_test.rb

@@ -0,0 +1,85 @@
+# frozen_string_literal: true
+
+require "service/shared_service_tests"
+
+module StorageTables
+  class Service
+    class MirrorServiceTest < ActiveSupport::TestCase
+      mirror_config = (1..3).to_h do |i|
+        ["mirror_#{i}",
+         { service: "Disk",
+           root: Dir.mktmpdir("active_storage_tests_mirror_#{i}") }]
+      end
+
+      config = mirror_config.merge \
+        mirror: { service: "Mirror", primary: "primary", mirrors: mirror_config.keys },
+        primary: { service: "Disk", root: Dir.mktmpdir("active_storage_tests_primary") }
+
+      SERVICE = StorageTables::Service.configure :mirror, config
+
+      include StorageTables::Service::SharedServiceTests
+      include ActiveJob::TestHelper
+
+      test "name" do
+        assert_equal :mirror, @service.name
+      end
+
+      test "uploading to all services" do
+        old_service = StorageTables::Blob.service
+        StorageTables::Blob.service = @service
+
+        data     = "Something else entirely!"
+        io       = StringIO.new(data)
+        checksum = generate_checksum(data)
+
+        assert_performed_jobs 1, only: StorageTables::MirrorJob do
+          @service.upload checksum, io.tap(&:read)
+        end
+
+        assert_equal data, @service.primary.download(checksum)
+        @service.mirrors.each do |mirror|
+          assert_equal data, mirror.download(checksum)
+        end
+      ensure
+        @service.delete checksum
+        StorageTables::Blob.service = old_service
+      end
+
+      test "downloading from primary service" do
+        data     = "Something else entirely!"
+        checksum = generate_checksum(data)
+
+        @service.primary.upload(checksum, StringIO.new(data))
+
+        assert_equal data, @service.download(checksum)
+      end
+
+      test "deleting from all services" do
+        @service.delete @checksum
+
+        assert_not SERVICE.primary.exist?(@checksum)
+        SERVICE.mirrors.each do |mirror|
+          assert_not mirror.exist?(@checksum)
+        end
+      end
+
+      test "mirroring a file from the primary service to secondary services where it doesn't exist" do
+        data     = "Something else entirely!"
+        checksum = generate_checksum(data)
+
+        @service.primary.upload(checksum, StringIO.new(data))
+        @service.mirrors.third.upload checksum, StringIO.new(data)
+
+        @service.mirror(checksum)
+
+        assert_equal data, @service.mirrors.first.download(checksum)
+        assert_equal data, @service.mirrors.second.download(checksum)
+        assert_equal data, @service.mirrors.third.download(checksum)
+      end
+
+      test "path for file in primary service" do
+        assert_equal @service.primary.path_for(@checksum), @service.path_for(@checksum)
+      end
+    end
+  end
+end