#434

backend/app/services/counts_axes.rb

@@ -72,14 +72,27 @@ class CountsAxes
     end
   end
 
-  def count_expression
-    return 'COUNT(DISTINCT v.id)' unless rows.any? { _1['field'] == 'tags' }
-
-    # Keep tag occurrences even when several private tags share one masked label.
-    values = ['v.id', "COALESCE(CONCAT('internal:', tag_entry.tag_id), " +
-        "CONCAT('external:', tag_entry.external_tag_id))"]
-    values << 'changed.field_key' if rows.any? { _1['field'] == 'changed_field' }
-    "COUNT(DISTINCT #{ values.join(', ') })"
+  def grouped_sql(scope, indexes = all.each_index.to_a)
+    keys = indexes.map { "a#{ _1 }" }
+    selected = indexes.map { "#{ expressions[_1] } AS a#{ _1 }" }
+    identities = ['v.id']
+    if rows.any? { _1['field'] == 'tags' }
+      # Raw identities stay inside COUNT; masked tag occurrences remain additive.
+      identities << "COALESCE(CONCAT('internal:', tag_entry.tag_id), " +
+          "CONCAT('external:', tag_entry.external_tag_id))"
+      identities << 'changed.field_key' if rows.any? { _1['field'] == 'changed_field' }
+    end
+    members = identities.each_index.map { "counts_member_#{ _1 }" }
+    selected += identities.each_with_index.map { |expression, index|
+      "#{ expression } AS #{ members[index] }"
+    }
+    # Evaluate correlated visibility subqueries before grouping. LIMIT prevents
+    # derived-table merging, so ONLY_FULL_GROUP_BY sees ordinary projected columns.
+    projection = scope.except(:select, :order).select(Arel.sql(selected.join(', ')))
+        .limit(18_446_744_073_709_551_615).to_sql
+    fields = keys + ["COUNT(DISTINCT #{ members.join(', ') }) AS count"]
+    grouping = keys.empty? ? '' : " GROUP BY #{ keys.join(', ') }"
+    "SELECT #{ fields.join(', ') } FROM (#{ projection }) counts_values#{ grouping }"
   end
 
   def joins(fields = all.map { _1['field'] })

backend/app/services/counts_candidates.rb

@@ -41,17 +41,16 @@ class CountsCandidates
           .to_sql
     end
     clauses = []
-    binds = []
     if @params['category'].present?
-      clauses << 'category = ?'
-      binds << @params['category']
+      clauses << ApplicationRecord.sanitize_sql_array(['category = ?', @params['category']])
     end
     where = clauses.empty? ? '' : "WHERE #{ clauses.join(' AND ') }"
     sql = <<~SQL
       SELECT /*+ MAX_EXECUTION_TIME(10000) */ * FROM (#{ sources.join(' UNION ALL ') }) choices
       #{ where } ORDER BY kind, id LIMIT #{ LIMIT + 1 } OFFSET #{ (@page - 1) * LIMIT }
     SQL
-    rows = connection.select_all(ApplicationRecord.sanitize_sql_array([sql, *binds])).to_a
+    # Sources already contain quoted search terms. Do not reinterpret their % or ? as binds.
+    rows = connection.select_all(sql).to_a
     rows.each { _1['names'] = JSON.parse(_1['names']) if _1['names'].is_a?(String) }
     { rows: rows.first(LIMIT), page: @page, more: rows.length > LIMIT,
       execution_at: internal.at.iso8601(6) }

backend/app/services/counts_exporter.rb

@@ -6,6 +6,13 @@ class CountsExporter
                 'summary_columns', 'nested_json', 'child_sort', 'labels'].freeze
   attr_reader :format, :resource
 
+  def self.with_read_snapshot(&block)
+    # A caller-owned transaction already owns the read view (including fixtures).
+    # MySQL cannot change isolation inside it. Standalone workers always start RR.
+    options = ApplicationRecord.connection.transaction_open? ? {} : { isolation: :repeatable_read }
+    ApplicationRecord.transaction(**options, &block)
+  end
+
   def initialize(params, language_code:)
     params = CountsResourceDefinition.normalise_params(params)
     if params.key?('definition')
@@ -52,7 +59,7 @@ class CountsExporter
     return @table_export.body(progress:, phase:) if @table_export
 
     Enumerator.new do |output|
-      ApplicationRecord.transaction(isolation: :repeatable_read) do
+      self.class.with_read_snapshot do
         processed = 0
         first_row = true
         started = false

backend/app/services/counts_pivot_totals.rb

@@ -41,32 +41,25 @@ class CountsPivotTotals
           }.join(' AND ') }.map { "(#{ _1 })" }.join(' OR ')
           scope = scope.where(predicate, *tuples.flatten)
         end
