PHRAS-3447_file-lock-in-workers_4.1

fix : exclusive lock for a worker to work on a subdef file

lib/Alchemy/Phrasea/Model/Entities/WorkerRunningJob.php

@@ -6,7 +6,12 @@ use Doctrine\ORM\Mapping as ORM;
 use Gedmo\Mapping\Annotation as Gedmo;
 
 /**
- * @ORM\Table(name="WorkerRunningJob")
+ * @ORM\Table(
+ *     name="WorkerRunningJob",
+ *     uniqueConstraints={
+ *          @ORM\uniqueConstraint(name="flock", columns={"databox_id", "record_id", "flock"})
+ *      }
+ * )
  * @ORM\Entity(repositoryClass="Alchemy\Phrasea\Model\Repositories\WorkerRunningJobRepository")
  */
 class WorkerRunningJob
@@ -14,7 +19,7 @@ class WorkerRunningJob
     const FINISHED = 'finished';
     const RUNNING  = 'running';
     const ERROR    = 'error';
-    const INTERRUPT = 'interrupted manually';
+    const INTERRUPT = 'canceled';
 
     const ATTEMPT  = 'attempt ';
 
@@ -41,12 +46,17 @@ class WorkerRunningJob
     private $recordId;
 
     /**
-     * @ORM\Column(type="string", name="work", nullable=true)
+     * @ORM\Column(type="string", length=64, name="flock", nullable=true)
+     */
+    private $flock;
+
+    /**
+     * @ORM\Column(type="string", length=64, name="work", nullable=true)
      */
     private $work;
 
     /**
-     * @ORM\Column(type="string", name="work_on", nullable=true)
+     * @ORM\Column(type="string", length=64, name="work_on", nullable=true)
      */
     private $workOn;
 
@@ -138,6 +148,25 @@ class WorkerRunningJob
 
     }
 
+    /**
+     * @return mixed
+     */
+    public function getFlock()
+    {
+        return $this->flock;
+    }
+
+    /**
+     * @param mixed $flock
+     * @return WorkerRunningJob
+     */
+    public function setFlock($flock)
+    {
+        $this->flock = $flock;
+
+        return $this;
+    }
+
     /**
      * @param $work
      * @return $this

lib/Alchemy/Phrasea/Model/Repositories/WorkerRunningJobRepository.php

@@ -2,78 +2,150 @@
 
 namespace Alchemy\Phrasea\Model\Repositories;
 
-use Alchemy\Phrasea\Core\PhraseaTokens;
 use Alchemy\Phrasea\Model\Entities\WorkerRunningJob;
 use Alchemy\Phrasea\WorkerManager\Queue\MessagePublisher;
+use DateTime;
 use Doctrine\ORM\EntityRepository;
+use Doctrine\ORM\OptimisticLockException;
+use Exception;
 
 class WorkerRunningJobRepository extends EntityRepository
 {
     /**
-     *  return true if we can create subdef
+     * Acquire a "lock" to create a subdef
-     * @param $subdefName
+     * @param array $payload
-     * @param $recordId
+     * @return WorkerRunningJob
-     * @param $databoxId
+     * @throws OptimisticLockException
-     * @return bool
      */
-    public function canCreateSubdef($subdefName, $recordId, $databoxId)
+    public function canCreateSubdef($payload)
     {
-        $rsm = $this->createResultSetMappingBuilder('w');
+        return $this->getLock($payload, MessagePublisher::SUBDEF_CREATION_TYPE);
-        $rsm->addScalarResult('work_on','work_on');
-
-        $sql = 'SELECT work_on
-            FROM WorkerRunningJob
-            WHERE ((work = :write_meta) OR ((work = :make_subdef) AND work_on = :work_on) ) 
-            AND record_id = :record_id 
-            AND databox_id = :databox_id
-            AND status = :status';
-
-        $query = $this->_em->createNativeQuery($sql, $rsm);
-        $query->setParameters([
-            'write_meta' => MessagePublisher::WRITE_METADATAS_TYPE,
-            'make_subdef'=> MessagePublisher::SUBDEF_CREATION_TYPE,
-            'work_on'    => $subdefName,
-            'record_id'  => $recordId,
-            'databox_id' => $databoxId,
-            'status'     => WorkerRunningJob::RUNNING
-            ]
-        );
-
-        return count($query->getResult()) == 0;
     }
 
     /**
-     * return true if we can write meta
+     * Acquire a "lock" to write meta into a subdef
-     *
+     * @param array $payload
-     * @param $subdefName
+     * @return WorkerRunningJob
-     * @param $recordId
+     * @throws OptimisticLockException
-     * @param $databoxId
-     * @return bool
      */
-    public function canWriteMetadata($subdefName, $recordId, $databoxId)
+    public function canWriteMetadata($payload)
     {
-        $rsm = $this->createResultSetMappingBuilder('w');
+        return $this->getLock($payload, MessagePublisher::WRITE_METADATAS_TYPE);
-        $rsm->addScalarResult('work_on','work_on');
+    }
 
