Merge pull request #131 from alucero270/feature/91-index-job-retries

feat(indexer): add retry policy and retry accounting for failed ingestion jobs

Author: alucero270

docs/api-surface-audit.md

@@ -23,7 +23,7 @@ explicit.
 | `POST /api/search` | `SearchController` | search indexed documents | intended product-facing endpoint | Core retrieval API for the current Strata slice. |
 | `GET /api/documents/{id}` | `DocumentsController` | fetch a known document | intended product-facing endpoint | Supports document viewing after search or direct lookup. |
 | `POST /api/index-jobs` | `IndexJobsController` | create indexing work | intended product-facing endpoint | Current request shape is intentionally minimal for the early product slice. |
-| `GET /api/index-jobs/{id}` | `IndexJobsController` | read indexing job status | intended product-facing endpoint | Supports current operational verification of indexing flow. |
+| `GET /api/index-jobs/{id}` | `IndexJobsController` | read indexing job status | intended product-facing endpoint | Supports current operational verification of indexing flow, including retry accounting. |
 | `GET /openapi/v1.json` | `MapOpenApi()` in `Program.cs` | development-time API description | development-only support endpoint | Present only in Development; useful for inspection, not part of the stable product contract. |
 
 ## Non-Product Or Limited-Scope Surface

docs/api.md

@@ -87,6 +87,13 @@ Create a new indexing job.
 
 Returns `201 Created` with the created job payload.
 
+### Behavior
+
+- retry policy is server-controlled for the current product slice
+- each job currently allows up to `3` total processing attempts
+- jobs return retry accounting metadata so operators can tell whether a
+  processing failure was the first attempt or a later retry
+
 ## `GET /api/index-jobs/{id}`
 
 Fetch the current status of an indexing job.
@@ -100,12 +107,21 @@ Fetch the current status of an indexing job.
   "requestedAt": "2026-04-10T08:20:00Z",
   "claimedAt": "2026-04-10T08:20:02Z",
   "completedAt": "2026-04-10T08:20:05Z",
+  "attemptCount": 1,
+  "maxAttempts": 3,
   "workerId": "host:1234:abcd",
   "errorMessage": null,
   "stats": null
 }
 ```
 
+### Retry Semantics
+
+- `attemptCount` increments each time a worker claims the job for processing
+- a failed attempt returns the job to `pending` while `attemptCount` is still
+  below `maxAttempts`
+- a job becomes terminally `failed` only when the final allowed attempt fails
+
 ## Notes
 
 - Use `/health` for basic readiness checks and product verification

docs/data-model.md

@@ -37,13 +37,17 @@ Stores indexing work items claimed by background workers.
 - `requested_at`: job creation timestamp
 - `claimed_at`: timestamp recorded when a worker claims the job
 - `completed_at`: timestamp recorded on successful completion
+- `attempt_count`: number of processing attempts claimed so far
+- `max_attempts`: server-controlled retry ceiling for the job
 - `worker_id`: worker identifier for claimed jobs
 - `error_message`: truncated failure detail when processing fails
 
 ### Notes
 
 - `ix_index_jobs_status_requested_at_id` supports pending-job polling
 - Job creation is API-driven; claiming and completion are background operations
+- failed attempts return to `pending` while `attempt_count < max_attempts`
+- terminal `failed` state means the last allowed attempt has already been used
 
 ## Configured Source Model
 

docs/operations.md

@@ -95,6 +95,9 @@ Get-Content -Raw ops/migrations/002_documents.sql |
 Get-Content -Raw ops/migrations/003_jobs.sql |
   docker compose -f ops/docker-compose.yml --env-file .env exec -T postgres `
   psql -v ON_ERROR_STOP=1 -U strata -d strata
+Get-Content -Raw ops/migrations/004_job_retries.sql |
+  docker compose -f ops/docker-compose.yml --env-file .env exec -T postgres `
+  psql -v ON_ERROR_STOP=1 -U strata -d strata
 ```
 
 If you change the PostgreSQL credentials in `.env`, update the `psql` arguments
