#434

backend/app/controllers/counts_exports_controller.rb

@@ -18,7 +18,7 @@ class CountsExportsController < ApplicationController
     render json: { error: 'private_condition' }, status: :forbidden
   rescue CountsTableQuery::TooManyColumns
     render json: { error: 'too_many_columns', max_columns: CountsTableQuery::MAX_COLUMNS },
-        status: :unprocessable_entity
+           status: :unprocessable_entity
   rescue ActiveRecord::StatementTimeout
     render json: { error: 'counts_query_timeout' }, status: :service_unavailable
   rescue ArgumentError, TypeError, KeyError, JSON::ParserError
@@ -86,68 +86,6 @@ class CountsExportsController < ApplicationController
               disposition: 'attachment'
   end
 
-  def confirmed_download
-    owner = session[:counts_export_owner]
-    arguments = { file_token: params[:token], attempt: params[:attempt] }
-    return head :not_found unless arguments[:attempt].present? &&
-        CountsExportJobs.download_attempt(params[:job], owner, **arguments)
-
-    file = CountsExportDownload.find(params[:token], owner)
-    unless file
-      CountsExportJobs.download_attempt(params[:job], owner, **arguments, status: 'unavailable')
-      return head :not_found
-    end
-    job = params[:job]
-    authority = HistoryAccess.user
-    first_check = true
-    stream = CountsExportDownload::ResponseBody.new(file.fetch('path')) do |status|
-      if status == 'active'
-        if first_check
-          first_check = false
-          fresh = authority && User.find_by(id: authority.id)
-          next false if authority && (!fresh || fresh.banned?)
-
-          authority = fresh
-          HistoryAccess.set(user: authority) do
-            CountsExportDownload.find(arguments[:file_token], owner) &&
-                CountsExportJobs.stream_active?(job, owner, **arguments)
-          end
-        else
-          CountsExportJobs.stream_active?(job, owner, **arguments)
-        end
-      else
-        HistoryAccess.set(user: authority) do
-          CountsExportJobs.download_attempt(job, owner, **arguments, status:)
-        end
-      end
-    end
-    unless CountsExportJobs.download_attempt(job, owner, **arguments, status: 'streaming')
-      stream.close
-      return head :conflict
-    end
-    response.headers['Cache-Control'] = 'private, no-store'
-    send_file_headers! filename: "counts-#{ file.fetch('resource') }.#{ file.fetch('format') }",
-        type: file['format'] == 'json' ? 'application/json' : 'text/csv',
-        disposition: 'attachment'
-    response.headers['Last-Modified'] = File.mtime(file.fetch('path')).httpdate
-    response.headers['X-Accel-Buffering'] = 'no'
-    self.response_body = stream
-  rescue IOError, SystemCallError
-    stream&.close
-    CountsExportJobs.download_attempt(params[:job], owner, **arguments, status: 'failed')
-    head :service_unavailable
-  end
-  private :confirmed_download
-
-  def job_state(cancel:)
-    result = CountsExportJobs.status(params[:token], session[:counts_export_owner], cancel:)
-    return head :not_found unless result
-
-    response.headers['Cache-Control'] = 'private, no-store'
-    render json: result
-  end
-  private :job_state
-
   def index
     locale = resolve_locale!
     payload = request.request_parameters.except('authenticity_token')
@@ -175,4 +113,73 @@ class CountsExportsController < ApplicationController
   rescue ArgumentError, TypeError, KeyError, JSON::ParserError
     head :bad_request
   end