-        grouped = expressions.each_with_index.map { |expression, index|
-          "#{ expression } AS v#{ index }"
+        grouped = indexes.each_with_index.map { |key, index|
+          "a#{ key } AS v#{ index }"
         }
         # Fixed-width result coordinates, independent of NULL and of translated labels.
         padding = (expressions.length...@axes.all.length).map { "NULL AS v#{ _1 }" }
         fields = ["#{ row_depth } AS row_depth", "#{ column_depth } AS column_depth"] +
-            grouped + padding + ['COUNT(DISTINCT v.id) AS count']
+            grouped + padding + ['count']
         if additive
           # Sum full row tuples, not page cells; preserve DISTINCT inside each tuple.
-          leaf = retained.map { @axes.expressions[_1] }
-          selected = leaf.each_with_index.map do |expression, index|
-            "#{ expression } AS k#{ index }"
-          end
-          selected << "#{ @axes.count_expression } AS count"
-          inner = scope.group(*leaf.map { Arel.sql(_1) })
-              .select(Arel.sql(selected.join(', '))).to_sql
-          coordinates = indexes.map { "k#{ retained.index(_1) }" }
+          inner = @axes.grouped_sql(scope, retained)
+          coordinates = indexes.map { "a#{ _1 }" }
           projected = coordinates.each_with_index.map { |key, index| "#{ key } AS v#{ index }" }
           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 }"
         else
-          scope = scope.group(*expressions.map { Arel.sql(_1) }) if expressions.any?
-          scope.optimizer_hints('MAX_EXECUTION_TIME(10000)')
-              .select(Arel.sql(fields.join(', '))).to_sql
+          inner = @axes.grouped_sql(scope, indexes)
+          "SELECT #{ fields.join(', ') } FROM (#{ inner }) grouped"
         end
       end
     end

backend/app/services/counts_table_export.rb

@@ -57,7 +57,7 @@ class CountsTableExport
 
   def body(progress: nil, phase: nil)
     Enumerator.new do |output|
-      ApplicationRecord.transaction(isolation: :repeatable_read) do
+      CountsExporter.with_read_snapshot do
         processed = 0
         started = false
         first = true

backend/app/services/counts_table_query.rb

@@ -84,19 +84,9 @@ class CountsTableQuery
           cells: page == 1 ? [{ values: [], count: record_counts[:visible] }] : [],
           axis_labels: {} }
     end
-    expressions = axes.expressions
-    selected = expressions.each_with_index.map do |expression, index|
-      "#{ expression } AS a#{ index }"
-    end
     source = summary_population.scope
     source = source.joins(axes.joins) unless axes.joins.empty?
-    sql = if expressions.empty?
-            source.except(:select).select('COUNT(DISTINCT v.id) AS count').to_sql
-          else
-            source.except(:select).group(*expressions.map { Arel.sql(_1) })
-                .select(Arel.sql((selected + ["#{ axes.count_expression } AS count"]).join(', ')))
-                .to_sql
-          end
+    sql = axes.grouped_sql(source)
     prefix = "WITH cells AS (#{ sql })"
     column_values = limited_columns(prefix)
     row_keys = axes.rows.each_index.map { "a#{ _1 }" }
@@ -354,7 +344,6 @@ class CountsTableQuery
   end
 
   def measure_relation values
-    row_expressions = axes.expressions.first(axes.rows.length)
     source = summary_population.scope.except(:select, :order)
     # Omitted multi-value column axes still determine full-cell membership.
     membership = summary_population.scope.except(:select, :order)
@@ -368,11 +357,7 @@ class CountsTableQuery
       expression = axes.expressions[axes.rows.length + index]
       source = source.where("(#{ expression }) <=> ?", value)
     end
-    selected = row_expressions.each_with_index.map { |expression, index|
-      "#{ expression } AS a#{ index }"
-    } + ["#{ axes.count_expression } AS count"]
-    source = source.group(*row_expressions.map { Arel.sql(_1) }) if row_expressions.any?
-    source.select(Arel.sql(selected.join(', '))).to_sql
+    axes.grouped_sql(source, axes.rows.each_index.to_a)
   end
 
   def axis_order_sql key, direction = 'asc', qualifier: nil

backend/spec/requests/counts_download_spec.rb

@@ -5,11 +5,15 @@ RSpec.describe 'Counts native GET acknowledgement', type: :request do
   include ActiveSupport::Testing::TimeHelpers
   around do |example|
     Dir.mktmpdir('counts-download-request') do |path|
-      allow(CountsExportDownload).to receive(:directory).and_return(Pathname.new(path))
+      @counts_download_dir = Pathname.new(path)
       example.run
     end
   end
 
