Task scheduling can be boring, but not with Symfony Scheduler

In the dynamic landscape of modern web application development, achieving accurate and timely task execution is pivotal for improving user experiences and optimizing backend processes. By enabling applications to run processes at optimal intervals, task scheduling reduces the need for manual intervention and promotes overall operational efficiency.

We will explore the capabilities of the Symfony Scheduler through a concrete use case that focuses on creating and maintaining a recurring task that is constrained by a predetermined schedule until a particular event happens. In this talk, we'll see how the Symfony Scheduler would have effortlessly enhanced the implementation of iterative automation in our case using some useful concepts from Symfony Messenger. We'll also take the opportunity to look at how Symfony Scheduler would have benefited our application, and what issues it might have raised.

Ready to be amazed by this captivating component? Let’s dive in.

Transcript

1. but not with Symfony Scheduler! Task scheduling can be boring…
2. Allison Guilhem ➔ Lead Developer at Les-Tilleuls.coop ➔ Contributor to Symfony/API Platform @Alli_g83 Allison E.Guilhem @alli83
3. Task Scheduling and cronJob 01 02 03 What is Symfony Scheduler? Symfony Scheduler via a use case scenario OUTLINE
4. Task Scheduling ➔ Technique for automating tasks based on a schedule ➔ Repeated regularly over time, or even indefinitely ➔ Such tasks may include backing-up databases, processing a queue, or creating system usage reports etc…
5. Task Scheduling in PHP ➔ Unix world: cronjob (as a tool coupled with a Symfony command, for example) ➔ Third-Party Scheduling Services in conjunction with AWS Lambda etc… ➔ Various bundles: for instance, to help you schedule via Cronjob/worker within your Symfony application: Guikingone/SchedulerBundle Cron/Symfony-Bundle zenstruck/schedule-bundle lavary/crunz
6. CronJob: Not an ideal situation ➔ Unsupported TimeZone specification (Kubernetes cronjob) ➔ The minimum time resolution is one minute ➔ CronJob can trigger concurrent jobs, regardless of options or previous completion ➔ CronJobs lack built-in handling for missed job executions ➔ Etc… * * * * * command-to-be-executed 45 5 15 * * command-to-be-executed 30 9 * * 1-5 command-to-be-executed 0 0 * * * command-to-be-executed
7. CronJob: Tools to overcome their limitations ➔ Anacron: ensures scheduled tasks run, even when the system is offline or idle ➔ Celery: Distributed task queue for scheduling and managing tasks in a distributed system, and it can be used alongside Cron ➔ Task Spooler (ts): Unix batch system for managing queued jobs, complementing Cron for task queuing ➔ Custom Scripts: in order to implement other features that Cron lacks ➔ Etc…
8. Does Symfony have a mechanism to automate tasks?
9. And what about Symfony Messenger?
10. Symfony Messenger (full stack application): quick reminders Dispatch Sync mode Async m ode Get from Dispatch Message MessageBus Handler Transport Worker
11. Symfony Messenger concepts: Quick reminders ➔ Message: Any PHP object that can be serialized ➔ Stamps: Metadata added to the envelope in which the message to be processed is inserted ➔ Bus: Where messages are dispatched ➔ Handlers: Service in charge of processing a type of message ➔ Transport: Generate the messages and makes them available
12. Symfony Messenger and the DelayStamp ➔ Supported by Symfony messenger transport (Doctrine / Redis / in memory transport etc…) ➔ It allows the message to be delayed for a certain amount of time before it's delivered ➔ But it's not a way to schedule and automate a repetitive task over a period of time
13. A need in Symfony? … A new feature is never too far away!
14. Task Scheduling and cronJob 01 02 03 What is Symfony Scheduler? Symfony Scheduler via a use case scenario OUTLINE
15. Symfony Scheduler component
16. Symfony Scheduler and the Debug command
17. Symfony Scheduler in a full stack application From Symfony Messenger to Symfony scheduler, it's only a short step... well, almost!
18. Symfony Scheduler in a full stack application: The basic principle Dispatch Sync mode Async m ode Get from Dispatch Message MessageBus Handler Scheduler Transport Worker GENERATE MESSAGES
19. how are they generated? But what are these messages?
20. Built-in cron expression trigger Symfony Scheduler: what is a Recurring Message? Messages associated with a trigger Via custom trigger Built-in PeriodicalTrigger via different frequency formats: string | int | \DateInterval
21. Recurring Message and Trigger RecurringMessage::every('10 seconds', new Message()) RecurringMessage::every('3 weeks', new Message()) RecurringMessage::every(‘first Monday of next month', new Message()) RecurringMessage::every(‘first Monday of next month', new Message(), $from, $until) RecurringMessage::cron(‘*****’, new Message()) RecurringMessage::trigger(your trigger, new Message()) Implementing TriggerInterface Also possible to add and define a timezone
22. Registered in a schedule And with the substantial message(s) wrapped in a MessageProviderInterface * WHY? Messages associated with a trigger Symfony Scheduler: what is a Recurring Message?
23. A dynamic vision for the messages generated
24. final class CallbackMessageProvider implements MessageProviderInterface { private \Closure $callback; /** * @param callable(MessageContext): iterable<object> $callback */ public function __construct(callable $callback, private string $id = '') { $this->callback = $callback(...); } public function getMessages(MessageContext $context): iterable { return ($this->callback)($context); } … } A dynamic vision for the messages generated via CallbackMessageProvider
25. Recurring Message and Trigger RecurringMessage::every( '9 seconds’, new CallbackMessageProvider( [$this,’generateDynamically'], ‘foo’ ) )
26. Recurring Message via an Attribute: another way #[AsPeriodicTask(frequency: 1, jitter: 1, arguments: [‘It works well!’])] class SomethingSimple { public function __invoke(string $message): void { echo "$message\n"; } }
27. Recurring Message via an Attribute: another way #[AsCronTask('* * * * *', arguments: 'test -v')] #[AsCommand(‘app:to-do')] class DoStuffCommand extends Command { … }
28. Recurring Message via an Attribute: another way Coupled with AsCommand Attribute Towards their respective handlers: RunCommandMessageHandler / ServiceCallbackMessageHandler AsCronTask / AsPeriodicTask ServiceCallMessage RunCommandMessage Will execute the method mentioned or __invoke Will run the command according to the options defined
29. When are they generated? Here comes the MessageGenerator Photo de Zdenek Machacek MessageGenerator For each message Check and yield Extract Build a RecurringMessage stack from the schedule or reused one previously created Scheduler Transport Calls
30. Everything is a question of timing! ➔ Possible to scale with multiple schedule/ worker ➔ Possible to keep track when the server is down : stateful ➔ Possible to add a lock and better handled concurrent jobs if multiple workers are involved #[AsSchedule(‘default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function __construct(…){} public function getSchedule(): Schedule { return (new Schedule()) ->with(RecurringMessage::every('5 seconds’), new Message()) ->stateful($this->cache) ->lock($this ->lockFactory->createLock( ‘my-lock’ ) messenger:consume scheduler_default
31. A Symfony and not a cacophony with cache and lock MyRecurringMessage MyRecurringMessage 10 min MyRecurringMessage 10 min MyRecurringMessage Worker1 Worker2 If lock: it won't be generated If no lock: it will be generated and it will be a duplicate 2:25 pm 2:35 pm 2:45 pm Pay close attention to how long it will take to process the message, otherwise all subsequent processing
32. Everything is a question of timing! #[AsSchedule(‘default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function __construct(…){} public function getSchedule(): Schedule { return (new Schedule()) ->with(RecurringMessage::every('5 seconds’), new RedispatchMessage(new Message(), ‘async’))) messenger:consume scheduler_default ➔ Possibility to specify that our message needs to be handled by a specified transport ➔ When the message is distributed, it will be intercepted by the RedispatchMessageHandler, which will redispatch it to the designated transport. ➔ Very practical for heavy tasks
33. But the Schedule still lack of dynamism…
34. A dynamic vision needed for the Schedule Photo de Zdenek Machacek For each message MessageGenerator WE NEED A MECHANISM TO FORCE THE STACK GENERATION AGAIN IF NEEDED. Check and yield Extract Build a stack of recurring messages from the schedule or reused one previously created
35. A dynamic vision needed for the Schedule
36. A dynamic vision needed for the Schedule final class Schedule implements ScheduleProviderInterface { … public function clear(): static { $this->messages = []; $this->setRestart(true); return $this; } }
37. A dynamic vision needed for the Schedule final class MessageGenerator implements MessageGeneratorInterface { public function getMessages(): \Generator { $checkpoint = $this->checkpoint(); if ($this->schedule?->shouldRestart()) { unset($this->triggerHeap); $this->waitUntil = new \DateTimeImmutable('@0'); $this->schedule->setRestart(false); } … } }
38. A dynamic vision needed for the Schedule final class MessageGenerator implements MessageGeneratorInterface { … private function heap(\DateTimeImmutable $time, \DateTimeImmutable $startTime): TriggerHeap { if (isset($this->triggerHeap) && $this->triggerHeap->time <= $time) { return $this->triggerHeap; } $heap = new TriggerHeap($time); foreach ($this->getSchedule()->getRecurringMessages() as $index => $recurringMessage) { … } } }
39. And now… a use case scenario
40. Task Scheduling and cronJob 01 02 03 What is Symfony Scheduler? Symfony Scheduler via a use case scenario OUTLINE
41. From a use case perspective ➔ Requests to send to third parties according to their sector ➔ Symfony Messenger seems like the way to go
42. Yes, but no… ➔ New requirements emerge • Requests to send to third parties on a regular basis until someone agrees • Schedule the cleaning of incoming requests according to a certain frequency
43. A Schedule with Recurring Messages #[AsSchedule('default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function getSchedule(): Schedule { return $this->schedule ??= (new Schedule()) ->with(RecurringMessage::every(‘6 hours', new CallbackMessageProvider([$this, 'gatherRequestsBySector'], 'foo'))) }) } public function gatherRequestsBySector(MessageContext $context) { // for example research all requests on a same sector dynamically yield new RequestsBySectorMessage([..], 'A1'); yield new RequestsBySectorMessage([..], 'A2'); } }
44. A Schedule with Recurring Messages #[AsSchedule('default')] class DefaultScheduleProvider implements ScheduleProviderInterface { … public function getSchedule(): Schedule { $this->cleanUpMessage = RecurringMessage::every('3 hours', new CleanUpMessage()); return $this->schedule ??= (new Schedule()) … ->add($this->cleanUpMessage) … }
45. As a reminder… Sync mode Get from Dispatch for each m essage MessageBus Handler Scheduler Transport Worker 1 2 3 Generate Messages
46. A dynamic schedule needed ➔ RequestsBySectorMessage case • If a request is accepted: be able to stop the generation of the message • Under certain circumstances: be able to stop the schedule of this recurring message ➔ CleanUpMessage case • Once the job is done: need to stop the schedule of this job
47. Stop the generation of cleanUpMessage #[AsMessageHandler] class CleanUpMessageHandler { public function __invoke(CleanUpMessage $message) { // do what you have to do if ($isFinished) { // access our schedule $this->mySchedule->removeCleanUpMessage(); } } }
48. #[AsSchedule('default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function getSchedule(): Schedule { … } public function removeCleanUpMessage() { $this->getSchedule()->getSchedule()->remove($this->cleanUpMessage); } } Will trigger setRestart => reset the stack of recurring messages Stop the generation of cleanUpMessage
49. A dynamic schedule Wouldn't be better, in certain situation, to separate the verification that a message should be processed from its processing?
50. A dynamic schedule Message x handled External Event causing a message to be deprogrammed How to avoid passing messages to handlers that should not be processed? Thanks to the events
51. Symfony Scheduler and events
52. Symfony Scheduler and events ➔ Before/after/onFailure PRE_RUN_EVENT - POST_RUN_EVENT - FAILURE_EVENT ➔ Need to have access to our schedule add or remove a type of recurring Message ➔ One of the objectives is to be able to update our schedule and to call the setRestart() method
53. Symfony Scheduler and events Worker Dispatch several events during the cycle WorkerMessageReceivedEvent Listeners listening to these events WorkerMessageHandledEvent WorkerMessageFailedEvent Dispatch corresponding SchedulerEvents
54. PreRunEvent particularity: can cancel a message class PreRunEvent { private bool $shouldCancel = false; public function __construct( private readonly ScheduleProviderInterface $schedule, private readonly MessageContext $messageContext, private readonly object $message, ) { … } public function shouldCancel(bool $shouldCancel = null): bool { if (null !== $shouldCancel) { $this->shouldCancel = $shouldCancel; } return $this->shouldCancel; } }
55. Once upon a time, a simple system of dispatch via Symfony Messenger #[AsSchedule('default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function getSchedule(): Schedule { return $this->schedule ??= (new Schedule($this->eventDispatcher)) ->with(RecurringMessage::every('6 hours', new CallbackMessageProvider([$this, 'gatherRequestsBySector'], 'foo'))) }) } ->before(function(PreRunEvent $event) { // Do what you want }
56. Once upon a time, a simple system of dispatch via Symfony Messenger #[AsSchedule('default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function getSchedule(): Schedule { … ->before(function(PreRunEvent $event) { $message = $event->getMessage(); $messageContext = $event->getMessageContext(); // can access the recurring message if needed $recurringMessages = $event->getSchedule()->getSchedule() ->getRecurringMessages(); //or target directly the one being processed $event->getSchedule()->getSchedule()->removeById($messageContext->id); //Allow to call the ShouldCancel() of the WorkerMessageReceivedEvent and avoid the message to be handled $event->shouldCancel(true); //do what you want } We can access : - Message, - MessageContext, - Schedule
57. Symfony Scheduler and events: In case of failure ➔ No built-in retry system in Symfony Scheduler ➔ FailureEvent ➔ Access to the Schedule / message / messageContext #[AsSchedule('default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function getSchedule(): Schedule { return (new Schedule()) ->with(…) ->OnFailure( function(FailureEvent $event) { // Do what you want }) } }
58. "I don't know what your task is, I don't know what you need. But if you are looking for efficient scheduling, I can tell you I am the Symfony Scheduler component. But what I do have are a very particular set of skills; skills I have acquired thanks to a great community. If you don't use me now, that'll be too bad. But if you do, I will help you explore options, I will help you find the best solution and I will make your scheduling nightmares disappear… " Symfony Scheduler has Taken control of time!
59. Thank you! Any questions?