Add guarded runtime activity lease renewal

.env.example

@@ -4,7 +4,7 @@ APP_DEBUG=true
 APP_URL=http://localhost:8080
 # APP_VERSION defaults to the checked-in source release identity.
 
-LOG_CHANNEL=stack
+LOG_CHANNEL=stderr
 LOG_LEVEL=debug
 
 DB_CONNECTION=mysql

.github/workflows/published-release-recovery.yml

@@ -66,6 +66,30 @@ jobs:
           TLS_TEST_ROOT: ${{ runner.temp }}/dw-server-tls-release-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
         run: bash scripts/ci/qualify-database-tls.sh
 
+      - name: Verify existing image first-run readiness and container logging
+        env:
+          DW_SERVER_IMAGE: ghcr.io/durable-workflow/server@${{ steps.release_verify.outputs.image_digest }}
+          DW_FIRST_RUN_SCOPE: recovery-first-run-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
+          RELEASE_SOURCE_COMMIT: ${{ inputs.release_commit }}
+        run: |
+          if git rev-parse --verify "${RELEASE_SOURCE_COMMIT}:scripts/ci/verify-container-logging.mjs" >/dev/null 2>&1; then
+            scripts/smoke-published-first-run.sh
+          else
+            echo "This immutable release predates container logging acceptance; historical verification remains unchanged."
+          fi
+
+      - name: Clean up existing image first-run resources
+        if: always()
+        env:
+          DW_FIRST_RUN_SCOPE: recovery-first-run-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
+          LOGGING_TEST_PROJECT: recovery-first-run-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}-logs
+        run: |
+          docker ps -aq --filter "label=com.docker.compose.project=$LOGGING_TEST_PROJECT" | xargs -r docker rm -f >/dev/null 2>&1 || true
+          docker network ls -q --filter "label=com.docker.compose.project=$LOGGING_TEST_PROJECT" | xargs -r docker network rm >/dev/null 2>&1 || true
+          docker volume ls -q --filter "label=com.docker.compose.project=$LOGGING_TEST_PROJECT" | xargs -r docker volume rm -f >/dev/null 2>&1 || true
+          docker rm -f "${DW_FIRST_RUN_SCOPE}-server" "${DW_FIRST_RUN_SCOPE}-bootstrap" "${DW_FIRST_RUN_SCOPE}-unbootstrapped" >/dev/null 2>&1 || true
+          docker volume rm "${DW_FIRST_RUN_SCOPE}-database" >/dev/null 2>&1 || true
+
       - name: Clean up published image database TLS fixtures
         if: always()
         env:
@@ -93,6 +117,7 @@ jobs:
     needs: verify
     if: >-
       always() &&
+      needs.verify.result == 'success' &&
       needs.verify.outputs.release_verification_outcome == 'success' &&
       needs.verify.outputs.published_database_tls_outcome == 'success' &&
       github.repository == 'durable-workflow/server' &&

.github/workflows/release.yml

@@ -296,9 +296,14 @@ jobs:
         if: always()
         env:
           DW_FIRST_RUN_SCOPE: release-first-run-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}
+          LOGGING_TEST_PROJECT: release-first-run-${{ github.run_id }}-${{ github.run_attempt }}-${{ github.job }}-logs
         run: |
+          docker ps -aq --filter "label=com.docker.compose.project=$LOGGING_TEST_PROJECT" | xargs -r docker rm -f >/dev/null 2>&1 || true
+          docker network ls -q --filter "label=com.docker.compose.project=$LOGGING_TEST_PROJECT" | xargs -r docker network rm >/dev/null 2>&1 || true
+          docker volume ls -q --filter "label=com.docker.compose.project=$LOGGING_TEST_PROJECT" | xargs -r docker volume rm -f >/dev/null 2>&1 || true
           docker rm -f \
             "${DW_FIRST_RUN_SCOPE}-server" \
+            "${DW_FIRST_RUN_SCOPE}-unbootstrapped" \
             "${DW_FIRST_RUN_SCOPE}-bootstrap" >/dev/null 2>&1 || true
           docker volume rm -f \
             "${DW_FIRST_RUN_SCOPE}-database" >/dev/null 2>&1 || true

