Ship S3-compatible external payload storage

.env.example

@@ -56,6 +56,15 @@ DW_EXTERNAL_PAYLOAD_MAX_OBJECTS_PER_NAMESPACE=
 DW_EXTERNAL_PAYLOAD_HARD_MAX_BYTES_PER_NAMESPACE=
 DW_EXTERNAL_PAYLOAD_HARD_MAX_OBJECTS_PER_NAMESPACE=
 DW_EXTERNAL_PAYLOAD_NAMESPACE_OVERRIDES={}
+# Shared object storage for namespaces whose external payload policy uses s3.
+# Access key and secret may be omitted when the runtime has an IAM role.
+DW_EXTERNAL_PAYLOAD_S3_ACCESS_KEY_ID=
+DW_EXTERNAL_PAYLOAD_S3_SECRET_ACCESS_KEY=
+DW_EXTERNAL_PAYLOAD_S3_SESSION_TOKEN=
+DW_EXTERNAL_PAYLOAD_S3_REGION=us-east-1
+DW_EXTERNAL_PAYLOAD_S3_BUCKET=
+DW_EXTERNAL_PAYLOAD_S3_ENDPOINT=
+DW_EXTERNAL_PAYLOAD_S3_USE_PATH_STYLE_ENDPOINT=false
 DW_DEFAULT_NAMESPACE=default
 #
 # Cluster discovery advertises the current node's deployment shape and process

.github/workflows/phpunit-feature.yml

@@ -165,15 +165,42 @@ jobs:
               && apt-get install -y --no-install-recommends git \
               && rm -rf /var/lib/apt/lists/*
           DOCKERFILE
+          docker network create \
+            "dw-server-feature-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}"
+          docker run --detach \
+            --name "dw-server-minio-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
+            --network "dw-server-feature-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
+            --env MINIO_ROOT_USER=dw-minio-access \
+            --env MINIO_ROOT_PASSWORD=dw-minio-secret-key \
+            minio/minio@sha256:14cea493d9a34af32f524e538b8346cf79f3321eff8e708c1e2960462bd8936e \
+            server /data --address :9000
           tar \
             --exclude=vendor \
             --exclude='*/vendor' \
             --exclude=build \
             --exclude='*/build' \
             -cf - . \
-            | docker run --rm -i -e CORPUS_BASE_REF -w /app --entrypoint sh durable-workflow-server-corpus-validator \
+            | docker run --rm -i \
+                --network "dw-server-feature-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
+                -e CORPUS_BASE_REF \
+                -e DW_TEST_S3_ENDPOINT="http://dw-server-minio-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}:9000" \
+                -e DW_TEST_S3_ACCESS_KEY_ID=dw-minio-access \
+                -e DW_TEST_S3_SECRET_ACCESS_KEY=dw-minio-secret-key \
+                -w /app \
+                --entrypoint sh \
+                durable-workflow-server-corpus-validator \
                 -lc 'set -eu; tar --no-same-owner -xf - -C /app; cp -a /app/workflow-package /workflow; php scripts/ci/prepare-release-workflow-composer-metadata.php; composer update durable-workflow/workflow --with-dependencies --minimal-changes --no-install --no-scripts --no-autoloader --no-interaction --no-progress --prefer-dist; composer install --no-interaction --no-progress --prefer-dist; git --version; vendor/bin/phpunit --version; git rev-parse --verify "$CORPUS_BASE_REF^{commit}"; php scripts/ci/check-worker-openapi-evolution.php "$CORPUS_BASE_REF"; vendor/bin/phpunit tests/Feature tests/Unit/NexusContractTest.php tests/Unit/CodecRegressionCorpusTest.php tests/Unit/PhpunitFeatureWorkflowContractTest.php tests/Unit/OpenApiDocumentEvolutionTest.php --no-progress --colors=never; python3 scripts/ci/validate-regression-corpus.py --base-ref "$CORPUS_BASE_REF" --verify-counterfactual'
 
