Rate-limit PEXPIRE on entity event streams

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
 

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

pkg/config/shared.go

@@ -1,3 +1,3 @@
 // Copyright © 2019 The Things Network Foundation, The Things Industries B.V.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
@@ -138,8 +138,9 @@
 		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 {

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
+}