Durable Batch Reviews with Spring AI and PostgreSQL

By seif.ibrahim AI & Development java spring September 14, 2026

In the previous series, I built a listing-quality API with Spring AI. A seller sends a listing, the application asks a model to review it, and the API returns a validated response.

That is a useful place to start. For a seller editing one listing, the synchronous endpoint may be all we need. But a catalog import changes the workflow: the seller submits many listings and wants to come back to the results. One model call is slow, another returns invalid output, and the application might restart halfway through.

Which listings finished? Which failed? Do we need to review the successful ones again?

Returning from the controller early is easy. Keeping a durable record of the work, recovering it after a restart, and letting the seller retrieve the result require a different contract.

In this article, we will add a durable batch API to our existing Spring Boot application. The seller gets one batch ID to track progress. Each listing gets its own background job and validated result. The API acknowledges the batch only after storing every child job and its scheduled task.

This is the first article in Reliable AI Workflows with Java and Spring. It builds on the Seller Listing Quality API, but you can use the current repository and the commands below without completing the previous series.

You will need Docker for PostgreSQL and a configured model provider. Adding durable jobs makes PostgreSQL a startup dependency of this deployable, including for synchronous reviews. I chose one application with one startup contract for this example. Keeping the synchronous API available independently of the jobs database would require a different deployment or feature-configuration boundary.

TL;DR – SmtC

Too Long; Didn’t Read – Show me the Code: https://github.com/iseif/listing-quality

What we will build

The batch workflow uses three endpoints:

Endpoint	Purpose
`POST /api/listings/review-batches`	Accept up to 100 listings atomically
`GET /api/listings/review-batches/{batchId}`	Read persisted progress counts
`GET /api/listings/review-batches/{batchId}/items`	Page through individual results and errors

The existing synchronous endpoint remains unchanged. This article also introduces POST /api/listings/review-jobs to submit an individual background review and GET /api/listings/review-jobs/{jobId} to check its status and retrieve the result. The batch endpoint builds on the same job-processing infrastructure. All execution paths use ListingReviewService for prompt rendering, provider selection, and response validation. The worker uses the stored listing snapshot; only idempotency comparison uses normalized values. Shared telemetry distinguishes synchronous calls from background execution with an entrypoint tag.

For the first increment, the API and worker run in the same application process. PostgreSQL stores the job and its result, and db-scheduler manages persistent task execution.

POST batch of listings
    |
    +--> Commit batch + N review jobs + N scheduled tasks together
    |        |
    |        +--> Shared workers --> One ListingReviewService call per job
    |                                    |
    |                                    +--> Persist each result or safe error
    |
    +--> 202 Accepted + batch status and items URLs

GET status URL --> Aggregate persisted child states
GET items URL  --> Page through results in accepted item order

The example uses Java 25, Spring Boot 4.1.1, Spring AI 2.0.1, db-scheduler 16.12.0, and PostgreSQL 18.6. These are the versions used for verification. The repository contains the full implementation and tests.

Start the database and application

From the repository root:

docker compose -f review-jobs/compose.yaml up -d --wait

This starts a dedicated PostgreSQL instance on localhost port 5434. Its named volume preserves jobs across container restarts. It is separate from the policy database introduced in the RAG article.

Set GEMINI_API_KEY in your environment, then start the application:

./mvnw -pl listing-quality-service spring-boot:run -Dspring-boot.run.profiles=gemini

You can also use the existing OpenAI, Ollama, or oMLX profiles. The background handler does not depend on a provider SDK.

The current version of the listing-quality service requires the jobs database at startup, including when calling its synchronous endpoint. Flyway applies the job migrations. If the database is unavailable, startup fails instead of accepting work the application cannot persist.

If port 5434 is occupied, configure JOBS_DB_PORT for Compose and the corresponding JOBS_DB_URL for the Java application. The runbook includes those settings.

How the code is organized

The application uses feature-first packages: listingreview, reviewjobs, and policy. Each feature groups its own technical layers instead of adding every controller and service to an application-wide folder.

dev.iseif.listingquality
├── listingreview
│   ├── config
│   ├── controller
│   ├── model
│   ├── observability
│   ├── prompt
│   └── service
├── reviewjobs
│   ├── config
│   ├── controller
│   ├── model
│   ├── persistence
│   ├── scheduling
│   └── service
├── policy
├── config
├── web
└── observability

The dependency goes from reviewjobs to listingreview. The review logic does not know about scheduler tasks or job tables. ListingReviewAiConfiguration and PolicyAiConfiguration each configure their own feature’s client. Application-wide API versioning stays in config, shared exception handling in web, and actuator observation filtering in observability. There is no generic shared package collecting unrelated classes.

Inside reviewjobs, ListingReviewBatchController handles the HTTP contract. ListingReviewBatchService owns atomic acceptance and batch idempotency. Its JDBC repository reads membership and progress. The existing job repository and handler still own each listing’s execution state. The scheduler adapter contains db-scheduler-specific code.

There is no second AI implementation. The worker calls our existing application service, which still delegates model access to ListingReviewGenerator. The executor owns local concurrency and timeout behavior. ReviewJobTelemetry owns job measurements. These are separate responsibilities we can test without replacing the business flow with mocks.

