Merge branch 'get-convex:main' into experiment-unverified-auth

Author: graham

.github/workflows/precompile.yml

@@ -29,7 +29,7 @@ jobs:
       - name: Node setup
         uses: actions/setup-node@v4
         with:
-          node-version-file: ".nvmrc.20"
+          node-version-file: ".nvmrc"
 
       - name: Install just
         uses: extractions/setup-just@v2
@@ -120,7 +120,7 @@ jobs:
       - name: Node setup
         uses: actions/setup-node@v4
         with:
-          node-version-file: ".nvmrc.20"
+          node-version-file: ".nvmrc"
 
       - name: NPM install globals
         run: npm ci --prefix scripts

.github/workflows/release_self_hosted_images.yml

@@ -10,9 +10,7 @@ on:
         required: true
   push:
     branches:
-      - main
-    paths:
-      - "crates/model/src/migrations.rs"
+      - release
 
 jobs:
   release_backend:

.github/workflows/test_self_hosted_backend.yml

@@ -56,7 +56,7 @@ jobs:
       - name: Node setup
         uses: actions/setup-node@v4
         with:
-          node-version-file: ".nvmrc.20"
+          node-version-file: ".nvmrc"
 
       - name: NPM install globals
         run: npm ci --prefix scripts

.nvmrc

@@ -1 +1 @@
-18.20.6
+20.19.5

.nvmrc.20

@@ -0,0 +1 @@
+22.19.0

.nvmrc.22

@@ -77,11 +77,11 @@ async_zip = { version = "0.0.17", default-features = false, features = [ "deflat
 async_zip_0_0_9 = { package = "async_zip", version = "0.0.9", default-features = false, features = [ "zstd", "deflate" ] }
 atomic_refcell = "0.1.13"
 aws-config = { version = "1.6", default-features = false, features = [ "client-hyper", "default-https-client", "rustls", "rt-tokio", "sso" ] }
+aws-credential-types = "1"
 aws-lc-rs = { version = "1.13", default-features = false, features = [ "aws-lc-sys", "prebuilt-nasm" ] }
 aws-sdk-s3 = { version = "1.83", default-features = false, features = [ "default-https-client", "rt-tokio", "sigv4a" ] }
 aws-smithy-http = "0.62.0"
 aws-smithy-types-convert = { version = "0.60", features = [ "convert-streams" ] }
-aws-credential-types = "1"
 aws-types = "1"
 axum = { version = "0.8", features = [ "ws", "original-uri", "macros", "multipart" ] }
 axum-extra = { version = "0.10", features = [ "typed-header", "cookie" ] }
@@ -103,7 +103,7 @@ compact_str = "0.9.0"
 const-oid = "0.9.6"
 const_format = { version = "0.2.34", features = [ "fmt" ] }
 criterion = "0.5"
-crossbeam-channel = "0.5"
+crossbeam-channel = "0.5.15"
 csf = "0.1.11"
 cstr = "0.2.11"
 csv-async = "1.2"

Cargo.lock

@@ -146,6 +146,9 @@ pub struct FunctionExecution {
     /// Usage statistics for this instance
     pub usage_stats: AggregatedFunctionUsageStats,
     pub action_memory_used_mb: Option<u64>,
+    /// Size of the returned value in bytes if the function execution was
+    /// successful, excluding HTTP actions.
+    pub return_bytes: Option<u64>,
 
     /// The Convex NPM package version pushed with the module version executed.
     pub udf_server_version: Option<semver::Version>,
@@ -282,6 +285,7 @@ impl FunctionExecution {
                     vector_index_read_bytes: self.usage_stats.vector_index_read_bytes,
                     vector_index_write_bytes: self.usage_stats.vector_index_write_bytes,
                     action_memory_used_mb: self.action_memory_used_mb,
+                    return_bytes: self.return_bytes,
                 },
             },
         }];
