File size: 15,336 Bytes
cc036ff | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 | """
Database Lock and Deadlock Tests
Tests for database transaction behavior under concurrent access.
These tests document SQLite limitations and PostgreSQL behavior.
Key Bugs Tested:
- Transaction rollback on deadlock
- SELECT FOR UPDATE locking behavior
- Isolation level differences
- Connection pool exhaustion
SQLite Limitations Documented:
- Only one writer at a time (serialized access)
- No true parallel write concurrency
- Limited deadlock detection
- For production, PostgreSQL provides:
- SERIALIZABLE isolation for phantom read prevention
- Real deadlock detection
- Row-level locking with SELECT FOR UPDATE
- Connection pooling with proper isolation
"""
import threading
import time
import pytest
import uuid
from datetime import datetime
from typing import List
from sqlalchemy import text
from sqlalchemy.orm import Session
from core.models import AgentRegistry, AgentStatus
from core.database import SessionLocal
class TestTransactionDeadlockHandling:
"""Test transaction behavior under concurrent access."""
def test_sqlite_concurrent_write_serialization(self, db_session: Session):
"""
CONCURRENT: SQLite serializes concurrent writes.
Documents that SQLite allows only one writer at a time.
Multiple threads writing will be serialized (not truly concurrent).
BUG_PATTERN: Assuming parallel writes on SQLite.
EXPECTED: Writes are serialized, no corruption.
"""
# Create test agents
agents = []
for i in range(5):
agent = AgentRegistry(
id=str(uuid.uuid4()),
name=f"Agent{i}",
category="test",
module_path="test.module",
class_name="TestAgent",
status=AgentStatus.INTERN.value,
confidence_score=0.6,
)
db_session.add(agent)
agents.append(agent)
db_session.commit()
errors = []
update_order = []
def update_agent(agent_id: int, new_confidence: float):
"""Update agent confidence score."""
local_db = SessionLocal()
try:
agent = local_db.query(AgentRegistry).filter(
AgentRegistry.id == agents[agent_id].id
).first()
if agent:
agent.confidence_score = new_confidence
local_db.commit()
update_order.append(agent_id)
except Exception as e:
errors.append(e)
local_db.rollback()
finally:
local_db.close()
# Launch concurrent updates (will be serialized by SQLite)
threads = []
for i in range(5):
thread = threading.Thread(target=update_agent, args=(i, 0.7 + i * 0.05))
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
# Verify no errors (SQLite serialized writes successfully)
assert len(errors) == 0, f"Errors during concurrent writes: {errors}"
# Verify agents were updated (SQLite serializes writes, so final value depends on last writer)
for agent in agents:
db_session.refresh(agent)
# Agent may have original value or one of the updates (last write wins in SQLite)
assert agent.confidence_score >= 0.6, f"Agent {agent.name} has unexpected value: {agent.confidence_score}"
def test_sqlite_read_write_concurrency(self, db_session: Session):
"""
CONCURRENT: SQLite allows one writer OR multiple readers.
Documents SQLite concurrency model: one writer blocks all other
operations (writes and reads). Multiple readers allowed only
when no writer is active.
BUG_PATTERN: Assuming reads can proceed during writes.
EXPECTED: Reads blocked during writes (serialized).
"""
# Create test agents
for i in range(10):
agent = AgentRegistry(
id=str(uuid.uuid4()),
name=f"Agent{i}",
category="test",
module_path="test.module",
class_name="TestAgent",
status=AgentStatus.INTERN.value,
confidence_score=0.6,
)
db_session.add(agent)
db_session.commit()
errors = []
read_count = [0]
write_count = [0]
def read_agents():
"""Read agents."""
local_db = SessionLocal()
try:
agents = local_db.query(AgentRegistry).all()
read_count[0] = len(agents)
except Exception as e:
errors.append(e)
finally:
local_db.close()
def write_agent():
"""Write agent."""
local_db = SessionLocal()
try:
agent = local_db.query(AgentRegistry).first()
if agent:
agent.confidence_score = 0.8
local_db.commit()
write_count[0] += 1
except Exception as e:
errors.append(e)
local_db.rollback()
finally:
local_db.close()
# Launch mixed reads and writes
threads = []
for i in range(5):
threads.append(threading.Thread(target=read_agents))
threads.append(threading.Thread(target=write_agent))
for thread in threads:
thread.start()
for thread in threads:
thread.join()
# Verify operations completed (serialized)
assert len(errors) == 0, f"Errors: {errors}"
assert read_count[0] > 0, "Should have read agents"
assert write_count[0] > 0, "Should have written agents"
class TestDatabaseIsolationLevels:
"""Test transaction isolation levels."""
def test_read_committed_isolation(self, db_session: Session):
"""
CONCURRENT: Read committed isolation behavior.
Documents that SQLite defaults to read committed (or higher).
Uncommitted changes from other transactions are not visible.
BUG_PATTERN: Assuming uncommitted reads are visible.
EXPECTED: Reads see only committed data.
"""
# Create test agent
agent = AgentRegistry(
id=str(uuid.uuid4()),
name="TestAgent",
category="test",
module_path="test.module",
class_name="TestAgent",
status=AgentStatus.INTERN.value,
confidence_score=0.6,
)
db_session.add(agent)
db_session.commit()
uncommitted_value = [None]
committed_value = [None]
errors = []
def write_uncommitted():
"""Write but don't commit."""
local_db = SessionLocal()
try:
agent = local_db.query(AgentRegistry).first()
agent.confidence_score = 0.9
# Don't commit yet
uncommitted_value[0] = agent.confidence_score
time.sleep(0.1) # Hold transaction open
local_db.commit() # Now commit
except Exception as e:
errors.append(e)
finally:
local_db.close()
def read_committed():
"""Read committed data."""
time.sleep(0.05) # Start after write begins but before commit
local_db = SessionLocal()
try:
agent = local_db.query(AgentRegistry).first()
committed_value[0] = agent.confidence_score
except Exception as e:
errors.append(e)
finally:
local_db.close()
# Launch threads
thread1 = threading.Thread(target=write_uncommitted)
thread2 = threading.Thread(target=read_committed)
thread1.start()
thread2.start()
thread1.join()
thread2.join()
# Verify no errors
assert len(errors) == 0, f"Errors: {errors}"
# Committed read can see any value depending on SQLite timing
# (0.6 original, 0.7-0.9 intermediate, or final 0.9)
assert 0.6 <= committed_value[0] <= 0.9, f"Unexpected value: {committed_value[0]}"
def test_serializable_isolation_note(self, db_session: Session):
"""
DOCUMENTATION: PostgreSQL SERIALIZABLE isolation.
Notes that PostgreSQL supports SERIALIZABLE isolation level
which prevents phantom reads. SQLite has limited isolation
level support.
For production with PostgreSQL:
engine = create_engine(
"postgresql://...",
isolation_level="SERIALIZABLE"
)
This ensures:
- No phantom reads
- No non-repeatable reads
- Full serializable execution
"""
# This is a documentation test
# SQLite doesn't support SET TRANSACTION ISOLATION LEVEL
# PostgreSQL does, and it's recommended for high-concurrency
# Verify we can query current isolation
result = db_session.execute(text("PRAGMA read_uncommitted")).scalar()
# SQLite returns 0 (False) for read committed (default)
assert result == 0, "SQLite should default to read committed"
class TestConnectionPoolBehavior:
"""Test connection pool under concurrent load."""
def test_connection_pool_exhaustion_handling(self, db_session: Session):
"""
CONCURRENT: Connection pool doesn't exhaust under load.
Tests that connection pool handles many concurrent connections
without exhaustion. SQLite uses StaticPool (single connection)
but tests should verify pool behavior.
BUG_PATTERN: Connection leak causing pool exhaustion.
EXPECTED: All connections released properly.
"""
# Create test agents
for i in range(20):
agent = AgentRegistry(
id=str(uuid.uuid4()),
name=f"Agent{i}",
category="test",
module_path="test.module",
class_name="TestAgent",
status=AgentStatus.INTERN.value,
confidence_score=0.6,
)
db_session.add(agent)
db_session.commit()
errors = []
successful_reads = [0]
def read_and_close():
"""Read from database and close connection."""
local_db = SessionLocal()
try:
# Multiple reads per connection
for _ in range(5):
agents = local_db.query(AgentRegistry).limit(5).all()
successful_reads[0] += len(agents)
local_db.commit()
except Exception as e:
errors.append(e)
finally:
local_db.close()
# Launch many concurrent DB operations
threads = []
for _ in range(20):
thread = threading.Thread(target=read_and_close)
threads.append(thread)
thread.start()
for thread in threads:
thread.join()
# Verify no connection pool errors
assert len(errors) == 0, f"Connection pool errors: {errors}"
# Verify all reads succeeded
assert successful_reads[0] > 0, "Should have completed reads"
class TestPostgreSQLBehaviorDocumentation:
"""Document expected PostgreSQL behavior for production."""
def test_select_for_update_pattern(self, db_session: Session):
"""
DOCUMENTATION: SELECT FOR UPDATE for pessimistic locking.
Documents the pattern for row-level locking in PostgreSQL.
SQLite doesn't support SELECT FOR UPDATE, but PostgreSQL does.
PostgreSQL Pattern:
with get_db_session() as db:
# Lock the row for update
agent = db.query(AgentRegistry).filter(
AgentRegistry.id == agent_id
).with_for_update().first()
# Update while holding lock
agent.confidence_score = 0.8
db.commit()
This prevents race conditions when updating the same row
from multiple transactions.
SQLite Behavior:
- Entire database locked during write
- SELECT FOR UPDATE not supported (ignored)
- Serialization happens at DB level, not row level
"""
# Document the pattern
# SQLite will ignore with_for_update()
agent = AgentRegistry(
id=str(uuid.uuid4()),
name="TestAgent",
category="test",
module_path="test.module",
class_name="TestAgent",
status=AgentStatus.INTERN.value,
confidence_score=0.6,
)
db_session.add(agent)
db_session.commit()
# Try SELECT FOR UPDATE (will be ignored by SQLite)
result = db_session.query(AgentRegistry).filter(
AgentRegistry.id == agent.id
).with_for_update().first()
assert result is not None, "Should find agent"
# Note: SQLite ignores FOR UPDATE, PostgreSQL would lock the row
def test_deadlock_detection_note(self, db_session: Session):
"""
DOCUMENTATION: PostgreSQL deadlock detection.
Notes that PostgreSQL automatically detects deadlocks and
rolls back one of the transactions. SQLite has limited
deadlock detection due to serialized writes.
PostgreSQL Deadlock Behavior:
- Automatically detects circular wait conditions
- Rolls back one transaction (returns error)
- Application should retry the transaction
Retry Pattern:
max_retries = 3
for attempt in range(max_retries):
try:
# Database operation
db.commit()
break
except OperationalError as e:
if "deadlock" in str(e).lower():
db.rollback()
if attempt < max_retries - 1:
time.sleep(0.1 * (attempt + 1))
continue
raise
SQLite Behavior:
- Rare true deadlocks (due to serialization)
- Lock timeouts more common than deadlocks
- Error: "database is locked" (not "deadlock")
"""
# This is a documentation test
# Document retry pattern for PostgreSQL deadlocks
retry_count = [0]
max_retries = 3
def operation_with_retry():
"""Example retry pattern."""
for attempt in range(max_retries):
try:
# Simulated operation
retry_count[0] = attempt + 1
return True
except Exception as e:
if "deadlock" in str(e).lower() and attempt < max_retries - 1:
time.sleep(0.1 * (attempt + 1))
continue
raise
result = operation_with_retry()
assert result is True
assert retry_count[0] == 1 # No retries needed in this example
|