Concurrent Operation Management

Techniques and patterns for safely handling simultaneous operations in systems where multiple processes or users may attempt to perform conflicting actions. Critical for data integrity and system stability in production applications.

Core Challenges

Race Conditions

Problem: Multiple operations accessing shared resources simultaneously Symptoms:

• Data corruption or inconsistent states
• Duplicate records creation
• Lost updates or partial writes
• System crashes under concurrent load

Example Scenario: Two users triggering data sync simultaneously

// Vulnerable: No concurrency control
export async function POST() {
  const data = await fetchExternalData();  // Long operation
  await clearDatabase();                   // Destructive operation
  await insertData(data);                  // State modification
  return success();
}

Resource Contention

Issues:

• Database connection exhaustion
• File system conflicts
• API rate limit violations
• Memory resource competition

Locking Mechanisms

Application-Level Locks

Mutex (Mutual Exclusion):

const syncMutex = new Mutex();

export async function POST() {
  const release = await syncMutex.acquire();
  try {
    await performSyncOperation();
  } finally {
    release();
  }
}

Semaphore: Control concurrent operation count

const syncSemaphore = new Semaphore(2); // Allow 2 concurrent operations

export async function POST() {
  await syncSemaphore.acquire();
  try {
    await performLimitedOperation();
  } finally {
    syncSemaphore.release();
  }
}

Database-Level Locks

Row-Level Locking:

SELECT * FROM products WHERE id = ? FOR UPDATE;
-- Exclusive lock until transaction commits

Table-Level Locking:

LOCK TABLE sync_operations WRITE;
-- Prevent concurrent table access

Advisory Locks:

SELECT pg_advisory_lock(12345);
-- Application-controlled database lock

Distributed Locking

Redis-Based Locks

const redlock = new Redlock([redis], {
  driftFactor: 0.01,
  retryCount: 10,
  retryDelay: 200
});

const lock = await redlock.acquire(['sync:operation'], 5000);
try {
  await performDistributedOperation();
} finally {
  await lock.release();
}

Database Advisory Locks

async function withPostgresLock(key: string, operation: () => Promise<void>) {
  await db.query('SELECT pg_advisory_lock($1)', [key]);
  try {
    await operation();
  } finally {
    await db.query('SELECT pg_advisory_unlock($1)', [key]);
  }
}

Atomic Operations

Database Transactions

async function atomicDataUpdate() {
  const transaction = await db.beginTransaction();
  try {
    await transaction.query('DELETE FROM old_data');
    await transaction.query('INSERT INTO new_data VALUES (...)');
    await transaction.commit();
  } catch (error) {
    await transaction.rollback();
    throw error;
  }
}

Optimistic Concurrency Control

interface VersionedRecord {
  id: string;
  data: any;
  version: number;
}

async function updateWithVersion(record: VersionedRecord) {
  const result = await db.query(`
    UPDATE records 
    SET data = $1, version = version + 1 
    WHERE id = $2 AND version = $3
  `, [record.data, record.id, record.version]);
  
  if (result.rowCount === 0) {
    throw new Error('Record was modified by another operation');
  }
}

Queue-Based Solutions

Operation Queuing

Message Queues: Sequential operation processing

// Producer: Add operations to queue
await messageQueue.enqueue('sync-operation', {
  type: 'data-sync',
  source: 'easybeer',
  timestamp: new Date()
});

// Consumer: Process operations sequentially
messageQueue.on('sync-operation', async (job) => {
  await performSyncOperation(job.data);
});

Job Scheduling: Distributed task management

const queue = new Bull('sync operations');

queue.process('data-sync', async (job) => {
  return await syncData(job.data);
});

// Add job to queue instead of immediate execution
await queue.add('data-sync', { source: 'easybeer' });

API Design Patterns

Idempotent Operations

Design APIs to be safely retryable:

// Idempotent: Same result regardless of retry count
export async function POST(request: Request) {
  const { operationId } = await request.json();
  
  // Check if operation already completed
  const existing = await db.findOperation(operationId);
  if (existing) {
    return Response.json(existing.result);
  }
  
  const result = await performOperation(operationId);
  await db.saveOperation(operationId, result);
  return Response.json(result);
}

Operation Status Tracking

async function startLongRunningOperation() {
  const operationId = generateId();
  
  // Set initial status
  await setOperationStatus(operationId, 'running');
  
  // Perform operation asynchronously
  performOperation(operationId)
    .then(() => setOperationStatus(operationId, 'completed'))
    .catch(error => setOperationStatus(operationId, 'failed', error));
  
  return { operationId, status: 'running' };
}

Monitoring and Detection

Concurrent Operation Detection

const runningOperations = new Set<string>();

export async function POST(request: Request) {
  const operationType = getOperationType(request);
  
  if (runningOperations.has(operationType)) {
    return Response.json(
      { error: 'Operation already in progress' }, 
      { status: 409 }
    );
  }
  
  runningOperations.add(operationType);
  try {
    return await performOperation();
  } finally {
    runningOperations.delete(operationType);
  }
}

Health Monitoring

• Lock Contention: Monitor waiting times
• Deadlock Detection: Identify circular dependencies
• Operation Duration: Track long-running operations
• Resource Usage: Monitor connection pools and memory

Best Practices

Design Principles

• Fail Fast: Quick error detection and recovery
• Graceful Degradation: Reduced functionality over total failure
• Timeout Handling: Prevent indefinite blocking
• Error Recovery: Automatic retry and rollback mechanisms

Implementation Guidelines

• Minimize Lock Scope: Hold locks for shortest time possible
• Avoid Nested Locks: Prevent deadlock scenarios
• Use Timeouts: Prevent indefinite waiting
• Log Lock Events: Monitor for debugging and optimization

Testing Strategies

• Load Testing: Verify behavior under concurrent load
• Race Condition Testing: Intentionally create conflicts
• Stress Testing: Test resource exhaustion scenarios
• Recovery Testing: Verify error handling and cleanup

See also

• Database Consistency
• Distributed Systems
• api-security
• System Reliability