The batch is not another scheduled task. I do not put a loop over the whole catalog inside one worker, and I do not send every listing in one model prompt. Independent child jobs give us a smaller recovery unit: a failed listing does not invalidate a completed sibling. They also share the existing worker limit instead of creating an executor per batch.

Accepted means the work was stored

Submit the three-listing example:

curl -i http://localhost:8080/api/listings/review-batches \
  -H 'Content-Type: application/json' \
  -H 'X-API-Version: 1' \
  -H 'Idempotency-Key: catalog-review-001' \
  --data-binary @review-jobs/catalog-batch.json

Each item wraps the existing ListingDraft contract with a caller-provided itemId. The complete example file contains a keyboard, mouse, and headset. Here is one item in the same request shape:

{
  "items": [
    {
      "itemId": "keyboard-001",
      "listing": {
        "title": "Wireless keyboard",
        "description": "Used wireless keyboard. Includes its USB receiver. All keys work.",
        "category": "Computer accessories",
        "price": 45.00,
        "attributes": {"brand": "KeyPro", "condition": "used"}
      }
    }
  ]
}

The API returns 202 Accepted, with a Location header pointing to the status resource. HTTP 202 means processing has been accepted but has not completed; it does not promise a successful review. HTTP semantics

The array must contain 1 to 100 nonnull items. Item IDs must be nonblank, at most 128 characters, and unique within the batch. Every nested listing must pass validation. One invalid item rejects the entire submission with HTTP 400 before any work is stored. That is different from a model failure after acceptance, which fails only that item’s job.

Here is an actual acceptance response from the deterministic recovery test described below. The test uses the same short keyboard fixture under three item IDs to isolate scheduling behavior from listing content. Your IDs, timestamps, and model results will differ.

{
  "batchId": "6befecd8-016e-4a00-a9a6-4ce1038ed5d0",
  "status": "QUEUED",
  "submittedAt": "2026-09-08T12:29:45.708702Z",
  "total": 3,
  "statusUrl": "/api/listings/review-batches/6befecd8-016e-4a00-a9a6-4ce1038ed5d0",
  "itemsUrl": "/api/listings/review-batches/6befecd8-016e-4a00-a9a6-4ce1038ed5d0/items"
}

The controller validates the request and delegates acceptance to the application service:

@PostMapping(version = "1")
public ResponseEntity<SubmissionResponse> submit(
    @RequestHeader("Idempotency-Key") @NotBlank @Size(max = 128) String key,
    @RequestBody @Valid ReviewBatchRequest request) {
  var batch = service.submit(key, request);
  String location = location(batch.batchId());
  return ResponseEntity.status(batch.status() == ReviewBatchStatus.COMPLETED ? HttpStatus.OK : HttpStatus.ACCEPTED)
      .location(URI.create(location)).cacheControl(CacheControl.noStore())
      .body(new SubmissionResponse(batch.batchId(), batch.status(), batch.submittedAt(), batch.total(),
          location, location + "/items"));
}

The location helper builds the relative batch resource path. The completed-state branch handles an equivalent repeated submission for an already completed batch, including one with failed items. We will come back to that shortly.

Why I added a persistent scheduler

Spring’s @Async and TaskExecutor let us execute work asynchronously. They do not, by themselves, add a durable record of accepted jobs. Spring task execution

For this application, I chose db-scheduler. Its task records live in the database, and it manages execution claiming and heartbeat-based recovery. Its documentation also explains how a scheduling client can participate in a Spring transaction through a transaction-aware datasource. db-scheduler documentation

The more useful comparison is with other durable options. My selection criteria here are independent listing jobs, atomic acceptance in PostgreSQL, and execution inside the current Spring application:

Option	Where I would consider it
Quartz with JDBCJobStore or JobRunr	Credible in-process alternatives for persistent scheduling or background jobs. I would favor an existing team standard; this example uses db-scheduler’s explicit task IDs and transaction-aware client.
Spring Batch	A record-processing pipeline organized around steps and chunks. Here, each seller listing needs an independently retrievable result and recovery lifecycle.
Transactional outbox with Kafka or SQS	A broker-based design when workers need a separate deployment boundary. It adds a relay and consumer lifecycle that this example does not require.
Temporal	Durable orchestration for workflows with multiple coordinated steps. I would reconsider it as the workflow grows; one review operation per job is enough here.

This is a scope decision, not a claim that db-scheduler is universally better. We reuse its execution machinery and own the seller-facing acceptance, result, and failure contracts.

That last detail matters. Our application has its own job table because a seller-facing result has a different lifecycle from an internal scheduler task. Saving the application row and scheduling its work must be atomic.

Consider two independently committed writes. The application stores a review job, then crashes before enqueueing it. The seller has a job that no worker will execute. Reversing the order introduces the opposite problem: a worker can receive an ID whose application record does not exist yet.

For a batch, the transaction must cover the parent, every child, and every scheduler row. Calling the public single-job submission method repeatedly would not give us that guarantee. Here is the batch service’s submission method:

