diff --git a/src/inkwell/sync.py b/src/inkwell/sync.py index 3c7516d..fd76235 100644 --- a/src/inkwell/sync.py +++ b/src/inkwell/sync.py @@ -19,6 +19,7 @@ from datetime import datetime, timezone from quart import Blueprint, g, jsonify, request from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from .auth import login_required from .colors import normalize_color @@ -446,6 +447,27 @@ async def _apply_child_delete(db, ch: dict, removed_files: list[str]) -> dict: return _result(str(child_id), entity, "applied") +async def _apply_raced(db, apply, previews: list[tuple[uuid.UUID, str]]) -> dict: + """Apply one note or label change in a savepoint, and once more if its insert lost + a race. + + Two devices can push the same new note or tag at the same moment, or one device + can retry a push whose reply it never got. Both requests look the id up, find + nothing, and insert, and the slower insert hits the primary key (or a live tag's + name). Rolling back to the savepoint drops only that one change, along with any + link previews it queued. Applying it again finds the row the other push committed, + so the change goes through the usual last-write-wins path instead of failing the + whole batch with a 500. + """ + queued = len(previews) + try: + async with db.begin_nested(): + return await apply() + except IntegrityError: + del previews[queued:] + return await apply() + + @bp.post("/push") @login_required async def push(): @@ -468,9 +490,9 @@ async def push(): continue entity = ch.get("entity") if entity == "note": - results.append(await _apply_note(db, ch, previews)) + results.append(await _apply_raced(db, lambda: _apply_note(db, ch, previews), previews)) elif entity == "label": - results.append(await _apply_label(db, ch)) + results.append(await _apply_raced(db, lambda: _apply_label(db, ch), previews)) elif entity in _CHILD_ENTITIES: results.append(await _apply_child_delete(db, ch, removed_files)) else: diff --git a/tests/test_integration.py b/tests/test_integration.py index 03b6d55..282c300 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -29,7 +29,7 @@ from inkwell.models.invite import Invite from inkwell.models.password_reset import PasswordReset from inkwell.models.settings import Setting from inkwell.models.share import Share -from inkwell.models.label import NoteLabel +from inkwell.models.label import Label, NoteLabel from inkwell.models.note import Note from inkwell.models.note_attachment import NoteAttachment from inkwell.models.user import User @@ -805,6 +805,54 @@ async def test_a_pushed_edit_keeps_the_clients_edit_time_when_a_tag_is_lifted(ap assert datetime.fromisoformat(note["updated_at"]) == datetime(2026, 1, 1, tzinfo=timezone.utc) +@pytest.mark.parametrize("entity", ["label", "note"]) +async def test_a_push_that_loses_a_create_race_applies_the_change_instead_of_failing(app_client, db, entity): + """Two devices push the same new tag or note at once. The other push's row is + held here, inserted but not committed, so this push finds no row, inserts, and + waits on it. Once it commits, this push's insert fails on the primary key. That + change has to be applied to the stored row, not fail the batch with a 500.""" + await _signed_in(app_client, f"race-{entity}") + owner_id = await db.scalar(select(User.id).where(User.email == f"race-{entity}@example.test")) + rid = uuid.uuid4() + first = datetime(2026, 1, 1, tzinfo=timezone.utc) + if entity == "label": + theirs = Label(id=rid, owner_id=owner_id, name="groceries", updated_at=first) + change = {"name": "groceries", "color": "teal"} + else: + theirs = Note(id=rid, owner_id=owner_id, body="pushed first", display_title="pushed first", updated_at=first) + change = {"body": "pushed second", "created_at": "2026-01-01T00:00:00Z"} + change |= {"entity": entity, "id": str(rid), "op": "upsert", "edited_at": "2026-01-02T00:00:00Z"} + + async with session_scope() as other: + other.add(theirs) + await other.flush() + push = asyncio.create_task(app_client.post("/api/sync/push", json={"changes": [change]})) + # Commit only once the push's insert is blocked on that row. Committing + # earlier would let it find the row and never race at all. A fresh session per + # look: pg_stat_activity is read once per transaction and then held. + for _ in range(100): + async with session_scope() as probe: + waiting = await probe.scalar(text( + "SELECT count(*) FROM pg_stat_activity" + " WHERE datname = current_database() AND wait_event_type = 'Lock'" + )) + if waiting or push.done(): + break + await asyncio.sleep(0.1) + assert waiting and not push.done(), "the push should be waiting on the other push's uncommitted row" + await other.commit() + resp = await asyncio.wait_for(push, timeout=10) + + assert resp.status_code == 200 + assert (await resp.get_json())["results"][0]["status"] == "applied" + model = Label if entity == "label" else Note + stored = await db.scalar(select(model).where(model.id == rid).execution_options(populate_existing=True)) + if entity == "label": + assert stored.color == "teal", "the later edit won" + else: + assert stored.body == "pushed second", "the later edit won" + + # --- attachments as a sync entity -----------------------------------------------