From 5274f1f11d8bf8b9bfa8a0c4b6df91006446fe20 Mon Sep 17 00:00:00 2001 From: Glen Patzlaff Date: Thu, 13 Aug 2026 09:28:44 -0700 Subject: [PATCH 1/2] SYM-7915: stop initial loads deadlocking silently after a JVM restart mid-extract An initial load whose extract is interrupted by a JVM restart comes back permanently stuck. sym_extract_request reads OK with extracted_rows=0 while its batches are still RQ. Every extraction selector filters on NE so nothing re-extracts it, extractBatch refuses to deliver an RQ batch, nothing reconciles the two, and nothing surfaces it. The only symptom in the product is a pair of INFO log lines, so the console shows an ACTIVE pipeline making no progress and it is indistinguishable from a slow load. At the reporting site it ran three days, was never recovered, and the load had to be discarded. The state comes from one place. extractOutgoingBatch marked the entire request -- all of its batches -- OK from a single batch's extract_row_count and extract_millis, with no check that anything was actually extracted and no check that the rest of the range had finished. A batch that skipped extraction because it was already extracted contributes a zero row count and a near-zero duration, which is exactly the row observed. Three layers: - Guard. isExtractRequestComplete gates the status write. When the request is not complete the statistics accumulate through a new incremental statement instead, so progress stays visible without claiming completion. - Reconciler. recoverStuckExtractRequests returns such requests to NE, called from queueWork inside the existing cluster lock, which InitialLoadExtractorJob already drives every 10s. That covers startup for free, and the NE requests it produces are picked up later in the same invocation. A startup-only check would not have helped the reporting site, whose node ran three days in this state. - restartExtractRequest now also zeroes extracted_rows / extracted_millis, which it previously left behind, so a recovered request stops reporting counters from the run that was interrupted. Detection is exact rather than heuristic. MultiBatchStagingWriter.close() advances every remaining batch in the range, so a finished extract leaves none at RQ. A request at OK with an RQ batch in its range is therefore an impossible state, not a healthy in-flight one, which is what makes both the guard and the reconciler free of false positives against a running extract. Requests whose range contains already-delivered batches are deliberately NOT restarted automatically. restartExtractRequest flips the whole range back to RQ through updateOutgoingBatchStatusSql, which carries no status predicate, so it would re-send rows already committed at the target -- and the target's own record of them can be missing, which is the data-integrity half of this defect. Those are reported at ERROR every pass with the recovery choices and require force. initial.load.extract.request.recovery.enabled (default true) and ...threshold.ms (default 300000, only to avoid racing an in-flight status write). recoverStuckExtractRequests is on IDataExtractorService so an operator surface can bind it; DataExtractorService is the only implementer and FileSyncExtractorService inherits it, and there is no PRO implementer, so nothing else has to change. 7 new tests pinning the recovery statements. 1111 tests pass across symmetric-core, symmetric-db, symmetric-jdbc and symmetric-io with 0 failures. The end-to-end kill-and-recover path needs a real database and is not covered here; it is on the ticket as manual QA. Also found while reading and deliberately not folded in: sym_extract_request rows are inserted as LS and flipped LS->NE by DataService, and nothing recovers an LS request either, so a kill during load setup strands a load the same way. The reconciler is the natural home for it, but it is a separate defect and should be its own ticket. Co-Authored-By: Claude Opus 5 (1M context) --- .../symmetric/common/ParameterConstants.java | 2 + .../service/IDataExtractorService.java | 9 ++ .../service/impl/DataExtractorService.java | 93 ++++++++++++++- .../impl/DataExtractorServiceSqlMap.java | 26 ++++- .../resources/symmetric-default.properties | 20 ++++ .../impl/DataExtractorServiceSqlMapTest.java | 107 ++++++++++++++++++ 6 files changed, 253 insertions(+), 4 deletions(-) create mode 100644 symmetric-core/src/test/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMapTest.java diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java index ca0ffbf3c1..104b9c8b44 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java @@ -159,6 +159,8 @@ private ParameterConstants() { public static final String INITIAL_LOAD_CONCAT_CSV_IN_SQL_ENABLED = "initial.load.concat.csv.in.sql.enabled"; public static final String INITIAL_LOAD_USE_COLUMN_TEMPLATES_ENABLED = "initial.load.use.column.templates.enabled"; public static final String INITIAL_LOAD_EXTRACT_THREAD_COUNT_PER_SERVER = "initial.load.extract.thread.per.server.count"; + public static final String INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_ENABLED = "initial.load.extract.request.recovery.enabled"; + public static final String INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_THRESHOLD_MS = "initial.load.extract.request.recovery.threshold.ms"; public static final String INITIAL_LOAD_EXTRACT_MAX_PROCESS_TIME_MS = "initial.load.extract.max.process.time.ms"; public static final String INITIAL_LOAD_EXTRACT_TIMEOUT_MS = "initial.load.extract.timeout.ms"; public static final String INITIAL_LOAD_EXTRACT_USE_TWO_PASS_LOB = "initial.load.extract.use.two.pass.lob"; diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataExtractorService.java index a27ff2be04..4b2d53c4d8 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataExtractorService.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IDataExtractorService.java @@ -67,6 +67,15 @@ public ExtractRequest requestExtractRequest(ISqlTransaction transaction, String public void resetExtractRequest(OutgoingBatch batch); + /** + * Return extract requests to NE when they are marked complete but their batches are still requested, which is the state a load is left in when the JVM is + * interrupted mid-extract. Called automatically from {@link #queueWork(boolean)}; exposed so an operator can drive it directly, and so a request whose + * range contains already-delivered batches can be recovered with {@code force} once the target has been dealt with. + * + * @return the number of requests restarted + */ + public int recoverStuckExtractRequests(boolean force); + public void removeBatchFromStaging(OutgoingBatch batch); public StagingFileLock acquireStagingFileLock(OutgoingBatch batch); diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java index 2c31dfa0df..a64547f4cb 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java @@ -1050,9 +1050,22 @@ protected OutgoingBatch extractOutgoingBatch(ProcessInfo extractInfo, Node targe } ExtractRequest extractRequest = getExtractRequestForBatch(currentBatch); if (extractRequest != null && extractRequest.getStatus() != ExtractStatus.OK) { - sqlTemplate.update(getSql("updateExtractRequestStatus"), ExtractStatus.OK.name(), new Date(), - currentBatch.getExtractRowCount(), currentBatch.getExtractMillis(), extractRequest.getRequestId()); - checkSendDeferredForeignKeys(extractRequest.getLoadId(), targetNode); + if (isExtractRequestComplete(extractRequest, currentBatch, mode)) { + sqlTemplate.update(getSql("updateExtractRequestStatus"), ExtractStatus.OK.name(), new Date(), + currentBatch.getExtractRowCount(), currentBatch.getExtractMillis(), extractRequest.getRequestId()); + checkSendDeferredForeignKeys(extractRequest.getLoadId(), targetNode); + } else { + /* + * One batch finishing does not mean the request did. Marking the whole request OK from a single batch's counters is how a request + * ends up at OK with extracted_rows=0 while its batches are still RQ: every extraction selector filters on NE so nothing + * re-extracts it, extractBatch refuses to deliver an RQ batch, and nothing reconciles the two or raises an error. Accumulate the + * statistics instead, so progress stays visible without claiming completion. + */ + sqlTemplate.update(getSql("updateExtractRequestExtractedStats"), currentBatch.getExtractRowCount(), + currentBatch.getExtractMillis(), new Date(), extractRequest.getRequestId()); + log.debug("Batch {} finished but extract request {} for table {} is not complete; leaving its status at {}", + currentBatch.getBatchId(), extractRequest.getRequestId(), extractRequest.getTableName(), extractRequest.getStatus()); + } } } } @@ -1727,6 +1740,9 @@ public RemoteNodeStatuses queueWork(boolean force) { if (identity != null) { if (force || clusterService.lock(ClusterConstants.INITIAL_LOAD_EXTRACT)) { try { + // Before assigning threads, put back any request that is marked complete but demonstrably is not. + // The NE requests this produces are picked up by getExtractRequestNodes() in this same invocation. + recoverStuckExtractRequests(false); updateExtractRequestsForThreading(); List nodes = getExtractRequestNodes(); for (NodeQueuePair pair : nodes) { @@ -2087,6 +2103,77 @@ public void execute(NodeCommunication nodeCommunication, RemoteNodeStatus status } } + /** + * Whether finishing this batch means the whole extract request is done. + *

+ * The detection is exact rather than heuristic: {@code MultiBatchStagingWriter.close()} advances every remaining batch in the range, so after a completed + * extract-job run none of them are left at {@code RQ}. A request at {@code OK} with any batch in its range still {@code RQ} is therefore an impossible + * state rather than a healthy in-flight one, which is what makes both this guard and {@link #recoverStuckExtractRequests(boolean)} safe from false + * positives. + */ + protected boolean isExtractRequestComplete(ExtractRequest request, OutgoingBatch currentBatch, ExtractMode mode) { + if (mode == ExtractMode.EXTRACT_ONLY) { + // changeBatchStatus does not persist in this mode, so the batch's own row would still read RQ. + return false; + } + if (countRequestedBatchesInRange(request) > 0) { + return false; + } + boolean multiBatch = request.getEndBatchId() > request.getStartBatchId(); + return !(multiBatch && currentBatch.getExtractRowCount() == 0 && request.getRows() > 0); + } + + protected int countRequestedBatchesInRange(ExtractRequest request) { + return sqlTemplate.queryForInt(getSql("countRequestedBatchesForExtractRequestSql"), request.getNodeId(), + request.getStartBatchId(), request.getEndBatchId()); + } + + /** + * Return extract requests to {@code NE} when they are marked complete but demonstrably are not, so a load interrupted mid-extract resumes instead of + * sitting silently forever. Runs from {@link #queueWork(boolean)} under the existing cluster lock, which also covers startup; a startup-only check would + * not have helped the reporting site, whose node ran three days in this state. + *

+ * Requests whose range contains already-delivered batches are not restarted automatically. {@code restartExtractRequest} flips the whole range + * back to {@code RQ} through a statement with no status predicate, so it would re-send rows already committed at the target, and the target's own record of + * them can be missing (that is the data-integrity half of this defect). Those are reported every pass and require {@code force}. + * + * @return the number of requests restarted + */ + public int recoverStuckExtractRequests(boolean force) { + if (!parameterService.is(ParameterConstants.INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_ENABLED, true)) { + return 0; + } + long thresholdMs = parameterService.getLong(ParameterConstants.INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_THRESHOLD_MS, 300000); + Date staleBefore = new Date(System.currentTimeMillis() - thresholdMs); + List stuck = sqlTemplateDirty.query(getSql("selectStuckExtractRequestsSql"), new ExtractRequestMapper(), + engine.getNodeId(), ExtractStatus.OK.name(), staleBefore); + int restarted = 0; + for (ExtractRequest request : stuck) { + int delivered = sqlTemplate.queryForInt(getSql("countDeliveredBatchesForExtractRequestSql"), request.getNodeId(), + request.getStartBatchId(), request.getEndBatchId()); + if (delivered > 0 && !force) { + log.error( + "Extract request {} for table {} (load {}, node {}) is marked {} with {} of {} rows extracted, but batches {} through {} are still " + + "requested. {} of them were already delivered, so restarting it would re-send rows that are already committed at the target. " + + "This load will not progress on its own: either truncate the target table and force recovery, or cancel the load.", + request.getRequestId(), request.getTableName(), request.getLoadId(), request.getNodeId(), ExtractStatus.OK.name(), + request.getExtractedRows(), request.getRows(), request.getStartBatchId(), request.getEndBatchId(), delivered); + continue; + } + log.warn("Extract request {} for table {} (load {}, node {}) is marked {} with {} of {} rows extracted while batches {} through {} are still " + + "requested, which cannot happen on a completed extract. Re-queuing it for extraction.", + request.getRequestId(), request.getTableName(), request.getLoadId(), request.getNodeId(), ExtractStatus.OK.name(), + request.getExtractedRows(), request.getRows(), request.getStartBatchId(), request.getEndBatchId()); + List batches = outgoingBatchService.getOutgoingBatchRange(request.getStartBatchId(), request.getEndBatchId()).getBatches(); + restartExtractRequest(batches, request, getExtractChildRequestsForNode(request)); + restarted++; + } + if (restarted > 0) { + log.warn("Recovered {} stuck extract request(s)", restarted); + } + return restarted; + } + protected void restartExtractRequest(List batches, ExtractRequest request, List childRequests) { /* * This extract request was interrupted and must start over diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMap.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMap.java index 8f9882de79..ae2ddbaee3 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMap.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMap.java @@ -89,10 +89,34 @@ public DataExtractorServiceSqlMap(IDatabasePlatform platform, putSql("updateOutgoingBatchesForSetupThreadSql", "update $(outgoing_batch) set thread_id = ? where load_id = ? and summary like ? and thread_id is null"); + // extracted_rows / extracted_millis are reset alongside the transfer and load statistics. Leaving them behind makes a + // restarted request report extraction counters from the run that was interrupted, which is the misleading state this + // recovery path exists to clear. putSql("restartExtractRequest", "update $(extract_request) set last_transferred_batch_id = null, transferred_rows = 0, transferred_millis = 0, " - + "last_loaded_batch_id = null, loaded_rows = 0, loaded_millis = 0, parent_request_id = 0, status = ? " + + "last_loaded_batch_id = null, loaded_rows = 0, loaded_millis = 0, extracted_rows = 0, extracted_millis = 0, " + + "parent_request_id = 0, status = ? " + "where request_id = ? and node_id = ?"); + putSql("updateExtractRequestExtractedStats", "update $(extract_request) set extracted_rows = extracted_rows + ?, " + + "extracted_millis = extracted_millis + ?, last_update_time = ? where request_id = ?"); + + putSql("countRequestedBatchesForExtractRequestSql", + "select count(*) from $(outgoing_batch) where node_id = ? and batch_id between ? and ? and status = 'RQ'"); + + putSql("countDeliveredBatchesForExtractRequestSql", + "select count(*) from $(outgoing_batch) where node_id = ? and batch_id between ? and ? and status in ('OK','IG')"); + + /* + * A request marked OK whose range still contains RQ batches cannot have completed: MultiBatchStagingWriter.close() advances every remaining batch, so a + * finished extract leaves none at RQ. loaded_time is null excludes requests that legitimately finished and were loaded, and the last_update_time bound + * avoids racing a status write that is still in flight. + */ + putSql("selectStuckExtractRequestsSql", "select * from $(extract_request) r where r.source_node_id = ? and r.status = ? " + + "and r.loaded_time is null and r.parent_request_id = 0 and r.last_update_time < ? " + + "and exists (select 1 from $(outgoing_batch) b where b.node_id = r.node_id " + + "and b.batch_id between r.start_batch_id and r.end_batch_id and b.status = 'RQ') " + + "order by r.load_id asc, r.request_id asc"); + putSql("cancelExtractRequests", "update $(extract_request) set status=?, last_update_time=?, loaded_time=? where load_id = ? and source_node_id = ? and (status != ? or loaded_time is null)"); putSql("countIncompleteExtractRequestsByLoadId", "select count(*) from $(extract_request) where load_id = ? and source_node_id = ? and parent_request_id = 0 and status != 'OK'"); diff --git a/symmetric-core/src/main/resources/symmetric-default.properties b/symmetric-core/src/main/resources/symmetric-default.properties index 9946ebafd2..664ac04a73 100644 --- a/symmetric-core/src/main/resources/symmetric-default.properties +++ b/symmetric-core/src/main/resources/symmetric-default.properties @@ -1037,6 +1037,26 @@ initial.load.use.extract.job.enabled=true # Type: integer initial.load.extract.thread.per.server.count=20 +# Return an extract request to a requested state when it is marked complete but its batches are still requested. +# That combination cannot occur on a completed extract, because closing the extract advances every batch in the +# range; it is the state a load is left in when the JVM is interrupted mid-extract, and nothing else recovers it. +# Left alone, the load makes no further progress and raises no error, so it is indistinguishable from a slow load. +# A request whose range contains batches that were already delivered is reported but not restarted automatically, +# because restarting re-sends rows that are already committed at the target. +# +# DatabaseOverridable: true +# Tags: load +# Type: boolean +initial.load.extract.request.recovery.enabled=true + +# How long an extract request must have been untouched before recovery will consider it stuck. This only exists to +# avoid racing a status update that is still in flight, so it does not need to be long. +# +# DatabaseOverridable: true +# Tags: load +# Type: integer +initial.load.extract.request.recovery.threshold.ms=300000 + # The number of milliseconds that the initial load extract job can run a thread # for processing extracting requests. # diff --git a/symmetric-core/src/test/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMapTest.java b/symmetric-core/src/test/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMapTest.java new file mode 100644 index 0000000000..c26d62aad7 --- /dev/null +++ b/symmetric-core/src/test/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMapTest.java @@ -0,0 +1,107 @@ +/** + * Licensed to JumpMind Inc under one or more contributor + * license agreements. See the NOTICE file distributed + * with this work for additional information regarding + * copyright ownership. JumpMind Inc licenses this file + * to you under the GNU General Public License, version 3.0 (GPLv3) + * (the "License"); you may not use this file except in compliance + * with the License. + * + * You should have received a copy of the GNU General Public License, + * version 3.0 (GPLv3) along with this library; if not, see + * . + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.jumpmind.symmetric.service.impl; + +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Map; + +import org.junit.jupiter.api.Test; + +/** + * SYM-7915. Pins the statements behind stuck-extract-request detection and recovery. + */ +class DataExtractorServiceSqlMapTest { + private DataExtractorServiceSqlMap sqlMap() { + return new DataExtractorServiceSqlMap(null, (Map) null); + } + + @Test + void restartExtractRequestResetsTheExtractionCounters() { + /* + * Without this a restarted request keeps the extraction counters from the interrupted run, so it still reports rows it no longer has -- the misleading + * state the recovery exists to clear. + */ + String sql = sqlMap().getSql("restartExtractRequest"); + assertTrue(sql.contains("extracted_rows = 0"), sql); + assertTrue(sql.contains("extracted_millis = 0"), sql); + assertTrue(sql.contains("status = ?"), sql); + } + + @Test + void restartExtractRequestStillResetsTransferAndLoadCounters() { + // Guard against the added resets displacing the ones that were already there. + String sql = sqlMap().getSql("restartExtractRequest"); + assertTrue(sql.contains("transferred_rows = 0"), sql); + assertTrue(sql.contains("loaded_rows = 0"), sql); + assertTrue(sql.contains("last_transferred_batch_id = null"), sql); + assertTrue(sql.contains("last_loaded_batch_id = null"), sql); + assertTrue(sql.contains("parent_request_id = 0"), sql); + } + + @Test + void stuckRequestDetectionRequiresAllFourConditions() { + /* + * The detection is only sound with every part present: status OK, no load time, a batch in the range still requested, and a settling period so a status + * write that is still in flight is not mistaken for a wedge. + */ + String sql = sqlMap().getSql("selectStuckExtractRequestsSql"); + assertTrue(sql.contains("r.status = ?"), sql); + assertTrue(sql.contains("r.loaded_time is null"), sql); + assertTrue(sql.contains("r.last_update_time < ?"), sql); + assertTrue(sql.contains("b.status = 'RQ'"), sql); + assertTrue(sql.contains("b.batch_id between r.start_batch_id and r.end_batch_id"), sql); + } + + @Test + void stuckRequestDetectionSkipsChildRequests() { + // Children are restarted through their parent, so selecting them independently would double up the recovery. + assertTrue(sqlMap().getSql("selectStuckExtractRequestsSql").contains("r.parent_request_id = 0")); + } + + @Test + void requestedAndDeliveredCountsLookAtDifferentStatuses() { + // The first decides whether a request is stuck; the second decides whether restarting it would re-send rows. + String requested = sqlMap().getSql("countRequestedBatchesForExtractRequestSql"); + String delivered = sqlMap().getSql("countDeliveredBatchesForExtractRequestSql"); + assertTrue(requested.contains("status = 'RQ'"), requested); + assertTrue(delivered.contains("status in ('OK','IG')"), delivered); + assertFalse(delivered.contains("'RQ'"), delivered); + } + + @Test + void extractedStatsUpdateAccumulatesRatherThanOverwriting() { + // A partial run must add to what previous batches recorded, or progress on a multi-batch request is lost. + String sql = sqlMap().getSql("updateExtractRequestExtractedStats"); + assertTrue(sql.contains("extracted_rows = extracted_rows + ?"), sql); + assertTrue(sql.contains("extracted_millis = extracted_millis + ?"), sql); + assertFalse(sql.contains("status"), "the incremental update must not touch status: " + sql); + } + + @Test + void updateExtractRequestStatusIsUnchangedForTheCompletionPath() { + // Still an absolute write, since execute() uses it when the request genuinely finished. + String sql = sqlMap().getSql("updateExtractRequestStatus"); + assertTrue(sql.contains("status=?"), sql); + assertTrue(sql.contains("extracted_rows=?"), sql); + } +} From 3203569a82515e70b9f14305178b3c130c4dc035 Mon Sep 17 00:00:00 2001 From: Glen Patzlaff Date: Thu, 13 Aug 2026 11:09:37 -0700 Subject: [PATCH 2/2] SYM-7915: stop the recovery scan and the completeness check doing needless work Quality pass, no behaviour change. All 781 tests in :symmetric-core still pass. Two real costs, both found by tracing call frequency: - isExtractRequestComplete ran a count(*) per delivered load batch on the pull path. The answer is only ever compared to zero, but a count has to visit every batch in the request's range, so across a load split into N batches the cost grew with N squared, on the same table the extract job is writing to. It now selects one row and stops at the first match, via the existing ISqlTemplate.query(sql, maxRowsToFetch, ...) overload. Same for the delivered-batch check. Both SQL keys renamed from count* to select* to match. - The free in-memory row-count check now runs before that query rather than after, so the query is skipped entirely in the case it would answer. - recoverStuckExtractRequests ran from queueWork every 10 seconds, but its own staleness threshold means a request cannot qualify until it has been untouched for 5 minutes. Twenty-nine of every thirty scans could not have a new answer, and each walked every completed request for the node plus a correlated probe into the outgoing batch table. It now skips until the threshold has elapsed. Placement: FileSyncExtractorService extends DataExtractorService and does not override queueWork, so the recovery was also running on the file-sync instance, where it would have scanned and restarted ordinary data extract requests and then resolved their staging through the file-sync getStagedResource override. It now returns 0 there, matching how updateExtractRequestsForThreading is already handled in that class for the same reason. Also: added the missing @Override; extracted the two nine-argument log statements that repeated the same sentence into describeStuckRequest; extracted the load-range plus child-requests plus restart idiom, which resetExtractRequest already performed, into a shared restartExtractRequest(request) that keeps the parent guard rather than relying on the SQL filter to imply it. Corrected the javadoc: it claimed the detection is "exact rather than heuristic", but the zero-row clause is a heuristic. The batch-status half is exact; the heuristic half is now labelled as one, with the case it misjudges named. Co-Authored-By: Claude Opus 5 (1M context) --- .../service/impl/DataExtractorService.java | 92 +++++++++++++------ .../impl/DataExtractorServiceSqlMap.java | 8 +- .../impl/FileSyncExtractorService.java | 9 ++ .../impl/DataExtractorServiceSqlMapTest.java | 42 ++++----- 4 files changed, 95 insertions(+), 56 deletions(-) diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java index a64547f4cb..c333076b23 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorService.java @@ -1733,6 +1733,9 @@ protected TransformWriter createTransformDataWriter(Node identity, Node targetNo return transformExtractWriter; } + /** When recovery last scanned for stuck extract requests, so it is not re-run faster than a request can become stale. */ + protected long lastStuckExtractRequestCheckMs; + @Override public RemoteNodeStatuses queueWork(boolean force) { final RemoteNodeStatuses statuses = new RemoteNodeStatuses(configurationService.getChannels(false)); @@ -2106,66 +2109,82 @@ public void execute(NodeCommunication nodeCommunication, RemoteNodeStatus status /** * Whether finishing this batch means the whole extract request is done. *

- * The detection is exact rather than heuristic: {@code MultiBatchStagingWriter.close()} advances every remaining batch in the range, so after a completed - * extract-job run none of them are left at {@code RQ}. A request at {@code OK} with any batch in its range still {@code RQ} is therefore an impossible - * state rather than a healthy in-flight one, which is what makes both this guard and {@link #recoverStuckExtractRequests(boolean)} safe from false - * positives. + * The batch-status half of the test is exact rather than heuristic: {@code MultiBatchStagingWriter.close()} advances every remaining batch in the range, so + * after a completed extract-job run none of them are left at {@code RQ}. A request at {@code OK} with any batch in its range still {@code RQ} is therefore + * an impossible state rather than a healthy in-flight one, which is what makes both this guard and {@link #recoverStuckExtractRequests(boolean)} safe from + * false positives. The zero-row check is a heuristic and is called out as such where it appears. */ protected boolean isExtractRequestComplete(ExtractRequest request, OutgoingBatch currentBatch, ExtractMode mode) { if (mode == ExtractMode.EXTRACT_ONLY) { // changeBatchStatus does not persist in this mode, so the batch's own row would still read RQ. return false; } - if (countRequestedBatchesInRange(request) > 0) { + if (request.getEndBatchId() > request.getStartBatchId() && currentBatch.getExtractRowCount() == 0 && request.getRows() > 0) { + /* + * Heuristic, and checked before the query because it costs nothing: a multi-batch request whose expected row count is non-zero cannot be finished + * by a batch that extracted nothing, because those rows belong to another batch in the range. A legitimately empty final batch on a request that + * expects rows is the case this misjudges, and it errs towards leaving the request open, which the reconciler then resolves. + */ return false; } - boolean multiBatch = request.getEndBatchId() > request.getStartBatchId(); - return !(multiBatch && currentBatch.getExtractRowCount() == 0 && request.getRows() > 0); + return !hasRequestedBatchesInRange(request); + } + + /** + * Whether any batch in the request's range is still requested. Deliberately not a {@code count(*)}: the answer is only ever used as a boolean, and a count + * has to visit every batch in the range, so across a load split into thousands of batches the cost would grow with the square of the batch count. Fetching + * one row stops at the first match, which in the common in-progress case is immediate. + */ + protected boolean hasRequestedBatchesInRange(ExtractRequest request) { + return !sqlTemplate.query(getSql("selectRequestedBatchesForExtractRequestSql"), 1, new LongMapper(), + new Object[] { request.getNodeId(), request.getStartBatchId(), request.getEndBatchId() }, null).isEmpty(); } - protected int countRequestedBatchesInRange(ExtractRequest request) { - return sqlTemplate.queryForInt(getSql("countRequestedBatchesForExtractRequestSql"), request.getNodeId(), - request.getStartBatchId(), request.getEndBatchId()); + /** Whether any batch in the request's range was already delivered, which makes an automatic restart unsafe. */ + protected boolean hasDeliveredBatchesInRange(ExtractRequest request) { + return !sqlTemplate.query(getSql("selectDeliveredBatchesForExtractRequestSql"), 1, new LongMapper(), + new Object[] { request.getNodeId(), request.getStartBatchId(), request.getEndBatchId() }, null).isEmpty(); } /** * Return extract requests to {@code NE} when they are marked complete but demonstrably are not, so a load interrupted mid-extract resumes instead of - * sitting silently forever. Runs from {@link #queueWork(boolean)} under the existing cluster lock, which also covers startup; a startup-only check would - * not have helped the reporting site, whose node ran three days in this state. + * sitting silently forever. Runs from {@link #queueWork(boolean)}, which also covers startup; a startup-only check would not have helped the reporting + * site, whose node ran three days in this state. *

* Requests whose range contains already-delivered batches are not restarted automatically. {@code restartExtractRequest} flips the whole range * back to {@code RQ} through a statement with no status predicate, so it would re-send rows already committed at the target, and the target's own record of - * them can be missing (that is the data-integrity half of this defect). Those are reported every pass and require {@code force}. + * them can be missing (that is the data-integrity half of this defect). Those are reported and require {@code force}. * * @return the number of requests restarted */ + @Override public int recoverStuckExtractRequests(boolean force) { if (!parameterService.is(ParameterConstants.INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_ENABLED, true)) { return 0; } long thresholdMs = parameterService.getLong(ParameterConstants.INITIAL_LOAD_EXTRACT_REQUEST_RECOVERY_THRESHOLD_MS, 300000); - Date staleBefore = new Date(System.currentTimeMillis() - thresholdMs); + long now = System.currentTimeMillis(); + /* + * A request cannot qualify until it has been untouched for the threshold, so scanning faster than that only re-runs a query that cannot have a new + * answer. queueWork runs every 10 seconds; at the default threshold that would be 29 wasted scans out of every 30, each walking every completed request + * for this node plus a correlated probe into the outgoing batch table. + */ + if (!force && now - lastStuckExtractRequestCheckMs < thresholdMs) { + return 0; + } + lastStuckExtractRequestCheckMs = now; List stuck = sqlTemplateDirty.query(getSql("selectStuckExtractRequestsSql"), new ExtractRequestMapper(), - engine.getNodeId(), ExtractStatus.OK.name(), staleBefore); + engine.getNodeId(), ExtractStatus.OK.name(), new Date(now - thresholdMs)); int restarted = 0; for (ExtractRequest request : stuck) { - int delivered = sqlTemplate.queryForInt(getSql("countDeliveredBatchesForExtractRequestSql"), request.getNodeId(), - request.getStartBatchId(), request.getEndBatchId()); - if (delivered > 0 && !force) { - log.error( - "Extract request {} for table {} (load {}, node {}) is marked {} with {} of {} rows extracted, but batches {} through {} are still " - + "requested. {} of them were already delivered, so restarting it would re-send rows that are already committed at the target. " - + "This load will not progress on its own: either truncate the target table and force recovery, or cancel the load.", - request.getRequestId(), request.getTableName(), request.getLoadId(), request.getNodeId(), ExtractStatus.OK.name(), - request.getExtractedRows(), request.getRows(), request.getStartBatchId(), request.getEndBatchId(), delivered); + if (hasDeliveredBatchesInRange(request) && !force) { + log.error("{} Some of those batches were already delivered, so restarting it would re-send rows that are already committed at the target. " + + "This load will not progress on its own: either truncate the target table and force recovery, or cancel the load.", + describeStuckRequest(request)); continue; } - log.warn("Extract request {} for table {} (load {}, node {}) is marked {} with {} of {} rows extracted while batches {} through {} are still " - + "requested, which cannot happen on a completed extract. Re-queuing it for extraction.", - request.getRequestId(), request.getTableName(), request.getLoadId(), request.getNodeId(), ExtractStatus.OK.name(), - request.getExtractedRows(), request.getRows(), request.getStartBatchId(), request.getEndBatchId()); - List batches = outgoingBatchService.getOutgoingBatchRange(request.getStartBatchId(), request.getEndBatchId()).getBatches(); - restartExtractRequest(batches, request, getExtractChildRequestsForNode(request)); + log.warn("{} That cannot happen on a completed extract, so it is being re-queued for extraction.", describeStuckRequest(request)); + restartExtractRequest(request); restarted++; } if (restarted > 0) { @@ -2174,6 +2193,19 @@ public int recoverStuckExtractRequests(boolean force) { return restarted; } + protected String describeStuckRequest(ExtractRequest request) { + return String.format("Extract request %d for table %s (load %d, node %s) is marked %s with %d of %d rows extracted, while batches %d through %d are " + + "still requested.", request.getRequestId(), request.getTableName(), request.getLoadId(), request.getNodeId(), ExtractStatus.OK.name(), + request.getExtractedRows(), request.getRows(), request.getStartBatchId(), request.getEndBatchId()); + } + + /** Load the batch range and any child requests, then hand off to {@link #restartExtractRequest(List, ExtractRequest, List)}. */ + protected void restartExtractRequest(ExtractRequest request) { + List batches = outgoingBatchService.getOutgoingBatchRange(request.getStartBatchId(), request.getEndBatchId()).getBatches(); + List childRequests = request.getParentRequestId() == 0 ? getExtractChildRequestsForNode(request) : null; + restartExtractRequest(batches, request, childRequests); + } + protected void restartExtractRequest(List batches, ExtractRequest request, List childRequests) { /* * This extract request was interrupted and must start over diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMap.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMap.java index ae2ddbaee3..44ae2b78d4 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMap.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMap.java @@ -100,11 +100,11 @@ public DataExtractorServiceSqlMap(IDatabasePlatform platform, putSql("updateExtractRequestExtractedStats", "update $(extract_request) set extracted_rows = extracted_rows + ?, " + "extracted_millis = extracted_millis + ?, last_update_time = ? where request_id = ?"); - putSql("countRequestedBatchesForExtractRequestSql", - "select count(*) from $(outgoing_batch) where node_id = ? and batch_id between ? and ? and status = 'RQ'"); + putSql("selectRequestedBatchesForExtractRequestSql", + "select batch_id from $(outgoing_batch) where node_id = ? and batch_id between ? and ? and status = 'RQ'"); - putSql("countDeliveredBatchesForExtractRequestSql", - "select count(*) from $(outgoing_batch) where node_id = ? and batch_id between ? and ? and status in ('OK','IG')"); + putSql("selectDeliveredBatchesForExtractRequestSql", + "select batch_id from $(outgoing_batch) where node_id = ? and batch_id between ? and ? and status in ('OK','IG')"); /* * A request marked OK whose range still contains RQ batches cannot have completed: MultiBatchStagingWriter.close() advances every remaining batch, so a diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/FileSyncExtractorService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/FileSyncExtractorService.java index 7fc5400118..86881daa61 100644 --- a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/FileSyncExtractorService.java +++ b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/impl/FileSyncExtractorService.java @@ -74,6 +74,15 @@ protected boolean isApplicable(NodeCommunication nodeCommunication) { protected void updateExtractRequestsForThreading() { } + /** + * Not applicable to file sync, for the same reason as {@link #updateExtractRequestsForThreading()}: this service inherits {@code queueWork} and would + * otherwise scan and restart ordinary data extract requests, then resolve their staging through the file-sync {@code getStagedResource} override. + */ + @Override + public int recoverStuckExtractRequests(boolean force) { + return 0; + } + @Override protected boolean canProcessExtractRequest(ExtractRequest request, CommunicationType communicationType) { return request.getTableName().equalsIgnoreCase(TableConstants.getTableName(tablePrefix, TableConstants.SYM_FILE_SNAPSHOT)); diff --git a/symmetric-core/src/test/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMapTest.java b/symmetric-core/src/test/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMapTest.java index c26d62aad7..2da934e524 100644 --- a/symmetric-core/src/test/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMapTest.java +++ b/symmetric-core/src/test/java/org/jumpmind/symmetric/service/impl/DataExtractorServiceSqlMapTest.java @@ -26,6 +26,8 @@ import java.util.Map; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; /** * SYM-7915. Pins the statements behind stuck-extract-request detection and recovery. @@ -35,27 +37,18 @@ private DataExtractorServiceSqlMap sqlMap() { return new DataExtractorServiceSqlMap(null, (Map) null); } - @Test - void restartExtractRequestResetsTheExtractionCounters() { + @ParameterizedTest + @ValueSource( + strings = { "extracted_rows = 0", "extracted_millis = 0", "transferred_rows = 0", "loaded_rows = 0", + "last_transferred_batch_id = null", "last_loaded_batch_id = null", "parent_request_id = 0", "status = ?" }) + void restartExtractRequestResetsEveryCounter(String fragment) { /* - * Without this a restarted request keeps the extraction counters from the interrupted run, so it still reports rows it no longer has -- the misleading - * state the recovery exists to clear. + * The extraction counters are the ones this change added: without them a restarted request keeps the counters from the interrupted run and still + * reports rows it no longer has, which is the misleading state the recovery exists to clear. The rest guard against the additions displacing what was + * already there. */ String sql = sqlMap().getSql("restartExtractRequest"); - assertTrue(sql.contains("extracted_rows = 0"), sql); - assertTrue(sql.contains("extracted_millis = 0"), sql); - assertTrue(sql.contains("status = ?"), sql); - } - - @Test - void restartExtractRequestStillResetsTransferAndLoadCounters() { - // Guard against the added resets displacing the ones that were already there. - String sql = sqlMap().getSql("restartExtractRequest"); - assertTrue(sql.contains("transferred_rows = 0"), sql); - assertTrue(sql.contains("loaded_rows = 0"), sql); - assertTrue(sql.contains("last_transferred_batch_id = null"), sql); - assertTrue(sql.contains("last_loaded_batch_id = null"), sql); - assertTrue(sql.contains("parent_request_id = 0"), sql); + assertTrue(sql.contains(fragment), sql); } @Test @@ -79,13 +72,18 @@ void stuckRequestDetectionSkipsChildRequests() { } @Test - void requestedAndDeliveredCountsLookAtDifferentStatuses() { - // The first decides whether a request is stuck; the second decides whether restarting it would re-send rows. - String requested = sqlMap().getSql("countRequestedBatchesForExtractRequestSql"); - String delivered = sqlMap().getSql("countDeliveredBatchesForExtractRequestSql"); + void requestedAndDeliveredLookupsUseDifferentStatusesAndDoNotCount() { + /* + * The first decides whether a request is stuck; the second decides whether restarting it would re-send rows. Both answers are booleans, so they select + * rows and stop at the first match rather than counting -- a count would have to visit the whole batch range. + */ + String requested = sqlMap().getSql("selectRequestedBatchesForExtractRequestSql"); + String delivered = sqlMap().getSql("selectDeliveredBatchesForExtractRequestSql"); assertTrue(requested.contains("status = 'RQ'"), requested); assertTrue(delivered.contains("status in ('OK','IG')"), delivered); assertFalse(delivered.contains("'RQ'"), delivered); + assertFalse(requested.contains("count("), requested); + assertFalse(delivered.contains("count("), delivered); } @Test