The Engineering Realities of High-Stakes Online Examinations
Building a Computer Based Testing (CBT) platform presents unique architectural challenges that standard CRUD applications never face:
- Clock Tampering & Network Latency: If the countdown timer relies on the client's browser clock, malicious candidates can pause JavaScript execution or rewind operating system clocks to gain unfair time extensions.
- Data Loss During Intermittent Dropouts: If a student's WiFi disconnects after 45 minutes of answering questions, submitting all answers in a single final payload invites catastrophic data loss.
- Proctoring Telemetry at Scale: School administrators and teachers require instantaneous visibility into student progress, active exam rooms, and anti-cheat alerts without hammering the primary database with aggressive 2-second HTTP polling.
In mfarim/fastapi-elearning-react, we engineered an asynchronous examination engine designed for zero-data-loss, server-authoritative timekeeping, and native WebSocket proctoring.
┌────────────────────────────────────────────────────────────────────────────────────────┐
│ CBT EXAM RUNNER & PROCTORING TOPOLOGY │
└────────────────────────────────────────────────────────────────────────────────────────┘
👨🎓 Candidate Client (React 19) 👨🏫 Teacher Proctor Screen
┌────────────────────────────┐ ┌────────────────────────────┐
│ • Fullscreen Lockdown │ │ • Live Candidate Grid │
│ • Tab-Switch Event Listener│ │ • Real-Time Progress Bars │
│ • Instant Debounced Autosave│ │ • Anti-Cheat Alert Badges │
└─────────────┬──────────────┘ └─────────────▲──────────────┘
│ │
HTTPS │ Answers / Violations WSS │ Telemetry Broadcast
▼ │
┌───────────────────────────────────────────────────────────────┴──────────────┐
│ FASTAPI ASGI BACKEND (Uvicorn) │
│ │
│ ┌───────────────────────────────┐ ┌──────────────────────────────────┐ │
│ │ ExamRunnerService │ │ ConnectionManager (WSS) │ │
│ │ • Server Timer Delta Check │────▶│ • Room-based Active WebSockets │ │
│ │ • Auto-Submit on Deadline │ │ • Dead Socket Garbage Collector │ │
│ │ • Violations Threshold (>=5) │ └──────────────────┬───────────────┘ │
│ └──────────────┬────────────────┘ │ │
└─────────────────┼─────────────────────────────────────────┼──────────────────┘
│ │
▼ ▼
┌──────────────────────┐ ┌──────────────────────┐
│ PostgreSQL 16 │ │ Redis 7 Pub/Sub │
│ (ExamAttempt/Answers)│ │ (Cluster Broadcast) │
└──────────────────────┘ └──────────────────────┘
1. Server-Authoritative Timer & Exam State Machine
A golden rule of secure exam architecture is that the client never dictates the remaining exam time.
In backend/app/api/v1/exam_runner.py, when a candidate launches an exam via /api/v1/exams/{id}/start, the backend records an immutable UTC timestamp started_at in the database:
PYTHON
# backend/app/api/v1/exam_runner.py
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import selectinload
from app.core.database import get_db
from app.models.exam import Examination, ExamAttempt
from app.schemas.exam import ExamStartResponse
from app.api.deps import require_roles
@router.post("/{id}/start", response_model=ApiResponse[ExamStartResponse])
async def start_exam(
id: int,
db: AsyncSession = Depends(get_db),
current_user = Depends(require_roles(["ROLE_STUDENT", "ROLE_ADMIN"]))
):
student_id = current_user.student_profile.id
now = datetime.utcnow()
# 1. Fetch examination details
stmt = (
select(Examination)
.options(selectinload(Examination.questions))
.where(Examination.id == id, Examination.deleted_at.is_(None))
)
exam = (await db.execute(stmt)).scalars().first()
if not exam or exam.status == "draft":
raise HTTPException(status_code=400, detail="Exam not available")
total_duration_sec = exam.duration_minutes * 60
# 2. Check for active attempts
att_stmt = (
select(ExamAttempt)
.options(selectinload(ExamAttempt.answers))
.where(ExamAttempt.examination_id == id, ExamAttempt.student_id == student_id)
.order_by(ExamAttempt.id.desc())
)
attempt = (await db.execute(att_stmt)).scalars().first()
if attempt and attempt.status in ["in_progress"]:
# Resume existing attempt: Calculate authoritative elapsed time
elapsed = (now - attempt.started_at).total_seconds()
remaining = max(0, int(total_duration_sec - elapsed))
# Enforce server-side hard cut-off
if remaining <= 0:
attempt.status = "submitted"
attempt.finished_at = now
await db.commit()
raise HTTPException(status_code=400, detail="Time limit reached for this exam")
else:
# Create brand-new attempt
attempt = ExamAttempt(
examination_id=id,
student_id=student_id,
status="in_progress",
started_at=now,
)
db.add(attempt)
await db.commit()
await db.refresh(attempt)
remaining = total_duration_sec
return ApiResponse.ok(
data=build_exam_start_dto(attempt, exam, remaining),
message="Exam session initialized"
)
Why this design is bulletproof:
- Network Dropout Immunity: If a student's laptop battery dies or connection drops for 5 minutes, resuming the session calculates: $$\text{Remaining Time} = (\text{Duration} \times 60) - (\text{Server Now} - \text{Started At})$$ The timer resumes strictly based on the server's master clock without granting illegal time extensions.
- Hard Deadline Expiry: If
remaining <= 0, any subsequent answer submission is immediately rejected, and the attempt is transitioned to"submitted".
2. Instant Autosave with Optimistic UI Synchronization
Submitting an entire exam at the very end is an anti-pattern. If a connection failure occurs on question 50, the candidate loses all prior work.
In our React client (frontend/src/pages/exams/ExamRunner.tsx), every option click triggers an immediate optimistic UI update (coloring the question palette green) while debouncing an asynchronous background sync:
PYTHON
# backend/app/api/v1/exam_runner.py
@router.post("/attempts/{attempt_id}/answers", response_model=ApiResponse[Dict[str, Any]])
async def submit_answer(
attempt_id: int,
payload: SubmitAnswerRequest,
db: AsyncSession = Depends(get_db),
current_user = Depends(require_roles(["ROLE_STUDENT", "ROLE_ADMIN"]))
):
attempt = await get_valid_in_progress_attempt(db, attempt_id, current_user.student_profile.id)
# 1. Check if an answer record already exists for this question (Upsert)
stmt = select(ExamAnswer).where(
ExamAnswer.exam_attempt_id == attempt_id,
ExamAnswer.question_id == payload.question_id
)
answer = (await db.execute(stmt)).scalars().first()
if not answer:
answer = ExamAnswer(
exam_attempt_id=attempt_id,
question_id=payload.question_id,
student_answer=payload.selected_option,
)
db.add(answer)
else:
answer.student_answer = payload.selected_option
answer.updated_at = datetime.utcnow()
await db.commit()
# 2. Broadcast live progress to supervising teachers via WebSockets
await ws_manager.broadcast_to_exam(
exam_id=attempt.examination_id,
event_type="ANSWER_SUBMITTED",
data={
"studentId": current_user.student_profile.id,
"attemptId": attempt_id,
"questionId": payload.question_id,
"totalAnswered": await count_answered_questions(db, attempt_id)
}
)
return ApiResponse.ok(message="Answer saved successfully")
3. Anti-Cheat Proctoring & Violation Trapping
Maintaining testing integrity in unproctored or remote computer labs requires aggressive client-side detection backed by server-side threshold enforcement.
Frontend Detection Hooks (ExamRunner.tsx)
TYPESCRIPT
// Detect window blur / switching to browser tabs or cheating tools
useEffect(() => {
const handleVisibilityChange = () => {
if (document.hidden) {
reportViolation("TAB_SWITCH_DETECTED");
}
};
const handleFullscreenChange = () => {
if (!document.fullscreenElement) {
reportViolation("FULLSCREEN_EXITED");
}
};
document.addEventListener("visibilitychange", handleVisibilityChange);
document.addEventListener("fullscreenchange", handleFullscreenChange);
return () => {
document.removeEventListener("visibilitychange", handleVisibilityChange);
document.removeEventListener("fullscreenchange", handleFullscreenChange);
};
}, []);
Server-Side Violation Threshold Handler
When a candidate attempts to switch tabs or exit fullscreen, the violation count increments. When violations reach 5 strikes, the server forcefully terminates the exam:
PYTHON
# backend/app/api/v1/exam_runner.py
@router.post("/attempts/{attempt_id}/violation", response_model=ApiResponse[Dict[str, Any]])
async def report_violation(
attempt_id: int,
payload: ViolationReportRequest,
db: AsyncSession = Depends(get_db),
current_user = Depends(require_roles(["ROLE_STUDENT", "ROLE_ADMIN"]))
):
attempt = await get_valid_in_progress_attempt(db, attempt_id, current_user.student_profile.id)
attempt.violations += 1
force_submitted = False
# Hard threshold: Auto-submit on 5th cheating infraction
if attempt.violations >= 5:
attempt.status = "force_finished"
attempt.finished_at = datetime.utcnow()
force_submitted = True
await db.commit()
# Instant alert to teacher proctoring dashboard
await ws_manager.broadcast_to_exam(
exam_id=attempt.examination_id,
event_type="VIOLATION_LOGGED",
data={
"studentId": current_user.student_profile.id,
"studentName": current_user.name,
"violationsCount": attempt.violations,
"violationType": payload.violation_type,
"forceSubmitted": force_submitted,
}
)
return ApiResponse.ok(
data={"violations": attempt.violations, "forceSubmitted": force_submitted},
message="Violation recorded"
)
4. Native ASGI WebSocket Architecture with ConnectionManager
Unlike synchronous frameworks that require separate WebSocket gateway services (such as Socket.IO microservices or Centrifugo), FastAPI supports native WebSockets directly on the ASGI event loop.
In backend/app/services/websocket_manager.py, we implement an in-memory room-based connection manager:
PYTHON
# backend/app/services/websocket_manager.py
import json
import logging
from typing import Dict, Set
from fastapi import WebSocket
logger = logging.getLogger(__name__)
class ConnectionManager:
def __init__(self):
# Maps examination_id -> Set of active supervisor WebSockets
self.active_rooms: Dict[int, Set[WebSocket]] = {}
async def connect(self, exam_id: int, websocket: WebSocket):
await websocket.accept()
if exam_id not in self.active_rooms:
self.active_rooms[exam_id] = set()
self.active_rooms[exam_id].add(websocket)
logger.info(f"Proctor connected to exam {exam_id}. Total: {len(self.active_rooms[exam_id])}")
def disconnect(self, exam_id: int, websocket: WebSocket):
if exam_id in self.active_rooms:
self.active_rooms[exam_id].discard(websocket)
if not self.active_rooms[exam_id]:
del self.active_rooms[exam_id]
async def broadcast_to_exam(self, exam_id: int, event_type: str, data: dict):
"""Broadcasts real-time events to all active teachers in the room"""
if exam_id not in self.active_rooms:
return
payload = {"eventType": event_type, "data": data}
message_str = json.dumps(payload, default=str)
dead_connections = set()
for connection in self.active_rooms[exam_id]:
try:
await connection.send_text(message_str)
except Exception as e:
logger.warning(f"Failed to push telemetry: {e}")
dead_connections.add(connection)
# Garbage collect dropped sockets
for dead in dead_connections:
self.disconnect(exam_id, dead)
ws_manager = ConnectionManager()
Horizontal Scaling with Redis Pub/Sub Backplane
In single-instance deployments, ConnectionManager runs entirely in memory with sub-millisecond dispatch times.
When scaling horizontally across multiple Uvicorn worker processes or Docker containers behind an Nginx load balancer, instances communicate through Redis Pub/Sub channels:
PYTHON
# Scaled architecture across multi-worker clusters:
async def publish_telemetry(exam_id: int, event_type: str, data: dict):
# 1. Publish to Redis channel
channel = f"exam:{exam_id}:telemetry"
await redis_client.publish(channel, json.dumps({"eventType": event_type, "data": data}))
# Background worker subscribing on each FastAPI node:
async def redis_listener(exam_id: int):
pubsub = redis_client.pubsub()
await pubsub.subscribe(f"exam:{exam_id}:telemetry")
async for msg in pubsub.listen():
if msg["type"] == "message":
# Broadcast to local WebSockets on this specific node
await ws_manager.broadcast_local(exam_id, msg["data"])
5. Teacher Live Proctoring Dashboard Integration
On the teacher's proctoring screen (frontend/src/pages/exams/ExamMonitor.tsx), incoming WebSocket messages trigger instantaneous reactive updates:
TYPESCRIPT
// frontend/src/pages/exams/ExamMonitor.tsx
useEffect(() => {
const ws = new WebSocket(`wss://${window.location.host}/ws/exams/${examId}/monitor`);
ws.onmessage = (event) => {
const payload = JSON.parse(event.data);
switch (payload.eventType) {
case "VIOLATION_LOGGED":
// Flash red badge and show audio/visual notification
updateCandidateViolation(payload.data.studentId, payload.data.violationsCount);
break;
case "ANSWER_SUBMITTED":
// Dynamically increment candidate's completion progress bar
updateCandidateProgress(payload.data.studentId, payload.data.totalAnswered);
break;
case "EXAM_FINISHED":
markCandidateCompleted(payload.data.studentId);
break;
}
};
return () => ws.close();
}, [examId]);
This transforms supervision from passive periodic checks to an active, real-time command center capable of monitoring hundreds of candidates with zero database polling pressure.
Key Takeaways for Engineering Leads
- Never trust client clocks: Calculate remaining time authoritatively from server-stamped UTC start dates on every interaction.
- Debounced autosave beats bulk submissions: Persisting individual answers eliminates WiFi disconnection panic and distributes database write loads evenly across the entire exam window.
- Set automated violation thresholds: Combining client-side
visibilitychangedetection with a server-side 5-strike auto-submit policy creates an effective, transparent deterrent against online cheating. - Leverage native ASGI WebSockets with dead connection cleanup: FastAPI manages thousands of persistent WebSocket channels with tiny RAM footprints (~15MB–25MB per 1,000 idle connections).
In Part 4, we tackle enterprise security architecture: Stateless JWTs, Role-Based Access Control (RBAC) via Dependency Injection, and 1-Click Administrative Impersonation with Zero-Reauth Session Recovery.
