fix(personhog): serialize splits of one person on its lifecycle mark

rust/personhog-replica/src/service/mod.rs

@@ -1495,9 +1495,6 @@ impl PersonHogReplica for PersonHogReplicaService {
             .await
             .map_err(|e| match &e {
                 storage::StorageError::NotFound(msg) => Status::not_found(msg.clone()),
-                storage::StorageError::FailedPrecondition(msg) => {
-                    Status::failed_precondition(msg.clone())
-                }
                 _ => log_and_convert_error(e, "split_person"),
             })?;
 

rust/personhog-replica/src/storage/postgres/person.rs

@@ -982,6 +982,14 @@ impl PersonLookup for PostgresStorage {
         // is constant (not per-distinct_id), so the lock window stays short; the
         // service-layer cap bounds the number of rows locked.
         let mut tx = self.bulk_primary_pool.begin().await?;
+        // Fail fast behind an in-flight merge, delete or split; the bound is per wait, not per call.
+        sqlx::query("SET LOCAL lock_timeout = '2s'")
+            .execute(&mut *tx)
+            .await?;
+
+        // The source's mark serializes splits and merges of this person, so two splits cannot both pass the
+        // keep-one check below. Claim it before any row lock; the manual detach_distinct_id job takes none.
+        let mark_op_id = claim_split_mark(&mut tx, team_id, person_id).await?;
 
         // No FOR UPDATE on the source person: deletes lock PDI rows before person
         // rows, so locking the person first here would invert that order and risk
@@ -1033,6 +1041,27 @@ impl PersonLookup for PostgresStorage {
             )));
         }
 
+        // A fresh statement after the claim, so it sees every split that committed while this one waited.
+        // An id derived to the source's own uuid counts as leaving.
+        let keeps_a_distinct_id: bool = sqlx::query_scalar!(
+            r#"
+            SELECT EXISTS (
+                SELECT 1 FROM posthog_persondistinctid
+                WHERE team_id = $1 AND person_id = $2 AND is_deleted = false AND NOT (distinct_id = ANY($3))
+            ) as "exists!"
+            "#,
+            team_id as i32,
+            person_id,
+            distinct_ids_to_split
+        )
+        .fetch_one(&mut *tx)
+        .await?;
+        if !keeps_a_distinct_id {
+            return Err(StorageError::FailedPrecondition(format!(
+                "the split would leave person_id={person_id} (team_id={team_id}) with no live distinct id"
+            )));
+        }
+
         let new_person_version = person_version + SPLIT_VERSION_OFFSET;
         // Build all arrays in request order so the response order matches the
         // input order (part of the RPC contract).
@@ -1054,12 +1083,13 @@ impl PersonLookup for PostgresStorage {
         // re-split, or a tombstone left by a merge or delete of the person
         // that once owned the distinct id. No ON CONFLICT — the partitioned
         // table has a unique index, not a unique constraint, so ON CONFLICT
-        // inference doesn't work.
+        // inference doesn't work. Lock in id order, as the tombstone RPC does, so the two cannot deadlock.
         let existing_persons = sqlx::query!(
             r#"
             SELECT id::bigint as "id!", uuid as "uuid!", created_at as "created_at!"
             FROM posthog_person
             WHERE team_id = $1 AND uuid = ANY($2)
+            ORDER BY id
             FOR UPDATE
             "#,
             team_id as i32,
@@ -1165,6 +1195,7 @@ impl PersonLookup for PostgresStorage {
             )));
         }
 
+        release_lifecycle_marks(&mut tx, team_id, mark_op_id, "merge").await?;
         tx.commit().await?;
 
         let results: Vec<SplitResult> = dids
@@ -1618,7 +1649,7 @@ async fn tombstone_persons_by_uuids(
     tombstones.extend(with_tombstoned_distinct_ids(&mut tx, team_id, already).await?);
 
     if let Some(op_id) = mark_op_id {
-        release_delete_marks(&mut tx, team_id, op_id).await?;
+        release_lifecycle_marks(&mut tx, team_id, op_id, "delete").await?;
     }
     tx.commit().await?;
     tombstones.sort_by(|a, b| a.uuid.cmp(&b.uuid));
@@ -1692,21 +1723,83 @@ async fn claim_delete_marks(
     Ok((Some(op_id), person_ids))
 }
 
-async fn release_delete_marks(
+/// Marks the split source as a 'merge' op, the type ingestion uses to move distinct ids.
+async fn claim_split_mark(
     tx: &mut Transaction<'_, Postgres>,
     team_id: i64,
-    op_id: Uuid,
-) -> StorageResult<()> {
+    person_id: i64,
+) -> StorageResult<Uuid> {
+    let person_uuid = sqlx::query_scalar!(
+        r#"
+        SELECT uuid as "uuid!"
+        FROM posthog_person
+        WHERE team_id = $1 AND id = $2
+        "#,
+        team_id as i32,
+        person_id
+    )
+    .fetch_optional(&mut **tx)
+    .await?
+    .ok_or_else(|| StorageError::NotFound(format!("person_id={person_id} (team_id={team_id})")))?;
+
+    let op_id = Uuid::new_v4();
     sqlx::query!(
         r#"
-        DELETE FROM lifecycle_op
-        WHERE op_id = $1 AND team_id = $2 AND op_type = 'delete'
+        INSERT INTO lifecycle_op (op_id, op_type, team_id, step, request, completed_at)
+        VALUES ($1, 'merge', $2, 'completed', '{"source": "personhog-replica-split"}'::jsonb, now())
         "#,
         op_id,
         team_id as i32
     )
     .execute(&mut **tx)
     .await?;
+    let claimed = sqlx::query_scalar!(
+        r#"
+        INSERT INTO lifecycle_op_person (op_id, team_id, person_id, person_uuid, role, status, mark_active)
+        VALUES ($1, $2, $3, $4, 'source', 'marked', true)
+        ON CONFLICT (team_id, person_id) WHERE mark_active DO NOTHING
+        RETURNING person_id
+        "#,
+        op_id,
+        team_id as i32,
+        person_id,
+        person_uuid
+    )
+    .fetch_optional(&mut **tx)
+    .await?;
+    if claimed.is_none() {
+        return Err(StorageError::FailedPrecondition(format!(
+            "person_id={person_id} (team_id={team_id}) is claimed by another lifecycle operation"
+        )));
+    }
+    Ok(op_id)
+}
+
+/// Deletes the op inside the claiming transaction, so committed state never holds its marks.
+async fn release_lifecycle_marks(
+    tx: &mut Transaction<'_, Postgres>,
+    team_id: i64,
+    op_id: Uuid,
+    op_type: &str,
+) -> StorageResult<()> {
+    let deleted = sqlx::query!(
+        r#"
+        DELETE FROM lifecycle_op
+        WHERE op_id = $1 AND team_id = $2 AND op_type = $3
+        "#,
+        op_id,
+        team_id as i32,
+        op_type
+    )
+    .execute(&mut **tx)
+    .await?
+    .rows_affected();
+    // Committing without the delete would leave the marks held for good.
+    if deleted != 1 {
+        return Err(StorageError::Query(format!(
+            "releasing lifecycle op {op_id} deleted {deleted} rows (team_id={team_id})"
+        )));
+    }
     Ok(())
 }