akadmin/chatwoot
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: captain auto sync scheduler resilience (#14379)

Browse Source 
# 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
This commit is contained in:
Aakash Bakhle
2026-05-06 18:07:22 +05:30
committed by GitHub
parent 815593eec9
commit d7d1e4113c
2 changed files with 126 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
enterprise/app/jobs/captain/documents/schedule_syncs_job.rb
+65 -16
View File
@@ -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)
spec/enterprise/jobs/captain/documents/schedule_syncs_job_spec.rb
+61
View File
@@ -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