public ReviewBatchSummary submit(String key, ReviewBatchRequest request) {
  var ids = new HashSet<String>();
  for (var item : request.items()) {
    if (!ids.add(item.itemId())) throw new InvalidReviewBatchException();
  }
  var prepared = request.items().stream().map(item -> {
    String json = payload.encode(item.listing());
    return new BatchItem(item.itemId(), json, payload.fingerprint(json));
  }).toList();
  String fingerprint = ReviewJobPayload.sha256(mapper.writeValueAsString(prepared.stream()
      .map(item -> new ItemIdentity(item.itemId(), item.fingerprint())).toList()));
  try {
    return transactions.execute(transaction -> {
      var submission = batches.createOrFind(key, fingerprint);
      if (!submission.fingerprint().equals(fingerprint)) throw new ReviewJobConflictException();
      if (submission.created()) {
        var jobIds = jobs.createBatchItems(submission.batchId(), prepared);
        scheduler.scheduleBatch(jobIds);
      }
      return batches.find(submission.batchId()).orElseThrow(ReviewBatchNotFoundException::new);
    });
  } catch (DataAccessException | TransactionException exception) {
    throw new ReviewJobStorageException(exception);
  }
}

TransactionTemplate makes the boundary explicit. The controller receives its result after the transaction commits.

BatchItem holds an item ID, the preserved JSON snapshot, and its fingerprint. ItemIdentity contains only the item ID and fingerprint. Their ordered list is the batch identity. Preparation happens before opening the database transaction; no model call is part of acceptance.

The repository inserts child rows with JDBC batchUpdate, and the scheduler adapter uses db-scheduler’s scheduleBatch. Both batched writes participate in the same transaction, keeping atomic acceptance while avoiding individual insert and schedule calls for each listing.

The scheduling client uses the same underlying jobs datasource through TransactionAwareDataSourceProxy:

@Bean(name = "jobsSchedulerClient")
@DependsOn("jobsFlyway")
SchedulerClient client(@Qualifier("jobsDataSource") HikariDataSource ds) {
  return SchedulerClient.Builder.create(new TransactionAwareDataSourceProxy(ds)).build();
}

This is easy to get wrong if an application has several datasources. I explicitly qualify the jobs datasource and transaction template. The optional policy index has its own connection pool and does not participate in job acceptance.

The batch rollback test throws after all 100 scheduled tasks have been written inside the transaction. It checks that review_batches, review_jobs, and scheduled_tasks are all empty afterward. Another test accepts 100 items, replays the request without creating more work, and pages through position 100. The HTTP tests reject item 101 before storage. A separate test submits twelve concurrent equivalent requests and verifies one batch, three children, and three tasks using PostgreSQL, rather than relying on mocked transaction behavior.

Put the state invariants in PostgreSQL

The first migration defines the child job lifecycle. It is small enough to read in full:

create table review_jobs (
  id uuid primary key,
  idempotency_key varchar(128) not null unique,
  request_hash char(64) not null,
  payload jsonb not null,
  status varchar(16) not null check (status in ('QUEUED', 'RUNNING', 'SUCCEEDED', 'FAILED')),
  submitted_at timestamptz not null default current_timestamp,
  started_at timestamptz,
  completed_at timestamptz,
  attempts integer not null default 0 check (attempts >= 0),
  execution_token uuid,
  result jsonb,
  error_code varchar(64),
  error_message varchar(256),
  check ((status in ('SUCCEEDED', 'FAILED')) = (completed_at is not null)),
  check ((status = 'SUCCEEDED') = (result is not null)),
  check ((status = 'FAILED') = (error_code is not null)),
  check ((status = 'RUNNING') = (execution_token is not null))
);

The CHECK constraints make invalid combinations unrepresentable in a committed row. A terminal job must have completed_at; only success has a result; only failure has an error code; only a running job has an execution token. Java still controls the transitions, but the database provides another boundary against incorrect writes.

The token is not a seller credential. It identifies the currently authorized execution attempt. Recovery replaces it, allowing us to reject a late response from an older attempt.

Each child job follows this lifecycle:

QUEUED -- claim, issue token --> RUNNING
                                  |
                                  +-- result + matching token --> SUCCEEDED
                                  |
                                  +-- error + matching token ---> FAILED
                                  |
                                  +-- recovery claim ----------> RUNNING
                                      replace token; increment attempt

A late completion with an old token leaves the row unchanged. Claiming also checks the attempt budget: an unfinished job at the limit becomes FAILED with EXECUTION_ATTEMPTS_EXHAUSTED, without another model call. Terminal jobs stay terminal on redelivery.

The migration also creates db-scheduler’s scheduled_tasks table and indexes using its versioned PostgreSQL schema. I keep that infrastructure schema separate from the seller-facing job contract. You can find both in V1__review_jobs.sql.

Batch membership is a separate migration, V2__review_batches.sql. We do not rewrite an already applied Flyway migration:

create table review_batches (
  id uuid primary key,
  idempotency_key varchar(128) not null unique,
  request_hash char(64) not null,
  submitted_at timestamptz not null default current_timestamp
);

