diff --git a/test/rb/test_ensemble.py b/test/rb/test_ensemble.py
index 2da22a87128..6030729387f 100644
--- a/test/rb/test_ensemble.py
+++ b/test/rb/test_ensemble.py
@@ -437,7 +437,7 @@ def test_rb_multidim(self, datatype, datadim, rbtype, storage_cls, sampler_cls):
         s = rb.sample()
         assert str(rb)
         if datatype in ("tensordict", "tensorclass"):
-            assert (s.exclude("index") == 1).all()
+            assert (s.exclude("index", "index_generation") == 1).all()
             assert s.numel() == 4
         else:
             for leaf in tree_iter(s):
diff --git a/test/rb/test_samplers.py b/test/rb/test_samplers.py
index 2e410f160a6..9abb6f14581 100644
--- a/test/rb/test_samplers.py
+++ b/test/rb/test_samplers.py
@@ -243,7 +243,7 @@ def test_sampler_without_rep_state_dict(self, backend):
             replay_buffer.extend(transition.clone())
         for _ in range(n_samples):
             s = replay_buffer.sample(batch_size=1)
-            assert (s.exclude("index") == 1).all()
+            assert (s.exclude("index", "index_generation") == 1).all()
 
         replay_buffer.extend(torch.zeros_like(transition))
 
@@ -257,7 +257,7 @@ def test_sampler_without_rep_state_dict(self, backend):
 
         new_replay_buffer.load_state_dict(state_dict)
         s = new_replay_buffer.sample(batch_size=1)
-        assert (s.exclude("index") == 0).all()
+        assert (s.exclude("index", "index_generation") == 0).all()
 
     def test_sampler_without_rep_dumps_loads(self, tmpdir):
         d0 = tmpdir + "/save0"
diff --git a/test/rb/test_storages.py b/test/rb/test_storages.py
index 228733e8a8f..76932a87cb0 100644
--- a/test/rb/test_storages.py
+++ b/test/rb/test_storages.py
@@ -306,7 +306,7 @@ def test_storage_state_dict(self, storage_in, storage_out, init_out, backend):
 
         new_replay_buffer.load_state_dict(state_dict)
         s = new_replay_buffer.sample()
