You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@burr.apache.org by "andreahlert (via GitHub)" <gi...@apache.org> on 2026/05/23 09:47:35 UTC
[PR] feat: durable execution (suspend/resume + ctx.durable journal) [burr]
andreahlert opened a new pull request, #786:
URL: https://github.com/apache/burr/pull/786
## Summary
Adds durable execution to Burr, unifying two approaches in one design:
- **A: action-boundary suspend/resume** via `__context.suspend(channel, schema=, metadata=)`. The signal propagates through `_Suspended(BaseException)`, is persisted as a `SuspensionRecord`, and replayed by `resume(...)` / `aresume(...)`.
- **B: `__context.durable(key, fn, ...)` / `__context.adurable(...)`** sub-step journal. Memoizes side effects across re-runs, with `DeterminismError` raised fail-loud on mismatch.
Closes the gap where Burr could not natively pause for human input, external events, or crash recovery without manual state-fork workarounds.
## What's new
- `burr/core/durable.py` — `_Suspended`, `SuspensionRecord`, `JournalEntry`, `DeterminismError`, `supports_durable_storage()`.
- `burr/core/resume.py` — `resume()` and `aresume()` rebuild the app from `graph` + `persister`, deliver the payload, run to the next halt.
- `ApplicationContext.suspend / durable / adurable` + `_handle_suspension` / `_ahandle_suspension` in the sync and async run loops.
- 5 optional persister methods (`save_suspension`, `load_suspension`, `save_journal_entry`, `load_journal`, `mark_suspension_resolved`) with a `bool` contract on `mark_*`.
- First-party persister overrides with dedicated tables/collections: in-memory, SQLite, psycopg2 (PostgreSQL), asyncpg, aiosqlite, redis (sync + async), pymongo. Deprecated shims inherit transparently.
- Lifecycle hooks: `PostActionSuspendHook` + async, `PreActionResumeHook` + async. `LocalTrackingClient` emits `SuspendEntryModel` so the Burr UI can render `status="suspended"`.
- Persister status `Literal["completed", "failed", "suspended"]` propagated across the persistence layer.
- Example at `examples/durable-execution/` (runnable HITL workflow with notebook + statemachine.png) and concept docs at `docs/concepts/durable-execution.rst`.
## Test plan
- [x] `BURR_CI_INTEGRATION_TESTS=true POSTGRES_PORT=… REDIS_DB=… pytest tests/core/test_durable.py tests/core/test_durable_integration.py tests/core/test_durable_persisters.py` — **90 passed** with real Postgres, asyncpg, redis sync+async, pymongo, mongo shim backends.
- [x] `pytest tests/core/ tests/lifecycle/ tests/tracking/ tests/integrations/persisters/ --ignore=tests/core/test_graphviz_display.py` — **425 passed**, no regressions. The 25 skipped are env-gated persister tests.
- [x] End-to-end real-LLM test with ollama (`qwen2.5:1.5b`) drives the HITL example, simulates a crash by deleting the in-memory app + persister, resumes from SQLite, and asserts the LLM is invoked exactly twice end-to-end — proving the journal prevents side-effect re-fire on replay.
- [x] Resume idempotency verified: second `resume()` call on the same suspension is a no-op (no LLM call, final state unchanged).
- [x] `python examples/validate_examples.py` accepts the new example directory.
## Determinism contract (from the design)
- `key` stable across re-runs per call site.
- Same order, same set of `durable` calls across re-runs.
- No `durable` behind a non-deterministic branch.
- No `suspend` inside a `durable` fn.
- Mismatch raises `DeterminismError` (fail-loud).
## Notes
- `_Suspended` inherits from `BaseException` — do not wrap `__context.suspend()` inside `asyncio.shield(...)` or task-cancellation guards that catch `BaseException`.
- Custom persisters work through the in-state fallback (correct, not optimized for resume-once at concurrent scale).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@burr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: durable execution (suspend/resume + ctx.durable journal) [burr]
Posted by "andreahlert (via GitHub)" <gi...@apache.org>.
andreahlert commented on PR #786:
URL: https://github.com/apache/burr/pull/786#issuecomment-4645735963
Thanks for reviewing it @vaquarkhan. We are having a discution about this implementation on Dev list. Could you take a look and give your opinion about it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@burr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: durable execution (suspend/resume + ctx.durable journal) [burr]
Posted by "skrawcz (via GitHub)" <gi...@apache.org>.
skrawcz commented on PR #786:
URL: https://github.com/apache/burr/pull/786#issuecomment-5059620588
we should align on the docs & the examples here to really drive everyone's understanding of the feature.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@burr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: durable execution (suspend/resume + ctx.durable journal) [burr]
Posted by "sangkyoonnam (via GitHub)" <gi...@apache.org>.
sangkyoonnam commented on code in PR #786:
URL: https://github.com/apache/burr/pull/786#discussion_r4175696015
##########
burr/core/application.py:
##########
@@ -885,6 +1011,10 @@ def _context_factory(self, action: Action, sequence_id: int) -> ApplicationConte
state_initializer=self._state_initializer,
state_persister=self._state_persister,
action_name=action.name if action else None, # Pass just the action name
+ # _journal_call_index is intentionally not forwarded; it starts at 0 per action context.
+ _resume_signals=self._resume_signals,
Review Comment:
`_loaded_journal` and `_resume_signals` survive the resumed step. After resuming `review` (`durable("draft")`, then `suspend("approval")`), the next action's `durable("publish")` raises `DeterminismError #0`. With the same key, it returns `review`'s result without calling `fn`. In an `ask -> revise -> ask` loop, the second `ask` gets the first payload instead of suspending, and the next `resume()` is a no-op on the resolved record, so that payload is dropped.
##########
burr/core/resume.py:
##########
@@ -0,0 +1,265 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Top-level resume helpers for durable execution."""
+
+import warnings
+from typing import Any, List, Optional
+
+from burr.core.durable import (
+ read_journal_from_state,
+ read_suspension_from_state,
+ supports_durable_storage,
+)
+from burr.core.state import State
+from burr.lifecycle.base import LifecycleAdapter
+
+
+def _load_suspension(persister, partition_key, app_id, channel):
+ if supports_durable_storage(persister):
+ return persister.load_suspension(partition_key, app_id, channel)
+ loaded = persister.load(partition_key, app_id)
+ if loaded is None:
+ return None
+ return read_suspension_from_state(loaded["state"], channel)
+
+
+def _load_journal(persister, partition_key, app_id, sequence_id, state):
+ if supports_durable_storage(persister):
+ return persister.load_journal(partition_key, app_id, sequence_id)
+ return read_journal_from_state(state)
+
+
+async def _aload_suspension(persister, partition_key, app_id, channel):
+ if supports_durable_storage(persister):
+ if persister.is_async():
+ return await persister.load_suspension(partition_key, app_id, channel)
+ return persister.load_suspension(partition_key, app_id, channel)
+ if persister.is_async():
+ loaded = await persister.load(partition_key, app_id)
+ else:
+ loaded = persister.load(partition_key, app_id)
+ if loaded is None:
+ return None
+ return read_suspension_from_state(loaded["state"], channel)
+
+
+async def _aload_journal(persister, partition_key, app_id, sequence_id, state):
+ if supports_durable_storage(persister):
+ if persister.is_async():
+ return await persister.load_journal(partition_key, app_id, sequence_id)
+ return persister.load_journal(partition_key, app_id, sequence_id)
+ return read_journal_from_state(state)
+
+
+async def _arebuild(persister, graph, app_id, partition_key, record, hooks: Optional[List[LifecycleAdapter]] = None):
+ from burr.core.application import ApplicationBuilder
+
+ builder = (
+ ApplicationBuilder()
+ .with_graph(graph)
+ .with_identifiers(app_id=app_id, partition_key=partition_key)
+ .with_entrypoint(record.position)
+ .with_state(State(record.state))
+ .with_state_persister(persister)
+ )
+ for hook in (hooks or []):
+ builder = builder.with_hooks(hook)
+ if persister.is_async():
+ return await builder.abuild()
+ return builder.build()
+
+
+def _validate_payload(schema_json, payload):
+ """Validate *payload* against *schema_json* using jsonschema.
+
+ Schema validation requires the optional ``jsonschema`` package. When it is
+ absent, validation is skipped and a warning is emitted.
+ """
+ if schema_json is None:
+ return
+ try:
+ import jsonschema
+ except ImportError:
+ warnings.warn(
+ "jsonschema is not installed; skipping resume payload schema validation. "
+ "Install jsonschema to enable validation.",
+ stacklevel=3,
+ )
+ return
+ jsonschema.validate(instance=payload, schema=schema_json)
+
+
+def _rebuild(persister, graph, app_id, partition_key, record, hooks: Optional[List[LifecycleAdapter]] = None):
+ from burr.core.application import ApplicationBuilder
+
+ builder = (
+ ApplicationBuilder()
+ .with_graph(graph)
+ .with_identifiers(app_id=app_id, partition_key=partition_key)
+ .with_entrypoint(record.position)
+ .with_state(State(record.state))
+ .with_state_persister(persister)
+ )
+ for hook in (hooks or []):
+ builder = builder.with_hooks(hook)
+ return builder.build()
+
+
+def resume(
+ *,
+ persister,
+ graph,
+ app_id: str,
+ partition_key: Optional[str],
+ channel: str,
+ payload: Any,
+ hooks: Optional[List[LifecycleAdapter]] = None,
+):
+ """Resume a suspended run by delivering ``payload`` to ``channel``.
+
+ Reloads the suspension, rebuilds the Application from ``graph`` + ``persister``,
+ re-runs the suspended action from the top (durable sub-steps replay from the
+ journal, ``suspend(channel)`` returns ``payload``), and runs to the next halt,
+ suspend, or completion.
+
+ Idempotency: resuming an already-resolved suspension is an idempotent no-op for
+ persisters with durable storage (those implementing ``save_suspension`` /
+ ``load_suspension`` / ``mark_suspension_resolved``). For persisters without
+ durable storage, the suspension lives in ``state['__burr_durable__']`` and is
+ overwritten as the resumed run progresses; a second ``resume()`` call after the
+ first completes raises ``ValueError``.
+
+ :param hooks: Optional lifecycle adapters to register on the rebuilt application
+ before firing ``pre_action_resume``. Pass any adapter implementing
+ :class:`~burr.lifecycle.PreActionResumeHook` (or its async variant) here.
+ """
+ record = _load_suspension(persister, partition_key, app_id, channel)
+ if record is None:
+ raise ValueError(
+ f"No suspension found for app_id={app_id!r} "
+ f"(never suspended, or already resolved on a persister without durable storage)."
+ )
+ if record.resolved:
+ loaded = persister.load(partition_key, app_id)
+ return loaded["state"] if loaded else State(record.state)
+
+ _validate_payload(record.schema_json, payload)
+
+ app = _rebuild(persister, graph, app_id, partition_key, record, hooks=hooks)
+ app._resume_signals = {channel: payload}
+ app._loaded_journal = _load_journal(
+ persister, partition_key, app_id, record.sequence_id, record.state
+ )
+ app._suspended = None
+
+ app._adapter_set.call_all_lifecycle_hooks_sync(
+ "pre_action_resume",
+ app_id=app_id,
+ partition_key=partition_key,
+ action=graph.get_action(record.position),
+ sequence_id=record.sequence_id,
+ channel=channel,
+ )
+
+ app.run(halt_after=[]) # run to completion or the next suspend
Review Comment:
The resumed action runs at N+1 while the record says N. With a non-null partition:
```python
ctx.durable("a", f); ctx.suspend("ch1"); ctx.durable("b", g); ctx.suspend("ch2")
```
the first resume journals `b` at N+1 and reaches `ch2`; the second loads only `b` and raises `DeterminismError #0`. Even with the combined journal it would re-suspend at `ch1`, since only the new channel's payload is installed. One candidate identity is the last completed sequence id plus action name, within an app and partition. Suspended steps are saved as `completed` today, though, and sub-applications would need their own stable ids.
##########
burr/core/application.py:
##########
@@ -584,6 +585,125 @@ def my_action(state: State, __context: ApplicationContext) -> State:
state_persister: Optional[BaseStateSaver]
action_name: Optional[str] # Store just the action name
+ # --- Durable execution runtime state (populated per-step) ---
+ _resume_signals: Dict[str, Any] = dataclasses.field(default_factory=dict)
+ _loaded_journal: list = dataclasses.field(default_factory=list)
+ _journal_sink: list = dataclasses.field(default_factory=list)
+ _journal_call_index: int = 0
+
+ def suspend(
+ self,
+ channel: str,
+ *,
+ schema: Optional[type] = None,
+ metadata: Optional[dict] = None,
+ ) -> Any:
+ """Pause this run until an external event arrives on ``channel``.
+
+ First execution: raises the internal suspend signal; the run loop
+ persists a suspension record and stops. On resume: returns the payload
+ delivered to that channel. When ``schema`` is supplied and the payload
+ is a dict, the payload is coerced into the schema type via
+ ``schema(**payload)``; a non-dict payload is returned unchanged.
+
+ Pure control flow, no IO. Do not call from inside a ``durable()`` fn.
+ """
+ if channel in self._resume_signals:
+ payload = self._resume_signals[channel]
+ if schema is not None and isinstance(payload, dict):
+ return schema(**payload)
+ return payload
+ schema_json = (
+ schema.model_json_schema()
+ if schema is not None and hasattr(schema, "model_json_schema")
+ else None
+ )
+ from burr.core.durable import _Suspended
+
+ raise _Suspended(channel, schema_json, metadata)
+
+ def durable(self, key: str, fn: Callable, *args, **kwargs) -> Any:
+ """Memoize a sub-step. First run: execute ``fn`` and journal its result.
+ On resume: replay the journaled result without executing ``fn`` again.
+
+ ``key`` must be stable and called in the same order across re-runs of
+ the same action invocation (see the determinism contract). Do not call
+ ``suspend()`` from inside ``fn``, and do not wrap ``durable()`` in a
+ try/except that varies the key on the exception branch: the call index
+ is consumed even when ``fn`` raises, so a retry with a different key
+ triggers a ``DeterminismError``.
+ """
+ from burr.core.durable import DeterminismError, JournalEntry, supports_durable_storage
+
+ idx = self._journal_call_index
+ self._journal_call_index += 1
+
+ if idx < len(self._loaded_journal):
+ recorded = self._loaded_journal[idx]
+ if recorded.step_key != key:
+ raise DeterminismError(
+ f"Durable sub-step #{idx} replayed as key {key!r} but the "
+ f"journal recorded key {recorded.step_key!r}. The action's "
+ f"durable() calls must occur in the same order with the same "
+ f"keys on every re-run."
+ )
+ return recorded.result
+
+ result = fn(*args, **kwargs)
Review Comment:
If the process dies after `fn` does its work but before the result is written at L666, or `fn` raises, that work can run again when a retry reaches the call. `durable-execution.rst` L31 says sub-steps "never execute twice." Suggested wording: persisted results replay; calls without one may run again, so side effects need a per-operation idempotency key that the receiving system dedupes.
##########
burr/core/persistence.py:
##########
@@ -586,6 +661,162 @@ def save(
raise
self.connection.commit()
+ def create_durable_tables_if_not_exist(self):
+ """Creates the durable-execution tables (suspensions + journal) if they don't exist."""
+ cursor = self.connection.cursor()
+ cursor.execute(
+ """
+ CREATE TABLE IF NOT EXISTS burr_suspensions (
+ suspension_id TEXT PRIMARY KEY,
+ partition_key TEXT,
+ app_id TEXT NOT NULL,
+ sequence_id INTEGER NOT NULL,
+ position TEXT NOT NULL,
+ channel TEXT NOT NULL,
+ schema_json TEXT,
+ metadata_json TEXT,
+ inputs_json TEXT,
+ state_json TEXT NOT NULL,
+ created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+ resolved INTEGER NOT NULL DEFAULT 0
+ )"""
+ )
+ cursor.execute(
+ """
+ CREATE TABLE IF NOT EXISTS burr_journal (
+ partition_key TEXT,
+ app_id TEXT NOT NULL,
+ sequence_id INTEGER NOT NULL,
+ step_key TEXT NOT NULL,
+ call_index INTEGER NOT NULL,
+ result_json TEXT NOT NULL,
+ created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+ PRIMARY KEY (partition_key, app_id, sequence_id, step_key)
Review Comment:
With the default `partition_key=None`, this breaks resume:
```python
ctx.durable("a", f); ctx.durable("b", g); ctx.suspend("ch")
```
SQLite treats NULLs in this key as distinct, so the re-save in `_handle_suspension` (application.py:1136) adds a second copy of each row; replay sees `a, a, b, b` and resume raises `DeterminismError` at index 1. With psycopg2, Postgres rejects the insert with `NotNullViolation` (later failed-state persistence can mask it). SQLite's `save()` already maps `None` to `PARTITION_KEY_DEFAULT` (L645). Doing the same in the four journal/suspension methods that take a partition key, and dropping the re-save, would fix the journal side; Postgres's existing `save()` still passes `None` through. `InMemoryPersister` appends, so the re-save duplicates there too. The persister fixtures use `"pk"`, and the `None` example test makes one durable call.
##########
burr/core/application.py:
##########
@@ -584,6 +585,125 @@ def my_action(state: State, __context: ApplicationContext) -> State:
state_persister: Optional[BaseStateSaver]
action_name: Optional[str] # Store just the action name
+ # --- Durable execution runtime state (populated per-step) ---
+ _resume_signals: Dict[str, Any] = dataclasses.field(default_factory=dict)
+ _loaded_journal: list = dataclasses.field(default_factory=list)
+ _journal_sink: list = dataclasses.field(default_factory=list)
+ _journal_call_index: int = 0
+
+ def suspend(
+ self,
+ channel: str,
+ *,
+ schema: Optional[type] = None,
+ metadata: Optional[dict] = None,
+ ) -> Any:
+ """Pause this run until an external event arrives on ``channel``.
+
+ First execution: raises the internal suspend signal; the run loop
+ persists a suspension record and stops. On resume: returns the payload
+ delivered to that channel. When ``schema`` is supplied and the payload
+ is a dict, the payload is coerced into the schema type via
+ ``schema(**payload)``; a non-dict payload is returned unchanged.
+
+ Pure control flow, no IO. Do not call from inside a ``durable()`` fn.
+ """
+ if channel in self._resume_signals:
+ payload = self._resume_signals[channel]
+ if schema is not None and isinstance(payload, dict):
+ return schema(**payload)
+ return payload
+ schema_json = (
+ schema.model_json_schema()
+ if schema is not None and hasattr(schema, "model_json_schema")
+ else None
+ )
+ from burr.core.durable import _Suspended
+
+ raise _Suspended(channel, schema_json, metadata)
+
+ def durable(self, key: str, fn: Callable, *args, **kwargs) -> Any:
+ """Memoize a sub-step. First run: execute ``fn`` and journal its result.
+ On resume: replay the journaled result without executing ``fn`` again.
+
+ ``key`` must be stable and called in the same order across re-runs of
+ the same action invocation (see the determinism contract). Do not call
+ ``suspend()`` from inside ``fn``, and do not wrap ``durable()`` in a
+ try/except that varies the key on the exception branch: the call index
+ is consumed even when ``fn`` raises, so a retry with a different key
+ triggers a ``DeterminismError``.
+ """
+ from burr.core.durable import DeterminismError, JournalEntry, supports_durable_storage
+
+ idx = self._journal_call_index
+ self._journal_call_index += 1
+
+ if idx < len(self._loaded_journal):
Review Comment:
This compares keys by index only. With a non-null partition, a key repeated in a loop is upserted into one SQL row: with `f(i) = i`, after `[ctx.durable("llm", f, i) for i in range(3)]` and a suspend, resume returns `[2, 1, 2]` and calls `f` again for items 1 and 2. The docs already require unique keys, so raising on a duplicate within one invocation would fail explicitly instead.
##########
burr/core/persistence.py:
##########
@@ -586,6 +661,162 @@ def save(
raise
self.connection.commit()
+ def create_durable_tables_if_not_exist(self):
+ """Creates the durable-execution tables (suspensions + journal) if they don't exist."""
+ cursor = self.connection.cursor()
+ cursor.execute(
+ """
+ CREATE TABLE IF NOT EXISTS burr_suspensions (
+ suspension_id TEXT PRIMARY KEY,
+ partition_key TEXT,
+ app_id TEXT NOT NULL,
+ sequence_id INTEGER NOT NULL,
+ position TEXT NOT NULL,
+ channel TEXT NOT NULL,
+ schema_json TEXT,
+ metadata_json TEXT,
+ inputs_json TEXT,
+ state_json TEXT NOT NULL,
+ created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+ resolved INTEGER NOT NULL DEFAULT 0
+ )"""
+ )
+ cursor.execute(
+ """
+ CREATE TABLE IF NOT EXISTS burr_journal (
+ partition_key TEXT,
+ app_id TEXT NOT NULL,
+ sequence_id INTEGER NOT NULL,
+ step_key TEXT NOT NULL,
+ call_index INTEGER NOT NULL,
+ result_json TEXT NOT NULL,
+ created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
+ PRIMARY KEY (partition_key, app_id, sequence_id, step_key)
+ )"""
+ )
+ self.connection.commit()
+
+ def save_suspension(self, record: SuspensionRecord) -> None:
+ """Persist a suspension record into the burr_suspensions table."""
+ cursor = self.connection.cursor()
+ cursor.execute(
+ """INSERT OR REPLACE INTO burr_suspensions
+ (suspension_id, partition_key, app_id, sequence_id, position,
+ channel, schema_json, metadata_json, inputs_json, state_json,
+ created_at, resolved)
+ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
+ (
+ record.suspension_id,
+ record.partition_key,
+ record.app_id,
+ record.sequence_id,
+ record.position,
+ record.channel,
+ json.dumps(record.schema_json),
+ json.dumps(serde.serialize(record.metadata, **self.serde_kwargs)),
+ json.dumps(serde.serialize(record.inputs, **self.serde_kwargs)),
+ json.dumps(serde.serialize(record.state, **self.serde_kwargs)),
+ record.created_at,
+ 1 if record.resolved else 0,
+ ),
+ )
+ self.connection.commit()
+
+ def load_suspension(
+ self, partition_key: Optional[str], app_id: str, channel: str
+ ) -> Optional[SuspensionRecord]:
+ """Load the most recent suspension record for (partition_key, app_id, channel).
+
+ Returns the record whether or not it is resolved; callers check
+ ``record.resolved`` for resume-once idempotency. Returns ``None``
+ when no record exists for this combination.
+ """
+ cursor = self.connection.cursor()
+ cursor.execute(
+ """SELECT suspension_id, partition_key, app_id, sequence_id, position,
+ channel, schema_json, metadata_json, inputs_json, state_json,
+ created_at, resolved
+ FROM burr_suspensions
+ WHERE partition_key IS ? AND app_id = ? AND channel = ?
+ ORDER BY created_at DESC LIMIT 1""",
+ (partition_key, app_id, channel),
+ )
+ row = cursor.fetchone()
+ if row is None:
+ return None
+ return SuspensionRecord(
+ suspension_id=row[0],
+ partition_key=row[1],
+ app_id=row[2],
+ sequence_id=row[3],
+ position=row[4],
+ channel=row[5],
+ schema_json=json.loads(row[6]) if row[6] is not None else None,
+ metadata=serde.deserialize(json.loads(row[7]), **self.serde_kwargs) if row[7] is not None else None,
+ inputs=serde.deserialize(json.loads(row[8]), **self.serde_kwargs),
+ state=serde.deserialize(json.loads(row[9]), **self.serde_kwargs),
+ created_at=row[10],
+ resolved=bool(row[11]),
+ )
+
+ def mark_suspension_resolved(self, suspension_id: str) -> bool:
+ """Mark a suspension consumed. Conditional UPDATE for resume-once idempotency.
+
+ :return: True if a row was updated (first call), False if already resolved (no-op).
+ """
+ cursor = self.connection.cursor()
+ cursor.execute(
+ "UPDATE burr_suspensions SET resolved = 1 "
+ "WHERE suspension_id = ? AND resolved = 0",
+ (suspension_id,),
+ )
+ self.connection.commit()
+ return cursor.rowcount > 0
+
+ def save_journal_entry(self, entry: JournalEntry) -> None:
+ """Persist one memoized sub-step into the burr_journal table."""
+ cursor = self.connection.cursor()
+ cursor.execute(
+ """INSERT OR REPLACE INTO burr_journal
+ (partition_key, app_id, sequence_id, step_key, call_index,
+ result_json)
+ VALUES (?, ?, ?, ?, ?, ?)""",
+ (
+ entry.partition_key,
+ entry.app_id,
+ entry.sequence_id,
+ entry.step_key,
+ entry.call_index,
+ json.dumps(serde.serialize(entry.result, **self.serde_kwargs)),
Review Comment:
Separate from execution: `serde.serialize` falls back to `str(value)`, so with the default serializer a tuple or `datetime` returned from `fn` replays as `str` on SQLite. Rejecting values that don't round-trip, or documenting the codec requirement, would make that visible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@burr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] feat: durable execution (suspend/resume + ctx.durable journal) [burr]
Posted by "vaquarkhan (via GitHub)" <gi...@apache.org>.
vaquarkhan commented on PR #786:
URL: https://github.com/apache/burr/pull/786#issuecomment-4639622444
Hey @andreahlert ,I have spent some time going through this properly. I checked out the branch at ececa4b, built it, ran the suite, and wrote a few throwaway scripts to confirm the behavior rather than just eyeballing the diff.
Overall it's good work. The suspend/resume flow is clean, the DeterminismError guard is a smart safety net, and the mark_suspension_resolved on each persister is already atomic. I did find a handful of things worth fixing. I've noted how I confirmed each one so you can repro. Everything below is against ececa4b.
**1. resume never uses the atomic claim (blocking)**
burr/core/resume.py, resume() L157-184, same in aresume() L230-263.
resume() reads record.resolved at L157, runs the action at L179, then calls mark_suspension_resolved() at L184 and throws away the return. So two resumes racing (webhook retry, two workers on the same queue) both see resolved=False and both run the post-suspension action.
Every persister already returns the right thing for this:
```
SQLite: UPDATE ... WHERE suspension_id=? AND resolved=0, returns rowcount > 0
Postgres: WHERE suspension_id=%s AND resolved=false, returns rowcount > 0
pymongo: update_one({"_id": id, "resolved": False}, ...), returns modified_count == 1
redis: SETNX
```
The claim is built and correct everywhere. resume just marks it after the run and ignores whether it won. It should claim before the run.
**Note**: I read the code and confirmed the order plus that resume runs the action. I did not run two workers concurrently, so the race is from the code, not observed under load.
**Fix**, claim before running:
```
_validate_payload(record.schema_json, payload)
if supports_durable_storage(persister):
if not persister.mark_suspension_resolved(record.suspension_id):
loaded = persister.load(partition_key, app_id)
return loaded["state"] if loaded else State(record.state)
app = _rebuild(...)
# deliver payload, then app.run(halt_after=[])
# drop the mark_suspension_resolved() at L184
Same in aresume(). No persister changes.
```
**2. durable() doesn't replay across a crash, only across resume() (blocking)**
burr/core/resume.py L165 (resume()) and L241 (aresume()), the only places _loaded_journal gets real data.
_loaded_journal starts empty and only gets filled in resume/aresume. The normal crash path, initialize_from(..., resume_at_next_action=True), never touches it. So a process that dies mid-action and recovers the usual way re-runs every durable() sub-step. That's the cost the journal is supposed to kill.
Repro: two-action graph, crash the second action after its first durable() ran and journaled, recover in a fresh process via initialize_from:
```
loaded_journal on recovery: []
manual load_journal(pk, app, seq): [('step_a', 'A')] # row is there, key matches
step_a re-fired on recovery? True
```
The row is saved and the key lines up (I checked the sequence_id, it matches, my first guess that it'd be off was wrong). Runtime just never loads it outside resume().
**Fix**: load the journal on the recovery path too when the persister is durable:
```
if supports_durable_storage(persister):
app._loaded_journal = persister.load_journal(partition_key, app_id, sequence_id)
Plus a test for exactly-once across initialize_from with no suspend/resume.
```
If the intent is that durable() only covers deliberate suspend/resume and not crashes, fine, but then the docs shouldn't call it crash recovery.
**3. in-state journal grows and gets rewritten every step**
burr/core/application.py, _step() L1064-1066 and _astep() L1358-1360.
```
merged = read_journal_from_state(self._state) + self._journal_sink
new_state = write_journal_into_state(new_state, merged)
```
Never cleared after an action finishes, so __burr_durable__.journal piles up for the whole run and the whole blob gets rewritten every step. N sub-steps means N-squared writes. Bad with large LLM state.
(The commit named "flush in-state journal on action completion" is the one that adds this merge-and-grow. It doesn't flush.)
Repro, three actions each with one durable() call, in-state persister:
after a1: journal has 1 -> ['k_a']
after a2: journal has 2 -> ['k_a', 'k_b']
after a3: journal has 3 -> ['k_a', 'k_b', 'k_c']
Fix: drop __burr_durable__.journal on successful completion, keep only what belongs to the suspended action.
**4. partial persister override treated as fully durable**
burr/core/durable.py, supports_durable_storage() L95-113.
Detection only checks save_suspension (L113). Override save_suspension but not save_journal_entry and this returns True, then you hit NotImplementedError at journal-write time, after the expensive sub-step already ran. Docstring admits it. Bad edge for anyone writing the custom DynamoDB/S3 persisters this is built for.
Repro, persister overriding only save_suspension:
```
supports_durable_storage(partial) = True
partial.save_journal_entry(...) -> NotImplementedError
```
**Fix**: check all five methods (save_suspension, load_suspension, save_journal_entry, load_journal, mark_suspension_resolved) at register/build time and fail clearly on partial override.
**5. suspend(schema=...) skips validation for non-Pydantic types**
burr/core/application.py, suspend() L617-623.
schema_json only computed if the schema has model_json_schema (L617-618). Pass a dataclass or TypedDict and it stays None, so _validate_payload does nothing and nobody is told. Confirmed a dataclass has no model_json_schema.
**Fix**: require a Pydantic model with a clear TypeError otherwise, or support dataclass/TypedDict, or at minimum warn so it isn't silent.
**6. two shim tests fail instead of skip without the DB driver (minor)**
tests/core/test_durable_persisters.py L544 and L559.
The test bodies import the persister module which does a plain import pymongo / import psycopg2. No driver means ModuleNotFoundError raised inside the test, so pytest marks them FAILED instead of skipping. Every other DB test in the file skips. Confirmed: ModuleNotFoundError: No module named 'psycopg2', reported as 1 failed.
**Fix**: pytest.importorskip("pymongo") / pytest.importorskip("psycopg2").
That's everything I caught. To be upfront about #1: I'm confident the race is real from reading the code, but I didn't actually run two workers against the same suspension to watch it happen, so treat that one as "proven on paper, not under load." Everything else I reproduced directly.
My take is that #1 and #2 are the ones to sort before this lands, since together they're what make the exactly-once story actually hold up under concurrency and after a crash.
The rest can follow whenever. Happy to send a small PR for #1 and #2 if that's easier than describing it, neither needs any persister changes. Either way, thanks for putting this together, it's a genuinely useful feature.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@burr.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org