Add the Scheduler component

.gitattributes

@@ -0,0 +1,4 @@
+/Tests export-ignore
+/phpunit.xml.dist export-ignore
+/.gitattributes export-ignore
+/.gitignore export-ignore

.gitignore

@@ -0,0 +1,3 @@
+vendor/
+composer.lock
+phpunit.xml

CHANGELOG.md

@@ -0,0 +1,7 @@
+CHANGELOG
+=========
+
+6.3
+---
+
+ * Add the component

DependencyInjection/SchedulerPass.php

@@ -0,0 +1,125 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\DependencyInjection;
+
+use Symfony\Component\Cache\CacheItem;
+use Symfony\Component\DependencyInjection\ChildDefinition;
+use Symfony\Component\DependencyInjection\Compiler\CompilerPassInterface;
+use Symfony\Component\DependencyInjection\ContainerBuilder;
+use Symfony\Component\DependencyInjection\Exception\RuntimeException;
+use Symfony\Component\DependencyInjection\Reference;
+use Symfony\Component\Lock\LockFactory;
+use Symfony\Component\Scheduler\Messenger\ScheduleTransportFactory;
+use Symfony\Contracts\Cache\CacheInterface;
+
+class SchedulerPass implements CompilerPassInterface
+{
+    public function process(ContainerBuilder $container)
+    {
+        $usedCachePools = [];
+        $usedLockFactories = [];
+
+        foreach ($container->findTaggedServiceIds('messenger.receiver') as $id => $tags) {
+            $transport = $container->getDefinition($id);
+            [$dsn, $options] = $transport->getArguments();
+            if (!ScheduleTransportFactory::isSupported($dsn)) {
+                continue;
+            }
+
+            if (\is_string($options['cache'] ?? null) && $options['cache']) {
+                $usedCachePools[] = $options['cache'];
+            }
+            if (\is_string($options['lock'] ?? null) && $options['lock']) {
+                $usedLockFactories[] = $options['lock'];
+            }
+            if (\is_array($options['lock'] ?? null) &&
+                \is_string($options['lock']['resource'] ?? null) &&
+                $options['lock']['resource']
+            ) {
+                $usedLockFactories[] = $options['lock']['resource'];
+            }
+        }
+
+        if ($usedCachePools) {
+            $this->locateCachePools($container, $usedCachePools);
+        }
+        if ($usedLockFactories) {
+            $this->locateLockFactories($container, $usedLockFactories);
+        }
+    }
+
+    /**
+     * @param string[] $cachePools
+     */
+    private function locateCachePools(ContainerBuilder $container, array $cachePools): void
+    {
+        if (!class_exists(CacheItem::class)) {
+            throw new \LogicException('You cannot use the "cache" option if the Cache Component is not available. Try running "composer require symfony/cache".');
+        }
+
+        $references = [];
+        foreach (array_unique($cachePools) as $name) {
+            if (!$this->isServiceInstanceOf($container, $id = $name, CacheInterface::class) &&
+                !$this->isServiceInstanceOf($container, $id = 'cache.'.$name, CacheInterface::class)
+            ) {
+                throw new RuntimeException(sprintf('The cache pool "%s" does not exist.', $name));
+            }
+
+            $references[$name] = new Reference($id);
+        }
+
+        $container->getDefinition('scheduler.cache_locator')
+            ->replaceArgument(0, $references);
+    }
+
+    /**
+     * @param string[] $lockFactories
+     */
+    private function locateLockFactories(ContainerBuilder $container, array $lockFactories): void
+    {
+        if (!class_exists(LockFactory::class)) {
+            throw new \LogicException('You cannot use the "lock" option if the Lock Component is not available. Try running "composer require symfony/lock".');
+        }
+
+        $references = [];
+        foreach (array_unique($lockFactories) as $name) {
+            if (!$this->isServiceInstanceOf($container, $id = $name, LockFactory::class) &&
+                !$this->isServiceInstanceOf($container, $id = 'lock.'.$name.'.factory', LockFactory::class)
+            ) {
+                throw new RuntimeException(sprintf('The lock resource "%s" does not exist.', $name));
+            }
+
+            $references[$name] = new Reference($id);
+        }
+
+        $container->getDefinition('scheduler.lock_locator')
+            ->replaceArgument(0, $references);
+    }
+
+    private function isServiceInstanceOf(ContainerBuilder $container, string $serviceId, string $className): bool
+    {
+        if (!$container->hasDefinition($serviceId)) {
+            return false;
+        }
+
+        while (true) {
+            $definition = $container->getDefinition($serviceId);
+            if (!$definition->getClass() && $definition instanceof ChildDefinition) {
+                $serviceId = $definition->getParent();
+
+                continue;
+            }
+
+            return $definition->getClass() && is_a($definition->getClass(), $className, true);
+        }
+    }
+}

Exception/ExceptionInterface.php

@@ -0,0 +1,19 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Exception;
+
+/**
+ * Base Scheduler component's exception.
+ */
+interface ExceptionInterface extends \Throwable
+{
+}

Exception/InvalidArgumentException.php

@@ -0,0 +1,16 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Exception;
+
+class InvalidArgumentException extends \InvalidArgumentException implements ExceptionInterface
+{
+}

Exception/LogicException.php

@@ -0,0 +1,16 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Exception;
+
+class LogicException extends \LogicException implements ExceptionInterface
+{
+}

Exception/LogicMessengerException.php

@@ -0,0 +1,19 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Exception;
+
+use Symfony\Component\Messenger\Exception\ExceptionInterface;
+
+// not sure about this
+class LogicMessengerException extends LogicException implements ExceptionInterface
+{
+}

LICENSE

@@ -0,0 +1,19 @@
+Copyright (c) 2023 Fabien Potencier
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is furnished
+to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.

Locator/ChainScheduleConfigLocator.php

@@ -0,0 +1,64 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Locator;
+
+use Psr\Container\NotFoundExceptionInterface;
+use Symfony\Component\Scheduler\Schedule\ScheduleConfig;
+
+final class ChainScheduleConfigLocator implements ScheduleConfigLocatorInterface
+{
+    /**
+     * @var ScheduleConfigLocatorInterface[]
+     */
+    private array $locators;
+
+    private array $lastFound = [];
+
+    /**
+     * @param iterable<ScheduleConfigLocatorInterface> $locators
+     */
+    public function __construct(iterable $locators)
+    {
+        $this->locators = (static fn (ScheduleConfigLocatorInterface ...$l) => $l)(...$locators);
+    }
+
+    public function get(string $id): ScheduleConfig
+    {
+        if ($locator = $this->findLocator($id)) {
+            return $locator->get($id);
+        }
+
+        throw new class(sprintf('You have requested a non-existent schedule "%s".', $id)) extends \InvalidArgumentException implements NotFoundExceptionInterface { };
+    }
+
+    public function has(string $id): bool
+    {
+        return null !== $this->findLocator($id);
+    }
+
+    private function findLocator(string $id): ?ScheduleConfigLocatorInterface
+    {
+        if (isset($this->lastFound[$id])) {
+            return $this->lastFound[$id];
+        }
+
+        foreach ($this->locators as $locator) {
+            if ($locator->has($id)) {
+                $this->lastFound = [$id => $locator];
+
+                return $locator;
+            }
+        }
+
+        return null;
+    }
+}

Locator/ScheduleConfigLocatorInterface.php

@@ -0,0 +1,20 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Locator;
+
+use Psr\Container\ContainerInterface;
+use Symfony\Component\Scheduler\Schedule\ScheduleConfig;
+
+interface ScheduleConfigLocatorInterface extends ContainerInterface
+{
+    public function get(string $id): ScheduleConfig;
+}

Messenger/ScheduleTransport.php

@@ -0,0 +1,50 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Messenger;
+
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Messenger\Transport\TransportInterface;
+use Symfony\Component\Scheduler\Exception\LogicMessengerException;
+use Symfony\Component\Scheduler\Schedule\ScheduleInterface;
+
+class ScheduleTransport implements TransportInterface
+{
+    private readonly array $stamps;
+
+    public function __construct(
+        private readonly ScheduleInterface $schedule,
+    ) {
+        $this->stamps = [new ScheduledStamp()];
+    }
+
+    public function get(): iterable
+    {
+        foreach ($this->schedule->getMessages() as $message) {
+            yield new Envelope($message, $this->stamps);
+        }
+    }
+
+    public function ack(Envelope $envelope): void
+    {
+        // ignore
+    }
+
+    public function reject(Envelope $envelope): void
+    {
+        throw new LogicMessengerException('Messages from ScheduleTransport must not be rejected.');
+    }
+
+    public function send(Envelope $envelope): Envelope
+    {
+        throw new LogicMessengerException('The ScheduleTransport cannot send messages.');
+    }
+}

Messenger/ScheduleTransportFactory.php

