mirror of
https://github.com/arsvendg/Stirling-PDF.git
synced 2026-07-16 11:23:10 +02:00
PAYG: process tracking + shadow charging engine (PR B-1) (#6477)
> 📌 **Stacked on [#6464](https://github.com/Stirling-Tools/Stirling-PDF/pull/6464)** (lineage primitives, still in review). #6469 has merged so its commits are no longer in this PR's diff. Once #6464 merges, a final rebase collapses the lineage-primitives commits out of this diff too — leaving only the B-1 work. ## What this is Process tracking + shadow charging engine. Bundles PR-I7 service half with the non-filter piece of PR-I7a so the pieces ship together — none of them is useful in isolation. **Review focus:** the new files in: - \`app/saas/src/main/java/stirling/software/saas/payg/job/\` (\`JobService\`, \`JobContext\`, \`JoinOrOpenResult\`, \`StaleJobCloser\`) - \`app/saas/src/main/java/stirling/software/saas/payg/charge/\` (\`JobChargeService\`, \`ChargeContext\`, \`ChargeOutcome\`, \`JobInput\`) - \`app/saas/src/main/java/stirling/software/saas/payg/lineage/LineagePruneScheduler.java\` - their tests The 8 files inherited from #6464 (lineage primitives) are unchanged from there — they ride along in this diff until #6464 lands. The remaining work for shadow-in-staging is the ingress/egress filter that wires controllers into this engine — that's PR B-2. ## Scope ### \`JobService\` — persistence + lineage policy - **\`joinOrOpen\`** — the multi-input "any-match-joins, newest wins" rule. Hash every input via the lineage detector; if any matches an open process in the workflow window, attach to the one with the freshest \`lastStepAt\`. Step-limit overflow on the matched job spawns a fresh process; the new job's input signatures are still recorded so \`mostRecentMatchWins\` routes future calls forward. - **\`recordOutput\`** — post-tool-success path. Records OUTPUT signatures so the next call that takes this file as input lineage-matches into the same process. - **\`appendStep\`** — audit-trail step row written after a tool completes. - **\`close\`** — idempotent; safe to call from multiple paths (explicit, FE on-unload, scheduler). Returns the same row on re-close, no state mutation. - **\`findStale\` / \`closeStale\`** — workflow-window-based stale closure used by the scheduler. ### \`JobChargeService\` — the orchestrator (shadow variant) \`openProcess\` resolves the effective policy via \`PricingPolicyService\` (now in main via #6469), derives the step-limit for the current \`JobSource\` (with a defensive fallback if the policy is missing an entry), delegates to \`JobService.joinOrOpen\`, and on OPENED runs the \`DocumentClassifier\` + writes a \`payg_shadow_charge\` row. Applies the policy-level \`minChargeUnits\` floor per design § 3.4. Shadow variant only — never debits the ledger, never posts a Stripe meter event. The real-charging follow-up reuses the same orchestration and swaps the side-effect. \`legacyCreditsCharged\` on the shadow row stays \`0\` until the legacy \`CreditService\` is wired in (PR B-2), where the comparison becomes meaningful. ### Schedulers (both plain \`@Scheduled\`) - **\`StaleJobCloser\`** — fixed-rate 60 s. Closes \`OPEN\` jobs idle past the workflow window. API users never have to call close explicitly — this is the safety net. - **\`LineagePruneScheduler\`** — hourly cron, retention 1 h. Deletes \`job_artifact_hash\` rows older than the retention window. - **No \`@SchedulerLock\` / no \`shedlock\` table** — consistent with the 5 existing unguarded \`@Scheduled\` tasks in \`:saas\` (\`CreditResetScheduler\` and friends, none of which are guarded today). Cluster-correctness across all 7 saas schedulers is tracked in design § 9 as a separate focused cleanup. Underlying operations are idempotent — duplicate firings on multi-pod would be wasted DB load, not data corruption. ### Records (call-shape glue for PR B-2's filter) - \`JobContext\` / \`JoinOrOpenResult\` — input/output for \`JobService\`. - \`ChargeContext\` / \`ChargeOutcome\` — input/output for \`JobChargeService\`. - \`JobInput\` — paired \`(MultipartFile, materialised Path)\` so the upcoming ingress filter can pass both views without re-materialising. ## Tests **26 new, all green.** - 14 × \`JobServiceTest\` — no-match → opens new, single-match → joins existing, multi-input any-match-joins, multi-match newest-wins (older job never even looked up), step-limit hit spawns fresh job (and original \`stepCount\` is NOT mutated), empty inputs reject, stale-signature handling, recordOutput delegation, close idempotency, closeStale, appendStep persistence. - 7 × \`JobChargeServiceTest\` — JOINED skips classifier + shadow write entirely, OPENED writes shadow row + classifies (single + multi file paths), \`minChargeUnits\` floor applied, step-limit resolved per-\`JobSource\` from policy, missing source entry falls back to conservative default of 10. - 2 × \`StaleJobCloserTest\`, 3 × \`LineagePruneSchedulerTest\` — scheduler-wiring smoke + constructor-validation tests. \`ENABLE_SAAS=true ./gradlew :saas:test\` — BUILD SUCCESSFUL. ## What's not in this PR (lands in PR B-2) - **Tool ingress/egress servlet filter.** The highest-risk piece — materialises the request body into a \`JobInput\`, calls \`JobChargeService.openProcess\` from every \`@AutoJobPostMapping\`, records OUTPUT after success. Edge cases to validate: multipart parts, async controllers, streaming responses, errored 5xx paths, very large files. Design decisions for the filter are being worked through in \`notes/PAYG_FILTER_DESIGN.md\` before any code is written. - **Wire shadow path into legacy \`CreditService\`.** Every legacy debit writes a comparison row carrying both the PAYG would-be units and the legacy actual credits, populating \`diffPct\`. ## Design doc \`notes/PAYG_DESIGN.md\` — PR-I7 + PR-I7a status updated to reflect this bundle. § 9 carries the cluster-correctness deferral note alongside the existing LISTEN/NOTIFY trade-off note. § 7.5.1 readiness summary shows the path now needs just **1 more PR** (the filter half + CreditService wire-in, both bundled into PR B-2).
This commit is contained in:
ConnorYoh
@@ -0,0 +1,26 @@
|
||||
package stirling.software.saas.payg.charge;
|
||||
|
||||
import stirling.software.saas.payg.model.JobSource;
|
||||
import stirling.software.saas.payg.model.ProcessType;
|
||||
|
||||
/**
|
||||
* Per-call context for {@link JobChargeService#openProcess}. Carries the caller's identity and what
|
||||
* kind of process this is. Does NOT carry policy fields — the charge service resolves the effective
|
||||
* policy from {@code PricingPolicyService} so a stale snapshot from the caller can't desync from
|
||||
* the live policy.
|
||||
*/
|
||||
public record ChargeContext(
|
||||
Long ownerUserId, Long ownerTeamId, JobSource source, ProcessType processType) {
|
||||
|
||||
public ChargeContext {
|
||||
if (ownerUserId == null) {
|
||||
throw new IllegalArgumentException("ownerUserId is required");
|
||||
}
|
||||
if (source == null) {
|
||||
throw new IllegalArgumentException("source is required");
|
||||
}
|
||||
if (processType == null) {
|
||||
throw new IllegalArgumentException("processType is required");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,19 @@
|
||||
package stirling.software.saas.payg.charge;
|
||||
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
* Result of {@link JobChargeService#openProcess}. {@code processId} is the job the caller's tool
|
||||
* call belongs to (newly opened or joined). {@code units} is what would have been debited had the
|
||||
* call been in PAYG (live) mode — for OPENED, the actual classification × policy units; for JOINED,
|
||||
* always 0 since the parent process already paid.
|
||||
*/
|
||||
public record ChargeOutcome(UUID processId, int units, Disposition disposition) {
|
||||
|
||||
public enum Disposition {
|
||||
/** Started a new process; {@code units} reflects the would-be charge. */
|
||||
OPENED,
|
||||
/** Joined an existing process within window; no incremental charge. */
|
||||
JOINED
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,143 @@
|
||||
package stirling.software.saas.payg.charge;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Path;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import stirling.software.saas.payg.docs.DocumentClassifier;
|
||||
import stirling.software.saas.payg.docs.DocumentMetrics;
|
||||
import stirling.software.saas.payg.job.JobContext;
|
||||
import stirling.software.saas.payg.job.JobService;
|
||||
import stirling.software.saas.payg.job.JoinOrOpenResult;
|
||||
import stirling.software.saas.payg.model.JobSource;
|
||||
import stirling.software.saas.payg.policy.PricingPolicy;
|
||||
import stirling.software.saas.payg.policy.PricingPolicyService;
|
||||
import stirling.software.saas.payg.repository.PaygShadowChargeRepository;
|
||||
import stirling.software.saas.payg.shadow.PaygShadowCharge;
|
||||
|
||||
/**
|
||||
* Orchestrates a tool call's open-process decision: look up the team's effective policy, resolve
|
||||
* the step-limit ceiling for this caller surface, delegate the join-or-open decision to {@link
|
||||
* JobService}, and — when a new process opens — compute the would-be charge and record a {@link
|
||||
* PaygShadowCharge} row for comparison against the legacy engine.
|
||||
*
|
||||
* <p>Shadow mode only: this service never debits the wallet ledger or posts a Stripe meter event.
|
||||
* The real-charging path lives in a separate follow-up and reuses the same orchestration — only the
|
||||
* side-effect (shadow row vs ledger entry + Stripe call) differs.
|
||||
*
|
||||
* <p>The {@code legacyCreditsCharged} field on the shadow row is set to {@code 0} here. When the
|
||||
* legacy {@code CreditService} is wired to call this service (separate PR), the legacy debit amount
|
||||
* becomes available and {@code diffPct} can be computed against it; until then the shadow row
|
||||
* captures the PAYG units only.
|
||||
*/
|
||||
@Service
|
||||
@Profile("saas")
|
||||
@Slf4j
|
||||
public class JobChargeService {
|
||||
|
||||
private final JobService jobService;
|
||||
private final PricingPolicyService policyService;
|
||||
private final DocumentClassifier classifier;
|
||||
private final PaygShadowChargeRepository shadowRepository;
|
||||
|
||||
public JobChargeService(
|
||||
JobService jobService,
|
||||
PricingPolicyService policyService,
|
||||
DocumentClassifier classifier,
|
||||
PaygShadowChargeRepository shadowRepository) {
|
||||
this.jobService = Objects.requireNonNull(jobService, "jobService");
|
||||
this.policyService = Objects.requireNonNull(policyService, "policyService");
|
||||
this.classifier = Objects.requireNonNull(classifier, "classifier");
|
||||
this.shadowRepository = Objects.requireNonNull(shadowRepository, "shadowRepository");
|
||||
}
|
||||
|
||||
/**
|
||||
* Open a process (or join an existing one) for this tool call. Side effects: persists a {@code
|
||||
* ProcessingJob} row plus input signatures, and — on OPENED — writes a {@code
|
||||
* payg_shadow_charge} row carrying the would-be PAYG units.
|
||||
*/
|
||||
@Transactional
|
||||
public ChargeOutcome openProcess(ChargeContext ctx, List<JobInput> inputs) throws IOException {
|
||||
Objects.requireNonNull(ctx, "ctx");
|
||||
Objects.requireNonNull(inputs, "inputs");
|
||||
if (inputs.isEmpty()) {
|
||||
throw new IllegalArgumentException("inputs must not be empty");
|
||||
}
|
||||
|
||||
PricingPolicy policy = policyService.getEffectivePolicy(ctx.ownerTeamId());
|
||||
int stepLimit = resolveStepLimit(policy, ctx.source());
|
||||
|
||||
JobContext jobCtx =
|
||||
new JobContext(
|
||||
ctx.ownerUserId(),
|
||||
ctx.ownerTeamId(),
|
||||
ctx.source(),
|
||||
ctx.processType(),
|
||||
policy.getId(),
|
||||
stepLimit);
|
||||
|
||||
List<Path> paths = inputs.stream().map(JobInput::path).toList();
|
||||
JoinOrOpenResult result = jobService.joinOrOpen(jobCtx, paths);
|
||||
|
||||
if (result.disposition() == JoinOrOpenResult.Disposition.JOINED) {
|
||||
return new ChargeOutcome(result.job().getId(), 0, ChargeOutcome.Disposition.JOINED);
|
||||
}
|
||||
|
||||
int units = computeUnits(inputs, policy);
|
||||
result.job().setDocUnits(units);
|
||||
|
||||
recordShadowRow(ctx, result.job().getId(), policy.getId(), units);
|
||||
|
||||
return new ChargeOutcome(result.job().getId(), units, ChargeOutcome.Disposition.OPENED);
|
||||
}
|
||||
|
||||
private int resolveStepLimit(PricingPolicy policy, JobSource source) {
|
||||
Integer fromPolicy =
|
||||
policy.getStepLimits() == null ? null : policy.getStepLimits().get(source);
|
||||
if (fromPolicy != null && fromPolicy > 0) {
|
||||
return fromPolicy;
|
||||
}
|
||||
// Defensive default — every JobSource should have an entry per the V12 seed, but a
|
||||
// hand-edited policy could be missing one. Fall back to the smallest documented limit
|
||||
// (10 — WEB/API/DESKTOP_APP default) so an admin slip-up never spawns unbounded chains.
|
||||
log.debug(
|
||||
"PricingPolicy {} missing stepLimit for source={}; using fallback of 10.",
|
||||
policy.getId(),
|
||||
source);
|
||||
return 10;
|
||||
}
|
||||
|
||||
private int computeUnits(List<JobInput> inputs, PricingPolicy policy) {
|
||||
List<MultipartFile> multiparts = inputs.stream().map(JobInput::multipart).toList();
|
||||
DocumentMetrics metrics =
|
||||
multiparts.size() == 1
|
||||
? classifier.classify(multiparts.get(0), policy)
|
||||
: classifier.classify(multiparts, policy);
|
||||
// Apply the policy-level minChargeUnits floor per design § 3.4. The classifier returns
|
||||
// raw docUnits with a "non-empty input → ≥1" floor; the charge formula's
|
||||
// max(min_charge_units, docUnits) layers on top.
|
||||
return Math.max(policy.getMinChargeUnits(), metrics.docUnits());
|
||||
}
|
||||
|
||||
private void recordShadowRow(
|
||||
ChargeContext ctx, java.util.UUID jobId, Long policyId, int units) {
|
||||
PaygShadowCharge row = new PaygShadowCharge();
|
||||
row.setTeamId(ctx.ownerTeamId());
|
||||
row.setJobId(jobId);
|
||||
row.setPolicyId(policyId);
|
||||
row.setPaygUnits(units);
|
||||
// No legacy comparison yet — wired when the shadow path is connected to the legacy
|
||||
// CreditService in the follow-up PR. Until then, diff stays at 0.
|
||||
row.setLegacyCreditsCharged(0);
|
||||
row.setDiffPct(0);
|
||||
shadowRepository.save(row);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
package stirling.software.saas.payg.charge;
|
||||
|
||||
import java.nio.file.Path;
|
||||
|
||||
import org.springframework.web.multipart.MultipartFile;
|
||||
|
||||
/**
|
||||
* One input file to a tool call, materialised. {@code path} is the on-disk copy the lineage
|
||||
* detector hashes; {@code multipart} carries the size + content-type metadata the classifier needs.
|
||||
* The pair is constructed by the ingress filter (PR-I7a filter) which materialises the request body
|
||||
* exactly once. Tests construct it from a fixture file plus a {@code MockMultipartFile} wrapping
|
||||
* the same bytes.
|
||||
*/
|
||||
public record JobInput(MultipartFile multipart, Path path) {
|
||||
|
||||
public JobInput {
|
||||
if (multipart == null) {
|
||||
throw new IllegalArgumentException("multipart is required");
|
||||
}
|
||||
if (path == null) {
|
||||
throw new IllegalArgumentException("path is required");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package stirling.software.saas.payg.job;
|
||||
|
||||
import stirling.software.saas.payg.model.JobSource;
|
||||
import stirling.software.saas.payg.model.ProcessType;
|
||||
|
||||
/**
|
||||
* Input to {@link JobService#joinOrOpen}: who owns the request, what process shape it is, which
|
||||
* policy version applies, and the step-limit ceiling derived from that policy for this caller
|
||||
* surface. {@code stepLimit} is resolved by the caller (typically {@code JobChargeService}) rather
|
||||
* than re-fetched here so the service stays free of policy-lookup concerns.
|
||||
*/
|
||||
public record JobContext(
|
||||
Long ownerUserId,
|
||||
Long ownerTeamId,
|
||||
JobSource source,
|
||||
ProcessType processType,
|
||||
Long policyId,
|
||||
int stepLimit) {
|
||||
|
||||
public JobContext {
|
||||
if (ownerUserId == null) {
|
||||
throw new IllegalArgumentException("ownerUserId is required");
|
||||
}
|
||||
if (source == null) {
|
||||
throw new IllegalArgumentException("source is required");
|
||||
}
|
||||
if (processType == null) {
|
||||
throw new IllegalArgumentException("processType is required");
|
||||
}
|
||||
if (policyId == null) {
|
||||
throw new IllegalArgumentException("policyId is required");
|
||||
}
|
||||
if (stepLimit <= 0) {
|
||||
throw new IllegalArgumentException("stepLimit must be > 0");
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,243 @@
|
||||
package stirling.software.saas.payg.job;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.nio.file.Path;
|
||||
import java.time.Duration;
|
||||
import java.time.LocalDateTime;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Comparator;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.UUID;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.stereotype.Service;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import stirling.software.saas.payg.lineage.HashLineageDetector;
|
||||
import stirling.software.saas.payg.lineage.LineageMatch;
|
||||
import stirling.software.saas.payg.model.ArtifactKind;
|
||||
import stirling.software.saas.payg.model.JobStatus;
|
||||
import stirling.software.saas.payg.model.JobStepStatus;
|
||||
import stirling.software.saas.payg.repository.ProcessingJobRepository;
|
||||
import stirling.software.saas.payg.repository.ProcessingJobStepRepository;
|
||||
|
||||
/**
|
||||
* Persistence + lineage policy layer for the PAYG process model. The hot-path entry point is {@link
|
||||
* #joinOrOpen}, which applies the "any-match-joins, newest wins" multi-input rule.
|
||||
*
|
||||
* <p>The charge service ({@code JobChargeService}) orchestrates around this — it resolves the
|
||||
* pricing policy, computes the step-limit ceiling for the current {@code JobSource}, calls {@link
|
||||
* #joinOrOpen}, and decides whether to record a shadow charge based on the resulting {@link
|
||||
* JoinOrOpenResult.Disposition}.
|
||||
*/
|
||||
@Service
|
||||
@Profile("saas")
|
||||
@Slf4j
|
||||
public class JobService {
|
||||
|
||||
private final HashLineageDetector detector;
|
||||
private final ProcessingJobRepository jobRepository;
|
||||
private final ProcessingJobStepRepository stepRepository;
|
||||
private final Duration workflowWindow;
|
||||
|
||||
public JobService(
|
||||
HashLineageDetector detector,
|
||||
ProcessingJobRepository jobRepository,
|
||||
ProcessingJobStepRepository stepRepository,
|
||||
@Value("${payg.lineage.workflow-window:PT5M}") Duration workflowWindow) {
|
||||
this.detector = Objects.requireNonNull(detector, "detector");
|
||||
this.jobRepository = Objects.requireNonNull(jobRepository, "jobRepository");
|
||||
this.stepRepository = Objects.requireNonNull(stepRepository, "stepRepository");
|
||||
Objects.requireNonNull(workflowWindow, "workflowWindow");
|
||||
if (workflowWindow.isNegative() || workflowWindow.isZero()) {
|
||||
throw new IllegalArgumentException(
|
||||
"payg.lineage.workflow-window must be positive, got " + workflowWindow);
|
||||
}
|
||||
this.workflowWindow = workflowWindow;
|
||||
}
|
||||
|
||||
/**
|
||||
* Decide whether {@code inputs} should join an existing open process or start a new one, and
|
||||
* persist that decision. Returns the job to attach to plus the disposition.
|
||||
*
|
||||
* <p>Rule: hash every input. If any input's signatures match an open process owned by the same
|
||||
* user within the workflow window, attach to that process. When multiple inputs match different
|
||||
* processes, the one with the freshest {@code lastStepAt} wins — preserves the
|
||||
* "most-recent-job-wins" invariant the lineage primitives test for. When the matched process is
|
||||
* already at its step-limit ceiling, fall through to opening a fresh process; the new job's
|
||||
* input signatures are still recorded so downstream calls lineage-match to it.
|
||||
*
|
||||
* <p>Race note: the read of candidate matches and the subsequent write happen in one
|
||||
* transaction at default isolation (READ COMMITTED in Postgres). A concurrent admin write or
|
||||
* another tool call could change the match set between read and commit. For shadow mode the
|
||||
* resulting diff is a low-grade comparison artefact; tighten isolation when real charging lands
|
||||
* if it materialises as a real-money issue.
|
||||
*/
|
||||
@Transactional
|
||||
public JoinOrOpenResult joinOrOpen(JobContext ctx, List<Path> inputs) throws IOException {
|
||||
Objects.requireNonNull(ctx, "ctx");
|
||||
Objects.requireNonNull(inputs, "inputs");
|
||||
if (inputs.isEmpty()) {
|
||||
throw new IllegalArgumentException("inputs must not be empty");
|
||||
}
|
||||
|
||||
Optional<LineageMatch> bestMatch = findBestMatch(ctx.ownerUserId(), inputs);
|
||||
|
||||
if (bestMatch.isPresent()) {
|
||||
ProcessingJob existing =
|
||||
jobRepository
|
||||
.findById(bestMatch.get().jobId())
|
||||
.orElseThrow(
|
||||
() ->
|
||||
new IllegalStateException(
|
||||
"Lineage match returned jobId="
|
||||
+ bestMatch.get().jobId()
|
||||
+ " but no such ProcessingJob row"
|
||||
+ " exists (stale signature?)"));
|
||||
if (existing.getStepCount() < ctx.stepLimit()) {
|
||||
return joinExisting(existing, inputs);
|
||||
}
|
||||
// Step-limit hit: spawn a fresh job. The new job will share input signatures with
|
||||
// the existing chain so future tool calls still lineage-match into the workflow,
|
||||
// but the freshest jobLastStepAt wins so they target the new one.
|
||||
log.debug(
|
||||
"Job {} hit step limit ({}); spawning new process within workflow.",
|
||||
existing.getId(),
|
||||
ctx.stepLimit());
|
||||
}
|
||||
|
||||
return openFresh(ctx, inputs);
|
||||
}
|
||||
|
||||
/**
|
||||
* Records {@code outputFile}'s signatures against {@code jobId} as {@code OUTPUT}. Called after
|
||||
* a tool runs successfully — failed tools don't record outputs. Multi-output tools call this
|
||||
* once per output.
|
||||
*/
|
||||
@Transactional
|
||||
public void recordOutput(UUID jobId, Path outputFile) throws IOException {
|
||||
Objects.requireNonNull(jobId, "jobId");
|
||||
Objects.requireNonNull(outputFile, "outputFile");
|
||||
detector.record(jobId, outputFile, ArtifactKind.OUTPUT);
|
||||
}
|
||||
|
||||
/** Appends an audit-trail step row after a tool call completes. */
|
||||
@Transactional
|
||||
public ProcessingJobStep appendStep(
|
||||
UUID jobId,
|
||||
String toolId,
|
||||
JobStepStatus status,
|
||||
Integer inputPages,
|
||||
Long inputBytes,
|
||||
String errorCode) {
|
||||
Objects.requireNonNull(jobId, "jobId");
|
||||
Objects.requireNonNull(toolId, "toolId");
|
||||
Objects.requireNonNull(status, "status");
|
||||
ProcessingJobStep step = new ProcessingJobStep();
|
||||
step.setJobId(jobId);
|
||||
step.setToolId(toolId);
|
||||
step.setStatus(status);
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
step.setStartedAt(now);
|
||||
step.setCompletedAt(now);
|
||||
step.setInputPages(inputPages);
|
||||
step.setInputBytes(inputBytes);
|
||||
step.setErrorCode(errorCode);
|
||||
return stepRepository.save(step);
|
||||
}
|
||||
|
||||
/**
|
||||
* Closes a job. Idempotent — closing an already-closed job is a silent no-op returning the
|
||||
* existing row, so multiple close paths (explicit caller, FE on-unload, stale scheduler) all
|
||||
* compose safely.
|
||||
*/
|
||||
@Transactional
|
||||
public ProcessingJob close(UUID jobId) {
|
||||
Objects.requireNonNull(jobId, "jobId");
|
||||
ProcessingJob job =
|
||||
jobRepository
|
||||
.findById(jobId)
|
||||
.orElseThrow(
|
||||
() ->
|
||||
new IllegalArgumentException(
|
||||
"No ProcessingJob with id " + jobId));
|
||||
if (job.getStatus() != JobStatus.OPEN) {
|
||||
return job;
|
||||
}
|
||||
job.setStatus(JobStatus.CLOSED);
|
||||
job.setClosedAt(LocalDateTime.now());
|
||||
return jobRepository.save(job);
|
||||
}
|
||||
|
||||
/** Returns open jobs whose {@code last_step_at} is older than the workflow window. */
|
||||
@Transactional(readOnly = true)
|
||||
public List<ProcessingJob> findStale() {
|
||||
return jobRepository.findStale(JobStatus.OPEN, LocalDateTime.now().minus(workflowWindow));
|
||||
}
|
||||
|
||||
/**
|
||||
* Closes all jobs returned by {@link #findStale}. Returns the count closed. Each transition
|
||||
* goes through {@link #close} semantics (idempotent), so re-running this on an empty stale set
|
||||
* is safe.
|
||||
*/
|
||||
@Transactional
|
||||
public int closeStale() {
|
||||
List<ProcessingJob> stale = findStale();
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
for (ProcessingJob j : stale) {
|
||||
j.setStatus(JobStatus.CLOSED);
|
||||
j.setClosedAt(now);
|
||||
}
|
||||
if (!stale.isEmpty()) {
|
||||
jobRepository.saveAll(stale);
|
||||
}
|
||||
return stale.size();
|
||||
}
|
||||
|
||||
private Optional<LineageMatch> findBestMatch(Long userId, List<Path> inputs)
|
||||
throws IOException {
|
||||
List<LineageMatch> matches = new ArrayList<>(inputs.size());
|
||||
for (Path input : inputs) {
|
||||
detector.detect(userId, input).ifPresent(matches::add);
|
||||
}
|
||||
return matches.stream().max(Comparator.comparing(LineageMatch::jobLastStepAt));
|
||||
}
|
||||
|
||||
private JoinOrOpenResult joinExisting(ProcessingJob existing, List<Path> inputs)
|
||||
throws IOException {
|
||||
existing.setStepCount(existing.getStepCount() + 1);
|
||||
existing.setLastStepAt(LocalDateTime.now());
|
||||
ProcessingJob saved = jobRepository.save(existing);
|
||||
recordAllInputs(saved.getId(), inputs);
|
||||
return new JoinOrOpenResult(saved, JoinOrOpenResult.Disposition.JOINED);
|
||||
}
|
||||
|
||||
private JoinOrOpenResult openFresh(JobContext ctx, List<Path> inputs) throws IOException {
|
||||
ProcessingJob fresh = new ProcessingJob();
|
||||
fresh.setId(UUID.randomUUID());
|
||||
fresh.setOwnerUserId(ctx.ownerUserId());
|
||||
fresh.setOwnerTeamId(ctx.ownerTeamId());
|
||||
fresh.setProcessType(ctx.processType());
|
||||
fresh.setSource(ctx.source());
|
||||
fresh.setPolicyId(ctx.policyId());
|
||||
fresh.setStepCount(1);
|
||||
LocalDateTime now = LocalDateTime.now();
|
||||
fresh.setStartedAt(now);
|
||||
fresh.setLastStepAt(now);
|
||||
fresh.setStatus(JobStatus.OPEN);
|
||||
ProcessingJob saved = jobRepository.save(fresh);
|
||||
recordAllInputs(saved.getId(), inputs);
|
||||
return new JoinOrOpenResult(saved, JoinOrOpenResult.Disposition.OPENED);
|
||||
}
|
||||
|
||||
private void recordAllInputs(UUID jobId, List<Path> inputs) throws IOException {
|
||||
for (Path input : inputs) {
|
||||
detector.record(jobId, input, ArtifactKind.INPUT);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package stirling.software.saas.payg.job;
|
||||
|
||||
/**
|
||||
* Output of {@link JobService#joinOrOpen}. {@code OPENED} means the charge layer should compute
|
||||
* units and record a shadow charge (or, post-shadow, actually debit). {@code JOINED} means the call
|
||||
* attached to an existing process — no further charge, just an audit-trail step append.
|
||||
*
|
||||
* <p>Step-limit overflow on a matched job surfaces as {@code OPENED}: the lineage tree is preserved
|
||||
* via shared input signatures, but the limit means we start a fresh billable process.
|
||||
*/
|
||||
public record JoinOrOpenResult(ProcessingJob job, Disposition disposition) {
|
||||
|
||||
public enum Disposition {
|
||||
OPENED,
|
||||
JOINED
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,36 @@
|
||||
package stirling.software.saas.payg.job;
|
||||
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
/**
|
||||
* Auto-closes {@code OPEN} jobs whose {@code last_step_at} is older than the workflow window. Runs
|
||||
* every minute. API users never have to call {@code close()} explicitly — this scheduler is the
|
||||
* safety net.
|
||||
*
|
||||
* <p>Single-fire only at V1: not {@code @SchedulerLock}'d, consistent with the other
|
||||
* {@code @Scheduled} tasks in {@code :saas} (none of them are guarded against multi-pod
|
||||
* double-fires today either). Multi-pod cluster-correctness for all schedulers is tracked in design
|
||||
* § 9 as a separate cleanup. The underlying {@code closeStale()} call is idempotent — duplicate
|
||||
* firings read an empty stale set on the second pod, no data corruption risk.
|
||||
*/
|
||||
@Component
|
||||
@Profile("saas")
|
||||
@RequiredArgsConstructor
|
||||
@Slf4j
|
||||
public class StaleJobCloser {
|
||||
|
||||
private final JobService jobService;
|
||||
|
||||
@Scheduled(fixedRateString = "${payg.job.stale-close-interval-ms:60000}")
|
||||
public void closeStale() {
|
||||
int closed = jobService.closeStale();
|
||||
if (closed > 0) {
|
||||
log.info("StaleJobCloser closed {} job(s) idle past the workflow window.", closed);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,54 @@
|
||||
package stirling.software.saas.payg.lineage;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.time.Instant;
|
||||
import java.util.Objects;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.Profile;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
/**
|
||||
* Deletes {@code job_artifact_hash} rows older than the configured retention window. The workflow
|
||||
* window is 5 minutes; we keep an extra hour of headroom as a safety margin (covers clock skew,
|
||||
* in-flight requests, brief outages of this scheduler). Runs hourly.
|
||||
*
|
||||
* <p>Like {@link stirling.software.saas.payg.job.StaleJobCloser}, this is not
|
||||
* {@code @SchedulerLock}'d — duplicate firings on multi-pod deploys would both run a {@code DELETE
|
||||
* WHERE created_at < cutoff}, the second seeing zero rows to delete. Idempotent, wasted IO at
|
||||
* worst.
|
||||
*/
|
||||
@Component
|
||||
@Profile("saas")
|
||||
@Slf4j
|
||||
public class LineagePruneScheduler {
|
||||
|
||||
private final JobLineageStore store;
|
||||
private final Duration retention;
|
||||
|
||||
public LineagePruneScheduler(
|
||||
JobLineageStore store, @Value("${payg.lineage.retention:PT1H}") Duration retention) {
|
||||
this.store = Objects.requireNonNull(store, "store");
|
||||
Objects.requireNonNull(retention, "retention");
|
||||
if (retention.isNegative() || retention.isZero()) {
|
||||
throw new IllegalArgumentException(
|
||||
"payg.lineage.retention must be positive, got " + retention);
|
||||
}
|
||||
this.retention = retention;
|
||||
}
|
||||
|
||||
@Scheduled(cron = "${payg.lineage.prune-cron:0 0 * * * *}", zone = "UTC")
|
||||
public void prune() {
|
||||
Instant cutoff = Instant.now().minus(retention);
|
||||
int deleted = store.pruneOlderThan(cutoff);
|
||||
if (deleted > 0) {
|
||||
log.info(
|
||||
"LineagePruneScheduler deleted {} job_artifact_hash row(s) older than {}.",
|
||||
deleted,
|
||||
cutoff);
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user