@@ -647,6 +651,7 @@ impl<RT: Runtime> FunctionExecutionLog<RT> {
         if outcome.path.is_system() {
             return;
         }
+        let return_bytes = outcome.result.as_ref().ok().map(|v| v.heap_size() as u64);
         let execution = FunctionExecution {
             params: UdfParams::Function {
                 error: match &outcome.result {
@@ -668,6 +673,7 @@ impl<RT: Runtime> FunctionExecutionLog<RT> {
             syscall_trace: outcome.syscall_trace.clone(),
             usage_stats: aggregated,
             action_memory_used_mb: None,
+            return_bytes,
             udf_server_version: outcome.udf_server_version.clone(),
             identity: outcome.identity.clone(),
             context,
@@ -801,6 +807,7 @@ impl<RT: Runtime> FunctionExecutionLog<RT> {
         if outcome.path.udf_path.is_system() {
             return;
         }
+        let return_bytes = outcome.result.as_ref().ok().map(|v| v.heap_size() as u64);
         let execution = FunctionExecution {
             params: UdfParams::Function {
                 error: outcome.result.err(),
@@ -819,6 +826,7 @@ impl<RT: Runtime> FunctionExecutionLog<RT> {
             syscall_trace: outcome.syscall_trace,
             usage_stats: aggregated,
             action_memory_used_mb: None,
+            return_bytes,
             udf_server_version: outcome.udf_server_version,
             identity: outcome.identity,
             context,
@@ -893,6 +901,7 @@ impl<RT: Runtime> FunctionExecutionLog<RT> {
         if outcome.path.udf_path.is_system() {
             return;
         }
+        let return_bytes = outcome.result.as_ref().ok().map(|v| v.heap_size() as u64);
         let execution = FunctionExecution {
             params: UdfParams::Function {
                 error: outcome.result.err(),
@@ -911,6 +920,7 @@ impl<RT: Runtime> FunctionExecutionLog<RT> {
             syscall_trace: outcome.syscall_trace,
             usage_stats: aggregated,
             action_memory_used_mb: Some(completion.memory_in_mb),
+            return_bytes,
             udf_server_version: outcome.udf_server_version,
             identity: outcome.identity,
             context: completion.context,
@@ -1053,6 +1063,7 @@ impl<RT: Runtime> FunctionExecutionLog<RT> {
             environment: ModuleEnvironment::Isolate,
             usage_stats: aggregated,
             action_memory_used_mb: Some(outcome.memory_in_mb()),
+            return_bytes: None,
             syscall_trace: outcome.syscall_trace,
             udf_server_version: outcome.udf_server_version,
             identity: outcome.identity,

Cargo.toml

@@ -122,6 +122,7 @@ impl<RT: Runtime> SystemTableCleanupWorker<RT> {
             self.runtime.clone(),
             Quota::per_second(*SYSTEM_TABLE_ROWS_PER_SECOND),
         );
+        let mut session_requests_delete_cursor = None;
         loop {
             // Jitter the wait between deletion runs to even out load.
             let delay = SYSTEM_TABLE_CLEANUP_FREQUENCY.mul_f32(self.runtime.rng().random());
@@ -140,15 +141,19 @@ impl<RT: Runtime> SystemTableCleanupWorker<RT> {
                 },
                 None => None,
             };
-            self.cleanup_system_table(
-                TableNamespace::Global,
-                &SESSION_REQUESTS_TABLE,
-                session_requests_cutoff
-                    .map_or(CreationTimeInterval::None, CreationTimeInterval::Before),
-                &rate_limiter,
-                *SESSION_CLEANUP_DELETE_CONCURRENCY,
-            )
-            .await?;
+            // Preserve the deletion cursor between runs. This helps skip index tombstones.
+            // Note that we only update the cursor after a successful run.
+            (_, session_requests_delete_cursor) = self
+                .cleanup_system_table(
+                    TableNamespace::Global,
+                    &SESSION_REQUESTS_TABLE,
+                    session_requests_cutoff
+                        .map_or(CreationTimeInterval::None, CreationTimeInterval::Before),
+                    &rate_limiter,
+                    *SESSION_CLEANUP_DELETE_CONCURRENCY,
+                    session_requests_delete_cursor,
+                )
+                .await?;
         }
     }
 
@@ -281,9 +286,9 @@ impl<RT: Runtime> SystemTableCleanupWorker<RT> {
         to_delete: CreationTimeInterval,
         rate_limiter: &RateLimiter<RT>,
         num_deleters: usize,
-    ) -> anyhow::Result<usize> {
+        mut cursor: Option<(CreationTime, ResolvedDocumentId)>,
+    ) -> anyhow::Result<(usize, Option<(CreationTime, ResolvedDocumentId)>)> {
         let _timer = system_table_cleanup_timer();
-        let mut cursor = None;
 
         let (tx, rx) = tokio::sync::mpsc::channel(1);
         let deleter = |chunk: Vec<ResolvedDocumentId>| async {
@@ -320,7 +325,7 @@ impl<RT: Runtime> SystemTableCleanupWorker<RT> {
         };
 
         let ((), deleted) = futures::try_join!(reader, deleters)?;
-        Ok(deleted)
+        Ok((deleted, cursor))
     }
 
     async fn cleanup_system_table_read_chunk(
@@ -521,13 +526,14 @@ mod tests {
         let rate_limiter =
             new_rate_limiter(rt.clone(), Quota::per_second(NonZeroU32::new(10).unwrap()));
 
-        let deleted = worker
+        let (deleted, _cursor) = worker
             .cleanup_system_table(
                 TableNamespace::Global,
                 &SESSION_REQUESTS_TABLE,
                 CreationTimeInterval::Before(cutoff),
                 &rate_limiter,
                 num_deleters,
+                None,
             )
             .await?;
         assert_eq!(deleted, 3);