mirror of
https://github.com/appwrite/appwrite.git
synced 2026-05-26 13:51:13 +00:00
Fix parallel deployment chunk uploads
This commit is contained in:
Torsten Dittmann
@@ -1191,6 +1191,144 @@ class FunctionsCustomServerTest extends Scope
|
||||
}, 120000, 500);
|
||||
}
|
||||
|
||||
public function testCreateDeploymentParallelChunksLargeFile(): void
|
||||
{
|
||||
$functionId = $this->setupFunction([
|
||||
'functionId' => ID::unique(),
|
||||
'name' => 'Test Parallel Chunk Deployment',
|
||||
'execute' => [Role::user($this->getUser()['$id'])->toString()],
|
||||
'runtime' => 'node-22',
|
||||
'entrypoint' => 'index.js',
|
||||
'timeout' => 10,
|
||||
]);
|
||||
|
||||
$deploymentId = ID::unique();
|
||||
$tmpDirectory = sys_get_temp_dir() . DIRECTORY_SEPARATOR . 'appwrite-parallel-function-deployment-' . $deploymentId;
|
||||
|
||||
mkdir($tmpDirectory);
|
||||
|
||||
try {
|
||||
copy(__DIR__ . '/../../../resources/functions/basic/index.js', $tmpDirectory . DIRECTORY_SEPARATOR . 'index.js');
|
||||
file_put_contents($tmpDirectory . DIRECTORY_SEPARATOR . 'large.bin', random_bytes(20 * 1024 * 1024));
|
||||
|
||||
$source = $tmpDirectory . DIRECTORY_SEPARATOR . 'code.tar.gz';
|
||||
Console::execute('cd ' . $tmpDirectory . ' && tar --exclude code.tar.gz -czf code.tar.gz .', '', $this->stdout, $this->stderr);
|
||||
|
||||
$totalSize = filesize($source);
|
||||
$chunkSize = 5 * 1024 * 1024;
|
||||
$chunksTotal = (int) ceil($totalSize / $chunkSize);
|
||||
|
||||
$this->assertGreaterThanOrEqual(4, $chunksTotal, 'Test deployment must span at least 4 chunks');
|
||||
|
||||
$requests = [];
|
||||
$sourceHandle = fopen($source, 'rb');
|
||||
$this->assertNotFalse($sourceHandle, 'Could not open deployment package');
|
||||
|
||||
try {
|
||||
for ($i = 0; $i < $chunksTotal; $i++) {
|
||||
$start = $i * $chunkSize;
|
||||
$end = min($start + $chunkSize, $totalSize) - 1;
|
||||
$length = $end - $start + 1;
|
||||
$chunkPath = $tmpDirectory . DIRECTORY_SEPARATOR . 'chunk-' . $i . '.part';
|
||||
|
||||
fseek($sourceHandle, $start);
|
||||
file_put_contents($chunkPath, fread($sourceHandle, $length));
|
||||
|
||||
$requests[] = [
|
||||
'headers' => [
|
||||
'x-appwrite-project' => $this->getProject()['$id'],
|
||||
'x-appwrite-key' => $this->getProject()['apiKey'],
|
||||
'x-appwrite-id' => $deploymentId,
|
||||
'content-range' => 'bytes ' . $start . '-' . $end . '/' . $totalSize,
|
||||
],
|
||||
'chunkPath' => $chunkPath,
|
||||
];
|
||||
}
|
||||
} finally {
|
||||
fclose($sourceHandle);
|
||||
}
|
||||
|
||||
$responses = [];
|
||||
$endpoint = parse_url($this->client->getEndpoint());
|
||||
$scheme = $endpoint['scheme'] ?? 'http';
|
||||
$host = $endpoint['host'] ?? 'appwrite';
|
||||
$port = $endpoint['port'] ?? ($scheme === 'https' ? 443 : 80);
|
||||
$basePath = rtrim($endpoint['path'] ?? '', '/');
|
||||
|
||||
\Swoole\Coroutine\run(function () use ($basePath, $functionId, $host, $port, $requests, $scheme, &$responses): void {
|
||||
$wg = new \Swoole\Coroutine\WaitGroup();
|
||||
|
||||
foreach ($requests as $index => $request) {
|
||||
$wg->add();
|
||||
\Swoole\Coroutine::create(function () use ($basePath, $functionId, $host, $index, $port, $request, &$responses, $scheme, $wg): void {
|
||||
try {
|
||||
for ($attempt = 0; $attempt < 3; $attempt++) {
|
||||
$client = new \Swoole\Coroutine\Http\Client($host, (int) $port, $scheme === 'https');
|
||||
$client->set([
|
||||
'timeout' => 300,
|
||||
'ssl_verify_peer' => false,
|
||||
'ssl_verify_host' => false,
|
||||
]);
|
||||
$client->setHeaders($request['headers']);
|
||||
$client->setMethod(Client::METHOD_POST);
|
||||
$client->setData([
|
||||
'entrypoint' => 'index.js',
|
||||
'activate' => true,
|
||||
]);
|
||||
$client->addFile($request['chunkPath'], 'code', 'application/x-gzip', 'code.tar.gz');
|
||||
$client->execute($basePath . '/functions/' . $functionId . '/deployments');
|
||||
|
||||
$responses[$index] = [
|
||||
'body' => $client->body,
|
||||
'error' => $client->errMsg,
|
||||
'headers' => $client->headers ?? [],
|
||||
'statusCode' => $client->statusCode,
|
||||
];
|
||||
|
||||
$client->close();
|
||||
|
||||
if ($responses[$index]['statusCode'] !== 429) {
|
||||
break;
|
||||
}
|
||||
|
||||
$retryAfter = (float) ($responses[$index]['headers']['retry-after'] ?? 0.1);
|
||||
\Swoole\Coroutine::sleep(max($retryAfter, 0.1));
|
||||
}
|
||||
} finally {
|
||||
$wg->done();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
$wg->wait();
|
||||
});
|
||||
|
||||
ksort($responses);
|
||||
|
||||
foreach ($responses as $response) {
|
||||
$this->assertSame('', $response['error']);
|
||||
$this->assertContains($response['statusCode'], [202], (string) $response['body']);
|
||||
}
|
||||
|
||||
$this->assertEventually(function () use ($functionId, $deploymentId) {
|
||||
$deployment = $this->getDeployment($functionId, $deploymentId);
|
||||
|
||||
$this->assertEquals(200, $deployment['headers']['status-code']);
|
||||
$this->assertEquals('ready', $deployment['body']['status']);
|
||||
$this->assertEquals($deploymentId, $deployment['body']['$id']);
|
||||
}, 120000, 500);
|
||||
} finally {
|
||||
$this->cleanupFunction($functionId);
|
||||
|
||||
if (is_dir($tmpDirectory)) {
|
||||
foreach (glob($tmpDirectory . DIRECTORY_SEPARATOR . '*') ?: [] as $file) {
|
||||
unlink($file);
|
||||
}
|
||||
rmdir($tmpDirectory);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public function testUpdateDeployment(): void
|
||||
{
|
||||
$data = $this->setupTestDeployment();
|
||||
|
||||
@@ -1034,6 +1034,145 @@ class SitesCustomServerTest extends Scope
|
||||
$this->cleanupSite($siteId);
|
||||
}
|
||||
|
||||
public function testCreateDeploymentParallelChunksLargeFile(): void
|
||||
{
|
||||
$siteId = $this->setupSite([
|
||||
'buildRuntime' => 'node-22',
|
||||
'fallbackFile' => '',
|
||||
'framework' => 'other',
|
||||
'name' => 'Test Site Parallel Chunk Deployment',
|
||||
'outputDirectory' => './',
|
||||
'providerBranch' => 'main',
|
||||
'providerRootDirectory' => './',
|
||||
'siteId' => ID::unique()
|
||||
]);
|
||||
|
||||
$deploymentId = ID::unique();
|
||||
$tmpDirectory = sys_get_temp_dir() . DIRECTORY_SEPARATOR . 'appwrite-parallel-site-deployment-' . $deploymentId;
|
||||
|
||||
mkdir($tmpDirectory);
|
||||
|
||||
try {
|
||||
file_put_contents($tmpDirectory . DIRECTORY_SEPARATOR . 'index.html', '<html><body>Hello World</body></html>');
|
||||
file_put_contents($tmpDirectory . DIRECTORY_SEPARATOR . 'large.bin', random_bytes(20 * 1024 * 1024));
|
||||
|
||||
$source = $tmpDirectory . DIRECTORY_SEPARATOR . 'code.tar.gz';
|
||||
Console::execute('cd ' . $tmpDirectory . ' && tar --exclude code.tar.gz -czf code.tar.gz .', '', $this->stdout, $this->stderr);
|
||||
|
||||
$totalSize = filesize($source);
|
||||
$chunkSize = 5 * 1024 * 1024;
|
||||
$chunksTotal = (int) ceil($totalSize / $chunkSize);
|
||||
|
||||
$this->assertGreaterThanOrEqual(4, $chunksTotal, 'Test deployment must span at least 4 chunks');
|
||||
|
||||
$requests = [];
|
||||
$sourceHandle = fopen($source, 'rb');
|
||||
$this->assertNotFalse($sourceHandle, 'Could not open deployment package');
|
||||
|
||||
try {
|
||||
for ($i = 0; $i < $chunksTotal; $i++) {
|
||||
$start = $i * $chunkSize;
|
||||
$end = min($start + $chunkSize, $totalSize) - 1;
|
||||
$length = $end - $start + 1;
|
||||
$chunkPath = $tmpDirectory . DIRECTORY_SEPARATOR . 'chunk-' . $i . '.part';
|
||||
|
||||
fseek($sourceHandle, $start);
|
||||
file_put_contents($chunkPath, fread($sourceHandle, $length));
|
||||
|
||||
$requests[] = [
|
||||
'headers' => [
|
||||
'x-appwrite-project' => $this->getProject()['$id'],
|
||||
'x-appwrite-key' => $this->getProject()['apiKey'],
|
||||
'x-appwrite-id' => $deploymentId,
|
||||
'content-range' => 'bytes ' . $start . '-' . $end . '/' . $totalSize,
|
||||
],
|
||||
'chunkPath' => $chunkPath,
|
||||
];
|
||||
}
|
||||
} finally {
|
||||
fclose($sourceHandle);
|
||||
}
|
||||
|
||||
$responses = [];
|
||||
$endpoint = parse_url($this->client->getEndpoint());
|
||||
$scheme = $endpoint['scheme'] ?? 'http';
|
||||
$host = $endpoint['host'] ?? 'appwrite';
|
||||
$port = $endpoint['port'] ?? ($scheme === 'https' ? 443 : 80);
|
||||
$basePath = rtrim($endpoint['path'] ?? '', '/');
|
||||
|
||||
\Swoole\Coroutine\run(function () use ($basePath, $host, $port, $requests, $scheme, $siteId, &$responses): void {
|
||||
$wg = new \Swoole\Coroutine\WaitGroup();
|
||||
|
||||
foreach ($requests as $index => $request) {
|
||||
$wg->add();
|
||||
\Swoole\Coroutine::create(function () use ($basePath, $host, $index, $port, $request, &$responses, $scheme, $siteId, $wg): void {
|
||||
try {
|
||||
for ($attempt = 0; $attempt < 3; $attempt++) {
|
||||
$client = new \Swoole\Coroutine\Http\Client($host, (int) $port, $scheme === 'https');
|
||||
$client->set([
|
||||
'timeout' => 300,
|
||||
'ssl_verify_peer' => false,
|
||||
'ssl_verify_host' => false,
|
||||
]);
|
||||
$client->setHeaders($request['headers']);
|
||||
$client->setMethod(Client::METHOD_POST);
|
||||
$client->setData([
|
||||
'activate' => true,
|
||||
]);
|
||||
$client->addFile($request['chunkPath'], 'code', 'application/x-gzip', 'code.tar.gz');
|
||||
$client->execute($basePath . '/sites/' . $siteId . '/deployments');
|
||||
|
||||
$responses[$index] = [
|
||||
'body' => $client->body,
|
||||
'error' => $client->errMsg,
|
||||
'headers' => $client->headers ?? [],
|
||||
'statusCode' => $client->statusCode,
|
||||
];
|
||||
|
||||
$client->close();
|
||||
|
||||
if ($responses[$index]['statusCode'] !== 429) {
|
||||
break;
|
||||
}
|
||||
|
||||
$retryAfter = (float) ($responses[$index]['headers']['retry-after'] ?? 0.1);
|
||||
\Swoole\Coroutine::sleep(max($retryAfter, 0.1));
|
||||
}
|
||||
} finally {
|
||||
$wg->done();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
$wg->wait();
|
||||
});
|
||||
|
||||
ksort($responses);
|
||||
|
||||
foreach ($responses as $response) {
|
||||
$this->assertSame('', $response['error']);
|
||||
$this->assertContains($response['statusCode'], [202], (string) $response['body']);
|
||||
}
|
||||
|
||||
$this->assertEventually(function () use ($siteId, $deploymentId) {
|
||||
$deployment = $this->getDeployment($siteId, $deploymentId);
|
||||
|
||||
$this->assertEquals(200, $deployment['headers']['status-code']);
|
||||
$this->assertEquals('ready', $deployment['body']['status']);
|
||||
$this->assertEquals($deploymentId, $deployment['body']['$id']);
|
||||
}, 120000, 500);
|
||||
} finally {
|
||||
$this->cleanupSite($siteId);
|
||||
|
||||
if (is_dir($tmpDirectory)) {
|
||||
foreach (glob($tmpDirectory . DIRECTORY_SEPARATOR . '*') ?: [] as $file) {
|
||||
unlink($file);
|
||||
}
|
||||
rmdir($tmpDirectory);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public function testCreateDeployment()
|
||||
{
|
||||
$siteId = $this->setupSite([
|
||||
|
||||
@@ -1376,7 +1376,7 @@ trait StorageBase
|
||||
|
||||
public function testCreateBucketFileParallelChunksLargeFile(): void
|
||||
{
|
||||
$totalSize = (int) ($_ENV['APPWRITE_TEST_PARALLEL_UPLOAD_SIZE'] ?? 20 * 1024 * 1024);
|
||||
$totalSize = 20 * 1024 * 1024;
|
||||
$chunkSize = 5 * 1024 * 1024;
|
||||
$chunksTotal = (int) ceil($totalSize / $chunkSize);
|
||||
|
||||
|
||||
Reference in New Issue
Block a user