diff --git a/docs/chunk-scheduling-redesign.md b/docs/chunk-scheduling-redesign.md index 1000e7e112..1804b5ed7c 100644 --- a/docs/chunk-scheduling-redesign.md +++ b/docs/chunk-scheduling-redesign.md @@ -285,10 +285,13 @@ what covers the ordinary case where they finish after. Both apply the cross-job both are no-ops for a job that has no termination chunk at all. Site A has to know whether the chunk it just removed was the termination chunk, so that the -job's barrier does not count itself. That falls out of the removal token if +job's barrier does not count itself. That falls out of the removal token, because `DependencyTrackingService.remove(TrackingKey)` returns the removed `DependencyTracking` -rather than a boolean: the winning caller gets the proof it won *and* the `is_termination` -flag of the row it removed, in one call and with no extra read. +rather than void: the winning caller gets the proof it won *and* the `is_termination` +flag of the entry it removed, in one call and with no extra read. The flag is a field on that +object as well as a column on the row, which is sound for a value decided when the row is created +and never changed afterwards. `gate_open` has no such field, deliberately, see [Who writes the gate +columns before Phase 9](#barrier-chunks--per-job-gate). **Who writes the gate columns before Phase 9.** Everything above is written as if `dependencytracking` were already a plain PostgreSQL table. It becomes one at [Phase 9]( @@ -422,13 +425,14 @@ why the increment must be a single atomic statement (`SET data_chunks_delivered = data_chunks_delivered + 1`) rather than a read followed by a write. -The removal of the chunk's `dependencytracking` row is the natural token, since exactly -one concurrent caller can perform it. The increment must therefore be conditioned on -having won that removal, not on a preceding read of the row - the current -`get`-then-`remove` sequence in `chunkDeliveringDone` is a non-atomic check-then-act that -two callers can both pass. Harmless for the idempotent work that follows it today, not -harmless for a counter. See [Phase 1](#phase-1--gate-and-ordered-dispatch-job-store-service) -for the required signature change. The `>=` in the pseudocode above is defensive only: +The removal of the chunk's `dependencytracking` row is the token, since exactly one +concurrent caller can perform it, and the increment is conditioned on having won that +removal rather than on a preceding read of the row. `chunkDeliveringDone` still reads the +entry first, to answer "is this chunk in `QUEUED_FOR_DELIVERY`" and to turn an unknown chunk +away, but that read is a filter and not the token: `get`-then-`remove` is a non-atomic +check-then-act that two callers can both pass. Harmless for the idempotent work that follows +it, not harmless for a counter. See [Phase 1](#phase-1--gate-and-ordered-dispatch-job-store-service) +for the signature. The `>=` in the pseudocode above is defensive only: it does not mitigate either hazard above, since double counting reaches the total exactly and a lost update leaves the counter below it forever. It guards only against the counter being pushed above `data_chunks_expected` by something other than the increment, such as a @@ -2033,7 +2037,11 @@ Two ordering constraints shape the sequence: transaction - **DI-3049** Change `DependencyTrackingService.remove(TrackingKey)` to return the removed `DependencyTracking`, so the caller learns whether *this* call - performed the removal, and condition the `data_chunks_delivered` increment on it. + performed the removal, and condition the `data_chunks_delivered` increment on it. Carry + `is_termination` on `DependencyTracking` as well, so the caller that won the removal reads the + branch off the entry it was handed rather than querying the row it has just removed. That is the + shape Phase 9 arrives at anyway, once the row is deleted synchronously and a read after the + removal would return nothing. It already computes this and discards it: `dependencyTracker.remove(key)` returns the previous value atomically per key and the method checks `removed == null`, but returns `void`, so no caller can learn it won the race. Without this the increment sits behind diff --git a/job-store-service/dependency-tracking.md b/job-store-service/dependency-tracking.md index c318329580..0f2847f018 100644 --- a/job-store-service/dependency-tracking.md +++ b/job-store-service/dependency-tracking.md @@ -117,7 +117,16 @@ depend on a row existing at a particular moment. Job-store owns the **gate column values**. `DependencyTrackingStore`'s upsert names five columns in its `on conflict ... do update set` clause, and cannot clobber a column it does not name. `is_termination` and `gate_open` are outside that list, which is what lets job-store write them in -plain synchronous SQL with no lag. **Never add those two columns to that clause.** +plain synchronous SQL with no lag. **Never add those two columns to that clause**, which +`JobGateIT.mapStoreDoesNotClobberGateColumns` is there to enforce. + +`is_termination` is also a field on `DependencyTracking`, and that is not a contradiction. The clause +still does not name it, so nothing about who writes the column changes, and a copy on the map value +cannot go stale for a value decided when the row is created and never changed. A delivery reads the +branch off the entry it removed rather than querying a row it has just removed, see +[Where the gate is decided](#where-the-gate-is-decided). **`gate_open` gets no such field.** Four +sites write it over a chunk's life, so a copy would be a second answer to a question with one, and a +stale open gate dispatches a job's end-of-job work ahead of the data it summarises. Three consequences follow: @@ -161,6 +170,16 @@ Five things about those sites are easy to get wrong: having a termination chunk, it would lose every chunk delivered before that chunk existed. As a read-then-write, it could lose an update and leave the counter permanently one short. Either way the gate never opens. +- **Site A runs once per chunk, and what makes that true is the removal.** `chunkDeliveringDone` is + called again by every redelivery, and the broker's failure detection can produce two genuinely + concurrent calls for one chunk, so the count hangs off the one thing only one caller can do: + `DependencyTrackingService.remove` hands the removed entry to whichever caller removed it and null + to every other. The read of the entry before it is a filter on status, not the token, since + `get`-then-`remove` is a check-then-act two callers can both pass. Counted twice, the counter still + lands on `data_chunks_expected` exactly, only while a data chunk is still in flight, so the failure + is a job that reads as complete rather than one that stalls. The rest of `chunkDeliveringDone` runs + for every caller: `removeFromWaitingOn` reports only the entries it changed, so the caller that + lost the removal finds nothing left to unblock. - **`data_chunks_expected = 0` means two opposite things**: the migration default on jobs that predate the gate, which must be ignored, and a genuine job with no data chunks, whose gate must be decided at once. `is_termination` is what tells them apart, so the gate keys on that and uses the @@ -275,6 +294,14 @@ Two ways a gate outlives its reason to be shut, each of them a job that never co loss on the termination branch never lifts the barrier, stalling every later job from that submitter. +That second one is inherent while `dependencytracking` is a Hazelcast map, and it is accepted rather +than overlooked. The removal is what both the count's once-only property and a redelivery's +early return hang off, so no choice of marker closes the window: it closes when the row is deleted in +the same transaction as the count, which is where the map goes away. Accepting it buys the far worse +failure being gone. A count that is lost leaves a job visibly stuck and swept within the hour, while +a count taken twice reaches the total early and lets end-of-job work run on data that never arrived, +with nothing in the job state saying so. + `AdminBean.recheckBlocks` sweeps for both hourly, which bounds the damage to one sweep interval. It lifts the barrier of any job left holding one with no termination row, then opens any gate closed with no earlier unlifted barrier, requiring additionally for a termination chunk that its own job's @@ -302,10 +329,10 @@ Delivery order and `gate_open` are read from PostgreSQL, while `status` is read the table's copy of it is written write-behind and lags. The bulk sweep therefore takes an ordered candidate list from SQL and re-checks each candidate against the map before dispatching it. -The query cannot be a Hazelcast predicate, and the reason is the ownership split above: -`gate_open` and `is_termination` are columns on the table and deliberately not fields on -`DependencyTracking`, so no predicate can see them. Putting them on the map value to make one -possible is the very thing that would let the MapStore clobber them. +The query cannot be a Hazelcast predicate, and the reason is the ownership split above: `gate_open` +is a column on the table and deliberately not a field on `DependencyTracking`, so no predicate can +see it. Putting it on the map value to make one possible is what would give a chunk two answers to +whether it may be dispatched, one of them written behind the other's back. ## Delivery watermark @@ -377,6 +404,12 @@ Each entry holds: - **`matchKeys`** — string keys derived from sequence analysis data plus an optional barrier key; used to find chunks this one must sequence after - **`waitFor`** — indexed form of matchKeys as `WaitFor(sinkId, submitter, key)` tuples, used for Hazelcast predicate queries - **`waitingOn`** — set of `TrackingKey`s this chunk is currently blocked by +- **`termination`** - whether this chunk is its job's termination chunk, mirroring the + `is_termination` column. The column is the authority and this is read back from it whenever an + entry is loaded, which is sound because the value is decided when the row is created and never + changes. It is here so that the caller who removes an entry on delivery can tell the two branches + of the gate apart from the entry it was handed, see [Where the gate is decided](#where-the-gate-is-decided). + **`gate_open` has no counterpart here, and must not get one**, see below - **`priority`**, **`lastModified`**, **`retries`** ## Chunk lifecycle diff --git a/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/DependencyTracking.java b/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/DependencyTracking.java index 9f6ebdd197..adec01f7cd 100644 --- a/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/DependencyTracking.java +++ b/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/DependencyTracking.java @@ -36,6 +36,7 @@ public class DependencyTracking implements DependencyTrackingRO, Serializable, C private final int submitter; private Instant lastModified = Instant.now(); private int retries = 0; + private boolean termination = false; public DependencyTracking(TrackingKey key, int sinkId, int submitter, Set sequenceData) { this.key = key; @@ -63,6 +64,7 @@ public DependencyTracking(ResultSet rs) throws SQLException { submitter = rs.getInt("submitter"); lastModified = rs.getTimestamp("lastmodified").toInstant(); retries = rs.getInt("retries"); + termination = rs.getBoolean("is_termination"); waitFor = toWaitForIndexSet(sinkId, submitter, matchKeys); } @@ -121,6 +123,28 @@ public int getSubmitter() { return submitter; } + /** + * Says whether this chunk is its job's termination chunk. + *

+ * The {@code is_termination} column is the authority and this is a copy of it, which is sound + * because the value is decided when the row is created and never changes afterwards. It is set + * on the entry the termination chunk is scheduled with, and read back from the column whenever + * an entry is loaded from the table. + *

+ * {@code gate_open} deliberately has no counterpart here. It is written by four sites over a + * chunk's life, so a copy on this object could be stale, and a stale open gate dispatches a + * job's end-of-job work ahead of the data it summarises. + */ + @Override + public boolean isTermination() { + return termination; + } + + public DependencyTracking setTermination(boolean termination) { + this.termination = termination; + return this; + } + @Override public int getPriority() { return priority; diff --git a/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/DependencyTrackingRO.java b/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/DependencyTrackingRO.java index 72628cd73a..928c5531d2 100644 --- a/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/DependencyTrackingRO.java +++ b/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/DependencyTrackingRO.java @@ -18,6 +18,8 @@ public interface DependencyTrackingRO { int getSubmitter(); + boolean isTermination(); + int getPriority(); Instant getLastModified(); diff --git a/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/hz/store/DependencyTrackingStore.java b/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/hz/store/DependencyTrackingStore.java index 36e2ccafea..0de3a71929 100644 --- a/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/hz/store/DependencyTrackingStore.java +++ b/job-store-service/distributed-objects/src/main/java/dk/dbc/dataio/jobstore/distributed/hz/store/DependencyTrackingStore.java @@ -43,6 +43,11 @@ public class DependencyTrackingStore implements MapStore Adding them would reset every termination chunk's gate on its next status * transition, dispatching job-end work ahead of the data it summarises. *

+ * {@link DependencyTracking#isTermination()} carries {@code is_termination} on the value object + * and does not change that. The column is written where the termination row is created and read + * back by {@link DependencyTracking#DependencyTracking(java.sql.ResultSet)}, so this store has + * no reason to name it in either list. + *

* See docs/chunk-scheduling-redesign.md, "Who writes the gate columns before Phase 9", and * {@code JobGateRepository} in the war module. */ diff --git a/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/dependencytracking/DependencyTrackingService.java b/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/dependencytracking/DependencyTrackingService.java index 5ae74b2144..325ba1434f 100644 --- a/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/dependencytracking/DependencyTrackingService.java +++ b/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/dependencytracking/DependencyTrackingService.java @@ -235,15 +235,28 @@ private StatusChangeEvent setStatus(TrackingKey key, ChunkSchedulingStatus newSt return statusChangeEvent; } - public void remove(TrackingKey key) { + /** + * Removes a chunk's entry and hands the removed entry to the caller that removed it. + *

+ * The return value is the once-only token for this removal. {@code IMap.remove} is atomic per + * key across the cluster, so however many callers ask to remove one chunk, exactly one is given + * the entry and every other is given null. A caller that has to act once per chunk asks that + * question here rather than by reading the entry first, which two callers can both pass. + * {@code JobSchedulerBean.chunkDeliveringDone} counts a delivered data chunk on that basis. + * + * @param key chunk to remove + * @return the removed entry, or null if this call did not remove it + */ + public DependencyTracking remove(TrackingKey key) { DependencyTracking removed = dependencyTracker.remove(key); - if(removed == null) return; + if(removed == null) return null; countersMap.executeOnKey(removed.getSinkId(), new UpdateCounter(removed.getStatus(), -1)); if(enableWaitForTracking) { PredicateBuilder.EntryObject o = Predicates.newPredicateBuilder().getEntryObject(); lastTracker.removeAll(o.get("jobId").equal(key.getJobId()).and(o.get("chunkId").equal(key.getChunkId()))); } LOGGER.info("Removed tracking key {} from dependency tracker", key.toChunkIdentifier()); + return removed; } public void remove(Predicate predicate) { diff --git a/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobGateBean.java b/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobGateBean.java index 83966bc449..a3177b771b 100644 --- a/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobGateBean.java +++ b/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobGateBean.java @@ -2,6 +2,7 @@ import dk.dbc.dataio.commons.types.interceptor.Stopwatch; import dk.dbc.dataio.jobstore.distributed.ChunkSchedulingStatus; +import dk.dbc.dataio.jobstore.distributed.DependencyTracking; import dk.dbc.dataio.jobstore.distributed.TrackingKey; import jakarta.annotation.Resource; import jakarta.ejb.EJB; @@ -121,21 +122,21 @@ private JobGateBean self() { * transaction commits before JAX-RS writes the response, so no sink can acknowledge a message * whose increment has not committed. *

- * Not idempotent, and nothing here detects a repeat. The caller arrives after a {@code get}, a - * status check and a {@code remove} on the chunk's dependency tracking entry, and those three - * are not atomic, so two concurrent acknowledgements of one chunk can both pass them and both - * be counted. + * Counts once per chunk, because it is reached only by the caller that removed the chunk's + * dependency tracking entry. That removal is atomic per key, so of two concurrent + * acknowledgements of one chunk only one arrives here and the other is told it did not remove + * anything. Every fact this method works from comes off the removed entry, including whether the + * chunk was its job's termination chunk. * - * @param key delivered chunk - * @param sinkId sink the chunk was delivered to - * @param submitter submitter the chunk's job belongs to + * @param removed the delivered chunk's removed dependency tracking entry */ @Stopwatch - public void advanceGateState(TrackingKey key, int sinkId, int submitter) { - if (jobGateRepository.isTerminationChunk(key)) { - liftBarrierAndRetrigger(key.getJobId(), sinkId, submitter); + public void advanceGateState(DependencyTracking removed) { + int jobId = removed.getKey().getJobId(); + if (removed.isTermination()) { + liftBarrierAndRetrigger(jobId, removed.getSinkId(), removed.getSubmitter()); } else { - countDataChunk(key.getJobId(), sinkId, submitter); + countDataChunk(jobId, removed.getSinkId(), removed.getSubmitter()); } } diff --git a/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobGateRepository.java b/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobGateRepository.java index 4d74af6099..12405689b7 100644 --- a/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobGateRepository.java +++ b/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobGateRepository.java @@ -172,22 +172,6 @@ public void upsertGateRow(TrackingKey key, int sinkId, int submitter, ChunkSched .executeUpdate(); } - /** - * @param key tracking key to ask about - * @return true if the chunk is its job's termination chunk - *

- * A missing row answers false, which is correct either way round: a chunk with no row is not a - * termination chunk, and a data chunk has to be counted whether or not its row is there. - */ - public boolean isTerminationChunk(TrackingKey key) { - return !entityManager.createNativeQuery( - "SELECT 1 FROM dependencytracking WHERE jobid = ?1 AND chunkid = ?2 AND is_termination") - .setParameter(1, key.getJobId()) - .setParameter(2, key.getChunkId()) - .getResultList() - .isEmpty(); - } - /** * @param sinkId sink the job delivers to * @param submitter the job's submitter @@ -299,8 +283,8 @@ public List laterClosedGates(int sinkId, int submitter, int jobId) *

* Leaving the width out buys two things. The set is read in one place, the scheduler, where the * job and its cached sink are already in hand, rather than at every call site that lifts a - * barrier, one of which holds only a {@code DependencyTrackingRO} and would need an extra job - * load per delivered termination chunk. And a row stays reopenable after the width that closed + * barrier, one of which holds only the removed dependency tracking entry and would need an extra + * job load per delivered termination chunk. And a row stays reopenable after the width that closed * it has gone: one closed by an earlier deployment, or by a sink whose type has since changed, * would otherwise be stranded closed with only the sweep to rescue it. *

diff --git a/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobSchedulerBean.java b/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobSchedulerBean.java index 9959cb6c43..0862a72b23 100644 --- a/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobSchedulerBean.java +++ b/job-store-service/war/src/main/java/dk/dbc/dataio/jobstore/service/ejb/JobSchedulerBean.java @@ -378,8 +378,12 @@ void createAndScheduleTerminationChunk(JobEntity jobEntity, Sink sink, int chunk // to PostgreSQL with a closed gate in the same transaction that writes the counters, ahead // of the map add below. A termination chunk carries no sequence analysis data, so its match // keys are the barrier key alone. + // The flag is set here rather than being read back from is_termination, so that the entry + // handed to whoever removes it on delivery answers for itself whether it was the job's + // termination chunk. createJobTerminationChunkEntity writes the column below. DependencyTracking endTracker = new DependencyTracking(key, sinkId, (int)jobEntity.getSpecification().getSubmitterId(), barrierMatchKey, Set.of()) - .setPriority(Priority.HIGH.getValue()); + .setPriority(Priority.HIGH.getValue()) + .setTermination(true); // chunkId is numberOfChunks as read in markJobAsPartitioned before this call, which is // exactly the job's data-chunk count. Passing it rather than re-reading it downstream is // what keeps data_chunks_expected reachable: createJobTerminationChunkEntity increments @@ -444,11 +448,27 @@ public void chunkDeliveringDone(Chunk chunk) { long startTime = System.currentTimeMillis(); int chunkDoneSinkId = chunkDone.getSinkId(); - dependencyTrackingService.remove(chunkDoneKey); + DependencyTracking removed = dependencyTrackingService.remove(chunkDoneKey); // Per-job gate: counts this delivery against the job's own gate, or lifts the job's // barrier and re-evaluates later jobs if the chunk was its termination chunk. - jobGateBean.advanceGateState(chunkDoneKey, chunkDoneSinkId, chunkDone.getSubmitter()); + // + // Only the caller that removed the entry gets to do that. The get, the status check and the + // remove above are three separate map operations, so two concurrent acknowledgements of one + // chunk can both reach this point, and a counter cannot survive being told twice: each call + // adds exactly 1, so the count still lands on data_chunks_expected, only too early, and the + // gate opens with data chunks still in flight. The removal is atomic per key, which is what + // makes it the once-only marker. + // + // The work below stays unconditional. removeFromWaitingOn reports only the entries it + // actually changed, so the caller that lost the removal finds nothing left to unblock and + // its dispatch loop does not run. + if (removed != null) { + jobGateBean.advanceGateState(removed); + } else { + LOGGER.info("chunkDeliveringDone: chunk {}/{} was removed by a concurrent call, so this one does not count it", + chunk.getJobId(), chunk.getChunkId()); + } StopWatch findChunksWaitingForMeStopWatch = new StopWatch(); Set unblocked = dependencyTrackingService.removeFromWaitingOn(chunkDoneKey); diff --git a/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobGateBeanTest.java b/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobGateBeanTest.java index 498e7995c4..34fd5abc74 100644 --- a/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobGateBeanTest.java +++ b/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobGateBeanTest.java @@ -1,6 +1,7 @@ package dk.dbc.dataio.jobstore.service.ejb; import dk.dbc.dataio.jobstore.distributed.ChunkSchedulingStatus; +import dk.dbc.dataio.jobstore.distributed.DependencyTracking; import dk.dbc.dataio.jobstore.distributed.TrackingKey; import org.junit.jupiter.api.Test; import org.mockito.InOrder; @@ -32,10 +33,9 @@ class JobGateBeanTest { @Test void advanceGateState_dataChunk_isCounted() { TrackingKey dataChunk = new TrackingKey(JOB_ID, 0); - when(jobGateRepository.isTerminationChunk(dataChunk)).thenReturn(false); when(jobGateRepository.dataChunksAccountedFor(JOB_ID)).thenReturn(false); - jobGateBean.advanceGateState(dataChunk, SINK_ID, SUBMITTER); + jobGateBean.advanceGateState(dataChunkEntry(dataChunk)); verify(jobGateRepository).incrementDataChunksDelivered(JOB_ID); verify(jobGateRepository, never()).markTerminationBarrierLifted(anyInt()); @@ -44,11 +44,10 @@ void advanceGateState_dataChunk_isCounted() { @Test void advanceGateState_terminationChunk_isNotCounted() { TrackingKey terminationChunk = new TrackingKey(JOB_ID, TERMINATION_CHUNK_ID); - when(jobGateRepository.isTerminationChunk(terminationChunk)).thenReturn(true); when(jobGateRepository.markTerminationBarrierLifted(JOB_ID)).thenReturn(1); when(jobGateRepository.laterClosedGates(SINK_ID, SUBMITTER, JOB_ID)).thenReturn(List.of()); - jobGateBean.advanceGateState(terminationChunk, SINK_ID, SUBMITTER); + jobGateBean.advanceGateState(terminationChunkEntry(terminationChunk)); verify(jobGateRepository, never()).incrementDataChunksDelivered(anyInt()); verify(jobGateRepository).markTerminationBarrierLifted(JOB_ID); @@ -57,11 +56,10 @@ void advanceGateState_terminationChunk_isNotCounted() { @Test void advanceGateState_jobWithoutTerminationChunk_countsButDoesNotEvaluate() { TrackingKey dataChunk = new TrackingKey(JOB_ID, 0); - when(jobGateRepository.isTerminationChunk(dataChunk)).thenReturn(false); when(jobGateRepository.dataChunksAccountedFor(JOB_ID)).thenReturn(true); when(jobGateRepository.closedTerminationChunkId(SINK_ID, SUBMITTER, JOB_ID)).thenReturn(OptionalInt.empty()); - jobGateBean.advanceGateState(dataChunk, SINK_ID, SUBMITTER); + jobGateBean.advanceGateState(dataChunkEntry(dataChunk)); verify(jobGateRepository).incrementDataChunksDelivered(JOB_ID); verify(jobGateRepository, never()).advisoryLock(anyInt(), anyInt()); @@ -71,13 +69,12 @@ void advanceGateState_jobWithoutTerminationChunk_countsButDoesNotEvaluate() { @Test void advanceGateState_lastDataChunk_opensGate() { TrackingKey dataChunk = new TrackingKey(JOB_ID, TERMINATION_CHUNK_ID - 1); - when(jobGateRepository.isTerminationChunk(dataChunk)).thenReturn(false); when(jobGateRepository.dataChunksAccountedFor(JOB_ID)).thenReturn(true); when(jobGateRepository.closedTerminationChunkId(SINK_ID, SUBMITTER, JOB_ID)) .thenReturn(OptionalInt.of(TERMINATION_CHUNK_ID)); when(jobGateRepository.hasEarlierUndeliveredTermination(SINK_ID, SUBMITTER, JOB_ID)).thenReturn(false); - jobGateBean.advanceGateState(dataChunk, SINK_ID, SUBMITTER); + jobGateBean.advanceGateState(dataChunkEntry(dataChunk)); verify(jobGateRepository).advisoryLock(SINK_ID, SUBMITTER); verify(jobGateRepository).openGate(new TrackingKey(JOB_ID, TERMINATION_CHUNK_ID)); @@ -86,13 +83,12 @@ void advanceGateState_lastDataChunk_opensGate() { @Test void advanceGateState_lastDataChunkButEarlierBarrierHolds_gateStaysClosed() { TrackingKey dataChunk = new TrackingKey(JOB_ID, TERMINATION_CHUNK_ID - 1); - when(jobGateRepository.isTerminationChunk(dataChunk)).thenReturn(false); when(jobGateRepository.dataChunksAccountedFor(JOB_ID)).thenReturn(true); when(jobGateRepository.closedTerminationChunkId(SINK_ID, SUBMITTER, JOB_ID)) .thenReturn(OptionalInt.of(TERMINATION_CHUNK_ID)); when(jobGateRepository.hasEarlierUndeliveredTermination(SINK_ID, SUBMITTER, JOB_ID)).thenReturn(true); - jobGateBean.advanceGateState(dataChunk, SINK_ID, SUBMITTER); + jobGateBean.advanceGateState(dataChunkEntry(dataChunk)); verify(jobGateRepository).advisoryLock(SINK_ID, SUBMITTER); verify(jobGateRepository, never()).openGate(new TrackingKey(JOB_ID, TERMINATION_CHUNK_ID)); @@ -102,14 +98,13 @@ void advanceGateState_lastDataChunkButEarlierBarrierHolds_gateStaysClosed() { void advanceGateState_terminationChunk_reTriggersLaterJobs() { TrackingKey terminationChunk = new TrackingKey(JOB_ID, TERMINATION_CHUNK_ID); TrackingKey laterJobTermination = new TrackingKey(JOB_ID + 1, 5); - when(jobGateRepository.isTerminationChunk(terminationChunk)).thenReturn(true); when(jobGateRepository.markTerminationBarrierLifted(JOB_ID)).thenReturn(1); when(jobGateRepository.laterClosedGates(SINK_ID, SUBMITTER, JOB_ID)) .thenReturn(List.of(laterJobTermination)); when(jobGateRepository.dataChunksAccountedFor(JOB_ID + 1)).thenReturn(true); when(jobGateRepository.hasEarlierUndeliveredTermination(SINK_ID, SUBMITTER, JOB_ID + 1)).thenReturn(false); - jobGateBean.advanceGateState(terminationChunk, SINK_ID, SUBMITTER); + jobGateBean.advanceGateState(terminationChunkEntry(terminationChunk)); verify(jobGateRepository).advisoryLock(SINK_ID, SUBMITTER); verify(jobGateRepository).markTerminationBarrierLifted(JOB_ID); @@ -120,13 +115,12 @@ void advanceGateState_terminationChunk_reTriggersLaterJobs() { void advanceGateState_terminationChunk_laterJobWithIncompleteCounterStaysClosed() { TrackingKey terminationChunk = new TrackingKey(JOB_ID, TERMINATION_CHUNK_ID); TrackingKey laterJobTermination = new TrackingKey(JOB_ID + 1, 5); - when(jobGateRepository.isTerminationChunk(terminationChunk)).thenReturn(true); when(jobGateRepository.markTerminationBarrierLifted(JOB_ID)).thenReturn(1); when(jobGateRepository.laterClosedGates(SINK_ID, SUBMITTER, JOB_ID)) .thenReturn(List.of(laterJobTermination)); when(jobGateRepository.dataChunksAccountedFor(JOB_ID + 1)).thenReturn(false); - jobGateBean.advanceGateState(terminationChunk, SINK_ID, SUBMITTER); + jobGateBean.advanceGateState(terminationChunkEntry(terminationChunk)); verify(jobGateRepository, never()).openGate(laterJobTermination); } @@ -253,4 +247,29 @@ void sweepUnliftedBarriers_liftsAndReTriggers() { verify(jobGateRepository).markTerminationBarrierLifted(JOB_ID); verify(jobGateRepository).openLaterDataChunkGates(SINK_ID, SUBMITTER, JOB_ID); } + + /** + * A termination chunk delivered again, after its barrier has already been lifted, still takes + * the termination branch and is still not counted. What stops it doing the work twice is the + * guarded update reporting no rows, not the branch declining to run. + */ + @Test + void advanceGateState_terminationChunkAgain_isStillNotCounted() { + TrackingKey terminationChunk = new TrackingKey(JOB_ID, TERMINATION_CHUNK_ID); + when(jobGateRepository.markTerminationBarrierLifted(JOB_ID)).thenReturn(0); + + jobGateBean.advanceGateState(terminationChunkEntry(terminationChunk)); + + verify(jobGateRepository, never()).incrementDataChunksDelivered(anyInt()); + verify(jobGateRepository, never()).advisoryLock(anyInt(), anyInt()); + verify(jobGateRepository, never()).openLaterDataChunkGates(anyInt(), anyInt(), anyInt()); + } + + private DependencyTracking dataChunkEntry(TrackingKey key) { + return new DependencyTracking(key, SINK_ID, SUBMITTER); + } + + private DependencyTracking terminationChunkEntry(TrackingKey key) { + return dataChunkEntry(key).setTermination(true); + } } diff --git a/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobGateIT.java b/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobGateIT.java index 0947bb1c08..dd3bcaabad 100644 --- a/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobGateIT.java +++ b/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobGateIT.java @@ -23,6 +23,7 @@ import java.sql.SQLException; import java.util.Set; import java.util.concurrent.Callable; +import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; @@ -41,6 +42,10 @@ * The assertions read the gate columns straight from PostgreSQL on their own connections, so they * see committed state and nothing else. What a closed gate withholds from dispatch is the dispatch * filter's contract and is asserted with it. + *

+ * The counter's two concurrency hazards are both here, since they are one property from opposite + * sides: an acknowledgement must not be lost, and an acknowledgement must not be counted twice. See + * "Same-item concurrent redelivery (job-store-service)" for where the duplicate comes from. */ public class JobGateIT extends AbstractJobStoreIT { private static final int SINK_ID = 4711; @@ -96,7 +101,7 @@ public void jobWithTerminationChunk() throws Exception { dataChunksDelivered(job.getId()), is(3)); // The termination chunk does not count itself. - deliverChunk(job.getId(), 3); + deliverTerminationChunk(job.getId(), 3); assertThat("termination chunk was not counted", dataChunksDelivered(job.getId()), is(3)); assertThat("barrier lifted on delivery", terminationBarrierLifted(job.getId()), is(true)); } @@ -198,11 +203,11 @@ public void reTriggerOpensTheNextJobOnly() throws Exception { assertThat("B closed", gateOpen(new TrackingKey(jobB.getId(), 1)), is(false)); assertThat("C closed", gateOpen(new TrackingKey(jobC.getId(), 1)), is(false)); - deliverChunk(jobA.getId(), 1); + deliverTerminationChunk(jobA.getId(), 1); assertThat("B opened by the re-trigger", gateOpen(new TrackingKey(jobB.getId(), 1)), is(true)); assertThat("C still behind B", gateOpen(new TrackingKey(jobC.getId(), 1)), is(false)); - deliverChunk(jobB.getId(), 1); + deliverTerminationChunk(jobB.getId(), 1); assertThat("C opened in turn", gateOpen(new TrackingKey(jobC.getId(), 1)), is(true)); } @@ -262,7 +267,7 @@ public void concurrentTerminationAndLastDataChunk_doesNotDeadlock() throws Excep JobGateBean terminationGate = new JobGateBean(new JobGateRepository().withEntityManager(terminationEm)); Future terminationDelivery = executor.submit(() -> runInTransaction(terminationEm, () -> { - terminationGate.advanceGateState(new TrackingKey(job.getId(), 2), SINK_ID, (int) SUBMITTER); + terminationGate.advanceGateState(terminationEntry(job.getId(), 2)); return null; })); @@ -314,7 +319,7 @@ public void terminationChunkInsertAgainstAnInFlightDelivery_gateEndsUpOpen() thr try { JobGateBean deliveryGate = new JobGateBean(new JobGateRepository().withEntityManager(deliveryEm)); deliveryEm.getTransaction().begin(); - deliveryGate.advanceGateState(new TrackingKey(job.getId(), 0), SINK_ID, (int) SUBMITTER); + deliveryGate.advanceGateState(dataEntry(job.getId(), 0)); Future partitioningDone = executor.submit(() -> { markJobAsPartitioned(job, partitioningEm); @@ -380,7 +385,7 @@ public void terminationInsertAgainstTheReTriggerOfAnEarlierJob_gateEndsUpOpen() }); JobGateBean reTriggerGate = new JobGateBean(new JobGateRepository().withEntityManager(reTriggerEm)); Future reTrigger = executor.submit(() -> runInTransaction(reTriggerEm, () -> { - reTriggerGate.advanceGateState(new TrackingKey(earlier.getId(), 1), SINK_ID, (int) SUBMITTER); + reTriggerGate.advanceGateState(terminationEntry(earlier.getId(), 1)); return null; })); @@ -455,7 +460,7 @@ public void dataChunkCloseAgainstTheReTriggerOfAnEarlierJob_gateEndsUpOpen() thr })); JobGateBean reTriggerGate = new JobGateBean(new JobGateRepository().withEntityManager(reTriggerEm)); Future reTrigger = executor.submit(() -> runInTransaction(reTriggerEm, () -> { - reTriggerGate.advanceGateState(new TrackingKey(earlier.getId(), 1), SINK_ID, (int) SUBMITTER); + reTriggerGate.advanceGateState(terminationEntry(earlier.getId(), 1)); return null; })); @@ -516,7 +521,7 @@ public void concurrentLastTwoDataChunks_counterLandsExactlyOnExpected() throws E JobGateBean secondDelivery = new JobGateBean(new JobGateRepository().withEntityManager(secondEm)); Future second = executor.submit(() -> runInTransaction(secondEm, () -> { - secondDelivery.advanceGateState(new TrackingKey(job.getId(), 1), SINK_ID, (int) SUBMITTER); + secondDelivery.advanceGateState(dataEntry(job.getId(), 1)); return null; })); @@ -539,6 +544,136 @@ public void concurrentLastTwoDataChunks_counterLandsExactlyOnExpected() throws E assertThat("gate opened by the delivery that unblocked", gateOpen(new TrackingKey(job.getId(), 2)), is(true)); } + /** + * The counter's other concurrency hazard, and the opposite one to the test above. Two + * acknowledgements of the *same* chunk must add 1 between them, where two acknowledgements of + * two different chunks must add 2. + *

+ * Counted twice, data_chunks_delivered still lands on data_chunks_expected exactly, only while a + * data chunk is still in flight. So the gate opens, the termination chunk is dispatched ahead of + * the data it summarises, and the job reads as completed. The failure is silent, which is what + * makes it worse than the stall a lost update causes. + *

+ * These four tests drive chunkDeliveringDone rather than advanceGateState, because the removal + * that makes the count once-only and the count itself are on opposite sides of that boundary. + */ + @org.junit.Test + public void repeatedAcknowledgementOfOneChunkCountsOnce() throws Exception { + JobEntity job = newPersistedTerminationJob(SUBMITTER, 2); + markJobAsPartitioned(job); + scheduleDataChunkForDelivery(job.getId(), 0); + + acknowledgeDelivery(job.getId(), 0); + acknowledgeDelivery(job.getId(), 0); + + assertThat("counted once", dataChunksDelivered(job.getId()), is(1)); + assertThat("gate closed while chunk 1 is outstanding", + gateOpen(new TrackingKey(job.getId(), 2)), is(false)); + } + + /** + * The same property under two genuinely concurrent calls, which is the case the removal token is + * for. A repeat arriving after the first call committed is turned away by the entry already + * being gone, but two callers that both read the entry before either removes it both get past + * that. + *

+ * The interleaving is arranged rather than hoped for, as in the test above and for the same + * reason. Two threads released together would often run one call to completion before the other + * started, and against a completed call nothing can be double counted however it is + * implemented. Here the serialisation point is a map operation rather than a row lock, so + * holding a transaction open does not reach it. The two callers meet inside remove instead, + * which puts both past their read of the entry before either removal takes effect. + */ + @org.junit.Test + public void concurrentAcknowledgementOfOneChunkCountsOnce() throws Exception { + JobEntity job = newPersistedTerminationJob(SUBMITTER, 2); + markJobAsPartitioned(job); + scheduleDataChunkForDelivery(job.getId(), 0); + + RendezvousInRemove trackingService = new RendezvousInRemove(); + EntityManager firstEm = entityManager.getEntityManagerFactory().createEntityManager(); + EntityManager secondEm = entityManager.getEntityManagerFactory().createEntityManager(); + ExecutorService executor = Executors.newFixedThreadPool(2); + try { + Future first = executor.submit(() -> runInTransaction(firstEm, + acknowledge(trackingService, firstEm, job.getId(), 0))); + Future second = executor.submit(() -> runInTransaction(secondEm, + acknowledge(trackingService, secondEm, job.getId(), 0))); + first.get(60, TimeUnit.SECONDS); + second.get(60, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + executor.awaitTermination(30, TimeUnit.SECONDS); + rollbackAndClose(firstEm); + rollbackAndClose(secondEm); + } + + assertThat("both callers met inside remove", trackingService.met, is(true)); + assertThat("counted once", dataChunksDelivered(job.getId()), is(1)); + assertThat("gate closed while chunk 1 is outstanding", + gateOpen(new TrackingKey(job.getId(), 2)), is(false)); + } + + /** + * What the count is for. A duplicated acknowledgement leaves the job still needing its remaining + * data chunk, so the gate opens on that chunk and not on the duplicate. + */ + @org.junit.Test + public void gateOpensOnlyAfterEveryDataChunkIsCounted() throws Exception { + JobEntity job = newPersistedTerminationJob(SUBMITTER, 2); + markJobAsPartitioned(job); + TrackingKey terminationChunk = new TrackingKey(job.getId(), 2); + + scheduleDataChunkForDelivery(job.getId(), 0); + acknowledgeDelivery(job.getId(), 0); + acknowledgeDelivery(job.getId(), 0); + assertThat("gate closed after the duplicate", gateOpen(terminationChunk), is(false)); + + scheduleDataChunkForDelivery(job.getId(), 1); + acknowledgeDelivery(job.getId(), 1); + + assertThat("every data chunk counted", dataChunksDelivered(job.getId()), is(2)); + assertThat("gate open once the job's data is delivered", gateOpen(terminationChunk), is(true)); + } + + /** + * The token's other branch. A termination chunk is not counted at all, and acknowledging it + * twice lifts its barrier once. + */ + @org.junit.Test + public void repeatedAcknowledgementOfTheTerminationChunkLiftsTheBarrierOnce() throws Exception { + JobEntity job = newPersistedTerminationJob(SUBMITTER, 2); + markJobAsPartitioned(job); + scheduleForDelivery(new TrackingKey(job.getId(), 2)); + + acknowledgeDelivery(job.getId(), 2); + acknowledgeDelivery(job.getId(), 2); + + assertThat("the termination chunk counts nothing", dataChunksDelivered(job.getId()), is(0)); + assertThat("barrier lifted", terminationBarrierLifted(job.getId()), is(true)); + } + + /** + * Both callers wait here until the other has arrived, so each is past its own read of the entry + * before either removal takes effect. The timeout is what turns a caller that never arrives into + * a failed test rather than a hung one. + */ + private static class RendezvousInRemove extends DependencyTrackingService { + private final CyclicBarrier barrier = new CyclicBarrier(2); + private volatile boolean met = false; + + @Override + public DependencyTracking remove(TrackingKey key) { + try { + barrier.await(60, TimeUnit.SECONDS); + met = true; + } catch (Exception e) { + throw new IllegalStateException("the other caller never reached remove", e); + } + return super.remove(key); + } + } + private void awaitAdvisoryLockWaiters(int expected) throws Exception { awaitLockWaiters("backends waiting on the barrier scope", expected, "advisory"); } @@ -597,6 +732,17 @@ private T runInTransaction(EntityManager em, Callable callable) throws Ex } } + /** + * Rolls back before closing, since a transaction left active returns its connection to the pool + * still holding its row locks, which hangs the next test's cleanup rather than failing this one. + */ + private void rollbackAndClose(EntityManager em) { + if (em.getTransaction().isActive()) { + em.getTransaction().rollback(); + } + em.close(); + } + private void markJobAsPartitioned(JobEntity job) throws JobStoreException { markJobAsPartitioned(job, entityManager); } @@ -607,19 +753,85 @@ private void markJobAsPartitioned(JobEntity job) throws JobStoreException { * manager, safe only while the other thread happens to be doing nothing. */ private void markJobAsPartitioned(JobEntity job, EntityManager em) throws JobStoreException { - JobSchedulerBean jobSchedulerBean = new JobSchedulerBean(em, - mock(JobSchedulerTransactionsBean.class), newPgJobStoreRepository(em), null, - new DependencyTrackingService().init(), newJobGateBean(em), newDeliveryDispatchRepository(em)); - jobSchedulerBean.markJobAsPartitioned(job); + newJobSchedulerBean(newDependencyTrackingService(), em).markJobAsPartitioned(job); + } + + /** + * Acknowledges a chunk's delivery the way a sink's callback does, through + * {@code chunkDeliveringDone} rather than by calling the gate directly. + */ + private void acknowledgeDelivery(int jobId, int chunkId) { + persistenceContext.run(() -> acknowledge(newDependencyTrackingService(), entityManager, jobId, chunkId).call()); + } + + private Callable acknowledge(DependencyTrackingService trackingService, EntityManager em, + int jobId, int chunkId) { + return () -> { + newJobSchedulerBean(trackingService, em) + .chunkDeliveringDone(new Chunk(jobId, chunkId, Chunk.Type.DELIVERED)); + return null; + }; + } + + /** + * Puts a data chunk into dependency tracking in the state an acknowledgement expects to find it + * in. Partitioning is what schedules these in production, and these tests do not run it. + */ + private void scheduleDataChunkForDelivery(int jobId, int chunkId) { + newDependencyTrackingService().add(new DependencyTracking(new TrackingKey(jobId, chunkId), + SINK_ID, (int) submitterOf(jobId)).setStatus(ChunkSchedulingStatus.QUEUED_FOR_DELIVERY)); + } + + /** + * The same, for the termination chunk that {@code markJobAsPartitioned} has already added. + */ + private void scheduleForDelivery(TrackingKey key) { + newDependencyTrackingService().setStatus(key, ChunkSchedulingStatus.QUEUED_FOR_DELIVERY); + } + + private JobSchedulerBean newJobSchedulerBean(DependencyTrackingService trackingService, EntityManager em) { + return new JobSchedulerBean(em, mock(JobSchedulerTransactionsBean.class), + newPgJobStoreRepository(em), null, trackingService, newJobGateBean(em), + newDeliveryDispatchRepository(em)); + } + + /** + * Constructed per call rather than held in a field, since the service resolves its Hazelcast + * maps in its own initialisers and this class's instance exists before Hazelcast is started. + */ + private DependencyTrackingService newDependencyTrackingService() { + return new DependencyTrackingService().init(); } private void deliverDataChunk(int jobId, int chunkId) { - deliverChunk(jobId, chunkId); + deliverChunk(jobId, chunkId, false); + } + + private void deliverTerminationChunk(int jobId, int chunkId) { + deliverChunk(jobId, chunkId, true); + } + + private void deliverChunk(int jobId, int chunkId, boolean termination) { + int submitter = (int) submitterOf(jobId); + persistenceContext.run(() -> newJobGateBean().advanceGateState( + entry(new TrackingKey(jobId, chunkId), submitter, termination))); + } + + /** + * The entry a delivery hands the gate, standing in for the one chunkDeliveringDone removes. + * Whether the chunk is its job's termination chunk is a fact about the entry rather than + * something the gate looks up, so a test that delivers one says which it is delivering. + */ + private DependencyTracking dataEntry(int jobId, int chunkId) { + return entry(new TrackingKey(jobId, chunkId), (int) SUBMITTER, false); + } + + private DependencyTracking terminationEntry(int jobId, int chunkId) { + return entry(new TrackingKey(jobId, chunkId), (int) SUBMITTER, true); } - private void deliverChunk(int jobId, int chunkId) { - persistenceContext.run(() -> newJobGateBean() - .advanceGateState(new TrackingKey(jobId, chunkId), SINK_ID, (int) submitterOf(jobId))); + private DependencyTracking entry(TrackingKey key, int submitter, boolean termination) { + return new DependencyTracking(key, SINK_ID, submitter).setTermination(termination); } private long submitterOf(int jobId) { diff --git a/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobSchedulerBeanTest.java b/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobSchedulerBeanTest.java index 1a7ef94aa5..0ff178e491 100644 --- a/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobSchedulerBeanTest.java +++ b/job-store-service/war/src/test/java/dk/dbc/dataio/jobstore/service/ejb/JobSchedulerBeanTest.java @@ -1,19 +1,32 @@ package dk.dbc.dataio.jobstore.service.ejb; +import dk.dbc.dataio.commons.types.Chunk; import dk.dbc.dataio.commons.types.SinkContent; +import dk.dbc.dataio.jobstore.distributed.ChunkSchedulingStatus; +import dk.dbc.dataio.jobstore.distributed.DependencyTracking; +import dk.dbc.dataio.jobstore.distributed.TrackingKey; +import dk.dbc.dataio.jobstore.service.dependencytracking.DependencyTrackingService; import org.junit.jupiter.api.Test; import java.util.Arrays; import java.util.List; +import java.util.Set; import static org.hamcrest.CoreMatchers.is; import static org.hamcrest.MatcherAssert.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; /** * The two sink type sets that decide barrier width, see docs/chunk-scheduling-redesign.md, - * "Barrier Width - Per-Sink-Type Job Isolation". + * "Barrier Width - Per-Sink-Type Job Isolation", and the once-only marker on a delivered chunk. */ class JobSchedulerBeanTest { + private static final int JOB_ID = 42; + private static final int CHUNK_ID = 0; + private static final int SINK_ID = 7; + private static final int SUBMITTER = 424242; /** * Holding a later job's data chunks behind a barrier is meaningless for a sink type that raises @@ -54,4 +67,29 @@ void terminationChunkSinkTypesAreUnchanged() { assertThat(withTermination, is(List.of(SinkContent.SinkType.MARCCONV, SinkContent.SinkType.PERIODIC_JOBS, SinkContent.SinkType.TICKLE))); } + + /** + * A delivery that did not remove the chunk's tracking entry does not reach the gate at all. + *

+ * Every caller of chunkDeliveringDone sees the entry before removing it, so a concurrent pair + * can both find it and both ask to remove it. Only one removal takes effect, and the count of + * the job's delivered data chunks belongs to that one: counted twice, it reaches + * data_chunks_expected while a data chunk is still in flight, and the gate opens early. + */ + @Test + void chunkDeliveringDone_removalLost_doesNotAdvanceTheGate() { + TrackingKey key = new TrackingKey(JOB_ID, CHUNK_ID); + DependencyTrackingService dependencyTrackingService = mock(DependencyTrackingService.class); + JobGateBean jobGateBean = mock(JobGateBean.class); + when(dependencyTrackingService.get(key)).thenReturn(new DependencyTracking(key, SINK_ID, SUBMITTER) + .setStatus(ChunkSchedulingStatus.QUEUED_FOR_DELIVERY)); + when(dependencyTrackingService.remove(key)).thenReturn(null); + when(dependencyTrackingService.removeFromWaitingOn(key)).thenReturn(Set.of()); + JobSchedulerBean jobSchedulerBean = new JobSchedulerBean(null, null, null, null, + dependencyTrackingService, jobGateBean, null); + + jobSchedulerBean.chunkDeliveringDone(new Chunk(JOB_ID, CHUNK_ID, Chunk.Type.DELIVERED)); + + verifyNoInteractions(jobGateBean); + } }