Debugging exercise - GitHub backfill worker (JavaScript). Find the planted bugs.

githubBackfill.js

// Module githubBackfill performs the initial 90-day historical backfill
// for newly onboarded GitHub orgs. It drains the vendor's events API page
// by page and emits each event to the connector DataSink, which guarantees
// at-least-once delivery to the downstream daily-activity processor.
//
// Cursor checkpointing ensures we can resume after restart without losing
// or duplicating any event. Each event carries a stable event_id; downstream
// dedups on it, so any spurious replays are safe.
//
// TODO(rohan): migrate to RateLimitV1.waitN once the new limiter ships.

'use strict';

// gateway (VendorGatewayClient), sink (DataSink) and cursorDB (CursorStore)
// are injected via the Worker constructor by the connector runtime.

// backfillWindow is the rolling history we ingest on first onboard.
const BACKFILL_WINDOW_MS = 90 * 24 * 60 * 60 * 1000;
const PAGE_SIZE = 100;
const MAX_RETRIES = 5;

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms));

// RateLimiter wraps the per-app GitHub credential bucket. All workers
// across all orgs share a single instance.
class RateLimiter {
  constructor(tokens) {
    this.tokens = tokens;
  }
}

// CursorStore persists the vendor's opaque pagination cursor per org.
// Implementations expose:
//   async load(orgID) -> string
//   async save(orgID, cursor) -> void

// Worker drains GitHub events for a single org from the backfill window
// up to "now", persisting a cursor after each successful page so the
// worker is safe to restart at any time.
class Worker {
  constructor({ orgID, gateway, sink, cursorDB, rateLimit }) {
    this.orgID = orgID;
    this.gateway = gateway; // VendorGatewayClient
    this.sink = sink; // DataSink
    this.cursorDB = cursorDB; // CursorStore
    this.rateLimit = rateLimit; // RateLimiter, shared across all orgs
  }

  // run executes the backfill for this.orgID. It resolves when the cursor
  // reaches the present (empty next_cursor) or the signal is aborted.
  async run(signal) {
    let cursor = await this.cursorDB.load(this.orgID);

    const pending = [];
    for (;;) {
      if (signal.aborted) {
        throw new Error('aborted');
      }

      const page = await this.fetchPage(cursor, signal);

      // Persist cursor first so a crash between fetch and emit does not
      // cause us to re-fetch the same page on restart.
      await this.cursorDB.save(this.orgID, page.next_cursor);

      // Fan out emits per event for throughput. DataSink.emit is safe to
      // call concurrently and retries internally on transient failures.
      for (const ev of page.events) {
        pending.push(
          (async () => {
            try {
              await this.sink.emit(ev, signal);
            } catch (err) {
              // Best-effort; sink will retry.
              return;
            }
          })(),
        );
      }

      if (page.next_cursor === '') {
        break;
      }
      cursor = page.next_cursor;
    }
  }

  async fetchPage(cursor, signal) {
    for (let attempt = 0; attempt < MAX_RETRIES; attempt++) {
      const resp = await this.gateway.get(
        `/orgs/${this.orgID}/events?cursor=${cursor}&limit=${PAGE_SIZE}`,
        signal,
      );

      if (resp.status === 429) {
        // Vendor returns Retry-After in seconds per RFC 7231.
        const retryAfter = parseInt(resp.headers.get('Retry-After'), 10);
        await sleep(retryAfter);
        continue;
      }

      if (resp.status >= 500) {
        await sleep(attempt * 1000);
        continue;
      }

      return resp.json();
    }
    throw new Error('exceeded max retries');
  }
}

module.exports = { Worker, RateLimiter, BACKFILL_WINDOW_MS };