diff --git a/.github/workflows/integration-test-cron.yml b/.github/workflows/integration-test-cron.yml
index 004e5c75..162cf723 100644
--- a/.github/workflows/integration-test-cron.yml
+++ b/.github/workflows/integration-test-cron.yml
@@ -127,6 +127,8 @@ jobs:
with:
repository: nextcloud/context_chat_backend
path: context_chat_backend/
+ # todo: remove this
+ ref: feat/vector-db-wipe
persist-credentials: false
- name: Checkout viewer
@@ -410,6 +412,73 @@ jobs:
kill -9 $WORKER1_PID
kill -9 $WORKER2_PID
+ - name: Wipe and reindex only the files provider
+ run: |
+ ./occ context_chat:wipe --force --provider files__default --reindex
+
+ stats=$(./occ context_chat:stats --json)
+ echo "$stats"
+ indexed=$(echo "$stats" | jq '.vectordb_document_counts.files__default // 0')
+ if [ "$indexed" != "0" ]; then
+ echo "Error: expected no indexed files right after the files wipe, got $indexed"
+ exit 1
+ fi
+
+ echo "::group::Waiting up to 10 minutes for the files to be indexed again"
+ for i in {1..60}; do
+ stats=$(timeout 30 ./occ context_chat:stats --json || true)
+ indexed=$(echo "$stats" | jq '.vectordb_document_counts.files__default // 0' 2>/dev/null || echo 0)
+ echo "Attempt $i: indexed=$indexed"
+ if [ "$indexed" -ge 10 ]; then
+ echo "::endgroup::"
+ echo "Files reindexed, $indexed documents"
+ exit 0
+ fi
+ sleep 10
+ done
+ echo "::endgroup::"
+ echo "Error: files were not indexed again after the files wipe"
+ exit 1
+
+ - name: Wipe all the indexed data and check that the queues and the vector DB are empty
+ run: |
+ ./occ context_chat:wipe --force
+
+ stats=$(./occ context_chat:stats --json)
+ echo "$stats"
+ indexed=$(echo "$stats" | jq '[.vectordb_document_counts | to_entries[].value] | add // 0')
+ queued=$(echo "$stats" | jq '[.queued_documents_counts | to_entries[].value] | add // 0')
+ actions=$(echo "$stats" | jq '.queued_actions_count')
+ fs_events=$(echo "$stats" | jq '.queued_fs_events_count')
+ # the queued counts leave out locked items, check those separately
+ queued_locked=$(echo "$stats" | jq '[.queued_documents_locked_counts | to_entries[].value] | add // 0')
+ actions_locked=$(echo "$stats" | jq '.queued_actions_locked_count')
+ backend_available=$(echo "$stats" | jq '.backend_available')
+ echo "indexed=$indexed queued=$queued queued_locked=$queued_locked actions=$actions actions_locked=$actions_locked fs_events=$fs_events backend_available=$backend_available"
+ if [ "$backend_available" != "true" ] || [ "$indexed" != "0" ] || [ "$queued" != "0" ] || [ "$queued_locked" != "0" ] || [ "$actions" != "0" ] || [ "$actions_locked" != "0" ] || [ "$fs_events" != "0" ]; then
+ echo "Error: expected an empty vector DB and empty queues after the wipe"
+ exit 1
+ fi
+
+ - name: Reindex and check that documents get indexed again
+ run: |
+ ./occ context_chat:reindex
+ echo "::group::Waiting up to 10 minutes for documents to be indexed again"
+ for i in {1..60}; do
+ stats=$(timeout 30 ./occ context_chat:stats --json || true)
+ indexed=$(echo "$stats" | jq '.vectordb_document_counts.files__default // 0' 2>/dev/null || echo 0)
+ echo "Attempt $i: indexed=$indexed"
+ if [ "$indexed" -ge 10 ]; then
+ echo "Reindex works, $indexed documents indexed again"
+ echo "::endgroup::"
+ exit 0
+ fi
+ sleep 10
+ done
+ echo "::endgroup::"
+ echo "Error: documents were not indexed again after the reindex"
+ exit 1
+
- name: Show nextcloud logs
if: always()
run: |
diff --git a/.github/workflows/integration-test-file-listener.yml b/.github/workflows/integration-test-file-listener.yml
index 3a5e2ac3..2664d0f0 100644
--- a/.github/workflows/integration-test-file-listener.yml
+++ b/.github/workflows/integration-test-file-listener.yml
@@ -115,6 +115,8 @@ jobs:
with:
repository: nextcloud/context_chat_backend
path: context_chat_backend/
+ # todo: remove this
+ ref: feat/vector-db-wipe
persist-credentials: false
- name: Checkout viewer
diff --git a/appinfo/info.xml b/appinfo/info.xml
index 2b8cda78..7075cb78 100644
--- a/appinfo/info.xml
+++ b/appinfo/info.xml
@@ -47,6 +47,7 @@ Refer to the [Context Chat Backend's readme](https://github.com/nextcloud/contex
OCA\ContextChat\Command\Search
OCA\ContextChat\Command\Statistics
OCA\ContextChat\Command\Reindex
+ OCA\ContextChat\Command\Wipe
diff --git a/lib/BackgroundJobs/InitialContentImportJob.php b/lib/BackgroundJobs/InitialContentImportJob.php
index dbf68163..4df3b05c 100644
--- a/lib/BackgroundJobs/InitialContentImportJob.php
+++ b/lib/BackgroundJobs/InitialContentImportJob.php
@@ -11,6 +11,7 @@
use OCA\ContextChat\Logger;
use OCA\ContextChat\Public\IContentProvider as OCAIContentProvider;
+use OCA\ContextChat\Service\FetchGuard;
use OCA\ContextChat\Service\ProviderConfigService;
use OCP\App\IAppManager;
use OCP\AppFramework\Utility\ITimeFactory;
@@ -29,6 +30,7 @@ public function __construct(
private IUserManager $userMan,
ITimeFactory $timeFactory,
private ?string $userId,
+ private FetchGuard $fetchGuard,
) {
parent::__construct($timeFactory);
}
@@ -56,16 +58,26 @@ protected function run($argument): void {
return;
}
- $registeredProviders = $this->providerConfig->getProviders();
$identifier = ProviderConfigService::getConfigKey($providerObj->getAppId(), $providerObj->getId());
- if (!isset($registeredProviders[$identifier])
- || $registeredProviders[$identifier]['isInitiated']
- ) {
- $this->logger->info('[InitialContentImportJob] Provider has already been initiated, skipping content import', ['provider' => $identifier]);
- return;
- }
+ $running = $this->fetchGuard->jobStarted($identifier, $this->getId(), FetchGuard::IMPORT_RUNNING_TTL);
+ try {
+ // a wipe of the provider removes this job, a re-index after it adds it again
+ if ($this->fetchGuard->isPaused($identifier)) {
+ return;
+ }
- $providerObj->triggerInitialImport();
- $this->providerConfig->updateProvider($providerObj->getAppId(), $providerObj->getId(), $argument, true);
+ $registeredProviders = $this->providerConfig->getProviders();
+ if (!isset($registeredProviders[$identifier])
+ || $registeredProviders[$identifier]['isInitiated']
+ ) {
+ $this->logger->info('[InitialContentImportJob] Provider has already been initiated, skipping content import', ['provider' => $identifier]);
+ return;
+ }
+
+ $providerObj->triggerInitialImport();
+ $this->providerConfig->updateProvider($providerObj->getAppId(), $providerObj->getId(), $argument, isInitiated: true);
+ } finally {
+ $this->fetchGuard->jobEnded($running);
+ }
}
}
diff --git a/lib/BackgroundJobs/SchedulerJob.php b/lib/BackgroundJobs/SchedulerJob.php
index 47b0ddb6..3abb335f 100644
--- a/lib/BackgroundJobs/SchedulerJob.php
+++ b/lib/BackgroundJobs/SchedulerJob.php
@@ -11,6 +11,8 @@
namespace OCA\ContextChat\BackgroundJobs;
use OCA\ContextChat\Logger;
+use OCA\ContextChat\Service\FetchGuard;
+use OCA\ContextChat\Service\ProviderConfigService;
use OCA\ContextChat\Service\StorageService;
use OCP\AppFramework\Services\IAppConfig;
use OCP\AppFramework\Utility\ITimeFactory;
@@ -26,6 +28,7 @@ public function __construct(
private IJobList $jobList,
private StorageService $storageService,
private IAppConfig $appConfig,
+ private FetchGuard $fetchGuard,
) {
parent::__construct($timeFactory);
}
@@ -34,17 +37,27 @@ public function __construct(
* @throws Exception
*/
protected function run($argument): void {
- $this->appConfig->setAppValueInt('last_indexed_time', 0, lazy: true);
- foreach ($this->storageService->getMounts() as $mount) {
- $this->logger->debug('Scheduling StorageCrawlJob storage_id=' . $mount['storage_id'] . ' root_id=' . $mount['root_id' ] . 'override_root=' . $mount['overridden_root']);
- $this->jobList->add(StorageCrawlJob::class, [
- 'storage_id' => $mount['storage_id'],
- 'root_id' => $mount['root_id' ],
- 'overridden_root' => $mount['overridden_root'],
- 'last_file_id' => 0,
- ]);
+ $filesKey = ProviderConfigService::getDefaultProviderKey();
+ $running = $this->fetchGuard->jobStarted($filesKey, $this->getId());
+ try {
+ // a wipe of the files removes this job, a re-index after it adds it again
+ if ($this->fetchGuard->isPaused($filesKey)) {
+ return;
+ }
+ foreach ($this->storageService->getMounts() as $mount) {
+ $this->logger->debug('Scheduling StorageCrawlJob storage_id=' . $mount['storage_id'] . ' root_id=' . $mount['root_id' ] . 'override_root=' . $mount['overridden_root']);
+ $this->jobList->add(StorageCrawlJob::class, [
+ 'storage_id' => $mount['storage_id'],
+ 'root_id' => $mount['root_id' ],
+ 'overridden_root' => $mount['overridden_root'],
+ 'last_file_id' => 0,
+ ]);
+ }
+
+ $this->jobList->remove(self::class);
+ $this->appConfig->setAppValueInt('last_indexed_time', 0, lazy: true);
+ } finally {
+ $this->fetchGuard->jobEnded($running);
}
-
- $this->jobList->remove(self::class);
}
}
diff --git a/lib/BackgroundJobs/StorageCrawlJob.php b/lib/BackgroundJobs/StorageCrawlJob.php
index 4c0bc8ed..25fbc11c 100644
--- a/lib/BackgroundJobs/StorageCrawlJob.php
+++ b/lib/BackgroundJobs/StorageCrawlJob.php
@@ -13,6 +13,8 @@
use OCA\ContextChat\Db\QueueFile;
use OCA\ContextChat\Logger;
use OCA\ContextChat\Service\DiagnosticService;
+use OCA\ContextChat\Service\FetchGuard;
+use OCA\ContextChat\Service\ProviderConfigService;
use OCA\ContextChat\Service\QueueService;
use OCA\ContextChat\Service\StorageService;
use OCP\AppFramework\Services\IAppConfig;
@@ -33,6 +35,7 @@ public function __construct(
private StorageService $storageService,
private DiagnosticService $diagnosticService,
private IAppConfig $appConfig,
+ private FetchGuard $fetchGuard,
) {
parent::__construct($timeFactory);
}
@@ -50,7 +53,13 @@ protected function run($argument): void {
// Remove current iteration
$this->jobList->remove(self::class, $argument);
+ $filesKey = ProviderConfigService::getDefaultProviderKey();
+ $running = $this->fetchGuard->jobStarted($filesKey, $this->getId());
try {
+ // wipe of everything/for files is running, skip the crawl
+ if ($this->fetchGuard->isPaused($filesKey)) {
+ return;
+ }
$this->diagnosticService->sendJobStart(static::class, $this->getId());
$this->diagnosticService->sendHeartbeat(static::class, $this->getId());
@@ -64,6 +73,7 @@ protected function run($argument): void {
$queueFile->setFileId($fileId);
$queueFile->setUpdate(false);
$this->diagnosticService->sendHeartbeat(static::class, $this->getId());
+ $this->fetchGuard->jobAlive($running);
try {
$dbFile = $this->queue->insertIntoQueue($queueFile);
$lastSuccessfulDbId = $dbFile->getId();
@@ -96,6 +106,7 @@ protected function run($argument): void {
}
} finally {
$this->diagnosticService->sendJobEnd(static::class, $this->getId());
+ $this->fetchGuard->jobEnded($running);
}
}
diff --git a/lib/Command/Reindex.php b/lib/Command/Reindex.php
index ca3de909..e3b2dc6f 100644
--- a/lib/Command/Reindex.php
+++ b/lib/Command/Reindex.php
@@ -9,10 +9,10 @@
namespace OCA\ContextChat\Command;
-use OCA\ContextChat\BackgroundJobs\SchedulerJob;
-use OCP\BackgroundJob\IJobList;
+use OCA\ContextChat\Service\ReindexService;
use Symfony\Component\Console\Command\Command;
use Symfony\Component\Console\Input\InputInterface;
+use Symfony\Component\Console\Input\InputOption;
use Symfony\Component\Console\Output\OutputInterface;
/**
@@ -20,32 +20,55 @@
*
* SchedulerJob -> StorageCrawlJob -> IndexerJob is seeded only by the repair step and
* removes itself once the initial crawl finishes. There is otherwise no way to re-enumerate
- * mounts (e.g. after installing on an instance whose files predate the app, or to recover a crawl
- * that did not complete) short of reinstalling the app. This command re-adds SchedulerJob so the
- * full enumeration runs again; it is a no-op if one is already scheduled, and already-indexed
- * files are skipped.
+ * mounts (e.g. to recover a crawl that did not complete, or after context_chat:wipe) short of
+ * reinstalling the app. This command re-adds SchedulerJob so the full enumeration runs again if one
+ * doesn't exist already.
+ * Content providers' initial import also gets scheduled (e.g. after context_chat:wipe) if it has not run yet.
*/
class Reindex extends Command {
public function __construct(
- private IJobList $jobList,
+ private ReindexService $reindexService,
) {
parent::__construct();
}
protected function configure() {
$this->setName('context_chat:reindex')
- ->setDescription('Schedule a full re-crawl of all the files in all the mounts. Indexed files are not re-indexed when compared against context_chat_backend\'s vector DB. Content providers are not re-indexed.');
+ ->setDescription(
+ 'Schedule a full re-crawl of all the files in all the mounts and the import of the content providers'
+ . ' that were not imported yet. Indexed files are not re-indexed when compared against context_chat_backend\'s vector DB.'
+ )
+ ->addOption(
+ 'provider',
+ 'p',
+ InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY,
+ 'Only reindex this provider, e.g. "files__default" (appId__providerId) for the files.'
+ . ' A named content provider is imported again even if it was imported before. Multiple values can be specified by repeating the option.',
+ );
}
protected function execute(InputInterface $input, OutputInterface $output): int {
- if ($this->jobList->has(SchedulerJob::class, null)) {
- $output->writeln('A full re-crawl is already scheduled; nothing to do.');
+ /** @var list $providerKeys */
+ $providerKeys = $input->getOption('provider');
+ $scope = $providerKeys === [] ? null : $providerKeys;
+
+ if ($scope !== null) {
+ $known = $this->reindexService->getRegisteredProviderKeys();
+ $unknown = array_diff($scope, $known);
+ if ($unknown !== []) {
+ $output->writeln('Unknown provider(s): ' . implode(', ', $unknown) . '');
+ $output->writeln('Registered providers: ' . implode(', ', $known));
+ return 1;
+ }
+ }
+
+ if (!$this->reindexService->scheduleReindex($scope)) {
+ $output->writeln('The re-index is already scheduled; nothing to do.');
return 0;
}
- $this->jobList->add(SchedulerJob::class);
- $output->writeln('Scheduled a full re-crawl. SchedulerJob will enumerate all mounts on its next run.');
+ $output->writeln('Scheduled the re-index. The background jobs will queue the documents on their next run.');
return 0;
}
}
diff --git a/lib/Command/Wipe.php b/lib/Command/Wipe.php
new file mode 100644
index 00000000..07334816
--- /dev/null
+++ b/lib/Command/Wipe.php
@@ -0,0 +1,142 @@
+setName('context_chat:wipe')
+ ->setDescription(
+ 'Delete the indexed documents from context_chat_backend\'s vector DB and clear the queues.'
+ . ' With --provider option, only the provider\'s data is deleted, the actions queue, and the indexing statistics are left intact.'
+ . ' Nothing gets indexed again until context_chat:reindex is run or the --reindex option is used.'
+ . ' While the wipe runs, no documents or actions are handed to context_chat_backend. If the command is killed'
+ . ' before it ends, this pause stays for up to a day, run the command again to lift it.'
+ )
+ ->addOption(
+ 'provider',
+ 'p',
+ InputOption::VALUE_REQUIRED | InputOption::VALUE_IS_ARRAY,
+ 'Only wipe this provider, e.g. "files__default" (appId__providerId) for the files. Multiple values can be specified by repeating the option.',
+ )
+ ->addOption(
+ 'reindex',
+ null,
+ InputOption::VALUE_NONE,
+ 'Schedule a re-crawl of the files and a re-import of the content providers that were wiped',
+ )
+ ->addOption(
+ 'ignore-handed-out',
+ null,
+ InputOption::VALUE_NONE,
+ 'Do not wait for the documents context_chat_backend is still indexing. Only use this when the backend was stopped while indexing them, otherwise they can be written to the vector DB after the wipe.',
+ )
+ ->addOption(
+ 'force',
+ 'f',
+ InputOption::VALUE_NONE,
+ 'Do not ask for confirmation',
+ );
+ }
+
+ protected function execute(InputInterface $input, OutputInterface $output): int {
+ /** @var list $providerKeys */
+ $providerKeys = $input->getOption('provider');
+ $scope = $providerKeys === [] ? null : $providerKeys;
+
+ if ($scope !== null) {
+ try {
+ $known = $this->reindexService->getWipeableProviderKeys();
+ } catch (\RuntimeException $e) {
+ $output->writeln('Could not list the indexed providers: ' . $e->getMessage() . '');
+ return 1;
+ }
+ $unknown = array_diff($scope, $known);
+ if ($unknown !== []) {
+ $output->writeln('Unknown provider(s): ' . implode(', ', $unknown) . '');
+ $output->writeln('Known providers: ' . implode(', ', $known));
+ return 1;
+ }
+ }
+
+ $scopeText = $scope === null ? '' : ' of ' . implode(', ', $scope);
+
+ if (!$input->getOption('force')) {
+ /** @var QuestionHelper $helper */
+ $helper = $this->getHelper('question');
+ $question = new ConfirmationQuestion(
+ 'All the indexed documents in Context Chat'
+ . $scopeText
+ . ' will be deleted. The original documents are not touched. Continue? [y/N] ',
+ false,
+ );
+ if (!$helper->ask($input, $output, $question)) {
+ $output->writeln('Aborted, nothing was deleted. Use --force to skip the confirmation.');
+ return 1;
+ }
+ }
+
+ try {
+ $this->reindexService->wipe(
+ $scope,
+ (bool)$input->getOption('ignore-handed-out'),
+ static function (string $message) use ($output): void {
+ $output->writeln('' . $message . '...');
+ },
+ );
+ } catch (\Throwable $e) {
+ $output->writeln('Wipe failed. The command can be safely run again once the problem is fixed.');
+ $output->writeln('Error: ' . $e->getMessage() . '');
+ $output->writeln('' . $e->getTraceAsString() . '');
+ return 1;
+ }
+ $output->writeln('Deleted all the indexed documents' . $scopeText . ' and cleared the queues.');
+
+ // providers that are not registered anymore cannot be indexed again
+ $reindexScope = $scope === null
+ ? null
+ : array_values(array_intersect($scope, $this->reindexService->getRegisteredProviderKeys()));
+ if ($reindexScope === []) {
+ return 0;
+ }
+
+ if ($input->getOption('reindex')) {
+ try {
+ $this->reindexService->scheduleReindex($reindexScope);
+ } catch (\Throwable $e) {
+ $output->writeln('Scheduling re-index failed. It can be tried again with context_chat:reindex.');
+ $output->writeln('Error: ' . $e->getMessage() . '');
+ $output->writeln('' . $e->getTraceAsString() . '');
+ return 1;
+ }
+ $output->writeln('Scheduled the re-index.');
+ } elseif ($reindexScope === null) {
+ $output->writeln('Run context_chat:reindex to index the documents again.');
+ } else {
+ $output->writeln('Run context_chat:reindex --provider ' . implode(' --provider ', $reindexScope) . ' to index the documents again.');
+ }
+
+ return 0;
+ }
+}
diff --git a/lib/Controller/QueueController.php b/lib/Controller/QueueController.php
index f35eaa14..d2bcb303 100644
--- a/lib/Controller/QueueController.php
+++ b/lib/Controller/QueueController.php
@@ -17,6 +17,7 @@
use OCA\ContextChat\Db\QueueContentItemMapper;
use OCA\ContextChat\Db\QueueFile;
use OCA\ContextChat\Db\QueueMapper;
+use OCA\ContextChat\Service\FetchGuard;
use OCA\ContextChat\Service\ProviderConfigService;
use OCA\ContextChat\Type\Source;
use OCP\AppFramework\Http;
@@ -47,6 +48,7 @@ public function __construct(
private IJobList $jobList,
private ITimeFactory $timeFactory,
private QueueMapper $queueMapper,
+ private FetchGuard $fetchGuard,
string $corsMethods = 'PUT, POST, GET, DELETE, PATCH',
string $corsAllowedHeaders = 'Authorization, Content-Type, Accept, OCS-APIRequest',
int $corsMaxAge = 1728000,
@@ -106,9 +108,21 @@ public function getDocumentsQueueItems(
if ($n > $maxN) {
$n = $maxN;
}
+
+ // hand out nothing, the backend waits and asks again
+ // wipe everything case, only files wipe or a content provider wipe does not pause here
+ if ($this->fetchGuard->isPaused()) {
+ return new DataResponse([
+ 'files' => new \stdClass(),
+ 'content_providers' => new \stdClass(),
+ ]);
+ }
+
try {
+ $filesKey = ProviderConfigService::getDefaultProviderKey();
$files = [];
- while (count($files) < $n) {
+ // a wipe of the files holds back the files only
+ while (count($files) < $n && !$this->fetchGuard->isPaused($filesKey)) {
$limit = $n - count($files);
$documents = $queueMapper->getFromQueue($limit);
if (empty($documents)) {
@@ -125,6 +139,7 @@ public function getDocumentsQueueItems(
}
}
}
+
$contentItems = [];
while (count($contentItems) < $n) {
$limit = $n - count($contentItems);
@@ -138,6 +153,23 @@ public function getDocumentsQueueItems(
}
}
}
+
+ // A wipe may have started after the pause check. Do a check again here before returning the
+ // response to see if that happened. If the pause did happen in between the checks,
+ // wipe would be waiting for the locks to unlock, which we do, or proceed with the cleanup, which
+ // is fine too since we don't return the docs to the backend, it won't be indexing them.
+ //
+ // Only the files and a full wipe pause. A wipe of a content provider deletes the provider's
+ // items that are not locked, so a fetch can no longer lock them, and waits for the locked ones.
+ if ($files !== [] && $this->fetchGuard->isPaused($filesKey)) {
+ $queueMapper->unlock(array_keys($files));
+ $files = [];
+ }
+ if ($contentItems !== [] && $this->fetchGuard->isPaused()) {
+ $queueContentItemMapper->unlock(array_keys($contentItems));
+ $contentItems = [];
+ }
+
return new DataResponse([
'files' => (object)$files,
'content_providers' => (object)$contentItems,
@@ -216,6 +248,10 @@ public function countDocumentsQueueItems(QueueMapper $queueMapper, QueueContentI
#[ExAppRequired]
#[ApiRoute(verb: 'GET', url: '/queues/actions/')]
public function getActionsQueueItems(QueueActionMapper $queueActionMapper, int $n = 512) : DataResponse {
+ if ($this->fetchGuard->isPaused()) {
+ // stdClass is used to represent an empty object in the response
+ return new DataResponse(['actions' => new \stdClass()]);
+ }
try {
$actions = [];
while (count($actions) < $n) {
@@ -230,6 +266,11 @@ public function getActionsQueueItems(QueueActionMapper $queueActionMapper, int $
}
}
}
+ // see getDocumentsQueueItems(), a full wipe may have started since the previous pause check above
+ if ($actions !== [] && $this->fetchGuard->isPaused()) {
+ $queueActionMapper->unlock(array_keys($actions));
+ $actions = [];
+ }
return new DataResponse(['actions' => (object)$actions]);
} catch (Exception $e) {
$this->logger->error($e->getMessage(), ['exception' => $e]);
diff --git a/lib/Db/QueueActionMapper.php b/lib/Db/QueueActionMapper.php
index ac367dfb..95699018 100644
--- a/lib/Db/QueueActionMapper.php
+++ b/lib/Db/QueueActionMapper.php
@@ -154,4 +154,17 @@ public function lock(int $id) : bool {
}
return false;
}
+
+ /**
+ * @throws Exception
+ */
+ public function unlock(array $ids): void {
+ foreach (array_chunk($ids, 1000) as $chunk) {
+ $qb = $this->db->getQueryBuilder();
+ $qb->update($this->getTableName())
+ ->set('locked_at', $qb->createNamedParameter(null, IQueryBuilder::PARAM_NULL))
+ ->where($qb->expr()->in('id', $qb->createNamedParameter($chunk, IQueryBuilder::PARAM_INT_ARRAY)))
+ ->executeStatement();
+ }
+ }
}
diff --git a/lib/Db/QueueContentItemMapper.php b/lib/Db/QueueContentItemMapper.php
index 2cc79cfc..6b3e82ae 100644
--- a/lib/Db/QueueContentItemMapper.php
+++ b/lib/Db/QueueContentItemMapper.php
@@ -71,6 +71,28 @@ public function removeFromQueue(array $ids): void {
}
}
+ /**
+ * @param bool $onlyUnlocked keep the items being indexed, i.e. locked within the lock timeout
+ * @throws \OCP\DB\Exception
+ */
+ public function removeProvider(string $appId, string $providerId, bool $onlyUnlocked = false): void {
+ $qb = $this->db->getQueryBuilder();
+ $qb->delete($this->getTableName())
+ ->where($qb->expr()->eq('app_id', $qb->createNamedParameter($appId)))
+ ->andWhere($qb->expr()->eq('provider_id', $qb->createNamedParameter($providerId)));
+ if ($onlyUnlocked) {
+ // the same items getFromQueue() can hand out
+ $qb->andWhere($qb->expr()->orX(
+ $qb->expr()->isNull('locked_at'),
+ $qb->expr()->lte('locked_at', $qb->createNamedParameter(
+ (new \DateTime())->sub(new \DateInterval('PT' . self::LOCK_TIMEOUT . 'S')),
+ IQueryBuilder::PARAM_DATETIME_MUTABLE
+ )),
+ ));
+ }
+ $qb->executeStatement();
+ }
+
/**
* @throws \OCP\DB\Exception
* @return array
@@ -174,4 +196,17 @@ public function lock(int $id) : bool {
}
return false;
}
+
+ /**
+ * @throws Exception
+ */
+ public function unlock(array $ids): void {
+ foreach (array_chunk($ids, 1000) as $chunk) {
+ $qb = $this->db->getQueryBuilder();
+ $qb->update($this->getTableName())
+ ->set('locked_at', $qb->createNamedParameter(null, IQueryBuilder::PARAM_NULL))
+ ->where($qb->expr()->in('id', $qb->createNamedParameter($chunk, IQueryBuilder::PARAM_INT_ARRAY)))
+ ->executeStatement();
+ }
+ }
}
diff --git a/lib/Db/QueueMapper.php b/lib/Db/QueueMapper.php
index c444e827..37cb3777 100644
--- a/lib/Db/QueueMapper.php
+++ b/lib/Db/QueueMapper.php
@@ -244,4 +244,17 @@ public function lock(int $id) : bool {
}
return false;
}
+
+ /**
+ * @throws Exception
+ */
+ public function unlock(array $ids): void {
+ foreach (array_chunk($ids, 1000) as $chunk) {
+ $qb = $this->db->getQueryBuilder();
+ $qb->update($this->getTableName())
+ ->set('locked_at', $qb->createNamedParameter(null, IQueryBuilder::PARAM_NULL))
+ ->where($qb->expr()->in('id', $qb->createNamedParameter($chunk, IQueryBuilder::PARAM_INT_ARRAY)))
+ ->executeStatement();
+ }
+ }
}
diff --git a/lib/Service/FetchGuard.php b/lib/Service/FetchGuard.php
new file mode 100644
index 00000000..5f105d5c
--- /dev/null
+++ b/lib/Service/FetchGuard.php
@@ -0,0 +1,276 @@
+" => until))
+ private const PROVIDER_KEY_PREFIX = 'paused_until_';
+ // a background job queueing items: ("running__" => until)
+ private const RUNNING_KEY_PREFIX = 'running_';
+ // pause indexing while wipe happens, max 1 day so a crashed process does not stop it indefinitely
+ private const PAUSE_DURATION = 24 * 60 * 60;
+ public const DEFAULT_DRAIN_TIMEOUT = 30 * 60; // 30 mins for the currently indexing documents to be done
+ // bg jobs except initial import refreshes its record while it runs, a killed job blocks wipes for at most this long
+ public const RUNNING_TTL = 10 * 60;
+ // an initial import job cannot refresh its record while the app imports, so it has to last for the whole import
+ public const IMPORT_RUNNING_TTL = 60 * 60;
+
+ public function __construct(
+ private IDBConnection $db,
+ private IAppConfig $appConfig,
+ private ITimeFactory $timeFactory,
+ private QueueMapper $queueMapper,
+ private QueueContentItemMapper $contentQueue,
+ ) {
+ }
+
+ /**
+ * Starts a wipe: what is queued for these providers is held back until resume()
+ *
+ * @param list|null $providerKeys null for a full wipe
+ * @return array the pause object, to be passed to resume()
+ */
+ public function pause(?array $providerKeys = null): array {
+ $until = $this->timeFactory->getTime() + self::PAUSE_DURATION;
+ $keys = $providerKeys === null ? [self::ALL_KEY] : array_map($this->getPauseKey(...), $providerKeys);
+ $pause = [];
+ foreach ($keys as $key) {
+ $this->appConfig->setAppValueInt($key, $until, lazy: true);
+ $pause[$key] = $until;
+ }
+ return $pause;
+ }
+
+ /**
+ * Lifts the pause, only the ones started by the caller, with a matching until
+ *
+ * @param array $pause
+ */
+ public function resume(array $pause): void {
+ foreach ($pause as $key => $until) {
+ if ($this->readUntil($key) === $until) {
+ $this->appConfig->deleteAppValue($key);
+ }
+ }
+ }
+
+ /**
+ * Whether a wipe holds back:
+ * - null: everything, nothing is handed out, not even actions. A full wipe is running.
+ * - the files: the files, they are neither handed out nor crawled. A full wipe or a wipe of the files is running.
+ * - a content provider: its import job, which must stop. A full wipe or a wipe of the provider is running.
+ * Its queued items are still handed out, unless it is a full wipe.
+ */
+ public function isPaused(?string $providerKey = null): bool {
+ $now = $this->timeFactory->getTime();
+ if ($this->readUntil(self::ALL_KEY) > $now) {
+ return true;
+ }
+ return $providerKey !== null && $this->readUntil($this->getPauseKey($providerKey)) > $now;
+ }
+
+ private function getPauseKey(string $providerKey): string {
+ // keep the key length within 64 chars
+ return $providerKey === ProviderConfigService::getDefaultProviderKey()
+ ? self::FILES_KEY
+ : self::PROVIDER_KEY_PREFIX . md5($providerKey);
+ }
+
+ /**
+ * Records a background job queueing items of a provider, until it calls jobEnded(). A wipe of the
+ * provider waits for it with waitForRunningJobs().
+ *
+ * The job records itself before it checks isPaused(), and a wipe pauses before it looks for running
+ * jobs, so either the wipe waits for the job or the job sees the wipe and stops.
+ *
+ * @return string the record, to be passed to jobAlive() and jobEnded()
+ */
+ public function jobStarted(string $providerKey, int|string $jobId, int $ttl = self::RUNNING_TTL): string {
+ $key = self::RUNNING_KEY_PREFIX . $this->getProviderConfigPartKey($providerKey) . '_' . $jobId;
+ $this->appConfig->setAppValueInt($key, $this->timeFactory->getTime() + $ttl, lazy: true);
+ return $key;
+ }
+
+ /**
+ * Extends the record of a job that is still running, at most once per half of its lifetime.
+ */
+ public function jobAlive(string $record): void {
+ $until = $this->appConfig->getAppValueInt($record, 0, lazy: true);
+ $now = $this->timeFactory->getTime();
+ if ($until - $now < self::RUNNING_TTL / 2) {
+ $this->appConfig->setAppValueInt($record, $now + self::RUNNING_TTL, lazy: true);
+ }
+ }
+
+ public function jobEnded(string $record): void {
+ $this->appConfig->deleteAppValue($record);
+ }
+
+ private function getProviderConfigPartKey(string $providerKey): string {
+ // a provider key can be longer than an app config key may be
+ return $providerKey === ProviderConfigService::getDefaultProviderKey() ? 'files' : md5($providerKey);
+ }
+
+ /**
+ * @param list $prefixes
+ */
+ private function countRunningJobs(array $prefixes): int {
+ $now = $this->timeFactory->getTime();
+ $count = 0;
+ foreach ($prefixes as $prefix) {
+ foreach ($this->readValues($prefix, isPrefix: true) as $key => $until) {
+ if ((int)$until > $now) {
+ $count++;
+ } else {
+ // left by a job that was killed
+ $this->appConfig->deleteAppValue($key);
+ }
+ }
+ }
+ return $count;
+ }
+
+ /**
+ * @param list|null $providerKeys
+ * @throws \OCP\DB\Exception
+ */
+ private function countHandedOutDocuments(?array $providerKeys): int {
+ $count = 0;
+ if ($providerKeys === null || in_array(ProviderConfigService::getDefaultProviderKey(), $providerKeys, true)) {
+ $count += $this->queueMapper->countLocked();
+ }
+ foreach ($this->contentQueue->countLocked() as $providerKey => $locked) {
+ if ($providerKeys === null || in_array($providerKey, $providerKeys, true)) {
+ $count += (int)$locked;
+ }
+ }
+ return $count;
+ }
+
+ /**
+ * Waits for the background jobs that queue items of these providers to end. A job that is still running
+ * would queue items, or mark its provider imported, after the wipe cleared them.
+ *
+ * @param list|null $providerKeys null for the jobs of all the providers
+ * @param ?callable(int): void $onWaiting called with the number of running jobs, once before waiting and again
+ * whenever that number changes
+ * @throws \RuntimeException when they do not end in time
+ */
+ public function waitForRunningJobs(?array $providerKeys = null, ?callable $onWaiting = null): void {
+ $prefixes = $providerKeys === null
+ ? [self::RUNNING_KEY_PREFIX]
+ : array_map(fn (string $key): string => self::RUNNING_KEY_PREFIX . $this->getProviderConfigPartKey($key) . '_', $providerKeys);
+ $this->waitUntilNone(
+ fn (): int => $this->countRunningJobs($prefixes),
+ $onWaiting,
+ 'background jobs are still queueing documents',
+ 'If they were killed, their records expire within an hour, or use --ignore-handed-out.',
+ );
+ }
+
+ /**
+ * Waits for the documents the backend took to be reported back. A document that is still being indexed
+ * would be written to the vector DB after it was wiped.
+ *
+ * @param list|null $providerKeys only wait for the documents of these providers, null for all of them
+ * @param ?callable(int): void $onWaiting called with the number of documents still being indexed,
+ * once before waiting and again whenever that number changes
+ * @throws \RuntimeException when they are not reported back in time
+ * @throws \OCP\DB\Exception
+ */
+ public function waitForHandedOutDocuments(?array $providerKeys = null, ?callable $onWaiting = null): void {
+ $this->waitUntilNone(
+ fn (): int => $this->countHandedOutDocuments($providerKeys),
+ $onWaiting,
+ 'documents are still being indexed by context_chat_backend',
+ 'If the backend was stopped while indexing them, they would only be released after a day, use --ignore-handed-out then.',
+ );
+ }
+
+ /**
+ * @param callable(): int $count what is left to wait for
+ * @param ?callable(int): void $onWaiting called with what is left, once before waiting and again whenever it changes
+ * @param string $progressInfoText what is being awaited, for the error message
+ * @param string $hint what to do if it never ends, for the error message
+ * @throws \RuntimeException when it is not done in time, receives the above error message
+ */
+ private function waitUntilNone(callable $count, ?callable $onWaiting, string $progressInfoText, string $hint): void {
+ $timeout = $this->getDrainTimeout();
+ $deadline = $this->timeFactory->getTime() + $timeout;
+ $reported = null;
+ while (($left = $count()) > 0) {
+ if ($onWaiting !== null && $left !== $reported) {
+ $onWaiting($left);
+ $reported = $left;
+ }
+ if ($this->timeFactory->getTime() >= $deadline) {
+ throw new \RuntimeException(
+ $left . ' ' . $progressInfoText . ' after waiting ' . $timeout . ' seconds. Try again later. ' . $hint
+ );
+ }
+ sleep(1);
+ }
+ }
+
+ private function getDrainTimeout(): int {
+ return $this->appConfig->getAppValueInt('wipe_drain_timeout', self::DEFAULT_DRAIN_TIMEOUT, lazy: true);
+ }
+
+ private function readUntil(string $key): int {
+ return (int)($this->readValues($key)[$key] ?? 0);
+ }
+
+ /**
+ * Read from the database instead of IAppConfig. IAppConfig values are cached by each web server
+ * for a few seconds, and all of them must stop handing out items before the wipe starts.
+ * This ensures that we get the most up-to-date value for the pause state.
+ *
+ * @return array configkey: value
+ */
+ private function readValues(string $key, bool $isPrefix = false): array {
+ $qb = $this->db->getQueryBuilder();
+ $qb->select('configkey', 'configvalue')
+ ->from('appconfig')
+ ->where($qb->expr()->eq('appid', $qb->createNamedParameter(Application::APP_ID)));
+ if ($isPrefix) {
+ $qb->andWhere($qb->expr()->like('configkey', $qb->createNamedParameter($this->db->escapeLikeParameter($key) . '%')));
+ } else {
+ $qb->andWhere($qb->expr()->eq('configkey', $qb->createNamedParameter($key)));
+ }
+ $result = $qb->executeQuery();
+ $values = [];
+ while (($row = $result->fetch()) !== false) {
+ $values[(string)$row['configkey']] = (string)$row['configvalue'];
+ }
+ $result->closeCursor();
+ return $values;
+ }
+}
diff --git a/lib/Service/LangRopeService.php b/lib/Service/LangRopeService.php
index 96150bcc..2485325d 100644
--- a/lib/Service/LangRopeService.php
+++ b/lib/Service/LangRopeService.php
@@ -180,6 +180,14 @@ private function requestToExApp(
}
}
+ if (!is_array($finalBody)) {
+ $this->logger->error('Malformed response from Context Chat Backend (ExApp), expected a JSON object or array', [
+ 'response' => $response->getBody(),
+ 'route' => $route,
+ 'method' => $method,
+ ]);
+ throw new RuntimeException('Malformed response from Context Chat Backend (ExApp) for ' . $route . ': expected a JSON object or array');
+ }
return $finalBody;
}
@@ -189,14 +197,13 @@ private function requestToExApp(
*/
public function getIndexedDocumentsCounts(): array {
$response = $this->requestToExApp('/countIndexedDocuments', 'POST', extraOptions: ['timeout' => 10]);
- if ($response === []) {
- // No documents indexed yet
- return [];
- }
- if (!isset($response[ProviderConfigService::getDefaultProviderKey()])) {
- throw new \RuntimeException("Malformed indexed documents count response from Context Chat Backend (ExApp): '"
- . ProviderConfigService::getDefaultProviderKey() . "' key is missing, response: " . strval(json_encode($response)));
+ foreach ($response as $count) {
+ if (!is_int($count)) {
+ throw new \RuntimeException('Malformed indexed documents count response from Context Chat Backend (ExApp): '
+ . strval(json_encode($response)));
+ }
}
+ // only the providers with indexed documents are listed
return $response;
}
@@ -224,6 +231,16 @@ public function deleteProvider(string $providerKey): void {
$this->requestToExApp('/deleteProvider', 'POST', $params);
}
+ /**
+ * Delete all the indexed documents from the backend's vector DB
+ *
+ * @return void
+ * @throws \RuntimeException
+ */
+ public function deleteAll(): void {
+ $this->requestToExApp('/deleteAll', 'POST');
+ }
+
/**
* @param string $userId
* @return void
diff --git a/lib/Service/ProviderConfigService.php b/lib/Service/ProviderConfigService.php
index 344c98e0..f513596d 100644
--- a/lib/Service/ProviderConfigService.php
+++ b/lib/Service/ProviderConfigService.php
@@ -112,6 +112,53 @@ public function updateProvider(
$this->appConfig->setAppValueString('providers', json_encode($providers), lazy: true);
}
+ /**
+ * Mark all the content providers, except the files provider, as not initiated
+ * so their initial import runs again
+ */
+ public function resetProvidersInitiation(): void {
+ $providers = $this->getProviders();
+ foreach ($providers as $key => $provider) {
+ if ($key !== self::getDefaultProviderKey()) {
+ $providers[$key]['isInitiated'] = false;
+ }
+ }
+ $this->appConfig->setAppValueString('providers', json_encode($providers), lazy: true);
+ }
+
+ /**
+ * Get the parts of a provider key
+ *
+ * @param string $providerKey The provider key in the format "appId__providerId"
+ * @return array{0: string, 1: string} The appId and providerId
+ * @throws \RuntimeException when the provider key format is invalid
+ */
+ public function getProviderParts(string $providerKey): array {
+ $identifierValues = explode('__', $providerKey, 2);
+ if (count($identifierValues) !== 2) {
+ throw new \RuntimeException('Invalid provider key format: ' . $providerKey);
+ }
+ return $identifierValues;
+ }
+
+ /**
+ * Mark a specific provider as initiated or not initiated
+ *
+ * @param string $providerKey The provider key in the format "appId__providerId"
+ * @param bool $isInitiated
+ * @return void
+ * @throws \RuntimeException when the provider key format is invalid
+ */
+ public function setProviderInitiated(string $providerKey, bool $isInitiated): void {
+ $provider = $this->getProvider($providerKey);
+ if ($provider === null) {
+ return;
+ }
+ [$appId, $providerId] = $this->getProviderParts($providerKey);
+ /** @psalm-suppress ArgumentTypeCoercion */
+ $this->updateProvider($appId, $providerId, $provider['classString'], $isInitiated);
+ }
+
/**
* @param string $appId
* @param ?string $providerId
diff --git a/lib/Service/ReindexService.php b/lib/Service/ReindexService.php
new file mode 100644
index 00000000..7c2cbf0f
--- /dev/null
+++ b/lib/Service/ReindexService.php
@@ -0,0 +1,300 @@
+
+ */
+ public function getRegisteredProviderKeys(): array {
+ return $this->uniqueSorted(array_merge(
+ [ProviderConfigService::getDefaultProviderKey()],
+ array_keys($this->providerConfig->getProviders()),
+ ));
+ }
+
+ /**
+ * Keys of the providers that can be wiped: the registered ones and the ones that still have
+ * documents in the vector DB, e.g. of an app that has been removed since.
+ *
+ * @return list
+ * @throws \RuntimeException when the backend could not be reached
+ */
+ public function getWipeableProviderKeys(): array {
+ return $this->uniqueSorted(array_merge(
+ $this->getRegisteredProviderKeys(),
+ array_keys($this->langRopeService->getIndexedDocumentsCounts()),
+ ));
+ }
+
+ /**
+ * @param array $keys
+ * @return list
+ */
+ private function uniqueSorted(array $keys): array {
+ $keys = array_values(array_unique(array_map(strval(...), $keys)));
+ sort($keys);
+ return $keys;
+ }
+
+ /**
+ * Schedule the full filesystem crawl and the initial import of the content providers.
+ * Jobs that are already scheduled are not scheduled again.
+ *
+ * @param list|null $providerKeys null means everything. Without a scope only the content
+ * providers that have not been imported yet are imported (which is all of them after a full wipe),
+ * content providers explicitly specified are set as not initiated and scheduled for import.
+ * @return bool whether anything new was scheduled
+ */
+ public function scheduleReindex(?array $providerKeys = null): bool {
+ $scheduled = false;
+ $filesKey = ProviderConfigService::getDefaultProviderKey();
+
+ foreach ($this->providerConfig->getProviders() as $key => $provider) {
+ if ($key === $filesKey || $provider['classString'] === '') {
+ continue;
+ }
+ if ($providerKeys === null) {
+ // a full wipe turns all providers as not initiated so they will be scheduled for import and not skipped
+ if ($provider['isInitiated']) {
+ continue;
+ }
+ } elseif (in_array($key, $providerKeys, true)) {
+ $this->providerConfig->setProviderInitiated($key, false);
+ } else {
+ continue;
+ }
+ if ($this->jobList->has(InitialContentImportJob::class, $provider['classString'])) {
+ continue;
+ }
+ $this->jobList->add(InitialContentImportJob::class, $provider['classString']);
+ $scheduled = true;
+ }
+
+ if (($providerKeys === null || in_array($filesKey, $providerKeys, true))
+ && !$this->jobList->has(SchedulerJob::class, null)
+ ) {
+ $this->jobList->add(SchedulerJob::class);
+ $scheduled = true;
+ }
+ return $scheduled;
+ }
+
+ /**
+ * Delete everything that was queued or indexed, here and in context_chat_backend's vector DB.
+ * The documents the backend is still indexing are waited for here, so that its workers, possibly running
+ * on Kubernetes, cannot write them to the vector DB after it was wiped.
+ *
+ * A full wipe holds back all the queues until it is over, a wipe of the files only holds the files queue.
+ * A content provider is not held back since its unlocked/untouched queued items are deleted before the wait,
+ * so once the wait is over, none of what was queued before the wipe can still be indexed. Items the provider
+ * submits during the wipe may be indexed before or after the delete in the backend.
+ * It is safe to run this again when the backend request fails.
+ *
+ * @param list|null $providerKeys null means everything including the actions queue, otherwise only the given providers
+ * @param bool $ignoreHandedOut do not wait for the documents the backend is indexing, for when some documents are stuck
+ * in the queue for 24h due to error or if the backend stopped while indexing them/having a lock
+ * @param ?callable(string): void $onStep called with a short description before each step, for progress output
+ * @throws \OCP\DB\Exception
+ * @throws \RuntimeException when the vector DB could not be wiped
+ */
+ public function wipe(?array $providerKeys = null, bool $ignoreHandedOut = false, ?callable $onStep = null): void {
+ $onStep = $onStep ?? static function (string $_): void {
+ };
+
+ $filesKey = ProviderConfigService::getDefaultProviderKey();
+ $providerKeys = $providerKeys === null ? null : array_values(array_unique($providerKeys));
+ $contentProviderKeys = $providerKeys === null ? [] : array_values(array_diff($providerKeys, [$filesKey]));
+
+ if ($providerKeys === null) {
+ $onStep('Pausing the hand-out of queued documents and actions to context_chat_backend');
+ } elseif (in_array($filesKey, $providerKeys, true)) {
+ $onStep('Pausing the hand-out of queued files to context_chat_backend');
+ }
+ // also stops the crawl and import jobs of the providers
+ $pause = $this->fetchGuard->pause($providerKeys);
+ try {
+ if ($ignoreHandedOut) {
+ $onStep('Not waiting for the background jobs and the documents context_chat_backend may still be indexing (--ignore-handed-out)');
+ } else {
+ // a running crawl or import would queue items after the queues are cleared
+ $onStep('Checking for background jobs still queueing documents');
+ $this->fetchGuard->waitForRunningJobs($providerKeys, static function (int $running) use ($onStep): void {
+ $onStep('Waiting for ' . $running . ' background jobs to finish queueing documents');
+ });
+ }
+
+ foreach ($contentProviderKeys as $providerKey) {
+ $onStep('Removing the import job and clearing the indexing queue of ' . $providerKey);
+ $this->clearContentProviderState($providerKey);
+ }
+
+ if (!$ignoreHandedOut) {
+ $onStep('Checking for documents context_chat_backend is still indexing');
+ $this->fetchGuard->waitForHandedOutDocuments($providerKeys, static function (int $locked) use ($onStep): void {
+ $onStep('Waiting for context_chat_backend to finish indexing ' . $locked . ' documents');
+ });
+ }
+
+ if ($providerKeys === null) {
+ $this->wipeAll($onStep);
+ return;
+ }
+
+ foreach ($providerKeys as $providerKey) {
+ $this->logger->info('Wiping the Context Chat data of provider ' . $providerKey);
+ if ($providerKey === $filesKey) {
+ $onStep('Removing the crawl jobs, clearing the files indexing queue and the file system events, resetting the indexing statistics');
+ $this->clearFilesState();
+ } else {
+ // `clearContentProviderState()` already ran so the content queue may be empty
+ // but clean it up again since with --ignore-handed-out locked rows are not waited for to be cleared
+ [$appId, $providerId] = $this->providerConfig->getProviderParts($providerKey);
+ $this->contentQueue->removeProvider($appId, $providerId);
+ }
+ $onStep('Deleting the documents of ' . $providerKey . ' from the vector DB');
+ $this->langRopeService->deleteProvider($providerKey);
+ }
+ } finally {
+ if ($providerKeys === null) {
+ $onStep('Resuming the hand-out of queued documents and actions');
+ } elseif (in_array($filesKey, $providerKeys, true)) {
+ $onStep('Resuming the hand-out of queued files');
+ }
+ $this->fetchGuard->resume($pause);
+ }
+ }
+
+ /**
+ * @param callable(string): void $onStep
+ */
+ private function wipeAll(callable $onStep): void {
+ $this->logger->info('Wiping all the Context Chat data');
+
+ $onStep('Removing the scheduled crawl and content import jobs');
+ // stop the crawl and the content provider imports from queueing more items
+ foreach ([SchedulerJob::class, StorageCrawlJob::class, InitialContentImportJob::class] as $jobClass) {
+ $this->jobList->remove($jobClass);
+ }
+ $onStep('Clearing the indexing, actions and file system events queues');
+ $this->clearTables(self::QUEUE_TABLES);
+ $onStep('Resetting the indexing statistics and the content providers\' import state');
+ $this->resetStats();
+ $this->providerConfig->resetProvidersInitiation();
+
+ $onStep('Deleting all the documents from the vector DB');
+ $this->langRopeService->deleteAll();
+ $this->logger->info('Wiped all the Context Chat data');
+ }
+
+ /**
+ * The actions queue is left alone, it is shared by all the providers and the users.
+ * Leftover actions for the wiped documents are skipped by the backend.
+ */
+ private function clearFilesState(): void {
+ foreach ([SchedulerJob::class, StorageCrawlJob::class] as $jobClass) {
+ $this->jobList->remove($jobClass);
+ }
+ $this->clearTables(self::FILES_QUEUE_TABLES);
+ // the stats only track the files
+ $this->resetStats();
+ }
+
+ /**
+ * Leaves the items the backend is indexing in the queue, so the wipe can wait for them, and deletes the rest.
+ * Deleted items can no longer be locked, so nothing more of the provider's queue can be handed out.
+ */
+ private function clearContentProviderState(string $providerKey): void {
+ $provider = $this->providerConfig->getProvider($providerKey);
+ if ($provider !== null && $provider['classString'] !== '') {
+ $this->jobList->remove(InitialContentImportJob::class, $provider['classString']);
+ }
+
+ [$appId, $providerId] = $this->providerConfig->getProviderParts($providerKey);
+ $this->contentQueue->removeProvider($appId, $providerId, onlyUnlocked: true);
+
+ if ($provider !== null) {
+ $this->providerConfig->setProviderInitiated($providerKey, false);
+ }
+ }
+
+ /**
+ * @param list $tables
+ * @throws \RuntimeException
+ */
+ private function clearTables(array $tables): void {
+ $this->db->beginTransaction();
+ try {
+ foreach ($tables as $table) {
+ $this->db->getQueryBuilder()->delete($table)->executeStatement();
+ }
+ $this->db->commit();
+ } catch (\Throwable $e) {
+ $this->db->rollBack();
+ throw new \RuntimeException('Failed to clear tables ' . implode(', ', $tables) . ': ' . $e->getMessage(), 0, $e);
+ }
+ }
+
+ private function resetStats(): void {
+ foreach (self::STATS_CONFIG_KEYS as $key) {
+ $this->appConfig->deleteAppValue($key);
+ }
+ // Indicate wipe as a different value, `last_indexed_time` set to 0 and queues empty triggers the initial
+ // indexing in QueueController to be set as completed.
+ // A re-index sets this back to 0, in SchedulerJob specifically.
+ $this->appConfig->setAppValueInt('last_indexed_time', -1, lazy: true);
+ // the time taken for the next complete index is measured from here
+ $this->appConfig->setAppValueInt('installed_time', $this->timeFactory->getTime(), lazy: true);
+ }
+}
diff --git a/lib/Service/StatisticsService.php b/lib/Service/StatisticsService.php
index 7d214ef0..cce67161 100644
--- a/lib/Service/StatisticsService.php
+++ b/lib/Service/StatisticsService.php
@@ -36,11 +36,12 @@ public function getStatistics(): array {
$stats = [];
$stats['installed_at'] = $this->appConfig->getAppValueInt('installed_time', 0, lazy: true);
- if ($this->appConfig->getAppValueInt('last_indexed_time', 0, lazy: true) === 0) {
- $stats['initial_indexing_complete'] = false;
- } else {
+ // 0 while the initial indexing runs, -1 after a wipe of the files until re-index starts again
+ if (($lastIndexedTime = $this->appConfig->getAppValueInt('last_indexed_time', 0, lazy: true)) > 0) {
$stats['initial_indexing_complete'] = true;
- $stats['intial_indexing_completed_at'] = $this->appConfig->getAppValueInt('last_indexed_time', 0, lazy: true);
+ $stats['intial_indexing_completed_at'] = $lastIndexedTime;
+ } else {
+ $stats['initial_indexing_complete'] = false;
}
$stats['eligible_files_count'] = $this->getEligibleFilesCount();
@@ -69,9 +70,7 @@ public function getStatistics(): array {
$this->logger->error($e->getMessage(), ['exception' => $e]);
$stats['backend_available'] = false;
}
- if (empty($stats['vectordb_document_counts'] ?? [])) {
- $stats['vectordb_document_counts'] = [ ProviderConfigService::getDefaultProviderKey() => 0 ];
- }
+ $stats['vectordb_document_counts'] = ($stats['vectordb_document_counts'] ?? []) + [ ProviderConfigService::getDefaultProviderKey() => 0 ];
try {
$queued_files_count = $this->queueService->count();
diff --git a/tests/integration/LangRopeServiceResponseTest.php b/tests/integration/LangRopeServiceResponseTest.php
new file mode 100644
index 00000000..6ed7c5e7
--- /dev/null
+++ b/tests/integration/LangRopeServiceResponseTest.php
@@ -0,0 +1,99 @@
+appConfig = (new Application())->getContainer()->get(IAppConfig::class);
+ $this->savedBackendInit = $this->appConfig->getAppValueString('backend_init', 'false', lazy: true);
+ $this->appConfig->setAppValueString('backend_init', 'true', lazy: true);
+ \OC::$server->registerService(PublicFunctions::class, static fn () => new PublicFunctions());
+ }
+
+ protected function tearDown(): void {
+ $this->appConfig->setAppValueString('backend_init', $this->savedBackendInit, lazy: true);
+ parent::tearDown();
+ }
+
+ private function service(): LangRopeService {
+ $service = Server::get(LangRopeService::class);
+ $appManager = $this->createMock(IAppManager::class);
+ $appManager->method('isEnabledForUser')->willReturn(true);
+ (new \ReflectionProperty($service, 'appManager'))->setValue($service, $appManager);
+ return $service;
+ }
+
+ private function backendAnswers(string $body): void {
+ $response = $this->createMock(IResponse::class);
+ $response->method('getHeader')->willReturnCallback(static fn (string $name) => $name === 'Content-Type' ? 'application/json' : '');
+ $response->method('getBody')->willReturn($body);
+ $response->method('getStatusCode')->willReturn(200);
+ PublicFunctions::$response = $response;
+ }
+
+ /** @return array */
+ public static function malformedBodies(): array {
+ return [
+ 'null' => ['null'],
+ 'a number' => ['3'],
+ 'a string' => ['"ok"'],
+ 'not JSON' => ['proxy error'],
+ ];
+ }
+
+ #[DataProvider('malformedBodies')]
+ public function testMalformedResponseIsReportedAsABackendError(string $body): void {
+ $this->backendAnswers($body);
+
+ $this->expectException(\RuntimeException::class);
+ $this->expectExceptionMessage('Malformed response from Context Chat Backend (ExApp) for /countIndexedDocuments');
+ $this->service()->getIndexedDocumentsCounts();
+ }
+
+ public function testCountsAreReturned(): void {
+ $this->backendAnswers('{"files__default": 3}');
+
+ $this->assertSame(['files__default' => 3], $this->service()->getIndexedDocumentsCounts());
+ }
+ }
+}
diff --git a/tests/integration/QueueControllerFetchGuardTest.php b/tests/integration/QueueControllerFetchGuardTest.php
new file mode 100644
index 00000000..983dd02b
--- /dev/null
+++ b/tests/integration/QueueControllerFetchGuardTest.php
@@ -0,0 +1,105 @@
+|null */
+ private ?array $pausedUntil = null;
+
+ protected function setUp(): void {
+ parent::setUp();
+ $this->controller = Server::get(QueueController::class);
+ $this->queueMapper = Server::get(QueueMapper::class);
+ $this->actionMapper = Server::get(QueueActionMapper::class);
+ $this->guard = (new Application())->getContainer()->get(FetchGuard::class);
+ $this->queueMapper->clearQueue();
+ $this->actionMapper->removeFromQueue($this->allActionIds());
+
+ $file = new QueueFile();
+ $file->setFileId(42);
+ $file->setStorageId(1);
+ $file->setRootId(1);
+ $file->setUpdate(false);
+ $this->queueMapper->insertIntoQueue($file);
+
+ $action = new QueueAction();
+ $action->setType(ActionType::DELETE_SOURCE_IDS);
+ $action->setPayload('{"sourceIds":["files__default: 42"]}');
+ $this->actionMapper->insertIntoQueue($action);
+ }
+
+ protected function tearDown(): void {
+ if ($this->pausedUntil !== null) {
+ $this->guard->resume($this->pausedUntil);
+ }
+ $this->queueMapper->clearQueue();
+ $this->actionMapper->removeFromQueue($this->allActionIds());
+ parent::tearDown();
+ }
+
+ /** @return list */
+ private function allActionIds(): array {
+ return array_map(static fn (QueueAction $a) => $a->getId(), $this->actionMapper->getFromQueue(1000));
+ }
+
+ public function testNothingIsHandedOutOrLockedWhilePaused(): void {
+ $this->pausedUntil = $this->guard->pause();
+
+ $documents = $this->controller->getDocumentsQueueItems(
+ Server::get(IRootFolder::class),
+ $this->queueMapper,
+ Server::get(QueueContentItemMapper::class),
+ Server::get(\OCP\Files\Config\IUserMountCache::class),
+ );
+ $actions = $this->controller->getActionsQueueItems($this->actionMapper);
+
+ $this->assertSame(Http::STATUS_OK, $documents->getStatus());
+ $this->assertEquals(['files' => new \stdClass(), 'content_providers' => new \stdClass()], $documents->getData());
+ $this->assertEquals(['actions' => new \stdClass()], $actions->getData());
+ $this->assertSame(0, $this->queueMapper->countLocked(), 'a paused fetch must not lock queue items');
+ $this->assertSame(1, $this->queueMapper->count());
+ $this->assertSame(0, $this->actionMapper->countLocked());
+ $this->assertSame(1, $this->actionMapper->count());
+ }
+
+ public function testActionsAreHandedOutAgainAfterResume(): void {
+ $this->pausedUntil = $this->guard->pause();
+ $this->guard->resume($this->pausedUntil);
+ $this->pausedUntil = null;
+
+ $actions = $this->controller->getActionsQueueItems($this->actionMapper);
+
+ $this->assertCount(1, (array)$actions->getData()['actions']);
+ }
+}
diff --git a/tests/integration/QueueControllerPauseTest.php b/tests/integration/QueueControllerPauseTest.php
new file mode 100644
index 00000000..5c7d7b9d
--- /dev/null
+++ b/tests/integration/QueueControllerPauseTest.php
@@ -0,0 +1,219 @@
+ */
+ private array $pauses = [];
+
+ protected function setUp(): void {
+ parent::setUp();
+ $this->db = Server::get(IDBConnection::class);
+ $this->fetchGuard = Server::get(FetchGuard::class);
+ $this->clearQueues();
+
+ $item = new QueueContentItem();
+ $item->setItemId('item1');
+ $item->setAppId('someapp');
+ $item->setProviderId('prov');
+ $item->setTitle('title');
+ $item->setContent('content');
+ $item->setDocumentType('text');
+ $item->setLastModified(new \DateTime());
+ $item->setUsers('admin');
+ Server::get(QueueContentItemMapper::class)->insert($item);
+
+ // a file that cannot be resolved: when it is handed out it is dropped from the queue
+ $file = new QueueFile();
+ $file->setFileId(PHP_INT_MAX >> 16);
+ $file->setStorageId(1);
+ $file->setRootId(1);
+ $file->setUpdate(false);
+ Server::get(QueueMapper::class)->insertIntoQueue($file);
+
+ $action = new QueueAction();
+ $action->setType(ActionType::DELETE_SOURCE_IDS);
+ $action->setPayload('{"sourceIds":["someapp__prov: item1"]}');
+ Server::get(QueueActionMapper::class)->insertIntoQueue($action);
+ }
+
+ protected function tearDown(): void {
+ foreach ($this->pauses as $pause) {
+ $this->fetchGuard->resume($pause);
+ }
+ $this->clearQueues();
+ parent::tearDown();
+ }
+
+ private function clearQueues(): void {
+ foreach (['context_chat_queue', 'context_chat_content_queue', 'context_chat_action_queue'] as $table) {
+ $this->db->getQueryBuilder()->delete($table)->executeStatement();
+ }
+ }
+
+ /**
+ * Arguments are resolved by type from the DI container, like the AppFramework dispatcher does.
+ *
+ * @param array $services passed instead of the container's
+ * @return array
+ */
+ private function call(string $endpoint, array $services = []): array {
+ $controller = Server::get(QueueController::class);
+ $method = new \ReflectionMethod($controller, $endpoint);
+ $args = [];
+ foreach ($method->getParameters() as $parameter) {
+ $type = $parameter->getType();
+ $args[] = ($type instanceof \ReflectionNamedType && !$type->isBuiltin())
+ ? ($services[$type->getName()] ?? Server::get($type->getName()))
+ : 64;
+ }
+ $response = $method->invokeArgs($controller, $args);
+ $this->assertSame(Http::STATUS_OK, $response->getStatus());
+ return $response->getData();
+ }
+
+ private function filesInQueue(): int {
+ return Server::get(QueueMapper::class)->count();
+ }
+
+ public function testFilesWipeHoldsBackOnlyTheFiles(): void {
+ $this->pauses[] = $this->fetchGuard->pause(['files__default']);
+
+ $response = $this->call('getDocumentsQueueItems');
+
+ $this->assertCount(1, (array)$response['content_providers']);
+ $this->assertSame(1, $this->filesInQueue(), 'the file must not be touched');
+ $this->assertCount(1, (array)$this->call('getActionsQueueItems')['actions']);
+ }
+
+ public function testContentProviderWipeHoldsBackNothing(): void {
+ // it only stops the provider's import, its wipe deletes the queued items it does not wait for
+ $this->pauses[] = $this->fetchGuard->pause(['someapp__prov']);
+
+ $response = $this->call('getDocumentsQueueItems');
+
+ $this->assertCount(1, (array)$response['content_providers']);
+ $this->assertSame(0, $this->filesInQueue());
+ $this->assertCount(1, (array)$this->call('getActionsQueueItems')['actions']);
+ }
+
+ public function testFullWipeHoldsBackEverything(): void {
+ $this->pauses[] = $this->fetchGuard->pause();
+
+ $response = $this->call('getDocumentsQueueItems');
+
+ $this->assertSame([], (array)$response['content_providers']);
+ $this->assertSame(1, $this->filesInQueue());
+ $this->assertSame([], (array)$this->call('getActionsQueueItems')['actions']);
+ }
+
+ public function testNothingIsHeldBackWithoutAWipe(): void {
+ $response = $this->call('getDocumentsQueueItems');
+
+ $this->assertCount(1, (array)$response['content_providers']);
+ $this->assertSame(0, $this->filesInQueue(), 'the file was handed out and dropped');
+ }
+
+ /**
+ * A full wipe that starts while a fetch is locking its batch counts the handed out documents before the fetch
+ * returns. The fetch must notice the pause after locking and put the items back, or they would be indexed
+ * after the wipe without the wipe having waited for them.
+ */
+ public function testItemsLockedAsAFullWipeStartsArePutBack(): void {
+ $real = Server::get(QueueContentItemMapper::class);
+ $contentQueue = $this->createMock(QueueContentItemMapper::class);
+ $contentQueue->method('getFromQueue')->willReturnCallback(fn (int $limit) => $real->getFromQueue($limit));
+ $contentQueue->method('lock')->willReturnCallback(function (int $id) use ($real): bool {
+ $locked = $real->lock($id);
+ $this->pauses[] = $this->fetchGuard->pause();
+ return $locked;
+ });
+ $contentQueue->method('unlock')->willReturnCallback(fn (array $ids) => $real->unlock($ids));
+
+ $response = $this->call('getDocumentsQueueItems', [QueueContentItemMapper::class => $contentQueue]);
+
+ $this->assertSame([], (array)$response['content_providers']);
+ $this->assertSame([], $real->countLocked(), 'the item must be back in the queue, the wipe would wait for it');
+ }
+
+ public function testActionsLockedAsAFullWipeStartsArePutBack(): void {
+ $real = Server::get(QueueActionMapper::class);
+ $actionQueue = $this->createMock(QueueActionMapper::class);
+ $actionQueue->method('getFromQueue')->willReturnCallback(fn (int $limit) => $real->getFromQueue($limit));
+ $actionQueue->method('lock')->willReturnCallback(function (int $id) use ($real): bool {
+ $locked = $real->lock($id);
+ $this->pauses[] = $this->fetchGuard->pause();
+ return $locked;
+ });
+ $actionQueue->method('unlock')->willReturnCallback(fn (array $ids) => $real->unlock($ids));
+
+ $response = $this->call('getActionsQueueItems', [QueueActionMapper::class => $actionQueue]);
+
+ $this->assertSame([], (array)$response['actions']);
+ $this->assertSame(0, $real->countLocked(), 'the action must be back in the queue');
+ }
+
+ public function testFilesLockedAsAFilesWipeStartsArePutBack(): void {
+ $owner = 'cc_pause_' . substr(md5(uniqid('', true)), 0, 8);
+ $user = Server::get(IUserManager::class)->createUser($owner, 'Pass-' . $owner . '-1234567890');
+ try {
+ $userFolder = Server::get(IRootFolder::class)->getUserFolder($owner);
+ $file = $userFolder->newFile('doc.txt', 'content');
+ $this->clearQueues();
+ $queued = new QueueFile();
+ $queued->setFileId($file->getId());
+ $queued->setStorageId($userFolder->getStorage()->getCache()->getNumericStorageId());
+ $queued->setRootId($userFolder->getId());
+ $queued->setUpdate(false);
+ $real = Server::get(QueueMapper::class);
+ $real->insertIntoQueue($queued);
+
+ $filesQueue = $this->createMock(QueueMapper::class);
+ $filesQueue->method('getFromQueue')->willReturnCallback(fn (int $limit) => $real->getFromQueue($limit));
+ $filesQueue->method('lock')->willReturnCallback(function (int $id) use ($real): bool {
+ $locked = $real->lock($id);
+ $this->pauses[] = $this->fetchGuard->pause(['files__default']);
+ return $locked;
+ });
+ $filesQueue->method('unlock')->willReturnCallback(fn (array $ids) => $real->unlock($ids));
+
+ $response = $this->call('getDocumentsQueueItems', [QueueMapper::class => $filesQueue]);
+
+ $this->assertSame([], (array)$response['files']);
+ $this->assertSame(1, $real->count(), 'the file must be back in the queue');
+ $this->assertSame(0, $real->countLocked());
+ } finally {
+ $user?->delete();
+ }
+ }
+}
diff --git a/tests/integration/ReindexServiceTest.php b/tests/integration/ReindexServiceTest.php
new file mode 100644
index 00000000..c4b452ba
--- /dev/null
+++ b/tests/integration/ReindexServiceTest.php
@@ -0,0 +1,644 @@
+> */
+ private array $savedConfig;
+
+ protected function setUp(): void {
+ parent::setUp();
+ $this->db = Server::get(IDBConnection::class);
+ $this->jobList = Server::get(IJobList::class);
+ // the app scoped config is only available from the app's container
+ $container = (new Application())->getContainer();
+ $this->appConfig = $container->get(IAppConfig::class);
+ $this->providerConfig = $container->get(ProviderConfigService::class);
+ $this->langRope = $this->createMock(LangRopeService::class);
+ $this->fetchGuard = $this->newGuard(Server::get(ITimeFactory::class));
+ $time = $this->createMock(ITimeFactory::class);
+ $time->method('getTime')->willReturn(self::NOW);
+
+ $this->service = new ReindexService(
+ $this->db,
+ $this->jobList,
+ $this->appConfig,
+ $this->providerConfig,
+ $this->langRope,
+ Server::get(QueueContentItemMapper::class),
+ $this->fetchGuard,
+ $time,
+ Server::get(Logger::class),
+ );
+
+ $this->savedConfig = $this->readConfig();
+ $this->clearState();
+ $this->appConfig->setAppValueString('providers', '', lazy: true);
+ }
+
+ protected function tearDown(): void {
+ $this->clearState();
+ // the wipe resets the stats and the providers, put back what the instance had before
+ $qb = $this->db->getQueryBuilder();
+ $qb->delete('appconfig')
+ ->where($qb->expr()->eq('appid', $qb->createNamedParameter(Application::APP_ID)))
+ ->executeStatement();
+ foreach ($this->savedConfig as $row) {
+ $qb = $this->db->getQueryBuilder();
+ $qb->insert('appconfig');
+ foreach ($row as $column => $value) {
+ $qb->setValue($column, $qb->createNamedParameter($value));
+ }
+ $qb->executeStatement();
+ }
+ Server::get(\OCP\IAppConfig::class)->clearCache(true);
+ parent::tearDown();
+ }
+
+ /** @return list> */
+ private function readConfig(): array {
+ $qb = $this->db->getQueryBuilder();
+ $result = $qb->select('*')->from('appconfig')
+ ->where($qb->expr()->eq('appid', $qb->createNamedParameter(Application::APP_ID)))
+ ->executeQuery();
+ $rows = $result->fetchAll();
+ $result->closeCursor();
+ return $rows;
+ }
+
+ private function clearState(): void {
+ foreach (ReindexService::QUEUE_TABLES as $table) {
+ $this->db->getQueryBuilder()->delete($table)->executeStatement();
+ }
+ foreach ([SchedulerJob::class, StorageCrawlJob::class, InitialContentImportJob::class] as $jobClass) {
+ $this->jobList->remove($jobClass);
+ }
+ }
+
+ private function seedIndexedState(): void {
+ $file = new QueueFile();
+ $file->setFileId(42);
+ $file->setStorageId(1);
+ $file->setRootId(1);
+ $file->setUpdate(false);
+ Server::get(QueueMapper::class)->insertIntoQueue($file);
+
+ foreach (['someapp', 'otherapp'] as $appId) {
+ $this->queueContentItem($appId, 'item1');
+ }
+
+ $action = new QueueAction();
+ $action->setType(ActionType::DELETE_SOURCE_IDS);
+ $action->setPayload('{"sourceIds":["files__default: 42"]}');
+ Server::get(QueueActionMapper::class)->insertIntoQueue($action);
+
+ Server::get(FsEventMapper::class)->insertRow(FsEventType::CREATE->value, 'admin', 42);
+
+ $this->jobList->add(StorageCrawlJob::class, ['storage_id' => 1, 'root_id' => 1, 'overridden_root' => 1, 'last_file_id' => 0]);
+ $this->jobList->add(InitialContentImportJob::class, self::PROVIDER_CLASS);
+ $this->jobList->add(InitialContentImportJob::class, self::OTHER_PROVIDER_CLASS);
+
+ $this->appConfig->setAppValueInt('last_indexed_time', 123, lazy: true);
+ $this->appConfig->setAppValueInt('last_enqueued_db_id', 7, lazy: true);
+ $this->appConfig->setAppValueInt('eligible_files_count', 99, lazy: true);
+ $this->appConfig->setAppValueInt('eligible_files_count_time', 456, lazy: true);
+
+ $this->providerConfig->updateProvider('files', 'default', '', true);
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, true);
+ $this->providerConfig->updateProvider('otherapp', 'prov', self::OTHER_PROVIDER_CLASS, true);
+ }
+
+ private function queueContentItem(string $appId, string $itemId): void {
+ $item = new QueueContentItem();
+ $item->setItemId($itemId);
+ $item->setAppId($appId);
+ $item->setProviderId('prov');
+ $item->setTitle('title');
+ $item->setContent('content');
+ $item->setDocumentType('text');
+ $item->setLastModified(new \DateTime());
+ $item->setUsers('admin');
+ Server::get(QueueContentItemMapper::class)->insert($item);
+ }
+
+ /** @return list appId/itemId of the content items in the queue */
+ private function contentQueueItems(): array {
+ $qb = $this->db->getQueryBuilder();
+ $result = $qb->select('app_id', 'item_id')->from('context_chat_content_queue')->executeQuery();
+ $items = [];
+ while (($row = $result->fetch()) !== false) {
+ $items[] = $row['app_id'] . '/' . $row['item_id'];
+ }
+ $result->closeCursor();
+ sort($items);
+ return $items;
+ }
+
+ /** @return list */
+ private function contentQueueAppIds(): array {
+ $qb = $this->db->getQueryBuilder();
+ $result = $qb->selectDistinct('app_id')->from('context_chat_content_queue')->executeQuery();
+ $appIds = [];
+ while (($appId = $result->fetchOne()) !== false) {
+ $appIds[] = $appId;
+ }
+ $result->closeCursor();
+ return $appIds;
+ }
+
+ private function rowCount(string $table): int {
+ $qb = $this->db->getQueryBuilder();
+ return (int)$qb->select($qb->func()->count('*'))->from($table)->executeQuery()->fetchOne();
+ }
+
+ public function testWipeClearsAllLocalStateAndTheBackend(): void {
+ $this->seedIndexedState();
+ $this->langRope->expects($this->once())->method('deleteAll');
+ $this->langRope->expects($this->never())->method('deleteProvider');
+
+ $this->service->wipe();
+
+ foreach (ReindexService::QUEUE_TABLES as $table) {
+ $this->assertSame(0, $this->rowCount($table), $table . ' should be empty');
+ }
+ $this->assertFalse($this->jobList->has(StorageCrawlJob::class, ['storage_id' => 1, 'root_id' => 1, 'overridden_root' => 1, 'last_file_id' => 0]));
+ $this->assertFalse($this->jobList->has(InitialContentImportJob::class, self::PROVIDER_CLASS));
+ $this->assertFalse($this->jobList->has(InitialContentImportJob::class, self::OTHER_PROVIDER_CLASS));
+ $this->assertSame(-1, $this->appConfig->getAppValueInt('last_indexed_time', 0, lazy: true), 'not complete, and not 0 until the crawl is scheduled');
+ $this->assertSame(-1, $this->appConfig->getAppValueInt('last_enqueued_db_id', -1, lazy: true));
+ $this->assertSame(0, $this->appConfig->getAppValueInt('eligible_files_count_time', 0, lazy: true));
+ $this->assertSame(self::NOW, $this->appConfig->getAppValueInt('installed_time', 0, lazy: true));
+
+ $providers = $this->providerConfig->getProviders();
+ $this->assertFalse($providers['someapp__prov']['isInitiated'], 'content providers must be imported again');
+ $this->assertTrue($providers[ProviderConfigService::getDefaultProviderKey()]['isInitiated'], 'files are re-crawled by SchedulerJob instead');
+ }
+
+ public function testWipeClearsLocalStateBeforeTheBackendFails(): void {
+ $this->seedIndexedState();
+ $this->langRope->method('deleteAll')->willThrowException(new \RuntimeException('backend down'));
+
+ try {
+ $this->service->wipe();
+ $this->fail('wipe() must report the backend failure');
+ } catch (\RuntimeException $e) {
+ $this->assertSame('backend down', $e->getMessage());
+ }
+
+ // nothing may be indexed or pruned against a half-wiped vector DB, and a re-run must work
+ foreach (ReindexService::QUEUE_TABLES as $table) {
+ $this->assertSame(0, $this->rowCount($table), $table . ' should be empty');
+ }
+ }
+
+ public function testScheduleReindexAfterWipeQueuesCrawlAndProviderImportsOnce(): void {
+ $this->seedIndexedState();
+ $this->service->wipe();
+
+ $this->assertTrue($this->service->scheduleReindex());
+ $this->assertFalse($this->service->scheduleReindex(), 'an already scheduled crawl is not scheduled again');
+
+ $this->assertTrue($this->jobList->has(SchedulerJob::class, null));
+ $this->assertTrue($this->jobList->has(InitialContentImportJob::class, self::PROVIDER_CLASS));
+ $importJobs = iterator_to_array($this->jobList->getJobsIterator(InitialContentImportJob::class, null, 0), false);
+ $this->assertCount(2, $importJobs);
+ }
+
+ public function testScheduleReindexSkipsInitiatedProviders(): void {
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, true);
+
+ $this->service->scheduleReindex();
+
+ $this->assertTrue($this->jobList->has(SchedulerJob::class, null));
+ $this->assertFalse($this->jobList->has(InitialContentImportJob::class, self::PROVIDER_CLASS));
+ }
+
+ public function testWipeFilesOnlyLeavesContentProvidersAndActions(): void {
+ $this->seedIndexedState();
+ $this->langRope->expects($this->never())->method('deleteAll');
+ $this->langRope->expects($this->once())->method('deleteProvider')->with('files__default');
+
+ $this->service->wipe(['files__default']);
+
+ $this->assertSame(0, $this->rowCount('context_chat_queue'));
+ $this->assertSame(0, $this->rowCount('context_chat_fs_events'));
+ $this->assertSame(2, $this->rowCount('context_chat_content_queue'), 'content providers must be kept');
+ $this->assertSame(1, $this->rowCount('context_chat_action_queue'), 'actions of other providers and users must be kept');
+ $this->assertFalse($this->jobList->has(StorageCrawlJob::class, ['storage_id' => 1, 'root_id' => 1, 'overridden_root' => 1, 'last_file_id' => 0]));
+ $this->assertTrue($this->jobList->has(InitialContentImportJob::class, self::PROVIDER_CLASS));
+ $this->assertSame(-1, $this->appConfig->getAppValueInt('last_indexed_time', 0, lazy: true));
+ $this->assertTrue($this->providerConfig->getProviders()['someapp__prov']['isInitiated']);
+ }
+
+ public function testWipeOneContentProviderLeavesTheRest(): void {
+ $this->seedIndexedState();
+ $this->langRope->expects($this->never())->method('deleteAll');
+ $this->langRope->expects($this->once())->method('deleteProvider')->with('someapp__prov');
+
+ $this->service->wipe(['someapp__prov']);
+
+ $this->assertSame(['otherapp'], $this->contentQueueAppIds());
+ $this->assertSame(1, $this->rowCount('context_chat_queue'), 'files must be kept');
+ $this->assertSame(1, $this->rowCount('context_chat_fs_events'), 'files must be kept');
+ $this->assertSame(1, $this->rowCount('context_chat_action_queue'));
+ $this->assertFalse($this->jobList->has(InitialContentImportJob::class, self::PROVIDER_CLASS));
+ $this->assertTrue($this->jobList->has(InitialContentImportJob::class, self::OTHER_PROVIDER_CLASS));
+ $this->assertSame(123, $this->appConfig->getAppValueInt('last_indexed_time', 0, lazy: true), 'file stats must be kept');
+
+ $providers = $this->providerConfig->getProviders();
+ $this->assertFalse($providers['someapp__prov']['isInitiated']);
+ $this->assertTrue($providers['otherapp__prov']['isInitiated']);
+ }
+
+ public function testWipeOfAnUnregisteredProviderOnlyDeletesInTheBackend(): void {
+ $this->seedIndexedState();
+ $this->langRope->expects($this->once())->method('deleteProvider')->with('removedapp__prov');
+
+ $this->service->wipe(['removedapp__prov']);
+
+ $this->assertSame(2, $this->rowCount('context_chat_content_queue'));
+ $this->assertArrayNotHasKey('removedapp__prov', $this->providerConfig->getProviders());
+ }
+
+ public function testScheduleReindexOfANamedProviderReimportsItEvenIfImported(): void {
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, true);
+ $this->providerConfig->updateProvider('otherapp', 'prov', self::OTHER_PROVIDER_CLASS, true);
+
+ $this->assertTrue($this->service->scheduleReindex(['someapp__prov']));
+
+ $this->assertTrue($this->jobList->has(InitialContentImportJob::class, self::PROVIDER_CLASS));
+ $this->assertFalse($this->jobList->has(InitialContentImportJob::class, self::OTHER_PROVIDER_CLASS));
+ $this->assertFalse($this->jobList->has(SchedulerJob::class, null), 'files are not in the scope');
+ // InitialContentImportJob skips initiated providers
+ $this->assertFalse($this->providerConfig->getProviders()['someapp__prov']['isInitiated']);
+ }
+
+ public function testScheduleReindexOfFilesOnly(): void {
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, false);
+
+ $this->assertTrue($this->service->scheduleReindex(['files__default']));
+ $this->assertFalse($this->service->scheduleReindex(['files__default']));
+
+ $this->assertTrue($this->jobList->has(SchedulerJob::class, null));
+ $this->assertFalse($this->jobList->has(InitialContentImportJob::class, self::PROVIDER_CLASS));
+ }
+
+ public function testWipeableProviderKeysIncludeProvidersOnlyKnownToTheBackend(): void {
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, true);
+ $this->langRope->method('getIndexedDocumentsCounts')->willReturn(['files__default' => 3, 'removedapp__prov' => 1]);
+
+ $this->assertSame(['files__default', 'someapp__prov'], $this->service->getRegisteredProviderKeys());
+ $this->assertSame(['files__default', 'removedapp__prov', 'someapp__prov'], $this->service->getWipeableProviderKeys());
+ }
+
+ private function newGuard(ITimeFactory $time): FetchGuard {
+ return new FetchGuard($this->db, $this->appConfig, $time, Server::get(QueueMapper::class), Server::get(QueueContentItemMapper::class));
+ }
+
+ /** @return ITimeFactory&MockObject a clock that advances one second per sleep(1) of the guard */
+ private function clock(int &$now): ITimeFactory {
+ $time = $this->createMock(ITimeFactory::class);
+ $time->method('getTime')->willReturnCallback(function () use (&$now) {
+ return $now;
+ });
+ return $time;
+ }
+
+ public function testFetchingIsPausedOnlyWhileTheBackendIsWiped(): void {
+ $this->seedIndexedState();
+ $seen = null;
+ $this->langRope->method('deleteAll')->willReturnCallback(function () use (&$seen) {
+ $seen = $this->fetchGuard->isPaused();
+ });
+
+ $this->service->wipe();
+
+ $this->assertTrue($seen, 'nothing may be handed out while the vector DB is wiped');
+ $this->assertFalse($this->fetchGuard->isPaused());
+ }
+
+ public function testFetchingResumesWhenTheBackendFails(): void {
+ $this->seedIndexedState();
+ $this->langRope->method('deleteAll')->willThrowException(new \RuntimeException('backend down'));
+
+ try {
+ $this->service->wipe();
+ } catch (\RuntimeException) {
+ }
+
+ $this->assertFalse($this->fetchGuard->isPaused());
+ }
+
+ public function testWipeIsRefusedWhileDocumentsAreStillBeingIndexed(): void {
+ $this->seedIndexedState();
+ $queueMapper = Server::get(QueueMapper::class);
+ $queueMapper->lock($queueMapper->getMaxId());
+ $this->appConfig->setAppValueInt('wipe_drain_timeout', 0, lazy: true);
+ $this->langRope->expects($this->never())->method('deleteAll');
+
+ try {
+ $this->service->wipe();
+ $this->fail('the wipe must not run while a document is being indexed');
+ } catch (\RuntimeException $e) {
+ $this->assertStringContainsString('1 documents are still being indexed', $e->getMessage());
+ } finally {
+ $this->appConfig->deleteAppValue('wipe_drain_timeout');
+ }
+
+ $this->assertSame(1, $queueMapper->countLocked(), 'nothing may be cleared when the wipe is refused');
+ $this->assertFalse($this->fetchGuard->isPaused());
+ }
+
+ public function testResumeDoesNotLiftANewerPause(): void {
+ $now = self::NOW;
+ $guard = $this->newGuard($this->clock($now));
+
+ $first = $guard->pause();
+ $now += 10;
+ $second = $guard->pause();
+ $guard->resume($first);
+
+ $this->assertTrue($guard->isPaused(), 'the second wipe is still running');
+ $guard->resume($second);
+ $this->assertFalse($guard->isPaused());
+ }
+
+ public function testResumeDoesNotLiftANewerPauseOfTheSameProvider(): void {
+ $now = self::NOW;
+ $guard = $this->newGuard($this->clock($now));
+
+ $first = $guard->pause(['someapp__prov', 'files__default']);
+ $now += 10;
+ $second = $guard->pause(['someapp__prov']);
+ $guard->resume($first);
+
+ $this->assertTrue($guard->isPaused('someapp__prov'), 'the second wipe is still running');
+ $this->assertFalse($guard->isPaused('files__default'));
+ $guard->resume($second);
+ $this->assertFalse($guard->isPaused('someapp__prov'));
+ }
+
+ public function testPauseExpiresOnItsOwn(): void {
+ $now = self::NOW;
+ $guard = $this->newGuard($this->clock($now));
+
+ $pause = $guard->pause();
+ $this->assertTrue($guard->isPaused());
+ $now += 24 * 60 * 60;
+
+ $this->assertFalse($guard->isPaused(), 'a wipe that died must not stop the indexing forever');
+ $guard->resume($pause);
+ }
+
+ public function testPauseOutlastsAScopedWipeOfManySlowProviders(): void {
+ $now = self::NOW;
+ $guard = $this->newGuard($this->clock($now));
+
+ $pause = $guard->pause(['files__default']);
+ // the drain timeout, then 20 deleteProvider requests that each run into the request timeout
+ $now += FetchGuard::DEFAULT_DRAIN_TIMEOUT + 20 * Application::CC_DEFAULT_REQUEST_TIMEOUT;
+
+ $this->assertTrue($guard->isPaused('files__default'), 'indexing must not restart while providers are still being wiped');
+ $guard->resume($pause);
+ }
+
+ public function testFullAndFilesPausesAreLiftedSeparately(): void {
+ $files = $this->fetchGuard->pause(['files__default']);
+ $all = $this->fetchGuard->pause();
+ $this->fetchGuard->resume($all);
+
+ $this->assertTrue($this->fetchGuard->isPaused('files__default'), 'the files wipe is still running');
+ $this->fetchGuard->resume($files);
+ $this->assertFalse($this->fetchGuard->isPaused('files__default'));
+ }
+
+ public function testFilesWipePausesOnlyTheFiles(): void {
+ $this->seedIndexedState();
+ $seen = null;
+ $this->langRope->method('deleteProvider')->willReturnCallback(function () use (&$seen) {
+ $seen = [$this->fetchGuard->isPaused('files__default'), $this->fetchGuard->isPaused()];
+ });
+
+ $this->service->wipe(['files__default']);
+
+ $this->assertSame([true, false], $seen);
+ $this->assertFalse($this->fetchGuard->isPaused('files__default'));
+ }
+
+ public function testContentProviderWipeOnlyWaitsForItsOwnDocuments(): void {
+ $this->seedIndexedState();
+ // a file is being indexed, unrelated to the provider being wiped
+ $queueMapper = Server::get(QueueMapper::class);
+ $queueMapper->lock($queueMapper->getMaxId());
+ $this->appConfig->setAppValueInt('wipe_drain_timeout', 0, lazy: true);
+ $this->langRope->expects($this->once())->method('deleteProvider')->with('someapp__prov');
+
+ try {
+ $this->service->wipe(['someapp__prov']);
+ } finally {
+ $this->appConfig->deleteAppValue('wipe_drain_timeout');
+ }
+
+ $this->assertSame(1, $queueMapper->countLocked(), 'the file being indexed is left alone');
+ }
+
+ /**
+ * Nothing pauses a content provider, so its queue is cleared before the wait: what is not handed out yet
+ * can no longer be, and what is handed out is waited for.
+ */
+ public function testContentProviderWipeClearsTheQueueBeforeWaitingForTheHandedOutItems(): void {
+ $this->seedIndexedState();
+ $contentQueue = Server::get(QueueContentItemMapper::class);
+ $this->queueContentItem('someapp', 'item2');
+ foreach ($contentQueue->getFromQueue(10) as $item) {
+ if ($item->getAppId() === 'someapp' && $item->getItemId() === 'item1') {
+ $contentQueue->lock($item->getId());
+ }
+ }
+ $this->appConfig->setAppValueInt('wipe_drain_timeout', 0, lazy: true);
+ $this->langRope->expects($this->never())->method('deleteProvider');
+
+ try {
+ $this->service->wipe(['someapp__prov']);
+ $this->fail('the wipe must wait for the item being indexed');
+ } catch (\RuntimeException $e) {
+ $this->assertStringContainsString('1 documents are still being indexed', $e->getMessage());
+ } finally {
+ $this->appConfig->deleteAppValue('wipe_drain_timeout');
+ }
+
+ $this->assertSame(['otherapp/item1', 'someapp/item1'], $this->contentQueueItems(), 'item2 must be gone, it could have been handed out during the wait');
+ }
+
+ public function testContentProviderWipeWithIgnoreHandedOutClearsItsWholeQueue(): void {
+ $this->seedIndexedState();
+ $contentQueue = Server::get(QueueContentItemMapper::class);
+ foreach ($contentQueue->getFromQueue(10) as $item) {
+ $contentQueue->lock($item->getId());
+ }
+
+ $this->service->wipe(['someapp__prov'], ignoreHandedOut: true);
+
+ $this->assertSame(['otherapp'], $this->contentQueueAppIds());
+ }
+
+ public function testWaitEndsOnceTheHandedOutBatchIsReportedBack(): void {
+ $queueMapper = Server::get(QueueMapper::class);
+ $file = new QueueFile();
+ $file->setFileId(43);
+ $file->setStorageId(1);
+ $file->setRootId(1);
+ $file->setUpdate(false);
+ $id = $queueMapper->insertIntoQueue($file)->getId();
+ $queueMapper->lock($id);
+
+ $checks = 0;
+ $time = $this->createMock(ITimeFactory::class);
+ $time->method('getTime')->willReturnCallback(function () use (&$checks, $queueMapper, $id) {
+ // the backend reports the batch back while the wipe is waiting
+ if (++$checks === 3) {
+ $queueMapper->removeFromQueue([$id]);
+ }
+ return self::NOW;
+ });
+
+ $reported = [];
+ $this->newGuard($time)->waitForHandedOutDocuments(null, function (int $locked) use (&$reported) {
+ $reported[] = $locked;
+ });
+
+ $this->assertSame(0, $queueMapper->countLocked());
+ $this->assertSame([1], $reported, 'the count is reported once, not on every poll');
+ }
+
+ /**
+ * The steps are reported in order, each one before its work: what the admin reads is what runs next.
+ */
+ public function testWipeReportsEachStepBeforeDoingIt(): void {
+ $this->seedIndexedState();
+ $events = [];
+ $this->langRope->method('deleteAll')->willReturnCallback(function () use (&$events) {
+ $events[] = 'backend:deleteAll';
+ });
+
+ $this->service->wipe(null, false, function (string $message) use (&$events) {
+ // mark the moment the queues are emptied, to check the order against the work
+ $events[] = $message . ($this->rowCount('context_chat_queue') === 0 ? ' [queues empty]' : '');
+ });
+
+ $this->assertSame([
+ 'Pausing the hand-out of queued documents and actions to context_chat_backend',
+ 'Checking for background jobs still queueing documents',
+ 'Checking for documents context_chat_backend is still indexing',
+ 'Removing the scheduled crawl and content import jobs',
+ 'Clearing the indexing, actions and file system events queues',
+ 'Resetting the indexing statistics and the content providers\' import state [queues empty]',
+ 'Deleting all the documents from the vector DB [queues empty]',
+ 'backend:deleteAll',
+ 'Resuming the hand-out of queued documents and actions [queues empty]',
+ ], $events);
+ }
+
+ public function testWipeOfOneProviderReportsItsSteps(): void {
+ $this->seedIndexedState();
+ $events = [];
+ $this->langRope->method('deleteProvider')->willReturnCallback(function (string $key) use (&$events) {
+ $events[] = 'backend:deleteProvider ' . $key;
+ });
+
+ $this->service->wipe(['someapp__prov'], true, function (string $message) use (&$events) {
+ $events[] = $message;
+ });
+
+ $this->assertSame([
+ 'Not waiting for the background jobs and the documents context_chat_backend may still be indexing (--ignore-handed-out)',
+ 'Removing the import job and clearing the indexing queue of someapp__prov',
+ 'Deleting the documents of someapp__prov from the vector DB',
+ 'backend:deleteProvider someapp__prov',
+ ], $events, 'a content provider wipe does not pause');
+ }
+
+ public function testWipeOfTheFilesReportsItsSteps(): void {
+ $this->seedIndexedState();
+ $events = [];
+ $this->langRope->method('deleteProvider')->willReturnCallback(function (string $key) use (&$events) {
+ $events[] = 'backend:deleteProvider ' . $key;
+ });
+
+ $this->service->wipe(['files__default'], true, function (string $message) use (&$events) {
+ $events[] = $message;
+ });
+
+ $this->assertSame([
+ 'Pausing the hand-out of queued files to context_chat_backend',
+ 'Not waiting for the background jobs and the documents context_chat_backend may still be indexing (--ignore-handed-out)',
+ 'Removing the crawl jobs, clearing the files indexing queue and the file system events, resetting the indexing statistics',
+ 'Deleting the documents of files__default from the vector DB',
+ 'backend:deleteProvider files__default',
+ 'Resuming the hand-out of queued files',
+ ], $events);
+ }
+
+ public function testResumeIsReportedWhenTheBackendFails(): void {
+ $this->seedIndexedState();
+ $this->langRope->method('deleteAll')->willThrowException(new \RuntimeException('backend down'));
+ $events = [];
+
+ try {
+ $this->service->wipe(null, false, function (string $message) use (&$events) {
+ $events[] = $message;
+ });
+ $this->fail('the backend failure must be reported');
+ } catch (\RuntimeException) {
+ }
+
+ $this->assertSame('Deleting all the documents from the vector DB', $events[count($events) - 2]);
+ $this->assertSame('Resuming the hand-out of queued documents and actions', end($events));
+ }
+}
diff --git a/tests/integration/WipeBackgroundJobsTest.php b/tests/integration/WipeBackgroundJobsTest.php
new file mode 100644
index 00000000..d03d7aaf
--- /dev/null
+++ b/tests/integration/WipeBackgroundJobsTest.php
@@ -0,0 +1,428 @@
+ 1, 'root_id' => 1, 'overridden_root' => 1];
+ private const PROVIDER_CLASS = 'OCA\\SomeApp\\ContentProvider';
+ // the indexing stats a files wipe resets, put back after each test
+ private const STATS_KEYS = ['last_indexed_time', 'installed_time', 'eligible_files_count', 'eligible_files_count_time'];
+
+ private IJobList $jobList;
+ private IAppConfig $appConfig;
+ private FetchGuard $fetchGuard;
+ private QueueMapper $queueMapper;
+ private ProviderConfigService $providerConfig;
+ private string $savedProviders;
+ /** @var array */
+ private array $savedStats = [];
+
+ protected function setUp(): void {
+ parent::setUp();
+ $this->jobList = Server::get(IJobList::class);
+ $container = (new Application())->getContainer();
+ $this->appConfig = $container->get(IAppConfig::class);
+ $this->providerConfig = $container->get(ProviderConfigService::class);
+ $this->fetchGuard = Server::get(FetchGuard::class);
+ $this->queueMapper = Server::get(QueueMapper::class);
+ $this->savedProviders = $this->appConfig->getAppValueString('providers', '', lazy: true);
+ foreach (self::STATS_KEYS as $key) {
+ $this->savedStats[$key] = $this->appConfig->hasAppKey($key, lazy: true) ? $this->appConfig->getAppValueInt($key, 0, lazy: true) : null;
+ }
+ $this->clearState();
+ }
+
+ protected function tearDown(): void {
+ $this->clearState();
+ $this->appConfig->setAppValueString('providers', $this->savedProviders, lazy: true);
+ foreach ($this->savedStats as $key => $value) {
+ $value === null ? $this->appConfig->deleteAppValue($key) : $this->appConfig->setAppValueInt($key, $value, lazy: true);
+ }
+ parent::tearDown();
+ }
+
+ private function clearState(): void {
+ $this->queueMapper->clearQueue();
+ foreach ([SchedulerJob::class, StorageCrawlJob::class, InitialContentImportJob::class] as $jobClass) {
+ $this->jobList->remove($jobClass);
+ }
+ $this->appConfig->deleteAppValue('last_enqueued_db_id');
+ $qb = Server::get(IDBConnection::class)->getQueryBuilder();
+ $qb->delete('appconfig')
+ ->where($qb->expr()->eq('appid', $qb->createNamedParameter(Application::APP_ID)))
+ ->andWhere($qb->expr()->orX(
+ $qb->expr()->like('configkey', $qb->createNamedParameter('running\_%')),
+ $qb->expr()->like('configkey', $qb->createNamedParameter('wipe\_%')),
+ $qb->expr()->in('configkey', $qb->createNamedParameter(['queue_fetch_paused_until', 'files_fetch_paused_until'], IQueryBuilder::PARAM_STR_ARRAY)),
+ ))
+ ->executeStatement();
+ }
+
+ /** @param ?callable(): void $meanwhile called between the two files of the batch */
+ private function runCrawl(?callable $meanwhile = null): void {
+ $storage = $this->createMock(StorageService::class);
+ $storage->method('getFilesInMount')->willReturnCallback(static function () use ($meanwhile): \Generator {
+ yield 101;
+ if ($meanwhile !== null) {
+ $meanwhile();
+ }
+ yield 102;
+ });
+ $job = Server::get(StorageCrawlJob::class);
+ $job->setId('7');
+ $this->replaceDependency($job, 'storageService', $storage);
+ $this->invokeRun($job, self::MOUNT + ['last_file_id' => 0]);
+ }
+
+ /** @param ?callable(): void $meanwhile called once the job runs, before it adds the crawl jobs */
+ private function runScheduler(?callable $meanwhile = null): void {
+ $storage = $this->createMock(StorageService::class);
+ $storage->method('getMounts')->willReturnCallback(static function () use ($meanwhile): \Generator {
+ if ($meanwhile !== null) {
+ $meanwhile();
+ }
+ yield self::MOUNT;
+ });
+ $job = Server::get(SchedulerJob::class);
+ $this->replaceDependency($job, 'storageService', $storage);
+ $this->invokeRun($job, null);
+ }
+
+ /** @param ?callable(): void $meanwhile called during the provider's import */
+ private function runImport(?callable $meanwhile = null): void {
+ $provider = $this->createMock(IContentProvider::class);
+ $provider->method('getAppId')->willReturn('someapp');
+ $provider->method('getId')->willReturn('prov');
+ $provider->method('triggerInitialImport')->willReturnCallback(static function () use ($meanwhile): void {
+ if ($meanwhile !== null) {
+ $meanwhile();
+ }
+ });
+ \OC::$server->registerService(self::PROVIDER_CLASS, static fn () => $provider);
+ $appManager = $this->createMock(IAppManager::class);
+ $appManager->method('isEnabledForUser')->willReturn(true);
+
+ $job = Server::get(InitialContentImportJob::class);
+ $this->replaceDependency($job, 'appManager', $appManager);
+ $this->invokeRun($job, self::PROVIDER_CLASS);
+ }
+
+ private function replaceDependency(object $object, string $property, object $value): void {
+ (new \ReflectionProperty($object, $property))->setValue($object, $value);
+ }
+
+ private function invokeRun(object $job, mixed $argument): void {
+ (new \ReflectionMethod($job, 'run'))->invoke($job, $argument);
+ }
+
+ private function countCrawlJobs(): int {
+ return count(iterator_to_array($this->jobList->getJobsIterator(StorageCrawlJob::class, null, 0), false));
+ }
+
+ private function isImported(): bool {
+ return $this->providerConfig->getProviders()['someapp__prov']['isInitiated'];
+ }
+
+ /** @return ITimeFactory a clock that starts at $now and moves when $now is changed */
+ private function clock(int &$now): ITimeFactory {
+ $time = $this->createMock(ITimeFactory::class);
+ $time->method('getTime')->willReturnCallback(static function () use (&$now): int {
+ return $now;
+ });
+ return $time;
+ }
+
+ private function guardWith(ITimeFactory $time): FetchGuard {
+ return new FetchGuard(
+ Server::get(IDBConnection::class), $this->appConfig, $time,
+ $this->queueMapper, Server::get(QueueContentItemMapper::class),
+ );
+ }
+
+ /**
+ * @param list|null $providerKeys
+ * @return int the running jobs a wipe of these providers would wait for
+ */
+ private function countRunningJobs(?array $providerKeys, ?FetchGuard $guard = null): int {
+ $counted = null;
+ $this->appConfig->setAppValueInt('wipe_drain_timeout', 0, lazy: true);
+ try {
+ ($guard ?? $this->fetchGuard)->waitForRunningJobs($providerKeys, static function (int $running) use (&$counted): void {
+ $counted = $running;
+ });
+ } catch (\RuntimeException) {
+ } finally {
+ $this->appConfig->deleteAppValue('wipe_drain_timeout');
+ }
+ return $counted ?? 0;
+ }
+
+ public function testCrawlQueuesAndContinuesWithoutAWipe(): void {
+ $this->runCrawl();
+
+ $this->assertSame(2, $this->queueMapper->count());
+ $this->assertSame(1, $this->countCrawlJobs(), 'the next batch is scheduled');
+ $this->assertSame(0, $this->countRunningJobs(null), 'the crawl is not recorded as running once it ended');
+ }
+
+ public function testCrawlStopsDuringAFilesWipe(): void {
+ $pause = $this->fetchGuard->pause(['files__default']);
+ try {
+ $this->runCrawl();
+ } finally {
+ $this->fetchGuard->resume($pause);
+ }
+
+ $this->assertSame(0, $this->queueMapper->count());
+ $this->assertSame(0, $this->countCrawlJobs());
+ }
+
+ public function testCrawlGoesOnDuringAWipeOfAContentProvider(): void {
+ $wipe = $this->fetchGuard->pause(['someapp__prov']);
+ try {
+ $this->runCrawl();
+ } finally {
+ $this->fetchGuard->resume($wipe);
+ }
+
+ $this->assertSame(2, $this->queueMapper->count());
+ }
+
+ public function testCrawlIsRecordedAsRunningWhileItQueues(): void {
+ $running = null;
+ $this->runCrawl(function () use (&$running) {
+ $running = [
+ $this->countRunningJobs([ProviderConfigService::getDefaultProviderKey()]),
+ $this->countRunningJobs(['someapp__prov']),
+ ];
+ });
+
+ $this->assertSame([1, 0], $running, 'a wipe of the files waits for the crawl, a wipe of another provider does not');
+ $this->assertSame(0, $this->countRunningJobs(null));
+ }
+
+ public function testCrawlRecordIsRemovedWhenTheCrawlFails(): void {
+ try {
+ $this->runCrawl(static function (): void {
+ throw new \RuntimeException('storage gone');
+ });
+ $this->fail('the failure must not be swallowed');
+ } catch (\RuntimeException $e) {
+ $this->assertSame('storage gone', $e->getMessage());
+ }
+
+ $this->assertSame(0, $this->countRunningJobs(null));
+ }
+
+ public function testSchedulerDoesNothingDuringAFullWipe(): void {
+ $pause = $this->fetchGuard->pause();
+ try {
+ $this->runScheduler();
+ } finally {
+ $this->fetchGuard->resume($pause);
+ }
+
+ $this->assertSame(0, $this->countCrawlJobs());
+ }
+
+ public function testSchedulerAddsTheCrawlJobsWithoutAWipe(): void {
+ $this->runScheduler();
+
+ $this->assertSame(1, $this->countCrawlJobs());
+ $this->assertSame(0, $this->countRunningJobs(null));
+ }
+
+ public function testImportMarksTheProviderImportedWithoutAWipe(): void {
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, false);
+
+ $this->runImport();
+
+ $this->assertTrue($this->isImported());
+ $this->assertSame(0, $this->countRunningJobs(null));
+ }
+
+ public function testImportStopsDuringAWipeOfItsProvider(): void {
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, false);
+ $imported = false;
+ $wipe = $this->fetchGuard->pause(['someapp__prov']);
+ try {
+ $this->runImport(static function () use (&$imported): void {
+ $imported = true;
+ });
+ } finally {
+ $this->fetchGuard->resume($wipe);
+ }
+
+ $this->assertFalse($imported, 'nothing may be submitted while the provider is wiped');
+ $this->assertFalse($this->isImported());
+ }
+
+ public function testImportStopsDuringAFullWipe(): void {
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, false);
+ $pause = $this->fetchGuard->pause();
+ try {
+ $this->runImport();
+ } finally {
+ $this->fetchGuard->resume($pause);
+ }
+
+ $this->assertFalse($this->isImported());
+ }
+
+ public function testImportIsRecordedAsRunningForItsProviderOnly(): void {
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, false);
+ $running = null;
+ $this->runImport(function () use (&$running) {
+ $running = [
+ $this->countRunningJobs(['someapp__prov']),
+ $this->countRunningJobs([ProviderConfigService::getDefaultProviderKey()]),
+ $this->countRunningJobs(null),
+ ];
+ });
+
+ $this->assertSame([1, 0, 1], $running);
+ }
+
+ public function testWipeWaitsForARunningImportBeforeClearingTheQueue(): void {
+ $this->providerConfig->updateProvider('someapp', 'prov', self::PROVIDER_CLASS, true);
+ $record = $this->fetchGuard->jobStarted('someapp__prov', '42', FetchGuard::IMPORT_RUNNING_TTL);
+ $this->appConfig->setAppValueInt('wipe_drain_timeout', 0, lazy: true);
+ $langRope = $this->createMock(LangRopeService::class);
+ $langRope->expects($this->never())->method('deleteProvider');
+ $service = Server::get(ReindexService::class);
+ $this->replaceDependency($service, 'langRopeService', $langRope);
+
+ try {
+ $service->wipe(['someapp__prov']);
+ $this->fail('the wipe must wait for the running import');
+ } catch (\RuntimeException $e) {
+ $this->assertStringContainsString('1 background jobs are still queueing documents', $e->getMessage());
+ } finally {
+ $this->appConfig->deleteAppValue('wipe_drain_timeout');
+ $this->fetchGuard->jobEnded($record);
+ }
+
+ $this->assertTrue($this->isImported(), 'nothing may be reset when the wipe gives up');
+ $this->assertFalse($this->fetchGuard->isPaused('someapp__prov'), 'the import may run again');
+ }
+
+ public function testWipeWithIgnoreHandedOutDoesNotWaitForRunningJobs(): void {
+ $record = $this->fetchGuard->jobStarted(ProviderConfigService::getDefaultProviderKey(), '42');
+ $langRope = $this->createMock(LangRopeService::class);
+ $langRope->expects($this->once())->method('deleteProvider');
+ $service = Server::get(ReindexService::class);
+ $this->replaceDependency($service, 'langRopeService', $langRope);
+
+ try {
+ $service->wipe([ProviderConfigService::getDefaultProviderKey()], ignoreHandedOut: true);
+ } finally {
+ $this->fetchGuard->jobEnded($record);
+ }
+ }
+
+ public function testAKilledJobBlocksWipesUntilItsRecordExpires(): void {
+ $now = 1_800_000_000;
+ $guard = $this->guardWith($this->clock($now));
+ $guard->jobStarted(ProviderConfigService::getDefaultProviderKey(), '42');
+
+ $now += FetchGuard::RUNNING_TTL - 1;
+ $this->assertSame(1, $this->countRunningJobs(null, $guard));
+
+ $now += 1;
+ $this->assertSame(0, $this->countRunningJobs(null, $guard), 'a job that never ended must not block wipes forever');
+ }
+
+ public function testARunningCrawlKeepsItsRecordAlive(): void {
+ $now = 1_800_000_000;
+ $guard = $this->guardWith($this->clock($now));
+ $record = $guard->jobStarted(ProviderConfigService::getDefaultProviderKey(), '42');
+
+ for ($i = 0; $i < 4; $i++) {
+ $now += FetchGuard::RUNNING_TTL / 2;
+ $guard->jobAlive($record);
+ }
+
+ $this->assertSame(1, $this->countRunningJobs(null, $guard), 'a crawl still working must not be taken as killed');
+ $guard->jobEnded($record);
+ }
+
+ /**
+ * The backend reports a batch back, which is when the initial indexing is checked for completion.
+ */
+ private function backendReportsABatch(): void {
+ $controller = Server::get(QueueController::class);
+ $method = new \ReflectionMethod($controller, 'deleteDocumentsQueueItems');
+ $args = [];
+ foreach ($method->getParameters() as $parameter) {
+ $type = $parameter->getType();
+ $args[] = ($type instanceof \ReflectionNamedType && !$type->isBuiltin()) ? Server::get($type->getName()) : [];
+ }
+ $method->invokeArgs($controller, $args);
+ }
+
+ private function isInitialIndexingComplete(): bool {
+ $stats = Server::get(StatisticsService::class);
+ $this->replaceDependency($stats, 'langRopeService', $this->createMock(LangRopeService::class));
+ $this->replaceDependency($stats, 'storageService', $this->createMock(StorageService::class));
+ return $stats->getStatistics()['initial_indexing_complete'];
+ }
+
+ public function testIndexingIsNotCompleteAfterAFilesWipeUntilTheFilesAreCrawledAgain(): void {
+ $langRope = $this->createMock(LangRopeService::class);
+ $service = Server::get(ReindexService::class);
+ $this->replaceDependency($service, 'langRopeService', $langRope);
+
+ $service->wipe([ProviderConfigService::getDefaultProviderKey()]);
+ $this->backendReportsABatch();
+ $this->assertFalse($this->isInitialIndexingComplete(), 'nothing was crawled yet');
+
+ // context_chat:reindex, then the scheduler adds the crawl jobs and the crawl queues the files
+ $service->scheduleReindex([ProviderConfigService::getDefaultProviderKey()]);
+ $this->runScheduler(function (): void {
+ $this->backendReportsABatch();
+ $this->assertFalse($this->isInitialIndexingComplete(), 'the crawl jobs are not added yet');
+ });
+ $this->backendReportsABatch();
+ $this->assertFalse($this->isInitialIndexingComplete(), 'the crawl jobs have not run yet');
+
+ $this->jobList->remove(StorageCrawlJob::class);
+ $this->backendReportsABatch();
+ $this->assertTrue($this->isInitialIndexingComplete(), 'the crawl is done and the queue is empty');
+ }
+}