|
| 1 | +# frozen_string_literal: true |
| 2 | + |
| 3 | +require "database_test_helper" |
| 4 | + |
| 5 | +class ActivationCandidatesTest < ActiveSupport::TestCase |
| 6 | + class CandidateActor < SolidObjects::Actor |
| 7 | + actor_type "activation-candidates" |
| 8 | + |
| 9 | + def run |
| 10 | + end |
| 11 | + end |
| 12 | + |
| 13 | + test "claimed candidates skip paused and live leases in claim order" do |
| 14 | + now = SolidObjects.database_adapter.database_now |
| 15 | + owner = create_process |
| 16 | + expired = claim_message("expired", claimed_at: now - 40.seconds, owner:, lease_expires_at: now - 1.second) |
| 17 | + unleased = claim_message("unleased", claimed_at: now - 30.seconds) |
| 18 | + first_tie = claim_message("first-tie", claimed_at: now - 20.seconds) |
| 19 | + second_tie = claim_message("second-tie", claimed_at: now - 20.seconds) |
| 20 | + unexpiring = claim_message("unexpiring", claimed_at: now - 10.seconds, owner:) |
| 21 | + claim_message("paused", claimed_at: now - 50.seconds, paused_at: now - 1.minute) |
| 22 | + claim_message("leased", claimed_at: now - 60.seconds, owner:, lease_expires_at: now + 1.minute) |
| 23 | + |
| 24 | + assert_equal [ expired, unleased, first_tie, second_tie, unexpiring ], claimed_instance_ids(now) |
| 25 | + end |
| 26 | + |
| 27 | + test "claimed candidates stop at the claim scan limit" do |
| 28 | + now = SolidObjects.database_adapter.database_now |
| 29 | + SolidObjects.configuration.claim_scan_limit = 2 |
| 30 | + oldest = claim_message("oldest", claimed_at: now - 30.seconds) |
| 31 | + older = claim_message("older", claimed_at: now - 20.seconds) |
| 32 | + claim_message("newest", claimed_at: now - 10.seconds) |
| 33 | + |
| 34 | + assert_equal [ oldest, older ], claimed_instance_ids(now) |
| 35 | + end |
| 36 | + |
| 37 | + test "SQLite reads claimed candidates from the claimed messages when they have no statistics" do |
| 38 | + skip "requires a SQLite query plan" unless database_family == :sqlite |
| 39 | + |
| 40 | + now = SolidObjects.database_adapter.database_now |
| 41 | + connection = SolidObjects::Record.connection |
| 42 | + restoring_sqlite_statistics(connection) do |
| 43 | + SolidObjects::Instance.insert_all!(Array.new(3_000) { |index| |
| 44 | + { actor_type: "activation-candidates", actor_id: "idle-#{index}", state: {}, created_at: now, updated_at: now } |
| 45 | + }) |
| 46 | + connection.execute("ANALYZE") |
| 47 | + analyzed_tables = connection.select_values("SELECT DISTINCT tbl FROM sqlite_stat1") |
| 48 | + |
| 49 | + assert_includes analyzed_tables, SolidObjects::Instance.table_name |
| 50 | + refute_includes analyzed_tables, SolidObjects::ClaimedMessage.table_name |
| 51 | + |
| 52 | + plan = sqlite_query_plan(connection, SolidObjects::ClaimedMessage.table_name) { claimed_instance_ids(now) } |
| 53 | + |
| 54 | + assert_match(/\A(SCAN|SEARCH) #{SolidObjects::ClaimedMessage.table_name}\b/, plan.first, plan.join("\n")) |
| 55 | + assert plan.none? { |step| step.start_with?("SCAN #{SolidObjects::Instance.table_name}") }, plan.join("\n") |
| 56 | + end |
| 57 | + end |
| 58 | + |
| 59 | + private |
| 60 | + |
| 61 | + def claimed_instance_ids(now) |
| 62 | + SolidObjects::ActivationManager.new(owner_id: SecureRandom.uuid).send(:claimed_instance_ids, now) |
| 63 | + end |
| 64 | + |
| 65 | + def claim_message(actor_id, claimed_at:, owner: nil, lease_expires_at: nil, paused_at: nil) |
| 66 | + message = SolidObjects::Message.find(CandidateActor.ref(actor_id).async.run.id) |
| 67 | + SolidObjects::ReadyMessage.where(message:).delete_all |
| 68 | + message.instance.update!( |
| 69 | + activation_owner_id: owner&.id, |
| 70 | + activation_token: owner && SecureRandom.uuid, |
| 71 | + activation_expires_at: lease_expires_at, |
| 72 | + paused_at: |
| 73 | + ) |
| 74 | + SolidObjects::ClaimedMessage.create!(message:, instance: message.instance, activation_generation: 1, claimed_at:) |
| 75 | + message.instance_id |
| 76 | + end |
| 77 | + |
| 78 | + def create_process |
| 79 | + SolidObjects::Process.create!( |
| 80 | + id: SecureRandom.uuid, |
| 81 | + kind: "worker", |
| 82 | + hostname: "test-host", |
| 83 | + pid: ::Process.pid, |
| 84 | + started_at: Time.current, |
| 85 | + last_heartbeat_at: Time.current, |
| 86 | + metadata: {} |
| 87 | + ) |
| 88 | + end |
| 89 | + |
| 90 | + def sqlite_query_plan(connection, table_name) |
| 91 | + statements = [] |
| 92 | + subscriber = ->(*arguments) { statements << arguments.last } |
| 93 | + ActiveSupport::Notifications.subscribed(subscriber, "sql.active_record") { yield } |
| 94 | + statement = statements.find { |payload| payload[:sql].match?(/\ASELECT .* FROM "#{table_name}"/) } |
| 95 | + assert statement, "the #{table_name} query was not captured" |
| 96 | + |
| 97 | + connection.select_all("EXPLAIN QUERY PLAN #{statement[:sql]}", "SQL", statement[:binds]).map { |row| row["detail"] } |
| 98 | + end |
| 99 | +end |
0 commit comments