fix(calendar): keep default creation transactional (#5806)

* fix(calendar): keep default creation transactional

* fix(calendar): serialize default calendar creation

* fix(calendar): handle renamed default id collisions

---------

Co-authored-by: Alexandre Teixeira <111787685+alteixeira20@users.noreply.github.com>
This commit is contained in:
RaresKeY
2026-08-12 04:51:52 +01:00
committed by GitHub
co-authored by Alexandre Teixeira
parent 858c872832
commit 3f9633c44f
3 changed files with 656 additions and 7 deletions
+115 -7
View File
@@ -10,6 +10,7 @@ from typing import Optional, List
from fastapi import APIRouter, HTTPException, Request, UploadFile, File from fastapi import APIRouter, HTTPException, Request, UploadFile, File
from pydantic import BaseModel from pydantic import BaseModel
from sqlalchemy import or_, and_ from sqlalchemy import or_, and_
from sqlalchemy.exc import IntegrityError
from dateutil.rrule import rrulestr from dateutil.rrule import rrulestr
from core.database import SessionLocal, CalendarCal, CalendarDeletedEvent, CalendarEvent from core.database import SessionLocal, CalendarCal, CalendarDeletedEvent, CalendarEvent
@@ -221,22 +222,125 @@ class EventUpdate(BaseModel):
# ── Helpers ── # ── Helpers ──
_DEFAULT_CALENDAR_NAMESPACE = uuid.UUID("4840613a-9847-4a3b-bd75-19e6bc5fc3ce")
def _default_calendar_id(owner: str, collision_index: int = 0) -> str:
"""Return one stable primary-key candidate for an owner's lazy default.
Slot zero preserves the original owner-derived identifier. Later slots
let a username be reused after its prior calendar was migrated to another
owner during a rename, without making concurrent first use choose random
and therefore divergent identifiers.
"""
if collision_index == 0:
candidate_name = owner
else:
candidate_name = json.dumps(
[owner, collision_index],
ensure_ascii=False,
separators=(",", ":"),
)
return str(uuid.uuid5(_DEFAULT_CALENDAR_NAMESPACE, candidate_name))
def _begin_sqlite_default_write(db) -> None:
"""Serialize an absent-default check with other SQLite writers.
SQLite's default deferred transactions allow two workers to both read an
empty calendar set before either writes. ``BEGIN IMMEDIATE`` acquires the
writer reservation before the second, authoritative lookup. We issue it
only when the driver has not already opened a write transaction; a caller
with a pending write already owns the required reservation.
"""
connection = db.connection()
dbapi_connection = connection.connection
driver_connection = getattr(
dbapi_connection,
"driver_connection",
dbapi_connection,
)
if not getattr(driver_connection, "in_transaction", False):
connection.exec_driver_sql("BEGIN IMMEDIATE")
def _ensure_default_calendar(db, owner: str = None) -> CalendarCal: def _ensure_default_calendar(db, owner: str = None) -> CalendarCal:
"""Create default calendar if none exist for this owner.""" """Return the owner's calendar, staging a default in the caller's transaction.
A stable owner-derived primary key makes concurrent first-use inserts
converge on one row on every SQL backend. SQLite additionally serializes
the absent-row check because its deferred transactions otherwise permit
both workers to read the gap before either writes. Other backends recover
a lost insert race inside a savepoint so the caller's event transaction
remains usable and atomic.
"""
owner = owner or FALLBACK_OWNER owner = owner or FALLBACK_OWNER
cal = db.query(CalendarCal).filter(CalendarCal.owner == owner).first() cal = db.query(CalendarCal).filter(CalendarCal.owner == owner).first()
if not cal: if cal:
return cal
dialect = db.get_bind().dialect.name
if dialect == "sqlite":
_begin_sqlite_default_write(db)
# Another worker may have committed while BEGIN IMMEDIATE waited.
cal = db.query(CalendarCal).filter(CalendarCal.owner == owner).first()
if cal:
return cal
collision_index = 0
while True:
default_id = _default_calendar_id(owner, collision_index)
if dialect == "sqlite":
# BEGIN IMMEDIATE above makes this occupancy check authoritative:
# another SQLite writer cannot rename, delete, or claim this slot
# until the caller commits or rolls back.
occupant = db.query(CalendarCal).filter(
CalendarCal.id == default_id,
).first()
if occupant is not None:
if occupant.owner == owner:
return occupant
collision_index += 1
continue
cal = CalendarCal( cal = CalendarCal(
id=str(uuid.uuid4()), id=default_id,
owner=owner, owner=owner,
name="Personal", name="Personal",
color="#5b8abf", color="#5b8abf",
source="local", source="local",
) )
db.add(cal)
db.commit() if dialect == "sqlite":
db.refresh(cal) db.add(cal)
return cal db.flush()
return cal
try:
# A uniqueness failure rolls back only this savepoint, not an event
# or reminder already staged by the caller's outer transaction.
with db.begin_nested():
db.add(cal)
db.flush()
return cal
except IntegrityError:
# Use a locking/current read so repeatable-read backends can observe
# the row that won after our transaction's original empty snapshot.
occupant = db.query(CalendarCal).filter(
CalendarCal.id == default_id,
).with_for_update().first()
if occupant is None:
# Do not misclassify an unrelated integrity failure as an ID
# collision and loop forever. A concurrently deleted winner is
# safe for the caller to retry as a fresh transaction.
raise
if occupant.owner == owner:
return occupant
# A renamed calendar owns this deterministic slot. Advance to the
# next stable slot; concurrent callers for this owner will still
# converge there.
collision_index += 1
# Per-request user time context. chat_routes sets this from browser timezone # Per-request user time context. chat_routes sets this from browser timezone
@@ -1015,6 +1119,9 @@ def setup_calendar_routes(upload_handler=None) -> APIRouter:
db = SessionLocal() db = SessionLocal()
try: try:
_ensure_default_calendar(db, owner) _ensure_default_calendar(db, owner)
# Listing calendars intentionally lazily creates a durable default.
# Other callers commit it with the event they are creating.
db.commit()
cals = db.query(CalendarCal).filter(CalendarCal.owner == owner).all() cals = db.query(CalendarCal).filter(CalendarCal.owner == owner).all()
return {"calendars": [ return {"calendars": [
{"name": c.name, "href": c.id, "color": c.color, "source": c.source} {"name": c.name, "href": c.id, "color": c.color, "source": c.source}
@@ -1023,6 +1130,7 @@ def setup_calendar_routes(upload_handler=None) -> APIRouter:
except HTTPException: except HTTPException:
raise raise
except Exception as e: except Exception as e:
db.rollback()
logger.error("Failed to list calendars: %s", e) logger.error("Failed to list calendars: %s", e)
raise HTTPException(500, "Failed to list calendars") raise HTTPException(500, "Failed to list calendars")
finally: finally:
+3
View File
@@ -196,6 +196,9 @@ async def do_manage_calendar(content: str, owner: Optional[str] = None) -> Dict:
try: try:
if action == "list_calendars": if action == "list_calendars":
_ensure_default_calendar(db, owner) _ensure_default_calendar(db, owner)
# This read path intentionally persists the lazily-created default;
# event creation commits it in the event's transaction instead.
db.commit()
cals = _calendar_query().all() cals = _calendar_query().all()
result = [{"name": c.name, "href": c.id} for c in cals] result = [{"name": c.name, "href": c.id} for c in cals]
if result: if result:
+538
View File
@@ -0,0 +1,538 @@
"""Default calendar creation belongs to the caller's transaction.
Before this regression, ``_ensure_default_calendar`` committed independently.
If event persistence then failed, the event rolled back but a new ``Personal``
calendar remained (``calendar_count=1``, ``event_count=0``).
"""
import json
import threading
from contextlib import contextmanager
from datetime import datetime, timedelta
from types import SimpleNamespace
import pytest
from fastapi import HTTPException
from sqlalchemy import create_engine, event
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session, sessionmaker
from sqlalchemy.pool import NullPool
from tests.helpers.import_state import clear_fake_database_modules
clear_fake_database_modules()
import core.database as cdb # noqa: E402
import routes.calendar_routes as calendar_routes # noqa: E402
from core.database import CalendarCal, CalendarEvent # noqa: E402
from routes.calendar_routes import EventCreate # noqa: E402
from routes.calendar_routes import ( # noqa: E402
_default_calendar_id,
_ensure_default_calendar,
)
class _RejectEventCommit(Session):
"""Reproduce an event commit failure after default-calendar creation."""
def commit(self):
if any(isinstance(row, CalendarEvent) for row in self.new):
raise RuntimeError("commit guard rejected event commit")
return super().commit()
@pytest.fixture
def session_factory(tmp_path, monkeypatch):
engine = create_engine(
f"sqlite:///{tmp_path / 'calendar.db'}",
connect_args={"check_same_thread": False},
poolclass=NullPool,
)
cdb.Base.metadata.create_all(engine)
factory = sessionmaker(
bind=engine,
autoflush=False,
autocommit=False,
class_=_RejectEventCommit,
)
monkeypatch.setattr(cdb, "SessionLocal", factory)
monkeypatch.setattr(calendar_routes, "SessionLocal", factory)
try:
yield factory
finally:
engine.dispose()
def _request():
return SimpleNamespace(state=SimpleNamespace(current_user="alice"))
def _endpoint(method, suffix):
router = calendar_routes.setup_calendar_routes()
for route in router.routes:
if route.path.endswith(suffix) and method in route.methods:
return route.endpoint
raise RuntimeError(f"{method} *{suffix} not found")
def _counts(factory):
db = factory()
try:
return db.query(CalendarCal).count(), db.query(CalendarEvent).count()
finally:
db.close()
async def test_route_event_failure_rolls_back_new_default_calendar(session_factory):
create_event = _endpoint("POST", "/events")
with pytest.raises(HTTPException) as caught:
await create_event(
_request(),
EventCreate(summary="Planning", dtstart="2126-07-20T09:00:00Z"),
)
assert caught.value.status_code == 500
assert _counts(session_factory) == (0, 0)
async def test_route_event_validation_failure_rolls_back_new_default_calendar(
session_factory,
):
create_event = _endpoint("POST", "/events")
with pytest.raises(HTTPException) as caught:
await create_event(
_request(),
EventCreate(summary="Planning", dtstart="not-a-datetime"),
)
assert caught.value.status_code == 500
assert _counts(session_factory) == (0, 0)
async def test_tool_event_failure_rolls_back_new_default_calendar(session_factory):
from src.tools.calendar import do_manage_calendar
result = await do_manage_calendar(
json.dumps({
"action": "create_event",
"summary": "Planning",
"dtstart": "2126-07-20T09:00:00Z",
}),
owner="alice",
)
assert result["exit_code"] == 1
assert "commit guard rejected event commit" in result["error"]
assert _counts(session_factory) == (0, 0)
async def test_tool_event_validation_failure_rolls_back_new_default_calendar(
session_factory,
):
from src.tools.calendar import do_manage_calendar
result = await do_manage_calendar(
json.dumps({
"action": "create_event",
"summary": "Planning",
"dtstart": "not-a-datetime",
}),
owner="alice",
)
assert result["exit_code"] == 1
assert "Could not parse dtstart" in result["error"]
assert _counts(session_factory) == (0, 0)
async def test_route_list_calendars_persists_lazy_default(session_factory):
list_calendars = _endpoint("GET", "/calendars")
result = await list_calendars(_request())
assert [calendar["name"] for calendar in result["calendars"]] == ["Personal"]
assert _counts(session_factory) == (1, 0)
async def test_tool_list_calendars_persists_lazy_default(session_factory):
from src.tools.calendar import do_manage_calendar
result = await do_manage_calendar(
json.dumps({"action": "list_calendars"}),
owner="alice",
)
assert result["exit_code"] == 0
assert [calendar["name"] for calendar in result["calendars"]] == ["Personal"]
assert _counts(session_factory) == (1, 0)
def test_repeated_rename_and_reuse_uses_stable_fallback_ids(tmp_path):
engine = create_engine(
f"sqlite:///{tmp_path / 'renamed-calendar.db'}",
connect_args={"check_same_thread": False},
poolclass=NullPool,
)
cdb.Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
db = factory()
try:
first = _ensure_default_calendar(db, "alice")
assert first.id == _default_calendar_id("alice")
db.commit()
# The supported user-rename migration changes owner columns while
# deliberately preserving durable row identifiers.
first.owner = "bob"
db.commit()
second = _ensure_default_calendar(db, "alice")
assert second.id == _default_calendar_id("alice", 1)
db.commit()
# Repeating the same lifecycle must advance deterministically instead
# of failing or choosing a random identifier.
second.owner = "carol"
db.commit()
third = _ensure_default_calendar(db, "alice")
assert third.id == _default_calendar_id("alice", 2)
db.commit()
rows = db.query(CalendarCal).order_by(CalendarCal.owner).all()
assert [(row.owner, row.id) for row in rows] == [
("alice", _default_calendar_id("alice", 2)),
("bob", _default_calendar_id("alice")),
("carol", _default_calendar_id("alice", 1)),
]
finally:
db.close()
engine.dispose()
def _assert_concurrent_first_use(tmp_path, occupied_owner=None):
engine = create_engine(
f"sqlite:///{tmp_path / 'concurrent-calendar.db'}",
connect_args={"check_same_thread": False, "timeout": 10},
poolclass=NullPool,
)
cdb.Base.metadata.create_all(engine)
factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
expected_collision_index = 0
if occupied_owner is not None:
seed = factory()
try:
seed.add(CalendarCal(
id=_default_calendar_id("alice"),
owner=occupied_owner,
name="Personal",
source="local",
))
seed.commit()
expected_collision_index = 1
finally:
seed.close()
first_staged = threading.Event()
second_selected = threading.Event()
errors = []
@event.listens_for(engine, "after_cursor_execute")
def observe_second_gap(conn, cursor, statement, parameters, context, executemany):
if (
threading.current_thread().name == "calendar-worker-second"
and statement.lstrip().upper().startswith("SELECT")
and "FROM calendars" in statement
):
second_selected.set()
def create_default(worker, hold=False):
db = factory()
try:
if not hold:
assert first_staged.wait(5)
cal = _ensure_default_calendar(db, "alice")
start = datetime(2126, 7, 20, 9 if hold else 10)
db.add(CalendarEvent(
uid=worker,
calendar_id=cal.id,
summary=f"Event {worker}",
dtstart=start,
dtend=start + timedelta(hours=1),
))
if hold:
first_staged.set()
# The second session has observed the uncommitted gap before
# this transaction releases its writer reservation.
assert second_selected.wait(5)
db.commit()
assert cal.id == _default_calendar_id("alice", expected_collision_index)
except BaseException as exc: # pragma: no cover - asserted below
errors.append((worker, exc))
db.rollback()
finally:
db.close()
first = threading.Thread(
target=create_default,
args=("first", True),
name="calendar-worker-first",
)
second = threading.Thread(
target=create_default,
args=("second",),
name="calendar-worker-second",
)
first.start()
second.start()
first.join(10)
second.join(10)
try:
assert not first.is_alive() and not second.is_alive()
assert errors == []
db = factory()
try:
rows = db.query(CalendarCal).filter(CalendarCal.owner == "alice").all()
assert [(row.id, row.name) for row in rows] == [
(_default_calendar_id("alice", expected_collision_index), "Personal")
]
assert db.query(CalendarEvent).count() == 2
if occupied_owner is not None:
occupied = db.query(CalendarCal).filter(
CalendarCal.id == _default_calendar_id("alice"),
).one()
assert occupied.owner == occupied_owner
finally:
db.close()
finally:
engine.dispose()
def test_concurrent_first_use_creates_one_sqlite_default(tmp_path):
_assert_concurrent_first_use(tmp_path)
def test_concurrent_first_use_after_rename_creates_one_fallback_default(tmp_path):
_assert_concurrent_first_use(tmp_path, occupied_owner="bob")
def test_sqlite_default_stays_in_callers_transaction(session_factory):
db = session_factory()
try:
cal = _ensure_default_calendar(db, "rollback-owner")
assert cal.id == _default_calendar_id("rollback-owner")
db.rollback()
finally:
db.close()
verify = session_factory()
try:
assert (
verify.query(CalendarCal)
.filter(CalendarCal.owner == "rollback-owner")
.count()
== 0
)
finally:
verify.close()
def test_sqlite_fallback_default_stays_in_callers_transaction(session_factory):
seed = session_factory()
try:
seed.add(CalendarCal(
id=_default_calendar_id("alice"),
owner="bob",
name="Personal",
source="local",
))
seed.commit()
finally:
seed.close()
db = session_factory()
try:
cal = _ensure_default_calendar(db, "alice")
assert cal.id == _default_calendar_id("alice", 1)
db.rollback()
finally:
db.close()
verify = session_factory()
try:
assert verify.query(CalendarCal).filter(CalendarCal.owner == "alice").count() == 0
assert verify.query(CalendarCal).filter(CalendarCal.owner == "bob").count() == 1
finally:
verify.close()
class _FakeDialect:
name = "postgresql"
class _FakeBind:
dialect = _FakeDialect()
class _FakeQuery:
def __init__(self, session):
self.session = session
def filter(self, *conditions):
return self
def with_for_update(self):
self.session.locking_read = True
return self
def first(self):
self.session.query_count += 1
if self.session.query_count == 1:
return None
return self.session.winner
class _GenericRaceSession:
"""Minimal non-SQLite session that loses the deterministic-ID race."""
def __init__(self):
self.query_count = 0
self.nested_entries = 0
self.locking_read = False
self.candidate = None
self.winner = CalendarCal(
id=_default_calendar_id("alice"),
owner="alice",
name="Personal",
source="local",
)
def get_bind(self):
return _FakeBind()
def query(self, model):
assert model is CalendarCal
return _FakeQuery(self)
@contextmanager
def begin_nested(self):
self.nested_entries += 1
yield
def add(self, row):
self.candidate = row
def flush(self):
raise IntegrityError("insert", {}, RuntimeError("duplicate primary key"))
def test_generic_backend_lost_race_recovers_inside_savepoint():
db = _GenericRaceSession()
winner = _ensure_default_calendar(db, "alice")
assert winner is db.winner
assert db.nested_entries == 1
assert db.locking_read is True
assert db.candidate.id == db.winner.id
def test_generic_backend_unattributed_integrity_error_is_not_retried():
db = _GenericRaceSession()
db.winner = None
with pytest.raises(IntegrityError):
_ensure_default_calendar(db, "alice")
assert db.nested_entries == 1
class _GenericRenamedSlotSession(_GenericRaceSession):
"""A different owner occupies slot zero; slot one remains available."""
def __init__(self):
super().__init__()
self.candidates = []
self.winner = CalendarCal(
id=_default_calendar_id("alice"),
owner="bob",
name="Personal",
source="local",
)
def add(self, row):
self.candidate = row
self.candidates.append(row)
def flush(self):
if len(self.candidates) == 1:
raise IntegrityError("insert", {}, RuntimeError("duplicate primary key"))
def test_generic_backend_renamed_slot_advances_inside_savepoint():
db = _GenericRenamedSlotSession()
fallback = _ensure_default_calendar(db, "alice")
assert fallback is db.candidates[-1]
assert fallback.id == _default_calendar_id("alice", 1)
assert fallback.owner == "alice"
assert db.nested_entries == 2
assert db.locking_read is True
assert db.winner.owner == "bob"
def test_generic_backend_fallback_keeps_outer_transaction_usable(tmp_path):
engine = create_engine(
f"sqlite:///{tmp_path / 'generic-savepoint-calendar.db'}",
poolclass=NullPool,
)
cdb.Base.metadata.create_all(engine)
# SQLite supplies a lightweight local SQL executor here; changing only the
# dispatch name exercises the real Session/savepoint branch used by
# PostgreSQL-style backends without pretending to validate their dialect.
engine.dialect.name = "postgresql"
factory = sessionmaker(bind=engine, autoflush=False, autocommit=False)
seed = factory()
try:
seed.add(CalendarCal(
id=_default_calendar_id("alice"),
owner="bob",
name="Personal",
source="local",
))
seed.commit()
finally:
seed.close()
db = factory()
try:
cal = _ensure_default_calendar(db, "alice")
start = datetime(2126, 7, 20, 9)
db.add(CalendarEvent(
uid="after-fallback",
calendar_id=cal.id,
summary="Atomic",
dtstart=start,
dtend=start + timedelta(hours=1),
))
db.commit()
finally:
db.close()
verify = factory()
try:
assert [
(row.owner, row.id)
for row in verify.query(CalendarCal).order_by(CalendarCal.owner).all()
] == [
("alice", _default_calendar_id("alice", 1)),
("bob", _default_calendar_id("alice")),
]
assert verify.query(CalendarEvent).count() == 1
finally:
verify.close()
engine.dispose()