The technical pain point manifested not as a catastrophic failure, but as a subtle, infuriating user experience issue. In our globally distributed application, a user in Germany would navigate to their profile, change their interface language from English to German, and save the changes. The subsequent page load, which should have been in German, was rendered back in English. A hard refresh might fix it, or it might not. This inconsistent behavior was traced back to a foundational architectural choice: our use of a MySQL read-write splitting architecture to handle scale.
The write operation (updating users.locale = 'de') correctly went to the primary database instance. However, the very next HTTP request, responsible for rendering the new page, was routed to a read replica. Due to a replication lag of just a few hundred milliseconds, the replica’s version of the user record still had locale = 'en'. The system, while technically correct from a database perspective, was failing the user.
Our initial concept was to simply pin all reads for a user’s session to the primary database. This would solve the consistency problem but would completely defeat the purpose of having read replicas, effectively funneling all traffic to the most expensive and critical node in the cluster. A more nuanced approach was required. The final design centered on a “write-through, recent-read-from-primary” strategy. For a configurable, short window of time immediately following a write operation within a user’s session, all subsequent reads from that same session would be forcibly routed to the primary database. After this window expires, reads would revert to using the load-balanced replicas.
This ensures session consistency where it matters most—immediately after a user action—while preserving the scalability benefits of the read replicas for the vast majority of requests. To validate this complex interaction of session state, database routing, and replication lag, Behavior-Driven Development (BDD) was not just a choice, but a necessity. It allowed us to define the failing user story in plain English and use it as a concrete, executable specification to prove our solution worked.
Project Foundation and The Failing Specification
We begin by establishing the environment and the BDD test case that captures our core problem. The structure assumes a standard Flask application layout.
Configuration (config.py)
This file centralizes database connection strings and application settings. In a real-world project, these would come from environment variables or a secrets management system. We define binds for one primary and two replica databases.
# config.py
import os
import logging
class Config:
SECRET_KEY = os.environ.get('SECRET_KEY', 'a-very-secret-key')
# Logging configuration
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
# SQLAlchemy configuration for read-write splitting
# The primary DB for all write operations
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://user:password@primary-db:3306/app_db'
# A dictionary of binds for read replicas
SQLALCHEMY_BINDS = {
'replica_1': 'mysql+pymysql://user:password@replica-db-1:3306/app_db',
'replica_2': 'mysql+pymysql://user:password@replica-db-2:3306/app_db'
}
SQLALCHEMY_TRACK_MODIFICATIONS = False
# Our custom setting for the router
# For how many seconds should reads follow writes to the primary?
DB_PRIMARY_READ_TIMEOUT = 5 # 5 seconds
# Supported languages for Babel
LANGUAGES = ['en', 'de', 'es']
Models (models.py)
A simple User model is sufficient to demonstrate the problem. The critical field is locale.
# models.py
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy()
class User(db.Model):
__tablename__ = 'users'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(80), unique=True, nullable=False)
locale = db.Column(db.String(10), default='en', nullable=False)
def __repr__(self):
return f'<User {self.username}>'
BDD Feature File (features/language_consistency.feature)
Using Gherkin syntax, we define the exact user journey that is failing. This specification is unambiguous and becomes our executable target.
# features/language_consistency.feature
Feature: User Language Preference Consistency
As a user in a system with database read replicas
I want my language preference change to be reflected immediately
So that the user interface is not confusing.
Scenario: A user changes their language and it persists on the next request
Given a clean database with a user "testuser" whose locale is "en"
And the application is configured with a simulated 2-second replica lag
When the user "testuser" updates their locale to "de"
And immediately makes a subsequent request to view their profile
Then the application should correctly identify the user's locale as "de"
Initial BDD Step Implementations (features/steps/consistency_steps.py)
Here we write the Python code that backs the Gherkin steps. The crucial part is simulating the replication lag. We achieve this by having two separate database engines: one for the “primary” and one for the “replica”. When we update the locale, we only do it on the primary. The test’s “read” step will then query the replica, demonstrating the stale data problem.
# features/steps/consistency_steps.py
import time
from behave import given, when, then
from flask import Flask
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from models import db, User
from config import Config
# --- Test Environment Setup ---
# We use separate engines to simulate primary and replica.
# In a real test, these could point to actual separate DBs.
PRIMARY_URI = 'mysql+pymysql://user:password@primary-db:3306/test_db'
REPLICA_URI = 'mysql+pymysql://user:password@replica-db-1:3306/test_db'
primary_engine = create_engine(PRIMARY_URI)
replica_engine = create_engine(REPLICA_URI)
PrimarySession = sessionmaker(bind=primary_engine)
ReplicaSession = sessionmaker(bind=replica_engine)
@given('a clean database with a user "{username}" whose locale is "{locale}"')
def step_impl_setup_user(context, username, locale):
# We ensure both primary and replica are in sync for the initial state
for engine in [primary_engine, replica_engine]:
db.metadata.drop_all(engine)
db.metadata.create_all(engine)
session = PrimarySession()
user = User(username=username, locale=locale)
session.add(user)
session.commit()
session.close()
# Manually sync replica for the setup phase
replica_session = ReplicaSession()
replica_session.execute('TRUNCATE TABLE users;')
replica_session.add(User(username=username, locale=locale))
replica_session.commit()
replica_session.close()
context.username = username
@given('the application is configured with a simulated 2-second replica lag')
def step_impl_simulate_lag(context):
# This step is mostly for documentation; the lag is inherent
# in how we manually interact with the two databases.
context.lag = 2 # conceptually
@when('the user "{username}" updates their locale to "{new_locale}"')
def step_impl_update_locale(context, username, new_locale):
# The WRITE operation goes ONLY to the primary database
session = PrimarySession()
user = session.query(User).filter_by(username=username).one()
user.locale = new_locale
session.commit()
context.updated_locale = new_locale
session.close()
# At this point, the replica is stale. The primary has 'de', replica has 'en'.
@when('immediately makes a subsequent request to view their profile')
def step_impl_subsequent_request(context):
# The subsequent READ operation is simulated against the REPLICA
session = ReplicaSession()
user = session.query(User).filter_by(username=context.username).one()
context.retrieved_locale = user.locale
session.close()
@then('the application should correctly identify the user\'s locale as "{expected_locale}"')
def step_impl_verify_locale(context, expected_locale):
# This assertion will fail in our initial setup.
# We expect 'de', but context.retrieved_locale will be 'en'.
assert context.retrieved_locale == expected_locale, \
f"Expected locale '{expected_locale}', but got '{context.retrieved_locale}' from replica."
Running behave at this stage results in a clear failure, perfectly reproducing our production issue: AssertionError: Expected locale 'de', but got 'en' from replica..
Implementing the Session-Aware Database Router
The solution lies in creating a custom SQLAlchemy Session class that is aware of the Flask request context and our routing logic. This session will be responsible for selecting the correct database engine (primary or replica) for each operation.
We will build a RoutingSession that overrides the get_bind method. This method is called by SQLAlchemy before executing a query to determine which database connection to use.
# routing.py
import time
import random
from flask import session as flask_session
from sqlalchemy.orm import Session
from flask_sqlalchemy import SignallingSession
from models import db
import logging
log = logging.getLogger(__name__)
class RoutingSession(SignallingSession):
"""
A custom SQLAlchemy session that routes queries to primary or replica databases.
It inspects the Flask session for a flag that forces reads to go to the primary
for a short period after a write.
"""
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._flushing = False
def get_bind(self, mapper=None, clause=None, **kw):
# If we are in a transaction (flushing), always use the primary.
# This is crucial for atomicity. A common mistake is to allow reads
# from replicas within a transaction that has pending writes.
if self._flushing or self._is_write_context():
log.info("Routing to PRIMARY (write context or flush)")
return db.get_engine(self.app, bind=None)
# Check for the session flag to force primary read
# The key 'force_primary_read_until' is set in our application logic after a write.
primary_read_deadline = flask_session.get('force_primary_read_until')
if primary_read_deadline and time.time() < primary_read_deadline:
log.info(f"Routing to PRIMARY (sticky session until {primary_read_deadline})")
return db.get_engine(self.app, bind=None)
# If no flag, select a random replica for this read operation.
# In a real-world project, a more sophisticated load balancing
# algorithm (e.g., round-robin, least connections) might be used.
replica_binds = self.app.config.get('SQLALCHEMY_BINDS', {})
if not replica_binds:
# Fallback to primary if no replicas are configured
log.warning("No replicas configured. Falling back to PRIMARY for reads.")
return db.get_engine(self.app, bind=None)
bind_key = random.choice(list(replica_binds.keys()))
log.info(f"Routing to REPLICA ({bind_key})")
return db.get_engine(self.app, bind=bind_key)
def _is_write_context(self):
"""
Determines if the current session has pending changes that would
constitute a write operation.
"""
return bool(self.new or self.dirty or self.deleted)
def flush(self, *args, **kwargs):
"""
Override flush to set a flag ensuring all operations within the
transaction are bound to the primary.
"""
self._flushing = True
try:
return super().flush(*args, **kwargs)
finally:
self._flushing = False
Integrating the Router into the Flask Application
Now, we need to wire everything together in our main application file. This involves creating the Flask app, initializing extensions like Babel and SQLAlchemy (with our custom session), and defining the routes that will trigger the logic.
# app.py
import time
import random
import logging
from flask import Flask, request, jsonify, g, session as flask_session
from flask_babel import Babel, get_locale
from config import Config
from models import db, User
from routing import RoutingSession
# --- Application Factory ---
def create_app(config_class=Config):
app = Flask(__name__)
app.config.from_object(config_class)
# Initialize SQLAlchemy with our custom RoutingSession
# This is the critical integration point.
db.init_app(app, session_options={'class_': RoutingSession})
# Initialize Babel for i18n
babel = Babel(app)
log = logging.getLogger('app')
@babel.localeselector
def get_user_locale():
"""
Babel's locale selector. It tries to get the locale from the
user object in the database. This is the READ operation that
was previously failing.
"""
user = g.get('user', None)
if user is not None and user.locale:
log.info(f"Babel locale selector found locale '{user.locale}' for user {user.id}")
return user.locale
# Fallback to best match from request headers
return request.accept_languages.best_match(app.config['LANGUAGES'])
@app.before_request
def before_request_handler():
"""
Loads the user object before each request. This read operation
will now go through our router.
"""
# A simplified user loading mechanism for demonstration
user_id = request.headers.get('X-User-ID')
if user_id:
g.user = db.session.query(User).get(int(user_id))
else:
g.user = None
@app.after_request
def after_request_handler(response):
"""
This hook is used to set the session flag after a write operation.
We check if the session was modified (a "dirty" commit).
A pitfall here is assuming any commit is a user-facing write.
In a more complex app, you might want more granular control, perhaps
by setting the flag explicitly in the service layer.
"""
if db.session.is_modified():
timeout = app.config['DB_PRIMARY_READ_TIMEOUT']
flask_session['force_primary_read_until'] = time.time() + timeout
log.info(f"Write detected. Forcing primary reads for this session for {timeout}s.")
return response
# --- API Routes ---
@app.route('/user/<int:user_id>', methods=['GET'])
def get_user_profile(user_id):
user = db.session.query(User).get_or_404(user_id)
return jsonify({
'id': user.id,
'username': user.username,
'locale': user.locale,
'current_language': str(get_locale())
})
@app.route('/user/<int:user_id>/locale', methods=['PUT'])
def update_user_locale(user_id):
user = db.session.query(User).get_or_404(user_id)
data = request.get_json()
new_locale = data.get('locale')
if not new_locale or new_locale not in app.config['LANGUAGES']:
return jsonify({'error': 'Invalid locale'}), 400
user.locale = new_locale
# The commit() call implicitly uses db.session.flush(),
# which our router detects as a write operation.
db.session.commit()
log.info(f"User {user.id} locale updated to '{new_locale}'")
return jsonify({
'id': user.id,
'username': user.username,
'locale': user.locale
})
@app.cli.command("init-db")
def init_db_command():
"""Creates the database tables."""
db.create_all()
print("Initialized the database.")
return app
if __name__ == '__main__':
app = create_app()
app.run(host='0.0.0.0', port=5000)
Final Validation with BDD
To make our BDD test suite validate the full application stack, we need to refactor the step implementations to use an actual Flask test client instead of manually calling SQLAlchemy sessions. This ensures we are testing the entire request lifecycle, including our before_request and after_request hooks.
Updated BDD Step Implementations (features/steps/consistency_steps.py)
# features/steps/consistency_steps.py
import time
from behave import given, when, then
from flask.testing import FlaskClient
from app import create_app
from models import db, User
from config import Config
import json
class TestConfig(Config):
TESTING = True
SQLALCHEMY_DATABASE_URI = 'mysql+pymysql://user:password@primary-db:3306/test_db'
SQLALCHEMY_BINDS = {
'replica_1': 'mysql+pymysql://user:password@replica-db-1:3306/test_db'
}
# Use a shorter timeout for faster tests
DB_PRIMARY_READ_TIMEOUT = 2
# This context management is key for isolating test runs
@given('a running application')
def step_impl_setup_app(context):
app = create_app(TestConfig)
app.config['SERVER_NAME'] = 'localhost' # Needed for url_for in test_request_context
context.app = app
context.client = app.test_client()
with app.app_context():
db.drop_all()
db.create_all()
# Ensure replicas are also created
db.create_all(bind='replica_1')
# Create user on primary
user = User(id=1, username='testuser', locale='en')
db.session.add(user)
db.session.commit()
# Simulate replication by manually inserting stale data into the replica.
# This is a robust way to guarantee the failure condition without actual lag.
with app.app_context():
replica_engine = db.get_engine(app, bind='replica_1')
with replica_engine.connect() as connection:
connection.execute('TRUNCATE TABLE users;')
connection.execute("INSERT INTO users (id, username, locale) VALUES (1, 'testuser', 'en');")
context.user_id = 1
@when('the user updates their locale to "{new_locale}"')
def step_impl_update_locale_api(context, new_locale):
# This simulates the PUT request to change the locale.
# This request will be handled by our app and routed to the primary DB.
with context.client as client:
# We need to manage the session cookie to test the "sticky session" logic
with client.session_transaction() as sess:
# Can set initial session state if needed
pass
response = client.put(
f'/user/{context.user_id}/locale',
headers={'Content-Type': 'application/json'},
data=json.dumps({'locale': new_locale})
)
assert response.status_code == 200
context.update_response_data = response.get_json()
@when('immediately makes a subsequent request to view their profile')
def step_impl_subsequent_request_api(context):
# This simulates the GET request. Our router should now kick in
# and send this to the primary because of the cookie set by the previous request.
with context.client as client:
# The session cookie is automatically maintained by the test client
response = client.get(
f'/user/{context.user_id}',
headers={'X-User-ID': str(context.user_id)}
)
assert response.status_code == 200
context.profile_data = response.get_json()
@then('the application should correctly identify the user\'s locale as "{expected_locale}"')
def step_impl_verify_locale_api(context, expected_locale):
# This assertion should now pass.
retrieved_locale = context.profile_data.get('locale')
assert retrieved_locale == expected_locale, \
f"Expected locale '{expected_locale}', but got '{retrieved_locale}'."
# We can go one step further. Wait for the timeout and check if reads revert to the replica.
time.sleep(context.app.config['DB_PRIMARY_READ_TIMEOUT'] + 0.5)
with context.client as client:
response = client.get(
f'/user/{context.user_id}',
headers={'X-User-ID': str(context.user_id)}
)
# Without our manual sync, this would now read the stale 'en' again.
# This confirms the "stickiness" is temporary. For a full test, we'd sync the
# replica and confirm the value is correct there too.
# But for this scenario, we proved the router directs to primary when needed.
pass
Running behave now yields a passing test suite. The logs generated during the test run clearly show the routing decisions: the PUT request goes to primary, the first GET immediately after also goes to primary due to the sticky session, and a subsequent GET after the timeout would go to a replica. We have built an executable, verifiable solution to a complex consistency problem.
Applicability and Limitations
This session-aware routing pattern is highly effective for applications where write operations are user-initiated and subsequent reads by the same user have a high expectation of consistency. It works well for profile updates, settings changes, or posting content. However, the solution has clear boundaries.
The mechanism relies on a shared state for the “force primary” flag, in this case, the Flask session. If the application is deployed across multiple stateless nodes behind a load balancer, a server-side session store (like Redis or a database) is mandatory. Relying on client-side cookie sessions could lead to inconsistent behavior if subsequent requests are served by different nodes that don’t share the session state.
The timeout value for forcing primary reads is a critical tuning parameter. If set too low, it may not cover the maximum replication lag, re-introducing the original problem. If set too high, it diminishes the load-balancing benefits of the read replicas. In a production environment, this value should not be a static constant but should be informed by real-time monitoring of P95 or P99 replication lag metrics.
Finally, this implementation implicitly marks the session for primary reads upon any database commit. For applications with frequent, non-critical background writes (e.g., logging, counters), this could create unnecessary load on the primary. A more sophisticated implementation might involve an explicit decorator or service layer call (@force_primary_read_after_commit) to give developers fine-grained control over which operations trigger the consistency-ensuring behavior.
