From 1dbfec800342e38ab1dcd44c542681b7c8fab53f Mon Sep 17 00:00:00 2001 From: pelikhan Date: Sun, 8 Nov 2026 23:06:40 -0800 Subject: [PATCH] Add work cancel obsolete operator with read-only preview Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> --- .../patch-cli-obsolete-work-cancellation.md | 5 + docs/src/content/docs/reference/work-queue.md | 16 ++ .../docs/specs/work-queue-specification.md | 9 + pkg/cli/work_command_obsolete.go | 123 +++++++++ pkg/cli/work_command_obsolete_test.go | 251 ++++++++++++++++++ pkg/cli/work_command_operations.go | 38 ++- pkg/workqueue/deployment.go | 25 ++ pkg/workqueue/obsolete_selection_test.go | 122 +++++++++ 8 files changed, 588 insertions(+), 1 deletion(-) create mode 100644 .changeset/patch-cli-obsolete-work-cancellation.md create mode 100644 pkg/cli/work_command_obsolete.go create mode 100644 pkg/cli/work_command_obsolete_test.go create mode 100644 pkg/workqueue/obsolete_selection_test.go diff --git a/.changeset/patch-cli-obsolete-work-cancellation.md b/.changeset/patch-cli-obsolete-work-cancellation.md new file mode 100644 index 00000000000..2490150832f --- /dev/null +++ b/.changeset/patch-cli-obsolete-work-cancellation.md @@ -0,0 +1,5 @@ +--- +"gh-aw": patch +--- + +Add `gh aw work cancel --obsolete [--limit N] [--dry-run]` using the JavaScript maintenance selection and checked cancellation semantics. Preserve execution pins, ownership, delivery barriers, native reservations and queue history; recover original targets on stable-request retries. diff --git a/docs/src/content/docs/reference/work-queue.md b/docs/src/content/docs/reference/work-queue.md index eee2312c37e..553b5473a93 100644 --- a/docs/src/content/docs/reference/work-queue.md +++ b/docs/src/content/docs/reference/work-queue.md @@ -241,6 +241,7 @@ Repository selection is `--repo`, then `GH_REPO`, then local `origin`, then `gh | `deploy --from-config` | Refresh compatible worker routes; optional `--pool`, `--worker-profile`. | | `policy --from-config --epoch EPOCH` | Update existing, quiescent Policy; `--file` is the mutually exclusive alternative. | | `control`, `cancel-work` | Apply authorized pause, cutover, or Work cancellation. | +| `cancel --obsolete [--limit N] [--dry-run]` | Select one checked obsolete-contract batch; `cancel` is an alias of `cancel-work`. | | `cancel-claim --claim-id ID[,ID...] --reason CODE` | Cancel exact current Claims and their Work, not native runs. | | `reprioritize --work-id ID[,ID...] --priority 1..5 --reason CODE` | Administrator-only priority override for future grants of available Work. | | `clear --reason CODE [--dry-run] [--hard]` | Interactive local cancellation or settled-queue reset; see below. | @@ -265,6 +266,21 @@ All timestamps are Unix milliseconds. Inspection reads do not refresh external d Metadata diagnostics exclude payloads, Result descriptors, request parameters, and receipt bodies. Retired pools remain visible with `policy_installed: false`; unknown admission ages are counted rather than inferred from client timestamps. +### Cancel obsolete worker-contract Work + +```bash +gh aw work --repo OWNER/REPO cancel --obsolete \ + --limit 256 --request-id obsolete-batch-1 --dry-run --json +``` + +`cancel` aliases `cancel-work`. `--obsolete` uses the JavaScript maintenance selection: only available, unclaimed, unpinned Work with barrier `none` and an admission contract incompatible with an available approved active Deployment is eligible. Expired Work, compatible Work, explicit execution pins, and missing/unavailable deployments are excluded. IDs are sorted in UTF-8 byte order; one batch contains at most `--limit` Work (default and maximum 256), further bounded by the installed Policy operation limit. It does not refresh routes or restore missing deployments. + +`--dry-run` reads and validates the ledger and reports the selection without checking administrator publication authority, recording a request, synchronizing Issues, or causing queue effects. Remove it to publish an administrator-authorized atomic `cancel_work` request with fixed reason `worker_contract_obsolete` and operation `version: 1`. Fresh publication checks reject the whole batch if a selected Work becomes owned, expired, compatible, or otherwise ineligible. Claims, Results, delivery barriers, reservations, policy/epoch/limits, fairness debt, history, and unselected Work are preserved; no native runs are stopped. + +`--work-id` and `--reason` cannot be combined with `--obsolete`; `--limit` and `--dry-run` require it. Use the same stable `--request-id` after an uncertain publication. An accepted request recovers only its original cancellation targets and fingerprint, including historical legacy receipts; it never sweeps newly obsolete Work. A smaller limit that cannot hold the original receipt is rejected. To select another batch, use a new request ID. No eligible Work returns `no_work` without recording a request. + +JSON output includes `status`, `request_id`, `snapshot_tip`, and metadata-only `work_ids`; authenticated publication additionally includes `publication`. Status is `dry_run`, `no_work`, `cancelled`, or `already_committed`. A preview does not guarantee later publication authority or eligibility. Inspect `explain --request-id ID` after an uncertain failure before retrying that same ID; a committed cancellation with pending Issue synchronization requires `sync-issues --request-id ID`, not another cancellation. + ### Clear unfinished Work ```bash diff --git a/docs/src/content/docs/specs/work-queue-specification.md b/docs/src/content/docs/specs/work-queue-specification.md index cbdf9e83796..f468d688a55 100644 --- a/docs/src/content/docs/specs/work-queue-specification.md +++ b/docs/src/content/docs/specs/work-queue-specification.md @@ -1021,6 +1021,15 @@ request MUST recover its original request and fingerprint, not publish a checked replacement. All existing authorization, ownership and native-reservation requirements still apply to both forms. +Native `work cancel --obsolete` MUST use the maintenance eligibility predicate, +current-time expiration, UTF-8 byte ordering, and a single batch bounded by 256 +and the installed operation limit. An optional smaller `--limit` only narrows +that bound. Publication uses the existing native administrator authority, not +an impersonated workflow reconciler. `--dry-run` MUST remain read-only and MUST +NOT assert authenticated publication authority. Retrying an accepted request +MUST recover its original targets and fingerprint rather than reselect a batch +from newly obsolete Work. No eligible targets MUST NOT record a request. + `WorkPriority` is published through an administrator `control` request in this same transaction log. Replay stores `effective_priority` separately from the immutable Work definition; native Go and JavaScript selection and historical diff --git a/pkg/cli/work_command_obsolete.go b/pkg/cli/work_command_obsolete.go new file mode 100644 index 00000000000..eb8b18ad9f7 --- /dev/null +++ b/pkg/cli/work_command_obsolete.go @@ -0,0 +1,123 @@ +package cli + +import ( + "encoding/json" + "errors" + "fmt" + "time" + + "github.com/github/gh-aw/pkg/workqueue" + "github.com/spf13/cobra" +) + +const workObsoleteReason = "worker_contract_obsolete" + +type workObsoleteResult struct { + Status string `json:"status"` + RequestID string `json:"request_id"` + SnapshotTip string `json:"snapshot_tip"` + WorkIDs []string `json:"work_ids"` + Publication *workqueue.Publication `json:"publication,omitempty"` +} + +func workObsoleteIntent(state workqueue.Projection, at int64, limit int, id string) (workqueue.OperationsParameters, []string, error) { + ids, err := workqueue.ObsoleteWorkIDs(state, at, limit) + if err != nil { + return workqueue.OperationsParameters{}, nil, err + } + if accepted, ok := state.Requests[id]; ok { + params, err := workRecoverClearIntent(accepted, workObsoleteReason) + if err != nil { + return params, nil, err + } + if len(params.Operations) > min(limit, state.Policy.Limits.Operations) { + return params, nil, errors.New("obsolete_limit: original cancellation exceeds this batch limit; retry with its original limit") + } + ids = []string{} + for _, operation := range params.Operations { + var fields struct { + WorkID string `json:"work_id"` + ClaimID string `json:"claim_id"` + } + if err := json.Unmarshal(operation, &fields); err != nil { + return params, nil, err + } + if fields.ClaimID != "" { + return params, nil, errors.New("request_reuse: accepted cancellation selected Claims; retry its original operator command") + } + ids = append(ids, fields.WorkID) + } + return params, ids, nil + } + targets := make([]workOperatorTarget, 0, len(ids)) + for _, id := range ids { + targets = append(targets, workOperatorTarget{WorkID: id}) + } + params, _, err := workOperatorIntent(state, targets, workObsoleteReason, 0, id) + return params, ids, err +} + +func workRunObsoleteCancellation(cmd *cobra.Command) error { + limit, _ := cmd.Flags().GetInt("limit") + if limit < 1 || limit > 256 { + return errors.New("obsolete_limit: --limit must be 1..256") + } + branch := workBranch(cmd) + commits, err := branch.Read(cmd.Context()) + if err != nil { + return err + } + if len(commits) == 0 { + return errors.New("queue_missing: obsolete cancellation requires an installed queue") + } + at := time.Now().UnixMilli() + state, err := workqueue.ReplayAt(commits, at) + if err != nil { + return err + } + id, err := workRequestID(cmd) + if err != nil { + return err + } + params, ids, err := workObsoleteIntent(state, at, limit, id) + if err != nil { + return err + } + result := workObsoleteResult{Status: "no_work", RequestID: id, SnapshotTip: state.Tip, WorkIDs: ids} + message := "No eligible obsolete Work; no request recorded" + if len(ids) == 0 { + return workPrint(cmd, result, message) + } + dryRun, _ := cmd.Flags().GetBool("dry-run") + _, accepted := state.Requests[id] + if dryRun { + result.Status = "dry_run" + message = fmt.Sprintf("Would cancel %d obsolete Work; read-only selection, publication authority not checked", len(ids)) + if accepted { + result.Status = "already_committed" + message = fmt.Sprintf("Original cancellation of %d Work already committed; read-only receipt recovery", len(ids)) + } + } else { + return workPublishObsoleteCancellation(cmd, params, result) + } + return workPrint(cmd, result, message+"; native runs, reservations, policy, debt and history retained; request="+workQueueText(id)) +} + +func workPublishObsoleteCancellation(cmd *cobra.Command, params workqueue.OperationsParameters, result workObsoleteResult) error { + id := result.RequestID + published, err := workPublishOperator(cmd, id, "cancel_work", params) + if err != nil { + return fmt.Errorf("obsolete cancellation failed; inspect the ledger with explain --request-id %s before retrying the same request: %w", workQueueText(id), err) + } + result.Publication = &published + result.Status = "cancelled" + message := fmt.Sprintf("Cancelled %d obsolete Work", len(result.WorkIDs)) + if !published.Changed { + result.Status = "already_committed" + message = fmt.Sprintf("Confirmed original cancellation of %d Work", len(result.WorkIDs)) + } + if _, err := workBranch(cmd).SynchronizeIssueTransition(cmd.Context(), id); err != nil { + return fmt.Errorf("queue cancellation committed; Issue synchronization pending; retry sync-issues --request-id %s: %w", workQueueText(id), err) + } + return workPrint(cmd, result, message+"; native runs, reservations, policy, debt and history retained; request="+workQueueText(id)) +} diff --git a/pkg/cli/work_command_obsolete_test.go b/pkg/cli/work_command_obsolete_test.go new file mode 100644 index 00000000000..dd273e62ad3 --- /dev/null +++ b/pkg/cli/work_command_obsolete_test.go @@ -0,0 +1,251 @@ +package cli + +import ( + "bytes" + "encoding/json" + "io" + "net/http" + "slices" + "strings" + "testing" + "time" + + "github.com/github/gh-aw/pkg/workqueue" + "github.com/spf13/cobra" +) + +func workObsoleteFixture(t *testing.T) ([]workqueue.QueueCommit, []string) { + t.Helper() + actor := workqueue.Actor{Role: "administrator", Principal: "1001", Repository: "owner/repo"} + policy := workqueue.DefaultPolicy("", actor.Repository) + policy.Authorization = "aw" + policy.Producers = map[string]workqueue.ProducerRule{} + pool := policy.Pools["default"] + profile := pool.Profiles["default"] + profile.Ref, profile.LogicalContract = strings.Repeat("a", 40), strings.Repeat("a", 64) + pool.Profiles["default"] = profile + other := profile + other.Workflow = ".github/workflows/other.lock.yml" + pool.Profiles["other"] = other + policy.Pools["default"] = pool + genesis, err := workqueue.Genesis(actor, policy, "installed", "genesis", 1000) + if err != nil { + t.Fatal(err) + } + nodes := []workqueue.WorkDefinition{} + ids := []string{} + for _, key := range []string{"old-one", "old-two", "pinned", "compatible", "expired"} { + work, err := workqueue.NewWork([]byte(`{"task":"PAYLOAD-MUST-STAY-PRIVATE"}`), "backlog", key, "default", policy, 2000) + if err != nil { + t.Fatal(err) + } + switch key { + case "pinned": + work.ExecutionRef = profile.Ref + case "compatible": + work.WorkerProfile = "other" + case "expired": + deadline := int64(9000) + work.ExpiresAt = &deadline + } + nodes, ids = append(nodes, work), append(ids, work.WorkID) + } + commits := workClearAppend(t, []workqueue.QueueCommit{genesis}, actor, "backlog", "submit", workqueue.SubmitParameters{Nodes: nodes}, 2000) + profile.Ref, profile.LogicalContract = strings.Repeat("b", 40), strings.Repeat("b", 64) + operation := workqueue.DeploymentOperation{ + Kind: "Deployment", Pool: "default", WorkerProfile: "default", + ExpectedRef: strings.Repeat("a", 40), ExpectedContract: strings.Repeat("a", 64), + Profile: profile, Available: true, Reason: "compiler_deployment", + } + commits = workClearAppend(t, commits, actor, "deployment", "deployment", workqueue.OperationsParameters{ + Operations: []workqueue.Operation{mustWorkQueueOperation(t, operation)}, + }, 3000) + return commits, ids +} + +func workObsoleteExecute(t *testing.T, extra ...string) (workObsoleteResult, error) { + t.Helper() + cmd := NewWorkCommand() + parent := &cobra.Command{Use: "gh-aw"} + parent.AddCommand(cmd) + var output bytes.Buffer + parent.SetOut(&output) + parent.SetErr(io.Discard) + parent.SetArgs(append([]string{"work", "--repo", "owner/repo", "--json", "cancel", "--obsolete", "--request-id", "obsolete-batch"}, extra...)) + err := parent.Execute() + var result workObsoleteResult + if err == nil { + if err := json.Unmarshal(output.Bytes(), &result); err != nil { + t.Fatalf("invalid obsolete output: %v (%s)", err, output.String()) + } + if strings.Contains(output.String(), "PAYLOAD-MUST") { + t.Fatal("obsolete command leaked payload") + } + } + return result, err +} + +func TestWorkCancelObsoleteDryRunAndPublication(t *testing.T) { + workClearRoleEnvironment(t, "admin") + commits, ids := workObsoleteFixture(t) + data, err := workqueue.Serialize(commits) + if err != nil { + t.Fatal(err) + } + queue := newWorkQueueGitRemote(t, string(data)) + initial := queue.head(t) + transport := http.DefaultTransport + operatorReads := 0 + http.DefaultTransport = workAPIRoundTripper(func(r *http.Request) (*http.Response, error) { + operatorReads++ + return transport.RoundTrip(r) + }) + preview, err := workObsoleteExecute(t, "--dry-run", "--limit", "1") + if err != nil || preview.Status != "dry_run" || len(preview.WorkIDs) != 1 || preview.Publication != nil { + t.Fatalf("obsolete preview failed: %+v %v", preview, err) + } + expected := slices.Clone(ids[:2]) + slices.Sort(expected) + if preview.WorkIDs[0] != expected[0] || queue.head(t) != initial || queue.pushes(t) != 0 || operatorReads != 0 { + t.Fatal("preview changed the ledger, authenticated operator authority, or selected outside the sorted bound") + } + result, err := workObsoleteExecute(t, "--limit", "1") + if err != nil || result.Status != "cancelled" || result.Publication == nil || !result.Publication.Changed { + t.Fatalf("obsolete publication failed: %+v %v", result, err) + } + next, err := workqueue.Parse([]byte(queue.read(t))) + if err != nil { + t.Fatal(err) + } + before, _ := workqueue.ReplayAt(commits, time.Now().UnixMilli()) + after, err := workqueue.Replay(next) + if err != nil { + t.Fatal(err) + } + selected := after.Works[expected[0]] + if selected.State != "cancelled" || selected.CancellationReason != workObsoleteReason || queue.pushes(t) != 1 { + t.Fatal("checked cancellation was not durable") + } + for _, id := range ids { + if id != expected[0] && !workObsoleteSameJSON(before.Works[id], after.Works[id]) { + t.Fatalf("unselected or protected Work changed: %s", id) + } + } + if !workObsoleteSameJSON(before.Policy, after.Policy) || before.PolicyEpoch != after.PolicyEpoch || + !workObsoleteSameJSON(before.Clocks, after.Clocks) || !workObsoleteSameJSON(before.Claims, after.Claims) || + !workObsoleteSameJSON(before.Dispatches, after.Dispatches) || !workObsoleteSameJSON(before.Deployments, after.Deployments) { + t.Fatal("retirement changed authority, reservations or economics") + } + head := queue.head(t) + for _, args := range [][]string{{"--limit", "1"}, {"--limit", "1", "--dry-run"}} { + retry, err := workObsoleteExecute(t, args...) + if err != nil || retry.Status != "already_committed" || !slices.Equal(retry.WorkIDs, result.WorkIDs) || + queue.head(t) != head || queue.pushes(t) != 1 { + t.Fatalf("retry swept newly eligible Work or repeated publication: %+v %v", retry, err) + } + } +} + +func workObsoleteSameJSON(a, b any) bool { + left, leftErr := json.Marshal(a) + right, rightErr := json.Marshal(b) + return leftErr == nil && rightErr == nil && bytes.Equal(left, right) +} + +func TestWorkCancelObsoleteAuthorityAndFreshPublication(t *testing.T) { + for _, scenario := range []string{"writer", "maintainer", "compatible", "unavailable", "claimed"} { + t.Run(scenario, func(t *testing.T) { + role := "admin" + switch scenario { + case "writer": + role = "write" + case "maintainer": + role = "maintain" + } + workClearRoleEnvironment(t, role) + commits, _ := workObsoleteFixture(t) + data, _ := workqueue.Serialize(commits) + queue := newWorkQueueGitRemote(t, string(data)) + transport := http.DefaultTransport + changed := false + http.DefaultTransport = workAPIRoundTripper(func(r *http.Request) (*http.Response, error) { + if r.URL.Path == "/user" && !changed && role == "admin" { + changed = true + state, _ := workqueue.Replay(commits) + actor := state.Requests["deployment"].Actor + if scenario == "claimed" { + worker := workqueue.FuturePolicy(state) + node, err := workqueue.NewWork([]byte(`{}`), "new", "root", "default", worker, 4000) + if err != nil { + t.Fatal(err) + } + commits = workClearAppend(t, commits, actor, "new", "submit", workqueue.SubmitParameters{Nodes: []workqueue.WorkDefinition{node}}, 4000) + } + deployment := state.Deployments["default"]["default"] + profile := deployment.Revisions[deployment.CurrentRef].Profile + available := scenario != "unavailable" + if scenario != "unavailable" { + profile = deployment.Revisions[strings.Repeat("a", 40)].Profile + } + op := workqueue.DeploymentOperation{ + Kind: "Deployment", Pool: "default", WorkerProfile: "default", ExpectedRef: deployment.CurrentRef, + ExpectedContract: deployment.CurrentContract, Profile: profile, Available: available, Reason: "compiler_deployment", + } + commits = workClearAppend(t, commits, actor, "route-race", "deployment", workqueue.OperationsParameters{ + Operations: []workqueue.Operation{mustWorkQueueOperation(t, op)}, + }, 5000) + if scenario == "claimed" { + commits = workClearAppend(t, commits, actor, "grant-race", "dispatch_next", workqueue.DispatchParameters{ + Pool: "default", MaxClaims: 1, MaxDispatches: 1, MaxBytes: 48 << 10, + }, 6000) + } + data, _ := workqueue.Serialize(commits) + queue.write(t, string(data)) + } + return transport.RoundTrip(r) + }) + _, err := workObsoleteExecute(t) + if err == nil || queue.pushes(t) != 0 { + t.Fatalf("unauthorized or stale obsolete batch published: %v", err) + } + code := "deployment_conflict:" + if role != "admin" { + code = "actor_unauthorized:" + } + if !strings.Contains(err.Error(), code) { + t.Fatalf("expected actionable %s, got %v", code, err) + } + }) + } +} + +func TestWorkCancelObsoleteNoWorkAndClosedFlags(t *testing.T) { + workClearRoleEnvironment(t, "write") + state, commits, _ := workOperatorFixture(t, false) + data, _ := workqueue.Serialize(commits) + queue := newWorkQueueGitRemote(t, string(data)) + result, err := workObsoleteExecute(t) + if err != nil || result.Status != "no_work" || len(result.WorkIDs) != 0 || queue.pushes(t) != 0 { + t.Fatalf("legacy compatible queue did not remain a no-op: %+v %v", result, err) + } + for _, extra := range [][]string{ + {"--work-id", "exact"}, {"--reason", workObsoleteReason}, {"--limit", "0"}, {"--limit", "257"}, + {"--all"}, {"--force"}, {"--execution-ref", "pin"}, + } { + if _, err := workObsoleteExecute(t, extra...); err == nil { + t.Fatalf("unsafe obsolete flags accepted: %v", extra) + } + } + for _, args := range [][]string{{"cancel", "--dry-run"}, {"cancel", "--limit", "1"}} { + cmd := NewWorkCommand() + cmd.SetOut(io.Discard) + cmd.SetErr(io.Discard) + cmd.SetArgs(append([]string{"--repo", "owner/repo"}, args...)) + if err := cmd.Execute(); err == nil { + t.Fatalf("obsolete-only flag silently ignored: %v", args) + } + } + if _, _, err := workObsoleteIntent(state, 10000, 256, "submit"); err == nil { + t.Fatal("obsolete request reused a noncancellation identity") + } +} diff --git a/pkg/cli/work_command_operations.go b/pkg/cli/work_command_operations.go index bb8d48f2660..976bc5ab767 100644 --- a/pkg/cli/work_command_operations.go +++ b/pkg/cli/work_command_operations.go @@ -14,7 +14,43 @@ import ( type workOperatorTarget struct{ WorkID, ClaimID string } -func workCancelCommand() *cobra.Command { return workOperatorCommand("cancel-work", false, false) } +func workCancelCommand() *cobra.Command { + cmd := workOperatorCommand("cancel-work", false, false) + cmd.Aliases = []string{"cancel"} + cmd.Long = "Terminally cancel exact Work IDs with --work-id and --reason, or use\n" + + "--obsolete to retire one bounded batch using the JavaScript maintenance semantics.\n\n" + + "--obsolete selects available, unclaimed, unpinned Work without delivery barriers\n" + + "whose admission contract differs from an available approved active deployment.\n" + + "Elapsed Work and missing/unavailable deployments are excluded. IDs are sorted\n" + + "in UTF-8 byte order, then bounded by --limit and the installed operation limit.\n" + + "The fixed reason is worker_contract_obsolete with operation version 1.\n\n" + + "--dry-run is a read-only obsolete selection preview; it neither authenticates\n" + + "operator authority nor reserves the request ID. Publication requires the same\n" + + "repository administrator permission as cancel-work and rechecks the current\n" + + "ledger at committed time; any stale target rejects the entire batch.\n" + + "Use a stable --request-id: retries recover only the original accepted targets,\n" + + "never newly obsolete Work. No eligible Work is a no-op without a receipt.\n" + + "Claims, Results, barriers, native reservations, policy, debt and history are\n" + + "retained; native runs are not stopped and worker routes are not changed." + cmd.Flags().Bool("obsolete", false, "Select one checked batch of obsolete worker-contract Work; excludes --work-id and --reason") + cmd.Flags().Int("limit", 256, "Maximum obsolete Work per batch (1..256, further bounded by Policy)") + cmd.Flags().Bool("dry-run", false, "Preview --obsolete selection without publishing or synchronizing Issues") + cmd.RunE = func(cmd *cobra.Command, _ []string) error { + obsolete, _ := cmd.Flags().GetBool("obsolete") + if obsolete { + if cmd.Flags().Changed("work-id") || cmd.Flags().Changed("reason") { + return errors.New("obsolete_selector_invalid: --obsolete cannot be combined with --work-id or --reason") + } + return workRunObsoleteCancellation(cmd) + } + if cmd.Flags().Changed("limit") || cmd.Flags().Changed("dry-run") { + return errors.New("obsolete_selector_invalid: --limit and --dry-run require --obsolete") + } + return workRunOperator(cmd, "work-id", false, false) + } + return cmd +} + func workCancelClaimCommand() *cobra.Command { return workOperatorCommand("cancel-claim", true, false) } func workPriorityCommand() *cobra.Command { return workOperatorCommand("reprioritize", false, true) } diff --git a/pkg/workqueue/deployment.go b/pkg/workqueue/deployment.go index dabded8ee24..345c8363872 100644 --- a/pkg/workqueue/deployment.go +++ b/pkg/workqueue/deployment.go @@ -144,6 +144,31 @@ func (state Projection) obsoleteWork(work *WorkState) bool { return reason == "worker_incompatible" } +// ObsoleteWorkIDs selects the same bounded, byte-ordered retirement batch as +// JavaScript maintenance. Publication must still validate each checked operation. +func ObsoleteWorkIDs(state Projection, at int64, limit int) ([]string, error) { + ids := []string{} + if at < 0 || at > MaxTimestamp { + return nil, queueError("timestamp_invalid", "invalid obsolete selection timestamp") + } + if limit < 1 || limit > 256 { + return nil, queueError("obsolete_limit", "obsolete cancellation limit must be 1..256") + } + if state.Policy == nil { + return nil, queueError("queue_missing", "obsolete cancellation requires an installed queue") + } + if state.Policy.Authorization != "aw" { + return ids, nil + } + for id, work := range state.Works { + if state.obsoleteWork(work) && !workExpired(work, max(at, state.timeWatermark)) { + ids = append(ids, id) + } + } + slices.Sort(ids) + return ids[:min(len(ids), limit, state.Policy.Limits.Operations)], nil +} + // FuturePolicy exposes current routes for trusted admission builders only. // It does not replace the installed policy or affect existing assignments. func FuturePolicy(state Projection) Policy { diff --git a/pkg/workqueue/obsolete_selection_test.go b/pkg/workqueue/obsolete_selection_test.go new file mode 100644 index 00000000000..3324ffdec30 --- /dev/null +++ b/pkg/workqueue/obsolete_selection_test.go @@ -0,0 +1,122 @@ +package workqueue + +import ( + "bytes" + "fmt" + "slices" + "strings" + "testing" +) + +func TestObsoleteWorkSelectionSpecification(t *testing.T) { + base, nodes := cancellationVersionFixture(t) + base = deployTestWorker(t, base, "incompatible", "default", strings.Repeat("b", 40), strings.Repeat("b", 64), true) + for _, name := range []string{ + "eligible", "claimed", "owner", "pin", "pending", "failed", "verified", "completed", "cancelled", + "missing", "missing-revision", "unavailable", "compatible", "legacy", "expired", "before-deadline", + "byte-order", "policy-limit", "caller-limit", + } { + state, err := Replay(base) + if err != nil { + t.Fatal(err) + } + work := state.Works[nodes[0].WorkID] + deployment := state.Deployments["default"]["default"] + revision := deployment.Revisions[deployment.CurrentRef] + limit := 256 + switch name { + case "claimed", "completed", "cancelled": + work.State = name + case "owner": + work.ClaimID = "owned" + case "pin": + work.ExecutionRef = nodes[1].ExecutionRef + case "pending", "failed", "verified": + work.Barrier = name + case "missing": + delete(state.Deployments["default"], "default") + case "missing-revision": + delete(deployment.Revisions, deployment.CurrentRef) + case "unavailable": + revision.Available = false + deployment.Revisions[deployment.CurrentRef] = revision + case "compatible": + revision.Profile.LogicalContract = work.AdmissionContract + deployment.Revisions[deployment.CurrentRef] = revision + case "legacy": + state.Policy.Authorization = "legacy" + case "expired", "before-deadline": + deadline := int64(9000) + if name == "before-deadline" { + deadline++ + } + work.ExpiresAt = &deadline + case "byte-order", "policy-limit", "caller-limit": + delete(state.Works, work.WorkID) + for i := range 260 { + copy := *work + copy.WorkID = fmt.Sprintf("work-%03d", 259-i) + state.Works[copy.WorkID] = © + } + switch name { + case "policy-limit": + state.Policy.Limits.Operations = 2 + case "caller-limit": + limit = 1 + } + } + before, _ := canonicalValue(state) + ids, err := ObsoleteWorkIDs(state, 9000, limit) + if err != nil { + t.Fatal(err) + } + after, _ := canonicalValue(state) + if !bytes.Equal(before, after) { + t.Fatal("selection mutated the authoritative projection") + } + expectedCount := 0 + switch name { + case "eligible", "before-deadline", "caller-limit": + expectedCount = 1 + case "policy-limit": + expectedCount = 2 + case "byte-order": + expectedCount = 256 + } + if len(ids) != expectedCount || !slices.IsSorted(ids) { + t.Fatalf("%s selection violates eligibility, ordering or limits: %v", name, ids) + } + if name == "eligible" || name == "before-deadline" { + if !slices.Equal(ids, []string{nodes[0].WorkID}) { + t.Fatalf("%s selected a compatible or pinned Work: %v", name, ids) + } + } + if name == "byte-order" && (ids[0] != "work-000" || ids[255] != "work-255") { + t.Fatalf("batch did not take the exact byte-ordered prefix: %v", ids) + } + if name == "policy-limit" && !slices.Equal(ids, []string{"work-000", "work-001"}) || + name == "caller-limit" && !slices.Equal(ids, []string{"work-000"}) { + t.Fatalf("%s did not take the exact bounded prefix: %v", name, ids) + } + } +} + +func TestObsoleteWorkSelectionRequiresValidBounds(t *testing.T) { + state, err := Replay(testGenesis(t, deploymentPolicy)) + if err != nil { + t.Fatal(err) + } + for _, limit := range []int{0, -1, 257} { + if _, err := ObsoleteWorkIDs(state, 1000, limit); err == nil { + t.Fatalf("invalid selection bound accepted: %d", limit) + } + } + for _, at := range []int64{-1, MaxTimestamp + 1} { + if _, err := ObsoleteWorkIDs(state, at, 256); err == nil { + t.Fatalf("invalid selection time accepted: %d", at) + } + } + if _, err := ObsoleteWorkIDs(Projection{}, 1000, 256); err == nil { + t.Fatal("absent queue silently treated as an empty retirement batch") + } +}