browser/platform/service/job_channel.go

package service

import (
	"context"
	"crypto/subtle"
	"fmt"
	"sort"
	"strings"
	"time"

	"browser.local/platform/domain"
	"browser.local/platform/validator"
)

const (
	defaultJobPollSeconds           = 1
	defaultJobMaxAttempts           = 3
	defaultJobInitialBackoffSeconds = 2
	defaultJobMaxBackoffSeconds     = 60
	defaultJobAckTimeout            = 15 * time.Second
	defaultJobLeaseDuration         = 60 * time.Second
)

func (svc *CoreService) ClaimRunJob(claim domain.RunJobClaim) (domain.RunJobClaimResult, error) {
	claim = domain.CopyRunJobClaim(claim)
	if err := validator.ValidateRunJobClaim(claim); err != nil {
		return domain.RunJobClaimResult{}, err
	}
	session, err := svc.validatedRunSession(claim.RunEndpointID, claim.SessionToken)
	if err != nil {
		return domain.RunJobClaimResult{}, err
	}
	if session.RequireSignedRequests {
		claim.Capabilities = withoutCapability(claim.Capabilities, domain.JobCapabilityDistributionBuild)
	}

	stamp := svc.now()
	svc.jobMu.Lock()
	defer svc.jobMu.Unlock()

	if err := svc.sweepExpiredJobs(claim.RunEndpointID, stamp); err != nil {
		return domain.RunJobClaimResult{}, err
	}
	if claim.Capacity.MaxJobs > 0 && claim.Capacity.RunningJobs >= claim.Capacity.MaxJobs {
		return emptyJobClaim(claim.RunEndpointID, stamp), nil
	}
	jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: claim.RunEndpointID})
	if err != nil {
		return domain.RunJobClaimResult{}, err
	}
	job, ok := firstEligibleSupportedJob(jobs, claim.Capabilities, stamp)
	if !ok {
		return emptyJobClaim(claim.RunEndpointID, stamp), nil
	}

	leaseToken, err := randomToken()
	if err != nil {
		return domain.RunJobClaimResult{}, err
	}
	job = normalizeJobScheduling(job, stamp)
	job.Attempt++
	job.State = domain.JobStateAccepted
	phase := job.Progress.Phase
	if job.ExecutionInput.Deployment != nil {
		phase = "claimed"
	}
	job.Progress = domain.JobProgress{Percent: 0, Phase: phase, Message: "claimed; awaiting Run acknowledgement"}
	job.NextAttemptAt = time.Time{}
	job.LeaseTokenHash = tokenHash(leaseToken)
	job.LeaseSessionGen = session.Generation
	job.AckDeadlineAt = stamp.Add(defaultJobAckTimeout)
	job.LeaseExpiresAt = stamp.Add(defaultJobLeaseDuration)
	job.LastProgressSeq = 0
	job.UpdatedAt = stamp
	if err := svc.updateScheduledJob(job); err != nil {
		return domain.RunJobClaimResult{}, err
	}
	assignment := assignmentFromJob(job, leaseToken)
	return domain.CopyRunJobClaimResult(domain.RunJobClaimResult{
		Accepted:        true,
		RunEndpointID:   claim.RunEndpointID,
		HasJob:          true,
		Job:             &assignment,
		NextPollSeconds: defaultJobPollSeconds,
		ServerTime:      stamp,
	}), nil
}

func (svc *CoreService) ClaimRunJobWithWait(ctx context.Context, claim domain.RunJobClaim) (domain.RunJobClaimResult, error) {
	claim = domain.CopyRunJobClaim(claim)
	result, err := svc.ClaimRunJob(claim)
	if err != nil || result.HasJob || claim.WaitSeconds <= 0 || runJobClaimAtCapacity(claim) {
		return result, err
	}
	waiter := svc.registerRunJobWaiter(claim.RunEndpointID)
	defer svc.unregisterRunJobWaiter(claim.RunEndpointID, waiter)
	result, err = svc.ClaimRunJob(claim)
	if err != nil || result.HasJob {
		return result, err
	}
	timer := time.NewTimer(time.Duration(claim.WaitSeconds) * time.Second)
	defer timer.Stop()
	select {
	case <-ctx.Done():
		return domain.RunJobClaimResult{}, ctx.Err()
	case <-waiter:
	case <-timer.C:
	}
	return svc.ClaimRunJob(claim)
}

func runJobClaimAtCapacity(claim domain.RunJobClaim) bool {
	return claim.Capacity.MaxJobs > 0 && claim.Capacity.RunningJobs >= claim.Capacity.MaxJobs
}

