notion-load-scale

What this skill does

# Notion Load & Scale

## Overview

Patterns for high-volume Notion API usage within the 3 requests/second rate limit. Covers parallel request orchestration with `p-queue`, worker queue architecture for background processing, full database pagination at scale (100K+ records), incremental sync using `last_edited_time` filters to avoid re-fetching unchanged data, and memory management for bulk operations using streaming and chunked processing.

## Prerequisites

- `@notionhq/client` v2.x installed (`npm install @notionhq/client`)
- `p-queue` for rate-limited concurrency (`npm install p-queue`)
- Python: `notion-client` installed (`pip install notion-client`)
- `NOTION_TOKEN` set (each token gets its own 3 req/s limit)
- Test database in Notion (dedicated for load testing)

## Instructions

### Step 1: Parallel Requests Within Rate Limits

Notion enforces 3 requests/second per integration token. Use `p-queue` to maximize throughput without hitting 429 errors.

```typescript
import { Client } from '@notionhq/client';
import PQueue from 'p-queue';

const notion = new Client({ auth: process.env.NOTION_TOKEN });

// Rate-limited queue: 3 requests per second, single concurrency
// Use intervalCap + interval instead of concurrency alone
const apiQueue = new PQueue({
  concurrency: 1,
  interval: 340,      // ~3 per second with safety margin
  intervalCap: 1,
});

// Metrics tracking
let totalRequests = 0;
let rateLimitHits = 0;
const startTime = Date.now();

function logThroughput() {
  const elapsed = (Date.now() - startTime) / 1000;
  console.log(`Throughput: ${(totalRequests / elapsed).toFixed(1)} req/s | Total: ${totalRequests} | 429s: ${rateLimitHits}`);
}

// Wrapper that tracks metrics and handles 429 automatically
async function rateLimitedCall<T>(label: string, fn: () => Promise<T>): Promise<T> {
  return apiQueue.add(async () => {
    totalRequests++;
    try {
      return await fn();
    } catch (error: any) {
      if (error.code === 'rate_limited') {
        rateLimitHits++;
        const retryAfter = parseInt(error.headers?.['retry-after'] ?? '1');
        console.warn(`[${label}] Rate limited, waiting ${retryAfter}s`);
        await new Promise(r => setTimeout(r, retryAfter * 1000));
        return fn(); // Single retry
      }
      throw error;
    }
  }) as Promise<T>;
}

// Example: query 5 databases in parallel (queued at 3/s)
const dbIds = ['db1', 'db2', 'db3', 'db4', 'db5'];
const results = await Promise.all(
  dbIds.map(id =>
    rateLimitedCall(`query-${id}`, () =>
      notion.databases.query({ database_id: id, page_size: 100 })
    )
  )
);
logThroughput();
```

```python
from notion_client import Client
import time
import threading

notion = Client(auth=os.environ["NOTION_TOKEN"])

class RateLimiter:
    """Simple token bucket rate limiter for 3 req/s."""
    def __init__(self, rate: float = 3.0):
        self.rate = rate
        self.tokens = rate
        self.last_time = time.monotonic()
        self.lock = threading.Lock()

    def acquire(self):
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_time
            self.tokens = min(self.rate, self.tokens + elapsed * self.rate)
            self.last_time = now

            if self.tokens < 1:
                sleep_time = (1 - self.tokens) / self.rate
                time.sleep(sleep_time)
                self.tokens = 0
            else:
                self.tokens -= 1

limiter = RateLimiter(rate=2.8)  # Slightly under 3/s for safety

def rate_limited_query(database_id: str, **kwargs):
    limiter.acquire()
    return notion.databases.query(database_id=database_id, **kwargs)
```

### Step 2: Worker Queue Architecture for Background Processing

For sustained high-volume operations, decouple API calls from user requests using a job queue.