@@ -188,6 +191,16 @@ Invoke-WebRequest -Uri http://localhost:8080/api/index-jobs -Method Post `
   -ContentType "application/json" -Body "{}"
 ```
 
+Verify index-job retry state:
+
+- poll `GET /api/index-jobs/{id}` after creating the job
+- expect `attemptCount` to increment each time the worker claims the job
+- if an attempt fails and retries remain, expect the job to return to `pending`
+  instead of remaining terminal immediately
+- expect a terminal `failed` job to report `attemptCount == maxAttempts`
+- the current retry policy is server-controlled and allows up to `3` total
+  attempts per job
+
 Platform readiness validation:
 
 ```powershell

ops/migrations/004_job_retries.sql

@@ -0,0 +1,46 @@
+-- Persist retry accounting so ingestion jobs can retry predictably.
+ALTER TABLE index_jobs
+    ADD COLUMN IF NOT EXISTS attempt_count INTEGER NOT NULL DEFAULT 0,
+    ADD COLUMN IF NOT EXISTS max_attempts INTEGER NOT NULL DEFAULT 3;
+
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1
+        FROM pg_constraint
+        WHERE conname = 'ck_index_jobs_attempt_count_non_negative'
+    ) THEN
+        ALTER TABLE index_jobs
+            ADD CONSTRAINT ck_index_jobs_attempt_count_non_negative
+                CHECK (attempt_count >= 0);
+    END IF;
+END
+$$;
+
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1
+        FROM pg_constraint
+        WHERE conname = 'ck_index_jobs_max_attempts_positive'
+    ) THEN
+        ALTER TABLE index_jobs
+            ADD CONSTRAINT ck_index_jobs_max_attempts_positive
+                CHECK (max_attempts >= 1);
+    END IF;
+END
+$$;
+
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1
+        FROM pg_constraint
+        WHERE conname = 'ck_index_jobs_attempt_count_within_limit'
+    ) THEN
+        ALTER TABLE index_jobs
+            ADD CONSTRAINT ck_index_jobs_attempt_count_within_limit
+                CHECK (attempt_count <= max_attempts);
+    END IF;
+END
+$$;

src/Codex.Api/Data/IndexJobsStore.cs