-        $sql = 'SELECT work_on
+    /**
-            FROM WorkerRunningJob
+     * Acquire a "lock" to work on a (sbid + rid + subdef) by inserting a row in WorkerRunningJob table.
-            WHERE ((work = :make_subdef) OR ((work = :write_meta) AND work_on = :work_on) ) 
+     * If it fails that means that another worker is already working on this file.
-            AND record_id = :record_id 
+     *
-            AND databox_id = :databox_id
+     * nb : this work only for "first try" where workerJobId is null (=insert).
-            AND status = :status';
+     *      for some retries (lock was acquired but worker job failed), the "count" of existing row is incremented (=update),
+     *      so many workers "could" update the same row...
+     *      __Luckily__, a rabbitmq message is consumed only once by a unique worker,
+     *      and different workers (write-meta, subdef) have their own queues and their own rows on table.
+     *      So working on a file always starts by a "first try", and concurency is not possible.
+     * todo : do not update, but insert a line for every try ?
+     *
+     * @param array $payload
+     * @param string $type
+     * @return WorkerRunningJob      the entity (created or updated) or null if file is already locked (duplicate key)
+     * @throws OptimisticLockException
+     */
+    private function getLock(array $payload, string $type)
+    {
+        if(!isset($payload['workerJobId'])) {
+            // insert a new row WorkerRunningJob : will fail if concurency
+            try {
+                $this->getEntityManager()->beginTransaction();
+                $date = new DateTime();
+                $workerRunningJob = new WorkerRunningJob();
+                $workerRunningJob
+                    ->setDataboxId($payload['databoxId'])
+                    ->setRecordId($payload['recordId'])
+                    ->setWork($type)
+                    ->setWorkOn($payload['subdefName'])
+                    ->setPayload([
+                        'message_type' => $type,
+                        'payload'      => $payload
+                    ])
+                    ->setPublished($date->setTimestamp($payload['published']))
+                    ->setStatus(WorkerRunningJob::RUNNING)
+                    ->setFlock($payload['subdefName']);
+                $this->getEntityManager()->persist($workerRunningJob);
 
-        $query = $this->_em->createNativeQuery($sql, $rsm);
+                $this->getEntityManager()->flush();
-        $query->setParameters([
+                $this->getEntityManager()->commit();
-                'make_subdef'=> MessagePublisher::SUBDEF_CREATION_TYPE,
-                'write_meta' => MessagePublisher::WRITE_METADATAS_TYPE,
-                'work_on'    => $subdefName,
-                'record_id'  => $recordId,
-                'databox_id' => $databoxId,
-                'status'     => WorkerRunningJob::RUNNING
-            ]
-        );
 
-        return count($query->getResult()) == 0;
+                return $workerRunningJob;
+            }
+            catch(Exception $e) {
+                // duplicate key ?
+                $this->getEntityManager()->rollback();
+                // for unpredicted other errors we can still ignore and return null (lock failed),
+                // because anyway the worker/rabbit retry-system will stop itself after n failures.
+            }
+        }
+        else {
+            // update an existing row : never fails (except bad id if row was purged)
+            try {
+                $this->getEntityManager()->beginTransaction();
+                $this->getEntityManager()->createQueryBuilder()
+                    ->update()
+                    ->set('info', ':info')->setParameter('info', WorkerRunningJob::ATTEMPT . $payload['count'])
+                    ->set('status', ':status')->setParameter('status', WorkerRunningJob::RUNNING)
+                    ->set('flock', ':flock')->setParameter('flock', $payload['subdefName'])
+                    ->where('id = :id')->setParameter('id', $payload['workerJobId'])
+                    ->getQuery()
+                    ->execute();
+
+                $this->getEntityManager()->flush();
+                $this->getEntityManager()->commit();
+
+                return $this->find($payload['workerJobId']);
+            }
+            catch (Exception $e) {
+                // really bad ? return null anyway
+                $this->getEntityManager()->rollback();
+                //$this->logger->error("Error persisting WorkerRunningJob !");
+            }
+        }
+
+        return null;
+    }
+
+
+    /**
+     * mark a job a "finished"
+     * nb : after a long job, connection may be lost so we reconnect.
+     *      But sometimes (?) a first commit fails (due to reconnect ?), while the second one is ok.
+     *      So here we try 2 times, just in case...
+     *
+     * @param WorkerRunningJob $workerRunningJob
+     * @param null $info
+     */
+    public function markFinished(WorkerRunningJob $workerRunningJob, $info = null)
+    {
+        $this->reconnect();
+        for($try=1; $try<=2; $try++) {
+            try {
+                $workerRunningJob->setStatus(WorkerRunningJob::FINISHED)
+                    ->setFinished(new DateTime('now'))
+                    ->setStatus(WorkerRunningJob::FINISHED)
+                    ->setFlock(null);
+                if (!is_null($info)) {
+                    $workerRunningJob->setInfo($info);
+                }
+
+                $this->getEntityManager()->beginTransaction();
+                $this->getEntityManager()->persist($workerRunningJob);
+                $this->getEntityManager()->flush();
+                $this->getEntityManager()->commit();
+
+                break;
+            }
+            catch (Exception $e) {
+                $this->getEntityManager()->rollback();
+            }
+        }
     }
 
     /**
@@ -134,8 +206,9 @@ class WorkerRunningJobRepository extends EntityRepository
         $platform = $connection->getDatabasePlatform();
         $this->_em->beginTransaction();
         try {
-        $connection->executeUpdate($platform->getTruncateTableSQL('WorkerRunningJob'));
+            $connection->executeUpdate($platform->getTruncateTableSQL('WorkerRunningJob'));
-        } catch (\Exception $e) {
+        }
+        catch (Exception $e) {
             $this->_em->rollback();
         }
     }
@@ -146,7 +219,8 @@ class WorkerRunningJobRepository extends EntityRepository
         try {
             $this->_em->getConnection()->delete('WorkerRunningJob', ['status' => WorkerRunningJob::FINISHED]);
             $this->_em->commit();
-        } catch (\Exception $e) {
+        }
+        catch (Exception $e) {
             $this->_em->rollback();
         }
     }

lib/Alchemy/Phrasea/WorkerManager/Controller/AdminConfigurationController.php

@@ -157,10 +157,10 @@ class AdminConfigurationController extends Controller
         /** @var WorkerRunningJob $workerRunningJob */
         $workerRunningJob = $repoWorker->find($workerId);
 
