Refactors replication.

Author: justinlittman

app/jobs/audit/replication_audit_job.rb

@@ -0,0 +1,32 @@
+# frozen_string_literal: true
+
+module Audit
+  # Job to audit replication for a PreservedObject
+  class ReplicationAuditJob < ApplicationJob
+    queue_as :moab_replication_audit
+    delegate :logger, to: Audit::ReplicationSupport
+
+    include UniqueJob
+
+    def perform(preserved_object)
+      @preserved_object = preserved_object
+
+      preserved_object.populate_zipped_moab_versions!
+
+      ::Replication::AuditService.call(preserved_object: preserved_object).each do |audit_results|
+        AuditResultsReporter.report_results(audit_results: audit_results, logger: logger)
+      end
+
+      preserved_object.update!(last_archive_audit: Time.current)
+
+      start_replication
+    end
+
+    attr_reader :preserved_object
+
+    def start_replication
+      return unless preserved_object.zipped_moab_versions.created.exists? || preserved_object.zipped_moab_versions.incomplete.exists?
+      ::ReplicationJob.perform_later(preserved_object)
+    end
+  end
+end

app/jobs/replication_job.rb

@@ -0,0 +1,17 @@
+# frozen_string_literal: true
+
+# Job to replicate a PreservedObject to cloud endpoints
+class ReplicationJob < ApplicationJob
+  queue_as :replication
+  include UniqueJob
+
+  def perform(preserved_object)
+    preserved_object.populate_zipped_moab_versions!
+
+    ::Replication::AuditService.call(preserved_object: preserved_object)
+
+    (1..preserved_object.current_version).each do |version|
+      Replication::ReplicateVersionService.call(preserved_object:, version:)
+    end
+  end
+end

app/models/moab_record.rb

@@ -14,11 +14,6 @@ class MoabRecord < ApplicationRecord
     'validity_unknown' => 6
   }
 
-  after_create :create_zipped_moab_versions!
-  # hook for creating archive zips is here and on PreservedObject, because version and current_version must be in sync, and
-  # even though both fields will usually be updated together in a single transaction, one has to be updated first.  latter
-  # of the two updates will actually trigger replication.
-  after_update :create_zipped_moab_versions!, if: :saved_change_to_version? # an ActiveRecord dynamic method
   after_save :validate_checksums!, if: proc { |moab_record| moab_record.saved_change_to_status? && moab_record.validity_unknown? }
 
   # NOTE: Since Rails 5.0, belongs_to adds the presence validator automatically, and explicit presence validation

app/models/preserved_object.rb

@@ -8,11 +8,6 @@
 class PreservedObject < ApplicationRecord
   PREFIX_RE = /druid:/i
 
-  # hook for creating archive zips is here and on MoabRecord, because version and current_version must be in sync, and
-  # even though both fields will usually be updated together in a single transaction, one has to be updated first.  latter
-  # of the two updates will actually trigger replication.
-  after_update :create_zipped_moab_versions!, if: :saved_change_to_current_version? # an ActiveRecord dynamic method
-
   has_one :moab_record, dependent: :restrict_with_exception, autosave: true
   has_many :zipped_moab_versions, dependent: :restrict_with_exception, inverse_of: :preserved_object
 
@@ -51,14 +46,27 @@ def create_zipped_moab_versions!
     end
   end
 
+  # Creates ZippedMoabVersion for each version for each ZipEndpoint for which a ZippedMoabVersion does not already exist.
+  # @return [Array<ZippedMoabVersion>] the ZippedMoabVersions that were created
+  def populate_zipped_moab_versions!
+    [].tap do |new_zipped_moab_versions|
+      (1..current_version).each do |version|
+        ZipEndpoint.find_each do |zip_endpoint|
+          next if zipped_moab_versions.exists?(version: version, zip_endpoint: zip_endpoint)
+          new_zipped_moab_versions << zipped_moab_versions.create!(version: version, zip_endpoint: zip_endpoint)
+        end
+      end
+    end
+  end
+
   def as_json(*)
     super.except('id')
   end
 
   # Queue a job that will check to see whether this PreservedObject has been
   # fully replicated to all target ZipEndpoints
   def audit_moab_version_replication!
