diff --git a/ARCHITECTURE_ANALYSIS.md b/ARCHITECTURE_ANALYSIS.md new file mode 100644 index 0000000..f3166a0 --- /dev/null +++ b/ARCHITECTURE_ANALYSIS.md @@ -0,0 +1,512 @@ +# SyncFlow Architecture Analysis + +**Generated:** 2026-08-12 +**Last updated:** 2026-08-17 (F11/F12 persistence extraction, S2 snapshot lock, F15 parallel chunking, P6 cursor typing — see item statuses below) +**Scope:** End-to-end codebase review (core, api, connectors, common, agent) + +--- + +## 1. Architecture Overview + +### 1.1 Module Structure + +| Module | Responsibility | Key Components | +|--------|----------------|----------------| +| `syncflow-common` | Shared primitives, tenant context, exceptions | `TenantContextHolder`, `SyncFlowException`, `Agent` domain | +| `syncflow-core` | Domain models, SPI interfaces, pipeline/snapshot/CDC logic | `Pipeline`, `SnapshotJob`, `CDCEvent`, connectors SPI | +| `syncflow-connectors` | Concrete connector implementations (JDBC, Debezium, MongoDB, writers) | `PostgresCdcConnector`, `MySqlCdcConnector`, `JdbcBatchWriter` | +| `syncflow-api` | REST API, orchestration, runtime state, multi-tenancy | `SnapshotExecutor`, `SyncOrchestrator`, `CaptureLifecycle` | +| `syncflow-agent` | Standalone agent for distributed execution | `AgentRegistrar`, `HeartbeatSender` | + +### 1.2 Data Flow Summary + +``` +┌──────────────┐ ┌─────────────────┐ ┌──────────────────┐ +│ Pipeline │────▶│ SnapshotExecutor│────▶│ Source Connector │ +│ Design │ │ (batch read + │ │ (keyset/offset │ +│ (DDL + map) │ │ transform + │ │ pagination) │ +└──────────────┘ │ write) │ └────────┬─────────┘ + └────────┬────────┘ │ + │ CDC events ▼ + ┌────────▼────────┐ ┌──────────────────┐ + │ CaptureLifecycle│────▶│ Debezium Engine │ + │ (start/stop/ │ │ (WAL/binlog │ + │ pause/resume) │ │ streaming) │ + └────────┬────────┘ └────────┬─────────┘ + │ CDC events │ + ┌────────▼────────┐ │ + │ SyncOrchestrator│◀────┐ │ + │ (queue + │ │ │ + │ process + │ │ │ + │ route + DLQ) │ │ │ + └────────┬────────┘ │ │ + │ │ │ + ┌────────▼────────┐ │ │ + │ DestinationRouter│───┘ │ + │ (writer per │ │ + │ connection) │ │ + └─────────────────┘ │ +``` + +--- + +## 2. Critical Problem Areas + +### 2.1 Architecture & Design Flaws + +| # | Issue | Location | Severity | Impact | +|---|-------|----------|----------|--------| +| **A1** | **In-memory runtime state maps** (`ConcurrentHashMap`) used for active jobs/captures | `SnapshotExecutor.cancellations`, `CaptureLifecycle.activeCaptures`, `SyncOrchestrator.eventQueues` | **HIGH** | State lost on pod restart; no HA; memory leaks if not cleaned; cannot scale horizontally | +| **A2** | **Virtual threads + ThreadLocal tenant context** — broken by design | `TenantContextHolder` (ThreadLocal) + `Thread.startVirtualThread()` | **HIGH** | Tenant leakage across requests; security boundary violation; `TenantSupport.workerContext()` hack required | +| **A3** | **Single-table assumption in SyncOrchestrator** | `SyncOrchestrator.start()` line 120: `pipeline.tableMappings().stream().findFirst()` | **HIGH** | Only first table mapping processed; multi-table pipelines silently broken | +| **A4** | **DELETE operations not implemented in writer** | `DestinationRouter.java:48` — `// ponytail: DELETE via writer not yet supported` | **MEDIUM** | Data drift; deletes not propagated to destination | +| **A5** | **No exactly-once semantics for CDC** | `CaptureLifecycle` + `SyncOrchestrator` — idempotency only at event level, not transaction | **MEDIUM** | Duplicate events on restart; no transaction boundary preservation | +| **A6** | **PipelineRepository is in-memory** | `InMemoryPipelineRepository` used in tests; no persistent impl visible in core | **MEDIUM** | Core module lacks persistence abstraction; API module has JPA entities but core doesn't define SPI | +| **A7** | **Tight coupling: API module imports core SPI + concrete domain** | `syncflow-api` depends on `syncflow-core` SPI and domain models | **MEDIUM** | Violates clean architecture; core should not know about API; API should depend on core interfaces only | +| **A8** | **No circuit breaker / backpressure on event queue** | `SyncOrchestrator` uses unbounded `LinkedBlockingQueue(10000)` | **MEDIUM** | OOME risk under burst; no flow control | + +--- + +### 2.2 Duplicate Logic + +| # | Duplicated Logic | Locations | Recommendation | +|---|------------------|-----------|----------------| +| **D1** | **ConnectionConfiguration construction** from `Connection` entity | `SnapshotExecutor.toConfig()`, `SyncOrchestrator.toConfig()`, `CaptureLifecycle.toConfig()`, `DestinationRouter.toConfig()`, `PipelineDesignerService.toConfig()` (5 copies) | Extract to `ConnectionMapper` utility in `syncflow-common` or `syncflow-api` | +| **D2** | **Offset store key = pipelineId** (hardcoded) | `CaptureLifecycle.start()`, `CaptureLifecycle.stop()`, `OffsetStore` interface | Make configurable; support multi-table offsets via composite key | +| **D3** | **Event publishing to publisher** (counter + publish) | `CaptureLifecycle.start()` line 100-105, `SnapshotExecutor` doesn't publish | Unify event emission via `EventPublisher` abstraction | +| **D4** | **BatchInformation cursor/offset calculation** | `SnapshotPlannerUnitTest` mirrors logic from `AbstractJdbcSnapshotConnector.readBatch()` | Move to shared `PaginationUtil` | +| **D5** | **ValidationResult pattern** (ok/failed) | `ValidationResult` in core SPI + `ValidationResult` in pipeline validation (different packages) | Unify into single `ValidationResult` in `syncflow-common` | +| **D6** | **Metrics counter/timer boilerplate** | Every executor/orchestrator repeats `meterRegistry.counter(...)` patterns | Create `MetricsHelper` with `incrementCounter()`, `recordTimer()` | + +--- + +### 2.3 Performance Bottlenecks + +| # | Bottleneck | Location | Why It Matters | +|---|------------|----------|----------------| +| **P1** | **Per-event writer connect/commit/close** | `DestinationRouter.write()` lines 32-58 | New DB connection + transaction per CDC event = catastrophic latency | +| **P2** | **No connection pooling in writers** | `JdbcBatchWriter.connect()` creates raw `DriverManager.getConnection()` | No pooling; connection storm under load | +| **P3** | **Virtual thread per pipeline** (unbounded) | `SnapshotExecutor.start()`, `SyncOrchestrator.start()` | Thread explosion with many pipelines; no pool sizing | +| **P4** | **Jackson ObjectMapper per connector instance** | `PostgresCdcConnector.MAPPER`, `MySqlCdcConnector.MAPPER` (static but per-class) | Acceptable but could be shared; minor | +| **P5** | **Synchronous flush/commit per batch in snapshot** | `SnapshotExecutor.executeInner()` lines 246-248 | Blocks virtual thread; should batch commits | +| **P6** | **Keyset pagination uses string cursor comparison** | `AbstractJdbcSnapshotConnector.readKeysetPage()` line 91: `stmt.setObject(1, cursor)` | Lexicographic comparison breaks for numeric/uuid PKs if cursor not same type. **Status (2026-08-17): ✅ fixed** — numeric cursors bound as `Long` (fixes `bigint >= character varying`) | +| **P7** | **No batching in SyncOrchestrator event processing** | `runInner()` drains max 100 events, processes one-by-one | Writer called per event (see P1); should batch writes | + +--- + +### 2.4 Scalability Risks + +| # | Risk | Details | +|---|------|---------| +| **S1** | **In-memory maps prevent horizontal scaling** | `activeCaptures`, `eventQueues`, `runningFlags`, `cancellations`, `tenantOf` — all `ConcurrentHashMap` in single JVM. ~~State lost on restart~~. **Status (2026-08-17):** durable shape done — jobs/status/statistics/checkpoints/captures/offsets/processed-events all persist (V12–V14 + earlier). Remaining `eventQueues`/`runningFlags`/`cancellations`/`tenantOf` are ephemeral by design (live queues + cancellation flags can't be persisted) | +| **S2** | **No distributed locking for pipeline operations** | Concurrent `start()` on same pipeline from different pods = duplicate CDC engines. **Status (2026-08-17): ✅ done** — `DistributedLockService` (Postgres advisory) guards CDC capture and snapshot start | +| **S3** | **Debezium offset store uses single table** | `debezium_offsets` table with `pipelineId` key — no partitioning; contention at scale. **Status:** unchanged; now tenant-keyed (`tenantId:pipelineId`) | +| **S4** | **Tenant context via ThreadLocal** | Fundamentally incompatible with virtual threads / reactive; breaks in any async boundary. **Status (2026-08-17): ✅ mitigated** — `TenantContext` threaded explicitly through orchestrators/workers; worker threads assert no ThreadLocal | +| **S5** | **Single writer connection per event** | `DestinationRouter` opens/closes connection per `write()` call — cannot scale. **Status (2026-08-17): ✅ done** — `PooledJdbcBatchWriter` + HikariCP, batched writes | +| **S6** | **No partitioning/sharding strategy for large tables** | Snapshot reads entire table sequentially; no parallel chunking. **Status (2026-08-17): ✅ done** — parallel PK-range chunking (F15) | + +--- + +### 2.5 Maintainability Issues + +| # | Issue | Impact | +|---|-------|--------| +| **M1** | **Two `ValidationResult` classes** | `com.syncflow.core.spi.ValidationResult` vs `com.syncflow.core.pipeline.validation.ValidationResult` — confusion, not unified | +| **M2** | **Two `ProcessingContext` classes** | `com.syncflow.core.snapshot.pipeline.ProcessingContext` vs `com.syncflow.core.sync.ProcessingContext` — same name, different packages | +| **M3** | **Core module has no persistence SPI** | `PipelineRepository` is interface but only `InMemoryPipelineRepository` in core; JPA entities only in API | +| **M4** | **`@Transactional` on read-only methods with validation that throws** | `PipelineDesignerService.validate()` explicitly avoids `@Transactional` due to `UnexpectedRollbackException` — symptom of wrong exception handling | +| **M5** | **Magic numbers / hardcoded values** | `QUEUE_CAPACITY=10000`, `MAX_RETRIES=3`, `BASE_DELAY_MS=1000`, checkpoint every 5 batches — not configurable | +| **M6** | **`ponytail:` comments indicate known debt** | `DestinationRouter:49` — "DELETE via writer not yet supported" | +| **M7** | **Inconsistent error handling** | Some methods throw `SyncFlowException`, others `IllegalArgumentException`, others `RuntimeException` — no unified strategy | +| **M8** | **`SnapshotExecutor` does too much** | 350+ lines: orchestration + persistence + metrics + tenant context + checkpointing + event emission — violates SRP | + +--- + +## 3. Clean Architecture Breakdown + +### 3.1 Current Layering (Problematic) + +``` +┌─────────────────────────────────────────────────────────────┐ +│ syncflow-api │ +│ Controllers, Orchestrators, JPA Entities, Repositories │ +│ ▼ depends on ▼ │ +│ syncflow-core │ +│ Domain Models, SPI Interfaces, In-Memory Repos │ +│ ▼ depends on ▼ │ +│ syncflow-common │ +│ TenantContext, Exceptions, Base Types │ +└─────────────────────────────────────────────────────────────┘ + ▲ ▲ + │ implements │ uses +┌─────────┴─────────┐ ┌─────────┴─────────┐ +│ syncflow-connectors │ (external) │ +│ Debezium, JDBC, │ │ +│ Writers, MongoDB │ │ +└───────────────────────┘ │ +``` + +**Violations:** +- API module contains business logic (`SyncOrchestrator`, `SnapshotExecutor`) — should be in core or a separate `syncflow-runtime` +- Core defines SPI but API implements it — inverted dependency +- Connectors depend on core SPI (correct) but core has no persistence SPI + +--- + +### 3.2 Recommended Clean Architecture + +``` +┌──────────────────────────────────────────────────────────────────┐ +│ syncflow-api (Thin) │ +│ REST Controllers, DTOs, OpenAPI, Security Config │ +│ ▼ uses ▼ │ +│ syncflow-runtime (NEW) │ +│ Orchestration: SnapshotExecutor, SyncOrchestrator, │ +│ CaptureLifecycle, CheckpointStore, EventPublisher │ +│ ▼ uses ▼ │ +│ syncflow-core │ +│ Domain Models (Pipeline, CDCEvent, SnapshotJob), │ +│ SPI Interfaces (Connector, Writer, Repository, OffsetStore), │ +│ Pure Domain Services (validation, transformation, mapping) │ +│ ▲ implements ▲ ▲ implements ▲ │ +│ ┌─────────────────┐ ┌─────────────────┐ │ +│ │ syncflow- │ │ syncflow- │ │ +│ │ connectors │ │ persistence │ (NEW) │ +│ │ (Debezium, JDBC,│ │ (JPA, │ │ +│ │ Writers, etc) │ │ repositories) │ │ +│ └─────────────────┘ └─────────────────┘ │ +└──────────────────────────────────────────────────────────────────┘ + ▲ + │ uses +┌─────────┴─────────┐ +│ syncflow-common │ +│ TenantContext, │ +│ Exceptions, │ +│ Base Types, │ +│ Utilities │ +└───────────────────┘ +``` + +**Key Changes:** +1. Extract runtime orchestration to `syncflow-runtime` module +2. Add `syncflow-persistence` module for JPA repositories +3. Core becomes pure domain + SPI (no Spring, no JPA) +4. API becomes thin controller layer only + +--- + +## 4. Refactoring Strategies + +### 4.1 Immediate Fixes (P0 — Security/Correctness) + +| # | Action | Files | +|---|--------|-------| +| **F1** | Replace `ThreadLocal` tenant context with `ContextPropagator` (Micrometer) or pass `TenantId` explicitly | `TenantContextHolder`, all callers in `SnapshotExecutor`, `SyncOrchestrator`, `CaptureLifecycle` | +| **F2** | Fix multi-table CDC processing in `SyncOrchestrator` | `SyncOrchestrator.start()` line 120 — iterate all `tableMappings` | +| **F3** | Implement DELETE in `DestinationRouter` / writers | `DestinationRouter.write()` case DELETE, `JdbcBatchWriter` add `deleteBatch()` | +| **F4** | Persist runtime state to DB (not in-memory maps) | Replace `ConcurrentHashMap` in `SnapshotExecutor`, `CaptureLifecycle`, `SyncOrchestrator` with JPA entities | +| **F5** | Add connection pooling to writers | `JdbcBatchWriter` → use `HikariDataSource` per connection config | + +--- + +### 4.2 Short-term (P1 — Scalability/Performance) + +| # | Action | Files | +|---|--------|-------| +| **F6** | Batch writes in `DestinationRouter` — accumulate events, flush periodically | `DestinationRouter`, `SyncOrchestrator.runInner()` | +| **F7** | Add circuit breaker + backpressure to event queues | `SyncOrchestrator.eventQueues` → use `Resilience4j` or custom | +| **F8** | Make all hardcoded constants configurable | `application.yml` + `@ConfigurationProperties` for queue size, retry policy, checkpoint interval | +| **F9** | Unify `ValidationResult` and `ProcessingContext` | Move to `syncflow-common` | +| **F10** | Extract `ConnectionConfiguration` mapper | New `ConnectionMapper` in `syncflow-common` or `syncflow-api` | + +--- + +### 4.3 Medium-term (P2 — Architecture) + +| # | Action | Modules | Status (2026-08-17) | +|---|--------|---------|----------------------| +| **F11** | Create `syncflow-runtime` module; move orchestrators out of API | New module | **Still deferred** — orchestrators stay in api; split gated on a second runtime consumer (see `docs/adr/MODULE_SPLIT_DEFERRED.md`) | +| **F12** | Create `syncflow-persistence` module; define `PipelineRepository`, `SnapshotJobRepository`, `SyncJobRepository` SPI in core | New module + core | ✅ **Done** — `syncflow-persistence` extracted; entities/repos/Flyway migrated; `PersistenceConfig`; `PipelineRepository` SPI exists in core | +| **F13** | Implement distributed locking for pipeline operations | `syncflow-runtime` + Redis/Postgres advisory locks | ✅ **Done (Postgres advisory)** — `DistributedLockService`; now also guards `SnapshotExecutor.start()` (S2) | +| **F14** | Add exactly-once CDC: transaction-aware idempotency + offset commit | `CaptureLifecycle`, `SyncOrchestrator`, `OffsetStore` | ✅ **Done** — `markProcessedIfAbsent` after write (F14 in ADR) | +| **F15** | Parallel snapshot: chunk tables, process chunks concurrently | `SnapshotExecutor`, `AbstractJdbcSnapshotConnector` | ✅ **Done** — PK-range chunking via `rangeChunks` + fixed pool `parallelism`; per-chunk checkpoints (V15) | + +--- + +### 4.4 Long-term (P3 — Platform) + +| # | Action | +|---|--------| +| **F16** | Migrate to reactive (Project Reactor) or structured concurrency for better resource control | +| **F17** | Add multi-region / geo-replication support | +| **F18** | Implement connector plugin system (dynamic loading) | +| **F19** | Add SQL-based transformation engine (push down to DB) | + +--- + +## 5. Improved Production-Grade Code Samples + +### 5.1 Fixed Tenant Context (No ThreadLocal) + +```java +// syncflow-common/src/main/java/com/syncflow/tenant/TenantContext.java +public record TenantContext(TenantId tenantId, String userId, Map attributes) { + public static TenantContext system(TenantId tenantId) { + return new TenantContext(tenantId, "system", Map.of()); + } +} + +// Usage: pass explicitly, no ThreadLocal +public interface TenantAware { + TenantId tenantId(); +} + +// In orchestrators: +public SnapshotJob start(String pipelineId, TenantContext ctx) { ... } +private void execute(TenantContext ctx, SnapshotJob job, PipelineDesign pipeline) { ... } +``` + +--- + +### 5.2 Batched Destination Router with Connection Pool + +```java +// syncflow-api/src/main/java/com/syncflow/api/sync/BatchedDestinationRouter.java +@Component +public class BatchedDestinationRouter { + + private final Map writerPools = new ConcurrentHashMap<>(); + private final ConnectionService connectionService; + private final WriterRegistry writerRegistry; + private final MeterRegistry meterRegistry; + + public WriteResult writeBatch(String connectionId, List events, List destColumns) { + var pool = writerPools.computeIfAbsent(connectionId, this::createPool); + var writer = pool.borrow(); + try { + var grouped = events.stream() + .collect(Collectors.groupingBy(e -> e.source().table())); + + for (var entry : grouped.entrySet()) { + var rows = entry.getValue().stream() + .map(this::extractRow) + .filter(Objects::nonNull) + .toList(); + if (!rows.isEmpty()) { + writer.writeBatch(entry.getKey(), rows, destColumns); + } + } + writer.flush(); + writer.commit(); + return new WriteResult(true, null); + } catch (Exception e) { + writer.rollback(); + return new WriteResult(false, e.getMessage()); + } finally { + pool.release(writer); + } + } + + private WriterPool createPool(String connectionId) { + var conn = connectionService.getWithDecryptedCredentials(connectionId); + var config = ConnectionMapper.toConfig(conn); + var writer = writerRegistry.get(ConnectorTypeMapper.toCore(conn.getProperties().type())) + .orElseThrow(); + return new WriterPool(() -> { + var w = writerRegistry.get(...).orElseThrow(); + w.connect(config); + return w; + }, 4); // pool size configurable + } +} +``` + +--- + +### 5.3 Persistent Runtime State (Replaces In-Memory Maps) + +```java +// syncflow-persistence/src/main/java/com/syncflow/persistence/entity/SnapshotJobEntity.java +@Entity +@Table(name = "snapshot_jobs") +public class SnapshotJobEntity { + @Id String id; + String tenantId; + String pipelineId; + @Enumerated(STRING) SnapshotStatus status; + @Column(columnDefinition = "jsonb") String payload; + Instant createdAt, updatedAt; + // Indexes on tenantId, pipelineId, status +} + +// syncflow-persistence/src/main/java/com/syncflow/persistence/entity/ActiveCaptureEntity.java +@Entity +@Table(name = "active_captures", + uniqueConstraints = @UniqueConstraint(columnNames = {"tenant_id", "pipeline_id"})) +public class ActiveCaptureEntity { + @Id String id; // pipelineId + String tenantId; + String pipelineId; + @Enumerated(STRING) CaptureStatus status; + @Column(columnDefinition = "jsonb") String offset; // serialized offset map + Instant startedAt, updatedAt; + // Replaces CaptureLifecycle.activeCaptures ConcurrentHashMap +} +``` + +--- + +### 5.4 Multi-Table Sync Orchestrator + +```java +// syncflow-runtime/src/main/java/com/syncflow/runtime/sync/MultiTableSyncOrchestrator.java +@Component +public class MultiTableSyncOrchestrator { + + public SyncJob start(String pipelineId, TenantContext ctx) { + var pipeline = pipelineService.get(pipelineId); + + // One worker per table mapping (or per pipeline with partitioned queue) + for (var tm : pipeline.tableMappings()) { + var queue = new LinkedBlockingQueue(queueCapacity); + var key = TenantKey.of(ctx.tenantId(), pipelineId, tm.sourceTable()); + eventQueues.put(key, queue); + + Thread.startVirtualThread(() -> + runTableWorker(ctx, pipelineId, tm, queue)); + } + } + + private void runTableWorker(TenantContext ctx, String pipelineId, + TableMapping tm, BlockingQueue queue) { + TenantContextHolder.set(ctx); // If still needed, else pass explicitly + try { + var writer = batchedRouter.forTable(tm.destinationTable()); + while (running) { + var batch = drainBatch(queue, 100); + if (batch.isEmpty()) continue; + + var rows = batch.stream() + .map(this::transform) + .filter(Objects::nonNull) + .toList(); + + if (!rows.isEmpty()) { + writer.writeBatch(tm.destinationTable(), rows, tm.columnMappings()); + } + // Checkpoint per table + checkpointStore.save(pipelineId, tm.sourceTable(), cursor); + } + } finally { + TenantContextHolder.clear(); + } + } +} +``` + +--- + +### 5.5 Unified ValidationResult & ProcessingContext + +```java +// syncflow-common/src/main/java/com/syncflow/common/validation/ValidationResult.java +public sealed interface ValidationResult permits ValidationResult.Ok, ValidationResult.Failed { + boolean valid(); + List errors(); + + record Ok() implements ValidationResult { + public boolean valid() { return true; } + public List errors() { return List.of(); } + public static Ok ok() { return new Ok(); } + } + + record Failed(List errors) implements ValidationResult { + public boolean valid() { return false; } + public static Failed failed(String... errors) { + return new Failed(List.of(errors)); + } + public static Failed failed(List errors) { + return new Failed(errors); + } + } +} + +// syncflow-common/src/main/java/com/syncflow/common/pipeline/ProcessingContext.java +public record ProcessingContext( + PipelineDesign pipeline, + TableMapping tableMapping, + TenantContext tenantContext +) { } +``` + +--- + +### 5.6 Configurable Constants + +```yaml +# application.yml +syncflow: + runtime: + snapshot: + queue-capacity: 10000 + checkpoint-interval-batches: 5 + batch-size: 1000 + writer-pool-size: 4 + cdc: + queue-capacity: 10000 + max-retries: 3 + base-retry-delay-ms: 1000 + offset-flush-interval-ms: 5000 + sync: + queue-capacity: 10000 + poll-timeout-ms: 500 + max-retries: 3 + base-retry-delay-ms: 1000 + writer-batch-size: 100 + writer-flush-interval-ms: 100 +``` + +```java +@ConfigurationProperties("syncflow.runtime") +public class RuntimeProperties { + private Snapshot snapshot = new Snapshot(); + private Cdc cdc = new Cdc(); + private Sync sync = new Sync(); + + // nested classes with defaults... +} +``` + +--- + +## 6. Optimization Summary + +| Area | Current | Target | Effort | +|------|---------|--------|--------| +| **Tenant isolation** | ThreadLocal (broken) | Explicit context passing | Medium | +| **Multi-table support** | First table only | Full pipeline parallelism | Medium | +| **DELETE propagation** | Not implemented | Full CRUD sync | Low | +| **Connection management** | Per-event new connection | Pooled, batched | Medium | +| **Runtime state** | In-memory maps | Persistent (DB) + distributed lock | High | +| **Exactly-once CDC** | At-least-once | Transactional idempotency | High | +| **Horizontal scaling** | Single JVM | Multi-pod with shared state | High | +| **Configuration** | Hardcoded | Externalized + validated | Low | +| **Code duplication** | 5+ Connection mappers, 2 ValidationResults | Single shared utilities | Low | + +--- + +## 7. Recommended Implementation Order + +> **Status (2026-08-17):** items 1–4 and item 5 are complete. F15 (parallel +> snapshot) is also done. Remaining roadmap below. + +1. ~~**Week 1-2**: F1, F2, F3, F9, F10 (correctness + deduplication)~~ ✅ done +2. ~~**Week 3-4**: F5, F6, F8 (performance + config)~~ ✅ done +3. ~~**Week 5-6**: F4, F7 (persistence + resilience)~~ ✅ done (runtime state durable, backpressure via DLQ) +4. ~~**Week 7-8**: F11, F12 (architecture extraction)~~ ✅ persistence extracted; `syncflow-runtime` still deferred (see ADR) +5. ~~**Week 9-10**: F13, F14 (distributed + exactly-once)~~ ✅ done (Postgres advisory locks, mark-after-write idempotency) +6. **Ongoing**: F15 ✅ done (parallel PK-range chunking); F16+ (reactive, geo-replication, plugin system, SQL-transform pushdown) still open + +--- + +*This analysis is based on code review as of 2026-08-12. No changes were made to the codebase.* \ No newline at end of file diff --git a/docs/adr/MODULE_SPLIT_DEFERRED.md b/docs/adr/MODULE_SPLIT_DEFERRED.md new file mode 100644 index 0000000..34817a7 --- /dev/null +++ b/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:`) 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. \ No newline at end of file diff --git a/settings.gradle b/settings.gradle index 802351e..8829712 100644 --- a/settings.gradle +++ b/settings.gradle @@ -20,6 +20,7 @@ include( "syncflow-common", "syncflow-core", "syncflow-api", + "syncflow-persistence", "syncflow-connectors", "syncflow-security", "syncflow-monitoring", diff --git a/syncflow-api/build.gradle b/syncflow-api/build.gradle index 5d60e5d..c7d00a4 100644 --- a/syncflow-api/build.gradle +++ b/syncflow-api/build.gradle @@ -15,6 +15,7 @@ dependencyManagement { dependencies { implementation project(":syncflow-core") + implementation project(":syncflow-persistence") implementation project(":syncflow-connectors") implementation project(":syncflow-security") implementation project(":syncflow-monitoring") @@ -22,7 +23,17 @@ dependencies { 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 @@ -30,9 +41,6 @@ dependencies { 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 diff --git a/syncflow-api/src/main/java/com/syncflow/api/agent/FleetManager.java b/syncflow-api/src/main/java/com/syncflow/api/agent/FleetManager.java index 706fd6a..a353462 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/agent/FleetManager.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/cdc/CaptureLifecycle.java b/syncflow-api/src/main/java/com/syncflow/api/cdc/CaptureLifecycle.java index 0bcf46a..b2a01df 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/cdc/CaptureLifecycle.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/cdc/OffsetStore.java b/syncflow-api/src/main/java/com/syncflow/api/cdc/OffsetStore.java index 34fdf23..c56e068 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/cdc/OffsetStore.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/config/RuntimeProperties.java b/syncflow-api/src/main/java/com/syncflow/api/config/RuntimeProperties.java index 6e20bf4..57ec34f 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/config/RuntimeProperties.java +++ b/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 diff --git a/syncflow-api/src/main/java/com/syncflow/api/connection/mapper/ConnectionMapper.java b/syncflow-api/src/main/java/com/syncflow/api/connection/mapper/ConnectionMapper.java index 4f95680..be47af7 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/connection/mapper/ConnectionMapper.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/connection/service/ConnectionService.java b/syncflow-api/src/main/java/com/syncflow/api/connection/service/ConnectionService.java index 6c668bd..22e7a70 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/connection/service/ConnectionService.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/controller/AuthController.java b/syncflow-api/src/main/java/com/syncflow/api/controller/AuthController.java index d34462e..47b5d9c 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/controller/AuthController.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/ops/alert/AlertEngine.java b/syncflow-api/src/main/java/com/syncflow/api/ops/alert/AlertEngine.java index de56c66..651c6dd 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/ops/alert/AlertEngine.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/PipelineDesignerService.java b/syncflow-api/src/main/java/com/syncflow/api/pipeline/PipelineDesignerService.java index 2eecf0a..253440b 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/PipelineDesignerService.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/mapper/PipelineDesignEntityMapper.java b/syncflow-api/src/main/java/com/syncflow/api/pipeline/mapper/PipelineDesignEntityMapper.java index 085275f..7bf821c 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/mapper/PipelineDesignEntityMapper.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/mapper/PipelineEntityMapper.java b/syncflow-api/src/main/java/com/syncflow/api/pipeline/mapper/PipelineEntityMapper.java index 1179e9d..1e07364 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/mapper/PipelineEntityMapper.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineRepositoryAdapter.java b/syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineRepositoryAdapter.java index 06233e4..6ca0d0b 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineRepositoryAdapter.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/AuthService.java b/syncflow-api/src/main/java/com/syncflow/api/security/AuthService.java index e69241b..f1b6d76 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/AuthService.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/DbUserDetailsService.java b/syncflow-api/src/main/java/com/syncflow/api/security/DbUserDetailsService.java index 0d548c5..305589b 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/DbUserDetailsService.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/apikey/ApiKeyStore.java b/syncflow-api/src/main/java/com/syncflow/api/security/apikey/ApiKeyStore.java index 152dda9..d6aba16 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/apikey/ApiKeyStore.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/audit/EnterpriseAuditStore.java b/syncflow-api/src/main/java/com/syncflow/api/security/audit/EnterpriseAuditStore.java index d156daf..27292e5 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/audit/EnterpriseAuditStore.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/quota/QuotaEngine.java b/syncflow-api/src/main/java/com/syncflow/api/security/quota/QuotaEngine.java index c3b916f..b067623 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/quota/QuotaEngine.java +++ b/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; diff --git a/syncflow-api/src/main/java/com/syncflow/api/snapshot/CheckpointStore.java b/syncflow-api/src/main/java/com/syncflow/api/snapshot/CheckpointStore.java index a9141ed..fc6b277 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/snapshot/CheckpointStore.java +++ b/syncflow-api/src/main/java/com/syncflow/api/snapshot/CheckpointStore.java @@ -1,15 +1,23 @@ package com.syncflow.api.snapshot; -import com.syncflow.api.snapshot.entity.SnapshotCheckpointEntity; -import com.syncflow.api.snapshot.repository.SnapshotCheckpointRepository; +import com.syncflow.persistence.snapshot.entity.SnapshotCheckpointEntity; +import com.syncflow.persistence.snapshot.repository.SnapshotCheckpointRepository; import com.syncflow.core.snapshot.SnapshotCheckpoint; -import com.syncflow.tenant.TenantSupport; import org.springframework.stereotype.Component; import org.springframework.transaction.annotation.Transactional; +import java.util.Optional; + /** * Resume checkpoints persisted to PostgreSQL; one row per - * tenant+pipeline+table. + * tenant+pipeline+table+chunk. + * + *

