a7d6d88f6f
CI / changes (push) Has been cancelled
CI / cd libs/checkpoint (push) Has been cancelled
CI / cd libs/checkpoint-conformance (push) Has been cancelled
CI / cd libs/checkpoint-postgres (push) Has been cancelled
CI / cd libs/checkpoint-sqlite (push) Has been cancelled
CI / cd libs/cli (push) Has been cancelled
CI / cd libs/prebuilt (push) Has been cancelled
CI / cd libs/sdk-py (push) Has been cancelled
CI / cd libs/langgraph (push) Has been cancelled
CI / Check SDK methods matching (push) Has been cancelled
CI / Check CLI schema hasn't changed #3.13 (push) Has been cancelled
CI / CLI integration test (push) Has been cancelled
CI / sdk-py integration test (push) Has been cancelled
CI / CI Success (push) Has been cancelled
baseline / benchmark (push) Has been cancelled
Deploy Redirects to GitHub Pages / deploy (push) Has been cancelled
430 lines
13 KiB
Python
430 lines
13 KiB
Python
"""Test SQLite store Time-To-Live (TTL) functionality."""
|
|
|
|
import asyncio
|
|
import os
|
|
import tempfile
|
|
import time
|
|
from collections.abc import Generator
|
|
|
|
import pytest
|
|
from langgraph.store.base import TTLConfig
|
|
|
|
from langgraph.store.sqlite import SqliteStore
|
|
from langgraph.store.sqlite.aio import AsyncSqliteStore
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_db_file() -> Generator[str, None, None]:
|
|
"""Create a temporary database file for testing."""
|
|
fd, path = tempfile.mkstemp()
|
|
os.close(fd)
|
|
yield path
|
|
os.unlink(path)
|
|
|
|
|
|
def test_ttl_basic(temp_db_file: str) -> None:
|
|
"""Test basic TTL functionality with synchronous API."""
|
|
ttl_seconds = 1
|
|
ttl_minutes = ttl_seconds / 60
|
|
|
|
with SqliteStore.from_conn_string(
|
|
temp_db_file, ttl={"default_ttl": ttl_minutes}
|
|
) as store:
|
|
store.setup()
|
|
|
|
store.put(("test",), "item1", {"value": "test"})
|
|
|
|
item = store.get(("test",), "item1")
|
|
assert item is not None
|
|
assert item.value["value"] == "test"
|
|
|
|
time.sleep(ttl_seconds + 1.0)
|
|
|
|
store.sweep_ttl()
|
|
|
|
item = store.get(("test",), "item1")
|
|
assert item is None
|
|
|
|
|
|
@pytest.mark.flaky(retries=3)
|
|
def test_ttl_refresh(temp_db_file: str) -> None:
|
|
"""Test TTL refresh on read."""
|
|
ttl_seconds = 1
|
|
ttl_minutes = ttl_seconds / 60
|
|
|
|
with SqliteStore.from_conn_string(
|
|
temp_db_file, ttl={"default_ttl": ttl_minutes, "refresh_on_read": True}
|
|
) as store:
|
|
store.setup()
|
|
|
|
# Store an item with TTL
|
|
store.put(("test",), "item1", {"value": "test"})
|
|
|
|
# Sleep almost to expiration
|
|
time.sleep(ttl_seconds - 0.5)
|
|
swept = store.sweep_ttl()
|
|
assert swept == 0
|
|
|
|
# Get the item and refresh TTL
|
|
item = store.get(("test",), "item1", refresh_ttl=True)
|
|
assert item is not None
|
|
|
|
time.sleep(ttl_seconds - 0.5)
|
|
swept = store.sweep_ttl()
|
|
assert swept == 0
|
|
|
|
# Get the item, should still be there
|
|
item = store.get(("test",), "item1")
|
|
assert item is not None
|
|
assert item.value["value"] == "test"
|
|
|
|
# Sleep again but don't refresh this time
|
|
time.sleep(ttl_seconds + 0.75)
|
|
|
|
swept = store.sweep_ttl()
|
|
assert swept == 1
|
|
|
|
# Item should be gone now
|
|
item = store.get(("test",), "item1")
|
|
assert item is None
|
|
|
|
|
|
def test_ttl_sweeper(temp_db_file: str) -> None:
|
|
"""Test TTL sweeper thread."""
|
|
ttl_seconds = 2
|
|
ttl_minutes = ttl_seconds / 60
|
|
|
|
ttl_config: TTLConfig = {
|
|
"default_ttl": ttl_minutes,
|
|
"sweep_interval_minutes": ttl_minutes / 2,
|
|
}
|
|
with SqliteStore.from_conn_string(
|
|
temp_db_file,
|
|
ttl=ttl_config,
|
|
) as store:
|
|
store.setup()
|
|
|
|
# Start the TTL sweeper
|
|
store.start_ttl_sweeper()
|
|
|
|
# Store an item with TTL
|
|
store.put(("test",), "item1", {"value": "test"})
|
|
|
|
# Item should be there initially
|
|
item = store.get(("test",), "item1")
|
|
assert item is not None
|
|
|
|
# Wait for TTL to expire and the sweeper to run
|
|
time.sleep(ttl_seconds + (ttl_seconds / 2) + 0.5)
|
|
|
|
# Item should be gone now (swept automatically)
|
|
item = store.get(("test",), "item1")
|
|
assert item is None
|
|
|
|
# Stop the sweeper
|
|
store.stop_ttl_sweeper()
|
|
|
|
|
|
@pytest.mark.flaky(retries=3)
|
|
def test_ttl_custom_value(temp_db_file: str) -> None:
|
|
"""Test TTL with custom value per item."""
|
|
with SqliteStore.from_conn_string(temp_db_file) as store:
|
|
store.setup()
|
|
|
|
# Store items with different TTLs
|
|
store.put(("test",), "item1", {"value": "short"}, ttl=1 / 60) # 1 second
|
|
store.put(("test",), "item2", {"value": "long"}, ttl=3 / 60) # 3 seconds
|
|
|
|
# Item with short TTL
|
|
time.sleep(2) # Wait for short TTL
|
|
store.sweep_ttl()
|
|
|
|
# Short TTL item should be gone, long TTL item should remain
|
|
item1 = store.get(("test",), "item1")
|
|
item2 = store.get(("test",), "item2")
|
|
assert item1 is None
|
|
assert item2 is not None
|
|
|
|
# Wait for the second item's TTL
|
|
time.sleep(4)
|
|
store.sweep_ttl()
|
|
|
|
# Now both should be gone
|
|
item2 = store.get(("test",), "item2")
|
|
assert item2 is None
|
|
|
|
|
|
@pytest.mark.flaky(retries=3)
|
|
def test_ttl_override_default(temp_db_file: str) -> None:
|
|
"""Test overriding default TTL at the item level."""
|
|
with SqliteStore.from_conn_string(
|
|
temp_db_file,
|
|
ttl={"default_ttl": 5 / 60}, # 5 seconds default
|
|
) as store:
|
|
store.setup()
|
|
|
|
# Store an item with shorter than default TTL
|
|
store.put(("test",), "item1", {"value": "override"}, ttl=1 / 60) # 1 second
|
|
|
|
# Store an item with default TTL
|
|
store.put(("test",), "item2", {"value": "default"}) # Uses default 5 seconds
|
|
|
|
# Store an item with no TTL
|
|
store.put(("test",), "item3", {"value": "permanent"}, ttl=None)
|
|
|
|
# Wait for the override TTL to expire
|
|
time.sleep(2)
|
|
store.sweep_ttl()
|
|
|
|
# Check results
|
|
item1 = store.get(("test",), "item1")
|
|
item2 = store.get(("test",), "item2")
|
|
item3 = store.get(("test",), "item3")
|
|
|
|
assert item1 is None # Should be expired
|
|
assert item2 is not None # Default TTL, should still be there
|
|
assert item3 is not None # No TTL, should still be there
|
|
|
|
# Wait for default TTL to expire
|
|
time.sleep(4)
|
|
store.sweep_ttl()
|
|
|
|
# Check results again
|
|
item2 = store.get(("test",), "item2")
|
|
item3 = store.get(("test",), "item3")
|
|
|
|
assert item2 is None # Default TTL item should be gone
|
|
assert item3 is not None # No TTL item should still be there
|
|
|
|
|
|
@pytest.mark.flaky(retries=3)
|
|
def test_search_with_ttl(temp_db_file: str) -> None:
|
|
"""Test TTL with search operations."""
|
|
ttl_seconds = 1
|
|
ttl_minutes = ttl_seconds / 60
|
|
|
|
with SqliteStore.from_conn_string(
|
|
temp_db_file, ttl={"default_ttl": ttl_minutes}
|
|
) as store:
|
|
store.setup()
|
|
|
|
# Store items
|
|
store.put(("test",), "item1", {"value": "apple"})
|
|
store.put(("test",), "item2", {"value": "banana"})
|
|
|
|
# Search before expiration
|
|
results = store.search(("test",), filter={"value": "apple"})
|
|
assert len(results) == 1
|
|
assert results[0].key == "item1"
|
|
|
|
# Wait for TTL to expire
|
|
time.sleep(ttl_seconds + 1)
|
|
store.sweep_ttl()
|
|
|
|
# Search after expiration
|
|
results = store.search(("test",), filter={"value": "apple"})
|
|
assert len(results) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_ttl_basic(temp_db_file: str) -> None:
|
|
"""Test basic TTL functionality with asynchronous API."""
|
|
ttl_seconds = 1
|
|
ttl_minutes = ttl_seconds / 60
|
|
|
|
async with AsyncSqliteStore.from_conn_string(
|
|
temp_db_file, ttl={"default_ttl": ttl_minutes}
|
|
) as store:
|
|
await store.setup()
|
|
|
|
# Store an item with TTL
|
|
await store.aput(("test",), "item1", {"value": "test"})
|
|
|
|
# Get the item before expiration
|
|
item = await store.aget(("test",), "item1")
|
|
assert item is not None
|
|
assert item.value["value"] == "test"
|
|
|
|
# Wait for TTL to expire
|
|
await asyncio.sleep(ttl_seconds + 1.0)
|
|
|
|
# Manual sweep needed without the sweeper thread
|
|
await store.sweep_ttl()
|
|
|
|
# Item should be gone now
|
|
item = await store.aget(("test",), "item1")
|
|
assert item is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.flaky(retries=3)
|
|
async def test_async_ttl_refresh(temp_db_file: str) -> None:
|
|
"""Test TTL refresh on read with async API."""
|
|
ttl_seconds = 1
|
|
ttl_minutes = ttl_seconds / 60
|
|
|
|
async with AsyncSqliteStore.from_conn_string(
|
|
temp_db_file, ttl={"default_ttl": ttl_minutes, "refresh_on_read": True}
|
|
) as store:
|
|
await store.setup()
|
|
|
|
# Store an item with TTL
|
|
await store.aput(("test",), "item1", {"value": "test"})
|
|
|
|
# Sleep almost to expiration
|
|
await asyncio.sleep(ttl_seconds - 0.5)
|
|
|
|
# Get the item and refresh TTL
|
|
item = await store.aget(("test",), "item1", refresh_ttl=True)
|
|
assert item is not None
|
|
|
|
# Sleep again - without refresh, would have expired by now
|
|
await asyncio.sleep(ttl_seconds - 0.5)
|
|
|
|
# Get the item, should still be there
|
|
item = await store.aget(("test",), "item1")
|
|
assert item is not None
|
|
assert item.value["value"] == "test"
|
|
|
|
# Sleep again but don't refresh this time
|
|
await asyncio.sleep(ttl_seconds + 1.0)
|
|
|
|
# Manual sweep
|
|
await store.sweep_ttl()
|
|
|
|
# Item should be gone now
|
|
item = await store.aget(("test",), "item1")
|
|
assert item is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_async_ttl_sweeper(temp_db_file: str) -> None:
|
|
"""Test TTL sweeper thread with async API."""
|
|
ttl_seconds = 2
|
|
ttl_minutes = ttl_seconds / 60
|
|
|
|
ttl_config: TTLConfig = {
|
|
"default_ttl": ttl_minutes,
|
|
"sweep_interval_minutes": ttl_minutes / 2,
|
|
}
|
|
|
|
async with AsyncSqliteStore.from_conn_string(
|
|
temp_db_file,
|
|
ttl=ttl_config,
|
|
) as store:
|
|
await store.setup()
|
|
|
|
# Start the TTL sweeper
|
|
await store.start_ttl_sweeper()
|
|
|
|
# Store an item with TTL
|
|
await store.aput(("test",), "item1", {"value": "test"})
|
|
|
|
# Item should be there initially
|
|
item = await store.aget(("test",), "item1")
|
|
assert item is not None
|
|
|
|
# Wait for TTL to expire and the sweeper to run
|
|
await asyncio.sleep(ttl_seconds + (ttl_seconds / 2) + 0.5)
|
|
|
|
# Item should be gone now (swept automatically)
|
|
item = await store.aget(("test",), "item1")
|
|
assert item is None
|
|
|
|
# Stop the sweeper
|
|
await store.stop_ttl_sweeper()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.flaky(retries=3)
|
|
async def test_async_search_with_ttl(temp_db_file: str) -> None:
|
|
"""Test TTL with search operations using async API."""
|
|
ttl_seconds = 1
|
|
ttl_minutes = ttl_seconds / 60
|
|
|
|
async with AsyncSqliteStore.from_conn_string(
|
|
temp_db_file, ttl={"default_ttl": ttl_minutes}
|
|
) as store:
|
|
await store.setup()
|
|
|
|
# Store items
|
|
await store.aput(("test",), "item1", {"value": "apple"})
|
|
await store.aput(("test",), "item2", {"value": "banana"})
|
|
|
|
# Search before expiration
|
|
results = await store.asearch(("test",), filter={"value": "apple"})
|
|
assert len(results) == 1
|
|
assert results[0].key == "item1"
|
|
|
|
# Wait for TTL to expire
|
|
await asyncio.sleep(ttl_seconds + 1)
|
|
await store.sweep_ttl()
|
|
|
|
# Search after expiration
|
|
results = await store.asearch(("test",), filter={"value": "apple"})
|
|
assert len(results) == 0
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.flaky(retries=3)
|
|
async def test_async_asearch_refresh_ttl(temp_db_file: str) -> None:
|
|
"""Test TTL refresh on asearch with async API."""
|
|
ttl_seconds = 4.0 # Increased TTL for less sensitivity to timing
|
|
ttl_minutes = ttl_seconds / 60.0
|
|
|
|
async with AsyncSqliteStore.from_conn_string(
|
|
temp_db_file, ttl={"default_ttl": ttl_minutes, "refresh_on_read": True}
|
|
) as store:
|
|
await store.setup()
|
|
|
|
namespace = ("docs", "user1")
|
|
# t=0: items put, expire at t=4.0s
|
|
await store.aput(namespace, "item1", {"text": "content1", "id": 1})
|
|
await store.aput(namespace, "item2", {"text": "content2", "id": 2})
|
|
|
|
# t=3.0s: (after sleep ttl_seconds * 0.75 = 3s)
|
|
await asyncio.sleep(ttl_seconds * 0.75)
|
|
|
|
# Perform asearch with refresh_ttl=True for item1.
|
|
# item1's TTL should be refreshed. New expiry: t=3.0s + 4.0s = t=7.0s.
|
|
# item2's TTL is not affected. Expires at t=4.0s.
|
|
searched_items = await store.asearch(
|
|
namespace, filter={"id": 1}, refresh_ttl=True
|
|
)
|
|
assert len(searched_items) == 1
|
|
assert searched_items[0].key == "item1"
|
|
|
|
# t=5.0s: (after sleep ttl_seconds * 0.5 = 2s more. Total elapsed: 3s + 2s = 5s)
|
|
await asyncio.sleep(ttl_seconds * 0.5)
|
|
# At this point:
|
|
# - item1 (refreshed by asearch) should expire at t=7.0s. Should be ALIVE.
|
|
# - item2 (original TTL) should have expired at t=4.0s. Should be GONE after sweep.
|
|
|
|
await store.sweep_ttl()
|
|
|
|
# Check item1 (should exist due to asearch refresh)
|
|
item1_check1 = await store.aget(namespace, "item1", refresh_ttl=False)
|
|
assert item1_check1 is not None, (
|
|
"Item1 should exist after asearch refresh and first sweep"
|
|
)
|
|
assert item1_check1.value["text"] == "content1"
|
|
|
|
# Check item2 (should be gone)
|
|
item2_check1 = await store.aget(namespace, "item2", refresh_ttl=False)
|
|
assert item2_check1 is None, (
|
|
"Item2 should be gone after its original TTL expired"
|
|
)
|
|
|
|
# t=7.5s: (after sleep ttl_seconds * 0.625 = 2.5s more. Total elapsed: 5s + 2.5s = 7.5s)
|
|
await asyncio.sleep(ttl_seconds * 0.625)
|
|
# At this point:
|
|
# - item1 (refreshed by asearch, expired at t=7.0s) should be GONE after sweep.
|
|
|
|
await store.sweep_ttl()
|
|
|
|
# Check item1 again (should be gone now)
|
|
item1_final_check = await store.aget(namespace, "item1", refresh_ttl=False)
|
|
assert item1_final_check is None, (
|
|
"Item1 should be gone after its refreshed TTL expired"
|
|
)
|