diff --git a/src/browser/webapi/storage/idb/IDBDatabase.zig b/src/browser/webapi/storage/idb/IDBDatabase.zig index c4dba4144..724492808 100644 --- a/src/browser/webapi/storage/idb/IDBDatabase.zig +++ b/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); } diff --git a/src/browser/webapi/storage/idb/IDBFactory.zig b/src/browser/webapi/storage/idb/IDBFactory.zig index d2ba352da..36abc8d73 100644 --- a/src/browser/webapi/storage/idb/IDBFactory.zig +++ b/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; } }; diff --git a/src/browser/webapi/storage/idb/IDBTransaction.zig b/src/browser/webapi/storage/idb/IDBTransaction.zig index f6dc80b85..6a1f2708f 100644 --- a/src/browser/webapi/storage/idb/IDBTransaction.zig +++ b/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; } }