feat: move the durable job substrate to PostgreSQL (#6809)

docs/internal/contributor/adr/README.md

@@ -85,6 +85,7 @@ This folder contains Architecture Decision Records (ADRs) for the Honua greenfie
 | [0078](0078-3d-scene-architecture.md) | 3D Scene Architecture — Canonical 3D Tiles, I3S Projection, and Runtime Resolution | Accepted | 2026-08 |
 | [0079](0079-protocol-route-scoping.md) | A protocol's route is scoped by its own addressable unit | Accepted | 2026-09 |
 | [0080](0080-fair-core-license.md) | Adopt the Fair Core License (FCL-1.0-ALv2) | Accepted | 2026-10 |
+| [0081](0081-postgres-job-substrate.md) | PostgreSQL is the durable job substrate | Proposed | 2026-11 |
 
 ## Template
 

eng/schema/schema-migration-hashes.v1.json

@@ -169,5 +169,6 @@
   "src/Honua.Server/Migrations/143_PromoteImportTypedColumns.sql": "C2C1C3BFE78F8A0D03C1A8186BB8CE07ECE88CA7B81204B91B3470D823E9EE22",
   "src/Honua.Server/Migrations/144_RenameGeoServicesRestSourceKind.sql": "89EEBC9062DF7381E6854B7B212EB539699845C0DFC43F4C4CA911E9EF01FAA9",
   "src/Honua.Server/Migrations/145_ProtectImportDependencies.sql": "5137ECDF8A2C23C7503B048BA79EA21CF504B89DED7E0E5A64417724D7DEAEB1",
-  "src/Honua.Server/Migrations/146_AddMigrationBatchPublicationTarget.sql": "0F09CE3C38D7C2FBF21A9497B6F575EC04C7B84ED5C74CFE510EDCF02D24AA7A"
+  "src/Honua.Server/Migrations/146_AddMigrationBatchPublicationTarget.sql": "0F09CE3C38D7C2FBF21A9497B6F575EC04C7B84ED5C74CFE510EDCF02D24AA7A",
+  "src/Honua.Server/Migrations/147_CreateJobSubstrate.sql": "9908DADFB95268F12A2FF16127DB1B469DBFB4D58EC915A237CCEA6F8E444A12"
 }

src/Honua.Core/Core/Features/Capabilities/DurableJobSubstrateOptions.cs

@@ -73,43 +73,28 @@ public sealed class DurableJobSubstrateOptions
     public string? RedisDurabilityStatus
         => RedisDurabilityStatuses.For(RedisDurabilityAttestation is not null, RedisDurabilityFailure);
 
+    // The job store and queue live in the system database, so a host that composed both runs jobs
+    // whatever its Redis state. Without them the Redis facts still classify the Redis-backed
+    // control plane (proposals, workflow operations), which callers ask about with neither present.
     public DurableJobSubstrateCause Classify(bool jobStorePresent, bool jobQueuePresent)
     {
-        // Entitlement is checked BEFORE durability, and fails closed (honua-server#4502). The
-        // durability inspection runs against whatever infrastructure Redis is connected, which in
-        // non-Development/Test deployments happens even when Redis is unentitled — so an
-        // unentitled deployment whose Redis DOES attest would otherwise report Available and
-        // advertise 'jobs.runner'. A present store never upgrades a missing entitlement.
-        if (!RedisEntitled && RedisConfigured)
-        {
-            return DurableJobSubstrateCause.RedisNotEntitled;
-        }
-
-        // Owner ruling (2026-10-10): a CONNECTED Redis enables the durable job runner and the
-        // governed proposal control plane whatever the durability attestation said. The
-        // attestation outcome is published (RedisDurabilityStatus) and logged, never classified
-        // here; only a Redis that is absent or did not connect withholds the capability.
         if (jobStorePresent && jobQueuePresent)
         {
-            if (!RedisConfigured)
-            {
-                return DurableJobSubstrateCause.RedisNotConfigured;
-            }
-
-            return RedisDurabilityFailure == DurableJobSubstrateCause.RedisUnreachable
-                ? DurableJobSubstrateCause.RedisUnreachable
-                : DurableJobSubstrateCause.Available;
+            return DurableJobSubstrateCause.Available;
         }
 
         if (!RedisConfigured)
         {
             return DurableJobSubstrateCause.RedisNotConfigured;
         }
 
-        return RedisEntitled
-            ? RedisDurabilityFailure == DurableJobSubstrateCause.RedisUnreachable
-                ? DurableJobSubstrateCause.RedisUnreachable
-                : DurableJobSubstrateCause.RuntimeIncomplete
-            : DurableJobSubstrateCause.RedisNotEntitled;
+        if (!RedisEntitled)
+        {
+            return DurableJobSubstrateCause.RedisNotEntitled;
+        }
+
+        return RedisDurabilityFailure == DurableJobSubstrateCause.RedisUnreachable
+            ? DurableJobSubstrateCause.RedisUnreachable
+            : DurableJobSubstrateCause.RuntimeIncomplete;
     }
 }