-        $workerRunningJob
+        $workerRunningJob->setStatus($request->request->get('status'));
-            ->setStatus($request->request->get('status'))
+        if($request->request->get('finished') == '1') {
-            ->setFinished(new \DateTime('now'))
+            $workerRunningJob->setFinished(new \DateTime('now'))->setFlock(null);
-        ;
+        }
 
         $em = $repoWorker->getEntityManager();
         $em->persist($workerRunningJob);

lib/Alchemy/Phrasea/WorkerManager/Subscriber/RecordSubscriber.php

@@ -119,12 +119,14 @@ class RecordSubscriber implements EventSubscriberInterface
                 $workerRunningJob
                     ->setInfo(WorkerRunningJob::ATTEMPT. ($event->getCount() - 1))
                     ->setStatus(WorkerRunningJob::ERROR)
+                    ->setFlock(null)            // unlock !
                 ;
 
                 $em->persist($workerRunningJob);
                 $em->flush();
                 $em->commit();
-            } catch (Exception $e) {
+            }
+            catch (Exception $e) {
                 $em->rollback();
             }
         }
@@ -240,12 +242,14 @@ class RecordSubscriber implements EventSubscriberInterface
                     $workerRunningJob
                         ->setInfo(WorkerRunningJob::ATTEMPT. ($event->getCount() - 1))
                         ->setStatus(WorkerRunningJob::ERROR)
+                        ->setFlock(null)    // unlock !
                     ;
 
                     $em->persist($workerRunningJob);
                     $em->flush();
                     $em->commit();
-                } catch (Exception $e) {
+                }
+                catch (Exception $e) {
                     $em->rollback();
                 }
             }
@@ -257,7 +261,8 @@ class RecordSubscriber implements EventSubscriberInterface
                 $event->getWorkerMessage()
             );
 
