Function withAdvisoryLockTransaction

packages/internal/src/db/transaction.ts:267–388

({
  callback,
  lockKey,
  context = {},
  logger,
  lockTimeoutMs = ADVISORY_LOCK_TIMEOUT_MS,
}: {
  callback: TransactionCallback<T>
  lockKey: string
  context: Record<string, unknown>
  logger: Logger
  lockTimeoutMs?: number
})

 * @returns Object containing the transaction result and lock wait time in milliseconds
 */
export async function withAdvisoryLockTransaction<T>({
  callback,
  lockKey,
  context = {},
  logger,
  lockTimeoutMs = ADVISORY_LOCK_TIMEOUT_MS,
}: {
  callback: TransactionCallback<T>
  lockKey: string
  context: Record<string, unknown>
  logger: Logger
  lockTimeoutMs?: number
}): Promise<AdvisoryLockTransactionResult<T>> {
  // Validate lock key to prevent bugs from null/empty keys
  if (!lockKey || typeof lockKey !== 'string' || lockKey.trim() === '') {
    throw new Error('lockKey must be a non-empty string')
  }

  return await withRetry(
    async () => {
      return await db.transaction(
        async (tx) => {
          // Set a statement timeout to prevent indefinite blocking if a lock holder hangs.
          // This timeout applies to the lock acquisition and subsequent statements.
          await tx.execute(
            sql`SET LOCAL statement_timeout = ${sql.raw(lockTimeoutMs.toString())}`,
          )

          // Acquire advisory lock - blocks until lock is available (or timeout).
          // We use MD5 to generate a 60-bit hash, dramatically reducing collision probability
          // compared to hashtext() which only produces 32 bits.
          // left(md5(key), 15) gives 15 hex chars (60 bits), which fits in a signed 64-bit bigint.
          const lockStart = Date.now()
          await tx.execute(
            sql`SELECT pg_advisory_xact_lock(('x' || left(md5(${lockKey}), 15))::bit(60)::bigint)`,
          )
          const lockWaitMs = Date.now() - lockStart

          // Log at WARN level only for significant waits (3+ seconds) to avoid excessive logging
          if (lockWaitMs > SIGNIFICANT_LOCK_WAIT_MS) {
            logger.warn(
              { ...context, lockKey, lockWaitMs },
              `Advisory lock contention: waited ${(lockWaitMs / 1000).toFixed(1)}s for lock`,
            )

            // Track in PostHog for analytics
            trackEvent({
              event: AnalyticsEvent.ADVISORY_LOCK_CONTENTION,
              userId: getUserIdForAnalytics(context, lockKey),
              properties: {
                ...context,
                lockKey,
                lockKeyType: lockKey.split(':')[0],
                lockWaitMs,
                lockWaitSeconds: lockWaitMs / 1000,
              },
              logger,
            })