fix(java): prevent executor, task, regex, and stream resource leaks (#7284)

# Description of Changes - Added graceful shutdown handling for service-owned executors in `JobExecutorService`, `PolicyEngine`, and `AsyncConfig`. - Added expiration and cleanup for abandoned pending jobs in `TaskManager`. - Replaced the unbounded regex pattern cache with a bounded cache limited to 512 entries. - Ensured `Files.walk()` is closed correctly in `MobileScannerService`. - These changes prevent unbounded heap growth, lingering virtual-thread executors, and file-descriptor leaks. - Added configurable pending-job expiration through `stirling.job.pendingExpiryMinutes`, defaulting to 24 hours. --- ## Checklist ### General - [ ] I have read the [Contribution Guidelines](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/CONTRIBUTING.md) - [ ] I have read the [Stirling-PDF Developer Guide](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/DeveloperGuide.md) (if applicable) - [ ] I have read the [How to add new languages to Stirling-PDF](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/devGuide/HowToAddNewLanguage.md) (if applicable) - [ ] I have performed a self-review of my own code - [ ] My changes generate no new warnings ### Documentation - [ ] I have updated relevant docs on [Stirling-PDF's doc repo](https://github.com/Stirling-Tools/Stirling-Tools.github.io/blob/main/docs/) (if functionality has heavily changed) - [ ] I have read the section [Add New Translation Tags](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/devGuide/HowToAddNewLanguage.md#add-new-translation-tags) (for new translation tags only) ### Translations (if applicable) - [ ] I ran [`scripts/counter_translation.py`](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/docs/counter_translation.md) ### UI Changes (if applicable) - [ ] Screenshots or videos demonstrating the UI changes are attached (e.g., as comments or direct attachments in the PR) ### Testing (if applicable) - [ ] I have run `task check` to verify linters, typechecks, and tests pass - [ ] I have tested my changes locally. Refer to the [Testing Guide](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/DeveloperGuide.md#7-testing) for more details.
2026-09-02 21:03:34 +03:00 · 2026-08-06 11:06:28 +00:00
parent 866e56728d
commit b10fc1b2de
6 changed files with 128 additions and 30 deletions
@@ -19,6 +19,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.web.multipart.MultipartFile;
 import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody;

+import jakarta.annotation.PreDestroy;
 import jakarta.servlet.http.HttpServletRequest;

 import lombok.extern.slf4j.Slf4j;
@@ -63,6 +64,21 @@ public class JobExecutorService {
                "Job executor configured with effective timeout of {} ms", this.effectiveTimeoutMs);
    }

