Operate high-volume ingestion and backfills

High-volume ingestion is an operations problem, not a larger for loop.

A safe GraphJSON backfill or sustained pipeline controls:

• source selection
• transformation
• event time
• batch size
• concurrency
• retry state
• checkpoints
• rejected rows
• retention and cost
• final reconciliation

Use the bulk endpoint for throughput, but keep the workflow restartable and observable.

Know the public guardrails#

The rate boundary is a guardrail, not a throughput target. Network latency, source reads, transformation, retries, and shared workspace traffic all reduce safe operating throughput.

Contact GraphJSON before a business-critical launch or unusually large backfill. Do not test limits by deliberately saturating production.

Separate live delivery from a backfill#

Live delivery and historical import have different risks:

Use separate queues, workers, and observability. A historical job should not consume all delivery capacity needed by current events.

When both target the same workspace, reserve headroom for live traffic.

Create an import manifest#

Before sending anything, record:

{
  "import_id": "imp_2026_07_orders_v1",
  "source": "orders_export_2026_07.ndjson",
  "source_sha256": "...",
  "target_collection": "orders",
  "source_time_zone": "UTC",
  "transform_version": "orders-v3",
  "first_source_time": 1751328000,
  "last_source_time": 1782863999,
  "expected_rows": 1284000
}

Keep the manifest outside GraphJSON in the migration system of record. It should survive a worker restart and identify the exact input and transform.

Do not silently replace an input file after an import starts.

Stage before delivery#

Recommended flow:

source snapshot
  → immutable staged input
    → parse and validate
      → transformed accepted rows
      → quarantined rejected rows
        → bounded GraphJSON sender
          → checkpoint
            → reconciliation

Staging gives you a stable source when:

• an upstream export expires
• the transformation has a defect
• a worker stops halfway through
• GraphJSON returns a retryable error
• reconciliation finds a gap

Protect staged data with access controls and retention appropriate to its sensitivity. Do not use a developer laptop as the only copy of a production migration.

Define a canonical import row#

Normalize every source into one internal type:

type ImportRow = {
  source_position: string;
  event_id: string;
  occurred_at: number;
  collection: string;
  event: Record<string, unknown>;
};

source_position might be:

• file name plus line number
• database primary key
• ordered cursor
• object key plus byte range
• provider export partition and record ID

It is for checkpointing and investigation. Do not add internal storage paths to the analytics event unless they are useful and safe.

Validate before batching#

For each row:

1. parse the source format
2. validate required source columns
3. resolve the original occurrence time
4. map to the approved event contract
5. reject forbidden fields
6. verify whole Unix seconds
7. verify serialized size
8. attach a stable event ID
9. route to the intended collection

Do not discover invalid rows only after combining them into a 50-event request.

Quarantine with a safe reason:

{
  "import_id": "imp_2026_07_orders_v1",
  "source_position": "orders-004.ndjson:18291",
  "event_id": "ord_evt_91",
  "reason": "invalid_currency",
  "transform_version": "orders-v3"
}

Keep the full rejected source row only in a restricted store designed for the migration.

Use full but bounded batches#

Fifty compatible events per request minimizes request overhead. Use smaller batches when:

• individual events approach the size limit
• the source is difficult to replay
• you are testing a new transform
• failures must be isolated more narrowly

Keep one collection per bulk request.

function* batches<T>(rows: T[], size = 50) {
  for (let index = 0; index < rows.length; index += size) {
    yield rows.slice(index, index + size);
  }
}

Batch by an ordered source position so a checkpoint has a clear meaning.

Start with conservative concurrency#

Begin sequentially. Increase to a small worker pool only after observing:

• successful response rate
• p50 and p95 request duration
• 429 frequency
• retry queue depth
• live ingestion health
• GraphJSON query freshness

Example bounded pool:

async function runPool<T>(
  items: T[],
  concurrency: number,
  work: (item: T) => Promise<void>
) {
  let next = 0;

  async function worker() {
    while (next < items.length) {
      const index = next++;
      await work(items[index]);
    }
  }

  await Promise.all(
    Array.from(
      { length: Math.min(concurrency, items.length) },
      worker
    )
  );
}

Do not set concurrency equal to the request-per-second guardrail. Concurrency and request rate are different units.

Apply rate control#

Use a token bucket or explicit scheduler in addition to concurrency.