func (svc *CoreService) registerRunJobWaiter(runEndpointID string) chan struct{} {
	waiter := make(chan struct{})
	svc.jobWaitMu.Lock()
	svc.jobWaiters[runEndpointID] = append(svc.jobWaiters[runEndpointID], waiter)
	svc.jobWaitMu.Unlock()
	return waiter
}

func (svc *CoreService) unregisterRunJobWaiter(runEndpointID string, waiter chan struct{}) {
	svc.jobWaitMu.Lock()
	waiters := svc.jobWaiters[runEndpointID]
	for index, candidate := range waiters {
		if candidate == waiter {
			waiters = append(waiters[:index], waiters[index+1:]...)
			break
		}
	}
	if len(waiters) == 0 {
		delete(svc.jobWaiters, runEndpointID)
	} else {
		svc.jobWaiters[runEndpointID] = waiters
	}
	svc.jobWaitMu.Unlock()
}

func (svc *CoreService) notifyRunJobWaiters(runEndpointID string) {
	svc.jobWaitMu.Lock()
	waiters := svc.jobWaiters[runEndpointID]
	delete(svc.jobWaiters, runEndpointID)
	svc.jobWaitMu.Unlock()
	for _, waiter := range waiters {
		close(waiter)
	}
	svc.publishRunControlEvent(runEndpointID, domain.RunControlEventTypeJobChanged)
}

func withoutCapability(capabilities []string, forbidden string) []string {
	filtered := make([]string, 0, len(capabilities))
	for _, capability := range capabilities {
		if capability != forbidden {
			filtered = append(filtered, capability)
		}
	}
	return filtered
}

func (svc *CoreService) AckRunJob(ack domain.RunJobAck) (domain.RunJobAckResult, error) {
	if err := validator.ValidateRunJobAck(ack); err != nil {
		return domain.RunJobAckResult{}, err
	}
	session, err := svc.validatedRunSession(ack.RunEndpointID, ack.SessionToken)
	if err != nil {
		return domain.RunJobAckResult{}, err
	}
	stamp := svc.now()
	svc.jobMu.Lock()
	defer svc.jobMu.Unlock()

	job, err := svc.fencedJob(session, ack.JobID, ack.LeaseToken, ack.Attempt)
	if err != nil {
		return domain.RunJobAckResult{}, err
	}
	if isTerminalJobState(job.State) {
		return domain.RunJobAckResult{}, validationError("late ack rejected for terminal job")
	}
	if job.State == domain.JobStateAccepted && deadlineExpired(job.AckDeadlineAt, stamp) {
		if err := svc.expireJobAttempt(&job, stamp, "Run acknowledgement deadline expired"); err != nil {
			return domain.RunJobAckResult{}, err
		}
		return domain.RunJobAckResult{}, validationError("ack deadline expired")
	}
	if job.State != domain.JobStateAccepted && job.State != domain.JobStateRunning {
		return domain.RunJobAckResult{}, validationError("job is not claimable for ack")
	}
	job.State = domain.JobStateRunning
	if strings.TrimSpace(ack.Message) != "" {
		job.Progress.Message = ack.Message
	}
	job.AckDeadlineAt = time.Time{}
	job.LeaseExpiresAt = stamp.Add(defaultJobLeaseDuration)
	job.UpdatedAt = stamp
	if err := svc.updateScheduledJob(job); err != nil {
		return domain.RunJobAckResult{}, err
	}
	return domain.RunJobAckResult{Accepted: true, Job: assignmentFromJob(job, ack.LeaseToken), ServerTime: stamp}, nil
}

