Deduplicator

The Deduplicator removes duplicate events using a deterministic SHA-256 key and an atomic Redis Lua script. Runs after normalization, before feature engineering.

---

Deduplication Key

Composition: SHA-256(message | type | template_id | host | source) → first 32 hex chars

def compute_dedup_key(event: CanonicalEvent, template_id: str) -> str:
    parts = [
        event.message or "",
        event.type.value,
        template_id or "",
        event.host or "",
        event.source or "",
    ]
    key = "|".join(parts)
    return hashlib.sha256(key.encode()).hexdigest()[:32]

Why these fields? - Same message + same template + same host + same source = duplicate - Different host/source = different event (even if message identical)

---

Redis Lua Script (Atomic)

-- dedup.lua
-- KEYS[1] = dedup:{dedup_key}
-- ARGV[1] = window_seconds (300)

local key = KEYS[1]
local window = tonumber(ARGV[1])

local count = redis.call('INCR', key)
if count == 1 then
    redis.call('EXPIRE', key, window)
    return {0, 1}  -- not duplicate, count=1
else
    local ttl = redis.call('TTL', key)
    if ttl < 0 then
        redis.call('EXPIRE', key, window)
    end
    return {1, count}  -- duplicate, current count
end

Execution:

# Python wrapper
async def check_and_increment(dedup_key: str, window: int = 300) -> tuple[bool, int]:
    result = await redis.eval(
        LUA_SCRIPT,
        1,
        f"dedup:{dedup_key}",
        str(window)
    )
    is_duplicate = bool(result[0])
    count = int(result[1])
    return is_duplicate, count

Properties: - Atomic: INCR + EXPIRE in single Redis call - Race-free: Multiple pipeline replicas safe - Low latency: < 1 ms per call - Sliding window: TTL resets on each hit

---

Pipeline Integration

flowchart LR
  NORM[Normalized Event] --> KEY[Compute dedup_key]
  KEY --> LUA[Redis EVAL<br/>INCR + EXPIRE]
  LUA -->|is_duplicate=0| UNIQUE[Unique Event]
  LUA -->|is_duplicate=1| DUP[Duplicate → Skip]
  UNIQUE --> FEAT[Feature Engineering]
  DUP -.-> STATS[duplicate_total++]

---

Window Configuration

Parameter	Default	Description
DEDUP_WINDOW_SECONDS	300	Sliding window (5 minutes)
Redis Key Format	dedup:{sha256}	Auto-expires after window

Tuning: - Shorter window (60s) → More events stored, less dedup - Longer window (3600s) → Better dedup, more Redis memory

---

PostgreSQL Backstop

Even if Redis dedup misses (race, restart, clock skew), PostgreSQL enforces uniqueness:

-- Partial unique index (migration 0015)
CREATE UNIQUE INDEX ux_events_dedup_key 
ON events (dedup_key) 
WHERE dedup_key IS NOT NULL;

Storage layer (storage.py) also does intra-batch merge:

# Merge events with same dedup_key within batch
merged = {}
for event in batch:
    key = event.dedup_key
    if key in merged:
        merged[key].occurrence_count += event.occurrence_count
    else:
        merged[key] = event

---

Performance

Metric	Value
Latency (Redis)	< 1 ms
Throughput	~2,100 checks/sec
Memory per key	~72 bytes (key + count + TTL)
Keys at 2K eps × 300s	~600K keys (~43 MB)

---

Monitoring

# Redis dedup keys
docker compose exec redis redis-cli --scan --pattern "dedup:*" | wc -l

# Key TTL sample
docker compose exec redis redis-cli --scan --pattern "dedup:*" | head -5 | xargs -I{} redis-cli TTL {}

# Pipeline stats
docker compose logs consumer-pipeline | grep duplicate_total

---

Troubleshooting

Symptom	Cause	Fix
Low dedup rate	Window too short	Increase DEDUP_WINDOW_SECONDS
High memory	Window too long / high cardinality	Decrease window, check for unmasked PII in messages
Duplicates in DB	Redis restarted / clock skew	PG unique index catches; check clock sync
Redis OOM	Too many keys	Reduce window, enable maxmemory-policy allkeys-lru

# Dedup rate from API
curl -s http://localhost:8000/api/kpis | jq .dedup_rate_pct
# Target: 50-80% depending on source repetitiveness