feat(snapshot): parallel PK-range chunks with per-chunk resume

docs/adr/MODULE_SPLIT_DEFERRED.md

@@ -0,0 +1,85 @@
+# Module Architecture Decision
+
+## Current state (2026-08-12)
+
+`syncflow-api` contains orchestration classes (`SyncOrchestrator`, `SnapshotExecutor`,
+`CaptureLifecycle`, `DestinationRouter`) plus all JPA entities and repositories.
+
+The original architecture analysis recommended extracting these into two
+separate modules:
+- `syncflow-runtime` — orchestrators
+- `syncflow-persistence` — JPA entities + repositories
+
+## Status (2026-08-17): persistence extracted; runtime split still deferred
+
+A module split is **not done** in this batch. Reasons:
+
+1. **Cost / benefit**: the API module is currently the only Spring Boot
+   application. Splitting requires moving ~30 classes, rewriting build.gradle
+   dependencies, and updating tests. The benefit is only realized when a
+   second consumer of the runtime exists (e.g. a CLI runner or a worker
+   binary). Today: one consumer, one module.
+
+2. **Risk**: module splits break tests that mock JPA repositories. The
+   current test suite has 490 passing tests; a split will require updating
+   many of them and may regress.
+
+3. **Alternative marker**: the JPA-related types live in
+   `com.syncflow.api.{cdc,sync}.entity` packages (e.g.
+   `ActiveCaptureEntity`, `SyncJobEntity`). If the module split ever lands,
+   these packages are the natural "persistence" boundary — moving them
+   becomes mechanical.
+
+## What IS done in this batch
+
+- New `ActiveCaptureEntity` + `ActiveCaptureRepository` to durably persist
+  CDC capture state (F4)
+- `DistributedLockService` for multi-pod coordination (F13)
+- `markProcessedIfAbsent` for exactly-once CDC semantics (F14)
+- `TenantContext` flows explicitly through every orchestrator method (F1)
+- `RuntimeProperties` externalized (F8)
+- `ConnectionMapper` extracted (F10)
+- `PooledJdbcBatchWriter` with HikariCP (F5)
+- Batched writes via `DestinationRouter.writeBatch` (F6)
+- Backpressure: queue-full → DLQ (F7)
+- SQL identifier sanitization (C1)
+- Unified SPI parameter order (F3)
+- Multi-table CDC dispatch with observability (F2)
+
+## Added in this follow-up batch (2026-08-17)
+
+- **`syncflow-persistence` module extracted** (F11/F12, the persistence half):
+  all JPA entities + Spring Data repositories + Flyway migrations
+  (`V1`–`V15`) moved out of `syncflow-api`, per-domain subpackages preserved.
+  `api` now depends on `:syncflow-persistence` and lists the formerly
+  transitive deps explicitly (`spring-tx`, `spring-data-commons`,
+  `spring-data-jpa`, `spring-jdbc`).
+  `PersistenceConfig` (`@EntityScan`/`@EnableJpaRepositories`) wires the new
+  packages; root app still scans `com.syncflow`.
+- **Snapshot start guard** (S2): `SnapshotExecutor.start()` now holds the
+  Postgres advisory lock (`snapshot:<pipeline>`) and returns an existing
+  RUNNING job instead of spawning a second worker.
+- **Parallel PK-range snapshot** (F15): `SnapshotCapableConnector.rangeChunks`
+  splits a numeric-PK table into disjoint `[start,end)` ranges; the executor
+  processes all (table, chunk) work items on a fixed pool sized by
+  `syncflow.runtime.snapshot.parallelism` (default 4), writes serialized on a
+  single `DestinationWriter`. Per-chunk resume checkpoints
+  (`V15__snapshot_chunk_checkpoints.sql` adds `chunk_index`).
+  Non-numeric single-column PKs (uuid/text/date) fall back to sequential.
+- **Keyset cursor typed binding** (P6): numeric cursors now bind as `Long`,
+  fixing `bigint >= character varying` for both chunked and sequential paths.
+- Residual in-memory state (`cancellations`, `eventQueues`, `runningFlags`) is
+  intentionally ephemeral (cancellation flags + live queues); the durable shape
+  (jobs, status, statistics, checkpoints) is all in Postgres.
+
+## When to revisit the module split
+
+Triggers that justify the cost:
+
+- A second runtime consumer (CLI / worker binary / embedded library)
+- Independent release cadence for runtime vs API
+- Build time becomes dominated by orchestrator recompilation
+- A clear ownership boundary between runtime team and persistence team
+
+Until one of those, the current single-module layout with package-level
+separation is the right trade-off.

settings.gradle

