receta
Async

Concurrency Control

Managing parallel operations, rate limits, and performance optimization in async workflows.

Overview

Concurrency control lets you execute multiple async operations efficiently while respecting system constraints like API rate limits, memory usage, and CPU capacity.

FunctionPurposeReal-World Analogy
mapAsync(items, fn, { concurrency })Transform items in parallel with limitProcessing orders with N workers
filterAsync(items, predicate, { concurrency })Filter items using async predicateChecking inventory across warehouses
parallel(tasks)Run all tasks simultaneouslyFetching data from multiple APIs at once
sequential(tasks)Run tasks one after anotherProcessing payments in order

mapAsync - Controlled Parallel Mapping

When to use: You need to transform many items using async operations, but want to limit how many run at once.

Signature

Signature
function mapAsync<T, U>(
  items: readonly T[],
  fn: (item: T, index: number) => Promise<U>,
  options?: { concurrency?: number }
): Promise<Result<U[], MapAsyncError>>

Leave concurrency out and every item starts at once.

Real-World: GitHub Bulk Repository Fetch

Problem: Fetch data for 1000 repositories, but GitHub limits you to 5000 requests/hour.

import { mapAsync } from '@ontologyio/receta/async'

type Repo = {
  name: string
  stars: number
  language: string
}

// Fetch repos with concurrency limit to avoid rate limit
const fetchRepos = async (repoNames: string[]) => {
  return mapAsync(
    repoNames,
    async (name) => {
      const [owner, repo] = name.split('/')
      const res = await fetch(`https://api.github.com/repos/${owner}/${repo}`, {
        headers: { Authorization: `token ${GITHUB_TOKEN}` }
      })
      const data = await res.json()
      return {
        name: data.name,
        stars: data.stargazers_count,
        language: data.language
      }
    },
    { concurrency: 10 } // Max 10 requests at once
  )
}

// Usage
const repos = await fetchRepos([
  'facebook/react',
  'microsoft/typescript',
  'vercel/next.js',
  // ... 997 more repos
])

// repos is a Result: Ok with every repo, or Err with the failure

Why concurrency: 10?

  • GitHub allows 5000 requests/hour = ~83/minute = ~1.4/second
  • 10 concurrent requests is safe and fast
  • Too high (100+) risks hitting rate limit
  • Too low (1-2) wastes time

Real-World: Stripe Batch Payment Processing

Problem: Process 500 customer payments, but Stripe limits to 100 requests/second.

import { mapAsync, type MapAsyncError } from '@ontologyio/receta/async'
import { isOk, type Result } from '@ontologyio/receta/result'

type Payment = {
  customerId: string
  amount: number
  currency: string
}

type PaymentResult = {
  id: string
  status: 'succeeded' | 'failed'
  customerId: string
}

const processPayments = async (
  payments: Payment[]
): Promise<Result<PaymentResult[], MapAsyncError>> => {
  return mapAsync(
    payments,
    async (payment) => {
      const charge = await stripe.charges.create({
        amount: payment.amount,
        currency: payment.currency,
        customer: payment.customerId
      })
      return {
        id: charge.id,
        status: charge.status as 'succeeded' | 'failed',
        customerId: payment.customerId
      }
    },
    { concurrency: 50 } // 50 concurrent = safe for 100/sec limit
  )
}

// Usage
const payments: Payment[] = [
  { customerId: 'cus_123', amount: 1000, currency: 'usd' },
  { customerId: 'cus_456', amount: 2500, currency: 'usd' },
  // ... 498 more
]

const results = await processPayments(payments)

if (isOk(results)) {
  const succeeded = results.value.filter(r => r.status === 'succeeded')
  console.log(`Processed ${succeeded.length} payments`)
}

Real-World: Image Processing Pipeline

Problem: Resize 1000 product images, but each resize uses 100MB RAM.

import sharp from 'sharp'
import fs from 'fs/promises'

const resizeImages = async (imagePaths: string[]) => {
  return mapAsync(
    imagePaths,
    async (path) => {
      const outputPath = path.replace('.jpg', '-thumb.jpg')

      await sharp(path)
        .resize(300, 300, { fit: 'cover' })
        .jpeg({ quality: 80 })
        .toFile(outputPath)

      return outputPath
    },
    { concurrency: 4 } // Only 4 images at once to limit memory
  )
}

// Usage
const images = await fs.readdir('./products')
const thumbnails = await resizeImages(
  images.map(img => `./products/${img}`)
)

Why concurrency: 4?

  • Each resize uses ~100MB RAM
  • 4 concurrent = ~400MB total
  • Prevents OOM errors on low-memory servers
  • Adjust based on available RAM

