Implement a Concurrent Payment Queue with Visibility Timeouts

Problem Statement

In high-throughput payment architectures, message queues must reliably handle processing failures, network latency, and worker crashes without double-charging or dropping transactions.

Implement an in-memory, thread-safe payment queue class, ConcurrentPaymentQueue, that handles message delivery with a visibility timeout, receipt-handle validation, and idempotent submissions.

Class Interface

class ConcurrentPaymentQueue:
    def __init__(self, visibility_timeout: float):
        """
        Initializes the queue with a visibility timeout in seconds.
        """
        pass

    def submit(self, payment_id: str, amount: float) -> bool:
        """
        Enqueues a payment. If a payment with the same payment_id has already
        been submitted and is currently pending, in-flight, or completed,
        the method must reject the duplicate and return False.
        Returns True if the payment was successfully accepted.
        """
        pass

    def receive(self) -> tuple[str, float, str] | None:
        """
        Atomically retrieves the next available payment message.
        Returns a tuple of (payment_id, amount, receipt_handle) if available,
        or None if no messages are currently ready to be processed.
        
        The message becomes invisible to subsequent receive() calls until
        visibility_timeout seconds have elapsed.
        """
        pass

    def ack(self, payment_id: str, receipt_handle: str) -> bool:
        """
        Acknowledges successful processing of a payment.
        Returns True if the ack succeeds and the message is permanently marked as completed.
        Returns False if the receipt_handle is invalid, expired, or superseded by a newer lease.
        """
        pass

Requirements and Semantics

1. Visibility Timeout: When a consumer calls receive(), the message becomes invisible to other consumers for visibility_timeout seconds. If visibility_timeout passes without a valid ack(), the message becomes available for receive() again.
2. Stale ACK Protection: Every receive() call on an unacknowledged message issues a distinct, unique receipt_handle (lease token). If consumer A's visibility timeout expires and consumer B receives the message with a new receipt_handle, any late ack() from consumer A using the old handle must be rejected (return False).
3. Idempotency: A payment ID cannot be re-enqueued if it is currently queued, currently in-flight, or has already been successfully ACKed.
4. Thread Safety: All methods must be safe to call concurrently from multiple threads without data races or deadlocks.

Constraints

• visibility_timeout > 0
• amount > 0
• Up to 100,000 total operations across multiple concurrent worker threads.
• You may use standard library concurrency primitives (e.g., threading.Lock, threading.Condition).

Example

q = ConcurrentPaymentQueue(visibility_timeout=2.0)

assert q.submit("pay_1", 100.0) == True
assert q.submit("pay_1", 100.0) == False  # Duplicate ID rejected

msg = q.receive()  # Returns ("pay_1", 100.0, "token_v1")
assert msg[0] == "pay_1"

# Queue is now empty (message is in-flight)
assert q.receive() is None

# Simulate timeout
time.sleep(2.1)

# Message is visible again after timeout
msg2 = q.receive()  # Returns ("pay_1", 100.0, "token_v2")
assert msg2 is not None

# Consumer 1 attempts late ACK with old token
assert q.ack("pay_1", msg[2]) == False

# Consumer 2 ACKs with valid token
assert q.ack("pay_1", msg2[2]) == True

# Subsequent submit or ack for pay_1 fails
assert q.submit("pay_1", 50.0) == False
assert q.ack("pay_1", msg2[2]) == False