如何在 Django 中防止竞态条件

Let's say you have enough credit left to generate one more image in an AI app. You submit a request in one browser tab, then submit another in a second tab before the first finishes. The app accepts both. Behind the scenes, each request passed the credit check. But the application accepted more work than your balance could pay for. What you just experienced is called a race condition. For a developer, this raises two questions: How can both requests pass the check when there's only enough credit for one? How do you prevent them from spending the same credit? In this guide, we’ll build a small Django credit system to explore those questions. We’ll reproduce the bug, fix it with database transactions and row locks, and test what happens when two requests compete for the last credit. What We’ll Cover Concurrency, race conditions, and the credit check How to set up the Django project How to reproduce the race condition How to protect the balance How to accept and test image requests Common mistakes and next steps Who This Guide Is For This guide is for developers who understand basic Django but are new to concurrency. Familiarity with models, migrations, and views will help you follow the examples. You’ll need Python 3.12 and Docker with Compose for the setup shown here. We’ll use Django 5.2, Django REST Framework 3.16, and PostgreSQL 17 because SQLite doesn't implement the row lock we’ll use. I'll explain the concurrency concepts before we apply them to the code. Image generation will be simulated throughout the tutorial. Our API will accept a prompt and return a simulated result. You won't need an AI provider account or a paid API key. This keeps the exercise focused on the credit decision and its database changes. You can follow the entire example without paying for an image request. What Concurrency and Race Conditions Mean What Is Concurrency? Concurrency refers to when two or more tasks make progress during overlapping periods of time. For example, a server can start processing Request B while Request A waits for a database response. Their instructions don't have to execute at the same instant. An event loop can switch between tasks while one waits, which is one way Python supports concurrent work. The important detail is that another operation can make progress before the first one finishes. What Is a Race Condition? A race condition is a flaw where the correctness of a result depends on the timing or order of concurrent operations. It can occur when operations share data, and the application doesn't coordinate their access adequately. Each operation may appear correct on its own, yet one can act on information another has already changed. A different execution order can then produce a different, incorrect outcome. Concurrency creates the opportunity for overlap, and a race condition is a bug that can arise from how the application handles that overlap. Where Else Can Race Conditions Happen? Race conditions can affect counters, user accounts, background jobs, and the results displayed in a browser. The participants can be two requests from one person, two different users, or automated tasks with no user action at all. The shared resource can be a database row, a file, an in-memory value, or the current state of a page. The examples below illustrate several ways the order can matter. Look for operations that share state and can interfere before either finishes. A View Counter Loses an Update Imagine two requests try to increase a post’s view count from 100. Both read 100 before either saves a change. Each adds one and writes 101. The counter should have reached 102, so one update disappears. This is a lost update, and no purchase or limited stock is involved. Two Signups Claim the Same Username A registration form can race if it relies only on a preliminary username check. Two requests check the same name, and both find it available. Each then attempts to create an account with that name. Without a database uniqueness rule, the application may create duplicate usernames. The protection here includes enforcing uniqueness in the database rather than trusting the earlier check. Two Workers Pick the Same Job Background workers can accidentally process the same pending job. Both workers read its status before either claims it. Each decides the job is available and begins the work. The application may then send a notification twice or generate the same report twice. A job-claim mechanism needs to coordinate ownership before the work begins. An Older Search Response Replaces a Newer One A browser can display the wrong search results because responses arrive out of order. You type “Django”, then change the search to “Django transactions” before the first response returns. The second response arrives first and displays the results you now want. If the first response arrives later and replaces them without a check, the page shows results for the old query. A request identifier or stale-response check addresses this case, so a database row lock wouldn't be the relevant fix. How the Credit Check Can Fail First, let’s define the rule for our example from the beginning of this tutorial: each image request costs one credit. Credits represent the app’s usage allowance. They're separate from the tokens a model processes. This is a rule we’re choosing for this tutorial. If an account starts with ten credits, nine accepted requests leave enough for one more. To accept a request, the backend must: Read the account’s balance. Check whether at least one credit remains. Deduct a credit and save the balance. Record the accepted request. When you test one request at a time, this process can appear correct. The first request saves a balance of zero. The next reads zero and stops. The next step is to examine the same operations when their execution overlaps. What Happens When Two Requests Overlap Suppose Request A reads the account and finds one credit. Before it updates the database, Request B reads the same account. It also finds one credit. Each request now has its own copy of the balance. Both pass the check, and both calculate 1 - 1 = 0. Request A saves zero and records a generation. Request B then saves zero and records another generation. The final balance is zero, but the application accepted two requests. A check for negative balances would miss this particular failure. The mistake is trusting a value after another request has had an opportunity to change it. To see this in practice, we’ll first build the version with that mistake. How to Set Up the Django Project Create a project directory and a virtual environment. The activation command below is for Linux and macOS: mkdir django-credit-demo cd django-credit-demo python3.12 -m venv .venv source .venv/bin/activate These commands give our example its own directory and Python environment. mkdir creates the directory, and cd moves you into it. The venv command creates an isolated environment named .venv. The source command activates it so subsequent package installations belong to this project. Keep this environment active while you run the commands below. On Windows, use py -3.12 -m venv .venv to create the environment and .venv\Scripts\Activate.ps1 to activate it in PowerShell. Create requirements.txt: Django>=5.2,=3.16,=3.2,= 1: account_a.balance -= 1 account_a.save(update_fields=["balance"]) Generation.objects.create(account=account_a, prompt="A garden") The first shell now checks and spends its copy of the balance. The if condition passes because account_a.balance is one. The subtraction changes it to zero, and save() writes zero to the account row. The final line creates a generation for the garden prompt. Press Enter on a blank line to finish the block before switching terminals. Then run the corresponding block in the second shell: if account_b.balance >= 1: account_b.balance -= 1 account_b.save(update_fields=["balance"]) Generation.objects.create(account=account_b, prompt="A beach") The second shell makes its decision using the object it loaded earlier. Its if condition still sees one because we have not refreshed account_b. It subtracts one and saves zero, overwriting the balance with the same value the first operation wrote. It then creates a separate generation for the beach prompt. The second operation has accepted work using a credit the first operation already spent. Finally, check the database from the second shell: account_b.refresh_from_db() print(account_b.balance) print(Generation.objects.filter(account=account_b).count()) This block checks the stored result of both operations. refresh_from_db() reloads the account so the print reflects the database value. The filtered count() counts only generations linked to that account. You should see zero credits and two generation records. The generation count reveals the failure that the balance alone would hide. The two shells let us reproduce the unsafe sequence deliberately. We paused after each read and then allowed both writes. A server can produce the same order when requests overlap, even though it won't do so on every attempt. To repeat the experiment, create a new username and account so previous records don't affect the count. We can now build protection around the exact gap we observed. How to Protect the Balance There are two database concerns in this function. The credit deduction and generation record should succeed together. Competing requests also need a coordinated way to check and update the account. We’ll address them in that order. Keep Related Changes in One Transaction A database transaction groups operations into a unit of work. Django’s transaction.atomic() commits the changes when the block completes successfully and rolls them back if an exception leaves the block. This matters because the unsafe function saves the balance before it creates the generation record. If record creation fails, the deduction can remain without an accepted generation. For illustration, wrapping the operations looks like this: from django.db import transaction @transaction.atomic def reserve_generation_atomic_only(user_id, prompt): account = CreditAccount.objects.get(user_id=user_id) if account.balance < 1: raise InsufficientCredits account.balance -= 1 account.save(update_fields=["balance"]) return Generation.objects.create(account=account, prompt=prompt) The decorator places the function’s database operations inside an atomic transaction. from django.db import transaction provides Django’s transaction tools. The function still reads the account, checks the balance, saves the deduction, and creates the generation in that order. If an exception escapes during record creation, the transaction rolls back the deduction too. This protects the relationship between the deduction and its generation record. The plain account lookup still leaves the credit check exposed. PostgreSQL uses Read Committed as its default isolation level. Under it, a plain read inside a transaction doesn't make another transaction wait before reading the row. Both functions can therefore read one credit before either updates it. We need to coordinate access before making the balance decision. Lock the Account Before the Balance Check Row locking is a database mechanism that restricts conflicting operations on selected rows while a transaction holds a lock. In this example, a row is the stored record for one credit account. A SELECT FOR UPDATE lock makes competing updates and conflicting lock requests wait until the lock is released. Ordinary reads can still proceed, and transactions can work on other account rows. This lets us protect one account while its balance is checked and changed. Add from django.db import transaction at the top of credits/services.py. Keep the unsafe function for comparison, then add this protected version: @transaction.atomic def reserve_generation(user_id, prompt): account = CreditAccount.objects.select_for_update().get(user_id=user_id) if account.balance < 1: raise InsufficientCredits account.balance -= 1 account.save(update_fields=["balance"]) return Generation.objects.create(account=account, prompt=prompt) This function acquires the account’s row lock before it checks the balance. select_for_update() requests the lock, and .get(user_id=user_id) executes the query for this account inside the transaction. Once it holds the lock, the function checks the balance and raises InsufficientCredits if necessary. Otherwise, it saves the deduction and creates the generation before the transaction completes. The decision and its database changes now happen while the account is protected. A competing call to this function must wait at the locking query. If the first call commits its deduction, the waiting call checks the updated balance under our Read Committed setup. If the first call rolls back, its deduction doesn't remain. Our configured timeout can also stop the wait with an error. For one credit and a successful first commit, the second call reads zero and rejects the request. This protection needs to cover every path that spends the balance. An older function could still read an account without requesting a lock. Its later update would wait while our lock is held, but it could then overwrite the balance using its stale value. Administrative adjustments and background jobs therefore need a safe update strategy, too. One protected function can't correct an unsafe writer elsewhere. Keep Image Generation Outside the Transaction The transaction should cover the credit reservation. A remote image request could take much longer than those database operations, so placing it inside the transaction would keep competing requests waiting unnecessarily. For our demo, add this function to credits/services.py: def simulate_generation(generation): # No external AI request is made in this tutorial. generation.status = "completed" generation.save(update_fields=["status"]) return "Simulated image generation completed." This function simulates completion of an accepted generation. It changes the supplied generation object’s status to completed. The save() call persists only that field. The return value is a message, so the function produces no image and makes no external request. We’ll call it after the reservation function returns. The call order keeps image work outside the reservation transaction. With these settings and no enclosing transaction, the reservation commits before the simulation starts. A failure after that point wouldn't automatically restore the credit. A real provider integration needs its own retry or refund policy. We’ll return to those limits after we test the reservation itself. How to Accept and Test Image Requests Now we can connect the protected function to an endpoint. Add this code to credits/views.py: from rest_framework import serializers, status from rest_framework.response import Response from rest_framework.views import APIView from .models import CreditAccount from .services import InsufficientCredits, reserve_generation, simulate_generation class GenerationInput(serializers.Serializer): prompt = serializers.CharField(max_length=500) class GenerateView(APIView): def post(self, request): serializer = GenerationInput(data=request.data) serializer.is_valid(raise_exception=True) try: generation = reserve_generation( request.user.pk, serializer.validated_data["prompt"] ) except CreditAccount.DoesNotExist: return Response({"detail": "Credit account not found."}, status=404) except InsufficientCredits: return Response({"detail": "Not enough credits."}, status=409) result = simulate_generation(generation) return Response( {"id": generation.pk, "status": generation.status, "result": result}, status=status.HTTP_201_CREATED, ) The serializer checks the prompt before the view spends a credit. GenerationInput declares a required text field with a maximum length of 500 characters. is_valid(raise_exception=True) rejects missing, blank, or invalid input with a validation response. The view then reads the cleaned value from validated_data and passes it with request.user.pk, the authenticated user’s database identifier, to the reservation function. The client supplies a prompt while the server chooses the account from the authenticated user. The view translates the reservation outcome into a response. A missing account produces404, and InsufficientCredits produces 409, our chosen response for the balance conflict. On success, the view calls the simulation after the reservation returns. It sends 201 with the record identifier, completion status, and simulated result. These branches let the caller distinguish accepted work from a rejected request. Replace config/urls.py with: from django.urls import path from credits.views import GenerateView urlpatterns = [ path("api/generate/", GenerateView.as_view()), ] This route connects the request address to our view. path() matches the api/generate/ part of the address. GenerateView.as_view() turns the class-based view into a callable Django can dispatch to. The view’s post() method handles a POST request at that route. The endpoint is now available at /api/generate/ when the server runs. Try One Request at a Time Open the Django shell and create a separate user for the endpoint demonstration: from django.contrib.auth import get_user_model from credits.models import CreditAccount user = get_user_model().objects.create_user( username="api-demo", password="local-example-password", ) CreditAccount.objects.create(user=user, balance=1) This block creates a user for the authenticated endpoint example. create_user() saves the username and hashes the supplied password. CreditAccount.objects.create() gives the new user one credit. A separate username keeps this check independent of the account used in the two-shell experiment. The credentials below belong only to this local demonstration account. Exit the shell and start the development server: python manage.py runserver This command starts Django’s development server. With no address argument, it listens at 127.0.0.1:8000. Requests at that address pass through the route configuration we just added. Keep this terminal open while you send the request from another terminal. This server is for the local demonstration. In another terminal, send a request: curl -i -u api-demo:local-example-password \ -H "Content-Type: application/json" \ -d '{"prompt": "A garden at sunrise"}' \ http://127.0.0.1:8000/api/generate/ This command submits a prompt to the endpoint with the demonstration user’s credentials. -i includes response headers, and -u supplies the username and password for Basic authentication. -H declares JavaScript Object Notation (JSON) as the request body format. -d supplies that body and makes curl send a POST request to the address shown. The first request should return 201 with the simulated completion result. Sending the command again checks the account after the first deduction. The next request should find zero credits. The view should return 409 with "detail": "Not enough credits.". Because we waited between requests, this verifies the ordinary sequence without exercising concurrency. Next, we’ll test overlapping operations. Set Up the Automated Tests Replace credits/tests.py with the following imports and helper class. We’ll add the test methods in the next steps. from concurrent.futures import ThreadPoolExecutor from threading import Barrier from unittest.mock import patch from django.contrib.auth import get_user_model from django.db import ( OperationalError, close_old_connections, connection, connections, transaction, ) from django.test import TransactionTestCase from rest_framework.test import APIClient from .models import CreditAccount, Generation from .services import InsufficientCredits, reserve_generation, reserve_generation_unsafe class ConcurrencyTests(TransactionTestCase): def setUp(self): if connection.vendor != "postgresql": self.skipTest("Run these concurrency tests on PostgreSQL.") self.user = get_user_model().objects.create_user(username="parallel-reader") self.account = CreditAccount.objects.create(user=self.user, balance=1) def run_two(self, action): def worker(): close_old_connections() try: return action() finally: connections.close_all() with ThreadPoolExecutor(max_workers=2) as pool: futures = [pool.submit(worker) for _ in range(2)] return [future.result(timeout=15) for future in futures] The test class prepares a fresh account before each concurrency test. setUp() skips the test when the connection isn't PostgreSQL. It then creates a user and an account with one credit. TransactionTestCase allows actual transaction boundaries, unlike regularTestCase, whose enclosing transactions can hide lock-usage mistakes. A skipped test on SQLite doesn't verify PostgreSQL’s lock behaviour. The run_two() helper gives the same action to two worker threads. ThreadPoolExecutor(max_workers=2) provides the workers, and each pool.submit(worker) schedules one call. A future represents that call’s eventual result, which future.result(timeout=15) retrieves or raises an error for. Each worker clears unusable old connections before the action and closes its own connections, including when the action fails. This lets the two operations reach the same database through separate connections. Confirm the Unsafe Behaviour Add this method insideConcurrencyTests, at the same indentation level as run_two(): def test_reproduce_unsafe_spending(self): both_have_read = Barrier(2) original_get = CreditAccount.objects.get def read_then_wait(*args, **kwargs): account = original_get(*args, **kwargs) both_have_read.wait(timeout=5) return account with patch( "credits.services.CreditAccount.objects.get", side_effect=read_then_wait, ): self.run_two( lambda: reserve_generation_unsafe(self.user.pk, "A garden").pk ) self.account.refresh_from_db() self.assertEqual(self.account.balance, 0) self.assertEqual(Generation.objects.count(), 2) This test forces both unsafe operations to read before either proceeds. original_get keeps the real lookup, and read_then_wait() calls it before waiting at Barrier(2). The barrier releases the workers only after both arrive, or raises an error if its wait times out. patch() temporarily replaces the service’s lookup with this wrapper for the duration of the with block. The database read stays real while the test controls the pause after it. The final assertions document the deliberately incorrect result. The small lambda calls the unsafe function and returns the created record’s identifier for each worker. After both return, refresh_from_db() reloads the account. The assertions expect a zero balance and two generation records in this isolated test. A pass here confirms reproduction of the bug, not correctness of the unsafe function. Check the Protected Endpoint Add these methods inside the same class: def concurrent_api_requests(self): ready = Barrier(2) def send_request(): client = APIClient() client.force_authenticate(self.user) ready.wait(timeout=5) return client.post( "/api/generate/", {"prompt": "A garden"}, format="json", ).status_code return self.run_two(send_request) def test_last_credit_accepts_only_one_request(self): self.assertEqual(sorted(self.concurrent_api_requests()), [201, 409]) self.account.refresh_from_db() self.assertEqual(self.account.balance, 0) self.assertEqual(Generation.objects.count(), 1) def test_two_credits_accept_both_requests(self): self.account.balance = 2 self.account.save(update_fields=["balance"]) self.assertEqual(self.concurrent_api_requests(), [201, 201]) self.account.refresh_from_db() self.assertEqual(self.account.balance, 0) self.assertEqual(Generation.objects.count(), 2) The request helper sends two authenticated requests through the protected view. Each worker creates its ownAPIClient, and force_authenticate() supplies the test user without a password exchange. The barrier sits before post() so both workers reach the request start together. Each call returns its response code to run_two(). This tests the endpoint’s credit behaviour without also testing the authentication mechanism. The two test methods cover both insufficient and sufficient shared credit. The one-credit test sorts the codes because either worker may finish first, then expects one 201 and one 409. It also checks zero remaining credits and exactly one generation record. The two-credit test changes the starting balance and expects two successes, two records, and a zero balance. Checking the records as well as the responses helps catch an incorrectly accepted generation. The barrier coordinates request starts without controlling every database operation. One request could still progress faster than the other. Moving the barrier after lock acquisition would make the first worker wait for a second worker that can't acquire its lock. The current placement avoids that artificial deadlock but doesn't prove every possible execution order. Use this regression test alongside the controlled reproduction and documented database behaviour. Verify the Service Encounters a Held Lock We can also test whether the reservation function waits for a lock before it checks the balance. The test below starts with zero credits so the balance check would immediately reject an unprotected read. The main connection acquires the account lock before it starts a worker. That worker calls the actual reservation function with a one-second lock timeout. Add this method inside ConcurrencyTests: def test_reservation_waits_for_account_lock(self): self.account.balance = 0 self.account.save(update_fields=["balance"]) user_id = self.user.pk def attempt_reservation(): close_old_connections() try: try: with transaction.atomic(): with connection.cursor() as cursor: cursor.execute("SET LOCAL lock_timeout = '1s'") reserve_generation(user_id, "A garden") except OperationalError as error: return error.__cause__.sqlstate except InsufficientCredits: return "insufficient_credits" return "accepted" finally: connections.close_all() with ThreadPoolExecutor(max_workers=1) as pool: with transaction.atomic(): CreditAccount.objects.select_for_update().get(pk=self.account.pk) result = pool.submit(attempt_reservation).result(timeout=5) self.assertEqual(result, "55P03") result = pool.submit(attempt_reservation).result(timeout=5) self.assertEqual(result, "insufficient_credits") self.account.refresh_from_db() self.assertEqual(self.account.balance, 0) self.assertEqual(Generation.objects.count(), 0) This test controls when the competing lock exists. The main connection keeps its transaction open while it waits for the worker’s result. Inside the worker’s transaction, SET LOCAL temporarily shortens the lock timeout. The worker should return PostgreSQL’s 55P03 error code, which means lock_not_available, after the lock wait times out. A balance rejection at this point would show that the service didn't wait for the account lock before its check. The second call checks the same operation after the main transaction releases its lock. This time, the worker can acquire the account lock and read the zero balance. It should return insufficient_credits rather than a database error. The final assertions confirm no credit or generation record changed during either attempt. Together, the two calls check contention and release without relying on two requests happening to overlap. The zero balance is deliberate and makes the test more specific. If we used one credit, an unprotected function might still encounter the lock when it eventually tried to update the row. With zero credits, that function would reject the request before any update and fail our expected timeout assertion. The test therefore checks the service’s lock-before-check behaviour, while the preceding endpoint tests check its spending outcomes. This controlled lock test passed as part of the companion project’s PostgreSQL test suite. Check Rollback After a Failure Outside ConcurrencyTests add another class: class CreditRollbackTests(TransactionTestCase): def test_failed_record_creation_restores_credit(self): user = get_user_model().objects.create_user(username="rollback-reader") account = CreditAccount.objects.create(user=user, balance=1) with patch( "credits.services.Generation.objects.create", side_effect=RuntimeError("Simulated record creation failure"), ): with self.assertRaises(RuntimeError): reserve_generation(user.pk, "A garden") account.refresh_from_db() self.assertEqual(account.balance, 1) self.assertEqual(Generation.objects.count(), 0) This test deliberately raises an exception during generation-record creation. It first creates an account with one credit. The patch makes Generation.objects.create() raise RuntimeError, and assertRaises() confirms that the error leaves the reservation function. After the transaction rolls back, the refreshed account should still have one credit and no generation record. This checks that a failed reservation doesn't leave a deduction behind. Run the tests with PostgreSQL still available: python manage.py test credits -v 2 This command runs the tests in the credits app. -v 2 asks Django to show each test and its outcome. Django creates a separate test database, so the database user needs permission to create it. The user from our local container has that permission, but an existing database setup may need configuration. Expect five passes on the intended PostgreSQL setup. Confirm them by execution before relying on the example. Common Mistakes and Next Steps The example now protects the credit decision, but a few details are easy to miss when you adapt it. Read the Account Inside the Lock Fetching an account before the transaction and continuing to use that object can leave you with an old balance. The protected function deliberately retrieves the account through select_for_update() before making its decision. Keep this order when you move the code into another service or endpoint. Passing in a user identifier makes the function responsible for its own fresh, locked read. Use the Database as the Shared Point of Coordination Disabling a submit button can reduce accidental clicks, but a second tab or another client can still send a request. A Python thread lock also coordinates only the code sharing that lock in the same process. If you run multiple application workers, the credit decision still needs protection at the shared database. Review other balance updates, including administrative adjustments, rather than assuming this endpoint is the only writer. Consider a Conditional Update for Simpler Counters Row locks are one approach. For a simple deduction, Django can also express the balance condition and subtraction in one database update using an F() expression. The condition matters: an unconditional subtraction doesn't enforce sufficient credit. If you also create a generation record, keep the deduction and record creation in one transaction. We used an explicit lock here because it makes the read, decision, and update easy to follow. You can explore conditional updates once you understand what the operation must protect. Separate Concurrent Requests from Retries Our example treats two submissions as two distinct requests. If the user has two credits, both should succeed. A retry introduces a different requirement. The app might accept a generation but lose the response before the client receives it. If the client resends the same logical request, you may want to return the original result without another charge. That requires idempotency: a way to identify a repeated operation and avoid applying its effect again. A typical design uses a request key, a database uniqueness rule, and a stored result. A balance lock alone doesn't identify duplicate intent. Plan for Provider Failures Once the reservation commits, a later provider failure doesn't automatically restore the credit. You need a policy for whether to retry the generation, refund it, or leave it pending for recovery. A real implementation also needs to recover if the application stops after it reserves the credit but before it starts the job. The reserved record gives you something to track, but this tutorial doesn't implement a durable job queue or recovery worker. Keep those concerns visible when you extend the example. Preventing concurrent overspending is one part of a complete credit system. Summary and Next Steps At the start of this guide, two requests could each pass the balance check and spend the same credit. The balance ended at zero, which made the failure easy to overlook. We reproduced the sequence, then protected the decision with a transaction and a row lock. We also checked the generation count, tested the case where both requests had enough credit, and added a rollback test for record creation failure. When you review a similar feature in your own application: State the rule the data must satisfy, such as one credit per accepted generation. Identify where separate requests can read and change the same value. Protect the decision and its related database changes. Test overlapping operations and failures on the database you actually use. The same reasoning applies to the last item in an online store or any shared allowance. A successful check is only useful if the application can safely act on it. References For a deeper look at retries, see Making retries safe with idempotent APIs.