#434

backend/app/services/counts_export_download.rb

@@ -13,15 +13,17 @@ class CountsExportDownload
   class TimedOut < StandardError; end
   class StorageFull < StandardError; end
 
-  def self.directory = Rails.root.join('tmp', 'counts_exports')
+  def self.directory
+    Rails.root.join(ENV.fetch('COUNTS_EXPORT_DIRECTORY', 'tmp/counts_exports'))
+  end
   def self.verifier = Rails.application.message_verifier('counts_export_download')
 
-  def self.prepare(exporter, owner)
+  def self.prepare(exporter, owner, progress: nil, check: nil, identifier: nil)
     FileUtils.mkdir_p(directory, mode: 0o700)
     slot = acquire_slot
 
     begin
-      generate(exporter, owner)
+      generate(exporter, owner, progress:, check:, identifier:)
     ensure
       slot.close
     end
@@ -38,7 +40,7 @@ class CountsExportDownload
     raise Busy
   end
 
-  def self.generate(exporter, owner)
+  def self.generate(exporter, owner, progress: nil, check: nil, identifier: nil)
     FileUtils.mkdir_p(directory, mode: 0o700)
     Dir.glob(directory.join('counts-*.export')).each do |path|
       File.unlink(path) if File.mtime(path) < Time.current - EXPIRY
@@ -53,14 +55,16 @@ class CountsExportDownload
     reserved = MAX_CONCURRENT_EXPORTS * MAX_FILE_BYTES
     raise StorageFull if stored + reserved > MAX_STORED_BYTES
 
-    file = Tempfile.create(['counts-', '.part'], directory)
+    prefix = identifier ? "counts-#{ identifier }-" : 'counts-'
+    file = Tempfile.create([prefix, '.part'], directory)
     complete = false
     bytes = 0
     deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + MAX_EXPORT_SECONDS
     path = file.path
     begin
       file.binmode
-      exporter.body.each do |chunk|
+      exporter.body(progress:).each do |chunk|
+        check&.call
         raise TimedOut if Process.clock_gettime(Process::CLOCK_MONOTONIC) > deadline
 
         bytes += chunk.bytesize
@@ -68,6 +72,8 @@ class CountsExportDownload
 
         file.write(chunk)
       end
+      file.flush
+      file.fsync
       file.close
       finished_path = path.sub(/\.part\z/, '.export')
       File.rename(path, finished_path)

backend/app/services/counts_export_jobs.rb