alter table review_jobs
  alter column idempotency_key drop not null,
  add column batch_id uuid references review_batches(id),
  add column item_id varchar(128),
  add column item_position integer,
  add constraint review_jobs_batch_item_unique unique (batch_id, item_id),
  add constraint review_jobs_batch_position_unique unique (batch_id, item_position),
  add constraint review_jobs_membership_check check (
    (idempotency_key is not null and batch_id is null and item_id is null and item_position is null)
    or
    (idempotency_key is null and batch_id is not null and item_id is not null
      and item_position is not null and item_position between 1 and 100)
  );

A job is either standalone with its own idempotency key, or a member with a batch ID, item ID, and position. The constraint disallows half-populated membership. The unique constraints also create indexes for batch membership and ordered page reads. We do not create artificial child idempotency strings in the caller’s standalone key namespace.

Notice what the parent does not contain: mutable status or progress counters. The child rows already record those facts. Updating a second set of counters would introduce another consistency problem during retries and recovery. The migration test also starts from V1 with an existing standalone job and verifies that V2 preserves it.

A worker that reuses the existing review service

Each scheduler task contains only the application job ID. The adapter builds and schedules the group together:

List<TaskInstance<?>> instances = jobIds.stream()
    .<TaskInstance<?>>map(id -> REVIEW_TASK.instance(id.toString()).build()).toList();
client.scheduleBatch(instances, Instant.now());

The worker loads the stored listing snapshot. It records an execution token and marks the job RUNNING in a short transaction, then calls ListingReviewService outside the transaction.

Here is the execution flow, including the failure and persistence boundaries:

public void execute(UUID id) {
  telemetry.observe(id, () -> executeAttempt(id));
}

private ReviewJobTelemetry.Outcome executeAttempt(UUID id) {
  var execution = repository.begin(id, maximumAttempts);
  if (execution.isEmpty()) {
    return ReviewJobTelemetry.Outcome.SKIPPED;
  }
  var attempt = execution.orElseThrow();
  telemetry.started(attempt.submittedAt(), attempt.startedAt());
  ListingReview result;
  try {
    var draft = payload.decode(attempt.payload());
    result = executor.call(() -> reviews.review(draft, ReviewEntrypoint.JOB));
  } catch (InterruptedException exception) {
    // Leave the execution recoverable when the worker is shutting down.
    throw new IllegalStateException("Review worker interrupted", exception);
  } catch (RuntimeException exception) {
    ReviewJobError error = classify(exception);
    log.warn("Background review failed: jobId={}, code={}", id, error.code());
    var completed = repository.fail(id, attempt.token(), error);
    completed.ifPresent(telemetry::completed);
    if (completed.isEmpty()) {
      return ReviewJobTelemetry.Outcome.STALE;
    }
    return switch (error.code()) {
      case "MODEL_TIMEOUT" -> ReviewJobTelemetry.Outcome.MODEL_TIMEOUT;
      case "MODEL_CAPACITY_UNAVAILABLE" -> ReviewJobTelemetry.Outcome.CAPACITY_UNAVAILABLE;
      default -> ReviewJobTelemetry.Outcome.FAILED;
    };
  }
  // Persistence failures escape to scheduler recovery; never turn them into model failures.
  var completed = repository.succeed(id, attempt.token(), result);
  completed.ifPresent(telemetry::completed);
  return completed.isPresent() ? ReviewJobTelemetry.Outcome.SUCCEEDED : ReviewJobTelemetry.Outcome.STALE;
}

The overload reviews.review(draft, ReviewEntrypoint.JOB) uses the same prompt, generator, and validator as the synchronous method. Its extra argument selects a bounded telemetry label, not a different model implementation.

The classify helper maps timeout, capacity, invalid-output, provider, and unexpected failures to our bounded public error codes. It does not store raw provider messages. Notice that a failed result write is outside the model-failure catch block: a database problem must remain recoverable rather than becoming a terminal model error.

A database transaction should not stay open while the application waits for inference. The tests check both the handler’s waiting thread and the model-call thread for an active transaction.

When storing the result, the update requires the current execution token:

update review_jobs set status = 'SUCCEEDED', result = ?::jsonb,
  completed_at = current_timestamp, execution_token = null
where id = ? and status = 'RUNNING' and execution_token = ?
returning *

After recovery, a new attempt gets a new token. If an older attempt eventually returns, its result cannot overwrite the newer attempt’s state. The repository returns the updated job, or an empty result when the fence rejects the write. It logs the rejected attempt, and the handler records a stale outcome. A stale response is discarded, not automatically sent back to the model.

Read progress from the database

Use the statusUrl returned by your own submission:

curl http://localhost:8080/api/listings/review-batches/6befecd8-016e-4a00-a9a6-4ce1038ed5d0

The UUID above belongs to the test run. Replace it with the one from your response.

The batch states are QUEUED, RUNNING, and COMPLETED. A batch is queued when every child is queued, completed when every child is terminal, and running otherwise. Completion does not mean every item succeeded. Polling is enough for this version: start with a one-second interval and gradually increase it to five seconds.

Here is the same batch after recovery and another application restart:

{
  "batchId": "6befecd8-016e-4a00-a9a6-4ce1038ed5d0",
  "submittedAt": "2026-09-08T12:29:45.708702Z",
  "completedAt": "2026-09-08T12:29:51.933979Z",
  "status": "COMPLETED",
  "total": 3,
  "queued": 0,
  "running": 0,
  "succeeded": 2,
  "failed": 1,
  "itemsUrl": "/api/listings/review-batches/6befecd8-016e-4a00-a9a6-4ce1038ed5d0/items"
}

The summary comes from one SQL statement:

select b.id, b.submitted_at, count(j.id) as total,
  count(*) filter (where j.status = 'QUEUED') as queued,
  count(*) filter (where j.status = 'RUNNING') as running,
  count(*) filter (where j.status = 'SUCCEEDED') as succeeded,
  count(*) filter (where j.status = 'FAILED') as failed,
  case when bool_and(j.status in ('SUCCEEDED', 'FAILED'))
    then max(j.completed_at) end as completed_at
from review_batches b join review_jobs j on j.batch_id = b.id
where b.id = ? group by b.id

All counts describe the same statement snapshot. We do not issue one query per counter and risk mixing observations from different moments. completedAt stays null until every child is terminal. There is no batch startedAt: a child’s start timestamp describes its current or latest attempt and changes during recovery.

Page through individual results

The summary is deliberately small. Fetch the items separately, using the URL from your own response:

curl 'http://localhost:8080/api/listings/review-batches/6befecd8-016e-4a00-a9a6-4ce1038ed5d0/items?after=0&limit=2'

The response contains batchId, items, and nextAfter. Each item contains its caller-provided ID, job ID, state, timestamps, validated result, and safe error. When nextAfter is nonnull, pass that value as after for the next page. In this test, the first page returned position 2; the second page returned the remaining item:

curl 'http://localhost:8080/api/listings/review-batches/6befecd8-016e-4a00-a9a6-4ce1038ed5d0/items?after=2&limit=2'

{
  "batchId": "6befecd8-016e-4a00-a9a6-4ce1038ed5d0",
  "items": [
    {
      "itemId": "headset-003",
      "jobId": "8709d0d4-20bd-4f0b-9a16-d311a7834cc6",
      "status": "FAILED",
      "submittedAt": "2026-09-08T12:29:45.708702Z",
      "startedAt": "2026-09-08T12:29:47.886649Z",
      "completedAt": "2026-09-08T12:29:47.946970Z",
      "result": null,
      "error": {
        "code": "AI_RESPONSE_INVALID",
        "message": "The model did not produce a valid review."
      }
    }
  ],
  "nextAfter": null
}

The failed item is intentional: the test generator returns an invalid score on one call. The successful siblings contain the validated ListingReview under result. This demonstrates persistence and failure isolation, not the quality or speed of any model.

The page query uses immutable submission positions, not changing status or completion timestamps:

select j.id, j.item_id, j.item_position, j.status, j.submitted_at, j.started_at,
  j.completed_at, j.result, j.error_code, j.error_message
from review_batches b
left join lateral (
  select id, item_id, item_position, status, submitted_at, started_at,
    completed_at, result, error_code, error_message
  from review_jobs where batch_id = b.id and item_position > ?
  order by item_position limit ?
) j on true
where b.id = ? order by j.item_position

The repository binds limit + 1, returns at most limit items, and uses the extra row only to decide whether another page exists. The left join distinguishes an unknown batch from an existing batch with no remaining items, without an extra existence query. The (batch_id, item_position) unique index supports this lookup.

after is bounded from 0 to 100, and limit from 1 to 100, defaulting to 25. Membership order is stable across pages; result states can still change between requests. This is not a frozen snapshot of the whole batch.

The query deliberately excludes the stored listing payload, request hash, and execution token. A PostgreSQL test reads a page through a role that cannot select those columns. Poll the small summary for progress; fetch result pages when you need the actual reviews rather than repeatedly downloading the whole result set.

Retrieving a completed batch with failures returns HTTP 200: the lookup succeeded, and individual errors explain what failed. Unknown batches return 404, including on the items endpoint. Unavailable storage returns 503 with a bounded ProblemDetail response. Batch input validation and duplicate item IDs both return INVALID_REVIEW_BATCH_REQUEST. All responses use Cache-Control: no-store. The existing single-job URL still works for a returned child job ID. Neither job errors nor request errors expose provider messages, prompts, or stack traces.

What if the seller submits twice?

The Idempotency-Key identifies one batch submission. We keep a unique database constraint on it and a fingerprint of the ordered item identities. Standalone job keys occupy a separate namespace, so reusing the same key string on the two endpoints does not create a conflict.

Repeated request	Response
Same key, equivalent ordered items, batch active	Same batch ID, HTTP 202
Same key, equivalent ordered items, batch completed	Same batch ID, HTTP 200
Same key, different payload	HTTP 409
New key	New batch and child jobs

Normalization sorts attribute keys and removes insignificant decimal scale. A price of 45.00 and 45 therefore identifies the same listing input. We preserve meaningful seller text and item IDs. Reordering items is a different request because the accepted order determines pagination. Replaying a completed batch does not resubmit failed items.