+  before do
+    allow(CountsExportDownload).to receive(:directory).and_return(@counts_download_dir)
+  end
+
   def ready_job(format)
     CountsExportJobs.pulse
     post '/counts/export/jobs', params: { resource: 'posts', definition: 'history',

backend/spec/services/counts_export_spec.rb

@@ -37,7 +37,6 @@ RSpec.describe 'Counts export contracts' do
         block.call(page)
       end
       allow(CountsTableQuery).to receive(:new).and_return(query)
-      allow(ApplicationRecord).to receive(:transaction).with(isolation: :repeatable_read).and_yield
     end
 
     it 'reports aggregation, actual output progress and regular-row metadata' do
@@ -175,11 +174,15 @@ end
 RSpec.describe 'Counts download acknowledgement' do
   around do |example|
     Dir.mktmpdir('counts-ack-spec') do |path|
-      allow(CountsExportDownload).to receive(:directory).and_return(Pathname.new(path))
+      @counts_ack_dir = Pathname.new(path)
       HistoryAccess.set(user: nil) { example.run }
     end
   end
 
+  before do
+    allow(CountsExportDownload).to receive(:directory).and_return(@counts_ack_dir)
+  end
+
   it 'binds an acknowledgement to the owner, job, file token and fresh attempt' do
     CountsExportJobs.pulse
     payload = { 'resource' => 'posts', 'definition' => 'history', 'mode' => 'details',

backend/spec/services/counts_pivot_security_spec.rb

@@ -132,11 +132,14 @@ RSpec.describe 'Counts masked pivot contracts' do
 
   it 'rejects private predicates and supplied measure coordinates before SQL labels escape' do
     probes = [{ 'resource_id' => PostVersion.first.post_id.to_s },
-        { 'tags' => 'A' }, { 'tag_refs.0.name' => 'A', 'tag_refs.0.tag_id' => tags.first.id.to_s },
-        { 'conditions.0.field' => 'title', 'conditions.0.operator' => 'eq',
+        { 'tags' => 'A' },
+        { 'tag_refs' => '', 'tag_refs.0.name' => 'A', 'tag_refs.0.tag_id' => tags.first.id.to_s },
+        { 'conditions' => '', 'conditions.0.field' => 'title', 'conditions.0.operator' => 'eq',
           'conditions.0.value' => 'hidden-title-0' },
-        { 'drill.0.field' => 'tags', 'drill.0.value' => "internal:#{ tags.first.id }" },
-        { 'cols' => 'tags', 'column_order.0.value' => CountsAxes::MASKED_VALUE }]
+        { 'drill' => '', 'drill.0.field' => 'tags', 'drill.0.value_type' => 'string',
+          'drill.0.value' => "internal:#{ tags.first.id }" },
+        { 'cols' => 'tags', 'column_order' => '', 'column_order.0.value_type' => 'string',
+          'column_order.0.value' => CountsAxes::MASKED_VALUE }]
     HistoryAccess.set(user: nil) do
       probes.each do |probe|
         expect { CountsTableQuery.new(params.merge(probe)) }
@@ -148,6 +151,7 @@ RSpec.describe 'Counts masked pivot contracts' do
   it 'keeps public column sorting usable while guessed private coordinates match nothing' do
     first = result(nil, 'rows' => 'event_type', 'cols' => 'tags')
     probe = result(nil, 'rows' => 'event_type', 'cols' => 'tags',
+        'column_order' => '', 'column_order.0.value_type' => 'string',
         'column_order.0.value' => "internal:#{ tags.first.id }")
     expect(probe[:column_values]).to eq(first[:column_values])
     expect(probe[:cells]).to eq(first[:cells])

backend/spec/services/shared_resource_id_allocator_mysql_spec.rb

@@ -8,6 +8,10 @@ RSpec.describe 'Shared resource IDs across MySQL connections', type: :model do
     expect(Rails.env.test?).to be(true)
     expect(ApplicationRecord.connection.adapter_name).to match(/mysql/i)
     expect(ApplicationRecord.connection_pool.size).to be >= 3
+    @main_connection = ApplicationRecord.connection
+    # A leaked fixture pin makes worker threads share the main session. Fail before
+    # starting a worker rather than diagnosing IS_USED_LOCK(NULL) after a timeout.
+    expect(@main_connection.open_transactions).to eq(0)
     @prefix = "shared_ids_#{ SecureRandom.hex(8) }"
     @tag = create(:tag, primary_name: "#{ @prefix }_owner")
     @group = TagGroup.create!(name: "#{ @prefix }_owner", version_no: 1)
@@ -42,13 +46,20 @@ RSpec.describe 'Shared resource IDs across MySQL connections', type: :model do
   end
 
   def await_queue queue
-    Timeout.timeout(15) { queue.pop }
+    value = Timeout.timeout(15) { queue.pop }
+    raise value if value.is_a?(Exception)
+
+    value
   end
 
   def worker &block
     Thread.new do
       Thread.current.report_on_exception = false
-      ApplicationRecord.connection_pool.with_connection(&block)
+      ApplicationRecord.connection_pool.with_connection do |connection|
+        raise 'Shared ID worker reused the main connection' if connection.equal?(@main_connection)
+
+        block.call(connection)
+      end
     end.tap { @threads << _1 }
   end
 
@@ -86,11 +97,16 @@ RSpec.describe 'Shared resource IDs across MySQL connections', type: :model do
         ActiveSupport::Notifications.subscribed(subscriber, 'sql.active_record') do
           id = create_resource(kind).id
         end
+        raise 'Shared ID creation did not acquire a named lock' unless lock_name
+
         ready << [id, connection.select_value('SELECT CONNECTION_ID()').to_i, lock_name]
         await_queue(@release)
         raise ActiveRecord::Rollback if rollback
       end
       id
+    rescue StandardError => error
+      ready << error
+      raise
     end
     [thread, await_queue(ready)]
   end