diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2312727735d..72ab0437da7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -8,6 +8,7 @@
 * [CHANGE] Removed the following deprecated config: `-querier.filter-queryables-enabled`, `-query-frontend.cache-samples-processed-stats`, `-ingest-storage.kafka.write-clients`, `-blocks-storage.tsdb.head-postings-for-matchers-cache-size`, `-blocks-storage.tsdb.block-postings-for-matchers-cache-size`. #16352
 * [CHANGE] The `bucket` label of the `thanos_objstore_bucket_*` metrics, previously always empty, is now set to the name of the bucket the metrics refer to. This lets a component that accesses more than one bucket report each of them separately. The `thanos_store_bucket_cache_*` and `cortex_bucket_index_load*` metrics gained a `bucket` label carrying the same bucket name, for the same reason. #16265
 * [CHANGE] Compactor: Stabilize `-compactor.first-level-compaction-skip-future-max-time` to `true` and `-compactor.first-level-compaction-ooo-wait-period` to 5 minutes, both of which have been shown to improve batching during the split phase and reduce the total volume of L2 blocks in deployments with lots of out-of-order writes. #16464
+* [ENHANCEMENT] Usage-tracker: Speed up the periodic removal of idle series. The scan that locates expired series now runs a whole group of slots at a time, using SSE2 on amd64 and NEON on arm64, so groups holding nothing to remove are skipped without inspecting each slot. This shortens the time a shard mutex is held during cleanup, which was showing up as latency spikes on series tracking in large partitions. Build with `-tags nosimd` to use the portable implementation. #16473
 * [ENHANCEMENT] Compactor: Add the experimental `-compactor.block-health-validation-concurrency` option to limit how many blocks are validated concurrently within a compaction job. #16269
 * [ENHANCEMENT] Query-frontend: Improve the stability of cardinality estimates and therefore sharding factors for queries when running splitting and caching inside MQE is enabled, or range vector splitting is enabled. #16274 #16301 #16305 #16311
   * When running splitting and caching inside MQE is enabled, the `cortex_query_frontend_cardinality_estimation_difference` metric will no longer be emitted.
