diff --git a/.github/workflows/integration-test.yml b/.github/workflows/integration-test.yml index 66838817..ba55560a 100644 --- a/.github/workflows/integration-test.yml +++ b/.github/workflows/integration-test.yml @@ -373,7 +373,17 @@ jobs: run: | NEW_FILES_COUNT=5 + # StatisticsService caches eligible_files_count in appconfig for an hour, because + # StorageService#countFiles() walks every mount. Both stats calls in this step would + # otherwise return the same stale number from before the scan, while the (uncached) + # vectordb counts do advance. Drop the cache so countFiles() runs again. + invalidate_eligible_files_cache() { + ./occ config:app:delete context_chat eligible_files_count + ./occ config:app:delete context_chat eligible_files_count_time + } + echo "::group::Stats before the empty files are added" + invalidate_eligible_files_cache prev_stats=$(./occ context_chat:stats --json) echo "$prev_stats" prev_scheduled=$(echo "$prev_stats" | jq '.queued_documents_counts.files__default // 0') @@ -432,6 +442,8 @@ jobs: # StorageService#countFiles filters size > 0, so empty files are NOT counted as eligible. # Only the 1 non-empty file is both eligible and indexed, so the gap must stay the same (diff=0). + invalidate_eligible_files_cache + stats=$(./occ context_chat:stats --json) eligible=$(echo "$stats" | jq '.eligible_files_count') indexed=$(echo "$stats" | jq '.vectordb_document_counts.files__default // 0') echo "prev: eligible=$prev_eligible indexed=$prev_indexed" diff --git a/context_chat_backend/controller.py b/context_chat_backend/controller.py index 12d86fa3..1d25efb8 100644 --- a/context_chat_backend/controller.py +++ b/context_chat_backend/controller.py @@ -86,19 +86,6 @@ app_enabled = threading.Event() -last_enabled_check: float | None = None -enabled_check_lock: threading.Lock = threading.Lock() -def get_enabled_state() -> bool: - global last_enabled_check - with enabled_check_lock: - if last_enabled_check is None or time.time() - last_enabled_check > 30: - nc = NextcloudApp() - if nc.enabled_state: - app_enabled.set() - else: - app_enabled.clear() - last_enabled_check = time.time() - return app_enabled.is_set() def enabled_handler(enabled: bool, nc: NextcloudApp | AsyncNextcloudApp) -> str: @@ -213,8 +200,7 @@ def enabled_handler(enabled: bool, nc: NextcloudApp | AsyncNextcloudApp) -> str: if THREAD_STOP_EVENT.is_set(): # If the threads were previously stopped, we start them again # otherwise the lifecycle handler has already started them - start_bg_threads(app_config, get_enabled_state) - THREAD_STOP_EVENT.clear() + start_bg_threads(app_config, app_enabled) else: app_enabled.clear() nc.providers.task_processing.unregister(SEARCH_PROVIDER_ID) @@ -237,8 +223,13 @@ async def lifespan(app: FastAPI): # k8s' rp role pulls tasks set_handlers(app, enabled_handler, models_to_fetch=models_to_fetch) - start_bg_threads(app_config, get_enabled_state) - logger.info(f'App enable state at startup: {get_enabled_state()}') + nc = NextcloudApp() + nc_app_enabled = nc.enabled_state + if nc_app_enabled: + app_enabled.set() + + start_bg_threads(app_config, app_enabled) + logger.info(f'App enable state at startup: {nc_app_enabled}') yield vectordb_loader.offload() wait_for_bg_threads() @@ -316,7 +307,7 @@ def decorator(func: Callable): @wraps(func) def wrapper(*args, **kwargs): disable_aaa = app.extra['CONFIG'].disable_aaa - if not disable_aaa and not get_enabled_state(): + if not disable_aaa and not app_enabled.is_set(): return JSONResponse('Context Chat is disabled, enable it from AppAPI to use it.', 503) return func(*args, **kwargs) @@ -337,7 +328,7 @@ def _(request: Request): @app.get('/enabled') def _(): - return JSONResponse(content={'enabled': get_enabled_state()}, status_code=200) + return JSONResponse(content={'enabled': app_enabled.is_set()}, status_code=200) @app.post('/countIndexedDocuments') diff --git a/context_chat_backend/task_fetcher.py b/context_chat_backend/task_fetcher.py index 69523979..ca2ab52a 100644 --- a/context_chat_backend/task_fetcher.py +++ b/context_chat_backend/task_fetcher.py @@ -70,7 +70,7 @@ class ThreadType(Enum): REQUEST_PROCESSING = 'request_processing' -def files_indexing_thread(app_config: TConfig, get_enabled_state) -> None: +def files_indexing_thread(app_config: TConfig, app_enabled: Event) -> None: try: network_em = NetworkEmbeddings(app_config) vectordb_loader = VectorDBLoader(app_config) @@ -145,7 +145,7 @@ def _load_sources(source_items: Mapping[int, SourceItem | ReceivedFileItem]) -> LOGGER.info('Files indexing thread is stopping due to stop event being set') return - if not get_enabled_state(): + if not app_enabled.is_set(): LOGGER.info('App is disabled, files indexing thread will sleep until next enabled state check') sleep(POLLING_COOLDOWN) continue @@ -295,7 +295,7 @@ def _load_sources(source_items: Mapping[int, SourceItem | ReceivedFileItem]) -> -def updates_processing_thread(app_config: TConfig, get_enabled_state) -> None: +def updates_processing_thread(app_config: TConfig, app_enabled: Event) -> None: try: vectordb_loader = VectorDBLoader(app_config) except LoaderException as e: @@ -308,7 +308,7 @@ def updates_processing_thread(app_config: TConfig, get_enabled_state) -> None: LOGGER.info('Updates processing thread is stopping due to stop event being set') return - if not get_enabled_state(): + if not app_enabled.is_set(): LOGGER.info('App is disabled, updates processing thread will sleep until next enabled state check') sleep(POLLING_COOLDOWN) continue @@ -481,7 +481,7 @@ def resolve_scope_list(source_ids: list[str], userId: str) -> list[str]: return ScopeList.model_validate(data).source_ids -def request_processing_thread(app_config: TConfig, get_enabled_state) -> None: +def request_processing_thread(app_config: TConfig, app_enabled: Event) -> None: LOGGER.info('Starting request processing thread') try: @@ -500,7 +500,7 @@ def request_processing_thread(app_config: TConfig, get_enabled_state) -> None: LOGGER.info('Request processing thread is stopping due to stop event being set') return - if not get_enabled_state(): + if not app_enabled.is_set(): LOGGER.info('App is disabled, request processing thread will sleep until next enabled state check') sleep(POLLING_COOLDOWN) continue @@ -727,7 +727,7 @@ def process_search_task( ) -def start_bg_threads(app_config: TConfig, get_enabled_state): +def start_bg_threads(app_config: TConfig, app_enabled: Event): THREAD_STOP_EVENT.clear() if APP_ROLE == AppRole.INDEXING or APP_ROLE == AppRole.NORMAL: @@ -737,7 +737,7 @@ def start_bg_threads(app_config: TConfig, get_enabled_state): THREADS[ThreadType.FILES_INDEXING] = Thread( target=files_indexing_thread, - args=(app_config,get_enabled_state), + args=(app_config, app_enabled), name='FilesIndexingThread', ) THREADS[ThreadType.FILES_INDEXING].start() @@ -749,7 +749,7 @@ def start_bg_threads(app_config: TConfig, get_enabled_state): THREADS[ThreadType.UPDATES_PROCESSING] = Thread( target=updates_processing_thread, - args=(app_config,get_enabled_state), + args=(app_config, app_enabled), name='UpdatesProcessingThread', ) THREADS[ThreadType.UPDATES_PROCESSING].start() @@ -761,7 +761,7 @@ def start_bg_threads(app_config: TConfig, get_enabled_state): THREADS[ThreadType.REQUEST_PROCESSING] = Thread( target=request_processing_thread, - args=(app_config,get_enabled_state), + args=(app_config, app_enabled), name='RequestProcessingThread', ) THREADS[ThreadType.REQUEST_PROCESSING].start() diff --git a/supervisord.conf b/supervisord.conf index 5b7baa44..791f34b5 100644 --- a/supervisord.conf +++ b/supervisord.conf @@ -22,6 +22,8 @@ stderr_logfile_maxbytes=0 autostart=true autorestart=true startsecs=120 +startretries=10 +priority=100 directory=/app command=/app/dockerfile_scripts/entrypoint.sh @@ -33,6 +35,8 @@ stderr_logfile_maxbytes=0 autostart=true autorestart=unexpected startsecs=120 +startretries=10 +priority=200 directory=/app command=/opt/venv/bin/python -u /app/main_em.py