perf(processing-folders): pace sweep runs and take smallest files first

A sweep fans out one run per file and dispatched all of them at once onto the
unbounded virtual-thread executor. Every run then converges on the pipeline's
slowest tool — on a desktop install, the local AI engine, which serves a
couple of requests at a time — so the whole folder sat "in progress" with
nothing visibly finishing until the end, then completed in clumps. Same total
time as pacing, with the worst possible feel.

Each sweep now carries an admission gate (policies.sweepConcurrency, default
2, 0 = unbounded): every run is still registered and reported to the caller
immediately, but only that many execute at once — parked runs sit honestly
pending on their virtual threads and start as slots free. Completions arrive
as a steady drip from the first file onward, which is also what feeds the
sweep-result delivery opening files into the workbench one after another.

The disk listing also orders a sweep smallest-file-first, so the first result
appears within seconds of approving rather than after the largest document in
the folder.

app/common/src/main/java/stirling/software/common/model/ApplicationProperties.java

@@ -215,6 +215,15 @@ public class ApplicationProperties {
          */
         private List<String> allowedFolderRoots = new java.util.ArrayList<>();
 
+        /**
+         * How many of one sweep's runs may execute at once; further runs queue (visible as pending)
+         * and start as slots free up. Sweeps fan out one run per file, and a folder of documents
+         * dispatched all at once just piles up at the pipeline's slowest tool (on a desktop
+         * install, the local AI engine) — same total time, but nothing visibly finishes until the
+         * end. A small cap keeps completions steady. 0 = unbounded.
+         */
+        private int sweepConcurrency = 2;
+
         /** How often (seconds) the schedule trigger checks for policies whose schedule is due. */
         private long scheduleSweepSeconds = 60;
 

app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java

@@ -7,6 +7,7 @@ import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Semaphore;
 
 import org.slf4j.MDC;
 import org.springframework.core.io.Resource;
@@ -130,7 +131,7 @@ public class PolicyEngine {
         // worker.
         String principal = currentActingPrincipal();
         return submitForPrincipal(
-                principal, principal, policyId, definition, inputs, listener, null, null);
+                principal, principal, policyId, definition, inputs, listener, null, null, null);
     }
 
     /** Run a stored policy on demand. {@code enabled} gates triggers, not explicit runs. */