```typescript
import { Client, isNotionClientError } from '@notionhq/client';
import PQueue from 'p-queue';

interface NotionJob {
  id: string;
  type: 'create' | 'update' | 'query' | 'append';
  payload: any;
  priority: number; // 0 = highest
  retries: number;
  maxRetries: number;
  createdAt: Date;
}

class NotionWorkerQueue {
  private notion: Client;
  private queue: PQueue;
  private deadLetter: NotionJob[] = [];
  private processed = 0;
  private failed = 0;

  constructor(token: string) {
    this.notion = new Client({ auth: token });
    this.queue = new PQueue({
      concurrency: 1,
      interval: 340,
      intervalCap: 1,
    });
  }

  async enqueue(job: Omit<NotionJob, 'id' | 'retries' | 'createdAt'>): Promise<string> {
    const fullJob: NotionJob = {
      ...job,
      id: crypto.randomUUID(),
      retries: 0,
      createdAt: new Date(),
    };

    this.queue.add(() => this.processJob(fullJob), { priority: job.priority });
    return fullJob.id;
  }

  private async processJob(job: NotionJob): Promise<void> {
    try {
      switch (job.type) {
        case 'create':
          await this.notion.pages.create(job.payload);
          break;
        case 'update':
          await this.notion.pages.update(job.payload);
          break;
        case 'query':
          await this.notion.databases.query(job.payload);
          break;
        case 'append':
          await this.notion.blocks.children.append(job.payload);
          break;
      }
      this.processed++;
    } catch (error) {
      job.retries++;
      if (isNotionClientError(error) && error.code === 'rate_limited') {
        const delay = Math.pow(2, job.retries) * 1000;
        await new Promise(r => setTimeout(r, delay));
        if (job.retries < job.maxRetries) {
          this.queue.add(() => this.processJob(job), { priority: job.priority });
          return;
        }
      }
      if (job.retries >= job.maxRetries) {
        this.deadLetter.push(job);
        this.failed++;
      } else {
        this.queue.add(() => this.processJob(job), { priority: job.priority });
      }
    }
  }

  getStats() {
    return {
      pending: this.queue.size,
      processed: this.processed,
      failed: this.failed,
      deadLetter: this.deadLetter.length,
    };
  }
}

// Usage: bulk create 500 pages in background
const worker = new NotionWorkerQueue(process.env.NOTION_TOKEN!);
const DB_ID = process.env.NOTION_DB_ID!;

for (let i = 0; i < 500; i++) {
  await worker.enqueue({
    type: 'create',
    priority: 1,
    maxRetries: 3,
    payload: {
      parent: { database_id: DB_ID },
      properties: {
        Name: { title: [{ text: { content: `Item ${i + 1}` } }] },
      },
    },
  });
}
// 500 pages at ~3/s = ~170 seconds
```

### Step 3: Full Pagination at Scale with Incremental Sync and Memory Management

For databases with 100K+ records, use streaming pagination and incremental sync to avoid re-fetching unchanged data.

```typescript
// Stream results instead of loading all into memory
async function* paginateDatabase(
  databaseId: string,
  filter?: any,
  sorts?: any[]
): AsyncGenerator<any[], void, unknown> {
  let cursor: string | undefined;
  let pageNum = 0;

  do {
    const response = await rateLimitedCall(`page-${pageNum}`, () =>
      notion.databases.query({
        database_id: databaseId,
        filter,
        sorts,
        page_size: 100,
        start_cursor: cursor,
      })
    );

    yield response.results;
    pageNum++;

    cursor = response.has_more ? (response.next_cursor ?? undefined) : undefined;
  } while (cursor);
}

// Process in chunks without loading everything into memory
async function processLargeDatabase(databaseId: string) {
  let totalProcessed = 0;

  for await (const batch of paginateDatabase(databaseId)) {
    for (const page of batch) {
      // Process each record immediately
      totalProcessed++;
    }

    if (totalProcessed % 1000 === 0) {
      console.log(`Processed ${totalProcessed} records...`);
      logThroughput();
    }
  }

  console.log(`Done: ${totalProcessed} total records processed`);
}

// Incremental sync: only fetch records modified since last sync