Dockerfile

@@ -164,7 +164,8 @@ ENV DW_WORKER_LONG_POLL_MAX_CONCURRENT=2 \
     DB_CONNECTION=sqlite \
     DB_DATABASE=/app/database/database.sqlite \
     QUEUE_CONNECTION=database \
-    CACHE_STORE=file
+    CACHE_STORE=file \
+    LOG_CHANNEL=stderr
 
 ENTRYPOINT ["server-entrypoint"]
 

app/Http/Controllers/Api/ActivityTaskController.php

@@ -32,6 +32,7 @@
 use Illuminate\Validation\Rule;
 use Illuminate\Validation\ValidationException;
 use InvalidArgumentException;
+use Workflow\V2\Contracts\ActivityLeaseRenewal;
 use Workflow\V2\Contracts\ActivityTaskBridge as ActivityTaskBridgeContract;
 use Workflow\V2\Enums\HistoryEventType;
 use Workflow\V2\Exceptions\ExternalPayloadIntegrityException;
@@ -219,6 +220,7 @@ public function poll(Request $request): JsonResponse
                 'external_executor' => $externalExecutor,
                 ...WorkflowObservabilityPoll::forActivityClaim((string) $namespace, $claim),
             ], static fn (mixed $v): bool => $v !== null)),
+            'server_time' => now()->toJSON(),
             'poll_status' => is_string($poll['poll_status'] ?? null)
                 ? $poll['poll_status']
                 : ($claim === null ? 'empty' : 'leased'),
@@ -494,12 +496,10 @@ public function status(Request $request, string $taskId): JsonResponse
                 $status['can_continue'] = false;
                 $status['reason'] = 'lease_expired';
             }
