fix(attendees): clear the EntityManager between pages and bound the timeout of SynchAllAttendeesStatus

app/Jobs/SynchAllAttendeesStatus.php

@@ -31,6 +31,9 @@ final class SynchAllAttendeesStatus implements ShouldQueue
 {
     public $tries = 1;
 
+    // 2 hours, bounded explicitly instead of relying on the worker default
+    public $timeout = 7200;
+
     use Dispatchable, InteractsWithQueue, Queueable, SerializesModels;
 
     /**

app/Services/Model/AttendeeService.php

@@ -24,6 +24,8 @@
 use Illuminate\Support\Facades\Cache;
 use Illuminate\Support\Facades\Config;
 use Illuminate\Support\Facades\Log;
+use LaravelDoctrine\ORM\Facades\Registry;
+use models\utils\SilverstripeBaseModel;
 use libs\utils\ITransactionService;
 use models\exceptions\EntityNotFoundException;
 use models\exceptions\ValidationException;
@@ -57,6 +59,11 @@
  */
 final class AttendeeService extends AbstractService implements IAttendeeService
 {
+    /**
+     * seconds, same bound as SynchAllAttendeesStatus::$timeout
+     */
+    const ResynchAttendeesStatusLockTTL = 7200;
+
     /**
      * @var ISummitAttendeeRepository
      */
@@ -1129,13 +1136,15 @@ public function resynchAttendeesStatusBySummit(int $summit_id): void
         Log::debug(sprintf("AttendeeService::resynchAttendeesStatusBySummit summit id %s", $summit_id));
         $key = sprintf("AttendeeService::resynchAttendeesStatusBySummit::%s", $summit_id);
 
-        if (Cache::has($key)) {
+        // owner-aware lock: the TTL lets a killed run (no finally) release it by itself, and release()
+        // in the finally only frees it if this run still owns it (a longer run can outlive the TTL)
+        $lock = Cache::lock($key, self::ResynchAttendeesStatusLockTTL);
+        if (!$lock->get()) {
             Log::debug(sprintf("AttendeeService::resynchAttendeesStatusBySummit summit id %s already running", $summit_id));
             return;
         }
 
         try {
-            Cache::add($key, $key);
             $summit = $this->summit_repository->getById($summit_id);
             if (!$summit instanceof Summit)
                 throw new EntityNotFoundException(sprintf("Summit %s not found", $summit_id));
@@ -1179,9 +1188,26 @@ public function resynchAttendeesStatusBySummit(int $summit_id): void
                     }
                 }
                 ++$current_page;
+
+                // release the identity map between pages; the manager is fetched from the registry on
+                // every page because DoctrineTransactionService replaces it after a failed transaction
+                $em = Registry::getManager(SilverstripeBaseModel::EntityManager);
+                if ($em->isOpen())
+                    $em->clear();
+
+                Log::debug
+                (
+                    sprintf
+                    (
+                        "AttendeeService::resynchAttendeesStatusBySummit summit id %s processed %s attendees so far, memory usage %s bytes",
+                        $summit_id,
+                        $count,
+                        memory_get_usage(true)
+                    )
+                );
             } while (true);
         } finally {
-            Cache::forget($key);
+            $lock->release();
         }
 
         Log::debug(sprintf("AttendeeService::resynchAttendeesStatusBySummit summit id %s processed %s attendees", $summit_id, $count));

tests/AttendeeServiceTest.php

@@ -15,8 +15,13 @@
 use App\Jobs\Emails\SummitAttendeeAllTicketsEditionEmail;
 use App\Jobs\Emails\SummitAttendeeRegistrationIncompleteReminderEmail;
 use App\Models\Foundation\Main\IGroup;
+use App\Services\Model\AttendeeService;
 use App\Services\Model\IAttendeeService;
+use Closure;
+use Illuminate\Contracts\Cache\Lock;
 use Illuminate\Support\Facades\App;
+use Illuminate\Support\Facades\Cache;
+use libs\utils\ITransactionService;
 use LaravelDoctrine\ORM\Facades\EntityManager;
 use models\summit\Summit;
 use models\summit\SummitAttendeeBadge;
@@ -54,6 +59,65 @@ public function testRedeemPromoCodes(){
         $service->updateRedeemedPromoCodes($summit);
     }
 
+    public function testResynchAttendeesStatusBySummitReleasesLock() {
+        $service = App::make(IAttendeeService::class);
+        $key = sprintf("AttendeeService::resynchAttendeesStatusBySummit::%s", self::$summit->getId());
+
+        $service->resynchAttendeesStatusBySummit(self::$summit->getId());
+
+        // the lock is free again: a new run can acquire it
+        $lock = Cache::lock($key, 60);
+        $this->assertTrue($lock->get());
+        $lock->release();
+    }
+
+    public function testResynchAttendeesStatusBySummitDoesNotReleaseAnotherRunsLock() {
+        $key = sprintf("AttendeeService::resynchAttendeesStatusBySummit::%s", self::$summit->getId());
+
+        // while the run is in progress (first transaction, lock already acquired), simulate its TTL
+        // expiring and a newer run taking the lock; the release() in the finally of the first run
+        // must leave the newer run's lock in place
+        $other = Cache::lock($key, 60, 'other-run');
+        $tx_service = new class(App::make(ITransactionService::class), $key, $other) implements ITransactionService {
+            public bool $taken = false;
+
+            public function __construct(private ITransactionService $inner, private string $key, private Lock $other) {}
+
+            public function transaction(Closure $callback, int $isolationLevel = 2)
+            {
+                if (!$this->taken) {
+                    Cache::lock($this->key)->forceRelease();
+                    $this->taken = $this->other->get();
+                }
+                return $this->inner->transaction($callback, $isolationLevel);
+            }
+        };
+        $service = App::make(AttendeeService::class, ['tx_service' => $tx_service]);
+
+        try {
+            $service->resynchAttendeesStatusBySummit(self::$summit->getId());
+            $this->assertTrue($tx_service->taken);
+            $this->assertTrue($other->isOwnedByCurrentProcess());
+        } finally {
+            $other->release();
+        }
+    }
+
+    public function testResynchAttendeesStatusBySummitSkipsWhenLocked() {
+        $service = App::make(IAttendeeService::class);
+        $key = sprintf("AttendeeService::resynchAttendeesStatusBySummit::%s", self::$summit->getId());
+        $lock = Cache::lock($key, 60);
+        $this->assertTrue($lock->get());
+
+        try {
+            $service->resynchAttendeesStatusBySummit(self::$summit->getId());
+            // the run that owns the lock must not be released by a skipped one
+            $this->assertTrue($lock->isOwnedByCurrentProcess());
+        } finally {
+            $lock->release();
+        }
+    }
+
     public function testSendAllAttendeeTickets() {
 
         $service = App::make(IAttendeeService::class);