delete_scheduled_functions_table endpoint (#43075)

emmaling27 · Convex, Inc. · commit dba5fa8548cb · 2025-11-14T19:34:13.000Z
Adds `/delete_scheduled_jobs_table` endpoint for fast O(1) deletion of all scheduled jobs. This helps when the scheduler gets in a loop, scheduling the same function recursively at such a high rate that the cancel all button doesn't work. This is destructive (removes all scheduled job state, unlike writing the scheduled job state as "canceled" in `/cancel_scheduled_jobs`).

GitOrigin-RevId: c1f0e99e57a30081fe5f821d7e35149930800a2e
diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs
@@ -271,7 +271,10 @@ use model::{
         },
         ModuleModel,
     },
-    scheduled_jobs::SchedulerModel,
+    scheduled_jobs::{
+        ScheduledJobsTable,
+        SchedulerModel,
+    },
     session_requests::types::SessionRequestIdentifier,
     snapshot_imports::types::{
         ImportFormat,
@@ -2953,6 +2956,30 @@ impl<RT: Runtime> Application<RT> {
         self.function_log.scheduled_job_lag(window)
     }
 
+    pub async fn delete_scheduled_jobs_table(
+        &self,
+        identity: Identity,
+        component_id: ComponentId,
+    ) -> anyhow::Result<()> {
+        anyhow::ensure!(
+            identity.is_admin() || identity.is_system(),
+            unauthorized_error("delete_scheduled_jobs_table")
+        );
+        let mut tx = self.begin(identity).await?;
+        let mut model = TableModel::new(&mut tx);
+        model
+            .replace_with_empty_table(ScheduledJobsTable, component_id.into())
+            .await?;
+        let component = tx.must_component_path(component_id)?;
+        self.commit_with_audit_log_events(
+            tx,
+            vec![DeploymentAuditLogEvent::DeleteScheduledJobsTable { component }],
+            "delete_scheduled_jobs_table",
+        )
+        .await?;
+        Ok(())
+    }
+
     pub async fn cancel_all_jobs(
         &self,
         component_id: ComponentId,
diff --git a/crates/application/src/tests/scheduled_jobs.rs b/crates/application/src/tests/scheduled_jobs.rs
@@ -401,3 +401,26 @@ async fn test_scheduled_job_retry(
     assert_eq!(state, ScheduledJobState::Success);
     Ok(())
 }
+
+#[convex_macro::test_runtime]
+async fn test_delete_scheduled_jobs_table(rt: TestRuntime) -> anyhow::Result<()> {
+    let application = Application::new_for_tests(&rt).await?;
+    let mut tx = application.begin(Identity::system()).await?;
+    create_scheduled_job(&rt, &mut tx, insert_object_path()).await?;
+    application.commit_test(tx).await?;
+
+    let mut tx = application.begin(Identity::system()).await?;
+    let mut model = SchedulerModel::new(&mut tx, TableNamespace::test_user());
+    let scheduled_jobs = model.list().await?;
+    assert_eq!(scheduled_jobs.len(), 1);
+
+    application
+        .delete_scheduled_jobs_table(Identity::system(), ComponentId::Root)
+        .await?;
+    let mut tx = application.begin(Identity::system()).await?;
+    let mut model = SchedulerModel::new(&mut tx, TableNamespace::test_user());
+    let scheduled_jobs = model.list().await?;
+    assert!(scheduled_jobs.is_empty());
+
+    Ok(())
+}
diff --git a/crates/database/src/bootstrap_model/table.rs b/crates/database/src/bootstrap_model/table.rs
@@ -217,26 +217,28 @@ impl<'a, RT: Runtime> TableModel<'a, RT> {
             .namespace(namespace)
             .id(&table_name)?;
         self.delete_table_by_id_bypassing_schema_enforcement(table_id_and_number.tablet_id)
-            .await
+            .await?;
+        Ok(())
     }
 
     pub async fn delete_hidden_table(&mut self, tablet_id: TabletId) -> anyhow::Result<()> {
         let table_metadata = self.get_table_metadata(tablet_id).await?;
         // We don't need to validate hidden table with the schema.
         anyhow::ensure!(table_metadata.state == TableState::Hidden);
         self.delete_table_by_id_bypassing_schema_enforcement(tablet_id)
-            .await
+            .await?;
+        Ok(())
     }
 
-    pub async fn delete_table(&mut self, tablet_id: TabletId) -> anyhow::Result<()> {
+    pub async fn delete_table(&mut self, tablet_id: TabletId) -> anyhow::Result<TableNumber> {
         self.delete_table_by_id_bypassing_schema_enforcement(tablet_id)
             .await
     }
 
     async fn delete_table_by_id_bypassing_schema_enforcement(
         &mut self,
         tablet_id: TabletId,
-    ) -> anyhow::Result<()> {
+    ) -> anyhow::Result<TableNumber> {
         for index in IndexModel::new(self.tx)
             .all_indexes_on_table(tablet_id)
             .await?
@@ -258,7 +260,7 @@ impl<'a, RT: Runtime> TableModel<'a, RT> {
         SystemMetadataModel::new_global(self.tx)
             .replace(table_doc_id, updated_table_metadata.try_into()?)
             .await?;
-        Ok(())
+        Ok(table_metadata.number)
     }
 
     pub async fn get_table_metadata(
@@ -418,7 +420,7 @@ impl<'a, RT: Runtime> TableModel<'a, RT> {
                 self.delete_table_by_id_bypassing_schema_enforcement(
                     existing_table_by_name.tablet_id,
                 )
-                .await?
+                .await?;
             }
             table_metadatas.push(table_metadata);
         }
@@ -453,6 +455,40 @@ impl<'a, RT: Runtime> TableModel<'a, RT> {
         Ok(())
     }
 
+    pub async fn replace_with_empty_table<S: SystemTable>(
+        &mut self,
+        _system_table: S,
+        namespace: TableNamespace,
+    ) -> anyhow::Result<()> {
+        let tablet_id = self
+            .tx
+            .table_mapping()
+            .namespace(namespace)
+            .name_to_tablet()(S::table_name().clone())?;
+
+        let table_number = self.delete_table(tablet_id).await?;
+        self._insert_table_metadata(
+            namespace,
+            S::table_name(),
+            Some(table_number),
+            TableState::Active,
+        )
+        .await?;
+        let mut index_model = IndexModel::new(self.tx);
+        for index in S::indexes() {
+            let index_metadata = IndexMetadata::new_enabled(
+                index
+                    .name
+                    .map_table(&|_| anyhow::Ok(S::table_name().clone()))?,
+                index.fields,
+            );
+            index_model
+                .add_system_index(namespace, index_metadata)
+                .await?;
+        }
+        Ok(())
+    }
+
     pub async fn insert_table_for_import(
         &mut self,
         namespace: TableNamespace,
diff --git a/crates/local_backend/src/dashboard.rs b/crates/local_backend/src/dashboard.rs
@@ -59,6 +59,10 @@ use crate::{
         export_value,
         UdfResponse,
     },
+    scheduling::{
+        __path_delete_scheduled_functions_table,
+        delete_scheduled_functions_table,
+    },
     schema::IndexMetadataResponse,
     LocalAppState,
 };
@@ -354,4 +358,5 @@ where
         .routes(utoipa_axum::routes!(delete_tables))
         .routes(utoipa_axum::routes!(delete_component))
         .routes(utoipa_axum::routes!(get_source_code))
+        .routes(utoipa_axum::routes!(delete_scheduled_functions_table))
 }
diff --git a/crates/local_backend/src/scheduling.rs b/crates/local_backend/src/scheduling.rs
@@ -11,7 +11,10 @@ use common::{
         ComponentPath,
     },
     http::{
-        extract::Json,
+        extract::{
+            Json,
+            MtState,
+        },
         HttpResponseError,
     },
 };
@@ -26,6 +29,7 @@ use serde::{
     Serialize,
 };
 use sync_types::Timestamp;
+use utoipa::ToSchema;
 use value::TableNamespace;
 
 use crate::{
@@ -140,3 +144,30 @@ pub async fn cancel_job(
 
     Ok(StatusCode::OK)
 }
+
+#[derive(Deserialize, ToSchema)]
+#[serde(rename_all = "camelCase")]
+pub struct DeleteScheduledFunctionsTableRequest {
+    pub component_id: Option<String>,
+}
+
+#[utoipa::path(
+    post,
+    path = "/delete_scheduled_functions_table",
+    request_body = DeleteScheduledFunctionsTableRequest,
+    responses((status = 200))
+)]
+pub async fn delete_scheduled_functions_table(
+    MtState(st): MtState<LocalAppState>,
+    ExtractIdentity(identity): ExtractIdentity,
+    Json(DeleteScheduledFunctionsTableRequest { component_id }): Json<
+        DeleteScheduledFunctionsTableRequest,
+    >,
+) -> Result<impl IntoResponse, HttpResponseError> {
+    must_be_admin_with_write_access(&identity)?;
+    let component_id = ComponentId::deserialize_from_string(component_id.as_deref())?;
+    st.application
+        .delete_scheduled_jobs_table(identity, component_id)
+        .await?;
+    Ok(StatusCode::OK)
+}
diff --git a/crates/model/src/deployment_audit_log/types.rs b/crates/model/src/deployment_audit_log/types.rs
@@ -194,6 +194,9 @@ pub enum DeploymentAuditLogEvent {
         table_names_deleted: BTreeMap<ComponentPath, Vec<TableName>>,
         table_count_deleted: u64,
     },
+    DeleteScheduledJobsTable {
+        component: ComponentPath,
+    },
 }
 
 impl From<IndexDiff> for DeploymentAuditLogEvent {
@@ -247,6 +250,9 @@ impl DeploymentAuditLogEvent {
             DeploymentAuditLogEvent::ChangeDeploymentState { .. } => "change_deployment_state",
             DeploymentAuditLogEvent::SnapshotImport { .. } => "snapshot_import",
             DeploymentAuditLogEvent::ClearTables => "clear_tables",
+            DeploymentAuditLogEvent::DeleteScheduledJobsTable { .. } => {
+                "delete_scheduled_jobs_table"
+            },
         }
     }
 