+
+  private
+
+  def confirmed_download
+    owner = session[:counts_export_owner]
+    arguments = { file_token: params[:token], attempt: params[:attempt] }
+    if (!(arguments[:attempt].present?) ||
+        !(CountsExportJobs.download_attempt(params[:job], owner, **arguments)))
+      return head :not_found
+    end
+
+    file = CountsExportDownload.find(params[:token], owner)
+    unless file
+      CountsExportJobs.download_attempt(params[:job], owner, **arguments, status: 'unavailable')
+      return head :not_found
+    end
+    job = params[:job]
+    authority = HistoryAccess.user
+    first_check = true
+    state = lambda do |status|
+      if status == 'active'
+        if first_check
+          first_check = false
+          fresh = authority && User.find_by(id: authority.id)
+          next false if authority && (!fresh || fresh.banned?)
+
+          authority = fresh
+          HistoryAccess.set(user: authority) do
+            CountsExportDownload.find(arguments[:file_token], owner) &&
+                CountsExportJobs.stream_active?(job, owner, **arguments)
+          end
+        else
+          CountsExportJobs.stream_active?(job, owner, **arguments)
+        end
+      else
+        HistoryAccess.set(user: authority) do
+          CountsExportJobs.download_attempt(job, owner, **arguments, status:)
+        end
+      end
+    end
+    unless CountsExportJobs.download_attempt(job, owner, **arguments, status: 'streaming')
+      return head :conflict
+    end
+    response.headers['Cache-Control'] = 'private, no-store'
+    response.headers['Last-Modified'] = File.mtime(file.fetch('path')).httpdate
+    response.headers['X-Accel-Buffering'] = 'no'
+    send_file file.fetch('path'),
+        filename: "counts-#{ file.fetch('resource') }.#{ file.fetch('format') }",
+        type: file['format'] == 'json' ? 'application/json' : 'text/csv',
+        disposition: 'attachment'
+    # Keep Rails' bounded FileBody reader, but hide to_path from Rack::Sendfile:
+    # offloading would bypass the GET acknowledgement and cancellation checks.
+    stream = CountsExportDownload::ResponseBody.new(response.stream, &state)
+    self.response_body = stream
+  rescue IOError, SystemCallError
+    stream&.close
+    CountsExportJobs.download_attempt(params[:job], owner, **arguments, status: 'failed')
+    raise if response.committed?
+
+    head :service_unavailable
+  end
+
+  def job_state cancel:
+    result = CountsExportJobs.status(params[:token], session[:counts_export_owner], cancel:)
+    return head :not_found unless result
+
+    response.headers['Cache-Control'] = 'private, no-store'
+    render json: result
+  end
 end

backend/app/controllers/counts_tables_controller.rb

@@ -25,6 +25,11 @@ class CountsTablesController < ApplicationController
   rescue CountsTableQuery::TooManyColumns
     render json: { error: 'too_many_columns', max_columns: CountsTableQuery::MAX_COLUMNS },
            status: :unprocessable_entity
+  rescue CountsPivotTotals::TimedOut
+    Rails.logger.warn('counts_query_timeout: pivot_totals')
+    render json: { error: 'counts_query_timeout', stage: 'totals',
+                   timeout_seconds: CountsTableQuery::SQL_TIMEOUT_SECONDS },
+           status: :service_unavailable
   rescue ActiveRecord::StatementTimeout => error
     Rails.logger.warn("counts_query_timeout: #{ error.class }")
     render json: { error: 'counts_query_timeout' }, status: :service_unavailable

backend/app/services/counts_export_download.rb

@@ -15,26 +15,38 @@ class CountsExportDownload
 
   # Rack body deliberately has no to_path: sendfile middleware must not bypass ACK.
   class ResponseBody
-    def initialize(path, &state)
-      File.open(path, 'rb') { |file| file.stat }
-      @path = path
-      @stream = nil
+    def initialize body, &state
+      @body = body
       @closed = false
       @state = state
     end
 
-    def each
+    def each &block
+      return enum_for(:each) unless block
       return if @closed
 
-      @stream = File.open(@path, 'rb')
+      transmit(&block)
+    end
+
+    def close
+      return if @closed
+
+      @closed = true
+      @body.close if @body.respond_to?(:close)
+    end
+
+    private
+
+    def transmit
       first = true
-      while (chunk = @stream.read(64 * 1024))
+      @body.each do |chunk|
         raise IOError, 'Download no longer authorised' unless @state.call('active')
 
         yield chunk
         if first
-          # This confirms yielding to Rack, never browser receipt or saving.
-          @state.call('started')
+          # Rack accepted a chunk; browser receipt and saving are not guaranteed.
+          raise IOError, 'Download acknowledgement failed' unless @state.call('started')
+
           first = false
         end
       end
@@ -46,11 +58,6 @@ class CountsExportDownload
     ensure
       close
     end
