lekhrocks · lekhrocks · Sep 6, 2026 · Sep 6, 2026 · Sep 6, 2026
diff --git 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
@@ -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
@@ -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<String, HikariDataSource> 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
@@ -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<Throwable> 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<WorkItem>(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<SnapshotCapableConnector>();
-            try {
-                for (int i = 0; i < poolSize; i++) {
-                    workerClones.add(connector.snapshotClone(sourceCtx));
-                }
-                var workers = new ArrayList<Thread>(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) {