Address CodeRabbit review: make batchConnect's rebuild atomic

Follow-up to 1515cb2: the lock-contention branch was reporting success
(true) even when it never ran a rebuild, so hasConnectedListeners()
could still be false right after Staff::login() called batchConnect().
More importantly, even the happy path could let a concurrent reader
observe a partially rebuilt listener set, since connect() commits each
row with its own flush() as the rebuild loop runs.

Wraps the rebuild (_disconnectUnavailable() plus the connect() loop)
in a single DB transaction via Connection::transactional(), so
hasConnectedListeners() can only ever observe the previous complete
set or the new complete set, never one row of a half-finished rebuild.
The lock-timeout branch now returns false instead of true, so a
caller that depends on the return value can tell initialization
didn't happen.

Staff::login() now retries once if the first batchConnect() call
reports failure, rather than firing the login event against a set we
know is incomplete. A single bounded retry, not a wait/poll loop -
GET_LOCK already blocks for its own timeout before failing, and
retrying indefinitely would risk turning this housekeeping step into
a login-availability problem.

Updated Hook's 'batch connects' test to stub Connection::transactional
(now required for the rebuild to run) and renamed/adjusted the
lock-contention test for the corrected false return. Added a Staff
test asserting login() retries batchConnect() once and still succeeds
when the second attempt connects the listeners.

src/modules/Hook/Service.php