A worker should not immediately refill every available connection after a 429. Reduce pressure, apply jitter, and allow other workspace traffic to continue.

Recommended response policy:

Preserve uncertain outcomes#

The endpoint does not provide a transport idempotency key. A timeout can happen after GraphJSON accepted the request.

For every imported event:

• keep the same event_id on retry
• keep the same occurrence timestamp
• keep the same payload
• assume at-least-once delivery
• deduplicate analytical queries when exact event uniqueness matters

Changing an event ID on every attempt makes later reconciliation harder.

Use bounded backoff#

function backoffMs(attempt: number) {
  const cap = 10_000;
  const base = Math.min(cap, 250 * 2 ** attempt);
  return Math.floor(base / 2 + Math.random() * (base / 2));
}

Set:

• maximum attempts per batch
• maximum elapsed retry time
• shutdown behavior
• dead-letter destination

Dead letters need an owner and replay procedure. They are not successful delivery.

Commit checkpoints after acknowledgement#

A checkpoint can contain:

{
  "import_id": "imp_2026_07_orders_v1",
  "partition": "orders-004.ndjson",
  "last_acknowledged_position": 18250,
  "accepted_rows": 928400,
  "rejected_rows": 118,
  "retry_batches": 7,
  "updated_at": "2026-07-26T20:18:00Z"
}

Commit only after the batch receives a successful response.

For multiple concurrent partitions, keep one ordered checkpoint per partition. Do not claim one global line number when workers can finish out of order.

Run a canary#

Before the full job:

1. import a tiny representative interval
2. inspect Samples
3. verify field names and types
4. verify first and last timestamps
5. run the intended chart and SQL
6. compare source and GraphJSON counts
7. verify retention and collection
8. inspect a rejected row
9. stop and review

Include difficult cases:

• missing optional values
• legacy schema versions
• daylight-saving boundaries
• largest payload
• repeated source ID
• non-ASCII text

Observe the campaign#

Track outside GraphJSON:

source rows read
valid rows
rejected rows by reason
batches attempted
batches acknowledged
retry attempts by status
dead-letter batches
oldest unprocessed source position
events per second
request duration
estimated completion time

You can send aggregate import progress events to a dedicated GraphJSON operational collection, but do not make the import’s only observability depend on the destination being imported into.

Plan retention and cost#

A backfill increases stored-event volume immediately.

Before starting:

• estimate accepted event count
• confirm target retention
• confirm which historical interval is useful
• remove obsolete or unsafe properties
• separate temporary validation data
• review the current pricing model

Do not import years of data merely because the export contains it. Import the history required for cohorts, comparisons, customer context, and contractual needs.

Reconcile in layers#

Transport reconciliation#

valid source rows
= acknowledged rows + unresolved rows

Destination reconciliation#

Compare GraphJSON counts by:

• day
• event name
• schema version
• source partition
• important business dimension

Business reconciliation#

Compare authoritative totals such as:

• completed orders
• activated accounts
• paid subscriptions
• processed jobs

A matching HTTP count does not prove that event meaning, time, currency, or identity mapping is correct.

Finish deliberately#

At completion:

1. stop new batch creation
2. drain retries
3. resolve or document dead letters
4. record the final checkpoint
5. run reconciliation
6. review dashboards and alerts
7. publish exclusions and known differences
8. expire staged sensitive data on schedule
9. archive the manifest and transform version
10. close the import only after an owner signs off

Abort conditions#

Stop automatically when:

• invalid-row rate exceeds the approved threshold
• 400 responses indicate a contract error
• destination timestamps move outside the manifest range
• a forbidden field is detected
• retries grow faster than successful acknowledgements
• live ingestion health degrades
• expected stored volume is exceeded materially

A paused import with a reliable checkpoint is safer than a fast import that must be untangled later.

Launch checklist#

• Input snapshot and checksum are recorded.
• Transform version is immutable.
• Event contract is executable.
• Source occurrence time is preserved.
• Stable event IDs survive retries.
• Batch size and concurrency are bounded.
• Request rate leaves live-traffic headroom.
• Checkpoints commit only acknowledged work.
• Invalid rows go to a restricted quarantine.
• Retry and dead-letter ownership is named.
• Canary results were reviewed.
• Retention and cost were estimated.
• Destination and business totals are reconciled.

Continue with Bulk logging, Reliable event delivery, and Migrate historical analytics data.