Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
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
4 changes: 2 additions & 2 deletions burr/core/persistence.py
Original file line number Diff line number Diff line change
Expand Up @@ -491,11 +491,11 @@ def load(
cursor = self.connection.cursor()
try:
if app_id is None:
# get latest for all app_ids
# CURRENT_TIMESTAMP has second precision; break ties by insertion order.
cursor.execute(
f"SELECT position, state, sequence_id, app_id, created_at, status FROM {self.table_name} "
f"WHERE partition_key = ? "
f"ORDER BY CREATED_AT DESC LIMIT 1",
f"ORDER BY created_at DESC, rowid DESC LIMIT 1",
(partition_key,),
)
elif sequence_id is None:
Expand Down
4 changes: 2 additions & 2 deletions burr/integrations/persisters/b_aiosqlite.py
Original file line number Diff line number Diff line change
Expand Up @@ -229,11 +229,11 @@ async def load(
logger.debug("Loading %s, %s, %s", partition_key, app_id, sequence_id)
cursor = await self.connection.cursor()
if app_id is None:
# get latest for all app_ids
# CURRENT_TIMESTAMP has second precision; break ties by insertion order.
await cursor.execute(
f"SELECT position, state, sequence_id, app_id, created_at, status FROM {self.table_name} "
f"WHERE partition_key = ? "
f"ORDER BY CREATED_AT DESC LIMIT 1",
f"ORDER BY created_at DESC, rowid DESC LIMIT 1",
(partition_key,),
)
elif sequence_id is None:
Expand Down
40 changes: 40 additions & 0 deletions tests/core/test_persistence.py
Original file line number Diff line number Diff line change
Expand Up @@ -131,6 +131,46 @@ def test_sqlite_persister_list_app_ids_without_initialize_raises_runtime_error()
persister.cleanup()


@pytest.mark.parametrize("partition_key", [None, "partition"])
@pytest.mark.parametrize("latest_app,latest_sequence", [("a-app", 12), ("z-app", 0)])
def test_sqlite_load_latest_breaks_timestamp_ties(
tmp_path, partition_key, latest_app, latest_sequence
):
db_path = str(tmp_path / "checkpoints.db")
with SQLLitePersister(db_path=db_path, table_name="test_table") as persister:
persister.initialize()
persister.save(partition_key, "a-app", 11, "first", State({"step": 1}), "completed")
persister.save(
partition_key, latest_app, latest_sequence, "second", State({"step": 2}), "completed"
)
persister.save("other-partition", "other-app", 999, "other", State({}), "completed")
# Model saves within the same second without depending on wall-clock timing.
persister.connection.execute("UPDATE test_table SET created_at = '2026-01-01 00:00:00'")
persister.connection.commit()

with SQLLitePersister(db_path=db_path, table_name="test_table") as persister:
loaded = persister.load(partition_key, None)
assert loaded["app_id"] == latest_app
assert loaded["sequence_id"] == latest_sequence
assert loaded["position"] == "second"
assert loaded["state"] == State({"step": 2})
assert loaded["status"] == "completed"
assert persister.load(partition_key, "a-app", 11)["state"] == State({"step": 1})


def test_sqlite_load_latest_prioritizes_timestamp(initializing_persistence):
persister = initializing_persistence
persister.initialize()
persister.save("partition", "a-app", 1, "first", State({"step": 1}), "completed")
persister.save("partition", "z-app", 0, "second", State({"step": 2}), "completed")
persister.connection.execute(
"UPDATE test_table SET created_at = CASE app_id "
"WHEN 'a-app' THEN '2026-01-02 00:00:00' ELSE '2026-01-01 00:00:00' END"
)
persister.connection.commit()
assert persister.load("partition", None)["state"] == State({"step": 1})


@pytest.mark.parametrize(
"method_name,kwargs",
[
Expand Down
46 changes: 46 additions & 0 deletions tests/integrations/persisters/test_b_aiosqlite.py
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,52 @@ async def test_async_persistence_lists_app_ids(async_persistence):
assert set(app_ids) == set(["app_id1", "app_id2"])


@pytest.mark.parametrize("partition_key", [None, "partition"])
@pytest.mark.parametrize("latest_app,latest_sequence", [("a-app", 12), ("z-app", 0)])
async def test_async_sqlite_load_latest_breaks_timestamp_ties(
tmp_path, partition_key, latest_app, latest_sequence
):
db_path = str(tmp_path / "checkpoints.db")
async with AsyncSQLitePersister.from_values(
db_path=db_path, table_name="test_table"
) as persister:
await persister.initialize()
await persister.save(partition_key, "a-app", 11, "first", State({"step": 1}), "completed")
await persister.save(
partition_key, latest_app, latest_sequence, "second", State({"step": 2}), "completed"
)
await persister.save("other-partition", "other-app", 999, "other", State({}), "completed")
# Model saves within the same second without depending on wall-clock timing.
await persister.connection.execute(
"UPDATE test_table SET created_at = '2026-01-01 00:00:00'"
)
await persister.connection.commit()

async with AsyncSQLitePersister.from_values(
db_path=db_path, table_name="test_table"
) as persister:
loaded = await persister.load(partition_key, None)
assert loaded["app_id"] == latest_app
assert loaded["sequence_id"] == latest_sequence
assert loaded["position"] == "second"
assert loaded["state"] == State({"step": 2})
assert loaded["status"] == "completed"
assert (await persister.load(partition_key, "a-app", 11))["state"] == State({"step": 1})


async def test_async_sqlite_load_latest_prioritizes_timestamp(async_persistence):
persister = async_persistence
await persister.initialize()
await persister.save("partition", "a-app", 1, "first", State({"step": 1}), "completed")
await persister.save("partition", "z-app", 0, "second", State({"step": 2}), "completed")
await persister.connection.execute(
"UPDATE test_table SET created_at = CASE app_id "
"WHEN 'a-app' THEN '2026-01-02 00:00:00' ELSE '2026-01-01 00:00:00' END"
)
await persister.connection.commit()
assert (await persister.load("partition", None))["state"] == State({"step": 1})


@pytest.mark.parametrize(
"method_name,kwargs",
[
Expand Down