feat(echo): run on-chain with metered payments

README.md

@@ -37,13 +37,13 @@ Need a schema that isn't here? [Open an issue](https://github.com/livepeer/runne
 
 ## Examples
 
-| Example                        | Goal                                                                                  | Registration | Mode                               | Transport         | Pricing           |
-| ------------------------------ | ------------------------------------------------------------------------------------- | ------------ | ---------------------------------- | ----------------- | ----------------- |
-| [`hello-world`](./hello-world) | The simplest app: one request, one response                                           | dynamic      | single-shot                        | HTTP (JSON)       | fixed             |
-| [`tiles`](./tiles)             | Capacity fan-out — one call per tile                                                  | dynamic      | single-shot                        | HTTP (base64 PNG) | fixed             |
-| [`api-proxy`](./api-proxy)     | Pass calls through to a hosted API — the operator holds the key, callers pay per call | static       | single-shot                        | HTTP (JPEG bytes) | fixed             |
-| [`echo`](./echo)               | Realtime video, transformed and echoed back                                           | dynamic      | persistent                         | trickle           | — (offchain only) |
-| [`vllm`](./vllm)               | Drop-in OpenAI API; the client stays unmodified                                       | static       | persistent (single-shot by nature) | HTTP + SSE        | hour              |
+| Example                        | Goal                                                                                  | Registration | Mode                               | Transport         | Pricing |
+| ------------------------------ | ------------------------------------------------------------------------------------- | ------------ | ---------------------------------- | ----------------- | ------- |
+| [`hello-world`](./hello-world) | The simplest app: one request, one response                                           | dynamic      | single-shot                        | HTTP (JSON)       | fixed   |
+| [`tiles`](./tiles)             | Capacity fan-out — one call per tile                                                  | dynamic      | single-shot                        | HTTP (base64 PNG) | fixed   |
+| [`api-proxy`](./api-proxy)     | Pass calls through to a hosted API — the operator holds the key, callers pay per call | static       | single-shot                        | HTTP (JPEG bytes) | fixed   |
+| [`echo`](./echo)               | Realtime video, transformed and echoed back                                           | dynamic      | persistent                         | trickle           | hour    |
+| [`vllm`](./vllm)               | Drop-in OpenAI API; the client stays unmodified                                       | static       | persistent (single-shot by nature) | HTTP + SSE        | hour    |
 
 Start with `hello-world` (the smallest end-to-end path); the others each layer on one new idea. More will follow, including a full example that exercises every feature. Each is self-contained and runs **offchain** (free, no wallet); most also run **on-chain** (paid) — see each README.
 
@@ -110,7 +110,7 @@ Each example is self-contained and its README has the run commands. Everything b
 
 ### Prerequisites
 