+    /** Stop the service-owned executor when the application context is closed or restarted. */
+    @PreDestroy
+    public void shutdown() {
+        log.debug("Shutting down job executor");
+        executor.shutdown();
+        try {
+            if (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
+                executor.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            executor.shutdownNow();
+        }
+    }
+
    public ResponseEntity<?> runJobGeneric(boolean async, Supplier<Object> work) {
        return runJobGeneric(async, work, -1);
    }
@@ -225,19 +225,21 @@ public class MobileScannerService {
                Path sessionDir = getSafeSessionDirectory(sessionId);
                if (Files.exists(sessionDir)) {
                    // Delete all files in session directory
-                    Files.walk(sessionDir)
-                            .sorted(
-                                    (a, b) ->
-                                            -a.compareTo(b)) // Reverse order to delete files before
-                            // directory
-                            .forEach(
-                                    path -> {
-                                        try {
-                                            Files.deleteIfExists(path);
-                                        } catch (IOException e) {
-                                            log.warn("Failed to delete file: {}", path, e);
-                                        }
-                                    });
+                    try (var paths = Files.walk(sessionDir)) {
+                        paths.sorted(
+                                        (a, b) ->
+                                                -a.compareTo(
+                                                        b)) // Reverse order to delete files before
+                                // directory
+                                .forEach(
+                                        path -> {
+                                            try {
+                                                Files.deleteIfExists(path);
+                                            } catch (IOException e) {
+                                                log.warn("Failed to delete file: {}", path, e);
+                                            }
+                                        });
+                    }
                }
                log.info("Deleted session: {}", sessionId);
            } catch (IllegalArgumentException e) {
@@ -48,6 +48,10 @@ public class TaskManager {
    @Value("${stirling.jobResultExpiryMinutes:30}")
    private int jobResultExpiryMinutes = 30;

+    /** Maximum age of a task that never reached a terminal state. */
+    @Value("${stirling.job.pendingExpiryMinutes:1440}")
+    private int pendingJobExpiryMinutes = 1440;
+
    private final FileStorage fileStorage;
    private final JobStore jobStore;
    private final ClusterBackplane clusterBackplane;
@@ -332,19 +336,32 @@ public class TaskManager {
        }
        LocalDateTime expiryThreshold =
                LocalDateTime.now().minus(jobResultExpiryMinutes, ChronoUnit.MINUTES);
+        LocalDateTime pendingExpiryThreshold =
+                LocalDateTime.now().minus(pendingJobExpiryMinutes, ChronoUnit.MINUTES);
        int removedCount = 0;

        try {
            for (Map.Entry<String, JobResult> entry : jobResults.entrySet()) {
                JobResult result = entry.getValue();

-                // Remove completed jobs that are older than the expiry threshold
-                if (result.isComplete()
-                        && result.getCompletedAt() != null
-                        && result.getCompletedAt().isBefore(expiryThreshold)) {
+                boolean expiredCompletedJob =
+                        result.isComplete()
+                                && result.getCompletedAt() != null
+                                && result.getCompletedAt().isBefore(expiryThreshold);
+                boolean abandonedPendingJob =
+                        !result.isComplete()
+                                && result.getCreatedAt() != null
+                                && result.getCreatedAt().isBefore(pendingExpiryThreshold);
+
+                // Remove old terminal results and abandoned pending jobs. Without the second
+                // branch, a client that starts a task and never completes it keeps its result in
+                // memory forever.
+                if (expiredCompletedJob || abandonedPendingJob) {

                    // Clean up file results
-                    cleanupJobFiles(result, entry.getKey());
+                    if (expiredCompletedJob) {
+                        cleanupJobFiles(result, entry.getKey());
+                    }

                    // Remove the job result
                    jobResults.remove(entry.getKey());
@@ -1,17 +1,22 @@
 package stirling.software.common.util;

 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
 import java.util.regex.Pattern;
 import java.util.regex.PatternSyntaxException;

+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.util.concurrent.UncheckedExecutionException;
+
 import lombok.extern.slf4j.Slf4j;

@Slf4j
 public final class RegexPatternUtils {

    private static final RegexPatternUtils INSTANCE = new RegexPatternUtils();
-    private final ConcurrentHashMap<PatternKey, Pattern> patternCache = new ConcurrentHashMap<>();
+    private static final long MAX_CACHED_PATTERNS = 512;
+    private final Cache<PatternKey, Pattern> patternCache =
+            CacheBuilder.newBuilder().maximumSize(MAX_CACHED_PATTERNS).build();

    private static final String WHITESPACE_REGEX = "\\s++";
    private static final String EXTENSION_REGEX = "\\.(?:[^.]*+)?$";
@@ -51,7 +56,7 @@ public final class RegexPatternUtils {
            throw new IllegalArgumentException("Regex pattern cannot be null");
        }

-        return patternCache.computeIfAbsent(new PatternKey(regex, 0), this::compilePattern);
+        return getOrCompile(new PatternKey(regex, 0));
    }

    /**
@@ -77,7 +82,7 @@ public final class RegexPatternUtils {
            throw new IllegalArgumentException("Regex pattern cannot be null");
        }

-        return patternCache.computeIfAbsent(new PatternKey(regex, flags), this::compilePattern);
+        return getOrCompile(new PatternKey(regex, flags));
    }

    /**
@@ -98,7 +103,7 @@ public final class RegexPatternUtils {
     * @return true if pattern is cached, false otherwise
     */
    public boolean isCached(String regex, int flags) {
-        return regex != null && patternCache.containsKey(new PatternKey(regex, flags));
+        return regex != null && patternCache.getIfPresent(new PatternKey(regex, flags)) != null;
    }

    /**
@@ -107,7 +112,7 @@ public final class RegexPatternUtils {
     * @return number of patterns currently cached
     */
    public int getCacheSize() {
-        return patternCache.size();
+        return (int) patternCache.size();
    }

    /**
@@ -115,7 +120,7 @@ public final class RegexPatternUtils {
     * useful for testing or memory cleanup in long-running applications.
     */
    public void clearCache() {
-        patternCache.clear();
+        patternCache.invalidateAll();
        log.debug("Regex pattern cache cleared");
    }

@@ -141,13 +146,32 @@ public final class RegexPatternUtils {
            return false;
        }
        PatternKey key = new PatternKey(regex, flags);
-        boolean removed = patternCache.remove(key) != null;
+        boolean removed = patternCache.getIfPresent(key) != null;
+        patternCache.invalidate(key);
        if (removed) {
            log.debug("Removed regex pattern from cache: {} (flags: {})", regex, flags);
        }
        return removed;
    }

+    private Pattern getOrCompile(PatternKey key) {
+        try {
+            return patternCache.get(key, () -> compilePattern(key));
+        } catch (UncheckedExecutionException e) {
+            Throwable cause = e.getCause();
+            if (cause instanceof PatternSyntaxException patternSyntaxException) {
+                throw patternSyntaxException;
+            }
+            throw e;
+        } catch (java.util.concurrent.ExecutionException e) {
+            Throwable cause = e.getCause();
+            if (cause instanceof PatternSyntaxException patternSyntaxException) {
+                throw patternSyntaxException;
+            }
+            throw new IllegalStateException("Failed to compile regex pattern", cause);
+        }
+    }
+
    /**
     * Internal method to compile a pattern and handle errors consistently.
     *
@@ -2,6 +2,7 @@ package stirling.software.proprietary.config;

 import java.util.Map;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;

 import org.slf4j.MDC;
@@ -12,10 +13,15 @@ import org.springframework.core.task.support.TaskExecutorAdapter;
 import org.springframework.scheduling.annotation.EnableAsync;
 import org.springframework.security.concurrent.DelegatingSecurityContextExecutor;

+import jakarta.annotation.PreDestroy;
+
@Configuration
@EnableAsync
 public class AsyncConfig {

+    private ExecutorService auditExecutorService;
+    private ExecutorService aiStreamExecutorService;
+
    /**
     * MDC context-propagating task decorator. Copies MDC context from the caller thread to the
     * virtual thread executing the task.
@@ -44,8 +50,8 @@ public class AsyncConfig {

    @Bean(name = "auditExecutor")
    public Executor auditExecutor() {
-        TaskExecutorAdapter adapter =
-                new TaskExecutorAdapter(Executors.newVirtualThreadPerTaskExecutor());
+        auditExecutorService = Executors.newVirtualThreadPerTaskExecutor();
+        TaskExecutorAdapter adapter = new TaskExecutorAdapter(auditExecutorService);
        adapter.setTaskDecorator(new MDCContextTaskDecorator());
        return adapter;
    }
@@ -53,9 +59,25 @@ public class AsyncConfig {
    /** Propagates the request's SecurityContext onto background AI-orchestration threads. */
    @Bean(name = "aiStreamExecutor")
    public Executor aiStreamExecutor() {
-        TaskExecutorAdapter adapter =
-                new TaskExecutorAdapter(Executors.newVirtualThreadPerTaskExecutor());
+        aiStreamExecutorService = Executors.newVirtualThreadPerTaskExecutor();
+        TaskExecutorAdapter adapter = new TaskExecutorAdapter(aiStreamExecutorService);
        adapter.setTaskDecorator(new MDCContextTaskDecorator());
        return new DelegatingSecurityContextExecutor(adapter);
    }
+
+    /**
+     * Close the underlying executors because the exposed Spring adapters do not own their
+     * lifecycle.
+     */
+    @PreDestroy
+    void shutdown() {
+        shutdownExecutor(auditExecutorService);
+        shutdownExecutor(aiStreamExecutorService);
+    }
+
+    private void shutdownExecutor(ExecutorService executor) {
+        if (executor != null) {
+            executor.shutdownNow();
+        }
+    }
 }
@@ -16,6 +16,8 @@ import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.stereotype.Service;
 import org.springframework.web.client.RestClientResponseException;

+import jakarta.annotation.PreDestroy;
+
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;

@@ -79,6 +81,21 @@ public class PolicyEngine {

    private final ExecutorService asyncExecutor = ExecutorFactory.newVirtualThreadExecutor();

+    /** Stop the service-owned executor when the application context is closed or restarted. */
+    @PreDestroy
+    void shutdown() {
+        log.debug("Shutting down policy engine executor");
+        asyncExecutor.shutdown();
+        try {
+            if (!asyncExecutor.awaitTermination(5, java.util.concurrent.TimeUnit.SECONDS)) {
+                asyncExecutor.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            asyncExecutor.shutdownNow();
+        }
+    }
+
    /**
     * Submit a pipeline to run asynchronously. The handle's run id scopes a {@link TaskManager} job
     * (status/notes/results observable via the job endpoints); its future resolves when the run