Fix stateful scheduler

symfony · fabpot · Sep 16, 2023 · Sep 13, 2023 · Sep 15, 2023 · 2d5856b6bee3f80444b7db49bfcdb981ee6edbb8
commit 2d5856b6bee3f80444b7db49bfcdb981ee6edbb8
@@ -10,6 +10,7 @@ CHANGELOG
 * Add `AbstractTriggerDecorator`
 * Make `ScheduledStamp` "send-able"
 * Add `ScheduledStamp` to `RedispatchMessage`
+ * Add `from()` to `CheckpointInterface`

 6.3
 ---

@@ -16,6 +16,7 @@

 final class Checkpoint implements CheckpointInterface
 {
+    private \DateTimeImmutable $from;
    private \DateTimeImmutable $time;
    private int $index = -1;
    private bool $reset = false;
@@ -41,14 +42,22 @@ public function acquire(\DateTimeImmutable $now): bool
            $this->save($now, -1);
        }

-        $this->time ??= $now;
        if ($this->cache) {
-            $this->save(...$this->cache->get($this->name, fn () => [$now, -1]));
+            [$this->time, $this->index, $this->from] = $this->cache->get($this->name, fn () => [$now, -1, $now]) + [2 => $now];
+            $this->save($this->time, $this->index);
        }

+        $this->time ??= $now;
+        $this->from ??= $now;
+
        return true;
    }

+    public function from(): \DateTimeImmutable
+    {
+        return $this->from;
+    }
+
    public function time(): \DateTimeImmutable
    {
        return $this->time;
@@ -63,7 +72,8 @@ public function save(\DateTimeImmutable $time, int $index): void
    {
        $this->time = $time;
        $this->index = $index;
-        $this->cache?->get($this->name, fn () => [$time, $index], \INF);
+        $this->from ??= $time;
+        $this->cache?->get($this->name, fn () => [$time, $index, $this->from], \INF);
    }

    /**

@@ -15,6 +15,8 @@ interface CheckpointInterface
 {
    public function acquire(\DateTimeImmutable $now): bool;

+    public function from(): \DateTimeImmutable;
+
    public function time(): \DateTimeImmutable;

    public function index(): int;

@@ -16,6 +16,7 @@
 use Symfony\Component\Scheduler\RecurringMessage;
 use Symfony\Component\Scheduler\Schedule;
 use Symfony\Component\Scheduler\ScheduleProviderInterface;
+use Symfony\Component\Scheduler\Trigger\StatefulTriggerInterface;

 final class MessageGenerator implements MessageGeneratorInterface
 {
@@ -43,9 +44,10 @@ public function getMessages(): \Generator
            return;
        }

+        $startTime = $checkpoint->from();
        $lastTime = $checkpoint->time();
        $lastIndex = $checkpoint->index();
-        $heap = $this->heap($lastTime);
+        $heap = $this->heap($lastTime, $startTime);

        while (!$heap->isEmpty() && $heap->top()[0] <= $now) {
            /** @var \DateTimeImmutable $time */
@@ -79,7 +81,7 @@ public function getMessages(): \Generator
        $checkpoint->release($now, $this->waitUntil);
    }