-
-    def close
-      @closed = true
-      @stream.close if @stream && !@stream.closed?
-    end
   end
 
   def self.directory

backend/app/services/counts_export_projection.rb

@@ -157,7 +157,7 @@ class CountsExportProjection
 
   def csv_display(value, key)
     if key.start_with?('aliases_') && value.is_a?(Array)
-      return value.map { fetch(_1, 'name') }.compact.join("\n")
+      return value.map { fetch(_1, 'name') }.compact.join(' ')
     end
     return @labels.fetch('bot', '') if key == 'created_by_user' && value.nil?
     return @labels.fetch("event_type:#{ value }", value) if key == 'event_type'

backend/app/services/counts_pivot_totals.rb

@@ -1,5 +1,6 @@
 # Re-aggregate distinct membership; tag row tuples alone use additive contributions.
 class CountsPivotTotals
+  class TimedOut < ActiveRecord::StatementTimeout; end
   def initialize(population, axes, row_headers, column_headers)
     @population = population
     @axes = axes
@@ -27,20 +28,26 @@ class CountsPivotTotals
         retained = additive ? (0...@axes.rows.length).to_a | indexes : indexes
         joins = @axes.joins(retained.map { @axes.all[_1].fetch('field') })
         scope = scope.joins(joins) unless joins.empty?
+        predicates = []
         if row_depth.positive?
           tuples = headers.map { _1[:values] }.uniq
           predicate = tuples.map { Array.new(row_depth) { |i|
-            "(#{ expressions[i] }) <=> ?"
+            "a#{ indexes[i] } <=> ?"
           }.join(' AND ') }.map { "(#{ _1 })" }.join(' OR ')
-          scope = scope.where(predicate, *tuples.flatten)
+          predicates << @population.definition.model.sanitize_sql_array([
+              tuples.empty? ? 'FALSE' : "(#{ predicate })", *tuples.flatten])
         end
         if column_depth.positive?
           tuples = columns.map { _1[:values] }.uniq
           predicate = tuples.map { Array.new(column_depth) { |i|
-            "(#{ expressions[row_depth + i] }) <=> ?"
+            "a#{ indexes[row_depth + i] } <=> ?"
           }.join(' AND ') }.map { "(#{ _1 })" }.join(' OR ')
-          scope = scope.where(predicate, *tuples.flatten)
+          predicates << @population.definition.model.sanitize_sql_array([
+              tuples.empty? ? 'FALSE' : "(#{ predicate })", *tuples.flatten])
         end
+        # Filter projected coordinates, not repeated masking/time expressions.
+        # Each member still participates in the complete DISTINCT population.
+        selection = predicates.empty? ? '' : " WHERE #{ predicates.join(' AND ') }"
         grouped = indexes.each_with_index.map { |key, index|
           "a#{ key } AS v#{ index }"
         }
@@ -56,19 +63,19 @@ class CountsPivotTotals
           output = ["#{ row_depth } AS row_depth", "#{ column_depth } AS column_depth"] +
               projected + padding + ['SUM(count) AS count']
           grouping = coordinates.empty? ? '' : " GROUP BY #{ coordinates.join(', ') }"
-          "SELECT #{ output.join(', ') } FROM (#{ inner }) leaf#{ grouping }"
+          "SELECT #{ output.join(', ') } FROM (#{ inner }) leaf#{ selection }#{ grouping }"
         else
           inner = @axes.grouped_sql(scope, indexes)
-          "SELECT #{ fields.join(', ') } FROM (#{ inner }) grouped"
+          "SELECT #{ fields.join(', ') } FROM (#{ inner }) grouped#{ selection }"
         end
       end
     end
     return [] if queries.empty?
 
     sql = queries.map { "(#{ _1 })" }.join(' UNION ALL ')
