From 40207402080a9ad2537eb4e0eeb0d8fe5b9b4b5c Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 1 Oct 2026 12:29:19 +0000 Subject: [PATCH] feat(job-array-one-based-index): control by identity validates first and dispatches once (task 7.4) ControlJobByID parses every selection with ParseIdentitySelection before any store access (first rejection: 400, nothing dispatched), collects the jobs to act on de-duplicated by unique_job_id plus one not-found result per missing number per selection (carrying the would-be element ID), validates each job once after the loop and dispatches each accepted job once. foundJob gains an explicit-index flag so [0] matches only element 0 and never falls back to the whole array or an ordinary job. Fixes bug-1899 (earlier selections were re-dispatched for every later one). Co-Authored-By: Claude Opus 5.5 (1M context) --- .../specs/job-array-one-based-index/tasks.md | 2 +- src/frontend/apiserver/controllers/jobs.go | 402 +++++++----------- .../controllers/jobs_handler_test.go | 258 +++++++++++ .../apiserver/controllers/jobs_test.go | 54 ++- 4 files changed, 475 insertions(+), 241 deletions(-) diff --git a/.kiro/specs/job-array-one-based-index/tasks.md b/.kiro/specs/job-array-one-based-index/tasks.md index f9bb1fcc..8c659913 100644 --- a/.kiro/specs/job-array-one-based-index/tasks.md +++ b/.kiro/specs/job-array-one-based-index/tasks.md @@ -163,7 +163,7 @@ - _Requirements: 5.2, 5.5, 5.6, 5.7, 5.8, 5.11, 5.12, 5.14, 5.16, 5.17_ - _Depends: 5.4, 7.2_ -- [ ] 7.4 Control jobs by identity: validate first, dispatch once (fixes bug-1899) +- [x] 7.4 Control jobs by identity: validate first, dispatch once (fixes bug-1899) - Replace the index-section parsing in this handler with the REST selection parser; validate all selections before matching; give the element matcher an explicit-index flag so `[0]` never falls back to the whole array or an ordinary job - Collect the de-duplicated jobs to act on and, separately, one not-found result per missing number per selection carrying the would-be element ID; validate each job once after the loop, then dispatch each accepted job once (a dispatch error stays a 500) - Update the existing element-matcher test to the new signature with a tagged fixture diff --git a/src/frontend/apiserver/controllers/jobs.go b/src/frontend/apiserver/controllers/jobs.go index 8785bfaf..c6fcbb13 100644 --- a/src/frontend/apiserver/controllers/jobs.go +++ b/src/frontend/apiserver/controllers/jobs.go @@ -2147,207 +2147,85 @@ func (j JobController) ControlJobByID(ctx *gin.Context) { return } - //1. get user input job list and validate the job existed or not - // Extract job IDs using the new function + //1. parse and validate every selection before any store query or + // dispatch (validate-all-first): the first rejected selection, in request + // order, rejects the whole command, which then acts on no job and reports + // no "not found" item. jobids := extractJobNames(jobIDsStr) for i := range jobids { jobids[i] = strings.TrimSpace(jobids[i]) } logger.Debug("The input jobids", zap.Any("jobids", jobids)) - mergeJqueryReply := make([]utils.OrderMap, 0) - - // store all user's input jobid - allFiltCdArrays := make([]utils.FilterConditionArray, 0) + sels := make([]Selection, 0, len(jobids)) for _, jobid := range jobids { - - compfilters := make(utils.CompFilterCondition) - - filtCdArrays := make([]utils.FilterConditionArray, 0) - subORFilters := make([]utils.FilterCondition, 0) - - var err error - - jobid = strings.TrimSpace(jobid) - // parse the jobid to support array list format - jobid, err := parseArrayName(jobid) - if err != nil { - logger.Error("job id format is invalid", zap.Error(err)) - ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) + sel, se := ParseIdentitySelection(jobid) + if se != nil { + returnSelectionError(ctx, jobid, se) return } - logger.Debug("GetJobByIDs after parseArrayName", zap.Any("jobid", jobid)) + sels = append(sels, sel) + } - // gitea#45 follow-up: accept the human-readable job ID form - // ("{cluster}-{apiserver}-{timestamp}-{seq}", e.g. what `jobs list` - // prints) in addition to the numeric one xcecli already translates - // to -- lets a caller hit this endpoint directly without - // reimplementing xcecli's ID packing. Numeric IDs are unaffected. - if !utils.IsNumeric(jobid.KeyID) { - if packed, ok := backend.ParseHumanJobID(jobid.KeyID); ok { - jobid.KeyID = strconv.FormatUint(packed, 10) - } else { - err := errors.New("bad jobid: " + jobid.KeyID) - logger.Error(err.Error()) - ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) - return - } - } + //2. match every selection against the store. The loop only collects: + // (a) the jobs to act on, de-duplicated by unique_job_id, in selection + // order, and (b) one "not found" item per missing number per selection + // (not de-duplicated). Nothing is validated or dispatched inside it + // (bug-1899: dispatching there re-sent earlier selections' jobs for every + // later selection). + actJobs := make([]utils.OrderMap, 0) + seenJobs := make(map[int64]bool) + notFoundJobs := []map[string]interface{}{} - mainFiltCond := utils.FilterCondition{ - Field: "job_id", - Operator: "eq", - Value: jobid.KeyID, - } - - // treat every section as an OR condition - if jobid.Indices != nil { - // handle every section in Indices list. - for _, jidxsec := range jobid.Indices { - st, ed, err := parseArrayIndexSection(jidxsec.Index) - if err != nil { - err := errors.New("bad job array index: " + jidxsec.Index) - logger.Error(err.Error()) - ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) - return - } - - if jidxsec.Step == 1 { - subORFilters = append(subORFilters, utils.FilterCondition{ - Field: "array_idx", - Operator: "between", - Value: "(" + strconv.Itoa(st) + "," + strconv.Itoa(ed) + ")", - }) - } else { // handle step > 1 - idxList := make([]string, 0) - for idx := st; idx <= ed; idx += jidxsec.Step { - idxList = append(idxList, strconv.Itoa(idx)) - } - subORFilters = append(subORFilters, utils.FilterCondition{ - Field: "array_idx", - Operator: "in", - Value: "(" + strings.Join(idxList, ",") + ")", - }) - } - } - } - filtCdArrays = append(filtCdArrays, utils.FilterConditionArray{ - MainFiltCond: mainFiltCond, - SubFiltConds: subORFilters, - }) - - allFiltCdArrays = append(allFiltCdArrays, filtCdArrays...) - - compfilters["job_id"] = filtCdArrays - var job models.JobInfo - compfiltercd, err := utils.ConvertToNewFilterArray(compfilters, &job) - if err != nil { - logger.Error(err.Error()) - // TODO: the msg used here is not proper - ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) + for _, sel := range sels { + jobid := int64(sel.Key) + rows, ok := queryControlRowsByJobID(ctx, sel, fields, sortOrders) + if !ok { return } - var bookmark string = "" - var limit int = utils.PER_PAGE_LIMIT_MAX - for { - queryOpts, err := utils.BuildQueryOptions(nil, compfiltercd, sortOrders, limit, bookmark) - if err != nil { - logger.Error(err.Error()) - ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) - return - } - logger.Debug("the job query request:", zap.Any("jquery request arrayjob:", compfiltercd), zap.Any("sortOrders", sortOrders), zap.Any("limit", limit), - zap.String("pagetoken", bookmark), zap.Any("fields", fields)) - jqueryReply, page_token, err := services.QueryData(services.TableJobInfo, queryOpts, fields) - if err != nil { - logger.Error("Failed to query job info.", zap.Error(err)) - ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) - return - } - // Here, page_token is escaped and will not be recognized by dbapi, - // for internal use, need add Unescape here. - bookmark, _ = url.QueryUnescape(page_token) - mergeJqueryReply = append(mergeJqueryReply, jqueryReply...) - if bookmark == "" { - break - } - } // end for fetch job loop - - } // end of for - logger.Debug("job list got:", zap.Any("jobInfoList", mergeJqueryReply)) - - //2. validate job status and control action matching - jobInfoList := mergeJqueryReply - - successJobs := []map[string]interface{}{} - failedJobs := []map[string]interface{}{} - - // check if specific jobids are exist in jobInfoList. - for _, ujobFilterCdArray := range allFiltCdArrays { - var jobid int64 - var arrayIdx int foundJobs := make([]utils.OrderMap, 0) - - jobid, _ = strconv.ParseInt(ujobFilterCdArray.MainFiltCond.Value, 10, 64) - - if len(ujobFilterCdArray.SubFiltConds) == 0 { // for non index section sepcified; - arrayIdx = 0 - fjobs, err := foundJob(jobid, arrayIdx, jobInfoList) + if sel.Form == SelBare { + // no element part: every element of the array and any ordinary + // job with that identity (5.6) + fjobs, err := foundJob(jobid, 0, false, rows) if err != nil { - failedJobs = append(failedJobs, map[string]interface{}{ - "job_id": jobid, - "array_idx": arrayIdx, - "msg": "job not found", + notFoundJobs = append(notFoundJobs, map[string]interface{}{ + "unique_job_id": jobid, + "job_id": jobid, + "array_idx": 0, + "msg": "job not found", }) continue } foundJobs = append(foundJobs, fjobs...) } else { - for _, subFiltCond := range ujobFilterCdArray.SubFiltConds { - if subFiltCond.Operator == "between" { - // (st, ed) - value := strings.Trim(subFiltCond.Value, "()") - values := strings.Split(value, ",") - st, _ := strconv.Atoi(values[0]) - ed, _ := strconv.Atoi(values[1]) - for i := st; i <= ed; i++ { - arrayIdx = i - fjobs, err := foundJob(jobid, arrayIdx, jobInfoList) - if err != nil { - failedJobs = append(failedJobs, map[string]interface{}{ - "job_id": jobid, - "array_idx": arrayIdx, - "msg": "job not found", - }) - } else { - foundJobs = append(foundJobs, fjobs...) - } - } - } else if subFiltCond.Operator == "in" { - // (1,4,7,...) - value := strings.Trim(subFiltCond.Value, "()") - values := strings.Split(value, ",") - for _, v := range values { - arrayidx, _ := strconv.ParseInt(v, 10, 64) - arrayIdx = int(arrayidx) - fjobs, err := foundJob(jobid, arrayIdx, jobInfoList) - if err != nil { - failedJobs = append(failedJobs, map[string]interface{}{ - "job_id": jobid, - "array_idx": arrayIdx, - "msg": "job not found", - }) - } else { - foundJobs = append(foundJobs, fjobs...) - } - } + // explicit element numbers: only array elements with that + // number, never the whole array or an ordinary job (5.2, 5.4, + // 8.2); each missing number is reported with its would-be + // element ID (5.3, 5.4) + for k := sel.Lo; k <= sel.Hi; k++ { + fjobs, err := foundJob(jobid, k, true, rows) + if err != nil { + notFoundJobs = append(notFoundJobs, map[string]interface{}{ + "unique_job_id": int64(backend.MakeElementID(sel.Key, k)), + "job_id": jobid, + "array_idx": k, + "msg": "job not found", + }) + continue } + foundJobs = append(foundJobs, fjobs...) } - } for _, jobInfo := range foundJobs { + uid, _ := jobInfo.Data["unique_job_id"].(int64) + if seenJobs[uid] { + continue + } + seenJobs[uid] = true + // check rbac if job_owner, exist := jobInfo.Data["submit_user"]; exist { jobPermission, err := GetRequireResourcePermission(submitter, "job:control") @@ -2380,58 +2258,66 @@ func (j JobController) ControlJobByID(ctx *gin.Context) { accessQueues = queuePermission.DomainValues } // check job:control permission - if err != nil || len(accessQueues) == 0 || !utils.ContainsField(accessQueues, job_queue.(string)) { + if len(accessQueues) == 0 || !utils.ContainsField(accessQueues, job_queue.(string)) { ReturnError(ctx, common.ERROR_AUTH_CODE, common.ERROR_ACCESS_FORBIDDEN_MSG, nil) return } } - // check job status to see if can do corresponding action - validatedSuccessJobs, validatedFailedJobs := validateJobControl(jobInfo, convertedReq.Action, convertedReq.Signal) + actJobs = append(actJobs, jobInfo) + } + } + logger.Debug("jobs to act on:", zap.Any("jobs", actJobs)) - for _, successJob := range validatedSuccessJobs { - successJobs = append(successJobs, map[string]interface{}{ - "unique_job_id": successJob["unique_job_id"].(int64), - "job_id": successJob["job_id"].(int64), - "array_idx": successJob["array_idx"].(int), - "msg": successJob["msg"].(string), - }) - } + //3. validate each job once against the action, then append the not-found + // items; every result item carries unique_job_id (5.1, 8.3). + successJobs := []map[string]interface{}{} + failedJobs := []map[string]interface{}{} + for _, jobInfo := range actJobs { + validatedSuccessJobs, validatedFailedJobs := validateJobControl(jobInfo, convertedReq.Action, convertedReq.Signal) - for _, failedJob := range validatedFailedJobs { - failedJobs = append(failedJobs, map[string]interface{}{ - "unique_job_id": failedJob["unique_job_id"].(int64), - "job_id": failedJob["job_id"].(int64), - "array_idx": failedJob["array_idx"].(int), - "msg": failedJob["msg"].(string), - }) - } + for _, successJob := range validatedSuccessJobs { + successJobs = append(successJobs, map[string]interface{}{ + "unique_job_id": successJob["unique_job_id"].(int64), + "job_id": successJob["job_id"].(int64), + "array_idx": successJob["array_idx"].(int), + "msg": successJob["msg"].(string), + }) } - if len(successJobs) > 0 { - for _, successJob := range successJobs { - signalInt, _ := strconv.Atoi(signalval) - jobCtrlEvt := models.JobControlEvent{ - UniqueJobID: successJob["unique_job_id"].(int64), - Signal: signalInt, - Comment: convertedReq.Comment, - ForceKill: convertedReq.ForceKill, - Method: convertedReq.Method, - } + for _, failedJob := range validatedFailedJobs { + failedJobs = append(failedJobs, map[string]interface{}{ + "unique_job_id": failedJob["unique_job_id"].(int64), + "job_id": failedJob["job_id"].(int64), + "array_idx": failedJob["array_idx"].(int), + "msg": failedJob["msg"].(string), + }) + } + } + failedJobs = append(failedJobs, notFoundJobs...) - jobCtrlEvtBytes, err := json.Marshal(jobCtrlEvt) - if err != nil { - logger.Error("encode job control event error", zap.Error(err)) - ReturnError(ctx, common.ERROR_CODE, "encode job control event error"+": "+err.Error(), nil) - return - } + //4. dispatch each accepted job once; a dispatch error is a 500. + signalInt, _ := strconv.Atoi(signalval) + for _, successJob := range successJobs { + jobCtrlEvt := models.JobControlEvent{ + UniqueJobID: successJob["unique_job_id"].(int64), + Signal: signalInt, + Comment: convertedReq.Comment, + ForceKill: convertedReq.ForceKill, + Method: convertedReq.Method, + } - err = controlJob(uint64(successJob["unique_job_id"].(int64)), acttype, string(jobCtrlEvtBytes)) - if err != nil { - logger.Error("send job control to backend error", zap.Error(err)) - ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) - return - } - } + jobCtrlEvtBytes, err := json.Marshal(jobCtrlEvt) + if err != nil { + logger.Error("encode job control event error", zap.Error(err)) + ReturnError(ctx, common.ERROR_CODE, "encode job control event error"+": "+err.Error(), nil) + return + } + + err = controlJob(uint64(successJob["unique_job_id"].(int64)), acttype, string(jobCtrlEvtBytes)) + if err != nil { + logger.Error("send job control to backend error", zap.Error(err)) + ReturnError(ctx, common.ERROR_CODE, common.ERROR_WRITE_DATA_MSG+": "+err.Error(), nil) + return } } @@ -2922,31 +2808,73 @@ func validateJobControl(jobinfo utils.OrderMap, action string, return successJobs, failedJobs } -/* check if the user specify job is found in job list - * when array_idx = 0, may be an array job, but without index section specify, will only compare jobId - */ -func foundJob(jobid int64, array_idx int, jobinfos []utils.OrderMap) ([]utils.OrderMap, error) { +// foundJob returns the rows of jobinfos a control selection names. +// +// With explicitIndex (the selection has an element part, "[k]" or +// "[a-b]"), only array elements -- rows whose unique_job_id carries the +// array marking -- of that array with array_idx == idx match, so "[0]" +// never falls back to the whole array or to an ordinary job with identity +// (5.4, 8.2). Without it, every row with that job_id matches: all +// elements of the array and any ordinary job (5.6). +func foundJob(jobID int64, idx int, explicitIndex bool, rows []utils.OrderMap) ([]utils.OrderMap, error) { jobs := make([]utils.OrderMap, 0) - for _, jobinfo := range jobinfos { - if array_idx > 0 { - // bug-1742: array_idx is a plain Go `int` on models.JobInfo, not - // `int64` -- utils.IntOrInt64 accepts either. - jobinfoArrayIdx, ok := utils.IntOrInt64(jobinfo.Data["array_idx"]) - if ok && jobinfo.Data["job_id"].(int64) == jobid && - jobinfoArrayIdx == array_idx { - jobs = append(jobs, jobinfo) - } - } else { - if jobinfo.Data["job_id"].(int64) == jobid { - jobs = append(jobs, jobinfo) - } + for _, row := range rows { + rowJobID, ok := rowUint64(row, "job_id") + if !ok || rowJobID != uint64(jobID) { + continue } + if explicitIndex && !rowElementInRange(row, idx, idx) { + continue + } + jobs = append(jobs, row) } - if len(jobs) > 0 { - return jobs, nil - } else { + if len(jobs) == 0 { return nil, errors.New("job not found") } + return jobs, nil +} + +// queryControlRowsByJobID fetches every stored row whose job_id is the +// selection's key (all pages), for ControlJobByID to match with foundJob. On +// a failure it writes the error reply and returns ok == false. +func queryControlRowsByJobID(ctx *gin.Context, sel Selection, fields []string, + sortOrders []utils.SortOrder) (rows []utils.OrderMap, ok bool) { + filtercds, err := utils.ConvertToNewFilter([]utils.FilterCondition{{ + Field: "job_id", + Operator: "eq", + Value: strconv.FormatUint(sel.Key, 10), + }}, &models.JobInfo{}) + if err != nil { + logger.Error(err.Error()) + ReturnError(ctx, common.INVALID_PARAMS_CODE, common.INVALID_PARAMS_MSG+": "+err.Error(), nil) + return nil, false + } + + rows = make([]utils.OrderMap, 0) + bookmark := "" + for { + queryOpts, err := utils.BuildQueryOptions(filtercds, nil, sortOrders, utils.PER_PAGE_LIMIT_MAX, bookmark) + if err != nil { + logger.Error(err.Error()) + ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) + return nil, false + } + logger.Debug("the job query request:", zap.String("selection", sel.Raw), zap.Any("filter", filtercds), + zap.String("pagetoken", bookmark), zap.Any("fields", fields)) + reply, pageToken, err := services.QueryData(services.TableJobInfo, queryOpts, fields) + if err != nil { + logger.Error("Failed to query job info.", zap.Error(err)) + ReturnError(ctx, common.ERROR_CODE, common.ERROR_QUERY_JOBINFO_MSG+": "+err.Error(), nil) + return nil, false + } + rows = append(rows, reply...) + // page_token is escaped; unescape it for internal use + bookmark, _ = url.QueryUnescape(pageToken) + if bookmark == "" { + break + } + } + return rows, true } func processJobData(rows []utils.OrderMap, pagetoken string) utils.QueryResult { diff --git a/src/frontend/apiserver/controllers/jobs_handler_test.go b/src/frontend/apiserver/controllers/jobs_handler_test.go index 53ac1d93..f1c83406 100644 --- a/src/frontend/apiserver/controllers/jobs_handler_test.go +++ b/src/frontend/apiserver/controllers/jobs_handler_test.go @@ -976,3 +976,261 @@ func TestJobHandler_GetJobByIDsSelections(t *testing.T) { assert400(t, "5.14", f.getJobs("64-0-1-0"), "wider than allowed") assert400(t, "5.5", f.getJobs(X+"[3-1]"), "reversed") } + +// ========== Control by identity (task 7.4, bug-1899) ========== + +// controlItem is one accepted or failed item of a control reply. +type controlItem struct { + UniqueJobID uint64 + HasUnique bool + JobID uint64 + ArrayIdx int + Msg string +} + +// controlReply is the data of a 202 control reply. +type controlReply struct { + Status string + Accepted []controlItem + Failed []controlItem +} + +// decodeControlReply decodes a successful (202) control reply. +func decodeControlReply(t *testing.T, w *httptest.ResponseRecorder) controlReply { + t.Helper() + if w.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body: %s", w.Code, w.Body.String()) + } + dec := json.NewDecoder(bytes.NewReader(w.Body.Bytes())) + dec.UseNumber() + var raw struct { + Code int `json:"code"` + Data struct { + Status string `json:"status"` + Accepted []map[string]interface{} `json:"accepted"` + Failed []map[string]interface{} `json:"failed"` + } `json:"data"` + } + if err := dec.Decode(&raw); err != nil { + t.Fatalf("decode control reply: %v; body: %s", err, w.Body.String()) + } + if raw.Code != 202 { + t.Fatalf("code = %d, want 202; body: %s", raw.Code, w.Body.String()) + } + num := func(m map[string]interface{}, k string) (uint64, bool) { + n, ok := m[k].(json.Number) + if !ok { + return 0, false + } + v, err := strconv.ParseUint(n.String(), 10, 64) + if err != nil { + // job_id/unique_job_id are int64 on the wire; keep the bit pattern + s, err2 := strconv.ParseInt(n.String(), 10, 64) + if err2 != nil { + t.Fatalf("%s %q: %v", k, n, err) + } + v = uint64(s) + } + return v, true + } + conv := func(in []map[string]interface{}) []controlItem { + out := make([]controlItem, 0, len(in)) + for _, m := range in { + var it controlItem + it.UniqueJobID, it.HasUnique = num(m, "unique_job_id") + it.JobID, _ = num(m, "job_id") + idx, _ := num(m, "array_idx") + it.ArrayIdx = int(idx) + it.Msg, _ = m["msg"].(string) + out = append(out, it) + } + return out + } + return controlReply{Status: raw.Data.Status, Accepted: conv(raw.Data.Accepted), Failed: conv(raw.Data.Failed)} +} + +// itemIDs returns the unique_job_id of every item, in reply order. +func itemIDs(t *testing.T, items []controlItem) []uint64 { + t.Helper() + ids := make([]uint64, 0, len(items)) + for _, it := range items { + if !it.HasUnique { + t.Errorf("result item without unique_job_id: %+v", it) + } + ids = append(ids, it.UniqueJobID) + } + return ids +} + +// assertExactIDs checks got equals want element by element (order kept). +func assertExactIDs(t *testing.T, name string, got, want []uint64) { + t.Helper() + if len(got) != len(want) { + t.Errorf("%s: %v, want %v", name, got, want) + return + } + for i := range got { + if got[i] != want[i] { + t.Errorf("%s: %v, want %v", name, got, want) + return + } + } +} + +// assertSeeded fails the test unless every id is in store.Jobs, so an +// act-on-no-job assertion cannot pass on an empty store. +func assertSeeded(t *testing.T, ids ...uint64) { + t.Helper() + for _, id := range ids { + if !store.Jobs.Has(id) { + t.Fatalf("fixture row %d missing from store.Jobs", id) + } + } +} + +// controlByIDEscaped posts a control action for a selection list, escaping +// the path segment. +func (f *handlerFixture) controlByIDEscaped(jobids, action string) *httptest.ResponseRecorder { + f.t.Helper() + return f.controlByID(url.PathEscape(jobids), action) +} + +// TestJobHandler_ControlTwoSelectionsDispatchOnce (bug-1899): a request with +// two selections dispatches each selected element exactly once and lists +// each once; the same holds for two ordinary jobs. +func TestJobHandler_ControlTwoSelectionsDispatchOnce(t *testing.T) { + f := newHandlerFixture(t) + X := handlerTestArrayBase + 30<<15 + Y := handlerTestArrayBase + 31<<15 + f.seedArrayElements(X, "foo", "RUN", 1, 2) + f.seedArrayElements(Y, "bar", "RUN", 1, 2) + A := handlerTestOrdinaryBase + 1 + B := handlerTestOrdinaryBase + 2 + f.seedOrdinaryJob(A, "a", "RUN") + f.seedOrdinaryJob(B, "b", "RUN") + + x1, y2 := backend.MakeElementID(X, 1), backend.MakeElementID(Y, 2) + r := decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+"[1],"+jobIDString(Y)+"[2]", "kill")) + assertExactIDs(t, "X[1],Y[2] dispatched", f.dispatchedIDs(), []uint64{x1, y2}) + assertExactIDs(t, "X[1],Y[2] accepted", itemIDs(t, r.Accepted), []uint64{x1, y2}) + if len(r.Failed) != 0 { + t.Errorf("X[1],Y[2] failed items: %+v", r.Failed) + } + + f.dispatched = nil + r = decodeControlReply(t, f.controlByIDEscaped(jobIDString(A)+","+jobIDString(B), "kill")) + assertExactIDs(t, "A,B dispatched", f.dispatchedIDs(), []uint64{A, B}) + assertExactIDs(t, "A,B accepted", itemIDs(t, r.Accepted), []uint64{A, B}) + + // a job selected by two selections is acted on once (de-duplicated by + // unique_job_id) + f.dispatched = nil + x2 := backend.MakeElementID(X, 2) + r = decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+","+jobIDString(X)+"[2]", "kill")) + assertExactIDs(t, "X,X[2] dispatched", f.dispatchedIDs(), []uint64{x1, x2}) + assertExactIDs(t, "X,X[2] accepted", itemIDs(t, r.Accepted), []uint64{x1, x2}) +} + +// TestJobHandler_ControlRangeReportsMissing (5.3): X[1-3] with element 2 +// missing dispatches elements 1 and 3 and reports 2 as not found, carrying +// its would-be element ID. +func TestJobHandler_ControlRangeReportsMissing(t *testing.T) { + f := newHandlerFixture(t) + X := handlerTestArrayBase + 32<<15 + f.seedArrayElements(X, "foo", "RUN", 1, 3, 4) + x1, x2, x3 := backend.MakeElementID(X, 1), backend.MakeElementID(X, 2), backend.MakeElementID(X, 3) + + r := decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+"[1-3]", "kill")) + assertExactIDs(t, "dispatched", f.dispatchedIDs(), []uint64{x1, x3}) + assertExactIDs(t, "accepted", itemIDs(t, r.Accepted), []uint64{x1, x3}) + assertExactIDs(t, "failed", itemIDs(t, r.Failed), []uint64{x2}) + if len(r.Failed) == 1 { + it := r.Failed[0] + if it.Msg != "job not found" || it.JobID != X || it.ArrayIdx != 2 { + t.Errorf("not-found item = %+v, want job_id %d, array_idx 2, msg \"job not found\"", it, X) + } + } + if r.Status != "partially_accepted" { + t.Errorf("status = %q, want partially_accepted", r.Status) + } + + // a range on an identity no array has: every number not found, no job + f.dispatched = nil + Z := handlerTestArrayBase + 33<<15 + r = decodeControlReply(t, f.controlByIDEscaped(jobIDString(Z)+"[1-2]", "kill")) + if got := f.dispatchedIDs(); len(got) != 0 { + t.Errorf("Z[1-2] dispatched %v, want none", got) + } + assertExactIDs(t, "Z[1-2] failed", itemIDs(t, r.Failed), + []uint64{backend.MakeElementID(Z, 1), backend.MakeElementID(Z, 2)}) +} + +// TestJobHandler_ControlElementZeroMissing (5.4): X[0] with no element 0 +// acts on no job -- not on the array, not on an ordinary job with identity +// X -- and reports element 0 as not found. +func TestJobHandler_ControlElementZeroMissing(t *testing.T) { + f := newHandlerFixture(t) + X := handlerTestArrayBase + 34<<15 + elems := f.seedArrayElements(X, "foo", "RUN", 1, 2) + // an ordinary job sharing the number X (array_idx 0) + f.seedOrdinaryJob(X, "ordinary", "RUN") + assertSeeded(t, uint64(elems[0].Unique_JobID), uint64(elems[1].Unique_JobID), X) + + for _, sel := range []string{"[0]", "[0-0]"} { + f.dispatched = nil + r := decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+sel, "kill")) + if got := f.dispatchedIDs(); len(got) != 0 { + t.Errorf("X%s dispatched %v, want none", sel, got) + } + if len(r.Accepted) != 0 { + t.Errorf("X%s accepted %+v, want none", sel, r.Accepted) + } + assertExactIDs(t, "X"+sel+" failed", itemIDs(t, r.Failed), []uint64{backend.MakeElementID(X, 0)}) + if len(r.Failed) == 1 && r.Failed[0].Msg != "job not found" { + t.Errorf("X%s failed msg = %q, want \"job not found\"", sel, r.Failed[0].Msg) + } + } +} + +// TestJobHandler_ControlPreUpgradeElementZero (8.2, 8.3): a pre-upgrade +// element 0 is dispatched alone, also when the same request selects another +// element of that array, and is reported with its own unique_job_id. +func TestJobHandler_ControlPreUpgradeElementZero(t *testing.T) { + f := newHandlerFixture(t) + X := handlerTestArrayBase + 35<<15 + f.seedArrayElements(X, "old", "RUN", 0, 1, 2) + x0, x2 := backend.MakeElementID(X, 0), backend.MakeElementID(X, 2) + + r := decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+"[0]", "kill")) + assertExactIDs(t, "X[0] dispatched", f.dispatchedIDs(), []uint64{x0}) + assertExactIDs(t, "X[0] accepted", itemIDs(t, r.Accepted), []uint64{x0}) + + f.dispatched = nil + r = decodeControlReply(t, f.controlByIDEscaped(jobIDString(X)+"[0],"+jobIDString(X)+"[2]", "kill")) + assertExactIDs(t, "X[0],X[2] dispatched", f.dispatchedIDs(), []uint64{x0, x2}) + assertExactIDs(t, "X[0],X[2] accepted", itemIDs(t, r.Accepted), []uint64{x0, x2}) +} + +// TestJobHandler_ControlValidThenInvalid (Terms): a valid selection followed +// by an invalid one rejects the whole command with 400 and dispatches +// nothing, without "not found" items. +func TestJobHandler_ControlValidThenInvalid(t *testing.T) { + f := newHandlerFixture(t) + X := handlerTestArrayBase + 36<<15 + elems := f.seedArrayElements(X, "foo", "RUN", 1, 2) + assertSeeded(t, uint64(elems[0].Unique_JobID)) + + cases := []struct{ sel, want string }{ + {jobIDString(X) + "[1]," + jobIDString(X) + "[40000]", "element numbers must be at most"}, + {jobIDString(X) + "[1]," + jobIDString(X) + "[3-1]", "reversed"}, + {jobIDString(X) + "[1]," + jobIDString(X) + "[a]", "invalid job identity or element selection"}, + {jobIDString(X) + "[1]," + jobIDString(uint64(elems[1].Unique_JobID)), "must be named as"}, + } + for _, c := range cases { + f.dispatched = nil + assert400(t, c.sel, f.controlByIDEscaped(c.sel, "kill"), c.want) + if got := f.dispatchedIDs(); len(got) != 0 { + t.Errorf("%s: dispatched %v, want none", c.sel, got) + } + } +} diff --git a/src/frontend/apiserver/controllers/jobs_test.go b/src/frontend/apiserver/controllers/jobs_test.go index a7628fa9..b9b5a88b 100644 --- a/src/frontend/apiserver/controllers/jobs_test.go +++ b/src/frontend/apiserver/controllers/jobs_test.go @@ -2,6 +2,7 @@ package controllers import ( "errors" + "apiserver/backend" "apiserver/models" "apiserver/store" "apiserver/utils" @@ -179,13 +180,17 @@ func TestValidateJobControl_RejectsNonKillableStatus(t *testing.T) { func TestFoundJob_MatchesRealArrayIdxType(t *testing.T) { resetJobSubmitState() + // an array element: its unique_job_id carries the array marking + const key = uint64(2)<<40 | uint64(1)<<20 + elem := backend.MakeElementID(key, 1) jobInfo := models.JobInfo{ - JobID: 1788186846, - ArrayIdx: 1, + Unique_JobID: int64(elem), + JobID: int64(key), + ArrayIdx: 1, } orderMap := store.StructToOrderMap(jobInfo) - found, err := foundJob(1788186846, 1, []utils.OrderMap{orderMap}) + found, err := foundJob(int64(key), 1, true, []utils.OrderMap{orderMap}) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -194,6 +199,49 @@ func TestFoundJob_MatchesRealArrayIdxType(t *testing.T) { } } +// TestFoundJob_ExplicitIndex (5.4, 5.6, 8.2): an explicit element number +// matches only array elements with that number -- never an ordinary job with +// array_idx 0 or the whole array -- while no element part matches every row +// with that job_id. +func TestFoundJob_ExplicitIndex(t *testing.T) { + const key = uint64(2)<<40 | uint64(1)<<20 + row := func(uid uint64, idx int) utils.OrderMap { + return store.StructToOrderMap(models.JobInfo{Unique_JobID: int64(uid), JobID: int64(key), ArrayIdx: idx}) + } + ordinary := row(key, 0) // an ordinary job sharing the number + e1 := row(backend.MakeElementID(key, 1), 1) + e2 := row(backend.MakeElementID(key, 2), 2) + withZero := []utils.OrderMap{ordinary, e1, e2, row(backend.MakeElementID(key, 0), 0)} + noZero := []utils.OrderMap{ordinary, e1, e2} + + uids := func(rows []utils.OrderMap) []int64 { + out := []int64{} + for _, r := range rows { + out = append(out, r.Data["unique_job_id"].(int64)) + } + return out + } + + if got, err := foundJob(int64(key), 0, true, noZero); err == nil { + t.Errorf("[0] without element 0 matched %v, want not found", uids(got)) + } + got, err := foundJob(int64(key), 0, true, withZero) + if err != nil || len(got) != 1 || uint64(got[0].Data["unique_job_id"].(int64)) != backend.MakeElementID(key, 0) { + t.Errorf("[0] with a pre-upgrade element 0 = %v, %v; want that element alone", uids(got), err) + } + got, err = foundJob(int64(key), 2, true, noZero) + if err != nil || len(got) != 1 || uint64(got[0].Data["unique_job_id"].(int64)) != backend.MakeElementID(key, 2) { + t.Errorf("[2] = %v, %v; want element 2 alone", uids(got), err) + } + if got, err := foundJob(int64(key), 3, true, noZero); err == nil { + t.Errorf("[3] matched %v, want not found", uids(got)) + } + got, err = foundJob(int64(key), 0, false, noZero) + if err != nil || len(got) != 3 { + t.Errorf("bare = %v, %v; want every row with that job_id", uids(got), err) + } +} + // TestPreCheckJobRequest_RejectsMalformedResreq (bug-1744): a resource request // the syntax check rejects must fail the submission instead of being accepted // and later silently dropped by the scheduler.