error('Cloud storage is not enabled (pixelfed.cloud_storage is false).'); return self::FAILURE; } 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()); return self::FAILURE; } if (! $this->cloudHost()) { $this->error('Cloud disk ('.config('filesystems.cloud').') is not configured (no resolvable URL).'); $this->line('Set AWS_URL / AWS_* in your .env before migrating to cloud.'); 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); } $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); } if ($limit > 0) { $files = array_slice($files, 0, $limit); } $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(); } 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(); } $bar?->finish(); $this->newLine(2); if ($moved > 0 && ! $this->option('dry-run')) { Cache::forget('pf:custom_emoji'); } $elapsed = max(0.001, microtime(true) - $startedAt); $rate = $moved / $elapsed; $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; } /** * Spawn N child worker processes, each handling a strided slice of the * files, and wait for them all to finish. */ protected function runParallel(int $workers): int { 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); } $failed = 0; foreach ($procs as $w => $process) { if (! $process->isSuccessful()) { $failed++; $this->error("Worker {$w} exited with code ".$process->getExitCode()); } } if (! $this->option('dry-run')) { Cache::forget('pf:custom_emoji'); } $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 $failed ? self::FAILURE : self::SUCCESS; } /** * 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. */ protected function runAsync(int $concurrency, $localDisk): int { $conf = config('filesystems.disks.s3'); $bucket = $conf['bucket'] ?? null; if (! $bucket) { $this->error('S3 bucket is not configured (filesystems.disks.s3.bucket).'); return self::FAILURE; } // Build an S3 client from the same disk config Flysystem uses. $args = [ 'version' => 'latest', 'region' => $conf['region'] ?? 'us-east-1', 'credentials' => [ 'key' => $conf['key'] ?? null, 'secret' => $conf['secret'] ?? null, ], ]; if (! empty($conf['endpoint'])) { $args['endpoint'] = $conf['endpoint']; } if (! empty($conf['use_path_style_endpoint'])) { $args['use_path_style_endpoint'] = true; } try { $client = new S3Client($args); } catch (\Throwable $e) { $this->error('Could not build S3 client: '.$e->getMessage()); 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; } $moved = 0; $failed = 0; $startedAt = microtime(true); $visibility = ($conf['visibility'] ?? 'public') === 'public' ? 'public-read' : 'private'; $bar = $this->output->createProgressBar($total); $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'); $commands = function () use ($client, $files, $bucket, $localDisk, $visibility, $sendAcl) { foreach ($files as $localPath) { $mediaPath = Str::after($localPath, 'public/'); $params = [ 'Bucket' => $bucket, 'Key' => $mediaPath, 'SourceFile' => $localDisk->path($localPath), ]; if ($sendAcl) { $params['ACL'] = $visibility; } yield $client->getCommand('PutObject', $params); } }; $pool = new CommandPool($client, $commands(), [ 'concurrency' => $concurrency, 'fulfilled' => function ($result, $iterKey) use (&$moved, $files, $localDisk, $keepLocal, $bar, $startedAt) { $moved++; $localPath = $files[$iterKey] ?? null; if ($localPath && ! $keepLocal) { $localDisk->delete($localPath); } $elapsed = max(0.001, microtime(true) - $startedAt); $bar->setMessage(sprintf('%.1f', $moved / $elapsed), 'rate'); $bar->advance(); }, 'rejected' => function ($reason, $iterKey) use (&$failed, $files, $bar) { $failed++; $localPath = $files[$iterKey] ?? '?'; $msg = $reason instanceof \Throwable ? $reason->getMessage() : (string) $reason; $this->warn(PHP_EOL.'Upload failed for '.$localPath.': '.$msg); $bar->advance(); }, ]); // 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 string one of moved|skipped|failed */ protected function migrateFile(string $localPath, string $mediaPath, $localDisk, $cloudDisk, bool $debug = false): string { // 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'; } try { $size = (int) $localDisk->size($localPath); $cloudDisk->put($mediaPath, $localDisk->get($localPath), 'public'); // 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.'); return 'failed'; } if (! $this->option('keep-local')) { $localDisk->delete($localPath); } $this->movedBytes += $size; return 'moved'; } catch (\Throwable $e) { $this->warn(PHP_EOL.'Error migrating '.$mediaPath.': '.$e->getMessage()); return 'failed'; } } /** * Verify the cloud copy matches the local source by size. Fails closed. */ protected function verify(string $localPath, string $cloudPath, $localDisk, $cloudDisk): bool { if (! $cloudDisk->exists($cloudPath)) { return false; } $localSize = $localDisk->size($localPath); $cloudSize = $cloudDisk->size($cloudPath); if ($localSize === false || $cloudSize === false || $localSize !== $cloudSize) { return false; } return true; } }