diff --git a/.github/markdown-links-config.json b/.github/markdown-links-config.json
index e4d81abb5..376a5636f 100644
--- a/.github/markdown-links-config.json
+++ b/.github/markdown-links-config.json
@@ -8,6 +8,9 @@
     },
     {
       "pattern": "^https://codecov.io/gh/Lightning-AI/litData/graph/badge.svg"
+    },
+    {
+      "pattern": "^https://devblog.pytorchlightning.ai/"
     }
   ],
   "httpHeaders": [
diff --git a/src/litdata/raw/dataset.py b/src/litdata/raw/dataset.py
index cf0a3b7de..86894dbaf 100644
--- a/src/litdata/raw/dataset.py
+++ b/src/litdata/raw/dataset.py
@@ -108,6 +108,182 @@
 # the bandwidth arm alone sizes the budget. Distinct from ``_HEDGE_MAX_BYTES``
 # (8 MiB duplicate-GET hedge policy).
 _LATENCY_MODEL_MAX_MEDIAN_BYTES = 1024 * 1024
+_LATENCY_OBSERVATION_MAX_BYTES = 256 * 1024
+_BANDWIDTH_OBSERVATION_MIN_BYTES = 64 * 1024
+_MIN_EMPIRICAL_SAMPLES = 5
+_LOW_BANDWIDTH_THRESHOLD_BPS = 10 * 1024 * 1024  # 10 MiB/s policy threshold
+_HIGH_LATENCY_THRESHOLD_S = 0.100  # 100 ms policy threshold
+# Minimum estimated request latency (seconds). Prevents the transfer-subtracted
+# latency estimate from collapsing to zero or going negative on fast / cached
+# responses. 1 ms is a reasonable floor: well below any real WAN RTT, but
+# large enough to keep EMA arithmetic well-conditioned.
+_LATENCY_EPSILON_S = 0.001
+_LATENCY_RTT_EPSILON = _LATENCY_EPSILON_S
+
+
+def _env_float(name: str, default: float) -> float:
+    val = os.getenv(name)
+    if val is None:
+        return default
+    try:
+        return float(val)
+    except ValueError:
+        return default
+
+
+def _env_int(name: str, default: int) -> int:
+    val = os.getenv(name)
+    if val is None:
+        return default
+    try:
+        return int(val)
+    except ValueError:
+        return default
+
+
+def _get_assumed_aggregate_bandwidth_bps() -> int:
+    return _env_int("LITDATA_ASSUMED_BANDWIDTH_BPS", _ASSUMED_AGGREGATE_BANDWIDTH_BPS)
+
+
+def _get_assumed_request_rate() -> float:
+    return _env_float("LITDATA_ASSUMED_REQUEST_RATE", _ASSUMED_REQUEST_RATE)
+
+
+def _get_assumed_request_latency_s() -> float:
+    return _env_float("LITDATA_ASSUMED_REQUEST_LATENCY_S", _ASSUMED_REQUEST_LATENCY_S)
+
+
+def _get_default_median_file_bytes() -> int:
+    return _env_int("LITDATA_DEFAULT_MEDIAN_FILE_BYTES", _DEFAULT_MEDIAN_FILE_BYTES)
+
+
+def _get_single_process_concurrency_cap() -> int:
+    return _env_int("LITDATA_SINGLE_PROCESS_CONCURRENCY_CAP", _SINGLE_PROCESS_CONCURRENCY_CAP)
+
+
+def _get_aggregate_concurrency_budget_cap() -> int:
+    return _env_int("LITDATA_AGGREGATE_CONCURRENCY_BUDGET_CAP", _AGGREGATE_CONCURRENCY_BUDGET_CAP)
+
+
+def _get_aggregate_concurrency_budget_floor() -> int:
+    return _env_int("LITDATA_AGGREGATE_CONCURRENCY_BUDGET_FLOOR", _AGGREGATE_CONCURRENCY_BUDGET_FLOOR)
+
+
+class BandwidthTracker:
+    """Thread-safe process-local empirical bandwidth and request latency tracker using EMA."""
+
+    def __init__(self, alpha: float = 0.2) -> None:
+        self.alpha = max(0.01, min(1.0, alpha))
+        self.bandwidth_bps_ema: float | None = None
+        self.request_latency_s_ema: float | None = None
+        self.bps_sample_count: int = 0
+        self.lat_sample_count: int = 0
+        self._sample_count: int = 0
+        self._lock = threading.Lock()
+
+    @property
+    def sample_count(self) -> int:
+        """Total GET requests observed by the tracker."""
+        with self._lock:
+            return self._sample_count
+
+    def record_observation(self, size_bytes: int, duration_s: float) -> None:
+        """Record an empirical GET observation and update EMA estimates.
+
+        Evaluation order is intentional and matters for correctness:
+
+        1. Read the **current** (pre-update) bandwidth EMA.
+        2. Estimate transfer-subtracted request latency using that prior estimate
+           (if empirical bandwidth sample threshold is met) and update ``request_latency_s_ema``.
+        3. Update ``bandwidth_bps_ema`` with the current observation.
+
+        This ordering avoids a circular estimation problem: if the bandwidth EMA
+        were updated first, the freshly-observed ``size / duration`` would be used
+        to explain its own transfer time, which collapses the latency estimate
+        towards ``_LATENCY_EPSILON_S`` on every sample — defeating the purpose
+        of the subtraction entirely.
+
+        The latency stored in ``request_latency_s_ema`` is *not* TCP RTT; it is
+        the estimated request latency after removing the payload-transfer
+        component from wall-clock GET time (connection setup, TLS, server
+        processing, scheduling, and proxy overhead are all included).
+        """
+        if size_bytes <= 0 or duration_s <= 0:
+            return
+        with self._lock:
+            self._sample_count += 1
+
+            # Step 1: capture the PREVIOUS bandwidth estimate before modifying it.
+            # This is the key invariant: the current sample must not be used to
+            # explain its own transfer time.
+            prev_bps_ema = self.bandwidth_bps_ema
+
+            # Step 2: update transfer-subtracted latency EMA for small objects.
+            if size_bytes < _LATENCY_OBSERVATION_MAX_BYTES:
+                self.lat_sample_count += 1
+                # Fall back to assumed aggregate bandwidth when empirical BPS sample
+                # threshold (_MIN_EMPIRICAL_SAMPLES = 5) has not yet been met.
+                effective_bps = (
+                    prev_bps_ema
+                    if self.bps_sample_count >= _MIN_EMPIRICAL_SAMPLES and prev_bps_ema is not None and prev_bps_ema > 0
+                    else _get_assumed_aggregate_bandwidth_bps()
+                )
+                transfer_time_s = float(size_bytes) / effective_bps
+                # Estimated request latency: wall-clock minus payload-transfer time.
+                # Clamped to _LATENCY_EPSILON_S to stay well-conditioned.
+                estimated_request_latency_s = max(_LATENCY_EPSILON_S, duration_s - transfer_time_s)
+                if self.request_latency_s_ema is None:
+                    self.request_latency_s_ema = estimated_request_latency_s
+                else:
+                    self.request_latency_s_ema = (
+                        self.alpha * estimated_request_latency_s + (1.0 - self.alpha) * self.request_latency_s_ema
+                    )
+
+            # Step 3: now update bandwidth EMA with the current observation.
+            if size_bytes >= _BANDWIDTH_OBSERVATION_MIN_BYTES:
+                self.bps_sample_count += 1
+                obs_bps = float(size_bytes) / duration_s
+                if self.bandwidth_bps_ema is None:
+                    self.bandwidth_bps_ema = obs_bps
+                else:
+                    self.bandwidth_bps_ema = self.alpha * obs_bps + (1.0 - self.alpha) * self.bandwidth_bps_ema
+
+    def get_metrics(self) -> tuple[float | None, float | None, int, int]:
+        """Returns (bandwidth_bps_ema, request_latency_s_ema, bps_sample_count, lat_sample_count)."""
+        with self._lock:
+            return (
+                self.bandwidth_bps_ema,
+                self.request_latency_s_ema,
+                self.bps_sample_count,
+                self.lat_sample_count,
+            )
+
+    def __getstate__(self) -> dict[str, Any]:
+        with self._lock:
+            return {
+                "alpha": self.alpha,
+                "bandwidth_bps_ema": self.bandwidth_bps_ema,
+                "request_latency_s_ema": self.request_latency_s_ema,
+                "bps_sample_count": self.bps_sample_count,
+                "lat_sample_count": self.lat_sample_count,
+                "sample_count": self._sample_count,
+            }
+
+    def __setstate__(self, state: dict[str, Any]) -> None:
+        self.alpha = state.get("alpha", 0.2)
+        self.bandwidth_bps_ema = state.get("bandwidth_bps_ema")
+        self.request_latency_s_ema = state.get("request_latency_s_ema")
+        self.bps_sample_count = state.get("bps_sample_count", 0)
+        self.lat_sample_count = state.get("lat_sample_count", 0)
+        self._sample_count = state.get("sample_count", self.bps_sample_count + self.lat_sample_count)
+        if "bps_sample_count" not in state and "sample_count" in state:
+            legacy_count = state.get("sample_count", 0)
+            if self.bandwidth_bps_ema is not None:
+                self.bps_sample_count = legacy_count
+            if self.request_latency_s_ema is not None:
+                self.lat_sample_count = legacy_count
+        self._lock = threading.Lock()
+
 
 _RUNNER_LOCK = threading.Lock()
 _RUNNER: _LoopRunner | None = None
@@ -191,7 +367,7 @@ class _LoopRunner:
     def __init__(self) -> None:
         self._pid = os.getpid()
         self.loop: asyncio.AbstractEventLoop = _create_event_loop()
-        self._executor = ThreadPoolExecutor(max_workers=32, thread_name_prefix="litdata-raw-pool")
+        self._executor = ThreadPoolExecutor(max_workers=32, thread_name_prefix="asyncio_litdata-raw-pool")
         self.loop.set_default_executor(self._executor)
         if _RAW_DEBUG:
             logger.warning(
@@ -413,52 +589,114 @@ def _median_file_bytes(files: Sequence[FileMetadata]) -> int | None:
     return int(statistics.median(sizes))
 
 
-def _aggregate_concurrency_budget(median_file_bytes: int | None) -> int:
+def _aggregate_concurrency_budget(
+    median_file_bytes: int | None,
+    tracker: BandwidthTracker | None = None,
+) -> int:
     """Aggregate in-flight download slots across all workers (size-aware, clamped).
 
-    Takes the max of two models then clamps to ``[floor, cap]``:
-
-    - **bandwidth**: ``(aggregate_bps × pipeline_s) // median_file_bytes`` — keep
-      ~50 MiB moving for large objects.
-    - **latency / Little's law**: ``target_rate × assumed_latency`` (~6000×0.040 ≈
-      240) **only when** ``median < _LATENCY_MODEL_MAX_MEDIAN_BYTES`` (1 MiB) so
-      tiny-object paths are not request-starved. Medians ≥1 MiB stay
-      bandwidth-bounded (avoids pinning at 240 slots → multi-GB in flight).
+    Calculates unconstrained baseline capacity: max(bandwidth model, Little's-law model).
+    `max` is used intentionally so the baseline is not constrained by single-model
+    underestimation.
 
-    Per-worker floor of 8 means realized aggregate is ``max(budget, 8 × num_workers)``.
+    If empirical measurements indicate congestion (latency > target), applies
+    stateless backoff factor (L_target / L_obs).
     """
-    median = median_file_bytes if median_file_bytes and median_file_bytes > 0 else _DEFAULT_MEDIAN_FILE_BYTES
-    target_bytes = int(_ASSUMED_AGGREGATE_BANDWIDTH_BPS * _CONCURRENCY_PIPELINE_SECONDS)
+    default_median = _get_default_median_file_bytes()
+    median = median_file_bytes if median_file_bytes and median_file_bytes > 0 else default_median
+
+    floor = _get_aggregate_concurrency_budget_floor()
+    cap = _get_aggregate_concurrency_budget_cap()
+
+    obs_bps: float | None = None
+    obs_lat: float | None = None
+    bps_samples: int = 0
+    lat_samples: int = 0
+    if tracker is not None:
+        bps_ema, lat_ema, bps_samples, lat_samples = tracker.get_metrics()
+        if bps_samples >= _MIN_EMPIRICAL_SAMPLES:
+            obs_bps = bps_ema
+        if lat_samples >= _MIN_EMPIRICAL_SAMPLES:
+            obs_lat = lat_ema
+
+    bandwidth_bps = obs_bps if obs_bps is not None and obs_bps > 0 else _get_assumed_aggregate_bandwidth_bps()
+    target_bytes = int(bandwidth_bps * _CONCURRENCY_PIPELINE_SECONDS)
     bandwidth_model = max(1, target_bytes // median)
+
     # Size-gate: Little's-law arm is for request-overhead-bound tiny objects only.
+    # Fixed baseline: target_rate * target_latency. Observed latency is NEVER multiplied in.
+    target_lat = _get_assumed_request_latency_s()
     if median < _LATENCY_MODEL_MAX_MEDIAN_BYTES:
-        latency_model = max(1, int(_ASSUMED_REQUEST_RATE * _ASSUMED_REQUEST_LATENCY_S))
+        req_rate = _get_assumed_request_rate()
+        latency_model = max(1, int(req_rate * target_lat))
     else:
         latency_model = 0
-    raw = max(bandwidth_model, latency_model)
-    return max(_AGGREGATE_CONCURRENCY_BUDGET_FLOOR, min(_AGGREGATE_CONCURRENCY_BUDGET_CAP, raw))
+
+    # max is intentional to avoid underestimating baseline capacity
+    baseline_budget = max(bandwidth_model, latency_model)
+
+    # Stateless congestion control backoff: latency > target => reduce budget
+    if obs_lat is not None and obs_lat > target_lat:
+        backoff_factor = min(1.0, target_lat / obs_lat)
+        computed_budget = max(1, int(baseline_budget * backoff_factor))
+    else:
+        computed_budget = baseline_budget
+
+    # Guarded adaptive floor reduction: require high-confidence combined evidence
+    if (
+        bps_samples >= _MIN_EMPIRICAL_SAMPLES
+        and lat_samples >= _MIN_EMPIRICAL_SAMPLES
+        and obs_bps is not None
+        and obs_bps < _LOW_BANDWIDTH_THRESHOLD_BPS
+        and obs_lat is not None
+        and obs_lat > _HIGH_LATENCY_THRESHOLD_S
+    ):
+        effective_floor = 1
+    else:
+        effective_floor = floor
+
+    # Enforce authoritative MAX cap (512) and effective MIN floor
+    return max(effective_floor, min(cap, computed_budget))
 
 
 def _effective_concurrency(
     max_concurrent_downloads: int | None,
     num_workers: int,
     median_file_bytes: int | None = None,
+    tracker: BandwidthTracker | None = None,
+    worker_id: int | None = None,
 ) -> int:
-    """Per-worker download permits for the Stage 1 static clamp.
+    """Per-worker download permits for the Stage 1 static/adaptive clamp.
 
     - ``max_concurrent_downloads is None`` (default): adaptive —
-      ``max(floor, budget // num_workers)`` with ``budget`` from
-      :func:`_aggregate_concurrency_budget`. When ``num_workers <= 1``, returns
-      ``min(budget, _SINGLE_PROCESS_CONCURRENCY_CAP)`` (unbenchmarked path).
+      budget split across workers while strictly maintaining aggregate budget
+      invariant: sum(worker_permits) <= aggregate_budget.
     - Explicit ``int``: **exactly** that many permits (no silent clamp). ``<= 0``
       collapses to 1.
     """
     if max_concurrent_downloads is not None:
         return 1 if max_concurrent_downloads <= 0 else max_concurrent_downloads
-    budget = _aggregate_concurrency_budget(median_file_bytes)
+    budget = _aggregate_concurrency_budget(median_file_bytes, tracker=tracker)
     if num_workers <= 1:
-        return min(budget, _SINGLE_PROCESS_CONCURRENCY_CAP)
-    return max(_MIN_CONCURRENCY_PER_WORKER, budget // num_workers)
+        return min(budget, _get_single_process_concurrency_cap())
+
+    if budget >= num_workers:
+        return budget // num_workers
+
+    # When aggregate budget < num_workers, allocate 1 permit to ranks < budget
+    w_id = worker_id
+    if w_id is None:
+        try:
+            from torch.utils.data import get_worker_info
+
+            info = get_worker_info()
+            if info is not None:
+                w_id = info.id
+        except ImportError:
+            pass
+    if w_id is not None and w_id >= budget:
+        return 0
+    return 1
 
 
 def _num_dataloader_workers() -> int:
@@ -557,6 +795,7 @@ def __init__(
         self.storage_options = storage_options or {}
         # Index median size (bytes); set by StreamingRawDataset after discovery.
         self._median_file_bytes: int | None = None
+        self._bandwidth_tracker = BandwidthTracker()
         self._downloader: Downloader | None = None
         self._downloader_pid: int | None = None
         self._downloader_loop: asyncio.AbstractEventLoop | None = None
@@ -610,6 +849,7 @@ def __getstate__(self) -> dict[str, Any]:
             "cache_dir": self.cache_dir,
             "storage_options": self.storage_options,
             "_median_file_bytes": self._median_file_bytes,
+            "_bandwidth_tracker": self._bandwidth_tracker,
             # Runtime — always fresh in the child.
             "_downloader": None,
             "_downloader_pid": None,
@@ -645,6 +885,7 @@ def __setstate__(self, state: dict[str, Any]) -> None:
         self._range_executor_pid = None
         self._hedge_fired = 0
         self._median_file_bytes = state.get("_median_file_bytes")
+        self._bandwidth_tracker = state.get("_bandwidth_tracker") or BandwidthTracker()
 
     def _shutdown_range_executor(self) -> None:
         if self._range_executor is not None:
@@ -720,6 +961,14 @@ def downloader(self) -> Downloader:
             self._downloader_loop = loop
         return self._downloader
 
+    def _record_download_observation(self, size_bytes: int, duration_s: float) -> None:
+        """Record an empirical GET transfer observation and refresh cached permits when needed."""
+        prev_count = self._bandwidth_tracker.sample_count
+        self._bandwidth_tracker.record_observation(size_bytes, duration_s)
+        new_count = self._bandwidth_tracker.sample_count
+        if (prev_count < 5 and new_count >= 5) or (new_count >= 5 and new_count % 10 == 0):
+            self._cached_permits = None
+
     def _effective_download_permits(self) -> int:
         """Worker-aware permit count for the download semaphore (Stage 1 static clamp).
 
@@ -733,6 +982,7 @@ def _effective_download_permits(self) -> int:
             self.max_concurrent_downloads,
             _num_dataloader_workers(),
             self._median_file_bytes,
+            tracker=self._bandwidth_tracker,
         )
         self._cached_permits = permits
         self._cached_permits_pid = pid
@@ -750,7 +1000,7 @@ def _get_semaphore(self) -> asyncio.Semaphore:
         if self._semaphore is None or self._semaphore_loop is not loop or self._semaphore_permits != permits:
             n_workers = _num_dataloader_workers()
             budget = (
-                _aggregate_concurrency_budget(self._median_file_bytes)
+                _aggregate_concurrency_budget(self._median_file_bytes, tracker=self._bandwidth_tracker)
                 if self.max_concurrent_downloads is None
                 else None
             )
@@ -1021,11 +1271,15 @@ async def fetch() -> bytes:
                 data = await fetch()
             return offset, data
 
+        t0 = time.monotonic()
         parts = await asyncio.gather(*(one(o, n) for o, n in ranges))
+        dur = time.monotonic() - t0
         parts.sort(key=lambda x: x[0])
         joined = b"".join(data for _, data in parts)
         if len(joined) != size:
             raise RuntimeError(f"Ranged download size mismatch for {file_path}: expected={size} got={len(joined)}")
+        if len(joined) > 0:
+            self._record_download_observation(len(joined), dur)
         return joined
 
     async def _fetch_bytes(self, file_path: str, size: int | None = None, *, gated: bool = True) -> bytes:
@@ -1048,14 +1302,24 @@ async def _fetch_bytes(self, file_path: str, size: int | None = None, *, gated:
         delay = _effective_hedge_delay(self.hedge_delay, size) if self._is_remote_object(file_path) else None
         # Pay-per-use: hedging off/ineligible → bare permit + download (batch enforces timeout).
         if delay is None:
+            t0 = time.monotonic()
             async with self._permit(gated):
-                return await self.downloader.adownload_fileobj(file_path)
+                data = await self.downloader.adownload_fileobj(file_path)
+            dur = time.monotonic() - t0
+            if len(data) > 0:
+                self._record_download_observation(len(data), dur)
+            return data
 
         async def once() -> bytes:
             async with self._permit(gated):
                 return await self.downloader.adownload_fileobj(file_path)
 
-        return await self._hedged(once, delay)
+        t0_h = time.monotonic()
+        data = await self._hedged(once, delay)
+        dur_h = time.monotonic() - t0_h
+        if len(data) > 0:
+            self._record_download_observation(len(data), dur_h)
+        return data
 
     def _schedule_write_behind(self, local_path: str, data: bytes) -> None:
         """Atomically publish ``data`` to ``local_path`` on a worker thread."""
@@ -1098,9 +1362,16 @@ async def _download_owned(self, file_path: str, local_path: str, size: int | Non
         try:
             if self._path_is_cached(local_path):
                 return local_path
+            t0 = time.monotonic()
             try:
                 # Hang protection is batch-level; keep the owned path bare.
                 await self.downloader.adownload_file(file_path, tmp_path)
+                dur = time.monotonic() - t0
+                st_size = (
+                    size if size and size > 0 else (os.path.getsize(tmp_path) if os.path.exists(tmp_path) else None)
+                )
+                if st_size and st_size > 0:
+                    self._record_download_observation(st_size, dur)
             except Exception as first_exc:
                 if self._is_non_retryable_download_error(first_exc):
                     raise
@@ -1112,6 +1383,7 @@ async def _download_owned(self, file_path: str, local_path: str, size: int | Non
                 with contextlib.suppress(OSError):
                     os.remove(tmp_path)
                 # Caller already holds the download semaphore — avoid nested acquire.
+                # Note: _fetch_bytes records download observation internally if successful.
                 data = await self._fetch_bytes(file_path, size=size, gated=False)
                 await asyncio.to_thread(Path(tmp_path).write_bytes, data)
             self._verify_tmp_size(tmp_path, size)
diff --git a/tests/raw/test_dataset.py b/tests/raw/test_dataset.py
index bead4c069..e6ef11738 100644
--- a/tests/raw/test_dataset.py
+++ b/tests/raw/test_dataset.py
@@ -91,7 +91,7 @@ def test_effective_prefetch_vs_num_workers(num_workers, max_prefetch, expected):
         (32, None, 100_000, 16),  # 512//32
         # Large objects (≥1 MiB): bandwidth-only (no Little's-law pin at 240)
         (4, None, 10 * 1024 * 1024, 8),  # budget=floor 32, 32//4=8
-        (16, None, 10 * 1024 * 1024, 8),  # 32//16=2 → floor 8
+        (16, None, 10 * 1024 * 1024, 2),  # budget=floor 32, 32//16=2
         # Unknown size uses default median (256KiB) → latency arm (240)
         (8, None, None, 30),  # 240//8
     ],
@@ -539,3 +539,355 @@ def transform(data):
     gds = GroupedDS(input_dir=str(tmp_path), transform=transform)
     gds.cache_manager.download_file_async = mock_download_file_async
     assert gds[0] == b"abc"
+
+
+def test_bandwidth_tracker_basic():
+    import pytest
+
+    from litdata.raw.dataset import (
+        _ASSUMED_AGGREGATE_BANDWIDTH_BPS,
+        _LATENCY_RTT_EPSILON,
+        BandwidthTracker,
+    )
+
+    assumed_bps = float(_ASSUMED_AGGREGATE_BANDWIDTH_BPS)
+    tracker = BandwidthTracker(alpha=0.2)
+    assert tracker.sample_count == 0
+
+    # --- Tiny GET (50 KB < 64 KB) ---
+    # Only updates latency EMA; no bandwidth observation recorded.
+    # prev_bps_ema = None → fallback to assumed bandwidth for transfer estimate.
+    tracker.record_observation(50_000, 0.010)
+    bps, lat, bps_count, lat_count = tracker.get_metrics()
+    assert tracker.sample_count == 1
+    assert bps_count == 0
+    assert lat_count == 1
+    assert bps is None
+    transfer1 = 50_000 / assumed_bps
+    expected_lat1 = max(_LATENCY_RTT_EPSILON, 0.010 - transfer1)
+    assert pytest.approx(lat, abs=1e-6) == expected_lat1
+    # Sanity: transfer-subtracted latency must be strictly less than raw duration.
+    assert lat < 0.010
+
+    # --- Small/Medium GET (100 KB: 64 KB <= size < 256 KB) ---
+    # Updates BOTH latency and bandwidth EMAs.
+    # prev_bps_ema is still None (previous obs was below bandwidth threshold) → fallback.
+    tracker.record_observation(100_000, 0.020)
+    bps, lat, bps_count, lat_count = tracker.get_metrics()
+    assert tracker.sample_count == 2
+    assert bps_count == 1
+    assert lat_count == 2
+    assert bps is not None
+    assert pytest.approx(bps, abs=1.0) == 5_000_000.0
+    transfer2 = 100_000 / assumed_bps  # prev_bps_ema still None before this obs
+    est_lat2 = max(_LATENCY_RTT_EPSILON, 0.020 - transfer2)
+    expected_lat2 = 0.2 * est_lat2 + 0.8 * expected_lat1
+    assert pytest.approx(lat, abs=1e-6) == expected_lat2
+
+    # --- Large GET (10 MiB >= 256 KB) ---
+    # Updates bandwidth EMA only; size >= _LATENCY_OBSERVATION_MAX_BYTES.
+    tracker.record_observation(10 * 1024 * 1024, 0.118)
+    bps, lat, bps_count, lat_count = tracker.get_metrics()
+    assert tracker.sample_count == 3
+    assert bps_count == 2
+    assert lat_count == 2  # unchanged — large GETs do not update latency EMA
+    assert bps is not None
+
+
+def test_bandwidth_tracker_pickle():
+    import pickle
+
+    from litdata.raw.dataset import BandwidthTracker
+
+    tracker = BandwidthTracker()
+    tracker.record_observation(100_000, 0.020)
+    tracker.record_observation(10 * 1024 * 1024, 0.200)
+    blob = pickle.dumps(tracker)
+    restored = pickle.loads(blob)  # noqa: S301
+    assert restored.sample_count == tracker.sample_count
+    assert restored.bps_sample_count == tracker.bps_sample_count
+    assert restored.lat_sample_count == tracker.lat_sample_count
+    assert restored.bandwidth_bps_ema == tracker.bandwidth_bps_ema
+    assert restored.request_latency_s_ema == tracker.request_latency_s_ema
+
+
+def test_concurrency_budget_warmup_gating():
+    from litdata.raw.dataset import BandwidthTracker, _aggregate_concurrency_budget
+
+    tracker = BandwidthTracker()
+    # 4 observations (under threshold of 5) -> uses default static budget
+    for _ in range(4):
+        tracker.record_observation(10 * 1024 * 1024, 0.001)
+
+    # Median 10MB -> default budget: (100MB/s * 0.5s) // 10MB = 5 -> floor 32
+    assert _aggregate_concurrency_budget(10 * 1024 * 1024, tracker=tracker) == 32
+
+    # 5th observation -> warm-up gate unlocks empirical EMA
+    tracker.record_observation(10 * 1024 * 1024, 0.001)
+    # Measured bandwidth is huge -> dynamic budget scales up from 32 to 500
+    assert _aggregate_concurrency_budget(10 * 1024 * 1024, tracker=tracker) == 500
+
+
+def test_concurrency_budget_high_and_low_bandwidth_adaptation():
+    from litdata.raw.dataset import BandwidthTracker, _aggregate_concurrency_budget
+
+    # High bandwidth scenario
+    high_tracker = BandwidthTracker()
+    for _ in range(5):
+        high_tracker.record_observation(10 * 1024 * 1024, 0.002)
+    budget_high = _aggregate_concurrency_budget(1 * 1024 * 1024, tracker=high_tracker)
+    assert budget_high == 512
+
+    # Low bandwidth scenario
+    low_tracker = BandwidthTracker()
+    for _ in range(5):
+        low_tracker.record_observation(10 * 1024 * 1024, 5.0)
+    budget_low = _aggregate_concurrency_budget(10 * 1024 * 1024, tracker=low_tracker)
+    assert budget_low == 32
+
+
+def test_class_gated_observation_isolation():
+    from litdata.raw.dataset import (
+        BandwidthTracker,
+        _aggregate_concurrency_budget,
+    )
+
+    # Scenario 1: 5 small GETs (< 64 KB) -> lat_sample_count = 5, bps_sample_count = 0
+    tracker_small = BandwidthTracker()
+    for _ in range(5):
+        tracker_small.record_observation(10_000, 0.010)
+
+    bps, lat, bps_cnt, lat_cnt = tracker_small.get_metrics()
+    assert bps_cnt == 0
+    assert lat_cnt == 5
+    assert bps is None
+
+    # Budget for large 10 MB objects must STILL use static default aggregate bandwidth because bps_sample_count < 5
+    # Clamped to floor 32
+    assert _aggregate_concurrency_budget(10 * 1024 * 1024, tracker=tracker_small) == 32
+
+    # Scenario 2: 5 large GETs (>= 256 KB) -> bps_sample_count = 5, lat_sample_count = 0
+    tracker_large = BandwidthTracker()
+    for _ in range(5):
+        tracker_large.record_observation(10 * 1024 * 1024, 0.001)
+
+    bps, lat, bps_cnt, lat_cnt = tracker_large.get_metrics()
+    assert bps_cnt == 5
+    assert lat_cnt == 0
+    assert lat is None
+
+
+def test_environment_variable_overrides(monkeypatch):
+    from litdata.raw.dataset import (
+        _aggregate_concurrency_budget,
+        _effective_concurrency,
+    )
+
+    monkeypatch.setenv("LITDATA_ASSUMED_BANDWIDTH_BPS", str(500 * 1024 * 1024))
+    monkeypatch.setenv("LITDATA_AGGREGATE_CONCURRENCY_BUDGET_CAP", "1024")
+    monkeypatch.setenv("LITDATA_AGGREGATE_CONCURRENCY_BUDGET_FLOOR", "16")
+    monkeypatch.setenv("LITDATA_SINGLE_PROCESS_CONCURRENCY_CAP", "256")
+
+    assert _aggregate_concurrency_budget(100_000) == 1024
+    assert _effective_concurrency(None, num_workers=1, median_file_bytes=100_000) == 256
+
+
+def test_no_latency_concurrency_inflation():
+    from litdata.raw.dataset import BandwidthTracker, _aggregate_concurrency_budget
+
+    tracker = BandwidthTracker()
+    # Record 5 latency observations with high latency (200ms vs assumed 40ms)
+    for _ in range(5):
+        tracker.record_observation(100_000, 0.200)
+
+    # Budget with high latency must not inflate above baseline (240 for sub-1MB)
+    budget = _aggregate_concurrency_budget(100_000, tracker=tracker)
+    assert budget <= 240
+
+
+def test_concurrency_latency_monotonicity():
+    from litdata.raw.dataset import BandwidthTracker, _aggregate_concurrency_budget
+
+    latencies = [0.040, 0.080, 0.200, 0.500]
+    budgets = []
+
+    for lat in latencies:
+        tracker = BandwidthTracker()
+        for _ in range(5):
+            tracker.record_observation(100_000, lat)
+        budgets.append(_aggregate_concurrency_budget(100_000, tracker=tracker))
+
+    # For latencies above target (40ms), increasing latency must not increase computed budget
+    for i in range(len(budgets) - 1):
+        assert budgets[i + 1] <= budgets[i]
+
+
+def test_concurrency_latency_recovery():
+    from litdata.raw.dataset import BandwidthTracker, _aggregate_concurrency_budget
+
+    tracker = BandwidthTracker()
+    # Inject high latency
+    for _ in range(5):
+        tracker.record_observation(100_000, 0.200)
+    budget_degraded = _aggregate_concurrency_budget(100_000, tracker=tracker)
+
+    # Now inject healthy latency
+    for _ in range(10):
+        tracker.record_observation(100_000, 0.040)
+    budget_recovered = _aggregate_concurrency_budget(100_000, tracker=tracker)
+
+    assert budget_recovered > budget_degraded
+
+
+def test_imagenet_bandwidth_observation():
+    from litdata.raw.dataset import BandwidthTracker
+
+    tracker = BandwidthTracker()
+    # ImageNet JPEG size ~150 KB — straddles both latency (<256 KB) and bandwidth (>=64 KB) thresholds.
+    imagenet_file_size = 150 * 1024
+    for _ in range(5):
+        tracker.record_observation(imagenet_file_size, 0.015)
+
+    bps, lat, bps_cnt, lat_cnt = tracker.get_metrics()
+    assert bps_cnt == 5
+    assert lat_cnt == 5
+
+    # ImageNet files MUST record both EMA estimates.
+    assert bps is not None and bps > 0
+    assert lat is not None and lat > 0
+
+    # Transfer-subtracted latency must be strictly less than raw wall-clock duration.
+    # (transfer time is non-zero for a 150 KB file)
+    assert lat < 0.015
+
+    # Epsilon floor must hold: estimated latency must not go below _LATENCY_RTT_EPSILON.
+    assert lat >= 0.001
+
+
+def test_guarded_adaptive_floor_reduction():
+    from litdata.raw.dataset import BandwidthTracker, _aggregate_concurrency_budget
+
+    tracker = BandwidthTracker()
+    # Low bandwidth (<10 MiB/s) AND high latency (>100 ms)
+    for _ in range(5):
+        tracker.record_observation(128 * 1024, 0.500)
+
+    # Under severe evidence (low bandwidth + high latency), budget can drop below default 32 floor
+    budget = _aggregate_concurrency_budget(10 * 1024 * 1024, tracker=tracker)
+    assert budget < 32
+
+
+def test_worker_allocation_respects_aggregate_budget():
+    from litdata.raw.dataset import BandwidthTracker, _aggregate_concurrency_budget, _effective_concurrency
+
+    test_cases = [
+        (512, 8, 100_000),
+        (32, 8, 10_000_000),
+        (32, 16, 10_000_000),
+        (32, 64, 10_000_000),
+    ]
+
+    for expected_budget_cap, workers, median_bytes in test_cases:
+        tracker = BandwidthTracker()
+        budget = _aggregate_concurrency_budget(median_bytes, tracker=tracker)
+        total_permits = sum(
+            _effective_concurrency(
+                None, num_workers=workers, median_file_bytes=median_bytes, tracker=tracker, worker_id=w
+            )
+            for w in range(workers)
+        )
+        # Sum of per-worker permits across all workers must not exceed aggregate budget
+        assert total_permits <= budget
+
+
+def test_transfer_subtracted_latency():
+    """Pre-seed the tracker with 5 large GETs to reach sample threshold and establish a known bandwidth EMA (20 MB/s),
+    then record a 200 KB request taking 30 ms. Verify latency EMA is approximately 30 ms - transfer_time (10 ms) = 20 ms.
+    """
+    import pytest
+
+    from litdata.raw.dataset import BandwidthTracker
+
+    tracker = BandwidthTracker(alpha=1.0)
+    prior_bps = 20 * 1024 * 1024  # 20 MB/s
+    large_size = 10 * 1024 * 1024  # 10 MiB (bandwidth-only GET)
+
+    # Record 5 observations to pass the _MIN_EMPIRICAL_SAMPLES = 5 threshold
+    for _ in range(5):
+        tracker.record_observation(large_size, large_size / prior_bps)
+
+    bps, lat, bps_cnt, lat_cnt = tracker.get_metrics()
+    assert bps_cnt == 5
+    assert lat_cnt == 0
+    assert pytest.approx(bps, rel=1e-5) == prior_bps
+
+    # Now record 200 KB GET taking 30 ms
+    size = 200 * 1024
+    duration = 0.030
+    tracker.record_observation(size, duration)
+
+    _, lat_after, _, _ = tracker.get_metrics()
+    expected_lat = duration - (size / prior_bps)  # 30 ms - 10 ms = 20 ms
+    assert lat_after is not None
+    assert pytest.approx(lat_after, abs=1e-4) == expected_lat
+
+
+def test_transfer_subtracted_latency_bootstrap():
+    """When no empirical bandwidth exists or bps_sample_count < 5, verify the configured/default bandwidth is used."""
+    import pytest
+
+    from litdata.raw.dataset import _ASSUMED_AGGREGATE_BANDWIDTH_BPS, BandwidthTracker
+
+    tracker = BandwidthTracker(alpha=1.0)
+    assert tracker.bandwidth_bps_ema is None
+
+    # Record a 50 KB GET taking 10 ms (sample 1, below 5 threshold)
+    size = 50 * 1024
+    duration = 0.010
+    tracker.record_observation(size, duration)
+
+    _, lat, _, _ = tracker.get_metrics()
+    expected_transfer = size / float(_ASSUMED_AGGREGATE_BANDWIDTH_BPS)
+    expected_lat = max(0.001, duration - expected_transfer)
+    assert lat is not None
+    assert pytest.approx(lat, abs=1e-6) == expected_lat
+
+
+def test_transfer_subtracted_latency_clamped():
+    """Provide an observation where estimated transfer time > observed duration (e.g. cached/fast GET).
+    Verify resulting latency is clamped to epsilon (0.001s) rather than becoming zero or negative.
+    """
+    import pytest
+
+    from litdata.raw.dataset import _LATENCY_EPSILON_S, BandwidthTracker
+
+    tracker = BandwidthTracker(alpha=1.0)
+    # Seed 5 large GETs at a slow prior BPS (1 MB/s)
+    slow_bps = 1 * 1024 * 1024
+    large_size = 10 * 1024 * 1024
+    for _ in range(5):
+        tracker.record_observation(large_size, large_size / slow_bps)
+
+    # Now record a 100 KB GET arriving in only 5 ms (faster than 100 ms transfer estimate)
+    tracker.record_observation(100 * 1024, 0.005)
+    _, lat, _, _ = tracker.get_metrics()
+
+    assert lat is not None
+    assert pytest.approx(lat, abs=1e-9) == _LATENCY_EPSILON_S
+
+
+def test_current_bandwidth_sample_does_not_explain_itself():
+    """Verify that the bandwidth calculated from the current observation is NOT used to calculate
+    that observation's own transfer time.
+    """
+    from litdata.raw.dataset import _LATENCY_EPSILON_S, BandwidthTracker
+
+    tracker = BandwidthTracker(alpha=1.0)
+    # Record a single 200 KB GET taking 30 ms (bps_sample_count = 0 before this sample).
+    # If circular, it would use 200 KB / 30 ms to calculate transfer = 30 ms -> lat = 0 -> clamped to epsilon (1 ms).
+    # Since it correctly uses fallback default (100 MB/s), transfer = 2 ms -> lat = 28 ms.
+    tracker.record_observation(200 * 1024, 0.030)
+    _, lat, _, _ = tracker.get_metrics()
+
+    assert lat is not None
+    assert lat > 10 * _LATENCY_EPSILON_S  # 28 ms is >> 1 ms epsilon