diff --git a/completions/ccas.bash b/completions/ccas.bash index 315c6eb..0412be2 100644 --- a/completions/ccas.bash +++ b/completions/ccas.bash @@ -62,10 +62,10 @@ _ccas() { *) COMPREPLY=(); return ;; esac ;; use-club) pos="slug" ;; - membership) opts="--server --trust-usernames --no-trust-usernames --club --all --no-progress" ;; - history) opts="--server --full --include-finished --refresh --refresh-min-hours --club --all --no-progress" ;; + membership) opts="--server --trust-usernames --no-trust-usernames --club --all --no-progress --detach" ;; + history) opts="--server --full --include-finished --refresh --refresh-min-hours --club --all --no-progress --detach" ;; recruit) opts="--server --alias --target --cumulative --source-clubs --time-limit-minutes --explore --no-explore --club --stdout --report --no-progress" ;; - stats) opts="--server --since --until --club --no-progress" ;; + stats) opts="--server --since --until --club --no-progress --detach" ;; jobs) opts="--server --limit" ;; logs) opts="--server --no-progress"; pos="jobid" ;; cancel) opts="--server"; pos="jobid" ;; diff --git a/src/main/scala/ccas/cli/CliCommand.scala b/src/main/scala/ccas/cli/CliCommand.scala index 3d2793d..8495077 100644 --- a/src/main/scala/ccas/cli/CliCommand.scala +++ b/src/main/scala/ccas/cli/CliCommand.scala @@ -54,7 +54,8 @@ object CliCommand { clubs: List[String], all: Boolean, trustUsernames: Option[Boolean], - noProgress: Boolean + noProgress: Boolean, + detach: Boolean ) extends ServerCommand final case class History( server: String, @@ -64,7 +65,8 @@ object CliCommand { includeFinished: Boolean, refresh: Boolean, refreshMinHours: Option[Int], - noProgress: Boolean + noProgress: Boolean, + detach: Boolean ) extends ServerCommand final case class Recruit( server: String, @@ -85,7 +87,8 @@ object CliCommand { club: Option[String], since: Option[String], until: Option[String], - noProgress: Boolean + noProgress: Boolean, + detach: Boolean ) 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 @@ -144,6 +147,11 @@ object CliCommand { private val noProgressOpt: Options[Boolean] = Options.boolean("no-progress") ?? "Don't render progress bars while following the job (plain log lines only)" + // Fire-and-forget: submit the job and return immediately without following it. Reattach later with `ccas logs `. + // Not offered on `recruit` (its result delivery / invite confirmation needs the follow) or `logs` (which IS the follow). + private val detachOpt: Options[Boolean] = + Options.boolean("detach") ?? "Submit the job and return immediately without following it (reattach with 'ccas logs ')" + private def intOpt(name: String): Options[Option[Int]] = Options.integer(name).map(_.toInt).optional // Single-club target. Absent → Dispatcher falls back to the config's `current_club`. @@ -190,10 +198,10 @@ object CliCommand { .map(Use.apply) private def membership(default: String): Command[CliCommand] = - Command("membership", serverOpt(default) ++ trustOpt ++ clubsOpt ++ allOpt ++ noProgressOpt) + Command("membership", serverOpt(default) ++ trustOpt ++ clubsOpt ++ allOpt ++ noProgressOpt ++ detachOpt) .withHelp("Submit a membership-sync job (current club, --club a,b, or --all managed clubs)") - .map { case (server, trust, clubs, all, noProgress) => - Membership(server, clubs, all, trust, noProgress) + .map { case (server, trust, clubs, all, noProgress, detach) => + Membership(server, clubs, all, trust, noProgress, detach) } private def history(default: String): Command[CliCommand] = @@ -204,10 +212,10 @@ object CliCommand { (Options.boolean("include-finished") ?? "Re-queue recently finished matches for refresh") ++ (Options.boolean("refresh") ?? "Refresh already-stored matches, not just newly seen ones") ++ (intOpt("refresh-min-hours") ?? "Skip refreshing a match seen within the last N hours") ++ - clubsOpt ++ allOpt ++ noProgressOpt + clubsOpt ++ allOpt ++ noProgressOpt ++ detachOpt ).withHelp("Submit a match-history crawl job (current club, --club a,b, or --all managed clubs)") - .map { case (server, full, includeFinished, refresh, refreshMinHours, clubs, all, noProgress) => - History(server, clubs, all, full, includeFinished, refresh, refreshMinHours, noProgress) + .map { case (server, full, includeFinished, refresh, refreshMinHours, clubs, all, noProgress, detach) => + History(server, clubs, all, full, includeFinished, refresh, refreshMinHours, noProgress, detach) } private val sourceClubsOpt: Options[List[String]] = @@ -243,9 +251,11 @@ object CliCommand { serverOpt(default) ++ (Options.text("since").optional ?? "Start of the date window (ISO-8601 date or instant)") ++ (Options.text("until").optional ?? "End of the date window (requires --since)") ++ - clubOpt ++ noProgressOpt + clubOpt ++ noProgressOpt ++ detachOpt ).withHelp("Submit a club performance-stats job") - .map { case (server, since, until, club, noProgress) => Stats(server, club, since, until, noProgress) } + .map { case (server, since, until, club, noProgress, detach) => + Stats(server, club, since, until, noProgress, detach) + } private def jobs(default: String): Command[CliCommand] = Command("jobs", serverOpt(default) ++ (intOpt("limit") ?? "Maximum number of recent jobs to list")) diff --git a/src/main/scala/ccas/cli/CompletionSpec.scala b/src/main/scala/ccas/cli/CompletionSpec.scala index d053780..8d81af6 100644 --- a/src/main/scala/ccas/cli/CompletionSpec.scala +++ b/src/main/scala/ccas/cli/CompletionSpec.scala @@ -50,13 +50,14 @@ object CompletionSpec { Leaf(List("use-club"), Nil, Nil, Slug), Leaf( List("membership"), - List(server, "--trust-usernames", "--no-trust-usernames", clubFlag, "--all", "--no-progress"), + List(server, "--trust-usernames", "--no-trust-usernames", clubFlag, "--all", "--no-progress", "--detach"), List(server, clubFlag), NoArgs ), Leaf( List("history"), - List(server, "--full", "--include-finished", "--refresh", "--refresh-min-hours", clubFlag, "--all", "--no-progress"), + List(server, "--full", "--include-finished", "--refresh", "--refresh-min-hours", clubFlag, "--all", + "--no-progress", "--detach"), List(server, "--refresh-min-hours", clubFlag), NoArgs ), @@ -71,7 +72,7 @@ object CompletionSpec { ), Leaf( List("stats"), - List(server, "--since", "--until", clubFlag, "--no-progress"), + List(server, "--since", "--until", clubFlag, "--no-progress", "--detach"), List(server, "--since", "--until", clubFlag), NoArgs ), diff --git a/src/main/scala/ccas/cli/Dispatcher.scala b/src/main/scala/ccas/cli/Dispatcher.scala index 47e42a5..5d21434 100644 --- a/src/main/scala/ccas/cli/Dispatcher.scala +++ b/src/main/scala/ccas/cli/Dispatcher.scala @@ -92,9 +92,9 @@ object Dispatcher { cmd: CliCommand.ServerCommand, currentClub: Option[String] ): Task[Int] = cmd match { - case CliCommand.Membership(_, clubs, all, trust, _) => + case CliCommand.Membership(_, clubs, all, trust, _, detach) => resolveClubs(api, clubs, all, currentClub).flatMap(slugs => - followEachClub(follower, slugs)(slug => + followEachClub(follower, slugs, detach)(slug => api.postJson[MembershipRequest, List[ClubJobResult]]( "/api/jobs/membership", MembershipRequest(NonEmptyChunk.single(slug), trust) @@ -102,9 +102,9 @@ object Dispatcher { ) ) - case CliCommand.History(_, clubs, all, full, includeFinished, refresh, refreshMinHours, _) => + case CliCommand.History(_, clubs, all, full, includeFinished, refresh, refreshMinHours, _, detach) => resolveClubs(api, clubs, all, currentClub).flatMap(slugs => - followEachClub(follower, slugs)(slug => + followEachClub(follower, slugs, detach)(slug => api.postJson[HistoryRequest, List[ClubJobResult]]( "/api/jobs/history", HistoryRequest(NonEmptyChunk.single(slug), flag(full), flag(includeFinished), flag(refresh), refreshMinHours) @@ -142,12 +142,15 @@ object Dispatcher { ) } - case CliCommand.Stats(_, club, since, until, _) => + case CliCommand.Stats(_, club, since, until, _, detach) => resolveClub(club, currentClub).flatMap(slug => api.postJson[StatsRequest, ClubJobResult]( "/api/jobs/stats", StatsRequest(slug, since, until) - ).flatMap(follower.handleClubSingle) + ).flatMap(result => + if (detach) { reportDetachedClub(result) } + else { follower.handleClubSingle(result) } + ) ) case CliCommand.Jobs(_, limit) => @@ -227,19 +230,40 @@ object Dispatcher { // `submitOne` posts a single-club job (the batch routes accept a one-element list) whose result `handleBatch` follows. // Near-free vs concurrent submission: the shared Chess.com client is gate-bound, so total wall-clock is ~unchanged. // + // With `detach` (#170) the follow is skipped: each club's job is submitted, its id printed, and the command returns — + // no log stream, no Ctrl-C-cancel. Reattach later with `ccas logs `. Serialization is irrelevant then (nothing is + // streamed), but the same per-club loop is reused so error handling and scoring stay identical. + // // Each club is best-effort: a submit / transport failure (a 4xx/5xx or a dropped connection for one club) is caught, // reported, and scored as a failure for that club WITHOUT aborting the rest — matching `handleBatch`'s handling of // per-club errors in a 200 body, and preserving the old batch design's "every club still runs" property. - private def followEachClub(follower: JobFollower, slugs: NonEmptyChunk[ClubSlug])( + private def followEachClub(follower: JobFollower, slugs: NonEmptyChunk[ClubSlug], detach: Boolean)( submitOne: ClubSlug => Task[List[ClubJobResult]] - ): Task[Int] = + ): Task[Int] = { + val handle: List[ClubJobResult] => Task[Int] = + if (detach) { results => + ZIO.foreach(results)(reportDetachedClub).map(codes => if (codes.forall(_ == 0)) { 0 } else { 1 }) + } else { follower.handleBatch } ZIO .foreach(slugs.toChunk.toList)(slug => submitOne(slug) - .flatMap(follower.handleBatch) + .flatMap(handle) .catchAll(e => Console.printLineError(s"${ClubSlug.unwrap(slug)}: ${rootMessage(e)}").orDie.as(1)) ) .map(codes => if (codes.forall(_ == 0)) { 0 } else { 1 }) + } + + // Detached submit of a single club-scoped job (#170): print its id (or its submit error) and DON'T follow. Cache the + // id for `ccas logs`/`ccas cancel` completion, and note the reattach command. Exit 1 for a per-club submit failure so + // a partial `--all --detach` batch still scores as failed overall (mirrors `handleClubSingle`'s error scoring). + private def reportDetachedClub(result: ClubJobResult): UIO[Int] = + (result.error, result.jobId) match { + case (Some(err), _) => Console.printLineError(s"${result.clubSlug}: $err").orDie.as(1) + case (_, Some(id)) => + CompletionCache.appendJob(id) *> + Console.printLine(s"${result.clubSlug} submitted (detached): $id — follow with 'ccas logs $id'").orDie.as(0) + case _ => Console.printLineError(s"${result.clubSlug}: server returned no job id").orDie.as(1) + } // Resolution lives in the pure, testable `ClubResolver`; the `--all` expansion's network call is injected here. private def resolveClub(explicit: Option[String], currentClub: Option[String]): IO[CliError, ClubSlug] = diff --git a/src/main/scala/ccas/cli/JobFollower.scala b/src/main/scala/ccas/cli/JobFollower.scala index 6c0b2a8..b820ad9 100644 --- a/src/main/scala/ccas/cli/JobFollower.scala +++ b/src/main/scala/ccas/cli/JobFollower.scala @@ -2,7 +2,7 @@ package ccas.cli import zio.* -import ccas.server.routes.JobRoutes.{ClubJobResult, JobResult, JobStatusResponse} +import ccas.server.routes.JobRoutes.{CancelResult, ClubJobResult, JobResult, JobStatusResponse} /** Drives a submitted job to completion by **following its log stream** (`GET /api/jobs/{id}/logs`), printing each * line as it arrives. The server holds the response open until the job is terminal and the tail reaches EOF, so the @@ -70,12 +70,32 @@ final class JobFollower( // Run the log follow (reconnecting across drops, bounded by `maxWait`) and return the raw outcome without printing any // terminal status: Some(Right)=EOF/terminal, Some(Left)=gave up after maxReconnects, None=timed out. `notice` sinks the // one-time reconnect message — stderr for the plain follow, `renderer.logLine` when bars are drawn. + // + // Ctrl-C during a follow cancels the server job (#170): the `.onInterrupt` is attached OUTSIDE `.timeout`, so a genuine + // `maxWait` expiry (which internally interrupts `follow` and returns `None` on the main path) does NOT trip it — only a + // real fiber interruption of this whole effect does. Both the plain and bars follow paths route through here, so the + // cancel hook covers both. It fires while the CLI's `Client` layer scope is still open (the follower runs inside it and + // finalizers unwind inner-to-outer), so the POST lands during teardown. Not scoped over `interpret`: once the stream + // reaches EOF the job is already terminal, so a Ctrl-C during the final status read has nothing to cancel. private def followResult( jobId: String, onLine: String => UIO[Unit], notice: String => UIO[Unit] ): Task[Option[Either[String, Unit]]] = - Ref.make(0).flatMap(printed => follow(jobId, onLine, printed, notice).timeout(maxWait)) + Ref.make(0).flatMap(printed => + follow(jobId, onLine, printed, notice).timeout(maxWait).onInterrupt(cancelOnInterrupt(jobId, notice)) + ) + + // Best-effort cancel of the followed job when the follow is interrupted (Ctrl-C). The notice routes through the + // follow's `notice` sink — the render-lock-safe `renderer.logLine` in bars mode, stderr otherwise — so it can't tear a + // live bar line (this finalizer runs BEFORE `withBars`' `renderer.clear`, which is the outer `.onExit`). It emits + // first so the operator sees it even if the POST is slow or fails; the POST itself is `.ignore`d — a 404 (the job + // already reached a terminal state before the interrupt landed) or a transport error during teardown must not turn a + // clean Ctrl-C into a stack trace. Cancellation is best-effort/asynchronous server-side (an in-flight blocking + // statement runs to completion first), so this requests the cancel and returns — it does not wait for `Cancelled`. + private def cancelOnInterrupt(jobId: String, notice: String => UIO[Unit]): UIO[Unit] = + notice(s"$jobId: cancelling on interrupt…") *> + api.postEmpty[CancelResult](s"/api/jobs/$jobId/cancel").ignore // Map the follow outcome to an exit code, printing the terminal status. In bars mode this runs AFTER the bars are // cleared (see `withBars`), so the message can't be jammed onto a live bar line. diff --git a/src/main/scala/ccas/server/jobs/JobRunner.scala b/src/main/scala/ccas/server/jobs/JobRunner.scala index 89dbacd..25b9203 100644 --- a/src/main/scala/ccas/server/jobs/JobRunner.scala +++ b/src/main/scala/ccas/server/jobs/JobRunner.scala @@ -149,14 +149,23 @@ object JobRunner { trigger: RunTrigger, effect: Option[JobRunId] => RIO[ProgressDisplay & ChessComClient & PostgresClient, Any] ): RIO[PostgresClient, JobRunId] = { - def conflictError = ConflictException( - s"A $kind job is already running" + clubId.fold("")(c => s" for club $c") - ) + // `cancelling` distinguishes a blocker that an operator has already asked to cancel (its id is in `cancelRequested`) + // but whose interrupt hasn't landed yet — an in-flight blocking statement runs to completion before the fiber + // flips to `Cancelled` and `release` clears it from the set. Without this, a `ccas ... ` immediately after a + // Ctrl-C cancel of the same (kind, club) reads the baffling "already running" for a job the operator just killed; + // the "finishing cancellation — retry in a moment" wording tells them it is self-resolving. (#170) + def conflictError(cancelling: Boolean): ConflictException = { + val forClub = clubId.fold("")(c => s" for club $c") + if (cancelling) { ConflictException(s"A $kind job$forClub is finishing cancellation — retry in a moment") } + else { ConflictException(s"A $kind job is already running$forClub") } + } (for { id <- withTransaction { for { existing <- JobRun.selectRunningForUpdate(kind, clubId) - _ <- ZIO.whenDiscard(existing.isDefined)(ZIO.fail(conflictError)) + _ <- ZIO.foreachDiscard(existing)(job => + cancelRequested.get.flatMap(pending => ZIO.fail(conflictError(pending.contains(job.id)))) + ) id = JobRunId.generate() now <- Clock.instant jobRun = JobRun(id, kind, clubId, trigger, JobRunStatus.Running, params, now, None, None) @@ -227,8 +236,9 @@ object JobRunner { } } yield id).catchSome { // Phantom-row race: two transactions both saw no running job, the unique partial index - // on (kind, COALESCE(club_id, -1)) WHERE status = 'Running' caught the second insert. - case e: SQLException if e.getSQLState == "23505" => ZIO.fail(conflictError) + // on (kind, COALESCE(club_id, -1)) WHERE status = 'Running' caught the second insert. A genuine concurrent + // submit (not a cancel-in-flight), so the plain "already running" wording is right. + case e: SQLException if e.getSQLState == "23505" => ZIO.fail(conflictError(cancelling = false)) } } diff --git a/src/test/scala/ccas/cli/TestCliParser.scala b/src/test/scala/ccas/cli/TestCliParser.scala index 2dde375..3e5e599 100644 --- a/src/test/scala/ccas/cli/TestCliParser.scala +++ b/src/test/scala/ccas/cli/TestCliParser.scala @@ -40,12 +40,12 @@ object TestCliParser extends ZIOSpecDefault { override def spec: Spec[Any, Any] = suite("TestCliParser")( test("membership parses comma-separated --club and defaults the server") { parsed("membership", "--club", "team-alpha,team-beta").map(c => - assertTrue(c.contains(CliCommand.Membership(DefaultServer, List("team-alpha", "team-beta"), false, None, false))) + assertTrue(c.contains(CliCommand.Membership(DefaultServer, List("team-alpha", "team-beta"), false, None, false, false))) ) }, test("membership --all parses with no explicit clubs") { parsed("membership", "--all").map(c => - assertTrue(c.contains(CliCommand.Membership(DefaultServer, Nil, true, None, false))) + assertTrue(c.contains(CliCommand.Membership(DefaultServer, Nil, true, None, false, false))) ) }, test("use-club parses the slug (local, no server)") { @@ -89,16 +89,32 @@ object TestCliParser extends ZIOSpecDefault { // No slug and no --club/--all now PARSES (empty clubs); the "no club" error is raised later by the Dispatcher // against the config's current_club, not at parse time. test("membership with no club parses to empty clubs") { - parsed("membership").map(c => assertTrue(c.contains(CliCommand.Membership(DefaultServer, Nil, false, None, false)))) + parsed("membership").map(c => assertTrue(c.contains(CliCommand.Membership(DefaultServer, Nil, false, None, false, false)))) }, test("--no-trust-usernames maps to Some(false)") { parsed("membership", "--no-trust-usernames", "--club", "team-alpha").map(c => - assertTrue(c.contains(CliCommand.Membership(DefaultServer, List("team-alpha"), false, Some(false), false))) + assertTrue(c.contains(CliCommand.Membership(DefaultServer, List("team-alpha"), false, Some(false), false, false))) ) }, test("--no-progress sets the flag on a follow command") { parsed("membership", "--no-progress", "--club", "team-alpha").map(c => - assertTrue(c.contains(CliCommand.Membership(DefaultServer, List("team-alpha"), false, None, true))) + assertTrue(c.contains(CliCommand.Membership(DefaultServer, List("team-alpha"), false, None, true, false))) + ) + }, + test("--detach sets the flag on membership/history/stats (#170)") { + for { + m <- parsed("membership", "--detach", "--club", "team-alpha") + h <- parsed("history", "--detach", "--club", "team-alpha") + s <- parsed("stats", "--detach", "--club", "team-alpha") + } yield assertTrue( + m.contains(CliCommand.Membership(DefaultServer, List("team-alpha"), false, None, false, true)), + h.contains(CliCommand.History(DefaultServer, List("team-alpha"), false, false, false, false, None, false, true)), + s.contains(CliCommand.Stats(DefaultServer, Some("team-alpha"), None, None, false, true)) + ) + }, + test("stats defaults --detach to false") { + parsed("stats", "--club", "team-alpha").map(c => + assertTrue(c.contains(CliCommand.Stats(DefaultServer, Some("team-alpha"), None, None, false, false))) ) }, test("recruit parses options, comma-separated source-clubs, and --club") { @@ -132,7 +148,7 @@ object TestCliParser extends ZIOSpecDefault { }, test("history flags parse") { parsed("history", "--full", "--include-finished", "--club", "team-alpha").map(c => - assertTrue(c.contains(CliCommand.History(DefaultServer, List("team-alpha"), false, true, true, false, None, false))) + assertTrue(c.contains(CliCommand.History(DefaultServer, List("team-alpha"), false, true, true, false, None, false, false))) ) }, test("blacklist add parses usernames and options") { diff --git a/src/test/scala/ccas/cli/TestJobFollower.scala b/src/test/scala/ccas/cli/TestJobFollower.scala index fc54450..0ef2a30 100644 --- a/src/test/scala/ccas/cli/TestJobFollower.scala +++ b/src/test/scala/ccas/cli/TestJobFollower.scala @@ -63,6 +63,29 @@ object TestJobFollower extends ZIOSpecDefault { override def streamProgress(path: String)(onFrame: ccas.utils.ProgressSnapshot => UIO[Unit]): Task[Unit] = progress } + /** Stub that records every `postEmpty` (the cancel POST) path and drives `streamLines` via `stream`, so an interrupt + * test can assert whether following a job fired the cancel-on-interrupt POST (#170). `getJson` replays a fixed status. + * `postEmpty` returns a valid `CancelResult` body decoded through the caller's own `Resp` decoder (the follower + * `.ignore`s it anyway), so no `CancelResult` codec is needed here. + */ + private final class CancelRecordingApi( + status: String, + cancelled: Ref[List[String]], + stream: (String => UIO[Unit]) => Task[Unit] + ) extends CcasApiClient { + override def getJson[Resp: JsonDecoder](path: String): Task[Resp] = + ZIO.fromEither(statusJson(status, None).fromJson[Resp]).mapError(m => CliError(s"stub decode failed: $m", 1)) + override def streamLines(path: String)(onLine: String => UIO[Unit]): Task[Unit] = stream(onLine) + override def streamProgress(path: String)(onFrame: ccas.utils.ProgressSnapshot => UIO[Unit]): Task[Unit] = ZIO.unit + override def postEmpty[Resp: JsonDecoder](path: String): Task[Resp] = + cancelled.update(path :: _) *> + ZIO.fromEither("""{"jobId":"job-1"}""".fromJson[Resp]).mapError(m => CliError(s"stub decode failed: $m", 1)) + override def postJson[Req: JsonEncoder, Resp: JsonDecoder](path: String, body: Req): Task[Resp] = + ZIO.die(new UnsupportedOperationException("postJson")) + override def postUnit[Req: JsonEncoder](path: String, body: Req): Task[Unit] = ZIO.unit + override def delete(path: String): Task[Unit] = ZIO.unit + } + private def statusJson(status: String, error: Option[String]): String = JobStatusResponse("job-1", "Membership", status, None, "2026-01-01T00:00:00Z", None, error, "Cli").toJson @@ -166,6 +189,34 @@ object TestJobFollower extends ZIOSpecDefault { err.exists(line => line.contains("ccas logs job-1") && line.contains("keeps running")) ) }, + test("an interrupt during a follow cancels the server job (#170)") { + for { + cancelled <- Ref.make(List.empty[String]) + started <- Promise.make[Nothing, Unit] + // The stream signals it is live, then hangs — so the follow is unambiguously inside `streamLines` (its + // onInterrupt installed) when we interrupt. `Running` status is irrelevant; interpret never runs. + api = new CancelRecordingApi("Running", cancelled, _ => started.succeed(()) *> ZIO.never) + fiber <- follower(api, 1.minute).followJob("job-1").fork + _ <- started.await + _ <- fiber.interrupt + recorded <- cancelled.get + } yield assertTrue(recorded.exists(p => p.contains("job-1") && p.contains("cancel"))) + }, + test("an interrupt during the post-follow recruit confirm does not cancel (#170)") { + for { + cancelled <- Ref.make(List.empty[String]) + inConfirm <- Promise.make[Nothing, Unit] + // The stream completes immediately (job terminal) so the follow returns before we interrupt; the interrupt lands + // in the `onComplete` (confirm) phase, which is OUTSIDE the follow's cancel-on-interrupt scope. No cancel fires. + api = new CancelRecordingApi("Completed", cancelled, _ => ZIO.unit) + fiber <- follower(api, 1.minute) + .handleRecruit("recruit", JobResult(Some("job-1"), None), logsToStderr = false, _ => inConfirm.succeed(()) *> ZIO.never) + .fork + _ <- inConfirm.await + _ <- fiber.interrupt + recorded <- cancelled.get + } yield assertTrue(recorded.isEmpty) + }, test("handleSingle short-circuits on a submission error") { for { follower <- followerWith("Completed", None, Nil) diff --git a/src/test/scala/ccas/server/jobs/TestJobRunner.scala b/src/test/scala/ccas/server/jobs/TestJobRunner.scala index 5465df3..76e9a5c 100644 --- a/src/test/scala/ccas/server/jobs/TestJobRunner.scala +++ b/src/test/scala/ccas/server/jobs/TestJobRunner.scala @@ -25,6 +25,7 @@ object TestJobRunner extends ZIOSpecDefault { testSubmitSucceeds, testSubmitRecordsFailed, testSubmitRejectsDuplicate, + testConflictWhileCancelInFlight, testConcurrentSubmitConflict, testSubmitAllowsDifferentClub, testSubmitAllowsDifferentKind, @@ -122,6 +123,38 @@ object TestJobRunner extends ZIOSpecDefault { ) } + // #170: a resubmit that races an in-flight cancel of the same (kind, club) must read the self-resolving "finishing + // cancellation — retry in a moment" wording, not the baffling "already running". The job parks in an UNINTERRUPTIBLE + // await so, after `cancel` registers it in `cancelRequested` and forks the interrupt, the interrupt stays pending: the + // row is still Running AND the id sits in `cancelRequested` — the exact window `submit` special-cases. Releasing the + // gate ends the uninterruptible region, the pending interrupt lands, and the job settles to Cancelled. + private def testConflictWhileCancelInFlight = + test("a resubmit racing an in-flight cancel gets the 'finishing cancellation' message") { + for { + _ <- deleteAllJobRuns + runner <- ZIO.service[JobRunner] + started <- Promise.make[Nothing, Unit] + gate <- Promise.make[Nothing, Unit] + id <- runner.submit( + JobKind.Membership, + Some(clubIdA), + None, + RunTrigger.Cli, + _ => (started.succeed(()) *> gate.await).uninterruptible + ) + _ <- started.await + _ <- runner.cancel(id) // registers id in cancelRequested; the forked interrupt can't land (uninterruptible) + result <- runner.submit(JobKind.Membership, Some(clubIdA), None, RunTrigger.Cli, _ => ZIO.unit).either + _ <- gate.succeed(()) // release the await → pending interrupt lands → job settles Cancelled + _ <- awaitStatus(runner, id) + } yield assertTrue( + result.left.exists { + case e: ConflictException => e.getMessage.contains("finishing cancellation") + case _ => false + } + ) + } + private def testConcurrentSubmitConflict = test("concurrent submits for same kind/club produce exactly one winner") { for { _ <- deleteAllJobRuns diff --git a/src/test/scala/ccas/server/routes/TestJobCancelWire.scala b/src/test/scala/ccas/server/routes/TestJobCancelWire.scala index 6e48c89..64cf81d 100644 --- a/src/test/scala/ccas/server/routes/TestJobCancelWire.scala +++ b/src/test/scala/ccas/server/routes/TestJobCancelWire.scala @@ -6,7 +6,7 @@ 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.cli.{CcasApiClient, CliError, JobFollower} import ccas.server.jobs.{JobKind, JobRun, JobRunner, JobRunStatus} import ccas.server.routes.JobRoutes.CancelResult import ccas.server.ServerTables @@ -26,7 +26,8 @@ object TestJobCancelWire extends ZIOSpecDefault { override def spec: Spec[Any, Throwable] = suite("TestJobCancelWire")( testCancelLiveJobOverHttp, - testCancelUnknownJobOverHttp + testCancelUnknownJobOverHttp, + testCancelViaFollowInterrupt ).provideShared( FreshSchemaLayer("test_cancel_wire", onInit = ServerTables.ensureTables), TestChessComClientSupport.dummyLayer, @@ -75,6 +76,35 @@ object TestJobCancelWire extends ZIOSpecDefault { ) } + // The #170 path: Ctrl-C during a `ccas logs`-style follow must cancel the server job. Drives the REAL `JobFollower` + // (the CLI's follow loop) against the real server+client, interrupts its fiber, and asserts the job reaches + // `Cancelled` — proving the follow's `.onInterrupt` cancel POST actually lands over real HTTP as the fiber tears down + // (something the stubbed `TestJobFollower` can't exercise). Readiness is deterministic: once the follow fiber is + // `Suspended` it has installed its onInterrupt hook and parked on the (silent) log stream, so the interrupt is + // guaranteed to land inside the cancel scope — no timing sleep. + private def testCancelViaFollowInterrupt = + test("interrupting a live follow cancels the server job over real HTTP") { + for { + runner <- ZIO.service[JobRunner] + port <- ZIO.service[ServerPort] + api <- apiFor(port) + started <- Promise.make[Nothing, Unit] + id <- runner.submit(JobKind.MatchRef, None, None, RunTrigger.Cli, _ => started.succeed(()) *> ZIO.never) + _ <- started.await + follower = JobFollower(api, maxWait = 60.seconds, reconnectBackoff = 1.second, maxReconnects = 1000, showProgress = false) + fiber <- follower.followJob(JobRunId.unwrap(id)).fork + _ <- (ZIO.sleep(20.millis) *> fiber.status) + .repeatUntil { case _: Fiber.Status.Suspended => true; case _ => false } + .timeoutFail(new Exception("follow never suspended"))(10.seconds) + _ <- fiber.interrupt + job <- awaitCancelled(runner, id) + } yield assertTrue( + 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 {