@@ -196,6 +196,168 @@ func TestRunnerHeartbeatClaimsQueuedJob(t *testing.T) { |
| 196 | } | 196 | } |
| 197 | } | 197 | } |
| 198 | | 198 | |
| | 199 | +func TestRunnerHeartbeatRespectsRepoConcurrencyCap(t *testing.T) { |
| | 200 | + ctx := context.Background() |
| | 201 | + pool := dbtest.NewTestDB(t) |
| | 202 | + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) |
| | 203 | + repoID, userID := setupRunnerAPIRepo(t, pool) |
| | 204 | + q := actionsdb.New() |
| | 205 | + if _, err := q.UpsertActionsRepoPolicy(ctx, pool, actionsdb.UpsertActionsRepoPolicyParams{ |
| | 206 | + RepoID: repoID, |
| | 207 | + ActionsEnabled: actionsdb.ActionsPolicyStateInherit, |
| | 208 | + MaxRepoConcurrentJobs: pgtype.Int4{Int32: 1, Valid: true}, |
| | 209 | + }); err != nil { |
| | 210 | + t.Fatalf("UpsertActionsRepoPolicy: %v", err) |
| | 211 | + } |
| | 212 | + |
| | 213 | + firstRunID := enqueueRunnerAPIRunWithTriggerID(t, pool, logger, repoID, userID, "repo-cap-1") |
| | 214 | + token, _ := registerRunnerForTest(t, pool, []string{"ubuntu-latest", "linux"}, 2) |
| | 215 | + router := newRunnerAPIRouter(t, pool, logger, runnerAPISigner(t, time.Now())) |
| | 216 | + firstClaim := claimRunnerHeartbeat(t, router, token, 2) |
| | 217 | + if firstClaim.Job.RunID != firstRunID { |
| | 218 | + t.Fatalf("first claim run_id=%d want %d", firstClaim.Job.RunID, firstRunID) |
| | 219 | + } |
| | 220 | + |
| | 221 | + secondRunID := enqueueRunnerAPIRunWithTriggerID(t, pool, logger, repoID, userID, "repo-cap-2") |
| | 222 | + req := httptest.NewRequest(http.MethodPost, "/api/v1/runners/heartbeat", |
| | 223 | + strings.NewReader(`{"labels":["ubuntu-latest","linux"],"capacity":2}`)) |
| | 224 | + req.Header.Set("Authorization", "Bearer "+token) |
| | 225 | + rr := httptest.NewRecorder() |
| | 226 | + router.ServeHTTP(rr, req) |
| | 227 | + if rr.Code != http.StatusNoContent { |
| | 228 | + t.Fatalf("capped heartbeat status: got %d, want 204; body=%s", rr.Code, rr.Body.String()) |
| | 229 | + } |
| | 230 | + jobs, err := q.ListJobsForRun(ctx, pool, secondRunID) |
| | 231 | + if err != nil { |
| | 232 | + t.Fatalf("ListJobsForRun: %v", err) |
| | 233 | + } |
| | 234 | + if len(jobs) != 1 || jobs[0].Status != actionsdb.WorkflowJobStatusQueued { |
| | 235 | + t.Fatalf("second run job should remain queued/unclaimed: %+v", jobs) |
| | 236 | + } |
| | 237 | + secondJob, err := q.GetWorkflowJobByID(ctx, pool, jobs[0].ID) |
| | 238 | + if err != nil { |
| | 239 | + t.Fatalf("GetWorkflowJobByID: %v", err) |
| | 240 | + } |
| | 241 | + if secondJob.RunnerID.Valid { |
| | 242 | + t.Fatalf("second run job should not have a runner: %+v", secondJob) |
| | 243 | + } |
| | 244 | + |
| | 245 | + if _, err := q.UpdateWorkflowJobStatus(ctx, pool, actionsdb.UpdateWorkflowJobStatusParams{ |
| | 246 | + ID: firstClaim.Job.ID, |
| | 247 | + Status: actionsdb.WorkflowJobStatusCompleted, |
| | 248 | + Conclusion: actionsdb.NullCheckConclusion{CheckConclusion: actionsdb.CheckConclusionSuccess, Valid: true}, |
| | 249 | + CompletedAt: pgtype.Timestamptz{ |
| | 250 | + Time: time.Now(), |
| | 251 | + Valid: true, |
| | 252 | + }, |
| | 253 | + }); err != nil { |
| | 254 | + t.Fatalf("UpdateWorkflowJobStatus: %v", err) |
| | 255 | + } |
| | 256 | + secondClaim := claimRunnerHeartbeat(t, router, token, 2) |
| | 257 | + if secondClaim.Job.RunID != secondRunID { |
| | 258 | + t.Fatalf("second claim run_id=%d want %d", secondClaim.Job.RunID, secondRunID) |
| | 259 | + } |
| | 260 | +} |
| | 261 | + |
| | 262 | +func TestRunnerHeartbeatRespectsOwnerConcurrencyCap(t *testing.T) { |
| | 263 | + ctx := context.Background() |
| | 264 | + pool := dbtest.NewTestDB(t) |
| | 265 | + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) |
| | 266 | + firstRepoID, userID := setupRunnerAPIRepo(t, pool) |
| | 267 | + secondRepo, err := reposdb.New().CreateRepo(ctx, pool, reposdb.CreateRepoParams{ |
| | 268 | + OwnerUserID: pgtype.Int8{Int64: userID, Valid: true}, |
| | 269 | + Name: "second", |
| | 270 | + DefaultBranch: "trunk", |
| | 271 | + Visibility: reposdb.RepoVisibilityPublic, |
| | 272 | + }) |
| | 273 | + if err != nil { |
| | 274 | + t.Fatalf("CreateRepo second: %v", err) |
| | 275 | + } |
| | 276 | + q := actionsdb.New() |
| | 277 | + if _, err := q.UpsertActionsRepoPolicy(ctx, pool, actionsdb.UpsertActionsRepoPolicyParams{ |
| | 278 | + RepoID: secondRepo.ID, |
| | 279 | + ActionsEnabled: actionsdb.ActionsPolicyStateInherit, |
| | 280 | + MaxOwnerConcurrentJobs: pgtype.Int4{Int32: 1, Valid: true}, |
| | 281 | + }); err != nil { |
| | 282 | + t.Fatalf("UpsertActionsRepoPolicy: %v", err) |
| | 283 | + } |
| | 284 | + |
| | 285 | + firstRunID := enqueueRunnerAPIRunWithTriggerID(t, pool, logger, firstRepoID, userID, "owner-cap-1") |
| | 286 | + token, _ := registerRunnerForTest(t, pool, []string{"ubuntu-latest", "linux"}, 2) |
| | 287 | + router := newRunnerAPIRouter(t, pool, logger, runnerAPISigner(t, time.Now())) |
| | 288 | + firstClaim := claimRunnerHeartbeat(t, router, token, 2) |
| | 289 | + if firstClaim.Job.RunID != firstRunID { |
| | 290 | + t.Fatalf("first claim run_id=%d want %d", firstClaim.Job.RunID, firstRunID) |
| | 291 | + } |
| | 292 | + |
| | 293 | + secondRunID := enqueueRunnerAPIRunWithTriggerID(t, pool, logger, secondRepo.ID, userID, "owner-cap-2") |
| | 294 | + req := httptest.NewRequest(http.MethodPost, "/api/v1/runners/heartbeat", |
| | 295 | + strings.NewReader(`{"labels":["ubuntu-latest","linux"],"capacity":2}`)) |
| | 296 | + req.Header.Set("Authorization", "Bearer "+token) |
| | 297 | + rr := httptest.NewRecorder() |
| | 298 | + router.ServeHTTP(rr, req) |
| | 299 | + if rr.Code != http.StatusNoContent { |
| | 300 | + t.Fatalf("owner-capped heartbeat status: got %d, want 204; body=%s", rr.Code, rr.Body.String()) |
| | 301 | + } |
| | 302 | + jobs, err := q.ListJobsForRun(ctx, pool, secondRunID) |
| | 303 | + if err != nil { |
| | 304 | + t.Fatalf("ListJobsForRun: %v", err) |
| | 305 | + } |
| | 306 | + if len(jobs) != 1 || jobs[0].Status != actionsdb.WorkflowJobStatusQueued { |
| | 307 | + t.Fatalf("second owner job should remain queued/unclaimed: %+v", jobs) |
| | 308 | + } |
| | 309 | + secondJob, err := q.GetWorkflowJobByID(ctx, pool, jobs[0].ID) |
| | 310 | + if err != nil { |
| | 311 | + t.Fatalf("GetWorkflowJobByID: %v", err) |
| | 312 | + } |
| | 313 | + if secondJob.RunnerID.Valid { |
| | 314 | + t.Fatalf("second owner job should not have a runner: %+v", secondJob) |
| | 315 | + } |
| | 316 | +} |
| | 317 | + |
| | 318 | +func TestRunnerHeartbeatSiteDisableOverridesRepoEnable(t *testing.T) { |
| | 319 | + ctx := context.Background() |
| | 320 | + pool := dbtest.NewTestDB(t) |
| | 321 | + logger := slog.New(slog.NewTextHandler(io.Discard, nil)) |
| | 322 | + repoID, userID := setupRunnerAPIRepo(t, pool) |
| | 323 | + q := actionsdb.New() |
| | 324 | + if _, err := q.UpsertActionsRepoPolicy(ctx, pool, actionsdb.UpsertActionsRepoPolicyParams{ |
| | 325 | + RepoID: repoID, |
| | 326 | + ActionsEnabled: actionsdb.ActionsPolicyStateEnabled, |
| | 327 | + }); err != nil { |
| | 328 | + t.Fatalf("UpsertActionsRepoPolicy: %v", err) |
| | 329 | + } |
| | 330 | + runID := enqueueRunnerAPIRunWithTriggerID(t, pool, logger, repoID, userID, "site-disable-claim") |
| | 331 | + if _, err := pool.Exec(ctx, `UPDATE actions_site_policy SET actions_enabled = false WHERE id = true`); err != nil { |
| | 332 | + t.Fatalf("disable site policy: %v", err) |
| | 333 | + } |
| | 334 | + |
| | 335 | + token, _ := registerRunnerForTest(t, pool, []string{"ubuntu-latest", "linux"}, 1) |
| | 336 | + router := newRunnerAPIRouter(t, pool, logger, runnerAPISigner(t, time.Now())) |
| | 337 | + req := httptest.NewRequest(http.MethodPost, "/api/v1/runners/heartbeat", |
| | 338 | + strings.NewReader(`{"labels":["ubuntu-latest","linux"],"capacity":1}`)) |
| | 339 | + req.Header.Set("Authorization", "Bearer "+token) |
| | 340 | + rr := httptest.NewRecorder() |
| | 341 | + router.ServeHTTP(rr, req) |
| | 342 | + if rr.Code != http.StatusNoContent { |
| | 343 | + t.Fatalf("site-disabled heartbeat status: got %d, want 204; body=%s", rr.Code, rr.Body.String()) |
| | 344 | + } |
| | 345 | + jobs, err := q.ListJobsForRun(ctx, pool, runID) |
| | 346 | + if err != nil { |
| | 347 | + t.Fatalf("ListJobsForRun: %v", err) |
| | 348 | + } |
| | 349 | + if len(jobs) != 1 || jobs[0].Status != actionsdb.WorkflowJobStatusQueued { |
| | 350 | + t.Fatalf("site-disabled job should remain queued: %+v", jobs) |
| | 351 | + } |
| | 352 | + job, err := q.GetWorkflowJobByID(ctx, pool, jobs[0].ID) |
| | 353 | + if err != nil { |
| | 354 | + t.Fatalf("GetWorkflowJobByID: %v", err) |
| | 355 | + } |
| | 356 | + if job.RunnerID.Valid { |
| | 357 | + t.Fatalf("site-disabled job should not have a runner: %+v", job) |
| | 358 | + } |
| | 359 | +} |
| | 360 | + |
| 199 | func TestRunnerHeartbeatDoesNotClaimWhenDraining(t *testing.T) { | 361 | func TestRunnerHeartbeatDoesNotClaimWhenDraining(t *testing.T) { |
| 200 | ctx := context.Background() | 362 | ctx := context.Background() |
| 201 | pool := dbtest.NewTestDB(t) | 363 | pool := dbtest.NewTestDB(t) |
@@ -1099,10 +1261,20 @@ func setupRunnerAPIRepo(t *testing.T, pool *pgxpool.Pool) (repoID, userID int64) |
| 1099 | | 1261 | |
| 1100 | func enqueueRunnerAPIRun(t *testing.T, pool *pgxpool.Pool, logger *slog.Logger, repoID, userID int64) int64 { | 1262 | func enqueueRunnerAPIRun(t *testing.T, pool *pgxpool.Pool, logger *slog.Logger, repoID, userID int64) int64 { |
| 1101 | t.Helper() | 1263 | t.Helper() |
| 1102 | - return enqueueRunnerAPIEventRun(t, pool, logger, repoID, userID, trigger.EventPush, map[string]any{"ref": "refs/heads/trunk"}) | 1264 | + return enqueueRunnerAPIRunWithTriggerID(t, pool, logger, repoID, userID, "runner-api-test:push") |
| | 1265 | +} |
| | 1266 | + |
| | 1267 | +func enqueueRunnerAPIRunWithTriggerID(t *testing.T, pool *pgxpool.Pool, logger *slog.Logger, repoID, userID int64, triggerID string) int64 { |
| | 1268 | + t.Helper() |
| | 1269 | + return enqueueRunnerAPIEventRunWithTriggerID(t, pool, logger, repoID, userID, trigger.EventPush, map[string]any{"ref": "refs/heads/trunk"}, triggerID) |
| 1103 | } | 1270 | } |
| 1104 | | 1271 | |
| 1105 | func enqueueRunnerAPIEventRun(t *testing.T, pool *pgxpool.Pool, logger *slog.Logger, repoID, userID int64, event trigger.EventKind, payload map[string]any) int64 { | 1272 | func enqueueRunnerAPIEventRun(t *testing.T, pool *pgxpool.Pool, logger *slog.Logger, repoID, userID int64, event trigger.EventKind, payload map[string]any) int64 { |
| | 1273 | + t.Helper() |
| | 1274 | + return enqueueRunnerAPIEventRunWithTriggerID(t, pool, logger, repoID, userID, event, payload, "runner-api-test:"+string(event)) |
| | 1275 | +} |
| | 1276 | + |
| | 1277 | +func enqueueRunnerAPIEventRunWithTriggerID(t *testing.T, pool *pgxpool.Pool, logger *slog.Logger, repoID, userID int64, event trigger.EventKind, payload map[string]any, triggerID string) int64 { |
| 1106 | t.Helper() | 1278 | t.Helper() |
| 1107 | wf, diags, err := workflow.Parse([]byte(`name: ci | 1279 | wf, diags, err := workflow.Parse([]byte(`name: ci |
| 1108 | on: push | 1280 | on: push |
@@ -1129,7 +1301,7 @@ jobs: |
| 1129 | EventKind: event, | 1301 | EventKind: event, |
| 1130 | EventPayload: payload, | 1302 | EventPayload: payload, |
| 1131 | ActorUserID: userID, | 1303 | ActorUserID: userID, |
| 1132 | - TriggerEventID: "runner-api-test:" + string(event), | 1304 | + TriggerEventID: triggerID, |
| 1133 | Workflow: wf, | 1305 | Workflow: wf, |
| 1134 | }) | 1306 | }) |
| 1135 | if err != nil { | 1307 | if err != nil { |
@@ -1138,6 +1310,34 @@ jobs: |
| 1138 | return res.RunID | 1310 | return res.RunID |
| 1139 | } | 1311 | } |
| 1140 | | 1312 | |
| | 1313 | +type runnerHeartbeatClaim struct { |
| | 1314 | + Token string `json:"token"` |
| | 1315 | + Job struct { |
| | 1316 | + ID int64 `json:"id"` |
| | 1317 | + RunID int64 `json:"run_id"` |
| | 1318 | + } `json:"job"` |
| | 1319 | +} |
| | 1320 | + |
| | 1321 | +func claimRunnerHeartbeat(t *testing.T, router http.Handler, token string, capacity int) runnerHeartbeatClaim { |
| | 1322 | + t.Helper() |
| | 1323 | + body := fmt.Sprintf(`{"labels":["ubuntu-latest","linux"],"capacity":%d}`, capacity) |
| | 1324 | + req := httptest.NewRequest(http.MethodPost, "/api/v1/runners/heartbeat", strings.NewReader(body)) |
| | 1325 | + req.Header.Set("Authorization", "Bearer "+token) |
| | 1326 | + rr := httptest.NewRecorder() |
| | 1327 | + router.ServeHTTP(rr, req) |
| | 1328 | + if rr.Code != http.StatusOK { |
| | 1329 | + t.Fatalf("heartbeat status: got %d, want 200; body=%s", rr.Code, rr.Body.String()) |
| | 1330 | + } |
| | 1331 | + var claim runnerHeartbeatClaim |
| | 1332 | + if err := json.Unmarshal(rr.Body.Bytes(), &claim); err != nil { |
| | 1333 | + t.Fatalf("decode heartbeat claim: %v", err) |
| | 1334 | + } |
| | 1335 | + if claim.Token == "" || claim.Job.ID == 0 || claim.Job.RunID == 0 { |
| | 1336 | + t.Fatalf("incomplete heartbeat claim: %+v", claim) |
| | 1337 | + } |
| | 1338 | + return claim |
| | 1339 | +} |
| | 1340 | + |
| 1141 | func registerRunnerForTest(t *testing.T, pool *pgxpool.Pool, labels []string, capacity int32) (token string, runnerID int64) { | 1341 | func registerRunnerForTest(t *testing.T, pool *pgxpool.Pool, labels []string, capacity int32) (token string, runnerID int64) { |
| 1142 | t.Helper() | 1342 | t.Helper() |
| 1143 | token, tokenHash, err := runnertoken.New() | 1343 | token, tokenHash, err := runnertoken.New() |