test-int/test_note_materialization_lock_order.py
Rama main · réplica de lectura del repositorio original.
"""Postgres coverage that materialization publish cannot anchor a row-lock cycle."""
from __future__ import annotations
import asyncio
from contextlib import suppress
from datetime import UTC, datetime
from typing import Any, override
import pytest
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import AsyncSession
from basic_memory import db
from basic_memory.indexing.note_materialization_runner import (
NoteMaterializationSessionLock,
RepositoryNoteMaterializationPublisher,
)
from basic_memory.indexing.accepted_note_write_runner import (
lock_accepted_note_content_for_entity_mutation,
)
from basic_memory.models import Entity, NoteContent, Project
from basic_memory.repository.note_content_repository import NoteContentRepository
from basic_memory.runtime.note_content import (
RuntimeNoteMaterializationJobRequest,
RuntimeNoteMaterializationStatus,
)
from basic_memory.runtime.note_materialization import (
RuntimeWrittenFileState,
plan_prepared_note_write,
)
class StartedMaterializationLock(NoteMaterializationSessionLock):
"""Expose that the publisher entered its transaction without adding a DB lock."""
def __init__(self) -> None:
self.started = asyncio.Event()
@override
async def lock_note_materialization(
self,
session: AsyncSession,
*,
project_id: int,
entity_id: int,
) -> None:
self.started.set()
@pytest.mark.asyncio
@pytest.mark.parametrize("move_during_wait", [False, True], ids=["stable-path", "moved-path"])
async def test_materialization_and_accepted_mutation_share_note_content_first_order(
engine_factory,
test_project: Project,
move_during_wait: bool,
) -> None:
"""An accepted mutation can take Entity while materialization waits on NoteContent."""
engine, session_maker = engine_factory
if engine.dialect.name != "postgresql":
pytest.skip("row-lock ordering requires PostgreSQL")
file_path = "notes/lock-order.md"
markdown = "# Lock order\n"
db_checksum = "accepted-checksum"
async with db.scoped_session(session_maker) as session:
entity = Entity(
project_id=test_project.id,
title="Lock order",
note_type="note",
content_type="text/markdown",
file_path=file_path,
checksum="previous-file-checksum",
)
session.add(entity)
await session.flush()
entity_id = entity.id
await NoteContentRepository(project_id=test_project.id).create(
session,
NoteContent(
entity_id=entity_id,
markdown_content=markdown,
db_version=1,
db_checksum=db_checksum,
file_version=None,
file_checksum="previous-file-checksum",
file_write_status="writing",
),
)
request = RuntimeNoteMaterializationJobRequest(
project_id=test_project.id,
entity_id=entity_id,
db_version=1,
db_checksum=db_checksum,
source="api",
)
attempted_at = datetime(2026, 8, 5, 1, 0, tzinfo=UTC)
prepared_write = plan_prepared_note_write(
request=request,
file_path=file_path,
markdown_content=markdown,
previous_file_checksum="previous-file-checksum",
previous_sync_checksum=None,
attempted_at=attempted_at,
)
written_file = RuntimeWrittenFileState(
file_path=file_path,
file_checksum="materialized-checksum",
file_updated_at=datetime(2026, 8, 5, 1, 1, tzinfo=UTC),
storage_checksum="materialized-etag",
)
session_lock = StartedMaterializationLock()
publisher = RepositoryNoteMaterializationPublisher(
session_maker=session_maker,
session_lock=session_lock,
)
publish_task: asyncio.Task[Any] | None = None
async with session_maker() as mutation_session:
await mutation_session.begin()
try:
await lock_accepted_note_content_for_entity_mutation(
mutation_session,
project_id=test_project.id,
entity_id=entity_id,
)
publish_task = asyncio.create_task(
publisher.publish_written_file_state(
request,
prepared_write,
written_file,
)
)
await asyncio.wait_for(session_lock.started.wait(), timeout=2)
# The publisher holds no row locks while it waits at its NoteContent
# CAS, so this Entity lock probe cannot close a lock cycle. Keep the
# timeout as a loud regression if publish-span read locks return.
await asyncio.sleep(0.1)
locked_entity = await asyncio.wait_for(
mutation_session.scalar(
select(Entity).where(Entity.id == entity_id).with_for_update()
),
timeout=2,
)
assert locked_entity is not None
if move_during_wait:
locked_entity.file_path = "notes/moved-during-materialization.md"
await mutation_session.execute(
update(NoteContent)
.where(NoteContent.entity_id == entity_id)
.values(file_path=locked_entity.file_path, db_version=2)
)
else:
locked_entity.title = "Accepted mutation completed"
await mutation_session.execute(
update(NoteContent)
.where(NoteContent.entity_id == entity_id)
.values(last_materialization_error="accepted mutation overlap")
)
await mutation_session.commit()
result = await asyncio.wait_for(publish_task, timeout=2)
finally:
if mutation_session.in_transaction():
await mutation_session.rollback()
if publish_task is not None and not publish_task.done():
publish_task.cancel()
with suppress(asyncio.CancelledError):
await publish_task
async with session_maker() as verification_session:
note_content = await verification_session.get(NoteContent, entity_id)
entity = await verification_session.get(Entity, entity_id)
assert note_content is not None
assert entity is not None
if move_during_wait:
assert result.status is RuntimeNoteMaterializationStatus.stale
assert result.written_file_orphaned
assert note_content.file_path == "notes/moved-during-materialization.md"
assert note_content.db_version == 2
assert note_content.file_version is None
assert note_content.file_checksum == "previous-file-checksum"
assert entity.file_path == "notes/moved-during-materialization.md"
assert entity.mtime != written_file.file_updated_at.timestamp()
else:
assert result.status is RuntimeNoteMaterializationStatus.written
assert entity.title == "Accepted mutation completed"
assert note_content.file_version == 1
assert note_content.file_checksum == "materialized-checksum"
assert note_content.file_write_status == "synced"
assert entity.mtime == written_file.file_updated_at.timestamp()
assert entity.size == len(markdown.encode("utf-8"))
@pytest.mark.asyncio
async def test_publish_cas_loss_never_reverts_newer_accepted_write(
engine_factory,
test_project: Project,
) -> None:
"""A publisher blocked at CAS must preserve a newer same-path accepted write."""
engine, session_maker = engine_factory
if engine.dialect.name != "postgresql":
pytest.skip("row-lock ordering requires PostgreSQL")
file_path = "notes/cas-loss.md"
original_markdown = "# Original\n"
async with db.scoped_session(session_maker) as session:
entity = Entity(
project_id=test_project.id,
title="CAS loss",
note_type="note",
content_type="text/markdown",
file_path=file_path,
checksum="previous-file-checksum",
)
session.add(entity)
await session.flush()
entity_id = entity.id
original_mtime = entity.mtime
original_size = entity.size
await NoteContentRepository(project_id=test_project.id).create(
session,
NoteContent(
entity_id=entity_id,
markdown_content=original_markdown,
db_version=1,
db_checksum="original-db-checksum",
file_version=None,
file_checksum="previous-file-checksum",
file_write_status="writing",
),
)
request = RuntimeNoteMaterializationJobRequest(
project_id=test_project.id,
entity_id=entity_id,
db_version=1,
db_checksum="original-db-checksum",
source="api",
)
prepared_write = plan_prepared_note_write(
request=request,
file_path=file_path,
markdown_content=original_markdown,
previous_file_checksum="previous-file-checksum",
previous_sync_checksum=None,
attempted_at=datetime(2026, 8, 5, 2, 0, tzinfo=UTC),
)
written_file = RuntimeWrittenFileState(
file_path=file_path,
file_checksum="stale-materialized-checksum",
file_updated_at=datetime(2026, 8, 5, 2, 1, tzinfo=UTC),
storage_checksum="stale-materialized-etag",
)
session_lock = StartedMaterializationLock()
publisher = RepositoryNoteMaterializationPublisher(
session_maker=session_maker,
session_lock=session_lock,
)
publish_task: asyncio.Task[Any] | None = None
async with session_maker() as mutation_session:
await mutation_session.begin()
try:
await lock_accepted_note_content_for_entity_mutation(
mutation_session,
project_id=test_project.id,
entity_id=entity_id,
)
publish_task = asyncio.create_task(
publisher.publish_written_file_state(
request,
prepared_write,
written_file,
)
)
await asyncio.wait_for(session_lock.started.wait(), timeout=2)
await asyncio.sleep(0.1)
assert not publish_task.done()
await mutation_session.execute(
update(NoteContent)
.where(NoteContent.entity_id == entity_id)
.values(
db_version=2,
db_checksum="newer-db-checksum",
markdown_content="# Newer accepted write\n",
file_write_status="pending",
)
)
await mutation_session.commit()
result = await asyncio.wait_for(publish_task, timeout=2)
finally:
if mutation_session.in_transaction():
await mutation_session.rollback()
if publish_task is not None and not publish_task.done():
publish_task.cancel()
with suppress(asyncio.CancelledError):
await publish_task
async with session_maker() as verification_session:
note_content = await verification_session.get(NoteContent, entity_id)
entity = await verification_session.get(Entity, entity_id)
assert result.status is RuntimeNoteMaterializationStatus.stale
assert result.written_file_orphaned is False
assert note_content is not None
assert note_content.db_version == 2
assert note_content.db_checksum == "newer-db-checksum"
assert note_content.markdown_content == "# Newer accepted write\n"
assert note_content.file_version is None
assert note_content.file_checksum == "previous-file-checksum"
assert note_content.file_write_status == "pending"
assert entity is not None
assert entity.mtime == original_mtime
assert entity.size == original_size