-    private function heap(\DateTimeImmutable $time): TriggerHeap
+    private function heap(\DateTimeImmutable $time, \DateTimeImmutable $startTime): TriggerHeap
    {
        if (isset($this->triggerHeap) && $this->triggerHeap->time <= $time) {
            return $this->triggerHeap;
@@ -88,7 +90,13 @@ private function heap(\DateTimeImmutable $time): TriggerHeap
        $heap = new TriggerHeap($time);

        foreach ($this->schedule()->getRecurringMessages() as $index => $recurringMessage) {
-            if (!$nextTime = $recurringMessage->getTrigger()->getNextRunDate($time)) {
+            $trigger  = $recurringMessage->getTrigger();
+
+            if ($trigger instanceof StatefulTriggerInterface) {
+                $trigger->continue($startTime);
+            }
+
+            if (!$nextTime = $trigger->getNextRunDate($time)) {
                continue;
            }


@@ -40,7 +40,7 @@ private function __construct(
     * @see https://en.wikipedia.org/wiki/ISO_8601#Durations
     * @see https://php.net/datetime.formats.relative
     */
-    public static function every(string|int|\DateInterval $frequency, object $message, string|\DateTimeImmutable $from = new \DateTimeImmutable(), string|\DateTimeImmutable $until = new \DateTimeImmutable('3000-01-01')): self
+    public static function every(string|int|\DateInterval $frequency, object $message, string|\DateTimeImmutable|null $from = null, string|\DateTimeImmutable $until = new \DateTimeImmutable('3000-01-01')): self
    {
        return new self(new PeriodicalTrigger($frequency, $from, $until), $message);
    }

@@ -48,7 +48,7 @@ public function testWithStateInitStateOnFirstAcquiring()
        $this->assertTrue($checkpoint->acquire($now));
        $this->assertEquals($now, $checkpoint->time());
        $this->assertEquals(-1, $checkpoint->index());
-        $this->assertEquals([$now, -1], $cache->get('cache', fn () => []));
+        $this->assertEquals([$now, -1, $now], $cache->get('cache', fn () => []));
    }

    public function testWithStateLoadStateOnAcquiring()
@@ -58,10 +58,10 @@ public function testWithStateLoadStateOnAcquiring()

        $cache->get('cache', fn () => [$now, 0], \INF);

-        $this->assertTrue($checkpoint->acquire($now->modify('1 min')));
+        $this->assertTrue($checkpoint->acquire($startedAt = $now->modify('1 min')));
        $this->assertEquals($now, $checkpoint->time());
        $this->assertEquals(0, $checkpoint->index());
-        $this->assertEquals([$now, 0], $cache->get('cache', fn () => []));
+        $this->assertEquals([$now, 0, $startedAt], $cache->get('cache', fn () => []));
    }

    public function testWithLockInitStateOnFirstAcquiring()
@@ -72,11 +72,12 @@ public function testWithLockInitStateOnFirstAcquiring()

        $this->assertTrue($checkpoint->acquire($now));
        $this->assertEquals($now, $checkpoint->time());
+        $this->assertEquals($now, $checkpoint->from());
        $this->assertEquals(-1, $checkpoint->index());
        $this->assertTrue($lock->isAcquired());
    }

-    public function testwithLockLoadStateOnAcquiring()
+    public function testWithLockLoadStateOnAcquiring()
    {
        $lock = new Lock(new Key('lock'), new InMemoryStore());
        $checkpoint = new Checkpoint('dummy', $lock);
@@ -86,6 +87,7 @@ public function testwithLockLoadStateOnAcquiring()

        $this->assertTrue($checkpoint->acquire($now->modify('1 min')));
        $this->assertEquals($now, $checkpoint->time());
+        $this->assertEquals($now, $checkpoint->from());
        $this->assertEquals(0, $checkpoint->index());
        $this->assertTrue($lock->isAcquired());
    }
@@ -105,12 +107,13 @@ public function testWithCacheSave()
    {
        $checkpoint = new Checkpoint('cache', new NoLock(), $cache = new ArrayAdapter());
        $now = new \DateTimeImmutable('2020-02-20 20:20:20Z');
-        $checkpoint->acquire($now->modify('-1 hour'));
+        $checkpoint->acquire($startedAt = $now->modify('-1 hour'));
        $checkpoint->save($now, 3);

        $this->assertSame($now, $checkpoint->time());
        $this->assertSame(3, $checkpoint->index());
-        $this->assertEquals([$now, 3], $cache->get('cache', fn () => []));
+        $this->assertSame($startedAt, $checkpoint->from());
+        $this->assertEquals([$now, 3, $startedAt], $cache->get('cache', fn () => []));
    }

    public function testWithLockSave()
@@ -119,11 +122,12 @@ public function testWithLockSave()
        $checkpoint = new Checkpoint('dummy', $lock);
        $now = new \DateTimeImmutable('2020-02-20 20:20:20Z');

-        $checkpoint->acquire($now->modify('-1 hour'));
+        $checkpoint->acquire($startTime = $now->modify('-1 hour'));
        $checkpoint->save($now, 3);

        $this->assertSame($now, $checkpoint->time());
        $this->assertSame(3, $checkpoint->index());
+        $this->assertSame($startTime, $checkpoint->from());
    }

    public function testWithLockAndCacheSave()
@@ -132,12 +136,12 @@ public function testWithLockAndCacheSave()
        $checkpoint = new Checkpoint('dummy', $lock, $cache = new ArrayAdapter());
        $now = new \DateTimeImmutable('2020-02-20 20:20:20Z');

-        $checkpoint->acquire($now->modify('-1 hour'));
+        $checkpoint->acquire($startTime = $now->modify('-1 hour'));
        $checkpoint->save($now, 3);

        $this->assertSame($now, $checkpoint->time());
        $this->assertSame(3, $checkpoint->index());
-        $this->assertEquals([$now, 3], $cache->get('dummy', fn () => []));
+        $this->assertEquals([$now, 3, $startTime], $cache->get('dummy', fn () => []));
    }

    public function testWithCacheFullCycle()
@@ -161,7 +165,7 @@ public function testWithCacheFullCycle()
        $this->assertSame(3, $lastIndex);
        $this->assertEquals($now, $checkpoint->time());
        $this->assertSame(0, $checkpoint->index());
-        $this->assertEquals([$now, 0], $cache->get('cache', fn () => []));
+        $this->assertEquals([$now, 0, $now], $cache->get('cache', fn () => []));
    }

    public function testWithLockResetStateAfterLockedAcquiring()