Real-World: Database Bulk Operations

Problem: Update 10,000 user records, but database connection pool has 20 connections.

const updateUsers = async (updates: Array<{ id: string; name: string }>) => {
  return mapAsync(
    updates,
    async (update) => {
      return db.user.update({
        where: { id: update.id },
        data: { name: update.name }
      })
    },
    { concurrency: 15 } // Leave 5 connections for other queries
  )
}

Performance Characteristics

// Benchmark: Fetching 100 URLs
const urls = Array.from({ length: 100 }, (_, i) => `https://api.example.com/item/${i}`)
const get = (url: string) => fetch(url)

// No limit (concurrency: Infinity) - 2.3 seconds, might fail
await mapAsync(urls, get) // All 100 at once!

// concurrency: 50 - 2.5 seconds, stable
await mapAsync(urls, get, { concurrency: 50 })

// concurrency: 10 - 4.1 seconds, very stable
await mapAsync(urls, get, { concurrency: 10 })

// concurrency: 1 - 23 seconds, sequential
await mapAsync(urls, get, { concurrency: 1 })

Rule of Thumb: Start with concurrency = 10, then adjust based on:

  • API rate limits (lower)
  • Memory constraints (lower)
  • Network capacity (higher)
  • Need for speed (higher)

Handling Failures with Result

When to use: Every mapAsync call. It returns a Result, so a failure is a value you handle rather than an exception you catch.

Error Type

Signature
type MapAsyncError = {
  type: 'map_async_error'
  index: number
  item: unknown
  error: unknown
}

index and item say which item failed.

What Counts as a Failure

mapAsync fails when the mapper throws or rejects. A mapper that returns a Result hasn't failed, whatever that Result holds, so you choose how a failure behaves:

  • Stop at the first failure: let the mapper throw. Under a limit, no new items start after a failure.
  • Keep going and see every outcome: have the mapper return a Result, then use partition or collect on the array.

Real-World: Stripe Batch Payments

import { mapAsync, retry } from '@ontologyio/receta/async'
import { type Result, isOk, map, mapErr, partition } from '@ontologyio/receta/result'
import * as R from 'remeda'

type Payment = { id: string; amount: number; status: 'success' | 'failed' }
type PaymentError = { type: 'insufficient_funds' | 'network'; message: string }

const processPayment = async (
  customerId: string
): Promise<Result<Payment, PaymentError>> => {
  // Returns Result instead of throwing
  const result = await retry(() =>
    stripe.charges.create({ customer: customerId, amount: 1000 })
  )

  return R.pipe(
    result,
    map(charge => ({ id: charge.id, amount: charge.amount, status: 'success' as const })),
    mapErr(error => ({ type: 'network' as const, message: String(error.lastError) }))
  )
}

// Process payments with concurrency control
const outcome = await mapAsync(
  customerIds,
  processPayment,
  { concurrency: 5 }
)

// processPayment never throws, so every customer gets an answer
if (isOk(outcome)) {
  const [paid, failed] = partition(outcome.value)
  console.log(`${paid.length} paid, ${failed.length} failed`)
  // Retry just the failed payments
}

Comparison: Throwing Mapper vs Result Mapper

FeatureMapper throwsMapper returns Result
After a failureStops starting new itemsKeeps going
You getErr with the first failureOk with one Result per item
TypeResult<U[], MapAsyncError>Result<Result<U, E>[], MapAsyncError>
Error typeunknown, in error.errorYour own E
Use caseAll-or-nothing workPartial success

Real-World: GitHub Repository Validation

type RepoData = { name: string; stars: number; language: string }
type RepoError = { type: 'not_found' | 'rate_limit' | 'network'; repo: string }

const fetchRepo = async (repoName: string): Promise<Result<RepoData, RepoError>> => {
  try {
    const [owner, repo] = repoName.split('/')
    const res = await fetch(`https://api.github.com/repos/${owner}/${repo}`, {
      headers: { Authorization: `token ${GITHUB_TOKEN}` }
    })

    if (res.status === 404) {
      return err({ type: 'not_found', repo: repoName })
    }

    if (res.status === 429) {
      return err({ type: 'rate_limit', repo: repoName })
    }

    if (!res.ok) {
      return err({ type: 'network', repo: repoName })
    }

    const data = await res.json()
    return ok({
      name: data.name,
      stars: data.stargazers_count,
      language: data.language
    })
  } catch {
    return err({ type: 'network', repo: repoName })
  }
}

