fix(transfer): abandon only stalled target requests, not slow ones

src/Migration/Target.php

@@ -16,8 +16,14 @@ abstract class Target
     public const int CONNECT_TIMEOUT = 10;
 
     /**
-     * Seconds a request may take in total. Requests are single API pages and
-     * file chunks, so a remote that has stopped answering fails within this.
+     * Seconds a request may go without receiving a byte before it is abandoned.
+     * Transfers of any length stay alive as long as data keeps arriving.
+     */
+    public const int STALL_TIMEOUT = 30;
+
+    /**
+     * Seconds an Appwrite SDK request may take in total. The SDK transport only
+     * takes a total timeout; its requests are single API pages and file chunks.
      */
     public const int REQUEST_TIMEOUT = 30;
 
@@ -156,10 +162,13 @@ protected function call(
      */
     protected function client(): Client
     {
-        return $this->client ??= (new Client(new CurlAdapter()))
+        return $this->client ??= (new Client(new CurlAdapter(options: [
+            \CURLOPT_TIMEOUT_MS => 0,
+            \CURLOPT_LOW_SPEED_LIMIT => 1,
+            \CURLOPT_LOW_SPEED_TIME => static::STALL_TIMEOUT,
+        ])))
             ->withFollowRedirects()
-            ->withConnectTimeout(static::CONNECT_TIMEOUT)
-            ->withTimeout(static::REQUEST_TIMEOUT);
+            ->withConnectTimeout(static::CONNECT_TIMEOUT);
     }
 
     /**

tests/Migration/Unit/General/ShortTimeoutTarget.php

@@ -0,0 +1,48 @@
+<?php
+
+namespace Utopia\Tests\Unit\General;
+
+use Override;
+use Utopia\Migration\Target;
+
+final class ShortTimeoutTarget extends Target
+{
+    public const int CONNECT_TIMEOUT = 1;
+
+    public const int STALL_TIMEOUT = 3;
+
+    public const int REQUEST_TIMEOUT = 1;
+
+    #[Override]
+    public static function getName(): string
+    {
+        return 'ShortTimeout';
+    }
+
+    #[Override]
+    public static function getSupportedResources(): array
+    {
+        return [];
+    }
+
+    #[Override]
+    public function run(array $resources, callable $callback, string $rootResourceId = ''): void
+    {
+    }
+
+    #[Override]
+    public function report(array $resources = [], array $resourceIds = []): array
+    {
+        return [];
+    }
+
+    /**
+     * @return array<mixed>|string
+     *
+     * @throws \Exception
+     */
+    public function fetch(string $url): array|string
+    {
+        return $this->call('GET', $url);
+    }
+}

tests/Migration/Unit/General/TargetTimeoutServerBehaviour.php

@@ -0,0 +1,14 @@
+<?php
+
+namespace Utopia\Tests\Unit\General;
+
+enum TargetTimeoutServerBehaviour: string
+{
+    public const int STREAMED_BYTES = 20;
+
+    public const int STREAM_INTERVAL_MICROSECONDS = 200_000;
+
+    case Streaming = 'streaming';
+    case Stalled = 'stalled';
+    case Silent = 'silent';
+}

tests/Migration/Unit/General/TargetTimeoutTest.php

