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
1 change: 1 addition & 0 deletions news/+disk-state-load-errors.bugfix.md
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
The disk state manager now logs a warning when an existing persisted state file cannot be read or unpickled, instead of silently starting from a fresh state. A file whose stored schema no longer matches the state class is still reset, and logged at debug level.
26 changes: 24 additions & 2 deletions reflex/istate/manager/disk.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from typing import Any, Generic, cast

from reflex_base.environment import state_manager_disk_debounce
from reflex_base.utils.exceptions import StateSchemaMismatchError
from typing_extensions import Unpack, override

from reflex.istate.manager import (
Expand Down Expand Up @@ -146,6 +147,11 @@ def token_path(self, token: StateToken) -> Path:
async def load_state(self, token: StateToken[TOKEN_TYPE]) -> TOKEN_TYPE | None:
"""Load a state object based on the provided token.

A missing state file returns None without logging. A state file that
cannot be loaded also returns None: one whose stored schema no longer
matches the state class is logged at debug level, one that cannot be
read or unpickled as a warning.

Args:
token: The token used to identify the state object.

Expand All @@ -158,8 +164,24 @@ async def load_state(self, token: StateToken[TOKEN_TYPE]) -> TOKEN_TYPE | None:
try:
with token_path.open(mode="rb") as file:
return token.deserialize(fp=file)
except Exception:
pass
except Exception as err:
# BaseState also reports a file it cannot unpickle as a schema
# mismatch, chained to the unpickling error.
if isinstance(err, StateSchemaMismatchError) and err.__cause__ is None:
logger.debug(
"Discarding persisted state for %s from %s: it no longer "
"matches the state class.",
token,
token_path,
)
return None
logger.warning(
"Failed to load persisted state for %s from %s, starting "
"from a fresh state.",
token,
token_path,
exc_info=True,
)
return None

async def populate_substates(
Expand Down
110 changes: 110 additions & 0 deletions tests/units/istate/manager/test_disk.py
Original file line number Diff line number Diff line change
@@ -1,16 +1,22 @@
"""Tests for the disk state manager."""

import logging
import math
import os
import pickle
from pathlib import Path

import pytest

from reflex.istate.manager import disk
from reflex.istate.manager.disk import StateManagerDisk
from reflex.istate.manager.token import BaseStateToken, StateToken
from reflex.state import BaseState
from reflex.utils import prerequisites

# A pickle of a global from a module that does not exist.
MISSING_MODULE_PICKLE = b"cmissing_module_for_disk_test\nThing\n."


class DiskPersistState(BaseState):
"""A state for testing disk persistence."""
Expand Down Expand Up @@ -167,3 +173,107 @@ def recording_with_cls(self, cls):

assert written == [DiskTreeFirst]
await state_manager.close()


@pytest.mark.asyncio
async def test_load_state_missing_file_is_not_logged(tmp_path, monkeypatch, caplog):
"""A state that was never persisted loads as None without logging.

Args:
tmp_path: A temporary directory.
monkeypatch: The pytest monkeypatch fixture.
caplog: The pytest log capture fixture.
"""
monkeypatch.setattr(prerequisites, "get_states_dir", lambda: tmp_path)
state_manager = StateManagerDisk(_write_debounce_seconds=0)
token = StateToken(ident="client", cls=int)

with caplog.at_level(logging.DEBUG, logger=disk.__name__):
assert await state_manager.load_state(token) is None

assert not caplog.records
await state_manager.close()


@pytest.mark.asyncio
@pytest.mark.parametrize(
("token", "contents", "fresh_state"),
[
(StateToken(ident="client", cls=int), b"not-a-pickle", 0),
(StateToken(ident="client", cls=int), b"", 0),
(BaseStateToken(ident="client", cls=DiskPersistState), b"not-a-pickle", None),
(BaseStateToken(ident="client", cls=DiskPersistState), b"", None),
(
BaseStateToken(ident="client", cls=DiskPersistState),
MISSING_MODULE_PICKLE,
None,
),
],
)
async def test_load_state_logs_unreadable_file(
tmp_path, monkeypatch, caplog, token, contents, fresh_state
):
"""A persisted state file that cannot be read or unpickled is logged as a warning.

Args:
tmp_path: A temporary directory.
monkeypatch: The pytest monkeypatch fixture.
caplog: The pytest log capture fixture.
token: The token whose state file is damaged.
contents: The damaged file contents.
fresh_state: The expected fresh state, or None for a DiskPersistState.
"""
monkeypatch.setattr(prerequisites, "get_states_dir", lambda: tmp_path)
state_manager = StateManagerDisk(_write_debounce_seconds=0)
token_path = state_manager.token_path(token)
token_path.write_bytes(contents)

with caplog.at_level(logging.DEBUG, logger=disk.__name__):
assert await state_manager.load_state(token) is None

[record] = caplog.records
assert record.levelno == logging.WARNING
assert str(token) in record.getMessage()
assert str(token_path) in record.getMessage()
Comment thread
cubic-dev-ai[bot] marked this conversation as resolved.
Resolved
assert record.exc_info is not None

caplog.clear()
with caplog.at_level(logging.WARNING, logger=disk.__name__):
state = await state_manager.get_state(token)
assert [r.levelno for r in caplog.records] == [logging.WARNING]
if fresh_state is None:
assert isinstance(state, DiskPersistState)
assert math.isclose(state.num, 3.15)
else:
assert state == fresh_state
await state_manager.close()


@pytest.mark.asyncio
async def test_load_state_logs_schema_mismatch_at_debug(tmp_path, monkeypatch, caplog):
"""A persisted BaseState whose class changed since it was written is logged at debug.

Args:
tmp_path: A temporary directory.
monkeypatch: The pytest monkeypatch fixture.
caplog: The pytest log capture fixture.
"""
monkeypatch.setattr(prerequisites, "get_states_dir", lambda: tmp_path)
state_manager = StateManagerDisk(_write_debounce_seconds=0)
token = BaseStateToken(ident="client", cls=DiskPersistState)
token_path = state_manager.token_path(token)
stale_state = DiskPersistState(_reflex_internal_init=True) # pyright: ignore [reportCallIssue]
stale_state.num = 9.5
token_path.write_bytes(pickle.dumps(("stale-schema", stale_state)))

with caplog.at_level(logging.DEBUG, logger=disk.__name__):
assert await state_manager.load_state(token) is None

[record] = caplog.records
assert record.levelno == logging.DEBUG
assert str(token_path) in record.getMessage()

state = await state_manager.get_state(token)
assert isinstance(state, DiskPersistState)
assert math.isclose(state.num, 3.15)
await state_manager.close()
Loading