func (svc *CoreService) UpdateRunJobProgress(progress domain.RunJobProgress) (domain.RunJobProgressResult, error) {
	if err := validator.ValidateRunJobProgress(progress); err != nil {
		return domain.RunJobProgressResult{}, err
	}
	session, err := svc.validatedRunSession(progress.RunEndpointID, progress.SessionToken)
	if err != nil {
		return domain.RunJobProgressResult{}, err
	}
	stamp := svc.now()
	svc.jobMu.Lock()
	defer svc.jobMu.Unlock()

	job, err := svc.fencedJob(session, progress.JobID, progress.LeaseToken, progress.Attempt)
	if err != nil {
		return domain.RunJobProgressResult{}, err
	}
	if job.State != domain.JobStateRunning {
		return domain.RunJobProgressResult{}, validationError("job is not running")
	}
	if deadlineExpired(job.LeaseExpiresAt, stamp) {
		if err := svc.expireJobAttempt(&job, stamp, "Run execution lease expired"); err != nil {
			return domain.RunJobProgressResult{}, err
		}
		return domain.RunJobProgressResult{}, validationError("job lease expired")
	}
	if progress.Sequence > 0 && progress.Sequence <= job.LastProgressSeq {
		return domain.RunJobProgressResult{}, validationError("progress sequence is stale")
	}
	job.Progress = domain.JobProgress{Percent: progress.Progress.Percent, Phase: progress.Progress.Phase, Message: progress.Progress.Message}
	if progress.Sequence > 0 {
		job.LastProgressSeq = progress.Sequence
	}
	job.LeaseExpiresAt = stamp.Add(defaultJobLeaseDuration)
	job.UpdatedAt = stamp
	if err := svc.updateScheduledJob(job); err != nil {
		return domain.RunJobProgressResult{}, err
	}
	if err := svc.projectServerDeploymentProgress(job, stamp); err != nil {
		return domain.RunJobProgressResult{}, err
	}
	if err := svc.projectDistributionBuildProgress(job, stamp); err != nil {
		return domain.RunJobProgressResult{}, err
	}
	if err := svc.projectDependencyAndRunUpdateProgress(job, stamp); err != nil {
		return domain.RunJobProgressResult{}, err
	}
	if err := svc.projectClientManagerLifecycleProgress(job, stamp); err != nil {
		return domain.RunJobProgressResult{}, err
	}
	return domain.RunJobProgressResult{Accepted: true, Job: assignmentFromJob(job, progress.LeaseToken), ServerTime: stamp}, nil
}

func (svc *CoreService) CompleteRunJob(result domain.RunJobResult) (domain.RunJobResultResult, error) {
	if err := validator.ValidateRunJobResult(result); err != nil {
		return domain.RunJobResultResult{}, err
	}
	session, err := svc.validatedRunSession(result.RunEndpointID, result.SessionToken)
	if err != nil {
		return domain.RunJobResultResult{}, err
	}
	stamp := svc.now()
	svc.jobMu.Lock()
	defer svc.jobMu.Unlock()

	job, err := svc.fencedJob(session, result.JobID, result.LeaseToken, result.Attempt)
	if err != nil {
		return domain.RunJobResultResult{}, err
	}
	fingerprint := terminalFingerprint(result)
	if isTerminalJobState(job.State) {
		if job.TerminalFingerprint == fingerprint {
			return domain.RunJobResultResult{Accepted: true, Job: assignmentFromJob(job, result.LeaseToken), ServerTime: stamp}, nil
		}
		return domain.RunJobResultResult{}, validationError("terminal result conflicts with existing job result")
	}
	if job.State != domain.JobStateAccepted && job.State != domain.JobStateRunning {
		return domain.RunJobResultResult{}, validationError("job attempt is no longer active")
	}
	if deadlineExpired(job.LeaseExpiresAt, stamp) {
		if err := svc.expireJobAttempt(&job, stamp, "Run execution lease expired"); err != nil {
			return domain.RunJobResultResult{}, err
		}
		return domain.RunJobResultResult{}, validationError("job lease expired")
	}
	if !job.CancelRequestedAt.IsZero() && result.State != domain.JobStateCancelled {
		return domain.RunJobResultResult{}, validationError("cancel intent requires a cancelled terminal result")
	}
	if err := validateExecutionResultForJob(job, result); err != nil {
		return domain.RunJobResultResult{}, err
	}

	if result.State == domain.JobStateFailed && result.Retryable && job.Attempt < job.RetryPolicy.MaxAttempts && job.CancelRequestedAt.IsZero() {
		job.Progress = domain.JobProgress{Percent: result.Progress.Percent, Phase: result.Progress.Phase, Message: terminalMessage(result)}
		if err := svc.scheduleJobRetry(&job, stamp, "retryable Run failure"); err != nil {
			return domain.RunJobResultResult{}, err
		}
		return domain.RunJobResultResult{Accepted: true, Job: assignmentFromJob(job, ""), ServerTime: stamp}, nil
	}

	job.State = result.State
	job.Progress = domain.JobProgress{Percent: result.Progress.Percent, Phase: result.Progress.Phase, Message: terminalMessage(result)}
	job.ResultRef = result.ResultRef
	job.ExecutionResult = result.ExecutionResult
	job.TerminalAt = stamp
	job.TerminalFingerprint = fingerprint
	job.AckDeadlineAt = time.Time{}
	job.LeaseExpiresAt = time.Time{}
	if result.State == domain.JobStateCancelled {
		job.CancelCompletedAt = stamp
		if job.CancelRequestedAt.IsZero() {
			job.CancelRequestedAt = stamp
			job.CancelReason = terminalMessage(result)
		}
	}
	job.UpdatedAt = stamp
	if err := validator.ValidateJob(job); err != nil {
		return domain.RunJobResultResult{}, err
	}
	if err := svc.validateDistributionBuildResult(job); err != nil {
		return domain.RunJobResultResult{}, err
	}
	if err := svc.updateScheduledJob(job); err != nil {
		return domain.RunJobResultResult{}, err
	}
	if err := svc.projectLifecycleJobResult(job, stamp); err != nil {
		return domain.RunJobResultResult{}, err
	}
	if err := svc.projectDistributionBuildResult(job, stamp); err != nil {
		return domain.RunJobResultResult{}, err
	}
	if err := svc.projectRemoteAdapterJobResult(job, stamp); err != nil {
		return domain.RunJobResultResult{}, err
	}
	if err := svc.projectDependencyAndRunUpdateResult(job, stamp); err != nil {
		return domain.RunJobResultResult{}, err
	}
	if err := svc.projectClientManagerLifecycleResult(job, stamp); err != nil {
		return domain.RunJobResultResult{}, err
	}
	if err := svc.projectPluginOperationsJobResult(job, stamp); err != nil {
		return domain.RunJobResultResult{}, err
	}
	return domain.RunJobResultResult{Accepted: true, Job: assignmentFromJob(job, result.LeaseToken), ServerTime: stamp}, nil
}

