mirror/browser
1
0
Fork 0
mirror of https://github.com/lightpanda-io/browser.git synced 2026-08-01 02:06:17 -04:00
Code Issues Packages Projects Releases Wiki Activity

idb: rework transaction model

Browse Source 
Previously, every operation would run synchronously, and resolve the value
on the next drain (schedule task run on the next tick).

With 1 connection per DB, this doesn't work with SQLite: you cannot have nested
transactions. Imagine:

add
  begin
    insert into
    JS callback (success)
      add
        begin <-- nested
          insert

This approach captures the requested operation and does all necessary validation
(since most validation errors are returned synchronously), but only executes
operations on drain. Hence, rather than IDBTransaction._requests being a
queue of result values to emit on drain, it is now a queue of operations to
execute and then emit.

This will also potentially make it easier to address serious memory
issues by better scoping the lifetime of things (particularly requests).
This commit is contained in:
Karl Seguin
2026-07-01 11:41:13 +08:00
parent b0bccbb4bc
commit 1dcd0dc6fb
6 changed files with 362 additions and 117 deletions
Download Patch File Download Diff File Expand all files Collapse all files

62
src/browser/tests/indexeddb.html
View File

@@ -935,4 +935,66 @@
await state.done((count) => testing.expectEqual(0, count));
}
</script>
<!-- Serialization: awaiting a request's success and then starting the next
transaction (the "wrapper await" pattern many IDB libraries use) resumes the
await *during* txN's drain, before it commits. Because operations run inside
each transaction's own drain, txN+1 waits its turn on the single connection
instead of colliding with txN's still-open transaction. -->
<script id="sequential_await_transactions" type=module>
{
const state = await testing.async();
const open = indexedDB.open("seq-db", 1);
open.onupgradeneeded = (e) => e.target.result.createObjectStore("s");
open.onsuccess = async (e) => {
const db = e.target.result;
try {
for (let i = 0; i < 5; i++) {
const tx = db.transaction("s", "readwrite");
const req = tx.objectStore("s").put("v" + i, "k" + i);
// Resumes mid-drain, before tx commits — the old eager `begin`
// would make the next iteration's begin fail here.
await new Promise((resolve, reject) => {
req.onsuccess = resolve;
req.onerror = () => reject(req.error);
});
}
const all = db.transaction("s", "readonly").objectStore("s").getAll();
all.onsuccess = () => state.resolve(all.result);
all.onerror = () => state.resolve({ err: all.error.name });
} catch (err) {
state.resolve({ err: err && err.name });
}
};
await state.done((values) => {
testing.expectEqual(5, values.length);
testing.expectEqual("v0", values[0]);
testing.expectEqual("v4", values[4]);
});
}
</script>
<!-- Two readwrite transactions opened before either drains. They serialize on the
shared connection (one drain fully commits before the next begins) rather than
the second's begin colliding with the first. -->
<script id="overlapping_transactions" type=module>
{
const state = await testing.async();
const open = indexedDB.open("overlap-db", 1);
open.onupgradeneeded = (e) => e.target.result.createObjectStore("s");
open.onsuccess = (e) => {
const db = e.target.result;
const txa = db.transaction("s", "readwrite");
txa.objectStore("s").put("a", "ka");
const txb = db.transaction("s", "readwrite");
txb.objectStore("s").put("b", "kb");
txb.oncomplete = () => {
const c = db.transaction("s", "readonly").objectStore("s").count();
c.onsuccess = () => state.resolve(c.result);
};
txb.onerror = () => state.resolve({ err: txb.error.name });
};
await state.done((count) => testing.expectEqual(2, count));
}
</script>
</body>

66
src/browser/webapi/storage/idb/IDBCursor.zig
View File