-    Audit::MoabReplicationAuditJob.perform_later(self)
+    Audit::ReplicationAuditJob.perform_later(self)
   end
 
   def total_size_of_moab_version(version)

app/models/replication/druid_version_zip.rb

@@ -36,14 +36,28 @@ def base_key
     # create it.
     # @raise [StandardError] if there's a zip file for this druid-version, but it looks too small to be complete.
     def find_or_create_zip!
-      if File.exist?(file_path)
+      if exist?
         raise "zip already exists, but size (#{total_part_size}) is smaller than the moab version size (#{moab_version_size})!" unless zip_size_ok?
         FileUtils.touch(file_path)
       else
         create_zip!
       end
     end
 
+    def exist?
+      File.exist?(file_path)
+    end
+
+    # @return [Boolean] true if there is a match between the zip part files and their md5 sidecar files
+    def complete?
+      # There is at least one part file
+      return false if part_paths.empty?
+      # The set of md5 sidecar files matches the set of part files
+      return false unless part_keys.to_set == part_keys_from_md5_sidecars.to_set
+      # Check each md5 sidecar file against the zip part file
+      druid_version_zip_parts.all?(&:md5_match?)
+    end
+
     # Creates a zip of Druid-Version content.
     # Changes directory so that the storage root (and druid tree) are not part of
     # the archival directory structure, just the object, e.g. starting at 'ab123cd4567/...' directory,
@@ -54,7 +68,7 @@ def create_zip!
       combined, status = Open3.capture2e(zip_command, chdir: work_dir.to_s)
       raise "zipmaker failure #{combined}" unless status.success?
       unless zip_size_ok?
-        part_cleanup_errors = cleanup_zip_parts!
+        part_cleanup_errors = cleanup_zip_parts_with_rescue!
         part_cleanup_err_msg = "\n-- errors cleaning up zip parts: #{part_cleanup_errors.map(&:inspect)}" if part_cleanup_errors.present?
         raise "zip size (#{total_part_size}) is smaller than the moab version size (#{moab_version_size})! zipmaker failure #{combined}#{part_cleanup_err_msg}"
       end
@@ -172,6 +186,18 @@ def moab_version_size
       moab_version_files.sum { |f| File.size(f) }
     end
 
+    # Deletes all zip part files and their md5 sidecar files from local zip storage
+    def cleanup_zip_parts!
+      parts_and_checksums_paths.each { |filepath| File.delete(filepath) }
+    end
+
+    # @return [Array<DruidVersionZipPart>] all parts for this DruidVersionZip
+    def druid_version_zip_parts
+      part_keys.map do |part_key|
+        Replication::DruidVersionZipPart.new(self, part_key)
+      end
+    end
+
     private
 
     # Throws an error if any of the files in the moab are not yet readable.  For example due to
@@ -184,7 +210,7 @@ def check_moab_version_readability!
       moab_version_files.map { |f| File.stat(f) }
     end
 
-    def cleanup_zip_parts!
+    def cleanup_zip_parts_with_rescue!
       errors = []
       parts_and_checksums_paths.map do |p|
         File.delete(p)
@@ -219,5 +245,14 @@ def fetch_zip_version
     def zip_version_regexp
       /This is (Zip \d+(\.\d)+\s*(\(.*\d{4}\))?)/
     end
+
+    # @return [Array<String>] relative paths, i.e. s3_part_keys for existing parts based on the md5 sidecar files
+    def part_keys_from_md5_sidecars
+      md5_sidecar_paths.map { |md5_path| md5_path.relative_path_from(zip_storage).to_s.delete_suffix('.md5') }
+    end
+
+    def md5_sidecar_paths
+      Pathname.glob(File.join(zip_storage, s3_key('.*.md5')))
+    end
   end
 end

