fix(sync): deliver snapshot tombstones across scopes

Snapshot publication follows the owning session's current project, while the deletion journal retains the immutable source project. Filtering deletions by that source label can strand identity rows in a destination that legitimately published the moved session.

Deliver snapshot tombstones to every revision scope and let PostgreSQL publication ownership or the scoped DuckDB mirror decide whether the row is present.

Author: wesm

internal/db/project_identity.go

@@ -57,7 +57,7 @@ type ProjectIdentityObservationKey struct {
 }
 
 // SessionProjectIdentitySnapshotKey identifies one immutable snapshot that a
-// downstream publication must remove. Project is retained to apply filters.
+// downstream publication must remove.
 type SessionProjectIdentitySnapshotKey struct {
 	SessionID string
 	Project   string
@@ -73,8 +73,11 @@ type ProjectIdentityPublicationDelta struct {
 }
 
 // LoadProjectIdentityPublicationDelta returns the compact identity changes in
-// (afterRevision, throughRevision]. Project filters are applied to both current
-// rows and tombstones so filtered targets can maintain independent cursors.
+// (afterRevision, throughRevision]. Project filters apply to current rows and
+// aggregate-observation tombstones. Snapshot tombstones remain unfiltered
+// because current rows are scoped by the owning session project while their
+// deletion journal retains the immutable snapshot project. Destinations decide
+// whether a tombstone applies from their resident rows or publication owners.
 func (db *DB) LoadProjectIdentityPublicationDelta(
 	ctx context.Context,
 	afterRevision, throughRevision int64,
@@ -223,7 +226,7 @@ func (db *DB) LoadProjectIdentityPublicationDelta(
 
 	tombstoneWhere, tombstoneArgs := projectIdentityPublicationChangeWhere(
 		"c", "c.project", afterRevision, throughRevision,
-		projects, excludeProjects,
+		nil, nil,
 	)
 	rows, err = db.getReader().QueryContext(ctx, `
 		SELECT c.session_id, c.project

internal/db/project_identity_test.go

@@ -225,7 +225,8 @@ func TestLoadProjectIdentityPublicationDeltaReturnsRowsAndTombstones(
 	assert.Empty(t, betaDelta.Observations)
 	assert.Equal(t, delta.ObservationDeletes, betaDelta.ObservationDeletes)
 	assert.Empty(t, betaDelta.Snapshots)
-	assert.Empty(t, betaDelta.SnapshotDeletes)
+	assert.Equal(t, delta.SnapshotDeletes, betaDelta.SnapshotDeletes,
+		"snapshot tombstones must reach scopes selected by the live session project")
 }
 
 func TestCopyArchiveIdentityFromPreservesLogicalArchiveAndNewGeneration(

internal/duckdb/sync_test.go

@@ -1660,7 +1660,6 @@ func TestFilteredIncrementalPushPublishesMovedSessionSourceSnapshot(t *testing.T
 	require.NoError(t, err)
 	mirror, err := OpenReadOnly(path)
 	require.NoError(t, err)
-	t.Cleanup(func() { require.NoError(t, mirror.Close()) })
 
 	var gotProject string
 	require.NoError(t, mirror.QueryRowContext(ctx,
@@ -1675,6 +1674,19 @@ func TestFilteredIncrementalPushPublishesMovedSessionSourceSnapshot(t *testing.T
 	).Scan(&snapshotProject))
 	assert.Equal(t, sourceProject, snapshotProject,
 		"the immutable snapshot keeps its source label while scope follows the session")
+	require.NoError(t, mirror.Close())
+
+	require.NoError(t, local.DeleteSession(sessionID))
+	deleted, err := Push(ctx, path, local, duckPushMachine, opts, false, nil)
+	require.NoError(t, err)
+	assert.False(t, deleted.Diagnostics.Full)
+	mirror, err = OpenReadOnly(path)
+	require.NoError(t, err)
+	t.Cleanup(func() { require.NoError(t, mirror.Close()) })
+	assertDuckDBCountWhere(t, mirror,
+		"source_session_project_identity_snapshots",
+		"source_session_id = ?", sessionID, 0,
+	)
 }
 
 func TestSyncPreservesAmbiguousIdentityAlongsideResolvedRemote(t *testing.T) {

internal/postgres/push_pgtest_test.go

@@ -516,6 +516,17 @@ func TestPushProjectMoveReconcilesFilteredScope(t *testing.T) {
 				SELECT COUNT(*) FROM source_session_project_identity_snapshots
 				WHERE project = $1`, targetProject).Scan(&count))
 			assert.Zero(t, count, "immutable snapshot must remain source-labelled")
+			if tc.wantSnapshot {
+				require.NoError(t, local.DeleteSession(sessionID))
+				_, err = syncer.Push(ctx, false, nil)
+				require.NoError(t, err)
+				require.NoError(t, syncer.pg.QueryRowContext(ctx, `
+					SELECT COUNT(*)
+					FROM source_session_project_identity_snapshots
+					WHERE source_session_id = $1`, sessionID).Scan(&count))
+				assert.Zero(t, count,
+					"hard deletion must remove the filtered source-labelled snapshot")
+			}
 		})
 	}
 }