feat: 完整游戏运维功能
This commit is contained in:
npc0-hue
+435
-167
@@ -1,6 +1,7 @@
|
||||
package service
|
||||
|
||||
import (
|
||||
"crypto/subtle"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
@@ -10,14 +11,22 @@ import (
|
||||
"browser.local/platform/validator"
|
||||
)
|
||||
|
||||
const defaultJobPollSeconds = 2
|
||||
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
|
||||
}
|
||||
if err := svc.validateRunSession(claim.RunEndpointID, claim.SessionToken); err != nil {
|
||||
session, err := svc.validatedRunSession(claim.RunEndpointID, claim.SessionToken)
|
||||
if err != nil {
|
||||
return domain.RunJobClaimResult{}, err
|
||||
}
|
||||
|
||||
@@ -25,31 +34,40 @@ func (svc *CoreService) ClaimRunJob(claim domain.RunJobClaim) (domain.RunJobClai
|
||||
svc.jobMu.Lock()
|
||||
defer svc.jobMu.Unlock()
|
||||
|
||||
jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: claim.RunEndpointID, State: domain.JobStateQueued})
|
||||
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 := firstSupportedJob(jobs, claim.Capabilities)
|
||||
job, ok := firstEligibleSupportedJob(jobs, claim.Capabilities, stamp)
|
||||
if !ok {
|
||||
return domain.RunJobClaimResult{
|
||||
Accepted: true,
|
||||
RunEndpointID: claim.RunEndpointID,
|
||||
NextPollSeconds: defaultJobPollSeconds,
|
||||
ServerTime: stamp,
|
||||
}, nil
|
||||
return emptyJobClaim(claim.RunEndpointID, stamp), nil
|
||||
}
|
||||
|
||||
lease := svc.newJobLease(job.ID, claim.RunEndpointID, claim.SessionToken, stamp)
|
||||
svc.jobLeases[job.ID] = lease
|
||||
leaseToken, err := randomToken()
|
||||
if err != nil {
|
||||
return domain.RunJobClaimResult{}, err
|
||||
}
|
||||
job = normalizeJobScheduling(job, stamp)
|
||||
job.Attempt++
|
||||
job.State = domain.JobStateAccepted
|
||||
job.Progress = domain.JobProgress{Percent: 0, 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 := validator.ValidateJob(job); err != nil {
|
||||
if err := svc.updateScheduledJob(job); err != nil {
|
||||
return domain.RunJobClaimResult{}, err
|
||||
}
|
||||
if err := svc.store.Jobs().Update(job); err != nil {
|
||||
return domain.RunJobClaimResult{}, err
|
||||
}
|
||||
assignment := assignmentFromJob(job, lease)
|
||||
assignment := assignmentFromJob(job, leaseToken)
|
||||
return domain.CopyRunJobClaimResult(domain.RunJobClaimResult{
|
||||
Accepted: true,
|
||||
RunEndpointID: claim.RunEndpointID,
|
||||
@@ -64,20 +82,26 @@ func (svc *CoreService) AckRunJob(ack domain.RunJobAck) (domain.RunJobAckResult,
|
||||
if err := validator.ValidateRunJobAck(ack); err != nil {
|
||||
return domain.RunJobAckResult{}, err
|
||||
}
|
||||
if err := svc.validateRunSession(ack.RunEndpointID, ack.SessionToken); err != nil {
|
||||
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, lease, err := svc.activeLeasedJob(ack.RunEndpointID, ack.SessionToken, ack.JobID, ack.LeaseToken, ack.Attempt)
|
||||
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{Accepted: true, Job: assignmentFromJob(job, lease), ServerTime: stamp}, nil
|
||||
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")
|
||||
@@ -86,92 +110,134 @@ func (svc *CoreService) AckRunJob(ack domain.RunJobAck) (domain.RunJobAckResult,
|
||||
if strings.TrimSpace(ack.Message) != "" {
|
||||
job.Progress.Message = ack.Message
|
||||
}
|
||||
job.AckDeadlineAt = time.Time{}
|
||||
job.LeaseExpiresAt = stamp.Add(defaultJobLeaseDuration)
|
||||
job.UpdatedAt = stamp
|
||||
if err := validator.ValidateJob(job); err != nil {
|
||||
if err := svc.updateScheduledJob(job); err != nil {
|
||||
return domain.RunJobAckResult{}, err
|
||||
}
|
||||
if err := svc.store.Jobs().Update(job); err != nil {
|
||||
return domain.RunJobAckResult{}, err
|
||||
}
|
||||
lease.UpdatedAt = stamp
|
||||
svc.jobLeases[job.ID] = lease
|
||||
return domain.RunJobAckResult{Accepted: true, Job: assignmentFromJob(job, lease), ServerTime: stamp}, nil
|
||||
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
|
||||
}
|
||||
if err := svc.validateRunSession(progress.RunEndpointID, progress.SessionToken); err != nil {
|
||||
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, lease, err := svc.activeLeasedJob(progress.RunEndpointID, progress.SessionToken, progress.JobID, progress.LeaseToken, progress.Attempt)
|
||||
job, err := svc.fencedJob(session, progress.JobID, progress.LeaseToken, progress.Attempt)
|
||||
if err != nil {
|
||||
return domain.RunJobProgressResult{}, err
|
||||
}
|
||||
if job.State != domain.JobStateAccepted && job.State != domain.JobStateRunning {
|
||||
return domain.RunJobProgressResult{}, validationError("job is not active")
|
||||
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.State = domain.JobStateRunning
|
||||
job.Progress = domain.JobProgress{Percent: progress.Progress.Percent, Message: progress.Progress.Message}
|
||||
job.UpdatedAt = stamp
|
||||
if err := validator.ValidateJob(job); err != nil {
|
||||
return domain.RunJobProgressResult{}, err
|
||||
if progress.Sequence > 0 {
|
||||
job.LastProgressSeq = progress.Sequence
|
||||
}
|
||||
if err := svc.store.Jobs().Update(job); err != nil {
|
||||
job.LeaseExpiresAt = stamp.Add(defaultJobLeaseDuration)
|
||||
job.UpdatedAt = stamp
|
||||
if err := svc.updateScheduledJob(job); err != nil {
|
||||
return domain.RunJobProgressResult{}, err
|
||||
}
|
||||
if err := svc.projectDistributionBuildProgress(job, stamp); err != nil {
|
||||
return domain.RunJobProgressResult{}, err
|
||||
}
|
||||
lease.UpdatedAt = stamp
|
||||
svc.jobLeases[job.ID] = lease
|
||||
return domain.RunJobProgressResult{Accepted: true, Job: assignmentFromJob(job, lease), ServerTime: stamp}, nil
|
||||
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
|
||||
}
|
||||
if err := svc.validateRunSession(result.RunEndpointID, result.SessionToken); err != nil {
|
||||
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, lease, err := svc.activeLeasedJob(result.RunEndpointID, result.SessionToken, result.JobID, result.LeaseToken, result.Attempt)
|
||||
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 lease.TerminalFingerprint != "" && lease.TerminalFingerprint == fingerprint {
|
||||
if err := svc.projectLifecycleJobResult(job, stamp); err != nil {
|
||||
return domain.RunJobResultResult{}, err
|
||||
}
|
||||
if err := svc.projectDistributionBuildResult(job, stamp); err != nil {
|
||||
return domain.RunJobResultResult{}, err
|
||||
}
|
||||
return domain.RunJobResultResult{Accepted: true, Job: assignmentFromJob(job, lease), ServerTime: stamp}, nil
|
||||
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, 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, 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.store.Jobs().Update(job); err != nil {
|
||||
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 {
|
||||
@@ -180,17 +246,84 @@ func (svc *CoreService) CompleteRunJob(result domain.RunJobResult) (domain.RunJo
|
||||
if err := svc.projectDistributionBuildResult(job, stamp); err != nil {
|
||||
return domain.RunJobResultResult{}, err
|
||||
}
|
||||
lease.TerminalFingerprint = fingerprint
|
||||
lease.UpdatedAt = stamp
|
||||
svc.jobLeases[job.ID] = lease
|
||||
return domain.RunJobResultResult{Accepted: true, Job: assignmentFromJob(job, lease), ServerTime: stamp}, nil
|
||||
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
|
||||
}
|
||||
return domain.RunJobResultResult{Accepted: true, Job: assignmentFromJob(job, result.LeaseToken), ServerTime: stamp}, nil
|
||||
}
|
||||
|
||||
func validateExecutionResultForJob(job domain.Job, result domain.RunJobResult) error {
|
||||
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()
|
||||
@@ -199,43 +332,59 @@ func (svc *CoreService) RequestRunJobCancel(request domain.RunJobCancelRequest)
|
||||
if err != nil {
|
||||
return domain.RunJobCancelRequestResult{}, err
|
||||
}
|
||||
if !isActiveJobState(job.State) {
|
||||
return domain.RunJobCancelRequestResult{}, validationError("job is not active")
|
||||
job = normalizeJobScheduling(job, stamp)
|
||||
if job.State == domain.JobStateCancelled {
|
||||
return cancelRequestResult(job), nil
|
||||
}
|
||||
lease, exists := svc.jobLeases[job.ID]
|
||||
if !exists {
|
||||
return domain.RunJobCancelRequestResult{}, validationError("job lease is missing")
|
||||
if isTerminalJobState(job.State) {
|
||||
return domain.RunJobCancelRequestResult{}, validationError("job is already terminal")
|
||||
}
|
||||
lease.CancelReason = request.Reason
|
||||
lease.CancelRequestedAt = stamp
|
||||
lease.UpdatedAt = stamp
|
||||
svc.jobLeases[job.ID] = lease
|
||||
return domain.RunJobCancelRequestResult{Accepted: true, JobID: job.ID, Reason: request.Reason, RequestedAt: stamp}, nil
|
||||
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
|
||||
}
|
||||
if err := svc.validateRunSession(poll.RunEndpointID, poll.SessionToken); err != nil {
|
||||
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()
|
||||
|
||||
lease, ok := svc.findCancelLease(poll)
|
||||
if !ok {
|
||||
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: lease.JobID,
|
||||
Reason: lease.CancelReason,
|
||||
RequestedAt: lease.CancelRequestedAt,
|
||||
JobID: job.ID,
|
||||
Reason: job.CancelReason,
|
||||
RequestedAt: job.CancelRequestedAt,
|
||||
ServerTime: stamp,
|
||||
}, nil
|
||||
}
|
||||
@@ -245,140 +394,198 @@ func (svc *CoreService) ReconcileRunJobs(reconcile domain.RunJobReconcile) (doma
|
||||
if err := validator.ValidateRunJobReconcile(reconcile); err != nil {
|
||||
return domain.RunJobReconcileResult{}, err
|
||||
}
|
||||
if err := svc.validateRunSession(reconcile.RunEndpointID, reconcile.SessionToken); err != nil {
|
||||
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
|
||||
}
|
||||
activeByID := map[string]domain.Job{}
|
||||
for _, job := range jobs {
|
||||
if isActiveJobState(job.State) {
|
||||
activeByID[job.ID] = job
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
activeJobs := make([]domain.RunJobAssignment, 0, len(activeByID))
|
||||
ids := make([]string, 0, len(activeByID))
|
||||
for id := range activeByID {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
sort.Strings(ids)
|
||||
for _, id := range ids {
|
||||
job := activeByID[id]
|
||||
lease := svc.jobLeases[job.ID]
|
||||
if lease.JobID == "" || lease.SessionToken != reconcile.SessionToken {
|
||||
lease = svc.newJobLease(job.ID, reconcile.RunEndpointID, reconcile.SessionToken, stamp)
|
||||
} else {
|
||||
lease.UpdatedAt = stamp
|
||||
}
|
||||
svc.jobLeases[job.ID] = lease
|
||||
activeJobs = append(activeJobs, assignmentFromJob(job, lease))
|
||||
}
|
||||
|
||||
unknown := make([]string, 0)
|
||||
for _, reportedID := range reconcile.ActiveJobIDs {
|
||||
if _, exists := activeByID[reportedID]; !exists {
|
||||
unknown = append(unknown, reportedID)
|
||||
}
|
||||
}
|
||||
sort.Strings(unknown)
|
||||
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,
|
||||
ActiveJobs: activeJobs,
|
||||
UnknownJobIDs: unknown,
|
||||
ConfirmedJobs: confirmed,
|
||||
DiscardJobIDs: discard,
|
||||
ServerTime: stamp,
|
||||
}), nil
|
||||
}
|
||||
|
||||
func (svc *CoreService) validateRunSession(runEndpointID string, sessionToken string) error {
|
||||
func (svc *CoreService) validatedRunSession(runEndpointID string, sessionToken string) (domain.RunControlSession, error) {
|
||||
svc.controlMu.Lock()
|
||||
defer svc.controlMu.Unlock()
|
||||
session, exists := svc.runSessions[runEndpointID]
|
||||
if !exists || session.SessionToken != sessionToken {
|
||||
return validationError("sessionToken is invalid")
|
||||
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) newJobLease(jobID string, runEndpointID string, sessionToken string, stamp time.Time) domain.RunJobLease {
|
||||
svc.jobLeaseSeq++
|
||||
return domain.RunJobLease{
|
||||
JobID: jobID,
|
||||
RunEndpointID: runEndpointID,
|
||||
SessionToken: sessionToken,
|
||||
LeaseToken: fmt.Sprintf("job-lease:%s:%d:%d", jobID, stamp.UnixNano(), svc.jobLeaseSeq),
|
||||
Attempt: int(svc.jobLeaseSeq),
|
||||
CreatedAt: stamp,
|
||||
UpdatedAt: stamp,
|
||||
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) activeLeasedJob(runEndpointID string, sessionToken string, jobID string, leaseToken string, attempt int) (domain.Job, domain.RunJobLease, error) {
|
||||
job, err := svc.store.Jobs().Get(jobID)
|
||||
if err != nil {
|
||||
return domain.Job{}, domain.RunJobLease{}, err
|
||||
}
|
||||
if job.RunEndpointID != runEndpointID {
|
||||
return domain.Job{}, domain.RunJobLease{}, validationError("job runEndpointId does not match request")
|
||||
}
|
||||
lease, exists := svc.jobLeases[jobID]
|
||||
if !exists || lease.SessionToken != sessionToken || lease.LeaseToken != leaseToken || lease.Attempt != attempt {
|
||||
return domain.Job{}, domain.RunJobLease{}, validationError("leaseToken is invalid")
|
||||
}
|
||||
return job, lease, nil
|
||||
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) findCancelLease(poll domain.RunJobCancelPoll) (domain.RunJobLease, bool) {
|
||||
if poll.JobID != "" {
|
||||
lease, exists := svc.jobLeases[poll.JobID]
|
||||
if !exists || lease.RunEndpointID != poll.RunEndpointID || lease.SessionToken != poll.SessionToken {
|
||||
return domain.RunJobLease{}, false
|
||||
}
|
||||
if poll.LeaseToken != "" && lease.LeaseToken != poll.LeaseToken {
|
||||
return domain.RunJobLease{}, false
|
||||
}
|
||||
return lease, lease.CancelReason != ""
|
||||
func (svc *CoreService) updateScheduledJob(job domain.Job) error {
|
||||
if err := validator.ValidateJob(job); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
ids := make([]string, 0, len(svc.jobLeases))
|
||||
for id := range svc.jobLeases {
|
||||
ids = append(ids, id)
|
||||
}
|
||||
sort.Strings(ids)
|
||||
for _, id := range ids {
|
||||
lease := svc.jobLeases[id]
|
||||
if lease.RunEndpointID == poll.RunEndpointID && lease.SessionToken == poll.SessionToken && lease.CancelReason != "" {
|
||||
return lease, true
|
||||
}
|
||||
}
|
||||
return domain.RunJobLease{}, false
|
||||
return svc.store.Jobs().Update(job)
|
||||
}
|
||||
|
||||
func firstSupportedJob(jobs []domain.Job, capabilities []string) (domain.Job, bool) {
|
||||
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 {
|
||||
if len(capabilitySet) == 0 {
|
||||
return job, true
|
||||
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 _, supported := capabilitySet[job.Capability]; supported {
|
||||
return job, true
|
||||
if len(capabilitySet) > 0 {
|
||||
if _, supported := capabilitySet[job.Capability]; !supported {
|
||||
continue
|
||||
}
|
||||
}
|
||||
return job, true
|
||||
}
|
||||
return domain.Job{}, false
|
||||
}
|
||||
|
||||
func assignmentFromJob(job domain.Job, lease domain.RunJobLease) domain.RunJobAssignment {
|
||||
func assignmentFromJob(job domain.Job, leaseToken string) domain.RunJobAssignment {
|
||||
return domain.RunJobAssignment{
|
||||
JobID: job.ID,
|
||||
ServerInstanceID: job.ServerInstanceID,
|
||||
@@ -390,15 +597,76 @@ func assignmentFromJob(job domain.Job, lease domain.RunJobLease) domain.RunJobAs
|
||||
State: job.State,
|
||||
Progress: domain.RunJobProgressReport{Percent: job.Progress.Percent, Message: job.Progress.Message},
|
||||
ResultRef: job.ResultRef,
|
||||
LeaseToken: lease.LeaseToken,
|
||||
Attempt: lease.Attempt,
|
||||
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},
|
||||
LeaseToken: leaseToken,
|
||||
Attempt: job.Attempt,
|
||||
MaxAttempts: job.RetryPolicy.MaxAttempts,
|
||||
AckDeadlineAt: job.AckDeadlineAt,
|
||||
LeaseExpiresAt: job.LeaseExpiresAt,
|
||||
NextAttemptAt: job.NextAttemptAt,
|
||||
ProgressSequence: job.LastProgressSeq,
|
||||
CreatedAt: job.CreatedAt,
|
||||
UpdatedAt: job.UpdatedAt,
|
||||
}
|
||||
}
|
||||
|
||||
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", result.State, result.Progress.Percent, result.ResultRef, result.Message, result.ErrorCode, result.Progress.Message)
|
||||
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 {
|
||||
|
||||
Reference in New Issue
Block a user