// Fetch multiple repos with Result handling
const outcome = await mapAsync(
  ['facebook/react', 'microsoft/typescript', 'invalid/repo'],
  fetchRepo,
  { concurrency: 10 }
)

if (isOk(outcome)) {
  // collect gives every repo, or the first failure
  const repos = collect(outcome.value)

  if (isErr(repos)) {
    const { repo, type } = repos.error
    console.error(`Failed at ${repo} (${type})`)
  } else {
    console.log(`Successfully fetched ${repos.value.length} repositories`)
  }
}

Real-World: File Processing Pipeline

import { isErr, isOk, ok, partition, tryCatchAsync, type Result } from '@ontologyio/receta/result'
type ProcessedFile = { path: string; size: number; hash: string }
type FileError = { type: 'read' | 'parse' | 'write'; path: string; message: string }

const processFile = async (path: string): Promise<Result<ProcessedFile, FileError>> => {
  // Read file
  const readResult = await tryCatchAsync(
    () => fs.readFile(path),
    (error) => ({ type: 'read' as const, path, message: String(error) })
  )

  if (isErr(readResult)) return readResult

  // Process file
  const processResult = await tryCatchAsync(
    () => processData(readResult.value),
    (error) => ({ type: 'parse' as const, path, message: String(error) })
  )

  if (isErr(processResult)) return processResult

  // Return result
  return ok({
    path,
    size: readResult.value.length,
    hash: processResult.value.hash
  })
}

// Process files with concurrency
const outcome = await mapAsync(
  filePaths,
  processFile,
  { concurrency: 4 }
)

// Handle errors gracefully
if (isOk(outcome)) {
  const [processed, failed] = partition(outcome.value)

  // Log the failures and keep what worked
  failed.forEach(({ path, message }) => console.error(`Failed to process ${path}:`, message))

  console.log(`Processed ${processed.length} files`)
  console.log(`Total size: ${processed.reduce((sum, f) => sum + f.size, 0)} bytes`)
}

Composing with Result Functions

Because mapAsync returns a Result, it composes naturally with other Result functions:

import { collect, flatMap, isErr, map } from '@ontologyio/receta/result'
import { mapAsync } from '@ontologyio/receta/async'
import * as R from 'remeda'

// Complex pipeline with Result composition
const processOrders = async (orderIds: string[]) => {
  const payments = R.pipe(
    // Fetch orders (returns Result)
    await mapAsync(orderIds, fetchOrder, { concurrency: 10 }),

    // Validate orders
    flatMap(orders =>
      collect(orders.map(validateOrder))
    ),

    // Transform to payment requests
    map(orders =>
      orders.map(order => ({ customerId: order.customerId, amount: order.total }))
    )
  )

  if (isErr(payments)) return payments

  // Process payments (returns Result)
  return mapAsync(payments.value, processPayment, { concurrency: 5 })
}

// Single error-handling point
const result = await processOrders(['order-1', 'order-2', 'order-3'])

if (isErr(result)) {
  console.error('Pipeline failed:', result.error)
} else {
  console.log(`Successfully processed ${result.value.length} orders`)
}

Why a Result

  1. No try-catch: Errors are values, not exceptions
  2. Type-safe: Result<U[], MapAsyncError> says the call can fail
  3. Detailed errors: Under a limit, know which item failed and why
  4. Composable: Works with all Result functions
  5. Fail-fast: Under a limit, stops starting new items after a failure

filterAsync - Async Filtering

When to use: Filter items using an async condition (like checking if file exists, or if product is in stock).

Signature

Signature
function filterAsync<T>(
  items: readonly T[],
  predicate: (item: T, index: number) => Promise<boolean>,
  options?: { concurrency?: number }
): Promise<Result<T[], FilterAsyncError>>

A predicate that throws gives an Err whose underlyingError is the MapAsyncError.

Real-World: Check Inventory Across Warehouses

Problem: Filter products that are in stock by checking multiple warehouse APIs.

import { filterAsync } from '@ontologyio/receta/async'

type Product = {
  id: string
  name: string
  sku: string
}

const isInStock = async (product: Product): Promise<boolean> => {
  // Check all warehouses in parallel
  const warehouses = ['US-EAST', 'US-WEST', 'EU-CENTRAL']

  const results = await Promise.all(
    warehouses.map(async (warehouse) => {
      const res = await fetch(
        `https://warehouse.api.com/${warehouse}/stock/${product.sku}`
      )
      const data = await res.json()
      return data.quantity > 0
    })
  )

  return results.some(inStock => inStock) // In stock if ANY warehouse has it
}

