feat: trigger urgent single-key compaction on ErrDeltaScanTruncated

adapter/redis.go

@@ -271,6 +271,11 @@ type RedisServer struct {
 	leaderClientsMu sync.RWMutex
 	leaderClients   map[string]*redis.Client
 
+	// compactor is the background DeltaCompactor for this node. When set,
+	// urgent compaction is triggered on ErrDeltaScanTruncated to unblock
+	// reads on hot keys faster than the regular compaction interval.
+	compactor *DeltaCompactor
+
 	route map[string]func(conn redcon.Conn, cmd redcon.Command)
 }
 
@@ -282,6 +287,14 @@ func WithRedisActiveTimestampTracker(tracker *kv.ActiveTimestampTracker) RedisSe
 	}
 }
 
+// WithRedisCompactor wires a DeltaCompactor to the RedisServer so that urgent
+// single-key compaction can be triggered when ErrDeltaScanTruncated is hit.
+func WithRedisCompactor(c *DeltaCompactor) RedisServerOption {
+	return func(r *RedisServer) {
+		r.compactor = c
+	}
+}
+
 // WithRedisRequestObserver enables Prometheus-compatible request metrics.
 func WithRedisRequestObserver(observer monitoring.RedisRequestObserver) RedisServerOption {
 	return func(r *RedisServer) {
@@ -467,6 +480,14 @@ func (r *RedisServer) pinReadTS(ts uint64) *kv.ActiveTimestampToken {
 	return r.readTracker.Pin(ts)
 }
 
+// triggerUrgentCompaction signals the DeltaCompactor to immediately compact
+// the given key, bypassing the regular interval. No-op when no compactor is wired.
+func (r *RedisServer) triggerUrgentCompaction(typeName string, key []byte) {
+	if r.compactor != nil {
+		r.compactor.TriggerUrgentCompaction(typeName, key)
+	}
+}
+
 func (r *RedisServer) dispatchCommand(conn redcon.Conn, name string, handler func(redcon.Conn, redcon.Command), cmd redcon.Command, start time.Time) {
 	switch {
 	case r.requestObserver != nil:
@@ -2682,7 +2703,8 @@ func (r *RedisServer) buildListPopElems(ctx context.Context, key []byte, meta st
 		claimEnd = meta.Tail
 	}
 	// Capacity: n claim keys + n Del(item) for found items + 1 for the delta key appended by caller.
-	elems := make([]*kv.Elem[kv.OP], 0, n+int64(len(kvps))+listPopDeltaOverhead)
+	// n is bounded by maxWideColumnItems (100_000) so the int conversion is safe.
+	elems := make([]*kv.Elem[kv.OP], 0, int(n)+len(kvps)+listPopDeltaOverhead)
 	for seq := claimStart; seq < claimEnd; seq++ {
 		elems = append(elems, &kv.Elem[kv.OP]{Op: kv.Put, Key: store.ListClaimKey(key, seq), Value: []byte{}})
 	}

adapter/redis_compat_helpers.go

@@ -742,6 +742,9 @@ func (r *RedisServer) resolveListMeta(ctx context.Context, key []byte, readTS ui
 		return d.LenDelta, errors.WithStack(unmarshalErr)
 	})
 	if err != nil {
+		if errors.Is(err, ErrDeltaScanTruncated) {
+			r.triggerUrgentCompaction("list", key)
+		}
 		return store.ListMeta{}, false, err
 	}
 	baseMeta.Len += lenSum
@@ -797,7 +800,7 @@ func (r *RedisServer) resolveCollectionLen(
 
 // resolveHashMeta aggregates the base hash metadata with all uncompacted Delta keys.
 func (r *RedisServer) resolveHashMeta(ctx context.Context, key []byte, readTS uint64) (int64, bool, error) {
-	return r.resolveCollectionLen(
+	n, exists, err := r.resolveCollectionLen(
 		ctx, key, readTS,
 		store.HashMetaKey(key),
 		store.HashMetaDeltaScanPrefix(key),
@@ -811,11 +814,15 @@ func (r *RedisServer) resolveHashMeta(ctx context.Context, key []byte, readTS ui
 		},
 		"resolveHashMeta: clamping negative Len to 0",
 	)
+	if errors.Is(err, ErrDeltaScanTruncated) {
+		r.triggerUrgentCompaction("hash", key)
+	}
+	return n, exists, err
 }
 
 // resolveSetMeta aggregates the base set metadata with all uncompacted Delta keys.
 func (r *RedisServer) resolveSetMeta(ctx context.Context, key []byte, readTS uint64) (int64, bool, error) {
-	return r.resolveCollectionLen(
+	n, exists, err := r.resolveCollectionLen(
 		ctx, key, readTS,
 		store.SetMetaKey(key),
 		store.SetMetaDeltaScanPrefix(key),
@@ -829,11 +836,15 @@ func (r *RedisServer) resolveSetMeta(ctx context.Context, key []byte, readTS uin
 		},
 		"resolveSetMeta: clamping negative Len to 0",
 	)
+	if errors.Is(err, ErrDeltaScanTruncated) {
+		r.triggerUrgentCompaction("set", key)
+	}
+	return n, exists, err
 }
 
 // resolveZSetMeta aggregates the base sorted set metadata with all uncompacted Delta keys.
 func (r *RedisServer) resolveZSetMeta(ctx context.Context, key []byte, readTS uint64) (int64, bool, error) {
-	return r.resolveCollectionLen(
+	n, exists, err := r.resolveCollectionLen(
 		ctx, key, readTS,
 		store.ZSetMetaKey(key),
 		store.ZSetMetaDeltaScanPrefix(key),
@@ -847,4 +858,8 @@ func (r *RedisServer) resolveZSetMeta(ctx context.Context, key []byte, readTS ui
 		},
 		"resolveZSetMeta: clamping negative Len to 0",
 	)
+	if errors.Is(err, ErrDeltaScanTruncated) {
+		r.triggerUrgentCompaction("zset", key)
+	}
+	return n, exists, err
 }