package service import ( "strings" "testing" "browser.local/platform/domain" ) func TestCoreServiceRunJobLifecycle(t *testing.T) { svc, sessionToken := newRegisteredRunJobService(t) createQueuedRunJob(t, svc, "job-1", "idem-1") claim, err := svc.ClaimRunJob(domain.RunJobClaim{ RunEndpointID: "run-local", SessionToken: sessionToken, Capabilities: []string{"process.start"}, Capacity: domain.RunCapacity{MaxJobs: 4}, }) if err != nil { t.Fatalf("claim job: %v", err) } if !claim.Accepted || !claim.HasJob || claim.Job.JobID != "job-1" || claim.Job.State != domain.JobStateAccepted { t.Fatalf("expected claimed job, got %+v", claim) } ack, err := svc.AckRunJob(domain.RunJobAck{ RunEndpointID: "run-local", SessionToken: sessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, Message: "starting", }) if err != nil { t.Fatalf("ack job: %v", err) } if ack.Job.State != domain.JobStateRunning || ack.Job.Progress.Message != "starting" { t.Fatalf("expected running ack job, got %+v", ack) } progress, err := svc.UpdateRunJobProgress(domain.RunJobProgress{ RunEndpointID: "run-local", SessionToken: sessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, Progress: domain.RunJobProgressReport{Percent: 50, Message: "half"}, }) if err != nil { t.Fatalf("progress job: %v", err) } if progress.Job.Progress.Percent != 50 || progress.Job.Progress.Message != "half" { t.Fatalf("expected progress update, got %+v", progress) } result, err := svc.CompleteRunJob(domain.RunJobResult{ RunEndpointID: "run-local", SessionToken: sessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, State: domain.JobStateSucceeded, Progress: domain.RunJobProgressReport{Percent: 100, Message: "done"}, ResultRef: "artifact://jobs/job-1/result", Message: "done", }) if err != nil { t.Fatalf("complete job: %v", err) } if result.Job.State != domain.JobStateSucceeded || result.Job.ResultRef != "artifact://jobs/job-1/result" { t.Fatalf("expected succeeded result, got %+v", result) } stored, err := svc.GetJob("job-1") if err != nil { t.Fatalf("get completed job: %v", err) } if stored.State != domain.JobStateSucceeded || stored.Progress.Percent != 100 { t.Fatalf("expected stored terminal job, got %+v", stored) } } func TestCoreServiceRunJobClaimNoJob(t *testing.T) { svc, sessionToken := newRegisteredRunJobService(t) claim, err := svc.ClaimRunJob(domain.RunJobClaim{ RunEndpointID: "run-local", SessionToken: sessionToken, Capabilities: []string{"process.start"}, Capacity: domain.RunCapacity{MaxJobs: 4}, }) if err != nil { t.Fatalf("claim no job: %v", err) } if !claim.Accepted || claim.HasJob || claim.Job != nil || claim.NextPollSeconds <= 0 { t.Fatalf("expected empty claim response, got %+v", claim) } } func TestCoreServiceRunJobClaimSkipsServerFileCapabilityWithoutDeclaration(t *testing.T) { svc := newTestCoreService() plugin, endpoint := createPluginAndRunEndpoint(t, svc) instance, err := svc.CreateServerInstance(domain.ServerInstance{ ID: "server-file-claim", PluginID: plugin.ID, RunEndpointID: endpoint.ID, Name: "File Claim Server", State: domain.ServerInstanceStateRunning, }) if err != nil { t.Fatalf("create server instance: %v", err) } helloRequest := validRunControlHello() helloRequest.CapabilityReport.Capabilities = []string{"control.hello", "control.heartbeat", "process.start"} helloRequest.CapabilityReport.Fingerprint = "cap-file-claim-no-list" hello, err := svc.RegisterRunHello(helloRequest) if err != nil { t.Fatalf("register run hello: %v", err) } _, err = svc.CreateJob(domain.Job{ ID: "job-file-list", ServerInstanceID: instance.ID, RunEndpointID: endpoint.ID, Capability: domain.JobCapabilityFilesList, TargetKey: "server-root", IdempotencyKey: "idem-file-list", }) if err != nil { t.Fatalf("create file list job: %v", err) } claim, err := svc.ClaimRunJob(domain.RunJobClaim{ RunEndpointID: endpoint.ID, SessionToken: hello.SessionToken, Capabilities: []string{"process.start"}, Capacity: domain.RunCapacity{MaxJobs: 4}, }) if err != nil { t.Fatalf("claim file list job: %v", err) } if !claim.Accepted || claim.HasJob || claim.Job != nil || claim.NextPollSeconds <= 0 { t.Fatalf("expected unsupported file job to remain queued, got %+v", claim) } jobs, err := svc.store.Jobs().List(domain.JobFilter{RunEndpointID: endpoint.ID}) if err != nil || len(jobs) != 1 || jobs[0].State != domain.JobStateQueued { t.Fatalf("expected unsupported file job to remain queued in storage, jobs=%+v err=%v", jobs, err) } } func TestCoreServiceRunJobRejectsInvalidSessionAndLease(t *testing.T) { svc, sessionToken := newRegisteredRunJobService(t) createQueuedRunJob(t, svc, "job-1", "idem-1") _, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: "run-local", SessionToken: "stale", Capacity: domain.RunCapacity{MaxJobs: 4}}) if err == nil || !strings.Contains(err.Error(), "sessionToken") { t.Fatalf("expected invalid session rejection, got %v", err) } claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: "run-local", SessionToken: sessionToken, Capacity: domain.RunCapacity{MaxJobs: 4}}) if err != nil { t.Fatalf("claim job: %v", err) } _, err = svc.UpdateRunJobProgress(domain.RunJobProgress{ RunEndpointID: "run-local", SessionToken: sessionToken, JobID: claim.Job.JobID, LeaseToken: "bad-lease", Attempt: claim.Job.Attempt, Progress: domain.RunJobProgressReport{Percent: 10}, }) if err == nil || !strings.Contains(err.Error(), "leaseToken") { t.Fatalf("expected invalid lease rejection, got %v", err) } } func TestCoreServiceRunJobRejectsInvalidProgress(t *testing.T) { svc, sessionToken := newRegisteredRunJobService(t) createQueuedRunJob(t, svc, "job-1", "idem-1") claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: "run-local", SessionToken: sessionToken, Capacity: domain.RunCapacity{MaxJobs: 4}}) if err != nil { t.Fatalf("claim job: %v", err) } _, err = svc.UpdateRunJobProgress(domain.RunJobProgress{ RunEndpointID: "run-local", SessionToken: sessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, Progress: domain.RunJobProgressReport{Percent: 101}, }) if err == nil || !strings.Contains(err.Error(), "progress.percent") { t.Fatalf("expected invalid progress rejection, got %v", err) } } func TestCoreServiceRunJobCancelPoll(t *testing.T) { svc, sessionToken := newRegisteredRunJobService(t) createQueuedRunJob(t, svc, "job-1", "idem-1") claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: "run-local", SessionToken: sessionToken, Capacity: domain.RunCapacity{MaxJobs: 4}}) if err != nil { t.Fatalf("claim job: %v", err) } if _, err := svc.AckRunJob(domain.RunJobAck{RunEndpointID: "run-local", SessionToken: sessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt}); err != nil { t.Fatalf("ack job: %v", err) } cancel, err := svc.RequestRunJobCancel(domain.RunJobCancelRequest{JobID: "job-1", Reason: "operator requested"}) if err != nil { t.Fatalf("request cancel: %v", err) } if !cancel.Accepted || cancel.Reason != "operator requested" { t.Fatalf("unexpected cancel request: %+v", cancel) } poll, err := svc.PollRunJobCancel(domain.RunJobCancelPoll{RunEndpointID: "run-local", SessionToken: sessionToken, JobID: "job-1", LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt}) if err != nil { t.Fatalf("poll cancel: %v", err) } if !poll.HasCancel || poll.JobID != "job-1" || poll.Reason != "operator requested" { t.Fatalf("expected cancel poll result, got %+v", poll) } } func TestCoreServiceRunJobTerminalResultIsIdempotent(t *testing.T) { svc, sessionToken := newRegisteredRunJobService(t) createQueuedRunJob(t, svc, "job-1", "idem-1") claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: "run-local", SessionToken: sessionToken, Capacity: domain.RunCapacity{MaxJobs: 4}}) if err != nil { t.Fatalf("claim job: %v", err) } request := domain.RunJobResult{ RunEndpointID: "run-local", SessionToken: sessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt, State: domain.JobStateSucceeded, Progress: domain.RunJobProgressReport{Percent: 100, Message: "done"}, ResultRef: "artifact://jobs/job-1/result", Message: "done", } first, err := svc.CompleteRunJob(request) if err != nil { t.Fatalf("complete first: %v", err) } second, err := svc.CompleteRunJob(request) if err != nil { t.Fatalf("complete duplicate: %v", err) } if second.Job.State != first.Job.State || second.Job.ResultRef != first.Job.ResultRef { t.Fatalf("expected duplicate result to be idempotent, got %+v %+v", first, second) } request.State = domain.JobStateFailed request.Message = "failed" _, err = svc.CompleteRunJob(request) if err == nil || !strings.Contains(err.Error(), "conflicts") { t.Fatalf("expected conflicting result rejection, got %v", err) } } func TestCoreServiceRunJobReconcile(t *testing.T) { svc, sessionToken := newRegisteredRunJobService(t) createQueuedRunJob(t, svc, "job-1", "idem-1") claim, err := svc.ClaimRunJob(domain.RunJobClaim{RunEndpointID: "run-local", SessionToken: sessionToken, Capacity: domain.RunCapacity{MaxJobs: 4}}) if err != nil { t.Fatalf("claim job: %v", err) } if _, err := svc.AckRunJob(domain.RunJobAck{RunEndpointID: "run-local", SessionToken: sessionToken, JobID: claim.Job.JobID, LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt}); err != nil { t.Fatalf("ack job: %v", err) } reconcile, err := svc.ReconcileRunJobs(domain.RunJobReconcile{RunEndpointID: "run-local", SessionToken: sessionToken, ActiveJobs: []domain.RunJobReconcileEntry{ {JobID: "job-1", LeaseToken: claim.Job.LeaseToken, Attempt: claim.Job.Attempt}, {JobID: "local-only", LeaseToken: "local-lease", Attempt: 1}, }}) if err != nil { t.Fatalf("reconcile jobs: %v", err) } if len(reconcile.ConfirmedJobs) != 1 || reconcile.ConfirmedJobs[0].JobID != "job-1" { t.Fatalf("expected platform active job, got %+v", reconcile) } if len(reconcile.DiscardJobIDs) != 1 || reconcile.DiscardJobIDs[0] != "local-only" { t.Fatalf("expected unknown local job, got %+v", reconcile.DiscardJobIDs) } } func newRegisteredRunJobService(t *testing.T) (*CoreService, string) { t.Helper() svc := newTestCoreService() helloRequest := validRunControlHello() helloRequest.CapabilityReport.Capabilities = append(helloRequest.CapabilityReport.Capabilities, "process.start") helloRequest.CapabilityReport.Fingerprint = "cap-jobs" hello, err := svc.RegisterRunHello(helloRequest) if err != nil { t.Fatalf("register run hello: %v", err) } return svc, hello.SessionToken } func createQueuedRunJob(t *testing.T, svc *CoreService, id string, idempotencyKey string) domain.Job { t.Helper() job, err := svc.CreateJob(domain.Job{ ID: id, RunEndpointID: "run-local", Capability: "process.start", IdempotencyKey: idempotencyKey, }) if err != nil { t.Fatalf("create queued job: %v", err) } return job }