✨ (ci): Reclaim zombie tasks when a runner vanishes
Changes
9 files changed, +262 -24
MODIFY
docs/admins/ci-runners.md
+8 -1
@@ -97,11 +97,18 @@
97
97
The `ci_runner*` tables hold no repository data (losing them only forces re-registration); the
98
98
`action_*` tables hold run history and logs, tied to their repository by cascade.
99
99
100
+## Task timeout and reclaim
101
+
102
+A task a runner claims must finish within `GITSHARK_CI_TASK_TIMEOUT` (default `1h`). A background sweep
103
+running every `GITSHARK_CI_ZOMBIE_RECLAIM_INTERVAL` (default `1m`) fails any task still running past its
104
+deadline — the assumption being the runner crashed or lost connectivity — and marks that runner
105
+`OFFLINE`. Raise the timeout if you legitimately run long jobs.
106
+
100
107
## Troubleshooting
101
108
102
109
| Symptom | Likely cause |
103
110
|---|---|
104
111
| `register` fails with `401`/`unauthenticated` | Registration token wrong, or deleted in the admin UI. Generate a fresh one. |
105
-| Runner registers but never runs anything | Expected in phase 1 — job execution is not implemented yet. |
112
+| Jobs are picked up but always end in failure after the timeout | The runner cannot report back (UpdateTask/UpdateLog blocked by the proxy), so the task is reclaimed as a zombie; forward the `x-runner-*` headers and allow POSTs to `/api/actions`. |
106
113
| `Declare` returns `401` after a working `Register` | Proxy is stripping `x-runner-uuid` / `x-runner-token`; forward them. |
107
114
| Runner cannot reach the instance | `--instance` must be the public origin; the runner appends `/api/actions` itself. |
MODIFY
docs/admins/getting-started.md
+2 -0
@@ -368,6 +368,8 @@
368
368
| `GITSHARK_FEDERATION_USER_RESYNC_INTERVAL` | — | `5m` | Re-scan followed users for new public repos |
369
369
| `GITSHARK_FEDERATION_DEV_ALLOW_INSECURE` | — | `false` | Dev only: allow http/loopback peers |
370
370
| `GITSHARK_ADMIN_HANDLES` | — | — | Comma-separated handles allowed into `/admin/*` (CI runner management); empty means no admins (see [CI runners](ci-runners.md)) |
371
+| `GITSHARK_CI_TASK_TIMEOUT` | — | `1h` | How long a claimed CI task may run before it is reclaimed as a zombie (see [CI runners](ci-runners.md)) |
372
+| `GITSHARK_CI_ZOMBIE_RECLAIM_INTERVAL` | — | `1m` | How often the sweep that fails timed-out CI tasks runs |
371
373
| `GITSHARK_GITEA_API_VERSION` | — | `1.13.0` | Version string reported by `GET /api/v1/version`. The `/api/v1` surface is Gitea-compatible; Gitea clients (Renovate, `tea`) gate features on this. Kept below `1.14.0` so they only call implemented endpoints — raise it as reviewer/label/status support lands |
372
374
373
375
### Optional: push mirrors
MODIFY
docs/maintainers/ci-runners.md
+7 -1
@@ -14,6 +14,7 @@
14
14
| Registration/presence | `ci/RunnerRegistrationService.java` | Token issue, runner register/declare/authenticate, list/delete. |
15
15
| Task dispatch | `ci/TaskDispatchService.java` | FetchTask: authenticate, claim the oldest PENDING task, flip task+run to RUNNING, set the runner ACTIVE and the task deadline — atomically. |
16
16
| Task progress | `ci/TaskProgressService.java` | UpdateTask (result → task status + run roll-up, runner back to IDLE) and UpdateLog (resume-safe log-row append, `ack_index`). |
17
+| Zombie reclaim | `ci/ZombieReclaimService.java` | Scheduled sweep failing RUNNING tasks past their deadline (vanished runner) and rolling up their runs. |
17
18
| Entities | `model/CiRunner.java`, `model/CiRunnerRegistrationToken.java` | Runner state (migration `V19`). |
18
19
| Run entities | `model/ActionRun.java`, `model/ActionTask.java`, `model/ActionLog.java` | Run/job/log-row persistence (migrations `V23`, `V24`). `ActionTask.seq` (`bigserial`) is the surrogate int64 `Task.id` for the wire. |
19
20
| Workflow ingest | `ci/WorkflowIngestService.java`, `ci/WorkflowRunFactory.java` | Post-receive hook: parse `.forgejo`/`.gitea` workflows at the pushed head, evaluate `on: push`, persist a run + its tasks. Rows are created but not yet dispatched. |
@@ -78,6 +79,12 @@
78
79
`FetchTaskTest` (claim oldest pending over the wire, empty queue, bad credentials, and two runners
79
80
racing one task → claimed at most once), `TaskProgressTest` (UpdateTask success rolls up task+run
80
81
and frees the runner, UpdateLog append + dedup/resume, cross-runner and bad-credential rejection).
82
+- **Zombie reclaim (`ZombieReclaimService`):** a scheduled sweep
83
+ (`gitshark.ci.zombie-reclaim-interval`, default 1m) fails any RUNNING task whose
84
+ `action_task.deadline` has passed — the runner is presumed gone — rolls its run up, and flags the
85
+ runner OFFLINE. The deadline is set at claim time from `gitshark.ci.task-timeout` (default 1h). A
86
+ late update from a runner cannot resurrect an already-terminal task. `ZombieReclaimTest` covers both
87
+ the reclaim (overdue → FAILURE, in-deadline left RUNNING) and the anti-resurrection guard.
81
88
82
89
## What still needs to be implemented
83
90
@@ -89,7 +96,6 @@
89
96
- **Trigger refinement:** only bare `on: push` is honored; branch/tag/path filters and other events
90
97
(tag push, `pull_request`) are not evaluated.
91
98
- **Run UI:** per-repository run list + run detail with live per-step status and logs.
92
-- **Task state machine:** timeout / zombie handling when a runner vanishes mid-task.
93
99
- **Real-runner integration test:** protocol round-trip against an actual `forgejo-runner` container
94
100
(the current endpoint test uses a hand-built protobuf client, not the binary).
95
101
- **Later phases:** secrets/variables delivery, label-based matching, concurrency/cancellation,
MODIFY
src/main/java/de/workaround/ci/TaskDispatchService.java
+24 -21
@@ -6,6 +6,8 @@
6
6
import java.util.Optional;
7
7
import java.util.UUID;
8
8
9
+import org.eclipse.microprofile.config.inject.ConfigProperty;
10
+
9
11
import de.workaround.model.ActionRun;
10
12
import de.workaround.model.ActionTask;
11
13
import de.workaround.model.CiRunner;
@@ -16,7 +18,7 @@
16
18
17
19
/**
18
20
* Hands PENDING tasks to runners over FetchTask (issue #2, phase 1). Claiming a task flips it (and its
19
- * run) to RUNNING, records the claiming runner and a {@link #TASK_TIMEOUT}-based {@link
21
+ * run) to RUNNING, records the claiming runner and a {@link #taskTimeout}-based {@link
20
22
* ActionTask#deadline} for later zombie reclaim, and marks the runner ACTIVE — all in one transaction
21
23
* so a task is never handed to two runners: the candidate row is selected {@code FOR UPDATE SKIP
22
24
* LOCKED}, so concurrent fetchers pick distinct rows (or none) rather than racing on the same one.
@@ -27,8 +29,9 @@
27
29
@ApplicationScoped
28
30
public class TaskDispatchService
29
31
{
30
- /** How long a claimed task may run before it is eligible for zombie reclaim (reclaim itself is a later slice). */
31
- static final Duration TASK_TIMEOUT = Duration.ofHours(1);
32
+ /** How long a claimed task may run before {@link ZombieReclaimService} may reclaim it. */
33
+ @ConfigProperty(name = "gitshark.ci.task-timeout", defaultValue = "1h")
34
+ Duration taskTimeout;
32
35
33
36
@Inject
34
37
RunnerRegistrationService runnerService;
@@ -63,6 +66,24 @@
63
66
return new Fetched(next, tasks.maxSeq());
64
67
}
65
68
69
+ private void claim(ActionTask task, CiRunner runner)
70
+ {
71
+ Instant now = Instant.now();
72
+ task.runner = runner;
73
+ task.status = ActionRun.Status.RUNNING;
74
+ task.startedAt = now;
75
+ task.deadline = now.plus(taskTimeout);
76
+
77
+ runner.status = CiRunner.Status.ACTIVE;
78
+
79
+ ActionRun run = task.run;
80
+ if (run.status == ActionRun.Status.PENDING)
81
+ {
82
+ run.status = ActionRun.Status.RUNNING;
83
+ run.startedAt = now;
84
+ }
85
+ }
86
+
66
87
/**
67
88
* Lock the oldest PENDING task's id with {@code FOR UPDATE SKIP LOCKED} so a concurrent fetcher in
68
89
* another transaction cannot claim the same row. Selecting only the id keeps the {@code FOR UPDATE}
@@ -77,22 +98,4 @@
77
98
return ids.isEmpty() ? Optional.empty() : Optional.of(ids.get(0));
78
99
}
79
100
80
- private static void claim(ActionTask task, CiRunner runner)
81
- {
82
- Instant now = Instant.now();
83
- task.runner = runner;
84
- task.status = ActionRun.Status.RUNNING;
85
- task.startedAt = now;
86
- task.deadline = now.plus(TASK_TIMEOUT);
87
-
88
- runner.status = CiRunner.Status.ACTIVE;
89
-
90
- ActionRun run = task.run;
91
- if (run.status == ActionRun.Status.PENDING)
92
- {
93
- run.status = ActionRun.Status.RUNNING;
94
- run.startedAt = now;
95
- }
96
- }
97
-
98
101
}
MODIFY
src/main/java/de/workaround/ci/TaskProgressService.java
+6 -1
@@ -45,6 +45,11 @@
45
45
CiRunner runner = authenticate(uuid, token);
46
46
ActionTask task = ownedTask(taskSeq, runner);
47
47
48
+ if (task.status.isTerminal())
49
+ {
50
+ // Already settled (e.g. reclaimed as a zombie); a late runner update must not resurrect it.
51
+ return task;
52
+ }
48
53
ActionRun.Status status = map(result);
49
54
task.status = status;
50
55
if (status.isTerminal())
@@ -111,7 +116,7 @@
111
116
}
112
117
113
118
/** A single-job run mirrors its task; a multi-job run is RUNNING until all tasks finish, then the worst outcome. */
114
- private void rollUpRun(ActionRun run)
119
+ public void rollUpRun(ActionRun run)
115
120
{
116
121
List<ActionTask> all = tasks.findByRun(run);
117
122
boolean anyRunning = all.stream().anyMatch(t -> !t.status.isTerminal());
ADD
src/main/java/de/workaround/ci/ZombieReclaimService.java
+61 -0
@@ -0,0 +1,61 @@
1
+package de.workaround.ci;
2
+
3
+import java.time.Instant;
4
+import java.util.List;
5
+
6
+import org.jboss.logging.Logger;
7
+
8
+import de.workaround.model.ActionRun;
9
+import de.workaround.model.ActionTask;
10
+import de.workaround.model.CiRunner;
11
+import io.quarkus.scheduler.Scheduled;
12
+import jakarta.enterprise.context.ApplicationScoped;
13
+import jakarta.inject.Inject;
14
+import jakarta.transaction.Transactional;
15
+
16
+/**
17
+ * Fails tasks whose runner vanished mid-run (issue #2, phase 1). A claimed task carries a {@link
18
+ * ActionTask#deadline}; once it passes with the task still RUNNING, the runner is presumed gone, so
19
+ * the task is marked FAILURE, its run rolled up, and the runner flagged OFFLINE. Runs on a schedule
20
+ * ({@code gitshark.ci.zombie-reclaim-interval}); {@link #reclaim(Instant)} is the testable core.
21
+ */
22
+@ApplicationScoped
23
+public class ZombieReclaimService
24
+{
25
+ private static final Logger LOG = Logger.getLogger(ZombieReclaimService.class);
26
+
27
+ @Inject
28
+ ActionTask.Repo tasks;
29
+
30
+ @Inject
31
+ TaskProgressService progress;
32
+
33
+ @Scheduled(every = "{gitshark.ci.zombie-reclaim-interval}", concurrentExecution = Scheduled.ConcurrentExecution.SKIP)
34
+ void sweep()
35
+ {
36
+ int reclaimed = reclaim(Instant.now());
37
+ if (reclaimed > 0)
38
+ {
39
+ LOG.infof("Reclaimed %d zombie CI task(s) past their deadline", reclaimed);
40
+ }
41
+ }
42
+
43
+ /** Fail every RUNNING task whose deadline is before {@code now}; returns how many were reclaimed. */
44
+ @Transactional
45
+ public int reclaim(Instant now)
46
+ {
47
+ List<ActionTask> overdue = tasks.listRunningPastDeadline(now);
48
+ for (ActionTask task : overdue)
49
+ {
50
+ task.status = ActionRun.Status.FAILURE;
51
+ task.finishedAt = now;
52
+ if (task.runner != null)
53
+ {
54
+ task.runner.status = CiRunner.Status.OFFLINE;
55
+ }
56
+ progress.rollUpRun(task.run);
57
+ }
58
+ return overdue.size();
59
+ }
60
+
61
+}
MODIFY
src/main/java/de/workaround/model/ActionTask.java
+3 -0
@@ -81,6 +81,9 @@
81
81
@HQL("select t from ActionTask t where t.status = PENDING order by t.createdAt asc")
82
82
List<ActionTask> listPending();
83
83
84
+ @HQL("select t from ActionTask t where t.status = RUNNING and t.deadline is not null and t.deadline < :now")
85
+ List<ActionTask> listRunningPastDeadline(Instant now);
86
+
84
87
default Optional<ActionTask> findOldestPending()
85
88
{
86
89
return listPending().stream().findFirst();
MODIFY
src/main/resources/application.properties
+7 -0
@@ -138,6 +138,13 @@
138
138
%test.gitshark.mirror.allow-insecure=true
139
139
%test.gitshark.mirror.drain-interval=1h
140
140
141
+# CI/CD runners (Forgejo/Gitea runner.v1). task-timeout bounds how long a claimed task may run before
142
+# the zombie-reclaim sweep fails it (a vanished runner). zombie-reclaim-interval is how often that
143
+# sweep runs; large in tests so the scheduler never races the test's own reclaim() call.
144
+gitshark.ci.task-timeout=${GITSHARK_CI_TASK_TIMEOUT:1h}
145
+gitshark.ci.zombie-reclaim-interval=${GITSHARK_CI_ZOMBIE_RECLAIM_INTERVAL:1m}
146
+%test.gitshark.ci.zombie-reclaim-interval=1h
147
+
141
148
# Federation (ActivityPub / ForgeFed) — disabled by default.
142
149
# base-url is the public origin of this instance (e.g. https://shark.example); actor IDs are
143
150
# absolute and permanent once published, so it must be a real, non-loopback URL when enabled.
ADD
src/test/java/de/workaround/ci/ZombieReclaimTest.java
+144 -0
@@ -0,0 +1,144 @@
1
+package de.workaround.ci;
2
+
3
+import java.time.Instant;
4
+import java.time.temporal.ChronoUnit;
5
+import java.util.List;
6
+import java.util.UUID;
7
+
8
+import org.junit.jupiter.api.Test;
9
+
10
+import de.workaround.ci.proto.runner.v1.Result;
11
+import de.workaround.git.GitRepositoryService;
12
+import de.workaround.model.ActionRun;
13
+import de.workaround.model.ActionTask;
14
+import de.workaround.model.CiRunner;
15
+import de.workaround.model.Repository;
16
+import de.workaround.model.User;
17
+import io.quarkus.test.junit.QuarkusTest;
18
+import jakarta.inject.Inject;
19
+import jakarta.transaction.Transactional;
20
+
21
+import static org.junit.jupiter.api.Assertions.assertEquals;
22
+
23
+/**
24
+ * Zombie reclaim (issue #2, phase 1): a RUNNING task whose deadline has passed — its runner vanished
25
+ * mid-task — is failed cleanly and its run rolled up, while a task still within its deadline is left
26
+ * running.
27
+ */
28
+@QuarkusTest
29
+class ZombieReclaimTest
30
+{
31
+ @Inject
32
+ ZombieReclaimService reclaim;
33
+
34
+ @Inject
35
+ RunnerRegistrationService runnerService;
36
+
37
+ @Inject
38
+ GitRepositoryService repositories;
39
+
40
+ @Inject
41
+ ActionRun.Repo runs;
42
+
43
+ @Inject
44
+ ActionTask.Repo tasks;
45
+
46
+ @Inject
47
+ CiRunner.Repo runners;
48
+
49
+ @Inject
50
+ TaskProgressService progress;
51
+
52
+ @Test
53
+ void overdueTaskIsFailedAndRunnerFreed()
54
+ {
55
+ Instant now = Instant.now();
56
+ CiRunner staleRunner = registerRunner().runner();
57
+ UUID overdue = seedRunningTask("zr-a", staleRunner.uuid, now.minus(5, ChronoUnit.MINUTES));
58
+
59
+ CiRunner freshRunner = registerRunner().runner();
60
+ UUID healthy = seedRunningTask("zr-b", freshRunner.uuid, now.plus(30, ChronoUnit.MINUTES));
61
+
62
+ int reclaimed = reclaim.reclaim(now);
63
+
64
+ assertEquals(1, reclaimed, "only the past-deadline task is reclaimed");
65
+
66
+ ActionTask failed = tasks.findById(overdue);
67
+ assertEquals(ActionRun.Status.FAILURE, failed.status);
68
+ assertEquals(ActionRun.Status.FAILURE, failed.run.status);
69
+ assertEquals(CiRunner.Status.OFFLINE, runners.findByUuid(staleRunner.uuid).orElseThrow().status);
70
+
71
+ ActionTask running = tasks.findById(healthy);
72
+ assertEquals(ActionRun.Status.RUNNING, running.status);
73
+ }
74
+
75
+ @Test
76
+ void lateUpdateCannotResurrectAReclaimedTask()
77
+ {
78
+ Instant now = Instant.now();
79
+ RunnerRegistrationService.RegisteredRunner reg = registerRunner();
80
+ UUID id = seedRunningTask("zr-res", reg.runner().uuid, now.minus(5, ChronoUnit.MINUTES));
81
+
82
+ reclaim.reclaim(now);
83
+ long seq = tasks.findById(id).seq;
84
+
85
+ // The presumed-dead runner reconnects and reports success — the task must stay FAILURE.
86
+ progress.updateTask(reg.runner().uuid, reg.plaintext(), seq, Result.RESULT_SUCCESS, null);
87
+
88
+ ActionTask task = tasks.findById(id);
89
+ assertEquals(ActionRun.Status.FAILURE, task.status);
90
+ assertEquals(ActionRun.Status.FAILURE, task.run.status);
91
+ }
92
+
93
+ private RunnerRegistrationService.RegisteredRunner registerRunner()
94
+ {
95
+ String regToken = runnerService.createRegistrationToken(persistUser("zr-admin-" + shortId())).plaintext();
96
+ return runnerService.register(regToken, "zr-runner", List.of("ubuntu-latest"), "v4.0.0", false);
97
+ }
98
+
99
+ @Transactional
100
+ UUID seedRunningTask(String repoName, String runnerUuid, Instant deadline)
101
+ {
102
+ User owner = persistUser(repoName + "-" + shortId());
103
+ Repository repo = repositories.create(owner, repoName, Repository.Visibility.PUBLIC, null);
104
+ CiRunner runner = runners.findByUuid(runnerUuid).orElseThrow();
105
+
106
+ ActionRun run = new ActionRun();
107
+ run.repository = repo;
108
+ run.number = runs.maxNumber(repo) + 1;
109
+ run.workflowName = "CI";
110
+ run.workflowFile = ".forgejo/workflows/ci.yml";
111
+ run.event = "push";
112
+ run.ref = "refs/heads/main";
113
+ run.commitSha = "0000000000000000000000000000000000000000";
114
+ run.status = ActionRun.Status.RUNNING;
115
+ run.persist();
116
+
117
+ ActionTask task = new ActionTask();
118
+ task.run = run;
119
+ task.name = "build";
120
+ task.payload = "on: push";
121
+ task.runner = runner;
122
+ task.status = ActionRun.Status.RUNNING;
123
+ task.startedAt = deadline.minus(1, ChronoUnit.HOURS);
124
+ task.deadline = deadline;
125
+ task.persist();
126
+ return task.id;
127
+ }
128
+
129
+ @Transactional
130
+ User persistUser(String name)
131
+ {
132
+ User user = new User();
133
+ user.oidcSub = name;
134
+ user.username = name;
135
+ user.persist();
136
+ return user;
137
+ }
138
+
139
+ private static String shortId()
140
+ {
141
+ return UUID.randomUUID().toString().substring(0, 8);
142
+ }
143
+
144
+}