func validateExecutionResultForJob(job domain.Job, result domain.RunJobResult) error {
	if definition := job.ExecutionInput.Deployment; definition != nil {
		receipt := result.ExecutionResult.DeploymentReceipt
		if result.State == domain.JobStateSucceeded && definition.Mode == domain.ServerDeploymentModeCustom && receipt == nil {
			return validationError("deployment execution receipt does not match leased definition")
		}
		if receipt != nil && (receipt.SchemaVersion != "1" || receipt.Revision != definition.Revision || receipt.Action != job.ExecutionInput.LifecycleOperation || receipt.Mode != definition.Mode || receipt.Shell != definition.Shell) {
			return validationError("deployment execution receipt does not match leased definition")
		}
	}
	if result.ExecutionResult.Kind == "" {
		return nil
	}
	switch job.Capability {
	case domain.JobCapabilityConfigWrite:
		if result.ExecutionResult.Kind != "file.write" {
			return validationError("config write result type is invalid")
		}
		if result.State != domain.JobStateSucceeded {
			return nil
		}
		if result.ExecutionResult.Version != job.ExecutionInput.ExpectedVersion+1 {
			return validationError("config write result version is invalid")
		}
		if result.ExecutionResult.Checksum == "" || result.ExecutionResult.Checksum != validator.BytesChecksum([]byte(job.ExecutionInput.Content)) {
			return validationError("config write result checksum is invalid")
		}
	case domain.JobCapabilityFilesList:
		if result.ExecutionResult.Kind != "file.list" {
			return validationError("file list result type is invalid")
		}
	case domain.JobCapabilityFilesRead:
		if result.ExecutionResult.Kind != "file.read" {
			return validationError("file read result type is invalid")
		}
	case domain.JobCapabilityFilesWrite:
		if result.ExecutionResult.Kind != "file.write" {
			return validationError("file write result type is invalid")
		}
	case domain.JobCapabilityDependenciesCheck:
		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "dependency.check" {
			return validationError("dependency check result type is invalid")
		}
	case domain.JobCapabilityDependenciesInstall:
		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "dependency.install" {
			return validationError("dependency install result type is invalid")
		}
	case domain.JobCapabilityRunSelfUpdate:
		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "run.update.staged" {
			return validationError("Run self-update result type is invalid")
		}
	case domain.JobCapabilityClientManagerDeploy:
		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "client-manager.deployed" {
			return validationError("client-manager deploy result type is invalid")
		}
	case domain.JobCapabilityClientManagerControl:
		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "client-manager.control" {
			return validationError("client-manager control result type is invalid")
		}
	case domain.JobCapabilityClientManagerUpdate:
		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "client-manager.updated" || result.State == domain.JobStateFailed && result.ExecutionResult.Kind != "" && result.ExecutionResult.Kind != "client-manager.rollback.restored" {
			return validationError("client-manager update result type is invalid")
		}
	case domain.JobCapabilityClientManagerRollback:
		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "client-manager.rolled-back" {
			return validationError("client-manager rollback result type is invalid")
		}
	case domain.JobCapabilityClientManagerUninstall:
		if result.State == domain.JobStateSucceeded && result.ExecutionResult.Kind != "client-manager.uninstalled" {
			return validationError("client-manager uninstall result type is invalid")
		}
	}
	return nil
}

