Implement DelayQueue with Idempotent Task Execution

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.

Idempotent Delay Queue with Min-Heap

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

Examples

Example 1

Input: operations = [
  {"op":"schedule","id":"a","run_at":5,"task":"A1"},
  {"op":"schedule","id":"b","run_at":3,"task":"B1"},
  {"op":"poll","now":2},
  {"op":"poll","now":4},
  {"op":"schedule","id":"a","run_at":6,"task":"A2"},
  {"op":"poll","now":10}
]
Output: [[], [["b","B1"]], [["a","A1"]]]

Explanation: Poll at 2 executes nothing. Poll at 4 executes b@3. Two a-tasks exist (run_at 5 and 6); only the first popped (A1) executes at poll 10; A2 is skipped due to idempotency.

Example 2

Input: operations = [
  {"op":"schedule","id":"alpha","run_at":10,"task":"A10"},
  {"op":"schedule","id":"alpha","run_at":5,"task":"A5"},
  {"op":"schedule","id":"beta","run_at":5,"task":"B5"},
  {"op":"poll","now":5},
  {"op":"poll","now":20}
]
Output: [[["alpha","A5"],["beta","B5"]], []]

Explanation: At now=5, alpha@5 is executed before beta@5 due to earlier schedule order among equal run_at; alpha@10 is later discarded since alpha already executed.

Hints

1. Maintain a min-heap keyed by (run_at, sequence_number) to fetch the next due task and break ties by insertion order.
2. Use a hash set to record executed IDs; skip any popped task whose ID is already in the set.
3. 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.
4. Return tasks executed in the order they are processed during the poll.

Approach

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.