src/Honua.Core/Core/Features/Capabilities/DurableJobSubstrateStartupGate.cs

@@ -3,12 +3,6 @@
 
 namespace Honua.Core.Features.Capabilities;
 
-// Entitlement decides whether the durable job substrate is composed; durability attestation only decides what the server advertises.
-public sealed class DurableJobSubstrateEntitlement
-{
-    public const string CapabilityId = CapabilityUnavailableCodes.DurableJobsCapability;
-}
-
 public sealed class JobDurabilityOptions
 {
     public const string SectionName = "Jobs";

src/Honua.Core/Core/Features/ControlPlane/Domain/OperationModels.cs

@@ -1181,7 +1181,7 @@ public sealed record PreMigrationRestoreContext
 /// <summary>
 /// Batch-compute execution specification.
 /// </summary>
-public sealed record ExecutionJobSpec
+public sealed record ExecutionJobSpec : Honua.Core.Serialization.ExtensibleJsonRecord
 {
     /// <summary>
     /// Server-owned GPServer binding, stamped after scoped execution authorization.
@@ -1297,7 +1297,7 @@ public sealed record ExecutionJobDefinition
 /// <summary>
 /// Durable execution-job record for run-to-completion workloads.
 /// </summary>
-public sealed record ExecutionJobRecord
+public sealed record ExecutionJobRecord : Honua.Core.Serialization.ExtensibleJsonRecord
 {
     /// <summary>
     /// Stable execution job identifier.

src/Honua.Geoprocessing/Core/Features/Geoprocessing/GeoprocessingServiceCollectionExtensions.cs

@@ -16,7 +16,6 @@
 using Microsoft.Extensions.Configuration;
 using Microsoft.Extensions.DependencyInjection.Extensions;
 using Microsoft.Extensions.Options;
-using StackExchange.Redis;
 
 namespace Honua.Geoprocessing;
 
@@ -83,41 +82,9 @@ public static IServiceCollection AddGeoprocessing(
         // process-local store the dispatcher records into and an admin view ranks.
         services.TryAddSingleton<IProcessUsageTelemetry, InMemoryProcessUsageTelemetry>();
 
-        // Execution job store (ticket #722).
-        //
-        // REGISTRATION CONTRACT (honua-server#4502): whenever the durable job substrate is
-        // ENTITLED and a Redis multiplexer is composed, IExecutionJobStore IS resolvable.
-        // Durability attestation decides what the server ADVERTISES, never whether the store
-        // exists. #4141 gated this block on the accepted RedisDurabilityAttestation while every
-        // consumer of IExecutionJobStore stayed registered alongside the multiplexer (the
-        // reconcilers, the backstop sweep, the event handler, the queue-depth collector, the
-        // metadata-release reconciler — the `connectedRedis != null` block in Program.cs), so a
-        // stock non-AOF Redis aborted ServiceProvider descriptor validation with 31
-        // unresolved-service failures and exit 139 before the server bound a port.
-        //
-        // A rejected attestation now degrades instead: the same Redis-backed store is composed,
-        // Program.cs logs one warning naming the typed cause and its consequence, the capability
-        // manifest reports that cause rather than advertising 'jobs.runner'
-        // (DurableJobSubstrateOptions.Classify), and only an explicit Jobs:RequireDurableStore=true
-        // may turn the rejection into a typed startup refusal.
-        //
-        // The entitlement marker is required SEPARATELY from the attestation and is not
-        // interchangeable with it. Infrastructure Redis is connected in non-Development/Test
-        // deployments even when unentitled (requiresDurableDistributedEvents forces it), so
-        // keying this on the multiplexer alone would compose — and, for a Redis that does attest,
-        // ADVERTISE — durable jobs for a deployment with no jobs entitlement.
-        if (services.Any(d => d.ServiceType == typeof(IConnectionMultiplexer))
-            && services.Any(d => d.ServiceType == typeof(DurableJobSubstrateEntitlement)))
+        if (services.TryAddPostgresExecutionJobStore(configuration))
         {
-            services.TryAddSingleton<IExecutionJobStore>(sp =>
-                new RedisExecutionJobStore(
-                    sp.GetRequiredService<IConnectionMultiplexer>(),
-                    sp.GetRequiredService<ILogger<RedisExecutionJobStore>>()));
-            services.TryAddSingleton<IGeoprocessingResultPackageStore>(sp =>
-                new RedisGeoprocessingResultPackageStore(
-                    sp.GetRequiredService<IConnectionMultiplexer>(),
-                    sp.GetRequiredService<IOptionsMonitor<GeoprocessingExecutorOptions>>(),
-                    sp.GetRequiredService<ILogger<RedisGeoprocessingResultPackageStore>>()));
+            services.TryAddSingleton<IGeoprocessingResultPackageStore, PostgresGeoprocessingResultPackageStore>();
         }
 
         // Execution admission controls (ticket #739) — rate, concurrency, cost, backpressure
@@ -168,9 +135,6 @@ public static IServiceCollection AddGeoprocessing(
         // tick dispatcher.
         services.AddGeoprocessingOutputStaging(configuration);
         services.TryAddSingleton<GeoprocessingRasterOutputRegistrar>();
-        // RedisExecutionJobStore is registered above before this check whenever
-        // IConnectionMultiplexer is present. AddJobOrchestration later composes the
-        // queue/log services around that existing store; it does not add the store.
         if (services.Any(d => d.ServiceType == typeof(IGeoprocessingOutputObjectStore))
             && services.Any(d => d.ServiceType == typeof(IExecutionJobStore)))
         {

src/Honua.Geoprocessing/Core/Features/Geoprocessing/PostgresGeoprocessingResultPackageStore.cs

@@ -0,0 +1,84 @@
+// Copyright (c) Honua Geo Inc. All rights reserved.
+// Licensed under the Fair Core License, Version 1.0, ALv2 Future License. See LICENSE in the project root.
+
+using System.Text.Json;
+using Honua.Core.Features.Geoprocessing.Domain;
+using Honua.Geoprocessing;
+using Microsoft.Extensions.Options;
+using Npgsql;
+using NpgsqlTypes;
+
+namespace Honua.ControlPlane;
+
+/// <summary>
+/// Geoprocessing result packages in PostgreSQL (ADR-0081). A package past its retention is not
+/// returned, whether or not the retention sweep has deleted it yet.
+/// </summary>
+internal sealed partial class PostgresGeoprocessingResultPackageStore(
+    JobStoreDatabase database,
+    IOptionsMonitor<GeoprocessingExecutorOptions> executorOptions,
+    ILogger<PostgresGeoprocessingResultPackageStore> logger) : IGeoprocessingResultPackageStore
+{
+    private readonly string _packages = database.Table("geoprocessing_result_packages");
+
+    public async Task<AnalysisResultPackage?> GetAsync(
+        string jobId,
+        CancellationToken cancellationToken = default)
+    {
+        ArgumentException.ThrowIfNullOrWhiteSpace(jobId);
+
+        return await database.ExecuteAsync(
+            async (connection, _, token) =>
+            {
+                await using var command = new NpgsqlCommand(
+                    $"SELECT package FROM {_packages} WHERE job_id = @job_id AND expires_at > now()",
+                    connection);
+                command.Parameters.Add(new NpgsqlParameter("job_id", NpgsqlDbType.Text) { Value = jobId });
+                return await command.ExecuteScalarAsync(token).ConfigureAwait(false) is string package
+                    ? JsonSerializer.Deserialize(package, ControlPlaneJsonContext.Default.AnalysisResultPackage)
+                    : null;
+            },
+            StatementRepeat.SafeToRepeat,
+            cancellationToken).ConfigureAwait(false);
+    }
+
+    public async Task SetAsync(
+        string jobId,
+        AnalysisResultPackage package,
+        TimeSpan? ttl = null,
+        CancellationToken cancellationToken = default)
+    {
+        ArgumentException.ThrowIfNullOrWhiteSpace(jobId);
+        ArgumentNullException.ThrowIfNull(package);
+
+        var payload = JsonSerializer.Serialize(package, ControlPlaneJsonContext.Default.AnalysisResultPackage);
+        await database.ExecuteAsync(
+            async (connection, token) =>
+            {
+                await using var command = new NpgsqlCommand(
+                    $"""
+                    INSERT INTO {_packages} (job_id, package, expires_at)
+                    VALUES (@job_id, @package, now() + @retention)
+                    ON CONFLICT (job_id) DO UPDATE SET package = EXCLUDED.package, expires_at = EXCLUDED.expires_at
+                    """,
+                    connection);
+                command.Parameters.Add(new NpgsqlParameter("job_id", NpgsqlDbType.Text) { Value = jobId });
+                command.Parameters.Add(new NpgsqlParameter("package", NpgsqlDbType.Json) { Value = payload });
+                command.Parameters.Add(new NpgsqlParameter("retention", NpgsqlDbType.Interval)
+                {
+                    Value = ttl ?? executorOptions.CurrentValue.ResultRetention
+                });
+                await command.ExecuteNonQueryAsync(token).ConfigureAwait(false);
+            },
+            StatementRepeat.SafeToRepeat,
+            cancellationToken).ConfigureAwait(false);
+
+        Log.ResultPackageStored(logger, jobId, package.ResultPackageId);
+    }
+
+    private static partial class Log
+    {
+        [LoggerMessage(9040, LogLevel.Debug, "Stored geoprocessing result package for job {JobId}: {ResultPackageId}")]
+        public static partial void ResultPackageStored(ILogger logger, string jobId, string resultPackageId);
+    }
+}

src/Honua.Jobs/Features/ControlPlane/ExecutionJobCursor.cs

@@ -0,0 +1,69 @@
+// Copyright (c) Honua Geo Inc. All rights reserved.
+// Licensed under the Fair Core License, Version 1.0, ALv2 Future License. See LICENSE in the project root.
+
+using System.Globalization;
+using System.Text;
+
+namespace Honua.ControlPlane;
+
+/// <summary>
+/// Position of the last job returned by a newest-first page: creation time at the database's
+/// microsecond precision, then operation id.
+/// </summary>
+internal readonly record struct ExecutionJobCursor(DateTimeOffset CreatedAt, string OperationId)
+{
+    private const char Separator = '|';
+    private const long TicksPerMicrosecond = TimeSpan.TicksPerMillisecond / 1000;
+
+    public static DateTimeOffset TruncateToMicroseconds(DateTimeOffset value)
+    {
+        var utc = value.ToUniversalTime();
+        return new DateTimeOffset(utc.Ticks - (utc.Ticks % TicksPerMicrosecond), TimeSpan.Zero);
+    }
+
+    public string Encode()
+    {
+        var microseconds = (TruncateToMicroseconds(CreatedAt).Ticks - DateTimeOffset.UnixEpoch.Ticks) / TicksPerMicrosecond;
+        var value = string.Concat(microseconds.ToString(CultureInfo.InvariantCulture), Separator.ToString(), OperationId);
+        return Convert.ToBase64String(Encoding.UTF8.GetBytes(value));
+    }
+
+    public static bool TryDecode(string cursor, out ExecutionJobCursor decoded)
+    {
+        decoded = default;
+        string text;
+        try
+        {
+            text = Encoding.UTF8.GetString(Convert.FromBase64String(cursor));
+        }
+        catch (FormatException)
+        {
+            return false;
+        }
+
+        var separator = text.IndexOf(Separator, StringComparison.Ordinal);
+        if (separator <= 0 || separator == text.Length - 1
+            || !long.TryParse(text.AsSpan(0, separator), NumberStyles.Integer, CultureInfo.InvariantCulture, out var microseconds))
+        {
+            return false;
+        }
+
+        long ticks;
+        try
+        {
+            ticks = checked(DateTimeOffset.UnixEpoch.Ticks + (microseconds * TicksPerMicrosecond));
+        }
+        catch (OverflowException)
+        {
+            return false;
+        }
+
+        if (ticks < DateTimeOffset.MinValue.Ticks || ticks > DateTimeOffset.MaxValue.Ticks)
+        {
+            return false;
+        }
+
+        decoded = new ExecutionJobCursor(new DateTimeOffset(ticks, TimeSpan.Zero), text[(separator + 1)..]);
+        return true;
+    }
+}