transcode

func (*Coordinator) Cancel ¶ added in v0.1.3

func (c *Coordinator) Cancel(batchID uuid.UUID) error

Cancel transitions a batch to `cancelled` AND tries to drain the pool of its remaining jobs. The pool doesn't currently support targeted dequeue (jobs are FIFO; the next worker pulls whichever is at the head), so the practical effect is:

• any worker that started a job from this batch finishes it (the variant lands on disk + in `track_variants`)
• any job still on the pool's pending channel runs when picked up (variant also lands)
• the batch row is marked `cancelled` so the admin Jobs page stops surfacing it as live

"Cancel" therefore means "stop tracking; stop counting" rather than "kill in-flight." A future enhancement can add per-batch dequeue if operators need a harder stop.

func (*Coordinator) OnJobComplete ¶ added in v0.1.3

func (c *Coordinator) OnJobComplete(path, variantID string, sampleRate, bitsPerSample int, durationSeconds float64, batchID uuid.UUID, completedAt time.Time)

OnJobComplete is the Pool-side callback invoked when a sox invocation succeeded AND `UpsertVariant` committed. Signature matches the pool's `SetOnJobComplete` parameter exactly.

`durationSeconds` is the wall-clock cost of the sox run (captured at worker startedAt → completedAt). Feeds the rolling throughput ring so ETAs reflect real bridge throughput; previously this path passed 0, leaving the ring empty even on a busy bridge (CodeRabbit high on PR #201).

Bumps `processed_files` on the owning batch (no-op when batchID is zero — legacy single-track jobs that didn't go through Submit don't have a batch to attribute to). Transitions the batch to `completed` when all expected jobs have reported back.

func (*Coordinator) OnJobFailed ¶ added in v0.1.3

func (c *Coordinator) OnJobFailed(path, variantID, errMsg string, durationSeconds float64, batchID uuid.UUID, failedAt time.Time)

OnJobFailed mirrors OnJobComplete for the failure side. Bumps `failed_files`; if every job in the batch has reported back and failed_files > 0, transitions the batch to `failed`.

func (*Coordinator) SetPublish ¶ added in v0.1.3

func (c *Coordinator) SetPublish(publish CoordinatorPublishFunc)

SetPublish wires (or rewires) the SSE broker emit closure. Construction-time callers may pass nil and call SetPublish later once the broker handle is in scope (the canonical wiring pattern in cmd/bridge/main.go — the Coordinator is built BEFORE the SSE broker so `RecoverInterruptedBatches` can run synchronously during startup). Concurrency-safe via the same coordinator mutex that guards the live-batch state.

func (*Coordinator) Submit ¶ added in v0.1.3

func (c *Coordinator) Submit(ctx context.Context, path string, targetRate, targetBits int, outputDir string) (*SubmitResult, error)

Submit walks every track under `path`, filters ineligible / already-covered, computes the projected variant size, refuses on insufficient disk headroom, inserts an `upscale_batches` row, and enqueues filtered jobs into the pool with their `BatchID` stamped. Returns ErrInsufficientDiskSpace (wrapped) if pre-flight refuses.

`targetRate` / `targetBits` come from the admin Settings (DB- backed via `Store.GetUpscaleTarget`); the caller is responsible for falling back to YAML bootstrap when scan_state is unseeded.

`outputDir` is `<dataDir>/transcoded` — same path the legacy per-track CLI / HTTP path uses. The Pool dedups on `(source, variant_id)` so a track that already has a variant at the same target is silently skipped (counted as `AlreadyCovered`).

func (*Coordinator) SubmitOptimize ¶ added in v0.1.4

func (c *Coordinator) SubmitOptimize(ctx context.Context, path string, outputDir string) (*SubmitResult, error)

SubmitOptimize is the CarPlay-targeted batch entry point. Pipeline: build candidates → disk-headroom preflight → insert batch row → enqueue jobs (with truncation handling). The pure-helpers split keeps cognitive complexity below the repo gate while preserving the structural pipeline upscale's `Submit` follows.

func (*Coordinator) Throughput ¶ added in v0.1.3

func (c *Coordinator) Throughput() ThroughputSnapshot

Throughput returns the rolling average jobs-per-hour + a forward projection for the currently-live batches' remaining files. Both numbers are zero when fewer than `throughputMinSamples` samples have landed.

func (*InsufficientDiskSpaceError) Error ¶ added in v0.1.3

func (e *InsufficientDiskSpaceError) Error() string

func (*InsufficientDiskSpaceError) Unwrap ¶ added in v0.1.3

func (e *InsufficientDiskSpaceError) Unwrap() error

func (*JobSpec) FreshnessFromFile ¶

func (j *JobSpec) FreshnessFromFile() error

FreshnessFromFile populates the SourceMTimeNS / SourceSize fields on a JobSpec by stat-ing SourceAbsPath. Used by the CLI to capture "what version of the source did we convert from" at the moment of conversion, which the variant-resolve path later uses to detect drift.

func (JobSpec) SidecarPath ¶

func (j JobSpec) SidecarPath() string

SidecarPath returns the absolute filesystem path the converted FLAC will land at. Filename pattern (v1.4):

<OutputDir>/<libRel-dirname>/<libRel-basename>.<VariantID>.flac

Example: SourceLibraryRel `Diana Krall/The Look of Love/01 Love Letters.flac` + VariantID `upscaled-v2-176400-24` →

<OutputDir>/Diana Krall/The Look of Love/01 Love Letters.flac.upscaled-v2-176400-24.flac

**Why source-path-mirrored layout**: operators with write access to their library can `mv` the OutputDir contents directly into the library and slot variants alongside source files. The variantID suffix on the filename is load-bearing (multiple targets coexist: `01 Love Letters.flac.upscaled-v2-176400-24.flac` next to `01 Love Letters.flac.upscaled-v2-352800-24.flac`).

**Why keep the original extension** (`.flac`) in the basename: a folder with `Track1.flac` AND `Track1.wav` (legit A/B testing case) would otherwise both write to `Track1.upscaled-…flac` — zero collision tolerance for that exact case is structural.

**Migration**: existing hash-flat sidecars on disk stay served by their existing `track_variants.sidecar_path` rows (absolute paths; DB lookup is the only resolver — never filesystem-walked). Only newly-transcoded variants use the new layout. Mixed-layout state is fine; `bridge variants relayout` (separate CLI) opt-in renames legacy variants to match.

**Filesystem safety**: long basenames + Windows-illegal characters + reserved names route through `safeVariantFilename` which:

• replaces `: * ? " < > |` with `_` (deterministic),
• appends a `~<sha8>.` disambiguation segment derived from the RAW (pre-sanitization) basename bytes whenever sanitization touched the name OR the total would exceed 255 bytes (ext4 / NTFS / exFAT cap). The raw-bytes hash is load-bearing: two source files differing only in FAT-illegal characters (`Track:A.flac` + `Track*A.flac`) sanitize to identical bytes, and pre-fix the hash was computed over the sanitized form so it collapsed too — silently overwriting one variant with the other via `os.Rename`. The current shape keeps distinct raw inputs distinct on disk; DO NOT move the hash computation past `sanitiseForFAT` at any new call site.

Dirname segments are NOT sanitised — they came from the source filesystem which already accepts them.

func (JobSpec) SoxArgs ¶

func (j JobSpec) SoxArgs() ([]string, string)

SoxArgs builds the argv for the sox invocation. Returns the exact slice exec.Command will receive (including the leading input/output sentinels) and a JSON-friendly settings string for `track_variants.sox_settings` (forensic record of what produced this sidecar). Pure function — no I/O — so unit tests can pin the exact command shape without invoking sox.

Quality preset → `rate` flag mapping:

QualityVeryHigh → "-v"  (very high; ~95 dB stopband, 32-bit internal)
QualityHigh     → "-h"
QualityMedium   → "-m"

`dither -s` is shaped TPDF dither when truncating the 32-bit internal float to the 24/32-bit integer FLAC output. Required for audible transparency at 24-bit and benign at 32-bit (the dither noise is below the LSB).

`-G` (gain-guard, leading global option) instructs sox to pre- scan the input for the headroom needed to prevent clipping through the rate-conversion + dither pipeline and apply a matching attenuation. Required for upscaling 0 dBFS-mastered material (most modern pop / loudness-war masters): without it, intersample peaks the rate filter reconstructs above 0 dBFS would clip on the integer FLAC output. The attenuation is computed from the actual peak — typically well under 1 dB and inaudible — and only fires when the source has the headroom problem. Cost is one extra peak-scan pass; invisible against the rate-conversion CPU dominating the run.

`rate … -L` pins LINEAR phase response on the resampler. This is already sox's default for the `rate` effect, so the pinned form is byte-identical to the prior unpinned `rate -v <Hz>` output — verified with deterministic `sox -R` runs (the signal-path md5 matches; only the always-present dither PRNG differs run-to-run). Pinning it is pure defense: a future sox release that changed the default phase could otherwise silently flip our cached variants to minimum/intermediate phase, and it lets the admin worker grid label the chain "linear phase" truthfully. Because the deterministic output is unchanged, NO VariantSchemaVersion bump / regeneration is required — existing sidecars stay valid.

func (JobSpec) VariantID ¶

func (j JobSpec) VariantID() string

VariantID returns the opaque identifier that uniquely names this JobSpec's output variant. Convention:

upscaled-<schemaVersion>-<targetRate>-<targetBits>   // JobKindUpscale (default)
optimized-<schemaVersion>-<targetRate>-<targetBits>  // JobKindOptimize

e.g. `upscaled-v2-176400-24` or `optimized-v2-44100-16`. iOS keys on the prefix to slot the variant into the share-level "prefer upscaled" toggle vs. the runtime CarPlay-routing path. Future variant kinds (e.g. PCM→DSD synthesis) get their own prefix.

Hot path during manifest scan + every pool callback. The finite (rate × bits) cross-product across all real DACs makes memoization a clean O(1) win — two parallel caches keep the prefix discriminator strict at the SQL boundary (see the docblock on the prefix constants).

**Don't share `variantIDCache` between kinds** — pre-seeded with `upscaled-v2-*` for `(44100, 16)`, returning that for an optimize job would silently emit the wrong variant ID into `track_variants`.

func (*Pool) ActiveWorkers ¶ added in v0.1.7

func (p *Pool) ActiveWorkers() []ActiveJobView

ActiveWorkers returns one snapshot per worker slot (WorkerID 0..N-1), Busy=false for idle workers. Lock-free: each slot is an atomic.Pointer read; the *ActiveJob behind it is immutable. Order is stable (slot index), so the SSE worker grid renders deterministically.

func (*Pool) DropInflight ¶ added in v0.1.3

func (p *Pool) DropInflight(matches func(sourcePath string) bool) int

DropInflight removes entries from the dedup `inflight` map whose source_path satisfies the supplied predicate. Returns the count dropped.

Used by the variant-delete handler (DELETE /v1/upscale/variants) and the integrity watcher to make sure a re-submission of an upscale request for a deleted path doesn't no-op against the stale dedup slot — without dropping the entry, a follow-up Enqueue would silently coalesce against an in-flight worker that's about to write a sidecar the caller intends to delete.

Does NOT cancel in-flight workers — there is no per-job cancellation primitive (workers run under the pool's stopCtx only). The unlink race is the caller's problem; the caller removes the sidecar file BEFORE the worker can write it, and if a worker beats the unlink the next `--gc` reverse pass / the integrity watcher reaps the orphan within ≤1 h. Document the race shape at the call site, not here.

Lock discipline: `p.mu` held only for the iteration window. Predicate is called synchronously under the lock; predicate authors keep predicates allocation-light (no DB calls, no map lookups on shared state). Caller proceeds to any I/O / SSE publishes AFTER this method returns, not under p.mu.

Map key parsing: keys are `source_path + "|" + variant_id` per the Enqueue contract. Splits on the FIRST `|` via strings.IndexByte (no SplitN allocation per iteration) and passes only the source_path segment to the predicate. The `|` byte is reserved (paths never contain it in practice, and the manifest scanner rejects upserts with malformed paths at the boundary); a defensive missing-pipe entry is left untouched rather than logged — it's never legitimately present.

func (*Pool) Enqueue ¶

func (p *Pool) Enqueue(spec JobSpec) error

Enqueue submits a JobSpec to the worker pool. Non-blocking: if the queue is full, returns ErrQueueFull immediately. Dedup is silent — a duplicate (same source_path + variant_id) returns nil without taking a slot. After Stop, returns ErrPoolClosed.

**Race-safe vs Stop**: pre-fix, Stop() did `close(p.optimizeJobs) / close(p.upscaleJobs)` concurrently with an Enqueue holding `inflight[dedup]` but not yet at the channel send — sending on a closed channel panics (Gemini high + Qodo bug 1 + CodeRabbit echo on PR #109). Fix: the channel-send branch runs INSIDE `p.mu` and Stop() takes the same mutex before close. The mutex serialises the two operations cheaply (the send is non-blocking via the select + default, so the lock window stays in microseconds), and the re-checked `closed` flag inside the lock catches a Stop that won the race for the mutex but hadn't yet closed the channel.

func (*Pool) SetOnJobComplete ¶ added in v0.1.3

func (p *Pool) SetOnJobComplete(fn func(path, variantID string, sampleRate, bitsPerSample int, durationSeconds float64, batchID uuid.UUID, completedAt time.Time))

SetOnJobComplete wires (or rewires) the per-job completion callback (see Pool.onJobComplete docstring for ordering invariants). nil disables notification. Set-once at cmd/bridge wiring time; race- safe vs concurrent calls.

PR 3 (v1.3) extended the signature with `batchID`. Existing consumers in cmd/bridge/main.go pass it through to the Coordinator's job-complete handler; pre-v1.3 callers that don't care about batch attribution can ignore the new parameter.

func (*Pool) SetOnJobFailed ¶ added in v0.1.3

func (p *Pool) SetOnJobFailed(fn func(path, variantID, errMsg string, durationSeconds float64, batchID uuid.UUID, failedAt time.Time))

SetOnJobFailed wires the per-job failure callback. nil disables. See Pool.onJobFailed docstring for threading invariants. Set-once at cmd/bridge wiring time alongside SetOnJobComplete.

func (*Pool) SetOnStateChange ¶

func (p *Pool) SetOnStateChange(fn func())

SetOnStateChange wires (or rewires) the callback fired after every observable state transition. nil disables notification (back-compat for tests / non-broker deployments). Called once during cmd/bridge wiring after the broker is up; subsequent calls are race-safe but in practice this is set-once.

func (*Pool) Stats ¶

func (p *Pool) Stats() PoolStats

Stats returns the current snapshot. Safe to call concurrently with Enqueue / worker activity.

**QueueLen AND QueueCap are both COMBINED across the two priority channels** so the ratio `QueueLen / QueueCap` stays bounded by 1.0 for monitoring tools that compute a fill-percentage. Pre-fix the snapshot reported a per-channel QueueCap alongside a combined QueueLen, which produced ratios >100% whenever both channels were partially full — caught by Gemini medium on PR #281.

**Per-channel back-pressure is still enforced** at Enqueue time: each channel independently has `p.queueCap` slots, and each independently triggers `ErrQueueFull` when full. The doubling here is presentational — it gives consumers a single combined capacity number to divide against the combined depth.

If a future surface needs per-channel split (e.g. an admin "optimize queue depth" indicator), extend PoolStats with sibling fields rather than retargeting either of the existing combined values — back-compat consumers depend on QueueLen + QueueCap staying ratio-coherent.

func (*Pool) Stop ¶

func (p *Pool) Stop()

Stop signals the workers to drain the queue and exit. Blocks until every in-flight conversion completes (either through success, failure, or sox-process kill on the cancelled context) AND the publisher has drained its remaining buffered events. The Pool can't be reused after Stop.

Idempotent: calling Stop twice is safe (the underlying CancelFunc is itself idempotent).

**Shutdown ordering is load-bearing** — the sequence below prevents BOTH "send on closed channel" panics AND publisher deadlocks:

1. close(p.optimizeJobs) + close(p.upscaleJobs) under p.mu so any in-flight Enqueue completes its channel send (or its dedup- rollback) before either close() lands. See Enqueue's docstring for the full race trace. Both channels share the same p.mu so ordering between them inside the lock is irrelevant.
2. stopCancel() so any sox subprocess waiting on the per-job context dies promptly.
3. p.wg.Wait() — block until every worker goroutine has returned. After this point, no worker can send to stateChangeChan or jobCompleteChan.
4. close() both publisher channels — safe ONLY because step 3 guaranteed no more sends are possible.
5. p.publisherWG.Wait() — let the publisher drain its remaining buffered events (especially `upscale.complete` events the iOS path-promotion path depends on), observe both channels closed, and return.

The publisher does NOT exit on stopCtx.Done() — workers blocking- send to jobCompleteChan, and an early publisher exit would deadlock the worker on a buffer-full send.