-        assert (s.exclude("index") == 1).all()
+        assert (s.exclude("index", "index_generation") == 1).all()
 
     @pytest.mark.skipif(
         TORCH_VERSION < version.parse("2.5.0"), reason="requires Torch >= 2.5.0"
diff --git a/test/rb/test_writers.py b/test/rb/test_writers.py
index ffe47a51ffb..1b218471311 100644
--- a/test/rb/test_writers.py
+++ b/test/rb/test_writers.py
@@ -403,6 +403,283 @@ def test_roundrobin_dumps_loads_write_count(self, tmp_path):
         assert writer2._write_count == 23
 
 
+class TestWriterGeneration:
+    def test_default_writer_tracks_generations(self):
+        rb = ReplayBuffer(storage=LazyTensorStorage(10))
+        assert rb._writer.tracks_generations is True
+        index = rb.extend(torch.arange(10))
+        gen = rb._writer.generations_of(index)
+        assert gen.dtype == torch.int64
+        assert gen.shape == index.shape
+        assert (gen == 0).all()
+
+    def test_non_tracking_writer_reports_minus_one(self):
+        writer = TensorDictMaxValueWriter(rank_key="key")
+        assert writer.tracks_generations is False
+        gen = writer.generations_of(torch.arange(4))
+        torch.testing.assert_close(gen, torch.full((4,), -1, dtype=torch.int64))
+
+    def test_generation_increments_on_reuse(self):
+        size = 4
+        rb = ReplayBuffer(storage=LazyTensorStorage(size))
+        rb.extend(torch.arange(size))
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size)),
+            torch.zeros(size, dtype=torch.int64),
+        )
+        rb.extend(torch.arange(size, size + 3))
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size)), torch.tensor([1, 1, 1, 0])
+        )
+
+    def test_generation_wraparound(self):
+        size = 5
+        rb = ReplayBuffer(storage=LazyTensorStorage(size))
+        rb.extend(torch.arange(2 * size))
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size)),
+            torch.full((size,), 1, dtype=torch.int64),
+        )
+
+    def test_generation_extend_wrapping_twice(self):
+        size = 4
+        rb = ReplayBuffer(storage=LazyTensorStorage(size))
+        # slots 0 and 1 are written three times, slots 2 and 3 twice
+        rb.extend(torch.arange(2 * size + 2))
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size)), torch.tensor([2, 2, 1, 1])
+        )
+
+    def test_generation_add(self):
+        size = 3
+        rb = ReplayBuffer(storage=LazyTensorStorage(size))
+        for i in range(size + 1):
+            rb.add(torch.tensor(i))
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size)), torch.tensor([1, 0, 0])
+        )
+
+    def test_generations_of_unwritten_reports_minus_one(self):
+        rb = ReplayBuffer(storage=LazyTensorStorage(4))
+        rb.extend(torch.arange(2))
+        gen = rb._writer.generations_of(torch.arange(4))
+        torch.testing.assert_close(gen, torch.tensor([0, 0, -1, -1]))
+
+    def test_generation_tensordict_writer(self):
+        size = 4
+        rb = TensorDictReplayBuffer(storage=LazyTensorStorage(size))
+        rb.extend(TensorDict({"a": torch.arange(2 * size)}, [2 * size]))
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size)),
+            torch.full((size,), 1, dtype=torch.int64),
+        )
+
+    def test_generation_write_at(self):
+        storage = LazyTensorStorage(4)
+        writer = RoundRobinWriter()
+        writer.register_storage(storage)
+        writer.extend(torch.arange(4))
+        writer.write_at(torch.tensor([0, 1]), torch.tensor([10, 11]))
+        torch.testing.assert_close(
+            writer.generations_of(torch.arange(4)), torch.tensor([1, 1, 0, 0])
+        )
+
+    def test_empty_is_monotonic(self):
+        rb = ReplayBuffer(storage=LazyTensorStorage(10))
+        index = rb.extend(torch.arange(10))
+        before = rb._writer.generations_of(index)
+        rb.empty()
+        rb.extend(torch.arange(10))
+        after = rb._writer.generations_of(index)
+        assert (after > before).all()
+
+    def test_empty_invalidates_handles_immediately(self):
+        rb = ReplayBuffer(storage=LazyTensorStorage(10))
+        index = rb.extend(torch.arange(10))
+        gen = rb._writer.generations_of(index)
+        rb.empty()
+        assert (rb._writer.generations_of(index) != gen).all()
+
+    def test_empty_preserves_unwritten_sentinel(self):
+        rb = ReplayBuffer(storage=LazyTensorStorage(4))
+        rb.extend(torch.arange(2))
+        rb.empty()
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(4)), torch.tensor([1, 1, -1, -1])
+        )
+
+    def test_generation_state_dict_roundtrip(self):
+        size = 4
+        rb = ReplayBuffer(storage=LazyTensorStorage(size))
+        rb.extend(torch.arange(size + 1))
+        sd = rb.state_dict()
+        rb2 = ReplayBuffer(storage=LazyTensorStorage(size))
+        rb2.load_state_dict(sd)
+        torch.testing.assert_close(
+            rb2._writer.generations_of(torch.arange(size)),
+            rb._writer.generations_of(torch.arange(size)),
+        )
+
+    def test_legacy_state_dict_without_generation_loads(self):
+        rb = ReplayBuffer(storage=LazyTensorStorage(10))
+        rb.extend(torch.arange(5))
+        sd = rb.state_dict()
+        del sd["_writer"]["_generation"]
+        rb2 = ReplayBuffer(storage=LazyTensorStorage(10))
+        rb2.load_state_dict(sd)
+        assert rb2._writer._cursor == 5
+
+    def test_generation_dumps_loads(self, tmp_path):
+        writer = RoundRobinWriter()
+        writer._cursor = 2
+        writer._write_count = 9
+        writer._generation = torch.tensor([3, 2, 2, 1])
+        writer.dumps(tmp_path)
+        writer2 = RoundRobinWriter()
+        writer2.loads(tmp_path)
+        assert writer2._cursor == 2
+        assert writer2._write_count == 9
+        torch.testing.assert_close(
+            writer2.generations_of(torch.arange(4)), torch.tensor([3, 2, 2, 1])
+        )
+
+    def test_sample_returns_generation(self):
+        size = 8
+        rb = ReplayBuffer(storage=LazyTensorStorage(size))
+        rb.extend(torch.arange(size))
+        _, info = rb.sample(4, return_info=True)
+        assert "index_generation" in info
+        gen = torch.as_tensor(info["index_generation"])
+        idx = torch.as_tensor(info["index"])
+        assert gen.shape == idx.shape
+        torch.testing.assert_close(gen, rb._writer.generations_of(idx))
+
+    def test_non_tracking_sample_has_no_generation(self):
+        rb = TensorDictReplayBuffer(
+            storage=LazyTensorStorage(10),
+            writer=TensorDictMaxValueWriter(rank_key="key"),
+        )
+        rb.extend(TensorDict({"key": torch.arange(10), "a": torch.arange(10)}, [10]))
+        _, info = rb.sample(4, return_info=True)
+        assert "index_generation" not in info
+
+    def test_tensordict_sample_has_generation_key(self):
+        size = 8
+        rb = TensorDictReplayBuffer(storage=LazyTensorStorage(size))
+        rb.extend(TensorDict({"a": torch.arange(size)}, [size]))
+        sample = rb.sample(4)
+        assert "index_generation" in sample.keys()
+        assert sample["index_generation"].shape[0] == 4
+
+    def test_wraparound_race_detectable(self):
+        size = 8
+        rb = ReplayBuffer(storage=LazyTensorStorage(size))
+        rb.extend(torch.arange(size))
+        _, info = rb.sample(4, return_info=True)
+        sampled_index = torch.as_tensor(info["index"])
+        sampled_generation = torch.as_tensor(info["index_generation"])
+        rb.extend(torch.arange(size, 2 * size))
+        current = rb._writer.generations_of(sampled_index)
+        assert (current != sampled_generation).all()
+
+    def test_partial_reuse_detectable(self):
+        size = 8
+        rb = ReplayBuffer(storage=LazyTensorStorage(size))
+        rb.extend(torch.arange(size))
+        _, info = rb.sample(size, return_info=True)
+        idx = torch.as_tensor(info["index"])
+        gen = torch.as_tensor(info["index_generation"])
+        rb.extend(torch.arange(size, size + 3))
+        stale = rb._writer.generations_of(idx) != gen
+        torch.testing.assert_close(stale, idx < 3)
+
+    @pytest.mark.parametrize("device", get_default_devices())
+    def test_generation_on_storage_device(self, device):
+        size = 8
+        rb = ReplayBuffer(storage=LazyTensorStorage(size, device=device))
+        rb.extend(torch.arange(size, device=device))
+        assert rb._writer._generation.device.type == device.type
+        _, info = rb.sample(4, return_info=True)
+        gen = info["index_generation"]
+        idx = torch.as_tensor(info["index"])
+        assert gen.device == idx.device
+        torch.testing.assert_close(gen, rb._writer.generations_of(idx))
+        rb.extend(torch.arange(size, 2 * size, device=device))
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size, device=device)),
+            torch.ones(size, dtype=torch.int64, device=device),
+        )
+
+    @pytest.mark.parametrize("device", get_default_devices())
+    def test_generation_add_on_storage_device(self, device):
+        size = 3
+        rb = ReplayBuffer(storage=LazyTensorStorage(size, device=device))
+        for i in range(size + 1):
+            rb.add(torch.tensor(i, device=device))
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size, device=device)),
+            torch.tensor([1, 0, 0], device=device),
+        )
+
+    @pytest.mark.gpu
+    @pytest.mark.skipif(not torch.cuda.is_available(), reason="needs CUDA")
+    def test_generation_cuda_data_into_cuda_storage(self):
+        size = 8
+        rb = ReplayBuffer(storage=LazyTensorStorage(size, device="cuda"))
+        rb.extend(torch.arange(size, device="cuda"))
+        assert rb._writer._generation.device.type == "cuda"
+        _, info = rb.sample(4, return_info=True)
+        idx = torch.as_tensor(info["index"])
+        assert info["index_generation"].device == idx.device
+        rb.extend(torch.arange(size, 2 * size, device="cuda"))
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size, device="cuda")),
+            torch.ones(size, dtype=torch.int64, device="cuda"),
+        )
+
+    @pytest.mark.gpu
+    @pytest.mark.skipif(not torch.cuda.is_available(), reason="needs CUDA")
+    def test_generation_cpu_data_into_cuda_storage(self):
+        size = 4
+        rb = TensorDictReplayBuffer(storage=LazyTensorStorage(size, device="cuda"))
+        rb.extend(TensorDict({"a": torch.arange(2 * size)}, [2 * size]))
+        assert rb._writer._generation.device.type == "cuda"
+        torch.testing.assert_close(
+            rb._writer.generations_of(torch.arange(size)),
+            torch.ones(size, dtype=torch.int64),
+        )
+
+    @pytest.mark.gpu
+    @pytest.mark.skipif(not torch.cuda.is_available(), reason="needs CUDA")
+    def test_generation_state_dict_roundtrip_cuda(self):
+        size = 4
+        rb = ReplayBuffer(storage=LazyTensorStorage(size, device="cuda"))
+        rb.extend(torch.arange(size + 1, device="cuda"))
+        rb2 = ReplayBuffer(storage=LazyTensorStorage(size, device="cuda"))
+        rb2.load_state_dict(rb.state_dict())
+        index = torch.arange(size, device="cuda")
+        assert rb2._writer._generation.device.type == "cuda"
+        torch.testing.assert_close(
+            rb2._writer.generations_of(index), rb._writer.generations_of(index)
+        )
+
+    @pytest.mark.gpu
+    @pytest.mark.skipif(not torch.cuda.is_available(), reason="needs CUDA")
+    def test_generation_dumps_loads_cuda(self, tmp_path):
+        writer = RoundRobinWriter()
+        writer.register_storage(LazyTensorStorage(4, device="cuda"))
+        writer._generation = torch.tensor([3, 2, 2, 1], device="cuda")
+        writer.dumps(tmp_path)
+        writer2 = RoundRobinWriter()
+        writer2.register_storage(LazyTensorStorage(4, device="cuda"))
+        writer2.loads(tmp_path)
+        assert writer2._generation.device.type == "cuda"
+        torch.testing.assert_close(
+            writer2.generations_of(torch.arange(4, device="cuda")),
+            torch.tensor([3, 2, 2, 1], device="cuda"),
+        )
+
+
 if __name__ == "__main__":
     args, unknown = argparse.ArgumentParser().parse_known_args()
     pytest.main([__file__, "--capture", "no", "--exitfirst"] + unknown)
