From 92236abf206337885f118df5f5f6326ad1782b50 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 29 Jul 2026 13:17:48 -0400 Subject: [PATCH 01/27] Comment out existing file replication --- backend/btrixcloud/background_jobs.py | 275 +++++++++--------- backend/btrixcloud/basecrawls.py | 82 +++--- .../migration_0052_profile_filenames.py | 21 +- backend/btrixcloud/operator/crawls.py | 3 +- backend/btrixcloud/profiles.py | 14 +- backend/btrixcloud/uploads.py | 5 +- 6 files changed, 198 insertions(+), 202 deletions(-) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index 732811085e..afef707565 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -99,136 +99,139 @@ def strip_bucket(self, endpoint_url: str) -> tuple[str, str]: parts.path[1:], "" ) - async def handle_replica_job_succeeded(self, job: CreateReplicaJob) -> None: - """Update replicas in corresponding file objects, based on type""" - res = None - if job.object_type in CRAWL_TYPES: - res = await self.base_crawl_ops.add_crawl_file_replica( - job.object_id, job.file_path, job.replica_storage - ) - elif job.object_type == "profile": - res = await self.profile_ops.add_profile_file_replica( - UUID(job.object_id), job.file_path, job.replica_storage - ) - if not res: - logger.debug( - "file_deleted_before_replication", - object_id=job.object_id, - file_path=job.file_path, - replica_storage=job.replica_storage, - oid=job.oid, - unstructured_message="File deleted before replication job started, ignoring", - ) + # TODO: Remove + # async def handle_replica_job_succeeded(self, job: CreateReplicaJob) -> None: + # """Update replicas in corresponding file objects, based on type""" + # res = None + # if job.object_type in CRAWL_TYPES: + # res = await self.base_crawl_ops.add_crawl_file_replica( + # job.object_id, job.file_path, job.replica_storage + # ) + # elif job.object_type == "profile": + # res = await self.profile_ops.add_profile_file_replica( + # UUID(job.object_id), job.file_path, job.replica_storage + # ) + # if not res: + # logger.debug( + # "file_deleted_before_replication", + # object_id=job.object_id, + # file_path=job.file_path, + # replica_storage=job.replica_storage, + # oid=job.oid, + # unstructured_message="File deleted before replication job started, ignoring", + # ) async def handle_delete_replica_job_finished(self, job: DeleteReplicaJob) -> None: """After successful replica deletion, delete cronjob if scheduled""" if job.schedule: await self.crawl_manager.delete_replica_deletion_scheduled_job(job.id) - async def create_replica_jobs( - self, oid: UUID, file: BaseFile, object_id: str, object_type: str - ) -> dict[str, bool | list[str]]: - """Create k8s background job to replicate a file to all replica storage locations.""" - org = await self.org_ops.get_org_by_id(oid) - - primary_storage = self.storage_ops.get_org_storage_by_ref(org, file.storage) - primary_endpoint, bucket_suffix = self.strip_bucket( - primary_storage.endpoint_url - ) - - primary_file_path = bucket_suffix + file.filename - - ids = [] - - for replica_ref in self.storage_ops.get_org_replicas_storage_refs(org): - job_id = await self.create_replica_job( - org, - file, - object_id, - object_type, - replica_ref, - primary_file_path, - primary_endpoint, - ) - ids.append(job_id) - - return {"added": True, "ids": ids} - - async def create_replica_job( - self, - org: Organization, - file: BaseFile, - object_id: str, - object_type: str, - replica_ref: StorageRef, - primary_file_path: str, - primary_endpoint: str, - existing_job_id: str | None = None, - ) -> str: - """Create k8s background job to replicate a file to a specific replica storage location.""" - replica_storage = self.storage_ops.get_org_storage_by_ref(org, replica_ref) - replica_endpoint, bucket_suffix = self.strip_bucket( - replica_storage.endpoint_url - ) - replica_file_path = bucket_suffix + file.filename - - job_type = BgJobType.CREATE_REPLICA.value - - try: - job_id, _ = await self.crawl_manager.run_replica_job( - oid=str(org.id), - job_type=job_type, - primary_storage=file.storage, - primary_file_path=primary_file_path, - primary_endpoint=primary_endpoint, - replica_storage=replica_ref, - replica_file_path=replica_file_path, - replica_endpoint=replica_endpoint, - delay_days=0, - existing_job_id=existing_job_id, - ) - if existing_job_id: - replication_job = await self.get_background_job(existing_job_id, org.id) - previous_attempt = { - "started": replication_job.started, - "finished": replication_job.finished, - } - if replication_job.previousAttempts: - replication_job.previousAttempts.append(previous_attempt) - else: - replication_job.previousAttempts = [previous_attempt] - replication_job.started = dt_now() - replication_job.finished = None - replication_job.success = None - else: - replication_job = CreateReplicaJob( - id=job_id, - oid=org.id, - started=dt_now(), - file_path=file.filename, - object_type=object_type, - object_id=object_id, - primary=file.storage, - replica_storage=replica_ref, - ) - - await self.jobs.find_one_and_update( - {"_id": job_id}, {"$set": replication_job.to_dict()}, upsert=True - ) - - return job_id - # pylint: disable=broad-exception-caught - except Exception as exc: - logger.warning( - "replica_job_start_failed", - object_type=object_type, - oid=org.id, - file=file, - exc_info=True, - unstructured_message=f"warning: replica job could not be started " - f"for {object_type} {file}: {exc}", - ) - return "" + # TODO: Remove + # async def create_replica_jobs( + # self, oid: UUID, file: BaseFile, object_id: str, object_type: str + # ) -> dict[str, bool | list[str]]: + # """Create k8s background job to replicate a file to all replica storage locations.""" + # org = await self.org_ops.get_org_by_id(oid) + + # primary_storage = self.storage_ops.get_org_storage_by_ref(org, file.storage) + # primary_endpoint, bucket_suffix = self.strip_bucket( + # primary_storage.endpoint_url + # ) + + # primary_file_path = bucket_suffix + file.filename + + # ids = [] + + # for replica_ref in self.storage_ops.get_org_replicas_storage_refs(org): + # job_id = await self.create_replica_job( + # org, + # file, + # object_id, + # object_type, + # replica_ref, + # primary_file_path, + # primary_endpoint, + # ) + # ids.append(job_id) + + # return {"added": True, "ids": ids} + + # TODO: Remove + # async def create_replica_job( + # self, + # org: Organization, + # file: BaseFile, + # object_id: str, + # object_type: str, + # replica_ref: StorageRef, + # primary_file_path: str, + # primary_endpoint: str, + # existing_job_id: str | None = None, + # ) -> str: + # """Create k8s background job to replicate a file to a specific replica storage location.""" + # replica_storage = self.storage_ops.get_org_storage_by_ref(org, replica_ref) + # replica_endpoint, bucket_suffix = self.strip_bucket( + # replica_storage.endpoint_url + # ) + # replica_file_path = bucket_suffix + file.filename + + # job_type = BgJobType.CREATE_REPLICA.value + + # try: + # job_id, _ = await self.crawl_manager.run_replica_job( + # oid=str(org.id), + # job_type=job_type, + # primary_storage=file.storage, + # primary_file_path=primary_file_path, + # primary_endpoint=primary_endpoint, + # replica_storage=replica_ref, + # replica_file_path=replica_file_path, + # replica_endpoint=replica_endpoint, + # delay_days=0, + # existing_job_id=existing_job_id, + # ) + # if existing_job_id: + # replication_job = await self.get_background_job(existing_job_id, org.id) + # previous_attempt = { + # "started": replication_job.started, + # "finished": replication_job.finished, + # } + # if replication_job.previousAttempts: + # replication_job.previousAttempts.append(previous_attempt) + # else: + # replication_job.previousAttempts = [previous_attempt] + # replication_job.started = dt_now() + # replication_job.finished = None + # replication_job.success = None + # else: + # replication_job = CreateReplicaJob( + # id=job_id, + # oid=org.id, + # started=dt_now(), + # file_path=file.filename, + # object_type=object_type, + # object_id=object_id, + # primary=file.storage, + # replica_storage=replica_ref, + # ) + + # await self.jobs.find_one_and_update( + # {"_id": job_id}, {"$set": replication_job.to_dict()}, upsert=True + # ) + + # return job_id + # # pylint: disable=broad-exception-caught + # except Exception as exc: + # logger.warning( + # "replica_job_start_failed", + # object_type=object_type, + # oid=org.id, + # file=file, + # exc_info=True, + # unstructured_message=f"warning: replica job could not be started " + # f"for {object_type} {file}: {exc}", + # ) + # return "" async def create_delete_replica_jobs( self, org: Organization, file: BaseFile, object_id: str, object_type: str @@ -688,8 +691,9 @@ async def job_finished( if job.type != job_type: raise HTTPException(status_code=400, detail="invalid_job_type") - if success and job_type == BgJobType.CREATE_REPLICA: - await self.handle_replica_job_succeeded(cast(CreateReplicaJob, job)) + # TODO: Remove + # if success and job_type == BgJobType.CREATE_REPLICA: + # await self.handle_replica_job_succeeded(cast(CreateReplicaJob, job)) if job_type == BgJobType.DELETE_REPLICA: await self.handle_delete_replica_job_finished(cast(DeleteReplicaJob, job)) @@ -902,24 +906,9 @@ async def retry_org_background_job( ) -> dict[str, bool | str | None]: """Retry background job specific to one org""" if job.type == BgJobType.CREATE_REPLICA: - job = cast(CreateReplicaJob, job) - file = await self.get_replica_job_file(job, org) - primary_storage = self.storage_ops.get_org_storage_by_ref(org, file.storage) - primary_endpoint, bucket_suffix = self.strip_bucket( - primary_storage.endpoint_url + raise HTTPException( + status_code=400, detail="create_replica_job_retry_not_supported" ) - primary_file_path = bucket_suffix + file.filename - await self.create_replica_job( - org, - file, - job.object_id, - job.object_type, - job.replica_storage, - primary_file_path, - primary_endpoint, - existing_job_id=job.id, - ) - return {"success": True} if job.type == BgJobType.DELETE_REPLICA: job = cast(DeleteReplicaJob, job) diff --git a/backend/btrixcloud/basecrawls.py b/backend/btrixcloud/basecrawls.py index f56461786d..0941e8c582 100644 --- a/backend/btrixcloud/basecrawls.py +++ b/backend/btrixcloud/basecrawls.py @@ -344,46 +344,48 @@ async def update_usernames(self, userid: UUID, updated_name: str) -> None: {"userid": userid}, {"$set": {"userName": updated_name}} ) - async def replicate_crawl_files( - self, crawl_id: str, org: Organization, type_: TYPE_CRAWL_TYPES - ): - """Replicate crawl files to configured replica locations""" - repl_logger = logger.bind(crawl_id=crawl_id, oid=org.id, type=type_) - - try: - crawl = await self.get_base_crawl(crawl_id, org, type_) - # pylint: disable=broad-exception-caught - except Exception: - repl_logger.warning( - "crawl_replicate_skipped_not_found", - unstructured_message=f"Not replicating files for crawl {crawl_id}: crawl not found", - ) - return - - for crawl_file in crawl.files: - try: - await self.background_job_ops.create_replica_jobs( - crawl.oid, crawl_file, crawl.id, type_ - ) - # pylint: disable=broad-exception-caught - except Exception as exc: - repl_logger.exception( - "crawl_replicate_failed", - unstructured_message=f"Replicate Exception {exc}", - ) - - async def add_crawl_file_replica( - self, crawl_id: str, filename: str, ref: StorageRef - ) -> dict[str, object]: - """Add replica StorageRef to existing CrawlFile""" - return await self.crawls.find_one_and_update( - {"_id": crawl_id, "files.filename": filename}, - { - "$addToSet": { - "files.$.replicas": {"name": ref.name, "custom": ref.custom} - } - }, - ) + # TODO: Remove + # async def replicate_crawl_files( + # self, crawl_id: str, org: Organization, type_: TYPE_CRAWL_TYPES + # ): + # """Replicate crawl files to configured replica locations""" + # repl_logger = logger.bind(crawl_id=crawl_id, oid=org.id, type=type_) + + # try: + # crawl = await self.get_base_crawl(crawl_id, org, type_) + # # pylint: disable=broad-exception-caught + # except Exception: + # repl_logger.warning( + # "crawl_replicate_skipped_not_found", + # unstructured_message=f"Not replicating files for crawl {crawl_id}: crawl not found", + # ) + # return + + # for crawl_file in crawl.files: + # try: + # await self.background_job_ops.create_replica_jobs( + # crawl.oid, crawl_file, crawl.id, type_ + # ) + # # pylint: disable=broad-exception-caught + # except Exception as exc: + # repl_logger.exception( + # "crawl_replicate_failed", + # unstructured_message=f"Replicate Exception {exc}", + # ) + + # TODO: Remove + # async def add_crawl_file_replica( + # self, crawl_id: str, filename: str, ref: StorageRef + # ) -> dict[str, object]: + # """Add replica StorageRef to existing CrawlFile""" + # return await self.crawls.find_one_and_update( + # {"_id": crawl_id, "files.filename": filename}, + # { + # "$addToSet": { + # "files.$.replicas": {"name": ref.name, "custom": ref.custom} + # } + # }, + # ) async def shutdown_crawl(self, crawl_id: str, org: Organization, graceful: bool): """placeholder, implemented in crawls, base version does nothing""" diff --git a/backend/btrixcloud/migrations/migration_0052_profile_filenames.py b/backend/btrixcloud/migrations/migration_0052_profile_filenames.py index 04876450e9..0f1542b8cc 100644 --- a/backend/btrixcloud/migrations/migration_0052_profile_filenames.py +++ b/backend/btrixcloud/migrations/migration_0052_profile_filenames.py @@ -67,16 +67,17 @@ async def migrate_up(self) -> None: profile.resource.filename = new_filename profile.resource.replicas = [] - logger.info( - "profile_replication_job_started", - profile_id=profile.id, - unstructured_message=( - f"Starting background jobs to replicate profile {profile.id}" - ), - ) - await self.background_job_ops.create_replica_jobs( - profile.oid, profile.resource, str(profile.id), "profile" - ) + # TODO: Remove + # logger.info( + # "profile_replication_job_started", + # profile_id=profile.id, + # unstructured_message=( + # f"Starting background jobs to replicate profile {profile.id}" + # ), + # ) + # await self.background_job_ops.create_replica_jobs( + # profile.oid, profile.resource, str(profile.id), "profile" + # ) # pylint: disable=broad-exception-caught except Exception: logger.exception( diff --git a/backend/btrixcloud/operator/crawls.py b/backend/btrixcloud/operator/crawls.py index 3afc76d4c1..55a0c433a9 100644 --- a/backend/btrixcloud/operator/crawls.py +++ b/backend/btrixcloud/operator/crawls.py @@ -2202,7 +2202,8 @@ async def do_crawl_finished_tasks( await self.coll_ops.add_successful_crawl_to_collections( crawl.id, crawl.cid, crawl.oid ) - await self.crawl_ops.replicate_crawl_files(crawl.id, crawl.org, "crawl") + # TODO: Remove + # await self.crawl_ops.replicate_crawl_files(crawl.id, crawl.org, "crawl") if stats and stats.profile_update and crawl.profileid: await self.crawl_config_ops.profiles.update_profile_from_crawl_upload( diff --git a/backend/btrixcloud/profiles.py b/backend/btrixcloud/profiles.py index 49ce65d4fc..42c8c4409b 100644 --- a/backend/btrixcloud/profiles.py +++ b/backend/btrixcloud/profiles.py @@ -351,9 +351,10 @@ async def do_commit_to_profile( {"_id": profile.id}, {"$set": profile.to_dict()}, upsert=True ) - await self.background_job_ops.create_replica_jobs( - org.id, profile_file, str(profileid), "profile" - ) + # TODO: Remove + # await self.background_job_ops.create_replica_jobs( + # org.id, profile_file, str(profileid), "profile" + # ) await self.orgs.inc_org_bytes_stored( org.id, file_size - prev_file_size, "profile" @@ -454,9 +455,10 @@ async def update_profile_from_crawl_upload( profile_file.hash = hash_ # update replica - await self.background_job_ops.create_replica_jobs( - org_id, profile_file, str(profileid), "profile" - ) + # TODO: Remove + # await self.background_job_ops.create_replica_jobs( + # org_id, profile_file, str(profileid), "profile" + # ) # update size stats, if changed if size != prev_file_size: diff --git a/backend/btrixcloud/uploads.py b/backend/btrixcloud/uploads.py index 1f8d0792dd..dc7ecb2557 100644 --- a/backend/btrixcloud/uploads.py +++ b/backend/btrixcloud/uploads.py @@ -280,6 +280,7 @@ async def _create_upload( upload_logger.debug( "upload_create", state="completed", quota_reached=quota_reached ) + return {"id": crawl_id, "added": True, "storageQuotaReached": quota_reached} async def post_process_upload( @@ -393,8 +394,8 @@ async def _check_and_maybe_split(file: CrawlFile) -> None: {"_id": crawl_id}, {"$set": {"state": "complete"}} ) - pp_logger.debug("post_process_upload", state="replicate_crawl_files") - await self.replicate_crawl_files(crawl_id, org, "upload") + # pp_logger.debug("post_process_upload", state="replicate_crawl_files") + # await self.replicate_crawl_files(crawl_id, org, "upload") pp_logger.debug( "post_process_upload", state="finished_processing_dispatching_webhook" From 1168b087bc625d0ed9234e84a4b0bcabca57c9eb Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 29 Jul 2026 13:29:12 -0400 Subject: [PATCH 02/27] Remove existing file replication code --- backend/btrixcloud/background_jobs.py | 133 ------------------ backend/btrixcloud/basecrawls.py | 43 ------ .../migration_0052_profile_filenames.py | 18 --- backend/btrixcloud/operator/crawls.py | 2 - backend/btrixcloud/profiles.py | 14 -- backend/btrixcloud/uploads.py | 3 - 6 files changed, 213 deletions(-) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index afef707565..735082fa40 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -99,140 +99,11 @@ def strip_bucket(self, endpoint_url: str) -> tuple[str, str]: parts.path[1:], "" ) - # TODO: Remove - # async def handle_replica_job_succeeded(self, job: CreateReplicaJob) -> None: - # """Update replicas in corresponding file objects, based on type""" - # res = None - # if job.object_type in CRAWL_TYPES: - # res = await self.base_crawl_ops.add_crawl_file_replica( - # job.object_id, job.file_path, job.replica_storage - # ) - # elif job.object_type == "profile": - # res = await self.profile_ops.add_profile_file_replica( - # UUID(job.object_id), job.file_path, job.replica_storage - # ) - # if not res: - # logger.debug( - # "file_deleted_before_replication", - # object_id=job.object_id, - # file_path=job.file_path, - # replica_storage=job.replica_storage, - # oid=job.oid, - # unstructured_message="File deleted before replication job started, ignoring", - # ) - async def handle_delete_replica_job_finished(self, job: DeleteReplicaJob) -> None: """After successful replica deletion, delete cronjob if scheduled""" if job.schedule: await self.crawl_manager.delete_replica_deletion_scheduled_job(job.id) - # TODO: Remove - # async def create_replica_jobs( - # self, oid: UUID, file: BaseFile, object_id: str, object_type: str - # ) -> dict[str, bool | list[str]]: - # """Create k8s background job to replicate a file to all replica storage locations.""" - # org = await self.org_ops.get_org_by_id(oid) - - # primary_storage = self.storage_ops.get_org_storage_by_ref(org, file.storage) - # primary_endpoint, bucket_suffix = self.strip_bucket( - # primary_storage.endpoint_url - # ) - - # primary_file_path = bucket_suffix + file.filename - - # ids = [] - - # for replica_ref in self.storage_ops.get_org_replicas_storage_refs(org): - # job_id = await self.create_replica_job( - # org, - # file, - # object_id, - # object_type, - # replica_ref, - # primary_file_path, - # primary_endpoint, - # ) - # ids.append(job_id) - - # return {"added": True, "ids": ids} - - # TODO: Remove - # async def create_replica_job( - # self, - # org: Organization, - # file: BaseFile, - # object_id: str, - # object_type: str, - # replica_ref: StorageRef, - # primary_file_path: str, - # primary_endpoint: str, - # existing_job_id: str | None = None, - # ) -> str: - # """Create k8s background job to replicate a file to a specific replica storage location.""" - # replica_storage = self.storage_ops.get_org_storage_by_ref(org, replica_ref) - # replica_endpoint, bucket_suffix = self.strip_bucket( - # replica_storage.endpoint_url - # ) - # replica_file_path = bucket_suffix + file.filename - - # job_type = BgJobType.CREATE_REPLICA.value - - # try: - # job_id, _ = await self.crawl_manager.run_replica_job( - # oid=str(org.id), - # job_type=job_type, - # primary_storage=file.storage, - # primary_file_path=primary_file_path, - # primary_endpoint=primary_endpoint, - # replica_storage=replica_ref, - # replica_file_path=replica_file_path, - # replica_endpoint=replica_endpoint, - # delay_days=0, - # existing_job_id=existing_job_id, - # ) - # if existing_job_id: - # replication_job = await self.get_background_job(existing_job_id, org.id) - # previous_attempt = { - # "started": replication_job.started, - # "finished": replication_job.finished, - # } - # if replication_job.previousAttempts: - # replication_job.previousAttempts.append(previous_attempt) - # else: - # replication_job.previousAttempts = [previous_attempt] - # replication_job.started = dt_now() - # replication_job.finished = None - # replication_job.success = None - # else: - # replication_job = CreateReplicaJob( - # id=job_id, - # oid=org.id, - # started=dt_now(), - # file_path=file.filename, - # object_type=object_type, - # object_id=object_id, - # primary=file.storage, - # replica_storage=replica_ref, - # ) - - # await self.jobs.find_one_and_update( - # {"_id": job_id}, {"$set": replication_job.to_dict()}, upsert=True - # ) - - # return job_id - # # pylint: disable=broad-exception-caught - # except Exception as exc: - # logger.warning( - # "replica_job_start_failed", - # object_type=object_type, - # oid=org.id, - # file=file, - # exc_info=True, - # unstructured_message=f"warning: replica job could not be started " - # f"for {object_type} {file}: {exc}", - # ) - # return "" - async def create_delete_replica_jobs( self, org: Organization, file: BaseFile, object_id: str, object_type: str ) -> dict[str, bool | list[str]]: @@ -691,10 +562,6 @@ async def job_finished( if job.type != job_type: raise HTTPException(status_code=400, detail="invalid_job_type") - # TODO: Remove - # if success and job_type == BgJobType.CREATE_REPLICA: - # await self.handle_replica_job_succeeded(cast(CreateReplicaJob, job)) - if job_type == BgJobType.DELETE_REPLICA: await self.handle_delete_replica_job_finished(cast(DeleteReplicaJob, job)) diff --git a/backend/btrixcloud/basecrawls.py b/backend/btrixcloud/basecrawls.py index 0941e8c582..d759ea46c9 100644 --- a/backend/btrixcloud/basecrawls.py +++ b/backend/btrixcloud/basecrawls.py @@ -344,49 +344,6 @@ async def update_usernames(self, userid: UUID, updated_name: str) -> None: {"userid": userid}, {"$set": {"userName": updated_name}} ) - # TODO: Remove - # async def replicate_crawl_files( - # self, crawl_id: str, org: Organization, type_: TYPE_CRAWL_TYPES - # ): - # """Replicate crawl files to configured replica locations""" - # repl_logger = logger.bind(crawl_id=crawl_id, oid=org.id, type=type_) - - # try: - # crawl = await self.get_base_crawl(crawl_id, org, type_) - # # pylint: disable=broad-exception-caught - # except Exception: - # repl_logger.warning( - # "crawl_replicate_skipped_not_found", - # unstructured_message=f"Not replicating files for crawl {crawl_id}: crawl not found", - # ) - # return - - # for crawl_file in crawl.files: - # try: - # await self.background_job_ops.create_replica_jobs( - # crawl.oid, crawl_file, crawl.id, type_ - # ) - # # pylint: disable=broad-exception-caught - # except Exception as exc: - # repl_logger.exception( - # "crawl_replicate_failed", - # unstructured_message=f"Replicate Exception {exc}", - # ) - - # TODO: Remove - # async def add_crawl_file_replica( - # self, crawl_id: str, filename: str, ref: StorageRef - # ) -> dict[str, object]: - # """Add replica StorageRef to existing CrawlFile""" - # return await self.crawls.find_one_and_update( - # {"_id": crawl_id, "files.filename": filename}, - # { - # "$addToSet": { - # "files.$.replicas": {"name": ref.name, "custom": ref.custom} - # } - # }, - # ) - async def shutdown_crawl(self, crawl_id: str, org: Organization, graceful: bool): """placeholder, implemented in crawls, base version does nothing""" diff --git a/backend/btrixcloud/migrations/migration_0052_profile_filenames.py b/backend/btrixcloud/migrations/migration_0052_profile_filenames.py index 0f1542b8cc..e845c0e632 100644 --- a/backend/btrixcloud/migrations/migration_0052_profile_filenames.py +++ b/backend/btrixcloud/migrations/migration_0052_profile_filenames.py @@ -26,8 +26,6 @@ async def migrate_up(self) -> None: """Perform migration up. Add oid prefix to profile resource filenames that don't already have it. - For any profiles that match, also delete the database record for any - existing replicas and then spawn new replication jobs. """ profiles_mdb = self.mdb["profiles"] @@ -59,25 +57,9 @@ async def migrate_up(self) -> None: { "$set": { "resource.filename": new_filename, - "resource.replicas": [], } }, ) - - profile.resource.filename = new_filename - profile.resource.replicas = [] - - # TODO: Remove - # logger.info( - # "profile_replication_job_started", - # profile_id=profile.id, - # unstructured_message=( - # f"Starting background jobs to replicate profile {profile.id}" - # ), - # ) - # await self.background_job_ops.create_replica_jobs( - # profile.oid, profile.resource, str(profile.id), "profile" - # ) # pylint: disable=broad-exception-caught except Exception: logger.exception( diff --git a/backend/btrixcloud/operator/crawls.py b/backend/btrixcloud/operator/crawls.py index 55a0c433a9..f70cb41c42 100644 --- a/backend/btrixcloud/operator/crawls.py +++ b/backend/btrixcloud/operator/crawls.py @@ -2202,8 +2202,6 @@ async def do_crawl_finished_tasks( await self.coll_ops.add_successful_crawl_to_collections( crawl.id, crawl.cid, crawl.oid ) - # TODO: Remove - # await self.crawl_ops.replicate_crawl_files(crawl.id, crawl.org, "crawl") if stats and stats.profile_update and crawl.profileid: await self.crawl_config_ops.profiles.update_profile_from_crawl_upload( diff --git a/backend/btrixcloud/profiles.py b/backend/btrixcloud/profiles.py index 42c8c4409b..d76e180c7d 100644 --- a/backend/btrixcloud/profiles.py +++ b/backend/btrixcloud/profiles.py @@ -351,11 +351,6 @@ async def do_commit_to_profile( {"_id": profile.id}, {"$set": profile.to_dict()}, upsert=True ) - # TODO: Remove - # await self.background_job_ops.create_replica_jobs( - # org.id, profile_file, str(profileid), "profile" - # ) - await self.orgs.inc_org_bytes_stored( org.id, file_size - prev_file_size, "profile" ) @@ -451,15 +446,6 @@ async def update_profile_from_crawl_upload( if profile_file.hash == hash_: return True - profile_file.size = size - profile_file.hash = hash_ - - # update replica - # TODO: Remove - # await self.background_job_ops.create_replica_jobs( - # org_id, profile_file, str(profileid), "profile" - # ) - # update size stats, if changed if size != prev_file_size: await self.orgs.inc_org_bytes_stored( diff --git a/backend/btrixcloud/uploads.py b/backend/btrixcloud/uploads.py index dc7ecb2557..4ea5c19c79 100644 --- a/backend/btrixcloud/uploads.py +++ b/backend/btrixcloud/uploads.py @@ -394,9 +394,6 @@ async def _check_and_maybe_split(file: CrawlFile) -> None: {"_id": crawl_id}, {"$set": {"state": "complete"}} ) - # pp_logger.debug("post_process_upload", state="replicate_crawl_files") - # await self.replicate_crawl_files(crawl_id, org, "upload") - pp_logger.debug( "post_process_upload", state="finished_processing_dispatching_webhook" ) From d7f219c0c0da8d413a7bc2622962c6e89a1fd9ec Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 29 Jul 2026 13:35:39 -0400 Subject: [PATCH 03/27] Modify generic replica_job template to only be for deletion --- backend/btrixcloud/background_jobs.py | 5 +-- backend/btrixcloud/crawlmanager.py | 9 +++-- ...plica_job.yaml => delete_replica_job.yaml} | 39 +------------------ 3 files changed, 7 insertions(+), 46 deletions(-) rename chart/app-templates/{replica_job.yaml => delete_replica_job.yaml} (59%) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index 735082fa40..4bdcae3f99 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -137,15 +137,12 @@ async def create_delete_replica_job( ) replica_file_path = bucket_suffix + file.filename - job_type = BgJobType.DELETE_REPLICA.value - delay_days = int(os.environ.get("REPLICA_DELETION_DELAY_DAYS", 0)) if force_start_immediately: delay_days = 0 - job_id, schedule = await self.crawl_manager.run_replica_job( + job_id, schedule = await self.crawl_manager.run_delete_replica_job( oid=str(org.id), - job_type=job_type, replica_storage=replica_ref, replica_file_path=replica_file_path, replica_endpoint=replica_endpoint, diff --git a/backend/btrixcloud/crawlmanager.py b/backend/btrixcloud/crawlmanager.py index e71963d085..0edfd7566a 100644 --- a/backend/btrixcloud/crawlmanager.py +++ b/backend/btrixcloud/crawlmanager.py @@ -81,10 +81,9 @@ async def run_profile_browser( return browserid - async def run_replica_job( + async def run_delete_replica_job( self, oid: str, - job_type: str, replica_storage: StorageRef, replica_file_path: str, replica_endpoint: str, @@ -96,6 +95,8 @@ async def run_replica_job( ) -> tuple[str, str | None]: """run job to replicate file from primary storage to replica storage""" + job_type = BgJobType.DELETE_REPLICA.value + if existing_job_id: job_id = existing_job_id else: @@ -119,13 +120,13 @@ async def run_replica_job( "BgJobType": BgJobType, } - if job_type == BgJobType.DELETE_REPLICA.value and delay_days > 0: + if delay_days > 0: # If replica deletion delay is configured, schedule as cronjob return await self.create_replica_deletion_scheduled_job( job_id, params, delay_days ) - data = self.templates.env.get_template("replica_job.yaml").render(params) + data = self.templates.env.get_template("delete_replica_job.yaml").render(params) await self.create_from_yaml(data) diff --git a/chart/app-templates/replica_job.yaml b/chart/app-templates/delete_replica_job.yaml similarity index 59% rename from chart/app-templates/replica_job.yaml rename to chart/app-templates/delete_replica_job.yaml index f902ed873c..b623fe7ba3 100644 --- a/chart/app-templates/replica_job.yaml +++ b/chart/app-templates/delete_replica_job.yaml @@ -27,40 +27,6 @@ spec: - name: rclone image: rclone/rclone:latest env: - -{% if job_type == BgJobType.CREATE_REPLICA %} - - name: RCLONE_CONFIG_PRIMARY_TYPE - value: "s3" - - - name: RCLONE_CONFIG_PRIMARY_ACCESS_KEY_ID - valueFrom: - secretKeyRef: - name: "{{ primary_secret_name }}" - key: STORE_ACCESS_KEY - - - name: RCLONE_CONFIG_PRIMARY_SECRET_ACCESS_KEY - valueFrom: - secretKeyRef: - name: "{{ primary_secret_name }}" - key: STORE_SECRET_KEY - - - name: RCLONE_CONFIG_PRIMARY_REGION - valueFrom: - secretKeyRef: - name: "{{ primary_secret_name }}" - key: STORE_REGION - - - name: RCLONE_CONFIG_PRIMARY_PROVIDER - valueFrom: - secretKeyRef: - name: "{{ primary_secret_name }}" - key: STORE_S3_PROVIDER - - - name: RCLONE_CONFIG_PRIMARY_ENDPOINT - value: "{{ primary_endpoint }}" - -{% endif %} - - name: RCLONE_CONFIG_REPLICA_TYPE value: "s3" @@ -91,11 +57,8 @@ spec: - name: RCLONE_CONFIG_REPLICA_ENDPOINT value: "{{ replica_endpoint }}" -{% if job_type == BgJobType.CREATE_REPLICA %} - command: ["rclone", "-vv", "copyto", "--checksum", "--error-on-no-transfer", "primary:{{ primary_file_path }}", "replica:{{ replica_file_path }}"] -{% elif job_type == BgJobType.DELETE_REPLICA %} command: ["rclone", "-vv", "delete", "replica:{{ replica_file_path }}"] -{% endif %} + resources: limits: memory: "200Mi" From db1c31d5eec00815fbac33cd52de460b5ac258a0 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 29 Jul 2026 13:41:31 -0400 Subject: [PATCH 04/27] Remove unused CRAWL_TYPES import --- backend/btrixcloud/background_jobs.py | 1 - 1 file changed, 1 deletion(-) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index 4bdcae3f99..d2c2aff5f4 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -14,7 +14,6 @@ from .crawlmanager import CrawlManager from .models import ( - CRAWL_TYPES, AnyJob, BackgroundJob, BaseFile, From 30ff192948800398c0eb230b7a84d0f96b4abaf5 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 29 Jul 2026 16:24:13 -0400 Subject: [PATCH 05/27] Add background cron job to spin up rclone jobs - no rclone jobs yet --- backend/btrixcloud/background_jobs.py | 21 ++++++++- backend/btrixcloud/crawlmanager.py | 63 +++++++++++++++++++++++++++ backend/btrixcloud/main_bg.py | 10 +++++ backend/btrixcloud/models.py | 13 +++++- chart/templates/configmap.yaml | 2 + chart/values.yaml | 4 ++ 6 files changed, 110 insertions(+), 3 deletions(-) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index d2c2aff5f4..10e79a8183 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -29,6 +29,7 @@ ReAddOrgPagesJob, RecalculateOrgStatsJob, RetryStuckUploadsJob, + ReplicateFilesCronJob, StorageRef, SuccessResponse, SuccessResponseId, @@ -509,6 +510,7 @@ async def ensure_cron_jobs_exist(self): """Ensure periodic background cron jobs exist""" await self.crawl_manager.ensure_cleanup_seed_file_cron_job_exists() await self.crawl_manager.ensure_retry_stuck_uploads_cron_job_exists() + await self.crawl_manager.ensure_file_replication_cron_job_exists() async def job_finished( self, @@ -524,7 +526,11 @@ async def job_finished( # For periodic cron jobs, no database record will exist for each # run before this point, so create it here - if job_type in (BgJobType.CLEANUP_SEED_FILES, BgJobType.RETRY_STUCK_UPLOADS): + if job_type in ( + BgJobType.CLEANUP_SEED_FILES, + BgJobType.RETRY_STUCK_UPLOADS, + BgJobType.REPLICATE_FILES_CRON + ): if not started: started = finished if job_type == BgJobType.CLEANUP_SEED_FILES: @@ -534,6 +540,13 @@ async def job_finished( finished=finished, success=success, ) + if job_type == BgJobType.REPLICATE_FILES_CRON: + cron_job = ReplicateFilesCronJob( + id=f"replicate-cron-{secrets.token_hex(5)}", + started=started, + finished=finished, + success=success, + ) else: cron_job = RetryStuckUploadsJob( id=f"stuck-uploads-{secrets.token_hex(5)}", @@ -607,6 +620,7 @@ async def get_background_job( | UpdateCollStatsJob | PostProcessUploadJob | RetryStuckUploadsJob + | ReplicateFilesCronJob ): """Get background job""" query: dict[str, object] = {"_id": job_id} @@ -649,6 +663,9 @@ def _get_job_by_type_from_data(self, data: dict[str, object]): if data["type"] == BgJobType.RETRY_STUCK_UPLOADS: return RetryStuckUploadsJob.from_dict(data) + if data["type"] == BgJobType.REPLICATE_FILES_CRON: + return ReplicateFilesCronJob.from_dict(data) + if data["type"] == BgJobType.DELETE_ORG: return DeleteOrgJob.from_dict(data) @@ -831,7 +848,7 @@ async def retry_org_background_job( ) return {"success": True} - if job.type == BgJobType.CLEANUP_SEED_FILES: + if job.type in (BgJobType.CLEANUP_SEED_FILES, BgJobType.REPLICATE_FILES_CRON): raise HTTPException(status_code=400, detail="cron_job_retry_not_supported") return {"success": False} diff --git a/backend/btrixcloud/crawlmanager.py b/backend/btrixcloud/crawlmanager.py index 0edfd7566a..55cda10171 100644 --- a/backend/btrixcloud/crawlmanager.py +++ b/backend/btrixcloud/crawlmanager.py @@ -414,6 +414,69 @@ async def _ensure_bg_cron_job_exists( cron_logger.info("bg_cron_job_creating") await self.create_from_yaml(data, namespace=DEFAULT_NAMESPACE) + async def ensure_file_replication_cron_job_exists(self): + """ensure cron background job to periodically replicate all files exists + + Each time the background job runs, it will spawn one additional background + job for each configured replica storage location, and run `rclone copy` from + primary storage to each replica location to keep them in sync without the + possibility of deleting anything from the destination. + + Deletions are handled separately with individual jobs and are subject to + the replica deletion delay. + """ + + job_id = "replicate-files-cron" + + # Default schedule is every 2 hours + default_schedule = "0 */2 * * *" + job_schedule = os.environ.get("REPLICATION_JOB_CRON_SCHEDULE", default_schedule) + + # Don't create a duplicate cron job if already exists + replication_logger = logger.bind(schedule=job_schedule) + + try: + cron_job = await self.batch_api.read_namespaced_cron_job( + name=job_id, + namespace=self.namespace, + ) + if cron_job: + replication_logger.info("replication_cron_job_exists") + + if cron_job.spec.schedule != job_schedule: + cron_job.spec.schedule = job_schedule + + await self.batch_api.patch_namespaced_cron_job( + name=cron_job.metadata.name, + namespace=self.namespace, + body=cron_job, + ) + replication_logger.info( + "replication_cron_job_updated", + prev_schedule=cron_job.spec.schedule, + ) + return + # pylint: disable=broad-exception-caught + except Exception: + pass + + replication_logger.info("replication_cron_job_creating") + + params = { + "id": job_id, + "job_type": BgJobType.REPLICATE_FILES_CRON.value, + "backend_image": os.environ.get("BACKEND_IMAGE", ""), + "pull_policy": os.environ.get("BACKEND_IMAGE_PULL_POLICY", ""), + "schedule": job_schedule, + "larger_resources": False, + } + + data = self.templates.env.get_template("background_cron_job.yaml").render( + params + ) + + await self.create_from_yaml(data) + async def create_crawl_job( self, crawlconfig: CrawlConfig, diff --git a/backend/btrixcloud/main_bg.py b/backend/btrixcloud/main_bg.py index 1930e86d09..21866d0d0b 100644 --- a/backend/btrixcloud/main_bg.py +++ b/backend/btrixcloud/main_bg.py @@ -107,6 +107,16 @@ async def main(): ) return ExitCode.ERROR + if job_type == BgJobType.REPLICATE_FILES_CRON: + try: + # Call method to launch actual rclone copy jobs - from storage_ops? + crawl_logger.info("replicate_files_cron_reached") + return ExitCode.SUCCESS + # pylint: disable=broad-exception-caught + except Exception: + crawl_logger.exception("bg_job_failed") + return ExitCode.ERROR + # Run job (org-specific) if not oid: crawl_logger.error( diff --git a/backend/btrixcloud/models.py b/backend/btrixcloud/models.py index ec5e78cb87..8de9d15827 100644 --- a/backend/btrixcloud/models.py +++ b/backend/btrixcloud/models.py @@ -3239,7 +3239,6 @@ class WebhookNotification(BaseMongoModel): class BgJobType(StrEnum): """Background Job Types""" - CREATE_REPLICA = "create-replica" DELETE_REPLICA = "delete-replica" DELETE_ORG = "delete-org" RECALCULATE_ORG_STATS = "recalculate-org-stats" @@ -3249,6 +3248,10 @@ class BgJobType(StrEnum): UPDATE_COLL_STATS = "update-coll-stats" POSTPROCESS_UPLOAD = "postprocess-upload" RETRY_STUCK_UPLOADS = "retry-stuck-uploads" + # Deprecated per-file replication jobs + CREATE_REPLICA = "create-replica" + # New replication job that spins up rclone copy jobs as needed + REPLICATE_FILES_CRON = "replicate-files-cron" # ============================================================================ @@ -3334,6 +3337,13 @@ class CleanupSeedFilesJob(BackgroundJob): type: Literal[BgJobType.CLEANUP_SEED_FILES] = BgJobType.CLEANUP_SEED_FILES +# ============================================================================ +class ReplicateFilesCronJob(BackgroundJob): + """Model for tracking cron jobs to replicate files""" + + type: Literal[BgJobType.REPLICATE_FILES_CRON] = BgJobType.REPLICATE_FILES_CRON + + # ============================================================================ class UpdateCollStatsJob(BackgroundJob): """Model for tracking jobs to readd pages for an org or single crawl""" @@ -3374,6 +3384,7 @@ class RetryStuckUploadsJob(BackgroundJob): | UpdateCollStatsJob | PostProcessUploadJob | RetryStuckUploadsJob + | ReplicateFilesCronJob ] diff --git a/chart/templates/configmap.yaml b/chart/templates/configmap.yaml index ebe29cbbab..ea50ca254d 100644 --- a/chart/templates/configmap.yaml +++ b/chart/templates/configmap.yaml @@ -121,6 +121,8 @@ data: CLEANUP_JOB_CRON_SCHEDULE: "{{ .Values.cleanup_job_cron_schedule }}" + REPLICATION_JOB_CRON_SCHEDULE: "{{ .Values.replication_job_cron_schedule }}" + CLEANUP_FILES_AFTER_MINUTES: "{{ .Values.cleanup_files_after_minutes | default 1440 }}" RETRY_STUCK_UPLOADS_CRON_SCHEDULE: "{{ .Values.stuck_uploads_cron_schedule }}" diff --git a/chart/values.yaml b/chart/values.yaml index 5abd71e493..72fb8706aa 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -192,6 +192,10 @@ cleanup_files_after_minutes: 1440 # without a running background job and retrying them stuck_uploads_cron_schedule: "0 * * * *" +# Cron schedule for periodically syncing replica storage +# locations from primary storage +replication_job_cron_schedule: "" + # Emails Image # ========================================= emails_image: "docker.io/webrecorder/browsertrix-emails:1.25.1" From 1d61c30e8c3beacd2894ce3271d14917f4878140 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 29 Jul 2026 16:29:04 -0400 Subject: [PATCH 06/27] Add missing default schedule --- chart/values.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/chart/values.yaml b/chart/values.yaml index 72fb8706aa..d565bca138 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -194,7 +194,7 @@ stuck_uploads_cron_schedule: "0 * * * *" # Cron schedule for periodically syncing replica storage # locations from primary storage -replication_job_cron_schedule: "" +replication_job_cron_schedule: "0 */2 * * *" # Emails Image # ========================================= From d6897db5988d7024c7dc931148d540a6b68f9b6f Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Fri, 31 Jul 2026 13:03:48 -0400 Subject: [PATCH 07/27] Add rclone copy bucket job and fix cron job Also removes unused arguments from run_delete_replica_job --- backend/btrixcloud/background_jobs.py | 124 ++++++++++++++++++- backend/btrixcloud/crawlmanager.py | 54 +++++--- backend/btrixcloud/main_bg.py | 5 +- backend/btrixcloud/models.py | 13 +- backend/btrixcloud/storages.py | 17 +++ chart/app-templates/background_cron_job.yaml | 2 +- chart/app-templates/copy_bucket_job.yaml | 95 ++++++++++++++ chart/app-templates/delete_replica_job.yaml | 2 +- chart/values.yaml | 4 +- 9 files changed, 293 insertions(+), 23 deletions(-) create mode 100644 chart/app-templates/copy_bucket_job.yaml diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index 10e79a8183..acb797c8c2 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -19,6 +19,7 @@ BaseFile, BgJobType, CleanupSeedFilesJob, + CopyBucketJob, CreateReplicaJob, DeleteOrgJob, DeleteReplicaJob, @@ -196,6 +197,95 @@ async def create_delete_replica_job( ) return "" + async def create_copy_bucket_jobs(self): + """Create background jobs to copy primary storage to each default replica location + + Note that this replicates default storages only, and not any org-specific + custom storage, which is not yet fully supported by Browsertrix. When custom + storage support is added, we will need to spin up additional copy jobs. + + Because rclone copy is only additive, these copy bucket jobs will ensure + that all files in primary storage also exist in the configured replica + locations without being able to delete any files from the replica location. + """ + primary_storage_ref = self.storage_ops.get_default_primary() + primary_storage = self.storage_ops.get_default_s3_storage(primary_storage_ref) + primary_endpoint, primary_bucket_suffix = self.strip_bucket( + primary_storage.endpoint_url + ) + + for default_replica_ref in self.storage_ops.get_default_replicas(): + await self.create_copy_bucket_job( + primary_storage_ref, + primary_endpoint, + primary_bucket_suffix, + default_replica_ref, + ) + + async def create_copy_bucket_job( + self, + primary_storage_ref: StorageRef, + primary_endpoint: str, + primary_bucket_suffix: str, + replica_ref: StorageRef, + existing_job_id: str | None = None, + ) -> str: + """Create background job to copy contents of bucket to replica location""" + replica_logger = logger.bind( + primary_storage=primary_storage_ref, replica_storage=replica_ref + ) + + try: + replica_storage = self.storage_ops.get_default_s3_storage(replica_ref) + replica_endpoint, replica_bucket_suffix = self.strip_bucket( + replica_storage.endpoint_url + ) + + job_id = await self.crawl_manager.run_copy_bucket_job( + primary_storage=primary_storage_ref, + replica_storage=replica_ref, + primary_endpoint=primary_endpoint, + primary_bucket_suffix=primary_bucket_suffix, + replica_endpoint=replica_endpoint, + replica_bucket_suffix=replica_bucket_suffix, + existing_job_id=existing_job_id, + ) + if existing_job_id: + copy_bucket_job = await self.get_background_job(existing_job_id) + previous_attempt = { + "started": copy_bucket_job.started, + "finished": copy_bucket_job.finished, + } + if copy_bucket_job.previousAttempts: + copy_bucket_job.previousAttempts.append(previous_attempt) + else: + copy_bucket_job.previousAttempts = [previous_attempt] + copy_bucket_job.started = dt_now() + copy_bucket_job.finished = None + copy_bucket_job.success = None + else: + copy_bucket_job = CopyBucketJob( + id=job_id, + started=dt_now(), + replica_storage=replica_ref, + ) + + await self.jobs.find_one_and_update( + {"_id": job_id}, {"$set": copy_bucket_job.to_dict()}, upsert=True + ) + + replica_logger.info("copy_bucket_job_started", job_id=job_id) + + return job_id + # pylint: disable=broad-exception-caught + except Exception as exc: + replica_logger.warning( + "copy_bucket_job_start_failed", + exc_info=True, + existing_job_id=existing_job_id, + ) + return "" + async def create_delete_org_job( self, org: Organization, @@ -621,6 +711,7 @@ async def get_background_job( | PostProcessUploadJob | RetryStuckUploadsJob | ReplicateFilesCronJob + | CopyBucketJob ): """Get background job""" query: dict[str, object] = {"_id": job_id} @@ -666,6 +757,9 @@ def _get_job_by_type_from_data(self, data: dict[str, object]): if data["type"] == BgJobType.REPLICATE_FILES_CRON: return ReplicateFilesCronJob.from_dict(data) + if data["type"] == BgJobType.COPY_BUCKET: + return CopyBucketJob.from_dict(data) + if data["type"] == BgJobType.DELETE_ORG: return DeleteOrgJob.from_dict(data) @@ -779,6 +873,26 @@ async def retry_background_job(self, job_id: str, org: Organization | None = Non ) return {"success": True} + if job.type == BgJobType.COPY_BUCKET: + job = cast(CopyBucketJob, job) + + primary_storage_ref = self.storage_ops.get_default_primary() + primary_storage = self.storage_ops.get_default_s3_storage( + primary_storage_ref + ) + primary_endpoint, primary_bucket_suffix = self.strip_bucket( + primary_storage.endpoint_url + ) + + await self.create_copy_bucket_job( + primary_storage_ref, + primary_endpoint, + primary_bucket_suffix, + job.replica_storage, + existing_job_id=job_id, + ) + return {"success": True} + return {"success": False} async def retry_org_background_job( @@ -787,7 +901,15 @@ async def retry_org_background_job( """Retry background job specific to one org""" if job.type == BgJobType.CREATE_REPLICA: raise HTTPException( - status_code=400, detail="create_replica_job_retry_not_supported" + status_code=400, detail="create_replica_job_retry_no_longer_supported" + ) + + if job.type in (BgJobType.CLEANUP_SEED_FILES, BgJobType.REPLICATE_FILES_CRON): + raise HTTPException(status_code=400, detail="cron_job_retry_not_supported") + + if job.type in (BgJobType.COPY_BUCKET, BgJobType.OPTIMIZE_PAGES): + raise HTTPException( + status_code=400, detail="non_org_specific_job_retry_not_supported" ) if job.type == BgJobType.DELETE_REPLICA: diff --git a/backend/btrixcloud/crawlmanager.py b/backend/btrixcloud/crawlmanager.py index 55cda10171..666c94ed51 100644 --- a/backend/btrixcloud/crawlmanager.py +++ b/backend/btrixcloud/crawlmanager.py @@ -81,6 +81,41 @@ async def run_profile_browser( return browserid + async def run_copy_bucket_job( + self, + primary_storage: StorageRef, + replica_storage: StorageRef, + primary_endpoint: str, + primary_bucket_suffix: str, + replica_endpoint: str, + replica_bucket_suffix: str, + existing_job_id: str | None = None, + ) -> str: + """run job to replicate primary storage bucket to replica location""" + job_type = BgJobType.COPY_BUCKET.value + + if existing_job_id: + job_id = existing_job_id + else: + job_id = f"{job_type}-{secrets.token_hex(5)}" + + params: dict[str, object] = { + "id": job_id, + "primary_secret_name": primary_storage.get_storage_secret_name(), + "primary_file_path": primary_bucket_suffix, + "primary_endpoint": primary_endpoint, + "replica_secret_name": replica_storage.get_storage_secret_name(), + "replica_file_path": replica_bucket_suffix, + "replica_endpoint": replica_endpoint, + "BgJobType": BgJobType, + } + + data = self.templates.env.get_template("copy_bucket_job.yaml").render(params) + + await self.create_from_yaml(data) + + return job_id + async def run_delete_replica_job( self, oid: str, @@ -88,9 +123,6 @@ async def run_delete_replica_job( replica_file_path: str, replica_endpoint: str, delay_days: int = 0, - primary_storage: StorageRef | None = None, - primary_file_path: str | None = None, - primary_endpoint: str | None = None, existing_job_id: str | None = None, ) -> tuple[str, str | None]: """run job to replicate file from primary storage to replica storage""" @@ -110,13 +142,6 @@ async def run_delete_replica_job( "replica_secret_name": replica_storage.get_storage_secret_name(oid), "replica_file_path": replica_file_path, "replica_endpoint": replica_endpoint, - "primary_secret_name": ( - primary_storage.get_storage_secret_name(oid) - if primary_storage - else None - ), - "primary_file_path": primary_file_path if primary_file_path else None, - "primary_endpoint": primary_endpoint if primary_endpoint else None, "BgJobType": BgJobType, } @@ -438,22 +463,23 @@ async def ensure_file_replication_cron_job_exists(self): try: cron_job = await self.batch_api.read_namespaced_cron_job( name=job_id, - namespace=self.namespace, + namespace=DEFAULT_NAMESPACE, ) if cron_job: replication_logger.info("replication_cron_job_exists") if cron_job.spec.schedule != job_schedule: + prev_schedule = cron_job.spec.schedule cron_job.spec.schedule = job_schedule await self.batch_api.patch_namespaced_cron_job( name=cron_job.metadata.name, - namespace=self.namespace, + namespace=DEFAULT_NAMESPACE, body=cron_job, ) replication_logger.info( "replication_cron_job_updated", - prev_schedule=cron_job.spec.schedule, + prev_schedule=prev_schedule, ) return # pylint: disable=broad-exception-caught @@ -475,7 +501,7 @@ async def ensure_file_replication_cron_job_exists(self): params ) - await self.create_from_yaml(data) + await self.create_from_yaml(data, namespace=DEFAULT_NAMESPACE) async def create_crawl_job( self, diff --git a/backend/btrixcloud/main_bg.py b/backend/btrixcloud/main_bg.py index 21866d0d0b..dd96d3ee9a 100644 --- a/backend/btrixcloud/main_bg.py +++ b/backend/btrixcloud/main_bg.py @@ -60,7 +60,7 @@ async def main(): coll_ops, _, _, - _, + bg_job_ops, _, user_manager, _, @@ -109,8 +109,7 @@ async def main(): if job_type == BgJobType.REPLICATE_FILES_CRON: try: - # Call method to launch actual rclone copy jobs - from storage_ops? - crawl_logger.info("replicate_files_cron_reached") + await bg_job_ops.create_copy_bucket_jobs() return ExitCode.SUCCESS # pylint: disable=broad-exception-caught except Exception: diff --git a/backend/btrixcloud/models.py b/backend/btrixcloud/models.py index 8de9d15827..867ef4fe1d 100644 --- a/backend/btrixcloud/models.py +++ b/backend/btrixcloud/models.py @@ -863,7 +863,7 @@ def __str__(self): return self.name return "cs-" + self.name - def get_storage_secret_name(self, oid: str) -> str: + def get_storage_secret_name(self, oid: str = "") -> str: """get k8s secret name for this storage and oid""" if not self.custom: return "storage-" + self.name @@ -3252,6 +3252,8 @@ class BgJobType(StrEnum): CREATE_REPLICA = "create-replica" # New replication job that spins up rclone copy jobs as needed REPLICATE_FILES_CRON = "replicate-files-cron" + # Per-bucket replication job + COPY_BUCKET = "copy-bucket" # ============================================================================ @@ -3344,6 +3346,14 @@ class ReplicateFilesCronJob(BackgroundJob): type: Literal[BgJobType.REPLICATE_FILES_CRON] = BgJobType.REPLICATE_FILES_CRON +# ============================================================================ +class CopyBucketJob(BackgroundJob): + """Model for tracking job to copy primary storage bucket to replica storage""" + + type: Literal[BgJobType.COPY_BUCKET] = BgJobType.COPY_BUCKET + replica_storage: StorageRef + + # ============================================================================ class UpdateCollStatsJob(BackgroundJob): """Model for tracking jobs to readd pages for an org or single crawl""" @@ -3385,6 +3395,7 @@ class RetryStuckUploadsJob(BackgroundJob): | PostProcessUploadJob | RetryStuckUploadsJob | ReplicateFilesCronJob + | CopyBucketJob ] diff --git a/backend/btrixcloud/storages.py b/backend/btrixcloud/storages.py index fc91e8b782..4a149cb282 100644 --- a/backend/btrixcloud/storages.py +++ b/backend/btrixcloud/storages.py @@ -287,6 +287,23 @@ def get_available_storages(self, org: Organization) -> list[StorageRef]: refs.append(StorageRef(name=name, custom=True)) return refs + def get_default_primary(self) -> StorageRef: + """return default primary storageref""" + if not self.default_primary: + raise HTTPException(status_code=400, detail="no_primary_storage") + return self.default_primary + + def get_default_replicas(self) -> list[StorageRef]: + """return default replica storage locations""" + return self.default_replicas + + def get_default_s3_storage(self, ref: StorageRef) -> S3Storage: + """return s3storage object for default storage ref (not org specific)""" + storage = self.default_storages.get(ref.name) + if not storage: + raise KeyError(f"Storage ref {ref.name} not found") + return storage + @asynccontextmanager async def get_s3_client( self, storage: S3Storage, for_presign=False diff --git a/chart/app-templates/background_cron_job.yaml b/chart/app-templates/background_cron_job.yaml index e4ae98cc6c..dbe79b7181 100644 --- a/chart/app-templates/background_cron_job.yaml +++ b/chart/app-templates/background_cron_job.yaml @@ -8,7 +8,7 @@ metadata: spec: concurrencyPolicy: Forbid - successfulJobsHistoryLimit: 0 + successfulJobsHistoryLimit: 1 failedJobsHistoryLimit: 2 schedule: "{{ schedule }}" diff --git a/chart/app-templates/copy_bucket_job.yaml b/chart/app-templates/copy_bucket_job.yaml new file mode 100644 index 0000000000..1b318cb575 --- /dev/null +++ b/chart/app-templates/copy_bucket_job.yaml @@ -0,0 +1,95 @@ +apiVersion: batch/v1 +kind: Job +metadata: + name: "{{ id }}" + labels: + role: "background-job" + job_type: {{ job_type }} + +spec: + ttlSecondsAfterFinished: 300 + backoffLimit: 5 + template: + spec: + restartPolicy: Never + priorityClassName: bg-job + podFailurePolicy: + rules: + - action: FailJob + onExitCodes: + containerName: rclone + operator: NotIn + values: [0] + containers: + - name: rclone + image: rclone/rclone:latest + env: + - name: RCLONE_CONFIG_PRIMARY_TYPE + value: "s3" + + - name: RCLONE_CONFIG_PRIMARY_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: "{{ primary_secret_name }}" + key: STORE_ACCESS_KEY + + - name: RCLONE_CONFIG_PRIMARY_SECRET_ACCESS_KEY + valueFrom: + secretKeyRef: + name: "{{ primary_secret_name }}" + key: STORE_SECRET_KEY + + - name: RCLONE_CONFIG_PRIMARY_REGION + valueFrom: + secretKeyRef: + name: "{{ primary_secret_name }}" + key: STORE_REGION + + - name: RCLONE_CONFIG_PRIMARY_PROVIDER + valueFrom: + secretKeyRef: + name: "{{ primary_secret_name }}" + key: STORE_S3_PROVIDER + + - name: RCLONE_CONFIG_PRIMARY_ENDPOINT + value: "{{ primary_endpoint }}" + + - name: RCLONE_CONFIG_REPLICA_TYPE + value: "s3" + + - name: RCLONE_CONFIG_REPLICA_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: "{{ replica_secret_name }}" + key: STORE_ACCESS_KEY + + - name: RCLONE_CONFIG_REPLICA_SECRET_ACCESS_KEY + valueFrom: + secretKeyRef: + name: "{{ replica_secret_name }}" + key: STORE_SECRET_KEY + + - name: RCLONE_CONFIG_REPLICA_REGION + valueFrom: + secretKeyRef: + name: "{{ replica_secret_name }}" + key: STORE_REGION + + - name: RCLONE_CONFIG_REPLICA_PROVIDER + valueFrom: + secretKeyRef: + name: "{{ replica_secret_name }}" + key: STORE_S3_PROVIDER + + - name: RCLONE_CONFIG_REPLICA_ENDPOINT + value: "{{ replica_endpoint }}" + + command: ["rclone", "-vv", "copy", "--checksum", "--error-on-no-transfer", "primary:{{ primary_file_path }}", "replica:{{ replica_file_path }}"] + + resources: + limits: + memory: "200Mi" + + requests: + memory: "200Mi" + cpu: "50m" diff --git a/chart/app-templates/delete_replica_job.yaml b/chart/app-templates/delete_replica_job.yaml index b623fe7ba3..63cbf9b998 100644 --- a/chart/app-templates/delete_replica_job.yaml +++ b/chart/app-templates/delete_replica_job.yaml @@ -11,7 +11,7 @@ metadata: spec: ttlSecondsAfterFinished: 0 - backoffLimit: 3 + backoffLimit: 5 template: spec: restartPolicy: Never diff --git a/chart/values.yaml b/chart/values.yaml index d565bca138..836bdaa271 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -192,8 +192,8 @@ cleanup_files_after_minutes: 1440 # without a running background job and retrying them stuck_uploads_cron_schedule: "0 * * * *" -# Cron schedule for periodically syncing replica storage -# locations from primary storage +# Cron schedule for periodically syncing replica storage from primary +# Defaults to every two hours replication_job_cron_schedule: "0 */2 * * *" # Emails Image From 4ae775d03990c312675f0bd3a9b2ca3e92a0171e Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Tue, 4 Aug 2026 15:36:12 -0400 Subject: [PATCH 08/27] Remove error-on-no-transfer option from copy bucket rclone command This option resulted in the background job being marked as failed any time rclone copy ran successfully but there were no new files to copy over. --- chart/app-templates/copy_bucket_job.yaml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/chart/app-templates/copy_bucket_job.yaml b/chart/app-templates/copy_bucket_job.yaml index 1b318cb575..36731c0c81 100644 --- a/chart/app-templates/copy_bucket_job.yaml +++ b/chart/app-templates/copy_bucket_job.yaml @@ -84,7 +84,7 @@ spec: - name: RCLONE_CONFIG_REPLICA_ENDPOINT value: "{{ replica_endpoint }}" - command: ["rclone", "-vv", "copy", "--checksum", "--error-on-no-transfer", "primary:{{ primary_file_path }}", "replica:{{ replica_file_path }}"] + command: ["rclone", "-vv", "copy", "--checksum", "primary:{{ primary_file_path }}", "replica:{{ replica_file_path }}"] resources: limits: From 480138d5931b1c2cb121b77d3fe07176a668253b Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Tue, 4 Aug 2026 16:13:13 -0400 Subject: [PATCH 09/27] Stop tracking replicas per-file and add migration to remove from db Also remove replicas from backend tests --- backend/btrixcloud/background_jobs.py | 4 +- backend/btrixcloud/basecrawls.py | 2 - backend/btrixcloud/db.py | 2 +- ...ation_0059_remove_file_replicas_from_db.py | 49 +++++++++++++++++++ backend/btrixcloud/models.py | 3 -- backend/btrixcloud/profiles.py | 10 ---- backend/test/conftest.py | 1 - backend/test/test_profiles.py | 3 -- 8 files changed, 52 insertions(+), 22 deletions(-) create mode 100644 backend/btrixcloud/migrations/migration_0059_remove_file_replicas_from_db.py diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index acb797c8c2..6f20e865f9 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -111,7 +111,7 @@ async def create_delete_replica_jobs( """Create a job to delete each replica for the given file""" ids = [] - for replica_ref in file.replicas or []: + for replica_ref in self.storage_ops.get_org_replicas_storage_refs(org): job_id = await self.create_delete_replica_job( org, file, object_id, object_type, replica_ref ) @@ -278,7 +278,7 @@ async def create_copy_bucket_job( return job_id # pylint: disable=broad-exception-caught - except Exception as exc: + except Exception: replica_logger.warning( "copy_bucket_job_start_failed", exc_info=True, diff --git a/backend/btrixcloud/basecrawls.py b/backend/btrixcloud/basecrawls.py index d759ea46c9..fd5e1674a8 100644 --- a/backend/btrixcloud/basecrawls.py +++ b/backend/btrixcloud/basecrawls.py @@ -634,7 +634,6 @@ async def bulk_presigned_files( hash=file["hash"], size=file["size"], crawlId=file["crawl_id"], - numReplicas=len(file.get("replicas") or []), expireAt=date_to_str( presigned["signedAt"] + self.storage_ops.signed_duration_delta @@ -676,7 +675,6 @@ async def bulk_presigned_files( hash=file["hash"], size=file["size"], crawlId=file["crawl_id"], - numReplicas=len(file.get("replicas") or []), expireAt=date_to_str(expire_at), ) ) diff --git a/backend/btrixcloud/db.py b/backend/btrixcloud/db.py index e8df8c8b00..32d9eb0e0e 100644 --- a/backend/btrixcloud/db.py +++ b/backend/btrixcloud/db.py @@ -39,7 +39,7 @@ ) = object -CURR_DB_VERSION = "0058" +CURR_DB_VERSION = "0059" MIN_DB_VERSION = 7.0 diff --git a/backend/btrixcloud/migrations/migration_0059_remove_file_replicas_from_db.py b/backend/btrixcloud/migrations/migration_0059_remove_file_replicas_from_db.py new file mode 100644 index 0000000000..d2c0460f6e --- /dev/null +++ b/backend/btrixcloud/migrations/migration_0059_remove_file_replicas_from_db.py @@ -0,0 +1,49 @@ +""" +Migration 0059 - Remove per-file tracking of file replicas in database +""" + +import structlog + +from btrixcloud.migrations import BaseMigration + +logger: structlog.stdlib.BoundLogger = structlog.get_logger(__name__) + +MIGRATION_VERSION = "0059" + + +class Migration(BaseMigration): + """Migration class.""" + + # pylint: disable=unused-argument + def __init__(self, mdb, **kwargs): + super().__init__(mdb, migration_version=MIGRATION_VERSION) + + async def migrate_up(self): + """Perform migration up. + + Remove replicas array from all crawl and profile files. + """ + crawls_mdb = self.mdb["crawls"] + profiles_mdb = self.mdb["profiles"] + + try: + res = await crawls_mdb.update_many( + {"files": {"$nin": [None, []]}}, + {"$unset": {"files.$[].replicas": 1}}, + ) + updated = res.modified_count + logger.info("updated_crawl_files", count=updated) + # pylint: disable=broad-exception-caught + except Exception: + logger.exception("failed_to_update_crawl_files") + + try: + res = await profiles_mdb.update_many( + {"resource": {"$ne": None}}, + {"$unset": {"resource.replicas": 1}}, + ) + updated = res.modified_count + logger.info("updated_profile_files", count=updated) + # pylint: disable=broad-exception-caught + except Exception: + logger.exception("failed_to_update_profile_files") diff --git a/backend/btrixcloud/models.py b/backend/btrixcloud/models.py index 867ef4fe1d..ed6aefd83a 100644 --- a/backend/btrixcloud/models.py +++ b/backend/btrixcloud/models.py @@ -896,8 +896,6 @@ class BaseFile(BaseModel): size: int storage: StorageRef - replicas: list[StorageRef] | None = [] - # ============================================================================ class CrawlFile(BaseFile): @@ -914,7 +912,6 @@ class CrawlFileOut(BaseModel): size: int crawlId: str | None = None - numReplicas: int = 0 expireAt: str | None = None fromDependency: bool = False diff --git a/backend/btrixcloud/profiles.py b/backend/btrixcloud/profiles.py index d76e180c7d..a6ba151dac 100644 --- a/backend/btrixcloud/profiles.py +++ b/backend/btrixcloud/profiles.py @@ -30,7 +30,6 @@ ProfilePingResponse, ProfileSearchValuesResponse, ProfileUpdate, - StorageRef, SuccessResponse, SuccessResponseStorageQuota, TagsResponse, @@ -648,15 +647,6 @@ async def _send_browser_req( return data or {} - async def add_profile_file_replica( - self, profileid: UUID, filename: str, ref: StorageRef - ) -> dict[str, object]: - """Add replica StorageRef to existing ProfileFile""" - return await self.profiles.find_one_and_update( - {"_id": profileid, "resource.filename": filename}, - {"$push": {"resource.replicas": {"name": ref.name, "custom": ref.custom}}}, - ) - async def calculate_org_profile_file_storage(self, oid: UUID) -> int: """Calculate and return total size of profile files in org""" total_size = 0 diff --git a/backend/test/conftest.py b/backend/test/conftest.py index 760544b56e..1c76246eb0 100644 --- a/backend/test/conftest.py +++ b/backend/test/conftest.py @@ -871,7 +871,6 @@ def profile_config_id(admin_auth_headers, default_org_id, profile_id): assert resource["size"] assert resource["storage"] assert resource["storage"]["name"] - assert resource.get("replicas") or resource.get("replicas") == [] # Use profile in a workflow r = requests.post( diff --git a/backend/test/test_profiles.py b/backend/test/test_profiles.py index acdcad153e..dcef5769ce 100644 --- a/backend/test/test_profiles.py +++ b/backend/test/test_profiles.py @@ -62,7 +62,6 @@ def test_get_profile(admin_auth_headers, default_org_id, profile_id, profile_con assert resource["size"] assert resource["storage"] assert resource["storage"]["name"] - assert "replicas" in resource and isinstance(resource["replicas"], list) assert "crawlconfigs" not in data assert data["inUse"] == True @@ -118,7 +117,6 @@ def test_list_profiles(admin_auth_headers, default_org_id, profile_id, profile_2 assert resource["size"] assert resource["storage"] assert resource["storage"]["name"] - assert "replicas" in resource and isinstance(resource["replicas"], list) # First profile should be listed second by default because it was # modified less recently @@ -145,7 +143,6 @@ def test_list_profiles(admin_auth_headers, default_org_id, profile_id, profile_2 assert resource["size"] assert resource["storage"] assert resource["storage"]["name"] - assert "replicas" in resource and isinstance(resource["replicas"], list) break except: From 35fff037fa059a00044bf48e0832f3eeb77f1002 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Tue, 4 Aug 2026 17:14:23 -0400 Subject: [PATCH 10/27] Update nightly tests to account for new replication routine --- backend/test_nightly/conftest.py | 22 ---------- backend/test_nightly/test_crawl_timeout.py | 32 ++++++++------ backend/test_nightly/test_upload_replicas.py | 43 ++++++++++--------- .../test_nightly/test_z_background_jobs.py | 15 ++++--- chart/test/test-nightly-addons.yaml | 3 ++ 5 files changed, 54 insertions(+), 61 deletions(-) diff --git a/backend/test_nightly/conftest.py b/backend/test_nightly/conftest.py index 7234e18d1e..b936ddf153 100644 --- a/backend/test_nightly/conftest.py +++ b/backend/test_nightly/conftest.py @@ -376,17 +376,6 @@ def deleted_crawl_id(admin_auth_headers, default_org_id): break time.sleep(5) - # Wait until replica background job completes - while True: - r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs/?jobType=create-replica&success=True", - headers=admin_auth_headers, - ) - assert r.status_code == 200 - if r.json()["total"] == 1: - break - time.sleep(5) - # Delete crawl r = requests.post( f"{API_PREFIX}/orgs/{default_org_id}/crawls/delete", @@ -395,15 +384,4 @@ def deleted_crawl_id(admin_auth_headers, default_org_id): ) assert r.status_code == 200 - # Wait until delete replica background job completes - while True: - r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs/?jobType=delete-replica&success=True", - headers=admin_auth_headers, - ) - assert r.status_code == 200 - if r.json()["total"] == 1: - break - time.sleep(5) - return crawl_id diff --git a/backend/test_nightly/test_crawl_timeout.py b/backend/test_nightly/test_crawl_timeout.py index ce1e430e10..95fbee6253 100644 --- a/backend/test_nightly/test_crawl_timeout.py +++ b/backend/test_nightly/test_crawl_timeout.py @@ -2,6 +2,8 @@ import requests +from btrixcloud.utils import dt_now + from .conftest import API_PREFIX from .utils import verify_file_replicated @@ -33,16 +35,20 @@ def test_crawl_timeout(admin_auth_headers, default_org_id, timeout_crawl): def test_crawl_files_replicated(admin_auth_headers, default_org_id, timeout_crawl): - time.sleep(20) + crawl_complete = dt_now() + + # Wait a few minutes so that replication jobs have time to run + time.sleep(300) - # Verify replication job was successful + # Verify copy bucket job has run and succeeded since crawl completed r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=create-replica", + f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=copy-bucket", headers=admin_auth_headers, ) assert r.status_code == 200 latest_job = r.json()["items"][0] - assert latest_job["type"] == "create-replica" + assert latest_job["type"] == "copy-bucket" + assert latest_job["started"] >= crawl_complete job_id = latest_job["id"] attempts = 0 @@ -62,22 +68,20 @@ def test_crawl_files_replicated(admin_auth_headers, default_org_id, timeout_craw assert job["success"] break - # Assert file was updated + # Verify crawlfiles are stored in replica location r = requests.get( f"{API_PREFIX}/orgs/{default_org_id}/crawls/{timeout_crawl}/replay.json", headers=admin_auth_headers, ) assert r.status_code == 200 data = r.json() + + oid = data["oid"] + assert oid + files = data.get("resources") assert files for file_ in files: - assert file_["numReplicas"] == 1 - - # Verify replica is stored - r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}", headers=admin_auth_headers - ) - assert r.status_code == 200 - data = r.json() - verify_file_replicated(data["file_path"]) + filename = file_["name"] + file_path = f"{oid}/{filename}" + verify_file_replicated(file_path) diff --git a/backend/test_nightly/test_upload_replicas.py b/backend/test_nightly/test_upload_replicas.py index 3f9c1337b5..330c08aeba 100644 --- a/backend/test_nightly/test_upload_replicas.py +++ b/backend/test_nightly/test_upload_replicas.py @@ -4,6 +4,7 @@ import structlog import requests +from btrixcloud.utils import dt_now from test.utils import read_in_chunks from .conftest import API_PREFIX @@ -33,16 +34,20 @@ def test_upload_stream(admin_auth_headers, default_org_id): def test_upload_file_replicated(admin_auth_headers, default_org_id): - time.sleep(20) + upload_complete = dt_now() + + # Wait a few minutes so that replication jobs have time to run + time.sleep(300) - # Verify replication job was successful + # Verify copy bucket job has run and succeeded since upload r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=-1&jobType=create-replica", + f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=copy-bucket", headers=admin_auth_headers, ) assert r.status_code == 200 latest_job = r.json()["items"][0] - assert latest_job["type"] == "create-replica" + assert latest_job["type"] == "copy-bucket" + assert latest_job["started"] >= upload_complete job_id = latest_job["id"] attempts = 0 @@ -62,30 +67,24 @@ def test_upload_file_replicated(admin_auth_headers, default_org_id): assert job["success"] break - # Verify file updated + # Verify upload file is stored r = requests.get( f"{API_PREFIX}/orgs/{default_org_id}/uploads/{upload_id}/replay.json", headers=admin_auth_headers, ) assert r.status_code == 200 data = r.json() + files = data.get("resources") assert files - for file_ in files: - assert file_["numReplicas"] == 1 - # Verify replica is stored - r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}", headers=admin_auth_headers - ) - assert r.status_code == 200 - job = r.json() - logger.info( - "upload_file_path", - file_path=job["file_path"], - unstructured_message=f"{job['file_path']}", - ) - verify_file_replicated(job["file_path"]) + file_ = files[0] + filename = file_["name"] + + global upload_file_path + upload_file_path = f"{default_org_id}/{filename}" + + verify_file_replicated(upload_file_path) def test_delete_upload_and_replicas(admin_auth_headers, default_org_id): @@ -145,4 +144,8 @@ def test_delete_upload_and_replicas(admin_auth_headers, default_org_id): ) assert r.status_code == 200 job = r.json() - verify_file_and_replica_deleted(job["file_path"]) + + job_file_path = job["file_path"] + assert job_file_path == upload_file_path + + verify_file_and_replica_deleted(job_file_path) diff --git a/backend/test_nightly/test_z_background_jobs.py b/backend/test_nightly/test_z_background_jobs.py index 28cd4d4b43..d58f12ba29 100644 --- a/backend/test_nightly/test_z_background_jobs.py +++ b/backend/test_nightly/test_z_background_jobs.py @@ -33,12 +33,11 @@ def test_background_jobs_list(admin_auth_headers, default_org_id, deleted_crawl_ assert job_id -@pytest.mark.parametrize("job_type", [("create-replica"), ("delete-replica")]) def test_background_jobs_list_filter_by_type( admin_auth_headers, default_org_id, deleted_crawl_id, job_type ): r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs/?jobType={job_type}", + f"{API_PREFIX}/orgs/{default_org_id}/jobs/?jobType=delete-replica", headers=admin_auth_headers, ) assert r.status_code == 200 @@ -49,7 +48,7 @@ def test_background_jobs_list_filter_by_type( assert len(items) == data["total"] for item in items: - assert item["type"] == job_type + assert item["type"] == "delete-replica" def test_background_jobs_list_filter_by_success( @@ -91,7 +90,12 @@ def test_get_background_job(admin_auth_headers, default_org_id, deleted_crawl_id data = r.json() assert data["id"] - assert data["type"] in ("create-replica", "delete-replica") + assert data["type"] in ( + "replicate-files-cron", + "copy-bucket", + "delete-replica", + "cleanup-seed-files", + ) assert data["oid"] == default_org_id assert data["success"] assert data["started"] @@ -99,7 +103,8 @@ def test_get_background_job(admin_auth_headers, default_org_id, deleted_crawl_id assert data["file_path"] assert data["object_type"] assert data["object_id"] - assert data["replica_storage"] + if data["type"] in ("delete-replica", "copy-bucket"): + assert data["replica_storage"] def test_retry_all_failed_bg_jobs_not_superuser(crawler_auth_headers, deleted_crawl_id): diff --git a/chart/test/test-nightly-addons.yaml b/chart/test/test-nightly-addons.yaml index 028c872137..ed9b5dc9aa 100644 --- a/chart/test/test-nightly-addons.yaml +++ b/chart/test/test-nightly-addons.yaml @@ -10,6 +10,9 @@ cleanup_job_cron_schedule: "* * * * *" # Clean up files > 1 minute old in testing cleanup_files_after_minutes: 1 +# Every 2 minutes, for use in testing file replication +replication_job_cron_schedule: "*/2 * * * *" + # max crawl queue size x concurrent crawls crawl_queue_limit_scale: 1 From 62f9dcbf263c16f59ed3f7f9e475e1d77fdbfb40 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 5 Aug 2026 14:39:59 -0400 Subject: [PATCH 11/27] Increase nightly test timeouts and make a few tweaks --- backend/test_nightly/conftest.py | 3 +++ backend/test_nightly/test_crawl_timeout.py | 2 ++ backend/test_nightly/test_upload_replicas.py | 5 +++++ backend/test_nightly/test_z_background_jobs.py | 1 + 4 files changed, 11 insertions(+) diff --git a/backend/test_nightly/conftest.py b/backend/test_nightly/conftest.py index b936ddf153..772ec36218 100644 --- a/backend/test_nightly/conftest.py +++ b/backend/test_nightly/conftest.py @@ -376,6 +376,9 @@ def deleted_crawl_id(admin_auth_headers, default_org_id): break time.sleep(5) + # Wait for it to replicate + time.sleep(300) + # Delete crawl r = requests.post( f"{API_PREFIX}/orgs/{default_org_id}/crawls/delete", diff --git a/backend/test_nightly/test_crawl_timeout.py b/backend/test_nightly/test_crawl_timeout.py index 95fbee6253..56c0712985 100644 --- a/backend/test_nightly/test_crawl_timeout.py +++ b/backend/test_nightly/test_crawl_timeout.py @@ -1,5 +1,6 @@ import time +import pytest import requests from btrixcloud.utils import dt_now @@ -34,6 +35,7 @@ def test_crawl_timeout(admin_auth_headers, default_org_id, timeout_crawl): attempts += 1 +@pytest.mark.timeout(1800) def test_crawl_files_replicated(admin_auth_headers, default_org_id, timeout_crawl): crawl_complete = dt_now() diff --git a/backend/test_nightly/test_upload_replicas.py b/backend/test_nightly/test_upload_replicas.py index 330c08aeba..e2daee5238 100644 --- a/backend/test_nightly/test_upload_replicas.py +++ b/backend/test_nightly/test_upload_replicas.py @@ -2,6 +2,7 @@ import time import structlog +import pytest import requests from btrixcloud.utils import dt_now @@ -17,6 +18,8 @@ curr_dir = os.path.dirname(os.path.realpath(__file__)) +upload_file_path = None + def test_upload_stream(admin_auth_headers, default_org_id): with open(os.path.join(curr_dir, "..", "test", "data", "example.wacz"), "rb") as fh: @@ -33,6 +36,7 @@ def test_upload_stream(admin_auth_headers, default_org_id): upload_id = r.json()["id"] +@pytest.mark.timeout(1800) def test_upload_file_replicated(admin_auth_headers, default_org_id): upload_complete = dt_now() @@ -87,6 +91,7 @@ def test_upload_file_replicated(admin_auth_headers, default_org_id): verify_file_replicated(upload_file_path) +@pytest.mark.timeout(1800) def test_delete_upload_and_replicas(admin_auth_headers, default_org_id): r = requests.post( f"{API_PREFIX}/orgs/{default_org_id}/uploads/delete", diff --git a/backend/test_nightly/test_z_background_jobs.py b/backend/test_nightly/test_z_background_jobs.py index d58f12ba29..ebb8ad7afa 100644 --- a/backend/test_nightly/test_z_background_jobs.py +++ b/backend/test_nightly/test_z_background_jobs.py @@ -8,6 +8,7 @@ job_id = None +@pytest.mark.timeout(1800) def test_background_jobs_list(admin_auth_headers, default_org_id, deleted_crawl_id): r = requests.get( f"{API_PREFIX}/orgs/{default_org_id}/jobs/", headers=admin_auth_headers From 5db473e6944207e5ea9733230b1dccd9cce62ae7 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 5 Aug 2026 15:04:50 -0400 Subject: [PATCH 12/27] Modify nightly tests to wait longer for replicaton job --- backend/test_nightly/test_crawl_timeout.py | 37 ++++++++++++------- backend/test_nightly/test_upload_replicas.py | 39 ++++++++++++-------- 2 files changed, 47 insertions(+), 29 deletions(-) diff --git a/backend/test_nightly/test_crawl_timeout.py b/backend/test_nightly/test_crawl_timeout.py index 56c0712985..bf4c8af68e 100644 --- a/backend/test_nightly/test_crawl_timeout.py +++ b/backend/test_nightly/test_crawl_timeout.py @@ -39,22 +39,31 @@ def test_crawl_timeout(admin_auth_headers, default_org_id, timeout_crawl): def test_crawl_files_replicated(admin_auth_headers, default_org_id, timeout_crawl): crawl_complete = dt_now() - # Wait a few minutes so that replication jobs have time to run - time.sleep(300) - # Verify copy bucket job has run and succeeded since crawl completed - r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=copy-bucket", - headers=admin_auth_headers, - ) - assert r.status_code == 200 - latest_job = r.json()["items"][0] - assert latest_job["type"] == "copy-bucket" - assert latest_job["started"] >= crawl_complete - job_id = latest_job["id"] + job_id = None + + # Give job up to 15 minutes to complete + attempts = 0 + while attempts < 15: + r = requests.get( + f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=copy-bucket", + headers=admin_auth_headers, + ) + assert r.status_code == 200 + jobs = r.json().get("items", []) + if jobs: + latest_job = jobs[0] + assert latest_job["type"] == "copy-bucket" + if latest_job["started"] >= crawl_complete: + job_id = latest_job["id"] + break + + attempts += 1 + time.sleep(60) + # Give job up to 5 minutes to finish attempts = 0 - while attempts < 5: + while attempts < 10: r = requests.get( f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}", headers=admin_auth_headers, @@ -64,7 +73,7 @@ def test_crawl_files_replicated(admin_auth_headers, default_org_id, timeout_craw finished = latest_job.get("finished") if not finished: attempts += 1 - time.sleep(10) + time.sleep(30) continue assert job["success"] diff --git a/backend/test_nightly/test_upload_replicas.py b/backend/test_nightly/test_upload_replicas.py index e2daee5238..c416f74d26 100644 --- a/backend/test_nightly/test_upload_replicas.py +++ b/backend/test_nightly/test_upload_replicas.py @@ -40,22 +40,31 @@ def test_upload_stream(admin_auth_headers, default_org_id): def test_upload_file_replicated(admin_auth_headers, default_org_id): upload_complete = dt_now() - # Wait a few minutes so that replication jobs have time to run - time.sleep(300) - - # Verify copy bucket job has run and succeeded since upload - r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=copy-bucket", - headers=admin_auth_headers, - ) - assert r.status_code == 200 - latest_job = r.json()["items"][0] - assert latest_job["type"] == "copy-bucket" - assert latest_job["started"] >= upload_complete - job_id = latest_job["id"] + # Verify copy bucket job has run and succeeded since crawl completed + job_id = None + # Give job up to 15 minutes to complete attempts = 0 - while attempts < 5: + while attempts < 15: + r = requests.get( + f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=copy-bucket", + headers=admin_auth_headers, + ) + assert r.status_code == 200 + jobs = r.json().get("items", []) + if jobs: + latest_job = jobs[0] + assert latest_job["type"] == "copy-bucket" + if latest_job["started"] >= upload_complete: + job_id = latest_job["id"] + break + + attempts += 1 + time.sleep(60) + + # Give job up to 5 minutes to finish + attempts = 0 + while attempts < 10: r = requests.get( f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}", headers=admin_auth_headers, @@ -65,7 +74,7 @@ def test_upload_file_replicated(admin_auth_headers, default_org_id): finished = latest_job.get("finished") if not finished: attempts += 1 - time.sleep(10) + time.sleep(30) continue assert job["success"] From 178a5489104909bc6ea751f7fb090adbc83d5e1d Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 5 Aug 2026 15:10:18 -0400 Subject: [PATCH 13/27] Remove backed up status from frontend now that we don't track per-file --- frontend/src/pages/org/browser-profiles/profile.ts | 14 -------------- frontend/src/types/collection.ts | 1 - frontend/src/types/crawler.ts | 6 ------ 3 files changed, 21 deletions(-) diff --git a/frontend/src/pages/org/browser-profiles/profile.ts b/frontend/src/pages/org/browser-profiles/profile.ts index 0053837677..0a815531a6 100644 --- a/frontend/src/pages/org/browser-profiles/profile.ts +++ b/frontend/src/pages/org/browser-profiles/profile.ts @@ -24,7 +24,6 @@ import { import type { ProfileUpdatedEvent } from "@/features/browser-profiles/types"; import type { WorkflowColumnName } from "@/features/crawl-workflows/workflow-list"; import { emptyMessage } from "@/layouts/emptyMessage"; -import { labelWithIcon } from "@/layouts/labelWithIcon"; import { pageHeader, pageNav } from "@/layouts/pageHeader"; import { panel, panelBody } from "@/layouts/panel"; import { OrgTab, WorkflowTab } from "@/routes"; @@ -548,19 +547,6 @@ export class BrowserProfilesProfilePage extends BtrixElement { : noData, )} - - ${this.renderDetail((profile) => { - const isBackedUp = - profile.resource?.replicas && - profile.resource.replicas.length > 0; - return labelWithIcon({ - label: isBackedUp ? msg("Backed Up") : msg("Not Backed Up"), - icon: html``, - }); - })} - `, }); diff --git a/frontend/src/types/collection.ts b/frontend/src/types/collection.ts index 692b83aea6..1763917b61 100644 --- a/frontend/src/types/collection.ts +++ b/frontend/src/types/collection.ts @@ -40,7 +40,6 @@ export const publicCollectionSchema = z.object({ hash: z.string(), size: z.number(), crawlId: z.string().nullable(), - numReplicas: z.number(), expireAt: z.string().datetime().nullable(), fromDependency: z.boolean(), }), diff --git a/frontend/src/types/crawler.ts b/frontend/src/types/crawler.ts index 4d870d41c7..cbb123bf30 100644 --- a/frontend/src/types/crawler.ts +++ b/frontend/src/types/crawler.ts @@ -131,11 +131,6 @@ export type ListWorkflow = Omit & { config: Workflow["config"] | null; }; -export type ProfileReplica = { - name: string; - custom?: boolean; -}; - export type Profile = { id: string; name: string; @@ -160,7 +155,6 @@ export type Profile = { path: string; hash: string; size: number; - replicas: ProfileReplica[] | null; }; crawlerChannel?: CrawlerChannelImage | AnyString; proxyId?: string; From b27ca7d63d2e505f9bba6974e79f418c591b0ed2 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 5 Aug 2026 17:12:32 -0400 Subject: [PATCH 14/27] Sort copy-bucket jobs by newest first --- backend/test_nightly/test_crawl_timeout.py | 2 +- backend/test_nightly/test_upload_replicas.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/backend/test_nightly/test_crawl_timeout.py b/backend/test_nightly/test_crawl_timeout.py index bf4c8af68e..47884a6366 100644 --- a/backend/test_nightly/test_crawl_timeout.py +++ b/backend/test_nightly/test_crawl_timeout.py @@ -46,7 +46,7 @@ def test_crawl_files_replicated(admin_auth_headers, default_org_id, timeout_craw attempts = 0 while attempts < 15: r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=copy-bucket", + f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=-1&jobType=copy-bucket", headers=admin_auth_headers, ) assert r.status_code == 200 diff --git a/backend/test_nightly/test_upload_replicas.py b/backend/test_nightly/test_upload_replicas.py index c416f74d26..9736158ba9 100644 --- a/backend/test_nightly/test_upload_replicas.py +++ b/backend/test_nightly/test_upload_replicas.py @@ -47,7 +47,7 @@ def test_upload_file_replicated(admin_auth_headers, default_org_id): attempts = 0 while attempts < 15: r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=1&jobType=copy-bucket", + f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=-1&jobType=copy-bucket", headers=admin_auth_headers, ) assert r.status_code == 200 From 707d81b4802eda234f7a879824818ab734d0758e Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 12 Aug 2026 10:55:11 -0400 Subject: [PATCH 15/27] Refactor method based on changes in multi-wacz PR, add todo --- backend/btrixcloud/background_jobs.py | 4 +- backend/btrixcloud/crawlmanager.py | 82 ++++++--------------------- 2 files changed, 20 insertions(+), 66 deletions(-) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index 6f20e865f9..ea9c5aa6be 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -29,8 +29,8 @@ PostProcessUploadJob, ReAddOrgPagesJob, RecalculateOrgStatsJob, - RetryStuckUploadsJob, ReplicateFilesCronJob, + RetryStuckUploadsJob, StorageRef, SuccessResponse, SuccessResponseId, @@ -619,7 +619,7 @@ async def job_finished( if job_type in ( BgJobType.CLEANUP_SEED_FILES, BgJobType.RETRY_STUCK_UPLOADS, - BgJobType.REPLICATE_FILES_CRON + BgJobType.REPLICATE_FILES_CRON, ): if not started: started = finished diff --git a/backend/btrixcloud/crawlmanager.py b/backend/btrixcloud/crawlmanager.py index 666c94ed51..d9b823cca4 100644 --- a/backend/btrixcloud/crawlmanager.py +++ b/backend/btrixcloud/crawlmanager.py @@ -392,6 +392,24 @@ async def ensure_retry_stuck_uploads_cron_job_exists(self): job_schedule, ) + async def ensure_file_replication_cron_job_exists(self): + """ensure cron background job to replica default storages exists""" + + # Default schedule is every 2 hours + default_schedule = "0 */2 * * *" + job_schedule = os.environ.get("REPLICATION_JOB_CRON_SCHEDULE", default_schedule) + + # TODO: Only create background job if default replica storage locations + # are configured. + # If replica locations were configured and then removed, do we want to + # do any cleanup around that as well? + + await self._ensure_bg_cron_job_exists( + "replicate-files-cron", + BgJobType.REPLICATE_FILES_CRON.value, + job_schedule, + ) + async def _ensure_bg_cron_job_exists( self, job_id: str, @@ -439,70 +457,6 @@ async def _ensure_bg_cron_job_exists( cron_logger.info("bg_cron_job_creating") await self.create_from_yaml(data, namespace=DEFAULT_NAMESPACE) - async def ensure_file_replication_cron_job_exists(self): - """ensure cron background job to periodically replicate all files exists - - Each time the background job runs, it will spawn one additional background - job for each configured replica storage location, and run `rclone copy` from - primary storage to each replica location to keep them in sync without the - possibility of deleting anything from the destination. - - Deletions are handled separately with individual jobs and are subject to - the replica deletion delay. - """ - - job_id = "replicate-files-cron" - - # Default schedule is every 2 hours - default_schedule = "0 */2 * * *" - job_schedule = os.environ.get("REPLICATION_JOB_CRON_SCHEDULE", default_schedule) - - # Don't create a duplicate cron job if already exists - replication_logger = logger.bind(schedule=job_schedule) - - try: - cron_job = await self.batch_api.read_namespaced_cron_job( - name=job_id, - namespace=DEFAULT_NAMESPACE, - ) - if cron_job: - replication_logger.info("replication_cron_job_exists") - - if cron_job.spec.schedule != job_schedule: - prev_schedule = cron_job.spec.schedule - cron_job.spec.schedule = job_schedule - - await self.batch_api.patch_namespaced_cron_job( - name=cron_job.metadata.name, - namespace=DEFAULT_NAMESPACE, - body=cron_job, - ) - replication_logger.info( - "replication_cron_job_updated", - prev_schedule=prev_schedule, - ) - return - # pylint: disable=broad-exception-caught - except Exception: - pass - - replication_logger.info("replication_cron_job_creating") - - params = { - "id": job_id, - "job_type": BgJobType.REPLICATE_FILES_CRON.value, - "backend_image": os.environ.get("BACKEND_IMAGE", ""), - "pull_policy": os.environ.get("BACKEND_IMAGE_PULL_POLICY", ""), - "schedule": job_schedule, - "larger_resources": False, - } - - data = self.templates.env.get_template("background_cron_job.yaml").render( - params - ) - - await self.create_from_yaml(data, namespace=DEFAULT_NAMESPACE) - async def create_crawl_job( self, crawlconfig: CrawlConfig, From e527a3eb21d758a61c1c48880999bd90497e004f Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 12 Aug 2026 11:06:46 -0400 Subject: [PATCH 16/27] Make sure replicate files cron job only exists if replicas configured --- backend/btrixcloud/background_jobs.py | 6 ++++- backend/btrixcloud/crawlmanager.py | 34 +++++++++++++++++++-------- 2 files changed, 29 insertions(+), 11 deletions(-) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index ea9c5aa6be..489c043581 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -600,7 +600,11 @@ async def ensure_cron_jobs_exist(self): """Ensure periodic background cron jobs exist""" await self.crawl_manager.ensure_cleanup_seed_file_cron_job_exists() await self.crawl_manager.ensure_retry_stuck_uploads_cron_job_exists() - await self.crawl_manager.ensure_file_replication_cron_job_exists() + + replicas_configured = True if self.storage_ops.get_default_replicas() else False + await self.crawl_manager.ensure_file_replication_cron_job_exists( + replicas_configured + ) async def job_finished( self, diff --git a/backend/btrixcloud/crawlmanager.py b/backend/btrixcloud/crawlmanager.py index d9b823cca4..ab99b11e51 100644 --- a/backend/btrixcloud/crawlmanager.py +++ b/backend/btrixcloud/crawlmanager.py @@ -392,23 +392,37 @@ async def ensure_retry_stuck_uploads_cron_job_exists(self): job_schedule, ) - async def ensure_file_replication_cron_job_exists(self): + async def ensure_file_replication_cron_job_exists( + self, replicas_configured: bool = False + ): """ensure cron background job to replica default storages exists""" # Default schedule is every 2 hours default_schedule = "0 */2 * * *" job_schedule = os.environ.get("REPLICATION_JOB_CRON_SCHEDULE", default_schedule) - # TODO: Only create background job if default replica storage locations - # are configured. - # If replica locations were configured and then removed, do we want to - # do any cleanup around that as well? + job_id = "replicate-files-cron" - await self._ensure_bg_cron_job_exists( - "replicate-files-cron", - BgJobType.REPLICATE_FILES_CRON.value, - job_schedule, - ) + if replicas_configured: + return await self._ensure_bg_cron_job_exists( + job_id, + BgJobType.REPLICATE_FILES_CRON.value, + job_schedule, + ) + + # If no replica locations are configured, make sure no replication cron + # job exists, as one could have been previously configured. + try: + await self.batch_api.delete_namespaced_cron_job( + name=job_id, + namespace=DEFAULT_NAMESPACE, + ) + logger.info( + "bg_cron_job_deleting", job_id=job_id, replicas_configured=False + ) + except ApiException as exc: + if exc.status != 404: + raise async def _ensure_bg_cron_job_exists( self, From efcea20e55623fbf795163ff5db4d2378ad77301 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 12 Aug 2026 11:08:33 -0400 Subject: [PATCH 17/27] Increase resources for copy bucket job --- chart/app-templates/copy_bucket_job.yaml | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/chart/app-templates/copy_bucket_job.yaml b/chart/app-templates/copy_bucket_job.yaml index 36731c0c81..480afadc47 100644 --- a/chart/app-templates/copy_bucket_job.yaml +++ b/chart/app-templates/copy_bucket_job.yaml @@ -88,8 +88,8 @@ spec: resources: limits: - memory: "200Mi" + memory: "1200Mi" requests: - memory: "200Mi" - cpu: "50m" + memory: "500Mi" + cpu: "200m" From df7d89184e3b3c885d0cc39192992e218cfa684b Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 12 Aug 2026 13:32:13 -0400 Subject: [PATCH 18/27] Include new jobs in bg job unit tests --- backend/test/test_unit_background_jobs.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/backend/test/test_unit_background_jobs.py b/backend/test/test_unit_background_jobs.py index 9728c317f8..8504648d19 100644 --- a/backend/test/test_unit_background_jobs.py +++ b/backend/test/test_unit_background_jobs.py @@ -12,6 +12,7 @@ from btrixcloud.models import ( BgJobType, CleanupSeedFilesJob, + CopyBucketJob, CreateReplicaJob, DeleteOrgJob, DeleteReplicaJob, @@ -19,6 +20,7 @@ PostProcessUploadJob, ReAddOrgPagesJob, RecalculateOrgStatsJob, + ReplicateFilesCronJob, RetryStuckUploadsJob, UpdateCollStatsJob, ) @@ -34,6 +36,8 @@ BgJobType.UPDATE_COLL_STATS: UpdateCollStatsJob, BgJobType.POSTPROCESS_UPLOAD: PostProcessUploadJob, BgJobType.RETRY_STUCK_UPLOADS: RetryStuckUploadsJob, + BgJobType.REPLICATE_FILES_CRON: ReplicateFilesCronJob, + BgJobType.COPY_BUCKET: CopyBucketJob, } _TYPE_EXTRAS = { @@ -49,6 +53,9 @@ "object_id": "test-object", "replica_storage": {"name": "test-storage"}, }, + BgJobType.COPY_BUCKET: { + "replica_storage": {"name": "test-storage"}, + }, } From ffdc45256d48df86214578e35317ae272845c3be Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 12 Aug 2026 13:41:17 -0400 Subject: [PATCH 19/27] Simplify determining if replica locations are configured --- backend/btrixcloud/background_jobs.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index 489c043581..f4156aca91 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -600,10 +600,8 @@ async def ensure_cron_jobs_exist(self): """Ensure periodic background cron jobs exist""" await self.crawl_manager.ensure_cleanup_seed_file_cron_job_exists() await self.crawl_manager.ensure_retry_stuck_uploads_cron_job_exists() - - replicas_configured = True if self.storage_ops.get_default_replicas() else False await self.crawl_manager.ensure_file_replication_cron_job_exists( - replicas_configured + bool(self.storage_ops.get_default_replicas()) ) async def job_finished( From 6d87945738b5c686d9161851e3e306f1a4c41e91 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Wed, 12 Aug 2026 15:14:43 -0400 Subject: [PATCH 20/27] Try to fix nightly tests --- backend/test_nightly/test_crawl_timeout.py | 32 +++++--------------- backend/test_nightly/test_upload_replicas.py | 32 +++++--------------- 2 files changed, 16 insertions(+), 48 deletions(-) diff --git a/backend/test_nightly/test_crawl_timeout.py b/backend/test_nightly/test_crawl_timeout.py index 47884a6366..5651c05ce1 100644 --- a/backend/test_nightly/test_crawl_timeout.py +++ b/backend/test_nightly/test_crawl_timeout.py @@ -42,42 +42,26 @@ def test_crawl_files_replicated(admin_auth_headers, default_org_id, timeout_craw # Verify copy bucket job has run and succeeded since crawl completed job_id = None - # Give job up to 15 minutes to complete + # Give copy bucket job (which is kicked off by cron replication job) + # up to 20 minutes to start and then complete attempts = 0 - while attempts < 15: + while attempts < 20: r = requests.get( f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=-1&jobType=copy-bucket", headers=admin_auth_headers, ) assert r.status_code == 200 jobs = r.json().get("items", []) - if jobs: - latest_job = jobs[0] - assert latest_job["type"] == "copy-bucket" - if latest_job["started"] >= crawl_complete: - job_id = latest_job["id"] + for job in jobs: + assert job["type"] == "copy-bucket" + if job.get("started") >= crawl_complete and job.get("finished"): + job_id = job["id"] break attempts += 1 time.sleep(60) - # Give job up to 5 minutes to finish - attempts = 0 - while attempts < 10: - r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}", - headers=admin_auth_headers, - ) - assert r.status_code == 200 - job = r.json() - finished = latest_job.get("finished") - if not finished: - attempts += 1 - time.sleep(30) - continue - - assert job["success"] - break + assert job_id # Verify crawlfiles are stored in replica location r = requests.get( diff --git a/backend/test_nightly/test_upload_replicas.py b/backend/test_nightly/test_upload_replicas.py index 9736158ba9..a5a08711d7 100644 --- a/backend/test_nightly/test_upload_replicas.py +++ b/backend/test_nightly/test_upload_replicas.py @@ -43,42 +43,26 @@ def test_upload_file_replicated(admin_auth_headers, default_org_id): # Verify copy bucket job has run and succeeded since crawl completed job_id = None - # Give job up to 15 minutes to complete + # Give copy bucket job (which is kicked off by cron replication job) + # up to 20 minutes to start and then complete attempts = 0 - while attempts < 15: + while attempts < 20: r = requests.get( f"{API_PREFIX}/orgs/{default_org_id}/jobs?sortBy=started&sortDirection=-1&jobType=copy-bucket", headers=admin_auth_headers, ) assert r.status_code == 200 jobs = r.json().get("items", []) - if jobs: - latest_job = jobs[0] - assert latest_job["type"] == "copy-bucket" - if latest_job["started"] >= upload_complete: - job_id = latest_job["id"] + for job in jobs: + assert job["type"] == "copy-bucket" + if job.get("started") >= upload_complete and job.get("finished"): + job_id = job["id"] break attempts += 1 time.sleep(60) - # Give job up to 5 minutes to finish - attempts = 0 - while attempts < 10: - r = requests.get( - f"{API_PREFIX}/orgs/{default_org_id}/jobs/{job_id}", - headers=admin_auth_headers, - ) - assert r.status_code == 200 - job = r.json() - finished = latest_job.get("finished") - if not finished: - attempts += 1 - time.sleep(30) - continue - - assert job["success"] - break + assert job_id # Verify upload file is stored r = requests.get( From 3d75b28c85c11e3c9b742eba1158adb91f546b72 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Thu, 13 Aug 2026 15:03:57 -0400 Subject: [PATCH 21/27] Disable stuck uploads cron to ease load in nightly CI --- backend/btrixcloud/background_jobs.py | 4 ++- backend/btrixcloud/crawlmanager.py | 46 +++++++++++++++++---------- chart/templates/configmap.yaml | 2 ++ chart/test/test-nightly-addons.yaml | 3 ++ chart/values.yaml | 4 +++ 5 files changed, 41 insertions(+), 18 deletions(-) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index f4156aca91..a039c76731 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -599,7 +599,9 @@ async def create_postprocess_upload_job( async def ensure_cron_jobs_exist(self): """Ensure periodic background cron jobs exist""" await self.crawl_manager.ensure_cleanup_seed_file_cron_job_exists() - await self.crawl_manager.ensure_retry_stuck_uploads_cron_job_exists() + await self.crawl_manager.ensure_retry_stuck_uploads_cron_job_exists( + bool(os.environ.get("DISABLE_STUCK_UPLOADS_CRON", False)) + ) await self.crawl_manager.ensure_file_replication_cron_job_exists( bool(self.storage_ops.get_default_replicas()) ) diff --git a/backend/btrixcloud/crawlmanager.py b/backend/btrixcloud/crawlmanager.py index ab99b11e51..40d8b658f0 100644 --- a/backend/btrixcloud/crawlmanager.py +++ b/backend/btrixcloud/crawlmanager.py @@ -378,7 +378,9 @@ async def ensure_cleanup_seed_file_cron_job_exists(self): larger_resources=True, ) - async def ensure_retry_stuck_uploads_cron_job_exists(self): + async def ensure_retry_stuck_uploads_cron_job_exists( + self, disable_job: bool = False + ): """ensure cron background job to retry stuck uploads exists""" default_schedule = "0 * * * *" @@ -386,11 +388,19 @@ async def ensure_retry_stuck_uploads_cron_job_exists(self): "RETRY_STUCK_UPLOADS_CRON_SCHEDULE", default_schedule ) - await self._ensure_bg_cron_job_exists( - "retry-stuck-uploads-cron", - BgJobType.RETRY_STUCK_UPLOADS.value, - job_schedule, - ) + job_id = "retry-stuck-uploads-cron" + + if not disable_job: + return await self._ensure_bg_cron_job_exists( + job_id, + BgJobType.RETRY_STUCK_UPLOADS.value, + job_schedule, + ) + + # If no replica locations are configured, make sure no replication cron + # job exists, as one could have been previously configured. + logger.info("bg_cron_job_deleting", job_id=job_id, disable_job=True) + await self._delete_cron_job_if_exists(job_id) async def ensure_file_replication_cron_job_exists( self, replicas_configured: bool = False @@ -412,17 +422,8 @@ async def ensure_file_replication_cron_job_exists( # If no replica locations are configured, make sure no replication cron # job exists, as one could have been previously configured. - try: - await self.batch_api.delete_namespaced_cron_job( - name=job_id, - namespace=DEFAULT_NAMESPACE, - ) - logger.info( - "bg_cron_job_deleting", job_id=job_id, replicas_configured=False - ) - except ApiException as exc: - if exc.status != 404: - raise + logger.info("bg_cron_job_deleting", job_id=job_id, replicas_configured=False) + await self._delete_cron_job_if_exists(job_id) async def _ensure_bg_cron_job_exists( self, @@ -471,6 +472,17 @@ async def _ensure_bg_cron_job_exists( cron_logger.info("bg_cron_job_creating") await self.create_from_yaml(data, namespace=DEFAULT_NAMESPACE) + async def _delete_cron_job_if_exists(self, job_id: str): + """Delete cron job by id if it exists""" + try: + await self.batch_api.delete_namespaced_cron_job( + name=job_id, + namespace=DEFAULT_NAMESPACE, + ) + except ApiException as exc: + if exc.status != 404: + raise + async def create_crawl_job( self, crawlconfig: CrawlConfig, diff --git a/chart/templates/configmap.yaml b/chart/templates/configmap.yaml index ea50ca254d..5280b73482 100644 --- a/chart/templates/configmap.yaml +++ b/chart/templates/configmap.yaml @@ -127,6 +127,8 @@ data: RETRY_STUCK_UPLOADS_CRON_SCHEDULE: "{{ .Values.stuck_uploads_cron_schedule }}" + DISABLE_STUCK_UPLOADS_CRON: "{{ .Values.disable_stuck_uploads_cron }}" + DEDUPE_IMPORTER_CHANNEL: "{{ .Values.dedupe.importer_channel }}" ENABLE_AUTO_RESIZE_INDEX_STORAGE: "{{ .Values.dedupe.enable_auto_resize }}" diff --git a/chart/test/test-nightly-addons.yaml b/chart/test/test-nightly-addons.yaml index ed9b5dc9aa..7fc4380744 100644 --- a/chart/test/test-nightly-addons.yaml +++ b/chart/test/test-nightly-addons.yaml @@ -13,6 +13,9 @@ cleanup_files_after_minutes: 1 # Every 2 minutes, for use in testing file replication replication_job_cron_schedule: "*/2 * * * *" +# Disable retry uploads cron job to ease resource usage on CI workers +disable_stuck_uploads_cron: true + # max crawl queue size x concurrent crawls crawl_queue_limit_scale: 1 diff --git a/chart/values.yaml b/chart/values.yaml index 836bdaa271..e1ec5c0fc5 100644 --- a/chart/values.yaml +++ b/chart/values.yaml @@ -192,6 +192,10 @@ cleanup_files_after_minutes: 1440 # without a running background job and retrying them stuck_uploads_cron_schedule: "0 * * * *" +# Option to disable stuck uploads cron job +# Generally should only be set true for testing in CI +disable_stuck_uploads_cron: false + # Cron schedule for periodically syncing replica storage from primary # Defaults to every two hours replication_job_cron_schedule: "0 */2 * * *" From 74424d20222e5a0b4836fe8736b4019e268c9ee1 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Thu, 20 Aug 2026 17:13:17 -0400 Subject: [PATCH 22/27] Raise backoffLimit for other related jobs to 6 as well --- chart/app-templates/copy_bucket_job.yaml | 2 +- chart/app-templates/delete_replica_job.yaml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/chart/app-templates/copy_bucket_job.yaml b/chart/app-templates/copy_bucket_job.yaml index 480afadc47..255ead8ba6 100644 --- a/chart/app-templates/copy_bucket_job.yaml +++ b/chart/app-templates/copy_bucket_job.yaml @@ -8,7 +8,7 @@ metadata: spec: ttlSecondsAfterFinished: 300 - backoffLimit: 5 + backoffLimit: 6 template: spec: restartPolicy: Never diff --git a/chart/app-templates/delete_replica_job.yaml b/chart/app-templates/delete_replica_job.yaml index 63cbf9b998..bea8b75a1e 100644 --- a/chart/app-templates/delete_replica_job.yaml +++ b/chart/app-templates/delete_replica_job.yaml @@ -11,7 +11,7 @@ metadata: spec: ttlSecondsAfterFinished: 0 - backoffLimit: 5 + backoffLimit: 6 template: spec: restartPolicy: Never From e21864b17f005d1ba63a00a07abf833a7fdaf91b Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Thu, 20 Aug 2026 17:14:43 -0400 Subject: [PATCH 23/27] Remove podFailurePolicy from copy_bucket and delete_replica jobs It was in the wrong place anyway, and we always want to retry jobs, at least for now. --- chart/app-templates/copy_bucket_job.yaml | 7 ------- chart/app-templates/delete_replica_job.yaml | 7 ------- 2 files changed, 14 deletions(-) diff --git a/chart/app-templates/copy_bucket_job.yaml b/chart/app-templates/copy_bucket_job.yaml index 255ead8ba6..f7084506c3 100644 --- a/chart/app-templates/copy_bucket_job.yaml +++ b/chart/app-templates/copy_bucket_job.yaml @@ -13,13 +13,6 @@ spec: spec: restartPolicy: Never priorityClassName: bg-job - podFailurePolicy: - rules: - - action: FailJob - onExitCodes: - containerName: rclone - operator: NotIn - values: [0] containers: - name: rclone image: rclone/rclone:latest diff --git a/chart/app-templates/delete_replica_job.yaml b/chart/app-templates/delete_replica_job.yaml index bea8b75a1e..eff0861226 100644 --- a/chart/app-templates/delete_replica_job.yaml +++ b/chart/app-templates/delete_replica_job.yaml @@ -16,13 +16,6 @@ spec: spec: restartPolicy: Never priorityClassName: bg-job - podFailurePolicy: - rules: - - action: FailJob - onExitCodes: - containerName: rclone - operator: NotIn - values: [0] containers: - name: rclone image: rclone/rclone:latest From a6b80bbc892bc0cf90d154dd4e4605baad11b1f2 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Thu, 20 Aug 2026 17:16:24 -0400 Subject: [PATCH 24/27] Quote strings in templates --- chart/app-templates/copy_bucket_job.yaml | 2 +- chart/app-templates/delete_replica_job.yaml | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/chart/app-templates/copy_bucket_job.yaml b/chart/app-templates/copy_bucket_job.yaml index f7084506c3..9e431fada1 100644 --- a/chart/app-templates/copy_bucket_job.yaml +++ b/chart/app-templates/copy_bucket_job.yaml @@ -4,7 +4,7 @@ metadata: name: "{{ id }}" labels: role: "background-job" - job_type: {{ job_type }} + job_type: "{{ job_type }}" spec: ttlSecondsAfterFinished: 300 diff --git a/chart/app-templates/delete_replica_job.yaml b/chart/app-templates/delete_replica_job.yaml index eff0861226..05259a4d90 100644 --- a/chart/app-templates/delete_replica_job.yaml +++ b/chart/app-templates/delete_replica_job.yaml @@ -6,8 +6,8 @@ metadata: - metacontroller.io/decoratorcontroller-background-job-operator labels: role: "background-job" - job_type: {{ job_type }} - btrix.org: {{ oid }} + job_type: "{{ job_type }}" + btrix.org: "{{ oid }}" spec: ttlSecondsAfterFinished: 0 From 301231c1ebce29d599730e831425b86af32fdba3 Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Mon, 24 Aug 2026 16:50:25 -0400 Subject: [PATCH 25/27] Add explicit finalizer to copy_bucket_job yaml --- chart/app-templates/copy_bucket_job.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/chart/app-templates/copy_bucket_job.yaml b/chart/app-templates/copy_bucket_job.yaml index 9e431fada1..a9025a6ec3 100644 --- a/chart/app-templates/copy_bucket_job.yaml +++ b/chart/app-templates/copy_bucket_job.yaml @@ -2,6 +2,8 @@ apiVersion: batch/v1 kind: Job metadata: name: "{{ id }}" + finalizers: + - metacontroller.io/decoratorcontroller-background-job-operator labels: role: "background-job" job_type: "{{ job_type }}" From fe880c739fb83d72df4c26e15821463f8ada776b Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Mon, 24 Aug 2026 16:58:22 -0400 Subject: [PATCH 26/27] Change if to elif --- backend/btrixcloud/background_jobs.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/btrixcloud/background_jobs.py b/backend/btrixcloud/background_jobs.py index a039c76731..3cd080be38 100644 --- a/backend/btrixcloud/background_jobs.py +++ b/backend/btrixcloud/background_jobs.py @@ -634,7 +634,7 @@ async def job_finished( finished=finished, success=success, ) - if job_type == BgJobType.REPLICATE_FILES_CRON: + elif job_type == BgJobType.REPLICATE_FILES_CRON: cron_job = ReplicateFilesCronJob( id=f"replicate-cron-{secrets.token_hex(5)}", started=started, From a27e446afc34bbbaa1c88791dc1c056e3c9d0e4f Mon Sep 17 00:00:00 2001 From: Tessa Walsh Date: Tue, 25 Aug 2026 14:28:51 -0400 Subject: [PATCH 27/27] Remove unused test function arg --- backend/test_nightly/test_z_background_jobs.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/test_nightly/test_z_background_jobs.py b/backend/test_nightly/test_z_background_jobs.py index ebb8ad7afa..81b4994730 100644 --- a/backend/test_nightly/test_z_background_jobs.py +++ b/backend/test_nightly/test_z_background_jobs.py @@ -35,7 +35,7 @@ def test_background_jobs_list(admin_auth_headers, default_org_id, deleted_crawl_ def test_background_jobs_list_filter_by_type( - admin_auth_headers, default_org_id, deleted_crawl_id, job_type + admin_auth_headers, default_org_id, deleted_crawl_id ): r = requests.get( f"{API_PREFIX}/orgs/{default_org_id}/jobs/?jobType=delete-replica",