@@ -125,46 +125,55 @@ class Service implements InjectionAwareInterface
     /**
      * Serializes batchConnect() runs so two concurrent callers (e.g. overlapping cron and
      * on-demand rebuilds triggered by a login) cannot interleave connect()'s check-then-insert
-     * and create duplicate listener rows, or leave a partially rebuilt set visible in between.
+     * and create duplicate listener rows. The rebuild itself runs in a single DB transaction,
+     * so hasConnectedListeners() can only ever observe the previous complete set or the new
+     * complete set, never a partially rebuilt one.
+     *
+     * Returns false if the lock could not be acquired within the timeout, so a failure to
+     * initialize is never mistaken for success - callers that need listeners connected before
+     * proceeding should check the return value rather than assume this always succeeds.
      */
     public function batchConnect($mod_name = null): bool
     {
         $connection = $this->di['em']->getConnection();
         if ((int) $connection->fetchOne('SELECT GET_LOCK(:name, 5)', ['name' => self::BATCH_CONNECT_LOCK]) !== 1) {
-            // Another process is already rebuilding the listener set; let it finish rather
-            // than racing it.
-            return true;
+            // Another process is already rebuilding the listener set and holding the lock
+            // past our wait. Report failure rather than claiming a rebuild we didn't run or
+            // wait for actually completed.
+            return false;
         }
 
         try {
-            // Clean up the existing list before we add to it
-            $this->_disconnectUnavailable();
-            $extensionService = $this->di['mod_service']('extension');
+            $connection->transactional(function () use ($mod_name): void {
+                // Clean up the existing list before we add to it
+                $this->_disconnectUnavailable();
+                $extensionService = $this->di['mod_service']('extension');
 
-            $mods = [];
-            if ($mod_name !== null) {
-                $mods[] = $mod_name;
-            } else {
-                $mods = $extensionService->getCoreAndActiveModules();
-            }
-
-            foreach ($mods as $m) {
-                $installed = $this->getExtensionRepository()->existsActiveByTypeAndName(Extension::TYPE_MOD, $m);
-                if (!$installed && !$extensionService->isCoreModule($m)) {
-                    continue;
+                $mods = [];
+                if ($mod_name !== null) {
+                    $mods[] = $mod_name;
+                } else {
+                    $mods = $extensionService->getCoreAndActiveModules();
                 }
 
-                $mod = $this->di['mod']($m);
-                if ($mod->hasService()) {
-                    $class = $mod->getService();
-                    $reflector = new \ReflectionClass($class);
-                    foreach ($reflector->getMethods() as $method) {
-                        if ($this->canBeConnected($method)) {
-                            $this->connect(['event' => $method->getName(), 'mod' => $mod->getName()]);
+                foreach ($mods as $m) {
+                    $installed = $this->getExtensionRepository()->existsActiveByTypeAndName(Extension::TYPE_MOD, $m);
+                    if (!$installed && !$extensionService->isCoreModule($m)) {
+                        continue;
+                    }
+
+                    $mod = $this->di['mod']($m);
+                    if ($mod->hasService()) {
+                        $class = $mod->getService();
+                        $reflector = new \ReflectionClass($class);
+                        foreach ($reflector->getMethods() as $method) {
+                            if ($this->canBeConnected($method)) {
+                                $this->connect(['event' => $method->getName(), 'mod' => $mod->getName()]);
+                            }
                         }
                     }
                 }
-            }
+            });
 
             return true;
         } finally {

src/modules/Hook/tests/Unit/ServiceTest.php

@@ -155,6 +155,9 @@ test('batch connects', function (): void {
     $expectation1->andReturn(false);
     $connection->shouldReceive('executeStatement')
         ->byDefault();
+    $connection->shouldReceive('transactional')
+        ->atLeast()->once()
+        ->andReturnUsing(fn (Closure $callback) => $callback($connection));
 
     $returnArr = [
         [
@@ -210,13 +213,14 @@ test('batch connects', function (): void {
     expect($result)->toBeTrue();
 });
 
-test('batch connect defers to another process already holding the lock', function (): void {
+test('batch connect reports failure when another process holds the lock', function (): void {
     $service = new Box\Mod\Hook\Service();
 
     $connection = Mockery::mock(Doctrine\DBAL\Connection::class);
     $connection->shouldReceive('fetchOne')
         ->with(Mockery::on(fn ($sql) => str_contains((string) $sql, 'GET_LOCK')), Mockery::any())
         ->andReturn(0);
+    $connection->shouldNotReceive('transactional');
     $connection->shouldNotReceive('fetchAllAssociative');
     $connection->shouldNotReceive('executeStatement')
         ->with(Mockery::on(fn ($sql) => str_contains((string) $sql, 'RELEASE_LOCK')), Mockery::any());
@@ -227,5 +231,5 @@ test('batch connect defers to another process already holding the lock', functio
 
     $result = $service->batchConnect('activity');
 
-    expect($result)->toBeTrue();
+    expect($result)->toBeFalse();
 });

src/modules/Staff/Service.php

@@ -123,9 +123,11 @@ class Service implements InjectionAwareInterface
         // connected by the cron job's hook_batch_connect task. Before cron has run for the
         // first time, no listeners are connected and the event fired below would silently do
         // nothing, so an admin's very first logins would go unrecorded. Connect them now so
-        // that gap does not exist.
+        // that gap does not exist. batchConnect() returns false if another process was still
+        // rebuilding the set when it gave up waiting; retry once rather than firing the event
+        // below against a set we know is incomplete.
         $hookService = $this->di['mod_service']('hook');
-        if (!$hookService->hasConnectedListeners()) {
+        if (!$hookService->hasConnectedListeners() && !$hookService->batchConnect()) {
             $hookService->batchConnect();
         }
 

src/modules/Staff/tests/Unit/ServiceTest.php

@@ -186,6 +186,58 @@ test('login returns admin details on successful login', function (): void {
     expect($result)->toBe($expected);
 });
 
+test('login retries connecting event listeners once before firing the login event', function (): void {
+    $email = 'email@domain.com';
+    $password = 'pass';
+    $ip = '127.0.0.1';
+
+    $admin = \Tests\Helpers\admin(['id' => 1, 'email' => $email, 'name' => 'Admin', 'pass' => 'hashedPassword']);
+
+    $emMock = Mockery::mock('\Box_EventManager');
+    $emMock->shouldReceive('fire')->atLeast()->once()
+        ->andReturn(true);
+
+    $adminRepository = Mockery::mock(AdminRepository::class);
+    $adminRepository->shouldReceive('findOneByEmailAndActive')->atLeast()->once()
+        ->with($email)
+        ->andReturn($admin);
+
+    $sessionMock = Mockery::mock(FOSSBilling\Session::class);
+    $sessionMock->shouldReceive('regenerateId')->atLeast()->once();
+    $sessionMock->shouldReceive('set')->atLeast()->once();
+
+    $passwordMock = Mockery::mock(FOSSBilling\PasswordManager::class);
+    $passwordMock->shouldReceive('verify')->atLeast()->once()
+        ->with($password, $admin->getPass())
+        ->andReturn(true);
+    $passwordMock->shouldReceive('needsRehash')->atLeast()->once()
+        ->andReturn(false);
+
+    // First attempt fails, as if another process held the rebuild lock past its timeout.
+    $hookServiceMock = Mockery::mock(Box\Mod\Hook\Service::class);
+    $hookServiceMock->shouldReceive('hasConnectedListeners')->atLeast()->once()->andReturn(false);
+    $hookServiceMock->shouldReceive('batchConnect')->twice()->andReturn(false, true);
+
+    $di = container();
+    $di['events_manager'] = $emMock;
+    $di['em']->shouldReceive('getRepository')->with(Admin::class)->andReturn($adminRepository);
+    $di['session'] = $sessionMock;
+    $di['logger'] = new Tests\Helpers\TestLogger();
+    $di['password'] = $passwordMock;
+    $di['mod_service'] = $di->protect(fn (string $name = ''): object => strtolower($name) === 'hook' ? $hookServiceMock : Mockery::mock()->shouldIgnoreMissing());
+
+    $service = new Service();
+    $service->setDi($di);
+
+    $result = $service->login($email, $password, $ip);
+
+    expect($result)->toBe([
+        'id' => 1,
+        'email' => $email,
+        'name' => 'Admin',
+    ]);
+});
+
 test('login throws exception when credentials are invalid', function (): void {
     $email = 'email@domain.com';
     $password = 'pass';