+      - name: Clean up S3-compatible test resources
+        if: always()
+        run: |
+          docker rm -f \
+            "dw-server-minio-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
+            >/dev/null 2>&1 || true
+          docker network rm \
+            "dw-server-feature-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}" \
+            >/dev/null 2>&1 || true
+
   qualification:
     name: Feature source qualification
     needs: [preflight, feature]

README.md

@@ -75,6 +75,9 @@ contract, so a workflow in one language can dispatch activities to another.
 Server supports SQLite for a single-node runtime and MySQL or PostgreSQL for
 shared durable state. Multi-node deployments use shared Redis for queue and
 coordination state. The database remains authoritative for workflow history.
+Namespaces using external payloads in a multi-node deployment should use the
+built-in [S3-compatible shared storage](docs/contracts/external-payload-storage.md#self-hosted-backing-storage)
+instead of node-local files.
 
 ## Capabilities
 

app/Http/Controllers/Api/HealthController.php

@@ -12,11 +12,11 @@
 use App\Support\ControlPlaneProtocol;
 use App\Support\ControlPlaneRequestContract;
 use App\Support\CoordinationHealthContract;
-use App\Support\FilesystemDiskAvailability;
 use App\Support\HeartbeatRuntimeContract;
 use App\Support\LegacyV1ProjectionContract;
 use App\Support\MessageStreamsContract;
 use App\Support\MigrationRuntimeContract;
+use App\Support\NamespaceExternalPayloadStorage;
 use App\Support\NamespaceRuntimeContract;
 use App\Support\NexusContract;
 use App\Support\PayloadCodecContract;
@@ -63,6 +63,7 @@ class HealthController
     public function __construct(
         private readonly ServerReadiness $readiness,
         private readonly TaskQueueBuildIdRolloutSnapshot $buildIdRollouts,
+        private readonly NamespaceExternalPayloadStorage $externalPayloadStorage,
     ) {}
 
     public function check(): JsonResponse
@@ -343,15 +344,16 @@ private function externalPayloadStoragePolicy(?WorkflowNamespace $ns): array
         $driver = $this->stringOrNull($policy['driver'] ?? null);
         $enabled = $policy !== [] && ($policy['enabled'] ?? true) !== false;
         $threshold = $policy['threshold_bytes'] ?? config('server.limits.max_payload_bytes', 2 * 1024 * 1024);
-        $config = is_array($policy['config'] ?? null) ? $policy['config'] : [];
-        $resolvedDriver = $enabled && $this->externalPayloadStorageResolvable($driver, $config);
+        $configurationError = $this->externalPayloadStorage->configurationErrorForPolicy($policy);
+        $resolvedDriver = $enabled && $this->externalPayloadStorage->policyResolvable($policy);
 
         return [
             'schema' => RuntimeExternalPayloadReference::SCHEMA,
             'version' => 1,
             'configured' => $policy !== [],
             'enabled' => $enabled,
             'status' => $this->externalPayloadStorageStatus($policy, $enabled, $resolvedDriver),
+            'configuration_error' => $enabled && ! $resolvedDriver ? $configurationError : null,
             'threshold_bytes' => (int) $threshold,
             'transport' => RuntimeExternalPayloadReference::transportManifest(),
             'provider_details_exposed' => false,
@@ -379,33 +381,6 @@ private function externalPayloadStorageStatus(array $policy, bool $enabled, bool
         return $resolved ? 'available' : 'driver_unavailable';
     }
 
-    /**
-     * @param  array<string, mixed>  $config
-     */
-    private function externalPayloadStorageResolvable(?string $driver, array $config): bool
-    {
-        if ($driver === 'local') {
-            return true;
-        }
-
-        if (! in_array($driver, ['s3', 'gcs', 'azure', 'custom'], true)) {
-            return false;
-        }
-
-        $disk = $this->stringOrNull($config['disk'] ?? null);
-        $bucket = $this->stringOrNull(
-            $config['bucket']
-            ?? $config['container']
-            ?? $config['name']
-            ?? null,
-        );
-        $scheme = $driver === 'custom'
-            ? $this->stringOrNull($config['scheme'] ?? null)
-            : $driver;
-
-        return FilesystemDiskAvailability::configured($disk) && $bucket !== null && $scheme !== null;
-    }
-
     private function stringOrNull(mixed $value): ?string
     {
         return is_string($value) && $value !== '' ? $value : null;

app/Http/Controllers/Api/StorageController.php

@@ -59,7 +59,10 @@ public function test(Request $request): JsonResponse
                 'The server can persist this storage policy, but the configured storage driver is not available in this runtime.',
                 $namespace,
                 $driver,
-                ['supported_diagnostic_drivers' => ['local', 's3', 'gcs', 'azure', 'custom']],
+                [
+                    'configuration_error' => $this->externalPayloadStorage->configurationErrorFor($namespace),
+                    'supported_diagnostic_drivers' => ['local', 's3', 'gcs', 'azure', 'custom'],
+                ],
             );
         }
 

app/Support/FilesystemDiskAvailability.php

@@ -2,18 +2,96 @@
 
 namespace App\Support;
 
+use League\Flysystem\AwsS3V3\AwsS3V3Adapter;
+
 class FilesystemDiskAvailability
 {
     public static function configured(mixed $disk): bool
+    {
+        return self::configurationError($disk) === null;
+    }
+
+    public static function configurationError(mixed $disk): ?string
+    {
+        if (! is_string($disk) || $disk === '') {
+            return 'filesystem_disk_name_missing';
+        }
+
+        $config = self::configuration($disk);
+        if ($config === null) {
+            return 'filesystem_disk_not_configured';
+        }
+
+        if (($config['driver'] ?? null) !== 's3') {
+            return null;
+        }
+
+        if (! class_exists(AwsS3V3Adapter::class)) {
+            return 's3_adapter_unavailable';
+        }
+
+        if (self::stringOrNull($config['bucket'] ?? null) === null) {
+            return 's3_bucket_missing';
+        }
+
+        if (self::stringOrNull($config['region'] ?? null) === null) {
+            return 's3_region_missing';
+        }
+
+        $key = self::stringOrNull($config['key'] ?? null);
+        $secret = self::stringOrNull($config['secret'] ?? null);
+        $token = self::stringOrNull($config['token'] ?? null);
+
+        if (($key === null) !== ($secret === null) || ($token !== null && ($key === null || $secret === null))) {
+            return 's3_credentials_incomplete';
+        }
+
+        return null;
+    }
+
+    public static function bucket(mixed $disk): ?string
+    {
+        if (! is_string($disk) || $disk === '') {
+            return null;
+        }
+
+        return self::stringOrNull(self::configuration($disk)['bucket'] ?? null);
+    }
+
+    public static function driver(mixed $disk): ?string
     {
         if (! is_string($disk) || $disk === '') {
-            return false;
+            return null;
         }
 
+        return self::stringOrNull(self::configuration($disk)['driver'] ?? null);
+    }
+
+    /**
+     * @return array<string, mixed>|null
+     */
+    private static function configuration(string $disk): ?array
+    {
         $configuredDisks = config('filesystems.disks');
 
-        return is_array($configuredDisks)
-            && array_key_exists($disk, $configuredDisks)
-            && is_array($configuredDisks[$disk]);
+        if (! is_array($configuredDisks)
+            || ! array_key_exists($disk, $configuredDisks)
+            || ! is_array($configuredDisks[$disk])
+        ) {
+            return null;
+        }
+
+        return $configuredDisks[$disk];
+    }
+
+    private static function stringOrNull(mixed $value): ?string
+    {
+        if (! is_string($value)) {
+            return null;
+        }
+
+        $value = trim($value);
+
+        return $value === '' ? null : $value;
     }
 }

app/Support/NamespaceExternalPayloadStorage.php

@@ -34,33 +34,61 @@ public function untrackedDriverFor(?string $namespace): ?RuntimeExternalPayloadS
         }
 
         if (in_array($driver, ['s3', 'gcs', 'azure', 'custom'], true)) {
-            $disk = $policy['config']['disk'] ?? null;
-            $bucket = $policy['config']['bucket']
-                ?? $policy['config']['container']
-                ?? $policy['config']['name']
-                ?? null;
-            $scheme = $driver === 'custom'
-                ? ($policy['config']['scheme'] ?? null)
-                : $driver;
-
-            if (! FilesystemDiskAvailability::configured($disk)
-                || ! is_string($bucket) || $bucket === ''
-                || ! is_string($scheme) || $scheme === ''
-            ) {
+            $filesystem = $this->filesystemPolicy($policy, $driver);
+
+            if ($filesystem['error'] !== null) {
                 return null;
             }
 
             return $this->guard(new FilesystemExternalPayloadStorage(
-                disk: $disk,
-                scheme: $scheme,
-                bucket: $bucket,
+                disk: $filesystem['disk'],
+                scheme: $filesystem['scheme'],
+                bucket: $filesystem['bucket'],
                 prefix: $this->prefix($policy),
             ));
         }
 
         return null;
     }
 