@@ -0,0 +1,236 @@
+require 'json'
+require 'fileutils'
+
+# A durable, bounded export spool. Web and supervised workers share the same POSIX volume.
+class CountsExportJobs
+  class Cancelled < StandardError; end
+  ACTIVE = ['queued', 'running', 'cancelling'].freeze
+
+  def self.directory = CountsExportDownload.directory.join('jobs')
+  def self.verifier = Rails.application.message_verifier('counts_export_job')
+
+  def self.synchronise
+    FileUtils.mkdir_p(directory, mode: 0o700)
+    File.open(directory.join('.registry.lock'), File::RDWR | File::CREAT, 0o600) do |lock|
+      lock.flock(File::LOCK_EX)
+      yield
+    end
+  end
+
+  def self.write(state)
+    path = directory.join("#{ state.fetch('id') }.json")
+    temporary = "#{ path }.#{ SecureRandom.hex(8) }.tmp"
+    File.open(temporary, 'w', 0o600) do |file|
+      file.write(JSON.generate(state))
+      file.flush
+      file.fsync
+    end
+    File.rename(temporary, path)
+  ensure
+    File.unlink(temporary) if temporary && File.exist?(temporary)
+  end
+
+  def self.read(id)
+    JSON.parse(File.read(directory.join("#{ id }.json")))
+  rescue Errno::ENOENT
+    nil
+  end
+
+  def self.remove_download(state)
+    name = state['download_file']
+    return unless name&.match?(/\Acounts-[a-zA-Z0-9_-]+\.export\z/)
+
+    File.unlink(CountsExportDownload.directory.join(name))
+  rescue Errno::ENOENT
+    nil
+  end
+
+  def self.remove_partial(id)
+    Dir.glob(CountsExportDownload.directory.join("counts-#{ id }-*")).each do |path|
+      File.unlink(path)
+    rescue Errno::ENOENT
+      nil
+    end
+  end
+
+  def self.clean
+    Dir.glob(directory.join('*.tmp')).each do |path|
+      File.unlink(path) if File.mtime(path) < Time.current - CountsExportDownload::EXPIRY
+    end
+    Dir.glob(directory.join('*.run')).each do |path|
+      id = File.basename(path, '.run')
+      next if directory.join("#{ id }.json").exist?
+
+      File.open(path, File::RDWR) do |lock|
+        File.unlink(path) if lock.flock(File::LOCK_EX | File::LOCK_NB)
+      end
+    end
+    Dir.glob(directory.join('*.json')).each do |path|
+      state = JSON.parse(File.read(path))
+      next if state.fetch('expires_at') > Time.current.to_f
+
+      run = directory.join("#{ state.fetch('id') }.run")
+      File.open(run, File::RDWR | File::CREAT, 0o600) do |lock|
+        unless lock.flock(File::LOCK_EX | File::LOCK_NB)
+          state['status'] = 'cancelling'
+          write(state)
+          next
+        end
+        remove_download(state)
+        remove_partial(state.fetch('id'))
+        File.unlink(path)
+        File.unlink(run)
+      end
+    end
+  end
+
+  def self.start(payload, owner, language)
+    payload = payload.dup
+    time_key = payload.key?('definition') ? 'execution_at' : 'at'
+    if payload[time_key].nil? || payload[time_key] == ''
+      payload[time_key] = Time.current.utc.iso8601(6)
+    end
+    exporter = CountsExporter.new(payload, language_code: language)
+    synchronise do
+      clean
+      active = Dir.glob(directory.join('*.json')).count do |path|
+        ACTIVE.include?(JSON.parse(File.read(path))['status'])
+      end
+      raise CountsExportDownload::Busy if active >= CountsExportDownload::MAX_CONCURRENT_EXPORTS
+
+      id = SecureRandom.hex(24)
+      state = { 'id' => id, 'owner' => owner, 'payload' => payload,
+          'language' => language, 'resource' => exporter.resource,
+          'format' => exporter.format, 'status' => 'queued',
+          'processed' => 0, 'total' => nil,
+          'expires_at' => (Time.current + CountsExportDownload::EXPIRY).to_f }
+      write(state)
+      verifier.generate({ 'id' => id, 'owner' => owner },
+          expires_in: CountsExportDownload::EXPIRY, purpose: 'counts_export_job')
+    end
+  end
+
+  def self.authorised_id(token, owner)
+    value = verifier.verified(token, purpose: 'counts_export_job')
+    return unless value.is_a?(Hash) && value['owner'] == owner
+    return unless value['id']&.match?(/\A[a-f0-9]{48}\z/)
+
+    value['id']
+  end
+
+  def self.status(token, owner, cancel: false)
+    id = authorised_id(token, owner)
+    return unless id
+
+    synchronise do
+      state = read(id)
+      return unless state && state['owner'] == owner && state['expires_at'] > Time.current.to_f
+
+      if cancel && ['queued', 'running', 'ready'].include?(state['status'])
+        state['status'] = state['status'] == 'running' ? 'cancelling' : 'cancelled'
+        remove_download(state)
+        state.delete('token')
+        write(state)
+      end
+      state.slice('status', 'processed', 'total', 'format', 'token', 'error')
+    end
+  end
+
+  def self.checkpoint(id, processed = nil, total = nil)
+    synchronise do
+      state = read(id)
+      unless state && state['status'] == 'running' && state['expires_at'] > Time.current.to_f
+        raise Cancelled
+      end
+      if processed
+        state.merge!('processed' => processed, 'total' => total)
+        write(state)
+      end
+    end
+  end
+
+  def self.work_one
+    FileUtils.mkdir_p(directory, mode: 0o700)
+    Dir.glob(directory.join('*.json')).sort.each do |path|
+      id = File.basename(path, '.json')
+      File.open(directory.join("#{ id }.run"), File::RDWR | File::CREAT, 0o600) do |lock|
+        next unless lock.flock(File::LOCK_EX | File::LOCK_NB)
+
+        state = synchronise do
+          candidate = read(id)
+          next unless candidate && ACTIVE.include?(candidate['status'])
+
+          if candidate['status'] != 'queued'
+            candidate['status'] = candidate['status'] == 'cancelling' ? 'cancelled' : 'failed'
+            remove_partial(id)
+            candidate['error'] = 'interrupted'
+            write(candidate)
+            next
+          end
+          candidate['status'] = 'running'
+          write(candidate)
+          candidate
+        end
+        next unless state
+
+        perform(state)
+        return true
+      end
+    end
+    synchronise { clean }
+    false
+  end
+
+  def self.perform(state)
+    id = state.fetch('id')
+    token = nil
+    begin
+      checkpoint(id)
+      exporter = CountsExporter.new(state.fetch('payload'), language_code: state.fetch('language'))
+      last = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+      progress = lambda do |processed, total|
+        now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        if processed.zero? || now - last >= 0.25 || processed == total
+          checkpoint(id, processed, total)
+          last = now
+        end
+      end
+      check = lambda do
+        now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
+        if now - last >= 0.25
+          checkpoint(id)
+          last = now
+        end
+      end
+      token = CountsExportDownload.prepare(exporter, state.fetch('owner'),
+          progress:, check:, identifier: id)
+      synchronise do
+        current = read(id)
+        if (!current || current['status'] != 'running' ||
+            current['expires_at'] <= Time.current.to_f)
+          file = CountsExportDownload.find(token, state.fetch('owner'))
+          File.unlink(file.fetch('path')) if file
+          raise Cancelled
+        end
+        file = CountsExportDownload.find(token, state.fetch('owner'))
+        current.merge!('status' => 'ready', 'token' => token,
+            'download_file' => File.basename(file.fetch('path')))
+        current['total'] = current['processed']
+        write(current)
+      end
+    rescue StandardError => error
+      remove_partial(id)
+      synchronise do
+        current = read(id)
+        if current
+          current['status'] = error.is_a?(Cancelled) ? 'cancelled' : 'failed'
+          current['error'] = error.class.name
+          write(current)
+        end
+      end
+      unless error.is_a?(Cancelled)
+        Rails.logger.error("counts_export_job #{ id }: #{ error.class }")
+      end
+    end
+  end
+end