@@ -84,7 +84,7 @@ const Source = union(enum) {
};
// How the next iterate() positions the cursor.
const Seek = union(enum) {
pub const Seek = union(enum) {
first, // start of the range, in the iteration direction
next, // strictly past the current position
to: []const u8, // continue(key): first record at/after an index/store key
@@ -109,7 +109,7 @@ pub fn openIndex(index: *IDBIndex, bounds: Engine.Bounds, direction: Direction,
}
fn create(store: *IDBObjectStore, txn: *IDBTransaction, index_id: ?i64, source: Source, bounds: Engine.Bounds, direction: Direction, key_only: bool, exec: *Execution) !*IDBRequest {
try txn.ensureBegun();
try txn.assertActive();
const request = try txn.newRequest();
const self = try exec._factory.create(IDBCursor{
@@ -136,8 +136,24 @@ fn create(store: *IDBObjectStore, txn: *IDBTransaction, index_id: ?i64, source:
// avoiding the bridge's pointer -> js.Value lookup each time.
self._js = try public.persist();
try self.iterate(.first, 0, exec);
return request;
// The first seek runs in the drain (or immediately, for a versionchange txn),
// not here — so the connection is only touched while the txn owns it.
return request.submit(.{ .cursor_iterate = .{ .cursor = self, .seek = .first, .offset = 0 } }, exec);
}
// Re-arm the cursor's existing request with its next seek and requeue it. Called by
// continue/advance from within a success handler; the drain loop picks the requeued
// request up in the same pass (a versionchange cursor runs it now).
fn reiterate(self: *IDBCursor, seek: Seek, offset: u32, exec: *Execution) !void {
_ = try self._request.submit(.{ .cursor_iterate = .{ .cursor = self, .seek = seek, .offset = offset } }, exec);
}
// Run the deferred seek, staging the positioned cursor (or null) on the request.
pub fn runIterate(self: *IDBCursor, seek: Seek, offset: u32, exec: *Execution) !void {
self.iterate(seek, offset, exec) catch |err| {
log.warn(.storage, "idb cursor iterate", .{ .err = err });
self._request.setError(err);
};
}
// Called by IDBRequest.deliver right before a cursor's success handler runs.
@@ -155,11 +171,10 @@ pub fn @"continue"(self: *IDBCursor, key_arg: ?js.Value, exec: *Execution) !void
if (if (self._direction.reverse()) order != .lt else order != .gt) {
return error.DataError;
}
try self.iterate(.{ .to = encoded }, 0, exec);
try self.reiterate(.{ .to = encoded }, 0, exec);
} else {
try self.iterate(.next, 0, exec);
try self.reiterate(.next, 0, exec);
}
try self._txn.enqueue(self._request);
}
pub fn continuePrimaryKey(self: *IDBCursor, key_arg: js.Value, primary_key_arg: js.Value, exec: *Execution) !void {
@@ -184,8 +199,7 @@ pub fn continuePrimaryKey(self: *IDBCursor, key_arg: js.Value, primary_key_arg:
};
if (!ok) return error.DataError;
try self.iterate(.{ .to_primary = .{ .key = key, .primary_key = primary_key } }, 0, exec);
try self._txn.enqueue(self._request);
try self.reiterate(.{ .to_primary = .{ .key = key, .primary_key = primary_key } }, 0, exec);
}
pub fn advance(self: *IDBCursor, count: u32, exec: *Execution) !void {
@@ -195,8 +209,7 @@ pub fn advance(self: *IDBCursor, count: u32, exec: *Execution) !void {
try self.prepareIterate();
// `count` records forward = skip count-1 past the immediate next.
try self.iterate(.next, count - 1, exec);
try self._txn.enqueue(self._request);
try self.reiterate(.next, count - 1, exec);
}
pub fn update(self: *IDBCursor, value: js.Value, exec: *Execution) !*IDBRequest {
@@ -224,20 +237,32 @@ pub fn update(self: *IDBCursor, value: js.Value, exec: *Execution) !*IDBRequest
}
}
const serialized = try value.serialize();
// Snapshot the record key: the write runs in the drain, by which point a
// `continue` could have moved the cursor's live position on.
const request = try self._txn.newRequest();
const value_global = try value.persist();
return request.submit(.{ .cursor_update = .{ .cursor = self, .key = key, .value = value_global } }, exec);
}
pub fn runUpdate(self: *IDBCursor, request: *IDBRequest, key: []const u8, value_global: js.Value.Global, exec: *Execution) !void {
const local = exec.js.local.?;
const value = value_global.local(local);
const serialized = value.serialize() catch |err| {
request.setError(err);
return;
};
defer serialized.deinit();
const request = try self._txn.newRequest();
self._store.writeAt(key, value, serialized.bytes(), exec) catch |err| {
log.warn(.storage, "idb cursor update", .{ .err = err });
request.setError(err);
return request;
return;
};
try request.setValue(try Key.decodeToJs(exec.call_arena, exec.js.local.?, key));
return request;
try request.setValue(try Key.decodeToJs(exec.call_arena, local, key));
}
pub fn delete(self: *IDBCursor) !*IDBRequest {
pub fn delete(self: *IDBCursor, exec: *Execution) !*IDBRequest {
if (self._txn._mode == .readonly) {
return error.ReadOnlyError;
}
@@ -250,14 +275,17 @@ pub fn delete(self: *IDBCursor) !*IDBRequest {
return error.InvalidStateError;
}
// Snapshot the record key (see update): the delete runs later, in the drain.
const key = self._primary_key orelse return error.InvalidStateError;
const request = try self._txn.newRequest();
return request.submit(.{ .cursor_delete = .{ .cursor = self, .key = key } }, exec);
}
pub fn runDelete(self: *IDBCursor, request: *IDBRequest, key: []const u8) !void {
self._store.deleteAt(key) catch |err| {
log.warn(.storage, "idb cursor delete", .{ .err = err });
request.setError(err);
};
return request;
}
pub fn getKey(self: *const IDBCursor, exec: *Execution) !?js.Value {

43
src/browser/webapi/storage/idb/IDBIndex.zig
View File

@@ -56,41 +56,44 @@ pub fn init(obj_store: *IDBObjectStore, info: Engine.IndexInfo, name: []const u8
fn txn(self: *IDBIndex) !*IDBTransaction {
const t = self._store._txn orelse return error.TransactionInactiveError;
try t.ensureBegun();
try t.assertActive();
return t;
}
pub fn get(self: *IDBIndex, query: js.Value, exec: *Execution) !*IDBRequest {
const t = try self.txn();
const arena = exec.call_arena;
const bounds = try IDBKeyRange.resolveQuery(arena, query, exec);
const bounds = try IDBKeyRange.resolveQuery(exec.arena, query, exec);
const request = try t.newRequest();
return request.submit(.{ .index_get = .{ .index = self, .bounds = bounds } }, exec);
}
pub fn runGet(self: *IDBIndex, request: *IDBRequest, bounds: Engine.Bounds, exec: *Execution) !void {
const arena = exec.call_arena;
const bytes = self._engine.indexGetRange(arena, self._store._store_id, self._index_id, bounds) catch |err| {
log.warn(.storage, "idb index get", .{ .err = err });
request.setError(err);
return request;
return;
};
const b = bytes orelse return request;
const b = bytes orelse return;
try request.setValue(try js.Value.deserialize(exec.js.local.?, b));
return request;
}
pub fn getKey(self: *IDBIndex, query: js.Value, exec: *Execution) !*IDBRequest {
const t = try self.txn();
const arena = exec.call_arena;
const bounds = try IDBKeyRange.resolveQuery(arena, query, exec);
const bounds = try IDBKeyRange.resolveQuery(exec.arena, query, exec);
const request = try t.newRequest();
return request.submit(.{ .index_get_key = .{ .index = self, .bounds = bounds } }, exec);
}
pub fn runGetKey(self: *IDBIndex, request: *IDBRequest, bounds: Engine.Bounds, exec: *Execution) !void {
const arena = exec.call_arena;
const bytes = self._engine.indexGetKeyRange(arena, self._index_id, bounds) catch |err| {
log.warn(.storage, "idb index getKey", .{ .err = err });
request.setError(err);
return request;
return;
};
const b = bytes orelse return request;
const b = bytes orelse return;
try request.setValue(try Key.decodeToJs(arena, exec.js.local.?, b));
return request;
}
pub fn getAll(self: *IDBIndex, query: ?js.Value, count_: ?u32, exec: *Execution) !*IDBRequest {
@@ -103,16 +106,18 @@ pub fn getAllKeys(self: *IDBIndex, query: ?js.Value, count_: ?u32, exec: *Execut
fn _getAll(self: *IDBIndex, query: ?js.Value, count_: ?u32, column: Engine.Column, exec: *Execution) !*IDBRequest {
const t = try self.txn();
const bounds = try IDBKeyRange.resolveQuery(exec.call_arena, query, exec);
const bounds = try IDBKeyRange.resolveQuery(exec.arena, query, exec);
const request = try t.newRequest();
return request.submit(.{ .index_get_all = .{ .index = self, .bounds = bounds, .column = column, .count = count_ } }, exec);
}
pub fn runGetAll(self: *IDBIndex, request: *IDBRequest, bounds: Engine.Bounds, column: Engine.Column, count_: ?u32, exec: *Execution) !void {
const arr = self.collectAll(bounds, count_, column, exec) catch |err| {
log.warn(.storage, "idb index getAll", .{ .err = err });
request.setError(err);
return request;
return;
};
try request.setValue(arr);
return request;
}
// Stream an index getAll/getAllKeys straight into a JS array: .value rows are
@@ -137,16 +142,18 @@ fn collectAll(self: *IDBIndex, bounds: Engine.Bounds, count_: ?u32, column: Engi
pub fn count(self: *IDBIndex, query: ?js.Value, exec: *Execution) !*IDBRequest {
const t = try self.txn();
const bounds = try IDBKeyRange.resolveQuery(exec.call_arena, query, exec);
const bounds = try IDBKeyRange.resolveQuery(exec.arena, query, exec);
const request = try t.newRequest();
return request.submit(.{ .index_count = .{ .index = self, .bounds = bounds } }, exec);
}
pub fn runCount(self: *IDBIndex, request: *IDBRequest, bounds: Engine.Bounds, exec: *Execution) !void {
const n = self._engine.indexCountRange(self._index_id, bounds) catch |err| {
log.warn(.storage, "idb index count", .{ .err = err });
request.setError(err);
return request;
return;
};
try request.setValue(try exec.js.local.?.zigValueToJs(n, .{}));
return request;
}
pub fn openCursor(self: *IDBIndex, query: ?js.Value, direction: ?IDBCursor.Direction, exec: *Execution) !*IDBRequest {

192
src/browser/webapi/storage/idb/IDBObjectStore.zig
View File

@@ -73,25 +73,25 @@ pub fn put(self: *IDBObjectStore, value: js.Value, key: ?js.Value, exec: *Execut
pub fn get(self: *IDBObjectStore, query: js.Value, exec: *Execution) !*IDBRequest {
const txn = self._txn orelse return error.TransactionInactiveError;
try txn.ensureBegun();
// Both the encoded key and the fetched bytes are consumed within this call
// (the bytes are deserialized below), so the per-call scratch arena suffices.
const arena = exec.call_arena;
const bounds = try IDBKeyRange.resolveQuery(arena, query, exec);
try txn.assertActive();
// Bounds are captured for deferred execution, so they must outlive this call —
// the page arena, not the per-call scratch arena.
const bounds = try IDBKeyRange.resolveQuery(exec.arena, query, exec);
const request = try txn.newRequest();
return request.submit(.{ .store_get = .{ .store = self, .bounds = bounds } }, exec);
}
pub fn runGet(self: *IDBObjectStore, request: *IDBRequest, bounds: Engine.Bounds, exec: *Execution) !void {
// The fetched bytes are deserialized here and then discarded, so the per-call
// scratch arena suffices for them.
const arena = exec.call_arena;
const bytes = self._engine.getRange(arena, self._store_id, bounds) catch |err| {
log.warn(.storage, "idb get", .{ .err = err });
request.setError(err);
return request;
return;
};
const b = bytes orelse return request;
const value = try js.Value.deserialize(exec.js.local.?, b);
try request.setValue(value);
return request;
const b = bytes orelse return; // no record -> undefined
try request.setValue(try js.Value.deserialize(exec.js.local.?, b));
}
pub fn delete(self: *IDBObjectStore, query: js.Value, exec: *Execution) !*IDBRequest {
@@ -99,16 +99,17 @@ pub fn delete(self: *IDBObjectStore, query: js.Value, exec: *Execution) !*IDBReq
if (txn._mode == .readonly) {
return error.ReadOnlyError;
}
try txn.ensureBegun();
const bounds = try IDBKeyRange.resolveQuery(exec.call_arena, query, exec);
try txn.assertActive();
const bounds = try IDBKeyRange.resolveQuery(exec.arena, query, exec);
const request = try txn.newRequest();
return request.submit(.{ .store_delete = .{ .store = self, .bounds = bounds } }, exec);
}
pub fn runDelete(self: *IDBObjectStore, request: *IDBRequest, bounds: Engine.Bounds, _: *Execution) !void {
self.deleteBounds(bounds) catch |err| {
log.warn(.storage, "idb delete", .{ .err = err });
request.setError(err);
};
return request;
}
fn deleteBounds(self: *IDBObjectStore, bounds: Engine.Bounds) !void {
@@ -116,19 +117,21 @@ fn deleteBounds(self: *IDBObjectStore, bounds: Engine.Bounds) !void {
try self._engine.deleteRange(self._store_id, bounds);
}
pub fn clear(self: *IDBObjectStore, _: *Execution) !*IDBRequest {
pub fn clear(self: *IDBObjectStore, exec: *Execution) !*IDBRequest {
const txn = self._txn orelse return error.TransactionInactiveError;
if (txn._mode == .readonly) {
return error.ReadOnlyError;
}
try txn.ensureBegun();
try txn.assertActive();
const request = try txn.newRequest();
return request.submit(.{ .store_clear = self }, exec);
}
pub fn runClear(self: *IDBObjectStore, request: *IDBRequest, _: *Execution) !void {
self.clearAll() catch |err| {
log.warn(.storage, "idb clear", .{ .err = err });
request.setError(err);
};
return request;
}
fn clearAll(self: *IDBObjectStore) !void {
@@ -138,17 +141,19 @@ fn clearAll(self: *IDBObjectStore) !void {
pub fn count(self: *IDBObjectStore, query: ?js.Value, exec: *Execution) !*IDBRequest {
const txn = self._txn orelse return error.TransactionInactiveError;
try txn.ensureBegun();
const bounds = try IDBKeyRange.resolveQuery(exec.call_arena, query, exec);
try txn.assertActive();
const bounds = try IDBKeyRange.resolveQuery(exec.arena, query, exec);
const request = try txn.newRequest();
return request.submit(.{ .store_count = .{ .store = self, .bounds = bounds } }, exec);
}
pub fn runCount(self: *IDBObjectStore, request: *IDBRequest, bounds: Engine.Bounds, exec: *Execution) !void {
const n = self._engine.countRange(self._store_id, bounds) catch |err| {
log.warn(.storage, "idb count", .{ .err = err });
request.setError(err);
return request;
return;
};
try request.setValue(try exec.js.local.?.zigValueToJs(n, .{}));
return request;
}
pub fn getAll(self: *IDBObjectStore, query: ?js.Value, count_: ?u32, exec: *Execution) !*IDBRequest {
@@ -157,18 +162,19 @@ pub fn getAll(self: *IDBObjectStore, query: ?js.Value, count_: ?u32, exec: *Exec
fn _getAll(self: *IDBObjectStore, query: ?js.Value, count_: ?u32, column: Engine.Column, exec: *Execution) !*IDBRequest {
const txn = self._txn orelse return error.TransactionInactiveError;
try txn.ensureBegun();
const bounds = try IDBKeyRange.resolveQuery(exec.call_arena, query, exec);
try txn.assertActive();
const bounds = try IDBKeyRange.resolveQuery(exec.arena, query, exec);
const request = try txn.newRequest();
return request.submit(.{ .store_get_all = .{ .store = self, .bounds = bounds, .column = column, .count = count_ } }, exec);
}
pub fn runGetAll(self: *IDBObjectStore, request: *IDBRequest, bounds: Engine.Bounds, column: Engine.Column, count_: ?u32, exec: *Execution) !void {
const arr = self.collectAll(exec, bounds, column, count_) catch |err| {
log.warn(.storage, "idb getAll", .{ .err = err });
request.setError(err);
return request;
return;
};
try request.setValue(arr);
return request;
}
// Stream a getAll/getAllKeys result straight into a JS array: .value rows
@@ -193,21 +199,21 @@ fn collectAll(self: *IDBObjectStore, exec: *Execution, bounds: Engine.Bounds, co
pub fn getKey(self: *IDBObjectStore, query: js.Value, exec: *Execution) !*IDBRequest {
const txn = self._txn orelse return error.TransactionInactiveError;
try txn.ensureBegun();
const arena = exec.call_arena;
const bounds = try IDBKeyRange.resolveQuery(arena, query, exec);
try txn.assertActive();
const bounds = try IDBKeyRange.resolveQuery(exec.arena, query, exec);
const request = try txn.newRequest();
return request.submit(.{ .store_get_key = .{ .store = self, .bounds = bounds } }, exec);
}
pub fn runGetKey(self: *IDBObjectStore, request: *IDBRequest, bounds: Engine.Bounds, exec: *Execution) !void {
const arena = exec.call_arena;
const found = self._engine.getKeyRange(arena, self._store_id, bounds) catch |err| {
log.warn(.storage, "idb getKey", .{ .err = err });
request.setError(err);
return request;
return;
};
const bytes = found orelse return request; // no record -> undefined
const bytes = found orelse return; // no record -> undefined
try request.setValue(try Key.decodeToJs(arena, exec.js.local.?, bytes));
return request;
}
pub fn getAllKeys(self: *IDBObjectStore, query: ?js.Value, count_: ?u32, exec: *Execution) !*IDBRequest {
@@ -240,86 +246,122 @@ pub fn getTransaction(self: *IDBObjectStore) ?*IDBTransaction {
return self._txn;
}
const WriteKind = enum { add, put };
pub const WriteKind = enum { add, put };
// The key that we're writing to. We need to validate this on the request side
// so we might as well store the result for the execution side.
pub const PreparedKey = union(enum) {
// generate number key + injsect into value
generate_in_line,
// generate number key, don't inject
generate_out_of_line,
// an explicit key, could be passed in or extracted from the value
explicit: struct { encoded: []const u8, bump: ?f64 },
};
fn write(self: *IDBObjectStore, value: js.Value, key_arg: ?js.Value, kind: WriteKind, exec: *Execution) !*IDBRequest {
const txn = self._txn orelse return error.TransactionInactiveError;
if (txn._mode == .readonly) {
return error.ReadOnlyError;
}
try txn.ensureBegun();
try txn.assertActive();
const local = exec.js.local.?;
var generated = false;
const key_value: js.Value = blk: {
// Resolve (and validate) the key now, so DataError / a throwing key getter
// throws synchronously. Encoded key bytes are captured on the page arena to
// outlive this call. The record write itself is deferred to runWrite.
const prepared: PreparedKey = blk: {
if (self._key_path) |kp| {
if (key_arg != null) {
// can't have an explicit key if we're configured for in-line keys
return error.DataError;
}
if (Key.evaluatePath(value, kp)) |extracted| {
break :blk extracted;
break :blk .{ .explicit = .{
.encoded = try Key.encodeValue(exec.arena, extracted),
.bump = if (self._auto_increment and extracted.isNumber()) try extracted.toF64() else null,
} };
}
// The keypath wasn't in the value...
if (self._auto_increment == false) {
// and auto-increment is disabled, no key, error.
return error.DataError;
}
if (Key.canInjectKey(value, kp) == false) {
return error.DataError;
}
generated = true;
const n = try self._engine.nextGeneratedKey(self._store_id);
const k = try local.newNumber(@floatFromInt(n));
try Key.injectKey(local, value, kp, k);
break :blk k;
break :blk .generate_in_line;
}
// Out-of-line keys.
if (key_arg) |k| {
break :blk k;
break :blk .{ .explicit = .{
.encoded = try Key.encodeValue(exec.arena, k),
.bump = if (self._auto_increment and k.isNumber()) try k.toF64() else null,
} };
}
if (self._auto_increment == false) {
return error.DataError;
}
generated = true;
const n = try self._engine.nextGeneratedKey(self._store_id);
break :blk try local.newNumber(@floatFromInt(n));
break :blk .generate_out_of_line;
};
const encoded = try Key.encodeValue(exec.call_arena, key_value);
const request = try txn.newRequest();
const value_global = try value.persist();
return request.submit(.{ .store_write = .{
.store = self,
.kind = kind,
.value = value_global,
.key = prepared,
} }, exec);
}
if (self._auto_increment and !generated and key_value.isNumber()) {
// auto-increment is enabled, but this was NOT a generated key, so we
// need to bump the generator so that future generated keys don't collide
try self._engine.maybeBumpGenerator(self._store_id, try key_value.toF64());
}
pub fn runWrite(self: *IDBObjectStore, request: *IDBRequest, kind: WriteKind, value_global: js.Value.Global, prepared: PreparedKey, exec: *Execution) !void {
self.writeInner(request, kind, value_global, prepared, exec) catch |err| {
log.warn(.storage, "idb write", .{ .err = err, .kind = kind });
request.setError(err);
};
}
fn writeInner(self: *IDBObjectStore, request: *IDBRequest, kind: WriteKind, value_global: js.Value.Global, prepared: PreparedKey, exec: *Execution) !void {
const local = exec.js.local.?;
const value = value_global.local(local);
// Resolve the encoded key + the JS value that becomes the request result. For
// generated keys, that's where the connection is finally touched.
const encoded: []const u8 = switch (prepared) {
.explicit => |e| blk: {
if (e.bump) |b| {
try self._engine.maybeBumpGenerator(self._store_id, b);
}
break :blk e.encoded;
},
.generate_out_of_line => blk: {
const n = try self._engine.nextGeneratedKey(self._store_id);
break :blk try Key.encodeValue(exec.call_arena, try local.newNumber(@floatFromInt(n)));
},
.generate_in_line => blk: {
const n = try self._engine.nextGeneratedKey(self._store_id);
const k = try local.newNumber(@floatFromInt(n));
try Key.injectKey(local, value, self._key_path.?, k);
break :blk try Key.encodeValue(exec.call_arena, k);
},
};
const serialized = try value.serialize();
defer serialized.deinit();
const request = try txn.newRequest();
// Record + index rows are atomic: a unique-index violation rolls the record
// write back too.
try self._engine.savepoint();
self.writeRecord(kind, encoded, serialized.bytes(), value, exec) catch |err| {
self._engine.rollbackSavepoint();
log.warn(.storage, "idb write", .{ .err = err, .kind = kind });
request.setError(err);
return request;
return err;
};
try self._engine.releaseSavepoint();
try request.setValue(key_value);
return request;
// The request result is the record's key (decoded back from the encoded bytes).
try request.setValue(try Key.decodeToJs(exec.call_arena, local, encoded));
}
fn writeRecord(self: *IDBObjectStore, kind: WriteKind, key: []const u8, bytes: []const u8, value: js.Value, exec: *Execution) !void {

82
src/browser/webapi/storage/idb/IDBRequest.zig
View File

@@ -26,8 +26,11 @@ const EventTarget = @import("../../EventTarget.zig");
const DOMException = @import("../../DOMException.zig");
const idb = @import("idb.zig");
const Engine = @import("Engine.zig");
const IDBIndex = @import("IDBIndex.zig");
const IDBCursor = @import("IDBCursor.zig");
const IDBDatabase = @import("IDBDatabase.zig");
const IDBObjectStore = @import("IDBObjectStore.zig");
const IDBTransaction = @import("IDBTransaction.zig");
const IDBVersionChangeEvent = @import("IDBVersionChangeEvent.zig");
@@ -37,11 +40,12 @@ const FunctionSetter = idb.FunctionSetter;
const IDBRequest = @This();
_proto: *EventTarget,
_result: Result = .{ .none = js.Undefined{} },
_op: Operation = .none,
_error: ?anyerror = null,
_txn: ?*IDBTransaction = null,
_cursor: ?*IDBCursor = null,
_ready_state: ReadyState = .pending,
_result: Result = .{ .none = js.Undefined{} },
_on_success: ?js.Function.Global = null,
_on_error: ?js.Function.Global = null,
@@ -184,6 +188,82 @@ fn getFunctionFromSetter(setter: ?FunctionSetter) ?js.Function.Global {
};
}
// A database operation, captured when a request method is called and run later.
pub const Operation = union(enum) {
none,
store_get: StoreQuery,
store_get_key: StoreQuery,
store_get_all: StoreGetAll,
store_count: StoreQuery,
store_delete: StoreQuery,
store_clear: *IDBObjectStore,
store_write: StoreWrite,
index_get: IndexQuery,
index_get_key: IndexQuery,
index_get_all: IndexGetAll,
index_count: IndexQuery,
cursor_iterate: CursorIterate,
cursor_update: CursorUpdate,
cursor_delete: CursorDelete,
const StoreQuery = struct { store: *IDBObjectStore, bounds: Engine.Bounds };
const StoreGetAll = struct { store: *IDBObjectStore, bounds: Engine.Bounds, column: Engine.Column, count: ?u32 };
const StoreWrite = struct { store: *IDBObjectStore, kind: IDBObjectStore.WriteKind, value: js.Value.Global, key: IDBObjectStore.PreparedKey };
const IndexQuery = struct { index: *IDBIndex, bounds: Engine.Bounds };
const IndexGetAll = struct { index: *IDBIndex, bounds: Engine.Bounds, column: Engine.Column, count: ?u32 };
const CursorIterate = struct { cursor: *IDBCursor, seek: IDBCursor.Seek, offset: u32 };
const CursorUpdate = struct { cursor: *IDBCursor, key: []const u8, value: js.Value.Global };
const CursorDelete = struct { cursor: *IDBCursor, key: []const u8 };
};
// Commit this request's operation: record it and queue the request on its
// transaction's drain.
pub fn submit(self: *IDBRequest, op: Operation, exec: *Execution) !*IDBRequest {
self._op = op;
const txn = self._txn.?;
try txn.enqueue(self);
if (txn.getMode() == .versionchange) {
// for "upgradeneeded" we run them immediately (and still queue them, but
// their .op == .noop at that point). This is necessary to keep this
// operation consistent / ordered with other "upgradeneeded" changes that
// happen synchronously (e.g. createIndex). And we queue it so that,
// while the re-execute will be noop, the events will still fire.
try self.execute(exec);
}
return self;
}
pub fn execute(self: *IDBRequest, exec: *Execution) !void {
const op = self._op;
if (op == .none) {
return;
}
self._op = .none;
if (self._txn) |txn| {
try txn.ensureBegun();
}
switch (op) {
.none => unreachable,
.store_get => |o| try o.store.runGet(self, o.bounds, exec),
.store_get_key => |o| try o.store.runGetKey(self, o.bounds, exec),
.store_get_all => |o| try o.store.runGetAll(self, o.bounds, o.column, o.count, exec),
.store_count => |o| try o.store.runCount(self, o.bounds, exec),
.store_delete => |o| try o.store.runDelete(self, o.bounds, exec),
.store_clear => |store| try store.runClear(self, exec),
.store_write => |o| try o.store.runWrite(self, o.kind, o.value, o.key, exec),
.index_get => |o| try o.index.runGet(self, o.bounds, exec),
.index_get_key => |o| try o.index.runGetKey(self, o.bounds, exec),
.index_get_all => |o| try o.index.runGetAll(self, o.bounds, o.column, o.count, exec),
.index_count => |o| try o.index.runCount(self, o.bounds, exec),
.cursor_iterate => |o| try o.cursor.runIterate(o.seek, o.offset, exec),
.cursor_update => |o| try o.cursor.runUpdate(self, o.key, o.value, exec),
.cursor_delete => |o| try o.cursor.runDelete(self, o.key),
}
}
pub const JsApi = struct {
pub const bridge = js.Bridge(IDBRequest);

34
src/browser/webapi/storage/idb/IDBTransaction.zig
View File

@@ -138,6 +138,10 @@ pub fn abort(self: *IDBTransaction, exec: *Execution) !void {
}
for (self._requests.items) |request| {
request._op = .none;
if (request._ready_state == .done) {
continue;
}
request.setError(error.AbortError);
request.deliver(exec) catch |err| {
log.warn(.storage, "idb abort deliver", .{ .err = err });
@@ -162,6 +166,13 @@ pub fn settle(self: *IDBTransaction, exec: *Execution) void {
self.fire(exec, comptime .wrap("complete"), self._on_complete);
}
// "is this transaction still usable". Once settled, it cannot be used.
pub fn assertActive(self: *const IDBTransaction) !void {
if (self._settled) {
return error.TransactionInactiveError;
}
}
pub fn ensureBegun(self: *IDBTransaction) !void {
if (self._settled) {
return error.TransactionInactiveError;
@@ -177,7 +188,6 @@ pub fn ensureBegun(self: *IDBTransaction) !void {
pub fn newRequest(self: *IDBTransaction) !*IDBRequest {
const request = try IDBRequest.init(self._exec);
request._txn = self;
try self.enqueue(request);
return request;
}
@@ -283,12 +293,28 @@ fn finalize(ctx: *anyopaque) void {
}
}
// Deliver every queued request in order. Re-reads len each iteration so a
// `success` handler can enqueue a follow-up request on this transaction.
fn deliverPending(self: *IDBTransaction, exec: *Execution) void {
// exec.js.local must be non-null. In some cases it is (e.g. tx.commit()
// called directly from JS), in others is isn't (drain schedule tasks).
// Easier to explicitly create one and then restore whatever was there before.
const prev_local = exec.js.local;
defer exec.js.local = prev_local;
var ls: js.Local.Scope = undefined;
exec.js.localScope(&ls);
defer ls.deinit();
exec.js.local = &ls.local;
var i: usize = 0;
while (i < self._requests.items.len) : (i += 1) {
self._requests.items[i].deliver(exec) catch |err| {
const request = self._requests.items[i];
request.execute(exec) catch |err| {
log.warn(.storage, "idb request execute", .{ .err = err });
request.setError(err);
};
request.deliver(exec) catch |err| {
log.warn(.storage, "idb request deliver", .{ .err = err });
};
}