-        } else {
+        }
+        else {
             $databoxId = $event->getRecord()->getDataboxId();
             $recordId = $event->getRecord()->getRecordId();
 

lib/Alchemy/Phrasea/WorkerManager/Worker/SubdefCreationWorker.php

@@ -6,7 +6,6 @@ use Alchemy\Phrasea\Application\Helper\ApplicationBoxAware;
 use Alchemy\Phrasea\Core\PhraseaTokens;
 use Alchemy\Phrasea\Filesystem\FilesystemService;
 use Alchemy\Phrasea\Media\SubdefGenerator;
-use Alchemy\Phrasea\Model\Entities\WorkerRunningJob;
 use Alchemy\Phrasea\Model\Repositories\WorkerRunningJobRepository;
 use Alchemy\Phrasea\SearchEngine\Elastic\Indexer;
 use Alchemy\Phrasea\WorkerManager\Event\StoryCreateCoverEvent;
@@ -14,8 +13,10 @@ use Alchemy\Phrasea\WorkerManager\Event\SubdefinitionCreationFailureEvent;
 use Alchemy\Phrasea\WorkerManager\Event\SubdefinitionWritemetaEvent;
 use Alchemy\Phrasea\WorkerManager\Event\WorkerEvents;
 use Alchemy\Phrasea\WorkerManager\Queue\MessagePublisher;
+use Exception;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\EventDispatcher\EventDispatcherInterface;
+use Throwable;
 
 class SubdefCreationWorker implements WorkerInterface
 {
@@ -53,182 +54,136 @@ class SubdefCreationWorker implements WorkerInterface
 
     public function process(array $payload)
     {
-        if (isset($payload['recordId']) && isset($payload['databoxId'])) {
+        if (!isset($payload['recordId']) || !isset($payload['databoxId']) || !isset($payload['subdefName'])) {
-            $recordId       = $payload['recordId'];
+            // bad payload
-            $databoxId      = $payload['databoxId'];
+            $this->logger->error(sprintf("%s (%s) : bad payload", __FILE__, __LINE__));
-            $wantedSubdef   = [$payload['subdefName']];
+            return;
+        }
 
-            $databox = $this->findDataboxById($databoxId);
+        $recordId       = $payload['recordId'];
-            $record = $databox->get_record($recordId);
+        $databoxId      = $payload['databoxId'];
+        $subdefName     = $payload['subdefName'];
 
-            $oldLogger = $this->subdefGenerator->getLogger();
+        $databox = $this->findDataboxById($databoxId);
+        $record = $databox->get_record($recordId);
 
-            $message = [
+        if ($record->isStory()) {
-                'message_type'  => MessagePublisher::SUBDEF_CREATION_TYPE,
+            return;
-                'payload'       => $payload
+        }
-            ];
 
-            if (!$record->isStory()) {
+        $oldLogger = $this->subdefGenerator->getLogger();
-                // check if there is a write meta running for the record or the same task running
-                $canCreateSubdef = $this->repoWorker->canCreateSubdef($payload['subdefName'], $recordId, $databoxId);
 
-                if (!$canCreateSubdef) {
+        // try to "lock" the file, will return null if already locked (key unicity)
-                    // the file is in used to write meta
+        // = insert a row with unqiue sbid + rid + subdefname (todo : replace the subdefname with a subdef_id ?)
+        $workerRunningJob = $this->repoWorker->canCreateSubdef($payload);
 
-                    $this->messagePublisher->publishDelayedMessage($message, MessagePublisher::SUBDEF_CREATION_TYPE);
+        if (is_null($workerRunningJob)) {
+            // the file was locked by another worker, delay to retry later
+            $this->messagePublisher->publishDelayedMessage(
+                [
+                    'message_type'  => MessagePublisher::SUBDEF_CREATION_TYPE,
+                    'payload'       => $payload
+                ],
+                MessagePublisher::SUBDEF_CREATION_TYPE
+            );
+            return ;
+        }
 
-                    return ;
+        // here the entity is "locked" (unique key)
-                }
 
-                // tell that a file is in used to create subdef
+        $this->subdefGenerator->setLogger($this->logger);
-                $em = $this->repoWorker->getEntityManager();
-                $this->repoWorker->reconnect();
 
-                if (isset($payload['workerJobId'])) {
+        try {
-                    /** @var WorkerRunningJob $workerRunningJob */
+            $this->subdefGenerator->generateSubdefs($record, [$subdefName]);
-                    $workerRunningJob = $this->repoWorker->find($payload['workerJobId']);
+        }
+        catch (Exception $e) {
+            $this->logger->error("Exception catched: " . $e->getMessage());
+        }
+        catch (Throwable $e) {
+            $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
+            $workerMessage = "Exception throwable catched when create subdef for the recordID: " .$recordId;
 
-                    if ($workerRunningJob == null) {
+            $this->logger->error($workerMessage);
-                        $this->logger->error("Given workerJobId not found !");
 
-                        return ;
+            /** @uses \Alchemy\Phrasea\WorkerManager\Subscriber\RecordSubscriber::onSubdefinitionCreationFailure() */
-                    }
+            $this->dispatcher->dispatch(WorkerEvents::SUBDEFINITION_CREATION_FAILURE, new SubdefinitionCreationFailureEvent(
+                $record,
+                $subdefName,
+                $workerMessage,
+                $count,
+                $workerRunningJob->getId()
+            ));
 
-                    $workerRunningJob
+            // the subscriber will "unlock" the row, no need to do it here
-                        ->setInfo(WorkerRunningJob::ATTEMPT . $payload['count'])
+            return ;
-                        ->setStatus(WorkerRunningJob::RUNNING);
+        }
 
-                    $em->persist($workerRunningJob);
+        // begin to check if the subdef is successfully generated
+        $subdef = $record->getDatabox()->get_subdef_structure()->getSubdefGroup($record->getType())->getSubdef($subdefName);
+        $filePathToCheck = null;
 
-                    $em->flush();
+        if ($record->has_subdef($subdefName) ) {
+            $filePathToCheck = $record->get_subdef($subdefName)->getRealPath();
+        }
 
-                } else {
+        $filePathToCheck = $this->filesystem->generateSubdefPathname($record, $subdef, $filePathToCheck);
-                    $em->beginTransaction();
-                    try {
-                        $date = new \DateTime();
-                        $workerRunningJob = new WorkerRunningJob();
-                        $workerRunningJob
-                            ->setDataboxId($databoxId)
-                            ->setRecordId($recordId)
-                            ->setWork(MessagePublisher::SUBDEF_CREATION_TYPE)
-                            ->setWorkOn($payload['subdefName'])
-                            ->setPayload($message)
-                            ->setPublished($date->setTimestamp($payload['published']))
-                            ->setStatus(WorkerRunningJob::RUNNING)
-                        ;
 
-                        $em->persist($workerRunningJob);
+        if (!$this->filesystem->exists($filePathToCheck)) {
-                        $em->flush();
 
-                        $em->commit();
+            $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
-                    } catch (\Exception $e) {
-                        $em->rollback();
-                    }
-                }
 
-                $this->subdefGenerator->setLogger($this->logger);
+            /** @uses \Alchemy\Phrasea\WorkerManager\Subscriber\RecordSubscriber::onSubdefinitionCreationFailure() */
+            $this->dispatcher->dispatch(WorkerEvents::SUBDEFINITION_CREATION_FAILURE, new SubdefinitionCreationFailureEvent(
+                $record,
+                $subdefName,
+                'Subdef generation failed !',
+                $count,
+                $workerRunningJob->getId()
+            ));
 
-                try {
+            $this->subdefGenerator->setLogger($oldLogger);
-                    $this->subdefGenerator->generateSubdefs($record, $wantedSubdef);
-                } catch (\Exception $e) {
-                    $this->logger->error("Exception catched: " . $e->getMessage());
 
-                } catch (\Throwable $e) {
+            // the subscriber will "unlock" the row, no need to do it here
-                    $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
+            return ;
-                    $workerMessage = "Exception throwable catched when create subdef for the recordID: " .$recordId;
+        }
 
-                    $this->logger->error($workerMessage);
+        // checking ended
 
-                    $this->dispatcher->dispatch(WorkerEvents::SUBDEFINITION_CREATION_FAILURE, new SubdefinitionCreationFailureEvent(
+        // order to write meta for the subdef if needed
-                        $record,
+        /** @uses \Alchemy\Phrasea\WorkerManager\Subscriber\RecordSubscriber::onSubdefinitionWritemeta() */
-                        $payload['subdefName'],
+        $this->dispatcher->dispatch(
-                        $workerMessage,
+            WorkerEvents::SUBDEFINITION_WRITE_META,
-                        $count,
+            new SubdefinitionWritemetaEvent(
-                        $workerRunningJob->getId()
+                $record,
-                    ));
+                $subdefName
+            )
+        );
 
-                    return ;
+        $this->subdefGenerator->setLogger($oldLogger);
-                }
 
-                // begin to check if the subdef is successfully generated
+        //  update jeton when subdef is created
-                $subdef = $record->getDatabox()->get_subdef_structure()->getSubdefGroup($record->getType())->getSubdef($payload['subdefName']);
+        $this->updateJeton($record);
-                $filePathToCheck = null;
 
-                if ($record->has_subdef($payload['subdefName']) ) {
+        $parents = $record->get_grouping_parents();
-                    $filePathToCheck = $record->get_subdef($payload['subdefName'])->getRealPath();
-                }
 
-                $filePathToCheck = $this->filesystem->generateSubdefPathname($record, $subdef, $filePathToCheck);
+        //  create a cover for a story
-
+        //  used when uploaded via uploader-service and grouped as a story
-                if (!$this->filesystem->exists($filePathToCheck)) {
+        if (!$parents->is_empty() && isset($payload['status']) && $payload['status'] == MessagePublisher::NEW_RECORD_MESSAGE  && in_array($payload['subdefName'], array('thumbnail', 'preview'))) {
-
+            foreach ($parents->get_elements() as $story) {
-                    $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
+                if (self::checkIfFirstChild($story, $record)) {
-
+                    $data = implode('_', [$databoxId, $story->getRecordId(), $recordId, $payload['subdefName']]);
-                    $this->dispatcher->dispatch(WorkerEvents::SUBDEFINITION_CREATION_FAILURE, new SubdefinitionCreationFailureEvent(
-                        $record,
-                        $payload['subdefName'],
-                        'Subdef generation failed !',
-                        $count,
-                        $workerRunningJob->getId()
-                    ));
-
-                    $this->subdefGenerator->setLogger($oldLogger);
-                    return ;
-                }
-                // checking ended
-
-                // order to write meta for the subdef if needed
-                $this->dispatcher->dispatch(WorkerEvents::SUBDEFINITION_WRITE_META, new SubdefinitionWritemetaEvent(
-                    $record,
-                    $payload['subdefName'])
-                );
-
-                $this->subdefGenerator->setLogger($oldLogger);
-
-                //  update jeton when subdef is created
-                $this->updateJeton($record);
-
-                $parents = $record->get_grouping_parents();
-
-                //  create a cover for a story
-                //  used when uploaded via uploader-service and grouped as a story
-                if (!$parents->is_empty() && isset($payload['status']) && $payload['status'] == MessagePublisher::NEW_RECORD_MESSAGE  && in_array($payload['subdefName'], array('thumbnail', 'preview'))) {
-                    foreach ($parents->get_elements() as $story) {
-                        if (self::checkIfFirstChild($story, $record)) {
-                            $data = implode('_', [$databoxId, $story->getRecordId(), $recordId, $payload['subdefName']]);
-
-                            $this->dispatcher->dispatch(WorkerEvents::STORY_CREATE_COVER, new StoryCreateCoverEvent($data));
-                        }
-                    }
-                }
-
-                // update elastic
-                $this->indexer->flushQueue();
-
-                // tell that we have finished to work on this file
-                $this->repoWorker->reconnect();
-                $em->getConnection()->beginTransaction();
-                try {
-                    $workerRunningJob->setStatus(WorkerRunningJob::FINISHED);
-                    $workerRunningJob->setFinished(new \DateTime('now'));
-                    $em->persist($workerRunningJob);
-                    $em->flush();
-                    $em->commit();
-                } catch (\Exception $e) {
-                    try {
-                        $em->getConnection()->beginTransaction();
-                        $workerRunningJob->setStatus(WorkerRunningJob::FINISHED);
-                        $em->persist($workerRunningJob);
-                        $em->flush();
-                        $em->commit();
-                    } catch (\Exception $e) {
-                        $this->messagePublisher->pushLog("rollback on recordID :" . $workerRunningJob->getRecordId());
-                        $em->rollback();
-                    }
 
+                    $this->dispatcher->dispatch(WorkerEvents::STORY_CREATE_COVER, new StoryCreateCoverEvent($data));
                 }
             }
         }
+
+        // update elastic
+        $this->indexer->flushQueue();
+
+        // tell that we have finished to work on this file
+        $this->repoWorker->markFinished($workerRunningJob);
+
     }
 
     public static function checkIfFirstChild(\record_adapter $story, \record_adapter $record)

lib/Alchemy/Phrasea/WorkerManager/Worker/WriteMetadatasWorker.php

@@ -7,7 +7,6 @@ use Alchemy\Phrasea\Application\Helper\DispatcherAware;
 use Alchemy\Phrasea\Application\Helper\EntityManagerAware;
 use Alchemy\Phrasea\Core\PhraseaTokens;
 use Alchemy\Phrasea\Metadata\TagFactory;
-use Alchemy\Phrasea\Model\Entities\WorkerRunningJob;
 use Alchemy\Phrasea\Model\Repositories\WorkerRunningJobRepository;
 use Alchemy\Phrasea\WorkerManager\Event\SubdefinitionWritemetaEvent;
 use Alchemy\Phrasea\WorkerManager\Event\WorkerEvents;
@@ -57,262 +56,227 @@ class WriteMetadatasWorker implements WorkerInterface
 
     public function process(array $payload)
     {
-        if (isset($payload['recordId']) && isset($payload['databoxId'])) {
+        // mandatory args
-            $recordId  = $payload['recordId'];
+        if (!isset($payload['recordId']) || !isset($payload['databoxId']) || !isset($payload['subdefName'])) {
-            $databoxId = $payload['databoxId'];
+            // bad payload
-
+            $this->logger->error(sprintf("%s (%s) : bad payload", __FILE__, __LINE__));
-            $MWG      = isset($payload['MWG']) ? $payload['MWG'] : false;
+            return;
-            $clearDoc = isset($payload['clearDoc']) ? $payload['clearDoc'] : false;
-            $databox = $this->findDataboxById($databoxId);
-
-            // check if there is a make subdef running for the record or the same task running
-            $canWriteMeta = $this->repoWorker->canWriteMetadata($payload['subdefName'], $recordId, $databoxId);
-
-            $message = [
-                'message_type'  => MessagePublisher::WRITE_METADATAS_TYPE,
-                'payload'       => $payload
-            ];
-
-            if (!$canWriteMeta) {
-                // the file is in used to generate subdef
-
-                $this->messagePublisher->publishDelayedMessage($message, MessagePublisher::WRITE_METADATAS_TYPE);
-
-                return ;
-            }
-
-            $record  = $databox->get_record($recordId);
-
-            if ($record->getMimeType() == 'image/svg+xml') {
-
-                $this->logger->error("Can't write meta on svg file!");
-
-                return;
-            }
-
-            // tell that a file is in used to create subdef
-            $em = $this->getEntityManager();
-            $this->repoWorker->reconnect();
-
-            if (isset($payload['workerJobId'])) {
-                /** @var WorkerRunningJob $workerRunningJob */
-                $workerRunningJob = $this->repoWorker->find($payload['workerJobId']);
-
-                if ($workerRunningJob == null) {
-                    $this->logger->error("Given workerJobId not found !");
-
-                    return ;
-                }
-
-                $workerRunningJob
-                    ->setInfo(WorkerRunningJob::ATTEMPT . $payload['count'])
-                    ->setStatus(WorkerRunningJob::RUNNING)
-                ;
-
-                $em->persist($workerRunningJob);
-
-                $em->flush();
-            } else {
-                $em->beginTransaction();
-
-                try {
-                    $date = new DateTime();
-                    $workerRunningJob = new WorkerRunningJob();
-                    $workerRunningJob
-                        ->setDataboxId($databoxId)
-                        ->setRecordId($recordId)
-                        ->setWork(MessagePublisher::WRITE_METADATAS_TYPE)
-                        ->setWorkOn($payload['subdefName'])
-                        ->setPayload($message)
-                        ->setPublished($date->setTimestamp($payload['published']))
-                        ->setStatus(WorkerRunningJob::RUNNING)
-                    ;
-
-                    $em->persist($workerRunningJob);
-                    $em->flush();
-
-                    $em->commit();
-                } catch (Exception $e) {
-                    $em->rollback();
-                    $this->logger->error("Error persisting WorkerRunningJob !");
-
-                    return ;
-                }
-            }
-
-            try {
-                $subdef = $record->get_subdef($payload['subdefName']);
-            } catch (Exception $e) {
-                $workerMessage = "Exception catched when try to get subdef " .$payload['subdefName']. " from DB for the recordID: " .$recordId;
-                $this->logger->error($workerMessage);
-
-                $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
-
-                $this->dispatch(WorkerEvents::SUBDEFINITION_WRITE_META, new SubdefinitionWritemetaEvent(
-                    $record,
-                    $payload['subdefName'],
-                    SubdefinitionWritemetaEvent::FAILED,
-                    $workerMessage,
-                    $count,
-                    $workerRunningJob->getId()
-                ));
-
-                return ;
-            }
-
-            if ($subdef->is_physically_present()) {
-                $metadata = new MetadataBag();
-
-                // add Uuid in metadatabag
-                if ($record->getUuid()) {
-                    $metadata->add(
-                        new Metadata(
-                            new Tag\XMPExif\ImageUniqueID(),
-                            new Mono($record->getUuid())
-                        )
-                    );
-                    $metadata->add(
-                        new Metadata(
-                            new Tag\ExifIFD\ImageUniqueID(),
-                            new Mono($record->getUuid())
-                        )
-                    );
-                    $metadata->add(
-                        new Metadata(
-                            new Tag\IPTC\UniqueDocumentID(),
-                            new Mono($record->getUuid())
-                        )
-                    );
-                }
-
-                // read document fields and add to metadatabag
-                $caption = $record->get_caption();
-                foreach ($databox->get_meta_structure() as $fieldStructure) {
-
-                    $tagName = $fieldStructure->get_tag()->getTagname();
-                    $fieldName = $fieldStructure->get_name();
-
-                    // skip fields with no src
-                    if ($tagName == '' || $tagName == 'Phraseanet:no-source') {
-                        continue;
-                    }
-
-                    // check exiftool known tags to skip Phraseanet:tf-*
-                    try {
-                        $tag = TagFactory::getFromRDFTagname($tagName);
-                        if(!$tag->isWritable()) {
-                            continue;
-                        }
-                    } catch (TagUnknown $e) {
-                        continue;
-                    }
-
-                    try {
-                        $field = $caption->get_field($fieldName);
-                        $fieldValues = $field->get_values();
-
-                        if ($fieldStructure->is_multi()) {
-                            $values = array();
-                            foreach ($fieldValues as $value) {
-                                $values[] = $this->removeNulChar($value->getValue());
-                            }
-
-                            $value = new Multi($values);
-                        } else {
-                            $fieldValue = array_pop($fieldValues);
-                            $value = $this->removeNulChar($fieldValue->getValue());
-
-                            // fix the dates edited into phraseanet
-                            if($fieldStructure->get_type() === $fieldStructure::TYPE_DATE) {
-                                try {
-                                    $value = self::fixDate($value); // will return NULL if the date is not valid
-                                }
-                                catch (Exception $e) {
-                                    $value = null;    // do NOT write back to iptc
-                                }
-                            }
-
-                            if($value !== null) {   // do not write invalid dates
-                                $value = new Mono($value);
-                            }
-                        }
-                    } catch(Exception $e) {
-                        // the field is not set in the record, erase it
-                        if ($fieldStructure->is_multi()) {
-                            $value = new Multi(array(''));
-                        }
-                        else {
-                            $value = new Mono('');
-                        }
-                    }
-
-                    if($value !== null) {   // do not write invalid data
-                        $metadata->add(
-                            new Metadata($fieldStructure->get_tag(), $value)
-                        );
-                    }
-                }
-
-                $this->writer->reset();
-
-                if ($MWG) {
-                    $this->writer->setModule(Writer::MODULE_MWG, true);
-                }
-
-                $this->writer->erase($subdef->get_name() != 'document' || $clearDoc, true);
-
-                // write meta in file
-                try {
-                    $this->writer->write($subdef->getRealPath(), $metadata);
-
-                    $this->messagePublisher->pushLog(sprintf('meta written for sbasid=%1$d - recordid=%2$d (%3$s)', $databox->get_sbas_id(), $recordId, $subdef->get_name() ));
-                } catch (Exception $e) {
-                    $workerMessage = sprintf('meta NOT written for sbasid=%1$d - recordid=%2$d (%3$s) because "%4$s"', $databox->get_sbas_id(), $recordId, $subdef->get_name() , $e->getMessage());
-                    $this->logger->error($workerMessage);
-
-                    $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
-
-                    $this->dispatch(WorkerEvents::SUBDEFINITION_WRITE_META, new SubdefinitionWritemetaEvent(
-                        $record,
-                        $payload['subdefName'],
-                        SubdefinitionWritemetaEvent::FAILED,
-                        $workerMessage,
-                        $count,
-                        $workerRunningJob->getId()
-                    ));
-                }
-
-                // mark write metas finished
-                $this->updateJeton($record);
-            } else {
-                $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
-
-                $this->dispatch(WorkerEvents::SUBDEFINITION_WRITE_META, new SubdefinitionWritemetaEvent(
-                    $record,
-                    $payload['subdefName'],
-                    SubdefinitionWritemetaEvent::FAILED,
-                    'Subdef is not physically present!',
-                    $count,
-                    $workerRunningJob->getId()
-                ));
-            }
-
-
-            // tell that we have finished to work on this file
-            $this->repoWorker->reconnect();
-            $em->beginTransaction();
-            try {
-                $workerRunningJob->setStatus(WorkerRunningJob::FINISHED);
-                $workerRunningJob->setFinished(new DateTime('now'));
-                $em->persist($workerRunningJob);
-                $em->flush();
-                $em->commit();
-            } catch (Exception $e) {
-                $em->rollback();
-            }
-
         }
 
+        $recordId    = $payload['recordId'];
+        $databoxId   = $payload['databoxId'];
+        $subdefName  = $payload['subdefName'];
+
+        $MWG         = $payload['MWG'] ?? false;
+        $clearDoc    = $payload['clearDoc'] ?? false;
+        $databox     = $this->findDataboxById($databoxId);
+
+        // try to "lock" the file, will return null if already locked (key unicity)
+        // = insert a row with unqiue sbid + rid + subdefname (todo : replace the subdefname with a subdef_id ?)
+        $workerRunningJob = $this->repoWorker->canWriteMetadata($payload);
+
+        if (is_null($workerRunningJob)) {
+            // the file was locked by another worker, delay to retry later
+            $this->messagePublisher->publishDelayedMessage(
+                [
+                    'message_type'  => MessagePublisher::WRITE_METADATAS_TYPE,
+                    'payload'       => $payload
+                ],
+                MessagePublisher::WRITE_METADATAS_TYPE
+            );
+            return ;
+        }
+
+        // here the entity is "locked" (unique key)
+
+        $record  = $databox->get_record($recordId);
+
+        if ($record->getMimeType() == 'image/svg+xml') {
+
+            $this->logger->error("Can't write meta on svg file!");
+
+            // tell that we have finished to work on this file ("unlock")
+            $this->repoWorker->markFinished($workerRunningJob, "Can't write meta on svg file!");
+
+            return;
+        }
+
+        $this->repoWorker->reconnect();
+
+        try {
+            $subdef = $record->get_subdef($subdefName);
+        }
+        catch (Exception $e) {
+            $workerMessage = "Exception catched when try to get subdef " .$subdefName. " from DB for the recordID: " .$recordId;
+            $this->logger->error($workerMessage);
+
+            $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
+
+            /** @uses \Alchemy\Phrasea\WorkerManager\Subscriber\RecordSubscriber::onSubdefinitionWritemeta() */
+            $this->dispatch(WorkerEvents::SUBDEFINITION_WRITE_META, new SubdefinitionWritemetaEvent(
+                $record,
+                $subdefName,
+                SubdefinitionWritemetaEvent::FAILED,
+                $workerMessage,
+                $count,
+                $workerRunningJob->getId()
+            ));
+
+            // the subscriber will "unlock" the row, no need to do it here
+            return ;
+        }
+
+        if (!$subdef->is_physically_present()) {
+            $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
+
+            /** @uses \Alchemy\Phrasea\WorkerManager\Subscriber\RecordSubscriber::onSubdefinitionWritemeta() */
+            $this->dispatch(WorkerEvents::SUBDEFINITION_WRITE_META, new SubdefinitionWritemetaEvent(
+                $record,
+                $subdefName,
+                SubdefinitionWritemetaEvent::FAILED,
+                'Subdef is not physically present!',
+                $count,
+                $workerRunningJob->getId()
+            ));
+
+            // the subscriber will "unlock" the row, no need to do it here
+            return ;
+        }
+
+        // here we can try to do the job
+
+        $metadata = new MetadataBag();
+
+        // add Uuid in metadatabag
+        if ($record->getUuid()) {
+            $metadata->add(
+                new Metadata(
+                    new Tag\XMPExif\ImageUniqueID(),
+                    new Mono($record->getUuid())
+                )
+            );
+            $metadata->add(
+                new Metadata(
+                    new Tag\ExifIFD\ImageUniqueID(),
+                    new Mono($record->getUuid())
+                )
+            );
+            $metadata->add(
+                new Metadata(
+                    new Tag\IPTC\UniqueDocumentID(),
+                    new Mono($record->getUuid())
+                )
+            );
+        }
+
+        // read document fields and add to metadatabag
+        $caption = $record->get_caption();
+        foreach ($databox->get_meta_structure() as $fieldStructure) {
+
+            $tagName = $fieldStructure->get_tag()->getTagname();
+            $fieldName = $fieldStructure->get_name();
+
+            // skip fields with no src
+            if ($tagName == '' || $tagName == 'Phraseanet:no-source') {
+                continue;
+            }
+
+            // check exiftool known tags to skip Phraseanet:tf-*
+            try {
+                $tag = TagFactory::getFromRDFTagname($tagName);
+                if(!$tag->isWritable()) {
+                    continue;
+                }
+            } catch (TagUnknown $e) {
+                continue;
+            }
+
+            try {
+                $field = $caption->get_field($fieldName);
+                $fieldValues = $field->get_values();
+
+                if ($fieldStructure->is_multi()) {
+                    $values = array();
+                    foreach ($fieldValues as $value) {
+                        $values[] = $this->removeNulChar($value->getValue());
+                    }
+
+                    $value = new Multi($values);
+                } else {
+                    $fieldValue = array_pop($fieldValues);
+                    $value = $this->removeNulChar($fieldValue->getValue());
+
+                    // fix the dates edited into phraseanet
+                    if($fieldStructure->get_type() === $fieldStructure::TYPE_DATE) {
+                        try {
+                            $value = self::fixDate($value); // will return NULL if the date is not valid
+                        }
+                        catch (Exception $e) {
+                            $value = null;    // do NOT write back to iptc
+                        }
+                    }
+
+                    if($value !== null) {   // do not write invalid dates
+                        $value = new Mono($value);
+                    }
+                }
+            }
+            catch(Exception $e) {
+                // the field is not set in the record, erase it
+                if ($fieldStructure->is_multi()) {
+                    $value = new Multi(array(''));
+                }
+                else {
+                    $value = new Mono('');
+                }
+            }
+
+            if($value !== null) {   // do not write invalid data
+                $metadata->add(
+                    new Metadata($fieldStructure->get_tag(), $value)
+                );
+            }
+        }
+
+        $this->writer->reset();
+
+        if ($MWG) {
+            $this->writer->setModule(Writer::MODULE_MWG, true);
+        }
+
+        $this->writer->erase($subdef->get_name() != 'document' || $clearDoc, true);
+
+        // write meta in file
+        try {
+            $this->writer->write($subdef->getRealPath(), $metadata);
+
+            $this->messagePublisher->pushLog(sprintf('meta written for sbasid=%1$d - recordid=%2$d (%3$s)', $databox->get_sbas_id(), $recordId, $subdef->get_name() ));
+        }
+        catch (Exception $e) {
+            $workerMessage = sprintf('meta NOT written for sbasid=%1$d - recordid=%2$d (%3$s) because "%4$s"', $databox->get_sbas_id(), $recordId, $subdef->get_name() , $e->getMessage());
+            $this->logger->error($workerMessage);
+
+            $count = isset($payload['count']) ? $payload['count'] + 1 : 2 ;
+
+            /** @uses \Alchemy\Phrasea\WorkerManager\Subscriber\RecordSubscriber::onSubdefinitionWritemeta() */
+            $this->dispatch(WorkerEvents::SUBDEFINITION_WRITE_META, new SubdefinitionWritemetaEvent(
+                $record,
+                $subdefName,
+                SubdefinitionWritemetaEvent::FAILED,
+                $workerMessage,
+                $count,
+                $workerRunningJob->getId()
+            ));
+
+            // the subscriber will "unlock" the row, no need to do it here
+            return ;
+        }
+
+        // mark write metas finished
+        $this->updateJeton($record);
+
+        // tell that we have finished to work on this file
+        $this->repoWorker->markFinished($workerRunningJob);
     }
 
     private function removeNulChar($value)

templates/web/admin/worker-manager/worker_info.html.twig

@@ -153,7 +153,8 @@
                 type: "POST",
                 dataType: 'json',
                 data: {
-                    status: '{{ constant("\\Alchemy\\Phrasea\\Model\\Entities\\WorkerRunningJob::INTERRUPT") }}'
+                    status: '{{ constant("\\Alchemy\\Phrasea\\Model\\Entities\\WorkerRunningJob::INTERRUPT") }}',
+                    finished: '1'     // manual interrupt also means "finished", it must update the date and unlock the row
                 },
                 url: "/admin/worker-manager/"+ workerId +"/change-status",
                 success: function (data) {