const filterInStockProducts = async (products: Product[]) => {
  return filterAsync(
    products,
    isInStock,
    { concurrency: 20 } // Check 20 products at once
  )
}

// Usage
const allProducts: Product[] = [
  { id: '1', name: 'iPhone', sku: 'IPHONE-14' },
  { id: '2', name: 'MacBook', sku: 'MACBOOK-PRO' },
  // ... 998 more
]

const inStock = await filterInStockProducts(allProducts)

if (isOk(inStock)) {
  console.log(`${inStock.value.length} products available`)
}

Real-World: Validate S3 Objects Exist

Problem: Filter a list of S3 keys to only those that exist.

import { S3Client, HeadObjectCommand } from '@aws-sdk/client-s3'

const s3 = new S3Client({ region: 'us-east-1' })

const objectExists = async (key: string): Promise<boolean> => {
  try {
    await s3.send(new HeadObjectCommand({
      Bucket: 'my-bucket',
      Key: key
    }))
    return true
  } catch {
    return false
  }
}

const filterExistingFiles = async (keys: string[]) => {
  return filterAsync(
    keys,
    objectExists,
    { concurrency: 30 } // S3 can handle high concurrency
  )
}

// Usage
const requestedFiles = [
  'uploads/avatar-123.jpg',
  'uploads/avatar-456.jpg',
  'uploads/deleted-file.jpg', // Doesn't exist
  // ... more files
]

const existingFiles = await filterExistingFiles(requestedFiles)

Real-World: Find Valid Email Addresses

Problem: Filter email list by actually checking if addresses are deliverable.

const isValidEmail = async (email: string): Promise<boolean> => {
  try {
    // Use email validation API
    const res = await fetch(
      `https://api.emailvalidation.com/verify?email=${email}`,
      { headers: { 'X-API-Key': API_KEY } }
    )
    const data = await res.json()
    return data.status === 'deliverable'
  } catch {
    return false
  }
}

const filterValidEmails = async (emails: string[]) => {
  return filterAsync(
    emails,
    isValidEmail,
    { concurrency: 5 } // Respect API rate limit
  )
}

// Usage
const emailList = ['user1@example.com', 'fake@invalid.xyz', /* ... */]
const validEmails = await filterValidEmails(emailList)

parallel - Run All Tasks Together

When to use: You have independent tasks that should all run simultaneously (fetching from multiple APIs, running independent calculations).

Signature

Signature
function parallel<T>(
  tasks: ReadonlyArray<() => Promise<T>>,
  options?: { concurrency?: number }
): Promise<Result<T[], MapAsyncError>>

parallel is mapAsync over a list of functions, so it takes the same concurrency limit. Every task shares one type T, and the values come back as an array inside the Result.

Real-World: Dashboard Data Loading

Problem: Load dashboard data from 5 different sources - should all load at once.

import { parallel } from '@ontologyio/receta/async'

type DashboardData = {
  userStats: { total: number; active: number }
  recentOrders: Order[]
  revenue: { today: number; month: number }
  notifications: Notification[]
  systemHealth: { status: 'healthy' | 'degraded' }
}

const loadDashboard = async () => {
  const loaded = await parallel([
    () => fetch('/api/users/stats').then(r => r.json()),
    () => fetch('/api/orders/recent').then(r => r.json()),
    () => fetch('/api/revenue').then(r => r.json()),
    () => fetch('/api/notifications').then(r => r.json()),
    () => fetch('/api/health').then(r => r.json())
  ])

  return map(
    loaded,
    ([userStats, recentOrders, revenue, notifications, systemHealth]): DashboardData => ({
      userStats, recentOrders, revenue, notifications, systemHealth
    })
  )
}

// Usage - All 5 APIs load in parallel
const data = await loadDashboard()
// Total time = slowest API (not sum of all APIs)

Real-World: Multi-Region Deployment

Problem: Deploy to 3 cloud regions simultaneously.

const deployToRegion = async (region: string, artifact: string) => {
  console.log(`Deploying to ${region}...`)
  // Upload artifact, update servers, health check
  await uploadArtifact(region, artifact)
  await updateServers(region)
  await healthCheck(region)
  console.log(`${region} deployed!`)
}

const deployGlobally = async (artifact: string) => {
  const regions = ['us-east-1', 'eu-west-1', 'ap-southeast-1']

  const deployed = await parallel(
    regions.map(region => () => deployToRegion(region, artifact))
  )

  if (isOk(deployed)) {
    console.log('Global deployment complete!')
  }
}

// All regions deploy at the same time
await deployGlobally('v2.3.0')

Real-World: Microservices Health Check

