diff --git a/AGENTS.md b/AGENTS.md index 9afdad5937..7e529e4be9 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -453,6 +453,7 @@ The frontend is organized with a clear separation of concerns: - **CRITICAL**: Always update translations in `en-US` only - all other languages (including `en-GB`) are handled separately - Translation files are located in `frontend/editor/public/locales/` +- After changing any translation file, run `task pre-commit:fix` ## Important Notes diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java index 3d076bb527..9494317751 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/controller/PolicyController.java @@ -48,6 +48,7 @@ import stirling.software.proprietary.policy.engine.PolicyRunHandle; import stirling.software.proprietary.policy.engine.PolicyRunRegistry; import stirling.software.proprietary.policy.engine.PolicyRunner; import stirling.software.proprietary.policy.engine.PolicyValidator; +import stirling.software.proprietary.policy.ledger.ProcessedLedger; import stirling.software.proprietary.policy.model.PipelineDefinition; import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.PolicyInputs; @@ -87,6 +88,7 @@ public class PolicyController { private final PolicyManagementAuthority policyManagementAuthority; private final PolicyTriggerManager policyTriggerManager; private final PolicyOverviewService policyOverviewService; + private final ProcessedLedger processedLedger; private final List policyTriggers; private final ApplicationProperties applicationProperties; private final TempFileManager tempFileManager; @@ -352,6 +354,7 @@ public class PolicyController { boolean accessible = policyStore.get(policyId).filter(policyAccessGuard::canAccess).isPresent(); if (accessible && policyStore.delete(policyId)) { + processedLedger.clearPolicy(policyId); // Cancel any now-orphaned folder watch promptly rather than leaving the WatchKey open // until the next reconcile sweep. policyTriggerManager.notifyPoliciesChanged(); @@ -360,6 +363,25 @@ public class PolicyController { return ResponseEntity.notFound().build(); } + @DeleteMapping("/{policyId}/processed-history") + @Operation( + summary = "Clear a policy's processed-file history", + description = + "Forgets which source files this policy has already processed, so its next" + + " sweep reprocesses everything currently in its sources. Does not" + + " touch the files themselves.") + public ResponseEntity clearProcessedHistory(@PathVariable String policyId) { + requirePolicyEditingAllowed(); + // Scope to the caller's team: a policy in another team reads as not-found. + boolean accessible = + policyStore.get(policyId).filter(policyAccessGuard::canAccess).isPresent(); + if (!accessible) { + return ResponseEntity.notFound().build(); + } + processedLedger.clearPolicy(policyId); + return ResponseEntity.noContent().build(); + } + @PostMapping(value = "/{policyId}/run", consumes = MediaType.MULTIPART_FORM_DATA_VALUE) @Operation( summary = "Run a stored policy", diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java index ff5d73efb3..98a6f90096 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java @@ -35,6 +35,7 @@ import stirling.software.proprietary.policy.model.Policy; import stirling.software.proprietary.policy.model.PolicyInputs; import stirling.software.proprietary.policy.model.PolicyRun; import stirling.software.proprietary.policy.model.WaitState; +import stirling.software.proprietary.policy.output.OutputDelivery; import stirling.software.proprietary.policy.output.PolicyOutputSink; import stirling.software.proprietary.policy.progress.PolicyProgressListener; import stirling.software.proprietary.service.DownstreamEntitlementError; @@ -203,7 +204,12 @@ public class PolicyEngine { PolicyExecutionResult result = stepExecutor.execute(run.getDefinition(), inputs, listener); OutputSpec output = run.getDefinition().output(); - List outputs = sinkFor(output).deliver(runId, result.files(), output); + List outputs = + sinkFor(output) + .deliver( + new OutputDelivery(runId, run.getPolicyId()), + result.files(), + output); taskManager.setMultipleFileResults(runId, outputs); taskManager.setComplete(runId); run.complete(outputs); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java index fa2a190e8f..5a35b879c0 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java @@ -13,6 +13,7 @@ import lombok.extern.slf4j.Slf4j; import stirling.software.proprietary.policy.input.InputSource; import stirling.software.proprietary.policy.input.ResolvedInput; +import stirling.software.proprietary.policy.ledger.ProcessedLedger; import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.PipelineDefinition; import stirling.software.proprietary.policy.model.Policy; @@ -27,7 +28,8 @@ import stirling.software.proprietary.policy.source.SourceStore; /** * Turns a policy's referenced sources into runs: each {@code sourceId} is resolved live to its * persisted {@link Source}, then to an {@link InputSpec}. Triggers decide when and call - * {@link #run(Policy)}; the controller uses the supplied-input and ad-hoc entry points. + * {@link #run(Policy)}; the controller uses the supplied-input and ad-hoc entry points. A {@link + * SweepKind#FULL} sweep also reconciles the processed-file ledger against what is present. */ @Slf4j @Service @@ -39,6 +41,12 @@ public class PolicyRunner { private final List inputSources; private final SourceStore sourceStore; private final SourceDocCounter docCounter; + private final ProcessedLedger processedLedger; + + /** Full-listing sweep: resolve every source, then reconcile the ledger. */ + public List run(Policy policy) { + return run(policy, SweepKind.FULL); + } /** * Trigger entry point. Pulls every referenced source; each yielded unit becomes its own run so @@ -47,15 +55,21 @@ public class PolicyRunner { * rest. Returns the ids of the runs it started (empty when sources yielded no work), so a * manual trigger can report back which runs to follow. */ - public List run(Policy policy) { + public List run(Policy policy, SweepKind sweep) { + long sweepStart = System.currentTimeMillis(); + PolicySweep context = new PolicySweep(policy.id(), sweep, processedLedger); + List runIds = new ArrayList<>(); List sourceIds = policy.sourceIds(); if (sourceIds.isEmpty()) { - return List.of(startRun(policy, PolicyInputs.of(List.of()), unused -> {})); + // 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. + runIds.add(startRun(policy, PolicyInputs.of(List.of()), unused -> {})); } - List runIds = new ArrayList<>(); for (String sourceId : sourceIds) { Source source = sourceStore.get(sourceId).orElse(null); if (source == null) { + // No veto: a deleted source's rows should age out via the cleanup below. log.warn("Policy {} references missing source {}; skipping", policy.id(), sourceId); continue; } @@ -65,9 +79,21 @@ public class PolicyRunner { sourceId, source.name(), policy.id()); + // Veto: a paused source's files cannot be stamped, so they must not be pruned. + context.vetoCleanup(); continue; } - runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec())); + runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec(), context)); + } + if (context.cleanupAllowed()) { + processedLedger.markSeen(policy.id(), context.presentIdentities()); + int removed = processedLedger.deleteUnseen(policy.id(), sweepStart); + if (removed > 0) { + log.debug( + "Pruned {} ledger row(s) for files no longer present (policy {})", + removed, + policy.id()); + } } return runIds; } @@ -86,26 +112,33 @@ public class PolicyRunner { /** * Resolves the source and starts a run per unit; records how many documents the source fed and - * returns the ids of the runs started. + * returns the ids of the runs started. Any source that could not be listed completely vetoes + * this sweep's ledger cleanup. */ - private List pullAndRun(Policy policy, String sourceId, InputSpec spec) { + private List pullAndRun( + Policy policy, String sourceId, InputSpec spec, PolicySweep context) { InputSource source = sourceFor(spec); if (source == null) { log.warn( "No input source for type '{}' (policy {}); skipping", spec.type(), policy.id()); + context.vetoCleanup(); return List.of(); } + if (!source.listsExhaustively()) { + context.vetoCleanup(); + } List work; try { - work = source.resolve(spec); + work = source.resolve(spec, context); } catch (IOException | RuntimeException e) { log.warn( "Failed to resolve source '{}' for policy {}: {}", spec.type(), policy.id(), e.getMessage()); + context.vetoCleanup(); return List.of(); } List runIds = new ArrayList<>(); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicySweep.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicySweep.java new file mode 100644 index 0000000000..5302bacdb3 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicySweep.java @@ -0,0 +1,89 @@ +package stirling.software.proprietary.policy.engine; + +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.Supplier; + +import stirling.software.proprietary.policy.input.ResolveContext; +import stirling.software.proprietary.policy.ledger.ClaimState; +import stirling.software.proprietary.policy.ledger.ProcessedFileStatus; +import stirling.software.proprietary.policy.ledger.ProcessedLedger; + +/** + * The {@link ResolveContext} for one policy sweep: scopes ledger calls to the policy, gathers the + * present-identity union across sources, prefetches claim state in bulk so per-file claims skip + * their row lookup, and vetoes presence cleanup when any source could not be listed completely + * (pruning would wrongly forget its files). + */ +final class PolicySweep implements ResolveContext { + + private final String policyId; + private final SweepKind kind; + private final ProcessedLedger ledger; + private final Set present = new HashSet<>(); + // Claim states loaded in bulk at reportPresent; a claim outside the prefetch falls back to a + // single lookup. A stale entry cannot double-claim (the ledger re-checks every transition), + // it can only defer a file to the next sweep. + private final Map prefetched = new HashMap<>(); + private final Set prefetchedIdentities = new HashSet<>(); + private boolean cleanupVetoed; + + PolicySweep(String policyId, SweepKind kind, ProcessedLedger ledger) { + this.policyId = policyId; + this.kind = kind; + this.ledger = ledger; + } + + @Override + public synchronized boolean claim(String identity, String gate, Supplier contentHash) { + ClaimState observed = + prefetchedIdentities.contains(identity) + ? prefetched.get(identity) + : ledger.statesFor(policyId, List.of(identity)).get(identity); + boolean claimed = ledger.claim(policyId, identity, gate, contentHash, observed); + if (claimed) { + // A nested source surfacing the same file later in this sweep sees it in flight + // without another lookup. + prefetchedIdentities.add(identity); + prefetched.put(identity, new ClaimState(ProcessedFileStatus.PROCESSING, gate, null)); + } + return claimed; + } + + @Override + public void settle( + String identity, String finalGate, String finalContentHash, boolean success) { + ledger.settle(policyId, identity, finalGate, finalContentHash, success); + } + + @Override + public boolean allSettledDone(String identity) { + // Deliberately not policy-scoped: consume deletion needs every claimant's consensus. + return ledger.allSettledDone(identity); + } + + @Override + public synchronized void reportPresent(Collection identities) { + if (kind == SweepKind.FULL) { + present.addAll(identities); + } + prefetched.putAll(ledger.statesFor(policyId, identities)); + prefetchedIdentities.addAll(identities); + } + + synchronized void vetoCleanup() { + cleanupVetoed = true; + } + + synchronized boolean cleanupAllowed() { + return kind == SweepKind.FULL && !cleanupVetoed; + } + + synchronized Set presentIdentities() { + return Set.copyOf(present); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/SweepKind.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/SweepKind.java new file mode 100644 index 0000000000..c2455749ae --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/SweepKind.java @@ -0,0 +1,10 @@ +package stirling.software.proprietary.policy.engine; + +/** + * How thorough a policy sweep is: {@link #FULL} (complete listing; also stamps presence and prunes + * the ledger) or {@link #LIGHT} (event-driven; claims only, cost proportional to what changed). + */ +public enum SweepKind { + FULL, + LIGHT +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/FolderInputSource.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/FolderInputSource.java index c00f5b4a33..9270cc0c1b 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/FolderInputSource.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/FolderInputSource.java @@ -1,12 +1,17 @@ package stirling.software.proprietary.policy.input; import java.io.IOException; +import java.io.UncheckedIOException; +import java.nio.file.FileVisitResult; import java.nio.file.Files; +import java.nio.file.NoSuchFileException; import java.nio.file.Path; -import java.nio.file.StandardCopyOption; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.attribute.BasicFileAttributes; import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.function.Supplier; import java.util.stream.Stream; import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; @@ -19,17 +24,20 @@ import lombok.extern.slf4j.Slf4j; import stirling.software.common.util.FileReadinessChecker; import stirling.software.proprietary.policy.config.FolderAccessGuard; +import stirling.software.proprietary.policy.ledger.FolderIdentities; import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.PolicyInputs; /** - * Reads input files from a directory; each ready file is its own unit of work so one failure does - * not affect the others. - * - *