@@ -184,7 +184,7 @@ public static function providerGetNextRunDateAgain(): iterable
        yield [
            $trigger,
            new \DateTimeImmutable('2020-02-20T01:59:00+02:00'),
-            new \DateTimeImmutable('2020-02-20T02:09:00+02:00'),
+            new \DateTimeImmutable('2020-02-20T02:00:00+02:00'),
        ];
        yield [
            $trigger,

@@ -14,12 +14,19 @@
 /**
 * @author Kevin Bond <kevinbond@gmail.com>
 */
-abstract class AbstractDecoratedTrigger implements TriggerInterface
+abstract class AbstractDecoratedTrigger implements StatefulTriggerInterface
 {
    public function __construct(private TriggerInterface $inner)
    {
    }

+    public function continue(\DateTimeImmutable $startedAt): void
+    {
+        if ($this->inner instanceof StatefulTriggerInterface) {
+            $this->inner->continue($startedAt);
+        }
+    }
+
    final public function inner(): TriggerInterface
    {
        $inner = $this->inner;

@@ -13,17 +13,18 @@

 use Symfony\Component\Scheduler\Exception\InvalidArgumentException;

-class PeriodicalTrigger implements TriggerInterface
+class PeriodicalTrigger implements StatefulTriggerInterface
 {
    private float $intervalInSeconds = 0.0;
-    private \DateTimeImmutable $from;
+    private ?\DateTimeImmutable $from;
    private \DateTimeImmutable $until;
    private \DatePeriod $period;
    private string $description;
+    private string|int|float|\DateInterval $interval;

    public function __construct(
        string|int|float|\DateInterval $interval,
-        string|\DateTimeImmutable $from = new \DateTimeImmutable(),
+        string|\DateTimeImmutable|null $from = null,
        string|\DateTimeImmutable $until = new \DateTimeImmutable('3000-01-01'),
    ) {
        $this->from = \is_string($from) ? new \DateTimeImmutable($from) : $from;
@@ -70,7 +71,7 @@ public function __construct(
                    $this->description = sprintf('every %s seconds', $this->intervalInSeconds);
                }
            } else {
-                $this->period = new \DatePeriod($this->from, $i, $this->until);
+                $this->interval = $i;
            }
        } catch (\Exception $e) {
            throw new InvalidArgumentException(sprintf('Invalid interval "%s": ', $interval instanceof \DateInterval ? 'instance of \DateInterval' : $interval).$e->getMessage(), 0, $e);
@@ -82,15 +83,22 @@ public function __toString(): string
        return $this->description;
    }

+    public function continue(\DateTimeImmutable $startedAt): void
+    {
+        $this->from ??= $startedAt;
+    }
+
    public function getNextRunDate(\DateTimeImmutable $run): ?\DateTimeImmutable
    {
+        $this->from ??= $run;
+
        if ($this->intervalInSeconds) {
            if ($this->until <= $run) {
                return null;
            }

-            $fromDate = min($this->from, $run);
-            $from = $fromDate->format('U.u');
+            $fromDate = $this->from;
+            $from = (float) $fromDate->format('U.u');
            $delta = $run->format('U.u') - $from;
            $recurrencesPassed = floor($delta / $this->intervalInSeconds);
            $nextRunTimestamp = sprintf('%.6F', ($recurrencesPassed + 1) * $this->intervalInSeconds + $from);
@@ -103,6 +111,7 @@ public function getNextRunDate(\DateTimeImmutable $run): ?\DateTimeImmutable
            return $this->until > $nextRun ? $nextRun : null;
        }

+        $this->period ??= new \DatePeriod($this->from, $this->interval, $this->until);
        $iterator = $this->period->getIterator();
        while ($run >= $next = $iterator->current()) {
            $iterator->next();
@@ -130,6 +139,6 @@ private function canBeConvertedToSeconds(\DateInterval $interval): bool

    private function calcInterval(\DateInterval $interval): float
    {
-        return $this->from->setTimestamp(0)->add($interval)->format('U.u');
+        return (float) (new \DateTimeImmutable('@0'))->add($interval)->format('U.u');
    }
 }
@@ -0,0 +1,8 @@
+<?php
+
+namespace Symfony\Component\Scheduler\Trigger;
+
+interface StatefulTriggerInterface extends TriggerInterface
+{
+    public function continue(\DateTimeImmutable $startedAt): void;
+}
-Original file line number
+Diff line change
@@ -15,6 +15,8 @@ interface CheckpointInterface
     {
         public function acquire(\DateTimeImmutable $now): bool;
+        public function from(): \DateTimeImmutable;
                   fabpot marked this conversation as resolved.
              

          
            Show resolved
            Hide resolved
        

                
  
            
  

  
    
  
    
      

              Uh oh!

              
There was an error while loading. Please reload this page.
         public function time(): \DateTimeImmutable;
         public function index(): int;
-          Expand Down