Streaming export: stop using the legacy cursor format for list_snapshot (#37219)

This fixes a bug where streaming export would fail with an error message (_cannot find table with id 10001_) or never complete when exporting instances with components.

### What was happening?

`list_snapshot` uses a cursor that is essentially the last document that was seen. Its table number is ambiguous without knowing the component of the document. When we added component support for streaming export, we added a new cursor format that contains a tablet ID for disambiguation. However, the part of the code that sends these cursors to the connector was still using the legacy format. This would cause cursors to always be interpreted as being in the root component.

### Deployment
Even though this PR technically introduces an API breaking change (because it drops support for the legacy cursor format), the worst scenario that can happen is the conductor deployment happening between several `list_snapshot` calls. In this case, the user can restart the operation and it will work. The connectors simply treat `cursor` as an arbitrary string and don’t need updates. Also, only new Fivetran/Airbyte exports (or full resyncs of them) call `list_snapshot`.

### Testing
- I tested manually the new behavior by hitting the HTTP endpoints and following the cursors I received.
- I also added a new `application` unit test. This test used to fail when the cursor being passed didn’t contain the tablet ID, which is the previous behavior we had when using the legacy format. The test used to fail with an invalid table number error or infinite loop depending on the order of the tablet IDs, which matches the behavior we had in production repros of this bug.
- This code path is also tested by a smoke test, that I’ve fixed in #37218

GitOrigin-RevId: 0f0c0ea511d5cb07d0a9f4615305f3b07ac2312b

Author: Nicolapps

crates/application/src/lib.rs

@@ -357,7 +357,6 @@ use value::{
     Namespace,
     ResolvedDocumentId,
     TableNamespace,
-    TabletId,
 };
 use vector::{
     PublicVectorSearchQueryResult,
@@ -925,7 +924,7 @@ impl<RT: Runtime> Application<RT> {
         &self,
         identity: Identity,
         snapshot: Option<Timestamp>,
-        cursor: Option<(Option<TabletId>, DeveloperDocumentId)>,
+        cursor: Option<ResolvedDocumentId>,
         table_filter: Option<TableName>,
         component_filter: Option<ComponentPath>,
     ) -> anyhow::Result<SnapshotPage> {

crates/application/src/tests/mod.rs

@@ -17,6 +17,7 @@ mod scheduled_jobs;
 mod schema;
 mod source_package;
 mod storage;
+mod streaming_export;
 
 const NODE_SOURCE: &str = r#"
 var nodeFunction = () => {};

crates/application/src/tests/streaming_export.rs

@@ -0,0 +1,91 @@
+use common::{
+    assert_obj,
+    components::ComponentPath,
+    runtime::Runtime,
+};
+use convex_macro::test_runtime;
+use database::StreamingExportTableFilter;
+use keybroker::Identity;
+use runtime::testing::TestRuntime;
+use value::{
+    TableName,
+    TableNamespace,
+};
+
+use crate::{
+    test_helpers::ApplicationTestExt,
+    Application,
+    BootstrapComponentsModel,
+    UserFacingModel,
+};
+
+pub fn table_name() -> TableName {
+    "table1".parse().unwrap()
+}
+
+#[test_runtime]
+async fn test_streaming_export_from_component(rt: TestRuntime) -> anyhow::Result<()> {
+    let application = Application::new_for_tests(&rt).await?;
+
+    application.load_component_tests_modules("mounted").await?;
+    let component_path = "component".parse()?;
+
+    // Insert documents in the component
+    let mut tx = application.begin(Identity::system()).await?;
+    insert_documents(&mut tx, component_path, &table_name()).await?;
+    application.commit_test(tx).await?;
+
+    // Get all documents through list_snapshot
+    let mut seen_documents = Vec::new();
+    let mut snapshot = None;
+    let mut cursor = None;
+
+    loop {
+        let snapshot_page = application
+            .database
+            .list_snapshot(
+                Identity::system(),
+                snapshot,
+                cursor,
+                StreamingExportTableFilter::default(),
+                5,
+                5,
+            )
+            .await?;
+
+        snapshot = Some(snapshot_page.snapshot);
+        cursor = snapshot_page.cursor;
+
+        for (_, _, _, doc) in snapshot_page.documents {
+            seen_documents.push(doc);
+        }
+
+        if !snapshot_page.has_more {
+            break;
+        }
+    }
+
+    // Verify we got all documents from both components
+    assert_eq!(seen_documents.len(), 10);
+
+    Ok(())
+}
+
+async fn insert_documents<RT: Runtime>(
+    tx: &mut crate::Transaction<RT>,
+    component_path: ComponentPath,
+    table_name: &TableName,
+) -> anyhow::Result<()> {
+    let mut components_model = BootstrapComponentsModel::new(tx);
+    let (_, component_id) = components_model.must_component_path_to_ids(&component_path)?;
+    let table_namespace = TableNamespace::from(component_id);
+    let mut user_facing_model = UserFacingModel::new(tx, table_namespace);
+
+    for i in 0..10 {
+        user_facing_model
+            .insert(table_name.clone(), assert_obj!("index" => i))
+            .await?;
+    }
+
+    Ok(())
+}

crates/database/src/database.rs

@@ -287,7 +287,7 @@ struct ListSnapshotTableIteratorCacheEntry {
     snapshot: Timestamp,
     tablet_id: TabletId,
     by_id: IndexId,
-    resolved_cursor: Option<ResolvedDocumentId>,
+    cursor: Option<ResolvedDocumentId>,
 }
 
 #[derive(Clone)]
@@ -1719,7 +1719,7 @@ impl<RT: Runtime> Database<RT> {
         &self,
         identity: Identity,
         snapshot: Option<Timestamp>,
-        cursor: Option<(Option<TabletId>, DeveloperDocumentId)>,
+        cursor: Option<ResolvedDocumentId>,
         table_filter: StreamingExportTableFilter,
         rows_read_limit: usize,
         rows_returned_limit: usize,
@@ -1752,16 +1752,6 @@ impl<RT: Runtime> Database<RT> {
         let table_mapping = self.snapshot_table_mapping(snapshot).await?;
         let by_id_indexes = self.snapshot_by_id_indexes(snapshot).await?;
         let component_paths = self.snapshot_component_paths(snapshot).await?;
-        let resolved_cursor = cursor
-            .map(|(tablet, developer_id)| match tablet {
-                Some(tablet_id) => Ok(ResolvedDocumentId::new(tablet_id, developer_id)),
-                None => developer_id.to_resolved(
-                    table_mapping
-                        .namespace(TableNamespace::by_component_TODO())
-                        .number_to_tablet(),
-                ),
-            })
-            .transpose()?;
         let tablet_ids: BTreeSet<_> = table_mapping
             .iter()
             .map(|(tablet_id, ..)| tablet_id)
@@ -1771,7 +1761,7 @@ impl<RT: Runtime> Database<RT> {
                     *tablet_id,
                     &table_mapping,
                     &component_paths,
-                ) && resolved_cursor
+                ) && cursor
                     .as_ref()
                     .map(|c| *tablet_id >= c.tablet_id)
                     .unwrap_or(true)
@@ -1798,7 +1788,7 @@ impl<RT: Runtime> Database<RT> {
                 snapshot: *snapshot,
                 tablet_id,
                 by_id,
-                resolved_cursor,
+                cursor,
             };
             if let Some((cache_key, _ds)) = &*cached
                 && *cache_key == expected_cache_key
@@ -1808,7 +1798,7 @@ impl<RT: Runtime> Database<RT> {
             } else {
                 let table_iterator = self.table_iterator(snapshot, 100);
                 table_iterator
-                    .stream_documents_in_table(tablet_id, by_id, resolved_cursor)
+                    .stream_documents_in_table(tablet_id, by_id, cursor)
                     .boxed()
             }
         };
@@ -1854,7 +1844,7 @@ impl<RT: Runtime> Database<RT> {
                 snapshot: *snapshot,
                 tablet_id,
                 by_id,
-                resolved_cursor: Some(new_cursor),
+                cursor: Some(new_cursor),
             };
             *self.list_snapshot_table_iterator_cache.lock() =
                 Some((new_cache_key, document_stream));

crates/database/src/tests/streaming_export_tests.rs

@@ -327,7 +327,7 @@ async fn test_snapshot_list(rt: TestRuntime) -> anyhow::Result<()> {
                         .list_snapshot(
                             Identity::system(),
                             snapshot,
-                            cursor.map(|c| (Some(c.tablet_id), c.developer_id)),
+                            cursor,
                             StreamingExportTableFilter {
                                 table_name: table_filter.clone(),
                                 ..Default::default()