diff --git a/app/config.ts b/app/config.ts
index 418d9a9..7912577 100644
--- a/app/config.ts
+++ b/app/config.ts
@@ -2,10 +2,11 @@ import type { ContrailConfig } from "../src/index";
 
 export const config: ContrailConfig = {
   namespace: "rsvp.atmo",
-  spaces: {
-    type: "tools.atmo.event.space",
-    serviceDid: "did:web:rsvp.atmo",
-  },
+  jetstreams: ["wss://jetstream1.us-east.bsky.network"],
+  // spaces: {
+  //   type: "tools.atmo.event.space",
+  //   serviceDid: "did:web:rsvp.atmo",
+  // },
   collections: {
     event: {
       collection: "community.lexicon.calendar.event",
diff --git a/src/core/db/records.ts b/src/core/db/records.ts
index 6bf0a18..c15f53d 100644
--- a/src/core/db/records.ts
+++ b/src/core/db/records.ts
@@ -428,7 +428,12 @@ export async function applyEvents(
     const short = config
       ? shortNameForNsid(config, e.collection) ?? (config.collections[e.collection] ? e.collection : null)
       : e.collection;
-    if (!short) continue; // unknown collection — skip silently
+    if (!short) {
+      (config?.logger ?? console).warn(
+        `[ingest] drop (unknown collection in applyEvents): ${e.operation} ${e.uri} collection=${e.collection}`
+      );
+      continue;
+    }
     const table = recordsTableName(short);
 
     if (e.operation === "delete") {
diff --git a/src/core/jetstream.ts b/src/core/jetstream.ts
index 88d6485..fa74282 100644
--- a/src/core/jetstream.ts
+++ b/src/core/jetstream.ts
@@ -37,33 +37,63 @@ export async function ingestEvents(
   const dependentCollections = new Set(getDependentNsids(config));
   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 subscription = new JetstreamSubscription({
     url: urls,
     wantedCollections: collections,
     ...(cursor !== null ? { cursor } : {}),
     onConnectionOpen() {
-      log.log("Connected to Jetstream");
+      connectCount++;
+      log.log(
+        `[ingest] connected to Jetstream #${connectCount} (url=${urls.join("|")}, cursor=${cursor ?? "none"}, wanted=${collections.join(",")})`
+      );
     },
     onConnectionClose(event) {
       log.log(
-        `Disconnected from Jetstream: ${event.code} ${event.reason}`
+        `[ingest] disconnected from Jetstream: ${event.code} ${event.reason}`
       );
     },
     onConnectionError(event) {
-      log.error("Jetstream error:", event.error);
+      log.error("[ingest] Jetstream error:", event.error);
     },
   });
 
   for await (const event of subscription) {
+    if (firstYieldedTimeUs === null) firstYieldedTimeUs = event.time_us;
+    lastYieldedTimeUs = event.time_us;
     if (event.kind === "commit") {
       const { commit } = event;
+      totalCommits++;
+
+      const uri = `at://${event.did}/${commit.collection}/${commit.rkey}`;
 
       if (dependentCollections.has(commit.collection) && knownDids) {
-        if (!knownDids.has(event.did)) continue;
+        if (!knownDids.has(event.did)) {
+          filteredUnknownDid++;
+          if (filteredDidSamples.size < 10) filteredDidSamples.add(event.did);
+          continue;
+        }
+      }
+
+      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`
+        );
+      } else {
+        seenUris.set(uri, event.time_us);
       }
 
       const now = Date.now();
-      const uri = `at://${event.did}/${commit.collection}/${commit.rkey}`;
 
       collected.push({
         uri,
@@ -80,23 +110,68 @@ export async function ingestEvents(
         indexed_at: now * 1000,
       });
 
+      log.log(
+        `[ingest] keep: ${commit.operation} ${uri} time_us=${event.time_us}`
+      );
+
       if (knownDids && !dependentCollections.has(commit.collection)) {
         knownDids.add(event.did);
       }
     }
 
     if (event.time_us >= startTimeUs) {
-      log.log("Caught up to present, stopping ingestion");
+      log.log(
+        `[ingest] caught up to present, stopping (last time_us=${event.time_us}, startTimeUs=${startTimeUs})`
+      );
       break;
     }
 
     if (Date.now() >= deadline) {
-      log.log("Safety timeout reached, stopping ingestion");
+      log.log(
+        `[ingest] safety timeout reached, stopping (deadline=${deadline}, collected=${collected.length})`
+      );
       break;
     }
   }
 
+  if (filteredUnknownDid > 0) {
+    const sample = [...filteredDidSamples].join(", ");
+    log.log(
+      `[ingest] ${filteredUnknownDid} events filtered (unknown did). sample dids: ${sample}`
+    );
+  }
   const lastCursor = subscription.cursor || null;
+
+  const cursorGap =
+    lastCursor !== null && lastYieldedTimeUs !== null
+      ? lastCursor - 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}, kept=${collected.length}, dupes=${duplicateUris.length}, connects=${connectCount}, first_yielded=${firstYieldedTimeUs ?? "none"}, last_yielded=${lastYieldedTimeUs ?? "none"}, subscription_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 — buffered events may be dropped`
+    );
+  }
+
+  if (connectCount > 1) {
+    log.warn(
+      `[ingest] RECONNECTED ${connectCount} times during cycle — each reconnect picks a URL at random and rolls cursor back 10s`
+    );
+  }
+
   return { events: collected, lastCursor };
 }
 
@@ -117,9 +192,13 @@ 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(
-    `Starting ingestion. Cursor: ${cursor ?? "none"}, Collections: ${collections.join(", ")}`
+    `[ingest] starting cycle. cursor=${cursor ?? "none"}${
+      lagMs !== null ? ` (lag=${lagMs}ms)` : ""
+    }, timeout=${timeoutMs}ms, collections=${collections.join(", ")}`
   );
 
   // Load known DIDs for filtering dependent collections
@@ -147,7 +226,18 @@ export async function runIngestCycle(
     knownDids
   );
 
-  log.log(`Received ${events.length} events from Jetstream`);
+  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`);
+  }
 
   for (let i = 0; i < events.length; i += BATCH_SIZE) {
     const batch = events.slice(i, i + BATCH_SIZE);
@@ -166,7 +256,13 @@ export async function runIngestCycle(
 
   if (lastCursor !== null) {
     await saveCursor(db, lastCursor);
-    log.log(`Saved cursor: ${lastCursor}`);
+    log.log(
+      `[ingest] saved cursor=${lastCursor} (advanced ${
+        cursor !== null ? lastCursor - cursor : "n/a"
+      }us)`
+    );
+  } else {
+    log.log(`[ingest] no cursor returned from subscription; not saving`);
   }
 
   // Prune feed items hourly
@@ -179,5 +275,5 @@ export async function runIngestCycle(
     s.lastFeedPruneMs = Date.now();
   }
 
-  log.log(`Ingestion complete. Stored ${events.length} events.`);
+  log.log(`[ingest] cycle complete. stored=${events.length}`);
 }
diff --git a/src/core/types.ts b/src/core/types.ts
index 219376a..56ee37d 100644
--- a/src/core/types.ts
+++ b/src/core/types.ts
@@ -128,9 +128,6 @@ export function deriveShortName(nsid: string): string {
 
 export const DEFAULT_JETSTREAMS = [
   "wss://jetstream1.us-east.bsky.network",
-  "wss://jetstream2.us-east.bsky.network",
-  "wss://jetstream1.us-west.bsky.network",
-  "wss://jetstream2.us-west.bsky.network",
 ];
 
 export const DEFAULT_RELAYS = [
diff --git a/wrangler.jsonc b/wrangler.jsonc
index 121abe1..2eb97e0 100644
--- a/wrangler.jsonc
+++ b/wrangler.jsonc
@@ -11,7 +11,6 @@
 			"binding": "DB",
 			"database_name": "contrail",
 			"database_id": "29cf6e32-d0b9-4646-ac52-bfbeef085c1a",
-			"remote": false
 		}
 	],
 	"triggers": {