✨ (ci): Serve UpdateTask and UpdateLog to complete the run loop
Changes
6 files changed, +439 -4
MODIFY
docs/maintainers/ci-runners.md
+12 -4
@@ -13,6 +13,7 @@
13
13
| Connect endpoint | `ci/ConnectRunnerResource.java` | JAX-RS resource serving the Connect unary RPCs under `/api/actions`. |
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
+| 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`). |
16
17
| Entities | `model/CiRunner.java`, `model/CiRunnerRegistrationToken.java` | Runner state (migration `V19`). |
17
18
| 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. |
18
19
| 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. |
@@ -64,18 +65,25 @@
64
65
row is locked `FOR UPDATE SKIP LOCKED` (id-only select, to keep the lock off the nullable `runner`
65
66
join) so concurrent fetchers never claim the same task. Auth failures return the Connect
66
67
`unauthenticated` error. No long-poll; `tasks_version` is a coarse max-`seq`.
68
+- **`UpdateTask` / `UpdateLog` progress (`TaskProgressService`):** UpdateTask records the reported
69
+ result, sends a finished task's runner back to IDLE, and rolls the owning run's status up from all
70
+ its tasks (RUNNING until every task is terminal, then the worst outcome). UpdateLog appends log rows
71
+ with resume-safe contiguous semantics — rows below `action_task.log_length` are ignored, a gap above
72
+ it stops the append, and the returned `ack_index` is the durable row count. Both reject a task not
73
+ assigned to the calling runner (`unauthenticated`) and an unknown task id (`not_found`).
67
74
- Tests: `RunnerRegistrationServiceTest` (service), `ConnectRunnerResourceTest` (protobuf-over-HTTP
68
75
round-trip for Ping/Register/Declare + auth failures), `AdminAccessTest` (admin gate),
69
76
`ActionRunPersistenceTest` (run/task/log persistence, per-repo run numbering, pending-task lookup),
70
77
`WorkflowIngestServiceTest` (push → run/task creation, non-push trigger and no-workflow are no-ops),
71
78
`FetchTaskTest` (claim oldest pending over the wire, empty queue, bad credentials, and two runners
72
- racing one task → claimed at most once).
79
+ racing one task → claimed at most once), `TaskProgressTest` (UpdateTask success rolls up task+run
80
+ and frees the runner, UpdateLog append + dedup/resume, cross-runner and bad-credential rejection).
73
81
74
82
## What still needs to be implemented
75
83
76
-- **Run loop (remaining):** `UpdateTask` (step/task state) and `UpdateLog` (offset / `ack_index`
77
- resume). `FetchTask` is served; these two report progress back. Add long-poll and a real
78
- state-driven `tasks_version` when they land.
84
+- **Long-poll & real `tasks_version`:** `FetchTask` returns immediately and `tasks_version` is a
85
+ coarse max-`seq` (bumps on creation, not state change), so with several simultaneous PENDING tasks a
86
+ runner may under-poll. Add server-side long-poll and a state-driven version counter.
79
87
- **Per-job payload expansion:** ingest/dispatch deliver the raw workflow YAML as `workflow_payload`;
80
88
it needs the single job isolated/expanded. No `needs`/`matrix` yet.
81
89
- **Trigger refinement:** only bare `on: push` is honored; branch/tag/path filters and other events
MODIFY
src/main/java/de/workaround/ci/ConnectRunnerResource.java
+56 -0
@@ -1,5 +1,6 @@
1
1
package de.workaround.ci;
2
2
3
+import java.time.Instant;
3
4
import java.util.Arrays;
4
5
import java.util.List;
5
6
@@ -17,6 +18,11 @@
17
18
import de.workaround.ci.proto.runner.v1.Runner;
18
19
import de.workaround.ci.proto.runner.v1.RunnerStatus;
19
20
import de.workaround.ci.proto.runner.v1.Task;
21
+import de.workaround.ci.proto.runner.v1.TaskState;
22
+import de.workaround.ci.proto.runner.v1.UpdateLogRequest;
23
+import de.workaround.ci.proto.runner.v1.UpdateLogResponse;
24
+import de.workaround.ci.proto.runner.v1.UpdateTaskRequest;
25
+import de.workaround.ci.proto.runner.v1.UpdateTaskResponse;
20
26
import de.workaround.model.ActionTask;
21
27
import de.workaround.model.CiRunner;
22
28
import jakarta.inject.Inject;
@@ -51,6 +57,9 @@
51
57
@Inject
52
58
TaskDispatchService dispatchService;
53
59
60
+ @Inject
61
+ TaskProgressService progressService;
62
+
54
63
@POST
55
64
@Path("/ping.v1.PingService/Ping")
56
65
public Response ping(byte[] body) throws InvalidProtocolBufferException
@@ -117,6 +126,53 @@
117
126
}
118
127
}
119
128
129
+ @POST
130
+ @Path("/runner.v1.RunnerService/UpdateTask")
131
+ public Response updateTask(@HeaderParam("x-runner-uuid") String uuid, @HeaderParam("x-runner-token") String token,
132
+ byte[] body) throws InvalidProtocolBufferException
133
+ {
134
+ UpdateTaskRequest request = UpdateTaskRequest.parseFrom(body);
135
+ TaskState state = request.getState();
136
+ Instant stoppedAt = state.hasStoppedAt()
137
+ ? Instant.ofEpochSecond(state.getStoppedAt().getSeconds(), state.getStoppedAt().getNanos())
138
+ : null;
139
+ try
140
+ {
141
+ progressService.updateTask(uuid, token, state.getId(), state.getResult(), stoppedAt);
142
+ return ok(UpdateTaskResponse.newBuilder().setState(state).build().toByteArray());
143
+ }
144
+ catch (RunnerAuthenticationException e)
145
+ {
146
+ return connectError(Response.Status.UNAUTHORIZED, "unauthenticated", e.getMessage());
147
+ }
148
+ catch (TaskNotFoundException e)
149
+ {
150
+ return connectError(Response.Status.NOT_FOUND, "not_found", e.getMessage());
151
+ }
152
+ }
153
+
154
+ @POST
155
+ @Path("/runner.v1.RunnerService/UpdateLog")
156
+ public Response updateLog(@HeaderParam("x-runner-uuid") String uuid, @HeaderParam("x-runner-token") String token,
157
+ byte[] body) throws InvalidProtocolBufferException
158
+ {
159
+ UpdateLogRequest request = UpdateLogRequest.parseFrom(body);
160
+ try
161
+ {
162
+ long ackIndex = progressService.appendLog(uuid, token, request.getTaskId(), request.getIndex(),
163
+ request.getRowsList());
164
+ return ok(UpdateLogResponse.newBuilder().setAckIndex(ackIndex).build().toByteArray());
165
+ }
166
+ catch (RunnerAuthenticationException e)
167
+ {
168
+ return connectError(Response.Status.UNAUTHORIZED, "unauthenticated", e.getMessage());
169
+ }
170
+ catch (TaskNotFoundException e)
171
+ {
172
+ return connectError(Response.Status.NOT_FOUND, "not_found", e.getMessage());
173
+ }
174
+ }
175
+
120
176
private static Task toProto(ActionTask task)
121
177
{
122
178
Task.Builder builder = Task.newBuilder().setId(task.seq);
ADD
src/main/java/de/workaround/ci/TaskNotFoundException.java
+10 -0
@@ -0,0 +1,10 @@
1
+package de.workaround.ci;
2
+
3
+/** A runner referenced a task id ({@code seq}) that does not exist. */
4
+public class TaskNotFoundException extends RuntimeException
5
+{
6
+ public TaskNotFoundException(String message)
7
+ {
8
+ super(message);
9
+ }
10
+}
ADD
src/main/java/de/workaround/ci/TaskProgressService.java
+152 -0
@@ -0,0 +1,152 @@
1
+package de.workaround.ci;
2
+
3
+import java.time.Instant;
4
+import java.util.List;
5
+
6
+import com.google.protobuf.Timestamp;
7
+
8
+import de.workaround.ci.proto.runner.v1.LogRow;
9
+import de.workaround.ci.proto.runner.v1.Result;
10
+import de.workaround.model.ActionLog;
11
+import de.workaround.model.ActionRun;
12
+import de.workaround.model.ActionTask;
13
+import de.workaround.model.CiRunner;
14
+import jakarta.enterprise.context.ApplicationScoped;
15
+import jakarta.inject.Inject;
16
+import jakarta.transaction.Transactional;
17
+
18
+/**
19
+ * Applies the progress a runner reports for its claimed task (issue #2, phase 1): UpdateTask records
20
+ * the task result and rolls the owning run's status up from all its tasks; UpdateLog appends log rows
21
+ * with resume-safe, contiguous-append semantics. Both authenticate the runner and reject a task that
22
+ * is not assigned to it.
23
+ */
24
+@ApplicationScoped
25
+public class TaskProgressService
26
+{
27
+ @Inject
28
+ RunnerRegistrationService runnerService;
29
+
30
+ @Inject
31
+ ActionTask.Repo tasks;
32
+
33
+ @Inject
34
+ ActionLog.Repo logs;
35
+
36
+ /**
37
+ * Record the result the runner reports for a task and roll the owning run's status up.
38
+ *
39
+ * @throws RunnerAuthenticationException if the credentials are bad or the task is not this runner's
40
+ * @throws TaskNotFoundException if no task has the given surrogate id
41
+ */
42
+ @Transactional
43
+ public ActionTask updateTask(String uuid, String token, long taskSeq, Result result, Instant stoppedAt)
44
+ {
45
+ CiRunner runner = authenticate(uuid, token);
46
+ ActionTask task = ownedTask(taskSeq, runner);
47
+
48
+ ActionRun.Status status = map(result);
49
+ task.status = status;
50
+ if (status.isTerminal())
51
+ {
52
+ task.finishedAt = stoppedAt != null ? stoppedAt : Instant.now();
53
+ runner.status = CiRunner.Status.IDLE;
54
+ }
55
+ rollUpRun(task.run);
56
+ return task;
57
+ }
58
+
59
+ /**
60
+ * Append log rows a runner streams for a task and return the durable row count (the {@code
61
+ * ack_index}). Rows already stored (index below the current length) are ignored; a gap above the
62
+ * current length stops the append so the runner resends contiguously.
63
+ */
64
+ @Transactional
65
+ public long appendLog(String uuid, String token, long taskSeq, long index, List<LogRow> rows)
66
+ {
67
+ CiRunner runner = authenticate(uuid, token);
68
+ ActionTask task = ownedTask(taskSeq, runner);
69
+
70
+ int length = task.logLength;
71
+ for (int i = 0; i < rows.size(); i++)
72
+ {
73
+ long line = index + i;
74
+ if (line < length)
75
+ {
76
+ continue; // already stored (runner resend)
77
+ }
78
+ if (line > length)
79
+ {
80
+ break; // gap: refuse to store non-contiguously, runner will resend from `length`
81
+ }
82
+ LogRow row = rows.get(i);
83
+ ActionLog log = new ActionLog();
84
+ log.task = task;
85
+ log.lineIndex = (int) line;
86
+ log.content = row.getContent();
87
+ log.timestamp = toInstant(row.getTime());
88
+ log.persist();
89
+ length++;
90
+ }
91
+ task.logLength = length;
92
+ return length;
93
+ }
94
+
95
+ private CiRunner authenticate(String uuid, String token)
96
+ {
97
+ CiRunner runner = runnerService.authenticate(uuid, token);
98
+ runner.lastSeen = Instant.now();
99
+ return runner;
100
+ }
101
+
102
+ private ActionTask ownedTask(long taskSeq, CiRunner runner)
103
+ {
104
+ ActionTask task = tasks.findBySeq(taskSeq)
105
+ .orElseThrow(() -> new TaskNotFoundException("No task with id " + taskSeq));
106
+ if (task.runner == null || !task.runner.uuid.equals(runner.uuid))
107
+ {
108
+ throw new RunnerAuthenticationException("Task " + taskSeq + " is not assigned to this runner");
109
+ }
110
+ return task;
111
+ }
112
+
113
+ /** 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)
115
+ {
116
+ List<ActionTask> all = tasks.findByRun(run);
117
+ boolean anyRunning = all.stream().anyMatch(t -> !t.status.isTerminal());
118
+ if (anyRunning)
119
+ {
120
+ run.status = ActionRun.Status.RUNNING;
121
+ return;
122
+ }
123
+ boolean anyFailure = all.stream().anyMatch(t -> t.status == ActionRun.Status.FAILURE);
124
+ boolean anyCancelled = all.stream().anyMatch(t -> t.status == ActionRun.Status.CANCELLED);
125
+ run.status = anyFailure ? ActionRun.Status.FAILURE
126
+ : anyCancelled ? ActionRun.Status.CANCELLED
127
+ : ActionRun.Status.SUCCESS;
128
+ run.finishedAt = Instant.now();
129
+ }
130
+
131
+ private static ActionRun.Status map(Result result)
132
+ {
133
+ return switch (result)
134
+ {
135
+ case RESULT_SUCCESS, RESULT_SKIPPED -> ActionRun.Status.SUCCESS;
136
+ case RESULT_FAILURE -> ActionRun.Status.FAILURE;
137
+ case RESULT_CANCELLED -> ActionRun.Status.CANCELLED;
138
+ // RESULT_UNSPECIFIED (and any unknown) is a mid-run heartbeat: still running.
139
+ default -> ActionRun.Status.RUNNING;
140
+ };
141
+ }
142
+
143
+ private static Instant toInstant(Timestamp timestamp)
144
+ {
145
+ if (timestamp == null || (timestamp.getSeconds() == 0 && timestamp.getNanos() == 0))
146
+ {
147
+ return Instant.now();
148
+ }
149
+ return Instant.ofEpochSecond(timestamp.getSeconds(), timestamp.getNanos());
150
+ }
151
+
152
+}
MODIFY
src/main/java/de/workaround/model/ActionTask.java
+3 -0
@@ -89,6 +89,9 @@
89
89
@Find
90
90
Optional<ActionTask> findByIdAndRunner(UUID id, CiRunner runner);
91
91
92
+ @Find
93
+ Optional<ActionTask> findBySeq(long seq);
94
+
92
95
/** The highest surrogate id issued so far; the coarse {@code tasks_version} handed to runners. */
93
96
@HQL("select coalesce(max(t.seq), 0) from ActionTask t")
94
97
long maxSeq();
ADD
src/test/java/de/workaround/ci/TaskProgressTest.java
+206 -0
@@ -0,0 +1,206 @@
1
+package de.workaround.ci;
2
+
3
+import java.util.List;
4
+import java.util.UUID;
5
+
6
+import org.junit.jupiter.api.Test;
7
+
8
+import com.google.protobuf.InvalidProtocolBufferException;
9
+
10
+import de.workaround.ci.proto.runner.v1.LogRow;
11
+import de.workaround.ci.proto.runner.v1.Result;
12
+import de.workaround.ci.proto.runner.v1.TaskState;
13
+import de.workaround.ci.proto.runner.v1.UpdateLogRequest;
14
+import de.workaround.ci.proto.runner.v1.UpdateLogResponse;
15
+import de.workaround.ci.proto.runner.v1.UpdateTaskRequest;
16
+import de.workaround.ci.proto.runner.v1.UpdateTaskResponse;
17
+import de.workaround.git.GitRepositoryService;
18
+import de.workaround.model.ActionLog;
19
+import de.workaround.model.ActionRun;
20
+import de.workaround.model.ActionTask;
21
+import de.workaround.model.CiRunner;
22
+import de.workaround.model.Repository;
23
+import de.workaround.model.User;
24
+import io.quarkus.test.junit.QuarkusTest;
25
+import jakarta.inject.Inject;
26
+import jakarta.transaction.Transactional;
27
+
28
+import static io.restassured.RestAssured.given;
29
+import static org.junit.jupiter.api.Assertions.assertEquals;
30
+
31
+/**
32
+ * Drives UpdateTask and UpdateLog over the Connect wire (issue #2, phase 1): a runner reports its
33
+ * claimed task's result (task+run go terminal, runner returns IDLE) and streams log rows with
34
+ * resume-safe append semantics (duplicate rows are ignored, ack_index is the durable row count).
35
+ */
36
+@QuarkusTest
37
+class TaskProgressTest
38
+{
39
+ private static final String PROTO = "application/proto";
40
+
41
+ @Inject
42
+ RunnerRegistrationService runnerService;
43
+
44
+ @Inject
45
+ GitRepositoryService repositories;
46
+
47
+ @Inject
48
+ ActionRun.Repo runs;
49
+
50
+ @Inject
51
+ ActionTask.Repo tasks;
52
+
53
+ @Inject
54
+ ActionLog.Repo logs;
55
+
56
+ @Inject
57
+ CiRunner.Repo runners;
58
+
59
+ @Inject
60
+ jakarta.persistence.EntityManager em;
61
+
62
+ @Test
63
+ void updateTaskWithSuccessMarksTaskAndRunDone() throws InvalidProtocolBufferException
64
+ {
65
+ RunnerRegistrationService.RegisteredRunner reg = registerRunner();
66
+ long seq = seedClaimedTask("tp-a", reg.runner().uuid);
67
+
68
+ byte[] bytes = given()
69
+ .contentType(PROTO)
70
+ .header("x-runner-uuid", reg.runner().uuid)
71
+ .header("x-runner-token", reg.plaintext())
72
+ .body(UpdateTaskRequest.newBuilder()
73
+ .setState(TaskState.newBuilder().setId(seq).setResult(Result.RESULT_SUCCESS))
74
+ .build().toByteArray())
75
+ .when().post("/api/actions/runner.v1.RunnerService/UpdateTask")
76
+ .then().statusCode(200)
77
+ .extract().asByteArray();
78
+
79
+ UpdateTaskResponse response = UpdateTaskResponse.parseFrom(bytes);
80
+ assertEquals(seq, response.getState().getId());
81
+
82
+ ActionTask task = tasks.findBySeq(seq).orElseThrow();
83
+ assertEquals(ActionRun.Status.SUCCESS, task.status);
84
+ assertEquals(ActionRun.Status.SUCCESS, task.run.status);
85
+ assertEquals(CiRunner.Status.IDLE, task.runner.status);
86
+ }
87
+
88
+ @Test
89
+ void updateLogAppendsRowsAndIsResumeSafe() throws InvalidProtocolBufferException
90
+ {
91
+ RunnerRegistrationService.RegisteredRunner reg = registerRunner();
92
+ long seq = seedClaimedTask("tp-b", reg.runner().uuid);
93
+
94
+ UpdateLogResponse first = UpdateLogResponse.parseFrom(sendLog(reg, seq, 0, List.of("line 0", "line 1")));
95
+ assertEquals(2, first.getAckIndex(), "two contiguous rows accepted from index 0");
96
+
97
+ // Runner resends line 1 (already stored) plus a new line 2 — the dup must be ignored.
98
+ UpdateLogResponse second = UpdateLogResponse.parseFrom(sendLog(reg, seq, 1, List.of("line 1", "line 2")));
99
+ assertEquals(3, second.getAckIndex(), "ack is the durable row count, dup ignored");
100
+
101
+ ActionTask task = tasks.findBySeq(seq).orElseThrow();
102
+ assertEquals(3, task.logLength);
103
+ List<ActionLog> stored = logs.findByTask(task);
104
+ assertEquals(3, stored.size());
105
+ assertEquals("line 0", stored.get(0).content);
106
+ assertEquals("line 2", stored.get(2).content);
107
+ }
108
+
109
+ @Test
110
+ void updateTaskByAnotherRunnerIsRejected()
111
+ {
112
+ RunnerRegistrationService.RegisteredRunner owner = registerRunner();
113
+ RunnerRegistrationService.RegisteredRunner other = registerRunner();
114
+ long seq = seedClaimedTask("tp-c", owner.runner().uuid);
115
+
116
+ given()
117
+ .contentType(PROTO)
118
+ .header("x-runner-uuid", other.runner().uuid)
119
+ .header("x-runner-token", other.plaintext())
120
+ .body(UpdateTaskRequest.newBuilder()
121
+ .setState(TaskState.newBuilder().setId(seq).setResult(Result.RESULT_SUCCESS))
122
+ .build().toByteArray())
123
+ .when().post("/api/actions/runner.v1.RunnerService/UpdateTask")
124
+ .then().statusCode(401);
125
+ }
126
+
127
+ @Test
128
+ void updateLogWithBadCredentialsIsUnauthenticated()
129
+ {
130
+ given()
131
+ .contentType(PROTO)
132
+ .header("x-runner-uuid", "nope")
133
+ .header("x-runner-token", "gsrt_bogus")
134
+ .body(UpdateLogRequest.newBuilder().setTaskId(1).build().toByteArray())
135
+ .when().post("/api/actions/runner.v1.RunnerService/UpdateLog")
136
+ .then().statusCode(401);
137
+ }
138
+
139
+ private byte[] sendLog(RunnerRegistrationService.RegisteredRunner reg, long seq, long index, List<String> lines)
140
+ {
141
+ UpdateLogRequest.Builder request = UpdateLogRequest.newBuilder().setTaskId(seq).setIndex(index);
142
+ for (String line : lines)
143
+ {
144
+ request.addRows(LogRow.newBuilder().setContent(line));
145
+ }
146
+ return given()
147
+ .contentType(PROTO)
148
+ .header("x-runner-uuid", reg.runner().uuid)
149
+ .header("x-runner-token", reg.plaintext())
150
+ .body(request.build().toByteArray())
151
+ .when().post("/api/actions/runner.v1.RunnerService/UpdateLog")
152
+ .then().statusCode(200)
153
+ .extract().asByteArray();
154
+ }
155
+
156
+ private RunnerRegistrationService.RegisteredRunner registerRunner()
157
+ {
158
+ String regToken = runnerService.createRegistrationToken(persistUser("tp-admin-" + shortId())).plaintext();
159
+ return runnerService.register(regToken, "tp-runner", List.of("ubuntu-latest"), "v4.0.0", false);
160
+ }
161
+
162
+ @Transactional
163
+ long seedClaimedTask(String repoName, String runnerUuid)
164
+ {
165
+ User owner = persistUser(repoName + "-" + shortId());
166
+ Repository repo = repositories.create(owner, repoName, Repository.Visibility.PUBLIC, null);
167
+ CiRunner runner = runners.findByUuid(runnerUuid).orElseThrow();
168
+
169
+ ActionRun run = new ActionRun();
170
+ run.repository = repo;
171
+ run.number = runs.maxNumber(repo) + 1;
172
+ run.workflowName = "CI";
173
+ run.workflowFile = ".forgejo/workflows/ci.yml";
174
+ run.event = "push";
175
+ run.ref = "refs/heads/main";
176
+ run.commitSha = "0000000000000000000000000000000000000000";
177
+ run.status = ActionRun.Status.RUNNING;
178
+ run.persist();
179
+
180
+ ActionTask task = new ActionTask();
181
+ task.run = run;
182
+ task.name = "build";
183
+ task.payload = "on: push";
184
+ task.runner = runner;
185
+ task.status = ActionRun.Status.RUNNING;
186
+ task.persist();
187
+ em.flush(); // populate the DB-generated seq before reading it
188
+ return task.seq;
189
+ }
190
+
191
+ @Transactional
192
+ User persistUser(String name)
193
+ {
194
+ User user = new User();
195
+ user.oidcSub = name;
196
+ user.username = name;
197
+ user.persist();
198
+ return user;
199
+ }
200
+
201
+ private static String shortId()
202
+ {
203
+ return UUID.randomUUID().toString().substring(0, 8);
204
+ }
205
+
206
+}