737 lines
28 KiB
Go
737 lines
28 KiB
Go
package service
|
|
|
|
import (
|
|
"crypto/subtle"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"browser.local/platform/domain"
|
|
"browser.local/platform/validator"
|
|
)
|
|
|
|
const (
|
|
defaultJobPollSeconds = 2
|
|
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
|
|
}
|
|
if session.RequireSignedRequests && len(claim.Capabilities) == 0 {
|
|
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 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.projectProtectedRequestJobResult(job, result, stamp); 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.projectProductionOpsJobResult(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.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.controlled" {
|
|
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
|
|
}
|
|
return svc.store.Jobs().Update(job)
|
|
}
|
|
|
|
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 {
|
|
fencingToken := uint64(0)
|
|
if isProtectedRequestCapability(job.Capability) {
|
|
fencingToken = uint64(job.Attempt)
|
|
}
|
|
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), 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: fencingToken,
|
|
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 ©
|
|
}
|
|
|
|
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
|
|
}
|