Problem: Check health of 10 microservices in parallel.

type ServiceHealth = {
  name: string
  status: 'healthy' | 'unhealthy'
  responseTime: number
}

const checkService = async (name: string, url: string): Promise<ServiceHealth> => {
  const start = Date.now()

  try {
    const res = await fetch(`${url}/health`, {
      signal: AbortSignal.timeout(5000)
    })
    const responseTime = Date.now() - start

    return {
      name,
      status: res.ok ? 'healthy' : 'unhealthy',
      responseTime
    }
  } catch {
    return { name, status: 'unhealthy', responseTime: Date.now() - start }
  }
}

const checkAllServices = async () => {
  const services = {
    auth: 'https://auth.api.com',
    payments: 'https://payments.api.com',
    inventory: 'https://inventory.api.com',
    shipping: 'https://shipping.api.com',
    notifications: 'https://notifications.api.com'
  }

  const results = await parallel(
    Object.entries(services).map(([name, url]) =>
      () => checkService(name, url)
    )
  )

  return results
}

// Usage
const health = await checkAllServices()

if (isOk(health)) {
  health.value.forEach(({ name, status, responseTime }) => {
    console.log(`${name}: ${status} (${responseTime}ms)`)
  })
}

Performance Characteristics

// Sequential (slow)
const slow = async () => {
  const a = await fetchA() // 1 second
  const b = await fetchB() // 1 second
  const c = await fetchC() // 1 second
  return [a, b, c]
}
await slow() // Total: 3 seconds

// Parallel (fast)
const fast = async () => {
  return parallel([fetchA, fetchB, fetchC])
}
await fast() // Total: 1 second (all run at once)

sequential - Run Tasks In Order

When to use: Tasks depend on each other, or you need guaranteed ordering (payment processing, database migrations).

Signature

Signature
function sequential<T>(
  tasks: ReadonlyArray<() => Promise<T>>
): Promise<Result<T[], SequentialError>>

The first task to fail stops the rest. SequentialError holds its taskIndex, the completedTasks count, and the error.

Real-World: Payment Processing Flow

Problem: Process payment steps in exact order - can't capture before authorization!

import { sequential } from '@ontologyio/receta/async'

type PaymentFlow = {
  authorized: AuthResult
  captured: CaptureResult
  receipt: ReceiptResult
  notification: NotificationResult
}

const processPayment = async (
  customerId: string,
  amount: number
) => {
  const steps = await sequential([
    // Step 1: Authorize card
    async () => {
      console.log('Authorizing...')
      return stripe.authorizations.create({
        customer: customerId,
        amount
      })
    },

    // Step 2: Capture funds (only after authorization!)
    async () => {
      console.log('Capturing...')
      return stripe.captures.create({ /* ... */ })
    },

    // Step 3: Generate receipt
    async () => {
      console.log('Generating receipt...')
      return generateReceipt(customerId, amount)
    },

    // Step 4: Send notification
    async () => {
      console.log('Sending notification...')
      return sendEmail(customerId, 'Payment received')
    }
  ])

  return map(
    steps,
    ([authorized, captured, receipt, notification]): PaymentFlow => ({
      authorized, captured, receipt, notification
    })
  )
}

// Each step waits for previous to complete
await processPayment('cus_123', 5000)

Real-World: Database Migration Pipeline

Problem: Run migrations in order - migration 003 depends on 002.

type Migration = {
  version: string
  name: string
  sql: string
}

const migrations: Migration[] = [
  { version: '001', name: 'create_users', sql: 'CREATE TABLE users...' },
  { version: '002', name: 'add_email_index', sql: 'CREATE INDEX...' },
  { version: '003', name: 'add_users_status', sql: 'ALTER TABLE users...' }
]

const runMigration = async (migration: Migration) => {
  console.log(`Running migration ${migration.version}: ${migration.name}`)
  await db.execute(migration.sql)
  await db.migrations.insert({ version: migration.version })
  console.log(`✓ Completed ${migration.version}`)
}

const migrate = async () => {
  const applied = await sequential(
    migrations.map(m => () => runMigration(m))
  )

  if (isErr(applied)) {
    console.error(`Stopped at migration ${applied.error.taskIndex}`)
  } else {
    console.log('All migrations complete!')
  }
}

// Runs in exact order: 001 → 002 → 003
await migrate()

Real-World: Video Processing Pipeline

Problem: Process video in stages - can't add subtitles before transcoding!

