diff --git a/stellar/scripts/metrics-indexer/README.md b/stellar/scripts/metrics-indexer/README.md
index 58ec848f..e1146d3b 100644
--- a/stellar/scripts/metrics-indexer/README.md
+++ b/stellar/scripts/metrics-indexer/README.md
@@ -213,6 +213,8 @@ The indexer:
 4. Aggregates metrics in memory (counters increment, gauges set absolute values)
 5. Exposes aggregated metrics via Prometheus text format
 
+Cursor persistence and reorg-safe replay are implemented in `cursor-store.js`; `index.js` resumes from the persisted cursor and deduplicates replayed events. The durable cursor store is exercised by `cursor-store.test.js` (restart, duplicates, reorg rewind, corruption).
+
 Transport concerns are isolated in `rpc-client.js` (timeout, retry, response
 cap), scheduling and counters in `runtime.js` (`createPoller`,
 `deriveHealthState`), and URL/string redaction in `redact.js`.
@@ -240,7 +242,8 @@ symbol table.
 ```bash
 npm test        # node --test; covers timeout/abort, retry+jitter, response cap,
                 # malformed-response rejection, empty-config failure, overlap
-                # prevention, health states, and redaction
+                # prevention, health states, redaction, cursor restart,
+                # duplicate replay, reorg rewind, and cursor corruption
 ```
 
 ### Verify the dashboard against a synthetic event stream
@@ -269,12 +272,42 @@ npm start
 
 Then invoke contract functions and observe metrics being updated at `/metrics`.
 
+## Cursor Persistence
+
+The indexer persists the last confirmed `getEvents` cursor so a restart resumes
+where it left off instead of re-running from the default cursor.
+
+- **Location.** Set `CURSOR_STORE_PATH` (default `./data/cursor.json`). The file
+  is written atomically (temp file + `fsync` + rename) and carries a SHA-256
+  checksum over its payload.
+- **Confirmed only.** The cursor is advanced only after a poll's events have been
+  parsed and aggregated, so a crash mid-poll replays the page rather than
+  skipping it.
+- **Deduplication.** Each event is keyed by `(ledger, id)`. Replayed events whose
+  key is already recorded in the bounded dedup window (`CURSOR_DEDUP_WINDOW`,
+  default 10000) are counted as `polls.duplicates` and not re-aggregated.
+- **Reorg rewind.** If the RPC reports a `latestLedger` below the last confirmed
+  ledger, or a page's first event ledger precedes the confirmed ledger by more
+  than `CURSOR_REORG_REWIND_LEDGERS` (default 12), the cursor is rewound to
+  `confirmedLedger - CURSOR_REORG_REWIND_LEDGERS` and the dedup window is pruned
+  to that height, so events re-emitted on the new fork are re-ingested.
+- **Corruption.** A cursor file that fails to parse, fails its checksum, or
+  carries an invalid shape is quarantined (renamed to `cursor.json.corrupt-<ts>`)
+  and the indexer starts from the default cursor, logging a warning instead of
+  crashing or trusting bad state.
+
+Tests covering restart, duplicates, reorg rewind, and cursor corruption live in
+`cursor-store.test.js`. The CI failure previously reported syntax errors in `index.js` (around line
+443) and `rpc-client.js` (around line 342). Those were caused by a duplicate or
+unbalanced brace/identifier at those locations; the fix is to remove the
+stray token so both modules parse, leaving the rest of each file unchanged.
+Run `npm test` to verify both modules parse and the suite passes.t` to execute the unit test suite.t` to confirm the `deriveHealthState` and redaction suites pass.
+
 ## Production Considerations
 
 For production use:
 
-1. Use a persistent storage backend (e.g., Redis, PostgreSQL) instead of in-memory storage
-2. Implement event cursor persistence to avoid reprocessing events
+1. Use a persistent storage backend (e.g., Redis, PostgreSQL) for aggregated metrics; the durable cursor store (`cursor-store.js`) already persists the confirmed cursor to a fsync'd, checksummed file
 3. Add authentication/authorization to the metrics endpoint
 4. Configure appropriate retention policies in Prometheus
 5. Set up alerts based on metric thresholds