@@ -0,0 +1,73 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Messenger;
+
+use Psr\Clock\ClockInterface;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use Symfony\Component\Messenger\Transport\TransportFactoryInterface;
+use Symfony\Component\Scheduler\Exception\InvalidArgumentException;
+use Symfony\Component\Scheduler\Locator\ScheduleConfigLocatorInterface;
+use Symfony\Component\Scheduler\Schedule\Schedule;
+use Symfony\Component\Scheduler\State\StateFactoryInterface;
+
+class ScheduleTransportFactory implements TransportFactoryInterface
+{
+    protected const DEFAULT_OPTIONS = [
+        'cache' => null,
+        'lock' => null,
+    ];
+
+    public function __construct(
+        private readonly ClockInterface $clock,
+        private readonly ScheduleConfigLocatorInterface $schedules,
+        private readonly StateFactoryInterface $stateFactory,
+    ) {
+    }
+
+    public function createTransport(string $dsn, array $options, SerializerInterface $serializer): ScheduleTransport
+    {
+        if ('schedule://' === $dsn) {
+            throw new InvalidArgumentException('The Schedule DSN must contains a name, e.g. "schedule://default".');
+        }
+        if (false === $scheduleName = parse_url($dsn, \PHP_URL_HOST)) {
+            throw new InvalidArgumentException(sprintf('The given Schedule DSN "%s" is invalid.', $dsn));
+        }
+
+        unset($options['transport_name']);
+        $options += static::DEFAULT_OPTIONS;
+        if (0 < \count($invalidOptions = array_diff_key($options, static::DEFAULT_OPTIONS))) {
+            throw new InvalidArgumentException(sprintf('Invalid option(s) "%s" passed to the Schedule Messenger transport.', implode('", "', array_keys($invalidOptions))));
+        }
+
+        if (!$this->schedules->has($scheduleName)) {
+            throw new InvalidArgumentException(sprintf('The schedule "%s" is not found.', $scheduleName));
+        }
+
+        return new ScheduleTransport(
+            new Schedule(
+                $this->clock,
+                $this->stateFactory->create($scheduleName, $options),
+                $this->schedules->get($scheduleName)
+            )
+        );
+    }
+
+    public function supports(string $dsn, array $options): bool
+    {
+        return self::isSupported($dsn);
+    }
+
+    final public static function isSupported(string $dsn): bool
+    {
+        return str_starts_with($dsn, 'schedule://');
+    }
+}

Messenger/ScheduledStamp.php

@@ -0,0 +1,18 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Messenger;
+
+use Symfony\Component\Messenger\Stamp\NonSendableStampInterface;
+
+final class ScheduledStamp implements NonSendableStampInterface
+{
+}

README.md