@@ -6,20 +6,24 @@ namespace Codex.Api.Data;
 
 public sealed class IndexJobsStore(NpgsqlDataSource dataSource, CodexSettings settings)
 {
+    private const int DefaultMaxAttempts = 3;
+
     // Current schema has no root_path column. Reference docs_root in the insert path
     // so job creation still depends on server-side configuration.
     private const string InsertJobSql = """
         WITH configured_root AS (
             SELECT @docs_root::text AS docs_root
         )
-        INSERT INTO index_jobs (status)
-        SELECT @status
+        INSERT INTO index_jobs (status, max_attempts)
+        SELECT @status, @max_attempts
         FROM configured_root
-        RETURNING id, status, requested_at, claimed_at, completed_at, worker_id, error_message;
+        RETURNING id, status, requested_at, claimed_at, completed_at, attempt_count,
+            max_attempts, worker_id, error_message;
         """;
 
     private const string SelectJobByIdSql = """
-        SELECT id, status, requested_at, claimed_at, completed_at, worker_id, error_message
+        SELECT id, status, requested_at, claimed_at, completed_at, attempt_count,
+            max_attempts, worker_id, error_message
         FROM index_jobs
         WHERE id = @id;
         """;
@@ -29,6 +33,8 @@ public async Task<IndexJobResponse> CreatePendingJobAsync(CancellationToken canc
         await using var command = dataSource.CreateCommand(InsertJobSql);
         command.Parameters.AddWithValue("docs_root", settings.DocsRoot);
         command.Parameters.AddWithValue("status", "pending");
+        // Retry policy remains server-controlled for the current product slice.
+        command.Parameters.AddWithValue("max_attempts", DefaultMaxAttempts);
 
         await using var reader = await command.ExecuteReaderAsync(cancellationToken);
         if (!await reader.ReadAsync(cancellationToken))
@@ -60,6 +66,8 @@ private static IndexJobResponse Map(NpgsqlDataReader reader)
     {
         var claimedAtOrdinal = reader.GetOrdinal("claimed_at");
         var completedAtOrdinal = reader.GetOrdinal("completed_at");
+        var attemptCountOrdinal = reader.GetOrdinal("attempt_count");
+        var maxAttemptsOrdinal = reader.GetOrdinal("max_attempts");
         var workerIdOrdinal = reader.GetOrdinal("worker_id");
         var errorMessageOrdinal = reader.GetOrdinal("error_message");
 
@@ -73,6 +81,8 @@ private static IndexJobResponse Map(NpgsqlDataReader reader)
             CompletedAt: reader.IsDBNull(completedAtOrdinal)
                 ? null
                 : reader.GetDateTime(completedAtOrdinal),
+            AttemptCount: reader.GetInt32(attemptCountOrdinal),
+            MaxAttempts: reader.GetInt32(maxAttemptsOrdinal),
             WorkerId: reader.IsDBNull(workerIdOrdinal)
                 ? null
                 : reader.GetString(workerIdOrdinal),

src/Codex.Contracts/IndexJobs/IndexJobResponse.cs

@@ -6,6 +6,8 @@ public sealed record IndexJobResponse(
     DateTime RequestedAt,
     DateTime? ClaimedAt,
     DateTime? CompletedAt,
+    int AttemptCount,
+    int MaxAttempts,
     string? WorkerId,
     string? ErrorMessage,
     object? Stats);

src/Codex.Indexer/Data/IndexJobsStore.cs

@@ -2,7 +2,16 @@
 
 namespace Codex.Indexer.Data;
 
-public sealed record ClaimedIndexJob(long Id);
+public sealed record ClaimedIndexJob(long Id, int AttemptCount, int MaxAttempts);
+
+public sealed record JobFailureDisposition(
+    long Id,
+    string Status,
+    int AttemptCount,
+    int MaxAttempts)
+{
+    public bool WillRetry => string.Equals(Status, "pending", StringComparison.Ordinal);
+}
 
 public sealed class IndexJobsStore(NpgsqlDataSource dataSource)
 {
@@ -19,11 +28,13 @@ LIMIT 1
         UPDATE index_jobs AS jobs
         SET status = 'processing',
             claimed_at = NOW(),
+            completed_at = NULL,
             worker_id = @worker_id,
-            error_message = NULL
+            error_message = NULL,
+            attempt_count = jobs.attempt_count + 1
         FROM next_job
         WHERE jobs.id = next_job.id
-        RETURNING jobs.id;
+        RETURNING jobs.id, jobs.attempt_count, jobs.max_attempts;
         """;
 
     private const string MarkJobCompletedSql = """
@@ -34,12 +45,27 @@ UPDATE index_jobs
         WHERE id = @id;
         """;
 
-    private const string MarkJobFailedSql = """
+    private const string RecordJobFailureSql = """
         UPDATE index_jobs
-        SET status = 'failed',
-            completed_at = NOW(),
+        SET status = CASE
+                WHEN attempt_count < max_attempts THEN 'pending'
+                ELSE 'failed'
+            END,
+            claimed_at = CASE
+                WHEN attempt_count < max_attempts THEN NULL
+                ELSE claimed_at
+            END,
+            completed_at = CASE
+                WHEN attempt_count < max_attempts THEN NULL
+                ELSE NOW()
+            END,
+            worker_id = CASE
+                WHEN attempt_count < max_attempts THEN NULL
+                ELSE worker_id
+            END,
             error_message = @error_message
-        WHERE id = @id;
+        WHERE id = @id
+        RETURNING id, status, attempt_count, max_attempts;
         """;
 
     public async Task<ClaimedIndexJob?> ClaimNextPendingJobAsync(
@@ -57,7 +83,10 @@ UPDATE index_jobs
         await using (var reader = await command.ExecuteReaderAsync(cancellationToken))
         {
             claimedJob = await reader.ReadAsync(cancellationToken)
-                ? new ClaimedIndexJob(reader.GetInt64(0))
+                ? new ClaimedIndexJob(
+                    reader.GetInt64(0),
+                    reader.GetInt32(1),
+                    reader.GetInt32(2))
                 : null;
         }
 
@@ -72,14 +101,25 @@ public async Task MarkJobCompletedAsync(long id, CancellationToken cancellationT
         await command.ExecuteNonQueryAsync(cancellationToken);
     }
 
-    public async Task MarkJobFailedAsync(
+    public async Task<JobFailureDisposition> RecordJobFailureAsync(
         long id,
         string errorMessage,
         CancellationToken cancellationToken)
     {
-        await using var command = dataSource.CreateCommand(MarkJobFailedSql);
+        await using var command = dataSource.CreateCommand(RecordJobFailureSql);
         command.Parameters.AddWithValue("id", id);
         command.Parameters.AddWithValue("error_message", errorMessage);
-        await command.ExecuteNonQueryAsync(cancellationToken);
+
+        await using var reader = await command.ExecuteReaderAsync(cancellationToken);
+        if (!await reader.ReadAsync(cancellationToken))
+        {
+            throw new InvalidOperationException($"Failed to record failure for job {id}.");
+        }
+
+        return new JobFailureDisposition(
+            reader.GetInt64(0),
+            reader.GetString(1),
+            reader.GetInt32(2),
+            reader.GetInt32(3));
     }
 }

src/Codex.Indexer/Worker.cs

@@ -56,13 +56,21 @@ private async Task PollOnceAsync(CancellationToken cancellationToken)
         }
 
         // Claiming commits in the store before this point, so processing is lock-free.
-        logger.LogInformation("Claimed index job {JobId}", claimedJob.Id);
+        logger.LogInformation(
+            "Claimed index job {JobId} (attempt {AttemptCount}/{MaxAttempts})",
+            claimedJob.Id,
+            claimedJob.AttemptCount,
+            claimedJob.MaxAttempts);
 
         try
         {
             await ProcessClaimedJobAsync(claimedJob, documentsStore, cancellationToken);
             await indexJobsStore.MarkJobCompletedAsync(claimedJob.Id, cancellationToken);
-            logger.LogInformation("Completed index job {JobId}", claimedJob.Id);
+            logger.LogInformation(
+                "Completed index job {JobId} on attempt {AttemptCount}/{MaxAttempts}",
+                claimedJob.Id,
+                claimedJob.AttemptCount,
+                claimedJob.MaxAttempts);
         }
         catch (Exception ex)
         {
@@ -73,8 +81,31 @@ private async Task PollOnceAsync(CancellationToken cancellationToken)
                 errorMessage = errorMessage[..1000];
             }
 
-            await indexJobsStore.MarkJobFailedAsync(claimedJob.Id, errorMessage, cancellationToken);
-            logger.LogError(ex, "Failed index job {JobId}", claimedJob.Id);
+            var failureDisposition =
+                await indexJobsStore.RecordJobFailureAsync(
+                    claimedJob.Id,
+                    errorMessage,
+                    cancellationToken);
+
+            if (failureDisposition.WillRetry)
+            {
+                logger.LogWarning(
+                    ex,
+                    "Index job {JobId} failed on attempt {AttemptCount}/{MaxAttempts}; " +
+                    "returned to pending for retry.",
+                    claimedJob.Id,
+                    failureDisposition.AttemptCount,
+                    failureDisposition.MaxAttempts);
+            }
+            else
+            {
+                logger.LogError(
+                    ex,
+                    "Index job {JobId} failed on final attempt {AttemptCount}/{MaxAttempts}",
+                    claimedJob.Id,
+                    failureDisposition.AttemptCount,
+                    failureDisposition.MaxAttempts);
+            }
         }
     }