Mode option: "consume" (default) claims each file by moving it into {@code - * .stirling/processing} then routes it to {@code .stirling/done} or {@code .stirling/error}, so - * each file runs once; "snapshot" reads without moving, so every run sees the full set. Readiness - * is checked first so files mid-write are skipped. + * Reads input files from a directory; each ready file is its own unit of work, claimed through the + * {@link ResolveContext} ledger rather than moved aside, so nothing accumulates in a work + * directory. Options: "mode" is "consume" (default: a processed file is removed once every policy + * that claimed it has settled successfully and it is still the version that ran; failures stay in + * place and are not retried until they change) or "snapshot" (stateless, every run sees the full + * set); "recursive" descends into subdirectories; "identity" is "stat" (default, any size/mtime + * change is a new version) or "hash" (content-verified, so a touch does not reprocess). Hidden + * files and directories, including the legacy {@code .stirling} work dir, are never picked up, and + * files mid-write are skipped by the readiness check. */ @Slf4j @Service @@ -38,11 +46,6 @@ import stirling.software.proprietary.policy.model.PolicyInputs; public class FolderInputSource implements InputSource { private static final String TYPE = FolderAccessGuard.FOLDER_TYPE; - // Bookkeeping lives under one hidden dir so the watched folder stays tidy. - private static final String WORK_SUBDIR = ".stirling"; - private static final String PROCESSING_SUBDIR = "processing"; - private static final String DONE_SUBDIR = "done"; - private static final String ERROR_SUBDIR = "error"; private final FileReadinessChecker readinessChecker; private final FolderAccessGuard accessGuard; @@ -68,70 +71,195 @@ public class FolderInputSource implements InputSource { } @Override - public List resolve(InputSpec spec) throws IOException { + public List resolve(InputSpec spec, ResolveContext ctx) throws IOException { FolderConfig config = FolderConfig.from(spec.options()); Path inputDir = accessGuard.requirePermitted(config.directory()); if (!Files.isDirectory(inputDir)) { - log.debug("Folder input dir does not exist: {}", inputDir); - return List.of(); + // Fail rather than return empty: an unmounted drive must read as "could not list", + // which vetoes the sweep's presence cleanup, not as "verifiably no files", which + // would wipe the policy's history and reprocess everything on remount. + throw new NoSuchFileException( + inputDir.toString(), null, "input directory does not exist"); + } + Path canonicalDir = FolderIdentities.canonicalDir(inputDir); + List present = listFiles(inputDir, config.recursive()); + + if (config.snapshot()) { + List work = new ArrayList<>(); + for (Path file : present) { + if (readinessChecker.isReady(file)) { + work.add(ResolvedInput.of(PolicyInputs.of(List.of(fileResource(file))))); + } + } + return work; } - List ready = new ArrayList<>(); - try (Stream entries = Files.list(inputDir)) { - entries.filter(Files::isRegularFile) - .filter(readinessChecker::isReady) - .forEach(ready::add); - } + ctx.reportPresent( + present.stream() + .map(file -> FolderIdentities.identity(canonicalDir, inputDir, file)) + .toList()); List work = new ArrayList<>(); - for (Path file : ready) { - if (config.snapshot()) { - work.add(ResolvedInput.of(PolicyInputs.of(List.of(fileResource(file))))); - } else { - Path claimed = claim(inputDir, file); - if (claimed == null) { - continue; // another sweep/process grabbed it - } - work.add( - new ResolvedInput( - PolicyInputs.of(List.of(fileResource(claimed))), - success -> route(inputDir, claimed, success))); + for (Path file : present) { + if (!readinessChecker.isReady(file)) { + continue; } + String identity = FolderIdentities.identity(canonicalDir, inputDir, file); + MemoizedContentHash contentHash = + config.hashIdentity() ? new MemoizedContentHash(file) : null; + String gate; + boolean claimed; + try { + gate = FolderIdentities.statGate(file); + claimed = ctx.claim(identity, gate, contentHash); + } catch (IOException | UncheckedIOException e) { + log.debug("Could not read {} for its version: {}", file, e.getMessage()); + continue; // vanished or unreadable mid-sweep; the next sweep sees the truth + } + if (!claimed) { + continue; + } + work.add( + new ResolvedInput( + PolicyInputs.of(List.of(fileResource(file))), + success -> + completeConsumed( + ctx, identity, file, gate, contentHash, success))); } return work; } - // Atomic move into processing/: only one sweep can win the claim, the rest see the file gone. - private Path claim(Path inputDir, Path file) { + /** + * Settle at the version this run claimed - never a re-read, so a file replaced mid-run reads as + * a new unclaimed version next sweep instead of being marked processed. Then remove the input + * only when it is still the processed version (a mid-run replacement must survive) and every + * policy that claimed it has settled DONE, so co-watching policies all read the original and + * one failure parks the file for everyone. A failed run settles ERROR and never deletes; the + * DONE row of a file that could not be deleted still stops reprocessing. + */ + private static void completeConsumed( + ResolveContext ctx, + String identity, + Path file, + String claimGate, + MemoizedContentHash contentHash, + boolean success) { + ctx.settle(identity, claimGate, claimedHash(file, claimGate, contentHash), success); + if (!success) { + return; + } try { - Path processingDir = workDir(inputDir, PROCESSING_SUBDIR); - Files.createDirectories(processingDir); - Path claimed = uniqueTarget(processingDir, file.getFileName().toString()); - Files.move(file, claimed, StandardCopyOption.ATOMIC_MOVE); - return claimed; + if (FolderIdentities.statGate(file).equals(claimGate) && ctx.allSettledDone(identity)) { + Files.deleteIfExists(file); + } + } catch (NoSuchFileException alreadyGone) { + // Removed by the user or a co-watching policy's own consensus delete: nothing to do. } catch (IOException e) { - log.debug("Could not claim {}: {}", file, e.getMessage()); + log.warn("Could not remove consumed input {}: {}", file, e.getMessage()); + } + } + + /** + * The claimed version's content hash: the value computed during the claim when the ledger + * consulted the verifier, else computed now while the file is still at the claimed gate (so the + * hash describes what actually ran), else null. Always null in stat mode. + */ + private static String claimedHash(Path file, String claimGate, MemoizedContentHash hash) { + if (hash == null) { return null; } + String computed = hash.valueIfComputed(); + if (computed != null) { + return computed; + } + try { + if (FolderIdentities.statGate(file).equals(claimGate)) { + return hash.get(); + } + } catch (IOException | UncheckedIOException e) { + log.debug("Could not hash {} at settle: {}", file, e.getMessage()); + } + return null; } - private void route(Path inputDir, Path claimed, boolean success) { - String subdir = success ? DONE_SUBDIR : ERROR_SUBDIR; - try { - Path destDir = workDir(inputDir, subdir); - Files.createDirectories(destDir); - Files.move( - claimed, - uniqueTarget(destDir, claimed.getFileName().toString()), - StandardCopyOption.ATOMIC_MOVE); - } catch (IOException e) { - log.warn( - "Could not move processed input {} to {}: {}", claimed, subdir, e.getMessage()); + /** Lazy verification tier: invoked at most once by the ledger, retained for the settle. */ + private static final class MemoizedContentHash implements Supplier { + + private final Path file; + private volatile String value; + + private MemoizedContentHash(Path file) { + this.file = file; + } + + @Override + public String get() { + if (value == null) { + try { + value = FolderIdentities.contentHash(file); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + return value; + } + + String valueIfComputed() { + return value; } } - private static Path workDir(Path inputDir, String subdir) { - return inputDir.resolve(WORK_SUBDIR).resolve(subdir); + /** Every non-hidden regular file in the source, readable or not. */ + private static List listFiles(Path inputDir, boolean recursive) throws IOException { + List files = new ArrayList<>(); + if (!recursive) { + try (Stream entries = Files.list(inputDir)) { + entries.filter(Files::isRegularFile) + .filter(file -> !hidden(file)) + .forEach(files::add); + } + return files; + } + // Hidden subtrees are pruned wholesale; symlinked directories are not followed. + Files.walkFileTree( + inputDir, + new SimpleFileVisitor<>() { + @Override + public FileVisitResult preVisitDirectory( + Path dir, BasicFileAttributes attributes) { + if (!dir.equals(inputDir) && hidden(dir)) { + return FileVisitResult.SKIP_SUBTREE; + } + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult visitFile(Path file, BasicFileAttributes attributes) { + if (attributes.isRegularFile() && !hidden(file)) { + files.add(file); + } + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult visitFileFailed(Path file, IOException e) { + log.debug("Skipping unreadable entry {}: {}", file, e.getMessage()); + return FileVisitResult.CONTINUE; + } + }); + return files; + } + + private static boolean hidden(Path path) { + Path name = path.getFileName(); + if (name != null && name.toString().startsWith(".")) { + return true; + } + try { + return Files.isHidden(path); + } catch (IOException e) { + return false; + } } private static Resource fileResource(Path path) { @@ -144,27 +272,15 @@ public class FolderInputSource implements InputSource { }; } - private static Path uniqueTarget(Path dir, String filename) { - Path candidate = dir.resolve(filename); - if (!Files.exists(candidate)) { - return candidate; - } - int dot = filename.lastIndexOf('.'); - String base = dot < 0 ? filename : filename.substring(0, dot); - String ext = dot < 0 ? "" : filename.substring(dot); - for (int n = 1; ; n++) { - Path next = dir.resolve(base + " (" + n + ")" + ext); - if (!Files.exists(next)) { - return next; - } - } - } - - record FolderConfig(Path directory, boolean snapshot) { + record FolderConfig(Path directory, boolean snapshot, boolean recursive, boolean hashIdentity) { private static final String DIRECTORY_OPTION = "directory"; private static final String MODE_OPTION = "mode"; private static final String MODE_SNAPSHOT = "snapshot"; + private static final String RECURSIVE_OPTION = "recursive"; + private static final String IDENTITY_OPTION = "identity"; + private static final String IDENTITY_STAT = "stat"; + private static final String IDENTITY_HASH = "hash"; static FolderConfig from(Map options) { Object directory = options.get(DIRECTORY_OPTION); @@ -173,7 +289,17 @@ public class FolderInputSource implements InputSource { } Object mode = options.get(MODE_OPTION); boolean snapshot = mode != null && MODE_SNAPSHOT.equals(mode.toString()); - return new FolderConfig(Path.of(directory.toString()), snapshot); + Object recursive = options.get(RECURSIVE_OPTION); + boolean recurse = recursive != null && Boolean.parseBoolean(recursive.toString()); + Object identity = options.get(IDENTITY_OPTION); + boolean hash = identity != null && IDENTITY_HASH.equals(identity.toString()); + if (identity != null + && !IDENTITY_STAT.equals(identity.toString()) + && !IDENTITY_HASH.equals(identity.toString())) { + throw new IllegalArgumentException( + "folder input 'identity' must be 'stat' or 'hash'"); + } + return new FolderConfig(Path.of(directory.toString()), snapshot, recurse, hash); } } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/InputSource.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/InputSource.java index 436f6c5263..d32c2fc546 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/InputSource.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/InputSource.java @@ -24,9 +24,21 @@ public interface InputSource { /** * Resolve the spec into zero or more units of work, each carrying one run's files and a - * completion hook. Empty list means nothing to run right now. + * completion hook. Empty list means nothing to run right now. Discovery is read-only - files + * stay where the user put them; "already processed" is tracked through {@code ctx} (claim on + * pickup, settle on completion, report what is present so stale ledger rows can be pruned). */ - List resolve(InputSpec spec) throws IOException; + List resolve(InputSpec spec, ResolveContext ctx) throws IOException; + + /** + * Whether {@link #resolve} observes everything in the source (a complete listing) rather than + * e.g. only what events surfaced. Presence cleanup of the ledger is skipped for the whole + * policy unless every enabled source says true - wrongly pruning history would reprocess a + * whole folder, while keeping a few stale rows costs nothing. + */ + default boolean listsExhaustively() { + return true; + } /** * Filesystem dirs this source draws from, for the folder-watch trigger. Advisory: resolving is diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/ResolveContext.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/ResolveContext.java new file mode 100644 index 0000000000..9282b44d5a --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/input/ResolveContext.java @@ -0,0 +1,36 @@ +package stirling.software.proprietary.policy.input; + +import java.util.Collection; +import java.util.function.Supplier; + +/** + * A source's policy-scoped window onto the processed-file ledger for one sweep. Thread-safe and + * valid for the lifetime of the work units the source issued ({@link #settle} fires from async run + * completions). + */ +public interface ResolveContext { + + /** + * Atomically claim a file at its current version; true means this sweep runs it. A null {@code + * contentHash} makes any gate change a new version; a non-null supplier is invoked at most + * once, only on a gate mismatch, and a matching hash refreshes the stored gate instead of + * reprocessing. Supplier exceptions propagate. + */ + boolean claim(String identity, String gate, Supplier contentHash); + + /** Record a claimed file's outcome at its final version ({@code finalContentHash} nullable). */ + void settle(String identity, String finalGate, String finalContentHash, boolean success); + + /** + * Whether every policy holding a ledger row for this identity has settled it DONE. Cross-policy + * by design: consume-mode deletion is a consensus of all claimants, so a shared input is + * removed only once nobody still needs it (in-flight, failed, and interrupted rows all veto). + */ + boolean allSettledDone(String identity); + + /** + * Report every identity present right now, readable or not; feeds presence cleanup of rows + * whose file is gone. + */ + void reportPresent(Collection identities); +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ClaimState.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ClaimState.java new file mode 100644 index 0000000000..4f3dc0312f --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ClaimState.java @@ -0,0 +1,9 @@ +package stirling.software.proprietary.policy.ledger; + +/** + * A row's claim-relevant state as read by {@link ProcessedLedger#statesFor}: what a sweep observed + * before deciding a claim. May be stale by the time the claim runs; every ledger transition + * re-checks the observed state in its WHERE clause, so staleness defers a claim to a later sweep + * rather than double-running one. + */ +public record ClaimState(ProcessedFileStatus status, String gate, String contentHash) {} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/FolderIdentities.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/FolderIdentities.java new file mode 100644 index 0000000000..8ece937147 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/FolderIdentities.java @@ -0,0 +1,41 @@ +package stirling.software.proprietary.policy.ledger; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.BasicFileAttributes; + +import stirling.software.proprietary.billing.ContentHasher; + +/** + * The folder backend's identity and version scheme, shared by {@code FolderInputSource} and {@code + * FolderOutputSink} so outputs are recorded under exactly the identity the next scan derives. + * Directories are canonicalised with {@code toRealPath()} so symlinked aliases agree. + */ +public final class FolderIdentities { + + private FolderIdentities() {} + + /** Canonical form of a configured directory; resolves symlinks, so the dir must exist. */ + public static Path canonicalDir(Path dir) throws IOException { + return dir.toRealPath(); + } + + /** Identity of {@code file} under {@code dir}: its path re-rooted onto the canonical dir. */ + public static String identity(Path canonicalDir, Path dir, Path file) { + return canonicalDir.resolve(dir.relativize(file)).normalize().toString(); + } + + /** The cheap version gate: a change to content length or mtime means "look closer". */ + public static String statGate(Path file) throws IOException { + BasicFileAttributes attributes = Files.readAttributes(file, BasicFileAttributes.class); + return attributes.size() + ":" + attributes.lastModifiedTime().toMillis(); + } + + /** + * The strong version token: distinguishes a real change from a touch, at the cost of a read. + */ + public static String contentHash(Path file) throws IOException { + return ContentHasher.sha256(file); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/IdentityHasher.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/IdentityHasher.java new file mode 100644 index 0000000000..32c30c0254 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/IdentityHasher.java @@ -0,0 +1,19 @@ +package stirling.software.proprietary.policy.ledger; + +import java.nio.charset.StandardCharsets; + +import stirling.software.proprietary.billing.ContentHasher; + +/** + * Fixed-width key form of a source-owned identity, so any identity length fits the ledger's primary + * key. Backend-agnostic: every source type's identities are keyed through here, which is why this + * does not live with the folder backend's {@link FolderIdentities}. + */ +public final class IdentityHasher { + + private IdentityHasher() {} + + public static String identityHash(String identity) { + return ContentHasher.sha256(identity.getBytes(StandardCharsets.UTF_8)); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/InProcessProcessedLedger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/InProcessProcessedLedger.java new file mode 100644 index 0000000000..436e62f2c8 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/InProcessProcessedLedger.java @@ -0,0 +1,229 @@ +package stirling.software.proprietary.policy.ledger; + +import java.util.Collection; +import java.util.HashMap; +import java.util.Map; +import java.util.Objects; +import java.util.function.Supplier; + +/** + * In-memory {@link ProcessedLedger} for tests and DB-less wiring; kept semantically identical to + * {@code JpaProcessedLedger} by the shared contract test. + */ +public class InProcessProcessedLedger implements ProcessedLedger { + + private final Map> rowsByPolicy = new HashMap<>(); + private final Supplier nowMillis; + + public InProcessProcessedLedger() { + this(System::currentTimeMillis); + } + + public InProcessProcessedLedger(Supplier nowMillis) { + this.nowMillis = nowMillis; + } + + @Override + public synchronized Map statesFor( + String policyId, Collection identities) { + Map rows = rowsByPolicy.getOrDefault(policyId, Map.of()); + Map states = new HashMap<>(); + for (String identity : identities) { + Row row = rows.get(identity); + if (row != null) { + states.put(identity, new ClaimState(row.status, row.gate, row.contentHash)); + } + } + return states; + } + + // Single-lock store: the live row is never staler than any observed snapshot, so decide + // against it directly; the conditional updates of the JPA ledger yield the same outcomes. + @Override + public synchronized boolean claim( + String policyId, + String identity, + String gate, + Supplier contentHash, + ClaimState observed) { + Map rows = rowsByPolicy.computeIfAbsent(policyId, key -> new HashMap<>()); + long now = nowMillis.get(); + Row row = rows.get(identity); + if (row == null) { + String hash = contentHash == null ? null : contentHash.get(); + rows.put(identity, new Row(gate, hash, ProcessedFileStatus.PROCESSING, 1, now)); + return true; + } + if (row.status == ProcessedFileStatus.PROCESSING) { + return false; + } + if (gate.equals(row.gate)) { + if (row.status == ProcessedFileStatus.INTERRUPTED && row.attempts < MAX_ATTEMPTS) { + row.status = ProcessedFileStatus.PROCESSING; + row.attempts++; + row.lastSeen = now; + return true; + } + return false; + } + if (contentHash == null) { + row.gate = gate; + row.contentHash = null; + row.status = ProcessedFileStatus.PROCESSING; + row.attempts = 1; + row.lastSeen = now; + return true; + } + String hash = contentHash.get(); + if (Objects.equals(hash, row.contentHash)) { + if (row.status == ProcessedFileStatus.INTERRUPTED && row.attempts < MAX_ATTEMPTS) { + row.gate = gate; + row.status = ProcessedFileStatus.PROCESSING; + row.attempts++; + row.lastSeen = now; + return true; + } + if (row.status != ProcessedFileStatus.INTERRUPTED) { + row.gate = gate; + row.lastSeen = now; + } + return false; + } + row.gate = gate; + row.contentHash = hash; + row.status = ProcessedFileStatus.PROCESSING; + row.attempts = 1; + row.lastSeen = now; + return true; + } + + @Override + public synchronized void settle( + String policyId, + String identity, + String finalGate, + String finalContentHash, + boolean success) { + upsertSettled( + policyId, + identity, + finalGate, + finalContentHash, + success ? ProcessedFileStatus.DONE : ProcessedFileStatus.ERROR); + } + + @Override + public synchronized void recordOutput( + String policyId, String identity, String gate, String contentHash) { + upsertSettled(policyId, identity, gate, contentHash, ProcessedFileStatus.DONE); + } + + @Override + public synchronized void forgetOutput(String policyId, String identity, String gate) { + Map rows = rowsByPolicy.get(policyId); + if (rows == null) { + return; + } + Row row = rows.get(identity); + if (row != null && row.status == ProcessedFileStatus.DONE && gate.equals(row.gate)) { + rows.remove(identity); + } + } + + private void upsertSettled( + String policyId, + String identity, + String gate, + String contentHash, + ProcessedFileStatus status) { + Map rows = rowsByPolicy.computeIfAbsent(policyId, key -> new HashMap<>()); + long now = nowMillis.get(); + Row row = rows.get(identity); + if (row == null) { + rows.put(identity, new Row(gate, contentHash, status, 1, now)); + return; + } + row.gate = gate; + row.contentHash = contentHash; + row.status = status; + row.lastSeen = now; + } + + @Override + public synchronized boolean allSettledDone(String identity) { + for (Map rows : rowsByPolicy.values()) { + Row row = rows.get(identity); + if (row != null && row.status != ProcessedFileStatus.DONE) { + return false; + } + } + return true; + } + + @Override + public synchronized void markSeen(String policyId, Collection identities) { + Map rows = rowsByPolicy.get(policyId); + if (rows == null) { + return; + } + long now = nowMillis.get(); + for (String identity : identities) { + Row row = rows.get(identity); + if (row != null) { + row.lastSeen = now; + } + } + } + + @Override + public synchronized int deleteUnseen(String policyId, long seenSinceMillis) { + Map rows = rowsByPolicy.get(policyId); + if (rows == null) { + return 0; + } + int before = rows.size(); + rows.values() + .removeIf( + row -> + row.lastSeen < seenSinceMillis + && row.status != ProcessedFileStatus.PROCESSING); + return before - rows.size(); + } + + @Override + public synchronized void clearPolicy(String policyId) { + rowsByPolicy.remove(policyId); + } + + @Override + public synchronized void recoverInterrupted() { + for (Map rows : rowsByPolicy.values()) { + for (Row row : rows.values()) { + if (row.status == ProcessedFileStatus.PROCESSING) { + row.status = ProcessedFileStatus.INTERRUPTED; + } + } + } + } + + private static final class Row { + private String gate; + private String contentHash; + private ProcessedFileStatus status; + private int attempts; + private long lastSeen; + + private Row( + String gate, + String contentHash, + ProcessedFileStatus status, + int attempts, + long lastSeen) { + this.gate = gate; + this.contentHash = contentHash; + this.status = status; + this.attempts = attempts; + this.lastSeen = lastSeen; + } + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/JpaProcessedLedger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/JpaProcessedLedger.java new file mode 100644 index 0000000000..43f0e6795d --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/JpaProcessedLedger.java @@ -0,0 +1,212 @@ +package stirling.software.proprietary.policy.ledger; + +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.event.EventListener; +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.stereotype.Service; + +import lombok.extern.slf4j.Slf4j; + +/** + * Durable {@link ProcessedLedger}; the runtime bean. A fresh claim is a flushed insert so a + * concurrent winner surfaces as a constraint violation; every other transition is a conditional + * update that re-checks the observed state, so a lost race reports 0 rows and the caller skips. + * Boot recovery assumes the single node the folder-watch trigger assumes: runs live in memory, so + * after a restart every PROCESSING row is stale. + */ +@Slf4j +@Service +@ConditionalOnBooleanProperty(name = "policies.enabled") +public class JpaProcessedLedger implements ProcessedLedger { + + private static final int STAMP_CHUNK = 500; + + private final ProcessedFileRepository repository; + private final Supplier nowMillis; + + @Autowired + public JpaProcessedLedger(ProcessedFileRepository repository) { + this(repository, System::currentTimeMillis); + } + + // Clock seam so tests can pin "now"; the runtime bean uses the wall clock above. + JpaProcessedLedger(ProcessedFileRepository repository, Supplier nowMillis) { + this.repository = repository; + this.nowMillis = nowMillis; + } + + @Override + public Map statesFor(String policyId, Collection identities) { + if (identities.isEmpty()) { + return Map.of(); + } + Map identityByHash = new HashMap<>(); + for (String identity : identities) { + identityByHash.put(IdentityHasher.identityHash(identity), identity); + } + Map states = new HashMap<>(); + List hashes = List.copyOf(identityByHash.keySet()); + for (int from = 0; from < hashes.size(); from += STAMP_CHUNK) { + List rows = + repository.findByPolicyIdAndIdentityHashIn( + policyId, + hashes.subList(from, Math.min(from + STAMP_CHUNK, hashes.size()))); + for (ProcessedFileEntity row : rows) { + states.put( + identityByHash.get(row.getIdentityHash()), + new ClaimState(row.getStatus(), row.getSignature(), row.getContentHash())); + } + } + return states; + } + + @Override + public boolean claim( + String policyId, + String identity, + String gate, + Supplier contentHash, + ClaimState observed) { + String identityHash = IdentityHasher.identityHash(identity); + long now = nowMillis.get(); + if (observed == null) { + try { + repository.saveAndFlush( + new ProcessedFileEntity( + policyId, + identityHash, + identity, + gate, + contentHash == null ? null : contentHash.get(), + ProcessedFileStatus.PROCESSING, + now)); + return true; + } catch (DataIntegrityViolationException concurrentClaim) { + return false; + } + } + if (observed.status() == ProcessedFileStatus.PROCESSING) { + return false; + } + if (gate.equals(observed.gate())) { + if (observed.status() == ProcessedFileStatus.INTERRUPTED) { + return repository.retryInterruptedAtGate( + policyId, identityHash, gate, MAX_ATTEMPTS, now) + > 0; + } + return false; + } + if (contentHash == null) { + return repository.reclaimAtNewGate(policyId, identityHash, gate, now) > 0; + } + String hash = contentHash.get(); + if (hash.equals(observed.contentHash())) { + if (observed.status() == ProcessedFileStatus.INTERRUPTED) { + return repository.retryInterruptedSameContent( + policyId, identityHash, gate, hash, MAX_ATTEMPTS, now) + > 0; + } + repository.refreshGate(policyId, identityHash, gate, hash, now); + return false; + } + return repository.reclaimAtNewContent(policyId, identityHash, gate, hash, now) > 0; + } + + @Override + public void settle( + String policyId, + String identity, + String finalGate, + String finalContentHash, + boolean success) { + upsertSettled( + policyId, + identity, + finalGate, + finalContentHash, + success ? ProcessedFileStatus.DONE : ProcessedFileStatus.ERROR); + } + + @Override + public void recordOutput(String policyId, String identity, String gate, String contentHash) { + upsertSettled(policyId, identity, gate, contentHash, ProcessedFileStatus.DONE); + } + + @Override + public void forgetOutput(String policyId, String identity, String gate) { + repository.deleteDoneAt(policyId, IdentityHasher.identityHash(identity), gate); + } + + /** + * Settle-or-insert: the row may have been presence-cleaned mid-run, and an output row may be + * brand new. + */ + private void upsertSettled( + String policyId, + String identity, + String gate, + String contentHash, + ProcessedFileStatus status) { + String identityHash = IdentityHasher.identityHash(identity); + long now = nowMillis.get(); + if (repository.settle(policyId, identityHash, gate, contentHash, status, now) > 0) { + return; + } + try { + ProcessedFileEntity row = + new ProcessedFileEntity( + policyId, identityHash, identity, gate, contentHash, status, now); + repository.saveAndFlush(row); + } catch (DataIntegrityViolationException concurrentInsert) { + repository.settle(policyId, identityHash, gate, contentHash, status, now); + } + } + + @Override + public boolean allSettledDone(String identity) { + return !repository.existsByIdentityHashAndStatusNot( + IdentityHasher.identityHash(identity), ProcessedFileStatus.DONE); + } + + @Override + public void markSeen(String policyId, Collection identities) { + if (identities.isEmpty()) { + return; + } + List hashes = identities.stream().map(IdentityHasher::identityHash).toList(); + long now = nowMillis.get(); + for (int from = 0; from < hashes.size(); from += STAMP_CHUNK) { + repository.stampSeen( + policyId, + hashes.subList(from, Math.min(from + STAMP_CHUNK, hashes.size())), + now); + } + } + + @Override + public int deleteUnseen(String policyId, long seenSinceMillis) { + return repository.deleteUnseen(policyId, seenSinceMillis); + } + + @Override + public void clearPolicy(String policyId) { + repository.deleteByPolicy(policyId); + } + + @Override + @EventListener(ApplicationReadyEvent.class) + public void recoverInterrupted() { + int recovered = repository.markAllProcessingInterrupted(nowMillis.get()); + if (recovered > 0) { + log.info("Recovered {} policy input file(s) interrupted by shutdown", recovered); + } + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileEntity.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileEntity.java new file mode 100644 index 0000000000..b05c8b1eec --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileEntity.java @@ -0,0 +1,97 @@ +package stirling.software.proprietary.policy.ledger; + +import java.io.Serializable; + +import org.springframework.data.domain.Persistable; + +import jakarta.persistence.Column; +import jakarta.persistence.Entity; +import jakarta.persistence.EnumType; +import jakarta.persistence.Enumerated; +import jakarta.persistence.Id; +import jakarta.persistence.IdClass; +import jakarta.persistence.Table; +import jakarta.persistence.Transient; + +import lombok.Getter; +import lombok.NoArgsConstructor; +import lombok.Setter; + +/** + * One processed-file ledger row: the version a policy last settled a file at, and where it is in + * the claim lifecycle. Keyed by SHA-256 of the source-owned identity so any identity length fits a + * fixed-width index. {@code isNew} is always true: the entity is only saved for fresh inserts + * (everything else is a conditional update), so a lost insert race surfaces as a constraint + * violation rather than a silent merge. + */ +@Entity +@Table(name = "policy_processed_files") +@IdClass(ProcessedFileId.class) +@NoArgsConstructor +@Getter +@Setter +public class ProcessedFileEntity implements Serializable, Persistable { + + private static final long serialVersionUID = 1L; + + @Id + @Column(name = "policy_id") + private String policyId; + + @Id + @Column(name = "identity_hash", length = 64) + private String identityHash; + + @Column(name = "identity", length = 4096) + private String identity; + + @Column(name = "signature") + private String signature; + + @Column(name = "content_hash", length = 64) + private String contentHash; + + @Enumerated(EnumType.STRING) + @Column(name = "status", length = 16) + private ProcessedFileStatus status; + + @Column(name = "attempts") + private int attempts; + + @Column(name = "last_seen") + private long lastSeen; + + @Column(name = "updated_at") + private long updatedAt; + + public ProcessedFileEntity( + String policyId, + String identityHash, + String identity, + String signature, + String contentHash, + ProcessedFileStatus status, + long nowMillis) { + this.policyId = policyId; + this.identityHash = identityHash; + this.identity = identity; + this.signature = signature; + this.contentHash = contentHash; + this.status = status; + this.attempts = 1; + this.lastSeen = nowMillis; + this.updatedAt = nowMillis; + } + + @Override + @Transient + public ProcessedFileId getId() { + return new ProcessedFileId(policyId, identityHash); + } + + @Override + @Transient + public boolean isNew() { + return true; + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileId.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileId.java new file mode 100644 index 0000000000..fd3d554022 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileId.java @@ -0,0 +1,37 @@ +package stirling.software.proprietary.policy.ledger; + +import java.io.Serializable; +import java.util.Objects; + +/** Composite key for {@link ProcessedFileEntity}: one row per policy per file identity. */ +public class ProcessedFileId implements Serializable { + + private static final long serialVersionUID = 1L; + + private String policyId; + private String identityHash; + + public ProcessedFileId() {} + + public ProcessedFileId(String policyId, String identityHash) { + this.policyId = policyId; + this.identityHash = identityHash; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof ProcessedFileId other)) { + return false; + } + return Objects.equals(policyId, other.policyId) + && Objects.equals(identityHash, other.identityHash); + } + + @Override + public int hashCode() { + return Objects.hash(policyId, identityHash); + } +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileRepository.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileRepository.java new file mode 100644 index 0000000000..def89821ae --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileRepository.java @@ -0,0 +1,195 @@ +package stirling.software.proprietary.policy.ledger; + +import java.util.Collection; +import java.util.List; + +import org.springframework.data.jpa.repository.JpaRepository; +import org.springframework.data.jpa.repository.Modifying; +import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; +import org.springframework.stereotype.Repository; +import org.springframework.transaction.annotation.Transactional; + +/** + * Conditional updates for the processed-file ledger: each claim variant re-checks in its WHERE + * clause the state it was decided against, so a racing claim loses cleanly with 0 rows updated. + * Transactional per call so the ledger can run them without an enclosing transaction. + */ +@Repository +public interface ProcessedFileRepository + extends JpaRepository { + + /** + * Re-claim a settled row at a new gate without content verification; clears the stored hash, + * which described content this claim never checked. + */ + @Modifying + @Transactional + @Query( + "update ProcessedFileEntity e set e.status =" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING," + + " e.signature = :gate, e.contentHash = null, e.attempts = 1," + + " e.lastSeen = :now, e.updatedAt = :now" + + " where e.policyId = :policyId and e.identityHash = :identityHash" + + " and e.status <>" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING" + + " and e.signature <> :gate") + int reclaimAtNewGate( + @Param("policyId") String policyId, + @Param("identityHash") String identityHash, + @Param("gate") String gate, + @Param("now") long now); + + /** Re-claim a settled row whose content verifiably changed (or was never hashed). */ + @Modifying + @Transactional + @Query( + "update ProcessedFileEntity e set e.status =" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING," + + " e.signature = :gate, e.contentHash = :contentHash, e.attempts = 1," + + " e.lastSeen = :now, e.updatedAt = :now" + + " where e.policyId = :policyId and e.identityHash = :identityHash" + + " and e.status <>" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING" + + " and (e.contentHash is null or e.contentHash <> :contentHash)") + int reclaimAtNewContent( + @Param("policyId") String policyId, + @Param("identityHash") String identityHash, + @Param("gate") String gate, + @Param("contentHash") String contentHash, + @Param("now") long now); + + /** The gate moved but the content did not: track the new gate without changing status. */ + @Modifying + @Transactional + @Query( + "update ProcessedFileEntity e set e.signature = :gate, e.lastSeen = :now," + + " e.updatedAt = :now" + + " where e.policyId = :policyId and e.identityHash = :identityHash" + + " and e.status <>" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING" + + " and e.contentHash = :contentHash and e.signature <> :gate") + int refreshGate( + @Param("policyId") String policyId, + @Param("identityHash") String identityHash, + @Param("gate") String gate, + @Param("contentHash") String contentHash, + @Param("now") long now); + + /** Bounded retry of an INTERRUPTED row at the same gate. */ + @Modifying + @Transactional + @Query( + "update ProcessedFileEntity e set e.status =" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING," + + " e.attempts = e.attempts + 1, e.lastSeen = :now, e.updatedAt = :now" + + " where e.policyId = :policyId and e.identityHash = :identityHash" + + " and e.status =" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.INTERRUPTED" + + " and e.signature = :gate and e.attempts < :maxAttempts") + int retryInterruptedAtGate( + @Param("policyId") String policyId, + @Param("identityHash") String identityHash, + @Param("gate") String gate, + @Param("maxAttempts") int maxAttempts, + @Param("now") long now); + + /** Bounded retry of an INTERRUPTED row whose gate moved but whose content is unchanged. */ + @Modifying + @Transactional + @Query( + "update ProcessedFileEntity e set e.status =" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING," + + " e.signature = :gate, e.attempts = e.attempts + 1, e.lastSeen = :now," + + " e.updatedAt = :now" + + " where e.policyId = :policyId and e.identityHash = :identityHash" + + " and e.status =" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.INTERRUPTED" + + " and e.contentHash = :contentHash and e.attempts < :maxAttempts") + int retryInterruptedSameContent( + @Param("policyId") String policyId, + @Param("identityHash") String identityHash, + @Param("gate") String gate, + @Param("contentHash") String contentHash, + @Param("maxAttempts") int maxAttempts, + @Param("now") long now); + + /** + * Unconditional settle (only the claiming run settles a row); returns 0 when the row was + * removed mid-run so the caller re-inserts. + */ + @Modifying + @Transactional + @Query( + "update ProcessedFileEntity e set e.status = :status, e.signature = :gate," + + " e.contentHash = :contentHash, e.lastSeen = :now, e.updatedAt = :now" + + " where e.policyId = :policyId and e.identityHash = :identityHash") + int settle( + @Param("policyId") String policyId, + @Param("identityHash") String identityHash, + @Param("gate") String gate, + @Param("contentHash") String contentHash, + @Param("status") ProcessedFileStatus status, + @Param("now") long now); + + /** Whether any policy's row at this identity is in a state other than {@code status}. */ + boolean existsByIdentityHashAndStatusNot(String identityHash, ProcessedFileStatus status); + + /** One policy's rows across a chunk of identity hashes, for a sweep's claim snapshot. */ + List findByPolicyIdAndIdentityHashIn( + String policyId, Collection identityHashes); + + /** + * Remove an output record whose rename never landed, only while still settled exactly as + * recorded; a row a claim has since taken over is left alone. + */ + @Modifying + @Transactional + @Query( + "delete from ProcessedFileEntity e where e.policyId = :policyId" + + " and e.identityHash = :identityHash and e.signature = :gate" + + " and e.status =" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.DONE") + int deleteDoneAt( + @Param("policyId") String policyId, + @Param("identityHash") String identityHash, + @Param("gate") String gate); + + /** Stamp presence for the given identities; chunked by the caller for very large folders. */ + @Modifying + @Transactional + @Query( + "update ProcessedFileEntity e set e.lastSeen = :now" + + " where e.policyId = :policyId and e.identityHash in :identityHashes") + int stampSeen( + @Param("policyId") String policyId, + @Param("identityHashes") Collection identityHashes, + @Param("now") long now); + + /** + * Presence cleanup: remove rows not stamped since the sweep began, keeping in-flight claims. + */ + @Modifying + @Transactional + @Query( + "delete from ProcessedFileEntity e where e.policyId = :policyId" + + " and e.lastSeen < :cutoff and e.status <>" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING") + int deleteUnseen(@Param("policyId") String policyId, @Param("cutoff") long cutoff); + + @Modifying + @Transactional + @Query("delete from ProcessedFileEntity e where e.policyId = :policyId") + int deleteByPolicy(@Param("policyId") String policyId); + + /** Boot recovery: after a restart every PROCESSING row is stale (single node). */ + @Modifying + @Transactional + @Query( + "update ProcessedFileEntity e set e.status =" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.INTERRUPTED," + + " e.updatedAt = :now" + + " where e.status =" + + " stirling.software.proprietary.policy.ledger.ProcessedFileStatus.PROCESSING") + int markAllProcessingInterrupted(@Param("now") long now); +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileStatus.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileStatus.java new file mode 100644 index 0000000000..68b8b5d00c --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedFileStatus.java @@ -0,0 +1,17 @@ +package stirling.software.proprietary.policy.ledger; + +/** Lifecycle of one {@code (policy, file)} ledger row. */ +public enum ProcessedFileStatus { + + /** Claimed; a run is in flight. */ + PROCESSING, + + /** Run completed at this version. */ + DONE, + + /** Run failed; skipped until the file changes (clear-history is the manual retry). */ + ERROR, + + /** Was PROCESSING when the JVM died; retried a bounded number of times. */ + INTERRUPTED +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedLedger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedLedger.java new file mode 100644 index 0000000000..848a764f60 --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/ledger/ProcessedLedger.java @@ -0,0 +1,100 @@ +package stirling.software.proprietary.policy.ledger; + +import java.util.Collection; +import java.util.List; +import java.util.Map; +import java.util.function.Supplier; + +/** + * Remembers which files a policy has processed, one row per {@code (policy, identity)}, so sources + * track files in place. Identities are opaque source-owned strings; versions are two-tier: a cheap + * gate compared every sweep plus an optional content hash consulted only when the gate moves. + * Presence reconciliation ({@link #markSeen} + {@link #deleteUnseen}) keeps the table bounded. + */ +public interface ProcessedLedger { + + /** + * Claims of one version before an {@link ProcessedFileStatus#INTERRUPTED} row stops retrying. + */ + int MAX_ATTEMPTS = 3; + + /** + * One-query snapshot of the rows for these identities, keyed by identity; identities with no + * row are absent. Feeds the {@code observed} parameter of {@link #claim(String, String, String, + * Supplier, ClaimState)} so a sweep decides its claims without a per-file lookup. + */ + Map statesFor(String policyId, Collection identities); + + /** + * Atomically claim a file at its current version, deciding against {@code observed} (this row's + * entry from {@link #statesFor}; null means no row was seen); true means this caller runs it. A + * stale {@code observed} cannot double-claim - every transition re-checks the observed state, + * so a lost race skips until a later sweep. A null {@code contentHash} makes any gate change a + * new version; a non-null supplier is invoked at most once, only on a gate mismatch, and a + * matching hash refreshes the stored gate instead of reprocessing. Supplier exceptions + * propagate. + */ + boolean claim( + String policyId, + String identity, + String gate, + Supplier contentHash, + ClaimState observed); + + /** Snapshot-then-claim convenience for a single file; sweeps batch via {@link #statesFor}. */ + default boolean claim( + String policyId, String identity, String gate, Supplier contentHash) { + return claim( + policyId, + identity, + gate, + contentHash, + statesFor(policyId, List.of(identity)).get(identity)); + } + + /** Record a claimed file's outcome at its final version ({@code finalContentHash} nullable). */ + void settle( + String policyId, + String identity, + String finalGate, + String finalContentHash, + boolean success); + + /** + * Record a produced file as {@link ProcessedFileStatus#DONE} so the policy skips its own + * outputs. Must be called before the file is visible at this identity; other policies have no + * row and still process it. + */ + void recordOutput(String policyId, String identity, String gate, String contentHash); + + /** + * Remove an output record whose file never became visible (its rename lost the name race to a + * concurrent writer), so whatever file actually owns that identity is claimable at any version. + * A no-op unless the row is still settled exactly as recorded, so a claim that took the row + * over in the meantime is left alone. + */ + void forgetOutput(String policyId, String identity, String gate); + + /** + * Whether every row at this identity - across all policies, by design - is {@link + * ProcessedFileStatus#DONE}. Consume-mode deletion gates on this so a shared input is removed + * only once every claimant has processed it; in-flight, failed, and interrupted rows all veto, + * parking the file. Vacuously true when no rows exist. + */ + boolean allSettledDone(String identity); + + /** Stamp presence for every identity a full-listing sweep observed. */ + void markSeen(String policyId, Collection identities); + + /** + * Remove rows not seen since {@code seenSinceMillis}, keeping in-flight claims. Only call after + * every enabled source listed completely; returns the number of rows removed. + */ + int deleteUnseen(String policyId, long seenSinceMillis); + + /** Forget everything for a policy. */ + void clearPolicy(String policyId); + + /** Boot recovery: flip stale in-flight claims to {@link ProcessedFileStatus#INTERRUPTED}. */ + void recoverInterrupted(); +} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java index 8821a2940b..ed3cf707ad 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/FolderOutputSink.java @@ -2,11 +2,18 @@ package stirling.software.proprietary.policy.output; import java.io.IOException; import java.io.InputStream; +import java.nio.file.FileAlreadyExistsException; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.security.DigestOutputStream; +import java.security.MessageDigest; +import java.time.Duration; +import java.time.Instant; import java.util.ArrayList; import java.util.List; import java.util.UUID; +import java.util.stream.Stream; import org.apache.commons.io.FilenameUtils; import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty; @@ -19,14 +26,18 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import stirling.software.common.model.job.ResultFile; +import stirling.software.proprietary.billing.ContentHasher; import stirling.software.proprietary.policy.config.FolderAccessGuard; +import stirling.software.proprietary.policy.ledger.FolderIdentities; +import stirling.software.proprietary.policy.ledger.ProcessedLedger; import stirling.software.proprietary.policy.model.OutputSpec; /** - * Writes a run's outputs to the {@code directory} given in the {@link OutputSpec}. Files are - * streamed (not buffered) and uniquely named to avoid clobbering. Returned {@link ResultFile}s - * carry a synthetic id since the deliverable is the file on disk, not a {@code FileStorage} entry, - * so folder outputs are not downloadable via {@code /files/{id}}. + * Writes a run's outputs to the {@code directory} given in the {@link OutputSpec}. Each output is + * staged under a hidden {@code .stirling/tmp} dir, recorded in the processed-file ledger, then + * atomically renamed into place, so the producing policy's row exists before the file is + * discoverable and half-written outputs are never visible. Returned {@link ResultFile}s carry a + * synthetic id since the deliverable is the file on disk, not a {@code FileStorage} entry. */ @Slf4j @Service @@ -37,7 +48,11 @@ public class FolderOutputSink implements PolicyOutputSink { static final String TYPE = FolderAccessGuard.FOLDER_TYPE; static final String DIRECTORY_OPTION = "directory"; + // Staging entries are renamed away within one delivery; anything older is a crash leftover. + private static final Duration STALE_TMP_AGE = Duration.ofDays(1); + private final FolderAccessGuard accessGuard; + private final ProcessedLedger processedLedger; @Override public String type() { @@ -55,20 +70,25 @@ public class FolderOutputSink implements PolicyOutputSink { } @Override - public List deliver(String runId, List outputs, OutputSpec spec) - throws IOException { + public List deliver( + OutputDelivery delivery, List outputs, OutputSpec spec) throws IOException { Path targetDir = accessGuard.requirePermitted(directoryOf(spec)); Files.createDirectories(targetDir); + Path canonicalDir = FolderIdentities.canonicalDir(targetDir); + Path tmpDir = canonicalDir.resolve(".stirling").resolve("tmp"); + Files.createDirectories(tmpDir); + sweepStaleTmp(tmpDir); List results = new ArrayList<>(); for (int i = 0; i < outputs.size(); i++) { Resource resource = outputs.get(i); String name = safeName(resource.getFilename(), i); - Path target = uniqueTarget(targetDir, name); - try (InputStream is = resource.getInputStream()) { - Files.copy(is, target); - } - long size = Files.size(target); + Path staged = tmpDir.resolve(UUID.randomUUID().toString()); + String contentHash = stage(resource, staged, delivery.policyId() != null); + long size = Files.size(staged); + // Size and mtime survive the rename. + String gate = FolderIdentities.statGate(staged); + Path target = moveIntoPlace(delivery, canonicalDir, name, staged, gate, contentHash); String contentType = MediaTypeFactory.getMediaType(name) .orElse(MediaType.APPLICATION_OCTET_STREAM) @@ -80,11 +100,95 @@ public class FolderOutputSink implements PolicyOutputSink { .contentType(contentType) .fileSize(size) .build()); - log.debug("Wrote policy run {} output to {}", runId, target); + log.debug("Wrote policy run {} output to {}", delivery.runId(), target); } return results; } + /** + * Stream the output to its staging path. For a recorded delivery (stored policy) the content + * hash is digested in the same pass, so the ledger gets both version tiers without re-reading a + * possibly huge output; ad-hoc runs record nothing and skip the digest entirely. + */ + private static String stage(Resource resource, Path staged, boolean hashed) throws IOException { + if (!hashed) { + try (InputStream is = resource.getInputStream()) { + Files.copy(is, staged); + } + return null; + } + MessageDigest digest = ContentHasher.newSha256(); + try (InputStream is = resource.getInputStream(); + DigestOutputStream out = + new DigestOutputStream(Files.newOutputStream(staged), digest)) { + is.transferTo(out); + } + return ContentHasher.toHex(digest.digest()); + } + + /** + * The ledger row must exist before the file is visible at its final path, or a sweep could + * claim the producing policy's own output in the gap. Losing the chosen name to a concurrent + * writer forgets the just-recorded row - whatever file actually owns that name must stay + * claimable at any version - then re-picks. + */ + private Path moveIntoPlace( + OutputDelivery delivery, + Path dir, + String name, + Path staged, + String gate, + String contentHash) + throws IOException { + while (true) { + Path target = uniqueTarget(dir, name); + if (delivery.policyId() != null) { + processedLedger.recordOutput( + delivery.policyId(), target.toString(), gate, contentHash); + } + try { + Files.move(staged, target, StandardCopyOption.ATOMIC_MOVE); + return target; + } catch (FileAlreadyExistsException raced) { + if (delivery.policyId() != null) { + processedLedger.forgetOutput(delivery.policyId(), target.toString(), gate); + } + log.debug("Output name {} taken concurrently; re-picking", target); + } + } + } + + /** Best-effort removal of staging leftovers from crashed deliveries. */ + private static void sweepStaleTmp(Path tmpDir) { + Instant cutoff = Instant.now().minus(STALE_TMP_AGE); + try (Stream entries = Files.list(tmpDir)) { + entries.filter(Files::isRegularFile) + .filter( + entry -> { + try { + return Files.getLastModifiedTime(entry) + .toInstant() + .isBefore(cutoff); + } catch (IOException e) { + return false; + } + }) + .forEach( + entry -> { + try { + Files.deleteIfExists(entry); + } catch (IOException e) { + log.debug( + "Could not remove stale staging file {}: {}", + entry, + e.getMessage()); + } + }); + } catch (IOException e) { + log.debug("Could not sweep staging dir {}: {}", tmpDir, e.getMessage()); + } + } + private static Path directoryOf(OutputSpec spec) { Object directory = spec.options().get(DIRECTORY_OPTION); if (directory == null || directory.toString().isBlank()) { diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/InlineOutputSink.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/InlineOutputSink.java index 0fcd7323ff..904eb20523 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/InlineOutputSink.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/InlineOutputSink.java @@ -41,8 +41,8 @@ public class InlineOutputSink implements PolicyOutputSink { } @Override - public List deliver(String runId, List outputs, OutputSpec spec) - throws IOException { + public List deliver( + OutputDelivery delivery, List outputs, OutputSpec spec) throws IOException { List results = new ArrayList<>(); for (int i = 0; i < outputs.size(); i++) { Resource resource = outputs.get(i); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/OutputDelivery.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/OutputDelivery.java new file mode 100644 index 0000000000..ef3b59bebe --- /dev/null +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/OutputDelivery.java @@ -0,0 +1,8 @@ +package stirling.software.proprietary.policy.output; + +/** + * Context for one run's output delivery. {@code policyId} is null for ad-hoc pipelines; when + * present, sinks record outputs in the processed-file ledger so the producing policy does not + * re-ingest them. + */ +public record OutputDelivery(String runId, String policyId) {} diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputSink.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputSink.java index 6e6c80ba4d..7a9744f9e7 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputSink.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/output/PolicyOutputSink.java @@ -25,6 +25,6 @@ public interface PolicyOutputSink { default void validate(OutputSpec spec) {} /** Persist/deliver the output files and return their descriptors. */ - List deliver(String runId, List outputs, OutputSpec spec) + List deliver(OutputDelivery delivery, List outputs, OutputSpec spec) throws IOException; } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/FolderWatchTrigger.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/FolderWatchTrigger.java index 79ba3e52e1..dee78e8488 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/FolderWatchTrigger.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/policy/trigger/FolderWatchTrigger.java @@ -29,6 +29,7 @@ import lombok.extern.slf4j.Slf4j; import stirling.software.common.model.ApplicationProperties; import stirling.software.proprietary.policy.config.FolderAccessGuard; import stirling.software.proprietary.policy.engine.PolicyRunner; +import stirling.software.proprietary.policy.engine.SweepKind; import stirling.software.proprietary.policy.input.InputSource; import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.Policy; @@ -202,7 +203,8 @@ public class FolderWatchTrigger implements PolicyTrigger { } if (dirs.stream().anyMatch(changedDirs::contains)) { log.debug("Folder-watch policy {} ({}) saw activity", policy.id(), policy.name()); - policyRunner.run(policy); + // Light: the periodic reconcile does the full sweep. + policyRunner.run(policy, SweepKind.LIGHT); } } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java b/app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java index 74f904c455..2e9c4d3e5b 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java +++ b/app/proprietary/src/main/java/stirling/software/proprietary/security/configuration/DatabaseConfig.java @@ -34,6 +34,7 @@ import stirling.software.common.model.exception.UnsupportedProviderException; "stirling.software.proprietary.workflow.repository", "stirling.software.proprietary.policy.store", "stirling.software.proprietary.policy.source", + "stirling.software.proprietary.policy.ledger", "stirling.software.proprietary.accountlink", "stirling.software.proprietary.access.repository", "stirling.software.proprietary.integration.repository", @@ -46,6 +47,7 @@ import stirling.software.common.model.exception.UnsupportedProviderException; "stirling.software.proprietary.workflow.model", "stirling.software.proprietary.policy.store", "stirling.software.proprietary.policy.source", + "stirling.software.proprietary.policy.ledger", "stirling.software.proprietary.accountlink", "stirling.software.proprietary.access.model", "stirling.software.proprietary.integration.model", diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java index 8a9a085653..f38263ad3e 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/controller/PolicyControllerTest.java @@ -34,6 +34,7 @@ import stirling.software.proprietary.policy.engine.PolicyRunHandle; import stirling.software.proprietary.policy.engine.PolicyRunRegistry; import stirling.software.proprietary.policy.engine.PolicyRunner; import stirling.software.proprietary.policy.engine.PolicyValidator; +import stirling.software.proprietary.policy.ledger.ProcessedLedger; import stirling.software.proprietary.policy.model.PipelineDefinition; import stirling.software.proprietary.policy.model.PipelineStep; import stirling.software.proprietary.policy.model.Policy; @@ -62,6 +63,8 @@ class PolicyControllerTest { private stirling.software.proprietary.policy.overview.PolicyOverviewService policyOverviewService; + @Mock private ProcessedLedger processedLedger; + @Mock private TempFileManager tempFileManager; @Mock private JobOwnershipService jobOwnershipService; @@ -89,6 +92,7 @@ class PolicyControllerTest { policyManagementAuthority, policyTriggerManager, policyOverviewService, + processedLedger, policyTriggers, applicationProperties, tempFileManager, @@ -398,6 +402,7 @@ class PolicyControllerTest { ResponseEntity response = controller.deletePolicy("a"); assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NO_CONTENT); + verify(processedLedger).clearPolicy("a"); verify(policyTriggerManager).notifyPoliciesChanged(); } @@ -431,6 +436,53 @@ class PolicyControllerTest { } } + @Nested + @DisplayName("clearProcessedHistory") + class ClearProcessedHistory { + + @Test + @DisplayName("clears an accessible policy's history") + void clears() { + applicationProperties.getSecurity().setEnableLogin(false); + Policy p = policy("a", 1L); + when(policyStore.get("a")).thenReturn(Optional.of(p)); + when(policyAccessGuard.canAccess(p)).thenReturn(true); + + ResponseEntity response = controller.clearProcessedHistory("a"); + + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NO_CONTENT); + verify(processedLedger).clearPolicy("a"); + } + + @Test + @DisplayName("returns 404 when policy is not accessible") + void notAccessible() { + applicationProperties.getSecurity().setEnableLogin(false); + Policy p = policy("a", 1L); + when(policyStore.get("a")).thenReturn(Optional.of(p)); + when(policyAccessGuard.canAccess(p)).thenReturn(false); + + ResponseEntity response = controller.clearProcessedHistory("a"); + + assertThat(response.getStatusCode()).isEqualTo(HttpStatus.NOT_FOUND); + verify(processedLedger, never()).clearPolicy(any()); + } + + @Test + @DisplayName("forbidden when login enabled and caller cannot edit") + void forbidden() { + applicationProperties.getSecurity().setEnableLogin(true); + when(policyManagementAuthority.canEditPolicies()).thenReturn(false); + + assertThatThrownBy(() -> controller.clearProcessedHistory("a")) + .isInstanceOf(ResponseStatusException.class) + .satisfies( + e -> + assertThat(((ResponseStatusException) e).getStatusCode()) + .isEqualTo(HttpStatus.FORBIDDEN)); + } + } + @Nested @DisplayName("runStoredPolicy") class RunStoredPolicy { diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java index b1ab05f8ba..13189b2481 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java @@ -4,15 +4,19 @@ import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertSame; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; +import java.io.IOException; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.CompletableFuture; import java.util.concurrent.atomic.AtomicBoolean; @@ -24,7 +28,9 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import stirling.software.proprietary.policy.input.InputSource; +import stirling.software.proprietary.policy.input.ResolveContext; import stirling.software.proprietary.policy.input.ResolvedInput; +import stirling.software.proprietary.policy.ledger.ProcessedLedger; import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.OutputSpec; import stirling.software.proprietary.policy.model.PipelineStep; @@ -39,15 +45,15 @@ import stirling.software.proprietary.policy.source.Source; import stirling.software.proprietary.policy.source.SourceStore; /** - * Tests for {@link PolicyRunner}: the one place that turns a policy's sources into runs. Verifies - * it pulls every source, runs one job per unit of work, feeds each unit's completion hook the run - * outcome, and that a generator (no sources) still runs once. + * Tests for {@link PolicyRunner}: the one place that turns a policy's sources into runs, and the + * orchestrator of ledger hygiene (presence stamping + cleanup on complete FULL sweeps). */ @ExtendWith(MockitoExtension.class) class PolicyRunnerTest { @Mock private PolicyEngine policyEngine; @Mock private InputSource folderSource; + @Mock private ProcessedLedger processedLedger; private final SourceStore sourceStore = new InProcessSourceStore(); private PolicyRunner runner; @@ -59,7 +65,8 @@ class PolicyRunnerTest { policyEngine, List.of(folderSource), sourceStore, - new InProcessSourceDocCounter()); + new InProcessSourceDocCounter(), + processedLedger); } @Test @@ -73,6 +80,9 @@ class PolicyRunnerTest { ArgumentCaptor inputs = ArgumentCaptor.forClass(PolicyInputs.class); verify(policyEngine).runPolicy(eq(policy), inputs.capture(), 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. + verify(processedLedger).deleteUnseen(eq("p1"), anyLong()); } @Test @@ -80,7 +90,7 @@ class PolicyRunnerTest { InputSpec spec = InputSpec.folder("/in"); Policy policy = policy(List.of(spec)); when(folderSource.supports(spec)).thenReturn(true); - when(folderSource.resolve(spec)) + when(folderSource.resolve(eq(spec), any())) .thenReturn( List.of( ResolvedInput.of(PolicyInputs.of(List.of())), @@ -100,7 +110,7 @@ class PolicyRunnerTest { AtomicBoolean outcome = new AtomicBoolean(false); ResolvedInput unit = new ResolvedInput(PolicyInputs.of(List.of()), outcome::set); when(folderSource.supports(spec)).thenReturn(true); - when(folderSource.resolve(spec)).thenReturn(List.of(unit)); + when(folderSource.resolve(eq(spec), any())).thenReturn(List.of(unit)); CompletableFuture completion = new CompletableFuture<>(); when(policyEngine.runPolicy(any(), any(), any())) .thenReturn(new PolicyRunHandle("r", completion)); @@ -121,7 +131,7 @@ class PolicyRunnerTest { AtomicBoolean outcome = new AtomicBoolean(true); ResolvedInput unit = new ResolvedInput(PolicyInputs.of(List.of()), outcome::set); when(folderSource.supports(spec)).thenReturn(true); - when(folderSource.resolve(spec)).thenReturn(List.of(unit)); + when(folderSource.resolve(eq(spec), any())).thenReturn(List.of(unit)); CompletableFuture completion = new CompletableFuture<>(); when(policyEngine.runPolicy(any(), any(), any())) .thenReturn(new PolicyRunHandle("r", completion)); @@ -143,6 +153,98 @@ class PolicyRunnerTest { verifyNoInteractions(policyEngine); } + @Test + void aFullSweepStampsPresenceAndPrunesUnseenRows() throws Exception { + InputSpec spec = InputSpec.folder("/in"); + Policy policy = policy(List.of(spec)); + when(folderSource.supports(spec)).thenReturn(true); + when(folderSource.listsExhaustively()).thenReturn(true); + when(folderSource.resolve(eq(spec), any())) + .thenAnswer( + invocation -> { + ResolveContext ctx = invocation.getArgument(1); + ctx.reportPresent(List.of("/in/a.pdf", "/in/b.pdf")); + return List.of(); + }); + + runner.run(policy); + + // Presence reporting also bulk-prefetches claim state: one lookup for the whole listing. + verify(processedLedger).statesFor(eq("p1"), eq(List.of("/in/a.pdf", "/in/b.pdf"))); + verify(processedLedger).markSeen("p1", Set.of("/in/a.pdf", "/in/b.pdf")); + verify(processedLedger).deleteUnseen(eq("p1"), anyLong()); + } + + @Test + void aLightSweepClaimsButSkipsLedgerHygiene() throws Exception { + InputSpec spec = InputSpec.folder("/in"); + Policy policy = policy(List.of(spec)); + 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())) + .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>())); + + runner.run(policy, SweepKind.LIGHT); + + verify(policyEngine).runPolicy(eq(policy), any(), any()); + verify(processedLedger, never()).markSeen(any(), any()); + verify(processedLedger, never()).deleteUnseen(any(), anyLong()); + } + + @Test + void aSourceThatFailsToResolveVetoesCleanupButOthersStillRun() throws Exception { + InputSpec broken = InputSpec.folder("/broken"); + InputSpec healthy = InputSpec.folder("/healthy"); + Policy policy = policy(List.of(broken, healthy)); + when(folderSource.supports(any())).thenReturn(true); + when(folderSource.listsExhaustively()).thenReturn(true); + 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())) + .thenReturn(new PolicyRunHandle("r", new CompletableFuture<>())); + + runner.run(policy); + + verify(policyEngine).runPolicy(eq(policy), any(), any()); // healthy source still ran + verify(processedLedger, never()).deleteUnseen(any(), anyLong()); // history preserved + } + + @Test + void aDisabledSourceVetoesCleanup() { + InputSpec spec = InputSpec.folder("/in"); + String pausedId = sourceStore.save(disabledSourceFrom(spec)).id(); + Policy policy = policyReferencing(List.of(pausedId)); + + runner.run(policy); + + verify(processedLedger, never()).deleteUnseen(any(), anyLong()); + } + + @Test + void aNonExhaustiveSourceVetoesCleanup() throws Exception { + InputSpec spec = InputSpec.folder("/in"); + Policy policy = policy(List.of(spec)); + when(folderSource.supports(spec)).thenReturn(true); + when(folderSource.listsExhaustively()).thenReturn(false); + when(folderSource.resolve(eq(spec), any())).thenReturn(List.of()); + + runner.run(policy); + + verify(processedLedger, never()).deleteUnseen(any(), anyLong()); + } + + @Test + void aMissingSourceDoesNotVetoCleanup() { + // A deleted source's rows age out precisely because cleanup still runs. + Policy policy = policyReferencing(List.of("ghost-source-id")); + + runner.run(policy); + + verify(processedLedger).deleteUnseen(eq("p1"), anyLong()); + } + @Test void runWithSuppliedInputsBypassesSources() { Policy policy = policy(List.of(InputSpec.folder("/in"))); @@ -159,6 +261,10 @@ class PolicyRunnerTest { private Policy policy(List sources) { List sourceIds = sources.stream().map(spec -> sourceStore.save(sourceFrom(spec)).id()).toList(); + return policyReferencing(sourceIds); + } + + private static Policy policyReferencing(List sourceIds) { return new Policy( "p1", "p", @@ -173,4 +279,8 @@ class PolicyRunnerTest { private static Source sourceFrom(InputSpec spec) { return new Source(null, "src", spec.type(), spec.options(), true, "owner", null); } + + private static Source disabledSourceFrom(InputSpec spec) { + return new Source(null, "src", spec.type(), spec.options(), false, "owner", null); + } } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/FolderInputSourceTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/FolderInputSourceTest.java index 62bfa3aabb..64178ce9bb 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/FolderInputSourceTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/input/FolderInputSourceTest.java @@ -1,17 +1,23 @@ package stirling.software.proprietary.policy.input; import static org.junit.jupiter.api.Assertions.assertEquals; -import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.when; import java.io.IOException; import java.nio.file.Files; +import java.nio.file.NoSuchFileException; import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.Map; +import java.util.function.Supplier; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -24,18 +30,26 @@ import org.springframework.core.env.StandardEnvironment; import stirling.software.common.model.ApplicationProperties; import stirling.software.common.util.FileReadinessChecker; import stirling.software.proprietary.policy.config.FolderAccessGuard; +import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger; import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.source.InProcessSourceStore; -/** Tests for {@link FolderInputSource}: consume (claim + route) and snapshot (read-only) modes. */ +/** + * Tests for {@link FolderInputSource}: consume mode tracks files in place through the ledger, + * snapshot stays stateless, and discovery skips hidden entries and honours the recursive option. + */ @ExtendWith(MockitoExtension.class) class FolderInputSourceTest { + private static final String POLICY = "p1"; + @Mock private FileReadinessChecker readinessChecker; @TempDir Path tempDir; private FolderInputSource source; + private InProcessProcessedLedger ledger; + private RecordingContext ctx; @BeforeEach void setUp() { @@ -45,73 +59,279 @@ class FolderInputSourceTest { new FolderAccessGuard( properties, new StandardEnvironment(), new InProcessSourceStore()); source = new FolderInputSource(readinessChecker, guard); + ledger = new InProcessProcessedLedger(); + ctx = new RecordingContext(); // Lenient: the missing-dir / nonexistent-dir cases return before any readiness check. lenient().when(readinessChecker.isReady(any())).thenReturn(true); } @Test - void consumeClaimsFilesAndRoutesToDoneOnSuccess() throws IOException { + void consumeRemovesTheFileOnceProcessed() throws IOException { Path inputDir = Files.createDirectories(tempDir.resolve("in")); - Files.writeString(inputDir.resolve("doc.pdf"), "data"); + Path file = inputDir.resolve("doc.pdf"); + Files.writeString(file, "data"); - List work = source.resolve(InputSpec.folder(inputDir.toString())); + List work = source.resolve(InputSpec.folder(inputDir.toString()), ctx); assertEquals(1, work.size()); assertEquals(1, work.get(0).inputs().primary().size()); - // Claimed out of the input dir. - assertFalse(Files.exists(inputDir.resolve("doc.pdf"))); - assertTrue( - Files.exists( - inputDir.resolve(".stirling").resolve("processing").resolve("doc.pdf"))); + // In flight: still on disk, but a second sweep does not pick it up again. + assertTrue(Files.exists(file)); + assertTrue(Files.notExists(inputDir.resolve(".stirling"))); + assertTrue(source.resolve(InputSpec.folder(inputDir.toString()), ctx).isEmpty()); work.get(0).onComplete().accept(true); - assertTrue(Files.exists(inputDir.resolve(".stirling").resolve("done").resolve("doc.pdf"))); - assertFalse( - Files.exists( - inputDir.resolve(".stirling").resolve("processing").resolve("doc.pdf"))); + assertTrue(Files.notExists(file)); + assertTrue(source.resolve(InputSpec.folder(inputDir.toString()), ctx).isEmpty()); } @Test - void consumeRoutesToErrorOnFailure() throws IOException { + void aFileReplacedMidRunSurvivesTheDeleteAndRunsAgain() throws IOException { Path inputDir = Files.createDirectories(tempDir.resolve("in")); - Files.writeString(inputDir.resolve("doc.pdf"), "data"); + Path file = inputDir.resolve("doc.pdf"); + Files.writeString(file, "data"); - List work = source.resolve(InputSpec.folder(inputDir.toString())); - work.get(0).onComplete().accept(false); + List work = source.resolve(InputSpec.folder(inputDir.toString()), ctx); + // The user saves a new version while the run is executing. + Files.writeString(file, "new data, different size"); + work.get(0).onComplete().accept(true); - assertTrue(Files.exists(inputDir.resolve(".stirling").resolve("error").resolve("doc.pdf"))); + // The delete is version-guarded: the replacement is not the file that ran, so it stays + // and is claimed as fresh work instead of being marked processed. + assertTrue(Files.exists(file)); + assertEquals(1, source.resolve(InputSpec.folder(inputDir.toString()), ctx).size()); } @Test - void snapshotReadsWithoutClaiming() throws IOException { + void aSharedFileIsRemovedOnlyOnceEveryPolicyHasProcessedIt() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Path file = inputDir.resolve("doc.pdf"); + Files.writeString(file, "data"); + InputSpec spec = InputSpec.folder(inputDir.toString()); + RecordingContext other = new RecordingContext("p2"); + + List mine = source.resolve(spec, ctx); + List theirs = source.resolve(spec, other); + assertEquals(1, mine.size()); + assertEquals(1, theirs.size()); + + mine.get(0).onComplete().accept(true); + // The other policy's claim is still in flight, so the first finisher must not delete. + assertTrue(Files.exists(file)); + + theirs.get(0).onComplete().accept(true); + assertTrue(Files.notExists(file)); + } + + @Test + void aSharedFileStaysParkedWhenAnyPolicyFailsOnIt() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Path file = inputDir.resolve("doc.pdf"); + Files.writeString(file, "data"); + InputSpec spec = InputSpec.folder(inputDir.toString()); + RecordingContext other = new RecordingContext("p2"); + + List mine = source.resolve(spec, ctx); + List theirs = source.resolve(spec, other); + + theirs.get(0).onComplete().accept(false); + mine.get(0).onComplete().accept(true); + + // The failure parks the file for everyone (retried when it changes), regardless of + // which policy settled last. + assertTrue(Files.exists(file)); + assertTrue(source.resolve(spec, ctx).isEmpty()); + assertTrue(source.resolve(spec, other).isEmpty()); + } + + @Test + void aReDroppedFileIsProcessedAgain() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Path file = inputDir.resolve("doc.pdf"); + Files.writeString(file, "data"); + + source.resolve(InputSpec.folder(inputDir.toString()), ctx).get(0).onComplete().accept(true); + Files.writeString(file, "data again"); + + assertEquals(1, source.resolve(InputSpec.folder(inputDir.toString()), ctx).size()); + } + + @Test + void aFailedFileStaysInPlaceAndIsNotRetriedUntilItChanges() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Path file = inputDir.resolve("doc.pdf"); + Files.writeString(file, "data"); + + source.resolve(InputSpec.folder(inputDir.toString()), ctx) + .get(0) + .onComplete() + .accept(false); + + assertTrue(Files.exists(file)); + assertTrue(source.resolve(InputSpec.folder(inputDir.toString()), ctx).isEmpty()); + + Files.setLastModifiedTime(file, FileTime.from(Instant.now().plusSeconds(60))); + assertEquals(1, source.resolve(InputSpec.folder(inputDir.toString()), ctx).size()); + } + + @Test + void statModeRetriesAFailureOnATouchButHashModeDoesNot() throws IOException { + Path statDir = Files.createDirectories(tempDir.resolve("stat")); + Path hashDir = Files.createDirectories(tempDir.resolve("hash")); + Path statFile = statDir.resolve("doc.pdf"); + Path hashFile = hashDir.resolve("doc.pdf"); + Files.writeString(statFile, "data"); + Files.writeString(hashFile, "data"); + InputSpec statSpec = InputSpec.folder(statDir.toString()); + InputSpec hashSpec = + new InputSpec( + "folder", Map.of("directory", hashDir.toString(), "identity", "hash")); + + source.resolve(statSpec, ctx).get(0).onComplete().accept(false); + source.resolve(hashSpec, ctx).get(0).onComplete().accept(false); + + FileTime touched = FileTime.from(Instant.now().plusSeconds(60)); + Files.setLastModifiedTime(statFile, touched); + Files.setLastModifiedTime(hashFile, touched); + + // Same content, new mtime: stat mode calls that a new version and retries; hash mode + // verifies the content is unchanged and keeps the failure parked. + assertEquals(1, source.resolve(statSpec, ctx).size()); + assertTrue(source.resolve(hashSpec, ctx).isEmpty()); + } + + @Test + void hashModeRetriesAFailureOnARealContentChange() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Path file = inputDir.resolve("doc.pdf"); + Files.writeString(file, "data"); + InputSpec spec = + new InputSpec( + "folder", Map.of("directory", inputDir.toString(), "identity", "hash")); + + source.resolve(spec, ctx).get(0).onComplete().accept(false); + Files.writeString(file, "data v2 - longer"); + + assertEquals(1, source.resolve(spec, ctx).size()); + } + + @Test + void snapshotReadsStatelesslyEverySweep() throws IOException { Path inputDir = Files.createDirectories(tempDir.resolve("in")); Files.writeString(inputDir.resolve("doc.pdf"), "data"); + InputSpec spec = + new InputSpec( + "folder", Map.of("directory", inputDir.toString(), "mode", "snapshot")); - List work = - source.resolve( - new InputSpec( - "folder", - Map.of("directory", inputDir.toString(), "mode", "snapshot"))); + List first = source.resolve(spec, ctx); + first.get(0).onComplete().accept(true); + List second = source.resolve(spec, ctx); + + assertEquals(1, first.size()); + assertEquals(1, second.size()); // no ledger involvement: every run sees the full set + assertTrue(ctx.present.isEmpty()); + assertTrue(Files.exists(inputDir.resolve("doc.pdf"))); + } + + @Test + void hiddenFilesAndTheLegacyWorkDirAreIgnored() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Files.writeString(inputDir.resolve("doc.pdf"), "data"); + Files.writeString(inputDir.resolve(".hidden.pdf"), "secret"); + Path legacy = Files.createDirectories(inputDir.resolve(".stirling").resolve("done")); + Files.writeString(legacy.resolve("old.pdf"), "processed long ago"); + + List work = source.resolve(InputSpec.folder(inputDir.toString()), ctx); assertEquals(1, work.size()); - // Not moved, and completing the run is a no-op. - assertTrue(Files.exists(inputDir.resolve("doc.pdf"))); - work.get(0).onComplete().accept(true); - assertTrue(Files.exists(inputDir.resolve("doc.pdf"))); + assertEquals(1, ctx.present.size()); + assertTrue(ctx.present.get(0).endsWith("doc.pdf")); + } + + @Test + void recursiveDiscoversSubdirectoriesButNotHiddenOnes() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Files.writeString(inputDir.resolve("top.pdf"), "a"); + Path sub = Files.createDirectories(inputDir.resolve("sub")); + Files.writeString(sub.resolve("nested.pdf"), "b"); + Path hiddenDir = Files.createDirectories(inputDir.resolve(".stirling")); + Files.writeString(hiddenDir.resolve("skipped.pdf"), "c"); + // Sink staging inside a watched subdirectory is pruned at any depth. + Path nestedStaging = Files.createDirectories(sub.resolve(".stirling").resolve("tmp")); + Files.writeString(nestedStaging.resolve("half-delivered"), "d"); + + InputSpec flat = InputSpec.folder(inputDir.toString()); + InputSpec recursive = + new InputSpec( + "folder", Map.of("directory", inputDir.toString(), "recursive", "true")); + + assertEquals(1, source.resolve(flat, ctx).size()); + assertEquals(1, source.resolve(recursive, ctx).size()); // top.pdf already claimed above + assertTrue(ctx.present.stream().anyMatch(identity -> identity.endsWith("nested.pdf"))); + assertTrue(ctx.present.stream().noneMatch(identity -> identity.endsWith("skipped.pdf"))); + assertTrue(ctx.present.stream().noneMatch(identity -> identity.endsWith("half-delivered"))); + } + + @Test + void unreadyFilesAreReportedPresentButNotClaimed() throws IOException { + Path inputDir = Files.createDirectories(tempDir.resolve("in")); + Path file = inputDir.resolve("mid-write.pdf"); + Files.writeString(file, "partial"); + when(readinessChecker.isReady(file)).thenReturn(false); + + List work = source.resolve(InputSpec.folder(inputDir.toString()), ctx); + + assertTrue(work.isEmpty()); + // Reported present so a full sweep does not prune its row while it settles on disk. + assertEquals(1, ctx.present.size()); + assertTrue(ctx.present.get(0).endsWith("mid-write.pdf")); + } + + @Test + void nestedSourcesShareThePolicysLedgerAndDoNotDoubleClaim() throws IOException { + Path parent = Files.createDirectories(tempDir.resolve("in")); + Path child = Files.createDirectories(parent.resolve("sub")); + Files.writeString(child.resolve("doc.pdf"), "data"); + InputSpec parentRecursive = + new InputSpec( + "folder", Map.of("directory", parent.toString(), "recursive", "true")); + InputSpec childFlat = InputSpec.folder(child.toString()); + + // Same sweep, same policy context: whichever source resolves first wins the file. + assertEquals(1, source.resolve(parentRecursive, ctx).size()); + assertTrue(source.resolve(childFlat, ctx).isEmpty()); } @Test void missingDirectoryOptionFails() { assertThrows( IllegalArgumentException.class, - () -> source.resolve(new InputSpec("folder", Map.of()))); + () -> source.resolve(new InputSpec("folder", Map.of()), ctx)); } @Test - void nonexistentDirectoryYieldsNoWork() throws IOException { - List work = - source.resolve(InputSpec.folder(tempDir.resolve("nope").toString())); - assertTrue(work.isEmpty()); + void anUnknownIdentityModeIsRejected() { + assertThrows( + IllegalArgumentException.class, + () -> + source.validate( + new InputSpec( + "folder", + Map.of( + "directory", + tempDir.toString(), + "identity", + "guesswork")))); + } + + @Test + void nonexistentDirectoryFailsResolveSoTheSweepVetoesCleanup() { + // An unreachable directory (e.g. unmounted drive) must surface as a failed listing, not + // an empty one: the runner vetoes presence cleanup on failure, keeping the history that + // an empty listing would wipe. + assertThrows( + NoSuchFileException.class, + () -> source.resolve(InputSpec.folder(tempDir.resolve("nope").toString()), ctx)); } @Test @@ -126,7 +346,7 @@ class FolderInputSourceTest { Path outside = tempDir.resolveSibling("not-allowed"); assertThrows( IllegalArgumentException.class, - () -> source.resolve(InputSpec.folder(outside.toString()))); + () -> source.resolve(InputSpec.folder(outside.toString()), ctx)); assertThrows( IllegalArgumentException.class, () -> source.validate(InputSpec.folder(outside.toString()))); @@ -137,4 +357,40 @@ class FolderInputSourceTest { Path inputDir = tempDir.resolve("in"); assertEquals(List.of(inputDir), source.watchTargets(InputSpec.folder(inputDir.toString()))); } + + /** Policy-scoped context backed by the in-process ledger, recording presence reports. */ + private class RecordingContext implements ResolveContext { + + private final String policyId; + private final List present = new ArrayList<>(); + + private RecordingContext() { + this(POLICY); + } + + private RecordingContext(String policyId) { + this.policyId = policyId; + } + + @Override + public boolean claim(String identity, String gate, Supplier contentHash) { + return ledger.claim(policyId, identity, gate, contentHash); + } + + @Override + public void settle( + String identity, String finalGate, String finalContentHash, boolean success) { + ledger.settle(policyId, identity, finalGate, finalContentHash, success); + } + + @Override + public boolean allSettledDone(String identity) { + return ledger.allSettledDone(identity); + } + + @Override + public void reportPresent(Collection identities) { + present.addAll(identities); + } + } } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/FolderIdentitiesTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/FolderIdentitiesTest.java new file mode 100644 index 0000000000..9f56411d9b --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/FolderIdentitiesTest.java @@ -0,0 +1,84 @@ +package stirling.software.proprietary.policy.ledger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.io.IOException; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.attribute.FileTime; +import java.time.Instant; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * Tests for {@link FolderIdentities}: identity derivation must agree for the same file, even + * through a symlinked alias of the directory. + */ +class FolderIdentitiesTest { + + @TempDir Path tempDir; + + @Test + void identityAgreesAcrossASymlinkedAliasOfTheDirectory() throws IOException { + Path real = Files.createDirectories(tempDir.resolve("real")); + Path alias = Files.createSymbolicLink(tempDir.resolve("alias"), real); + Files.writeString(real.resolve("doc.pdf"), "data"); + + String viaReal = + FolderIdentities.identity( + FolderIdentities.canonicalDir(real), real, real.resolve("doc.pdf")); + String viaAlias = + FolderIdentities.identity( + FolderIdentities.canonicalDir(alias), alias, alias.resolve("doc.pdf")); + + assertEquals(viaReal, viaAlias); + } + + @Test + void identityOfANestedFileKeepsItsRelativePath() throws IOException { + Path dir = Files.createDirectories(tempDir.resolve("in")); + Path nested = Files.createDirectories(dir.resolve("sub")).resolve("doc.pdf"); + Files.writeString(nested, "data"); + + String identity = + FolderIdentities.identity(FolderIdentities.canonicalDir(dir), dir, nested); + + assertTrue(identity.endsWith("sub" + java.io.File.separator + "doc.pdf")); + } + + @Test + void theGateTracksSizeAndMtime() throws IOException { + Path file = tempDir.resolve("doc.pdf"); + Files.writeString(file, "data"); + String before = FolderIdentities.statGate(file); + + Files.setLastModifiedTime(file, FileTime.from(Instant.now().plusSeconds(60))); + + assertNotEquals(before, FolderIdentities.statGate(file)); + } + + @Test + void theContentHashIgnoresMtimeButTracksContent() throws IOException { + Path file = tempDir.resolve("doc.pdf"); + Files.writeString(file, "data"); + String before = FolderIdentities.contentHash(file); + + Files.setLastModifiedTime(file, FileTime.from(Instant.now().plusSeconds(60))); + assertEquals(before, FolderIdentities.contentHash(file)); + + Files.writeString(file, "different"); + assertNotEquals(before, FolderIdentities.contentHash(file)); + } + + @Test + void identityHashIsAStableFixedWidthKey() { + String hash = IdentityHasher.identityHash("/in/doc.pdf"); + + assertEquals(64, hash.length()); + assertEquals(hash, IdentityHasher.identityHash("/in/doc.pdf")); + assertNotEquals(hash, IdentityHasher.identityHash("/in/other.pdf")); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/InProcessProcessedLedgerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/InProcessProcessedLedgerTest.java new file mode 100644 index 0000000000..08ac10061d --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/InProcessProcessedLedgerTest.java @@ -0,0 +1,12 @@ +package stirling.software.proprietary.policy.ledger; + +import java.util.function.Supplier; + +/** {@link InProcessProcessedLedger} against the shared {@link ProcessedLedger} contract. */ +class InProcessProcessedLedgerTest extends ProcessedLedgerContractTest { + + @Override + ProcessedLedger newLedger(Supplier nowMillis) { + return new InProcessProcessedLedger(nowMillis); + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/JpaProcessedLedgerDbTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/JpaProcessedLedgerDbTest.java new file mode 100644 index 0000000000..8c1bd85c94 --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/JpaProcessedLedgerDbTest.java @@ -0,0 +1,36 @@ +package stirling.software.proprietary.policy.ledger; + +import java.util.function.Supplier; + +import org.junit.jupiter.api.AfterEach; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringBootConfiguration; +import org.springframework.boot.autoconfigure.AutoConfigurationPackage; +import org.springframework.boot.data.jpa.test.autoconfigure.DataJpaTest; + +/** + * {@link JpaProcessedLedger} against the shared contract on a real (H2) database. The inherited + * tests run outside {@code @DataJpaTest}'s per-test transaction (the transaction attribute resolves + * against the declaring class, the plain contract base), so every ledger call commits in its own + * transaction as at runtime; state is wiped explicitly instead of relying on rollback. + */ +@DataJpaTest +class JpaProcessedLedgerDbTest extends ProcessedLedgerContractTest { + + @Autowired private ProcessedFileRepository repository; + + @AfterEach + void wipeLedger() { + // deleteAll() skips entities whose isNew() is hardcoded true, so use the bulk form. + repository.deleteAllInBatch(); + } + + @Override + ProcessedLedger newLedger(Supplier nowMillis) { + return new JpaProcessedLedger(repository, nowMillis); + } + + @SpringBootConfiguration + @AutoConfigurationPackage + static class TestApp {} +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/ProcessedLedgerContractTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/ProcessedLedgerContractTest.java new file mode 100644 index 0000000000..800e8a6bde --- /dev/null +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/ledger/ProcessedLedgerContractTest.java @@ -0,0 +1,360 @@ +package stirling.software.proprietary.policy.ledger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Supplier; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +/** The {@link ProcessedLedger} contract, run against every implementation so they cannot drift. */ +abstract class ProcessedLedgerContractTest { + + static final String POLICY = "p1"; + static final String OTHER_POLICY = "p2"; + static final String FILE = "/in/doc.pdf"; + static final String GATE = "100:1111"; + static final String NEW_GATE = "100:2222"; + static final String HASH = "hash-aaa"; + static final String NEW_HASH = "hash-bbb"; + + final AtomicLong clock = new AtomicLong(1_000_000L); + + ProcessedLedger ledger; + + abstract ProcessedLedger newLedger(Supplier nowMillis); + + @BeforeEach + void createLedger() { + ledger = newLedger(clock::get); + } + + @Test + void aFileIsClaimedOnceAndSkippedWhileInFlight() { + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); + assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, null)); // new version waits for settle + } + + @Test + void aSettledFileIsSkippedAtTheSameGate() { + ledger.claim(POLICY, FILE, GATE, null); + ledger.settle(POLICY, FILE, GATE, null, true); + + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); + } + + @Test + void aMovedGateIsReclaimedInGateOnlyMode() { + ledger.claim(POLICY, FILE, GATE, null); + ledger.settle(POLICY, FILE, GATE, null, true); + + assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null)); + } + + @Test + void settlingAtTheOutputsVersionStopsAnInPlaceOverwriteLooping() { + ledger.claim(POLICY, FILE, GATE, null); + // The run overwrote the input; settle re-reads and lands on the produced version. + ledger.settle(POLICY, FILE, NEW_GATE, null, true); + + assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, null)); // own output: skip + assertTrue(ledger.claim(POLICY, FILE, "100:3333", null)); // later user edit: reprocess + } + + @Test + void aFailedFileIsNotRetriedUntilItChanges() { + ledger.claim(POLICY, FILE, GATE, null); + ledger.settle(POLICY, FILE, GATE, null, false); + + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); + assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null)); + } + + @Test + void aTouchedButUnchangedFileRefreshesTheGateInsteadOfReprocessing() { + ledger.claim(POLICY, FILE, GATE, hash(HASH)); + ledger.settle(POLICY, FILE, GATE, HASH, true); + + // Same content under a new gate (touch / identical re-copy): verified, not reprocessed. + assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH))); + + // The gate was refreshed, so the next sweep takes the cheap path: no content read at all. + CountingSupplier counting = new CountingSupplier(HASH); + assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, counting)); + assertEquals(0, counting.invocations.get()); + } + + @Test + void theVerificationTierIsNotConsultedWhileTheGateMatches() { + ledger.claim(POLICY, FILE, GATE, hash(HASH)); + ledger.settle(POLICY, FILE, GATE, HASH, true); + + CountingSupplier counting = new CountingSupplier(HASH); + assertFalse(ledger.claim(POLICY, FILE, GATE, counting)); + assertEquals(0, counting.invocations.get()); + } + + @Test + void aRealContentChangeUnderANewGateIsReprocessed() { + ledger.claim(POLICY, FILE, GATE, hash(HASH)); + ledger.settle(POLICY, FILE, GATE, HASH, true); + + assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, hash(NEW_HASH))); + } + + @Test + void aGateOnlySettledRowCannotBeContentVerifiedSoItReprocesses() { + ledger.claim(POLICY, FILE, GATE, null); + ledger.settle(POLICY, FILE, GATE, null, true); + + // The row stored no hash, so "same content" is unprovable: reprocess on gate change. + assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH))); + } + + @Test + void aFailedFileStaysParkedThroughATouch() { + ledger.claim(POLICY, FILE, GATE, hash(HASH)); + ledger.settle(POLICY, FILE, GATE, HASH, false); + + // A touch must not resurrect an ERROR row; only a real content change does. + assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH))); + assertTrue(ledger.claim(POLICY, FILE, "100:3333", hash(NEW_HASH))); + } + + @Test + void interruptedRunsAreRetriedABoundedNumberOfTimes() { + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); // attempt 1 dies with the JVM + ledger.recoverInterrupted(); + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); // attempt 2 + ledger.recoverInterrupted(); + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); // attempt 3, the last + ledger.recoverInterrupted(); + + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); // parked: no crash-loop + } + + @Test + void aNewGateResetsTheInterruptRetryBudgetInGateOnlyMode() { + for (int attempt = 0; attempt < ProcessedLedger.MAX_ATTEMPTS; attempt++) { + ledger.claim(POLICY, FILE, GATE, null); + ledger.recoverInterrupted(); + } + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); + + assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null)); + ledger.recoverInterrupted(); + assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null)); // fresh budget at the new version + } + + @Test + void aTouchDoesNotResetTheInterruptRetryBudgetWhenContentIsVerified() { + assertTrue(ledger.claim(POLICY, FILE, GATE, hash(HASH))); // attempt 1 + ledger.recoverInterrupted(); + // Same content, moved gate: still the interrupted work, still bounded. + assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH))); // attempt 2 + ledger.recoverInterrupted(); + assertTrue(ledger.claim(POLICY, FILE, "100:3333", hash(HASH))); // attempt 3 + ledger.recoverInterrupted(); + + assertFalse(ledger.claim(POLICY, FILE, "100:4444", hash(HASH))); // parked + assertTrue(ledger.claim(POLICY, FILE, "100:5555", hash(NEW_HASH))); // real change: fresh + } + + @Test + void recoveryOnlyTouchesInFlightRows() { + ledger.claim(POLICY, FILE, GATE, null); + ledger.settle(POLICY, FILE, GATE, null, true); + ledger.recoverInterrupted(); + + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); // still DONE, not retried + } + + @Test + void anOutputIsSkippedByItsProducerButSeenByOtherPolicies() { + ledger.recordOutput(POLICY, FILE, GATE, HASH); + + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); // producer skips its own output + assertTrue(ledger.claim(OTHER_POLICY, FILE, GATE, null)); // chaining still works + } + + @Test + void anOutputIsSkippedByAHashVerifyingProducerEvenIfTheGateMoved() { + ledger.recordOutput(POLICY, FILE, GATE, HASH); + + assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, hash(HASH))); + } + + @Test + void policiesTrackTheSameFileIndependently() { + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); + assertTrue(ledger.claim(OTHER_POLICY, FILE, GATE, null)); + } + + @Test + void statesForSnapshotsOnlyExistingRows() { + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); + + Map states = ledger.statesFor(POLICY, List.of(FILE, "/in/other.pdf")); + + assertEquals(1, states.size()); + assertEquals(ProcessedFileStatus.PROCESSING, states.get(FILE).status()); + assertEquals(GATE, states.get(FILE).gate()); + } + + @Test + void aStaleAbsentSnapshotLosesTheClaimRaceInsteadOfDoubleClaiming() { + ClaimState absent = ledger.statesFor(POLICY, List.of(FILE)).get(FILE); // no row yet + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); // another sweep wins meanwhile + + assertFalse(ledger.claim(POLICY, FILE, GATE, null, absent)); + } + + @Test + void aStaleSettledSnapshotCannotReclaimAnInFlightRow() { + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); + ledger.settle(POLICY, FILE, GATE, null, true); + ClaimState settled = ledger.statesFor(POLICY, List.of(FILE)).get(FILE); + assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null)); // a fresh sweep reclaims first + + assertFalse(ledger.claim(POLICY, FILE, "100:3333", null, settled)); + } + + @Test + void aForgottenOutputIsClaimableAtAnyVersion() { + ledger.recordOutput(POLICY, FILE, GATE, HASH); + ledger.forgetOutput(POLICY, FILE, GATE); + + // Even a byte-identical file at that identity is fresh work: the record is gone. + assertTrue(ledger.claim(POLICY, FILE, GATE, hash(HASH))); + } + + @Test + void forgetOutputLeavesARowReclaimedInTheMeantime() { + ledger.recordOutput(POLICY, FILE, GATE, HASH); + assertTrue(ledger.claim(POLICY, FILE, NEW_GATE, null)); // a real claim took the row over + + ledger.forgetOutput(POLICY, FILE, GATE); + + assertFalse(ledger.claim(POLICY, FILE, NEW_GATE, null)); // still in flight, not deleted + } + + @Test + void deletionConsensusNeedsEveryClaimantSettledDone() { + assertTrue(ledger.allSettledDone(FILE)); // vacuous: no rows yet + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); + assertTrue(ledger.claim(OTHER_POLICY, FILE, GATE, null)); + ledger.settle(POLICY, FILE, GATE, null, true); + assertFalse(ledger.allSettledDone(FILE)); // the other claim is still in flight + ledger.settle(OTHER_POLICY, FILE, GATE, null, true); + assertTrue(ledger.allSettledDone(FILE)); + } + + @Test + void aFailedClaimVetoesDeletionConsensus() { + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); + assertTrue(ledger.claim(OTHER_POLICY, FILE, GATE, null)); + ledger.settle(POLICY, FILE, GATE, null, true); + ledger.settle(OTHER_POLICY, FILE, GATE, null, false); + assertFalse(ledger.allSettledDone(FILE)); + } + + @Test + void anInterruptedClaimVetoesDeletionConsensus() { + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); + ledger.recoverInterrupted(); + assertFalse(ledger.allSettledDone(FILE)); + } + + @Test + void settleRecreatesARowRemovedMidRun() { + ledger.claim(POLICY, FILE, GATE, null); + ledger.clearPolicy(POLICY); // e.g. a clear-history while the run is in flight + ledger.settle(POLICY, FILE, GATE, null, true); + + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); + } + + @Test + void presenceCleanupRemovesOnlyUnseenSettledRows() { + String inFlight = "/in/in-flight.pdf"; + String stillPresent = "/in/still-present.pdf"; + String deleted = "/in/deleted.pdf"; + ledger.claim(POLICY, inFlight, GATE, null); + ledger.recordOutput(POLICY, stillPresent, GATE, HASH); + ledger.recordOutput(POLICY, deleted, GATE, HASH); + + clock.addAndGet(10_000); + long sweepStart = clock.get(); + ledger.markSeen(POLICY, List.of(inFlight, stillPresent)); // deleted.pdf is gone from disk + assertEquals(1, ledger.deleteUnseen(POLICY, sweepStart)); + + assertFalse(ledger.claim(POLICY, inFlight, GATE, null)); // in-flight row survived + assertFalse(ledger.claim(POLICY, stillPresent, GATE, null)); // stamped row survived + assertTrue(ledger.claim(POLICY, deleted, GATE, null)); // forgotten: a re-drop reprocesses + } + + @Test + void presenceCleanupNeverRemovesInFlightRowsEvenUnstamped() { + ledger.claim(POLICY, FILE, GATE, null); + clock.addAndGet(10_000); + + assertEquals(0, ledger.deleteUnseen(POLICY, clock.get())); + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); + } + + @Test + void rowsWrittenDuringTheSweepSurviveItsCleanup() { + long sweepStart = clock.get(); + // Recorded after the sweep's cutoff: unseen by it, but newer, so it must survive. + clock.addAndGet(5); + ledger.recordOutput(POLICY, FILE, GATE, HASH); + + assertEquals(0, ledger.deleteUnseen(POLICY, sweepStart)); + assertFalse(ledger.claim(POLICY, FILE, GATE, null)); + } + + @Test + void markSeenOnUnknownIdentitiesIsANoOp() { + ledger.markSeen(POLICY, List.of("/never/claimed.pdf")); + assertEquals(0, ledger.deleteUnseen(POLICY, clock.get())); + } + + @Test + void clearPolicyForgetsOnlyThatPolicy() { + ledger.claim(POLICY, FILE, GATE, null); + ledger.settle(POLICY, FILE, GATE, null, true); + ledger.claim(OTHER_POLICY, FILE, GATE, null); + ledger.settle(OTHER_POLICY, FILE, GATE, null, true); + + ledger.clearPolicy(POLICY); + + assertTrue(ledger.claim(POLICY, FILE, GATE, null)); + assertFalse(ledger.claim(OTHER_POLICY, FILE, GATE, null)); + } + + static Supplier hash(String value) { + return () -> value; + } + + static final class CountingSupplier implements Supplier { + final AtomicInteger invocations = new AtomicInteger(); + private final String value; + + CountingSupplier(String value) { + this.value = value; + } + + @Override + public String get() { + invocations.incrementAndGet(); + return value; + } + } +} diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/FolderOutputSinkTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/FolderOutputSinkTest.java index f1b8b599ac..f9e1c92403 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/FolderOutputSinkTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/output/FolderOutputSinkTest.java @@ -10,6 +10,7 @@ import java.nio.file.Files; import java.nio.file.Path; import java.util.List; import java.util.Map; +import java.util.stream.Stream; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -21,24 +22,35 @@ import org.springframework.core.io.Resource; import stirling.software.common.model.ApplicationProperties; import stirling.software.common.model.job.ResultFile; import stirling.software.proprietary.policy.config.FolderAccessGuard; +import stirling.software.proprietary.policy.ledger.FolderIdentities; +import stirling.software.proprietary.policy.ledger.InProcessProcessedLedger; import stirling.software.proprietary.policy.model.OutputSpec; import stirling.software.proprietary.policy.source.InProcessSourceStore; -/** Tests for {@link FolderOutputSink}: outputs are written to the configured directory on disk. */ +/** + * Tests for {@link FolderOutputSink}: outputs are staged hidden, recorded in the ledger, then + * atomically renamed into the configured directory. + */ class FolderOutputSinkTest { + private static final OutputDelivery AD_HOC = new OutputDelivery("run-1", null); + private static final OutputDelivery POLICY_RUN = new OutputDelivery("run-1", "p1"); + @TempDir Path tempDir; private FolderOutputSink sink; + private InProcessProcessedLedger ledger; @BeforeEach void setUp() { ApplicationProperties properties = new ApplicationProperties(); properties.getPolicies().setAllowedFolderRoots(List.of(tempDir.toString())); + ledger = new InProcessProcessedLedger(); sink = new FolderOutputSink( new FolderAccessGuard( - properties, new StandardEnvironment(), new InProcessSourceStore())); + properties, new StandardEnvironment(), new InProcessSourceStore()), + ledger); } @Test @@ -46,13 +58,80 @@ class FolderOutputSinkTest { Path out = tempDir.resolve("out"); List outputs = List.of(named("a.pdf", "aaa"), named("b.pdf", "bb")); - List results = - sink.deliver("run-1", outputs, OutputSpec.folder(out.toString())); + List results = sink.deliver(AD_HOC, outputs, OutputSpec.folder(out.toString())); assertEquals(2, results.size()); assertTrue(Files.exists(out.resolve("a.pdf"))); assertEquals("aaa", Files.readString(out.resolve("a.pdf"))); assertEquals("bb", Files.readString(out.resolve("b.pdf"))); + // Nothing left behind in the staging dir. + try (Stream staged = Files.list(out.resolve(".stirling").resolve("tmp"))) { + assertEquals(0, staged.count()); + } + } + + @Test + void recordsThePolicysOutputsSoOnlyOtherPoliciesReprocessThem() throws IOException { + Path out = tempDir.resolve("out"); + + sink.deliver(POLICY_RUN, List.of(named("a.pdf", "aaa")), OutputSpec.folder(out.toString())); + + Path delivered = FolderIdentities.canonicalDir(out).resolve("a.pdf"); + String gate = FolderIdentities.statGate(delivered); + assertFalse(ledger.claim("p1", delivered.toString(), gate, null)); // producer skips it + assertTrue(ledger.claim("p2", delivered.toString(), gate, null)); // chaining still works + } + + @Test + void aHashVerifyingProducerSkipsItsOwnOutputEvenIfTheGateMoved() throws IOException { + Path out = tempDir.resolve("out"); + + sink.deliver(POLICY_RUN, List.of(named("a.pdf", "aaa")), OutputSpec.folder(out.toString())); + + // A hash-verifying reader matches on content even when the stat moved. + Path delivered = FolderIdentities.canonicalDir(out).resolve("a.pdf"); + assertFalse( + ledger.claim( + "p1", + delivered.toString(), + "999:12345", + () -> { + try { + return FolderIdentities.contentHash(delivered); + } catch (IOException e) { + throw new java.io.UncheckedIOException(e); + } + })); + } + + @Test + void recordsAnOutputBeforeItBecomesVisible() throws IOException { + Path out = tempDir.resolve("out"); + VisibilityAssertingLedger orderedLedger = new VisibilityAssertingLedger(); + ApplicationProperties properties = new ApplicationProperties(); + properties.getPolicies().setAllowedFolderRoots(List.of(tempDir.toString())); + FolderOutputSink orderedSink = + new FolderOutputSink( + new FolderAccessGuard( + properties, new StandardEnvironment(), new InProcessSourceStore()), + orderedLedger); + + orderedSink.deliver( + POLICY_RUN, List.of(named("a.pdf", "aaa")), OutputSpec.folder(out.toString())); + + assertTrue(orderedLedger.recorded); + assertTrue(Files.exists(out.resolve("a.pdf"))); + } + + @Test + void adHocDeliveriesRecordNothing() throws IOException { + Path out = tempDir.resolve("out"); + + sink.deliver(AD_HOC, List.of(named("a.pdf", "aaa")), OutputSpec.folder(out.toString())); + + Path delivered = FolderIdentities.canonicalDir(out).resolve("a.pdf"); + // No row was recorded, so any policy (including a hypothetical producer) may claim it. + assertTrue(ledger.claim("p1", delivered.toString(), "any-gate", null)); } @Test @@ -60,7 +139,7 @@ class FolderOutputSinkTest { Path out = tempDir.resolve("out"); List outputs = List.of(named("a.pdf", "first"), named("a.pdf", "second")); - sink.deliver("run-1", outputs, OutputSpec.folder(out.toString())); + sink.deliver(AD_HOC, outputs, OutputSpec.folder(out.toString())); assertTrue(Files.exists(out.resolve("a.pdf"))); assertTrue(Files.exists(out.resolve("a (1).pdf"))); @@ -72,7 +151,7 @@ class FolderOutputSinkTest { assertThrows(IllegalArgumentException.class, () -> sink.validate(noDir)); assertThrows( IllegalArgumentException.class, - () -> sink.deliver("run-1", List.of(named("a.pdf", "x")), noDir)); + () -> sink.deliver(AD_HOC, List.of(named("a.pdf", "x")), noDir)); } @Test @@ -81,7 +160,7 @@ class FolderOutputSinkTest { assertThrows(IllegalArgumentException.class, () -> sink.validate(outside)); assertThrows( IllegalArgumentException.class, - () -> sink.deliver("run-1", List.of(named("a.pdf", "x")), outside)); + () -> sink.deliver(AD_HOC, List.of(named("a.pdf", "x")), outside)); } @Test @@ -90,7 +169,7 @@ class FolderOutputSinkTest { List outputs = List.of(named("../escape.pdf", "x"), named("nested/deep.pdf", "y")); - sink.deliver("run-1", outputs, OutputSpec.folder(out.toString())); + sink.deliver(AD_HOC, outputs, OutputSpec.folder(out.toString())); // Each name is reduced to its bare form inside the target dir; nothing escapes. assertTrue(Files.exists(out.resolve("escape.pdf"))); @@ -106,4 +185,20 @@ class FolderOutputSinkTest { } }; } + + /** Fails the delivery if an output is visible at its final path before being recorded. */ + private static class VisibilityAssertingLedger extends InProcessProcessedLedger { + + private boolean recorded; + + @Override + public synchronized void recordOutput( + String policyId, String identity, String gate, String contentHash) { + assertFalse( + Files.exists(Path.of(identity)), + "output must be recorded before it is visible at its final path"); + recorded = true; + super.recordOutput(policyId, identity, gate, contentHash); + } + } } diff --git a/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/FolderWatchTriggerTest.java b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/FolderWatchTriggerTest.java index 58c5123327..2c3adf691e 100644 --- a/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/FolderWatchTriggerTest.java +++ b/app/proprietary/src/test/java/stirling/software/proprietary/policy/trigger/FolderWatchTriggerTest.java @@ -3,6 +3,7 @@ package stirling.software.proprietary.policy.trigger; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -26,6 +27,7 @@ import org.mockito.junit.jupiter.MockitoExtension; import stirling.software.common.model.ApplicationProperties; import stirling.software.proprietary.policy.engine.PolicyRunner; +import stirling.software.proprietary.policy.engine.SweepKind; import stirling.software.proprietary.policy.input.InputSource; import stirling.software.proprietary.policy.model.InputSpec; import stirling.software.proprietary.policy.model.OutputSpec; @@ -100,8 +102,8 @@ class FolderWatchTriggerTest { trigger.runForChangedDirs(Set.of(normalized("/in/a"))); - verify(policyRunner).run(a); - verify(policyRunner, never()).run(b); + verify(policyRunner).run(a, SweepKind.LIGHT); + verify(policyRunner, never()).run(eq(b), any()); } @Test @@ -112,8 +114,8 @@ class FolderWatchTriggerTest { trigger.runForChangedDirs(Set.of(normalized("/in/a"))); - verify(policyRunner).run(good); - verify(policyRunner, never()).run(bad); + verify(policyRunner).run(good, SweepKind.LIGHT); + verify(policyRunner, never()).run(eq(bad), any()); } @Test diff --git a/app/saas/src/main/resources/db/migration/saas/V32__policy_processed_files.sql b/app/saas/src/main/resources/db/migration/saas/V32__policy_processed_files.sql new file mode 100644 index 0000000000..f3a81de753 --- /dev/null +++ b/app/saas/src/main/resources/db/migration/saas/V32__policy_processed_files.sql @@ -0,0 +1,30 @@ +-- Per-policy processed-file ledger: +-- +-- policy_processed_files one row per (policy, file identity) recording the version a policy +-- last settled that file at, so folder sources track files in place +-- instead of moving them into a work directory. signature is a cheap +-- version gate (folder: size:mtime); content_hash an optional strong +-- token consulted only when the gate moves. Rows are claimed into +-- PROCESSING, settled to DONE/ERROR, flipped to INTERRUPTED at boot if +-- a run died with the JVM, and pruned once the file is gone from all of +-- the policy's sources, so the table stays near the set of files +-- currently present. +-- +-- Gated by policies.enabled like the rest of the subsystem; Hibernate ddl-auto would also create +-- this, but the migration keeps the schema explicit for the Flyway-managed deployments. + +CREATE TABLE IF NOT EXISTS policy_processed_files ( + policy_id VARCHAR(255) NOT NULL, + identity_hash VARCHAR(64) NOT NULL, + identity VARCHAR(4096), + signature VARCHAR(255) NOT NULL, + content_hash VARCHAR(64), + status VARCHAR(16) NOT NULL, + attempts SMALLINT NOT NULL DEFAULT 1, + last_seen BIGINT NOT NULL DEFAULT 0, + updated_at BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (policy_id, identity_hash) +); + +CREATE INDEX IF NOT EXISTS idx_processed_files_policy_seen + ON policy_processed_files (policy_id, last_seen); diff --git a/frontend/editor/public/locales/en-US/translation.toml b/frontend/editor/public/locales/en-US/translation.toml index 5f07f5d462..2daef44903 100644 --- a/frontend/editor/public/locales/en-US/translation.toml +++ b/frontend/editor/public/locales/en-US/translation.toml @@ -7327,6 +7327,7 @@ showMore = "Show more" sources = "Sources" [portal.policies.detail.actions] +clearHistory = "Clear processed history" delete = "Delete" editSettings = "Edit settings" pause = "Pause" @@ -7896,6 +7897,14 @@ helperText = "Absolute path Stirling watches for files to process." label = "Directory path" placeholder = "/data/incoming" +[portal.sources.types.folder.fields.identity] +helperText = "Content check reads each changed file, so renames and touches that don't alter content are not reprocessed." +label = "Change detection" + +[portal.sources.types.folder.fields.identity.options] +hash = "Size, date and content check" +stat = "Size and date modified" + [portal.sources.types.folder.fields.mode] label = "Read mode" @@ -7903,6 +7912,13 @@ label = "Read mode" consume = "Consume: process each file once" snapshot = "Snapshot: re-read the folder every run" +[portal.sources.types.folder.fields.recursive] +label = "Folder depth" + +[portal.sources.types.folder.fields.recursive.options] +all = "Include subfolders" +top = "Top level only" + [portal.sources.types.unknown] label = "Source" diff --git a/frontend/editor/src/portal/api/policies.ts b/frontend/editor/src/portal/api/policies.ts index e225449a16..a28b5d2889 100644 --- a/frontend/editor/src/portal/api/policies.ts +++ b/frontend/editor/src/portal/api/policies.ts @@ -177,6 +177,19 @@ export async function deletePolicy(id: string): Promise { ); } +/** + * DELETE /api/v1/policies/{id}/processed-history — forget which source files + * the policy has processed, so its next sweep reprocesses everything present. + */ +export async function clearProcessedHistory(id: string): Promise { + await apiClient.local.json( + `/api/v1/policies/${encodeURIComponent(id)}/processed-history`, + { + method: "DELETE", + }, + ); +} + // ── Wire-build helpers (so Policies.tsx doesn't need codec knowledge) ──────── const DEFAULT_RETRIES = 3; diff --git a/frontend/editor/src/portal/components/policies/PolicyDetailPanel.tsx b/frontend/editor/src/portal/components/policies/PolicyDetailPanel.tsx index d638cc4b7a..3faa37d076 100644 --- a/frontend/editor/src/portal/components/policies/PolicyDetailPanel.tsx +++ b/frontend/editor/src/portal/components/policies/PolicyDetailPanel.tsx @@ -23,6 +23,7 @@ interface PolicyDetailPanelProps { onRun?: () => void; onTogglePause: () => void; onDelete: () => void; + onClearHistory?: () => void; onRetry?: (item: PolicyActivityItem) => void; } @@ -106,6 +107,7 @@ export function PolicyDetailPanel({ onRun, onTogglePause, onDelete, + onClearHistory, onRetry, }: PolicyDetailPanelProps) { const { t } = useTranslation(); @@ -113,6 +115,9 @@ export function PolicyDetailPanel({ const { category, config, state, steps, stats, activity } = policy; const isPaused = state.status === "paused"; const canDelete = state.isDefault !== true; + // Processed history only exists for watched sources; editor uploads are never ledgered. + const canClearHistory = + onClearHistory !== undefined && state.sources.some((s) => s !== "editor"); const enforceItems = steps.length > 0 ? steps.map((s) => s.operation) : null; const hasEditorSource = state.sources.includes("editor"); @@ -161,6 +166,16 @@ export function PolicyDetailPanel({ {t("portal.policies.detail.actions.runNow")} )} + {canClearHistory && ( + + )}