From c87f8821a21c386320eafdd677f9bf8338f00c19 Mon Sep 17 00:00:00 2001 From: shimon Date: Tue, 3 Nov 2026 11:23:46 +0200 Subject: [PATCH 1/4] feat: parsers describe segments already shipped off disk M3u8::read() and Mpd::read() take an optional ledger of segment sizes by name. A segment named by the manifest but gone from disk is described with the recorded size instead of failing the package as missing; one that is neither on disk nor in the ledger still fails. Co-Authored-By: Claude Opus 5.5 --- src/Video/Parser/M3u8.php | 16 ++++++--- src/Video/Parser/Mpd.php | 38 +++++++++++++++----- tests/Video/Unit/ParserM3u8Test.php | 31 +++++++++++++++++ tests/Video/Unit/ParserMpdTest.php | 54 +++++++++++++++++++++++++++++ 4 files changed, 125 insertions(+), 14 deletions(-) diff --git a/src/Video/Parser/M3u8.php b/src/Video/Parser/M3u8.php index a1cda5d..15fde71 100644 --- a/src/Video/Parser/M3u8.php +++ b/src/Video/Parser/M3u8.php @@ -181,9 +181,13 @@ public static function media(string $path): array /** * Reads a master playlist and everything it points at. * + * @param array $shipped Bytes of segments already moved off + * this disk, by name. One of those is + * not missing; it is described with + * the size recorded before it went. * @return array{variants: list, metadata: array, playlists: list} */ - public static function read(string $master, string $dir): array + public static function read(string $master, string $dir, array $shipped = []): array { $dir = \rtrim($dir, '/'); $variants = []; @@ -207,7 +211,7 @@ public static function read(string $master, string $dir): array $number = 0; foreach ($media['segments'] as $segment) { - $segments[] = self::segment($entry['id'], $dir, $segment, $number); + $segments[] = self::segment($entry['id'], $dir, $segment, $number, $shipped); if (! $segment['init']) { $number++; @@ -237,12 +241,14 @@ public static function read(string $master, string $dir): array /** * @param array{file: string, duration: float, init: bool} $segment + * @param array $shipped */ - private static function segment(string $variant, string $dir, array $segment, int $number): Segment + private static function segment(string $variant, string $dir, array $segment, int $number, array $shipped): Segment { $path = $dir.'/'.$segment['file']; + $size = \is_file($path) ? (int) \filesize($path) : ($shipped[$segment['file']] ?? null); - if (! \is_file($path)) { + if ($size === null) { throw new Runtime('Segment "'.$segment['file'].'" is missing from the package'); } @@ -253,7 +259,7 @@ private static function segment(string $variant, string $dir, array $segment, in duration: $segment['duration'], init: $segment['init'], number: $segment['init'] ? 0 : $number, - size: (int) \filesize($path), + size: $size, ); } diff --git a/src/Video/Parser/Mpd.php b/src/Video/Parser/Mpd.php index d382266..8ec7115 100644 --- a/src/Video/Parser/Mpd.php +++ b/src/Video/Parser/Mpd.php @@ -24,9 +24,11 @@ final class Mpd { /** + * @param array $shipped Bytes of segments already moved off + * this disk, by name. See M3u8::read(). * @return array{variants: list, metadata: array} */ - public static function read(string $path, string $dir): array + public static function read(string $path, string $dir, array $shipped = []): array { $dir = \rtrim($dir, '/'); $document = self::load($path); @@ -63,6 +65,7 @@ public static function read(string $path, string $dir): array $dir, $duration, $target, + $shipped, ); } } @@ -70,6 +73,9 @@ public static function read(string $path, string $dir): array return ['variants' => $variants, 'metadata' => $metadata]; } + /** + * @param array $shipped + */ private static function variant( DOMElement $set, DOMElement $representation, @@ -77,6 +83,7 @@ private static function variant( string $dir, float $duration, float $target, + array $shipped, ): Variant { $id = $representation->getAttribute('id'); $id = $id !== '' ? $id : $group; @@ -102,6 +109,7 @@ private static function variant( $dir, $duration, $bandwidth, + $shipped, ); $media = \array_filter( @@ -134,6 +142,7 @@ private static function variant( } /** + * @param array $shipped * @return array{0: list, 1: int, 2: int} */ private static function segments( @@ -143,6 +152,7 @@ private static function segments( string $dir, float $duration, int $bandwidth, + array $shipped, ): array { // A Representation describes its own segments when it says anything at // all, and only falls back to the AdaptationSet when it says nothing. @@ -152,13 +162,13 @@ private static function segments( $list = self::child($element, 'SegmentList'); if ($list !== null) { - return self::listed($list, $id, $dir); + return self::listed($list, $id, $dir, $shipped); } $template = self::child($element, 'SegmentTemplate'); if ($template !== null) { - return self::templated($template, $id, $dir, $duration, $bandwidth); + return self::templated($template, $id, $dir, $duration, $bandwidth, $shipped); } } @@ -168,9 +178,10 @@ private static function segments( /** * Every segment written out one by one. * + * @param array $shipped * @return array{0: list, 1: int, 2: int} */ - private static function listed(DOMElement $list, string $id, string $dir): array + private static function listed(DOMElement $list, string $id, string $dir, array $shipped): array { $timescale = (int) $list->getAttribute('timescale'); $start = (int) $list->getAttribute('startNumber'); @@ -185,13 +196,13 @@ private static function listed(DOMElement $list, string $id, string $dir): array $init = self::child($list, 'Initialization') ?? self::descendant($list, 'Initialization'); if ($init !== null && $init->getAttribute('sourceURL') !== '') { - $segments[] = self::segment($id, $dir, $init->getAttribute('sourceURL'), 0.0, true, 0); + $segments[] = self::segment($id, $dir, $init->getAttribute('sourceURL'), 0.0, true, 0, $shipped); } $number = $start; foreach ($list->getElementsByTagName('SegmentURL') as $url) { - $segments[] = self::segment($id, $dir, $url->getAttribute('media'), $duration, false, $number); + $segments[] = self::segment($id, $dir, $url->getAttribute('media'), $duration, false, $number, $shipped); $number++; } @@ -201,6 +212,7 @@ private static function listed(DOMElement $list, string $id, string $dir): array /** * Segment names described by formula, resolved into real filenames. * + * @param array $shipped * @return array{0: list, 1: int, 2: int} */ private static function templated( @@ -209,6 +221,7 @@ private static function templated( string $dir, float $duration, int $bandwidth, + array $shipped, ): array { $timescale = (int) $template->getAttribute('timescale'); $timescale = $timescale > 0 ? $timescale : 1; @@ -220,7 +233,7 @@ private static function templated( $init = $template->getAttribute('initialization'); if ($init !== '') { - $segments[] = self::segment($id, $dir, self::expand($init, $id, 0, $bandwidth), 0.0, true, 0); + $segments[] = self::segment($id, $dir, self::expand($init, $id, 0, $bandwidth), 0.0, true, 0, $shipped); } $timeline = self::child($template, 'SegmentTimeline'); @@ -254,6 +267,7 @@ private static function templated( $length, false, $number, + $shipped, ); $number++; $time += $units; @@ -281,6 +295,7 @@ private static function templated( $length, false, $number, + $shipped, ); } @@ -322,6 +337,9 @@ public static function expand( return \str_replace('$$', '$', $name); } + /** + * @param array $shipped + */ private static function segment( string $variant, string $dir, @@ -329,11 +347,13 @@ private static function segment( float $duration, bool $init, int $number, + array $shipped, ): Segment { $file = \basename($file); $path = $dir.'/'.$file; + $size = \is_file($path) ? (int) \filesize($path) : ($shipped[$file] ?? null); - if (! \is_file($path)) { + if ($size === null) { throw new Runtime('Segment "'.$file.'" is missing from the package'); } @@ -344,7 +364,7 @@ private static function segment( duration: $duration, init: $init, number: $number, - size: (int) \filesize($path), + size: $size, ); } diff --git a/tests/Video/Unit/ParserM3u8Test.php b/tests/Video/Unit/ParserM3u8Test.php index 3987697..24129af 100644 --- a/tests/Video/Unit/ParserM3u8Test.php +++ b/tests/Video/Unit/ParserM3u8Test.php @@ -171,6 +171,37 @@ public function testMissingSegmentIsAFailure(): void M3u8::read($master, $this->dir); } + /** + * A segment shipped and deleted while the run was going is not missing: + * it is described with the size recorded before it left. + */ + public function testAShippedSegmentIsDescribedFromItsLedger(): void + { + $this->write('v720_0001.ts', 'bb'); + $this->write('v720.m3u8', "#EXTM3U\n#EXTINF:6.000000,\nv720_0000.ts\n#EXTINF:2.000000,\nv720_0001.ts\n"); + $master = $this->write('master.m3u8', "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1\nv720.m3u8\n"); + + $read = M3u8::read($master, $this->dir, ['v720_0000.ts' => 1234]); + $segments = $read['variants'][0]->segments; + + $this->assertSame('v720_0000.ts', $segments[0]->file); + $this->assertSame($this->dir.'/v720_0000.ts', $segments[0]->path); + $this->assertSame(1234, $segments[0]->size); + $this->assertSame(6.0, $segments[0]->duration); + $this->assertSame(2, $segments[1]->size, 'a file still on disk is measured, not looked up'); + } + + public function testASegmentNeitherOnDiskNorShippedIsStillMissing(): void + { + $this->write('v720.m3u8', "#EXTM3U\n#EXTINF:6.000000,\nmissing.ts\n"); + $master = $this->write('master.m3u8', "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1\nv720.m3u8\n"); + + $this->expectException(Runtime::class); + $this->expectExceptionMessage('missing.ts'); + + M3u8::read($master, $this->dir, ['other.ts' => 10]); + } + public function testMissingPlaylistIsAFailure(): void { $master = $this->write('master.m3u8', "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=1\ngone.m3u8\n"); diff --git a/tests/Video/Unit/ParserMpdTest.php b/tests/Video/Unit/ParserMpdTest.php index a72f5a4..83307e5 100644 --- a/tests/Video/Unit/ParserMpdTest.php +++ b/tests/Video/Unit/ParserMpdTest.php @@ -502,6 +502,60 @@ public function testMissingSegmentIsAFailure(): void Mpd::read($manifest, $this->dir); } + public function testShippedSegmentsAreDescribedFromTheirLedger(): void + { + $this->write('stream_chunk_0_00002.m4s', 'bb'); + $manifest = $this->write('manifest.mpd', <<<'XML' + + + + + + + + + + + XML); + + $read = Mpd::read($manifest, $this->dir, [ + 'stream_init_0.m4s' => 800, + 'stream_chunk_0_00001.m4s' => 4096, + ]); + $segments = $read['variants'][0]->segments; + + $this->assertCount(3, $segments); + $this->assertTrue($segments[0]->init); + $this->assertSame(800, $segments[0]->size); + $this->assertSame(4096, $segments[1]->size); + $this->assertSame(2, $segments[2]->size); + } + + public function testASegmentNeitherOnDiskNorShippedIsStillMissing(): void + { + $manifest = $this->write('manifest.mpd', <<<'XML' + + + + + + + + + + + + + XML); + + $this->expectException(Runtime::class); + $this->expectExceptionMessage('gone.m4s'); + + Mpd::read($manifest, $this->dir, ['other.m4s' => 10]); + } + public function testUnreadableManifestIsAFailure(): void { $this->expectException(Runtime::class); From 88e6f8b63024b3cc32dd4e0118a3bda6157066c7 Mon Sep 17 00:00:00 2001 From: shimon Date: Tue, 3 Nov 2026 11:23:46 +0200 Subject: [PATCH 2/4] feat: report finished segments during pack() and accept URL sources Packager::SEGMENT fires for each segment ffmpeg has finished while the rest are still encoding, with its name, path and size. A listener may upload and delete it; the package still describes it from that size. A segment counts as finished once a higher-numbered sibling exists in its sequence, and HLS now writes `.tmp` until rename (temp_file is a default flag). Init segments and the last of each sequence are reported when the run ends, manifests never, and nothing that was in the directory, unchanged, before the run. Without a listener nothing is watched. Sources may be an http(s) URL, which ffmpeg and ffprobe read with range requests, so a presigned object storage link is probed, packed and tiled without a local copy. Every other scheme is refused, and so is a local name that starts like a protocol: is_file() followed stream wrappers, so file:///etc/passwd passed the old check and ftp:// reached the network. Messages name a URL source without its query string. Co-Authored-By: Claude Opus 5.5 --- README.md | 14 ++ docs/design.md | 14 ++ src/Video/Adapter.php | 6 +- src/Video/Adapter/FFmpeg.php | 46 ++++-- src/Video/Adapter/Mock.php | 24 ++- src/Video/Adapter/Observable.php | 11 +- src/Video/Adapter/Reads.php | 3 +- src/Video/Adapter/Watcher.php | 173 ++++++++++++++++++++ src/Video/Encoder.php | 4 +- src/Video/Output/Hls.php | 10 +- src/Video/Packager.php | 7 +- src/Video/Source.php | 66 ++++++++ src/Video/Written.php | 22 +++ tests/Video/Adapters.php | 39 +++++ tests/Video/E2E/Base.php | 83 ++++++++++ tests/Video/E2E/CmafTest.php | 9 ++ tests/Video/E2E/DashTest.php | 9 ++ tests/Video/E2E/HlsTest.php | 14 ++ tests/Video/E2E/RemoteTest.php | 173 ++++++++++++++++++++ tests/Video/E2E/serve.php | 80 ++++++++++ tests/Video/Unit/ArgumentsHlsTest.php | 2 +- tests/Video/Unit/EncoderTest.php | 17 ++ tests/Video/Unit/ImmutabilityTest.php | 6 +- tests/Video/Unit/MockTest.php | 16 ++ tests/Video/Unit/OutputTest.php | 2 +- tests/Video/Unit/PackagerTest.php | 22 +++ tests/Video/Unit/SourceTest.php | 97 ++++++++++++ tests/Video/Unit/WatcherTest.php | 219 ++++++++++++++++++++++++++ 28 files changed, 1153 insertions(+), 35 deletions(-) create mode 100644 src/Video/Adapter/Watcher.php create mode 100644 src/Video/Source.php create mode 100644 src/Video/Written.php create mode 100644 tests/Video/E2E/RemoteTest.php create mode 100644 tests/Video/E2E/serve.php create mode 100644 tests/Video/Unit/SourceTest.php create mode 100644 tests/Video/Unit/WatcherTest.php diff --git a/README.md b/README.md index c7505fa..086b83d 100644 --- a/README.md +++ b/README.md @@ -97,6 +97,20 @@ $package->files(); // everything produced, ready to upload playlist tree, so either kind of player downloads the same bytes. Use `Hls` or `Dash` to write just one of them. +To keep a long encode from holding its whole output on disk, listen for `Packager::SEGMENT`. It +fires for each segment as soon as it is finished, while the rest are still encoding, and the +listener may upload and delete the file; the package returned at the end still describes it: + +```php +$packager->on(Packager::SEGMENT, function (Written $segment) use ($bucket): void { + $bucket->upload($segment->path, $segment->file); + unlink($segment->path); +}); +``` + +A source can also be an `http(s)://` URL, such as a presigned object storage link. ffmpeg reads it +with range requests, so it is probed, packed and tiled without being copied to local disk first. + A segment can only start where a keyframe already is, so packaging forces a keyframe every segment unless you ask for something else with `keyframe()`. Nothing has to be kept in step by hand, and a keyframe interval longer than a segment — which cannot produce segments of the length asked for — is diff --git a/docs/design.md b/docs/design.md index 8d333e1..f953cdf 100644 --- a/docs/design.md +++ b/docs/design.md @@ -168,6 +168,20 @@ consumer's job: the library reports every progress block it gets. How *much* arr adapter's display level (below), not a filter in the facade — the backend decides what to print, and everything it prints is forwarded. +`SEGMENT` → `fn (Written $w)` fires during `pack()` for each segment the backend has finished +(`file`, `path`, `size`), while the rest are still encoding. A segment counts as finished once a +higher-numbered sibling exists in its sequence (HLS also writes `.tmp` until rename, via the default +`temp_file` flag); the last of each sequence and every init segment are reported when the run ends. +Manifests are never reported, and neither is anything that was in the directory, unchanged, before +the run. A listener may upload and delete the file: the package still lists it, and the parsers +describe it from the size recorded before the event (`M3u8::read()`/`Mpd::read()` take that ledger). +Without a `SEGMENT` listener nothing is watched and nothing changes. + +Sources are a local file or an `http(s)://` URL (`Source::reachable()`), which ffmpeg and ffprobe +read with range requests, so a source in object storage needs no local copy. Any other scheme, and a +local name that starts like a protocol (`concat:`, `file:`), is refused: PHP's `is_file()` follows +stream wrappers and ffmpeg reads such a prefix as a protocol, whatever is on disk. + ### Adapter interfaces Two parent interfaces hold what every backend shares, so nothing is declared twice: diff --git a/src/Video/Adapter.php b/src/Video/Adapter.php index 2ebd2d4..2284867 100644 --- a/src/Video/Adapter.php +++ b/src/Video/Adapter.php @@ -323,12 +323,14 @@ protected function forget(): void } /** + * A local file, or an http(s) URL the backend reads itself. See Source. + * * @throws Input */ protected function source(string $path): string { - if (! \is_file($path)) { - throw new Input($this->getName().': source "'.$path.'" does not exist'); + if (! Source::reachable($path)) { + throw new Input($this->getName().': source "'.Source::name($path).'" does not exist'); } return $path; diff --git a/src/Video/Adapter/FFmpeg.php b/src/Video/Adapter/FFmpeg.php index 4424cf6..2bab569 100644 --- a/src/Video/Adapter/FFmpeg.php +++ b/src/Video/Adapter/FFmpeg.php @@ -23,10 +23,12 @@ use Utopia\Video\Parser\Mpd; use Utopia\Video\Progress; use Utopia\Video\Representation; +use Utopia\Video\Source; use Utopia\Video\Spritesheet; use Utopia\Video\Thumb; use Utopia\Video\Tile; use Utopia\Video\Track; +use Utopia\Video\Written; /** * Everything the ffmpeg binary can do, behind one class. @@ -185,10 +187,21 @@ public function pack(string $dir): Package $args[] = $builder->target(); + // Only watched when someone is listening: without a listener every + // segment stays where ffmpeg wrote it, exactly as before. + $watcher = ($this->listeners[Observable::SEGMENT] ?? []) === [] + ? null + : new Watcher($dir, function (Written $written): void { + $this->emit(Observable::SEGMENT, $written); + }); + try { - $this->execute($args, $info->duration); + $this->execute($args, $info->duration, $watcher); + + // Only a run that succeeded has a last segment worth reporting. + $watcher?->scan(final: true); - $package = $this->result($dir, $output, $info, $info->duration); + $package = $this->result($dir, $output, $info, $info->duration, $watcher?->ledger() ?? []); $this->reportSuccess(self::NAME.': packed '.$dir); return $package; @@ -209,7 +222,7 @@ public function grab(string $path, string $output, ?Thumb $options = null): stri $cover = $info->cover; if (! $info->hasVideo && $cover === null) { - throw new Input(self::NAME.': source "'.$path.'" carries no image to grab'); + throw new Input(self::NAME.': source "'.Source::name($path).'" carries no image to grab'); } $this->directory(\dirname($output)); @@ -277,7 +290,7 @@ public function subtitle(string $path, int $stream, string $output, ?int $timeou // Image subtitles (PGS, VobSub) are tracks too, but ffmpeg cannot turn // pictures into text; that failure surfaces from the run itself. if (! $found) { - throw new Input(self::NAME.': source "'.$path.'" has no subtitle stream '.$stream); + throw new Input(self::NAME.': source "'.Source::name($path).'" has no subtitle stream '.$stream); } $this->directory(\dirname($output)); @@ -306,11 +319,11 @@ public function tile(string $path, string $dir, ?Tile $options = null): Spritesh $info = $this->prober()->read($path); if (! $info->hasVideo || ! $info->width || ! $info->height) { - throw new Input(self::NAME.': source "'.$path.'" carries no video to tile'); + throw new Input(self::NAME.': source "'.Source::name($path).'" carries no video to tile'); } if ($info->duration <= 0) { - throw new Input(self::NAME.': source "'.$path.'" has no measurable duration'); + throw new Input(self::NAME.': source "'.Source::name($path).'" has no measurable duration'); } $dir = $this->directory($dir); @@ -376,9 +389,13 @@ protected function prefix(bool $progress = true): array /** * Runs ffmpeg, translating its progress stream into events. * + * A watcher is scanned at the end of every progress block, which ffmpeg + * writes about twice a second, so finished segments are reported while the + * run continues. A listener that throws stops the run like any other. + * * @param list $args */ - protected function execute(array $args, float $duration): void + protected function execute(array $args, float $duration, ?Watcher $watcher = null): void { $frame = 0; $fps = 0.0; @@ -386,7 +403,7 @@ protected function execute(array $args, float $duration): void $this->process( $args, - function (string $line) use (&$frame, &$fps, &$speed, $duration): void { + function (string $line) use (&$frame, &$fps, &$speed, $duration, $watcher): void { $split = \strpos($line, '='); if ($split === false) { @@ -423,6 +440,8 @@ function (string $line) use (&$frame, &$fps, &$speed, $duration): void { break; case 'progress': + $watcher?->scan(); + if ($value === 'end') { $this->emit(Observable::PROGRESS, new Progress( percent: 100.0, @@ -552,8 +571,11 @@ private function sheet( /** * Reads back what was written and describes it. + * + * @param array $shipped Segments reported while the run went on, + * which a listener may have removed. */ - private function result(string $dir, Output $output, Info $info, float $duration): Package + private function result(string $dir, Output $output, Info $info, float $duration, array $shipped = []): Package { $manifests = []; @@ -564,7 +586,7 @@ private function result(string $dir, Output $output, Info $info, float $duration self::independent($master); self::spoken($master, $info); - $read = Mpd::read($mpd, $dir); + $read = Mpd::read($mpd, $dir, $shipped); $manifests[] = new Manifest(Manifest::DASH, $mpd, true); if (\is_file($master)) { @@ -576,7 +598,7 @@ private function result(string $dir, Output $output, Info $info, float $duration } } elseif ($output instanceof Dash) { $mpd = $dir.'/'.$output->manifestFile(); - $read = Mpd::read($mpd, $dir); + $read = Mpd::read($mpd, $dir, $shipped); $manifests[] = new Manifest(Manifest::DASH, $mpd, true); } elseif ($output instanceof Hls) { $master = $dir.'/'.$output->masterFile(); @@ -584,7 +606,7 @@ private function result(string $dir, Output $output, Info $info, float $duration self::independent($master); self::spoken($master, $info); - $read = M3u8::read($master, $dir); + $read = M3u8::read($master, $dir, $shipped); $manifests[] = new Manifest(Manifest::HLS, $master, true); foreach ($read['playlists'] as $playlist) { diff --git a/src/Video/Adapter/Mock.php b/src/Video/Adapter/Mock.php index 372cadc..f88db5a 100644 --- a/src/Video/Adapter/Mock.php +++ b/src/Video/Adapter/Mock.php @@ -14,11 +14,13 @@ use Utopia\Video\Package; use Utopia\Video\Progress; use Utopia\Video\Segment; +use Utopia\Video\Source; use Utopia\Video\Spritesheet; use Utopia\Video\Thumb; use Utopia\Video\Tile; use Utopia\Video\Track; use Utopia\Video\Variant; +use Utopia\Video\Written; /** * A backend that pretends, so tests do not need real tools. @@ -89,7 +91,7 @@ public function read(string $path): Info public function valid(string $path): bool { - return \is_file($path) && $this->duration > 0; + return Source::reachable($path) && $this->duration > 0; } public function info(): ?Info @@ -164,15 +166,21 @@ private function packed(string $dir, Output $output): Package foreach ([0, 1] as $number) { $file = $output->base().'_'.$rep->name.'_'.$number.'.ts'; - \file_put_contents($dir.'/'.$file, 'mock segment'); + $path = $dir.'/'.$file; + \file_put_contents($path, 'mock segment'); + $size = (int) \filesize($path); + + // Reported as each is finished, like the real backend, so a + // listener that ships and deletes segments can be proved here. + $this->emit(Observable::SEGMENT, new Written($file, $path, $size)); $segments[] = new Segment( variant: (string) $position, file: $file, - path: $dir.'/'.$file, + path: $path, duration: $this->duration / 2, number: $number, - size: (int) \filesize($dir.'/'.$file), + size: $size, ); } @@ -210,7 +218,7 @@ public function grab(string $path, string $output, ?Thumb $options = null): stri $this->source($path); if ($this->width <= 0 || $this->height <= 0) { - throw new Input(self::NAME.': source "'.$path.'" carries no image to grab'); + throw new Input(self::NAME.': source "'.Source::name($path).'" carries no image to grab'); } $this->directory(\dirname($output)); @@ -237,7 +245,7 @@ public function subtitle(string $path, int $stream, string $output, ?int $timeou } if (! $found) { - throw new Input(self::NAME.': source "'.$path.'" has no subtitle stream '.$stream); + throw new Input(self::NAME.': source "'.Source::name($path).'" has no subtitle stream '.$stream); } $this->directory(\dirname($output)); @@ -258,11 +266,11 @@ public function tile(string $path, string $dir, ?Tile $options = null): Spritesh // The same refusals the real adapter makes, so code proved against this // double behaves the same way once ffmpeg is behind it. if ($this->width <= 0 || $this->height <= 0) { - throw new Input(self::NAME.': source "'.$path.'" carries no video to tile'); + throw new Input(self::NAME.': source "'.Source::name($path).'" carries no video to tile'); } if ($this->duration <= 0) { - throw new Input(self::NAME.': source "'.$path.'" has no measurable duration'); + throw new Input(self::NAME.': source "'.Source::name($path).'" has no measurable duration'); } $dir = $this->directory($dir); diff --git a/src/Video/Adapter/Observable.php b/src/Video/Adapter/Observable.php index 4312f32..40274ef 100644 --- a/src/Video/Adapter/Observable.php +++ b/src/Video/Adapter/Observable.php @@ -16,7 +16,16 @@ interface Observable public const LOG = 'log'; /** - * Register a listener for PROGRESS or LOG. + * A segment the backend has finished writing, while the job is still + * running. Receives a Written. + * + * A listener may move or delete the file: its size is recorded before the + * event fires, and the package returned at the end still describes it. + */ + public const SEGMENT = 'segment'; + + /** + * Register a listener for PROGRESS, LOG or SEGMENT. * * @param callable $listener */ diff --git a/src/Video/Adapter/Reads.php b/src/Video/Adapter/Reads.php index 16e5a42..3c7d1ad 100644 --- a/src/Video/Adapter/Reads.php +++ b/src/Video/Adapter/Reads.php @@ -6,6 +6,7 @@ use Utopia\Video\Exception\Input; use Utopia\Video\Exception\Runtime; +use Utopia\Video\Source; /** * The shared answer to "can this file be used at all?". @@ -20,7 +21,7 @@ trait Reads { public function valid(string $path): bool { - if (! \is_file($path)) { + if (! Source::reachable($path)) { return false; } diff --git a/src/Video/Adapter/Watcher.php b/src/Video/Adapter/Watcher.php new file mode 100644 index 0000000..19510ed --- /dev/null +++ b/src/Video/Adapter/Watcher.php @@ -0,0 +1,173 @@ + */ + private const MANIFESTS = ['m3u8', 'mpd']; + + /** + * Bytes of every file reported so far, by name. + * + * @var array + */ + private array $ledger = []; + + /** + * The highest number seen in each group, kept across scans because a + * reported file may already be gone. + * + * @var array + */ + private array $highest = []; + + /** + * What was in the directory before the run, by name. + * + * @var array + */ + private array $before = []; + + /** + * @param Closure(Written): void $report + */ + public function __construct( + private readonly string $dir, + private readonly Closure $report, + ) { + foreach (\scandir($dir) ?: [] as $name) { + $mark = self::mark($dir.'/'.$name); + + if ($mark !== null) { + $this->before[$name] = $mark; + } + } + } + + /** + * Report what has been finished since the last scan. + * + * @param bool $final The run has ended, so everything left is finished. + */ + public function scan(bool $final = false): void + { + $found = []; + + foreach (\scandir($this->dir) ?: [] as $name) { + $path = $this->dir.'/'.$name; + + if ($name[0] === '.' || ! \is_file($path)) { + continue; + } + + if (isset($this->before[$name]) && $this->before[$name] === self::mark($path)) { + continue; + } + + $temporary = \str_ends_with($name, self::TEMPORARY); + $plain = $temporary ? \substr($name, 0, -\strlen(self::TEMPORARY)) : $name; + $group = null; + $number = null; + + if (\preg_match(self::NUMBERED, $plain, $match) === 1) { + $group = $match[1].$match[3]; + $number = (int) $match[2]; + $this->highest[$group] = \max($this->highest[$group] ?? $number, $number); + } + + // A file being written still proves the one before it is done, which + // is why it counts towards the highest number before it is skipped. + if ($temporary || $this->manifest($plain) || isset($this->ledger[$plain])) { + continue; + } + + $found[] = [$plain, $path, $group, $number]; + } + + foreach ($found as [$name, $path, $group, $number]) { + $done = $final || ( + $group !== null + && ! \str_contains(\strtolower($name), 'init') + && $this->highest[$group] > $number + ); + + if (! $done) { + continue; + } + + \clearstatcache(true, $path); + $size = \filesize($path); + + if ($size === false) { + continue; + } + + $this->ledger[$name] = $size; + ($this->report)(new Written($name, $path, $size)); + } + } + + /** + * Bytes of every file reported, by name. + * + * @return array + */ + public function ledger(): array + { + return $this->ledger; + } + + /** + * Enough of a file's state to tell whether it has been written since. + * A muxer that renames a finished file gives it a new inode, and one that + * rewrites in place changes its size or modification time. + */ + private static function mark(string $path): ?string + { + \clearstatcache(true, $path); + $stat = @\stat($path); + + if ($stat === false || ! \is_file($path)) { + return null; + } + + return $stat['ino'].':'.$stat['size'].':'.$stat['mtime'].':'.$stat['ctime']; + } + + private function manifest(string $name): bool + { + return \in_array(\strtolower(\pathinfo($name, PATHINFO_EXTENSION)), self::MANIFESTS, true); + } +} diff --git a/src/Video/Encoder.php b/src/Video/Encoder.php index d3ff836..b1c3457 100644 --- a/src/Video/Encoder.php +++ b/src/Video/Encoder.php @@ -118,8 +118,8 @@ public function subtitle(string $path, int $stream, string $output, ?int $timeou */ public function open(string $path): static { - if (! \is_file($path)) { - throw new Exception\Input('Source "'.$path.'" does not exist'); + if (! Source::reachable($path)) { + throw new Exception\Input('Source "'.Source::name($path).'" does not exist'); } // Passed straight through: the adapter's own open() makes the same diff --git a/src/Video/Output/Hls.php b/src/Video/Output/Hls.php index 553051b..b3dbe10 100644 --- a/src/Video/Output/Hls.php +++ b/src/Video/Output/Hls.php @@ -32,8 +32,14 @@ final class Hls extends Output private string $url = ''; - /** @var list */ - private array $flags = ['independent_segments']; + /** + * temp_file has ffmpeg write each segment as `.tmp` and rename it once + * closed, so a finished name is never a half-written file. That is what + * lets a SEGMENT listener ship segments while the run is still going. + * + * @var list + */ + private array $flags = ['independent_segments', 'temp_file']; public function type(): string { diff --git a/src/Video/Packager.php b/src/Video/Packager.php index 313250e..c4b2291 100644 --- a/src/Video/Packager.php +++ b/src/Video/Packager.php @@ -40,6 +40,9 @@ class Packager /** Raw backend output, useful for debugging. Receives a string. */ public const LOG = Observable::LOG; + /** A segment finished while the job runs. Receives a Written. See Observable::SEGMENT. */ + public const SEGMENT = Observable::SEGMENT; + protected readonly PackagerAdapter $adapter; /** Encodes the rungs when the packager cannot do it itself. */ @@ -122,8 +125,8 @@ public function valid(string $path): bool */ public function open(string $path): static { - if (! \is_file($path)) { - throw new Input('Source "'.$path.'" does not exist'); + if (! Source::reachable($path)) { + throw new Input('Source "'.Source::name($path).'" does not exist'); } $this->source = $path; diff --git a/src/Video/Source.php b/src/Video/Source.php new file mode 100644 index 0000000..b1efd65 --- /dev/null +++ b/src/Video/Source.php @@ -0,0 +1,66 @@ +packager() + ->open($this->source()) + ->format($this->format()) + ->add(new Representation(320, 240, 400, 64)) + ->output((new Hls())->segment(2)) + ->on(Packager::SEGMENT, function (mixed $written) use (&$shipped): void { + $this->assertInstanceOf(Written::class, $written); + $this->assertArrayNotHasKey($written->file, $shipped, $written->file.' was reported twice'); + $this->assertSame($written->size, (int) \filesize($written->path)); + $shipped[$written->file] = $written->size; + \unlink($written->path); + }) + ->pack($this->dir); + + $this->assertNotEmpty($package->segments()); + + $described = []; + + foreach ($package->segments() as $segment) { + $this->assertArrayHasKey($segment->file, $shipped, $segment->file.' was never reported'); + $this->assertSame($shipped[$segment->file], $segment->size); + $this->assertGreaterThan(0, $segment->size); + $this->assertFileDoesNotExist($segment->path); + $described[$segment->file] = $segment->size; + } + + $this->assertEqualsCanonicalizing(\array_keys($shipped), \array_keys($described), 'only segments are reported'); + $this->assertSame(\array_sum($shipped), $package->size()); + } + public function testRefusesToPackWithoutAnOutput(): void { $this->expectExceptionMessage('output format'); diff --git a/tests/Video/E2E/Base.php b/tests/Video/E2E/Base.php index 94efbac..3a58756 100644 --- a/tests/Video/E2E/Base.php +++ b/tests/Video/E2E/Base.php @@ -5,7 +5,14 @@ namespace Utopia\Tests\E2E; use PHPUnit\Framework\TestCase; +use Utopia\Video\Format\X264; +use Utopia\Video\Output; +use Utopia\Video\Package; +use Utopia\Video\Packager; use Utopia\Video\Process; +use Utopia\Video\Progress; +use Utopia\Video\Representation; +use Utopia\Video\Written; /** * Shared fixtures for the tests that drive a real encoder. @@ -83,6 +90,82 @@ protected static function video(): string return $path; } + /** + * Thirty seconds, long enough that ffmpeg reports progress several times + * before it finishes a 720p rung. + */ + protected static function long(): string + { + $path = self::$root.'/long.mp4'; + + if (\is_file($path)) { + return $path; + } + + Process::run([ + 'ffmpeg', '-y', '-hide_banner', '-loglevel', 'error', + '-f', 'lavfi', '-i', 'testsrc=duration=30:size=640x480:rate=25', + '-f', 'lavfi', '-i', 'sine=frequency=440:duration=30', + '-c:v', 'libx264', '-preset', 'ultrafast', '-pix_fmt', 'yuv420p', + '-force_key_frames', 'expr:gte(t,n_forced*2)', + '-c:a', 'aac', '-shortest', + $path, + ], timeout: 120); + + return $path; + } + + /** + * Packs the long source while a listener ships every finished segment and + * deletes it, the way a worker with little scratch disk would. + * + * @return array{package: Package, during: int, peak: int, shipped: array} + * during: segments reported before ffmpeg said it was done; + * peak: most segment files on disk at once + */ + protected function ship(Output $output): array + { + $ended = false; + $during = 0; + $peak = 0; + $shipped = []; + $dir = $this->dir; + + $package = (new Packager()) + ->open(self::long()) + ->format((new X264())->crf(30)->keyframe(2.0)->params(['-preset', 'slow'])) + ->add(new Representation(1280, 720, 2500, 64)) + ->output($output->segment(2)) + ->on(Packager::PROGRESS, function (Progress $progress) use (&$ended): void { + $ended = $ended || $progress->percent >= 100.0; + }) + ->on(Packager::SEGMENT, function (Written $written) use (&$ended, &$during, &$peak, &$shipped, $dir): void { + $peak = \max($peak, \count(\glob($dir.'/*.{ts,m4s,mp4}', GLOB_BRACE) ?: [])); + $during += $ended ? 0 : 1; + $shipped[$written->file] = $written->size; + \unlink($written->path); + }) + ->pack($this->dir); + + return ['package' => $package, 'during' => $during, 'peak' => $peak, 'shipped' => $shipped]; + } + + /** + * What ship() promises, for any output. + * + * @param array{package: Package, during: int, peak: int, shipped: array} $run + */ + protected function assertShippedWhileRunning(array $run): void + { + $segments = $run['package']->segments(); + + $this->assertGreaterThan(0, $run['during'], 'nothing left the disk before ffmpeg finished'); + $this->assertLessThan(\count($segments), $run['peak'], 'every segment was on disk at once'); + $this->assertSame([], \glob($this->dir.'/*.{ts,m4s,mp4}', GLOB_BRACE) ?: []); + $this->assertCount(\count($segments), $run['shipped']); + $this->assertSame(\array_sum($run['shipped']), $run['package']->size()); + } + protected static function silent(): string { $path = self::$root.'/silent.mp4'; diff --git a/tests/Video/E2E/CmafTest.php b/tests/Video/E2E/CmafTest.php index f7abdab..7794692 100644 --- a/tests/Video/E2E/CmafTest.php +++ b/tests/Video/E2E/CmafTest.php @@ -195,4 +195,13 @@ public function testRejectsACodecItCannotDescribeTwice(): void ->output(new Cmaf()) ->pack($this->dir); } + + /** + * A segment leaves the disk while ffmpeg is still encoding the rest, so a + * long encode never holds its whole output at once. + */ + public function testSegmentsLeaveTheDiskWhileTheRestAreStillEncoding(): void + { + $this->assertShippedWhileRunning($this->ship(new Cmaf())); + } } diff --git a/tests/Video/E2E/DashTest.php b/tests/Video/E2E/DashTest.php index 29a2c17..f74b781 100644 --- a/tests/Video/E2E/DashTest.php +++ b/tests/Video/E2E/DashTest.php @@ -191,4 +191,13 @@ public function testPackagesVp9(): void $this->assertWritten($this->dir.'/manifest.mpd'); $this->assertNotEmpty($package->segments()); } + + /** + * A segment leaves the disk while ffmpeg is still encoding the rest, so a + * long encode never holds its whole output at once. + */ + public function testSegmentsLeaveTheDiskWhileTheRestAreStillEncoding(): void + { + $this->assertShippedWhileRunning($this->ship(new Dash())); + } } diff --git a/tests/Video/E2E/HlsTest.php b/tests/Video/E2E/HlsTest.php index 82ffece..e58bbd2 100644 --- a/tests/Video/E2E/HlsTest.php +++ b/tests/Video/E2E/HlsTest.php @@ -277,4 +277,18 @@ public function testNamingFlowsThroughToArtifacts(): void $this->assertStringStartsWith('movie_', $segment->file); } } + + /** + * A segment leaves the disk while ffmpeg is still encoding the rest, so a + * long encode never holds its whole output at once. + */ + public function testSegmentsLeaveTheDiskWhileTheRestAreStillEncoding(): void + { + $this->assertShippedWhileRunning($this->ship(new Hls())); + } + + public function testFragmentedSegmentsLeaveTheDiskWhileTheRestAreStillEncoding(): void + { + $this->assertShippedWhileRunning($this->ship((new Hls())->segments(Hls::FMP4))); + } } diff --git a/tests/Video/E2E/RemoteTest.php b/tests/Video/E2E/RemoteTest.php new file mode 100644 index 0000000..3136971 --- /dev/null +++ b/tests/Video/E2E/RemoteTest.php @@ -0,0 +1,173 @@ + ['pipe', 'r'], 1 => ['file', '/dev/null', 'w'], 2 => ['file', '/dev/null', 'w']], + $pipes, + null, + [ + 'UTOPIA_SERVE_ROOT' => self::$root, + 'UTOPIA_SERVE_LOG' => self::$log, + 'PHP_CLI_SERVER_WORKERS' => '4', + ], + ); + + if ($server === false) { + self::fail('the range server could not be started'); + } + + self::$server = $server; + + for ($i = 0; $i < 50; $i++) { + $socket = @\stream_socket_client('tcp://'.$address, $code, $message, 0.1); + + if ($socket !== false) { + \fclose($socket); + + return; + } + + \usleep(100000); + } + + self::fail('the range server never started'); + } + + public static function tearDownAfterClass(): void + { + if (\is_resource(self::$server)) { + \proc_terminate(self::$server); + \proc_close(self::$server); + } + + self::$server = null; + } + + protected function setUp(): void + { + parent::setUp(); + @\unlink(self::$log); + } + + private function url(): string + { + return self::$url.'/source.mp4?X-Amz-Signature=ignored&X-Amz-Expires=60'; + } + + /** + * @return list + */ + private function ranges(): array + { + return \is_file(self::$log) ? \file(self::$log, FILE_IGNORE_NEW_LINES) ?: [] : []; + } + + public function testAUrlIsProbed(): void + { + $packager = new Packager(); + + $this->assertTrue($packager->valid($this->url())); + $this->assertEqualsWithDelta(8.0, $packager->probe($this->url())->duration, 0.2); + } + + public function testAUrlThatAnswersNotFoundIsNotUsable(): void + { + $this->assertFalse((new Packager())->valid(self::$url.'/nothing.mp4')); + } + + /** + * The source's index sits at its end, so a reader that cannot seek would + * have to download all of it first. Ranged answers prove it seeks. + */ + public function testALadderIsPackedFromAUrl(): void + { + $package = (new Packager()) + ->open($this->url()) + ->format((new X264())->crf(30)->keyframe(2.0)->params(['-preset', 'ultrafast'])) + ->add(new Representation(320, 240, 400, 64)) + ->output((new Hls())->segment(2)) + ->pack($this->dir); + + $total = 0.0; + + foreach ($package->segments() as $segment) { + $this->assertWritten($segment->path); + } + + foreach ($package->segments($package->variants()[0]->id) as $segment) { + $total += $segment->duration; + } + + $this->assertEqualsWithDelta(8.0, $total, 1.0); + $this->assertNotEmpty($this->ranges(), 'ffmpeg never asked for a byte range'); + $this->assertSame([], \glob($this->dir.'/*.mp4') ?: [], 'the source was copied into the output'); + } + + public function testATimelineIsTiledFromAUrl(): void + { + $sheet = (new Encoder())->tile($this->url(), $this->dir, new Tile()); + + $this->assertNotEmpty($sheet->images()); + $this->assertWritten($sheet->images()[0]); + } + + public function testASingleFileIsEncodedFromAUrl(): void + { + $path = (new Encoder()) + ->open($this->url()) + ->format((new X264())->crf(30)->params(['-preset', 'ultrafast'])) + ->encode($this->dir.'/out.mp4'); + + $this->assertWritten($path); + } +} diff --git a/tests/Video/E2E/serve.php b/tests/Video/E2E/serve.php new file mode 100644 index 0000000..1c2fcf4 --- /dev/null +++ b/tests/Video/E2E/serve.php @@ -0,0 +1,80 @@ + $end) { + \http_response_code(416); + \header('Content-Range: bytes */'.$size); + + return true; + } + + \http_response_code(206); + \header('Content-Range: bytes '.$start.'-'.$end.'/'.$size); + + if ($log !== '') { + \file_put_contents($log, $start.'-'.$end."\n", FILE_APPEND | LOCK_EX); + } +} + +\header('Content-Length: '.($end - $start + 1)); + +if ($_SERVER['REQUEST_METHOD'] === 'HEAD') { + return true; +} + +$handle = \fopen($path, 'rb'); + +if ($handle === false) { + return true; +} + +\fseek($handle, $start); +$left = $end - $start + 1; + +while ($left > 0 && ! \feof($handle)) { + $chunk = (string) \fread($handle, \min(65536, $left)); + $left -= \strlen($chunk); + echo $chunk; +} + +\fclose($handle); + +return true; diff --git a/tests/Video/Unit/ArgumentsHlsTest.php b/tests/Video/Unit/ArgumentsHlsTest.php index 1c6b517..a17ac7e 100644 --- a/tests/Video/Unit/ArgumentsHlsTest.php +++ b/tests/Video/Unit/ArgumentsHlsTest.php @@ -123,7 +123,7 @@ public function testSegmentsAreDeclaredIndependent(): void { $argv = $this->build($this->info(), [new Representation(1280, 720, 2538)]); - $this->assertStringContainsString('-hls_flags independent_segments', $argv); + $this->assertStringContainsString('-hls_flags independent_segments+temp_file', $argv); } public function testVodPlaylistKeepsEverySegment(): void diff --git a/tests/Video/Unit/EncoderTest.php b/tests/Video/Unit/EncoderTest.php index 87b0ecc..d009de0 100644 --- a/tests/Video/Unit/EncoderTest.php +++ b/tests/Video/Unit/EncoderTest.php @@ -111,6 +111,23 @@ public function testOpeningAgainForgetsThePreviousJob(): void $this->assertSame([], $adapter->reps); } + public function testAnHttpSourceCanBeOpened(): void + { + $adapter = new FakeEncoder(); + $url = 'https://bucket.example/video.mp4?X-Amz-Signature=abc'; + + $this->encoder($adapter)->open($url); + + $this->assertSame($url, $adapter->opened); + } + + public function testASourceOnAnyOtherProtocolIsRefused(): void + { + $this->expectException(Input::class); + + $this->encoder()->open('ftp://example.com/video.mp4'); + } + public function testOpeningAMissingSourceIsAFailure(): void { $this->expectException(Input::class); diff --git a/tests/Video/Unit/ImmutabilityTest.php b/tests/Video/Unit/ImmutabilityTest.php index c173ae0..bdfbeb5 100644 --- a/tests/Video/Unit/ImmutabilityTest.php +++ b/tests/Video/Unit/ImmutabilityTest.php @@ -115,10 +115,10 @@ public static function setters(): array ], 'Hls::flags' => [ fn () => new Hls(), - fn (Hls $it) => $it->flags(['temp_file']), + fn (Hls $it) => $it->flags(['delete_segments']), fn (Hls $it) => $it->hlsFlags(), - ['independent_segments'], - ['temp_file'], + ['independent_segments', 'temp_file'], + ['delete_segments'], ], 'Dash::template' => [ fn () => new Dash(), diff --git a/tests/Video/Unit/MockTest.php b/tests/Video/Unit/MockTest.php index c24c402..72c8f27 100644 --- a/tests/Video/Unit/MockTest.php +++ b/tests/Video/Unit/MockTest.php @@ -84,6 +84,22 @@ public function testDescribesTheSourceItWasToldToPretend(): void * bug: it is what code gets proved against. FFmpeg raises Input for a source * with no picture in it, so this has to as well — it used to divide by zero. */ + /** + * A presigned URL is a credential, and a refusal's message ends up in logs. + */ + public function testARefusalNamesAUrlSourceWithoutItsSignature(): void + { + $mock = (new Mock())->pretend(width: 0, height: 0); + + try { + (new Encoder($mock, $mock))->tile('https://bucket.example/v.mp4?X-Amz-Signature=secret', $this->dir); + $this->fail('a source with no video was tiled'); + } catch (Input $e) { + $this->assertStringContainsString('https://bucket.example/v.mp4', $e->getMessage()); + $this->assertStringNotContainsString('secret', $e->getMessage()); + } + } + public function testRefusesToTileASourceWithNoVideo(): void { $adapter = (new Mock())->pretend(duration: 8.0, width: 0, height: 0, audio: true); diff --git a/tests/Video/Unit/OutputTest.php b/tests/Video/Unit/OutputTest.php index c2efa1a..9fed78b 100644 --- a/tests/Video/Unit/OutputTest.php +++ b/tests/Video/Unit/OutputTest.php @@ -45,7 +45,7 @@ public function testHlsDefaultsToTransportStream(): void $this->assertFalse($hls->fragmented()); $this->assertSame('ts', $hls->extension()); $this->assertSame('master.m3u8', $hls->masterFile()); - $this->assertSame(['independent_segments'], $hls->hlsFlags()); + $this->assertSame(['independent_segments', 'temp_file'], $hls->hlsFlags()); } /** diff --git a/tests/Video/Unit/PackagerTest.php b/tests/Video/Unit/PackagerTest.php index 19bde3e..cb14bfb 100644 --- a/tests/Video/Unit/PackagerTest.php +++ b/tests/Video/Unit/PackagerTest.php @@ -310,6 +310,28 @@ public function testPackingWithoutALadderIsAFailure(): void $this->packager()->open($this->file)->output(new Cmaf())->pack('/out'); } + public function testAnHttpSourceIsHandedToTheAdapterAsItIs(): void + { + $adapter = new FusedPackager(); + $url = 'https://bucket.example/video.mp4?X-Amz-Signature=abc'; + + $this->packager($adapter) + ->open($url) + ->format(new X264()) + ->add(new Representation(320, 240, 400)) + ->output(new Cmaf()) + ->pack($this->dir.'/out'); + + $this->assertSame([$url], $adapter->opened); + } + + public function testASourceOnAnyOtherProtocolIsRefused(): void + { + $this->expectException(Input::class); + + $this->packager()->open('file:///etc/passwd'); + } + public function testOpeningAMissingSourceIsAFailure(): void { $this->expectException(Input::class); diff --git a/tests/Video/Unit/SourceTest.php b/tests/Video/Unit/SourceTest.php new file mode 100644 index 0000000..4d6ee0e --- /dev/null +++ b/tests/Video/Unit/SourceTest.php @@ -0,0 +1,97 @@ +assertTrue(Source::reachable('https://bucket.example/app/video.mp4?X-Amz-Signature=abc&X-Amz-Expires=60')); + $this->assertTrue(Source::reachable('http://minio:9000/bucket/video.mp4')); + $this->assertTrue(Source::reachable('HTTPS://EXAMPLE.COM/a.mp4')); + } + + public function testALocalFileIsReachable(): void + { + $path = \tempnam(\sys_get_temp_dir(), 'utopia-source-'); + $this->assertIsString($path); + + try { + $this->assertTrue(Source::reachable($path)); + } finally { + \unlink($path); + } + } + + /** + * ffmpeg speaks many protocols; a source naming any other one could make + * it read whatever that protocol reaches. + */ + public function testEveryOtherProtocolIsRefused(): void + { + foreach ([ + 'file:///etc/passwd', + 'ftp://example.com/video.mp4', + 'concat:a.mp4|b.mp4', + 'pipe:0', + 'tcp://example.com:80', + 'data:video/mp4;base64,AAAA', + ] as $path) { + $this->assertFalse(Source::reachable($path), $path); + } + } + + public function testAUrlCannotCarryWhitespaceOrControlCharacters(): void + { + $this->assertFalse(Source::reachable('https://example.com/a b.mp4')); + $this->assertFalse(Source::reachable("https://example.com/a.mp4\n-i /etc/passwd")); + $this->assertFalse(Source::reachable("https://example.com/a.mp4\0")); + $this->assertFalse(Source::reachable('https://')); + } + + /** + * PHP's is_file() follows stream wrappers, so file:///etc/passwd is a file + * to it, and ffmpeg reads a `concat:` name as a protocol even when a file + * of that name exists. A local source is a plain path or nothing. + */ + public function testALocalFileNamedLikeAProtocolIsRefused(): void + { + $dir = \sys_get_temp_dir().'/utopia-source-'.\bin2hex(\random_bytes(6)); + \mkdir($dir); + $cwd = (string) \getcwd(); + \chdir($dir); + + try { + \file_put_contents('concat:a.mp4', 'x'); + $this->assertTrue(\is_file('concat:a.mp4')); + $this->assertFalse(Source::reachable('concat:a.mp4')); + $this->assertTrue(Source::reachable('./concat:a.mp4'), 'a plain relative path is fine'); + $this->assertTrue(Source::reachable($dir.'/concat:a.mp4'), 'and so is an absolute one'); + } finally { + \unlink($dir.'/concat:a.mp4'); + \chdir($cwd); + \rmdir($dir); + } + } + + public function testAUrlIsNamedWithoutItsQueryString(): void + { + $this->assertSame( + 'https://bucket.example/app/video.mp4', + Source::name('https://bucket.example/app/video.mp4?X-Amz-Credential=a%2Fb&X-Amz-Signature=abc'), + ); + $this->assertSame('/storage/video.mp4', Source::name('/storage/video.mp4')); + $this->assertSame('/storage/what?.mp4', Source::name('/storage/what?.mp4'), 'only a URL has a query'); + } + + public function testAMissingLocalPathIsNotReachable(): void + { + $this->assertFalse(Source::reachable('/nowhere/nothing.mp4')); + $this->assertFalse(Source::reachable('')); + } +} diff --git a/tests/Video/Unit/WatcherTest.php b/tests/Video/Unit/WatcherTest.php new file mode 100644 index 0000000..f2ba374 --- /dev/null +++ b/tests/Video/Unit/WatcherTest.php @@ -0,0 +1,219 @@ + */ + private array $reported = []; + + private Watcher $watcher; + + protected function setUp(): void + { + $dir = \sys_get_temp_dir().'/utopia-watcher-'.\bin2hex(\random_bytes(6)); + \mkdir($dir, 0o755, true); + $this->dir = $dir; + $this->reported = []; + $this->watcher = new Watcher($dir, function (Written $written): void { + $this->reported[] = $written; + }); + } + + protected function tearDown(): void + { + foreach (\glob($this->dir.'/{,.}*', GLOB_BRACE) ?: [] as $path) { + if (\is_file($path)) { + \unlink($path); + } + } + + @\rmdir($this->dir); + } + + private function write(string $name, string $body = 'bytes'): void + { + \file_put_contents($this->dir.'/'.$name, $body); + } + + /** + * @return list + */ + private function names(): array + { + return \array_map(static fn (Written $written): string => $written->file, $this->reported); + } + + public function testASegmentIsFinishedOnceTheNextOneExists(): void + { + $this->write('stream_0_0000.ts', 'aaaa'); + $this->watcher->scan(); + $this->assertSame([], $this->names(), 'the newest segment may still be growing'); + + $this->write('stream_0_0001.ts'); + $this->watcher->scan(); + $this->assertSame(['stream_0_0000.ts'], $this->names()); + $this->assertSame(4, $this->reported[0]->size); + $this->assertSame($this->dir.'/stream_0_0000.ts', $this->reported[0]->path); + } + + public function testTheLastSegmentWaitsForTheEndOfTheRun(): void + { + $this->write('stream_0_0000.ts'); + $this->write('stream_0_0001.ts'); + $this->watcher->scan(); + + $this->assertSame(['stream_0_0000.ts'], $this->names()); + + $this->watcher->scan(final: true); + $this->assertSame(['stream_0_0000.ts', 'stream_0_0001.ts'], $this->names()); + } + + /** + * Each rung of a ladder numbers its own segments, so one rung running ahead + * must not release another rung's newest file. + */ + public function testEveryRungKeepsItsOwnSequence(): void + { + $this->write('stream_0_0000.ts'); + $this->write('stream_0_0001.ts'); + $this->write('stream_1_0000.ts'); + $this->watcher->scan(); + + $this->assertSame(['stream_0_0000.ts'], $this->names()); + } + + public function testAFileStillBeingWrittenIsNeverReportedButProvesTheOneBefore(): void + { + $this->write('stream_0_0000.ts'); + $this->write('stream_0_0001.ts.tmp'); + $this->watcher->scan(); + + $this->assertSame(['stream_0_0000.ts'], $this->names()); + + $this->watcher->scan(final: true); + $this->assertSame(['stream_0_0000.ts'], $this->names(), 'a .tmp left at the end is not a segment'); + } + + public function testManifestsAreNeverReported(): void + { + $this->write('master.m3u8'); + $this->write('stream_0.m3u8'); + $this->write('stream_1.m3u8'); + $this->write('manifest.mpd'); + $this->watcher->scan(final: true); + + $this->assertSame([], $this->names()); + } + + /** + * DASH numbers its init segments by rung, which looks like a sequence; an + * init segment is only released at the end however many rungs there are. + */ + public function testInitialisationSegmentsWaitForTheEnd(): void + { + $this->write('stream_init_0.m4s'); + $this->write('stream_init_1.m4s'); + $this->write('stream_0_init.mp4'); + $this->watcher->scan(); + + $this->assertSame([], $this->names()); + + $this->watcher->scan(final: true); + $this->assertEqualsCanonicalizing( + ['stream_init_0.m4s', 'stream_init_1.m4s', 'stream_0_init.mp4'], + $this->names(), + ); + } + + /** + * A listener is free to delete what it was given. The sequence must still + * advance, and the ledger must still know what was shipped and how big. + */ + public function testAListenerMayDeleteWhatItIsGiven(): void + { + $watcher = new Watcher($this->dir, function (Written $written): void { + $this->reported[] = $written; + \unlink($written->path); + }); + + $this->write('stream_chunk_0_00001.m4s', 'aaa'); + $this->write('stream_chunk_0_00002.m4s', 'bb'); + $watcher->scan(); + $this->write('stream_chunk_0_00003.m4s', 'c'); + $watcher->scan(); + $watcher->scan(final: true); + + $this->assertSame( + ['stream_chunk_0_00001.m4s', 'stream_chunk_0_00002.m4s', 'stream_chunk_0_00003.m4s'], + $this->names(), + ); + $this->assertSame( + ['stream_chunk_0_00001.m4s' => 3, 'stream_chunk_0_00002.m4s' => 2, 'stream_chunk_0_00003.m4s' => 1], + $watcher->ledger(), + ); + $this->assertSame([], \glob($this->dir.'/*.m4s')); + } + + public function testNothingIsReportedTwice(): void + { + $this->write('stream_0_0000.ts'); + $this->write('stream_0_0001.ts'); + $this->watcher->scan(); + $this->watcher->scan(); + $this->watcher->scan(final: true); + $this->watcher->scan(final: true); + + $this->assertSame(['stream_0_0000.ts', 'stream_0_0001.ts'], $this->names()); + } + + /** + * A source kept next to the output must never reach a listener that deletes + * what it is given, and an earlier, longer run's segments must not count as + * proof that this run's are finished. + */ + public function testFilesFromBeforeTheRunAreLeftAlone(): void + { + $this->write('source.mp4', 'source'); + $this->write('stream_chunk_0_00001.m4s', 'old one'); + $this->write('stream_chunk_0_00002.m4s', 'old two'); + $this->write('stream_chunk_0_00003.m4s', 'old three'); + + $watcher = new Watcher($this->dir, function (Written $written): void { + $this->reported[] = $written; + }); + + // This run rewrites the first segment and is part way into the second. + \unlink($this->dir.'/stream_chunk_0_00001.m4s'); + $this->write('stream_chunk_0_00001.m4s', 'new'); + $watcher->scan(); + + $this->assertSame([], $this->names(), 'the old second segment proves nothing about the new first'); + + \unlink($this->dir.'/stream_chunk_0_00002.m4s'); + $this->write('stream_chunk_0_00002.m4s', 'new two'); + $watcher->scan(); + $watcher->scan(final: true); + + $this->assertSame(['stream_chunk_0_00001.m4s', 'stream_chunk_0_00002.m4s'], $this->names()); + $this->assertSame(3, $this->reported[0]->size); + } + + public function testHiddenEntriesAndDirectoriesAreIgnored(): void + { + \mkdir($this->dir.'/.staging'); + $this->write('.hidden_0001.ts'); + $this->watcher->scan(final: true); + @\rmdir($this->dir.'/.staging'); + + $this->assertSame([], $this->names()); + } +} From 4ccc499a3decb41661ada1e9e568b1067978de69 Mon Sep 17 00:00:00 2001 From: shimon Date: Tue, 3 Nov 2026 11:55:18 +0200 Subject: [PATCH 3/4] test: progress keeps climbing while segments ship The shipping E2E records progress next to segments and asserts it stays monotonic, reaches 100, and is still reported after the first segment has left. Seen red with mid-run progress suppressed while watching. Co-Authored-By: Claude Opus 5.5 --- tests/Video/E2E/Base.php | 28 ++++++++++++++++++++++------ 1 file changed, 22 insertions(+), 6 deletions(-) diff --git a/tests/Video/E2E/Base.php b/tests/Video/E2E/Base.php index 3a58756..11a0fa5 100644 --- a/tests/Video/E2E/Base.php +++ b/tests/Video/E2E/Base.php @@ -119,7 +119,7 @@ protected static function long(): string * Packs the long source while a listener ships every finished segment and * deletes it, the way a worker with little scratch disk would. * - * @return array{package: Package, during: int, peak: int, shipped: array} + * @return array{package: Package, during: int, peak: int, shipped: array, progress: list, after: int} * during: segments reported before ffmpeg said it was done; * peak: most segment files on disk at once */ @@ -129,6 +129,8 @@ protected function ship(Output $output): array $during = 0; $peak = 0; $shipped = []; + $progress = []; + $events = []; $dir = $this->dir; $package = (new Packager()) @@ -136,10 +138,13 @@ protected function ship(Output $output): array ->format((new X264())->crf(30)->keyframe(2.0)->params(['-preset', 'slow'])) ->add(new Representation(1280, 720, 2500, 64)) ->output($output->segment(2)) - ->on(Packager::PROGRESS, function (Progress $progress) use (&$ended): void { - $ended = $ended || $progress->percent >= 100.0; + ->on(Packager::PROGRESS, function (Progress $report) use (&$ended, &$progress, &$events): void { + $ended = $ended || $report->percent >= 100.0; + $progress[] = $report->percent; + $events[] = $report->percent < 100.0 ? 'progress' : 'done'; }) - ->on(Packager::SEGMENT, function (Written $written) use (&$ended, &$during, &$peak, &$shipped, $dir): void { + ->on(Packager::SEGMENT, function (Written $written) use (&$ended, &$during, &$peak, &$shipped, &$events, $dir): void { + $events[] = 'segment'; $peak = \max($peak, \count(\glob($dir.'/*.{ts,m4s,mp4}', GLOB_BRACE) ?: [])); $during += $ended ? 0 : 1; $shipped[$written->file] = $written->size; @@ -147,13 +152,18 @@ protected function ship(Output $output): array }) ->pack($this->dir); - return ['package' => $package, 'during' => $during, 'peak' => $peak, 'shipped' => $shipped]; + // Progress reported once a segment had already left: shipping must not + // swallow or stall the progress stream a worker drives its status from. + $first = \array_search('segment', $events, true); + $after = $first === false ? 0 : \count(\array_keys(\array_slice($events, $first), 'progress', true)); + + return ['package' => $package, 'during' => $during, 'peak' => $peak, 'shipped' => $shipped, 'progress' => $progress, 'after' => $after]; } /** * What ship() promises, for any output. * - * @param array{package: Package, during: int, peak: int, shipped: array} $run + * @param array{package: Package, during: int, peak: int, shipped: array, progress: list, after: int} $run */ protected function assertShippedWhileRunning(array $run): void { @@ -164,6 +174,12 @@ protected function assertShippedWhileRunning(array $run): void $this->assertSame([], \glob($this->dir.'/*.{ts,m4s,mp4}', GLOB_BRACE) ?: []); $this->assertCount(\count($segments), $run['shipped']); $this->assertSame(\array_sum($run['shipped']), $run['package']->size()); + + $sorted = $run['progress']; + \sort($sorted); + $this->assertSame($sorted, $run['progress'], 'progress went backwards while segments shipped'); + $this->assertSame(100.0, \end($run['progress']), 'progress never reached 100'); + $this->assertGreaterThan(0, $run['after'], 'no progress was reported once segments began to ship'); } protected static function silent(): string From c33765c69e033102cccd6e89f6c826384520760a Mon Sep 17 00:00:00 2001 From: shimon Date: Tue, 3 Nov 2026 14:22:25 +0200 Subject: [PATCH 4/4] fix: progress no longer drops back to 0 on an N/A timestamp ffmpeg writes out_time_us=N/A for a progress block with no output timestamp yet, which a float cast read as 0, so progress fell from ~77% back to 0 before reaching 100 (seen on DASH and CMAF under ffmpeg 8.1.2). Non-numeric and negative times are now skipped. A consumer writing progress to storage would otherwise publish that 0. Also: tests no longer use GLOB_BRACE, which musl does not define, and the watcher documents the one rewrite it cannot tell apart (same second, same size, reused inode). Co-Authored-By: Claude Opus 5.5 --- src/Video/Adapter/FFmpeg.php | 7 ++ src/Video/Adapter/Watcher.php | 4 + tests/Video/E2E/Base.php | 18 ++++- tests/Video/Unit/ProgressStreamTest.php | 97 +++++++++++++++++++++++++ tests/Video/Unit/WatcherTest.php | 11 ++- 5 files changed, 131 insertions(+), 6 deletions(-) create mode 100644 tests/Video/Unit/ProgressStreamTest.php diff --git a/src/Video/Adapter/FFmpeg.php b/src/Video/Adapter/FFmpeg.php index 2bab569..e2c6718 100644 --- a/src/Video/Adapter/FFmpeg.php +++ b/src/Video/Adapter/FFmpeg.php @@ -428,6 +428,13 @@ function (string $line) use (&$frame, &$fps, &$speed, $duration, $watcher): void break; case 'out_time_us': case 'out_time_ms': + // ffmpeg writes N/A for a block with no output timestamp + // yet, which a float cast reads as 0: progress would fall + // back to the start before climbing again. + if (! \is_numeric($value) || (float) $value < 0) { + break; + } + // Both keys are reported in microseconds. $time = ((float) $value) / 1000000; $this->emit(Observable::PROGRESS, new Progress( diff --git a/src/Video/Adapter/Watcher.php b/src/Video/Adapter/Watcher.php index 19510ed..8208c04 100644 --- a/src/Video/Adapter/Watcher.php +++ b/src/Video/Adapter/Watcher.php @@ -25,6 +25,10 @@ * taken as proof that anything before it is finished: a source kept beside * the output must not be handed to a listener that deletes, and a longer * earlier run's segments must not release this run's before they are done. + * A file counts as rewritten when its inode, size or whole-second times + * change, so one rewritten in the same second at the same size on a reused + * inode looks untouched: it is not reported, but stays on disk and in the + * package. A fresh directory per run avoids the question entirely. * * @internal */ diff --git a/tests/Video/E2E/Base.php b/tests/Video/E2E/Base.php index 11a0fa5..e38f49a 100644 --- a/tests/Video/E2E/Base.php +++ b/tests/Video/E2E/Base.php @@ -145,7 +145,7 @@ protected function ship(Output $output): array }) ->on(Packager::SEGMENT, function (Written $written) use (&$ended, &$during, &$peak, &$shipped, &$events, $dir): void { $events[] = 'segment'; - $peak = \max($peak, \count(\glob($dir.'/*.{ts,m4s,mp4}', GLOB_BRACE) ?: [])); + $peak = \max($peak, \count(self::segments($dir))); $during += $ended ? 0 : 1; $shipped[$written->file] = $written->size; \unlink($written->path); @@ -171,7 +171,7 @@ protected function assertShippedWhileRunning(array $run): void $this->assertGreaterThan(0, $run['during'], 'nothing left the disk before ffmpeg finished'); $this->assertLessThan(\count($segments), $run['peak'], 'every segment was on disk at once'); - $this->assertSame([], \glob($this->dir.'/*.{ts,m4s,mp4}', GLOB_BRACE) ?: []); + $this->assertSame([], self::segments($this->dir)); $this->assertCount(\count($segments), $run['shipped']); $this->assertSame(\array_sum($run['shipped']), $run['package']->size()); @@ -182,6 +182,20 @@ protected function assertShippedWhileRunning(array $run): void $this->assertGreaterThan(0, $run['after'], 'no progress was reported once segments began to ship'); } + /** + * Segment files in a directory. Not glob with braces: GLOB_BRACE does not + * exist on musl, which is what the test image runs. + * + * @return list + */ + protected static function segments(string $dir): array + { + return \array_values(\array_filter( + \scandir($dir) ?: [], + static fn (string $name): bool => \preg_match('/\.(ts|m4s|mp4)$/', $name) === 1, + )); + } + protected static function silent(): string { $path = self::$root.'/silent.mp4'; diff --git a/tests/Video/Unit/ProgressStreamTest.php b/tests/Video/Unit/ProgressStreamTest.php new file mode 100644 index 0000000..3d9bfaf --- /dev/null +++ b/tests/Video/Unit/ProgressStreamTest.php @@ -0,0 +1,97 @@ +dir = \sys_get_temp_dir().'/utopia-progress-'.\bin2hex(\random_bytes(6)); + \mkdir($this->dir, 0o755, true); + } + + protected function tearDown(): void + { + foreach (\scandir($this->dir) ?: [] as $name) { + if (\is_file($this->dir.'/'.$name)) { + \unlink($this->dir.'/'.$name); + } + } + + @\rmdir($this->dir); + } + + /** + * A stand-in ffmpeg: prints the given progress blocks on stdout, then writes + * its last argument, which is where the real one puts its output. + * + * @param list $lines + */ + private function binary(array $lines): string + { + $path = $this->dir.'/ffmpeg'; + \file_put_contents($path, "#!/bin/sh\n" + ."cat <<'EOF'\n".\implode("\n", $lines)."\nEOF\n" + .'for last; do :; done; printf x > "$last"'."\n"); + \chmod($path, 0o755); + + return $path; + } + + /** + * @param list $lines + * @return list + */ + private function percentages(array $lines): array + { + $source = $this->dir.'/source.mp4'; + \file_put_contents($source, 'source'); + $probe = (new Mock())->pretend(duration: 10.0); + $seen = []; + + (new Encoder(new FFmpeg($this->binary($lines)), $probe)) + ->on(Encoder::PROGRESS, function (Progress $progress) use (&$seen): void { + $seen[] = $progress->percent; + }) + ->open($source) + ->encode($this->dir.'/out.mp4'); + + return $seen; + } + + public function testABlockWithoutATimestampDoesNotSendProgressBackToTheStart(): void + { + $seen = $this->percentages([ + 'out_time_us=2000000', 'out_time_ms=2000000', 'progress=continue', + 'out_time_us=7000000', 'out_time_ms=7000000', 'progress=continue', + 'out_time_us=N/A', 'out_time_ms=N/A', 'progress=continue', + 'out_time_us=9900000', 'out_time_ms=9900000', 'progress=end', + ]); + + $this->assertSame([20.0, 20.0, 70.0, 70.0, 99.0, 99.0, 100.0], $seen); + } + + public function testANegativeTimestampIsIgnoredToo(): void + { + $seen = $this->percentages([ + 'out_time_us=-9223372036854775807', 'progress=continue', + 'out_time_us=5000000', 'progress=end', + ]); + + $this->assertSame([50.0, 100.0], $seen); + } +} diff --git a/tests/Video/Unit/WatcherTest.php b/tests/Video/Unit/WatcherTest.php index f2ba374..daa43ae 100644 --- a/tests/Video/Unit/WatcherTest.php +++ b/tests/Video/Unit/WatcherTest.php @@ -30,9 +30,10 @@ protected function setUp(): void protected function tearDown(): void { - foreach (\glob($this->dir.'/{,.}*', GLOB_BRACE) ?: [] as $path) { - if (\is_file($path)) { - \unlink($path); + // scandir rather than glob: GLOB_BRACE does not exist on musl. + foreach (\scandir($this->dir) ?: [] as $name) { + if (\is_file($this->dir.'/'.$name)) { + \unlink($this->dir.'/'.$name); } } @@ -198,8 +199,10 @@ public function testFilesFromBeforeTheRunAreLeftAlone(): void $this->assertSame([], $this->names(), 'the old second segment proves nothing about the new first'); + // A real rewrite differs in size or second; see the Watcher's own note + // on what it cannot tell apart. \unlink($this->dir.'/stream_chunk_0_00002.m4s'); - $this->write('stream_chunk_0_00002.m4s', 'new two'); + $this->write('stream_chunk_0_00002.m4s', 'new second'); $watcher->scan(); $watcher->scan(final: true);