diff --git a/CHANGELOG.md b/CHANGELOG.md
index 95bf08ac54..2a1fc830e5 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,12 +15,14 @@ For details about compatibility between different releases, see the **Commitment
 - `ttgc.managed-gateway-euis` configuration option: Gateway EUI prefixes of managed gateways, defaulting to the EUI prefix of The Things Industries managed gateways. Gateways outside these prefixes are reported as not managed in the claiming info.
 - `ttgc.lbscups.lns-port` configuration option: the LoRa Basics Station LNS port of the Gateway Server, defaulting to `8887`.
 - Downlink scheduling on all antennas of a gateway. Previously, only the first antenna could be used as a downlink path, and uplinks received on any other antenna had their downlink path disabled.
+- `events.redis.store.expire-limiter-size` configuration option: how many entity event streams the `PEXPIRE` rate limiter tracks, at 8 bytes each, defaulting to `4194304` (32 MiB). A negative value refreshes the TTL on every event.
 
 ### Changed
 
 - In the Semtech UDP Packet Forwarder protocol, `PUSH_ACK` and `PULL_ACK` are only sent after the gateway has connected to the Gateway Server and the gateway's `PUSH_DATA` or `PULL_DATA` respectively has been accepted.
 - Don't log a `Task failed` warning in GS for every task attached to a gateway connection when that connection is closed. The tasks that run for the lifetime of a gateway connection now stop without an error when the connection is closed, and the disconnection is logged once, as `Disconnected`, including the reason. This removes several duplicate warnings per gateway disconnection, which were particularly noisy for gateways on unreliable backhaul.
 - Websocket close errors on the LoRa Basics Station frontend (such as `websocket: close 1006 (abnormal closure): unexpected EOF`, which is what a gateway disappearing without a close handshake looks like) are now reported as the defined error `pkg/gatewayserver/io/semtechws:websocket_closed`, with the close code as an attribute and the original error as the cause. As a result, the `gs.gateway.disconnect` event for these disconnections now carries structured error details instead of a plain string; consumers that parse the event data should expect the `ErrorDetails` format.
+- The events Redis store now refreshes each per-entity event stream's TTL at most once per half of the configured `events.redis.store.entity-ttl`, instead of on every published event, to reduce `PEXPIRE` command load (and therefore CPU) on the events Redis. The refresh is rate limited with a fixed-size table allocated at startup, so memory does not grow with the number of entities. Event history retention is unchanged.
 
 ### Deprecated
 