const processVideo = async (videoId: string) => {
  const stages = await sequential([
    // Step 1: Transcode to different resolutions
    async () => {
      console.log('Transcoding video...')
      return transcode(videoId, ['720p', '1080p', '4k'])
    },

    // Step 2: Generate thumbnail from transcoded video
    async () => {
      console.log('Generating thumbnail...')
      return generateThumbnail(videoId)
    },

    // Step 3: Generate subtitles
    async () => {
      console.log('Generating subtitles...')
      return generateSubtitles(videoId)
    },

    // Step 4: Upload everything to CDN
    async () => {
      console.log('Uploading to CDN...')
      return uploadToCDN(videoId)
    }
  ])

  return map(stages, ([transcoded, thumbnail, subtitles, uploaded]) => ({
    transcoded, thumbnail, subtitles, uploaded
  }))
}

Concurrency Patterns

Pattern 1: Rate-Limited API Calls

Problem: GitHub allows 5000 requests/hour. Need to fetch 1000 repos.

import { mapAsync } from '@ontologyio/receta/async'

const GITHUB_RATE_LIMIT = 5000 // requests per hour
const SAFE_CONCURRENCY = 10    // requests at once

const fetchGitHubRepos = async (repos: string[]) => {
  return mapAsync(
    repos,
    async (repo) => {
      const res = await fetch(`https://api.github.com/repos/${repo}`)
      return res.json()
    },
    { concurrency: SAFE_CONCURRENCY }
  )
}

// Alternative: Add delay between requests
const fetchWithDelay = async (url: string) => {
  await new Promise(resolve => setTimeout(resolve, 100)) // 100ms delay
  return fetch(url).then(r => r.json())
}

Pattern 2: Memory-Constrained Processing

Problem: Process 1000 large files, but only 2GB RAM available.

import { mapAsync } from '@ontologyio/receta/async'

const processLargeFiles = async (files: string[]) => {
  // Each file uses ~500MB when processing
  // 2GB / 500MB = 4 files max
  return mapAsync(
    files,
    async (file) => {
      const data = await readLargeFile(file) // 500MB
      const processed = await processData(data)
      await writeOutput(processed)
      return processed
    },
    { concurrency: 3 } // Leave some RAM headroom
  )
}

Pattern 3: Batch Processing with Progress

Problem: Process 10,000 items and show progress.

const processBatch = async <T, U>(
  items: T[],
  fn: (item: T) => Promise<U>,
  onProgress: (completed: number, total: number) => void
) => {
  let completed = 0

  const results = await mapAsync(
    items,
    async (item) => {
      const result = await fn(item)
      completed++
      onProgress(completed, items.length)
      return result
    },
    { concurrency: 20 }
  )

  return results
}

// Usage
await processBatch(
  users,
  updateUser,
  (completed, total) => {
    console.log(`Progress: ${completed}/${total} (${(completed/total*100).toFixed(1)}%)`)
  }
)

Pattern 4: Retry with Exponential Backoff

Problem: API calls sometimes fail - retry with increasing delays.

import { mapAsync, retry } from '@ontologyio/receta/async'

const fetchWithRetry = async (urls: string[]) => {
  return mapAsync(
    urls,
    (url) =>
      retry(() => fetch(url).then(r => r.json()), {
        maxAttempts: 4,
        delay: 1000, // 1s, 2s, 4s
        backoff: 2,
        onRetry: (_error, attempt, delay) => {
          console.log(`Retry ${attempt}/3 after ${delay}ms`)
        }
      }),
    { concurrency: 10 }
  )
}

Pattern 5: Priority Queue Processing

Problem: Process high-priority items first, then low-priority.

type Task = {
  id: string
  priority: 'high' | 'low'
  fn: () => Promise<void>
}

const processByPriority = async (tasks: Task[]) => {
  const high = tasks.filter(t => t.priority === 'high')
  const low = tasks.filter(t => t.priority === 'low')

  // Process high priority first (parallel)
  await mapAsync(high, t => t.fn(), { concurrency: 10 })

  // Then process low priority
  await mapAsync(low, t => t.fn(), { concurrency: 5 })
}

Comparison Table

FunctionExecutionUse CaseConcurrency ControlError Handling
mapAsyncParallel with limitTransform many items✅ Yes (configurable)Result<U[], MapAsyncError>
filterAsyncParallel with limitFilter with async check✅ Yes (configurable)Result<T[], FilterAsyncError>
parallelAll at once, or with limitIndependent tasks✅ Yes (configurable)Result<T[], MapAsyncError>
sequentialOne by oneDependent tasks❌ No (always sequential)Result<T[], SequentialError>

When to Use Each

// mapAsync - Transform with rate limit
const results = await mapAsync(items, fetchData, { concurrency: 10 })

