From 5e50428341760374b0c556871892f059d885b5a9 Mon Sep 17 00:00:00 2001 From: ABCxFF <79597906+abcxff@users.noreply.github.com> Date: Tue, 1 Sep 2026 14:16:02 -0400 Subject: [PATCH] feat(rivetkit): isolate includeState transaction reads with a committed snapshot MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An includeState state transaction mutated actor state in place, so a concurrent action reading state mid-transaction observed the owner's uncommitted writes (a dirty read); the writes only reverted on rollback. That gave atomic commit but not read isolation. The same held for hibernatable connection state — and worse: a background save tick (serializeForTick, used for periodic/sleep saves) could serialize and durably persist uncommitted connection bytes, which rollback does not undo in storage. Snapshot the committed actor and connection state when the transaction opens. The transaction owner keeps mutating live state (so commit/rollback, onStateChange, and retained-proxy semantics are unchanged), but every non-owner context — actions, runtime save ticks, the inspector, sleep saves — reads the snapshot instead: - actor state: ActorContextHandleAdapter#readState returns the snapshot for non-owner contexts. - connection state: NativeConnAdapter#readState returns the committed connection snapshot for non-owner contexts (gated by a new ownsActiveStateTransaction() predicate threaded like assertCanMutateState), and serializeForTick serializes the committed connection bytes for non-owner contexts so a background save can't durably flush uncommitted connection state. Snapshots are torn down on transaction exit. Note: the driver-suite state-transaction tests require the native engine and could not be run in this environment (they fail identically on unmodified main — internal_error on stateTransactionCommit); validated by typecheck, the mock-provider unit tests, biome, and review. --- .../driver-test-suite/actor-db-raw.ts | 1 + .../rivetkit/src/common/database/config.ts | 6 +- .../packages/rivetkit/src/registry/native.ts | 123 ++++++++++++++++-- .../rivetkit/tests/driver/actor-db.test.ts | 32 +++++ 4 files changed, 150 insertions(+), 12 deletions(-) diff --git a/rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/actor-db-raw.ts b/rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/actor-db-raw.ts index 731f95cccc..cd35a576c5 100644 --- a/rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/actor-db-raw.ts +++ b/rivetkit-typescript/packages/rivetkit/fixtures/driver-test-suite/actor-db-raw.ts @@ -400,6 +400,7 @@ export const dbActorRaw = actor({ } await c.vars.stateTransactionStarted.promise; }, + readAtomicStateValue: (c) => c.state.atomicStateValue, mutateStateDuringTransaction: async (c, value: string) => { try { c.state.atomicStateValue = value; diff --git a/rivetkit-typescript/packages/rivetkit/src/common/database/config.ts b/rivetkit-typescript/packages/rivetkit/src/common/database/config.ts index 7ff08c39e3..f8ff914dea 100644 --- a/rivetkit-typescript/packages/rivetkit/src/common/database/config.ts +++ b/rivetkit-typescript/packages/rivetkit/src/common/database/config.ts @@ -39,7 +39,11 @@ export interface SqliteTransactionOptions { * Atomically includes actor and hibernatable connection state. * Only single-statement `execute` calls are supported in the transaction. * Concurrent actions that try to mutate state while the transaction is - * active fail with `actor.state_transaction_conflict`. + * active fail with `actor.state_transaction_conflict`. Concurrent reads + * of actor or connection state observe the committed values (a snapshot + * taken when the transaction opened), and background saves persist that + * snapshot, so the transaction's uncommitted writes are never observed + * or durably flushed until it commits. */ includeState?: boolean; }; diff --git a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts b/rivetkit-typescript/packages/rivetkit/src/registry/native.ts index 4d3a55b45e..887a64e377 100644 --- a/rivetkit-typescript/packages/rivetkit/src/registry/native.ts +++ b/rivetkit-typescript/packages/rivetkit/src/registry/native.ts @@ -333,6 +333,17 @@ type NativePersistActorState = { pendingStateTransactionOwners?: Set; stateTransactionTail?: Promise; stateTransactionSaveDeferred?: boolean; + // Present only while an includeState transaction is active and state is + // enabled. Holds a structured clone of the state as of transaction start. + // The owner keeps mutating the live `state`; every other context reads this + // snapshot so it observes only committed values until the owner commits. + committedStateSnapshot?: { value: unknown }; + // Committed hibernatable connection state (connId -> encoded bytes) captured + // when an includeState transaction opens. Non-owner contexts read and + // serialize these bytes instead of the connection's live state, so a + // concurrent read or a background save never sees or durably persists the + // transaction's uncommitted connection writes. + committedConnStateSnapshot?: Map; }; type NativeDestroyGate = { destroyCompletion?: Promise; @@ -1309,6 +1320,7 @@ class NativeConnAdapter { #ctx?: ActorContextHandle; #queueHibernationRemoval?: (connId: string) => void; #assertCanMutateState?: () => void; + #isStateTransactionOwner?: () => boolean; #stateProxy?: unknown; #stateProxyTarget?: unknown; @@ -1319,6 +1331,7 @@ class NativeConnAdapter { ctx?: ActorContextHandle, queueHibernationRemoval?: (connId: string) => void, assertCanMutateState?: () => void, + isStateTransactionOwner?: () => boolean, ) { this.#runtime = runtime; this.#conn = conn; @@ -1326,6 +1339,7 @@ class NativeConnAdapter { this.#ctx = ctx; this.#queueHibernationRemoval = queueHibernationRemoval; this.#assertCanMutateState = assertCanMutateState; + this.#isStateTransactionOwner = isStateTransactionOwner; ( this as NativeConnAdapter & { [CONN_STATE_MANAGER_SYMBOL]?: unknown; @@ -1443,6 +1457,25 @@ class NativeConnAdapter { return decodeValue(this.#runtime.connState(this.#conn)); } + // While another context's includeState transaction is mutating this + // connection's state, non-owner readers observe the committed snapshot + // rather than the owner's uncommitted writes. The owner keeps reading + // its live state. + const snapshot = getNativePersistState( + this.#runtime, + this.#ctx, + ).committedConnStateSnapshot; + if ( + snapshot !== undefined && + this.#isStateTransactionOwner !== undefined && + !this.#isStateTransactionOwner() + ) { + const committedBytes = snapshot.get(this.id); + if (committedBytes !== undefined) { + return decodeValue(committedBytes); + } + } + const connState = getNativeConnPersistState( this.#runtime, this.#ctx, @@ -2609,17 +2642,20 @@ class NativeConnectionMap implements ReadonlyMap { #ctx: ActorContextHandle; #schemas: NativeValidationConfig; #assertCanMutateState: () => void; + #isStateTransactionOwner?: () => boolean; constructor( runtime: CoreRuntime, ctx: ActorContextHandle, schemas: NativeValidationConfig, assertCanMutateState: () => void, + isStateTransactionOwner?: () => boolean, ) { this.#runtime = runtime; this.#ctx = ctx; this.#schemas = schemas; this.#assertCanMutateState = assertCanMutateState; + this.#isStateTransactionOwner = isStateTransactionOwner; } #connToAdapter(conn: ConnHandle): NativeConnAdapter { @@ -2636,6 +2672,7 @@ class NativeConnectionMap implements ReadonlyMap { ), ), this.#assertCanMutateState, + this.#isStateTransactionOwner, ); } @@ -2951,6 +2988,7 @@ export class ActorContextHandleAdapter { this.#ctx, this.#schemas, () => this.#assertCanMutateState(), + () => this.ownsActiveStateTransaction(), ); } return this.#connMap; @@ -3120,21 +3158,36 @@ export class ActorContextHandleAdapter { pendingOwners.delete(this.#stateTransactionOwner); actorState.activeStateTransactionOwner = this.#stateTransactionOwner; - return { - actorContext: this, - actorStateBaseline: this.#stateEnabled - ? structuredClone(this.#readState()) - : undefined, - connectionStateBaselines: new Map( - callNativeSync(() => - this.#runtime.actorConns(this.#ctx), - ).map((conn) => [ + // Snapshot the committed state up front. The owner mutates the live + // `state` in place; every non-owner context reads this snapshot + // instead, so actions observe only committed values while the + // transaction is open. Doubles as the rollback baseline. + const actorStateBaseline = this.#stateEnabled + ? structuredClone(this.#readState()) + : undefined; + if (this.#stateEnabled) { + actorState.committedStateSnapshot = { + value: actorStateBaseline, + }; + } + // Snapshot committed connection state too. Non-owner reads and + // background saves use these bytes instead of the connection's live + // (possibly uncommitted) state; also the rollback baseline. + const connectionStateBaselines = new Map( + callNativeSync(() => this.#runtime.actorConns(this.#ctx)).map( + (conn) => [ callNativeSync(() => this.#runtime.connId(conn)), new Uint8Array( callNativeSync(() => this.#runtime.connState(conn)), ), - ]), + ], ), + ); + actorState.committedConnStateSnapshot = connectionStateBaselines; + return { + actorContext: this, + actorStateBaseline, + connectionStateBaselines, committed: false, release, }; @@ -3158,6 +3211,10 @@ export class ActorContextHandleAdapter { this.#restoreStateTransactionBaseline(scope); } } finally { + // Tear down the read snapshots so non-owner contexts see the + // committed (or restored) live state again. + actorState.committedStateSnapshot = undefined; + actorState.committedConnStateSnapshot = undefined; if ( actorState.activeStateTransactionOwner === this.#stateTransactionOwner @@ -3299,13 +3356,27 @@ export class ActorContextHandleAdapter { this.#stateEnabled && this.#readState() !== undefined ? encodeValue(this.#readState()) : undefined; + // When another context's includeState transaction is mutating connection + // state, serialize the committed snapshot rather than the live bytes, so + // a background save can't durably persist connection state the + // transaction may still roll back. The owner (e.g. the commit path) is + // exempt so it flushes the values it is committing. + const isStateTransactionOwner = + actorState.activeStateTransactionOwner === + this.#stateTransactionOwner; + const connSnapshot = isStateTransactionOwner + ? undefined + : actorState.committedConnStateSnapshot; const connHibernation = callNativeSync(() => this.#runtime.actorDirtyHibernatableConns(this.#ctx), ).map((conn) => { const connId = callNativeSync(() => this.#runtime.connId(conn)); + const committedBytes = connSnapshot?.get(connId); return { connId, - bytes: callNativeSync(() => this.#runtime.connState(conn)), + bytes: + committedBytes ?? + callNativeSync(() => this.#runtime.connState(conn)), }; }); @@ -3473,6 +3544,17 @@ export class ActorContextHandleAdapter { callNativeSync(() => this.#runtime.actorState(this.#ctx)), ); } + // While a transaction owner is mutating the live state, every other + // context reads the committed snapshot so it never observes the owner's + // uncommitted writes. The owner itself keeps reading the live state. + const snapshot = actorState.committedStateSnapshot; + if ( + snapshot !== undefined && + actorState.activeStateTransactionOwner !== + this.#stateTransactionOwner + ) { + return snapshot.value; + } return actorState.state; } @@ -3510,6 +3592,20 @@ export class ActorContextHandleAdapter { this.#assertCanMutateState(); } + /** + * @internal + * True when this context owns the active includeState transaction. Used by + * paired connection adapters to decide whether they read live connection + * state (owner) or the committed snapshot (non-owner). + */ + ownsActiveStateTransaction(): boolean { + const actorState = getNativePersistState(this.#runtime, this.#ctx); + return ( + actorState.activeStateTransactionOwner === + this.#stateTransactionOwner + ); + } + // Coalesce the request-save and onStateChange work to once per event loop // tick. A synchronous burst of mutations (for example // `Object.assign(c.state, ...)`) would otherwise cross the NAPI boundary and @@ -3813,6 +3909,7 @@ function withConnContext( runtime.actorQueueHibernationRemoval(ctx, connId), ), () => actorContext.assertCanMutateState(), + () => actorContext.ownsActiveStateTransaction(), ), }); } @@ -4837,6 +4934,7 @@ export function buildNativeFactory( ), ), () => actorCtx.assertCanMutateState(), + () => actorCtx.ownsActiveStateTransaction(), ); try { const nextConnState = hasStaticConnState @@ -4895,6 +4993,7 @@ export function buildNativeFactory( ), ), () => actorCtx.assertCanMutateState(), + () => actorCtx.ownsActiveStateTransaction(), ); try { await config.onConnect( @@ -4939,6 +5038,8 @@ export function buildNativeFactory( ), ), () => actorCtx.assertCanMutateState(), + () => + actorCtx.ownsActiveStateTransaction(), ), ); } diff --git a/rivetkit-typescript/packages/rivetkit/tests/driver/actor-db.test.ts b/rivetkit-typescript/packages/rivetkit/tests/driver/actor-db.test.ts index 304969f794..77aa8fa050 100644 --- a/rivetkit-typescript/packages/rivetkit/tests/driver/actor-db.test.ts +++ b/rivetkit-typescript/packages/rivetkit/tests/driver/actor-db.test.ts @@ -825,6 +825,38 @@ describeDriverMatrix( dbTestTimeout, ); + test( + "exposes only committed state to concurrent reads during a state transaction", + async (c) => { + const { client } = await setupDriverTest( + c, + driverTestConfig, + ); + const actor = getDbActor(client, variant).getOrCreate([ + `db-${variant}-state-tx-read-iso-${crypto.randomUUID()}`, + ]); + await actor.reset(); + // Commit a known baseline so reads have a committed value. + await actor.stateTransactionCommit("committed"); + + const rollback = + actor.stateTransactionHoldAndRollback("held"); + await actor.waitForStateTransaction(); + // A concurrent (non-owner) action reads the committed value, + // never the owner's uncommitted "held" write. + expect(await actor.readAtomicStateValue()).toBe( + "committed", + ); + await actor.releaseStateTransaction(); + expect(await rollback).toBe("committed"); + // The committed value is still what reads observe afterward. + expect(await actor.readAtomicStateValue()).toBe( + "committed", + ); + }, + dbTestTimeout, + ); + test( "queues state transactions from separate actions", async (c) => {