diff --git a/torchrl/data/replay_buffers/replay_buffers.py b/torchrl/data/replay_buffers/replay_buffers.py
index a5d781e71df..64942fbace7 100644
--- a/torchrl/data/replay_buffers/replay_buffers.py
+++ b/torchrl/data/replay_buffers/replay_buffers.py
@@ -1516,6 +1516,8 @@ def _sample(self, batch_size: int) -> tuple[Any, dict]:
         with self._replay_lock if not is_comp else nc, self._write_lock if not is_comp else nc:
             index, info = self._sampler.sample(self._storage, batch_size)
             info["index"] = index
+            if self._writer.tracks_generations:
+                info["index_generation"] = self._writer.generations_of(index)
             data = self._storage.get(_storage_index(index, self._storage))
         if not isinstance(index, INT_CLASSES):
             data = self._collate_fn(data)
@@ -2135,6 +2137,8 @@ def _sample(self, batch_size: int) -> tuple[Any, dict]:
         ):
             index, info = self.prioritized_sampler.sample(self._storage, batch_size)
             info["index"] = index
+            if self._writer.tracks_generations:
+                info["index_generation"] = self._writer.generations_of(index)
             data = self._storage.get(_storage_index(index, self._storage))
         if not isinstance(index, INT_CLASSES):
             data = self._collate_fn(data)