-            foreach (['heartbeat', 'start_to_close', 'schedule_to_close'] as $kind) {
-                if ($status['can_continue'] === true && isset($deadlines[$kind])
-                    && ! CarbonImmutable::parse($deadlines[$kind])->isAfter(now())) {
-                    $status['can_continue'] = false;
-                    $status['reason'] = $kind.'_timeout';
-                }
+            $expiredReason = $this->expiredDeadlineReason($deadlines);
+            if ($status['can_continue'] === true && $expiredReason !== null) {
+                $status['can_continue'] = false;
+                $status['reason'] = $expiredReason;
             }
             if ($status['can_continue'] === true && $workerSession !== null) {
                 $sessionLease = $workerSession['lease_expires_at'] ?? null;
@@ -694,6 +694,103 @@ private function activityCancellationReceipt(
         ];
     }
 
+    public function renewLease(Request $request, string $taskId): JsonResponse
+    {
+        if ($response = WorkerProtocol::rejectUnsupported($request)) {
+            return $response;
+        }
+
+        $bridge = app(ActivityTaskBridgeContract::class);
+        if (! $bridge instanceof ActivityLeaseRenewal) {
+            return WorkerProtocol::json([
+                'lease_renewed' => false,
+                'can_continue' => false,
+                'reason' => 'activity_lease_renewal_unavailable',
+                'error' => 'The configured activity bridge does not support runtime lease renewal.',
+            ], 422);
+        }
+
+        try {
+            $response = $this->renewOwnedActivityLease($request, $taskId, $bridge);
+        } catch (\Throwable $exception) {
+            if (! BackendLockPressure::is($exception)) {
+                throw $exception;
+            }
+
+            $response = BackendLockPressure::workerOperationResponse($request, false);
+        }
+
+        return $response;
+    }
+
+    private function renewOwnedActivityLease(Request $request, string $taskId, ActivityLeaseRenewal $bridge): JsonResponse
+    {
+        $validated = $request->validate([
+            'activity_attempt_id' => ['required', 'string'],
+            'lease_owner' => ['required', 'string'],
+        ]);
+        $namespace = $request->attributes->get('namespace');
+        $observed = null;
+        if ($response = $this->leaseRenewalRefusal($namespace, $taskId, $validated, $observed)) {
+            return $response;
+        }
+
+        $workerSession = $this->storageMutations->run(
+            fn (): ?array => $this->workerSessions->heartbeatForAttempt(
+                $namespace,
+                $validated['activity_attempt_id'],
+                $validated['lease_owner'],
+            ),
+        );
+        if ($workerSession !== null && ($workerSession['admitted'] ?? false) !== true) {
+            return WorkerProtocol::json([
+                ...$observed,
+                'lease_renewed' => false,
+                'can_continue' => false,
+                'reason' => $workerSession['reason'] ?? 'worker_session_renewal_failed',
+                'worker_session' => $this->workerSessionSnapshotFromRenewal($workerSession),
+                'worker_session_renewal' => $this->workerSessionRenewalStatusPayload($workerSession),
+            ], $this->workerSessionRenewalHttpStatus($workerSession));
+        }
+
+        $status = $this->storageMutations->run(
+            static fn (): array => $bridge->renewLease($validated['activity_attempt_id'], $validated['lease_owner']),
+        );
+
+        return WorkerProtocol::json([
+            ...$status,
+            'task_id' => $taskId,
+            'worker_session' => $workerSession['session'] ?? null,
+        ], $this->outcomeStatus($status['reason'] ?? null));
+    }
+
+    private function leaseRenewalRefusal(string $namespace, string $taskId, array $validated, ?array &$observed): ?JsonResponse
+    {
+        if ($response = $this->guardAttemptOwnership(
+            $namespace,
+            $taskId,
+            $validated['activity_attempt_id'],
+            $validated['lease_owner'],
+            $observed,
+        )) {
+            return $response;
+        }
+
+        if (($observed['can_continue'] ?? false) !== true) {
+            return WorkerProtocol::json([
+                ...$observed, 'lease_renewed' => false,
+            ], $this->outcomeStatus($observed['reason'] ?? null));
+        }
+
+        $expiredReason = $this->expiredDeadlineReason(
+            $this->executionDeadlines($observed['activity_execution_id'] ?? null) ?? [],
+        );
+
+        return $expiredReason === null ? null : WorkerProtocol::json([
+            ...$observed, 'lease_renewed' => false, 'can_continue' => false, 'reason' => $expiredReason,
+        ], 409);
+    }
+
     /**
      * Heartbeat an in-progress activity task.
      *
@@ -1031,18 +1128,13 @@ private function guardOutcomeAuthority(string $taskId, string $attemptId, array
                 $reason = 'lease_expired';
             } else {
                 $deadlines = $this->executionDeadlines($status['activity_execution_id'] ?? null) ?? [];
-                foreach (['heartbeat', 'start_to_close', 'schedule_to_close'] as $kind) {
-                    if (isset($deadlines[$kind]) && ! CarbonImmutable::parse($deadlines[$kind])->isAfter(now())) {
-                        $reason = $kind.'_timeout';
-                        break;
-                    }
-                }
+                $reason = $this->expiredDeadlineReason($deadlines);
             }
         }
 
         // Closed, cancelled and reclaimed attempts retain their existing handling,
         // including the committed completion receipt and cancellation diagnostics.
-        if (! in_array($reason, ['lease_expired', 'heartbeat_timeout', 'start_to_close_timeout', 'schedule_to_close_timeout'], true)) {
+        if (! in_array($reason, ['lease_expired', 'heartbeat_timeout', 'start_to_close_timeout', 'schedule_to_close_timeout', 'run_timeout', 'execution_timeout'], true)) {
             return null;
         }
 
@@ -1184,7 +1276,7 @@ private function executionDeadlines(?string $executionId): ?array
         }
 
         /** @var ActivityExecution|null $execution */
-        $execution = ActivityExecution::query()->find($executionId);
+        $execution = ActivityExecution::query()->with('run')->find($executionId);
 
         if (! $execution) {
             return null;
@@ -1195,11 +1287,25 @@ private function executionDeadlines(?string $executionId): ?array
             'start_to_close' => $this->deadlineTimestamp($execution->close_deadline_at),
             'schedule_to_close' => $this->deadlineTimestamp($execution->schedule_to_close_deadline_at),
             'heartbeat' => $this->deadlineTimestamp($execution->heartbeat_deadline_at),
+            'run' => $this->deadlineTimestamp($execution->run?->run_deadline_at),
+            'execution' => $this->deadlineTimestamp($execution->run?->execution_deadline_at),
         ], static fn (mixed $v): bool => $v !== null);
 
         return $deadlines !== [] ? $deadlines : null;
     }
 
