Upgrade to Pro — share decks privately, control downloads, hide ads and more …

SymfonyLive Paris 2023: Scheduler

SymfonyLive Paris 2023: Scheduler

SymfonyLive Paris 2023: Scheduler

Fabien Potencier

March 24, 2023
Tweet

More Decks by Fabien Potencier

Other Decks in Programming

Transcript

  1. Out-of-band activities Like sending an email When doing something that

    should not be 
 in the HTTP request/response hot-path
  2. 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
  3. 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
  4. 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
  5. 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
  6. 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
  7. 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); } }
  8. 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
  9. 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
  10. 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
  11. 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
  12. 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
  13. Recurring messages A message that should be handled 
 over

    and over again, on a pre-defined schedule
  14. 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
  15. 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
  16. 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 ❤
  17. 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
  18. 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
  19. 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
  20. 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!
  21. 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
  22. 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
  23. 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
  24. 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 ---
  25. 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
  26. 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
  27. 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
  28. 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
  29. Rule of thumb 
 Con fi gure a frequency that

    is greater 
 than the average time 
 it takes to handle a message
  30. 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
  31. 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
  32. 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.
  33. 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
  34. 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
  35. 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
  36. 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
  37. 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
  38. 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 //
  39. #[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!
  40. Cron vs Scheduler? 1. Cron can only run "executables" think

    Symfony Console commands 2. Cron support on Windows? 3. Cron is yet another moving part to monitor 4. Cron max frequency limited to one minute 5. Cron needs tooling for reporting (errors, ...) 6. Cron does not support multiple "workers", locks, states, ... Cron is still probably best 
 for any recurring command not defined in Symfony
  41. 🙋 I need your help ‣ Test before 6.3 fi

    nal and provide feedback! ‣ ProcessMessage and ProcessMessageHandler ‣ RecurringRule support for triggers (I have a branch for that) ‣ debug:scheduler command ‣ ... Next Steps? The main concepts 
 and low-level infrastructure 
 are well-defined or don't complain :)
  42. 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