Async job hooks

xtraplatform-features-sql/src/main/java/de/ii/xtraplatform/features/sql/app/FeatureQueryEncoderSql.java

@@ -124,6 +124,7 @@ private SqlQueryBatch encode(FeatureQuery query, Map<String, String> 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)

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<SqlRow> getSourceStream(
                       if (queryBatch.isSingleFeature() || queryBatch.isUnpaged()) {
                         boolean unpaged = queryBatch.isUnpaged();
                         List<SqlQuerySet> querySets = queryBatch.getQuerySets();
+                        Optional<JobHook> 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<SqlRow> 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);
                             };
 

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<JobHook> getJobHook();
+
   long getLimit();
 
   long getOffset();

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')

xtraplatform-features/src/main/java/de/ii/xtraplatform/features/domain/FeatureStreamImpl.java

@@ -229,6 +229,10 @@ public CompletionStage<Result> runWith(
           source =
               source.via(new FeatureTokenTransformerHooks(resultBuilder, onCollectionMetadata));
 
+          if (query.getJobHook().isPresent()) {
+            source = source.via(new FeatureTokenTransformerProgress(query.getJobHook().get()));
+          }
+
           Reactive.BasicStream<?, Void> basicStream =
               sink instanceof Reactive.SinkTransformed
                   ? source.to((Reactive.SinkTransformed<Object, ?>) sink)
@@ -330,6 +334,10 @@ public <X> CompletionStage<ResultReduced<X>> runWith(
           source =
               source.via(new FeatureTokenTransformerHooks(resultBuilder, onCollectionMetadata));
 
+          if (query.getJobHook().isPresent()) {
+            source = source.via(new FeatureTokenTransformerProgress(query.getJobHook().get()));
+          }
+
           Reactive.BasicStream<?, X> basicStream =
               sink instanceof Reactive.SinkReducedTransformed
                   ? source.to((Reactive.SinkReducedTransformed<Object, ?, X>) sink)

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<FeatureSchema, SchemaMapping> 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<FeatureSchema, SchemaMapping> context) {
+    jobHook.checkpoint();
+
+    if (countFeatures) {
+      jobHook.update(1);
+    }
+
+    super.onFeatureStart(context);
+  }
+}

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");
+  }
+}

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.
+ *
+ * <p>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();
+    }
+  }
+}

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;
+  }
+}

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<EpsgCrs> getCrs();
 
+  @JsonIgnore
+  @Value.Auxiliary
+  Optional<JobHook> getJobHook();
+
   @Value.Default
   default double getMaxAllowableOffset() {
     return 0;

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<Object> 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<Integer> inits = []
+    List<Integer> 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
+    }
+}