@@ -2564,6 +2568,8 @@ def _sample(self, batch_size: int) -> tuple[Any, dict]:
         with self._replay_lock if not is_comp else nc, self._write_lock if not is_comp else nc:
             index, info = self._sampler.sample(self._storage, batch_size)
             info["index"] = index
+            if self._writer.tracks_generations:
+                info["index_generation"] = self._writer.generations_of(index)
             data = self._storage.get(_storage_index(index, self._storage))
         if not isinstance(index, INT_CLASSES):
             data = self._collate_fn(data)
@@ -2929,6 +2935,8 @@ def _sample(self, batch_size: int) -> tuple[Any, dict]:
         ):
             index, info = self.prioritized_sampler.sample(self._storage, batch_size)
             info["index"] = index
+            if self._writer.tracks_generations:
+                info["index_generation"] = self._writer.generations_of(index)
             data = self._storage.get(_storage_index(index, self._storage))
         if not isinstance(index, INT_CLASSES):
             data = self._collate_fn(data)
diff --git a/torchrl/data/replay_buffers/writers.py b/torchrl/data/replay_buffers/writers.py
index 9726e9861dd..59932b42773 100644
--- a/torchrl/data/replay_buffers/writers.py
+++ b/torchrl/data/replay_buffers/writers.py
@@ -19,7 +19,7 @@
 from tensordict import is_tensor_collection, MemoryMappedTensor, TensorDictBase
 from tensordict.utils import expand_as_right, is_tensorclass
 from torch import multiprocessing as mp
