feat: add `on_error` to Image/Audio/Video features and IterableDatase…

…t.decode()

Adds a new `on_error` parameter that controls how decoding failures are handled. This addresses the long-standing problem where a single corrupted sample in a streaming dataset (bad EXIF, truncated bytes, unidentified format, ...) raises an exception inside the generator frame and silently kills the entire iterator
- making large web-scale pipelines non-resumable.

Refs: #8165 (issue), #8173 (narrower EXIF-only patch).

API (three modes, applied uniformly across Image, Audio, Video):
  - "raise"                 - re-raise the exception (default; preserves current behavior)
  - "return_none"           - silently yield None for the failing example
  - "warn_and_return_none"  - emit a per-sample UserWarning and yield None

The parameter is exposed at two levels:
  1. As a dataclass field on each feature, e.g. `Image(on_error="return_none")`. Validated in `__post_init__`. Wraps the entire `decode_example` body in `except Exception as e:` so all failure modes (OSError, ZeroDivisionError, UnidentifiedImageError, UnicodeDecodeError, ...) are covered uniformly.
  2. As a keyword on `IterableDataset.decode(on_error=...)`. When set, it is propagated to every Image/Audio/Video feature in the schema via the existing `_visit` walker, in both the threaded and single-threaded branches.

Backward compatibility: unchanged. Default is `"raise"` everywhere; existing code that relies on the exception propagating continues to work identically.

Tests (24 new):
  - Image / Audio / Video: invalid value rejection, all 3 modes on a happy decode, all 3 modes on synthetic-bad bytes (b"not an image" / similar).
  - IterableDataset: `decode(on_error=...)` propagates to nested features.
  - Real-world Image scenarios:
    * Reproduces #8165's `ZeroDivisionError` from `PIL.ImageOps.exif_transpose` and verifies all 3 modes handle it correctly.
    * Streaming IterableDataset with a corrupted-EXIF sample mid-batch must keep iterating after `.decode(on_error=...)` instead of dropping the rest.
    * Truncated real JPEG bytes (mimics partial download / disk corruption).
    * Mixed good/bad/good batch end-to-end through `Dataset` (non-streaming).
    * Mixed good/bad/good batch end-to-end through `IterableDataset` (streaming).


Hi @lhoestq — this PR implements your design from issue #8165. Quick alignment check:

- **Q1 (Exception scope)**: catch-all `except Exception` in all three feature decoders (Image/Audio/Video) ✅
- **Q2 (Default)**: `on_error="raise"` preserves current behavior ✅
- **Q3 (Naming)**: exposed via `IterableDataset.decode(on_error=...)` (also settable per-feature on the dataclass) ✅
- **Q4 (Visibility)**: `return_none` is silent; `warn_and_return_none` emits one warning per failed sample ✅

Tests include the original #8165 reproducer (PIL `exif_transpose` `ZeroDivisionError`) plus truncated-JPEG end-to-end through both `Dataset` and `IterableDataset`. Happy to adjust naming or scope if you'd prefer something different.

## End-to-end validation of `on_error` on Flux training (torchtitan)

Validated this PR against torchtitan's Flux trainer (`pixparse/cc12m-wds` streaming + `IterableDataset.decode(on_error="warn_and_return_none")`) with 5 complementary tests. All passed.

A `custom_config_module` swaps the local try/except wrapper out of `FluxDataset` and replaces it with `FluxDatasetNoWrapper`, which calls `self._data.decode(on_error="warn_and_return_none")` at init — so all decode failure handling is delegated to this PR. `PYTHONWARNINGS=always` is set to defeat the `(file, lineno)` dedup.

### Test 1 — API contract (`scripts/test_pr8174_minimal_repro.py`)

| # | Check | Expected | Actual | Match |
|---|---|---|---|---|
| 1 | `Image()` default on a corrupt sample | raises `UnidentifiedImageError` (back-compat preserved) | raised `UnidentifiedImageError` | ✅ |
| 2 | `Image(on_error="return_none")` on corrupt sample | returns `None`, no warning | returned `None`, no warning | ✅ |
| 3 | `Image(on_error="warn_and_return_none")` on corrupt sample | returns `None` + emits "Failed to decode image" warning | returned `None` + 1 warning containing "Failed to decode image" | ✅ |
| 4 | `IterableDataset.decode(on_error="warn_and_return_none")` on 3-row stream (good/bad/good) | 3 rows iterated, bad row's `image` is `None`, sibling `txt` preserved, exactly 1 warning | got 3 rows; bad row `image=None`, `txt` preserved; 1 warning | ✅ |

**Outcome**: 4/4 match. The PR's API behaves exactly as documented at both the feature level (`Image(on_error=...)`) and the dataset level (`IterableDataset.decode(on_error=...)`).

### Test 2 — Production code path, isolated unit (`scripts/test_pr8174_production_path.py`)

5-sample HF `Dataset` (good / good / **BAD** / good / good, cc12m schema) fed through `FluxDatasetNoWrapper`.

| Check | Expected | Actual | Match |
|---|---|---|---|
| HF "Failed to decode image" warnings | 1 (one bad sample) | 1 | ✅ |
| `FluxDatasetNoWrapper` "skipping bad sample" logs | 1 (downstream filter catches `None`) | 1, with `key=k2_BAD` | ✅ |
| Yielded samples | 4 (red, green, blue, yellow), BAD absent, order preserved | 4 in order red→green→blue→yellow, no BAD | ✅ |
| Process exit | clean, no crash | clean | ✅ |

