Detect Recent Duplicate Records with Bounded Memory

Problem Statement

In financial news distribution systems, incoming news flashes and press releases can be duplicated across multiple feeds. You are tasked with designing a streaming duplicate detector that tracks incoming records and matches duplicates against the most recent occurrence within an inclusive time window window_size.

Each record contains a unique integer identifier record_id, a non-decreasing integer timestamp, a string title, and a string text. A record is defined as a duplicate of an earlier record if both records have identical (title, text) pairs and current_timestamp - previous_timestamp <= window_size.

Because the stream is unbounded, your system must maintain bounded memory: any content keys whose most recent occurrences have aged past window_size must be evicted so that memory usage is proportional to the number of active records within the sliding window, rather than the lifetime of the stream.

Implement the DuplicateDetector class:

class DuplicateDetector:
    def __init__(self, window_size: int):
        """
        Initialize the detector with the maximum age (inclusive) for duplicates.
        """
        pass

    def process_record(self, record_id: int, timestamp: int, title: str, text: str) -> int | None:
        """
        Process an incoming record.
        If this record matches a previous record within [timestamp - window_size, timestamp],
        return the record_id of the most recent matching record.
        Otherwise, return None.
        """
        pass

Constraints

• window_size >= 0
• 0 = timestamp_prev).
• 1 <= len(title), len(text) <= 500
• Total calls to process_record up to 2 * 10^5.
• At any given time, memory consumption must be O(K), where K is the number of records received within the last window_size time units.

Examples

Example 1

detector = DuplicateDetector(window_size=10)

# t=0: First record, no previous record exists
detector.process_record(1, 0, "Earnings", "Company A beats EPS")
# Output: None

# t=5: Identical content within window (5 - 0 <= 10)
detector.process_record(2, 5, "Earnings", "Company A beats EPS")
# Output: 1 (matched against record 1)

# t=12: Same content again. Window covers [2, 12]. Record 2 (t=5) is valid.
detector.process_record(3, 12, "Earnings", "Company A beats EPS")
# Output: 2 (matched against the most recent duplicate, record 2)

# t=25: Same content. Window covers [15, 25]. Record 3 was at t=12 (expired).
detector.process_record(4, 25, "Earnings", "Company A beats EPS")
# Output: None (all previous duplicates expired)

Example 2

detector = DuplicateDetector(window_size=0)

detector.process_record(10, 100, "Alert", "Market Open")
# Output: None

# Duplicate at the exact same timestamp
detector.process_record(11, 100, "Alert", "Market Open")
# Output: 10

# Duplicate at a later timestamp when window_size is 0
detector.process_record(12, 101, "Alert", "Market Open")
# Output: None (101 - 100 > 0)

Edge Cases to Consider

• window_size = 0: Only matches records occurring at the exact same timestamp.
• Multiple identical records arriving in rapid succession: Always match against the immediately preceding (most recent) record.
• Large time gaps between consecutive records: Proper bulk eviction must occur even if many window periods have elapsed without input.