aboutsummaryrefslogtreecommitdiffstats
path: root/lib/private/TaskProcessing
diff options
context:
space:
mode:
Diffstat (limited to 'lib/private/TaskProcessing')
-rw-r--r--lib/private/TaskProcessing/Db/Task.php10
-rw-r--r--lib/private/TaskProcessing/Db/TaskMapper.php25
-rw-r--r--lib/private/TaskProcessing/Manager.php94
-rw-r--r--lib/private/TaskProcessing/RemoveOldTasksBackgroundJob.php44
4 files changed, 137 insertions, 36 deletions
diff --git a/lib/private/TaskProcessing/Db/Task.php b/lib/private/TaskProcessing/Db/Task.php
index 4d919deaf94..05c0ae9ac74 100644
--- a/lib/private/TaskProcessing/Db/Task.php
+++ b/lib/private/TaskProcessing/Db/Task.php
@@ -45,6 +45,8 @@ use OCP\TaskProcessing\Task as OCPTask;
* @method int getStartedAt()
* @method setEndedAt(int $endedAt)
* @method int getEndedAt()
+ * @method setAllowCleanup(int $allowCleanup)
+ * @method int getAllowCleanup()
*/
class Task extends Entity {
protected $lastUpdated;
@@ -63,16 +65,17 @@ class Task extends Entity {
protected $scheduledAt;
protected $startedAt;
protected $endedAt;
+ protected $allowCleanup;
/**
* @var string[]
*/
- public static array $columns = ['id', 'last_updated', 'type', 'input', 'output', 'status', 'user_id', 'app_id', 'custom_id', 'completion_expected_at', 'error_message', 'progress', 'webhook_uri', 'webhook_method', 'scheduled_at', 'started_at', 'ended_at'];
+ public static array $columns = ['id', 'last_updated', 'type', 'input', 'output', 'status', 'user_id', 'app_id', 'custom_id', 'completion_expected_at', 'error_message', 'progress', 'webhook_uri', 'webhook_method', 'scheduled_at', 'started_at', 'ended_at', 'allow_cleanup'];
/**
* @var string[]
*/
- public static array $fields = ['id', 'lastUpdated', 'type', 'input', 'output', 'status', 'userId', 'appId', 'customId', 'completionExpectedAt', 'errorMessage', 'progress', 'webhookUri', 'webhookMethod', 'scheduledAt', 'startedAt', 'endedAt'];
+ public static array $fields = ['id', 'lastUpdated', 'type', 'input', 'output', 'status', 'userId', 'appId', 'customId', 'completionExpectedAt', 'errorMessage', 'progress', 'webhookUri', 'webhookMethod', 'scheduledAt', 'startedAt', 'endedAt', 'allowCleanup'];
public function __construct() {
@@ -94,6 +97,7 @@ class Task extends Entity {
$this->addType('scheduledAt', 'integer');
$this->addType('startedAt', 'integer');
$this->addType('endedAt', 'integer');
+ $this->addType('allowCleanup', 'integer');
}
public function toRow(): array {
@@ -122,6 +126,7 @@ class Task extends Entity {
'scheduledAt' => $task->getScheduledAt(),
'startedAt' => $task->getStartedAt(),
'endedAt' => $task->getEndedAt(),
+ 'allowCleanup' => $task->getAllowCleanup() ? 1 : 0,
]);
return $taskEntity;
}
@@ -144,6 +149,7 @@ class Task extends Entity {
$task->setScheduledAt($this->getScheduledAt());
$task->setStartedAt($this->getStartedAt());
$task->setEndedAt($this->getEndedAt());
+ $task->setAllowCleanup($this->getAllowCleanup() !== 0);
return $task;
}
}
diff --git a/lib/private/TaskProcessing/Db/TaskMapper.php b/lib/private/TaskProcessing/Db/TaskMapper.php
index 91fd68820ae..fee96534633 100644
--- a/lib/private/TaskProcessing/Db/TaskMapper.php
+++ b/lib/private/TaskProcessing/Db/TaskMapper.php
@@ -183,16 +183,39 @@ class TaskMapper extends QBMapper {
/**
* @param int $timeout
+ * @param bool $force If true, ignore the allow_cleanup flag
* @return int the number of deleted tasks
* @throws Exception
*/
- public function deleteOlderThan(int $timeout): int {
+ public function deleteOlderThan(int $timeout, bool $force = false): int {
$qb = $this->db->getQueryBuilder();
$qb->delete($this->tableName)
->where($qb->expr()->lt('last_updated', $qb->createPositionalParameter($this->timeFactory->getDateTime()->getTimestamp() - $timeout)));
+ if (!$force) {
+ $qb->andWhere($qb->expr()->eq('allow_cleanup', $qb->createPositionalParameter(1, IQueryBuilder::PARAM_INT)));
+ }
return $qb->executeStatement();
}
+ /**
+ * @param int $timeout
+ * @param bool $force If true, ignore the allow_cleanup flag
+ * @return \Generator<Task>
+ * @throws Exception
+ */
+ public function getTasksToCleanup(int $timeout, bool $force = false): \Generator {
+ $qb = $this->db->getQueryBuilder();
+ $qb->select(Task::$columns)
+ ->from($this->tableName)
+ ->where($qb->expr()->lt('last_updated', $qb->createPositionalParameter($this->timeFactory->getDateTime()->getTimestamp() - $timeout)));
+ if (!$force) {
+ $qb->andWhere($qb->expr()->eq('allow_cleanup', $qb->createPositionalParameter(1, IQueryBuilder::PARAM_INT)));
+ }
+ foreach ($this->yieldEntities($qb) as $entity) {
+ yield $entity;
+ };
+ }
+
public function update(Entity $entity): Entity {
$entity->setLastUpdated($this->timeFactory->now()->getTimestamp());
return parent::update($entity);
diff --git a/lib/private/TaskProcessing/Manager.php b/lib/private/TaskProcessing/Manager.php
index 11fb2bed559..e288f2981a8 100644
--- a/lib/private/TaskProcessing/Manager.php
+++ b/lib/private/TaskProcessing/Manager.php
@@ -30,6 +30,7 @@ use OCP\Files\IRootFolder;
use OCP\Files\Node;
use OCP\Files\NotPermittedException;
use OCP\Files\SimpleFS\ISimpleFile;
+use OCP\Files\SimpleFS\ISimpleFolder;
use OCP\Http\Client\IClientService;
use OCP\IAppConfig;
use OCP\ICache;
@@ -78,6 +79,8 @@ class Manager implements IManager {
'ai.taskprocessing_provider_preferences',
];
+ public const MAX_TASK_AGE_SECONDS = 60 * 60 * 24 * 30 * 4; // 4 months
+
/** @var list<IProvider>|null */
private ?array $providers = null;
@@ -1449,6 +1452,97 @@ class Manager implements IManager {
}
/**
+ * @param Task $task
+ * @return list<int>
+ * @throws NotFoundException
+ */
+ public function extractFileIdsFromTask(Task $task): array {
+ $ids = [];
+ $taskTypes = $this->getAvailableTaskTypes();
+ if (!isset($taskTypes[$task->getTaskTypeId()])) {
+ throw new NotFoundException('Could not find task type');
+ }
+ $taskType = $taskTypes[$task->getTaskTypeId()];
+ foreach ($taskType['inputShape'] + $taskType['optionalInputShape'] as $key => $descriptor) {
+ if (in_array(EShapeType::getScalarType($descriptor->getShapeType()), [EShapeType::File, EShapeType::Image, EShapeType::Audio, EShapeType::Video], true)) {
+ /** @var int|list<int> $inputSlot */
+ $inputSlot = $task->getInput()[$key];
+ if (is_array($inputSlot)) {
+ $ids = array_merge($inputSlot, $ids);
+ } else {
+ $ids[] = $inputSlot;
+ }
+ }
+ }
+ if ($task->getOutput() !== null) {
+ foreach ($taskType['outputShape'] + $taskType['optionalOutputShape'] as $key => $descriptor) {
+ if (in_array(EShapeType::getScalarType($descriptor->getShapeType()), [EShapeType::File, EShapeType::Image, EShapeType::Audio, EShapeType::Video], true)) {
+ /** @var int|list<int> $outputSlot */
+ $outputSlot = $task->getOutput()[$key];
+ if (is_array($outputSlot)) {
+ $ids = array_merge($outputSlot, $ids);
+ } else {
+ $ids[] = $outputSlot;
+ }
+ }
+ }
+ }
+ return $ids;
+ }
+
+ /**
+ * @param ISimpleFolder $folder
+ * @param int $ageInSeconds
+ * @return \Generator
+ */
+ public function clearFilesOlderThan(ISimpleFolder $folder, int $ageInSeconds = self::MAX_TASK_AGE_SECONDS): \Generator {
+ foreach ($folder->getDirectoryListing() as $file) {
+ if ($file->getMTime() < time() - $ageInSeconds) {
+ try {
+ $fileName = $file->getName();
+ $file->delete();
+ yield $fileName;
+ } catch (NotPermittedException $e) {
+ $this->logger->warning('Failed to delete a stale task processing file', ['exception' => $e]);
+ }
+ }
+ }
+ }
+
+ /**
+ * @param int $ageInSeconds
+ * @return \Generator
+ * @throws Exception
+ * @throws InvalidPathException
+ * @throws NotFoundException
+ * @throws \JsonException
+ * @throws \OCP\Files\NotFoundException
+ */
+ public function cleanupTaskProcessingTaskFiles(int $ageInSeconds = self::MAX_TASK_AGE_SECONDS): \Generator {
+ $taskIdsToCleanup = [];
+ foreach ($this->taskMapper->getTasksToCleanup($ageInSeconds) as $task) {
+ $taskIdsToCleanup[] = $task->getId();
+ $ocpTask = $task->toPublicTask();
+ $fileIds = $this->extractFileIdsFromTask($ocpTask);
+ foreach ($fileIds as $fileId) {
+ // only look for output files stored in appData/TaskProcessing/
+ $file = $this->rootFolder->getFirstNodeByIdInPath($fileId, '/' . $this->rootFolder->getAppDataDirectoryName() . '/core/TaskProcessing/');
+ if ($file instanceof File) {
+ try {
+ $fileId = $file->getId();
+ $fileName = $file->getName();
+ $file->delete();
+ yield ['task_id' => $task->getId(), 'file_id' => $fileId, 'file_name' => $fileName];
+ } catch (NotPermittedException $e) {
+ $this->logger->warning('Failed to delete a stale task processing file', ['exception' => $e]);
+ }
+ }
+ }
+ }
+ return $taskIdsToCleanup;
+ }
+
+ /**
* Make a request to the task's webhookUri if necessary
*
* @param Task $task
diff --git a/lib/private/TaskProcessing/RemoveOldTasksBackgroundJob.php b/lib/private/TaskProcessing/RemoveOldTasksBackgroundJob.php
index 42d073a024d..52fc204b752 100644
--- a/lib/private/TaskProcessing/RemoveOldTasksBackgroundJob.php
+++ b/lib/private/TaskProcessing/RemoveOldTasksBackgroundJob.php
@@ -10,17 +10,14 @@ use OC\TaskProcessing\Db\TaskMapper;
use OCP\AppFramework\Utility\ITimeFactory;
use OCP\BackgroundJob\TimedJob;
use OCP\Files\AppData\IAppDataFactory;
-use OCP\Files\NotFoundException;
-use OCP\Files\NotPermittedException;
-use OCP\Files\SimpleFS\ISimpleFolder;
use Psr\Log\LoggerInterface;
class RemoveOldTasksBackgroundJob extends TimedJob {
- public const MAX_TASK_AGE_SECONDS = 60 * 60 * 24 * 30 * 4; // 4 months
private \OCP\Files\IAppData $appData;
public function __construct(
ITimeFactory $timeFactory,
+ private Manager $taskProcessingManager,
private TaskMapper $taskMapper,
private LoggerInterface $logger,
IAppDataFactory $appDataFactory,
@@ -32,48 +29,29 @@ class RemoveOldTasksBackgroundJob extends TimedJob {
$this->appData = $appDataFactory->get('core');
}
-
/**
* @inheritDoc
*/
protected function run($argument): void {
try {
- $this->taskMapper->deleteOlderThan(self::MAX_TASK_AGE_SECONDS);
- } catch (\OCP\DB\Exception $e) {
- $this->logger->warning('Failed to delete stale task processing tasks', ['exception' => $e]);
+ iterator_to_array($this->taskProcessingManager->cleanupTaskProcessingTaskFiles());
+ } catch (\Exception $e) {
+ $this->logger->warning('Failed to delete stale task processing tasks files', ['exception' => $e]);
}
try {
- $this->clearFilesOlderThan($this->appData->getFolder('text2image'), self::MAX_TASK_AGE_SECONDS);
- } catch (NotFoundException $e) {
- // noop
+ $this->taskMapper->deleteOlderThan(Manager::MAX_TASK_AGE_SECONDS);
+ } catch (\OCP\DB\Exception $e) {
+ $this->logger->warning('Failed to delete stale task processing tasks', ['exception' => $e]);
}
try {
- $this->clearFilesOlderThan($this->appData->getFolder('audio2text'), self::MAX_TASK_AGE_SECONDS);
- } catch (NotFoundException $e) {
+ iterator_to_array($this->taskProcessingManager->clearFilesOlderThan($this->appData->getFolder('text2image')));
+ } catch (\OCP\Files\NotFoundException $e) {
// noop
}
try {
- $this->clearFilesOlderThan($this->appData->getFolder('TaskProcessing'), self::MAX_TASK_AGE_SECONDS);
- } catch (NotFoundException $e) {
+ iterator_to_array($this->taskProcessingManager->clearFilesOlderThan($this->appData->getFolder('audio2text')));
+ } catch (\OCP\Files\NotFoundException $e) {
// noop
}
}
-
- /**
- * @param ISimpleFolder $folder
- * @param int $ageInSeconds
- * @return void
- */
- private function clearFilesOlderThan(ISimpleFolder $folder, int $ageInSeconds): void {
- foreach ($folder->getDirectoryListing() as $file) {
- if ($file->getMTime() < time() - $ageInSeconds) {
- try {
- $file->delete();
- } catch (NotPermittedException $e) {
- $this->logger->warning('Failed to delete a stale task processing file', ['exception' => $e]);
- }
- }
- }
- }
-
}
generated by cgit v1.2.3 (git 2.39.1) at 2025-08-16 03:55:20 +0000