Fix Gap 4: asset-level deletion invisible to backfill staleness checks (#612)

rakhiagr · claude · web-flow · commit aa6ba4ed5ca8 · 2026-04-15T15:14:17.000-04:00
* Fix Gap 4: asset-level deletion invisible to backfill staleness checks

When an entity is deleted via softDeleteAsset(), deleted_ts is set on the
entity row but batchGetUnion() unconditionally filters deleted_ts IS NULL,
making deleted rows invisible. getLatest() returns null (as if entity never
existed), allowing stale backfill events to resurrect deleted entities.

Fix:
- SQLStatementUtils.createAspectReadSql(): only append deleted_ts IS NULL
  when includeSoftDeleted=false (was unconditional before)
- SQL_READ_ASPECT_WITH_SOFT_DELETED_TEMPLATE: add deleted_ts to SELECT
- EBeanDAOUtils.readSqlRow(): when deleted_ts is non-null, mark aspect as
  soft-deleted using deleted_ts as the deletion timestamp

See docs/backfill-gap4-asset-level-deletion.md for detailed explanation.

Co-Authored-By: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;

* Fix Gap 4: ensure asset-level deletion clears deleted_ts on legitimate writes

* clarify soft delete behavior in documentation and tests

---------

Co-authored-by: Claude Opus 4.6 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/utils/EBeanDAOUtils.java b/dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/utils/EBeanDAOUtils.java
@@ -382,6 +382,11 @@ private static <ASPECT extends RecordTemplate> EbeanMetadataAspect readSqlRow(Sq
     } catch (URISyntaxException e) {
       throw new RuntimeException("Invalid urn format: " + urn, e);
     }
+    // Check for asset-level deletion: deleted_ts is non-null means the entire entity was deleted
+    // via softDeleteAsset(). deleted_ts is only in the SELECT when includeSoftDeleted=true.
+    final Timestamp assetDeletedTs = sqlRow.keySet().contains("deleted_ts") ? sqlRow.getTimestamp("deleted_ts") : null;
+    final boolean isAssetLevelDeleted = assetDeletedTs != null;
+
     if (isSoftDeletedAspect(sqlRow, columnName)) {
       primaryKey = new EbeanMetadataAspect.PrimaryKey(urn, aspectClass.getCanonicalName(), LATEST_VERSION);
 
@@ -399,6 +404,14 @@ private static <ASPECT extends RecordTemplate> EbeanMetadataAspect readSqlRow(Sq
       ebeanMetadataAspect.setCreatedOn(deletionTimestamp);
       ebeanMetadataAspect.setCreatedFor(sqlRow.getString("createdfor"));
       ebeanMetadataAspect.setMetadata(sqlRow.getString(columnName));
+    } else if (isAssetLevelDeleted) {
+      // Asset-level deletion: aspect column still has its value but entity row has deleted_ts set.
+      // Mark as soft-deleted using deleted_ts as the deletion timestamp.
+      primaryKey = new EbeanMetadataAspect.PrimaryKey(urn, aspectClass.getCanonicalName(), LATEST_VERSION);
+      ebeanMetadataAspect.setCreatedBy(sqlRow.getString("lastmodifiedby"));
+      ebeanMetadataAspect.setCreatedOn(assetDeletedTs);
+      ebeanMetadataAspect.setCreatedFor(sqlRow.getString("createdfor"));
+      ebeanMetadataAspect.setMetadata(DELETED_VALUE);
     } else {
       AuditedAspect auditedAspect = RecordUtils.toRecordTemplate(AuditedAspect.class, sqlRow.getString(columnName));
       primaryKey = new EbeanMetadataAspect.PrimaryKey(urn, auditedAspect.getCanonicalName(), LATEST_VERSION);
diff --git a/dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/utils/SQLStatementUtils.java b/dao-impl/ebean-dao/src/main/java/com/linkedin/metadata/dao/utils/SQLStatementUtils.java
@@ -104,20 +104,22 @@ public class SQLStatementUtils {
   // "JSON_EXTRACT(%s, '$.gma_deleted') IS NOT NULL" is used to exclude soft-deleted entity which has no lastmodifiedon.
   // for details, see the known limitations on https://github.com/linkedin/datahub-gma/pull/311. Same reason for
   // SQL_UPDATE_ASPECT_WITH_URN_TEMPLATE
+  // All UPDATE templates include deleted_ts = NULL to ensure any successful write revives an asset-deleted entity.
+  // Without this, the optimistic locking UPDATE path leaves deleted_ts set, making the entity invisible to reads.
   private static final String SQL_UPDATE_ASPECT_TEMPLATE =
-      "UPDATE %s SET %s = :metadata, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby "
+      "UPDATE %s SET %s = :metadata, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby, deleted_ts = NULL "
           + "WHERE urn = :urn and (JSON_EXTRACT(%s, '$.lastmodifiedon') = :oldTimestamp OR JSON_EXTRACT(%s, '$.gma_deleted') IS NOT NULL);";
 
   private static final String SQL_UPDATE_ASPECT_WITH_URN_TEMPLATE =
-      "UPDATE %s SET %s = :metadata, a_urn = :a_urn, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby "
+      "UPDATE %s SET %s = :metadata, a_urn = :a_urn, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby, deleted_ts = NULL "
           + "WHERE urn = :urn and (JSON_EXTRACT(%s, '$.lastmodifiedon') = :oldTimestamp OR JSON_EXTRACT(%s, '$.gma_deleted') IS NOT NULL);";
 
   private static final String SQL_UPDATE_ASPECT_TEMPLATE_WITH_SOFT_DELETE_OVERWRITE =
-      "UPDATE %s SET %s = :metadata, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby "
+      "UPDATE %s SET %s = :metadata, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby, deleted_ts = NULL "
           + "WHERE urn = :urn ;";
 
   private static final String SQL_UPDATE_ASPECT_WITH_URN_TEMPLATE_WITH_SOFT_DELETE_OVERWRITE = "UPDATE %s SET %s = "
-      + ":metadata, a_urn = :a_urn, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby WHERE urn = :urn;";
+      + ":metadata, a_urn = :a_urn, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby, deleted_ts = NULL WHERE urn = :urn;";
 
   private static final String SQL_READ_ASPECT_TEMPLATE =
       String.format("SELECT urn, %%s, lastmodifiedon, lastmodifiedby FROM %%s WHERE %s AND urn IN (", SOFT_DELETED_CHECK);
@@ -137,7 +139,7 @@ public class SQLStatementUtils {
           + "as _total_count FROM %%s WHERE %s LIMIT %%s OFFSET %%s", NONNULL_CHECK,  NONNULL_CHECK);
 
   private static final String SQL_READ_ASPECT_WITH_SOFT_DELETED_TEMPLATE =
-      "SELECT urn, %s, lastmodifiedon, lastmodifiedby FROM %s WHERE urn IN (";
+      "SELECT urn, %s, lastmodifiedon, lastmodifiedby, deleted_ts FROM %s WHERE urn IN (";
 
   private static final String INDEX_GROUP_BY_CRITERION = "SELECT count(*) as COUNT, %s FROM %s";
 
@@ -229,8 +231,10 @@ public static <ASPECT extends RecordTemplate> String createAspectReadSql(@Nonnul
     stringBuilder.append(String.format(sqlTemplate, columnName, tableName, columnName));
     stringBuilder.append(urnList);
     stringBuilder.append(RIGHT_PARENTHESIS);
-    stringBuilder.append(" AND ");
-    stringBuilder.append(DELETED_TS_IS_NULL_CHECK);
+    if (!includeSoftDeleted) {
+      stringBuilder.append(" AND ");
+      stringBuilder.append(DELETED_TS_IS_NULL_CHECK);
+    }
     return stringBuilder.toString();
   }
 
diff --git a/dao-impl/ebean-dao/src/test/java/com/linkedin/metadata/dao/EbeanLocalAccessTest.java b/dao-impl/ebean-dao/src/test/java/com/linkedin/metadata/dao/EbeanLocalAccessTest.java
@@ -924,4 +924,82 @@ public void testBatchSoftDeleteAssetsRejectsSqlInjection() {
     List<FooUrn> urns = Collections.singletonList(makeFooUrn(0));
     _ebeanLocalAccessFoo.batchSoftDeleteAssets(urns, "'; DROP TABLE metadata_entity_foo; --", false);
   }
+
+  // ===== Gap 4: Asset-level deletion visibility in batchGetUnion =====
+
+  @Test
+  public void testAssetDeletedEntityVisibleWithIncludeSoftDeleted() {
+    // Gap 4: After softDeleteAsset(), batchGetUnion(includeSoftDeleted=true) should return the row
+    // so that shouldBackfill() can detect the entity is deleted and reject stale backfills.
+    // softDeleteAsset() only sets deleted_ts; readSqlRow() synthesizes the soft-delete marker in memory.
+    FooUrn fooUrn = makeFooUrn(400);
+    AspectFoo aspectFoo = new AspectFoo().setValue("gap4test");
+    AuditStamp auditStamp = makeAuditStamp("actor", System.currentTimeMillis());
+
+    // Step 1: Create entity with an aspect
+    _ebeanLocalAccessFoo.add(fooUrn, aspectFoo, AspectFoo.class, auditStamp, null, false);
+
+    // Verify: aspect is readable normally
+    AspectKey<FooUrn, AspectFoo> aspectKey = new AspectKey<>(AspectFoo.class, fooUrn, 0L);
+    List<EbeanMetadataAspect> results =
+        _ebeanLocalAccessFoo.batchGetUnion(Collections.singletonList(aspectKey), 1000, 0, false, false);
+    assertEquals(1, results.size());
+    assertFalse(EBeanDAOUtils.isSoftDeletedMetadata(results.get(0).getMetadata()));
+
+    // Step 2: Asset-level delete (sets deleted_ts only; aspect columns are untouched at DB level)
+    int deleted = _ebeanLocalAccessFoo.softDeleteAsset(fooUrn, false);
+    assertEquals(1, deleted);
+
+    // Step 3: batchGetUnion with includeSoftDeleted=false should NOT return the row (deleted_ts filters it)
+    results = _ebeanLocalAccessFoo.batchGetUnion(Collections.singletonList(aspectKey), 1000, 0, false, false);
+    assertEquals(0, results.size());
+
+    // Step 4: batchGetUnion with includeSoftDeleted=true SHOULD return the row (Gap 4 fix —
+    // no longer filtered by deleted_ts IS NULL)
+    results = _ebeanLocalAccessFoo.batchGetUnion(Collections.singletonList(aspectKey), 1000, 0, true, false);
+    assertEquals(1, results.size());
+
+    // Step 5: The returned aspect should be marked as soft-deleted (readSqlRow() synthesizes {"gma_deleted": true})
+    EbeanMetadataAspect result = results.get(0);
+    assertEquals(fooUrn.toString(), result.getKey().getUrn());
+    assertTrue(EBeanDAOUtils.isSoftDeletedMetadata(result.getMetadata()));
+  }
+
+  @Test
+  public void testWriteToAssetDeletedEntityClearsDeletedTs() {
+    // Verify that a legitimate write to an asset-deleted entity clears deleted_ts,
+    // reviving the entity and making it visible to normal reads again.
+    FooUrn fooUrn = makeFooUrn(401);
+    AspectFoo aspectFoo = new AspectFoo().setValue("gap4_revive_test");
+    AuditStamp auditStamp = makeAuditStamp("actor", System.currentTimeMillis());
+
+    // Step 1: Create entity with an aspect
+    _ebeanLocalAccessFoo.add(fooUrn, aspectFoo, AspectFoo.class, auditStamp, null, false);
+
+    // Step 2: Asset-level delete
+    int deleted = _ebeanLocalAccessFoo.softDeleteAsset(fooUrn, false);
+    assertEquals(1, deleted);
+
+    // Verify deleted_ts is set in DB
+    SqlRow row = _server.createSqlQuery(
+        "SELECT deleted_ts FROM metadata_entity_foo WHERE urn = '" + fooUrn + "'").findOne();
+    assertNotNull(row.getTimestamp("deleted_ts"));
+
+    // Step 3: Write a new aspect value (simulates a legitimate, non-stale write)
+    AspectFoo updatedAspect = new AspectFoo().setValue("gap4_revived");
+    AuditStamp newAuditStamp = makeAuditStamp("actor", System.currentTimeMillis() + 1000);
+    _ebeanLocalAccessFoo.add(fooUrn, updatedAspect, AspectFoo.class, newAuditStamp, null, false);
+
+    // Step 4: Verify deleted_ts is cleared in DB
+    row = _server.createSqlQuery(
+        "SELECT deleted_ts FROM metadata_entity_foo WHERE urn = '" + fooUrn + "'").findOne();
+    assertNull(row.getTimestamp("deleted_ts"));
+
+    // Step 5: Entity should now be visible to normal reads (includeSoftDeleted=false)
+    AspectKey<FooUrn, AspectFoo> aspectKey = new AspectKey<>(AspectFoo.class, fooUrn, 0L);
+    List<EbeanMetadataAspect> results =
+        _ebeanLocalAccessFoo.batchGetUnion(Collections.singletonList(aspectKey), 1000, 0, false, false);
+    assertEquals(1, results.size());
+    assertFalse(EBeanDAOUtils.isSoftDeletedMetadata(results.get(0).getMetadata()));
+  }
 }
diff --git a/dao-impl/ebean-dao/src/test/java/com/linkedin/metadata/dao/utils/SQLStatementUtilsTest.java b/dao-impl/ebean-dao/src/test/java/com/linkedin/metadata/dao/utils/SQLStatementUtilsTest.java
@@ -146,12 +146,11 @@ public void testCreateAspectReadSql() {
             + "AND deleted_ts IS NULL";
     assertEquals(SQLStatementUtils.createAspectReadSql(AspectFoo.class, set, false, false), expectedSql);
 
-    //test when includedSoftDeleted is true
+    //test when includedSoftDeleted is true — should NOT filter deleted_ts, and should SELECT deleted_ts
     expectedSql =
-        "SELECT urn, a_aspectfoo, lastmodifiedon, lastmodifiedby "
+        "SELECT urn, a_aspectfoo, lastmodifiedon, lastmodifiedby, deleted_ts "
             + "FROM metadata_entity_foo "
-            + "WHERE urn IN ('urn:li:foo:1', 'urn:li:foo:2') "
-            + "AND deleted_ts IS NULL";
+            + "WHERE urn IN ('urn:li:foo:1', 'urn:li:foo:2')";
     assertEquals(SQLStatementUtils.createAspectReadSql(AspectFoo.class, set, true, false), expectedSql);
   }
 
@@ -707,18 +706,33 @@ public void testUpdateAspectWithOptimisticLockSql() {
     FooUrn fooUrn = makeFooUrn(1);
     String expectedSql =
         "UPDATE metadata_entity_foo SET a_aspectfoo = :metadata, a_urn = :a_urn, lastmodifiedon = :lastmodifiedon, "
-            + "lastmodifiedby = :lastmodifiedby WHERE urn = :urn and (JSON_EXTRACT(a_aspectfoo, '$.lastmodifiedon') = "
+            + "lastmodifiedby = :lastmodifiedby, deleted_ts = NULL WHERE urn = :urn and (JSON_EXTRACT(a_aspectfoo, '$.lastmodifiedon') = "
             + ":oldTimestamp OR JSON_EXTRACT(a_aspectfoo, '$.gma_deleted') IS NOT NULL);";
     assertEquals(SQLStatementUtils.createAspectUpdateWithOptimisticLockSql(fooUrn, AspectFoo.class, true, false, false),
         expectedSql);
 
     expectedSql =
         "UPDATE metadata_entity_foo SET a_aspectfoo = :metadata, lastmodifiedon = :lastmodifiedon, lastmodifiedby = "
-            + ":lastmodifiedby WHERE urn = :urn and (JSON_EXTRACT(a_aspectfoo, '$.lastmodifiedon') = :oldTimestamp "
+            + ":lastmodifiedby, deleted_ts = NULL WHERE urn = :urn and (JSON_EXTRACT(a_aspectfoo, '$.lastmodifiedon') = :oldTimestamp "
             + "OR JSON_EXTRACT(a_aspectfoo, '$.gma_deleted') IS NOT NULL);";
     assertEquals(
         SQLStatementUtils.createAspectUpdateWithOptimisticLockSql(fooUrn, AspectFoo.class, false, false, false),
         expectedSql);
+
+    // softDeleteOverwrite=true with urnExtraction — should include deleted_ts = NULL
+    expectedSql =
+        "UPDATE metadata_entity_foo SET a_aspectfoo = "
+            + ":metadata, a_urn = :a_urn, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby, deleted_ts = NULL WHERE urn = :urn;";
+    assertEquals(SQLStatementUtils.createAspectUpdateWithOptimisticLockSql(fooUrn, AspectFoo.class, true, false, true),
+        expectedSql);
+
+    // softDeleteOverwrite=true without urnExtraction — should include deleted_ts = NULL
+    expectedSql =
+        "UPDATE metadata_entity_foo SET a_aspectfoo = :metadata, lastmodifiedon = :lastmodifiedon, lastmodifiedby = :lastmodifiedby, deleted_ts = NULL "
+            + "WHERE urn = :urn ;";
+    assertEquals(
+        SQLStatementUtils.createAspectUpdateWithOptimisticLockSql(fooUrn, AspectFoo.class, false, false, true),
+        expectedSql);
   }
 
   @Test
diff --git a/docs/backfill-gap4-asset-level-deletion.md b/docs/backfill-gap4-asset-level-deletion.md
@@ -0,0 +1,121 @@
+# Gap 4: Asset-Level Deletion Invisible to Backfill Staleness Checks
+
+## Problem
+
+When an entire entity is deleted via `deleteAll()`, each aspect is first marked with `{"gma_deleted": true}` (via
+individual `delete()` calls), then `softDeleteAsset()` sets `deleted_ts = NOW()` on the entity row. The row becomes
+invisible to read queries because `batchGetUnion()` unconditionally filters `deleted_ts IS NULL` — even when
+`includeSoftDeleted=true`.
+
+When a stale backfill event (e.g., DLQ replay) arrives for an aspect on this deleted entity:
+
+1. `getLatest()` calls `batchGetUnion(..., includeSoftDeleted=true)` to fetch the current state
+2. `batchGetUnion()` generates SQL via `createAspectReadSql()`
+3. **Bug**: `createAspectReadSql()` unconditionally appends `AND deleted_ts IS NULL` regardless of the
+   `includeSoftDeleted` parameter
+4. The deleted row is filtered out, query returns empty
+5. `getLatest()` returns `AspectEntry(null, null)` — null aspect, `isSoftDeleted=false`
+6. `shouldBackfill()` sees `oldValue == null && !isSoftDeleted` → "aspect never existed" → allows backfill
+7. The write resets `deleted_ts` to NULL, **resurrecting the deleted entity** with stale data
+
+Note: Even though `deleteAll()` also sets aspect-level `{"gma_deleted": true}` markers (which PR #602's Gap 1 fix would
+catch), those markers are never reached because the entire row is invisible at the SQL level.
+
+## Root Cause
+
+**File**: `SQLStatementUtils.java`, method `createAspectReadSql()`
+
+```java
+stringBuilder.append(urnList);
+stringBuilder.append(RIGHT_PARENTHESIS);
+stringBuilder.append(" AND ");
+stringBuilder.append(DELETED_TS_IS_NULL_CHECK);  // Always appended — ignores includeSoftDeleted!
+```
+
+## Fix
+
+**One change in `SQLStatementUtils.createAspectReadSql()`**: make `AND deleted_ts IS NULL` conditional on
+`!includeSoftDeleted`:
+
+```java
+stringBuilder.append(urnList);
+stringBuilder.append(RIGHT_PARENTHESIS);
+if (!includeSoftDeleted) {
+    stringBuilder.append(" AND ");
+    stringBuilder.append(DELETED_TS_IS_NULL_CHECK);
+}
+```
+
+This matches the pattern already used correctly in `createListAspectByUrnSql()`.
+
+### Generated SQL (after fix)
+
+```sql
+-- includeSoftDeleted=false (normal reads — unchanged)
+SELECT urn, a_aspectfoo, lastmodifiedon, lastmodifiedby
+FROM metadata_entity_foo
+WHERE JSON_EXTRACT(a_aspectfoo, '$.gma_deleted') IS NULL
+AND urn IN ('urn:li:foo:1')
+AND deleted_ts IS NULL
+
+-- includeSoftDeleted=true (backfill path — now includes deleted rows + deleted_ts)
+SELECT urn, a_aspectfoo, lastmodifiedon, lastmodifiedby, deleted_ts
+FROM metadata_entity_foo
+WHERE urn IN ('urn:li:foo:1')
+```
+
+## How This Solves the Issue
+
+After the fix, when a stale backfill arrives for an asset-deleted entity:
+
+1. `getLatest()` calls `batchGetUnion(..., includeSoftDeleted=true)`
+2. `createAspectReadSql()` generates SQL **without** `deleted_ts IS NULL` filter
+3. The deleted row is returned — if `deleteAll()` was used, aspects have `{"gma_deleted": true}` markers which the
+   existing aspect-level soft-delete detection in `readSqlRow()` handles
+4. If `softDeleteAsset()` was called directly (without aspect markers), `readSqlRow()` detects `deleted_ts` is non-null
+   and marks the aspect as soft-deleted using `deleted_ts` as the deletion timestamp
+5. `getLatest()` returns `AspectEntry(null, extraInfo, isSoftDeleted=true)`
+6. `shouldBackfill()` (from PR #602) compares `emitTime > deletionTimestamp` → stale backfill rejected
+
+Additionally, all UPDATE SQL templates now include `deleted_ts = NULL` so that any successful write (including via the
+optimistic locking / changelog path in `saveLatest()`) properly revives an asset-deleted entity. Without this, the
+`saveLatest()` changelog path would write the aspect but leave `deleted_ts` set, making the entity invisible to
+subsequent reads.
+
+## Files Changed
+
+| File                                                                    | Change                                                                                           |
+| ----------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------ |
+| `dao-impl/ebean-dao/src/main/java/.../utils/SQLStatementUtils.java`     | Conditional `deleted_ts IS NULL` filter + `deleted_ts` in SELECT + `deleted_ts = NULL` in UPDATE |
+| `dao-impl/ebean-dao/src/main/java/.../utils/EBeanDAOUtils.java`         | Handle asset-level deletion in `readSqlRow()` when `deleted_ts` is non-null                      |
+| `dao-impl/ebean-dao/src/test/java/.../dao/EbeanLocalAccessTest.java`    | New test: asset-deleted entity visible with `includeSoftDeleted=true`                            |
+| `dao-impl/ebean-dao/src/test/java/.../utils/SQLStatementUtilsTest.java` | Update expected SQL for `includeSoftDeleted=true` and optimistic lock UPDATE                     |
+
+## Relationship to Other Gaps
+
+| Gap       | What                                                   | Status           |
+| --------- | ------------------------------------------------------ | ---------------- |
+| Gap 1     | `shouldBackfill()` blind to soft-deletes               | Fixed in PR #602 |
+| Gap 2     | Missing per-aspect deletion timestamp                  | Fixed in PR #602 |
+| Gap 3     | Old schema loses emitTime on deletion                  | Fixed by Gap 2   |
+| **Gap 4** | **Asset-level deletion invisible to staleness checks** | **This PR**      |
+
+Gap 4 depends on PR #602: once the row is visible (Gap 4 fix), the aspect-level `{"gma_deleted": true}` markers are
+detected by `readSqlRow()`, and PR #602's `shouldBackfill()` logic rejects stale backfills.
+
+## EI Verification (April 13, 2026)
+
+**Entity:** `urn:li:demo:rakagraw_gap4_test_004` **Environment:** ei-ltx1 (mg_db_1_ei)
+
+Confirmed the bug on current EI (without PR #602 or #612):
+
+1. Created entity with Status + DemoInfo
+2. Asset-level delete (`delete` with empty aspectTypes) — `deleted_ts` set, both aspects marked `{"gma_deleted": true}`
+3. GET returns NOT FOUND (correct)
+4. Stale backfill (emitTime=1000, before deletion) — **entity resurrected**:
+   - `deleted_ts` reset to NULL
+   - `a_demo_info` overwritten with stale data
+   - `a_status` still `{"gma_deleted": true}` (only backfilled aspect resurrected)
+
+Full test results:
+[Backfill Gap Fix Verification Results](https://docs.google.com/document/d/17XNuwhBBanPuT5w5Bo-40ozN5bhTSSqyHxSTYaiXybs/edit)