diff --git a/pkg/usagetracker/clock/minutes.go b/pkg/usagetracker/clock/minutes.go
index 70ec5c1db6e..b96d330ffa0 100644
--- a/pkg/usagetracker/clock/minutes.go
+++ b/pkg/usagetracker/clock/minutes.go
@@ -22,6 +22,10 @@ func ToMinutes(t time.Time) Minutes {
 // This value only makes sense within the last hour.
 type Minutes uint8
 
+// Cycle is the number of values Minutes takes before wrapping back to zero.
+// ToMinutes always returns a value below it, and comparisons are only defined for such values.
+const Cycle Minutes = 2 * 60
+
 // GreaterThan returns true if this value is greater than other on a four-hour clock face assuming that none of the values is ever older than 1h.
 func (m Minutes) GreaterThan(other Minutes) bool {
 	if m > other {
diff --git a/pkg/usagetracker/tenantshard/expiry.go b/pkg/usagetracker/tenantshard/expiry.go
new file mode 100644
index 00000000000..41055ca0404
--- /dev/null
+++ b/pkg/usagetracker/tenantshard/expiry.go
@@ -0,0 +1,106 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package tenantshard
+
+import (
+	"math/bits"
+	"unsafe"
+
+	"github.com/grafana/mimir/pkg/usagetracker/clock"
+)
+
+// Finding the entries to evict during Cleanup used to be a per-slot loop with three
+// unpredictable branches per slot, which dominated the cost of Cleanup. This file turns
+// that test into a branchless, whole-group operation, so that groups holding nothing to
+// evict are skipped without ever branching per slot.
+//
+// # Turning the clock comparison into a range check
+//
+// An entry is evicted when watermark.GreaterOrEqualThan(value) holds, where value is
+// clock.Minutes in [0, 120) and watermark is also in [0, 120). Expanding
+// clock.Minutes.GreaterThan, that condition is (watermark-value) mod 120 < 60, so the
+// values that expire are the 60 consecutive minutes ending at the watermark, wrapping
+// around the 120 minute clock face.
+//
+// Slots hold the value xor-ed (see xorData), that is x = 255-value, so the expiring
+// values map to the byte range starting at lo = 255-watermark and running upwards,
+// wrapping at 256. Two cases:
+//
+//   - watermark >= 59: the minutes do not wrap, and x lands in [lo, lo+59], entirely
+//     inside [136, 255].
+//   - watermark < 59: the minutes wrap, and x lands in [lo, 255] plus [136, 194-watermark].
+//     Going up from lo and wrapping at 256, those two pieces are joined by [0, 135], which
+//     holds no valid value, so the whole thing is still one range: [lo, lo+195].
+//
+// Both cases are therefore "uint8(x-lo) <= length", differing only in length. The second
+// case sweeps up the empty (0) and tombstone (1) markers as a side effect, so the result
+// is always intersected with a separate occupied test.
+//
+// # Domain
+//
+// This is exact for values and watermarks in [0, 120), which is what clock.ToMinutes
+// produces. Snapshots are the only path that can introduce a byte outside that range, and
+// loadSnapshot rejects those before they reach the map.
+
+// expiredRange returns the range of xorData byte values that Cleanup must evict for the
+// given watermark: a slot expires when uint8(x-lo) <= length and the slot is occupied.
+func expiredRange(watermark clock.Minutes) (lo, length uint8) {
+	lo = ^uint8(watermark) // 255 - watermark
+	if watermark >= 59 {
+		// The 60 minute window ending at the watermark does not wrap around the clock face.
+		return lo, 59
+	}
+	return lo, 195
+}
+
+// matchExpired returns a bitset with the high bit of lane j set when slot j of the group
+// is occupied and its value falls in the expired range described by lo and length.
+func (d *data) matchExpired(lo, length uint8) bitset {
+	return matchExpiredWord(castUint64Data(d), lo, length)
+}
+
+// matchExpiredWord is the portable implementation of matchExpired over one 8 byte group.
+// It mirrors, lane for lane, what the assembly implementations compute with a byte-wise
+// subtract, an unsigned compare and an occupied test.
+func matchExpiredWord(x uint64, lo, length uint8) bitset {
+	y := subBytes(x, lo)
+	// Occupied means the slot holds neither empty (0) nor tombstone (1), that is x > 1,
+	// which for bytes is the same as x &^ 1 being non-zero.
+	notOccupied := findZeroBytes(x &^ loBits)
+	return leBytes(y, length) &^ notOccupied
+}
+
+// subBytes subtracts n from every byte lane of x, wrapping at 256 per lane instead of
+// borrowing into the next lane.
+func subBytes(x uint64, n uint8) uint64 {
+	nb := uint64(n) * loBits
+	// Setting the high bit of every lane of x guarantees each lane is at least as large as
+	// the corresponding lane of nb&^hiBits, so the subtraction never borrows across lanes.
+	// The final xor restores the high bits to what a per-lane subtraction would produce.
+	return ((x | hiBits) - (nb &^ hiBits)) ^ ((x ^ ^nb) & hiBits)
+}
+
+// leBytes sets the high bit of every byte lane of x that is less than or equal to n,
+// unsigned, and clears the others.
+func leBytes(x uint64, n uint8) bitset {
+	// Compare the low 7 bits of each lane by borrowing into the spare high bit.
+	low := (((uint64(n&0x7f) * loBits) | hiBits) - (x &^ hiBits)) & hiBits
+	if n < 0x80 {
+		// x <= n also requires the high bit of the lane to be clear.
+		return bitset(low & ^x)
+	}
+	// Any lane with its high bit clear is below 0x80 and therefore below n.
+	return bitset((^x & hiBits) | low)
+}
+
+// lastMatch clears and returns the index of the highest set lane in the given bitset.
+// It is the counterpart of nextMatch and assumes the bitset is non-zero.
+func lastMatch(b *bitset) uint32 {
+	s := uint32(63 - bits.LeadingZeros64(uint64(*b)))
+	*b &= ^(1 << s)
+	return s >> 3
+}
+
+func castUint64Data(d *data) uint64 {
+	return *(*uint64)(unsafe.Pointer(d))
+}
diff --git a/pkg/usagetracker/tenantshard/expiry_amd64.go b/pkg/usagetracker/tenantshard/expiry_amd64.go
new file mode 100644
index 00000000000..9360d29a68a
--- /dev/null
+++ b/pkg/usagetracker/tenantshard/expiry_amd64.go
@@ -0,0 +1,16 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build amd64 && !nosimd
+
+package tenantshard
+
+// simdImpl names the scanExpired implementation compiled into this binary. It is only used
+// for reporting in tests and benchmarks.
+const simdImpl = "sse2"
+
+// scanExpired returns the index of the first group in p[:n] that holds at least one
+// occupied, expired slot, or n when there is none. See matchExpiredWord for the predicate
+// it evaluates, and expiry_amd64.s for the implementation.
+//
+//go:noescape
+func scanExpired(p *data, n int, lo, length uint8) int
diff --git a/pkg/usagetracker/tenantshard/expiry_amd64.s b/pkg/usagetracker/tenantshard/expiry_amd64.s
new file mode 100644
index 00000000000..2eb0eaa58ca
--- /dev/null
+++ b/pkg/usagetracker/tenantshard/expiry_amd64.s
@@ -0,0 +1,87 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build amd64 && !nosimd
+
+#include "textflag.h"
+
+// func scanExpired(p *data, n int, lo, length uint8) int
+//
+// Evaluates matchExpiredWord over two groups at a time. SSE2 has no unsigned byte compare,
+// so "x-lo <= length" is done with a saturating subtract: it lands on zero exactly when the
+// lane is within range. The same trick against 1 identifies the empty and tombstone
+// markers, which PANDN then removes from the result.
+//
+// Everything here is SSE2, which is part of the amd64 baseline, so there is no feature
+// detection to do.
+TEXT ·scanExpired(SB), NOSPLIT, $0-32
+	MOVQ	p+0(FP), SI
+	MOVQ	n+8(FP), CX
+	MOVBLZX	lo+16(FP), AX
+	MOVBLZX	length+17(FP), DX
+
+	XORQ	DI, DI                  // index of the group under the cursor
+	MOVQ	$0x0101010101010101, R8
+
+	// Broadcast each constant across all 16 lanes by splatting it into a general purpose
+	// register first, which avoids the SSE2 unpack dance.
+	IMULQ	R8, AX
+	MOVQ	AX, X1                  // lo in every lane
+	MOVLHPS	X1, X1
+	IMULQ	R8, DX
+	MOVQ	DX, X2                  // length in every lane
+	MOVLHPS	X2, X2
+	MOVQ	R8, X3                  // tombstone marker in every lane
+	MOVLHPS	X3, X3
+	PXOR	X0, X0
+
+loop16:
+	CMPQ	CX, $2
+	JLT	tail
+	MOVOU	(SI), X4
+	MOVO	X4, X5
+	PSUBB	X1, X5                  // X5 = x - lo, wrapping per lane
+	PSUBUSB	X2, X5                  // saturates to zero when x-lo <= length
+	PCMPEQB	X0, X5                  // X5 = lanes within range
+	MOVO	X4, X6
+	PSUBUSB	X3, X6                  // saturates to zero when x <= 1
+	PCMPEQB	X0, X6                  // X6 = lanes holding empty or tombstone
+	PANDN	X5, X6                  // X6 = in range and occupied
+	PMOVMSKB X6, BX
+	TESTL	BX, BX
+	JNZ	found
+	ADDQ	$16, SI
+	ADDQ	$2, DI
+	SUBQ	$2, CX
+	JMP	loop16
+
+found:
+	TESTL	$0xff, BX
+	JNZ	done                    // the first of the two groups hit
+	ADDQ	$1, DI
+	JMP	done
+
+	// An odd number of groups leaves a single 8 byte group to check. Loading it into the
+	// low half of the register zeroes the high half, and zero lanes read as empty, so they
+	// can never produce a spurious hit.
+tail:
+	TESTQ	CX, CX
+	JZ	notfound
+	MOVQ	(SI), X4
+	MOVO	X4, X5
+	PSUBB	X1, X5
+	PSUBUSB	X2, X5
+	PCMPEQB	X0, X5
+	MOVO	X4, X6
+	PSUBUSB	X3, X6
+	PCMPEQB	X0, X6
+	PANDN	X5, X6
+	PMOVMSKB X6, BX
+	TESTL	$0xff, BX
+	JNZ	done
+
+notfound:
+	MOVQ	n+8(FP), DI
+
+done:
+	MOVQ	DI, ret+24(FP)
+	RET
diff --git a/pkg/usagetracker/tenantshard/expiry_arm64.go b/pkg/usagetracker/tenantshard/expiry_arm64.go
new file mode 100644
index 00000000000..fe0790832b3
--- /dev/null
+++ b/pkg/usagetracker/tenantshard/expiry_arm64.go
@@ -0,0 +1,16 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build arm64 && !nosimd
+
+package tenantshard
+
+// simdImpl names the scanExpired implementation compiled into this binary. It is only used
+// for reporting in tests and benchmarks.
+const simdImpl = "neon"
+
+// scanExpired returns the index of the first group in p[:n] that holds at least one
+// occupied, expired slot, or n when there is none. See matchExpiredWord for the predicate
+// it evaluates, and expiry_arm64.s for the implementation.
+//
+//go:noescape
+func scanExpired(p *data, n int, lo, length uint8) int
diff --git a/pkg/usagetracker/tenantshard/expiry_arm64.s b/pkg/usagetracker/tenantshard/expiry_arm64.s
new file mode 100644
index 00000000000..82ec332922b
--- /dev/null
+++ b/pkg/usagetracker/tenantshard/expiry_arm64.s
@@ -0,0 +1,71 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build arm64 && !nosimd
+
+#include "textflag.h"
+
+// func scanExpired(p *data, n int, lo, length uint8) int
+//
+// Evaluates matchExpiredWord over two groups at a time: subtract lo from every byte lane,
+// keep the lanes that land within length of it, and drop the lanes holding the empty or
+// tombstone markers.
+//
+// The unsigned byte comparisons are expressed with UMIN and UMAX rather than with CMHS and
+// CMHI, because the assembler only learned the latter in Go 1.27: "y <= n" is
+// "umin(y, n) == y" and "x >= 2" is "umax(x, 2) == x". Two extra instructions per iteration,
+// and it assembles on the Go version Mimir currently builds with.
+TEXT ·scanExpired(SB), NOSPLIT|NOFRAME, $0-32
+	MOVD	p+0(FP), R0
+	MOVD	n+8(FP), R1
+	MOVBU	lo+16(FP), R2
+	MOVBU	length+17(FP), R3
+
+	VDUP	R2, V0.B16              // lo in every lane
+	VDUP	R3, V1.B16              // length in every lane
+	MOVD	$2, R4
+	VDUP	R4, V2.B16              // lowest occupied byte value in every lane
+
+	MOVD	$0, R5                  // index of the group under the cursor
+
+loop16:
+	CMP	$2, R1
+	BLT	tail
+	VLD1	(R0), [V4.B16]
+	VSUB	V0.B16, V4.B16, V5.B16  // V5 = x - lo, wrapping per lane
+	VUMIN	V1.B16, V5.B16, V6.B16
+	VCMEQ	V5.B16, V6.B16, V6.B16  // V6 = lanes where x-lo <= length, unsigned
+	VUMAX	V2.B16, V4.B16, V7.B16
+	VCMEQ	V4.B16, V7.B16, V7.B16  // V7 = lanes where x >= 2, that is occupied
+	VAND	V6.B16, V7.B16, V7.B16
+	VMOV	V7.D[0], R6             // lanes of the first group
+	VMOV	V7.D[1], R7             // lanes of the second group
+	CBNZ	R6, done
+	CBZ	R7, next16
+	ADD	$1, R5
+	B	done
+
+next16:
+	ADD	$16, R0
+	ADD	$2, R5
+	SUB	$2, R1
+	B	loop16
+
+	// An odd number of groups leaves a single 8 byte group to check.
+tail:
+	CBZ	R1, notfound
+	VLD1	(R0), [V4.B8]
+	VSUB	V0.B8, V4.B8, V5.B8
+	VUMIN	V1.B8, V5.B8, V6.B8
+	VCMEQ	V5.B8, V6.B8, V6.B8
+	VUMAX	V2.B8, V4.B8, V7.B8
+	VCMEQ	V4.B8, V7.B8, V7.B8
+	VAND	V6.B8, V7.B8, V7.B8
+	VMOV	V7.D[0], R6
+	CBNZ	R6, done
+
+notfound:
+	MOVD	n+8(FP), R5
+
+done:
+	MOVD	R5, ret+24(FP)
+	RET
diff --git a/pkg/usagetracker/tenantshard/expiry_generic.go b/pkg/usagetracker/tenantshard/expiry_generic.go
new file mode 100644
index 00000000000..e082e11fea5
--- /dev/null
+++ b/pkg/usagetracker/tenantshard/expiry_generic.go
@@ -0,0 +1,22 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+//go:build (!amd64 && !arm64) || nosimd
+
+package tenantshard
+
+import "unsafe"
+
+// simdImpl names the scanExpired implementation compiled into this binary. It is only used
+// for reporting in tests and benchmarks.
+const simdImpl = "generic"
+
+// scanExpired returns the index of the first group in p[:n] that holds at least one
+// occupied, expired slot, or n when there is none.
+func scanExpired(p *data, n int, lo, length uint8) int {
+	for i, d := range unsafe.Slice(p, n) {
+		if d.matchExpired(lo, length) != 0 {
+			return i
+		}
+	}
+	return n
+}
diff --git a/pkg/usagetracker/tenantshard/expiry_test.go b/pkg/usagetracker/tenantshard/expiry_test.go
new file mode 100644
index 00000000000..88f9830481f
--- /dev/null
+++ b/pkg/usagetracker/tenantshard/expiry_test.go
@@ -0,0 +1,332 @@
+// SPDX-License-Identifier: AGPL-3.0-only
+
+package tenantshard
+
+import (
+	"fmt"
+	"math/rand"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/grafana/mimir/pkg/usagetracker/clock"
+)
+
+// TestMatchExpiredExhaustive checks the branchless range test against the clock comparison
+// it replaces, for every watermark and every value clock.ToMinutes can produce.
+func TestMatchExpiredExhaustive(t *testing.T) {
+	for w := 0; w < 120; w++ {
+		watermark := clock.Minutes(w)
+		lo, length := expiredRange(watermark)
+
+		for v := 0; v < 120; v++ {
+			value := clock.Minutes(v)
+			x := uint64(xor(value)) * loBits // same value in all 8 lanes
+			want := watermark.GreaterOrEqualThan(value)
+			got := matchExpiredWord(x, lo, length)
+
+			if want {
+				require.Equal(t, bitset(hiBits), got, "watermark=%d value=%d: every lane should be expired", w, v)
+			} else {
+				require.Zero(t, got, "watermark=%d value=%d: no lane should be expired", w, v)
+			}
+		}
+
+		// Empty and tombstone markers must never be reported, whatever the watermark.
+		require.Zero(t, matchExpiredWord(uint64(empty), lo, length), "watermark=%d: empty slot reported as expired", w)
+		require.Zero(t, matchExpiredWord(uint64(tombstone)*loBits, lo, length), "watermark=%d: tombstone reported as expired", w)
+	}
+}
+
+// TestMatchExpiredMixedLanes checks that lanes are independent: each slot of a group is
+// evaluated on its own value, not on its neighbours.
+func TestMatchExpiredMixedLanes(t *testing.T) {
+	r := rand.New(rand.NewSource(1))
+	for i := 0; i < 20000; i++ {
+		watermark := clock.Minutes(r.Intn(120))
+		lo, length := expiredRange(watermark)
+
+		var d data
+		var want bitset
+		for j := range d {
+			switch r.Intn(4) {
+			case 0:
+				d[j] = empty
+			case 1:
+				d[j] = tombstone
+			default:
+				value := clock.Minutes(r.Intn(120))
+				d[j] = xor(value)
+				if watermark.GreaterOrEqualThan(value) {
+					want |= bitset(0x80) << (8 * j)
+				}
+			}
+		}
+
+		require.Equal(t, want, d.matchExpired(lo, length), "watermark=%d data=%v", watermark, d)
+	}
+}
+
+func TestLastMatch(t *testing.T) {
+	b := bitset(0)
+	for _, lane := range []uint32{0, 3, 7} {
+		b |= bitset(0x80) << (8 * lane)
+	}
+	require.Equal(t, uint32(7), lastMatch(&b))
+	require.Equal(t, uint32(3), lastMatch(&b))
+	require.Equal(t, uint32(0), lastMatch(&b))
+	require.Zero(t, b)
+}
+
+// TestCleanupPreservesLookups is the property that matters after Cleanup rearranges slots:
+// every entry that should survive must still be found by a probe, and every entry that
+// should be gone must be re-created on the next Put. It exercises maps that are dense
+// enough to spill probes across groups and to force tombstones.
+func TestCleanupPreservesLookups(t *testing.T) {
+	r := rand.New(rand.NewSource(1))
+
+	for round := 0; round < 300; round++ {
+		size := 1 + r.Intn(400)
+		m := New(uint32(size))
+
+		// Insert more than the map was sized for, so groups fill up and probes spill over.
+		entries := map[uint64]clock.Minutes{}
+		for i := 0; i < size*2; i++ {
+			key := r.Uint64()
+			value := clock.Minutes(r.Intn(120))
+			m.Put(key, value, nil, nil, false)
+			if _, ok := entries[key]; !ok {
+				entries[key] = value
+			} else {
+				// Put keeps the newest value when tracking, which is what we asked for above.
+				entries[key] = value
+			}
+		}
+
+		watermark := clock.Minutes(r.Intn(120))
+		survivors := map[uint64]clock.Minutes{}
+		for key, value := range entries {
+			if !watermark.GreaterOrEqualThan(value) {
+				survivors[key] = value
+			}
+		}
+
+		removed := m.Cleanup(watermark, nil)
+		require.Equal(t, len(entries)-len(survivors), removed, "round %d: unexpected number of removals", round)
+		require.Equal(t, len(survivors), m.Count(), "round %d: unexpected count", round)
+
+		// Items() must report exactly the survivors, with their values intact.
+		got := map[uint64]clock.Minutes{}
+		_, items := m.Items()
+		for key, value := range items {
+			got[key] = value
+		}
+		require.Equal(t, survivors, got, "round %d: unexpected contents", round)
+
+		// A probe for a survivor must find it, so Put reports it as already existing.
+		// A probe for a removed key must not find it, so Put reports it as created.
+		for key, value := range entries {
+			created, _ := m.Put(key, value, nil, nil, false)
+			_, survived := survivors[key]
+			require.Equal(t, !survived, created, "round %d: key %d lookup after cleanup", round, key)
+		}
+	}
+}
+
+// scanExpiredReference is the obvious implementation that scanExpired must agree with on
+// every architecture.
+func scanExpiredReference(d []data, lo, length uint8) int {
+	for i := range d {
+		if d[i].matchExpired(lo, length) != 0 {
+			return i
+		}
+	}
+	return len(d)
+}
+
+// TestScanExpiredMatchesReference covers the assembly implementations against the portable
+// one, including the odd-length tail and the empty input.
+func TestScanExpiredMatchesReference(t *testing.T) {
+	t.Logf("scanExpired implementation: %s", simdImpl)
+	r := rand.New(rand.NewSource(1))
+
+	for round := 0; round < 5000; round++ {
+		n := r.Intn(9) // covers 0, the 8 byte tail, and several 16 byte iterations
+		d := make([]data, n+1)
+		for i := 0; i < n; i++ {
+			for j := range d[i] {
+				switch r.Intn(6) {
+				case 0:
+					d[i][j] = empty
+				case 1:
+					d[i][j] = tombstone
+				default:
+					d[i][j] = xor(clock.Minutes(r.Intn(120)))
+				}
+			}
+		}
+		// Poison the group past the end so an over-read would be caught.
+		for j := range d[n] {
+			d[n][j] = xor(0)
+		}
+
+		watermark := clock.Minutes(r.Intn(120))
+		lo, length := expiredRange(watermark)
+
+		want := scanExpiredReference(d[:n], lo, length)
+		got := scanExpired(&d[0], n, lo, length)
+		require.Equal(t, want, got, "round %d: n=%d watermark=%d data=%v", round, n, watermark, d[:n])
+	}
+}
+
+// TestScanExpiredAllExpired and its counterpart pin down the two extremes, where the branch
+// in the scan loop always goes the same way.
+func TestScanExpiredExtremes(t *testing.T) {
+	lo, length := expiredRange(60)
+
+	for n := 1; n <= 9; n++ {
+		none := make([]data, n)
+		for i := range none {
+			for j := range none[i] {
+				none[i][j] = empty
+			}
+		}
+		require.Equal(t, n, scanExpired(&none[0], n, lo, length), "n=%d: empty groups must report no hit", n)
+
+		all := make([]data, n)
+		for i := range all {
+			for j := range all[i] {
+				all[i][j] = xor(60)
+			}
+		}
+		require.Equal(t, 0, scanExpired(&all[0], n, lo, length), "n=%d: first group must hit", n)
+
+		// A hit in the last group only, which is the tail group when n is odd.
+		last := make([]data, n)
+		for i := range last {
+			for j := range last[i] {
+				last[i][j] = empty
+			}
+		}
+		last[n-1][3] = xor(60)
+		require.Equal(t, n-1, scanExpired(&last[0], n, lo, length), "n=%d: last group must hit", n)
+	}
+}
+
+// cleanupLegacy is the per-slot Cleanup loop that scanExpired replaced. It is kept here to
+// compare behaviour and speed against the current implementation. The trailing rehash is
+// left out so that both can be measured on the same map.
+func (m *Map) cleanupLegacy(watermark clock.Minutes) int {
+	removed := 0
+groups:
+	for i := range m.data {
+		for j := uint32(0); j < groupSize; {
+			if m.data[i][j] == empty {
+				continue groups
+			}
+			if m.data[i][j] == tombstone {
+				j++
+				continue
+			}
+			if watermark.GreaterOrEqualThan(m.data[i][j].clockMinutes()) {
+				removed++
+
+				if emptySlots := m.index[i].matchEmpty(); emptySlots != 0 {
+					m.resident--
+					e := nextMatch(&emptySlots)
+					if e == j+1 {
+						m.index[i][j] = empty
+						m.keys[i][j] = 0
+						m.data[i][j] = empty
+						continue groups
+					}
+
+					m.index[i][j], m.index[i][e-1] = m.index[i][e-1], empty
+					m.keys[i][j], m.keys[i][e-1] = m.keys[i][e-1], 0
+					m.data[i][j], m.data[i][e-1] = m.data[i][e-1], empty
+					continue
+				}
+
+				m.index[i][j] = tombstone
+				m.keys[i][j] = 0
+				m.data[i][j] = tombstone
+				m.dead++
+			}
+			j++
+		}
+	}
+	return removed
+}
+
+// TestCleanupMatchesLegacy runs both implementations over identical maps and requires them
+// to agree on what survives and on the bookkeeping counters. The new one visits slots in a
+// different order, so surviving entries may sit in different slots of the same group, which
+// is why the comparison is on contents rather than on the raw arrays.
+func TestCleanupMatchesLegacy(t *testing.T) {
+	r := rand.New(rand.NewSource(1))
+
+	for round := 0; round < 500; round++ {
+		size := 1 + r.Intn(300)
+		seed := r.Int63()
+
+		build := func() *Map {
+			m := New(uint32(size))
+			br := rand.New(rand.NewSource(seed))
+			for i := 0; i < size*2; i++ {
+				m.Put(br.Uint64(), clock.Minutes(br.Intn(120)), nil, nil, false)
+			}
+			return m
+		}
+
+		watermark := clock.Minutes(r.Intn(120))
+		legacy, current := build(), build()
+
+		wantRemoved := legacy.cleanupLegacy(watermark)
+		gotRemoved := current.Cleanup(watermark, nil)
+
+		require.Equal(t, wantRemoved, gotRemoved, "round %d: removed count", round)
+		require.Equal(t, legacy.resident, current.resident, "round %d: resident", round)
+		require.Equal(t, legacy.dead, current.dead, "round %d: dead", round)
+		require.Equal(t, legacy.Count(), current.Count(), "round %d: count", round)
+
+		collect := func(m *Map) map[uint64]clock.Minutes {
+			got := map[uint64]clock.Minutes{}
+			_, items := m.Items()
+			for key, value := range items {
+				got[key] = value
+			}
+			return got
+		}
+		require.Equal(t, collect(legacy), collect(current), "round %d: contents", round)
+	}
+}
+
+// BenchmarkMapCleanupImpl compares the per-slot loop against the scanExpired one on the
+// same data, at the two expired fractions that bracket real usage.
+func BenchmarkMapCleanupImpl(b *testing.B) {
+	b.Logf("scanExpired implementation: %s", simdImpl)
+	for _, size := range []int{16e6} {
+		for _, fraction := range []float64{0, 0.01, 0.05, 0.25, 1} {
+			m, keys, values, watermark := buildMapForCleanup(size, fraction, 1)
+
+			run := func(b *testing.B, cleanup func()) {
+				b.ResetTimer()
+				for i := 0; i < b.N; i++ {
+					cleanup()
+					b.StopTimer()
+					refill(m, keys, values)
+					b.StartTimer()
+				}
+				b.StopTimer()
+				b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N)/float64(size), "ns/entry")
+			}
+
+			b.Run(fmt.Sprintf("size=%d/expired=%.0f%%/impl=legacy", size, fraction*100), func(b *testing.B) {
+				run(b, func() { m.cleanupLegacy(watermark) })
+			})
+			b.Run(fmt.Sprintf("size=%d/expired=%.0f%%/impl=scan", size, fraction*100), func(b *testing.B) {
+				run(b, func() { m.Cleanup(watermark, nil) })
+			})
+		}
+	}
+}
diff --git a/pkg/usagetracker/tenantshard/map.go b/pkg/usagetracker/tenantshard/map.go
index 4c78189dd20..5714f298407 100644
--- a/pkg/usagetracker/tenantshard/map.go
+++ b/pkg/usagetracker/tenantshard/map.go
@@ -223,59 +223,21 @@ func (m *Map) Stats() Stats {
 	}
 }
 