-- **Docker** for the end-to-end demos. They use the mainline `livepeer/go-livepeer:v0.9.0` release image, so there is nothing to build.
+- **Docker** for the end-to-end demos, so there is nothing to build. They currently run a mainline `livepeer/go-livepeer` master build rather than a release image: metered pricing needs the session-scoped payment URL added after v0.9.0 ([#4008](https://github.com/livepeer/go-livepeer/pull/4008)). They move back to a release tag once one carries it.
 - **Python 3.12+** and [`uv`](https://docs.astral.sh/uv/) for the client.
 - The **`livepeer-gateway` SDK** from the `ja/live-runner` branch (not yet on PyPI):
 

api-proxy/.env.example

@@ -23,8 +23,8 @@ ORCH_ETH_PASSWORD=your-operator-keystore-password
 ORCH_ONCHAIN_ADDR=0xYourRegisteredOrchestrator
 
 # The runner's price lives in runners.json (static runner): USD billed once per
-# call (fixed pricing). Keep it under ~0.00019: the signer signs at most 100
-# tickets per payment and the demo orchestrator runs -ticketEV=1e9.
+# call (fixed pricing). Keep it under ~0.0019: the signer signs at most 100
+# tickets per payment and the demo orchestrator runs -ticketEV=1e10.
 # Signer's max-price cap (payer side) is per billing unit, here one call, so it
 # must exceed the runners.json price.
 MAX_PRICE_PER_UNIT=0.000111USD

api-proxy/compose.onchain.yml

@@ -40,7 +40,7 @@ services:
       # Price comes from the app's registration; -pricePerUnit only sets the
       # unused legacy base price, but an on-chain O won't boot without it.
       - -pricePerUnit=0
-      - -ticketEV=1000000000
+      - -ticketEV=10000000000
       - -maxFaceValue=4000000000000000
       - -liveRunnerConfig=/config/runners.json
       - -monitor=false

compose.onchain.yml

@@ -13,7 +13,8 @@
 
 services:
   signer:
-    image: livepeer/go-livepeer:v0.9.0
+    # Master build, matching the orchestrator; see compose.orchestrator.yml.
+    image: livepeer/go-livepeer:sha-cc49228
     container_name: example_apps_signer
     command:
       - -remoteSigner
@@ -50,7 +51,10 @@ services:
       # Price comes from the app's registration; -pricePerUnit only sets the
       # unused legacy base price, but an on-chain O won't boot without it.
       - -pricePerUnit=0
-      - -ticketEV=1000000000
+      # Expected value per ticket. A payment needs fee / ticketEV tickets and the
+      # signer signs at most 100 at once, which is what caps the price. At 1e10
+      # echo's 0.10 USD per hour signs 15 tickets upfront, 6 per top-up.
+      - -ticketEV=10000000000
       - -maxFaceValue=4000000000000000
       - -monitor=false
       - -v=6

compose.orchestrator.yml

@@ -10,10 +10,15 @@
 #         service: orchestrator
 #
 # The go-livepeer image entrypoint is `livepeer`, so `command:` is flags only.
+#
+# Image: a master build, not a release tag. Metered sessions need the
+# session-scoped payment URL in the runner payment challenge (go-livepeer #4008),
+# which landed after v0.9.0; without it the SDK cannot keep a session funded.
+# Re-pin to the release tag once one ships with that change.
 
 services:
   orchestrator:
-    image: livepeer/go-livepeer:v0.9.0
+    image: livepeer/go-livepeer:sha-cc49228
     container_name: example_apps_orchestrator
     # Two distinct addresses:
     #   serviceAddr    advertised to external clients in discovery responses,

echo/.env.example

@@ -0,0 +1,25 @@
+# Copy to .env (gitignored) and fill in. Never commit secrets.
+# Keystore dirs: absolute paths OUTSIDE this repo, mounted read-only.
+
+NETWORK=arbitrum-one-mainnet
+ETH_RPC_URL=https://arb1.arbitrum.io/rpc
+
+# Signer (payer): needs an on-chain deposit + reserve.
+SIGNER_KEYSTORE_DIR=/absolute/path/to/signer-keystore
+SIGNER_ETH_ACCT=0xYourSignerAddress
+SIGNER_ETH_PASSWORD=your-signer-keystore-password
+
+# Orchestrator operating key (split-key): needs ETH for gas to redeem tickets.
+ORCH_KEYSTORE_DIR=/absolute/path/to/operator-keystore
+ORCH_ETH_ACCT=0xYourOperatorAddress
+ORCH_ETH_PASSWORD=your-operator-keystore-password
+# Registered orch = ticket recipient (-ethOrchAddr); empty = use the operating key.
+ORCH_ONCHAIN_ADDR=0xYourRegisteredOrchestrator
+
+# Runner price (on-chain): USD per hour, metered per second while the client
+# holds the session. Keep under ~0.67: the signer signs at most 100 tickets
+# per payment, and the demo orchestrator runs -ticketEV=1e10 (fee / ticketEV).
+PRICE=0.10
+# Signer's max-price cap (payer side), per billing unit. Metered here, so the
+# unit is one second and must exceed PRICE / 3600 (0.000111USD is ~0.40/hour).
+MAX_PRICE_PER_UNIT=0.000111USD

echo/README.md

@@ -8,7 +8,7 @@ A realtime video app on the Livepeer network: it receives a live video stream ov
 | Runner mode  | persistent (held-open session)       |
 | Registration | dynamic (self-registers via the SDK) |
 | Transport    | trickle (realtime video in/out)      |
-| Pricing      | none (offchain only)                 |
+| Pricing      | hour (metered per second)            |
 | Port         | 8989                                 |
 
 Prerequisites (Docker, `uv`, the not-yet-released SDK) and the shared setup are in the [repo README](../README.md).
@@ -19,6 +19,18 @@ The app is **dynamically registered**: it self-registers with the orchestrator v
 
 echo registers **dynamically** as the natural fit for a stateful app that already embeds the SDK (heartbeats, capacity, lifecycle). Trickle itself isn't tied to dynamic, though: `create_trickle_channels` rides the orchestrator's per-request `Livepeer-Session-Control` header, so a static runner exposing the same endpoints could open channels too.
 
+## Held-open sessions — what this shows
+
+**A trickle session is a pipe the client holds open, not a call it makes.** `hello-world`, `tiles` and `api-proxy` each answer one request and are finished. echo reserves a session once and then streams through it: frames go into the `in` channel and come back transformed from `out` continuously, with no request boundary in between.
+
+Two things follow from that, and they are what this example exists to show:
+
+- **The runner keeps state.** Each session owns its current transform and blur radius, which is why `POST /update` can change the effect mid-stream while frames keep flowing. A single-shot runner has nowhere to keep that between calls.
+- **Billing becomes a lifecycle.** There is no call to bill against, so the session is metered per second for as long as it is held, and payment repeats for the life of the stream instead of settling once. The [on-chain section](#run-on-chain-paid) below exercises exactly that.
+
+> [!NOTE]
+> The session ends when the client releases it, not when the input runs out, which is why the client calls `stop_runner_session` on the way out. echo registers the default capacity of 1, so one held session occupies the runner and `/discovery` reports `capacity_available: 0` until it is released. See [`tiles`](../tiles) for what capacity does under fan-out.
+
 ## Run offchain (free)
 
 Start the stack and confirm the runner registered:
@@ -30,10 +42,11 @@ curl -sk https://localhost:8935/discovery | jq '.[].runners[].app'   # confirm l
 
 The input is a file path, or `-` to read an MPEG-TS stream from stdin (so you can pipe in anything ffmpeg produces); the output is a file, or `-` to write the echoed stream to stdout (pipe it to a player).
 
-**From a file** — writes the result to `echo-out.ts`:
+**From a file** — writes the result to `echo-out.ts`. Any video works; the first command makes a 30s one (`-t` sets the length):
 
 ```sh
-uv run client.py --mode blur --discovery https://localhost:8935/discovery ~/samples/bbb_720p.mp4
+ffmpeg -f lavfi -i testsrc=size=1280x720:rate=30 -t 30 -c:v libx264 -preset ultrafast -pix_fmt yuv420p sample.mp4   # skip if you have a video
+uv run client.py --mode blur --discovery https://localhost:8935/discovery sample.mp4
 ```
 
 **Live from ffmpeg's test pattern** — no file needed; watch the test counter echo back in real time:
@@ -65,3 +78,32 @@ The `ffplay` low-delay flags (`-fflags nobuffer -flags low_delay -framedrop`) ke
 - `--radius N` sets the initial blur strength, `--max-frames N` stops early.
 
 Stop the stack with `docker compose down`.
+
+## Run on-chain (paid)
+
+Layer `compose.onchain.yml` to run the orchestrator on-chain with a remote signer paying for the session. This example showcases **metered pricing**: `PRICE` is USD per hour, billed per second for as long as the client holds the session, the natural fit for a stream with no fixed length. For the required RPC and wallets see [On-chain (paid) setup](../README.md#on-chain-paid-setup) in the repo README.
+
+```sh
+cp .env.example .env   # fill in RPC, network, keystore paths, accounts, pricing
+docker compose -f compose.yml -f compose.onchain.yml up -d --build
+uv run client.py sample.mp4 --mode blur \
+  --discovery https://localhost:8935/discovery \
+  --signer http://localhost:7936
+docker compose -f compose.yml -f compose.onchain.yml down
+```
+
+The ffmpeg and webcam pipes above work the same way on-chain: add `--signer` to the `client.py` in the pipeline. Without it the client stops at the payment challenge (`Live runner paid call requires signer_url`), which closes the pipe and leaves ffmpeg reporting `Broken pipe` — the paid stack refusing an unpaid caller, not a broken camera.
+
+A metered session pays **more than once**. The upfront payment that answers the 402 challenge only buys the signer's preroll (about ten seconds), while the orchestrator keeps debiting every few seconds and releases the session on the first debit it cannot cover. So `reserve_session` keeps the session funded in the background for as long as the client holds it, and leaving the client's `async with session` block stops that before the session is released. A stream that outlives the preroll is the whole point of this example on-chain, so use a clip of at least a few tens of seconds (the 30s `sample.mp4` above is enough): watch `docker compose logs -f orchestrator` and you should see repeated payments, not one.
+
+## Run without Docker
+
+Start an orchestrator built from go-livepeer `v0.9.0` or newer (see [Build from source](https://docs.livepeer.org/v1/orchestrators/guides/install-go-livepeer#build-from-source)), then the app and client directly:
+
+```sh
+./livepeer -orchestrator -useLiveRunners -serviceAddr localhost:8935 -orchSecret abcdef -v 6
+uv run runner.py --orchestrator https://localhost:8935 --orchSecret abcdef
+uv run client.py --mode blur sample.mp4
+```
+
+The paid path needs a newer orchestrator than the offchain one: metered sessions rely on the session-scoped payment URL added after `v0.9.0` ([#4008](https://github.com/livepeer/go-livepeer/pull/4008)), which is why the compose files pin a master build.

echo/client.py

@@ -6,9 +6,15 @@
 chain `ffmpeg -> client -> ffplay`.
 
 Livepeer integration (grep `# Livepeer:`):
-  1. reserve_session()        — discover the runner, reserve a session
+  1. reserve_session()        — discover the runner, reserve a session. On the paid
+                                path it answers the 402 challenge and then keeps the
+                                session funded in the background, because the price is
+                                metered: the orchestrator debits every few seconds and
+                                releases the session on the first debit it cannot cover.
   2. MediaPublish/MediaOutput — publish frames to `in`, read echoed frames from `out`
-  3. stop_runner_session()    — end the session (settles payment on-chain)
+  3. stop_runner_session()    — end the session. Leaving the `async with` stops the
+                                funding first, so nothing is paid for a session that
+                                is about to go away.
 """
 
 from __future__ import annotations
@@ -51,6 +57,9 @@ def _parse_args() -> argparse.Namespace:
         ),
     )
     parser.add_argument("--discovery", default=DEFAULT_DISCOVERY)
+    parser.add_argument(
+        "--signer", default="", help="Remote signer base URL (on-chain/paid path)."
+    )
     parser.add_argument(
         "--output",
         default=DEFAULT_OUTPUT,
@@ -189,41 +198,47 @@ async def main() -> None:
     session = None
 
     try:
-        session = await reserve_session(
-            discovery_url=args.discovery, app=APP_ID
-        )  # Livepeer: 1
+        session = await reserve_session(  # Livepeer: 1
+            discovery_url=args.discovery,  # omit if the signer does discovery itself
+            app=APP_ID,
+            signer_url=args.signer.strip() or None,
+        )
         log.info("session_id=%s app_url=%s", session.session_id, session.app_url)
 
-        echo = await post_json(
-            f"{session.app_url.rstrip('/')}/echo",
-            {"radius": args.radius, "mode": args.mode},
-        )
-        in_url = _channel_url(echo, "in")
-        out_url = _channel_url(echo, "out")
-        log.info("in=%s out=%s", in_url, out_url)
-
-        with (
-            nullcontext(sys.stdout.buffer) if output_stdout else output_path.open("wb")
-        ) as fh:
-
-            def _write_chunk(chunk: bytes) -> None:
-                fh.write(chunk)
-                if output_stdout:
-                    fh.flush()
-
-            async with MediaOutput(
-                out_url, on_bytes=_write_chunk
-            ):  # Livepeer: 2 (read echoed frames)
-                await _publish_video(
-                    input_source,
-                    in_url,
-                    max_frames=max(0, args.max_frames),
-                    app_url=session.app_url,
-                    mode=args.mode,
-                    blur_period=args.blur_period,
-                )
-                log.info("publish complete; waiting for output to drain...")
-            fh.flush()
+        # The session funds itself while it is held; leaving this block stops that.
+        async with session:
+            echo = await post_json(
+                f"{session.app_url.rstrip('/')}/echo",
+                {"radius": args.radius, "mode": args.mode},
+            )
+            in_url = _channel_url(echo, "in")
+            out_url = _channel_url(echo, "out")
+            log.info("in=%s out=%s", in_url, out_url)
+
+            with (
+                nullcontext(sys.stdout.buffer)
+                if output_stdout
+                else output_path.open("wb")
+            ) as fh:
+
+                def _write_chunk(chunk: bytes) -> None:
+                    fh.write(chunk)
+                    if output_stdout:
+                        fh.flush()
+
+                async with MediaOutput(
+                    out_url, on_bytes=_write_chunk
+                ):  # Livepeer: 2 (read echoed frames)
+                    await _publish_video(
+                        input_source,
+                        in_url,
+                        max_frames=max(0, args.max_frames),
+                        app_url=session.app_url,
+                        mode=args.mode,
+                        blur_period=args.blur_period,
+                    )
+                    log.info("publish complete; waiting for output to drain...")
+                fh.flush()
     except LivepeerGatewayError as exc:
         raise SystemExit(f"ERROR: {exc}") from exc
     finally:

echo/compose.onchain.yml

@@ -0,0 +1,34 @@
+# On-chain payment overlay for echo. Layer it on the offchain base:
+#   docker compose -f compose.yml -f compose.onchain.yml up -d --build
+#
+# Adds the shared remote signer, re-points the orchestrator on-chain (see
+# ../compose.onchain.yml), and registers the app with a price so the orchestrator
+# issues a payment challenge. Requires a local .env (gitignored); copy .env.example
+# and fill it in. The session is metered, so a stream is billed for as long as it
+# runs. Then pay through the signer:
+#   uv run client.py sample.mp4 --mode blur \
+#     --discovery https://localhost:8935/discovery \
+#     --signer http://localhost:7936
+
+services:
+  signer:
+    extends:
+      file: ../compose.onchain.yml
+      service: signer
+    ports:
+      - "7936:7936"
+
+  orchestrator:
+    extends:
+      file: ../compose.onchain.yml
+      service: orchestrator
+
+  # Re-declare the command to advertise a price (base file registers free).
+  app:
+    command:
+      - --host=0.0.0.0
+      - --orchestrator=https://orchestrator:8935
+      - --orchSecret=abcdef
+      - --runner-url=http://app:8989
+      # Billed per second of session (metered); price cap in .env.example.
+      - --price=${PRICE}

echo/compose.yml

@@ -4,7 +4,7 @@
 # pulled in with `extends`; this file only adds the app. Once up, drive it from
 # the host with the SDK client and a sample video:
 #   docker compose up -d --build
-#   uv run client.py --blur ~/samples/clip.mp4   # writes echo-out.ts
+#   uv run client.py --mode blur sample.mp4   # writes echo-out.ts
 
 services:
   orchestrator: