diff --git a/completions/ccas.bash b/completions/ccas.bash index 2dcae3d..a4c51c7 100644 --- a/completions/ccas.bash +++ b/completions/ccas.bash @@ -62,12 +62,12 @@ _ccas() { *) COMPREPLY=(); return ;; esac ;; use-club) pos="slug" ;; - membership) opts="--server --trust-usernames --no-trust-usernames --club --all" ;; - history) opts="--server --full --include-finished --refresh --refresh-min-hours --club --all" ;; - recruit) opts="--server --alias --target --cumulative --source-clubs --time-limit-minutes --explore --no-explore --club --stdout --report" ;; - stats) opts="--server --since --until --club" ;; + 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" ;; + 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" ;; jobs) opts="--server --limit" ;; - logs) opts="--server"; pos="jobid" ;; + logs) opts="--server --no-progress"; pos="jobid" ;; blacklist) case "$sub" in "") COMPREPLY=( $(compgen -W "add list remove --help" -- "$cur") ); return ;; diff --git a/src/main/scala/ccas/analysis/tables/AppSetting.scala b/src/main/scala/ccas/analysis/tables/AppSetting.scala index b7a35dd..b49385a 100644 --- a/src/main/scala/ccas/analysis/tables/AppSetting.scala +++ b/src/main/scala/ccas/analysis/tables/AppSetting.scala @@ -33,8 +33,16 @@ object AppSetting { /** Retention window (days) for per-job log files in `${JOB_LOGS_DIR}`, applied by `JobRunner.live` on startup. */ val JobLogRetentionDays: Key[Int] = Key("job_log_retention_days", 14, _.toIntOption, _.toString) + /** Sample spacing (milliseconds) for progress-bar frames on `GET /api/jobs/{id}/progress`. The server samples the + * merged bar state at this interval and de-duplicates, so a busy job's updates collapse to at most one encoded + + * sent frame per interval (an idle job sends none). Caps server work + follower traffic while still delivering the + * latest state within one interval. Default 100ms (10 fps): smooth for a bar, light on the server; floored at 16ms. + * Read per subscribe in `JobRunner`. + */ + val ProgressRefreshIntervalMillis: Key[Int] = Key("progress_refresh_interval_ms", 100, _.toIntOption, _.toString) + /** Every known key — for discoverability (e.g. a future `ccas settings list`). */ - val all: List[Key[?]] = List(CacheRetentionDays, JobLogRetentionDays) + val all: List[Key[?]] = List(CacheRetentionDays, JobLogRetentionDays, ProgressRefreshIntervalMillis) def createTable: ZIO[PostgresClient, SQLException, Int] = connectZIO { diff --git a/src/main/scala/ccas/cli/CcasApiClient.scala b/src/main/scala/ccas/cli/CcasApiClient.scala index 7189d70..a5c8477 100644 --- a/src/main/scala/ccas/cli/CcasApiClient.scala +++ b/src/main/scala/ccas/cli/CcasApiClient.scala @@ -6,6 +6,7 @@ import zio.json.* import zio.stream.{ZPipeline, ZStream} import ccas.server.jobs.JobLogStream +import ccas.utils.ProgressSnapshot import ccas.utils.errors.ErrorResponse /** Thin HTTP client to a local `CcasServer`. Every method fails with [[CliError]] (carrying an exit code) so the @@ -22,6 +23,12 @@ trait CcasApiClient { * follow a job's live log output; scoped internally so callers need no `Scope`. */ def streamLines(path: String)(onLine: String => UIO[Unit]): Task[Unit] + + /** Stream a job's `/progress` NDJSON endpoint, invoking `onFrame` for each decoded [[ProgressSnapshot]]. A line that + * fails to decode is skipped (never fails the stream) — bar rendering is best-effort. Transport errors propagate as + * for [[streamLines]]; the follow path treats them as non-fatal (bars just stop). + */ + def streamProgress(path: String)(onFrame: ProgressSnapshot => UIO[Unit]): Task[Unit] } object CcasApiClient { @@ -121,5 +128,10 @@ object CcasApiClient { } } } yield () + + // Each `/progress` line is one `ProgressSnapshot` JSON object; a decode failure (should never happen) is dropped so + // a malformed frame can't kill live bar rendering. `streamLines` already filters the keepalive frames. + override def streamProgress(path: String)(onFrame: ProgressSnapshot => UIO[Unit]): Task[Unit] = + streamLines(path)(line => ZIO.fromEither(line.fromJson[ProgressSnapshot]).foldZIO(_ => ZIO.unit, onFrame)) } } diff --git a/src/main/scala/ccas/cli/CliCommand.scala b/src/main/scala/ccas/cli/CliCommand.scala index 2357299..e77cb0b 100644 --- a/src/main/scala/ccas/cli/CliCommand.scala +++ b/src/main/scala/ccas/cli/CliCommand.scala @@ -49,8 +49,13 @@ object CliCommand { // Server commands — dispatched as HTTP calls by `Dispatcher`. Club-targeting fields hold the *parsed* request // (`clubs`/`club` may be empty/None); `Dispatcher` resolves them against `--all` and the config's `current_club`. - final case class Membership(server: String, clubs: List[String], all: Boolean, trustUsernames: Option[Boolean]) - extends ServerCommand + final case class Membership( + server: String, + clubs: List[String], + all: Boolean, + trustUsernames: Option[Boolean], + noProgress: Boolean + ) extends ServerCommand final case class History( server: String, clubs: List[String], @@ -58,7 +63,8 @@ object CliCommand { full: Boolean, includeFinished: Boolean, refresh: Boolean, - refreshMinHours: Option[Int] + refreshMinHours: Option[Int], + noProgress: Boolean ) extends ServerCommand final case class Recruit( server: String, @@ -71,12 +77,18 @@ object CliCommand { explore: Option[Boolean], stdout: Boolean, report: Boolean, - runId: Option[Int] + runId: Option[Int], + noProgress: Boolean + ) extends ServerCommand + final case class Stats( + server: String, + club: Option[String], + since: Option[String], + until: Option[String], + noProgress: Boolean ) extends ServerCommand - final case class Stats(server: String, club: Option[String], since: Option[String], until: Option[String]) - extends ServerCommand final case class Jobs(server: String, limit: Option[Int]) extends ServerCommand - final case class Logs(server: String, jobId: String) extends ServerCommand + final case class Logs(server: String, jobId: String, noProgress: Boolean) extends ServerCommand final case class BlacklistAdd( server: String, club: Option[String], @@ -126,6 +138,11 @@ object CliCommand { case _ => None } + // Opt out of live progress bars while following a job (plain log lines only). Bars otherwise render on an interactive + // terminal; a non-TTY (piped/redirected) already suppresses them regardless of this flag. + private val noProgressOpt: Options[Boolean] = + Options.boolean("no-progress") ?? "Don't render progress bars while following the job (plain log lines only)" + 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`. @@ -172,10 +189,10 @@ object CliCommand { .map(Use.apply) private def membership(default: String): Command[CliCommand] = - Command("membership", serverOpt(default) ++ trustOpt ++ clubsOpt ++ allOpt) + Command("membership", serverOpt(default) ++ trustOpt ++ clubsOpt ++ allOpt ++ noProgressOpt) .withHelp("Submit a membership-sync job (current club, --club a,b, or --all managed clubs)") - .map { case (server, trust, clubs, all) => - Membership(server, clubs, all, trust) + .map { case (server, trust, clubs, all, noProgress) => + Membership(server, clubs, all, trust, noProgress) } private def history(default: String): Command[CliCommand] = @@ -186,10 +203,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 + clubsOpt ++ allOpt ++ noProgressOpt ).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) => - History(server, clubs, all, full, includeFinished, refresh, refreshMinHours) + .map { case (server, full, includeFinished, refresh, refreshMinHours, clubs, all, noProgress) => + History(server, clubs, all, full, includeFinished, refresh, refreshMinHours, noProgress) } private val sourceClubsOpt: Options[List[String]] = @@ -207,14 +224,16 @@ object CliCommand { (intOpt("time-limit-minutes") ?? "Stop scouting after roughly N minutes") ++ exploreOpt ++ clubOpt ++ (Options.boolean("stdout") ?? "Print invited usernames (bare, newline-separated) to stdout for piping to a clipboard tool (e.g. wl-copy); logs go to stderr. Auto-confirms invites") ++ - (Options.boolean("report") ?? "Show a past run's invited usernames instead of scouting (the club's latest run, or the run id given as an argument)"), + (Options.boolean("report") ?? "Show a past run's invited usernames instead of scouting (the club's latest run, or the run id given as an argument)") ++ + noProgressOpt, (Args.integer("run-id") ?? "With --report, the run id to show (default: the club's latest run)") .atMost(1) .map(_.headOption.map(_.toInt)) ).withHelp("Submit a recruitment scouting job for a club (interactive runs confirm invites before marking them)") // Options and the optional `[run-id]` arg combine as (optionsTuple, argValue), so destructure the two levels. - .map { case ((server, alias, target, cumulative, sourceClubs, timeLimitMinutes, explore, club, stdout, report), runId) => - Recruit(server, club, alias, target, cumulative, sourceClubs, timeLimitMinutes, explore, stdout, report, runId) + .map { + case ((server, alias, target, cumulative, sourceClubs, timeLimitMinutes, explore, club, stdout, report, noProgress), runId) => + Recruit(server, club, alias, target, cumulative, sourceClubs, timeLimitMinutes, explore, stdout, report, runId, noProgress) } private def stats(default: String): Command[CliCommand] = @@ -223,9 +242,9 @@ 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 + clubOpt ++ noProgressOpt ).withHelp("Submit a club performance-stats job") - .map { case (server, since, until, club) => Stats(server, club, since, until) } + .map { case (server, since, until, club, noProgress) => Stats(server, club, since, until, noProgress) } private def jobs(default: String): Command[CliCommand] = Command("jobs", serverOpt(default) ++ (intOpt("limit") ?? "Maximum number of recent jobs to list")) @@ -233,9 +252,9 @@ object CliCommand { .map { case (server, limit) => Jobs(server, limit) } private def logs(default: String): Command[CliCommand] = - Command("logs", serverOpt(default), Args.text("jobId") ?? "Job run id to poll") + Command("logs", serverOpt(default) ++ noProgressOpt, Args.text("jobId") ?? "Job run id to poll") .withHelp("Poll a job's status and logs until it finishes") - .map { case (server, jobId) => Logs(server, jobId) } + .map { case ((server, noProgress), jobId) => Logs(server, jobId, noProgress) } private def blacklistAdd(default: String): Command[CliCommand] = Command( diff --git a/src/main/scala/ccas/cli/ClientProgressRenderer.scala b/src/main/scala/ccas/cli/ClientProgressRenderer.scala new file mode 100644 index 0000000..7aacce3 --- /dev/null +++ b/src/main/scala/ccas/cli/ClientProgressRenderer.scala @@ -0,0 +1,72 @@ +package ccas.cli + +import zio.{Ref, UIO, ZIO} + +import ccas.utils.{BarSnapshot, ProgressBar, ProgressDisplay, ProgressSnapshot} + +/** Renders the bar frames streamed from `GET /api/jobs/{id}/progress` onto a local [[ProgressDisplay]] while the CLI + * follows a job. Each frame is a full latest-wins snapshot, so [[render]] reconciles it against the bars currently + * shown: update a bar we already track (skipping the redraw when its raw fields are unchanged), add one that's new, + * finish one that dropped out of the frame. The server sends raw `current` / `total` / `text`; [[ProgressBar.print]] + * re-renders the block bar + percentage at this terminal's width (the server can't, it doesn't know the width). + * + * `bars` maps a server-assigned bar id to the local [[ProgressBar]] standing in for it plus the last snapshot rendered + * for it — the two id spaces are independent, so this indirection is what lets successive frames address the same bar, + * and the stored snapshot lets an unchanged bar skip a redundant full-display redraw. + */ +final class ClientProgressRenderer private[cli] ( + display: ProgressDisplay, + bars: Ref[Map[Int, ClientProgressRenderer.Tracked]] +) { + import ClientProgressRenderer.Tracked + + /** Reconcile one snapshot frame onto the display. */ + def render(frame: ProgressSnapshot): UIO[Unit] = + bars.get.flatMap { current => + val incomingIds = frame.bars.map(_.id).toSet + val stale = current.keySet -- incomingIds + ZIO.foreachDiscard(frame.bars)(upsert(current, _)) *> + ZIO.foreachDiscard(stale)(drop(current, _)) + } + + // Update the local bar for this snapshot's id, or create one if it's new. A frame carries every live bar even when + // only one changed, so skip the print (and its whole-display redraw) when the raw fields are identical to last time. + private def upsert(current: Map[Int, Tracked], snap: BarSnapshot): UIO[Unit] = + current.get(snap.id) match { + case Some(t) if t.last == snap => ZIO.unit + case Some(t) => + t.bar.print(snap.current, snap.total, snap.text) *> bars.update(_.updated(snap.id, Tracked(t.bar, snap))) + case None => + for { + bar <- display.addBar + _ <- bars.update(_ + (snap.id -> Tracked(bar, snap))) + _ <- bar.print(snap.current, snap.total, snap.text) + } yield () + } + + // A bar absent from the latest frame is finished (erased) and forgotten. + private def drop(current: Map[Int, Tracked], id: Int): UIO[Unit] = + ZIO.foreachDiscard(current.get(id))(_.bar.finish) *> bars.update(_ - id) + + /** Finish every remaining bar and forget them — called when the follow ends so the terminal is left clean (the + * `/progress` stream may close before a final empty frame arrives, leaving bars drawn). Idempotent. + */ + def clear: UIO[Unit] = + bars.getAndSet(Map.empty).flatMap(m => ZIO.foreachDiscard(m.values)(_.bar.finish)) + + /** Print a job-log line above the bars, interleaved cleanly via the display's render lock. */ + def logLine(line: String): UIO[Unit] = + display.logLineAboveBars(line) +} + +object ClientProgressRenderer { + + /** A tracked bar: the local [[ProgressBar]] plus the last [[BarSnapshot]] rendered onto it (for change detection). */ + final case class Tracked(bar: ProgressBar, last: BarSnapshot) + + /** Build a renderer over a fresh [[ProgressDisplay]] drawing to `System.out`. Only ever built when bars are wanted + * (interactive TTY), so the display is always enabled. + */ + def make: UIO[ClientProgressRenderer] = + Ref.make(Map.empty[Int, Tracked]).map(new ClientProgressRenderer(ProgressDisplay.make(enabled = true), _)) +} diff --git a/src/main/scala/ccas/cli/CompletionSpec.scala b/src/main/scala/ccas/cli/CompletionSpec.scala index dc98ee1..dce9107 100644 --- a/src/main/scala/ccas/cli/CompletionSpec.scala +++ b/src/main/scala/ccas/cli/CompletionSpec.scala @@ -50,28 +50,33 @@ object CompletionSpec { Leaf(List("use-club"), Nil, Nil, Slug), Leaf( List("membership"), - List(server, "--trust-usernames", "--no-trust-usernames", clubFlag, "--all"), + List(server, "--trust-usernames", "--no-trust-usernames", clubFlag, "--all", "--no-progress"), List(server, clubFlag), NoArgs ), Leaf( List("history"), - List(server, "--full", "--include-finished", "--refresh", "--refresh-min-hours", clubFlag, "--all"), + List(server, "--full", "--include-finished", "--refresh", "--refresh-min-hours", clubFlag, "--all", "--no-progress"), List(server, "--refresh-min-hours", clubFlag), NoArgs ), Leaf( List("recruit"), List(server, "--alias", "--target", "--cumulative", "--source-clubs", "--time-limit-minutes", "--explore", - "--no-explore", clubFlag, "--stdout", "--report"), + "--no-explore", clubFlag, "--stdout", "--report", "--no-progress"), List(server, "--alias", "--target", "--source-clubs", "--time-limit-minutes", clubFlag), // `--report`'s optional `[run-id]` is a bare numeric positional with nothing to complete, so NoArgs (suggest // nothing) is the right shell behavior. NoArgs ), - Leaf(List("stats"), List(server, "--since", "--until", clubFlag), List(server, "--since", "--until", clubFlag), NoArgs), + Leaf( + List("stats"), + List(server, "--since", "--until", clubFlag, "--no-progress"), + List(server, "--since", "--until", clubFlag), + NoArgs + ), Leaf(List("jobs"), List(server, "--limit"), List(server, "--limit"), NoArgs), - Leaf(List("logs"), List(server), List(server), JobId), + Leaf(List("logs"), List(server, "--no-progress"), List(server), JobId), Leaf( List("blacklist", "add"), List(server, "--reason", "--months", clubFlag), diff --git a/src/main/scala/ccas/cli/Dispatcher.scala b/src/main/scala/ccas/cli/Dispatcher.scala index fa74238..89c7775 100644 --- a/src/main/scala/ccas/cli/Dispatcher.scala +++ b/src/main/scala/ccas/cli/Dispatcher.scala @@ -40,8 +40,14 @@ object Dispatcher { CcasApiClient .live(cmd.server) .flatMap(api => - runCommand(api, JobFollower(api, MaxJobWait, ReconnectBackoff, MaxReconnects), cmd, currentClub) - .tap(_ => refreshClubsCache(api)) + // Bars render only on an interactive terminal (hasTty) and unless `--no-progress` was passed; piped/redirected + // output stays plain lines regardless. + runCommand( + api, + JobFollower(api, MaxJobWait, ReconnectBackoff, MaxReconnects, showProgress = hasTty && !noProgressFor(cmd)), + cmd, + currentClub + ).tap(_ => refreshClubsCache(api)) ) .provide(Client.default) .catchAll { @@ -56,6 +62,16 @@ object Dispatcher { private def rootMessage(e: Throwable): String = Option(e.getMessage).getOrElse(e.getClass.getSimpleName) + // Whether `--no-progress` was set — only the job-following commands carry the flag; the rest never draw bars. + private def noProgressFor(cmd: CliCommand.ServerCommand): Boolean = cmd match { + case c: CliCommand.Membership => c.noProgress + case c: CliCommand.History => c.noProgress + case c: CliCommand.Recruit => c.noProgress + case c: CliCommand.Stats => c.noProgress + case c: CliCommand.Logs => c.noProgress + case _ => false + } + // Best-effort, staleness-gated refresh of the completion club-slug cache after a successful command. Fully ignored: // it never blocks the result or alters the exit code (the `.tap` runs only on the success channel). private def refreshClubsCache(api: CcasApiClient): UIO[Unit] = @@ -75,23 +91,27 @@ object Dispatcher { cmd: CliCommand.ServerCommand, currentClub: Option[String] ): Task[Int] = cmd match { - case CliCommand.Membership(_, clubs, all, trust) => + case CliCommand.Membership(_, clubs, all, trust, _) => resolveClubs(api, clubs, all, currentClub).flatMap(slugs => - api.postJson[MembershipRequest, List[ClubJobResult]]( - "/api/jobs/membership", - MembershipRequest(slugs, trust) - ).flatMap(follower.handleBatch) + followEachClub(follower, slugs)(slug => + api.postJson[MembershipRequest, List[ClubJobResult]]( + "/api/jobs/membership", + MembershipRequest(NonEmptyChunk.single(slug), trust) + ) + ) ) - case CliCommand.History(_, clubs, all, full, includeFinished, refresh, refreshMinHours) => + case CliCommand.History(_, clubs, all, full, includeFinished, refresh, refreshMinHours, _) => resolveClubs(api, clubs, all, currentClub).flatMap(slugs => - api.postJson[HistoryRequest, List[ClubJobResult]]( - "/api/jobs/history", - HistoryRequest(slugs, flag(full), flag(includeFinished), flag(refresh), refreshMinHours) - ).flatMap(follower.handleBatch) + followEachClub(follower, slugs)(slug => + api.postJson[HistoryRequest, List[ClubJobResult]]( + "/api/jobs/history", + HistoryRequest(NonEmptyChunk.single(slug), flag(full), flag(includeFinished), flag(refresh), refreshMinHours) + ) + ) ) - case CliCommand.Recruit(_, club, alias, target, cumulative, sourceClubs, timeLimitMinutes, explore, stdout, report, runId) => + case CliCommand.Recruit(_, club, alias, target, cumulative, sourceClubs, timeLimitMinutes, explore, stdout, report, runId, _) => // Guard the read-only flag: a run-id argument without `--report` would otherwise be silently dropped and launch // a fresh scout — and with `--stdout` it would auto-confirm that scout's invites. Fail before submitting. if (runId.isDefined && !report) { ZIO.fail(CliError("a run id can only be given with --report", 2)) } @@ -121,7 +141,7 @@ object Dispatcher { ) } - case CliCommand.Stats(_, club, since, until) => + case CliCommand.Stats(_, club, since, until, _) => resolveClub(club, currentClub).flatMap(slug => api.postJson[StatsRequest, ClubJobResult]( "/api/jobs/stats", @@ -132,7 +152,7 @@ object Dispatcher { case CliCommand.Jobs(_, limit) => api.getJson[List[JobStatusResponse]]("/api/jobs").flatMap(all => printJobs(limit.fold(all)(all.take)).as(0)) - case CliCommand.Logs(_, jobId) => + case CliCommand.Logs(_, jobId, _) => follower.followJob(jobId) case CliCommand.BlacklistAdd(_, club, usernames, reason, months) => @@ -194,6 +214,26 @@ object Dispatcher { ZIO.foreachDiscard(clubs)(c => Console.printLine(s"${c.slug} ${c.name} marked=${c.markedAt}").orDie) } + // Run each club through submit-then-follow ONE at a time, so a multi-club (`--all`) run streams each club's logs and + // bars live in turn rather than following the first while the rest run blind and dump their logs at the end. The next + // club isn't submitted until the current one's follow completes; overall exit is 0 only if every club succeeded. Each + // `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. + // + // 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])( + submitOne: ClubSlug => Task[List[ClubJobResult]] + ): Task[Int] = + ZIO + .foreach(slugs.toChunk.toList)(slug => + submitOne(slug) + .flatMap(follower.handleBatch) + .catchAll(e => Console.printLineError(s"${ClubSlug.unwrap(slug)}: ${rootMessage(e)}").orDie.as(1)) + ) + .map(codes => if (codes.forall(_ == 0)) { 0 } else { 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] = ClubResolver.single(explicit, currentClub) diff --git a/src/main/scala/ccas/cli/JobFollower.scala b/src/main/scala/ccas/cli/JobFollower.scala index 8f5a84a..f3919dd 100644 --- a/src/main/scala/ccas/cli/JobFollower.scala +++ b/src/main/scala/ccas/cli/JobFollower.scala @@ -18,27 +18,87 @@ import ccas.server.routes.JobRoutes.{ClubJobResult, JobResult, JobStatusResponse * generous values. Exit codes: 0 when the job reaches `Completed`, 1 when it reaches `Failed`, could not be submitted, * did not finish within `maxWait`, or the follow was lost past `maxReconnects` reconnects. */ -final class JobFollower(api: CcasApiClient, maxWait: Duration, reconnectBackoff: Duration, maxReconnects: Int) { +final class JobFollower( + api: CcasApiClient, + maxWait: Duration, + reconnectBackoff: Duration, + maxReconnects: Int, + showProgress: Boolean +) { def followJob(jobId: String): Task[Int] = - followJob(jobId, line => Console.printLine(line).orDie) - - // `onLine` sinks each streamed log line; `--stdout` routes it to stderr so stdout carries only the username payload. - private def followJob(jobId: String, onLine: String => UIO[Unit]): Task[Int] = - Ref.make(0).flatMap { printed => - follow(jobId, onLine, printed).timeout(maxWait).flatMap { - case Some(Right(())) => finalExitCode(jobId) - case Some(Left(cause)) => streamLost(jobId, cause).as(1) - case None => stillRunning(jobId).as(1) + followWith(jobId, line => Console.printLine(line).orDie, bars = showProgress) + + // Follow a job, optionally rendering its live progress bars above the streamed log lines. When not `bars`, log lines + // go straight to `sink` (the pre-bars behaviour: `--stdout` routes to stderr, everything else stdout) and the reconnect + // notice to stderr. When `bars`, delegate to `withBars`. `interpret` maps the raw follow outcome to an exit code. + private def followWith(jobId: String, sink: String => UIO[Unit], bars: Boolean): Task[Int] = + if (!bars) { followResult(jobId, sink, stderrNotice).flatMap(interpret(jobId, _)) } + else { withBars(jobId) } + + // Bars branch: draw progress bars from `/api/jobs/{id}/progress` above the log lines. The progress consumer runs on a + // scope-owned fiber (`forkScoped`, so it can't leak in the window between fork and cleanup registration) and + // auto-reconnects across the server's ~60s read-idle drop (#161) just like the log follow — otherwise the bars would + // freeze partway through a long job. `onExit` interrupts that fiber and clears the bars BEFORE `interpret` prints any + // terminal status, so the final message lands on a quiescent terminal (not jammed onto a bar line). The one-time + // reconnect notice for the *log* follow routes through `renderer.logLine` (the render lock), so it can't tear a live + // bar redraw either. Both log lines and bars write through the one display's lock, so they never interleave mid-line. + private def withBars(jobId: String): Task[Int] = + ZIO.scoped { + for { + renderer <- ClientProgressRenderer.make + progress <- consumeProgress(jobId, renderer).forkScoped + result <- followResult(jobId, renderer.logLine, renderer.logLine) + .onExit(_ => progress.interrupt *> renderer.clear) + code <- interpret(jobId, result) + } yield code + } + + // Consume the progress stream, reconnecting on a mid-stream drop — the same ~60s read-idle reap the log follow handles + // (#161) — so bars stay live for the whole job. Frames are latest-wins, so a reconnect needs no replay: the next frame + // re-syncs every bar. A clean end (the server closes the stream at job completion) or any non-drop error stops without + // retrying; the parallel log follow surfaces any real failure. Bounded by this fiber's scope (interrupted in `withBars` + // once the follow ends), so the reconnect loop can't outlive the job. + private def consumeProgress(jobId: String, renderer: ClientProgressRenderer): UIO[Unit] = + api + .streamProgress(s"/api/jobs/$jobId/progress")(renderer.render) + .catchAll { + case StreamDropped(_) => ZIO.sleep(reconnectBackoff) *> consumeProgress(jobId, renderer) + case _ => ZIO.unit } + + // 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. + 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)) + + // 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. + private def interpret(jobId: String, result: Option[Either[String, Unit]]): Task[Int] = + result match { + case Some(Right(())) => finalExitCode(jobId) + case Some(Left(cause)) => streamLost(jobId, cause).as(1) + case None => stillRunning(jobId).as(1) } + private val stderrNotice: String => UIO[Unit] = line => Console.printLineError(line).orDie + /** Follow the log stream, auto-reconnecting across a mid-stream drop (#161). Each (re)connect replays the log from * the start of the file (`FileTail.subscribe` begins at offset 0), so [[skipReplay]] drops the lines already shown * and only genuinely new lines print. `Right(())` = the stream reached EOF (the job is terminal); `Left(cause)` = we * gave up after `maxReconnects` drops. A non-drop error (e.g. a 404 for a GC'd job) propagates unchanged. */ - private def follow(jobId: String, onLine: String => UIO[Unit], printed: Ref[Int]): Task[Either[String, Unit]] = { + private def follow( + jobId: String, + onLine: String => UIO[Unit], + printed: Ref[Int], + notice: String => UIO[Unit] + ): Task[Either[String, Unit]] = { def attempt(reconnects: Int): Task[Either[String, Unit]] = for { startPrinted <- printed.get @@ -47,7 +107,7 @@ final class JobFollower(api: CcasApiClient, maxWait: Duration, reconnectBackoff: .streamLines(s"/api/jobs/$jobId/logs")(skipReplay(onLine, printed, seen, startPrinted)) .foldZIO( { - case StreamDropped(cause) => afterDrop(jobId, cause, reconnects, () => attempt(reconnects + 1)) + case StreamDropped(cause) => afterDrop(jobId, cause, reconnects, notice, () => attempt(reconnects + 1)) case other => ZIO.fail(other) }, _ => ZIO.succeed(Right(())) @@ -79,22 +139,21 @@ final class JobFollower(api: CcasApiClient, maxWait: Duration, reconnectBackoff: jobId: String, cause: String, reconnects: Int, + notice: String => UIO[Unit], retry: () => Task[Either[String, Unit]] ): Task[Either[String, Unit]] = if (reconnects >= maxReconnects) { ZIO.succeed(Left(cause)) } else { for { - _ <- ZIO.whenDiscard(reconnects == 0)(reconnectNotice(jobId)) + _ <- ZIO.whenDiscard(reconnects == 0)(notice(reconnectMessage(jobId))) _ <- ZIO.sleep(reconnectBackoff) result <- retry() } yield result } - private def reconnectNotice(jobId: String): UIO[Unit] = - Console - .printLineError(s"$jobId: log stream dropped on a silent phase — reconnecting (the job keeps running)…") - .orDie + private def reconnectMessage(jobId: String): String = + s"$jobId: log stream dropped on a silent phase — reconnecting (the job keeps running)…" private def streamLost(jobId: String, cause: String): UIO[Unit] = Console @@ -150,10 +209,12 @@ final class JobFollower(api: CcasApiClient, maxWait: Duration, reconnectBackoff: val onLine: String => UIO[Unit] = if (logsToStderr) { line => Console.printLineError(line).orDie } else { line => Console.printLine(line).orDie } + // `--stdout` (logsToStderr) keeps stdout clean for the username payload, so no bars there; an interactive run + // gets bars when the terminal supports them. for { _ <- CompletionCache.appendJob(id) _ <- notice - code <- followJob(id, onLine) + code <- followWith(id, onLine, bars = showProgress && !logsToStderr) _ <- ZIO.whenDiscard(code == 0)(onComplete(id)) } yield code case _ => Console.printLineError(s"$label: server returned no job id").orDie.as(1) diff --git a/src/main/scala/ccas/server/jobs/JobRunner.scala b/src/main/scala/ccas/server/jobs/JobRunner.scala index a9cfbfb..c97079f 100644 --- a/src/main/scala/ccas/server/jobs/JobRunner.scala +++ b/src/main/scala/ccas/server/jobs/JobRunner.scala @@ -8,14 +8,15 @@ import com.typesafe.config.ConfigFactory import ccas.utils.sql.PostgresClient import ccas.utils.sql.PostgresClient.withTransaction -import zio.stream.ZStream -import zio.{Clock, Duration, Promise, RIO, RLayer, Ref, Scope, UIO, ZIO, ZLayer, durationInt} +import zio.json.EncoderOps +import zio.stream.{SubscriptionRef, ZStream} +import zio.{Clock, Duration, 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} import ccas.utils.client.ChessComClient import ccas.utils.errors.{ConflictException, safeMessage} -import ccas.utils.{JobLogSink, ProgressDisplay} +import ccas.utils.{BarSnapshot, JobLogSink, ProgressDisplay, ProgressSnapshot} /** Asynchronous job executor that runs analysis tasks as forked fibers. * @@ -47,6 +48,12 @@ trait JobRunner { * per-job log file live and closes once the job is terminal and the tail has reached EOF. */ def logStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] + + /** Stream a job's live progress as latest-wins [[ccas.utils.ProgressSnapshot]] JSON frames (one per line), or `None` + * if no such job exists. Each frame merges the job's own app bars with the shared client's global API gauge. The + * stream closes when the job is terminal — bars are ephemeral, so nothing is persisted or replayed. + */ + def progressStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] } object JobRunner { @@ -54,6 +61,10 @@ object JobRunner { // How often the file-tail transport re-polls a job's log file for newly-appended lines while the job is running. private val LogTailPollInterval: Duration = 250.millis + // Floor for the `/progress` sample interval (see `AppSetting.ProgressRefreshIntervalMillis`) — guards a mis-set + // 0/negative value from busy-looping the sampler. ~60 fps, far tighter than any sensible configured cap. + private val MinRefreshInterval: Duration = 16.millis + val live: RLayer[ProgressDisplay & ChessComClient & PostgresClient, JobRunner] = ZLayer.scoped { for { @@ -61,6 +72,9 @@ object JobRunner { client <- ZIO.service[ChessComClient] pgClient <- ZIO.service[PostgresClient] completions <- Ref.make(Map.empty[JobRunId, Promise[Nothing, Unit]]) + // 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]) // 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. @@ -78,7 +92,7 @@ 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, layerScope, logDir) + } yield new JobRunnerLive(display, client, pgClient, completions, jobChannels, layerScope, logDir) } private class JobRunnerLive( @@ -86,6 +100,7 @@ object JobRunner { client: ChessComClient, pgClient: PostgresClient, completions: Ref[Map[JobRunId, Promise[Nothing, Unit]]], + jobChannels: Ref[Map[JobRunId, ProgressDisplay.BarChannel]], layerScope: Scope, logDir: Path ) extends JobRunner { @@ -133,26 +148,37 @@ object JobRunner { for { sink <- FileSink.make(logDir, JobRunId.unwrap(id)) promise <- Promise.make[Nothing, Unit] + // The per-job bar channel: app bars created inside this job (via `currentChannel.locally` below) publish + // here; `progressStream` merges it with the global API gauge to serve this job's `/progress`. + channel <- SubscriptionRef.make(Map.empty[Int, BarSnapshot]) // Runs in the CHILD fiber when the job ends (success, failure, or interruption): close the sink (flush + // close the held-open writer) FIRST, then fire the completion promise, then de-register — so a tailer // 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. release = for { _ <- sink.close() _ <- promise.succeed(()) _ <- completions.update(_ - id) + _ <- jobChannels.update(_ - id) } yield () _ <- completions.update(_ + (id -> promise)) + _ <- jobChannels.update(_ + (id -> channel)) _ <- 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 // without this, HTTP-submitted jobs logged to the default logger and their per-job file stayed empty (#132). + // `currentChannel.locally` scopes bar publishing to this job's channel for the whole run (forked children + // inherit the FiberRef), mirroring `currentSink` for log lines. display.installLogger( ProgressDisplay.sourced(label)( - JobLogSink.currentSink - .locally(sink)( - runJob(id, effect(Some(id))).ensuring(release) - ) + ProgressDisplay.currentChannel.locally(Some(channel))( + JobLogSink.currentSink + .locally(sink)( + runJob(id, effect(Some(id))).ensuring(release) + ) + ) ) ) ).forkIn(layerScope) @@ -195,5 +221,52 @@ object JobRunner { override def logStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = status(id).map(_.map(_ => transport.subscribe(id))) + + override def progressStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = + status(id).flatMap { + case None => ZIO.none + case Some(_) => + for { + chans <- jobChannels.get + comps <- completions.get + // The refresh cap is a non-essential tuning knob for best-effort bars — a failed settings read (transient DB + // error) falls back to the compiled-in default rather than failing the `/progress` subscribe. + ms <- AppSetting + .get(AppSetting.ProgressRefreshIntervalMillis) + .orElseSucceed(AppSetting.ProgressRefreshIntervalMillis.default) + } yield Some(progressFrames(chans.get(id), comps.get(id), ms.millis)) + } + + // Build the `/progress` frame stream by SAMPLING the merged bar state (the shared global gauge channel plus this + // job's channel — later keys win; a job never shares a bar id with the gauge, so the union is disjoint) at + // `refreshInterval` and de-duplicating consecutive-identical samples. Sampling + `changes` gives conflation: a busy + // job's rapid updates collapse to at most one frame per interval (bounding the encode + send the operator asked to + // cap), an idle job emits nothing (consecutive samples are equal), and — unlike dropping excess — the LATEST state + // is always delivered within one interval, so a bar that ticks once then goes quiet still reaches the follower. + // Frames render bar-id-ordered (creation order: gauge first, then app bars). The stream ends when the job completes; + // a terminal job (no live channel / promise) emits a single settling frame and closes. + private def progressFrames( + jobChannel: Option[ProgressDisplay.BarChannel], + completion: Option[Promise[Nothing, Unit]], + refreshInterval: Duration + ): ZStream[Any, Throwable, String] = { + val readState: UIO[Map[Int, BarSnapshot]] = + for { + global <- display.globalBarSnapshot + job <- jobChannel.fold(ZIO.succeed(Map.empty[Int, BarSnapshot]))(_.get) + } yield global ++ job + // Floor the interval so a mis-set 0/negative can't busy-loop the sampler. + val interval = if (refreshInterval.toNanos > MinRefreshInterval.toNanos) { refreshInterval } else { MinRefreshInterval } + val frames = + ZStream + .fromZIO(readState) + .repeat(Schedule.spaced(interval)) + .changes + .map(m => ProgressSnapshot(m.values.toList.sortBy(_.id)).toJson) + completion match { + case Some(p) => frames.interruptWhen(p.await) + case None => frames.take(1) + } + } } } diff --git a/src/main/scala/ccas/server/routes/JobRoutes.scala b/src/main/scala/ccas/server/routes/JobRoutes.scala index 981f003..4a03552 100644 --- a/src/main/scala/ccas/server/routes/JobRoutes.scala +++ b/src/main/scala/ccas/server/routes/JobRoutes.scala @@ -288,6 +288,27 @@ object JobRoutes { ) }).pipe(withErrorHandling) }, + // Chunked NDJSON stream of a job's live progress: one `ProgressSnapshot` per line (latest-wins), merging the job's + // app bars with the shared client's API gauge. Live-only — closes when the job is terminal. The following CLI opens + // this only when it wants bars (interactive TTY, not `--no-progress`); it never affects the `/logs` follow. + Method.GET / "api" / "jobs" / string("jobId") / "progress" -> handler { (jobId: String, _: Request) => + (for { + runner <- ZIO.service[JobRunner] + streamOpt <- runner.progressStream(JobRunId.wrap(jobId)) + } yield streamOpt match { + case None => Response.text(s"Job $jobId not found").status(Status.NotFound) + case Some(frames) => + Response( + status = Status.Ok, + headers = Headers(Header.ContentType(MediaType.text.`plain`, charset = Some(StandardCharsets.UTF_8))), + // Same keepalive tick as `/logs`: a job phase with no bar changes for >50s must not idle the follower shut. + body = Body.fromCharSequenceStreamChunked( + JobLogStream.withKeepAlive(frames).map(_ + "\n"), + StandardCharsets.UTF_8 + ) + ) + }).pipe(withErrorHandling) + }, // Invited usernames for the recruitment run linked to a job — the paste-ready payload the CLI fetches once the // job is terminal (the `ccas recruit --stdout` auto-confirm path). 404 if the job id has no recruitment run. // Scope is THIS run only, deliberately: a `--cumulative` top-up returns just its new invites so the operator diff --git a/src/main/scala/ccas/utils/ProgressBar.scala b/src/main/scala/ccas/utils/ProgressBar.scala index 68a4b3e..7dbe34c 100644 --- a/src/main/scala/ccas/utils/ProgressBar.scala +++ b/src/main/scala/ccas/utils/ProgressBar.scala @@ -23,7 +23,9 @@ class ProgressBar private[utils] (id: Int, display: ProgressDisplay) { val bar = "\u2588" * filled + "\u2591" * (20 - filled) val full = f"$text $bar $pct%.1f%%" val lineCount = full.count(_ == '\n') + 1 - display.render(id, full, lineCount) + // Pass both the rendered line (for the local terminal draw) and the raw fields (mirrored to this bar's channel as + // an unrendered `BarSnapshot`, so a following CLI re-renders the block bar at its own terminal width). + display.render(id, full, lineCount, current, total, text) } /** Remove this bar from the display. No-op if already removed. */ diff --git a/src/main/scala/ccas/utils/ProgressDisplay.scala b/src/main/scala/ccas/utils/ProgressDisplay.scala index bab9ee8..803014a 100644 --- a/src/main/scala/ccas/utils/ProgressDisplay.scala +++ b/src/main/scala/ccas/utils/ProgressDisplay.scala @@ -1,14 +1,16 @@ package ccas.utils import java.io.PrintStream +import java.nio.charset.StandardCharsets import java.time.format.DateTimeFormatter import java.time.{Instant, ZoneId} import java.util.concurrent.atomic.AtomicInteger import java.util.regex.Pattern -import zio.{Cause, FiberId, FiberRef, FiberRefs, LogLevel, LogSpan, Ref, Runtime, Scope, Trace, UIO, URIO, URLayer, ZIO, ZLayer, ZLogger} +import zio.stream.SubscriptionRef +import zio.{Cause, FiberId, FiberRef, FiberRefs, LogLevel, LogSpan, Ref, Runtime, Scope, Trace, UIO, Unsafe, URIO, URLayer, ZIO, ZLayer, ZLogger} -/** Manages progress bars rendered as a single line on stdout, overwritten in-place via `\r`. +/** Manages progress bars rendered one-per-line on stdout; a redraw moves the cursor up over the block and repaints it. * * State (`bars`) is guarded by a Java intrinsic monitor (`lock`) so the synchronous `ZLogger` callback installed by * `ProgressDisplay.live` and the ZIO-effect entry points (`render`, `removeBar`, `finishAllSync`) all serialise their @@ -32,23 +34,44 @@ import zio.{Cause, FiberId, FiberRef, FiberRefs, LogLevel, LogSpan, Ref, Runtime final class ProgressDisplay private[utils] ( private val enabled: Boolean, private val out: PrintStream, - private val err: PrintStream + private val err: PrintStream, + private val globalChannel: Option[ProgressDisplay.BarChannel] ) { private val lock = new Object private var bars: List[ProgressDisplay.BarState] = Nil private val idGen = new AtomicInteger(0) + // Physical terminal lines the last draw occupied — so a redraw moves the cursor back to the top of the bar block and + // repaints it. Only touched inside `lock.synchronized`. + private var lastDrawnLines: Int = 0 + + /** Current snapshot of the process-wide global bar channel — the bars created outside any job's + * [[ProgressDisplay.currentChannel]] scope (e.g. the shared `ChessComClient` API gauge). `progressStream` samples this + * (merged with a job's own channel) to build each `/progress` frame so the gauge shows alongside a job's app bars. + * Read-only (`.get`), so no caller can mutate the shared gauge state; empty when the display has no global channel + * (the sync test factory / server console). + */ + private[ccas] def globalBarSnapshot: UIO[Map[Int, BarSnapshot]] = + globalChannel.fold(ZIO.succeed(Map.empty[Int, BarSnapshot]))(_.get) // --------------------------------------------------------------------------- // Bar lifecycle // --------------------------------------------------------------------------- - /** Create a new progress bar appended to the bottom of the display. */ - def addBar: UIO[ProgressBar] = ZIO.succeed { - val id = idGen.getAndIncrement() - lock.synchronized { bars = bars :+ ProgressDisplay.BarState(id, 0, "") } - new ProgressBar(id, this) - } + /** Create a new progress bar appended to the bottom of the display. + * + * The bar's publish target is captured **now**, from the creating fiber's [[ProgressDisplay.currentChannel]] (a job's + * channel when created inside a job effect; otherwise the shared [[globalChannel]]). Capturing at creation — not at + * each `render` — keeps a bar's snapshots flowing to one channel for its whole life, even though later updates may run + * on other fibers (e.g. the API gauge is created outside any job but updated from job fibers). + */ + def addBar: UIO[ProgressBar] = + ProgressDisplay.currentChannel.get.map { jobChannel => + val id = idGen.getAndIncrement() + val target = jobChannel.orElse(globalChannel) + lock.synchronized { bars = bars :+ ProgressDisplay.BarState(id, 0, "", 0, 0, "", target) } + new ProgressBar(id, this) + } /** Create a scoped progress bar — automatically removed when the scope closes. */ def addBarScoped: ZIO[Scope, Nothing, ProgressBar] = @@ -61,29 +84,59 @@ final class ProgressDisplay private[utils] ( // `addBarScoped` finalisers stay consistent (otherwise disabled-mode bars would leak in `bars`). // --------------------------------------------------------------------------- - /** Update one bar's output and re-render the whole display. */ - private[utils] def render(barId: Int, output: String, lineCount: Int): UIO[Unit] = ZIO.succeed { - lock.synchronized { - bars = bars.map(b => - if (b.id == barId) ProgressDisplay.BarState(barId, lineCount, output) else b - ) - drawAllSync() + /** Update one bar's output and re-render the whole display, then republish the bar's channel from `bars`. + * + * `output` is the already-rendered line (block bar + percentage) used for the local terminal draw; the raw + * `current` / `total` / `text` are stored so [[publishChannel]] can emit an unrendered [[BarSnapshot]] (the following + * CLI re-renders it at its own width). Publishing runs outside the terminal lock (a `SubscriptionRef` write is an + * effect) but re-reads `bars` under the lock at publish time, so it can't diverge from local state. + */ + private[utils] def render(barId: Int, output: String, lineCount: Int, current: Int, total: Int, text: String): UIO[Unit] = + ZIO.succeed { + lock.synchronized { + val target = bars.find(_.id == barId).flatMap(_.target) + bars = bars.map(b => + if (b.id == barId) b.copy(lineCount = lineCount, lastOutput = output, current = current, total = total, text = text) + else b + ) + clearBlockSync() + drawAllSync() + target + } + }.flatMap(publishChannel) + + /** Remove a bar from the display, then republish its channel from `bars` (the bar is now gone, so it drops out). */ + private[utils] def removeBar(barId: Int): UIO[Unit] = + ZIO.succeed { + lock.synchronized { + val target = bars.find(_.id == barId).flatMap(_.target) + clearBlockSync() + bars = bars.filterNot(_.id == barId) + drawAllSync() + target + } + }.flatMap(publishChannel) + + // Republish `target`'s snapshot by re-deriving it from `bars` (the lock-guarded source of truth) at publish time and + // `set`-ing the whole map — NOT an incremental delta. So even when concurrent render/removeBar publishes for the same + // channel reorder relative to their locked mutations, the last publish to run re-reads the final `bars` and the + // channel converges to it (no sticky ghost bar). No-op when the bar has no channel (sync `make` / server console). + private def publishChannel(target: Option[ProgressDisplay.BarChannel]): UIO[Unit] = + ZIO.foreachDiscard(target) { channel => + ZIO.succeed { + lock.synchronized { + bars.collect { + case b if b.lineCount > 0 && b.target.exists(_ eq channel) => + b.id -> BarSnapshot(b.id, b.current, b.total, b.text) + }.toMap + } + }.flatMap(channel.set) } - } - - /** Remove a bar from the display. */ - private[utils] def removeBar(barId: Int): UIO[Unit] = ZIO.succeed { - lock.synchronized { - clearLineSync() - bars = bars.filterNot(_.id == barId) - drawAllSync() - } - } /** Finish all bars — erase without redraw. Called from the live layer's release block. */ private[utils] def finishAllSync(): Unit = lock.synchronized { - clearLineSync() + clearBlockSync() bars = Nil } @@ -107,12 +160,27 @@ final class ProgressDisplay private[utils] ( private[utils] def logAboveBarsSync(sink: JobLogSink, msg: String): Unit = { sink.writeFileSync(msg) lock.synchronized { - clearLineSync() + clearBlockSync() sink.writeConsoleSync(msg) drawAllSync() } } + /** Print a log line above the bars on this display's own `out` stream — the CLI-follow analog of [[logAboveBarsSync]] + * (which routes through a [[JobLogSink]]). The CLI `JobFollower` uses this to interleave each streamed job-log line + * above the bars it renders from `/api/jobs/{id}/progress`. Clears the bar block, prints the line, redraws — all under + * the render lock, so it can't tear against a concurrent bar [[render]] on the same display. When `enabled` is false + * (bars suppressed / non-TTY), it degrades to a plain `out.println(line)`. + */ + private[ccas] def logLineAboveBars(line: String): UIO[Unit] = + ZIO.succeed { + lock.synchronized { + clearBlockSync() + out.println(line) + drawAllSync() + } + } + // --------------------------------------------------------------------------- // ZLogger — installed by `live` via `Runtime.removeDefaultLoggers ++ ZIO.withLoggerScoped`. Companion-visible // (`private[ProgressDisplay]`) so the layer's `withLoggerScoped` call can reach it without exposing the logger @@ -163,24 +231,101 @@ final class ProgressDisplay private[utils] ( } // --------------------------------------------------------------------------- - // Internal rendering — single-line, \r-based (no cursor-up needed). The `enabled` check lives in these - // helpers so callers don't repeat it; both also short-circuit on empty bar lists. Always called from inside - // a `lock.synchronized` block so the read of `bars` is consistent. + // Internal rendering — multi-line: each bar draws on its own line, and a redraw moves the cursor up over the block + // (CSI n A) then clears to end of screen (CSI J) before repainting. The `enabled` check lives in these helpers so + // callers don't repeat it. Always called from inside a `lock.synchronized` block so the reads of `bars` and + // `lastDrawnLines` stay consistent. ANSI cursor control was already assumed (the old \r-overwrite used CSI K), so + // this needs no capability beyond what a disabled / non-TTY display already gated out. `Esc` is built via 27.toChar + // (equivalent to a unicode-escape ESC, as the colour literals in the companion use) — both avoid a raw control byte; + // 27.toChar just keeps the string interpolations below free of escape noise. // --------------------------------------------------------------------------- - private def clearLineSync(): Unit = - if (enabled && bars.exists(_.lineCount > 0)) out.print("\r\u001b[K") + private val Esc: String = 27.toChar.toString + + // The visible bars as physical lines: each bar's output split on any embedded newlines, trimmed, and truncated to the + // detected terminal width so a too-wide bar can't wrap onto a second row and desync the cursor-up count. + private def barLinesSync(): List[String] = + bars.filter(_.lineCount > 0).flatMap(_.lastOutput.split("\n", -1)).map(l => truncateToWidth(l.trim)) + + // Move the cursor to column 0 of the FIRST bar line — up `lastDrawnLines - 1` rows — when a block is currently drawn. + private def moveToBlockTopSync(): Unit = + if (lastDrawnLines > 0) { + out.print("\r") + if (lastDrawnLines > 1) { out.print(s"$Esc[${lastDrawnLines - 1}A") } + } + // Erase the currently-drawn bar block, leaving the cursor at the top of where it was (so a log line prints there and + // the bars redraw below it). Resets the line count. + private def clearBlockSync(): Unit = + if (enabled && lastDrawnLines > 0) { + moveToBlockTopSync() + out.print(s"$Esc[J") // clear from cursor to end of screen + lastDrawnLines = 0 + } + + // Draw every visible bar one per line and record how many physical lines that took. Each line is `\r`-anchored and + // cleared to end of line (CSI K); lines are separated by `\n`. Assumes the block region is already clear (callers + // pair this with `clearBlockSync`), so a shrinking block leaves no orphaned rows. private def drawAllSync(): Unit = if (enabled) { - val parts = bars.filter(_.lineCount > 0).map(_.lastOutput.trim) - if (parts.nonEmpty) out.print("\r" + parts.mkString(" ") + "\u001b[K") + val lines = barLinesSync() + if (lines.nonEmpty) { + out.print(lines.mkString("\r", s"$Esc[K\n\r", s"$Esc[K")) + lastDrawnLines = lines.size + } } + + // Best-effort terminal width, resolved once at construction (the window is stable enough over a short interactive run). + // Prefer an exported COLUMNS, else ask the controlling terminal directly — bash does NOT export COLUMNS to a child + // JVM, so it's usually absent and a fixed 80 would mis-truncate on a narrower window — else fall back to 80. Computed + // eagerly (not lazily on first draw) so the `sttyCols` sub-process spawn runs here, off the render lock, rather than + // inside `lock.synchronized` on the first bar. Only an `enabled` display probes; the server / non-TTY path stays at 80. + private val terminalWidth: Int = + if (enabled) { Option(System.getenv("COLUMNS")).flatMap(_.toIntOption).filter(_ > 0).orElse(sttyCols).getOrElse(80) } + else { 80 } + + // The controlling terminal's column count via `stty size` (prints " "), reading /dev/tty so it reflects + // the real window even when this process's stdin/stdout are redirected. Any failure (no tty, stty absent) yields None. + private def sttyCols: Option[Int] = + try { + val proc = new ProcessBuilder("sh", "-c", "stty size < /dev/tty 2>/dev/null").start() + val output = new String(proc.getInputStream.readAllBytes(), StandardCharsets.UTF_8).trim + proc.waitFor() + output.split("\\s+").lift(1).flatMap(_.toIntOption).filter(_ > 0) + } catch { case _: Throwable => None } + + // Truncate a bar line to the detected width so it can't wrap onto a second physical row and desync the cursor-up count. + private def truncateToWidth(line: String): String = + if (line.length <= terminalWidth) { line } else { line.take(terminalWidth - 1) } // leave the last column, avoid wrap } object ProgressDisplay { - private case class BarState(id: Int, lineCount: Int, lastOutput: String) + /** A per-job (or the shared global) bar channel: the current set of that scope's bars keyed by bar id, published as a + * latest-wins snapshot. `GET /api/jobs/{id}/progress` streams `.changes`; removal is a key drop. + */ + type BarChannel = SubscriptionRef[Map[Int, BarSnapshot]] + + /** Per-fiber active bar channel — the job-scoped destination for bars created on this fiber (and, via FiberRef + * inheritance, its forked children). [[ccas.server.jobs.JobRunner]] sets it per job with `currentChannel.locally`, + * exactly as it does [[JobLogSink.currentSink]] for log lines, so a job's app bars land in that job's channel. + * `None` (the default) routes a bar to the display's shared global channel instead. + */ + val currentChannel: FiberRef[Option[BarChannel]] = + Unsafe.unsafe(implicit u => FiberRef.unsafe.make[Option[BarChannel]](None)) + + // `lastOutput` is the rendered line for the local terminal draw; `current`/`total`/`text` are the raw fields mirrored + // into `target` (this bar's channel) as an unrendered `BarSnapshot`. `target` is captured at `addBar` and fixed for + // the bar's life. `lineCount > 0` marks a bar that has actually been printed (drawn + snapshot-eligible). + private case class BarState( + id: Int, + lineCount: Int, + lastOutput: String, + current: Int, + total: Int, + text: String, + target: Option[BarChannel] + ) /** Synchronous factory — no IO, just allocates the lock + state. Use `live` instead in production code; `make` is * intended for tests that need a noop / quiet display without spinning up the layer machinery. @@ -189,13 +334,19 @@ object ProgressDisplay { * `true` for stdout rendering. `false` suppresses every stdout side-effect — bar lifecycle (`addBar`/`removeBar`) * still tracks state correctly so that `addBarScoped` finalisers behave the same in both modes. */ - def make(enabled: Boolean): ProgressDisplay = makeWith(enabled, System.out, System.err) + def make(enabled: Boolean): ProgressDisplay = makeWith(enabled, System.out, System.err, None) /** Test-only factory that injects the bar-redraw (`out`) and defect (`err`) streams so suites can capture output - * without mutating process-global `System.out` / `System.err`. Production code uses [[make]] / [[live]]. + * without mutating process-global `System.out` / `System.err`, and the optional global bar channel. Production code + * uses [[make]] / [[live]]. */ - private[utils] def makeWith(enabled: Boolean, out: PrintStream, err: PrintStream): ProgressDisplay = - new ProgressDisplay(enabled, out, err) + private[ccas] def makeWith( + enabled: Boolean, + out: PrintStream, + err: PrintStream, + globalChannel: Option[BarChannel] + ): ProgressDisplay = + new ProgressDisplay(enabled, out, err, globalChannel) /** Provides a `ProgressDisplay` service AND replaces ZIO's default console logger with the progress-display * `ZLogger` for the layer's scope. ZIO's default console logger is therefore inactive while this layer is alive — @@ -218,8 +369,9 @@ object ProgressDisplay { private[utils] def liveWith(showProgress: Boolean, out: PrintStream, err: PrintStream): URLayer[Scope, ProgressDisplay] = Runtime.removeDefaultLoggers ++ ZLayer.scoped { for { + channel <- SubscriptionRef.make(Map.empty[Int, BarSnapshot]) d <- ZIO.acquireRelease( - ZIO.succeed(makeWith(showProgress, out, err)) + ZIO.succeed(makeWith(showProgress, out, err, Some(channel))) )(d => ZIO.succeed(d.finishAllSync())) _ <- ZIO.withLoggerScoped(d.asZLogger) } yield d diff --git a/src/main/scala/ccas/utils/ProgressSnapshot.scala b/src/main/scala/ccas/utils/ProgressSnapshot.scala new file mode 100644 index 0000000..5408435 --- /dev/null +++ b/src/main/scala/ccas/utils/ProgressSnapshot.scala @@ -0,0 +1,28 @@ +package ccas.utils + +import zio.json.{DeriveJsonDecoder, DeriveJsonEncoder, JsonDecoder, JsonEncoder} + +/** One progress bar's raw, unrendered state — the wire shape streamed by `GET /api/jobs/{id}/progress`. + * + * Deliberately semantic (`current` / `total` / `text`), NOT a pre-rendered `████ 57%` string: the server can't render + * a width-correct bar because it doesn't know the following terminal's column count, so the block-bar + percentage are + * computed client-side ([[ProgressBar.render]]). `id` is the display-assigned bar id, stable for a bar's lifetime, so a + * client reconciles successive snapshots (update existing, add new, drop absent) without positional guessing. + */ +final case class BarSnapshot(id: Int, current: Int, total: Int, text: String) + +object BarSnapshot { + given JsonEncoder[BarSnapshot] = DeriveJsonEncoder.gen[BarSnapshot] + given JsonDecoder[BarSnapshot] = DeriveJsonDecoder.gen[BarSnapshot] +} + +/** A full, latest-wins snapshot of every bar currently live for a job (its own app bars plus the shared client's API + * gauge, merged at the route). Removal is implicit: a bar absent from a newer frame is gone. Wrapped in an object + * rather than sent as a bare array so the frame can gain fields later without a breaking wire change. + */ +final case class ProgressSnapshot(bars: List[BarSnapshot]) + +object ProgressSnapshot { + given JsonEncoder[ProgressSnapshot] = DeriveJsonEncoder.gen[ProgressSnapshot] + given JsonDecoder[ProgressSnapshot] = DeriveJsonDecoder.gen[ProgressSnapshot] +} diff --git a/src/test/scala/ccas/cli/TestCliParser.scala b/src/test/scala/ccas/cli/TestCliParser.scala index bfc567e..7712982 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))) + assertTrue(c.contains(CliCommand.Membership(DefaultServer, List("team-alpha", "team-beta"), false, None, false))) ) }, test("membership --all parses with no explicit clubs") { parsed("membership", "--all").map(c => - assertTrue(c.contains(CliCommand.Membership(DefaultServer, Nil, true, None))) + assertTrue(c.contains(CliCommand.Membership(DefaultServer, Nil, true, None, false))) ) }, test("use-club parses the slug (local, no server)") { @@ -84,11 +84,16 @@ 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)))) + parsed("membership").map(c => assertTrue(c.contains(CliCommand.Membership(DefaultServer, Nil, false, None, 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)))) + assertTrue(c.contains(CliCommand.Membership(DefaultServer, List("team-alpha"), false, Some(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))) ) }, test("recruit parses options, comma-separated source-clubs, and --club") { @@ -96,19 +101,19 @@ object TestCliParser extends ZIOSpecDefault { assertTrue( c.contains( CliCommand - .Recruit(DefaultServer, Some("team-alpha"), None, Some(5), false, List("x", "y"), None, Some(false), false, false, None) + .Recruit(DefaultServer, Some("team-alpha"), None, Some(5), false, List("x", "y"), None, Some(false), false, false, None, false) ) ) ) }, test("recruit with no --club parses to None") { parsed("recruit").map(c => - assertTrue(c.contains(CliCommand.Recruit(DefaultServer, None, None, None, false, Nil, None, None, false, false, None))) + assertTrue(c.contains(CliCommand.Recruit(DefaultServer, None, None, None, false, Nil, None, None, false, false, None, false))) ) }, test("recruit --stdout sets stdout") { parsed("recruit", "--stdout").map(c => - assertTrue(c.contains(CliCommand.Recruit(DefaultServer, None, None, None, false, Nil, None, None, true, false, None))) + assertTrue(c.contains(CliCommand.Recruit(DefaultServer, None, None, None, false, Nil, None, None, true, false, None, false))) ) }, test("recruit --report parses, with an optional run-id argument") { @@ -116,13 +121,13 @@ object TestCliParser extends ZIOSpecDefault { r <- parsed("recruit", "--report") n <- parsed("recruit", "--report", "42") } yield assertTrue( - r.contains(CliCommand.Recruit(DefaultServer, None, None, None, false, Nil, None, None, false, true, None)), - n.contains(CliCommand.Recruit(DefaultServer, None, None, None, false, Nil, None, None, false, true, Some(42))) + r.contains(CliCommand.Recruit(DefaultServer, None, None, None, false, Nil, None, None, false, true, None, false)), + n.contains(CliCommand.Recruit(DefaultServer, None, None, None, false, Nil, None, None, false, true, Some(42), false)) ) }, 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))) + assertTrue(c.contains(CliCommand.History(DefaultServer, List("team-alpha"), false, true, true, false, None, false))) ) }, test("blacklist add parses usernames and options") { diff --git a/src/test/scala/ccas/cli/TestClientProgressRenderer.scala b/src/test/scala/ccas/cli/TestClientProgressRenderer.scala new file mode 100644 index 0000000..14c2bcc --- /dev/null +++ b/src/test/scala/ccas/cli/TestClientProgressRenderer.scala @@ -0,0 +1,103 @@ +package ccas.cli + +import java.io.{ByteArrayOutputStream, PrintStream} + +import zio.{Ref, ZIO} +import zio.test.{assertTrue, Spec, ZIOSpecDefault} + +import ccas.utils.{BarSnapshot, ProgressDisplay, ProgressSnapshot} + +/** Unit tests for [[ClientProgressRenderer]] — the reconcile-and-draw of `/progress` snapshot frames onto a local + * display. Uses `ProgressDisplay.makeWith` with a capture buffer (no `System.out` swap), so bar output is observable + * without touching process-global streams. + */ +object TestClientProgressRenderer extends ZIOSpecDefault { + + private def withRenderer[A]( + use: (ClientProgressRenderer, Ref[Map[Int, ClientProgressRenderer.Tracked]]) => ZIO[Any, Throwable, A] + ): ZIO[Any, Throwable, (A, String)] = + ZIO.suspendSucceed { + val baos = new ByteArrayOutputStream + val ps = new PrintStream(baos, true, "UTF-8") + val display = ProgressDisplay.makeWith(enabled = true, ps, ps, None) + Ref.make(Map.empty[Int, ClientProgressRenderer.Tracked]).flatMap { ref => + use(new ClientProgressRenderer(display, ref), ref).map { a => + ps.flush() + (a, baos.toString("UTF-8")) + } + } + } + + private def frame(bars: BarSnapshot*): ProgressSnapshot = ProgressSnapshot(bars.toList) + + override def spec: Spec[Any, Throwable] = suite("TestClientProgressRenderer")( + test("renders a new bar, re-renders on update, and re-derives the block bar at the client from raw fields") { + withRenderer { (renderer, ref) => + for { + _ <- renderer.render(frame(BarSnapshot(0, 5, 10, "Working"))) + afterAdd <- ref.get + _ <- renderer.render(frame(BarSnapshot(0, 8, 10, "Working"))) + } yield afterAdd + }.map { case (afterAdd, out) => + assertTrue( + afterAdd.keySet == Set(0), // tracked under the server-assigned bar id + out.contains("Working"), + out.contains("50.0%"), // 5/10 rendered client-side + out.contains("80.0%"), // updated to 8/10 + out.contains("█"), // block bar drawn locally from raw current/total + out.contains("░") + ) + } + }, + test("a bar absent from the next frame is finished and forgotten (implicit removal)") { + withRenderer { (renderer, ref) => + for { + _ <- renderer.render(frame(BarSnapshot(0, 1, 10, "A"), BarSnapshot(1, 2, 10, "B"))) + afterTwo <- ref.get + _ <- renderer.render(frame(BarSnapshot(1, 3, 10, "B"))) // bar 0 dropped + afterRemove <- ref.get + } yield (afterTwo, afterRemove) + }.map { case ((afterTwo, afterRemove), _) => + assertTrue( + afterTwo.keySet == Set(0, 1), + afterRemove.keySet == Set(1) // only the surviving bar remains tracked + ) + } + }, + test("clear finishes every remaining bar and forgets them") { + withRenderer { (renderer, ref) => + for { + _ <- renderer.render(frame(BarSnapshot(0, 1, 10, "A"), BarSnapshot(1, 2, 10, "B"))) + _ <- renderer.clear + afterClear <- ref.get + } yield afterClear + }.map { case (afterClear, _) => + assertTrue(afterClear.isEmpty) + } + }, + test("logLine is interleaved on the display's stream") { + withRenderer { (renderer, _) => + renderer.render(frame(BarSnapshot(0, 1, 10, "Bar"))) *> renderer.logLine("a log line") + }.map { case (_, out) => + assertTrue(out.contains("a log line"), out.contains("Bar")) + } + }, + test("re-rendering an identical frame writes nothing further (unchanged bars are not redrawn)") { + ZIO.suspendSucceed { + val baos = new ByteArrayOutputStream + val ps = new PrintStream(baos, true, "UTF-8") + val display = ProgressDisplay.makeWith(enabled = true, ps, ps, None) + for { + ref <- Ref.make(Map.empty[Int, ClientProgressRenderer.Tracked]) + renderer = new ClientProgressRenderer(display, ref) + _ <- renderer.render(frame(BarSnapshot(0, 5, 10, "X"))) + _ <- ZIO.succeed(ps.flush()) + lenAfterFirst <- ZIO.succeed(baos.size()) + _ <- renderer.render(frame(BarSnapshot(0, 5, 10, "X"))) // byte-identical frame + _ <- ZIO.succeed(ps.flush()) + lenAfterSecond <- ZIO.succeed(baos.size()) + } yield assertTrue(lenAfterSecond == lenAfterFirst) + } + } + ) +} diff --git a/src/test/scala/ccas/cli/TestJobFollower.scala b/src/test/scala/ccas/cli/TestJobFollower.scala index f4a2b23..0877d25 100644 --- a/src/test/scala/ccas/cli/TestJobFollower.scala +++ b/src/test/scala/ccas/cli/TestJobFollower.scala @@ -18,9 +18,15 @@ object TestJobFollower extends ZIOSpecDefault { private case object Hang extends StreamBehaviour private final case class AlwaysFail(err: Throwable) extends StreamBehaviour private final case class Attempts(ref: Ref[List[(List[String], Boolean)]]) extends StreamBehaviour + // Stay open until `gate` completes (a progress-reconnect signal), then reach EOF — models a /logs stream that + // outlives the /progress reconnect so the test can observe it. + private final case class AwaitThenComplete(gate: Promise[Nothing, Unit]) extends StreamBehaviour - /** Stub that replays a fixed `JobStatusResponse` body for `getJson` and drives `streamLines` per [[StreamBehaviour]]. */ - private final class StubApi(script: Ref[List[String]], behaviour: StreamBehaviour) extends CcasApiClient { + /** Stub that replays a fixed `JobStatusResponse` body for `getJson`, drives `streamLines` per [[StreamBehaviour]], and + * runs `progress` (re-evaluated per call, so a Ref-backed effect can script a drop-then-reconnect) for `streamProgress`. + */ + private final class StubApi(script: Ref[List[String]], behaviour: StreamBehaviour, progress: Task[Unit]) + extends CcasApiClient { override def getJson[Resp: JsonDecoder](path: String): Task[Resp] = script.modify { case head :: tail => (head, tail) @@ -32,6 +38,7 @@ object TestJobFollower extends ZIOSpecDefault { case ReplayOnce(lines) => ZIO.foreachDiscard(lines)(onLine) case Hang => ZIO.never case AlwaysFail(err) => ZIO.fail(err) + case AwaitThenComplete(gate) => gate.await case Attempts(ref) => ref.modify { case head :: tail => (Some(head), tail) @@ -50,19 +57,27 @@ object TestJobFollower extends ZIOSpecDefault { ZIO.die(new UnsupportedOperationException("postEmpty")) override def postUnit[Req: JsonEncoder](path: String, body: Req): Task[Unit] = ZIO.unit override def delete(path: String): Task[Unit] = ZIO.unit + + // Rendering is exercised in TestClientProgressRenderer; here `progress` scripts the transport (default ZIO.unit = + // one clean pass; a Ref-backed effect scripts a drop-then-reconnect). onFrame is unused — no frames are emitted. + override def streamProgress(path: String)(onFrame: ccas.utils.ProgressSnapshot => UIO[Unit]): Task[Unit] = progress } private def statusJson(status: String, error: Option[String]): String = JobStatusResponse("job-1", "Membership", status, None, "2026-01-01T00:00:00Z", None, error, "Cli").toJson - // Fast reconnect tuning for tests: negligible backoff, tiny cap so the give-up path resolves quickly. + // Fast reconnect tuning for tests: negligible backoff, tiny cap so the give-up path resolves quickly. Bars off by + // default (plain log lines, asserted via TestConsole); `barsFollower` flips `showProgress` on for the bars-path test. private def follower(api: CcasApiClient, maxWait: Duration): JobFollower = - JobFollower(api, maxWait, reconnectBackoff = 1.milli, maxReconnects = 3) + JobFollower(api, maxWait, reconnectBackoff = 1.milli, maxReconnects = 3, showProgress = false) + + private def barsFollower(api: CcasApiClient, maxWait: Duration): JobFollower = + JobFollower(api, maxWait, reconnectBackoff = 1.milli, maxReconnects = 3, showProgress = true) private def followerWith(status: String, error: Option[String], logLines: List[String]): UIO[JobFollower] = Ref .make(List(statusJson(status, error))) - .map(ref => follower(new StubApi(ref, ReplayOnce(logLines)), 1.minute)) + .map(ref => follower(new StubApi(ref, ReplayOnce(logLines), ZIO.unit), 1.minute)) override def spec: Spec[Any, Throwable] = suite("TestJobFollower")( test("followJob streams the log lines and returns 0 when the job completes") { @@ -77,10 +92,44 @@ object TestJobFollower extends ZIOSpecDefault { 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 + // follow still resolves to 0 — i.e. the bars wiring doesn't break or hang the log follow. + for { + ref <- Ref.make(List(statusJson("Completed", None))) + code <- barsFollower(new StubApi(ref, ReplayOnce(List("log line")), ZIO.unit), 1.minute).followJob("job-1") + } yield assertTrue(code == 0) + }, + test("bars mode reconnects the /progress stream after a mid-stream drop") { + // The server reaps /progress at its read-idle timeout just like /logs (#161); the consumer must reconnect or the + // bars freeze for the rest of the job. /progress drops on its 1st subscribe, then on reconnect signals `reconnected` + // and ends; /logs stays open until that signal, so the reconnect is observed before the follow finishes. + for { + reconnected <- Promise.make[Nothing, Unit] + progressCall <- Ref.make(0) + progress = progressCall.updateAndGet(_ + 1).flatMap(n => + if (n == 1) { ZIO.fail(StreamDropped("read-idle reap")) } + else { reconnected.succeed(()).unit } + ) + ref <- Ref.make(List(statusJson("Completed", None))) + code <- barsFollower(new StubApi(ref, AwaitThenComplete(reconnected), progress), 1.minute).followJob("job-1") + calls <- progressCall.get + } yield assertTrue(code == 0, calls >= 2) + }, + test("bars mode times out cleanly and still reports 'still running'") { + // Bug-fix guard: on the timeout path the bars are cleared (onExit) BEFORE interpret prints the terminal status, so + // the follow resolves to 1 and the message is emitted rather than lost/hung. + for { + ref <- Ref.make(List(statusJson("Running", None))) + code <- barsFollower(new StubApi(ref, Hang, ZIO.unit), 30.millis).followJob("job-1") + err <- TestConsole.outputErr + } yield assertTrue(code == 1, err.exists(_.contains("still running"))) + }, test("followJob times out when the log stream never ends") { for { ref <- Ref.make(List(statusJson("Running", None))) - code <- follower(new StubApi(ref, Hang), 30.millis).followJob("job-1") + code <- follower(new StubApi(ref, Hang, ZIO.unit), 30.millis).followJob("job-1") } yield assertTrue(code == 1) }, test("followJob reconnects across a mid-stream drop and resumes without re-printing seen lines") { @@ -88,7 +137,7 @@ object TestJobFollower extends ZIOSpecDefault { // Attempt 1 shows a,b then drops; attempt 2 replays a,b (skipped as already-shown) and finishes with c,d. attempts <- Ref.make(List(List("a", "b") -> true, List("a", "b", "c", "d") -> false)) ref <- Ref.make(List(statusJson("Completed", None))) - code <- follower(new StubApi(ref, Attempts(attempts)), 1.minute).followJob("job-1") + code <- follower(new StubApi(ref, Attempts(attempts), ZIO.unit), 1.minute).followJob("job-1") out <- TestConsole.output err <- TestConsole.outputErr } yield assertTrue( @@ -104,7 +153,7 @@ object TestJobFollower extends ZIOSpecDefault { for { ref <- Ref.make(List(statusJson("Running", None))) // Always drops → never reaches EOF → give up after maxReconnects and surface the reattach hint. - code <- follower(new StubApi(ref, AlwaysFail(StreamDropped("idle timeout"))), 1.minute).followJob("job-1") + code <- follower(new StubApi(ref, AlwaysFail(StreamDropped("idle timeout")), ZIO.unit), 1.minute).followJob("job-1") err <- TestConsole.outputErr } yield assertTrue( code == 1, diff --git a/src/test/scala/ccas/server/jobs/TestJobRunner.scala b/src/test/scala/ccas/server/jobs/TestJobRunner.scala index 582d303..4ce9d61 100644 --- a/src/test/scala/ccas/server/jobs/TestJobRunner.scala +++ b/src/test/scala/ccas/server/jobs/TestJobRunner.scala @@ -36,7 +36,9 @@ object TestJobRunner extends ZIOSpecDefault { testLogStreamTailsLiveJob, testLogStreamCarriesPartialLineAcrossTicks, testLogStreamJoinsMultibyteCharAcrossTicks, - testLogStreamReassemblesCjkAndEmojiTornMidChar + testLogStreamReassemblesCjkAndEmojiTornMidChar, + testProgressStreamTailsLiveJobAndCloses, + testProgressStreamUnknownReturnsNone ).provideShared( FreshSchemaLayer("test_job_runner", onInit = ServerTables.ensureTables), TestChessComClientSupport.dummyLayer, @@ -204,6 +206,53 @@ object TestJobRunner extends ZIOSpecDefault { } yield assertTrue(result.isEmpty) } + // End-to-end for `progressStream`: submit → `currentChannel.locally` → a job bar publishes into the per-job channel → + // the stream emits its snapshot → the job completes → `interruptWhen(promise)` closes the stream. We wait for a bar + // frame *before* releasing the job (deterministic — otherwise the job could finish and drop the bar before the + // subscriber observes it), then assert the stream ended on completion. Global-gauge merge is None here (the suite's + // `ProgressDisplay.make` has no global channel), so this covers the `(None, Some(job))` path + the completion halt. + private def testProgressStreamTailsLiveJobAndCloses = + test("progressStream emits a running job's bar snapshots and closes when it finishes") { + for { + _ <- deleteAllJobRuns + runner <- ZIO.service[JobRunner] + gate <- Promise.make[Nothing, Unit] + id <- runner.submit( + JobKind.MatchRef, + None, + None, + RunTrigger.Cli, + _ => + ZIO.scoped { + for { + bar <- ProgressDisplay.progressBar + _ <- bar.print(1, 2, "working") + _ <- gate.await // stay Running until the subscriber has seen the bar frame + } yield () + } + ) + streamOpt <- runner.progressStream(id) + stream <- ZIO.fromOption(streamOpt).orElseFail(new Exception("expected a live progress stream")) + seen <- Promise.make[Nothing, Unit] + framesRef <- Ref.make(List.empty[String]) + collect <- stream + .runForeach(f => framesRef.update(f :: _) *> ZIO.whenDiscard(f.contains("working"))(seen.succeed(()).unit)) + .fork + _ <- seen.await.timeoutFail(new Exception("no bar frame observed"))(10.seconds) + _ <- gate.succeed(()) + _ <- awaitStatus(runner, id) + _ <- collect.join // returns only if the stream ended — proves close-on-completion + frames <- framesRef.get + } yield assertTrue(frames.exists(_.contains("\"text\":\"working\""))) + } + + private def testProgressStreamUnknownReturnsNone = test("progressStream returns None for unknown id") { + for { + runner <- ZIO.service[JobRunner] + result <- runner.progressStream(JobRunId.wrap("does-not-exist")) + } yield assertTrue(result.isEmpty) + } + private def testLogStreamTailsLiveJob = test("logStream tails a running job and closes when it finishes") { for { _ <- deleteAllJobRuns diff --git a/src/test/scala/ccas/server/routes/TestRoutes.scala b/src/test/scala/ccas/server/routes/TestRoutes.scala index 8823a73..a69f5fe 100644 --- a/src/test/scala/ccas/server/routes/TestRoutes.scala +++ b/src/test/scala/ccas/server/routes/TestRoutes.scala @@ -81,6 +81,10 @@ object TestRoutes extends ZIOSpecDefault { override def logStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = jobs.get.map(_.get(id).map(_ => ZStream.fromIterable(List("alpha", "beta")))) + // Canned single progress frame for any known job; None for unknown — pins the /progress route's 200/404 + framing. + override def progressStream(id: JobRunId): RIO[PostgresClient, Option[ZStream[Any, Throwable, String]]] = + jobs.get.map(_.get(id).map(_ => ZStream.fromIterable(List("""{"bars":[]}""")))) + def setNextAction(action: Action): UIO[Unit] = nextAction.set(action) def prePopulate(jobRun: JobRun): UIO[Unit] = jobs.update(_ + (jobRun.id -> jobRun)) @@ -150,6 +154,8 @@ object TestRoutes extends ZIOSpecDefault { testGetJobByIdReturns404, testGetJobLogsReturns200, testGetJobLogsReturns404, + testGetJobProgressReturns200, + testGetJobProgressReturns404, testStatsWithInvalidDateReturns400, testStatsWithPartialDatesReturns400, testRecruitmentInvitedAndFound, @@ -376,15 +382,15 @@ object TestRoutes extends ZIOSpecDefault { 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( - JobRunId.wrap("logs-id"), - JobKind.Membership, - None, - RunTrigger.Cli, - JobRunStatus.Completed, - None, - t0, - Some(t0), - None + id = JobRunId.wrap("logs-id"), + kind = JobKind.Membership, + clubId = None, + trigger = RunTrigger.Cli, + status = JobRunStatus.Completed, + params = None, + startedAt = t0, + completedAt = Some(t0), + error = None ) for { fake <- getFakeRunner @@ -408,6 +414,43 @@ object TestRoutes extends ZIOSpecDefault { ) } + private def testGetJobProgressReturns200 = + test("GET /api/jobs/:id/progress streams chunked NDJSON for an existing job") { + val t0 = LocalDateTime.of(2025, 6, 1, 0, 0).toInstant(ZoneOffset.UTC) + val job = JobRun( + id = JobRunId.wrap("progress-id"), + kind = JobKind.Membership, + clubId = None, + trigger = RunTrigger.Cli, + status = JobRunStatus.Completed, + params = None, + startedAt = t0, + completedAt = Some(t0), + error = None + ) + for { + fake <- getFakeRunner + _ <- fake.prePopulate(job) + response <- JobRoutes.routes.runZIO(jsonRequest(Method.GET, "/api/jobs/progress-id/progress")) + body <- response.body.asString + } yield assertTrue( + response.status == Status.Ok, + response.header(Header.ContentType).exists(_.mediaType == MediaType.text.`plain`), + body == "{\"bars\":[]}\n" + ) + } + + private def testGetJobProgressReturns404 = + test("GET /api/jobs/:id/progress returns 404 plain text for unknown job") { + for { + response <- JobRoutes.routes.runZIO(jsonRequest(Method.GET, "/api/jobs/nonexistent/progress")) + body <- response.body.asString + } yield assertTrue( + response.status == Status.NotFound, + body.contains("not found") + ) + } + private def testStatsWithInvalidDateReturns400 = test("POST /api/jobs/stats with invalid date returns 400") { for { _ <- ensureClubs diff --git a/src/test/scala/ccas/server/scheduler/TestJobScheduler.scala b/src/test/scala/ccas/server/scheduler/TestJobScheduler.scala index c992305..45a146e 100644 --- a/src/test/scala/ccas/server/scheduler/TestJobScheduler.scala +++ b/src/test/scala/ccas/server/scheduler/TestJobScheduler.scala @@ -66,6 +66,7 @@ object TestJobScheduler extends ZIOSpecDefault { override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none 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 } /** JobRunner stub that captures the raw `params` string of each submission. */ @@ -82,6 +83,7 @@ object TestJobScheduler extends ZIOSpecDefault { override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none 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 } /** JobRunner stub that tracks which clubIds are submitted. */ @@ -98,6 +100,7 @@ object TestJobScheduler extends ZIOSpecDefault { override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none 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 } private def testPollFiberStopsOnScopeClose = test("poll fiber stops when enclosing scope closes") { @@ -274,6 +277,7 @@ object TestJobScheduler extends ZIOSpecDefault { override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none 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 } scheduler = new JobScheduler.JobSchedulerLive(failingRunner, pgClient, pollInterval) @@ -313,6 +317,7 @@ object TestJobScheduler extends ZIOSpecDefault { override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none 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 } scheduler = new JobScheduler.JobSchedulerLive(runner, pgClient, pollInterval) @@ -360,6 +365,7 @@ object TestJobScheduler extends ZIOSpecDefault { override def status(id: JobRunId): RIO[PostgresClient, Option[JobRun]] = ZIO.none 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 } scheduler = new JobScheduler.JobSchedulerLive(conflictingRunner, pgClient, pollInterval) diff --git a/src/test/scala/ccas/utils/TestProgressBar.scala b/src/test/scala/ccas/utils/TestProgressBar.scala index 6e1678d..da2edf9 100644 --- a/src/test/scala/ccas/utils/TestProgressBar.scala +++ b/src/test/scala/ccas/utils/TestProgressBar.scala @@ -2,6 +2,7 @@ package ccas.utils import java.io.{ByteArrayOutputStream, PrintStream} +import zio.stream.SubscriptionRef import zio.{LogLevel, Promise, Ref, ZIO} import zio.test.{assertCompletes, assertTrue, Spec, ZIOSpecDefault} @@ -16,6 +17,8 @@ object TestProgressBar extends ZIOSpecDefault { testDisplayMultipleBars, testDisabledBarIsNoOp, testDisabledBarRemovedFromList, + testRenderPublishesRawSnapshotToChannel, + testMultiLineBarsUseCursorUp, testLogAboveBarsRoutesThroughDisplay, testZioLogInfoRoutesThroughLiveLogger, testCurrentLogLevelFiltersDebug, @@ -43,7 +46,7 @@ object TestProgressBar extends ZIOSpecDefault { ZIO.suspendSucceed { val baos = new ByteArrayOutputStream val ps = new PrintStream(baos, true, "UTF-8") - val display = ProgressDisplay.makeWith(enabled, ps, ps) + val display = ProgressDisplay.makeWith(enabled, ps, ps, None) val sink = new JobLogSink { override def writeConsoleSync(line: String): Unit = ps.println(line) } use(display, sink).map { a => ps.flush() @@ -195,6 +198,59 @@ object TestProgressBar extends ZIOSpecDefault { }.map(_._1) } + /** A bar created under an active [[ProgressDisplay.currentChannel]] (as `JobRunner` sets per job) mirrors its raw, + * unrendered `current` / `total` / `text` into that channel on each `print` — keyed by bar id — and drops the key on + * `finish`. This is the state `GET /api/jobs/{id}/progress` streams. Uses a disabled display: publishing is + * independent of terminal drawing, so no capture stream is needed. + */ + private def testRenderPublishesRawSnapshotToChannel = + test("bar print publishes a raw snapshot to the active channel; finish drops it") { + val display = ProgressDisplay.makeWith(enabled = false, System.out, System.err, None) + for { + channel <- SubscriptionRef.make(Map.empty[Int, BarSnapshot]) + afterPrint <- ProgressDisplay.currentChannel.locally(Some(channel)) { + ZIO.scoped { + for { + bar <- display.addBarScoped + _ <- bar.print(3, 12, " Working") + snap <- channel.get + } yield snap + } + } + afterFinish <- channel.get + } yield assertTrue( + afterPrint.size == 1, + afterPrint.values.head == BarSnapshot(afterPrint.keys.head, 3, 12, " Working"), + afterFinish.isEmpty + ) + } + + /** Two bars draw on their own lines (a `\n` between them), and re-rendering a two-line block moves the cursor up over + * it (`ESC[1A`) before repainting — the multi-line behaviour that replaced the old single `\r`-line. `ESC` is built + * via `27.toChar` here (no source escape), matching the display. + */ + private def testMultiLineBarsUseCursorUp = test("multiple bars draw on separate lines and redraw via cursor-up") { + withCapture(enabled = true) { (display, _) => + ZIO.scoped { + for { + b1 <- display.addBarScoped + b2 <- display.addBarScoped + _ <- b1.print(1, 10, "Alpha") + _ <- b2.print(2, 10, "Beta") // now a two-line block + _ <- b1.print(3, 10, "Alpha") // redraw => cursor up over the two lines + } yield () + } + }.map { case (_, out) => + val esc = 27.toChar.toString + assertTrue( + out.contains("Alpha"), + out.contains("Beta"), + out.contains("\n"), // bars occupy their own lines + out.contains(s"$esc[1A") // moved up one row to repaint the two-line block + ) + } + } + private def testLogAboveBarsRoutesThroughDisplay = test("logAboveBarsSync interleaves above the active bar") { withCapture(enabled = true) { (display, sink) => ZIO.scoped {