aiden
/
pixelfed
mirror of https://github.com/pixelfed/pixelfed
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Update EmojiMoveStorageLocalToCloud.php

Browse Source
pull/6970/head
Shlee 4 weeks ago committed by GitHub
parent ef8cf16cfa
commit 948da00f0c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
1 changed files with 121 additions and 326 deletions
Show Stats Download Patch File Download Diff File

447
app/Console/Commands/Admin/EmojiMoveStorageLocalToCloud.php
Unescape Escape View File

@ -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<string>
*/
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<string> $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<string> $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')));
}
}

Write Preview
Loading…
Cancel
Save