diff --git a/cmd/internal/shared/config.go b/cmd/internal/shared/config.go
index 3241daeeb9..4f0ffc4b60 100644
--- a/cmd/internal/shared/config.go
+++ b/cmd/internal/shared/config.go
@@ -115,6 +115,7 @@ var DefaultEventsConfig = func() config.Events {
 	c.Redis.Store.EntityCount = 100
 	c.Redis.Store.CorrelationIDCount = 100
 	c.Redis.Store.StreamPartitionSize = 64
+	c.Redis.Store.ExpireLimiterSize = 1 << 22
 	c.Redis.Workers = 16
 	c.Redis.Publish.QueueSize = 8192
 	c.Redis.Publish.MaxWorkers = 1024
diff --git a/pkg/config/shared.go b/pkg/config/shared.go
index 3b5ae810bf..79103acd9b 100644
--- a/pkg/config/shared.go
+++ b/pkg/config/shared.go
@@ -138,8 +138,9 @@ type RedisEvents struct {
 		TTL                 time.Duration `name:"ttl" description:"How long event payloads are retained"`
 		EntityCount         int           `name:"entity-count" description:"How many events are indexed for a entity ID"`
 		EntityTTL           time.Duration `name:"entity-ttl" description:"How long events are indexed for a entity ID"`
-		CorrelationIDCount  int           `name:"correlation-id-count" description:"How many events are indexed for a correlation ID"`     //nolint:lll
-		StreamPartitionSize int           `name:"stream-partition-size" description:"How many streams to listen to in a single partition"` //nolint:lll
+		CorrelationIDCount  int           `name:"correlation-id-count" description:"How many events are indexed for a correlation ID"`                                                                                 //nolint:lll
+		StreamPartitionSize int           `name:"stream-partition-size" description:"How many streams to listen to in a single partition"`                                                                             //nolint:lll
+		ExpireLimiterSize   int           `name:"expire-limiter-size" description:"How many entity event streams the PEXPIRE rate limiter tracks, at 8 bytes each; a negative value refreshes the TTL on every event"` //nolint:lll
 	} `name:"store"`
 	Workers int `name:"workers"`
 	Publish struct {
diff --git a/pkg/events/redis/expire_limiter.go b/pkg/events/redis/expire_limiter.go
new file mode 100644
index 0000000000..c20cc80cf4
--- /dev/null
+++ b/pkg/events/redis/expire_limiter.go
@@ -0,0 +1,111 @@
+// Copyright © 2026 The Things Network Foundation, The Things Industries B.V.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package redis
+
+import (
+	"hash/maphash"
+	"math/bits"
+	"sync/atomic"
+	"time"
+)
+
+const (
+	// Slots when events.redis.store.expire-limiter-size is unset: 8 bytes each, so
+	// a flat 32 MiB, holding the ideal PEXPIRE rate to about a million streams.
+	defaultExpireLimiterSlots = 1 << 22
+	// Slots a key may occupy. Eight of 8 bytes are one cache line, so scanning a
+	// whole set is a single fetch.
+	expireLimiterWays = 8
+)
+
+// Slot layout: a non-zero hash tag, then the tick the key was last allowed at,
+// so an all-zero slot reads as unused. Ages are computed modulo the tick
+// counter, so it never runs out. A tick is a fraction of the interval, so 24
+// bits span a million intervals and the tag takes the rest: two keys sharing a
+// set and a tag are limited as one, stranding the quieter stream without a TTL.
+const (
+	expireTicksPerInterval = 16
+	expireTickBits         = 24
+	expireTickMask         = 1<<expireTickBits - 1
+)
+
+// expireLimiter holds PEXPIRE down to one command per key per interval, in a
+// fixed pointer-free table. A key may use any way of its set, so two busy keys
+// do not evict each other. Anything uncertain allows a PEXPIRE, never skips one.
+type expireLimiter struct {
+	slots    []atomic.Uint64
+	setMask  uint64
+	seed     maphash.Seed
+	tickUnit time.Duration
+	epoch    time.Time
+}
+
+// newExpireLimiter allows one PEXPIRE per key per interval over slots entries,
+// rounded up to whole sets. A non-positive size, or an interval too short to
+// divide into ticks, returns nil, which allows everything.
+func newExpireLimiter(slots int, interval time.Duration, epoch time.Time) *expireLimiter {
+	tickUnit := interval / expireTicksPerInterval
+	if slots <= 0 || tickUnit <= 0 {
+		return nil
+	}
+	sets := uint64((slots + expireLimiterWays - 1) / expireLimiterWays)
+	sets = 1 << bits.Len64(sets-1) // Round up to a power of two, so setMask works.
+	return &expireLimiter{
+		slots:    make([]atomic.Uint64, sets*expireLimiterWays),
+		setMask:  sets - 1,
+		seed:     maphash.MakeSeed(),
+		tickUnit: tickUnit,
+		epoch:    epoch,
+	}
+}
+
+// allow reports whether key needs a refresh at now, recording it only if so.
+func (l *expireLimiter) allow(key string, now time.Time) bool {
+	if l == nil {
+		return true
+	}
+	elapsed := now.Sub(l.epoch)
+	if elapsed < 0 {
+		return true
+	}
+	tick := uint64(int64(elapsed/l.tickUnit) & expireTickMask)
+	h := maphash.String(l.seed, key)
+	tag, base := h>>expireTickBits, (h&l.setMask)*expireLimiterWays
+	if tag == 0 {
+		tag = 1 // Zero is reserved for unused slots, so no key may claim it.
+	}
+
+	var target, oldestAge uint64
+	for way := range uint64(expireLimiterWays) {
+		slot := l.slots[base+way].Load()
+		last := slot & expireTickMask
+		age := uint64(expireTickMask) // An unused slot is evicted before any live one.
+		if slot != 0 {
+			age = (tick - last) & expireTickMask
+		}
+		if slot>>expireTickBits == tag {
+			if age < expireTicksPerInterval {
+				return false
+			}
+			target = way
+			break
+		}
+		if age > oldestAge {
+			target, oldestAge = way, age
+		}
+	}
+	l.slots[base+target].Store(tag<<expireTickBits | tick)
+	return true
+}
diff --git a/pkg/events/redis/expire_limiter_test.go b/pkg/events/redis/expire_limiter_test.go
new file mode 100644
index 0000000000..b1678a719e
--- /dev/null
+++ b/pkg/events/redis/expire_limiter_test.go
@@ -0,0 +1,272 @@
+// Copyright © 2026 The Things Network Foundation, The Things Industries B.V.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// The limiter is unexported, so its tests live in the package under test.
+package redis
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/smarty/assertions"
+	"go.thethings.network/lorawan-stack/v3/pkg/util/test/assertions/should"
+)
+
+const (
+	testExpireInterval = time.Hour
+	testExpireTick     = testExpireInterval / expireTicksPerInterval
+	expireWrapPeriod   = (expireTickMask + 1) * testExpireTick
+)
+
+var testExpireEpoch = time.Date(2026, time.September, 15, 12, 0, 0, 0, time.UTC)
+
+func BenchmarkExpireLimiter(b *testing.B) {
+	l := newExpireLimiter(defaultExpireLimiterSlots, testExpireInterval, testExpireEpoch)
+	keys := make([]string, 1024)
+	for i := range keys {
+		keys[i] = fmt.Sprintf("ttn:v3:event_stream:device:app.dev-%d", i)
+	}
+
+	for i := 0; b.Loop(); i++ {
+		l.allow(keys[i%len(keys)], testExpireEpoch)
+	}
+}
+
+// TestExpireLimiterAllocations is not parallel, because testing.AllocsPerRun
+// panics while a parallel test is running.
+func TestExpireLimiterAllocations(t *testing.T) { //nolint:paralleltest
+	a := assertions.New(t)
+	l := newExpireLimiter(1024, testExpireInterval, testExpireEpoch)
+
+	a.So(testing.AllocsPerRun(100, func() {
+		l.allow("stream", testExpireEpoch)
+	}), should.BeZeroValue)
+}
+
+// newTestExpireLimiter returns a limiter with a single set, so that every key
+// competes for the same expireLimiterWays slots.
+func newTestExpireLimiter() *expireLimiter {
+	return newExpireLimiter(expireLimiterWays, testExpireInterval, testExpireEpoch)
+}
+
+func TestExpireLimiter(t *testing.T) {
+	t.Parallel()
+
+	t.Run("AllowsOncePerInterval", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+		l := newTestExpireLimiter()
+
+		a.So(l.allow("stream", testExpireEpoch), should.BeTrue)
+		a.So(l.allow("stream", testExpireEpoch), should.BeFalse)
+		a.So(l.allow("stream", testExpireEpoch.Add(testExpireInterval-testExpireTick)), should.BeFalse)
+		a.So(l.allow("stream", testExpireEpoch.Add(testExpireInterval)), should.BeTrue)
+		a.So(l.allow("stream", testExpireEpoch.Add(testExpireInterval)), should.BeFalse)
+	})
+
+	t.Run("AllowsKeyPublishingWithinEveryInterval", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+		l := newTestExpireLimiter()
+
+		// A key publishing twice per interval must still be refreshed once per
+		// interval, or its stream expires in Redis.
+		allowed := 0
+		for i := range 20 {
+			if l.allow("stream", testExpireEpoch.Add(time.Duration(i)*testExpireInterval/2)) {
+				allowed++
+			}
+		}
+		a.So(allowed, should.Equal, 10)
+	})
+
+	t.Run("KeepsDistinctKeysApart", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+		l := newTestExpireLimiter()
+
+		for i := range expireLimiterWays {
+			a.So(l.allow(fmt.Sprintf("stream-%d", i), testExpireEpoch), should.BeTrue)
+		}
+		for i := range expireLimiterWays {
+			a.So(l.allow(fmt.Sprintf("stream-%d", i), testExpireEpoch), should.BeFalse)
+		}
+	})
+
+	t.Run("AllowsAfterEviction", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+		l := newTestExpireLimiter()
+
+		// Filling every way of the only set evicts the oldest key, which must then
+		// be allowed again rather than silently suppressed.
+		for i := range expireLimiterWays + 1 {
+			at := testExpireEpoch.Add(time.Duration(i) * testExpireTick)
+			a.So(l.allow(fmt.Sprintf("stream-%d", i), at), should.BeTrue)
+		}
+		a.So(l.allow("stream-0", testExpireEpoch.Add(expireLimiterWays*testExpireTick)), should.BeTrue)
+	})
+
+	t.Run("RateLimitsAcrossTickWrap", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+		l := newTestExpireLimiter()
+
+		// Ages are computed modulo the tick counter, so limiting must survive the
+		// counter wrapping rather than degrading to one refresh per event. Each
+		// period is added separately: several of them overflow a time.Duration.
+		at := testExpireEpoch
+		for range 2 {
+			at = at.Add(expireWrapPeriod)
+			just := at.Add(-testExpireTick) // Just before the counter wraps.
+			a.So(l.allow("wrap", just), should.BeTrue)
+			a.So(l.allow("wrap", just.Add(testExpireInterval-testExpireTick)), should.BeFalse)
+			a.So(l.allow("wrap", just.Add(testExpireInterval)), should.BeTrue)
+		}
+	})
+
+	t.Run("AllowsBeforeEpoch", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+		l := newTestExpireLimiter()
+
+		a.So(l.allow("stream", testExpireEpoch.Add(-testExpireTick)), should.BeTrue)
+		a.So(l.allow("stream", testExpireEpoch.Add(-testExpireTick)), should.BeTrue)
+	})
+
+	t.Run("AllowsWhenDisabled", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+
+		for _, l := range []*expireLimiter{
+			nil,
+			newExpireLimiter(-1, testExpireInterval, testExpireEpoch),
+			newExpireLimiter(0, testExpireInterval, testExpireEpoch),
+			newExpireLimiter(expireLimiterWays, 0, testExpireEpoch),
+		} {
+			a.So(l, should.BeNil)
+			a.So(l.allow("stream", testExpireEpoch), should.BeTrue)
+			a.So(l.allow("stream", testExpireEpoch), should.BeTrue)
+		}
+	})
+
+	t.Run("RefreshesWellInsideTheJitteredTTL", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+
+		// A tick rounds the next refresh up, so it must land inside publish's
+		// jittered PEXPIRE, and no sooner than the interval it holds.
+		for _, ttl := range []time.Duration{
+			time.Second,
+			2*time.Second + time.Millisecond,
+			time.Minute,
+			24 * time.Hour,
+		} {
+			l := newExpireLimiter(expireLimiterWays, ttl/2, testExpireEpoch)
+			a.So(l, should.NotBeNil)
+
+			// The epoch sits on a tick boundary, the worst phase to be refreshed at.
+			a.So(l.allow("stream", testExpireEpoch), should.BeTrue)
+			step, next := ttl/1000, ttl
+			for at := time.Duration(0); at < ttl; at += step {
+				if l.allow("stream", testExpireEpoch.Add(at)) {
+					next = at
+					break
+				}
+			}
+			a.So(next, should.BeGreaterThanOrEqualTo, ttl/2-step)
+			a.So(next, should.BeLessThan, time.Duration(float64(ttl)*(1-ttlJitter)))
+		}
+	})
+
+	t.Run("DisablesItselfBelowOneTick", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+
+		// A tick unit that rounds to zero disables the limiter rather than dividing
+		// by it.
+		const oneTick = expireTicksPerInterval * time.Nanosecond
+		a.So(newExpireLimiter(expireLimiterWays, oneTick-time.Nanosecond, testExpireEpoch), should.BeNil)
+		a.So(newExpireLimiter(expireLimiterWays, oneTick, testExpireEpoch), should.NotBeNil)
+	})
+
+	t.Run("ClampsSmallSizesToOneSet", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+
+		// A positive size below one set limits rather than silently disabling.
+		for slots := 1; slots < expireLimiterWays; slots++ {
+			l := newExpireLimiter(slots, testExpireInterval, testExpireEpoch)
+			a.So(len(l.slots), should.Equal, expireLimiterWays)
+			a.So(l.allow("stream", testExpireEpoch), should.BeTrue)
+			a.So(l.allow("stream", testExpireEpoch), should.BeFalse)
+		}
+	})
+
+	t.Run("RoundsTableUpToWholeSets", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+
+		for slots, want := range map[int]int{
+			expireLimiterWays:         1 * expireLimiterWays,
+			expireLimiterWays + 1:     2 * expireLimiterWays,
+			3 * expireLimiterWays:     4 * expireLimiterWays,
+			3*expireLimiterWays + 1:   4 * expireLimiterWays,
+			4 * expireLimiterWays:     4 * expireLimiterWays,
+			defaultExpireLimiterSlots: defaultExpireLimiterSlots,
+		} {
+			l := newExpireLimiter(slots, testExpireInterval, testExpireEpoch)
+			a.So(len(l.slots), should.Equal, want)
+		}
+	})
+
+	t.Run("KeepsTableSizeConstant", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+		l := newExpireLimiter(1024, testExpireInterval, testExpireEpoch)
+		size := len(l.slots)
+
+		for i := range 100_000 {
+			l.allow(fmt.Sprintf("stream-%d", i), testExpireEpoch)
+		}
+		a.So(len(l.slots), should.Equal, size)
+	})
+
+	t.Run("AllowsEveryKeyUnderConcurrency", func(t *testing.T) {
+		t.Parallel()
+		a := assertions.New(t)
+		l := newExpireLimiter(1024, testExpireInterval, testExpireEpoch)
+
+		const keys, publishers = 64, 8
+		allowed := make([]atomic.Int64, keys)
+		wg := sync.WaitGroup{}
+		for range publishers {
+			wg.Go(func() {
+				for i := range keys {
+					if l.allow(fmt.Sprintf("stream-%d", i), testExpireEpoch) {
+						allowed[i].Add(1)
+					}
+				}
+			})
+		}
+		wg.Wait()
+
+		for i := range keys {
+			a.So(allowed[i].Load(), should.BeGreaterThanOrEqualTo, int64(1))
+		}
+	})
+}
diff --git a/pkg/events/redis/redis.go b/pkg/events/redis/redis.go
index ae3146b5a7..ed43209a0f 100644
--- a/pkg/events/redis/redis.go
+++ b/pkg/events/redis/redis.go
@@ -111,6 +111,12 @@ func NewPubSub(
 		pss.streamPartitionSize = 64
 	}
 
