idb: handle reentrant upgradeneeded callbacks

Add handling for upgradeneeded callbacks enqueuing more work.

src/browser/webapi/storage/idb/IDBDatabase.zig

@@ -61,13 +61,17 @@ const CreateObjectStoreOptions = struct {
     autoIncrement: bool = false,
 };
 
-// Only callable during upgradeneeded, hence the _txn check
+// Only callable while the upgrade transaction is live and active, hence the checks
 pub fn createObjectStore(
     self: *IDBDatabase,
     name: []const u8,
     options: ?CreateObjectStoreOptions,
 ) !*IDBObjectStore {
     const txn = self._txn orelse return error.InvalidStateError;
+    if (txn._settled) {
+        return error.InvalidStateError;
+    }
+    try txn.assertActive();
 
     const opts = options orelse CreateObjectStoreOptions{};
     const store_id = self._engine.createObjectStore(
@@ -87,9 +91,13 @@ pub fn createObjectStore(
     return store;
 }
 
-// Only callable during upgradeneeded, hence the _txn check
+// Only callable while the upgrade transaction is live and active, hence the checks
 pub fn deleteObjectStore(self: *IDBDatabase, name: []const u8, _: *Execution) !void {
     const txn = self._txn orelse return error.InvalidStateError;
+    if (txn._settled) {
+        return error.InvalidStateError;
+    }
+    try txn.assertActive();
     try self._engine.deleteObjectStore(self._database_id, name);
     txn.uncacheStore(name);
 }

src/browser/webapi/storage/idb/IDBFactory.zig

@@ -73,10 +73,20 @@ const OpenContext = struct {
     // destruction then. When parked on the gate instead, cancelParked owns it.
     _scheduled: bool = true,
 
+    // If an callback queued more requests, we need to process those requests
+    // on the next tick, and thus need to hold onto the transaction (which pins
+    // that transaction (_rc++) so that it does't get cleaned up from under us).
+    _upgrade: ?*IDBTransaction = null,
+
     fn cancelled(ctx: *anyopaque) void {
         // What if we're gated? Well, A scheduled task is only canceled on
         // teardown, which would have already called Engine.detach(js_ctx).
         const self: *OpenContext = @ptrCast(@alignCast(ctx));
+        if (self._upgrade) |txn| {
+            self.request._txn = .none;
+            txn._db._txn = null;
+            txn.releaseRef(self.exec.page);
+        }
         self.exec._factory.destroy(self);
     }
 
@@ -85,6 +95,14 @@ const OpenContext = struct {
     // in the wake->run window the task finalizer does.
     fn cancelParked(waiter: *Engine.GateWaiter) void {
         const self: *OpenContext = @fieldParentPtr("_gate_waiter", waiter);
+        if (self._upgrade) |txn| {
+            if (txn._begun) {
+                txn._engine.rollback();
+                txn._begun = false;
+            }
+            txn._settled = true;
+            return;
+        }
         if (!self._scheduled) {
             self.exec._factory.destroy(self);
         }
@@ -94,6 +112,10 @@ const OpenContext = struct {
         const self: *OpenContext = @ptrCast(@alignCast(ctx));
         self._scheduled = false;
 
+        if (self._upgrade != null) {
+            return self.drainUpgrade();
+        }
+
         const engine = self.resolveEngine() catch |err| {
             self.exec._factory.destroy(self);
             self.request.setError(err);
@@ -107,17 +129,59 @@ const OpenContext = struct {
         if (!engine.acquireGate(&self._gate_waiter)) {
             return null; // parked; not destroyed
         }
-        defer self.exec._factory.destroy(self);
-        defer _ = engine.releaseGate(&self._gate_waiter);
 
-        self.runOpen(engine) catch |err| {
+        const upgrading = self.runOpen(engine) catch |err| blk: {
             log.warn(.storage, "idb open", .{ .err = err, .name = self.name });
             self.request.setError(err);
             self.request.deliver(self.exec) catch {};
+            break :blk false;
         };
+        if (upgrading) {
+            self._scheduled = true;
+            return 1; // the versionchange drain continues next turn; keep the gate
+        }
+
+        _ = engine.releaseGate(&self._gate_waiter);
+        self.exec._factory.destroy(self);
         return null;
     }
 
+    // One turn of the versionchange drain: deliver a batch of request events;
+    // handlers may enqueue more. Once the transaction settles — the queue
+    // stayed empty (committed, `complete` fired) or a handler aborted —
+    // deliver the open request's outcome and clean up.
+    fn drainUpgrade(self: *OpenContext) !?u32 {
+        const txn = self._upgrade.?;
+        if (txn.settleStep(self.exec)) {
+            self._scheduled = true;
+            return 1;
+        }
+
+        self._upgrade = null;
+        const engine = txn._engine;
+        defer self.exec._factory.destroy(self);
+        defer _ = engine.releaseGate(&self._gate_waiter);
+        try self.finishUpgrade(txn);
+        return null;
+    }
+
+    // The versionchange transaction settled (committed or aborted): sever the
+    // upgrade wiring, drop our pin (may free the transaction), and deliver the
+    // open request's outcome.
+    fn finishUpgrade(self: *OpenContext, txn: *IDBTransaction) !void {
+        const exec = self.exec;
+        const aborted = txn.aborted();
+        self.request._txn = .none;
+        txn._db._txn = null;
+        txn.releaseRef(exec.page);
+
+        if (aborted) {
+            self.request.setError(error.AbortError);
+            return self.request.deliver(exec);
+        }
+        return self.request.fireSuccess(exec);
+    }
+
     // Scheduler wake-up: the connection gate was handed to us, so re-run.
     fn wakeUp(waiter: *Engine.GateWaiter) void {
         const self: *OpenContext = @fieldParentPtr("_gate_waiter", waiter);
@@ -141,7 +205,10 @@ const OpenContext = struct {
         return self.exec.session.idb.engineForOrigin(origin);
     }
 
-    fn runOpen(self: *OpenContext, engine: *Engine) !void {
+    // Returns true when an upgrade drain is now pending: the versionchange
+    // transaction has queued requests, the gate stays held and drainUpgrade
+    // takes over on the next turns.
+    fn runOpen(self: *OpenContext, engine: *Engine) !bool {
         const exec = self.exec;
         const existing = try engine.databaseVersion(self.name);
 
@@ -153,21 +220,22 @@ const OpenContext = struct {
             if (requested < current) {
                 self.request.setError(error.VersionError);
                 self.request.deliver(exec) catch {};
-                return;
+                return false;
             }
 
             if (requested == current) {
                 const database_id = (try engine.databaseId(self.name)).?;
                 const db = try IDBDatabase.init(exec, engine, database_id, self.name, current);
                 self.request.setDatabaseResult(db);
-                return self.request.fireSuccess(exec);
+                try self.request.fireSuccess(exec);
+                return false;
             }
         }
 
         // New database or an upgrade to a higher version. Run a versionchange
         // transaction so user JS can evolve the schema during `upgradeneeded`;
-        // it's exposed as `request.transaction` and committed here once the
-        // handler returns.
+        // it's exposed as `request.transaction` and committed once its request
+        // queue stays empty.
         try engine.begin();
 
         var closed = false;
@@ -181,34 +249,41 @@ const OpenContext = struct {
 
         const txn = try IDBTransaction.initVersionChange(db, exec);
         txn.acquireRef();
-        defer txn.releaseRef(exec.page);
-
-        self.request._txn = .{ .borrowed = txn };
-        // The request is page-scoped and must never outlive this pointer; the
-        // success/abort paths below null it before delivering, this covers the
-        // error paths.
-        defer self.request._txn = .none;
 
         {
+            // The wiring below outlives this call on the drain path; on an
+            // error it must be severed here, with our pin.
+            errdefer {
+                self.request._txn = .none;
+                db._txn = null;
+                txn.releaseRef(exec.page);
+            }
+            self.request._txn = .{ .borrowed = txn };
             db._txn = txn;
-            defer db._txn = null;
             const old_version: u64 = @intCast(existing orelse 0);
             try self.request.fireUpgradeNeeded(exec, old_version, @intCast(requested));
         }
 
-        if (txn.aborted()) {
-            // updateneeded handler called abort() (what a jerk!) — abort() already
-            // rolled back.
+        if (!txn.aborted() and txn._queue.items.len > 0) {
+            // The handler left requests pending (e.g. a keep-alive loop).
+            // Deliver their events one batch per scheduler turn — never
+            // synchronously — so timer tasks can interleave and observe the
+            // transaction as inactive. The drain owns the sqlite txn's fate now.
             closed = true;
-            self.request._txn = .none;
-            self.request.setError(error.AbortError);
-            return self.request.deliver(exec);
+            self._upgrade = txn;
+            return true;
         }
 
-        txn.settle(exec);
+        if (!txn.aborted()) {
+            // Nothing queued: settle synchronously (commit + fire `complete`).
+            txn.settle(exec);
+        }
+        // An aborted transaction — the upgradeneeded handler called abort()
+        // (what a jerk!) — already rolled back; finishUpgrade delivers its
+        // AbortError.
         closed = true;
-        self.request._txn = .none;
-        return self.request.fireSuccess(exec);
+        try self.finishUpgrade(txn);
+        return false;
     }
 };
 

src/browser/webapi/storage/idb/IDBTransaction.zig

@@ -160,6 +160,7 @@ pub fn initVersionChange(db: *IDBDatabase, exec: *Execution) !*IDBTransaction {
         ._mode = .versionchange,
         ._arena = arena,
         ._begun = true,
+        ._active_turn = exec.js.scheduler.generation,
         ._queue = undefined,
         ._gate_waiter = undefined,
     });
@@ -261,23 +262,35 @@ pub fn abort(self: *IDBTransaction, exec: *Execution) !void {
 }
 
 pub fn settle(self: *IDBTransaction, exec: *Execution) void {
+    // Deliver batches until the queue stays empty — a handler may enqueue more.
+    while (self.settleStep(exec)) {}
+}
+
+// One settle turn: deliver the pending batch of request events (handlers may
+// enqueue more) and, once the queue stays empty, commit and fire `complete`.
+// Returns true while more batches remain.
+pub fn settleStep(self: *IDBTransaction, exec: *Execution) bool {
     if (comptime IS_DEBUG) {
         // non versionchange mode goes through the scheduler + drain
         std.debug.assert(self._mode == .versionchange);
     }
 
     if (self._settled) {
-        return;
+        return false;
     }
-    // Deliver batches until the queue stays empty — a handler may enqueue more.
-    while (self._queue.items.len > 0) {
+
+    if (self._queue.items.len > 0) {
         self.deliverBatch(exec);
         if (self._settled) {
-            // a request handler might have settled this (e.g. called abort)
-            return;
+            // a request handler settled this (e.g. called abort) mid-batch
+            return false;
+        }
+        if (self._queue.items.len > 0) {
+            return true;
         }
     }
     self.commitAndComplete(exec);
+    return false;
 }
 
 // Commit the underlying sqlite transaction (if begun), release the connection
@@ -300,13 +313,14 @@ fn commitAndComplete(self: *IDBTransaction, exec: *Execution) void {
 
 // "is this transaction still usable". Once settled or explicitly committing, it
 // no longer accepts new requests; nor does it outside its active turn (a request
-// made from an unrelated task). A versionchange transaction runs synchronously
-// during upgradeneeded and stays active until settled, so it skips the turn check.
+// made from an unrelated task). The turn is stamped at creation (which covers
+// the upgradeneeded dispatch for a versionchange transaction) and again by each
+// delivered batch.
 pub fn assertActive(self: *const IDBTransaction) !void {
     if (self._settled or self._committing) {
         return error.TransactionInactiveError;
     }
-    if (self._mode != .versionchange and self._active_turn != self._exec.js.scheduler.generation) {
+    if (self._active_turn != self._exec.js.scheduler.generation) {
         return error.TransactionInactiveError;
     }
 }