backend/app/services/counts_export_projection.rb

@@ -24,6 +24,22 @@ class CountsExportProjection
       'aliases' => ['languageCode', 'name', 'scriptCode'],
       'parents' => ['id'] }.freeze
 
+  def self.default_columns(resource)
+    fields = { 'posts' => ['title', 'url', 'tags', 'parents', 'video_ms'],
+        'tags' => ['name', 'category', 'aliases', 'parents'],
+        'nico_tags' => ['name', 'linked_tags'],
+        'tag_descriptions' => ['language_code', 'body', 'source_language_code'],
+        'tag_groups' => ['name', 'description', 'members'],
+        'materials' => ['url', 'tag_id', 'source_kind'] }.fetch(resource)
+    ['resource_id', 'version_no'] + fields + ['event_type', 'user', 'created_at']
+  end
+
+  def self.default_children(resource)
+    { 'posts' => ['tags.name', 'parents.id'],
+        'tags' => ['aliases.name', 'parents.id'],
+        'tag_groups' => ['members.tagId'] }.fetch(resource, [])
+  end
+
   attr_reader :columns, :headers
 
   def initialize(params, resource:, summary:, tags:)
@@ -37,7 +53,8 @@ class CountsExportProjection
               else
                 CountsQuery::COLUMNS.fetch(resource)
               end
-    @columns = params.fetch('columns', allowed.join(',')).split(',')
+    defaults = summary ? allowed : self.class.default_columns(resource)
+    @columns = params.fetch('columns', defaults.join(',')).split(',')
     raise ArgumentError if (@columns - allowed).any? || @columns.uniq != @columns
 
     @headers = JSON.parse(params.fetch('headers', JSON.generate(@columns)))
