From 6b187060de0d9fa117493d101137d419e04f6889 Mon Sep 17 00:00:00 2001 From: let5sne Date: Sun, 12 Jul 2026 04:03:16 +0800 Subject: [PATCH] fix(personal): retain upload cleanup tombstones --- .../service/PersonalIngestionService.java | 80 ++++++++++++++--- .../support/PersonalKnowledgeProperties.java | 1 + .../PersonalIngestionServiceTest.java | 85 ++++++++++++++++--- .../personal/PersonalSpaceServiceTest.java | 1 + 4 files changed, 143 insertions(+), 24 deletions(-) diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java index 7cd419c3..1b8209f2 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/service/PersonalIngestionService.java @@ -260,13 +260,21 @@ public class PersonalIngestionService { for (Map row : staleUploadRows("CLEANING", cleaningCutoff)) { finalizeStaleCleanup(intent(row), cleaningCutoff); } + LocalDateTime tombstoneCutoff = now.minusMinutes(properties.getUploadTombstoneRetentionMinutes()); + for (Map row : tombstoneRows()) { + UploadIntent intent = intent(row); + if (deleteKnownObject(intent) && uploadUpdatedAt(row).isBefore(tombstoneCutoff)) { + phaseTransaction.executeWithoutResult(status -> deleteTombstoneMetadata(intent)); + } + } } private List> staleUploadRows(String state, LocalDateTime cutoff) { return jdbcTemplate.queryForList(""" select i.tenant_id, i.owner_user_id, i.space_id, i.id item_id, i.oss_id, i.size_bytes, i.mime_type, o.file_name, o.service, - json_unquote(json_extract(o.ext1, '$.uploadToken')) upload_token + json_unquote(json_extract(o.ext1, '$.uploadToken')) upload_token, + o.update_time upload_updated_at from aihr_personal_item i join sys_oss o on o.oss_id = i.oss_id and binary o.tenant_id = binary i.tenant_id and o.create_by = i.owner_user_id @@ -278,6 +286,22 @@ public class PersonalIngestionService { """, cutoff, state); } + private List> tombstoneRows() { + return jdbcTemplate.queryForList(""" + select i.tenant_id, i.owner_user_id, i.space_id, i.id item_id, i.oss_id, + i.size_bytes, i.mime_type, o.file_name, o.service, + json_unquote(json_extract(o.ext1, '$.uploadToken')) upload_token, + o.update_time upload_updated_at + from sys_oss o + join aihr_personal_item i on i.oss_id = o.oss_id + and binary i.tenant_id = binary o.tenant_id and i.owner_user_id = o.create_by + where i.status = 'DELETED' + and json_unquote(json_extract(o.ext1, '$.uploadState')) = 'TOMBSTONE' + and json_unquote(json_extract(o.ext1, '$.uploadToken')) is not null + order by o.update_time + """); + } + private void beginCleanup(UploadIntent intent, LocalDateTime cutoff) { Integer claimed = phaseTransaction.execute(status -> claimState( intent, UploadState.PENDING, UploadState.CLEANING, cutoff)); @@ -286,7 +310,8 @@ public class PersonalIngestionService { return; } UploadState current = uploadState(intent); - if (current == UploadState.CLEANING || current == UploadState.MISSING) { + if (current == UploadState.CLEANING || current == UploadState.TOMBSTONE + || current == UploadState.MISSING) { deleteKnownObject(intent); } } @@ -331,7 +356,8 @@ public class PersonalIngestionService { UploadState current = uploadState(intent); if (current == UploadState.PENDING) { beginCleanup(intent, null); - } else if (current == UploadState.CLEANING || current == UploadState.MISSING) { + } else if (current == UploadState.CLEANING || current == UploadState.TOMBSTONE + || current == UploadState.MISSING) { deleteKnownObject(intent); } } @@ -386,24 +412,43 @@ public class PersonalIngestionService { and used_bytes >= ? and item_count > 0 """, intent.sizeBytes(), intent.owner().tenantId(), intent.owner().userId(), intent.spaceId(), intent.sizeBytes()); - int ossDeleted = jdbcTemplate.update(""" - delete from sys_oss + int tombstoned = jdbcTemplate.update(""" + update sys_oss + set ext1 = json_set(ext1, '$.uploadState', 'TOMBSTONE'), url = '', update_time = now() where tenant_id = ? and oss_id = ? and create_by = ? and file_name = ? and json_unquote(json_extract(ext1, '$.itemId')) = ? and json_unquote(json_extract(ext1, '$.uploadToken')) = ? and json_unquote(json_extract(ext1, '$.uploadState')) = 'CLEANING' + and service = ? """, intent.owner().tenantId(), intent.ossId(), intent.owner().userId(), intent.objectKey(), - String.valueOf(intent.itemId()), intent.uploadToken()); - if (itemDeleted != 1 || counterUpdated != 1 || ossDeleted != 1) { + String.valueOf(intent.itemId()), intent.uploadToken(), intent.serviceKey()); + if (itemDeleted != 1 || counterUpdated != 1 || tombstoned != 1) { throw new ServiceException("PERSONAL_UPLOAD_CLEANUP_FAILED"); } } + private void deleteTombstoneMetadata(UploadIntent intent) { + jdbcTemplate.update(""" + delete from sys_oss + where tenant_id = ? and oss_id = ? and create_by = ? and file_name = ? + and json_unquote(json_extract(ext1, '$.itemId')) = ? + and json_unquote(json_extract(ext1, '$.uploadToken')) = ? + and json_unquote(json_extract(ext1, '$.uploadState')) = 'TOMBSTONE' + and service = ? + """, intent.owner().tenantId(), intent.ossId(), intent.owner().userId(), intent.objectKey(), + String.valueOf(intent.itemId()), intent.uploadToken(), intent.serviceKey()); + } + private ItemCreatedResponse duplicate(PersonalOwner owner, long spaceId, String hash) { List> rows = jdbcTemplate.queryForList(""" - select id, status from aihr_personal_item - where tenant_id = ? and owner_user_id = ? and space_id = ? and content_hash = ? - and status <> 'DELETED' order by id desc limit 1 + select i.id, i.status + from aihr_personal_item i + join sys_oss o on o.oss_id = i.oss_id and binary o.tenant_id = binary i.tenant_id + and o.create_by = i.owner_user_id + where i.tenant_id = ? and i.owner_user_id = ? and i.space_id = ? and i.content_hash = ? + and i.status <> 'DELETED' + and json_unquote(json_extract(o.ext1, '$.uploadState')) = 'READY' + order by i.id desc limit 1 """, owner.tenantId(), owner.userId(), spaceId, hash); if (rows.isEmpty()) return null; long id = ((Number) rows.get(0).get("id")).longValue(); @@ -444,6 +489,13 @@ public class PersonalIngestionService { return number.longValue(); } + private static LocalDateTime uploadUpdatedAt(Map row) { + Object value = row.get("upload_updated_at"); + if (value instanceof LocalDateTime dateTime) return dateTime; + if (value instanceof java.sql.Timestamp timestamp) return timestamp.toLocalDateTime(); + throw new ServiceException("PERSONAL_UPLOAD_CLEANUP_FAILED"); + } + private void validateFile(MultipartFile file) { if (file == null || file.isEmpty() || file.getSize() <= 0) throw new ServiceException("PERSONAL_FILE_EMPTY"); validateSize(file.getSize()); @@ -466,8 +518,12 @@ public class PersonalIngestionService { } private static void validateRecoveryWindows(PersonalKnowledgeProperties properties) { + long cleanupWindow = (long) properties.getUploadCleanupAgeMinutes() + + properties.getCleanupFinalizeGraceMinutes(); if (properties.getUploadCleanupAgeMinutes() < MIN_UPLOAD_CLEANUP_AGE_MINUTES - || properties.getCleanupFinalizeGraceMinutes() <= 0) { + || properties.getCleanupFinalizeGraceMinutes() <= 0 + || properties.getUploadTombstoneRetentionMinutes() < 60 + || properties.getUploadTombstoneRetentionMinutes() <= cleanupWindow) { throw new IllegalArgumentException("invalid personal upload recovery windows"); } } @@ -557,7 +613,7 @@ public class PersonalIngestionService { } private record PhaseOne(ItemCreatedResponse duplicate, UploadIntent intent) {} - private enum UploadState { PENDING, READY, CLEANING, MISSING } + private enum UploadState { PENDING, READY, CLEANING, TOMBSTONE, MISSING } private record UploadIntent(PersonalOwner owner, long spaceId, long itemId, long ossId, String objectKey, String suffix, String mimeType, long sizeBytes, String serviceKey, diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/support/PersonalKnowledgeProperties.java b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/support/PersonalKnowledgeProperties.java index 378bce29..f5dd4fd5 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/support/PersonalKnowledgeProperties.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/main/java/org/dromara/aihr/personal/support/PersonalKnowledgeProperties.java @@ -23,4 +23,5 @@ public class PersonalKnowledgeProperties { private int maxParseAttempts = 3; private int uploadCleanupAgeMinutes = 15; private int cleanupFinalizeGraceMinutes = 15; + private int uploadTombstoneRetentionMinutes = 1440; } diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java index 242ba6b0..0790a16b 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalIngestionServiceTest.java @@ -115,7 +115,7 @@ class PersonalIngestionServiceTest { } @Test - void duplicateIsResolvedUnderOwnerLockWithoutCreatingUploadIntent() { + void duplicateIsResolvedOnlyFromReadyUploadUnderOwnerLock() { Fixture fixture = fixture(); when(fixture.spaces.reserve(OWNER, 4L)).thenReturn(7L); when(fixture.jdbc.queryForList(contains("content_hash"), eq("000000"), eq(101L), eq(7L), anyString())) @@ -126,6 +126,9 @@ class PersonalIngestionServiceTest { assertEquals(77L, response.itemId()); assertEquals(77L, response.duplicateOf()); + ArgumentCaptor dedupeSql = ArgumentCaptor.forClass(String.class); + verify(fixture.jdbc).queryForList(dedupeSql.capture(), eq("000000"), eq(101L), eq(7L), anyString()); + assertTrue(dedupeSql.getValue().contains("$.uploadState')) = 'READY'")); verify(fixture.store).requirePrivateService(); verify(fixture.store, never()).uploadPhysical(anyString(), anyString(), anyString(), any(byte[].class)); verify(fixture.jdbc, never()).update(contains("insert into sys_oss"), any(), any(), any(), any(), any(), @@ -162,13 +165,14 @@ class PersonalIngestionServiceTest { } @Test - void staleCleanupIsIdempotentAndDoesNotDecrementCountersTwice() { + void cleaningFinalizeCompensatesOnceAndRetainsTombstoneIntent() { Fixture fixture = fixture(); Map stale = staleIntent(); when(fixture.jdbc.queryForList(contains("uploadState')) = ?"), any(LocalDateTime.class), eq("PENDING"))) - .thenReturn(List.of(stale), List.of(), List.of()); + .thenReturn(List.of(stale), List.of()); when(fixture.jdbc.queryForList(contains("uploadState')) = ?"), any(LocalDateTime.class), eq("CLEANING"))) - .thenReturn(List.of(), List.of(stale), List.of(stale)); + .thenReturn(List.of(), List.of(stale)); + when(fixture.jdbc.queryForList(contains("uploadState')) = 'TOMBSTONE'"))).thenReturn(List.of()); when(fixture.jdbc.update(contains("json_set"), eq("CLEANING"), eq("000000"), eq(101L), eq(101L), eq("personal/key.txt"), eq("100"), eq("upload-token"), eq("PENDING"), any(LocalDateTime.class))) .thenReturn(1); @@ -178,13 +182,41 @@ class PersonalIngestionServiceTest { when(fixture.spaces.lockForUpdate(any(PersonalOwner.class))).thenReturn(7L); stubFinalizeCleanup(fixture); - fixture.service.recoverStaleUploadIntents(); fixture.service.recoverStaleUploadIntents(); fixture.service.recoverStaleUploadIntents(); verify(fixture.store, times(2)).deletePhysical("personal-private", "personal/key.txt"); verify(fixture.jdbc, times(1)).update(contains("used_bytes = used_bytes -"), eq(4L), eq("000000"), eq(101L), eq(7L), eq(4L)); + verify(fixture.jdbc).update(contains("'$.uploadState', 'TOMBSTONE'"), eq("000000"), eq(101L), + eq(101L), eq("personal/key.txt"), eq("100"), eq("upload-token"), eq("personal-private")); + verify(fixture.jdbc, never()).update(contains("delete from sys_oss"), any(), any(), any(), any(), any(), + any(), any()); + } + + @Test + void tombstoneIsDeletedOnEveryScanAndMetadataRemovedOnlyAfterRetention() { + Fixture fixture = fixture(); + Map fresh = staleIntent(LocalDateTime.now()); + Map expired = staleIntent(LocalDateTime.now().minusDays(2)); + when(fixture.jdbc.queryForList(contains("uploadState')) = ?"), any(LocalDateTime.class), eq("PENDING"))) + .thenReturn(List.of()); + when(fixture.jdbc.queryForList(contains("uploadState')) = ?"), any(LocalDateTime.class), eq("CLEANING"))) + .thenReturn(List.of()); + when(fixture.jdbc.queryForList(contains("uploadState')) = 'TOMBSTONE'"))) + .thenReturn(List.of(fresh), List.of(fresh), List.of(expired)); + when(fixture.jdbc.update(contains("delete from sys_oss"), eq("000000"), eq(101L), eq(101L), + eq("personal/key.txt"), eq("100"), eq("upload-token"), eq("personal-private"))).thenReturn(1); + + fixture.service.recoverStaleUploadIntents(); + fixture.service.recoverStaleUploadIntents(); + fixture.service.recoverStaleUploadIntents(); + + verify(fixture.store, times(3)).deletePhysical("personal-private", "personal/key.txt"); + verify(fixture.jdbc, times(1)).update(contains("delete from sys_oss"), eq("000000"), eq(101L), eq(101L), + eq("personal/key.txt"), eq("100"), eq("upload-token"), eq("personal-private")); + verify(fixture.jdbc, never()).update(contains("used_bytes = used_bytes -"), any(), any(), any(), any(), + any()); } @Test @@ -310,6 +342,23 @@ class PersonalIngestionServiceTest { any()); } + @Test + void tombstoneActivationFailureDeletesKnownLateObjectWithoutReclaimingIntent() { + Fixture fixture = fixture(); + stubPhaseOne(fixture, 4L); + when(fixture.store.uploadPhysical(eq("personal-private"), anyString(), eq("text/plain"), any(byte[].class))) + .thenReturn("https://private.invalid/object"); + when(fixture.jdbc.queryForList(contains("select json_unquote"), eq("000000"), eq(101L), eq(101L), + anyString(), eq("100"), anyString())).thenReturn(List.of(Map.of("upload_state", "TOMBSTONE"))); + + assertThrows(ServiceException.class, () -> fixture.service.createFile(OWNER, + new MockMultipartFile("file", "notes.txt", "text/plain", new byte[]{1, 2, 3, 4}), null, null)); + + verify(fixture.store).deletePhysical(eq("personal-private"), anyString()); + verify(fixture.jdbc, never()).update(contains("json_set"), any(), any(), any(), any(), any(), any(), any(), + any()); + } + @Test void unsafeCleanupWindowsAreRejected() { PersonalKnowledgeProperties properties = new PersonalKnowledgeProperties(); @@ -318,6 +367,13 @@ class PersonalIngestionServiceTest { properties.setUploadCleanupAgeMinutes(15); properties.setCleanupFinalizeGraceMinutes(0); assertThrows(IllegalArgumentException.class, () -> fixture(properties)); + properties.setCleanupFinalizeGraceMinutes(15); + properties.setUploadTombstoneRetentionMinutes(30); + assertThrows(IllegalArgumentException.class, () -> fixture(properties)); + properties.setUploadCleanupAgeMinutes(5); + properties.setCleanupFinalizeGraceMinutes(1); + properties.setUploadTombstoneRetentionMinutes(59); + assertThrows(IllegalArgumentException.class, () -> fixture(properties)); } private static Fixture fixture() { @@ -356,16 +412,21 @@ class PersonalIngestionServiceTest { eq(101L))).thenReturn(1); when(fixture.jdbc.update(contains("used_bytes = used_bytes -"), eq(4L), eq("000000"), eq(101L), eq(7L), eq(4L))).thenReturn(1); - when(fixture.jdbc.update(contains("delete from sys_oss"), eq("000000"), eq(101L), eq(101L), - anyString(), eq("100"), eq("upload-token"))).thenReturn(1); + when(fixture.jdbc.update(contains("'$.uploadState', 'TOMBSTONE'"), eq("000000"), eq(101L), eq(101L), + anyString(), eq("100"), eq("upload-token"), eq("personal-private"))).thenReturn(1); } private static Map staleIntent() { - return Map.of( - "tenant_id", "000000", "owner_user_id", 101L, "space_id", 7L, "item_id", 100L, - "oss_id", 101L, "size_bytes", 4L, "mime_type", "text/plain", - "file_name", "personal/key.txt", "service", "personal-private", - "upload_token", "upload-token" + return staleIntent(LocalDateTime.now()); + } + + private static Map staleIntent(LocalDateTime updatedAt) { + return Map.ofEntries( + Map.entry("tenant_id", "000000"), Map.entry("owner_user_id", 101L), Map.entry("space_id", 7L), + Map.entry("item_id", 100L), Map.entry("oss_id", 101L), Map.entry("size_bytes", 4L), + Map.entry("mime_type", "text/plain"), Map.entry("file_name", "personal/key.txt"), + Map.entry("service", "personal-private"), Map.entry("upload_token", "upload-token"), + Map.entry("upload_updated_at", updatedAt) ); } diff --git a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalSpaceServiceTest.java b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalSpaceServiceTest.java index 57df43ec..65f5fc7c 100644 --- a/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalSpaceServiceTest.java +++ b/backend/ruoyi-modules/ruoyi-aihr/src/test/java/org/dromara/aihr/personal/PersonalSpaceServiceTest.java @@ -56,6 +56,7 @@ class PersonalSpaceServiceTest { assertEquals(3, properties.getMaxParseAttempts()); assertEquals(15, properties.getUploadCleanupAgeMinutes()); assertEquals(15, properties.getCleanupFinalizeGraceMinutes()); + assertEquals(1440, properties.getUploadTombstoneRetentionMinutes()); } @Test