Implement DelayQueue with Idempotent Task Execution
Company: Pinterest
Role: Data Scientist
Category: Coding & Algorithms
Difficulty: Medium
Interview Round: Onsite
##### Scenario
Message broker offers DelayQueue where tasks execute at future timestamps, ensuring idempotency on duplicate IDs.
##### Question
Implement a delay queue supporting schedule(id, run_at, task) and poll(now). Follow-up: when two tasks share the same ID but different run_at times, guarantee that exactly one executes.
##### Hints
Min-heap ordered by run_at + hash set of executed IDs; atomic check-and-set before execution.
Quick Answer: This question evaluates implementation skills in time-based task scheduling, idempotent execution semantics, and concurrency control, emphasizing data-structure selection and synchronization trade-offs.
Design a delay queue that schedules tasks to execute not earlier than their run_at time and guarantees idempotent execution by task ID. Implement a function process_delay_queue that processes a sequence of operations and returns the results of poll operations. Each operation is a dictionary: (1) {"op":"schedule","id":string,"run_at":int,"task":string} enqueues a task with a unique ID and the time it becomes eligible; multiple tasks may share the same ID. (2) {"op":"poll","now":int} executes all due tasks whose run_at <= now and returns a list of [id, task] pairs executed during this poll. Each ID must be executed at most once overall; if multiple tasks share the same ID, exactly one executes (the first one popped when due), and the rest are discarded when they are popped. Execution order within a poll must be ascending by run_at, breaking ties by schedule insertion order.
Constraints
- 1 <= len(operations) <= 200000
- run_at and now are integers in the range [0, 10^12]
- id and task are non-empty strings of length <= 64
- Within a single poll, tasks execute in ascending run_at; ties broken by schedule order
- Each ID executes at most once across all polls; subsequent tasks with the same ID are skipped when popped
- Use O(n) additional space; aim for O(log n) schedule and pop operations
Solution
import heapq
def process_delay_queue(operations: list) -> list:
heap = [] # (run_at, seq, id, task)
executed = set()
results = []
seq = 0
for op in operations:
t = op.get("op")
if t == "schedule":
run_at = op["run_at"]
tid = op["id"]
task = op["task"]
heapq.heappush(heap, (run_at, seq, tid, task))
seq += 1
elif t == "poll":
now = op["now"]
out = []
while heap and heap[0][0] <= now:
run_at, s, tid, task = heapq.heappop(heap)
if tid in executed:
continue
executed.add(tid)
out.append([tid, task])
results.append(out)
else:
raise ValueError("Unknown operation: " + str(t))
return results
Explanation
Use a min-heap prioritized by run_at to efficiently fetch due tasks. A monotonically increasing sequence number ensures deterministic ordering among tasks with equal run_at based on their scheduling order. Maintain a set of executed IDs. During each poll(now), repeatedly pop from the heap while the smallest run_at is <= now; if the task's ID has not executed, mark it executed and include it in the output; otherwise skip it. This guarantees at-most-once execution per ID even if multiple tasks with the same ID are scheduled at different times.
Time complexity: Each schedule is O(log n). Each poll pops k tasks in O(k log n). Across all operations, total time is O((S + P) log S), where S is schedules and P is total popped tasks.. Space complexity: O(S) for the heap and executed-ID set, where S is the number of scheduled tasks..
Hints
- Maintain a min-heap keyed by (run_at, sequence_number) to fetch the next due task and break ties by insertion order.
- Use a hash set to record executed IDs; skip any popped task whose ID is already in the set.
- On poll(now), pop while heap top run_at <= now; for each popped task, atomically check-and-add the ID to the executed set before executing.
- Return tasks executed in the order they are processed during the poll.