But the snapshot and the fingerprint have different jobs. The snapshot preserves the accepted typed input for execution. The fingerprint answers whether a repeated request is equivalent. Here is the codec:

public class ReviewJobPayload {
  private final ObjectMapper mapper;

  public ReviewJobPayload(ObjectMapper mapper) {
    this.mapper = mapper;
  }

  public String encode(ListingDraft draft) {
    return mapper.writeValueAsString(draft);
  }

  public ListingDraft decode(String json) {
    return mapper.readValue(json, ListingDraft.class);
  }

  public String fingerprint(String json) {
    ListingDraft draft = decode(json);
    String canonical = mapper.writeValueAsString(new ListingDraft(draft.title(), draft.description(),
        draft.category(), draft.price().stripTrailingZeros(),
        draft.attributes() == null ? null : new TreeMap<>(draft.attributes())));
    return sha256(canonical);
  }

  static String sha256(String canonical) {
    try {
      return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256")
          .digest(canonical.getBytes(StandardCharsets.UTF_8)));
    } catch (NoSuchAlgorithmException exception) {
      throw new IllegalStateException("SHA-256 is required by the Java runtime", exception);
    }
  }
}

An earlier implementation reused the normalized representation for both purposes. It changed 100.00 to 100 by the time the stored JSONB reached the worker’s prompt. The monetary value had not changed, but the submitted representation had. Separating the two responsibilities fixes that without weakening idempotency.

The regression tests cover 45.00, 100.00, 1200.00, and 19.99 through submission, real PostgreSQL storage, execution, and prompt rendering. PostgreSQL can reorder JSON object keys; the snapshot contract is the typed values, not a byte-for-byte copy of the original HTTP body. An equivalent repeated request keeps the first accepted snapshot.

This decision is enforced by the database, including concurrent submissions. A check in a Java map or a separate lookup followed by an unconditional insert would not provide that protection across application instances.

Keys remain reserved while the corresponding batch or standalone job is retained. This increment keeps work until explicit cleanup, so an intentional new submission needs a new key.

The job stores its listing input, but execution uses the currently deployed review implementation, prompt, and provider configuration. A deployment that changes those settings can change the result of queued or recovered work. If a workflow needs reproducible processing across releases, version those execution inputs explicitly.

Bound execution and be precise about retries

The default configuration has two scheduler workers and two model-call slots per application instance, shared by all batches and standalone jobs. The model executor has no waiting queue. This prevents an accumulation of background model calls in JVM memory. A batch of 100 listings does not create 100 local model threads, but it can still produce 100 review operations and their associated provider charges.

The execution settings are bound and validated through a configuration-properties record:

@Validated
@ConfigurationProperties("listing-quality.review-jobs")
public record ListingReviewJobProperties(
    @Valid @DefaultValue Database datasource,
    @Min(1) @Max(32) @DefaultValue("2") int workers,
    @Min(1) @Max(10) @DefaultValue("3") int maximumAttempts,
    @DefaultValue("120s") Duration modelTimeout,
    @DefaultValue("2s") Duration pollingInterval,
    @DefaultValue("5s") Duration heartbeatInterval,
    @Min(4) @DefaultValue("6") int missedHeartbeats,
    @DefaultValue("5s") Duration shutdownWait) {
  public ListingReviewJobProperties {
    for (Duration duration : new Duration[]{modelTimeout, pollingInterval, heartbeatInterval, shutdownWait}) {
      if (duration == null || duration.toMillis() < 1) {
        throw new IllegalArgumentException("Job durations must be at least one millisecond");
      }
    }
  }

  public record Database(
      @NotBlank @DefaultValue("jdbc:postgresql://localhost:5434/listing_quality_jobs") String url,
      @NotBlank @DefaultValue("listing_quality_jobs") String username,
      @NotBlank @DefaultValue("listing_quality_jobs") String password,
      @Min(2) @DefaultValue("8") int maximumPoolSize) {}
}

The scheduler bean uses those settings directly:

@Bean(name = "jobsScheduler", initMethod = "start", destroyMethod = "stop")
@DependsOn("jobsFlyway")
Scheduler scheduler(@Qualifier("jobsDataSource") HikariDataSource ds, OneTimeTask<Void> listingReviewTask,
    ListingReviewJobProperties properties) {
  return Scheduler.create(ds, listingReviewTask).threads(properties.workers())
      .pollingInterval(properties.pollingInterval()).heartbeatInterval(properties.heartbeatInterval())
      .missedHeartbeatsLimit(properties.missedHeartbeats()).shutdownMaxWait(properties.shutdownWait())
      .build();
}

The handler waits up to 120 seconds for its model result. On timeout, it attempts cancellation and stores a safe MODEL_TIMEOUT outcome. Cancellation is cooperative: an SDK can ignore interruption, and a provider can continue processing and billing. Such a call keeps its local slot until it ends. If all slots remain occupied, another job can fail with MODEL_CAPACITY_UNAVAILABLE.

