Designing an Audit Logging System That Can Handle Millions of Writes

For financial systems, healthcare platforms, and regulated SaaS applications, audit logging is legally mandated (SOC2, HIPAA, GDPR).

Every meaningful state mutation must be recorded:

• Who performed the action (actor_id)?
• What changed (before_state vs after_state)?
• When did it happen (timestamp)?
• From where (ip_address)?

However, in high-throughput applications, inserting a synchronous audit row for every business action can double database write latency and quickly overwhelm storage tables with tens of millions of rows.

Here is how to design a scalable, tamper-evident audit logging architecture.

Designing Partitioned Audit Tables in PostgreSQL

CREATE TABLE audit_logs (
    id UUID DEFAULT gen_random_uuid(),
    actor_id VARCHAR(64) NOT NULL,
    action VARCHAR(64) NOT NULL,
    entity_type VARCHAR(64) NOT NULL,
    entity_id VARCHAR(64) NOT NULL,
    diff JSONB NOT NULL,
    ip_address INET,
    created_at TIMESTAMP WITH TIME ZONE NOT NULL,
    PRIMARY KEY (id, created_at)
) PARTITION BY RANGE (created_at);

CREATE TABLE audit_logs_2026_09 PARTITION OF audit_logs
    FOR VALUES FROM ('2026-09-01') TO ('2026-10-01');

CREATE TABLE audit_logs_2026_10 PARTITION OF audit_logs
    FOR VALUES FROM ('2026-10-01') TO ('2026-11-01');

CREATE INDEX idx_audit_entity ON audit_logs(entity_type, entity_id);

Why Partitioning Matters:

• Instant Deletion: When data retention policies require deleting 2-year-old audit logs, executing DROP TABLE audit_logs_2024_09; is an instantaneous $O(1)$ filesystem operation, compared to a devastating DELETE FROM audit_logs ... that locks tables and generates massive vacuum overhead.

Asynchronous High-Throughput Buffering in Python

import queue
import threading
import time
import json
from datetime import datetime

class AuditLogger:
    def __init__(self, db_engine, batch_size=500, flush_interval=2.0):
        self.queue = queue.Queue(maxsize=50000)
        self.db_engine = db_engine
        self.batch_size = batch_size
        self.flush_interval = flush_interval

        self.worker_thread = threading.Thread(target=self._flush_loop, daemon=True)
        self.worker_thread.start()

    def record_event(self, actor_id: str, action: str, entity_type: str, entity_id: str, diff: dict, ip: str):
        event = {
            "actor_id": actor_id,
            "action": action,
            "entity_type": entity_type,
            "entity_id": entity_id,
            "diff": json.dumps(diff),
            "ip_address": ip,
            "created_at": datetime.utcnow().isoformat()
        }
        try:
            self.queue.put_nowait(event)
        except queue.Full:
            print("ALERT: Audit queue full!")

    def _flush_loop(self):
        while True:
            batch = []
            start_time = time.time()
            while len(batch) < self.batch_size and (time.time() - start_time) < self.flush_interval:
                try:
                    item = self.queue.get(timeout=0.1)
                    batch.append(item)
                except queue.Empty:
                    break

            if batch:
                with self.db_engine.connect() as conn:
                    conn.execute(
                        """
                        INSERT INTO audit_logs (actor_id, action, entity_type, entity_id, diff, ip_address, created_at)
                        VALUES (:actor_id, :action, :entity_type, :entity_id, :diff, :ip_address, :created_at)
                        """,
                        batch
                    )
                    conn.commit()

Security: Enforcing Append-Only Immutability

Revoke UPDATE and DELETE permissions from the application database user:

REVOKE UPDATE, DELETE, TRUNCATE ON audit_logs FROM app_user;
GRANT INSERT, SELECT ON audit_logs TO app_user;

Even if an attacker gains SQL injection access through your web application, they cannot alter or erase previous audit logs to cover their tracks.