From 948da00f0cb1efc57b6bd13f18f79358c8c34a8e Mon Sep 17 00:00:00 2001 From: Shlee Date: Sun, 30 Aug 2026 13:56:31 +0930 Subject: [PATCH] Update EmojiMoveStorageLocalToCloud.php --- .../Admin/EmojiMoveStorageLocalToCloud.php | 447 +++++------------- 1 file changed, 121 insertions(+), 326 deletions(-) diff --git a/app/Console/Commands/Admin/EmojiMoveStorageLocalToCloud.php b/app/Console/Commands/Admin/EmojiMoveStorageLocalToCloud.php index 15beb063f..462c8e294 100644 --- a/app/Console/Commands/Admin/EmojiMoveStorageLocalToCloud.php +++ b/app/Console/Commands/Admin/EmojiMoveStorageLocalToCloud.php @@ -4,14 +4,12 @@ namespace App\Console\Commands\Admin; use App\Console\Commands\Concerns\ManagesMediaStorageEnv; use App\Models\CustomEmoji; -use App\Util\Lexer\PrettyNumber; use Aws\CommandPool; use Aws\S3\S3Client; use Illuminate\Console\Command; use Illuminate\Support\Facades\Cache; use Illuminate\Support\Facades\Storage; use Illuminate\Support\Str; -use Symfony\Component\Process\Process; class EmojiMoveStorageLocalToCloud extends Command { @@ -25,15 +23,10 @@ class EmojiMoveStorageLocalToCloud extends Command protected $signature = 'admin:EmojiMoveStorageLocalToCloud {--limit=0 : Max files to process this run (0 = no limit, process all)} {--offset=0 : Skip this many files before processing (for manual chunking)} - {--workers=1 : Spawn N parallel worker processes to upload concurrently} - {--concurrency=0 : Use the async S3 SDK with N concurrent in-flight uploads (0 = disabled, use sync path)} - {--no-acl : Do not send an ACL header on async uploads (some S3-compatible stores reject it)} - {--stride=1 : Internal: total worker count for strided sharding} - {--shard=0 : Internal: this worker index (0-based) for strided sharding} - {--skip-verify : Do not re-check the cloud copy size after upload (faster)} - {--skip-cloud-check : Do not HEAD the cloud object first; always upload (faster, idempotent)} - {--dry-run : Report what would happen without copying or writing} - {--keep-local : Do not delete local files after verifying the cloud copy} + {--concurrency=100 : Concurrent in-flight S3 uploads via the async SDK (0 = simple synchronous fallback)} + {--no-acl : Do not send an ACL header on uploads (some S3-compatible stores reject it)} + {--keep-local : Do not delete local files after a successful upload} + {--dry-run : Report what would happen without uploading or deleting} {--debug : Print detailed diagnostics} {--force : Skip confirmation prompts}'; @@ -42,9 +35,7 @@ class EmojiMoveStorageLocalToCloud extends Command * * @var string */ - protected $description = 'Migrate local custom emoji to cloud storage: copy up, verify by size, then delete the local copy.'; - - protected int $movedBytes = 0; + protected $description = 'Migrate local custom emoji to cloud storage (concurrent async uploads), then delete the local copy.'; public function handle(): int { @@ -61,9 +52,8 @@ class EmojiMoveStorageLocalToCloud extends Command try { $localDisk = Storage::disk('local'); - $cloudDisk = Storage::disk(config('filesystems.cloud')); } catch (\Throwable $e) { - $this->error('Cloud disk ('.config('filesystems.cloud').') could not be resolved: '.$e->getMessage()); + $this->error('Local disk could not be resolved: '.$e->getMessage()); return self::FAILURE; } @@ -75,237 +65,79 @@ class EmojiMoveStorageLocalToCloud extends Command return self::FAILURE; } - $debug = (bool) $this->option('debug'); - - // Diagnostics: what does the environment/config look like? - if ($debug) { - $this->line('--- debug: config ---'); - $this->line(' config(pixelfed.cloud_storage): '.var_export(config('pixelfed.cloud_storage'), true)); - $this->line(' config_cache(pixelfed.cloud_storage): '.var_export(config_cache('pixelfed.cloud_storage'), true)); - $this->line(' filesystems.cloud: '.config('filesystems.cloud')); - $this->line(' cloud host: '.($this->cloudHost() ?? 'null')); - $this->line(' local disk root: '.$localDisk->path('')); - - $this->line('--- debug: custom_emoji table ---'); - $this->line(' total rows: '.CustomEmoji::count()); - $this->line(' uri IS NULL: '.CustomEmoji::whereNull('uri')->count()); - $this->line(' uri NOT NULL: '.CustomEmoji::whereNotNull('uri')->count()); - $this->line(' media_path NOT NULL: '.CustomEmoji::whereNotNull('media_path')->count()); - $this->line(' media_path LIKE http%: '.CustomEmoji::where('media_path', 'like', 'http%')->count()); - - $this->line('--- debug: local emoji directory (public/emoji) ---'); - $this->line(' dir exists: '.var_export($localDisk->exists('public/emoji'), true)); - $this->line(' file count: '.count($localDisk->files('public/emoji'))); - } - - $concurrency = max(0, (int) $this->option('concurrency')); - - // Fastest path: async S3 SDK with many concurrent in-flight PutObject - // requests from a single process. This hides the high per-request - // latency of the object store far better than sequential uploads or - // process-level workers. - if ($concurrency > 0) { - return $this->runAsync($concurrency, $localDisk); + if ($this->option('debug')) { + $this->printDebug($localDisk); } - $workers = max(1, (int) $this->option('workers')); - - // When asked to parallelise, spawn N child processes that each handle a - // strided slice of the file list (worker w processes indexes where - // index % workers === w). This gives real concurrency for the - // I/O-bound S3 uploads without any extra dependencies. - if ($workers > 1) { - return $this->runParallel($workers); - } - - if (! $this->option('dry-run') && ! $this->option('force')) { - if (! $this->confirm('Begin migrating local custom emoji to cloud?', true)) { - $this->comment('Aborted.'); - - return self::SUCCESS; - } - } - - $limit = (int) $this->option('limit'); - $offset = max(0, (int) $this->option('offset')); - $stride = max(1, (int) $this->option('stride')); - $shard = max(0, (int) $this->option('shard')); - $moved = 0; - $skipped = 0; - $failed = 0; - $startedAt = microtime(true); - - // Disk-driven: enumerate the actual emoji files on the local disk and - // migrate each one. We intentionally do NOT filter by DB columns here — - // the file existing locally is the source of truth for "needs moving". - // Federated emoji have their media stored locally too (with a uri set), - // so a DB filter on uri would wrongly exclude them. - $files = $localDisk->exists('public/emoji') ? $localDisk->files('public/emoji') : []; - - if ($offset > 0) { - $files = array_slice($files, $offset); - } + // Build the list of local emoji files to migrate. Disk-driven: the file + // existing locally is the source of truth for "needs moving". We do NOT + // filter by DB columns — federated emoji have their media stored locally + // too (with a uri set), so a DB filter would wrongly exclude them. + $files = $this->collectFiles($localDisk); + $total = count($files); - if ($limit > 0) { - $files = array_slice($files, 0, $limit); - } + if ($total === 0) { + $this->info('No emoji files to migrate.'); - $showBar = $stride === 1 && ! $debug; - $bar = null; - if ($showBar) { - $bar = $this->output->createProgressBar(count($files)); - $bar->setFormat(" %current%/%max% [%bar%] %percent:3s%% %rate% up/s\n %message%"); - $bar->setMessage(''); - $bar->setMessage('0.0', 'rate'); - $bar->start(); + return self::SUCCESS; } - foreach ($files as $i => $localPath) { - // Strided sharding for parallel workers: only handle our slice. - if ($stride > 1 && ($i % $stride) !== $shard) { - continue; - } - - $filename = basename($localPath); - - // Preserve dotfiles such as a directory .gitignore, and the - // missing.png placeholder which the frontend hardcodes as a local - // /storage/emoji/missing.png onerror fallback (must stay local). - if (str_starts_with($filename, '.') || $filename === 'missing.png') { - $skipped++; - $bar?->advance(); - - continue; - } - - // media_path is the local path without the public/ disk prefix. - $mediaPath = Str::after($localPath, 'public/'); - $result = $this->migrateFile($localPath, $mediaPath, $localDisk, $cloudDisk, $debug); - match ($result) { - 'moved' => $moved++, - 'skipped' => $skipped++, - default => $failed++, - }; - - if ($bar) { - $liveElapsed = max(0.001, microtime(true) - $startedAt); - $bar->setMessage(sprintf('%.1f', $moved / $liveElapsed), 'rate'); - } - $bar?->advance(); - } + $concurrency = max(0, (int) $this->option('concurrency')); + $mode = $concurrency > 0 ? "async S3 (concurrency={$concurrency})" : 'synchronous'; - $bar?->finish(); - $this->newLine(2); + if ($this->option('dry-run')) { + $this->info("[dry-run] Would upload {$total} emoji to cloud using {$mode}."); - if ($moved > 0 && ! $this->option('dry-run')) { - Cache::forget('pf:custom_emoji'); + return self::SUCCESS; } - $elapsed = max(0.001, microtime(true) - $startedAt); - $rate = $moved / $elapsed; + if (! $this->option('force') && ! $this->confirm("Upload {$total} emoji to cloud using {$mode}?", true)) { + $this->comment('Aborted.'); - $this->info(($this->option('dry-run') ? '[dry-run] ' : '').'Done. moved='.$moved.' skipped='.$skipped.' failed='.$failed.'.'); - $this->info(sprintf('Elapsed %.1fs, %.1f uploads/sec.', $elapsed, $rate)); - if ($this->movedBytes) { - $this->info('Transferred '.PrettyNumber::size($this->movedBytes).' to cloud storage.'); + return self::SUCCESS; } - return self::SUCCESS; + return $concurrency > 0 + ? $this->uploadAsync($files, $concurrency, $localDisk) + : $this->uploadSync($files, $localDisk); } /** - * Spawn N child worker processes, each handling a strided slice of the - * files, and wait for them all to finish. + * Enumerate local emoji files, applying offset/limit and skipping the + * missing.png placeholder (hardcoded local /storage/emoji/missing.png + * onerror fallback) and dotfiles. + * + * @return list */ - protected function runParallel(int $workers): int + protected function collectFiles($localDisk): array { - if (! $this->option('force') && ! $this->confirm("Begin migrating local custom emoji to cloud using {$workers} parallel workers?", true)) { - $this->comment('Aborted.'); - - return self::SUCCESS; - } - - $php = PHP_BINARY; - $artisan = base_path('artisan'); - - // Forwarded flags each worker should inherit. - $forward = ['--force']; - foreach (['dry-run', 'keep-local', 'skip-verify', 'skip-cloud-check', 'debug'] as $flag) { - if ($this->option($flag)) { - $forward[] = '--'.$flag; - } - } - if ((int) $this->option('limit') > 0) { - $forward[] = '--limit='.(int) $this->option('limit'); - } - if ((int) $this->option('offset') > 0) { - $forward[] = '--offset='.(int) $this->option('offset'); - } - - $this->info("Launching {$workers} workers..."); - - $startedAt = microtime(true); - $totalMoved = 0; - - $procs = []; - for ($w = 0; $w < $workers; $w++) { - $cmd = array_merge( - [$php, $artisan, 'admin:EmojiMoveStorageLocalToCloud'], - $forward, - ['--stride='.$workers, '--shard='.$w] - ); - - $process = new Process($cmd); - $process->setTimeout(null); - $process->start(); - $procs[$w] = $process; - } - - // Stream each worker's output prefixed with its shard id. - while (array_filter($procs, fn ($p) => $p->isRunning())) { - foreach ($procs as $w => $process) { - if ($out = $process->getIncrementalOutput()) { - foreach (explode("\n", rtrim($out, "\n")) as $line) { - if ($line !== '') { - $this->line("[worker {$w}] ".$line); - if (preg_match('/moved=(\d+)/', $line, $m)) { - $totalMoved += (int) $m[1]; - } - } - } - } - } - usleep(200000); - } + $files = $localDisk->exists('public/emoji') ? $localDisk->files('public/emoji') : []; - $failed = 0; - foreach ($procs as $w => $process) { - if (! $process->isSuccessful()) { - $failed++; - $this->error("Worker {$w} exited with code ".$process->getExitCode()); - } + $offset = max(0, (int) $this->option('offset')); + if ($offset > 0) { + $files = array_slice($files, $offset); } - if (! $this->option('dry-run')) { - Cache::forget('pf:custom_emoji'); + $limit = (int) $this->option('limit'); + if ($limit > 0) { + $files = array_slice($files, 0, $limit); } - $elapsed = max(0.001, microtime(true) - $startedAt); - $rate = $totalMoved / $elapsed; - - $this->newLine(); - $this->info('All workers finished'.($failed ? " ({$failed} failed)" : '.')); - $this->info(sprintf('Total moved=%d across %d workers in %.1fs, %.1f uploads/sec.', $totalMoved, $workers, $elapsed, $rate)); + return array_values(array_filter($files, function ($p) { + $name = basename($p); - return $failed ? self::FAILURE : self::SUCCESS; + return ! str_starts_with($name, '.') && $name !== 'missing.png'; + })); } /** * Upload via the async S3 SDK with a fixed number of concurrent in-flight * PutObject requests. A successful PutObject response is the confirmation - * (no separate HEAD verify), and the local copy is deleted on success. + * (no separate HEAD verify); the local copy is deleted on success. + * + * @param list $files */ - protected function runAsync(int $concurrency, $localDisk): int + protected function uploadAsync(array $files, int $concurrency, $localDisk): int { $conf = config('filesystems.disks.s3'); $bucket = $conf['bucket'] ?? null; @@ -316,7 +148,6 @@ class EmojiMoveStorageLocalToCloud extends Command return self::FAILURE; } - // Build an S3 client from the same disk config Flysystem uses. $args = [ 'version' => 'latest', 'region' => $conf['region'] ?? 'us-east-1', @@ -340,67 +171,26 @@ class EmojiMoveStorageLocalToCloud extends Command return self::FAILURE; } - $limit = (int) $this->option('limit'); - $offset = max(0, (int) $this->option('offset')); $keepLocal = (bool) $this->option('keep-local'); - $dryRun = (bool) $this->option('dry-run'); - - $files = $localDisk->exists('public/emoji') ? $localDisk->files('public/emoji') : []; - if ($offset > 0) { - $files = array_slice($files, $offset); - } - if ($limit > 0) { - $files = array_slice($files, 0, $limit); - } - - // Filter out placeholders/dotfiles up front. - $files = array_values(array_filter($files, function ($p) { - $name = basename($p); - - return ! str_starts_with($name, '.') && $name !== 'missing.png'; - })); - - $total = count($files); - - if ($total === 0) { - $this->info('No emoji files to migrate.'); - - return self::SUCCESS; - } - - if (! $dryRun && ! $this->option('force')) { - if (! $this->confirm("Upload {$total} emoji to cloud using async S3 (concurrency={$concurrency})?", true)) { - $this->comment('Aborted.'); - - return self::SUCCESS; - } - } - - if ($dryRun) { - $this->info("[dry-run] Would upload {$total} files via async S3 (concurrency={$concurrency})."); - - return self::SUCCESS; - } + $sendAcl = ! $this->option('no-acl'); + $visibility = ($conf['visibility'] ?? 'public') === 'public' ? 'public-read' : 'private'; $moved = 0; $failed = 0; $startedAt = microtime(true); - $visibility = ($conf['visibility'] ?? 'public') === 'public' ? 'public-read' : 'private'; - $bar = $this->output->createProgressBar($total); + $bar = $this->output->createProgressBar(count($files)); $bar->setFormat(' %current%/%max% [%bar%] %percent:3s%% %rate% up/s'); $bar->setMessage('0.0', 'rate'); $bar->start(); - // Lazily yield a PutObject command per file so CommandPool pulls work - // as concurrency slots free up (keeps memory flat over large runs). - $sendAcl = ! $this->option('no-acl'); + // Lazily yield a PutObject command per file so CommandPool pulls work as + // concurrency slots free up (keeps memory flat over large runs). $commands = function () use ($client, $files, $bucket, $localDisk, $visibility, $sendAcl) { foreach ($files as $localPath) { - $mediaPath = Str::after($localPath, 'public/'); $params = [ 'Bucket' => $bucket, - 'Key' => $mediaPath, + 'Key' => Str::after($localPath, 'public/'), 'SourceFile' => $localDisk->path($localPath), ]; if ($sendAcl) { @@ -431,93 +221,98 @@ class EmojiMoveStorageLocalToCloud extends Command }, ]); - // Block until all queued uploads settle. $pool->promise()->wait(); $bar->finish(); $this->newLine(2); - if ($moved > 0) { - Cache::forget('pf:custom_emoji'); - } - - $elapsed = max(0.001, microtime(true) - $startedAt); - $this->info(sprintf('Done. moved=%d failed=%d in %.1fs, %.1f uploads/sec (concurrency=%d).', $moved, $failed, $elapsed, $moved / $elapsed, $concurrency)); - - return $failed ? self::FAILURE : self::SUCCESS; + return $this->finish($moved, $failed, $startedAt, $concurrency); } /** - * @return string one of moved|skipped|failed + * Simple synchronous fallback (concurrency=0): upload one file at a time via + * the cloud disk. Slower, but has no dependency on the S3 SDK internals. + * + * @param list $files */ - protected function migrateFile(string $localPath, string $mediaPath, $localDisk, $cloudDisk, bool $debug = false): string + protected function uploadSync(array $files, $localDisk): int { - // The upfront cloud HEAD is an extra S3 round-trip per file. Skip it - // with --skip-cloud-check for a faster, always-upload (idempotent) run. - if (! $this->option('skip-cloud-check') && $cloudDisk->exists($mediaPath)) { - if ($debug) { - $this->newLine(); - $this->line(' [skip] already on cloud: '.$mediaPath); - } - - // Present on cloud already; remove the local copy unless asked not to. - if (! $this->option('dry-run') && ! $this->option('keep-local')) { - $localDisk->delete($localPath); - } - - return 'skipped'; - } - - if ($this->option('dry-run')) { - if ($debug) { - $this->newLine(); - $this->line(' [dry-run] would move: '.$localPath.' -> '.$mediaPath); - } - - return 'moved'; - } + $cloudDisk = Storage::disk(config('filesystems.cloud')); + $keepLocal = (bool) $this->option('keep-local'); - try { - $size = (int) $localDisk->size($localPath); - $cloudDisk->put($mediaPath, $localDisk->get($localPath), 'public'); + $moved = 0; + $failed = 0; + $startedAt = microtime(true); - // Verify is another S3 round-trip; skippable with --skip-verify. - if (! $this->option('skip-verify') && ! $this->verify($localPath, $mediaPath, $localDisk, $cloudDisk)) { - $this->warn(PHP_EOL.'Verify failed for '.$mediaPath.'; left local copy intact.'); + $bar = $this->output->createProgressBar(count($files)); + $bar->setFormat(' %current%/%max% [%bar%] %percent:3s%% %rate% up/s'); + $bar->setMessage('0.0', 'rate'); + $bar->start(); - return 'failed'; - } + foreach ($files as $localPath) { + $mediaPath = Str::after($localPath, 'public/'); - if (! $this->option('keep-local')) { - $localDisk->delete($localPath); + try { + $cloudDisk->put($mediaPath, $localDisk->get($localPath), 'public'); + if (! $keepLocal) { + $localDisk->delete($localPath); + } + $moved++; + } catch (\Throwable $e) { + $failed++; + $this->warn(PHP_EOL.'Upload failed for '.$localPath.': '.$e->getMessage()); } - $this->movedBytes += $size; + $elapsed = max(0.001, microtime(true) - $startedAt); + $bar->setMessage(sprintf('%.1f', $moved / $elapsed), 'rate'); + $bar->advance(); + } - return 'moved'; - } catch (\Throwable $e) { - $this->warn(PHP_EOL.'Error migrating '.$mediaPath.': '.$e->getMessage()); + $bar->finish(); + $this->newLine(2); - return 'failed'; - } + return $this->finish($moved, $failed, $startedAt, 0); } /** - * Verify the cloud copy matches the local source by size. Fails closed. + * Bust the emoji cache and print the run summary. */ - protected function verify(string $localPath, string $cloudPath, $localDisk, $cloudDisk): bool + protected function finish(int $moved, int $failed, float $startedAt, int $concurrency): int { - if (! $cloudDisk->exists($cloudPath)) { - return false; + if ($moved > 0) { + Cache::forget('pf:custom_emoji'); } - $localSize = $localDisk->size($localPath); - $cloudSize = $cloudDisk->size($cloudPath); + $elapsed = max(0.001, microtime(true) - $startedAt); + $this->info(sprintf( + 'Done. moved=%d failed=%d in %.1fs, %.1f uploads/sec%s.', + $moved, + $failed, + $elapsed, + $moved / $elapsed, + $concurrency > 0 ? " (concurrency={$concurrency})" : '' + )); - if ($localSize === false || $cloudSize === false || $localSize !== $cloudSize) { - return false; - } + return $failed ? self::FAILURE : self::SUCCESS; + } - return true; + protected function printDebug($localDisk): void + { + $this->line('--- debug: config ---'); + $this->line(' config(pixelfed.cloud_storage): '.var_export(config('pixelfed.cloud_storage'), true)); + $this->line(' config_cache(pixelfed.cloud_storage): '.var_export(config_cache('pixelfed.cloud_storage'), true)); + $this->line(' filesystems.cloud: '.config('filesystems.cloud')); + $this->line(' cloud host: '.($this->cloudHost() ?? 'null')); + $this->line(' local disk root: '.$localDisk->path('')); + + $this->line('--- debug: custom_emoji table ---'); + $this->line(' total rows: '.CustomEmoji::count()); + $this->line(' uri IS NULL: '.CustomEmoji::whereNull('uri')->count()); + $this->line(' uri NOT NULL: '.CustomEmoji::whereNotNull('uri')->count()); + $this->line(' media_path NOT NULL: '.CustomEmoji::whereNotNull('media_path')->count()); + + $this->line('--- debug: local emoji directory (public/emoji) ---'); + $this->line(' dir exists: '.var_export($localDisk->exists('public/emoji'), true)); + $this->line(' file count: '.count($localDisk->files('public/emoji'))); } }