@@ -55,7 +72,8 @@ class CountsExportProjection
 
       [path, direction]
     end
-    @selected = params.key?('children') ? params['children'].split(',').to_set : nil
+    @selected = params.key?('children') ? params['children'].split(',').to_set :
+        self.class.default_children(resource).to_set
     valid = CHILDREN.flat_map do |key, fields|
       paths(fields, key)
     end

backend/app/services/counts_exporter.rb

@@ -47,14 +47,17 @@ class CountsExporter
     end
   end
 
-  def body
-    return @table_export.body if @table_export
+  def body(progress: nil)
+    return @table_export.body(progress:) if @table_export
 
     Enumerator.new do |output|
       ApplicationRecord.transaction(isolation: :repeatable_read) do
+        processed = 0
         first_row = true
         started = false
         pages do |page|
+          total = @mode == 'details' ? page[:count] : nil
+          progress&.call(processed, total)
           unless started
             meta = { resource:, mode: @mode, basis: @history ? 'history' : 'state',
                      at: @query_params['at'], boundary: 'inclusive',
@@ -83,8 +86,11 @@ class CountsExporter
               output << CSV.generate_line(values, row_sep: "\r\n")
             end
             first_row = false
+            processed += 1
+            progress&.call(processed, total)
           end
         end
+        progress&.call(processed, processed)
         output << ']}' if format == 'json'
       end
     end

backend/app/services/counts_table_export.rb

@@ -10,6 +10,10 @@ class CountsTableExport
     raise ArgumentError if (params.keys - CountsTableQuery::KEYS - EXTRA).any?
     raise ArgumentError unless params.values.all? { _1.is_a?(String) }
 
+    unless params.key?('children')
+      params = params.merge('children' => CountsExportProjection.default_children(
+          params.fetch('resource', 'posts')).join(','))
+    end
     @params = params
     @labels = JSON.parse(params.fetch('labels', '{}'))
     raise ArgumentError unless @labels.is_a?(Hash)
@@ -27,8 +31,11 @@ class CountsTableExport
     @query = CountsTableQuery.new(params.slice(*CountsTableQuery::KEYS)
         .merge('page' => '1', 'limit' => '100'))
     @summary = @query.mode == 'summary'
-    @columns = params.fetch('columns', '').split(',')
-    @headers = JSON.parse(params.fetch('headers', '[]'))
+    default_columns = @summary ? @query.axes.all.map { _1.fetch('field') } + ['count'] :
+        CountsExportProjection.default_columns(resource)
+    @columns = params.fetch('columns', default_columns.join(',')).split(',')
+    default_headers = @columns.map { @labels.fetch(_1, _1) }
+    @headers = JSON.parse(params.fetch('headers', default_headers.to_json))
     raise ArgumentError unless @headers.is_a?(Array) && @headers.all? { _1.is_a?(String) }
     if @summary && @headers.any? && @headers.length != @query.axes.all.length + 1
       raise ArgumentError
@@ -45,13 +52,16 @@ class CountsTableExport
     end
   end
 
-  def body
+  def body(progress: nil)
     Enumerator.new do |output|
       ApplicationRecord.transaction(isolation: :repeatable_read) do
         @language_names = Language.pluck(:code, :name).to_h
+        processed = 0
         started = false
         first = true
         @query.each_page do |page|
+          total = @summary ? nil : page[:count]
+          progress&.call(processed, total)
           unless started
             metadata = page.except(:versions, :cells, :axis_labels, :column_values).merge(
                 resource:, filters: @params.slice(*CountsTableQuery::KEYS),
@@ -84,8 +94,11 @@ class CountsTableExport
               output << CSV.generate_line(values.map { safe(_1) }, row_sep: "\r\n")
             end
             first = false
+            processed += 1
+            progress&.call(processed, total)
           end
         end
+        progress&.call(processed, processed)
         output << ']}' if format == 'json'
       end
     end