+	expireLimiterSize := conf.Store.ExpireLimiterSize
+	if expireLimiterSize == 0 {
+		expireLimiterSize = defaultExpireLimiterSlots
+	}
+	pss.expireLimiter = newExpireLimiter(expireLimiterSize, pss.entityHistoryTTL/2, time.Now())
+
 	pss.publisher = events.PublishFunc(pss.publish)
 	if batchConf.Enable {
 		targetSize, delay := batchConf.TargetSize, batchConf.Delay
diff --git a/pkg/events/redis/store.go b/pkg/events/redis/store.go
index d55e064a86..ad60ecefce 100644
--- a/pkg/events/redis/store.go
+++ b/pkg/events/redis/store.go
@@ -51,6 +51,10 @@ type PubSubStore struct {
 	entityHistoryTTL          time.Duration
 	correlationIDHistoryCount int
 	streamPartitionSize       int
+
+	// expireLimiter keeps PEXPIRE on entity event streams down to one command per
+	// stream per entityHistoryTTL/2.
+	expireLimiter *expireLimiter
 }
 
 func (ps *PubSubStore) eventDataKey(_ context.Context, uid string) string {
@@ -532,7 +536,11 @@ func (ps *PubSubStore) publish(evs ...events.Event) {
 	}
 
 	entityHistoryTTL := random.Jitter(ps.entityHistoryTTL, ttlJitter)
+	now := time.Now()
 	for eventStream := range eventStreams {
+		if !ps.expireLimiter.allow(eventStream, now) {
+			continue
+		}
 		tx.PExpire(ps.ctx, eventStream, entityHistoryTTL)
 	}