+    /** @param array<string, string> $deadlines */
+    private function expiredDeadlineReason(array $deadlines): ?string
+    {
+        foreach (['heartbeat', 'start_to_close', 'schedule_to_close', 'run', 'execution'] as $kind) {
+            if (isset($deadlines[$kind]) && ! CarbonImmutable::parse($deadlines[$kind])->isAfter(now())) {
+                return $kind.'_timeout';
+            }
+        }
+
+        return null;
+    }
+
     private function deadlineTimestamp(?CarbonInterface $deadline): ?string
     {
         return $deadline?->copy()

app/Http/Controllers/Api/WorkerDeregistrationController.php

@@ -22,8 +22,21 @@ public function destroy(Request $request, string $workerId): JsonResponse
             return $response;
         }
 
+        return $this->completeLegacyDeregistration($request, $workerId);
+    }
+
+    private function completeLegacyDeregistration(Request $request, string $workerId): JsonResponse
+    {
         $namespace = (string) $request->attributes->get('namespace');
-        $recoveredWorkflowTaskCount = $this->registrar->deregister($request, $namespace, $workerId);
+        try {
+            $recoveredWorkflowTaskCount = $this->registrar->deregister($request, $namespace, $workerId);
+        } catch (\Throwable $exception) {
+            if (! BackendLockPressure::is($exception)) {
+                throw $exception;
+            }
+
+            return BackendLockPressure::workerOperationResponse($request, false);
+        }
 
         if ($recoveredWorkflowTaskCount === null) {
             return WorkerProtocol::json([

app/Providers/AppServiceProvider.php

@@ -19,6 +19,7 @@
 use App\Observers\WorkflowHistoryEventObserver;
 use App\Observers\WorkflowTaskObserver;
 use App\Observers\WorkflowUpdateValidationObserver;
+use App\Support\ActivityTaskLeaseConfiguration;
 use App\Support\DispatchObservability;
 use App\Support\ExternalWorkflowUpdateAdmission;
 use App\Support\NamespaceExternalPayloadStorage;
@@ -202,6 +203,7 @@ public function boot(): void
         WorkflowPackageApiFloor::assert();
 
         WorkflowTaskLeaseConfiguration::apply();
+        ActivityTaskLeaseConfiguration::apply();
         WorkflowMemoRollingCompatibility::register();
 
         config([

app/Support/ActivityTaskLeaseConfiguration.php

@@ -0,0 +1,30 @@
+<?php
+
+namespace App\Support;
+
+use Workflow\V2\Support\ActivityLease;
+
+final class ActivityTaskLeaseConfiguration
+{
+    public static function apply(): int
+    {
+        $configured = config('server.lease.activity_task_timeout', 300);
+        $seconds = is_int($configured) || is_string($configured)
+            ? filter_var($configured, FILTER_VALIDATE_INT, [
+                'options' => ['min_range' => 1, 'max_range' => 86_400],
+            ]) : false;
+        if ($seconds === false) {
+            throw new ActivityTaskLeaseConfigurationException('DW_ACTIVITY_TASK_TIMEOUT must be an integer between 1 and 86400 seconds.');
+        }
+
+        config(['server.lease.activity_task_timeout' => $seconds]);
+        if (defined(ActivityLease::class.'::CONFIG_KEY') && method_exists(ActivityLease::class, 'seconds')) {
+            config([ActivityLease::CONFIG_KEY => $seconds]);
+            if (ActivityLease::seconds() !== $seconds) {
+                throw new ActivityTaskLeaseConfigurationException('The configured activity-task lease does not match the Workflow runtime.');
+            }
+        }
+
+        return $seconds;
+    }
+}

app/Support/ActivityTaskLeaseConfigurationException.php

@@ -0,0 +1,7 @@
+<?php
+
+namespace App\Support;
+
+use RuntimeException;
+
+final class ActivityTaskLeaseConfigurationException extends RuntimeException {}