-from torchrl._utils import _STRDTYPE2DTYPE
+from torchrl._utils import _make_ordinal_device, _STRDTYPE2DTYPE
 
 try:
     from torch.compiler import disable as compile_disable
@@ -39,6 +39,10 @@ def tree_leaves(data):  # noqa: D103
 from torchrl.data.replay_buffers.storages import Storage
 from torchrl.data.replay_buffers.utils import _is_int, _reduce
 
+# Storage capacities at or above this value are treated as unbounded (lazy
+# storages report a sentinel max size), triggering dynamic generation growth.
+_GENERATION_UNBOUNDED = 2**40
+
 
 class Writer(ABC):
     """A ReplayBuffer base Writer class."""
@@ -50,9 +54,22 @@ def __init__(self, compilable: bool = False) -> None:
         self._storage = None
         self._compilable = compilable
 
+    #: Whether this writer type stamps storage slots with a reuse generation.
+    tracks_generations: bool = False
+
     def register_storage(self, storage: Storage) -> None:
         self._storage = storage
 
+    def generations_of(self, index: int | torch.Tensor) -> torch.Tensor:
+        """Returns the generation stamp for each physical slot in ``index``.
+
+        The stamp advances once per write to that slot, so a single ``extend``
+        that wraps the storage advances a reused slot once per write it
+        receives. Writers that do not track slot reuse report ``-1``.
+        """
+        index = torch.as_tensor(index)
+        return torch.full(index.shape, -1, dtype=torch.int64, device=index.device)
+
     @abstractmethod
     def add(self, data: Any) -> int:
         """Inserts one piece of data at an appropriate index, and returns that index."""