+ * The tenant is passed explicitly rather than read from a ThreadLocal. + * Snapshot workers run on pool/virtual threads that never set + * {@code TenantContextHolder}; keying on the ThreadLocal there would resolve + * to {@code TenantId.DEFAULT} and attribute every tenant's checkpoints to the + * default tenant (cross-tenant cursor reuse on resume). */ @Component public class CheckpointStore { @@ -21,14 +29,13 @@ public CheckpointStore(SnapshotCheckpointRepository repository) { } @Transactional - public void save(SnapshotCheckpoint checkpoint) { - var entity = repository - .findByTenantIdAndPipelineIdAndSourceTable( - tenantId(), checkpoint.pipelineId(), checkpoint.sourceTable()) + public void save(String tenantId, SnapshotCheckpoint checkpoint) { + var entity = find(tenantId, checkpoint.pipelineId(), checkpoint.sourceTable(), checkpoint.chunkIndex()) .orElseGet(SnapshotCheckpointEntity::new); - entity.setTenantId(tenantId()); + entity.setTenantId(tenantId); entity.setPipelineId(checkpoint.pipelineId()); entity.setSourceTable(checkpoint.sourceTable()); + entity.setChunkIndex(checkpoint.chunkIndex()); entity.setLastBatchNumber(checkpoint.lastBatchNumber()); entity.setRowsProcessed(checkpoint.rowsProcessed()); entity.setCursorPos(checkpoint.cursor()); @@ -37,31 +44,25 @@ public void save(SnapshotCheckpoint checkpoint) { } @Transactional(readOnly = true) - public SnapshotCheckpoint get(String pipelineId, String sourceTable) { - return repository - .findByTenantIdAndPipelineIdAndSourceTable(tenantId(), pipelineId, sourceTable) + public SnapshotCheckpoint get(String tenantId, String pipelineId, String sourceTable, int chunkIndex) { + return find(tenantId, pipelineId, sourceTable, chunkIndex) .map(this::toDomain) .orElse(null); } @Transactional - public void delete(String pipelineId, String sourceTable) { - repository - .findByTenantIdAndPipelineIdAndSourceTable(tenantId(), pipelineId, sourceTable) - .ifPresent(repository::delete); + public void deleteAll(String tenantId, String pipelineId) { + repository.deleteAllForPipeline(tenantId, pipelineId); } - @Transactional - public void deleteAll(String pipelineId) { - repository.deleteAllForPipeline(tenantId(), pipelineId); + private Optional find(String tenantId, String pipelineId, + String sourceTable, int chunkIndex) { + return repository.findByTenantIdAndPipelineIdAndSourceTableAndChunkIndex( + tenantId, pipelineId, sourceTable, chunkIndex); } private SnapshotCheckpoint toDomain(SnapshotCheckpointEntity e) { - return new SnapshotCheckpoint(e.getPipelineId(), e.getSourceTable(), + return new SnapshotCheckpoint(e.getPipelineId(), e.getSourceTable(), e.getChunkIndex(), e.getLastBatchNumber(), e.getRowsProcessed(), e.getCursorPos()); } - - private String tenantId() { - return TenantSupport.tenantId(); - } } diff --git a/syncflow-api/src/main/java/com/syncflow/api/snapshot/SnapshotExecutor.java b/syncflow-api/src/main/java/com/syncflow/api/snapshot/SnapshotExecutor.java index 10c1fe0..15dcfc7 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/snapshot/SnapshotExecutor.java +++ b/syncflow-api/src/main/java/com/syncflow/api/snapshot/SnapshotExecutor.java @@ -3,21 +3,25 @@ import com.syncflow.api.config.RuntimeProperties; import com.syncflow.api.connection.ConnectionMapper; import com.syncflow.api.connection.service.ConnectionService; +import com.syncflow.api.lock.DistributedLockService; import com.syncflow.api.metadata.ConnectorTypeMapper; import com.syncflow.api.pipeline.PipelineDesignerService; import com.syncflow.api.runtimestate.RuntimeStateJson; -import com.syncflow.api.snapshot.entity.SnapshotJobEntity; -import com.syncflow.api.snapshot.repository.SnapshotJobRepository; +import com.syncflow.persistence.snapshot.entity.SnapshotJobEntity; +import com.syncflow.persistence.snapshot.repository.SnapshotJobRepository; import com.syncflow.api.sse.StatusBroadcaster; import com.syncflow.core.model.ConnectionConfiguration; import com.syncflow.core.pipeline.PipelineDesign; import com.syncflow.core.pipeline.mapping.ColumnMapping; +import com.syncflow.core.pipeline.mapping.TableMapping; import com.syncflow.core.registry.ConnectorRegistry; import com.syncflow.core.snapshot.BatchInformation; +import com.syncflow.core.snapshot.ChunkRange; import com.syncflow.core.snapshot.SnapshotCheckpoint; import com.syncflow.core.snapshot.SnapshotError; import com.syncflow.core.snapshot.SnapshotJob; import com.syncflow.core.snapshot.SnapshotProgress; +import com.syncflow.core.snapshot.SnapshotStatus; import com.syncflow.core.snapshot.SnapshotStatistics; import com.syncflow.core.snapshot.pipeline.FilterProcessor; import com.syncflow.core.snapshot.pipeline.ProcessingContext; @@ -33,7 +37,10 @@ import org.springframework.stereotype.Component; import org.springframework.transaction.annotation.Transactional; +import java.math.BigDecimal; +import java.time.Duration; import java.time.Instant; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.NoSuchElementException; @@ -42,12 +49,14 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; @Component public class SnapshotExecutor { private final PipelineDesignerService pipelineService; private final ConnectionService connectionService; + private final DistributedLockService lockService; private final ConnectorRegistry connectorRegistry; private final WriterRegistry writerRegistry; private final CheckpointStore checkpointStore; @@ -65,6 +74,7 @@ public class SnapshotExecutor { public SnapshotExecutor(PipelineDesignerService pipelineService, ConnectionService connectionService, + DistributedLockService lockService, ConnectorRegistry connectorRegistry, WriterRegistry writerRegistry, CheckpointStore checkpointStore, @@ -75,6 +85,7 @@ public SnapshotExecutor(PipelineDesignerService pipelineService, RuntimeProperties runtime) { this.pipelineService = pipelineService; this.connectionService = connectionService; + this.lockService = lockService; this.connectorRegistry = connectorRegistry; this.writerRegistry = writerRegistry; this.checkpointStore = checkpointStore; @@ -87,6 +98,32 @@ public SnapshotExecutor(PipelineDesignerService pipelineService, public SnapshotJob start(String pipelineId, TenantContext tenantContext) { TenantContext.require(tenantContext); + return lockService.withLock("snapshot:" + pipelineId, tenantContext, + Duration.ofSeconds(30), () -> doStart(pipelineId, tenantContext)); + } + + /** + * Race-safe start behind the distributed lock: two pods cannot start the + * same pipeline's snapshot concurrently (S2). If a RUNNING job already + * exists for this tenant+pipeline, return it instead of spawning a second + * worker (mirrors {@code SyncOrchestrator.start()}). + */ + private SnapshotJob doStart(String pipelineId, TenantContext tenantContext) { + // A RUNNING snapshot only short-circuits start when this JVM actually + // owns a live worker for it (the existing RUNNING row's snapshotId is + // in `cancellations` and not flagged cancelled). A stale RUNNING row + // left by a crashed pod has no entry — treat it as startable, not as + // "already running", so snapshots can't get stuck RUNNING forever. + var tenantId = tenantContext.tenantId().value(); + var existing = jobRepository + .findByTenantIdAndPipelineIdOrderByCreatedAtDesc(tenantId, pipelineId) + .stream().findFirst().map(this::toDomain).orElse(null); + if (existing != null && existing.getStatus() == SnapshotStatus.RUNNING) { + var flag = cancellations.get(existing.getId().value()); + if (flag != null && !flag.get()) { + return existing; + } + } var pipeline = pipelineService.get(pipelineId); var job = new SnapshotJob(pipelineId).withRunning(); var snapshotId = job.getId().value(); @@ -118,15 +155,32 @@ public List list(TenantContext tenantContext) { public SnapshotJob cancel(String snapshotId, TenantContext tenantContext) { TenantContext.require(tenantContext); var flag = cancellations.get(snapshotId); - if (flag != null) - flag.set(true); var job = Optional.ofNullable(findOwned(snapshotId, tenantContext)) .map(this::toDomain) .orElseThrow(() -> new NoSuchElementException("Snapshot not found: " + snapshotId)); + // A cancel is only meaningful for a job that is still running. If the + // worker already finished (COMPLETED) or the job is otherwise terminal, + // cancel must not overwrite that status — the caller was too late. + if (job.getStatus() == SnapshotStatus.COMPLETED + || job.getStatus() == SnapshotStatus.FAILED + || job.getStatus() == SnapshotStatus.CANCELLED) { + return job; + } var cancelled = job.withCancelled(); - persist(cancelled, tenantContext); - // A cancelled snapshot is terminal; release its in-memory state. - remove(snapshotId); + // Serialize the cancel flag-set and the CANCELLED persist with the + // worker's terminal COMPLETED persist (both take progressLock), so the + // two terminal states cannot race: a cancel that lands mid-terminal wins + // BEFORE the worker commits, instead of overriding the status after the + // commit. The worker re-checks isCancelled() inside the same lock. + synchronized (progressLock) { + if (flag != null) + flag.set(true); + persist(cancelled, tenantContext); + } + // Do NOT release the in-memory cancel flag here. The worker must still + // observe it at its terminal check to keep the CANCELLED status from + // being overridden by a COMPLETED commit; the worker clears it when it + // finishes (executeInner's terminal branches call remove()). return cancelled; } @@ -157,6 +211,10 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex var sample = Timer.start(meterRegistry); var rowsProcessed = new AtomicLong(0); var batchesDone = new AtomicLong(0); + // DestinationWriter is single-connection and not thread-safe; writes + // across parallel chunk workers serialize on this monitor. + var writerLock = new Object(); + AtomicReference failure = new AtomicReference<>(); DestinationWriter writer = null; try { @@ -164,6 +222,7 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex var destCfg = buildDestConfig(pipeline); var connector = resolveSourceConnector(pipeline); writer = resolveWriter(pipeline); + final DestinationWriter sharedWriter = writer; writer.connect(destCfg); @@ -179,90 +238,92 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex var progress = SnapshotProgress.starting(totalRows); persist(job.withProgress(progress), tenantContext); - + final long finalTotalRows = totalRows; + final long finalTotalBatches = totalBatches; + + var parallelism = runtime.getSnapshot().getParallelism(); + var maxChunks = runtime.getSnapshot().getMaxChunks(); + // One task per (table, chunk-range) work item. Tables without a + // numeric single-column PK yield one whole-table chunk (sequential). + var workItems = new ArrayList(); for (var tm : pipeline.tableMappings()) { - if (isCancelled(job)) - break; - var ctx = new ProcessingContext(pipeline, tm); - - var checkpoint = checkpointStore.get(pipeline.id().value(), tm.sourceTable()); - // Resume from the last checkpointed cursor; else start fresh. - String cursor = (checkpoint != null) ? checkpoint.cursor() : null; - int batchNumber = (checkpoint != null) ? checkpoint.lastBatchNumber() + 1 : 0; - - var batchInfo = new BatchInformation(batchNumber, pipeline.settings().batchSize(), - tm.sourceTable(), cursor); - var page = connector.readBatch(sourceCtx, pipeline.source().schema(), - tm.sourceTable(), batchInfo); - - var chain = new FilterProcessor().andThen(new TransformProcessor()); - - while (page != null && !page.rows().isEmpty() && !isCancelled(job)) { - var batch = page.rows().stream() - .map(r -> chain.process(r, ctx)) - .filter(Objects::nonNull) - .toList(); - - if (!batch.isEmpty()) { - var destCols = tm.columnMappings().stream() - .map(ColumnMapping::destinationColumn) - .toList(); - writer.writeBatch(tm.destinationTable() != null - ? tm.destinationTable() - : tm.destinationCollection(), - destCols, batch); - } - - rowsProcessed.addAndGet(batch.size()); - batchesDone.incrementAndGet(); - var pct = totalRows > 0 ? (double) rowsProcessed.get() / totalRows * 100 : 0; - var updated = job.withProgress(new SnapshotProgress( - (int) batchesDone.get(), (int) totalBatches, - rowsProcessed.get(), totalRows, pct, 0)); - persist(updated, tenantContext); - emit(job.getId().value(), updated, tenantContext); - - meterRegistry.counter("syncflow.snapshot.rows", - "pipeline", pipeline.id().value()).increment(batch.size()); + var ranges = connector.rangeChunks(sourceCtx, pipeline.source().schema(), + tm.sourceTable(), maxChunks); + for (var range : ranges) { + workItems.add(new WorkItem(tm, range)); + } + } - // Checkpoint every N batches (configurable) — captures the keyed cursor so a - // resume continues exactly at the next row (no OFFSET drift). - if (batchesDone.get() % runtime.getSnapshot().getCheckpointIntervalBatches() == 0) { - checkpointStore.save(new SnapshotCheckpoint( - pipeline.id().value(), tm.sourceTable(), - (int) batchesDone.get(), rowsProcessed.get(), - page.nextCursor())); + var poolSize = Math.max(1, Math.min(parallelism, workItems.size())); + // Each worker thread owns ONE exclusive connector clone for its whole + // lifetime and drains a shared work queue, so no two in-flight tasks + // ever share a JDBC Connection (which is not thread-safe). A 64-chunk + // table with 4 workers still opens only 4 DB connections. + var workQueue = new java.util.concurrent.LinkedBlockingQueue(workItems); + var workerClones = new ArrayList(); + try { + for (int i = 0; i < poolSize; i++) { + workerClones.add(connector.snapshotClone(sourceCtx)); + } + var workers = new ArrayList(poolSize); + for (var workerConnector : workerClones) { + workers.add(Thread.startVirtualThread(() -> { + while (true) { + var wi = workQueue.poll(); + if (wi == null) { + break; // the queue is drained + } + try { + snapshotRange(job, pipeline, tenantContext, workerConnector, + sharedWriter, writerLock, wi.table(), wi.range(), sourceCtx, + rowsProcessed, batchesDone, finalTotalRows, finalTotalBatches); + } catch (Throwable t) { + failure.compareAndSet(null, t); + } + } + })); + } + for (var worker : workers) { + worker.join(); + } + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + } finally { + for (var clone : workerClones) { + try { + clone.disconnect(); + } catch (Exception ignored) { } - - // Next read continues from this page's cursor. - var nextBatchInfo = new BatchInformation( - (int) batchesDone.get(), pipeline.settings().batchSize(), - tm.sourceTable(), page.nextCursor()); - page = connector.readBatch(sourceCtx, pipeline.source().schema(), - tm.sourceTable(), nextBatchInfo); } } - if (isCancelled(job)) { - // Do not commit partial writes on cancel — a later resume would - // duplicate the already-written rows. - writer.rollback(); - } else { - writer.flush(); - writer.commit(); + if (failure.get() != null) { + throw new RuntimeException("Snapshot range failed", failure.get()); } var elapsed = sample.stop(timer); - if (!isCancelled(job)) { - var stats = new SnapshotStatistics(totalRows, rowsProcessed.get(), - batchesDone.get(), totalBatches, 0, 0, - job.getCreatedAt(), Instant.now(), elapsed / 1_000_000); - var completed = job.withCompleted(stats); - persist(completed, tenantContext); - emit(job.getId().value(), completed, tenantContext); - checkpointStore.deleteAll(pipeline.id().value()); - // Terminal and durable; release worker state so the in-memory - // maps cannot grow unbounded across snapshots. + // Decide the terminal state under progressLock so it cannot race the + // cancel() path. The in-memory flag is re-checked inside the lock: + // if cancel() ran between the loop's check and here, the snapshot is + // CANCELLED and must not be overridden by COMPLETED. + synchronized (progressLock) { + if (isCancelled(job)) { + // Do not commit partial writes on cancel — a later resume would + // duplicate the already-written rows. + writer.rollback(); + } else { + writer.flush(); + writer.commit(); + var stats = new SnapshotStatistics(totalRows, rowsProcessed.get(), + batchesDone.get(), totalBatches, 0, 0, + job.getCreatedAt(), Instant.now(), elapsed / 1_000_000); + var completed = job.withCompleted(stats); + persist(completed, tenantContext); + emit(job.getId().value(), completed, tenantContext); + checkpointStore.deleteAll(tenantContext.tenantId().value(), pipeline.id().value()); + } + // Release worker state once the worker has decided its terminal + // state (COMPLETED or CANCELLED stood). remove(job.getId().value()); } } catch (Exception e) { @@ -275,15 +336,191 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex } var error = new SnapshotError("SNAPSHOT_FAILED", e.getMessage(), (int) batchesDone.get(), Instant.now()); - var failed = job.withFailed(List.of(error)); - persist(failed, tenantContext); - emit(job.getId().value(), failed, tenantContext); - remove(job.getId().value()); + // A cancellation outranks a failure — the user asked to stop, so the + // CANCELLED status (already persisted by cancel()) must not be + // overwritten by FAILED. Same lock discipline as the completion path. + synchronized (progressLock) { + if (!isCancelled(job)) { + var failed = job.withFailed(List.of(error)); + persist(failed, tenantContext); + emit(job.getId().value(), failed, tenantContext); + } + remove(job.getId().value()); + } meterRegistry.counter("syncflow.snapshot.errors", "pipeline", pipeline.id().value()).increment(); } } + /** + * Snapshot one PK-range chunk of one table: keyset-paginate the range, + * filter/transform, and batch-write to the destination. Writes serialize on + * {@code writerLock}; read/transform run in parallel across chunk workers. + * Resume continues from the chunk's own checkpoint cursor. + */ + private void snapshotRange(SnapshotJob job, PipelineDesign pipeline, TenantContext tenantContext, + SnapshotCapableConnector connector, DestinationWriter writer, Object writerLock, + TableMapping tm, ChunkRange range, ConnectorContext sourceCtx, + AtomicLong rowsProcessed, AtomicLong batchesDone, long totalRows, long totalBatches) { + var ctx = new ProcessingContext(pipeline, tm); + var checkpoint = checkpointStore.get( + tenantContext.tenantId().value(), pipeline.id().value(), tm.sourceTable(), range.index()); + // Resume only from a checkpoint cursor that still lies inside this + // chunk's bounds. A legacy whole-table checkpoint shares chunk_index=0 + // with chunk 0 — resuming from its (potentially out-of-range) cursor + // would silently skip rows in [start, cursor). Out-of-range → start the + // chunk fresh. + String cursor = (checkpoint != null && cursorWithinRange(checkpoint.cursor(), range)) + ? checkpoint.cursor() + : null; + int batchNumber = (checkpoint != null) ? checkpoint.lastBatchNumber() + 1 : 0; + // Per-chunk batch counter for checkpoint/progress cadence. It is local + // to this worker (resumed from the chunk's checkpoint) so cadence is + // accurate per chunk rather than diluted across parallel workers sharing + // the global counter. It starts one below the FIRST read's ordinal: + // each increment yields the ordinal of the page just read (fresh: 0,1,2… + // resume: L+1, L+2…), and the next page is chunkBatch + 1 — this is what + // makes offset-based connectors paginate correctly on resume. + var chunkBatchCounter = new AtomicLong(batchNumber - 1); + + // Destination columns, table, and upsert keys are constant for the whole + // chunk — derive once and reuse across every page instead of rebuilding + // them per batch. + var destCols = tm.columnMappings().stream() + .map(ColumnMapping::destinationColumn) + .toList(); + var destTable = tm.destinationTable() != null + ? tm.destinationTable() + : tm.destinationCollection(); + var keyCols = tm.primaryKey() != null ? tm.primaryKey().destinationColumns() : null; + var useUpsert = keyCols != null && !keyCols.isEmpty(); + var chain = new FilterProcessor().andThen(new TransformProcessor()); + + // Do not read the first page at all if the snapshot was already + // cancelled — otherwise a cancel landing before the loop-top check would + // still write (and auto-commit) this chunk's first batch. + if (isCancelled(job)) { + return; + } + var batchInfo = new BatchInformation(batchNumber, pipeline.settings().batchSize(), + tm.sourceTable(), cursor, range); + var page = connector.readBatch(sourceCtx, pipeline.source().schema(), + tm.sourceTable(), batchInfo); + + while (page != null && !page.rows().isEmpty() && !isCancelled(job)) { + var batch = page.rows().stream() + .map(r -> chain.process(r, ctx)) + .filter(Objects::nonNull) + .toList(); + + if (!batch.isEmpty()) { + synchronized (writerLock) { + // Upsert when a destination PK is mapped so a resume that + // re-reads already-committed rows (the pooled writer + // auto-commits each batch; there is no transaction to roll + // back) is idempotent instead of inserting duplicates or + // tripping a constraint violation. + if (useUpsert) { + writer.upsertBatch(destTable, destCols, batch, keyCols); + } else { + writer.writeBatch(destTable, destCols, batch); + } + } + } + + rowsProcessed.addAndGet(batch.size()); + batchesDone.incrementAndGet(); + // Per-chunk batch counter for checkpoint cadence — the shared global + // counter would spread checkpoints unevenly across parallel workers. + var chunkBatch = chunkBatchCounter.incrementAndGet(); + meterRegistry.counter("syncflow.snapshot.rows", + "pipeline", pipeline.id().value()).increment(batch.size()); + + // Checkpoint every N batches (configurable) — captures the chunk's + // keyed cursor so a resume continues exactly at the next row. + if (chunkBatch % runtime.getSnapshot().getCheckpointIntervalBatches() == 0) { + checkpointStore.save(tenantContext.tenantId().value(), new SnapshotCheckpoint( + pipeline.id().value(), tm.sourceTable(), range.index(), + (int) chunkBatch, rowsProcessed.get(), page.nextCursor())); + } + + // Publish live progress every N batches, serialized so the shared + // (job, progress) read-modify-write cannot lose updates across + // parallel workers. + synchronized (progressLock) { + // Re-check cancellation under the lock before publishing. A + // cancel() that landed after the loop-top check persists + // CANCELLED under this same monitor; publishing progress here + // would write a RUNNING-status job over it and strand a zombie + // RUNNING row with no live worker. + if (!isCancelled(job) + && chunkBatch % runtime.getSnapshot().getProgressPublishIntervalBatches() == 0) { + var pct = totalRows > 0 ? (double) rowsProcessed.get() / totalRows * 100 : 0; + var updated = job.withProgress(new SnapshotProgress( + (int) batchesDone.get(), (int) totalBatches, + rowsProcessed.get(), totalRows, pct, 0)); + persist(updated, tenantContext); + emit(job.getId().value(), updated, tenantContext); + } + } + + // Next read continues from this page's cursor within this chunk. Its + // batchNumber is the ordinal of the NEXT batch: this page was + // chunkBatch (after the increment above), so the next is + // chunkBatch + 1. Keeping the ordinal per-chunk (not the shared + // global counter) means offset-based connectors (Mongo, PK-less + // JDBC) paginate by batchNumber * batchSize without skipping or + // re-reading rows. + var nextBatchInfo = new BatchInformation( + (int) chunkBatch + 1, pipeline.settings().batchSize(), + tm.sourceTable(), page.nextCursor(), range); + page = connector.readBatch(sourceCtx, pipeline.source().schema(), + tm.sourceTable(), nextBatchInfo); + } + } + + /** + * Aggregate live-progress publication across parallel chunk workers is + * serialized on this monitor; {@link #persist} reads the whole job payload + * and writes it back, so two workers persisting concurrently would clobber + * each other's progress. + */ + private final Object progressLock = new Object(); + + /** A (table mapping, chunk range) work item for the parallel snapshot. */ + private record WorkItem(TableMapping table, ChunkRange range) { + } + + /** + * A resume cursor is meaningful for a chunk only if it lies within + * {@code [start, end)}. Numeric cursors/bounds compare by value; anything + * non-numeric (uuid/text) or out of range is not a valid resume point. + */ + private static boolean cursorWithinRange(String cursor, ChunkRange range) { + if (cursor == null || range == null) { + return false; + } + if (range.start() instanceof Number start) { + try { + BigDecimal c = new BigDecimal(cursor); + BigDecimal lo = new BigDecimal(start.toString()); + // Start-inclusive. The end bound, when present, is exclusive; + // a null end (open-ended last chunk / whole table) accepts any + // cursor at or after start. + if (c.compareTo(lo) < 0) { + return false; + } + return range.end() == null + || !(range.end() instanceof Number end) + || c.compareTo(new BigDecimal(end.toString())) < 0; + } catch (NumberFormatException e) { + return false; + } + } + // Whole-table (unbounded start) ranges: any cursor is valid. + return range.start() == null; + } + /** Live-status event emitted on every progress/state change for a snapshot. */ private void emit(String snapshotId, SnapshotJob job, TenantContext tenantContext) { // Tenant-scoped SSE key (matching the tenantOf map) so streams don't cross. diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/DeadLetterQueue.java b/syncflow-api/src/main/java/com/syncflow/api/sync/DeadLetterQueue.java index 1894fd4..4fb0ce4 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/DeadLetterQueue.java +++ b/syncflow-api/src/main/java/com/syncflow/api/sync/DeadLetterQueue.java @@ -2,8 +2,8 @@ import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.ObjectMapper; -import com.syncflow.api.sync.entity.DeadLetterEventEntity; -import com.syncflow.api.sync.repository.DeadLetterEventRepository; +import com.syncflow.persistence.sync.entity.DeadLetterEventEntity; +import com.syncflow.persistence.sync.repository.DeadLetterEventRepository; import com.syncflow.core.cdc.CDCEvent; import com.syncflow.core.sync.FailureReason; import com.syncflow.core.sync.dlq.DeadLetterEvent; diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/EventIdempotencyStore.java b/syncflow-api/src/main/java/com/syncflow/api/sync/EventIdempotencyStore.java index af2ba83..7dae786 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/EventIdempotencyStore.java +++ b/syncflow-api/src/main/java/com/syncflow/api/sync/EventIdempotencyStore.java @@ -1,7 +1,7 @@ package com.syncflow.api.sync; -import com.syncflow.api.sync.entity.ProcessedEventEntity; -import com.syncflow.api.sync.repository.ProcessedEventRepository; +import com.syncflow.persistence.sync.entity.ProcessedEventEntity; +import com.syncflow.persistence.sync.repository.ProcessedEventRepository; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.scheduling.annotation.Scheduled; diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/SyncOrchestrator.java b/syncflow-api/src/main/java/com/syncflow/api/sync/SyncOrchestrator.java index 3177aa9..00cba62 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/SyncOrchestrator.java +++ b/syncflow-api/src/main/java/com/syncflow/api/sync/SyncOrchestrator.java @@ -4,8 +4,8 @@ import com.syncflow.api.pipeline.PipelineDesignerService; import com.syncflow.api.runtimestate.RuntimeStateJson; import com.syncflow.api.sse.StatusBroadcaster; -import com.syncflow.api.sync.entity.SyncJobEntity; -import com.syncflow.api.sync.repository.SyncJobRepository; +import com.syncflow.persistence.sync.entity.SyncJobEntity; +import com.syncflow.persistence.sync.repository.SyncJobRepository; import com.syncflow.api.config.RuntimeProperties; import com.syncflow.core.cdc.CDCEvent; import com.syncflow.core.cdc.CDCOperation; diff --git a/syncflow-api/src/main/java/com/syncflow/api/user/UserService.java b/syncflow-api/src/main/java/com/syncflow/api/user/UserService.java index 70fa1fe..a3de4f6 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/user/UserService.java +++ b/syncflow-api/src/main/java/com/syncflow/api/user/UserService.java @@ -1,7 +1,7 @@ package com.syncflow.api.user; -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 org.springframework.security.crypto.password.PasswordEncoder; import org.springframework.stereotype.Service; diff --git a/syncflow-api/src/main/java/com/syncflow/api/workflow/WorkflowScheduler.java b/syncflow-api/src/main/java/com/syncflow/api/workflow/WorkflowScheduler.java index a4ff895..524c9c8 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/workflow/WorkflowScheduler.java +++ b/syncflow-api/src/main/java/com/syncflow/api/workflow/WorkflowScheduler.java @@ -2,8 +2,8 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.syncflow.api.runtimestate.RuntimeStateJson; -import com.syncflow.api.workflow.entity.WorkflowInstanceEntity; -import com.syncflow.api.workflow.repository.WorkflowInstanceRepository; +import com.syncflow.persistence.workflow.entity.WorkflowInstanceEntity; +import com.syncflow.persistence.workflow.repository.WorkflowInstanceRepository; import com.syncflow.core.workflow.TaskExecution; import com.syncflow.core.workflow.WorkflowId; import com.syncflow.core.workflow.WorkflowInstance; diff --git a/syncflow-api/src/test/java/com/syncflow/api/cdc/CaptureLifecycleUnitTest.java b/syncflow-api/src/test/java/com/syncflow/api/cdc/CaptureLifecycleUnitTest.java index aac74bc..5a9d26d 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/cdc/CaptureLifecycleUnitTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/cdc/CaptureLifecycleUnitTest.java @@ -1,7 +1,7 @@ package com.syncflow.api.cdc; import com.fasterxml.jackson.databind.ObjectMapper; -import com.syncflow.api.cdc.repository.ActiveCaptureRepository; +import com.syncflow.persistence.cdc.repository.ActiveCaptureRepository; import com.syncflow.api.connection.service.ConnectionService; import com.syncflow.api.kafka.KafkaCdcConsumer; import com.syncflow.api.lock.DistributedLockService; diff --git a/syncflow-api/src/test/java/com/syncflow/api/cdc/OffsetStoreTest.java b/syncflow-api/src/test/java/com/syncflow/api/cdc/OffsetStoreTest.java index 95173f3..13d1022 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/cdc/OffsetStoreTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/cdc/OffsetStoreTest.java @@ -1,8 +1,8 @@ package com.syncflow.api.cdc; import com.fasterxml.jackson.databind.ObjectMapper; -import com.syncflow.api.cdc.entity.CdcOffsetEntity; -import com.syncflow.api.cdc.repository.CdcOffsetRepository; +import com.syncflow.persistence.cdc.entity.CdcOffsetEntity; +import com.syncflow.persistence.cdc.repository.CdcOffsetRepository; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Nested; diff --git a/syncflow-api/src/test/java/com/syncflow/api/db/DatabaseMigrationValidationTest.java b/syncflow-api/src/test/java/com/syncflow/api/db/DatabaseMigrationValidationTest.java index 3c11a65..d9e4e5a 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/db/DatabaseMigrationValidationTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/db/DatabaseMigrationValidationTest.java @@ -1,7 +1,7 @@ package com.syncflow.api.db; import com.syncflow.api.config.AbstractIntegrationTest; -import com.syncflow.api.connection.repository.ConnectionRepository; +import com.syncflow.persistence.connection.repository.ConnectionRepository; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.jdbc.core.JdbcTemplate; diff --git a/syncflow-api/src/test/java/com/syncflow/api/pipeline/PipelineDesignerServiceTest.java b/syncflow-api/src/test/java/com/syncflow/api/pipeline/PipelineDesignerServiceTest.java index 9bdf4e3..a31e133 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/pipeline/PipelineDesignerServiceTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/pipeline/PipelineDesignerServiceTest.java @@ -4,13 +4,13 @@ import com.fasterxml.jackson.databind.SerializationFeature; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; 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.mapper.PipelineDesignEntityMapperImpl; -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.DestinationReference; import com.syncflow.core.pipeline.PipelineDesign; import com.syncflow.core.pipeline.PipelineName; diff --git a/syncflow-api/src/test/java/com/syncflow/api/pipeline/mapper/PipelineEntityMapperTest.java b/syncflow-api/src/test/java/com/syncflow/api/pipeline/mapper/PipelineEntityMapperTest.java index 491d946..7b3d56d 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/pipeline/mapper/PipelineEntityMapperTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/pipeline/mapper/PipelineEntityMapperTest.java @@ -3,7 +3,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.SerializationFeature; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; -import com.syncflow.api.pipeline.entity.PipelineEntity; +import com.syncflow.persistence.pipeline.entity.PipelineEntity; import com.syncflow.core.model.ConnectionConfiguration; import com.syncflow.core.model.ConnectorType; import com.syncflow.core.model.Pipeline; diff --git a/syncflow-api/src/test/java/com/syncflow/api/security/AuthServiceTest.java b/syncflow-api/src/test/java/com/syncflow/api/security/AuthServiceTest.java index be6a5f8..83ba92d 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/security/AuthServiceTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/security/AuthServiceTest.java @@ -5,8 +5,8 @@ import com.nimbusds.jose.jwk.OctetSequenceKey; import com.nimbusds.jose.jwk.source.ImmutableJWKSet; import com.syncflow.api.config.JwtProperties; -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 org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.security.authentication.AuthenticationManager; diff --git a/syncflow-api/src/test/java/com/syncflow/api/security/DbUserDetailsServiceTest.java b/syncflow-api/src/test/java/com/syncflow/api/security/DbUserDetailsServiceTest.java index ac4d4c1..9018045 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/security/DbUserDetailsServiceTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/security/DbUserDetailsServiceTest.java @@ -1,7 +1,7 @@ package com.syncflow.api.security; -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 org.junit.jupiter.api.Test; import java.util.Optional; diff --git a/syncflow-api/src/test/java/com/syncflow/api/sync/MultiTableDispatchTest.java b/syncflow-api/src/test/java/com/syncflow/api/sync/MultiTableDispatchTest.java index e201ec3..65477d2 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/sync/MultiTableDispatchTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/sync/MultiTableDispatchTest.java @@ -5,7 +5,7 @@ import com.syncflow.api.pipeline.PipelineDesignerService; import com.syncflow.api.runtimestate.RuntimeStateJson; import com.syncflow.api.sse.StatusBroadcaster; -import com.syncflow.api.sync.repository.SyncJobRepository; +import com.syncflow.persistence.sync.repository.SyncJobRepository; import com.syncflow.tenant.TenantContext; import com.syncflow.tenant.TenantId; import com.fasterxml.jackson.databind.ObjectMapper; diff --git a/syncflow-api/src/test/java/com/syncflow/api/sync/SyncEngineUnitTest.java b/syncflow-api/src/test/java/com/syncflow/api/sync/SyncEngineUnitTest.java index 0d1c432..9bd30a4 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/sync/SyncEngineUnitTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/sync/SyncEngineUnitTest.java @@ -6,10 +6,10 @@ import com.syncflow.tenant.TenantId; import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; -import com.syncflow.api.sync.entity.DeadLetterEventEntity; -import com.syncflow.api.sync.entity.ProcessedEventEntity; -import com.syncflow.api.sync.repository.DeadLetterEventRepository; -import com.syncflow.api.sync.repository.ProcessedEventRepository; +import com.syncflow.persistence.sync.entity.DeadLetterEventEntity; +import com.syncflow.persistence.sync.entity.ProcessedEventEntity; +import com.syncflow.persistence.sync.repository.DeadLetterEventRepository; +import com.syncflow.persistence.sync.repository.ProcessedEventRepository; import com.syncflow.core.cdc.CDCEvent; import com.syncflow.core.cdc.CDCOperation; import com.syncflow.core.cdc.EventHeader; diff --git a/syncflow-api/src/test/java/com/syncflow/api/user/UserServiceTest.java b/syncflow-api/src/test/java/com/syncflow/api/user/UserServiceTest.java index 298663a..288a11c 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/user/UserServiceTest.java +++ b/syncflow-api/src/test/java/com/syncflow/api/user/UserServiceTest.java @@ -1,7 +1,7 @@ package com.syncflow.api.user; -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 org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder; diff --git a/syncflow-connectors/src/main/java/com/syncflow/connector/snapshot/AbstractJdbcSnapshotConnector.java b/syncflow-connectors/src/main/java/com/syncflow/connector/snapshot/AbstractJdbcSnapshotConnector.java index e56d9e5..350e92a 100644 --- a/syncflow-connectors/src/main/java/com/syncflow/connector/snapshot/AbstractJdbcSnapshotConnector.java +++ b/syncflow-connectors/src/main/java/com/syncflow/connector/snapshot/AbstractJdbcSnapshotConnector.java @@ -2,13 +2,17 @@ import com.syncflow.connector.metadata.AbstractJdbcMetadataConnector; import com.syncflow.core.snapshot.BatchInformation; +import com.syncflow.core.snapshot.ChunkRange; import com.syncflow.core.spi.ConnectorContext; import com.syncflow.core.spi.SnapshotCapableConnector; +import java.math.BigDecimal; +import java.math.MathContext; import java.sql.ResultSetMetaData; import java.sql.SQLException; import java.util.ArrayList; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; public abstract class AbstractJdbcSnapshotConnector @@ -69,26 +73,138 @@ public Page readBatch(ConnectorContext ctx, String schema, String table, } /** - * Read a page using a keyset cursor over the PK: {@code WHERE pk > :cursor - * ORDER BY pk LIMIT size}. The last PK value becomes the next cursor, so resume - * and concurrent writes stay consistent. + * A disjoint, already-connected clone for a parallel snapshot worker. Each + * worker gets its own {@code java.sql.Connection} via the concrete + * connector's no-arg constructor, so a single Connection is never shared + * across threads (JDBC Connection is not thread-safe). + */ + @Override + public SnapshotCapableConnector snapshotClone(ConnectorContext ctx) { + try { + var clone = getClass().getDeclaredConstructor().newInstance(); + clone.connect(ctx); + return clone; + } catch (ReflectiveOperationException e) { + throw new IllegalStateException( + "Snapshot connector " + getClass().getSimpleName() + " needs a no-arg constructor to clone", e); + } + } + + /** + * Split a table into PK-range chunks (F15). Returns the whole table as one + * chunk when there is no single-column PK, or when the PK is not numeric + * (uuid/text/date ranges cannot be split arithmetically). Chunks are + * disjoint and gapless: {@code [min, max]} split evenly for a numeric PK. + */ + @Override + public List rangeChunks(ConnectorContext ctx, String schema, String table, + int chunkCount) { + ensureConnected(ctx); + var pkCol = primaryKeyColumn(ctx, schema, table); + if (pkCol == null) { + return List.of(ChunkRange.whole()); + } + var minMax = minMaxPk(schema, table, pkCol); + if (minMax == null || minMax[0] == null || minMax[1] == null + || !(minMax[0] instanceof Number left) || !(minMax[1] instanceof Number right)) { + // Non-numeric PK (uuid / text / date) — cannot split by value ranges. + return List.of(ChunkRange.whole()); + } + // Numeric PKs split into disjoint [start, end) value ranges. Integer PKs + // (BIGINT/INT) split arithmetically; a decimal (NUMERIC) PK splits on + // exact BigDecimal so fractional values are not truncated and dropped. + if (left instanceof BigDecimal) { + return decimalRanges((BigDecimal) left, (BigDecimal) right, chunkCount); + } + long min = left.longValue(); + long max = right.longValue(); + int chunks = Math.max(1, chunkCount); + var list = new ArrayList(chunks); + // Split the [min, max] domain arithmetically in BigDecimal so neither + // `max - min + 1` nor `max + 1` can overflow a signed long (a BIGINT PK + // may span Long.MIN_VALUE..Long.MAX_VALUE). Bounds are emitted back as + // native Longs so the driver binds them to the bigint column. A chunk + // whose boundary passes max is the effective last chunk and carries a + // null end (open-ended) — this also sidesteps end == max + 1 overflow. + var hi = BigDecimal.valueOf(max); + var span = hi.subtract(BigDecimal.valueOf(min)).add(BigDecimal.ONE); + var step = span.divide(BigDecimal.valueOf(chunks), MathContext.DECIMAL128) + .setScale(0, java.math.RoundingMode.CEILING) + .max(BigDecimal.ONE); + var start = BigDecimal.valueOf(min); + for (int i = 0; i < chunks; i++) { + var boundary = start.add(step); + Long end; + if (boundary.compareTo(hi) > 0) { + // Remaining domain [start, max] fits in this last chunk — open end. + end = null; + } else { + end = boundary.longValueExact(); + } + list.add(new ChunkRange(i, start.longValueExact(), end)); + if (end == null) { + break; + } + start = boundary; + } + return list; + } + + /** Split a BigDecimal PK range into disjoint [start, end) chunks exactly. */ + private static List decimalRanges(BigDecimal min, BigDecimal max, int chunkCount) { + int chunks = Math.max(1, chunkCount); + var span = max.subtract(min); + var step = span.divide(BigDecimal.valueOf(chunks), MathContext.DECIMAL128) + .max(BigDecimal.ONE); + var list = new ArrayList(chunks); + var start = min; + for (int i = 0; i < chunks; i++) { + var end = i == chunks - 1 ? max.add(BigDecimal.ONE) : start.add(step); + list.add(new ChunkRange(i, start, end)); + if (end.compareTo(max) > 0) { + break; + } + start = end; + } + return list; + } + + /** + * Read a page within a chunk range: {@code WHERE pk >= start AND pk < end + * (plus > cursor on resume) ORDER BY pk LIMIT size}. */ private Page readKeysetPage(ConnectorContext ctx, String schema, String table, BatchInformation batchInfo, String pkCol) { var cursor = batchInfo.cursor(); - String sql = "SELECT * FROM " + schema + "." + table + - " WHERE " + pkCol + (cursor == null ? " IS NOT NULL" : " > ?") + - " ORDER BY " + pkCol + - " LIMIT " + batchInfo.batchSize(); + var chunk = batchInfo.chunkRange(); + var clause = new StringBuilder(" WHERE " + pkCol); + if (cursor != null) { + clause.append(" > ?"); + } else if (chunk != null && chunk.start() != null) { + clause.append(" >= ?"); + } else { + clause.append(" IS NOT NULL"); + } + if (chunk != null && chunk.end() != null) { + clause.append(" AND " + pkCol + " < ?"); + } + String sql = "SELECT * FROM " + schema + "." + table + clause + + " ORDER BY " + pkCol + " LIMIT " + batchInfo.batchSize(); var rows = new ArrayList>(); Object lastPk = null; try (var stmt = jdbcConnection.prepareStatement(sql)) { + int param = 1; if (cursor != null) { - // The cursor round-trips through a String (SPI contract). Binding with - // setObject lets the driver coerce to the PK's column type; lexically - // the value is compared to a seekable key, which holds for int, bigint, - // uuid, and text PKs — the types these connectors support. - stmt.setObject(1, cursor); + // The cursor round-trips through a String (SPI contract). For a + // numeric PK the range bounds are native Numbers, so coerce the + // cursor to Long to match — a raw String fails the type check + // ("operator does not exist: bigint >= character varying"). + stmt.setObject(param++, coerceCursor(cursor, chunk)); + } else if (chunk != null && chunk.start() != null) { + stmt.setObject(param++, chunk.start()); + } + if (chunk != null && chunk.end() != null) { + stmt.setObject(param++, chunk.end()); } var rs = stmt.executeQuery(); var meta = rs.getMetaData(); @@ -113,6 +229,60 @@ private Page readKeysetPage(ConnectorContext ctx, String schema, String table, return rows.isEmpty() ? Page.empty() : Page.of(rows, nextCursor); } + /** + * Bind the keyset cursor as the PK's native type. The SPI cursor is a + * String; for numeric PKs the driver rejects a bare String against a + * bigint column ("operator does not exist: bigint >= character varying"), + * so parse a numeric cursor to {@link Long}. Non-numeric PKs (uuid/text) + * fall through to the raw String, which the driver handles. + * + *

+ * Whether the PK is numeric is determined by the chunk BOUNDS, not by + * the cursor's appearance: {@code rangeChunks} returns a whole chunk + * (null bounds) for every non-numeric PK (uuid/text/date) as well as for + * no-PK tables. Coercing an all-digit cursor to Long on such a chunk would + * break a TEXT keyseek — e.g. a VARCHAR PK storing '00123' bound as Long + * 123 fails the type check on Postgres and silently skips rows on engines + * that coerce. So only coerce when the chunk bounds are real Numbers. + */ + private static Object coerceCursor(String cursor, ChunkRange chunk) { + if (cursor == null || cursor.isEmpty()) { + return cursor; + } + // Numeric-range chunks carry Number bounds; coerce the cursor to the + // same numeric type so the comparison operator matches the PK column. + if (chunk != null && chunk.start() instanceof BigDecimal) { + try { + return new BigDecimal(cursor); + } catch (NumberFormatException e) { + return cursor; + } + } + if (chunk != null && chunk.start() instanceof Number) { + try { + return Long.valueOf(cursor); + } catch (NumberFormatException e) { + return cursor; + } + } + // Whole-chunk (non-numeric PK / no PK): keep the raw String cursor. + return cursor; + } + + /** MIN/MAX of the PK column as bound driver values, or null on failure. */ + private Object[] minMaxPk(String schema, String table, String pkCol) { + var sql = "SELECT MIN(" + pkCol + "), MAX(" + pkCol + ") FROM " + schema + "." + table; + try (var stmt = jdbcConnection.createStatement(); + var rs = stmt.executeQuery(sql)) { + if (rs.next()) { + return new Object[]{rs.getObject(1), rs.getObject(2)}; + } + } catch (SQLException e) { + return null; + } + return null; + } + /** OFFSET/LIMIT fallback for tables without a single-column PK. */ private Page readOffsetPage(String schema, String table, BatchInformation batchInfo) { int offset = batchInfo.cursor() != null diff --git a/syncflow-connectors/src/main/java/com/syncflow/connector/writer/JdbcBatchWriter.java b/syncflow-connectors/src/main/java/com/syncflow/connector/writer/JdbcBatchWriter.java index c0839e7..eebe5d4 100644 --- a/syncflow-connectors/src/main/java/com/syncflow/connector/writer/JdbcBatchWriter.java +++ b/syncflow-connectors/src/main/java/com/syncflow/connector/writer/JdbcBatchWriter.java @@ -70,11 +70,15 @@ public void writeBatch(String table, List columns, List columns, List= 1000) { @@ -261,6 +266,19 @@ public void rollback() { connection.rollback(); } catch (SQLException e) { throw new RuntimeException("Rollback failed", e); + } finally { + // Discard any buffered (not-yet-flushed) rows so a failed run's + // residual buffer cannot leak into the next pipeline's destination + // when its first writeBatch hits a different table. The buffered + // rows were never committed; the cursor checkpoint sits before + // them and resume re-reads them. + buffer.clear(); + deleteBuffer.clear(); + deleteColumns.clear(); + currentTable = null; + currentColumns = null; + currentUpsertKeys = null; + currentInsertSql = null; } } diff --git a/syncflow-connectors/src/main/java/com/syncflow/connector/writer/MySqlWriter.java b/syncflow-connectors/src/main/java/com/syncflow/connector/writer/MySqlWriter.java index ee2a8b4..f90a94e 100644 --- a/syncflow-connectors/src/main/java/com/syncflow/connector/writer/MySqlWriter.java +++ b/syncflow-connectors/src/main/java/com/syncflow/connector/writer/MySqlWriter.java @@ -3,7 +3,9 @@ import com.syncflow.core.model.ConnectionConfiguration; import org.springframework.stereotype.Component; +import java.util.List; import java.util.Properties; +import java.util.stream.Collectors; @Component public class MySqlWriter extends PooledJdbcBatchWriter { @@ -26,4 +28,31 @@ protected Properties jdbcProperties(ConnectionConfiguration config) { props.setProperty("password", config.password()); return props; } + + /** + * MySQL UPSERT: {@code INSERT ... ON DUPLICATE KEY UPDATE ...}. The base + * {@link JdbcBatchWriter#upsertSql} emits Postgres {@code ON CONFLICT} + * syntax, which MySQL rejects; without this override the snapshot/CDC + * upsert path would fail at write time for a MySQL destination. + */ + @Override + protected String upsertSql(String table, List columns, List keyColumns) { + var cols = String.join(", ", columns); + var params = "?" + ", ?".repeat(columns.size() - 1); + var updateClause = columns.stream() + .filter(c -> !keyColumns.contains(c)) + .map(c -> c + " = VALUES(" + c + ")") + .collect(Collectors.joining(", ")); + // VALUES(col) is the only ON DUPLICATE KEY UPDATE form that works on + // every MySQL (5.7 .. 8.0) and all MariaDB. The row-alias form + // INSERT ... AS new ... c = new.c requires MySQL >= 8.0.19 and is + // unsupported on MariaDB, so it would fail at write time for those + // targets. VALUES(col) is deprecated (not removed) since 8.0.20 — + // the deprecation warning is acceptable vs a hard parse error. + return "INSERT INTO " + table + " (" + cols + ") VALUES (" + params + ")" + + " ON DUPLICATE KEY UPDATE " + + (updateClause.isEmpty() + ? columns.getFirst() + " = VALUES(" + columns.getFirst() + ")" + : updateClause); + } } diff --git a/syncflow-connectors/src/main/java/com/syncflow/connector/writer/PooledJdbcBatchWriter.java b/syncflow-connectors/src/main/java/com/syncflow/connector/writer/PooledJdbcBatchWriter.java index 3d2d847..dd98bb4 100644 --- a/syncflow-connectors/src/main/java/com/syncflow/connector/writer/PooledJdbcBatchWriter.java +++ b/syncflow-connectors/src/main/java/com/syncflow/connector/writer/PooledJdbcBatchWriter.java @@ -103,7 +103,10 @@ public void commit() { @Override public void rollback() { - // Auto-commit per batch — nothing to roll back. + // Auto-commit per batch — nothing to roll back in the store. But + // discard any buffered (not-yet-flushed) rows so a failed run's + // residual buffer cannot leak into the next pipeline's destination. + super.rollback(); } @Override diff --git a/syncflow-connectors/src/test/java/com/syncflow/connector/snapshot/ChunkedSnapshotCorrectnessTest.java b/syncflow-connectors/src/test/java/com/syncflow/connector/snapshot/ChunkedSnapshotCorrectnessTest.java new file mode 100644 index 0000000..e6f5605 --- /dev/null +++ b/syncflow-connectors/src/test/java/com/syncflow/connector/snapshot/ChunkedSnapshotCorrectnessTest.java @@ -0,0 +1,118 @@ +package com.syncflow.connector.snapshot; + +import com.syncflow.connector.metadata.PostgresMetadataConnector; +import com.syncflow.core.model.ConnectionConfiguration; +import com.syncflow.core.model.ConnectorType; +import com.syncflow.core.snapshot.BatchInformation; +import com.syncflow.core.snapshot.ChunkRange; +import com.syncflow.core.spi.ConnectorContext; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledIfSystemProperty; +import org.testcontainers.containers.PostgreSQLContainer; +import org.testcontainers.junit.jupiter.Container; +import org.testcontainers.junit.jupiter.Testcontainers; + +import java.sql.Connection; +import java.sql.DriverManager; +import java.util.HashSet; +import java.util.Map; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * F15 correctness: parallel PK-range chunking must produce exactly the rows of + * the whole table — no duplicates, no gaps — for both a fresh chunked read and + * a resume-with-cursor within a chunk. A PG table with a numeric PK is split + * into several ranges; each range is paginated with keyset; the union must + * equal the full table's PK set. + */ +@Testcontainers +@Tag("integration") +@EnabledIfSystemProperty(named = "tests.integration", matches = "true") +class ChunkedSnapshotCorrectnessTest { + + @Container + static PostgreSQLContainer postgres = new PostgreSQLContainer<>("postgres:16-alpine") + .withDatabaseName("testdb") + .withUsername("testuser") + .withPassword("testpass"); + + static PostgresMetadataConnector connector = new PostgresMetadataConnector(); + static ConnectorContext ctx; + + @BeforeAll + static void seed() throws Exception { + try (Connection conn = DriverManager.getConnection( + postgres.getJdbcUrl(), postgres.getUsername(), postgres.getPassword()); + var stmt = conn.createStatement()) { + stmt.execute("CREATE TABLE chunked_rows (id BIGINT PRIMARY KEY, label TEXT)"); + // 1000 rows, ids 1..1000. + stmt.execute("INSERT INTO chunked_rows SELECT g, 'row-' || g FROM generate_series(1, 1000) g"); + } + var config = new ConnectionConfiguration(ConnectorType.POSTGRESQL, postgres.getHost(), + postgres.getMappedPort(5432), "testdb", "testuser", "testpass", Map.of()); + ctx = new ConnectorContext(config, Map.of()); + connector.connect(ctx); + } + + @Test + void chunkedReadCoversWholeTableExactly() { + var ranges = connector.rangeChunks(ctx, "public", "chunked_rows", 8); + assertTrue(ranges.size() > 1, "expected >1 chunk for a 1000-row table, got " + ranges.size()); + + Set seen = new HashSet<>(); + for (var range : ranges) { + readRange(range, null, seen); + } + assertEquals(1000, seen.size(), "chunked read must cover all 1000 rows exactly"); + for (long i = 1; i <= 1000; i++) { + assertTrue(seen.contains(i), "missing id " + i); + } + } + + @Test + void resumeWithinChunkDoesNotDuplicateOrSkip() { + var ranges = connector.rangeChunks(ctx, "public", "chunked_rows", 4); + var target = ranges.get(1); // a middle chunk + + // Read the chunk fully, remembering every id. + Set full = new HashSet<>(); + readRange(target, null, full); + + // Re-read the same chunk but stop after the first page, then resume from + // that page's cursor — the union must equal the full chunk's ids. + var firstPage = connector.readBatch(ctx, "public", "chunked_rows", + new BatchInformation(0, 100, "chunked_rows", null, target)); + Set resumed = new HashSet<>(); + firstPage.rows().forEach(r -> resumed.add((Long) r.get("id"))); + var cursor = firstPage.nextCursor(); + assertTrue(cursor != null, "first page should have a next cursor"); + readRange(target, cursor, resumed); + + assertEquals(full, resumed, + "resume within a chunk must not duplicate or skip rows"); + } + + /** Paginate a chunk from an optional starting cursor, collecting PK ids. */ + private void readRange(ChunkRange range, String fromCursor, Set into) { + String cursor = fromCursor; + int batch = fromCursor == null ? 0 : 1; + while (true) { + var info = new BatchInformation(batch, 100, "chunked_rows", cursor, range); + var page = connector.readBatch(ctx, "public", "chunked_rows", info); + if (page.rows().isEmpty()) { + break; + } + page.rows().forEach(r -> into.add((Long) r.get("id"))); + if (page.nextCursor() == null) { + break; + } + cursor = page.nextCursor(); + batch++; + } + } +} diff --git a/syncflow-connectors/src/test/java/com/syncflow/connector/writer/JdbcBatchWriterTest.java b/syncflow-connectors/src/test/java/com/syncflow/connector/writer/JdbcBatchWriterTest.java index 421fcb7..d30cb9e 100644 --- a/syncflow-connectors/src/test/java/com/syncflow/connector/writer/JdbcBatchWriterTest.java +++ b/syncflow-connectors/src/test/java/com/syncflow/connector/writer/JdbcBatchWriterTest.java @@ -185,6 +185,30 @@ void writeBatchForDifferentTablesTracksLatestTable() { "cross-table switch must move the pending scope, not mix tables"); } + @Test + void mysqlUpsertUsesOnDuplicateKeyUpdate() { + // The MySQL dialect override must emit ON DUPLICATE KEY UPDATE, not the + // Postgres ON CONFLICT shape — a MySQL destination would otherwise fail + // at write time with a Postgres-only statement. + var w = new MySqlWriter(); + var sql = mysqlUpsertSql(w, "users", List.of("id", "email"), List.of("id")); + assertTrue(sql.contains("ON DUPLICATE KEY UPDATE"), "missing ON DUPLICATE KEY UPDATE"); + assertTrue(sql.contains(" = VALUES(email)"), "missing VALUES() assignment"); + assertTrue(!sql.contains("ON CONFLICT"), "MySQL upsert must not use ON CONFLICT"); + assertTrue(!sql.contains("AS new"), "row-alias form is not portable to MariaDB/MySQL<8.0.19"); + } + + /** Invoke the protected MySQL upsertSql for assertion. */ + private static String mysqlUpsertSql(MySqlWriter w, String table, List cols, List keys) { + try { + var m = MySqlWriter.class.getDeclaredMethod("upsertSql", String.class, List.class, List.class); + m.setAccessible(true); + return (String) m.invoke(w, table, cols, keys); + } catch (Exception e) { + throw new RuntimeException(e); + } + } + @Test void upsertBatchSetsConflictTargetFromKeyColumns() { // R3: upsertBatch must produce an ON CONFLICT statement keyed by the diff --git a/syncflow-core/src/main/java/com/syncflow/core/snapshot/BatchInformation.java b/syncflow-core/src/main/java/com/syncflow/core/snapshot/BatchInformation.java index 7970190..0badc5f 100644 --- a/syncflow-core/src/main/java/com/syncflow/core/snapshot/BatchInformation.java +++ b/syncflow-core/src/main/java/com/syncflow/core/snapshot/BatchInformation.java @@ -4,5 +4,14 @@ public record BatchInformation( int batchNumber, int batchSize, String sourceTable, - String cursor) { + String cursor, + ChunkRange chunkRange) { + + /** + * Sequential (whole-table) page request — no chunk range. Kept for the + * non-parallel path and callers that never chunk. + */ + public BatchInformation(int batchNumber, int batchSize, String sourceTable, String cursor) { + this(batchNumber, batchSize, sourceTable, cursor, null); + } } diff --git a/syncflow-core/src/main/java/com/syncflow/core/snapshot/ChunkRange.java b/syncflow-core/src/main/java/com/syncflow/core/snapshot/ChunkRange.java new file mode 100644 index 0000000..e35fb40 --- /dev/null +++ b/syncflow-core/src/main/java/com/syncflow/core/snapshot/ChunkRange.java @@ -0,0 +1,24 @@ +package com.syncflow.core.snapshot; + +/** + * A single PK-range of a table for parallel snapshot (F15). Bounds are + * inclusive on start, exclusive on end; a null bound is open-ended (whole + * table / to the end). + * + * Bounds carry the driver-native PK value type (e.g. {@link Long} for a + * bigint PK) so they bind correctly to the column — a String bound against a + * numeric column fails ("operator does not exist: bigint >= character + * varying"). The {@code index} is the chunk ordinal within the table, used to + * key per-chunk resume checkpoints. + */ +public record ChunkRange(int index, Object start, Object end) { + + /** The whole table as a single chunk — used by the non-chunked path. */ + public static ChunkRange whole() { + return new ChunkRange(0, null, null); + } + + public boolean isWhole() { + return start() == null && end() == null; + } +} diff --git a/syncflow-core/src/main/java/com/syncflow/core/snapshot/SnapshotCheckpoint.java b/syncflow-core/src/main/java/com/syncflow/core/snapshot/SnapshotCheckpoint.java index 027b1e1..4bd5b3f 100644 --- a/syncflow-core/src/main/java/com/syncflow/core/snapshot/SnapshotCheckpoint.java +++ b/syncflow-core/src/main/java/com/syncflow/core/snapshot/SnapshotCheckpoint.java @@ -3,7 +3,17 @@ public record SnapshotCheckpoint( String pipelineId, String sourceTable, + int chunkIndex, int lastBatchNumber, long rowsProcessed, String cursor) { + + /** + * Whole-table (chunk 0) checkpoint — the pre-F15 shape, kept so existing + * callers and tests constructing a checkpoint without a chunk compile. + */ + public SnapshotCheckpoint(String pipelineId, String sourceTable, + int lastBatchNumber, long rowsProcessed, String cursor) { + this(pipelineId, sourceTable, 0, lastBatchNumber, rowsProcessed, cursor); + } } diff --git a/syncflow-core/src/main/java/com/syncflow/core/spi/SnapshotCapableConnector.java b/syncflow-core/src/main/java/com/syncflow/core/spi/SnapshotCapableConnector.java index 813a4d1..153710a 100644 --- a/syncflow-core/src/main/java/com/syncflow/core/spi/SnapshotCapableConnector.java +++ b/syncflow-core/src/main/java/com/syncflow/core/spi/SnapshotCapableConnector.java @@ -1,6 +1,7 @@ package com.syncflow.core.spi; import com.syncflow.core.snapshot.BatchInformation; +import com.syncflow.core.snapshot.ChunkRange; import java.util.List; import java.util.Map; import java.util.stream.Stream; @@ -20,6 +21,27 @@ public interface SnapshotCapableConnector extends MetadataCapableConnector { Page readBatch(ConnectorContext context, String schema, String table, BatchInformation batchInfo); + /** + * Split a table into PK-range chunks for parallel snapshot (F15). + * The default returns a single whole-table chunk (sequential path); + * JDBC connectors override with real MIN/MAX-based splitting. Connectors + * without a single-column PK (Mongo, Redis) keep the sequential path. + */ + default List rangeChunks(ConnectorContext context, String schema, + String table, int chunkCount) { + return List.of(ChunkRange.whole()); + } + + /** + * Return a fresh, already-connected connector instance for one parallel + * snapshot worker. The default reuses the singleton; JDBC connectors + * override so each worker owns its own {@code java.sql.Connection} + * instead of sharing a single non-thread-safe connection across threads. + */ + default SnapshotCapableConnector snapshotClone(ConnectorContext context) { + return this; + } + /** * Stream all rows from a table. The default reads batches internally. * Override for database-native streaming (Postgres CURSOR, MySQL streaming). diff --git a/syncflow-persistence/build.gradle b/syncflow-persistence/build.gradle new file mode 100644 index 0000000..fefb0b6 --- /dev/null +++ b/syncflow-persistence/build.gradle @@ -0,0 +1,27 @@ +plugins { + alias(libs.plugins.spring.dependency.management) +} + +dependencyManagement { + imports { + mavenBom "org.springframework.boot:spring-boot-dependencies:${libs.versions.spring.boot.get()}" + } +} + +dependencies { + implementation project(":syncflow-core") + + implementation libs.spring.boot.starter.data.jpa + implementation libs.flyway.core + implementation libs.flyway.database.postgresql + implementation libs.postgresql + + compileOnly libs.lombok + annotationProcessor libs.lombok + + testImplementation libs.junit.jupiter + testImplementation libs.junit.platform.launcher + testImplementation(libs.spring.boot.starter.test) { + exclude group: "org.junit.vintage", module: "junit-vintage-engine" + } +} \ No newline at end of file diff --git a/syncflow-api/src/main/java/com/syncflow/api/agent/entity/AgentEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/agent/entity/AgentEntity.java similarity index 97% rename from syncflow-api/src/main/java/com/syncflow/api/agent/entity/AgentEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/agent/entity/AgentEntity.java index 3eb0326..578ea4e 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/agent/entity/AgentEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/agent/entity/AgentEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.agent.entity; +package com.syncflow.persistence.agent.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/agent/repository/AgentRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/agent/repository/AgentRepository.java similarity index 67% rename from syncflow-api/src/main/java/com/syncflow/api/agent/repository/AgentRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/agent/repository/AgentRepository.java index ae655c8..83144e8 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/agent/repository/AgentRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/agent/repository/AgentRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.agent.repository; +package com.syncflow.persistence.agent.repository; -import com.syncflow.api.agent.entity.AgentEntity; +import com.syncflow.persistence.agent.entity.AgentEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; diff --git a/syncflow-api/src/main/java/com/syncflow/api/cdc/entity/ActiveCaptureEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/entity/ActiveCaptureEntity.java similarity index 97% rename from syncflow-api/src/main/java/com/syncflow/api/cdc/entity/ActiveCaptureEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/entity/ActiveCaptureEntity.java index 05fb836..17b3b51 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/cdc/entity/ActiveCaptureEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/entity/ActiveCaptureEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.cdc.entity; +package com.syncflow.persistence.cdc.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/cdc/entity/CdcOffsetEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/entity/CdcOffsetEntity.java similarity index 95% rename from syncflow-api/src/main/java/com/syncflow/api/cdc/entity/CdcOffsetEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/entity/CdcOffsetEntity.java index eeb067f..51177a2 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/cdc/entity/CdcOffsetEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/entity/CdcOffsetEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.cdc.entity; +package com.syncflow.persistence.cdc.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/cdc/repository/ActiveCaptureRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/repository/ActiveCaptureRepository.java similarity index 76% rename from syncflow-api/src/main/java/com/syncflow/api/cdc/repository/ActiveCaptureRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/repository/ActiveCaptureRepository.java index 4999adb..56c73c1 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/cdc/repository/ActiveCaptureRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/repository/ActiveCaptureRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.cdc.repository; +package com.syncflow.persistence.cdc.repository; -import com.syncflow.api.cdc.entity.ActiveCaptureEntity; +import com.syncflow.persistence.cdc.entity.ActiveCaptureEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.Optional; diff --git a/syncflow-api/src/main/java/com/syncflow/api/cdc/repository/CdcOffsetRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/repository/CdcOffsetRepository.java similarity index 91% rename from syncflow-api/src/main/java/com/syncflow/api/cdc/repository/CdcOffsetRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/repository/CdcOffsetRepository.java index a4cff2e..a380e28 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/cdc/repository/CdcOffsetRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/cdc/repository/CdcOffsetRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.cdc.repository; +package com.syncflow.persistence.cdc.repository; -import com.syncflow.api.cdc.entity.CdcOffsetEntity; +import com.syncflow.persistence.cdc.entity.CdcOffsetEntity; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; diff --git a/syncflow-persistence/src/main/java/com/syncflow/persistence/config/PersistenceConfig.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/config/PersistenceConfig.java new file mode 100644 index 0000000..c010a8d --- /dev/null +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/config/PersistenceConfig.java @@ -0,0 +1,16 @@ +package com.syncflow.persistence.config; + +import org.springframework.boot.autoconfigure.domain.EntityScan; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.jpa.repository.config.EnableJpaRepositories; + +/** + * Scans the persistence module for JPA entities and Spring Data repositories. + * The root application uses {@code scanBasePackages = "com.syncflow"}, so this + * configuration is picked up automatically. + */ +@Configuration +@EntityScan(basePackages = "com.syncflow.persistence") +@EnableJpaRepositories(basePackages = "com.syncflow.persistence") +public class PersistenceConfig { +} diff --git a/syncflow-api/src/main/java/com/syncflow/api/connection/entity/ConnectionEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/connection/entity/ConnectionEntity.java similarity index 97% rename from syncflow-api/src/main/java/com/syncflow/api/connection/entity/ConnectionEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/connection/entity/ConnectionEntity.java index aafcca5..9a10ca2 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/connection/entity/ConnectionEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/connection/entity/ConnectionEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.connection.entity; +package com.syncflow.persistence.connection.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/connection/repository/ConnectionRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/connection/repository/ConnectionRepository.java similarity index 73% rename from syncflow-api/src/main/java/com/syncflow/api/connection/repository/ConnectionRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/connection/repository/ConnectionRepository.java index f72d72f..c28692e 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/connection/repository/ConnectionRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/connection/repository/ConnectionRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.connection.repository; +package com.syncflow.persistence.connection.repository; -import com.syncflow.api.connection.entity.ConnectionEntity; +import com.syncflow.persistence.connection.entity.ConnectionEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; diff --git a/syncflow-api/src/main/java/com/syncflow/api/ops/alert/entity/AlertEventEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/ops/alert/entity/AlertEventEntity.java similarity index 96% rename from syncflow-api/src/main/java/com/syncflow/api/ops/alert/entity/AlertEventEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/ops/alert/entity/AlertEventEntity.java index b5e3474..932ca4e 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/ops/alert/entity/AlertEventEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/ops/alert/entity/AlertEventEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.ops.alert.entity; +package com.syncflow.persistence.ops.alert.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/ops/alert/repository/AlertEventRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/ops/alert/repository/AlertEventRepository.java similarity index 79% rename from syncflow-api/src/main/java/com/syncflow/api/ops/alert/repository/AlertEventRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/ops/alert/repository/AlertEventRepository.java index 8a13997..6f76008 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/ops/alert/repository/AlertEventRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/ops/alert/repository/AlertEventRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.ops.alert.repository; +package com.syncflow.persistence.ops.alert.repository; -import com.syncflow.api.ops.alert.entity.AlertEventEntity; +import com.syncflow.persistence.ops.alert.entity.AlertEventEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/entity/PipelineDesignEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/entity/PipelineDesignEntity.java similarity index 97% rename from syncflow-api/src/main/java/com/syncflow/api/pipeline/entity/PipelineDesignEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/entity/PipelineDesignEntity.java index 8557b25..64a0ec4 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/entity/PipelineDesignEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/entity/PipelineDesignEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.pipeline.entity; +package com.syncflow.persistence.pipeline.entity; import jakarta.persistence.CascadeType; import jakarta.persistence.Column; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/entity/PipelineDesignVersionEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/entity/PipelineDesignVersionEntity.java similarity index 96% rename from syncflow-api/src/main/java/com/syncflow/api/pipeline/entity/PipelineDesignVersionEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/entity/PipelineDesignVersionEntity.java index ac808a1..06b617c 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/entity/PipelineDesignVersionEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/entity/PipelineDesignVersionEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.pipeline.entity; +package com.syncflow.persistence.pipeline.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/entity/PipelineEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/entity/PipelineEntity.java similarity index 96% rename from syncflow-api/src/main/java/com/syncflow/api/pipeline/entity/PipelineEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/entity/PipelineEntity.java index eb8ccb4..5b36f0c 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/entity/PipelineEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/entity/PipelineEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.pipeline.entity; +package com.syncflow.persistence.pipeline.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineDesignJpaRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/repository/PipelineDesignJpaRepository.java similarity index 74% rename from syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineDesignJpaRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/repository/PipelineDesignJpaRepository.java index b9db869..ba5b1a0 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineDesignJpaRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/repository/PipelineDesignJpaRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.pipeline.repository; +package com.syncflow.persistence.pipeline.repository; -import com.syncflow.api.pipeline.entity.PipelineDesignEntity; +import com.syncflow.persistence.pipeline.entity.PipelineDesignEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineDesignVersionJpaRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/repository/PipelineDesignVersionJpaRepository.java similarity index 78% rename from syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineDesignVersionJpaRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/repository/PipelineDesignVersionJpaRepository.java index 4e2e9c8..2b9cb79 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineDesignVersionJpaRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/repository/PipelineDesignVersionJpaRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.pipeline.repository; +package com.syncflow.persistence.pipeline.repository; -import com.syncflow.api.pipeline.entity.PipelineDesignVersionEntity; +import com.syncflow.persistence.pipeline.entity.PipelineDesignVersionEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; diff --git a/syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineJpaRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/repository/PipelineJpaRepository.java similarity index 66% rename from syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineJpaRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/repository/PipelineJpaRepository.java index c471950..ff709a4 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/pipeline/repository/PipelineJpaRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/pipeline/repository/PipelineJpaRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.pipeline.repository; +package com.syncflow.persistence.pipeline.repository; -import com.syncflow.api.pipeline.entity.PipelineEntity; +import com.syncflow.persistence.pipeline.entity.PipelineEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/apikey/entity/ApiKeyEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/apikey/entity/ApiKeyEntity.java similarity index 94% rename from syncflow-api/src/main/java/com/syncflow/api/security/apikey/entity/ApiKeyEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/security/apikey/entity/ApiKeyEntity.java index 742e386..9b40835 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/apikey/entity/ApiKeyEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/apikey/entity/ApiKeyEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.security.apikey.entity; +package com.syncflow.persistence.security.apikey.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/apikey/repository/ApiKeyRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/apikey/repository/ApiKeyRepository.java similarity index 66% rename from syncflow-api/src/main/java/com/syncflow/api/security/apikey/repository/ApiKeyRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/security/apikey/repository/ApiKeyRepository.java index 4b3a157..d0e92fe 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/apikey/repository/ApiKeyRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/apikey/repository/ApiKeyRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.security.apikey.repository; +package com.syncflow.persistence.security.apikey.repository; -import com.syncflow.api.security.apikey.entity.ApiKeyEntity; +import com.syncflow.persistence.security.apikey.entity.ApiKeyEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.Optional; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/audit/entity/AuditRecordEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/audit/entity/AuditRecordEntity.java similarity index 95% rename from syncflow-api/src/main/java/com/syncflow/api/security/audit/entity/AuditRecordEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/security/audit/entity/AuditRecordEntity.java index 734293e..2070b99 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/audit/entity/AuditRecordEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/audit/entity/AuditRecordEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.security.audit.entity; +package com.syncflow.persistence.security.audit.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/audit/repository/AuditRecordRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/audit/repository/AuditRecordRepository.java similarity index 72% rename from syncflow-api/src/main/java/com/syncflow/api/security/audit/repository/AuditRecordRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/security/audit/repository/AuditRecordRepository.java index f70c61f..4dca2d9 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/audit/repository/AuditRecordRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/audit/repository/AuditRecordRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.security.audit.repository; +package com.syncflow.persistence.security.audit.repository; -import com.syncflow.api.security.audit.entity.AuditRecordEntity; +import com.syncflow.persistence.security.audit.entity.AuditRecordEntity; import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/quota/entity/QuotaEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/quota/entity/QuotaEntity.java similarity index 92% rename from syncflow-api/src/main/java/com/syncflow/api/security/quota/entity/QuotaEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/security/quota/entity/QuotaEntity.java index 67ed42d..875747c 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/quota/entity/QuotaEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/quota/entity/QuotaEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.security.quota.entity; +package com.syncflow.persistence.security.quota.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/security/quota/repository/QuotaRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/quota/repository/QuotaRepository.java similarity index 53% rename from syncflow-api/src/main/java/com/syncflow/api/security/quota/repository/QuotaRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/security/quota/repository/QuotaRepository.java index 8e17177..9ade38c 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/security/quota/repository/QuotaRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/security/quota/repository/QuotaRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.security.quota.repository; +package com.syncflow.persistence.security.quota.repository; -import com.syncflow.api.security.quota.entity.QuotaEntity; +import com.syncflow.persistence.security.quota.entity.QuotaEntity; import org.springframework.data.jpa.repository.JpaRepository; public interface QuotaRepository extends JpaRepository { diff --git a/syncflow-api/src/main/java/com/syncflow/api/snapshot/entity/SnapshotCheckpointEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/entity/SnapshotCheckpointEntity.java similarity index 90% rename from syncflow-api/src/main/java/com/syncflow/api/snapshot/entity/SnapshotCheckpointEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/entity/SnapshotCheckpointEntity.java index 42f1d2b..156d649 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/snapshot/entity/SnapshotCheckpointEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/entity/SnapshotCheckpointEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.snapshot.entity; +package com.syncflow.persistence.snapshot.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; @@ -31,6 +31,9 @@ public class SnapshotCheckpointEntity { @Column(name = "source_table", nullable = false, length = 255) private String sourceTable; + @Column(name = "chunk_index", nullable = false) + private int chunkIndex; + @Column(name = "last_batch_number", nullable = false) private int lastBatchNumber; diff --git a/syncflow-api/src/main/java/com/syncflow/api/snapshot/entity/SnapshotJobEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/entity/SnapshotJobEntity.java similarity index 95% rename from syncflow-api/src/main/java/com/syncflow/api/snapshot/entity/SnapshotJobEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/entity/SnapshotJobEntity.java index 26940e3..ef67f05 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/snapshot/entity/SnapshotJobEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/entity/SnapshotJobEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.snapshot.entity; +package com.syncflow.persistence.snapshot.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/snapshot/repository/SnapshotCheckpointRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/repository/SnapshotCheckpointRepository.java similarity index 81% rename from syncflow-api/src/main/java/com/syncflow/api/snapshot/repository/SnapshotCheckpointRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/repository/SnapshotCheckpointRepository.java index 857ba61..1a838d4 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/snapshot/repository/SnapshotCheckpointRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/repository/SnapshotCheckpointRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.snapshot.repository; +package com.syncflow.persistence.snapshot.repository; -import com.syncflow.api.snapshot.entity.SnapshotCheckpointEntity; +import com.syncflow.persistence.snapshot.entity.SnapshotCheckpointEntity; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; @@ -10,8 +10,8 @@ public interface SnapshotCheckpointRepository extends JpaRepository { - Optional findByTenantIdAndPipelineIdAndSourceTable( - String tenantId, String pipelineId, String sourceTable); + Optional findByTenantIdAndPipelineIdAndSourceTableAndChunkIndex( + String tenantId, String pipelineId, String sourceTable, int chunkIndex); @Modifying @Query(value = """ diff --git a/syncflow-api/src/main/java/com/syncflow/api/snapshot/repository/SnapshotJobRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/repository/SnapshotJobRepository.java similarity index 75% rename from syncflow-api/src/main/java/com/syncflow/api/snapshot/repository/SnapshotJobRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/repository/SnapshotJobRepository.java index f5c8957..3c03e9f 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/snapshot/repository/SnapshotJobRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/snapshot/repository/SnapshotJobRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.snapshot.repository; +package com.syncflow.persistence.snapshot.repository; -import com.syncflow.api.snapshot.entity.SnapshotJobEntity; +import com.syncflow.persistence.snapshot.entity.SnapshotJobEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/entity/DeadLetterEventEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/entity/DeadLetterEventEntity.java similarity index 97% rename from syncflow-api/src/main/java/com/syncflow/api/sync/entity/DeadLetterEventEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/sync/entity/DeadLetterEventEntity.java index 883f900..ffba610 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/entity/DeadLetterEventEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/entity/DeadLetterEventEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.sync.entity; +package com.syncflow.persistence.sync.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/entity/ProcessedEventEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/entity/ProcessedEventEntity.java similarity index 95% rename from syncflow-api/src/main/java/com/syncflow/api/sync/entity/ProcessedEventEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/sync/entity/ProcessedEventEntity.java index 9dd4bfc..8271bc5 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/entity/ProcessedEventEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/entity/ProcessedEventEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.sync.entity; +package com.syncflow.persistence.sync.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/entity/SyncJobEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/entity/SyncJobEntity.java similarity index 96% rename from syncflow-api/src/main/java/com/syncflow/api/sync/entity/SyncJobEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/sync/entity/SyncJobEntity.java index 1861fbf..ba5ec19 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/entity/SyncJobEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/entity/SyncJobEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.sync.entity; +package com.syncflow.persistence.sync.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/repository/DeadLetterEventRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/repository/DeadLetterEventRepository.java similarity index 90% rename from syncflow-api/src/main/java/com/syncflow/api/sync/repository/DeadLetterEventRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/sync/repository/DeadLetterEventRepository.java index 10dbf08..b19ccc1 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/repository/DeadLetterEventRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/repository/DeadLetterEventRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.sync.repository; +package com.syncflow.persistence.sync.repository; -import com.syncflow.api.sync.entity.DeadLetterEventEntity; +import com.syncflow.persistence.sync.entity.DeadLetterEventEntity; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/repository/ProcessedEventRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/repository/ProcessedEventRepository.java similarity index 92% rename from syncflow-api/src/main/java/com/syncflow/api/sync/repository/ProcessedEventRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/sync/repository/ProcessedEventRepository.java index ac2e264..4658c72 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/repository/ProcessedEventRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/repository/ProcessedEventRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.sync.repository; +package com.syncflow.persistence.sync.repository; -import com.syncflow.api.sync.entity.ProcessedEventEntity; +import com.syncflow.persistence.sync.entity.ProcessedEventEntity; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Modifying; import org.springframework.data.jpa.repository.Query; diff --git a/syncflow-api/src/main/java/com/syncflow/api/sync/repository/SyncJobRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/repository/SyncJobRepository.java similarity index 77% rename from syncflow-api/src/main/java/com/syncflow/api/sync/repository/SyncJobRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/sync/repository/SyncJobRepository.java index 987f349..6eb22cb 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/sync/repository/SyncJobRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/sync/repository/SyncJobRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.sync.repository; +package com.syncflow.persistence.sync.repository; -import com.syncflow.api.sync.entity.SyncJobEntity; +import com.syncflow.persistence.sync.entity.SyncJobEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; diff --git a/syncflow-api/src/main/java/com/syncflow/api/user/entity/UserEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/user/entity/UserEntity.java similarity index 96% rename from syncflow-api/src/main/java/com/syncflow/api/user/entity/UserEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/user/entity/UserEntity.java index 4c37cc6..bbedbfe 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/user/entity/UserEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/user/entity/UserEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.user.entity; +package com.syncflow.persistence.user.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/user/repository/UserRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/user/repository/UserRepository.java similarity index 72% rename from syncflow-api/src/main/java/com/syncflow/api/user/repository/UserRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/user/repository/UserRepository.java index dcd5e2d..48fc90f 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/user/repository/UserRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/user/repository/UserRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.user.repository; +package com.syncflow.persistence.user.repository; -import com.syncflow.api.user.entity.UserEntity; +import com.syncflow.persistence.user.entity.UserEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.Optional; diff --git a/syncflow-api/src/main/java/com/syncflow/api/workflow/entity/WorkflowInstanceEntity.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/workflow/entity/WorkflowInstanceEntity.java similarity index 96% rename from syncflow-api/src/main/java/com/syncflow/api/workflow/entity/WorkflowInstanceEntity.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/workflow/entity/WorkflowInstanceEntity.java index 00a0107..390cf56 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/workflow/entity/WorkflowInstanceEntity.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/workflow/entity/WorkflowInstanceEntity.java @@ -1,4 +1,4 @@ -package com.syncflow.api.workflow.entity; +package com.syncflow.persistence.workflow.entity; import jakarta.persistence.Column; import jakarta.persistence.Entity; diff --git a/syncflow-api/src/main/java/com/syncflow/api/workflow/repository/WorkflowInstanceRepository.java b/syncflow-persistence/src/main/java/com/syncflow/persistence/workflow/repository/WorkflowInstanceRepository.java similarity index 68% rename from syncflow-api/src/main/java/com/syncflow/api/workflow/repository/WorkflowInstanceRepository.java rename to syncflow-persistence/src/main/java/com/syncflow/persistence/workflow/repository/WorkflowInstanceRepository.java index 5c9e4ff..9128828 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/workflow/repository/WorkflowInstanceRepository.java +++ b/syncflow-persistence/src/main/java/com/syncflow/persistence/workflow/repository/WorkflowInstanceRepository.java @@ -1,6 +1,6 @@ -package com.syncflow.api.workflow.repository; +package com.syncflow.persistence.workflow.repository; -import com.syncflow.api.workflow.entity.WorkflowInstanceEntity; +import com.syncflow.persistence.workflow.entity.WorkflowInstanceEntity; import org.springframework.data.jpa.repository.JpaRepository; import java.util.List; diff --git a/syncflow-api/src/main/resources/db/migration/V10__must_change_password.sql b/syncflow-persistence/src/main/resources/db/migration/V10__must_change_password.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V10__must_change_password.sql rename to syncflow-persistence/src/main/resources/db/migration/V10__must_change_password.sql diff --git a/syncflow-api/src/main/resources/db/migration/V11__tenant_id.sql b/syncflow-persistence/src/main/resources/db/migration/V11__tenant_id.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V11__tenant_id.sql rename to syncflow-persistence/src/main/resources/db/migration/V11__tenant_id.sql diff --git a/syncflow-api/src/main/resources/db/migration/V12__runtime_state_persistence.sql b/syncflow-persistence/src/main/resources/db/migration/V12__runtime_state_persistence.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V12__runtime_state_persistence.sql rename to syncflow-persistence/src/main/resources/db/migration/V12__runtime_state_persistence.sql diff --git a/syncflow-api/src/main/resources/db/migration/V13__debezium_offsets.sql b/syncflow-persistence/src/main/resources/db/migration/V13__debezium_offsets.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V13__debezium_offsets.sql rename to syncflow-persistence/src/main/resources/db/migration/V13__debezium_offsets.sql diff --git a/syncflow-api/src/main/resources/db/migration/V14__active_captures.sql b/syncflow-persistence/src/main/resources/db/migration/V14__active_captures.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V14__active_captures.sql rename to syncflow-persistence/src/main/resources/db/migration/V14__active_captures.sql diff --git a/syncflow-persistence/src/main/resources/db/migration/V15__snapshot_chunk_checkpoints.sql b/syncflow-persistence/src/main/resources/db/migration/V15__snapshot_chunk_checkpoints.sql new file mode 100644 index 0000000..64675c1 --- /dev/null +++ b/syncflow-persistence/src/main/resources/db/migration/V15__snapshot_chunk_checkpoints.sql @@ -0,0 +1,13 @@ +-- Per-chunk resume checkpoints for parallel PK-range snapshot (F15). +-- Adds chunk_index so each chunk of a table resumes from its own cursor. +ALTER TABLE snapshot_checkpoints + ADD COLUMN IF NOT EXISTS chunk_index INTEGER NOT NULL DEFAULT 0; + +-- Widen the unique key to include the chunk so multiple chunks of one table +-- can hold distinct checkpoints concurrently. +ALTER TABLE snapshot_checkpoints + DROP CONSTRAINT IF EXISTS uq_checkpoint_pipeline_table; + +ALTER TABLE snapshot_checkpoints + ADD CONSTRAINT uq_checkpoint_pipeline_table_chunk + UNIQUE (tenant_id, pipeline_id, source_table, chunk_index); \ No newline at end of file diff --git a/syncflow-api/src/main/resources/db/migration/V1__init.sql b/syncflow-persistence/src/main/resources/db/migration/V1__init.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V1__init.sql rename to syncflow-persistence/src/main/resources/db/migration/V1__init.sql diff --git a/syncflow-api/src/main/resources/db/migration/V2__connections.sql b/syncflow-persistence/src/main/resources/db/migration/V2__connections.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V2__connections.sql rename to syncflow-persistence/src/main/resources/db/migration/V2__connections.sql diff --git a/syncflow-api/src/main/resources/db/migration/V3__governance.sql b/syncflow-persistence/src/main/resources/db/migration/V3__governance.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V3__governance.sql rename to syncflow-persistence/src/main/resources/db/migration/V3__governance.sql diff --git a/syncflow-api/src/main/resources/db/migration/V4__pipeline_designs.sql b/syncflow-persistence/src/main/resources/db/migration/V4__pipeline_designs.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V4__pipeline_designs.sql rename to syncflow-persistence/src/main/resources/db/migration/V4__pipeline_designs.sql diff --git a/syncflow-api/src/main/resources/db/migration/V5__cdc_offsets.sql b/syncflow-persistence/src/main/resources/db/migration/V5__cdc_offsets.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V5__cdc_offsets.sql rename to syncflow-persistence/src/main/resources/db/migration/V5__cdc_offsets.sql diff --git a/syncflow-api/src/main/resources/db/migration/V6__kafka_sync_persistence.sql b/syncflow-persistence/src/main/resources/db/migration/V6__kafka_sync_persistence.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V6__kafka_sync_persistence.sql rename to syncflow-persistence/src/main/resources/db/migration/V6__kafka_sync_persistence.sql diff --git a/syncflow-api/src/main/resources/db/migration/V7__dlq_nullable_event.sql b/syncflow-persistence/src/main/resources/db/migration/V7__dlq_nullable_event.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V7__dlq_nullable_event.sql rename to syncflow-persistence/src/main/resources/db/migration/V7__dlq_nullable_event.sql diff --git a/syncflow-api/src/main/resources/db/migration/V8__dlq_replay_count.sql b/syncflow-persistence/src/main/resources/db/migration/V8__dlq_replay_count.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V8__dlq_replay_count.sql rename to syncflow-persistence/src/main/resources/db/migration/V8__dlq_replay_count.sql diff --git a/syncflow-api/src/main/resources/db/migration/V9__users.sql b/syncflow-persistence/src/main/resources/db/migration/V9__users.sql similarity index 100% rename from syncflow-api/src/main/resources/db/migration/V9__users.sql rename to syncflow-persistence/src/main/resources/db/migration/V9__users.sql diff --git a/syncflow-api/src/test/java/com/syncflow/api/user/UserEntityMappingTest.java b/syncflow-persistence/src/test/java/com/syncflow/persistence/user/UserEntityMappingTest.java similarity index 88% rename from syncflow-api/src/test/java/com/syncflow/api/user/UserEntityMappingTest.java rename to syncflow-persistence/src/test/java/com/syncflow/persistence/user/UserEntityMappingTest.java index ae93336..07e6292 100644 --- a/syncflow-api/src/test/java/com/syncflow/api/user/UserEntityMappingTest.java +++ b/syncflow-persistence/src/test/java/com/syncflow/persistence/user/UserEntityMappingTest.java @@ -1,6 +1,6 @@ -package com.syncflow.api.user; +package com.syncflow.persistence.user; -import com.syncflow.api.user.entity.UserEntity; +import com.syncflow.persistence.user.entity.UserEntity; import jakarta.persistence.Table; import org.junit.jupiter.api.Test;