diff --git a/.changeset/bounded-scheduled-ingest.md b/.changeset/bounded-scheduled-ingest.md
new file mode 100644
index 0000000..3cb5619
--- /dev/null
+++ b/.changeset/bounded-scheduled-ingest.md
@@ -0,0 +1,5 @@
+---
+"@atmo-dev/contrail": patch
+---
+
+Bound scheduled Jetstream cycles by retained candidate count and serialized bytes, drop exact transport observations before admission, preserve same-timestamp observations across capped restarts, capture empty initial cursors safely, reject rollback-prone endpoint pools in scheduled mode, and emit one bounded aggregate cycle summary.
diff --git a/docs/01-indexing.md b/docs/01-indexing.md
index f4117ba..001cae2 100644
--- a/docs/01-indexing.md
+++ b/docs/01-indexing.md
@@ -106,7 +106,21 @@ async scheduled(_ev, env, ctx) {
 }
 ```
 
-`ingest()` connects to Jetstream, streams events since the saved cursor, stops when caught up. Running every minute is fine — the next fire resumes where this one left off. Each cycle is bounded, so it can't blow past the Worker time limit.
+`ingest()` connects to Jetstream, streams events since the saved cursor, and stops at the live edge or the first scheduled collection threshold. Defaults are 25 seconds, 250 retained unique commit candidates, and 4 MiB of serialized candidate data. Byte accounting uses the UTF-8 record body plus a fixed 512-byte event-metadata allowance; the threshold-crossing candidate is retained, while deletes use only the allowance. Exact transport duplicates and source-filtered or identity events do not consume candidate/byte limits, but remain accounted for checkpointing.
+
+Override the limits when profiling a deployment:
+
+```ts
+await contrail.ingest({
+  maxDrainMs: 20_000,
+  maxCandidates: 200,
+  maxSerializedBytes: 3 * 1024 * 1024,
+}, env.DB);
+```
+
+Running every minute is fine—the next fire resumes from the last fully accounted cursor. Contrail stores bounded exact-observation hashes at the current microsecond cursor and resumes one microsecond earlier, so a cap can split equal-cursor items without skipping or recounting them. The cycle summary reports its stop reason, work counters, thresholds, safe cursor, and committed database sub-batches.
+
+Scheduled ingestion requires exactly one pinned Jetstream endpoint. Atcute intentionally rolls pooled connections back ten seconds on each new subscription; in a dense overlap that replay can consume every scheduled cap without moving forward. Use `runPersistent()` for multi-endpoint failover. Persistent ingestion is otherwise unchanged and does not inherit the scheduled count or byte defaults.
 
 **Local dev:** wrangler's cron scheduler only runs in deployed production. For local dev use `pnpm contrail dev` — it runs `wrangler dev --test-scheduled`, fires `/__scheduled` on your configured cron interval, and offers to start or resume backfill whenever known work remains.
 
diff --git a/packages/contrail/README.md b/packages/contrail/README.md
index 8a1e40d..7e042c1 100644
--- a/packages/contrail/README.md
+++ b/packages/contrail/README.md
@@ -61,6 +61,13 @@ HTTP routes expose the same pipeline, including relationship/reference hydration
 ```ts
 await contrail.ingest(); // bounded Jetstream cycle
 
