diff --git a/ARCHITECTURE_ANALYSIS.md b/ARCHITECTURE_ANALYSIS.md index dd137f6..ea26f6b 100644 --- a/ARCHITECTURE_ANALYSIS.md +++ b/ARCHITECTURE_ANALYSIS.md @@ -1,7 +1,7 @@ # SyncFlow Architecture Analysis **Generated:** 2026-08-12 -**Last updated:** 2026-09-05 (M1/M2/M6/M7/M8 maintainability, D6 metrics dedup — see item statuses below) +**Last updated:** 2026-09-06 (F16 structured concurrency — see item statuses below) **Scope:** End-to-end codebase review (core, api, connectors, common, agent) --- @@ -237,7 +237,7 @@ | # | Action | |---|--------| -| **F16** | Migrate to reactive (Project Reactor) or structured concurrency for better resource control | +| **F16** | Migrate to reactive (Project Reactor) or structured concurrency for better resource control | ✅ **Done** — `StructuredTaskScope` for snapshot fan-out, `ReentrantLock` replaces all `synchronized`, `spring.threads.virtual.enabled: true`. Reactive path (WebFlux/Reactor) deferred: virtual threads + structured concurrency deliver equivalent resource control without the servlet→reactive ecosystem migration. Revisit if backpressure becomes a requirement. | | **F17** | Add multi-region / geo-replication support | | **F18** | Implement connector plugin system (dynamic loading) | | **F19** | Add SQL-based transformation engine (push down to DB) | @@ -505,7 +505,7 @@ public class RuntimeProperties { 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 +6. **Ongoing**: F15 ✅ done (parallel PK-range chunking); F16 ✅ done (structured concurrency: `StructuredTaskScope`, `ReentrantLock`, virtual threads); F17+ (geo-replication, plugin system, SQL-transform pushdown) still open --- diff --git a/docs/adr/ADR-004-virtual-threads.md b/docs/adr/ADR-004-virtual-threads.md index 5411543..f618db3 100644 --- a/docs/adr/ADR-004-virtual-threads.md +++ b/docs/adr/ADR-004-virtual-threads.md @@ -6,24 +6,28 @@ SyncFlow executes concurrent operations: multiple snapshot batches, parallel CDC connectors, simultaneous sync workers. Traditional thread-per-request with a fixed thread pool requires careful sizing and often leads to thread starvation under load. ## Decision -Use JDK 25 Virtual Threads (`Thread.startVirtualThread()`) for all concurrent execution paths. +Use JDK 25 Virtual Threads (`Thread.startVirtualThread()`) for all concurrent execution paths. Structured concurrency (`StructuredTaskScope`) for fan-out workloads. `ReentrantLock` instead of `synchronized` to avoid carrier-thread pinning. ## Usage in SyncFlow -- **Snapshot engine**: Each snapshot runs on its own virtual thread (`SnapshotExecutor.java:134`). -- **CDC engine**: Debezium engine runs on a virtual thread (`DebeziumCdcConnector.java:175`). -- **CDC (MongoDB)**: Change Streams cursor runs on a virtual thread (`MongoDbCdcConnector.java:105`). -- **Sync orchestrator**: Each pipeline's event consumer runs on a virtual thread (`SyncOrchestrator.java:105`). +- **Snapshot engine**: Parallel chunk fan-out uses `StructuredTaskScope.open(Joiner.allSuccessfulOrThrow())` — one forked task per work item, structured lifecycle, automatic shutdown on failure (`SnapshotExecutor.java`). +- **CDC engine**: Debezium engine runs on a virtual thread (`DebeziumCdcConnector.java`). +- **CDC (MongoDB)**: Change Streams cursor runs on a virtual thread (`MongoDbCdcConnector.java`). +- **Sync orchestrator**: Each pipeline's event consumer runs on a virtual thread (`SyncOrchestrator.java`). +- **Tomcat**: `spring.threads.virtual.enabled: true` handles HTTP requests on virtual threads. ## Rationale - **No thread pool sizing**: Virtual threads are cheap (~1KB stack) — start as many as needed. - **Blocking I/O is fine**: JDBC calls, HTTP requests, and Kafka client calls all release the underlying carrier thread while waiting. +- **Structured concurrency**: `StructuredTaskScope` replaces manual thread-join + work-queue patterns with structured lifecycle management and automatic failure propagation. - **Simple code**: No `CompletableFuture` chaining — synchronous code on virtual threads is easier to read and debug. -- **Future-proof**: JDK 25 virtual threads are mature (incubated since JDK 19, final in JDK 21). ## Consequences - All blocking operations must go through virtual-thread-aware APIs (JDBC, HTTP client, etc.). -- `synchronized` blocks should be replaced with `ReentrantLock` where they might pin carrier threads. -- Thread pool executors are not needed — `Executors.newVirtualThreadPerTaskExecutor()` replaces them. +- `synchronized` blocks replaced with `ReentrantLock` to avoid pinning carrier threads (`SnapshotExecutor`, `SnapshotWorker`, `RegionalDataSourceFactory`). +- `StructuredTaskScope` is final in JDK 25 (was incubating in `jdk.incubator.concurrent` through JDK 24). ## Links - `spring.threads.virtual.enabled: true` in `application.yml` +- `SnapshotExecutor.java` — `StructuredTaskScope` fan-out +- `SnapshotWorker.java` — `ReentrantLock` for writer/progress serialization +- `RegionalDataSourceFactory.java` — `ReentrantLock` for failover diff --git a/syncflow-api/src/main/java/com/syncflow/api/region/RegionalDataSourceFactory.java b/syncflow-api/src/main/java/com/syncflow/api/region/RegionalDataSourceFactory.java index 4d42756..888a443 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/region/RegionalDataSourceFactory.java +++ b/syncflow-api/src/main/java/com/syncflow/api/region/RegionalDataSourceFactory.java @@ -4,6 +4,7 @@ import com.zaxxer.hikari.HikariDataSource; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReentrantLock; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; @@ -35,6 +36,7 @@ public class RegionalDataSourceFactory { private final RegionalProperties regionalProperties; private final Map datasources = new ConcurrentHashMap<>(); private volatile String currentPrimaryRegion; + private final ReentrantLock failoverLock = new ReentrantLock(); public RegionalDataSourceFactory(RegionalProperties regionalProperties) { this.regionalProperties = regionalProperties; @@ -112,16 +114,21 @@ public HikariDataSource getReadDataSource(String localRegion) { * @param newPrimaryRegion * region to promote (e.g., "eu-west-1") */ - public synchronized void promoteReplicaToPrimary(String newPrimaryRegion) { - if (!datasources.containsKey(newPrimaryRegion)) { - throw new IllegalArgumentException("Region not configured: " + newPrimaryRegion); - } + public void promoteReplicaToPrimary(String newPrimaryRegion) { + failoverLock.lock(); + try { + if (!datasources.containsKey(newPrimaryRegion)) { + throw new IllegalArgumentException("Region not configured: " + newPrimaryRegion); + } - logger.info( - "Promoting region {} to primary (was: {})", - newPrimaryRegion, - currentPrimaryRegion); - currentPrimaryRegion = newPrimaryRegion; + logger.info( + "Promoting region {} to primary (was: {})", + newPrimaryRegion, + currentPrimaryRegion); + currentPrimaryRegion = newPrimaryRegion; + } finally { + failoverLock.unlock(); + } } /** @@ -134,26 +141,31 @@ public synchronized void promoteReplicaToPrimary(String newPrimaryRegion) { * @param connectionString * new connection string */ - public synchronized void updateRegionalDataSource(String region, String connectionString) { - // Close old datasource - var old = datasources.get(region); - if (old != null && !old.isClosed()) { + public void updateRegionalDataSource(String region, String connectionString) { + failoverLock.lock(); + try { + // Close old datasource + var old = datasources.get(region); + if (old != null && !old.isClosed()) { + try { + old.close(); + logger.info("Closed datasource for region: {}", region); + } catch (Exception e) { + logger.warn("Error closing datasource for region: {}", region, e); + } + } + + // Create new datasource try { - old.close(); - logger.info("Closed datasource for region: {}", region); + var newDs = createDataSource(region, connectionString); + datasources.put(region, newDs); + logger.info("Updated datasource for region: {} with new connection", region); } catch (Exception e) { - logger.warn("Error closing datasource for region: {}", region, e); + logger.error("Failed to update datasource for region: {}", region, e); + throw new RuntimeException("Cannot update datasource for " + region, e); } - } - - // Create new datasource - try { - var newDs = createDataSource(region, connectionString); - datasources.put(region, newDs); - logger.info("Updated datasource for region: {} with new connection", region); - } catch (Exception e) { - logger.error("Failed to update datasource for region: {}", region, e); - throw new RuntimeException("Cannot update datasource for " + region, e); + } finally { + failoverLock.unlock(); } } 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 cf773c5..e83eaa8 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 @@ -42,7 +42,9 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.atomic.AtomicReference; +import java.util.concurrent.locks.ReentrantLock; +import java.util.concurrent.StructuredTaskScope; +import java.util.concurrent.StructuredTaskScope.Joiner; @Component public class SnapshotExecutor { @@ -165,10 +167,13 @@ public SnapshotJob cancel(String snapshotId, TenantContext tenantContext) { // 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) { + progressLock.lock(); + try { if (flag != null) flag.set(true); persist(cancelled, tenantContext); + } finally { + progressLock.unlock(); } // 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 @@ -205,9 +210,8 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex 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<>(); + // across parallel chunk workers serialize on this lock. + var writerLock = new ReentrantLock(); DestinationWriter writer = null; try { @@ -247,40 +251,27 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex } } - 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); + // StructuredTaskScope with Joiner.allSuccessfulOrThrow: fork one task + // per work item, shut down on first failure, propagate that failure. + // Each task gets its own connector clone (JDBC connections are not + // thread-safe). 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(); + try (var scope = StructuredTaskScope.open(Joiner.allSuccessfulOrThrow())) { + for (var wi : workItems) { + var clone = connector.snapshotClone(sourceCtx); + workerClones.add(clone); + scope.fork(() -> { + snapshotRange(job, pipeline, tenantContext, clone, + sharedWriter, writerLock, wi.table(), wi.range(), sourceCtx, + rowsProcessed, batchesDone, finalTotalRows, finalTotalBatches); + return true; + }); } + scope.join(); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); + } catch (Throwable t) { + throw new RuntimeException("Snapshot range failed", t); } finally { for (var clone : workerClones) { try { @@ -290,16 +281,13 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex } } - if (failure.get() != null) { - throw new RuntimeException("Snapshot range failed", failure.get()); - } - var elapsed = sample.stop(timer); // 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) { + progressLock.lock(); + try { if (isCancelled(job)) { // Do not commit partial writes on cancel — a later resume would // duplicate the already-written rows. @@ -318,6 +306,8 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex // Release worker state once the worker has decided its terminal // state (COMPLETED or CANCELLED stood). remove(job.getId().value()); + } finally { + progressLock.unlock(); } } catch (Exception e) { sample.stop(timer); @@ -332,13 +322,16 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex // 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) { + progressLock.lock(); + try { if (!isCancelled(job)) { var failed = job.withFailed(List.of(error)); persist(failed, tenantContext); emit(job.getId().value(), failed, tenantContext); } remove(job.getId().value()); + } finally { + progressLock.unlock(); } MetricsHelper.increment(meterRegistry, "syncflow.snapshot.errors", "pipeline", pipeline.id().value()); @@ -350,7 +343,7 @@ private void executeInner(SnapshotJob job, PipelineDesign pipeline, TenantContex * {@link SnapshotWorker} for the actual read/transform/write loop. */ private void snapshotRange(SnapshotJob job, PipelineDesign pipeline, TenantContext tenantContext, - SnapshotCapableConnector connector, DestinationWriter writer, Object writerLock, + SnapshotCapableConnector connector, DestinationWriter writer, ReentrantLock writerLock, TableMapping tm, ChunkRange range, ConnectorContext sourceCtx, AtomicLong rowsProcessed, AtomicLong batchesDone, long totalRows, long totalBatches) { SnapshotWorker.snapshotRange(job, pipeline, tenantContext, @@ -365,11 +358,12 @@ private void snapshotRange(SnapshotJob job, PipelineDesign pipeline, TenantConte /** * Aggregate live-progress publication across parallel chunk workers is - * serialized on this monitor; {@link #persist} reads the whole job payload + * serialized on this lock; {@link #persist} reads the whole job payload * and writes it back, so two workers persisting concurrently would clobber - * each other's progress. + * each other's progress. Uses ReentrantLock instead of synchronized to + * avoid pinning virtual threads on carrier threads. */ - private final Object progressLock = new Object(); + private final ReentrantLock progressLock = new ReentrantLock(); /** A (table mapping, chunk range) work item for the parallel snapshot. */ private record WorkItem(TableMapping table, ChunkRange range) { diff --git a/syncflow-api/src/main/java/com/syncflow/api/snapshot/SnapshotWorker.java b/syncflow-api/src/main/java/com/syncflow/api/snapshot/SnapshotWorker.java index ba5254e..528edfe 100644 --- a/syncflow-api/src/main/java/com/syncflow/api/snapshot/SnapshotWorker.java +++ b/syncflow-api/src/main/java/com/syncflow/api/snapshot/SnapshotWorker.java @@ -21,6 +21,7 @@ import java.math.BigDecimal; import java.util.Objects; import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.locks.ReentrantLock; import java.util.function.BooleanSupplier; /** @@ -44,11 +45,11 @@ interface TriConsumer { * batch-write to the destination. Writes serialize on {@code writerLock}. */ static void snapshotRange(SnapshotJob job, PipelineDesign pipeline, TenantContext tenantContext, - SnapshotCapableConnector connector, DestinationWriter writer, Object writerLock, + SnapshotCapableConnector connector, DestinationWriter writer, ReentrantLock writerLock, TableMapping tm, ChunkRange range, ConnectorContext sourceCtx, AtomicLong rowsProcessed, AtomicLong batchesDone, long totalRows, long totalBatches, CheckpointStore checkpointStore, RuntimeProperties runtime, - Object progressLock, BooleanSupplier isCancelled, + ReentrantLock progressLock, BooleanSupplier isCancelled, java.util.function.BiConsumer persist, TriConsumer emit, MeterRegistry meterRegistry) { @@ -90,12 +91,15 @@ static void snapshotRange(SnapshotJob job, PipelineDesign pipeline, TenantContex .toList(); if (!batch.isEmpty()) { - synchronized (writerLock) { + writerLock.lock(); + try { if (useUpsert) { writer.upsertBatch(destTable, destCols, batch, keyCols); } else { writer.writeBatch(destTable, destCols, batch); } + } finally { + writerLock.unlock(); } } @@ -111,7 +115,8 @@ static void snapshotRange(SnapshotJob job, PipelineDesign pipeline, TenantContex (int) chunkBatch, rowsProcessed.get(), page.nextCursor())); } - synchronized (progressLock) { + progressLock.lock(); + try { if (!isCancelled.getAsBoolean() && chunkBatch % runtime.getSnapshot().getProgressPublishIntervalBatches() == 0) { var pct = totalRows > 0 ? (double) rowsProcessed.get() / totalRows * 100 : 0; @@ -121,6 +126,8 @@ static void snapshotRange(SnapshotJob job, PipelineDesign pipeline, TenantContex persist.accept(updated, tenantContext); emit.accept(job.getId().value(), updated, tenantContext); } + } finally { + progressLock.unlock(); } var nextBatchInfo = new BatchInformation( diff --git a/syncflow-api/src/main/resources/application.yml b/syncflow-api/src/main/resources/application.yml index 76370a0..2d411dd 100644 --- a/syncflow-api/src/main/resources/application.yml +++ b/syncflow-api/src/main/resources/application.yml @@ -4,6 +4,9 @@ server: spring: application: name: syncflow + threads: + virtual: + enabled: true datasource: url: jdbc:postgresql://localhost:5432/syncflow username: syncflow