@@ -155,16 +172,123 @@ class RoundRobinWriter(Writer):
 
     """
 
+    tracks_generations: bool = True
+
     def __init__(self, compilable: bool = False) -> None:
         super().__init__(compilable=compilable)
         self._cursor = 0
         self._write_count  # noqa
+        self._generation = None
+
+    def register_storage(self, storage: Storage) -> None:
+        super().register_storage(storage)
+        self._align_generation_device()
+
+    def _generation_device(self, index: int | torch.Tensor) -> torch.device:
+        # The generation buffer follows the storage: sampled indices are built on
+        # the storage device, so lookups stay sync-free on the sampling path.
+        device = getattr(self._storage, "device", None)
+        if device is None or device == "auto":
+            if isinstance(index, torch.Tensor):
+                return _make_ordinal_device(index.device)
+            return torch.device("cpu")
+        return _make_ordinal_device(torch.device(device))
+
+    def _align_generation_device(self) -> None:
+        generation = self._generation
+        if generation is None:
+            return
+        device = self._generation_device(generation)
+        if generation.device != device:
+            self._generation = generation.to(device)
+
+    def _ensure_generation(
+        self, capacity: int, min_size: int, device: torch.device
+    ) -> None:
+        # Bounded storages allocate to capacity once (stable shape, so the
+        # ``torch.compile`` extend/sample path does not recompile); lazy storages
+        # report a sentinel capacity and instead grow geometrically.
+        generation = self._generation
+        if generation is not None and generation.device != device:
+            generation = generation.to(device)
+            self._generation = generation
+        current = 0 if generation is None else generation.numel()
+        if current >= min_size:
+            return
+        size = (
+            capacity if capacity < _GENERATION_UNBOUNDED else max(min_size, current * 2)
+        )
+        new_generation = torch.full((size,), -1, dtype=torch.int64, device=device)
+        if generation is not None:
+            new_generation[:current] = generation
+        if not self._compilable and new_generation.device.type == "cpu":
+            new_generation.share_memory_()
+        self._generation = new_generation
+
+    def _bump_generation(self, index: int | torch.Tensor, data: Any) -> None:
+        device = self._generation_device(index)
+        if _is_int(index):
+            capacity = self._storage._max_size_along_dim0(single_data=data)
+            self._ensure_generation(capacity, int(index) + 1, device)
+            self._generation[int(index)] += 1
+        else:
+            index = torch.as_tensor(index, dtype=torch.long).reshape(-1)
+            if index.numel() == 0:
+                return
+            capacity = self._storage._max_size_along_dim0(batched_data=data)
+            min_size = (
+                capacity if capacity < _GENERATION_UNBOUNDED else int(index.max()) + 1
+            )
+            self._ensure_generation(capacity, min_size, device)
+            index = index.to(device)
+            self._generation.index_put_(
+                (index,), torch.ones_like(index), accumulate=True
+            )
+
+    def generations_of(self, index: int | torch.Tensor) -> torch.Tensor:
+        if isinstance(index, tuple):
+            index = index[0]
+        elif (
+            isinstance(index, torch.Tensor)
+            and index.ndim
+            and self._storage is not None
+            and self._storage.ndim > 1
+            and index.shape[-1] == self._storage.ndim
+        ):
+            index = index[..., 0]
+        index = torch.as_tensor(index, dtype=torch.long)
+        if self._generation is None:
+            return torch.full(index.shape, -1, dtype=torch.int64, device=index.device)
+        idx = index.to(self._generation.device)
+        n = self._generation.numel()
+        gen = self._generation[idx.clamp(max=n - 1)]
+        gen = torch.where(idx < n, gen, torch.full_like(gen, -1))
+        return gen.to(index.device)
 
     def dumps(self, path):
         path = Path(path).absolute()
         path.mkdir(exist_ok=True)
+        metadata = {
+            "cursor": self._cursor,
+            "write_count": self._write_count,
+        }
+        generation = self._generation
+        if generation is not None:
+            generation = generation.cpu()
+            try:
+                MemoryMappedTensor.from_filename(
+                    filename=path / "generation.memmap",
+                    shape=generation.shape,
+                    dtype=generation.dtype,
+                ).copy_(generation)
+            except FileNotFoundError:
+                MemoryMappedTensor.from_tensor(
+                    generation, filename=path / "generation.memmap"
+                )
+            metadata["generation_shape"] = list(generation.shape)
+            metadata["generation_dtype"] = str(generation.dtype)
         with open(path / "metadata.json", "w") as file:
-            json.dump({"cursor": self._cursor, "write_count": self._write_count}, file)
+            json.dump(metadata, file)
 
     def loads(self, path):
         path = Path(path).absolute()
@@ -174,6 +298,17 @@ def loads(self, path):
             write_count = metadata.get("write_count")
             if write_count is not None:
                 self._write_count = write_count
+            generation_shape = metadata.get("generation_shape")
+        if generation_shape is not None:
+            generation = MemoryMappedTensor.from_filename(
+                filename=path / "generation.memmap",
+                dtype=_STRDTYPE2DTYPE[metadata["generation_dtype"]],
+                shape=torch.Size(generation_shape),
+            ).clone()
+            if not self._compilable:
+                generation.share_memory_()
+            self._generation = generation
+            self._align_generation_device()
 
     def add(self, data: Any) -> int | torch.Tensor:
         index = self._cursor
@@ -186,6 +321,7 @@ def add(self, data: Any) -> int | torch.Tensor:
         # Replicate index requires the shape of the storage to be known
         # Other than that, a "flat" (1d) index is ok to write the data
         self._storage.set(_cursor, data)
+        self._bump_generation(_cursor, data)
         index = self._replicate_index(index)
         self._mark_update_entities(index)
         return index
@@ -214,12 +350,17 @@ def extend(self, data: Sequence) -> torch.Tensor:
         # Replicate index requires the shape of the storage to be known
         # Other than that, a "flat" (1d) index is ok to write the data
         self._storage.set(index, data)
+        self._bump_generation(index, data)
         index = self._replicate_index(index)
         self._mark_update_entities(index)
         return index
 
     def write_at(self, index: int | torch.Tensor, data: Any) -> int | torch.Tensor:
-        """Writes data at explicit storage indices without moving the cursor."""
+        """Writes data at explicit storage indices without moving the cursor.
+
+        The generation of every written slot is bumped, so handles previously
+        handed out for those slots are stale once this returns.
+        """
         if _is_int(index):
             batch_size = 1
         else:
@@ -229,6 +370,7 @@ def write_at(self, index: int | torch.Tensor, data: Any) -> int | torch.Tensor:
             batch_size = index.numel()
         self._write_count += batch_size
         self._storage.set(index, data, set_cursor=False)
+        self._bump_generation(index, data)
         self._update_storage_len_for_write_at(index)
         index = self._replicate_index(index)
         self._mark_update_entities(index)
@@ -249,16 +391,30 @@ def _update_storage_len_for_write_at(self, index: int | torch.Tensor) -> None:
         )
 
     def state_dict(self) -> dict[str, Any]:
-        return {"_cursor": self._cursor, "_write_count": self._write_count}
+        state_dict = {"_cursor": self._cursor, "_write_count": self._write_count}
+        if self._generation is not None:
+            state_dict["_generation"] = self._generation.clone()
+        return state_dict
 
     def load_state_dict(self, state_dict: dict[str, Any]) -> None:
         self._cursor = state_dict["_cursor"]
         write_count = state_dict.get("_write_count")
         if write_count is not None:
             self._write_count = write_count
+        generation = state_dict.get("_generation")
+        if generation is not None:
+            generation = generation.clone()
+            if not self._compilable and generation.device.type == "cpu":
+                generation.share_memory_()
+            self._generation = generation
+            self._align_generation_device()
 
     def _empty(self, empty_write_count: bool = True) -> None:
         self._cursor = 0
+        generation = self._generation
+        if generation is not None:
+            # never-written slots keep the -1 sentinel
+            generation[generation >= 0] += 1
         if empty_write_count:
             self._write_count = 0
 
@@ -363,6 +519,7 @@ def add(self, data: Any) -> int | torch.Tensor:
                 ),
             )
         self._storage.set(index, data)
+        self._bump_generation(index, data)
         index = self._replicate_index(index)
         self._mark_update_entities(index)
         return index
@@ -392,6 +549,7 @@ def extend(self, data: Sequence) -> torch.Tensor:
         # Replicate index requires the shape of the storage to be known
         # Other than that, a "flat" (1d) index is ok to write the data
         self._storage.set(index, data)
+        self._bump_generation(index, data)
         index = self._replicate_index(index)
         self._mark_update_entities(index)
         return index
@@ -407,6 +565,7 @@ def write_at(self, index: int | torch.Tensor, data: Any) -> int | torch.Tensor:
         if not is_tensorclass(data):
             data.set("index", expand_as_right(index_tensor, data))
         self._storage.set(index_tensor, data, set_cursor=False)
+        self._bump_generation(index_tensor, data)
         self._update_storage_len_for_write_at(index_tensor)
         index = self._replicate_index(index_tensor)
         self._mark_update_entities(index)