diff --git a/backend/app/services/counts_axes.rb b/backend/app/services/counts_axes.rb index cdd2907..3bb8dfb 100644 --- a/backend/app/services/counts_axes.rb +++ b/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'] }) diff --git a/backend/app/services/counts_candidates.rb b/backend/app/services/counts_candidates.rb index 5a26c2a..ded9d1d 100644 --- a/backend/app/services/counts_candidates.rb +++ b/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) } diff --git a/backend/app/services/counts_exporter.rb b/backend/app/services/counts_exporter.rb index 09cd584..8af545a 100644 --- a/backend/app/services/counts_exporter.rb +++ b/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 diff --git a/backend/app/services/counts_pivot_totals.rb b/backend/app/services/counts_pivot_totals.rb index 42125c7..d516171 100644 --- a/backend/app/services/counts_pivot_totals.rb +++ b/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 diff --git a/backend/app/services/counts_table_export.rb b/backend/app/services/counts_table_export.rb index bd03bd8..eab1d50 100644 --- a/backend/app/services/counts_table_export.rb +++ b/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 diff --git a/backend/app/services/counts_table_query.rb b/backend/app/services/counts_table_query.rb index 049dc72..51e646b 100644 --- a/backend/app/services/counts_table_query.rb +++ b/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 diff --git a/backend/spec/requests/counts_download_spec.rb b/backend/spec/requests/counts_download_spec.rb index 0b1d72a..9d03297 100644 --- a/backend/spec/requests/counts_download_spec.rb +++ b/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', diff --git a/backend/spec/services/counts_export_spec.rb b/backend/spec/services/counts_export_spec.rb index b0671cb..052bf2e 100644 --- a/backend/spec/services/counts_export_spec.rb +++ b/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', diff --git a/backend/spec/services/counts_pivot_security_spec.rb b/backend/spec/services/counts_pivot_security_spec.rb index 5bb234e..6548e36 100644 --- a/backend/spec/services/counts_pivot_security_spec.rb +++ b/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]) diff --git a/backend/spec/services/shared_resource_id_allocator_mysql_spec.rb b/backend/spec/services/shared_resource_id_allocator_mysql_spec.rb index e6b2aa4..8487c30 100644 --- a/backend/spec/services/shared_resource_id_allocator_mysql_spec.rb +++ b/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