**Outcome**: 4/4 match. Confirms the PR call site (`self._data.decode(on_error="warn_and_return_none")` inside `FluxDatasetNoWrapper.__init__`) wires through end-to-end: bad sample → HF warns + returns None → downstream filter skips → iteration continues.

### Test 3 — Real Flux training, step 0 → 800

Standard `cc12m-wds` stream, 8 ranks FSDP, no injection.

| Check | Expected | Actual | Match |
|---|---|---|---|
| `[PR #8174 validation]` init markers | 8 (one per rank) | 8 | ✅ |
| Crashes (SIGABRT / unhandled exception) | 0 | 0 | ✅ |
| `Failed to decode image` warnings | 0 or more — depends on whether the cc12m window contains PIL-unrecoverable samples; either is acceptable | 0 (window happened to be clean) | ✅ |
| Training reaches step 800 | yes | yes | ✅ |

**Outcome**: matches expectation. PR introduces no regression on real Flux training. The 0 warnings is *not* evidence the PR works — the window simply had no unrecoverable samples; that question is answered by Test 4 + Test 5.

### Test 4 — Continuation training, step 720 → 1500

Resumed from full-state checkpoint at step 720 to check the PR path is re-applied after a checkpoint resume.

| Check | Expected | Actual | Match |
|---|---|---|---|
| `[PR #8174 validation]` markers across the run | ≥8 (init on every rank, again on epoch reloop) | 16 (8 ranks × 2: init + epoch reloop) | ✅ |
| Crashes | 0 | 0 | ✅ |
| `Failed to decode image` (HF on_error) | 0 or more, see Test 3 caveat | 0 | ✅ |
| PIL-recoverable warnings (`Truncated File Read`, `Corrupt EXIF`) | non-zero, NOT routed through HF on_error (PIL self-recovers) | 1890 PIL warnings, 0 HF on_error warnings — correctly disjoint | ✅ |
| Training reaches step 1500 | yes | yes | ✅ |

**Outcome**: matches expectation. Two non-trivial signals:

1. PIL-recoverable warnings (1890) and HF `Failed to decode image` (0) are disjoint — the PR correctly fires only when PIL gives up, not on every PIL warning.
2. `FluxDatasetNoWrapper` deliberately **removes** the local `except Exception` around `next(dataset_iterator)`. If the PR path failed to swallow a decode error, the unhandled exception would crash an FSDP rank → SIGABRT. 1500 steps without a crash is *consistent* with the PR working, but not by itself proof — the cc12m shuffle window may genuinely have contained no unrecoverable samples. Test 5 forces the issue.

### Test 5 — Production smoke with injected bad sample, 8 GPU FSDP

Forces the on_error path under real torchrun + FSDP + T5/CLIP/VAE. `FluxDatasetNoWrapperInjectBad` chains one schema-matched bad sample (`b"INJECTED_BAD_SAMPLE_NOT_A_REAL_IMAGE"`) at the start of `_get_data_iter`, routed through the same `decode(on_error=...)` API the PR introduces.

| Check | Expected | Actual | Match |
|---|---|---|---|
| `[PR #8174 INJECT validation]` init markers | 8 (one per rank) | 8 | ✅ |
| `[PR #8174 INJECT validation]` prepending markers | 8 (each rank prepends 1 bad sample) | 8 | ✅ |
| HF `Failed to decode image` warnings (the PR's user-facing warning) | ≥8 — must fire for every rank's bad sample | 16 (each rank fired twice: DataLoader prefetch + real iter; pairs cleanly with the 8 downstream skips) | ✅ |
| Downstream `skipping bad sample key=INJECTED_BAD` log (post-skip confirmation) | 8 (one per rank, paired with the warning) | 8 | ✅ |
| `step` outputs (training continued past the bad sample) | 40 (5 steps × 8 ranks) | 40, with loss 1.9982 → 2.1704 → 1.7060 → 1.7070 → 1.5591 (all distinct → consuming different samples, not the same one looped) | ✅ |
| **No re-iteration from index 0 after skip** | After skip, iterator continues into `real_iter`; `_get_data_iter` must NOT be called again | All 8 prepending markers occur 07:13:45–07:13:48 (dataloader init); step 1 starts at 07:14:00; **0 prepending / 0 Failed to decode / 0 skipping after step 1 begins** | ✅ |
| Crashes (SIGABRT / Aborted / HIP OOM) | 0 | 0 | ✅ |

**Outcome**: 7/7 match. The full chain fires under real production conditions: HF on_error emits the documented warning → returns None → downstream filter catches None and emits its own skip log → iterator continues forward into the real cc12m stream (no replay from index 0) → training proceeds through 5 distinct-loss steps without any rank crashing. Direct end-to-end evidence the PR works.

### Conclusion

`on_error` works as documented in both isolated unit tests and full-stack production training (FSDP, real T5/CLIP/VAE, real cc12m streaming). Bad sample → warn + None → downstream filter → iteration continues. 1500 real-training steps with the PR active produced zero crashes across 8 ranks; an injected bad sample produced the expected `Failed to decode image` warning on every rank that saw it.