diff --git a/src/main/kotlin/de/hoennig/werkator/server/ServerWatcherLifecycle.kt b/src/main/kotlin/de/hoennig/werkator/server/ServerWatcherLifecycle.kt index 3254c82..bb0bd10 100644 --- a/src/main/kotlin/de/hoennig/werkator/server/ServerWatcherLifecycle.kt +++ b/src/main/kotlin/de/hoennig/werkator/server/ServerWatcherLifecycle.kt @@ -1,6 +1,6 @@ package de.hoennig.werkator.server -import de.hoennig.werkator.repo.RepoContext +import de.hoennig.werkator.repo.RepoRegistry import de.hoennig.werkator.watcher.Watcher import jakarta.annotation.PreDestroy import org.springframework.boot.context.event.ApplicationReadyEvent @@ -9,7 +9,7 @@ import org.springframework.context.event.EventListener import org.springframework.stereotype.Component /** - * Starts the watcher poll loop over the served repository once the server context + * Starts the watcher poll loop over the served repositories once the server context * is ready and stops it on shutdown. Only in the `server` profile — CLI commands * and tests never start the loop (see [Watcher]). */ @@ -17,11 +17,11 @@ import org.springframework.stereotype.Component @Profile("server") class ServerWatcherLifecycle( private val watcher: Watcher, - private val repo: RepoContext, + private val registry: RepoRegistry, ) { @EventListener(ApplicationReadyEvent::class) fun onApplicationReady() { - watcher.start(repo) + watcher.start(registry.all()) } @PreDestroy diff --git a/src/main/kotlin/de/hoennig/werkator/watcher/Watcher.kt b/src/main/kotlin/de/hoennig/werkator/watcher/Watcher.kt index 78dab00..2f1fcd1 100644 --- a/src/main/kotlin/de/hoennig/werkator/watcher/Watcher.kt +++ b/src/main/kotlin/de/hoennig/werkator/watcher/Watcher.kt @@ -26,10 +26,12 @@ import java.util.concurrent.ScheduledExecutorService import java.util.concurrent.TimeUnit /** - * Replaces the legacy blocking main loop: a non-blocking fixed-delay poll cycle that - * fetches origin, enqueues due branches via the async [BuildExecutor], and prunes - * retention — it never waits for a build and never builds in the primary checkout - * (whose branch refs it does fast-forward, see [fastForwardLocalRefs]). + * Replaces the legacy blocking main loop: a non-blocking fixed-delay poll cycle that, + * for every served repository, fetches origin, enqueues due branches via the async + * [BuildExecutor], and prunes retention — it never waits for a build and never builds + * in the primary checkout (whose branch refs it does fast-forward, see + * [fastForwardLocalRefs]). One repository's failure never reaches another: each is + * polled in its own guard and reports on its own in [WatcherState.repositories]. * The loop only runs after an explicit [start] (server/watch mode, step 07); * nothing is scheduled during CLI commands or tests. */ @@ -78,24 +80,35 @@ class Watcher( } /** - * Runs the startup recovery and schedules the poll loop with the fixed delay - * `watcher.pollInterval`; the first poll runs immediately. + * Runs the startup recovery of every repository and schedules the poll loop with the + * fixed delay `watcher.pollInterval` — one loop, one delay: the instance's setting, + * which every repository's effective config carries; the first poll runs immediately. */ @Synchronized - fun start(repo: RepoContext) { + fun start(repos: List) { check(scheduler == null) { "watcher is already running" } - recoverOnStartup(repo) - val interval = DurationParser.parse(configLoader.load(repo.workingDir).watcher.pollInterval) + require(repos.isNotEmpty()) { "no repository to watch" } + repos.forEach { recoverSafely(it) } + val interval = DurationParser.parse(configLoader.load(repos.first().workingDir).watcher.pollInterval) scheduler = Executors .newSingleThreadScheduledExecutor { runnable -> Thread(runnable, "werkator-watcher").apply { isDaemon = true } }.also { - it.scheduleWithFixedDelay({ pollSafely(repo) }, 0, interval.toMillis(), TimeUnit.MILLISECONDS) + it.scheduleWithFixedDelay({ pollAll(repos) }, 0, interval.toMillis(), TimeUnit.MILLISECONDS) } state = state.copy(running = true) } + /** A repository whose recovery crashes is still polled; the others' recovery is never skipped. */ + private fun recoverSafely(repo: RepoContext) { + try { + recoverOnStartup(repo) + } catch (e: Exception) { + log.error("[{}] startup recovery failed", repo.name, e) + } + } + @Synchronized fun stop() { scheduler?.shutdownNow() @@ -144,31 +157,73 @@ class Watcher( } } + /** One poll cycle over a single repository; see [pollAll]. */ + fun poll(repo: RepoContext) = pollAll(listOf(repo)) + /** - * One poll cycle, never blocking on a build: fetch origin (on failure: log once per - * message, expose in [state], retry next cycle), enqueue due branches — changed local branches - * first, then recent new origin branches, then due auto-build slots — then - * fast-forward the local branch refs, and finally prune results, artifacts, and - * worktrees of branches gone from origin. + * One poll cycle over all served repositories, never blocking on a build. Each + * repository is polled in its own guard — a crash or an unreachable origin is that + * repository's report, and the next one is polled regardless — and the cycle's + * [state] aggregates the reports: the top-level fields read as before with one + * repository, and name the repository in front of every message with several. */ - fun poll(repo: RepoContext) { + fun pollAll(repos: List) { val startedAt = clock.instant() + val reports = repos.map { pollSafely(it, startedAt) } + val several = repos.size > 1 + + fun named( + report: RepoWatcherState, + message: String, + ): String = if (several) "${report.name}: $message" else message + state = + state.copy( + lastPollAt = startedAt, + lastFetchError = reports.mapNotNull { report -> report.lastFetchError?.let { named(report, it) } }.joinOrNull(), + lastPollError = reports.mapNotNull { report -> report.lastPollError?.let { named(report, it) } }.joinOrNull(), + queuedBranches = reports.flatMap { it.queuedBranches }, + repositories = reports, + ) + } + + private fun List.joinOrNull(): String? = takeIf { it.isNotEmpty() }?.joinToString("; ") + + private fun pollSafely( + repo: RepoContext, + startedAt: Instant, + ): RepoWatcherState = + try { + pollRepo(repo, startedAt) + } catch (e: Exception) { + log.error("[{}] poll cycle failed", repo.name, e) + RepoWatcherState(repo.name, lastPollAt = startedAt, lastPollError = e.message ?: e.javaClass.simpleName) + } + + /** + * One repository's poll: fetch origin (on failure: log once per message, report it, + * retry next cycle), enqueue due branches — changed local branches first, then recent + * new origin branches, then due auto-build slots — then fast-forward the local branch + * refs, and finally prune results, artifacts, and worktrees of branches gone from origin. + */ + private fun pollRepo( + repo: RepoContext, + startedAt: Instant, + ): RepoWatcherState { val workingDir = repo.workingDir val watch = watchOf(repo) try { gitService.fetchOrigin(workingDir) if (watch.loggedFetchError != null) { - log.info("fetching origin succeeded again") + log.info("[{}] fetching origin succeeded again", repo.name) watch.loggedFetchError = null } } catch (e: Exception) { val failure = e.message ?: e.javaClass.simpleName if (watch.loggedFetchError != failure) { - log.warn("fetching origin failed; retrying every cycle until it succeeds: {}", failure) + log.warn("[{}] fetching origin failed; retrying every cycle until it succeeds: {}", repo.name, failure) watch.loggedFetchError = failure } - state = state.copy(lastPollAt = startedAt, lastFetchError = failure) - return + return RepoWatcherState(repo.name, lastPollAt = startedAt, lastFetchError = failure) } val config = configLoader.load(workingDir) val originBranches = gitService.originBranches(workingDir) @@ -177,26 +232,15 @@ class Watcher( fastForwardLocalRefs(workingDir) } prune(repo, config, originBranches) - state = - state.copy( - lastPollAt = startedAt, - lastFetchError = null, - lastPollError = null, - queuedBranches = - repo.results - .latestPerName() - .filter { it.status == BuildStatus.PENDING || it.status == BuildStatus.RUNNING } - .map { it.name }, - ) - } - - private fun pollSafely(repo: RepoContext) { - try { - poll(repo) - } catch (e: Exception) { - log.error("poll cycle failed", e) - state = state.copy(lastPollAt = clock.instant(), lastPollError = e.message ?: e.javaClass.simpleName) - } + return RepoWatcherState( + repo.name, + lastPollAt = startedAt, + queuedBranches = + repo.results + .latestPerName() + .filter { it.status == BuildStatus.PENDING || it.status == BuildStatus.RUNNING } + .map { it.name }, + ) } /** @@ -343,7 +387,7 @@ class Watcher( log.info("not enqueueing branch {}: no pull request has head commit {}", branch, commit) return false } - log.info("enqueueing build {} of branch {} at commit {}", build, branch, commit) + log.info("[{}] enqueueing build {} of branch {} at commit {}", repo.name, build, branch, commit) buildExecutor.startBuild(repo, branch, commit, build) return true } diff --git a/src/main/kotlin/de/hoennig/werkator/watcher/WatcherState.kt b/src/main/kotlin/de/hoennig/werkator/watcher/WatcherState.kt index 62bd7ea..d733b94 100644 --- a/src/main/kotlin/de/hoennig/werkator/watcher/WatcherState.kt +++ b/src/main/kotlin/de/hoennig/werkator/watcher/WatcherState.kt @@ -2,16 +2,32 @@ package de.hoennig.werkator.watcher import java.time.Instant -/** Observable watcher health for status endpoints (step 07) and the UI (step 08). */ +/** + * Observable watcher health for status endpoints (step 07) and the UI (step 08). + * The top-level fields describe the whole poll cycle — with one repository they are + * that repository's, with several they aggregate [repositories], where each served + * repository reports on its own (ADR 0009). + */ data class WatcherState( /** Whether the poll loop is scheduled. */ val running: Boolean = false, /** When the last poll cycle started, successful or not. */ val lastPollAt: Instant? = null, - /** Why the last `fetchOrigin` failed; null after a successful fetch. */ + /** Why the last `fetchOrigin` failed; null after a successful fetch. With several repositories, `: ` per failure. */ + val lastFetchError: String? = null, + /** Why the last poll cycle crashed after a successful fetch; null after a clean cycle. Named per repository like [lastFetchError]. */ + val lastPollError: String? = null, + /** Branches whose latest build was PENDING or RUNNING at the end of the last poll, over all repositories. */ + val queuedBranches: List = emptyList(), + /** The same per served repository, in registry order. */ + val repositories: List = emptyList(), +) + +/** One repository's part of the last poll cycle. */ +data class RepoWatcherState( + val name: String, + val lastPollAt: Instant? = null, val lastFetchError: String? = null, - /** Why the last poll cycle crashed after a successful fetch; null after a clean cycle. */ val lastPollError: String? = null, - /** Branches whose latest build was PENDING or RUNNING at the end of the last poll. */ val queuedBranches: List = emptyList(), ) diff --git a/src/test/kotlin/de/hoennig/werkator/watcher/WatcherTest.kt b/src/test/kotlin/de/hoennig/werkator/watcher/WatcherTest.kt index d99fdf5..be58238 100644 --- a/src/test/kotlin/de/hoennig/werkator/watcher/WatcherTest.kt +++ b/src/test/kotlin/de/hoennig/werkator/watcher/WatcherTest.kt @@ -767,19 +767,62 @@ class WatcherTest : FunSpec() { Files.exists(busyWorktree).shouldBeTrue() } + test("one repository's unreachable origin neither stops nor silences the other") { + val harness = Harness() + val otherDir = Files.createTempDirectory("werkator-watcher-other") + val other = + RepoContext( + "other", + otherDir, + FileBuildResultRepository(otherDir.resolve(".git/werkator/build-results.json")), + harness.artifactStore, + ) + every { harness.gitService.fetchOrigin(harness.workingDir) } throws RuntimeException("origin unreachable") + every { harness.gitService.originBranches(otherDir) } returns listOf("main") + every { harness.gitService.originBranchHeads(otherDir) } returns mapOf("main" to "commit-other") + every { harness.gitService.localBranches(otherDir) } returns listOf("main") + every { harness.gitService.hasNewCommits("main", otherDir) } returns true + every { harness.gitService.originHeadCommit("main", otherDir) } returns "commit-other" + + harness.watcher.pollAll(listOf(harness.repo, other)) + + verify { harness.buildExecutor.startBuild(other, "main", "commit-other", BuildDefinition.DEFAULT) } + verify(exactly = 0) { harness.buildExecutor.startBuild(harness.repo, any(), any(), any()) } + val state = harness.watcher.state() + state.lastFetchError shouldBe "test: origin unreachable" + state.lastPollError shouldBe null + state.repositories.map { it.name } shouldBe listOf("test", "other") + state.repositories[0].lastFetchError shouldBe "origin unreachable" + state.repositories[1].lastFetchError shouldBe null + } + + test("a repository whose poll crashes reports it by name and the cycle goes on") { + val harness = Harness() + val otherDir = Files.createTempDirectory("werkator-watcher-other") + val other = RepoContext("other", otherDir, harness.repository, harness.artifactStore) + every { harness.gitService.originBranches(otherDir) } throws IllegalStateException("corrupt refs") + + harness.watcher.pollAll(listOf(harness.repo, other)) + + val state = harness.watcher.state() + state.lastPollError shouldBe "other: corrupt refs" + state.repositories[1].lastPollError shouldBe "corrupt refs" + state.repositories[0].lastPollError shouldBe null + } + test("start runs recovery plus an immediate first poll; stop halts the loop") { val harness = Harness() val fetches = CountDownLatch(2) every { harness.gitService.fetchOrigin(any()) } answers { fetches.countDown() } - harness.watcher.start(harness.repo) + harness.watcher.start(listOf(harness.repo)) fetches.await(5, TimeUnit.SECONDS).shouldBeTrue() harness.watcher .state() .running .shouldBeTrue() - shouldThrow { harness.watcher.start(harness.repo) } + shouldThrow { harness.watcher.start(listOf(harness.repo)) } harness.watcher.stop()