Skip to content

Recipe： Resumable mass operations

Jump to bottom
Eugene Lazutkin edited this page Apr 23, 2026 · 1 revision

Recipe: Resumable mass operations

The SQL-equivalent question this answers: DELETE FROM events WHERE tenant = 'acme' AND created_at < '2025-01-01'; UPDATE orders SET status = 'archived' WHERE ...; INSERT INTO new_table SELECT * FROM old_table. Bulk operations that touch thousands-to-millions of rows, executing inside a bounded runtime (Lambda 15-minute cap, Fargate task, cron container) with zero tolerance for "started-but-didn't-finish" state. In SQL the database owns the transaction and resumes are someone else's problem; in DynamoDB the client walks pages and you own resumption yourself.

Pattern: every paginated mass op (deleteListByParams, cloneListByParams, moveListByParams, editListByParams) accepts MassOpOptions (maxItems, resumeToken, asOf, ifNotExists, ifExists) and returns MassOpResult (processed, skipped, failed, conflicts, cursor?). When cursor is present the op stopped at a page boundary — re-invoke with {resumeToken: cursor} to continue. When cursor is absent the op finished. The primitive loop is "call, check cursor, loop until absent."

Why DynamoDB makes you compose

DynamoDB Query/Scan returns ≤ 1 MB per page. A multi-million-item operation is many pages; inside a runtime-bounded environment (especially Lambda's 15-minute cap) any single invocation may not finish. Three options:

  • Run in an unbounded environment (Fargate / EC2) — available, expensive, and defeats the serverless value proposition.
  • Spawn per-page jobs via Step Functions — solves it cleanly, but the orchestration overhead is high for a one-off backfill.
  • Return a resumption token from each invocation, hand it back on the next invocation. The Lambda/EventBridge loop is "call, check cursor, dispatch next, loop."

The toolkit picks #3 and wraps it: every mass op that walks pages honors maxItems (soft cap, page-boundary enforced) and emits a cursor when the cap stops it mid-scan. The cursor carries the underlying ExclusiveStartKey plus any phase bookkeeping (for multi-phase ops like cascade delete); callers treat it as opaque.

The contract

interface MassOpOptions {
  ifNotExists?: boolean;         // per-item attribute_not_exists; CCF → skipped
  ifExists?: boolean;            // per-item attribute_exists; CCF → skipped
  maxItems?: number;             // soft cap, page-boundary enforced
  resumeToken?: string;          // opaque cursor from a prior MassOpResult.cursor
  asOf?: Date | string | number; // scope-freeze; requires createdAtField declared
  strategy?: 'native' | 'sequential';
  params?: Record<string, unknown>;
  includeDescriptor?: boolean;   // let the descriptor row through (default: filtered)
  filter?: FilterClause[];       // additional FE clauses (via applyFilter)
  search?: string;               // free-form substring via buildSearch
  allowKeyChange?: boolean;      // editListByParams only — auto-promote to move on keyField change
}

interface MassOpResult {
  processed: number;             // items written (or deleted) successfully
  skipped: number;               // items the op passed on — mapFn returned falsy, OC noop, ifExists/ifNotExists CCF, edit-with-no-diff
  failed: {key, reason, details?, sdkError?}[];   // per-item ValidationException / throughput / unknown
  conflicts: {key, reason: 'VersionConflict', sdkError?}[];   // populated only when versionField declared
  cursor?: string;               // page-boundary stop — absent when op finished
}

Worked example 1 — bounded delete sweep

Delete every event older than 30 days. Run on EventBridge every 5 minutes with a 5-minute Lambda timeout; each invocation makes progress, emits a cursor if it ran out of budget, hands off to the next.

import {Adapter, stampCreatedAtEpoch} from 'dynamodb-toolkit';

const events = new Adapter({
  client: docClient,
  table: 'events',
  keyFields: ['tenantId', {name: 'eventId', type: 'string'}],
  structuralKey: '-sk',
  technicalPrefix: '-',
  createdAtField: '-createdAt',            // enables asOf
  hooks: {
    prepare: stampCreatedAtEpoch('-createdAt')
  }
});

// Lambda handler — invoked by EventBridge every 5 minutes.
// Event body optionally carries `{resumeToken}` from the prior run.
export const handler = async event => {
  const cutoff = Date.now() - 30 * 24 * 60 * 60 * 1000;   // 30 days ago

  const result = await events.deleteListByParams(
    {TableName: events.table},
    {
      asOf: cutoff,                       // AND-merges `-createdAt <= cutoff` into the Scan FE
      maxItems: 5000,                     // soft cap sized for the 5-min Lambda budget
      resumeToken: event.resumeToken      // opaque cursor from the prior invocation (or undefined on first run)
    }
  );

  console.log(JSON.stringify({
    processed: result.processed,
    failed: result.failed.length,
    resuming: Boolean(result.cursor)
  }));

  if (result.failed.length) {
    // Per-item errors are bucketed, not thrown — inspect them explicitly.
    for (const f of result.failed) console.error('delete failed', f.key, f.reason);
  }

  // Kick the next invocation if we stopped mid-scan.
  if (result.cursor) {
    // Option A: return the cursor; the EventBridge target treats the return value as the next event.
    return {resumeToken: result.cursor};
    // Option B: immediately self-invoke (InvocationType: 'Event' on Lambda).
    // Option C: write the cursor to a "sweep-state" DDB row and let the next tick pick it up.
  }

  return {done: true};
};

Three properties that make this safe:

  • Idempotent on re-processing. DynamoDB DeleteItem is idempotent — running the op on the same item twice is a no-op. deleteListByParams doesn't need exactly-once page delivery because the underlying primitive can't break under duplicates.
  • asOf is monotonic. -createdAt <= cutoff only gains eligible rows over time (new rows have later timestamps). A resumed sweep catches any rows that crossed the cutoff during the earlier invocation.
  • The cursor is enough. It encodes LastEvaluatedKey. No per-item state has to be carried across invocations.

Worked example 2 — resumable backfill with OC

Backfill a tier field on every customer row. Read-modify-write via editListByParams — each update conditions on versionField, so a concurrent external writer doesn't lose its changes.

const customers = new Adapter({
  client, table: 'customers',
  keyFields: ['customerId'],
  technicalPrefix: '-',
  versionField: '-version'
});

const classify = customer => {
  if (customer.totalSpent >= 10000) return 'platinum';
  if (customer.totalSpent >= 1000)  return 'gold';
  return 'silver';
};

const backfill = async resumeToken => {
  const result = await customers.editListByParams(
    {TableName: customers.table},
    customer => {
      const tier = classify(customer);
      if (customer.tier === tier) return null;          // mapFn falsy → skipped (no write, no WCU)
      return {...customer, tier};
    },
    {
      maxItems: 5000,
      resumeToken
    }
  );

  return result;   // {processed, skipped, failed, conflicts, cursor?}
};

// Driver loop — invocable from a local script or a Fargate task where the 15-min limit doesn't bite.
let cursor;
for (;;) {
  const r = await backfill(cursor);
  console.log(`processed ${r.processed}, skipped ${r.skipped}, conflicts ${r.conflicts.length}, failed ${r.failed.length}`);
  if (!r.cursor) break;
  cursor = r.cursor;
}

Semantics of the buckets:

  • processed — mapFn returned a changed item and the conditional write landed.
  • skipped — mapFn returned falsy (the common "already classified" case) OR the diff was empty (mapFn returned the same shape). No WCU spent.
  • conflicts — versionField declared AND the write hit ConditionalCheckFailed. Interpretation: someone else wrote the row mid-edit. Typical response: rerun on just the conflict keys, or accept the loss.
  • failed — per-item ValidationException / ProvisionedThroughputExceededException / unknown SDK error. Inspect .reason to triage.

Conflict recovery. A second pass with keys = result.conflicts.map(c => c.key) + cloneByKeys / per-key reads + retries picks up the missed rows; because the mapFn is idempotent (same inputs → same output), re-running is safe.

Worked example 3 — conditional clone (don't overwrite)

Copy every active customer to a customers-archive table, but never clobber an existing archive row:

const sourceParams = {
  TableName: customers.table,
  FilterExpression: '#s = :s',
  ExpressionAttributeNames: {'#s': 'status'},
  ExpressionAttributeValues: {':s': 'active'}
};

const result = await customers.cloneListByParams(
  sourceParams,
  item => ({...item, archivedAt: Date.now()}),
  {
    ifNotExists: true,           // per-item attribute_not_exists(pk) — CCF → skipped
    maxItems: 5000,
    resumeToken
  }
);
// result.processed: first-time copies
// result.skipped:   items already in the archive (CCF from ifNotExists)
// result.failed:    per-item validation / throughput failures

ifNotExists / ifExists switch cloneListByParams from BatchWriteItem (no conditions possible) to per-item PutItem with ConditionExpression. You lose batching; you gain "don't overwrite" / "only update existing" semantics. Use when the cost of re-writing an already-copied row (data loss, metadata reset) is higher than the RCU savings of BatchWriteItem.

Cursor lifecycle and opacity

The cursor is the only state a caller needs to carry. It's base64-encoded JSON; treat it as opaque across process restarts and versions:

import {decodeCursor} from 'dynamodb-toolkit/mass';

// Debug / test only — NOT a stable public contract:
console.log(decodeCursor(result.cursor));
// → {LastEvaluatedKey: {customerId: 'c-12345', '-sk': '...'}, meta?: {...}}

The inner shape will grow as mass ops gain new phases (cascade ops already stuff a meta payload for phase tracking). Callers that parse the cursor directly will break on internal changes without notice.

Storage patterns:

  • Event body / EventBridge — pass cursor as part of the next event. Cheap and stateless.
  • Step Functions state — store cursor in the state object; Choice state picks "continue" vs "done" based on presence.
  • DDB state row — write the cursor to a dedicated {jobId, status, cursor} row. Lets external observers see progress; useful for admin UIs.
  • SQS redrive — emit {cursor} to the queue; next consumer pops it and runs. Natural backpressure when the queue fills.

maxItems is a soft cap

maxItems stops the run at the first page boundary after the running total (processed + skipped + failed.length + conflicts.length) meets or exceeds the cap. Pages aren't split mid-way because the page handler is a single batch write / transaction. Practical implications:

  • maxItems: 5000 with a page of 1500 items can finish at 6000 (cap reached at page start → finish the page → stop).
  • Set maxItems lower than your true budget to leave headroom. If a page takes 20s to process, and the Lambda timeout is 60s, targeting maxItems = 2 pages × avg-page-size gives you margin for the final page.
  • Pages size is bounded by DynamoDB (1 MB of raw item bytes per Query/Scan page). For ~500-byte items that's ~2000 items/page; for ~5 KB items, ~200/page. Page count and per-page latency are the real budget dimensions, not maxItems alone.

Composition with asOf (scope-freeze)

Every paginated mass op accepts options.asOf. When createdAtField is declared, the toolkit AND-merges <createdAtField> <= :asOf into the FilterExpression:

// "Everything as of 2025-04-01" — rows created after that timestamp are invisible to this run.
await customers.cloneListByParams(
  {TableName: customers.table},
  item => ({...item, migratedAt: Date.now()}),
  {
    asOf: '2025-04-01T00:00:00Z',
    maxItems: 5000,
    resumeToken
  }
);

Critical for long-running backfills: rows created after the job started don't need to be re-processed mid-run. Combined with idempotent maps, this makes re-runs trivially safe.

Without createdAtField declared the option throws CreatedAtFieldNotDeclared — the toolkit won't silently skip it. For new tables, declare createdAtField from day one; it's free (one attribute per item) and unlocks the scope-freeze recipe everywhere.

Composition with applyFilter / buildSearch

options.filter and options.search thread through the same page handler:

await customers.deleteListByParams(
  {TableName: customers.table},
  {
    filter: [
      {field: 'status', op: 'eq', value: 'archived'},
      {field: 'lastLogin', op: 'lt', value: '2020-01-01'}
    ],
    search: 'inactive',
    asOf: Date.now() - 90 * 24 * 60 * 60 * 1000,
    maxItems: 5000,
    resumeToken
  }
);

All three (filter, search, asOf) AND-compose into the FilterExpression. applyFilter clauses that match a pk/sk auto-promote to KeyConditionExpression when the target params are a Query; see the filter URL grammar recipe.

Error buckets, not throws

The page handler catches per-item errors and routes them to failed / conflicts / skipped:

Scenario Bucket Re-run safety
ValidationException / ProvisionedThroughputExceededException on a single item failed Retryable after throughput calms / input fixed.
ConditionalCheckFailedException on ifNotExists / ifExists clone skipped By design — item already existed / already absent.
ConditionalCheckFailedException on editListByParams with versionField conflicts Re-read the row and retry the edit.
ConditionalCheckFailedException on editListByParams without versionField skipped Rare (item was deleted mid-edit).
editListByParams mapFn returned the same shape as input skipped No-op; no WCU.
editListByParams mapFn touched a keyField without {allowKeyChange: true} failed with reason: 'Unknown' Fix the mapFn or pass allowKeyChange: true to auto-promote the edit to a move.
Network / auth / non-per-item SDK error thrown (page aborts) Intentional — the failure isn't item-scoped and the caller should see it.

Any page-level throw aborts the whole run; the caller sees a bare exception. Running with a cursor on the retry resumes at the last successful page boundary.

Long-running orchestration patterns

Three common wire-ups:

A — EventBridge self-loop (Lambda)

EventBridge rule → Lambda (once per minute) → deleteListByParams
  cursor present → return {resumeToken: cursor} → EventBridge retry with payload
  cursor absent → emit "done" CloudWatch event → detach the rule

Cheap, stateless. Fine for sweeps that finish within a few hours. Downsides: EventBridge payload size limits; "done" notification needs explicit wiring; no external visibility into progress without logging.

B — Step Functions

Map state → Lambda(deleteListByParams) → Choice state
  cursor present → loop back into Lambda
  cursor absent → success

More overhead to set up; clean state machine visible in console; handles retries + dead-letter inbuilt. Right for anything operations will observe.

C — Self-paced driver (Fargate / local script)

let cursor;
for (;;) {
  const r = await adapter.deleteListByParams({TableName}, {maxItems: 10000, resumeToken: cursor});
  if (!r.cursor) break;
  cursor = r.cursor;
}

Simplest. No Lambda timeout to fight. Right for one-off backfills, manual ops, data-migration scripts. Downsides: no observer, a script crash loses progress unless you persist the cursor.

Cost model

Per mass op, you pay for:

  • Reads — 1 RCU per 4 KB of item data (or 2 for strong reads). Page size, not maxItems, determines how many pages you pay for.
  • Writes — 1 WCU per 1 KB per item written. For deletes that's 1 WCU per item regardless of size; for puts/updates it scales with item size.
  • Conditions — a CCF'd PutItem still costs 1 WCU (the condition was evaluated before the reject). ifNotExists: true on a clone that skips 10,000 items costs 10,000 WCU — budget for it.
  • BatchWriteItem batching — 25 items per call; unprocessed items retry with jitter (toolkit's backoff). Saves per-request overhead, not per-item WCU.

The options.strategy knob toggles 'native' (BatchWriteItem — cheaper per call, no conditions) vs 'sequential' (per-item PutItem — supports conditions, slower). Clone with ifNotExists/ifExists auto-switches to sequential regardless of strategy.

When this pattern fits

  • Bulk operation that's expected to span thousands-to-millions of items.
  • Caller has a bounded runtime (Lambda, cron container, scheduled task).
  • The mapFn (for edit/clone/move) is idempotent — same input produces same output regardless of how many times it runs.
  • Ordering doesn't matter. DDB pages come back in partition-key order, not any logical order.

When it doesn't fit

  • Transactional multi-item updates. applyTransaction handles ≤ 100 items atomically; beyond that, mass ops sacrifice atomicity. If you need "all items update or none," split into ≤100-item transactions and coordinate externally.
  • Ordered processing. The run sees items in DDB's page order (partition-key, then sort-key). Enforce ordering client-side after collection, or use a different primitive.
  • Cross-partition set math. Mass ops walk one table / one index. Joining two sources needs two scans + client-side merge.
  • Hot-partition writes. Mass writes batched into BatchWriteItem can overwhelm a single partition's 1,000 WCU/sec limit. Spread writes with a strategy: 'sequential' + backoff, or pre-shard the pk.

Related

🔍 Search the wiki

dynamodb-toolkit

Start here

Guides

Adapter

Expression builders

Batch / transactions / mass / paths

REST surface

Framework adapters

Recipes

History

Clone this wiki locally