+// Optional per-cycle overrides (defaults: 25s, 250 candidates, 4 MiB).
+await contrail.ingest({
+  maxDrainMs: 20_000,
+  maxCandidates: 200,
+  maxSerializedBytes: 3 * 1024 * 1024,
+});
+
 await contrail.runPersistent({
   signal: abortController.signal,
   batchSize: 50,
@@ -70,7 +77,7 @@ await contrail.runPersistent({
 
 After a write to a user's PDS, `contrail.notify(uri)` can fetch the authoritative record immediately. Only an authoritative not-found response deletes local state; rate limits, server errors, timeouts, malformed responses, and network failures leave it unchanged. Authentication and abuse controls for the public HTTP operation remain under design.
 
-Contrail stores source event time, repository revision, source cursor, CID, and local index time separately from record/application time. Durable tombstones reject stale resurrection, and live Jetstream projection commits its exact yielded cursor in the same transaction. A successful PDS `listRecords` page is a current authoritative observation, so it supersedes older durable state without a redundant version read; its version writes and page cursor still commit atomically. Tombstones are retained indefinitely; authoritative rebuild/retention tooling is planned separately.
+Contrail stores source event time, repository revision, source cursor, CID, and local index time separately from record/application time. Scheduled collection drops exact Jetstream observations before admission, counts UTF-8 record bodies plus a fixed 512-byte metadata allowance, and retains the threshold-crossing candidate before stopping. Identity and source-filtered observations consume neither candidate nor byte limits, but their yielded cursors remain accounted. Bounded exact-observation hashes at the current microsecond cursor let restarts replay one microsecond of overlap without skipping equal-cursor siblings or recounting earlier ones. Durable tombstones reject stale resurrection, and live Jetstream projection commits its exact accounted cursor in the same transaction. Scheduled mode requires one pinned Jetstream endpoint; use `runPersistent()` for an Atcute failover pool. Persistent ingestion retains its streaming batch lifecycle and does not inherit the scheduled count/byte defaults. A successful PDS `listRecords` page is a current authoritative observation, so it supersedes older durable state without a redundant version read; its version writes and page cursor still commit atomically. Tombstones are retained indefinitely; authoritative rebuild/retention tooling is planned separately.
 
 ## Local development
 
diff --git a/packages/contrail/src/contrail.ts b/packages/contrail/src/contrail.ts
index bba7572..82f03c5 100644
--- a/packages/contrail/src/contrail.ts
+++ b/packages/contrail/src/contrail.ts
@@ -22,8 +22,10 @@ import {
 } from "./core/db/records";
 import {
   createIngestState,
+  resolveScheduledIngestBudget,
   runIngestCycle,
   type IngestState,
+  type ScheduledIngestOptions,
 } from "./core/jetstream";
 import {
   backfillPending,
@@ -115,16 +117,17 @@ export class Contrail {
   }
 
   /** Run one ingestion cycle: catches up records from Jetstream and — when
-   *  `config.labels` is set — labels from each configured labeler in parallel.
-   *  Both share the same `timeoutMs` budget; they're independent network
-   *  operations so concurrency is free. */
-  async ingest(options?: { timeoutMs?: number }, db?: Database): Promise<void> {
+   * `config.labels` is set — labels from each configured labeler in parallel.
+   * Scheduled record collection is independently bounded by drain time,
+   * retained candidates, and serialized bytes. */
+  async ingest(options?: ScheduledIngestOptions, db?: Database): Promise<void> {
     const d = this.getDb(db);
+    const budget = resolveScheduledIngestBudget(options);
     const tasks: Promise<void>[] = [
-      runIngestCycle(d, this.config, options?.timeoutMs, this._ingestState),
+      runIngestCycle(d, this.config, budget, this._ingestState),
     ];
     if (this.config.labels) {
-      tasks.push(runLabelIngestCycle(d, this.config, options?.timeoutMs));
+      tasks.push(runLabelIngestCycle(d, this.config, budget.maxDrainMs));
     }
     await Promise.all(tasks);
   }
diff --git a/packages/contrail/src/core/db/index.ts b/packages/contrail/src/core/db/index.ts
index c772276..770eb3c 100644
--- a/packages/contrail/src/core/db/index.ts
+++ b/packages/contrail/src/core/db/index.ts
@@ -1,6 +1,6 @@
 export { initSchema, CONTRAIL_SCHEMA_VERSION } from "./schema";
 export { getMeta, setMeta, getMetaNumber } from "./meta";
 export { optimizeDatabase } from "./optimize";
-export { assertServingSourceCompatibility, getLastCursor, getServingSourcePosition, orderedSourcePosition, saveCursor, saveCursorStatement, saveOrderedSourcePositionStatement, saveServingSourcePositionStatement, lookupExistingRecords, queryRecords, pruneFeedItems, pruneActorFeed, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./records";
+export { assertServingSourceCompatibility, getLastCursor, getServingSourcePosition, orderedSourcePosition, saveCursor, saveCursorStatement, getCursorObservations, saveCursorObservationStatements, saveOrderedSourcePositionStatement, saveServingSourcePositionStatement, lookupExistingRecords, queryRecords, pruneFeedItems, pruneActorFeed, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./records";
 export type { QueryOptions, SortOption, ExistingRecordInfo, FeedSweepResult, ServingSourcePosition } from "./records";
 export type { RecordSource } from "../types";
diff --git a/packages/contrail/src/core/db/records.ts b/packages/contrail/src/core/db/records.ts
index e06cac0..04206ce 100644
--- a/packages/contrail/src/core/db/records.ts
+++ b/packages/contrail/src/core/db/records.ts
@@ -663,10 +663,50 @@ export function saveCursorStatement(
     .bind(timeUs);
 }
 
+/** Exact source observations already accounted for at the current coarse
+ * cursor. Scheduled Jetstream resumes one microsecond earlier and uses these
+ * hashes to avoid both skipping same-cursor siblings and recounting prior ones. */
+export async function getCursorObservations(
+  db: Database,
+  timeUs: number,
+): Promise<Set<string>> {
+  const rows = await db
+    .prepare(
+      "SELECT observation FROM cursor_observations WHERE time_us = ?",
+    )
+    .bind(timeUs)
+    .all<{ observation: string }>();
+  return new Set((rows.results ?? []).map((row) => row.observation));
+}
+
+/** Statements that atomically retire observations behind the monotonic cursor
+ * and union observations accounted for at its current timestamp. Inserts are
+ * conditional so an older concurrent cycle cannot attach hashes to a newer
+ * checkpoint. */
+export function saveCursorObservationStatements(
+  db: Database,
+  timeUs: number,
+  observations: Iterable<string>,
+): Statement[] {
+  return [
+    db.prepare(
+      "DELETE FROM cursor_observations WHERE time_us < (SELECT time_us FROM cursor WHERE id = 1)",
+    ),
+    ...[...new Set(observations)].map((observation) =>
+      db
+        .prepare(
+          "INSERT INTO cursor_observations (time_us, observation) SELECT ?, ? WHERE (SELECT time_us FROM cursor WHERE id = 1) = ? ON CONFLICT(time_us, observation) DO NOTHING",
+        )
+        .bind(timeUs, observation, timeUs),
+    ),
+  ];
+}
+
 export async function saveCursor(
   db: Database,
   timeUs: number,
   orderedSource?: OrderedSourceConfig,
+  observations: Iterable<string> = [],
 ): Promise<void> {
   const statements = [saveCursorStatement(db, timeUs)];
   if (orderedSource) {
@@ -674,6 +714,7 @@ export async function saveCursor(
       saveOrderedSourcePositionStatement(db, orderedSource, timeUs),
     );
   }
+  statements.push(...saveCursorObservationStatements(db, timeUs, observations));
   await db.batch(statements);
 }
 
diff --git a/packages/contrail/src/core/db/schema.ts b/packages/contrail/src/core/db/schema.ts
index 95fa42d..77cc78a 100644
--- a/packages/contrail/src/core/db/schema.ts
+++ b/packages/contrail/src/core/db/schema.ts
@@ -702,16 +702,32 @@ export async function initSchema(
   const indexes = buildDynamicIndexes(config, dialect);
   const feeds = buildFeedTables(config, dialect);
   const fts = buildFtsTables(config, dialect);
-  const fingerprint = schemaFingerprint(config, dialect, {
+  const baseFingerprint = schemaFingerprint(config, dialect, {
     base,
     collections,
     indexes,
     feeds,
     fts,
   });
-
-  if ((await getMeta(db, SCHEMA_FINGERPRINT_KEY)) === fingerprint) {
+  const fingerprint = `${baseFingerprint}:cursor-observations-v1`;
+  const cursorObservationsDdl = `CREATE TABLE IF NOT EXISTS cursor_observations (
+    time_us ${dialect.bigintType} NOT NULL,
+    observation TEXT NOT NULL,
+    PRIMARY KEY (time_us, observation)
+  )`;
+  const storedFingerprint = await getMeta(db, SCHEMA_FINGERPRINT_KEY);
+
+  if (storedFingerprint === fingerprint) {
+    for (const apply of options.extraSchemas ?? []) await apply(db);
+    return;
+  }
+  if (storedFingerprint === baseFingerprint) {
+    // This independent additive table must not force an otherwise-current
+    // deployment through the FTS rebuild path. Upgrade the prior fingerprint
+    // directly after its idempotent DDL succeeds.
+    await runIdempotentDdl(db, cursorObservationsDdl);
     for (const apply of options.extraSchemas ?? []) await apply(db);
+    await setMeta(db, SCHEMA_FINGERPRINT_KEY, fingerprint);
     return;
   }
 
@@ -725,6 +741,7 @@ export async function initSchema(
     }
   }
 
+  await runIdempotentDdl(db, cursorObservationsDdl);
   await applyFtsTables(db, config, dialect);
 
   const hasFeeds = !!(config.feeds && Object.keys(config.feeds).length > 0);
diff --git a/packages/contrail/src/core/ingest.ts b/packages/contrail/src/core/ingest.ts
index 48527cf..defbb11 100644
--- a/packages/contrail/src/core/ingest.ts
+++ b/packages/contrail/src/core/ingest.ts
@@ -3,6 +3,7 @@ import type {
   ContrailConfig,
   Database,
   IngestEvent,
+  Logger,
   MutationSource,
   Statement,
 } from "./types";
@@ -99,6 +100,13 @@ export function recordTimeUs(
   return microseconds > fallbackUs ? fallbackUs : microseconds;
 }
 
+export interface IngestWarningSamples {
+  /** Fixed maximum number of warning strings retained by the caller. */
+  maxSamples: number;
+  samples: string[];
+  omitted: number;
+}
+
 export interface IngestRecordsOptions {
   skipReplayDetection?: boolean;
   skipFeedFanout?: boolean;
@@ -117,6 +125,8 @@ export interface IngestRecordsOptions {
   /** @internal Aggregate private diagnostics for one bulk run. The caller
    * flushes this bounded object once after concurrent page processing. */
   aggregateDiagnostics?: IngestDiagnosticCounts;
+  /** Collect bounded warning details instead of logging per-record lines. */
+  warningSamples?: IngestWarningSamples;
 }
 
 export interface IngestDropCounts {
@@ -140,6 +150,22 @@ export interface IngestRecordsResult {
   discoveredDids: string[];
 }
 
+function emitIngestWarning(
+  logger: Pick<Logger, "warn">,
+  samples: IngestWarningSamples | undefined,
+  message: string,
+): void {
+  if (!samples) {
+    logger.warn(message);
+    return;
+  }
+  if (samples.samples.length >= samples.maxSamples) {
+    samples.omitted++;
+    return;
+  }
+  samples.samples.push(message.slice(0, 320));
+}
+
 /**
  * The single admission and projection path for records from every source.
  *
@@ -178,7 +204,9 @@ export async function ingestRecords(
     const shortName = resolveCollectionKey(config, event.collection);
     if (!shortName) {
       dropped.unknownCollection++;
-      logger.warn(
+      emitIngestWarning(
+        logger,
+        options.warningSamples,
         `[ingest] drop unknown collection: ${event.operation} ${event.uri} collection=${event.collection}`,
       );
       continue;
@@ -187,7 +215,11 @@ export async function ingestRecords(
       event.operation === "delete" ? null : parseRecord(event.record);
     if (event.operation !== "delete" && !record) {
       dropped.invalidRecord++;
-      logger.warn(`[ingest] drop invalid record: ${event.uri}`);
+      emitIngestWarning(
+        logger,
+        options.warningSamples,
+        `[ingest] drop invalid record: ${event.uri}`,
+      );
       continue;
     }
     candidates.push({ event, shortName, record });
@@ -249,7 +281,11 @@ export async function ingestRecords(
         try {
           keep = filter(record);
         } catch (error) {
-          logger.warn(`[ingest] recordFilter threw for ${event.uri}: ${error}`);
+          emitIngestWarning(
+            logger,
+            options.warningSamples,
+            `[ingest] recordFilter threw for ${event.uri}: ${error}`,
+          );
         }
         if (!keep) {
           dropped.recordFilter++;
@@ -270,7 +306,9 @@ export async function ingestRecords(
     dropped.cidEncoding +
     dropped.missingCid;
   if (validationDropTotal > 0 && !options.aggregateDiagnostics) {
-    logger.warn(
+    emitIngestWarning(
+      logger,
+      options.warningSamples,
       `[ingest] dropped ${validationDropTotal} record(s) during validation ` +
         `(lexicon=${dropped.lexiconValidation}, cid_mismatch=${dropped.cidMismatch}, ` +
         `cid_encoding=${dropped.cidEncoding}, missing_cid=${dropped.missingCid})`,
diff --git a/packages/contrail/src/core/jetstream.ts b/packages/contrail/src/core/jetstream.ts
index b29c8d1..5e953aa 100644
--- a/packages/contrail/src/core/jetstream.ts
+++ b/packages/contrail/src/core/jetstream.ts
@@ -1,6 +1,7 @@
 import { JetstreamSubscription } from "@atcute/jetstream";
 import type { ContrailConfig, IngestEvent, Database, Logger } from "./types";
 import {
+  DEFAULT_JETSTREAMS,
   getCollectionNsids,
   getDependentNsids,
   jetstreamUrlOption,
@@ -10,12 +11,121 @@ import {
   optimizeIntervalMs,
   optimizeAnalysisLimit,
 } from "./types";
-import { initSchema, getLastCursor, saveCursor, saveCursorStatement, saveOrderedSourcePositionStatement, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor, getMetaNumber, setMeta, optimizeDatabase } from "./db";
-import { createIngestEvent, ingestRecords, recordTimeUs } from "./ingest";
+import { initSchema, getLastCursor, saveCursor, saveCursorStatement, getCursorObservations, saveCursorObservationStatements, saveOrderedSourcePositionStatement, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor, getMetaNumber, setMeta, optimizeDatabase } from "./db";
+import {
+  createIngestEvent,
+  ingestRecords,
+  recordTimeUs,
+  type IngestDropCounts,
+  type IngestWarningSamples,
+} from "./ingest";
 import { refreshStaleIdentities, applyIdentityEvent } from "./identity";
 import { backfillFollowersFromConstellation } from "./constellation";
 
 const BATCH_SIZE = 50;
+
+/** Fixed estimate for the URI, source position, revision, CID, and other
+ * metadata retained beside each serialized record body. Deletes consume only
+ * this allowance. The byte budget is intentionally an admission threshold: the
+ * candidate that reaches it is retained, bounding overshoot to one record plus
+ * this allowance. */
+export const SCHEDULED_INGEST_METADATA_BYTES = 512;
+
+/** Conservative defaults for one D1/Worker scheduled drain. Persistent
+ * ingestion has its own streaming lifecycle and does not use these limits. */
+export const DEFAULT_SCHEDULED_INGEST_BUDGET = Object.freeze({
+  maxDrainMs: 25_000,
+  maxCandidates: 250,
+  maxSerializedBytes: 4 * 1024 * 1024,
+}) satisfies ScheduledIngestBudget;
+
+export interface ScheduledIngestBudget {
+  maxDrainMs: number;
+  maxCandidates: number;
+  maxSerializedBytes: number;
+}
+
+export interface ScheduledIngestOptions {
+  /** Maximum wall time spent requesting source items. Default: 25 seconds. */
+  maxDrainMs?: number;
+  /** Maximum unique commit candidates retained by one drain. Default: 250. */
+  maxCandidates?: number;
+  /** UTF-8 record bytes plus metadata allowances. Default: 4 MiB. */
+  maxSerializedBytes?: number;
+  /** @deprecated Compatibility alias for maxDrainMs. */
+  timeoutMs?: number;
+}
+
+export type ScheduledIngestStopReason =
+  | "head"
+  | "idle"
+  | "count"
+  | "bytes"
+  | "drain-time"
+  | "cancelled";
+
+export interface ScheduledIngestCollectionStats {
+  observedSourceItems: number;
+  commitObservations: number;
+  identityObservations: number;
+  retainedCandidates: number;
+  exactDuplicatesDropped: number;
+  cursorBoundaryDuplicatesDropped: number;
+  resumeOverlapDropped: number;
+  sourceScopeFiltered: number;
+  sourceInconsistencies: number;
+  serializedCandidateBytes: number;
+  startingCursor: number | null;
+  lastAccountedCursor: number | null;
+  safeEndingCursor: number | null;
+  stopReason: ScheduledIngestStopReason;
+  connections: number;
+  connectionCloses: number;
+  connectionErrors: number;
+  diagnosticSamples: string[];
+  diagnosticSamplesOmitted: number;
+}
+
+function positiveInteger(value: number, label: string): number {
+  if (!Number.isFinite(value) || !Number.isInteger(value) || value <= 0) {
+    throw new TypeError(`${label} must be a positive finite integer`);
+  }
+  return value;
+}
+
+/** Resolve and validate every scheduled collection threshold. */
+export function resolveScheduledIngestBudget(
+  value?: ScheduledIngestOptions | ScheduledIngestBudget | number,
+): ScheduledIngestBudget {
+  if (typeof value === "number") {
+    return {
+      ...DEFAULT_SCHEDULED_INGEST_BUDGET,
+      maxDrainMs: positiveInteger(value, "maxDrainMs"),
+    };
+  }
+  const options = (value ?? {}) as ScheduledIngestOptions;
+  if (options.timeoutMs !== undefined) {
+    positiveInteger(options.timeoutMs, "timeoutMs");
+  }
+  return {
+    maxDrainMs: positiveInteger(
+      options.maxDrainMs ??
+        options.timeoutMs ??
+        DEFAULT_SCHEDULED_INGEST_BUDGET.maxDrainMs,
+      "maxDrainMs",
+    ),
+    maxCandidates: positiveInteger(
+      options.maxCandidates ?? DEFAULT_SCHEDULED_INGEST_BUDGET.maxCandidates,
+      "maxCandidates",
+    ),
+    maxSerializedBytes: positiveInteger(
+      options.maxSerializedBytes ??
+        DEFAULT_SCHEDULED_INGEST_BUDGET.maxSerializedBytes,
+      "maxSerializedBytes",
+    ),
+  };
+}
+
 /** Distinct actors pruned per ingest tick by the rolling feed sweep. Each
  *  actor costs a handful of index-backed O(cap) deletes, so this bounds the
  *  prune's per-tick CPU regardless of how large feed_items grows. */
@@ -161,17 +271,15 @@ function getLogger(config: ContrailConfig): Logger {
 const INGEST_TIMEOUT = Symbol("ingest-timeout");
 
 /** Await the iterator's next value, but give up after `ms`. Without this a
- *  quiet Jetstream (the async iterator blocks forever waiting for an event that
- *  never arrives) holds the cycle past its safety timeout until the caller's
- *  hard timeout kills the isolate — so the batch and cursor are never written. */
+ * quiet Jetstream can hold a scheduled drain past its deadline. */
 function nextWithDeadline<T>(
   iterator: AsyncIterator<T>,
-  ms: number
+  ms: number,
 ): Promise<IteratorResult<T> | typeof INGEST_TIMEOUT> {
   let timer: ReturnType<typeof setTimeout>;
   const next = iterator.next();
-  // If the timeout wins this race the next() promise stays pending; swallow a
-  // later rejection so it can't surface as an unhandled rejection.
+  // The transport currently has no structural cancellation seam. If the timer
+  // wins, swallow a later rejection while the iterator is closed best-effort.
   next.catch(() => {});
   const timeout = new Promise<typeof INGEST_TIMEOUT>((resolve) => {
     timer = setTimeout(() => resolve(INGEST_TIMEOUT), ms);
@@ -179,71 +287,257 @@ function nextWithDeadline<T>(
   return Promise.race([next, timeout]).finally(() => clearTimeout(timer));
 }
 
+const utf8 = new TextEncoder();
+const MAX_DIAGNOSTIC_SAMPLES = 5;
+const MAX_DIAGNOSTIC_SAMPLE_LENGTH = 320;
+
+function addDiagnosticSample(
+  stats: Pick<
+    ScheduledIngestCollectionStats,
+    "diagnosticSamples" | "diagnosticSamplesOmitted"
+  >,
+  message: string,
+): void {
+  if (stats.diagnosticSamples.length >= MAX_DIAGNOSTIC_SAMPLES) {
+    stats.diagnosticSamplesOmitted++;
+    return;
+  }
+  stats.diagnosticSamples.push(message.slice(0, MAX_DIAGNOSTIC_SAMPLE_LENGTH));
+}
+
+/** JSON normalization used only for transport-observation fingerprints. The
+ * Jetstream payload has already been decoded from JSON, so sorting object keys
+ * makes semantically identical payloads stable across decoder/property order. */
+function normalizedJson(value: unknown): string {
+  if (value === null || typeof value !== "object") {
+    const encoded = JSON.stringify(value);
+    return encoded === undefined ? "null" : encoded;
+  }
+  if (Array.isArray(value)) {
+    return `[${value.map((item) => normalizedJson(item)).join(",")}]`;
+  }
+  const fields: string[] = [];
+  for (const key of Object.keys(value as Record<string, unknown>).sort()) {
+    const field = (value as Record<string, unknown>)[key];
+    if (field === undefined) continue;
+    fields.push(`${JSON.stringify(key)}:${normalizedJson(field)}`);
+  }
+  return `{${fields.join(",")}}`;
+}
+
+async function observationHash(value: string): Promise<string> {
+  const subtle = (
+    globalThis as typeof globalThis & {
+      crypto?: {
+        subtle?: {
+          digest(
+            algorithm: string,
+            data: Uint8Array,
+          ): Promise<ArrayBuffer>;
+        };
+      };
+    }
+  ).crypto?.subtle;
+  if (!subtle) throw new Error("Web Crypto SHA-256 is required for ingestion");
+  const digest = await subtle.digest("SHA-256", utf8.encode(value));
+  return [...new Uint8Array(digest)]
+    .map((byte) => byte.toString(16).padStart(2, "0"))
+    .join("");
+}
+
+interface JetstreamCommitObservation {
+  /** One logical Jetstream source slot, before source ID/epoch prefixing. */
+  key: string;
+  /** Stable operation/CID/normalized-payload identity for that slot. */
+  fingerprint: string;
+  uri: string;
+}
+
+/** Jetstream owns the transport-specific observation identity. Core collection
+ * adds the configured source ID and epoch before cycle-local deduplication. */
+function jetstreamCommitObservation(
+  event: {
+    did: string;
+    time_us: number;
+    commit: {
+      rev?: string;
+      collection: string;
+      rkey: string;
+      operation: string;
+      cid?: string;
+      record?: unknown;
+    };
+  },
+): JetstreamCommitObservation {
+  const { commit } = event;
+  const uri = `at://${event.did}/${commit.collection}/${commit.rkey}`;
+  return {
+    key: JSON.stringify([event.time_us, uri, commit.rev ?? null]),
+    fingerprint: JSON.stringify([
+      commit.operation,
+      commit.cid ?? null,
+      commit.operation === "delete" ? null : normalizedJson(commit.record),
+    ]),
+    uri,
+  };
+}
+
 export async function ingestEvents(
   config: ContrailConfig,
   cursor: number | null,
-  safetyTimeoutMs: number = 25_000,
-  knownDids?: Set<string>
+  budgetInput: ScheduledIngestOptions | ScheduledIngestBudget | number =
+    DEFAULT_SCHEDULED_INGEST_BUDGET,
+  knownDids?: Set<string>,
+  startingCursorObservations: ReadonlySet<string> = new Set(),
 ): Promise<{
   events: IngestEvent[];
   lastCursor: number | null;
+  cursorObservations: Set<string>;
   identityUpdates: Map<string, string>;
+  stats: ScheduledIngestCollectionStats;
 }> {
-  const log = getLogger(config);
+  const budget = resolveScheduledIngestBudget(budgetInput);
   const startTimeUs = Date.now() * 1000;
-  const deadline = Date.now() + safetyTimeoutMs;
+  const deadline = Date.now() + budget.maxDrainMs;
   const collected: IngestEvent[] = [];
 
   const collections = getCollectionNsids(config);
   const dependentCollections = new Set(getDependentNsids(config));
   const provisionalKnownDids = knownDids ? new Set(knownDids) : undefined;
-  const urls = config.jetstreams ?? [];
-
-  let totalCommits = 0;
-  let filteredUnknownDid = 0;
-  const filteredDidSamples = new Set<string>();
-  let lastYieldedTimeUs: number | null = null;
-  let firstYieldedTimeUs: number | null = null;
-  let connectCount = 0;
-  const seenUris = new Map<string, number>(); // uri -> time_us of first occurrence
-  const duplicateUris: string[] = [];
+  const urls = config.jetstreams ?? DEFAULT_JETSTREAMS;
+  const sourceId = config.orderedSource?.source ?? "jetstream";
+  const sourceEpoch = config.orderedSource?.epoch ?? null;
+
+  const seenObservations = new Map<string, Set<string>>();
   const identityUpdates = new Map<string, string>();
+  const stats: ScheduledIngestCollectionStats = {
+    observedSourceItems: 0,
+    commitObservations: 0,
+    identityObservations: 0,
+    retainedCandidates: 0,
+    exactDuplicatesDropped: 0,
+    cursorBoundaryDuplicatesDropped: 0,
+    resumeOverlapDropped: 0,
+    sourceScopeFiltered: 0,
+    sourceInconsistencies: 0,
+    serializedCandidateBytes: 0,
+    startingCursor: cursor,
+    lastAccountedCursor: null,
+    safeEndingCursor: cursor,
+    stopReason: "idle",
+    connections: 0,
+    connectionCloses: 0,
+    connectionErrors: 0,
+    diagnosticSamples: [],
+    diagnosticSamplesOmitted: 0,
+  };
 
+  // A durable timestamp cursor is resumed one microsecond earlier. This works
+  // with inclusive and exclusive timestamp APIs: all observations at the
+  // coarse cursor are replayed, while persisted hashes suppress only the exact
+  // items already accounted for there. It also protects a captured empty-start
+  // cursor from a later item that happens to share its microsecond.
+  const requestedCursor =
+    cursor === null ? null : Math.max(0, cursor - 1);
   const subscription = new JetstreamSubscription({
     // A single-instance config is handed over as a string so @atcute skips its
-    // array-only first-connect cursor rollback (see jetstreamUrlOption). On the
-    // cron model that rollback would otherwise re-ingest 10s every cycle.
+    // array-only first-connect cursor rollback (see jetstreamUrlOption).
     url: jetstreamUrlOption(urls),
     wantedCollections: collections,
-    ...(cursor !== null ? { cursor } : {}),
+    ...(requestedCursor !== null ? { cursor: requestedCursor } : {}),
     onConnectionOpen() {
-      connectCount++;
-      log.log(
-        `[ingest] connected to Jetstream #${connectCount} (url=${urls.join("|")}, cursor=${cursor ?? "none"}, wanted=${collections.join(",")})`
-      );
+      stats.connections++;
     },
-    onConnectionClose(event) {
-      log.log(
-        `[ingest] disconnected from Jetstream: ${event.code} ${event.reason}`
-      );
+    onConnectionClose() {
+      stats.connectionCloses++;
     },
     onConnectionError(event) {
-      log.error("[ingest] Jetstream error:", event.error);
+      stats.connectionErrors++;
+      addDiagnosticSample(stats, `Jetstream connection error: ${String(event.error)}`);
     },
   });
 
+  // Capture Atcute's constructor cursor before iteration can buffer frames and
+  // move it ahead. With no durable cursor this is the subscription's effective
+  // lower bound and must be persisted even when the first drain stays empty.
+  const effectiveStartCursor = cursor ?? subscription.cursor ?? null;
+  stats.safeEndingCursor = effectiveStartCursor;
+  let boundaryCursor = effectiveStartCursor;
+  let boundaryObservations = new Set(startingCursorObservations);
+  // Only hashes newly accounted for this cycle are returned for insertion. The
+  // database unions them with existing boundary rows, keeping each DB batch
+  // bounded even if many cycles share one coarse cursor.
+  let cursorObservations = new Set<string>();
+  const singleEndpoint = urls.length === 1;
+
   const iterator = subscription[Symbol.asyncIterator]();
   type Ev = typeof subscription extends AsyncIterable<infer V> ? V : never;
 
-  // Collect or skip a single event. Filtering uses early `return` rather than
-  // the loop's `continue` so the loop's exit checks still run after a filtered
-  // event (a stream of all-filtered events must not skip the deadline).
-  const handleEvent = (event: Ev): void => {
+  const accountSourceItem = async (
+    event: Ev,
+    identity: string,
+  ): Promise<boolean> => {
+    // The scheduled path requires one pinned endpoint. Its one-microsecond
+    // boundary overlap is known-complete below the durable cursor and must not
+    // consume candidate/byte budgets on every restart.
+    if (
+      singleEndpoint &&
+      effectiveStartCursor !== null &&
+      event.time_us < effectiveStartCursor
+    ) {
+      stats.resumeOverlapDropped++;
+      return true;
+    }
+
+    const hash = await observationHash(
+      JSON.stringify([sourceId, sourceEpoch, identity]),
+    );
+    if (boundaryCursor === null || event.time_us > boundaryCursor) {
+      boundaryCursor = event.time_us;
+      boundaryObservations = new Set([hash]);
+      cursorObservations = new Set([hash]);
+    } else if (event.time_us === boundaryCursor) {
+      const alreadyAccounted = boundaryObservations.has(hash);
+      boundaryObservations.add(hash);
+      if (!alreadyAccounted) cursorObservations.add(hash);
+    }
+
+    if (
+      effectiveStartCursor !== null &&
+      event.time_us === effectiveStartCursor &&
+      startingCursorObservations.has(hash)
+    ) {
+      stats.cursorBoundaryDuplicatesDropped++;
+      return true;
+    }
+    return false;
+  };
+
+  // Fully account for one yielded item before the loop considers any stop
+  // threshold. Exact boundary observations are suppressed before evolving
+  // source-scope policy; ordinary cheap filters still precede cycle dedupe.
+  const handleEvent = async (event: Ev): Promise<void> => {
     if (event.kind === "commit") {
       const { commit } = event;
-      totalCommits++;
-
-      const uri = `at://${event.did}/${commit.collection}/${commit.rkey}`;
+      stats.commitObservations++;
+      const observation = jetstreamCommitObservation(event);
+      const dedupeKey = JSON.stringify([
+        sourceId,
+        sourceEpoch,
+        observation.key,
+      ]);
+      if (
+        await accountSourceItem(
+          event,
+          JSON.stringify([
+            "commit",
+            observation.key,
+            observation.fingerprint,
+          ]),
+        )
+      ) {
+        return;
+      }
 
       if (
         provisionalKnownDids &&
@@ -257,89 +551,104 @@ export async function ingestEvents(
         provisionalKnownDids &&
         !provisionalKnownDids.has(event.did)
       ) {
-        filteredUnknownDid++;
-        if (filteredDidSamples.size < 10) filteredDidSamples.add(event.did);
+        stats.sourceScopeFiltered++;
         return;
       }
 
-      const prev = seenUris.get(uri);
-      if (prev !== undefined) {
-        duplicateUris.push(uri);
-        log.warn(
-          `[ingest] DUPLICATE in cycle: ${uri} first time_us=${prev}, again=${event.time_us}, delta=${event.time_us - prev}us`
+      const fingerprints = seenObservations.get(dedupeKey);
+      if (fingerprints?.has(observation.fingerprint)) {
+        stats.exactDuplicatesDropped++;
+        return;
+      }
+      if (fingerprints) {
+        stats.sourceInconsistencies++;
+        addDiagnosticSample(
+          stats,
+          `source slot changed payload: ${observation.uri} time_us=${event.time_us}`,
         );
+        fingerprints.add(observation.fingerprint);
       } else {
-        seenUris.set(uri, event.time_us);
+        seenObservations.set(dedupeKey, new Set([observation.fingerprint]));
       }
 
-      collected.push(
-        createIngestEvent({
-          did: event.did,
-          timeUs:
-            commit.operation === "delete"
-              ? event.time_us
-              : recordTimeUs(
-                  commit.record,
-                  commit.collection,
-                  config,
-                  event.time_us,
-                ),
-          collection: commit.collection,
-          operation: commit.operation,
-          rkey: commit.rkey,
-          cid: commit.operation === "delete" ? null : commit.cid,
-          value: commit.operation === "delete" ? undefined : commit.record,
-          source: {
-            id: config.orderedSource?.source ?? "jetstream",
-            ...(config.orderedSource
-              ? { epoch: config.orderedSource.epoch }
-              : {}),
-            time_us: event.time_us,
-            revision: commit.rev,
-            cursor: String(event.time_us),
-          },
-        }),
-      );
-
-      log.log(
-        `[ingest] candidate: ${commit.operation} ${uri} time_us=${event.time_us}`
-      );
-
+      const candidate = createIngestEvent({
+        did: event.did,
+        timeUs:
+          commit.operation === "delete"
+            ? event.time_us
+            : recordTimeUs(
+                commit.record,
+                commit.collection,
+                config,
+                event.time_us,
+              ),
+        collection: commit.collection,
+        operation: commit.operation,
+        rkey: commit.rkey,
+        cid: commit.operation === "delete" ? null : commit.cid,
+        value: commit.operation === "delete" ? undefined : commit.record,
+        source: {
+          id: sourceId,
+          ...(sourceEpoch === null ? {} : { epoch: sourceEpoch }),
+          time_us: event.time_us,
+          revision: commit.rev,
+          cursor: String(event.time_us),
+        },
+      });
+      collected.push(candidate);
+      stats.retainedCandidates++;
+      stats.serializedCandidateBytes +=
+        SCHEDULED_INGEST_METADATA_BYTES +
+        (candidate.record === null ? 0 : utf8.encode(candidate.record).byteLength);
     } else if (event.kind === "identity") {
+      stats.identityObservations++;
+      if (await accountSourceItem(event, normalizedJson(event))) return;
       identityUpdates.set(event.did, event.identity.handle);
+    } else {
+      await accountSourceItem(event, normalizedJson(event));
     }
   };
 
   for (;;) {
-    // Run the exit checks BEFORE awaiting the next event and regardless of
-    // whether the previous event was filtered — otherwise a quiet stream blocks
-    // forever and an all-filtered flood never reaches the deadline check.
-    if (Date.now() >= deadline) {
-      log.log(
-        `[ingest] safety timeout reached, stopping (deadline=${deadline}, collected=${collected.length})`
-      );
+    // Check the deadline before requesting another item. A hot iterator must not
+    // get one extra next() after any threshold has been reached.
+    const remainingMs = deadline - Date.now();
+    if (remainingMs <= 0) {
+      stats.stopReason = "drain-time";
       break;
     }
 
-    const step = await nextWithDeadline(iterator, Math.max(0, deadline - Date.now()));
+    const step = await nextWithDeadline(iterator, remainingMs);
     if (step === INGEST_TIMEOUT) {
-      log.log(
-        `[ingest] safety timeout reached, stopping (deadline=${deadline}, collected=${collected.length})`
-      );
+      stats.stopReason = "drain-time";
+      break;
+    }
+    if (step.done) {
+      stats.stopReason = "idle";
       break;
     }
-    if (step.done) break;
     const event = step.value;
+    stats.observedSourceItems++;
+
+    await handleEvent(event);
+    // handleEvent either retained, filtered, deduplicated, or deliberately
+    // handled the item. Only now is its cursor safe to checkpoint. Keep the
+    // representable checkpoint monotonic across deliberate overlap replay.
+    stats.lastAccountedCursor = Math.max(
+      stats.lastAccountedCursor ?? event.time_us,
+      event.time_us,
+    );
 
-    if (firstYieldedTimeUs === null) firstYieldedTimeUs = event.time_us;
-    lastYieldedTimeUs = event.time_us;
-
-    handleEvent(event);
-
+    if (stats.retainedCandidates >= budget.maxCandidates) {
+      stats.stopReason = "count";
+      break;
+    }
+    if (stats.serializedCandidateBytes >= budget.maxSerializedBytes) {
+      stats.stopReason = "bytes";
+      break;
+    }
     if (event.time_us >= startTimeUs) {
-      log.log(
-        `[ingest] caught up to present, stopping (last time_us=${event.time_us}, startTimeUs=${startTimeUs})`
-      );
+      stats.stopReason = "head";
       break;
     }
   }
@@ -348,74 +657,98 @@ export async function ingestEvents(
   // iterator's return on a quiet stream could itself block (the hang we fix).
   Promise.resolve(iterator.return?.()).catch(() => {});
 
-  if (filteredUnknownDid > 0) {
-    const sample = [...filteredDidSamples].join(", ");
-    log.log(
-      `[ingest] ${filteredUnknownDid} events filtered (unknown did). sample dids: ${sample}`
-    );
-  }
-  const subscriptionCursor = subscription.cursor || null;
-  // @atcute may advance its internal cursor when an event enters its buffer,
-  // before the async iterator yields that event. Persist only through the last
-  // event Contrail actually observed; anything buffered is deliberately replayed.
+  // Never read Atcute's cursor again here: it may have moved when a frame was
+  // buffered but not yielded. The constructor cursor captured above and the
+  // maximum fully-accounted yielded cursor are the only safe positions.
   const lastCursor =
-    lastYieldedTimeUs === null
-      ? subscriptionCursor
-      : subscriptionCursor === null
-        ? lastYieldedTimeUs
-        : Math.min(subscriptionCursor, lastYieldedTimeUs);
-
-  const cursorGap =
-    subscriptionCursor !== null && lastYieldedTimeUs !== null
-      ? subscriptionCursor - lastYieldedTimeUs
-      : null;
-
-  // Detect the library's internal cursor rollback (picks a different URL → rolls
-  // back 10s → first event comes in BEFORE the cursor we asked it to start from).
-  const rolledBackUs =
-    cursor !== null && firstYieldedTimeUs !== null && firstYieldedTimeUs < cursor
-      ? cursor - firstYieldedTimeUs
-      : 0;
-
-  log.log(
-    `[ingest] jetstream loop done. commits_seen=${totalCommits}, filtered=${filteredUnknownDid}, candidates=${collected.length}, dupes=${duplicateUris.length}, connects=${connectCount}, first_yielded=${firstYieldedTimeUs ?? "none"}, last_yielded=${lastYieldedTimeUs ?? "none"}, subscription_cursor=${subscriptionCursor ?? "none"}, safe_cursor=${lastCursor ?? "none"}, cursor_gap=${cursorGap ?? "n/a"}us, rolled_back=${rolledBackUs}us`
-  );
-
-  if (cursorGap !== null && cursorGap > 1000) {
-    log.warn(
-      `[ingest] CURSOR GAP: subscription cursor is ${cursorGap}us (${Math.floor(
-        cursorGap / 1000
-      )}ms) ahead of last yielded event — saving safe_cursor=${lastCursor ?? "none"}; buffered events will replay`
-    );
+    effectiveStartCursor === null
+      ? stats.lastAccountedCursor
+      : Math.max(
+          effectiveStartCursor,
+          stats.lastAccountedCursor ?? effectiveStartCursor,
+        );
+  stats.safeEndingCursor = lastCursor;
+  if (lastCursor !== boundaryCursor) {
+    // This occurs only for an older deliberate overlap. Leave the durable
+    // boundary unchanged rather than attaching old hashes to a newer cursor.
+    cursorObservations = new Set();
   }
 
-  if (connectCount > 1) {
-    if (urls.length > 1) {
-      // Multi-instance pool: each reconnect picks a URL at random, and @atcute
-      // rolls the cursor back up to 10s on a fresh instance to absorb clock skew.
-      log.warn(
-        `[ingest] RECONNECTED ${connectCount} times during cycle across a ${urls.length}-instance pool — each reconnect picks a URL at random and may roll the cursor back up to 10s (rolled_back=${rolledBackUs}us this cycle)`
-      );
-    } else {
-      // Single fixed instance (see jetstreamUrlOption): reconnects resume on the
-      // same instance from the saved cursor, so there is no rollback.
-      log.log(
-        `[ingest] reconnected ${connectCount} times during cycle to the single fixed instance — no cursor rollback (rolled_back=${rolledBackUs}us)`
-      );
-    }
+  return {
+    events: collected,
+    lastCursor,
+    cursorObservations,
+    identityUpdates,
+    stats,
+  };
+}
+
+function emptyDropCounts(): IngestDropCounts {
+  return {
+    unknownCollection: 0,
+    invalidRecord: 0,
+    lexiconValidation: 0,
+    cidMismatch: 0,
+    cidEncoding: 0,
+    missingCid: 0,
+    recordFilter: 0,
+    unknownActor: 0,
+    unknownSubject: 0,
+    superseded: 0,
+  };
+}
+
+function addDropCounts(target: IngestDropCounts, value: IngestDropCounts): void {
+  for (const key of Object.keys(target) as Array<keyof IngestDropCounts>) {
+    target[key] += value[key];
   }
+}
 
-  return { events: collected, lastCursor, identityUpdates };
+function admissionFilteredCount(dropped: IngestDropCounts): number {
+  return Object.entries(dropped).reduce(
+    (total, [key, value]) => key === "superseded" ? total : total + value,
+    0,
+  );
+}
+
+function runtimeCpuUsage(): (() => number) | null {
+  const runtimeProcess = (
+    globalThis as typeof globalThis & {
+      process?: {
+        cpuUsage?: (previous?: { user: number; system: number }) => {
+          user: number;
+          system: number;
+        };
+      };
+    }
+  ).process;
+  if (!runtimeProcess?.cpuUsage) return null;
+  const start = runtimeProcess.cpuUsage();
+  return () => {
+    const elapsed = runtimeProcess.cpuUsage!(start);
+    return Math.round((elapsed.user + elapsed.system) / 100) / 10;
+  };
 }
 
 // Run a full ingest cycle: init schema, load cursor, ingest, apply, save cursor
 export async function runIngestCycle(
   db: Database,
   config: ContrailConfig,
-  timeoutMs: number = 25_000,
+  budgetInput: ScheduledIngestOptions | ScheduledIngestBudget | number =
+    DEFAULT_SCHEDULED_INGEST_BUDGET,
   state?: IngestState,
 ): Promise<void> {
   const log = getLogger(config);
+  const budget = resolveScheduledIngestBudget(budgetInput);
+  const scheduledUrls = config.jetstreams ?? DEFAULT_JETSTREAMS;
+  if (scheduledUrls.length !== 1) {
+    throw new TypeError(
+      "scheduled ingestion requires exactly one pinned Jetstream endpoint; " +
+        "use runPersistent() for a failover pool",
+    );
+  }
+  const wallStartedAt = Date.now();
+  const finishCpuUsage = runtimeCpuUsage();
   const s = state ?? createIngestState();
 
   if (!s.schemaInitialized) {
@@ -424,15 +757,8 @@ export async function runIngestCycle(
   }
 
   const cursor = await getLastCursor(db);
-  const collections = getCollectionNsids(config);
-  const nowUs = Date.now() * 1000;
-  const lagMs = cursor !== null ? Math.floor((nowUs - cursor) / 1000) : null;
-
-  log.log(
-    `[ingest] starting cycle. cursor=${cursor ?? "none"}${
-      lagMs !== null ? ` (lag=${lagMs}ms)` : ""
-    }, timeout=${timeoutMs}ms, collections=${collections.join(", ")}`
-  );
+  const startingCursorObservations =
+    cursor === null ? new Set<string>() : await getCursorObservations(db, cursor);
 
   // Load known DIDs for filtering dependent collections
   const dependentCollections = getDependentNsids(config);
@@ -441,44 +767,44 @@ export async function runIngestCycle(
   if (dependentCollections.length > 0) {
     if (s.cachedKnownDids) {
       knownDids = s.cachedKnownDids;
-      log.log(`Using cached known DIDs (${knownDids.size} users)`);
     } else {
       const result = await db
         .prepare("SELECT did FROM identities")
         .all<{ did: string }>();
       knownDids = new Set((result.results ?? []).map((r) => r.did));
       s.cachedKnownDids = knownDids;
-      log.log(`Loaded ${knownDids.size} known DIDs from database`);
     }
   }
 
-  const { events, lastCursor, identityUpdates } = await ingestEvents(
+  const {
+    events,
+    lastCursor,
+    cursorObservations,
+    identityUpdates,
+    stats,
+  } = await ingestEvents(
     config,
     cursor,
-    timeoutMs,
-    knownDids
+    budget,
+    knownDids,
+    startingCursorObservations,
   );
 
-  if (events.length > 0) {
-    const breakdown: Record<string, number> = {};
-    for (const e of events) {
-      const key = `${e.collection}:${e.operation}`;
-      breakdown[key] = (breakdown[key] ?? 0) + 1;
-    }
-    log.log(
-      `[ingest] received ${events.length} events. breakdown=${JSON.stringify(breakdown)}`
-    );
-  } else {
-    log.log(`[ingest] received 0 events from Jetstream`);
-  }
-
   const accepted: IngestEvent[] = [];
   const newlyKnownDids: string[] = [];
+  const dropped = emptyDropCounts();
+  let databaseSubBatchesCommitted = 0;
+  const warningSamples: IngestWarningSamples = {
+    maxSamples: MAX_DIAGNOSTIC_SAMPLES,
+    samples: stats.diagnosticSamples,
+    omitted: stats.diagnosticSamplesOmitted,
+  };
   for (let i = 0; i < events.length; i += BATCH_SIZE) {
     const batch = events.slice(i, i + BATCH_SIZE);
     const isFinalBatch = i + BATCH_SIZE >= events.length;
     const result = await ingestRecords(db, batch, config, {
       knownDids,
+      warningSamples,
       // Earlier batches may commit without moving the cursor. A crash replays
       // them safely; the final batch atomically commits the exact source cursor.
       trailingStatements:
@@ -494,10 +820,17 @@ export async function runIngestCycle(
                     ),
                   ]
                 : []),
+              ...saveCursorObservationStatements(
+                db,
+                lastCursor,
+                cursorObservations,
+              ),
             ]
           : undefined,
     });
+    databaseSubBatchesCommitted++;
     accepted.push(...result.accepted);
+    addDropCounts(dropped, result.dropped);
     if (knownDids) {
       for (const did of result.discoveredDids) {
         knownDids.add(did);
@@ -506,54 +839,97 @@ export async function runIngestCycle(
     }
   }
 
-  // Apply handle changes from #identity events. UPDATE-only, so unknown
-  // DIDs are no-ops — we don't want to create partial rows lacking PDS.
-  if (identityUpdates.size > 0) {
-    for (const [did, handle] of identityUpdates) {
-      try {
-        await applyIdentityEvent(db, did, handle);
-      } catch (err) {
-        log.warn(`[ingest] identity update failed for ${did}: ${err}`);
+  // Apply handle changes from #identity events. UPDATE-only, so unknown DIDs
+  // are no-ops. Failures are sampled into the bounded cycle summary.
+  let identityUpdateFailures = 0;
+  for (const [did, handle] of identityUpdates) {
+    try {
+      await applyIdentityEvent(db, did, handle);
+    } catch (error) {
+      identityUpdateFailures++;
+      if (warningSamples.samples.length < warningSamples.maxSamples) {
+        warningSamples.samples.push(
+          `identity update failed for ${did}: ${String(error)}`.slice(
+            0,
+            MAX_DIAGNOSTIC_SAMPLE_LENGTH,
+          ),
+        );
+      } else {
+        warningSamples.omitted++;
       }
     }
-    log.log(`[ingest] applied ${identityUpdates.size} identity event(s)`);
   }
 
   // A commit batch saves its source cursor in the projection transaction above.
-  // An identity-only or fully filtered stream has no canonical record batch, so
-  // its cursor can advance independently after best-effort identity handling.
-  if (lastCursor !== null && events.length === 0) {
-    await saveCursor(db, lastCursor, config.orderedSource);
-  }
-  if (lastCursor !== null) {
-    log.log(
-      `[ingest] saved cursor=${lastCursor} (advanced ${
-        cursor !== null ? lastCursor - cursor : "n/a"
-      }us, atomic=${events.length > 0})`,
+  // Identity-only, filtered-only, or otherwise candidate-free accounted ranges
+  // checkpoint only after best-effort identity handling. No observed item means
+  // there is no new source work to save, even if the subscription buffered ahead.
+  if (
+    lastCursor !== null &&
+    events.length === 0 &&
+    // Save an accounted range, or capture Atcute's constructor cursor for an
+    // empty first drain so the next cron cannot silently start later.
+    (stats.lastAccountedCursor !== null || cursor === null)
+  ) {
+    await saveCursor(
+      db,
+      lastCursor,
+      config.orderedSource,
+      cursorObservations,
     );
-  } else {
-    log.log(`[ingest] no cursor returned from subscription; not saving`);
+    databaseSubBatchesCommitted++;
   }
 
   // Refresh stale/missing identities for DIDs in this batch (best-effort; runs
   // after the cursor save so its network latency can't strand forward progress).
   const uniqueDids = [...new Set(accepted.map((e) => e.did))];
+  let identityRefreshFailures = 0;
   if (uniqueDids.length > 0) {
     try {
       await refreshStaleIdentities(db, uniqueDids, config);
-    } catch (err) {
-      log.warn(`Identity refresh failed: ${err}`);
+    } catch (error) {
+      identityRefreshFailures++;
+      if (warningSamples.samples.length < warningSamples.maxSamples) {
+        warningSamples.samples.push(
+          `identity refresh failed: ${String(error)}`.slice(
+            0,
+            MAX_DIAGNOSTIC_SAMPLE_LENGTH,
+          ),
+        );
+      } else {
+        warningSamples.omitted++;
+      }
     }
   }
 
-  // Newly-discovered DIDs: ask Constellation for back-edges so they
-  // immediately appear in existing followers' feeds (best-effort, opt-out).
+  // Newly-discovered DIDs: ask Constellation for back-edges. Suppress helper
+  // per-subject logs and report bounded aggregate results in the cycle summary.
+  let constellationFailures = 0;
+  let constellationInserted = 0;
   if (config.feeds && newlyKnownDids.length > 0) {
+    const quietConfig = {
+      ...config,
+      logger: { log() {}, warn() {}, error() {} },
+    };
     for (const subj of newlyKnownDids) {
       try {
-        await backfillFollowersFromConstellation(db, config, subj);
-      } catch (err) {
-        log.warn(`[constellation] subject=${subj} failed: ${err}`);
+        constellationInserted += await backfillFollowersFromConstellation(
+          db,
+          quietConfig,
+          subj,
+        );
+      } catch (error) {
+        constellationFailures++;
+        if (warningSamples.samples.length < warningSamples.maxSamples) {
+          warningSamples.samples.push(
+            `constellation failed for ${subj}: ${String(error)}`.slice(
+              0,
+              MAX_DIAGNOSTIC_SAMPLE_LENGTH,
+            ),
+          );
+        } else {
+          warningSamples.omitted++;
+        }
       }
     }
   }
@@ -586,5 +962,41 @@ export async function runIngestCycle(
   // config.maintenance.optimize is set).
   await maybeOptimize(db, config, log);
 
-  log.log(`[ingest] cycle complete. stored=${accepted.length}`);
+  const summary = {
+    observed_source_items: stats.observedSourceItems,
+    commit_observations: stats.commitObservations,
+    identity_observations: stats.identityObservations,
+    retained_candidates: stats.retainedCandidates,
+    exact_duplicates_dropped: stats.exactDuplicatesDropped,
+    cursor_boundary_duplicates_dropped:
+      stats.cursorBoundaryDuplicatesDropped,
+    resume_overlap_dropped: stats.resumeOverlapDropped,
+    source_scope_filtered: stats.sourceScopeFiltered,
+    admission_policy_filtered: admissionFilteredCount(dropped),
+    candidates_superseded: dropped.superseded,
+    source_inconsistencies: stats.sourceInconsistencies,
+    serialized_candidate_bytes: stats.serializedCandidateBytes,
+    max_candidates: budget.maxCandidates,
+    max_serialized_bytes: budget.maxSerializedBytes,
+    max_drain_ms: budget.maxDrainMs,
+    starting_cursor: cursor,
+    last_accounted_cursor: stats.lastAccountedCursor,
+    safe_ending_cursor: stats.safeEndingCursor,
+    stop_reason: stats.stopReason,
+    database_sub_batches_committed: databaseSubBatchesCommitted,
+    projected_mutations_accepted: accepted.length,
+    identity_updates_attempted: identityUpdates.size,
+    identity_update_failures: identityUpdateFailures,
+    identity_refresh_failures: identityRefreshFailures,
+    constellation_inserted: constellationInserted,
+    constellation_failures: constellationFailures,
+    connections: stats.connections,
+    connection_closes: stats.connectionCloses,
+    connection_errors: stats.connectionErrors,
+    wall_ms: Date.now() - wallStartedAt,
+    cpu_ms: finishCpuUsage?.() ?? null,
+    diagnostic_samples: warningSamples.samples,
+    diagnostic_samples_omitted: warningSamples.omitted,
+  };
+  log.log(`[ingest] cycle summary ${JSON.stringify(summary)}`);
 }
diff --git a/packages/contrail/src/core/types.ts b/packages/contrail/src/core/types.ts
index c472054..f46691e 100644
--- a/packages/contrail/src/core/types.ts
+++ b/packages/contrail/src/core/types.ts
@@ -275,10 +275,11 @@ export interface ContrailConfig {
   /** Jetstream endpoints to ingest from (defaults to {@link DEFAULT_JETSTREAMS}).
    *  Prefer a single endpoint: one instance has no clock skew, so `@atcute` takes
    *  no cursor rollback (see {@link jetstreamUrlOption}) — important for the cron
-   *  model, which rebuilds the subscription every cycle. Use 2+ only for failover
-   *  across interchangeable endpoints, and ideally only with a persistent
-   *  connection (`runPersistent`), where the per-switch 10s skew rollback fires
-   *  about once rather than every cycle. */
+   *  model, which rebuilds the subscription every cycle. Scheduled ingestion
+   *  requires exactly one pinned endpoint because a pooled connection rolls its
+   *  timestamp cursor back on every new cycle and cannot make bounded progress
+   *  through a dense overlap. Use 2+ only with `runPersistent()`, where the
+   *  connection and its failover cursor remain long-lived. */
   jetstreams?: string[];
   /** Identity of the ordered source consumed by live ingestion. Its opaque
    * cursor is persisted atomically with projected mutations and may be exposed
diff --git a/packages/contrail/src/worker/index.ts b/packages/contrail/src/worker/index.ts
index a75f0a0..17f97ff 100644
--- a/packages/contrail/src/worker/index.ts
+++ b/packages/contrail/src/worker/index.ts
@@ -18,6 +18,7 @@ import { Contrail } from "../contrail.js";
 import { createHandler } from "../server.js";
 import type { ContrailConfig, Database } from "../core/types.js";
 import type { BackfillRetryOptions } from "../core/backfill.js";
+import type { ScheduledIngestOptions } from "../core/jetstream.js";
 import {
   normalizePublicServiceEndpoint,
   validatePublicServiceAuthEndpoint,
@@ -33,6 +34,8 @@ export interface CreateWorkerOptions {
   lexicons?: object[];
   /** Enable stable discovery and Lexicon routes for anonymous remote clients. */
   publicService?: PublicServiceOptions;
+  /** Count, byte, and drain-time limits for each scheduled Jetstream cycle. */
+  scheduledIngest?: ScheduledIngestOptions;
   /** Bounded pending-account retry slice after each scheduled ingest. Enabled
    *  by default; pass `false` to disable or options to tune its budget. */
   backfillRetries?: BackfillRetryOptions | false;
@@ -84,7 +87,7 @@ export function createWorker(
       // failures. A database lease prevents overlap with a manual backfill.
       ctx.waitUntil(
         (async () => {
-          await contrail.ingest({}, db);
+          await contrail.ingest(options.scheduledIngest, db);
           if (options.backfillRetries !== false) {
             await contrail.retryBackfill(options.backfillRetries, db);
           }
diff --git a/packages/contrail/tests/ingest-hang.test.ts b/packages/contrail/tests/ingest-hang.test.ts
index 8d7ee34..2c7bebd 100644
--- a/packages/contrail/tests/ingest-hang.test.ts
+++ b/packages/contrail/tests/ingest-hang.test.ts
@@ -27,7 +27,11 @@ vi.mock("@atcute/jetstream", () => {
   return { JetstreamSubscription: MockJetstreamSubscription };
 });
 
-import { ingestEvents } from "../src/index";
+import {
+  ingestEvents,
+  resolveScheduledIngestBudget,
+  SCHEDULED_INGEST_METADATA_BYTES,
+} from "../src/index";
 import { resolveConfig } from "../src/index";
 import type { ContrailConfig } from "../src/index";
 
@@ -38,22 +42,44 @@ function commitEvent(
   collection: string,
   time_us: number,
   rkey: string,
+  options: {
+    revision?: string;
+    record?: Record<string, unknown>;
+    cid?: string;
+  } = {},
 ) {
   return {
     kind: "commit" as const,
     time_us,
     did,
     commit: {
-      rev: String(time_us),
+      rev: options.revision ?? String(time_us),
       collection,
       operation: "create" as const,
       rkey,
-      cid: "bafy" + rkey,
-      record: { name: "Test Event", startsAt: "2026-04-01T10:00:00Z", mode: "online" },
+      cid: options.cid ?? "bafy" + rkey,
+      record: options.record ?? {
+        name: "Test Event",
+        startsAt: "2026-04-01T10:00:00Z",
+        mode: "online",
+      },
     },
   };
 }
 
+function budget(overrides: Partial<{
+  maxDrainMs: number;
+  maxCandidates: number;
+  maxSerializedBytes: number;
+}> = {}) {
+  return {
+    maxDrainMs: 5_000,
+    maxCandidates: 100,
+    maxSerializedBytes: 1024 * 1024,
+    ...overrides,
+  };
+}
+
 /** A discoverable-only config — events flow straight into `collected`. */
 function discoverableConfig(): ContrailConfig {
   return {
@@ -127,6 +153,7 @@ describe("ingestEvents — bounded by the safety timeout (om-dua7)", () => {
     });
     // The replayed cursor must come back so the caller can persist it.
     expect(result.lastCursor).toBe(1_000_000);
+    expect(result.stats.stopReason).toBe("drain-time");
   });
 
   it("never checkpoints past the last yielded event when the subscription buffers ahead", async () => {
@@ -152,6 +179,7 @@ describe("ingestEvents — bounded by the safety timeout (om-dua7)", () => {
 
     expect(result.events).toHaveLength(1);
     expect(result.lastCursor).toBe(1_000_000);
+    expect(result.stats.lastAccountedCursor).toBe(1_000_000);
   });
 
   it("returns by the safety timeout even when every arriving event is filtered out", async () => {
@@ -177,6 +205,8 @@ describe("ingestEvents — bounded by the safety timeout (om-dua7)", () => {
       // ...but the cursor still advanced, so the caller persists forward progress.
       expect(result.lastCursor).not.toBeNull();
       expect(result.lastCursor).toBeGreaterThan(1_000_000);
+      expect(result.stats.stopReason).toBe("drain-time");
+      expect(result.stats.lastAccountedCursor).toBe(result.lastCursor);
     } finally {
       jetstream.abort = true;
     }
@@ -237,6 +267,7 @@ describe("ingestEvents — bounded by the safety timeout (om-dua7)", () => {
 
     expect(result.events.map((e) => e.rkey)).toEqual(["hist", "live"]);
     expect(result.lastCursor).toBe(liveUs);
+    expect(result.stats.stopReason).toBe("head");
   });
 
   it("caught-up break fires even on a filtered live event, deferring a following kept event to the next cycle", async () => {
@@ -315,4 +346,244 @@ describe("ingestEvents — bounded by the safety timeout (om-dua7)", () => {
     expect(result.identityUpdates.get("did:plc:author")).toBe("alice.test");
     expect(result.events).toHaveLength(0); // identity events are not record commits
   });
+
+  it("validates every scheduled work threshold", () => {
+    expect(() => resolveScheduledIngestBudget({ maxCandidates: 0 })).toThrow(
+      "maxCandidates must be a positive finite integer",
+    );
+    expect(() =>
+      resolveScheduledIngestBudget({ maxSerializedBytes: Number.NaN }),
+    ).toThrow("maxSerializedBytes must be a positive finite integer");
+    expect(() => resolveScheduledIngestBudget({ maxDrainMs: 1.5 })).toThrow(
+      "maxDrainMs must be a positive finite integer",
+    );
+  });
+
+  it("stops an infinite hot stream at exactly maxCandidates without another next()", async () => {
+    let produced = 0;
+    jetstream.script = async function* (self) {
+      for (let index = 0; !jetstream.abort; index++) {
+        produced++;
+        self.cursor = 1_000_000 + index;
+        yield commitEvent(
+          "did:plc:author",
+          "community.lexicon.calendar.event",
+          self.cursor,
+          `hot-${index}`,
+        );
+      }
+    };
+
+    const result = await ingestEvents(
+      discoverableConfig(),
+      999_999,
+      budget({ maxCandidates: 3 }),
+    );
+
+    expect(result.events).toHaveLength(3);
+    expect(produced).toBe(3);
+    expect(result.stats).toMatchObject({
+      observedSourceItems: 3,
+      retainedCandidates: 3,
+      stopReason: "count",
+      lastAccountedCursor: 1_000_002,
+      safeEndingCursor: 1_000_002,
+    });
+  });
+
+  it("retains the byte-threshold-crossing candidate and does not split equal cursors", async () => {
+    const firstRecord = { value: "a" };
+    const secondRecord = { value: "bbbb" };
+    const encoder = new TextEncoder();
+    const firstBytes =
+      SCHEDULED_INGEST_METADATA_BYTES +
+      encoder.encode(JSON.stringify(firstRecord)).byteLength;
+    const secondBytes =
+      SCHEDULED_INGEST_METADATA_BYTES +
+      encoder.encode(JSON.stringify(secondRecord)).byteLength;
+    const byteThreshold = firstBytes + 1;
+    let produced = 0;
+    jetstream.script = async function* (self) {
+      self.cursor = 2_000_000;
+      produced++;
+      yield commitEvent(
+        "did:plc:author",
+        "community.lexicon.calendar.event",
+        2_000_000,
+        "same-cursor-1",
+        { record: firstRecord },
+      );
+      produced++;
+      yield commitEvent(
+        "did:plc:author",
+        "community.lexicon.calendar.event",
+        2_000_000,
+        "same-cursor-2",
+        { record: secondRecord },
+      );
+      produced++;
+      yield commitEvent(
+        "did:plc:author",
+        "community.lexicon.calendar.event",
+        2_000_001,
+        "not-requested",
+      );
+    };
+
+    const result = await ingestEvents(
+      discoverableConfig(),
+      999_999,
+      budget({ maxSerializedBytes: byteThreshold }),
+    );
+
+    expect(result.events.map((event) => event.rkey)).toEqual([
+      "same-cursor-1",
+      "same-cursor-2",
+    ]);
+    expect(produced).toBe(2);
+    expect(result.stats.stopReason).toBe("bytes");
+    expect(result.stats.serializedCandidateBytes).toBe(firstBytes + secondBytes);
+    expect(result.stats.serializedCandidateBytes - byteThreshold).toBeLessThanOrEqual(
+      secondBytes,
+    );
+    expect(result.lastCursor).toBe(2_000_000);
+  });
+
+  it("drops only exact source observations and retains genuine revisions", async () => {
+    jetstream.script = async function* (self) {
+      self.cursor = 3_000_000;
+      yield commitEvent(
+        "did:plc:author",
+        "community.lexicon.calendar.event",
+        3_000_000,
+        "record",
+        { revision: "rev-1", record: { b: 2, a: 1 } },
+      );
+      // Same source slot and normalized payload, despite object key order.
+      yield commitEvent(
+        "did:plc:author",
+        "community.lexicon.calendar.event",
+        3_000_000,
+        "record",
+        { revision: "rev-1", record: { a: 1, b: 2 } },
+      );
+      self.cursor = 3_000_001;
+      yield commitEvent(
+        "did:plc:author",
+        "community.lexicon.calendar.event",
+        3_000_001,
+        "record",
+        { revision: "rev-2", record: { a: 2, b: 2 } },
+      );
+    };
+
+    const result = await ingestEvents(
+      discoverableConfig(),
+      999_999,
+      budget(),
+    );
+
+    expect(result.events.map((event) => event.source?.revision)).toEqual([
+      "rev-1",
+      "rev-2",
+    ]);
+    expect(result.stats.exactDuplicatesDropped).toBe(1);
+    expect(result.stats.retainedCandidates).toBe(2);
+    expect(result.stats.lastAccountedCursor).toBe(3_000_001);
+  });
+
+  it("advances the accounted cursor through filtered and duplicate observations", async () => {
+    const knownDids = new Set<string>();
+    jetstream.script = async function* (self) {
+      self.cursor = 4_000_000;
+      const original = commitEvent(
+        "did:plc:author",
+        "community.lexicon.calendar.event",
+        4_000_000,
+        "kept",
+      );
+      yield original;
+      yield original;
+      self.cursor = 4_000_001;
+      yield commitEvent(
+        "did:plc:stranger",
+        "app.bsky.graph.follow",
+        4_000_001,
+        "filtered",
+      );
+    };
+
+    const result = await ingestEvents(
+      dependentConfig(),
+      999_999,
+      budget(),
+      knownDids,
+    );
+
+    expect(result.events).toHaveLength(1);
+    expect(result.stats.exactDuplicatesDropped).toBe(1);
+    expect(result.stats.sourceScopeFiltered).toBe(1);
+    expect(result.stats.observedSourceItems).toBe(3);
+    expect(result.lastCursor).toBe(4_000_001);
+  });
+
+  it("never regresses the safe cursor when a pooled transport replays older overlap", async () => {
+    const startingCursor = 6_000_000;
+    jetstream.script = async function* (self) {
+      self.cursor = startingCursor - 500;
+      yield commitEvent(
+        "did:plc:author",
+        "community.lexicon.calendar.event",
+        startingCursor - 500,
+        "rolled-back",
+      );
+    };
+    const config = {
+      ...discoverableConfig(),
+      jetstreams: ["wss://one.test", "wss://two.test"],
+    };
+
+    const result = await ingestEvents(
+      config,
+      startingCursor,
+      budget({ maxCandidates: 1 }),
+    );
+
+    expect(result.events).toHaveLength(1);
+    expect(result.stats.lastAccountedCursor).toBe(startingCursor - 500);
+    expect(result.lastCursor).toBe(startingCursor);
+    expect(result.stats.safeEndingCursor).toBe(startingCursor);
+  });
+
+  it("keeps source-inconsistency diagnostics and normal logs bounded", async () => {
+    const logs: unknown[][] = [];
+    const config = {
+      ...discoverableConfig(),
+      logger: {
+        log: (...args: unknown[]) => logs.push(args),
+        warn: (...args: unknown[]) => logs.push(args),
+        error: (...args: unknown[]) => logs.push(args),
+      },
+    };
+    jetstream.script = async function* (self) {
+      self.cursor = 5_000_000;
+      for (let index = 0; index < 50; index++) {
+        yield commitEvent(
+          "did:plc:author",
+          "community.lexicon.calendar.event",
+          5_000_000,
+          "inconsistent",
+          { revision: "same", record: { value: index } },
+        );
+      }
+    };
+
+    const result = await ingestEvents(config, 999_999, budget());
+
+    expect(result.events).toHaveLength(50);
+    expect(result.stats.sourceInconsistencies).toBe(49);
+    expect(result.stats.diagnosticSamples).toHaveLength(5);
+    expect(result.stats.diagnosticSamplesOmitted).toBe(44);
+    expect(logs).toHaveLength(0);
+  });
 });
diff --git a/packages/contrail/tests/scheduled-ingest.test.ts b/packages/contrail/tests/scheduled-ingest.test.ts
new file mode 100644
index 0000000..12d9391
--- /dev/null
+++ b/packages/contrail/tests/scheduled-ingest.test.ts
@@ -0,0 +1,323 @@
+import { beforeEach, describe, expect, it, vi } from "vitest";
+
+const source = vi.hoisted(() => ({
+  initialCursor: 0,
+  requestedCursors: [] as Array<number | null>,
+  events: [] as Array<{
+    kind: "commit";
+    time_us: number;
+    did: string;
+    commit: {
+      rev: string;
+      collection: string;
+      operation: "create";
+      rkey: string;
+      cid: string;
+      record: unknown;
+    };
+  }>,
+}));
+
+vi.mock("@atcute/jetstream", () => {
+  class MockJetstreamSubscription {
+    cursor: number | null;
+    private readonly start: number | null;
+
+    constructor(options: { cursor?: number }) {
+      this.start = options.cursor ?? source.initialCursor;
+      this.cursor = this.start;
+      source.requestedCursors.push(this.start);
+    }
+
+    async *[Symbol.asyncIterator]() {
+      for (const event of source.events) {
+        // Model the supported source's resume coordinate: only events after the
+        // committed cursor are delivered on the next scheduled connection.
+        if (this.start !== null && event.time_us <= this.start) continue;
+        this.cursor = event.time_us;
+        yield event;
+      }
+    }
+  }
+
+  return { JetstreamSubscription: MockJetstreamSubscription };
+});
+
+import {
+  getLastCursor,
+  resolveConfig,
+  runIngestCycle,
+  type Logger,
+} from "../src/index";
+import { createTestDb } from "./helpers";
+
+const COLLECTION = "com.example.event";
+
+function commit(time_us: number, rkey: string) {
+  return {
+    kind: "commit" as const,
+    time_us,
+    did: "actor",
+    commit: {
+      rev: String(time_us),
+      collection: COLLECTION,
+      operation: "create" as const,
+      rkey,
+      cid: `bafy-${rkey}`,
+      record: { $type: COLLECTION, value: rkey },
+    },
+  };
+}
+
+function config(logger: Logger, dependent = false) {
+  return resolveConfig({
+    namespace: "com.example",
+    profiles: [],
+    constellation: false,
+    collections: dependent
+      ? {
+          event: { collection: COLLECTION },
+          follow: { collection: "app.bsky.graph.follow", discover: false },
+        }
+      : { event: { collection: COLLECTION } },
+    logger,
+  });
+}
+
+function logger() {
+  const lines: Array<{ level: "log" | "warn" | "error"; text: string }> = [];
+  return {
+    lines,
+    value: {
+      log: (...values: unknown[]) =>
+        lines.push({ level: "log" as const, text: values.map(String).join(" ") }),
+      warn: (...values: unknown[]) =>
+        lines.push({ level: "warn" as const, text: values.map(String).join(" ") }),
+      error: (...values: unknown[]) =>
+        lines.push({ level: "error" as const, text: values.map(String).join(" ") }),
+    },
+  };
+}
+
+const budget = {
+  maxDrainMs: 5_000,
+  maxCandidates: 2,
+  maxSerializedBytes: 1024 * 1024,
+};
+
+describe("bounded scheduled ingest cycles", () => {
+  beforeEach(() => {
+    source.initialCursor = 0;
+    source.requestedCursors = [];
+    source.events = [];
+  });
+
+  it("commits a capped cycle cursor and converges after restart", async () => {
+    const db = createTestDb();
+    const output = logger();
+    const configured = config(output.value);
+    const first = commit(1_000_001, "one");
+    source.events = [first, first, commit(1_000_002, "two"), commit(1_000_003, "three")];
+
+    await runIngestCycle(db, configured, budget);
+
+    expect(await getLastCursor(db)).toBe(1_000_002);
+    expect(
+      (await db.prepare(`SELECT rkey FROM records_event ORDER BY time_us`).all<{ rkey: string }>())
+        .results?.map((row) => row.rkey),
+    ).toEqual(["one", "two"]);
+
+    const firstSummary = output.lines.filter((line) =>
+      line.text.startsWith("[ingest] cycle summary "),
+    );
+    expect(firstSummary).toHaveLength(1);
+    expect(firstSummary[0]?.text).toContain('"stop_reason":"count"');
+    expect(firstSummary[0]?.text).toContain('"exact_duplicates_dropped":1');
+    expect(output.lines.some((line) => /candidate:|DUPLICATE/.test(line.text))).toBe(false);
+
+    await runIngestCycle(db, configured, budget);
+
+    expect(await getLastCursor(db)).toBe(1_000_003);
+    expect(
+      (await db.prepare(`SELECT rkey FROM records_event ORDER BY time_us`).all<{ rkey: string }>())
+        .results?.map((row) => row.rkey),
+    ).toEqual(["one", "two", "three"]);
+  });
+
+  it("resumes before a capped coarse cursor without skipping or recounting siblings", async () => {
+    const db = createTestDb();
+    const output = logger();
+    const configured = config(output.value);
+    source.events = [
+      commit(1_100_000, "same-cursor-a"),
+      commit(1_100_000, "same-cursor-b"),
+      commit(1_100_001, "after"),
+    ];
+    const oneCandidate = { ...budget, maxCandidates: 1 };
+
+    await runIngestCycle(db, configured, oneCandidate);
+    expect(await getLastCursor(db)).toBe(1_100_000);
+    expect(
+      (await db.prepare("SELECT rkey FROM records_event ORDER BY rkey").all<{ rkey: string }>())
+        .results?.map((row) => row.rkey),
+    ).toEqual(["same-cursor-a"]);
+
+    await runIngestCycle(db, configured, oneCandidate);
+    expect(await getLastCursor(db)).toBe(1_100_000);
+    expect(
+      (await db.prepare("SELECT rkey FROM records_event ORDER BY rkey").all<{ rkey: string }>())
+        .results?.map((row) => row.rkey),
+    ).toEqual(["same-cursor-a", "same-cursor-b"]);
+
+    await runIngestCycle(db, configured, oneCandidate);
+    expect(await getLastCursor(db)).toBe(1_100_001);
+    expect(
+      (await db.prepare("SELECT rkey FROM records_event ORDER BY rkey").all<{ rkey: string }>())
+        .results?.map((row) => row.rkey),
+    ).toEqual(["after", "same-cursor-a", "same-cursor-b"]);
+    expect(source.requestedCursors).toEqual([0, 1_099_999, 1_099_999]);
+
+    const summaries = output.lines
+      .filter((line) => line.text.startsWith("[ingest] cycle summary "))
+      .map((line) =>
+        JSON.parse(line.text.slice("[ingest] cycle summary ".length)) as {
+          retained_candidates: number;
+          cursor_boundary_duplicates_dropped: number;
+        },
+      );
+    expect(summaries.map((summary) => summary.retained_candidates)).toEqual([
+      1,
+      1,
+      1,
+    ]);
+    expect(summaries[1]?.cursor_boundary_duplicates_dropped).toBe(1);
+    expect(summaries[2]?.cursor_boundary_duplicates_dropped).toBe(2);
+  });
+
+  it("persists Atcute's captured initial cursor after an empty first drain", async () => {
+    const db = createTestDb();
+    const output = logger();
+    source.initialCursor = 7_000_000;
+
+    await runIngestCycle(db, config(output.value), budget);
+
+    expect(await getLastCursor(db)).toBe(7_000_000);
+    const summary = output.lines.find((line) =>
+      line.text.startsWith("[ingest] cycle summary "),
+    );
+    expect(summary?.text).toContain('"starting_cursor":null');
+    expect(summary?.text).toContain('"safe_ending_cursor":7000000');
+
+    // An event sharing the captured microsecond is not skipped by an exclusive
+    // resume API: the next cycle requests one microsecond earlier.
+    source.events = [commit(7_000_000, "between-empty-cycles")];
+    await runIngestCycle(db, config(output.value), budget);
+    expect(source.requestedCursors).toEqual([7_000_000, 6_999_999]);
+    expect(
+      await db
+        .prepare("SELECT rkey FROM records_event WHERE rkey = ?")
+        .bind("between-empty-cycles")
+        .first<{ rkey: string }>(),
+    ).toEqual({ rkey: "between-empty-cycles" });
+  });
+
+  it("rejects endpoint pools for scheduled ingestion before rollback can starve the cap", async () => {
+    const db = createTestDb();
+    const output = logger();
+    const configured = resolveConfig({
+      ...config(output.value),
+      jetstreams: ["wss://one.test", "wss://two.test"],
+    });
+
+    await expect(runIngestCycle(db, configured, budget)).rejects.toThrow(
+      "scheduled ingestion requires exactly one pinned Jetstream endpoint",
+    );
+    expect(source.requestedCursors).toHaveLength(0);
+  });
+
+  it("keeps admission diagnostics bounded as source volume grows", async () => {
+    const db = createTestDb();
+    const output = logger();
+    const configured = config(output.value);
+    source.events = Array.from({ length: 20 }, (_, index) => ({
+      ...commit(1_250_000 + index, `invalid-${index}`),
+      commit: {
+        ...commit(1_250_000 + index, `invalid-${index}`).commit,
+        record: `not-an-object-${index}`,
+      },
+    }));
+
+    await runIngestCycle(db, configured, {
+      ...budget,
+      maxCandidates: 20,
+    });
+
+    expect(output.lines.filter((line) => line.level === "warn")).toHaveLength(0);
+    const summaryLine = output.lines.find((line) =>
+      line.text.startsWith("[ingest] cycle summary "),
+    );
+    const summary = JSON.parse(
+      summaryLine!.text.slice("[ingest] cycle summary ".length),
+    ) as {
+      admission_policy_filtered: number;
+      diagnostic_samples: string[];
+      diagnostic_samples_omitted: number;
+    };
+    expect(summary.admission_policy_filtered).toBe(20);
+    expect(summary.diagnostic_samples).toHaveLength(5);
+    expect(summary.diagnostic_samples_omitted).toBe(15);
+  });
+
+  it("retains and orders two genuine revisions of one URI", async () => {
+    const db = createTestDb();
+    const output = logger();
+    const configured = config(output.value);
+    source.events = [
+      commit(1_500_001, "same"),
+      {
+        ...commit(1_500_002, "same"),
+        commit: {
+          ...commit(1_500_002, "same").commit,
+          cid: "bafy-newer",
+          record: { $type: COLLECTION, value: "newer" },
+        },
+      },
+    ];
+
+    await runIngestCycle(db, configured, budget);
+
+    const row = await db
+      .prepare("SELECT cid, record FROM records_event WHERE rkey = ?")
+      .bind("same")
+      .first<{ cid: string; record: string }>();
+    expect(row?.cid).toBe("bafy-newer");
+    expect(JSON.parse(row!.record)).toMatchObject({ value: "newer" });
+    expect(await getLastCursor(db)).toBe(1_500_002);
+  });
+
+  it("checkpoints a filtered-only accounted range", async () => {
+    const db = createTestDb();
+    const output = logger();
+    const configured = config(output.value, true);
+    source.events = [
+      {
+        ...commit(2_000_001, "filtered"),
+        did: "unknown-actor",
+        commit: {
+          ...commit(2_000_001, "filtered").commit,
+          collection: "app.bsky.graph.follow",
+        },
+      },
+    ];
+
+    await runIngestCycle(db, configured, budget);
+
+    expect(await getLastCursor(db)).toBe(2_000_001);
+    const summary = output.lines.find((line) =>
+      line.text.startsWith("[ingest] cycle summary "),
+    );
+    expect(summary?.text).toContain('"retained_candidates":0');
+    expect(summary?.text).toContain('"source_scope_filtered":1');
+    expect(summary?.text).toContain('"safe_ending_cursor":2000001');
+  });
+});
diff --git a/packages/contrail/tests/schema-fingerprint-gate.test.ts b/packages/contrail/tests/schema-fingerprint-gate.test.ts
index bd31c61..64b768f 100644
--- a/packages/contrail/tests/schema-fingerprint-gate.test.ts
+++ b/packages/contrail/tests/schema-fingerprint-gate.test.ts
@@ -1,6 +1,6 @@
 import { describe, it, expect } from "vitest";
 import { createSqliteDatabase } from "../src/adapters/sqlite";
-import { initSchema, getMeta } from "../src/index";
+import { initSchema, getMeta, setMeta } from "../src/index";
 import { resolveConfig } from "../src/index";
 import type { Database, Statement } from "../src/index";
 
@@ -53,6 +53,24 @@ describe("schema fingerprint gate", () => {
     expect(await getMeta(real, "schema_fingerprint")).toBe(fp); // unchanged
   });
 
+  it("installs cursor observations additively without rebuilding current projections", async () => {
+    const real = createSqliteDatabase(":memory:");
+    await initSchema(real, CONFIG);
+    const current = (await getMeta(real, "schema_fingerprint"))!;
+    const previous = current.replace(/:cursor-observations-v1$/, "");
+    await real.prepare("DROP TABLE cursor_observations").run();
+    await setMeta(real, "schema_fingerprint", previous);
+
+    const upgraded = recordingDb(real);
+    await initSchema(upgraded.db, CONFIG);
+
+    expect(
+      upgraded.prepares.some((sql) => /CREATE TABLE IF NOT EXISTS cursor_observations/.test(sql)),
+    ).toBe(true);
+    expect(upgraded.prepares.some((sql) => /fts_event|records_event/.test(sql))).toBe(false);
+    expect(await getMeta(real, "schema_fingerprint")).toBe(current);
+  });
+
   it("re-applies when the generated schema changes (fingerprint busts)", async () => {
     const real = createSqliteDatabase(":memory:");
     await initSchema(real, CONFIG);
-- 
2.51.2


From 0ec7a003af98e0b079c5d18d0454052d87d41e76 Mon Sep 17 00:00:00 2001
From: Florian <45694132+flo-bit@users.noreply.github.com>
Date: Fri, 28 Aug 2026 13:23:21 +0200
Subject: [PATCH 2/2] fixes

---
 .changeset/bounded-scheduled-ingest.md        |  2 +-
 packages/contrail/README.md                   |  6 +-
 packages/contrail/src/core/bootstrap.ts       | 24 +----
 packages/contrail/src/core/db/index.ts        |  2 +-
 packages/contrail/src/core/db/records.ts      | 24 +++++
 packages/contrail/src/core/identity.ts        | 18 +++-
 packages/contrail/src/core/jetstream.ts       | 61 +++++++++---
 packages/contrail/src/core/persistent.ts      |  7 +-
 packages/contrail/src/worker/index.ts         |  3 +-
 packages/contrail/tests/ingest-hang.test.ts   | 86 +++++++++++++++++
 .../contrail/tests/scheduled-ingest.test.ts   | 96 +++++++++++++++++++
 11 files changed, 281 insertions(+), 48 deletions(-)

diff --git a/.changeset/bounded-scheduled-ingest.md b/.changeset/bounded-scheduled-ingest.md
index 3cb5619..c796f73 100644
--- a/.changeset/bounded-scheduled-ingest.md
+++ b/.changeset/bounded-scheduled-ingest.md
@@ -2,4 +2,4 @@
 "@atmo-dev/contrail": patch
 ---
 
-Bound scheduled Jetstream cycles by retained candidate count and serialized bytes, drop exact transport observations before admission, preserve same-timestamp observations across capped restarts, capture empty initial cursors safely, reject rollback-prone endpoint pools in scheduled mode, and emit one bounded aggregate cycle summary.
+Bound scheduled Jetstream cycles by retained candidate count, distinct identity updates, and serialized bytes; batch identity writes; drop exact transport observations before admission; preserve same-timestamp observations and durable actor scope across capped restarts; capture empty initial cursors safely; reject rollback-prone endpoint pools in scheduled mode; and emit one bounded aggregate cycle summary.
diff --git a/packages/contrail/README.md b/packages/contrail/README.md
index f63608e..1ad89cb 100644
--- a/packages/contrail/README.md
+++ b/packages/contrail/README.md
@@ -61,10 +61,12 @@ HTTP routes expose the same pipeline, including relationship/reference hydration
 ```ts
 await contrail.ingest(); // bounded Jetstream cycle
 
-// Optional per-cycle overrides (defaults: 25s, 250 candidates, 4 MiB).
+// Optional per-cycle overrides (defaults: 25s, 250 candidates,
+// 250 distinct identity updates, and 4 MiB).
 await contrail.ingest({
   maxDrainMs: 20_000,
   maxCandidates: 200,
+  maxIdentityUpdates: 200,
   maxSerializedBytes: 3 * 1024 * 1024,
 });
 
@@ -77,7 +79,7 @@ await contrail.runPersistent({
 
 After a write to a user's PDS, `contrail.notify(uri)` can fetch the authoritative record immediately. Only an authoritative not-found response deletes local state; rate limits, server errors, timeouts, malformed responses, and network failures leave it unchanged. Authentication and abuse controls for the public HTTP operation remain under design.
 
-Contrail stores source event time, repository revision, source cursor, CID, and local index time separately from record/application time. Scheduled collection drops exact Jetstream observations before admission, counts UTF-8 record bodies plus a fixed 512-byte metadata allowance, and retains the threshold-crossing candidate before stopping. Identity and source-filtered observations consume neither candidate nor byte limits, but their yielded cursors remain accounted. Bounded exact-observation hashes at the current microsecond cursor let restarts replay one microsecond of overlap without skipping equal-cursor siblings or recounting earlier ones. Durable tombstones reject stale resurrection, and live Jetstream projection commits its exact accounted cursor in the same transaction. Scheduled mode requires one pinned Jetstream endpoint; use `runPersistent()` for an Atcute failover pool. Persistent ingestion retains its streaming batch lifecycle and does not inherit the scheduled count/byte defaults. A successful PDS `listRecords` page is a current authoritative observation, so it supersedes older durable state without a redundant version read; its version writes and page cursor still commit atomically. Tombstones are retained indefinitely; authoritative rebuild/retention tooling is planned separately.
+Contrail stores source event time, repository revision, source cursor, CID, and local index time separately from record/application time. Scheduled collection drops exact Jetstream observations before admission, counts UTF-8 record bodies plus a fixed 512-byte metadata allowance, and retains the threshold-crossing candidate before stopping. Identity updates have an independent retained-count limit, coalesce by DID, and commit in database batches; excess global identity traffic is accounted but omitted so it cannot starve record progress. Source-filtered observations consume neither candidate nor byte limits, but their yielded cursors remain accounted. Durable actor scope comes from identities, relay discovery, and visible discoverable records, so a failed identity refresh cannot make a restarted isolate filter equal-cursor dependent siblings. Bounded exact-observation hashes at the current microsecond cursor let restarts replay one microsecond of overlap without skipping equal-cursor siblings or recounting earlier ones. Durable tombstones reject stale resurrection, and live Jetstream projection commits its exact accounted cursor in the same transaction. Scheduled mode requires one pinned Jetstream endpoint; use `runPersistent()` for an Atcute failover pool. Persistent ingestion retains its streaming batch lifecycle and does not inherit the scheduled count/byte defaults. A successful PDS `listRecords` page is a current authoritative observation, so it supersedes older durable state without a redundant version read; its version writes and page cursor still commit atomically. Tombstones are retained indefinitely; authoritative rebuild/retention tooling is planned separately.
 
 Projection winner selection is guarded again inside the write transaction. Overlapping cron, persistent, notify, or backfill writers cannot commit a stale canonical row, derived projection, tombstone, or source checkpoint; a changed predecessor rolls the complete attempt back and retries from fresh durable state.
 
diff --git a/packages/contrail/src/core/bootstrap.ts b/packages/contrail/src/core/bootstrap.ts
index e4e5757..f42ff73 100644
--- a/packages/contrail/src/core/bootstrap.ts
+++ b/packages/contrail/src/core/bootstrap.ts
@@ -1,7 +1,8 @@
 import type { ContrailConfig, Database, IngestEvent, Statement } from "./types";
 import { recordTimeUs, createIngestEvent, ingestRecords } from "./ingest";
-import { getDependentNsids, recordsTableName } from "./types";
+import { getDependentNsids } from "./types";
 import {
+  loadKnownActorDids,
   rebuildDerivedProjections,
   saveCursorStatement,
   saveServingSourcePositionStatement,
@@ -511,25 +512,8 @@ export class DatabaseBootstrapTarget implements BootstrapTarget {
     if (getDependentNsids(this.config).length === 0) return undefined;
     if (this.knownDidsLoaded) return this.knownDids ?? new Set();
     const known = this.knownDids ?? new Set<string>();
-    const identityRows = await this.db
-      .prepare("SELECT did FROM identities")
-      .all<{ did: string }>();
-    for (const row of identityRows.results ?? []) known.add(row.did);
-    // Relay discovery defines acquisition scope before concurrent PDS workers
-    // necessarily resolve every identity. Load it directly so a dependent
-    // collection arriving first is not mistaken for an unknown actor.
-    const backfillRows = await this.db
-      .prepare("SELECT DISTINCT did FROM backfills")
-      .all<{ did: string }>();
-    for (const row of backfillRows.results ?? []) known.add(row.did);
-    for (const [shortName, collection] of Object.entries(
-      this.config.collections,
-    )) {
-      if (collection.discover === false) continue;
-      const rows = await this.db
-        .prepare(`SELECT DISTINCT did FROM ${recordsTableName(shortName)}`)
-        .all<{ did: string }>();
-      for (const row of rows.results ?? []) known.add(row.did);
+    for (const did of await loadKnownActorDids(this.db, this.config)) {
+      known.add(did);
     }
     this.knownDids = known;
     this.knownDidsLoaded = true;
diff --git a/packages/contrail/src/core/db/index.ts b/packages/contrail/src/core/db/index.ts
index 770eb3c..78b38a2 100644
--- a/packages/contrail/src/core/db/index.ts
+++ b/packages/contrail/src/core/db/index.ts
@@ -1,6 +1,6 @@
 export { initSchema, CONTRAIL_SCHEMA_VERSION } from "./schema";
 export { getMeta, setMeta, getMetaNumber } from "./meta";
 export { optimizeDatabase } from "./optimize";
-export { assertServingSourceCompatibility, getLastCursor, getServingSourcePosition, orderedSourcePosition, saveCursor, saveCursorStatement, getCursorObservations, saveCursorObservationStatements, saveOrderedSourcePositionStatement, saveServingSourcePositionStatement, lookupExistingRecords, queryRecords, pruneFeedItems, pruneActorFeed, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./records";
+export { assertServingSourceCompatibility, getLastCursor, loadKnownActorDids, getServingSourcePosition, orderedSourcePosition, saveCursor, saveCursorStatement, getCursorObservations, saveCursorObservationStatements, saveOrderedSourcePositionStatement, saveServingSourcePositionStatement, lookupExistingRecords, queryRecords, pruneFeedItems, pruneActorFeed, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor } from "./records";
 export type { QueryOptions, SortOption, ExistingRecordInfo, FeedSweepResult, ServingSourcePosition } from "./records";
 export type { RecordSource } from "../types";
diff --git a/packages/contrail/src/core/db/records.ts b/packages/contrail/src/core/db/records.ts
index db11b3b..1567b9a 100644
--- a/packages/contrail/src/core/db/records.ts
+++ b/packages/contrail/src/core/db/records.ts
@@ -655,6 +655,30 @@ export async function getLastCursor(db: Database): Promise<number | null> {
   return row ? row.time_us : null;
 }
 
+/** Load durable actor-acquisition scope. Identity resolution is best-effort, so
+ * visible discoverable records and relay backfill rows are also authoritative
+ * evidence that an actor is known. This keeps dependent collection filtering
+ * stable across process restarts even when profile resolution failed. */
+export async function loadKnownActorDids(
+  db: Database,
+  config: ContrailConfig,
+): Promise<Set<string>> {
+  const known = new Set<string>();
+  const durableRows = await db
+    .prepare("SELECT did FROM identities UNION SELECT did FROM backfills")
+    .all<{ did: string }>();
+  for (const row of durableRows.results ?? []) known.add(row.did);
+
+  for (const [shortName, collection] of Object.entries(config.collections)) {
+    if (collection.discover === false) continue;
+    const rows = await db
+      .prepare(`SELECT DISTINCT did FROM ${recordsTableName(shortName)}`)
+      .all<{ did: string }>();
+    for (const row of rows.results ?? []) known.add(row.did);
+  }
+  return known;
+}
+
 export function saveCursorStatement(
   db: Database,
   timeUs: number,
diff --git a/packages/contrail/src/core/identity.ts b/packages/contrail/src/core/identity.ts
index a18120d..fbd46e8 100644
--- a/packages/contrail/src/core/identity.ts
+++ b/packages/contrail/src/core/identity.ts
@@ -1,5 +1,5 @@
 import type { Did } from "@atcute/lexicons";
-import type { ContrailConfig, Database, Logger } from "./types";
+import type { ContrailConfig, Database, Logger, Statement } from "./types";
 import { isDid, isHandle } from "@atcute/lexicons/syntax";
 import { resolvePDS } from "./client";
 
@@ -134,15 +134,23 @@ export async function resolveActor(
  * partial rows confuse the rest of the pipeline). PDS column is left
  * untouched; it gets refreshed lazily via `getPDS` / next slingshot resolve.
  */
+export function applyIdentityEventStatement(
+  db: Database,
+  did: string,
+  handle: string,
+  updatedAt = Date.now(),
+): Statement {
+  return db
+    .prepare("UPDATE identities SET handle = ?, resolved_at = ? WHERE did = ?")
+    .bind(handle, updatedAt, did);
+}
+
 export async function applyIdentityEvent(
   db: Database,
   did: string,
   handle: string
 ): Promise<void> {
-  await db
-    .prepare("UPDATE identities SET handle = ?, resolved_at = ? WHERE did = ?")
-    .bind(handle, Date.now(), did)
-    .run();
+  await applyIdentityEventStatement(db, did, handle).run();
 }
 
 export async function refreshStaleIdentities(
diff --git a/packages/contrail/src/core/jetstream.ts b/packages/contrail/src/core/jetstream.ts
index e285c67..d0f3c07 100644
--- a/packages/contrail/src/core/jetstream.ts
+++ b/packages/contrail/src/core/jetstream.ts
@@ -11,7 +11,7 @@ import {
   optimizeIntervalMs,
   optimizeAnalysisLimit,
 } from "./types";
-import { initSchema, getLastCursor, saveCursor, saveCursorStatement, getCursorObservations, saveCursorObservationStatements, saveOrderedSourcePositionStatement, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor, getMetaNumber, setMeta, optimizeDatabase } from "./db";
+import { initSchema, getLastCursor, loadKnownActorDids, saveCursor, saveCursorStatement, getCursorObservations, saveCursorObservationStatements, saveOrderedSourcePositionStatement, sweepFeedItems, getFeedPruneCursor, saveFeedPruneCursor, getMetaNumber, setMeta, optimizeDatabase } from "./db";
 import {
   createIngestEvent,
   ingestRecords,
@@ -19,7 +19,10 @@ import {
   type IngestDropCounts,
   type IngestWarningSamples,
 } from "./ingest";
-import { refreshStaleIdentities, applyIdentityEvent } from "./identity";
+import {
+  refreshStaleIdentities,
+  applyIdentityEventStatement,
+} from "./identity";
 import { backfillFollowersFromConstellation } from "./constellation";
 
 const BATCH_SIZE = 50;
@@ -36,12 +39,14 @@ export const SCHEDULED_INGEST_METADATA_BYTES = 512;
 export const DEFAULT_SCHEDULED_INGEST_BUDGET = Object.freeze({
   maxDrainMs: 25_000,
   maxCandidates: 250,
+  maxIdentityUpdates: 250,
   maxSerializedBytes: 4 * 1024 * 1024,
 }) satisfies ScheduledIngestBudget;
 
 export interface ScheduledIngestBudget {
   maxDrainMs: number;
   maxCandidates: number;
+  maxIdentityUpdates: number;
   maxSerializedBytes: number;
 }
 
@@ -50,6 +55,8 @@ export interface ScheduledIngestOptions {
   maxDrainMs?: number;
   /** Maximum unique commit candidates retained by one drain. Default: 250. */
   maxCandidates?: number;
+  /** Maximum distinct handle updates retained by one drain. Default: 250. */
+  maxIdentityUpdates?: number;
   /** UTF-8 record bytes plus metadata allowances. Default: 4 MiB. */
   maxSerializedBytes?: number;
   /** @deprecated Compatibility alias for maxDrainMs. */
@@ -69,6 +76,8 @@ export interface ScheduledIngestCollectionStats {
   commitObservations: number;
   identityObservations: number;
   retainedCandidates: number;
+  retainedIdentityUpdates: number;
+  identityUpdatesOmitted: number;
   exactDuplicatesDropped: number;
   cursorBoundaryDuplicatesDropped: number;
   resumeOverlapDropped: number;
@@ -118,6 +127,11 @@ export function resolveScheduledIngestBudget(
       options.maxCandidates ?? DEFAULT_SCHEDULED_INGEST_BUDGET.maxCandidates,
       "maxCandidates",
     ),
+    maxIdentityUpdates: positiveInteger(
+      options.maxIdentityUpdates ??
+        DEFAULT_SCHEDULED_INGEST_BUDGET.maxIdentityUpdates,
+      "maxIdentityUpdates",
+    ),
     maxSerializedBytes: positiveInteger(
       options.maxSerializedBytes ??
         DEFAULT_SCHEDULED_INGEST_BUDGET.maxSerializedBytes,
@@ -415,6 +429,8 @@ export async function ingestEvents(
     commitObservations: 0,
     identityObservations: 0,
     retainedCandidates: 0,
+    retainedIdentityUpdates: 0,
+    identityUpdatesOmitted: 0,
     exactDuplicatesDropped: 0,
     cursorBoundaryDuplicatesDropped: 0,
     resumeOverlapDropped: 0,
@@ -603,7 +619,17 @@ export async function ingestEvents(
     } else if (event.kind === "identity") {
       stats.identityObservations++;
       if (await accountSourceItem(event, normalizedJson(event))) return;
-      identityUpdates.set(event.did, event.identity.handle);
+      if (
+        identityUpdates.has(event.did) ||
+        identityUpdates.size < budget.maxIdentityUpdates
+      ) {
+        identityUpdates.set(event.did, event.identity.handle);
+        stats.retainedIdentityUpdates = identityUpdates.size;
+      } else {
+        // Handle updates are best-effort and do not define record projection.
+        // Account overflow so global identity traffic cannot starve commits.
+        stats.identityUpdatesOmitted++;
+      }
     } else {
       await accountSourceItem(event, normalizedJson(event));
     }
@@ -768,10 +794,7 @@ export async function runIngestCycle(
     if (s.cachedKnownDids) {
       knownDids = s.cachedKnownDids;
     } else {
-      const result = await db
-        .prepare("SELECT did FROM identities")
-        .all<{ did: string }>();
-      knownDids = new Set((result.results ?? []).map((r) => r.did));
+      knownDids = await loadKnownActorDids(db, config);
       s.cachedKnownDids = knownDids;
     }
   }
@@ -840,17 +863,26 @@ export async function runIngestCycle(
     }
   }
 
-  // Apply handle changes from #identity events. UPDATE-only, so unknown DIDs
-  // are no-ops. Failures are sampled into the bounded cycle summary.
+  // Apply the independently capped handle changes in bounded database batches.
+  // UPDATE-only statements make unknown DIDs no-ops without spending one D1
+  // round-trip apiece. Failures are sampled into the bounded cycle summary.
   let identityUpdateFailures = 0;
-  for (const [did, handle] of identityUpdates) {
+  const identityEntries = [...identityUpdates];
+  for (let index = 0; index < identityEntries.length; index += BATCH_SIZE) {
+    const chunk = identityEntries.slice(index, index + BATCH_SIZE);
     try {
-      await applyIdentityEvent(db, did, handle);
+      const updatedAt = Date.now();
+      await db.batch(
+        chunk.map(([did, handle]) =>
+          applyIdentityEventStatement(db, did, handle, updatedAt),
+        ),
+      );
+      databaseSubBatchesCommitted++;
     } catch (error) {
-      identityUpdateFailures++;
+      identityUpdateFailures += chunk.length;
       if (warningSamples.samples.length < warningSamples.maxSamples) {
         warningSamples.samples.push(
-          `identity update failed for ${did}: ${String(error)}`.slice(
+          `identity update batch failed (${chunk.length}, first=${chunk[0]?.[0]}): ${String(error)}`.slice(
             0,
             MAX_DIAGNOSTIC_SAMPLE_LENGTH,
           ),
@@ -968,6 +1000,8 @@ export async function runIngestCycle(
     commit_observations: stats.commitObservations,
     identity_observations: stats.identityObservations,
     retained_candidates: stats.retainedCandidates,
+    retained_identity_updates: stats.retainedIdentityUpdates,
+    identity_updates_omitted: stats.identityUpdatesOmitted,
     exact_duplicates_dropped: stats.exactDuplicatesDropped,
     cursor_boundary_duplicates_dropped:
       stats.cursorBoundaryDuplicatesDropped,
@@ -978,6 +1012,7 @@ export async function runIngestCycle(
     source_inconsistencies: stats.sourceInconsistencies,
     serialized_candidate_bytes: stats.serializedCandidateBytes,
     max_candidates: budget.maxCandidates,
+    max_identity_updates: budget.maxIdentityUpdates,
     max_serialized_bytes: budget.maxSerializedBytes,
     max_drain_ms: budget.maxDrainMs,
     starting_cursor: cursor,
diff --git a/packages/contrail/src/core/persistent.ts b/packages/contrail/src/core/persistent.ts
index 877526b..6b19d96 100644
--- a/packages/contrail/src/core/persistent.ts
+++ b/packages/contrail/src/core/persistent.ts
@@ -8,7 +8,7 @@ import {
   jetstreamUrlOption,
   resolveConfig,
 } from "./types";
-import { initSchema, getLastCursor, saveCursorStatement, saveOrderedSourcePositionStatement } from "./db";
+import { initSchema, getLastCursor, loadKnownActorDids, saveCursorStatement, saveOrderedSourcePositionStatement } from "./db";
 import { createIngestEvent, ingestRecords, recordTimeUs } from "./ingest";
 import { refreshStaleIdentities, applyIdentityEvent } from "./identity";
 import { backfillFollowersFromConstellation } from "./constellation";
@@ -66,10 +66,7 @@ export async function runPersistent(
   const dependentCollections: Set<string> = new Set(getDependentNsids(config));
   let knownDids: Set<string> | undefined;
   if (dependentCollections.size > 0) {
-    const result = await db
-      .prepare("SELECT did FROM identities")
-      .all<{ did: string }>();
-    knownDids = new Set((result.results ?? []).map((r) => r.did));
+    knownDids = await loadKnownActorDids(db, config);
     state.cachedKnownDids = knownDids;
     log.log(`Loaded ${knownDids.size} known DIDs from database`);
   }
diff --git a/packages/contrail/src/worker/index.ts b/packages/contrail/src/worker/index.ts
index 8918a7b..8cc81af 100644
--- a/packages/contrail/src/worker/index.ts
+++ b/packages/contrail/src/worker/index.ts
@@ -43,7 +43,8 @@ export interface CreateWorkerOptions<Env extends WorkerEnv = WorkerEnv> {
   lexicons?: object[];
   /** Enable stable discovery and Lexicon routes for anonymous remote clients. */
   publicService?: PublicServiceOptions;
-  /** Count, byte, and drain-time limits for each scheduled Jetstream cycle. */
+  /** Record count, identity count, byte, and drain-time limits for each
+   * scheduled Jetstream cycle. */
   scheduledIngest?: ScheduledIngestOptions;
   /** Bounded pending-account retry slice after each scheduled ingest. Enabled
    *  by default; pass `false` to disable or options to tune its budget. */
diff --git a/packages/contrail/tests/ingest-hang.test.ts b/packages/contrail/tests/ingest-hang.test.ts
index 2c7bebd..1dd0782 100644
--- a/packages/contrail/tests/ingest-hang.test.ts
+++ b/packages/contrail/tests/ingest-hang.test.ts
@@ -70,11 +70,13 @@ function commitEvent(
 function budget(overrides: Partial<{
   maxDrainMs: number;
   maxCandidates: number;
+  maxIdentityUpdates: number;
   maxSerializedBytes: number;
 }> = {}) {
   return {
     maxDrainMs: 5_000,
     maxCandidates: 100,
+    maxIdentityUpdates: 100,
     maxSerializedBytes: 1024 * 1024,
     ...overrides,
   };
@@ -351,6 +353,9 @@ describe("ingestEvents — bounded by the safety timeout (om-dua7)", () => {
     expect(() => resolveScheduledIngestBudget({ maxCandidates: 0 })).toThrow(
       "maxCandidates must be a positive finite integer",
     );
+    expect(() =>
+      resolveScheduledIngestBudget({ maxIdentityUpdates: 0 }),
+    ).toThrow("maxIdentityUpdates must be a positive finite integer");
     expect(() =>
       resolveScheduledIngestBudget({ maxSerializedBytes: Number.NaN }),
     ).toThrow("maxSerializedBytes must be a positive finite integer");
@@ -391,6 +396,87 @@ describe("ingestEvents — bounded by the safety timeout (om-dua7)", () => {
     });
   });
 
+  it("caps distinct identity updates without letting global identity traffic stop record progress", async () => {
+    let produced = 0;
+    jetstream.script = async function* (self) {
+      for (let index = 0; index < 5; index++) {
+        produced++;
+        self.cursor = 1_500_000 + index;
+        yield {
+          kind: "identity" as const,
+          time_us: self.cursor,
+          did: `did:plc:identity-${index}`,
+          identity: {
+            did: `did:plc:identity-${index}`,
+            handle: `identity-${index}.test`,
+            seq: index,
+            time: "2026-04-01T10:00:00Z",
+          },
+        };
+      }
+      self.cursor = 1_500_005;
+      produced++;
+      yield commitEvent(
+        "did:plc:author",
+        "community.lexicon.calendar.event",
+        self.cursor,
+        "after-identities",
+      );
+    };
+
+    const result = await ingestEvents(
+      discoverableConfig(),
+      999_999,
+      budget({ maxIdentityUpdates: 3 }),
+    );
+
+    expect(result.identityUpdates.size).toBe(3);
+    expect(result.events.map((event) => event.rkey)).toEqual([
+      "after-identities",
+    ]);
+    expect(produced).toBe(6);
+    expect(result.stats).toMatchObject({
+      observedSourceItems: 6,
+      identityObservations: 5,
+      retainedIdentityUpdates: 3,
+      identityUpdatesOmitted: 2,
+      stopReason: "idle",
+      lastAccountedCursor: 1_500_005,
+      safeEndingCursor: 1_500_005,
+    });
+  });
+
+  it("coalesces repeated identity updates without spending the distinct-update cap", async () => {
+    jetstream.script = async function* (self) {
+      for (let index = 0; index < 3; index++) {
+        self.cursor = 1_600_000 + index;
+        yield {
+          kind: "identity" as const,
+          time_us: self.cursor,
+          did: "did:plc:identity",
+          identity: {
+            did: "did:plc:identity",
+            handle: `identity-${index}.test`,
+            seq: index,
+            time: "2026-04-01T10:00:00Z",
+          },
+        };
+      }
+    };
+
+    const result = await ingestEvents(
+      discoverableConfig(),
+      999_999,
+      budget({ maxIdentityUpdates: 2 }),
+    );
+
+    expect(result.identityUpdates).toEqual(
+      new Map([["did:plc:identity", "identity-2.test"]]),
+    );
+    expect(result.stats.retainedIdentityUpdates).toBe(1);
+    expect(result.stats.stopReason).toBe("idle");
+  });
+
   it("retains the byte-threshold-crossing candidate and does not split equal cursors", async () => {
     const firstRecord = { value: "a" };
     const secondRecord = { value: "bbbb" };
diff --git a/packages/contrail/tests/scheduled-ingest.test.ts b/packages/contrail/tests/scheduled-ingest.test.ts
index 12d9391..8092753 100644
--- a/packages/contrail/tests/scheduled-ingest.test.ts
+++ b/packages/contrail/tests/scheduled-ingest.test.ts
@@ -45,8 +45,10 @@ vi.mock("@atcute/jetstream", () => {
 
 import {
   getLastCursor,
+  initSchema,
   resolveConfig,
   runIngestCycle,
+  type Database,
   type Logger,
 } from "../src/index";
 import { createTestDb } from "./helpers";
@@ -194,6 +196,40 @@ describe("bounded scheduled ingest cycles", () => {
     expect(summaries[2]?.cursor_boundary_duplicates_dropped).toBe(2);
   });
 
+  it("restores actor scope from a projected record before replaying dependent siblings", async () => {
+    const db = createTestDb();
+    const output = logger();
+    const configured = config(output.value, true);
+    const timeUs = 1_200_000;
+    source.events = [
+      commit(timeUs, "discover-actor"),
+      {
+        ...commit(timeUs, "dependent-sibling"),
+        commit: {
+          ...commit(timeUs, "dependent-sibling").commit,
+          collection: "app.bsky.graph.follow",
+        },
+      },
+    ];
+    const oneCandidate = { ...budget, maxCandidates: 1 };
+
+    await runIngestCycle(db, configured, oneCandidate);
+    expect(
+      await db.prepare("SELECT did FROM identities WHERE did = ?").bind("actor").first(),
+    ).toBeNull();
+    expect(
+      await db.prepare("SELECT rkey FROM records_event").first<{ rkey: string }>(),
+    ).toEqual({ rkey: "discover-actor" });
+
+    // runIngestCycle receives no shared IngestState, modeling a fresh scheduled
+    // isolate. The exact discovery event is dropped at the cursor boundary, so
+    // durable projection scope must still admit its dependent sibling.
+    await runIngestCycle(db, configured, oneCandidate);
+    expect(
+      await db.prepare("SELECT rkey FROM records_follow").first<{ rkey: string }>(),
+    ).toEqual({ rkey: "dependent-sibling" });
+  });
+
   it("persists Atcute's captured initial cursor after an empty first drain", async () => {
     const db = createTestDb();
     const output = logger();
@@ -235,6 +271,66 @@ describe("bounded scheduled ingest cycles", () => {
     expect(source.requestedCursors).toHaveLength(0);
   });
 
+  it("batches the independently capped identity updates", async () => {
+    const real = createTestDb();
+    const output = logger();
+    const configured = config(output.value);
+    await initSchema(real, configured);
+    await real
+      .prepare(
+        "INSERT INTO identities (did, handle, pds, resolved_at) VALUES (?, NULL, NULL, 0), (?, NULL, NULL, 0)",
+      )
+      .bind("actor-a", "actor-b")
+      .run();
+
+    let batchCalls = 0;
+    const db: Database = {
+      ...real,
+      batch(statements) {
+        batchCalls++;
+        return real.batch(statements);
+      },
+    };
+    source.events = [
+      {
+        kind: "identity",
+        time_us: 1_000_001,
+        did: "actor-a",
+        identity: {
+          did: "actor-a",
+          handle: "a.test",
+          seq: 1,
+          time: "2026-04-01T10:00:00Z",
+        },
+      },
+      {
+        kind: "identity",
+        time_us: 1_000_002,
+        did: "actor-b",
+        identity: {
+          did: "actor-b",
+          handle: "b.test",
+          seq: 2,
+          time: "2026-04-01T10:00:00Z",
+        },
+      },
+    ] as unknown as typeof source.events;
+
+    await runIngestCycle(db, configured, {
+      ...budget,
+      maxIdentityUpdates: 2,
+    });
+
+    expect(batchCalls).toBe(2); // one identity batch, then one cursor batch
+    expect(
+      (await real.prepare("SELECT did, handle FROM identities ORDER BY did").all())
+        .results,
+    ).toEqual([
+      { did: "actor-a", handle: "a.test" },
+      { did: "actor-b", handle: "b.test" },
+    ]);
+  });
+
   it("keeps admission diagnostics bounded as source volume grows", async () => {
     const db = createTestDb();
     const output = logger();