func (svc *CoreService) RequestRunJobCancel(request domain.RunJobCancelRequest) (domain.RunJobCancelRequestResult, error) {
	if err := validator.ValidateRunJobCancelRequest(request); err != nil {
		return domain.RunJobCancelRequestResult{}, err
	}
	stamp := svc.now()
	svc.jobMu.Lock()
	defer svc.jobMu.Unlock()

	job, err := svc.store.Jobs().Get(request.JobID)
	if err != nil {
		return domain.RunJobCancelRequestResult{}, err
	}
	job = normalizeJobScheduling(job, stamp)
	if job.State == domain.JobStateCancelled {
		return cancelRequestResult(job), nil
	}
	if isTerminalJobState(job.State) {
		return domain.RunJobCancelRequestResult{}, validationError("job is already terminal")
	}
	if job.CancelRequestedAt.IsZero() {
		job.CancelReason = request.Reason
		job.CancelRequestedAt = stamp
	}
	if job.State == domain.JobStateQueued || job.State == domain.JobStateRetrying {
		terminalizeCancelled(&job, stamp, job.CancelReason)
	}
	job.UpdatedAt = stamp
	if err := svc.updateScheduledJob(job); err != nil {
		return domain.RunJobCancelRequestResult{}, err
	}
	return cancelRequestResult(job), nil
}

func (svc *CoreService) PollRunJobCancel(poll domain.RunJobCancelPoll) (domain.RunJobCancelPollResult, error) {
	if err := validator.ValidateRunJobCancelPoll(poll); err != nil {
		return domain.RunJobCancelPollResult{}, err
	}
	session, err := svc.validatedRunSession(poll.RunEndpointID, poll.SessionToken)
	if err != nil {
		return domain.RunJobCancelPollResult{}, err
	}
	stamp := svc.now()
	svc.jobMu.Lock()
	defer svc.jobMu.Unlock()

	job, err := svc.fencedJob(session, poll.JobID, poll.LeaseToken, poll.Attempt)
	if err != nil {
		return domain.RunJobCancelPollResult{}, err
	}
	if jobAttemptExpired(job, stamp) {
		if err := svc.expireJobAttempt(&job, stamp, "Run job deadline expired before cancel poll"); err != nil {
			return domain.RunJobCancelPollResult{}, err
		}
		return domain.RunJobCancelPollResult{}, validationError("job lease expired")
	}
	if job.CancelRequestedAt.IsZero() || !isActiveJobState(job.State) {
		return domain.RunJobCancelPollResult{Accepted: true, RunEndpointID: poll.RunEndpointID, ServerTime: stamp}, nil
	}
	return domain.RunJobCancelPollResult{
		Accepted:      true,
		RunEndpointID: poll.RunEndpointID,
		HasCancel:     true,
		JobID:         job.ID,
		Reason:        job.CancelReason,
		RequestedAt:   job.CancelRequestedAt,
		ServerTime:    stamp,
	}, nil
}