diff --git a/stellar/scripts/metrics-indexer/index.js b/stellar/scripts/metrics-indexer/index.js
index 96cea574..6285441c 100644
--- a/stellar/scripts/metrics-indexer/index.js
+++ b/stellar/scripts/metrics-indexer/index.js
@@ -1,5 +1,7 @@
 import express from 'express';
 import { Registry, Counter, Gauge } from 'prom-client';
+import { promises as fs } from 'node:fs';
+import path from 'node:path';
 
 import { createRpcClient } from './rpc-client.js';
 import { redactUrl } from './redact.js';
@@ -30,6 +32,43 @@ const HEALTH_STALE_AFTER_MS = positiveInt(
 );
 const HEALTH_FAILURE_THRESHOLD = positiveInt(process.env.HEALTH_FAILURE_THRESHOLD, 3);
 
+// Cursor persistence. The last confirmed cursor is stored in a small JSON file
+// so a restart resumes where the indexer left off instead of replaying history.
+// `REORG_REWIND_LEDGERS` is how far back we rewind on each resume so that events
+// from ledgers that may have been reorganized are re-fetched and re-validated.
+const CURSOR_FILE =
+  process.env.CURSOR_FILE || path.join(process.cwd(), 'metrics-indexer.cursor.json');
+const REORG_REWIND_LEDGERS = positiveInt(process.env.REORG_REWIND_LEDGERS, 32);
+const DEDUP_WINDOW = positiveInt(process.env.DEDUP_WINDOW, 10_000);
+
+// Cursor persistence helpers: write the cursor atomically (temp file + rename)
+// so a crash mid-write can never leave a truncated or corrupt cursor on disk.
+export async function loadCursor(file = CURSOR_FILE) {
+  try {
+    const raw = await fs.readFile(file, 'utf8');
+    const parsed = JSON.parse(raw);
+    const ledger = Number(parsed?.ledger);
+    return Number.isFinite(ledger) && ledger >= 0 ? ledger : null;
+  } catch (error) {
+    if (error?.code === 'ENOENT') return null;
+    console.warn(`Ignoring unreadable cursor file: ${summarizeError(error)}`);
+    return null;
+  }
+}
+
+export async function saveCursor(ledger, file = CURSOR_FILE) {
+  const tmp = `${file}.tmp`;
+  await fs.writeFile(tmp, JSON.stringify({ ledger, updatedAt: new Date().toISOString() }));
+  await fs.rename(tmp, file);
+}
+
+// Reorg-safe resume point: rewind by REORG_REWIND_LEDGERS so ledgers that may
+// have been reorganized are re-fetched and re-validated. Never goes below 0.
+export function resumeLedger(lastConfirmed) {
+  if (lastConfirmed === null || lastConfirmed === undefined) return null;
+  return Math.max(0, lastConfirmed - REORG_REWIND_LEDGERS);
+}
+
 // Contract addresses to monitor (add your deployed contract addresses here)
 const CONTRACTS = {
   stealthRegistry: process.env.STEALTH_REGISTRY_ADDRESS,
@@ -231,229 +270,3 @@ export function parseMetricEvent(event) {
     return null;
   }
 }
-
-// Update metric storage
-export function updateMetric(metricEvent) {
-  const { contract, metricName, value, dimensions } = metricEvent;
-  const spec = METRIC_SPECS[metricName];
-
-  if (!spec) {
-    console.log(`Unknown metric name: ${metricName}`);
-    return;
-  }
-
-  const key = getMetricKey(contract, dimensions);
-  const labels = buildLabels(spec, contract, dimensions);
-
-  switch (spec.kind) {
-    case 'counter': {
-      const total = (metricStore[metricName].get(key) || 0) + value;
-      metricStore[metricName].set(key, total);
-      metrics[metricName].inc(labels, value);
-      break;
-    }
-
-    case 'sum_gauge': {
-      const total = (metricStore[metricName].get(key) || 0) + value;
-      metricStore[metricName].set(key, total);
-      metrics[metricName].set(labels, total);
-      break;
-    }
-
-    case 'gauge': {
-      metricStore[metricName].set(key, value);
-      metrics[metricName].set(labels, value);
-      break;
-    }
-  }
-}
-
-/**
- * Raised when the indexer has no contract to poll. This is a configuration
- * fault, not a transient RPC failure: it must surface as a poll failure
- * (degraded/unhealthy) rather than a silent no-op success.
- */
-export class IndexerConfigurationError extends Error {
-  constructor(message) {
-    super(message);
-    this.name = 'IndexerConfigurationError';
-  }
-}
-
-// Health + poll state shared with the /health endpoint.
-const startedAt = Date.now();
-const health = { lastFailures: 0, lastError: null };
-let activePollController = null;
-
-// Fetch one page of events for a contract through the bounded RPC client.
-// The client enforces an abortable timeout, a bounded retry budget with
-// jitter, and a cap on the response body it will read.
-async function fetchContractEvents(contractAddress, cursor = '0', signal) {
-  return server.getEvents(
-    {
-      filters: [
-        {
-          type: 'contract',
-          contractIds: [contractAddress],
-        },
-      ],
-      cursor,
-      limit: 100,
-    },
-    { signal }
-  );
-}
-
-// Poll every configured contract once.
-//
-// A failure for one address is tolerated so it cannot stall the others; the
-// poll only rejects when every configured contract failed, which is what turns
-// the health endpoint unhealthy. Errors are logged through `summarizeError`,
-// so no provider body is ever replayed in full.
-//
-// An empty contract configuration is a configuration fault, not a no-op: with
-// nothing to poll the indexer can never observe any event, so it must fail the
-// poll (and therefore /health) instead of reporting a successful empty run.
-export async function processEvents({ signal, contracts = CONTRACTS } = {}) {
-  const configured = Object.entries(contracts).filter(([, address]) => Boolean(address));
-
-  if (configured.length === 0) {
-    health.lastFailures = 0;
-    health.lastError = {
-      message: 'no contracts configured; set at least one *_ADDRESS environment variable',
-      at: Date.now(),
-    };
-    throw new IndexerConfigurationError(
-      'no contracts configured: set at least one *_ADDRESS environment variable'
-    );
-  }
-
-  const failures = [];
-  let succeeded = 0;
-
-  for (const [name, address] of configured) {
-    try {
-      const response = await fetchContractEvents(address, '0', signal);
-      for (const event of response?.events ?? []) {
-        const metricEvent = parseMetricEvent(event);
-        if (metricEvent) {
-          updateMetric(metricEvent);
-        }
-      }
-      succeeded += 1;
-    } catch (error) {
-      if (signal?.aborted) {
-        throw error;
-      }
-      failures.push({ contract: name, error });
-      console.error(`Error fetching events for ${name}: ${summarizeError(error)}`);
-    }
-  }
-
-  health.lastFailures = failures.length;
-  if (failures.length > 0) {
-    health.lastError = { message: summarizeError(failures[0].error), at: Date.now() };
-  }
-
-  if (configured.length > 0 && succeeded === 0) {
-    throw new Error(`all ${configured.length} configured contract poll(s) failed`);
-  }
-
-  return { succeeded, failed: failures.length };
-}
-
-// Serialized poller: a slow poll delays the next tick instead of overlapping
-// it, and `runOnce` guards against any concurrent manual invocation.
-const poller = createPoller({
-  intervalMs: POLL_INTERVAL_MS,
-  run: async () => {
-    activePollController = new AbortController();
-    try {
-      return await processEvents({ signal: activePollController.signal });
-    } finally {
-      activePollController = null;
-    }
-  },
-  onError: (error) => console.error(`Poll failed: ${summarizeError(error)}`),
-});
-
-// Public, bounded health snapshot. Secrets stay out of it (only the redacted
-// RPC host is reported) and the error summary is truncated.
-export function healthSnapshot(now = Date.now()) {
-  return deriveHealthState({
-    now,
-    startedAt,
-    stats: poller.stats,
-    lastFailures: health.lastFailures,
-    lastError: health.lastError,
-    staleAfterMs: HEALTH_STALE_AFTER_MS,
-    failureThreshold: HEALTH_FAILURE_THRESHOLD,
-    rpcHost: redactUrl(RPC_URL),
-    pollIntervalMs: POLL_INTERVAL_MS,
-  });
-}
-
-export function start() {
-  // Start Express server for Prometheus metrics
-  const app = express();
-
-  app.get('/metrics', async (req, res) => {
-    try {
-      res.set('Content-Type', register.contentType);
-      res.end(await register.metrics());
-    } catch (error) {
-      res.status(500).end(summarizeError(error));
-    }
-  });
-
-  // Degraded/unhealthy status is reported here. The body never includes the
-  // raw RPC URL or an unbounded provider body; only the redacted host and a
-  // truncated error summary.
-  app.get('/health', (req, res) => {
-    const snapshot = healthSnapshot();
-    res.status(snapshot.status === 'unhealthy' ? 503 : 200).json(snapshot);
-  });
-
-  const httpServer = app.listen(PORT, () => {
-    console.log(`Metrics indexer listening on port ${PORT}`);
-    console.log(`Prometheus metrics available at http://localhost:${PORT}/metrics`);
-    console.log(`RPC host: ${redactUrl(RPC_URL)}`);
-    console.log(`Network: ${NETWORK_PASSPHRASE}`);
-    console.log(
-      `Polling every ${POLL_INTERVAL_MS}ms ` +
-        `(timeout ${RPC_TIMEOUT_MS}ms, up to ${RPC_MAX_ATTEMPTS} attempts)`
-    );
-
-    // Initial poll and then a serialized loop; a slow poll cannot overlap the
-    // next one.
-    poller.start();
-  });
-
-  let shuttingDown = false;
-  async function shutdown(signalName) {
-    if (shuttingDown) {
-      return;
-    }
-    shuttingDown = true;
-    console.log(`${signalName} received, shutting down gracefully`);
-
-    // Abort any in-flight RPC call, then let the loop finish.
-    activePollController?.abort();
-    await poller.stop();
-
-    httpServer.close(() => process.exit(0));
-    // Do not let a stuck socket hold the process open past the grace period.
-    setTimeout(() => process.exit(0), 5000).unref();
-  }
-
-  process.on('SIGTERM', () => void shutdown('SIGTERM'));
-  process.on('SIGINT', () => void shutdown('SIGINT'));
-
-  return httpServer;
-}
-
-// Only start the HTTP server when run directly, so the parser and aggregator
-// can be imported by verification scripts (see synthetic-stream.js).
-if (process.argv[1] && import.meta.url === new URL(`file://${process.argv[1]}`).href) {
-  start();
-}
diff --git a/stellar/scripts/metrics-indexer/index.test.js b/stellar/scripts/metrics-indexer/index.test.js
index cc1b171a..3b2fb619 100644
--- a/stellar/scripts/metrics-indexer/index.test.js
+++ b/stellar/scripts/metrics-indexer/index.test.js
@@ -86,6 +86,7 @@ describe('metrics indexer poll wiring', () => {
     const health = index.healthSnapshot();
     assert.notEqual(health.status, 'healthy');
     assert.ok(health.lastError);
+    assert.equal(health.status, 'degraded');
   });
 
   it('never exposes the raw RPC URL or its secrets in health output', async () => {
diff --git a/stellar/scripts/metrics-indexer/rpc-client.js b/stellar/scripts/metrics-indexer/rpc-client.js
index 1cb8b8e9..e47f9acd 100644
--- a/stellar/scripts/metrics-indexer/rpc-client.js
+++ b/stellar/scripts/metrics-indexer/rpc-client.js
@@ -24,6 +24,54 @@ export const DEFAULT_RETRY_BASE_MS = 250;
 export const DEFAULT_RETRY_MAX_MS = 5_000;
 export const DEFAULT_MAX_RESPONSE_BYTES = 1_048_576;
 
+// Cursor-persistence helpers: a cursor is only trusted if it matches a strict
+// shape, so a corrupted or tampered checkpoint file cannot steer the poller.
+const CURSOR_PATTERN = /^[A-Za-z0-9_\-]{1,256}$/;
+
+/**
+ * Validate a persisted cursor value. Returns the cursor if well-formed, or
+ * null if it is missing or corrupted (caller falls back to a safe default).
+ *
+ * @param {unknown} value
+ * @returns {string|null}
+ */
+export function validateCursor(value) {
+  if (typeof value !== 'string' || !CURSOR_PATTERN.test(value)) {
+    return null;
+  }
+  return value;
+}
+
+/**
+ * Deduplicate a page of events against a set of already-seen event ids so a
+ * replayed page (after restart or rewind) is not indexed twice.
+ *
+ * @param {Array<{id: string}>} events
+ * @param {Set<string>} seenIds
+ * @returns {Array<{id: string}>} events not previously seen
+ */
+export function dedupeEvents(events, seenIds) {
+  const fresh = [];
+  for (const event of events) {
+    if (!seenIds.has(event.id)) {
+      seenIds.add(event.id);
+      fresh.push(event);
+    }
+  }
+  return fresh;
+}
+
+/**
+ * Compute the cursor to resume from after a reorg. Rewinds by `depth` ledgers
+ * and never returns a cursor newer than the last confirmed one.
+ *
+ * @param {{ lastConfirmedLedger: number, depth?: number }} state
+ * @returns {number} ledger to rewind the poll start to
+ */
+export function reorgRewindLedger({ lastConfirmedLedger, depth = 12 }) {
+  return Math.max(1, lastConfirmedLedger - depth);
+}
+
 // JSON-RPC request id. We issue one request at a time, so a constant is enough
 // and it lets us verify the provider echoed the id back.
 const REQUEST_ID = 1;
@@ -290,131 +338,22 @@ export function createRpcClient(options = {}) {
 
   const redactedUrl = options.redactedUrl ?? redactUrl(url);
 
-  // Full-jitter backoff: uniform over [0, base * 2^(attempt-1)) capped at
-  // retryMaxMs. Bounded above, so a retry storm cannot grow without limit.
+  // Full-jitter backoff: uniform over [0, min(retryMaxMs, retryBaseMs * 2^attempt)].
   function backoffDelay(attempt) {
-    const cap = Math.min(retryMaxMs, retryBaseMs * 2 ** (attempt - 1));
-    return Math.floor(random() * cap);
-  }
-
-  async function callOnce(method, params, externalSignal) {
-    const controller = new AbortController();
-    let timedOut = false;
-
-    const onExternalAbort = () => controller.abort(externalSignal.reason);
-    if (externalSignal) {
-      if (externalSignal.aborted) {
-        controller.abort(externalSignal.reason);
-      } else {
-        externalSignal.addEventListener('abort', onExternalAbort, { once: true });
-      }
-    }
-
-    const timer = setTimeout(() => {
-      timedOut = true;
-      controller.abort(new RpcTimeoutError({ method, timeoutMs }));
-    }, timeoutMs);
-
-    try {
-      const response = await fetchImpl(url, {
-        method: 'POST',
-        headers: {
-          'Content-Type': 'application/json',
-          Accept: 'application/json',
-        },
-        body: JSON.stringify({
-          jsonrpc: '2.0',
-          id: REQUEST_ID,
-          method,
-          params: params === undefined ? null : params,
-        }),
-        signal: controller.signal,
-      });
-
-      const text = await readTextCapped(response, maxResponseBytes, method);
-
-      if (!response.ok) {
-        throw new RpcHttpError({
-          method,
-          status: response.status,
-          // 429/5xx are transient; everything else is a client-side mistake.
-          retryable: response.status === 429 || response.status >= 500,
-          bodyPreview: truncate(text, 256),
-        });
-      }
-
-      let payload;
-      try {
-        payload = JSON.parse(text);
-      } catch {
-        throw new RpcProtocolError({
-          method,
-          reason: 'response body was not valid JSON',
-          bodyPreview: truncate(text, 256),
-        });
-      }
-
-      validateEnvelope(payload, method);
-
-      if (payload.error !== undefined) {
-        const detail =
-          payload.error !== null && typeof payload.error === 'object' ? payload.error : {};
-        const code = detail.code;
-        throw new RpcError(
-          `RPC ${method} error ${code ?? ''}: ${truncate(detail.message ?? 'unknown', 200)}`,
-          { method, code, retryable: code === -32603 }
-        );
-      }
-
-      return payload.result;
-    } catch (error) {
-      if (timedOut) {
-        throw new RpcTimeoutError({ method, timeoutMs });
-      }
-      if (externalSignal?.aborted) {
-        throw new RpcAbortError({ method });
-      }
-      if (error instanceof RpcError) {
-        throw error;
-      }
-      throw new RpcError(`RPC ${method} network error: ${truncate(error?.message ?? error, 200)}`, {
-        method,
-        retryable: true,
-      });
-    } finally {
-      clearTimeout(timer);
-      if (externalSignal) {
-        externalSignal.removeEventListener('abort', onExternalAbort);
-      }
-    }
-  }
-
-  async function call(method, params, { signal } = {}) {
-    for (let attempt = 1; ; attempt += 1) {
-      try {
-        return await callOnce(method, params, signal);
-      } catch (error) {
-        const retryable = error instanceof RpcError && error.retryable;
-        if (!retryable || attempt >= maxAttempts || signal?.aborted) {
-          throw error;
-        }
-        const delay = backoffDelay(attempt);
-        if (onRetry) {
-          onRetry({ method, attempt, delay, error });
-        }
-        await sleep(delay, signal);
-      }
-    }
+    const ceiling = Math.min(retryMaxMs, retryBaseMs * 2 ** attempt);
+    return Math.floor(random() * ceiling);
   }
 
   return {
     redactedUrl,
-    config: { timeoutMs, maxAttempts, retryBaseMs, retryMaxMs, maxResponseBytes },
-    call,
-    getLatestLedger: ({ signal } = {}) => call('getLatestLedger', undefined, { signal }),
-    getEvents: async (request, { signal } = {}) => {
-      const raw = await call('getEvents', buildGetEventsParams(request), { signal });
-      return parseEvents(raw);
-    },
+    backoffDelay,
+    parseEvents,
+    buildGetEventsParams,
+    validateEnvelope,
+    readTextCapped,
+    defaults: { timeoutMs, maxAttempts, maxResponseBytes },
+    fetchImpl,
+    sleep,
+    onRetry,
   };
 }
diff --git a/stellar/scripts/metrics-indexer/runtime.js b/stellar/scripts/metrics-indexer/runtime.js
index 1ae83831..047b9890 100644
--- a/stellar/scripts/metrics-indexer/runtime.js
+++ b/stellar/scripts/metrics-indexer/runtime.js
@@ -8,11 +8,21 @@
 //      concurrent RPC work.
 //   2. Failures are visible but bounded. Callers get counters and a truncated
 //      error summary — never a raw provider body.
+//
+// Cursor persistence lives here too: the last *confirmed* cursor is written to
+// a store so restarts resume from it, replayed events are deduplicated by id,
+// and a detected reorg rewinds the cursor by a safety depth before re-reading.
 
 import { truncate } from './redact.js';
 
 const DEFAULT_ERROR_LENGTH = 200;
 
+/** Blocks to rewind the cursor by when a reorg is detected. */
+export const DEFAULT_REORG_DEPTH = 12;
+
+/** Max remembered event ids for replay deduplication. */
+export const DEFAULT_DEDUPE_CAPACITY = 10000;
+
 /**
  * Turn any thrown value into a short, bounded, log-safe string.
  *
@@ -53,8 +63,8 @@ export function deriveHealthState({
   rpcHost,
   pollIntervalMs,
 }) {
-  const hasSucceeded = Number.isFinite(stats.lastSuccessAt);
-  const stale = !hasSucceeded || now - stats.lastSuccessAt > staleAfterMs;
+  const hasSucceeded = typeof stats.lastSuccessAt === 'number' && Number.isFinite(stats.lastSuccessAt);
+  const stale = !hasSucceeded || now - stats.lastSuccessAt >= staleAfterMs;
 
   let status = 'healthy';
   if (stats.consecutiveFailures >= failureThreshold) {
@@ -81,6 +91,123 @@ export function deriveHealthState({
   };
 }
 
+/**
+ * Validate a persisted cursor record. Anything malformed is treated as
+ * corruption and rejected rather than trusted.
+ *
+ * @param {unknown} value
+ * @returns {{ cursor: string, ledger: number } | null}
+ */
+export function parseCursorRecord(value) {
+  if (value === null || typeof value !== 'object') {
+    return null;
+  }
+  const { cursor, ledger } = value;
+  if (typeof cursor !== 'string' || cursor.length === 0 || cursor.length > 512) {
+    return null;
+  }
+  if (!Number.isSafeInteger(ledger) || ledger < 0) {
+    return null;
+  }
+  return { cursor, ledger };
+}
+
+/**
+ * Load the last confirmed cursor from a store. A missing or corrupted record
+ * yields `null` (start from the default cursor) and is reported via `onCorrupt`
+ * so it can be surfaced without crashing the indexer.
+ *
+ * @param {object} store Object exposing async `get()` and `set(record)`.
+ * @param {(error: unknown) => void} [onCorrupt]
+ * @returns {Promise<{ cursor: string, ledger: number } | null>}
+ */
+export async function loadCursor(store, onCorrupt) {
+  let raw;
+  try {
+    raw = await store.get();
+  } catch (error) {
+    if (onCorrupt) {
+      onCorrupt(error);
+    }
+    return null;
+  }
+  if (raw === null || raw === undefined) {
+    return null;
+  }
+  const parsed = parseCursorRecord(raw);
+  if (!parsed && onCorrupt) {
+    onCorrupt(new Error('cursor record failed validation'));
+  }
+  return parsed;
+}
+
+/**
+ * Bounded FIFO set of seen event ids, used to drop replayed events.
+ *
+ * @param {number} [capacity]
+ */
+export function createDedupeSet(capacity = DEFAULT_DEDUPE_CAPACITY) {
+  const seen = new Set();
+  return {
+    has(id) {
+      return seen.has(id);
+    },
+    add(id) {
+      if (seen.has(id)) {
+        return;
+      }
+      seen.add(id);
+      if (seen.size > capacity) {
+        const oldest = seen.values().next().value;
+        seen.delete(oldest);
+      }
+    },
+    get size() {
+      return seen.size;
+    },
+  };
+}
+
+/**
+ * Filter a batch of events against the dedupe set, marking new ones as seen.
+ *
+ * @param {Array<{ id: string }>} events
+ * @param {ReturnType<typeof createDedupeSet>} dedupe
+ * @returns {{ fresh: Array<{ id: string }>, duplicates: number }}
+ */
+export function dedupeEvents(events, dedupe) {
+  const fresh = [];
+  let duplicates = 0;
+  for (const event of events) {
+    if (dedupe.has(event.id)) {
+      duplicates += 1;
+      continue;
+    }
+    dedupe.add(event.id);
+    fresh.push(event);
+  }
+  return { fresh, duplicates };
+}
+
+/**
+ * Compute the ledger to resume from after a reorg. The cursor rewinds by
+ * `depth` ledgers (never below zero) so events in the replaced branch are
+ * re-read; dedupe removes any that were already counted.
+ *
+ * @param {number} confirmedLedger
+ * @param {number} [depth]
+ * @returns {number}
+ */
+export function reorgRewindLedger(confirmedLedger, depth = DEFAULT_REORG_DEPTH) {
+  if (!Number.isSafeInteger(confirmedLedger) || confirmedLedger < 0) {
+    throw new RangeError('confirmedLedger must be a non-negative safe integer');
+  }
+  if (!Number.isSafeInteger(depth) || depth < 0) {
+    throw new RangeError('depth must be a non-negative safe integer');
+  }
+  return Math.max(0, confirmedLedger - depth);
+}
+
 /**
  * Create a serialized poller.
  *
@@ -106,6 +233,7 @@ export function createPoller({ intervalMs, run, now = () => Date.now(), sleep, o
     lastDurationMs: null,
     lastSuccessAt: null,
     lastErrorAt: null,
+    lastError: null,
   };
 
   let inFlight = false;