diff --git a/Cargo.lock b/Cargo.lock index 8f77d76..69e9bd2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1973,6 +1973,7 @@ dependencies = [ name = "transcodarr-server" version = "0.1.0" dependencies = [ + "blake3", "serde", "serde_json", "tempfile", diff --git a/NEXT-SESSION.md b/NEXT-SESSION.md new file mode 100644 index 0000000..5e8bd61 --- /dev/null +++ b/NEXT-SESSION.md @@ -0,0 +1,138 @@ + + + + + +# Goal: a real transcode on the GPU node, both transports, audio and video + +**Still not achieved.** No transcode has run on U1 by either method. What +changed on 2026-08-12 is that two of the three blockers are gone, and the third +is now precisely scoped rather than mysterious. + +## Done 2026-08-12 + +### The hung `cargo test --workspace` — root-caused, fixed, merged (`c75a72c`) + +It was never infrastructure. Last night's run was still alive after **26 hours** +at 0% CPU; sampling it named the frame outright. + +`Shutdown::stop` signalled with `Notify::notify_waiters`, which wakes only tasks +**already parked** at that instant and stores no permit. Both waits — the +session loop and the reconnect backoff — awaited `notified()` directly with no +level-triggered read of the flag `stop` had already set. When `stop` landed +while the client was anywhere else (mid-recovery, mid-register, mid-dispatch) +the wakeup was lost, and the session loop parked forever on a notification that +had already been and gone. + +`Shutdown::cancelled` now registers as a waiter *before* reading the flag, and +`stop` writes the flag *before* notifying — two opposed orderings, so no +interleaving misses both. + +| measurement | result | +|---|---| +| the one test ×260, pre-fix branch | 5 hangs | +| same test ×260, pre-merge `main` | 0 (latent, not absent) | +| **with fix**, whole `connect_client` binary ×150 | **0** | +| CI `Test Rust` | 2m00s, was 20+ min | + +**Do not repeat the mistaken lesson:** the stated suspects (the +`FetchSource`/`PushOutput` stubs) were inert. And a single-run bisect against +`main` would have wrongly convicted the branch — the defect is latent there too. +For an intermittent fault, measure a *rate*, and sample a live hung process +rather than reasoning about it. + +### The agent runs on Windows — cross-compiled and executed on U1 + +`x86_64-pc-windows-gnu` builds clean after `brew install mingw-w64`. `ring` is +in the tree (via `tonic`'s `tls` → `rustls`) and cross-built without trouble, so +no workspace feature surgery was needed. 11 MB PE32+ binary, copied to +`C:\Users\jdfalk\transcodarr-agent.exe`, and its capability survey **ran on the +box**: + +```text +platform windows classes audio, gpu transport stream +encoders hevc_nvenc, eac3, ac3, aac mounts none +``` + +Note what is *absent*: `av1_nvenc`. transcodarr's trial-probe correctly refused +the codec that ffmpeg falsely advertises. The trap held. + +Rebuild with: + +```sh +export CARGO_TARGET_X86_64_PC_WINDOWS_GNU_LINKER=x86_64-w64-mingw32-gcc +export CC_x86_64_pc_windows_gnu=x86_64-w64-mingw32-gcc +export AR_x86_64_pc_windows_gnu=x86_64-w64-mingw32-ar +cargo build --release --target x86_64-pc-windows-gnu -p transcodarr-agent --bin transcodarr-agent +``` + +ffmpeg on that box lives at `C:\Users\jdfal\bin\ffmpeg.exe` (note: `jdfal`, not +`jdfalk`). + +### Streaming groundwork + the spec (PR #80) + +`transcodarr-server::transfer` — chunked read, and a receiving `Sink` that +checks `offset` and verifies a whole-file blake3 before the bytes are used. +`distributed-architecture.md` now documents both transports; it previously had +zero mentions of `TransportMode`, `FetchSource`, `PushOutput`, upload, download +or byte ranges. + +## What is left, in order + +1. **Wire the RPCs.** `AgentSession::fetch_source` / `push_output` still return + `unimplemented`. `fetch_source` is small: check the epoch the way + `CommitReport` does, look the job's source path up, hand it to + `transfer::source_stream`. +2. **Server-side install for `push_output`.** Receive into a staging path with + `transfer::Sink`, then follow `runner.rs:281–350` *literally*, including the + ordering the comments there defend: `CommitIntentRepo::grant_op` **before** + the ritual touches anything, `resolve_op` after whatever happened, the + `trash_entry` row **last**. `AgentSession` will need a `WorkArea` + + `CommitRitual` (constructor change; 3 call sites). +3. **The agent-side stream path** in `worker.rs`. Under `Stream`, fetch instead + of reading `source_path`, and **do not** run the local ritual + (`commit_blocking`, worker.rs:373). + **The trap:** the intent journal exists to recover half-finished *installs*, + and a streaming agent never installs. Reusing mount-mode journalling as-is + writes `IntentRecord`s that can never resolve and hands the server + `live_intents` for work it must not fence on. Decide what `live_intents()` + returns under `Stream` — plausibly empty — and write down why. +4. **Prove it locally before Windows.** Server and agent both on the Mac, agent + `--transport stream`, a real audio transcode end to end. None of that needs + the GPU node. If it works locally, Windows is deployment; if it does not, you + debug byte plumbing without mingw, NVENC and logon sessions confounding it. +5. **Then U1:** stream-mode audio, then video with `hevc_nvenc`. +6. **Mount mode last** — see below, it needs hands on the box. + +## Mount mode needs the owner at the console + +Not a software problem. Windows drive mappings are per-user **and +per-logon-session**, and SSH lands in a separate elevated one: all three SMB +mounts show `Unavailable`, UNC is denied, `cmdkey` is empty. So the agent must +run **in the interactive session** — a scheduled task set to "run only when the +user is logged on", an autologon, or a startup shortcut. That cannot be arranged +over SSH, so it is a request for the owner, not a task for the next session. + +Prefer UNC paths over drive letters in the mount table regardless: a drive +letter is a per-session alias, a UNC path is not (though it still needs +credentials in whatever session resolves it). + +## Traps still standing + +- Turing NVDEC cannot decode AV1 (exit 69, ~1 KB truncated) or 10-bit H.264 + (silent software fallback). Use an **8-bit H.264 source**, software decode, + NVENC encode. Gate hardware decode per codec, never globally. +- **Probe by trial, never by asking ffmpeg what it lists.** Confirmed again + today: that ffmpeg lists `av1_nvenc` and the hardware cannot do it. +- That ffmpeg build has **no libx264** — no software video fallback on the box. +- ICMP is filtered on U1. `ping` failing proves nothing; test port 22. +- The `FakeServer` in `connect_client.rs` refuses to serve bytes on purpose. + Do not weaken it to make a streaming test pass — a fake that returned an empty + stream would let a streaming test pass while moving no bytes. + +## The rule that keeps being right + +An invariant verified where it is cheap to measure will be violated where it +matters. A green suite is evidence about the tests. Today's version: NVENC +"confirmed working" by standalone ffmpeg still has not transcoded anything +through transcodarr. **Run the thing.** diff --git a/changelog.d/20260812-streaming-bytes-and-spec.md b/changelog.d/20260812-streaming-bytes-and-spec.md new file mode 100644 index 0000000..bf733f5 --- /dev/null +++ b/changelog.d/20260812-streaming-bytes-and-spec.md @@ -0,0 +1,35 @@ + + + + + +### Added + +#### The byte plumbing for the streaming transport + +`transcodarr-server::transfer` reads a source into an ordered chunk stream and +receives chunks into a file the server can vouch for. Nothing is wired to the +RPCs yet — `FetchSource` and `PushOutput` still refuse explicitly — so this adds +no behaviour an agent can reach. + +Two properties carry the correctness, and both have tests that fail without +them. `offset` is checked rather than trusted, so a restarted stream is detected +instead of being appended to. And the final chunk carries a blake3 of the whole +file, checked before the bytes are used for anything: a truncated transfer is +*smaller*, and size is never an accept criterion here, so without the hash a +half-received encode would install cleanly over a good original. + +### Documentation + +#### The architecture document describes both transports + +`distributed-architecture.md` documented only direct mounted access. It +contained no mention of upload, download, byte ranges, `TransportMode`, +`FetchSource` or `PushOutput`, and stated `MountCovers` as an unconditional +requirement — which made the second transport impossible to express. A node with +no usable share was undispatchable no matter how good its encoder was. + +That was a defect in the document, not a decision: the code faithfully +implemented a spec that had dropped the requirement. The new *Transport modes* +section sets out both modes, why the server performs the commit ritual when the +agent cannot, and why `MountCovers` applies only to mount-mode nodes. diff --git a/crates/transcodarr-server/Cargo.toml b/crates/transcodarr-server/Cargo.toml index 1ed4989..a6e74ef 100644 --- a/crates/transcodarr-server/Cargo.toml +++ b/crates/transcodarr-server/Cargo.toml @@ -16,6 +16,7 @@ homepage.workspace = true # The only crate that links the store. An agent must stay copyable to the # Windows node without dragging SQLite along with it. [dependencies] +blake3 = { workspace = true } serde = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } diff --git a/crates/transcodarr-server/src/lib.rs b/crates/transcodarr-server/src/lib.rs index 39dbab0..fd2d8b6 100644 --- a/crates/transcodarr-server/src/lib.rs +++ b/crates/transcodarr-server/src/lib.rs @@ -31,6 +31,7 @@ pub mod schedule; pub mod serve; pub mod session; pub mod summary; +pub mod transfer; pub use capacity::{AgentLimits, CapacityLedger, Grant, Refusal}; pub use dispatch::{AgentEntry, Assignment, Blocked, DispatchRound, Dispatcher, QueuedJob}; diff --git a/crates/transcodarr-server/src/transfer.rs b/crates/transcodarr-server/src/transfer.rs new file mode 100644 index 0000000..d8e614a --- /dev/null +++ b/crates/transcodarr-server/src/transfer.rs @@ -0,0 +1,318 @@ +// file: crates/transcodarr-server/src/transfer.rs +// version: 1.0.0 +// guid: 9c1f6b28-4a70-4de3-8f52-6b0d7ae94c11 +// last-edited: 2026-08-12 +//! Moving file bytes for [`TransportMode::Stream`]. +//! +//! A streaming agent never resolves a canonical path. It is handed the source +//! bytes, works on a local copy, and sends the result back for the server to +//! install. This module is only the byte plumbing; the install itself is the +//! ordinary commit ritual, performed server-side because a streaming agent has +//! no path to the destination and could not perform it if it wanted to. +//! +//! ## Why a chunk carries its offset and a signature +//! +//! `offset` is carried rather than implied so a short read is detectable. A +//! receiver that has written 900 MB and is then handed a chunk claiming offset +//! zero knows the stream restarted, instead of appending and producing a +//! corrupt file that happens to be a plausible length. +//! +//! The last chunk carries a blake3 of the whole file, and [`Sink`] refuses the +//! transfer when it does not match. This is the one gate that matters: a +//! truncated transfer is *smaller*, and size is never an accept criterion in +//! this project. Without the hash a half-received encode would install +//! cleanly and silently replace a good original with a broken file. + +use std::io::{Read, Write}; +use std::path::{Path, PathBuf}; + +use tokio::sync::mpsc; +use tokio_stream::wrappers::ReceiverStream; +use tonic::Status; + +use transcodarr_proto::pb; + +/// Bytes per chunk. +/// +/// Well under tonic's 4 MiB default message ceiling, with room for the frame +/// and the other fields: a chunk that trips `max_decoding_message_size` fails +/// the whole transfer, and picking a size near the limit makes that a function +/// of how large the other fields happen to be. +pub const CHUNK_BYTES: usize = 512 * 1024; + +/// How many chunks may be buffered ahead of a slow consumer. +const QUEUE_DEPTH: usize = 4; + +/// Read `path` and produce its chunks in order. +/// +/// The read happens on the blocking pool: these are multi-gigabyte files, and +/// doing that on a runtime worker would stall every other agent's stream. +pub fn source_stream( + job_id: String, + attempt: u32, + path: PathBuf, +) -> ReceiverStream> { + let (tx, rx) = mpsc::channel(QUEUE_DEPTH); + + tokio::task::spawn_blocking(move || { + let mut file = match std::fs::File::open(&path) { + Ok(f) => f, + Err(e) => { + // Blocking send: the receiver is a live gRPC stream, and + // dropping the error would hand the agent an empty file. + let _ = tx.blocking_send(Err(Status::not_found(format!( + "source {} cannot be opened: {e}", + path.display() + )))); + return; + } + }; + + let mut hasher = blake3::Hasher::new(); + let mut buf = vec![0u8; CHUNK_BYTES]; + let mut offset: u64 = 0; + + loop { + let read = match file.read(&mut buf) { + Ok(n) => n, + Err(e) => { + let _ = tx.blocking_send(Err(Status::internal(format!( + "reading {} at offset {offset}: {e}", + path.display() + )))); + return; + } + }; + + if read == 0 { + // The terminator is its own chunk carrying the signature. An + // explicit end beats "the stream closed", which is + // indistinguishable from the sender dying mid-file. + let _ = tx.blocking_send(Ok(pb::FileChunk { + job_id, + attempt, + offset, + data: Vec::new(), + last: true, + content_sig: hasher.finalize().to_hex().to_string(), + })); + return; + } + + hasher.update(&buf[..read]); + let chunk = pb::FileChunk { + job_id: job_id.clone(), + attempt, + offset, + data: buf[..read].to_vec(), + last: false, + content_sig: String::new(), + }; + offset += read as u64; + + if tx.blocking_send(Ok(chunk)).is_err() { + // The agent hung up. Nothing to report to: stop reading rather + // than burn I/O on a transfer nobody will receive. + return; + } + } + }); + + ReceiverStream::new(rx) +} + +/// Receives chunks into a file, refusing anything it cannot vouch for. +pub struct Sink { + path: PathBuf, + file: std::fs::File, + hasher: blake3::Hasher, + written: u64, + finished: bool, +} + +impl Sink { + /// Open `path` for a new transfer, truncating any previous attempt. + pub fn create(path: &Path) -> std::io::Result { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent)?; + } + Ok(Self { + path: path.to_path_buf(), + file: std::fs::File::create(path)?, + hasher: blake3::Hasher::new(), + written: 0, + finished: false, + }) + } + + /// Bytes accepted so far. + pub fn written(&self) -> u64 { + self.written + } + + /// Where the bytes are landing. + pub fn path(&self) -> &Path { + &self.path + } + + /// Accept one chunk. Returns `true` once the transfer is complete. + /// + /// Every rejection here leaves a partial file on disk on purpose — the + /// caller removes it. Returning an error while pretending the file is gone + /// would be a worse lie than the partial file is a mess. + pub fn accept(&mut self, chunk: &pb::FileChunk) -> Result { + if self.finished { + return Err("a chunk arrived after the transfer was declared complete".into()); + } + // The gap check. Out-of-order or restarted streams are the failure this + // exists to catch, and appending regardless is how you get a file of + // the right length and the wrong contents. + if chunk.offset != self.written { + return Err(format!( + "chunk claims offset {} but {} bytes have been written; the stream restarted or lost a chunk", + chunk.offset, self.written + )); + } + + if !chunk.data.is_empty() { + self.file + .write_all(&chunk.data) + .map_err(|e| format!("writing at offset {}: {e}", self.written))?; + self.hasher.update(&chunk.data); + self.written += chunk.data.len() as u64; + } + + if !chunk.last { + return Ok(false); + } + + self.file + .flush() + .map_err(|e| format!("flushing {}: {e}", self.path.display()))?; + self.file + .sync_all() + .map_err(|e| format!("syncing {}: {e}", self.path.display()))?; + + let actual = self.hasher.finalize().to_hex().to_string(); + if actual != chunk.content_sig { + return Err(format!( + "content signature mismatch after {} bytes: sender said {}, received bytes hash to {}", + self.written, chunk.content_sig, actual + )); + } + + self.finished = true; + Ok(true) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn chunk(offset: u64, data: &[u8], last: bool, sig: &str) -> pb::FileChunk { + pb::FileChunk { + job_id: "job-1".into(), + attempt: 0, + offset, + data: data.to_vec(), + last, + content_sig: sig.into(), + } + } + + fn sig_of(bytes: &[u8]) -> String { + blake3::hash(bytes).to_hex().to_string() + } + + #[test] + fn a_whole_transfer_lands_and_verifies() { + let dir = tempfile::TempDir::new().unwrap(); + let dest = dir.path().join("out.mkv"); + let mut sink = Sink::create(&dest).unwrap(); + + assert!(!sink.accept(&chunk(0, b"hello ", false, "")).unwrap()); + assert!(!sink.accept(&chunk(6, b"world", false, "")).unwrap()); + assert!( + sink.accept(&chunk(11, b"", true, &sig_of(b"hello world"))) + .unwrap() + ); + + assert_eq!(std::fs::read(&dest).unwrap(), b"hello world"); + assert_eq!(sink.written(), 11); + } + + /// The failure the signature exists for. A truncated transfer produces a + /// *smaller* file, so length proves nothing — this is the same rule as the + /// encode side, where a truncated output is never accepted on size. + #[test] + fn a_truncated_transfer_is_refused_rather_than_installed() { + let dir = tempfile::TempDir::new().unwrap(); + let mut sink = Sink::create(&dir.path().join("out.mkv")).unwrap(); + + sink.accept(&chunk(0, b"hello ", false, "")).unwrap(); + // The sender died and something closed the transfer early: the bytes + // present are a clean prefix, and only the hash reveals it. + let err = sink + .accept(&chunk(6, b"", true, &sig_of(b"hello world"))) + .unwrap_err(); + + assert!(err.contains("content signature mismatch"), "{err}"); + } + + /// A restarted stream must not be appended to the bytes already written. + #[test] + fn a_chunk_at_the_wrong_offset_is_refused() { + let dir = tempfile::TempDir::new().unwrap(); + let mut sink = Sink::create(&dir.path().join("out.mkv")).unwrap(); + + sink.accept(&chunk(0, b"hello ", false, "")).unwrap(); + let err = sink.accept(&chunk(0, b"hello ", false, "")).unwrap_err(); + + assert!(err.contains("claims offset 0"), "{err}"); + } + + #[tokio::test] + async fn a_file_round_trips_through_the_stream_and_the_sink() { + use tokio_stream::StreamExt; + + let dir = tempfile::TempDir::new().unwrap(); + // Deliberately larger than one chunk, and not a multiple of it: an + // off-by-one in the final partial read is the bug this shape catches. + let body: Vec = (0..(CHUNK_BYTES * 2 + 12345)) + .map(|i| (i % 251) as u8) + .collect(); + let src = dir.path().join("in.mkv"); + std::fs::write(&src, &body).unwrap(); + + let mut stream = source_stream("job-1".into(), 0, src); + let dest = dir.path().join("out.mkv"); + let mut sink = Sink::create(&dest).unwrap(); + + let mut done = false; + let mut chunks = 0; + while let Some(next) = stream.next().await { + chunks += 1; + done = sink.accept(&next.unwrap()).unwrap(); + } + + assert!( + done, + "the stream must terminate with an explicit last chunk" + ); + assert!(chunks >= 4, "expected several chunks, got {chunks}"); + assert_eq!(std::fs::read(&dest).unwrap(), body); + } + + #[tokio::test] + async fn a_missing_source_reports_rather_than_streaming_nothing() { + use tokio_stream::StreamExt; + + let dir = tempfile::TempDir::new().unwrap(); + let mut stream = source_stream("job-1".into(), 0, dir.path().join("absent.mkv")); + + let first = stream.next().await.expect("an error, not an empty stream"); + let status = first.expect_err("a missing source must not look like an empty file"); + assert_eq!(status.code(), tonic::Code::NotFound); + } +} diff --git a/docs/design/distributed-architecture.md b/docs/design/distributed-architecture.md index 95d0c36..bfc5177 100644 --- a/docs/design/distributed-architecture.md +++ b/docs/design/distributed-architecture.md @@ -1,7 +1,7 @@ - + - + # transcodarr — Distributed Transcode Orchestrator @@ -978,6 +978,45 @@ The agent replays its fsynced `IntentJournal` and `InflightJournal` **before** s One unary RPC for the handshake and one long-lived bidirectional stream for everything else. A single stream means connection liveness *is* agent liveness, ordering is guaranteed, and there is no way for one channel to be healthy while another is wedged. +### Transport modes + +There are **two** ways an agent reaches media, chosen per node by the operator with `--transport`. Earlier revisions of this document described only the first and stated `MountCovers` as an unconditional requirement, which made the second impossible to express. That was a defect in this document, not a decision: a node with no usable share was undispatchable no matter how good its encoder was. + +| | `TM_MOUNT` | `TM_STREAM` | +|---|---|---| +| Reaches media by | a share, with per-node path translation | bytes over the agent connection | +| Advertises mounts | yes | **no** — it resolves no canonical paths | +| Needs server storage layout | yes | no | +| Performs the commit ritual | the agent | **the server** | +| `MountCovers` applies | yes | no | + +`TM_MOUNT` is the zero value on purpose. proto3 gives an absent field the zero value, so an agent that has never heard of transports keeps behaving exactly as it did. Silently switching an existing agent to streaming because a new field defaulted the other way would be the worst possible way to introduce this. + +#### Why the server commits in streaming mode + +A streaming agent has no path to the destination. It receives bytes, encodes, and returns bytes; it *cannot* perform the nine-step ritual — rename original to trash, install replacement, resolve the intent — because every path in it is one the agent cannot see. So under `TM_STREAM` the ritual moves server-side and runs exactly as it does for `admin run`, with the same ordering: the ledger intent is granted **before** anything on disk is touched, resolved afterwards whatever happened, and the `trash_entry` row written only once the original really is in the trash. + +This is not a new shape. It is the one already specified for a WSL2 node failing `RenameProbe`, where "the GPU agent becomes produce-only and a U0-local agent performs commits". Produce-only *is* streaming mode's agent. + +#### The two byte-moving RPCs + +```protobuf +rpc FetchSource(FetchSourceRequest) returns (stream FileChunk); +rpc PushOutput(stream FileChunk) returns (PushOutputResponse); +``` + +Both are `TM_STREAM` only; a `TM_MOUNT` agent never calls either. `FetchSourceRequest` carries a `fencing_epoch` and is rejected when stale, exactly as a `CommitReport` is — an agent fenced out must not be able to pull bytes for work it no longer owns. + +`FileChunk` carries its own `offset` rather than implying it, so a short read is detectable: a receiver that has written 900 MB and is handed a chunk claiming offset zero knows the stream restarted, instead of appending and producing a corrupt file of an entirely plausible length. The final chunk sets `last` and carries a blake3 of the whole file, and the receiver verifies before acting. + +That hash is the gate that matters. **A truncated transfer is smaller, and size is never an accept criterion** — the same rule the duration gate enforces on the encode side. Without the hash, a half-received encode installs cleanly over a good original. + +An explicit `last` beats "the stream closed", which is indistinguishable from the sender dying mid-file. + +#### Consequence for dispatch + +`Requirement::MountCovers` is satisfied automatically by a `TM_STREAM` agent, because coverage of a path it never resolves is not a meaningful question. This is a dispatch-matching change, not merely a file-copying feature: it is what makes a mountless node eligible for work at all. + ### The `.proto` file `crates/transcodarr-proto/proto/transcodarr/v1/agent.proto`: