diff --git a/app/Jobs/SynchAllAttendeesStatus.php b/app/Jobs/SynchAllAttendeesStatus.php index d5a4792d9..b0fa8c9c3 100644 --- a/app/Jobs/SynchAllAttendeesStatus.php +++ b/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; /** diff --git a/app/Services/Model/AttendeeService.php b/app/Services/Model/AttendeeService.php index b0dee3d57..bac370f56 100644 --- a/app/Services/Model/AttendeeService.php +++ b/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)); diff --git a/tests/AttendeeServiceTest.php b/tests/AttendeeServiceTest.php index 7ff1a67cf..c2ece3289 100644 --- a/tests/AttendeeServiceTest.php +++ b/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);