From 10b6e6c3d9593ea08b07ced1e4730c12d72bf2f8 Mon Sep 17 00:00:00 2001 From: Wallace Date: Fri, 17 Jul 2026 16:46:48 +0100 Subject: [PATCH] feat(server): cancel a running job by id A CLI-submitted job could not be stopped once launched: the work runs as a fiber forked into the server's layer scope, so Ctrl-C in the CLI only detached the log follow while the server-side job ran to completion. Unlike an app run directly (IntelliJ Stop = root-fiber interrupt), there was no channel to reach that fiber. This adds cancellation by id. JobRunner now retains a per-process handle to each job fiber (`runningFibers`, a Promise registered *before* the fork so `release`'s de-register can't race ahead of the register). `cancel(id)` registers the id in `cancelRequested`, then `interruptFork`s the fiber so the HTTP handler returns promptly. The job's own `.onInterrupt` records the new `Cancelled` terminal state -- but ONLY if its id is in `cancelRequested`. That gate is load-bearing: `layerScope` interrupts every in-flight job on server shutdown too, and those must stay `Running` so the next boot's `markOrphansAsFailed` records them as `Failed`/"Service restarted", not a spurious "Cancelled by operator". The interrupt-observing write also fixes a latent bug where an interrupted job bypassed `foldZIO` and left its row stuck at `Running`. Cancellation is best-effort/async: an in-flight uninterruptible blocking JDBC statement runs to completion before the interrupt lands, and `markCancelled`'s `WHERE status = Running` guard makes the terminal write a no-op if the job reached a terminal state first. `cancel` is per-process, so it inherits the single-server-per-DB assumption (#110/#111); an absent local fiber returns 404 rather than blindly marking a row that may be Running elsewhere. Surface: `POST /api/jobs/{jobId}/cancel` (200 CancelResult / 404) and the `ccas cancel ` subcommand. A job cancelled out from under an active `ccas logs` follow renders "was cancelled" and exits non-zero. Also unify all `completed_at` writers (`updateStatus`, `markCancelled`, `markOrphansAsFailed`) on a caller-supplied `Clock.instant` instead of SQL `NOW()`, so every job timestamp comes from the one testable app clock and is coherent with `started_at` (no app-vs-DB host skew). Tested: cancel interrupts a live job -> Cancelled (real JobRunner.live); the shutdown-style self-interrupt stays Running; markCancelled guard; the cancel route 200/404; CLI parse + completion drift; follow exit code; and an end-to-end wire smoke (TestJobCancelWire) driving the real CcasApiClient over a real HTTP server against real JobRoutes + JobRunner.live. Fixes #168 Co-authored-by: Claude Opus 4.8 (1M context) --- CLAUDE.md | 4 +- README.md | 7 +- completions/ccas.bash | 3 +- src/main/scala/ccas/cli/CliCommand.scala | 7 ++ src/main/scala/ccas/cli/CompletionSpec.scala | 4 +- src/main/scala/ccas/cli/Dispatcher.scala | 7 ++ src/main/scala/ccas/cli/JobFollower.scala | 3 + src/main/scala/ccas/server/jobs/JobRun.scala | 18 +++- .../scala/ccas/server/jobs/JobRunStatus.scala | 2 +- .../scala/ccas/server/jobs/JobRunner.scala | 84 +++++++++++++++-- .../scala/ccas/server/routes/JobRoutes.scala | 20 +++++ .../scala/ccas/cli/TestCcasCompletion.scala | 1 + src/test/scala/ccas/cli/TestCliParser.scala | 5 ++ src/test/scala/ccas/cli/TestJobFollower.scala | 6 ++ .../ccas/server/jobs/TestJobRunSql.scala | 37 +++++++- .../ccas/server/jobs/TestJobRunner.scala | 44 +++++++++ .../server/routes/TestJobCancelWire.scala | 89 +++++++++++++++++++ .../scala/ccas/server/routes/TestRoutes.scala | 45 ++++++++++ .../server/scheduler/TestJobScheduler.scala | 8 +- 19 files changed, 377 insertions(+), 17 deletions(-) create mode 100644 src/test/scala/ccas/server/routes/TestJobCancelWire.scala diff --git a/CLAUDE.md b/CLAUDE.md index c4692cbc..8d5efa74 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -85,8 +85,8 @@ Each variant carries a lazy `getValue: Task[T]` so callers can branch on `isUnch ### Backend Server `CcasServer` (`ccas.server`) is a zio-http server that exposes REST endpoints and runs background jobs: -- **Routes:** `HealthRoutes` (health/readiness), `JobRoutes` (submit and query jobs, plus recruitment-result delivery: `GET /api/jobs/{jobId}/recruitment/invited` for a completed run's paste-ready invited usernames, `GET .../recruitment/found` for still-`Deferred` candidates, the mutating `POST .../recruitment/confirm` which flips that run's `Deferred` candidates to `Invited` in one transaction and returns `ConfirmResult{marked, usernames}`, and `GET /api/recruitment/clubs/{slug}/latest/invited` + `GET /api/recruitment/runs/{runId}/invited` for reporting a past run — these drive the `ccas recruit` `--stdout` / interactive-confirm / `--report` delivery modes, where an interactive scout sends `autoConfirm=false` so candidates stay `Deferred` until the operator confirms), `ScheduleRoutes` (CRUD for scheduled jobs), `BlacklistRoutes` (synchronous CRUD for `RecruitmentBlacklist` entries — delegates to `BlacklistApp` directly without going through `JobRunner`), `RecruitmentCriteriaRoutes` (synchronous `POST /api/recruitment-criteria` to set criteria, `GET .../{slug}/{alias}` to show, `GET .../{slug}` to list aliases — delegates to `RecruitmentCriteriaApp`). Route handlers use inline JSON codecs for request/response types. User-facing errors (`BadRequestException`, `NotFoundException`, `ConflictException`) form a sealed `UserFacingError` trait (`ccas.utils.errors`) that carries the HTTP status and a pre-encoded JSON body; `RouteHelpers.withErrorHandling` renders them uniformly (default body shape is `{"error": ""}`, but `.of[B: JsonEncoder](body, msg)` companion constructors support structured payloads). Escaping `HttpStatusException`s from the Chess.com client render as 502. Anything else — including defects — collapses to a generic 500 with the full cause logged via `ZIO.logErrorCause` so operational visibility isn't lost; pure-interrupt causes are re-propagated so shutdown / client-disconnect noise stays out of the error log. -- **JobRunner:** Trait-based (`JobRunnerLive`) async executor that forks fibers per job, tracks state in `JobRun` table (ULID IDs via `ulid-creator`). Marks orphaned running jobs as failed on startup. The `submit` method takes an `Option[JobRunId] => RIO[..., Any]` effect function, passing the generated job run ID so analysis apps can link their run records back via `job_run_id`. Job kinds: `Recruitment`, `Membership`, `MatchRef`, `History`, `Stats`, `ClubData` (the latter has no HTTP route — scheduler only). +- **Routes:** `HealthRoutes` (health/readiness), `JobRoutes` (submit, query, and cancel jobs — `POST /api/jobs/{jobId}/cancel` interrupts a running job's fiber in this process (200 `CancelResult`, or 404 when no live job fiber exists here: unknown id, already-terminal, or owned by another instance), plus recruitment-result delivery: `GET /api/jobs/{jobId}/recruitment/invited` for a completed run's paste-ready invited usernames, `GET .../recruitment/found` for still-`Deferred` candidates, the mutating `POST .../recruitment/confirm` which flips that run's `Deferred` candidates to `Invited` in one transaction and returns `ConfirmResult{marked, usernames}`, and `GET /api/recruitment/clubs/{slug}/latest/invited` + `GET /api/recruitment/runs/{runId}/invited` for reporting a past run — these drive the `ccas recruit` `--stdout` / interactive-confirm / `--report` delivery modes, where an interactive scout sends `autoConfirm=false` so candidates stay `Deferred` until the operator confirms), `ScheduleRoutes` (CRUD for scheduled jobs), `BlacklistRoutes` (synchronous CRUD for `RecruitmentBlacklist` entries — delegates to `BlacklistApp` directly without going through `JobRunner`), `RecruitmentCriteriaRoutes` (synchronous `POST /api/recruitment-criteria` to set criteria, `GET .../{slug}/{alias}` to show, `GET .../{slug}` to list aliases — delegates to `RecruitmentCriteriaApp`). Route handlers use inline JSON codecs for request/response types. User-facing errors (`BadRequestException`, `NotFoundException`, `ConflictException`) form a sealed `UserFacingError` trait (`ccas.utils.errors`) that carries the HTTP status and a pre-encoded JSON body; `RouteHelpers.withErrorHandling` renders them uniformly (default body shape is `{"error": ""}`, but `.of[B: JsonEncoder](body, msg)` companion constructors support structured payloads). Escaping `HttpStatusException`s from the Chess.com client render as 502. Anything else — including defects — collapses to a generic 500 with the full cause logged via `ZIO.logErrorCause` so operational visibility isn't lost; pure-interrupt causes are re-propagated so shutdown / client-disconnect noise stays out of the error log. +- **JobRunner:** Trait-based (`JobRunnerLive`) async executor that forks fibers per job, tracks state in `JobRun` table (ULID IDs via `ulid-creator`). Marks orphaned running jobs as failed on startup. The `submit` method takes an `Option[JobRunId] => RIO[..., Any]` effect function, passing the generated job run ID so analysis apps can link their run records back via `job_run_id`. Job kinds: `Recruitment`, `Membership`, `MatchRef`, `History`, `Stats`, `ClubData` (the latter has no HTTP route — scheduler only). `JobRunStatus` is `Running | Completed | Failed | Cancelled`. `cancel(id)` interrupts the job's forked fiber (handles retained per-process in a `runningFibers` map registered before the fork so de-register can't race the register): it registers the id in a `cancelRequested` set then `interruptFork`s the fiber, and the job's own `.onInterrupt` records `Cancelled` **only if** its id is in that set. This gates operator cancellation apart from the `layerScope` interrupt fired at every in-flight job on server **shutdown** — those are left `Running` so the next boot's `markOrphansAsFailed` records them as `Failed`/"Service restarted" (which also still covers hard crashes / SIGKILL, where finalizers never run). Cancellation is best-effort/asynchronous — an in-flight uninterruptible blocking JDBC statement runs to completion before the interrupt lands (`markCancelled`'s `WHERE status = Running` guard makes the terminal write a no-op if the job reached Completed/Failed first). All `completed_at` writers (`updateStatus`, `markCancelled`, `markOrphansAsFailed`) take an `Instant` from the caller's `Clock.instant` rather than SQL `NOW()`, so every job timestamp comes from the one testable app clock, coherent with `started_at`. Being per-process, `cancel` inherits the single-server-per-DB assumption (#110/#111). The CLI surface is `ccas cancel `; a job cancelled out from under an active `ccas logs` follow renders "was cancelled" and exits non-zero. - **JobScheduler:** Polling daemon (configurable interval) that checks enabled `JobSchedule` entries and submits due jobs to the `JobRunner`. - **ServerTables:** Ensures both analysis and server database tables exist on startup. diff --git a/README.md b/README.md index 84563b10..cbdeb99f 100644 --- a/README.md +++ b/README.md @@ -57,6 +57,7 @@ sbt "runMain ccas.cli.Main server up" # boots CcasServer on 127.0.0.1:8080 # Terminal 2 — commands. No environment needed; they just call the server. ccas use-club team-alpha # set the current club once… ccas jobs +ccas cancel # request cancellation of a running job ccas membership # …then commands target it with no slug ccas membership --club team-beta # override for one command ccas membership --all # every managed club @@ -66,7 +67,9 @@ ccas --help # full command tree ccas --help # per-command flags ``` -Commands: `server {up|down|status}`, `use`, `membership`, `history`, `recruit`, `stats`, `jobs`, `logs`, `blacklist {add|list|remove}`, `schedule {list|add|remove}`, `club {add|remove|list}`. +Commands: `server {up|down|status}`, `use`, `membership`, `history`, `recruit`, `stats`, `jobs`, `logs`, `cancel`, `blacklist {add|list|remove}`, `schedule {list|add|remove}`, `club {add|remove|list}`. + +**Cancelling a job.** `ccas cancel ` requests cancellation of a running job — it interrupts the job's fiber on the server, which records the run as `Cancelled`. Cancellation is best-effort: an in-flight blocking database statement runs to completion first, so the job stops at its next interruptible point rather than instantly. A cancelled recruitment run leaves the candidates it found so far as **deferred** (nothing invited), so you can review and confirm them afterwards. Cancel is single-server-scoped — it reaches jobs running on the server it is sent to. **Club targeting.** Slug-requiring commands take the club via `--club ` rather than a positional argument; `membership`/`history` accept a comma-separated `--club a,b` or `--all` (every managed club) — but not both at once (`--all` with `--club` is rejected as a likely mistake). When neither is given, the command falls back to the **current club** set with `ccas use-club ` (stored as `current_club` in the [config file](#cli-config-file)); an explicit `--club`/`--all` always wins. `ccas use-club` is a local config write — no server call — so it works offline and only warns (does not reject) if the slug isn't among the cached clubs. @@ -92,7 +95,7 @@ The detached server writes its pid to `${XDG_STATE_HOME:-~/.local/state}/ccas/cc ### Shell completion -`ccas completion ` prints a self-contained, **pure-shell** completion script. It boots the JVM once when you install it; afterwards every `` runs entirely in the shell (no JVM, no network), so completion is instant. The scripts complete subcommands and flags, plus **club slugs** (after `--club`, and for the `ccas use-club` argument) and **recent job ids** (for `ccas logs`) read from cache files. +`ccas completion ` prints a self-contained, **pure-shell** completion script. It boots the JVM once when you install it; afterwards every `` runs entirely in the shell (no JVM, no network), so completion is instant. The scripts complete subcommands and flags, plus **club slugs** (after `--club`, and for the `ccas use-club` argument) and **recent job ids** (for `ccas logs` and `ccas cancel`) read from cache files. ```bash # bash — install once (bash-completion auto-loads it) diff --git a/completions/ccas.bash b/completions/ccas.bash index a4c51c75..315c6ebd 100644 --- a/completions/ccas.bash +++ b/completions/ccas.bash @@ -31,7 +31,7 @@ _ccas() { words=("${COMP_WORDS[@]}") fi - local top="server use-club membership history recruit stats jobs logs blacklist schedule club config completion" + local top="server use-club membership history recruit stats jobs logs cancel blacklist schedule club config completion" local global="--help --version" local envkeys="CCAS_CONTACT_EMAIL DATABASE_URL DB_HOST DB_PORT DB_NAME DB_USER DB_PASSWORD DB_SCHEMA SERVER_PORT SERVER_HOST JOB_LOGS_DIR SCHEDULER_POLL_MINUTES SCHEDULER_MATCHREF_INTERVAL_HOURS SCHEDULER_MATCHREF_ENABLED SCHEDULER_CLUBDATA_INTERVAL_HOURS SCHEDULER_CLUBDATA_ENABLED SCHEDULER_HISTORY_INTERVAL_HOURS SCHEDULER_HISTORY_ENABLED SCHEDULER_MEMBERSHIP_INTERVAL_HOURS SCHEDULER_MEMBERSHIP_ENABLED DB_POOL_MAX DB_POOL_MIN_IDLE DB_POOL_CONNECTION_TIMEOUT DB_POOL_IDLE_TIMEOUT DB_POOL_MAX_LIFETIME DB_POOL_KEEPALIVE_TIME DB_POOL_CONNECTION_TEST_QUERY DB_POOL_INIT_FAIL_TIMEOUT DB_SOCKET_TIMEOUT_SECONDS DB_CONNECT_TIMEOUT_SECONDS DB_TCP_KEEP_ALIVE DB_RETRY_BASE_DELAY_MS DB_RETRY_MAX_RETRIES CHESS_COM_API_RECOVERY_TIERS CHESS_COM_API_COOLDOWN_SECONDS CHESS_COM_API_CF_COOLDOWN_SECONDS CHESS_COM_API_FAILURE_WINDOW_SIZE CHESS_COM_API_FAILURE_THRESHOLD CHESS_COM_API_MIN_SAMPLE_SIZE CHESS_COM_API_MIN_REQUEST_DELAY_MS CHESS_COM_API_EMA_TAU_MS CHESS_COM_API_MIN_TIER_OBSERVATION_SECONDS CHESS_COM_API_RETRY_BASE_SECONDS CHESS_COM_API_CF_RETRY_DELAY_SECONDS CHESS_COM_API_CONNECTION_RETRY_BASE_SECONDS CHESS_COM_API_MAX_429_RETRIES CHESS_COM_API_MAX_CF_RETRIES CHESS_COM_API_MAX_CONNECTION_RETRIES CHESS_COM_API_STATS_FLUSH_INTERVAL_SECONDS" @@ -68,6 +68,7 @@ _ccas() { stats) opts="--server --since --until --club --no-progress" ;; jobs) opts="--server --limit" ;; logs) opts="--server --no-progress"; pos="jobid" ;; + cancel) opts="--server"; pos="jobid" ;; blacklist) case "$sub" in "") COMPREPLY=( $(compgen -W "add list remove --help" -- "$cur") ); return ;; diff --git a/src/main/scala/ccas/cli/CliCommand.scala b/src/main/scala/ccas/cli/CliCommand.scala index e77cb0b9..3d2793d9 100644 --- a/src/main/scala/ccas/cli/CliCommand.scala +++ b/src/main/scala/ccas/cli/CliCommand.scala @@ -89,6 +89,7 @@ object CliCommand { ) extends ServerCommand final case class Jobs(server: String, limit: Option[Int]) extends ServerCommand final case class Logs(server: String, jobId: String, noProgress: Boolean) extends ServerCommand + final case class Cancel(server: String, jobId: String) extends ServerCommand final case class BlacklistAdd( server: String, club: Option[String], @@ -256,6 +257,11 @@ object CliCommand { .withHelp("Poll a job's status and logs until it finishes") .map { case ((server, noProgress), jobId) => Logs(server, jobId, noProgress) } + private def cancel(default: String): Command[CliCommand] = + Command("cancel", serverOpt(default), Args.text("jobId") ?? "Job run id to cancel") + .withHelp("Request cancellation of a running job (the job stops as soon as it reaches an interruptible point)") + .map { case (server, jobId) => Cancel(server, jobId) } + private def blacklistAdd(default: String): Command[CliCommand] = Command( "add", @@ -415,6 +421,7 @@ object CliCommand { stats(defaultServer), jobs(defaultServer), logs(defaultServer), + cancel(defaultServer), blacklist(defaultServer), schedule(defaultServer), clubs(defaultServer), diff --git a/src/main/scala/ccas/cli/CompletionSpec.scala b/src/main/scala/ccas/cli/CompletionSpec.scala index dce9107d..d0537807 100644 --- a/src/main/scala/ccas/cli/CompletionSpec.scala +++ b/src/main/scala/ccas/cli/CompletionSpec.scala @@ -77,6 +77,7 @@ object CompletionSpec { ), Leaf(List("jobs"), List(server, "--limit"), List(server, "--limit"), NoArgs), Leaf(List("logs"), List(server, "--no-progress"), List(server), JobId), + Leaf(List("cancel"), List(server), List(server), JobId), Leaf( List("blacklist", "add"), List(server, "--reason", "--months", clubFlag), @@ -120,7 +121,7 @@ object CompletionSpec { /** Top-level subcommand names, in tree order (matches `CliCommand.command(...).subcommands`). */ val topLevel: List[String] = - List("server", "use-club", "membership", "history", "recruit", "stats", "jobs", "logs", + List("server", "use-club", "membership", "history", "recruit", "stats", "jobs", "logs", "cancel", "blacklist", "schedule", "club", "config", "completion") /** Flags available on every command. */ @@ -154,6 +155,7 @@ object CompletionSpec { List("stats") -> "Submit a club performance-stats job", List("jobs") -> "List recent jobs and their status", List("logs") -> "Poll a job's status and logs until it finishes", + List("cancel") -> "Request cancellation of a running job (the job stops as soon as it reaches an interruptible point)", List("completion") -> "Emit a shell completion script (bash, zsh, or fish)", List("club", "add") -> "Mark a club as one you manage with CCAS", List("club", "remove") -> "Remove a club from the ones you manage", diff --git a/src/main/scala/ccas/cli/Dispatcher.scala b/src/main/scala/ccas/cli/Dispatcher.scala index 89c77754..47e42a55 100644 --- a/src/main/scala/ccas/cli/Dispatcher.scala +++ b/src/main/scala/ccas/cli/Dispatcher.scala @@ -11,6 +11,7 @@ import ccas.api.player.ApiPlayer import ccas.server.routes.BlacklistRoutes.{BlacklistEntryResponse, CreateBlacklistRequest} import ccas.server.routes.ManagedClubRoutes.{ManagedClubResponse, MarkManagedRequest} import ccas.server.routes.JobRoutes.{ + CancelResult, ClubJobResult, ConfirmResult, HistoryRequest, @@ -155,6 +156,12 @@ object Dispatcher { case CliCommand.Logs(_, jobId, _) => follower.followJob(jobId) + // A 404 (no running job to cancel) surfaces via the client's decode path as a CliError, rendered by `dispatch`'s + // catchAll as "error: …" with exit 1 — so success here means the interrupt was dispatched. + case CliCommand.Cancel(_, jobId) => + api.postEmpty[CancelResult](s"/api/jobs/$jobId/cancel") + .flatMap(_ => Console.printLine(s"cancellation requested for job $jobId").orDie.as(0)) + case CliCommand.BlacklistAdd(_, club, usernames, reason, months) => resolveClub(club, currentClub).flatMap(slug => api.postUnit[CreateBlacklistRequest]( diff --git a/src/main/scala/ccas/cli/JobFollower.scala b/src/main/scala/ccas/cli/JobFollower.scala index f3919dd4..6c0b2a88 100644 --- a/src/main/scala/ccas/cli/JobFollower.scala +++ b/src/main/scala/ccas/cli/JobFollower.scala @@ -175,6 +175,9 @@ final class JobFollower( job.status match { case "Completed" => ZIO.succeed(0) case "Failed" => Console.printLineError(s"$jobId failed: ${job.error.getOrElse("unknown error")}").orDie.as(1) + // A followed job cancelled out from under us (e.g. `ccas cancel` from another terminal): a clean terminal + // outcome, not an "unexpected status". Non-zero — the job did not complete. + case "Cancelled" => Console.printLineError(s"$jobId was cancelled").orDie.as(1) case other => Console.printLineError(s"$jobId: unexpected status '$other'").orDie.as(1) } } diff --git a/src/main/scala/ccas/server/jobs/JobRun.scala b/src/main/scala/ccas/server/jobs/JobRun.scala index a471064c..8b17b277 100644 --- a/src/main/scala/ccas/server/jobs/JobRun.scala +++ b/src/main/scala/ccas/server/jobs/JobRun.scala @@ -95,11 +95,25 @@ object JobRun { WHERE id = $id""".update.run() } - def markOrphansAsFailed: ZIO[PostgresClient, SQLException, Int] = + // `completedAt` is supplied by the caller (from `Clock.instant`) rather than SQL `NOW()` so every job timestamp comes + // from the one testable app clock — consistent with `startedAt` / `updateStatus`, and skew-free against the DB host. + def markOrphansAsFailed(completedAt: Instant): ZIO[PostgresClient, SQLException, Int] = connectZIO { val failed = JobRunStatus.Failed val running = JobRunStatus.Running - sql"""UPDATE job_run SET status = $failed, completed_at = NOW(), error = 'Service restarted' + sql"""UPDATE job_run SET status = $failed, completed_at = $completedAt, error = 'Service restarted' WHERE status = $running""".update.run() } + + // Flip a job to Cancelled — the terminal write when an operator interrupts a running job (see JobRunner.cancel). The + // `WHERE status = Running` guard makes it a no-op if the job already reached Completed/Failed between the interrupt + // request and this write, so a cancel can never clobber a genuine terminal outcome. Returns rows affected (0 or 1). + // `completedAt` comes from the caller's `Clock.instant` for the same clock-consistency reason as `markOrphansAsFailed`. + def markCancelled(id: JobRunId, completedAt: Instant): ZIO[PostgresClient, SQLException, Int] = + connectZIO { + val cancelled = JobRunStatus.Cancelled + val running = JobRunStatus.Running + sql"""UPDATE job_run SET status = $cancelled, completed_at = $completedAt, error = 'Cancelled by operator' + WHERE id = $id AND status = $running""".update.run() + } } diff --git a/src/main/scala/ccas/server/jobs/JobRunStatus.scala b/src/main/scala/ccas/server/jobs/JobRunStatus.scala index 24767400..fb91ecab 100644 --- a/src/main/scala/ccas/server/jobs/JobRunStatus.scala +++ b/src/main/scala/ccas/server/jobs/JobRunStatus.scala @@ -3,7 +3,7 @@ package ccas.server.jobs import ccas.utils.sql.EnumSql enum JobRunStatus { - case Running, Completed, Failed + case Running, Completed, Failed, Cancelled } object JobRunStatus extends EnumSql[JobRunStatus] diff --git a/src/main/scala/ccas/server/jobs/JobRunner.scala b/src/main/scala/ccas/server/jobs/JobRunner.scala index c97079f2..89dbacd1 100644 --- a/src/main/scala/ccas/server/jobs/JobRunner.scala +++ b/src/main/scala/ccas/server/jobs/JobRunner.scala @@ -10,7 +10,7 @@ import ccas.utils.sql.PostgresClient import ccas.utils.sql.PostgresClient.withTransaction import zio.json.EncoderOps import zio.stream.{SubscriptionRef, ZStream} -import zio.{Clock, Duration, Promise, RIO, RLayer, Ref, Schedule, Scope, UIO, ZIO, ZLayer, durationInt} +import zio.{Clock, Duration, Fiber, Promise, RIO, RLayer, Ref, Schedule, Scope, UIO, ZIO, ZLayer, durationInt} import ccas.analysis.tables.{AppSetting, Club, RunTrigger} import ccas.api.misc.subtypes.{ClubId, ClubSlug, JobRunId} @@ -41,6 +41,15 @@ trait JobRunner { /** Look up a job by ID, returning `None` if no such job exists. */ def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] + /** Request cancellation of a running job by interrupting its fiber. Returns `true` if a live job fiber was found in + * THIS process and the interrupt was dispatched, `false` if no such running fiber exists here (unknown id, an + * already-terminal job, or — under the unsupported multi-server model — a job owned by another instance). The + * interrupt is best-effort and asynchronous: the fiber's own finalizer records the `Cancelled` terminal state (a + * blocking JDBC statement in flight runs to completion first), so a `true` result means "cancellation requested", + * not "already stopped". + */ + def cancel(id: JobRunId): UIO[Boolean] + /** Return the most recent jobs ordered by start time descending, up to `limit`. */ def recentJobs(limit: Int): RIO[PostgresClient, List[JobRun]] @@ -75,6 +84,18 @@ object JobRunner { // Per-job progress channels, registered on submit and dropped on completion — the in-memory analog of // `completions`, read by `progressStream` to serve `GET /api/jobs/{id}/progress`. jobChannels <- Ref.make(Map.empty[JobRunId, ProgressDisplay.BarChannel]) + // Per-job forked-fiber handles, read by `cancel` to interrupt a running job. Held via a Promise (completed with + // the fiber the instant `forkIn` returns it) rather than the fiber directly: like `completions`/`jobChannels` + // the map entry is registered BEFORE the fork, so `release`'s de-register can't race ahead of the register and + // strand a handle. A `cancel` that arrives in the sub-millisecond gap before the fiber exists simply awaits the + // Promise. The map is per-process, so `cancel` only reaches fibers in THIS server (see the trait doc / #110). + runningFibers <- Ref.make(Map.empty[JobRunId, Promise[Nothing, Fiber.Runtime[Throwable, Unit]]]) + // Ids for which an operator `cancel` has been requested. `cancel` adds an id before interrupting its fiber; the + // job's `onInterrupt` writes `Cancelled` ONLY if its id is here. This distinguishes an operator cancel from the + // shutdown interrupt `layerScope` fires at every in-flight job on server stop — the latter must leave the row + // `Running` so the next boot's `markOrphansAsFailed` records it as `Failed`/"Service restarted", not a spurious + // "Cancelled by operator". `release` clears the id, so the set only ever holds live-and-cancelling jobs. + cancelRequested <- Ref.make(Set.empty[JobRunId]) // The layer's own scope owns the job fibers (see `submit`): forking into it detaches each job from the // short-lived request fiber that submitted it, and interrupts any still in flight when the server shuts down. // `forkIn` self-cleans — a job's child scope closes when its fiber exits — so completed jobs don't accumulate. @@ -84,7 +105,8 @@ object JobRunner { Files.createDirectories(dir) dir }.orDie - _ <- JobRun.markOrphansAsFailed.provideEnvironment(zio.ZEnvironment(pgClient)) + now <- Clock.instant + _ <- JobRun.markOrphansAsFailed(now).provideEnvironment(zio.ZEnvironment(pgClient)) days <- AppSetting.get(AppSetting.JobLogRetentionDays).provideEnvironment(zio.ZEnvironment(pgClient)) cutoff <- Clock.instant.map(_.minus(days.toLong, ChronoUnit.DAYS)) swept <- FileSink @@ -92,7 +114,17 @@ object JobRunner { .tapError(t => ZIO.logWarning(s"Job-log retention sweep failed: ${t.safeMessage}")) .orElseSucceed(0) _ <- ZIO.logInfo(s"Swept $swept job log(s) older than $days day(s) from $logDir").when(swept > 0) - } yield new JobRunnerLive(display, client, pgClient, completions, jobChannels, layerScope, logDir) + } yield new JobRunnerLive( + display, + client, + pgClient, + completions, + jobChannels, + runningFibers, + cancelRequested, + layerScope, + logDir + ) } private class JobRunnerLive( @@ -101,6 +133,8 @@ object JobRunner { pgClient: PostgresClient, completions: Ref[Map[JobRunId, Promise[Nothing, Unit]]], jobChannels: Ref[Map[JobRunId, ProgressDisplay.BarChannel]], + runningFibers: Ref[Map[JobRunId, Promise[Nothing, Fiber.Runtime[Throwable, Unit]]]], + cancelRequested: Ref[Set[JobRunId]], layerScope: Scope, logDir: Path ) extends JobRunner { @@ -156,15 +190,21 @@ object JobRunner { // observes a fully written, closed file before it sees the job done, and always gets EOF, incl. on shutdown. // The channel de-registers too: `/progress`'s `interruptWhen(promise)` already ends live followers, and a // later subscribe finds no channel (job terminal) and emits a single settling frame. + // Registered before the fork (see `runningFibers`) so the fiber handle is addressable the instant the job is + // live; `fiberSlot` is completed with the fiber below. + fiberSlot <- Promise.make[Nothing, Fiber.Runtime[Throwable, Unit]] release = for { _ <- sink.close() _ <- promise.succeed(()) _ <- completions.update(_ - id) _ <- jobChannels.update(_ - id) + _ <- runningFibers.update(_ - id) + _ <- cancelRequested.update(_ - id) } yield () _ <- completions.update(_ + (id -> promise)) _ <- jobChannels.update(_ + (id -> channel)) - _ <- restore( + _ <- runningFibers.update(_ + (id -> fiberSlot)) + fiber <- restore( // `installLogger` forces ProgressDisplay's ZLogger onto the job fiber itself, so the per-job FileSink is // written regardless of which fiber submitted the job. `forkIn` inherits the submitter's loggers, and an // HTTP request handler carries only the default logger (not the app-scoped ProgressDisplay one) — so @@ -182,6 +222,7 @@ object JobRunner { ) ) ).forkIn(layerScope) + _ <- fiberSlot.succeed(fiber) } yield () } } yield id).catchSome { @@ -211,11 +252,44 @@ object JobRunner { .unit.catchAll(e => ZIO.logError(s"Failed to record job completion: ${e.safeMessage}")) ) - effect.provideEnvironment(env).foldZIO(onFailure, _ => onSuccess) + // Interruption is a `Cause`, not a typed error, so it bypasses `foldZIO`'s branches entirely. This hook fires on + // ANY interruption — but only an operator `cancel` (which registers the id in `cancelRequested` before + // interrupting) should record `Cancelled`. The other interrupt source is `layerScope` tearing down every in-flight + // job on server shutdown; those must be left `Running` so the next boot's `markOrphansAsFailed` records them as + // `Failed`/"Service restarted", not "Cancelled by operator". The finalizer runs uninterruptibly so the write lands; + // `markCancelled`'s `WHERE status = Running` guard makes it a no-op if the job reached a terminal state first. + def onCancelled: UIO[Unit] = + ZIO.whenZIODiscard(cancelRequested.get.map(_.contains(id)))( + Clock.instant.flatMap(now => + JobRun.markCancelled(id, now) + .provideEnvironment(env) + .unit.catchAll(e => ZIO.logError(s"Failed to record job cancellation: ${e.safeMessage}")) + ) + ) + + effect.provideEnvironment(env).foldZIO(onFailure, _ => onSuccess).onInterrupt(onCancelled) override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = JobRun.selectId(id) + // Interrupt the job's fiber if it is running in this process. Registers the id in `cancelRequested` first so the + // fiber's `onInterrupt` knows this is an operator cancel (not a shutdown) and records `Cancelled`; then + // `interruptFork` dispatches the interrupt in the background so the caller (the HTTP handler) returns promptly + // rather than blocking until the fiber unwinds. `slot.await` resolves the fiber the instant the fork registered it. + // `fiber.poll` guards the narrow window where the job already finished but `release` hasn't de-registered the slot + // yet: a completed fiber returns `false` (nothing to cancel) rather than a misleading `true`. + override def cancel(id: JobRunId): UIO[Boolean] = + runningFibers.get.map(_.get(id)).flatMap { + case None => ZIO.succeed(false) + case Some(slot) => + slot.await.flatMap(fiber => + fiber.poll.flatMap { + case Some(_) => ZIO.succeed(false) // already terminal — release just hasn't cleared the slot yet + case None => (cancelRequested.update(_ + id) *> fiber.interruptFork).as(true) + } + ) + } + override def recentJobs(limit: Int): RIO[PostgresClient, List[JobRun]] = JobRun.selectRecent(limit) diff --git a/src/main/scala/ccas/server/routes/JobRoutes.scala b/src/main/scala/ccas/server/routes/JobRoutes.scala index 4a035521..3b91ef30 100644 --- a/src/main/scala/ccas/server/routes/JobRoutes.scala +++ b/src/main/scala/ccas/server/routes/JobRoutes.scala @@ -100,6 +100,13 @@ object JobRoutes { given JsonCodec[ConfirmResult] = DeriveJsonCodec.gen } + /** Result of a cancel request: `jobId` echoed back. A 200 means the interrupt was dispatched to a live job fiber + * ("cancellation requested"); a missing / already-terminal job is a 404 instead. */ + private[ccas] case class CancelResult(jobId: String) + object CancelResult { + given JsonCodec[CancelResult] = DeriveJsonCodec.gen + } + private[ccas] case class JobStatusResponse( id: String, kind: String, @@ -268,6 +275,19 @@ object JobRoutes { case None => jsonResponse(Status.NotFound, ErrorResponse(s"Job $jobId not found")) }).pipe(withErrorHandling) }, + // Request cancellation of a running job: interrupt its fiber (best-effort, async — the job records `Cancelled` + // itself as it unwinds). 200 if a live job fiber was found and interrupted; 404 if the id is unknown, already + // terminal, or (unsupported multi-server) owned by another instance. POST, not DELETE: the row is retained, this + // is a state transition, not a resource removal. + Method.POST / "api" / "jobs" / string("jobId") / "cancel" -> handler { (jobId: String, _: Request) => + (for { + runner <- ZIO.service[JobRunner] + cancelled <- runner.cancel(JobRunId.wrap(jobId)) + } yield + if (cancelled) { jsonResponse(Status.Ok, CancelResult(jobId)) } + else { jsonResponse(Status.NotFound, ErrorResponse(s"No running job $jobId to cancel")) } + ).pipe(withErrorHandling) + }, // Chunked `text/plain` stream of a job's log lines. Stays open while the job runs (lines arrive as emitted) and // closes once the job is terminal and the tail reaches EOF, so a client can treat body-close as "job finished". Method.GET / "api" / "jobs" / string("jobId") / "logs" -> handler { (jobId: String, _: Request) => diff --git a/src/test/scala/ccas/cli/TestCcasCompletion.scala b/src/test/scala/ccas/cli/TestCcasCompletion.scala index 52f2fa4f..74ce21c6 100644 --- a/src/test/scala/ccas/cli/TestCcasCompletion.scala +++ b/src/test/scala/ccas/cli/TestCcasCompletion.scala @@ -99,6 +99,7 @@ object TestCcasCompletion extends ZIOSpecDefault { positionalOf("history") == PositionalKind.NoArgs, positionalOf("recruit") == PositionalKind.NoArgs, positionalOf("logs") == PositionalKind.JobId, + positionalOf("cancel") == PositionalKind.JobId, positionalOf("completion") == PositionalKind.Shell, positionalOf("blacklist", "add") == PositionalKind.Other, positionalOf("config", "get") == PositionalKind.EnvKey, diff --git a/src/test/scala/ccas/cli/TestCliParser.scala b/src/test/scala/ccas/cli/TestCliParser.scala index 7712982b..2dde375d 100644 --- a/src/test/scala/ccas/cli/TestCliParser.scala +++ b/src/test/scala/ccas/cli/TestCliParser.scala @@ -73,6 +73,11 @@ object TestCliParser extends ZIOSpecDefault { assertTrue(c.contains(CliCommand.Jobs("http://example:9000", None))) ) }, + test("cancel parses the jobId (options before the positional)") { + parsed("cancel", "--server", "http://example:9000", "job-123").map(c => + assertTrue(c.contains(CliCommand.Cancel("http://example:9000", "job-123"))) + ) + }, test("config api_url becomes the default when --server is absent") { serverOf("http://config:1234")("jobs").map(s => assertTrue(s.contains("http://config:1234"))) }, diff --git a/src/test/scala/ccas/cli/TestJobFollower.scala b/src/test/scala/ccas/cli/TestJobFollower.scala index 0877d257..fc544506 100644 --- a/src/test/scala/ccas/cli/TestJobFollower.scala +++ b/src/test/scala/ccas/cli/TestJobFollower.scala @@ -92,6 +92,12 @@ object TestJobFollower extends ZIOSpecDefault { code <- follower.followJob("job-1") } yield assertTrue(code == 1) }, + test("followJob returns 1 when the job was cancelled") { + for { + follower <- followerWith("Cancelled", Some("Cancelled by operator"), List("partial output")) + code <- follower.followJob("job-1") + } yield assertTrue(code == 1) + }, test("followJob with progress bars on forks the progress consumer, follows, and completes cleanly") { // showProgress = true takes the bars branch: build a client ProgressDisplay, fork the (empty) /progress consumer, // follow the log stream through the display, then tear the consumer down and clear bars on completion. Asserts the diff --git a/src/test/scala/ccas/server/jobs/TestJobRunSql.scala b/src/test/scala/ccas/server/jobs/TestJobRunSql.scala index ac2eeb2f..ce0e62b4 100644 --- a/src/test/scala/ccas/server/jobs/TestJobRunSql.scala +++ b/src/test/scala/ccas/server/jobs/TestJobRunSql.scala @@ -21,7 +21,9 @@ object TestJobRunSql extends ZIOSpecDefault { testSelectRunningForUpdateNullClub, testSelectRunningForUpdateIgnoresNonRunning, testSelectRecentOrdering, - testMarkOrphansAsFailed + testMarkOrphansAsFailed, + testMarkCancelledFlipsRunning, + testMarkCancelledIgnoresTerminal ).provideShared( FreshSchemaLayer("test_job_run", onInit = ServerTables.ensureTables) ) @@ TestAspect.sequential @@ -178,7 +180,7 @@ object TestJobRunSql extends ZIOSpecDefault { _ <- JobRun.insert(running1) _ <- JobRun.insert(running2) _ <- JobRun.insert(completed) - count <- JobRun.markOrphansAsFailed + count <- JobRun.markOrphansAsFailed(Times.t2) r0 <- JobRun.selectId(id0) r1 <- JobRun.selectId(id1) r2 <- JobRun.selectId(id2) @@ -186,8 +188,39 @@ object TestJobRunSql extends ZIOSpecDefault { count == 2, r0.get.status == JobRunStatus.Failed, r0.get.error.contains("Service restarted"), + r0.get.completedAt.contains(Times.t2), r1.get.status == JobRunStatus.Failed, r2.get.status == JobRunStatus.Completed ) } + + private def testMarkCancelledFlipsRunning = test("markCancelled marks a Running job Cancelled with completedAt + error") { + for { + _ <- deleteAll + _ <- JobRun.insert(run0) // Running + count <- JobRun.markCancelled(id0, Times.t1) + result <- JobRun.selectId(id0) + } yield assertTrue( + count == 1, + result.get.status == JobRunStatus.Cancelled, + result.get.completedAt.contains(Times.t1), + result.get.error.contains("Cancelled by operator") + ) + } + + private def testMarkCancelledIgnoresTerminal = + test("markCancelled is a no-op on an already-terminal job (guards against clobbering Completed/Failed)") { + val completed = + JobRun(id0, JobKind.Recruitment, None, RunTrigger.Cli, JobRunStatus.Completed, None, Times.t0, Some(Times.t1), None) + for { + _ <- deleteAll + _ <- JobRun.insert(completed) + count <- JobRun.markCancelled(id0, Times.t1) + result <- JobRun.selectId(id0) + } yield assertTrue( + count == 0, + result.get.status == JobRunStatus.Completed, + result.get.error.isEmpty + ) + } } diff --git a/src/test/scala/ccas/server/jobs/TestJobRunner.scala b/src/test/scala/ccas/server/jobs/TestJobRunner.scala index 4ce9d619..5465df33 100644 --- a/src/test/scala/ccas/server/jobs/TestJobRunner.scala +++ b/src/test/scala/ccas/server/jobs/TestJobRunner.scala @@ -29,6 +29,9 @@ object TestJobRunner extends ZIOSpecDefault { testSubmitAllowsDifferentClub, testSubmitAllowsDifferentKind, testStatusUnknown, + testCancelInterruptsRunningJob, + testCancelUnknownReturnsFalse, + testNonOperatorInterruptDoesNotMarkCancelled, testRecentJobsOrdered, testSubmitWritesJobLog, testLogStreamReplaysCompletedJob, @@ -162,6 +165,47 @@ object TestJobRunner extends ZIOSpecDefault { } yield assertTrue(result.isEmpty) } + private def testCancelInterruptsRunningJob = test("cancel interrupts a running job and marks it Cancelled") { + for { + _ <- deleteAllJobRuns + runner <- ZIO.service[JobRunner] + started <- Promise.make[Nothing, Unit] + // The job announces it is running, then blocks forever — so it is unambiguously live when we cancel. + id <- runner.submit(JobKind.MatchRef, None, None, RunTrigger.Cli, _ => started.succeed(()) *> ZIO.never) + _ <- started.await + cancelled <- runner.cancel(id) + job <- awaitStatus(runner, id) // returns once the job leaves Running (here: Cancelled via onInterrupt) + } yield assertTrue( + cancelled, + job.status == JobRunStatus.Cancelled, + job.completedAt.isDefined, + job.error.exists(_.contains("Cancelled")) + ) + } + + private def testCancelUnknownReturnsFalse = test("cancel returns false for unknown id") { + for { + runner <- ZIO.service[JobRunner] + result <- runner.cancel(JobRunId.wrap("does-not-exist")) + } yield assertTrue(!result) + } + + // Guards the cancel-vs-shutdown distinction: the job interrupts ITSELF (stands in for the `layerScope` interrupt fired + // at every in-flight job on server shutdown), so no operator `cancel` ran and no id is in `cancelRequested`. The + // onInterrupt hook must therefore NOT write Cancelled — the row stays Running for the next boot's orphan sweep. Were + // the gate absent, this would flip to Cancelled/"Cancelled by operator", mislabeling a shutdown as an operator cancel. + private def testNonOperatorInterruptDoesNotMarkCancelled = + test("a job interrupted without an operator cancel is left Running, not marked Cancelled") { + for { + _ <- deleteAllJobRuns + runner <- ZIO.service[JobRunner] + id <- runner.submit(JobKind.MatchRef, None, None, RunTrigger.Cli, _ => ZIO.interrupt) + // The fiber self-interrupts immediately; wait past its unwind + release, then confirm the row never left Running. + _ <- ZIO.sleep(300.millis) + job <- runner.status(id) + } yield assertTrue(job.exists(j => j.status == JobRunStatus.Running && j.completedAt.isEmpty)) + } + private def testSubmitWritesJobLog = test("submit routes job log lines into the per-job file") { val logDir = Paths.get(ConfigFactory.load().getString("job-logs.directory")) for { diff --git a/src/test/scala/ccas/server/routes/TestJobCancelWire.scala b/src/test/scala/ccas/server/routes/TestJobCancelWire.scala new file mode 100644 index 00000000..6e48c891 --- /dev/null +++ b/src/test/scala/ccas/server/routes/TestJobCancelWire.scala @@ -0,0 +1,89 @@ +package ccas.server.routes + +import zio.* +import zio.http.{Client, Server} +import zio.test.{assertTrue, Spec, TestAspect, ZIOSpecDefault} + +import ccas.analysis.tables.RunTrigger +import ccas.api.misc.subtypes.JobRunId +import ccas.cli.{CcasApiClient, CliError} +import ccas.server.jobs.{JobKind, JobRun, JobRunner, JobRunStatus} +import ccas.server.routes.JobRoutes.CancelResult +import ccas.server.ServerTables +import ccas.utils.client.TestChessComClientSupport +import ccas.utils.sql.{FreshSchemaLayer, PostgresClient} +import ccas.utils.ProgressDisplay + +/** End-to-end wire smoke for job cancellation: a REAL zio-http server bound to an ephemeral port serving the REAL + * `JobRoutes`, driven by the REAL `CcasApiClient` (the CLI's HTTP client) over the loopback, backed by the REAL + * `JobRunner.live`. Exercises the whole `ccas cancel` path — client JSON encode/route → `runner.cancel` → fiber + * interrupt → `Cancelled` terminal write → `CancelResult` decode — and the 404 branch. The one non-production seam is + * the job body (`ZIO.never`, to hold a job Running) and that it is submitted via the runner rather than an HTTP submit + * route; both submit routes and the analysis apps are covered by their own suites. `Server.install` binds the shared + * server once (the real port even from configured port 0) and keeps serving under the `Server` layer's lifetime. + */ +object TestJobCancelWire extends ZIOSpecDefault { + + override def spec: Spec[Any, Throwable] = suite("TestJobCancelWire")( + testCancelLiveJobOverHttp, + testCancelUnknownJobOverHttp + ).provideShared( + FreshSchemaLayer("test_cancel_wire", onInit = ServerTables.ensureTables), + TestChessComClientSupport.dummyLayer, + JobRunner.live, + ZLayer.succeed(ProgressDisplay.make(enabled = false)), + Server.defaultWith(_.port(0)), + boundPort, + Client.default + ) @@ TestAspect.sequential @@ TestAspect.withLiveClock @@ TestAspect.timeout(60.seconds) + + // The shared server's actual bound port, wrapped so it can't collide with any stray `Int` in the environment. + private final case class ServerPort(value: Int) + + // Install the real JobRoutes on the shared server ONCE and expose its actual bound port (real port even from the + // configured port 0). `install` needs no caller Scope — the server keeps serving under the `Server` layer's lifetime — + // so `fromZIO` fits; installing per-test would re-register the routes and log a benign "Duplicate routes" warning. + private val boundPort = ZLayer.fromZIO(Server.install(JobRoutes.routes).map(ServerPort(_))) + + // 127.0.0.1 (not "localhost") to avoid an IPv6/IPv4 loopback mismatch against the IPv4-bound test server. + private def apiFor(port: ServerPort): URIO[Client, CcasApiClient] = + CcasApiClient.live(s"http://127.0.0.1:${port.value}") + + private def awaitCancelled(runner: JobRunner, id: JobRunId): ZIO[PostgresClient, Throwable, JobRun] = + runner.status(id).flatMap { + case Some(job) if job.status != JobRunStatus.Running => ZIO.succeed(job) + case _ => ZIO.sleep(50.millis) *> awaitCancelled(runner, id) + }.timeoutFail(new Exception(s"job $id did not leave Running"))(10.seconds) + + private def testCancelLiveJobOverHttp = + test("POST /api/jobs/{id}/cancel over real HTTP cancels a live job and returns CancelResult") { + for { + runner <- ZIO.service[JobRunner] + port <- ZIO.service[ServerPort] + api <- apiFor(port) + started <- Promise.make[Nothing, Unit] + // A real live job: announces it is running, then blocks forever until the interrupt lands. + id <- runner.submit(JobKind.MatchRef, None, None, RunTrigger.Cli, _ => started.succeed(()) *> ZIO.never) + _ <- started.await + result <- api.postEmpty[CancelResult](s"/api/jobs/${JobRunId.unwrap(id)}/cancel") + job <- awaitCancelled(runner, id) + } yield assertTrue( + result.jobId == JobRunId.unwrap(id), + job.status == JobRunStatus.Cancelled, + job.completedAt.isDefined, + job.error.exists(_.contains("Cancelled")) + ) + } + + private def testCancelUnknownJobOverHttp = + test("POST /api/jobs/{id}/cancel for an unknown job surfaces the 404 as a CliError") { + for { + port <- ZIO.service[ServerPort] + api <- apiFor(port) + result <- api.postEmpty[CancelResult]("/api/jobs/does-not-exist/cancel").either + } yield assertTrue(result match { + case Left(e: CliError) => e.message.contains("No running job") && e.exitCode == 1 + case _ => false + }) + } +} diff --git a/src/test/scala/ccas/server/routes/TestRoutes.scala b/src/test/scala/ccas/server/routes/TestRoutes.scala index a69f5fe0..13a4138b 100644 --- a/src/test/scala/ccas/server/routes/TestRoutes.scala +++ b/src/test/scala/ccas/server/routes/TestRoutes.scala @@ -73,6 +73,17 @@ object TestRoutes extends ZIOSpecDefault { override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = jobs.get.map(_.get(id)) + // Flip a stored Running job to Cancelled (returns true); unknown or already-terminal → false. Enough to pin the + // cancel route's 200/404 without the real fiber-interrupt mechanics (covered against JobRunner.live in TestJobRunner). + override def cancel(id: JobRunId): UIO[Boolean] = + jobs.modify { m => + m.get(id) match { + case Some(job) if job.status == JobRunStatus.Running => + (true, m + (id -> job.copy(status = JobRunStatus.Cancelled, completedAt = Some(Instant.now())))) + case _ => (false, m) + } + } + override def recentJobs(limit: Int): RIO[PostgresClient, List[JobRun]] = jobs.get.map(_.values.toList.sortBy(_.startedAt)(using Ordering[Instant].reverse).take(limit)) @@ -152,6 +163,8 @@ object TestRoutes extends ZIOSpecDefault { testGetJobsReturnsList, testGetJobByIdReturns200, testGetJobByIdReturns404, + testCancelJobReturns200, + testCancelJobReturns404, testGetJobLogsReturns200, testGetJobLogsReturns404, testGetJobProgressReturns200, @@ -379,6 +392,38 @@ object TestRoutes extends ZIOSpecDefault { } yield assertTrue(response.status == Status.NotFound) } + private def testCancelJobReturns200 = test("POST /api/jobs/:id/cancel returns 200 and cancels a running job") { + val t0 = LocalDateTime.of(2025, 6, 1, 0, 0).toInstant(ZoneOffset.UTC) + val job = JobRun( + id = JobRunId.wrap("cancel-id"), + kind = JobKind.Membership, + clubId = None, + trigger = RunTrigger.Cli, + status = JobRunStatus.Running, + params = None, + startedAt = t0, + completedAt = None, + error = None + ) + for { + fake <- getFakeRunner + _ <- fake.prePopulate(job) + response <- JobRoutes.routes.runZIO(jsonRequest(Method.POST, "/api/jobs/cancel-id/cancel")) + body <- response.body.asString + after <- fake.status(JobRunId.wrap("cancel-id")) + } yield assertTrue( + response.status == Status.Ok, + body.contains("cancel-id"), + after.exists(_.status == JobRunStatus.Cancelled) + ) + } + + private def testCancelJobReturns404 = test("POST /api/jobs/:id/cancel returns 404 for an unknown or terminal job") { + for { + response <- JobRoutes.routes.runZIO(jsonRequest(Method.POST, "/api/jobs/nonexistent/cancel")) + } yield assertTrue(response.status == Status.NotFound) + } + private def testGetJobLogsReturns200 = test("GET /api/jobs/:id/logs streams chunked text/plain for an existing job") { val t0 = LocalDateTime.of(2025, 6, 1, 0, 0).toInstant(ZoneOffset.UTC) val job = JobRun( diff --git a/src/test/scala/ccas/server/scheduler/TestJobScheduler.scala b/src/test/scala/ccas/server/scheduler/TestJobScheduler.scala index 45a146ec..b4bbe257 100644 --- a/src/test/scala/ccas/server/scheduler/TestJobScheduler.scala +++ b/src/test/scala/ccas/server/scheduler/TestJobScheduler.scala @@ -2,7 +2,7 @@ package ccas.server.scheduler import java.time.temporal.ChronoUnit -import zio.{durationInt, Clock, Duration, Ref, RIO, ZIO} +import zio.{durationInt, Clock, Duration, Ref, RIO, UIO, ZIO} import zio.stream.ZStream import zio.test.{assertCompletes, assertTrue, Spec, TestAspect, TestClock, ZIOSpecDefault} @@ -64,6 +64,7 @@ object TestJobScheduler extends ZIOSpecDefault { submissions.update(_ + 1).as(JobRunId.generate()) override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none + override def cancel(id: JobRunId): UIO[Boolean] = ZIO.succeed(false) override def recentJobs(limit: Int): RIO[PostgresClient, List[JobRun]] = ZIO.succeed(Nil) override def logStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none override def progressStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none @@ -81,6 +82,7 @@ object TestJobScheduler extends ZIOSpecDefault { captured.update(params :: _).as(JobRunId.generate()) override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none + override def cancel(id: JobRunId): UIO[Boolean] = ZIO.succeed(false) override def recentJobs(limit: Int): RIO[PostgresClient, List[JobRun]] = ZIO.succeed(Nil) override def logStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none override def progressStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none @@ -98,6 +100,7 @@ object TestJobScheduler extends ZIOSpecDefault { submitted.update(clubId :: _).as(JobRunId.generate()) override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none + override def cancel(id: JobRunId): UIO[Boolean] = ZIO.succeed(false) override def recentJobs(limit: Int): RIO[PostgresClient, List[JobRun]] = ZIO.succeed(Nil) override def logStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none override def progressStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none @@ -275,6 +278,7 @@ object TestJobScheduler extends ZIOSpecDefault { callCount.update(_ + 1) *> ZIO.fail(new RuntimeException("boom")) override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none + override def cancel(id: JobRunId): UIO[Boolean] = ZIO.succeed(false) override def recentJobs(limit: Int): RIO[PostgresClient, List[JobRun]] = ZIO.succeed(Nil) override def logStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none override def progressStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none @@ -315,6 +319,7 @@ object TestJobScheduler extends ZIOSpecDefault { ZIO.when(clubId.contains(failClubId))(ZIO.fail(new RuntimeException("boom"))).as(JobRunId.generate()) override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none + override def cancel(id: JobRunId): UIO[Boolean] = ZIO.succeed(false) override def recentJobs(limit: Int): RIO[PostgresClient, List[JobRun]] = ZIO.succeed(Nil) override def logStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none override def progressStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none @@ -363,6 +368,7 @@ object TestJobScheduler extends ZIOSpecDefault { callCount.update(_ + 1) *> ZIO.fail(ConflictException(s"A $kind job is already running")) override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none + override def cancel(id: JobRunId): UIO[Boolean] = ZIO.succeed(false) override def recentJobs(limit: Int): RIO[PostgresClient, List[JobRun]] = ZIO.succeed(Nil) override def logStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none override def progressStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = ZIO.none