mi/werkator
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Step 22 C: the watcher multiplexes the registry

Browse Source 
start takes the served repositories; one poll cycle polls each in its own guard
and aggregates the reports into WatcherState, which gains a per-repository list.
With one repository the top-level fields read exactly as before; with several,
every message is prefixed by the repository name. Log lines carry the name too.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
mhoennig
2026-09-02 09:00:52 +02:00
co-authored by Claude Fable 5.1
parent b16c9c2e27
commit a5027acfaf
4 changed files with 154 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/kotlin/de/hoennig/werkator/server/ServerWatcherLifecycle.kt
+4 -4
View File
@@ -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
src/main/kotlin/de/hoennig/werkator/watcher/Watcher.kt
+85 -41
View File
@@ -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<RepoContext>) {
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<RepoContext>) {
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<String>.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
}
src/main/kotlin/de/hoennig/werkator/watcher/WatcherState.kt
+20 -4
View File
@@ -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, `<name>: <reason>` 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<String> = emptyList(),
/** The same per served repository, in registry order. */
val repositories: List<RepoWatcherState> = 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<String> = emptyList(),
)
src/test/kotlin/de/hoennig/werkator/watcher/WatcherTest.kt
+45 -2
View File
@@ -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<IllegalStateException> { harness.watcher.start(harness.repo) }
shouldThrow<IllegalStateException> { harness.watcher.start(listOf(harness.repo)) }
harness.watcher.stop()