Concurrent Human Approval Queue With Real Race-Condition Safety
Two real reviewers act on the same item at nearly the same moment. Without a guard, both silently believe they succeeded — and one's decision is silently lost. This project measures that failure directly, then fixes it.
Problem Statement
A content moderation queue uses Module 8's interrupt() mechanism to pause each item for real human review. In a genuine production deployment with multiple real reviewers working simultaneously, two people can open the same pending item at nearly the same moment and both submit a decision before either sees the other's action — a real, classic race condition, not a hypothetical one. This project builds this exact scenario using real, concurrent Python threads, measures directly that both reviewers' actions silently appear to succeed while one decision is actually lost, then builds and verifies the real fix: an atomic guard ensuring exactly one decision wins and the other reviewer is correctly, honestly informed. It also directly measures a second, related failure: whether one slow or unresponsive reviewer can block the entire queue, and confirms the real, architectural fix already available in this course's own patterns.
Dataset
Simulated Concurrent Review Queue
A small, real queue of content items pending human approval, each represented by its own genuine LangGraph thread using Module 8's interrupt() mechanism, tested under real, concurrent access using Python's threading module — not simulated or mocked concurrency, genuine simultaneous execution.
Architecture Decisions
This project deliberately uses Python's real threading module to create genuine, simultaneous execution rather than simulating concurrency sequentially, since a race condition is fundamentally a timing bug that only manifests under real, overlapping execution. The fix uses a real threading.Lock guarding the entire check-then-act sequence atomically, the standard, correct primitive for exactly this class of problem. The queue-blocking test reuses this course's own proven pattern directly: giving each queue item its own, independent thread_id, exactly as Module 5 established, confirming this existing architectural choice already solves the second failure mode without any additional code.
Built On
- •Module 8 — interrupt() and Human Approval Gates
- •Module 5 — Checkpointers and thread_id Persistence
- •Module 14 — Parallel Fan-Out and Result Aggregation
- •Module 16 — Node-Level Error Handling, Timeouts, and Recovery
Measuring the Real Race Condition With Genuine Concurrent Threads
A shared, in-memory record represents one real queue item's current status — exactly the shape of shared state a real production system's approval queue would hold. Two real, separate threads represent two real reviewers, Alice and Bob, each independently checking the item's current status, then, after a small, deliberate delay representing real processing time, acting on what they saw. Because both threads check the status while it is still 'pending' before either one has written its decision, both proceed to act — and Bob's write silently overwrites Alice's, even though Alice's own action log entry incorrectly reports success. This is run with Python's real threading module, producing genuine, non-simulated concurrent execution.
1import threading
2import time
3import warnings
4warnings.filterwarnings("ignore")
5
6print("=== A REAL, SHARED QUEUE ITEM, ACTED ON BY TWO GENUINE, CONCURRENT THREADS ===\n")
7
8queue_item = {"id": "content-42", "status": "pending", "approved_by": None}
9action_log = []
10
11def approver_acts(approver_name: str, decision: str):
12 """A REALISTIC check-then-act sequence -- exactly what a real
13 reviewer's action would look like: read the current status, then
14 (after some real processing time) act on what was read."""
15 current_status = queue_item["status"]
16 time.sleep(0.05) # the real, dangerous gap between CHECK and ACT
17 if current_status == "pending":
18 queue_item["status"] = decision
19 queue_item["approved_by"] = approver_name
20 action_log.append(approver_name + " successfully set status to " + decision)
21 else:
22 action_log.append(approver_name + " saw status was already " + current_status + ", took no action")
23
24print("=== RUNNING TWO REAL REVIEWERS, GENUINELY CONCURRENTLY ===\n")
25
26thread_alice = threading.Thread(target=approver_acts, args=("Alice", "approved"))
27thread_bob = threading.Thread(target=approver_acts, args=("Bob", "rejected"))
28
29thread_alice.start()
30thread_bob.start()
31thread_alice.join()
32thread_bob.join()
33
34print("Action log:")
35for entry in action_log:
36 print(" " + entry)
37
38print("\nFinal item state: " + str(queue_item) + "\n")
39
40both_claim_success = sum(1 for e in action_log if "successfully" in e) == 2
41
42print(f"""
43=== THE RESULT, CONFIRMED DIRECTLY ===
44
45Both reviewers' action log entries claim success: {both_claim_success}
46
47Report your own observed output above. This is the real, dangerous
48part of this bug: BOTH Alice's and Bob's log entries say "successfully
49set status" -- but the final, actual item state only reflects ONE of
50their decisions, since the second write silently overwrote the first.
51Alice's system may have shown her a success message for a decision
52that was, moments later, silently erased with no notification to her
53at all.
54
55This is a genuine, classic race condition: a check-then-act sequence
56with no protection against another thread acting in between the
57check and the act.
58""")Gotchas
- ⚠The 0.05-second sleep() between the check and the act is deliberately inserted to make this race condition reliably observable in a short script — in a real production system, this same dangerous gap exists naturally, caused by real network latency, real database read/write timing, or real processing time between a reviewer opening an item and submitting their decision.
- ⚠This bug is genuinely dangerous specifically because it fails SILENTLY — neither Alice's nor Bob's system shows an error; both appear to succeed. A real production incident caused by this exact bug class would likely go unnoticed until someone manually audits a decision and finds it doesn't match what a reviewer remembers submitting.
- ⚠This exact check-then-act race condition pattern is a well-known, general concurrency bug class, not specific to LangGraph or approval queues — it applies anywhere multiple real, concurrent actors can read shared state, then act on what they read, without atomic protection around the whole sequence.
The Real Fix: An Atomic Guard Around the Entire Sequence
The fix wraps the entire check-then-act sequence in a real Python threading.Lock, making it atomic — no other thread can execute any part of this sequence while one thread already holds the lock. Running the identical two-reviewer scenario with this real lock in place confirms directly: exactly one reviewer's action succeeds, and the other reviewer's action log now correctly, honestly reports that the item was already handled, rather than falsely claiming its own success.
1import threading
2import time
3import warnings
4warnings.filterwarnings("ignore")
5
6print("=== THE SAME SCENARIO, NOW WITH A REAL LOCK GUARDING THE FULL SEQUENCE ===\n")
7
8queue_item = {"id": "content-42", "status": "pending", "approved_by": None}
9action_log = []
10
11# THE REAL FIX: a genuine lock, guarding the ENTIRE check-then-act
12# sequence as one atomic unit -- no other thread can interleave
13item_lock = threading.Lock()
14
15def approver_acts_safely(approver_name: str, decision: str):
16 with item_lock:
17 current_status = queue_item["status"]
18 time.sleep(0.05) # the identical real delay -- but now safely
19 # protected, since no other thread can act
20 # while this one holds the lock
21 if current_status == "pending":
22 queue_item["status"] = decision
23 queue_item["approved_by"] = approver_name
24 action_log.append(approver_name + " successfully set status to " + decision)
25 else:
26 action_log.append(
27 approver_name + " correctly saw status was already " + current_status + ", took no action"
28 )
29
30print("=== RUNNING THE IDENTICAL TWO-REVIEWER SCENARIO ===\n")
31
32thread_alice = threading.Thread(target=approver_acts_safely, args=("Alice", "approved"))
33thread_bob = threading.Thread(target=approver_acts_safely, args=("Bob", "rejected"))
34
35thread_alice.start()
36thread_bob.start()
37thread_alice.join()
38thread_bob.join()
39
40print("Action log:")
41for entry in action_log:
42 print(" " + entry)
43
44print("\nFinal item state: " + str(queue_item) + "\n")
45
46exactly_one_success = sum(1 for e in action_log if "successfully" in e) == 1
47
48print(f"""
49=== THE RESULT, CONFIRMED DIRECTLY ===
50
51Exactly one reviewer's action succeeded: {exactly_one_success}
52
53This time, the losing reviewer's log entry HONESTLY reports the item
54was already handled, rather than falsely claiming success. A real
55system built this way could correctly notify the losing reviewer
56directly -- "this item was already reviewed by Alice" -- instead of
57silently discarding their decision with no explanation at all.
58
59This is the real, correct fix: not preventing two reviewers from
60looking at the same item (that's often genuinely fine), but ensuring
61only ONE of their decisions can ever actually take effect, and both
62reviewers get an honest, accurate report of what actually happened.
63""")Gotchas
- ⚠threading.Lock() is Python's real, standard primitive for exactly this problem — a real production system might instead use a database-level lock, a real atomic compare-and-swap operation, or a distributed lock service, depending on where the actual shared state lives, but the underlying principle (atomic protection around the full check-then-act sequence) is identical.
- ⚠The fix does not prevent both reviewers from opening and viewing the same item — that's often genuinely fine and even useful. It specifically prevents both of their DECISIONS from silently taking effect, ensuring only one decision wins and both reviewers receive an honest, accurate outcome.
- ⚠This same real lock-based pattern applies directly to any LangGraph node resuming a real interrupt() from multiple possible sources — if two separate systems might both attempt to call Command(resume=...) on the same thread_id, the same atomic-guard principle protects against exactly this class of silent, double-processed decision.
Confirming a Slow Reviewer Never Blocks the Rest of the Queue
A second, related real failure worth checking directly: does one slow or unresponsive reviewer block progress on every other item in the queue? This project builds three real queue items, each using Module 8's interrupt() mechanism with its own independent thread_id, following exactly the pattern Module 5 established. One item is deliberately left unresumed — simulating a reviewer who is away, slow, or unavailable — while the other two are resumed directly. This confirms, by directly inspecting each item's real state, that the two resumed items complete correctly and independently, while the deliberately-unresumed item remains genuinely, safely paused on its own, never blocking or being blocked by the other two.
1from typing_extensions import TypedDict
2from langgraph.graph import StateGraph, START, END
3from langgraph.checkpoint.memory import InMemorySaver
4from langgraph.types import interrupt, Command
5import warnings
6warnings.filterwarnings("ignore")
7
8print("=== CONFIRMING A SLOW REVIEWER NEVER BLOCKS THE REST OF THE QUEUE ===\n")
9
10class ReviewState(TypedDict):
11 item_id: str
12 decision: str
13
14def request_review(state: ReviewState) -> dict:
15 decision = interrupt({"item_id": state["item_id"], "message": "Approve this item?"})
16 return {"decision": decision}
17
18graph_builder = StateGraph(ReviewState)
19graph_builder.add_node("request_review", request_review)
20graph_builder.add_edge(START, "request_review")
21graph_builder.add_edge("request_review", END)
22
23checkpointer = InMemorySaver()
24compiled_graph = graph_builder.compile(checkpointer=checkpointer)
25
26# Each queue item gets its OWN, independent thread_id -- Module 5's
27# exact pattern, confirmed here directly to solve this second problem
28items = ["item-A", "item-B", "item-C"]
29configs = {item: {"configurable": {"thread_id": item}} for item in items}
30
31print("=== STARTING ALL THREE ITEMS -- EACH PAUSES INDEPENDENTLY ===\n")
32
33for item in items:
34 result = compiled_graph.invoke({"item_id": item, "decision": ""}, config=configs[item])
35 paused = "__interrupt__" in result
36 print(item + " genuinely paused: " + str(paused))
37
38print("\n=== item-B IS DELIBERATELY LEFT UNRESUMED (the slow, unavailable reviewer) ===\n")
39print("=== RESUMING item-A AND item-C DIRECTLY, CONFIRMING THEY PROCEED NORMALLY ===\n")
40
41result_a = compiled_graph.invoke(Command(resume="approved"), config=configs["item-A"])
42print("item-A final decision: " + result_a["decision"])
43
44result_c = compiled_graph.invoke(Command(resume="rejected"), config=configs["item-C"])
45print("item-C final decision: " + result_c["decision"])
46
47state_b = compiled_graph.get_state(configs["item-B"])
48
49print(f"""
50
51=== THE RESULT, CONFIRMED DIRECTLY ===
52
53item-B's real state, still genuinely paused: {state_b.next}
54
55item-A and item-C were both resumed and completed correctly and
56independently, with ZERO interference from item-B being left
57completely unresumed. item-B itself remains genuinely, safely
58paused, exactly where it was left -- not blocking anything, and not
59being affected by the other two items completing around it.
60
61=== WHY THIS WORKS: EACH ITEM HAS ITS OWN, TRULY INDEPENDENT THREAD ===
62
63This is not a new mechanism built specifically for this project --
64it is Module 5's exact thread_id isolation pattern, confirmed here
65directly to already solve this second, related real failure mode.
66A queue architecture giving every item its own independent thread
67structurally cannot have one slow item block another, since there is
68no shared execution path between them at all.
69
70=== PROJECT COMPLETE ===
71
72Section 1 measured a real, genuine race condition using actual
73concurrent threads, confirmed directly through real execution.
74Section 2 built and verified the real fix -- an atomic lock ensuring
75exactly one decision wins, with an honest outcome for both reviewers.
76Section 3 confirmed a second, related failure mode -- one slow
77reviewer blocking the queue -- is already solved by this course's
78own established thread_id isolation pattern, requiring no new code
79at all, only correct architectural awareness of why it already works.
80""")Gotchas
- ⚠This test deliberately never resumes item-B at all, confirming directly that a genuinely, indefinitely paused item causes no real problem for the rest of the system — the real risk this section rules out is architectural coupling between items, not simply a review taking a long time.
- ⚠This section's real finding is genuinely reassuring: this course's standard, established pattern for building any per-item pausable workflow (a unique thread_id per real, independent unit of work) already provides this safety property, with no additional design needed specifically for a queue use case.