@@ -151,6 +152,23 @@ public class PolicyEngine {
             PolicyProgressListener listener,
             String sourceId,
             String fileIdentity) {
+        return runPolicy(policy, inputs, listener, sourceId, fileIdentity, null);
+    }
+
+    /**
+     * As above, additionally pacing execution through {@code admission}: the run is registered and
+     * visible immediately (pending), but its work only proceeds while holding a permit. A sweep
+     * passes one gate for all its runs so a folderful of files executes a few at a time — the
+     * pipeline's slowest tool serializes them anyway, and paced runs finish steadily instead of all
+     * sitting in-flight until the end. Null means ungated.
+     */
+    public PolicyRunHandle runPolicy(
+            Policy policy,
+            PolicyInputs inputs,
+            PolicyProgressListener listener,
+            String sourceId,
+            String fileIdentity,
+            Semaphore admission) {
         // Bill the policy owner: trigger-fired runs have no security context, and the async worker
         // doesn't inherit the caller's, so the owner (stamped at policy creation) is the reliable
         // billing identity — and for org-wide policies the org/owner is meant to pay. But own the
@@ -179,7 +197,8 @@ public class PolicyEngine {
                 resolved,
                 listener,
                 sourceId,
-                fileIdentity);
+                fileIdentity,
+                admission);
     }
 
     private PolicyRunHandle submitForPrincipal(
@@ -190,7 +209,8 @@ public class PolicyEngine {
             PolicyInputs inputs,
             PolicyProgressListener listener,
             String sourceId,
-            String fileIdentity) {
+            String fileIdentity,
+            Semaphore admission) {
         // Scope the run id to the current user (this request thread) so the file-download
         // ownership check passes. No-op when security is off.
         String runId = jobOwnershipService.createScopedJobKey(UUID.randomUUID().toString());
@@ -214,7 +234,27 @@ public class PolicyEngine {
                                 billingPrincipal,
                                 fileOwner,
                                 definition.name(),
-                                () -> runToCompletion(run, inputs, tracking, completion));
+                                () -> {
+                                    // Pacing gate: park (cheap on a virtual thread, and the run
+                                    // honestly reads as pending) until a slot frees up.
+                                    if (admission != null) {
+                                        try {
+                                            admission.acquire();
+                                        } catch (InterruptedException e) {
+                                            // Shutdown while parked: never ran, never will.
+                                            Thread.currentThread().interrupt();
+                                            completion.completeExceptionally(e);
+                                            return;
+                                        }
+                                    }
+                                    try {
+                                        runToCompletion(run, inputs, tracking, completion);
+                                    } finally {
+                                        if (admission != null) {
+                                            admission.release();
+                                        }
+                                    }
+                                });
 
         // One admission unit per run; steps run synchronously within it, so this gates heavy work
         // without the pool-within-pool risk of queueing each tool call.

app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java

@@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.engine;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.Semaphore;
 import java.util.function.Consumer;
 
 import org.springframework.stereotype.Service;
@@ -10,6 +11,7 @@ import org.springframework.stereotype.Service;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
+import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.policy.input.InputSource;
 import stirling.software.proprietary.policy.input.ResolvedInput;
 import stirling.software.proprietary.policy.ledger.ProcessedLedger;
@@ -42,6 +44,18 @@ public class PolicyRunner {
     private final SourceStore sourceStore;
     private final SourceDocCounter docCounter;
     private final ProcessedLedger processedLedger;
+    private final ApplicationProperties applicationProperties;
+
+    /**
+     * One admission gate per sweep: every run still starts (and is visible) immediately, but only
+     * this many execute at once. A sweep fans out one run per file, and a folderful dispatched all
+     * at once just queues at the pipeline's slowest tool — same total time, nothing visibly done
+     * until the end. Paced, completions arrive steadily from the first file onward.
+     */
+    private Semaphore sweepAdmission() {
+        int concurrency = applicationProperties.getPolicies().getSweepConcurrency();
+        return concurrency > 0 ? new Semaphore(concurrency) : null;
+    }
 
     /** Full-listing sweep over every input: resolve each source, then reconcile the ledger. */
     public SweepOutcome run(Policy policy) {
@@ -74,13 +88,21 @@ public class PolicyRunner {
     public SweepOutcome run(Policy policy, List<PipelineInput> inputs, SweepKind sweep) {
         long sweepStart = System.currentTimeMillis();
         PolicySweep context = new PolicySweep(policy.id(), sweep, processedLedger);
+        Semaphore admission = sweepAdmission();
         List<String> runIds = new ArrayList<>();
         if (inputs.isEmpty()) {
             // Generator pipeline: one run with no input. Still fall through to the cleanup
             // below so rows recorded for its folder outputs are pruned like anything else,
             // instead of accumulating until the policy is deleted.
             // Generator pipeline: no input, so neither a source nor a document to attribute to.
-            runIds.add(startRun(policy, null, null, PolicyInputs.of(List.of()), unused -> {}));
+            runIds.add(
+                    startRun(
+                            policy,
+                            null,
+                            null,
+                            PolicyInputs.of(List.of()),
+                            unused -> {},
+                            admission));
         }
         for (PipelineInput input : inputs) {
             String sourceId = input.sourceId();
@@ -100,7 +122,7 @@ public class PolicyRunner {
                 context.vetoCleanup();
                 continue;
             }
-            runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec(), context));
+            runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec(), context, admission));
         }
         boolean fullPolicy = inputs.size() == policy.inputs().size();
         if (fullPolicy && context.cleanupAllowed()) {
@@ -140,7 +162,11 @@ public class PolicyRunner {
      * this sweep's ledger cleanup.
      */
     private List<String> pullAndRun(
-            Policy policy, String sourceId, InputSpec spec, PolicySweep context) {
+            Policy policy,
+            String sourceId,
+            InputSpec spec,
+            PolicySweep context,
+            Semaphore admission) {
         InputSource source = sourceFor(spec);
         if (source == null) {
             log.warn(
@@ -174,7 +200,8 @@ public class PolicyRunner {
                             sourceId,
                             unit.fileIdentity(),
                             unit.inputs(),
-                            unit.onComplete()));
+                            unit.onComplete(),
+                            admission));
             docsFed += unit.inputs().primary().size();
         }
         docCounter.record(sourceId, docsFed);
@@ -186,11 +213,17 @@ public class PolicyRunner {
             String sourceId,
             String fileIdentity,
             PolicyInputs inputs,
-            Consumer<Boolean> onComplete) {
+            Consumer<Boolean> onComplete,
+            Semaphore admission) {
         log.info("Running policy {} ({})", policy.id(), policy.name());
         PolicyRunHandle handle =
                 policyEngine.runPolicy(
-                        policy, inputs, PolicyProgressListener.NOOP, sourceId, fileIdentity);
+                        policy,
+                        inputs,
+                        PolicyProgressListener.NOOP,
+                        sourceId,
+                        fileIdentity,
+                        admission);
         handle.completion()
                 .whenComplete((run, throwable) -> onComplete.accept(succeeded(run, throwable)));
         return handle.runId();

app/proprietary/src/main/java/stirling/software/proprietary/policy/input/FolderInputSource.java

@@ -9,6 +9,7 @@ import java.nio.file.Path;
 import java.nio.file.SimpleFileVisitor;
 import java.nio.file.attribute.BasicFileAttributes;
 import java.util.ArrayList;
+import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Supplier;
@@ -83,6 +84,10 @@ public class FolderInputSource implements InputSource {
         }
         Path canonicalDir = FolderIdentities.canonicalDir(inputDir);
         List<Path> present = listFiles(inputDir, config.recursive());
+        // Smallest first: a sweep's first results should appear within seconds of it starting,
+        // not after the largest document in the folder. Unsizeable entries (vanished mid-listing)
+        // sort last and resolve their own fate at claim time.
+        present.sort(Comparator.comparingLong(FolderInputSource::sizeForOrdering));
 
         if (config.snapshot()) {
             List<ResolvedInput> work = new ArrayList<>();
@@ -238,6 +243,15 @@ public class FolderInputSource implements InputSource {
     }
 
     /** Every non-hidden regular file in the source, readable or not. */
+    /** The file's size for sweep ordering; unreadable reads as largest, sorting it last. */
+    private static long sizeForOrdering(Path file) {
+        try {
+            return Files.size(file);
+        } catch (IOException e) {
+            return Long.MAX_VALUE;
+        }
+    }
+
     private static List<Path> listFiles(Path inputDir, boolean recursive) throws IOException {
         List<Path> files = new ArrayList<>();
         if (!recursive) {

app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java

@@ -29,6 +29,7 @@ import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 import org.springframework.core.io.ByteArrayResource;
 
+import stirling.software.common.model.ApplicationProperties;
 import stirling.software.proprietary.policy.input.InputSource;
 import stirling.software.proprietary.policy.input.ResolveContext;
 import stirling.software.proprietary.policy.input.ResolvedInput;
@@ -72,19 +73,20 @@ class PolicyRunnerTest {
                         List.of(folderSource),
                         sourceStore,
                         docCounter,
-                        processedLedger);
+                        processedLedger,
+                        new ApplicationProperties());
     }
 
     @Test
     void runsOnceWithNoFilesWhenThePolicyHasNoSources() {
         Policy policy = policy(List.of());
-        when(policyEngine.runPolicy(eq(policy), any(), any(), any(), any()))
+        when(policyEngine.runPolicy(eq(policy), any(), any(), any(), any(), any()))
                 .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>()));
 
         runner.run(policy);
 
         ArgumentCaptor<PolicyInputs> inputs = ArgumentCaptor.forClass(PolicyInputs.class);
-        verify(policyEngine).runPolicy(eq(policy), inputs.capture(), any(), any(), any());
+        verify(policyEngine).runPolicy(eq(policy), inputs.capture(), any(), any(), any(), any());
         assertTrue(inputs.getValue().primary().isEmpty());
         // Ledger hygiene still runs: rows recorded for a generator policy's folder outputs
         // are pruned by its own sweeps rather than accumulating until the policy is deleted.
@@ -100,7 +102,8 @@ class PolicyRunnerTest {
                         List.of(folderSource),
                         sourceStore,
                         new InProcessSourceDocCounter(),
-                        ledger);
+                        ledger,
+                        new ApplicationProperties());
         InputSpec spec = InputSpec.folder("/in");
         Policy policy = policy(List.of(spec));
         // One file already processed at its current version, one parked by a failed run.
@@ -137,12 +140,12 @@ class PolicyRunnerTest {
                         List.of(
                                 ResolvedInput.of(PolicyInputs.of(List.of())),
                                 ResolvedInput.of(PolicyInputs.of(List.of()))));
-        when(policyEngine.runPolicy(any(), any(), any(), any(), any()))
+        when(policyEngine.runPolicy(any(), any(), any(), any(), any(), any()))
                 .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>()));
 
         runner.run(policy);
 
