Out-of-band activities, the Scheduler component

Transcript

1. Out of band activities Fabien Potencier
2. Out-of-band activities Like sending an email When doing something that should not be 
 in the HTTP request/response hot-path
3. Handle the message out-of-band $bus->dispatch(new SendEmailMessage($email)); Moving activities out-of-band $ symfony console messenger:consume async A message identifies the task 
 to be done and contains 
 the task's context Message handling can happen asynchronously
4. class SomeMessage { public function __construct(public string $message) { } } #[AsMessageHandler] class SomeHandler { public function __invoke(SomeMessage $message) { error_log($message->message); } } Handling messages The handling logic 
 is wrapped in 
 another service A serializable 
 data object
5. Handling messages This pattern is also useful 
 as an abstraction layer 
 even without async processing $bus->dispatch(new SignupMessage($user)); The business logic is 
 abstracted in the handler
6. Async but handled almost immediately $ symfony console messenger:consume async 
 For quick activities, messages are handled 
 almost simultaneously as the "main" flow Messages are handled as they come in
7. Delaying handling is an option $msg = new EndOfTrialMessage($userId); End of trial email shouldn't be sent now... Envelopes add metadata to the message $endOfTrial = (new \DateTimeImmutable())->modify('+30 days'); $msg = Envelope::wrap($msg)->with(DelayStamp::delayUntil($endOfTrial)); ... but scheduled in 30 days
8. Message as data objects class EndOfTrialMessage { public function __construct(public int $userId) { } } #[AsMessageHandler] class EndOfTrialMessageHandler { public function __invoke(EndOfTrialMessage $message) { // Get the User from DB and send them an email error_log('Processing '.$message->userId); } }
9. Detour: Quickly testing message handling $transport = new InMemoryTransport(); $bus = new MessageBus([ new SendMessageMiddleware( new SendersLocator([ EndOfTrialMessage::class => ['mem'], ], new class(['mem' => fn () => $transport]) implements ContainerInterface { use ServiceLocatorTrait; }, , ), new HandleMessageMiddleware(new HandlersLocator([ EndOfTrialMessage::class => [new EndOfTrialMessageHandler()], ])), ]); A non-persistent 
 transport Message/transport wiring Message/handler wiring
10. Detour of the detour $transport = new InMemoryTransport(); DelayStamp support as of 6.3 Implemented just for this keynote https://github.com/symfony/symfony/pull/49725
11. Detour: The simplest possible Container $transport = new InMemoryTransport(); $bus = new MessageBus([ new SendMessageMiddleware( new SendersLocator([ EndOfTrialMessage::class => ['mem'], ], new class(['mem' => fn () => $transport]) implements ContainerInterface { use ServiceLocatorTrait; }, , ), new HandleMessageMiddleware(new HandlersLocator([ EndOfTrialMessage::class => [new EndOfTrialMessageHandler()], ])), ]); PSR Container Implements the interface as an array of callables
12. Detour: Quickly testing message handling $bus->dispatch($msg); error_log('Nothing happens before running the workers'); $receivers = ['mem' => $transport]; $worker = new Worker($receivers, $bus); $worker->run(); Dispatch messages Process incoming messages forever Check it actually works
13. Clean-up ended trials? $cleanupDate = (new \DateTimeImmutable())->modify('+30 days'); $msg = Envelope::wrap($msg)->with(DelayStamp::delayUntil($cleanupDate)); Is it the best way? 1. Should discard converted trials 2. No need to act precisely at +30 days 3. More e ff i cient as batch
14. Recurring messages A message that should be handled 
 over and over again, on a pre-defined schedule
15. Scheduler Landed in 6.3 
 as an experimental component
16. Recurring messages RecurringMessage::every('1 week', new EndOfTrialMessage()) $schedule = (new Schedule())->add($msg); A trigger that defines 
 the frequency A message to handle 
 periodically A schedule describes the config 
 of one or more recurring messages
17. Running the scheduler $scheduler = new Scheduler( [EndOfTrialMessage::class => new EndOfTrialMessageHandler()], [$schedule], ); $scheduler->run(); Runs forever and will handle messages as per the trigger definition Only useful when using 
 the component standalone
18. Reusing Messenger infrastructure? 1. Learn it fast as it uses the same Messenger concepts 
 Messages, handlers, envelopes, stamps, ... 2. The scheduler can be integrated with Messenger via a special transport 
 Triggers and message generators 3. Scheduler uses the same Messenger worker: messenger:consume 
 Time limit, memory management, signal handling, ... The beauty of the new component ❤
19. Detour: Beautiful contributions Credits to Sergey Rabochiy ❤
20. The Scheduler component in the context of a Symfony full-stack app
21. Register a Schedule provider #[AsSchedule('default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function getSchedule(): Schedule { return (new Schedule()) ->add(RecurringMessage::every('1 week', new EndOfTrialMessage())) ; } } This automatically wires the Scheduler with Messenger Like before 1
22. Run the Messenger consumer #[AsSchedule('default')] symfony console messenger:consume -v scheduler_default The name is used for the receiver name Equivalent to registering scheduler://default 2
23. Enjoy! 3
24. Triggers: "every()" Frequency as text RecurringMessage::every('10 seconds', $msg) RecurringMessage::every('2 weeks', $msg) RecurringMessage::every('1 day', $msg)
25. Triggers: "every()" RecurringMessage::every('next tuesday', $msg) RecurringMessage::every('last day of next month', $msg) RecurringMessage::every('first monday of next month', $msg) Relative date format https://php.net/datetime.formats.relative
26. Triggers: Quickly testing $msg = RecurringMessage::every('next tuesday', new stdClass()); $i = 0; $next = new \DateTimeImmutable(); while ($i++ < 20) { $next = $msg->getTrigger()->getNextRunDate($next); if (!$next) { break; } echo $next->format('Y-m-d H:i:s')."\n"; } That's what makes the Scheduler tick!
27. Triggers: "every()" Time of the day RecurringMessage::every('1 day', $msg, from: '13:47') 2023-03-20 13:47:00 2023-03-21 13:47:00 2023-03-22 13:47:00 2023-03-23 13:47:00 2023-03-24 13:47:00 2023-03-25 13:47:00
28. Triggers: "every()" Timezone RecurringMessage::every('1 day', $msg, from: '13:47+0400') 2023-03-20 13:47:00 +04:00 2023-03-21 13:47:00 +04:00 2023-03-22 13:47:00 +04:00 2023-03-23 13:47:00 +04:00 2023-03-24 13:47:00 +04:00 2023-03-25 13:47:00 +04:00
29. Triggers: "every()" Can be a DateTimeImmutable $from = new \DateTimeImmutable('13:47', new \DateTimeZone('Europe/Paris')); $msg = RecurringMessage::every('1 day', new stdClass(), from: $from); 2023-03-20 13:47:00 Europe/Paris 2023-03-21 13:47:00 Europe/Paris 2023-03-22 13:47:00 Europe/Paris 2023-03-23 13:47:00 Europe/Paris 2023-03-24 13:47:00 Europe/Paris 2023-03-25 13:47:00 Europe/Paris
30. Triggers: "every()" End date RecurringMessage::every('1 day', new stdClass(), until: '2023-03-23'); 2023-03-20 13:45:00 2023-03-21 13:45:00 2023-03-22 13:45:00 ---
31. Triggers: CronExpression RecurringMessage::cron('* * * * *', $msg) ⚠ Max frequency is 1 minute
32. Triggers: Your choice RecurringMessage::trigger(TriggerInterface, $msg) interface TriggerInterface { public function getNextRunDate(\DateTimeImmutable $run): ?\DateTimeImmutable; }
33. Triggers: Be different 😭 Midnight On top of the hour At any "round number" Add some randomness to your times 🤩 Use 12:13 or 11:54 
 not 12:00 Prefer :17/:23/:47 
 not :15/:25/:45
34. // jitter on any trigger (add up to X seconds) $schedule->add(RecurringMessage::every('1 hour', $msg)->withJitter(60)); // Hashed cron expressions: sometimes between midnight and 7PM $schedule->add(RecurringMessage::cron('# #(0-7) * * *', $msg)); Triggers: Add some randomness
35. Each Schedule is a Messenger transport No stateful "transport" 
 Messages are handled immediately when generated (like the sync transport) Messages are never "dispatched" 
 They are generated by the transport directly Messages are handled one by one 
 For all recurring messages configured on a schedule Scale by using di ff erent workers for di ff erence schedules 
 Or use more than one worker for a schedule
36. Multiple schedules No guarantee that a message is executed right 
 on time or at the required frequency 
 It's up to you to configure schedules properly Create more than one schedule to con fi gure them di ff erently 
 You can then scale each worker and schedule independently
37. Messages are handled synchronously one by one EndOfTrialCleanupMessage 2023-03-24 13:10 EndOfTrialCleanupMessage EndOfTrialCleanupMessage 2023-03-24 13:00 handled Next message will be handled 
 just after the current one A message that takes 
 a lot of time to handle 10 minutes 10 minutes Recurring message EndOfTrialCleanupMessage 2023-03-24 13:20 EndOfTrialCleanupMessage handled The backlog keeps growing 
 until the worker is restarted
38. Rule of thumb 
 Con fi gure a frequency that is greater 
 than the average time 
 it takes to handle a message
39. messenger:consume downtime? SignupEmailMessage 
 for User 1337 queued Messenger worker down SignupEmailMessage 
 for User 1337 handled SignupEmailMessage 
 for User 42 queued SignupEmailMessage 
 for User 42 handled "Regular" message Another message
40. messenger:consume downtime? Messenger worker down EndOfTrialCleanupMessage generated EndOfTrialCleanupMessage handled not generated EndOfTrialCleanupMessage EndOfTrialCleanupMessage EndOfTrialCleanupMessage handled generated 10 minutes 10 minutes Recurring message 2023-03-24 13:00 2023-03-25 13:10 2023-03-26 13:20
41. Stateful schedules #[AsSchedule('default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function __construct( private CacheInterface $cache, ) { } public function getSchedule(): Schedule { $msg = new EndOfTrialMessage(); return (new Schedule()) ->add(RecurringMessage::every('2 seconds', $msg)) ->stateful($this->cache) ; } } When missing messages 
 is not an option, 
 Stateful schedules will catch up.
42. messenger:consume downtime? Messenger worker down EndOfTrialCleanupMessage generated EndOfTrialCleanupMessage handled EndOfTrialCleanupMessage EndOfTrialCleanupMessage handled generated Recurring message generated EndOfTrialCleanupMessage handled Stateful schedules catch up 10 minutes 10 minutes 2023-03-24 13:00 2023-03-25 13:10 EndOfTrialCleanupMessage 2023-03-26 13: 20
43. Many messenger:consume in //? EndOfTrialCleanupMessage EndOfTrialCleanupMessage EndOfTrialCleanupMessage 2023-03-24 13:10 Scaling by running 
 2 workers running at the same Worker 1 Worker 2 handled EndOfTrialCleanupMessage handled Oops, the message 
 is generated 
 and handled 
 twice! 2023-03-24 13:10
44. Locking schedules #[AsSchedule('default')] class DefaultScheduleProvider implements ScheduleProviderInterface { public function __construct( private CacheInterface $cache, private LockFactory $lockFactory, ) { } public function getSchedule(): Schedule { $msg = new EndOfTrialMessage(); return (new Schedule()) ->add(RecurringMessage::every('2 seconds', $msg)) ->stateful($this->cache) ->lock($this->lockFactory->createLock('default-scheduler')) ; } } Don't use a lock without a cache
45. Many messenger:consume in //? EndOfTrialCleanupMessage EndOfTrialCleanupMessage EndOfTrialCleanupMessage Scaling by running 
 2 workers running at the same Worker 1 Worker 2 handled not generated The lock prevents the message to be generated multiple times 2023-03-24 13:10 2023-03-24 13:10
46. Frequent but slow message handling? EndOfTrialCleanupMessage EndOfTrialCleanupMessage 2023-03-24 13:00 handled A message that takes 
 a lot of time to execute EndOfTrialCleanupMessage 2023-03-24 13:10 not generated The lock prevents the next message to be generated as another is still handled Worker 1 Worker 2
47. Send to another Messenger transport $msg = new RedispatchMessage(new UpdateDocMessage(), 'async'); Message will be handled asynchronously ⚠ Lock is only partially supported The same message is probably not generated twice But several messages can be handled in //
48. #[AsMessageHandler()] class UpdateDocMessageHandler { private LockInterface $lock; public function __construct( LockFactory $lockFactory, ) { $this->lock = $lockFactory->createLock('update-doc'); } public function __invoke(UpdateDocMessage $message) { if (!$this->lock->acquire(false)) { echo "Doc is already being generated, skip...\n"; return; } try { // do something } finally { $this->lock->release(); } } } Locking schedules Handling the lock in the handler 
 works great!
49. Recap For a schedule No cache No lock Cache Cache Lock On a single 
 worker Catch up missed messages when downtime Not needed On several 
 workers Not recommended Not recommended Avoid messages to be executed several times
50. Cron vs Scheduler? 1. Cron can only run "executables" think Symfony Console commands 2. Cron is yet another moving part to monitor 3. Cron max frequency limited to one minute 4. Cron needs tooling for reporting (errors, ...) 5. Cron does not support multiple "workers", locks, states, ... Cron is still probably best 
 for any recurring command not defined in Symfony
51. https://symfony.com/sponsor Sponsor Symfony Thank you!