-    @population.definition.model.connection.select_all(
-        "WITH population_members AS (#{ basis }) " +
-        "SELECT /*+ MAX_EXECUTION_TIME(10000) */ * FROM (#{ sql }) totals").map do |record|
+    statement = "WITH population_members AS (#{ basis }) " +
+        "SELECT /*+ #{ CountsTableQuery::SQL_HINTS.join(' ') } */ * FROM (#{ sql }) totals"
+    @population.definition.model.connection.select_all(statement).map do |record|
       row_depth = record['row_depth'].to_i
       column_depth = record['column_depth'].to_i
       { row: (0...row_depth).map { record["v#{ _1 }"] },
@@ -78,5 +85,7 @@ class CountsPivotTotals
             row_depth, column_depth),
         count: record['count'].to_i }
     end
+  rescue ActiveRecord::StatementTimeout
+    raise TimedOut, 'Counts pivot totals exceeded the query time limit'
   end
 end

backend/app/services/counts_table_query.rb

@@ -1,6 +1,7 @@
 # Common entrypoint for six version resources; legacy endpoints keep their contracts.
 class CountsTableQuery
-  SQL_HINTS = ['MAX_EXECUTION_TIME(10000)',
+  SQL_TIMEOUT_SECONDS = 30
+  SQL_HINTS = ["MAX_EXECUTION_TIME(#{ SQL_TIMEOUT_SECONDS * 1000 })",
                "SET_VAR(optimizer_switch='semijoin=off')"].freeze
   MAX_COLUMNS = 100
   class TooManyColumns < ArgumentError; end
@@ -176,7 +177,7 @@ class CountsTableQuery
   end
 
   def details
-    query = population.scope.select(Arel.sql(
+    query = population.scope.optimizer_hints(*SQL_HINTS).select(Arel.sql(
         "(#{ population.current_latest_sql }) AS counts_latest"))
     order = sort_sql { detail_expression(_1) }
     query = query.order(Arel.sql((order + ['v.id ASC']).join(', ')))
@@ -373,7 +374,9 @@ class CountsTableQuery
       return "#{ column } IS NULL ASC, (#{ column } <=> #{ masked }) ASC, " +
           "#{ expression } #{ direction.upcase }"
     end
-    if field == 'category'
+    if field == 'event_type'
+      expression = event_order_sql(column)
+    elsif field == 'category'
       # Tag's enum preserves the same category order as TagsController.
       cases = Tag.categories.keys.each_with_index.map { |category, index|
         "WHEN #{ @connection.quote(category) } THEN #{ index }"
@@ -398,6 +401,7 @@ class CountsTableQuery
 
   def detail_expression(key)
     return "(#{ population.current_latest_sql })" if key == 'latest'
+    return event_order_sql(@definition.expression(key)) if key == 'event_type'
     return @definition.name_expression(key) if @definition.resource == 'tags' &&
         @definition.name_column(key)
     aliases = { 'tags' => 'tags_json', 'parents' => @definition.resource == 'posts' ?
@@ -409,6 +413,11 @@ class CountsTableQuery
     "CAST(#{ expression } AS CHAR) COLLATE utf8mb4_bin"
   end
 
+  def event_order_sql(expression)
+    "CASE #{ expression } WHEN 'create' THEN 0 WHEN 'update' THEN 1 " +
+        "WHEN 'discard' THEN 2 WHEN 'restore' THEN 3 ELSE 4 END"
+  end
+
   def sort_sql
     @sort.flat_map do |key, direction|
       expression = yield key
@@ -423,6 +432,7 @@ class CountsTableQuery
       values = conditions.map { _1['value'] }.compact.uniq
       labels = case field
                when 'tags' then tag_labels(values)
+               when 'parent_tag_ids' then parent_tag_labels(values)
                when 'user'
                  actors = HistoryAccess.visible(population.scope, 'v').except(:select, :order)
                      .select('v.created_by_user_id')
@@ -452,6 +462,7 @@ class CountsTableQuery
                  labels = tag_labels(values)
                  @axis_names["a#{ index }"] = @tag_names
                  labels
+               when 'parent_tag_ids' then parent_tag_labels(values)
                else {}
                end
       result["a#{ index }"] = labels
@@ -459,6 +470,17 @@ class CountsTableQuery
     result
   end
 
+  def parent_tag_labels(values)
+    ids = values.flat_map { _1.to_s.split }.uniq
+    tag_labels(ids.map { "internal:#{ _1 }" })
+    values.to_h do |value|
+      names = value.to_s.split.map do |id|
+        @tag_names.fetch("internal:#{ id }", {}).dig(locale.language_code, 'name')
+      end
+      [value.to_s, names.compact.join(' ')]
+    end
+  end
+
   def tag_labels(values)
     internal = values.grep(/\Ainternal:/).map { _1.split(':').last.to_i }
     external = values.grep(/\Aexternal:/).map { _1.split(':').last.to_i }

backend/spec/requests/counts_download_spec.rb

@@ -14,7 +14,7 @@ RSpec.describe 'Counts native GET acknowledgement', type: :request do
     allow(CountsExportDownload).to receive(:directory).and_return(@counts_download_dir)
   end
 
-  def ready_job(format)
+  def ready_job(format, payload: 'payload')
     CountsExportJobs.pulse
     post '/counts/export/jobs', params: { resource: 'posts', definition: 'history',
         mode: 'details', columns: 'resource_id', locale: 'ja', format: }
@@ -22,7 +22,7 @@ RSpec.describe 'Counts native GET acknowledgement', type: :request do
     job = response.parsed_body.fetch('job')
     signature = CountsExportJobs.verifier.verified(job, purpose: 'counts_export_job')
     owner = signature.fetch('owner')
-    exporter = instance_double(CountsExporter, resource: 'posts', format:, body: ['payload'])
+    exporter = instance_double(CountsExporter, resource: 'posts', format:, body: [payload])
     file = HistoryAccess.set(user: nil) { CountsExportDownload.prepare(exporter, owner) }
     state = CountsExportJobs.read(signature.fetch('id'))
     state.merge!('status' => 'ready', 'token' => file)
@@ -33,6 +33,28 @@ RSpec.describe 'Counts native GET acknowledgement', type: :request do
   end
 
   ['csv', 'json'].each do |format|
+    it "delivers every byte of a multi-chunk #{ format } file through HTTP" do
+      sign_in_as(create(:user, :member))
+      rows = Array.new(5000) { |index| { id: index, name: '日本語,名称' } }
+      payload = if format == 'json'
+                  rows.to_json
+                else
+                  "id,name\r\n" + rows.map { CSV.generate_line(_1.values) }.join
+                end
+      job, file, attempt = ready_job(format, payload:)
+      owner = CountsExportJobs.verifier.verified(job, purpose: 'counts_export_job').fetch('owner')
+      metadata = HistoryAccess.set(user: nil) { CountsExportDownload.find(file, owner) }
+      expected = File.binread(metadata.fetch('path'))
+      expect(expected.bytesize).to be > 64 * 1024
+      get "/counts/export/files/#{ file }", params: { job:, attempt: }
+      expect(response).to have_http_status(:ok)
+      expect(response.body.bytesize).to eq(expected.bytesize)
+      expect(response.body.b).to eq(expected)
+      expect(response.body.b).to eq(payload.b)
+      get "/counts/export/jobs/#{ job }"
+      expect(response.parsed_body['download_status']).to eq('started')
+    end
+
     it "acknowledges only an authorised #{ format } GET body, not HEAD or acceptance" do
       job, file, attempt = ready_job(format)
       get "/counts/export/jobs/#{ job }"

backend/spec/requests/counts_table_spec.rb

@@ -50,6 +50,32 @@ RSpec.describe 'Counts table API', type: :request do
 
   def versions(result) = result.fetch('versions').map { _1.fetch('version_no') }
 
+  it 'identifies a totals timeout without exposing SQL in the error response' do
+    allow_any_instance_of(CountsPivotTotals).to receive(:call)
+        .and_raise(CountsPivotTotals::TimedOut, 'private SQL')
+    get '/counts/query', params: base.merge(mode: 'summary', rows: 'created_at:month')
+    expect(response).to have_http_status(:service_unavailable)
+    expect(response.parsed_body).to eq('error' => 'counts_query_timeout',
+        'stage' => 'totals', 'timeout_seconds' => 30)
+    expect(response.body).not_to include('private SQL')
+  end
+
+  it 'applies the same optimizer safeguards to the paginated detail SELECT' do
+    statements = []
+    listener = lambda do |_name, _start, _finish, _id, payload|
+      statements << payload[:sql]
+    end
+    ActiveSupport::Notifications.subscribed(listener, 'sql.active_record') do
+      result = query(tags: 'alpha', limit: 1)
+      expect(result['count']).to eq(2)
+      expect(versions(result)).to eq([1])
+    end
+    details = statements.find { _1.include?('AS counts_latest') }
+    expect(details).not_to be_nil
+    expect(details).to include('MAX_EXECUTION_TIME(30000)',
+        "SET_VAR(optimizer_switch='semijoin=off')")
+  end
+
   it 'separates latest and history and defaults to stable detail ordering and limit 100' do
     history = query
     expect(versions(history)).to eq([1, 2, 3])

backend/spec/services/counts_export_spec.rb

@@ -211,18 +211,37 @@ RSpec.describe 'Counts download acknowledgement' do
 end
 
 RSpec.describe CountsExportDownload::ResponseBody do
+  it 'survives Rails buffer enumeration without closing before the first byte' do
+    Dir.mktmpdir('counts-body-spec') do |directory|
+      path = File.join(directory, 'file')
+      payload = 'data' * 20_000
+      File.binwrite(path, payload)
+      states = []
+      source = ActionDispatch::Response::FileBody.new(path)
+      body = described_class.new(source) { |state| states << state; true }
+      response = ActionDispatch::Response.new
+      response.body = body
+      deferred = response.stream.to_ary
+      expect(states).to eq([])
+      expect(deferred.to_a.join).to eq(payload)
+      expect(states.count('started')).to eq(1)
+    end
+  end
+
   it 'acknowledges only after the first Rack yield and closes its bounded file body' do
     Dir.mktmpdir('counts-body-spec') do |directory|
       path = File.join(directory, 'file')
       File.binwrite(path, 'x' * (64 * 1024 + 3))
       states = []
-      body = described_class.new(path) { |state| states << state; true }
+      source = ActionDispatch::Response::FileBody.new(path)
+      body = described_class.new(source) { |state| states << state; true }
       chunks = []
       body.each do |chunk|
         expect(states).not_to include('started') if chunks.empty?
         chunks << chunk
       end
-      expect(chunks.map(&:bytesize)).to eq([64 * 1024, 3])
+      expect(chunks.join).to eq('x' * (64 * 1024 + 3))
+      expect(chunks.map(&:bytesize).max).to be <= 64 * 1024
       expect(states.count('started')).to eq(1)
       expect(body).not_to respond_to(:to_path)
       expect { body.close }.not_to raise_error
@@ -234,7 +253,8 @@ RSpec.describe CountsExportDownload::ResponseBody do
       path = File.join(directory, 'file')
       File.binwrite(path, 'data')
       states = []
-      body = described_class.new(path) { |state| states << state; true }
+      source = ActionDispatch::Response::FileBody.new(path)
+      body = described_class.new(source) { |state| states << state; true }
       expect { body.each { raise IOError, 'connection closed' } }.to raise_error(IOError)
       expect(states).to include('failed')
       expect(states).not_to include('started')

backend/spec/services/counts_pivot_security_spec.rb

@@ -44,6 +44,20 @@ RSpec.describe 'Counts masked pivot contracts' do
     result.fetch(:pivot_totals).find { _1[:row].empty? && _1[:column].empty? }.fetch(:count)
   end
 
+  it 'preserves monthly name subtotals and grand totals across pages' do
+    settings = { 'resource' => 'tags', 'rows' => 'created_at:month name_ja' }
+    complete = result(admin, settings)
+    paged = result(admin, settings.merge('limit' => '1'))
+    expect(complete[:group_count]).to eq(3)
+    expect(grand(complete)).to eq(3)
+    expect(grand(paged)).to eq(3)
+    subtotal = paged[:pivot_totals].find do |cell|
+      cell[:row] == ['2023-01'] && cell[:column].empty?
+    end
+    expect(subtotal[:count]).to eq(3)
+    expect(paged[:cells].sum { _1[:count] }).to eq(1)
+  end
+
   it 'preserves five tag occurrences after merging private tags and across pages' do
     [nil, guest, member, admin].each do |user|
       page = result(user)