JobList.php 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442
  1. <?php
  2. /**
  3. * SPDX-FileCopyrightText: 2016-2024 Nextcloud GmbH and Nextcloud contributors
  4. * SPDX-FileCopyrightText: 2016 ownCloud, Inc.
  5. * SPDX-License-Identifier: AGPL-3.0-only
  6. */
  7. namespace OC\BackgroundJob;
  8. use Doctrine\DBAL\Platforms\MySQLPlatform;
  9. use OCP\AppFramework\QueryException;
  10. use OCP\AppFramework\Utility\ITimeFactory;
  11. use OCP\AutoloadNotAllowedException;
  12. use OCP\BackgroundJob\IJob;
  13. use OCP\BackgroundJob\IJobList;
  14. use OCP\BackgroundJob\IParallelAwareJob;
  15. use OCP\DB\Exception;
  16. use OCP\DB\QueryBuilder\IQueryBuilder;
  17. use OCP\IConfig;
  18. use OCP\IDBConnection;
  19. use Psr\Log\LoggerInterface;
  20. use function get_class;
  21. use function json_encode;
  22. use function md5;
  23. use function strlen;
  24. class JobList implements IJobList {
  25. protected IDBConnection $connection;
  26. protected IConfig $config;
  27. protected ITimeFactory $timeFactory;
  28. protected LoggerInterface $logger;
  29. public function __construct(IDBConnection $connection, IConfig $config, ITimeFactory $timeFactory, LoggerInterface $logger) {
  30. $this->connection = $connection;
  31. $this->config = $config;
  32. $this->timeFactory = $timeFactory;
  33. $this->logger = $logger;
  34. }
  35. public function add($job, $argument = null, ?int $firstCheck = null): void {
  36. if ($firstCheck === null) {
  37. $firstCheck = $this->timeFactory->getTime();
  38. }
  39. if ($job instanceof IJob) {
  40. $class = get_class($job);
  41. } else {
  42. $class = $job;
  43. }
  44. $argumentJson = json_encode($argument);
  45. if (strlen($argumentJson) > 4000) {
  46. throw new \InvalidArgumentException('Background job arguments can\'t exceed 4000 characters (json encoded)');
  47. }
  48. $query = $this->connection->getQueryBuilder();
  49. if (!$this->has($job, $argument)) {
  50. $query->insert('jobs')
  51. ->values([
  52. 'class' => $query->createNamedParameter($class),
  53. 'argument' => $query->createNamedParameter($argumentJson),
  54. 'argument_hash' => $query->createNamedParameter(md5($argumentJson)),
  55. 'last_run' => $query->createNamedParameter(0, IQueryBuilder::PARAM_INT),
  56. 'last_checked' => $query->createNamedParameter($firstCheck, IQueryBuilder::PARAM_INT),
  57. ]);
  58. } else {
  59. $query->update('jobs')
  60. ->set('reserved_at', $query->expr()->literal(0, IQueryBuilder::PARAM_INT))
  61. ->set('last_checked', $query->createNamedParameter($firstCheck, IQueryBuilder::PARAM_INT))
  62. ->set('last_run', $query->createNamedParameter(0, IQueryBuilder::PARAM_INT))
  63. ->where($query->expr()->eq('class', $query->createNamedParameter($class)))
  64. ->andWhere($query->expr()->eq('argument_hash', $query->createNamedParameter(md5($argumentJson))));
  65. }
  66. $query->executeStatement();
  67. }
  68. public function scheduleAfter(string $job, int $runAfter, $argument = null): void {
  69. $this->add($job, $argument, $runAfter);
  70. }
  71. /**
  72. * @param IJob|string $job
  73. * @param mixed $argument
  74. */
  75. public function remove($job, $argument = null): void {
  76. if ($job instanceof IJob) {
  77. $class = get_class($job);
  78. } else {
  79. $class = $job;
  80. }
  81. $query = $this->connection->getQueryBuilder();
  82. $query->delete('jobs')
  83. ->where($query->expr()->eq('class', $query->createNamedParameter($class)));
  84. if (!is_null($argument)) {
  85. $argumentJson = json_encode($argument);
  86. $query->andWhere($query->expr()->eq('argument_hash', $query->createNamedParameter(md5($argumentJson))));
  87. }
  88. // Add galera safe delete chunking if using mysql
  89. // Stops us hitting wsrep_max_ws_rows when large row counts are deleted
  90. if ($this->connection->getDatabasePlatform() instanceof MySQLPlatform) {
  91. // Then use chunked delete
  92. $max = IQueryBuilder::MAX_ROW_DELETION;
  93. $query->setMaxResults($max);
  94. do {
  95. $deleted = $query->execute();
  96. } while ($deleted === $max);
  97. } else {
  98. // Dont use chunked delete - let the DB handle the large row count natively
  99. $query->execute();
  100. }
  101. }
  102. public function removeById(int $id): void {
  103. $query = $this->connection->getQueryBuilder();
  104. $query->delete('jobs')
  105. ->where($query->expr()->eq('id', $query->createNamedParameter($id, IQueryBuilder::PARAM_INT)));
  106. $query->executeStatement();
  107. }
  108. /**
  109. * check if a job is in the list
  110. *
  111. * @param IJob|class-string<IJob> $job
  112. * @param mixed $argument
  113. */
  114. public function has($job, $argument): bool {
  115. if ($job instanceof IJob) {
  116. $class = get_class($job);
  117. } else {
  118. $class = $job;
  119. }
  120. $argument = json_encode($argument);
  121. $query = $this->connection->getQueryBuilder();
  122. $query->select('id')
  123. ->from('jobs')
  124. ->where($query->expr()->eq('class', $query->createNamedParameter($class)))
  125. ->andWhere($query->expr()->eq('argument_hash', $query->createNamedParameter(md5($argument))))
  126. ->setMaxResults(1);
  127. $result = $query->executeQuery();
  128. $row = $result->fetch();
  129. $result->closeCursor();
  130. return (bool) $row;
  131. }
  132. public function getJobs($job, ?int $limit, int $offset): array {
  133. $iterable = $this->getJobsIterator($job, $limit, $offset);
  134. if (is_array($iterable)) {
  135. return $iterable;
  136. } else {
  137. return iterator_to_array($iterable);
  138. }
  139. }
  140. /**
  141. * @param IJob|class-string<IJob>|null $job
  142. * @return iterable<IJob> Avoid to store these objects as they may share a Singleton instance. You should instead use these IJobs instances while looping on the iterable.
  143. */
  144. public function getJobsIterator($job, ?int $limit, int $offset): iterable {
  145. $query = $this->connection->getQueryBuilder();
  146. $query->select('*')
  147. ->from('jobs')
  148. ->setMaxResults($limit)
  149. ->setFirstResult($offset);
  150. if ($job !== null) {
  151. if ($job instanceof IJob) {
  152. $class = get_class($job);
  153. } else {
  154. $class = $job;
  155. }
  156. $query->where($query->expr()->eq('class', $query->createNamedParameter($class)));
  157. }
  158. $result = $query->executeQuery();
  159. while ($row = $result->fetch()) {
  160. $job = $this->buildJob($row);
  161. if ($job) {
  162. yield $job;
  163. }
  164. }
  165. $result->closeCursor();
  166. }
  167. /**
  168. * @inheritDoc
  169. */
  170. public function getNext(bool $onlyTimeSensitive = false, ?array $jobClasses = null): ?IJob {
  171. $query = $this->connection->getQueryBuilder();
  172. $query->select('*')
  173. ->from('jobs')
  174. ->where($query->expr()->lte('reserved_at', $query->createNamedParameter($this->timeFactory->getTime() - 12 * 3600, IQueryBuilder::PARAM_INT)))
  175. ->andWhere($query->expr()->lte('last_checked', $query->createNamedParameter($this->timeFactory->getTime(), IQueryBuilder::PARAM_INT)))
  176. ->orderBy('last_checked', 'ASC')
  177. ->setMaxResults(1);
  178. if ($onlyTimeSensitive) {
  179. $query->andWhere($query->expr()->eq('time_sensitive', $query->createNamedParameter(IJob::TIME_SENSITIVE, IQueryBuilder::PARAM_INT)));
  180. }
  181. if ($jobClasses !== null && count($jobClasses) > 0) {
  182. $orClasses = $query->expr()->orx();
  183. foreach ($jobClasses as $jobClass) {
  184. $orClasses->add($query->expr()->eq('class', $query->createNamedParameter($jobClass, IQueryBuilder::PARAM_STR)));
  185. }
  186. $query->andWhere($orClasses);
  187. }
  188. $result = $query->executeQuery();
  189. $row = $result->fetch();
  190. $result->closeCursor();
  191. if ($row) {
  192. $job = $this->buildJob($row);
  193. if ($job instanceof IParallelAwareJob && !$job->getAllowParallelRuns() && $this->hasReservedJob(get_class($job))) {
  194. $this->logger->debug('Skipping ' . get_class($job) . ' job with ID ' . $job->getId() . ' because another job with the same class is already running', ['app' => 'cron']);
  195. $update = $this->connection->getQueryBuilder();
  196. $update->update('jobs')
  197. ->set('last_checked', $update->createNamedParameter($this->timeFactory->getTime() + 1))
  198. ->where($update->expr()->eq('id', $update->createParameter('jobid')));
  199. $update->setParameter('jobid', $row['id']);
  200. $update->executeStatement();
  201. return $this->getNext($onlyTimeSensitive);
  202. }
  203. $update = $this->connection->getQueryBuilder();
  204. $update->update('jobs')
  205. ->set('reserved_at', $update->createNamedParameter($this->timeFactory->getTime()))
  206. ->set('last_checked', $update->createNamedParameter($this->timeFactory->getTime()))
  207. ->where($update->expr()->eq('id', $update->createParameter('jobid')))
  208. ->andWhere($update->expr()->eq('reserved_at', $update->createParameter('reserved_at')))
  209. ->andWhere($update->expr()->eq('last_checked', $update->createParameter('last_checked')));
  210. $update->setParameter('jobid', $row['id']);
  211. $update->setParameter('reserved_at', $row['reserved_at']);
  212. $update->setParameter('last_checked', $row['last_checked']);
  213. $count = $update->executeStatement();
  214. if ($count === 0) {
  215. // Background job already executed elsewhere, try again.
  216. return $this->getNext($onlyTimeSensitive, $jobClasses);
  217. }
  218. if ($job === null) {
  219. // set the last_checked to 12h in the future to not check failing jobs all over again
  220. $reset = $this->connection->getQueryBuilder();
  221. $reset->update('jobs')
  222. ->set('reserved_at', $reset->expr()->literal(0, IQueryBuilder::PARAM_INT))
  223. ->set('last_checked', $reset->createNamedParameter($this->timeFactory->getTime() + 12 * 3600, IQueryBuilder::PARAM_INT))
  224. ->where($reset->expr()->eq('id', $reset->createNamedParameter($row['id'], IQueryBuilder::PARAM_INT)));
  225. $reset->executeStatement();
  226. // Background job from disabled app, try again.
  227. return $this->getNext($onlyTimeSensitive, $jobClasses);
  228. }
  229. return $job;
  230. } else {
  231. return null;
  232. }
  233. }
  234. /**
  235. * @return ?IJob The job matching the id. Beware that this object may be a singleton and may be modified by the next call to buildJob.
  236. */
  237. public function getById(int $id): ?IJob {
  238. $row = $this->getDetailsById($id);
  239. if ($row) {
  240. return $this->buildJob($row);
  241. }
  242. return null;
  243. }
  244. public function getDetailsById(int $id): ?array {
  245. $query = $this->connection->getQueryBuilder();
  246. $query->select('*')
  247. ->from('jobs')
  248. ->where($query->expr()->eq('id', $query->createNamedParameter($id, IQueryBuilder::PARAM_INT)));
  249. $result = $query->executeQuery();
  250. $row = $result->fetch();
  251. $result->closeCursor();
  252. if ($row) {
  253. return $row;
  254. }
  255. return null;
  256. }
  257. /**
  258. * get the job object from a row in the db
  259. *
  260. * @param array{class:class-string<IJob>, id:mixed, last_run:mixed, argument:string} $row
  261. * @return ?IJob the next job to run. Beware that this object may be a singleton and may be modified by the next call to buildJob.
  262. */
  263. private function buildJob(array $row): ?IJob {
  264. try {
  265. try {
  266. // Try to load the job as a service
  267. /** @var IJob $job */
  268. $job = \OCP\Server::get($row['class']);
  269. } catch (QueryException $e) {
  270. if (class_exists($row['class'])) {
  271. $class = $row['class'];
  272. $job = new $class();
  273. } else {
  274. // Remove job from disabled app or old version of an app
  275. $this->removeById($row['id']);
  276. return null;
  277. }
  278. }
  279. if (!($job instanceof IJob)) {
  280. // This most likely means an invalid job was enqueued. We can ignore it.
  281. return null;
  282. }
  283. $job->setId((int) $row['id']);
  284. $job->setLastRun((int) $row['last_run']);
  285. $job->setArgument(json_decode($row['argument'], true));
  286. return $job;
  287. } catch (AutoloadNotAllowedException $e) {
  288. // job is from a disabled app, ignore
  289. return null;
  290. }
  291. }
  292. /**
  293. * set the job that was last ran
  294. */
  295. public function setLastJob(IJob $job): void {
  296. $this->unlockJob($job);
  297. $this->config->setAppValue('backgroundjob', 'lastjob', (string)$job->getId());
  298. }
  299. /**
  300. * Remove the reservation for a job
  301. */
  302. public function unlockJob(IJob $job): void {
  303. $query = $this->connection->getQueryBuilder();
  304. $query->update('jobs')
  305. ->set('reserved_at', $query->expr()->literal(0, IQueryBuilder::PARAM_INT))
  306. ->where($query->expr()->eq('id', $query->createNamedParameter($job->getId(), IQueryBuilder::PARAM_INT)));
  307. $query->executeStatement();
  308. }
  309. /**
  310. * set the lastRun of $job to now
  311. */
  312. public function setLastRun(IJob $job): void {
  313. $query = $this->connection->getQueryBuilder();
  314. $query->update('jobs')
  315. ->set('last_run', $query->createNamedParameter(time(), IQueryBuilder::PARAM_INT))
  316. ->where($query->expr()->eq('id', $query->createNamedParameter($job->getId(), IQueryBuilder::PARAM_INT)));
  317. if ($job instanceof \OCP\BackgroundJob\TimedJob
  318. && !$job->isTimeSensitive()) {
  319. $query->set('time_sensitive', $query->createNamedParameter(IJob::TIME_INSENSITIVE));
  320. }
  321. $query->executeStatement();
  322. }
  323. /**
  324. * @param int $timeTaken
  325. */
  326. public function setExecutionTime(IJob $job, $timeTaken): void {
  327. $query = $this->connection->getQueryBuilder();
  328. $query->update('jobs')
  329. ->set('execution_duration', $query->createNamedParameter($timeTaken, IQueryBuilder::PARAM_INT))
  330. ->set('reserved_at', $query->createNamedParameter(0, IQueryBuilder::PARAM_INT))
  331. ->where($query->expr()->eq('id', $query->createNamedParameter($job->getId(), IQueryBuilder::PARAM_INT)));
  332. $query->executeStatement();
  333. }
  334. /**
  335. * Reset the $job so it executes on the next trigger
  336. *
  337. * @since 23.0.0
  338. */
  339. public function resetBackgroundJob(IJob $job): void {
  340. $query = $this->connection->getQueryBuilder();
  341. $query->update('jobs')
  342. ->set('last_run', $query->createNamedParameter(0, IQueryBuilder::PARAM_INT))
  343. ->set('reserved_at', $query->createNamedParameter(0, IQueryBuilder::PARAM_INT))
  344. ->where($query->expr()->eq('id', $query->createNamedParameter($job->getId()), IQueryBuilder::PARAM_INT));
  345. $query->executeStatement();
  346. }
  347. public function hasReservedJob(?string $className = null): bool {
  348. $query = $this->connection->getQueryBuilder();
  349. $query->select('*')
  350. ->from('jobs')
  351. ->where($query->expr()->gt('reserved_at', $query->createNamedParameter($this->timeFactory->getTime() - 6 * 3600, IQueryBuilder::PARAM_INT)))
  352. ->setMaxResults(1);
  353. if ($className !== null) {
  354. $query->andWhere($query->expr()->eq('class', $query->createNamedParameter($className)));
  355. }
  356. try {
  357. $result = $query->executeQuery();
  358. $hasReservedJobs = $result->fetch() !== false;
  359. $result->closeCursor();
  360. return $hasReservedJobs;
  361. } catch (Exception $e) {
  362. $this->logger->debug('Querying reserved jobs failed', ['exception' => $e]);
  363. return false;
  364. }
  365. }
  366. public function countByClass(): array {
  367. $query = $this->connection->getQueryBuilder();
  368. $query->select('class')
  369. ->selectAlias($query->func()->count('id'), 'count')
  370. ->from('jobs')
  371. ->orderBy('count')
  372. ->groupBy('class');
  373. $result = $query->executeQuery();
  374. $jobs = [];
  375. while (($row = $result->fetch()) !== false) {
  376. /**
  377. * @var array{count:int, class:class-string} $row
  378. */
  379. $jobs[] = $row;
  380. }
  381. return $jobs;
  382. }
  383. }