Poll again right after a full batch instead of waiting for the next signal

src/Internal/functions.php

@@ -42,6 +42,7 @@ function consume(
         $watcher,
         $completionMarker,
         $context,
+        $polls,
     ): void {
         $watcher->watch();
 
@@ -66,6 +67,12 @@ function consume(
                 }
 
                 $tx->commit();
+
+                // A full batch means the queue may hold more messages. Inserts that already happened will not notify
+                // again, so waiting for the next signal would leave the backlog idle for up to $pollInterval.
+                if (\count($messages) === $config->batch && !$polls->isComplete()) {
+                    $polls->pushAsync(null)->ignore();
+                }
             } catch (\Throwable $e) {
                 $tx->rollback();
                 $completionMarker->error($e);

tests/PgmqTest.php

@@ -8,6 +8,7 @@
 use Amp\Postgres\PostgresConnection;
 use Amp\Postgres\PostgresConnectionPool;
 use Amp\Postgres\PostgresQueryError;
+use Amp\TimeoutCancellation;
 use PHPUnit\Framework\Attributes\CoversClass;
 use PHPUnit\Framework\Attributes\DoesNotPerformAssertions;
 use PHPUnit\Framework\TestCase;
@@ -326,6 +327,38 @@ static function (array $messages, ConsumeController $ctrl) use (&$consumed): voi
         self::assertSame(0, $queue->metrics()->length);
     }
 
+    public function testConsumerDrainsBacklogWithoutWaitingForPollInterval(): void
+    {
+        $queue = createQueue($this->pg, $this->randomQueueName());
+        $queue->sendBatch(array_fill(0, 25, new SendMessage(self::TESTING_MESSAGE)));
+
+        $consumed = 0;
+
+        $consumer = createConsumer($this->pg);
+        $context = $consumer->consume(
+            static function (array $messages, ConsumeController $ctrl) use (&$consumed): void {
+                $consumed += \count($messages);
+                $ctrl->ack($messages);
+
+                if ($consumed === 25) {
+                    $ctrl->stop();
+                }
+            },
+            new ConsumeConfig(
+                queue: $queue->name,
+                batch: 10,
+                pollInterval: TimeSpan::fromSeconds(10),
+            ),
+        );
+
+        // The backlog is already there: no further inserts will wake the consumer up, so a full batch must
+        // trigger the next read immediately instead of waiting for the poll interval (2 x 10 s for 25 messages).
+        $context->awaitCompletion(new TimeoutCancellation(2));
+
+        self::assertSame(25, $consumed);
+        self::assertSame(0, $queue->metrics()->length);
+    }
+
     public function testNackBatch(): void
     {
         $queue = createQueue($this->pg, $this->randomQueueName());