// filterAsync - Filter with async validation
const valid = await filterAsync(items, isValid, { concurrency: 20 })

// parallel - Load independent data
const loaded = await parallel([
  fetchUsers,
  fetchProducts,
  fetchOrders
])

// sequential - Steps depend on each other
const steps = await sequential([
  authenticate,
  fetchData,
  processData
])

Common Pitfalls and Solutions

Pitfall 1: Too High Concurrency

// ❌ Bad - Will hit rate limit
await mapAsync(repos, fetchRepo) // Default: unlimited!

// ✅ Good - Respect rate limit
await mapAsync(repos, fetchRepo, { concurrency: 10 })

Pitfall 2: Sequential When Should Parallel

// ❌ Bad - Slow (3 seconds total)
const a = await fetchA() // 1 second
const b = await fetchB() // 1 second
const c = await fetchC() // 1 second

// ✅ Good - Fast (1 second total)
const results = await parallel([fetchA, fetchB, fetchC])

Pitfall 3: Parallel When Should Sequential

// ❌ Bad - Might capture before authorization completes!
const results = await parallel([
  authorizeCard,
  capturePayment
])

// ✅ Good - Capture only after authorization
const results = await sequential([
  authorizeCard,
  capturePayment
])

Pitfall 4: Not Handling Errors

// ❌ Bad - One failure stops everything
await mapAsync(items, processItem)

// ✅ Good - Each item gets its own Result
await mapAsync(items, (item) => tryCatchAsync(() => processItem(item)))

Pitfall 5: Memory Leak in Long-Running Tasks

// ❌ Bad - Loads all files into memory at once
const files = await Promise.all(filePaths.map(readFile))

// ✅ Good - Process in batches
const processInBatches = async (items: string[], batchSize = 100) => {
  for (let i = 0; i < items.length; i += batchSize) {
    const batch = items.slice(i, i + batchSize)
    await mapAsync(batch, processFile, { concurrency: 10 })
  }
}

Performance Guidelines

Choosing the Right Concurrency Limit

// API Rate Limits
const concurrency = {
  github: 10,        // 5000/hour = ~1.4/second, 10 is safe
  stripe: 50,        // 100/second, 50 is safe
  twitter: 5,        // Strict limits
  internal: 100      // Your own API, can be higher
}

// System Resources
const concurrency = {
  imageProcessing: 4,   // CPU-intensive
  fileIO: 20,           // I/O bound
  database: 15,         // Based on connection pool size
  network: 50           // Network I/O
}

// Rule of thumb
const safeConcurrency = Math.min(
  rateLimit / 2,           // Half of rate limit
  connectionPoolSize - 5,  // Leave connections for other tasks
  availableRAM / taskMemory // Don't exceed RAM
)

Benchmarking Your Operations

const benchmark = async () => {
  const items = Array.from({ length: 100 }, (_, i) => i)

  const concurrencies = [1, 5, 10, 20, 50, 100]

  for (const concurrency of concurrencies) {
    const start = Date.now()

    await mapAsync(
      items,
      async (item) => {
        await new Promise(resolve => setTimeout(resolve, 100))
        return item * 2
      },
      { concurrency }
    )

    const duration = Date.now() - start
    console.log(`Concurrency ${concurrency}: ${duration}ms`)
  }
}

// Results:
// Concurrency 1: 10,000ms
// Concurrency 5: 2,000ms
// Concurrency 10: 1,000ms ← Sweet spot
// Concurrency 20: 500ms
// Concurrency 50: 200ms
// Concurrency 100: 100ms ← Diminishing returns

Memory Usage Monitoring

const monitorMemory = async () => {
  const before = process.memoryUsage()

  await mapAsync(
    largeDataset,
    processItem,
    { concurrency: 10 }
  )

  const after = process.memoryUsage()
  const diff = after.heapUsed - before.heapUsed

  console.log(`Memory used: ${(diff / 1024 / 1024).toFixed(2)}MB`)

  // If too high, reduce concurrency
}

Optimal Concurrency Formula

const calculateOptimalConcurrency = (params: {
  rateLimit?: number        // Requests per second allowed
  taskDuration?: number     // Average task duration in ms
  memoryPerTask?: number    // MB per task
  availableMemory?: number  // Total MB available
  poolSize?: number         // Connection pool size
}) => {
  const limits = [
    params.rateLimit ? params.rateLimit * 0.8 : Infinity,
    params.availableMemory && params.memoryPerTask
      ? Math.floor(params.availableMemory / params.memoryPerTask)
      : Infinity,
    params.poolSize ? params.poolSize * 0.8 : Infinity,
    100 // Never go above 100
  ]

  return Math.floor(Math.min(...limits))
}

