Give every process group one collective timeout that survives rank-0-only work

[luciaquirke]

Give every process group one collective timeout that survives rank-0-only work

EK-FAC runs abort at larger dataset sizes with

Watchdog caught collective operation timeout
checkTimeout at ProcessGroupNCCL.cpp:733

always immediately after "Collecting gradients: 100%", and never during the
gradient loop itself. The loop is not the problem -- allocate_batches pads the
batch count to a multiple of the world size and asserts every rank gets an equal
number, so ranks leave the loop together.

What follows the loop is not symmetric:

run_with_collector_hooks()                    collector/collector.py
    ... gradient loop (symmetric) ...
    self.collector.teardown()                 hessians/autocorrelation.py
        process_autocorrelation_matrices(...)   `if rank == 0:` ...
        if self.rank == 0: processor.save(...)
    dist.all_reduce(total_processed)          <- other ranks block HERE

Rank 0 reduces and eigendecomposes the Grams and writes them to storage while
every other rank is already waiting in the all_reduce. That is not a deadlock --
rank 0 does arrive -- but the rank-0 work grows with the dataset and the write
lands on shared storage, so past the timeout the watchdog kills a run that was
making normal progress. We see it at 64k documents where 32k with the same
config and the same hardware completes.

The timeouts in place were inconsistent and, in the build path, too short:

build.py         30 minutes   <- the index/EK-FAC path, where this fires
distributed.py   1 hour  (init_dist) / none (parent_barrier)
score/score.py   1 hour
magic/cli.py     none -> NCCL default, 10 minutes
magic/metasmoothness.py  none -> 10 minutes

This replaces all five with a single DIST_TIMEOUT in bergson/distributed.py,
defaulting to 1 hour and overridable per-run with BERGSON_DIST_TIMEOUT_MIN.

Worth stating explicitly: TORCH_NCCL_HEARTBEAT_TIMEOUT_SEC does NOT help here.
It governs the watchdog monitor thread, not the collective timeout, and the
collective timeout has no environment override at all -- it is an
init_process_group argument. That is why this has to be a code change.

This raises the ceiling rather than removing the asymmetry. Doing the all_reduce
before teardown, or sharding the Hessian save across ranks, would shrink the
window itself; both are larger changes and neither is needed to stop healthy
runs being killed.

[luciaquirke]

Measurement update: raised the default from 1 h to 3 h.

The original hour was a guess made before I had a number for the rank-0 section. On a 64k-document row I have now observed it still healthy at 114 minutes — rank 0 pinned at 100% GPU running process_autocorrelation_matrices and processor.save(), rank 1 at 0% blocked in dist.all_reduce(total_processed). An hour would abort exactly the runs this PR is meant to protect.

The two-GPU utilisation split is also the cleanest way to tell this state from a real hang: one rank saturated and the rest idle is healthy rank-0 work.

Still worth saying that this raises a ceiling rather than removing the asymmetry — doing the all_reduce before teardown, or sharding the Hessian save across ranks, would shrink the window itself.