feat: Lease 기반 멱등 Sync Dispatcher 구현

backend/src/main/java/com/opensource/docgrid/domain/embedding/repository/EmbeddingJobRepository.java

@@ -4,6 +4,7 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Optional;
+import java.util.UUID;
 
 import jakarta.persistence.LockModeType;
 
@@ -25,6 +26,13 @@
  */
 public interface EmbeddingJobRepository extends JpaRepository<EmbeddingJob, Long> {
 
+    Optional<EmbeddingJob> findBySourceEventId(UUID sourceEventId);
+
+    Optional<EmbeddingJob> findTopByDocumentVersionIdAndEmbeddingModelIdOrderByIdDesc(
+        Long documentVersionId,
+        Long embeddingModelId
+    );
+
     /**
      * 관리자 목록 화면에 필요한 연관관계를 함께 조회하면서 선택 필터와 Pagination을 적용한다.
      *

backend/src/main/java/com/opensource/docgrid/domain/embedding/repository/EmbeddingRepository.java

@@ -48,6 +48,19 @@ int markActiveAsStaleByDocumentVersionId(
         @Param("documentVersionId") Long documentVersionId
     );
 
+    /**
+     * 삭제 문서의 모든 ACTIVE Vector를 즉시 검색 대상에서 제외한다.
+     */
+    @Modifying(flushAutomatically = true)
+    @Query(value = """
+        UPDATE embeddings
+        SET status = 'STALE',
+            updated_at = CURRENT_TIMESTAMP
+        WHERE document_id = :documentId
+          AND status = 'ACTIVE'
+        """, nativeQuery = true)
+    int markActiveAsStaleByDocumentId(@Param("documentId") Long documentId);
+
     /**
      * 수동 재처리로 다시 인덱싱할 Version의 Embedding 행을 한 SQL로 제거한다.
      *

backend/src/main/java/com/opensource/docgrid/domain/permission/service/command/CollectionPermissionCommandService.java

@@ -27,6 +27,8 @@
 import com.opensource.docgrid.domain.user.repository.DepartmentRepository;
 import com.opensource.docgrid.domain.user.repository.RoleRepository;
 import com.opensource.docgrid.domain.user.repository.UserRepository;
+import com.opensource.docgrid.domain.sync.enums.SyncPermissionOperation;
+import com.opensource.docgrid.domain.sync.service.command.SyncEventWriter;
 import com.opensource.docgrid.global.exception.DocGridException;
 import com.opensource.docgrid.global.exception.ErrorCode;
 
@@ -46,6 +48,7 @@ public class CollectionPermissionCommandService {
     private final DepartmentRepository departmentRepository;
     private final PermissionConverter permissionConverter;
     private final PermissionQueryService permissionQueryService;
+    private final SyncEventWriter syncEventWriter;
 
     // 컬렉션 권한 부여
     public CollectionPermissionResponse grantPermission(Long collectionId, Long grantorId,
@@ -99,6 +102,13 @@ public CollectionPermissionResponse grantPermission(Long collectionId, Long gran
             updateCacheForCollection(collectionId, targetUser, permissions, permission.getId(), request.expiresAt());
         }
 
+        // 권한 원장과 같은 Transaction에 컬렉션 캐시 재투영 의도를 기록한다.
+        syncEventWriter.recordPermissionCacheRefresh(
+            AccessSourceType.DIRECT_COLLECTION_PERMISSION,
+            permission.getId(),
+            SyncPermissionOperation.GRANTED
+        );
+
         return permissionConverter.toCollectionPermissionResponse(permission);
     }
 
@@ -121,6 +131,11 @@ public void revokePermission(Long collectionId, Long permissionId, Long revokerI
         }
 
         collectionPermissionRepository.delete(permission);
+        syncEventWriter.recordPermissionCacheRefresh(
+            AccessSourceType.DIRECT_COLLECTION_PERMISSION,
+            permissionId,
+            SyncPermissionOperation.REVOKED
+        );
     }
 
     // targetType과 ID 필드 조합 유효성 검사

backend/src/main/java/com/opensource/docgrid/domain/permission/service/command/DocumentPermissionCommandService.java

@@ -22,6 +22,8 @@
 import com.opensource.docgrid.domain.user.repository.DepartmentRepository;
 import com.opensource.docgrid.domain.user.repository.RoleRepository;
 import com.opensource.docgrid.domain.user.repository.UserRepository;
+import com.opensource.docgrid.domain.sync.enums.SyncPermissionOperation;
+import com.opensource.docgrid.domain.sync.service.command.SyncEventWriter;
 import com.opensource.docgrid.global.exception.DocGridException;
 import com.opensource.docgrid.global.exception.ErrorCode;
 
@@ -40,6 +42,7 @@ public class DocumentPermissionCommandService {
     private final DepartmentRepository departmentRepository;
     private final PermissionConverter permissionConverter;
     private final PermissionQueryService permissionQueryService;
+    private final SyncEventWriter syncEventWriter;
 
     // 문서 단건 예외 권한 부여
     public DocumentPermissionResponse grantPermission(Long documentId, Long grantorId,
@@ -94,6 +97,13 @@ public DocumentPermissionResponse grantPermission(Long documentId, Long grantorI
                     AccessSourceType.DIRECT_DOCUMENT_PERMISSION, permission.getId(), request.expiresAt());
         }
 
+        // 권한 원장과 같은 Transaction에 캐시 재투영 의도를 남겨 후속 누락을 복구할 수 있게 한다.
+        syncEventWriter.recordPermissionCacheRefresh(
+            AccessSourceType.DIRECT_DOCUMENT_PERMISSION,
+            permission.getId(),
+            SyncPermissionOperation.GRANTED
+        );
+
         return permissionConverter.toDocumentPermissionResponse(permission);
     }
 
@@ -116,6 +126,11 @@ public void revokePermission(Long documentId, Long permissionId, Long revokerId)
         }
 
         documentPermissionRepository.delete(permission);
+        syncEventWriter.recordPermissionCacheRefresh(
+            AccessSourceType.DIRECT_DOCUMENT_PERMISSION,
+            permissionId,
+            SyncPermissionOperation.REVOKED
+        );
     }
 
     // targetType과 ID 필드 조합 유효성 검사

backend/src/main/java/com/opensource/docgrid/domain/sync/config/SyncDispatcherProperties.java

@@ -0,0 +1,69 @@
+package com.opensource.docgrid.domain.sync.config;
+
+import java.time.Duration;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+import org.springframework.validation.annotation.Validated;
+
+import jakarta.validation.constraints.AssertTrue;
+import jakarta.validation.constraints.Min;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * Sync Dispatcher의 실행 여부, Polling, Lease, 복구 Batch와 Retry Backoff 설정을 바인딩한다.
+ *
+ * <p>API 전용 인스턴스에서는 enabled를 false로 유지할 수 있고, 다중 Dispatcher는 서로 다른 name을
+ * 사용해 Event의 현재 소유 인스턴스를 운영 화면에서 식별한다.
+ */
+@Getter
+@Setter
+@Validated
+@Component
+@ConfigurationProperties(prefix = "sync.dispatcher")
+public class SyncDispatcherProperties {
+
+    private boolean enabled = false;
+
+    @NotBlank
+    private String name = "sync-dispatcher";
+
+    @NotNull
+    private Duration pollingInterval = Duration.ofSeconds(1);
+
+    @NotNull
+    private Duration leaseDuration = Duration.ofSeconds(30);
+
+    @NotNull
+    private Duration leaseRecoveryInterval = Duration.ofSeconds(10);
+
+    @Min(1)
+    private int leaseRecoveryBatchSize = 100;
+
+    @NotNull
+    private Duration retryInitialDelay = Duration.ofSeconds(5);
+
+    @NotNull
+    private Duration retryMaxDelay = Duration.ofMinutes(1);
+
+    @AssertTrue(message = "Sync Dispatcher의 Polling과 Lease 시간은 0보다 커야 합니다.")
+    public boolean isTimingValid() {
+        return isPositive(pollingInterval)
+            && isPositive(leaseDuration)
+            && isPositive(leaseRecoveryInterval);
+    }
+
+    @AssertTrue(message = "Sync Retry 최대 지연은 양수인 초기 지연보다 짧을 수 없습니다.")
+    public boolean isRetryDelayValid() {
+        return isPositive(retryInitialDelay)
+            && retryMaxDelay != null
+            && retryMaxDelay.compareTo(retryInitialDelay) >= 0;
+    }
+
+    private boolean isPositive(Duration duration) {
+        return duration != null && !duration.isZero() && !duration.isNegative();
+    }
+}

backend/src/main/java/com/opensource/docgrid/domain/sync/config/SyncSchedulingConfig.java

@@ -0,0 +1,14 @@
+package com.opensource.docgrid.domain.sync.config;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+/**
+ * Sync Dispatcher가 활성화된 실행 인스턴스에서만 Polling과 Lease Recovery Scheduler를 켠다.
+ */
+@Configuration
+@EnableScheduling
+@ConditionalOnProperty(prefix = "sync.dispatcher", name = "enabled", havingValue = "true")
+public class SyncSchedulingConfig {
+}

backend/src/main/java/com/opensource/docgrid/domain/sync/dto/ClaimedSyncEvent.java

@@ -0,0 +1,12 @@
+package com.opensource.docgrid.domain.sync.dto;
+
+import java.util.UUID;
+
+/**
+ * Dispatcher가 후속 처리 Transaction에 전달하는 최소 Event 소유권 Snapshot이다.
+ *
+ * <p>Payload와 도메인 상태는 Dispatch 시 DB에서 다시 읽고, 이 DTO는 Event ID와 현재 Claim Token만
+ * 전달해 오래된 Scheduler 실행이 새 소유권으로 작업하지 못하게 한다.
+ */
+public record ClaimedSyncEvent(UUID eventId, UUID claimToken) {
+}

backend/src/main/java/com/opensource/docgrid/domain/sync/entity/SyncOutboxEvent.java

@@ -141,4 +141,167 @@ public SyncOutboxEvent(
         this.occurredAt = occurredAt;
         this.maxRetryCount = maxRetryCount;
     }
+
+    /**
+     * 실행 가능한 PENDING Event에 Dispatcher Lease 소유권을 원자적으로 부여한다.
+     */
+    public void claim(
+        String dispatcherName,
+        UUID newClaimToken,
+        LocalDateTime claimedAt,
+        LocalDateTime newLockExpiresAt
+    ) {
+        // 1. 이미 Claim됐거나 끝난 Event의 소유권을 덮어쓰지 않는다.
+        if (status != SyncEventStatus.PENDING || availableAt == null || availableAt.isAfter(claimedAt)) {
+            throw new IllegalStateException("실행 가능한 PENDING Event만 Claim할 수 있습니다.");
+        }
+        // 2. 빈 소유자나 유효하지 않은 Lease가 영속화되지 않게 입력 계약을 검증한다.
+        if (dispatcherName == null
+            || dispatcherName.isBlank()
+            || newClaimToken == null
+            || claimedAt == null
+            || newLockExpiresAt == null
+            || !newLockExpiresAt.isAfter(claimedAt)) {
+            throw new IllegalArgumentException("유효한 Dispatcher 소유권과 Lease가 필요합니다.");
+        }
+
+        // 3. 처리 상태와 현재 Claim 세대의 소유권을 함께 반영한다.
+        status = SyncEventStatus.PROCESSING;
+        lockedBy = dispatcherName;
+        claimToken = newClaimToken;
+        lockExpiresAt = newLockExpiresAt;
+    }
+
+    /**
+     * 현재 Claim이 Handler 부작용까지 Commit할 준비가 됐을 때 Event를 완료한다.
+     */
+    public void complete(UUID currentClaimToken, LocalDateTime completedAt) {
+        validateActiveOwnership(currentClaimToken, completedAt);
+        status = SyncEventStatus.PROCESSED;
+        processedAt = completedAt;
+        lastErrorCode = null;
+        lastErrorMessage = null;
+        clearOwnership();
+    }
+
+    /**
+     * 장시간 Handler가 현재 Claim 세대를 유지한 채 Lease 만료 시각만 연장한다.
+     */
+    public void renewLease(
+        UUID currentClaimToken,
+        LocalDateTime renewedAt,
+        LocalDateTime renewedLockExpiresAt
+    ) {
+        validateActiveOwnership(currentClaimToken, renewedAt);
+        if (renewedLockExpiresAt == null || !renewedLockExpiresAt.isAfter(lockExpiresAt)) {
+            throw new IllegalArgumentException("새 Lease 만료 시각은 현재 Lease보다 늦어야 합니다.");
+        }
+        lockExpiresAt = renewedLockExpiresAt;
+    }
+
+    /**
+     * 현재 처리 실패를 기록하고 지정 시각 이후 다시 Claim 가능한 Queue 상태로 되돌린다.
+     */
+    public void scheduleRetry(
+        UUID currentClaimToken,
+        String errorCode,
+        String errorMessage,
+        LocalDateTime failedAt,
+        LocalDateTime nextAvailableAt
+    ) {
+        validateActiveOwnership(currentClaimToken, failedAt);
+        if (retryCount + 1 >= maxRetryCount || nextAvailableAt == null || nextAvailableAt.isBefore(failedAt)) {
+            throw new IllegalStateException("남은 Retry와 다음 실행 시각이 필요합니다.");
+        }
+        status = SyncEventStatus.PENDING;
+        retryCount++;
+        availableAt = nextAvailableAt;
+        lastErrorCode = errorCode;
+        lastErrorMessage = errorMessage;
+        clearOwnership();
+    }
+
+    /**
+     * Retry를 모두 소진한 현재 Claim을 최종 실패 상태로 종결한다.
+     */
+    public void markFailed(
+        UUID currentClaimToken,
+        String errorCode,
+        String errorMessage,
+        LocalDateTime failedAt
+    ) {
+        validateActiveOwnership(currentClaimToken, failedAt);
+        status = SyncEventStatus.FAILED;
+        retryCount++;
+        lastErrorCode = errorCode;
+        lastErrorMessage = errorMessage;
+        clearOwnership();
+    }
+
+    /**
+     * 관리자가 최종 실패 Event에 실행 기회 한 번을 추가해 Queue로 되돌린다.
+     */
+    public void requeueFailed(LocalDateTime requeuedAt) {
+        if (status != SyncEventStatus.FAILED || requeuedAt == null) {
+            throw new IllegalStateException("FAILED Event만 수동 재처리할 수 있습니다.");
+        }
+        status = SyncEventStatus.PENDING;
+        availableAt = requeuedAt;
+        maxRetryCount++;
+        lastErrorCode = null;
+        lastErrorMessage = null;
+        processedAt = null;
+        clearOwnership();
+    }
+
+    /**
+     * 만료된 PROCESSING Lease를 Retry Queue 또는 최종 실패 상태로 회수한다.
+     */
+    public void recoverExpiredLease(
+        String errorCode,
+        String errorMessage,
+        LocalDateTime recoveredAt,
+        LocalDateTime nextAvailableAt
+    ) {
+        // 1. 아직 유효하거나 이미 다른 흐름이 끝낸 Event를 오래된 복구 Snapshot으로 변경하지 않는다.
+        if (status != SyncEventStatus.PROCESSING
+            || claimToken == null
+            || lockExpiresAt == null
+            || recoveredAt == null
+            || lockExpiresAt.isAfter(recoveredAt)) {
+            throw new IllegalStateException("만료된 PROCESSING Event만 회수할 수 있습니다.");
+        }
+
+        // 2. 이번 만료를 실패 횟수에 반영하고 남은 기회에 따라 Queue 또는 최종 실패로 전환한다.
+        retryCount++;
+        lastErrorCode = errorCode;
+        lastErrorMessage = errorMessage;
+        if (retryCount >= maxRetryCount) {
+            status = SyncEventStatus.FAILED;
+        } else {
+            if (nextAvailableAt == null || nextAvailableAt.isBefore(recoveredAt)) {
+                throw new IllegalArgumentException("다음 실행 시각은 복구 시각보다 빠를 수 없습니다.");
+            }
+            status = SyncEventStatus.PENDING;
+            availableAt = nextAvailableAt;
+        }
+        clearOwnership();
+    }
+
+    private void validateActiveOwnership(UUID currentClaimToken, LocalDateTime operatedAt) {
+        if (status != SyncEventStatus.PROCESSING
+            || currentClaimToken == null
+            || !currentClaimToken.equals(claimToken)
+            || operatedAt == null
+            || lockExpiresAt == null
+            || !lockExpiresAt.isAfter(operatedAt)) {
+            throw new IllegalStateException("유효한 현재 Sync Event 소유권이 필요합니다.");
+        }
+    }
+
+    private void clearOwnership() {
+        lockedBy = null;
+        claimToken = null;
+        lockExpiresAt = null;
+    }
 }