+    public function configurationErrorFor(?string $namespace): ?string
+    {
+        $namespace = $namespace ?: (string) config('server.default_namespace', 'default');
+
+        return $this->configurationErrorForPolicy($this->policyFor($namespace));
+    }
+
+    /**
+     * @param  array<string, mixed>  $policy
+     */
+    public function configurationErrorForPolicy(array $policy): ?string
+    {
+        if ($policy === [] || ($policy['enabled'] ?? true) === false) {
+            return null;
+        }
+
+        $driver = $policy['driver'] ?? null;
+        if ($driver === 'local') {
+            return null;
+        }
+
+        if (! is_string($driver) || ! in_array($driver, ['s3', 'gcs', 'azure', 'custom'], true)) {
+            return 'external_payload_storage_driver_unsupported';
+        }
+
+        return $this->filesystemPolicy($policy, $driver)['error'];
+    }
+
+    /**
+     * @param  array<string, mixed>  $policy
+     */
+    public function policyResolvable(array $policy): bool
+    {
+        return $policy !== []
+            && ($policy['enabled'] ?? true) !== false
+            && $this->configurationErrorForPolicy($policy) === null;
+    }
+
     public function thresholdBytesFor(?string $namespace): ?int
     {
         $namespace = $namespace ?: (string) config('server.default_namespace', 'default');
@@ -119,6 +147,49 @@ private function prefix(array $policy): string
         return trim($prefix, '/').'/';
     }
 
