diff --git a/app/console b/app/console index b1a81a390a..af3d741ae8 160000 --- a/app/console +++ b/app/console @@ -1 +1 @@ -Subproject commit b1a81a390a05746701651fca49e0d853f430677c +Subproject commit af3d741ae8f02c2e16b8b4ea4664a3f8970290fd diff --git a/app/controllers/api/edge.php b/app/controllers/api/edge.php index 1450e54aaa..d7ef222b1e 100644 --- a/app/controllers/api/edge.php +++ b/app/controllers/api/edge.php @@ -39,9 +39,9 @@ App::post('/v1/edge/sync') ->inject('queueForCacheSyncIn') ->action(function (array $keys, Request $request, Response $response, Client $queueForCacheSyncIn) { - //if (empty($keys)) { + if (empty($keys)) { throw new Exception(Exception::KEY_NOT_FOUND); - //} + } foreach ($keys as $sync) { $queueForCacheSyncIn diff --git a/app/workers/sync-out.php b/app/workers/sync-out.php index 9dc3092e19..39e76a5d77 100644 --- a/app/workers/sync-out.php +++ b/app/workers/sync-out.php @@ -35,9 +35,9 @@ const MAX_CURL_SEND_ATTEMPTS = 4; * @param string $url * @param string $token * @param array $payload - * @return array + * @return int */ -function call(string $url, string $token, array $payload): array +function call(string $url, string $token, array $payload): int { $ch = curl_init($url); @@ -82,12 +82,13 @@ function handle($dbForConsole, $regions, $payload): void $status = call($region['domain'] . '/v1/edge/sync', $token, ['keys' => $payload]); if ($status !== Response::STATUS_CODE_OK) { Console::error("[{$time}] Request to {$code} has failed"); + foreach ($payload as $sync) { $dbForConsole->createDocument('syncs', new Document([ 'region' => App::getEnv('_APP_REGION'), 'target' => $code, 'type' => $sync['type'], - 'key' => $sync['key'], + 'key' => ['key' => $sync['key']], 'status' => $status, ])); } @@ -100,11 +101,6 @@ $server->job() ->action(function (Message $message) use (&$stack, &$failures) { $payload = $message->getPayload() ?? []; - $type = $payload['type'] ?? null; - - if (empty($type)) { - return; - } //Get failed requests if (!empty($payload['region']) && !empty($payload['keys'])) { @@ -122,6 +118,10 @@ $server->job() return; } + if (empty($payload['type'])) { + return; + } + if (!empty($payload['key'])) { $stack['keys'][] = [ 'type' => $payload['type'], @@ -154,9 +154,8 @@ $server } return; } - //var_dump($stack['keys']); + $chunk = array_slice($stack['keys'], 0, CHUNK_MAX_KEYS, true); - //var_dump($chunk); array_splice($stack['keys'], 0, CHUNK_MAX_KEYS); Console::log("[{$time}] Sending " . count($chunk) . " remains " . count($stack['keys'])); handle($dbForConsole, $stack['regions'], $chunk); diff --git a/src/Appwrite/Platform/Tasks/EdgeSync.php b/src/Appwrite/Platform/Tasks/EdgeSync.php index cdc99136b1..7ffd8e24f2 100644 --- a/src/Appwrite/Platform/Tasks/EdgeSync.php +++ b/src/Appwrite/Platform/Tasks/EdgeSync.php @@ -47,7 +47,7 @@ class EdgeSync extends Action Console::success("[{$time}] New task every {$interval} seconds"); - foreach ($regions as $target) { + foreach ($regions as $code => $region) { $count = 0; $chunk = 0; $limit = 50; @@ -58,16 +58,17 @@ class EdgeSync extends Action $results = $dbForConsole->find('syncs', [ Query::equal('region', [App::getEnv('_APP_REGION')]), - Query::equal('target', [$target]), + Query::equal('target', [$code]), Query::limit($limit) ]); $sum = count($results); if ($sum > 0) { foreach ($results as $document) { + $key = $document->getAttribute('key'); $keys[] = [ 'type' => $document->getAttribute('type'), - 'key' => $document->getAttribute('key') + 'key' => $key['key'] ]; $dbForConsole->deleteDocument('syncs', $document->getId()); $count++; @@ -76,14 +77,14 @@ class EdgeSync extends Action } if (!empty($keys)) { - Console::info("[{$time}] Enqueueing keys chunk {$count} to {$target}"); + Console::info("[{$time}] Enqueueing keys chunk {$count} to region {$code}"); $queueForCacheSyncOut ->enqueue([ - 'region' => $target, + 'region' => $code, 'keys' => $keys ]); } else { - Console::info("[{$time}] No cache keys where found."); + Console::info("[{$time}] No keys where found for region {$code}."); } } }, $interval);