Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 12 additions & 0 deletions .github/workflows/integration-test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -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')
Expand Down Expand Up @@ -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"
Expand Down
29 changes: 10 additions & 19 deletions context_chat_backend/controller.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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)
Expand All @@ -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()
Expand Down Expand Up @@ -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)
Expand All @@ -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')
Expand Down
20 changes: 10 additions & 10 deletions context_chat_backend/task_fetcher.py
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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
Expand Down Expand Up @@ -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:
Expand All @@ -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()
Expand All @@ -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()
Expand All @@ -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()
Expand Down
4 changes: 4 additions & 0 deletions supervisord.conf
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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

Expand Down
Loading