feat(webcodecs): skip decoding/rendering frames when decoder/renderer can't keep up

.vscode/settings.json

@@ -58,10 +58,13 @@
         "tinyh",
         "transferables",
         "tsbuildinfo",
+        "unconfigured",
         "undici",
         "versioncode",
         "webadb",
         "webcodecs",
+        "webglcontextlost",
+        "webglcontextrestored",
         "webm",
         "websockify",
         "webusb",

libraries/adb-credential-web/src/storage/password.ts

@@ -1,5 +1,5 @@
 import type { MaybeError } from "@yume-chan/adb";
-import { encodeUtf8 } from "@yume-chan/adb";
+import { encodeUtf8, toLocalUint8Array } from "@yume-chan/adb";
 import type { MaybePromiseLike } from "@yume-chan/async";
 import {
     buffer,
@@ -81,7 +81,7 @@ export class TangoPasswordProtectedStorage implements TangoKeyStorage {
     }
 
     async save(
-        privateKey: Uint8Array<ArrayBuffer>,
+        privateKey: Uint8Array,
         name: string | undefined,
     ): Promise<undefined> {
         const password = await this.#requestPassword("save", name);
@@ -93,7 +93,7 @@ export class TangoPasswordProtectedStorage implements TangoKeyStorage {
         const encrypted = await crypto.subtle.encrypt(
             { name: "AES-GCM", iv },
             aesKey,
-            privateKey,
+            toLocalUint8Array(privateKey),
         );
 
         const bundle = Bundle.serialize({

libraries/adb-credential-web/src/storage/prf/storage.ts

@@ -1,4 +1,4 @@
-import type { MaybeError } from "@yume-chan/adb";
+import { toLocalUint8Array, type MaybeError } from "@yume-chan/adb";
 import {
     buffer,
     struct,
@@ -85,7 +85,7 @@ export class TangoPrfStorage implements TangoKeyStorage {
     }
 
     async save(
-        privateKey: Uint8Array<ArrayBuffer>,
+        privateKey: Uint8Array,
         name: string | undefined,
     ): Promise<undefined> {
         const prfInput = new Uint8Array(PrfInputLength);
@@ -122,7 +122,7 @@ export class TangoPrfStorage implements TangoKeyStorage {
         const encrypted = await crypto.subtle.encrypt(
             { name: "AES-GCM", iv },
             aesKey,
-            privateKey,
+            toLocalUint8Array(privateKey),
         );
 
         const bundle = Bundle.serialize({

libraries/adb/src/commands/subprocess/none/pty.ts

@@ -7,9 +7,9 @@ import type {
 import { MaybeConsumable } from "@yume-chan/stream-extra";
 
 import type { AdbSocket } from "../../../adb.js";
-import type { AdbPtyProcess } from "../pty.js";
+import type { AdbPty } from "../pty.js";
 
-export class AdbNoneProtocolPtyProcess implements AdbPtyProcess<undefined> {
+export class AdbNoneProtocolPty implements AdbPty<undefined> {
     readonly #socket: AdbSocket;
     readonly #writer: WritableStreamDefaultWriter<MaybeConsumable<Uint8Array>>;
 

libraries/adb/src/commands/subprocess/none/service.ts

@@ -1,7 +1,7 @@
 import type { Adb } from "../../../adb.js";
 
 import { AdbNoneProtocolProcessImpl } from "./process.js";
-import { AdbNoneProtocolPtyProcess } from "./pty.js";
+import { AdbNoneProtocolPty } from "./pty.js";
 import { adbNoneProtocolSpawner } from "./spawner.js";
 
 export class AdbNoneProtocolSubprocessService {
@@ -42,7 +42,7 @@ export class AdbNoneProtocolSubprocessService {
 
     async pty(
         command?: string | readonly string[],
-    ): Promise<AdbNoneProtocolPtyProcess> {
+    ): Promise<AdbNoneProtocolPty> {
         let service = "shell:";
 
         if (typeof command === "string") {
@@ -52,8 +52,6 @@ export class AdbNoneProtocolSubprocessService {
             service += command.join(" ");
         }
 
-        return new AdbNoneProtocolPtyProcess(
-            await this.#adb.createSocket(service),
-        );
+        return new AdbNoneProtocolPty(await this.#adb.createSocket(service));
     }
 }

libraries/adb/src/commands/subprocess/pty.ts

@@ -5,7 +5,7 @@ import type {
     WritableStream,
 } from "@yume-chan/stream-extra";
 
-export interface AdbPtyProcess<TExitCode> {
+export interface AdbPty<TExitCode> {
     get input(): WritableStream<MaybeConsumable<Uint8Array>>;
     get output(): ReadableStream<Uint8Array>;
     get exited(): Promise<TExitCode>;

libraries/adb/src/commands/subprocess/shell/pty.ts

@@ -13,11 +13,11 @@ import {
 import { encodeUtf8 } from "@yume-chan/struct";
 
 import type { AdbSocket } from "../../../adb.js";
-import type { AdbPtyProcess } from "../pty.js";
+import type { AdbPty } from "../pty.js";
 
 import { AdbShellProtocolId, AdbShellProtocolPacket } from "./shared.js";
 
-export class AdbShellProtocolPtyProcess implements AdbPtyProcess<number> {
+export class AdbShellProtocolPty implements AdbPty<number> {
     readonly #socket: AdbSocket;
     readonly #writer: WritableStreamDefaultWriter<MaybeConsumable<Uint8Array>>;
 

libraries/adb/src/commands/subprocess/shell/service.ts

@@ -2,7 +2,7 @@ import type { Adb } from "../../../adb.js";
 import { AdbFeature } from "../../../features.js";
 
 import { AdbShellProtocolProcessImpl } from "./process.js";
-import { AdbShellProtocolPtyProcess } from "./pty.js";
+import { AdbShellProtocolPty } from "./pty.js";
 import { adbShellProtocolSpawner } from "./spawner.js";
 
 export class AdbShellProtocolSubprocessService {
@@ -39,7 +39,7 @@ export class AdbShellProtocolSubprocessService {
     async pty(options?: {
         command?: string | readonly string[] | undefined;
         terminalType?: string;
-    }): Promise<AdbShellProtocolPtyProcess> {
+    }): Promise<AdbShellProtocolPty> {
         const { command, terminalType } = options ?? {};
 
         let service = "shell,v2,pty";
@@ -59,8 +59,6 @@ export class AdbShellProtocolSubprocessService {
             service += command.join(" ");
         }
 
-        return new AdbShellProtocolPtyProcess(
-            await this.#adb.createSocket(service),
-        );
+        return new AdbShellProtocolPty(await this.#adb.createSocket(service));
     }
 }

libraries/adb/src/daemon/dispatcher.ts

@@ -146,13 +146,20 @@ export class AdbPacketDispatcher implements Closeable {
                                 this.#handleOkay(packet);
                                 break;
                             case AdbCommand.Open:
-                                await this.#handleOpen(packet);
+                                // Don't await
+                                // The handler may take a long time to accept the socket,
+                                // don't block other sockets' packet processing.
+                                this.#handleOpen(packet).catch((e) => {
+                                    // Propagate fatal errors to consumer
+                                    controller.error(e);
+                                });
                                 break;
                             case AdbCommand.Write:
-                                // Don't await - let each socket handle its own backpressure
-                                // without blocking other sockets' packet processing.
-                                // Fatal errors are propagated via WritableStream's controller.
+                                // Don't await
+                                // The socket might be stalled because of backpressure,
+                                // don't block other sockets' packet processing.
                                 this.#handleWrite(packet).catch((e) => {
+                                    // Propagate fatal errors to consumer
                                     controller.error(e);
                                 });
                                 break;

libraries/android-bin/src/bug-report.ts

@@ -258,10 +258,10 @@ export class BugReport extends AdbServiceBase {
      */
     bugReportZStream(): ReadableStream<Uint8Array> {
         return new PushReadableStream(async (controller) => {
-            const process = await this.adb.subprocess.shellProtocol!.spawn([
-                "bugreportz",
-                "-s",
-            ]);
+            const process = await this.adb.subprocess.shellProtocol!.spawn(
+                ["bugreportz", "-s"],
+                controller.abortSignal,
+            );
             process.stdout
                 .pipeTo(
                     new WritableStream({

libraries/media-codec/src/index.ts

@@ -1,4 +1,5 @@
 export * from "./av1.js";
+export * from "./format.js";
 export * as H264 from "./h264.js";
 export * as H265 from "./h265.js";
 export * from "./nalu.js";

libraries/scrcpy-decoder-tinyh264/src/decoder.ts

@@ -1,9 +1,6 @@
 import { PromiseResolver } from "@yume-chan/async";
 import { H264 } from "@yume-chan/media-codec";
-import type {
-    ScrcpyMediaStreamConfigurationPacket,
-    ScrcpyMediaStreamPacket,
-} from "@yume-chan/scrcpy";
+import type { ScrcpyMediaStreamConfigurationPacket } from "@yume-chan/scrcpy";
 import {
     AndroidAvcLevel,
     AndroidAvcProfile,
@@ -17,13 +14,16 @@ import type {
     ScrcpyVideoDecoder,
     ScrcpyVideoDecoderCapability,
 } from "./types.js";
-import { createCanvas, glIsSupported } from "./utils/index.js";
-import { PauseControllerImpl } from "./utils/pause.js";
-import { PerformanceCounterImpl } from "./utils/performance.js";
+import {
+    createCanvas,
+    glIsSupported,
+    PauseController,
+    PerformanceCounter,
+} from "./utils/index.js";
 import type { TinyH264Wrapper } from "./wrapper.js";
 import { createTinyH264Wrapper } from "./wrapper.js";
 
-const noop = () => {
+export const noop = () => {
     // no-op
 };
 
@@ -41,6 +41,15 @@ export class TinyH264Decoder implements ScrcpyVideoDecoder {
         return this.#canvas;
     }
 
+    #pause = new PauseController();
+    get paused() {
+        return this.#pause.paused;
+    }
+
+    get writable() {
+        return this.#pause.writable;
+    }
+
     #size = new ScrcpyVideoSizeImpl();
     get width() {
         return this.#size.width;
@@ -52,25 +61,32 @@ export class TinyH264Decoder implements ScrcpyVideoDecoder {
         return this.#size.sizeChanged;
     }
 
-    #counter = new PerformanceCounterImpl();
-    get framesDrawn() {
-        return this.#counter.framesDrawn;
+    #counter = new PerformanceCounter();
+    /**
+     * Gets the number of frames that have been drawn on the renderer.
+     */
+    get framesRendered() {
+        return this.#counter.framesRendered;
     }
+    /**
+     * Gets the number of frames that's visible to the user.
+     *
+     * Multiple frames might be rendered during one vertical sync interval,
+     * but only the last of them is represented to the user.
+     * This costs some performance but reduces latency by 1 frame.
+     *
+     * Might be `0` if the renderer is in a nested Web Worker on Chrome due to a Chrome bug.
+     * https://issues.chromium.org/issues/41483010
+     */
     get framesPresented() {
         return this.#counter.framesPresented;
     }
-    get framesSkipped() {
-        return this.#counter.framesSkipped;
-    }
-
-    #pause: PauseControllerImpl;
-    get paused() {
-        return this.#pause.paused;
-    }
-
-    #writable: WritableStream<ScrcpyMediaStreamPacket>;
-    get writable() {
-        return this.#writable;
+    /**
+     * Gets the number of frames that wasn't drawn on the renderer
+     * because the renderer can't keep up
+     */
+    get framesSkippedRendering() {
+        return this.#counter.framesSkippedRendering;
     }
 
     #renderer: YuvCanvas | undefined;
@@ -95,29 +111,31 @@ export class TinyH264Decoder implements ScrcpyVideoDecoder {
             }),
         });
 
-        this.#pause = new PauseControllerImpl(
-            this.#configure,
-            async (packet) => {
-                if (!this.#decoder) {
-                    throw new Error("Decoder not configured");
-                }
-
-                // TinyH264 decoder doesn't support associating metadata
-                // with each frame's input/output
-                // so skipping frames when resuming from pause is not supported
-
-                const decoder = await this.#decoder;
-
-                // `packet.data` might be from a `BufferCombiner` so we have to copy it using `slice`
-                decoder.feed(packet.data.slice().buffer);
-            },
-        );
-
-        this.#writable = new WritableStream<ScrcpyMediaStreamPacket>({
-            write: this.#pause.write,
-            // Nothing can be disposed when the stream is aborted/closed
-            // No new frames will arrive, but some frames might still be decoding and/or rendering
-        });
+        void this.#pause.readable
+            .pipeTo(
+                new WritableStream({
+                    write: async (packet) => {
+                        if (packet.type === "configuration") {
+                            await this.#configure(packet);
+                            return;
+                        }
+
+                        if (!this.#decoder) {
+                            throw new Error("Decoder not configured");
+                        }
+
+                        // TinyH264 decoder doesn't support associating metadata
+                        // with each frame's input/output
+                        // so skipping frames when resuming from pause is not supported
+
+                        const decoder = await this.#decoder;
+
+                        // `packet.data` might be from a `BufferCombiner` so we have to copy it using `slice`
+                        decoder.feed(packet.data.slice().buffer);
+                    },
+                }),
+            )
+            .catch(noop);
     }
 
     #configure = async ({
@@ -164,6 +182,9 @@ export class TinyH264Decoder implements ScrcpyVideoDecoder {
             const uPlaneOffset = encodedWidth * encodedHeight;
             const vPlaneOffset = uPlaneOffset + chromaWidth * chromaHeight;
             decoder.onPictureReady(({ data }) => {
+                // TinyH264 doesn't pass any frame metadata to `onPictureReady`
+                // so frames marked as skipped (by pause controller) can't be skipped
+
                 const array = new Uint8Array(data);
                 const frame = YuvBuffer.frame(
                     format,
@@ -184,7 +205,7 @@ export class TinyH264Decoder implements ScrcpyVideoDecoder {
 
                 // Can't know if yuv-canvas is dropping frames or not
                 this.#renderer!.drawFrame(frame);
-                this.#counter.increaseFramesDrawn();
+                this.#counter.increaseFramesRendered();
             });
 
             decoder.feed(data.slice().buffer);
@@ -199,8 +220,8 @@ export class TinyH264Decoder implements ScrcpyVideoDecoder {
         this.#pause.pause();
     }
 
-    resume(): Promise<undefined> {
-        return this.#pause.resume();
+    resume(): undefined {
+        this.#pause.resume();
     }
 
     trackDocumentVisibility(document: Document): () => undefined {