intoduce new async_slow queue for image deletion

compose.yaml

@@ -32,7 +32,7 @@ services:
             context: .
             target: prod
         restart: unless-stopped
-        command: bin/console messenger:consume scheduler_default old async outbox deliver inbox resolve receive failed --time-limit=3600
+        command: bin/console messenger:consume scheduler_default old async async_slow outbox deliver inbox resolve receive failed --time-limit=3600
         healthcheck:
             test: ['CMD-SHELL', "ps aux | grep 'messenger[:]consume' || exit 1"]
         env_file: .env

config/packages/messenger.yaml

@@ -23,6 +23,23 @@ framework:
                     max_delay: 76800000
                     jitter: 0
                 serializer: messenger.transport.symfony_serializer
+            async_slow:
+                dsn: "%env(MESSENGER_TRANSPORT_DSN)%"
+                options:
+                    queues:
+                        async_slow:
+                            arguments:
+                                x-queue-version: 2
+                                x-queue-type: 'classic'
+                    exchange:
+                        name: async_slow
+                retry_strategy:
+                    max_retries: 5
+                    delay: 300000
+                    multiplier: 4
+                    max_delay: 76800000
+                    jitter: 0
+                serializer: messenger.transport.symfony_serializer
             inbox:
                 dsn: "%env(MESSENGER_TRANSPORT_DSN)%"
                 options:
@@ -137,6 +154,7 @@ framework:
         routing:
             # Route your messages to the transports
             App\Message\Contracts\AsyncMessageInterface: async
+            App\Message\Contracts\AsyncSlowMessageInterface: async_slow
             App\Message\Contracts\ActivityPubInboxInterface: inbox
             App\Message\Contracts\ActivityPubInboxReceiveInterface: receive
             App\Message\Contracts\ActivityPubOutboxDeliverInterface: deliver

docs/02-admin/01-installation/01-bare_metal.md

