From 697aa012de4319543ad416fe79c22ed04b62def3 Mon Sep 17 00:00:00 2001 From: Eli Reisman Date: Tue, 13 Oct 2026 18:56:38 -0700 Subject: [PATCH 1/6] fix(persons): reject a split that leaves the person with no distinct ids A split that moves every distinct id off a person, or that names a main distinct id the person does not own, leaves a live person no event can reach and no cleanup job picks up. The API now returns 400 for both requests, and the split task re-checks before it moves anything. Co-Authored-By: Claude Opus 5.5 --- posthog/api/person.py | 5 +++ posthog/api/test/test_person.py | 18 +++++++++++ posthog/models/person/person.py | 31 +++++++++++++++++-- .../models/person/test/test_split_person.py | 28 +++++++++++++++++ 4 files changed, 80 insertions(+), 2 deletions(-) diff --git a/posthog/api/person.py b/posthog/api/person.py index d3f311067a82..70696df727a6 100644 --- a/posthog/api/person.py +++ b/posthog/api/person.py @@ -1343,6 +1343,11 @@ def split(self, request: request.Request, pk=None, **kwargs) -> response.Respons unknown = set(distinct_ids_to_split) - set(distinct_ids) if unknown: raise ValidationError({"distinct_ids_to_split": f"not on this person: {sorted(unknown)}"}) + if set(distinct_ids) <= set(distinct_ids_to_split): + raise ValidationError({"distinct_ids_to_split": "must leave at least one distinct ID on this person"}) + elif main_distinct_id and main_distinct_id not in distinct_ids: + # An unknown main ID would split off every distinct ID and leave the person unreachable. + raise ValidationError({"main_distinct_id": "not on this person"}) split_person.delay( person.id, diff --git a/posthog/api/test/test_person.py b/posthog/api/test/test_person.py index c043708b397b..82e336cd09e3 100644 --- a/posthog/api/test/test_person.py +++ b/posthog/api/test/test_person.py @@ -1385,6 +1385,24 @@ def test_split_people_partial_rejects_unknown_distinct_id(self) -> None: assert original is not None self.assertCountEqual(original.distinct_ids, ["a", "b"]) + @parameterized.expand( + [ + ("every_id_listed", {"distinct_ids_to_split": ["b", "a", "a"]}, "must leave at least one distinct ID"), + ("unknown_main_id", {"main_distinct_id": "not_on_this_person"}, "not on this person"), + ] + ) + def test_split_people_rejects_a_split_that_leaves_no_distinct_id( + self, _name: str, payload: dict, message: str + ) -> None: + person1 = _create_person(team=self.team, distinct_ids=["a", "b"], properties={}, immediate=True) + + with mock.patch("posthog.api.person.split_person.delay") as split_task: + response = self.client.post(f"/api/person/{person1.pk}/split/", payload, content_type="application/json") + + self.assertEqual(response.status_code, 400) + self.assertIn(message, response.content.decode()) + split_task.assert_not_called() + def test_split_people_partial_rejects_combined_with_main_distinct_id(self) -> None: person1 = _create_person( team=self.team, diff --git a/posthog/models/person/person.py b/posthog/models/person/person.py index f29e9c371b4d..82e7c75ca277 100644 --- a/posthog/models/person/person.py +++ b/posthog/models/person/person.py @@ -271,16 +271,31 @@ def split_person( ) if distinct_ids_to_split is not None: - self._split_explicit_ids(distinct_ids_to_split) + self._split_explicit_ids(client, distinct_ids_to_split) else: + if main_distinct_id: + self._require_own_distinct_id(client, main_distinct_id) self._split_all_ids(client, main_distinct_id, max_splits) - def _split_explicit_ids(self, distinct_ids_to_split: list[str]) -> None: + def _require_own_distinct_id(self, client: Any, distinct_id: str) -> None: + """A split must keep one distinct ID on the person: a live person without one is unreachable and + is never cleaned up. An unowned main ID would split off every ID, so reject it before any batch runs.""" + from posthog.personhog_client.proto import GetPersonByDistinctIdRequest + + resp = client.get_person_by_distinct_id( + GetPersonByDistinctIdRequest(team_id=self.team_id, distinct_id=distinct_id) + ) + if not resp.person or resp.person.id != self.pk: + raise ValueError(f"main_distinct_id is not on person_id={self.pk} (team_id={self.team_id})") + + def _split_explicit_ids(self, client: Any, distinct_ids_to_split: list[str]) -> None: """Partial split: caller specifies exactly which IDs to move. The RPC validates that every ID belongs to this person — no need to fetch all distinct IDs upfront. """ + from posthog.personhog_client.proto import GetDistinctIdsForPersonRequest + seen: set[str] = set() distinct_ids_to_process: list[str] = [] for did in distinct_ids_to_split: @@ -291,6 +306,18 @@ def _split_explicit_ids(self, distinct_ids_to_split: list[str]) -> None: if not distinct_ids_to_process: return + # One more ID than the request names: if every returned ID is in the request, none would remain. + remaining = client.get_distinct_ids_for_person( + GetDistinctIdsForPersonRequest( + team_id=self.team_id, person_id=self.pk, limit=len(distinct_ids_to_process) + 1 + ) + ) + if all(d.distinct_id in seen for d in remaining.distinct_ids): + raise ValueError( + f"split would leave no distinct ID on person_id={self.pk} (team_id={self.team_id}), " + "so the person would be unreachable" + ) + logger.info( "split_person will split explicit distinct_ids", person_id=self.pk, diff --git a/posthog/models/person/test/test_split_person.py b/posthog/models/person/test/test_split_person.py index 8b1d24e0fb9b..6b94e3321c79 100644 --- a/posthog/models/person/test/test_split_person.py +++ b/posthog/models/person/test/test_split_person.py @@ -3,9 +3,12 @@ from posthog.test.base import BaseTest from unittest.mock import patch +from parameterized import parameterized + from posthog.models import Person from posthog.models.person.missing_person import uuidFromDistinctId from posthog.models.person.util import get_person_by_distinct_id, get_person_by_id +from posthog.models.utils import UUIDT from posthog.personhog_client.fake_client import FakePersonHogClient, fake_personhog_client from posthog.test.persons import add_distinct_id, create_person @@ -233,6 +236,31 @@ def test_partial_split_rejects_unknown_distinct_id(self, mock_create_pdi, mock_c mock_create_pdi.assert_not_called() mock_create_person.assert_not_called() + @parameterized.expand( + [ + ("every_id", {"main_distinct_id": None, "distinct_ids_to_split": ["id2", "id1", "id1"]}), + ("unknown_main_id", {"main_distinct_id": "not_on_this_person"}), + ("main_id_of_another_person", {"main_distinct_id": "other_persons_id"}), + ] + ) + def test_split_rejects_leaving_the_person_with_no_distinct_id( + self, mock_create_pdi, mock_create_person, _name: str, split_kwargs: dict + ): + with fake_personhog_client() as fake: + person = self._setup_person(fake, ["id1", "id2"], mock_create_pdi, mock_create_person) + fake.add_person( + team_id=self.team.id, person_id=person.id + 1, uuid=str(UUIDT()), distinct_ids=["other_persons_id"] + ) + + with self.assertRaises(ValueError): + person.split_person(**split_kwargs) + + fake.assert_not_called("split_person") + assert fake._persons_by_distinct_id[(self.team.id, "id1")].id == person.id + assert fake._persons_by_distinct_id[(self.team.id, "id2")].id == person.id + mock_create_pdi.assert_not_called() + mock_create_person.assert_not_called() + def test_partial_split_ignores_main_distinct_id_and_max_splits(self, mock_create_pdi, mock_create_person): with fake_personhog_client() as fake: person = self._setup_person(fake, ["a", "b", "c", "d"], mock_create_pdi, mock_create_person) From 285c0cfd797264e335af907384e02dd72b177f63 Mon Sep 17 00:00:00 2001 From: Eli Reisman Date: Tue, 13 Oct 2026 21:06:19 -0700 Subject: [PATCH 2/6] fix(personhog): serialize splits of one person on its lifecycle mark Two splits of the same person could each see an id the other was about to move and together leave the person with no live distinct id. The SplitPerson RPC now claims the source person's lifecycle mark before any row lock, as ingestion merges and the tombstone RPC do, and then refuses with FailedPrecondition a split that would leave no live distinct id. A held mark or row fails the call within a 2s lock timeout. Co-Authored-By: Claude Opus 5.5 --- ...528f8a52565944012c4d53908164a308870e4.json | 24 +++ ...24ca22eabd036425f5a34f73479e5da56cb7.json} | 7 +- ...0ea04ae5796a362f973adce0a925345e13392.json | 23 +++ ...192aaf71583c5583b80a60d690f0386a8026a.json | 15 ++ ...55577b09524dbc0b751db57d27968cd23ef7f.json | 25 +++ .../src/storage/postgres/person.rs | 108 ++++++++++++- rust/personhog-replica/tests/storage_tests.rs | 142 +++++++++++++++++- 7 files changed, 329 insertions(+), 15 deletions(-) create mode 100644 rust/personhog-replica/.sqlx/query-373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4.json rename rust/personhog-replica/.sqlx/{query-138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a.json => query-459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7.json} (57%) create mode 100644 rust/personhog-replica/.sqlx/query-96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392.json create mode 100644 rust/personhog-replica/.sqlx/query-cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a.json create mode 100644 rust/personhog-replica/.sqlx/query-d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f.json diff --git a/rust/personhog-replica/.sqlx/query-373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4.json b/rust/personhog-replica/.sqlx/query-373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4.json new file mode 100644 index 000000000000..bddf5c61264c --- /dev/null +++ b/rust/personhog-replica/.sqlx/query-373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4.json @@ -0,0 +1,24 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT EXISTS (\n SELECT 1 FROM posthog_persondistinctid\n WHERE team_id = $1 AND person_id = $2 AND is_deleted = false AND NOT (distinct_id = ANY($3))\n ) as \"exists!\"\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "exists!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + "Int4", + "Int8", + "TextArray" + ] + }, + "nullable": [ + null + ] + }, + "hash": "373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4" +} diff --git a/rust/personhog-replica/.sqlx/query-138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a.json b/rust/personhog-replica/.sqlx/query-459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7.json similarity index 57% rename from rust/personhog-replica/.sqlx/query-138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a.json rename to rust/personhog-replica/.sqlx/query-459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7.json index b0642aac9ba6..978beb104c6b 100644 --- a/rust/personhog-replica/.sqlx/query-138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a.json +++ b/rust/personhog-replica/.sqlx/query-459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7.json @@ -1,15 +1,16 @@ { "db_name": "PostgreSQL", - "query": "\n DELETE FROM lifecycle_op\n WHERE op_id = $1 AND team_id = $2 AND op_type = 'delete'\n ", + "query": "\n DELETE FROM lifecycle_op\n WHERE op_id = $1 AND team_id = $2 AND op_type = $3\n ", "describe": { "columns": [], "parameters": { "Left": [ "Uuid", - "Int4" + "Int4", + "Text" ] }, "nullable": [] }, - "hash": "138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a" + "hash": "459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7" } diff --git a/rust/personhog-replica/.sqlx/query-96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392.json b/rust/personhog-replica/.sqlx/query-96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392.json new file mode 100644 index 000000000000..48098485a409 --- /dev/null +++ b/rust/personhog-replica/.sqlx/query-96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT uuid as \"uuid!\"\n FROM posthog_person\n WHERE team_id = $1 AND id = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "uuid!", + "type_info": "Uuid" + } + ], + "parameters": { + "Left": [ + "Int4", + "Int8" + ] + }, + "nullable": [ + false + ] + }, + "hash": "96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392" +} diff --git a/rust/personhog-replica/.sqlx/query-cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a.json b/rust/personhog-replica/.sqlx/query-cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a.json new file mode 100644 index 000000000000..dfe5f23521d1 --- /dev/null +++ b/rust/personhog-replica/.sqlx/query-cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO lifecycle_op (op_id, op_type, team_id, step, request, completed_at)\n VALUES ($1, 'merge', $2, 'completed', '{\"source\": \"personhog-replica-split\"}'::jsonb, now())\n ", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "Uuid", + "Int4" + ] + }, + "nullable": [] + }, + "hash": "cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a" +} diff --git a/rust/personhog-replica/.sqlx/query-d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f.json b/rust/personhog-replica/.sqlx/query-d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f.json new file mode 100644 index 000000000000..84a12f37e2dd --- /dev/null +++ b/rust/personhog-replica/.sqlx/query-d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f.json @@ -0,0 +1,25 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO lifecycle_op_person (op_id, team_id, person_id, person_uuid, role, status, mark_active)\n VALUES ($1, $2, $3, $4, 'source', 'marked', true)\n ON CONFLICT (team_id, person_id) WHERE mark_active DO NOTHING\n RETURNING person_id\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "person_id", + "type_info": "Int8" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Int4", + "Int8", + "Uuid" + ] + }, + "nullable": [ + false + ] + }, + "hash": "d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f" +} diff --git a/rust/personhog-replica/src/storage/postgres/person.rs b/rust/personhog-replica/src/storage/postgres/person.rs index 8d90bfe2f009..e98e018e6f2b 100644 --- a/rust/personhog-replica/src/storage/postgres/person.rs +++ b/rust/personhog-replica/src/storage/postgres/person.rs @@ -982,6 +982,17 @@ 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?; + // A held mark or row means a merge, delete or split in flight: fail fast, rolled back, under the + // router's deadline instead of committing after the caller gave up and retried. + sqlx::query("SET LOCAL lock_timeout = '2s'") + .execute(&mut *tx) + .await?; + + // The source's mark serializes this split with the other writers that move its distinct ids away + // (other splits, ingestion merges, the tombstone RPC), so the check below that one id stays cannot + // pass in two splits at once. The manual detach_distinct_id job claims no mark. Marks go before any + // row lock, as everywhere else. + 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 +1044,27 @@ impl PersonLookup for PostgresStorage { ))); } + // A fresh statement after the claim, so it sees every split that committed while this one waited. + // A requested id whose derived uuid is the source's own stays on the source, but 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). @@ -1165,6 +1197,7 @@ impl PersonLookup for PostgresStorage { ))); } + release_lifecycle_marks(&mut tx, team_id, mark_op_id, "merge").await?; tx.commit().await?; let results: Vec = dids @@ -1618,7 +1651,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 +1725,84 @@ async fn claim_delete_marks( Ok((Some(op_id), person_ids)) } -async fn release_delete_marks( +/// Claims the split source's mark as a 'merge' op, the type ingestion uses for moving distinct ids between +/// persons. Fails like the tombstone RPC when another lifecycle operation holds the person. +async fn claim_split_mark( tx: &mut Transaction<'_, Postgres>, team_id: i64, - op_id: Uuid, -) -> StorageResult<()> { + person_id: i64, +) -> StorageResult { + 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(()) } diff --git a/rust/personhog-replica/tests/storage_tests.rs b/rust/personhog-replica/tests/storage_tests.rs index 45205de5abbf..fcfb08a15f7a 100644 --- a/rust/personhog-replica/tests/storage_tests.rs +++ b/rust/personhog-replica/tests/storage_tests.rs @@ -620,10 +620,7 @@ async fn wait_for_a_mark_claim_blocked_by(pool: &sqlx::PgPool, holder_pid: i32) if waiting > 0 { return; } - assert!( - Instant::now() < deadline, - "the delete never waited on the held mark" - ); + assert!(Instant::now() < deadline, "nothing waited on the held mark"); tokio::time::sleep(Duration::from_millis(10)).await; } } @@ -2871,6 +2868,139 @@ async fn test_split_person_creates_new_persons(#[case] num_splits: usize) { .expect("Lookup should succeed") .expect("Keeper DID should still resolve to original"); assert_eq!(original_by_did.id, person.id); + assert_eq!(lifecycle_op_count(&ctx.pool, ctx.team_id).await, 0); + + ctx.cleanup().await.ok(); +} + +async fn live_distinct_ids(pool: &sqlx::PgPool, team_id: i64, person_id: i64) -> Vec { + sqlx::query_scalar( + "SELECT distinct_id FROM posthog_persondistinctid + WHERE team_id = $1 AND person_id = $2 AND NOT is_deleted ORDER BY distinct_id", + ) + .bind(team_id as i32) + .bind(person_id) + .fetch_all(pool) + .await + .unwrap() +} + +fn assert_failed_precondition( + result: Result, +) { + assert!( + matches!( + result, + Err(personhog_replica::storage::StorageError::FailedPrecondition(_)) + ), + "expected FailedPrecondition, got {result:?}" + ); +} + +#[tokio::test] +async fn test_split_person_rejects_moving_every_distinct_id() { + let ctx = TestContext::new().await; + let person = ctx.insert_person("split_all_a", None).await.unwrap(); + ctx.add_distinct_id_to_person(person.id, "split_all_b") + .await + .unwrap(); + + let result = ctx + .storage + .split_person( + ctx.team_id, + person.id, + &["split_all_a".to_string(), "split_all_b".to_string()], + ) + .await; + + assert_failed_precondition(result); + assert_eq!( + live_distinct_ids(&ctx.pool, ctx.team_id, person.id).await, + vec!["split_all_a", "split_all_b"] + ); + assert_eq!(lifecycle_op_count(&ctx.pool, ctx.team_id).await, 0); + + ctx.cleanup().await.ok(); +} + +#[tokio::test] +async fn test_split_person_waits_for_a_concurrent_move_and_keeps_the_last_distinct_id() { + let ctx = TestContext::new().await; + let person = ctx.insert_person("race_keep_a", None).await.unwrap(); + ctx.add_distinct_id_to_person(person.id, "race_keep_b") + .await + .unwrap(); + let other = ctx.insert_person("race_other", None).await.unwrap(); + + // A concurrent split or merge, in flight: it holds the person's mark and has moved "a" away. + let mut mover = ctx.pool.begin().await.unwrap(); + let mover_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") + .fetch_one(&mut *mover) + .await + .unwrap(); + let op_id = claim_merge_mark(&mut mover, ctx.team_id, &person).await; + sqlx::query( + "UPDATE posthog_persondistinctid SET person_id = $1, version = version + 1 + WHERE team_id = $2 AND distinct_id = 'race_keep_a'", + ) + .bind(other.id) + .bind(ctx.team_id as i32) + .execute(&mut *mover) + .await + .unwrap(); + sqlx::query("DELETE FROM lifecycle_op WHERE op_id = $1") + .bind(op_id) + .execute(&mut *mover) + .await + .unwrap(); + + let storage = ctx.storage.clone(); + let team_id = ctx.team_id; + let person_id = person.id; + let split = tokio::spawn(async move { + storage + .split_person(team_id, person_id, &["race_keep_b".to_string()]) + .await + }); + + // Commit the move only once the split waits on the mark, so the split cannot finish first. + wait_for_a_mark_claim_blocked_by(&ctx.pool, mover_pid).await; + mover.commit().await.unwrap(); + + assert_failed_precondition(split.await.unwrap()); + assert_eq!( + live_distinct_ids(&ctx.pool, ctx.team_id, person.id).await, + vec!["race_keep_b"] + ); + assert_eq!(lifecycle_op_count(&ctx.pool, ctx.team_id).await, 0); + + ctx.cleanup().await.ok(); +} + +#[tokio::test] +async fn test_split_person_fails_when_a_lifecycle_op_holds_the_person() { + let ctx = TestContext::new().await; + let person = ctx.insert_person("split_held_a", None).await.unwrap(); + ctx.add_distinct_id_to_person(person.id, "split_held_b") + .await + .unwrap(); + let mut conn = ctx.pool.acquire().await.unwrap(); + claim_merge_mark(&mut conn, ctx.team_id, &person).await; + drop(conn); + + let result = ctx + .storage + .split_person(ctx.team_id, person.id, &["split_held_b".to_string()]) + .await; + + assert_failed_precondition(result); + assert_eq!( + live_distinct_ids(&ctx.pool, ctx.team_id, person.id).await, + vec!["split_held_a", "split_held_b"] + ); + // Only the holder's op: the split's own claim rolled back. + assert_eq!(lifecycle_op_count(&ctx.pool, ctx.team_id).await, 1); ctx.cleanup().await.ok(); } @@ -2898,9 +3028,9 @@ async fn test_split_person_version_arithmetic() { .await .unwrap(); - // Insert PDIs with non-zero versions + // Insert PDIs with non-zero versions; the split must leave one on the person. let pdi_version: i64 = 7; - for did in &["v_did_a", "v_did_b"] { + for did in &["v_did_a", "v_did_b", "v_did_keep"] { sqlx::query( r#"INSERT INTO posthog_persondistinctid (distinct_id, person_id, team_id, version) VALUES ($1, $2, $3, $4)"#, From 010c4fbc887db9f0de316a00f178202faecbafbc Mon Sep 17 00:00:00 2001 From: Eli Reisman Date: Tue, 13 Oct 2026 21:14:13 -0700 Subject: [PATCH 3/6] fix(personhog): lock split target persons in id order The tombstone RPC locks persons in id order, and the split locked its target persons in no particular order, so a split and a tombstone call naming two of the same persons could deadlock. The comment on the lock timeout now says it bounds each wait, not the whole call. Co-Authored-By: Claude Opus 5.5 --- ...89e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e.json} | 4 ++-- rust/personhog-replica/src/storage/postgres/person.rs | 8 +++++--- 2 files changed, 7 insertions(+), 5 deletions(-) rename rust/personhog-replica/.sqlx/{query-49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78.json => query-0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e.json} (83%) diff --git a/rust/personhog-replica/.sqlx/query-49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78.json b/rust/personhog-replica/.sqlx/query-0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e.json similarity index 83% rename from rust/personhog-replica/.sqlx/query-49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78.json rename to rust/personhog-replica/.sqlx/query-0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e.json index f9a13ecd40ef..52546426dc78 100644 --- a/rust/personhog-replica/.sqlx/query-49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78.json +++ b/rust/personhog-replica/.sqlx/query-0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT id::bigint as \"id!\", uuid as \"uuid!\", created_at as \"created_at!\"\n FROM posthog_person\n WHERE team_id = $1 AND uuid = ANY($2)\n FOR UPDATE\n ", + "query": "\n SELECT id::bigint as \"id!\", uuid as \"uuid!\", created_at as \"created_at!\"\n FROM posthog_person\n WHERE team_id = $1 AND uuid = ANY($2)\n ORDER BY id\n FOR UPDATE\n ", "describe": { "columns": [ { @@ -31,5 +31,5 @@ false ] }, - "hash": "49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78" + "hash": "0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e" } diff --git a/rust/personhog-replica/src/storage/postgres/person.rs b/rust/personhog-replica/src/storage/postgres/person.rs index e98e018e6f2b..35eb9155fa1d 100644 --- a/rust/personhog-replica/src/storage/postgres/person.rs +++ b/rust/personhog-replica/src/storage/postgres/person.rs @@ -982,8 +982,9 @@ 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?; - // A held mark or row means a merge, delete or split in flight: fail fast, rolled back, under the - // router's deadline instead of committing after the caller gave up and retried. + // A held mark or row means a merge, delete or split in flight: fail fast and roll back instead of + // waiting out the statement timeout. The bound is per wait, so a split can still outlast the + // router's deadline when several waits add up. sqlx::query("SET LOCAL lock_timeout = '2s'") .execute(&mut *tx) .await?; @@ -1086,12 +1087,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, From c5eff82cec23ce3ced602dbd139a3428fe2871a0 Mon Sep 17 00:00:00 2001 From: Eli Reisman Date: Tue, 13 Oct 2026 21:20:23 -0700 Subject: [PATCH 4/6] chore(persons): move the personhog split guard to its own pr The merge queue keeps Python and Rust changes in separate lanes, so the SplitPerson lifecycle mark change moves to #116730. This branch keeps the API and task checks only. Co-Authored-By: Claude Opus 5.5 --- ...b6fef5a1750e00e4797687861e4d0f1be82a.json} | 7 +- ...528f8a52565944012c4d53908164a308870e4.json | 24 --- ...2e4a004a39a064354ea5ef7dd6ae4e5afb78.json} | 4 +- ...0ea04ae5796a362f973adce0a925345e13392.json | 23 --- ...192aaf71583c5583b80a60d690f0386a8026a.json | 15 -- ...55577b09524dbc0b751db57d27968cd23ef7f.json | 25 --- .../src/storage/postgres/person.rs | 112 +------------- rust/personhog-replica/tests/storage_tests.rs | 142 +----------------- 8 files changed, 18 insertions(+), 334 deletions(-) rename rust/personhog-replica/.sqlx/{query-459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7.json => query-138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a.json} (57%) delete mode 100644 rust/personhog-replica/.sqlx/query-373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4.json rename rust/personhog-replica/.sqlx/{query-0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e.json => query-49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78.json} (83%) delete mode 100644 rust/personhog-replica/.sqlx/query-96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392.json delete mode 100644 rust/personhog-replica/.sqlx/query-cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a.json delete mode 100644 rust/personhog-replica/.sqlx/query-d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f.json diff --git a/rust/personhog-replica/.sqlx/query-459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7.json b/rust/personhog-replica/.sqlx/query-138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a.json similarity index 57% rename from rust/personhog-replica/.sqlx/query-459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7.json rename to rust/personhog-replica/.sqlx/query-138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a.json index 978beb104c6b..b0642aac9ba6 100644 --- a/rust/personhog-replica/.sqlx/query-459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7.json +++ b/rust/personhog-replica/.sqlx/query-138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a.json @@ -1,16 +1,15 @@ { "db_name": "PostgreSQL", - "query": "\n DELETE FROM lifecycle_op\n WHERE op_id = $1 AND team_id = $2 AND op_type = $3\n ", + "query": "\n DELETE FROM lifecycle_op\n WHERE op_id = $1 AND team_id = $2 AND op_type = 'delete'\n ", "describe": { "columns": [], "parameters": { "Left": [ "Uuid", - "Int4", - "Text" + "Int4" ] }, "nullable": [] }, - "hash": "459ab338d573c90e1cea2116fc4424ca22eabd036425f5a34f73479e5da56cb7" + "hash": "138b31578ae24294373586142495b6fef5a1750e00e4797687861e4d0f1be82a" } diff --git a/rust/personhog-replica/.sqlx/query-373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4.json b/rust/personhog-replica/.sqlx/query-373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4.json deleted file mode 100644 index bddf5c61264c..000000000000 --- a/rust/personhog-replica/.sqlx/query-373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4.json +++ /dev/null @@ -1,24 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT EXISTS (\n SELECT 1 FROM posthog_persondistinctid\n WHERE team_id = $1 AND person_id = $2 AND is_deleted = false AND NOT (distinct_id = ANY($3))\n ) as \"exists!\"\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "exists!", - "type_info": "Bool" - } - ], - "parameters": { - "Left": [ - "Int4", - "Int8", - "TextArray" - ] - }, - "nullable": [ - null - ] - }, - "hash": "373f3eccba1f2a9a6627543a0de528f8a52565944012c4d53908164a308870e4" -} diff --git a/rust/personhog-replica/.sqlx/query-0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e.json b/rust/personhog-replica/.sqlx/query-49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78.json similarity index 83% rename from rust/personhog-replica/.sqlx/query-0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e.json rename to rust/personhog-replica/.sqlx/query-49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78.json index 52546426dc78..f9a13ecd40ef 100644 --- a/rust/personhog-replica/.sqlx/query-0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e.json +++ b/rust/personhog-replica/.sqlx/query-49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78.json @@ -1,6 +1,6 @@ { "db_name": "PostgreSQL", - "query": "\n SELECT id::bigint as \"id!\", uuid as \"uuid!\", created_at as \"created_at!\"\n FROM posthog_person\n WHERE team_id = $1 AND uuid = ANY($2)\n ORDER BY id\n FOR UPDATE\n ", + "query": "\n SELECT id::bigint as \"id!\", uuid as \"uuid!\", created_at as \"created_at!\"\n FROM posthog_person\n WHERE team_id = $1 AND uuid = ANY($2)\n FOR UPDATE\n ", "describe": { "columns": [ { @@ -31,5 +31,5 @@ false ] }, - "hash": "0ebe83ceec43b02189e46d1adbc9fc3a70c8b613c7b3bdfdcb053bb84534a57e" + "hash": "49bcc915e71cec48ea8e91fac80f2e4a004a39a064354ea5ef7dd6ae4e5afb78" } diff --git a/rust/personhog-replica/.sqlx/query-96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392.json b/rust/personhog-replica/.sqlx/query-96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392.json deleted file mode 100644 index 48098485a409..000000000000 --- a/rust/personhog-replica/.sqlx/query-96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392.json +++ /dev/null @@ -1,23 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n SELECT uuid as \"uuid!\"\n FROM posthog_person\n WHERE team_id = $1 AND id = $2\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "uuid!", - "type_info": "Uuid" - } - ], - "parameters": { - "Left": [ - "Int4", - "Int8" - ] - }, - "nullable": [ - false - ] - }, - "hash": "96ae2fe1f4c10d259e4151924c80ea04ae5796a362f973adce0a925345e13392" -} diff --git a/rust/personhog-replica/.sqlx/query-cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a.json b/rust/personhog-replica/.sqlx/query-cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a.json deleted file mode 100644 index dfe5f23521d1..000000000000 --- a/rust/personhog-replica/.sqlx/query-cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO lifecycle_op (op_id, op_type, team_id, step, request, completed_at)\n VALUES ($1, 'merge', $2, 'completed', '{\"source\": \"personhog-replica-split\"}'::jsonb, now())\n ", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Uuid", - "Int4" - ] - }, - "nullable": [] - }, - "hash": "cd8cc89f4662b7ea42e9f7e58ef192aaf71583c5583b80a60d690f0386a8026a" -} diff --git a/rust/personhog-replica/.sqlx/query-d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f.json b/rust/personhog-replica/.sqlx/query-d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f.json deleted file mode 100644 index 84a12f37e2dd..000000000000 --- a/rust/personhog-replica/.sqlx/query-d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f.json +++ /dev/null @@ -1,25 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "\n INSERT INTO lifecycle_op_person (op_id, team_id, person_id, person_uuid, role, status, mark_active)\n VALUES ($1, $2, $3, $4, 'source', 'marked', true)\n ON CONFLICT (team_id, person_id) WHERE mark_active DO NOTHING\n RETURNING person_id\n ", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "person_id", - "type_info": "Int8" - } - ], - "parameters": { - "Left": [ - "Uuid", - "Int4", - "Int8", - "Uuid" - ] - }, - "nullable": [ - false - ] - }, - "hash": "d536dcec71c04e794120461f6f655577b09524dbc0b751db57d27968cd23ef7f" -} diff --git a/rust/personhog-replica/src/storage/postgres/person.rs b/rust/personhog-replica/src/storage/postgres/person.rs index 35eb9155fa1d..8d90bfe2f009 100644 --- a/rust/personhog-replica/src/storage/postgres/person.rs +++ b/rust/personhog-replica/src/storage/postgres/person.rs @@ -982,18 +982,6 @@ 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?; - // A held mark or row means a merge, delete or split in flight: fail fast and roll back instead of - // waiting out the statement timeout. The bound is per wait, so a split can still outlast the - // router's deadline when several waits add up. - sqlx::query("SET LOCAL lock_timeout = '2s'") - .execute(&mut *tx) - .await?; - - // The source's mark serializes this split with the other writers that move its distinct ids away - // (other splits, ingestion merges, the tombstone RPC), so the check below that one id stays cannot - // pass in two splits at once. The manual detach_distinct_id job claims no mark. Marks go before any - // row lock, as everywhere else. - 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 @@ -1045,27 +1033,6 @@ impl PersonLookup for PostgresStorage { ))); } - // A fresh statement after the claim, so it sees every split that committed while this one waited. - // A requested id whose derived uuid is the source's own stays on the source, but 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). @@ -1087,13 +1054,12 @@ 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. Lock in id order, as the tombstone RPC does, so the two cannot deadlock. + // inference doesn't work. 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, @@ -1199,7 +1165,6 @@ impl PersonLookup for PostgresStorage { ))); } - release_lifecycle_marks(&mut tx, team_id, mark_op_id, "merge").await?; tx.commit().await?; let results: Vec = dids @@ -1653,7 +1618,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_lifecycle_marks(&mut tx, team_id, op_id, "delete").await?; + release_delete_marks(&mut tx, team_id, op_id).await?; } tx.commit().await?; tombstones.sort_by(|a, b| a.uuid.cmp(&b.uuid)); @@ -1727,84 +1692,21 @@ async fn claim_delete_marks( Ok((Some(op_id), person_ids)) } -/// Claims the split source's mark as a 'merge' op, the type ingestion uses for moving distinct ids between -/// persons. Fails like the tombstone RPC when another lifecycle operation holds the person. -async fn claim_split_mark( - tx: &mut Transaction<'_, Postgres>, - team_id: i64, - person_id: i64, -) -> StorageResult { - 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#" - 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( +async fn release_delete_marks( tx: &mut Transaction<'_, Postgres>, team_id: i64, op_id: Uuid, - op_type: &str, ) -> StorageResult<()> { - let deleted = sqlx::query!( + sqlx::query!( r#" DELETE FROM lifecycle_op - WHERE op_id = $1 AND team_id = $2 AND op_type = $3 + WHERE op_id = $1 AND team_id = $2 AND op_type = 'delete' "#, op_id, - team_id as i32, - op_type + team_id as i32 ) .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})" - ))); - } + .await?; Ok(()) } diff --git a/rust/personhog-replica/tests/storage_tests.rs b/rust/personhog-replica/tests/storage_tests.rs index fcfb08a15f7a..45205de5abbf 100644 --- a/rust/personhog-replica/tests/storage_tests.rs +++ b/rust/personhog-replica/tests/storage_tests.rs @@ -620,7 +620,10 @@ async fn wait_for_a_mark_claim_blocked_by(pool: &sqlx::PgPool, holder_pid: i32) if waiting > 0 { return; } - assert!(Instant::now() < deadline, "nothing waited on the held mark"); + assert!( + Instant::now() < deadline, + "the delete never waited on the held mark" + ); tokio::time::sleep(Duration::from_millis(10)).await; } } @@ -2868,139 +2871,6 @@ async fn test_split_person_creates_new_persons(#[case] num_splits: usize) { .expect("Lookup should succeed") .expect("Keeper DID should still resolve to original"); assert_eq!(original_by_did.id, person.id); - assert_eq!(lifecycle_op_count(&ctx.pool, ctx.team_id).await, 0); - - ctx.cleanup().await.ok(); -} - -async fn live_distinct_ids(pool: &sqlx::PgPool, team_id: i64, person_id: i64) -> Vec { - sqlx::query_scalar( - "SELECT distinct_id FROM posthog_persondistinctid - WHERE team_id = $1 AND person_id = $2 AND NOT is_deleted ORDER BY distinct_id", - ) - .bind(team_id as i32) - .bind(person_id) - .fetch_all(pool) - .await - .unwrap() -} - -fn assert_failed_precondition( - result: Result, -) { - assert!( - matches!( - result, - Err(personhog_replica::storage::StorageError::FailedPrecondition(_)) - ), - "expected FailedPrecondition, got {result:?}" - ); -} - -#[tokio::test] -async fn test_split_person_rejects_moving_every_distinct_id() { - let ctx = TestContext::new().await; - let person = ctx.insert_person("split_all_a", None).await.unwrap(); - ctx.add_distinct_id_to_person(person.id, "split_all_b") - .await - .unwrap(); - - let result = ctx - .storage - .split_person( - ctx.team_id, - person.id, - &["split_all_a".to_string(), "split_all_b".to_string()], - ) - .await; - - assert_failed_precondition(result); - assert_eq!( - live_distinct_ids(&ctx.pool, ctx.team_id, person.id).await, - vec!["split_all_a", "split_all_b"] - ); - assert_eq!(lifecycle_op_count(&ctx.pool, ctx.team_id).await, 0); - - ctx.cleanup().await.ok(); -} - -#[tokio::test] -async fn test_split_person_waits_for_a_concurrent_move_and_keeps_the_last_distinct_id() { - let ctx = TestContext::new().await; - let person = ctx.insert_person("race_keep_a", None).await.unwrap(); - ctx.add_distinct_id_to_person(person.id, "race_keep_b") - .await - .unwrap(); - let other = ctx.insert_person("race_other", None).await.unwrap(); - - // A concurrent split or merge, in flight: it holds the person's mark and has moved "a" away. - let mut mover = ctx.pool.begin().await.unwrap(); - let mover_pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()") - .fetch_one(&mut *mover) - .await - .unwrap(); - let op_id = claim_merge_mark(&mut mover, ctx.team_id, &person).await; - sqlx::query( - "UPDATE posthog_persondistinctid SET person_id = $1, version = version + 1 - WHERE team_id = $2 AND distinct_id = 'race_keep_a'", - ) - .bind(other.id) - .bind(ctx.team_id as i32) - .execute(&mut *mover) - .await - .unwrap(); - sqlx::query("DELETE FROM lifecycle_op WHERE op_id = $1") - .bind(op_id) - .execute(&mut *mover) - .await - .unwrap(); - - let storage = ctx.storage.clone(); - let team_id = ctx.team_id; - let person_id = person.id; - let split = tokio::spawn(async move { - storage - .split_person(team_id, person_id, &["race_keep_b".to_string()]) - .await - }); - - // Commit the move only once the split waits on the mark, so the split cannot finish first. - wait_for_a_mark_claim_blocked_by(&ctx.pool, mover_pid).await; - mover.commit().await.unwrap(); - - assert_failed_precondition(split.await.unwrap()); - assert_eq!( - live_distinct_ids(&ctx.pool, ctx.team_id, person.id).await, - vec!["race_keep_b"] - ); - assert_eq!(lifecycle_op_count(&ctx.pool, ctx.team_id).await, 0); - - ctx.cleanup().await.ok(); -} - -#[tokio::test] -async fn test_split_person_fails_when_a_lifecycle_op_holds_the_person() { - let ctx = TestContext::new().await; - let person = ctx.insert_person("split_held_a", None).await.unwrap(); - ctx.add_distinct_id_to_person(person.id, "split_held_b") - .await - .unwrap(); - let mut conn = ctx.pool.acquire().await.unwrap(); - claim_merge_mark(&mut conn, ctx.team_id, &person).await; - drop(conn); - - let result = ctx - .storage - .split_person(ctx.team_id, person.id, &["split_held_b".to_string()]) - .await; - - assert_failed_precondition(result); - assert_eq!( - live_distinct_ids(&ctx.pool, ctx.team_id, person.id).await, - vec!["split_held_a", "split_held_b"] - ); - // Only the holder's op: the split's own claim rolled back. - assert_eq!(lifecycle_op_count(&ctx.pool, ctx.team_id).await, 1); ctx.cleanup().await.ok(); } @@ -3028,9 +2898,9 @@ async fn test_split_person_version_arithmetic() { .await .unwrap(); - // Insert PDIs with non-zero versions; the split must leave one on the person. + // Insert PDIs with non-zero versions let pdi_version: i64 = 7; - for did in &["v_did_a", "v_did_b", "v_did_keep"] { + for did in &["v_did_a", "v_did_b"] { sqlx::query( r#"INSERT INTO posthog_persondistinctid (distinct_id, person_id, team_id, version) VALUES ($1, $2, $3, $4)"#, From 6a16710819fbba39306d0e0a5e364634f9ff2ef5 Mon Sep 17 00:00:00 2001 From: Eli Reisman Date: Tue, 13 Oct 2026 22:10:02 -0700 Subject: [PATCH 5/6] fix(persons): retry a split batch after a database timeout A split that waits behind a merge of the same person can hit a database lock or statement timeout. The server rolls the whole batch back, so the task now sends it again, up to three times, instead of failing after the user already saw the split succeed. Other errors are not retried. Co-Authored-By: Claude Opus 5.5 --- posthog/models/person/person.py | 32 +++++++++------- .../models/person/test/test_split_person.py | 38 ++++++++++++++++++- posthog/personhog_client/interceptor.py | 10 +++++ 3 files changed, 65 insertions(+), 15 deletions(-) diff --git a/posthog/models/person/person.py b/posthog/models/person/person.py index 82e7c75ca277..94f6a602a3ae 100644 --- a/posthog/models/person/person.py +++ b/posthog/models/person/person.py @@ -1,3 +1,4 @@ +import time from datetime import UTC, datetime from typing import Any, NamedTuple, Optional from uuid import UUID @@ -19,6 +20,8 @@ # Mirrors MAX_SPLIT_BATCH_SIZE enforced by the personhog SplitPerson RPC. PERSONHOG_SPLIT_BATCH_SIZE = 250 +SPLIT_DB_TIMEOUT_ATTEMPTS = 3 +SPLIT_DB_TIMEOUT_BACKOFF_SECONDS = 0.5 class SplitOutcome(NamedTuple): @@ -278,8 +281,7 @@ def split_person( self._split_all_ids(client, main_distinct_id, max_splits) def _require_own_distinct_id(self, client: Any, distinct_id: str) -> None: - """A split must keep one distinct ID on the person: a live person without one is unreachable and - is never cleaned up. An unowned main ID would split off every ID, so reject it before any batch runs.""" + """An unowned main ID would split off every ID and strand the person, so reject it before any batch runs.""" from posthog.personhog_client.proto import GetPersonByDistinctIdRequest resp = client.get_person_by_distinct_id( @@ -289,11 +291,7 @@ def _require_own_distinct_id(self, client: Any, distinct_id: str) -> None: raise ValueError(f"main_distinct_id is not on person_id={self.pk} (team_id={self.team_id})") def _split_explicit_ids(self, client: Any, distinct_ids_to_split: list[str]) -> None: - """Partial split: caller specifies exactly which IDs to move. - - The RPC validates that every ID belongs to this person — no need to - fetch all distinct IDs upfront. - """ + """Partial split: the caller names the IDs to move. The RPC checks ownership; a bounded fetch checks one ID stays.""" from posthog.personhog_client.proto import GetDistinctIdsForPersonRequest seen: set[str] = set() @@ -398,7 +396,10 @@ def _split_distinct_ids_batch(self, distinct_ids: list[str]) -> list[SplitOutcom posthog/models/person/util.py (_delete_person) and rust/personhog-replica/src/storage/postgres/person.rs (SPLIT_VERSION_OFFSET). """ + import grpc + from posthog.personhog_client.client import get_personhog_client + from posthog.personhog_client.interceptor import is_db_timeout_error from posthog.personhog_client.proto import SplitPersonRequest client = get_personhog_client() @@ -407,13 +408,16 @@ def _split_distinct_ids_batch(self, distinct_ids: list[str]) -> list[SplitOutcom "split_person requires personhog, but the client is not configured (PERSONHOG_ADDR is unset)" ) - response = client.split_person( - SplitPersonRequest( - team_id=self.team_id, - person_id=self.pk, - distinct_ids_to_split=distinct_ids, - ) - ) + request = SplitPersonRequest(team_id=self.team_id, person_id=self.pk, distinct_ids_to_split=distinct_ids) + for attempt in range(1, SPLIT_DB_TIMEOUT_ATTEMPTS + 1): + try: + response = client.split_person(request) + break + except grpc.RpcError as exc: + # A database timeout rolled the whole batch back, often behind a merge of the same person. + if attempt == SPLIT_DB_TIMEOUT_ATTEMPTS or not is_db_timeout_error(exc): + raise + time.sleep(SPLIT_DB_TIMEOUT_BACKOFF_SECONDS * attempt) if len(response.splits) != len(distinct_ids): logger.error( "split_person RPC returned unexpected number of splits", diff --git a/posthog/models/person/test/test_split_person.py b/posthog/models/person/test/test_split_person.py index 6b94e3321c79..fc47cef0614c 100644 --- a/posthog/models/person/test/test_split_person.py +++ b/posthog/models/person/test/test_split_person.py @@ -1,8 +1,9 @@ from datetime import UTC, datetime from posthog.test.base import BaseTest -from unittest.mock import patch +from unittest.mock import MagicMock, patch +import grpc from parameterized import parameterized from posthog.models import Person @@ -313,6 +314,41 @@ def test_split_paginates_fetch_and_splits(self, mock_create_pdi, mock_create_per assert mock_create_person.call_count == 5 assert mock_create_pdi.call_count == 5 + @parameterized.expand( + [ + ("db_timeout_then_ok", 1, (("x-db-timeout", "lock_timeout"),), 2, True), + ("db_timeout_every_attempt", 99, (("x-db-timeout", "lock_timeout"),), 3, False), + ("other_deadline_exceeded", 1, (), 1, False), + ] + ) + def test_split_retries_a_batch_only_after_a_database_timeout( + self, mock_create_pdi, mock_create_person, _name, failures, metadata, expected_calls, succeeds + ): + with fake_personhog_client() as fake, patch("posthog.models.person.person.time.sleep"): + person = self._setup_person(fake, ["id1", "id2"], mock_create_pdi, mock_create_person) + real_split = fake.split_person + calls = [] + + def flaky_split(request, timeout=None): + calls.append(request) + if len(calls) <= failures: + error = grpc.RpcError() + error.code = MagicMock(return_value=grpc.StatusCode.DEADLINE_EXCEEDED) + error.details = MagicMock(return_value="timeout") + error.trailing_metadata = MagicMock(return_value=metadata) + raise error + return real_split(request, timeout=timeout) + + fake.split_person = flaky_split + if succeeds: + person.split_person(main_distinct_id="id1") + else: + with self.assertRaises(grpc.RpcError): + person.split_person(main_distinct_id="id1") + + assert len(calls) == expected_calls + assert mock_create_pdi.call_count == (1 if succeeds else 0) + def test_split_raises_when_rpc_fails(self, mock_create_pdi, mock_create_person): # No ORM fallback: an RPC failure propagates and nothing is published. with fake_personhog_client() as fake: diff --git a/posthog/personhog_client/interceptor.py b/posthog/personhog_client/interceptor.py index e64d6021d378..2b9b014ee367 100644 --- a/posthog/personhog_client/interceptor.py +++ b/posthog/personhog_client/interceptor.py @@ -179,6 +179,16 @@ def _is_db_timeout(exc: grpc.RpcError) -> bool: return any(key == _DB_TIMEOUT_METADATA_KEY for key, _ in exc.trailing_metadata() or ()) +def is_db_timeout_error(exc: BaseException) -> bool: + """Whether ``exc``, or the error it was raised from, is a database timeout, which the server rolled back.""" + current: BaseException | None = exc + while current is not None: + if isinstance(current, grpc.RpcError): + return callable(getattr(current, "trailing_metadata", None)) and _is_db_timeout(current) + current = current.__cause__ + return False + + def is_transient_rpc_error(exc: BaseException, codes: Collection[grpc.StatusCode] = _RETRYABLE_CODES) -> bool: """Whether ``exc``, or the error it was raised from, is a gRPC failure with a status in ``codes``. From 2f22581e8cd554499d03d3be46ede7d7cc8b33a4 Mon Sep 17 00:00:00 2001 From: Eli Reisman Date: Wed, 14 Oct 2026 14:12:45 -0700 Subject: [PATCH 6/6] fix(persons): drop the split batch retry on database timeouts Co-Authored-By: Claude Opus 5.5 --- posthog/models/person/person.py | 23 ++++------- .../models/person/test/test_split_person.py | 38 +------------------ posthog/personhog_client/interceptor.py | 10 ----- 3 files changed, 8 insertions(+), 63 deletions(-) diff --git a/posthog/models/person/person.py b/posthog/models/person/person.py index 94f6a602a3ae..d35529623825 100644 --- a/posthog/models/person/person.py +++ b/posthog/models/person/person.py @@ -1,4 +1,3 @@ -import time from datetime import UTC, datetime from typing import Any, NamedTuple, Optional from uuid import UUID @@ -20,8 +19,6 @@ # Mirrors MAX_SPLIT_BATCH_SIZE enforced by the personhog SplitPerson RPC. PERSONHOG_SPLIT_BATCH_SIZE = 250 -SPLIT_DB_TIMEOUT_ATTEMPTS = 3 -SPLIT_DB_TIMEOUT_BACKOFF_SECONDS = 0.5 class SplitOutcome(NamedTuple): @@ -396,10 +393,7 @@ def _split_distinct_ids_batch(self, distinct_ids: list[str]) -> list[SplitOutcom posthog/models/person/util.py (_delete_person) and rust/personhog-replica/src/storage/postgres/person.rs (SPLIT_VERSION_OFFSET). """ - import grpc - from posthog.personhog_client.client import get_personhog_client - from posthog.personhog_client.interceptor import is_db_timeout_error from posthog.personhog_client.proto import SplitPersonRequest client = get_personhog_client() @@ -408,16 +402,13 @@ def _split_distinct_ids_batch(self, distinct_ids: list[str]) -> list[SplitOutcom "split_person requires personhog, but the client is not configured (PERSONHOG_ADDR is unset)" ) - request = SplitPersonRequest(team_id=self.team_id, person_id=self.pk, distinct_ids_to_split=distinct_ids) - for attempt in range(1, SPLIT_DB_TIMEOUT_ATTEMPTS + 1): - try: - response = client.split_person(request) - break - except grpc.RpcError as exc: - # A database timeout rolled the whole batch back, often behind a merge of the same person. - if attempt == SPLIT_DB_TIMEOUT_ATTEMPTS or not is_db_timeout_error(exc): - raise - time.sleep(SPLIT_DB_TIMEOUT_BACKOFF_SECONDS * attempt) + response = client.split_person( + SplitPersonRequest( + team_id=self.team_id, + person_id=self.pk, + distinct_ids_to_split=distinct_ids, + ) + ) if len(response.splits) != len(distinct_ids): logger.error( "split_person RPC returned unexpected number of splits", diff --git a/posthog/models/person/test/test_split_person.py b/posthog/models/person/test/test_split_person.py index fc47cef0614c..6b94e3321c79 100644 --- a/posthog/models/person/test/test_split_person.py +++ b/posthog/models/person/test/test_split_person.py @@ -1,9 +1,8 @@ from datetime import UTC, datetime from posthog.test.base import BaseTest -from unittest.mock import MagicMock, patch +from unittest.mock import patch -import grpc from parameterized import parameterized from posthog.models import Person @@ -314,41 +313,6 @@ def test_split_paginates_fetch_and_splits(self, mock_create_pdi, mock_create_per assert mock_create_person.call_count == 5 assert mock_create_pdi.call_count == 5 - @parameterized.expand( - [ - ("db_timeout_then_ok", 1, (("x-db-timeout", "lock_timeout"),), 2, True), - ("db_timeout_every_attempt", 99, (("x-db-timeout", "lock_timeout"),), 3, False), - ("other_deadline_exceeded", 1, (), 1, False), - ] - ) - def test_split_retries_a_batch_only_after_a_database_timeout( - self, mock_create_pdi, mock_create_person, _name, failures, metadata, expected_calls, succeeds - ): - with fake_personhog_client() as fake, patch("posthog.models.person.person.time.sleep"): - person = self._setup_person(fake, ["id1", "id2"], mock_create_pdi, mock_create_person) - real_split = fake.split_person - calls = [] - - def flaky_split(request, timeout=None): - calls.append(request) - if len(calls) <= failures: - error = grpc.RpcError() - error.code = MagicMock(return_value=grpc.StatusCode.DEADLINE_EXCEEDED) - error.details = MagicMock(return_value="timeout") - error.trailing_metadata = MagicMock(return_value=metadata) - raise error - return real_split(request, timeout=timeout) - - fake.split_person = flaky_split - if succeeds: - person.split_person(main_distinct_id="id1") - else: - with self.assertRaises(grpc.RpcError): - person.split_person(main_distinct_id="id1") - - assert len(calls) == expected_calls - assert mock_create_pdi.call_count == (1 if succeeds else 0) - def test_split_raises_when_rpc_fails(self, mock_create_pdi, mock_create_person): # No ORM fallback: an RPC failure propagates and nothing is published. with fake_personhog_client() as fake: diff --git a/posthog/personhog_client/interceptor.py b/posthog/personhog_client/interceptor.py index 2b9b014ee367..e64d6021d378 100644 --- a/posthog/personhog_client/interceptor.py +++ b/posthog/personhog_client/interceptor.py @@ -179,16 +179,6 @@ def _is_db_timeout(exc: grpc.RpcError) -> bool: return any(key == _DB_TIMEOUT_METADATA_KEY for key, _ in exc.trailing_metadata() or ()) -def is_db_timeout_error(exc: BaseException) -> bool: - """Whether ``exc``, or the error it was raised from, is a database timeout, which the server rolled back.""" - current: BaseException | None = exc - while current is not None: - if isinstance(current, grpc.RpcError): - return callable(getattr(current, "trailing_metadata", None)) and _is_db_timeout(current) - current = current.__cause__ - return False - - def is_transient_rpc_error(exc: BaseException, codes: Collection[grpc.StatusCode] = _RETRYABLE_CODES) -> bool: """Whether ``exc``, or the error it was raised from, is a gRPC failure with a status in ``codes``.