@@ -0,0 +1,56 @@
+Scheduler Component
+====================
+
+Provides basic scheduling through the Symfony Messenger.
+
+Getting Started
+---------------
+
+```
+$ composer require symfony/scheduler
+```
+
+Full DSN with schedule name: `schedule://<name>`
+
+```yaml
+# messenger.yaml
+framework:
+  messenger:
+    transports:
+      schedule_default: 'schedule://default'
+```
+
+```php
+<?php
+
+use Symfony\Component\Scheduler\ScheduleConfig;
+use Symfony\Component\Scheduler\Trigger\PeriodicalTrigger;
+
+class ExampleLocator implements ScheduleConfigLocatorInterface
+{
+    public function get(string $id): ScheduleConfig
+    {
+        return (new ScheduleConfig())
+            ->add(
+                // do the MaintenanceJob every night at 3 a.m. UTC
+                PeriodicalTrigger::create('P1D', '03:00:00+00'),
+                new MaintenanceJob()
+            )
+        ;
+    }
+
+    public function has(string $id): bool
+    {
+        return 'default' === $id;
+    }
+}
+```
+
+Resources
+---------
+
+* [Documentation](https://symfony.com/doc/current/scheduler.html)
+* [Contributing](https://symfony.com/doc/current/contributing/index.html)
+* [Report issues](https://github.com/symfony/symfony/issues) and
+  [send Pull Requests](https://github.com/symfony/symfony/pulls)
+  in the [main Symfony repository](https://github.com/symfony/symfony)

Schedule/Schedule.php

@@ -0,0 +1,95 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Schedule;
+
+use Psr\Clock\ClockInterface;
+use Symfony\Component\Scheduler\State\StateInterface;
+use Symfony\Component\Scheduler\Trigger\TriggerInterface;
+
+final class Schedule implements ScheduleInterface
+{
+    /**
+     * @var array<int, array{TriggerInterface, object}>
+     */
+    private readonly array $schedule;
+    private ScheduleHeap $scheduleHeap;
+    private ?\DateTimeImmutable $waitUntil;
+
+    public function __construct(
+        private readonly ClockInterface $clock,
+        private readonly StateInterface $state,
+        ScheduleConfig $scheduleConfig,
+    ) {
+        $this->schedule = $scheduleConfig->getSchedule();
+        $this->waitUntil = new \DateTimeImmutable('@0');
+    }
+
+    public function getMessages(): \Generator
+    {
+        if (!$this->waitUntil ||
+            $this->waitUntil > ($now = $this->clock->now()) ||
+            !$this->state->acquire($now)
+        ) {
+            return;
+        }
+
+        $lastTime = $this->state->time();
+        $lastIndex = $this->state->index();
+        $heap = $this->heap($lastTime);
+
+        while (!$heap->isEmpty() && $heap->top()[0] <= $now) {
+            /** @var TriggerInterface $trigger */
+            [$time, $index, $trigger, $message] = $heap->extract();
+            $yield = true;
+
+            if ($time < $lastTime) {
+                $time = $lastTime;
+                $yield = false;
+            } elseif ($time == $lastTime && $index <= $lastIndex) {
+                $yield = false;
+            }
+
+            if ($nextTime = $trigger->nextTo($time)) {
+                $heap->insert([$nextTime, $index, $trigger, $message]);
+            }
+
+            if ($yield) {
+                yield $message;
+                $this->state->save($time, $index);
+            }
+        }
+
+        $this->waitUntil = $heap->isEmpty() ? null : $heap->top()[0];
+
+        $this->state->release($now, $this->waitUntil);
+    }
+
+    private function heap(\DateTimeImmutable $time): ScheduleHeap
+    {
+        if (isset($this->scheduleHeap) && $this->scheduleHeap->time <= $time) {
+            return $this->scheduleHeap;
+        }
+
+        $heap = new ScheduleHeap($time);
+
+        foreach ($this->schedule as $index => [$trigger, $message]) {
+            /** @var TriggerInterface $trigger */
+            if (!$nextTime = $trigger->nextTo($time)) {
+                continue;
+            }
+
+            $heap->insert([$nextTime, $index, $trigger, $message]);
+        }
+
+        return $this->scheduleHeap = $heap;
+    }
+}

Schedule/ScheduleConfig.php

@@ -0,0 +1,47 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Schedule;
+
+use Symfony\Component\Scheduler\Trigger\TriggerInterface;
+
+final class ScheduleConfig
+{
+    /**
+     * @var array<int, array{TriggerInterface, object}>
+     */
+    private array $schedule = [];
+
+    /**
+     * @param iterable<array{TriggerInterface, object}> $schedule
+     */
+    public function __construct(iterable $schedule = [])
+    {
+        foreach ($schedule as $args) {
+            $this->add(...$args);
+        }
+    }
+
+    public function add(TriggerInterface $trigger, object $message): self
+    {
+        $this->schedule[] = [$trigger, $message];
+
+        return $this;
+    }
+
+    /**
+     * @return array<int, array{TriggerInterface, object}>
+     */
+    public function getSchedule(): array
+    {
+        return $this->schedule;
+    }
+}

Schedule/ScheduleHeap.php

@@ -0,0 +1,36 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Schedule;
+
+use Symfony\Component\Scheduler\Trigger\TriggerInterface;
+
+/**
+ * @internal
+ *
+ * @extends \SplHeap<array{\DateTimeImmutable, int, TriggerInterface, object}>
+ */
+final class ScheduleHeap extends \SplHeap
+{
+    public function __construct(
+        public \DateTimeImmutable $time,
+    ) {
+    }
+
+    /**
+     * @param array{\DateTimeImmutable, int, TriggerInterface, object} $value1
+     * @param array{\DateTimeImmutable, int, TriggerInterface, object} $value2
+     */
+    protected function compare(mixed $value1, mixed $value2): int
+    {
+        return $value2[0] <=> $value1[0] ?: $value2[1] <=> $value1[1];
+    }
+}

Schedule/ScheduleInterface.php

@@ -0,0 +1,17 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Schedule;
+
+interface ScheduleInterface
+{
+    public function getMessages(): iterable;
+}

State/CacheStateDecorator.php

@@ -0,0 +1,56 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\State;
+
+use Symfony\Contracts\Cache\CacheInterface;
+
+final class CacheStateDecorator implements StateInterface
+{
+    public function __construct(
+        private readonly StateInterface $inner,
+        private readonly CacheInterface $cache,
+        private readonly string $name,
+    ) {
+    }
+
+    public function acquire(\DateTimeImmutable $now): bool
+    {
+        if (!$this->inner->acquire($now)) {
+            return false;
+        }
+
+        $this->inner->save(...$this->cache->get($this->name, fn () => [$now, -1]));
+
+        return true;
+    }
+
+    public function time(): \DateTimeImmutable
+    {
+        return $this->inner->time();
+    }
+
+    public function index(): int
+    {
+        return $this->inner->index();
+    }
+
+    public function save(\DateTimeImmutable $time, int $index): void
+    {
+        $this->inner->save($time, $index);
+        $this->cache->get($this->name, fn () => [$time, $index], \INF);
+    }
+
+    public function release(\DateTimeImmutable $now, ?\DateTimeImmutable $nextTime): void
+    {
+        $this->inner->release($now, $nextTime);
+    }
+}

State/LockStateDecorator.php

@@ -0,0 +1,73 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\State;
+
+use Symfony\Component\Lock\LockInterface;
+
+final class LockStateDecorator implements StateInterface
+{
+    private bool $reset = false;
+
+    public function __construct(
+        private readonly State $inner,
+        private readonly LockInterface $lock,
+    ) {
+    }
+
+    public function acquire(\DateTimeImmutable $now): bool
+    {
+        if (!$this->lock->acquire()) {
+            // Reset local state if a `Lock` is acquired by another `Worker`.
+            $this->reset = true;
+
+            return false;
+        }
+
+        if ($this->reset) {
+            $this->reset = false;
+            $this->inner->save($now, -1);
+        }
+
+        return $this->inner->acquire($now);
+    }
+
+    public function time(): \DateTimeImmutable
+    {
+        return $this->inner->time();
+    }
+
+    public function index(): int
+    {
+        return $this->inner->index();
+    }
+
+    public function save(\DateTimeImmutable $time, int $index): void
+    {
+        $this->inner->save($time, $index);
+    }
+
+    /**
+     * Releases `State`, not `Lock`.
+     *
+     * It tries to keep a `Lock` as long as a `Worker` is alive.
+     */
+    public function release(\DateTimeImmutable $now, ?\DateTimeImmutable $nextTime): void
+    {
+        $this->inner->release($now, $nextTime);
+
+        if (!$nextTime) {
+            $this->lock->release();
+        } elseif ($remaining = $this->lock->getRemainingLifetime()) {
+            $this->lock->refresh((float) $nextTime->format('U.u') - (float) $now->format('U.u') + $remaining);
+        }
+    }
+}

State/State.php

@@ -0,0 +1,48 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\State;
+
+final class State implements StateInterface
+{
+    private \DateTimeImmutable $time;
+    private int $index = -1;
+
+    public function acquire(\DateTimeImmutable $now): bool
+    {
+        if (!isset($this->time)) {
+            $this->time = $now;
+        }
+
+        return true;
+    }
+
+    public function time(): \DateTimeImmutable
+    {
+        return $this->time;
+    }
+
+    public function index(): int
+    {
+        return $this->index;
+    }
+
+    public function save(\DateTimeImmutable $time, int $index): void
+    {
+        $this->time = $time;
+        $this->index = $index;
+    }
+
+    public function release(\DateTimeImmutable $now, ?\DateTimeImmutable $nextTime): void
+    {
+        // skip
+    }
+}

State/StateFactory.php

@@ -0,0 +1,91 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\State;
+
+use Psr\Container\ContainerInterface;
+use Symfony\Component\Lock\LockFactory;
+use Symfony\Component\Lock\LockInterface;
+use Symfony\Component\Scheduler\Exception\LogicException;
+use Symfony\Contracts\Cache\CacheInterface;
+
+final class StateFactory implements StateFactoryInterface
+{
+    public function __construct(
+        private readonly ContainerInterface $lockFactories,
+        private readonly ContainerInterface $caches,
+    ) {
+    }
+
+    public function create(string $scheduleName, array $options): StateInterface
+    {
+        $name = 'messenger.schedule.'.$scheduleName;
+        $state = new State();
+
+        if ($lock = $this->createLock($scheduleName, $name, $options)) {
+            $state = new LockStateDecorator($state, $lock);
+        }
+        if ($cache = $this->createCache($scheduleName, $options)) {
+            $state = new CacheStateDecorator($state, $cache, $name);
+        }
+
+        return $state;
+    }
+
+    private function createLock(string $scheduleName, string $resourceName, array $options): ?LockInterface
+    {
+        if (!($options['lock'] ?? false)) {
+            return null;
+        }
+
+        if (\is_string($options['lock'])) {
+            $options['lock'] = ['resource' => $options['lock']];
+        }
+
+        if (\is_array($options['lock']) && \is_string($resource = $options['lock']['resource'] ?? null)) {
+            if (!$this->lockFactories->has($resource)) {
+                throw new LogicException(sprintf('The lock resource "%s" does not exist.', $resource));
+            }
+
+            /** @var LockFactory $lockFactory */
+            $lockFactory = $this->lockFactories->get($resource);
+
+            $args = ['resource' => $resourceName];
+            if (isset($options['lock']['ttl'])) {
+                $args['ttl'] = (float) $options['lock']['ttl'];
+            }
+            if (isset($options['lock']['auto_release'])) {
+                $args['autoRelease'] = (float) $options['lock']['auto_release'];
+            }
+
+            return $lockFactory->createLock(...$args);
+        }
+
+        throw new LogicException(sprintf('Invalid lock configuration for "%s" schedule.', $scheduleName));
+    }
+
+    private function createCache(string $scheduleName, array $options): ?CacheInterface
+    {
+        if (!($options['cache'] ?? false)) {
+            return null;
+        }
+
+        if (\is_string($options['cache'])) {
+            if (!$this->caches->has($options['cache'])) {
+                throw new LogicException(sprintf('The cache pool "%s" does not exist.', $options['cache']));
+            }
+
+            return $this->caches->get($options['cache']);
+        }
+
+        throw new LogicException(sprintf('Invalid cache configuration for "%s" schedule.', $scheduleName));
+    }
+}

State/StateFactoryInterface.php

@@ -0,0 +1,20 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\State;
+
+interface StateFactoryInterface
+{
+    /**
+     * @param array<string, int|float|string|bool|null> $options
+     */
+    public function create(string $scheduleName, array $options): StateInterface;
+}

State/StateInterface.php

@@ -0,0 +1,25 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\State;
+
+interface StateInterface
+{
+    public function acquire(\DateTimeImmutable $now): bool;
+
+    public function time(): \DateTimeImmutable;
+
+    public function index(): int;
+
+    public function save(\DateTimeImmutable $time, int $index): void;
+
+    public function release(\DateTimeImmutable $now, ?\DateTimeImmutable $nextTime): void;
+}

Tests/Locator/ChainScheduleConfigLocatorTest.php

@@ -0,0 +1,49 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\Locator;
+
+use PHPUnit\Framework\TestCase;
+use Psr\Container\NotFoundExceptionInterface;
+use Symfony\Component\Scheduler\Locator\ChainScheduleConfigLocator;
+use Symfony\Component\Scheduler\Locator\ScheduleConfigLocatorInterface;
+use Symfony\Component\Scheduler\Schedule\ScheduleConfig;
+
+class ChainScheduleConfigLocatorTest extends TestCase
+{
+    public function testExists()
+    {
+        $schedule = new ScheduleConfig();
+
+        $empty = $this->createMock(ScheduleConfigLocatorInterface::class);
+        $empty->expects($this->once())->method('has')->with('exists')->willReturn(false);
+        $empty->expects($this->never())->method('get');
+
+        $full = $this->createMock(ScheduleConfigLocatorInterface::class);
+        $full->expects($this->once())->method('has')->with('exists')->willReturn(true);
+        $full->expects($this->once())->method('get')->with('exists')->willReturn($schedule);
+
+        $locator = new ChainScheduleConfigLocator([$empty, $full]);
+
+        $this->assertTrue($locator->has('exists'));
+        $this->assertSame($schedule, $locator->get('exists'));
+    }
+
+    public function testNonExists()
+    {
+        $locator = new ChainScheduleConfigLocator([$this->createMock(ScheduleConfigLocatorInterface::class)]);
+
+        $this->assertFalse($locator->has('non-exists'));
+        $this->expectException(NotFoundExceptionInterface::class);
+
+        $locator->get('non-exists');
+    }
+}

Tests/Messenger/ScheduleTransportFactoryTest.php

@@ -0,0 +1,125 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\Messenger;
+
+use PHPUnit\Framework\TestCase;
+use Psr\Clock\ClockInterface;
+use Symfony\Component\Messenger\Transport\Serialization\SerializerInterface;
+use Symfony\Component\Scheduler\Exception\InvalidArgumentException;
+use Symfony\Component\Scheduler\Locator\ScheduleConfigLocatorInterface;
+use Symfony\Component\Scheduler\Messenger\ScheduleTransport;
+use Symfony\Component\Scheduler\Messenger\ScheduleTransportFactory;
+use Symfony\Component\Scheduler\Schedule\Schedule;
+use Symfony\Component\Scheduler\Schedule\ScheduleConfig;
+use Symfony\Component\Scheduler\State\StateFactoryInterface;
+use Symfony\Component\Scheduler\State\StateInterface;
+use Symfony\Component\Scheduler\Trigger\TriggerInterface;
+
+class ScheduleTransportFactoryTest extends TestCase
+{
+    public function testCreateTransport()
+    {
+        $trigger = $this->createMock(TriggerInterface::class);
+        $serializer = $this->createMock(SerializerInterface::class);
+        $clock = $this->createMock(ClockInterface::class);
+        $container = new class() extends \ArrayObject implements ScheduleConfigLocatorInterface {
+            public function get(string $id): ScheduleConfig
+            {
+                return $this->offsetGet($id);
+            }
+
+            public function has(string $id): bool
+            {
+                return $this->offsetExists($id);
+            }
+        };
+
+        $stateFactory = $this->createMock(StateFactoryInterface::class);
+        $stateFactory
+            ->expects($this->exactly(2))
+            ->method('create')
+            ->withConsecutive(
+                ['default', ['cache' => null, 'lock' => null]],
+                ['custom', ['cache' => 'app', 'lock' => null]]
+            )
+            ->willReturn($state = $this->createMock(StateInterface::class));
+
+        $container['default'] = new ScheduleConfig([[$trigger, (object) ['id' => 'default']]]);
+        $container['custom'] = new ScheduleConfig([[$trigger, (object) ['id' => 'custom']]]);
+        $default = new ScheduleTransport(new Schedule($clock, $state, $container['default']));
+        $custom = new ScheduleTransport(new Schedule($clock, $state, $container['custom']));
+
+        $factory = new ScheduleTransportFactory($clock, $container, $stateFactory);
+
+        $this->assertEquals($default, $factory->createTransport('schedule://default', [], $serializer));
+        $this->assertEquals($custom, $factory->createTransport('schedule://custom', ['cache' => 'app'], $serializer));
+    }
+
+    public function testInvalidDsn()
+    {
+        $factory = $this->makeTransportFactoryWithStubs();
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('The given Schedule DSN "schedule://#wrong" is invalid.');
+
+        $factory->createTransport('schedule://#wrong', [], $this->createMock(SerializerInterface::class));
+    }
+
+    public function testNoName()
+    {
+        $factory = $this->makeTransportFactoryWithStubs();
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('The Schedule DSN must contains a name, e.g. "schedule://default".');
+
+        $factory->createTransport('schedule://', [], $this->createMock(SerializerInterface::class));
+    }
+
+    public function testInvalidOption()
+    {
+        $factory = $this->makeTransportFactoryWithStubs();
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('Invalid option(s) "invalid" passed to the Schedule Messenger transport.');
+
+        $factory->createTransport('schedule://name', ['invalid' => true], $this->createMock(SerializerInterface::class));
+    }
+
+    public function testNotFound()
+    {
+        $factory = $this->makeTransportFactoryWithStubs();
+
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('The schedule "not-exists" is not found.');
+
+        $factory->createTransport('schedule://not-exists', [], $this->createMock(SerializerInterface::class));
+    }
+
+    public function testSupports()
+    {
+        $factory = $this->makeTransportFactoryWithStubs();
+
+        $this->assertTrue($factory->supports('schedule://', []));
+        $this->assertTrue($factory->supports('schedule://name', []));
+        $this->assertFalse($factory->supports('', []));
+        $this->assertFalse($factory->supports('string', []));
+    }
+
+    private function makeTransportFactoryWithStubs(): ScheduleTransportFactory
+    {
+        return new ScheduleTransportFactory(
+            $this->createMock(ClockInterface::class),
+            $this->createMock(ScheduleConfigLocatorInterface::class),
+            $this->createMock(StateFactoryInterface::class)
+        );
+    }
+}

Tests/Messenger/ScheduleTransportTest.php

@@ -0,0 +1,67 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\Messenger;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Messenger\Envelope;
+use Symfony\Component\Scheduler\Exception\LogicMessengerException;
+use Symfony\Component\Scheduler\Messenger\ScheduledStamp;
+use Symfony\Component\Scheduler\Messenger\ScheduleTransport;
+use Symfony\Component\Scheduler\Schedule\ScheduleInterface;
+
+class ScheduleTransportTest extends TestCase
+{
+    public function testGetFromIterator()
+    {
+        $messages = [
+            (object) ['id' => 'first'],
+            (object) ['id' => 'second'],
+        ];
+        $scheduler = $this->createConfiguredMock(ScheduleInterface::class, [
+            'getMessages' => $messages,
+        ]);
+        $transport = new ScheduleTransport($scheduler);
+
+        foreach ($transport->get() as $envelope) {
+            $this->assertInstanceOf(Envelope::class, $envelope);
+            $this->assertNotNull($envelope->last(ScheduledStamp::class));
+            $this->assertSame(array_shift($messages), $envelope->getMessage());
+        }
+
+        $this->assertEmpty($messages);
+    }
+
+    public function testAckIgnored()
+    {
+        $transport = new ScheduleTransport($this->createMock(ScheduleInterface::class));
+
+        $transport->ack(new Envelope(new \stdClass()));
+
+        $this->assertTrue(true); // count coverage
+    }
+
+    public function testRejectException()
+    {
+        $transport = new ScheduleTransport($this->createMock(ScheduleInterface::class));
+
+        $this->expectException(LogicMessengerException::class);
+        $transport->reject(new Envelope(new \stdClass()));
+    }
+
+    public function testSendException()
+    {
+        $transport = new ScheduleTransport($this->createMock(ScheduleInterface::class));
+
+        $this->expectException(LogicMessengerException::class);
+        $transport->send(new Envelope(new \stdClass()));
+    }
+}

Tests/Schedule/ScheduleConfigTest.php

@@ -0,0 +1,61 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\Schedule;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Scheduler\Schedule\ScheduleConfig;
+use Symfony\Component\Scheduler\Trigger\TriggerInterface;
+
+class ScheduleConfigTest extends TestCase
+{
+    public function testEmpty()
+    {
+        $config = new ScheduleConfig();
+
+        $this->assertSame([], $config->getSchedule());
+    }
+
+    public function testAdd()
+    {
+        $config = new ScheduleConfig();
+
+        $config->add($t1 = $this->createMock(TriggerInterface::class), $o1 = (object) ['name' => 'first']);
+        $config->add($t2 = $this->createMock(TriggerInterface::class), $o2 = (object) ['name' => 'second']);
+
+        $expected = [
+            [$t1, $o1],
+            [$t2, $o2],
+        ];
+
+        $this->assertSame($expected, $config->getSchedule());
+    }
+
+    public function testFromIterator()
+    {
+        $expected = [
+            [$this->createMock(TriggerInterface::class), (object) ['name' => 'first']],
+            [$this->createMock(TriggerInterface::class), (object) ['name' => 'second']],
+        ];
+
+        $config = new ScheduleConfig(new \ArrayObject($expected));
+
+        $this->assertSame($expected, $config->getSchedule());
+    }
+
+    public function testFromBadIterator()
+    {
+        $this->expectException(\TypeError::class);
+        $this->expectExceptionMessage('must be of type Symfony\Component\Scheduler\Trigger\TriggerInterface');
+
+        new ScheduleConfig([new \ArrayObject(['wrong'])]);
+    }
+}

Tests/Schedule/ScheduleTest.php

@@ -0,0 +1,180 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\Schedule;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Clock\ClockInterface;
+use Symfony\Component\Scheduler\Schedule\Schedule;
+use Symfony\Component\Scheduler\Schedule\ScheduleConfig;
+use Symfony\Component\Scheduler\State\State;
+use Symfony\Component\Scheduler\Trigger\TriggerInterface;
+
+class ScheduleTest extends TestCase
+{
+    public function messagesProvider(): \Generator
+    {
+        $first = (object) ['id' => 'first'];
+        $second = (object) ['id' => 'second'];
+        $third = (object) ['id' => 'third'];
+
+        yield 'first' => [
+            'startTime' => '22:12:00',
+            'runs' => [
+                '22:12:00' => [],
+                '22:12:01' => [],
+                '22:13:00' => [$first],
+                '22:13:01' => [],
+            ],
+            'schedule' => [
+                $this->makeSchedule($first, '22:13:00', '22:14:00'),
+            ],
+        ];
+
+        yield 'microseconds' => [
+            'startTime' => '22:12:00',
+            'runs' => [
+                '22:12:59.999999' => [],
+                '22:13:00' => [$first],
+                '22:13:01' => [],
+            ],
+            'schedule' => [
+                $this->makeSchedule($first, '22:13:00', '22:14:00', '22:15:00'),
+            ],
+        ];
+
+        yield 'skipped' => [
+            'startTime' => '22:12:00',
+            'runs' => [
+                '22:14:01' => [$first, $first],
+            ],
+            'schedule' => [
+                $this->makeSchedule($first, '22:13:00', '22:14:00', '22:15:00'),
+            ],
+        ];
+
+        yield 'sequence' => [
+            'startTime' => '22:12:00',
+            'runs' => [
+                '22:12:59' => [],
+                '22:13:00' => [$first],
+                '22:13:01' => [],
+                '22:13:59' => [],
+                '22:14:00' => [$first],
+                '22:14:01' => [],
+            ],
+            'schedule' => [
+                $this->makeSchedule($first, '22:13:00', '22:14:00', '22:15:00'),
+            ],
+        ];
+
+        yield 'concurrency' => [
+            'startTime' => '22:12:00',
+            'runs' => [
+                '22:12:00.555' => [],
+                '22:13:01.555' => [$third, $first, $first, $second, $first],
+                '22:13:02.000' => [$first],
+                '22:13:02.555' => [],
+            ],
+            'schedule' => [
+                $this->makeSchedule($first, '22:12:59', '22:13:00', '22:13:01', '22:13:02', '22:13:03'),
+                $this->makeSchedule($second, '22:13:00', '22:14:00'),
+                $this->makeSchedule($third, '22:12:30', '22:13:30'),
+            ],
+        ];
+
+        yield 'parallel' => [
+            'startTime' => '22:12:00',
+            'runs' => [
+                '22:12:59' => [],
+                '22:13:59' => [$first, $second],
+                '22:14:00' => [$first, $second],
+                '22:14:01' => [],
+            ],
+            'schedule' => [
+                $this->makeSchedule($first, '22:13:00', '22:14:00', '22:15:00'),
+                $this->makeSchedule($second, '22:13:00', '22:14:00', '22:15:00'),
+            ],
+        ];
+
+        yield 'past' => [
+            'startTime' => '22:12:00',
+            'runs' => [
+                '22:12:01' => [],
+            ],
+            'schedule' => [
+                [$this->createMock(TriggerInterface::class), $this],
+            ],
+        ];
+    }
+
+    /**
+     * @dataProvider messagesProvider
+     */
+    public function testGetMessages(string $startTime, array $runs, array $schedule)
+    {
+        // for referencing
+        $now = $this->makeDateTime($startTime);
+
+        $clock = $this->createMock(ClockInterface::class);
+        $clock->method('now')->willReturnReference($now);
+
+        $scheduler = new Schedule($clock, new State(), new ScheduleConfig($schedule));
+
+        // Warmup. The first run is always returns nothing.
+        $this->assertSame([], iterator_to_array($scheduler->getMessages()));
+
+        foreach ($runs as $time => $expected) {
+            $now = $this->makeDateTime($time);
+            $this->assertSame($expected, iterator_to_array($scheduler->getMessages()));
+        }
+    }
+
+    private function makeDateTime(string $time): \DateTimeImmutable
+    {
+        return new \DateTimeImmutable('2020-02-20T'.$time, new \DateTimeZone('UTC'));
+    }
+
+    /**
+     * @return array{TriggerInterface, object}
+     */
+    private function makeSchedule(object $message, string ...$runs): array
+    {
+        $runs = array_map(fn ($time) => $this->makeDateTime($time), $runs);
+        sort($runs);
+
+        $ticks = [$this->makeDateTime(''), 0];
+
+        $trigger = $this->createMock(TriggerInterface::class);
+        $trigger
+            ->method('nextTo')
+            ->willReturnCallback(function (\DateTimeImmutable $lastTick) use ($runs, &$ticks): \DateTimeImmutable {
+                [$tick, $count] = $ticks;
+                if ($lastTick > $tick) {
+                    $ticks = [$lastTick, 1];
+                } elseif ($lastTick == $tick && $count < 2) {
+                    $ticks = [$lastTick, ++$count];
+                } else {
+                    $this->fail(sprintf('Invalid tick %s', $lastTick->format(\DateTimeImmutable::RFC3339_EXTENDED)));
+                }
+
+                foreach ($runs as $run) {
+                    if ($lastTick < $run) {
+                        return $run;
+                    }
+                }
+
+                $this->fail(sprintf('There is no next run for tick %s', $lastTick->format(\DateTimeImmutable::RFC3339_EXTENDED)));
+            });
+
+        return [$trigger, $message];
+    }
+}

Tests/State/CacheStateDecoratorTest.php

@@ -0,0 +1,115 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\State;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Cache\Adapter\ArrayAdapter;
+use Symfony\Component\Scheduler\State\CacheStateDecorator;
+use Symfony\Component\Scheduler\State\State;
+use Symfony\Component\Scheduler\State\StateInterface;
+use Symfony\Contracts\Cache\CacheInterface;
+
+class CacheStateDecoratorTest extends TestCase
+{
+    private ArrayAdapter $cache;
+    private State $inner;
+    private CacheStateDecorator $state;
+    private \DateTimeImmutable $now;
+
+    protected function setUp(): void
+    {
+        $this->cache = new ArrayAdapter(storeSerialized: false);
+        $this->inner = new State();
+        $this->state = new CacheStateDecorator($this->inner, $this->cache, 'cache');
+        $this->now = new \DateTimeImmutable('2020-02-20 20:20:20Z');
+    }
+
+    public function testInitStateOnFirstAcquiring()
+    {
+        [$cache, $state, $now] = [$this->cache, $this->state, $this->now];
+
+        $this->assertTrue($state->acquire($now));
+        $this->assertEquals($now, $state->time());
+        $this->assertEquals(-1, $state->index());
+        $this->assertEquals([$now, -1], $cache->get('cache', fn () => []));
+    }
+
+    public function testLoadStateOnAcquiring()
+    {
+        [$cache, $inner, $state, $now] = [$this->cache, $this->inner, $this->state, $this->now];
+
+        $cache->get('cache', fn () => [$now, 0], \INF);
+
+        $this->assertTrue($state->acquire($now->modify('1 min')));
+        $this->assertEquals($now, $state->time());
+        $this->assertEquals(0, $state->index());
+        $this->assertEquals([$now, 0], $cache->get('cache', fn () => []));
+    }
+
+    public function testCannotAcquereIfInnerAcquered()
+    {
+        $inner = $this->createMock(StateInterface::class);
+        $inner->method('acquire')->willReturn(false);
+        $state = new CacheStateDecorator($inner, $this->cache, 'cache');
+
+        $this->assertFalse($state->acquire($this->now));
+    }
+
+    public function testSave()
+    {
+        [$cache, $inner, $state, $now] = [$this->cache, $this->inner, $this->state, $this->now];
+
+        $state->acquire($now->modify('-1 hour'));
+        $state->save($now, 3);
+
+        $this->assertSame($now, $state->time());
+        $this->assertSame(3, $state->index());
+        $this->assertSame($inner->time(), $state->time());
+        $this->assertSame($inner->index(), $state->index());
+        $this->assertSame([$now, 3], $cache->get('cache', fn () => []));
+    }
+
+    public function testRelease()
+    {
+        $now = $this->now;
+        $later = $now->modify('1 min');
+        $cache = $this->createMock(CacheInterface::class);
+        $inner = $this->createMock(StateInterface::class);
+        $inner->expects($this->once())->method('release')->with($now, $later);
+        $state = new CacheStateDecorator($inner, $cache, 'cache');
+
+        $state->release($now, $later);
+    }
+
+    public function testFullCycle()
+    {
+        [$cache, $inner, $state, $now] = [$this->cache, $this->inner, $this->state, $this->now];
+
+        // init
+        $cache->get('cache', fn () => [$now->modify('-1 min'), 3], \INF);
+
+        // action
+        $acquired = $state->acquire($now);
+        $lastTime = $state->time();
+        $lastIndex = $state->index();
+        $state->save($now, 0);
+        $state->release($now, null);
+
+        // asserting
+        $this->assertTrue($acquired);
+        $this->assertEquals($now->modify('-1 min'), $lastTime);
+        $this->assertSame(3, $lastIndex);
+        $this->assertEquals($now, $inner->time());
+        $this->assertSame(0, $inner->index());
+        $this->assertEquals([$now, 0], $cache->get('cache', fn () => []));
+    }
+}

Tests/State/LockStateDecoratorTest.php

@@ -0,0 +1,172 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\State;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Lock\Key;
+use Symfony\Component\Lock\Lock;
+use Symfony\Component\Lock\LockInterface;
+use Symfony\Component\Lock\Store\InMemoryStore;
+use Symfony\Component\Scheduler\State\LockStateDecorator;
+use Symfony\Component\Scheduler\State\State;
+
+class LockStateDecoratorTest extends TestCase
+{
+    private InMemoryStore $store;
+    private Lock $lock;
+    private State $inner;
+    private LockStateDecorator $state;
+    private \DateTimeImmutable $now;
+
+    protected function setUp(): void
+    {
+        $this->store = new InMemoryStore();
+        $this->lock = new Lock(new Key('lock'), $this->store);
+        $this->inner = new State();
+        $this->state = new LockStateDecorator($this->inner, $this->lock);
+        $this->now = new \DateTimeImmutable('2020-02-20 20:20:20Z');
+    }
+
+    public function testSave()
+    {
+        [$inner, $state, $now] = [$this->inner, $this->state, $this->now];
+
+        $state->acquire($now->modify('-1 hour'));
+        $state->save($now, 3);
+
+        $this->assertSame($now, $state->time());
+        $this->assertSame(3, $state->index());
+        $this->assertSame($inner->time(), $state->time());
+        $this->assertSame($inner->index(), $state->index());
+    }
+
+    public function testInitStateOnFirstAcquiring()
+    {
+        [$lock, $state, $now] = [$this->lock, $this->state, $this->now];
+
+        $this->assertTrue($state->acquire($now));
+        $this->assertEquals($now, $state->time());
+        $this->assertEquals(-1, $state->index());
+        $this->assertTrue($lock->isAcquired());
+    }
+
+    public function testLoadStateOnAcquiring()
+    {
+        [$lock, $inner, $state, $now] = [$this->lock, $this->inner, $this->state, $this->now];
+
+        $inner->save($now, 0);
+
+        $this->assertTrue($state->acquire($now->modify('1 min')));
+        $this->assertEquals($now, $state->time());
+        $this->assertEquals(0, $state->index());
+        $this->assertTrue($lock->isAcquired());
+    }
+
+    public function testCannotAcquereIfLocked()
+    {
+        [$state, $now] = [$this->state, $this->now];
+
+        $this->concurrentLock();
+
+        $this->assertFalse($state->acquire($now));
+    }
+
+    public function testResetStateAfterLockedAcquiring()
+    {
+        [$lock, $inner, $state, $now] = [$this->lock, $this->inner, $this->state, $this->now];
+
+        $concurrentLock = $this->concurrentLock();
+        $inner->save($now->modify('-2 min'), 0);
+        $state->acquire($now->modify('-1 min'));
+        $concurrentLock->release();
+
+        $this->assertTrue($state->acquire($now));
+        $this->assertEquals($now, $state->time());
+        $this->assertEquals(-1, $state->index());
+        $this->assertTrue($lock->isAcquired());
+        $this->assertFalse($concurrentLock->isAcquired());
+    }
+
+    public function testKeepLock()
+    {
+        [$lock, $state, $now] = [$this->lock, $this->state, $this->now];
+
+        $state->acquire($now->modify('-1 min'));
+        $state->release($now, $now->modify('1 min'));
+
+        $this->assertTrue($lock->isAcquired());
+    }
+
+    public function testReleaseLock()
+    {
+        [$lock, $state, $now] = [$this->lock, $this->state, $this->now];
+
+        $state->acquire($now->modify('-1 min'));
+        $state->release($now, null);
+
+        $this->assertFalse($lock->isAcquired());
+    }
+
+    public function testRefreshLock()
+    {
+        $lock = $this->createMock(LockInterface::class);
+        $lock->method('acquire')->willReturn(true);
+        $lock->method('getRemainingLifetime')->willReturn(120.0);
+        $lock->expects($this->once())->method('refresh')->with(120.0 + 60.0);
+        $lock->expects($this->never())->method('release');
+
+        $state = new LockStateDecorator(new State(), $lock);
+        $now = $this->now;
+
+        $state->acquire($now->modify('-10 sec'));
+        $state->release($now, $now->modify('60 sec'));
+    }
+
+    public function testFullCycle()
+    {
+        [$lock, $inner, $state, $now] = [$this->lock, $this->inner, $this->state, $this->now];
+
+        // init
+        $inner->save($now->modify('-1 min'), 3);
+
+        // action
+        $acquired = $state->acquire($now);
+        $lastTime = $state->time();
+        $lastIndex = $state->index();
+        $state->save($now, 0);
+        $state->release($now, null);
+
+        // asserting
+        $this->assertTrue($acquired);
+        $this->assertEquals($now->modify('-1 min'), $lastTime);
+        $this->assertSame(3, $lastIndex);
+        $this->assertEquals($now, $inner->time());
+        $this->assertSame(0, $inner->index());
+        $this->assertFalse($lock->isAcquired());
+    }
+
+    // No need to unlock after test, because the `InMemoryStore` is deleted
+    private function concurrentLock(): Lock
+    {
+        $lock = new Lock(
+            key: new Key('lock'),
+            store: $this->store,
+            autoRelease: false
+        );
+
+        if (!$lock->acquire()) {
+            throw new \LogicException('Already locked.');
+        }
+
+        return $lock;
+    }
+}

Tests/State/StateFactoryTest.php

@@ -0,0 +1,176 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\State;
+
+use PHPUnit\Framework\TestCase;
+use Psr\Container\ContainerInterface;
+use Symfony\Component\Lock\LockFactory;
+use Symfony\Component\Lock\Store\InMemoryStore;
+use Symfony\Component\Scheduler\Exception\LogicException;
+use Symfony\Component\Scheduler\State\CacheStateDecorator;
+use Symfony\Component\Scheduler\State\LockStateDecorator;
+use Symfony\Component\Scheduler\State\State;
+use Symfony\Component\Scheduler\State\StateFactory;
+use Symfony\Contracts\Cache\CacheInterface;
+
+class StateFactoryTest extends TestCase
+{
+    public function testCreateSimple()
+    {
+        $factory = new StateFactory(
+            $this->makeContainer([]),
+            $this->makeContainer([])
+        );
+
+        $expected = new State();
+
+        $this->assertEquals($expected, $factory->create('name', []));
+    }
+
+    public function testCreateWithCache()
+    {
+        $cache = $this->createMock(CacheInterface::class);
+        $cache->method('get')->willReturnCallback(fn ($key, \Closure $f) => $f());
+
+        $factory = new StateFactory(
+            $this->makeContainer([]),
+            $this->makeContainer(['app' => $cache]),
+        );
+
+        $state = new State();
+        $expected = new CacheStateDecorator($state, $cache, 'messenger.schedule.name');
+
+        $this->assertEquals($expected, $factory->create('name', ['cache' => 'app']));
+    }
+
+    public function testCreateWithCacheAndLock()
+    {
+        $cache = $this->createMock(CacheInterface::class);
+        $cache->method('get')->willReturnCallback(fn ($key, \Closure $f) => $f());
+
+        $lockFactory = new LockFactory(new InMemoryStore());
+
+        $factory = new StateFactory(
+            $this->makeContainer(['unlock' => $lockFactory]),
+            $this->makeContainer(['app' => $cache]),
+        );
+
+        $lock = $lockFactory->createLock($name = 'messenger.schedule.name');
+        $state = new State();
+        $state = new LockStateDecorator($state, $lock);
+        $expected = new CacheStateDecorator($state, $cache, $name);
+
+        $this->assertEquals($expected, $factory->create('name', ['cache' => 'app', 'lock' => 'unlock']));
+    }
+
+    public function testCreateWithConfiguredLock()
+    {
+        $cache = $this->createMock(CacheInterface::class);
+        $cache->method('get')->willReturnCallback(fn ($key, \Closure $f) => $f());
+
+        $lockFactory = new LockFactory(new InMemoryStore());
+
+        $factory = new StateFactory(
+            $this->makeContainer(['unlock' => $lockFactory]),
+            $this->makeContainer([]),
+        );
+
+        $lock = $lockFactory->createLock('messenger.schedule.name', $ttl = 77.7, false);
+        $state = new State();
+        $expected = new LockStateDecorator($state, $lock);
+
+        $cfg = [
+            'resource' => 'unlock',
+            'ttl' => $ttl,
+            'auto_release' => false,
+        ];
+        $this->assertEquals($expected, $factory->create('name', ['lock' => $cfg]));
+    }
+
+    public function testInvalidCacheName()
+    {
+        $factory = new StateFactory(
+            $this->makeContainer([]),
+            $this->makeContainer([])
+        );
+
+        $this->expectException(LogicException::class);
+        $this->expectExceptionMessage('The cache pool "wrong-cache" does not exist.');
+
+        $factory->create('name', ['cache' => 'wrong-cache']);
+    }
+
+    public function testInvalidLockName()
+    {
+        $factory = new StateFactory(
+            $this->makeContainer([]),
+            $this->makeContainer([])
+        );
+
+        $this->expectException(LogicException::class);
+        $this->expectExceptionMessage('The lock resource "wrong-lock" does not exist.');
+
+        $factory->create('name', ['lock' => 'wrong-lock']);
+    }
+
+    public function testInvalidConfiguredLockName()
+    {
+        $factory = new StateFactory(
+            $this->makeContainer([]),
+            $this->makeContainer([])
+        );
+
+        $this->expectException(LogicException::class);
+        $this->expectExceptionMessage('The lock resource "wrong-lock" does not exist.');
+
+        $factory->create('name', ['lock' => ['resource' => 'wrong-lock']]);
+    }
+
+    public function testInvalidCacheOption()
+    {
+        $factory = new StateFactory(
+            $this->makeContainer([]),
+            $this->makeContainer([]),
+        );
+
+        $this->expectException(LogicException::class);
+        $this->expectExceptionMessage('Invalid cache configuration for "default" schedule.');
+        $factory->create('default', ['cache' => true]);
+    }
+
+    public function testInvalidLockOption()
+    {
+        $factory = new StateFactory(
+            $this->makeContainer([]),
+            $this->makeContainer([]),
+        );
+
+        $this->expectException(LogicException::class);
+        $this->expectExceptionMessage('Invalid lock configuration for "default" schedule.');
+        $factory->create('default', ['lock' => true]);
+    }
+
+    private function makeContainer(array $services): ContainerInterface|\ArrayObject
+    {
+        return new class($services) extends \ArrayObject implements ContainerInterface {
+            public function get(string $id): mixed
+            {
+                return $this->offsetGet($id);
+            }
+
+            public function has(string $id): bool
+            {
+                return $this->offsetExists($id);
+            }
+        };
+    }
+}

Tests/State/StateTest.php

@@ -0,0 +1,36 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\State;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Scheduler\State\State;
+
+class StateTest extends TestCase
+{
+    public function testState()
+    {
+        $now = new \DateTimeImmutable('2020-02-20 20:20:20Z');
+        $later = $now->modify('1 hour');
+        $state = new State();
+
+        $this->assertTrue($state->acquire($now));
+        $this->assertSame($now, $state->time());
+        $this->assertSame(-1, $state->index());
+
+        $state->save($later, 7);
+
+        $this->assertSame($later, $state->time());
+        $this->assertSame(7, $state->index());
+
+        $state->release($later, null);
+    }
+}

Tests/Trigger/ExcludeTimeTriggerTest.php

@@ -0,0 +1,38 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\Trigger;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Scheduler\Trigger\ExcludeTimeTrigger;
+use Symfony\Component\Scheduler\Trigger\TriggerInterface;
+
+class ExcludeTimeTriggerTest extends TestCase
+{
+    public function testGetNextRun()
+    {
+        $inner = $this->createMock(TriggerInterface::class);
+        $inner
+            ->method('nextTo')
+            ->willReturnCallback(static fn (\DateTimeImmutable $d) => $d->modify('+1 sec'));
+
+        $scheduled = new ExcludeTimeTrigger(
+            $inner,
+            new \DateTimeImmutable('2020-02-20T02:02:02Z'),
+            new \DateTimeImmutable('2020-02-20T20:20:20Z')
+        );
+
+        $this->assertEquals(new \DateTimeImmutable('2020-02-20T02:02:01Z'), $scheduled->nextTo(new \DateTimeImmutable('2020-02-20T02:02:00Z')));
+        $this->assertEquals(new \DateTimeImmutable('2020-02-20T20:20:21Z'), $scheduled->nextTo(new \DateTimeImmutable('2020-02-20T02:02:02Z')));
+        $this->assertEquals(new \DateTimeImmutable('2020-02-20T20:20:21Z'), $scheduled->nextTo(new \DateTimeImmutable('2020-02-20T20:20:20Z')));
+        $this->assertEquals(new \DateTimeImmutable('2020-02-20T22:22:23Z'), $scheduled->nextTo(new \DateTimeImmutable('2020-02-20T22:22:22Z')));
+    }
+}

Tests/Trigger/OnceTriggerTest.php

@@ -0,0 +1,29 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\Trigger;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Scheduler\Trigger\OnceTrigger;
+
+class OnceTriggerTest extends TestCase
+{
+    public function testNextTo()
+    {
+        $time = new \DateTimeImmutable('2020-02-20 20:00:00');
+        $schedule = new OnceTrigger($time);
+
+        $this->assertEquals($time, $schedule->nextTo(new \DateTimeImmutable('@0'), ''));
+        $this->assertEquals($time, $schedule->nextTo($time->modify('-1 sec'), ''));
+        $this->assertNull($schedule->nextTo($time, ''));
+        $this->assertNull($schedule->nextTo($time->modify('+1 sec'), ''));
+    }
+}

Tests/Trigger/PeriodicalTriggerTest.php

@@ -0,0 +1,140 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Tests\Trigger;
+
+use PHPUnit\Framework\TestCase;
+use Symfony\Component\Scheduler\Exception\InvalidArgumentException;
+use Symfony\Component\Scheduler\Trigger\PeriodicalTrigger;
+
+class PeriodicalTriggerTest extends TestCase
+{
+    public function providerNextTo(): iterable
+    {
+        $periodicalMessage = new PeriodicalTrigger(
+            600,
+            new \DateTimeImmutable('2020-02-20T02:00:00+02'),
+            new \DateTimeImmutable('2020-02-20T03:00:00+02')
+        );
+
+        yield [
+            $periodicalMessage,
+            new \DateTimeImmutable('@0'),
+            new \DateTimeImmutable('2020-02-20T02:00:00+02'),
+        ];
+        yield [
+            $periodicalMessage,
+            new \DateTimeImmutable('2020-02-20T01:59:59.999999+02'),
+            new \DateTimeImmutable('2020-02-20T02:00:00+02'),
+        ];
+        yield [
+            $periodicalMessage,
+            new \DateTimeImmutable('2020-02-20T02:00:00+02'),
+            new \DateTimeImmutable('2020-02-20T02:10:00+02'),
+        ];
+        yield [
+            $periodicalMessage,
+            new \DateTimeImmutable('2020-02-20T02:05:00+02'),
+            new \DateTimeImmutable('2020-02-20T02:10:00+02'),
+        ];
+        yield [
+            $periodicalMessage,
+            new \DateTimeImmutable('2020-02-20T02:49:59.999999+02'),
+            new \DateTimeImmutable('2020-02-20T02:50:00+02'),
+        ];
+        yield [
+            $periodicalMessage,
+            new \DateTimeImmutable('2020-02-20T02:50:00+02'),
+            null,
+        ];
+        yield [
+            $periodicalMessage,
+            new \DateTimeImmutable('2020-02-20T03:00:00+02'),
+            null,
+        ];
+
+        $periodicalMessage = new PeriodicalTrigger(
+            600,
+            new \DateTimeImmutable('2020-02-20T02:00:00Z'),
+            new \DateTimeImmutable('2020-02-20T03:01:00Z')
+        );
+
+        yield [
+            $periodicalMessage,
+            new \DateTimeImmutable('2020-02-20T02:59:59.999999Z'),
+            new \DateTimeImmutable('2020-02-20T03:00:00Z'),
+        ];
+        yield [
+            $periodicalMessage,
+            new \DateTimeImmutable('2020-02-20T03:00:00Z'),
+            null,
+        ];
+    }
+
+    /**
+     * @dataProvider providerNextTo
+     */
+    public function testNextTo(PeriodicalTrigger $periodicalMessage, \DateTimeImmutable $lastRun, ?\DateTimeImmutable $expected)
+    {
+        $this->assertEquals($expected, $periodicalMessage->nextTo($lastRun));
+    }
+
+    public function testConstructors()
+    {
+        $firstRun = new \DateTimeImmutable($now = '2222-02-22');
+        $priorTo = new \DateTimeImmutable($farFuture = '3000-01-01');
+        $day = new \DateInterval('P1D');
+
+        $message = new PeriodicalTrigger(86400, $firstRun, $priorTo);
+
+        $this->assertEquals($message, PeriodicalTrigger::create(86400, $firstRun, $priorTo));
+        $this->assertEquals($message, PeriodicalTrigger::create('86400', $firstRun, $priorTo));
+        $this->assertEquals($message, PeriodicalTrigger::create('P1D', $firstRun, $priorTo));
+        $this->assertEquals($message, PeriodicalTrigger::create($day, $now, $farFuture));
+        $this->assertEquals($message, PeriodicalTrigger::create($day, $now));
+
+        $this->assertEquals($message, PeriodicalTrigger::fromPeriod(new \DatePeriod($firstRun, $day, $priorTo)));
+        $this->assertEquals($message, PeriodicalTrigger::fromPeriod(new \DatePeriod($firstRun->sub($day), $day, $priorTo, \DatePeriod::EXCLUDE_START_DATE)));
+        $this->assertEquals($message, PeriodicalTrigger::fromPeriod(new \DatePeriod($firstRun, $day, 284107)));
+        $this->assertEquals($message, PeriodicalTrigger::fromPeriod(new \DatePeriod($firstRun->sub($day), $day, 284108, \DatePeriod::EXCLUDE_START_DATE)));
+    }
+
+    public function testTooBigInterval()
+    {
+        $this->expectException(InvalidArgumentException::class);
+        $this->expectExceptionMessage('The interval for a periodical message is too big');
+
+        PeriodicalTrigger::create(\PHP_INT_MAX.'0', new \DateTimeImmutable('2002-02-02'));
+    }
+
+    public function getInvalidIntervals(): iterable
+    {
+        yield ['wrong'];
+        yield ['3600.5'];
+        yield [0];
+        yield [-3600];
+    }
+
+    /**
+     * @dataProvider getInvalidIntervals
+     */
+    public function testInvalidInterval($interval)
+    {
+        $this->expectException(InvalidArgumentException::class);
+        PeriodicalTrigger::create($interval, $now = new \DateTimeImmutable(), $now->modify('1 day'));
+    }
+
+    public function testNegativeInterval()
+    {
+        $this->expectException(InvalidArgumentException::class);
+        PeriodicalTrigger::create('wrong', $now = new \DateTimeImmutable(), $now->modify('1 day'));
+    }
+}

Trigger/ExcludeTimeTrigger.php

@@ -0,0 +1,32 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Trigger;
+
+final class ExcludeTimeTrigger implements TriggerInterface
+{
+    public function __construct(
+        private readonly TriggerInterface $inner,
+        private readonly \DateTimeImmutable $from,
+        private readonly \DateTimeImmutable $to,
+    ) {
+    }
+
+    public function nextTo(\DateTimeImmutable $run): \DateTimeImmutable
+    {
+        $nextRun = $this->inner->nextTo($run);
+        if ($nextRun >= $this->from && $nextRun <= $this->to) {
+            return $this->inner->nextTo($this->to);
+        }
+
+        return $nextRun;
+    }
+}

Trigger/OnceTrigger.php

@@ -0,0 +1,25 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Trigger;
+
+final class OnceTrigger implements TriggerInterface
+{
+    public function __construct(
+        private readonly \DateTimeImmutable $time,
+    ) {
+    }
+
+    public function nextTo(\DateTimeImmutable $run): ?\DateTimeImmutable
+    {
+        return $run < $this->time ? $this->time : null;
+    }
+}

Trigger/PeriodicalTrigger.php

@@ -0,0 +1,112 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Trigger;
+
+use Symfony\Component\Scheduler\Exception\InvalidArgumentException;
+
+final class PeriodicalTrigger implements TriggerInterface
+{
+    public function __construct(
+        private readonly int $intervalInSeconds,
+        private readonly \DateTimeImmutable $firstRun,
+        private readonly \DateTimeImmutable $priorTo,
+    ) {
+        if (0 >= $this->intervalInSeconds) {
+            throw new InvalidArgumentException('The `$intervalInSeconds` argument must be greater then zero.');
+        }
+    }
+
+    public static function create(
+        string|int|\DateInterval $interval,
+        string|\DateTimeImmutable $firstRun,
+        string|\DateTimeImmutable $priorTo = new \DateTimeImmutable('3000-01-01'),
+    ): self {
+        if (\is_string($firstRun)) {
+            $firstRun = new \DateTimeImmutable($firstRun);
+        }
+        if (\is_string($priorTo)) {
+            $priorTo = new \DateTimeImmutable($priorTo);
+        }
+        if (\is_string($interval)) {
+            if ('P' === $interval[0]) {
+                $interval = new \DateInterval($interval);
+            } elseif (ctype_digit($interval)) {
+                self::ensureIntervalSize($interval);
+                $interval = (int) $interval;
+            } else {
+                throw new InvalidArgumentException(sprintf('The interval "%s" for a periodical message is invalid.', $interval));
+            }
+        }
+        if (!\is_int($interval)) {
+            $interval = self::calcInterval($firstRun, $firstRun->add($interval));
+        }
+
+        return new self($interval, $firstRun, $priorTo);
+    }
+
+    public static function fromPeriod(\DatePeriod $period): self
+    {
+        $startDate = \DateTimeImmutable::createFromInterface($period->getStartDate());
+        $nextDate = $startDate->add($period->getDateInterval());
+        $firstRun = $period->include_start_date ? $startDate : $nextDate;
+        $interval = self::calcInterval($startDate, $nextDate);
+
+        $priorTo = $period->getEndDate()
+            ? \DateTimeImmutable::createFromInterface($period->getEndDate())
+            : $startDate->modify($period->getRecurrences() * $interval.' seconds');
+
+        return new self($interval, $firstRun, $priorTo);
+    }
+
+    private static function calcInterval(\DateTimeImmutable $from, \DateTimeImmutable $to): int
+    {
+        if (8 <= \PHP_INT_SIZE) {
+            return $to->getTimestamp() - $from->getTimestamp();
+        }
+
+        // @codeCoverageIgnoreStart
+        $interval = $to->format('U') - $from->format('U');
+        self::ensureIntervalSize(abs($interval));
+
+        return (int) $interval;
+        // @codeCoverageIgnoreEnd
+    }
+
+    private static function ensureIntervalSize(string|float $interval): void
+    {
+        if ($interval > \PHP_INT_MAX) {
+            throw new InvalidArgumentException('The interval for a periodical message is too big. If you need to run it once, use `$priorTo` argument.');
+        }
+    }
+
+    public function nextTo(\DateTimeImmutable $run): ?\DateTimeImmutable
+    {
+        if ($this->firstRun > $run) {
+            return $this->firstRun;
+        }
+        if ($this->priorTo <= $run) {
+            return null;
+        }
+
+        $delta = (float) $run->format('U.u') - (float) $this->firstRun->format('U.u');
+        $recurrencesPassed = (int) ($delta / $this->intervalInSeconds);
+        $nextRunTimestamp = ($recurrencesPassed + 1) * $this->intervalInSeconds + $this->firstRun->getTimestamp();
+        /** @var \DateTimeImmutable $nextRun */
+        $nextRun = \DateTimeImmutable::createFromFormat('U.u', $nextRunTimestamp.$this->firstRun->format('.u'));
+
+        if ($this->priorTo <= $nextRun) {
+            return null;
+        }
+
+        return $nextRun;
+    }
+}

Trigger/TriggerInterface.php

@@ -0,0 +1,17 @@
+<?php
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Scheduler\Trigger;
+
+interface TriggerInterface
+{
+    public function nextTo(\DateTimeImmutable $run): ?\DateTimeImmutable;
+}

composer.json

@@ -0,0 +1,42 @@
+{
+    "name": "symfony/scheduler",
+    "type": "library",
+    "description": "Provides basic scheduling through the Symfony Messenger",
+    "keywords": ["scheduler", "schedule", "cron"],
+    "homepage": "https://symfony.com",
+    "license": "MIT",
+    "authors": [
+        {
+            "name": "Sergey Rabochiy",
+            "email": "upyx.00@gmail.com"
+        },
+        {
+            "name": "Symfony Community",
+            "homepage": "https://symfony.com/contributors"
+        }
+    ],
+    "require": {
+        "php": ">=8.1",
+        "psr/clock-implementation": "^1.0",
+        "psr/container-implementation": "^1.1|^2.0",
+        "symfony/cache-implementation": "^1.1|^2.0|^3.0",
+        "symfony/messenger": "^5.4|^6.0"
+    },
+    "require-dev": {
+        "symfony/cache": "^5.4|^6.0",
+        "symfony/clock": "^6.2",
+        "symfony/dependency-injection": "^5.4|^6.0",
+        "symfony/lock": "^5.4|^6.0"
+    },
+    "suggest": {
+        "symfony/cache": "For saving state between restarts.",
+        "symfony/lock": "For preventing workers race."
+    },
+    "autoload": {
+        "psr-4": { "Symfony\\Component\\Scheduler\\": "" },
+        "exclude-from-classmap": [
+            "/Tests/"
+        ]
+    },
+    "minimum-stability": "dev"
+}

phpunit.xml.dist

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+
+<phpunit xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:noNamespaceSchemaLocation="http://schema.phpunit.de/9.3/phpunit.xsd"
+         backupGlobals="false"
+         colors="true"
+         bootstrap="vendor/autoload.php"
+         failOnRisky="true"
+         failOnWarning="true"
+>
+    <php>
+        <ini name="error_reporting" value="-1" />
+    </php>
+
+    <testsuites>
+        <testsuite name="Symfony Scheduler Component Test Suite">
+            <directory>./Tests/</directory>
+        </testsuite>
+    </testsuites>
+
+    <coverage>
+        <include>
+            <directory>./</directory>
+        </include>
+        <exclude>
+            <directory>./Tests</directory>
+            <directory>./vendor</directory>
+        </exclude>
+    </coverage>
+</phpunit>