Skip to content

Transactions fixes #129

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
onelxj wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Transactions fixes #129

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f9e8928
Key `ongoing_transaction_states` by transaction
onelxj Aug 14, 2026
dc744e8
Take a state exclusively when a transaction will persist to it
onelxj Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion documentation/docs/learn_more/auth.mdx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,7 @@ async def verify_token(

<TabItem value="typescript" label="TypeScript">
<!-- MARKDOWN-AUTO-DOCS:START
(CODE:src=../../../reboot/nodejs/index.ts&lines=664-667) -->
(CODE:src=../../../reboot/nodejs/index.ts&lines=668-671) -->
<!-- The below code snippet is automatically added from ../../../reboot/nodejs/index.ts -->

```ts
Expand Down
5 changes: 5 additions & 0 deletions rbt/v1alpha1/nodejs.proto
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,11 @@ message Context {
bool app_internal = 6;
optional bytes auth = 7;
optional string workflow_id = 8;
// The outermost transaction this call belongs to, if any. Several
// transactions may be joined to one state at the same time, so
// anything caching per-state data for the duration of a call must
// key on this too.
optional string transaction_root_id = 9;
}

////////////////////////////////////////////////////////////////////////
Expand Down
10 changes: 8 additions & 2 deletions reboot/aio/state_managers.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -286,9 +286,15 @@ def from_context(
# A `writer` is always exclusive; a `reader` is
# shared, and a `transaction` starts shared and may
# upgrade later if it requires exclusive.
#
# An idempotency key is itself something to persist,
# so a transaction carrying one starts exclusive to
# ensure it is persisted properly.
mode=(
Lock.Mode.EXCLUSIVE
if isinstance(context, WriterContext) else Lock.Mode.SHARED
Lock.Mode.EXCLUSIVE if (
isinstance(context, WriterContext) or
context.idempotency_key is not None
) else Lock.Mode.SHARED
),
idempotency_key=context.idempotency_key,
using_restart_detection=using_restart_detection,
Expand Down
8 changes: 8 additions & 0 deletions reboot/nodejs/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -312,6 +312,7 @@ export class Context {
readonly appInternal: boolean;
readonly auth: Auth | null;
readonly workflowId: string | null;
readonly transactionRootId: string | null;

constructor({
external,
Expand All @@ -323,6 +324,7 @@ export class Context {
appInternal,
auth,
workflowId,
transactionRootId,
cancelled,
}: {
external: any;
Expand All @@ -334,6 +336,7 @@ export class Context {
appInternal: boolean;
auth: Auth | null;
workflowId: string | null;
transactionRootId: string | null;
cancelled: Promise<void>;
}) {
if (!Context.#isInternalConstructing) {
Expand All @@ -351,6 +354,7 @@ export class Context {
this.appInternal = appInternal;
this.auth = auth;
this.workflowId = workflowId;
this.transactionRootId = transactionRootId;
this.cancelled = cancelled;
}

Expand Down Expand Up @@ -685,6 +689,10 @@ export abstract class TokenVerifier {
auth: null,
workflowId:
call.context.workflowId !== undefined ? call.context.workflowId : null,
transactionRootId:
call.context.transactionRootId !== undefined
? call.context.transactionRootId
: null,
cancelled,
}) as ReaderContext;

Expand Down
8 changes: 8 additions & 0 deletions reboot/nodejs/python.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -315,6 +315,10 @@ async def authorize(
None if context.workflow_id is None else
str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None else
str(context.transaction_root_id)
),
),
state=(
None if state is None else state.SerializeToString()
Expand Down Expand Up @@ -385,6 +389,10 @@ async def verify_token(
None if context.workflow_id is None else
str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None else
str(context.transaction_root_id)
),
),
token=token,
).SerializeToString(),
Expand Down
26 changes: 23 additions & 3 deletions reboot/templates/reboot.py.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,14 +143,22 @@ ongoing_transaction_states: dict[str, IMPORT_reboot_api.Model] = {}


# Helper to get the `ongoing_transaction_states` dictionary key.
# The key should contain both the state ID and the state type name
# to avoid conflicts when multiple states share the same ID.
# The key contains the state type name and the state ID to avoid
# conflicts when multiple states share the same ID, and the root
# transaction ID because several transactions may be joined to one
# state at the same time (they join it "shared", and only upgrade to
# "exclusive" once they write). Without the transaction ID they would
# share a single entry, and whichever registered last would receive
# the others' writer updates, silently losing their effects.
def ongoing_transaction_state_key(
context: IMPORT_reboot_aio_contexts.ReaderContext
| IMPORT_reboot_aio_contexts.WriterContext
| IMPORT_reboot_aio_contexts.TransactionContext,
) -> str:
return f"{context.state_type_name}/{context.state_id}"
return (
f"{context.state_type_name}/{context.state_id}/"
f"{context.transaction_root_id}"
)
{% endif %}

{# Since Pydantic model fields might be 'Optional' or have defaults, we
Expand Down Expand Up @@ -7722,6 +7730,10 @@ class {{ state.proto.name }}ServicerNodeAdaptor({{ state.proto.name }}.singleton
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
state=state.SerializeToString(),
request=request.SerializeToString(),
Expand Down Expand Up @@ -7836,6 +7848,10 @@ class {{ state.proto.name }}ServicerNodeAdaptor({{ state.proto.name }}.singleton
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
state=state.SerializeToString(),
request=request.SerializeToString(),
Expand Down Expand Up @@ -7956,6 +7972,10 @@ class {{ state.proto.name }}ServicerNodeAdaptor({{ state.proto.name }}.singleton
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
request=request.SerializeToString(),
).SerializeToString()
Expand Down
21 changes: 18 additions & 3 deletions reboot/templates/reboot.ts.j2
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,12 +110,17 @@ import { api } from "{{ options.proto.zod }}{% if nodejs_extensions %}.js{% endi
const ongoingTransactionStates: { [id: string] : any; } = {};

// Helper to get the `ongoingTransactionStates` dictionary key.
// The key should contain both the state ID and the state type name
// to avoid conflicts when multiple states share the same ID.
// The key contains the state type name and the state ID to avoid
// conflicts when multiple states share the same ID, and the root
// transaction ID because several transactions may be joined to one
// state at the same time (they join it "shared", and only upgrade to
// "exclusive" once they write). Without the transaction ID they would
// share a single entry, and whichever registered last would receive
// the others' writer updates, silently losing their effects.
const ongoingTransactionStateKey = (
context: Context
): string => {
return `${context.stateTypeName}/${context.stateId}`;
return `${context.stateTypeName}/${context.stateId}/${context.transactionRootId}`;
};

// Track state IDs that are being _constructed_ in a transaction
Expand Down Expand Up @@ -750,6 +755,11 @@ export abstract class {{ state.proto.name }}BaseServicer extends reboot.Servicer
? call.context.workflowId
: null
),
transactionRootId: (
call.context.transactionRootId !== undefined
? call.context.transactionRootId
: null
),
cancelled,
});

Expand Down Expand Up @@ -1345,6 +1355,11 @@ export class {{ state.proto.name }}Authorizer extends reboot.Authorizer<{{ state
? call.context.workflowId
: null
),
transactionRootId: (
call.context.transactionRootId !== undefined
? call.context.transactionRootId
: null
),
cancelled,
}) as reboot.ReaderContext;

Expand Down
56 changes: 56 additions & 0 deletions tests/reboot/echo_rbt.golden.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28961,6 +28961,10 @@ async def Reply(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
state=state.SerializeToString(),
request=request.SerializeToString(),
Expand Down Expand Up @@ -29082,6 +29086,10 @@ async def Replay(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
state=state.SerializeToString(),
request=request.SerializeToString(),
Expand Down Expand Up @@ -29218,6 +29226,10 @@ async def SearchAndReplace(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
state=state.SerializeToString(),
request=request.SerializeToString(),
Expand Down Expand Up @@ -29353,6 +29365,10 @@ async def FailOnceShouldBeRetried(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
state=state.SerializeToString(),
request=request.SerializeToString(),
Expand Down Expand Up @@ -29474,6 +29490,10 @@ async def FailOnceShouldBeRetriedWorkflow(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
request=request.SerializeToString(),
).SerializeToString()
Expand Down Expand Up @@ -29587,6 +29607,10 @@ async def TooManyTasks(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
state=state.SerializeToString(),
request=request.SerializeToString(),
Expand Down Expand Up @@ -29708,6 +29732,10 @@ async def Hanging(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
request=request.SerializeToString(),
).SerializeToString()
Expand Down Expand Up @@ -29807,6 +29835,10 @@ async def ReactiveWorkflow(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
request=request.SerializeToString(),
).SerializeToString()
Expand Down Expand Up @@ -29906,6 +29938,10 @@ async def ControlLoop(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
request=request.SerializeToString(),
).SerializeToString()
Expand Down Expand Up @@ -30005,6 +30041,10 @@ async def AtMostOnceWorkflow(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
request=request.SerializeToString(),
).SerializeToString()
Expand Down Expand Up @@ -30104,6 +30144,10 @@ async def WorkflowCallingWorkflow(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
request=request.SerializeToString(),
).SerializeToString()
Expand Down Expand Up @@ -30217,6 +30261,10 @@ async def RaiseValueError(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
state=state.SerializeToString(),
request=request.SerializeToString(),
Expand Down Expand Up @@ -30352,6 +30400,10 @@ async def RaiseSpecifiedError(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
state=state.SerializeToString(),
request=request.SerializeToString(),
Expand Down Expand Up @@ -30473,6 +30525,10 @@ async def FailingWorkflow(
None if context.workflow_id is None
else str(context.workflow_id)
),
transaction_root_id=(
None if context.transaction_root_id is None
else str(context.transaction_root_id)
),
),
request=request.SerializeToString(),
).SerializeToString()
Expand Down
Loading
Loading