A push that loses a create race applies its change instead of failing

Two devices pushing the same new note or tag at the same moment both found
no row and both inserted it; the slower insert hit the primary key and the
whole push answered 500. Each pushed note or label change now runs in a
savepoint. If its insert loses the race, only that change is rolled back,
and it is applied again against the row the other push committed.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

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:

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 -----------------------------------------------