func (svc *CoreService) ReconcileRunJobs(reconcile domain.RunJobReconcile) (domain.RunJobReconcileResult, error) {
	reconcile = domain.CopyRunJobReconcile(reconcile)
	if err := validator.ValidateRunJobReconcile(reconcile); err != nil {
		return domain.RunJobReconcileResult{}, err
	}
	session, err := svc.validatedRunSession(reconcile.RunEndpointID, reconcile.SessionToken)
	if err != nil {
		return domain.RunJobReconcileResult{}, err
	}
	stamp := svc.now()
	svc.jobMu.Lock()
	defer svc.jobMu.Unlock()

	confirmed := make([]domain.RunJobAssignment, 0, len(reconcile.ActiveJobs))
	discard := make([]string, 0)
	confirmedIDs := map[string]struct{}{}
	for _, entry := range reconcile.ActiveJobs {
		job, getErr := svc.store.Jobs().Get(entry.JobID)
		if getErr != nil || job.RunEndpointID != reconcile.RunEndpointID || !isActiveJobState(job.State) || job.Attempt != entry.Attempt || !leaseTokenMatches(job.LeaseTokenHash, entry.LeaseToken) || jobAttemptExpired(job, stamp) {
			discard = append(discard, entry.JobID)
			continue
		}
		job = normalizeJobScheduling(job, stamp)
		job.LeaseSessionGen = session.Generation
		job.LeaseExpiresAt = stamp.Add(defaultJobLeaseDuration)
		job.LastReconciledAt = stamp
		job.ReconcileCount++
		job.ReconcileOutcome = "confirmed active attempt"
		job.UpdatedAt = stamp
		if err := svc.updateScheduledJob(job); err != nil {
			return domain.RunJobReconcileResult{}, err
		}
		confirmedIDs[job.ID] = struct{}{}
		confirmed = append(confirmed, assignmentFromJob(job, entry.LeaseToken))
	}

	jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: reconcile.RunEndpointID})
	if err != nil {
		return domain.RunJobReconcileResult{}, err
	}
	for _, job := range jobs {
		if !isActiveJobState(job.State) {
			continue
		}
		if _, ok := confirmedIDs[job.ID]; ok {
			continue
		}
		job = normalizeJobScheduling(job, stamp)
		job.LastReconciledAt = stamp
		job.ReconcileCount++
		job.ReconcileOutcome = "missing from Run journal"
		if err := svc.expireJobAttempt(&job, stamp, "active attempt missing during Run reconciliation"); err != nil {
			return domain.RunJobReconcileResult{}, err
		}
	}
	sort.Strings(discard)
	sort.Slice(confirmed, func(i, j int) bool { return confirmed[i].JobID < confirmed[j].JobID })
	return domain.CopyRunJobReconcileResult(domain.RunJobReconcileResult{
		Accepted:      true,
		RunEndpointID: reconcile.RunEndpointID,
		ConfirmedJobs: confirmed,
		DiscardJobIDs: discard,
		ServerTime:    stamp,
	}), nil
}

func (svc *CoreService) validatedRunSession(runEndpointID string, sessionToken string) (domain.RunControlSession, error) {
	svc.controlMu.Lock()
	defer svc.controlMu.Unlock()
	return svc.currentRunSession(runEndpointID, sessionToken)
}

func (svc *CoreService) validateRunSession(runEndpointID string, sessionToken string) error {
	_, err := svc.validatedRunSession(runEndpointID, sessionToken)
	return err
}

func (svc *CoreService) fencedJob(session domain.RunControlSession, jobID string, leaseToken string, attempt int) (domain.Job, error) {
	job, err := svc.store.Jobs().Get(jobID)
	if err != nil {
		return domain.Job{}, err
	}
	job = normalizeJobScheduling(job, svc.now())
	if job.RunEndpointID != session.RunEndpointID {
		return domain.Job{}, validationError("job runEndpointId does not match request")
	}
	if job.Attempt != attempt || job.LeaseSessionGen != session.Generation || !leaseTokenMatches(job.LeaseTokenHash, leaseToken) {
		return domain.Job{}, validationError("attempt or leaseToken is invalid")
	}
	return job, nil
}

func (svc *CoreService) sweepExpiredJobs(runEndpointID string, stamp time.Time) error {
	jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: runEndpointID})
	if err != nil {
		return err
	}
	for _, job := range jobs {
		job = normalizeJobScheduling(job, stamp)
		expired := job.State == domain.JobStateAccepted && deadlineExpired(job.AckDeadlineAt, stamp)
		expired = expired || job.State == domain.JobStateRunning && deadlineExpired(job.LeaseExpiresAt, stamp)
		if expired {
			if err := svc.expireJobAttempt(&job, stamp, "Run job deadline expired"); err != nil {
				return err
			}
		}
	}
	return nil
}

func (svc *CoreService) expireJobAttempt(job *domain.Job, stamp time.Time, reason string) error {
	if !job.CancelRequestedAt.IsZero() {
		terminalizeCancelled(job, stamp, job.CancelReason)
		return svc.updateScheduledJob(*job)
	}
	if job.Attempt < job.RetryPolicy.MaxAttempts {
		return svc.scheduleJobRetry(job, stamp, reason)
	}
	job.State = domain.JobStateFailed
	job.Progress = domain.JobProgress{Percent: job.Progress.Percent, Message: reason + "; retry budget exhausted"}
	job.TerminalAt = stamp
	job.TerminalFingerprint = fmt.Sprintf("scheduler-failed|%d|%s", job.Attempt, reason)
	job.AckDeadlineAt = time.Time{}
	job.LeaseExpiresAt = time.Time{}
	job.UpdatedAt = stamp
	return svc.updateScheduledJob(*job)
}