@@ -375,6 +381,12 @@ impl DeploymentAuditLogEvent {
                 )
             },
             DeploymentAuditLogEvent::ClearTables => obj!(),
+            DeploymentAuditLogEvent::DeleteScheduledJobsTable { component } => {
+                let component = component.serialize();
+                obj!(
+                    "component" => component
+                )
+            },
         }
     }
 
@@ -513,6 +525,12 @@ impl TryFrom<ConvexObject> for DeploymentAuditLogEvent {
                     table_count_deleted: remove_int64(&mut fields, "table_count_deleted")? as u64,
                 }
             },
+            "delete_scheduled_jobs_table" => {
+                let component = ComponentPath::deserialize(
+                    remove_nullable_string(&mut fields, "component")?.as_deref(),
+                )?;
+                DeploymentAuditLogEvent::DeleteScheduledJobsTable { component }
+            },
             _ => anyhow::bail!("action {action} unrecognized"),
         };
         Ok(event)
diff --git a/npm-packages/dashboard/dashboard-deployment-openapi.json b/npm-packages/dashboard/dashboard-deployment-openapi.json
@@ -157,6 +157,26 @@
         }
       }
     },
+    "/delete_scheduled_functions_table": {
+      "post": {
+        "operationId": "delete_scheduled_functions_table",
+        "requestBody": {
+          "content": {
+            "application/json": {
+              "schema": {
+                "$ref": "#/components/schemas/DeleteScheduledFunctionsTableRequest"
+              }
+            }
+          },
+          "required": true
+        },
+        "responses": {
+          "200": {
+            "description": ""
+          }
+        }
+      }
+    },
     "/check_admin_key": {
       "get": {
         "tags": [
@@ -191,6 +211,17 @@
           }
         }
       },
+      "DeleteScheduledFunctionsTableRequest": {
+        "type": "object",
+        "properties": {
+          "componentId": {
+            "type": [
+              "string",
+              "null"
+            ]
+          }
+        }
+      },
       "DeleteTableArgs": {
         "type": "object",
         "required": [