The executor bounds local model work and how long the handler waits. Configure connection and read/request timeouts in the provider’s HTTP transport as well, using the settings supported by that SDK. Choose those limits together with SDK retries so their budget fits inside the handler deadline. The executor is a backstop, not a substitute for transport timeouts, and closing a connection still cannot guarantee that remote processing or billing stops. For example, the JDK HTTP client exposes a per-request timeout through HttpRequest.Builder); other transports have their own configuration.

The three-argument constructor receives the application’s observation registry; the two-argument overload delegates with a no-op registry for standalone use.

public ReviewJobModelExecutor(int concurrency, Duration timeout, ObservationRegistry registry) {
  this.timeout = timeout;
  this.registry = registry;
  this.executor = new ThreadPoolExecutor(concurrency, concurrency, 0, TimeUnit.SECONDS,
      new SynchronousQueue<>(), Thread.ofPlatform().daemon().name("review-model-", 0).factory(),
      new ThreadPoolExecutor.AbortPolicy());
}

public ListingReview call(Callable<ListingReview> operation) throws InterruptedException {
  Future<ListingReview> result;
  try {
    Observation parent = registry.getCurrentObservation();
    result = executor.submit(() -> Observation.tryScopedChecked(parent, operation::call));
  } catch (RejectedExecutionException exception) {
    throw new ReviewJobExecutionException("MODEL_CAPACITY_UNAVAILABLE", "Model execution capacity is unavailable.");
  }
  try {
    return result.get(timeout.toMillis(), TimeUnit.MILLISECONDS);
  } catch (TimeoutException exception) {
    result.cancel(true);
    throw new ReviewJobExecutionException("MODEL_TIMEOUT", "The review exceeded its time limit.");
  } catch (InterruptedException exception) {
    result.cancel(true);
    Thread.currentThread().interrupt();
    throw exception;
  } catch (ExecutionException exception) {
    if (exception.getCause() instanceof RuntimeException cause) {
      throw cause;
    }
    if (exception.getCause() instanceof Error cause) {
      throw cause;
    }
    throw new ReviewJobExecutionException("REVIEW_FAILED", "The review could not be completed.");
  }
}

@Override
public void close() {
  executor.shutdownNow();
}

SynchronousQueue has no storage capacity. Submission must hand work to an available thread or fail. AbortPolicy makes that failure explicit, and the adapter translates it into MODEL_CAPACITY_UNAVAILABLE.

Scheduler workers and model slots have the same configured count. Normal queued work waits in PostgreSQL; capacity rejection is particularly useful when timed-out calls keep occupying model slots. Rejection can also occur during executor shutdown or thread handoff, so it is a safe local failure signal rather than a precise measure of backlog.

The captured observation is scoped on the executor thread while the callable runs, then restored. It is not restarted or stopped there. This keeps the existing feature observation underneath the job attempt without making the executor responsible for the attempt’s lifecycle.

These limits apply per instance. Adding replicas increases concurrency. A persistent queue can also keep growing even with a small worker pool, so public deployment needs backlog admission limits and capacity planning.

Known model failures are terminal in this increment, after any retries inside the provider or structured-output validation layer. That includes a transient provider outage that reaches the handler: the item becomes FAILED. Replaying the original idempotency key retrieves that submission; it does not retry failed work. To try again today, submit only the failed listings as a new batch under a new key, or use the individual-job endpoint with a new key. Successful siblings stay untouched. The next article will introduce a deliberate retry policy for these failures.

Storage failures are different. If writing a result fails, the worker lets the exception reach the scheduler so the job remains recoverable. Storage-related deliveries use a ten-second retry delay. There are at most three recorded handler attempts that can call the model; a subsequent delivery marks an unfinished job as exhausted. When the database itself is unreachable, persistence cannot record attempts, so storage retries continue until it can.

What background execution does to your dashboard

In the previous series, we added metrics and traces. Reusing the review service also reuses its instrumentation, but synchronous requests and background work should not silently share a latency objective.

Listing reviews now add entrypoint=sync or entrypoint=job. The feature recording rules preserve that dimension. The existing example alerts exclude background execution:

listing_quality:feature_latency_p95_seconds_5m{entrypoint!="job"} > 10

The inequality also retains existing synchronous enrichment series that do not yet have that label. route=direct still describes model routing. It is not overloaded to mean synchronous execution.

The listing and policy features share one meter name, so their tag-key sets must also agree for Prometheus registration. Policy reviews now include entrypoint=sync; listing reviews include marketplace=none and locale=none. Those values mean the dimensions do not apply, not that we inferred marketplace or language from the listing.

A separate job observation surrounds each handler delivery:

public void observe(UUID id, Supplier<Outcome> operation) {
  Observation observation = Observation.createNotStarted("listing.quality.review.job.attempt", observations)
      .contextualName("background listing review")
      .highCardinalityKeyValue("job.id", id.toString())
      .lowCardinalityKeyValue("outcome", "recoverable_failure")
      .start();
  try (var ignored = observation.openScope()) {
    Outcome outcome = operation.get();
    observation.lowCardinalityKeyValue("outcome", outcome.name().toLowerCase(Locale.ROOT));
  } finally {
    // Do not attach exception messages or seller data to exported job observations.
    observation.stop();
  }
}

