fix: disabling transactions for jobs and event subscribers

integration/testdata/audit_logs/tests.test.ts

@@ -558,7 +558,7 @@ test("job function with identity - audit table populated", async () => {
   expect(weddingUpdateAudit.data.headcount).toEqual(1);
 });
 
-test("job function with error and default rollback - audit table is also rolled back", async () => {
+test("job function with error and no rollback - audit table is not rolled back", async () => {
   const identity = await models.identity.create({
     email: "keelson@keel.xyz",
     issuer: "https://keel.so",
@@ -594,7 +594,7 @@ test("job function with error and default rollback - audit table is also rolled
   >`SELECT * FROM keel_audit where table_name = 'wedding_invitee'`.execute(
     useDatabase()
   );
-  expect(inviteesAudits.rows.length).toEqual(3);
+  expect(inviteesAudits.rows.length).toEqual(4);
 
   const keelsonAudit = inviteesAudits.rows.at(0)!;
   expect(keelsonAudit.id).toHaveLength(27);
@@ -617,12 +617,22 @@ test("job function with error and default rollback - audit table is also rolled
   expect(weavetonAudit.identityId).toBeNull();
   expect(weavetonAudit.data.id).toEqual(prisma.id);
 
+  const keelerDeleteAudit = inviteesAudits.rows.at(3)!;
+  expect(keelerDeleteAudit.id).toHaveLength(27);
+  expect(keelerDeleteAudit.tableName).toEqual("wedding_invitee");
+  expect(keelerDeleteAudit.op).toEqual("delete");
+  expect(keelerDeleteAudit.identityId).toEqual(identity.id);
+  expect(keelerDeleteAudit.data.id).toEqual(keeler.id);
+  expect(keelerDeleteAudit.data.firstName).toEqual(keeler.firstName);
+  expect(keelerDeleteAudit.data.status).toEqual(keeler.status);
+  expect(keelerDeleteAudit.data.isFamily).toEqual(keeler.isFamily);
+
   const weddingAudits = await sql<
     Audit<Wedding>
   >`SELECT * FROM keel_audit where table_name = 'wedding'`.execute(
     useDatabase()
   );
-  expect(weddingAudits.rows.length).toEqual(1);
+  expect(weddingAudits.rows.length).toEqual(2);
 
   const weddingAudit = weddingAudits.rows.at(0)!;
   expect(weddingAudit.id).toHaveLength(27);
@@ -632,6 +642,15 @@ test("job function with error and default rollback - audit table is also rolled
   expect(weddingAudit.data.id).toEqual(wedding.id);
   expect(weddingAudit.data.name).toEqual(wedding.name);
   expect(weddingAudit.data.headcount).toEqual(0);
+
+  const weddingUpdateAudit = weddingAudits.rows.at(1)!;
+  expect(weddingUpdateAudit.id).toHaveLength(27);
+  expect(weddingUpdateAudit.tableName).toEqual("wedding");
+  expect(weddingUpdateAudit.op).toEqual("update");
+  expect(weddingUpdateAudit.identityId).toEqual(identity.id);
+  expect(weddingUpdateAudit.data.id).toEqual(wedding.id);
+  expect(weddingUpdateAudit.data.name).toEqual(wedding.name);
+  expect(weddingUpdateAudit.data.headcount).toEqual(1);
 });
 
 test("job function using kysely with identity - audit table populated", async () => {

integration/testdata/events_basic/tests.test.ts

@@ -72,8 +72,11 @@ test("events from failed job", async () => {
 
   const persons = await models.person.findMany();
 
-  // Due to default rollback
-  expect(persons).toHaveLength(0);
+  expect(persons).toHaveLength(2);
+  expect(persons[0].verifiedEmail).toBeTruthy();
+  expect(persons[1].verifiedEmail).toBeTruthy();
+  expect(persons[0].verifiedUpdate).toBeTruthy();
+  expect(persons[1].verifiedUpdate).toBeTruthy();
 });
 
 test("events from failed custom function with rollback", async () => {

integration/testdata/jobs_permissions/tests.test.ts

@@ -211,7 +211,7 @@ test("job - allowed in job code - permitted", async () => {
   expect(await jobRan(id)).toBeTruthy();
 });
 
-test("job - denied in job code - not permitted with default rollback transaction", async () => {
+test("job - denied in job code - not permitted without rollback transaction", async () => {
   const { id } = await models.trackJob.create({ didJobRun: false });
   const identity = await models.identity.create({
     email: "keel@keel.so",
@@ -222,11 +222,11 @@ test("job - denied in job code - not permitted with default rollback transaction
     jobs.withIdentity(identity).manualJobDeniedInCode({ id, denyIt: true })
   ).toHaveAuthorizationError();
 
-  // This would be true if the transaction didn't roll back.
-  expect(await jobRan(id)).toBeFalsy();
+  // This would be false if a transaction rolled back.
+  expect(await jobRan(id)).toBeTruthy();
 });
 
-test("job - exception - internal error with default rollback transaction", async () => {
+test("job - exception - internal error without rollback transaction", async () => {
   const { id } = await models.trackJob.create({ didJobRun: false });
   const identity = await models.identity.create({
     email: "keel@keel.so",
@@ -240,8 +240,8 @@ test("job - exception - internal error with default rollback transaction", async
     message: "something bad has happened!",
   });
 
-  // This would be true if the transaction didn't roll back.
-  expect(await jobRan(id)).toBeFalsy();
+  // This would be false if a transaction rolled back.
+  expect(await jobRan(id)).toBeTruthy();
 });
 
 test("scheduled job - without identity - permitted", async () => {

integration/testdata/subscribers_basic/tests.test.ts

@@ -33,7 +33,7 @@ test("subscriber - mutating field", async () => {
   expect(updatedMary?.verified).toBeTruthy();
 });
 
-test("subscriber - exception - internal error with default rollback transaction", async () => {
+test("subscriber - exception - internal error without rollback transaction", async () => {
   await models.trackSubscriber.create({ didSubscriberRun: false });
 
   const mary = await models.member.create({
@@ -56,8 +56,8 @@ test("subscriber - exception - internal error with default rollback transaction"
     message: "something bad has happened!",
   });
 
-  // This would be true if the transaction didn't roll back.
-  expect(await subscriberRan()).toBeFalsy();
+  // This would be false if a transaction rolled back.
+  expect(await subscriberRan()).toBeTruthy();
 });
 
 test("subscriber - with env vars - successful", async () => {

packages/functions-runtime/src/database.js

@@ -3,7 +3,6 @@ const neonserverless = require("@neondatabase/serverless");
 const { AsyncLocalStorage } = require("async_hooks");
 const { AuditContextPlugin } = require("./auditing");
 const pg = require("pg");
-const { PROTO_ACTION_TYPES } = require("./consts");
 const { withSpan } = require("./tracing");
 const ws = require("ws");
 

packages/functions-runtime/src/tryExecuteJob.js

@@ -7,7 +7,7 @@ const { PermissionError } = require("./errors");
 // and handle any permissions checks. If a permission check fails, then an Error will be thrown and the catch block will be hit.
 function tryExecuteJob({ db, permitted, request, functionConfig }, cb) {
   return withPermissions(permitted, async ({ getPermissionState }) => {
-    let requiresTransaction = true;
+    let requiresTransaction = false;
     if (functionConfig?.dbTransaction !== undefined) {
       requiresTransaction = functionConfig.dbTransaction;
     }

packages/functions-runtime/src/tryExecuteSubscriber.js

@@ -3,7 +3,7 @@ const { withAuditContext } = require("./auditing");
 
 // tryExecuteSubscriber will create a new database connection and execute the function call.
 function tryExecuteSubscriber({ request, db, functionConfig }, cb) {
-  let requiresTransaction = true;
+  let requiresTransaction = false;
   if (functionConfig?.dbTransaction !== undefined) {
     requiresTransaction = functionConfig.dbTransaction;
   }