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