app/models/replication/druid_version_zip_part.rb

@@ -73,5 +73,10 @@ def write_md5
     def read_md5
       File.read(md5_path)
     end
+
+    # @return [Boolean] whether the md5 from the md5 sidecar file matches the computed md5
+    def md5_match?
+      read_md5 == hexdigest
+    end
   end
 end

app/models/zip_endpoint.rb

@@ -63,6 +63,11 @@ def audit_class
     raise "Failed to return audit class based on setting for #{endpoint_name}.  Check setting string for accuracy."
   end
 
+  # @return [Aws::S3::Bucket] S3 bucket object for this zip endpoint
+  def bucket
+    @bucket ||= Replication::ProviderFactory.create(zip_endpoint: self).bucket
+  end
+
   def to_s
     endpoint_name
   end

app/models/zip_part.rb

@@ -5,7 +5,7 @@
 # This model's data is populated by Replication::DeliveryDispatcherJob.
 class ZipPart < ApplicationRecord
   belongs_to :zipped_moab_version, inverse_of: :zip_parts
-  delegate :zip_endpoint, :preserved_object, to: :zipped_moab_version
+  delegate :zip_endpoint, :preserved_object, :druid_version_zip, to: :zipped_moab_version
 
   # @note Hash values cannot be modified without migrating any associated persisted data.
   # @see [enum docs] http://api.rubyonrails.org/classes/ActiveRecord/Enum.html
@@ -37,11 +37,15 @@ def suffixes_in_set
     druid_version_zip.expected_part_keys(parts_count).map { |key| File.extname(key) }
   end
 
-  def druid_version_zip
-    @druid_version_zip ||= Replication::DruidVersionZip.new(preserved_object.druid, zipped_moab_version.version)
+  def druid_version_zip_part
+    @druid_version_zip_part ||= Replication::DruidVersionZipPart.new(druid_version_zip, s3_key)
   end
 
   def s3_key
     druid_version_zip.s3_key(suffix)
   end
+
+  def s3_part
+    @s3_part ||= zip_endpoint.bucket.object(s3_key)
+  end
 end

app/models/zipped_moab_version.rb

@@ -14,10 +14,22 @@ class ZippedMoabVersion < ApplicationRecord
 
   validates :version, presence: true
 
+  # If zip_parts_count is present, must be greater than zero
+  validates :zip_parts_count, numericality: { only_integer: true, greater_than: 0 }, allow_nil: true
+
+  enum :status, {
+    'ok' => 0,
+    'incomplete' => 1,
+    'created' => 2, # DB-level default
+    'failed' => 3
+  }
+
   scope :by_druid, lambda { |druid|
     joins(:preserved_object).where(preserved_objects: { druid: druid })
   }
 
+  before_save :update_status_updated_at
+
   # ideally, there should be only one distinct parts_count value among a set of sibling
   # zip_parts.  if there's variation in the count, that implies the zip was remade, and that
   # the part count differed between the zip invocations (which may imply a zip implementation
@@ -41,4 +53,15 @@ def all_parts_replicated?
   def total_part_size
     zip_parts.sum(&:size)
   end
+
+  def update_status_updated_at
+    self.status_updated_at = Time.current if status_changed?
+  end
+
+  def druid_version_zip
+    @druid_version_zip ||= Replication::DruidVersionZip.new(
+      # In tests, a PreservedObject may not have a MoabRecord, hence the safe navigation.
+      preserved_object.druid, version, preserved_object&.moab_record&.moab_storage_root&.storage_location # rubocop:disable Style/SafeNavigationChainLength
+    )
+  end
 end

app/services/audit/replication_support.rb

@@ -41,7 +41,7 @@ def self.check_child_zip_part_attributes(zmv, results)
       if unreplicated_parts.any?
         results.add_result(
           Audit::Results::ZIP_PARTS_NOT_ALL_REPLICATED,
-          base_hash.merge(unreplicated_parts_list: unreplicated_parts.to_a)
+          base_hash
         )
       end