diff --git a/enterprise/app/jobs/captain/documents/schedule_syncs_job.rb b/enterprise/app/jobs/captain/documents/schedule_syncs_job.rb index 3c2c55132..4103fd271 100644 --- a/enterprise/app/jobs/captain/documents/schedule_syncs_job.rb +++ b/enterprise/app/jobs/captain/documents/schedule_syncs_job.rb @@ -3,12 +3,13 @@ class Captain::Documents::ScheduleSyncsJob < ApplicationJob PER_ACCOUNT_HOURLY_CAP = 50 GLOBAL_HOURLY_CAP = 1000 + DUE_DOCUMENT_BATCH_SIZE = PER_ACCOUNT_HOURLY_CAP * 2 # Inspite of skipping, we should at least reach the hourly cap SYNC_STALE_TIMEOUT = Captain::Document::SYNC_STALE_TIMEOUT def perform @remaining_global_capacity = GLOBAL_HOURLY_CAP sync_intervals = Enterprise::Account.captain_document_sync_intervals - stats = { accounts_scanned: 0, accounts_enabled: 0, accounts_scheduled: 0, documents_enqueued: 0 } + stats = { accounts_scanned: 0, accounts_enabled: 0, accounts_scheduled: 0, documents_enqueued: 0, documents_skipped: 0 } Account.joins(:captain_documents).distinct.find_each(batch_size: 100) do |account| break if @remaining_global_capacity <= 0 @@ -21,7 +22,9 @@ class Captain::Documents::ScheduleSyncsJob < ApplicationJob next unless interval stats[:accounts_scheduled] += 1 - stats[:documents_enqueued] += enqueue_due_documents(account, interval) + result = enqueue_due_documents(account, interval) + stats[:documents_enqueued] += result[:enqueued] + stats[:documents_skipped] += result[:skipped] end log_scheduler_summary(stats) @@ -30,28 +33,74 @@ class Captain::Documents::ScheduleSyncsJob < ApplicationJob private def enqueue_due_documents(account, interval) + per_account_limit = [PER_ACCOUNT_HOURLY_CAP, @remaining_global_capacity].min + result = { enqueued: 0, skipped: 0 } + skipped_document_ids = [] + + while result[:enqueued] < per_account_limit + + documents = due_documents(account, interval, skipped_document_ids).limit(DUE_DOCUMENT_BATCH_SIZE).to_a + break if documents.empty? + + documents.each do |document| + break if result[:enqueued] >= per_account_limit + + process_due_document(document, result, skipped_document_ids) + end + end + + result + end + + def process_due_document(document, result, skipped_document_ids) + return unless document.syncable? + + # Reserve the sync slot before enqueueing so later scheduler runs skip this document while the job is queued. + unless reserve_sync_slot(document) + result[:skipped] += 1 + skipped_document_ids << document.id + return + end + + Captain::Documents::PerformSyncJob.perform_later(document) + @remaining_global_capacity -= 1 + result[:enqueued] += 1 + end + + def due_documents(account, interval, skipped_document_ids) syncing = Captain::Document.sync_statuses[:syncing] synced = Captain::Document.sync_statuses[:synced] failed = Captain::Document.sync_statuses[:failed] - stale_cutoff = SYNC_STALE_TIMEOUT.ago - per_account_limit = [PER_ACCOUNT_HOURLY_CAP, @remaining_global_capacity].min - enqueued_count = 0 - account.captain_documents.syncable.where(status: :available).where( + documents = account.captain_documents.syncable.where(status: :available).where( '(sync_status = ? AND last_synced_at < ?) OR (sync_status = ? AND last_sync_attempted_at < ?) OR ' \ '(sync_status = ? AND last_sync_attempted_at < ?)', - synced, interval.ago, failed, interval.ago, syncing, stale_cutoff - ).order(Arel.sql('last_sync_attempted_at ASC NULLS FIRST'), :id).limit(per_account_limit).each do |document| - next unless document.syncable? + synced, interval.ago, failed, interval.ago, syncing, SYNC_STALE_TIMEOUT.ago + ) + documents = documents.where.not(id: skipped_document_ids) if skipped_document_ids.present? + documents.order(Arel.sql('last_sync_attempted_at ASC NULLS FIRST'), :id) + end - # Reserve the sync slot before enqueueing so later scheduler runs skip this document while the job is queued. - document.update!(sync_status: :syncing, last_sync_attempted_at: Time.current) - Captain::Documents::PerformSyncJob.perform_later(document) - @remaining_global_capacity -= 1 - enqueued_count += 1 - end + def reserve_sync_slot(document) + document.update!(sync_status: :syncing, last_sync_attempted_at: Time.current) + true + rescue ActiveRecord::RecordInvalid => e + log_document_skip(document, e) + false + end - enqueued_count + def log_document_skip(document, error) + payload = { + event: 'document_skipped', + document_id: document.id, + account_id: document.account_id, + assistant_id: document.assistant_id, + error_class: error.class.name, + error_message: error.message, + validation_errors: document.errors.full_messages + } + + Rails.logger.warn("[Captain::Documents::ScheduleSyncsJob] #{payload.to_json}") end def log_scheduler_summary(stats) diff --git a/spec/enterprise/jobs/captain/documents/schedule_syncs_job_spec.rb b/spec/enterprise/jobs/captain/documents/schedule_syncs_job_spec.rb index 66828ef9f..76604a0b5 100644 --- a/spec/enterprise/jobs/captain/documents/schedule_syncs_job_spec.rb +++ b/spec/enterprise/jobs/captain/documents/schedule_syncs_job_spec.rb @@ -122,6 +122,67 @@ RSpec.describe Captain::Documents::ScheduleSyncsJob, type: :job do expect { described_class.new.perform } .to have_enqueued_job(Captain::Documents::PerformSyncJob).with(document) end + + it 'skips invalid legacy documents without counting them against the account cap' do + stub_const("#{described_class}::PER_ACCOUNT_HOURLY_CAP", 1) + create( + :captain_document, + assistant: assistant, + account: account, + status: :in_progress, + content: nil, + external_link: 'https://example.com' + ) + invalid_document = build( + :captain_document, + assistant: assistant, + account: account, + status: :available, + sync_status: :synced, + last_synced_at: 2.days.ago, + last_sync_attempted_at: 2.days.ago, + external_link: 'https://example.com/' + ) + invalid_document.save!(validate: false) + valid_document = create(:captain_document, assistant: assistant, account: account, status: :available) + valid_document.update!(sync_status: :synced, last_synced_at: 2.days.ago, last_sync_attempted_at: 2.days.ago) + clear_enqueued_jobs + + expect { described_class.new.perform }.not_to raise_error + expect(Captain::Documents::PerformSyncJob).not_to have_been_enqueued.with(invalid_document) + expect(Captain::Documents::PerformSyncJob).to have_been_enqueued.with(valid_document) + end + + it 'keeps paging due documents when invalid documents fill the first batch' do + stub_const("#{described_class}::PER_ACCOUNT_HOURLY_CAP", 1) + stub_const("#{described_class}::DUE_DOCUMENT_BATCH_SIZE", 1) + create( + :captain_document, + assistant: assistant, + account: account, + status: :in_progress, + content: nil, + external_link: 'https://example.com' + ) + invalid_document = build( + :captain_document, + assistant: assistant, + account: account, + status: :available, + sync_status: :synced, + last_synced_at: 2.days.ago, + last_sync_attempted_at: 3.days.ago, + external_link: 'https://example.com/' + ) + invalid_document.save!(validate: false) + valid_document = create(:captain_document, assistant: assistant, account: account, status: :available) + valid_document.update!(sync_status: :synced, last_synced_at: 2.days.ago, last_sync_attempted_at: 2.days.ago) + clear_enqueued_jobs + + described_class.new.perform + + expect(Captain::Documents::PerformSyncJob).to have_been_enqueued.with(valid_document) + end end context 'when more documents are due than the account cap allows' do