func (svc *CoreService) scheduleJobRetry(job *domain.Job, stamp time.Time, reason string) error {
	job.State = domain.JobStateRetrying
	job.Progress.Message = reason
	job.NextAttemptAt = stamp.Add(jobRetryBackoff(job.RetryPolicy, job.Attempt))
	job.LeaseTokenHash = ""
	job.LeaseSessionGen = 0
	job.AckDeadlineAt = time.Time{}
	job.LeaseExpiresAt = time.Time{}
	job.LastProgressSeq = 0
	job.UpdatedAt = stamp
	return svc.updateScheduledJob(*job)
}

func (svc *CoreService) updateScheduledJob(job domain.Job) error {
	if err := validator.ValidateJob(job); err != nil {
		return err
	}
	if err := svc.store.Jobs().Update(job); err != nil {
		return err
	}
	svc.notifyRunJobWaiters(job.RunEndpointID)
	return nil
}

func normalizeJobScheduling(job domain.Job, stamp time.Time) domain.Job {
	if job.RetryPolicy.MaxAttempts <= 0 {
		job.RetryPolicy.MaxAttempts = defaultJobMaxAttempts
	}
	if job.RetryPolicy.InitialBackoffSeconds <= 0 {
		job.RetryPolicy.InitialBackoffSeconds = defaultJobInitialBackoffSeconds
	}
	if job.RetryPolicy.MaxBackoffSeconds < job.RetryPolicy.InitialBackoffSeconds {
		job.RetryPolicy.MaxBackoffSeconds = defaultJobMaxBackoffSeconds
	}
	if job.QueueEligibleAt.IsZero() {
		if !job.CreatedAt.IsZero() {
			job.QueueEligibleAt = job.CreatedAt
		} else {
			job.QueueEligibleAt = stamp
		}
	}
	return job
}

func firstEligibleSupportedJob(jobs []domain.Job, capabilities []string, stamp time.Time) (domain.Job, bool) {
	capabilitySet := map[string]struct{}{}
	for _, capability := range capabilities {
		capabilitySet[capability] = struct{}{}
	}
	sort.SliceStable(jobs, func(i, j int) bool {
		if jobs[i].CreatedAt.Equal(jobs[j].CreatedAt) {
			return jobs[i].ID < jobs[j].ID
		}
		return jobs[i].CreatedAt.Before(jobs[j].CreatedAt)
	})
	for _, job := range jobs {
		job = normalizeJobScheduling(job, stamp)
		eligible := job.State == domain.JobStateQueued && !stamp.Before(job.QueueEligibleAt)
		eligible = eligible || job.State == domain.JobStateRetrying && !stamp.Before(job.NextAttemptAt)
		if !eligible || !job.CancelRequestedAt.IsZero() {
			continue
		}
		if len(capabilitySet) > 0 {
			if _, supported := capabilitySet[job.Capability]; !supported {
				continue
			}
		}
		return job, true
	}
	return domain.Job{}, false
}

func assignmentFromJob(job domain.Job, leaseToken string) domain.RunJobAssignment {
	return domain.RunJobAssignment{
		JobID:            job.ID,
		ServerInstanceID: job.ServerInstanceID,
		RunEndpointID:    job.RunEndpointID,
		Capability:       job.Capability,
		TargetKey:        job.TargetKey,
		InputRef:         job.InputRef,
		IdempotencyKey:   job.IdempotencyKey,
		State:            job.State,
		Progress:         domain.RunJobProgressReport{Percent: job.Progress.Percent, Phase: job.Progress.Phase, Message: job.Progress.Message},
		ResultRef:        job.ResultRef,
		ExecutionInput:   domain.JobExecutionInput{WorkspaceScope: job.ExecutionInput.WorkspaceScope, Content: job.ExecutionInput.Content, ExpectedVersion: job.ExecutionInput.ExpectedVersion, ExpectedChecksum: job.ExecutionInput.ExpectedChecksum, MaxReadBytes: job.ExecutionInput.MaxReadBytes, RemoteAdapterKey: job.ExecutionInput.RemoteAdapterKey, RemoteAdapterKind: job.ExecutionInput.RemoteAdapterKind, TimeoutSeconds: job.ExecutionInput.TimeoutSeconds, PluginID: job.ExecutionInput.PluginID, LifecycleOperation: job.ExecutionInput.LifecycleOperation, TargetVersion: job.ExecutionInput.TargetVersion, Inputs: domain.CopyStringMap(job.ExecutionInput.Inputs), LogSource: domain.CopyRuntimeLogSourcePtr(job.ExecutionInput.LogSource), LogSources: domain.CopyRuntimeLogSources(job.ExecutionInput.LogSources), DLLExtensions: append([]domain.RuntimeDLLExtensionPlan(nil), job.ExecutionInput.DLLExtensions...), SourceRCON: domain.CopyRuntimeSourceRCONPlan(job.ExecutionInput.SourceRCON), Deployment: deploymentPlanForDispatchValue(job.ExecutionInput.Deployment), ServerDeploymentPlan: domain.CopyServerDeploymentPlan(job.ExecutionInput.ServerDeploymentPlan)},
		LeaseToken:       leaseToken,
		Attempt:          job.Attempt,
		FencingToken:     0,
		MaxAttempts:      job.RetryPolicy.MaxAttempts,
		AckDeadlineAt:    job.AckDeadlineAt,
		LeaseExpiresAt:   job.LeaseExpiresAt,
		NextAttemptAt:    job.NextAttemptAt,
		ProgressSequence: job.LastProgressSeq,
		CreatedAt:        job.CreatedAt,
		UpdatedAt:        job.UpdatedAt,
	}
}

