feat(rivetkit): isolate includeState transaction reads with a committed snapshot

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;

rivetkit-typescript/packages/rivetkit/src/common/database/config.ts

@@ -39,7 +39,9 @@ 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
+		 * observe the committed state (a snapshot taken when the transaction
+		 * opened), never the transaction's uncommitted writes.
 		 */
 		includeState?: boolean;
 	};

rivetkit-typescript/packages/rivetkit/src/registry/native.ts

@@ -307,6 +307,11 @@ type NativePersistActorState = {
 	pendingStateTransactionOwners?: Set<symbol>;
 	stateTransactionTail?: Promise<void>;
 	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 };
 };
 type NativeDestroyGate = {
 	destroyCompletion?: Promise<void>;
@@ -3094,11 +3099,21 @@ export class ActorContextHandleAdapter {
 			pendingOwners.delete(this.#stateTransactionOwner);
 			actorState.activeStateTransactionOwner =
 				this.#stateTransactionOwner;
+			// 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,
+				};
+			}
 			return {
 				actorContext: this,
-				actorStateBaseline: this.#stateEnabled
-					? structuredClone(this.#readState())
-					: undefined,
+				actorStateBaseline,
 				connectionStateBaselines: new Map(
 					callNativeSync(() =>
 						this.#runtime.actorConns(this.#ctx),
@@ -3132,6 +3147,9 @@ export class ActorContextHandleAdapter {
 				this.#restoreStateTransactionBaseline(scope);
 			}
 		} finally {
+			// Tear down the read snapshot so non-owner contexts see the
+			// committed (or restored) live state again.
+			actorState.committedStateSnapshot = undefined;
 			if (
 				actorState.activeStateTransactionOwner ===
 				this.#stateTransactionOwner
@@ -3447,6 +3465,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;
 	}
 

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) => {