The job ID is a high-cardinality trace attribute, not a metric label. Each recovered attempt gets a new observation and can be found using the same job ID. We do not persist the original HTTP trace context in this increment, and we do not keep its span open while a job waits in PostgreSQL.

The runnable Grafana dashboard now separates the measurements:

Measurement	What it tells us
Feature execution, split by entrypoint	Time inside the shared review operation
Job attempt duration and outcome	Time spent in a handler delivery, including its persistence boundary
Job age at attempt start	Submission-to-start duration; on recovery, includes earlier execution and recovery delay
Worker-completed job turnaround	Submission-to-completion duration for a committed worker success or model failure

The histogram ranges cover longer background work: 180 seconds for feature and attempt duration, and one hour for age and turnaround. The runbook documents the sampling boundaries, including which completion paths contribute to turnaround.

A timed-out attempt stays model_timeout even if its model call later succeeds. That call’s feature span may also finish after the parent attempt span. Read these together: the handler stopped waiting, while model work continued. PostgreSQL remains the authoritative source for the persisted job outcome; a process crash can lose telemetry after a commit.

Start the existing dashboard stack and run the application with trace export enabled:

docker compose -f observability/compose.yaml up -d
./mvnw -pl listing-quality-service spring-boot:run -Dspring-boot.run.profiles=gemini,observability

If the application is already running, restart it with the additional profile rather than starting another instance on the same port. Prometheus is available on localhost port 9090 and Grafana on port 3000. Submit jobs, allow several scrapes, and use the new review-job panels. The application selector applies to those panels; the feature selector applies to the shared feature panels.

The tests run real Prometheus rule evaluation with healthy synchronous traffic and slow, failing background traffic. Neither synchronous alert may fire because of the background traffic. I would choose a separate background SLO from observed turnaround and queue behavior, not copy the HTTP latency threshold.

Run those rule tests from the repository root:

docker run --rm --entrypoint /bin/promtool \
  -v "$PWD/observability/prometheus:/work:ro" -w /work/tests \
  prom/prometheus:v3.12.0 test rules review-jobs.test.yml

Test the restart, including the uncomfortable window

Run the offline tests and the Docker integration suite:

./mvnw -o test
./mvnw -o -pl listing-quality-service -Preview-jobs-integration-tests verify

On a fresh checkout, omit -o initially so Maven can download dependencies. Docker must be running for the integration suite.

The process-recovery test starts the actual application in a child JVM. It replaces only the external model generator, then performs this sequence:

Submit one batch containing three items.
Let one item complete, block the next model call, and leave the remaining item queued.
Verify the persisted summary contains one succeeded, one running, and one queued item.
Forcibly terminate the JVM.
Start a replacement against the same PostgreSQL database.
Make one remaining model call return invalid output; verify two successes and one failure overall.
Restart again and retrieve the summary and both pages, including their stored result/error contents.

The completed item remained at one handler attempt. The interrupted item recorded two; the queued item recorded one. This proves completed work was not regenerated during recovery. The test uses accelerated heartbeat settings; normal recovery follows the configured heartbeat threshold and polling, and RUNNING can remain visible during that detection interval.

Other tests cover V1-to-V2 migration, membership constraints, transaction rollback after partial scheduling, concurrent duplicate submissions, canonical batch identity, stable pagination, all-success/all-failure progress, shared worker concurrency, exhausted attempts, stale result writes, price preservation through the worker prompt, cross-thread observation parentage, and safe errors. The batch recovery test writes its HTTP evidence to listing-quality-service/target/review-batches-process-evidence.json. The existing standalone-job crash test remains in the suite.

The JSON examples in this article come from a retained September 8, 2026 test run. Your run generates new IDs and timestamps; compare the states, pagination, results, and errors rather than those run-specific values.

There is still a window we cannot remove with a database transaction: the provider completes a paid request, but the application crashes before storing the response. Recovery can call the provider again. We also simulate a result-persistence failure and verify that two inference calls can lead to one stored result.

If the result was already committed, redelivery sees a terminal job and skips inference. That protects completed application work, but it is not an exactly-once guarantee for external model calls.

Before exposing this outside your laptop

The sample endpoints run in a trusted local environment and do not implement authentication. In a marketplace, batch, page, and child-job lookups all need ownership checks, and idempotency keys should be scoped to the authenticated account. Neither UUIDs nor caller item IDs are authorization.

Job snapshots and results also need an explicit retention policy that preserves batch membership. The 100-item request bound is not a bound on retained backlog or total cost. The complete deployment needs HTTP body-size limits, account admission limits, and provider quota controls based on its expected workload. Batching changes how work is accepted and tracked; it does not make inference cheaper or guarantee fair scheduling between sellers.

Where we go next

The seller can now submit a group of listings, leave the page, and return to its progress and individual results. The application has a persisted record of what it accepted, and each listing is an independent recovery unit. The synchronous API is still there when one interactive review is the better fit.

In the next article, we will make the failure policy more deliberate: which operations to retry, how long to keep trying, and how to avoid repeating completed steps as the workflow grows.