From d63d3febd7da4f415c3937b461712017006fbe06 Mon Sep 17 00:00:00 2001 From: Clemens Portele Date: Tue, 18 Aug 2026 12:09:52 +0200 Subject: [PATCH 1/2] features: job hooks for progress and cooperative cancellation Queries can carry a JobHook that execution code reports progress to and polls for cooperative cancellation; the default NOOP hook leaves the pipeline unchanged. - JobHook (init/update/isCancelRequested/checkpoint) and JobCancelledException in the features domain - Query.getJobHook(), default NOOP, auxiliary - FeatureTokenTransformerProgress: init from numberReturned when available, one update per feature, per-feature cancellation checkpoint; inserted into the pipeline only when a hook is attached - SqlQueryBatch / FeatureQueryEncoderSql: the hook travels with the encoded query batch - SqlConnector: unpaged single-shot queries report one unit per completed sub-query (weak counter, no meta phase) - specs: JobHookSpec, FeatureTokenTransformerProgressSpec --- .../sql/app/FeatureQueryEncoderSql.java | 2 + .../features/sql/domain/SqlConnector.java | 13 +- .../features/sql/domain/SqlQueryBatch.java | 5 + .../features/domain/FeatureStreamImpl.java | 8 ++ .../FeatureTokenTransformerProgress.java | 43 +++++++ .../domain/JobCancelledException.java | 20 +++ .../xtraplatform/features/domain/JobHook.java | 36 ++++++ .../xtraplatform/features/domain/Query.java | 5 + ...FeatureTokenTransformerProgressSpec.groovy | 115 ++++++++++++++++++ .../features/domain/JobHookSpec.groovy | 69 +++++++++++ 10 files changed, 315 insertions(+), 1 deletion(-) create mode 100644 xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureTokenTransformerProgress.java create mode 100644 xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobCancelledException.java create mode 100644 xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobHook.java create mode 100644 xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/FeatureTokenTransformerProgressSpec.groovy create mode 100644 xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/JobHookSpec.groovy diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureQueryEncoderSql.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureQueryEncoderSql.java index 56f1f2843..6873de2ff 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureQueryEncoderSql.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureQueryEncoderSql.java @@ -124,6 +124,7 @@ private SqlQueryBatch encode(FeatureQuery query, Map additionalQ // reuse SqlQuerySet instances instead of copying them; this is expensive and unnecessary, since // they are immutable return new ImmutableSqlQueryBatch.Builder() + .jobHook(query.getJobHook()) .limit(query.getLimit()) .offset(query.getOffset()) .chunkSize(chunkSize) @@ -175,6 +176,7 @@ private SqlQueryBatch encode( // reuse SqlQuerySet instances instead of copying them; this is expensive and unnecessary, since // they are immutable return new ImmutableSqlQueryBatch.Builder() + .jobHook(query.getJobHook()) .limit(query.getLimit()) .offset(query.getOffset()) .chunkSize(chunkSize) diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java index cf807662b..9f396a118 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlConnector.java @@ -11,6 +11,7 @@ import de.ii.xtraplatform.base.domain.resiliency.Volatile2; import de.ii.xtraplatform.base.domain.util.Tuple; import de.ii.xtraplatform.features.domain.FeatureProviderConnector; +import de.ii.xtraplatform.features.domain.JobHook; import de.ii.xtraplatform.features.sql.domain.ImmutableSqlRowMeta.Builder; import de.ii.xtraplatform.streams.domain.Reactive; import de.ii.xtraplatform.streams.domain.Reactive.Source; @@ -220,6 +221,13 @@ default Reactive.Source getSourceStream( if (queryBatch.isSingleFeature() || queryBatch.isUnpaged()) { boolean unpaged = queryBatch.isUnpaged(); List querySets = queryBatch.getQuerySets(); + Optional jobHook = queryBatch.getJobHook(); + // unpaged queries have no meta phase and therefore no numberReturned to + // count features against; report sub-queries instead (weak progress, + // completion order may differ from emission order) + if (unpaged) { + jobHook.ifPresent(hook -> hook.init(querySets.size())); + } // a single-shot query computes neither count; -1 marks both as absent (the // decoder leaves numberReturned unset and numberMatched stays empty) // instead of reporting 0 @@ -277,7 +285,10 @@ default Reactive.Source getSourceStream( int tables = querySets.get(index).getTableSchemas().size(); return Reactive.Source.guarded( () -> getConnectionBudget().acquireUninterruptibly(tables), - () -> getConnectionBudget().release(tables), + () -> { + getConnectionBudget().release(tables); + jobHook.ifPresent(hook -> hook.update(1)); + }, merged); }; diff --git a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryBatch.java b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryBatch.java index 15acb2d6d..10f419e01 100644 --- a/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryBatch.java +++ b/xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/domain/SqlQueryBatch.java @@ -7,12 +7,17 @@ */ package de.ii.xtraplatform.features.sql.domain; +import de.ii.xtraplatform.features.domain.JobHook; import java.util.List; +import java.util.Optional; import org.immutables.value.Value; @Value.Immutable public interface SqlQueryBatch { + @Value.Auxiliary + Optional getJobHook(); + long getLimit(); long getOffset(); diff --git a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureStreamImpl.java b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureStreamImpl.java index d4f797a9f..ad364e6ac 100644 --- a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureStreamImpl.java +++ b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureStreamImpl.java @@ -208,6 +208,10 @@ public CompletionStage runWith( source = source.via(new FeatureTokenTransformerHooks(resultBuilder, onCollectionMetadata)); + if (query.getJobHook().isPresent()) { + source = source.via(new FeatureTokenTransformerProgress(query.getJobHook().get())); + } + Reactive.BasicStream basicStream = sink instanceof Reactive.SinkTransformed ? source.to((Reactive.SinkTransformed) sink) @@ -308,6 +312,10 @@ public CompletionStage> runWith( source = source.via(new FeatureTokenTransformerHooks(resultBuilder, onCollectionMetadata)); + if (query.getJobHook().isPresent()) { + source = source.via(new FeatureTokenTransformerProgress(query.getJobHook().get())); + } + Reactive.BasicStream basicStream = sink instanceof Reactive.SinkReducedTransformed ? source.to((Reactive.SinkReducedTransformed) sink) diff --git a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureTokenTransformerProgress.java b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureTokenTransformerProgress.java new file mode 100644 index 000000000..3156a8e22 --- /dev/null +++ b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureTokenTransformerProgress.java @@ -0,0 +1,43 @@ +/* + * Copyright 2026 interactive instruments GmbH + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ +package de.ii.xtraplatform.features.domain; + +public class FeatureTokenTransformerProgress extends FeatureTokenTransformer { + + private final JobHook jobHook; + private boolean countFeatures; + + public FeatureTokenTransformerProgress(JobHook jobHook) { + this.jobHook = jobHook; + this.countFeatures = false; + } + + @Override + public void onStart(ModifiableContext context) { + // features are only counted when the total is known (paged queries with a meta phase); + // unpaged queries report sub-query progress in the connector instead + if (context.metadata().getNumberReturned().isPresent()) { + long numberReturned = context.metadata().getNumberReturned().getAsLong(); + jobHook.init((int) Math.min(Integer.MAX_VALUE, numberReturned)); + this.countFeatures = true; + } + + super.onStart(context); + } + + @Override + public void onFeatureStart(ModifiableContext context) { + jobHook.checkpoint(); + + if (countFeatures) { + jobHook.update(1); + } + + super.onFeatureStart(context); + } +} diff --git a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobCancelledException.java b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobCancelledException.java new file mode 100644 index 000000000..8e41c2aea --- /dev/null +++ b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobCancelledException.java @@ -0,0 +1,20 @@ +/* + * Copyright 2026 interactive instruments GmbH + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ +package de.ii.xtraplatform.features.domain; + +/** + * Thrown by {@link JobHook#checkpoint()} when cancellation was requested. Inside a token pipeline + * it rides the regular stream-error path, which cancels the upstream subscription and closes the + * database resources. + */ +public class JobCancelledException extends RuntimeException { + + public JobCancelledException() { + super("The request was cancelled"); + } +} diff --git a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobHook.java b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobHook.java new file mode 100644 index 000000000..63670ee85 --- /dev/null +++ b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobHook.java @@ -0,0 +1,36 @@ +/* + * Copyright 2026 interactive instruments GmbH + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ +package de.ii.xtraplatform.features.domain; + +/** + * Receives progress from and requests cancellation of a running query or mutation. Queries carry + * the hook as an {@code Optional}; an instance exists only when there is a real hook, an absent + * hook means no instrumentation. + * + *

All methods may be called from stream/executor threads; implementations must be thread-safe + * and {@link #isCancelRequested()} must be cheap enough to poll per feature. + */ +public interface JobHook { + + /** The total the progress reported via {@link #update(int)} counts towards. */ + default void init(int total) {} + + /** Adds {@code delta} to the current progress. */ + default void update(int delta) {} + + default boolean isCancelRequested() { + return false; + } + + /** Cooperative cancellation point. */ + default void checkpoint() { + if (isCancelRequested()) { + throw new JobCancelledException(); + } + } +} diff --git a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/Query.java b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/Query.java index 4332028c6..3dbc9045b 100644 --- a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/Query.java +++ b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/Query.java @@ -7,6 +7,7 @@ */ package de.ii.xtraplatform.features.domain; +import com.fasterxml.jackson.annotation.JsonIgnore; import com.google.common.collect.ImmutableList; import de.ii.xtraplatform.crs.domain.EpsgCrs; import de.ii.xtraplatform.features.domain.FeatureStream.PipelineSteps; @@ -18,6 +19,10 @@ public interface Query { Optional getCrs(); + @JsonIgnore + @Value.Auxiliary + Optional getJobHook(); + @Value.Default default double getMaxAllowableOffset() { return 0; diff --git a/xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/FeatureTokenTransformerProgressSpec.groovy b/xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/FeatureTokenTransformerProgressSpec.groovy new file mode 100644 index 000000000..35d3e1070 --- /dev/null +++ b/xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/FeatureTokenTransformerProgressSpec.groovy @@ -0,0 +1,115 @@ +/* + * Copyright 2026 interactive instruments GmbH + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ +package de.ii.xtraplatform.features.domain + +import spock.lang.Specification + +class FeatureTokenTransformerProgressSpec extends Specification { + + RecordingJobHook jobHook + FeatureTokenReader tokenReader + List tokens + + def setup() { + jobHook = new RecordingJobHook() + FeatureTokenTransformerProgress transformer = new FeatureTokenTransformerProgress(jobHook) + FeatureQuery query = ImmutableFeatureQuery.builder().type("test").build() + FeatureEventHandler.ModifiableContext context = transformer.createContext() + .setQuery(query) + .setMappings([test: FeatureSchemaFixtures.BIOTOP_MAPPING]) + .setType('test') + .setIsUseTargetPaths(true) + + tokenReader = new FeatureTokenReader(transformer, context) + tokens = [] + transformer.init(token -> tokens.add(token)) + } + + def 'init from numberReturned and one update per feature'() { + + given: + + when: + + FeatureTokenFixtures.COLLECTION.forEach(token -> tokenReader.onToken(token)) + + then: + + jobHook.inits == [3] + jobHook.updates == [1, 1, 1] + + } + + def 'no init and no updates without numberReturned'() { + + given: + + when: + + FeatureTokenFixtures.SINGLE_FEATURE.forEach(token -> tokenReader.onToken(token)) + + then: + + jobHook.inits == [] + jobHook.updates == [] + + } + + def 'tokens pass through unchanged'() { + + given: + + when: + + FeatureTokenFixtures.COLLECTION.forEach(token -> tokenReader.onToken(token)) + + then: + + tokens == FeatureTokenFixtures.COLLECTION + + } + + def 'cancellation is requested before the next feature'() { + + given: + + jobHook.cancelRequested = true + + when: + + FeatureTokenFixtures.COLLECTION.forEach(token -> tokenReader.onToken(token)) + + then: + + thrown(JobCancelledException) + jobHook.updates == [] + + } +} + +class RecordingJobHook implements JobHook { + + List inits = [] + List updates = [] + boolean cancelRequested = false + + @Override + void init(int total) { + inits.add(total) + } + + @Override + void update(int delta) { + updates.add(delta) + } + + @Override + boolean isCancelRequested() { + return cancelRequested + } +} diff --git a/xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/JobHookSpec.groovy b/xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/JobHookSpec.groovy new file mode 100644 index 000000000..e7fb8d7da --- /dev/null +++ b/xtraplatform-features/src/test/groovy/de/ii/xtraplatform/features/domain/JobHookSpec.groovy @@ -0,0 +1,69 @@ +/* + * Copyright 2026 interactive instruments GmbH + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ +package de.ii.xtraplatform.features.domain + +import spock.lang.Specification + +class JobHookSpec extends Specification { + + def 'the default methods are inert'() { + + given: + + JobHook hook = new JobHook() {} + + when: + + hook.init(100) + hook.update(1) + hook.checkpoint() + + then: + + noExceptionThrown() + !hook.isCancelRequested() + + } + + def 'checkpoint throws when cancellation was requested'() { + + given: + + JobHook hook = new JobHook() { + @Override + boolean isCancelRequested() { + return true + } + } + + when: + + hook.checkpoint() + + then: + + thrown(JobCancelledException) + + } + + def 'checkpoint passes when cancellation was not requested'() { + + given: + + JobHook hook = new JobHook() {} + + when: + + hook.checkpoint() + + then: + + noExceptionThrown() + + } +} From dcec2d577292967aaa4164fbc9bec260791a8a8d Mon Sep 17 00:00:00 2001 From: Andreas Zahnen Date: Wed, 21 Oct 2026 13:18:52 +0200 Subject: [PATCH 2/2] cancellable job hooks --- xtraplatform-features/build.gradle | 1 + .../features/domain/JobHookCancellable.java | 34 +++++++++++++++++++ 2 files changed, 35 insertions(+) create mode 100644 xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobHookCancellable.java diff --git a/xtraplatform-features/build.gradle b/xtraplatform-features/build.gradle index f5387471f..ab24ed3d9 100644 --- a/xtraplatform-features/build.gradle +++ b/xtraplatform-features/build.gradle @@ -9,6 +9,7 @@ dependencies { provided 'de.interactive_instruments:xtraplatform-streams' provided 'de.interactive_instruments:xtraplatform-values' provided 'de.interactive_instruments:xtraplatform-services' + provided 'de.interactive_instruments:xtraplatform-xtralink' provided project(':xtraplatform-codelists') provided project(':xtraplatform-cql') provided project(':xtraplatform-crs') diff --git a/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobHookCancellable.java b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobHookCancellable.java new file mode 100644 index 000000000..50cf1b458 --- /dev/null +++ b/xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/JobHookCancellable.java @@ -0,0 +1,34 @@ +/* + * Copyright 2026 interactive instruments GmbH + * + * This Source Code Form is subject to the terms of the Mozilla Public + * License, v. 2.0. If a copy of the MPL was not distributed with this + * file, You can obtain one at http://mozilla.org/MPL/2.0/. + */ +package de.ii.xtraplatform.features.domain; + +import de.ii.xtraplatform.xtralink.domain.JobProcessing; + +public abstract class JobHookCancellable implements JobHook { + + protected final JobProcessing jobs; + protected final String jobId; + private boolean cancelRequested; + + public JobHookCancellable(JobProcessing jobs, String jobId) { + this.jobs = jobs; + this.jobId = jobId; + this.cancelRequested = false; + } + + @Override + public final boolean isCancelRequested() { + if (cancelRequested) { + return true; + } + if (jobs.isCancelled(jobId)) { + this.cancelRequested = true; + } + return cancelRequested; + } +}