@@ -20,6 +20,7 @@ include(
     "syncflow-common",
     "syncflow-core",
     "syncflow-api",
+    "syncflow-persistence",
     "syncflow-connectors",
     "syncflow-security",
     "syncflow-monitoring",

syncflow-api/build.gradle

@@ -15,24 +15,32 @@ dependencyManagement {
 
 dependencies {
     implementation project(":syncflow-core")
+    implementation project(":syncflow-persistence")
     implementation project(":syncflow-connectors")
     implementation project(":syncflow-security")
     implementation project(":syncflow-monitoring")
     implementation project(":syncflow-plugin-api")
 
     implementation libs.spring.boot.starter.web
     implementation libs.spring.boot.starter.validation
-    implementation libs.spring.boot.starter.data.jpa
+
+    // @Transactional and Spring Data Pageable come from spring-tx / spring-data-commons.
+    // Previously transitively via starter-data-jpa (now in :syncflow-persistence); the
+    // Spring Boot BOM still manages their versions.
+    implementation "org.springframework:spring-tx"
+    implementation "org.springframework.data:spring-data-commons"
+    // JdbcTemplate for DistributedLockService, and spring-data-jpa for the
+    // repository interface types used in signatures (adapter, services). Both
+    // were transitive via starter-data-jpa (now in :syncflow-persistence).
+    implementation "org.springframework:spring-jdbc"
+    implementation "org.springframework.data:spring-data-jpa"
 
     implementation libs.spring.boot.starter.actuator
     implementation libs.spring.boot.starter.security
     implementation libs.spring.security.oauth2.jose
     implementation libs.spring.security.oauth2.resource.server
     implementation libs.spring.doc.openapi.starter.webmvc.ui
     implementation libs.kafka.clients
-    implementation libs.flyway.core
-    implementation libs.flyway.database.postgresql
-    implementation libs.postgresql
     implementation libs.micrometer.registry.prometheus
     implementation libs.logback.classic
     implementation libs.mapstruct

syncflow-api/src/main/java/com/syncflow/api/agent/FleetManager.java

@@ -5,8 +5,8 @@
 import com.syncflow.agent.domain.AgentId;
 import com.syncflow.agent.domain.AgentStatus;
 import com.syncflow.agent.domain.HardwareMetrics;
-import com.syncflow.api.agent.entity.AgentEntity;
-import com.syncflow.api.agent.repository.AgentRepository;
+import com.syncflow.persistence.agent.entity.AgentEntity;
+import com.syncflow.persistence.agent.repository.AgentRepository;
 import com.syncflow.api.ops.metrics.MetricsRegistry;
 import com.syncflow.api.runtimestate.RuntimeStateJson;
 import com.syncflow.tenant.TenantSupport;

syncflow-api/src/main/java/com/syncflow/api/cdc/CaptureLifecycle.java

@@ -1,8 +1,8 @@
 package com.syncflow.api.cdc;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.syncflow.api.cdc.entity.ActiveCaptureEntity;
-import com.syncflow.api.cdc.repository.ActiveCaptureRepository;
+import com.syncflow.persistence.cdc.entity.ActiveCaptureEntity;
+import com.syncflow.persistence.cdc.repository.ActiveCaptureRepository;
 import com.syncflow.api.connection.ConnectionMapper;
 import com.syncflow.api.connection.service.ConnectionService;
 import com.syncflow.api.kafka.KafkaCdcConsumer;

syncflow-api/src/main/java/com/syncflow/api/cdc/OffsetStore.java

@@ -3,7 +3,7 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.syncflow.api.cdc.repository.CdcOffsetRepository;
+import com.syncflow.persistence.cdc.repository.CdcOffsetRepository;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Component;

syncflow-api/src/main/java/com/syncflow/api/config/RuntimeProperties.java

@@ -2,6 +2,7 @@
 
 import java.time.Duration;
 
+import jakarta.validation.constraints.Max;
 import jakarta.validation.constraints.Min;
 import jakarta.validation.constraints.NotNull;
 import jakarta.validation.constraints.Positive;
@@ -61,6 +62,32 @@ public static class Snapshot {
         @Min(value = 1, message = "checkpointIntervalBatches must be at least 1")
         private int checkpointIntervalBatches = 5;
 
+        /**
+         * Publish live progress (persist + SSE) every N batches aggregated
+         * across the parallel chunk workers. Serialized so concurrent workers
+         * cannot clobber the JSON-payload progress write.
+         */
+        @Min(value = 1, message = "progressPublishIntervalBatches must be at least 1")
+        private int progressPublishIntervalBatches = 10;
+
+        /**
+         * Number of parallel PK-range chunk workers per snapshot (F15). One per
+         * chunk; tables with non-numeric PKs ignore this and run sequentially.
+         */
+        @Min(value = 1, message = "parallelism must be at least 1")
+        private int parallelism = 4;
+
+        /**
+         * PK ranges split a table into at most this many chunks (throttle on the
+         * batching emphasis: chunk count = batchSize, capped here). The upper
+         * bound guards the per-chunk range allocation: each chunk is one
+         * ChunkRange + one prepared statement, so a huge misconfiguration (e.g.
+         * 1e9) would allocate gigabytes and stall snack-start.
+         */
+        @Min(value = 1, message = "maxChunks must be at least 1")
+        @Max(value = 1024, message = "maxChunks must be at most 1024")
+        private int maxChunks = 64;
+
     }
 
     @Setter

syncflow-api/src/main/java/com/syncflow/api/connection/mapper/ConnectionMapper.java

@@ -1,7 +1,7 @@
 package com.syncflow.api.connection.mapper;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.syncflow.api.connection.entity.ConnectionEntity;
+import com.syncflow.persistence.connection.entity.ConnectionEntity;
 import com.syncflow.core.connection.Connection;
 import com.syncflow.core.connection.ConnectionId;
 import com.syncflow.core.connection.ConnectionMetadata;
@@ -38,7 +38,7 @@
     @Mapping(target = "lastChecked", expression = "java(domain.getMetadata().lastChecked())")
     @Mapping(target = "createdAt", expression = "java(domain.getCreatedAt())")
     @Mapping(target = "updatedAt", expression = "java(domain.getUpdatedAt())")
     public abstract ConnectionEntity toEntity(Connection domain,
             String encryptedUsername,
             String encryptedPassword);
 

syncflow-api/src/main/java/com/syncflow/api/connection/service/ConnectionService.java

@@ -1,9 +1,9 @@
 package com.syncflow.api.connection.service;
 
 import com.syncflow.api.connection.encryption.EncryptionService;
-import com.syncflow.api.connection.entity.ConnectionEntity;
+import com.syncflow.persistence.connection.entity.ConnectionEntity;
 import com.syncflow.api.connection.mapper.ConnectionMapper;
-import com.syncflow.api.connection.repository.ConnectionRepository;
+import com.syncflow.persistence.connection.repository.ConnectionRepository;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.syncflow.common.exception.SyncFlowException;
 import com.syncflow.core.connection.Connection;

syncflow-api/src/main/java/com/syncflow/api/controller/AuthController.java

@@ -2,8 +2,8 @@
 
 import com.syncflow.api.security.AuthService;
 import com.syncflow.api.user.UserService;
-import com.syncflow.api.user.entity.UserEntity;
-import com.syncflow.api.user.repository.UserRepository;
+import com.syncflow.persistence.user.entity.UserEntity;
+import com.syncflow.persistence.user.repository.UserRepository;
 import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotBlank;
 import org.springframework.http.ResponseEntity;

syncflow-api/src/main/java/com/syncflow/api/ops/alert/AlertEngine.java

@@ -1,7 +1,7 @@
 package com.syncflow.api.ops.alert;
 
-import com.syncflow.api.ops.alert.entity.AlertEventEntity;
-import com.syncflow.api.ops.alert.repository.AlertEventRepository;
+import com.syncflow.persistence.ops.alert.entity.AlertEventEntity;
+import com.syncflow.persistence.ops.alert.repository.AlertEventRepository;
 import com.syncflow.tenant.TenantSupport;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;

syncflow-api/src/main/java/com/syncflow/api/pipeline/PipelineDesignerService.java

@@ -1,12 +1,12 @@
 package com.syncflow.api.pipeline;
 
 import com.syncflow.api.metadata.MetadataDiscoveryService;
-import com.syncflow.api.pipeline.entity.PipelineDesignEntity;
-import com.syncflow.api.pipeline.entity.PipelineDesignVersionEntity;
+import com.syncflow.persistence.pipeline.entity.PipelineDesignEntity;
+import com.syncflow.persistence.pipeline.entity.PipelineDesignVersionEntity;
 import com.syncflow.api.pipeline.mapper.JsonMapper;
 import com.syncflow.api.pipeline.mapper.PipelineDesignEntityMapper;
-import com.syncflow.api.pipeline.repository.PipelineDesignJpaRepository;
-import com.syncflow.api.pipeline.repository.PipelineDesignVersionJpaRepository;
+import com.syncflow.persistence.pipeline.repository.PipelineDesignJpaRepository;
+import com.syncflow.persistence.pipeline.repository.PipelineDesignVersionJpaRepository;
 import com.syncflow.core.pipeline.AuditInformation;
 import com.syncflow.core.pipeline.DestinationReference;
 import com.syncflow.core.pipeline.PipelineDesign;

syncflow-api/src/main/java/com/syncflow/api/pipeline/mapper/PipelineDesignEntityMapper.java

@@ -1,6 +1,6 @@
 package com.syncflow.api.pipeline.mapper;
 
-import com.syncflow.api.pipeline.entity.PipelineDesignEntity;
+import com.syncflow.persistence.pipeline.entity.PipelineDesignEntity;
 import com.syncflow.core.pipeline.AuditInformation;
 import com.syncflow.core.pipeline.PipelineDesign;
 import com.syncflow.core.pipeline.PipelineId;
@@ -29,7 +29,7 @@
     @Mapping(target = "createdAt", expression = "java(design.audit().createdAt())")
     @Mapping(target = "updatedAt", expression = "java(design.audit().updatedAt())")
     @Mapping(target = "versions", ignore = true)
     PipelineDesignEntity toEntity(PipelineDesign design,
             @org.mapstruct.Context JsonMapper jsonMapper);
 
     /**
@@ -48,7 +48,7 @@
     @Mapping(target = "version", expression = "java(design.audit().version())")
     @Mapping(target = "createdBy", expression = "java(design.audit().createdBy())")
     @Mapping(target = "updatedAt", expression = "java(design.audit().updatedAt())")
     void updateEntity(@MappingTarget PipelineDesignEntity entity, PipelineDesign design,
             @org.mapstruct.Context JsonMapper jsonMapper);
 
     default PipelineDesign toDomain(PipelineDesignEntity e,

syncflow-api/src/main/java/com/syncflow/api/pipeline/mapper/PipelineEntityMapper.java

@@ -1,6 +1,6 @@
 package com.syncflow.api.pipeline.mapper;
 
-import com.syncflow.api.pipeline.entity.PipelineEntity;
+import com.syncflow.persistence.pipeline.entity.PipelineEntity;
 import com.syncflow.core.model.Pipeline;
 import org.mapstruct.Context;
 import org.mapstruct.Mapper;

syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineRepositoryAdapter.java

@@ -5,6 +5,7 @@
 import com.syncflow.core.model.Pipeline;
 import com.syncflow.core.model.PipelineStatus;
 import com.syncflow.core.repository.PipelineRepository;
+import com.syncflow.persistence.pipeline.repository.PipelineJpaRepository;
 import org.springframework.context.annotation.Primary;
 import org.springframework.stereotype.Repository;
 

syncflow-api/src/main/java/com/syncflow/api/security/AuthService.java

@@ -1,7 +1,7 @@
 package com.syncflow.api.security;
 
 import com.syncflow.api.config.JwtProperties;
-import com.syncflow.api.user.repository.UserRepository;
+import com.syncflow.persistence.user.repository.UserRepository;
 import com.syncflow.tenant.TenantId;
 import org.springframework.security.authentication.AuthenticationManager;
 import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;

syncflow-api/src/main/java/com/syncflow/api/security/DbUserDetailsService.java

@@ -1,6 +1,6 @@
 package com.syncflow.api.security;
 
-import com.syncflow.api.user.repository.UserRepository;
+import com.syncflow.persistence.user.repository.UserRepository;
 import org.springframework.security.core.authority.SimpleGrantedAuthority;
 import org.springframework.security.core.userdetails.User;
 import org.springframework.security.core.userdetails.UserDetails;

syncflow-api/src/main/java/com/syncflow/api/security/apikey/ApiKeyStore.java

@@ -1,7 +1,7 @@
 package com.syncflow.api.security.apikey;
 
-import com.syncflow.api.security.apikey.entity.ApiKeyEntity;
-import com.syncflow.api.security.apikey.repository.ApiKeyRepository;
+import com.syncflow.persistence.security.apikey.entity.ApiKeyEntity;
+import com.syncflow.persistence.security.apikey.repository.ApiKeyRepository;
 import com.syncflow.tenant.TenantId;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Repository;

syncflow-api/src/main/java/com/syncflow/api/security/audit/EnterpriseAuditStore.java

@@ -1,7 +1,7 @@
 package com.syncflow.api.security.audit;
 
-import com.syncflow.api.security.audit.entity.AuditRecordEntity;
-import com.syncflow.api.security.audit.repository.AuditRecordRepository;
+import com.syncflow.persistence.security.audit.entity.AuditRecordEntity;
+import com.syncflow.persistence.security.audit.repository.AuditRecordRepository;
 import com.syncflow.tenant.TenantId;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.stereotype.Component;

syncflow-api/src/main/java/com/syncflow/api/security/quota/QuotaEngine.java

@@ -2,8 +2,8 @@
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.syncflow.api.runtimestate.RuntimeStateJson;
-import com.syncflow.api.security.quota.entity.QuotaEntity;
-import com.syncflow.api.security.quota.repository.QuotaRepository;
+import com.syncflow.persistence.security.quota.entity.QuotaEntity;
+import com.syncflow.persistence.security.quota.repository.QuotaRepository;
 import com.syncflow.tenant.TenantId;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;