func deploymentPlanForDispatchValue(definition *domain.ServerDeploymentDefinition) *domain.ServerDeploymentDefinition {
	if definition == nil {
		return nil
	}
	copy := domain.CopyServerDeploymentDefinition(*definition)
	return &copy
}

func emptyJobClaim(runEndpointID string, stamp time.Time) domain.RunJobClaimResult {
	return domain.RunJobClaimResult{Accepted: true, RunEndpointID: runEndpointID, NextPollSeconds: defaultJobPollSeconds, ServerTime: stamp}
}

func cancelRequestResult(job domain.Job) domain.RunJobCancelRequestResult {
	return domain.RunJobCancelRequestResult{
		Accepted: true, JobID: job.ID, Reason: job.CancelReason, RequestedAt: job.CancelRequestedAt,
		CompletedAt: job.CancelCompletedAt, State: job.State,
	}
}

func terminalizeCancelled(job *domain.Job, stamp time.Time, reason string) {
	job.State = domain.JobStateCancelled
	job.Progress = domain.JobProgress{Percent: job.Progress.Percent, Message: reason}
	job.CancelCompletedAt = stamp
	job.TerminalAt = stamp
	job.TerminalFingerprint = fmt.Sprintf("scheduler-cancelled|%d|%s", job.Attempt, reason)
	job.NextAttemptAt = time.Time{}
	job.AckDeadlineAt = time.Time{}
	job.LeaseExpiresAt = time.Time{}
}

func leaseTokenMatches(expectedHash string, token string) bool {
	if expectedHash == "" || strings.TrimSpace(token) == "" {
		return false
	}
	actual := tokenHash(token)
	return subtle.ConstantTimeCompare([]byte(expectedHash), []byte(actual)) == 1
}

func deadlineExpired(deadline time.Time, stamp time.Time) bool {
	return deadline.IsZero() || !stamp.Before(deadline)
}

func jobAttemptExpired(job domain.Job, stamp time.Time) bool {
	if job.State == domain.JobStateAccepted {
		return deadlineExpired(job.AckDeadlineAt, stamp)
	}
	if job.State == domain.JobStateRunning {
		return deadlineExpired(job.LeaseExpiresAt, stamp)
	}
	return false
}

func jobRetryBackoff(policy domain.JobRetryPolicy, attempt int) time.Duration {
	delay := int64(policy.InitialBackoffSeconds)
	for current := 1; current < attempt && delay < int64(policy.MaxBackoffSeconds); current++ {
		delay *= 2
		if delay > int64(policy.MaxBackoffSeconds) {
			delay = int64(policy.MaxBackoffSeconds)
		}
	}
	return time.Duration(delay) * time.Second
}

func terminalFingerprint(result domain.RunJobResult) string {
	return fmt.Sprintf("%s|%d|%s|%s|%s|%s|%t", result.State, result.Progress.Percent, result.ResultRef, result.Message, result.ErrorCode, result.Progress.Message, result.Retryable)
}

func terminalMessage(result domain.RunJobResult) string {
	if strings.TrimSpace(result.Message) != "" {
		return result.Message
	}
	return result.Progress.Message
}

func isActiveJobState(state domain.JobState) bool {
	return state == domain.JobStateAccepted || state == domain.JobStateRunning
}

func isTerminalJobState(state domain.JobState) bool {
	return state == domain.JobStateSucceeded || state == domain.JobStateFailed || state == domain.JobStateCancelled
}