// Example usage
const concurrency = calculateOptimalConcurrency({
  rateLimit: 100,        // 100 req/sec
  memoryPerTask: 50,     // 50MB per task
  availableMemory: 1000, // 1GB available
  poolSize: 20           // 20 connections
})

console.log(`Optimal concurrency: ${concurrency}`)
// Output: 16 (limited by poolSize: 20 * 0.8 = 16)

Real-World Example: E-Commerce Order Processing

Putting it all together - processing orders with multiple async steps:

import { mapAsync, sequential, parallel } from '@ontologyio/receta/async'

type Order = {
  id: string
  items: Array<{ productId: string; quantity: number }>
  customerId: string
  total: number
}

const processOrder = async (order: Order) => {
  // Sequential: Must happen in order
  const steps = await sequential([
    // Step 1: Validate inventory
    async () => {
      console.log(`Validating order ${order.id}`)
      const inStock = await validateInventory(order.items)
      if (!inStock) throw new Error('Out of stock')
      return true
    },

    // Step 2: Charge customer (only after validation!)
    async () => {
      console.log(`Charging customer for order ${order.id}`)
      return stripe.charges.create({
        customer: order.customerId,
        amount: order.total
      })
    },

    // Step 3: Fulfill order (only after payment!)
    async () => {
      console.log(`Fulfilling order ${order.id}`)

      // Parallel: These can happen at the same time
      const fulfilled = await parallel([
        () => createShipment(order),
        () => generateReceipt(order),
        () => sendOrderConfirmation(order.customerId)
      ])

      // Throw so sequential sees the failure and stops
      if (isErr(fulfilled)) throw fulfilled.error

      const [shipment, receipt, notification] = fulfilled.value
      return { shipment, receipt, notification }
    }
  ])

  // Throw so mapAsync sees the failure
  if (isErr(steps)) throw steps.error

  const [validated, charged, fulfilled] = steps.value
  return { validated, charged, fulfilled }
}

const processOrders = async (orders: Order[]) => {
  // Process multiple orders concurrently (but each order is sequential internally)
  return mapAsync(
    orders,
    processOrder,
    { concurrency: 10 } // Process 10 orders at once
  )
}

// Usage
const orders: Order[] = [/* ... 100 orders ... */]
const results = await processOrders(orders)

if (isOk(results)) {
  console.log(`Processed ${results.value.length} orders`)
}

Next Steps

Overview

Async utilities provide battle-tested patterns for handling concurrent operations, rate limits, retries, and timeouts—making your async code reliable without manual Promise juggling.

Retry and Timeout Patterns

Next Page

On this page

OverviewmapAsync - Controlled Parallel MappingSignatureReal-World: GitHub Bulk Repository FetchReal-World: Stripe Batch Payment ProcessingReal-World: Image Processing PipelineReal-World: Database Bulk OperationsPerformance CharacteristicsHandling Failures with ResultError TypeWhat Counts as a FailureReal-World: Stripe Batch PaymentsComparison: Throwing Mapper vs Result MapperReal-World: GitHub Repository ValidationReal-World: File Processing PipelineComposing with Result FunctionsWhy a ResultfilterAsync - Async FilteringSignatureReal-World: Check Inventory Across WarehousesReal-World: Validate S3 Objects ExistReal-World: Find Valid Email Addressesparallel - Run All Tasks TogetherSignatureReal-World: Dashboard Data LoadingReal-World: Multi-Region DeploymentReal-World: Microservices Health CheckPerformance Characteristicssequential - Run Tasks In OrderSignatureReal-World: Payment Processing FlowReal-World: Database Migration PipelineReal-World: Video Processing PipelineConcurrency PatternsPattern 1: Rate-Limited API CallsPattern 2: Memory-Constrained ProcessingPattern 3: Batch Processing with ProgressPattern 4: Retry with Exponential BackoffPattern 5: Priority Queue ProcessingComparison TableWhen to Use EachCommon Pitfalls and SolutionsPitfall 1: Too High ConcurrencyPitfall 2: Sequential When Should ParallelPitfall 3: Parallel When Should SequentialPitfall 4: Not Handling ErrorsPitfall 5: Memory Leak in Long-Running TasksPerformance GuidelinesChoosing the Right Concurrency LimitBenchmarking Your OperationsMemory Usage MonitoringOptimal Concurrency FormulaReal-World Example: E-Commerce Order ProcessingNext Steps