From d7d1e4113cd09e1d08f5002652ff6f34291ca28d Mon Sep 17 00:00:00 2001 From: Aakash Bakhle <48802744+aakashb95@users.noreply.github.com> Date: Wed, 6 May 2026 18:07:22 +0530 Subject: [PATCH] fix: captain auto sync scheduler resilience (#14379) # Pull Request Template ## Description skip documents that fail with ActiveRecord errors possibly due to stale/corrupt data and not crash scheduler How did we find out about this error? before October 28th, 2025, we did not have url normalisation. so we had document rows as: id: 123 `https://example.com` status: `in_progress` --> likely stuck crawl id 234: `https://example.com/` status: `available` When the schedule sync job ran, it ran an `document.update!(sync_status: :syncing, last_sync_attempted_at: Time.current)` on the 234 one since it was `available` now `update!` runs `before_validation :normalize_external_link` so `https://example.com/` became `https://example.com` which invalidated: `validates :external_link, uniqueness: { scope: :assistant_id }` so the scheduler crashed. This PR logs the skipped ones with their errors and continues to pick other documents to scheduler doesn't crash ## Type of change - [x] Bug fix (non-breaking change which fixes an issue) ## How Has This Been Tested? Please describe the tests that you ran to verify your changes. Provide instructions so we can reproduce. Please also list any relevant details for your test configuration. spec ## Checklist: - [x] My code follows the style guidelines of this project - [x] I have performed a self-review of my code - [x] I have commented on my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [x] Any dependent changes have been merged and published in downstream modules --- .../captain/documents/schedule_syncs_job.rb | 81 +++++++++++++++---- .../documents/schedule_syncs_job_spec.rb | 61 ++++++++++++++ 2 files changed, 126 insertions(+), 16 deletions(-) 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