-        verify(policyEngine, times(2)).runPolicy(eq(policy), any(), any(), any(), any());
+        verify(policyEngine, times(2)).runPolicy(eq(policy), any(), any(), any(), any(), any());
     }
 
     @Test
@@ -154,7 +157,7 @@ class PolicyRunnerTest {
         when(folderSource.supports(spec)).thenReturn(true);
         when(folderSource.resolve(eq(spec), any())).thenReturn(List.of(unit));
         CompletableFuture<PolicyRun> completion = new CompletableFuture<>();
-        when(policyEngine.runPolicy(any(), any(), any(), any(), any()))
+        when(policyEngine.runPolicy(any(), any(), any(), any(), any(), any()))
                 .thenReturn(new PolicyRunHandle("r", completion));
 
         runner.run(policy);
@@ -175,7 +178,7 @@ class PolicyRunnerTest {
         when(folderSource.supports(spec)).thenReturn(true);
         when(folderSource.resolve(eq(spec), any())).thenReturn(List.of(unit));
         CompletableFuture<PolicyRun> completion = new CompletableFuture<>();
-        when(policyEngine.runPolicy(any(), any(), any(), any(), any()))
+        when(policyEngine.runPolicy(any(), any(), any(), any(), any(), any()))
                 .thenReturn(new PolicyRunHandle("r", completion));
 
         runner.run(policy);
@@ -224,12 +227,12 @@ class PolicyRunnerTest {
         when(folderSource.supports(spec)).thenReturn(true);
         when(folderSource.resolve(eq(spec), any()))
                 .thenReturn(List.of(ResolvedInput.of(PolicyInputs.of(List.of()))));
-        when(policyEngine.runPolicy(any(), any(), any(), any(), any()))
+        when(policyEngine.runPolicy(any(), any(), any(), any(), any(), any()))
                 .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>()));
 
         runner.run(policy, SweepKind.LIGHT);
 
-        verify(policyEngine).runPolicy(eq(policy), any(), any(), any(), any());
+        verify(policyEngine).runPolicy(eq(policy), any(), any(), any(), any(), any());
         verify(processedLedger, never()).markSeen(any(), any());
         verify(processedLedger, never()).deleteUnseen(any(), anyLong());
     }
@@ -244,13 +247,14 @@ class PolicyRunnerTest {
         when(folderSource.resolve(eq(broken), any())).thenThrow(new IOException("mount gone"));
         when(folderSource.resolve(eq(healthy), any()))
                 .thenReturn(List.of(ResolvedInput.of(PolicyInputs.of(List.of()))));
-        when(policyEngine.runPolicy(any(), any(), any(), any(), any()))
+        when(policyEngine.runPolicy(any(), any(), any(), any(), any(), any()))
                 .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>()));
 
         runner.run(policy);
 
         verify(policyEngine)
-                .runPolicy(eq(policy), any(), any(), any(), any()); // healthy source still ran
+                .runPolicy(
+                        eq(policy), any(), any(), any(), any(), any()); // healthy source still ran
         verify(processedLedger, never()).deleteUnseen(any(), anyLong()); // history preserved
     }