+// Cleanup removes every entry whose timestamp is at or before the watermark, and returns
+// how many were removed. Groups are located with scanExpired, so groups that hold nothing
+// to remove are skipped without inspecting their slots one by one.
 func (m *Map) Cleanup(watermark clock.Minutes, limit *atomic.Uint64) int {
 	removed := 0
-groups:
-	for i := range m.data {
-		for j := uint32(0); j < groupSize; {
-			if m.data[i][j] == empty {
-				// There's nothing here.
-				// Hence, there's nothing in the next slots.
-				continue groups
-			}
-			if m.data[i][j] == tombstone {
-				// Already deleted, skip.
-				j++
-				continue
-			}
-			if watermark.GreaterOrEqualThan(m.data[i][j].clockMinutes()) {
-				removed++
-
-				// We want to avoid creating tombstones. Every time we create a tombstone, we get closer to the rehash of the map.
-				// Rehash of the map is slow and creates garbage, slowing down the entire service.
-				if emptySlots := m.index[i].matchEmpty(); emptySlots != 0 {
-					// We target groups to be half-full, so it's likely that there are empty slots in this group so far.
-					// If there's an empty slot in this group, it means that no elements that were originally targeting this group
-					// have been written to a next group, which in turn means that we can safely move the elements in this group.
-					m.resident--
-					e := nextMatch(&emptySlots)
-					if e == j+1 {
-						// This is the last element in the group, just mark it as empty and move to the next group.
-						m.index[i][j] = empty
-						m.keys[i][j] = 0
-						m.data[i][j] = empty
-						continue groups
-					}
-
-					// There are more elements in the group, move the last element in the group to this position.
-					// Set that element position to empty.
-					m.index[i][j], m.index[i][e-1] = m.index[i][e-1], empty
-					m.keys[i][j], m.keys[i][e-1] = m.keys[i][e-1], 0
-					m.data[i][j], m.data[i][e-1] = m.data[i][e-1], empty
-
-					// Continue checking the same position again.
-					continue
-				}
-
-				// Bad luck, the group is full, just set a tombstone and keep checking.
-				m.index[i][j] = tombstone
-				m.keys[i][j] = 0
-				m.data[i][j] = tombstone
-				m.dead++
-			}
-			j++
+	lo, length := expiredRange(watermark)
+	for i := 0; i < len(m.data); {
+		hit := i + scanExpired(&m.data[i], len(m.data)-i, lo, length)
+		if hit >= len(m.data) {
+			break
 		}
+		removed += m.cleanupGroup(hit, lo, length)
+		i = hit + 1
 	}
+
 	if m.dead > m.limit/2 {
 		var lim uint64
 		if limit != nil {
@@ -286,6 +248,52 @@ groups:
 	return removed
 }
 
+// cleanupGroup removes every occupied, expired slot of group i and returns how many were
+// removed.
+//
+// Slots are visited from the highest index down. Removing a slot fills the hole with the
+// last occupied slot of the group, and going downwards guarantees that this last slot is a
+// live entry: every expired slot above the one being removed has already been dealt with.
+func (m *Map) cleanupGroup(i int, lo, length uint8) int {
+	removed := 0
+	for expired := m.data[i].matchExpired(lo, length); expired != 0; {
+		j := lastMatch(&expired)
+		removed++
+
+		// We want to avoid creating tombstones. Every time we create a tombstone, we get closer to the rehash of the map.
+		// Rehash of the map is slow and creates garbage, slowing down the entire service.
+		emptySlots := m.index[i].matchEmpty()
+		if emptySlots == 0 {
+			// Bad luck, the group is full, just set a tombstone.
+			m.index[i][j] = tombstone
+			m.keys[i][j] = 0
+			m.data[i][j] = tombstone
+			m.dead++
+			continue
+		}
+
+		// We target groups to be half-full, so it's likely that there are empty slots in this group so far.
+		// If there's an empty slot in this group, it means that no elements that were originally targeting this group
+		// have been written to a next group, which in turn means that we can safely move the elements in this group.
+		m.resident--
+		e := nextMatch(&emptySlots)
+		if e == j+1 {
+			// This is the last occupied slot of the group, just mark it as empty.
+			m.index[i][j] = empty
+			m.keys[i][j] = 0
+			m.data[i][j] = empty
+			continue
+		}
+
+		// There are more elements in the group, move the last element in the group to this position.
+		// Set that element position to empty.
+		m.index[i][j], m.index[i][e-1] = m.index[i][e-1], empty
+		m.keys[i][j], m.keys[i][e-1] = m.keys[i][e-1], 0
+		m.data[i][j], m.data[i][e-1] = m.data[i][e-1], empty
+	}
+	return removed
+}
+
 // EnsureCapacity ensure that the map has enough capacity to store |n| elements.
 // This does not mean that the map will have n empty slots, there might be already n elements in the map and 0 spare capacity.
 // If there's no enough capacity, the map is rehashed to accommodate at least |n| elements.
diff --git a/pkg/usagetracker/tenantshard/map_test.go b/pkg/usagetracker/tenantshard/map_test.go
index 03369c19bf7..84538fcc96a 100644
--- a/pkg/usagetracker/tenantshard/map_test.go
+++ b/pkg/usagetracker/tenantshard/map_test.go
@@ -678,3 +678,132 @@ func BenchmarkMapTrackCleanupGarbage(b *testing.B) {
 	}
 	b.Logf("Rehashes: %d, rehashes per iteration %.2f", m.rehashes, float64(m.rehashes)/float64(b.N))
 }
+
+// buildMapForCleanup fills a Map with size entries whose timestamps are spread so that
+// approximately expiredFraction of them are older than the returned watermark.
+// Keys are returned so callers can refill the removed entries between timed iterations.
+func buildMapForCleanup(size int, expiredFraction float64, seed int64) (m *Map, keys []uint64, values []clock.Minutes, watermark clock.Minutes) {
+	// Live entries get minute 60, expired ones get minute 0. Watermark 30 splits them.
+	// All values stay well inside the 1h window clock.Minutes comparisons require.
+	const (
+		expiredMinute = clock.Minutes(0)
+		watermarkTs   = clock.Minutes(30)
+		liveMinute    = clock.Minutes(60)
+	)
+
+	m = New(uint32(size))
+	keys = make([]uint64, size)
+	values = make([]clock.Minutes, size)
+	r := rand.New(rand.NewSource(seed))
+	expiredUpTo := int(float64(size) * expiredFraction)
+	for i := 0; i < size; i++ {
+		keys[i] = r.Uint64()
+		if i < expiredUpTo {
+			values[i] = expiredMinute
+		} else {
+			values[i] = liveMinute
+		}
+		m.Put(keys[i], values[i], nil, nil, false)
+	}
+	return m, keys, values, watermarkTs
+}
+
+// refill re-inserts every key, restoring the map to the state it had before Cleanup ran.
+func refill(m *Map, keys []uint64, values []clock.Minutes) {
+	for i, k := range keys {
+		m.Put(k, values[i], nil, nil, false)
+	}
+}
+
+func logCleanupStats(b *testing.B, m *Map, size, removed int) {
+	s := m.Stats()
+	b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N)/float64(size), "ns/entry")
+	b.Logf("size=%d removed=%d resident=%d dead=%d limit=%d groups=%d rehashes=%d bytes=%d",
+		size, removed, s.Resident, s.Dead, s.Limit, s.Length, s.Rehashes, s.Length*80)
+}
+
+// BenchmarkMapCleanupBySize measures Cleanup as the shard grows. 64e6 entries per shard
+// corresponds to ~1e9 series per partition (16 shards).
+func BenchmarkMapCleanupBySize(b *testing.B) {
+	for _, size := range []int{1e6, 4e6, 16e6, 64e6} {
+		b.Run(fmt.Sprintf("size=%d", size), func(b *testing.B) {
+			m, keys, values, watermark := buildMapForCleanup(size, 0.25, 1)
+			removed := 0
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				removed = m.Cleanup(watermark, nil)
+				b.StopTimer()
+				refill(m, keys, values)
+				b.StartTimer()
+			}
+			b.StopTimer()
+			logCleanupStats(b, m, size, removed)
+		})
+	}
+}
+
+// BenchmarkMapCleanupByExpiredFraction isolates the cost of the plain scan (0% expired,
+// only the data array is read) from the cost of the per-removal compaction.
+func BenchmarkMapCleanupByExpiredFraction(b *testing.B) {
+	const size = 16e6
+	for _, fraction := range []float64{0, 0.01, 0.05, 0.25, 1} {
+		b.Run(fmt.Sprintf("expired=%.0f%%", fraction*100), func(b *testing.B) {
+			m, keys, values, watermark := buildMapForCleanup(size, fraction, 1)
+			removed := 0
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				removed = m.Cleanup(watermark, nil)
+				b.StopTimer()
+				refill(m, keys, values)
+				b.StartTimer()
+			}
+			b.StopTimer()
+			logCleanupStats(b, m, size, removed)
+		})
+	}
+}
+
+// BenchmarkMapRehashRealistic measures the rehash that Cleanup triggers when
+// dead > limit/2. Unlike BenchmarkMapRehash, it rehashes into the group count nextSize()
+// would actually pick, so the allocation figure matches production.
+func BenchmarkMapRehashRealistic(b *testing.B) {
+	for _, size := range []int{1e6, 16e6, 64e6} {
+		b.Run(fmt.Sprintf("size=%d", size), func(b *testing.B) {
+			m, _, _, _ := buildMapForCleanup(size, 0, 1)
+			target := m.nextSize(0)
+			b.Logf("entries=%d groups now=%d nextSize=%d", size, len(m.index), target)
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				m.rehash(target)
+			}
+			b.StopTimer()
+			b.ReportMetric(float64(b.Elapsed().Nanoseconds())/float64(b.N)/float64(size), "ns/entry")
+		})
+	}
+}
+
+// BenchmarkMapCleanupSparse measures an oversized map holding few live entries, which is
+// what a shrinking tenant leaves behind: Cleanup only shrinks via the dead > limit/2 rehash.
+func BenchmarkMapCleanupSparse(b *testing.B) {
+	const groups = 16e6 / maxAvgGroupLoad
+	for _, live := range []int{0, 1e3, 1e6} {
+		b.Run(fmt.Sprintf("live=%d", live), func(b *testing.B) {
+			m := New(16e6)
+			keys := make([]uint64, live)
+			values := make([]clock.Minutes, live)
+			r := rand.New(rand.NewSource(1))
+			for i := 0; i < live; i++ {
+				keys[i] = r.Uint64()
+				values[i] = clock.Minutes(60)
+				m.Put(keys[i], values[i], nil, nil, false)
+			}
+			b.ResetTimer()
+			for i := 0; i < b.N; i++ {
+				m.Cleanup(clock.Minutes(30), nil)
+			}
+			b.StopTimer()
+			s := m.Stats()
+			b.Logf("live=%d groups=%d (allocated for %d) resident=%d", live, s.Length, int(groups), s.Resident)
+		})
+	}
+}
diff --git a/pkg/usagetracker/tracker_store_bench_test.go b/pkg/usagetracker/tracker_store_bench_test.go
index b1ba4e0501d..c229432f6cc 100644
--- a/pkg/usagetracker/tracker_store_bench_test.go
+++ b/pkg/usagetracker/tracker_store_bench_test.go
@@ -6,6 +6,7 @@ import (
 	"context"
 	"fmt"
 	"math/rand"
+	"slices"
 	"strconv"
 	"sync"
 	"testing"
@@ -156,3 +157,118 @@ func generateSnapshot(b *testing.B, tenantsCount int, totalSeries int, now time.
 	}
 	return snapshots, lim
 }