@@ -576,7 +576,7 @@ With the following content:
 
 ```ini
 [program:messenger]
-command=php /var/www/mbin/bin/console messenger:consume scheduler_default old async outbox deliver inbox resolve receive failed --time-limit=3600
+command=php /var/www/mbin/bin/console messenger:consume scheduler_default old async async_slow outbox deliver inbox resolve receive failed --time-limit=3600
 #stdout_logfile=NONE
 #redirect_stderr=true
 user=www-data

docs/02-admin/04-running-mbin/04-messenger.md

@@ -3,21 +3,22 @@
 The symphony messengers are background workers for a lot of different task, the biggest one being handling all the ActivityPub traffic.  
 We have a few different queues:
 
-1. `receive` [RabbitMQ]: everything any remote instance sends to us will first end up in this queue.
-   When processing it will be determined what kind of message it is (creation of a thread, a new comment, etc.)
-2. `inbox` [RabbitMQ]: messages from `receive` with the determined kind of incoming message will end up here and the necessary actions will be executed.
-   This is the place where the thread or comment will actually be created
-3. `outbox` [RabbitMQ]: when a user creates a thread or a comment, a message will be created and send to the outbox queue
+1. `receive` [RabbitMQ]: Everything any remote instance sends to us will first end up in this queue.
+   When processing it will be determined what kind of message it is (creation of a thread, a new comment, etc.).
+2. `inbox` [RabbitMQ]: Messages from `receive` with the determined kind of incoming message will end up here and the necessary actions will be executed.
+   This is the place where the thread or comment will actually be created.
+3. `outbox` [RabbitMQ]: When a user creates a thread or a comment, a message will be created and send to the `outbox` queue
    to build the ActivityPub object that will be sent to remote instances.
    After the object is built and the inbox addresses of all the remote instances who are interested in the message are gathered,
-   we will create a `DeliverMessage` for every one of them, which will be sent to the `deliver` queue
-4. `deliver` [RabbitMQ]: Actually sending out the ActivityPub objects to other instances
+   we will create a `DeliverMessage` for every one of them, which will be sent to the `deliver` queue.
+4. `deliver` [RabbitMQ]: Actually sending out the ActivityPub objects to other instances.
 5. `resolve` [RabbitMQ]: Resolving dependencies or ActivityPub actors.
-   For example if your instance gets a like message for a post that is not on your instance a message resolving that dependency will be dispatched to this queue
-6. `async` [RabbitMQ]: messages in async are local actions that are relevant to this instance, e.g. creating notifications, fetching embedded images, etc.
-7. `old` [RabbitMQ]: the standard messages queue that existed before. This exists solely for compatibility purposes and might be removed later on
-8. `failed` [PostgreSQL]: jobs from the other queues that have been retried, but failed. They get retried a few times again, before they end up in
-9. `dead` [PostgreSQL]: dead jobs that will not be retried
+   For example if your instance gets a like message for a post that is not on your instance a message resolving that dependency will be dispatched to this queue.
+6. `async` [RabbitMQ]: Messages in `async` are local actions that are relevant to this instance, e.g. creating notifications, fetching embedded images, etc.
+7. `async_slow` [RabbitMQ]: Similar to the `async` queue, but for messages which might take a long time (up to hours) to process. It is recommended to run a separate messenger for this queue.
+8. `old` [RabbitMQ]: The standard messages queue that existed before. This exists solely for compatibility purposes and might be removed later on.
+9. `failed` [PostgreSQL]: Jobs from the other queues that have been retried, but failed. They get retried a few times again, before they end up in `dead`.
+10. `dead` [PostgreSQL]: Dead jobs that will not be retried.
 
 We need the `dead` queue so that messages that throw a `UnrecoverableMessageHandlingException`, which is used to indicate that a message should not be retried and go straight to the supplied failure queue
 

src/Message/Contracts/AsyncSlowMessageInterface.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Message\Contracts;
+
+interface AsyncSlowMessageInterface extends MessageInterface
+{
+}

src/Message/DeleteImageMessage.php

@@ -6,6 +6,9 @@
 
 use App\Message\Contracts\AsyncMessageInterface;
 
+/**
+ * Attempts to delete the image entity from the database, but not the file from storage.
+ */
 class DeleteImageMessage implements AsyncMessageInterface
 {
     public function __construct(public int $id, public bool $force = false)

src/Message/DeleteImageV2Message.php

@@ -0,0 +1,37 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\Message;
+
+use App\Entity\Image;
+use App\Message\Contracts\AsyncSlowMessageInterface;
+
+/**
+ * Will check for every provided image if it is not referenced anywhere
+ * and if this is the case deletes the entity from the database & the file from storage.
+ */
+class DeleteImageV2Message implements AsyncSlowMessageInterface
+{
+    /**
+     * @param array<string, ?string> $images a list of mappings where the key is the sha256 (hex) of the image entity and the value is its filepath
+     */
+    public function __construct(
+        public array $images,
+    ) {
+    }
+
+    public static function fromImage(Image $img): self
+    {
+        return new self([bin2hex($img->sha256) => $img->filePath]);
+    }
+
+    public static function fromImages(array $imgs): self
+    {
+        $content = [];
+        foreach ($imgs as $img) {
+            $content[bin2hex($img->sha256)] = $img->filePath;
+        }
+        return new self($content);
+    }
+}

src/MessageHandler/DeleteImageV2Handler.php

@@ -0,0 +1,96 @@
+<?php
+
+declare(strict_types=1);
+
+namespace App\MessageHandler;
+
+use App\Entity\Image;
+use App\Message\DeleteImageV2Message;
+use App\Repository\ImageRepository;
+use App\Service\ImageManagerInterface;
+use Doctrine\ORM\EntityManagerInterface;
+use Psr\Log\LoggerInterface;
+use Symfony\Component\HttpKernel\KernelInterface;
+use Symfony\Component\Messenger\Attribute\AsMessageHandler;
+
+#[AsMessageHandler]
+readonly class DeleteImageV2Handler
+{
+    private int $chunkSize;
+
+    public function __construct(
+        KernelInterface $kernel,
+        private EntityManagerInterface $entityManager,
+        private ImageRepository $imageRepository,
+        private ImageManagerInterface $imageManager,
+        private LoggerInterface $logger,
+    ) {
+        $this->chunkSize = ('test' !== $kernel->getEnvironment()) ? 256 : 2;
+    }
+
+    public function __invoke(DeleteImageV2Message $message): void
+    {
+        try {
+            /*
+             * Split workload into chunks to balance overhead with probability of race-conditions. A race-condition can look like the following:
+             *   1. one image in a chunk is detected as an orphan
+             *   2. entity is deleted and transaction commited
+             *   3. new content is created with the same image (resulting in the same file path) while $filesToDelete is iterated
+             *   4. the file is deleted; the new image entity created by the new content now has an invalid path
+             */
+            $batches = array_chunk($message->images, $this->chunkSize, true);
+            foreach ($batches as $batch) {
+                $this->processBatch($batch);
+            }
+        } finally {
+            gc_collect_cycles();
+        }
+    }
+
+    private function processBatch(array $batch): void
+    {
+        $conn = $this->entityManager->getConnection();
+        $conn->getNativeConnection(); // calls connect() internally
+
+        /** @var string[] $filesToDelete */
+        $filesToDelete = $conn->transactional(function () use ($batch) {
+            $filesToDelete = [];
+            $hashes = array_map(fn ($str) => hex2bin($str), array_keys($batch));
+            $images = $this->imageRepository->findMultipleBySha256AndLock($hashes);
+
+            // images which not exist in DB can be deleted
+            foreach ($batch as $hash => $filepath) {
+                $hashBin = hex2bin($hash);
+                if (!array_any($images, fn ($img) => $img->sha256 === $hashBin)) {
+                    $filesToDelete[] = $filepath;
+                }
+            }
+
+            // images which are not referenced can be deleted
+            $referenced = $this->imageRepository->areImagesReferenced($images);
+            foreach ($referenced as $imgId => $isReferenced) {
+                if (!$isReferenced) {
+                    $img = array_find($images, fn ($img) => $img->getId() === $imgId);
+                    \assert($img instanceof Image);
+
+                    $filesToDelete[] = $img->filePath;
+                    $this->entityManager->remove($img);
+                }
+            }
+
+            $this->entityManager->flush();
+            return $filesToDelete;
+        });
+
+        foreach ($filesToDelete as $path) {
+            try {
+                $this->imageManager->remove($path);
+            } catch (\Exception $e) {
+                $this->logger->error('[DeleteImageV2Handler]: an error occurred when deleting an image file: {type} - {message}', [
+                    'message' => $e->getMessage(),
+                    'type' => \get_class($e),
+                ]);
+            }
+        }
+    }
+}

src/MessageHandler/DeleteUserHandler.php

@@ -8,6 +8,7 @@
 use App\Entity\User;
 use App\Message\ActivityPub\Outbox\DeliverMessage;
 use App\Message\Contracts\MessageInterface;
+use App\Message\DeleteImageV2Message;
 use App\Message\DeleteUserMessage;
 use App\Service\ActivityPub\ActivityJsonBuilder;
 use App\Service\ActivityPub\Wrapper\DeleteWrapper;
@@ -83,13 +84,11 @@ public function doWork(MessageInterface $message): void
         } catch (\Exception|\Error $e) {
             $this->logger->error("[ClearDeletedUserHandler::__invoke] Couldn't delete the cover of {user} at '{path}': {message}", ['user' => $user->username, 'path' => $user->cover?->filePath, 'message' => \get_class($e).': '.$e->getMessage()]);
         }
-        $filePathsOfUser = $this->userManager->getAllImageFilePathsOfUser($user);
-        foreach ($filePathsOfUser as $path) {
-            try {
-                $this->imageManager->remove($path);
-            } catch (\Exception|\Error $e) {
-                $this->logger->error("[ClearDeletedUserHandler::__invoke] Couldn't delete image of {user} at '{path}': {message}", ['user' => $user->username, 'path' => $path, 'message' => \get_class($e).': '.$e->getMessage()]);
-            }
+
+        $imagesOfUser = $this->userManager->getAllImagesShaAndFilepathOfUser($user);
+        $deleteImagesPayload = [];
+        foreach ($imagesOfUser as $row) {
+            $deleteImagesPayload[$row[0]] = $row[1];
         }
 
         $this->entityManager->beginTransaction();
@@ -122,6 +121,10 @@ public function doWork(MessageInterface $message): void
 
             throw $e;
         }
+
+        // dispatch at end or else reference-check would keep images
+        // because of the reference check this call can be safely placed outside the try{}
+        $this->bus->dispatch(new DeleteImageV2Message($deleteImagesPayload));
     }
 
     private function sendDeleteMessages(array $targetInboxes, User $deletedUser): void

src/Repository/ImageRepository.php

@@ -13,8 +13,11 @@
 use App\Pagination\Transformation\ContentPopulationTransformer;
 use App\Service\ImageManagerInterface;
 use App\Utils\ImageOrigin;
+use App\Utils\SqlHelpers;
 use Doctrine\Bundle\DoctrineBundle\Repository\ServiceEntityRepository;
+use Doctrine\DBAL\ArrayParameterType;
 use Doctrine\DBAL\Exception;
+use Doctrine\DBAL\LockMode;
 use Doctrine\Persistence\ManagerRegistry;
 use kornrunner\Blurhash\Blurhash;
 use Psr\EventDispatcher\EventDispatcherInterface;
@@ -40,6 +43,21 @@ public function __construct(
         parent::__construct($registry, Image::class);
     }
 
+    /**
+     * @param array $hashes array of hashes (binary)
+     *
+     * @return Image[]
+     */
+    public function findMultipleBySha256AndLock(array $hashes): array
+    {
+        return $this->createQueryBuilder('i')
+            ->where('i.sha256 IN (:hashes)')
+            ->setParameter('hashes', $hashes, ArrayParameterType::BINARY)
+            ->getQuery()
+            ->setLockMode(LockMode::PESSIMISTIC_READ)
+            ->getResult();
+    }
+
     /**
      * Process and store an uploaded image.
      *
@@ -305,4 +323,47 @@ public function redownloadImagesIfNecessary(array $images): void
         }
         $this->getEntityManager()->flush();
     }
+
+    /**
+     * @param Image[] $images
+     *
+     * @return array<int, bool> image id => is referenced
+     */
+    public function areImagesReferenced(array $images): array
+    {
+        $sql = '
+            SELECT i.id, (CASE WHEN
+                e.image_id IS NOT NULL
+                OR ec.image_id IS NOT NULL
+                OR p.image_id IS NOT NULL
+                OR pc.image_id IS NOT NULL
+                OR u.avatar_id IS NOT NULL
+                OR u.cover_id IS NOT NULL
+                OR m.icon_id IS NOT NULL
+                OR m.banner_id IS NOT NULL
+                OR oc.image_id IS NOT NULL
+            THEN TRUE ELSE FALSE END) AS referenced FROM image i
+            LEFT OUTER JOIN entry e ON i.id = e.image_id
+            LEFT OUTER JOIN entry_comment ec ON i.id = ec.image_id
+            LEFT OUTER JOIN post p ON i.id = p.image_id
+            LEFT OUTER JOIN post_comment pc ON i.id = pc.image_id
+            LEFT OUTER JOIN "user" u ON i.id = u.avatar_id OR i.id = u.cover_id
+            LEFT OUTER JOIN magazine m ON i.id = m.icon_id OR i.id = m.banner_id
+            LEFT OUTER JOIN oauth2_client oc ON i.id = oc.image_id
+            WHERE i.id IN (:ids);
+        ';
+
+        $sql = SqlHelpers::rewriteArrayParameters(['ids' => array_map(fn ($img) => $img->getId(), $images)], $sql);
+        $stmt = $this->getEntityManager()->getConnection()->prepare($sql['sql']);
+        foreach ($sql['parameters'] as $param => $value) {
+            $stmt->bindValue($param, $value, SqlHelpers::getSqlType($value));
+        }
+
+        $rows = $stmt->executeQuery()->fetchAllAssociative();
+        $ret = [];
+        foreach ($rows as $row) {
+            $ret[$row['id']] = $row['referenced'];
+        }
+        return $ret;
+    }
 }