feat: report finished segments during pack() and accept URL sources

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

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:

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;

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);
 
-            $package = $this->result($dir, $output, $info, $info->duration);
+            // Only a run that succeeded has a last segment worth reporting.
+            $watcher?->scan(final: true);
+
+            $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,17 +389,21 @@ 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<string>  $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;
         $speed = 0.0;
 
         $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) {
@@ -411,6 +428,13 @@ function (string $line) use (&$frame, &$fps, &$speed, $duration): 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(
@@ -423,6 +447,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 +578,11 @@ private function sheet(
 
     /**
      * Reads back what was written and describes it.
+     *
+     * @param  array<string, int>  $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 +593,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,15 +605,15 @@ 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();
 
             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) {

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);

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
      */

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;
         }