+
+// BenchmarkTrackerStoreCleanupLockStall measures how long trackSeries() blocks while
+// cleanup() holds a shard mutex. trackerStore.cleanup takes shard.Lock() for the whole
+// Map.Cleanup call, so the stall is bounded by the per-shard cleanup time, and the
+// minTimeBetweenShardsCleanup delay does not break up a single hold.
+func BenchmarkTrackerStoreCleanupLockStall(b *testing.B) {
+	for _, seriesPerShard := range []int{250e3, 1e6, 4e6} {
+		b.Run(fmt.Sprintf("seriesPerShard=%d", seriesPerShard), func(b *testing.B) {
+			// 5% is what a steady state looks like: cleanup runs every minute and only the
+			// series that crossed the idle timeout in that minute are removed. 20% stands in
+			// for a burst of churn.
+			for _, expiredFraction := range []float64{0.05, 0.2} {
+				b.Run(fmt.Sprintf("expired=%.0f%%", expiredFraction*100), func(b *testing.B) {
+					for _, interShardDelay := range []time.Duration{0, 25 * time.Millisecond} {
+						b.Run(fmt.Sprintf("interShardDelay=%s", interShardDelay), func(b *testing.B) {
+							benchmarkCleanupLockStall(b, seriesPerShard, expiredFraction, interShardDelay)
+						})
+					}
+				})
+			}
+		})
+	}
+}
+
+func benchmarkCleanupLockStall(b *testing.B, seriesPerShard int, expiredFraction float64, interShardDelay time.Duration) {
+	const tenant = "0"
+	total := seriesPerShard * shards
+	lim := limiterMock{tenant: uint64(total) * 2}
+	t := newTrackerStore(testIdleTimeout, 85, log.NewNopLogger(), lim, noopEvents{}, false, interShardDelay)
+
+	// Spread creation timestamps far enough back that the requested fraction of them falls
+	// outside the idle timeout when cleanup runs.
+	now := time.Now()
+	spread := time.Duration(float64(testIdleTimeout) / (1 - expiredFraction))
+	batches := 25
+	perBatch := total / batches
+	r := rand.New(rand.NewSource(1))
+	for i := 0; i < batches; i++ {
+		series := make([]uint64, perBatch)
+		for j := range series {
+			series[j] = r.Uint64()
+		}
+		ts := now.Add(-spread + time.Duration(i)*spread/time.Duration(batches))
+		_, err := t.trackSeries(context.Background(), tenant, series, ts)
+		require.NoError(b, err)
+	}
+
+	// Steady-state series used by the latency probe. They are already tracked, so each
+	// probe call is a pure update: no allocation, no event publishing.
+	probe := make([]uint64, 100)
+	copy(probe, make([]uint64, 100))
+	for j := range probe {
+		probe[j] = r.Uint64()
+	}
+	_, err := t.trackSeries(context.Background(), tenant, probe, now)
+	require.NoError(b, err)
+
+	measure := func(withCleanup bool) []time.Duration {
+		var latencies []time.Duration
+		stop := make(chan struct{})
+		done := make(chan struct{})
+		go func() {
+			defer close(done)
+			buf := make([]uint64, len(probe))
+			for {
+				select {
+				case <-stop:
+					return
+				default:
+				}
+				copy(buf, probe)
+				t0 := time.Now()
+				_, err := t.trackSeries(context.Background(), tenant, buf, now)
+				latencies = append(latencies, time.Since(t0))
+				if err != nil {
+					panic(err)
+				}
+			}
+		}()
+
+		if withCleanup {
+			t.cleanup(now)
+		} else {
+			// Idle for a comparable amount of time to collect a baseline.
+			time.Sleep(100 * time.Millisecond)
+		}
+		close(stop)
+		<-done
+		return latencies
+	}
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		base := measure(false)
+		during := measure(true)
+		b.StopTimer()
+		reportLatencies(b, "baseline", base)
+		reportLatencies(b, "cleanup", during)
+		b.StartTimer()
+	}
+}
+
+func reportLatencies(b *testing.B, name string, latencies []time.Duration) {
+	if len(latencies) == 0 {
+		b.Logf("%s: no samples", name)
+		return
+	}
+	sorted := slices.Clone(latencies)
+	slices.Sort(sorted)
+	pct := func(p float64) time.Duration {
+		i := int(float64(len(sorted)-1) * p)
+		return sorted[i]
+	}
+	b.Logf("%s: n=%d p50=%s p99=%s max=%s", name, len(sorted), pct(0.5), pct(0.99), sorted[len(sorted)-1])
+}
diff --git a/pkg/usagetracker/tracker_store_snapshot.go b/pkg/usagetracker/tracker_store_snapshot.go
index 3c09ec32413..ac6bebf5ae1 100644
--- a/pkg/usagetracker/tracker_store_snapshot.go
+++ b/pkg/usagetracker/tracker_store_snapshot.go
@@ -146,6 +146,11 @@ func (t *trackerStore) loadSnapshot(data []byte, now time.Time) error {
 			if err := snapshot.Err(); err != nil {
 				return fmt.Errorf("failed to read series timestamp %d: %w", i, err)
 			}
+			if snapshotTs >= clock.Cycle {
+				// The file is corrupted: no timestamp we ever write is out of the clock face.
+				// Loading it would store a series that comparisons cannot reason about, so drop it.
+				continue
+			}
 			if expirationWatermark.GreaterThan(snapshotTs) {
 				// We're not interested in this series, it was about to be evicted.
 				continue