+    /**
+     * @param  array<string, mixed>  $policy
+     * @return array{disk: string, bucket: string, scheme: string, error: ?string}
+     */
+    private function filesystemPolicy(array $policy, string $driver): array
+    {
+        $config = is_array($policy['config'] ?? null) ? $policy['config'] : [];
+        $disk = $config['disk'] ?? null;
+
+        if ((! is_string($disk) || $disk === '') && $driver === 's3') {
+            $disk = (string) config('server.external_payload_transport.s3_disk', 'external-payload-s3');
+        }
+
+        $diskError = FilesystemDiskAvailability::configurationError($disk);
+        $bucket = $config['bucket']
+            ?? $config['container']
+            ?? $config['name']
+            ?? ($driver === 's3' ? FilesystemDiskAvailability::bucket($disk) : null);
+        $scheme = $driver === 'custom' ? ($config['scheme'] ?? null) : $driver;
+
+        $error = $diskError;
+        if ($error === null && (! is_string($bucket) || $bucket === '')) {
+            $error = 'external_payload_storage_bucket_missing';
+        }
+        if ($error === null && (! is_string($scheme) || $scheme === '')) {
+            $error = 'external_payload_storage_scheme_missing';
+        }
+        if ($error === null
+            && $driver === 's3'
+            && FilesystemDiskAvailability::driver($disk) === 's3'
+            && ! hash_equals((string) FilesystemDiskAvailability::bucket($disk), (string) $bucket)
+        ) {
+            $error = 's3_bucket_mismatch';
+        }
+
+        return [
+            'disk' => is_string($disk) ? $disk : '',
+            'bucket' => is_string($bucket) ? $bucket : '',
+            'scheme' => is_string($scheme) ? $scheme : '',
+            'error' => $error,
+        ];
+    }
+
     private function guard(RuntimeExternalPayloadStorageDriver $driver): RuntimeExternalPayloadStorageDriver
     {
         return new GuardedExternalPayloadStorage($driver);