@@ -0,0 +1,101 @@
+<?php
+
+namespace Utopia\Tests\Unit\General;
+
+use Override;
+use PHPUnit\Framework\Attributes\DataProvider;
+use PHPUnit\Framework\TestCase;
+use Utopia\Migration\Exception;
+use Utopia\Tests\Unit\General\TargetTimeoutServerBehaviour as Behaviour;
+
+final class TargetTimeoutTest extends TestCase
+{
+    private const string HTTP = 'http';
+
+    private const string HTTPS = 'https';
+
+    private const float GIVE_UP_WITHIN_SECONDS = 10.0;
+
+    private const float TIMER_TOLERANCE_SECONDS = 0.1;
+
+    /** @var resource|null */
+    private $server = null;
+
+    #[Override]
+    protected function tearDown(): void
+    {
+        if (\is_resource($this->server)) {
+            \proc_terminate($this->server, 9);
+            \proc_close($this->server);
+        }
+
+        $this->server = null;
+    }
+
+    public function testTransferOutlastingEveryTimeoutSucceedsWhileDataKeepsArriving(): void
+    {
+        $url = $this->serve(Behaviour::Streaming, self::HTTP);
+
+        $started = \microtime(true);
+        $body = (new ShortTimeoutTarget())->fetch($url);
+        $elapsed = \microtime(true) - $started;
+
+        $this->assertSame(\str_repeat('x', Behaviour::STREAMED_BYTES), $body);
+        $this->assertGreaterThan(
+            \max(ShortTimeoutTarget::REQUEST_TIMEOUT, ShortTimeoutTarget::STALL_TIMEOUT),
+            $elapsed,
+            'The transfer must outlast the request and stall timeouts to show neither caps a progressing transfer.',
+        );
+    }
+
+    /**
+     * @return array<string, array{Behaviour, string, int, float}>
+     */
+    public static function silentRemotes(): array
+    {
+        return [
+            'accepts and never answers' => [Behaviour::Silent, self::HTTP, ShortTimeoutTarget::STALL_TIMEOUT, self::GIVE_UP_WITHIN_SECONDS],
+            'stops sending mid-body' => [Behaviour::Stalled, self::HTTP, ShortTimeoutTarget::STALL_TIMEOUT, self::GIVE_UP_WITHIN_SECONDS],
+            'never completes the connection' => [Behaviour::Silent, self::HTTPS, ShortTimeoutTarget::CONNECT_TIMEOUT, ShortTimeoutTarget::STALL_TIMEOUT],
+        ];
+    }
+
+    #[DataProvider('silentRemotes')]
+    public function testSilentRemoteFailsOnItsTimeout(Behaviour $behaviour, string $scheme, int $timeout, float $before): void
+    {
+        $url = $this->serve($behaviour, $scheme);
+
+        $started = \microtime(true);
+        $error = null;
+
+        try {
+            (new ShortTimeoutTarget())->fetch($url);
+        } catch (\Exception $caught) {
+            $error = $caught;
+        }
+
+        $elapsed = \microtime(true) - $started;
+
+        $this->assertInstanceOf(\Exception::class, $error, 'A remote that sends nothing must fail the request.');
+        $this->assertSame(Exception::CODE_INTERNAL, $error->getCode(), $error->getMessage());
+        $this->assertGreaterThanOrEqual(
+            $timeout - self::TIMER_TOLERANCE_SECONDS,
+            $elapsed,
+            'The request must wait out its timeout rather than fail for another reason: '.$error->getMessage(),
+        );
+        $this->assertLessThan($before, $elapsed, 'The request must fail on its own timeout: '.$error->getMessage());
+    }
+
+    private function serve(Behaviour $behaviour, string $scheme): string
+    {
+        $script = \dirname(__DIR__, 2).'/resources/timeout/server.php';
+        $server = \proc_open([PHP_BINARY, $script, $behaviour->value], [1 => ['pipe', 'w']], $pipes);
+        $this->assertIsResource($server);
+        $this->server = $server;
+
+        $port = \trim((string) \fgets($pipes[1]));
+        $this->assertMatchesRegularExpression('/^\d+$/', $port, 'The test server did not report its port.');
+
+        return "{$scheme}://127.0.0.1:{$port}/";
+    }
+}

tests/Migration/resources/timeout/server.php

@@ -0,0 +1,50 @@
+<?php
+
+require \dirname(__DIR__, 4).'/vendor/autoload.php';
+
+use Utopia\Tests\Unit\General\TargetTimeoutServerBehaviour as Behaviour;
+
+const LIFETIME_SECONDS = 120;
+
+$behaviour = Behaviour::from($argv[1] ?? '');
+$server = \stream_socket_server('tcp://127.0.0.1:0', $errorCode, $errorMessage);
+
+if ($server === false) {
+    \fwrite(STDERR, $errorMessage);
+    exit(1);
+}
+
+$address = (string) \stream_socket_get_name($server, false);
+\fwrite(STDOUT, \substr($address, \strrpos($address, ':') + 1).PHP_EOL);
+\fflush(STDOUT);
+
+$connection = \stream_socket_accept($server, LIFETIME_SECONDS);
+
+if ($connection === false) {
+    exit(1);
+}
+
+if ($behaviour === Behaviour::Silent) {
+    \sleep(LIFETIME_SECONDS);
+    exit(0);
+}
+
+$request = '';
+while (!\str_contains($request, "\r\n\r\n") && !\feof($connection)) {
+    $request .= (string) \fread($connection, 8192);
+}
+
+\fwrite($connection, "HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: ".Behaviour::STREAMED_BYTES."\r\nConnection: close\r\n\r\n");
+
+if ($behaviour === Behaviour::Stalled) {
+    \fwrite($connection, 'x');
+    \sleep(LIFETIME_SECONDS);
+    exit(0);
+}
+
+for ($i = 0; $i < Behaviour::STREAMED_BYTES; $i++) {
+    \usleep(Behaviour::STREAM_INTERVAL_MICROSECONDS);
+    \fwrite($connection, 'x');
+}
+
+\fclose($connection);