feat(index): indexing repo in shard when more than 20 (#4366)

* feat(index): indexing repo in shard when more than 20

* [autofix.ci] apply automated fixes

* [autofix.ci] apply automated fixes (attempt 2/3)

* chore: add env for enabling index repo in shard

# Conflicts:
#	ee/tabby-webserver/src/service/background_job/mod.rs

* [autofix.ci] apply automated fixes

* fix(job): utilize to_std err to check whether the time is past (#4370)

* [autofix.ci] apply automated fixes

* fix(test): use cloudflare workers-ai llms.txt for test

---------

Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>

Author: zwpaper

crates/tabby-crawler/src/lib.rs

@@ -225,7 +225,7 @@ mod tests {
     #[tokio::test]
     #[traced_test]
     async fn test_crawler_llms_success_developers_cloudflare_with_url() {
-        let base_url = "https://developers.cloudflare.com";
+        let base_url = "https://developers.cloudflare.com/workers-ai";
         let result = crawler_llms(base_url).await;
         assert!(result.is_ok(), "Expected success from {base_url}");
         let docs = result.unwrap();

ee/tabby-webserver/src/service/background_job/git.rs

@@ -9,7 +9,10 @@ use tabby_inference::Embedding;
 use tabby_schema::{job::JobService, repository::GitRepositoryService};
 use tracing::debug;
 
-use super::{helper::Job, index_commits, BackgroundJobEvent};
+use super::{
+    calculate_current_shard, helper::Job, index_commits, should_process_repository,
+    BackgroundJobEvent,
+};
 
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub struct SchedulerGitJob {
@@ -50,7 +53,7 @@ impl SchedulerGitJob {
     }
 
     pub async fn cron(
-        _now: DateTime<Utc>,
+        now: DateTime<Utc>,
         git_repository: Arc<dyn GitRepositoryService>,
         job: Arc<dyn JobService>,
     ) -> tabby_schema::Result<()> {
@@ -76,10 +79,19 @@ impl SchedulerGitJob {
             .chain(config_repositories.into_iter())
             .collect();
 
-        for repository in repositories {
+        let number_of_repo = repositories.len();
+        let current_shard = calculate_current_shard(number_of_repo, now.timestamp());
+
+        for (i, repository) in repositories.iter().enumerate() {
+            if !should_process_repository(i, current_shard, number_of_repo) {
+                continue;
+            }
+
             debug!("Scheduling git repository sync for {:?}", repository);
             let _ = job
-                .trigger(BackgroundJobEvent::SchedulerGitRepository(repository).to_command())
+                .trigger(
+                    BackgroundJobEvent::SchedulerGitRepository(repository.clone()).to_command(),
+                )
                 .await;
         }
         Ok(())

ee/tabby-webserver/src/service/background_job/helper/cron.rs

@@ -1,7 +1,6 @@
 use chrono::{DateTime, TimeZone, Utc};
 pub use cron::Schedule;
 use futures::Stream;
-use tracing::warn;
 
 /// Represents a stream from a cron schedule with a timezone
 #[derive(Clone, Debug)]
@@ -35,14 +34,10 @@ where
                 match next {
                     Some(next) => {
                         let to_sleep = next.clone() - timezone.from_utc_datetime(&Utc::now().naive_utc());
-                        if to_sleep.num_seconds() < 0 {
-                            // If the next time is in the past, skip it and get the next one.
-                            continue
-                        }
                         let to_sleep = match to_sleep.to_std() {
                             Ok(to_sleep) => to_sleep,
-                            Err(err) => {
-                                warn!("Failed to convert to std duration: {}", err);
+                            Err(_) => {
+                                // If the next time is in the past or conversion fails, skip it and get the next one.
                                 continue;
                             }
                         };

ee/tabby-webserver/src/service/background_job/mod.rs

@@ -11,7 +11,7 @@ mod license_check;
 mod third_party_integration;
 mod web_crawler;
 
-use std::{str::FromStr, sync::Arc};
+use std::{env, str::FromStr, sync::Arc};
 
 use cron::Schedule;
 use daily::DailyJob;
@@ -47,6 +47,41 @@ pub use web_crawler::WebCrawlerJob;
 
 use self::third_party_integration::SyncIntegrationJob;
 
+// Sharding configuration constants
+pub const REPOSITORIES_PER_SHARD: usize = 7;
+pub const SHARDING_THRESHOLD: usize = 20;
+
+/// Calculate the current shard for repository processing
+/// Returns Some(shard) if sharding should be used, None otherwise
+fn calculate_current_shard(number_of_repo: usize, timestamp_seconds: i64) -> Option<usize> {
+    // Only run on TABBY_INDEX_REPO_IN_SHARD is not empty and number_of_repo > SHARDING_THRESHOLD
+    // otherwise return None
+    if !(env::var("TABBY_INDEX_REPO_IN_SHARD").is_ok_and(|v| !v.is_empty())
+        && number_of_repo > SHARDING_THRESHOLD)
+    {
+        return None;
+    }
+
+    // `number_of_repo + REPOSITORIES_PER_SHARD - 1` because we should ceil number_of_repo
+    let number_of_shard = number_of_repo.div_ceil(REPOSITORIES_PER_SHARD);
+    let timestamp = timestamp_seconds as usize;
+    Some((timestamp / 3600) % number_of_shard)
+}
+
+/// Check if a repository should be processed based on sharding
+fn should_process_repository(
+    repo_index: usize,
+    current_shard: Option<usize>,
+    number_of_repo: usize,
+) -> bool {
+    let Some(current_shard) = current_shard else {
+        return true; // No sharding, process all repositories
+    };
+
+    let number_of_shard = number_of_repo.div_ceil(REPOSITORIES_PER_SHARD); // Math.ceil
+    repo_index % number_of_shard == current_shard
+}
+
 #[derive(Debug, Serialize, Deserialize, Clone)]
 pub enum BackgroundJobEvent {
     SchedulerGitRepository(CodeRepository),

ee/tabby-webserver/src/service/background_job/third_party_integration.rs

@@ -17,7 +17,10 @@ use tabby_schema::{
 };
 use tracing::debug;
 
-use super::{helper::Job, index_commits, BackgroundJobEvent};
+use super::{
+    calculate_current_shard, helper::Job, index_commits, should_process_repository,
+    BackgroundJobEvent,
+};
 
 pub mod error;
 mod issues;
@@ -242,17 +245,26 @@ impl SchedulerGithubGitlabJob {
     }
 
     pub async fn cron(
-        _now: DateTime<Utc>,
+        now: DateTime<Utc>,
         repository: Arc<dyn ThirdPartyRepositoryService>,
         job: Arc<dyn JobService>,
     ) -> tabby_schema::Result<()> {
-        for repository in repository
+        let repositories = repository
             .list_repositories_with_filter(None, None, Some(true), None, None, None, None)
-            .await?
-        {
+            .await?;
+
+        let number_of_repo = repositories.len();
+        let current_shard = calculate_current_shard(number_of_repo, now.timestamp());
+
+        for (i, repository) in repositories.iter().enumerate() {
+            if !should_process_repository(i, current_shard, number_of_repo) {
+                continue;
+            }
+
             let _ = job
                 .trigger(
-                    BackgroundJobEvent::SchedulerGithubGitlabRepository(repository.id).to_command(),
+                    BackgroundJobEvent::SchedulerGithubGitlabRepository(repository.id.clone())
+                        .to_command(),
                 )
                 .await;
         }